plan_status.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/plan_status.py
 
# OCT-LINT: disable=tests_exist reason="Covered by tests/test_plan_status.py, tests/test_221_plan_status.py and tests/test_270_standalone_dispatch.py"
"""
Per-worktree ``.PLAN_STATUS.md`` lifecycle helper + CLI.
 
Purpose
-------
Read, validate, write, and transition the per-worktree status file
introduced in plan 062. The file lives at the worktree root, carries
YAML frontmatter + a Markdown body, and tracks a single plan's
lifecycle from ``Queued``/``Implementing`` through one of three
terminal states (``Completed`` / ``Partially Completed`` / ``Failed``).
 
Responsibilities
----------------
- Pydantic models for the schema v1.0: :class:`PlanStatus`,
  :class:`Progress`, :class:`AcceptanceChecks`, :class:`Status`.
- :func:`read` / :func:`write` — frontmatter + body round-trip with
  atomic writes (temp-then-rename).
- :func:`update` / :func:`transition` — mutator helpers that enforce
  the lifecycle state machine and timestamp invariants.
- ``init`` / ``set`` / ``show`` CLI subcommands for hand-driven use.
- ``report`` — the plan-231 worker-report lane writer: one JSON event in
  the comms bus carrying the worker's full closing summary verbatim.
 
The worker-report lane (plan 231, proposal V1 §5)
-------------------------------------------------
``report`` writes ONE file into ``<comms-root>/lanes/worker-reports/`` and
does nothing else. It holds no credential, opens no socket, and — by
design and by a structural test — **never imports oc_connect**: a worker
reaching AWAITING GATE must not need connector knowledge to be heard. The
orchestrator-side relay reads that lane and puts the summary on the
operator's phone; the event schema it validates against is documented in
``oc_connect/oc_connect/worker_reports.py``, which is the authority for
the contract these two sides share without sharing an import.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (read, write, transition)
    L3 - semantic details (invalid transitions, validation errors)
    L4 - deep tracing (per-field parse / emit)
 
Contracts
---------
- :func:`write` is atomic (temp file in the same dir, then ``replace``).
- :func:`transition` enforces the state machine: forward-only, no
  skipping ``Implementing``, terminal states are sinks. Raises
  :class:`ValueError` on illegal transitions.
- ``finished_at >= started_at >= queued_at`` when set; Pydantic
  validates the relationship at model construction time.
- Terminal-state body is recommended but not required at write time
  (workers may set status first and fill body in a follow-up call).
 
The observers (plan 257 P1, default inverted by plan 270 P3)
-----------------------------------------------------------
Every lifecycle event this tool records used to be mirrored into an
orchestration registry by code that lived right here, which made that
registry the dispatch layer's largest outbound edge. Plan 257 inverted
the direction: this module announces events through
:class:`oc_orchestrator.state_observer.StateObserver` and knows nothing
about who listens. Plan 270 inverted the DEFAULT to match, because the
registry belongs to a consumer and not to this package: passing no
observers now means no observer, and a consumer that wants the mirror
registers it. ``oc_chat``'s ``scripts/plan_status.py`` shim does exactly
that, so nothing about its behaviour changed.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.plan_inventory (PlanPaths -- where plan files live)
- oc_orchestrator.plan_metadata (the one plan-header reader)
- oc_orchestrator.state_observer (the plan-state observation contract)
- pydantic (v2 model validation)
- yaml (frontmatter parse/emit)
- oc_orchestrator.agent_instances (lazy; the default agent id check and live-instance lookup)
"""
from __future__ import annotations
 
import argparse
import hashlib
import json
import secrets
import sys
import tempfile
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Callable, Literal, Sequence
 
import yaml
from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.plan_inventory import PlanPaths
from oc_orchestrator.plan_metadata import (
    PlanMetadataProvider,
    resolve_provider,
    split_frontmatter,
)
from oc_orchestrator.state_observer import StateObserver
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
SCHEMA_VERSION = "2.0"
LEGACY_SCHEMA_VERSION = "1.0"
DEFAULT_FILENAME = ".PLAN_STATUS.md"
 
# ── Worker-report lane constants (plan 231; proposal V1 §3 and §5) ───────
# These mirror oc_connect's own values on purpose. The mirror is what buys
# the worker its credential-free, dependency-free write path; the drift risk
# it creates is paid for by a cross-implementation test that generates the
# same identifier from both sides and compares the bytes.
WORKSPACE_MARKER = ".option_c_workspace.json"
PROJECT_MARKER = ".option_c"
COMMS_ANCHOR = "oc_chat"
COMMS_SUBPATH = (".option_c", "comms")
WORKER_REPORT_LANE = "worker-reports"
WORKER_REPORT_SOURCE = "worker"
WORKER_REPORT_SCHEMA_VERSION = "1.0"
WORKER_REPORT_STATUSES = ("awaiting_gate", "blocked")
DEFAULT_REPORT_FILENAME = "report.md"
 
