diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 2e9f6beac0..f16a4a5cf2 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -956,6 +956,9 @@ jobs: - name: Qualify the native Windows CLI output differential timeout-minutes: 12 + env: + LOOPX_CLI_OUTPUT_BASE_REF: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.sha }} + LOOPX_CLI_OUTPUT_MAIN_REF: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.sha }} run: python examples/control_plane/cli-output-budget-regression-smoke.py - name: Run native Windows lifecycle tests diff --git a/AGENTS.md b/AGENTS.md index 5948b2603b..5680fe7700 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -336,10 +336,10 @@ local delivery (a File-only performance miss must be disclosed but does not bloc SQLite qualification), while correctness, durability, authority and recovery checks remain required for every affected backend. -### Bounded Future-Facing Refactoring +### Refactoring And Convergence During development and before approving or merging, apply the -[scope and refactoring](docs/development/README.md#engineering-scope-and-refactoring) +[convergence and refactoring](docs/development/README.md#engineering-scope-and-refactoring) to the touched behavior and adjacent owner. Include a same-domain, reversible, behavior-preserving improvement when locally reviewable and covered by parity; report applied/deferred/unnecessary with the concrete boundary. diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index 0c94752e61..b8674ec1ee 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -155,7 +155,7 @@ its own image provenance, and a later retag can change what gets launched. ## Worker profiles The worker profiles are `official`, `single`, `native-goal`, `heartbeat-resume`, -`heartbeat-explore`, and `turn-resume`. `official` uses SForge's Codex Stop +`heartbeat-explore`, `turn-resume`, and `turn-controller-worker`. `official` uses SForge's Codex Stop hook and outer recovery. `single` executes once with neither. `native-goal` uses the shared native Codex Goal transport. The heartbeat profiles use the shared LoopX scheduler and resume store. `heartbeat-explore` additionally enables the @@ -174,9 +174,15 @@ Both `heartbeat-resume` and `turn-resume` use the external quota scheduler, planned task entry and one Agent-scoped Codex session by default. Turn sends the captured work to the Host, runs the validator and settles through Core. For a comparison, freeze both source revisions, task, feedback, model, effort -and effective solve time. Heartbeat's native context delivery defaults on. -Turn uses its product Host request and reserves 160 seconds for managed cleanup. -Allow for that difference when choosing the phase budget. +and phase limit. Record solver and controller elapsed time separately. Compare +the judge's selected round using the task's declared selection policy, native +session usage including cached input, and verified quota spends. For example, +`pass_rate_first` retains the first round at the highest partial pass rate; +a later raw score can improve without changing the selected round. Heartbeat's +native context delivery defaults on. +Turn uses its product Host request. Both workers use the remaining phase time +for planning, admission and execution. Turn also runs its required validator +and settles the original transaction within the same trial budget. A lost reply leaves a private pending marker. The next Turn wake looks up the original journal by Goal, Agent and wake identity. A committed replay produces @@ -185,8 +191,49 @@ the wake for reconciliation. Older markers without a validator digest also require reconciliation against the original validator; the current command cannot establish that history. Retain the original trial state while diagnosing a failure. To return to heartbeat, select `heartbeat-resume` for a new trial. -The Controller–Worker profile and its matched comparison are specified in the -[delivery RFC](../../docs/architecture/rfcs/controller-executor-unified-contract-v0.md#from-current-code-to-a-useful-product-dependencies-and-delivery-slices). +A rejected Host result permits one automatic receipt correction on the original +Turn. A second contract rejection stops the scheduler and retains the pending +transaction for explicit repair. A lost corrective reply permits journal +readback without another automatic model attempt. +The benchmark passes the topology to the public Turn provider. Other callers use +`loopx turn run-once --host codex-cli --execution-mode isolated-headless --topology controller-worker` +with their own project, Goal, validator and phase budget. + +Use the same task, validator and model with `--worker turn-controller-worker` to +run two fresh Codex CLI sessions inside the admitted Turn. The Controller sees +the captured decision and task. On its first dispatch the Host passes the +pinned task file to the Worker verbatim. After the Worker returns a finding and +artifact paths, the Controller can request further work on the original Worker +session while the admission and phase budget remain live, or send its receipt +to Core. Both roles use `codex exec` and resume by exact session ID. +The guest needs a kernel that supports the native Codex CLI filesystem sandbox. +The sandbox enforces the two roles' file and tool grants at execution. +Both roles can use the caller's ordinary workspace tools, capped at workspace +write, and hand off that workspace serially. Only receipt repair restricts the +Controller to reading. This provider does not yet deliver feedback while the +Worker's native Turn is still running. + +From the heartbeat trial entry, specify +`--worker heartbeat-resume --turn-topology single-worker` or `--worker heartbeat-resume --turn-topology controller-worker` and +supply the same validator. Both use the existing heartbeat scheduler, quota decision +and public Turn transaction. The unqualified heartbeat profile keeps its native +context transport for baseline comparisons. Receipts distinguish the requested +entry from the effective treatment. During fresh execution, Turn's built-in +preparation owner performs the exact ordinary claim supplied by quota and rereads +the same Todo before dispatch. This needs no extra preparation switch. Existing +native heartbeat consumes the same claim contract. + +The original Turn’s private `turns/controller-worker/` record includes role evidence with native thread and turn IDs, +message digests, step wall time and reported cumulative token use per thread. +For role cost, take the last cumulative count for each thread once. A missing +count remains unknown. Include planning, validation, retries, idle time and +operator interventions in a matched comparison. Cold trials accept +`--container-cpus N` (default 4) for both work and judge containers and record it +in the runtime settings. Use the same CPU limit across compared arms. +The trial's Goal, pending identity, original decision and validator survive a +lost acknowledgement. Roles start fresh for the next admitted Turn. The +longer-lived role binding and matched comparison are specified in the +[delivery RFC](../../docs/architecture/rfcs/controller-executor-unified-contract-v0.md#controller-delivery-cadence). All profiles use the same explicitly staged Codex binary and model/effort. This permits an updated Codex version with the native scaffold; it is not a claim of @@ -444,8 +491,8 @@ qualification runs. The shared Harbor defaults are unchanged. Native task internet policy is retained. Each attempt requires a new output directory. Heartbeat and native Goal workers have no independent per-call time limit. -The shared worker uses the remaining trial budget, retaining 160 seconds for -startup/cleanup; natural completion determines continuation boundaries. A second +Heartbeat and Turn can use the entire remaining trial budget. Native Goal +retains a 160-second reserve. Natural completion determines continuation boundaries. A second scheduler wake alone does not prove resume: verify another actual model invocation with the same session identity. Native Goal owns its continuation without an outer resume loop. Heartbeat profiles now also disable SForge outer diff --git a/benchmark/edgebench/export_report.py b/benchmark/edgebench/export_report.py index 09a8ac39ad..7d14071f25 100644 --- a/benchmark/edgebench/export_report.py +++ b/benchmark/edgebench/export_report.py @@ -31,9 +31,21 @@ def _validate_feedback_payload(runner, profile): raise ValueError("Unsupported feedback payload for this mode") -def _validate_turn_validator(runner, profile): +def _validate_worker_settings(runner, profile): + if "container_cpus" in runner and (type(runner["container_cpus"]) is not int or runner["container_cpus"] < 1): + raise ValueError("Invalid runtime container CPU limit") + if "requested_worker" in runner: + from benchmark.runtime.codex import governed_worker_profile + + effective = governed_worker_profile(runner["requested_worker"], runner.get("turn_topology", "single-worker")) + if effective != runner["worker"]: + raise ValueError("Requested entry disagrees with governed Turn topology") key = "validation_argv_sha256" - if runner["worker"] == "turn-resume": + dual = runner["worker"] == "turn-controller-worker" + for source in (runner, profile): + if source.get("turn_topology") != ("controller-worker" if dual else None): + raise ValueError("Worker topology missing or mismatched") + if dual or runner["worker"] == "turn-resume": digest = runner.get(key) if not isinstance(digest, str) or not re.fullmatch("[0-9a-f]{64}", digest) or profile.get(key) != digest: raise ValueError("Turn validator argv digest missing or mismatched") @@ -141,7 +153,7 @@ def _validate_run_data(row, config, summary, points): raise ValueError(f"Settings identity mismatch: {key}") runner, profile = config["runner"], config["worker_profile"] _validate_feedback_payload(runner, profile) - _validate_turn_validator(runner, profile) + _validate_worker_settings(runner, profile) if ( runner["runner_commit"] != row["runner_revision"] or runner["model"] != row["model_id"] @@ -269,7 +281,7 @@ def export(runs_root: Path, selection: list, output: Path, *, observed_at: str): if row.get("runner_revision") != receipt["runner_commit"]: raise ValueError("Board and runner revision disagree") _validate_feedback_payload(receipt, profile) - _validate_turn_validator(receipt, profile) + _validate_worker_settings(receipt, profile) if receipt.get("planning_horizon_goal") != profile.get("planning_horizon_goal"): raise ValueError("Profile disagrees with runtime: planning_horizon_goal") for source in (receipt, profile): @@ -347,9 +359,13 @@ def export(runs_root: Path, selection: list, output: Path, *, observed_at: str): } # Old notification-only attempts retain absent/unknown payload metadata; # never relabel them using today's default or operator-supplied settings. - if "validation_argv_sha256" in receipt: - for target in ("runner", "worker_profile"): - config[target]["validation_argv_sha256"] = receipt["validation_argv_sha256"] + for key in ("validation_argv_sha256", "turn_topology"): + if key in receipt: + for target in ("runner", "worker_profile"): + config[target][key] = receipt[key] + for key in ("requested_worker", "container_cpus"): + if key in receipt: + config["runner"][key] = receipt[key] if "feedback_payload" in receipt: config["runner"]["feedback_payload"] = receipt["feedback_payload"] config["worker_profile"]["feedback_payload"] = profile["feedback_payload"] diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index e5bde74a56..26c34d568a 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -33,6 +33,7 @@ from benchmark.runtime.codex import TASK_ENTRIES, turn_envelope_receipt from loopx.reasoning_effort import REASONING_EFFORTS from loopx.control_plane.work_items.stride_policy_codec import stride_strategy_options +from benchmark.runtime.codex import governed_worker_profile from benchmark.runtime.sforge import ( DEFAULT_TIMEOUT_SECONDS, PROFILES, SForgeWorker, profile_execution, ) @@ -118,8 +119,10 @@ def main(argv=None): parser.add_argument("--log-dir", required=True, type=Path) parser.add_argument("--run-id", required=True) parser.add_argument("--worker", choices=PROFILES, required=True) + parser.add_argument("--turn-topology", choices=("single-worker", "controller-worker"), + help="Execute the quota-admitted heartbeat or Turn through the shared governed Turn host") parser.add_argument("--validation-command-json", type=json.loads, - help="For turn-resume, trusted guest argv that independently checks the work postcondition") + help="For governed Turn workers, trusted guest argv that independently checks the work postcondition") parser.add_argument("--feedback", choices=FEEDBACK_MODES, default="native", help="New-run default: native public feedback; blind/best-only are explicit controls") parser.add_argument("--managed-run-json", type=Path, @@ -149,6 +152,7 @@ def main(argv=None): help="Per-attempt cap for explicit native readiness setup, at most 25s; " "omitted keeps 25s (not a model retry)") # Preserve the existing unique abbreviation after adding --managed-run-json. + parser.add_argument("--container-cpus", type=int, help="CPU limit per cold work and judge container (default 4)") parser.add_argument("--model", "--m", required=True) parser.add_argument("--effort", choices=REASONING_EFFORTS, required=True) parser.add_argument("--provider-rate-limit-retry-seconds", type=int, default=0, @@ -207,6 +211,13 @@ def main(argv=None): help="Host services run on Docker's default bridge (e.g. runner inside a VM); " "pin host.docker.internal there instead of Docker's host-gateway alias") args = parser.parse_args(argv) + requested_worker = args.worker + try: + args.worker = governed_worker_profile(requested_worker, args.turn_topology) + except ValueError as error: + parser.error(str(error)) + if args.container_cpus is not None and (args.container_cpus < 1 or args.restore_checkpoint): + parser.error("--container-cpus requires a positive limit on a cold trial") if args.native_result_comparison and ( args.worker != "heartbeat-explore" or args.feedback != "native"): parser.error("--native-result-comparison requires native heartbeat-explore") @@ -412,10 +423,10 @@ def main(argv=None): agent_api_base_url=api_base_url, agent_api_key=api_key, agent_model=args.model, agent_effort=args.effort, agent_timeout=args.timeout, log_dir=args.log_dir, tasks_dir=args.tasks_dir, - work_cpu_limit=4, work_mem_limit="16g", + work_cpu_limit=args.container_cpus or 4, work_mem_limit="16g", # A restored judge already owns its captured resource profile. Cold # defaults must not replace or conflict with that live registration. - judge_cpu_limit=None if args.restore_checkpoint else 4, + judge_cpu_limit=None if args.restore_checkpoint else (args.container_cpus or 4), judge_mem_limit=None if args.restore_checkpoint else "8g", ) try: @@ -539,6 +550,8 @@ def main(argv=None): receipt = { "run_id": args.run_id, "task": args.task, "worker": args.worker, "task_entry": agent.task_entry, + **({"requested_worker": requested_worker} if args.turn_topology is not None else {}), + **({"container_cpus": config.work_cpu_limit} if not args.restore_checkpoint else {}), **execution.validation_receipt(), **({"planning_timeout_seconds": None} if agent.task_entry == "loopx-planned" else {}), diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index c3e607e910..fc294c11c3 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -426,14 +426,15 @@ state or force the agent active. An unresolved Turn must be recovered before another phase can replace its task input. These wait/recovery rules apply to both entry policies; they correct the earlier unconditional phase reset. Every scheduler wake caps its host timeout against the remaining phase budget -before opening an execution. Heartbeat workers may use the entire positive +before opening an execution. Heartbeat and Turn workers may use the entire positive remainder, including a continuation shorter than 160 seconds. Their bounded process cleanup is not a solver-time deduction. The adapters do not derive an independent wake limit from the initial trial budget; an explicitly supplied Harbor wake limit remains an independent limit and retains its failure semantics. -Native Goal and managed-turn transports retain their existing reserve. -When the applicable work allowance is exhausted, the worker records a -budget-exhausted no-op without creating a pending Turn. In the +Native Goal retains its existing reserve. A deadline reached before dispatch +records a budget-exhausted no-op without creating a pending Turn. A deadline +reached during a Turn interrupts the owned process and keeps its pending +identity for original-journal recovery, including a lost acknowledgement. In the heartbeat/turn scheduler path, that receipt exits the wake with code 75 and the configured shell worker stops normally. It does not re-admit empty wakes, mark the task complete or spend quota. Direct plain/native-Goal calls retain exit 0 @@ -734,7 +735,7 @@ also needs installed Codex, the native Harbor backend and independently checked task output. Unit tests establish no score or model-uplift claim. Validate small jobs through each benchmark's native configuration before launching a study. -By default, worker calls have no independent turn deadline. Harbor derives their available time from the remaining total phase budget, reserving cleanup and settlement time. An explicit `turn_timeout_sec` remains supported as an operator override. +By default, Harbor gives heartbeat and Turn workers the remaining phase time. Native Goal retains its runtime reserve. An explicit `turn_timeout_sec` can shorten the Host window. Process cleanup can finish after the model deadline. ### Native SForge task entry diff --git a/benchmark/runtime/codex.py b/benchmark/runtime/codex.py index 5610f9ec65..3e38d0a228 100644 --- a/benchmark/runtime/codex.py +++ b/benchmark/runtime/codex.py @@ -31,6 +31,7 @@ class Execution: turn_envelope: bool = False context_delivery: str | None = None native_child_limit: int | None = None + controller_worker: bool = False complete_turn: bool = False def __post_init__(self) -> None: @@ -38,8 +39,13 @@ def __post_init__(self) -> None: type(self.native_child_limit) is not int or self.native_child_limit < 1 ): raise ValueError("native_child_limit must be a positive integer or None") + if type(self.controller_worker) is not bool or ( + self.controller_worker and (self.mode != "turn" or self.native_child_limit is not None)): + raise ValueError("controller-worker requires a governed Turn without native child delegation") + if self.controller_worker and self.context not in {None, "fresh"}: + raise ValueError("controller-worker role threads are fresh per admission") if self.context is None: - object.__setattr__(self, "context", "resume" if self.mode in {"heartbeat", "turn"} else "fresh") + object.__setattr__(self, "context", "resume" if self.mode in {"heartbeat", "turn"} and not self.controller_worker else "fresh") if self.context_delivery is None: object.__setattr__(self, "context_delivery", "on" if self.mode == "heartbeat" else "off") if self.context_delivery not in ("off", "on") or (self.context_delivery != "off" and self.mode != "heartbeat"): @@ -78,7 +84,8 @@ def validation_receipt(self) -> dict[str, str]: if self.mode != "turn": return {} encoded = json.dumps(self.validation_command, separators=(",", ":")).encode() - return {"validation_argv_sha256": hashlib.sha256(encoded).hexdigest()} + return {"validation_argv_sha256": hashlib.sha256(encoded).hexdigest(), + **({"turn_topology": "controller-worker"} if self.controller_worker else {})} @property def uses_loopx(self) -> bool: @@ -92,6 +99,18 @@ def native_goal(self) -> bool: def uses_complete_heartbeat_guard(self) -> bool: return self.mode == "heartbeat" and not self.turn_envelope +def governed_worker_profile(profile, topology): + if topology is None: + return profile + if profile not in {"heartbeat-resume", "turn-resume", "turn-controller-worker"}: + raise ValueError("explicit Turn topology requires a plain governed heartbeat or Turn profile") + if topology == "controller-worker": + return "turn-controller-worker" + if topology == "single-worker" and profile != "turn-controller-worker": + return "turn-resume" + raise ValueError("unsupported or conflicting Turn topology") + + def turn_envelope_receipt(execution: Execution) -> dict: """Record the actual heartbeat source choice independently of capture.""" diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 6cc1ee6cca..0351dfe159 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -60,6 +60,7 @@ def __init__( self, *args, execution_mode="heartbeat", + controller_worker=False, iteration_context=None, codex_sandbox="danger-full-access", validation_command=None, @@ -104,6 +105,7 @@ def __init__( turn_envelope, context_delivery, native_child_limit, + controller_worker=controller_worker, complete_turn=complete_turn, ) if type(provider_request_admission) is not bool: @@ -319,6 +321,16 @@ async def _capture_task_path(self, environment: BaseEnvironment) -> None: async def install(self, environment: BaseEnvironment) -> None: await self._capture_task_path(environment) await super().install(environment) + if self.execution.controller_worker: + # Probe the staged CLI as the actual solver user before loading + # trial credentials or paying for a role that cannot run tools. + probe = await environment.exec( + command="/usr/local/bin/codex sandbox -P :workspace -- /bin/sh -c 'exit 0'", + timeout_sec=30, + ) + if probe.return_code: + raise RuntimeError("Controller–Worker requires a working native Codex workspace sandbox " + "inside the task guest; check guest kernel and namespace support") loopx_src = Path(os.environ["LOOPX_SRC_DIR"]).resolve() node_root = Path(os.environ["LOOPX_NODE_DIR"]).resolve() @@ -333,6 +345,11 @@ async def install(self, environment: BaseEnvironment) -> None: ), timeout_sec=180, ) + if self.execution.mode == "turn": + # Core protects the pending Turn marker's parent. Mode 0777 on a + # root-owned directory still forbids the guest user from chmod. + owner = shlex.quote(str(environment.default_user or "root")) + await self.exec_as_root(environment, command=f"chown {owner} {_LOOPX_RUNTIME} && chmod 0700 {_LOOPX_RUNTIME}") actual_commit = await self._stage_source(environment, loopx_src) await stage_portable_python(environment, _PYTHON) if self.execution.uses_loopx: @@ -704,6 +721,7 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]: **({"LOOPX_NATIVE_GOAL_CONTINUE": "1"} if self.native_goal_continue else {}), "LOOPX_ITERATION_CONTEXT": self.execution.context, "LOOPX_CODEX_SANDBOX": self.execution.sandbox, + "LOOPX_TURN_TOPOLOGY": "controller-worker" if self.execution.controller_worker else "single-worker", "LOOPX_VALIDATION_COMMAND_JSON": json.dumps( self.execution.validation_command ), diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 0723d449e3..2e4efdc9a8 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -14,13 +14,14 @@ # Private process transport between the benchmark wake and its shell scheduler. # The benchmark worker remains the owner of phase-budget exhaustion. -BUDGET_EXHAUSTED_EXIT_CODE = 75 +WAKE_REQUESTED_STOP_EXIT_CODE = 75 +BUDGET_EXHAUSTED_EXIT_CODE = WAKE_REQUESTED_STOP_EXIT_CODE PHASE_WORK_RESERVE_SECONDS = 160 def phase_work_reserve(mode: str) -> int: - """Heartbeat cleanup must not consume the remaining solver allowance.""" - return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS + """Governed wakes share the phase clock through dispatch and cancellation.""" + return 0 if mode in {"heartbeat", "turn"} else PHASE_WORK_RESERVE_SECONDS def phase_state_name(root: PurePath, phase_deadline: float) -> str: @@ -368,7 +369,7 @@ def worker_command(env: dict[str, str], *, python: str, source: str, "--runtime-profile", "generic_cli", "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"], "--state-file", state_file, "--wake-cmd", "exec " + shlex.join(wake), - "--wake-stop-exit-code", str(BUDGET_EXHAUSTED_EXIT_CODE), + "--wake-stop-exit-code", str(WAKE_REQUESTED_STOP_EXIT_CODE), "--wake-timeout-seconds", str(host_timeout + 150), "--quota-timeout-seconds", "30", "--error-backoff-seconds", "15", *(["--wake-host-failure"] if env["LOOPX_EXECUTION_MODE"] == "heartbeat" else []), diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index 6887532256..b769b6c7bd 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -38,6 +38,7 @@ "heartbeat-resume": "heartbeat", "heartbeat-explore": "heartbeat", "turn-resume": "turn", + "turn-controller-worker": "turn", } PROFILES = tuple(PROFILE_MODES) DEFAULT_TIMEOUT_SECONDS = 18 * 60 * 60 @@ -51,7 +52,8 @@ def profile_execution(profile, *, task_entry=None, context_delivery=None, if profile not in PROFILE_MODES: raise ValueError("Unknown benchmark worker profile") return Execution( - mode=PROFILE_MODES[profile], task_entry=task_entry, context_delivery=context_delivery, + mode=PROFILE_MODES[profile], controller_worker=profile == "turn-controller-worker", + task_entry=task_entry, context_delivery=context_delivery, turn_envelope=turn_envelope, complete_turn=complete_turn, native_child_limit=native_child_limit, validation_command=() if validation_command is None else validation_command, ) @@ -181,20 +183,20 @@ def __init__(self, config, *, profile: str, cwd: str, if type(replan_after_todos) is not int or not 1 <= replan_after_todos <= 5: raise ValueError("replan_after_todos must be an integer between 1 and 5") if not self.execution.uses_loopx: - raise ValueError("replan_after_todos requires a heartbeat profile or turn-resume") + raise ValueError("replan_after_todos requires a heartbeat profile or a Turn profile") if replan_after_turns is not None: if (type(replan_after_turns) is not int or not 1 <= replan_after_turns <= 6): raise ValueError("replan_after_turns must be an integer between 1 and 6") if not self.execution.uses_loopx: - raise ValueError("replan_after_turns requires a heartbeat profile or turn-resume") + raise ValueError("replan_after_turns requires a heartbeat profile or a Turn profile") if (replan_after_turns is None and replan_after_todos is None and self.execution.uses_loopx): replan_after_turns = DEFAULT_REPLAN_AFTER_TURNS self.replan_after_turns = replan_after_turns self.replan_after_todos = replan_after_todos if stride_strategy is not None and not self.execution.uses_loopx: - raise ValueError("stride_strategy requires a heartbeat profile or turn-resume") + raise ValueError("stride_strategy requires a heartbeat profile or a Turn profile") if stride_strategy is not None: from loopx.control_plane.work_items.stride_policy_codec import normalize_stride_strategy stride_strategy = normalize_stride_strategy(stride_strategy) @@ -202,7 +204,7 @@ def __init__(self, config, *, profile: str, cwd: str, if not isinstance(planning_horizon_goal, bool): raise ValueError("planning_horizon_goal must be a boolean") if not self.execution.uses_loopx: - raise ValueError("planning_horizon_goal requires a heartbeat profile or turn-resume") + raise ValueError("planning_horizon_goal requires a heartbeat profile or a Turn profile") self.planning_horizon_goal = ( DEFAULT_PLANNING_HORIZON_GOAL if planning_horizon_goal is None and self.execution.uses_loopx else planning_horizon_goal @@ -417,6 +419,7 @@ def _install_worker(self, backend, handle, log_dir, logger): self.runtime = BenchmarkCodex( **common, execution_mode=self.execution.mode, iteration_context=self.execution.context, + controller_worker=self.execution.controller_worker, validation_command=self.execution.validation_command, turn_timeout_sec=None, scheduler_timeout_sec=self.timeout_seconds, diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 9791a34f40..56df60814a 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -299,6 +299,8 @@ def pending_turn(path: Path, turn_id: str, execution: Execution) -> dict: atomic_write_private_json(path, pending) return pending pending = read_json(path.read_text()) + if pending.get("turn_topology") != validator.get("turn_topology"): + raise ValueError("Pending Turn requires its original worker topology") if pending.get("validation_argv_sha256") != validator["validation_argv_sha256"]: raise ValueError("Pending Turn requires its original validation command") return pending @@ -309,20 +311,26 @@ def turn_command( execution: Execution, turn_id: str, resume_turn_key: str | None = None, - *, recover_instance: bool = False, + *, recover_instance: bool = False, retry_failed: bool = True, ) -> list[str]: address = ["--turn-instance-id", turn_id] if resume_turn_key: - address = ["--resume-turn-key", resume_turn_key, "--retry-failed-turn"] + address = ["--resume-turn-key", resume_turn_key, *(["--retry-failed-turn"] if retry_failed else [])] elif recover_instance: address = ["--resume-turn-instance-id", turn_id, "--retry-failed-turn"] + host = ["--host", "codex-cli", "--session-scope", "agent", + "--codex-reasoning-effort", env["REASONING_EFFORT"], "--codex-bin", env["CODEX_BIN"], + "--codex-model", env["MODEL_NAME"], "--codex-sandbox", execution.sandbox] + if execution.controller_worker: + host.extend(["--topology", "controller-worker"]) + if env.get("LOOPX_TASK_DOC"): + host.extend(["--task-material", env["LOOPX_TASK_DOC"]]) return loopx_command(env) + [ "turn", "run-once", "--execute", "--no-global-sync", - "--host", - "codex-cli", + *host, "--project", env["LOOPX_PROJECT"], "--goal-id", @@ -334,16 +342,6 @@ def turn_command( *address, "--iteration-context", execution.context, - "--session-scope", - "agent", - "--codex-reasoning-effort", - env["REASONING_EFFORT"], - "--codex-bin", - env["CODEX_BIN"], - "--codex-model", - env["MODEL_NAME"], - "--codex-sandbox", - execution.sandbox, *(["--timeout-seconds", str(execution.timeout_seconds)] if execution.timeout_seconds is not None else []), "--validation-command-json", @@ -529,7 +527,11 @@ def run_once(env: dict[str, str]) -> dict: if env.get("LOOPX_CONTEXT_DELIVERY", "off") not in {"off", "on"}: raise ValueError("LOOPX_CONTEXT_DELIVERY supports only off or on") display = explicit_turn_display(env) + topology = env.get("LOOPX_TURN_TOPOLOGY", "single-worker") + if topology not in {"single-worker", "controller-worker"}: + raise ValueError("unknown governed Turn topology") execution = Execution( + controller_worker=topology == "controller-worker", mode=env.get("LOOPX_EXECUTION_MODE", "heartbeat"), context=env.get("LOOPX_ITERATION_CONTEXT"), sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access"), @@ -615,14 +617,11 @@ def run_once(env: dict[str, str]) -> dict: receipt["recovery_handoff_sha256"] = handoff["sha256"] if stage == "execute" and env.get("LOOPX_PHASE_DEADLINE_EPOCH"): remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() - # Heartbeat work uses the whole remaining phase. Cleanup is bounded - # by its existing host transport, not subtracted from each wake. - # Native Goal and managed-turn transports retain their own reserve. reserve = phase_work_reserve(execution.mode) if remaining <= reserve: receipt.update(ok=True, budget_exhausted=True, host_invoked=False) return receipt - if execution.mode == "heartbeat": + if reserve == 0: phase_work_deadline = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) execution = replace( execution, timeout_seconds=(remaining - reserve if execution.timeout_seconds is None @@ -713,6 +712,7 @@ def run_once(env: dict[str, str]) -> dict: process_evidence_path=wake / "native-process.json") else: pending = {} + retry_failed = True if execution.mode == "turn" and stage == "execute": from loopx.file_lock import try_exclusive_file_lock @@ -721,13 +721,22 @@ def run_once(env: dict[str, str]) -> dict: raise RuntimeError("Another benchmark wake owns the pending Turn") pending = pending_turn(pending_path, turn_id, execution) receipt["turn_instance_id"] = pending["turn_instance_id"] + if pending.get("validation_stage") == "host_result_contract": + from loopx.control_plane.runtime.document_io import atomic_write_private_json + + # Reserve a single corrective Host call on the original journal. + # A lost reply after reservation permits readback only. + retry_failed = not pending.get("correction_started", False) + if retry_failed: + pending["correction_started"] = True + atomic_write_private_json(pending_path, pending) command = ( turn_command( env, execution, pending["turn_instance_id"], pending.get("resume_turn_key"), - recover_instance=pending["turn_instance_id"] != turn_id, + recover_instance=pending["turn_instance_id"] != turn_id, retry_failed=retry_failed, ) if execution.mode == "turn" and stage == "execute" else native_command(env, execution, stage, wake, session_wake) @@ -784,6 +793,12 @@ def run_once(env: dict[str, str]) -> dict: # Product recovery owns eligibility and retry limits. # Never disguise a failed transaction as a fresh Turn. pending["resume_turn_key"] = result["resume_turn_key"] + if pending.get("correction_started"): + # Any returned failure on the reserved correction requires + # repair. Lost replies still read back the original journal. + receipt["turn_repair_required"] = True + elif result.get("validation_stage") == "host_result_contract": + pending["validation_stage"] = "host_result_contract" from loopx.control_plane.runtime.document_io import atomic_write_private_json atomic_write_private_json(pending_path, pending) @@ -837,10 +852,10 @@ def cancelled(signum, frame): print(json.dumps(receipt)) if receipt.get("native_goal_outcome") == "cancelled": return 130 - if receipt.get("budget_exhausted") and receipt["mode"] in {"heartbeat", "turn"}: - from benchmark.runtime.scheduler import BUDGET_EXHAUSTED_EXIT_CODE + if (receipt.get("budget_exhausted") or receipt.get("turn_repair_required")) and receipt["mode"] in {"heartbeat", "turn"}: + from benchmark.runtime.scheduler import WAKE_REQUESTED_STOP_EXIT_CODE - return BUDGET_EXHAUSTED_EXIT_CODE + return WAKE_REQUESTED_STOP_EXIT_CODE return 124 if receipt["timed_out"] else (0 if receipt["ok"] else 1) diff --git a/benchmark/tests/test_controller_worker.py b/benchmark/tests/test_controller_worker.py new file mode 100644 index 0000000000..399e7a9642 --- /dev/null +++ b/benchmark/tests/test_controller_worker.py @@ -0,0 +1,17 @@ +"""The benchmark selects the product Turn provider and pins its pending topology.""" +import pytest + +from benchmark.runtime.codex import Execution +from benchmark.runtime.worker import pending_turn + + +@pytest.mark.parametrize("original", [False, True]) +def test_recovery_pins_original_topology(tmp_path, original): + marker = tmp_path / "pending.json" + def execution(dual): + return Execution(mode="turn", controller_worker=dual, validation_command=["check"]) + + first = pending_turn(marker, "first", execution(original)) + with pytest.raises(ValueError, match="original worker topology"): + pending_turn(marker, "retry", execution(not original)) + assert pending_turn(marker, "retry", execution(original)) == first diff --git a/benchmark/tests/test_edgebench_registration.py b/benchmark/tests/test_edgebench_registration.py index fa40b725b8..456b543300 100644 --- a/benchmark/tests/test_edgebench_registration.py +++ b/benchmark/tests/test_edgebench_registration.py @@ -39,7 +39,7 @@ def native_cli(tmp_path, monkeypatch): @pytest.mark.parametrize("profile,complete_turn", [ ("official", False), ("single", False), ("native-goal", False), - ("heartbeat-resume", False), ("heartbeat-explore", False), ("turn-resume", False), + ("heartbeat-resume", False), ("heartbeat-explore", False), ("turn-resume", False), ("turn-controller-worker", False), ("heartbeat-resume", True), ]) @pytest.mark.parametrize("custom_registration", [False, True]) @@ -98,7 +98,7 @@ def native_provider(*, host_registration_callback=None, **kwargs): with pytest.raises(RuntimeError, match="synthetic worker boundary"): run.main([*args, "--worker", profile, *(["--complete-turn"] if complete_turn else []), - *(["--validation-command-json", '["trusted-validator"]'] if profile == "turn-resume" else [])]) + *(["--validation-command-json", '["trusted-validator"]'] if profile.startswith("turn-") else [])]) if custom_registration: assert events == ["prepare", "register", "bind"] else: @@ -108,12 +108,12 @@ def native_provider(*, host_registration_callback=None, **kwargs): assert receipt["status"] == "runner_failed" assert receipt["feedback"] == "native" assert receipt["local_docker_images"] == _IMAGES - assert ("validation_argv_sha256" in receipt) is (profile == "turn-resume") + assert ("validation_argv_sha256" in receipt) is profile.startswith("turn-") assert ("complete_turn" in receipt) is complete_turn assert "checkpoint_export_directory" not in receipt -@pytest.mark.parametrize("profile", ["official", "native-goal", "turn-resume"]) +@pytest.mark.parametrize("profile", ["official", "native-goal", "turn-resume", "turn-controller-worker"]) def test_registration_preparation_failure_prevents_native_admission(native_cli, monkeypatch, profile): run, args, trial = native_cli events = [] @@ -129,7 +129,28 @@ def prepare(provider): pytest.fail("failed preparation must prevent registration and all worker effects")) with pytest.raises(RuntimeError, match="synthetic registration unavailable"): run.main([*args, "--worker", profile, - *(["--validation-command-json", '["trusted-validator"]'] if profile == "turn-resume" else [])]) + *(["--validation-command-json", '["trusted-validator"]'] if profile.startswith("turn-") else [])]) assert events == ["prepare"] receipt = json.loads((trial / "runtime-receipt.json").read_text()) assert receipt["status"] == "runner_failed" and receipt["error_kind"] == "RuntimeError" + + +@pytest.mark.parametrize("topology,profile", [ + ("single-worker", "turn-resume"), + ("controller-worker", "turn-controller-worker"), +]) +def test_heartbeat_entry_records_effective_turn_arm(native_cli, monkeypatch, topology, profile): + run, args, trial = native_cli + monkeypatch.setattr(run, "RecordingDockerBackend", lambda **kw: SimpleNamespace( + observe_local_images=lambda **keys: {}, image_exists=lambda image: True, + registration_options=lambda provider: {})) + def stop(**kwargs): + raise RuntimeError("trial prepared") + monkeypatch.setattr(run, "run_agent", stop) + with pytest.raises(RuntimeError, match="trial prepared"): + run.main([*args, "--worker", "heartbeat-resume", "--turn-topology", topology, + "--validation-command-json", '["check"]', "--container-cpus", "2"]) + receipt = json.loads((trial / "runtime-receipt.json").read_text()) + assert (receipt["requested_worker"], receipt["worker"]) == ("heartbeat-resume", profile) + assert receipt.get("turn_topology", "single-worker") == topology + assert receipt["container_cpus"] == 2 diff --git a/benchmark/tests/test_edgebench_report.py b/benchmark/tests/test_edgebench_report.py index fb718b9580..75d2422e53 100644 --- a/benchmark/tests/test_edgebench_report.py +++ b/benchmark/tests/test_edgebench_report.py @@ -704,3 +704,27 @@ def test_turn_report_requires_the_same_validator_argv(trial, profile_digest): else: with pytest.raises(ValueError, match="validator argv"): export(runs, selection, output, observed_at="2026-01-02T00:00:00+00:00") + + +@pytest.mark.parametrize("observed", ["controller-worker", None, "single-worker"]) +def test_heartbeat_turn_export_binds_role_topology(trial, observed): + runs, selection, output, _ = trial + task = runs / "run" / "case" + for name, key in (("runtime-receipt", "worker"), ("worker-profile", "profile")): + path = task / (name + ".json") + record = json.loads(path.read_text()) + record[key] = "turn-controller-worker" + record["validation_argv_sha256"] = "c" * 64 + if name == "runtime-receipt": + record.update(turn_topology="controller-worker", requested_worker="heartbeat-resume", container_cpus=2) + elif observed is not None: + record["turn_topology"] = observed + write(path, record) + if observed != "controller-worker": + with pytest.raises(ValueError, match="topology"): + export(runs, selection, output, observed_at="2026-01-02T00:00:00+00:00") + return + export(runs, selection, output, observed_at="2026-01-02T00:00:00+00:00") + recorded = json.loads((output / "settings/run.json").read_text())["runner"] + assert (recorded["requested_worker"], recorded["worker"]) == ("heartbeat-resume", "turn-controller-worker") + assert recorded["turn_topology"] == observed and recorded["container_cpus"] == 2 diff --git a/benchmark/tests/test_sforge_turn.py b/benchmark/tests/test_sforge_turn.py index 07e5402a7b..50cede7edf 100644 --- a/benchmark/tests/test_sforge_turn.py +++ b/benchmark/tests/test_sforge_turn.py @@ -6,6 +6,7 @@ import shlex import subprocess import sys +import time import pytest @@ -14,6 +15,8 @@ @pytest.mark.parametrize("args", [ ["--worker", "turn-resume"], + ["--worker", "turn-controller-worker"], + ["--worker", "turn-controller-worker", "--validation-command-json", '["check"]', "--container-cpus", "0"], ["--worker", "turn-resume", "--validation-command-json", '"shell text"'], ["--worker", "turn-resume", "--validation-command-json", '[""]'], ["--worker", "single", "--validation-command-json", '["validator"]'], @@ -36,8 +39,11 @@ def test_unsupported_arm_rejected_before_trial_reservation(tmp_path, args, capsy assert not (tmp_path / "runs").exists() -@pytest.mark.parametrize("lose_reply", [False, True], ids=["reject-invalid-artifact", "recover-committed"]) -def test_turn_arm_settles_only_independently_validated_original_work(tmp_path, monkeypatch, lose_reply): +@pytest.mark.parametrize("scenario", ["reject-invalid-artifact", "recover-committed", + "correct-receipt", "hold-repeated-rejection"]) +def test_turn_arm_settles_only_independently_validated_original_work(tmp_path, monkeypatch, scenario): + lose_reply = scenario == "recover-committed" + correct_receipt = scenario in {"correct-receipt", "hold-repeated-rejection"} pytest.importorskip("sforge") pytest.importorskip("harbor") from sforge.harness.config import SForgeConfig @@ -53,6 +59,16 @@ def test_turn_arm_settles_only_independently_validated_original_work(tmp_path, m initialize_canonical_source_fixture(runtime, goal, state_path=project / ".codex" / "goals" / goal / "ACTIVE_GOAL_STATE.md", provider="sqlite") codex = fixture["_write_fake_codex"](tmp_path) + if correct_receipt: + # The real Codex adapter supplies its structured request to the fixture. + # Repeating the same input cannot repair this Host's receipt. + host = codex.read_text() + evidence = '"work_evidence": ["artifact:turn-e2e-marker"]' + assert evidence in host + corrected_input = repr('"correction":') + allow_correction = scenario == "correct-receipt" + response = f'{evidence} if {allow_correction} and {corrected_input} in prompt else ["/tmp/private-trial-artifact"]' + codex.write_text(host.replace(evidence, response)) skills = tmp_path / "skills" skills.mkdir() workspace_marker = project / fixture["MARKER_NAME"] @@ -90,8 +106,26 @@ async def installed(self, environment): "LOOPX_CODEX_HOME": str(tmp_path / "codex-home"), "LOOPX_SHARED_SKILLS": str(skills), "LOOPX_WAKE_LOG_DIR": str(wake_logs), "LOOPX_CODEX_SANDBOX": "workspace-write", "OPENAI_BASE_URL": "http://127.0.0.1:1/v1", "OPENAI_API_KEY": "fixture", + # The real CLI must still validate or recover with less than the former 160s reserve. + "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + 120), "CODEX_BIN": str(codex), "LOOPX_USAGE_PING": "0", } + if correct_receipt: + # Planned entry binds this native thread before the first execution wake. + from dataclasses import replace + from benchmark.runtime.codex import prepare_codex_home + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession + + prepare_codex_home(Path(env["LOOPX_CODEX_HOME"]), + execution=replace(worker.execution, sandbox=env["LOOPX_CODEX_SANDBOX"]), workspace=project, + model=env["MODEL_NAME"], effort=env["REASONING_EFFORT"], base_url=env["OPENAI_BASE_URL"], + api_key=env["OPENAI_API_KEY"], wire_api=env.get("CODEX_WIRE_API", "responses"), skills=skills) + + planning_output = tmp_path / "planning-output.jsonl" + planning_output.write_text(json.dumps({"type": "thread.started", "thread_id": "session-fixture-0001"})) + planning_env = env | {"CODEX_HOME": env["LOOPX_CODEX_HOME"]} + CodexAgentSession(planning_env, context="resume", sandbox=env["LOOPX_CODEX_SANDBOX"], + receipt={}).observe(planning_output) command = worker_command(env, python=sys.executable, source=str(src), state_file=str(tmp_path / "scheduler.json"), host_timeout=90) + ["--once"] def tick(): @@ -103,6 +137,26 @@ def tick(): assert workspace_marker.read_text() == fixture["_marker_value"](1) assert fixture["_quota_spend_count"](runtime) == int(lose_reply) pending = runtime / "benchmark-pending-turn.json" + if correct_receipt: + assert first["turn_execution"]["validation_stage"] == "host_result_contract" + if scenario == "correct-receipt": + drop.touch() # Recover a committed correction after losing the reply. + corrected = tick() + assert corrected["turn_instance_id"] == first["turn_instance_id"] + assert workspace_marker.read_text() == fixture["_marker_value"](2) + if scenario == "correct-receipt": + assert not corrected["ok"] and fixture["_quota_spend_count"](runtime) == 1 + replay = tick() + assert replay["ok"] and replay["turn_execution"]["replayed"] and not pending.exists() + assert not any(replay["turn_execution"]["effects"].values()) + else: + assert corrected["turn_repair_required"] and not corrected["ok"] + assert json.loads(pending.read_text())["correction_started"] + assert fixture["_quota_spend_count"](runtime) == 0 + readback = tick() + assert readback["turn_repair_required"] and not readback["turn_execution"]["effects"]["host_invoked"] + assert workspace_marker.read_text() == fixture["_marker_value"](2) + return if not lose_reply: assert first["turn_execution"]["validation"]["status"] == "failed" assert json.loads(pending.read_text())["resume_turn_key"] == first["turn_execution"]["resume_turn_key"] diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 2d69b82d6d..ad0eb737bc 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -462,6 +462,8 @@ def test_scheduler_cancel_reaps_detached_worker_host_and_keeps_receipt(tmp_path) ) cli = tmp_path / "quota" payload = { + "ok": True, + "interaction_contract": {"agent_channel": {"must_attempt": True}}, "should_run": True, "effective_action": "run_now", "scheduler_hint": { @@ -549,15 +551,46 @@ def test_turn_uses_public_cli_and_core_session_policy(tmp_path): assert parsed.codex_sandbox == "danger-full-access" -def test_turn_defaults_and_retired_context_rejection(): +@pytest.mark.parametrize("entry", ["plan", "run-once"]) +def test_turn_defaults_and_retired_context_rejection(tmp_path, monkeypatch, entry): + import runpy from loopx.cli import build_parser - - parser = build_parser() - argv = ["turn", "plan", "--goal-id", "fixture-goal", "--agent-id", "fixture-agent"] - parsed = parser.parse_args(argv) - assert parsed.iteration_context == "resume" and parsed.session_scope == "agent" + from loopx.cli_commands import turn + + src = Path(__file__).resolve().parents[2] + fixture = runpy.run_path(str(src / "examples/loopx-turn-codex-cli-e2e-smoke.py")) + project, runtime, _, registry = fixture["_write_fixture"](tmp_path, turn_count=1) + goal = fixture["GOAL_ID"] + fixture["initialize_canonical_source_fixture"](runtime, goal, provider="sqlite", + state_path=project / ".codex" / "goals" / goal / "ACTIVE_GOAL_STATE.md") + argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "turn", entry, + "--goal-id", goal, "--agent-id", fixture["AGENT_ID"], "--host", "codex-cli", + "--execution-mode", "isolated-headless"] + if entry == "run-once": + argv.extend(["--project", str(project), "--timeout-seconds", "20"]) + else: + argv.append("--include-transaction-detail") + + # The preview is intentionally compact. Observe the real plan consumed by + # both public entry points, including the mode passed to the Host. + make_plan = turn.build_loopx_turn_plan + policies = [] + def observe_plan(*args, **kwargs): + plan = make_plan(*args, **kwargs) + policy = plan["session"]["context_policy"] + policies.append((policy["mode"], policy["binding_scope"])) + return plan + + monkeypatch.setattr(turn, "build_loopx_turn_plan", observe_plan) + for options in ([], ["--topology", "controller-worker"], + ["--topology", "controller-worker", "--iteration-context", "fresh"]): + code, _ = fixture["_run_cli"]([*argv, *options]) + assert code == 0 + code, held = fixture["_run_cli"]([*argv, "--topology", "controller-worker", "--iteration-context", "resume"]) + assert code != 0 and "fresh role threads" in held["error"] + assert policies == [("resume", "agent"), ("fresh", "agent"), ("fresh", "agent")] with pytest.raises(SystemExit): - parser.parse_args([*argv, "--iteration-context", "resume-if-available"]) + build_parser().parse_args([*argv, "--iteration-context", "resume-if-available"]) def test_failed_turn_restarts_same_transaction_until_core_recovers(tmp_path): diff --git a/benchmark/tests/test_task_entry.py b/benchmark/tests/test_task_entry.py index aaab181ea6..a0ccee9cf2 100644 --- a/benchmark/tests/test_task_entry.py +++ b/benchmark/tests/test_task_entry.py @@ -755,7 +755,7 @@ async def unexpected_write(*args, **kwargs): ) -def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, monkeypatch): +def test_expired_scheduler_wake_does_not_open_a_turn(planning_env, monkeypatch): from benchmark.runtime import worker env = planning_env | { @@ -763,7 +763,7 @@ def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, mo "LOOPX_TASK_STAGE": "execute", "LOOPX_VALIDATION_COMMAND_JSON": '["python", "check.py"]', "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", - "LOOPX_PHASE_DEADLINE_EPOCH": "260", + "LOOPX_PHASE_DEADLINE_EPOCH": "100", } monkeypatch.setattr(worker.time, "time", lambda: 100) monkeypatch.setattr(worker, "prepare_codex_home", lambda *a, **kw: pytest.fail("late wake must not launch a host")) @@ -774,6 +774,27 @@ def test_late_scheduler_wake_does_not_open_an_unfinishable_turn(planning_env, mo assert not (Path(env["LOOPX_RUNTIME_ROOT"]) / "benchmark-pending-turn.json").exists() +def test_turn_interrupts_inflight_cli_at_phase_deadline(planning_env, tmp_path): + cli = tmp_path / "sleeping-cli" + pid_file = tmp_path / "invoked" + cli.write_text( + f"#!{sys.executable}\nimport os, pathlib, time\n" + f"pathlib.Path({str(pid_file)!r}).write_text(str(os.getpid()))\ntime.sleep(90)\n" + ) + cli.chmod(0o755) + receipt = run_once(planning_env | { + "LOOPX_EXECUTION_MODE": "turn", "LOOPX_TASK_STAGE": "execute", + "LOOPX_CLI": str(cli), "LOOPX_VALIDATION_COMMAND_JSON": '["validator"]', + "LOOPX_CODEX_TURN_TIMEOUT_SEC": "60", "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + 2), + }) + assert receipt["budget_exhausted"] and receipt["timed_out"] and not receipt["ok"] + assert "turn_execution" not in receipt + with pytest.raises(ProcessLookupError): + os.kill(int(pid_file.read_text()), 0) + pending = Path(planning_env["LOOPX_RUNTIME_ROOT"]) / "benchmark-pending-turn.json" + assert json.loads(pending.read_text())["turn_instance_id"] == receipt["turn_id"] + + @pytest.mark.parametrize("mode", ["heartbeat", "turn", "plain", "native-goal", "loopx-goal"]) def test_real_scheduler_stops_after_one_budget_exhausted_wake(planning_env, mode): from benchmark.runtime.scheduler import worker_command @@ -781,7 +802,7 @@ def test_real_scheduler_stops_after_one_budget_exhausted_wake(planning_env, mode env = planning_env | { "LOOPX_EXECUTION_MODE": mode, "LOOPX_TASK_STAGE": "execute", "LOOPX_TASK_ENTRY": "seeded-todo", - "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + (0 if mode == "heartbeat" else 100)), + "LOOPX_PHASE_DEADLINE_EPOCH": str(time.time() + (0 if mode in {"heartbeat", "turn"} else 100)), } if mode == "turn": env["LOOPX_VALIDATION_COMMAND_JSON"] = '["python", "check.py"]' @@ -981,19 +1002,14 @@ def test_shared_harbor_rejects_retired_planning_limit(tmp_path, legacy_limit): BenchmarkCodex(logs_dir=tmp_path, model_name="openai/fixture", planning_timeout_sec=None) -@pytest.mark.parametrize("mode,planning_elapsed,host_timeout", [ - ("heartbeat", 400, 500), ("heartbeat", 850, 150), ("heartbeat", 1000, None), - ("turn", 400, 440), ("turn", 840, None), ("turn", 850, None), -]) -def test_harbor_planning_uses_the_phase_deadline_without_a_stage_cap( - tmp_path, monkeypatch, mode, planning_elapsed, host_timeout, -): +@pytest.mark.parametrize("planning_elapsed", [400, 850, 1000]) +@pytest.mark.parametrize("mode", ["heartbeat", "turn"]) +def test_harbor_planning_uses_the_phase_deadline_without_a_stage_cap(tmp_path, monkeypatch, planning_elapsed, mode): pytest.importorskip("harbor") from benchmark.runtime import harbor agent = harbor.BenchmarkCodex(logs_dir=tmp_path, model_name="fixture", - execution_mode=mode, - validation_command=["trusted-validator"] if mode == "turn" else (), + execution_mode=mode, validation_command=["validator"] if mode == "turn" else None, turn_timeout_sec=500, scheduler_timeout_sec=1000) clock = [0.0] planning = [] @@ -1021,9 +1037,7 @@ async def result(**kwargs): monkeypatch.setattr(agent, "exec_as_agent", execute) monkeypatch.setattr(agent, "_populate_context", lambda *args: None) invocation = agent.run("Synthetic task", SimpleNamespace(exec=result, is_mounted=True), SimpleNamespace()) - # Heartbeat retains the whole solver window; Turn reserves 160 seconds. - # These expected budgets are independent of the helper under test. - if host_timeout is None: + if planning_elapsed == 1000: with pytest.raises(TimeoutError, match="before execution handoff"): asyncio.run(invocation) assert len(planning) == 1 and not execution @@ -1035,8 +1049,9 @@ async def result(**kwargs): assert "LOOPX_PLANNING_TIMEOUT_SEC" not in env assert "LOOPX_PHASE_DEADLINE_EPOCH=$(( $(date +%s) + 1000 ))" in command assert options["timeout_sec"] >= 1000 - assert f"--kill-after=30 {1000 - planning_elapsed}s" in execution[0][0] - assert float(execution[0][1]["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == host_timeout + remaining = 1000 - planning_elapsed + assert f"--kill-after=30 {remaining}s" in execution[0][0] + assert float(execution[0][1]["LOOPX_CODEX_TURN_TIMEOUT_SEC"]) == min(500, remaining) def test_sforge_post_plan_deadline_never_marks_execution_started(tmp_path, monkeypatch): diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 0f60f221d3..e75e01b349 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -192,7 +192,7 @@ qualified dual-session journey or benefit. | Unified work contract | [#7430](https://github.com/loopx-project/loopx/pull/7430) merged and was reverted by [#8265](https://github.com/loopx-project/loopx/pull/8265). [#8266](https://github.com/loopx-project/loopx/pull/8266) carries the C1 replacement. [#8238](https://github.com/loopx-project/loopx/pull/8238) and [#8323](https://github.com/loopx-project/loopx/pull/8323) landed individual settlement-read and narrative-read improvements. [#8304](https://github.com/loopx-project/loopx/pull/8304) landed the staged replan path. [#8499](https://github.com/loopx-project/loopx/pull/8499) delivers the single-worker EdgeBench entry and original-wake recovery independently of the C1 writer cutover. | Keep moving useful fixes into main. Qualify the replacement against original single-session quota before profile cutover. Earlier transport or writeback measurements remain scoped to their measured path. | | Common Host transport | [#7568](https://github.com/loopx-project/loopx/pull/7568) remains Draft, inspected at `e3a3c06524fae99519a72d751a16a5e8dd6f9a5d` | A candidate adapter; bounded calls do not establish bidirectional long-lived sessions/asynchronous control. OAR or an all-Host rewrite is not a prerequisite | -The benchmark integration on C1 `853eb8f56c1a` exercised planned task entry, quota admission and two independently validated Turn settlements on isolated SQLite. Native Codex used a scripted model endpoint. The Host received the v1 work contract. The raw quota compatibility view identifies its signed capture by hash. These checks cover dispatch and recovery. A matched trial with the native judge is still needed to compare outcomes and cost. +The benchmark integration on C1 `853eb8f56c1a` exercised planned task entry, quota admission and two independently validated Turn settlements on isolated SQLite. Native Codex used a scripted model endpoint. The Host received the v1 work contract. The raw quota compatibility view identifies its signed capture by hash. These checks cover dispatch and recovery. Section 11 records the subsequent native single-case comparison and the remaining multi-case qualification. The annex below retains older field audits and experiments at their original revisions. Historical Draft/test records do not replace this status, and this @@ -1067,6 +1067,146 @@ No model busy-polling loop or new recurring automation follows from this RFC. + + +### Proposed refinement: one run entry and execution core + +**Decision.** Converge both execution entrypoints and their implementation. The +target is one generic run operation, exposed by the existing Turn surface as +`loopx turn run` and by the same typed service to products. This command is +proposed, not shipped by the current implementation checkpoint. Benchmark, +Managed Host, Chat and Steward all call this operation. One admitted iteration +or policy-governed continuation until a deadline/stop are run settings, as are +single-worker and controller-worker topology. Heartbeat becomes the internal +wake/continuation mechanism; quota remains the shared policy, admission and +observation owner. Neither retains an independent normal execution pipeline. +The name Turn does not preserve the present Turn implementation: absorb useful +heartbeat behavior, redesign internal interfaces and retire superseded owners. + +This is one logical service entry, not a requirement for every caller to spawn +the CLI. App/Chat provide their existing native session binding and Host effects; +the CLI is another adapter. Benchmark supplies tasks, budgets and evaluation, +not a private orchestration mode. Existing heartbeat and run-once commands can +delegate during a named compatibility window, then be deprecated or removed +according to their real consumers. Quota inspection stays a non-executing view +of the same policy/state. Do not force a read-only quota query to start work. + +The current candidate exposes a useful but incomplete convergence. Both benchmark +entries use the same external scheduler and phase clock. Heartbeat runs the native +conversation with context delivery on and model-side typed CLI writeback; the +single-worker Turn Host delivers C1 and requests a structured result for independent +validation and Core settlement. Context delivery is off on that entry. These are +real behavior differences, beyond the command name. The current Turn design is a +candidate to refine, not a required replacement for every heartbeat behavior. + +```mermaid +flowchart LR + B[Benchmark] --> R[One run service / loopx turn run] + P[Managed Host / Chat / Steward] --> R + R --> Q[Quota policy and observation] + Q --> A[Admit and prepare original Turn] + A --> X[Host: single or controller-worker] + X --> V[Validate and commit one result] + V --> W[Continue / wait / stop] + W -->|authorized wake| Q +``` + +A scheduler observation may decide whether to wake; it is not another work +admission. The execution owner captures the authoritative decision once, claims +and rereads the same Todo/lease, and binds the selected Host and original attempt. +A monitor/no-work tick does not manufacture a model call or Spend. After a result, +continuation returns to current quota policy for a fresh admission; receipt +recovery keeps the original identity. + +**What becomes common.** One owner for each state transition, context decision, +result command, budget rule and recovery rule; these remain cohesive modules in +the typed control plane. Reuse or extend the established TypeScript owners; +Python/native adapters provide transport and effects. A new facade over two +decision implementations does not satisfy this decision. + +- **Source and context delivery — C1/C2.** Resolve the canonical source, + capture/view, Progressive/Delta choices and recovery through the same owner. + Choose and record a supported context profile explicitly; do not silently make + delivery on/off a consequence of the caller or a former command name. The first + parity comparison uses one native Worker and aligns this factor before adding + another role. Ordinary-quota and explicit-C1 evidence remain separate. +- **Execution and receipt — C4/C5.** Reuse preparation, Host launch/observation, + independent validation, typed writeback and once-only settlement. A model can + supply its judgment through supported typed CLI operations or a structured Host + result, but both are transport adapters into one typed result command and the + same original Turn/result owner. Inventory and retire duplicated normal writers; + never interpret free-form prose as authority + or perform both settlement paths. The host fills mechanical identity and + execution facts rather than asking the Worker to reproduce bookkeeping. +- **Work and topology — C3/C6.** Deliver the original objective, acceptance and + task materials, with ordinary tools and room to choose the method. A dual + Controller authors the Core-facing judgment; its Worker does not inherit a + second set of Core duties. Execution-time observation and feedback use existing + session, request revision and interruption owners, with actual artifact + adoption. A blocking serial handoff does not satisfy live feedback. +- **Budget and recovery — C4/C7.** Use the same absolute phase deadline across + preparation, all roles and work. Do not introduce an entry-specific early-stop + reserve to make receipts look complete. Cleanup can finish after cancellation; + new model work cannot. Preserve the last verifiable artifact separately from an + incomplete receipt. An unconfirmed native identity at the deadline remains + unresolved; neither a new session nor a fabricated success repairs it. + +**Host differences remain explicit.** A headless CLI can supply owned +create/resume/observe/interrupt callbacks. An already-running App or Chat session +must supply its verified native binding and supported callbacks to the same +lifecycle; wrapping its heartbeat in a second CLI process would duplicate the +worker and lose the original conversation. Missing ownership, interruption or +role isolation remains unsupported. No new authentication, MCP registration or +copied native session database follows from this proposal. The topology provider +belongs to LoopX; benchmark supplies task/evaluation settings, while Managed Host, +Chat and Steward supply their existing caller and return routes. + +**Cutover and acceptance.** Keep the present heartbeat path as a frozen reference +during migration, not a permanent alternative product runner. Qualify the shared +operation on explicit new admissions, then move benchmark and each product to it. +Internal breaking changes are allowed within a coordinated caller migration; +versioned public contracts, persisted attempts and rollback still need explicit +handling. First inventory and +align source/context, model/tools, task entry, session policy, result authoring, +validator and budget. Then compare one-Worker before/after outcomes and complete +costs; introduce dual topology as a separate treatment. One task or fewer native +tokens cannot establish non-regression or default promotion. + +The same public-entry tests must cover plan/observation without work claims, Host +dispatch or debit; ordinary claim and same-Todo readback, foreign/stale leases, +changed input, failed +validation, lost replies and exactly one settlement, Stop with no surviving +native tool or delayed write, and deadline/restart without a fresh fork. Live +qualification follows evidence into subsequent work and independently accepted +outcomes, including negative results and justified continuation. C8 adoption must +show the same authorized operation, busy/error/recovery state and result readback +in the relevant product journey. C9 retires each replaced normal producer/caller +only after its replacement passes; retain versioned historical readers. Rollback +selects the qualified release for new admissions and reconciles pending attempts +through their original Host and receipt. Completion requires caller-search and +real-entry evidence that benchmark and supported products use the one operation, +old normal execution/settlement producers are gone, and changing a common rule +requires one owner edit. Unsupported Hosts stay explicit; they are not silently +routed through a second runner. Apply the repository's +[structural convergence practice](../../development/README.md#structural-convergence--结构融合): +keep task/authority/outcome boundaries stable, incorporate experimental learning, +and remove the redundant structure after its replacement is qualified. + +Reuse existing managed-operation, Turn, session and receipt identities for run +observation, continuation and cancellation; do not add a parallel run ledger. +The first migration inventory is concrete: + +| Current boundary | Final owner/caller migration | Retirement evidence | +| --- | --- | --- | +| `benchmark/runtime/worker.py` selects heartbeat versus Turn and owns separate launch/recovery branches | Task/evaluation adapter calls the common run service | No benchmark-specific normal LoopX launch, context or receipt branch; native/plain controls retain their distinct research protocol | +| `benchmark/runtime/scheduler.py` supplies wakes and phase budget | External experiment clock supplies the budget; generic runtime owns authorized continuation | Scheduling one LoopX run does not reproduce policy or start a second loop | +| `turn_run_once.py` / `turn_driver/executor.py` and heartbeat launch/settlement callers | One admitted lifecycle and typed result owner, with transport-specific Host effects | Existing native continuation/context behavior adopted; replaced normal writers and duplicate failure classifiers removed | +| Managed Host / Chat / Steward callers | Same operation with existing binding, request revisions and original-audience return | Real packaged operation, Stop, failure/recovery and result readback without a second CLI Worker or caller-specific decision owner | + +This inventory is migration work, not a statement that the current PR has +completed those retirements. Keep its acceptance in C1–C9 rather than opening a +second architecture program. + ### From current code to a useful product: dependencies and delivery slices ```mermaid @@ -1083,7 +1223,11 @@ flowchart TB ``` **First consumer and reference.** Run the public Turn provider through EdgeBench. -Keep standard `heartbeat-resume` with its ordinary quota decision as the reference. +Keep standard `heartbeat-resume` as the reference and record its actual decision +source. The current C1 candidate defaults new benchmark heartbeats to the complete +Turn source; ordinary CLI quota/heartbeat still requires explicit C1 activation. +A three-arm comparison on that candidate measures topology and entry behavior. +It does not replace C1's separate comparison against the original quota source. At a new admission, select `--turn-topology single-worker` or `controller-worker` on the same benchmark entry. The standalone product entry is `loopx turn run-once --host codex-cli --execution-mode isolated-headless --topology controller-worker`. @@ -1134,6 +1278,27 @@ provides independent acceptance. The private Turn record binds the role steps and message digests to their native sessions. An operator can match those digests to retained native messages without duplicating transcripts in Core. +**Shared workspace checkpoint.** The isolated CLI provider now gives both roles +the caller's ordinary workspace tools, capped at workspace write. A native CLI +qualification with a controlled Responses backend has the Controller read the Worker's observation, write a checked +artifact and send feedback that changes the Worker's independently validated +output. This checks transport and adoption, not model judgment or outcome benefit. +Both roles still fail to read private Core state. A read-only grant stays +read-only; the original Controller's receipt-only retry stays read-only and does +not rerun the completed Worker. Role execution is serial, so this qualifies +handoff-time access, not concurrent writing or inspection during active work. +Stop uses the native interrupt-and-drain path; qualification checks actual shell +and child process exit and absence of their delayed write, rather than only the +CLI exit code. Failure to confirm native cleanup remains an unresolved stop. + +The session discovery and shared-environment requirements added in +[the session-discovery update](https://github.com/loopx-project/loopx/pull/8655) +remain the next C3/C6 acceptance boundary. Current role IDs and message digests +are retained evidence, not a shipped discovery/read journey. Bounded authorized +session/event reads, feedback while a Worker is executing, request revisions, +and App/Chat/Steward adoption remain unqualified. Reuse their existing owners; +do not add a benchmark-specific controller or another session database. + **Same effective budget.** Give all three arms the full standard heartbeat phase deadline, including preparation, dispatch, work, validation and settlement. An expired phase stops new work and interrupts the active Host. Preserve its @@ -1143,6 +1308,12 @@ solver time and end-to-end time, scored rounds, native score and pass rate, verifiable artifacts, calls, input/cached/output tokens per distinct thread, retries, quota spend and human intervention. Keep unavailable counts unknown. Native/plain controls retain their own protocol. +For a claimed feedback benefit, follow the existing +[decision-episode readback](long-horizon-harness-benchmark-research-program-v0.md#action-value-episode): +connect the observed state, prior choice, actual work, returned evidence and next +decision to the frozen terminal outcome. Justified continuation counts; a missing +prior forecast stays missing. Session survival, a delivered message or a synthetic +feedback test alone cannot establish a model-outcome gain. **What the first case establishes.** With a 600-second solver budget on the public permutation case, standard heartbeat and unified single worker used @@ -1177,6 +1348,26 @@ call count, full tokens, and Core settlement separately. The implementation PR records the observed outcomes for each topology and the guest qualification boundary. [Experiment readback](https://github.com/loopx-project/loopx/pull/8484) +**One-hour follow-up.** Three sequential 1,200-second native diagnostics at +`771b8ec8b0` used the same public task, model/effort, offline two-CPU work image +and phase deadline. Heartbeat, single-worker Turn and controller-worker scored +1.1265403166, 0.4637035725 and 1.6457014069; all artifacts passed independent +cost recomputation. Observed cumulative native tokens, including cached input, +were 4,569,942, 2,211,239 and 1,321,569; these are not billing-complete costs. +Core spends were 1, 0 and 0. Both Turn runs reached the deadline without a +completed receipt. Heartbeat's final resume began 28 ms before the deadline and +ended before native identity confirmation; it remains unresolved, not successful. +The dual Controller dispatched in 10.738 seconds, then the Worker used the phase +without returning an intermediate report: this is not evidence of feedback gain. +The single-worker result still falls below the live reference. Context delivery +and result-authoring differences prevent isolating topology as the cause. +This is one case per arm, with native-host isolation not fully attested and the +same pristine-judge 4/5 limitation; no result is an official promotion claim. +Two failed setup preflights additionally cost 229.99 seconds and 707,673 observed +tokens. Retain their failures separately. The next acceptance is the +[one-Worker convergence comparison](#heartbeat-turn-convergence), followed by +execution-time feedback qualification, not an automatic dual-role default. + **Product adoption.** Managed Host, Chat and Steward consume the same admitted Turn provider after qualification. Both headless roles use `codex exec`, the same CLI entry as standard heartbeat and single-worker Turn. Each role resumes its own explicit diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index f973ed84f8..7083970a74 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -158,7 +158,7 @@ flowchart TB | 统一工作契约 | [#7430](https://github.com/loopx-project/loopx/pull/7430) 已合入，随后由 [#8265](https://github.com/loopx-project/loopx/pull/8265) 回退；C1 替换由 [#8266](https://github.com/loopx-project/loopx/pull/8266) 继续。[#8238](https://github.com/loopx-project/loopx/pull/8238) 和 [#8323](https://github.com/loopx-project/loopx/pull/8323) 已分别改善结算读取和叙事读取。[#8304](https://github.com/loopx-project/loopx/pull/8304) 已合入分阶段重规划路径。 [#8499](https://github.com/loopx-project/loopx/pull/8499) 独立交付 EdgeBench 单 Worker 入口和按原唤醒恢复，C1 写入格式切换另行推进。 | 继续将可独立验收的修复合入主干。切换 profile 前，按原 quota 单会话验收替换路径。此前传输或写回测量只适用于各自测过的路径。 | | 通用 Host transport | [#7568](https://github.com/loopx-project/loopx/pull/7568) 仍为 Draft，本次核对头 `e3a3c06524fae99519a72d751a16a5e8dd6f9a5d` | 可作为适配器候选；bounded 调用不证明双向长期 session/异步控制，不把 OAR 或全 Host 重写设为先决条件 | -在 C1 `853eb8f56c1a` 上，benchmark 已走通任务规划、quota 准入，以及在隔离 SQLite 环境下经过独立验证的两次 Turn 结算。原生 Codex 连接脚本化模型端点，Host 收到了 v1 工作契约。quota 原有兼容视图通过哈希标识签名 capture。这些检查覆盖派发和恢复。比较成果与成本仍需在同等条件下运行原生评分器。 +在 C1 `853eb8f56c1a` 上，benchmark 已走通任务规划、quota 准入，以及在隔离 SQLite 环境下经过独立验证的两次 Turn 结算。原生 Codex 连接脚本化模型端点，Host 收到了 v1 工作契约。quota 原有兼容视图通过哈希标识签名 capture。这些检查覆盖派发和恢复。§11 记录了同题原生对照及仍待验证的多题效果与完整成本。 旧的逐字段审计与实验保留原版本，位于下面的技术附录；其中历史 Draft/测试记录不能 代替本表的状态，也不能用本表给旧结果重新标注版本。下一次认领须重查实时 PR 与任务。 @@ -856,6 +856,107 @@ backoff 继续归当前 scheduler owner。本文不引入模型空轮询或新 + + +### 提议细化：一个运行入口与执行内核 + +**决定。** 执行入口与内部实现一并融合。目标是一个通用运行操作，在既有 Turn 产品面 +暴露为 `loopx turn run`，产品通过同一个类型化服务调用。该命令仍是提案，不是当前 +实现 checkpoint 已交付的功能。Benchmark、Managed Host、Chat、Steward 都调用它。 +执行一次准入迭代，或按策略持续到截止/停止，是运行设置；single-worker 与 +controller-worker 也是运行设置。Heartbeat 成为内部唤醒与续跑机制，quota 继续拥有 +共同策略、准入和观察；两者均不再保留独立的正常执行链路。采用 Turn 名称不意味着 +保留现有 Turn 实现：吸收 heartbeat 的有效行为，重塑内部接口并退役被替代的 owner。 + +这里统一的是逻辑服务入口，不要求所有调用方启动 CLI。App/Chat 提供已有原生会话 +绑定与 Host effects，CLI 是另一适配层。Benchmark 只提供任务、预算、评测，不拥有 +专用编排模式。旧 heartbeat 与 run-once 命令可在明确的兼容窗口内转调，再按真实 +消费者弃用或移除。Quota 查询继续只是同一策略/状态的非执行视图，不因只读查询启动工作。 + +当前候选已有部分收敛：两种 benchmark 入口使用相同的外部调度器和阶段时钟。但 +heartbeat 在原生会话中开启 context delivery，由模型通过类型化 CLI 操作写回； +单 Worker Turn 则由 Host 交付 C1，要求结构化结果，再独立验证并交给 Core 结算， +该入口的 context delivery 为 off。差异已经影响行为，远不只是命令名称。现有 Turn +设计本身也需要改进，不能预设它应替代 heartbeat 的全部有效行为。 + +```mermaid +flowchart LR + B[Benchmark] --> R[统一运行服务 / loopx turn run] + P[Managed Host / Chat / Steward] --> R + R --> Q[Quota 策略与观察] + Q --> A[准入并准备原 Turn] + A --> X[Host：单 Worker 或双角色] + X --> V[验证并提交一份结果] + V --> W[继续 / 等待 / 停止] + W -->|授权唤醒| Q +``` + +调度观察可以决定是否唤醒，但不成为另一笔工作准入。执行 owner 捕获一次权威决策， +领取并回读同一 Todo/lease，再绑定所选 Host 与原 attempt。监控或无工作 tick 不制造 +模型调用或 Spend。结果返回后，继续工作重新按当前 quota 策略准入；回执恢复保留 +原身份。 + +**共同部分。** 状态迁移、上下文决策、结果命令、预算与恢复规则各有一个 owner， +在类型化控制面内保持职责内聚的模块。复用或扩展既有 TypeScript owner，Python/ +原生适配器提供传输与 effects。给两套决策实现包一个 facade 不满足本决定。 + +- **来源与上下文交付——C1/C2。** 同一 owner 解析 canonical source、capture/view、 + Progressive/Delta 选择与恢复。显式选择并记录支持的上下文 profile，不再因调用方 + 或旧命令名称而静默改变 on/off。首次一致性比较先使用单个原生 Worker， + 对齐这一因素后再引入第二个角色。普通 quota 与显式 C1 的证据仍分别保留。 +- **执行与回执——C4/C5。** 复用准备、Host 启动/观察、独立验证、类型化写回和一次结算。 + 模型可通过支持的类型化 CLI 操作或结构化 Host result 提交判断，但两者只是进入同一 + typed result command 和原 Turn/result owner 的传输适配。逐项盘点并退休重复的 + 正常 writer；不从自由文本推断权限，也 + 不同时执行两条结算路径。身份与执行事实等机械信息由 Host 补齐，不让 Worker 重复 + 制造记账信息。 +- **工作与拓扑——C3/C6。** 交付原始目标、验收条件和任务材料，提供普通工具并保留方法 + 选择空间。双角色由 Controller 提交面向 Core 的判断，Worker 不再继承另一套 Core + 职责。执行中的观察与反馈复用 session、request revision 和中断 owner，并验证反馈 + 实际进入产物；阻塞式串行交接不能算作实时反馈。 +- **预算与恢复——C4/C7。** 准备、所有角色和工作共用同一绝对阶段截止。不为了凑齐回执 + 给某个入口偷偷扣除提前结束窗口。取消后的清理可以越过截止，新模型工作不可以。 + 最后可验证产物与未完成回执分开保留。截止时无法确认原生身份仍需对账，不能通过 + 新开会话或伪造成功来“恢复”。 + +**Host 差异明确保留。** Headless CLI 可以提供自己拥有的 create/resume/observe/ +interrupt 回调。已运行的 App 或 Chat 会话应将经过验证的原生绑定及支持的回调交给 +同一生命周期；在其 heartbeat 外再包一次 CLI，会重复创建 Worker 并丢失原对话。 +缺少所有权、中断或角色隔离能力时明确不支持。本提案不新增认证、MCP 注册或原生 +会话库复制。拓扑 provider 属于 LoopX；benchmark 提供任务与评测设置，Managed Host、 +Chat、Steward 提供既有调用方及结果返回路径。 + +**切换与验收。** 迁移期间冻结当前 heartbeat 作为对照，不把它永久保留为另一个产品 +runner。在显式的新准入上验证统一操作，再把 benchmark 与各产品迁入。允许在同步迁移 +调用方时进行内部 breaking change；版本化公开契约、持久化 attempt 和回退仍需明确处理。 +先盘点并对齐来源/上下文、模型/工具、任务入口、会话策略、结果作者、validator 与预算， +再比较单 Worker 前后成果和完整成本；双角色作为另一个 treatment 引入。一题或更少的 +原生 token 不能证明无回退，更不能直接推广默认行为。 + +相同公开入口测试须覆盖：plan/观察不领取工作、不派发 Host、不扣额，常规领取与 +同 Todo 回读，外部/过期 lease，输入变化，验证失败，回复丢失与只结算一次，Stop 后实际原生工具退出且 +无延迟写入，以及截止/重启不另开会话。真实验证继续追踪证据如何进入后续工作和独立 +验收成果，包含负面结果与有依据的继续。C8 采用须在相关产品流程中展示相同的授权 +操作、忙碌/错误/恢复状态和结果回读。C9 在替代路径通过后逐项退休旧正常 producer/ +caller，并保留版本化历史 reader。回退只为新准入选择已验收版本；待处理 attempt +仍通过其原 Host 和回执对账。完成时需以调用方检索和真实入口证据证明：benchmark 与 +受支持产品已采用同一个操作，旧正常执行/结算 producer 已删除，修改共同规则只需修改 +一个 owner。不支持的 Host 明确保留为不支持，不静默转到第二个 runner。沿用仓库的 +[结构融合范式](../../development/README.md#structural-convergence--结构融合)：固定任务、 +权限与结果边界，吸收实验所得，在替代路径通过后移除冗余结构。 + +运行观察、续跑与取消复用已有 managed-operation、Turn、session 和 receipt 身份， +不再新增并行的 run 账本。首批迁移对象具体如下： + +| 当前边界 | 最终 owner / 调用方迁移 | 退役证据 | +| --- | --- | --- | +| `benchmark/runtime/worker.py` 选择 heartbeat/Turn，分别启动与恢复 | 任务/评测适配器调用统一运行服务 | 不再保留 benchmark 专用的正常 LoopX 启动、上下文和回执分支；native/plain 对照仍保留独立研究协议 | +| `benchmark/runtime/scheduler.py` 提供唤醒和阶段预算 | 实验外部时钟提供预算，通用 runtime 拥有授权续跑 | 调度一次 LoopX run 不复制策略、不再启动第二个循环 | +| `turn_run_once.py` / `turn_driver/executor.py` 与 heartbeat 启动/结算调用方 | 一个准入后生命周期与 typed result owner，Host effects 按传输适配 | 采用已有原生续跑/上下文行为，删除被替代的正常 writer 和重复失败分类 | +| Managed Host / Chat / Steward 调用方 | 同一操作接入已有绑定、request revision 与原受众返回 | 真实打包操作、Stop、失败恢复和结果读回，无第二个 CLI Worker 或调用方专用决策 owner | + +这些是迁移任务，不代表当前 PR 已完成相应退役。验收继续归 C1–C9，不再另开一套架构计划。 + ### 从现状到可用产品：依赖图与交付切片 ```mermaid @@ -872,7 +973,10 @@ flowchart TB ``` **首个调用方与对照。** 先用 EdgeBench 调用公开 Turn provider，保留标准 -`heartbeat-resume` 及其常规 quota 决策作对照。在同一 benchmark 入口的新准入上用 +`heartbeat-resume` 作对照，并记录它实际使用的决策来源。当前 C1 候选已将新 benchmark +heartbeat 默认切到完整 Turn 来源，普通 CLI quota/heartbeat 仍需显式启用 C1。在该候选 +上的三组比较衡量拓扑与入口行为，不能替代 C1 对原 quota 来源的独立比较。 +在同一 benchmark 入口的新准入上用 `--turn-topology single-worker` 或 `controller-worker` 选择拓扑。独立产品入口为 `loopx turn run-once --host codex-cli --execution-mode isolated-headless --topology controller-worker`。 默认使用单 Worker。Turn 提供执行与恢复能力，benchmark 适配器传入拓扑、任务验证器和 judge 设置。 @@ -910,12 +1014,33 @@ Worker 的回读为 Controller 判断是否返工提供证据，独立验收由 私有 Turn 记录把角色步骤、消息摘要绑定到原生会话。原生记录仍在时，操作者可以按摘要核对 原文，Core 无需再存一份完整对话。 +**共享工作区检查点。** 隔离 CLI provider 现将调用方的普通工作区工具授予两个角色， +最高为工作区写权限。使用受控 Responses 后端的原生 CLI 验证中，Controller 读取 +Worker 的观察、写入核验产物，再通过 +反馈改变 Worker 的输出，最终由独立 validator 验收。这验证传输与采用，未证明模型判断或 +成果收益。两个角色仍无法读取 Core 私有状态。 +只读授权保持只读；原 Controller 的纯回执修复也保持只读，不重跑已完成的 Worker。 +角色串行执行，因此本次验证覆盖交接时访问，未证明并发写入或工作进行中的检查。 +Stop 使用原生中断与进程清理路径，验证实际 shell 和子进程退出、未出现延迟写入， +而非只看 CLI 退出码。无法确认原生清理时，停止仍属于待对账状态。 + +[会话发现更新](https://github.com/loopx-project/loopx/pull/8655) 中的会话发现与共享环境 +要求仍是 C3/C6 的下一验收边界。当前保存的角色 ID 和消息摘要属于留存证据，并非已交付 +的发现与阅读流程。有界的授权会话/事件读取、Worker 执行中的反馈、request revision， +以及 App/Chat/Steward 采用仍未验收。继续复用已有 owner，不新增 benchmark 专用 +controller 或第二套会话数据库。 + **相同的有效预算。** 三组均可用到标准 heartbeat 的阶段截止，包括准备、派发、工作、 验证和结算。到期停止发起新工作并中断运行中的 Host，保留待恢复身份。固定任务入口、 模型及推理强度、沙箱、CPU、反馈与评测频率、提交冷却、评测器、验证器及外部调度器。 记录求解与全程时间、计分轮次、原生分数和通过率、可验证产物、调用次数、按独立线程 统计的输入／缓存／输出 token、重试、配额和人工介入。无法观测的指标保留为未知。 原生及 plain 对照沿用各自协议。 +声称反馈带来收益时，沿用既有 +[决策片段回读](long-horizon-harness-benchmark-research-program-v0.zh-CN.md#action-value-episode)， +把当时状态、事前选择、实际工作、返回证据与下一决策连接到冻结终点的结果。有依据的继续 +也算采用；缺失的事前预测保持缺失。会话存活、消息送达或合成反馈测试，均不能单独证明 +模型成果收益。 **首题验证范围。** 在公开排列题的 600 秒求解预算下，标准 heartbeat 和统一单 Worker 实际分别使用 602.27 与 602.24 秒，原生 token 为 1,613,614 与 1,624,174， @@ -939,6 +1064,20 @@ shell 临时文件受限、模型服务容量不足，以及 Controller 将自 judge 评分。分别记录最后验证的分数、黑盒调用次数、完整 token 和 Core 结算。实现 PR 保留各拓扑的实验读回与标准 guest 的验证边界。[实验记录](https://github.com/loopx-project/loopx/pull/8484) +**一小时补充实验。** 在 `771b8ec8b0` 上串行完成三组各 1,200 秒的原生诊断， +使用相同公开题目、模型/推理强度、离线双核工作镜像和阶段截止。Heartbeat、单 Worker +Turn、双角色得分依次为 1.1265403166、0.4637035725、1.6457014069，产物均通过独立 +成本复算。包含缓存输入的原生累计 token 依次为 4,569,942、2,211,239、1,321,569， +不等于完整账单成本；Core Spend 为 1、0、0。两组 Turn 均到期而未完成回执。 +Heartbeat 最后一次续跑在截止前 28 毫秒发起，原生身份尚未确认即到期，保留待对账， +不当作成功。双角色 Controller 用 10.738 秒完成派工，随后 Worker 用完阶段、未返回 +中间报告，因此未证明反馈收益。单 Worker 结果仍低于参考路径；上下文交付和结果撰写 +差异也使本次无法把效果归因于拓扑。每组仅一题一次，原生 Host 隔离未完整证明，且有 +原样 judge 的 4/5 限制，均不作为官方成绩或默认推广依据。另有两次失败的启动预检， +合计 229.99 秒与 707,673 个观测 token，单独保留失败与成本。下一验收是 +[单 Worker 融合对照](#heartbeat-turn-convergence)，再验证执行中的反馈，不自动推广 +双角色默认值。 + **产品接入。** Managed Host、Chat、Steward 在资格验证后复用同一 Turn provider。 两个 headless 角色均使用 `codex exec`，与标准 heartbeat 和单 Worker Turn 的 CLI 入口一致。后续调用用 `codex exec resume` 指定各自的原会话 ID。角色通过有界指令和 diff --git a/docs/development/README.md b/docs/development/README.md index 27cbbdf210..dfebbbc384 100644 --- a/docs/development/README.md +++ b/docs/development/README.md @@ -73,12 +73,15 @@ not need these documents to start LoopX. ## Change Loop / 变更闭环 -Keep one shipped behavior as the source of truth. Characterize it first, make a -small change in its owning module, then choose validation by risk. Do not create -a second product path solely for a test. +Keep the accepted outcome and its observable boundary as the source of truth. +Characterize the current behavior, deliver a complete outcome slice in its owning +boundary, then choose validation by risk. The slice can reshape internal modules; +its size follows the outcome, not a preference for small diffs. Do not create a +second product path solely for a test. -始终只保留一套真实交付行为作为 source of truth：先刻画现状，在行为所属模块中做 -小变更，再按风险选择验证层。不要为了测试而维护第二套产品路径。 +以已接受的目标和可观察边界为准：先刻画现状，在行为所属边界交付一个完整成果切片， +再按风险选择验证层。切片可以重塑内部模块，范围由成果决定，不以小 diff 为目标。 +不要为了测试而维护第二套产品路径。 The usual loop is: @@ -102,6 +105,62 @@ apply the scope-fit review below. During development and again before approval or merge, apply the bounded refactoring pass. These are engineering decisions within the accepted task, not additional user-approval gates. +### Structural Convergence / 结构融合 + +Hold the task's input and accepted outcome stable while improving the structure +between them. The boundary includes authority, side effects, failure/recovery, +latency and complete cost, not only matching input/output JSON. Corrected user +requirements can change that boundary explicitly; an implementation's current +shape is not the specification. + +固定当前任务的输入与验收结果，持续改进中间结构。边界包含权限、副作用、失败恢复、 +时延和完整成本，不只是输入输出 JSON 相同。用户修正需求时显式更新边界，不能把现有 +实现形状当成规范。 + +Start with the smallest extensible capability that serves a real caller. New +structures are useful experiments when they expose a demonstrated gap. After +learning from them, absorb their valid behavior into the owning structure and +retire redundant entrypoints, writers, state and branches. Keep failed approaches +as bounded evidence or independent regression oracles, not permanently active +fallback implementations. Convergence is complete when the next change to a +shared rule has one owner and affected callers use it; a wrapper, moved files, +fewer lines or more modules does not establish that outcome. Independently +evolving rules remain separate modules rather than one universal orchestrator. +Keep exploratory interfaces internal or explicitly experimental until repeated +real caller use justifies a stable public contract; do not promise compatibility +for every intermediate abstraction merely because it appeared in a prototype. + +从能服务真实调用方的最小可扩展能力出发。新增结构可以用于暴露和解决已证明的缺口； +获得证据后，把有效行为吸收到所属结构，退役重复入口、writer、状态和分支。失败方案 +保留为有界证据或独立回归判据，不永久保留成运行中的 fallback。下一次修改共同规则时 +只有一个 owner，相关调用方也实际采用它，才算完成融合；包一层、移文件、减行数或 +拆模块均不能单独证明。不同原因驱动的规则仍分模块演进，不汇成万能编排器。 +探索中的接口留在内部或明确标为实验，经过真实调用方反复使用后再稳定公开；不能因为 +某个中间抽象曾出现在原型里，就永久承担兼容成本。 + +For an authorized convergence change, reuse the delivery brief and existing +`repository_reuse.rule_ownership`, `observable_semantics` and validation evidence: + +- State the fixed start/end behavior and the gap that justifies restructuring. +- Map each affected entry, rule, writer and recovery path to its final owner; + name what disappears and any real compatibility consumer that remains. +- Change internal APIs decisively when needed, migrate active callers together, + and validate success, failure, cancellation and pending-state recovery through + the real entry. Public contracts and durable state need an explicit migration; + internal breaking changes do not authorize silently breaking deployed users. +- Compare outcomes and complete costs at the same boundary. Preserve negative + results; improve the design rather than weakening acceptance to fit it. +- Close the slice by removing superseded normal producers. If a staged cutover + is needed, retain one existing checkpoint with the caller, exit condition and + next integration step; temporary coexistence is not the target architecture. + +在已授权的融合任务中，复用交付简报和现有 review 证据：明确固定的起终点与缺口； +逐项映射入口、规则、writer、恢复路径及退役项；允许必要的内部 breaking change， +同步迁移调用方并实测成功、失败、取消和在途恢复；同边界比较成果与完整成本；完成后 +删除被替代的正常 producer。对外契约和持久状态另有明确迁移，不能无声破坏已部署用户。 +需要分阶段时，在既有 checkpoint 写清调用方、退出条件和下一接入步骤，不把长期并存 +作为终态。无需新增表单、状态词汇或逐 PR 的仪式性重构指标。 + ### Bounded Future-Facing Refactoring During development and again before approving or merging each PR, explicitly @@ -119,7 +178,10 @@ reason, remains locally reviewable and reversible, and is covered by characterization or parity validation; it need not be strictly required for the immediate fix. Do not use this principle to justify broad migration or unrelated cleanup. When the valuable related refactor is larger than a bounded -companion change, record a focused follow-up instead. PR review and merge notes +companion change, record a focused follow-up instead. When restructuring itself +is the accepted task, apply the structural convergence rule above and deliver a +complete owner/caller migration; this companion-refactor limit is not a reason +to stop at an adapter or repeatedly defer the requested integration. PR review and merge notes should state whether this future-facing pass was applied, deferred, or found unnecessary, with the concrete boundary considered. diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md index 9a452f94c2..6b5e8fb28f 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/docs/reference/protocols/loopx-turn-v0.md @@ -77,6 +77,60 @@ document, knowledge-maintenance, and other long-running workflows. The agent CLI remains responsible for model and tool execution; it does not become the authority for goal state or completion. +## Select the Controller and Worker on a headless Turn + +Any caller with a registered Goal, an admitted Todo, a Codex CLI with native +workspace sandbox support and an independent validator can select the product's +Controller–Worker provider. From the same configured registry and runtime: + +```sh +loopx turn plan --goal-id GOAL --agent-id AGENT --host codex-cli \ + --execution-mode isolated-headless --topology controller-worker +loopx turn run-once --execute --goal-id GOAL --agent-id AGENT \ + --project "$PWD" --host codex-cli --execution-mode isolated-headless \ + --topology controller-worker --timeout-seconds 300 --codex-sandbox workspace-write \ + --validation-command-json '["./check-work"]' +``` + +The default single-worker path keeps its existing Host and session policy. +The dual provider opens two fresh native threads per admission. The Controller +receives the complete original decision inline, delegates task work and assesses +the return. The Host verifies the journal capture against the delivered C1 work +contract and marks this read fulfilled for the Controller. Private Core files +remain outside both role sandboxes. During ordinary execution, both roles have +the caller's workspace grant capped at workspace write. They share the same +working directory, files and tools, with separate native contexts. The provider +executes the roles serially: the Controller can check or update an artifact +after the Worker returns and before handing work back. A read-only caller keeps +both roles read-only. Use `--task-material FILE` to give the Worker the original UTF-8 task unchanged +after the Controller chooses `dispatch`. Use `--controller-material FILE` for +Controller-only context. These options share a 128 KiB limit. The original +admission pins the content digest and its audience. +Core retains selection, leases, validation, Spend and the original journal. +The result identifies its topology. Resume the same Turn with its original Host +configuration. Switch topology on a new Turn. Missing native sandbox support +prevents role inference. Role IDs, causal steps and usage are in the original +Goal's private `turns/controller-worker/` record. +If Core rejects the structured receipt after task work completes, an authorized +`--retry-failed-turn` resumes the original read-only Controller to repair that +receipt. The native thread, model, provider, original work and result kind stay +bound. Core independently validates the artifact and settles once. A completed +repair replays its saved result. An interrupted repair requires reconciliation. +Previously completed role records remain readable. Repairing a pre-CLI attempt or +an attempt without its original model binding requires reconciliation. + +The reference provider starts both roles with `codex exec --json` and continues each +with `codex exec resume` and its original session ID. It uses the selected native +account and model provider. No MCP service is required for role handoff. The Host +reads the names and transport types of configured MCP sources to disable them for +the invocation. The user's configuration stays with Codex. The Host forwards the Controller's task instruction verbatim. The native sandbox enforces +its policy and the role driver omits the permissions briefing. A writable Worker +uses its workspace for shell temporary files. A caller +able to run the headless Turn may use it from a CLI, scheduler or external Host. +An interactive Codex App heartbeat or an existing Chat/Steward connection owns +its current conversation. Adopting this profile there requires an authorized +secondary execution surface and result return to the original conversation. + ## Generic Agent CLI Quick Start An agent CLI does not need native LoopX support. It needs a thin host adapter diff --git a/examples/control_plane/cli-output-probe-runner.py b/examples/control_plane/cli-output-probe-runner.py index a4a4c8d7be..d86e5206b0 100644 --- a/examples/control_plane/cli-output-probe-runner.py +++ b/examples/control_plane/cli-output-probe-runner.py @@ -154,8 +154,18 @@ def _quota_control_observation(value: Any) -> dict[str, Any] | None: def digest(child): return hashlib.sha256(json.dumps(child, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() - return {"source_schema": version, "controls_sha256": digest(controls), - "field_sha256": {key: digest(child) for key, child in controls.items()}} + witness = {"source_schema": version, "controls_sha256": digest(controls), + "field_sha256": {key: digest(child) for key, child in controls.items()}} + # Compare the intentional reward-authority clarification in its forward + # direction. Hash every other original control, including scope and rule. + if boundary["stop_condition"] == ( + "stop before unauthorized reward assignment, gate approval, write-control changes, or production access" + ): + prior = {**controls["boundary"], "stop_condition": + "stop if the next action needs reward, gate approval, write control, or production access"} + witness["prior_reward_boundary"] = {"boundary": digest(prior), + "controls_sha256": digest({**controls, "boundary": prior})} + return witness def _quota_markdown_field_observations(text: str) -> dict[str, str]: diff --git a/examples/control_plane/quota-plan-smoke.py b/examples/control_plane/quota-plan-smoke.py index 408af749cd..0ac68741be 100644 --- a/examples/control_plane/quota-plan-smoke.py +++ b/examples/control_plane/quota-plan-smoke.py @@ -863,7 +863,7 @@ def assert_attention_queue_overrides_stale_run_history() -> None: def assert_project_asset_backed_no_evidence_should_run() -> None: goal_id = "platform-migration-material-registry" expected_next = "Refresh the public-safe material registry summary." - expected_stop = "stop if the next action needs reward, gate approval, write control, or production access" + expected_stop = "stop before unauthorized reward assignment, gate approval, write-control changes, or production access" expected_user_todo = "Confirm whether owner review is fresh enough to resume delivery." expected_agent_todo = "Run read-only map and report material freshness without internal links." raw_next = "raw queue says ask for an old gate" diff --git a/loopx/capabilities/decision_context/runtime.py b/loopx/capabilities/decision_context/runtime.py index 51a84ba383..b5181a835b 100644 --- a/loopx/capabilities/decision_context/runtime.py +++ b/loopx/capabilities/decision_context/runtime.py @@ -7,8 +7,8 @@ from pathlib import Path from typing import Any -from ..context_providers import build_context_provider from ..context_providers.base import ContextProvider +from ..context_providers.factory import build_context_provider from ...control_plane.runtime.public_safety import public_safe_compact_text from .extension_provider import ( EXTENSION_CONTEXT_PROVIDER_ID, diff --git a/loopx/capabilities/pr_review_queue/README.md b/loopx/capabilities/pr_review_queue/README.md index 21173cc08d..84b40ea230 100644 --- a/loopx/capabilities/pr_review_queue/README.md +++ b/loopx/capabilities/pr_review_queue/README.md @@ -1231,6 +1231,16 @@ declared fields; `validation_matrix.items[].case_id` proves coverage of the packet's typed required cases. The checker validates this completeness only, never the truth of their contents. +For LoopX convergence work, apply the repository's +[structural convergence procedure](../../../docs/development/README.md#structural-convergence--结构融合) +through these existing rows. Verify the fixed input/outcome and authority/recovery +boundary, caller migration into one rule owner, actual retirement of superseded +normal producers, and comparable outcome/cost evidence. One facade over two +normal implementations is an incomplete migration. An authorized internal API +break with migrated callers is not itself a defect; deployed contracts and pending +state still require migration and recovery. This adds no generic review fields, +new ledger, line-count target or obligation to refactor unrelated work. + Behavioral qualification lives in `tests/capabilities/test_pr_review_behavior.py`: paired synthetic cases include valid designs as well as counterexamples. The optional live no-tools test uses the existing Doubao transport with a runtime-injected key: diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 7bbf6e0205..59e5da4fc1 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -81,6 +81,15 @@ def project_preparation_effects(value: dict[str, Any]) -> None: if preparation_state_written is not False and effects.get("state_written") is not True: effects["state_written"] = preparation_state_written try: + dual = getattr(args, "topology", "single-worker") == "controller-worker" + if getattr(args, "iteration_context", None) is None: + args.iteration_context = "fresh" if dual else "resume" + if dual and (args.host != "codex-cli" or args.execution_mode != "isolated-headless" + or args.iteration_context != "fresh" or getattr(args, "codex_operation_tools", False) + or getattr(args, "delegation_context_json", None)): + raise ValueError("controller-worker requires isolated-headless codex-cli with fresh role threads and no operation transport") + if not dual and (getattr(args, "controller_material", None) or getattr(args, "task_material", None)): + raise ValueError("role task material requires controller-worker topology") if (native_host_runner is not None or native_session_binding_resolver is not None) and ( args.host != "codex-app-server" or native_host_runner is None or native_session_binding_resolver is None ): @@ -150,6 +159,11 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: on_claim=record_claim, ) + topology_config = None + if dual and args.turn_command == "run-once": + from .turn_host_request import controller_worker_launch + + _, topology_config = controller_worker_launch(args, registry_path, runtime_root) resume_instance = getattr(args, "resume_turn_instance_id", None) if resume_instance is not None: from ..control_plane.turn_driver.journal_store import find_loopx_turn_key_by_instance @@ -243,6 +257,10 @@ def execution_preparer(decision: Mapping[str, Any]) -> str | None: turn_key=args.resume_turn_key, agent_id=args.agent_id, ) + recorded_host = payload.get("host") or {} + if (recorded_host.get("topology", "single-worker") != getattr(args, "topology", "single-worker") + or recorded_host.get("topology_config_sha256") != topology_config): + raise ValueError("Recovery requires the original Turn topology and Host configuration") goal_admission = FirstPartyHostGoalAdmission.for_plan( registry_path=registry_path, goal_id=args.goal_id, @@ -398,6 +416,10 @@ def read_decision_owner(): inspect_plan=(args.turn_command == "plan" and not args.include_transaction_detail and output_format(args) == "json"), ) + if dual: + payload["host"] = {**payload["host"], "topology": "controller-worker"} + if topology_config is not None: + payload["host"]["topology_config_sha256"] = topology_config if delegated_context and delegated_context.get("request_subject") is not None: import json # Private operation metadata stays outside model input. A resumed diff --git a/loopx/cli_commands/turn_host_request.py b/loopx/cli_commands/turn_host_request.py index 34852557a3..ee8a6357de 100644 --- a/loopx/cli_commands/turn_host_request.py +++ b/loopx/cli_commands/turn_host_request.py @@ -14,6 +14,43 @@ from ..control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission +def controller_worker_launch(args, registry: Path, runtime: Path) -> tuple[list[str], str]: + """Bind the product's isolated native provider to the caller's Turn configuration.""" + import hashlib + import math + import os + import sys + + if args.timeout_seconds is None or not math.isfinite(args.timeout_seconds) or args.timeout_seconds <= 0: + raise ValueError("controller-worker execution requires a finite positive --timeout-seconds") + project = Path(args.project).expanduser().resolve() + argv = [sys.executable, "-P", "-m", "loopx.control_plane.turn_driver.controller_worker", + "--project", str(project), "--registry", str(registry.resolve()), "--runtime", str(runtime.resolve()), + "--codex-bin", args.codex_bin, "--sandbox", args.codex_sandbox] + for option, value in (("--model", args.codex_model), ("--effort", args.codex_reasoning_effort)): + if value is not None: + argv.extend([option, value]) + task_material = getattr(args, "task_material", None) + material = task_material or getattr(args, "controller_material", None) + if material is not None: + from ..control_plane.turn_driver.controller_worker import read_material + + source = material.expanduser().resolve(strict=True) + if not source.is_file(): + raise ValueError("Role material must be a regular UTF-8 file") + data = read_material(source) + if task_material and not data.strip(): + raise ValueError("Worker task material is empty") + argv.extend(["--material", str(source), "--material-sha256", hashlib.sha256(data).hexdigest()]) + if task_material: + argv.append("--dispatch-task-material") + # The deadline may shrink on recovery. Model, permissions, workspace and + # private binding paths belong to the original admission. + home = str(Path(os.environ.get("CODEX_HOME", "~/.codex")).expanduser().resolve()) + key = hashlib.sha256(json.dumps([home, *argv]).encode()).hexdigest() + return [*argv, "--timeout", str(args.timeout_seconds)], key + + def prepare_turn_host_arguments( args: argparse.Namespace, *, native_host_runner: Callable[..., dict[str, Any]] | None, @@ -112,7 +149,7 @@ def run_built_in_host( "model": args.codex_model, "reasoning_effort": args.codex_reasoning_effort, "mcp_server": args.codex_mcp_server_json, - "timeout_seconds": None if args.timeout_seconds is None else max(1.0, args.timeout_seconds - 5.0), + "timeout_seconds": args.timeout_seconds, } options["goal_admission"] = ( goal_admission or FirstPartyHostGoalAdmission.for_plan( @@ -125,6 +162,8 @@ def run_built_in_host( run_codex_operation_host, ) + if args.timeout_seconds is not None: + options["timeout_seconds"] = max(1.0, args.timeout_seconds - 5.0) return run_codex_operation_host( request, registry_path=registry_path, diff --git a/loopx/cli_commands/turn_registration.py b/loopx/cli_commands/turn_registration.py index ddba5db3f2..5908191df1 100644 --- a/loopx/cli_commands/turn_registration.py +++ b/loopx/cli_commands/turn_registration.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse +from pathlib import Path import json from collections.abc import Callable @@ -169,6 +170,15 @@ def register_turn_commands( execution_mode_choices=["isolated-headless"], default_execution_mode="isolated-headless", ) + for entry in (plan, run_once): + entry.set_defaults(iteration_context=None) + entry.add_argument("--topology", choices=("single-worker", "controller-worker"), + default="single-worker", + help="Execute with one Host session or a Controller and Worker on an isolated codex-cli Host.") + material = run_once.add_mutually_exclusive_group() + material.add_argument("--task-material", type=Path, help="Original task for the Worker, forwarded verbatim after Controller dispatch.") + material.add_argument("--controller-material", type=Path, + help="An additional UTF-8 task file for the Controller (up to 128 KiB, pinned at admission).") run_once.add_argument("--project", required=True) run_once.add_argument( "--delegation-context-json", diff --git a/loopx/cli_commands/turn_rendering.py b/loopx/cli_commands/turn_rendering.py index c480835c0f..1ad1d258ec 100644 --- a/loopx/cli_commands/turn_rendering.py +++ b/loopx/cli_commands/turn_rendering.py @@ -159,6 +159,7 @@ def render_loopx_turn_plan_markdown( "# LoopX Turn Plan", *(_turn_plan_readback_lines(payload) if include_transaction_detail else []), f"- host: {host.get('kind')}", + *([f"- topology: {host['topology']}"] if host.get("topology") else []), f"- execution_mode: {host.get('execution_mode')}", f"- route: {route.get('kind')}", f"- would_invoke_host: {route.get('would_invoke_host')}", @@ -223,6 +224,8 @@ def render_loopx_turn_execution_markdown(payload: dict[str, object]) -> str: [ "# LoopX Turn Run Once", f"- status: {payload.get('status')}", + *(["- topology: controller-worker"] if isinstance(payload.get("host"), dict) + and payload["host"].get("topology") == "controller-worker" else []), f"- result_kind: {payload.get('result_kind')}", *( [f"- interval_reason: {admission.get('reason')}", diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index b93a5df82f..66cb450ef0 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -767,6 +767,7 @@ def scheduler(_spend_payload: dict[str, object]) -> dict[str, object]: host_runner: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None session_binding_resolver = None + adapter_schema = getattr(args, "host_adapter_request_schema", None) if args.host == "codex-app-server": def native_launch_decision(current_registry: Path) -> Mapping[str, Any]: return current_turn_host_launch_decision(args, registry_path=current_registry, @@ -786,6 +787,15 @@ def run_native_host(request: Mapping[str, Any]) -> dict[str, Any]: host_runner = run_native_host session_binding_resolver = native_session_binding_resolver + elif (payload.get("host") or {}).get("topology") == "controller-worker": + from ..control_plane.quota.turn_envelope import COMPLETE_TURN_HOST_REQUEST_SCHEMA + from .turn_host_request import controller_worker_launch + + adapter_schema = COMPLETE_TURN_HOST_REQUEST_SCHEMA + + raw_argv, binding = controller_worker_launch(args, registry_path, runtime_root) + if binding != payload["host"].get("topology_config_sha256"): + raise ValueError("Controller-worker launch differs from the captured Host configuration") elif args.host == "codex-cli": def current_launch_decision(current_registry: Path) -> Mapping[str, Any]: @@ -891,7 +901,7 @@ def on_managed_start_admitted() -> None: decision_capture_payload=decision, require_decision_capture=True, host_argv=raw_argv, - host_adapter_request_schema=getattr(args, "host_adapter_request_schema", None), + host_adapter_request_schema=adapter_schema, host_runner=host_runner, session_binding_resolver=session_binding_resolver, project=project, diff --git a/loopx/control_plane/testing/cli_output_differential.py b/loopx/control_plane/testing/cli_output_differential.py index 903cb30a18..9af417433b 100644 --- a/loopx/control_plane/testing/cli_output_differential.py +++ b/loopx/control_plane/testing/cli_output_differential.py @@ -717,12 +717,21 @@ def _matching_quota_controls(base: dict[str, Any], candidate: dict[str, Any]) -> # The frozen probe owns the source-field inventory; its independent tests # pin every field and mutation. Compare its bounded witness here without # creating another inventory/consumer of retired source fields. - return bool(isinstance(fields, dict) and len(fields) == 19 - and all(isinstance(key, str) and key for key in fields) - and all(isinstance(value, str) and len(value) == 64 for value in fields.values()) - and fields == after.get("field_sha256") - and isinstance(before.get("controls_sha256"), str) and len(before["controls_sha256"]) == 64 - and before["controls_sha256"] == after.get("controls_sha256")) + def valid(witness: dict[str, Any]) -> bool: + values = witness.get("field_sha256") + return bool(isinstance(values, dict) and len(values) == 19 + and all(isinstance(key, str) and key for key in values) + and all(isinstance(value, str) and len(value) == 64 for value in values.values()) + and isinstance(witness.get("controls_sha256"), str) and len(witness["controls_sha256"]) == 64) + if not valid(before) or not valid(after): + return False + current = after["field_sha256"] + if fields == current: + return before["controls_sha256"] == after["controls_sha256"] + prior = after.get("prior_reward_boundary") + return bool(isinstance(prior, dict) and "boundary" in current + and fields == {**current, "boundary": prior.get("boundary")} + and before["controls_sha256"] == prior.get("controls_sha256")) def _complete_quota_source_migration(base: dict[str, Any], candidate: dict[str, Any]) -> bool: diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py index 875785699b..ad5ac7f5e3 100644 --- a/loopx/control_plane/turn_driver/codex_cli.py +++ b/loopx/control_plane/turn_driver/codex_cli.py @@ -352,6 +352,37 @@ def _checked_prompt_context( return checked +def codex_receipt_instructions() -> tuple[str, ...]: + """Instructions for the role producing the Core receipt.""" + return ( + "Return only the schema-constrained result. For validated_progress, repair_required, or replan_required, fill every material field with public-safe evidence.", + "For those material results, preserve the actual delivery_batch_scale and delivery_outcome selected from the signed execution_profile's work and outcome requirements. Return work_evidence and validation_evidence as separate non-empty arrays of short public-safe strings; name the artifact or observed work in the first and the independent command or readback in the second. Use workspace-relative artifact names and describe scratch artifacts by purpose. Keep absolute paths, private URLs and raw output out of both fields. For legacy requests without execution_profile, use null when evidence is not supplied. For stop results, set both evidence fields to null.", + "For those material results, set path_delta_mode=material_replan only when this Turn changes a prior assumption, route, scope, acceptance rule, or stops prior work; then provide a complete bounded agent vision packet with goal_path_delta_v0 in agent_vision_json and leave vision_unchanged_reason empty.", + "For routine continuation, retry, successor creation, or no-change replanning, set path_delta_mode=unchanged, leave agent_vision_json empty, and provide vision_unchanged_reason.", + "When replan_action_packet requires a semantic delta, return progress_observation_json as a JSON-encoded typed_progress_observation_v0 with result_class, the observed surface_id/hypothesis_id/probe_kind or blocker/coverage fields, and opaque evidence_ids. Report only independently checked evidence; a plan summary or a vision change alone cannot substitute for the required observation. Otherwise leave this field empty.", + "For user_action_required, wait, or iteration_failed, leave material-only fields empty and explain the stop in summary. iteration_failed ends only this iteration and never requests a retry or successor.", + 'completed_phases must be exactly ["host_execute","typed_result"], and turn_key must match the request.', + ) + + +def prepare_codex_controller_input(request: Mapping[str, Any]) -> tuple[str, dict[str, Any]]: + """Deliver the verified C1 source once. Receipt guidance follows the first Worker return.""" + from ..quota.turn_envelope import captured_turn_host_input + + checked = _checked_prompt_context(request) + if checked is None: + raise ValueError("Controller dispatch requires the complete original Turn contract") + source = json.dumps(captured_turn_host_input(request), ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return ( + "You are the Controller. Use the complete work_contract and its outstanding task materials. " + "The Host verified the captured decision and delivered it inline. Give the Worker the original objective, " + "success criteria and useful task material. Leave the approach to the Worker unless the task specifies it. " + "You own the Core receipt. Set instruction to delegate or result to return your decision.\nTurn request:\n" + + source, + _result_schema(request, checked=checked), + ) + + def prepare_codex_cli_input(request: Mapping[str, Any]) -> tuple[str, dict[str, Any]]: """Prepare one launch's prompt and schema from one complete-source check. @@ -394,13 +425,7 @@ def _render_prompt(request: Mapping[str, Any], *, checked: Mapping[str, Any] | N ("Use work_contract as the complete source of truth. Its interaction_contract.agent_channel carries work_context, required_reads and primary_action; its cli_channel describes Core effects owned by the outer adapter. " if complete else ("Use turn_authority as the verified authority for this Turn. " if staged else "Use the TurnEnvelope as the source of truth. ")) + "Read work_context and remaining required_reads before primary_action; do not repeat fulfilled reads; perform work only when its contract allows it. The selected Todo summary is not its full requirements, and completing that work does not prove Goal completion.", "Do not write LoopX state, spend quota, or apply scheduler changes; the adapter owns those effects.", - "Return only the schema-constrained result. For validated_progress, repair_required, or replan_required, fill every material field with public-safe evidence.", - "For those material results, preserve the actual delivery_batch_scale and delivery_outcome selected from the signed execution_profile's work and outcome requirements. Return work_evidence and validation_evidence as separate non-empty arrays of short public-safe strings; name the artifact or observed work in the first and the independent command or readback in the second. For legacy requests without execution_profile, use null when evidence is not supplied. For stop results, set both evidence fields to null.", - "For those material results, set path_delta_mode=material_replan only when this Turn changes a prior assumption, route, scope, acceptance rule, or stops prior work; then provide a complete bounded agent vision packet with goal_path_delta_v0 in agent_vision_json and leave vision_unchanged_reason empty.", - "For routine continuation, retry, successor creation, or no-change replanning, set path_delta_mode=unchanged, leave agent_vision_json empty, and provide vision_unchanged_reason.", - "When replan_action_packet requires a semantic delta, return progress_observation_json as a JSON-encoded typed_progress_observation_v0 with result_class, the observed surface_id/hypothesis_id/probe_kind or blocker/coverage fields, and opaque evidence_ids. Report only independently checked evidence; a plan summary or a vision change alone cannot substitute for the required observation. Otherwise leave this field empty.", - "For user_action_required, wait, or iteration_failed, leave material-only fields empty and explain the stop in summary. iteration_failed ends only this iteration and never requests a retry or successor.", - 'completed_phases must be exactly ["host_execute","typed_result"], and turn_key must match the request.', + *codex_receipt_instructions(), ] if complete: from .host_candidate import TURN_EXECUTION_BOUNDARY_RULE, goal_acceptance_prompt_instruction @@ -765,35 +790,22 @@ def _codex_command( project: Path, schema_path: Path | None, output_path: Path | None, - sandbox: str, + sandbox: str | None, model: str | None, reasoning_effort: str | None, session_id: str | None, mcp_server: Mapping[str, Any] | None, + config_args: tuple[str, ...] = (), ) -> list[str]: + command = [codex_bin, "exec", "--skip-git-repo-check"] + if sandbox is not None and not session_id: + command.extend(["--sandbox", sandbox]) + command.extend(["-C", str(project)]) if session_id: - command = [ - codex_bin, - "exec", - "--skip-git-repo-check", - "-C", - str(project), - "resume", - "-c", - f'sandbox_mode="{sandbox}"', - "--json", - ] - else: - command = [ - codex_bin, - "exec", - "--skip-git-repo-check", - "--sandbox", - sandbox, - "-C", - str(project), - "--json", - ] + command.append("resume") + if sandbox is not None: + command.extend(["-c", f'sandbox_mode="{sandbox}"']) + command.extend(["--json", *config_args]) if schema_path is not None: command.extend(["--output-schema", str(schema_path)]) if output_path is not None: diff --git a/loopx/control_plane/turn_driver/controller_worker.py b/loopx/control_plane/turn_driver/controller_worker.py new file mode 100644 index 0000000000..ca2bf8613b --- /dev/null +++ b/loopx/control_plane/turn_driver/controller_worker.py @@ -0,0 +1,465 @@ +"""Codex Controller and Worker sessions serving one admitted Turn over the v1 Host ABI.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import os +import signal +import sqlite3 +import subprocess +import sys +import tempfile +import time +from collections.abc import Callable +from contextlib import closing, contextmanager +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Any + +from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission +from loopx.control_plane.quota.turn_envelope import COMPLETE_TURN_HOST_REQUEST_SCHEMA +from loopx.control_plane.runtime.document_io import atomic_write_private_json +from loopx.control_plane.turn_driver.codex_cli import ( + CodexFailureObservation, _codex_command, codex_cli_event_session_id, codex_cli_result_schema, + codex_receipt_instructions, normalize_codex_result, prepare_codex_controller_input, +) +from loopx.control_plane.turn_driver.host_process_transport import HostOutputLines, run_host_process +from loopx.control_plane.turn_driver.decision_capture import read_turn_observation_text +from loopx.control_plane.turn_driver.executor import MATERIAL_HOST_RESULT_KINDS +from loopx.control_plane.turn_driver.host_candidate import extract_turn_work_identity +from loopx.control_plane.turn_driver.journal_store import turn_journal_path +from loopx.file_lock import try_exclusive_file_lock + + +def digest(value: Any) -> str: + raw = json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + return hashlib.sha256(raw.encode()).hexdigest() + + +def object_schema(fields): + return {"type": "object", "properties": fields, "required": list(fields), "additionalProperties": False} + + +WORK_REPORT = object_schema({"finding": {"type": "string"}, + "artifacts": {"type": "array", "items": {"type": "string"}}}) + + +@dataclass(frozen=True) +class RoleHost: + project: Path + runtime: Path + registry: Path + codex_home: Path + codex_bin: str + model: str | None + effort: str | None + sandbox: str + timeout: float + material: Path | None = None + material_sha256: str | None = None + goal_state: Path | None = None + dispatch_task_material: bool = False + controller_writable: bool = False + + @property + def workspace_writable(self) -> bool: + return self.sandbox != "read-only" + + @property + def protected(self) -> tuple[Path, ...]: + return (self.runtime, self.registry, self.codex_home, self.project / ".codex", + self.project / ".loopx", self.project / ".git", self.project / "ACTIVE_GOAL_STATE.md", + *((self.goal_state,) if self.goal_state is not None else ())) + + +def read_material(path: Path) -> bytes: + # The caller selects the audience and pins these exact UTF-8 bytes at admission. + with path.open("rb") as stream: + data = stream.read(128 * 1024 + 1) + if len(data) > 128 * 1024: + raise ValueError("Role material exceeds 128 KiB") + data.decode("utf-8") + return data + + +def role_policy(role: str, host: RoleHost) -> tuple[str, dict[str, Any]]: + project = host.project + writable = host.workspace_writable and (role == "worker" or host.controller_writable) + context = { + "kind": "project_workspace", "workspace_path": str(project), "project_ref": digest(str(project))[:24], + "audience": "local_owner", "grant": "workspace_write" if writable else "workspace_read", + "filesystem_scope": "workspace_only", + } + identity = effect_runtime_result("collaboration.project.session_identity", {"context": context}) + profile = identity["permissions_profile"] + config = identity["host_config"] + filesystem = config["permissions"][profile]["filesystem"] + fs = filesystem[":workspace_roots"] + for path in host.protected: + if project.is_relative_to(path): + raise ValueError("worker workspace overlaps private Host state") + if path.is_relative_to(project): + fs[str(path.relative_to(project))] = "deny" + else: + filesystem[str(path)] = "deny" + return profile, config + + +def toml_value(value: Any) -> str: + # Codex -c takes TOML, including quoted filesystem keys, never shell text. + if isinstance(value, dict): + return "{" + ", ".join(json.dumps(key) + " = " + toml_value(item) for key, item in value.items()) + "}" + return json.dumps(value) + + +def role_cli_options(role: str, host: RoleHost, timeout: float) -> tuple[str, ...]: + _, config = role_policy(role, host) + if host.workspace_writable and (role == "worker" or host.controller_writable): + # Shell here-documents, compilers and tempfile users need scratch space. + # Use the existing write grant; the shared system temp directory stays denied. + shell = config["shell_environment_policy"] + scratch = {"TMPDIR": str(host.project), "TMP": str(host.project), "TEMP": str(host.project), + "TMPPREFIX": str(host.project / ".zsh-tmp")} + shell.setdefault("set", {}).update(scratch) + shell["include_only"] = [*shell["include_only"], *scratch] + config.update(approval_policy="never", include_permissions_instructions=False, features={ + "apps": False, "multi_agent": False, "multi_agent_v2": False, "hooks": False, "plugins": False, + "remote_plugin": False, "browser_use": False, "computer_use": False, "goals": False, "js_repl": False, + }) + # Native configuration can include project and managed MCP sources. Only + # names leave the discovery response; never persist its commands or secrets. + listing = subprocess.run([host.codex_bin, "mcp", "list", "--json"], cwd=host.project, + capture_output=True, timeout=min(15, timeout)) + if listing.returncode: + raise RuntimeError("cannot inspect configured Codex CLI tools") + servers = json.loads(listing.stdout) + if not isinstance(servers, list) or not all(isinstance(row, dict) and isinstance(row.get("name"), str) + and row["name"] for row in servers): + raise ValueError("invalid Codex CLI tool inventory") + # Codex validates each override before merging native configuration. + # Mask only enabled, existing entries. Match their discriminator with + # an empty address so there is no executable, endpoint or credential. + masks = {} + for server in servers: + if server.get("enabled") is False: + continue + kind = server["transport"]["type"] + if kind not in {"stdio", "streamable_http"}: + raise ValueError("unsupported native MCP transport") + address = "command" if kind == "stdio" else "url" + masks[server["name"]] = {"enabled": False, address: ""} + if masks: + config["mcp_servers"] = masks + overrides = [key + "=" + toml_value(value) for key, value in config.items()] + return ("--ignore-rules", *(arg for override in overrides for arg in ("-c", override))) + + +def cli_model_binding(host: RoleHost, thread: str) -> dict[str, str]: + # The exec JSONL stream identifies the thread. Read only the native model + # and provider chosen for that exact id; neither transcripts nor auth enter Core. + for db in sorted(host.codex_home.glob("state_*.sqlite"), reverse=True): + with closing(sqlite3.connect(db.as_uri() + "?mode=ro", uri=True, timeout=1)) as conn: + row = conn.execute("SELECT model, model_provider, cwd, approval_mode FROM threads WHERE id=?", + (thread,)).fetchone() + if row and row[0] and row[1] and Path(row[2]).resolve() == host.project and row[3] == "never": + return {"model": row[0], "modelProvider": row[1]} + raise RuntimeError("native role model binding unconfirmed") + + +class RoleTransport: + def __init__(self, host: RoleHost, deadline: float, admission: FirstPartyHostGoalAdmission, + evidence: dict[str, Any], save: Callable[[], None]) -> None: + self.host, self.deadline, self.admission = host, deadline, admission + self.evidence, self.save = evidence, save + + def remaining(self) -> float: + seconds = self.deadline - time.monotonic() + if seconds <= 0: + raise TimeoutError("controller-worker Turn deadline") + return seconds + + def check_running(self) -> None: + self.remaining() + with self.admission.launch_lifetime(operation="turn_role_continue"): + pass + + def send(self, role: str, prompt: str, schema: dict[str, Any]) -> dict[str, Any]: + threads = self.evidence["threads"] + previous = threads.get(role) + binding = self.evidence.get("host_models", {}).get(role) + options = role_cli_options(role, self.host, self.remaining()) + if previous: + if (self.evidence.get("transport") != "codex-exec" or not isinstance(binding, dict) + or binding != cli_model_binding(self.host, previous)): + raise RuntimeError("original native role binding unavailable") + options += ("-c", "model_provider=" + toml_value(binding["modelProvider"])) + step: dict[str, Any] = {"role": role, "command": "exec resume" if previous else "exec", "input_sha256": digest(prompt), + "thread_token_usage_total": None} + self.evidence["steps"].append(step) + self.save() + seen: str | None = None + completed = False + failures = CodexFailureObservation() + + def observe(line: str) -> None: + nonlocal seen, completed + event = json.loads(line) + candidate = codex_cli_event_session_id(event) + if candidate: + if (seen is not None or previous and candidate != previous + or any(candidate == value for key, value in threads.items() if key != role)): + raise RuntimeError("native role identity changed") + seen = threads[role] = candidate + self.save() + failures.observe_event(event) + if event.get("type") == "turn.completed": + if not seen or completed: + raise RuntimeError("ambiguous native role completion") + completed = True + usage = event.get("usage") or {} + keys = ("input_tokens", "output_tokens") + if all(type(usage.get(k)) is int and usage[k] >= 0 for k in keys): + step["thread_token_usage_total"] = { + "totalTokens": sum(usage[k] for k in keys), "inputTokens": usage["input_tokens"], + "outputTokens": usage["output_tokens"], + **{dest: usage[src] for src, dest in (("cached_input_tokens", "cachedInputTokens"), + ("reasoning_output_tokens", "reasoningOutputTokens")) if type(usage.get(src)) is int}, + } + + began = time.monotonic() + try: + with tempfile.TemporaryDirectory() as temp: + schema_path, output = Path(temp) / "schema.json", Path(temp) / "result.json" + schema_path.write_text(json.dumps(schema), encoding="utf-8") + command = _codex_command(codex_bin=self.host.codex_bin, project=self.host.project, + schema_path=schema_path, output_path=output, sandbox=None, model=binding["model"] if binding else self.host.model, reasoning_effort=self.host.effort, session_id=previous, + mcp_server=None, config_args=options) + events = HostOutputLines(observe) + diagnostics = HostOutputLines(failures.observe_stderr) + run = run_host_process(command, project=self.host.project, input_text=prompt, + timeout_seconds=self.remaining(), on_stdout=events.feed, on_stderr=diagnostics.feed, + launch_guard=lambda: self.admission.launch_lifetime(operation="turn_role_turn_start"), + check_running=self.check_running, interrupt_before_terminate=True) + events.finish() + diagnostics.finish() + if (run["outcome"] != "exited" or run["returncode"] != 0 or not run["output_complete"] + or not events.complete or not completed or seen is None): + step["native_status"] = run["outcome"] if run["outcome"] != "exited" else "failed" + step["codex_error_info"] = failures.category(complete=diagnostics.complete) + raise RuntimeError("native role Turn did not complete") + observed = cli_model_binding(self.host, seen) + if previous and binding != observed: + raise RuntimeError("native role model binding changed") + self.evidence.setdefault("host_models", {})[role] = observed + with output.open("rb") as stream: + data = stream.read(128 * 1024 + 1) + if len(data) > 128 * 1024: + raise ValueError("oversized native role result") + result = json.loads(data) + if not isinstance(result, dict): + raise ValueError("invalid native role result") + step.update(native_status="completed", output_sha256=digest(result)) + return result + finally: + step["elapsed_seconds"] = round(time.monotonic() - began, 3) + self.save() + + +def run_roles(native, request, material=None, *, dispatch_task_material=False): + prompt, result_schema = prepare_codex_controller_input(request) + decision_schema = object_schema({"instruction": {"type": ["string", "null"]}, + "result": {"anyOf": [result_schema, {"type": "null"}]}}) + if dispatch_task_material and not material: + raise ValueError("original Worker task material is missing") + original_task = material if dispatch_task_material else None + if original_task is not None: + prompt += '\nThe caller supplied the Worker task below. Use instruction="dispatch" to send it unchanged.' + if material is not None: + prompt += "\nOperator-supplied task material:\n" + material + worker_returned = False + while True: + schema = decision_schema + if original_task is not None and not worker_returned: + schema = object_schema({**decision_schema["properties"], "instruction": {"enum": ["dispatch", None], + "type": ["string", "null"]}}) + decision = native.send("controller", prompt, schema) + instruction, result = decision.get("instruction"), decision.get("result") + if result is not None: + if instruction is not None or not isinstance(result, dict): + raise ValueError("controller returned ambiguous decision") + if not worker_returned and result.get("result_kind") in MATERIAL_HOST_RESULT_KINDS: + raise ValueError("controller returned a material result without a Worker return") + return normalize_codex_result(request, result) + if not isinstance(instruction, str) or not instruction.strip() or len(instruction) > 16000: + raise ValueError("controller returned an invalid task") + if original_task is not None and not worker_returned: + if instruction != "dispatch": + raise ValueError("expected dispatch of the original Worker task") + instruction = original_task + returned = native.send("worker", instruction, WORK_REPORT) + if (not isinstance(returned.get("finding"), str) or not returned["finding"].strip() + or not isinstance(returned.get("artifacts"), list) + or not all(isinstance(path, str) for path in returned["artifacts"])): + raise ValueError("worker returned an invalid report") + prompt = "Worker return:\n" + json.dumps(returned, ensure_ascii=False) + if not worker_returned: + prompt += "\nFor your Core receipt:\n" + "\n".join(codex_receipt_instructions()) + worker_returned = True + + +@contextmanager +def interrupt_on_stop(): + def stop(*_): + raise InterruptedError("controller-worker Host stopped") + previous = signal.signal(signal.SIGTERM, stop) + try: + yield + finally: + signal.signal(signal.SIGTERM, previous) + + +def prepare_role_attempt(path, journal, request): + """Select the original result or the one Core-authorized receipt repair under the role lock.""" + request_hash = digest(request) + if not path.exists(): + if "correction" in request["result_contract"]: + raise RuntimeError("original role attempt requires reconciliation") + evidence = {"transport": "codex-exec", "turn_key": request["turn_key"], "input_sha256": request_hash, "threads": {}, "steps": []} + return evidence, evidence + + evidence = json.loads(path.read_text()) + if not isinstance(evidence.get("result"), dict): + raise RuntimeError("original role attempt requires reconciliation") + if evidence["input_sha256"] == request_hash: + return evidence, evidence + + # Only the result-contract feedback may differ from the completed attempt. + contract = dict(request["result_contract"]) + feedback = contract.pop("correction", None) + if (not isinstance(feedback, str) or not feedback + or digest({**request, "result_contract": contract}) != evidence["input_sha256"]): + raise RuntimeError("original role attempt requires reconciliation") + correction = evidence.get("correction") + if correction is not None: + if correction.get("input_sha256") == request_hash and isinstance(correction.get("result"), dict): + return evidence, correction + raise RuntimeError("original receipt correction requires reconciliation") + + audit = json.loads(journal.read_text()).get("recovery_audit") or {} + planned = audit.get("planned") or {} + if (planned.get("retry_failed") is not True or planned.get("reinvoke_host") is not True + or planned.get("resume_from") != "host_execute" or (audit.get("actual") or {}).get("status") != "started"): + raise RuntimeError("original Turn has no authorized Host retry") + if evidence.get("transport") != "codex-exec": + raise RuntimeError("original role transport requires reconciliation") + correction = {"input_sha256": request_hash} + evidence["correction"] = correction + return evidence, correction + + +def correct_original_receipt(native, request, evidence): + controller = evidence["threads"].get("controller") + if not isinstance(controller, str) or not controller or controller == evidence["threads"].get("worker"): + raise RuntimeError("original Controller identity unavailable") + feedback = request["result_contract"]["correction"] + reply = native.send("controller", feedback + " Return the corrected typed Turn result.", + codex_cli_result_schema(request)) + if reply.get("result_kind") != evidence["result"].get("result_kind"): + raise ValueError("receipt correction changed the original result kind") + return normalize_codex_result(request, reply) + + +def serve(request, host): + if request.get("schema_version") != COMPLETE_TURN_HOST_REQUEST_SCHEMA: + raise ValueError("controller-worker requires a complete C1 Turn request") + identity = extract_turn_work_identity(request) + project, runtime = host.project, host.runtime + key = request["turn_key"] + journal = turn_journal_path(runtime, goal_id=identity["goal_id"], turn_key=key) + capture = journal.parent / "decisions" / journal.name + reference = request.get("detail_ref", {}).get("captured_decision", {}) + if reference.get("path") != str(capture) or capture.resolve(strict=True) != capture: + raise ValueError("original Turn decision capture unavailable") + # The Codex input preparation validates the complete contract and its source hash. + # Compare the actual journal bytes as JSON before attesting that the inline + # work_contract is the original. Neither role needs access to private Core files. + if digest(json.loads(read_turn_observation_text(capture))) != digest(request["work_contract"]): + raise ValueError("original decision differs from the delivered Turn work contract") + from loopx.registry import find_registry_goal, load_registry, resolve_state_file + + goal = find_registry_goal(load_registry(host.registry), identity["goal_id"]) + if not goal: + raise ValueError("Controller-worker requires its original registered Goal") + state = resolve_state_file(Path(goal["repo"]), goal.get("state_file")) + host = replace(host, goal_state=state.resolve() if state else None) + material = None + if host.material is not None: + source = host.material.resolve(strict=True) + if not source.is_file() or any(source.is_relative_to(path) for path in host.protected): + raise ValueError("Role material overlaps protected state or is not a file") + data = read_material(source) + if hashlib.sha256(data).hexdigest() != host.material_sha256: + raise ValueError("Role material changed after admission") + material = data.decode("utf-8") + path = journal.parent / "controller-worker" / journal.name + timeout = host.timeout + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError("controller-worker requires a finite execution budget") + deadline = time.monotonic() + timeout + admission = FirstPartyHostGoalAdmission.for_plan(registry_path=host.registry, + goal_id=identity["goal_id"], planned_goal_ref=request.get("goal_ref")) + with try_exclusive_file_lock(path) as lock: + if lock is None: + raise RuntimeError("original role attempt is still active") + evidence, attempt = prepare_role_attempt(path, journal, request) + if isinstance(attempt.get("result"), dict): + return attempt["result"] + # Test the selected executable and workspace sandbox before paid inference. + # Turn supervision also enforces the outer deadline over this preflight. + probe = subprocess.run([host.codex_bin, "sandbox", "-P", ":workspace", "--", + sys.executable, "-I", "-c", "pass"], cwd=project, + capture_output=True, timeout=min(15, timeout)) + if probe.returncode: + raise RuntimeError("Native Codex workspace sandbox is unavailable") + def save(): + atomic_write_private_json(path, evidence) + save() + # Role execution is serial. Both roles share the caller's workspace + # grant during work. A receipt-only retry remains read-only. + host = replace(host, controller_writable=attempt is evidence) + native = RoleTransport(host, deadline, admission, evidence, save) + if attempt is evidence: + result = run_roles(native, request, material, dispatch_task_material=host.dispatch_task_material) + else: + result = correct_original_receipt(native, request, evidence) + attempt["result"] = result + save() + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + for name in ("project", "runtime", "registry"): + parser.add_argument(f"--{name}", type=Path, required=True) + parser.add_argument("--codex-bin", default="codex") + parser.add_argument("--model") + parser.add_argument("--effort") + parser.add_argument("--sandbox", choices=("read-only", "workspace-write", "danger-full-access"), required=True) + parser.add_argument("--timeout", type=float, required=True) + parser.add_argument("--material", type=Path) + parser.add_argument("--material-sha256") + parser.add_argument("--dispatch-task-material", action="store_true") + args = parser.parse_args() + home = Path(os.environ.get("CODEX_HOME", "~/.codex")).expanduser().resolve() + host = RoleHost(args.project.resolve(), args.runtime.resolve(), args.registry.resolve(), home, + args.codex_bin, args.model, args.effort, args.sandbox, args.timeout, args.material, args.material_sha256, + dispatch_task_material=args.dispatch_task_material) + with interrupt_on_stop(): + print(json.dumps(serve(json.load(sys.stdin), host), ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 3ca076c661..bf63fa6cf1 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -1485,7 +1485,10 @@ def _prepare_host_invocation( raise ValueError("run-once accepts either host_argv or host_runner, not both") if host_runner is None: argv = normalize_host_argv(host_argv or []) - return argv, {"executable": Path(argv[0]).name, "argv_count": len(argv)} + projection: dict[str, str | int] = {"executable": Path(argv[0]).name, "argv_count": len(argv)} + if (plan.get("host") or {}).get("topology") == "controller-worker": + projection.update(kind="codex-cli", topology="controller-worker") + return argv, projection planned_host = plan.get("host") if isinstance(plan.get("host"), dict) else {} return None, {"executable": "built-in", "kind": str(planned_host.get("kind") or "codex-cli")} @@ -1510,6 +1513,40 @@ def _host_attempt_requirements( return needs_host, retry_validation, None +def _prepare_failed_turn_retry( + journal: dict[str, Any], request: dict[str, Any], *, validation_reinvokes_host: bool, +) -> dict[str, Any]: + """Reset the failed phase and return the original request with any applicable feedback.""" + receipt = journal.get("receipt") + if isinstance(receipt, dict) and receipt.get("failed_phase") == "validation": + if validation_reinvokes_host: + if journal.get("validation_stage") == "host_result_contract": + # Feedback is about the previous response, outside the immutable + # work contract. Never send rejected private values back on this wire. + request = {**request, "result_contract": { + **request["result_contract"], + "correction": "The previous result failed the Host result contract. " + "Read back existing work and correct its structured receipt. " + "Use public-safe artifact names and observed checks in evidence. " + "Keep private locations and raw output in the workspace.", + }} + journal.pop("host_result", None) + journal.pop("result_kind", None) + journal["completed_phases"] = ( + list(TRANSACTION_PHASES[:2]) + if isinstance(journal.get("host_result"), dict) + else [] + ) + journal.pop("task_validation", None) + journal.pop("validation_stage", None) + journal.pop("reason", None) + journal.pop("receipt", None) + journal.pop("host_recovery", None) + journal.pop("host_failure", None) + journal["status"] = "in_progress" + return request + + @single_executor_per_turn_lane(execution_payload) def run_loopx_turn_once( plan: Mapping[str, Any], @@ -1634,6 +1671,9 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None: ): pass if recorded_plan is not None: + requested_host, original_host = plan.get("host") or {}, recorded_plan.get("host") or {} + if any(requested_host.get(k) != original_host.get(k) for k in ("topology", "topology_config_sha256")): + raise ValueError("Recovery requires the original Turn topology and Host configuration") # Recovery uses the recorded source through Host dispatch, # cached-result validation and settlement. plan, request = recorded_plan, build_loopx_turn_host_request(recorded_plan) @@ -1736,27 +1776,7 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None: persist_journal(journal) if journal and journal.get("status") == "failed": - receipt = ( - journal.get("receipt") - if isinstance(journal.get("receipt"), dict) - else {} - ) - if receipt.get("failed_phase") == "validation": - if validation_reinvokes_host: - journal.pop("host_result", None) - journal.pop("result_kind", None) - journal["completed_phases"] = ( - list(TRANSACTION_PHASES[:2]) - if isinstance(journal.get("host_result"), dict) - else [] - ) - journal.pop("task_validation", None) - journal.pop("validation_stage", None) - journal.pop("reason", None) - journal.pop("receipt", None) - journal.pop("host_recovery", None) - journal.pop("host_failure", None) - journal["status"] = "in_progress" + request = _prepare_failed_turn_retry(journal, request, validation_reinvokes_host=validation_reinvokes_host) persist_journal(journal) reserved_start = admission is not None and admission.get("reserved") is True if journal is None or reserved_start: diff --git a/loopx/control_plane/work_items/project_asset.py b/loopx/control_plane/work_items/project_asset.py index c518bcb1ba..435de84356 100644 --- a/loopx/control_plane/work_items/project_asset.py +++ b/loopx/control_plane/work_items/project_asset.py @@ -86,7 +86,7 @@ def project_asset_stop_condition( return monitor_display_stop_condition if agent_command: return "stop if the command fails or needs write, production, or additional approval" - return "stop if the next action needs reward, gate approval, write control, or production access" + return "stop before unauthorized reward assignment, gate approval, write-control changes, or production access" def project_asset_support_mode( diff --git a/tests/control_plane/test_cli_output_differential.py b/tests/control_plane/test_cli_output_differential.py index a0c3255e99..56642a8597 100644 --- a/tests/control_plane/test_cli_output_differential.py +++ b/tests/control_plane/test_cli_output_differential.py @@ -2306,16 +2306,17 @@ def test_equal_but_malformed_quota_witnesses_do_not_qualify(damage): def test_thin_source_read_guidance_requires_complete_ordered_recovery_contract(): - from pathlib import Path from loopx.control_plane.testing.cli_output_semantics import continuation_instruction_observations as observe from loopx.control_plane.testing.cli_output_differential import _compare_row - from loopx.heartbeat_prompt import build_heartbeat_prompt # Independent before observation: the former instruction required current # context before work but had no explicit original-section recovery block. before = {"source_read_before": True, "source_read_after": False} - body = build_heartbeat_prompt(goal_id="fixture", active_state=Path("goal.md"), - agent_id="worker-a", thin=True)["task_body"] + # This growth allowance belongs only to the historical wording. The + # current public prompt has its own recovery checks in the transport suite. + body = """Read work_context and required_reads before acting (applicable goals/criteria/constraints/stops). +For selection/replan or missing details, follow returned read/recovery instructions with original capture/identity; never re-admit. +Honor explicit full reads; keep capture/handoff flags on selection/same-Turn repair.""" after = observe(body) assert after["source_read_after"] and not after["source_read_before"] assert observe(json.dumps({"task_body": body}))["source_read_after"] diff --git a/tests/control_plane/test_cli_output_probe_runner.py b/tests/control_plane/test_cli_output_probe_runner.py index 14e97252dd..aaf1855372 100644 --- a/tests/control_plane/test_cli_output_probe_runner.py +++ b/tests/control_plane/test_cli_output_probe_runner.py @@ -83,6 +83,26 @@ def test_quota_observation_preserves_current_controls_across_declared_carriers() assert observe({**complete, "schema_version": "unknown"}) is None +def test_reward_boundary_refinement_is_directional_and_preserves_all_other_controls(): + from loopx.control_plane.testing.cli_output_differential import _matching_quota_controls + observe = runpy.run_path(str(RUNNER))["_quota_control_observation"] + def row(payload): + return {"quota_controls": observe(payload)} + before = _quota_control_fixture() + before["goal_boundary"]["stop_condition"] = ( + "stop if the next action needs reward, gate approval, write control, or production access") + after = copy.deepcopy(before) + after["goal_boundary"]["stop_condition"] = ( + "stop before unauthorized reward assignment, gate approval, write-control changes, or production access") + assert _matching_quota_controls(row(before), row(after)) + assert not _matching_quota_controls(row(after), row(before)) + for path, value in ((("goal_boundary", "stop_condition"), "stop only on reward"), + (("goal_boundary", "write_scope"), ["**"]), (("selected_todo", "claimed_by"), "other")): + changed = copy.deepcopy(after) + changed[path[0]][path[1]] = value + assert not _matching_quota_controls(row(before), row(changed)) + + @pytest.mark.parametrize("path,replacement", [ (("goal_id",), "wrong-goal"), (("agent_identity", "agent_id"), "other-worker"), (("decision",), "skip"), (("should_run",), False), (("effective_action",), "quota_skip"), diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index acf28b7ca3..5b24709a1f 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -575,7 +575,9 @@ def observe_guard(command, **kwargs): prepared = "--prepared-capture-reference" in command if prepared: assert ("--complete-turn" in command) is (replace_stage in {"version", "downgrade"}) - result = run(command, **kwargs) + # Reach the injected capture replacement even under suite contention. + # This test exercises identity refusal, not the hook's wall-clock budget. + result = run(command, **{**kwargs, "timeout": 30}) if prepared: verified["writer"] = json.loads(result.stdout) if replace_stage == "guard": @@ -730,9 +732,9 @@ def interrupt_first_guard(command, **kwargs): guards.append(command[command.index("--turn-instance-id") + 1]) if not after_ready and len(guards) == 1: raise subprocess.TimeoutExpired(command, kwargs["timeout"]) - # Reach the deliberate post-ready fault rather than accidentally - # injecting the separate pre-commit timeout under test contention. - result = run(command, **{**kwargs, "timeout": 30} if after_ready else kwargs) + # Complete prerequisites before the explicit guard/ready fault above; + # accidental claim timeouts would test a different recovery boundary. + result = run(command, **{**kwargs, "timeout": 30}) if "claim" in command: claims.append(json.loads(result.stdout)) return result @@ -1141,3 +1143,51 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( timeout=15, ) assert repeated.returncode == 0 and json.loads(repeated.stdout) == {} + + +@pytest.mark.parametrize("lease,ready", [ + (None, False), ("own", True), ("foreign", False), +]) +def test_benchmark_turn_and_heartbeat_share_selected_work_and_lease(tmp_path, monkeypatch, lease, ready): + """The two real entry decisions must agree before either Host can act.""" + leases = [] if lease is None else [_lease(owner=AGENT if lease == "own" else "agent-b")] + registry, runtime = tmp_path / "registry.json", tmp_path / "runtime" + instance = "parity-wake" + capabilities = ["--available-capability", "shell", "--available-capability", "filesystem_write"] + heartbeat = _run(tmp_path, monkeypatch, "sqlite", leases, "--runtime-profile", "generic_cli", + "--turn-instance-id", instance, *capabilities) + _, turn = run_json_cli_result( + "turn", "plan", "--goal-id", GOAL, "--agent-id", AGENT, "--host", "codex-cli", + "--execution-mode", "isolated-headless", "--scheduler-owner", "outer_controller", + "--iteration-context", "resume", "--session-scope", "agent", "--turn-instance-id", instance, + "--include-transaction-detail", *capabilities, registry_path=registry, runtime_root=runtime, + ) + captured = turn["work_contract"] + for field in ("should_run", "selected_todo", "execution_preparation", "normal_delivery_allowed"): + assert captured[field] == heartbeat[field], field + assert captured["selected_todo"]["todo_id"] == TODO + assert captured["execution_preparation"]["delivery_allowed"] is ready + assert turn["route"]["would_invoke_host"] is ready + before = heartbeat["interaction_contract"]["agent_channel"] + after = captured["interaction_contract"]["agent_channel"] + assert before["primary_action"] == after["primary_action"] + if not ready: + assert before["required_reads"] == after["required_reads"] + assert before["work_context"] == after["work_context"] + return + before_sources = {item["source"]: item for item in before["work_context"]["sources"]} + after_sources = {item["source"]: item for item in after["work_context"]["sources"]} + for source in before_sources: + assert before_sources[source] == after_sources[source] + outstanding = list(before["required_reads"]) + for item in after["required_reads"]: + outstanding.remove(item) + # Native heartbeat reads the canonical text on demand. C1 may discharge + # that read only by delivering the actual source before primary work. + read, = outstanding + assert read["source"] == "goal_state" and read["ordering"] == "before_work" + delivered = after_sources["goal_state"] + assert delivered["command"] == read["command"] and delivered["ordering"] == read["ordering"] + content = delivered["content"] + assert content["full_source"] is True and Path(content["state_file"]) == tmp_path / "state.md" + assert content["text"] == (tmp_path / "state.md").read_text() diff --git a/tests/control_plane/test_turn_controller_worker_host.py b/tests/control_plane/test_turn_controller_worker_host.py new file mode 100644 index 0000000000..12c0484b9d --- /dev/null +++ b/tests/control_plane/test_turn_controller_worker_host.py @@ -0,0 +1,374 @@ +"""One Core admission, two native identities and independently checked work.""" +import json +import os +from contextlib import nullcontext +from pathlib import Path +import runpy +import shutil +import subprocess +import sys +import time + +import pytest + +from loopx.control_plane.turn_driver import controller_worker as roles +from loopx.file_lock import try_exclusive_file_lock +from tests.control_plane.host_process_fixture import write_python_host_fixture + + +def test_revoked_role_drains_silent_exec_process(tmp_path, monkeypatch): + project = tmp_path / "project" + project.mkdir() + binary = write_python_host_fixture(tmp_path / "codex", f"""#!{sys.executable} +from pathlib import Path +import signal +import sys +import time +from loopx.file_lock import try_exclusive_file_lock +signal.signal(signal.SIGINT, lambda *_: sys.exit(130)) +with try_exclusive_file_lock(Path('active-role')) as lock: + assert lock is not None + Path('started').touch() + time.sleep(45) +""") + monkeypatch.setattr(roles, "role_cli_options", lambda *args: ()) + + class Admission: + def launch_lifetime(self, *, operation): + if operation == "turn_role_continue" and (project / "started").exists(): + raise InterruptedError("operator stop") + return nullcontext() + + host = roles.RoleHost(project, tmp_path / "runtime", tmp_path / "registry", tmp_path / "home", + str(binary), None, "low", "read-only", 10) + state = {"threads": {}, "steps": []} + native = roles.RoleTransport(host, time.monotonic() + 10, Admission(), state, lambda: None) + with pytest.raises(InterruptedError, match="operator stop"): + native.send("worker", "check", {}) + assert (project / "started").is_file() + with try_exclusive_file_lock(project / "active-role") as lock: + assert lock is not None + assert len(state["steps"]) == 1 and state["steps"][0]["thread_token_usage_total"] is None + + +def test_worker_can_iterate_until_the_host_deadline(monkeypatch): + from types import SimpleNamespace + + instructions = [f"Explore alternative {i}." for i in range(5)] + reports = iter(instructions) + worker_inputs = [] + + def send(role, prompt, schema): + if role == "worker": + worker_inputs.append(prompt) + return {"finding": "Measured another candidate.", "artifacts": []} + instruction = next(reports, None) + return {"instruction": instruction, "result": None if instruction else {"result_kind": "validated_progress"}} + + monkeypatch.setattr(roles, "prepare_codex_controller_input", lambda _: ("task", {})) + monkeypatch.setattr(roles, "normalize_codex_result", lambda _, result: result) + result = roles.run_roles(SimpleNamespace(send=send), {}) + assert result["result_kind"] == "validated_progress" and worker_inputs == instructions + + +@pytest.mark.parametrize("permission", ["danger-full-access", "read-only"]) +def test_native_role_permissions_cap_the_callers_grant(tmp_path, permission): + project = tmp_path / "task" + project.mkdir() + host = roles.RoleHost(project, tmp_path / "state", tmp_path / "registry", tmp_path / "home", + "codex", None, "low", permission, 10) + from dataclasses import replace + host = replace(host, controller_writable=True) + for role in ("controller", "worker"): + name, config = roles.role_policy(role, host) + rules = config["permissions"][name] + root = rules["filesystem"][":workspace_roots"] + assert root["."] == ("write" if permission != "read-only" else "read") + assert root[".loopx"] == root[".codex"] == "deny" + assert rules["filesystem"][str(host.runtime)] == "deny" + assert rules["filesystem"][str(host.codex_home)] == "deny" + assert rules["network"]["enabled"] is False + + +def test_correction_readback_never_starts_another_uncertain_or_completed_attempt(tmp_path): + path, journal = tmp_path / "roles.json", tmp_path / "turn.json" + request = {"turn_key": "original", "work_contract": {"todo": "selected"}, "result_contract": {}} + evidence, _ = roles.prepare_role_attempt(path, journal, request) + evidence["result"] = {"result_kind": "validated_progress"} + path.write_text(json.dumps(evidence)) + recovery = {**request, "result_contract": {"correction": "Correct the original receipt."}} + with pytest.raises(RuntimeError, match="original role attempt"): + roles.prepare_role_attempt(tmp_path / "missing-original.json", journal, recovery) + journal.write_text(json.dumps({"status": "committed"})) + with pytest.raises(RuntimeError, match="no authorized Host retry"): + roles.prepare_role_attempt(path, journal, recovery) + + journal.write_text(json.dumps({"recovery_audit": { + "planned": {"retry_failed": True, "reinvoke_host": True, "resume_from": "host_execute"}, + "actual": {"status": "started"}, + }})) + with pytest.raises(RuntimeError, match="original role attempt"): + roles.prepare_role_attempt(path, journal, {**recovery, "work_contract": {"todo": "other"}}) + historical = dict(evidence) + del historical["transport"] + path.write_text(json.dumps(historical)) + assert roles.prepare_role_attempt(path, journal, request)[1]["result"] == historical["result"] + with pytest.raises(RuntimeError, match="original role transport"): + roles.prepare_role_attempt(path, journal, recovery) + path.write_text(json.dumps(evidence)) + evidence, correction = roles.prepare_role_attempt(path, journal, recovery) + path.write_text(json.dumps(evidence)) + with pytest.raises(RuntimeError, match="correction requires reconciliation"): + roles.prepare_role_attempt(path, journal, recovery) + correction["result"] = {"result_kind": "validated_progress", "work_evidence": ["artifact"]} + path.write_text(json.dumps(evidence)) + assert roles.prepare_role_attempt(path, journal, recovery)[1]["result"] == correction["result"] + assert roles.prepare_role_attempt(path, journal, request)[1]["result"] == evidence["result"] + + +@pytest.mark.skipif(os.environ.get("LOOPX_TEST_NATIVE_ROLE_HOST") != "1", + reason="opt-in installed Codex with native filesystem isolation") +@pytest.mark.parametrize("entry", ["turn", "benchmark", "receipt-repair", "goal-stop"]) +def test_real_native_feedback_changes_the_checked_artifact(tmp_path, entry): + from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture + from tests.control_plane.codex_checkpoint_fixture import local_inference_provider + + src = Path(__file__).resolve().parents[2] + fixture = runpy.run_path(str(src / "examples/loopx-turn-codex-cli-e2e-smoke.py")) + project, runtime, _, registry = fixture["_write_fixture"](tmp_path, turn_count=2) + goal, agent = fixture["GOAL_ID"], fixture["AGENT_ID"] + initialize_canonical_source_fixture(runtime, goal, + state_path=project / ".codex" / "goals" / goal / "ACTIVE_GOAL_STATE.md", provider="sqlite") + (tmp_path / "skills").mkdir() + marker, expected = fixture["MARKER_NAME"], fixture["_marker_value"](1) + (project / "observation").write_text(expected) + source_task = "Inspect the observation.\nReport the checksum." + task_file = tmp_path / "worker-task.txt" + task_file.write_text(source_task) + pinned_task = entry != "turn" + if entry == "goal-stop": + (project / "docs").mkdir(exist_ok=True) + tool_started = project / "docs" / "tool-started" + late_write = project / "docs" / "after-stop" + key = None + inputs = [] + def respond(request, index, message): + nonlocal key + assert not any("" in part.get("text", "") + for item in request["input"] if item.get("role") == "developer" + for part in item["content"] if isinstance(part, dict)) + text = [part["text"] for item in request["input"] if item.get("role") == "user" + for part in item["content"] if "text" in part][-1] + if entry == "goal-stop" and index == 2: + command = ("/bin/sleep 6 & tool_pid=$!; " + "printf '%s\\n' \"$$ $tool_pid\" > docs/tool-started; " + "wait \"$tool_pid\"; /usr/bin/touch docs/after-stop") + return {"type": "function_call", "id": "fc_stop", "call_id": "call_stop", + "name": "exec_command", "arguments": json.dumps({"cmd": command, "yield_time_ms": 1000})} + if entry == "goal-stop" and index > 2: + # Keep the Worker waiting for its tool while the Goal is stopped. + time.sleep(6) + return message + if entry == "receipt-repair" and index == 9: + return {"type": "function_call", "id": "fc_repair_write", "call_id": "call_repair_write", + "name": "exec_command", "arguments": json.dumps({"cmd": "touch docs/receipt-write-probe"})} + if entry == "receipt-repair" and index == 10: + output = str(request["input"][-1]["output"]).lower() + assert "denied" in output or "not permitted" in output + index -= 1 + if index == 4: + return {"type": "function_call", "id": "fc_controller_check", "call_id": "call_controller_check", + "name": "exec_command", "arguments": json.dumps({"cmd": "cat observation > docs/controller-check; if cat .loopx/registry.json >/dev/null; then exit 17; fi; cat docs/controller-check"})} + if index == 5: + output = str(request["input"][-1]["output"]).lower() + assert expected.lower() in output + assert "denied" in output or "not permitted" in output + if index >= 5: + index -= 1 + inputs.append(text) + if index == 1: + contract, _ = json.JSONDecoder().raw_decode(text.split("Turn request:\n", 1)[1]) + key = contract["turn_key"] + answer = {"instruction": "dispatch" if pinned_task else "Inspect the observation.", "result": None} + elif index in (2, 5): + command = ("if cat .loopx/registry.json >/dev/null; then exit 17; fi; cat observation" + if index == 2 else "tr -d '\\n' > " + marker + " < 0 for s in observation["steps"]) + assert fixture["_quota_spend_count"](runtime) == 1 + + if entry == "turn": + from loopx.control_plane.turn_driver import build_loopx_turn_host_request + + journal = json.loads((record.parent.parent / record.name).read_text()) + original = build_loopx_turn_host_request(journal["plan"]) + capture = Path(original["detail_ref"]["captured_decision"]["path"]) + changed_source = json.loads(capture.read_text()) + changed_source["recommended_action"] = "An unrelated action." + capture.write_text(json.dumps(changed_source)) + host = roles.RoleHost(project, runtime, registry, home, binary, None, None, "workspace-write", 5) + with pytest.raises(ValueError, match="original decision differs"): + roles.serve(original, host) diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index b5b0b3464d..a0cb10666f 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -4194,7 +4194,8 @@ def recording_spend_quota_slot(*args: object, **kwargs: object) -> dict[str, obj host_calls: list[str] = [] - def fake_codex_host(request: dict[str, object], **_kwargs: object) -> dict[str, object]: + def fake_codex_host(request: dict[str, object], **kwargs: object) -> dict[str, object]: + assert kwargs["timeout_seconds"] == 3.0 host_calls.append(str(request["turn_key"])) return { "schema_version": "loopx_turn_result_v0", @@ -4244,6 +4245,8 @@ def fake_codex_host(request: dict[str, object], **_kwargs: object) -> dict[str, "codex-fixture", "--host", "codex-cli", + "--timeout-seconds", + "3", "--project", str(project), "--validation-command-json", diff --git a/tests/test_public_source_reader.py b/tests/test_public_source_reader.py index b461adc648..564462d296 100644 --- a/tests/test_public_source_reader.py +++ b/tests/test_public_source_reader.py @@ -678,7 +678,8 @@ def test_source_only_host_worker_needs_no_site_installed_loopx(tmp_path, monkeyp import sys import venv host = tmp_path / "source-only-host" - venv.EnvBuilder(with_pip=False).create(host) + # Standalone POSIX Python keeps its shared library beside the real executable. + venv.EnvBuilder(with_pip=False, symlinks=sys.platform != "win32").create(host) interpreter = host / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python") absent = subprocess.run([str(interpreter), "-I", "-c", "import importlib.util; assert importlib.util.find_spec('loopx') is None"],