dispatch.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/dispatch.py
 
# OCT-LINT: disable=tests_exist reason="Covered by test_221_dispatch.py"
"""
Fail-closed, one-shot plan dispatch orchestration.
 
Purpose
-------
Plan 221 P2/P3 (design §4.1). Every step of dispatching a plan either
goes through this module — which fills it correctly — or refuses loudly
with the reason named. It resolves the plan by number, validates its
``requires:`` block, creates branch and worktree idempotently, allocates
the ``agent-NNN`` identity, renders the launch file from the ONE
committed template, and refuses the WHOLE dispatch if any template field
would render unfilled. Hand-authoring a launch file is a workflow
violation from the moment this landed.
 
Responsibilities
----------------
- :class:`DispatchRequest` / :class:`DispatchResult` — the operator's
  inputs and the committed outputs of one attempt.
- :class:`DispatchHooks` / :data:`NO_HOOKS` — the plan-259 seam: one
  announcement after a dispatch commits, and an empty default.
- :class:`DispatchError` — the named refusal at the dispatch boundary.
- The refusal matrix (W3): missing/incomplete ``requires:``, plan absent
  from master, an open instance without ``--again``.
- All-or-nothing rendering (W4): no instance allocated and no launch
  file written when anything refuses.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (dispatch attempted, refused, completed).
    L3 — semantic details (plan resolution, template rendering).
    L4 — deep tracing.
 
Contracts
---------
- BASE_SHA is the point the branch actually diverges from master,
  recorded in the launch file and never re-derived: master's SHA for a
  branch created by this dispatch, and the merge-base of master and the
  branch tip for a pre-existing one (plan 223 P4). The two values
  coincide at creation, so a fresh dispatch is unchanged.
- Refusals raise :class:`DispatchError` BEFORE any state mutates; the
  registry allocation and the launch-file write share one transaction
  (via :meth:`AgentInstanceRegistry.dispatch_allocation`).
- Git operations run against the caller's ``repo_root`` only.
- The plan-259 hook is told AFTER everything has committed and its
  failure is reported on stderr, never raised: by then the worktree
  exists, and unmaking a real worktree because a log was unreachable
  would be the worse failure.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.agent_instances (the runtime instance registry)
- yaml (plan frontmatter parsing; already a project dependency)
- oc_orchestrator.worktree_env (the git anchors, the per-worktree venv prepare and the dispatch timing ledger)
"""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
from pathlib import Path
import re
import string
import subprocess
import sys
import tempfile
from typing import Any, Protocol, runtime_checkable
 
import yaml
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.agent_instances import (
    AgentInstance,
    AgentInstanceRegistry,
    HarnessKind,
)
from oc_orchestrator.worktree_env import (
    PrepareRequest,
    PrepareResult,
    WorktreeEnvError,
    discover_packages,
    ensure_branch_and_worktree,
    prepare_worktree,
    venv_scripts_dir,
)
 
_dbg("OC_CHAT.MAIN", "module", "dispatch loaded", 4)
 
 
#: The launch briefs ship WITH this package (plan 270 P3): they are the
#: data half of what a dispatch produces, so a consumer that installs the
#: package gets a working default and overrides it per dispatch through
#: ``DispatchRequest.template_path``. Their prose still names oc_chat's
#: module paths, which is the shims' business and not this module's.
_TEMPLATES_DIR = Path(__file__).resolve().parent / "templates"
DEFAULT_LAUNCH_TEMPLATE = _TEMPLATES_DIR / "launch.md"
#: The pre-plan-241 brief, kept byte-identical so ``--shared-venv`` renders
#: the legacy form exactly rather than a lookalike (plan 241 P4/W5).
SHARED_VENV_LAUNCH_TEMPLATE = _TEMPLATES_DIR / "launch_shared_venv.md"
_REQUIRED_REQUIRES = ("tier", "repo", "thinking", "session")
_VALID_THINKING = frozenset({"off", "low", "medium", "high", "max"})
_VALID_SESSION = frozenset({"fresh", "continue", "either"})
_LEFTOVER_TOKEN_RE = re.compile(r"\{[^{}\n]+\}|<[^<>\n]+>")
_PLAN_NUMBER_RE = re.compile(r"^[0-9]{3}$")
 
 
class DispatchError(RuntimeError):
    """A named refusal at the plan-dispatch boundary."""
 
 
