work_queue_manage.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/work_queue_manage.py
 
# OCT-LINT: disable=tests_exist reason="Covered by tests/test_258_queue_manage_cli.py and tests/test_271_standalone_queue.py"
"""
The queue's operator surface: list, add, move, dequeue, complete, next.
 
Purpose
-------
Plan 258 P4. The queue CLI plan 223 retired is revived in the SHAPE
that 140 dispatches taught it, and only in that shape. What comes back
is ``list``, ``add``, ``move``, ``complete``, ``health``, ``reconcile``
and ``history``, plus ``dequeue`` and ``next``. What does NOT come back
is everything the letter model or the dispatcher now owns:
 
``launch``
    ``core/dispatch.py`` renders launch files.
``prep-next``
    ``next`` prints what the operator would dispatch; ``dispatch.ps1``
    and ``prepare_worktree`` create the branch, the worktree and its
    venv. This command never creates either, and never runs anything.
``status-scan``
    ``plan_manage agents active`` and the per-worktree
    ``.PLAN_STATUS.md`` files cover it.
``auto-assign`` / ``add-agent`` / ``list-agents``
    The roster, the letters, the pools and the Plan-Type suggesters are
    retired by operator ruling and are not revived.
 
**This command dispatches nothing.** The autonomy boundary in the
factory-decomposition design note §6 reserves the close and every
ruling for the human, and automated dispatch is a later plan's work.
``next`` prints the line an operator would run; that is the whole of
it, and this module does not import ``subprocess`` so it could not do
more if it tried.
 
Responsibilities
----------------
- :func:`build_parser` -- the command shape, exposed so a witness can
  read the defaults off it.
- :func:`main` -- argument dispatch and the exit-code policy. It carries
  the two consumer seams (plan 271 P2): ``prog``, the name the CLI gives
  itself in ``--help``, and ``backend_factory``, which builds the
  ``--backend registry`` backend. Both default to this package standing
  alone: it names itself, and it has no registry to offer.
- :func:`format_health_report` / :func:`format_reconcile_report` --
  the operator-facing text, kept out of the command bodies.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (which subcommand ran)
    L3 - semantic details (a refusal and its exit code)
    L4 - deep tracing (module load)
 
Contracts
---------
- EXIT CODES: ``0`` success, ``1`` a semantic refusal (the operator
  asked for something that does not make sense -- an unknown item, a
  duplicate, a bad position, a missing plan file), ``2`` a BACKEND
  refusal (:class:`~oc_orchestrator.work_queue.QueueBackendError` -- a
  corrupt log, a storage constraint, a missing registry argument).
  Argparse's own usage errors keep argparse's ``2`` via ``SystemExit``.
- ``--backend file`` is the DEFAULT, and the file backend is the mode
  the package's own tests run in (design note §7).
- ``--backend registry`` is a PLUGIN POINT and not this package's
  business: the registry that implements it is a consumer's, so this
  module names none of it, lazily or otherwise, and refuses the flag
  with :data:`NO_BACKEND_FACTORY` when no consumer supplied a factory.
- Every path is injected: ``--queue-dir`` for the queue,
  ``--plans-dir`` for plan files, ``--agent-registry`` for the instance
  registry, ``--registry`` for the orchestration registry. The
  defaults are today's values and nothing is discovered.
- ``add`` reads a plan's header through
  ``plan_metadata.py``'s provider -- the dispatch layer's one
  plan-header reader -- and never parses a plan file itself.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.work_queue (the contract)
- oc_orchestrator.work_queue_file (the default backend)
- oc_orchestrator.plan_inventory (the plans-dir literal and the terminal states)
- oc_orchestrator.plan_metadata (the one plan-header reader)
- oc_orchestrator.agent_instances (the instance registry ``health`` reads)
"""
 
from __future__ import annotations
 
