From 8a563b4a733f2a17de76cb0008e7e962811ab5b5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 06:03:27 +0800 Subject: [PATCH 1/2] fix(chat): continue a refused native monitor observation once Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 4 + .../capabilities/native_chat/steward_turn.py | 3 +- .../collaboration/steward_managed_mode.ts | 17 ++- tests/test_native_steward_monitor.py | 132 ++++++++++++++++++ 4 files changed, 154 insertions(+), 2 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 9f60f3e359..5aea6d094b 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -257,6 +257,10 @@ operation. Monitoring requires the separate explicit setting below. 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. +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. 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 0d59b7b1c1..97998af318 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -160,13 +160,14 @@ def monitor_restart_basis(self, request: Mapping[str, Any]) -> dict[str, Any]: 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, + "contexts": [row["payload"] for row in events if row["kind"] == "manager.context"], "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" + if (original.get("status") != "failed" or original.get("error_code") not in {"server_restarted", "runtime_error"} 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"]}} diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 3e3210cd89..23454b7adf 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -28,10 +28,25 @@ function admitMonitorRestartRecovery(params: JsonObject, request: JsonObject, se 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; + // Compatibility with persisted Chat failures: the exact error alone is not + // proof. Require the original collector's single, empty typed refusal too. + // Fresh scope/provider checks below still decide the new observation. + let scopeRefused = false; + if (original.error_code === "runtime_error" && original.error === "the portfolio observation omitted authorized Goals" + && requireJsonObject(prior.actor, "original watch actor").schema_version === "loopx_steward_managed_actor_v0" + && Array.isArray(basis.contexts) && basis.contexts.length === 1) { + const context = requireJsonObject(basis.contexts[0], "refused watch context"); + const coverage = requireJsonObject(context.coverage, "refused watch coverage"); + scopeRefused = context.schema_version === "manager_turn_context_v1" + && Array.isArray(context.goals) && context.goals.length === 0 + && Array.isArray(context.warnings) && context.warnings.length === 1 && context.warnings[0] === "external_authorization_changed" + && coverage.discovered === null && coverage.verified === 0 && coverage.complete === false + && context.authorization_scope_id === undefined; + } 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.status !== "failed" || !(original.error_code === "server_restarted" || scopeRefused) || 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/test_native_steward_monitor.py b/tests/test_native_steward_monitor.py index 58c45206bb..60364b7e9a 100644 --- a/tests/test_native_steward_monitor.py +++ b/tests/test_native_steward_monitor.py @@ -374,6 +374,138 @@ def stopped_watch(steward, tmp_path, monkeypatch, provider, audience): return sid, actor, todo_id, original_turn +def scope_refused_watch(steward, tmp_path, monkeypatch, provider): + import loopx.chat_manager_context as context_owner + store, runtime, _, transport, *_ = steward + sid, actor, monitor = prepared(steward, tmp_path, monkeypatch, provider) + transport.reconcile() + scope_id = context_owner.manager_authorization_scope_id_for_registry + + def changed_snapshot(*args, **kwargs): + # Exercise the production collector's refusal of a changed registry + # snapshot, then the real Chat worker and typed observation owner. + return None if kwargs.get("expected_inventory_revision") else scope_id(*args, **kwargs) + + with monkeypatch.context() as changed: + changed.setattr(context_owner, "manager_authorization_scope_id_for_registry", changed_snapshot) + original_turn = pump_steward_monitors(runtime)[0] + failed = finish(runtime, original_turn) + assert failed["status"] == "failed" + assert failed["error"] == "the portfolio observation omitted authorized Goals" + 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): + 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) + 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) + 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 + wake = DelegationWakeService(reopened, interval=.02).start() + try: + deadline = time.monotonic() + 15 + while local_watch_row(reopened, actor, todo_id).get("last_checked_at") == before.get("last_checked_at"): + assert time.monotonic() < deadline, "the existing host must continue the refused observation" + time.sleep(.02) + after = local_watch_row(reopened, actor, todo_id) + wake.close() + turns = [json.loads(path.read_text()) for path in (reopened.store._session_dir(sid) / "turns").glob("*.json")] + turns = [row for row in turns if (row.get("loopx_request") or {}).get("monitor_restart_recovery")] + assert len(turns) == 1 + done = finish(reopened, turns[0]) + assert done["status"] == "completed" and done["response"]["native_loopx"]["status"] == "observed" + assert done["loopx_request"]["turn_instance_id"] == failed["loopx_request"]["turn_instance_id"] + assert done["loopx_request"]["monitor_deadline"] == before["next_due_at"] + assert after["next_due_at"] != before["next_due_at"] and after.get("result_hash") + assert reopened.store._turn_path(sid, failed["turn_id"]).read_bytes() == retained + reopened._steward_monitor_next_check.clear() + assert not pump_steward_monitors(reopened) + rows = [json.loads(line) for line in (reopened.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") == failed["loopx_request"]["turn_instance_id"]]) == 1 + assert capture.read_bytes() == calls and steward[2].writes == writes + assert spent_slots(StewardMonitor(reopened), actor) == used + wait_for_next_observation_tick(after) + edit_local_watch(reopened, actor, todo_id, "--next-due-at", (datetime.now(timezone.utc) - timedelta(seconds=2)).isoformat()) + next_turns = pump_steward_monitors(reopened) + assert len(next_turns) == 1 and finish(reopened, next_turns[0])["status"] == "completed" + assert "monitor_restart_recovery" not in next_turns[0]["loopx_request"] + assert capture.read_bytes() == calls and spent_slots(StewardMonitor(reopened), actor) == used + finally: + wake.close() + reopened.close() + + +@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"]) +def test_scope_refusal_retains_unknown_effects_and_current_authority(steward, tmp_path, monkeypatch, provider, 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, binding, capture, _ = steward + sid, actor, todo_id, failed = scope_refused_watch(steward, tmp_path, monkeypatch, provider) + path = store._event_path(sid, failed["turn_id"]) + if fault in {"missing_context", "different_context", "duplicate_context"}: + import json + rows = [json.loads(line) for line in path.read_text().splitlines()] + context = next(row for row in rows if row["kind"] == "manager.context") + if fault == "missing_context": + rows.remove(context) + elif fault == "different_context": + context["payload"]["warnings"] = ["source_unavailable"] + else: + rows.append(context) + path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + elif fault in {"different_error", "model_started"}: + store.update_turn(sid, failed["turn_id"], **({"error": "Unknown observation failure."} + if fault == "different_error" else {"upstream_turn_id": "unknown-model-turn"})) + elif fault == "poll_receipt": + request = failed["loopx_request"] + effect = _monitor_poll_effect_id(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], turn_instance_id=request["turn_instance_id"], todo_id=todo_id, + target_key=None, goal_ref=None) + receipt = _monitor_poll_receipt_path(runtime.coordination_runtime_root, actor["goal_id"], effect) + receipt.parent.mkdir(parents=True, exist_ok=True) + receipt.write_text('{"status":"provider_pending"}') + elif fault == "broken_history": + with path.open("ab") as output: + output.write(b'\n{"unknown":') + elif fault == "queued": + store.create_queued_turn(sid, client_turn_id="scope-refusal-user-priority", message="Do my current task first.", origin="web") + elif fault == "paused": + edit_local_watch(runtime, actor, todo_id, "--status", "blocked") + elif fault == "revoked": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + 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(steward[2].writes) + runtime._steward_monitor_next_check.clear() + guard = exclusive_file_lock(store.turn_execution_lock_path(sid, failed["turn_id"]), timeout_seconds=0) if fault == "worker_busy" else nullcontext() + with guard: + resumed = pump_steward_monitors(runtime) + if fault == "retry_fails": + assert len(resumed) == 1 and finish(runtime, resumed[0])["status"] == "failed" + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + else: + assert resumed == [] + 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 steward[2].writes == writes + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("audience", ["steward", "managed"]) @pytest.mark.parametrize("released_original_lease", [False, True]) From ef598681fd440928d32f2ff1079037d77072661e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 06:13:14 +0800 Subject: [PATCH 2/2] fix(chat): retain legacy monitor refusal recovery precedence Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/native_chat/steward_turn.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 97998af318..1cfedcd960 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -465,7 +465,10 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None 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 + # Keep the separately proven legacy admission-refusal route + # below ahead of broader runtime-error observation candidates. + restart_request = (self.monitor_restart_request(existing) + if work_kind == "monitor" and not self._monitor_selection_failed_before_poll(existing) else None) if restart_request is not None: client_id += ".restart-recovery" continued = self.store.turn_for_client(actor["session_id"], client_id)