orchestrator.py

"""orchestrator.py — The Owned Spine (Orchestrator v1).

The acquisition ladder (ladder.py) pointed at LABOR instead of KNOWLEDGE. Where the ladder
climbs a Null cheapest-first and writes back the answer, the orchestrator routes a Task to a
worker, gates the returned work, and escalates on failure — same skeleton, same seam law:

    ladder:        Null  -> climb   -> verify -> write_back
    orchestrator:  Task  -> route   -> dispatch -> gate -> escalate -> log

Design: orchestrator-v1-design.md (absorbs dispatch-v1-design.md). It is CODE, owned — no
resident model, no rented LLM at the wheel (New Brain facts 69/71/75). The local qwen judge sits
ONLY in the judgment seams: qwen recommends, code decides, Aaron approves; qwen abstains climb to
Aaron (fact 76). Adding a model/provider/task mapping is a roster.json data edit with ZERO code
change (fact 91) — transport adapters are generic per PROVIDER protocol, never per model.

STDOUT LAW: diagnostics to stderr (this module can touch the New Brain store / MCP transport).

Run:  .venv/Scripts/python.exe orchestrator.py        # demo: watch dispatch/, ground the queue
"""

from __future__ import annotations

import json
import os
import shlex
import shutil
import subprocess
import sys
import threading
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import nullcontext
from dataclasses import dataclass, field, replace
from pathlib import Path

HERE = Path(__file__).resolve().parent
DISPATCH_DIR = HERE / "dispatch"
ROSTER = HERE / "roster.json"
DISPATCH_LOG = HERE / "dispatch_log.jsonl"
# Worker LIVENESS log (task 32). dispatch_log.jsonl records OUTCOMES only (done/escalated/returned) —
# it says nothing while a worker is mid-flight. The swarm HUD needs to show what is RUNNING right now,
# so the spine appends lifecycle events here as an ADDITIVE side-channel: 'running' lands BEFORE the
# synchronous dispatch blocks (inside null_gate), and every running reaches a terminal 'finished' (or
# 'requeued' on a rate limit) in run_task. It sits beside its dispatch_log: derived from log_path so a
# test's temp log never writes into the repo.
WORKER_STATUS = HERE / "worker_status.jsonl"


def _status_path_for(log_path: Path) -> Path:
    """The liveness log that rides beside a given dispatch_log (task 32). The HUD reads the repo's
    worker_status.jsonl; tests pass a temp log_path and this keeps their liveness writes in the temp
    dir — the status side-channel is additive and never touches the repo from a test."""
    return WORKER_STATUS if Path(log_path) == DISPATCH_LOG else Path(log_path).with_name("worker_status.jsonl")


def write_status(task: Task, phase: str, octx: OCtx, log_path: Path = DISPATCH_LOG) -> None:
    """Append ONE worker-lifecycle event so the HUD can show live/finished workers (task 32). phase is
    'running' (written BEFORE the synchronous dispatch call blocks — the worker is about to launch, so
    the HUD shows it live for the whole model run), 'finished' (a terminal gate/escalation outcome
    landed), or 'requeued' (subscription rate limit — the task went back to the queue; NOT live, so the
    HUD must not ring it). Every 'running' reaches a terminal phase on every defined run_task path.
    Additive-only: the spine's routing/gating/logging is unchanged; this is a pure observability
    side-write, guarded by the same log_lock as dispatch_log so parallel fan-out appends never tear.
    Best-effort — a status-write failure never derails a real dispatch (observability must not break
    the spine)."""
    ran = task.roster_entry or {}
    rec = {"ts": octx.now, "task_id": task.id, "type": task.type, "phase": phase,
           "model": ran.get("model"), "provider": ran.get("provider"), "title": task.title}
    try:
        with (octx.log_lock or nullcontext()):
            with _status_path_for(log_path).open("a", encoding="utf-8") as f:
                f.write(json.dumps(rec) + "\n")
    except Exception as e:                                    # observability must never break the spine
        _log(f"  (worker_status write skipped for {task.id}: {e})")

# The five task types dispatch v1 routes (design "Routing table"). web-research is a routing
# TARGET for primarily-research tasks only (fact 82-85) — NOT the gate on who may touch the web;
# every worker carries ladder access as standing equipment. judge/triage are the qwen seam types.
TASK_TYPES = {"design", "implement", "fix", "verify", "sweep", "web-research", "judge", "triage"}


# ---------------------------------------------------------------------------
# the task record — the unit of dispatch (dispatch v1 schema + DeepSpark abort_criteria)
# ---------------------------------------------------------------------------
@dataclass
class Task:
    """One unit of dispatchable labor. receipts + done_means are the GROUNDING FLOOR: a task
    missing either does not dispatch, it returns to its spawner for grounding (dispatch v1 law).
    abort_criteria (DeepSpark) are the mid-task stop conditions the handoff prompt carries — the
    worker escalates the partial the moment confidence is lost, never finishes a doomed artifact."""
    id: str
    title: str
    type: str
    source: str                                      # what spawned it (flag, abstain, audit, Aaron)
    receipts: list[str] = field(default_factory=list)      # evidence the task is real (file:line, log ref)
    done_means: list[str] = field(default_factory=list)    # explicit checkable completion criteria
    gates: list[str] = field(default_factory=list)         # mechanical checks that verdict the work
    abort_criteria: list[str] = field(default_factory=list)  # stop-mid-task-and-escalate conditions
    context: str = ""                                # free-text order-of-work detail for the handoff prompt
    read_first: list[str] = field(default_factory=list)    # context pointers (paths/refs) the worker reads first
    hold: str = ""                                   # policy hold (e.g. fact-9 eval gate): held tasks never dispatch
    status: str = "queued"                           # queued|dispatched|gated|escalated|done|returned|held
    attempts: list[dict] = field(default_factory=list)     # [{roster_entry, outcome, gate_results, notes}]
    roster_entry: dict | None = None                 # assigned by the router (fills at route time)

    @property
    def grounded(self) -> bool:
        """A task is dispatchable exactly when it carries BOTH receipts and done_means. Type must
        also be one the router knows — an unknown type is a malformed record, not a routing null."""
        return bool(self.receipts) and bool(self.done_means) and self.type in TASK_TYPES

    def grounding_gaps(self) -> list[str]:
        """The specific reasons a task is not grounded — carried back to the spawner verbatim so the
        return is actionable, never a bare 'rejected'."""
        gaps = []
        if not self.receipts:
            gaps.append("no receipts (evidence the task is real: file:line, capture ref, log entry)")
        if not self.done_means:
            gaps.append("no done_means (explicit checkable completion criteria)")
        if self.type not in TASK_TYPES:
            gaps.append(f"unknown type {self.type!r} (known: {sorted(TASK_TYPES)})")
        return gaps


# ---------------------------------------------------------------------------
# queue — task records live as JSON files under dispatch/
# ---------------------------------------------------------------------------
def task_from_dict(d: dict) -> Task:
    """Build a Task from a record dict, tolerating extra keys (forward-compat) and missing optional
    lists. Required keys id/title/type/source raise if absent — a record without them is not a task."""
    known = {f for f in Task.__dataclass_fields__}
    return Task(**{k: v for k, v in d.items() if k in known})


