fix(chat): recover watches after identity-read refusal

docs/product/periodic-reports.md

@@ -275,8 +275,12 @@ idle pump can create one separately correlated Chat continuation. It preserves
 the failed request and uses the same native observation identity and deadline.
 For a bound steward, this also covers a portfolio read explicitly refused
 because its authorization snapshot changed. The exact original failure and a
-single empty `external_authorization_changed` context must both be present;
-other runtime errors or missing, conflicting or unreadable context do not qualify.
+single empty `external_authorization_changed` context must both be present.
+An App identity read that fails before context publication can also qualify,
+but only with its exact original error, one recorded `manager_context` phase,
+no published context and the independent no-effect checks below. The current
+provider must verify successfully before acceptance; a matching error alone,
+other runtime errors or missing, conflicting or unreadable evidence do not qualify.
 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

loopx/capabilities/native_chat/steward_turn.py

@@ -162,6 +162,7 @@ def monitor_restart_basis(self, request: Mapping[str, Any]) -> dict[str, Any]:
             return {"monitor_restart_basis": {"original": original,
                 "event_kinds": [row["kind"] for row in events], "poll_absent": not present,
                 "contexts": [row["payload"] for row in events if row["kind"] == "manager.context"],
+                "phases": [row["payload"] for row in events if row["kind"] == "agent.phase"],
                 "old_worker_excluded": True, "lease_available": available}}
 
     @staticmethod

loopx/control_plane/collaboration/steward_managed_mode.ts

@@ -43,10 +43,22 @@ function admitMonitorRestartRecovery(params: JsonObject, request: JsonObject, se
       && coverage.discovered === null && coverage.verified === 0 && coverage.complete === false
       && context.authorization_scope_id === undefined;
   }
+  // A provider identity read may fail inside the original read-only context
+  // collector, before it publishes any context. Retained legacy errors have no
+  // typed failure code: require the collector's exact phase independently of
+  // the error, plus the strict no-effect evidence below. Current binding and
+  // provider verification still run before a new carrier can be accepted.
+  const identityRefused = original.error_code === "runtime_error"
+    && original.error === "the selected App identity could not be verified"
+    && requireJsonObject(prior.actor, "original watch actor").schema_version === "loopx_steward_managed_actor_v0"
+    && Array.isArray(basis.contexts) && basis.contexts.length === 0
+    && Array.isArray(basis.phases) && basis.phases.length === 1
+    && requireJsonObject(basis.phases[0], "original watch phase").phase === "manager_context"
+    && Array.isArray(events) && events.filter(kind => kind === "agent.phase").length === 1;
   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" || scopeRefused)
+      || original.status !== "failed" || !(original.error_code === "server_restarted" || scopeRefused || identityRefused)
       || 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

tests/control_plane_ts/steward_managed_mode.test.ts

@@ -48,6 +48,39 @@ test("only the canonical open watch can run without a model, and it cannot reuse
     turn: {...input.turn, loopx_request: modelRequest}}));
 });
 
