fix(chat): recover frozen reports after native validation failure

docs/product/periodic-reports.md

@@ -316,6 +316,22 @@ 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.
 
+Frozen generation is checked independently inside the original host using the
+same installed report provider. An embedded Python host's bootstrap does not
+imply that its bare child interpreter can import LoopX; no second installation
+or ambient import path is required. Ordinary CLI command validators are unchanged.
+
+If the original model has returned and the frozen generation validates, the
+original pump can resume a failed native **validation** phase without invoking
+the model again. The typed owner requires the same live subscription, actor,
+window and open generation Todo, the canonical cached host/result phases, and
+an explicit no-host recovery decision. The native driver revalidates the frozen
+artifacts and current authorization, reacquires the original task lease and
+performs ordinary completion and settlement. A paused subscription, changed
+identity or grant, invalid generation or unknown model effect stays blocked.
+The historical Chat response is retained; only the native committed receipt
+allows the existing original-audience sender to proceed.
+
 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_reports.py

@@ -263,12 +263,13 @@ def frontier(self, binding: dict[str, Any]) -> tuple[dict[str, Any] | None, bool
                 "--target-key", window["window_id"], "--continuation-policy", "same_agent_non_delivery",
                 "--execute"])
             rows = [row for row in StewardMonitor(self.controller).todos(actor) if row.get("target_key") == window["window_id"]]
-        if len(rows) == 1 and rows[0]["status"] == "done":
+        folder = attempt_directory(self.root, window)
+        if len(rows) == 1 and (rows[0]["status"] == "done"
+                or rows[0]["status"] == "open" and (folder / "private-generation.json").exists()):
             StewardTurnHost(self.controller).recover_report_settlement(actor, window, rows[0])
             return None, False
         if len(rows) != 1 or rows[0]["status"] != "open":
             return None, False
-        folder = attempt_directory(self.root, window)
         validator = [sys.executable, "-c", "from pathlib import Path; from loopx.capabilities.periodic_report.private_generation import validate_private_generation; "
             + f"validate_private_generation(Path({str(folder)!r}))"]
         return StewardTurnHost(self.controller).submit(binding["binding_id"],

loopx/capabilities/native_chat/steward_turn.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
-from contextlib import contextmanager, ExitStack
+from contextlib import contextmanager, ExitStack, nullcontext
 import hashlib
 import json
 from pathlib import Path
@@ -125,6 +125,74 @@ def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] |
             "host_attempt_count": (journal or {}).get("host_attempt_count"),
             "generation_present": window is None or (attempt_directory(root, window) / "private-generation.json").exists()}
 
+    def _report_validator(self, request: Mapping[str, Any]):
+        """Validate with the owning installed package, outside model execution.
+
+        An embedded host may import LoopX through its bootstrap rather than the
+        Python interpreter's site-packages. Its bare child interpreter is not
+        an independent installation of the same package.
+        """
+        def validate(_plan: Mapping[str, Any], _result: Mapping[str, Any]) -> dict[str, Any]:
+            from ..periodic_report.private_generation import attempt_directory, validate_private_generation
+            from .steward_reports import StewardReports
+            reports = StewardReports(self.controller)
+            binding_id = request["actor"]["goal_id"].removeprefix("steward-agent-")
+            try:
+                subscription, window = reports.current_window(binding_id)
+                if (not subscription or not window or window["window_id"] != request["report_window_id"]
+                        or window["subscription_revision"] != request["report_subscription_revision"]):
+                    raise ValueError("the original report window changed")
+                validate_private_generation(attempt_directory(reports.root, window))
+                if not reports.delivery_current(binding_id, window):
+                    raise ValueError("the original report authorization changed")
+            except (OSError, ValueError, RuntimeError):
+                return {"status": "failed", "validator_kind": "callback",
+                    "summary": "independent original report validation failed", "recovery_kind": "repair_required"}
+            return {"status": "passed", "validator_kind": "callback",
+                "summary": "independent original report validation passed"}
+        return validate
+
+    def _report_lease_proof(self, request: Mapping[str, Any], lease: Mapping[str, Any] | None) -> dict[str, Any]:
+        from .steward_commands import native_command
+        observed = native_command(self.controller, ["task-lease", "inspect", "--goal-id", request["actor"]["goal_id"],
+            "--todo-id", request["todo_id"]]).get("lease")
+        if (not lease or not observed or observed.get("status") != "active"
+                or any(observed.get(key) != lease.get(key) for key in
+                    ("owner", "idempotency_key", "lease_epoch", "version"))):
+            raise ValueError("the original report execution lease changed")
+        return {"task_lease_idempotency_key": observed["idempotency_key"],
+            "task_lease_expected_version": observed["version"]}
+
+    @contextmanager
+    def _report_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, recovery: bool = False):
+        """Both generation and cached validation use the original Todo owner."""
+        from .steward_commands import native_command
+        actor = request["actor"]
+        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") 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",
+            "--expected-version", str(version)])["lease"]
+        try:
+            yield lease
+        finally:
+            # Completion retires this lease. Never release a later owner's.
+            try:
+                retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"],
+                    "--todo-id", request["todo_id"]]).get("lease")
+                if retained and retained.get("status") == "active":
+                    proof = self._report_lease_proof(request, lease)
+                    native_command(self.controller, ["task-lease", "release", "--goal-id", actor["goal_id"],
+                        "--todo-id", request["todo_id"], "--owner", actor["agent_id"],
+                        "--idempotency-key", proof["task_lease_idempotency_key"],
+                        "--expected-version", str(proof["task_lease_expected_version"])])
+            except (OSError, ValueError, RuntimeError):
+                self.store.append_event(actor["session_id"], turn_id, kind="native_loopx.report_lease_release_unconfirmed",
+                    payload={"todo_id": request["todo_id"], "lease_epoch": lease["lease_epoch"]})
+
     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.
 