def load_queue(dispatch_dir: Path = DISPATCH_DIR) -> list[Task]:
    """Read every *.json task record under dispatch/ (sorted by filename for stable order). A record
    that fails to parse is reported to stderr and skipped — one bad file never blocks the queue."""
    if not dispatch_dir.exists():
        return []
    tasks = []
    for path in sorted(dispatch_dir.glob("*.json")):
        try:
            tasks.append(task_from_dict(json.loads(path.read_text(encoding="utf-8"))))
        except (json.JSONDecodeError, TypeError, ValueError) as e:
            _log(f"  SKIP malformed task record {path.name}: {e}")
    return tasks


def ground(tasks: list[Task]) -> tuple[list[Task], list[Task]]:
    """Split the queue: (dispatchable, returned-for-grounding). The grounding check runs BEFORE any
    routing or dispatch — an ungrounded task never reaches a worker (dispatch v1 law, kept)."""
    dispatchable, returned = [], []
    for t in tasks:
        if t.grounded:
            dispatchable.append(t)
        else:
            t.status = "returned"
            returned.append(t)
    return dispatchable, returned


def _log(msg: str) -> None:
    print(msg, file=sys.stderr, flush=True)


# ---------------------------------------------------------------------------
# router — type -> roster entry, ENTIRELY data-driven (fact 91 zero-code-alteration)
# ---------------------------------------------------------------------------
def load_roster(path: Path = ROSTER) -> list[dict]:
    """Read the roster entries. Empty if the file is absent — an empty roster routes everything to
    None (every task becomes a routing null for Aaron), never a crash."""
    if not path.exists():
        return []
    return json.loads(path.read_text(encoding="utf-8")).get("entries", [])


def route(task: Task, roster: list[dict]) -> dict | None:
    """Return the first roster entry whose 'tasks' list claims task.type. None when nothing is
    rostered for the type — a ROUTING NULL, which surfaces to Aaron via the null-gate (later step),
    NOT a silent drop and NOT a code-side fallback. The type->model mapping lives ONLY in the data:
    this function hardcodes no model, no provider, no type. Adding a mapping is a roster.json edit
    (fact 91). Order is the tie-break — first capable entry wins (cheapest-capable-first is a data
    ordering choice, per the roster comment)."""
    for entry in roster:
        if task.type in entry.get("tasks", []):
            return entry
    return None


# ---------------------------------------------------------------------------
# transport — SUBSCRIPTION-FIRST (facts 109-111, 113, 115). No paid API by default.
# ---------------------------------------------------------------------------
# Primary transport = subscription-authenticated provider CLIs; the command line is roster.json DATA,
# so a new CLI provider (Gemini, Codex) is a data-only add (facts 110/91). Hard money rules: (a) no
# paid API billing by default — a subscription rate limit makes tasks QUEUE AND WAIT, never spill
# (fact 111); (b) ANY use of a billing:'api' transport FLAGS AARON and asks permission first
# (facts 109/115). The subscription path runs with ANTHROPIC_API_KEY scrubbed from the subprocess
# env, so no subprocess can silently bill the paid API — it uses the subscription token or fails loud.
SUBSCRIPTION, FREE, API = "subscription", "free", "api"
RATE_LIMIT_MARKERS = ("rate limit", "rate-limit", "usage limit", "limit reached", "limit exceeded",
                      "too many requests", "quota", "429", "overloaded")
DEFAULT_CLAUDE_CMD = ["claude", "-p", "--model", "{model}", "--output-format", "json", "{prompt}"]


@dataclass
class WorkerResult:
    ok: bool
    text: str
    cost: dict = field(default_factory=dict)         # per-provider cost fields (feeds dispatch_log)
    note: str = ""
    rate_limited: bool = False                       # subscription limit hit -> QUEUE AND WAIT (fact 111), no spill


def _ollama_generate(model: str, prompt: str, *, num_predict: int = 512, timeout: int = 300) -> str:
    """Spine's Ollama primitive. checker._ollama_generate is capped at 12 tokens (one-word judge
    verdicts); the seams here need room for a drafted recommendation, so num_predict is a param.
    temperature 0 — the judgment seams are deterministic, not creative."""
    payload = json.dumps({"model": model, "prompt": prompt, "stream": False,
                          "options": {"temperature": 0, "num_predict": num_predict}}).encode()
    req = urllib.request.Request("http://localhost:11434/api/generate", data=payload,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read())["response"]


def _default_runner(cmd: list[str], timeout: int, env: dict | None = None) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env)


def _subscription_env() -> dict:
    """Env for a subscription CLI dispatch with the paid-API credentials SCRUBBED. Load-bearing guard
    behind facts 109/115: even if an API key sits in the ambient env, the subprocess cannot silently
    bill the paid API — it must use the subscription token (claude setup-token) or fail loud with
    'Not logged in'. No silent API, ever."""
    env = dict(os.environ)
    env.pop("ANTHROPIC_API_KEY", None)
    env.pop("ANTHROPIC_AUTH_TOKEN", None)
    return env


def _resolve_cmd(entry: dict, prompt: str) -> list[str]:
    """Build argv from the roster entry's cmd TEMPLATE (data, not code — facts 110/91). {model} and
    {prompt} are substituted; argv[0] is resolved to a full path (Windows .cmd shims that CreateProcess
    can't find by bare name; shutil.which honors PATHEXT)."""
    template = entry.get("cmd") or DEFAULT_CLAUDE_CMD
    exe = shutil.which(template[0]) or template[0]
    def sub(a: str) -> str:
        return a.replace("{model}", entry.get("model", "")).replace("{prompt}", prompt)
    return [exe] + [sub(a) for a in template[1:]]


def _is_rate_limit(text: str) -> bool:
    return any(m in (text or "").lower() for m in RATE_LIMIT_MARKERS)


def _parse_cli(proc: subprocess.CompletedProcess, entry: dict) -> WorkerResult:
    """Parse a `claude -p --output-format json` result. Structured payload is present even on failure
    (rc 1 + is_error + the reason in `result`), so parse stdout FIRST — the autopsy carries the true
    reason. A rate-limit response sets rate_limited so the spine QUEUES AND WAITS (fact 111) rather
    than failing/escalating/spilling."""
    cost = {"provider": entry.get("provider"), "model": entry.get("model"),
            "billing": entry.get("billing", SUBSCRIPTION)}
    payload = None
    try:
        payload = json.loads(proc.stdout) if proc.stdout.strip() else None
    except json.JSONDecodeError:
        payload = None
    if payload is not None:
        text = payload.get("result", proc.stdout)
        cost |= {"dollars": payload.get("total_cost_usd"), "usage": payload.get("usage")}
        if payload.get("is_error") and _is_rate_limit(text):
            return WorkerResult(False, text, cost, note=f"rate-limited: {text[:160]}", rate_limited=True)
        ok = proc.returncode == 0 and not payload.get("is_error")
        return WorkerResult(ok, text, cost, note="" if ok else f"worker error: {text[:200]}")
    raw = proc.stdout or proc.stderr or ""
    if _is_rate_limit(raw):
        return WorkerResult(False, raw, cost, note=f"rate-limited: {raw[:160]}", rate_limited=True)
    ok = proc.returncode == 0
    return WorkerResult(ok, raw, cost, note="" if ok else f"cli exit {proc.returncode}: {raw[:200]}")


