diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 20e25c1b01..911b03561a 100644 --- a/docs/quota-allocation.md +++ b/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`. diff --git a/loopx/control_plane/work_items/project_asset.py b/loopx/control_plane/work_items/project_asset.py index c518bcb1ba..7b07aa09ee 100644 --- a/loopx/control_plane/work_items/project_asset.py +++ b/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, ) ) diff --git a/loopx/control_plane/work_items/replan_history.ts b/loopx/control_plane/work_items/replan_history.ts index 53f30c0ccf..5692587b32 100644 --- a/loopx/control_plane/work_items/replan_history.ts +++ b/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" && diff --git a/loopx/control_plane/work_items/replan_history_codec.py b/loopx/control_plane/work_items/replan_history_codec.py index c2f2d446d9..18fad82150 100644 --- a/loopx/control_plane/work_items/replan_history_codec.py +++ b/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")), diff --git a/tests/control_plane/test_replan_history_policy.py b/tests/control_plane/test_replan_history_policy.py index 666127473a..85b50f8093 100644 --- a/tests/control_plane/test_replan_history_policy.py +++ b/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()}] diff --git a/tests/control_plane/test_replan_history_transport.py b/tests/control_plane/test_replan_history_transport.py index d6ecd90831..73dfcc3c46 100644 --- a/tests/control_plane/test_replan_history_transport.py +++ b/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" diff --git a/tests/control_plane_ts/replan_history.test.ts b/tests/control_plane_ts/replan_history.test.ts index 9f251af020..1c33c5d0b3 100644 --- a/tests/control_plane_ts/replan_history.test.ts +++ b/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");