# ── The hub seam (plan 259 P6) ───────────────────────────────────────────
#
# The plan-257 idiom, verbatim: a Protocol, a ``NO_*`` default, and the
# live default registered by the CALLER rather than by this layer. A
# dispatch is a hub KIND -- the orchestrator's picture of the fleet is
# the log and not a guess -- but this module announces that one thing
# happened and knows nothing about who is listening, so ``core/hub.py``
# never appears in the dispatch layer's own imports.
 
 
@runtime_checkable
class DispatchHooks(Protocol):
    """What a listener is told about one completed dispatch."""
 
    def dispatched(self, result: "DispatchResult") -> None:
        """A plan was dispatched: the instance exists and the launch is written."""
 
 
class _NoHooks:
    """The default listener: it does nothing, and that is the whole point."""
 
    def dispatched(self, result: "DispatchResult") -> None:
        """Ignore the dispatch. A caller that wants it recorded injects hooks."""
 
 
#: The default when a caller wants no announcement at all, so ``dispatch``
#: behaves exactly as it did before plan 259.
NO_HOOKS: DispatchHooks = _NoHooks()
 
 
@dataclass(frozen=True)
class DispatchRequest:
    """All operator-controlled inputs to one dispatch attempt."""
 
    plan_number: str
    plans_dir: Path
    repo_root: Path
    registry_path: Path
    launch_dir: Path
    model_attested: str
    thinking: str | None
    harness: HarnessKind
    template_path: Path | None = None
    again_note: str | None = None
    shared_venv: bool = False
    uv_executable: str = "uv"
    uv_cache_dir: Path | None = None
    #: Plan 259 P6. Told once, after the dispatch has committed. The
    #: default reaches nothing.
    hooks: DispatchHooks = field(default=NO_HOOKS)
 
 
@dataclass(frozen=True)
class DispatchResult:
    """The committed outputs from one successful dispatch."""
 
    instance: AgentInstance
    plan_id: str
    launch_path: Path
    branch: str
    worktree: Path
    base_sha: str
    tier: str
    thinking: str
    session: str
    prepare: PrepareResult | None = None
 
 
@dataclass(frozen=True)
class _PlanDispatchData:
    plan_id: str
    branch: str
    worktree: Path
    tier: str
    thinking: str
    session: str
    path: Path
 
 
def _run_git(repo_root: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        ["git", *args],
        cwd=repo_root,
        text=True,
        capture_output=True,
        check=False,
    )
    if check and result.returncode != 0:
        detail = (result.stderr or result.stdout).strip()
        raise DispatchError(f"git {' '.join(args)} failed: {detail}")
    return result
 
 
def _resolve_plan(plans_dir: Path, plan_number: str) -> Path:
    if not _PLAN_NUMBER_RE.fullmatch(plan_number):
        raise DispatchError(
            f"invalid plan number {plan_number!r}; expected exactly three digits"
        )
    matches = sorted(Path(plans_dir).glob(f"{plan_number}-*.md"))
    if not matches:
        raise DispatchError(
            f"no active plan file resolves from plan number {plan_number}"
        )
    if len(matches) != 1:
        names = ", ".join(path.name for path in matches)
        raise DispatchError(
            f"plan number {plan_number} is ambiguous: {names}"
        )
    return matches[0].resolve()
 
 
