plan_manage.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/plan_manage.py
 
# OCT-LINT: disable=tests_exist reason="Covered by tests/test_plan_manage_cli.py, tests/test_241_worktree_env.py, tests/test_243_worktree_extras.py and oc_chat's test_plan_manage_manifest.py"
"""
CLI for direct plan-inventory operations (Phase J).
 
Purpose
-------
Lets humans query and heal the plan inventory without needing OC Chat
to be running. Subcommands:
 
    list           Print every record (or filter by --status).
    status PLAN    Print one record's full metadata.
    reconcile      Heal inventory/filesystem drift.
    complete PLAN  Mark a plan complete and rename file to ``*_complete.md``.
    archive        Move ``*_complete.md`` files to ``plans/old/``.
    prepare-worktree NNN
                   Create a plan's branch, worktree and its own uv venv;
                   ``--smoke`` also runs the worker's first two commands
                   inside it (plan 243 P4).
    teardown-worktree NNN
                   Remove that worktree and its venv (plan 241 P5),
                   resolving the plan from ``plans/`` and then from
                   ``plans/completed/`` (plan 243 P5).
 
Responsibilities
----------------
- Thin wrapper around :mod:`oc_orchestrator.plan_inventory` and
  :mod:`oc_orchestrator.plans`. No new business logic.
- Exit code 0 on success, 1 on usage error or unrecoverable failure.
- :func:`main` carries the two consumer seams (plan 270 P4): ``prog``,
  the name the CLI gives itself in ``--help``, and ``dispatch_hooks``,
  the factory that decides who is told about a dispatch. Both default
  to this package standing alone.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — CLI lifecycle events
    L3 — per-subcommand semantic details
    L4 — per-record traces
 
Contracts
---------
- Stable interface: option names won't change without a deprecation
  cycle (matches the project convention for ``migrate_session_names``).
- ``--inventory`` and ``--plans-dir`` overrides are provided for tests
  and for power users with non-default project layouts.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.plan_inventory (the JSON inventory, PlanPaths, the status vocabulary and reconcile)
- oc_orchestrator.plans (the default plans directory and the list, find and complete operations)
- option_c_agent_contracts (lazy; PlanManifest from the ``[manifests]`` extra, imported inside :func:`_load_contracts` only: every other sub-command runs without it)
- oc_orchestrator.plan_metadata (PlanMetadataProvider and resolve_provider - the plan-header reader for manifest derivation)
- oc_orchestrator.agent_instances (lazy; agent id validation and the instance registry for the attested model, the agents view and dispatch)
- oc_orchestrator.dispatch (lazy; plan anchor resolution for the worktree sub-commands, and the dispatch workflow)
- oc_orchestrator.worktree_env (lazy; worktree prepare and teardown)
- option_c_agent_contracts.canonical (lazy; canonical_json for the manifest bytes, from the ``[manifests]`` extra)
- oc_orchestrator.plan_status (not imported; loaded by name through importlib for the worktree status file and the plan number)
"""
from __future__ import annotations
 
import argparse
import hashlib
import importlib
import re
import sys
import tempfile
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import Any
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.plan_metadata import PlanMetadataProvider, resolve_provider
from oc_orchestrator.plan_inventory import (
    JsonPlanInventory,
    PlanInventory,
    PlanPaths,
    VALID_STATUSES,
    reconcile,
)
from oc_orchestrator.plans import (
    DEFAULT_PLANS_DIR,
    complete_plan,
    find_plan_by_id,
    list_plans,
)
 
 
def _build_inventory(args: argparse.Namespace) -> tuple[PlanInventory, Path]:
    """Construct an inventory instance per CLI args."""
    plans_dir = Path(args.plans_dir) if args.plans_dir else DEFAULT_PLANS_DIR
    inv_path = (
        Path(args.inventory) if args.inventory
        else plans_dir / ".inventory.json"
    )
    return JsonPlanInventory(inv_path, plans_dir=plans_dir), plans_dir
 
 
# ── M1.4: PlanManifest commands ────────────────────────────────────────────
#
# ``init-manifest`` derives a :class:`option_c_agent_contracts.PlanManifest`
# from a plan markdown file's header + queue state and writes it as canonical
# JSON to ``oc_chat/plans/manifests/<plan_id>.manifest.json``. ``validate-
# manifest`` re-loads that file, Pydantic-validates it, and compares it
# field-by-field against the live-derived manifest. See plan 093 (M1.4).
 
MANIFESTS_DIRNAME = "manifests"
MANIFEST_SUFFIX = ".manifest.json"
 
# Only numbered plan files participate; the standing templates and the queue
# file are excluded (plan 093 §3.1).
_PLAN_FILE_RE = re.compile(r"^\d{3}-.*\.md$")
_EXCLUDED_PLAN_FILES = frozenset(
    {"PLAN_TEMPLATE.md", "BUMP_ROADMAP_TEMPLATE.md", "agent_queue.md"}
)
 
# The ``**Branch:**`` / ``**Worktree:**`` parsers and the frontmatter path
# moved to plan_metadata.py (plan 257 P2); this module reads them back
# through the shared provider. ``**Suggested model:**`` stays here: it is
# not one of PlanMeta's four fields.
_MODEL_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="markdown-header parsing regexes (**Suggested model:** etc.) flag as high-entropy; this CLI carries no secrets"
    r"\*\*Suggested model:\*\*\s*`?(?P<value>[A-Za-z0-9._-]+)", re.MULTILINE
)
 