def cli_dispatch(entry: dict, prompt: str, *, runner=None, timeout: int = 1800) -> WorkerResult:
    """Generic subscription/free CLI transport. Command line from roster DATA (facts 110/91). The
    subscription path runs with the API key scrubbed (subscription-only, no silent API). `runner` is
    an injectable seam (default subprocess.run) — tests stub it so no worker launches and no spend
    happens in the test path. billing:'api' never reaches here — dispatch() gates it first."""
    cmd = _resolve_cmd(entry, prompt)
    if runner is not None:
        proc = runner(cmd, timeout)                          # stubbed path: (cmd, timeout), env irrelevant
    else:
        env = _subscription_env() if entry.get("billing", SUBSCRIPTION) == SUBSCRIPTION else None
        proc = _default_runner(cmd, timeout, env)
    return _parse_cli(proc, entry)


def ollama_api(entry: dict, prompt: str, *, num_predict: int = 512, **_) -> WorkerResult:
    """Local model over the Ollama HTTP API (the judge/triage seam already serving in Ollama). Free,
    owned — the cost field records $0 so the log's per-provider accounting is complete, not blank."""
    text = _ollama_generate(entry["model"], prompt, num_predict=num_predict)
    return WorkerResult(True, text, {"provider": "ollama", "model": entry["model"],
                                     "billing": FREE, "dollars": 0.0})


def dispatch(entry: dict, prompt: str, *, runner=None, approve_api=None) -> WorkerResult:
    """THE single transport entry point — selects by provider/billing and enforces the money rules:
      - ollama            -> local HTTP, free.
      - billing == 'api'  -> FLAG AARON (approve_api seam); proceed ONLY on an explicit yes
                             (facts 109/115). No API client is wired in v1, so even a granted yes
                             returns a defined seam-only note — nothing bills silently, ever.
      - else (subscription/free CLI) -> cli_dispatch, API key scrubbed on the subscription path.
    """
    if entry.get("provider") == "ollama":
        return ollama_api(entry, prompt)
    if entry.get("billing") == API:
        granted = bool(approve_api(entry)) if callable(approve_api) else False
        note = ("API permitted by Aaron, but no API client is wired in v1 (seam only, fact 90)"
                if granted else
                "API transport — needs Aaron's explicit permission first (facts 109/115); not granted")
        return WorkerResult(False, "", {"provider": entry.get("provider"), "billing": API}, note=note)
    return cli_dispatch(entry, prompt, runner=runner)


# ---------------------------------------------------------------------------
# seam context — where the ONLY model-in-the-loop (qwen) and the yes-gate live
# ---------------------------------------------------------------------------
@dataclass
class OCtx:
    """The orchestrator's injectable seams (mirrors ladder.Ctx). Every place a model or Aaron enters
    the loop is a seam here, defaulting to the SAFE behavior: no auto-dispatch, no fabricated
    recommendation. Tests inject stubs; the live session injects Aaron's real yes."""
    roster: list[dict] = field(default_factory=load_roster)
    approve: object = None            # (ask: DispatchAsk) -> bool  — THE YES-GATE. None -> auto-DENY (safe default)
    approve_api: object = None        # (entry) -> bool  — the API-PERMISSION gate (facts 109/115). None -> auto-DENY
    recommender: object = None        # (task, entry) -> dict       — qwen seam. None -> code-only rationale, no model
    store: object = None              # NewBrainStore | None
    runner: object = None             # subprocess runner seam for headless dispatch (tests stub; None -> real)
    gate_runner: object = None        # subprocess runner seam for command gates (tests stub; None -> real)
    verify: object = None             # (claim) -> bool|None  — null-checker seam (rift-verify); None -> skipped
    diff: object = None               # (paths) -> bool|None  — diff-scope seam; None -> skipped
    gate_judge: object = None         # (name, detail) -> bool — qwen interprets AMBIGUOUS gate results; None -> climbs
    log_lock: object = None           # threading.Lock guarding the dispatch_log append under fan-out; None -> no lock
    batch_approve: object = None       # (FanoutPlan) -> {task_id: bool} — the BATCHED yes-gate; None -> auto-deny all
    now: str = field(default_factory=lambda: __import__("datetime").datetime.now(
        __import__("datetime").timezone.utc).isoformat())


@dataclass
class DispatchAsk:
    """The decision-grade yes/no the null-gate surfaces to Aaron (facts 19-21, 72): issue, why,
    honest recommendation, pros/cons. A blocking ask interrupts now; a non-blocking one batches to
    the next touchpoint. qwen_abstained means the recommendation is code-only — Aaron decides with
    that flag visible, never a model verdict laundered as certainty."""
    task: Task
    entry: dict | None
    issue: str
    why: str
    recommendation: str
    pros: list[str]
    cons: list[str]
    blocking: bool = False
    qwen_abstained: bool = False

    def render(self) -> str:
        """Decision-grade text (the format Aaron reads). Same shape whether it batches or interrupts."""
        tgt = f"{self.entry['model']} ({self.entry['provider']}/{self.entry['transport']})" \
            if self.entry else "NONE — nothing rostered for this type"
        lines = [f"DISPATCH NULL{' [BLOCKING]' if self.blocking else ' [batched]'}: {self.task.title}",
                 f"  issue: {self.issue}",
                 f"  why:   {self.why}",
                 f"  route: {tgt}",
                 f"  recommendation: {self.recommendation}"
                 + ("  (qwen abstained — code-only rationale)" if self.qwen_abstained else "")]
        for p in self.pros:
            lines.append(f"    + {p}")
        for c in self.cons:
            lines.append(f"    - {c}")
        lines.append(f"  -> yes dispatches to {tgt}; the yes is yours, the labor is shed (fact 72).")
        return "\n".join(lines)