import argparse
import sys
from collections.abc import Callable
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.agent_instances import (
    AgentInstanceRegistry,
    AgentRegistryError,
    default_agent_registry,
)
from oc_orchestrator.plan_inventory import (
    DEFAULT_PLANS_DIR,
    TERMINAL_STATUSES,
    PlanPaths,
)
from oc_orchestrator.plan_metadata import resolve_provider
from oc_orchestrator.work_queue import (
    KIND_PLAN,
    QueueBackend,
    QueueBackendError,
    QueueError,
    QueueState,
    WorkItem,
)
from oc_orchestrator.work_queue_file import (
    DEFAULT_QUEUE_DIR,
    SECTIONS,
    FileQueueBackend,
    QueuePaths,
)
 
_dbg("OC_CHAT.MAIN", "module", "work_queue_manage loaded", 4)
 
 
#: Success, a semantic refusal, a backend refusal.
EXIT_OK = 0
EXIT_REFUSED = 1
EXIT_BACKEND = 2
 
#: What ``--backend registry`` says when no consumer supplied a factory.
#: It is a BACKEND refusal -- the same arm that already refused a missing
#: operator slot -- so it keeps that arm's ``EXIT_BACKEND`` and its
#: one-line ``backend refused: ...`` stderr shape.
NO_BACKEND_FACTORY = (
    "--backend registry needs a consumer-supplied backend; "
    "oc_chat supplies one through its shim"
)
 
#: What ``next`` prints. ``<model>`` stays a placeholder on purpose:
#: the model is the OPERATOR'S attestation, not a queue field, and the
#: queue has no business guessing it.
DISPATCH_LINE = "scripts/dispatch.ps1 {number} <model>"
 
 
# ── Plan files ───────────────────────────────────────────────────────────
 
 
def _plan_file(paths: PlanPaths, plan_id: str) -> Path | None:
    """The ACTIVE plan file for *plan_id*, or ``None``."""
    candidate = paths.plans_dir / f"{plan_id}.md"
    return candidate if candidate.is_file() else None
 
 
def _terminal_plan_file(paths: PlanPaths, plan_id: str) -> Path | None:
    """A state-suffixed plan file in a TERMINAL state, or ``None``.
 
    ``plan_inventory.reconcile`` renames a plan's file to
    ``<plan_id>_<status>.md`` when its status changes, so the
    filesystem alone answers this -- no inventory singleton, no cwd.
    """
    for status in TERMINAL_STATUSES:
        candidate = paths.plans_dir / f"{plan_id}_{status}.md"
        if candidate.is_file():
            return candidate
    return None
 
 
def _plan_worktree(paths: PlanPaths, plan_id: str) -> Path | None:
    """The worktree the plan's header declares, through the one reader."""
    plan_path = _plan_file(paths, plan_id) or _terminal_plan_file(paths, plan_id)
    if plan_path is None:
        return None
    meta = resolve_provider(None).metadata(plan_path, plan_id=plan_id)
    return Path(meta.worktree) if meta.worktree else None
 
 
def _plan_title(paths: PlanPaths, plan_id: str) -> str | None:
    """The plan's ``# `` heading, through the one reader."""
    plan_path = _plan_file(paths, plan_id)
    if plan_path is None:
        return None
    return resolve_provider(None).metadata(plan_path, plan_id=plan_id).title
 
 
def _plan_number(plan_id: str) -> str:
    """The ``NNN`` of a ``NNN-stem`` plan id, or the id when it has none."""
    head = plan_id.split("-", 1)[0]
    return head if head.isdigit() else plan_id
 
 
