From 329fa5232f270742e719ad59b59a5502beb81979 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Wed, 16 Sep 2026 21:03:54 +0800 Subject: [PATCH] fix(replan): keep checkpoint replan ACKs visible past the run window The long Todo-chain rule is edge-triggered: an accepted ACK suppresses the obligation while its exact Todo frontier revision matches, and the rule re-arms only after a material revision. The ACK projection stopped scanning after AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW (20) material runs, so a checkpoint-bearing ACK was dropped while its checkpoint still described the current frontier. Frontier consumers that read that projection then saw no ACK and re-issued the long-chain obligation for an unchanged frontier, duplicating the periodic review cadence and reporting a "long chain" trigger that no longer matched the state. A checkpoint-bearing ACK is revision-gated by the typed evaluator, and the obligation-clearing path also requires the exact obligation id, so an old ACK cannot suppress or clear a materially changed frontier. Keep such an ACK visible for the whole scanned history, and keep the material-run window for legacy ACKs without checkpoints, which is also the window the separate periodic review rule owns. Observed on the loopx-meta review lane: 22 material runs after the newest accepted ACK made latest_autonomous_replan_ack_for_projection return None even though the ACK carried a long_todo_chain checkpoint, and the same frontier revision was acknowledged 12 hours apart under two different obligation ids. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../work_items/autonomous_replan_ack.py | 42 +++++++++- .../test_autonomous_replan_ack.py | 79 +++++++++++++++++++ 2 files changed, 118 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/work_items/autonomous_replan_ack.py b/loopx/control_plane/work_items/autonomous_replan_ack.py index cabe07fbf0..a89be06bc2 100644 --- a/loopx/control_plane/work_items/autonomous_replan_ack.py +++ b/loopx/control_plane/work_items/autonomous_replan_ack.py @@ -8,6 +8,31 @@ AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW = 20 +def ack_binds_trigger_checkpoints(ack: dict[str, Any] | None) -> bool: + """Return whether an ACK binds exact typed trigger revisions. + + A checkpoint-bearing ACK is revision-gated: the typed frontier evaluator + suppresses a trigger only when the ACK names that trigger's exact revision, + and the obligation-clearing path additionally requires the exact obligation + id. Such an ACK therefore cannot hide a materially changed frontier, which + is why it stays visible for the whole scanned history instead of expiring + with the material-run window. + """ + + semantic_delta = ack.get("semantic_delta") if isinstance(ack, dict) else None + if not isinstance(semantic_delta, dict): + return False + checkpoints = semantic_delta.get("trigger_checkpoints") + if not isinstance(checkpoints, list): + return False + return any( + isinstance(checkpoint, dict) + and str(checkpoint.get("kind") or "").strip() + and str(checkpoint.get("frontier_revision") or "").strip() + for checkpoint in checkpoints + ) + + def normalize_projected_autonomous_replan_ack( ack: dict[str, Any] | None, ) -> dict[str, Any] | None: @@ -275,15 +300,25 @@ def latest_autonomous_replan_ack_for_projection( *, neutral_classifications: set[str], ) -> dict[str, Any] | None: - """Return a recent durable replan ACK within the material review window.""" + """Return the newest durable replan ACK this lane may still use. + + A legacy ACK (one without typed trigger checkpoints) keeps the material-run + review window that produced it. A checkpoint-bearing ACK stays visible while + the scanned history still carries it: expiring it on run count alone re-armed + the long Todo-chain rule for an unchanged frontier, which is exactly the + level trigger the checkpoint contract exists to prevent. + """ material_run_count = 0 + window_expired = False for run in latest_runs or []: if not isinstance(run, dict): continue replan_ack = compact_autonomous_replan_ack(run) if replan_ack: - return replan_ack + if not window_expired or ack_binds_trigger_checkpoints(replan_ack): + return replan_ack + return None classification = str(run.get("classification") or "").strip() if not classification: continue @@ -293,5 +328,6 @@ def latest_autonomous_replan_ack_for_projection( continue material_run_count += 1 if material_run_count >= AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW: - return None + window_expired = True + return None return None diff --git a/tests/control_plane/test_autonomous_replan_ack.py b/tests/control_plane/test_autonomous_replan_ack.py index 5949131438..0b72857841 100644 --- a/tests/control_plane/test_autonomous_replan_ack.py +++ b/tests/control_plane/test_autonomous_replan_ack.py @@ -6,7 +6,9 @@ from loopx.control_plane.goals.goal_vision import normalize_goal_vision_packet from loopx.control_plane.work_items.autonomous_replan_ack import ( + AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW, compact_autonomous_replan_ack, + latest_autonomous_replan_ack_for_projection, ) from loopx.control_plane.work_items.progress_observation import ( semantic_delta_from_writeback, @@ -17,6 +19,83 @@ FRESH_VISION_PATH_OUTCOME = "fresh_vision_path_outcome" GOAL_ID = "goal-replan-3338" AGENT_ID = "codex-replan-agent" +NEUTRAL_CLASSIFICATIONS = {"quota_slot_spent"} +LONG_CHAIN_REVISION = "todo_frontier_revision_v0:0123456789abcdef01234567" +LONG_CHAIN_OBLIGATION_ID = "replan-0123456789abcdef" + + +def _material_run(index: int) -> dict[str, Any]: + return { + "classification": f"bounded_progress_{index}", + "generated_at": f"2026-08-19T12:{index:02d}:00+08:00", + "agent_id": AGENT_ID, + } + + +def _long_chain_ack_run(*, checkpoints: bool) -> dict[str, Any]: + semantic_delta: dict[str, Any] = { + "schema_version": "replan_semantic_delta_v0", + "accepted": True, + "outcomes": ["new_surface"], + "satisfying_outcomes": ["new_surface"], + "required_any_of": ["new_surface"], + "trigger_kinds": ["long_todo_chain"], + "obligation_id": LONG_CHAIN_OBLIGATION_ID, + } + if checkpoints: + semantic_delta["trigger_checkpoints"] = [ + {"kind": "long_todo_chain", "frontier_revision": LONG_CHAIN_REVISION} + ] + return { + "classification": "bounded_replan_progress", + "generated_at": "2026-08-19T13:00:00+08:00", + "agent_id": AGENT_ID, + "autonomous_replan_ack": { + "schema_version": "autonomous_replan_ack_v0", + "recorded": True, + "source": "refresh_state_semantic_delta", + "semantic_delta": semantic_delta, + }, + } + + +def _projection_ack(*, checkpoints: bool, newer_material_runs: int) -> dict[str, Any] | None: + """Newest-first history: fresh material runs, then the ACK, then older runs.""" + + latest_runs = [_material_run(index) for index in range(newer_material_runs)] + latest_runs.append(_long_chain_ack_run(checkpoints=checkpoints)) + latest_runs.extend(_material_run(index) for index in range(100, 103)) + return latest_autonomous_replan_ack_for_projection( + latest_runs, + neutral_classifications=NEUTRAL_CLASSIFICATIONS, + ) + + +def test_checkpoint_ack_stays_visible_beyond_the_material_run_window() -> None: + # The frontier rules compare the exact revision, so an old checkpoint ACK + # cannot suppress a changed frontier; it must not expire on run count alone. + ack = _projection_ack( + checkpoints=True, + newer_material_runs=AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW * 2, + ) + + assert ack is not None + assert ack["semantic_delta"]["obligation_id"] == LONG_CHAIN_OBLIGATION_ID + assert ack["semantic_delta"]["trigger_checkpoints"] == [ + {"kind": "long_todo_chain", "frontier_revision": LONG_CHAIN_REVISION} + ] + + +def test_legacy_ack_still_expires_with_the_material_run_window() -> None: + expired = _projection_ack( + checkpoints=False, + newer_material_runs=AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW, + ) + visible = _projection_ack(checkpoints=False, newer_material_runs=2) + + assert expired is None + assert visible is not None + assert visible["semantic_delta"]["obligation_id"] == LONG_CHAIN_OBLIGATION_ID def _validated_agent_vision(path_outcome: str) -> dict[str, Any]: