diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 4411040095..544f5aa985 100644 --- a/docs/product/periodic-reports.md +++ b/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 diff --git a/loopx/capabilities/native_chat/steward_reports.py b/loopx/capabilities/native_chat/steward_reports.py index 780090d09b..4e41e07626 100644 --- a/loopx/capabilities/native_chat/steward_reports.py +++ b/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"], diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 637eb8352e..e3764f4c9d 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/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 diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index f2f13622f5..aa61b7740c 100644 --- a/loopx/cli_commands/turn.py +++ b/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") diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 61d44be49f..8d48263823 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/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) diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index aa9a5d3428..dc2e205a5e 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/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"); } diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 990854e75b..d35e7e6b21 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -78,6 +78,20 @@ test("a model report requires the exact live calendar window and original task, journal_consistent: true, journal_status: "failed", recovery_decision: {action: "continue", reinvoke_host: false, resume_from: "quota_spend"}}}; assert.equal(admitStewardManagedTurn(recovery).observation_only, false); + const validation = {...input, report_settlement_recovery: {...recovery.report_settlement_recovery, + completed_phases: ["host_execute", "typed_result"], recovery_decision: + {action: "continue", reinvoke_host: false, resume_from: "validation"}}}; + assert.equal(admitStewardManagedTurn(validation).observation_only, false); + for (const invalid of [ + {...validation, preparing_model_context: true}, + {...validation, report_todo: {...todo, todo_id: "another-task"}}, + {...validation, report_subscription: {...subscription, enabled: false}}, + {...validation, report_settlement_recovery: {...validation.report_settlement_recovery, journal_status: "in_progress"}}, + ...[[], ["host_execute"], ["typed_result", "host_execute"]].map(completed_phases => + ({...validation, report_settlement_recovery: {...validation.report_settlement_recovery, completed_phases}})), + {...validation, report_settlement_recovery: {...validation.report_settlement_recovery, recovery_decision: + {action: "continue", reinvoke_host: true, resume_from: "validation"}}}, + ]) assert.throws(() => admitStewardManagedTurn(invalid)); for (const invalid of [ {...recovery, report_settlement_recovery: {...recovery.report_settlement_recovery, journal_consistent: false}}, {...recovery, report_todo: {...recovery.report_todo, completion_turn_key: "another-original-turn"}}, diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 9fbfa494cc..44d65bea90 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -87,6 +87,104 @@ def install_model(steward, tmp_path, sid, *, result_kind="validated_completion") runtime.adapters.pop(sid).close_session() # Resume the same native thread with the synthetic response. +def embedded_report_python(tmp_path, monkeypatch): + """A real child interpreter without the embedding host's LoopX bootstrap.""" + import subprocess + import sys + import os + from types import SimpleNamespace + from loopx.capabilities.native_chat import steward_reports + folder = tmp_path / "embedded-python" + subprocess.run([sys.executable, "-m", "venv", "--without-pip", str(folder)], check=True) + python = folder / ("Scripts/python.exe" if os.name == "nt" else "bin/python") + missing = subprocess.run([str(python), "-I", "-c", "import loopx"], + cwd=tmp_path, capture_output=True, text=True) + assert missing.returncode != 0 and "ModuleNotFoundError" in missing.stderr + monkeypatch.setattr(steward_reports, "sys", SimpleNamespace(executable=str(python))) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_embedded_host_validates_report_without_a_second_python_installation(steward, tmp_path, monkeypatch, provider): + _, runtime, _, transport, binding, _, _ = steward + sid, _, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + install_model(steward, tmp_path, sid) + embedded_report_python(tmp_path, monkeypatch) + result = finish(runtime, pump_steward_monitors(runtime)[0]) + assert result["response"]["native_loopx"].get("status") == "committed", result + assert transport.steward_reports.reconcile() == 1 + assert reports.health(binding["binding_id"])["publication_verified"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("stop", [None, "pause", "tamper"]) +def test_original_frozen_report_recovers_validation_without_repeating_model_or_business(steward, tmp_path, monkeypatch, provider, stop): + from loopx.cli_commands import turn as turn_command + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + from loopx.control_plane.turn_driver.journal_store import find_loopx_turn_key_by_settlement_identity, 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) + embedded_report_python(tmp_path, monkeypatch) + command = turn_command.handle_turn_command + + def legacy_validator(*args, **kwargs): + # Seed the exact historical command failure, not a fabricated journal. + kwargs.pop("native_task_validator", None) + return command(*args, **kwargs) + + with monkeypatch.context() as historical: + historical.setattr(turn_command, "handle_turn_command", legacy_validator) + original = pump_steward_monitors(runtime)[0] + failed = finish(runtime, original) + native = failed["response"]["native_loopx"] + assert native["result_kind"] == "validation_failed" and native["receipt"]["next_phase"] == "validation" + todo = next(row for row in StewardMonitor(runtime).todos(actor) if row["todo_id"] == original["loopx_request"]["todo_id"]) + assert todo["status"] == "open" + key = find_loopx_turn_key_by_settlement_identity(reports.root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], todo_id=todo["todo_id"], turn_instance_id=original["loopx_request"]["turn_instance_id"]) + path = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=key) + journal = json.loads(path.read_text()) + assert journal["completed_phases"] == ["host_execute", "typed_result"] + window = read_cadence_journal(runtime_root=reports.root, goal_id=actor["goal_id"])["window"] + folder = attempt_directory(reports.root, window) + generation = validate_private_generation(folder) + body, history, calls, writes = (folder / "report.md").read_bytes(), store.load_turn(sid, original["turn_id"]), capture.read_bytes(), list(bot.writes) + assert transport.steward_reports.reconcile() == 0 + if stop == "pause": + preview = reports.configure(binding["binding_id"], operation="pause") + reports.configure(binding["binding_id"], operation="pause", execute=True, expected_plan_id=preview["plan"]["plan_id"]) + elif stop == "tamper": + (folder / "report.md").write_bytes(body + b"altered") + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert capture.read_bytes() == calls and bot.writes == writes + if stop: + assert transport.steward_reports.reconcile() == 0 + assert json.loads(path.read_text()) == journal + assert not (folder / "private-delivery.json").exists() + return + recovered = json.loads(path.read_text()) + assert recovered["status"] == "committed", recovered + assert recovered["turn_key"] == key and recovered["host_attempt_count"] == journal["host_attempt_count"] + assert store.load_turn(sid, original["turn_id"]) == history + task = next(row for row in StewardMonitor(runtime).todos(actor) if row["todo_id"] == todo["todo_id"]) + assert task["status"] == "done" and task["completion_turn_key"] == original["loopx_request"]["turn_instance_id"] + events = [row["payload"] for row in store.events_after(sid, original["turn_id"], None) + if row["kind"] == "native_loopx.result" and row["payload"].get("status") == "committed"] + assert len(events) == 1 and events[0]["recovery"]["actual"]["host_invoked"] is False + assert transport.steward_reports.reconcile() == 1 + assert (folder / "report.md").read_bytes() == body + delivery = json.loads((folder / "private-delivery.json").read_text()) + proof = delivery["deliveries"]["periodic_report." + window["window_id"]] + assert proof["verified"] and proof["text"] == body.decode().strip() + cursor = read_periodic_report_publication_cursor(runtime_root=reports.root, goal_id=actor["goal_id"], agent_id=actor["agent_id"]) + assert cursor["generation_id"] == generation["generation_id"] and cursor["covered_until"] == window["due_at"] + sent = list(bot.writes) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) and transport.steward_reports.reconcile() == 0 + assert capture.read_bytes() == calls and bot.writes == sent + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("result_kind", ["validated_progress", "validated_completion"]) def test_original_due_report_freezes_once_and_returns_full_body_after_native_settlement(steward, tmp_path, monkeypatch, provider, result_kind): diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index 229b14aea3..689d0bca2d 100644 --- a/tests/test_native_steward_turn_host.py +++ b/tests/test_native_steward_turn_host.py @@ -15,6 +15,20 @@ steward = steward_fixture +def test_native_task_validator_cannot_replace_a_cli_host(tmp_path): + from loopx.cli import build_parser + from loopx.cli_commands.turn import handle_turn_command + args = build_parser().parse_args(["turn", "run-once", "--goal-id", "example", "--agent-id", "worker", + "--host", "codex-cli", "--project", str(tmp_path), "--execute"]) + payload, calls = {}, [] + handle_turn_command(args, registry_path=tmp_path / "registry.json", + runtime_root_arg=str(tmp_path / "runtime"), output_format=lambda *_a, **_kw: "json", + print_payload=lambda value, *_a: payload.update(value), + native_task_validator=lambda *_a: calls.append("called")) + assert not payload["ok"] and "requires its original host connection" in payload["error"] + assert calls == [] and not (tmp_path / "registry.json").exists() + + def cli(runtime, *args): out = io.StringIO() with contextlib.redirect_stdout(out):