def qwen_recommend(task: Task, entry: dict | None, octx: OCtx) -> dict:
    """qwen PRE-DRAFTS the routing recommendation the yes/no ask carries (seam law: qwen recommends,
    code decides, Aaron approves). It does NOT choose the route — route() already did that in code;
    qwen only drafts the human rationale + pros/cons. An injected recommender overrides (tests). With
    no recommender and no reachable qwen, returns abstained=True — a code-only rationale, never a
    fabricated model verdict (an abstained seam climbs to Aaron like any null, fact 76)."""
    if octx.recommender is not None:
        return octx.recommender(task, entry)
    if entry is None:
        return {"rationale": "no roster entry claims this task type — Aaron's call whether to roster "
                             "one, redesign the task, or handle it in conversation (design tier).",
                "pros": [], "cons": [], "abstained": True}
    prompt = (
        "You are a routing advisor. In ONE sentence, say whether sending this task to this worker is "
        "sound, and give at most two pros and two cons. Be terse.\n"
        f"TASK TYPE: {task.type}\nTASK: {task.title}\n"
        f"WORKER: {entry['model']} (cost {entry.get('cost','?')}, trust {entry.get('trust','?')})\n"
        "Format:\nRATIONALE: <one sentence>\nPRO: <...>\nPRO: <...>\nCON: <...>\nCON: <...>")
    try:
        raw = _ollama_generate(entry["model"] if entry["provider"] == "ollama" else "qwen2.5:7b",
                               prompt, num_predict=200)
    except Exception as e:                                    # qwen unreachable -> abstain, don't fake
        return {"rationale": f"(qwen unreachable: {e}) routing by code table only", "pros": [],
                "cons": [], "abstained": True}
    rationale, pros, cons = "", [], []
    for line in raw.splitlines():
        s = line.strip()
        if s.upper().startswith("RATIONALE:"):
            rationale = s.split(":", 1)[1].strip()
        elif s.upper().startswith("PRO:"):
            pros.append(s.split(":", 1)[1].strip())
        elif s.upper().startswith("CON:"):
            cons.append(s.split(":", 1)[1].strip())
    abstained = not rationale
    return {"rationale": rationale or "(qwen returned no rationale)", "pros": pros, "cons": cons,
            "abstained": abstained}


def build_dispatch_ask(task: Task, octx: OCtx) -> DispatchAsk:
    """Turn a grounded, routed task into the decision-grade yes/no. Blocking iff any abort_criterion
    marks it actively-blocking OR the task itself is flagged blocking via a receipt convention —
    v1 keeps it simple: a task is blocking if it declares gates it cannot self-satisfy (interrupt),
    else batched (facts 19-21)."""
    entry = route(task, octx.roster)
    rec = qwen_recommend(task, entry, octx)
    issue = f"queued {task.type} task needs a worker" if entry else \
            f"queued {task.type} task has NO rostered worker"
    why = f"grounded (receipts + done_means present); router -> " + \
          (f"{entry['model']}" if entry else "unrouted (routing null)")
    return DispatchAsk(task=task, entry=entry, issue=issue, why=why,
                       recommendation=rec["rationale"], pros=rec.get("pros", []),
                       cons=rec.get("cons", []), blocking=bool(task.gates) and entry is None,
                       qwen_abstained=rec.get("abstained", False))


# ---------------------------------------------------------------------------
# handoff prompt — built BY THE ORCHESTRATOR on Aaron's yes (fact 72), Phase 2 template
# ---------------------------------------------------------------------------
# Sections mirror the proven Fable->Opus handoff format (this build's own brief is the exemplar):
# role, context pointers, order of work, hard rules, done-means, decision-block protocol, abort
# criteria. STANDING hard rules apply to EVERY worker; task fields fill the specifics.
STANDING_HARD_RULES = [
    "Ladder access (L0 local -> L1 adjacent -> L2 web) is your standing equipment — any task can "
    "hit a knowledge null mid-flight; climb it, don't guess (facts 82-85).",
    "Surgical changes: touch only what the task names; every changed line traces to the task.",
    "STDOUT LAW where it applies: diagnostics to stderr; stdout stays clean for machine consumers.",
    "Do NOT touch the New Brain extractor — it is hard-gated behind the eval set (fact 9).",
    "Abstain-upward is honored: if the task is beyond your brief, escalate the partial with the "
    "sticking point named — never fake done (the only sin).",
]
DECISION_BLOCK_PROTOCOL = (
    "Decision-block protocol: a genuine decision-null the brief doesn't cover -> surface it "
    "decision-grade (issue, why, honest recommendation, pros/cons) and STOP for that decision; "
    "don't guess a yes or a no.")


def build_handoff_prompt(task: Task, entry: dict) -> str:
    """The worker's prompt, generated from the task on Aaron's yes. This is the artifact fact 72
    says the orchestrator builds ITSELF — Aaron keeps the yes, the spine does the drafting."""
    ptrs = task.read_first or task.receipts or ["(no explicit pointers — start from the task title)"]
    aborts = task.abort_criteria or ["lost confidence the approach is correct — stop and escalate the partial"]
    gates = task.gates or ["(no gates declared — done_means checklist is the bar)"]
    S = []
    S.append(f"You are {entry['model']}, dispatched by the orchestrator for a {task.type} task.")
    S.append(f"TASK: {task.title}")
    if task.context:
        S.append(f"\n{task.context}")
    S.append("\nCONTEXT POINTERS (read first):")
    S += [f"  - {p}" for p in ptrs]
    S.append("\nORDER OF WORK:")
    S += [f"  {i+1}. {d}" for i, d in enumerate(task.done_means)]  # done_means double as the work list
    S.append("\nHARD RULES:")
    S += [f"  - {r}" for r in STANDING_HARD_RULES]
    S.append(f"\nDONE MEANS (each must be explicitly confirmed or explicitly failed):")
    S += [f"  - {d}" for d in task.done_means]
    S.append("\nGATES (mechanical — your work is verdicted by these, not by opinion):")
    S += [f"  - {g}" for g in gates]
    S.append(f"\n{DECISION_BLOCK_PROTOCOL}")
    S.append("\nABORT CRITERIA (stop mid-task and escalate the partial if):")
    S += [f"  - {a}" for a in aborts]
    S.append(f"\nSOURCE: {task.source}  |  TASK ID: {task.id}")
    return "\n".join(S)


def null_gate(task: Task, octx: OCtx, log_path: Path | None = None) -> dict:
    """The fact-72 flow: dispatch-need is a null -> decision-grade ask -> Aaron's yes -> the
    orchestrator BUILDS THE PROMPT and dispatches. Returns a structured outcome. On no/abstain
    NOTHING dispatches — the yes is the thing Aaron keeps. The `approve` seam is the yes-gate: None
    means auto-deny (v1 gates every dispatch; nothing launches without an explicit yes).

    log_path (task 32): when set, a 'running' liveness event is written BEFORE dispatch() blocks —
    dispatch is SYNCHRONOUS, so a write after it returns would mark the worker running only after it
    had already finished (the HUD would miss the entire model run). Only run_task passes it, and
    run_task guarantees the matching terminal event; a direct null_gate call writes no liveness (so
    nothing can strand a 'running' with no terminal)."""
    ask = build_dispatch_ask(task, octx)
    task.roster_entry = ask.entry                             # stamp the route so the log + escalation see it
    approved = bool(octx.approve(ask)) if callable(octx.approve) else False
    if not approved:
        return {"dispatched": False, "ask": ask, "reason": "no yes (batched/denied — nothing launched)"}
    if ask.entry is None:
        return {"dispatched": False, "ask": ask, "reason": "approved but unrouted — cannot dispatch a routing null"}
    prompt = build_handoff_prompt(task, ask.entry)
    task.status = "dispatched"
    if log_path is not None:                                  # LIVENESS: mark running BEFORE the blocking dispatch
        write_status(task, "running", octx, log_path)
    result = dispatch(ask.entry, prompt, runner=octx.runner, approve_api=octx.approve_api)
    return {"dispatched": True, "ask": ask, "prompt": prompt, "entry": ask.entry, "result": result}