@@ -169,11 +237,15 @@ def no_model(*_args):
                 "--host", "generic-cli", "--resume-turn-key", key, "--retry-failed-turn", "--execute"])
             args.host, args.session_scope = "codex-app-server", "agent"
             result: dict[str, Any] = {}
-            handle_turn_command(args, registry_path=self.controller.registry_path,
-                runtime_root_arg=str(reports.root), output_format=lambda *_a, **_kw: "json",
-                print_payload=lambda value, *_a: result.update(value), native_host_runner=no_model,
-                native_session_binding_resolver=lambda _envelope: {"schema_version": "loopx_turn_session_binding_v0",
-                    "goal_id": actor["goal_id"], "agent_id": actor["agent_id"]})
+            with (self._report_execution_lease(request, original["turn_id"], recovery=True)
+                    if todo["status"] == "open" else nullcontext(None)) as lease:
+                handle_turn_command(args, registry_path=self.controller.registry_path,
+                    runtime_root_arg=str(reports.root), output_format=lambda *_a, **_kw: "json",
+                    print_payload=lambda value, *_a: result.update(value), native_host_runner=no_model,
+                    native_session_binding_resolver=lambda _envelope: {"schema_version": "loopx_turn_session_binding_v0",
+                        "goal_id": actor["goal_id"], "agent_id": actor["agent_id"]},
+                    native_task_validator=self._report_validator(request),
+                    **({"native_completion_lease_resolver": lambda: self._report_lease_proof(request, lease)} if lease else {}))
             # Retain the historical outer failure. The actual native owner
             # supplies the new event; no status or publication proof is forged.
             self.store.append_event(actor["session_id"], original["turn_id"], kind="native_loopx.result", payload=result)
@@ -286,15 +358,7 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any,
         report_context = ""
 
         def report_lease_proof():
-            from .steward_commands import native_command
-            observed = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"],
-                "--todo-id", request["todo_id"]]).get("lease")
-            if (not report_lease or not observed or observed.get("status") != "active"
-                    or any(observed.get(key) != report_lease.get(key) for key in
-                        ("owner", "idempotency_key", "lease_epoch", "version"))):
-                raise ValueError("the original report execution lease changed")
-            return {"task_lease_idempotency_key": observed["idempotency_key"],
-                "task_lease_expected_version": observed["version"]}
+            return self._report_lease_proof(request, report_lease)
 
         def current():
             fresh = self.store.load_turn(session_id, turn_id)
@@ -370,16 +434,7 @@ def launch():
         # The kernel records the actual app-server host and independent profile.
         args.host, args.session_scope = "codex-app-server", "agent"
         result: dict[str, Any] = {}
-        if report_work:
-            from .steward_commands import native_command
-            retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"],
-                "--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"] + (".admission-retry" if request.get("resume_turn_key") else ""),
-                "--ttl-seconds", "3600",
-                "--expected-version", str(retained["version"] if retained else 0)])["lease"]
-        try:
+        with (self._report_execution_lease(request, turn_id) if report_work else nullcontext(None)) as report_lease:
             if report_work:
                 from .steward_reports import StewardReports
                 # Portfolio collection can take minutes and legitimately see
@@ -395,24 +450,8 @@ def launch():
                 output_format=lambda *_a, **_kw: "json",
                 print_payload=lambda value, *_a: result.update(value),
                 native_host_runner=run, native_session_binding_resolver=session_binding,
-                **({"native_completion_lease_resolver": report_lease_proof} if report_work else {}))
-        finally:
-            if report_lease:
-                # Completion atomically retires this lease. On a non-completion
-                # release only our unchanged claim, never a later owner's lease.
-                try:
-                    retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"],
-                        "--todo-id", request["todo_id"]]).get("lease")
-                    if retained and retained.get("status") == "active":
-                        proof = report_lease_proof()
-                        native_command(self.controller, ["task-lease", "release", "--goal-id", actor["goal_id"],
-                            "--todo-id", request["todo_id"], "--owner", actor["agent_id"],
-                            "--idempotency-key", proof["task_lease_idempotency_key"],
-                            "--expected-version", str(proof["task_lease_expected_version"])])
-                except (OSError, ValueError, RuntimeError):
-                    self.store.append_event(session_id, turn_id,
-                        kind="native_loopx.report_lease_release_unconfirmed",
-                        payload={"todo_id": request["todo_id"], "lease_epoch": report_lease["lease_epoch"]})
+                **({"native_completion_lease_resolver": report_lease_proof,
+                    "native_task_validator": self._report_validator(request)} if report_work else {}))
         self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=result)
         if request.get("work_kind") == "report" and result.get("status") == "committed":
             from ..periodic_report.private_generation import attempt_directory, validate_private_generation

