From 536673503648aabe7639d9e1d2de713750e8f39b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 16:52:32 +0800 Subject: [PATCH] fix(chat): recover watches after a proven identity-read refusal Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 8 +- .../capabilities/native_chat/steward_turn.py | 1 + .../collaboration/steward_managed_mode.ts | 14 +++- .../steward_managed_mode.test.ts | 33 ++++++++ tests/test_native_steward_monitor.py | 76 ++++++++++++++++++- 5 files changed, 127 insertions(+), 5 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 0397590114..acb9176bbe 100644 --- a/docs/product/periodic-reports.md +++ b/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 diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 0c93e830f7..a99220b9ef 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/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 diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 096bdd011c..64bf60096e 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/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 diff --git a/tests/control_plane_ts/steward_managed_mode.test.ts b/tests/control_plane_ts/steward_managed_mode.test.ts index 3495dbb86c..4f6738b8b8 100644 --- a/tests/control_plane_ts/steward_managed_mode.test.ts +++ b/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}); diff --git a/tests/test_native_steward_monitor.py b/tests/test_native_steward_monitor.py index 60364b7e9a..869a950ae7 100644 --- a/tests/test_native_steward_monitor.py +++ b/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"])