diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 64338a29fc..3c0cbb5926 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -196,6 +196,18 @@ operation. Monitoring requires the separate explicit setting below. ### Native status monitoring +If the host restarts before a native watch observes anything, the original +idle pump can create one separately correlated Chat continuation. It preserves +the failed request and uses the same native observation identity and deadline. +The shared typed admission requires an untouched observation-only request, an +available original worker lock, complete event/history reads and no pending or +committed native poll receipt. It rechecks current authority, lease generation +and user work; it never starts a model, repeats business or spends execution +quota. Unknown effects, unreadable history, an active old worker, revoked or +paused watches and another failed continuation remain held for reconciliation. +This bounded recovery applies to both local managed Sessions and the prepared +steward through the existing host, without another scheduler or consumer. + After preparation, choose **Enable monitoring** in the same connection panel. The native continuous-monitor Todo owns enable/pause, the 30–1440 minute cadence, last observation, result hash and next deadline. The panel's single gesture diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index 8eb1a92592..aacdc29332 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -286,6 +286,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}), **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}), **StewardTurnHost(self.controller).report_quota_basis(request), + **StewardTurnHost(self.controller).monitor_restart_basis(request), "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) def inspect(self, session_id: str, *, agent_id: str | None = None, @@ -365,6 +366,14 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, client_id += ".quota-recovery" existing = None selection_recovery = host.report_selection_recovery(existing) if existing is not None and report_window is not None else None + restart_request = host.monitor_restart_request(existing) if existing is not None and monitor_deadline is not None else None + if restart_request is not None: + client_id += ".restart-recovery" + continued = self.store.turn_for_client(session_id, client_id) + if continued is not None: + return continued, False + request, existing = restart_request, None + request["monitor_lease_version"] = retained["version"] if retained else 0 if existing is not None: if selection_recovery is None: return existing, False # Failed/unknown work is never replayed. diff --git a/loopx/capabilities/native_chat/steward_monitor.py b/loopx/capabilities/native_chat/steward_monitor.py index 2ed450c194..8e44b0f940 100644 --- a/loopx/capabilities/native_chat/steward_monitor.py +++ b/loopx/capabilities/native_chat/steward_monitor.py @@ -245,12 +245,16 @@ def observe(self, *, host: Any, request: dict[str, Any], turn: dict[str, Any], # the watch as a heartbeat's advancement task. # Freeze admission's generation. A pause/re-enable between collection # and acquisition must not turn an old observation into fresh work. - lease = self._acquire(actor, todo, request["turn_instance_id"], + # A new carrier cannot borrow an old acquire receipt after that lease + # expired or was released. The native poll identity stays unchanged. + recovery = request.get("monitor_restart_recovery") + lease_key = "monitor-recovery." + recovery["original_turn_id"] if recovery else request["turn_instance_id"] + lease = self._acquire(actor, todo, lease_key, expected_version=request["monitor_lease_version"])["lease"] try: result = self._commit_poll(host, request, turn, context, ["quota", "monitor-poll", *route, "--target-key", todo["target_key"], "--result-hash", observation["result_hash"], - "--task-lease-idempotency-key", request["turn_instance_id"], + "--task-lease-idempotency-key", lease_key, "--task-lease-expected-version", str(lease["version"]), "--execute", *(["--material-change"] if observation["material_change"] else [])]) finally: diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 82478a1a74..4faeed3b11 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -112,11 +112,65 @@ def scope_id(ids): **({"report_launch_recovery": launch_recovery} if launch_recovery is not None else {}), **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}), **self.report_quota_basis(request), + **self.monitor_restart_basis(request), **({"preparing_model_context": True} if prepare_context else {}), **report}) if admitted["refresh_original_scope"]: self.store.update_session(session["session_id"], manager_authorization_scope_id=scope["current"]) return {**admitted, "observed_scope_id": scope["current"] if scope else None} + def monitor_restart_basis(self, request: Mapping[str, Any]) -> dict[str, Any]: + """Read original worker/journals without dispatching or repairing effects. + + Any pending, committed, unreadable or legacy poll evidence withholds a + new carrier. The shared typed admission owner interprets these facts. + """ + recovery = request.get("monitor_restart_recovery") + if recovery is None: + return {} + from ...chat_store import _read_jsonl + from ...control_plane.quota.monitor_poll import _monitor_poll_effect_id, _monitor_poll_receipt_path + from .steward_commands import native_command + actor = request["actor"] + original = self.store.load_turn(actor["session_id"], recovery["original_turn_id"]) + if original is None: + raise ValueError("the original stopped watch is unavailable") + root = self.controller.coordination_runtime_root + with exclusive_file_lock(self.store.turn_execution_lock_path(actor["session_id"], original["turn_id"]), + timeout_seconds=0, operation="observe_stopped_native_watch"): + # Strict reads are essential: an unreadable/compacted journal is + # unknown, never an empty observation or proof that work did not run. + events = _read_jsonl(self.store._event_path(actor["session_id"], original["turn_id"]), + raise_on_error=True, strict=True) + rows = _read_jsonl(root / "goals" / actor["goal_id"] / "runs" / "index.jsonl", + raise_on_error=True, strict=True) + present = any(row.get("classification") == "quota_monitor_poll" + and row.get("turn_instance_id") == request["turn_instance_id"] for row in rows) + from .steward_monitor import TARGET + target = ("managed-session-watch." + actor["session_id"] + if actor.get("schema_version") == "loopx_managed_session_actor_v0" else TARGET) + for todo, key in [(request["todo_id"], target), (None, target), (None, None)]: + effect = _monitor_poll_effect_id(runtime_root=root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], turn_instance_id=request["turn_instance_id"], + todo_id=todo, target_key=key, goal_ref=None) + present = present or _monitor_poll_receipt_path(root, actor["goal_id"], effect).exists() + lease = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"], + "--todo-id", request["todo_id"]]) + held = lease.get("lease") or {} + available = lease.get("active") is False or (held.get("owner") == actor["agent_id"] + and held.get("idempotency_key") == "monitor-recovery." + original["turn_id"]) + return {"monitor_restart_basis": {"original": original, + "event_kinds": [row["kind"] for row in events], "poll_absent": not present, + "old_worker_excluded": True, "lease_available": available}} + + @staticmethod + def monitor_restart_request(original: Mapping[str, Any]) -> dict[str, Any] | None: + """Prepare correlation only; current typed admission decides recovery.""" + prior = original.get("loopx_request") or {} + if (original.get("status") != "failed" or original.get("error_code") != "server_restarted" + or prior.get("work_kind") != "monitor" or prior.get("monitor_restart_recovery") is not None): + return None + return {**prior, "monitor_restart_recovery": {"original_turn_id": original["turn_id"]}} + def report_quota_basis(self, request: Mapping[str, Any]) -> dict[str, Any]: """Transport original journal observations; the typed owner admits work.""" recovery = request.get("report_quota_recovery") @@ -409,6 +463,14 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None request = quota_request client_id = native_steward_client_id(actor, request["turn_instance_id"]) existing = None + if existing is not None: + restart_request = self.monitor_restart_request(existing) if work_kind == "monitor" else None + if restart_request is not None: + client_id += ".restart-recovery" + continued = self.store.turn_for_client(actor["session_id"], client_id) + if continued is not None: + return continued, False + request, existing = restart_request, None if existing is not None: monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing) if not monitor_retry and work_kind != "report": diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 9aba7f43e4..3e3210cd89 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -13,6 +13,38 @@ import {goalAgentConversationIdentity, resolveConversationScope} from "./convers import {readTurnSelectionRejection} from "../turn_driver/selection_rejection.ts"; import {EffectiveAction} from "../quota/effective_action.generated.ts"; +/** A stopped observation may get one new Chat carrier, never a new native + * observation identity. The host independently reads the old worker lock, + * complete journals and native poll receipts; normal admission still checks + * current ownership, deadline, permissions and user work on every reentry. */ +function admitMonitorRestartRecovery(params: JsonObject, request: JsonObject, session: JsonObject): void { + if (request.monitor_restart_recovery === undefined) { + if (params.monitor_restart_basis != null) throw new EffectRuntimeRequestError("watch recovery lacks its original request"); + return; + } + const recovery = requireJsonObject(request.monitor_restart_recovery, "original watch continuation"); + const basis = requireJsonObject(params.monitor_restart_basis, "original watch observation"); + const original = requireJsonObject(basis.original, "original stopped watch"); + const prior = requireJsonObject(original.loopx_request, "original watch request"); + const keys = Object.keys(request).filter(key => key !== "monitor_restart_recovery"); + const events = basis.event_kinds; + if (Object.keys(recovery).join() !== "original_turn_id" + || recovery.original_turn_id !== original.turn_id || typeof original.turn_id !== "string" + || !/^[a-f0-9]{32}$/.test(original.turn_id) || original.session_id !== session.session_id + || original.status !== "failed" || original.error_code !== "server_restarted" + || original.origin !== "native_loopx" || original.loopx_execution !== true + || original.response !== null || original.upstream_turn_id !== null || original.delta_count !== 0 + || prior.work_kind !== "monitor" || request.work_kind !== "monitor" || request.validation_command !== null + || prior.monitor_restart_recovery !== undefined || basis.old_worker_excluded !== true + || basis.poll_absent !== true || basis.lease_available !== true || !Array.isArray(events) || !events.includes("turn.failed") + || events.some(kind => !["turn.queued", "agent.phase", "manager.context", "turn.failed"].includes(String(kind))) + || Object.keys(prior).length !== keys.length + || keys.some(key => key !== "monitor_lease_version" && JSON.stringify(request[key]) !== JSON.stringify(prior[key])) + || Number(params.queued_turn_count) !== 0) { + throw new EffectRuntimeRequestError("watch restart requires independent proof before any native observation"); + } +} + /** One new report Turn after a provider-proven untouched quota refusal. * This does not reopen the original Chat/kernel Turn or relax its recovery. * Native metadata and usable allowance are host observations, never model input. @@ -208,7 +240,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { const reportWork = request.work_kind === "report"; const scheduledWork = request.work_kind === "scheduled_task"; const requestFields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", - ...(observationOnly ? ["monitor_deadline", "monitor_lease_version"] : []), + ...(observationOnly ? ["monitor_deadline", "monitor_lease_version", "monitor_restart_recovery"] : []), ...(reportWork ? ["report_window_id", "report_subscription_revision", "report_quota_recovery"] : []), ...(scheduledWork ? ["schedule_window_id", "schedule_subscription_revision"] : [])]; if (Object.keys(request).some(key => !requestFields.includes(key)) @@ -229,6 +261,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("a local watch must retain its original Agent, Session, Todo and deadline"); } } + admitMonitorRestartRecovery(params, request, session); if (reportWork) { const window = requireJsonObject(params.report_window, "original report window"); const subscription = requireJsonObject(params.report_subscription, "current report subscription"); @@ -396,7 +429,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { } const request = requireJsonObject(params.request, "native steward Turn request"); const fields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", - "monitor_deadline", "monitor_lease_version", "report_window_id", "report_subscription_revision", "resume_turn_key", "report_quota_recovery"]; + "monitor_deadline", "monitor_lease_version", "monitor_restart_recovery", "report_window_id", "report_subscription_revision", "resume_turn_key", "report_quota_recovery"]; if (Object.keys(request).some(key => !fields.includes(key)) || request.operation !== "native_turn" || request.host !== "codex-app-server" || ![undefined, "native_turn", "monitor", "report"].includes(request.work_kind as string | undefined) @@ -412,6 +445,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { } admitReportSelectionRecovery(params, request, current.sessionId); admitReportQuotaRecovery(params, request, current.session); + admitMonitorRestartRecovery(params, request, current.session); if (turn && (turn.session_id !== current.sessionId || turn.turn_id !== current.session.active_turn_id || !["starting", "running"].includes(String(turn.status)) || turn.origin !== "native_loopx" || turn.loopx_execution !== true || JSON.stringify(turn.loopx_request) !== JSON.stringify(request))) { diff --git a/tests/test_native_steward_monitor.py b/tests/test_native_steward_monitor.py index 50a519329c..58c45206bb 100644 --- a/tests/test_native_steward_monitor.py +++ b/tests/test_native_steward_monitor.py @@ -353,6 +353,154 @@ def prepared(steward, tmp_path, monkeypatch, provider="sqlite"): return sid, actor, monitor +def stopped_watch(steward, tmp_path, monkeypatch, provider, audience): + store, runtime, _, transport, _, capture, _ = steward + if audience == "managed": + _, session, actor, todo_id, _ = local_watch(steward, tmp_path, monkeypatch, provider) + sid = session["session_id"] + else: + sid, actor, monitor = prepared(steward, tmp_path, monkeypatch, provider) + todo_id = monitor.current(actor)["todo_id"] + transport.reconcile() # deliver setup's original request, before the watch + # A crash after acceptance, before an observation or native poll. The + # original failed request remains immutable; only the host pump may recover. + with monkeypatch.context() as stopped_worker: + stopped_worker.setattr(runtime, "_start_accepted_turn_worker", lambda **_kw: None) + original_turn = pump_steward_monitors(runtime)[0] + store.update_turn(sid, original_turn["turn_id"], status="failed", error_code="server_restarted", + error="The original host stopped before observation.", response=None, upstream_turn_id=None) + store.append_event(sid, original_turn["turn_id"], kind="turn.failed", payload={"error_code": "server_restarted"}) + store.update_session(sid, active_turn_id=None, status="ready") + return sid, actor, todo_id, original_turn + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("audience", ["steward", "managed"]) +@pytest.mark.parametrize("released_original_lease", [False, True]) +def test_restart_before_observation_recovers_the_original_due_watch_once(steward, tmp_path, monkeypatch, provider, audience, released_original_lease): + import json + store, runtime, _, transport, _, capture, _ = steward + sid, actor, todo_id, original_turn = stopped_watch(steward, tmp_path, monkeypatch, provider, audience) + calls, writes = capture.read_bytes(), list(steward[2].writes) + used = spent_slots(StewardMonitor(runtime), actor) + if released_original_lease: + monitor = StewardMonitor(runtime) + watch = local_watch_row(runtime, actor, todo_id) + old_lease = monitor._acquire(actor, watch, original_turn["loopx_request"]["turn_instance_id"])["lease"] + monitor._release(actor, watch, old_lease) + retained = store.load_turn(sid, original_turn["turn_id"]) + watch = local_watch_row(runtime, actor, todo_id) + runtime._steward_monitor_next_check.clear() + recovered = pump_steward_monitors(runtime) + assert len(recovered) == 1, "a proven pre-observation restart must not strand a due watch" + assert recovered[0]["turn_id"] != retained["turn_id"] + assert recovered[0]["loopx_request"]["turn_instance_id"] == retained["loopx_request"]["turn_instance_id"] + completed = finish(runtime, recovered[0]) + assert completed["status"] == "completed", completed + assert completed["response"]["native_loopx"]["status"] == "observed" + after = local_watch_row(runtime, actor, todo_id) + assert after["next_due_at"] != watch["next_due_at"] and after.get("result_hash") + assert store.load_turn(sid, retained["turn_id"]) == retained + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + transport.reconcile() + assert steward[2].writes == writes and capture.read_bytes() == calls + assert spent_slots(StewardMonitor(runtime), actor) == used + # Native once-only semantics, not just the ephemeral pump cache. + rows = [json.loads(line) for line in (runtime.coordination_runtime_root / "goals" / actor["goal_id"] / "runs" / "index.jsonl").read_text().splitlines()] + assert len([row for row in rows if row.get("classification") == "quota_monitor_poll" + and row.get("turn_instance_id") == retained["loopx_request"]["turn_instance_id"]]) == 1 + wait_for_next_observation_tick(after) + edit_local_watch(runtime, actor, todo_id, "--next-due-at", (datetime.now(timezone.utc) - timedelta(seconds=2)).isoformat()) + following = pump_steward_monitors(runtime) + assert len(following) == 1 + next_observation = finish(runtime, following[0]) + assert next_observation["status"] == "completed", next_observation + assert "monitor_restart_recovery" not in following[0]["loopx_request"] + assert store.load_turn(sid, retained["turn_id"]) == retained and capture.read_bytes() == calls + assert spent_slots(StewardMonitor(runtime), actor) == used + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("audience", ["steward", "managed"]) +@pytest.mark.parametrize("fault", ["unknown_error", "native_started", "delta", "unknown_event", "broken_history", + "pending_poll", "legacy_poll", "queued", "worker_busy", "active_lease", "retry_fails"]) +def test_stopped_watch_recovery_preserves_unknown_effects_user_priority_and_one_attempt(steward, tmp_path, monkeypatch, provider, audience, fault): + from contextlib import nullcontext + from loopx.control_plane.quota.monitor_poll import _monitor_poll_effect_id, _monitor_poll_receipt_path + from loopx.file_lock import exclusive_file_lock + store, runtime, _, transport, _, capture, _ = steward + sid, actor, todo_id, original_turn = stopped_watch(steward, tmp_path, monkeypatch, provider, audience) + request = original_turn["loopx_request"] + root = runtime.coordination_runtime_root + if fault in {"unknown_error", "native_started", "delta"}: + changes = {"unknown_error": {"error_code": "runtime_error"}, + "native_started": {"upstream_turn_id": "unknown-native-turn"}, "delta": {"delta_count": 1}}[fault] + store.update_turn(sid, original_turn["turn_id"], **changes) + elif fault == "unknown_event": + store.append_event(sid, original_turn["turn_id"], kind="native_loopx.monitor_poll", payload={"unknown": True}) + elif fault == "broken_history": + index = root / "goals" / actor["goal_id"] / "runs" / "index.jsonl" + index.parent.mkdir(parents=True, exist_ok=True) + with index.open("ab") as output: + output.write(b'\n{"unknown":') + elif fault in {"pending_poll", "legacy_poll"}: + effect = _monitor_poll_effect_id(runtime_root=root, goal_id=actor["goal_id"], agent_id=actor["agent_id"], + turn_instance_id=request["turn_instance_id"], todo_id=None if fault == "legacy_poll" else todo_id, + target_key=None, goal_ref=None) + receipt = _monitor_poll_receipt_path(root, actor["goal_id"], effect) + receipt.parent.mkdir(parents=True, exist_ok=True) + receipt.write_text('{"status":"provider_pending"}') # even incomplete evidence must hold + elif fault == "queued": + store.create_queued_turn(sid, client_turn_id="new-user-work", message="Handle my current request first.", origin="web") + elif fault == "active_lease": + monitor = StewardMonitor(runtime) + monitor._acquire(actor, local_watch_row(runtime, actor, todo_id), "independent-watch-owner") + elif fault == "retry_fails": + def unavailable(*_args, **_kwargs): + raise OSError("Synthetic source failed before the new poll.") + monkeypatch.setattr(StewardMonitor, "_commit_poll", unavailable) + original = store.load_turn(sid, original_turn["turn_id"]) + watch, calls, writes = local_watch_row(runtime, actor, todo_id), capture.read_bytes(), list(steward[2].writes) + runtime._steward_monitor_next_check.clear() + guard = (exclusive_file_lock(store.turn_execution_lock_path(sid, original_turn["turn_id"]), timeout_seconds=0) + if fault == "worker_busy" else nullcontext()) + with guard: + recovered = pump_steward_monitors(runtime) + if fault == "retry_fails": + assert len(recovered) == 1 and finish(runtime, recovered[0])["status"] == "failed" + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + else: + assert recovered == [] + assert store.load_turn(sid, original_turn["turn_id"]) == original + assert local_watch_row(runtime, actor, todo_id) == watch + transport.reconcile() + assert capture.read_bytes() == calls and steward[2].writes == writes + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("audience", ["steward", "managed"]) +def test_reopened_controller_recovers_only_the_persisted_untouched_watch(steward, tmp_path, monkeypatch, provider, audience): + from loopx.chat_store import ChatSessionStore + from loopx.chat_runtime import ChatRuntimeController + store, runtime, _, _, _, capture, _ = steward + sid, actor, todo_id, original = stopped_watch(steward, tmp_path, monkeypatch, provider, audience) + before, retained = capture.read_bytes(), store.load_turn(sid, original["turn_id"]) + runtime.close() + reopened = ChatRuntimeController(store=ChatSessionStore(store.root.parent), codex_bin=runtime.codex_bin, + registry_path=runtime.registry_path, project_contexts=runtime.project_contexts) + reopened.manager_scope_resolver = runtime.manager_scope_resolver + try: + turns = pump_steward_monitors(reopened) + assert len(turns) == 1 and finish(reopened, turns[0])["status"] == "completed" + assert local_watch_row(reopened, actor, todo_id).get("last_checked_at") + assert reopened.store.load_turn(sid, original["turn_id"]) == retained + assert not pump_steward_monitors(reopened) and capture.read_bytes() == before + finally: + reopened.close() + + @pytest.mark.parametrize("unreadable", ["session", "queued_turn"]) @pytest.mark.parametrize("storage_provider", ["file", "sqlite"]) def test_unreadable_conversation_does_not_starve_another_native_monitor(steward, tmp_path, monkeypatch, unreadable, storage_provider):