#: Crockford base32, the ULID alphabet: no I, L, O or U.
# OCT-LINT: disable=no_hardcoded_secrets reason="Public standard encoding alphabet; its 5.00 bits/char is the arithmetic maximum for any 32-distinct-character string and carries no credential. Operator-ruled 2026-09-01 (plan 230 W1 escalation), same constant as oc_connect.comms"
_CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
 
 
# ── Agent-identity seams (plan 257 P4) ───────────────────────────────────
#
# Two separate things wear the name "agent validation" here, and only one
# of them touches a store. The FORMAT check is a pure regex; the LIVE
# check opens the runtime instance registry. Both are injectable, so this
# module names ``oc_orchestrator.agent_instances`` only inside the two
# defaults, and a consumer with a roster instead of a registry can supply
# its own without touching the dispatch layer.
 
 
class AgentLookupError(RuntimeError):
    """A refusal from the injected live-instance lookup."""
 
 
#: ``(agent_registry_argument, agent_id) -> None``; raises to refuse.
AgentLookup = Callable[[str | None, str], None]
 
#: ``(value) -> canonical value``; raises :class:`ValueError` to refuse.
AgentIdValidator = Callable[[str], str]
 
 
def _default_agent_id_validator(value: str) -> str:
    """Today's format check: ``agent_instances.validate_agent_id``."""
    from oc_orchestrator.agent_instances import validate_agent_id
 
    return validate_agent_id(value)
 
 
def _default_agent_lookup(registry_arg: str | None, agent_id: str) -> None:
    """Today's live check: the instance must exist and not be closed.
 
    The refusal is re-raised as :class:`AgentLookupError` carrying the
    original message unchanged, so the caller's ``agent validation
    failed: {exc}`` line stays character-for-character what it was.
    Path resolution stays inside this default because it reads
    ``Path.cwd()`` at call time, and stays outside the ``try`` because
    ``default_agent_registry`` cannot raise the error being caught.
    """
    func = "plan_status._default_agent_lookup"
    from oc_orchestrator.agent_instances import (
        AgentInstanceRegistry,
        AgentRegistryError,
        default_agent_registry,
    )
 
    registry_path = (
        Path(registry_arg) if registry_arg else default_agent_registry(Path.cwd())
    )
    try:
        AgentInstanceRegistry(registry_path).require_live(agent_id)
    except AgentRegistryError as exc:
        # VALIDATION_ERROR: an unknown or closed instance is an expected,
        # handled refusal; the command reports it and returns 1.
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 3)
        raise AgentLookupError(str(exc)) from exc
 
 
#: Read at call time, never captured, so a substitution is honoured by
#: both the pydantic model validator and the argparse type.
_agent_id_validator: AgentIdValidator = _default_agent_id_validator
 
 
# ── Public types ─────────────────────────────────────────────────────────
 
 
class Status(str, Enum):
    """Lifecycle status of a plan in a worktree."""
 
    QUEUED = "Queued"
    IMPLEMENTING = "Implementing"
    COMPLETED = "Completed"
    BLOCKED = "Blocked"
    PARTIALLY_COMPLETED = "Partially Completed"
    FAILED = "Failed"
 
 
TERMINAL_STATUSES: frozenset[Status] = frozenset(
    {Status.COMPLETED, Status.PARTIALLY_COMPLETED, Status.FAILED}
)
 
 
_ALLOWED_TRANSITIONS: dict[Status | None, frozenset[Status]] = {
    None: frozenset({Status.QUEUED, Status.IMPLEMENTING}),
    Status.QUEUED: frozenset({Status.IMPLEMENTING}),
    Status.IMPLEMENTING: frozenset({Status.BLOCKED, *TERMINAL_STATUSES}),
    Status.BLOCKED: frozenset({Status.IMPLEMENTING, Status.COMPLETED}),
    Status.COMPLETED: frozenset(),
    Status.PARTIALLY_COMPLETED: frozenset(),
    Status.FAILED: frozenset(),
}
 
 
class Progress(BaseModel):
    model_config = ConfigDict(extra="forbid")
    steps_total: int | None = None
    steps_done: int | None = None
    current_step: str | None = None
 
 
class AcceptanceChecks(BaseModel):
    model_config = ConfigDict(extra="forbid")
    total: int
    passed: int
    failed: int
    remaining: int
 
 
class PlanStatus(BaseModel):
    """Validated `.PLAN_STATUS.md` frontmatter (schema v1.0)."""
 
    model_config = ConfigDict(extra="forbid", use_enum_values=False)
 
    schema_version: str = SCHEMA_VERSION
    plan_id: str
    branch: str
    worktree: str
    agent: str
    model: str
    thinking: Literal["off", "low", "medium", "high", "max"]
    session: Literal["fresh", "continue", "either"]
    status: Status
    queued_at: datetime | None = None
    started_at: datetime | None = None
    finished_at: datetime | None = None
    progress: Progress | None = None
    acceptance_checks: AcceptanceChecks | None = None
 
    @field_validator("schema_version")
    @classmethod
    def _check_schema_version(cls, value: str) -> str:
        if value not in {LEGACY_SCHEMA_VERSION, SCHEMA_VERSION}:
            raise ValueError(
                f"unsupported schema_version {value!r}; this helper reads "
                f"{LEGACY_SCHEMA_VERSION!r} and {SCHEMA_VERSION!r}"
            )
        return value
 
    @model_validator(mode="after")
    def _check_agent_identity(self) -> "PlanStatus":
        if self.schema_version == LEGACY_SCHEMA_VERSION:
            if not self.agent or not self.agent.isalpha() or not self.agent.isupper():
                raise ValueError(
                    "legacy schema agent must be uppercase letters, "
                    f"got {self.agent!r}"
                )
        else:
            try:
                _agent_id_validator(self.agent)
            except ValueError as exc:
                raise ValueError(str(exc)) from exc
        return self
 
    @model_validator(mode="after")
    def _check_timestamp_order(self) -> "PlanStatus":
        ordered = [
            ("queued_at", self.queued_at),
            ("started_at", self.started_at),
            ("finished_at", self.finished_at),
        ]
        prev_name, prev_ts = None, None
        for name, ts in ordered:
            if ts is None:
                continue
            if prev_ts is not None and ts < prev_ts:
                raise ValueError(
                    f"timestamp order violated: {name}={ts!r} precedes "
                    f"{prev_name}={prev_ts!r}"
                )
            prev_name, prev_ts = name, ts
        return self
 
 