_VALID_TIER = frozenset({"frontier", "strong", "fast"})
_VALID_THINKING = frozenset({"off", "low", "medium", "high", "max"})
_VALID_SESSION = frozenset({"fresh", "continue", "either"})
 
 
def _load_contracts() -> tuple[type, Any]:
    """Import ``PlanManifest`` + ``canonical_json`` from the M1.2 package.
 
    ``option_c_agent_contracts`` is installed editable as a monorepo sibling
    (Plan 151 T7/B1 promoted the contracts packages to ``contracts/``), so a
    plain import resolves the real package directly. The flat-layout sys.path
    bootstrap was retired when the shadowing project dir left the subproject
    root.
    """
    from option_c_agent_contracts import PlanManifest  # type: ignore
    from option_c_agent_contracts.canonical import canonical_json  # type: ignore
 
    return PlanManifest, canonical_json
 
 
def _manifests_dir(plans_dir: Path) -> Path:
    return plans_dir / MANIFESTS_DIRNAME
 
 
def _manifest_path(plans_dir: Path, plan_id: str) -> Path:
    return _manifests_dir(plans_dir) / f"{plan_id}{MANIFEST_SUFFIX}"
 
 
def _derive_assigned_agents(yaml_data: dict[str, object]) -> list[str]:
    """Derive ``PlanManifest.assigned_agents`` from plan frontmatter.
 
    Plan 222 ruling 2 (design §4.6 — refuse, never default): an ``agent:``
    value is either a registry-valid ``agent-NNN`` or a loud refusal naming
    the input. The legacy A-D letter mapping is GONE: a letter no longer
    resolves to ``agent_c`` or to anything else, because substituting an
    identity the operator did not attest is the exact failure §4.6 exists
    to prevent. An absent ``agent:`` key yields the empty list — a plan
    that has not been dispatched has no assigned instance, which asserts
    nothing false about who worked it.
    """
    func = "_derive_assigned_agents"
    from oc_orchestrator.agent_instances import validate_agent_id
 
    agent_identity = str(yaml_data.get("agent") or "").strip()
    if not agent_identity:
        _dbg("OC_CHAT.MAIN", func, "no agent in frontmatter; assigned_agents=[]", 3)
        return []
 
    # Raises ValueError naming the offending input (legacy letters included).
    validate_agent_id(agent_identity)
    _dbg("OC_CHAT.MAIN", func, f"assigned_agents=[{agent_identity}]", 3)
    return [agent_identity]
 
 
def _derive_requires(yaml_data: dict[str, object], plan_id: str) -> dict[str, str]:
    """Derive ``PlanManifest.requires`` from the frontmatter ``requires:`` block.
 
    Plan 222 ruling 3: the block is the authority and is mandatory — every
    active plan has carried one since 2026-08-13. A plan without it is a
    named refusal, consistent with dispatch, rather than a silent default.
 
    Only ``tier`` / ``thinking`` / ``session`` are lifted; sibling keys such
    as ``repo:`` belong to dispatch, and ``PlanRequirements`` is
    ``extra="forbid"``.
    """
    func = "_derive_requires"
    raw = yaml_data.get("requires")
    if not isinstance(raw, dict):
        raise ValueError(
            f"plan {plan_id} has no 'requires:' frontmatter block "
            "(expected tier/thinking/session); refusing to default it"
        )
 
    resolved: dict[str, str] = {}
    for key, valid in (
        ("tier", _VALID_TIER),
        ("thinking", _VALID_THINKING),
        ("session", _VALID_SESSION),
    ):
        supplied = raw.get(key)
        value = str(supplied or "").strip().lower()
        if value not in valid:
            raise ValueError(
                f"plan {plan_id} has invalid requires.{key} {supplied!r}; "
                f"expected one of {sorted(valid)}"
            )
        resolved[key] = value
 
    _dbg("OC_CHAT.MAIN", func, f"requires={resolved}", 3)
    return resolved
 
 
def _parse_plan_header(
    text: str,
    plan_id: str,
    *,
    attested_model: str | None = None,
    provider: PlanMetadataProvider | None = None,
) -> dict[str, Any]:
    """Derive header fields from a plan file's text.
 
    Supports YAML frontmatter (preferred — ``branch``/``worktree``/``agent``/
    ``requires`` keys) and the ``**Branch:**`` / ``**Worktree:**`` markdown
    form, with deterministic branch/worktree derivations so every numbered
    plan yields a structurally valid manifest. Reads only *text* (hermetic —
    honours a ``--plans-dir`` override), except for *attested_model*, which
    the caller resolves.
 
    Plan 223 P2b — the ``suggested_*`` re-keying. These three manifest fields
    used to read TOP-LEVEL ``model:`` / ``thinking:`` / ``session:``
    frontmatter keys that no modern plan carries, so every modern manifest
    came out internally inconsistent: ``suggested_thinking="medium"`` (the
    fallback) sitting beside ``requires.thinking="high"`` (the plan's actual
    declaration). ``thinking`` and ``session`` now read the ``requires:``
    block, which plan 222 made the mandatory authority, so the two halves of
    the manifest can no longer disagree. ``suggested_model`` has no
    plan-level source at all by design (D1: a plan declares a TIER, and the
    model is the operator's attestation at dispatch), so it takes the
    dispatch-attested value when the caller supplies one and stays
    ``"unknown"`` otherwise — never a guess derived from the tier.
    Legacy ``model:`` / ``**Suggested model:**`` headers are still read for
    historical plans that carry them.
    """
    meta = resolve_provider(provider).metadata_from_text(text)
    if meta.frontmatter_error is not None:
        # Re-raised rather than handled: a plan whose frontmatter is not
        # strict YAML has always surfaced here as a named derivation
        # failure, and the shared provider returns the error instead of
        # raising so that plan_status, which never parsed YAML at all,
        # keeps working on the same file.
        raise meta.frontmatter_error
    yaml_data: Mapping[str, object] = meta.frontmatter
 
    def _md(rx: re.Pattern[str]) -> str | None:
        match = rx.search(text)
        if not match:
            return None
        return match.group("value").strip().strip("`").strip()
 
    branch = meta.branch
    if not branch:
        stem = re.sub(r"^\d{3}-", "", plan_id)
        branch = f"feat/{stem}"
 
    worktree = meta.worktree
    if not worktree:
        slug = branch.split("/", 1)[-1].replace("-", "_")
        worktree = f"G:\\PythonApps\\OC_Chat_{slug}"
 
    assigned_agents = _derive_assigned_agents(yaml_data)
    requires = _derive_requires(yaml_data, plan_id)
 
    model = (
        str(attested_model or "").strip()
        or str(yaml_data.get("model") or _md(_MODEL_RE) or "").strip()
        or "unknown"
    )
 
    return {
        "branch": branch,
        "worktree": worktree,
        "assigned_agents": assigned_agents,
        "requires": requires,
        "suggested_model": model,
        "suggested_thinking": requires["thinking"],
        "suggested_session": requires["session"],
    }
 
 