# ---------------------------------------------------------------------------
# gate runner — MECHANICAL ONLY (dispatch v1 law). No model's opinion is ever a gate.
# ---------------------------------------------------------------------------
# A gate verdicts the returned work. Clear pass/fail needs no model. An AMBIGUOUS result — a gate
# that RAN but whose outcome isn't a clean boolean ("test output changed shape") — is the ONE place
# the qwen judge seam may interpret (design "judgment seams"); qwen abstaining climbs to Aaron.
GATE_KINDS = {"command", "null-checker", "diff-scope", "done_means"}


def normalize_gate(g) -> dict:
    """A gate spec is a bare string (a shell COMMAND, name == the command) or a dict with a 'name'
    and a kind-specific field. Recognized names map to their kind; anything with a 'cmd' (or a bare
    string) is a command gate."""
    if isinstance(g, str):
        return {"name": g, "kind": "command", "cmd": g}
    name = g.get("name", "")
    if "cmd" in g:
        return {"name": name or g["cmd"], "kind": "command", "cmd": g["cmd"]}
    kind = name if name in GATE_KINDS else "command"
    return {"name": name, "kind": kind, **{k: v for k, v in g.items() if k != "name"}}


def _default_gate_runner(cmd: str, timeout: int) -> subprocess.CompletedProcess:
    """Run a gate command (a shell string in task data, e.g. '.venv/Scripts/python.exe ladder_test.py')
    as an argv list: shlex-parse (strips quotes), resolve argv[0] to an absolute/looked-up path, and
    anchor cwd at the repo. Raw Windows subprocess can't find a relative .exe by forward-slash path
    (WinError 2) and cmd.exe mis-parses forward slashes — resolving argv[0] sidesteps both."""
    parts = shlex.split(cmd, posix=True) if isinstance(cmd, str) else list(cmd)
    if parts:
        exe = parts[0]
        if ("/" in exe) or ("\\" in exe):
            p = Path(exe)
            parts[0] = str(p if p.is_absolute() else (HERE / p))
        else:
            parts[0] = shutil.which(exe) or exe
    return subprocess.run(parts, capture_output=True, text=True, timeout=timeout, cwd=str(HERE))


def _run_command_gate(spec: dict, octx: OCtx) -> tuple[bool | None, str]:
    """rc 0 -> pass, nonzero -> fail. A timeout/spawn error is AMBIGUOUS (None) — the gate could not
    render a verdict, which is not the same as a fail. Uses the gate_runner seam (tests stub it)."""
    runner = octx.gate_runner or _default_gate_runner
    cmd = spec["cmd"] if isinstance(spec["cmd"], str) else " ".join(spec["cmd"])
    try:
        proc = runner(cmd, timeout=1800)
    except Exception as e:
        return None, f"gate could not run: {e}"
    return (proc.returncode == 0, f"exit {proc.returncode}")


def _run_nullchecker_gate(spec: dict, result: WorkerResult, octx: OCtx) -> tuple[bool | None, str]:
    """Every factual claim in the worker's output runs through verify (rift-verify seam). Any flag
    blocks the gate. No verify seam wired -> skipped-with-note (None), never a silent pass."""
    if octx.verify is None:
        return None, "null-checker seam not wired (verify) — skipped, not passed"
    claims = [ln.strip() for ln in (result.text or "").splitlines() if len(ln.strip()) > 12]
    flags = [c for c in claims if octx.verify(c) is False]
    return (not flags, f"{len(flags)} flagged of {len(claims)} claim(s)")


def make_rift_verify():
    """Build the OCtx.verify callable, backed by the rift-verify confident-wrong verifier. Loads the
    heavy typed index + judge ONCE (as verify_server does), then per claim returns:
      False -> at least one extracted fact is CONTRADICTED by source (a flag)
      True  -> at least one is confirmed and none contradicted
      None  -> the claim yields only abstains (can't verify) — skipped, never a silent pass
    Uses rift-verify's OWN typed extraction (runtime_verify), NOT the New Brain extractor — so it is
    clear of the fact-9 gate. The live orchestrator wires this into OCtx.verify; tests stub the seam
    directly (no heavy load), the same way transport and the judge are stubbed."""
    from runtime_verify import verify as _rv, TypedChecker
    from checker import JudgeChecker
    typed, judge = TypedChecker(), JudgeChecker()

    def _verify(claim: str) -> bool | None:
        verdicts = {r["verdict"] for r in _rv(claim, typed, judge)}
        if "flag" in verdicts:
            return False
        if "pass" in verdicts:
            return True
        return None
    return _verify


def _run_diffscope_gate(spec: dict, octx: OCtx) -> tuple[bool | None, str]:
    """The diff touches only the paths the task named. Seam (diff) — None if unwired (aporia-engine
    git state is not assumed). paths come from the gate spec."""
    if octx.diff is None:
        return None, "diff-scope seam not wired (diff) — skipped, not passed"
    ok = octx.diff(spec.get("paths", []))
    return ok, "in-scope" if ok else "out-of-scope diff"


def run_gates(task: Task, result: WorkerResult, octx: OCtx) -> dict:
    """Run each gate, resolve AMBIGUOUS results through the qwen interpret-seam (abstain climbs to
    Aaron), then the done_means checklist. Returns {passed: bool, gate_results: [...], climbs: [...]}.
    A failed WORKER result (ok=False) fails the gate set outright — a doomed/aborted artifact is
    never gated as if complete (DeepSpark abort semantics)."""
    gate_results, climbs = [], []
    if not result.ok:
        return {"passed": False, "gate_results": [{"name": "worker", "verdict": "fail",
                "detail": result.note or "worker returned not-ok (aborted/failed)"}], "climbs": []}

    for g in (normalize_gate(x) for x in task.gates):
        if g["kind"] == "null-checker":
            passed, detail = _run_nullchecker_gate(g, result, octx)
        elif g["kind"] == "diff-scope":
            passed, detail = _run_diffscope_gate(g, octx)
        elif g["kind"] == "done_means":
            continue                                          # handled as the checklist below
        else:
            passed, detail = _run_command_gate(g, octx)

        if passed is None:                                    # AMBIGUOUS -> qwen interprets, else climb
            if callable(octx.gate_judge):
                passed = bool(octx.gate_judge(g["name"], detail))
                detail += f" | qwen: {'pass' if passed else 'fail'}"
            else:
                climbs.append({"gate": g["name"], "detail": detail})
                gate_results.append({"name": g["name"], "verdict": "ambiguous", "detail": detail})
                continue
        gate_results.append({"name": g["name"], "verdict": "pass" if passed else "fail", "detail": detail})

    # done_means checklist: each criterion explicitly confirmed or explicitly failed. v1 mechanical
    # proxy — a criterion is confirmed iff no gate failed (the gates ARE the checkable evidence);
    # anything the gates don't cover is a climb, never an assumed pass.
    hard_fail = any(r["verdict"] == "fail" for r in gate_results)
    for crit in task.done_means:
        gate_results.append({"name": f"done_means: {crit[:48]}",
                             "verdict": "fail" if hard_fail else "confirmed",
                             "detail": "tied to gate outcomes"})
    passed = not hard_fail and not climbs
    return {"passed": passed, "gate_results": gate_results, "climbs": climbs}