def _frontmatter(path: Path) -> dict[str, Any]:
    text = path.read_text(encoding="utf-8")
    lines = text.splitlines(keepends=True)
    if not lines or lines[0].rstrip("\r\n") != "---":
        raise DispatchError(f"plan {path.name} has no YAML frontmatter")
    end = next(
        (index for index, line in enumerate(lines[1:], start=1) if line.rstrip("\r\n") == "---"),
        None,
    )
    if end is None:
        raise DispatchError(f"plan {path.name} has unclosed YAML frontmatter")
    try:
        loaded = yaml.safe_load("".join(lines[1:end]))
    except yaml.YAMLError as exc:
        raise DispatchError(f"plan {path.name} frontmatter is invalid YAML: {exc}") from exc
    if not isinstance(loaded, dict):
        raise DispatchError(f"plan {path.name} frontmatter must be a mapping")
    return loaded
 
 
def _nonblank(data: dict[str, Any], field: str, *, context: str) -> str:
    value = str(data.get(field) or "").strip()
    if not value:
        raise DispatchError(f"{context} is missing required field {field!r}")
    return value
 
 
def _load_plan(path: Path, plan_number: str, thinking_override: str | None) -> _PlanDispatchData:
    data = _frontmatter(path)
    plan_id = _nonblank(data, "plan_id", context=f"plan {path.name}")
    if not plan_id.startswith(f"{plan_number}-"):
        raise DispatchError(
            f"plan number {plan_number} does not match frontmatter plan_id {plan_id!r}"
        )
    branch = _nonblank(data, "branch", context=f"plan {path.name}")
    worktree = Path(_nonblank(data, "worktree", context=f"plan {path.name}")).resolve()
    requires = data.get("requires")
    if not isinstance(requires, dict):
        raise DispatchError(
            f"plan {plan_id!r} requires: is missing or is not a mapping"
        )
    missing = [
        field
        for field in _REQUIRED_REQUIRES
        if not str(requires.get(field) or "").strip()
    ]
    if missing:
        raise DispatchError(
            f"plan {plan_id!r} has incomplete requires:; missing "
            + ", ".join(missing)
        )
    tier = str(requires["tier"]).strip()
    thinking = (thinking_override or str(requires["thinking"])).strip().lower()
    session = str(requires["session"]).strip().lower()
    if thinking not in _VALID_THINKING:
        raise DispatchError(
            f"plan {plan_id!r} has invalid thinking requirement {thinking!r}"
        )
    if session not in _VALID_SESSION:
        raise DispatchError(
            f"plan {plan_id!r} has invalid session requirement {session!r}"
        )
    return _PlanDispatchData(
        plan_id=plan_id,
        branch=branch,
        worktree=worktree,
        tier=tier,
        thinking=thinking,
        session=session,
        path=path,
    )
 
 
@dataclass(frozen=True)
class PlanAnchors:
    """A plan's declared git anchors, read straight from its frontmatter."""
 
    plan_id: str
    branch: str
    worktree: Path
 
 
def resolve_plan_anchors(plans_dir: Path, plan_number: str) -> PlanAnchors:
    """Resolve one plan number to the branch and worktree it declares.
 
    Plan 241 P1/P5. ``prepare-worktree`` and ``teardown-worktree`` act on
    exactly the anchors ``dispatch`` would use, from the same parser, so the
    three commands can never disagree about which tree they mean.
    """
    func = "resolve_plan_anchors"
    plan = _load_plan(_resolve_plan(plans_dir, plan_number), plan_number, None)
    _dbg("OC_CHAT.MAIN", func, f"{plan.plan_id} -> {plan.branch}", 3)
    return PlanAnchors(
        plan_id=plan.plan_id, branch=plan.branch, worktree=plan.worktree
    )
 
 
