test(control-plane): exercise native child admission with current fixtures

tests/control_plane/test_native_child_closeout_cli.py

@@ -39,11 +39,31 @@ def reject(*args: str, reason: str) -> None:
         "--replan-obligation-id", guard["autonomous_replan_obligation"]["obligation_id"])
     assert added["replan_transition"]["recorded"] is True
     binding = ("--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", TODO, "--turn-instance-id", TURN)
+    vision = tmp_path / "replan-vision.json"
+    vision.write_text(json.dumps({
+        "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed",
+        "vision_patch": {
+            "acceptance_summary": "Validate each source slice before dependent work proceeds.",
+            "advancement_policy": "as_needed",
+        },
+        "path_delta": {
+            "schema_version": "goal_path_delta_v0", "outcome": "replan",
+            "prior_assumption": "The long source chain needed a bounded review.",
+            "observed_reality": "The reviewed chain has an independent source validation slice.",
+            "retained": ["Validate the original source"],
+            "changed": ["Proceed with the independent validation slice"],
+            "evidence_refs": ["evidence:source-audit"],
+        },
+    }))
+    # Long-chain replans require an evidence-linked path; unchanged prose cannot ACK them.
     refreshed = call("refresh-state", *binding, "--classification", "bounded_replan_progress",
         "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress",
-        "--vision-unchanged-reason", "The source validation remains open; its independent successor changes the path.",
+        "--agent-vision-json", str(vision),
         "--no-global-sync", "--suppress-external-sinks")
     assert refreshed["settlement_progress"]["state"] == "spend_required"
+    ack = json.loads(Path(refreshed["json_path"]).read_text())["autonomous_replan_ack"]
+    assert ack["recorded"] is True
+    assert "fresh_vision_path_outcome" in ack["semantic_delta"]["outcomes"]
     reject(*base, "record", "--operation-id", "op-new", "--stage", "decision",
         "--operation", "spawn", "--outcome", "started", "--entrypoint-id", "generic_host",
         "--execute", reason="open, work-admitted")
@@ -90,6 +110,7 @@ def reject(*args: str, reason: str) -> None:
     assert index.read_bytes() == closed_index
     rows = [json.loads(line) for line in index.read_text().splitlines()]
     assert sum(row.get("classification") == "quota_slot_spent" for row in rows) == 1
+    assert sum(row.get("autonomous_replan_ack", {}).get("recorded") is True for row in rows) == 1
     assert call("todo", "list", "--goal-id", GOAL, "--todo-id", TODO)["todo"]["status"] == "open"
     assert call("todo", "list", "--goal-id", GOAL, "--todo-id", added["todo_id"])["todo"]["status"] == "open"
     assert original.get("todo_id") == TODO

tests/control_plane/test_native_child_replan_guard_cli.py

@@ -31,14 +31,22 @@ def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, tod
     project, runtime = tmp_path / "project", tmp_path / "runtime"
     project.mkdir()
     state = project / "ACTIVE_GOAL_STATE.md"
-    state.write_text("---\nstatus: active\n---\n\n# Synthetic Goal\n\n## Agent Todo\n" + (
+    # Fifteen selectable Todos trigger long-chain replan and require an explicit choice.
+    source_todos = (
         "\n- [ ] [P1] Validate the original source.\n"
         f"  <!-- loopx:todo todo_id={TODO} status=open task_class=advancement_task "
         f"claimed_by={AGENT} action_kind=validate validation_command=pytest "
         "continuation_policy=same_agent_non_delivery "
         "required_capabilities=shell%2Cfilesystem_read -->\n"
+        + "".join(
+            f"\n- [ ] [P1] Validate bounded slice {i}.\n"
+            f"  <!-- loopx:todo todo_id=todo_chain_{i:012d} status=open "
+            f"task_class=advancement_task claimed_by={AGENT} action_kind=validate -->\n"
+            for i in range(14)
+        )
         if todo_bound else ""
-    ))
+    )
+    state.write_text("---\nstatus: active\n---\n\n# Synthetic Goal\n\n## Agent Todo\n" + source_todos)
     index = runtime / "goals" / GOAL / "runs" / "index.jsonl"
     index.parent.mkdir(parents=True)
     evidence, report = index.parent / "source.json", index.parent / "source.md"