# ---------------------------------------------------------------------------
# escalation trinary (dispatch v1): one rung up w/ autopsy · two fails -> spawner · abstain-up honored
# ---------------------------------------------------------------------------
COST_RANK = {"free": 0, "low": 1, "mid": 2, "high": 3}


def escalate_target(current: dict, roster: list[dict]) -> dict | None:
    """The next rung UP the capability ladder (by cost) — a more capable worker retries the SAME
    task with the autopsy attached. None if current is already the top rung (then it returns to the
    spawner as malformed, not thrown away). Cost ordering: free < low < mid < high."""
    cur = COST_RANK.get(current.get("cost", ""), -1)
    higher = [e for e in roster if COST_RANK.get(e.get("cost", ""), -1) > cur]
    return min(higher, key=lambda e: COST_RANK.get(e.get("cost", ""), 99)) if higher else None


def gate_and_escalate(task: Task, result: WorkerResult, octx: OCtx) -> dict:
    """Gate the returned work; on failure apply the trinary. An attempt (worker + gate autopsy) is
    ALWAYS recorded on the task and travels with it on escalation — the next rung starts from the
    autopsy, not from zero. Abstain-upward (worker self-escalates before/for the task) is honored:
    it escalates, it is never punished as a fail — faking done is the only sin."""
    gr = run_gates(task, result, octx)
    self_escalated = (not result.ok) and any(
        k in (result.note or "").lower() for k in ("escalate", "beyond", "abstain"))
    task.attempts.append({"roster_entry": task.roster_entry, "outcome": "pass" if gr["passed"] else
                          ("self-escalate" if self_escalated else "fail"),
                          "gate_results": gr["gate_results"], "note": result.note})

    if gr["passed"]:
        task.status = "done"
        return {"status": "done", "gate_results": gr["gate_results"], "climbs": gr["climbs"]}

    if gr["climbs"] and not self_escalated:                   # ambiguous gate, no qwen verdict -> Aaron
        task.status = "gated"
        return {"status": "climb", "climbs": gr["climbs"], "gate_results": gr["gate_results"]}

    fails = sum(1 for a in task.attempts if a["outcome"] in ("fail", "self-escalate"))
    current = task.roster_entry or {}
    target = escalate_target(current, octx.roster)
    if target is not None and fails < 2:
        task.status = "escalated"
        task.roster_entry = target
        return {"status": "escalated", "to": target, "autopsy": gr["gate_results"],
                "from": current.get("model"), "reason": "self-escalate" if self_escalated else "gate fail"}
    # two failed rungs, or no higher rung -> back to the spawner as malformed (wrong scope/done_means)
    task.status = "returned"
    return {"status": "returned_to_spawner", "autopsy": gr["gate_results"],
            "reason": f"{fails} failed rung(s); "
                      + ("no higher rung" if target is None else "escalation budget spent")}


# ---------------------------------------------------------------------------
# outcome log — dispatch_log.jsonl. Acceptance rate is the headline metric (DeepSpark).
# ---------------------------------------------------------------------------
def log_outcome(task: Task, outcome: dict, result: WorkerResult | None, octx: OCtx,
                log_path: Path = DISPATCH_LOG) -> dict:
    """Append ONE dispatch outcome. Cost fields carry per-provider (from the worker result) so the
    log's accounting is complete for every rung — free/owned rungs record $0, not blank. NOTHING is
    computed on the log in v1: it accumulates like the field logs until the density instrument earns
    its fair-chance experiment (fact 9 discipline). acceptance rate is the future learned router's
    optimization target, recorded now."""
    # attribute the outcome to the model that ACTUALLY RAN (the last attempt), not task.roster_entry
    # — escalation advances roster_entry to the NEXT rung, so reading it here would credit a failure
    # to the wrong model and mistrain the future learned router (acceptance rate is the headline metric).
    ran = (task.attempts[-1]["roster_entry"] if task.attempts and task.attempts[-1].get("roster_entry")
           else task.roster_entry) or {}
    rec = {"ts": octx.now, "task_id": task.id, "type": task.type,
           "model": ran.get("model"), "provider": ran.get("provider"),
           "status": outcome.get("status"),
           "gates_passed": outcome.get("status") == "done",
           "escalated_to": (outcome.get("to") or {}).get("model") if outcome.get("status") == "escalated" else None,
           "returned": outcome.get("status") in ("returned", "returned_to_spawner"),
           "attempts": len(task.attempts),
           "cost": result.cost if result else None}
    # Under fan-out, many threads append here at once. The lock seam serializes the write so lines
    # never interleave/tear (no-op nullcontext for the sequential spine — run_task's tests pass None).
    with (octx.log_lock or nullcontext()):
        with log_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec) + "\n")
    return rec


def acceptance_rate(log_path: Path = DISPATCH_LOG) -> dict:
    """Headline metric: gates-passed rate per (type, model). Reported, not optimized (v1). Reads the
    log; empty/absent log -> empty report, never a crash."""
    if not log_path.exists():
        return {}
    buckets: dict[tuple, list[int]] = {}
    for line in log_path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        r = json.loads(line)
        key = (r.get("type"), r.get("model"))
        buckets.setdefault(key, []).append(1 if r.get("gates_passed") else 0)
    return {f"{t}/{m}": {"accepted": sum(v), "total": len(v), "rate": round(sum(v) / len(v), 3)}
            for (t, m), v in buckets.items()}