# ── Backends ─────────────────────────────────────────────────────────────
 
 
def _build_backend(
    args: argparse.Namespace,
    backend_factory: Callable[[argparse.Namespace], QueueBackend] | None = None,
) -> QueueBackend:
    """The backend the flags ask for. ``file`` unless told otherwise.
 
    The ``registry`` arm is the PLUGIN POINT plan 258's receipt predicted
    and plan 271 P2 built. Before the move this function reached four
    ``oc_chat.registry`` names on four consecutive lines, and the receipt
    modelled them as a named cut precisely because a later extraction
    would replace them with a seam. This is that seam: the package holds
    no registry, so a consumer that wants one passes *backend_factory*
    and the package names nothing of it at any depth.
    """
    func = "_build_backend"
    if args.backend == "file":
        _dbg("OC_CHAT.MAIN", func, f"file backend at {args.queue_dir}", 2)
        return FileQueueBackend(QueuePaths(queue_dir=Path(args.queue_dir)))
 
    if backend_factory is None:
        # Raised bare, exactly as the pre-271 registry arm raised its
        # missing-operator-slot refusal: `main`'s QueueBackendError handler
        # owns the stderr line and the L1 trace, so logging here as well
        # would print the same refusal twice.
        raise QueueBackendError(NO_BACKEND_FACTORY)
    _dbg("OC_CHAT.MAIN", func, "registry backend from the consumer's factory", 2)
    return backend_factory(args)
 
 
def _paths(args: argparse.Namespace) -> PlanPaths:
    return PlanPaths(plans_dir=Path(args.plans_dir))
 
 
# ── Rendering ────────────────────────────────────────────────────────────
 
 
def _describe(item: WorkItem) -> str:
    """One work item as one line of operator-facing text."""
    parts = [item.item_id]
    extras = [
        f"{key}={value}"
        for key, value in (
            ("lane", item.lane),
            ("instance", item.instance),
        )
        if value
    ]
    if extras:
        parts.append(f"[{', '.join(extras)}]")
    if item.note:
        parts.append(f"- {item.note}")
    return " ".join(parts)
 
 
def format_state(state: QueueState) -> str:
    """The three buckets, as ``list`` prints them."""
    lines: list[str] = []
    for title, attribute in SECTIONS:
        lines.append(f"## {title}")
        items = getattr(state, attribute)
        if not items:
            lines.append("  (none)")
        for position, item in enumerate(items, start=1):
            lines.append(f"  {position}. {_describe(item)}")
        lines.append("")
    return "\n".join(lines).rstrip()
 
 
def format_health_report(
    state: QueueState,
    paths: PlanPaths,
    agent_registry: Path | None,
) -> str:
    """The two forgetting cases, plus the counts.
 
    A pending item is stale when its plan file is gone or when the plan
    reached a terminal state and nobody dequeued it. An in-progress item
    is stale when its instance is closed in ``agents.sqlite`` -- the
    worker went away without the queue being told -- or when the
    worktree its plan declares holds no ``.PLAN_STATUS.md``.
    """
    func = "format_health_report"
    lines = ["## Queue health", ""]
    lines.append(
        f"pending {len(state.pending)}  in progress {len(state.in_progress)}  "
        f"completed {len(state.completed)}"
    )
 
    stale_pending: list[str] = []
    for item in state.pending:
        if _terminal_plan_file(paths, item.item_id) is not None:
            stale_pending.append(
                f"- {item.item_id}: the plan file is in a terminal state; "
                "it was never dequeued or completed"
            )
        elif _plan_file(paths, item.item_id) is None:
            stale_pending.append(
                f"- {item.item_id}: missing plan file under {paths.plans_dir}"
            )
    if stale_pending:
        lines.extend(["", "## Pending items that need attention", *stale_pending])
 
    registry = _open_instance_registry(agent_registry)
    stale_running: list[str] = []
    for item in state.in_progress:
        if registry is not None and item.instance:
            state_text = _instance_state(registry, item.instance)
            if state_text == "closed":
                stale_running.append(
                    f"- {item.item_id}: instance {item.instance} is closed; "
                    "nobody completed the item"
                )
                continue
        worktree = _plan_worktree(paths, item.item_id)
        if worktree is not None and not (worktree / ".PLAN_STATUS.md").is_file():
            stale_running.append(
                f"- {item.item_id}: no .PLAN_STATUS.md under {worktree}"
            )
    if stale_running:
        lines.extend(["", "## In-progress items that need attention", *stale_running])
 
    if not stale_pending and not stale_running:
        lines.extend(["", "Nothing needs attention."])
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"{len(stale_pending)} pending, {len(stale_running)} running flagged",
        3,
    )
    return "\n".join(lines)
 
 