def _require_committed_on_master(repo_root: Path, plan_path: Path) -> str:
    try:
        relative = plan_path.relative_to(repo_root).as_posix()
    except ValueError as exc:
        raise DispatchError(
            f"plan file {plan_path} is outside repository {repo_root}"
        ) from exc
    probe = _run_git(
        repo_root,
        "cat-file",
        "-e",
        f"master:{relative}",
        check=False,
    )
    if probe.returncode != 0:
        raise DispatchError(
            f"plan {relative} is not committed on master"
        )
    unstaged = _run_git(
        repo_root,
        "diff",
        "--quiet",
        "master",
        "--",
        relative,
        check=False,
    )
    staged = _run_git(
        repo_root,
        "diff",
        "--cached",
        "--quiet",
        "master",
        "--",
        relative,
        check=False,
    )
    if unstaged.returncode != 0 or staged.returncode != 0:
        raise DispatchError(
            f"plan {relative} differs from its committed master version"
        )
    return _run_git(repo_root, "rev-parse", "master").stdout.strip()
 
 
def _branch_exists(repo_root: Path, branch: str) -> bool:
    probe = _run_git(
        repo_root,
        "show-ref",
        "--verify",
        f"refs/heads/{branch}",
        check=False,
    )
    return probe.returncode == 0
 
 
def _base_sha(repo_root: Path, branch: str, master_sha: str) -> str:
    """Return the SHA the worker's BASE diff should be measured against.
 
    Plan 223 P4 (from the 225 gate finding). Recording master's CURRENT SHA
    for a branch that was cut earlier is a lie the moment master moves: the
    2026-09-01 dispatch rendered ``344fe8ce`` for a branch cut at
    ``b7962f5f``, so the worker's "diff against BASE" silently spanned every
    commit master had gained in between. The truthful anchor for an existing
    branch is where it diverged — the merge-base.
 
    A branch this dispatch is about to CREATE has no divergence yet, so
    master's SHA is both values at once and the fresh path is unchanged.
    """
 
    func = "_base_sha"
    if not _branch_exists(repo_root, branch):
        _dbg("OC_CHAT.MAIN", func, f"{branch} is new; base = master {master_sha[:8]}", 3)
        return master_sha
    probe = _run_git(repo_root, "merge-base", "master", branch, check=False)
    merge_base = probe.stdout.strip()
    if probe.returncode != 0 or not merge_base:
        detail = (probe.stderr or probe.stdout).strip()
        raise DispatchError(
            f"branch {branch!r} has no merge-base with master; "
            f"refusing to record a BASE_SHA that names no shared history: {detail}"
        )
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"{branch} exists; base = merge-base {merge_base[:8]} (master {master_sha[:8]})",
        3,
    )
    return merge_base
 
 
def _template_field_names(template: str) -> set[str]:
    names: set[str] = set()
    try:
        parts = string.Formatter().parse(template)
        for _literal, field_name, format_spec, conversion in parts:
            if field_name is None:
                continue
            if not field_name or not field_name.isidentifier():
                raise DispatchError(
                    f"launch template contains invalid field {field_name!r}"
                )
            if format_spec or conversion:
                raise DispatchError(
                    f"launch template field {field_name!r} may not use formatting or conversion"
                )
            names.add(field_name)
    except ValueError as exc:
        raise DispatchError(f"launch template syntax is invalid: {exc}") from exc
    return names
 
 
def render_launch_template(template_path: Path, fields: dict[str, str]) -> str:
    """Render the single launch template or refuse any unresolved field."""
 
    try:
        template = Path(template_path).read_text(encoding="utf-8")
    except OSError as exc:
        raise DispatchError(f"cannot read launch template {template_path}: {exc}") from exc
    required = _template_field_names(template)
    missing = sorted(name for name in required if not str(fields.get(name) or "").strip())
    if missing:
        raise DispatchError(
            "launch template has unfillable field(s): " + ", ".join(missing)
        )
    rendered = template.format_map(fields)
    leftover = _LEFTOVER_TOKEN_RE.search(rendered)
    if leftover:
        raise DispatchError(
            f"rendered launch contains placeholder token {leftover.group(0)!r}"
        )
    return rendered
 
 