+test("identity-refused watches need independent collector evidence and retain current admission", () => {
+  const {params, goal} = fixture();
+  const actor = bindStewardManagedActor({...params, goal});
+  const prior = {operation: "native_turn", host: "codex-app-server", actor,
+    work_kind: "monitor", turn_instance_id: "monitor.original", todo_id: "todo_watch", validation_command: null,
+    monitor_deadline: "2026-10-23T10:00:00Z", monitor_lease_version: 0};
+  const original = {turn_id: "1".repeat(32), session_id: params.session.session_id,
+    status: "failed", error_code: "runtime_error", error: "the selected App identity could not be verified",
+    origin: "native_loopx", loopx_execution: true, upstream_turn_id: null, response: null,
+    delta_count: 0, loopx_request: prior};
+  const basis = {original, contexts: [], phases: [{phase: "manager_context"}],
+    event_kinds: ["turn.queued", "agent.phase", "turn.failed"],
+    poll_absent: true, old_worker_excluded: true, lease_available: true};
+  const monitor = {todo_id: "todo_watch", task_class: "continuous_monitor", target_key: "steward-portfolio-watch",
+    watch_only: true, claimed_by: "steward", status: "open", next_due_at: prior.monitor_deadline};
+  const input = {...params, goal, goal_active: true, binding: {...params.binding, managed_actor: actor},
+    request: {...prior, monitor_restart_recovery: {original_turn_id: original.turn_id}},
+    monitor_todo: monitor, monitor_restart_basis: basis};
+  assert.equal(admitStewardManagedTurn(input).observation_only, true);
+  for (const changed of [
+    {phases: []}, {phases: [{phase: "model_start"}]}, {phases: [{phase: "manager_context"}, {phase: "manager_context"}]},
+    {contexts: [{}]}, {event_kinds: ["turn.queued", "turn.failed"]},
+    {event_kinds: [...basis.event_kinds, "agent.phase"]},
+    {poll_absent: false}, {old_worker_excluded: false}, {lease_available: false},
+    {original: {...original, error: "Unknown identity failure"}},
+    {original: {...original, upstream_turn_id: "unknown-model-turn"}},
+  ]) assert.throws(() => admitStewardManagedTurn({...input, monitor_restart_basis: {...basis, ...changed}}));
+  for (const changed of [{queued_turn_count: 1}, {goal_active: false},
+    {monitor_todo: {...monitor, status: "blocked"}}, {observation: {...params.observation, verified: false}}]) {
+    assert.throws(() => admitStewardManagedTurn({...input, ...changed}));
+  }
+});
+
 test("monitor configuration retains the original actor and native Todo authority", () => {
   const {params, goal} = fixture();
   const actor = bindStewardManagedActor({...params, goal});

tests/test_native_steward_monitor.py

@@ -395,13 +395,42 @@ def changed_snapshot(*args, **kwargs):
     return sid, actor, monitor.current(actor)["todo_id"], failed
 
 
+def identity_refused_watch(steward, tmp_path, monkeypatch, provider):
+    import loopx.chat_manager_context as context_owner
+    store, runtime, source, transport, *_ = steward
+    sid, actor, monitor = prepared(steward, tmp_path, monkeypatch, provider)
+    transport.reconcile()
+    collect = context_owner.collect_manager_turn_context
+
+    def fail_during_collection(*args, **kwargs):
+        # Admission and scope collection verify the provider. Fail the later
+        # commission-evidence read before the host publishes its context.
+        context = collect(*args, **kwargs)
+        source.bot_ready = False
+        return context
+
+    try:
+        with monkeypatch.context() as changed:
+            changed.setattr(context_owner, "collect_manager_turn_context", fail_during_collection)
+            failed = finish(runtime, pump_steward_monitors(runtime)[0])
+    finally:
+        source.bot_ready = True
+    assert failed["status"] == "failed"
+    assert failed["error"] == "the selected App identity could not be verified"
+    events = store.events_after(sid, failed["turn_id"], None)
+    assert [row["kind"] for row in events] == ["turn.queued", "agent.phase", "turn.failed"]
+    return sid, actor, monitor.current(actor)["todo_id"], failed
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_scope_refusal_is_automatically_reobserved_once_after_host_reopen(steward, tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("failure", ["authorization", "identity"])
+def test_refused_watch_is_automatically_reobserved_once_after_host_reopen(steward, tmp_path, monkeypatch, provider, failure):
     import json
     from loopx.chat_runtime import ChatRuntimeController
     from loopx.chat_store import ChatSessionStore
     store, runtime, _, transport, _, capture, _ = steward
-    sid, actor, todo_id, failed = scope_refused_watch(steward, tmp_path, monkeypatch, provider)
+    refused = scope_refused_watch if failure == "authorization" else identity_refused_watch
+    sid, actor, todo_id, failed = refused(steward, tmp_path, monkeypatch, provider)
     retained = store._turn_path(sid, failed["turn_id"]).read_bytes()
     before = local_watch_row(runtime, actor, todo_id)
     calls, writes, used = capture.read_bytes(), list(steward[2].writes), spent_slots(StewardMonitor(runtime), actor)
@@ -444,6 +473,49 @@ def test_scope_refusal_is_automatically_reobserved_once_after_host_reopen(stewar
         reopened.close()
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("fault", ["missing_phase", "different_phase", "duplicate_phase", "context_present",
+    "different_error", "provider_unavailable", "retry_fails"])
+def test_identity_refusal_requires_original_collector_evidence_and_current_provider(steward, tmp_path, monkeypatch, provider, fault):
+    import json
+    store, runtime, source, _, _, capture, _ = steward
+    sid, actor, todo_id, failed = identity_refused_watch(steward, tmp_path, monkeypatch, provider)
+    path = store._event_path(sid, failed["turn_id"])
+    if fault in {"missing_phase", "different_phase", "duplicate_phase"}:
+        rows = [json.loads(line) for line in path.read_text().splitlines()]
+        phase = next(row for row in rows if row["kind"] == "agent.phase")
+        if fault == "missing_phase":
+            rows.remove(phase)
+        elif fault == "different_phase":
+            phase["payload"]["phase"] = "model_start"
+        else:
+            rows.append(phase)
+        path.write_text("".join(json.dumps(row) + "\n" for row in rows))
+    elif fault == "context_present":
+        store.append_event(sid, failed["turn_id"], kind="manager.context", payload={"unknown": True})
+    elif fault == "different_error":
+        store.update_turn(sid, failed["turn_id"], error="Unknown failure before context.")
+    elif fault == "provider_unavailable":
+        source.bot_ready = False
+    elif fault == "retry_fails":
+        def unavailable(*_args, **_kwargs):
+            raise OSError("Source unavailable in the one continuation.")
+        monkeypatch.setattr(StewardMonitor, "_commit_poll", unavailable)
+    retained = store._turn_path(sid, failed["turn_id"]).read_bytes()
+    watch, calls, writes = local_watch_row(runtime, actor, todo_id), capture.read_bytes(), list(source.writes)
+    runtime._steward_monitor_next_check.clear()
+    continued = pump_steward_monitors(runtime)
+    if fault == "retry_fails":
+        assert len(continued) == 1 and finish(runtime, continued[0])["status"] == "failed"
+        runtime._steward_monitor_next_check.clear()
+        assert not pump_steward_monitors(runtime)
+    else:
+        assert continued == []
+    assert store._turn_path(sid, failed["turn_id"]).read_bytes() == retained
+    assert local_watch_row(runtime, actor, todo_id) == watch
+    assert capture.read_bytes() == calls and source.writes == writes
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("fault", ["missing_context", "different_context", "duplicate_context", "different_error",
     "model_started", "poll_receipt", "broken_history", "worker_busy", "queued", "paused", "revoked", "retry_fails"])