@@ -82,18 +90,24 @@ def _admitted_guard(call, todo_bound: bool) -> dict:
                   "--agent-id", AGENT, "--turn-instance-id", TURN)
     guard = call(*guard_args)
     if todo_bound:
-        # The planning recommendation has no settlement authority. An explicit
-        # choice may be retained during hard replan and bound only on reentry.
+        assert guard["decision"] == "autonomous_replan_required"
+        assert guard["interaction_contract"]["cli_channel"]["selection_required"] is True
+        # A recommendation cannot bind a multi-candidate Turn before choice.
         assert "settlement_identity" not in guard["heartbeat_receipt"]
+        assert guard["heartbeat_receipt"]["settlement_binding_owed"] is True
         rejected = call("native-child", "--goal-id", GOAL, "--agent-id", AGENT,
             "--turn-instance-id", TURN, "record", "--operation-id", "op-before-choice",
             "--stage", "decision", "--operation", "spawn", "--outcome", "started",
             "--entrypoint-id", "generic_host", "--execute", expected_code=1)
         assert "admitted" in rejected["error"]
         guard = call(*guard_args, "--todo-id", TODO)
         assert guard["normal_delivery_allowed"] is False
+        assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO
         assert guard["heartbeat_receipt"]["pending_action_selection"]["settlement_bound"] is True
         assert guard["retained_action_selection"]["disposition"] == "preserve_retained_todo"
+        replay = call(*guard_args)
+        assert replay["heartbeat_receipt"]["settlement_identity"] == guard["heartbeat_receipt"]["settlement_identity"]
+        assert replay["heartbeat_receipt"]["event_id"] == guard["heartbeat_receipt"]["event_id"]
     return guard
 
 

tests/control_plane/test_replan_successor_guard_reentry.py

@@ -234,16 +234,20 @@ def test_invalidated_successor_cannot_rebind_or_close_original_turn(
     original = _guard(call)
     identity = original["heartbeat_receipt"]["settlement_identity"]
     added = _add(call, identity["replan_obligation_id"])
-    prior_actions = _guard(call)["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    prior_guard = _guard(call)
+    prior_actions = prior_guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
     edit = {"wrong_owner": ["--claimed-by", "different-agent"],
             "deferred": ["--status", "deferred", "--resume-when", "resume_at:2099-01-01T00:00:00Z"],
             "unclaimed": ["--clear-claim"]}[invalidation]
     call("todo", "update", "--goal-id", GOAL, "--agent-id", AGENT,
          "--todo-id", added["todo_id"], *edit)
-    # Invalidation may leave the original duty open, which is a legal guard
-    # read. It must not grant successor settlement from a stale creation ACK.
-    guarded = _guard(call, expected_code=1 if later_vision and invalidation == "unclaimed" else 0)
+    # An unclaimed successor loses its original-lane proof. Fail closed while
+    # retaining the original receipt; a stale creation ACK cannot settle it.
+    guarded = _guard(call, expected_code=1 if invalidation == "unclaimed" else 0)
     assert guarded["heartbeat_receipt"]["settlement_identity"] == identity
+    if invalidation == "unclaimed":
+        assert "settlement identity conflicts" in guarded["reason"]
+        assert guarded["heartbeat_receipt"]["event_id"] == prior_guard["heartbeat_receipt"]["event_id"]
     assert guarded.get("selected_todo") is None
     assert (guarded.get("autonomous_replan_obligation") or {}).get("resolution_mode") != "receipt_bound_replan_settlement"
     assert (guarded.get("replan_action_packet") or {}).get("settlement_only") is not True