From fdd6567368a1ed05c7b550aab8f103d9c93f4851 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 24 Oct 2026 11:25:44 +0800 Subject: [PATCH] fix(chat): resume proven prelaunch report refusals through the original Turn Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 12 +++ .../capabilities/native_chat/steward_turn.py | 60 +++++++++++++-- .../collaboration/steward_managed_mode.ts | 47 +++++++++++- .../control_plane_ts/steward_reports.test.ts | 57 ++++++++++++++ tests/test_native_steward_reports.py | 74 +++++++++++++++++++ 5 files changed, 241 insertions(+), 9 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 6db57ff8d0..4411040095 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -304,6 +304,18 @@ source and execution lease are checked after collection and again at dispatch; authority changes after plan capture still reject the launch. This ordering does not authorize replay of a historical attempt with unknown host effects. +The original pump can resume one retained report after the first-party launch +owner's exact `stale_goal_authority` refusal. That owner rejects before publishing +`turn/start`; a missing answer or missing events alone is insufficient. The +typed admission checks the original Chat request, canonical native transaction, +one recorded attempt, no generation or settlement, and the current original +subscription and grant. A separately correlated Chat attempt resumes the same +native journal, window, Todo and settlement identity through `run-once` recovery. +The historical outer response is retained. A failed recovery, unknown transport +outcome, real model start, changed identity or revoked permission cannot trigger +another automatic model attempt. Existing no-model settlement recovery and +independent full-body delivery readback still apply. + Once generation is frozen, delivery checks the original source Goal identities against the current authorized scope. Adding an authorized Goal does not revoke the frozen report or rewrite its source/body; later windows include the new diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 31a6ce9767..637eb8352e 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -45,7 +45,8 @@ def _monitor_selection_failed_before_poll(self, turn: dict[str, Any]) -> bool: for row in self.store.events_after(turn["session_id"], turn["turn_id"], None))) def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, - prepare_context: bool = False, settlement_recovery: dict[str, Any] | None = None) -> dict[str, Any]: + prepare_context: bool = False, settlement_recovery: dict[str, Any] | None = None, + launch_recovery: dict[str, Any] | None = None) -> dict[str, Any]: if self.controller.closed.is_set(): raise ValueError("the original Chat host is stopped") actor = request["actor"] @@ -87,11 +88,43 @@ def scope_id(ids): "goal": self.managed._goal(actor["goal_id"]), "goal_active": bool(goal and goal.get("status") == "active" and not goal_is_stopped(goal)), "request": request, "turn": turn, "scope": scope, "monitor_todo": monitor, **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}), + **({"report_launch_recovery": launch_recovery} if launch_recovery is not None else {}), **({"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 _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None: + """Read the original typed launch refusal and its canonical transaction. + + Missing model events alone prove nothing. Only the native launch + owner's specific pre-publication rejection can qualify this path; + the typed conversation owner checks it with the current journal. + """ + result = (original.get("response") or {}).get("native_loopx") or {} + if result.get("error_code") != "stale_goal_authority": + return None + from ...control_plane.turn_driver import inspect_loopx_turn_journal + from ...control_plane.turn_driver.journal_store import ( + find_loopx_turn_key_by_settlement_identity, load_turn_journal, turn_journal_path, + ) + from ..periodic_report.private_generation import attempt_directory + from .steward_reports import StewardReports + request = original["loopx_request"] + actor = request["actor"] + root = self.controller.coordination_runtime_root + key = find_loopx_turn_key_by_settlement_identity(root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], todo_id=request["todo_id"], turn_instance_id=request["turn_instance_id"]) + if key is None: + raise ValueError("the original launch rejection has no canonical Turn") + inspection = inspect_loopx_turn_journal(root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], turn_key=key, retry_failed=True) + journal = load_turn_journal(turn_journal_path(root, goal_id=actor["goal_id"], turn_key=key)) + _, window = StewardReports(self.controller).current_window(actor["goal_id"].removeprefix("steward-agent-")) + return {"original": original, "inspection": inspection, "turn_key": key, + "host_attempt_count": (journal or {}).get("host_attempt_count"), + "generation_present": window is None or (attempt_directory(root, window) / "private-generation.json").exists()} + def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any], todo: dict[str, Any]) -> None: """Resume only cached, validated native effects of the original report. @@ -173,6 +206,7 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None report_subscription_revision=report_window["subscription_revision"]) client_id = native_steward_client_id(actor, turn_instance_id) existing = self.store.turn_for_client(actor["session_id"], client_id) + launch_recovery = None if existing is not None: # Never replay a failed/unknown model Turn. A committed Chat # request is observed; native journal recovery remains separate. @@ -181,7 +215,8 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None frozen.setdefault("work_kind", "native_turn") if frozen != request or work_kind == "monitor" and existing["loopx_request"].get("monitor_deadline") != monitor_deadline: raise ValueError("native steward request identity conflict") - if work_kind != "monitor" or not self._monitor_selection_failed_before_poll(existing): + monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing) + if not monitor_retry and work_kind != "report": return existing, False # One separately correlated retry after the proven admission # failure. Retain the failed Chat Turn and original native @@ -190,6 +225,11 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None retry = self.store.turn_for_client(actor["session_id"], client_id) if retry is not None: return retry, False + if work_kind == "report": + launch_recovery = self._report_launch_recovery(existing) + if launch_recovery is None: + return existing, False + request["resume_turn_key"] = existing["response"]["native_loopx"].get("resume_turn_key") if work_kind == "monitor": from .steward_commands import native_command if not todo_id or not monitor_deadline: @@ -198,7 +238,7 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None "--todo-id", todo_id]).get("lease") request.update(monitor_deadline=monitor_deadline, monitor_lease_version=retained["version"] if retained else 0) - admitted = self._admit(request) + admitted = self._admit(request, launch_recovery=launch_recovery) session = self.store.load_session(actor["session_id"]) accepted = self.store.accept_managed_turn(actor["session_id"], client_turn_id=client_id, message="Advance the currently admitted native LoopX work.", @@ -314,9 +354,14 @@ def launch(): **({"allow_completion": True} if report_work else {})) argv = ["turn", "run-once", "--goal-id", actor["goal_id"], "--agent-id", actor["agent_id"], - "--project", admitted["workspace_path"], "--host", "generic-cli", "--execute", - "--turn-instance-id", request["turn_instance_id"]] - if request["todo_id"] is not None: + "--project", admitted["workspace_path"], "--host", "generic-cli", "--execute"] + if request.get("resume_turn_key"): + # Keep the original plan, Todo, native Turn and settlement identity. + # The kernel rechecks its journal and current launch authority. + argv += ["--resume-turn-key", request["resume_turn_key"], "--retry-failed-turn"] + else: + argv += ["--turn-instance-id", request["turn_instance_id"]] + if request["todo_id"] is not None and not request.get("resume_turn_key"): argv += ["--todo-id", request["todo_id"]] if request["validation_command"] is not None: argv += ["--validation-command-json", json.dumps(request["validation_command"])] @@ -331,7 +376,8 @@ def launch(): "--todo-id", request["todo_id"]]).get("lease") report_lease = native_command(self.controller, ["task-lease", "acquire", "--goal-id", actor["goal_id"], "--todo-id", request["todo_id"], "--owner", actor["agent_id"], - "--idempotency-key", request["turn_instance_id"], "--ttl-seconds", "3600", + "--idempotency-key", request["turn_instance_id"] + (".admission-retry" if request.get("resume_turn_key") else ""), + "--ttl-seconds", "3600", "--expected-version", str(retained["version"] if retained else 0)])["lease"] try: if report_work: diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index d92bfc1bd9..aa9a5d3428 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -87,7 +87,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"]; + "monitor_deadline", "monitor_lease_version", "report_window_id", "report_subscription_revision", "resume_turn_key"]; 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) @@ -146,8 +146,51 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { || todo.action_kind !== "generate_periodic_report" || !(todo.status === "open" && !recovery || settlementOnly)) { throw new EffectRuntimeRequestError("report execution requires the original live subscription, window and task"); } + if (request.resume_turn_key !== undefined) { + if (typeof request.resume_turn_key !== "string" || !/^sha256:[a-f0-9]{64}$/.test(request.resume_turn_key)) { + throw new EffectRuntimeRequestError("report recovery requires the original native Turn key"); + } + // An already accepted worker retains its ordinary active-Turn fence. + // Before acceptance, qualify one exact transport refusal against native + // history. This code is emitted only before turn/start publication by + // the first-party launch owner; absent events or a timeout do not qualify. + if (!turn) { + const proof = requireJsonObject(params.report_launch_recovery, "original launch rejection"); + const original = requireJsonObject(proof.original, "original rejected Chat Turn"); + const prior = requireJsonObject(original.loopx_request, "original rejected request"); + const result = requireJsonObject(requireJsonObject(original.response, "original response").native_loopx, "native rejection"); + const journal = requireJsonObject(proof.inspection, "canonical original Turn inspection"); + const decision = requireJsonObject(journal.recovery_decision, "original native recovery decision"); + const effects = requireJsonObject(journal.recorded_effects, "original native effects"); + const invocation = requireJsonObject(result.effects, "original invocation effects"); + const keys = Object.keys(request).filter(key => key !== "resume_turn_key"); + if (original.session_id !== current.sessionId || original.origin !== "native_loopx" + || original.loopx_execution !== true || original.status !== "completed" + || original.delta_count !== 0 || original.upstream_turn_id !== null + || Object.keys(prior).length !== keys.length + || keys.some(key => JSON.stringify(prior[key]) !== JSON.stringify(request[key])) + || result.schema_version !== "loopx_turn_execution_v0" || result.ok !== false + || result.mode !== "run_once" || result.effects_scope !== "current_invocation" + || result.error_code !== "stale_goal_authority" + || result.error !== "first-party Host runtime rejected: stale_goal_authority" + || result.resume_turn_key !== request.resume_turn_key + || proof.turn_key !== request.resume_turn_key || Object.values(invocation).some(value => value === true) + || journal.schema_version !== "loopx_turn_journal_inspection_v1" || journal.ok !== true + || journal.goal_matches !== true || journal.owner_matches !== true || journal.turn_key_matches !== true + || journal.journal_consistent !== true || journal.journal_status !== "in_progress" + || !Array.isArray(journal.completed_phases) || journal.completed_phases.length !== 0 + || proof.host_attempt_count !== 1 || proof.generation_present !== false + || effects.host_invoked !== null || effects.state_written !== false + || effects.quota_spent !== false || effects.scheduler_acknowledged !== false + || decision.action !== "continue" || decision.reinvoke_host !== true || decision.resume_from !== "host_execute") { + throw new EffectRuntimeRequestError("report launch recovery is not proven safe by the original native rejection"); + } + } + } else if (params.report_launch_recovery != null) { + throw new EffectRuntimeRequestError("launch recovery must retain the original native Turn key"); + } } else if (request.report_window_id !== undefined || request.report_subscription_revision !== undefined - || params.report_settlement_recovery != null) { + || request.resume_turn_key !== undefined || params.report_settlement_recovery != null || params.report_launch_recovery != null) { throw new EffectRuntimeRequestError("report fields belong only to an original calendar report"); } let refreshScope = false; diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 7dbfccd4d5..990854e75b 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -112,3 +112,60 @@ test("a model report requires the exact live calendar window and original task, {...input, report_todo: {...todo, status: "done"}}, ]) assert.throws(() => admitStewardManagedTurn(invalid)); }); + +test("a retained pre-publication launch refusal can resume only its original native transaction", () => { + const p = original(); + const prior = {operation: "native_turn", host: "codex-app-server", actor: p.actor, + turn_instance_id: "report.cadence_original", todo_id: "todo_report", validation_command: ["python", "validate.py"], + work_kind: "report", report_window_id: "cadence_original", report_subscription_revision: "sha256:original"}; + const key = "sha256:" + "a".repeat(64); + const request = {...prior, resume_turn_key: key}; + const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false, + error_code: "stale_goal_authority", error: "first-party Host runtime rejected: stale_goal_authority", + effects_scope: "current_invocation", effects: {host_invoked: null, state_written: null, quota_spent: null}, + resume_turn_key: key}; + const source = {session_id: p.session.session_id, status: "completed", origin: "native_loopx", loopx_execution: true, + delta_count: 0, upstream_turn_id: null, loopx_request: prior, response: {native_loopx: result}}; + const inspection = {schema_version: "loopx_turn_journal_inspection_v1", ok: true, + goal_matches: true, owner_matches: true, turn_key_matches: true, journal_consistent: true, + journal_status: "in_progress", completed_phases: [], + recorded_effects: {host_invoked: null, state_written: false, quota_spent: false, scheduler_acknowledged: false}, + recovery_decision: {action: "continue", reinvoke_host: true, resume_from: "host_execute"}}; + const proof = {original: source, inspection, turn_key: key, host_attempt_count: 1, generation_present: false}; + const input = {...p, request, report_launch_recovery: proof, + report_window: {window_id: prior.report_window_id, subscription_revision: prior.report_subscription_revision, + goal_id: p.actor.goal_id, agent_id: p.actor.agent_id}, + report_subscription: {enabled: true, profile_preset: "portfolio-progress", + route_ref: `steward-private:${p.binding.binding_id}:${p.source}`, effective_revision: prior.report_subscription_revision}, + report_todo: {todo_id: prior.todo_id, target_key: prior.report_window_id, + claimed_by: "steward", action_kind: "generate_periodic_report", status: "open"}}; + assert.equal(admitStewardManagedTurn(input).observation_only, false); + for (const invalid of [ + {...input, report_launch_recovery: null}, + {...input, request: prior}, + {...input, request: {...request, resume_turn_key: "sha256:" + "b".repeat(64)}}, + {...input, report_launch_recovery: {...proof, turn_key: "sha256:" + "b".repeat(64)}}, + {...input, report_launch_recovery: {...proof, host_attempt_count: 2}}, + {...input, report_launch_recovery: {...proof, generation_present: true}}, + ...["failed", "running", "interrupted"].map(status => ({...input, + report_launch_recovery: {...proof, original: {...source, status}}})), + {...input, report_launch_recovery: {...proof, original: {...source, delta_count: 1}}}, + {...input, report_launch_recovery: {...proof, original: {...source, upstream_turn_id: "actual-start"}}}, + {...input, report_launch_recovery: {...proof, original: {...source, + loopx_request: {...prior, todo_id: "todo_other"}}}}, + ...["transport_lost", "goal_launch_not_allowed", "unknown"].map(error_code => ({...input, + report_launch_recovery: {...proof, original: {...source, response: {native_loopx: {...result, error_code}}}}})), + {...input, report_launch_recovery: {...proof, original: {...source, response: {native_loopx: + {...result, effects: {state_written: true}}}}}}, + ...["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].map(key => ({...input, + report_launch_recovery: {...proof, inspection: {...inspection, [key]: false}}})), + {...input, report_launch_recovery: {...proof, inspection: {...inspection, completed_phases: ["host_execute"]}}}, + {...input, report_launch_recovery: {...proof, inspection: {...inspection, + recorded_effects: {...inspection.recorded_effects, quota_spent: null}}}}, + {...input, report_launch_recovery: {...proof, inspection: {...inspection, + recorded_effects: {...inspection.recorded_effects, host_invoked: true}}}}, + {...input, report_subscription: {...input.report_subscription, enabled: false}}, + {...input, goal_active: false}, + {...input, queued_turn_count: 1}, + ]) assert.throws(() => admitStewardManagedTurn(invalid)); +}); diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 572855131c..9fbfa494cc 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -162,6 +162,80 @@ def collect(self, *args, **kwargs): assert store.load_session(sid)["upstream_thread_id"] +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("reject_recovery", [False, True]) +def test_original_pump_resumes_exact_prelaunch_rejection_once_without_replaying_unknown_work(steward, tmp_path, monkeypatch, provider, reject_recovery): + from loopx.chat_agent import CodexChatAgentSession + from loopx.control_plane.turn_driver.journal_store import turn_journal_path + store, runtime, bot, transport, binding, capture, _ = steward + sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + install_model(steward, tmp_path, sid) + history, calls = store.messages(sid), capture.read_bytes() + write = CodexChatAgentSession._write + + def concurrent_registry_update(self, payload, **kwargs): + if payload.get("method") == "turn/start": + # The genuine legacy witness check rejects immediately before + # transport publication, after source collection and native plan. + runtime.registry_path.write_bytes(runtime.registry_path.read_bytes() + b"\n") + return write(self, payload, **kwargs) + + with monkeypatch.context() as fault: + fault.setattr(CodexChatAgentSession, "_write", concurrent_registry_update) + first = pump_steward_monitors(runtime)[0] + rejected = finish(runtime, first) + failure = rejected["response"]["native_loopx"] + assert failure["error_code"] == "stale_goal_authority", rejected + assert rejected["delta_count"] == 0 and rejected["upstream_turn_id"] is None + native = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=failure["resume_turn_key"]) + assert json.loads(native.read_text())["host_attempt_count"] == 1 + assert json.loads(native.read_text())["completed_phases"] == [] + original_turn = store._turn_path(sid, first["turn_id"]).read_bytes() + assert not any(json.loads(line).get("method") == "turn/start" for line in capture.read_bytes()[len(calls):].splitlines()) + assert transport.steward_reports.reconcile() == 0 + # A new host adapter after idle/restart uses the original producer and + # canonical journal, rather than replaying the failed outer Chat request. + runtime._steward_monitor_next_check.clear() + with monkeypatch.context() as retry_fault: + if reject_recovery: + retry_fault.setattr(CodexChatAgentSession, "_write", concurrent_registry_update) + retry = pump_steward_monitors(runtime)[0] + assert retry["turn_id"] != first["turn_id"] + assert retry["loopx_request"]["turn_instance_id"] == first["loopx_request"]["turn_instance_id"] + assert retry["loopx_request"]["resume_turn_key"] == failure["resume_turn_key"] + completed = finish(runtime, retry) + assert completed["status"] == "completed", {key: completed.get(key) for key in ("status", "error", "error_code", "response")} + if reject_recovery: + assert completed["response"]["native_loopx"]["error_code"] == "stale_goal_authority" + before = capture.read_bytes(), list(bot.writes), native.read_bytes() + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == 0 + assert (capture.read_bytes(), bot.writes, native.read_bytes()) == before + assert store._turn_path(sid, first["turn_id"]).read_bytes() == original_turn + return + assert completed["response"]["native_loopx"].get("status") == "committed", completed + assert completed["response"]["native_loopx"]["resume_turn_key"] == failure["resume_turn_key"] + assert store._turn_path(sid, first["turn_id"]).read_bytes() == original_turn + assert json.loads(native.read_text())["host_attempt_count"] == 2 + # The existing no-model settlement recovery supplies the original return + # reader with the committed native event; publication remains separate. + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == 1 + assert reports.health(binding["binding_id"])["publication_verified"] + assert store.messages(sid)[:len(history)] == history + tail = [json.loads(line) for line in capture.read_bytes()[len(calls):].splitlines()] + assert sum(row.get("method") == "turn/start" for row in tail) == 1 + sent, calls = list(bot.writes), capture.read_bytes() + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == 0 + assert bot.writes == sent and capture.read_bytes() == calls + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_report_source_revoked_during_collection_never_enters_native_execution(steward, tmp_path, monkeypatch, provider): _, runtime, bot, _, binding, capture, _ = steward