def _derive_status(worktree: str) -> str:
    """Map the worktree's ``.PLAN_STATUS.md`` to a manifest status.
 
    Default ``pending`` (the registry spelling of "queued", per ADR-00Z)
    when no worktree or no status file exists.
    """
    plan_status = importlib.import_module("oc_orchestrator.plan_status")
    try:
        status_path = Path(worktree) / plan_status.DEFAULT_FILENAME
        if status_path.is_file():
            status_obj, _body = plan_status.read(status_path)
            return plan_status._registry_status(status_obj.status)
    except Exception:  # noqa: BLE001 - status derivation is best-effort
        pass
    return "pending"
 
 
def _attested_model(plan_id: str, agent_registry: str | None) -> str | None:
    """Return the model attested for *plan_id* at its latest dispatch.
 
    Plan 223 P2b. The runtime instance registry is the only place the
    operator's actual assignment is recorded, and reading it is OPT-IN via
    an explicit path so manifest derivation stays hermetic by default:
    ``validate-manifest`` compares an on-disk manifest against a live
    re-derivation, and a derivation that silently read mutable runtime state
    would report drift every time an instance changed state.
 
    Best-effort by design — an unreadable or absent registry yields ``None``
    and the manifest records ``"unknown"``, which is the truthful answer for
    a plan that has never been dispatched.
    """
    func = "_attested_model"
    if not agent_registry:
        return None
    try:
        from oc_orchestrator.agent_instances import AgentInstanceRegistry
 
        instances = [
            instance
            for instance in AgentInstanceRegistry(Path(agent_registry)).list()
            if instance.plan_id == plan_id
        ]
    except Exception:  # noqa: BLE001 - registry read is best-effort
        _dbg("OC_CHAT.MAIN", func, f"{plan_id}: registry unreadable", 3)
        return None
    if not instances:
        return None
    # ``list()`` is ordered by agent id, so the last entry is the most recent
    # instance to have served this plan (re-dispatches append).
    latest = instances[-1].model_attested.strip()
    _dbg("OC_CHAT.MAIN", func, f"{plan_id}: attested {latest!r}", 3)
    return latest or None
 
 
def _derive_manifest(
    plan_id: str,
    plans_dir: Path,
    *,
    agent_registry: str | None = None,
    paths: PlanPaths | None = None,
) -> Any:
    """Build a ``PlanManifest`` for *plan_id* from live plan + instance state.
 
    ``queue_position`` is always ``None``: the Markdown queue that supplied
    it retired with the queue machinery (plan 223 P3). The registry's own
    ``plans.queue_position`` column is untouched and still owns registry-side
    ordering; what died is this module's derivation of it from a Markdown
    file that no longer exists.
 
    *plans_dir* says where to READ the plan from and honours a
    ``--plans-dir`` override; *paths* says what the plan's body path IS,
    which is a stable relative identity recorded in the manifest and must
    not move when a test or a power user reads from somewhere else. They
    are two different questions and plan 270 P3 keeps them apart.
    """
    paths = paths or PlanPaths()
    plan_status = importlib.import_module("oc_orchestrator.plan_status")
    PlanManifest, _ = _load_contracts()
 
    plan_path = plans_dir / f"{plan_id}.md"
    if not plan_path.is_file():
        raise FileNotFoundError(f"plan file not found: {plan_path}")
 
    raw = plan_path.read_bytes()
    body_sha256 = hashlib.sha256(raw).hexdigest()
    header = _parse_plan_header(
        raw.decode("utf-8"),
        plan_id,
        attested_model=_attested_model(plan_id, agent_registry),
    )
 
    return PlanManifest(
        plan_id=plan_id,
        plan_number=plan_status._plan_number(plan_id),
        branch=header["branch"],
        worktree=header["worktree"],
        assigned_agents=header["assigned_agents"],
        requires=header["requires"],
        suggested_model=header["suggested_model"],
        suggested_thinking=header["suggested_thinking"],
        suggested_session=header["suggested_session"],
        status=_derive_status(header["worktree"]),
        # Plan 270 P3: read off PlanPaths rather than a hard-coded
        # consumer layout; the default is byte-identical to the literal
        # ``Path("oc_chat") / "plans" / f"{plan_id}.md"`` it replaces.
        body_path=paths.plan_body_relpath(plan_id).as_posix(),
        body_sha256=body_sha256,
        queue_position=None,
    )
 
 
def _serialize_manifest(manifest: Any) -> str:
    """Canonical-JSON the manifest (sorted keys, no whitespace; ADR-00X §3.4)."""
    _, canonical_json = _load_contracts()
    return canonical_json(manifest.model_dump(mode="json"))
 
 