def _open_instance_registry(path: Path | None) -> AgentInstanceRegistry | None:
    """The instance registry, or ``None`` when there is no file to read."""
    if path is None or not Path(path).is_file():
        return None
    return AgentInstanceRegistry(Path(path))
 
 
def _instance_state(registry: AgentInstanceRegistry, agent_id: str) -> str | None:
    """*agent_id*'s state, or ``None`` when the registry does not know it."""
    func = "_instance_state"
    try:
        return registry.get(agent_id).state
    except (AgentRegistryError, ValueError) as exc:
        # EXTERNAL_ERROR (L2): health is a report, not a gate. An
        # unknown or malformed id is worth a trace and nothing more.
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {agent_id}: {exc}", 3)
        return None
 
 
def format_reconcile_report(drifts: tuple[object, ...], applied: bool) -> str:
    """The drift list, as ``reconcile`` prints it."""
    lines = ["## Queue reconcile", ""]
    if not drifts:
        lines.append("No drift detected.")
        return "\n".join(lines)
    for drift in drifts:
        mark = "+" if getattr(drift, "fixable", False) else "!"
        lines.append(
            f"{mark} {getattr(drift, 'item_id', '?')}: "
            f"{getattr(drift, 'kind', '?')} - {getattr(drift, 'detail', '')}"
        )
    unfixable = [d for d in drifts if not getattr(d, "fixable", False)]
    lines.append("")
    if applied:
        lines.append(
            f"Applied {len(drifts) - len(unfixable)} edit(s) to the log as "
            "operations attributed to reconcile."
        )
        if unfixable:
            lines.append(
                f"{len(unfixable)} deletion(s) could not be applied: the "
                "operation set has no remove. The projection was re-rendered "
                "from the log, so those bullets are back."
            )
    else:
        lines.append("(dry run; pass --apply to record these as operations)")
    return "\n".join(lines)
 
 
# ── Commands ─────────────────────────────────────────────────────────────
 
 
def _cmd_list(args: argparse.Namespace, backend: QueueBackend) -> int:
    print(format_state(backend.status()))
    return EXIT_OK
 
 
def _cmd_add(args: argparse.Namespace, backend: QueueBackend) -> int:
    """Queue a plan, refusing an id with no active plan file."""
    func = "_cmd_add"
    paths = _paths(args)
    if _plan_file(paths, args.plan_id) is None:
        terminal = _terminal_plan_file(paths, args.plan_id)
        detail = (
            f"its only file is {terminal.name}, a terminal state"
            if terminal is not None
            else f"no {args.plan_id}.md"
        )
        print(
            f"refused: {args.plan_id} has no active plan file under "
            f"{paths.plans_dir} ({detail})",
            file=sys.stderr,
        )
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {args.plan_id}", 3)
        return EXIT_REFUSED
    item = WorkItem(
        kind=KIND_PLAN,
        item_id=args.plan_id,
        lane=args.lane,
        note=args.note or _plan_title(paths, args.plan_id),
    )
    backend.enqueue(item, position=args.position)
    print(f"Queued {_describe(item)}")
    return EXIT_OK
 
 
def _cmd_move(args: argparse.Namespace, backend: QueueBackend) -> int:
    backend.move(args.plan_id, args.position, reason=args.reason)
    print(f"Moved {args.plan_id} to position {args.position}.")
    return EXIT_OK
 
 
def _cmd_dequeue(args: argparse.Namespace, backend: QueueBackend) -> int:
    backend.dequeue(args.plan_id, instance=args.instance, reason=args.reason)
    print(f"Dequeued {args.plan_id} to {args.instance}.")
    return EXIT_OK
 
 
