fix: report outcome gaps immediately and correct terminal recovery

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -1079,8 +1079,8 @@ def derive_goal_frontier_replan_obligation_from_summaries(
                     "priority": "P0",
                     "todo_ids": settlement_todo_ids,
                     "text": (
-                        "settle the exact completed Todos through no-follow-up or "
-                        "link each to a real runnable successor"
+                        "settle the completed scope using its original completion identity, or "
+                        "record a genuine dependency on the mutable successor; do not update terminal links"
                     ),
                 },
                 {
@@ -1100,8 +1100,10 @@ def derive_goal_frontier_replan_obligation_from_summaries(
             recommended_action=(
                 "settle the exact completed Todo(s) "
                 f"{settlement_summary} through loopx todo complete --no-follow-up "
-                "when no real work remains; otherwise link a real runnable successor. "
-                "do not invent a user gate or add a lifecycle-only filler Todo"
+                "with the original completion identity when no follow-up remains in that scope. "
+                "Independent open/deferred work remains active. For genuine dependent work, "
+                "record its real readiness dependency on the mutable successor, preserving gates; "
+                "never add links to a terminal Todo; do not invent a user gate or lifecycle-only filler Todo"
             ),
             extra_fields={
                 "resolution_mode": TODO_LIFECYCLE_SETTLEMENT_RESOLUTION_MODE,

loopx/control_plane/goals/goal_frontier/outcome_continuity.py

@@ -121,50 +121,60 @@ def latest_outcome_vision_checkpoint_from_status_payload(
         goal_id=goal_id,
         agent_id=agent_id,
     ):
-        checkpoint = run.get("vision_checkpoint")
-        if not isinstance(checkpoint, dict):
-            continue
-        checkpoint_agent_id = str(
-            checkpoint.get("agent_id") or run.get("agent_id") or ""
-        ).strip()
-        if agent_id and checkpoint_agent_id != agent_id:
-            continue
-        if not agent_id and checkpoint_agent_id:
-            continue
-        trigger_kinds = {
-            str(trigger.get("kind") or "").strip()
-            for trigger in (checkpoint.get("triggers") or [])
-            if isinstance(trigger, dict)
-            and str(trigger.get("kind") or "").strip()
-        }
-        if "material_delivery_outcome" not in trigger_kinds:
-            continue
-        result: dict[str, Any] = {
-            "schema_version": checkpoint.get("schema_version"),
-            "goal_id": goal_id,
-            "agent_id": checkpoint_agent_id or agent_id,
-            "required": checkpoint.get("required") is True,
-            "satisfied": checkpoint.get("satisfied") is True,
-            "decision": checkpoint.get("decision"),
-            "triggers": checkpoint.get("triggers")
-            if isinstance(checkpoint.get("triggers"), list)
-            else [],
+        checkpoint = outcome_vision_checkpoint_from_run(run, goal_id=goal_id, agent_id=agent_id)
+        if checkpoint is not None:
+            return checkpoint
+    return None
+
+
+def outcome_vision_checkpoint_from_run(
+    run: dict[str, Any], *, goal_id: str, agent_id: str | None,
+) -> dict[str, Any] | None:
+    """Project the same checkpoint facts for immediate and historical readback."""
+
+    checkpoint = run.get("vision_checkpoint")
+    if not isinstance(checkpoint, dict):
+        return None
+    checkpoint_agent_id = str(
+        checkpoint.get("agent_id") or run.get("agent_id") or ""
+    ).strip()
+    if agent_id and checkpoint_agent_id != agent_id:
+        return None
+    if not agent_id and checkpoint_agent_id:
+        return None
+    trigger_kinds = {
+        str(trigger.get("kind") or "").strip()
+        for trigger in (checkpoint.get("triggers") or [])
+        if isinstance(trigger, dict)
+        and str(trigger.get("kind") or "").strip()
+    }
+    if "material_delivery_outcome" not in trigger_kinds:
+        return None
+    result: dict[str, Any] = {
+        "schema_version": checkpoint.get("schema_version"),
+        "goal_id": goal_id,
+        "agent_id": checkpoint_agent_id or agent_id,
+        "required": checkpoint.get("required") is True,
+        "satisfied": checkpoint.get("satisfied") is True,
+        "decision": checkpoint.get("decision"),
+        "triggers": checkpoint.get("triggers")
+        if isinstance(checkpoint.get("triggers"), list)
+        else [],
+        "generated_at": run.get("generated_at"),
+    }
+    continuity_basis = checkpoint.get("continuity_basis")
+    if isinstance(continuity_basis, dict):
+        result["continuity_basis"] = dict(continuity_basis)
+    run_vision = run.get("agent_vision")
+    if isinstance(run_vision, dict):
+        result["qualification_agent_vision"] = {
+            "agent_id": run_vision.get("agent_id") or checkpoint_agent_id,
+            "state": run_vision.get("state"),
+            "vision_patch": _dict_field(run_vision, "vision_patch"),
+            "path_delta": _dict_field(run_vision, "path_delta"),
             "generated_at": run.get("generated_at"),
         }
-        continuity_basis = checkpoint.get("continuity_basis")
-        if isinstance(continuity_basis, dict):
-            result["continuity_basis"] = dict(continuity_basis)
-        run_vision = run.get("agent_vision")
-        if isinstance(run_vision, dict):
-            result["qualification_agent_vision"] = {
-                "agent_id": run_vision.get("agent_id") or checkpoint_agent_id,
-                "state": run_vision.get("state"),
-                "vision_patch": _dict_field(run_vision, "vision_patch"),
-                "path_delta": _dict_field(run_vision, "path_delta"),
-                "generated_at": run.get("generated_at"),
-            }
-        return result
-    return None
+    return result
 
 
 def acceptance_gaps_from_vision_checkpoint(

loopx/control_plane/work_items/interaction_contract.ts

@@ -242,7 +242,7 @@ export function selectPrimaryActionInstruction(value: unknown, mode: string): st
   if (mode === "autonomous_replan") {
     const lifecycle = object("lifecycle");
     if (requireBoolean(lifecycle.present, "lifecycle.present")) return text(lifecycle, "action") ||
-      "settle the exact completed Todo lifecycle or link a real runnable successor; do not create lifecycle-only filler";
+      "follow the exact Todo lifecycle recovery actions; preserve independent work and gates, and do not add links to terminal Todos or create lifecycle-only filler";
     const packet = object("replan_packet");
     if (requireBoolean(packet.settlement_only, "settlement_only")) return text(packet, "rule");
     if (requireBoolean(packet.has_required_outcome, "has_required_outcome") && requireBoolean(packet.has_successor_command, "has_successor_command")) {

loopx/control_plane/work_items/replan_settlement.ts

@@ -159,8 +159,13 @@ export function projectTodoLifecycleSettlementReentry(value: unknown): JsonObjec
     );
   });
   nextCliActions.push(
-    "otherwise link or create one real runnable successor for the exact " +
-      "completed Todo; do not create lifecycle-only filler",
+    "A terminal Todo cannot accept new successor links through complete or update. " +
+      "For genuine dependent work, author its dependency on the mutable successor " +
+      "(resume_when=todo_done:<completed-id>) only when that is its real readiness condition; " +
+      "preserve existing deferral, claims and gates. Independent work needs no artificial edge: " +
+      "when this completed scope has no remaining follow-up, use its original completion identity " +
+      "with --no-follow-up above. This closes only that scope, not the Goal or other Todos. " +
+      "Do not create lifecycle-only filler",
     `${commandPrefix} --format json quota should-run --goal-id ${shellArgument(goalId)}` +
       argumentSuffix(quotaScopedArgs),
   );

loopx/state_refresh.py

@@ -106,6 +106,10 @@
     prepare_vision_refresh,
 )
 from .control_plane.goals.goal_frontier import latest_agent_vision_from_runs
+from .control_plane.goals.goal_frontier.outcome_continuity import (
+    acceptance_gaps_from_outcome_checkpoint,
+    outcome_vision_checkpoint_from_run,
+)
 from .control_plane.goals.checkpoint_context_io import (
     checkpoint_commit_guard, commit_checkpoint_run, require_complete_checkpoint_index, inspect_checkpoint_replay,
 )
@@ -604,6 +608,10 @@ def render_state_refresh_markdown(payload: dict[str, Any]) -> str:
                 "- vision_checkpoint_required_resolution: "
                 f"{','.join(str(item) for item in required_resolution)}"
             )
+        for gap in vision_checkpoint.get("acceptance_gaps", []):
+            if isinstance(gap, dict):
+                failed = [name for name, passed in (gap.get("component_checks") or {}).items() if not passed]
+                lines.append(f"- outcome qualification gap: {', '.join(failed)}. {gap.get('resolution_hint', '')}")
         lines.extend(render_first_refresh_checkpoint_hint(payload))
 
     projection_gap = (
@@ -1356,6 +1364,17 @@ def read_explore_context(read_goal_id: str, read_agent_id: str | None):
             settlement_identity=settlement_identity,
             todo_fields=todo_fields,
         )
+        # Report qualification gaps in the write receipt, using the same owner
+        # as quota/history. A committed checkpoint is not outcome acceptance;
+        # this read-only diagnostic neither rejects nor repairs the write.
+        outcome_checkpoint = outcome_vision_checkpoint_from_run(
+            record, goal_id=safe_goal_id, agent_id=normalized_agent_id or None,
+        )
+        outcome_gaps = acceptance_gaps_from_outcome_checkpoint(
+            agent_vision or existing_agent_vision, outcome_checkpoint,
+        )
+        if outcome_gaps:
+            record["vision_checkpoint"] = {**vision_checkpoint, "acceptance_gaps": outcome_gaps}
         source_snapshot = (
             prior_writeback_run.get("todo_source")
             if checkpoint_supplement and prior_writeback_run

tests/control_plane/test_goal_frontier_replan_rules.py

@@ -447,15 +447,15 @@ def test_todo_succession_gap_prefers_exact_lifecycle_settlement() -> None:
     assert obligation is not None
     assert obligation["resolution_mode"] == "todo_lifecycle_settlement"
     assert obligation["guidance_actions"][0] == "record_no_followup"
-    assert obligation["todo_actions"][0] == {
+    action = obligation["todo_actions"][0]
+    assert "original completion identity" in action["text"]
+    assert "mutable successor" in action["text"]
+    assert "do not update terminal links" in action["text"]
+    assert {key: value for key, value in action.items() if key != "text"} == {
         "action": "settle",
         "role": "agent",
         "priority": "P0",
         "todo_ids": ["todo_111111111111", "todo_222222222222"],
-        "text": (
-            "settle the exact completed Todos through no-follow-up or link each "
-            "to a real runnable successor"
-        ),
     }
     assert "todo_111111111111, todo_222222222222" in obligation[
         "recommended_action"

tests/control_plane/test_refresh_state_replan_gate.py

@@ -1015,7 +1015,9 @@ def test_todo_lifecycle_rejection_projects_exact_direct_settlement() -> None:
     assert all("spend-slot" not in action for action in actions)
 
 
+@pytest.mark.parametrize("independent_deferred", [False, True])
 def test_todo_complete_then_refresh_cli_projects_direct_settlement(
+    independent_deferred: bool,
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],
 ) -> None:
@@ -1110,6 +1112,15 @@ def test_todo_complete_then_refresh_cli_projects_direct_settlement(
         for index in (1, 2)
     ]
 
+    independent = None
+    if independent_deferred:
+        independent = add_goal_todo(
+            registry_path=registry_path, goal_id=GOAL_ID, role="agent",
+            text="Evaluate independent work when its window opens.",
+            task_class="advancement_task", claimed_by=AGENT_ID,
+            status="deferred", resume_when="resume_at:2099-01-01T00:00:00Z",
+        )
+
     for todo in todos:
         assert cli_main(
             [
@@ -1183,6 +1194,16 @@ def test_todo_complete_then_refresh_cli_projects_direct_settlement(
     assert all("refresh-state" not in action for action in actions)
     assert all("spend-slot" not in action for action in actions)
 
+    if independent:
+        assert cli_main([
+            "--registry", str(registry_path), "--format", "json", "todo", "complete",
+            "--goal-id", GOAL_ID, "--todo-id", str(todos[0]["todo_id"]),
+            "--agent-id", AGENT_ID, "--successor-todo-id", str(independent["todo_id"]),
+        ]) == 1
+        rejected = json.loads(capsys.readouterr().out)
+        assert "operation id already names a different coordination request" in rejected["error"]
+        assert any("terminal Todo cannot accept new successor links" in action for action in actions)
+
     for action in lifecycle_actions:
         tokens = shlex.split(action)
         loopx_index = tokens.index("loopx")
@@ -1214,7 +1235,22 @@ def test_todo_complete_then_refresh_cli_projects_direct_settlement(
     terminal_quota = json.loads(capsys.readouterr().out)
     assert terminal_exit == 1
     assert terminal_quota["should_run"] is False
-    assert terminal_quota["effective_action"] == "terminal_no_followup"
+    if independent_deferred:
+        assert terminal_quota["effective_action"] != "terminal_no_followup"
+    else:
+        assert terminal_quota["effective_action"] == "terminal_no_followup"
+    if independent:
+        assert cli_main(["--registry", str(registry_path), "--format", "json",
+                         "todo", "list", "--goal-id", GOAL_ID,
+                         "--todo-id", str(independent["todo_id"])]) == 0
+        readback = json.loads(capsys.readouterr().out)
+        # Full readback must keep the independent wait, not merely avoid a
+        # terminal recommendation by replacing it with an invented blocker.
+        retained = readback["todo"]
+        assert retained["todo_id"] == independent["todo_id"]
+        assert retained["status"] == "deferred"
+        assert retained["resume_when"] == "resume_at:2099-01-01T00:00:00Z"
+        assert retained["claimed_by"] == AGENT_ID
 
 
 @pytest.mark.parametrize(

tests/control_plane/test_vision_budget_cli.py

@@ -232,16 +232,21 @@ def test_misplaced_delta_rejects_before_write_and_corrected_packet_roundtrips(tm
     assert json.loads(index.read_text().splitlines()[-1])["agent_vision"]["path_delta"] == delta
 
 
-@pytest.mark.parametrize("omitted", [None, "path_delta", "evidence_refs"])
-def test_authoring_example_material_closeout_has_no_missing_path_replan(tmp_path, omitted):
+@pytest.mark.parametrize("omitted", [None, "path_delta", "evidence_refs", "wait"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_authoring_example_material_closeout_has_no_missing_path_replan(tmp_path, monkeypatch, omitted, provider):
+    from canonical_authority_fixture import isolate_sqlite_runtime
+    from loopx.state_refresh import render_state_refresh_markdown
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
     from loopx.control_plane.work_items.progress_observation import replan_writeback_requirements
 
     source = Path(__file__).resolve().parents[2] / "examples/project/goal-vision-refresh-state-budget-smoke.py"
     spec = importlib.util.spec_from_file_location("vision_closeout_fixture", source)
     fixture = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(fixture)
     registry, runtime, _ = fixture.write_fixture(tmp_path)
-    import_fixture_authority(registry, fixture.GOAL_ID, provider="file")
+    import_fixture_authority(registry, fixture.GOAL_ID, provider=provider)
     contract = replan_writeback_requirements({
         "satisfying_semantic_outcomes": ["fresh_vision_path_outcome"],
     })["writeback_contract"]["vision_authoring"]
@@ -252,6 +257,8 @@ def test_authoring_example_material_closeout_has_no_missing_path_replan(tmp_path
         packet.pop("path_delta", None)
     elif omitted == "evidence_refs":
         packet.get("path_delta", {}).pop("evidence_refs", None)
+    if omitted == "wait":
+        packet["path_delta"]["outcome"] = "wait"
     path = tmp_path / "vision.json"
     fixture.write_json(path, packet)
     result = fixture.payload(fixture.run_cli(
@@ -263,9 +270,11 @@ def test_authoring_example_material_closeout_has_no_missing_path_replan(tmp_path
     gaps = quota["goal_frontier_projection"]["acceptance_gaps"]
     outcome_gaps = [g for g in gaps if g["kind"] == "vision_outcome_checkpoint_required"]
     assert bool(outcome_gaps) is (omitted is not None)
+    assert result["vision_checkpoint"].get("acceptance_gaps", []) == outcome_gaps
+    assert ("outcome qualification gap:" in render_state_refresh_markdown(result)) is (omitted is not None)
     if omitted:
         checks = outcome_gaps[0]["component_checks"]
-        assert checks["evidence_refs_present"] is False
+        assert checks["evidence_refs_present"] is (omitted == "wait")
         assert checks["path_outcome_valid"] is (omitted == "evidence_refs")
     else:
         persisted = result["agent_vision"]