def _atomic_write(path: Path, content: str) -> None:
    """Write *content* to *path* via a temp file + rename (atomic)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=path.parent,
        prefix=".manifest_",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(content)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
 
 
def _resolve_target_plan_ids(args: argparse.Namespace, plans_dir: Path) -> list[str]:
    """Return the plan-id stems a manifest command should act on."""
    if getattr(args, "all", False):
        ids: list[str] = []
        for path in sorted(plans_dir.glob("*.md")):
            if path.name in _EXCLUDED_PLAN_FILES:
                continue
            if not _PLAN_FILE_RE.match(path.name):
                continue
            ids.append(path.stem)
        return ids
    if getattr(args, "plan", None):
        return [args.plan]
    return []
 
 
def _cmd_init_manifest(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_init_manifest"
    _, plans_dir = _build_inventory(args)
    if not args.all and not args.plan:
        print("error: must specify --plan PLAN_ID or --all", file=sys.stderr)
        return 1
    plan_ids = _resolve_target_plan_ids(args, plans_dir)
    if not plan_ids:
        print("(no matching plan files)")
        return 0
 
    exit_code = 0
    written = 0
    for plan_id in plan_ids:
        try:
            manifest = _derive_manifest(
                plan_id, plans_dir, agent_registry=args.agent_registry
            )
        except FileNotFoundError as exc:
            print(f"{plan_id}: {exc}", file=sys.stderr)
            exit_code = 1
            continue
        except Exception as exc:  # noqa: BLE001 - report and continue under --all
            print(f"{plan_id}: cannot derive manifest: {exc}", file=sys.stderr)
            exit_code = 1
            continue
        content = _serialize_manifest(manifest)
        out_path = _manifest_path(plans_dir, plan_id)
        if out_path.exists():
            existing = out_path.read_text(encoding="utf-8")
            if existing == content:
                print(f"{plan_id}: up to date ({out_path})")
                continue
            if not args.force:
                print(
                    f"{plan_id}: divergent manifest exists; pass --force to "
                    f"overwrite ({out_path})",
                    file=sys.stderr,
                )
                exit_code = 1
                continue
        _atomic_write(out_path, content)
        written += 1
        print(f"{plan_id}: wrote {out_path}")
    _dbg("OC_CHAT.MAIN", func, f"init-manifest wrote {written}/{len(plan_ids)}", 2)
    return exit_code
 
 
def _validate_one(
    plan_id: str, plans_dir: Path, agent_registry: str | None
) -> tuple[int, list[str]]:
    """Validate one manifest. Returns ``(exit_code, report_lines)``.
 
    Codes: 0 valid + in sync, 2 missing/unreadable, 3 Pydantic failure,
    4 structurally valid but drifted from live state.
    """
    PlanManifest, _ = _load_contracts()
    manifest_path = _manifest_path(plans_dir, plan_id)
    if not manifest_path.is_file():
        return 2, [f"{plan_id}: manifest file missing ({manifest_path})"]
    try:
        raw = manifest_path.read_text(encoding="utf-8")
    except OSError as exc:
        return 2, [f"{plan_id}: cannot read manifest: {exc}"]
    try:
        manifest = PlanManifest.model_validate_json(raw)
    except Exception as exc:  # noqa: BLE001 - Pydantic ValidationError + JSON errors
        return 3, [f"{plan_id}: Pydantic validation failed: {exc}"]
    try:
        live = _derive_manifest(plan_id, plans_dir, agent_registry=agent_registry)
    except Exception as exc:  # noqa: BLE001 - missing/garbled live plan
        return 4, [f"{plan_id}: cannot re-derive live state: {exc}"]
 
    diffs: list[str] = []
    for field_name in PlanManifest.model_fields:
        on_disk = getattr(manifest, field_name)
        live_value = getattr(live, field_name)
        if on_disk != live_value:
            diffs.append(
                f"{plan_id}: {field_name}: manifest={on_disk!r} live={live_value!r}"
            )
    if diffs:
        return 4, diffs
    return 0, [f"{plan_id}: valid and matches live state"]
 
 
def _cmd_validate_manifest(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_validate_manifest"
    _, plans_dir = _build_inventory(args)
 
    if args.all:
        manifests_dir = _manifests_dir(plans_dir)
        manifest_files = (
            sorted(manifests_dir.glob(f"*{MANIFEST_SUFFIX}"))
            if manifests_dir.is_dir()
            else []
        )
        if not manifest_files:
            print("(no manifests found)")
            return 0
        worst = 0
        passed = 0
        for manifest_file in manifest_files:
            plan_id = manifest_file.name[: -len(MANIFEST_SUFFIX)]
            code, lines = _validate_one(plan_id, plans_dir, args.agent_registry)
            for line in lines:
                print(line)
            if code == 0:
                passed += 1
            else:
                worst = max(worst, code)
        print(f"\n{passed}/{len(manifest_files)} manifest(s) valid + in sync.")
        _dbg("OC_CHAT.MAIN", func, f"validate-all passed={passed} worst={worst}", 2)
        return worst
 
    if not args.plan:
        print("error: must specify --plan PLAN_ID or --all", file=sys.stderr)
        return 1
    code, lines = _validate_one(args.plan, plans_dir, args.agent_registry)
    for line in lines:
        print(line)
    _dbg("OC_CHAT.MAIN", func, f"validate {args.plan} -> {code}", 2)
    return code
 
 
def _cmd_list(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_list"
    inv, _ = _build_inventory(args)
    records = list_plans(inventory=inv)
    if args.status:
        records = [r for r in records if r.status == args.status]
    if not records:
        print("(no plans)")
        return 0
    for r in records:
        desc = r.description if len(r.description) <= 60 else r.description[:57] + "..."
        print(f"{r.plan_id:<35} {r.status:<9} {r.created_at}  {desc}")
    _dbg("OC_CHAT.MAIN", func, f"listed {len(records)} record(s)", 2)
    return 0
 
 
def _cmd_status(args: argparse.Namespace) -> int:
    inv, _ = _build_inventory(args)
    rec = find_plan_by_id(args.plan_id, inventory=inv)
    if rec is None:
        print(f"No plan found for {args.plan_id!r}", file=sys.stderr)
        return 1
    print(f"plan_id:      {rec.plan_id}")
    print(f"status:       {rec.status}")
    print(f"description:  {rec.description}")
    print(f"created_at:   {rec.created_at}")
    print(f"completed_at: {rec.completed_at}")
    print(f"aborted_at:   {rec.aborted_at}")
    print(f"session_name: {rec.session_name}")
    return 0
 
 
def _cmd_reconcile(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_reconcile"
    inv, plans_dir = _build_inventory(args)
    result = reconcile(inv, plans_dir)
    for renamed in result["renamed"]:
        print(f"renamed: {renamed}")
    for absorbed in result["orphans_absorbed"]:
        print(f"absorbed orphan: {absorbed}")
    for missing in result["missing_files"]:
        print(f"missing file for inventory record: {missing}")
    n_changes = (
        len(result["renamed"])
        + len(result["orphans_absorbed"])
        + len(result["missing_files"])
    )
    if n_changes == 0:
        print("(inventory and filesystem agree)")
    _dbg("OC_CHAT.MAIN", func, f"reconcile: {n_changes} report line(s)", 2)
    return 0
 
 
def _cmd_complete(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_complete"
    inv, plans_dir = _build_inventory(args)
    if inv.get(args.plan_id) is None:
        inv.register(args.plan_id, description="", session_name=None)
        _dbg("OC_CHAT.MAIN", func, f"registered orphan {args.plan_id!r}", 3)
    rec = inv.get(args.plan_id)
    if rec is None:
        print(f"unknown plan_id: {args.plan_id!r}", file=sys.stderr)
        return 1
    if rec.status == "aborted":
        print(
            f"plan {args.plan_id!r} is in terminal status 'aborted'; "
            f"cannot complete",
            file=sys.stderr,
        )
        return 1
    if rec.status == "complete":
        print(
            f"{rec.plan_id}: already complete "
            f"(completed_at={rec.completed_at})"
        )
        _dbg("OC_CHAT.MAIN", func, f"already complete: {args.plan_id!r}", 3)
        return 0
    rec = complete_plan(args.plan_id, inventory=inv, plans_dir=plans_dir)
    print(f"{rec.plan_id}: status={rec.status} completed_at={rec.completed_at}")
    _dbg("OC_CHAT.MAIN", func, f"marked complete: {args.plan_id!r}", 2)
    return 0
 
 
def _cmd_archive(args: argparse.Namespace) -> int:
    func = "plan_manage._cmd_archive"
    inv, plans_dir = _build_inventory(args)
 
    if args.all_complete and args.plan_id:
        print(
            "error: cannot combine --all-complete with a PLAN_ID positional",
            file=sys.stderr,
        )
        return 1
    if not args.all_complete and not args.plan_id:
        print(
            "error: must specify a PLAN_ID or pass --all-complete",
            file=sys.stderr,
        )
        return 1
 
    if args.all_complete:
        records = inv.list_by_status("complete")
    else:
        rec = inv.get(args.plan_id)
        if rec is None:
            print(f"unknown plan_id: {args.plan_id!r}", file=sys.stderr)
            return 1
        if rec.status != "complete":
            print(
                f"plan {args.plan_id!r} status is {rec.status!r}; "
                f"run 'complete' first",
                file=sys.stderr,
            )
            return 1
        records = [rec]
 
    old_dir = plans_dir / "old"
    old_dir.mkdir(parents=True, exist_ok=True)
 
    moved = 0
    for rec in records:
        src_name = f"{rec.plan_id}_complete.md"
        src = plans_dir / src_name
        dst = old_dir / src_name
        if dst.exists() and not src.exists():
            continue  # already archived — idempotent
        if not src.exists():
            print(f"missing source file: {src}", file=sys.stderr)
            continue
        src.rename(dst)
        print(f"archived: {src_name} -> old/")
        moved += 1
    print(f"({moved} archived)")
    _dbg("OC_CHAT.MAIN", func, f"archived {moved} file(s)", 2)
    return 0
 
 
def _find_repo_root(start: Path) -> Path:
    """Walk upward to the nearest git worktree root."""
    current = start.resolve()
    for candidate in (current, *current.parents):
        if (candidate / ".git").exists():
            return candidate
    raise ValueError(f"no git worktree found above {start}")
 
 
def _cmd_agents(args: argparse.Namespace) -> int:
    """Render the runtime agent-instance registry without mutating it."""
    from oc_orchestrator.agent_instances import (
        AgentInstanceRegistry,
        AgentRegistryError,
    )
 
    try:
        # The cwd is consulted only when nothing else names a registry
        # (plan 278 P2): an explicit --registry is an answer, and a view
        # that already has one has no business refusing because of where
        # the process happens to stand.
        if args.registry:
            registry_path = Path(args.registry).resolve()
        else:
            registry_path = (
                _find_repo_root(Path.cwd())
                / "oc_chat"
                / ".option_c"
                / "orchestration"
                / "agents.sqlite"
            )
        active_only = args.agents_view == "active"
        instances = AgentInstanceRegistry(registry_path).list(
            active_only=active_only
        )
    except (AgentRegistryError, OSError, ValueError) as exc:
        print(f"agents view refused: {exc}", file=sys.stderr)
        return 1
 
    if not instances:
        qualifier = "active " if active_only else ""
        print(f"(no {qualifier}agent instances)")
        return 0
 
    print(
        f"{'AGENT':<12} {'STATE':<14} {'PLAN':<30} "
        f"{'MODEL':<24} {'THINKING':<9} HARNESS"
    )
    for instance in instances:
        print(
            f"{instance.agent_id:<12} {instance.state:<14} "
            f"{(instance.plan_id or '-'):<30} "
            f"{instance.model_attested:<24} {instance.thinking:<9} "
            f"{instance.harness}"
        )
    return 0
 
 
def _cmd_prepare_worktree(args: argparse.Namespace) -> int:
    """Create a plan's branch, worktree and per-worktree uv venv (plan 241)."""
    func = "plan_manage._cmd_prepare_worktree"
    from oc_orchestrator.dispatch import DispatchError, resolve_plan_anchors
    from oc_orchestrator.worktree_env import (
        PrepareRequest,
        WorktreeEnvError,
        prepare_worktree,
    )
 
    try:
        repo_root = (
            Path(args.repo_root).resolve()
            if args.repo_root
            else _find_repo_root(Path.cwd())
        )
        plans_dir = Path(args.plans_dir or DEFAULT_PLANS_DIR).resolve()
        anchors = resolve_plan_anchors(plans_dir, args.plan_number)
        result = prepare_worktree(
            PrepareRequest(
                repo_root=repo_root,
                branch=anchors.branch,
                worktree=anchors.worktree,
                base_sha=args.base,
                uv_executable=args.uv,
                uv_cache_dir=Path(args.uv_cache_dir) if args.uv_cache_dir else None,
                smoke=bool(getattr(args, "smoke", False)),
            )
        )
    except (DispatchError, WorktreeEnvError, OSError, ValueError) as exc:
        print(f"prepare-worktree refused: {exc}", file=sys.stderr)
        return 1
    for line in result.report_lines():
        print(line)
    _dbg("OC_CHAT.MAIN", func, f"prepared {anchors.plan_id}", 2)
    return 0
 
 