def _prepare_worktree_env(
    request: DispatchRequest, plan: _PlanDispatchData, *, use_worktree_venv: bool
) -> PrepareResult:
    """Create the git anchors and, by default, the worktree's own venv.
 
    Plan 241 P1/P4. ``--shared-venv`` (``use_worktree_venv=False``) keeps
    the legacy behaviour — anchors only, the shared main venv in the brief.
    Otherwise the anchors and the per-worktree environment are made in one
    step and every refusal inside it aborts the whole dispatch: a silent
    fallback to the shared venv would re-open the F17 shadow.
    """
    func = "_prepare_worktree_env"
    try:
        if not use_worktree_venv:
            ensure_branch_and_worktree(
                request.repo_root.resolve(), plan.branch, plan.worktree
            )
            return PrepareResult(
                worktree=plan.worktree,
                branch=plan.branch,
                venv=None,
                python=None,
                installed=(),
                verified=(),
                uv_cache_dir=None,
                link_mode="n/a",
                already_prepared=False,
                repaired=(),
                skipped_reason="--shared-venv: the legacy shared-venv brief was requested",
                elapsed_seconds=0.0,
            )
        result = prepare_worktree(
            PrepareRequest(
                repo_root=request.repo_root.resolve(),
                branch=plan.branch,
                worktree=plan.worktree,
                uv_executable=request.uv_executable,
                uv_cache_dir=request.uv_cache_dir,
                # Plan 243 P4: every dispatched worktree is smoke-tested. The
                # brief tells the worker to run plan_status at STEP-0, so
                # dispatch runs it first: a brief handed out before that
                # command is known to work is the hand-repair loop ENV-02
                # exists to end.
                smoke=True,
            )
        )
    except WorktreeEnvError as exc:
        raise DispatchError(str(exc)) from exc
    if result.skipped_reason is not None:
        # The brief has already been rendered naming <worktree>\.venv, because
        # the repo root declared packages. A worktree that turns out to carry
        # none would leave that brief pointing at an interpreter nothing
        # built — the brief lying about the environment is the exact failure
        # this plan exists to end, so it fails closed instead.
        raise DispatchError(
            f"prepare skipped after the worktree-venv brief was rendered "
            f"({result.skipped_reason}); refusing to hand a worker a brief "
            f"naming an interpreter that was never built"
        )
    _dbg("OC_CHAT.MAIN", func, f"env ready for {plan.plan_id}", 3)
    return result
 
 
