fix: preserve intentional watches in replan history

docs/quota-allocation.md

@@ -1177,6 +1177,16 @@ receipts have no concrete executed-monitor identity and do not count toward
 this threshold; distinct heartbeat turn ids are idempotency evidence, not
 monitor execution evidence.
 
+A current open Monitor explicitly registered with `watch_only=true` represents
+an intentional continuing watch. Matching unchanged history does not trigger
+`dead_monitor_repeat`: the Todo id (when recorded), target key and agent scope
+must match, and an expiring watch must still be active at the latest observation.
+The same watch is excluded from current `monitor_no_change_streak` selection.
+Missing or false watch state keeps ordinary finite-monitor detection; another
+Todo, target or peer cannot discharge it. Independent typed-progress,
+blocked-successor and periodic work reviews remain effective. This does not
+change cadence, expiry, observation counters, receipts or spend rules.
+
 The current Todo `consecutive_no_change` detector shares the Monitor lifetime
 rule with scheduling. At one explicit runtime observation, `expires_at` at or
 before that instant excludes that Monitor from `monitor_no_change_streak`.

loopx/control_plane/work_items/project_asset.py

@@ -319,6 +319,9 @@ def attach_active_state_project_asset_fields(
         else None
     )
     if replan_obligation is None and autonomous_replan_obligation_from_runs is not None:
+        shared_todos = item.get("agent_todos")
+        monitor_todos = {"monitor_open_items":
+            shared_todos.get("monitor_open_items", []) if isinstance(shared_todos, dict) else []}
         agent_ids = sorted(
             {
                 agent_id
@@ -333,9 +336,10 @@ def attach_active_state_project_asset_fields(
             if (
                 obligation := autonomous_replan_obligation_from_runs(
                     latest_runs or [],
-                    # The shared todo summary is goal-scoped here. Quota derives
-                    # todo-based replans later from its agent-filtered summary.
-                    agent_todos=None,
+                    # The TS history owner matches monitor intent by Todo,
+                    # target and lane. Share those facts without importing
+                    # goal-scoped advancement/resume decisions into this lane.
+                    agent_todos=monitor_todos,
                     agent_id=agent_id,
                 )
             )

loopx/control_plane/work_items/replan_history.ts

@@ -46,6 +46,7 @@ interface MonitorTodo {
   claim: string | null;
   publicClaim: string | null;
   target: string;
+  watchOnly: boolean;
   count: number;
   due: number | null;
   expiry: number | null;
@@ -176,6 +177,7 @@ function decode(value: unknown): Request {
         id: optionalNonEmptyString(row.id, "monitor.id"), claim: optionalNonEmptyString(row.claim, "claim"),
         publicClaim: optionalNonEmptyString(row.public_claim, "public_claim"),
         target: text(row.target, "monitor.target"),
+        watchOnly: row.watch_only === undefined ? false : requireBoolean(row.watch_only, "watch_only"),
         count: integer(row.no_change_count, "no_change_count", Number.MIN_SAFE_INTEGER),
         due: time(row.due_at, "due_at"), expiry: time(row.expires_at, "expires_at"),
       })),
@@ -323,6 +325,7 @@ function monitorTrigger(runs: readonly Run[], request: Request): Trigger | null
     };
   }
   const repeated = signals.filter(run => (run.monitor.todo || run.monitor.key) &&
+    !isIntentionalWatch(run, request, signals[0]!.observedAt) &&
     ["due_monitor_observed_without_material_transition", "external_monitor_observed_without_material_transition"]
       .includes(run.monitor.mode!)).slice(0, request.monitor);
   const target = sole(repeated.map(run => run.monitor.target));
@@ -333,9 +336,19 @@ function monitorTrigger(runs: readonly Run[], request: Request): Trigger | null
     latest_generated_at: repeated[0]!.generatedAt, agent_id: sole(repeated.map(run => run.agent)),
   };
 }