#: Where ``teardown-worktree`` looks for a plan, in order (plan 243 P5).
#: The close ceremony moves a plan into ``completed/`` at step 7b and only
#: then removes the worktree at step 9, so resolving the active directory
#: alone refused exactly when the tool was wanted.
_TEARDOWN_PLAN_DIRS = ("plans/", "plans/completed/")
 
 
def _resolve_teardown_plan(plans_dir: Path, plan_number: str) -> tuple[Any, str]:
    """Resolve a plan's anchors from ``plans/``, then ``plans/completed/``.
 
    Returns the anchors and the label of the directory they came from, so
    the report line says which one answered. A plan in neither place keeps
    the canonical refusal (including "expected exactly three digits" for a
    malformed number) and names the completed directory as well.
    """
    from oc_orchestrator.dispatch import DispatchError, resolve_plan_anchors
 
    completed = plans_dir / "completed"
    for directory, label in zip((plans_dir, completed), _TEARDOWN_PLAN_DIRS):
        if sorted(directory.glob(f"{plan_number}-*.md")):
            return resolve_plan_anchors(directory, plan_number), label
    try:
        resolve_plan_anchors(plans_dir, plan_number)
    except DispatchError as exc:
        raise DispatchError(f"{exc}; {completed} holds no matching plan either") from exc
    raise DispatchError(
        f"plan number {plan_number} resolves in neither {plans_dir} nor {completed}"
    )
 
 