# ---------------------------------------------------------------------------
# run_task — the full spine loop for ONE task (detect -> yes -> prompt -> dispatch -> gate -> log)
# ---------------------------------------------------------------------------
def run_task(task: Task, octx: OCtx, log_path: Path = DISPATCH_LOG) -> dict:
    """One task through the whole spine. Ungrounded -> returned (never dispatched). Grounded ->
    null-gate (Aaron's yes) -> on yes the orchestrator builds the prompt and dispatches -> gates run
    -> escalation trinary -> outcome logged. Returns the outcome dict; every terminal path logs."""
    if not task.grounded:
        task.status = "returned"
        outcome = {"status": "returned", "reason": "ungrounded", "gaps": task.grounding_gaps()}
        log_outcome(task, outcome, None, octx, log_path)
        return outcome

    if task.hold:                                            # policy hold (e.g. fact-9 eval gate) — never dispatches
        task.status = "held"
        outcome = {"status": "held", "reason": task.hold}
        log_outcome(task, outcome, None, octx, log_path)
        return outcome

    # LIVENESS (task 32): passing log_path makes null_gate write 'running' BEFORE its synchronous
    # dispatch blocks (dispatch only returns once the worker completed — a write here would be too
    # late and the HUD would miss the whole model run). Every path below lands the terminal event.
    gate = null_gate(task, octx, log_path)
    if not gate["dispatched"]:
        outcome = {"status": "not_dispatched", "reason": gate["reason"], "ask": gate["ask"].render()}
        log_outcome(task, outcome, None, octx, log_path)
        return outcome

    # subscription rate limit -> QUEUE AND WAIT (fact 111). The task returns to 'queued' and the loop
    # moves on; it is NOT gated/escalated and NEVER spills to a paid API. It retries a later cycle.
    # LIVENESS: 'requeued' terminal — the worker is NOT live and the HUD must not ring a queued task.
    if gate["result"].rate_limited:
        task.status = "queued"
        outcome = {"status": "waiting", "reason": gate["result"].note, "requeue": True}
        log_outcome(task, outcome, gate["result"], octx, log_path)
        write_status(task, "requeued", octx, log_path)
        return outcome

    # LIVENESS: the dispatch completed — close the 'running' null_gate opened, whatever the verdict.
    outcome = gate_and_escalate(task, gate["result"], octx)
    log_outcome(task, outcome, gate["result"], octx, log_path)
    write_status(task, "finished", octx, log_path)
    return outcome


# ---------------------------------------------------------------------------
# fan-out (task 30) — dispatch N tasks CONCURRENTLY, each worker in an ISOLATED git worktree.
# ---------------------------------------------------------------------------
# Clean-room, inspired by Nate B Jones's Ringer (New Brain facts 171/173/174 — idea, not code; the
# Ringer source is never read). The spine (run_task) is reused UNCHANGED per task; fan-out only adds
# the concurrency wrapper + the isolated execution environment. Three ratified laws shape it:
#   - Isolation is EARNED, never assumed (Fable ruling 2026-07-05): a task runs in parallel only if
#     every gate is on the parallel-safe allowlist (roster.json data). Unknown gate -> sequential +
#     flag. "Do not fake isolation" — the first unclassified DB-touching gate would corrupt the shared
#     new_brain_test DB, so the conservative default is sequential (facts 206-208).
#   - Concurrency cap + the allowlist are roster.json DATA (fact 91, zero-code-alteration).
#   - Fan-out parallelizes EXECUTION, never approval (fact 73): approvals are COLLECTED FIRST as one
#     batched ask (per-item yes/no, facts 19-21), THEN the approved set executes concurrently. The
#     per-task yes-gate seam is kept internally; the batching is presentation-layer.
FANOUT_DEFAULTS = {"max_concurrency": 3, "parallel_safe_gates": [], "worktree_root": None}


def load_fanout_config(path: Path = ROSTER) -> dict:
    """The fan-out knobs live as DATA in roster.json's optional 'fanout' block (fact 91): the
    concurrency cap, the parallel-safe gate allowlist, and the worktree root. Absent block -> the
    conservative defaults (cap 3, empty allowlist => everything sequential until a gate is opted in)."""
    if not path.exists():
        return dict(FANOUT_DEFAULTS)
    data = json.loads(path.read_text(encoding="utf-8")).get("fanout", {})
    return {**FANOUT_DEFAULTS, **data}


def isolation_class(task: Task, config: dict) -> dict:
    """Conservative opt-IN classification (Fable ruling 2026-07-05). A task is PARALLEL-safe only when
    it declares gates AND every gate name matches the parallel_safe_gates allowlist (roster.json data,
    e.g. 'orchestrator_test', 'ladder_test' — the deterministic gates, fact 206). Any unlisted gate,
    or no gates at all, is SEQUENTIAL-only and flagged — isolation for it is unverified, and an
    unverified gate is treated as shared-state-touching (facts 207-208). Returns the verdict + the
    human reason so the classification is never hidden."""
    allow = config.get("parallel_safe_gates", [])
    gate_names = [normalize_gate(g)["name"] for g in task.gates]
    if not gate_names:
        return {"klass": "sequential", "parallel_safe": False, "flagged": True, "unlisted": [],
                "reason": "no gates declared — isolation unverifiable; sequential-only (conservative)"}
    unlisted = [g for g in gate_names if not any(pat in g for pat in allow)]
    if unlisted:
        return {"klass": "sequential", "parallel_safe": False, "flagged": True, "unlisted": unlisted,
                "reason": f"gate(s) not on the parallel-safe allowlist: {unlisted} — sequential-only"}
    return {"klass": "parallel", "parallel_safe": True, "flagged": False, "unlisted": [],
            "reason": "every gate is on the parallel-safe allowlist"}


def _git(args: list[str], cwd: Path = HERE) -> subprocess.CompletedProcess:
    """A git primitive anchored at the repo (git -C semantics via cwd). Read-only calls (rev-parse,
    worktree list) and the worktree add/remove mechanics all route through here."""
    return subprocess.run(["git", *args], cwd=str(cwd), capture_output=True, text=True)


@dataclass
class Worktree:
    """An isolated working copy for one worker — filesystem isolation AND the revert path from ONE
    mechanism (task 30 done_means). `git worktree add <path> -b task/<id> <base>`; the worker + gate
    operate in <path>; on a passed gate the worktree is REMOVED (the branch persists for review), on a
    failed gate it is KEPT for autopsy. Removal runs without --force: a dirty 'passed' tree refuses to
    remove, which is itself a flag (the worker never committed) — surfaced, never force-deleted
    (blueprint rule 9: no destructive shortcuts by workers)."""
    task_id: str
    base: str
    root: Path
    repo: Path = HERE

    @property
    def branch(self) -> str:
        return f"task/{self.task_id}"

    @property
    def path(self) -> Path:
        return self.root / self.task_id

    def add(self) -> "Worktree":
        self.root.mkdir(parents=True, exist_ok=True)
        proc = _git(["worktree", "add", str(self.path), "-b", self.branch, self.base], self.repo)
        if proc.returncode != 0:
            raise RuntimeError(f"git worktree add failed for {self.task_id}: {proc.stderr.strip()}")
        return self

    def remove(self) -> tuple[bool, str]:
        proc = _git(["worktree", "remove", str(self.path)], self.repo)
        return (proc.returncode == 0, proc.stderr.strip())


def worktree_worker_runner(worktree_path: Path):
    """The transport runner seam bound to a worktree: the worker CLI runs with cwd=<worktree> so its
    file edits land in the isolated copy. The paid-API key stays scrubbed (subscription-only, facts
    109/115) — the same guarantee the sequential path gives."""
    def _run(cmd, timeout):
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout,
                              cwd=str(worktree_path), env=_subscription_env())
    return _run


