fix(chat): recover untouched native watches after host restart

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

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.

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:

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":

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))) {