loopx/cli_commands/turn.py

@@ -50,6 +50,7 @@ def handle_turn_command(
     native_host_runner: Callable[..., dict[str, Any]] | None = None,
     native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
     native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None,
+    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None,
 ) -> int | None:
     if args.command != "turn":
         return None
@@ -78,6 +79,8 @@ def handle_turn_command(
             raise ValueError("an app-server Turn requires its original host connection and session resolver")
         if native_completion_lease_resolver is not None and native_host_runner is None:
             raise ValueError("a native completion lease requires its original host connection")
+        if native_task_validator is not None and native_host_runner is None:
+            raise ValueError("a native task validator requires its original host connection")
         if getattr(args, "todo_id", None) is not None and (
             getattr(args, "resume_turn_key", None)
             or any(getattr(args, key, None) for key in ("resume_goal_id", "resume_agent_id", "resume_todo_id"))
@@ -344,6 +347,7 @@ def handle_turn_command(
                 native_host_runner=native_host_runner,
                 native_session_binding_resolver=native_session_binding_resolver,
                 native_completion_lease_resolver=native_completion_lease_resolver,
+                native_task_validator=native_task_validator,
             )
         else:
             raise ValueError("turn requires the `plan` or `run-once` subcommand")

loopx/cli_commands/turn_run_once.py

@@ -77,6 +77,7 @@ def execute_turn_run_once(
     native_host_runner: Callable[..., dict[str, Any]] | None = None,
     native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None,
     native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None,
+    native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None,
 ) -> dict[str, Any]:
     """Adapt one prepared CLI Turn plan to the bounded Turn driver."""
     execution_started = False
@@ -112,6 +113,9 @@ def execute_turn_run_once(
             if args.host_command_json:
                 raise ValueError(f"{args.host} does not accept --host-command-json")
             raw_argv = None
+        if native_task_validator is not None and args.host != "codex-app-server":
+            raise ValueError("a native task validator cannot replace a selected CLI validator")
+        task_validator = native_task_validator
         if args.validation_command_json:
             raw_validation_argv = json.loads(args.validation_command_json)
             if not isinstance(raw_validation_argv, list) or not all(
@@ -120,14 +124,13 @@ def execute_turn_run_once(
                 raise ValueError(
                     "--validation-command-json must be a JSON string array"
                 )
-            task_validator = build_loopx_turn_command_validator(
-                raw_validation_argv,
-                project=project,
-                timeout_seconds=args.validation_timeout_seconds,
-                failure_recovery_kind=args.validation_failure_kind,
-            )
-        else:
-            task_validator = None
+            if task_validator is None:
+                task_validator = build_loopx_turn_command_validator(
+                    raw_validation_argv,
+                    project=project,
+                    timeout_seconds=args.validation_timeout_seconds,
+                    failure_recovery_kind=args.validation_failure_kind,
+                )
         envelope = (
             payload.get("turn_envelope")
             if isinstance(payload.get("turn_envelope"), dict)

loopx/control_plane/collaboration/steward_managed_mode.ts

@@ -126,12 +126,17 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject {
     let settlementOnly = false;
     if (recovery) {
       const decision = requireJsonObject(recovery.recovery_decision, "native journal recovery decision");
+      const cachedGeneration = todo.status === "open" && recovery.journal_status === "failed"
+        && decision.action === "continue" && decision.resume_from === "validation"
+        && Array.isArray(recovery.completed_phases) && recovery.completed_phases.length === 2
+        && recovery.completed_phases[0] === "host_execute" && recovery.completed_phases[1] === "typed_result";
+      const completedGeneration = todo.status === "done" && todo.completion_turn_key === request.turn_instance_id
+        && (decision.action === "continue" && ["durable_writeback", "quota_spend", "scheduler_apply", "scheduler_ack"].includes(String(decision.resume_from))
+          || decision.action === "return_existing" && recovery.journal_status === "committed");
       settlementOnly = recovery.schema_version === "loopx_turn_journal_inspection_v1" && recovery.ok === true
         && recovery.goal_matches === true && recovery.owner_matches === true && recovery.turn_key_matches === true
         && recovery.journal_consistent === true && decision.reinvoke_host === false
-        && (decision.action === "continue" && ["durable_writeback", "quota_spend", "scheduler_apply", "scheduler_ack"].includes(String(decision.resume_from))
-          || decision.action === "return_existing" && recovery.journal_status === "committed")
-        && todo.status === "done" && todo.completion_turn_key === request.turn_instance_id;
+        && (cachedGeneration || completedGeneration);
       if (!settlementOnly || params.preparing_model_context === true) {
         throw new EffectRuntimeRequestError("original report recovery requires validated remaining effects without a model");
       }