def worktree_gate_runner(worktree_path: Path):
    """The gate runner seam bound to a worktree. Mirrors _default_gate_runner EXCEPT a relative
    '.venv/...' interpreter resolves against the MAIN repo, not the worktree: .venv is gitignored so it
    is ABSENT from a fresh worktree (dogfooding lesson 3). The interpreter is location-independent and
    imports resolve from the worktree cwd — so the gate runs the worktree's code with the main venv."""
    def _run(cmd, timeout):
        parts = shlex.split(cmd, posix=True) if isinstance(cmd, str) else list(cmd)
        if parts:
            exe = parts[0]
            if ("/" in exe) or ("\\" in exe):
                p = Path(exe)
                parts[0] = str(p if p.is_absolute() else (HERE / p))   # MAIN repo venv, not the worktree
            else:
                parts[0] = shutil.which(exe) or exe
        return subprocess.run(parts, capture_output=True, text=True, timeout=timeout,
                              cwd=str(worktree_path))
    return _run


@dataclass
class FanoutPlan:
    """The batched decision surface: every grounded+routed dispatch presented TOGETHER as one ask,
    each carrying its own yes/no (facts 19-21). This is the object the batch_approve seam receives —
    live it renders to Aaron (the HUD 'Needs your yes (N)' band); tests stub it. Fan-out collects the
    whole batch here BEFORE any worker launches — execution parallelizes, approval does not (fact 73)."""
    asks: list  # list[DispatchAsk]

    def render(self) -> str:
        head = f"BATCHED DISPATCH — {len(self.asks)} task(s) need your yes (per-item):"
        return "\n\n".join([head] + [a.render() for a in self.asks])


def run_fanout(tasks, octx: OCtx, log_path: Path = DISPATCH_LOG, config: dict | None = None,
               worktree_factory=None) -> dict:
    """Dispatch a queue CONCURRENTLY with per-worker worktree isolation. The flow honors the three
    ratified laws: ground -> ONE batched approval -> classify the approved set for isolation safety ->
    execute (parallel-safe concurrently up to the cap; sequential-only serialized). Every approved task
    runs through the UNCHANGED run_task with a per-task OCtx whose runner+gate_runner are bound to its
    worktree and whose approve seam returns the already-granted batch decision. A shared log_lock keeps
    dispatch_log appends from interleaving. Returns a structured summary (never raises for one bad
    task — its worktree is kept for autopsy and its outcome recorded).

    worktree_factory is a test seam: (task_id) -> a Worktree-like object with .path and .remove();
    defaults to the real git-backed Worktree."""
    config = {**FANOUT_DEFAULTS, **(config or load_fanout_config())}
    cap = max(1, int(config.get("max_concurrency", 3)))
    root = Path(config["worktree_root"]) if config.get("worktree_root") else (HERE.parent / "aporia-worktrees")
    base = config.get("base") or _git(["rev-parse", "HEAD"]).stdout.strip()
    log_lock = octx.log_lock or threading.Lock()
    make_wt = worktree_factory or (lambda task_id: Worktree(task_id, base, root).add())

    # --- ground: ungrounded tasks never dispatch; they log as returned (spine law, kept) ---
    dispatchable, returned = ground(list(tasks))
    for t in returned:
        log_outcome(t, {"status": "returned", "reason": "ungrounded", "gaps": t.grounding_gaps()},
                    None, replace(octx, log_lock=log_lock), log_path)

    # --- batched approval: collect the WHOLE batch first (approval is never parallelized) ---
    asks = [build_dispatch_ask(t, octx) for t in dispatchable]
    for t, ask in zip(dispatchable, asks):
        t.roster_entry = ask.entry
    decisions = octx.batch_approve(FanoutPlan(asks)) if callable(octx.batch_approve) else {}
    decisions = decisions or {}
    approved = [t for t in dispatchable if decisions.get(t.id)]
    denied = [t for t in dispatchable if not decisions.get(t.id)]
    for t in denied:
        ask = next(a for a in asks if a.task.id == t.id)
        log_outcome(t, {"status": "not_dispatched", "reason": "no yes (batched)", "ask": ask.render()},
                    None, replace(octx, log_lock=log_lock), log_path)

    # --- classify the approved set for isolation safety (the crux) ---
    classes = {t.id: isolation_class(t, config) for t in approved}

    results, kept = {}, []
    kept_lock = threading.Lock()

    def _execute(task: Task) -> dict:
        wt = make_wt(task.id)
        # Bind the worker + gate to the worktree cwd (the isolation). An already-injected seam passes
        # through unchanged — the canary stubs the worker (no spend) while keeping the REAL worktree
        # gate; the deterministic tests stub both. Production leaves both None -> fully isolated.
        per = replace(octx, approve=lambda ask: True,           # granted in the batch; per-task seam kept
                      runner=octx.runner or worktree_worker_runner(wt.path),
                      gate_runner=octx.gate_runner or worktree_gate_runner(wt.path), log_lock=log_lock)
        outcome = run_task(task, per, log_path)
        if outcome.get("status") == "done":
            ok, err = wt.remove()
            if not ok:                                          # dirty passed tree -> a flag, not a force-delete
                with kept_lock:
                    kept.append({"task": task.id, "path": str(wt.path), "why": f"remove refused: {err}"})
        else:
            with kept_lock:
                kept.append({"task": task.id, "path": str(wt.path), "why": f"gate {outcome.get('status')} — autopsy"})
        return outcome

    # --- sequential-only tasks: strictly serial (their gates touch shared state — fact 207) ---
    for t in [t for t in approved if classes[t.id]["klass"] == "sequential"]:
        results[t.id] = _execute(t)

    # --- parallel-safe tasks: concurrent up to the cap (roster.json data) ---
    par = [t for t in approved if classes[t.id]["klass"] == "parallel"]
    if par:
        with ThreadPoolExecutor(max_workers=cap) as ex:
            futs = {ex.submit(_execute, t): t for t in par}
            for fut in as_completed(futs):
                results[futs[fut].id] = fut.result()

    return {"cap": cap, "base": base,
            "dispatched": [t.id for t in approved], "denied": [t.id for t in denied],
            "returned": [t.id for t in returned],
            "classes": classes, "results": results, "kept_worktrees": kept}


# ---------------------------------------------------------------------------
# demo — watch the queue, ground it, route it, report (full loop is opt-in below)
# ---------------------------------------------------------------------------
def main() -> int:
    tasks = load_queue()
    dispatchable, returned = ground(tasks)
    roster = load_roster()
    _log(f"queue: {len(tasks)} record(s) | {len(dispatchable)} grounded | {len(returned)} returned")
    for t in dispatchable:
        entry = route(t, roster)
        t.roster_entry = entry
        target = f"{entry['model']} ({entry['provider']}/{entry['transport']})" if entry \
            else "UNROUTED -> routing null (surfaces to Aaron)"
        _log(f"  GROUNDED  [{t.type:11}] {t.id}: {t.title}  ->  {target}")
    for t in returned:
        _log(f"  RETURNED  [{t.type:11}] {t.id}: {t.title}")
        for gap in t.grounding_gaps():
            _log(f"              - {gap}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())