def _cmd_complete(args: argparse.Namespace, backend: QueueBackend) -> int:
    """Finish an item. Already-complete is success, as the retired CLI had it."""
    already = [
        item.item_id for item in backend.status().completed
    ]
    if args.plan_id in already:
        print(f"{args.plan_id} is already complete.")
        return EXIT_OK
    extra: dict[str, object] = {}
    if args.round_no is not None:
        extra["round_no"] = args.round_no
    if args.merge_commit_sha is not None:
        extra["merge_commit_sha"] = args.merge_commit_sha
    backend.complete(args.plan_id, reason=args.reason, **extra)
    print(f"Completed {args.plan_id}.")
    return EXIT_OK
 
 
def _cmd_next(args: argparse.Namespace, backend: QueueBackend) -> int:
    """Print what the operator would dispatch. Never dispatch it.
 
    The design note's autonomy boundary is the reason this prints
    rather than runs, and automated dispatch is the hub's plan. The
    ``<model>`` placeholder stays a placeholder: the model is the
    operator's attestation.
    """
    func = "_cmd_next"
    state = backend.status()
    if not state.pending:
        print("Nothing pending.")
        return EXIT_OK
    item = state.pending[0]
    print(f"next: {_describe(item)}")
    print(f"dispatch: {DISPATCH_LINE.format(number=_plan_number(item.item_id))}")
    print("(this command prints the line; it does not run it)")
    _dbg("OC_CHAT.MAIN", func, f"next is {item.item_id}", 2)
    return EXIT_OK
 
 
def _cmd_health(args: argparse.Namespace, backend: QueueBackend) -> int:
    registry_path = (
        Path(args.agent_registry) if args.agent_registry else default_agent_registry()
    )
    print(format_health_report(backend.status(), _paths(args), registry_path))
    return EXIT_OK
 
 
def _cmd_reconcile(args: argparse.Namespace, backend: QueueBackend) -> int:
    """Compare the projection with the log. Dry-run unless ``--apply``."""
    func = "_cmd_reconcile"
    reconcile = getattr(backend, "reconcile", None)
    if reconcile is None:
        print(
            f"refused: the {args.backend} backend has no projection to reconcile",
            file=sys.stderr,
        )
        return EXIT_REFUSED
    drifts = reconcile(apply=args.apply)
    print(format_reconcile_report(drifts, applied=args.apply))
    _dbg("OC_CHAT.MAIN", func, f"{len(drifts)} drift(s), apply={args.apply}", 2)
    return EXIT_OK
 
 
def _cmd_history(args: argparse.Namespace, backend: QueueBackend) -> int:
    ops = backend.history(args.plan_id)
    if not ops:
        print(f"No recorded operations for {args.plan_id}.")
        return EXIT_OK
    for op in ops:
        print(f"{op.seq:>4}  {op.at}  {op.op:<9} {op.actor}  {dict(op.payload)}")
    return EXIT_OK
 
 
