test(replan): qualify the explicit completed-Todo periodic fallback

examples/autonomous-replan-obligation-smoke.py

@@ -379,7 +379,7 @@ def assert_new_typed_surface_closes_exact_obligation() -> None:
         assert "autonomous_replan_obligation" not in after, after
 
 
-def assert_typed_monitor_and_periodic_thresholds_remain_explicit() -> None:
+def assert_typed_monitor_threshold_remains_explicit() -> None:
     with tempfile.TemporaryDirectory(prefix="loopx-monitor-replan-") as tmp:
         registry_path, runtime = write_fixture(Path(tmp), monitor_repeat_count=6)
         item = attention_item(run_cli("status", registry_path=registry_path, runtime=runtime))
@@ -400,14 +400,38 @@ def assert_typed_monitor_and_periodic_thresholds_remain_explicit() -> None:
             "dead_monitor_repeat"
         ), item
 
+
+def assert_explicit_completed_todo_periodic_fallback() -> None:
     with tempfile.TemporaryDirectory(prefix="loopx-periodic-replan-") as tmp:
         registry_path, runtime = write_fixture(Path(tmp), periodic_run_count=20)
+        # Twenty durable records have no settlement evidence. They cannot
+        # satisfy the default effective-Turn cadence. The retained 20-record
+        # fallback belongs only to the explicit completed-Todo unit.
+        item = attention_item(
+            run_cli("status", "--limit", "30", registry_path=registry_path, runtime=runtime)
+        )
+        assert "autonomous_replan_obligation" not in item["project_asset"], item
+        run_cli(
+            "configure-goal", "--goal-id", GOAL_ID,
+            "--execution-replan-after-todos", "5", "--execute",
+            registry_path=registry_path, runtime=runtime,
+        )
         item = attention_item(
             run_cli("status", "--limit", "30", registry_path=registry_path, runtime=runtime)
         )
         obligation = item["project_asset"]["autonomous_replan_obligation"]
         assert obligation["triggers"][0]["kind"] == "periodic_review_due", obligation
         assert obligation["triggers"][0]["run_count"] == 20, obligation
+        assert obligation["triggers"][0]["threshold"] == 20, obligation
+        run_cli(
+            "configure-goal", "--goal-id", GOAL_ID,
+            "--clear-execution-replan-after-todos", "--execute",
+            registry_path=registry_path, runtime=runtime,
+        )
+        item = attention_item(
+            run_cli("status", "--limit", "30", registry_path=registry_path, runtime=runtime)
+        )
+        assert "autonomous_replan_obligation" not in item["project_asset"], item
 
 
 def main() -> int:
@@ -419,7 +443,8 @@ def main() -> int:
     assert_typed_repeat_requires_two_equivalent_observations()
     assert_equivalent_observation_is_rejected_before_write()
     assert_new_typed_surface_closes_exact_obligation()
-    assert_typed_monitor_and_periodic_thresholds_remain_explicit()
+    assert_typed_monitor_threshold_remains_explicit()
+    assert_explicit_completed_todo_periodic_fallback()
     print("autonomous-replan-obligation-smoke ok")
     return 0
 

tests/control_plane/test_replan_history_provider.py

@@ -8,6 +8,9 @@
 from test_canonical_frontier_revision import _fixture
 from test_goal_amendment_proposal import _write_fixture, _stall_runs, _ack_run, GOAL_ID
 from loopx.control_plane.testing.canary_harness import run_json_cli_result
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.todos.active_state_todo_parser import parse_todo_source
+from loopx.control_plane.todos.todo_summary import structured_todo_item
 from loopx.status import active_state_todo_fields, autonomous_replan_obligation_from_runs
 
 
@@ -42,3 +45,66 @@ def test_complex_provider_snapshot_replan_and_quota_readback(tmp_path, monkeypat
     assert "typed_progress_repeat" in json.dumps(result)
     assert active_state_todo_fields(goal, runtime_root=paths["runtime"])["agent_todos"] == summary
     assert not paths["state_file"].exists()
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_explicit_completed_todo_fallback_retains_twenty_record_threshold(
+    tmp_path, monkeypatch, provider,
+):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    paths = _write_fixture(tmp_path, runs=[])
+    registry = json.loads(paths["registry"].read_text())
+    registry["goals"][0]["adapter"] = {"kind": "harness_self_improvement"}
+    paths["registry"].write_text(json.dumps(registry))
+    todos, _, sections = parse_todo_source(paths["state_file"].read_text())
+    projection = build_todo_runtime_shadow_projection(
+        goal_id=GOAL_ID,
+        todos=[structured_todo_item(row, role=role, source_section=sections[role], text_limit=None)
+               for role in ("user", "agent") for row in todos[role]],
+        handoff_mode="soft_claim",
+    )
+    initialize_canonical_authority(paths["runtime"], GOAL_ID, projection,
+                                   state_path=paths["state_file"], provider=provider)
+    paths["state_file"].unlink()
+    index = paths["runtime"] / "goals" / GOAL_ID / "runs" / "index.jsonl"
+    index.parent.mkdir(parents=True, exist_ok=True)
+
+    def call(*args):
+        code, result = run_json_cli_result(*args, registry_path=paths["registry"])
+        assert code == 0, result
+        return result
+
+    def obligation():
+        status = call("status", "--limit", "30", "--scan-path", str(paths["project"]))
+        return next((obligation for row in status["attention_queue"]["items"]
+                     if row["goal_id"] == GOAL_ID
+                     if (obligation := row.get("autonomous_replan_obligations_by_agent", {})
+                         .get("agent-a")) is not None), None)
+
+    def history(count, *, retry=False, peer=False):
+        index.write_text("".join(json.dumps({
+            "goal_id": GOAL_ID, "agent_id": "agent-b" if peer else "agent-a",
+            "generated_at": f"2026-09-01T03:{n:02d}:00Z",
+            "turn_instance_id": "one-retried-turn" if retry else f"turn-{n}",
+            "classification": "bounded_iteration", "delivery_outcome": "outcome_progress",
+        }) + "\n" for n in range(count)))
+
+    history(20)
+    assert obligation() is None, "unsettled history is not effective-Turn evidence"
+    call("configure-goal", "--goal-id", GOAL_ID,
+         "--execution-replan-after-todos", "5", "--execute")
+    history(19)
+    assert obligation() is None
+    history(20, retry=True)
+    assert obligation() is None, "retries cannot meet the twenty-Turn fallback"
+    history(20, peer=True)
+    assert obligation() is None, "a peer cannot trigger the current Agent's review"
+    history(20)
+    trigger = obligation()["triggers"][0]
+    assert trigger["kind"] == "periodic_review_due"
+    assert trigger["run_count"] == trigger["threshold"] == 20
+    assert trigger["agent_id"] == "agent-a"
+    call("configure-goal", "--goal-id", GOAL_ID,
+         "--clear-execution-replan-after-todos", "--execute")
+    assert obligation() is None
+    assert not paths["state_file"].exists(), "readback must retain canonical authority"