# ── Frontmatter parse / emit ─────────────────────────────────────────────
 
 
def _now_local() -> datetime:
    # TZ-aware local datetime with explicit offset; emitted via
    # .isoformat() gives e.g. '2026-05-17T00:25:42+02:00'.
    # Per CHANGELOG-format.md §"Timestamps".
    return datetime.now().astimezone().replace(microsecond=0)
 
 
#: The ``---`` fence split now lives beside the other plan-header
#: parsing (plan 257 P2). The private name stays bound here because
#: ``scripts/plan_manage.py`` and ``registry/migrate.py`` both reach it
#: as ``plan_status._split_frontmatter``.
_split_frontmatter = split_frontmatter
 
 
def _emit_frontmatter(status: PlanStatus) -> str:
    """Emit the YAML frontmatter for *status*, preserving field order."""
    data = status.model_dump(mode="json")
    return yaml.safe_dump(data, sort_keys=False, default_flow_style=False)
 
 
# ── Public functions ─────────────────────────────────────────────────────
 
 
def read(path: Path) -> tuple[PlanStatus, str]:
    """Read ``.PLAN_STATUS.md`` and return ``(PlanStatus, markdown_body)``."""
    func = "plan_status.read"
    text = Path(path).read_text(encoding="utf-8")
    frontmatter, body = _split_frontmatter(text)
    data = yaml.safe_load(frontmatter)
    if not isinstance(data, dict):
        raise ValueError("frontmatter must be a YAML mapping")
    status = PlanStatus(**data)
    _dbg("OC_CHAT.MAIN", func, f"read {path} status={status.status.value}", 2)
    return status, body
 
 
def write(path: Path, status: PlanStatus, body: str = "") -> None:
    """Atomically write ``.PLAN_STATUS.md`` to *path*."""
    func = "plan_status.write"
    if status.schema_version != SCHEMA_VERSION:
        raise ValueError(
            f"schema {status.schema_version!r} is read-only; "
            f"new writes require {SCHEMA_VERSION!r}"
        )
    path = Path(path)
    frontmatter = _emit_frontmatter(status)
    payload = f"---\n{frontmatter}---\n\n{body.lstrip()}" if body else f"---\n{frontmatter}---\n"
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=path.parent,
        prefix=".plan_status_",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(payload)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
    _dbg("OC_CHAT.MAIN", func, f"wrote {path} ({len(payload)}B)", 2)
 
 
def update(path: Path, body: str | None = None, **fields: object) -> PlanStatus:
    """Read, apply *fields* updates, validate, write back. Returns the new status."""
    func = "plan_status.update"
    current, current_body = read(path)
    if "status" in fields:
        _validate_transition(current.status, Status(fields["status"]))
    data = current.model_dump(mode="json")
    if "progress" in fields and isinstance(fields["progress"], dict):
        progress_data = dict(data.get("progress") or {})
        progress_data.update(fields["progress"])
        data["progress"] = progress_data
        fields = {k: v for k, v in fields.items() if k != "progress"}
    data.update(fields)
    new_status = PlanStatus(**data)
    write(path, new_status, body if body is not None else current_body)
    _dbg("OC_CHAT.MAIN", func, f"updated {path}: {current.status.value}→{new_status.status.value}", 2)
    return new_status
 
 
def transition(path: Path, new_status: Status, body: str | None = None) -> PlanStatus:
    """Transition the status file at *path* to *new_status*.
 
    Sets ``started_at`` if leaving ``Queued`` (or if the file had no
    prior start timestamp), and ``finished_at`` on transitions into a
    terminal state. Raises :class:`ValueError` on illegal transitions.
    """
    func = "plan_status.transition"
    current, current_body = read(path)
    _validate_transition(current.status, new_status)
    data = current.model_dump(mode="json")
    data["status"] = new_status.value
    now_iso = _now_local().isoformat()
    if new_status == Status.IMPLEMENTING and current.started_at is None:
        data["started_at"] = now_iso
    if new_status in TERMINAL_STATUSES and current.finished_at is None:
        data["finished_at"] = now_iso
        if current.started_at is None and data.get("started_at") is None:
            data["started_at"] = now_iso
    new_status_obj = PlanStatus(**data)
    write(path, new_status_obj, body if body is not None else current_body)
    _dbg("OC_CHAT.MAIN", func, f"{current.status.value} → {new_status.value} at {path}", 2)
    return new_status_obj
 
 
