interface:
  display_name: "Orchestrate"
  short_description: "Coordinator mode: route real work to benchmarked workers"
  default_prompt: "Use $orchestrate to act as a coordinator. Explicit GPT-6 Astra executor admission does not change model ladders or reviewer eligibility. For delegated workflow work, the coordinator dispatches every mandatory reviewer, resumes the same worker with the verified verdict, and missing review evidence is unavailable."

# Claude worker dispatch

Use this adapter for every delegated Claude worker: exploration,
implementation, review, and follow-up. Do not dispatch a Claude worker through
the Agent tool, the Workflow tool, or a background agent — those routes are
retired for delegated work by work order 149. `skills/drivers/orchestrate/scripts/claude-worker.py`
is the only dispatch path, mirroring `codex-worker.py`'s command surface:
`start`, `resume`, `stop`, `verify`, `--state`, `--model`, `--effort`,
`--sandbox read-only|workspace-write`, `--cwd`, `--control-checkout`, and
`--claude` (defaults to `claude`, the CLI binary on PATH) in place of
`codex-worker.py`'s `--codex`.

## Command surface

Run `claude-worker.py start` for a new worker. Give read work a resolved
repository or worktree path and `--sandbox read-only`. Give write work a
resolved isolated worktree path, `--sandbox workspace-write`, and the
coordinator checkout in `--control-checkout`; the adapter rejects a
`--cwd` inside the control checkout, exactly as codex-worker.py does. Persist
one state file per worker. Use `resume` with that state file for retry
findings and ordinary follow-ups; it restores the captured model, sandbox,
effort, and canonical cwd rather than taking replacements — the same contract
as codex-worker's resume.

## Hosted source access

`start --network` opts the worker into provider-hosted source retrieval. The
adapter maps that capability to `--allowedTools WebSearch,WebFetch`; the default
path enforces offline behavior with `--disallowedTools WebSearch,WebFetch`. The
adapter persists the boolean in lifecycle state, replays it on `resume` without
a replacement flag, and reports it in successful output. Omitting the option
leaves hosted source tools explicitly denied and reports `network: false`;
existing state without the field remains valid and resumes offline. This
capability promises hosted web search and fetch only; shell-command networking,
command egress,
private-network access, credentials, and a wider filesystem sandbox are outside
the contract.

## Prompt on stdin, not as an argument

`claude-worker.py` writes the prompt to the worker's stdin and closes it,
unlike `codex-worker.py`, which still takes the prompt positionally. The
`claude` CLI's `-p` invocation carries variadic flags (`--tools`,
`--allowedTools`, `--disallowedTools`) that would swallow a trailing
positional prompt, so a prompt-on-stdin is a fact about the `claude` CLI, not
a stylistic choice — never pass the prompt as a trailing argv token to a
`claude-worker.py`-launched command.

## Sandbox modes

Both modes are generated settings files the adapter owns and writes to a
temp path at launch — it never reads a settings shape out of `docs/`, because
`docs/` is not part of an installed skill copy.

- **`read-only`**: `sandbox.enabled: true`, `allowUnsandboxedCommands: false`,
  `filesystem.denyWrite: ["/", "~/"]` (writes denied filesystem-wide), and
  `permissions.deny: ["Write", "Edit", "NotebookEdit"]` (the edit tools
  denied). `allowUnsandboxedCommands: false` is load-bearing: it disables the
  `dangerouslyDisableSandbox` retry, so a command the sandbox blocked cannot
  be re-run outside it.
- **`workspace-write`**: `sandbox.enabled: true`, `allowUnsandboxedCommands:
  false`, `filesystem.allowWrite: [<the worker's own --cwd>]`, and
  `permissions.allow: ["Write", "Edit", "NotebookEdit"]` (the edit tools
  allowed). `allowWrite` is what does the confining, and it must stand alone:
  a real run of #149 launched with no `filesystem` block wrote straight
  through to `$HOME/.cache`, and pairing `allowWrite` with a `denyWrite` was
  rejected on disk twice because `deny` beats `allow` for the same path,
  silently re-blocking the very cwd `allowWrite` exists to carve out.

  The real boundary is **cwd plus the session temp directory**, which the
  sandbox documents as writable. A write under `$TMPDIR` is therefore expected
  and is not a confinement failure; a write into the operator's home directory
  or into the control checkout is. `start` additionally refuses a
  `workspace-write` `--cwd` that sits inside `--control-checkout` before any
  worker launches. Do not describe this mode as "the worktree and nowhere
  else" — that phrasing was measured false. The captured runs are in
  `docs/scope/149-probes/run-log.md`.

## Effort

Every delegation carries `--effort`, defaulting to `medium` for every model
because no effort benchmarking exists yet (see `references/routing-table.md`
Effort notes). `claude-worker.py`'s enum is `low|medium|high|xhigh|max`,
captured in `docs/scope/149-probes/effort-enums.md` from `claude --help`; it
is not the same set as `codex-worker.py`'s `none|low|medium|high|xhigh|max`
(probed against the live API on 2026-08-27).
The chosen effort is persisted in state and replayed on `resume`.

## Liveness is process identity only

A Claude worker has no rollout file and no headroom fields — there is no
Claude analogue of Codex's `~/.codex/sessions/*.jsonl` or its
`token_count.rate_limits` payload. `claude-worker.py` does not invent one:
liveness is proving the recorded PID/PGID/SID/cwd/birth identity still
matches, the same process-family probe codex-worker.py uses, and nothing
more. Do not infer a Claude worker's health from elapsed time or output
volume; there is no equivalent "CPU time growing" heuristic documented for
this adapter.

## Interrupted workers

The coordinator that launched a worker owns its recovery, exactly as with
Codex workers: run `claude-worker.py stop --state STATE --cwd WORKTREE`, then
`claude-worker.py verify --state STATE --cwd WORKTREE` and require success
before a successor receives the worktree. A successor never discovers or
cleans unknown processes; only the adapter's recorded dedicated process group
is in scope.

## Output