_COMMANDS = {
    "list": _cmd_list,
    "add": _cmd_add,
    "move": _cmd_move,
    "dequeue": _cmd_dequeue,
    "complete": _cmd_complete,
    "next": _cmd_next,
    "health": _cmd_health,
    "reconcile": _cmd_reconcile,
    "history": _cmd_history,
}
 
 
# ── Parser ───────────────────────────────────────────────────────────────
 
 
#: How this CLI names itself in ``--help``. A consumer that exposes the
#: tool under its own module path passes its own (plan 271 P2): 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.work_queue_manage"
 
 
def build_parser(prog: str = DEFAULT_PROG) -> argparse.ArgumentParser:
    """The command shape, exposed so a witness can read the defaults."""
    parser = argparse.ArgumentParser(
        prog=prog,
        description="Manage the OC Chat work queue. This command never dispatches.",
    )
    parser.add_argument(
        "--backend",
        choices=["file", "registry"],
        default="file",
        help="Queue backend (default: file, the tested mode)",
    )
    parser.add_argument(
        "--queue-dir",
        default=str(DEFAULT_QUEUE_DIR),
        help=f"File-backend queue directory (default: {DEFAULT_QUEUE_DIR})",
    )
    parser.add_argument(
        "--plans-dir",
        default=str(DEFAULT_PLANS_DIR),
        help=f"Where plan files live (default: {DEFAULT_PLANS_DIR})",
    )
    parser.add_argument("--registry", help="Registry-backend SQLite path")
    parser.add_argument("--slot", help="Registry queue slot (default: the actor slot)")
    parser.add_argument(
        "--actor-slot", help="Registry actor slot (default: the active operator)"
    )
    parser.add_argument(
        "--agent-registry", help="agents.sqlite path, read by health"
    )
    sub = parser.add_subparsers(dest="command", required=True)
 
    sub.add_parser("list", help="Print pending, in-progress and completed items")
 
    add = sub.add_parser("add", help="Queue a plan")
    add.add_argument("plan_id")
    add.add_argument("--lane", help="Free-text preference, e.g. a model family")
    add.add_argument("--position", type=int, help="1-based position (default: last)")
    add.add_argument("--note", help="Free text (default: the plan's title)")
 
    move = sub.add_parser("move", help="Move a pending item to another position")
    move.add_argument("plan_id")
    move.add_argument("--position", type=int, required=True)
    move.add_argument("--reason", required=True)
 
    dequeue = sub.add_parser("dequeue", help="Take an item into progress")
    dequeue.add_argument("plan_id")
    dequeue.add_argument("--instance", required=True, help="agent-NNN")
    dequeue.add_argument("--reason", required=True)
 
    complete = sub.add_parser("complete", help="Finish an item")
    complete.add_argument("plan_id")
    complete.add_argument("--reason", required=True)
    complete.add_argument(
        "--round-no", type=int, help="Required by the registry backend"
    )
    complete.add_argument(
        "--merge-commit-sha", help="Required by the registry backend"
    )
 
    sub.add_parser(
        "next",
        help="Print the first pending item and the dispatch line. Never runs it.",
    )
    sub.add_parser("health", help="Report items that look forgotten")
 
    reconcile = sub.add_parser(
        "reconcile", help="Compare the projection with the log (dry run by default)"
    )
    reconcile.add_argument(
        "--apply",
        action="store_true",
        help="Record the hand edits as operations attributed to reconcile",
    )
 
    history = sub.add_parser("history", help="Print every operation for one item")
    history.add_argument("plan_id")
    return parser
 
 
def main(
    argv: list[str] | None = None,
    *,
    prog: str = DEFAULT_PROG,
    backend_factory: Callable[[argparse.Namespace], QueueBackend] | None = None,
) -> int:
    """Run one subcommand and return its exit code.
 
    *prog* and *backend_factory* are the consumer seams (plan 271 P2).
    Both default to this package standing alone: it names itself, and
    ``--backend registry`` refuses because this package has no registry
    to build one from. ``oc_chat``'s shim passes its own module path and
    a factory that builds ``RegistryQueueBackend``, which is what keeps
    every output and every queue transition identical across the split.
    """
    func = "main"
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(errors="replace")
        except (AttributeError, ValueError):
            pass  # non-reconfigurable stream (tests, pipes wrapped by harnesses)
 
    args = build_parser(prog).parse_args(argv)
    _dbg("OC_CHAT.MAIN", func, f"subcommand {args.command}", 2)
    try:
        backend = _build_backend(args, backend_factory)
        return _COMMANDS[args.command](args, backend)
    except QueueBackendError as exc:
        # The BACKEND refused: a corrupt log, a storage constraint, an
        # argument only that backend demands. A different backend would
        # not have refused, so it gets its own exit code.
        print(f"backend refused: {exc}", file=sys.stderr)
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: {exc}", 1, override=True)
        return EXIT_BACKEND
    except QueueError as exc:
        print(f"refused: {exc}", file=sys.stderr)
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 3)
        return EXIT_REFUSED
 
 
if __name__ == "__main__":
    sys.exit(main())