def _validate_transition(current: Status | None, target: Status) -> None:
    func = "_validate_transition"
    allowed = _ALLOWED_TRANSITIONS.get(current, frozenset())
    if target not in allowed:
        _dbg("OC_CHAT.MAIN", func, f"illegal: {current} → {target}", 3)
        raise ValueError(
            f"illegal status transition: {current.value if current else 'None'} "
            f"→ {target.value}"
        )
 
 
# ── Worker-report lane (plan 231) ────────────────────────────────────────
 
 
class WorkerReportError(ValueError):
    """A worker-report event could not be written as specified."""
 
 
class EmptySummaryError(WorkerReportError):
    """The worker's closing summary was blank, so there is nothing to relay."""
 
 
class CommsRootNotFoundError(WorkerReportError):
    """No workspace or project marker was found above the starting path."""
 
 
def _encode(value: int, length: int) -> str:
    digits = []
    for _ in range(length):
        value, remainder = divmod(value, 32)
        digits.append(_CROCKFORD[remainder])
    return "".join(reversed(digits))
 
 
def new_ulid(now: datetime, *, entropy: bytes | None = None) -> str:
    """Return a 26-character ULID: 48 bits of time, 80 bits of randomness.
 
    Byte-for-byte the algorithm in ``oc_connect.comms.new_ulid``; see the
    module docstring for why the worker side re-implements rather than
    imports it. Time-first encoding makes lexical order arrival order,
    which is what lets the relay read the lane with a plain directory sort.
    """
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    milliseconds = int(now.astimezone(timezone.utc).timestamp() * 1000)
    raw = entropy if entropy is not None else secrets.token_bytes(10)
    if len(raw) != 10:
        raise ValueError("ULID entropy must be exactly ten bytes")
    return _encode(milliseconds, 10) + _encode(int.from_bytes(raw, "big"), 16)
 
 
def utc_stamp(now: datetime) -> str:
    """Return a filename-safe UTC stamp; colons are illegal on Windows."""
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    return now.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
 
 
def _utc_timestamp(now: datetime) -> str:
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    return now.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
def find_comms_root(start: Path) -> Path:
    """Return the workspace-anchored comms-bus root above *start*.
 
    The rules are ``oc_connect.workspace.comms_root``'s, restated: the
    workspace marker wins over a nearer project marker because connector
    state is workspace-wide, and the bus sits beside the dev engine's
    orchestration state when an ``oc_chat`` subproject is present. Guessing
    a root would post a worker's report somewhere nobody is listening, so a
    failed walk raises instead.
    """
    origin = Path(start).expanduser().resolve()
    project_fallback: Path | None = None
    anchor: Path | None = None
    for parent in (origin, *origin.parents):
        if (parent / WORKSPACE_MARKER).is_file():
            anchor = parent
            break
        if project_fallback is None and (parent / PROJECT_MARKER).is_dir():
            project_fallback = parent
    if anchor is None:
        anchor = project_fallback
    if anchor is None:
        raise CommsRootNotFoundError(
            f"no Option C workspace marker ({WORKSPACE_MARKER}) or project "
            f"marker ({PROJECT_MARKER}/) was found at or above {origin}"
        )
    if (anchor / COMMS_ANCHOR).is_dir():
        return anchor.joinpath(COMMS_ANCHOR, *COMMS_SUBPATH)
    return anchor.joinpath(*COMMS_SUBPATH)
 
 
def worker_report_event(
    *,
    plan_id: str,
    agent: str,
    status: str,
    worktree: Path,
    report_path: Path,
    summary: str,
    now: datetime,
    entropy: bytes | None = None,
) -> dict[str, object]:
    """Build one lane event; refuse by name when the summary is empty.
 
    An empty summary is refused rather than defaulted because the whole
    point of the lane is that the operator reads the worker's own words: a
    report event with nothing in it would announce a gate and then say
    nothing, which is worse than no event at all.
    """
    if status not in WORKER_REPORT_STATUSES:
        raise WorkerReportError(
            f"worker report status must be one of "
            f"{', '.join(WORKER_REPORT_STATUSES)}; got {status!r}"
        )
    if not summary.strip():
        raise EmptySummaryError(
            "refusing to write a worker report event with an empty summary; "
            "the lane exists to carry the worker's own closing words verbatim"
        )
    return {
        "schema_version": WORKER_REPORT_SCHEMA_VERSION,
        "event_id": new_ulid(now, entropy=entropy),
        "lane": WORKER_REPORT_LANE,
        "direction": "inbound",
        "risk_class": "low",
        "plan_id": plan_id,
        "agent": agent,
        "status": status,
        "worktree": str(worktree),
        "report_path": str(report_path),
        "created_at_utc": _utc_timestamp(now),
        "summary": summary,
        "summary_sha256": hashlib.sha256(summary.encode("utf-8")).hexdigest(),
        "relayed_at_utc": None,
        "relay_chunk_count": None,
    }
 
 