On success the adapter emits one public JSON object: `session_id`, `model`,
`sandbox`, `cwd`, `effort`, `network`, `final_message` (the CLI's `result` field), and
`permission_denials` (whatever the CLI's JSON payload reported as denied, so
the coordinator can see what the sandbox refused — an empty list when
nothing was denied). `is_error: true` in the CLI's own JSON is treated as a
dispatch failure, not a worker answer; the adapter fails instead of emitting.

## Presence check

Before dispatching any Claude worker, the coordinator runs `command -v claude`.
If neither `codex` nor `claude` is present on PATH, the coordinator
cannot dispatch at all: report the blocker to the operator and stop — there
is no third route to fall back to.

# Claude Code parent dispatching a Codex worker

Use this reference only when the interactive coordinator is a Claude Code
parent dispatching a **Codex** worker for review. `dispatch-codex.md` remains
Codex-UI-parent-only. This document adds no route or reviewer precedence of its
own.

## Review admission

Read [`review-routing.md`](review-routing.md) and apply its four-row matrix for
reviewer classification and initial adapter/model selection. That matrix
composes the selected review skill's area with `routing-table.md` and the Codex
presence/headroom gate in `SKILL.md`; this adapter reference does not duplicate
their precedence.

When the matrix selects Codex, dispatch the review read-only through
`codex-worker.py start` and persist one state file for that worker. Retry a
model-quality failure once through `codex-worker.py resume` against the same
state file, carrying the specific finding; do not start a second worker for the
retry. Review is a read task, so `workspace-write` is never correct for it.

Codex fast mode is available on this path as `codex-worker.py start --fast`,
but review is a load-bearing verdict: leave fast mode off for it. An explicit
operator selection of GPT-6 Astra (`--model gpt-6-astra`) is an executor
admission and never a reviewer selection; the matrix above still picks the
reviewer.

## Infrastructure failure vs. model-quality failure

A worker that failed to launch, hung before session start, lost its rollout,
or was refused for headroom is a dispatch failure, matching the rule
`dispatch-codex.md` already states. It is not evidence that Luna reviewed
badly, so it consumes neither the one same-session retry in `Review admission`
above nor an escalation rung. See `dispatch-codex.md`'s "Worker liveness" and
"Interrupted workers" sections for the shared mechanics — they are not
duplicated here. The launching coordinator owns stop-then-verify before any
successor touches the worktree.

## Boundary

This document does not cover dispatching a **Claude** worker; use
[`dispatch-claude.md`](dispatch-claude.md) for that adapter.

# Codex UI parent dispatch (v0)

Use this adapter only when the interactive coordinator is a Codex UI parent.
All delegated exploration, implementation, review, and follow-up work runs
through the executable helper; do not use native `spawn_agent`. CLI workers are
the validated path and let the coordinator enforce both `cwd` and sandbox.

Run `skills/drivers/orchestrate/scripts/codex-worker.py start` for a new worker. Give
read work a resolved repository or worktree path and `--sandbox read-only`.
Give write work a resolved isolated worktree path, `--sandbox workspace-write`,
and the coordinator checkout in `--control-checkout`; the helper rejects the
control checkout. Persist one state file per worker. Use `resume` with that
state file for retry findings and ordinary follow-ups; it restores the captured
model, sandbox, and canonical cwd rather than taking replacements.

The adapter's successful probe establishes dispatch capability, not a review
ranking. An explicitly admitted Astra coordinator may dispatch the routes their
existing locks select, but this page neither admits Astra as a reviewer nor changes
the review-routing precedence.

## Hosted source access

`start --network` opts the worker into provider-hosted source retrieval. The
adapter maps that capability to `-c web_search=live -c tools.web_search=true`,
persists the boolean in lifecycle state, replays it on `resume` without a
replacement flag, and reports it in successful output. Omitting the option
keeps the provider argv offline and reports `network: false`; existing state
without the field remains valid and resumes offline. This capability promises
hosted web search and fetch only; shell-command networking, command egress,
private-network access, credentials, and a wider filesystem sandbox are outside
the contract.

When the worker must judge rendered evidence, pass every evidence image as a
repeated `--image /absolute/path/to/file` argument on `start`. A path mentioned
only in the prompt is not an attachment. `resume` accepts the same repeated
option when a follow-up introduces new or revised evidence. Do not dispatch a
rendered-evidence review until the selected model accepts image input and every
image the verdict depends on is attached.

The helper closes a worker's stdin itself. Any hand-rolled background `codex
exec` must redirect stdin from /dev/null (or pipe a prompt deliberately and let
it reach EOF): an inherited open stdin is a permanent pre-session hang, because
codex reads stdin for an appended `<stdin>` block and blocks until EOF even when
the prompt was passed as an argument.

On success the helper emits one public JSON object. Read its `final_message`
field as the current worker's answer; do not infer an answer from its session or
headroom metadata.

## Fast mode

`start --fast` runs the worker in Codex fast mode. The adapter maps the option to
`--enable fast_mode`, persists the boolean in lifecycle state, replays it on
`resume` without a replacement flag, and reports it as `fast` in successful
output. Omitting the option passes `--disable fast_mode` explicitly, so a worker
never inherits whatever `features.fast_mode` the operator's `~/.codex/config.toml`
happens to set; existing state without the field resumes with fast mode off.

Fast mode trades reasoning depth for latency. Ask for it when the operator asks
for it, or for a bounded, low-judgment run such as the headroom probe. Do not
use it for review, plan/spec writing, or any load-bearing verdict, and never read
a fast-mode result as evidence about a model's benchmarked rung.

## Explicit Astra selection

An operator may explicitly select GPT-6 Astra as the worker model; pass
`--model gpt-6-astra` to the helper. This is an operator admission to execute,
not a benchmarked route: it never enters the admission table below, never
selects a reviewer, and never escalates to or from a benchmarked rung. Astra
takes `--effort` like any other Codex model, so state the effort the order
requires rather than inferring a hidden one. `review-routing.md` remains the sole
authority for review precedence.

## Worker liveness

When the adapter is still running but has no terminal output or session ID, wait
another minute and check again. `session_id: ""` while `lifecycle: running` is
indeterminate, not a pre-session failure. Silence, PID presence, or low parent CPU
alone does not prove a hang and does not authorize stopping the worker.

A PID appearing in `ps` proves nothing. A healthy worker accrues CPU time
within a minute and writes `~/.codex/sessions/<date>/rollout-*.jsonl` at session
start. Before trusting a long-running worker, check that `ps -o time` is growing
and that the rollout file exists; a worker with neither hung before session
start. Such a worker burned no tokens. The coordinator stops it through the
recovery below and reports; it may then relaunch into a fresh directory, so a
late-waking zombie cannot clobber the new run.

## Interrupted workers

The coordinator that launched a worker owns its recovery. Before handing a
preserved worktree to a successor, run `codex-worker.py stop --state STATE --cwd
WORKTREE`, then run `codex-worker.py verify --state STATE --cwd WORKTREE` and
require success. State is retained for interrupted, failed, and completed runs.
Only the helper's recorded dedicated process group is in scope; a successor must
never search for or clean unknown processes, tests, providers, sessions, or
descendants. `stop` and `verify` reject legacy state. Ordinary `resume` can read
a completed legacy state, but only from a terminal state with a session ID.

The adapter reports session-bound headroom only when the matching persisted
rollout contains rate limits. `unknown`, headroom at or below 5%, and rate-limit
failures block a Codex UI parent: stop dispatching and report the blocker. It
cannot switch to Claude workers. Infrastructure failures (including a missing
rollout) are dispatch failures, not evidence that a model tier failed its task.

## Served evidence and the exec bridge

Field-derived provenance (source: #130; **field-validated (provisional)**): the
exec bridge reaps long-lived and detached processes, so a server started in one
invocation is gone by the next. Instruct workers to use one invocation to start
the server, wait for LISTEN, run the check, and kill the server. Do not read
repeated served-evidence failure as a model-tier weakness; it can bias workers
toward bounded runs instead of full replays.

Field-derived provenance (source: #130; **field-validated (provisional)**): a
worker lost to `invalid JSONL on line <n>` is a dispatch failure. Its session is
unrecoverable but committed on-disk work survives; recover the worktree under
"Interrupted workers" and resume it. Do not re-route.

## Codex-only admission routes

These are the only validated initial routes in Codex-only mode, plus the one
provisional implementation-escalation exception below:

| Area | Initial route |
|---|---|
| Bounded exploration | Luna (`gpt-5.6-luna`) |
| Hermetic implementation | Terra (`gpt-5.6-terra`) |
| Plan / spec writing | Terra (`gpt-5.6-terra`) |
| Prototyping | Sol (`gpt-5.6-sol`) |
| Default brainstorming | Terra (`gpt-5.6-terra`) |
| Documentation | Luna (`gpt-5.6-luna`) |
| Implementation escalation (not an initial admission) | Sol (`gpt-5.6-sol`) — field-validated (provisional; source: #130) |

Full-system exploration, novelty-as-deliverable brainstorming, and
other unlisted admissions are **NO_VALIDATED_ROUTE** until benchmarked.
Do not invent an escalation for those admissions. Each admitted v0 route is one
validated rung: retry once in the same worker session, then stop with
**NO_VALIDATED_ROUTE**. Never escalate Terra, Luna, or Sol to Sonnet or Opus.
Pass the exact parenthesized CLI model ID to the helper's `--model` argument.

Field-derived provenance (source: #130; **field-validated (provisional)**): for
hermetic implementation only, Terra may escalate to Sol once after its
in-session retry fails. At Sol, stop and surface. This exception changes no
benchmarked ladder.

Review admissions are owned by [`review-routing.md`](review-routing.md). Its
matrix includes the Codex UI parent's routine routes and explicit-operator
exception; do not treat review as a generic admission above.

Field-derived provenance (source: #130; **field-validated (provisional)**):
Codex-only load-bearing review remains **NO_VALIDATED_ROUTE** by default. An
operator may explicitly choose the unvalidated Codex exception in
`review-routing.md`; label the resulting review unvalidated rather than
silently promoting it into the benchmark table. Rendered evidence is a transport
constraint, not a model-family ban: use the adapter's `--image` arguments and
require image-input support from the selected model. `review-routing.md` remains
the sole authority for review precedence.

# Reviewer routing

## Reviewer selection contract

This reference is the sole live authority for reviewer classification, reviewer
eligibility, and reviewer-model precedence. The benchmark authority remains
[`routing-table.md`](routing-table.md).

For a review with a work order, effective review depth is the routing input:
Focused and Targeted are routine; Full is load-bearing. An order with no stamped
depth retains [review-depth.md](../../ticket/references/review-depth.md)'s Targeted
default.

For every review without a work order—a bare diff, chat plan, file-backed PRD, design document, or GitHub issue—the dispatcher judges the subject against [review-depth.md](../../ticket/references/review-depth.md)'s sensitivity floor: any of its four categories makes the subject load-bearing; otherwise it is routine.

Precedence is fixed: effective depth, or the judged sensitivity floor when there
is no order, produces routine/load-bearing routing stakes; the selected review
skill supplies its routing-table area; `routing-table.md` supplies that row's
candidate model or ladder; parent policy and the Codex presence/headroom gate
remove unavailable candidates. Builder tier is never an input, and a fallback is
never borrowed from another row.

Haiku never reviews.

Reviewer-routing stakes and plan-review's plan stakes tier are independent.
Neither derives from, overrides, or rewrites the other.

An explicit Astra executor/coordinator admission is not reviewer evidence. It
does not alter this matrix, the Full-review route, or headroom handling. When the
selected review route cannot run, report it as unresolved; never promote Astra or
silently downgrade the review.

| Review skill | Routing stakes | Initial route |
|---|---|---|
| code-review | routine | Run the Codex presence/headroom gate first. With usable Codex, use Luna from the Code review row. On absent, unknown, ≤5%, or rate-limited Codex, enter Claude-only mode at Sonnet from the Code review row and make no second Codex attempt for the session. |
| code-review | load-bearing | Use Opus directly from the Code review row; select no Codex rung. |
| plan-review | routine | Run the Codex presence/headroom gate first. With usable Codex, use Terra from the Plan / spec writing row. On absent, unknown, ≤5%, or rate-limited Codex, enter Claude-only mode at Opus from the Plan / spec writing row, which has no Sonnet rung, and make no second Codex attempt for the session. |
| plan-review | load-bearing | Use Opus directly from the Plan / spec writing row; select no Codex rung. |

For a Codex UI parent, routine `code-review` uses Luna and routine `plan-review`
uses Terra, subject to the same presence/headroom gate. Load-bearing review has
no benchmark-validated Codex route. When the operator explicitly directs the
workflow to use Codex anyway, honor that product choice with Luna for
`code-review` or Terra for `plan-review`, label the review **unvalidated**, and
retain the ordinary same-session retry cap. An explicit choice does not promote
the route or alter the benchmark table.

Rendered evidence does not change these stakes. Before admitting a Codex
reviewer, verify that the selected model accepts image input and attach every
required render through `codex-worker.py --image`; a filename in the prompt is
not an attachment. If the images are unavailable, report an evidence-transport
blocker instead of a model-capability blocker.

Opus in the Plan / spec writing ladder is an availability rung, not a benchmarked
plan-writing win.

The matrix chooses only the initial reviewer adapter/model. Existing read-only
sandboxing, explicit effort, same-session retry, escalation, liveness, recovery,
and worker-state mechanics remain owned by orchestrate and the merged #145, #149,
#150, #151, and #152 work.

## Related authorities

Use `routing-table.md` for benchmarked areas, ladders, scores, and effort notes;
use `review-depth.md` for depth, sensitivity, hardening, blocking, and whole-diff
behavior.

# Model routing table

Benchmarked 2026-08-03 against real replayed tasks from the operator's repos: the
public agentflow fleet daemon plus two private apps (a telemetry/tuning app and a
recipe collection). Ground truth: the actually-merged fixes, a locked mockup spec,
and a planted-defect review fixture. Models: Claude Opus 5, Sonnet 5, Haiku 4.5; GPT-5.6-Sol, -Terra, -Luna,
GPT-5.3-Codex-Spark (full suite); GPT-5.5, GPT-5.4, GPT-5.4-Mini (light pass: plan +
review only). Codex runs at `model_reasoning_effort=medium`. Scores 1–5, judged blind by
the coordinator against ground truth; published-eval data was a secondary signal only.
Re-benchmark when a new model ships — and note Claude routes dispatch via unversioned aliases (`opus`/`sonnet`/`haiku`), so re-verify this stamp when Anthropic rolls a snapshot, not only on a named launch. See `benchmark/README.md`.

Cost tiers as of 2026-08-03 — routes were selected at these prices; re-check on a
price move (per M tokens in/out): Opus $5/$25 · Sonnet $3/$15 · Haiku $1/$5 ·
Sol $5/$30 · Terra $2.50/$15 · Luna $1/$6. Spark, GPT-5.5, GPT-5.4, and
GPT-5.4-Mini are flat-rate under the operator's ChatGPT plan (price them normally if you pay
per token) — under flat rate they sit outside the cheapest-clears-bar rule and are chosen only where a route
below names them explicitly (Spark for latency; GPT-5.4 as a plan alternate).

## Provenance labels

**Provenance labels (source: #130).** Unlabeled routes are benchmarked (blind-judged
against ground truth, stamp above). A route or note marked **field-validated
(provisional)** comes from production observation only: uncontrolled, unpaired,
and small-N. Provisional entries may be used, but are promoted to benchmarked only
by replay, and may not displace a benchmarked route's ordering.

## Review-consumer classification

Before applying their own named area row, `code-review` and `plan-review` obtain
routine/load-bearing routing stakes and their initial route directly from
[`review-routing.md`](review-routing.md).

Opus in the Plan / spec writing ladder is an availability rung, not a benchmarked
plan-writing win.

Explicit GPT-6 Astra executor/coordinator admission is deliberately outside this
table. It consumes authoritative current-session metadata and required adapter
capabilities; it changes no route, ladder, effort default, benchmark result, or
reviewer eligibility.

## Routes (cheapest that clears the bar) and escalation ladders

Escalate one step at a time along the row's ladder; at the last rung, stop and
surface both failed attempts to the operator.

| Area | Route | Ladder | Why |
|---|---|---|---|
| Exploration / codebase-mapping | **Luna** for bounded lookups; **Sonnet** for full-system maps | Luna → Sonnet → Opus | Sonnet tied Opus at 5/5 with fully verified citations at 60% of the cost; Luna scored 4 at a fraction of both. Haiku fabricated a citation — do not use for exploration you won't verify. |
| Hermetic implementation | **Terra** | Terra → Sonnet → Opus | Terra matched the merged fix exactly (incl. the window-bounds subtlety) at $2.50; Sonnet/Opus scored 5 with richer tests — escalate for correctness-critical or gnarly changes. Luna/Haiku/Spark all missed a subtle placement decision. Field-derived provenance (sources: #144's session-fit comment and epic ledger PR #136): Terra failed canonical byte-for-byte prose-contract work on #151 and #152, required escalation, and informed #144's stamp; this observation changes neither Route nor Ladder. Field-derived provenance (source: #130; **field-validated (provisional)**): across 11 tickets, Terra was substantively correct first pass on 10 and made zero fabrications in approximately 20 sessions, twice refusing to report unmeasured output. Watch records/test discipline (verify evidence provenance, not just presence) and aggregate-context test bugs that pass bounded runs but fail a full replay; require a full-aggregate run before accepting a replay story. |
| Plan / spec writing | **Terra** | Terra → Sol → Opus | The Codex family owns this area: Terra 5/5 (tightest, correct fail-closed), Sol/Luna/GPT-5.4 ≈4.8. Opus wrote the prettiest spec with a load-bearing polarity error — never route specs to Claude models without a fail-safe review. Field-derived provenance (sources: #144's session-fit comment and epic ledger PR #136): Terra failed canonical byte-for-byte prose-contract work on #151 and #152, required escalation, and informed #144's stamp; this observation changes neither Route nor Ladder. |
| Prototyping (incl. UI mockups) | **Sol** | Sol → Opus → none (top of ladder; Sol first despite Opus's lower sticker price — Sol's per-task token volume ran leaner and its output resolved a spec tension Opus ignored) | Sol, Opus, and Spark all hit 5; Sol resolved a spec tension the others ignored. Spark ties when repo context (an existing lock/design system) exists to reuse — and it's near-instant. Luna is banned here (1/5: no page geometry, invented UI, leaked never-print content). |
| Novel-solution brainstorming | **Terra**; **Opus** when novelty is the deliverable | Terra → Opus → none (top of ladder) | Opus 5/5 with the most novel idea of the whole benchmark; Terra 4.5 at half the price. Haiku and Spark produce generic-ML re-skins — don't route ideation there. |
| Documentation writing | **Haiku** (default; Luna equal-scored alternate) | Haiku → Opus → Sol (Opus first: equal score, lower price) | Both scored 4 at ~$1; Opus and Sol scored 5 — escalate for load-bearing ADRs. Sonnet (3) narrated implementation identifiers; Spark (2) fabricated a cross-reference. |
| Code review | **Luna** for routine PRs; **Opus** for load-bearing/safety review | Luna → Sonnet → Opus; Opus route: none (top of ladder) | Opus was the only model to catch all 3 planted defects (incl. a silently weakened test). Luna caught 2/3 with zero false positives at the lowest cost. GPT-5.5 confidently reported a nonexistent syntax error; GPT-5.4-Mini missed a blatant inverted guard — avoid both for review. Field-derived provenance (source: epic ledger PR #136's 2026-08-25 rounds): Sol produced zero hallucinated findings across approximately 20 Full-depth and cold reviews in one epic session, with every blocking finding reproduced against the tree, grounding the standing Codex-first review practice. **Rendered evidence:** require an image-capable reviewer and attach every required render through the selected adapter. Codex image input is admitted through `codex-worker.py --image`; do not mistake missing evidence transport for a model-family limitation. `review-routing.md` remains the review-precedence authority. Field-derived provenance (source: #190; **field-validated (provisional)**): Luna returned five PASS contract-review verdicts with file:line citations and zero false positives, corroborating its existing routine-review route without changing it. |

## Benchmarked replay — 2026-08-27 review fixture

**Benchmarked replay (2026-08-27; settles #130; N=1 regenerated private review
fixture).** The Luna anchor caught 2 of 3 planted defects with zero false
positives, reproducing its benchmarked 2026-08-03 review profile and calibrating
the regenerated fixture. Sol caught the same two defects, missed the same subtle
boundary defect, and added one false positive. Sol's mechanical score was 1
against Luna's 2 (catches minus confident false positives), so Sol does not
out-detect Luna on planted defects at five times the price.

This measured result supports retaining Luna as the routine-review route and
gives no reason to revise Opus for load-bearing/safety review: neither Codex
model caught the subtle boundary defect. Codex-only load-bearing review remains
**NO_VALIDATED_ROUTE**; Sol is not promoted.

This is one fixture and one run per model, not a re-benchmark. The un-tipped
prompt omitted the phrase that says there are exactly three planted defects for
both models; Luna's anchor reproduced its prior profile, which calibrates that
deviation. Sol's false positive was a soft over-demand for coverage of
explicitly unchanged behavior. It does not contradict the existing field note
crediting Sol with zero hallucinated findings across approximately 20 reviews;
the narrow replay claim is the planted-defect and price result above.

## Effort notes (coarse, per spec decision 8)
- Every delegation carries an effort dial (per ADR 149,
  `docs/adr/adr-149-pack-owned-model-dispatch.md`), defaulting to medium for
  every model — overridable per delegation, never left implicit. The default
  is uniform because no effort benchmarking exists yet: all scored runs used
  Codex medium effort, and it was sufficient everywhere tested. Changing a
  model's default effort is a benchmark result, not a preference, and gets set
  only when a replay measures one; escalation changes the model tier, not the
  effort dial.
- The two adapters validate different enums, each a literal in its own
  script: `claude-worker.py` accepts `low|medium|high|xhigh|max`;
  `codex-worker.py` accepts `none|low|medium|high|xhigh|max` (unvalidated
  locally by the Codex CLI itself). A live Codex 5.6 API probe recorded this
  set on 2026-08-27. See
  `docs/scope/149-probes/effort-enums.md`.
- Spark's value is latency: use for tight edit-test loops and mockup iteration, not judgment.
- Opus spends ~4–5× Haiku's tokens on the same exploration prompt; route it only where the depth is the point.

## Light-tier verdicts (plan + review probes only)
- **GPT-5.5**: competent but padded; one confident false positive. No niche the 5.6 family doesn't fill better.
- **GPT-5.4**: genuinely strong spec-writer (4.7, line-level verified citations) — viable Terra alternate for plans.
- **GPT-5.4-Mini**: clean prose, weak review (missed a blatant inversion), invents scope. Avoid.

## Cross-cutting empirical findings
- Self-assessments were directionally honest about weaknesses (every model self-flagged visual prototyping; all were right except Sol/Spark, who beat their own expectations) but inflated on strengths — Sol claimed 5s across the board and scored 3 on brainstorming; Terra's claimed 5s on plan/docs held up, its review 4 didn't (3.5).
- The Claude models' edge is depth-with-verification (review, exploration); the Codex 5.6 family's edge is disciplined, executable specs and cheap accuracy.
- Never delegate to Fable (coordinator tier only — spec decision 4).
- Field-derived provenance (source: #130; **field-validated (provisional)**): a Terra → Sol escalation succeeded 2/2 for hard debugging after measured-failure handback. Sol also found review blockers including a live-reproduced unlabeled-stale race; its two false positives came from batch-level mandates omitted from the brief. Brief reviewers with every batch-level mandate that overrides repo defaults, or they can correctly identify a spec violation.
- Field-derived provenance (source: #190; **field-validated (provisional)**): **Executable-evidence repair** routes on execution access, not model tier. Terra for two rounds and Sol for two rounds could only reduce browser failures from 11 to 8 to 3 without execution; Opus with Chromium fixed the remaining test defect in one shot. Do not dispatch blind repair to a worker that cannot run the affected suite; when no admissible worker can execute it, surface the access blocker rather than consuming a model ladder.
- Field-derived provenance (source: #190; **field-validated (provisional)**): Sol can deepen hard frontend debugging without execution access, but it authored new executable evidence it could not run; both residual failures were defects in that new evidence.

#!/usr/bin/env python3
"""Launch, resume, stop, and verify one durable Claude worker process family."""

from __future__ import annotations

import argparse
import json
import os
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
import worker_lifecycle as lifecycle


EFFORT_LEVELS = {"low", "medium", "high", "xhigh", "max"}
DEFAULT_EFFORT = "medium"


def fail(message: str, code: int = 1) -> int:
    print(f"claude-worker: {message}", file=sys.stderr)
    return code


def parse_result(output: str) -> tuple[dict[str, Any] | None, str | None]:
    """Parse claude's single `--output-format json` object (not Codex's JSONL)."""
    try:
        payload = json.loads(output)
    except json.JSONDecodeError:
        return None, "invalid JSON in Claude output"
    if not isinstance(payload, dict):
        return None, "Claude output is not a JSON object"
    return payload, None


def final_result_message(payload: dict[str, Any]) -> tuple[str | None, str | None]:
    if payload.get("is_error"):
        return None, "Claude reported is_error"
    result = payload.get("result")
    if not isinstance(result, str):
        return None, "missing result field in Claude output"
    return result, None


def captured_session_id(payload: dict[str, Any]) -> tuple[str | None, str | None]:
    session_id = payload.get("session_id")
    if not isinstance(session_id, str) or not session_id:
        return None, "missing session ID in Claude output"
    return session_id, None


def effort_of(state: dict[str, Any]) -> str:
    return state.get("effort", DEFAULT_EFFORT)


def network_arguments(enabled: bool) -> list[str]:
    option = "--allowedTools" if enabled else "--disallowedTools"
    return [option, "WebSearch,WebFetch"]


def parse(output: str) -> tuple[str | None, str | None, Any, str | None]:
    payload, error = parse_result(output)
    if error:
        return None, None, None, error
    assert payload is not None
    session_id, error = captured_session_id(payload)
    if error:
        return None, None, None, error
    message, error = final_result_message(payload)
    return session_id, message, payload.get("permission_denials"), error


def emit(state: dict[str, Any], final_message: str, permission_denials: Any) -> None:
    print(json.dumps({
        "session_id": state["session_id"],
        "model": state["model"],
        "sandbox": state["sandbox"],
        "cwd": state["cwd"],
        "effort": effort_of(state),
        "network": state.get("network", False),
        "final_message": final_message,
        "permission_denials": permission_denials if permission_denials is not None else [],
    }))


def sandbox_settings(sandbox: str, cwd: Path) -> dict[str, Any]:
    """Build the installed Claude CLI's two sandbox settings shapes."""
    if sandbox == "read-only":
        return {
            "sandbox": {
                "enabled": True,
                "allowUnsandboxedCommands": False,
                "filesystem": {"denyWrite": ["/", "~/"]},
            },
            "permissions": {"deny": ["Write", "Edit", "NotebookEdit"]},
        }
    return {
        "sandbox": {
            "enabled": True,
            "allowUnsandboxedCommands": False,
            "filesystem": {"allowWrite": [str(cwd)]},
        },
        "permissions": {"allow": ["Write", "Edit", "NotebookEdit"]},
    }


def write_settings_file(sandbox: str, cwd: Path) -> Path:
    descriptor, path = tempfile.mkstemp(prefix="claude-worker-settings-", suffix=".json")
    with os.fdopen(descriptor, "w", encoding="utf-8") as file:
        json.dump(sandbox_settings(sandbox, cwd), file)
    return Path(path)


def start(args: argparse.Namespace) -> int:
    state, error = lifecycle.prepare_start(
        args, effort_levels=EFFORT_LEVELS, default_effort=DEFAULT_EFFORT
    )
    if error:
        return fail(error)
    assert state is not None
    settings = write_settings_file(args.sandbox, args.cwd)
    session_id = str(uuid.uuid4())
    command = [
        args.claude, "-p", "--model", args.model, "--effort", args.effort,
        "--permission-mode", "dontAsk", "--settings", str(settings),
        *network_arguments(state.get("network", False)),
        "--session-id", session_id, "--output-format", "json",
    ]
    return lifecycle.run_lifecycle(
        args, command, state, parse=parse, emit=emit, fail=fail,
        stdin_text=args.prompt, effort_levels=EFFORT_LEVELS,
    )


def resume(args: argparse.Namespace) -> int:
    fresh, expected, error = lifecycle.prepare_resume(
        args, effort_levels=EFFORT_LEVELS, default_effort=DEFAULT_EFFORT
    )
    if error:
        return fail(error)
    assert fresh is not None and expected is not None
    cwd = Path(fresh["cwd"])
    effort = effort_of(fresh)
    settings = write_settings_file(fresh["sandbox"], cwd)
    command = [
        args.claude, "-p", "--resume", fresh["session_id"], "--model", fresh["model"],
        "--effort", effort, "--permission-mode", "dontAsk", "--settings", str(settings),
        *network_arguments(fresh.get("network", False)),
        "--output-format", "json",
    ]
    return lifecycle.run_lifecycle(
        args, command, fresh, expected=expected, parse=parse, emit=emit, fail=fail,
        stdin_text=args.prompt, effort_levels=EFFORT_LEVELS,
    )


def verify(args: argparse.Namespace) -> int:
    code, error = lifecycle.verify_worker(args, effort_levels=EFFORT_LEVELS)
    return fail(error) if error else (code or 0)


def stop(args: argparse.Namespace) -> int:
    code, error = lifecycle.stop_worker(args, effort_levels=EFFORT_LEVELS)
    return fail(error) if error else (code or 0)


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    commands = result.add_subparsers(dest="command", required=True)
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--claude", default="claude")
    common.add_argument("--state", type=Path, required=True)
    common.add_argument("prompt", nargs="?")
    start_parser = commands.add_parser("start", parents=[common]); start_parser.add_argument("--model", required=True); start_parser.add_argument("--sandbox", choices=("read-only", "workspace-write"), required=True); start_parser.add_argument("--effort", default=DEFAULT_EFFORT); start_parser.add_argument("--network", action="store_true"); start_parser.add_argument("--cwd", type=lifecycle.resolved_directory, required=True); start_parser.add_argument("--control-checkout", type=lifecycle.resolved_directory); start_parser.set_defaults(handler=start)
    resume_parser = commands.add_parser("resume", parents=[common]); resume_parser.set_defaults(handler=resume)
    for name, handler in (("stop", stop), ("verify", verify)):
        command = commands.add_parser(name, parents=[common]); command.add_argument("--cwd", type=lifecycle.resolved_directory, required=True); command.add_argument("--grace-seconds", type=float, default=1.0); command.set_defaults(handler=handler)
    return result


if __name__ == "__main__":
    arguments = parser().parse_args()
    if arguments.command in {"start", "resume"} and not arguments.prompt:
        raise SystemExit(fail("prompt is required"))
    raise SystemExit(arguments.handler(arguments))

#!/usr/bin/env python3
"""Launch, resume, stop, and verify one durable Codex worker process family."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent))
import worker_lifecycle as lifecycle


# The 2026-08-27 live API probe recorded in docs/scope/149-probes/effort-enums.md
# accepted none, low, medium, high, xhigh, and max for Codex 5.6. The Codex CLI
# validates nothing locally, so this adapter owns the local guard. Not shared
# with claude-worker.py's enum.
EFFORT_LEVELS = {"none", "low", "medium", "high", "xhigh", "max"}
DEFAULT_EFFORT = "medium"


def fail(message: str, code: int = 1) -> int:
    print(f"codex-worker: {message}", file=sys.stderr)
    return code


def parse_jsonl(output: str) -> tuple[list[dict[str, Any]], str | None]:
    items: list[dict[str, Any]] = []
    for number, line in enumerate(output.splitlines(), 1):
        if not line.strip():
            continue
        try:
            item = json.loads(line)
        except json.JSONDecodeError:
            return [], f"invalid JSONL on line {number}"
        if not isinstance(item, dict):
            return [], f"JSONL item on line {number} is not an object"
        items.append(item)
    return items, None


def captured_thread_id(items: list[dict[str, Any]]) -> tuple[str | None, str | None]:
    if any(item.get("type") == "item.completed" and isinstance(item.get("item"), dict) and item["item"].get("type") == "error" for item in items):
        return None, "Codex reported an error item"
    ids = {item["thread_id"] for item in items if item.get("type") == "thread.started" and isinstance(item.get("thread_id"), str) and item["thread_id"]}
    return (ids.pop(), None) if len(ids) == 1 else (None, "missing or ambiguous thread ID in Codex JSONL")


def final_agent_message(items: list[dict[str, Any]]) -> tuple[str | None, str | None]:
    messages = [item["item"]["text"] for item in items if item.get("type") == "item.completed" and isinstance(item.get("item"), dict) and item["item"].get("type") == "agent_message" and isinstance(item["item"].get("text"), str)]
    return (messages[-1], None) if messages else (None, "missing completed agent message in Codex JSONL")


def latest_rate_limits(session_id: str) -> dict[str, Any] | None:
    root = Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")) / "sessions"
    matches = []
    for rollout in root.rglob("*.jsonl") if root.exists() else ():
        try:
            entries, error = parse_jsonl(rollout.read_text(encoding="utf-8"))
        except OSError:
            continue
        if error is None and any(entry.get("type") == "session_meta" and isinstance(entry.get("payload"), dict) and entry["payload"].get("session_id") == session_id for entry in entries):
            matches.append(rollout)
    if not matches:
        return None
    entries, error = parse_jsonl(max(matches, key=lambda item: item.stat().st_mtime_ns).read_text(encoding="utf-8"))
    if error:
        return None
    limits = None
    for entry in entries:
        payload = entry.get("payload")
        if entry.get("type") == "event_msg" and isinstance(payload, dict) and payload.get("type") == "token_count":
            limits = payload.get("rate_limits") if isinstance(payload.get("rate_limits"), dict) else None
    return limits


def effort_of(state: dict[str, Any]) -> str:
    return state.get("effort", DEFAULT_EFFORT)


def existing_file(value: str) -> Path:
    path = Path(value).resolve(strict=True)
    if not path.is_file():
        raise argparse.ArgumentTypeError("must be an existing file")
    return path


def image_arguments(paths: list[Path]) -> list[str]:
    return [value for path in paths for value in ("--image", str(path))]


def network_arguments(enabled: bool) -> list[str]:
    return ["-c", "web_search=live", "-c", "tools.web_search=true"] if enabled else []


# The operator's ~/.codex/config.toml may set features.fast_mode either way, so
# this adapter always states the choice rather than inheriting an ambient one.
def fast_arguments(enabled: bool) -> list[str]:
    return ["--enable" if enabled else "--disable", "fast_mode"]


def parse(output: str) -> tuple[str | None, str | None, Any, str | None]:
    items, error = parse_jsonl(output)
    if error:
        return None, None, None, error
    session_id, error = captured_thread_id(items)
    if error:
        return None, None, None, error
    message, error = final_agent_message(items)
    return session_id, message, None, error


def emit(state: dict[str, Any], final_message: str, _metadata: Any = None) -> None:
    limits = latest_rate_limits(state["session_id"])
    primary = limits.get("primary") if isinstance(limits, dict) else None
    remaining = 100 - primary["used_percent"] if isinstance(primary, dict) and isinstance(primary.get("used_percent"), (int, float)) else None
    print(json.dumps({"session_id": state["session_id"], "model": state["model"], "sandbox": state["sandbox"], "cwd": state["cwd"], "effort": effort_of(state), "network": state.get("network", False), "fast": state.get("fast", False), "final_message": final_message, "headroom": remaining, "headroom_status": "known" if remaining is not None else "unknown"}))


def start(args: argparse.Namespace) -> int:
    state, error = lifecycle.prepare_start(
        args, effort_levels=EFFORT_LEVELS, default_effort=DEFAULT_EFFORT
    )
    if error:
        return fail(error)
    assert state is not None
    effort = effort_of(state)
    command = [args.codex, "exec", "-m", args.model, "-c", f"model_reasoning_effort={effort}", *network_arguments(state.get("network", False)), *fast_arguments(state.get("fast", False)), "--sandbox", args.sandbox, "--skip-git-repo-check", "-C", str(args.cwd), "--json", args.prompt, *image_arguments(getattr(args, "image", []))]
    return lifecycle.run_lifecycle(
        args, command, state, parse=parse, emit=emit, fail=fail,
        stdin_text=None, effort_levels=EFFORT_LEVELS,
    )


def resume(args: argparse.Namespace) -> int:
    fresh, expected, error = lifecycle.prepare_resume(
        args, effort_levels=EFFORT_LEVELS, default_effort=DEFAULT_EFFORT
    )
    if error:
        return fail(error)
    assert fresh is not None and expected is not None
    effort = effort_of(fresh)
    command = [args.codex, "exec", "resume", fresh["session_id"], "-m", fresh["model"], "-c", f'sandbox_mode="{fresh["sandbox"]}"', "-c", f"model_reasoning_effort={effort}", *network_arguments(fresh.get("network", False)), *fast_arguments(fresh.get("fast", False)), "--skip-git-repo-check", "--json", args.prompt, *image_arguments(getattr(args, "image", []))]
    return lifecycle.run_lifecycle(
        args, command, fresh, expected=expected, parse=parse, emit=emit, fail=fail,
        stdin_text=None, effort_levels=EFFORT_LEVELS,
    )


def verify(args: argparse.Namespace) -> int:
    code, error = lifecycle.verify_worker(args, effort_levels=EFFORT_LEVELS)
    return fail(error) if error else (code or 0)


def stop(args: argparse.Namespace) -> int:
    code, error = lifecycle.stop_worker(args, effort_levels=EFFORT_LEVELS)
    return fail(error) if error else (code or 0)


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    commands = result.add_subparsers(dest="command", required=True)
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--codex", default="codex")
    common.add_argument("--state", type=Path, required=True)
    common.add_argument("--image", action="append", default=[], type=existing_file)
    common.add_argument("prompt", nargs="?")
    start_parser = commands.add_parser("start", parents=[common]); start_parser.add_argument("--model", required=True); start_parser.add_argument("--sandbox", choices=("read-only", "workspace-write"), required=True); start_parser.add_argument("--effort", default=DEFAULT_EFFORT); start_parser.add_argument("--network", action="store_true"); start_parser.add_argument("--fast", action="store_true"); start_parser.add_argument("--cwd", type=lifecycle.resolved_directory, required=True); start_parser.add_argument("--control-checkout", type=lifecycle.resolved_directory); start_parser.set_defaults(handler=start)
    resume_parser = commands.add_parser("resume", parents=[common]); resume_parser.set_defaults(handler=resume)
    for name, handler in (("stop", stop), ("verify", verify)):
        command = commands.add_parser(name, parents=[common]); command.add_argument("--cwd", type=lifecycle.resolved_directory, required=True); command.add_argument("--grace-seconds", type=float, default=1.0); command.set_defaults(handler=handler)
    return result


if __name__ == "__main__":
    arguments = parser().parse_args()
    if arguments.command in {"start", "resume"} and not arguments.prompt:
        raise SystemExit(fail("prompt is required"))
    raise SystemExit(arguments.handler(arguments))

"""Shared durable worker process-family lifecycle."""

from __future__ import annotations

import argparse
import ctypes
import fcntl
import json
import os
import signal
import struct
import subprocess
import sys
import tempfile
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Callable


STATE_VERSION = 2
UNSUPPORTED = "UNSUPPORTED_PROCESS_FAMILY_SEMANTICS"
FAMILY_SEMANTICS_UNSUPPORTED = "unsupported"
TERMINAL = {"stopped", "exited"}
TRANSITIONS = {
    "launching": {"running", "stopping", "exited"},
    "running": {"stopping", "exited"},
    "stopping": {"stopped", "exited"},
    "stopped": set(),
    "exited": set(),
}
PROC_PIDTBSDINFO = 3
PROC_PIDVNODEPATHINFO = 9
BSD_SIZE = 136
VNODE_SIZE = 2352
VNODE_CWD_OFFSET = 152
PID_MAX = 2**31 - 1
UINT64_MAX = 2**64 - 1

Fail = Callable[[str], int]
Parse = Callable[[str], tuple]
Emit = Callable[[dict[str, Any], str, Any], None]


def resolved_directory(value: str) -> Path:
    path = Path(value).resolve(strict=True)
    if not path.is_dir():
        raise argparse.ArgumentTypeError("must be an existing directory")
    return path


def is_within(path: Path, directory: Path) -> bool:
    return path == directory or directory in path.parents


def _control_checkout_refusal(cwd: Path, control_checkout: Path) -> str | None:
    if is_within(cwd, control_checkout):
        return "workspace-write refuses the control checkout"
    return None


@contextmanager
def state_lock(path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    lock = path.with_name(path.name + ".lock")
    with lock.open("a+") as handle:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        yield


def atomic_write(path: Path, state: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as file:
            json.dump(state, file, indent=2, sort_keys=True)
            file.write("\n")
            file.flush()
            os.fsync(file.fileno())
        os.replace(temporary, path)
        parent = os.open(path.parent, os.O_RDONLY)
        try:
            os.fsync(parent)
        finally:
            os.close(parent)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def read_state(path: Path, *, family_required: bool = False) -> dict[str, Any] | None:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    if not isinstance(value, dict):
        return None
    if family_required and value.get("version") != STATE_VERSION:
        return None
    return value


def transition(path: Path, state: dict[str, Any], lifecycle: str) -> dict[str, Any]:
    previous = state.get("lifecycle")
    if previous not in TRANSITIONS or lifecycle not in TRANSITIONS[previous]:
        raise ValueError(f"illegal lifecycle transition {previous!r} to {lifecycle!r}")
    updated = dict(state)
    updated["lifecycle"] = lifecycle
    atomic_write(path, updated)
    return updated


def _bounded_integer(value: Any, minimum: int, maximum: int) -> bool:
    return type(value) is int and minimum <= value <= maximum


def _canonical_path(value: Any) -> bool:
    if not isinstance(value, str) or not value or not Path(value).is_absolute():
        return False
    try:
        return str(Path(value).resolve()) == value
    except (OSError, RuntimeError, ValueError):
        return False


BASE_STATE_FIELDS = {"version", "lifecycle", "session_id", "model", "sandbox", "cwd"}


def _valid_common_schema(
    state: dict[str, Any], allowed: set[str], *, effort_levels: set[str]
) -> bool:
    if not BASE_STATE_FIELDS.issubset(state):
        return False
    if not set(state).issubset(allowed):
        return False
    if type(state["version"]) is not int or state["version"] != STATE_VERSION:
        return False
    if not isinstance(state["lifecycle"], str) or state["lifecycle"] not in TRANSITIONS:
        return False
    if not isinstance(state["session_id"], str):
        return False
    if not isinstance(state["model"], str) or not state["model"]:
        return False
    if not isinstance(state["sandbox"], str) or state["sandbox"] not in {"read-only", "workspace-write"}:
        return False
    if not _canonical_path(state["cwd"]):
        return False
    if "effort" in state and (not isinstance(state["effort"], str) or state["effort"] not in effort_levels):
        return False
    if "network" in state and type(state["network"]) is not bool:
        return False
    if "fast" in state and type(state["fast"]) is not bool:
        return False
    control = state.get("control_checkout")
    if control is not None and not _canonical_path(control):
        return False
    if state["sandbox"] == "workspace-write":
        if control is None or is_within(Path(state["cwd"]), Path(control)):
            return False
    return True


def valid_family_schema(state: dict[str, Any], *, effort_levels: set[str]) -> bool:
    identity = {"pid", "pgid", "sid", "birth"}
    if not identity.issubset(state):
        return False
    allowed = BASE_STATE_FIELDS | identity | {"control_checkout", "effort", "network", "fast"}
    if not _valid_common_schema(state, allowed, effort_levels=effort_levels):
        return False
    for field in ("pid", "pgid", "sid"):
        if not _bounded_integer(state[field], 1, PID_MAX):
            return False
    birth = state["birth"]
    if not isinstance(birth, dict) or set(birth) != {"seconds", "microseconds"}:
        return False
    if not _bounded_integer(birth["seconds"], 1, UINT64_MAX):
        return False
    if not _bounded_integer(birth["microseconds"], 0, 999_999):
        return False
    return True


def valid_portable_schema(state: dict[str, Any], *, effort_levels: set[str]) -> bool:
    allowed = BASE_STATE_FIELDS | {"control_checkout", "family_semantics", "generation", "effort", "network", "fast"}
    if not _valid_common_schema(state, allowed, effort_levels=effort_levels):
        return False
    if state["lifecycle"] != "exited":
        return False
    if state.get("family_semantics") != FAMILY_SEMANTICS_UNSUPPORTED:
        return False
    if not _bounded_integer(state.get("generation"), 1, UINT64_MAX):
        return False
    return True


def _libproc() -> ctypes.CDLL | None:
    if sys.platform != "darwin":
        return None
    try:
        library = ctypes.CDLL("/usr/lib/libproc.dylib", use_errno=True)
        library.proc_pidinfo.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_uint64, ctypes.c_void_p, ctypes.c_int]
        library.proc_pidinfo.restype = ctypes.c_int
        library.proc_listpgrppids.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_int]
        library.proc_listpgrppids.restype = ctypes.c_int
        return library
    except OSError:
        return None


def live_identity(pid: int) -> dict[str, Any] | None:
    library = _libproc()
    if library is None:
        return None
    bsd = ctypes.create_string_buffer(BSD_SIZE)
    ctypes.set_errno(0)
    if library.proc_pidinfo(pid, PROC_PIDTBSDINFO, 0, bsd, BSD_SIZE) != BSD_SIZE:
        return None
    cwd = ctypes.create_string_buffer(VNODE_SIZE)
    ctypes.set_errno(0)
    if library.proc_pidinfo(pid, PROC_PIDVNODEPATHINFO, 0, cwd, VNODE_SIZE) != VNODE_SIZE:
        return None
    try:
        returned_pid = struct.unpack_from("<I", bsd.raw, 12)[0]
        pgid = struct.unpack_from("<I", bsd.raw, 100)[0]
        seconds, microseconds = struct.unpack_from("<QQ", bsd.raw, 120)
        sid = os.getsid(pid)
        directory = cwd.raw[VNODE_CWD_OFFSET:].split(b"\0", 1)[0].decode("utf-8")
        canonical_cwd = str(Path(directory).resolve(strict=True))
    except (OSError, UnicodeDecodeError, struct.error):
        return None
    if returned_pid != pid or not directory:
        return None
    return {
        "pid": pid, "pgid": pgid, "sid": sid, "cwd": canonical_cwd,
        "birth": {"seconds": seconds, "microseconds": microseconds},
    }


def group_members(pgid: int) -> list[int] | None:
    library = _libproc()
    if library is None:
        return None
    capacity = 64
    while capacity <= 65536:
        values = (ctypes.c_int * capacity)()
        ctypes.set_errno(0)
        count = library.proc_listpgrppids(pgid, values, ctypes.sizeof(values))
        if count < 0 or (count == 0 and ctypes.get_errno()):
            return None
        members = list(values[:count])
        if count < capacity:
            return members
        capacity *= 2
    return None


def gated_process(
    command: list[str], cwd: Path, *, stdin_text: str | None
) -> tuple[subprocess.Popen[str], int]:
    """Exec a session-leading gate wrapper; Popen can return before the real exec."""
    gate_read, gate_write = os.pipe()
    wrapper = (
        "import json,os,sys; "
        "os.setsid(); os.chdir(sys.argv[2]); "
        "os.read(int(sys.argv[1]), 1) == b'R' or os._exit(125); "
        "os.execvp(json.loads(sys.argv[3])[0], json.loads(sys.argv[3]))"
    )
    process = subprocess.Popen(
        [sys.executable, "-c", wrapper, str(gate_read), str(cwd), json.dumps(command)],
        text=True,
        stdin=subprocess.PIPE if stdin_text is not None else subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        pass_fds=(gate_read,),
    )
    os.close(gate_read)
    return process, gate_write


def establish_family(
    args: argparse.Namespace,
    command: list[str],
    state: dict[str, Any],
    *,
    fail: Fail,
    stdin_text: str | None,
) -> tuple[subprocess.Popen[str] | None, int | None]:
    """Persist and release one gated family while the caller holds the state lock."""
    process, write_fd = gated_process(command, Path(state["cwd"]), stdin_text=stdin_text)
    pid = process.pid
    try:
        identity = None
        deadline = time.monotonic() + 1
        while time.monotonic() < deadline:
            identity = live_identity(pid)
            if identity and identity["pgid"] == pid and identity["sid"] == pid and identity["cwd"] == state["cwd"]:
                break
            time.sleep(0.01)
        if identity is None or identity["pgid"] != pid or identity["sid"] != pid or identity["cwd"] != state["cwd"]:
            os.kill(pid, signal.SIGKILL)
            process.communicate()
            return None, fail(
                f"could not establish dedicated worker process family: observed={identity!r} expected_cwd={state['cwd']!r}"
            )
        state = {**state, **identity}
        atomic_write(args.state, state)
        os.write(write_fd, b"R")
        transition(args.state, state, "running")
    finally:
        os.close(write_fd)
    return process, None


def finish_lifecycle(
    args: argparse.Namespace,
    process: subprocess.Popen[str],
    *,
    parse: Parse,
    emit: Emit,
    fail: Fail,
    stdin_text: str | None,
) -> int:
    stdout, stderr = process.communicate(input=stdin_text) if stdin_text is not None else process.communicate()
    returncode = process.returncode
    with state_lock(args.state):
        current = read_state(args.state, family_required=True)
        if current and current.get("lifecycle") in {"launching", "running"}:
            state = transition(args.state, current, "running") if current["lifecycle"] == "launching" else current
            transition(args.state, state, "exited")
    if returncode:
        sys.stdout.write(stdout)
        sys.stderr.write(stderr)
        return returncode
    session_id, message, metadata, error = parse(stdout)
    if error:
        return fail(error)
    with state_lock(args.state):
        current = read_state(args.state, family_required=True)
        if current is None:
            return fail("state file was lost during worker execution")
        current["session_id"] = session_id
        atomic_write(args.state, current)
    emit(current, message or "", metadata)
    return 0


def run_portable(
    args: argparse.Namespace,
    command: list[str],
    state: dict[str, Any],
    generation: int,
    *,
    parse: Parse,
    emit: Emit,
    fail: Fail,
    stdin_text: str | None,
) -> int:
    """Run under the state lock and persist no recoverable process-family claim."""
    run_arguments: dict[str, Any] = {
        "cwd": Path(state["cwd"]),
        "text": True,
        "stdout": subprocess.PIPE,
        "stderr": subprocess.PIPE,
        "check": False,
    }
    if stdin_text is None:
        run_arguments["stdin"] = subprocess.DEVNULL
    else:
        run_arguments["input"] = stdin_text
    result = subprocess.run(command, **run_arguments)
    terminal = {
        **state,
        "lifecycle": "exited",
        "family_semantics": FAMILY_SEMANTICS_UNSUPPORTED,
        "generation": generation,
    }
    if result.returncode:
        atomic_write(args.state, terminal)
        sys.stdout.write(result.stdout)
        sys.stderr.write(result.stderr)
        return result.returncode
    session_id, message, metadata, error = parse(result.stdout)
    if session_id is not None:
        terminal["session_id"] = session_id
    atomic_write(args.state, terminal)
    if error is not None:
        return fail(error)
    emit(terminal, message or "", metadata)
    return 0


def run_lifecycle(
    args: argparse.Namespace,
    command: list[str],
    state: dict[str, Any],
    *,
    expected: dict[str, Any] | None = None,
    parse: Parse,
    emit: Emit,
    fail: Fail,
    stdin_text: str | None,
    effort_levels: set[str],
) -> int:
    with state_lock(args.state):
        if expected is not None and read_state(args.state) != expected:
            return fail("resume requires an unchanged terminal worker state")
        if _libproc() is None:
            previous_generation = (
                expected["generation"]
                if expected is not None and valid_portable_schema(expected, effort_levels=effort_levels)
                else 0
            )
            if previous_generation == UINT64_MAX:
                return fail("portable state generation exhausted")
            return run_portable(
                args, command, state, previous_generation + 1,
                parse=parse, emit=emit, fail=fail, stdin_text=stdin_text,
            )
        process, error = establish_family(
            args, command, state, fail=fail, stdin_text=stdin_text
        )
        if error is not None:
            return error
    assert process is not None
    return finish_lifecycle(
        args, process, parse=parse, emit=emit, fail=fail, stdin_text=stdin_text
    )


def prepare_start(
    args: argparse.Namespace, *, effort_levels: set[str], default_effort: str
) -> tuple[dict[str, Any] | None, str | None]:
    if args.sandbox == "workspace-write":
        if args.control_checkout is None:
            return None, "workspace-write requires --control-checkout"
        error = _control_checkout_refusal(args.cwd, args.control_checkout)
        if error:
            return None, error
    effort = getattr(args, "effort", default_effort)
    if effort not in effort_levels:
        return None, f"--effort must be one of {sorted(effort_levels)}"
    state: dict[str, Any] = {
        "version": STATE_VERSION,
        "lifecycle": "launching",
        "model": args.model,
        "sandbox": args.sandbox,
        "cwd": str(args.cwd),
        "session_id": "",
        "network": bool(getattr(args, "network", False)),
    }
    if getattr(args, "fast", False):
        state["fast"] = True
    if effort != default_effort:
        state["effort"] = effort
    if args.control_checkout:
        state["control_checkout"] = str(args.control_checkout)
    return state, None


def prepare_resume(
    args: argparse.Namespace, *, effort_levels: set[str], default_effort: str
) -> tuple[dict[str, Any] | None, dict[str, Any] | None, str | None]:
    snapshot = read_state(args.state)
    with state_lock(args.state):
        state = read_state(args.state)
        if state != snapshot:
            return None, None, "resume requires an unchanged terminal worker state"
        if state is None:
            return None, None, "state file is malformed or incomplete"
        if "version" not in state:
            legacy = {"session_id", "model", "sandbox", "cwd"}
            if not legacy.issubset(state) or not set(state).issubset(legacy | {"control_checkout", "effort", "network", "fast"}):
                return None, None, "state file is malformed or incomplete"
            if not all(isinstance(state[key], str) and state[key] for key in legacy):
                return None, None, "state file is malformed or incomplete"
            if "network" in state and type(state["network"]) is not bool:
                return None, None, "state file is malformed or incomplete"
            if "fast" in state and type(state["fast"]) is not bool:
                return None, None, "state file is malformed or incomplete"
        elif not (
            valid_family_schema(state, effort_levels=effort_levels)
            or valid_portable_schema(state, effort_levels=effort_levels)
        ):
            return None, None, "state file is malformed or incomplete"
        elif state["lifecycle"] not in TERMINAL or not state["session_id"]:
            return None, None, "resume requires a terminal worker state with a session ID"
        try:
            cwd = resolved_directory(state["cwd"])
        except (OSError, argparse.ArgumentTypeError):
            return None, None, "state file has an invalid cwd"
        sandbox = state.get("sandbox")
        if sandbox not in {"read-only", "workspace-write"}:
            return None, None, "state file has an invalid sandbox"
        if sandbox == "workspace-write":
            try:
                control = resolved_directory(state["control_checkout"])
            except (KeyError, OSError, argparse.ArgumentTypeError):
                return None, None, "state file is missing the control checkout"
            error = _control_checkout_refusal(cwd, control)
            if error:
                return None, None, error
        effort = state.get("effort", default_effort)
        if effort not in effort_levels:
            return None, None, "state file has an invalid effort"
        fresh = {
            "version": STATE_VERSION,
            "lifecycle": "launching",
            "session_id": state["session_id"],
            "model": state["model"],
            "sandbox": sandbox,
            "cwd": str(cwd),
            "network": state.get("network", False),
        }
        if state.get("fast", False):
            fresh["fast"] = True
        if effort != default_effort:
            fresh["effort"] = effort
        if sandbox == "workspace-write":
            fresh["control_checkout"] = str(control)
    return fresh, state, None


def family_state(
    path: Path, expected: Path, *, effort_levels: set[str]
) -> tuple[dict[str, Any] | None, str | None]:
    state = read_state(path)
    if state is None:
        return None, "state is missing, corrupt, or legacy"
    version = state.get("version")
    if "version" not in state or (type(version) is int and version != STATE_VERSION):
        return None, "state is missing, corrupt, or legacy"
    if not (
        valid_family_schema(state, effort_levels=effort_levels)
        or valid_portable_schema(state, effort_levels=effort_levels)
    ):
        return None, "state is malformed"
    if state["cwd"] != str(expected):
        return None, f"cwd mismatch: recorded={state['cwd']!r} expected={str(expected)!r}"
    return state, None


def matching_leader(state: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None]:
    observed = live_identity(state["pid"])
    if observed is None:
        return None, "leader identity probe failed"
    recorded = {key: state[key] for key in ("pid", "pgid", "sid", "cwd", "birth")}
    if observed != recorded:
        return None, f"identity mismatch: recorded={recorded!r} observed={observed!r}"
    return observed, None


def verify_worker(
    args: argparse.Namespace, *, effort_levels: set[str]
) -> tuple[int | None, str | None]:
    with state_lock(args.state):
        state, error = family_state(args.state, args.cwd, effort_levels=effort_levels)
        if error:
            return None, error
        if _libproc() is None:
            return None, UNSUPPORTED
        assert state is not None
        if valid_portable_schema(state, effort_levels=effort_levels):
            return None, "state has no recoverable process family"
        members = group_members(state["pgid"])
        if members is None:
            return None, "process-group probe failed"
        if members:
            return None, f"worker process group still has members: {members}"
        if state["lifecycle"] not in TERMINAL:
            if state["lifecycle"] != "stopping":
                state = transition(args.state, state, "stopping")
            transition(args.state, state, "stopped")
    return 0, None


def stop_worker(
    args: argparse.Namespace, *, effort_levels: set[str]
) -> tuple[int | None, str | None]:
    with state_lock(args.state):
        state, error = family_state(args.state, args.cwd, effort_levels=effort_levels)
        if error:
            return None, error
        if _libproc() is None:
            return None, UNSUPPORTED
        assert state is not None
        if valid_portable_schema(state, effort_levels=effort_levels):
            return None, "state has no recoverable process family"
        _, error = matching_leader(state)
        if error and state["lifecycle"] not in TERMINAL:
            if "identity mismatch:" in error:
                return None, error
            members = group_members(state["pgid"])
            if members is None:
                return None, "process-group probe failed"
            if not members:
                if state["lifecycle"] != "stopping":
                    state = transition(args.state, state, "stopping")
                transition(args.state, state, "stopped")
                return 0, None
            state = transition(args.state, state, "stopping") if state["lifecycle"] != "stopping" else state
            try:
                os.killpg(state["pgid"], signal.SIGKILL)
            except OSError as exc:
                return None, f"KILL refused: {exc}"
            return 0, None
        members = group_members(state["pgid"])
        if members is None:
            return None, "process-group probe failed"
        if not members:
            if state["lifecycle"] not in TERMINAL:
                if state["lifecycle"] != "stopping":
                    state = transition(args.state, state, "stopping")
                transition(args.state, state, "stopped")
            return 0, None
        if state["lifecycle"] in TERMINAL:
            try:
                os.killpg(state["pgid"], signal.SIGKILL)
            except OSError as exc:
                return None, f"KILL refused: {exc}"
            deadline = time.monotonic() + args.grace_seconds
            while time.monotonic() < deadline:
                members = group_members(state["pgid"])
                if members is None:
                    return None, "process-group probe failed"
                if not members:
                    return verify_worker(args, effort_levels=effort_levels)
                time.sleep(0.05)
            return None, f"worker process group still has members: {members}"
        if state["lifecycle"] != "stopping":
            state = transition(args.state, state, "stopping")
        try:
            os.killpg(state["pgid"], signal.SIGTERM)
        except OSError as exc:
            return None, f"TERM refused: {exc}"
    deadline = time.monotonic() + args.grace_seconds
    while time.monotonic() < deadline:
        members = group_members(state["pgid"])
        if members is None:
            return None, "process-group probe failed"
        if not members:
            with state_lock(args.state):
                current = read_state(args.state, family_required=True)
                if current and current["lifecycle"] == "stopping":
                    transition(args.state, current, "stopped")
            return 0, None
        time.sleep(0.05)
    members = group_members(state["pgid"])
    if members is None:
        return None, "process-group probe failed"
    if not members:
        return 0, None
    try:
        os.killpg(state["pgid"], signal.SIGKILL)
    except OSError as exc:
        return None, f"KILL refused: {exc}"
    deadline = time.monotonic() + args.grace_seconds
    while time.monotonic() < deadline:
        members = group_members(state["pgid"])
        if members is None:
            return None, "process-group probe failed"
        if not members:
            break
        time.sleep(0.05)
    return verify_worker(args, effort_levels=effort_levels)