def _cmd_teardown_worktree(args: argparse.Namespace) -> int:
    """Remove a plan's worktree and venv, naming any residue (plan 241 P5)."""
    func = "plan_manage._cmd_teardown_worktree"
    from oc_orchestrator.dispatch import DispatchError
    from oc_orchestrator.worktree_env import WorktreeEnvError, teardown_worktree
 
    origin = None
    try:
        repo_root = (
            Path(args.repo_root).resolve()
            if args.repo_root
            else _find_repo_root(Path.cwd())
        )
        if args.worktree:
            worktree = Path(args.worktree)
            branch = None
        else:
            plans_dir = Path(args.plans_dir or DEFAULT_PLANS_DIR).resolve()
            anchors, origin = _resolve_teardown_plan(plans_dir, args.plan_number)
            worktree = anchors.worktree
            branch = anchors.branch
        result = teardown_worktree(repo_root, worktree, branch=branch)
    except (DispatchError, WorktreeEnvError, OSError, ValueError) as exc:
        print(f"teardown-worktree refused: {exc}", file=sys.stderr)
        return 1
    stream = sys.stdout if result.removed else sys.stderr
    if origin is not None:
        print(f"Resolved plan {args.plan_number} from {origin}", file=stream)
    for line in result.report_lines():
        print(line, file=stream)
    _dbg("OC_CHAT.MAIN", func, f"teardown removed={result.removed}", 2)
    return 0 if result.removed else 1
 
 
def _dispatch_hooks(repo_root: Path):
    """The hooks one dispatch announces through: none, by default.
 
    Plan 259 P6 made the listener something the CALLER registers:
    ``dispatch.py`` declares a Protocol and a ``NO_HOOKS`` default and
    names neither the hub nor the queue, exactly as it declares nothing
    about the registry. Plan 270 P3 moved the registration OUT of this
    module, along with the hub and the queue it named: this package has
    neither, so its default announces nothing and a standalone dispatch
    reaches no log at all.
 
    A consumer passes its own factory to :func:`main` as
    ``dispatch_hooks``. ``oc_chat``'s ``scripts/plan_manage.py`` shim
    passes the live hub hooks, so a dispatch there still records the
    ``dispatch`` message and the ``dequeue`` it always did.
    """
    func = "plan_manage._dispatch_hooks"
    from oc_orchestrator.dispatch import NO_HOOKS
 
    _dbg("OC_CHAT.MAIN", func, f"no dispatch hooks registered for {repo_root}", 4)
    return NO_HOOKS
 
 
