fix(chat): resume proven prelaunch report refusals

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

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:

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;

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));
+});