+function isIntentionalWatch(run: Run, request: Request, observedAt: number | null): boolean {
+  const agent = run.monitorAgent ?? request.monitorAgent;
+  if (!agent) return false;
+  // Poll target_id is a digest; target_key (or a legacy plain target) names
+  // the canonical target. A key match cannot override a conflicting Todo id.
+  const target = run.monitor.key ?? run.monitor.target;
+  return request.todos.monitors.some(row => row.watchOnly && (!row.claim || row.claim === agent) &&
+    row.target === target && (run.monitor.todo ? row.id === run.monitor.todo : !!run.monitor.key) &&
+    (row.expiry === null || observedAt !== null && monitorIsActive({expiresAt: row.expiry}, observedAt)));
+}
 function monitorStreak(request: Request): Trigger | null {
   const frontier = request.operation === "frontier_monitor_streak";
-  const stalled = request.todos.monitors.filter(row => row.count >= request.streak &&
+  const stalled = request.todos.monitors.filter(row => !row.watchOnly && row.count >= request.streak &&
     (request.observedAt === null || monitorIsActive({expiresAt: row.expiry}, request.observedAt)))
     .sort((a, b) => b.count - a.count || (frontier ? authorityUnicodeCompare(a.target, b.target) : 0))[0];
   if (!stalled || !frontier && request.todos.advancements.some(row => row.status === "open" &&

loopx/control_plane/work_items/replan_history_codec.py

@@ -16,7 +16,10 @@
 from ..effect_runtime import MAX_REQUEST_BYTES, EffectRuntimeRejected, effect_runtime_result
 from ..runtime.time import now_utc, parse_timestamp
 from ..goals.goal_vision_policy import normalize_effective_turn_replan_threshold
-from ..todos.contract import normalize_todo_claimed_by, normalize_todo_id, normalize_todo_id_list
+from ..todos.contract import (
+    normalize_todo_claimed_by, normalize_todo_id, normalize_todo_id_list,
+    normalize_todo_watch_only,
+)
 from ..todos.resume_planning import build_todo_resume_planning_request
 
 REPLAN_HISTORY_NEUTRAL_CLASSIFICATIONS = {
@@ -119,6 +122,7 @@ def _todo_facts(value: Any, *, include_resume: bool, frontier: bool = False) ->
             "claim": normalize_todo_claimed_by(item.get("claimed_by")),
             "public_claim": str(item.get("claimed_by") or "").strip() or None,
             "target": str(item.get("target_key") or item.get("todo_id") or "monitor").strip(),
+            "watch_only": normalize_todo_watch_only(item.get("watch_only")) is True,
             "no_change_count": count,
             "due_at": _timestamp(item.get("next_due_at")),
             "expires_at": _timestamp(item.get("expires_at")),

tests/control_plane/test_replan_history_policy.py

@@ -134,6 +134,19 @@ def test_six_distinct_polls_retain_dead_monitor_contract() -> None:
     assert result["dead_monitor_detector"]["run_count"] == 6
 
 
+def test_watch_only_state_reaches_the_history_policy_owner() -> None:
+    rows = [monitor(n, turn=f"poll-{n}") for n in range(6, 0, -1)]
+    for watch_only in (True, "true", False, "false", None):
+        todos = {"monitor_open_items": [{"todo_id": "todo_monitor_fixture",
+                 "target_key": "target-fixture", "claimed_by": AGENT,
+                 "watch_only": watch_only}]}
+        result = autonomous_replan_obligation_from_runs(rows, agent_todos=todos, agent_id=AGENT)
+        if watch_only in (True, "true"):
+            assert result is None
+        else:
+            assert result["triggers"][0]["kind"] == "dead_monitor_repeat"
+
+
 def test_genuinely_unknown_work_still_breaks_progress_streak() -> None:
     rows = [{**run(3), "progress_observation": observation()}, run(2),
             {**run(1), "progress_observation": observation()}]

tests/control_plane/test_replan_history_transport.py

@@ -1,11 +1,41 @@
 """Production CLI settlement with a synthetic long-lived history, no live state."""
 import json
+import pytest
 
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, TODO_ID, _run_cli, _spend_run_count, _write_fixture,
+    DUE_MONITOR_TODO_ID, _append_newly_due_monitor, _initialize_fixture_authority,
 )
 
 
+@pytest.mark.parametrize("watch_only", [True, False])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_cli_history_preserves_registered_watch_intent(tmp_path, watch_only, provider):
+    project, runtime, registry = _write_fixture(tmp_path)
+    _append_newly_due_monitor(project, watch_only=watch_only)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl"
+    index.parent.mkdir(parents=True, exist_ok=True)
+    original = "".join(json.dumps({
+        "agent_id": AGENT_ID, "classification": "quota_monitor_poll",
+        "generated_at": f"2026-01-01T00:00:{n:02d}Z", "turn_instance_id": f"poll-{n}",
+        "todo_id": DUE_MONITOR_TODO_ID, "target_key": "due-monitor-fixture",
+        "monitor_target": {"target_id": "target-digest", "agent_id": AGENT_ID,
+            "monitor_mode": "due_monitor_observed_without_material_transition"},
+        "monitor_event": {"material_change": False},
+    }) + "\n" for n in range(1, 7)).encode()
+    index.write_bytes(original)
+    code, guard = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", "inspect-watch",
+        "--available-capability", "network", "--available-capability", "external_evidence_poll",
+        cwd=project)
+    assert code == 0, guard
+    kinds = [row["kind"] for row in (guard.get("autonomous_replan_obligation") or {}).get("triggers", [])]
+    assert ("dead_monitor_repeat" in kinds) is (not watch_only)
+    assert _spend_run_count(runtime) == 0
+    assert index.read_bytes().startswith(original)
+
+
 def test_large_history_refresh_replay_and_spend_once(tmp_path):
     project, runtime, registry = _write_fixture(tmp_path, provider="sqlite")
     index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl"

tests/control_plane_ts/replan_history.test.ts

@@ -239,6 +239,57 @@ function monitorTodo(patch: JsonObject = {}): JsonObject {
   return { id: "todo_watch", claim: "worker-a", public_claim: "worker-a", target: "watch",
     no_change_count: 5, due_at: null, expires_at: null, ...patch };
 }
+test("intentional watches do not become dead monitors after unchanged observations", () => {
+  for (const watch_only of [true, false, undefined]) {
+    const todos = {monitors: [monitorTodo({watch_only})], advancements: [], resume: null};
+    const before = structuredClone(todos);
+    assert.equal(trigger(many(6, poll), {todos})?.kind ?? null,
+      watch_only === true ? null : "dead_monitor_repeat");
+    assert.deepEqual(todos, before);
+  }
+});
+
+test("watch exemptions require the observed Todo, target and lane", () => {
+  for (const patch of [{id: "other"}, {target: "other"}, {claim: "peer"}, {expires_at: 6}]) {
+    const todos = {monitors: [monitorTodo({watch_only: true, ...patch})], advancements: [], resume: null};
+    assert.equal(trigger(many(6, poll), {todos})?.kind, "dead_monitor_repeat");
+  }
+  const todos = {monitors: [monitorTodo({watch_only: true})], advancements: [], resume: null};
+  assert.equal(trigger(many(6, n => poll(n, {target_id: "target-digest", target_key: "watch"})), {todos}), null,
+    "the recorded target digest is distinct from the canonical target key");
+  assert.equal(trigger(many(6, n => poll(n, {todo_id: null, target_key: "watch"})), {todos}), null);
+  assert.equal(trigger(many(6, n => poll(n, {todo_id: "other", target_key: "watch"})), {todos})?.kind,
+    "dead_monitor_repeat", "a matching key cannot override a different Todo");
+});
+
+test("intentional watch state does not discharge independent work reviews", () => {
+  const todos = {monitors: [monitorTodo({watch_only: true})], advancements: [], resume: null};
+  assert.equal(trigger(many(2, n => run(n, {progress: observation})), {todos})?.kind,
+    "typed_progress_repeat");
+  assert.equal(trigger(many(20), {todos})?.kind, "periodic_review_due");
+  assert.equal(trigger(many(2, n => poll(n, {
+    mode: "blocked_successor_wait_without_material_transition", frontier: "frontier-a",
+  })), {todos})?.kind, "blocked_successor_no_progress_repeat");
+});
+
+test("both current streak projections skip intentional watches and retain finite monitors", () => {
+  for (const operation of ["monitor_streak", "frontier_monitor_streak"]) {
+    const todos = {monitors: [monitorTodo({watch_only: true, no_change_count: 99})],
+      advancements: [], resume: null};
+    assert.equal(trigger([], {operation, observed_at: 100, todos}), null);
+    assert.equal(trigger([], {operation, observed_at: 100, todos: {...todos,
+      monitors: [...todos.monitors, monitorTodo({id: "finite", target: "finite", watch_only: false})],
+    }})?.monitor_target_id, "finite");
+  }
+});
+
+test("malformed watch state is rejected at the typed history boundary", () => {
+  for (const watch_only of ["true", 1, null]) {
+    const todos = {monitors: [monitorTodo({watch_only})], advancements: [], resume: null};
+    assert.throws(() => trigger(many(6, poll), {todos}), /watch_only/);
+  }
+});
+
 test("persisted monitor streak retains stable tie order and advancement ownership", () => {
   const todos = { monitors: [monitorTodo(), monitorTodo({ target: "second" })], advancements: [], resume: null };
   assert.equal(trigger([], { operation: "monitor_streak", todos })?.monitor_target_id, "watch");