fix(chat): continue reports after host-only admission stops

docs/product/periodic-reports.md

@@ -282,10 +282,17 @@ or historical results.
 
 If the native guard refuses the exact report Task before execution because a
 semantic replan is due, the original idle worker can admit one correlated
-continuation of that window. It first runs the original Agent's native replan,
-without choosing another business Task. The canonical writeback preview must
-independently accept its semantic delta before writeback and quota settlement;
-only a committed replan permits a fresh admission of the retained report Task.
+continuation of that window. It rechecks native admission without choosing
+another business Task. When governance is still due, the canonical writeback
+preview must independently accept the original Agent's semantic delta before
+writeback and quota settlement. When the fresh guard no longer owes governance,
+the host-only admission stop can request a new exact admission of the original
+report Task without a governance model call. A different advisory Task in that
+probe is never executed and grants no authority over the report. A retained stop
+from the earlier host can receive one separately correlated continuation only
+when its canonical journal proves that exact host-only path and no model work;
+missing output alone is insufficient. Both paths retain the original native
+report identity, task validator, report window and publication owner.
 Ordinary delivery remains withheld during governance. The original return
 reader then verifies the full report through its existing publication owner.
 The refused Chat request remains in history. Unknown execution, partial effects,

loopx/capabilities/native_chat/steward_managed_mode.py

@@ -357,7 +357,13 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str,
                 client_id += ".admission-retry"
                 retry = self.store.turn_for_client(session_id, client_id)
                 if retry is not None:
-                    return retry, False
+                    selection_recovery = StewardTurnHost(controller).report_selection_recovery(existing, probe_turn=retry)
+                    if selection_recovery is None or "probe" not in selection_recovery:
+                        return retry, False
+                    client_id = "managed." + operation_id + ".admission-resume"
+                    resumed = self.store.turn_for_client(session_id, client_id)
+                    if resumed is not None:
+                        return resumed, False
             commit = managed_session_goal_commit(store=self.store,
                 registry_path=controller.registry_path, session_id=session_id)
             with commit():

loopx/capabilities/native_chat/steward_turn.py

@@ -101,7 +101,8 @@ def scope_id(ids):
             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_selection_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None:
+    def report_selection_recovery(self, original: dict[str, Any], *,
+                                  probe_turn: dict[str, Any] | None = None) -> dict[str, Any] | None:
         """The typed owner qualifies this observation before any new worker."""
         result = (original.get("response") or {}).get("native_loopx") or {}
         refusal = result.get("selection_rejection")
@@ -114,9 +115,28 @@ def report_selection_recovery(self, original: dict[str, Any]) -> dict[str, Any]
         request = original["loopx_request"]
         reports, owner = report_service(self.controller, request["actor"])
         _, window = reports.current_window(owner)