def _cmd_dispatch(
    args: argparse.Namespace,
    *,
    dispatch_hooks: Callable[[Path], Any] | None = None,
) -> int:
    """Run the plan-221 one-shot dispatch workflow."""
    from oc_orchestrator.agent_instances import AgentRegistryError
    from oc_orchestrator.dispatch import (
        DispatchError,
        DispatchRequest,
        dispatch,
        format_dispatch_suggestion,
    )
 
    try:
        repo_root = (
            Path(args.repo_root).resolve()
            if args.repo_root
            else _find_repo_root(Path.cwd())
        )
        plans_dir = Path(args.plans_dir or DEFAULT_PLANS_DIR).resolve()
        registry_path = (
            Path(args.registry).resolve()
            if args.registry
            else repo_root / "oc_chat" / ".option_c" / "orchestration" / "agents.sqlite"
        )
        launch_dir = (
            Path(args.launch_dir).resolve()
            if args.launch_dir
            else repo_root / "oc_chat" / ".oc_chat" / "launch"
        )
        request = DispatchRequest(
            hooks=(dispatch_hooks or _dispatch_hooks)(repo_root),
            plan_number=args.plan_number,
            plans_dir=plans_dir,
            repo_root=repo_root,
            registry_path=registry_path,
            launch_dir=launch_dir,
            model_attested=args.model,
            thinking=args.thinking,
            harness=args.harness,
            template_path=(
                Path(args.template).resolve() if args.template else None
            ),
            again_note=args.again,
            shared_venv=args.shared_venv,
            uv_executable=args.uv,
            uv_cache_dir=Path(args.uv_cache_dir) if args.uv_cache_dir else None,
        )
        result = dispatch(request)
    except (AgentRegistryError, DispatchError, OSError, ValueError) as exc:
        print(f"dispatch refused: {exc}", file=sys.stderr)
        return 1
    for line in format_dispatch_suggestion(
        result,
        model_attested=args.model,
        registry_path=registry_path,
    ):
        print(line)
    return 0
 
 
def _add_uv_arguments(parser: argparse.ArgumentParser) -> None:
    """Attach the shared plan-241 uv knobs to a sub-parser."""
    parser.add_argument(
        "--uv",
        default="uv",
        help="uv executable name or path (refusal witnesses patch this)",
    )
    parser.add_argument(
        "--uv-cache-dir",
        help=(
            "uv cache directory; must be on the venv's drive (default: "
            "<repo root>/.uv-cache)"
        ),
    )
 
 
