fix(quota): bind watch ACKs to causal monitor frontier (#3037)

Author: huangruiteng

loopx/control_plane/goals/goal_frontier/__init__.py

@@ -18,7 +18,10 @@
     MONITOR_NO_CHANGE_STREAK_THRESHOLD,
     build_autonomous_replan_obligation_payload,
 )
-from ...work_items.repair_delta import repair_delta_kinds_have_frontier_delta
+from ...work_items.repair_delta import (
+    repair_delta_kinds_have_frontier_delta,
+    validate_repair_delta_claims,
+)
 from ..goal_vision_policy import goal_vision_repeats_advancement_until_closed
 from ..goal_vision_state import (
     goal_vision_state_is_closed,
@@ -243,6 +246,8 @@ def _watch_lane_ack_covers_dead_monitor_repeat(
     *,
     replan_obligation: dict[str, Any] | None,
     acceptance_gaps: list[dict[str, Any]],
+    agent_todo_summary: dict[str, Any] | None,
+    agent_id: str | None,
 ) -> bool:
     """Keep an as-needed watch ACK valid across unchanged heartbeat receipts."""
 
@@ -262,15 +267,66 @@ def _watch_lane_ack_covers_dead_monitor_repeat(
         delta_kind="watch_lane_continuation",
     ):
         return False
+    if not isinstance(ack, dict):
+        return False
     if not acceptance_gaps or any(
         gap.get("kind") != "vision_acceptance_gap"
         for gap in acceptance_gaps
     ):
         return False
-    return not any(
+    if any(
         goal_vision_repeats_advancement_until_closed(gap.get("advancement_policy"))
         for gap in acceptance_gaps
+    ):
+        return False
+    frontier_identity = str(
+        replan_obligation.get("frontier_identity")
+        if isinstance(replan_obligation, dict)
+        else ""
+    ).strip()
+    if (
+        not frontier_identity
+        or str(ack.get("frontier_identity") or "").strip() != frontier_identity
+    ):
+        return False
+    delta_contract = ack.get("delta_contract")
+    evidence_todo_ids = {
+        str(todo_id).strip()
+        for evidence in (
+            delta_contract.get("auto_evidence") or []
+            if isinstance(delta_contract, dict)
+            else []
+        )
+        if isinstance(evidence, dict)
+        and evidence.get("kind") == "watch_lane_continuation"
+        for todo_id in (evidence.get("todo_ids") or [])
+        if str(todo_id or "").strip()
+    }
+    if not evidence_todo_ids:
+        return False
+    summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {}
+    exact_watch_items_by_id: dict[str, dict[str, Any]] = {}
+    for lane in (
+        "items",
+        "current_agent_claimed_monitor_items",
+        "claimed_monitor_open_items",
+        "monitor_open_items",
+    ):
+        for item in summary.get(lane) or []:
+            if not isinstance(item, dict):
+                continue
+            todo_id = str(item.get("todo_id") or "").strip()
+            if todo_id in evidence_todo_ids:
+                exact_watch_items_by_id[todo_id] = item
+    accepted, _, _ = validate_repair_delta_claims(
+        ["watch_lane_continuation"],
+        agent_todo_summary={"items": list(exact_watch_items_by_id.values())},
+        agent_id=agent_id,
+        advancement_policy="as_needed",
+        next_action_changed=False,
+        vision_patch_written=False,
     )
+    return "watch_lane_continuation" in accepted
 
 
 def _watch_lane_ack_covers_blocked_successor_repeat(
@@ -1670,6 +1726,8 @@ def build_goal_frontier_projection_context_from_status(
             effective_replan_ack,
             replan_obligation=replan_obligation,
             acceptance_gaps=acceptance_gaps,
+            agent_todo_summary=agent_todo_summary,
+            agent_id=agent_id,
         )
     )
     watch_lane_ack_covers_blocked_successor_repeat = (
@@ -1692,6 +1750,7 @@ def build_goal_frontier_projection_context_from_status(
                 replan_obligation,
             )
             or watch_lane_ack_covers_blocked_successor_repeat
+            or watch_lane_ack_covers_dead_monitor_repeat
         )
         and (
             not acceptance_gaps

loopx/control_plane/goals/goal_frontier/outcome_continuity.py

@@ -5,7 +5,10 @@
 
 from ...agents.agent_scope import agent_scope_item_claimed_by
 from ...work_items.repair_delta import repair_delta_kinds_have_frontier_delta
-from ..goal_vision_policy import COMPLETED_TODO_CHAIN_REPLAN_THRESHOLD
+from ..goal_vision_policy import (
+    COMPLETED_TODO_CHAIN_REPLAN_THRESHOLD,
+    goal_vision_repeats_advancement_until_closed,
+)
 from ..goal_vision_state import goal_vision_state_is_closed
 
 VISION_OUTCOME_CHECKPOINT_REQUIRED_TRIGGER = "vision_outcome_checkpoint_required"
@@ -360,8 +363,40 @@ def _checkpoint_covers_completed_todo(
         for value in (checkpoint.get("repair_delta_kinds") or [])
         if str(value or "").strip()
     }
+    if repair_delta_kinds & set(REPEAT_VISION_REPLAN_SATISFYING_DELTA_KINDS):
+        return True
+    qualification_vision_value = checkpoint.get("qualification_agent_vision")
+    qualification_vision = (
+        qualification_vision_value
+        if isinstance(qualification_vision_value, dict)
+        else agent_vision
+    )
+    qualification_patch = _dict_field(qualification_vision, "vision_patch")
+    if goal_vision_repeats_advancement_until_closed(
+        qualification_patch.get("advancement_policy")
+    ):
+        return False
+    path_delta = _dict_field(qualification_vision, "path_delta")
+    evidence_refs = [
+        value
+        for value in (path_delta.get("evidence_refs") or [])
+        if _compact_text(value, limit=140)
+    ]
+    fresh_vision_patch = bool(
+        checkpoint.get("decision") == "patched"
+        and checkpoint.get("generated_at")
+        and checkpoint.get("generated_at")
+        == qualification_vision.get("generated_at")
+    )
     return bool(
-        repair_delta_kinds & set(REPEAT_VISION_REPLAN_SATISFYING_DELTA_KINDS)
+        fresh_vision_patch
+        and _compact_text(qualification_patch.get("acceptance_summary"), limit=420)
+        and _compact_text(path_delta.get("outcome"), limit=32) == "replan"
+        and evidence_refs
+        and {
+            "goal_vision_patch",
+            "watch_lane_continuation",
+        }.issubset(repair_delta_kinds)
     )
 
 

loopx/control_plane/work_items/autonomous_replan_ack.py

@@ -22,10 +22,12 @@ def compact_autonomous_replan_ack(run: dict[str, Any] | None) -> dict[str, Any]
     if not isinstance(run, dict) or not autonomous_replan_ack_recorded(run):
         return None
     ack = run.get("autonomous_replan_ack")
+    if not isinstance(ack, dict):
+        return None
     delta_contract = ack.get("delta_contract") if isinstance(ack, dict) else {}
     if not isinstance(delta_contract, dict):
         return None
-    compact_delta = {
+    compact_delta: dict[str, Any] = {
         "schema_version": delta_contract.get("schema_version"),
         "delta_present": bool(delta_contract.get("delta_present")),
         "delta_kinds": [
@@ -34,6 +36,24 @@ def compact_autonomous_replan_ack(run: dict[str, Any] | None) -> dict[str, Any]
             if str(item or "").strip()
         ],
     }
+    watch_evidence: list[dict[str, Any]] = []
+    for item in delta_contract.get("auto_evidence") or []:
+        if not isinstance(item, dict) or item.get("kind") != "watch_lane_continuation":
+            continue
+        todo_ids = [
+            str(todo_id).strip()
+            for todo_id in (item.get("todo_ids") or [])
+            if str(todo_id or "").strip()
+        ][:4]
+        if todo_ids:
+            watch_evidence.append(
+                {
+                    "kind": "watch_lane_continuation",
+                    "todo_ids": todo_ids,
+                }
+            )
+    if watch_evidence:
+        compact_delta["auto_evidence"] = watch_evidence
     result = {
         "schema_version": ack.get("schema_version"),
         "recorded": True,
@@ -56,6 +76,33 @@ def latest_blocked_successor_frontier_identity(
     latest_runs: list[dict[str, Any]] | None,
     *,
     agent_id: str | None = None,
+) -> str | None:
+    return _latest_monitor_replan_frontier_identity(
+        latest_runs,
+        agent_id=agent_id,
+        include_generic_watch=False,
+    )
+
+
+def latest_monitor_replan_frontier_identity(
+    latest_runs: list[dict[str, Any]] | None,
+    *,
+    agent_id: str | None = None,
+) -> str | None:
+    """Return the latest monitor identity that a durable replan ACK must bind."""
+
+    return _latest_monitor_replan_frontier_identity(
+        latest_runs,
+        agent_id=agent_id,
+        include_generic_watch=True,
+    )
+
+
+def _latest_monitor_replan_frontier_identity(
+    latest_runs: list[dict[str, Any]] | None,
+    *,
+    agent_id: str | None,
+    include_generic_watch: bool,
 ) -> str | None:
     normalized_agent_id = str(agent_id or "").strip()
     for run in latest_runs or []:
@@ -84,7 +131,14 @@ def latest_blocked_successor_frontier_identity(
         if target.get("monitor_mode") != (
             "blocked_successor_wait_without_material_transition"
         ):
-            continue
+            if not (
+                include_generic_watch
+                and target.get("monitor_mode")
+                == "monitor_quiet_until_material_transition"
+            ):
+                continue
+            target_identity = str(target.get("target_id") or "").strip()
+            return target_identity or None
         frontier_identity = str(target.get("frontier_identity") or "").strip()
         return frontier_identity or None
     return None

loopx/control_plane/work_items/autonomous_replan_obligation.py

@@ -604,6 +604,10 @@ def build_autonomous_replan_obligation(
         extra_fields["frontier_identity"] = blocked_successor_evidence.get(
             "frontier_identity"
         )
+    elif dead_monitor_evidence and dead_monitor_evidence.get("monitor_target_id"):
+        extra_fields["frontier_identity"] = dead_monitor_evidence.get(
+            "monitor_target_id"
+        )
 
     result = build_autonomous_replan_obligation_payload(
         schema_version=autonomous_replan_schema_version,

loopx/state_refresh.py

@@ -32,7 +32,7 @@
     validate_repair_delta_claims,
 )
 from .control_plane.work_items.autonomous_replan_ack import (
-    latest_blocked_successor_frontier_identity,
+    latest_monitor_replan_frontier_identity,
 )
 from .control_plane.runtime.shared_runtime_refresh_projection import (
     build_shared_runtime_projection,
@@ -1163,7 +1163,7 @@ def refresh_state_run(
         )
         if autonomous_replan_recorded:
             autonomous_replan_frontier_identity = (
-                latest_blocked_successor_frontier_identity(
+                latest_monitor_replan_frontier_identity(
                     newest_first_runs,
                     agent_id=normalized_agent_id,
                 )

skills/loopx-self-repair/references/repair-patterns.md

@@ -63,6 +63,7 @@ teaches a reusable control-plane lesson.
 | `stale_active_next_priority_override` | A lower-priority todo named by durable `Next Action` becomes `agent_lane_next_action` even though the same peer has a higher-priority capability-runnable todo, commonly after a deferred P0 resumes or another peer refreshes shared state. | scoped quota payload, `capability_gate.runnable_candidates`, `agent_lane_next_action`, active-state `Next Action`, todo priority/resume/claim metadata, recent peer writes. | The active-next hint was ranked before todo priority, so stale steering text acted as a global override instead of a same-priority tie-breaker. | Preserve peer claim/profile routing, then rank todo priority before active-next preference; keep active-next as a same-priority tie-breaker. Cover a resume-ready P0 plus lower-priority active-next and concurrent other-peer work. |
 | `status_projection_history_neutral_gap` | `quota should-run` reports no user action or quiet monitor behavior, but `status` / `diagnose --limit N` misreports a stale controller/user gate or older connected-without-run state after recent monitor/spend/readiness events. | quota payload, diagnose/status payload, recent history classifications, neutral-run classification sets, UI `--limit`, latest meaningful state run. | Status/history/quota disagree on which runs are status-neutral, or a short display limit is reused as the control-plane reasoning window. | Share the neutral classification contract across history/status/quota and reason over a wider internal state window before trimming displayed runs; add a regression with consecutive neutral runs before a meaningful state transition. |
 | `monitor_replan_noop_loop` | Recent runs are monitor-only, replan, or self-repair-adjacent, but the same monitor/action recommendation keeps returning and no runnable todo, blocker, successor, supersede, user gate, capability change, or monitor-target change appears. A sharper form is `goal_frontier_projection` showing a monitor-only lane with zero current, unclaimed, or other-agent advancement frontier. | quota payload, `goal_frontier_projection`, interaction contract, todo summaries, recent history classifications, latest replan ACK, monitor target, active state todos. | Self-repair/replan was recorded as activity but did not change the machine-visible work frontier, or generic vision/no-follow-up deltas silently converted an empty active-goal frontier into a monitor wait. | Require a repair delta contract: classify no-delta repairs as no-ops, then add a blocker, supersede stale monitor work, create a successor runnable todo, or record an explicit `watch_lane_continuation` with expiry. Generic vision, no-follow-up, and next-action deltas must not suppress empty-frontier replan by themselves. |
+| `watch_lane_ack_causality_gap` | An accepted, expiring `watch_lane_continuation` clears one replan, but two later quiet heartbeat receipts reopen the same dead-monitor obligation before the watch is due; a stale ACK may also suppress a changed or expired monitor. | dead-monitor target id, compact ACK frontier identity and watch Todo ids, current scoped monitor target/cadence/due/expiry, agent vision advancement policy, and completed-Todo outcome checkpoint. | Dead-monitor obligations had no causal frontier identity, compact ACKs dropped their exact watch evidence, or a completed-Todo checkpoint treated every watch-only replan as invalid regardless of an as-needed vision. | Bind the obligation and ACK to the same monitor target, retain bounded exact watch Todo ids, revalidate that watch and expiry on every suppression, and allow an evidence-linked `as_needed` watch replan to close the completed-Todo checkpoint. Missing ACK/evidence, target drift, expiry, and `repeat_until_closed` remain replan triggers. |
 | `vision_replan_writeback_gap` | The user or agent identifies a missing outcome, acceptance condition, or product bottleneck, but later `quota should-run` shows no `goal_frontier_projection.acceptance_gaps[]`, no per-agent `agent_vision` / `vision_checkpoint_v0`, and no replan obligation; the agent keeps following older runnable todos or monitor waits. | recent user correction, `loopx history --goal-id`, latest runs' `agent_vision` and `vision_checkpoint`, quota `goal_frontier_projection.acceptance_gaps`, active-state todos, bad-case report or incident note. | The insight stayed in chat, prose, or a report instead of being converted into a bounded `goal_vision_replan_contract_v0` packet, per-agent checkpoint decision, or concrete successor todo. | Write `loopx refresh-state --agent-id <agent-id> --vision-summary ... --vision-replan-trigger ...` or `--agent-vision-json <packet>` with a compact `replan_trigger_summary`; use `--vision-unchanged-reason` when the current per-agent vision still applies; and add/link a successor todo when the next executable step is known. Then rerun quota so the vision gap or `vision_checkpoint_missing` can become `autonomous_replan_required` before local quiet/wait states. |
 | `active_goal_closed_vision_succession_gap` | A bounded stage is correctly marked `vision_closed`, but the registry goal remains active; quota shows no acceptance gap or exposes one without requiring replan, then continues ordinary or resume-ready deferred work or monitor quiet without establishing the next vision. | latest per-agent `agent_vision`, registry goal status, quota `goal_frontier_projection`, remaining open/deferred advancement todos, terminal lane rationale. | Closed vision was treated as terminal, or a generic ready-deferred shortcut returned before the successor-vision gap was evaluated, so closing one accepted stage erased or demoted the long-horizon vision frontier. | Treat `vision_closed` as stage closure: for an active registry goal, project `vision_successor_required` and run replan even when ordinary or resume-ready deferred advancement already exists. Explicit user/blocking handoff gates still take precedence. Only `retired`, `superseded`, or `no_followup` may terminate the lane without another vision. Cover active/inactive goal and runnable/deferred/empty frontier cases. |
 | `scheduler_liveness_backoff_gap` | `quota should-run` says `should_run=false`, `quiet_noop_allowed=true`, no spend, or `automation_action=keep_active_quiet`, but the host keeps polling at the bootstrap cadence for hours; Codex CLI TUI or Claude Code `/loop` also keeps repeating unchanged checks. | quota payload `automation_liveness`, `interaction_contract`, `heartbeat_recommendation`, host automation cadence, CLI/Claude loop logs or statusline, recent no-spend poll history. | LoopX separated execution permission from user notification, but did not expose a machine-readable next-wakeup/backoff/final-check/exit policy for host runtimes. | Add or repair `quota should-run.scheduler_hint`; host runtimes must apply Codex App cadence backoff and CLI/Claude unchanged-poll final quota/replan check before self-stop, without quota spend. Prompt/docs should treat fixed 3-minute cadence as bootstrap only. |