From 18e2d8366241f77a5f4ab649e7e297e60c95fe7c Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 15 Sep 2026 10:12:41 +0800 Subject: [PATCH 1/4] feat(turn): report the managed executor and fail closed when it cannot launch `loopx turn plan` and `loopx turn run-once` now carry a typed `managed_executor` block naming the planned executor, whether it is bound to an operator credential or to an individual CLI host, and whether LoopX can prove it launches here. A `run-once --execute` whose planned host reports `available: false` fails closed with status `unavailable`: it invokes no host, writes no journal, and spends no quota slot, so a Turn never moves onto another executor on its own. The credential-resolved default also has to pair its host with a schedulable execution mode: a managed default now plans `isolated-headless` instead of a visible interactive mode that the `outer_controller` scheduler context rejects, so the shipped default is usable end to end. Coverage: a readback matrix for the binding, executor coverage for the refusal and preview paths, CLI default-mode coverage, and a hermetic smoke that runs both the readback and the fail-closed refusal through the CLI. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...opx-turn-managed-executor-binding-smoke.py | 274 ++++++++++++++++++ loopx/cli_commands/turn.py | 9 + loopx/cli_commands/turn_registration.py | 16 +- loopx/control_plane/turn_driver/executor.py | 25 ++ .../control_plane/turn_driver/host_binding.py | 89 +++++- tests/test_loopx_turn_executor.py | 92 ++++++ tests/test_turn_default_host_binding.py | 22 ++ tests/test_turn_managed_executor_binding.py | 126 ++++++++ 8 files changed, 650 insertions(+), 3 deletions(-) create mode 100644 examples/loopx-turn-managed-executor-binding-smoke.py create mode 100644 tests/test_turn_managed_executor_binding.py diff --git a/examples/loopx-turn-managed-executor-binding-smoke.py b/examples/loopx-turn-managed-executor-binding-smoke.py new file mode 100644 index 0000000000..aec89953a1 --- /dev/null +++ b/examples/loopx-turn-managed-executor-binding-smoke.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +"""Prove the managed executor readback and the fail-closed start it drives.""" + +from __future__ import annotations + +import contextlib +import importlib.abc +import io +import json +import os +import sys +import tempfile +from collections.abc import Iterator +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 +from loopx.control_plane.turn_driver import executor as turn_executor # noqa: E402 +from loopx.control_plane.turn_driver.host_binding import ( # noqa: E402 + DSH_RUNTIME_UNAVAILABLE, + EXECUTOR_KIND_INDIVIDUAL, + EXECUTOR_KIND_MANAGED, +) + + +GOAL_ID = "loopx-turn-managed-executor-fixture" +AGENT_ID = "codex-managed-executor-fixture" +TODO_ID = "todo_managedexec01" +CREDENTIAL_ENV = "DEEPSEEK_API_KEY" + + +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) + + 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", + "---", + "", + "# LoopX Managed Executor Fixture", + "", + "## Next Action", + "", + "Run one bounded managed Turn on the planned executor only.", + "", + "## Agent Todo", + "", + "- [ ] [P1] Run one bounded managed Turn without leaving the planned executor.", + ( + 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-public-fixture", + "status": "active", + "repo": str(project), + "state_file": str(state.relative_to(project)), + "adapter": {"kind": "fixture_v0", "status": "connected-delivery"}, + "quota": {"compute": 10.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 + + +class _UnavailableHarnessRuntime(importlib.abc.MetaPathFinder): + """Make the DeepSeek Harness runtime unimportable for one bounded call.""" + + def find_spec(self, fullname, path=None, target=None): + if fullname == "deepseek_harness" or fullname.startswith("deepseek_harness."): + raise ModuleNotFoundError(fullname) + return None + + +@contextlib.contextmanager +def _harness_runtime_unavailable() -> Iterator[None]: + finder = _UnavailableHarnessRuntime() + sys.meta_path.insert(0, finder) + try: + yield + finally: + sys.meta_path.remove(finder) + + +@contextlib.contextmanager +def _operator_credential(value: str | None) -> Iterator[None]: + previous = os.environ.get(CREDENTIAL_ENV) + if value is None: + os.environ.pop(CREDENTIAL_ENV, None) + else: + os.environ[CREDENTIAL_ENV] = value + try: + yield + finally: + if previous is None: + os.environ.pop(CREDENTIAL_ENV, None) + else: + os.environ[CREDENTIAL_ENV] = previous + + +def _run_cli(argv: list[str]) -> tuple[int, dict[str, Any]]: + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main(argv) + return exit_code, json.loads(output.getvalue()) + + +def _plan_command(registry: Path, runtime: Path, project: Path) -> list[str]: + return [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "plan", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--scan-root", + str(project), + ] + + +def _run_once_command(registry: Path, runtime: Path, project: Path, workspace: 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-executor-fail-closed", + "--project", + str(workspace), + "--scan-root", + str(project), + "--no-global-sync", + "--execute", + ] + + +def _expect_managed_binding(payload: dict[str, Any]) -> dict[str, Any]: + binding = payload["managed_executor"] + assert binding["schema_version"] == "managed_executor_binding_v0", binding + assert binding["executor"] == payload["host"]["kind"], binding + assert binding["executor_kind"] == EXECUTOR_KIND_MANAGED, binding + assert binding["credential_env"] == CREDENTIAL_ENV, binding + assert isinstance(binding["available"], bool), binding + assert (binding["unavailable_reason"] is None) is binding["available"], binding + return binding + + +def main() -> int: + with tempfile.TemporaryDirectory(prefix="loopx-turn-managed-executor-") as directory: + root = Path(directory) + project, runtime, workspace, registry = _write_fixture(root) + + # 1. No operator credential: the default stays an individual CLI host and + # the readback makes no launch claim for it. + with _operator_credential(None): + exit_code, payload = _run_cli(_plan_command(registry, runtime, project)) + assert exit_code == 0, payload + assert payload["host"]["kind"] == "codex-cli", payload + individual = payload["managed_executor"] + assert individual["executor_kind"] == EXECUTOR_KIND_INDIVIDUAL, individual + assert individual["available"] is None, individual + assert individual["unavailable_reason"] is None, individual + + # 2. A configured credential selects the managed executor, and the + # readback names the operator environment it is bound to. + with _operator_credential("sk-fixture-operator"): + exit_code, payload = _run_cli(_plan_command(registry, runtime, project)) + assert exit_code == 0, payload + assert payload["host"]["kind"] == "dsh", payload + _expect_managed_binding(payload) + + # 3. With the runtime genuinely missing, the same plan reports an + # unavailable managed executor instead of promising a launch. + with _operator_credential("sk-fixture-operator"): + with _harness_runtime_unavailable(): + exit_code, payload = _run_cli(_plan_command(registry, runtime, project)) + assert exit_code == 0, payload + assert payload["host"]["kind"] == "dsh", payload + unavailable = _expect_managed_binding(payload) + assert unavailable["available"] is False, unavailable + assert unavailable["unavailable_reason"] == DSH_RUNTIME_UNAVAILABLE, unavailable + + # 4. Executing that Turn fails closed: typed status, no host invocation, + # no journal, and no quota slot spend. + with _operator_credential("sk-fixture-operator"): + with _harness_runtime_unavailable(): + exit_code, refusal = _run_cli( + _run_once_command(registry, runtime, project, workspace) + ) + assert exit_code == 1, refusal + assert refusal["ok"] is False, refusal + assert refusal["status"] == "unavailable", refusal + assert refusal["reason"] == DSH_RUNTIME_UNAVAILABLE, refusal + assert refusal["effects"] == { + "host_invoked": False, + "state_written": False, + "quota_spent": False, + "scheduler_acknowledged": False, + }, refusal + assert refusal["quota_slot_spend_count"] == 0, refusal + journal = turn_executor.turn_journal_path( + runtime, + goal_id=GOAL_ID, + turn_key=str(refusal["resume_turn_key"]), + ) + assert journal.exists() is False, journal + + print("managed executor binding smoke passed") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index a07209d970..f95247bff0 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -55,6 +55,7 @@ run_loopx_turn_once, selected_turn_todo, ) +from ..control_plane.turn_driver.host_binding import managed_executor_binding from ..quota import spend_quota_slot from ..state_refresh import refresh_state_run from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status @@ -239,6 +240,14 @@ def build_turn_decision( turn_instance_id=args.turn_instance_id, iteration_context_policy=args.iteration_context.replace("-", "_"), ) + # The executor readback names where this Turn's model work runs and + # whether that host can launch here, so a caller never has to infer it + # from the host id. The explicit runner hook is the one launchability + # fact only this command layer knows. + payload["managed_executor"] = managed_executor_binding( + args.host, + dsh_runner_configured=bool(getattr(args, "dsh_runner", None)), + ) if ( args.turn_command == "run-once" and args.execute diff --git a/loopx/cli_commands/turn_registration.py b/loopx/cli_commands/turn_registration.py index c6e3bde8d9..eb32769fb1 100644 --- a/loopx/cli_commands/turn_registration.py +++ b/loopx/cli_commands/turn_registration.py @@ -5,7 +5,10 @@ import argparse from collections.abc import Callable -from ..control_plane.turn_driver.host_binding import resolve_default_turn_host +from ..control_plane.turn_driver.host_binding import ( + HOST_WITH_OPERATOR_CREDENTIAL, + resolve_default_turn_host, +) from ..paths import default_public_scan_root # Explicit host choices stay per-command: planning may name any host the Turn @@ -47,10 +50,19 @@ def register_turn_commands( help="Build one typed read-only host decision without launching or writing.", ) add_subcommand_format(plan) + # The default host and the default execution mode are one decision: a + # managed host runs bounded headless Turns, so pairing it with a visible + # interactive mode would produce a default plan that cannot be scheduled. + resolved_default_host = resolve_default_turn_host() _add_turn_decision_arguments( plan, - default_host=resolve_default_turn_host(), + default_host=resolved_default_host, host_choices=list(PLANNED_TURN_HOST_CHOICES), + default_execution_mode=( + "isolated-headless" + if resolved_default_host == HOST_WITH_OPERATOR_CREDENTIAL + else "interactive-visible" + ), ) plan.add_argument( "--include-transaction-detail", diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 634b26a854..e986fdf200 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -778,6 +778,11 @@ def _execution_payload( "status": journal.get("status"), "execution_mode": planned_host.get("execution_mode"), "host": journal.get("host"), + **( + {"managed_executor": dict(plan["managed_executor"])} + if isinstance(plan.get("managed_executor"), Mapping) + else {} + ), "result_kind": journal.get("result_kind"), "validation": journal.get("task_validation"), "receipt": journal.get("receipt"), @@ -1306,6 +1311,26 @@ def run_loopx_turn_once( "quota_spent": False, "scheduler_acknowledged": False, } + managed_executor = ( + plan.get("managed_executor") + if isinstance(plan.get("managed_executor"), Mapping) + else {} + ) + if execute and managed_executor.get("available") is False: + # Fail closed on an executor LoopX can prove cannot launch here: report + # the planned executor and stop before the journal, the host, and quota + # so the Turn cannot quietly move onto a different executor instead. + return _execution_payload( + plan, + { + "status": "unavailable", + "host": host_projection, + "reason": str(managed_executor.get("unavailable_reason") or ""), + }, + execute=True, + replayed=False, + effects=empty_effects, + ) if not execute: preview = { "schema_version": LOOPX_TURN_JOURNAL_SCHEMA_VERSION, diff --git a/loopx/control_plane/turn_driver/host_binding.py b/loopx/control_plane/turn_driver/host_binding.py index 5afab83a51..d8f59a5052 100644 --- a/loopx/control_plane/turn_driver/host_binding.py +++ b/loopx/control_plane/turn_driver/host_binding.py @@ -1,4 +1,4 @@ -"""Credential-resolved default Turn host binding. +"""Credential-resolved default Turn host binding and managed executor readback. The default host for ``loopx turn plan`` and ``loopx turn run-once`` is decided by what the operator configured, not by the harness alone: @@ -11,12 +11,21 @@ Resolution is a pure function of the environment so the command defaults, the Turn plan readback, and tests quote one rule instead of drifting apart. + +``managed_executor_binding`` turns the same facts into the readback a caller can +act on before a Turn runs: which executor the plan would use, whether that +executor is bound to an operator credential or to an individual CLI host, and +whether the managed host can actually launch here. An executor LoopX can prove +cannot launch is reported as unavailable so the Turn fails closed instead of +drifting onto another executor. """ from __future__ import annotations +import importlib.util import os from collections.abc import Mapping +from typing import Any, Callable HOST_WITH_OPERATOR_CREDENTIAL = "dsh" HOST_WITHOUT_OPERATOR_CREDENTIAL = "codex-cli" @@ -27,6 +36,23 @@ OPERATOR_CREDENTIAL_ENV_VARS = ("DEEPSEEK_API_KEY",) OPERATOR_ENDPOINT_ENV_VAR = "DEEPSEEK_BASE_URL" +MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION = "managed_executor_binding_v0" +# Executor kinds name where a Turn's model work is billed and bounded rather +# than which adapter is launched: a managed executor runs on an +# operator-supplied credential, an individual executor on one person's own CLI +# login, and a generic executor on a caller-supplied adapter command. +EXECUTOR_KIND_MANAGED = "managed" +EXECUTOR_KIND_INDIVIDUAL = "individual" +EXECUTOR_KIND_GENERIC = "generic" +INDIVIDUAL_CLI_HOSTS = frozenset({"codex-cli", "claude-code"}) +MANAGED_HOST = HOST_WITH_OPERATOR_CREDENTIAL + +# The built-in dsh host launches the DeepSeek Harness runtime unless the caller +# supplies the explicit runner hook, so that module being importable is the +# launchability fact this projection checks without side effects. +DSH_RUNTIME_MODULE = "deepseek_harness" +DSH_RUNTIME_UNAVAILABLE = "dsh_runtime_unavailable" + def configured_operator_credential( environ: Mapping[str, str] | None = None, @@ -46,3 +72,64 @@ def resolve_default_turn_host(environ: Mapping[str, str] | None = None) -> str: if configured_operator_credential(environ) is not None: return HOST_WITH_OPERATOR_CREDENTIAL return HOST_WITHOUT_OPERATOR_CREDENTIAL + + +def _configured_env_name(name: str, environ: Mapping[str, str] | None) -> str | None: + source = os.environ if environ is None else environ + return name if str(source.get(name, "") or "").strip() else None + + +def dsh_runtime_importable( + module_probe: Callable[[str], bool] | None = None, +) -> bool: + """Whether the DeepSeek Harness runtime the built-in dsh host launches exists.""" + + if module_probe is not None: + return bool(module_probe(DSH_RUNTIME_MODULE)) + try: + return importlib.util.find_spec(DSH_RUNTIME_MODULE) is not None + except (ImportError, ValueError): + return False + + +def managed_executor_binding( + host: str, + *, + environ: Mapping[str, str] | None = None, + dsh_runner_configured: bool = False, + module_probe: Callable[[str], bool] | None = None, +) -> dict[str, Any]: + """Project the executor one planned Turn would run on. + + ``available`` is ``False`` only when LoopX can prove the planned executor + cannot launch here, which is what a caller has to fail closed on. ``None`` + records that this projection does not probe that executor kind, so it makes + no claim rather than an unproven ``True``. + """ + + if host == MANAGED_HOST: + launchable = bool( + dsh_runner_configured or dsh_runtime_importable(module_probe) + ) + return { + "schema_version": MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION, + "executor": host, + "executor_kind": EXECUTOR_KIND_MANAGED, + "credential_env": configured_operator_credential(environ), + "endpoint_env": _configured_env_name(OPERATOR_ENDPOINT_ENV_VAR, environ), + "available": launchable, + "unavailable_reason": None if launchable else DSH_RUNTIME_UNAVAILABLE, + } + return { + "schema_version": MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION, + "executor": host, + "executor_kind": ( + EXECUTOR_KIND_INDIVIDUAL + if host in INDIVIDUAL_CLI_HOSTS + else EXECUTOR_KIND_GENERIC + ), + "credential_env": None, + "endpoint_env": None, + "available": None, + "unavailable_reason": None, + } diff --git a/tests/test_loopx_turn_executor.py b/tests/test_loopx_turn_executor.py index 2bc97abaa5..5191f8b791 100644 --- a/tests/test_loopx_turn_executor.py +++ b/tests/test_loopx_turn_executor.py @@ -24,7 +24,9 @@ BuiltInHostError, LOOPX_TURN_JOURNAL_SCHEMA_VERSION, _task_validation_stage, + turn_journal_path, ) +from loopx.control_plane.turn_driver.host_binding import managed_executor_binding from loopx.control_plane.turn_driver.settlement import execute_turn_driver_settlement from loopx.control_plane.turn_driver.transaction import TRANSACTION_PHASES @@ -77,6 +79,25 @@ def _codex_plan() -> dict[str, object]: ) +def _managed_plan(*, runtime_available: bool) -> dict[str, object]: + """One dsh plan carrying the executor readback the command layer attaches.""" + + plan = _plan() + envelope = plan["turn_envelope"] + assert isinstance(envelope, dict) + managed = build_loopx_turn_plan( + envelope, + host="dsh", + execution_mode="isolated-headless", + ) + managed["managed_executor"] = managed_executor_binding( + "dsh", + environ={"DEEPSEEK_API_KEY": "fixture-operator-credential"}, + module_probe=lambda _module: runtime_available, + ) + return managed + + def _adaptive_observation_plan( *, required_write_scopes: list[str] | None = None, @@ -2501,3 +2522,74 @@ def scheduler(_spend: dict[str, object]) -> dict[str, object]: turn_key=str(transaction["turn_key"]), ) assert audited["last_recovery"] == resumed["recovery"] + + +def test_run_once_fails_closed_when_the_managed_executor_cannot_launch(tmp_path): + plan = _managed_plan(runtime_available=False) + transaction = plan["transaction"] + assert isinstance(transaction, dict) + runtime_root = tmp_path / "runtime" + journal = turn_journal_path( + runtime_root, + goal_id="fixture-goal", + turn_key=str(transaction["turn_key"]), + ) + + payload = run_loopx_turn_once( + plan, + host_runner=lambda _request: pytest.fail("an unavailable executor must not run"), + project=tmp_path, + runtime_root=runtime_root, + goal_id="fixture-goal", + timeout_seconds=5, + execute=True, + ) + + assert payload["ok"] is False + assert payload["status"] == "unavailable" + assert payload["reason"] == "dsh_runtime_unavailable" + assert payload["effects"] == { + "host_invoked": False, + "state_written": False, + "quota_spent": False, + "scheduler_acknowledged": False, + } + assert payload["quota_slot_spend_count"] == 0 + assert payload["managed_executor"] == plan["managed_executor"] + assert journal.exists() is False + + +def test_run_once_preview_reports_the_managed_executor_without_refusing(tmp_path): + plan = _managed_plan(runtime_available=False) + + payload = run_loopx_turn_once( + plan, + host_runner=lambda _request: pytest.fail("preview must not run the host"), + project=tmp_path, + runtime_root=tmp_path / "runtime", + goal_id="fixture-goal", + timeout_seconds=5, + execute=False, + ) + + assert payload["ok"] is True + assert payload["status"] == "preview" + assert payload["managed_executor"]["available"] is False + assert payload["managed_executor"]["unavailable_reason"] == "dsh_runtime_unavailable" + + +def test_run_once_does_not_refuse_a_launchable_managed_executor(tmp_path): + plan = _managed_plan(runtime_available=True) + + # The refusal is the only guard under test here: without writeback, spend, + # and scheduler callbacks the executor stops at its own contract instead. + with pytest.raises(ValueError, match="requires writeback, spend, and scheduler"): + run_loopx_turn_once( + plan, + host_runner=lambda _request: pytest.fail("host must not run without callbacks"), + project=tmp_path, + runtime_root=tmp_path / "runtime", + goal_id="fixture-goal", + timeout_seconds=5, + execute=True, + ) diff --git a/tests/test_turn_default_host_binding.py b/tests/test_turn_default_host_binding.py index 74e3cb9a50..7a928c5fc8 100644 --- a/tests/test_turn_default_host_binding.py +++ b/tests/test_turn_default_host_binding.py @@ -75,3 +75,25 @@ def test_explicit_host_still_wins_over_the_credential_default(monkeypatch): ) assert args.host == "generic-cli" + + +@pytest.mark.parametrize("command", ["plan", "run-once"]) +def test_default_execution_mode_matches_the_resolved_default_host( + command, monkeypatch +): + monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-operator") + + managed = build_parser().parse_args(_turn_argv(command)) + + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + individual = build_parser().parse_args(_turn_argv(command)) + + # A managed host plans bounded headless Turns; pairing it with a visible + # interactive mode would make the shipped default unusable. run-once only + # ships the isolated-headless mode, so it keeps that mode either way. + assert managed.host == HOST_WITH_OPERATOR_CREDENTIAL + assert managed.execution_mode == "isolated-headless" + assert individual.host == HOST_WITHOUT_OPERATOR_CREDENTIAL + assert individual.execution_mode == ( + "interactive-visible" if command == "plan" else "isolated-headless" + ) diff --git a/tests/test_turn_managed_executor_binding.py b/tests/test_turn_managed_executor_binding.py new file mode 100644 index 0000000000..b4ce70e596 --- /dev/null +++ b/tests/test_turn_managed_executor_binding.py @@ -0,0 +1,126 @@ +"""The managed executor readback names the executor and whether it can launch.""" + +from __future__ import annotations + +import pytest + +from loopx.control_plane.turn_driver.host_binding import ( + DSH_RUNTIME_UNAVAILABLE, + EXECUTOR_KIND_GENERIC, + EXECUTOR_KIND_INDIVIDUAL, + EXECUTOR_KIND_MANAGED, + MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION, + configured_operator_credential, + managed_executor_binding, + resolve_default_turn_host, +) + +_NO_RUNTIME = lambda _module: False # noqa: E731 - tiny probe fixture +_RUNTIME = lambda _module: True # noqa: E731 - tiny probe fixture + + +def test_managed_executor_reports_the_operator_credential_and_endpoint(): + binding = managed_executor_binding( + "dsh", + environ={ + "DEEPSEEK_API_KEY": "sk-operator", + "DEEPSEEK_BASE_URL": "https://example.invalid", + }, + module_probe=_RUNTIME, + ) + + assert binding == { + "schema_version": MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION, + "executor": "dsh", + "executor_kind": EXECUTOR_KIND_MANAGED, + "credential_env": "DEEPSEEK_API_KEY", + "endpoint_env": "DEEPSEEK_BASE_URL", + "available": True, + "unavailable_reason": None, + } + + +def test_managed_executor_fails_closed_when_the_runtime_is_missing(): + binding = managed_executor_binding( + "dsh", + environ={"DEEPSEEK_API_KEY": "sk-operator"}, + module_probe=_NO_RUNTIME, + ) + + assert binding["available"] is False + assert binding["unavailable_reason"] == DSH_RUNTIME_UNAVAILABLE + + +def test_configured_runner_hook_makes_the_managed_host_launchable(): + binding = managed_executor_binding( + "dsh", + environ={"DEEPSEEK_API_KEY": "sk-operator"}, + dsh_runner_configured=True, + module_probe=_NO_RUNTIME, + ) + + assert binding["available"] is True + assert binding["unavailable_reason"] is None + + +def test_managed_executor_reports_an_unconfigured_credential_without_inventing_one(): + binding = managed_executor_binding("dsh", environ={}, module_probe=_RUNTIME) + + assert binding["credential_env"] is None + assert binding["endpoint_env"] is None + assert binding["executor_kind"] == EXECUTOR_KIND_MANAGED + + +@pytest.mark.parametrize( + ("host", "expected_kind"), + [ + ("codex-cli", EXECUTOR_KIND_INDIVIDUAL), + ("claude-code", EXECUTOR_KIND_INDIVIDUAL), + ("generic-cli", EXECUTOR_KIND_GENERIC), + ], +) +def test_other_hosts_make_no_launch_claim_and_carry_no_operator_env( + host, expected_kind +): + binding = managed_executor_binding( + host, + environ={"DEEPSEEK_API_KEY": "sk-operator"}, + module_probe=_RUNTIME, + ) + + assert binding["executor_kind"] == expected_kind + assert binding["available"] is None + assert binding["unavailable_reason"] is None + assert binding["credential_env"] is None + assert binding["endpoint_env"] is None + + +@pytest.mark.parametrize( + "environ", + [{}, {"DEEPSEEK_API_KEY": ""}, {"DEEPSEEK_API_KEY": " "}], +) +def test_default_resolution_and_executor_kind_agree(environ): + default_host = resolve_default_turn_host(environ) + binding = managed_executor_binding( + default_host, + environ=environ, + module_probe=_RUNTIME, + ) + + # A machine with an operator credential resolves to the managed executor and + # one without it resolves to the individual CLI host, so the readback never + # claims managed execution the default would not actually select. + expected_kind = ( + EXECUTOR_KIND_MANAGED + if configured_operator_credential(environ) + else EXECUTOR_KIND_INDIVIDUAL + ) + assert binding["executor_kind"] == expected_kind + + +def test_endpoint_without_credential_is_reported_but_does_not_switch_host(): + environ = {"DEEPSEEK_BASE_URL": "https://example.invalid"} + binding = managed_executor_binding("codex-cli", environ=environ) + + assert resolve_default_turn_host(environ) == "codex-cli" + assert binding["endpoint_env"] is None From 5740c4f8e2835a55af9724a26b7fb726d222cae1 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 15 Sep 2026 10:12:54 +0800 Subject: [PATCH 2/4] docs(turn): document the managed executor readback and fail-closed start Describe the shipped `managed_executor` block, what `executor_kind` means, when `available` is false versus null, and the fail-closed contract of an explicitly executing Turn whose planned host cannot launch. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../deepseek-harness-connector.md | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/docs/integrations/deepseek-harness-connector.md b/docs/integrations/deepseek-harness-connector.md index 9d5a52d30b..91b7729edd 100644 --- a/docs/integrations/deepseek-harness-connector.md +++ b/docs/integrations/deepseek-harness-connector.md @@ -88,6 +88,40 @@ the resolved value is what `loopx turn plan` reports back. `loopx turn run-once` accepts the `codex-cli`, `dsh`, and `generic-cli` hosts it ships adapters for; `loopx turn plan` additionally accepts the planning-only `claude-code` host. +## Managed Executor Readback + +Both `loopx turn plan` and `loopx turn run-once` report a +`managed_executor` block, so a caller reads the planned executor instead of +inferring it from a host id: + +```json +{ + "schema_version": "managed_executor_binding_v0", + "executor": "dsh", + "executor_kind": "managed", + "credential_env": "DEEPSEEK_API_KEY", + "endpoint_env": "DEEPSEEK_BASE_URL", + "available": true, + "unavailable_reason": null +} +``` + +`executor_kind` names where the Turn's model work is billed and bounded: +`managed` for a host bound to an operator credential, `individual` for a host +that runs on one person's own CLI login, and `generic` for a caller-supplied +adapter command. `available` is `false` only when LoopX can prove the planned +host cannot launch here; it is `null` for executors this projection does not +probe rather than an unproven claim. + +A `run-once --execute` whose planned host reports `available: false` fails +closed: it reports the status `unavailable` with the typed +`unavailable_reason`, invokes no host, writes no journal, and spends no quota +slot. The Turn never moves onto a different executor on its own; an operator who +wants another host names it explicitly. `--dsh-runner` records that the caller +supplied its own runner, so a hermetic test hook counts as a launchable managed +host. Without that flag the built-in host needs the DeepSeek Harness runtime +from the `loopx[deepseek-harness]` extra. + ## Onboard ```bash From cbc08e36a664a24f75d03421344ef6e31f55663b Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:45:16 +0800 Subject: [PATCH 3/4] refactor(turn): quote the planned managed executor from the binding owner Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- loopx/control_plane/turn_driver/executor.py | 33 ++++++---------- .../control_plane/turn_driver/host_binding.py | 38 +++++++++++++++++++ 2 files changed, 49 insertions(+), 22 deletions(-) diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index e986fdf200..854586f32d 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -28,6 +28,10 @@ reward_memory_reflection_digest, ) from .driver import selected_turn_todo +from .host_binding import ( + managed_executor_payload_entry, + managed_executor_unavailable_payload, +) from .host_failure import BuiltInHostError, project_host_failure, record_host_failure from .journal_store import ( LOOPX_TURN_JOURNAL_SCHEMA_VERSION, @@ -778,11 +782,7 @@ def _execution_payload( "status": journal.get("status"), "execution_mode": planned_host.get("execution_mode"), "host": journal.get("host"), - **( - {"managed_executor": dict(plan["managed_executor"])} - if isinstance(plan.get("managed_executor"), Mapping) - else {} - ), + **managed_executor_payload_entry(plan), "result_kind": journal.get("result_kind"), "validation": journal.get("task_validation"), "receipt": journal.get("receipt"), @@ -1311,25 +1311,14 @@ def run_loopx_turn_once( "quota_spent": False, "scheduler_acknowledged": False, } - managed_executor = ( - plan.get("managed_executor") - if isinstance(plan.get("managed_executor"), Mapping) - else {} + fail_closed = managed_executor_unavailable_payload( + plan, execute=execute, host_projection=host_projection ) - if execute and managed_executor.get("available") is False: - # Fail closed on an executor LoopX can prove cannot launch here: report - # the planned executor and stop before the journal, the host, and quota - # so the Turn cannot quietly move onto a different executor instead. + if fail_closed is not None: + # Fail closed on an executor LoopX can prove cannot launch: report the + # planned executor and stop before the journal, host, and quota. return _execution_payload( - plan, - { - "status": "unavailable", - "host": host_projection, - "reason": str(managed_executor.get("unavailable_reason") or ""), - }, - execute=True, - replayed=False, - effects=empty_effects, + plan, fail_closed, execute=True, replayed=False, effects=empty_effects ) if not execute: preview = { diff --git a/loopx/control_plane/turn_driver/host_binding.py b/loopx/control_plane/turn_driver/host_binding.py index d8f59a5052..a43c4fe9f6 100644 --- a/loopx/control_plane/turn_driver/host_binding.py +++ b/loopx/control_plane/turn_driver/host_binding.py @@ -133,3 +133,41 @@ def managed_executor_binding( "available": None, "unavailable_reason": None, } + + +def managed_executor_payload_entry(plan: Mapping[str, Any]) -> dict[str, Any]: + """Return the execution payload's ``managed_executor`` entry, when planned. + + Reading the entry from the plan keeps one authority for the executor + identity: the payload quotes the binding the plan resolved instead of + re-deriving an executor from the launched host. + """ + + binding = plan.get("managed_executor") + return {"managed_executor": dict(binding)} if isinstance(binding, Mapping) else {} + + +def managed_executor_unavailable_payload( + plan: Mapping[str, Any], + *, + execute: bool, + host_projection: Mapping[str, Any], +) -> dict[str, Any] | None: + """Return the fail-closed execution payload for an unlaunchable executor. + + ``None`` means the plan makes no claim that its executor cannot launch, so + the Turn continues normally. A returned payload stops the Turn before the + journal, the host, and quota with no effect recorded, so a bounded Turn + cannot quietly move onto a different executor than the plan read back. + """ + + if not execute: + return None + binding = plan.get("managed_executor") + if not isinstance(binding, Mapping) or binding.get("available") is not False: + return None + return { + "status": "unavailable", + "host": dict(host_projection), + "reason": str(binding.get("unavailable_reason") or ""), + } From f3cd2bd88a007e726e3448eacda418ef882f2540 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:45:26 +0800 Subject: [PATCH 4/4] test(cli-output): declare the bounded managed executor readback budget Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../control_plane/cli-output-probe-runner.py | 3 ++ .../testing/cli_output_differential.py | 45 +++++++++++++++++++ .../testing/cli_output_semantics.py | 22 +++++++++ .../test_cli_output_differential.py | 45 +++++++++++++++++++ 4 files changed, 115 insertions(+) diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py index 1ad861fb4f..878187ccf1 100644 --- a/examples/control_plane/cli-output-probe-runner.py +++ b/examples/control_plane/cli-output-probe-runner.py @@ -108,6 +108,9 @@ def _receipt_row( "reward_memory_outcome_prompt_revision": ( semantics.reward_memory_outcome_prompt_revision(text) ), + "managed_executor_binding_revision": ( + semantics.managed_executor_binding_revision(text) + ), "guided_todo_delta_schema_versions": ( semantics.guided_todo_delta_schema_versions(payload) if isinstance(payload, dict) diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py index b1315e5357..67baab035c 100644 --- a/loopx/control_plane/testing/cli_output_differential.py +++ b/loopx/control_plane/testing/cli_output_differential.py @@ -177,6 +177,45 @@ class GrowthAllowance: "compact_payload_chars": 640, } +# The Turn plan readback adds one bounded managed-executor binding so a caller +# sees which executor a planned Turn would use and whether it can launch here, +# instead of inferring it from the host id. The allowance is bound to the +# declared none-to-v0 binding transition and to the Turn surfaces that quote +# it; quota, status, and every other agent-facing surface keep the ordinary +# budget, and once v0 is the baseline a v0-to-v0 change receives no allowance. +_MANAGED_EXECUTOR_BINDING_V0_MIGRATION_GROWTH_ALLOWANCE: dict[Metric, int] = { + "chars": 512, + "utf8_bytes": 512, + "lines": 12, + "compact_payload_chars": 448, +} + +_MANAGED_EXECUTOR_BINDING_SURFACES = frozenset( + { + "loopx_turn_plan", + "loopx_turn_plan_transaction_detail", + "loopx_turn_run_once_preview", + } +) + + +def _managed_executor_binding_allowance( + row_id: str, + base: Mapping[str, Any], + candidate: Mapping[str, Any], + metric: Metric, +) -> int: + surface = row_id.partition("/")[2].partition("/")[0] + if ( + row_id.startswith(("surface/", "variant/")) + and surface in _MANAGED_EXECUTOR_BINDING_SURFACES + and base.get("managed_executor_binding_revision") is None + and candidate.get("managed_executor_binding_revision") + == "managed_executor_binding_v0" + ): + return _MANAGED_EXECUTOR_BINDING_V0_MIGRATION_GROWTH_ALLOWANCE[metric] + return 0 + def _reward_memory_outcome_prompt_allowance( row_id: str, @@ -545,6 +584,12 @@ def _compare_row(base: dict[str, Any], candidate: dict[str, Any]) -> dict[str, A candidate, metric, ), + _managed_executor_binding_allowance( + row_id, + base, + candidate, + metric, + ), ) # Thin installed prompts contain bilingual lifecycle instructions. A # small character-level clarification can cost three bytes per CJK diff --git a/loopx/control_plane/testing/cli_output_semantics.py b/loopx/control_plane/testing/cli_output_semantics.py index 412f3bea19..fa32f1f19f 100644 --- a/loopx/control_plane/testing/cli_output_semantics.py +++ b/loopx/control_plane/testing/cli_output_semantics.py @@ -41,6 +41,28 @@ def reward_memory_outcome_prompt_revision(text: str) -> str | None: else None ) + +def managed_executor_binding_revision(text: str) -> str | None: + """Attribute the managed-executor binding readback on a Turn surface. + + This is qualification evidence for the exact projection, never a runtime + classifier: the binding key alone would match prose, so the revision also + requires the executor identity, its launchability claim, and the typed + reason slot that only this readback renders. + """ + + required = ( + '"managed_executor"', + '"executor_kind"', + '"available"', + '"unavailable_reason"', + ) + return ( + "managed_executor_binding_v0" + if all(fragment in text for fragment in required) + else None + ) + _MARKDOWN_HEADING = re.compile(r"^#{1,6}\s+.+$") _RUNTIME_ROOT_COMMAND_ROUTE = re.compile( r"(?m)(?:^|[\"'`])[^\r\n\S]*loopx\s+--runtime-root\s+" diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py index 8cc785ca51..2055d13dac 100644 --- a/tests/control_plane/test_cli_output_differential.py +++ b/tests/control_plane/test_cli_output_differential.py @@ -146,6 +146,51 @@ def test_reward_memory_outcome_prompt_budget_is_one_time_bounded_and_prompt_only ] +def test_managed_executor_binding_budget_is_one_time_bounded_and_turn_only() -> None: + from loopx.control_plane.testing.cli_output_differential import _compare_row + from loopx.control_plane.testing.cli_output_semantics import ( + managed_executor_binding_revision, + ) + + projection = ( + '{\n "managed_executor": {\n' + ' "executor_kind": "managed",\n' + ' "available": true,\n' + ' "unavailable_reason": null\n }\n}' + ) + assert ( + managed_executor_binding_revision(projection) + == "managed_executor_binding_v0" + ) + assert ( + managed_executor_binding_revision( + projection.replace('"unavailable_reason"', '"reason"') + ) + is None + ) + + base = _row(row_id="variant/loopx_turn_run_once_preview/small/json") + current = { + **base, + "chars": base["chars"] + 254, + "utf8_bytes": base["utf8_bytes"] + 254, + "lines": base["lines"] + 9, + "compact_payload_chars": base["compact_payload_chars"] + 205, + "managed_executor_binding_revision": "managed_executor_binding_v0", + } + assert not _compare_row(base, current)["failures"] + assert _compare_row(base, {**current, "chars": base["chars"] + 513})[ + "failures" + ] + assert _compare_row(current, {**current, "chars": current["chars"] + 254})[ + "failures" + ] + other = {**base, "row_id": "surface/status/small/json"} + assert _compare_row(other, {**current, "row_id": other["row_id"]})[ + "failures" + ] + + def test_regular_integration_pr_keeps_requested_cli_output_base() -> None: ancestors = { ("origin/main", "HEAD"),