#: How this CLI names itself in ``--help``. A consumer that exposes the
#: tool under its own module path passes its own (plan 270 P4): the
#: shim's help output must stay byte-identical to what it printed before
#: the split, and ``prog`` is the only line in it that could differ.
DEFAULT_PROG = "python -m oc_orchestrator.plan_manage"
 
 
def main(
    argv: list[str] | None = None,
    *,
    prog: str = DEFAULT_PROG,
    dispatch_hooks: Callable[[Path], Any] | None = None,
) -> int:
    """Entry point. Returns an exit code (0 success, 1 failure).
 
    *prog* and *dispatch_hooks* are the consumer seams (plan 270 P4).
    Both default to this package standing alone: it names itself, and a
    dispatch announces itself to nobody. ``oc_chat``'s shim passes its
    own module path and the live hub hooks, which is what keeps every
    output, every hub entry and every queue transition identical across
    the split.
    """
    parser = argparse.ArgumentParser(
        prog=prog,
        description="Query and heal the OC Chat plan inventory.",
    )
    parser.add_argument(
        "--inventory",
        help=f"Path to inventory JSON (default: {DEFAULT_PLANS_DIR}/.inventory.json)",
    )
    parser.add_argument(
        "--plans-dir",
        help=f"Path to plans directory (default: {DEFAULT_PLANS_DIR})",
    )
    sub = parser.add_subparsers(dest="sub", required=True)
 
    p_list = sub.add_parser("list", help="List plans, optionally filtered by status")
    p_list.add_argument(
        "--status",
        choices=VALID_STATUSES,
        help="Filter to plans with this status only",
    )
 
    p_status = sub.add_parser("status", help="Show full record for one plan")
    p_status.add_argument("plan_id", help="Plan id or 3-digit prefix")
 
    sub.add_parser("reconcile", help="Heal disk/inventory drift")
 
    p_complete = sub.add_parser(
        "complete", help="Mark a plan complete and rename file to *_complete.md"
    )
    p_complete.add_argument(
        "plan_id",
        help="Plan id (full stem, e.g. 030-enhanced-terminal-ui-with-realtime-stats)",
    )
 
    p_archive = sub.add_parser(
        "archive", help="Move *_complete.md files to plans/old/"
    )
    p_archive.add_argument(
        "plan_id",
        nargs="?",
        help="Plan id to archive (omit when using --all-complete)",
    )
    p_archive.add_argument(
        "--all-complete",
        action="store_true",
        help="Archive every record with status='complete'",
    )
 
    p_agents = sub.add_parser(
        "agents", help="Inspect runtime agent instances"
    )
    agents_sub = p_agents.add_subparsers(dest="agents_view", required=True)
    for view, help_text in (
        ("list", "List every runtime agent instance"),
        ("active", "List non-closed runtime agent instances"),
    ):
        p_agents_view = agents_sub.add_parser(view, help=help_text)
        p_agents_view.add_argument(
            "--registry", help="agents.sqlite path override"
        )
 
    p_dispatch = sub.add_parser(
        "dispatch",
        help="Create/reuse git anchors, allocate an instance, and render one launch file",
    )
    p_dispatch.add_argument("plan_number", help="Three-digit operator plan handle")
    p_dispatch.add_argument(
        "--model", required=True, help="Operator-attested model assignment"
    )
    p_dispatch.add_argument(
        "--thinking",
        choices=["off", "low", "medium", "high", "max"],
        help="Override requires.thinking (defaults to the plan requirement)",
    )
    p_dispatch.add_argument(
        "--harness",
        default="other",
        choices=["claude-code", "oc_chat", "openai", "other"],
        help="Harness kind recorded in the runtime instance registry",
    )
    p_dispatch.add_argument(
        "--again",
        metavar="NOTE",
        help="Close the plan's dead open instance with NOTE and allocate a successor",
    )
    p_dispatch.add_argument("--repo-root", help="Git root override (tests/operators)")
    p_dispatch.add_argument("--registry", help="agents.sqlite path override")
    p_dispatch.add_argument("--launch-dir", help="Rendered launch directory override")
    p_dispatch.add_argument(
        "--template",
        help="Committed launch-template override (used by refusal witnesses)",
    )
    p_dispatch.add_argument(
        "--shared-venv",
        action="store_true",
        help=(
            "Render the legacy shared-venv brief and skip building the "
            "worktree's own uv environment (plan 241 P4)"
        ),
    )
    _add_uv_arguments(p_dispatch)
 
    p_prepare = sub.add_parser(
        "prepare-worktree",
        help=(
            "Create a plan's branch and worktree and build a uv venv inside "
            "it whose editable installs point at that worktree"
        ),
    )
    p_prepare.add_argument("plan_number", help="Three-digit operator plan handle")
    p_prepare.add_argument("--repo-root", help="Git root override (tests/operators)")
    p_prepare.add_argument(
        "--base",
        help=(
            "Commit the branch must start at; refuses when an existing branch "
            "sits elsewhere (default: reuse an existing branch, else master)"
        ),
    )
    p_prepare.add_argument(
        "--smoke",
        action="store_true",
        help=(
            "After verification, run the worker's own first commands inside "
            "the worktree (plan_status --help and the dev-engine imports) and "
            "refuse by name when one fails; dispatch always does (plan 243 P4)"
        ),
    )
    _add_uv_arguments(p_prepare)
 
    p_teardown = sub.add_parser(
        "teardown-worktree",
        help=(
            "Remove a plan's worktree and its venv, naming any residue; "
            "resolves the plan from plans/ and then plans/completed/"
        ),
    )
    p_teardown.add_argument("plan_number", help="Three-digit operator plan handle")
    p_teardown.add_argument("--repo-root", help="Git root override (tests/operators)")
    p_teardown.add_argument(
        "--worktree",
        help="Worktree path override (skips plan-frontmatter resolution)",
    )
 
    p_init_manifest = sub.add_parser(
        "init-manifest",
        help=(
            "Derive a PlanManifest from a plan file's header + queue state "
            "and write canonical JSON to plans/manifests/<plan_id>.manifest.json"
        ),
    )
    p_init_manifest.add_argument(
        "--plan", help="Plan id stem to derive (e.g. 093-m1-4-plan-manage-...)"
    )
    p_init_manifest.add_argument(
        "--all",
        action="store_true",
        help="Process every numbered plan file under the plans directory",
    )
    p_init_manifest.add_argument(
        "--force",
        action="store_true",
        help="Overwrite an existing manifest whose content differs",
    )
    p_init_manifest.add_argument(
        "--agent-registry",
        default=None,
        help=(
            "agents.sqlite path; when given, suggested_model takes the "
            "dispatch-attested value for this plan instead of 'unknown'"
        ),
    )
 
    p_validate_manifest = sub.add_parser(
        "validate-manifest",
        help=(
            "Validate a manifest with option_c_agent_contracts.PlanManifest "
            "and compare it field-by-field against live plan + queue state"
        ),
    )
    p_validate_manifest.add_argument(
        "--plan", help="Plan id stem to validate"
    )
    p_validate_manifest.add_argument(
        "--all",
        action="store_true",
        help="Validate every manifest on disk under plans/manifests/",
    )
    p_validate_manifest.add_argument(
        "--agent-registry",
        default=None,
        help=(
            "agents.sqlite path; when given, suggested_model takes the "
            "dispatch-attested value for this plan instead of 'unknown'"
        ),
    )
 
    args = parser.parse_args(argv)
    if args.sub == "list":
        return _cmd_list(args)
    if args.sub == "status":
        return _cmd_status(args)
    if args.sub == "reconcile":
        return _cmd_reconcile(args)
    if args.sub == "complete":
        return _cmd_complete(args)
    if args.sub == "archive":
        return _cmd_archive(args)
    if args.sub == "agents":
        return _cmd_agents(args)
    if args.sub == "dispatch":
        return _cmd_dispatch(args, dispatch_hooks=dispatch_hooks)
    if args.sub == "prepare-worktree":
        return _cmd_prepare_worktree(args)
    if args.sub == "teardown-worktree":
        return _cmd_teardown_worktree(args)
    if args.sub == "init-manifest":
        return _cmd_init_manifest(args)
    if args.sub == "validate-manifest":
        return _cmd_validate_manifest(args)
    parser.error(f"unknown sub-command: {args.sub}")
    return 1
 
 
if __name__ == "__main__":
    sys.exit(main())