def write_worker_report(
    comms_root: Path,
    event: dict[str, object],
    *,
    now: datetime,
) -> Path:
    """Write one event into ``lanes/worker-reports/`` and return its path.
 
    The lane directory is created when absent: a worker must be able to
    report from a workspace where the bus has never been initialised.
    Nothing outside the lane is created, read, or written.
    """
    func = "plan_status.write_worker_report"
    lane_dir = Path(comms_root) / "lanes" / WORKER_REPORT_LANE
    lane_dir.mkdir(parents=True, exist_ok=True)
    name = f"{utc_stamp(now)}_{WORKER_REPORT_SOURCE}_{event['event_id']}.json"
    path = lane_dir / name
    if path.exists():
        raise WorkerReportError(
            f"refusing to overwrite an existing worker report event at {path}"
        )
    payload = json.dumps(event, indent=2, sort_keys=True) + "\n"
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        newline="\n",
        dir=lane_dir,
        prefix=".worker_report_",
        suffix=".tmp",
        delete=False,
    ) as tmp:
        tmp.write(payload)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
    _dbg("OC_CHAT.MAIN", func, f"wrote worker report event {path.name}", 2)
    return path
 
 
# ── CLI ──────────────────────────────────────────────────────────────────
 
 
def _canonical_agent_id(value: str) -> str:
    """Argparse type for the registry-allocated ``agent-NNN`` form."""
    try:
        return _agent_id_validator(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(str(exc)) from exc
 
 
def _resolve_path(args: argparse.Namespace) -> Path:
    """Where the status file lives: ``--path`` > ``--worktree-path`` >
    the recorded absolute ``--worktree`` > cwd.
 
    The third rung is the 2026-09-06 correction: the launch brief runs
    workers from ``<worktree>/oc_chat`` and passes ``--worktree`` (the
    field recorded INSIDE the file), so before this the file landed one
    level below the worktree root where the gate reads it, and four plans
    were wrongly recorded as having skipped STEP-0.
    """
    if getattr(args, "path", None):
        return Path(args.path)
    explicit = getattr(args, "worktree_path", None)
    if explicit:
        return Path(explicit) / DEFAULT_FILENAME
    recorded = getattr(args, "worktree", None)
    if recorded and Path(recorded).is_absolute():
        return Path(recorded) / DEFAULT_FILENAME
    return Path.cwd() / DEFAULT_FILENAME
 
 
def _resolve_worktree_field(args: argparse.Namespace) -> Path:
    raw = getattr(args, "worktree", None)
    if raw:
        candidate = Path(raw)
        if candidate.is_absolute():
            return candidate.resolve()
    return Path(getattr(args, "worktree_path", None) or Path.cwd()).resolve()
 
 
def _registry_status(status: Status) -> str:
    return {
        Status.QUEUED: "pending",
        Status.IMPLEMENTING: "implementing",
        Status.BLOCKED: "implementing",
        Status.COMPLETED: "ready_for_merge",
        Status.PARTIALLY_COMPLETED: "partial",
        Status.FAILED: "failed",
    }[status]
 
 
def _plan_number(plan_id: str) -> int:
    prefix = plan_id.split("-", 1)[0]
    try:
        value = int(prefix)
    except ValueError as exc:
        raise ValueError(f"plan id {plan_id!r} does not start with a number") from exc
    if value <= 0:
        raise ValueError(f"plan id {plan_id!r} does not start with a positive number")
    return value
 
 
def _plan_metadata(
    plan_id: str,
    worktree: Path,
    *,
    provider: PlanMetadataProvider | None = None,
    paths: PlanPaths | None = None,
) -> dict[str, str | None]:
    """Title, type and body hash for the plan file *plan_id* names.
 
    The header fields come from the shared provider (plan 257 P2); the
    body hash stays here because it is a property of the bytes on disk,
    not of the header. The bytes are read once and reused for both.
 
    Plan 270 P3: where the plan file lives is read off :class:`PlanPaths`
    rather than from a ``Path("oc_chat") / "plans"`` literal, because a
    standalone package cannot hard-code a consumer's layout. The default
    is byte-identical to that literal.
    """
    func = "plan_status._plan_metadata"
    relative_path = (paths or PlanPaths()).plan_body_relpath(plan_id)
    plan_path = worktree / relative_path
    if not plan_path.is_file():
        _dbg("OC_CHAT.MAIN", func, f"no plan file at {plan_path}", 3)
        return {
            "title": plan_id,
            "body_path": relative_path.as_posix(),
            "body_sha256": hashlib.sha256(b"").hexdigest(),
            "plan_type": None,
        }
    body = plan_path.read_bytes()
    text = body.decode("utf-8", errors="replace")
    meta = resolve_provider(provider).metadata_from_text(text, plan_id=plan_id)
    return {
        "title": meta.title or plan_id,
        "body_path": relative_path.as_posix(),
        "body_sha256": hashlib.sha256(body).hexdigest(),
        "plan_type": meta.plan_type,
    }
 
 
def _resolve_observers(
    observers: Sequence[StateObserver] | None,
) -> tuple[StateObserver, ...]:
    """The observers to notify for one operation.
 
    Plan 257 P1 made this the seam; plan 270 P3 inverted its default.
    ``None`` and an empty sequence now mean the same thing -- write the
    status file and reach nothing else -- because a standalone package
    has no registry to mirror into and must not invent one. The
    consumer that HAS a registry passes its observer in; ``oc_chat``'s
    ``scripts/plan_status.py`` shim passes ``RegistryMirrorObserver()``
    on every command, so its behaviour is unchanged.
 
    This is the line plan 257 called "the one line a split replaces",
    and this is the split that replaced it: nothing in this package
    names a registry any more.
    """
    func = "plan_status._resolve_observers"
    if observers is None:
        _dbg("OC_CHAT.MAIN", func, "no observers registered", 4)
        return ()
    return tuple(observers)
 
 
def _notify(
    observers: Sequence[StateObserver] | None,
    emit: Callable[[StateObserver], int],
) -> int:
    """Run *emit* over every observer; the first non-zero code wins.
 
    A consumer typically registers exactly one observer, so this
    reproduces the single return code the pre-257 ``_mirror_*`` helpers
    produced. A non-zero code aborts the command, exactly as the
    mirror's ``2`` did. With no observers registered there is nothing to
    run and the answer is ``0``: writing the status file succeeded and
    nobody asked to be told.
    """
    func = "plan_status._notify"
    for observer in _resolve_observers(observers):
        code = emit(observer)
        if code:
            _dbg("OC_CHAT.MAIN", func, f"observer reported code {code}", 2)
            return code
    return 0
 
 
def _body_note(body: str | None, fallback: str) -> str:
    if body:
        for line in body.splitlines():
            stripped = line.strip()
            if stripped:
                return stripped
    return fallback
 
 
def _status_observed_at(status: PlanStatus) -> str | None:
    for value in (status.finished_at, status.started_at, status.queued_at):
        if value is not None:
            return value.isoformat()
    return None
 
 
def _mirror_init(
    status: PlanStatus,
    *,
    observers: Sequence[StateObserver] | None = None,
) -> int:
    """Announce that *status* was initialised for its plan."""
    target = _registry_status(status.status)
    worktree = Path(status.worktree)
    metadata = _plan_metadata(status.plan_id, worktree)
    return _notify(
        observers,
        lambda observer: observer.plan_created(
            plan_id=status.plan_id,
            title=str(metadata["title"]),
            plan_type=metadata["plan_type"],
            body_path=str(metadata["body_path"]),
            body_sha256=str(metadata["body_sha256"]),
            body_observed_at=_status_observed_at(status),
            model=status.model,
            thinking=status.thinking,
            session=status.session,
            branch=status.branch,
            worktree=status.worktree,
            target_status=target,
        ),
    )
 
 
def _mirror_transition(
    previous: Status,
    current: PlanStatus,
    body: str | None,
    *,
    observers: Sequence[StateObserver] | None = None,
) -> int:
    """Announce that the plan moved from *previous* to *current*.
 
    A transition that does not change the REGISTRY-side spelling of the
    status is not an event: ``Implementing`` and ``Blocked`` both map to
    ``implementing``, so no observer is notified and nothing is reached.
    """
    target = _registry_status(current.status)
    if target == _registry_status(previous):
        return 0
    note = None
    if target in {"partial", "failed"}:
        note = _body_note(body, f"legacy status transition to {current.status.value}")
    return _notify(
        observers,
        lambda observer: observer.status_changed(
            plan_id=current.plan_id,
            from_status=_registry_status(previous),
            to_status=target,
            note=note,
        ),
    )
 
 
def _cmd_init(
    args: argparse.Namespace,
    *,
    observers: Sequence[StateObserver] | None = None,
    lookup: AgentLookup | None = None,
) -> int:
    func = "plan_status._cmd_init"
    path = _resolve_path(args)
    target_status = Status(args.status)
    try:
        (lookup or _default_agent_lookup)(args.agent_registry, args.agent)
    except AgentLookupError as exc:
        print(f"agent validation failed: {exc}", file=sys.stderr)
        return 1
    now_iso = _now_local().isoformat()
    fields = dict(
        schema_version=SCHEMA_VERSION,
        plan_id=args.plan,
        branch=args.branch,
        worktree=str(_resolve_worktree_field(args)),
        agent=args.agent,
        model=args.model,
        thinking=args.thinking,
        session=args.session,
        status=target_status.value,
    )
    if target_status == Status.QUEUED:
        fields["queued_at"] = now_iso
    elif target_status == Status.IMPLEMENTING:
        fields["started_at"] = now_iso
    status = PlanStatus(**fields)
    write(path, status)
    mirror_rc = _mirror_init(status, observers=observers)
    if mirror_rc:
        return mirror_rc
    print(f"Initialised {path} (status={target_status.value}).")
    _dbg("OC_CHAT.MAIN", func, f"init {path} ({target_status.value})", 2)
    return 0
 
 
def _cmd_set(
    args: argparse.Namespace,
    *,
    observers: Sequence[StateObserver] | None = None,
) -> int:
    func = "plan_status._cmd_set"
    path = _resolve_path(args)
    body: str | None = None
    if args.body_file:
        body = Path(args.body_file).read_text(encoding="utf-8")
    if args.status:
        previous, _previous_body = read(path)
        result = transition(path, Status(args.status), body=body)
        mirror_rc = _mirror_transition(
            previous.status, result, body, observers=observers
        )
        if mirror_rc:
            return mirror_rc
        print(f"{path}: status -> {result.status.value}")
    progress_updates: dict[str, object] = {}
    if args.current_step is not None:
        progress_updates["current_step"] = args.current_step
    if args.steps_done is not None:
        progress_updates["steps_done"] = args.steps_done
    if args.steps_total is not None:
        progress_updates["steps_total"] = args.steps_total
    if progress_updates:
        result = update(path, progress=progress_updates)
        print(f"{path}: progress -> {result.progress}")
    if not args.status and not progress_updates and body is not None:
        update(path, body=body)
        print(f"{path}: body updated")
    if args.status and result.status is Status.BLOCKED:
        # Blocked is the moment the operator must act, so it is heard on the
        # worker-report lane without the worker knowing the lane exists
        # (operator ruling 2026-09-04: a Blocked worker was silent because the
        # lane only fired at AWAITING GATE). Fail-SOFT by design: recording
        # Blocked must never depend on the comms bus being reachable.
        _emit_blocked_event(args, result, body)
    _dbg("OC_CHAT.MAIN", func, f"set {path}", 2)
    return 0
 
 
def _emit_blocked_event(
    args: argparse.Namespace, current: PlanStatus, body: str | None
) -> None:
    """Write one ``blocked`` lane event for *current*; warn, never fail."""
    func = "plan_status._emit_blocked_event"
    # Only a reason given ON THIS transition counts: a stale progress step from
    # an earlier call is not why the worker is blocked now.
    step = getattr(args, "current_step", None)
    summary = (
        body
        if body and body.strip()
        else step
        if step and step.strip()
        else f"{current.plan_id} entered Blocked; no reason was given on the transition."
    )
    worktree = Path(current.worktree)
    now = datetime.now(timezone.utc).replace(microsecond=0)
    try:
        event = worker_report_event(
            plan_id=current.plan_id,
            agent=current.agent,
            status="blocked",
            worktree=worktree,
            report_path=worktree / DEFAULT_REPORT_FILENAME,
            summary=summary,
            now=now,
        )
        comms_root = (
            Path(args.comms_root).expanduser().resolve()
            if getattr(args, "comms_root", None)
            else find_comms_root(worktree)
        )
        written = write_worker_report(comms_root, event, now=now)
    except (WorkerReportError, OSError) as exc:
        print(
            f"warning: worker report for Blocked not written ({exc}); "
            f"the status file is updated regardless",
            file=sys.stderr,
        )
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: blocked event not written: {exc}", 2)
        return
    print(f"Wrote worker report {event['event_id']} (Blocked) to {written}.")
    _dbg("OC_CHAT.MAIN", func, f"blocked event {current.plan_id} -> {written}", 2)
 
 
def _cmd_show(args: argparse.Namespace) -> int:
    func = "plan_status._cmd_show"
    path = _resolve_path(args)
    status, _body = read(path)
    progress_text = "?/?"
    if status.progress is not None:
        done = status.progress.steps_done if status.progress.steps_done is not None else "?"
        total = status.progress.steps_total if status.progress.steps_total is not None else "?"
        progress_text = f"{done}/{total}"
    started_text = status.started_at.isoformat() if status.started_at else "(not started)"
    print(f"{status.plan_id} | {status.status.value} | {progress_text} | started {started_text}")
    _dbg("OC_CHAT.MAIN", func, f"show {path}", 2)
    return 0
 
 
def _resolve_summary(args: argparse.Namespace) -> str:
    if args.summary_file:
        return Path(args.summary_file).read_text(encoding="utf-8")
    return args.summary or ""
 
 
def _cmd_report(args: argparse.Namespace) -> int:
    func = "plan_status._cmd_report"
    worktree = _resolve_worktree_field(args)
    report_path = (
        Path(args.report_path).expanduser().resolve()
        if args.report_path
        else worktree / DEFAULT_REPORT_FILENAME
    )
    now = datetime.now(timezone.utc).replace(microsecond=0)
    try:
        summary = _resolve_summary(args)
        event = worker_report_event(
            plan_id=args.plan,
            agent=args.agent,
            status=args.status,
            worktree=worktree,
            report_path=report_path,
            summary=summary,
            now=now,
        )
        comms_root = (
            Path(args.comms_root).expanduser().resolve()
            if args.comms_root
            else find_comms_root(worktree)
        )
        path = write_worker_report(comms_root, event, now=now)
    except (WorkerReportError, OSError) as exc:
        print(f"worker report refused: {exc}", file=sys.stderr)
        return 2
    print(
        f"Wrote worker report {event['event_id']} "
        f"({args.plan}, {args.agent}, {args.status}) to {path}."
    )
    _dbg("OC_CHAT.MAIN", func, f"report {args.plan} status={args.status}", 2)
    return 0
 
 
#: 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_status"
 
 
def main(
    argv: list[str] | None = None,
    *,
    prog: str = DEFAULT_PROG,
    observers: Sequence[StateObserver] | None = None,
) -> int:
    """Run one sub-command. 0 on success, non-zero on a named refusal.
 
    *prog* and *observers* are the consumer seams (plan 270 P4). Both
    default to this package standing alone: it names itself, and it
    tells nobody. ``oc_chat``'s shim passes its own module path and its
    registry mirror, which is what keeps every output and every registry
    row identical across the split.
    """
    func = "plan_status.main"
    parser = argparse.ArgumentParser(
        prog=prog,
        # OCT-LINT: disable=no_hardcoded_secrets reason="Status enum values, model IDs, and CLI choice strings are not secrets"
        description="Read/write/transition .PLAN_STATUS.md lifecycle files.",
    )
    sub = parser.add_subparsers(dest="sub", required=True)
 
    p_init = sub.add_parser("init", help="Initialise a fresh .PLAN_STATUS.md")
    p_init.add_argument("--plan", required=True, help="Plan id (filename stem)")
    p_init.add_argument("--branch", required=True, help="Git branch name")
    p_init.add_argument(
        "--worktree",
        help=(
            "Worktree absolute path written into the YAML field. "
            "Defaults to the resolved --worktree-path (cwd). "
            "Relative or mangled values are ignored — the cwd fallback is used "
            "instead so markdown-stripped paths from launch prompts don't poison the field."
        ),
    )
    p_init.add_argument(
        "--agent",
        required=True,
        type=_canonical_agent_id,
        metavar="AGENT_ID",
        help=(
            "Registry-allocated instance identity (for example agent-001). "
            "The instance must exist and must not be closed."
        ),
    )
    p_init.add_argument(
        "--agent-registry",
        help=(
            "agents.sqlite path used to validate --agent "
            "(defaults to .option_c/orchestration/agents.sqlite from cwd)"
        ),
    )
    p_init.add_argument("--model", required=True, help="Provider-native model id")
    p_init.add_argument("--thinking", required=True, choices=["off", "low", "medium", "high", "max"])
    p_init.add_argument("--session", required=True, choices=["fresh", "continue", "either"])
    p_init.add_argument("--status", default="Implementing", choices=[s.value for s in Status])
    p_init.add_argument("--path", help="Override target .PLAN_STATUS.md path")
    p_init.add_argument("--worktree-path", help="Worktree directory (default: cwd)")
 
    p_set = sub.add_parser("set", help="Update status / progress / body of an existing file")
    p_set.add_argument("--status", choices=[s.value for s in Status])
    p_set.add_argument("--current-step", help="Set progress.current_step")
    p_set.add_argument("--steps-done", type=int, help="Set progress.steps_done")
    p_set.add_argument("--steps-total", type=int, help="Set progress.steps_total")
    p_set.add_argument("--body-file", help="Read Markdown body from FILE")
    p_set.add_argument("--path", help="Override target .PLAN_STATUS.md path")
    p_set.add_argument("--worktree-path", help="Worktree directory (default: cwd)")
    p_set.add_argument(
        "--worktree",
        help="The recorded worktree root; locates the status file when neither "
             "--path nor --worktree-path is given (the launch brief runs you from "
             "<worktree>/oc_chat)",
    )
    p_set.add_argument(
        "--comms-root",
        help=(
            "Comms-bus root for the Blocked lane event (default: discovered from "
            "the status file's worktree); a Blocked transition emits one event"
        ),
    )
 
    p_show = sub.add_parser("show", help="Print a one-line summary of the status file")
    p_show.add_argument("--path", help="Override target .PLAN_STATUS.md path")
    p_show.add_argument("--worktree-path", help="Worktree directory (default: cwd)")
    p_show.add_argument(
        "--worktree",
        help="The recorded worktree root; locates the status file when neither "
             "--path nor --worktree-path is given",
    )
 
    p_report = sub.add_parser(
        "report",
        help="Post this worker's closing summary to the comms worker-report lane",
    )
    p_report.add_argument("--plan", required=True, help="Plan id (filename stem)")
    p_report.add_argument(
        "--agent",
        required=True,
        type=_canonical_agent_id,
        metavar="AGENT_ID",
        help="Registry-allocated instance identity (for example agent-001)",
    )
    p_report.add_argument(
        "--status",
        required=True,
        choices=list(WORKER_REPORT_STATUSES),
        help="Both statuses emit an event (operator ruling, proposal V1 §10.3)",
    )
    summary_source = p_report.add_mutually_exclusive_group(required=True)
    summary_source.add_argument(
        "--summary-file",
        help="Read the FULL closing summary from FILE (normally report.md)",
    )
    summary_source.add_argument(
        "--summary",
        help="The FULL closing summary as a literal argument",
    )
    p_report.add_argument(
        "--report-path",
        help=(
            "Absolute path of report.md in this worktree "
            f"(default: <worktree>/{DEFAULT_REPORT_FILENAME})"
        ),
    )
    p_report.add_argument(
        "--worktree",
        help="Worktree absolute path recorded in the event (default: cwd)",
    )
    p_report.add_argument("--worktree-path", help="Worktree directory (default: cwd)")
    p_report.add_argument(
        "--comms-root",
        help=(
            "Comms bus root to write into. Defaults to the workspace-anchored "
            "root discovered above the worktree; pass it explicitly when the "
            "worktree's own workspace marker is not the bus the orchestrator reads."
        ),
    )
 
    args = parser.parse_args(argv)
    _dbg("OC_CHAT.MAIN", func, f"dispatch sub-command: {args.sub}", 2)
    if args.sub == "init":
        return _cmd_init(args, observers=observers)
    if args.sub == "set":
        return _cmd_set(args, observers=observers)
    if args.sub == "show":
        return _cmd_show(args)
    if args.sub == "report":
        return _cmd_report(args)
    parser.error(f"unknown sub-command: {args.sub}")
    return 1
 
 
if __name__ == "__main__":
    sys.exit(main())