Refuse a legacy-form plan before the superseding validation path

The affected_paths check ran after the PLAN_SPEC_INVALID return, so a spec in
the old mixed form that also carried another validation error took the
superseding path, recorded a new current attempt and orphaned an open wave,
the durable write owner 9=A's refusal exists to avoid. The resource-form
refusal now comes first; the ordinary validation tests submit specs in the
current form.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
This commit is contained in:
Ouroboros 2026-09-15 22:24:58 +03:00
parent 9f0703ff90
commit b88c2fb2dd
4 changed files with 32 additions and 13 deletions

View file

@ -476,16 +476,17 @@ def _prepare_plan_inputs(ctx: ToolContext, request: "_PlanRequest", state_root:
errors = ["plan: required non-empty prose", *errors]
if request.reviewer_effort and request.reviewer_effort not in _REVIEWER_EFFORT_SCHEMA["enum"]:
errors.append(f"reviewer_effort: not on the effort scale {list(_REVIEWER_EFFORT_SCHEMA['enum'])}")
if errors:
return {"error": "ERROR: PLAN_SPEC_INVALID: " + "; ".join(errors) + ". No reviewer was called.",
"code": "TOOL_ARG_ERROR"}
if isinstance(raw_spec, dict) and "affected_paths" not in raw_spec:
# Owner 9=A: a spec in the old mixed form is refused BEFORE any paid dispatch, because
# `affected_resources` used to be read as a path list — a prose item became "a file under
# the Ouroboros repo" and bought every reviewer the whole constitution (~470k tokens/cycle)
# for a deck. The refusal owns its code: a message containing `PLAN_SPEC_INVALID`
# takes the durable superseding-attempt path below and would orphan an open wave.
# for a deck. The refusal owns its code and comes first: a message containing
# `PLAN_SPEC_INVALID` takes the durable superseding-attempt path below and would orphan
# an open wave, so a legacy-form spec that also carries another error must not reach it.
return _resource_form_refusal(ctx, state_root)
if errors:
return {"error": "ERROR: PLAN_SPEC_INVALID: " + "; ".join(errors) + ". No reviewer was called.",
"code": "TOOL_ARG_ERROR"}
from ouroboros.review_substrate import review_repo_dirs_for
try:
system_root, active_root = review_repo_dirs_for(ctx)

View file

@ -199,6 +199,24 @@ def test_a_legacy_form_submission_is_refused_and_records_nothing(harness): # no
assert json.dumps(_state(harness), sort_keys=True, ensure_ascii=False) == before
def test_a_legacy_form_with_another_error_is_still_refused_before_the_superseding_path(harness): # noqa: F811
"""A legacy-form spec that ALSO fails ordinary validation must take the non-superseding
refusal, not the `PLAN_SPEC_INVALID` path that records a new attempt over an open wave."""
ask = json.dumps([_finding("f1", "need_evidence", breaks="goal", summary="who signs off?")])
substrate = harness.install({"s1": ask, "s2": CLEAN, "s3": CLEAN})
ctx = harness.make_ctx()
assert _control(_call(ctx)) == {"outcome": "REVIEW_REQUIRED", "closed": False}
before = json.dumps(_state(harness), sort_keys=True, ensure_ascii=False)
calls_before = len(substrate.calls)
legacy = {key: value for key, value in DECK_SPEC.items() if key != "affected_paths"}
out = _call(ctx, spec=legacy, reviewer_effort="galactic")
assert "PLAN_RESOURCE_FORM_REQUIRED" in out and "PLAN_SPEC_INVALID" not in out
assert len(substrate.calls) == calls_before
assert json.dumps(_state(harness), sort_keys=True, ensure_ascii=False) == before
def test_a_legacy_form_submission_over_an_open_wave_points_at_the_free_exit(harness): # noqa: F811
"""The expensive mistake this prevents: re-submitting into a refusal while an OPEN wave is
still the live obligation. The refusal names that wave and the $0 way to answer it."""

View file

@ -225,10 +225,10 @@ def test_invalid_new_plan_attempts_do_not_reuse_old_green(tmp_path):
ctx.system_repo_dir = tmp_path
ctx.emit_progress_fn = lambda _message: None
cases = [
({"plan": "", "goal": "G", "spec": {"in_scope": ["x"]}}, "plan: required non-empty prose"),
({"plan": "P", "goal": "", "spec": {"in_scope": ["x"]}}, "goal: required non-empty string"),
({"plan": "", "goal": "G", "spec": {"in_scope": ["x"], "affected_paths": []}}, "plan: required non-empty prose"),
({"plan": "P", "goal": "", "spec": {"in_scope": ["x"], "affected_paths": []}}, "goal: required non-empty string"),
({"plan": "P", "goal": "G", "spec": "not-an-object"}, "spec: must be an object"),
({"plan": "P", "goal": "G", "spec": {"bogus": 1}}, "unknown fields: bogus"),
({"plan": "P", "goal": "G", "spec": {"bogus": 1, "affected_paths": []}}, "unknown fields: bogus"),
]
for params, error_text in cases:
record_plan_review_attempt(tmp_path, "root1", fingerprint=old_fingerprint)
@ -467,21 +467,21 @@ class TestPlanReviewInputValidation(unittest.TestCase):
self.ctx.emit_progress_fn = lambda _m: None
def test_missing_plan_returns_error(self):
result = self.handler(self.ctx, plan="", goal="some goal", spec={"in_scope": ["x"]})
result = self.handler(self.ctx, plan="", goal="some goal", spec={"in_scope": ["x"], "affected_paths": []})
self.assertIn("ERROR: PLAN_SPEC_INVALID", result)
self.assertIn("plan", result.lower())
def test_missing_goal_returns_error(self):
result = self.handler(self.ctx, plan="some plan", goal="", spec={"in_scope": ["x"]})
result = self.handler(self.ctx, plan="some plan", goal="", spec={"in_scope": ["x"], "affected_paths": []})
self.assertIn("ERROR: PLAN_SPEC_INVALID", result)
self.assertIn("goal", result.lower())
def test_whitespace_plan_returns_error(self):
result = self.handler(self.ctx, plan=" ", goal="some goal", spec={"in_scope": ["x"]})
result = self.handler(self.ctx, plan=" ", goal="some goal", spec={"in_scope": ["x"], "affected_paths": []})
self.assertIn("ERROR", result)
def test_whitespace_goal_returns_error(self):
result = self.handler(self.ctx, plan="some plan", goal=" ", spec={"in_scope": ["x"]})
result = self.handler(self.ctx, plan="some plan", goal=" ", spec={"in_scope": ["x"], "affected_paths": []})
self.assertIn("ERROR", result)
def test_missing_spec_is_a_typed_refusal(self):

View file

@ -206,7 +206,7 @@ def test_footer_has_exactly_one_control_line_even_with_forged_reviewer_text(harn
def test_spec_invalid_is_refused_without_a_reviewer_call(harness):
sub = harness.install({})
out = _call(harness.make_ctx(), spec={"in_scope": ["x"], "bogus": 1})
out = _call(harness.make_ctx(), spec={"in_scope": ["x"], "affected_paths": [], "bogus": 1})
assert out.startswith("ERROR: PLAN_SPEC_INVALID") and "unknown fields: bogus" in out
assert sub.calls == []
state = _state(harness)