From d0b61e06eec0f200f0c4c343bf1470e65d59e786 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 14 Sep 2026 03:36:12 +0800 Subject: [PATCH 1/4] feat(turn): consume retry_continuation as a managed step The pure Turn Loop Controller answers `wait` with a typed `retry_continuation` block when the host failed for a retryable reason, but nothing in production consumed it: the outer scheduler was left to infer retryability and backoff from prose. `loopx turn run-once` mints the record and stops. Add `loopx turn managed-step`, the first production consumer. It reads one canonical Turn journal, rebuilds its validated receipt through the existing `ValidatedTurnReceipt` contract, projects the current control-plane decision as a fresh `loopx_turn_envelope_v0`, and returns `loopx_turn_managed_step_v0` with the disposition and, on `wait`, the bounded same-Turn continuation. The command is read-only by construction: it never invokes a host, writes state, spends quota, sleeps, or mints a Turn. The Turn journal stays the sole authority for the attempt count and retry ceiling, so `--observed-attempt` and `--observed-max-attempts` are reconciled against it and refused on disagreement rather than adopted. A journal that is not a finished failed Turn, whose typed host failure is not retryable, or whose recovery plan is `blocked` is refused before the transition is reached. Carrying `--retry-failed-turn` / `--resume-turn-key` on the next `run-once` remains the caller's decision. Because the Turn decision chain is now shared, `run-once` and `managed-step` build their envelope through one module instead of two copies that could drift. The new hermetic smoke drives the whole loop through the public CLI with a fake dsh runner: a first attempt raising `insufficient_capacity` fails with a typed retryable failure and spends nothing, the managed step answers `wait(30s, 1/3)` without touching the ledger, and replaying that exact Turn commits `validated_progress` with exactly one quota slot, where a further replay reports `replayed` and does not spend again. Signed-off-by: song --- .../protocols/turn-loop-controller-v0.md | 44 +- ...loopx-turn-managed-step-self-heal-smoke.py | 418 ++++++++++++++++++ loopx/cli_commands/turn.py | 78 ++-- loopx/cli_commands/turn_decision.py | 202 +++++++++ loopx/cli_commands/turn_managed_step.py | 103 +++++ loopx/cli_commands/turn_registration.py | 56 +++ loopx/cli_commands/turn_rendering.py | 49 ++ loopx/control_plane/turn_driver/__init__.py | 14 +- .../control_plane/turn_driver/managed_step.py | 292 ++++++++++++ tests/test_loopx_turn_managed_step.py | 343 ++++++++++++++ 10 files changed, 1544 insertions(+), 55 deletions(-) create mode 100644 examples/loopx-turn-managed-step-self-heal-smoke.py create mode 100644 loopx/cli_commands/turn_decision.py create mode 100644 loopx/cli_commands/turn_managed_step.py create mode 100644 loopx/control_plane/turn_driver/managed_step.py create mode 100644 tests/test_loopx_turn_managed_step.py diff --git a/docs/reference/protocols/turn-loop-controller-v0.md b/docs/reference/protocols/turn-loop-controller-v0.md index 5cd584a02d..72192535f7 100644 --- a/docs/reference/protocols/turn-loop-controller-v0.md +++ b/docs/reference/protocols/turn-loop-controller-v0.md @@ -12,11 +12,14 @@ host-specific wake adapters, and operator presentation are later slices in the Turn Loop Controller plan. The controller is an exported transition API, not a loop implicitly started by -`loopx turn run-once`. The CLI does not currently call `decide_loop_disposition` -or persist a `BoundedTurnBudget`. Both `max_turns` and `completed_turns` must be -supplied by an integrating caller; there is no CLI or product default of three -Turns. The budget applies to continued `validated_progress` on the same Todo, -not a chain of completed Todos and not fine-grained planning mode. +`loopx turn run-once`. `loopx turn managed-step` calls `decide_loop_disposition` +for one already-journaled failed Turn and returns the typed answer without +executing anything, which is the first production consumer of the transition. +The CLI still does not persist a `BoundedTurnBudget`; both `max_turns` and +`completed_turns` must be supplied by an integrating caller, and there is no CLI +or product default of three Turns. The budget applies to continued +`validated_progress` on the same Todo, not a chain of completed Todos and not +fine-grained planning mode. ## Inputs @@ -172,6 +175,37 @@ with an open acceptance gap, a terminal/obsolete/incompatible selected todo, validated negative evidence, or two eligible turns without material progress all require replan rather than another delivery attempt. +## Managed Step Surface + +`loopx turn managed-step` is the CLI surface that consumes this transition for +one already-journaled Turn: + +```bash +loopx turn managed-step \ + --goal-id \ + --agent-id \ + --turn-key \ + --format json +``` + +It rebuilds the `ValidatedTurnReceipt` from the canonical Journal, projects the +current control-plane decision as a fresh `loopx_turn_envelope_v0`, and returns +`loopx_turn_managed_step_v0`: the disposition, its reason, the goal/agent/Todo +lineage, and, on `wait`, the typed `retry_continuation` block. + +The command is read-only and grants no authority of its own. It never launches +a host, writes state, spends quota, sleeps, or mints a Turn. On `wait` the +answer only describes the bounded backoff after which the outer scheduler may +wake the *same* Turn; carrying the existing `--retry-failed-turn` / +`--resume-turn-key` flags on the next `run-once` stays the caller's decision. + +The Turn Journal remains the sole authority for the attempt count and retry +ceiling. `--observed-attempt` and `--observed-max-attempts` are reconciled +against it and refused on disagreement, so a caller's bookkeeping can be +checked but never substituted. A Journal that is not a finished failed Turn, +whose typed host failure is not retryable, or whose recovery plan is `blocked` +is refused before the transition is reached. + ## Boundary The controller is a pure function. It must not invoke a model, sleep, mutate a diff --git a/examples/loopx-turn-managed-step-self-heal-smoke.py b/examples/loopx-turn-managed-step-self-heal-smoke.py new file mode 100644 index 0000000000..ed2ec603ae --- /dev/null +++ b/examples/loopx-turn-managed-step-self-heal-smoke.py @@ -0,0 +1,418 @@ +#!/usr/bin/env python3 +"""Managed-step demonstration: fail once on capacity, then self-heal same-Turn. + +This is the first production consumer of the controller's typed +``retry_continuation``. It drives the public CLI end to end, with no model and +no DeepSeek Harness SDK required: + +1. ``turn run-once`` against a hermetic fake dsh runner whose first attempt + raises ``provider_capacity``. The Turn must fail with a typed, retryable + host failure and must not spend quota. +2. ``turn managed-step`` on that exact Turn key. It must answer ``wait`` with + the bounded continuation the journal proves (30s, attempt 1/3), and must + itself write nothing and spend nothing. +3. ``turn run-once --retry-failed-turn --resume-turn-key ...`` with the fake + runner now succeeding. The same Turn key must commit, the independent + validator must pass, and the goal must have spent exactly one quota slot. + +The point is the accounting: a retryable failure costs nothing, the retry costs +exactly one slot, and both facts come from the journal rather than from the +caller's bookkeeping. +""" + +from __future__ import annotations + +import argparse +import contextlib +import io +import json +import os +import sys +import tempfile +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +from loopx.cli import main as cli_main # noqa: E402 + +GOAL_ID = "loopx-turn-managed-step" +AGENT_ID = "managed-step-agent" +TODO_ID = "todo_managedstep01" +MARKER_NAME = "docs/managed-step-marker.txt" +MARKER_VALUE = "loopx-managed-step-self-healed" + +# The fake runner reads this state file to decide whether to fail. It is the +# only thing that changes between the first attempt and the retry. +ATTEMPTS_FILE = "attempts.json" + + +def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + project = root / "project" + runtime = root / "runtime" + workspace = root / "workspace" + runtime.mkdir(parents=True) + workspace.mkdir(parents=True) + (workspace / "docs").mkdir() + state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" + state.parent.mkdir(parents=True) + state.write_text( + "\n".join( + [ + "---", + "status: active", + "updated_at: 2026-01-01T00:00:00+00:00", + "---", + "", + "# Managed Step Self-Heal", + "", + "## Agent Todo", + "", + ( + f"- [ ] [P0] Produce the managed-step marker; a capacity " + f"failure must self-heal on the same Turn. `{MARKER_NAME}`" + ), + ( + f" " + ), + "", + ] + ), + encoding="utf-8", + ) + registry = project / ".loopx" / "registry.json" + registry.parent.mkdir(parents=True) + registry.write_text( + json.dumps( + { + "schema_version": 1, + "common_runtime_root": str(runtime), + "goals": [ + { + "id": GOAL_ID, + "domain": "loopx-turn-managed-step-fixture", + "status": "active", + "repo": str(project), + "state_file": str(state.relative_to(project)), + "adapter": { + "kind": "fixture_v0", + "status": "connected-delivery", + }, + "quota": {"compute": 1.0, "window_hours": 24}, + "coordination": { + "agent_model": "peer_v1", + "registered_agents": [AGENT_ID], + "agent_profiles": { + AGENT_ID: { + "schema_version": "agent_profile_v1", + "profile_role": "fixture", + "scope": "public qualification", + } + }, + "write_scope": ["docs/**"], + }, + } + ], + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + return project, runtime, workspace, registry + + +def _write_fake_runner(root: Path, workspace: Path) -> Path: + """A runner hook that fails the first attempt and succeeds afterwards. + + ``run_dsh_host`` classifies the raised error through the real + ``classify_dsh_failure`` path, so this exercises the production failure + channel rather than a test-only shortcut. + """ + + runner = root / "fake_dsh_runner.py" + runner.write_text( + "\n".join( + [ + "from __future__ import annotations", + "", + "import json", + "from pathlib import Path", + "", + f"ATTEMPTS = Path({str(workspace / ATTEMPTS_FILE)!r})", + "", + "", + "def run_dsh_turn(*, prompt, session_id=None, **kwargs): # noqa: ANN001,ARG001", + " attempts = (", + " int(json.loads(ATTEMPTS.read_text(encoding='utf-8'))['count'])", + " if ATTEMPTS.is_file()", + " else 0", + " )", + " attempts += 1", + " ATTEMPTS.write_text(json.dumps({'count': attempts}), encoding='utf-8')", + " if attempts == 1:", + " error = RuntimeError('provider at capacity')", + " error.code = 'insufficient_capacity'", + " raise error", + " marker = Path(__file__).resolve().parent / " + f"{str(Path('workspace') / MARKER_NAME)!r}", + " marker.parent.mkdir(parents=True, exist_ok=True)", + f" marker.write_text({MARKER_VALUE!r}, encoding='utf-8')", + " payload = {", + " 'result_kind': 'validated_progress',", + " 'classification': 'managed_step_fake_runner',", + " 'summary': 'Same-Turn retry produced the marker.',", + " 'recommended_action': 'Inspect the marker.',", + " 'next_action': 'Confirm exactly one quota slot was spent.',", + " 'vision_unchanged_reason': 'The objective path is unchanged.',", + " }", + " return json.dumps(payload)", + "", + ] + ), + encoding="utf-8", + ) + return runner + + +def _validator_command() -> list[str]: + program = ( + "import json,pathlib,sys; " + "json.load(sys.stdin); " + f"p=pathlib.Path({MARKER_NAME!r}); " + "raise SystemExit(0 if p.is_file() and " + f"p.read_text(encoding='utf-8').strip() == {MARKER_VALUE!r} else 9)" + ) + return [sys.executable, "-c", program] + + +def _run_cli(argv: list[str]) -> tuple[int, dict[str, Any]]: + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main(argv) + payload = json.loads(output.getvalue()) + assert isinstance(payload, dict), payload + return exit_code, payload + + +def _resume_argv(base_argv: list[str], turn_key: str) -> list[str]: + """Resume the exact journaled Turn: the plan is not recomputed. + + ``--turn-instance-id`` identifies a *new* logical Turn, so it must be + dropped once the caller is replaying a journaled one. + """ + + argv: list[str] = [] + skip_next = False + for item in base_argv: + if skip_next: + skip_next = False + continue + if item == "--turn-instance-id": + skip_next = True + continue + argv.append(item) + return [ + *argv, + "--retry-failed-turn", + "--resume-turn-key", + turn_key, + ] + + +def _base_argv( + *, + registry: Path, + runtime: Path, + workspace: Path, + runner: Path, + dsh_home: Path, +) -> list[str]: + return [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "run-once", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--turn-instance-id", + "managed-step-self-heal-turn-1", + "--host", + "dsh", + "--execution-mode", + "isolated-headless", + "--project", + str(workspace), + "--dsh-runner", + str(runner), + "--dsh-home", + str(dsh_home), + "--dsh-model", + "mock-model", + "--validation-command-json", + json.dumps(_validator_command()), + "--validation-failure-kind", + "repair_required", + "--scan-root", + str(registry.parent.parent), + "--no-global-sync", + "--timeout-seconds", + "60", + "--execute", + ] + + +def _quota_spend_count(runtime: Path) -> int: + index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + if not index.is_file(): + return 0 + return sum( + 1 + for line in index.read_text(encoding="utf-8").splitlines() + if json.loads(line).get("classification") == "quota_slot_spent" + ) + + +def _journal_path(runtime: Path, turn_key: str) -> Path: + return ( + runtime + / "goals" + / GOAL_ID + / "turns" + / f"{turn_key.removeprefix('sha256:')}.json" + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.parse_args() + + with tempfile.TemporaryDirectory(prefix="loopx-managed-step-") as directory: + root = Path(directory) + _project, runtime, workspace, registry = _write_fixture(root) + runner = _write_fake_runner(root, workspace) + dsh_home = root / "dsh-home" + dsh_home.mkdir(parents=True) + old_env = { + key: os.environ.get(key) + for key in ("DSH_CWD", "DSH_HOME", "DSH_SESSION_ROOT") + } + try: + for key in ("DSH_CWD", "DSH_HOME", "DSH_SESSION_ROOT"): + os.environ.pop(key, None) + + base = _base_argv( + registry=registry, + runtime=runtime, + workspace=workspace, + runner=runner, + dsh_home=dsh_home, + ) + + # 1. The capacity failure must be typed and must not spend. + exit_code, failed = _run_cli(base) + assert exit_code == 1, (exit_code, failed) + assert failed["ok"] is False, failed + assert failed["status"] == "failed", failed + assert failed["result_kind"] == "host_failure", failed + host_failure = failed["host_failure"] + assert host_failure["kind"] == "provider_capacity", host_failure + assert host_failure["retryable"] is True, host_failure + assert host_failure["attempt"] == 1, host_failure + assert failed["effects"]["quota_spent"] is False, failed + assert failed["effects"]["state_written"] is False, failed + assert _quota_spend_count(runtime) == 0, "the failure spent a slot" + turn_key = failed["resume_turn_key"] + assert _journal_path(runtime, turn_key).is_file(), turn_key + spends_after_failure = _quota_spend_count(runtime) + + # 2. The managed step must answer wait without touching anything. + _, step = _run_cli( + [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "managed-step", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--turn-key", + turn_key, + "--host", + "dsh", + "--execution-mode", + "isolated-headless", + "--scan-root", + str(registry.parent.parent), + "--observed-attempt", + "1", + "--observed-max-attempts", + "3", + ] + ) + assert step["ok"] is True, step + assert step["disposition"] == "wait", step + continuation = step["retry_continuation"] + assert continuation["same_turn"] is True, continuation + assert continuation["retry_failed_turn"] is True, continuation + assert continuation["strategy"] == "same_configuration", continuation + assert continuation["retry_after_seconds"] == 30, continuation + assert continuation["attempt"] == 1, continuation + assert continuation["max_attempts"] == 3, continuation + assert continuation["fresh_envelope_required"] is True, continuation + assert continuation["model_fallback_allowed"] is False, continuation + assert _quota_spend_count(runtime) == spends_after_failure, ( + "the managed step changed the spend ledger" + ) + + # 3. Replaying the same Turn must self-heal and spend exactly once. + exit_code, healed = _run_cli(_resume_argv(base, turn_key)) + assert exit_code == 0, (exit_code, healed) + assert healed["ok"] is True, healed + assert healed["resume_turn_key"] == turn_key, healed + assert healed["status"] == "committed", healed + assert healed["result_kind"] == "validated_progress", healed + assert healed["effects"]["quota_spent"] is True, healed + assert healed["effects"]["state_written"] is True, healed + assert healed["quota_slot_spend_count"] == 1, healed + assert _quota_spend_count(runtime) == 1, "the retry did not spend once" + assert (workspace / MARKER_NAME).read_text(encoding="utf-8").strip() == ( + MARKER_VALUE + ) + + # 4. Idempotent replay: the healed Turn must not spend again. + exit_code, replayed = _run_cli(_resume_argv(base, turn_key)) + assert exit_code == 0, (exit_code, replayed) + assert replayed["replayed"] is True, replayed + assert _quota_spend_count(runtime) == 1, "the replay double-spent" + finally: + for key, value in old_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + print( + "managed-step self-heal: provider_capacity -> wait(30s, 1/3) -> " + "same-Turn validated_progress, exactly one quota slot" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index a07209d970..9d3236d65f 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -18,7 +18,6 @@ run_configured_turn_outcome_ingest_fail_open, ) from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch -from ..capabilities.periodic_report.pending_intent import periodic_report_pending_intent_interaction_hook from ..control_plane.quota.live_decision import build_live_quota_should_run_decision from ..control_plane.agents.workspace_guard import capture_delivery_workspace from ..control_plane.quota.heartbeat_receipt import ( @@ -66,11 +65,16 @@ from .turn_dsh_host import build_dsh_host_runner from .turn_registration import register_turn_commands as register_turn_commands from .turn_inspection import handle_turn_journal_inspection +from .turn_decision import ( + apply_controller_advisory_primary, + build_turn_decision_builder, + collect_turn_status_payload, +) +from .turn_managed_step import handle_turn_managed_step from .turn_rendering import ( render_loopx_turn_execution_markdown as _render_loopx_turn_execution_markdown, render_loopx_turn_plan_markdown as _render_loopx_turn_plan_markdown, ) -from .turn_selection import turn_controller_advisory_primary from .turn_todo_writeback import ( write_turn_repair_update, write_turn_validated_completion, @@ -108,10 +112,16 @@ def handle_turn_command( ) if inspection_result is not None: return inspection_result + managed_step_result = handle_turn_managed_step( + args, + registry_path=registry_path, + runtime_root_arg=runtime_root_arg, + output_format=output_format, + print_payload=print_payload, + ) + if managed_step_result is not None: + return managed_step_result try: - scan_roots = [Path(item).expanduser() for item in args.scan_path] - if not scan_roots: - scan_roots = [Path(args.scan_root).expanduser()] runtime_root = resolve_status_projection_cache_runtime_root( registry_path=registry_path, runtime_root_override=runtime_root_arg, @@ -144,60 +154,30 @@ def handle_turn_command( operator_inbox_urgency_projector = build_lark_operator_inbox_urgency_projector( runtime_root_arg=runtime_root, ) - status_payload = collect_status( - registry_path=registry_path, - runtime_root_override=runtime_root_arg, - scan_roots=scan_roots, - limit=max(max(0, args.limit), AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK), - goal_id=args.goal_id, - available_capabilities=args.available_capabilities, - ) + scan_roots = [Path(item).expanduser() for item in args.scan_path] + if not scan_roots: + scan_roots = [Path(args.scan_root).expanduser()] scheduler_context = scheduler_execution_context_for_turn( host=args.host, execution_mode=args.execution_mode, scheduler_owner=args.scheduler_owner, ) - def build_turn_decision( - *, requested_action_todo_id: str | None = None - ) -> dict[str, Any]: - return build_live_quota_should_run_decision( - status_payload, - goal_id=args.goal_id, - agent_id=args.agent_id, - available_capabilities=args.available_capabilities, - include_scheduler_detail=False, - codex_app_current_rrule=None, + # Decision building is shared with the managed step so both owners + # always answer from the same control-plane projection. + decision = apply_controller_advisory_primary( + build_turn_decision_builder( + args, registry_path=registry_path, runtime_root=runtime_root, - route_source="loopx_turn_plan", - scheduler_execution_context=scheduler_context, - operator_inbox_urgency_projector=operator_inbox_urgency_projector, - bounded_research_frontier_projector=( - project_live_explore_composition_frontier + runtime_root_arg=runtime_root_arg, + status_payload=collect_turn_status_payload( + args, + registry_path=registry_path, + runtime_root_arg=runtime_root_arg, ), - requested_action_todo_id=requested_action_todo_id, turn_start_hook_dispatch=turn_start_hook_dispatch, - interaction_projection_hooks=(periodic_report_pending_intent_interaction_hook( - registry_path=registry_path, runtime_root=runtime_root, - goal_id=args.goal_id, agent_id=args.agent_id),), - ) - - decision = build_turn_decision() - controller_default = turn_controller_advisory_primary(decision) - if controller_default is not None: - primary_todo_id, advisory_portfolio = controller_default - decision = build_turn_decision( - requested_action_todo_id=primary_todo_id, ) - selected_todo = decision.get("selected_todo") - if not isinstance(selected_todo, dict) or ( - selected_todo.get("todo_id") != primary_todo_id - ): - raise ValueError( - "Turn controller advisory primary failed current eligibility" - ) - selected_todo["selected_by"] = "turn_controller_advisory_primary" - decision["action_portfolio"] = advisory_portfolio + ) resume_identity = { "goal_id": args.resume_goal_id, "agent_id": args.resume_agent_id, diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py new file mode 100644 index 0000000000..231a6caac7 --- /dev/null +++ b/loopx/cli_commands/turn_decision.py @@ -0,0 +1,202 @@ +"""One shared fresh Turn decision for every Turn subcommand that needs one. + +``run-once`` and ``managed-step`` must agree on what "the current governing +decision" means: read the same live status, apply the same controller advisory +primary, and sign the same ``loopx_turn_envelope_v0``. Duplicating that chain +would let the two drift, so both owners build through this module. + +Nothing here executes, writes, or spends: it only projects the control plane's +current decision into the envelope the loop controller consumes. +""" + +from __future__ import annotations + +import argparse +from collections.abc import Callable, Mapping +from pathlib import Path +from typing import Any + +from ..capabilities.explore.composition_frontier import ( + project_live_explore_composition_frontier, +) +from ..capabilities.periodic_report.pending_intent import ( + periodic_report_pending_intent_interaction_hook, +) +from ..control_plane.quota.live_decision import build_live_quota_should_run_decision +from ..control_plane.quota.turn_envelope import build_turn_envelope +from ..control_plane.scheduler.execution_context import ( + scheduler_execution_context_for_turn, +) +from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status +from .lark_inbox import build_lark_operator_inbox_urgency_projector +from .turn_selection import turn_controller_advisory_primary + +#: The route source every Turn owner attributes its decision to. +TURN_DECISION_ROUTE_SOURCE = "loopx_turn_plan" + + +def collect_turn_status_payload( + args: argparse.Namespace, + *, + registry_path: Path, + runtime_root_arg: str | None, +) -> dict[str, Any]: + """Read the live status the Turn decision is derived from.""" + + scan_roots = [Path(item).expanduser() for item in args.scan_path] + if not scan_roots: + scan_roots = [Path(args.scan_root).expanduser()] + return collect_status( + registry_path=registry_path, + runtime_root_override=runtime_root_arg, + scan_roots=scan_roots, + limit=max(max(0, args.limit), AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK), + goal_id=args.goal_id, + available_capabilities=args.available_capabilities, + ) + + +def build_turn_decision_builder( + args: argparse.Namespace, + *, + registry_path: Path, + runtime_root: Path, + runtime_root_arg: str | None, + status_payload: Mapping[str, Any], + turn_start_hook_dispatch: Mapping[str, Any] | None = None, +) -> Callable[..., dict[str, Any]]: + """Return the shared ``build_turn_decision`` used by the Turn owners. + + ``turn_start_hook_dispatch`` is the caller's business: an executing Turn + may publish Go/No-Go hooks before deciding, while a read-only managed step + must not. Passing the projection in keeps that choice with the caller. + """ + + scheduler_context = scheduler_execution_context_for_turn( + host=args.host, + execution_mode=args.execution_mode, + scheduler_owner=args.scheduler_owner, + ) + operator_inbox_urgency_projector = build_lark_operator_inbox_urgency_projector( + runtime_root_arg=runtime_root_arg, + ) + + def build_turn_decision( + *, requested_action_todo_id: str | None = None + ) -> dict[str, Any]: + return build_live_quota_should_run_decision( + status_payload, + goal_id=args.goal_id, + agent_id=args.agent_id, + available_capabilities=args.available_capabilities, + include_scheduler_detail=False, + codex_app_current_rrule=None, + registry_path=registry_path, + runtime_root=runtime_root, + route_source=TURN_DECISION_ROUTE_SOURCE, + scheduler_execution_context=scheduler_context, + operator_inbox_urgency_projector=operator_inbox_urgency_projector, + bounded_research_frontier_projector=( + project_live_explore_composition_frontier + ), + requested_action_todo_id=requested_action_todo_id, + turn_start_hook_dispatch=dict(turn_start_hook_dispatch or {}), + interaction_projection_hooks=( + periodic_report_pending_intent_interaction_hook( + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=args.goal_id, + agent_id=args.agent_id, + ), + ), + ) + + return build_turn_decision + + +def apply_controller_advisory_primary( + build_turn_decision: Callable[..., dict[str, Any]], +) -> dict[str, Any]: + """Build the decision, preferring the controller's advisory primary Todo.""" + + decision = build_turn_decision() + controller_default = turn_controller_advisory_primary(decision) + if controller_default is None: + return decision + primary_todo_id, advisory_portfolio = controller_default + decision = build_turn_decision(requested_action_todo_id=primary_todo_id) + selected_todo = decision.get("selected_todo") + if not isinstance(selected_todo, dict) or ( + selected_todo.get("todo_id") != primary_todo_id + ): + raise ValueError( + "Turn controller advisory primary failed current eligibility" + ) + selected_todo["selected_by"] = "turn_controller_advisory_primary" + decision["action_portfolio"] = advisory_portfolio + return decision + + +def fresh_turn_envelope( + decision: Mapping[str, Any], + *, + scheduler_execution_context: Any, +) -> dict[str, Any]: + """Sign the decision into the envelope the loop controller consumes. + + The causal link to a failed predecessor is *not* written here: the envelope + schema is a signed contract, so the managed step passes its predecessor + beside the envelope instead of mutating it. + """ + + return build_turn_envelope( + decision, + scheduler_execution_context=scheduler_execution_context, + ) + + +def build_fresh_envelope_for_managed_step( + args: argparse.Namespace, + *, + registry_path: Path, + runtime_root: Path, + runtime_root_arg: str | None, +) -> dict[str, Any]: + """Project the current decision as the managed step's fresh envelope. + + Read-only: no Turn-start hooks are dispatched and no Turn instance is + minted, because the managed step only decides whether the outer scheduler + may wake the same failed Turn again. + """ + + status_payload = collect_turn_status_payload( + args, + registry_path=registry_path, + runtime_root_arg=runtime_root_arg, + ) + build_turn_decision = build_turn_decision_builder( + args, + registry_path=registry_path, + runtime_root=runtime_root, + runtime_root_arg=runtime_root_arg, + status_payload=status_payload, + ) + decision = apply_controller_advisory_primary(build_turn_decision) + return fresh_turn_envelope( + decision, + scheduler_execution_context=scheduler_execution_context_for_turn( + host=args.host, + execution_mode=args.execution_mode, + scheduler_owner=args.scheduler_owner, + ), + ) + + +__all__ = [ + "TURN_DECISION_ROUTE_SOURCE", + "apply_controller_advisory_primary", + "build_fresh_envelope_for_managed_step", + "build_turn_decision_builder", + "collect_turn_status_payload", + "fresh_turn_envelope", +] diff --git a/loopx/cli_commands/turn_managed_step.py b/loopx/cli_commands/turn_managed_step.py new file mode 100644 index 0000000000..bc85c96c17 --- /dev/null +++ b/loopx/cli_commands/turn_managed_step.py @@ -0,0 +1,103 @@ +"""CLI surface for the managed-step consumer of same-Turn continuation. + +Split from ``turn.py`` to respect the CLI command-owner size budget. The +subcommand resolves the canonical Turn journal, rebuilds its validated receipt, +projects the current control-plane decision as a fresh envelope, asks the pure +controller for a disposition, and prints a typed answer. It never launches a +host, writes state, or spends quota. +""" + +from __future__ import annotations + +import argparse +from collections.abc import Callable +from pathlib import Path + +from ..control_plane.runtime.status_projection_cache import ( + resolve_status_projection_cache_runtime_root, +) +from ..control_plane.turn_driver import ( + load_turn_journal, + turn_journal_path, +) +from ..control_plane.turn_driver.managed_step import ( + LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, + decide_managed_step, +) +from .turn_decision import build_fresh_envelope_for_managed_step +from .turn_rendering import render_loopx_turn_managed_step_markdown + +PrintPayload = Callable[ + [dict[str, object], str, Callable[[dict[str, object]], str]], + None, +] +FormatSelector = Callable[..., str] + + +def _load_journal( + runtime_root: Path, + *, + goal_id: str, + turn_key: str, +) -> dict[str, object]: + """Read the canonical journal, or refuse when it does not exist.""" + + path = turn_journal_path(runtime_root, goal_id=goal_id, turn_key=turn_key) + journal = load_turn_journal(path) + if journal is None: + raise ValueError("LoopX Turn journal does not exist") + return journal + + +def handle_turn_managed_step( + args: argparse.Namespace, + *, + registry_path: Path, + runtime_root_arg: str | None, + output_format: FormatSelector, + print_payload: PrintPayload, +) -> int | None: + """Handle ``loopx turn managed-step`` and return its exit code.""" + + if args.turn_command != "managed-step": + return None + try: + runtime_root = resolve_status_projection_cache_runtime_root( + registry_path=registry_path, + runtime_root_override=runtime_root_arg, + ) + journal = _load_journal( + runtime_root, + goal_id=args.goal_id, + turn_key=args.turn_key, + ) + fresh_decision = build_fresh_envelope_for_managed_step( + args, + registry_path=registry_path, + runtime_root=runtime_root, + runtime_root_arg=runtime_root_arg, + ) + payload: dict[str, object] = { + "ok": True, + **decide_managed_step( + journal, + fresh_decision, + goal_id=args.goal_id, + agent_id=args.agent_id, + turn_key=args.turn_key, + observed_attempt=args.observed_attempt, + observed_max_attempts=args.observed_max_attempts, + ), + } + except Exception as exc: # noqa: BLE001 - typed CLI failure boundary + payload = { + "ok": False, + "schema_version": LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, + "error": str(exc), + } + print_payload( + payload, + output_format(args), + render_loopx_turn_managed_step_markdown, + ) + return 0 if payload.get("ok") else 1 diff --git a/loopx/cli_commands/turn_registration.py b/loopx/cli_commands/turn_registration.py index b11e9fbcbd..524693460b 100644 --- a/loopx/cli_commands/turn_registration.py +++ b/loopx/cli_commands/turn_registration.py @@ -61,6 +61,62 @@ def register_turn_commands( ) plan.add_argument("--limit", type=int, default=5) + managed_step = command_sub.add_parser( + "managed-step", + help=( + "Decide one bounded same-Turn continuation for a failed Turn " + "without executing it." + ), + description=( + "Read one canonical Turn journal, rebuild its validated receipt, " + "and ask the pure Turn Loop Controller for a disposition against " + "the current decision. Grants no execution authority: it never " + "launches a host, writes state, or spends quota. The Turn journal " + "remains the authority for the attempt count and retry budget." + ), + ) + add_subcommand_format(managed_step) + _add_turn_decision_arguments( + managed_step, + default_host="dsh", + host_choices=["codex-cli", "dsh", "generic-cli"], + execution_mode_choices=["isolated-headless"], + default_execution_mode="isolated-headless", + ) + managed_step.add_argument( + "--turn-key", + required=True, + help="Exact sha256 Turn key of the failed Turn to decide about.", + ) + managed_step.add_argument( + "--observed-attempt", + type=int, + help=( + "Caller's observed attempt count, reconciled against the Turn " + "journal. A disagreement is refused rather than adopted." + ), + ) + managed_step.add_argument( + "--observed-max-attempts", + type=int, + help=( + "Caller's observed retry ceiling, reconciled against the Turn " + "journal retry policy." + ), + ) + managed_step.add_argument( + "--scan-root", + default=default_public_scan_root(), + help="Public files to scan for obvious private material.", + ) + managed_step.add_argument( + "--scan-path", + action="append", + default=[], + help="Specific public file or directory to scan. Repeatable.", + ) + managed_step.add_argument("--limit", type=int, default=5) + run_once = command_sub.add_parser( "run-once", help=( diff --git a/loopx/cli_commands/turn_rendering.py b/loopx/cli_commands/turn_rendering.py index de219522cf..22ad87c834 100644 --- a/loopx/cli_commands/turn_rendering.py +++ b/loopx/cli_commands/turn_rendering.py @@ -160,3 +160,52 @@ def render_loopx_turn_journal_inspection_markdown( "- effects: none", ] ) + + +def render_loopx_turn_managed_step_markdown(payload: dict[str, object]) -> str: + if not payload.get("ok"): + error = payload.get("error") or "Turn managed step failed" + return f"LoopX Turn managed step failed: {error}" + continuation = ( + payload.get("retry_continuation") + if isinstance(payload.get("retry_continuation"), dict) + else {} + ) + lineage = ( + payload.get("lineage") if isinstance(payload.get("lineage"), dict) else {} + ) + return "\n".join( + [ + "# LoopX Turn Managed Step", + f"- disposition: {payload.get('disposition')}", + f"- reason: {payload.get('reason')}", + f"- turn_key: {payload.get('turn_key')}", + f"- attempt: {payload.get('attempt')}" + + ( + f"/{payload.get('max_attempts')}" + if payload.get("max_attempts") is not None + else "" + ), + f"- goal_id: {lineage.get('goal_id') or 'none'}", + f"- agent_id: {lineage.get('agent_id') or 'none'}", + f"- todo_id: {lineage.get('todo_id') or 'none'}", + *( + [ + f"- retry_after_seconds: {continuation.get('retry_after_seconds')}", + f"- retry_strategy: {continuation.get('strategy')}", + "- same_turn: " + f"{continuation.get('same_turn')}; " + "retry_failed_turn: " + f"{continuation.get('retry_failed_turn')}", + "- fresh_envelope_required: " + f"{continuation.get('fresh_envelope_required')}; " + "model_fallback_allowed: " + f"{continuation.get('model_fallback_allowed')}", + ] + if continuation + else ["- continuation: none"] + ), + "- execution_authority: none", + "- effects: none", + ] + ) diff --git a/loopx/control_plane/turn_driver/__init__.py b/loopx/control_plane/turn_driver/__init__.py index 726b42109c..f24344b8e8 100644 --- a/loopx/control_plane/turn_driver/__init__.py +++ b/loopx/control_plane/turn_driver/__init__.py @@ -27,7 +27,15 @@ run_loopx_turn_once, validate_loopx_turn_host_result, ) -from .journal_store import load_loopx_turn_plan_from_journal +from .journal_store import ( + load_loopx_turn_plan_from_journal, + load_turn_journal, + turn_journal_path, +) +from .managed_step import ( + LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, + decide_managed_step, +) from .recovery import TurnRecoveryBlockedError from .loop_controller import ( BOUNDED_TURN_BUDGET_SCHEMA_VERSION, @@ -56,6 +64,7 @@ "LOOPX_ITERATION_CONTEXT_POLICY_SCHEMA_VERSION", "LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION", "LOOPX_TURN_JOURNAL_INSPECTION_SCHEMA_VERSION", + "LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION", "LOOPX_TURN_RESULT_SCHEMA_VERSION", "LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION", "LOOPX_TURN_TASK_VALIDATION_SCHEMA_VERSION", @@ -75,9 +84,12 @@ "codex_cli_session_binding", "codex_cli_session_id_from_jsonl", "decide_loop_disposition", + "decide_managed_step", "load_codex_cli_session", "inspect_loopx_turn_journal", "load_loopx_turn_plan_from_journal", + "load_turn_journal", + "turn_journal_path", "loopx_turn_execution_committed", "loopx_turn_execution_has_durable_effects", "loopx_turn_execution_recovery_required", diff --git a/loopx/control_plane/turn_driver/managed_step.py b/loopx/control_plane/turn_driver/managed_step.py new file mode 100644 index 0000000000..9ca3d5d98f --- /dev/null +++ b/loopx/control_plane/turn_driver/managed_step.py @@ -0,0 +1,292 @@ +"""The managed-step consumer of one bounded same-Turn continuation. + +``loopx turn run-once`` commits a governed Turn and stops. When the host +failed for a retryable reason, the pure Turn Loop Controller already answers +``wait`` with a typed ``retry_continuation`` block, but nothing in production +consumed it: the outer scheduler was left to infer retryability from prose. + +This module is that consumer. It rebuilds one validated receipt from the +canonical Turn journal, asks the controller for a disposition against a fresh +``loopx_turn_envelope_v0`` decision, and returns the typed answer. It is +read-only by construction: + +- it never invokes a host, writes state, spends quota, or sleeps; +- the Turn journal stays the sole authority for attempt and max_attempts, so a + caller-supplied observation is only ever reconciled, never believed; +- a disagreement between the observation and the journal is a ``ValueError`` at + the typed-input boundary rather than a silently adopted number. + +The caller decides whether to carry the existing ``--retry-failed-turn`` / +``--resume-turn-key`` flags on the next ``run-once``; this module grants no +execution authority of its own. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from .executor import LOOPX_TURN_EXECUTION_SCHEMA_VERSION +from .host_failure import normalize_host_failure_record, project_host_failure +from .loop_controller import ( + LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION, + ValidatedTurnReceipt, + decide_loop_disposition, +) +from .transaction import ( + LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION, + LoopXTurnResultKind, +) + +LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION = "loopx_turn_managed_step_v0" + +# A blocked recovery audit means the journal is not resumable for this +# identity; the controller decision must never be reached in that case. +_REPLAY_BLOCKED_ACTION = "blocked" + + +def _mapping(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _require_positive_int(value: Any, *, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise ValueError(f"{field} must be a positive integer") + return value + + +def _validated_turn_receipt(journal: Mapping[str, Any]) -> ValidatedTurnReceipt: + """Rebuild the controller receipt from the canonical journal alone. + + The journal stores the plan and the validated receipt rather than an + execution payload, so the projection here is reconstructed from committed + fields only. ``ValidatedTurnReceipt.from_execution`` then re-enforces the + material-effect and lineage contracts, which keeps this reader honest + without duplicating them. + """ + + plan = journal.get("plan") + if not isinstance(plan, Mapping): + raise TypeError("Turn journal has no stored plan to reconcile") + stored_transaction = _mapping(plan.get("transaction")) + turn_key = str(journal.get("turn_key") or "") + if not turn_key or str(stored_transaction.get("turn_key") or "") != turn_key: + raise ValueError("Turn journal turn_key does not match its stored transaction") + + receipt = journal.get("receipt") + if not isinstance(receipt, Mapping): + raise TypeError("Turn journal has no validated receipt to reconcile") + result_kind = str(receipt.get("result_kind") or "") + if receipt.get("schema_version") != LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION: + raise ValueError("Turn journal receipt has an unsupported schema") + if receipt.get("ok") is not True: + raise ValueError("Turn journal receipt is not a validated receipt") + + execution: dict[str, Any] = { + "ok": journal.get("status") == "committed", + "schema_version": LOOPX_TURN_EXECUTION_SCHEMA_VERSION, + "mode": "managed_step", + "status": journal.get("status"), + "result_kind": result_kind, + "receipt": dict(receipt), + "scheduler": _mapping(journal.get("scheduler")), + **project_host_failure(journal), + } + settlement_result = journal.get("settlement_result") + if isinstance(settlement_result, Mapping): + execution["settlement_result"] = dict(settlement_result) + writeback = _mapping(journal.get("writeback")) + completion = writeback.get("completion") + if isinstance(completion, Mapping): + execution["todo_completion"] = dict(completion) + return ValidatedTurnReceipt.from_execution(execution) + + +def managed_step_receipt_from_journal( + journal: Mapping[str, Any], + *, + goal_id: str, + agent_id: str, + turn_key: str, +) -> ValidatedTurnReceipt: + """Qualify one journaled Turn for the managed-step transition. + + Fails closed unless the journal is a finished failed Turn for exactly this + goal/agent/turn identity whose typed host failure is retryable. A recovery + audit whose planned action is ``blocked`` is refused here rather than at the + controller, because an unsafe journal must not reach the transition at all. + """ + + if journal.get("status") != "failed": + raise ValueError("managed step requires a failed Turn journal") + if str(journal.get("turn_key") or "") != turn_key: + raise ValueError("Turn journal turn_key does not match the requested Turn") + result_kind = str(journal.get("result_kind") or "") + if result_kind != LoopXTurnResultKind.HOST_FAILURE.value: + raise ValueError("managed step requires a typed host failure Turn") + + recovery = _mapping(journal.get("recovery_audit")) + planned = _mapping(recovery.get("planned")) + action = str(planned.get("action") or "") + if action == _REPLAY_BLOCKED_ACTION: + raise ValueError("Turn journal replay is blocked for this identity") + + failure = normalize_host_failure_record(journal.get("host_failure")) + if failure.get("retryable") is not True: + raise ValueError( + f"host failure {failure.get('kind')} is not retryable; repair instead" + ) + + receipt = _validated_turn_receipt(journal) + lineage = receipt.lineage + if lineage["goal_id"] != goal_id or lineage["agent_id"] != agent_id: + raise ValueError("Turn journal lineage does not match the requested goal/agent") + return receipt + + +def reconcile_observed_attempt( + journal: Mapping[str, Any], + *, + observed_attempt: int | None, + observed_max_attempts: int | None = None, +) -> None: + """Reconcile a caller-supplied observation against the journal authority. + + The journal owns the attempt count. An observation that disagrees with it is + a contract violation, not a correction, so this raises instead of preferring + either side. + """ + + authority = _require_positive_int( + journal.get("host_attempt_count"), + field="Turn journal host_attempt_count", + ) + failure = normalize_host_failure_record(journal.get("host_failure")) + retry = _mapping(failure.get("retry")) + # A non-retryable failure carries no retry policy, so there is no ceiling to + # reconcile against; the caller is refused for the failure kind itself. + max_attempts = ( + _require_positive_int( + retry.get("max_attempts"), + field="Turn journal retry max_attempts", + ) + if failure.get("retryable") is True + else None + ) + if observed_attempt is not None: + _require_positive_int(observed_attempt, field="observed_attempt") + if observed_attempt != authority: + raise ValueError( + "observed_attempt disagrees with the Turn journal attempt authority" + ) + if observed_max_attempts is not None: + _require_positive_int(observed_max_attempts, field="observed_max_attempts") + if max_attempts is not None and observed_max_attempts != max_attempts: + raise ValueError( + "observed_max_attempts disagrees with the Turn journal retry policy" + ) + + +def _managed_payload( + disposition: Mapping[str, Any], + *, + turn_key: str, + journal: Mapping[str, Any], +) -> dict[str, Any]: + """Project the controller decision plus its journal-proven provenance.""" + + payload: dict[str, Any] = { + "schema_version": LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, + "disposition": disposition.get("disposition"), + "reason": disposition.get("reason"), + "turn_key": turn_key, + "attempt": _require_positive_int( + journal.get("host_attempt_count"), + field="Turn journal host_attempt_count", + ), + } + for key in ( + "lineage", + "stop_scope", + "goal_terminal", + "continuation_required", + "successor_todo_ids", + "capability_action", + "capability_action_required", + "user_action", + ): + if key in disposition: + payload[key] = disposition[key] + retry_continuation = disposition.get("retry_continuation") + if isinstance(retry_continuation, Mapping): + payload["retry_continuation"] = dict(retry_continuation) + # The journal is the authority for the budget; echoing it here keeps a + # caller from having to trust the controller's copy. + payload["max_attempts"] = _require_positive_int( + retry_continuation.get("max_attempts"), + field="retry_continuation max_attempts", + ) + return payload + + +def decide_managed_step( + journal: Mapping[str, Any], + fresh_decision: Mapping[str, Any], + *, + goal_id: str, + agent_id: str, + turn_key: str, + predecessor_turn_key: str | None = None, + observed_attempt: int | None = None, + observed_max_attempts: int | None = None, +) -> dict[str, Any]: + """Decide the managed continuation for one failed Turn. + + ``fresh_decision`` is a fresh ``loopx_turn_envelope_v0`` from the control + plane, exactly as ``run-once`` builds one. ``predecessor_turn_key`` carries + the causal link to the failed Turn and is deliberately *not* written into + the signed envelope; it defaults to the Turn under decision, which is the + only successor relationship a managed step can prove on its own. + + The answer is typed and grants no execution authority: ``wait`` means "wake + the same Turn after the bounded backoff", never "spend or write now". + """ + + journal = _mapping(journal) + # Refuse a non-failed or non-host-failure Turn before any reconciliation, so + # the caller's first error is about the input it actually supplied. + if journal.get("status") != "failed": + raise ValueError("managed step requires a failed Turn journal") + if str(journal.get("result_kind") or "") != LoopXTurnResultKind.HOST_FAILURE.value: + raise ValueError("managed step requires a typed host failure Turn") + # Reconcile before deciding so a forged observation cannot influence which + # branch is taken. + reconcile_observed_attempt( + journal, + observed_attempt=observed_attempt, + observed_max_attempts=observed_max_attempts, + ) + receipt = managed_step_receipt_from_journal( + journal, + goal_id=goal_id, + agent_id=agent_id, + turn_key=turn_key, + ) + envelope = _mapping(fresh_decision) + if str(envelope.get("goal_id") or "") not in {"", goal_id}: + raise ValueError("fresh decision goal_id does not match the managed step") + if str(envelope.get("agent_id") or "") not in {"", agent_id}: + raise ValueError("fresh decision agent_id does not match the managed step") + disposition = decide_loop_disposition( + turn_receipt=receipt, + quota_decision=envelope, + predecessor_turn_key=( + turn_key if predecessor_turn_key is None else predecessor_turn_key + ), + ) + if str(disposition.get("schema_version") or "") not in { + "", + LOOP_CONTROLLER_DISPOSITION_SCHEMA_VERSION, + }: + raise RuntimeError("Turn Loop Controller returned an unsupported schema") + return _managed_payload(disposition, turn_key=turn_key, journal=journal) diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py new file mode 100644 index 0000000000..32171d8e41 --- /dev/null +++ b/tests/test_loopx_turn_managed_step.py @@ -0,0 +1,343 @@ +"""Typed tests for the managed-step consumer of same-Turn continuation. + +Each test drives ``decide_managed_step`` off a journal shaped exactly as +``loopx turn run-once`` leaves one after a retryable host failure, paired with a +fresh ``loopx_turn_envelope_v0`` decision. The reader must stay pure: it grants +no execution authority, never spends or writes, and treats the Turn journal as +the sole authority for the attempt budget. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from loopx.control_plane.turn_driver import ( + LOOPX_TURN_RESULT_SCHEMA_VERSION, + LoopXTurnResultKind, + build_loopx_turn_transaction_plan, + validate_loopx_turn_receipt, +) +from loopx.control_plane.turn_driver.host_failure import build_host_failure_record +from loopx.control_plane.turn_driver.managed_step import ( + LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION, + decide_managed_step, + managed_step_receipt_from_journal, + reconcile_observed_attempt, +) + +GOAL_ID = "goal-managed-step" +AGENT_ID = "agent-managed-step" +TODO_ID = "todo-managed-step" + + +def _lineage() -> dict[str, str]: + return {"goal_id": GOAL_ID, "agent_id": AGENT_ID, "todo_id": TODO_ID} + + +def _plan() -> dict[str, Any]: + """A stored Turn plan shaped as ``build_loopx_turn_plan`` returns it.""" + + transaction = build_loopx_turn_transaction_plan( + planned=True, + lineage=_lineage(), + host="dsh", + execution_mode="interactive-visible", + session_action="resume", + ) + return {"transaction": transaction} + + +def _failed_journal( + *, + kind: str = "provider_capacity", + attempt: int = 1, + result_kind: LoopXTurnResultKind = LoopXTurnResultKind.HOST_FAILURE, + status: str = "failed", + lineage: dict[str, str] | None = None, +) -> dict[str, Any]: + transaction = build_loopx_turn_transaction_plan( + planned=True, + lineage=_lineage() if lineage is None else lineage, + host="dsh", + execution_mode="interactive-visible", + session_action="resume", + ) + plan = {"transaction": transaction} + failure_kind = result_kind in { + LoopXTurnResultKind.HOST_FAILURE, + LoopXTurnResultKind.VALIDATION_FAILED, + LoopXTurnResultKind.WRITEBACK_FAILED, + LoopXTurnResultKind.QUOTA_SPEND_FAILED, + } + if failure_kind: + completed: list[str] = [] + result: dict[str, Any] = { + "schema_version": LOOPX_TURN_RESULT_SCHEMA_VERSION, + "turn_key": transaction["turn_key"], + "result_kind": result_kind.value, + "completed_phases": completed, + "failed_phase": "host_execute", + } + else: + completed = ["host_execute", "typed_result", "validation"] + result = { + "schema_version": LOOPX_TURN_RESULT_SCHEMA_VERSION, + "turn_key": transaction["turn_key"], + "result_kind": result_kind.value, + "completed_phases": completed, + } + receipt = validate_loopx_turn_receipt(transaction, result) + assert receipt["ok"] is True, receipt + journal: dict[str, Any] = { + "schema_version": "loopx_turn_journal_v0", + "status": status, + "turn_key": transaction["turn_key"], + "result_kind": result_kind.value, + "plan": plan, + "receipt": receipt, + "completed_phases": completed, + "host_attempt_count": attempt, + } + if result_kind is LoopXTurnResultKind.HOST_FAILURE: + journal["host_failure"] = build_host_failure_record(kind, attempt=attempt) + return journal + + +def _envelope( + *, + should_run: bool = True, + effective_action: str = "deliver", + selected_todo_id: str | None = TODO_ID, + user_action_required: bool = False, + lineage: dict[str, str] | None = None, + predecessor_turn_key: str | None = None, +) -> dict[str, Any]: + lin = _lineage() if lineage is None else lineage + envelope: dict[str, Any] = { + "schema_version": "loopx_turn_envelope_v0", + "goal_id": lin["goal_id"], + "agent_id": lin["agent_id"], + "should_run": should_run, + "effective_action": effective_action, + "action_signature": {"matches": True, "source_hash": "sha256:test", "envelope_hash": "sha256:test"}, + "compaction": {"within_budget": True}, + "action": { + "delivery_allowed": True, + "must_attempt": True, + "quiet_noop_allowed": False, + "selected_todo": ( + {"todo_id": selected_todo_id} if selected_todo_id else None + ), + }, + "user": {"action_required": user_action_required}, + } + if predecessor_turn_key is not None: + envelope["predecessor_turn_key"] = predecessor_turn_key + return envelope + + +def _decide( + journal: dict[str, Any], + *, + envelope: dict[str, Any] | None = None, + **kwargs: Any, +) -> dict[str, Any]: + """Decide one managed step, binding the fresh decision to the receipt.""" + + return decide_managed_step( + journal, + ( + _envelope(predecessor_turn_key=str(journal["turn_key"])) + if envelope is None + else envelope + ), + goal_id=GOAL_ID, + agent_id=AGENT_ID, + turn_key=str(journal["turn_key"]), + **kwargs, + ) + + +def test_retryable_failure_with_budget_returns_wait_with_typed_continuation() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + payload = _decide(journal) + + assert payload["schema_version"] == LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION + assert payload["disposition"] == "wait" + assert payload["turn_key"] == journal["turn_key"] + assert payload["attempt"] == 1 + continuation = payload["retry_continuation"] + assert continuation == { + "same_turn": True, + "retry_failed_turn": True, + "strategy": "same_configuration", + "retry_after_seconds": 30, + "attempt": 1, + "max_attempts": 3, + "fresh_envelope_required": True, + "model_fallback_allowed": False, + } + assert payload["max_attempts"] == 3 + + +def test_exhausted_budget_requests_repair_instead_of_waiting() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=3) + + payload = _decide(journal) + + assert payload["disposition"] == "repair" + assert "retry_continuation" not in payload + assert payload["attempt"] == 3 + + +def test_non_retryable_failure_is_refused_before_the_controller() -> None: + journal = _failed_journal(kind="auth_failed", attempt=1) + + with pytest.raises(ValueError, match="not retryable"): + _decide(journal) + + +def test_committed_journal_is_not_a_managed_step_input() -> None: + journal = _failed_journal( + result_kind=LoopXTurnResultKind.VALIDATED_PROGRESS, + status="committed", + ) + + with pytest.raises(ValueError, match="failed Turn journal"): + _decide(journal) + + +def test_forged_observed_attempt_is_refused() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + with pytest.raises(ValueError, match="observed_attempt disagrees"): + _decide(journal, observed_attempt=99) + + +def test_forged_observed_max_attempts_is_refused() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + with pytest.raises(ValueError, match="observed_max_attempts disagrees"): + _decide(journal, observed_max_attempts=99) + + +def test_matching_observation_is_accepted() -> None: + journal = _failed_journal(kind="rate_limited", attempt=2) + + payload = _decide(journal, observed_attempt=2, observed_max_attempts=3) + + assert payload["disposition"] == "wait" + # Backoff doubles per attempt off the policy base (60s for rate_limited). + assert payload["retry_continuation"]["retry_after_seconds"] == 120 + + +def test_observation_must_be_a_positive_integer() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + with pytest.raises(ValueError, match="positive integer"): + _decide(journal, observed_attempt=0) + with pytest.raises(ValueError, match="positive integer"): + _decide(journal, observed_attempt=True) + + +def test_foreign_lineage_is_refused() -> None: + journal = _failed_journal( + lineage={"goal_id": "other-goal", "agent_id": AGENT_ID, "todo_id": TODO_ID} + ) + + with pytest.raises(ValueError, match="lineage does not match"): + _decide(journal) + + +def test_blocked_recovery_audit_is_refused() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + journal["recovery_audit"] = { + "schema_version": "loopx_turn_recovery_audit_v0", + "planned": { + "schema_version": "loopx_turn_recovery_decision_v0", + "action": "blocked", + "can_continue": False, + "resume_from": None, + "reinvoke_host": False, + "reason": "journal consistency violation", + "retry_failed": True, + "checks": [], + }, + "actual": { + "status": "finished", + "journal_status": "failed", + "completed_phases": [], + "host_invoked": True, + }, + } + + with pytest.raises(ValueError, match="replay is blocked"): + _decide(journal) + + +def test_fresh_decision_must_agree_on_goal_and_agent_lineage() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + foreign = _envelope( + lineage={"goal_id": "other-goal", "agent_id": AGENT_ID, "todo_id": TODO_ID} + ) + + with pytest.raises(ValueError, match="fresh decision goal_id"): + _decide(journal, envelope=foreign) + + +def test_managed_step_never_spends_or_writes() -> None: + """The reader is pure: no effects, no host invocation, no quota spend.""" + + journal = _failed_journal(kind="provider_capacity", attempt=1) + + payload = _decide(journal) + + for forbidden in ("effects", "quota_slot_spend_count", "host_invoked", "writeback"): + assert forbidden not in payload + # A wait decision must not carry any field that could be mistaken for + # authority to execute now. + assert "accepted_turn_keys" not in payload + assert payload["disposition"] == "wait" + + +def test_reconcile_accepts_omitted_observations() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + assert reconcile_observed_attempt( + journal, observed_attempt=None, observed_max_attempts=None + ) is None + + +def test_receipt_from_journal_projects_lineage_and_kind() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + + receipt = managed_step_receipt_from_journal( + journal, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + turn_key=str(journal["turn_key"]), + ) + + assert receipt.result_kind is LoopXTurnResultKind.HOST_FAILURE + assert receipt.lineage["todo_id"] == TODO_ID + assert receipt.host_failure is not None + assert receipt.host_failure["kind"] == "provider_capacity" + + +def test_journal_without_stored_plan_is_refused() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + journal.pop("plan") + + with pytest.raises(TypeError, match="no stored plan"): + _decide(journal) + + +def test_journal_turn_key_mismatch_is_refused() -> None: + journal = _failed_journal(kind="provider_capacity", attempt=1) + journal["turn_key"] = "sha256:" + "0" * 64 + + with pytest.raises(ValueError, match="turn_key"): + _decide(journal) From b6183f7aa69333596332b9d94d1b5d534e582bed Mon Sep 17 00:00:00 2001 From: song Date: Mon, 14 Sep 2026 09:04:02 +0800 Subject: [PATCH 2/4] fix(turn): read the resolved runtime root for the Lark activation check The shared Turn decision builder wired the operator-inbox activation check to the raw `--runtime-root` argument instead of the resolved runtime root. When a registry declares `common_runtime_root` and the command omits the flag, that raw value is None, so the check read the operator's global `~/.codex/loopx/extensions/state.json` rather than this registry's extension state. Verified on a registry-scoped fixture that installs the extension under its own runtime root: the unpatched path reported `extension loopx-lark is not installed` while the resolved path read the registry root. Both owners of this builder are affected; `run-once` happened to pass the resolved value at its own call site, so only `managed-step` showed the symptom. The regression test pins the argument the check receives, and reverting the wiring fails it. Signed-off-by: song --- loopx/cli_commands/turn_decision.py | 6 +++- tests/test_loopx_turn_managed_step.py | 46 +++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 231a6caac7..8ad90e24d1 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -77,8 +77,12 @@ def build_turn_decision_builder( execution_mode=args.execution_mode, scheduler_owner=args.scheduler_owner, ) + # Use the resolved runtime root, not the raw CLI argument. When a registry + # declares `common_runtime_root` and the command omits `--runtime-root`, + # the raw value is None and the activation check would silently read the + # global default instead of this registry's own extension state. operator_inbox_urgency_projector = build_lark_operator_inbox_urgency_projector( - runtime_root_arg=runtime_root_arg, + runtime_root_arg=runtime_root, ) def build_turn_decision( diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index 32171d8e41..c145e760a7 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -341,3 +341,49 @@ def test_journal_turn_key_mismatch_is_refused() -> None: with pytest.raises(ValueError, match="turn_key"): _decide(journal) + + +def test_shared_decision_builder_reads_the_resolved_runtime_root(monkeypatch) -> None: + """The activation check must read this registry's root, not the global default. + + A registry that declares ``common_runtime_root`` while the command omits + ``--runtime-root`` passes ``None`` as the raw argument. Wiring the check to + that raw value would silently read the operator's global extension state. + """ + + import argparse + from pathlib import Path + + from loopx.cli_commands import turn_decision + + seen: list[object] = [] + + def _record(*, runtime_root_arg): + seen.append(runtime_root_arg) + return lambda **_: {"schema_version": "lark_event_inbox_urgency_v0"} + + monkeypatch.setattr( + turn_decision, "build_lark_operator_inbox_urgency_projector", _record + ) + args = argparse.Namespace( + goal_id=GOAL_ID, + agent_id="agent-a", + host="codex-cli", + execution_mode=None, + scheduler_owner=None, + available_capabilities=[], + ) + resolved_root = Path("/tmp/registry-scoped-runtime-root") + + turn_decision.build_turn_decision_builder( + args, + registry_path=Path("/tmp/registry.json"), + runtime_root=resolved_root, + runtime_root_arg=None, + status_payload={"ok": True, "attention_queue": {"items": []}, "run_history": {"goals": []}}, + ) + + assert seen == [resolved_root], ( + "the activation check must receive the resolved runtime root, not the raw " + f"argument that is None when the registry declares common_runtime_root: {seen}" + ) From e811fe85f7e4b2a00605f05e8985d6a0155d22d2 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 14 Sep 2026 17:18:43 +0800 Subject: [PATCH 3/4] fix(turn): cross-bind the journal attempt fields before deciding The review found the exact counterexample: a journal with `host_attempt_count=3` and `host_failure.attempt=1` still returned `disposition=wait`, with the projection reporting `attempt=3` while `retry_continuation.attempt` said `1`. The two persisted attempt fields could therefore diverge and the retry ceiling was bypassed. The journal records the attempt twice by design: the executor increments `host_attempt_count`, and `record_host_failure` copies that value into the typed `host_failure` record the controller reads for the ceiling. They are written from one value, so a disagreement means the journal was edited or corrupted. `managed_step_receipt_from_journal` now requires the two to match and fails closed otherwise, before any reconciliation or controller call, so no branch is taken on a journal whose presentation and authority disagree. Verified: the counterexample now raises; an agreeing journal still returns `wait`; removing the binding fails the new regression; the managed-step suite is 18/18, the wider turn suites are 167/167, the self-heal smoke passes, and Ruff is clean. Signed-off-by: song --- .../control_plane/turn_driver/managed_step.py | 35 +++++++++++++++++++ tests/test_loopx_turn_managed_step.py | 31 ++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/loopx/control_plane/turn_driver/managed_step.py b/loopx/control_plane/turn_driver/managed_step.py index 9ca3d5d98f..739cff1409 100644 --- a/loopx/control_plane/turn_driver/managed_step.py +++ b/loopx/control_plane/turn_driver/managed_step.py @@ -55,6 +55,30 @@ def _require_positive_int(value: Any, *, field: str) -> int: return value +def _require_matching_attempts(journal_attempt: Any, failure_attempt: Any) -> None: + """Refuse a journal whose two persisted attempt fields disagree. + + The executor increments ``host_attempt_count`` and ``record_host_failure`` + writes the same value into ``host_failure.attempt``. The controller reads + the nested field for the retry ceiling while the projection reports the top + level, so a divergence would present one attempt count and authorize on + another. + """ + + authority = _require_positive_int( + journal_attempt, + field="Turn journal host_attempt_count", + ) + nested = _require_positive_int( + failure_attempt, + field="Turn journal host_failure attempt", + ) + if nested != authority: + raise ValueError( + "Turn journal host_failure attempt disagrees with host_attempt_count" + ) + + def _validated_turn_receipt(journal: Mapping[str, Any]) -> ValidatedTurnReceipt: """Rebuild the controller receipt from the canonical journal alone. @@ -136,6 +160,17 @@ def managed_step_receipt_from_journal( raise ValueError( f"host failure {failure.get('kind')} is not retryable; repair instead" ) + # The journal records the attempt twice: once as the top-level + # `host_attempt_count` the executor increments, and once inside the typed + # `host_failure` record the controller reads to decide whether the retry + # ceiling is reached. `record_host_failure` writes them from one value, so a + # disagreement means the journal was edited or corrupted. Preferring either + # side would let the presentation report `3/3` while the controller still + # authorized a retry, so refuse the Turn instead. + _require_matching_attempts( + journal.get("host_attempt_count"), + failure.get("attempt"), + ) receipt = _validated_turn_receipt(journal) lineage = receipt.lineage diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index c145e760a7..fc9163a1b6 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -243,6 +243,37 @@ def test_observation_must_be_a_positive_integer() -> None: _decide(journal, observed_attempt=True) +def test_journal_attempt_divergence_is_refused() -> None: + """The two persisted attempt fields must agree, in both directions. + + The executor increments ``host_attempt_count`` while the controller reads + ``host_failure.attempt`` for the retry ceiling. A journal that reports one + value at the top level and another inside the typed failure would present + ``3/3`` while still authorizing a retry, so it must fail closed. + """ + + consumed = _failed_journal(kind="provider_capacity", attempt=3) + consumed["host_failure"] = build_host_failure_record("provider_capacity", attempt=1) + + with pytest.raises(ValueError, match="host_failure attempt disagrees"): + _decide(consumed) + + # The opposite direction is refused the same way: a nested attempt ahead of + # the journal's own counter. + ahead = _failed_journal(kind="provider_capacity", attempt=1) + ahead["host_failure"] = build_host_failure_record("provider_capacity", attempt=3) + + with pytest.raises(ValueError, match="host_failure attempt disagrees"): + _decide(ahead) + + # An agreeing journal still decides normally, so the binding is not a blanket + # rejection of the retryable path. + agreeing = _failed_journal(kind="provider_capacity", attempt=1) + decided = _decide(agreeing) + assert decided["disposition"] == "wait" + assert decided["attempt"] == 1 + + def test_foreign_lineage_is_refused() -> None: journal = _failed_journal( lineage={"goal_id": "other-goal", "agent_id": AGENT_ID, "todo_id": TODO_ID} From 9a154124c07f1b1faa5f933defefd971461df336 Mon Sep 17 00:00:00 2001 From: song Date: Mon, 14 Sep 2026 20:43:59 +0800 Subject: [PATCH 4/4] fix(turn): validate the current journal through its canonical interpreter Signed-off-by: song --- .../protocols/turn-loop-controller-v0.md | 7 +++-- ...loopx-turn-managed-step-self-heal-smoke.py | 29 ++++++++++++++++-- .../control_plane/turn_driver/managed_step.py | 30 ++++++++++--------- tests/test_loopx_turn_managed_step.py | 26 ++++++++++++++-- 4 files changed, 71 insertions(+), 21 deletions(-) diff --git a/docs/reference/protocols/turn-loop-controller-v0.md b/docs/reference/protocols/turn-loop-controller-v0.md index 72192535f7..26669c80c9 100644 --- a/docs/reference/protocols/turn-loop-controller-v0.md +++ b/docs/reference/protocols/turn-loop-controller-v0.md @@ -203,8 +203,11 @@ The Turn Journal remains the sole authority for the attempt count and retry ceiling. `--observed-attempt` and `--observed-max-attempts` are reconciled against it and refused on disagreement, so a caller's bookkeeping can be checked but never substituted. A Journal that is not a finished failed Turn, -whose typed host failure is not retryable, or whose recovery plan is `blocked` -is refused before the transition is reached. +whose typed host failure is not retryable, or whose current snapshot fails the +canonical TypeScript journal consistency checks is refused before the transition +is reached. A stored recovery audit describes an earlier attempt, not current +eligibility. The eventual `run-once` still revalidates host-session binding and +execution authority before retrying. ## Boundary diff --git a/examples/loopx-turn-managed-step-self-heal-smoke.py b/examples/loopx-turn-managed-step-self-heal-smoke.py index ed2ec603ae..2a90cfa292 100644 --- a/examples/loopx-turn-managed-step-self-heal-smoke.py +++ b/examples/loopx-turn-managed-step-self-heal-smoke.py @@ -337,8 +337,7 @@ def main() -> int: spends_after_failure = _quota_spend_count(runtime) # 2. The managed step must answer wait without touching anything. - _, step = _run_cli( - [ + step_argv = [ "--registry", str(registry), "--runtime-root", @@ -364,7 +363,10 @@ def main() -> int: "--observed-max-attempts", "3", ] - ) + journal_path = _journal_path(runtime, turn_key) + journal_before = journal_path.read_bytes() + _, step = _run_cli(step_argv) + assert journal_path.read_bytes() == journal_before assert step["ok"] is True, step assert step["disposition"] == "wait", step continuation = step["retry_continuation"] @@ -380,6 +382,27 @@ def main() -> int: "the managed step changed the spend ledger" ) + # A previous audit is not proof that the current journal is safe. + # Mutate only this disposable fixture, then restore it for the real + # successful retry below. Never substitute the checker result. + from loopx.control_plane.turn_driver import inspect_loopx_turn_journal + corrupt = json.loads(journal_before) + corrupt["completed_phases"] = ["quota_spend"] + try: + journal_path.write_text(json.dumps(corrupt), encoding="utf-8") + inspection = inspect_loopx_turn_journal( + runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, + turn_key=turn_key, retry_failed=True, + ) + assert inspection["recovery_decision"]["action"] == "blocked" + code, rejected = _run_cli(step_argv) + assert code == 1 and rejected["ok"] is False, rejected + assert "completed_phases_not_ordered_prefix" in rejected["error"] + assert "retry_continuation" not in rejected + assert _quota_spend_count(runtime) == spends_after_failure + finally: + journal_path.write_bytes(journal_before) + # 3. Replaying the same Turn must self-heal and spend exactly once. exit_code, healed = _run_cli(_resume_argv(base, turn_key)) assert exit_code == 0, (exit_code, healed) diff --git a/loopx/control_plane/turn_driver/managed_step.py b/loopx/control_plane/turn_driver/managed_step.py index 739cff1409..5f2770a0be 100644 --- a/loopx/control_plane/turn_driver/managed_step.py +++ b/loopx/control_plane/turn_driver/managed_step.py @@ -24,7 +24,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any +from typing import Any, cast from .executor import LOOPX_TURN_EXECUTION_SCHEMA_VERSION from .host_failure import normalize_host_failure_record, project_host_failure @@ -33,6 +33,7 @@ ValidatedTurnReceipt, decide_loop_disposition, ) +from .turn_journal_runtime import interpret_turn_journal_projection from .transaction import ( LOOPX_TURN_RECEIPT_VALIDATION_SCHEMA_VERSION, LoopXTurnResultKind, @@ -40,10 +41,6 @@ LOOPX_TURN_MANAGED_STEP_SCHEMA_VERSION = "loopx_turn_managed_step_v0" -# A blocked recovery audit means the journal is not resumable for this -# identity; the controller decision must never be reached in that case. -_REPLAY_BLOCKED_ACTION = "blocked" - def _mapping(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, Mapping) else {} @@ -136,9 +133,9 @@ def managed_step_receipt_from_journal( """Qualify one journaled Turn for the managed-step transition. Fails closed unless the journal is a finished failed Turn for exactly this - goal/agent/turn identity whose typed host failure is retryable. A recovery - audit whose planned action is ``blocked`` is refused here rather than at the - controller, because an unsafe journal must not reach the transition at all. + goal/agent/turn identity whose typed host failure is retryable. The current + snapshot must pass the canonical TS journal consistency check before the + controller is called; historical recovery audits never establish eligibility. """ if journal.get("status") != "failed": @@ -149,12 +146,6 @@ def managed_step_receipt_from_journal( if result_kind != LoopXTurnResultKind.HOST_FAILURE.value: raise ValueError("managed step requires a typed host failure Turn") - recovery = _mapping(journal.get("recovery_audit")) - planned = _mapping(recovery.get("planned")) - action = str(planned.get("action") or "") - if action == _REPLAY_BLOCKED_ACTION: - raise ValueError("Turn journal replay is blocked for this identity") - failure = normalize_host_failure_record(journal.get("host_failure")) if failure.get("retryable") is not True: raise ValueError( @@ -176,6 +167,17 @@ def managed_step_receipt_from_journal( lineage = receipt.lineage if lineage["goal_id"] != goal_id or lineage["agent_id"] != agent_id: raise ValueError("Turn journal lineage does not match the requested goal/agent") + # Historical recovery_audit is explanatory, not proof about this snapshot. + # Reuse the same TS consistency rules as inspect-journal/run-once. This is + # not a retry request: fresh host-session validation remains with run-once, + # while the controller below owns the bounded retry disposition. + inspection = interpret_turn_journal_projection( + journal, goal_id=goal_id, agent_id=agent_id, turn_key=turn_key, + ) + if inspection["journal_consistent"] is not True: + raise ValueError( + "Turn journal replay is blocked: " + ", ".join(cast(list[str], inspection["violations"])) + ) return receipt diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index fc9163a1b6..895fd8b92e 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -64,7 +64,10 @@ def _failed_journal( execution_mode="interactive-visible", session_action="resume", ) - plan = {"transaction": transaction} + plan = {"transaction": transaction, "turn_envelope": { + **(_lineage() if lineage is None else lineage), + "action": {"selected_todo": {"todo_id": (_lineage() if lineage is None else lineage)["todo_id"]}}, + }} failure_kind = result_kind in { LoopXTurnResultKind.HOST_FAILURE, LoopXTurnResultKind.VALIDATION_FAILED, @@ -92,6 +95,7 @@ def _failed_journal( assert receipt["ok"] is True, receipt journal: dict[str, Any] = { "schema_version": "loopx_turn_journal_v0", + "goal_id": (_lineage() if lineage is None else lineage)["goal_id"], "status": status, "turn_key": transaction["turn_key"], "result_kind": result_kind.value, @@ -283,7 +287,7 @@ def test_foreign_lineage_is_refused() -> None: _decide(journal) -def test_blocked_recovery_audit_is_refused() -> None: +def test_historical_recovery_audit_does_not_override_current_journal() -> None: journal = _failed_journal(kind="provider_capacity", attempt=1) journal["recovery_audit"] = { "schema_version": "loopx_turn_recovery_audit_v0", @@ -305,6 +309,24 @@ def test_blocked_recovery_audit_is_refused() -> None: }, } + assert _decide(journal)["disposition"] == "wait" + # The inverse matters too: a previous success audit cannot admit a now + # inconsistent journal. The TS owner independently defines phase order. + journal["recovery_audit"]["planned"]["action"] = "continue" + journal["completed_phases"] = ["quota_spend"] + with pytest.raises(ValueError, match="completed_phases_not_ordered_prefix"): + _decide(journal) + + +@pytest.mark.parametrize("mutation", ["goal", "receipt_key", "settlement"]) +def test_current_journal_uses_canonical_identity_checks(mutation): + journal = _failed_journal() + if mutation == "goal": + journal["goal_id"] = "other-goal" + elif mutation == "receipt_key": + journal["receipt"]["turn_key"] = "sha256:" + "0" * 64 + else: + journal["plan"]["transaction"]["settlement_plan"]["identity"]["effect_id"] = "invalid" with pytest.raises(ValueError, match="replay is blocked"): _decide(journal)