-        return {"original": original, "generation_present": window is None
+        proof = {"original": original, "generation_present": window is None
             or window["window_id"] != request["report_window_id"]
             or (attempt_directory(reports.root, window) / "private-generation.json").exists()}
+        if probe_turn is not None:
+            from ...control_plane.turn_driver.executor import inspect_loopx_turn_journal
+            from ...control_plane.turn_driver.journal_store import (
+                load_turn_journal, turn_journal_path,
+            )
+            actor = request["actor"]
+            keys = {row["payload"].get("resume_turn_key")
+                for row in self.store.events_after(actor["session_id"], probe_turn["turn_id"], None)
+                if row.get("kind") == "native_loopx.replan_result"}
+            if len(keys) != 1 or not isinstance(next(iter(keys)), str):
+                return None
+            key = keys.pop()
+            # Transport canonical observations to the shared typed admission
+            # owner. Neither a missing model event nor Python prose decides.
+            proof["probe"] = {"turn": probe_turn,
+                "inspection": inspect_loopx_turn_journal(reports.root, goal_id=actor["goal_id"],
+                    agent_id=actor["agent_id"], turn_key=key),
+                "journal": load_turn_journal(turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=key))}
+        return proof
 
     def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None:
         """Read the original typed launch refusal and its canonical transaction.
@@ -202,7 +222,9 @@ def _task_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, rec
         retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"],
             "--todo-id", request["todo_id"]]).get("lease")
         version = retained["version"] if retained else 0
-        suffix = ".validation-recovery." + str(version) if recovery else ".admission-retry" if request.get("resume_turn_key") or admission_retry else ""
+        suffix = (".validation-recovery." + str(version) if recovery else
+            ".admission-retry." + turn_id if admission_retry else
+            ".admission-retry" if request.get("resume_turn_key") else "")
         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"] + suffix, "--ttl-seconds", "3600",
@@ -327,8 +349,16 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None
                 client_id = native_steward_client_id(actor, turn_instance_id + ".admission-retry")
                 retry = self.store.turn_for_client(actor["session_id"], client_id)
                 if retry is not None:
-                    return retry, False
-                if work_kind == "report":
+                    if work_kind != "report":
+                        return retry, False
+                    selection_recovery = self.report_selection_recovery(existing, probe_turn=retry)
+                    if selection_recovery is None or "probe" not in selection_recovery:
+                        return retry, False
+                    client_id = native_steward_client_id(actor, turn_instance_id + ".admission-resume")
+                    resumed = self.store.turn_for_client(actor["session_id"], client_id)
+                    if resumed is not None:
+                        return resumed, False
+                if work_kind == "report" and selection_recovery is None:
                     selection_recovery = self.report_selection_recovery(existing)
                     if selection_recovery is None:
                         launch_recovery = self._report_launch_recovery(existing)
@@ -383,6 +413,19 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any,
         original = self.store.turn_for_client(session_id, original_client) if request.get("work_kind") == "report" else None
         selection_recovery = (self.report_selection_recovery(original)
             if original and original["turn_id"] != turn_id else None)
+        resumed_client = (original_client + ".admission-resume"
+            if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0"
+            else native_steward_client_id(request["actor"], request["turn_instance_id"] + ".admission-resume"))
+        if selection_recovery and turn["client_turn_id"] == resumed_client:
+            retry_client = (original_client + ".admission-retry"
+                if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0"
+                else native_steward_client_id(request["actor"], request["turn_instance_id"] + ".admission-retry"))
+            retry = self.store.turn_for_client(session_id, retry_client)
+            if retry is None:
+                raise ValueError("the original admission probe is unavailable")
+            selection_recovery = self.report_selection_recovery(original, probe_turn=retry)
+            if selection_recovery is None:
+                raise ValueError("the original admission probe has no canonical journal")
         admitted = self._admit(request, turn=turn, selection_recovery=selection_recovery)
         actor = admitted["actor"]
         local_session = actor.get("schema_version") == "loopx_managed_session_actor_v0"
@@ -516,7 +559,7 @@ def launch():
         result: dict[str, Any] = {}
         with ExitStack() as managed_leases, (self._task_execution_lease(request, turn_id,
                 admission_retry=bool(selection_recovery)) if report_work else nullcontext(None)) as report_lease:
-            if selection_recovery:
+            if selection_recovery and "probe" not in selection_recovery:
                 # Re-enter the native guard without selecting another business
                 # Todo. Only its signed replan packet may invoke the original
                 # Agent. The native executor owns semantic validation/settlement.
@@ -531,10 +574,17 @@ def launch():
                     native_host_runner=run, native_session_binding_resolver=session_binding)
                 self.store.append_event(session_id, turn_id, kind="native_loopx.replan_result", payload=governance)
                 if governance.get("status") != "committed" or governance.get("result_kind") != "replan_required":
-                    retained = selection_recovery["original"]["response"]["native_loopx"]
-                    self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=retained)
-                    return {"schema_version": "loopx_chat_agent_response_v0", "message": "原生重规划尚未完成；保留原报告及执行证据。",
-                        "proposals": [], "gate": None, "native_loopx": retained}
+                    probe = self.report_selection_recovery(original, probe_turn=self.store.load_turn(session_id, turn_id))
+                    try:
+                        if probe is None or "probe" not in probe:
+                            raise ValueError("no canonical admission probe")
+                        self._admit(request, turn=self.store.load_turn(session_id, turn_id), selection_recovery=probe)
+                    except (OSError, ValueError, RuntimeError):
+                        retained = selection_recovery["original"]["response"]["native_loopx"]
+                        self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=retained)
+                        return {"schema_version": "loopx_chat_agent_response_v0", "message": "原生重规划尚未完成；保留原报告及执行证据。",
+                            "proposals": [], "gate": None, "native_loopx": retained}
+                    selection_recovery = probe
                 governance_work = False
             if report_work:
                 # Portfolio collection can take minutes and legitimately see

loopx/control_plane/collaboration/steward_managed_mode.ts

@@ -11,6 +11,7 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
 import {goalAgentConversationIdentity, resolveConversationScope} from "./conversation_scope.ts";
 import {readTurnSelectionRejection} from "../turn_driver/selection_rejection.ts";
+import {EffectiveAction} from "../quota/effective_action.generated.ts";
 
 /** A native refusal before execution can owe governance, never a business replay. */
 function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, sessionId: unknown): void {
@@ -40,6 +41,55 @@ function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, s
       || proof.generation_present !== false) {
     throw new EffectRuntimeRequestError("report selection recovery requires the original proven pre-execution replan refusal");
   }
+  if (proof.probe == null) return;
+  // The original host returned before calling its model when a fresh guard
+  // no longer owed governance. Absence of output alone cannot prove that.
+  const probe = requireJsonObject(proof.probe, "original host admission probe");
+  const observed = requireJsonObject(probe.turn, "original probe Chat Turn");
+  const journal = requireJsonObject(probe.journal, "canonical probe journal");
+  const inspection = requireJsonObject(probe.inspection, "canonical probe inspection");
+  const recovery = requireJsonObject(inspection.recovery_decision, "probe recovery decision");
+  const recorded = requireJsonObject(inspection.recorded_effects, "probe effects");
+  const plan = requireJsonObject(journal.plan, "probe plan");
+  const route = requireJsonObject(plan.route, "probe route");
+  const envelope = requireJsonObject(plan.turn_envelope, "probe envelope");
+  const action = requireJsonObject(envelope.action, "probe action");
+  const selected = requireJsonObject(action.selected_todo, "probe selected task");
+  const identity = requireJsonObject(requireJsonObject(requireJsonObject(plan.transaction, "probe transaction")
+    .settlement_plan, "probe settlement").identity, "probe identity");
+  const host = requireJsonObject(journal.host, "probe host");
+  const stopped = requireJsonObject(journal.host_result, "probe host result");
+  const runningHere = params.turn != null && requireJsonObject(params.turn, "current Chat Turn").turn_id === observed.turn_id
+    && ["starting", "running"].includes(String(observed.status));
+  if (observed.session_id !== sessionId || typeof observed.turn_id !== "string" || !/^[a-f0-9]{32}$/.test(observed.turn_id)
+      || observed.turn_id === original.turn_id
+      || observed.origin !== "native_loopx" || observed.loopx_execution !== true
+      || observed.delta_count !== 0 || observed.upstream_turn_id !== null || observed.error_code != null || observed.error != null
+      || JSON.stringify(observed.loopx_request) !== JSON.stringify(request)
+      || !(runningHere || observed.status === "completed"
+        && JSON.stringify(requireJsonObject(observed.response, "probe response").native_loopx) === JSON.stringify(result))
+      || journal.schema_version !== "loopx_turn_journal_v0" || !ENVELOPED_SHA256_PATTERN.test(String(journal.turn_key))
+      || journal.status !== "stopped" || journal.goal_id !== envelope.goal_id
+      || journal.host_attempt_count !== 1 || journal.result_kind !== "wait"
+      || JSON.stringify(journal.completed_phases) !== JSON.stringify(["host_execute", "typed_result", "validation"])
+      || host.kind !== "codex-app-server" || host.executable !== "built-in"
+      || stopped.schema_version !== "loopx_turn_result_v0" || stopped.turn_key !== journal.turn_key
+      || stopped.result_kind !== "wait" || stopped.classification !== "report_replan_not_admitted"
+      || inspection.schema_version !== "loopx_turn_journal_inspection_v1" || inspection.ok !== true
+      || inspection.goal_matches !== true || inspection.owner_matches !== true || inspection.turn_key_matches !== true
+      || inspection.journal_consistent !== true || inspection.journal_status !== "stopped"
+      || recovery.action !== "return_existing" || recovery.reinvoke_host !== false
+      || recorded.host_invoked !== true || ["state_written", "quota_spent", "scheduler_acknowledged"].some(key => recorded[key] !== false)
+      || route.kind !== "ready_for_host" || route.effective_action !== EffectiveAction.NORMAL_RUN
+      || envelope.effective_action !== EffectiveAction.NORMAL_RUN || envelope.replan_action_packet != null
+      || action.must_attempt !== true || action.delivery_allowed !== true || selected.todo_id !== identity.todo_id
+      || identity.goal_id !== envelope.goal_id || identity.agent_id !== envelope.agent_id
+      || identity.goal_id !== requireJsonObject(request.actor, "report actor").goal_id
+      || identity.agent_id !== requireJsonObject(request.actor, "report actor").agent_id
+      || typeof identity.todo_id !== "string" || !/^todo_[A-Za-z0-9_-]{1,100}$/.test(identity.todo_id)
+      || identity.turn_instance_id !== request.turn_instance_id + ".replan") {
+    throw new EffectRuntimeRequestError("report continuation requires its canonical host-only admission stop; retain all model or unknown work");
+  }
 }
 
 /** Registration and current Goal lifetime are observed by the host, never by