def _write_launch(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=path.parent,
        prefix=".launch_",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(text)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
 
 
def _launch_fields(
    request: DispatchRequest,
    plan: _PlanDispatchData,
    *,
    base_sha: str,
    launch_path: Path,
    agent_id: str,
    use_worktree_venv: bool,
) -> dict[str, str]:
    repo_root = request.repo_root.resolve()
    try:
        plan_path = plan.path.relative_to(repo_root).as_posix()
    except ValueError:
        plan_path = str(plan.path)
    # Plan 241 P3: the brief names the interpreter whose editable installs
    # point at THIS worktree, so the worker never has to juggle PYTHONPATH.
    venv = plan.worktree / ".venv" if use_worktree_venv else repo_root / ".venv"
    scripts = venv_scripts_dir(venv)
    return {
        "agent_id": agent_id,
        "base_sha": base_sha,
        "branch": plan.branch,
        "launch_path": str(launch_path),
        "model_attested": request.model_attested.strip(),
        "plan_id": plan.plan_id,
        "plan_number": request.plan_number,
        "plan_path": plan_path,
        "registry_path": str(request.registry_path.resolve()),
        "session": plan.session,
        "thinking": plan.thinking,
        "venv_oct": str(scripts / "oct.exe"),
        "venv_python": str(scripts / "python.exe"),
        "worktree": str(plan.worktree),
    }
 
 
def dispatch(request: DispatchRequest) -> DispatchResult:
    """Validate, prepare git state, allocate an instance, and render launch."""
 
    from oc_orchestrator.worktree_env import _engine_timing
    func = "dispatch"
    with _engine_timing(request.repo_root, "engine.dispatch", request.plan_number):
        _dbg("OC_CHAT.MAIN", func, f"dispatch attempt for plan {request.plan_number}", 2)
        if not request.model_attested.strip():
            raise DispatchError("--model must be an operator-attested non-empty value")
        repo_root = request.repo_root.resolve()
        if not (repo_root / ".git").exists():
            raise DispatchError(f"repository root is not a git worktree: {repo_root}")
        plan_path = _resolve_plan(request.plans_dir, request.plan_number)
        plan = _load_plan(plan_path, request.plan_number, request.thinking)
        master_sha = _require_committed_on_master(repo_root, plan.path)
        # Derived BEFORE the branch is created so a fresh dispatch reads master
        # rather than a merge-base against a branch this call is about to make.
        base_sha = _base_sha(repo_root, plan.branch, master_sha)
        _dbg("OC_CHAT.MAIN", func, f"resolved {plan.plan_id} at base {base_sha[:8]}", 3)
        launch_path = request.launch_dir.resolve() / f"{request.plan_number}.md"
 
        # Plan 241: the environment shape is decided BEFORE anything is rendered
        # or created, so the preflight render exercises the template the real
        # render will use. A tree carrying none of the declared packages (every
        # dispatch fixture) has no editable shadow to close, so it keeps the
        # legacy brief rather than being told about a venv it will not get.
        use_worktree_venv = not request.shared_venv and bool(
            discover_packages(repo_root)
        )
        template_path = request.template_path or (
            DEFAULT_LAUNCH_TEMPLATE if use_worktree_venv else SHARED_VENV_LAUNCH_TEMPLATE
        )
 
        preflight_fields = _launch_fields(
            request,
            plan,
            base_sha=base_sha,
            launch_path=launch_path,
            agent_id="agent-000",
            use_worktree_venv=use_worktree_venv,
        )
        render_launch_template(template_path, preflight_fields)
        prepared = _prepare_worktree_env(
            request, plan, use_worktree_venv=use_worktree_venv
        )
 
        registry = AgentInstanceRegistry(request.registry_path)
        previous = launch_path.read_text(encoding="utf-8") if launch_path.exists() else None
        launch_written = False
        try:
            with registry.dispatch_allocation(
                plan_id=plan.plan_id,
                model_attested=request.model_attested.strip(),
                thinking=plan.thinking,
                harness=request.harness,
                again_note=request.again_note,
            ) as instance:
                fields = _launch_fields(
                    request,
                    plan,
                    base_sha=base_sha,
                    launch_path=launch_path,
                    agent_id=instance.agent_id,
                    use_worktree_venv=use_worktree_venv,
                )
                rendered = render_launch_template(template_path, fields)
                _write_launch(launch_path, rendered)
                launch_written = True
                _dbg("OC_CHAT.MAIN", func,
                     f"dispatched {instance.agent_id} for {plan.plan_id}", 2)
        except Exception:
            if launch_written:
                if previous is None:
                    launch_path.unlink(missing_ok=True)
                else:
                    _write_launch(launch_path, previous)
            raise
        result = DispatchResult(
            instance=instance,
            plan_id=plan.plan_id,
            launch_path=launch_path,
            branch=plan.branch,
            worktree=plan.worktree,
            base_sha=base_sha,
            tier=plan.tier,
            thinking=plan.thinking,
            session=plan.session,
            prepare=prepared,
        )
        _announce_dispatched(request.hooks, result)
        return result
 
 
def _announce_dispatched(hooks: DispatchHooks, result: DispatchResult) -> None:
    """Tell the hooks once, after the dispatch has committed. Never raise.
 
    Plan 259 P6. By the time this runs the branch, the worktree, its venv,
    the instance row and the launch file all exist; a listener that cannot
    record the fact is a reason to say so, not a reason to unmake any of
    that. Undoing a real worktree because a log was unreachable would be
    the worse failure by a wide margin.
    """
    func = "_announce_dispatched"
    try:
        hooks.dispatched(result)
    except Exception as exc:  # noqa: BLE001 - EXTERNAL_ERROR (L2); the dispatch stands
        print(
            f"warning: dispatch hook for {result.plan_id} failed "
            f"({type(exc).__name__}: {exc}); the dispatch stands and "
            f"{result.launch_path} is written",
            file=sys.stderr,
        )
        _dbg("OC_CHAT.MAIN", func,
             f"EXTERNAL_ERROR: hook failed for {result.plan_id}: {exc}", 2)
        return
    _dbg("OC_CHAT.MAIN", func, f"announced {result.instance.agent_id}", 3)
 
 
#: What each ``requires.session`` value CONSTRAINS, in the operator's terms.
#: Plan 223 P4b: ``session=continue`` read like it named a target session; it
#: never did. It is the plan's requirement, and the operator picks a session
#: that satisfies it.
_SESSION_EXPLANATIONS = {
    "continue": "an existing worker session may be reused; fresh also works",
    "fresh": "a NEW session is required; an existing one may not be reused",
    "either": "any session satisfies this plan",
}
 
 
def format_dispatch_suggestion(
    result: DispatchResult,
    *,
    model_attested: str,
    registry_path: Path,
) -> list[str]:
    """Build the operator-facing suggestion lines for a completed dispatch.
 
    Plan 223 P4b. The suggestion line spells out what the session field
    constrains, and a following line names the instances that are actually
    open right now — the concrete reuse candidates, rather than leaving the
    operator to remember which windows are still alive.
 
    NOTE (scope): these are candidate INSTANCES, not sessions. Naming the
    actual session to continue requires ``agents.sqlite.session_uuid`` to be
    populated, which is plan 224's session-linkage territory and deliberately
    out of scope here; until then the operator maps instance to window.
    """
 
    func = "format_dispatch_suggestion"
    session_note = _SESSION_EXPLANATIONS.get(result.session, "")
    session_text = f"session={result.session}"
    if session_note:
        session_text += f" - {session_note}"
    lines = [
        *(result.prepare.report_lines() if result.prepare is not None else []),
        f"Suggested assignment: {model_attested} "
        f"(tier={result.tier}, thinking={result.thinking}, {session_text}).",
        f"Open instances: {_open_instance_summary(registry_path, result.instance.agent_id)}",
        f"Allocated {result.instance.agent_id} (state={result.instance.state}).",
        f"Read {result.launch_path} and execute end-to-end",
    ]
    _dbg("OC_CHAT.MAIN", func, f"suggestion rendered for {result.plan_id}", 3)
    return lines
 
 
def _open_instance_summary(registry_path: Path, exclude_agent_id: str) -> str:
    """Render open instances as ``agent-NNN (plan-id, state)``, or ``none``.
 
    The instance this dispatch just allocated is excluded: it is the thing
    being launched, not a candidate to reuse for it. A registry that cannot
    be read reports that fact rather than claiming an empty pool, because
    "none" and "unknown" are different answers to the operator's question.
    """
 
    func = "_open_instance_summary"
    try:
        instances = AgentInstanceRegistry(registry_path).list(active_only=True)
    except Exception as exc:  # noqa: BLE001 - a view must never fail a dispatch
        _dbg("OC_CHAT.MAIN", func, f"registry unreadable: {exc}", 3)
        return f"(registry unreadable: {exc})"
    rendered = [
        f"{instance.agent_id} ({instance.plan_id or '-'}, {instance.state})"
        for instance in instances
        if instance.agent_id != exclude_agent_id
    ]
    return ", ".join(rendered) if rendered else "none"
 
 
__all__ = [
    "DEFAULT_LAUNCH_TEMPLATE",
    "NO_HOOKS",
    "SHARED_VENV_LAUNCH_TEMPLATE",
    "DispatchError",
    "DispatchHooks",
    "DispatchRequest",
    "DispatchResult",
    "dispatch",
    "format_dispatch_suggestion",
    "render_launch_template",
    "resolve_plan_anchors",
]