hub_manage.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/hub_manage.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_orchestrator/tests/test_259_hub_manage_cli.py"
"""
The hub's operator surface: send, ask, rule, inbox, asks, fleet, history.
 
Purpose
-------
Plan 259 P8 and P7. This is where a human, a worker and the orchestrator
all reach the hub. Two of its properties are the point; the rest is
plumbing around them.
 
**Who the process is comes from the BACKEND, never from a flag.** There
is no ``--as``. In file mode the actor is the operator at the control
centre and the correlation is the process user and working directory; in
registry mode it is the active operator slot. A flag that set provenance
would be exactly the mechanism by which an agent relays a yes, and the
whole point of the hub's provenance rule is that such a yes is
distinguishable. So the flag does not exist, and asking for one is an
argparse usage error.
 
**``fleet`` says "waiting" before anyone says "stalled".** One line per
non-closed instance: what it is, what plan it carries, what state the
registry has it in, and either ``waiting on ask <id> since <time>`` or
``no message since <time>``. A worker blocked on a human answer is
waiting on a decision, not going round in circles, and automated
dispatch that could not tell the difference would halt every worker that
ever asked a question. That is the fleet-level half of design-note
invariant 3; the in-session stall detector's half belongs to the Tier-2
follow-up and is not touched here.
 
**This command dispatches nothing and relays nothing.** ``relay-check``
prints the ``oc_connect relay hub`` line an operator would run, and that
is the whole of it -- this module does not import ``subprocess``, so it
could not do more if it tried. The same autonomy boundary
``work_queue_manage next`` observes.
 
**Two consumer seams, and nothing else outward** (plan 272 P2). This
package holds no registry and declares no connector, so the two places
the pre-272 module reached for one are now seams:
:func:`main`'s ``prog``, the name the CLI gives itself in ``--help``, and
``log_factory``, which builds the log ``--backend registry`` asks for.
With neither supplied the package stands alone: it names itself, and
``--backend registry`` refuses by name rather than pretending. The
connector is the ``[telegram]`` extra, reached lazily from the two
``rule --from-inbox`` sites and nowhere else, so a ``hub_manage asks``
in a venv that has never heard of ``oc_connect`` runs exactly as it
always did.
 
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, plus the
  two consumer seams.
- :func:`default_hooks` / :func:`hooks_for_dispatch` -- the live
  :class:`~oc_orchestrator.dispatch.DispatchHooks` P6's caller registers,
  which records a ``dispatch`` message and, for a PENDING item only, one
  ``dequeue`` with ``reason="dispatched"``.
- :func:`format_fleet_report` -- P7's lines, kept out of the command body.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (which subcommand ran, a hook firing)
    L3 - semantic details (a refusal and its exit code)
    L4 - deep tracing (module load)
 
Contracts
---------
- EXIT CODES, matching ``work_queue_manage``: ``0`` success, ``1`` a
  SEMANTIC refusal (an unknown address, an unknown kind, a ruling on
  nothing), ``2`` a BACKEND refusal (a corrupt log, a missing registry).
  Argparse's own usage errors keep argparse's ``2`` via ``SystemExit``.
- ``--backend file`` is the DEFAULT and the mode the package's own tests
  run in (design note §7).
- Every path is injected: ``--queue-dir``, ``--comms-root``,
  ``--registry``, ``--agent-registry``, ``--actor-slot``. The defaults
  are today's values and nothing is discovered except the comms root,
  which is walked from the working directory exactly as every other lane
  writer walks it.
- :func:`_build_log`'s registry arm is a PLUGIN POINT, not an import.
  Before plan 272 it reached four ``oc_chat.registry`` names on four
  consecutive lines; they now live in the consumer's factory, exactly as
  plan 271 did for the queue's ``_build_backend``. This module names no
  registry at any depth, lazily or otherwise.
- The connector is reached from :func:`_reply_from_inbox` and
  :func:`_consume_inbound` and NOWHERE else, inside the function bodies,
  and a missing ``[telegram]`` extra is a named refusal rather than a
  traceback. ``tests/test_t1_leaf_guard.py`` holds both halves.
- ``ingest`` runs implicitly before ``inbox``, ``asks`` and ``fleet``, so
  a blocked worker is an open ask before anyone looks for one.
- NO ``subprocess`` import, anywhere, at any depth.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.hub (the contract)
- oc_orchestrator.hub_file (the default backend and the lane carrier)
- oc_orchestrator.agent_instances (the instance registry ``fleet`` reads)
- oc_orchestrator.work_queue (the refusal class ``main`` maps)
- oc_orchestrator.work_queue_file (the queue the dispatch hook records into)
- oc_connect.audit (lazy; the ``[telegram]`` extra's audit store, for ``rule --from-inbox`` only)
- oc_connect.hub_lane (lazy; the ``[telegram]`` extra's reply-to-hub-message correlation, for ``rule --from-inbox`` only)
- oc_connect.inbound.inbox (lazy; the ``[telegram]`` extra's inbound store, read and marked acted by ``rule --from-inbox`` only)
"""
 
from __future__ import annotations
 
import argparse
import getpass
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.agent_instances import (
    AgentInstanceRegistry,
    AgentRegistryError,
    default_agent_registry,
)
from oc_orchestrator.hub import (
    ADDRESS_OPERATOR_LOCAL,
    ADDRESS_ORCHESTRATOR,
    KIND_ASK,
    KIND_DISPATCH,
    RULING_ANSWERS,
    SUPPORTED_KINDS,
    SUPPORTED_REQUIRES,
    Hub,
    HubEntry,
    HubError,
    Provenance,
    new_message,
    parse_address,
)
from oc_orchestrator.hub_file import (
    DEFAULT_QUEUE_DIR,
    CorruptHubLogError,
    FileHubLog,
    FileLaneCarrier,
    HubPaths,
    find_comms_root,
    ingest_lanes,
)
from oc_orchestrator.work_queue import QueueError
from oc_orchestrator.work_queue_file import FileQueueBackend, QueuePaths
 
_dbg("OC_CHAT.MAIN", "module", "hub_manage loaded", 4)
 
 
#: Success, a semantic refusal, a backend refusal.
EXIT_OK = 0
EXIT_REFUSED = 1
EXIT_BACKEND = 2
 
#: What ``relay-check`` prints. It is a LINE, not a command this module
#: runs: relaying holds a credential, and the autonomy boundary keeps
#: that in the operator's own process.
RELAY_LINE = "python -m oc_connect relay hub"
 
#: The ``QUEUE_DEQUEUE_REASONS`` member a dispatch records (plan 259 P4,
#: closing the 258 residual). The queue's own vocabulary now has a word
#: for what happened, so the hook does not have to borrow one.
DEQUEUE_REASON_DISPATCHED = "dispatched"
 
#: What ``--backend registry`` says when no consumer supplied a factory
#: (plan 272 P2, in plan 271's shape). It is a BACKEND refusal -- the
#: same arm that already refused an unopenable registry -- so it keeps
#: that arm's :data:`EXIT_BACKEND` and its one-line ``backend refused:``
#: stderr shape.
NO_LOG_FACTORY = (
    "--backend registry needs a consumer-supplied log; "
    "oc_chat supplies one through its shim"
)
 
#: What the two connector sites say when the ``[telegram]`` extra is not
#: installed. The same refusal shape for the same reason: the package
#: declares the connector optional, so its absence is a fact about this
#: venv and not a crash.
NO_TELEGRAM_EXTRA = (
    "rule --from-inbox needs the [telegram] extra: "
    "pip install oc_orchestrator[telegram]"
)
 
 
# ── Backends ─────────────────────────────────────────────────────────────
 
 
def _build_log(
    args: argparse.Namespace,
    log_factory: "Callable[[argparse.Namespace], Any] | None" = None,
) -> Any:
    """The log the flags ask for. ``file`` unless told otherwise.
 
    The ``registry`` arm is the PLUGIN POINT plan 272 P2 built, in the
    shape plan 271 gave the queue's ``_build_backend``. Before the move
    this function reached four ``oc_chat.registry`` names on four
    consecutive lines, and the plan-259 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 *log_factory* and the package names nothing of
    it at any depth.
 
    Raises:
        CorruptHubLogError: SYSTEM_ERROR (L1 for the caller) --
            ``--backend registry`` with no factory to fill it. ``main``'s
            handler owns the stderr line and the trace, so the refusal is
            raised bare exactly as the pre-272 registry arm raised its
            missing-operator-slot one.
    """
    func = "_build_log"
    if args.backend == "file":
        _dbg("OC_CHAT.MAIN", func, f"file log at {args.queue_dir}", 2)
        return FileHubLog(HubPaths(queue_dir=Path(args.queue_dir)))
 
    if log_factory is None:
        raise CorruptHubLogError(NO_LOG_FACTORY)
    _dbg("OC_CHAT.MAIN", func, "registry log from the consumer's factory", 2)
    return log_factory(args)
 
 
def _comms_root(args: argparse.Namespace) -> Path | None:
    """The bus root the flags ask for, or the walk, or ``None``.
 
    ``None`` means the lane carrier is not built: the log still records
    every message, so a hub with no reachable bus is quieter rather than
    broken.
    """
    func = "_comms_root"
    if getattr(args, "comms_root", None):
        return Path(args.comms_root).expanduser().resolve()
    try:
        return find_comms_root(Path.cwd())
    except CorruptHubLogError as exc:
        # EXTERNAL_ERROR (L2): running outside a workspace is a fact
        # about where the operator is, not a refusal to make.
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {exc}", 2)
        return None
 
 
def _build_hub(args: argparse.Namespace, log: Any) -> Hub:
    """The hub the flags ask for, with the file lane as its one carrier.
 
    The bus carrier is deliberately absent from a CLI process: nothing is
    attached in a command that runs for a hundred milliseconds, so it
    would never win a routing decision. A long-lived control centre
    injects one; this tool is not that.
    """
    root = _comms_root(args)
    carriers = (FileLaneCarrier(comms_root=root),) if root is not None else ()
    return Hub(log=log, carriers=carriers)
 
 
def _actor_provenance() -> Provenance:
    """Who this PROCESS is, in the hub's own terms.
 
    The correlation is the process user and the working directory --
    the two things that say which terminal this ruling was typed into.
    There is no flag anywhere that can change any of these four values;
    see the module docstring for why.
    """
    return Provenance(
        actor=ADDRESS_OPERATOR_LOCAL,
        actor_type="human_operator",
        channel="cli",
        correlation={"user": _process_user(), "cwd": str(Path.cwd())},
    )
 
 
def _process_user() -> str:
    """The OS user, or a named stand-in when the platform will not say."""
    try:
        return getpass.getuser()
    except Exception:  # noqa: BLE001 - EXTERNAL_ERROR (L3); a name, not a gate
        return "unknown-user"
 
 
# ── Rendering ────────────────────────────────────────────────────────────
 
 
def _describe(entry: HubEntry) -> str:
    """One recorded message as one line of operator-facing text."""
    message = entry.message
    parts = [
        f"{entry.seq:>4}",
        message.at,
        f"{message.kind:<11}",
        f"{message.sender} -> {message.recipient}",
    ]
    if message.subject:
        parts.append(f"[{message.subject}]")
    if message.requires:
        parts.append(f"requires={message.requires}")
    if message.answer:
        parts.append(f"answer={message.answer}")
    parts.append(message.message_id)
    return "  ".join(parts)
 
 
def format_entries(entries: tuple[HubEntry, ...], *, empty: str) -> str:
    """A list of messages, with each body indented under its own header.
 
    The dense form: ``history`` keeps it because that is the audit view.
    ``inbox`` and ``asks`` render cards (:func:`format_cards`) instead.
    """
    if not entries:
        return empty
    lines: list[str] = []
    for entry in entries:
        lines.append(_describe(entry))
        for line in entry.message.body.splitlines():
            lines.append(f"      {line}")
    return "\n".join(lines)
 
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Operator-facing footer prose naming the rule verb and the four reply-vocabulary answers; scores 4.95 bits/char and carries no credential. Operator-ruled 2026-09-13 (hub_manage card-rendering chore escalation, line-scoped)"
RULE_HINT = "To answer:  hub_manage rule <id> --answer YES|NO|ACK|HOLD [--body \"...\"]"
 
 
def _age(at: str, *, now: datetime | None = None) -> str:
    """``(3 days ago)`` for an ISO timestamp, or ``""`` when it will not parse."""
    try:
        then = datetime.fromisoformat(at.replace("Z", "+00:00"))
    except ValueError:
        return ""
    if then.tzinfo is None:
        then = then.replace(tzinfo=timezone.utc)
    moment = now or datetime.now(timezone.utc)
    seconds = max(0, int((moment - then).total_seconds()))
    if seconds < 3600:
        return f"({seconds // 60} min ago)"
    if seconds < 86400:
        return f"({seconds // 3600} h ago)"
    days = seconds // 86400
    return f"({days} day{'s' if days != 1 else ''} ago)"
 
 
def _card_body(message: Any, *, full: bool) -> list[str]:
    """The body as card lines: the first line and a count, or all of it."""
    body_lines = message.body.splitlines()
    if full or len(body_lines) <= 1:
        return [f"    {line}" for line in body_lines] or ["    (no body)"]
    first = next((line for line in body_lines if line.strip()), "")
    rest = len(body_lines) - 1
    where = f"`history {message.subject}`" if message.subject else "`--full`"
    return [f"    {first}  (+{rest} lines; {where} for the body)"]
 
 
def format_cards(
    entries: tuple[HubEntry, ...],
    *,
    title: str,
    empty: str,
    source: str,
    full: bool = False,
    now: datetime | None = None,
) -> str:
    """One card per message under a header that says what the list is.
 
    Header: ``## <title> (<count>)   <source>``. Each card opens with a
    bracketed ordinal, the kind and the id, then one aligned field per
    line, then the body's first line with a line count (all of it under
    ``full``). A blank line separates cards, so the seam between two
    items is visible without reading either. When any card is an ask
    the footer names the verb that answers it.
    """
    header = f"## {title} ({len(entries)})   {source}"
    if not entries:
        return f"{header}\n\n{empty}"
    blocks: list[str] = [header]
    for index, entry in enumerate(entries, start=1):
        message = entry.message
        lines = [
            f"[{index}] {message.kind} {message.message_id}   seq {entry.seq}",
            f"    from      {message.sender} -> {message.recipient}",
        ]
        if message.subject:
            lines.append(f"    subject   {message.subject}")
        if message.requires:
            lines.append(f"    requires  {message.requires}")
        if message.answer:
            lines.append(f"    answer    {message.answer}")
        if message.in_reply_to:
            lines.append(f"    answers   {message.in_reply_to}")
        lines.append(f"    sent      {message.at}  {_age(message.at, now=now)}".rstrip())
        lines.extend(_card_body(message, full=full))
        blocks.append("\n".join(lines))
    if any(entry.message.kind == KIND_ASK for entry in entries):
        blocks.append(RULE_HINT)
    return "\n\n".join(blocks)
 
 
def entries_as_json(entries: tuple[HubEntry, ...], *, command: str, source: str) -> str:
    """The same list as one JSON document, for agents and tools."""
    return json.dumps(
        {
            "command": command,
            "source": source,
            "count": len(entries),
            "entries": [
                {"seq": entry.seq, **entry.message.as_dict()} for entry in entries
            ],
        },
        indent=2,
        sort_keys=False,
    )
 
 
def _source_label(args: argparse.Namespace) -> str:
    """Where the listing came from, so nobody wonders which store this is."""
    if args.backend == "file":
        return f"backend=file   log={Path(args.queue_dir) / 'hub.ops.jsonl'}"
    return f"backend=registry   registry={args.registry or 'discovered'}"
 
 
def _emit(
    args: argparse.Namespace,
    entries: tuple[HubEntry, ...],
    *,
    command: str,
    title: str,
    empty: str,
) -> None:
    """Print a listing as cards, or as JSON when ``--json`` was given."""
    source = _source_label(args)
    if getattr(args, "json", False):
        print(entries_as_json(entries, command=command, source=source))
        return
    print(format_cards(
        entries, title=title, empty=empty, source=source,
        full=getattr(args, "full", False),
    ))
 
 
def format_fleet_report(
    instances: list[Any],
    entries: tuple[HubEntry, ...],
) -> str:
    """P7: one line per non-closed instance, saying whether it is WAITING.
 
    The distinction the design note requires, made before anyone can call
    a worker stalled: an instance with an open ask it sent is waiting on
    a decision. The line names the ask and when it was sent, so the
    operator can answer it rather than investigate a silence.
    """
    func = "format_fleet_report"
    lines = ["## Fleet", ""]
    if not instances:
        lines.append("No open instances.")
        return "\n".join(lines)
 
    ruled = {
        entry.message.in_reply_to
        for entry in entries
        if entry.message.kind == "ruling"
    }
    waiting = 0
    for instance in instances:
        open_ask = next(
            (
                entry
                for entry in reversed(entries)
                if entry.message.kind == KIND_ASK
                and entry.message.sender == instance.agent_id
                and entry.message.message_id not in ruled
            ),
            None,
        )
        last = next(
            (
                entry
                for entry in reversed(entries)
                if instance.agent_id in (entry.message.sender, entry.message.recipient)
            ),
            None,
        )
        head = (
            f"- {instance.agent_id}  {instance.plan_id or '-'}  "
            f"state={instance.state}"
        )
        if last is not None:
            head += f"  last={last.message.kind} at {last.message.at}"
        if open_ask is not None:
            waiting += 1
            lines.append(
                f"{head}  waiting on ask {open_ask.message.message_id} since "
                f"{open_ask.message.at}"
            )
        elif last is not None:
            lines.append(f"{head}  no message since {last.message.at}")
        else:
            lines.append(f"{head}  no message since it was launched")
 
    lines.extend(["", f"{waiting} of {len(instances)} waiting on a decision."])
    _dbg("OC_CHAT.MAIN", func, f"{waiting}/{len(instances)} waiting", 3)
    return "\n".join(lines)
 
 
# ── Commands ─────────────────────────────────────────────────────────────
 
 
def _ingest(args: argparse.Namespace, log: Any) -> int:
    """Fold the two existing lanes in. Returns how many were new."""
    root = _comms_root(args)
    if root is None:
        return 0
    return len(ingest_lanes(root, log))
 
 
def _cmd_send(args: argparse.Namespace, log: Any) -> int:
    hub = _build_hub(args, log)
    message = new_message(
        kind=args.kind,
        sender=args.sender,
        recipient=args.recipient,
        subject=args.subject,
        body=args.body,
        requires=args.requires if args.kind == KIND_ASK else None,
        provenance=_actor_provenance(),
    )
    _deliver(hub, message)
    print(f"Recorded {message.kind} {message.message_id} -> {message.recipient}.")
    return EXIT_OK
 
 
def _cmd_ask(args: argparse.Namespace, log: Any) -> int:
    """A ``send`` shorthand: the kind is fixed and an answerer is required."""
    args.kind = KIND_ASK
    return _cmd_send(args, log)
 
 
def _cmd_rule(args: argparse.Namespace, log: Any) -> int:
    """Answer an open ask, from this process or from a stored reply."""
    func = "_cmd_rule"
    hub = _build_hub(args, log)
    if args.from_inbox:
        ask_id, provenance, answer = _reply_from_inbox(args)
        body = args.body or f"{answer} (from the operator's reply)"
    else:
        if not args.ask_id:
            print("rule needs an ask id, or --from-inbox", file=sys.stderr)
            return EXIT_REFUSED
        ask_id, provenance, answer = args.ask_id, _actor_provenance(), args.answer
        body = args.body or f"{answer} (ruled at the control centre)"
 
    entry = hub.rule(ask_id, answer=answer, body=body, provenance=provenance)
    if args.from_inbox:
        _consume_inbound(args, args.from_inbox, entry.message.message_id)
    print(
        f"Ruled {answer} on {ask_id} as {entry.message.message_id} "
        f"(channel={provenance.channel})."
    )
    _dbg("OC_CHAT.MAIN", func, f"{answer} on {ask_id}", 2)
    return EXIT_OK
 
 
def _cmd_inbox(args: argparse.Namespace, log: Any) -> int:
    """What is addressed to one party and not yet read. Advances the cursor."""
    func = "_cmd_inbox"
    parse_address(args.address)
    _ingest(args, log)
    if args.all:
        entries = log.entries(recipient=args.address)
        _emit(args, entries, command="inbox", title=f"Inbox for {args.address}, everything",
              empty=f"Nothing addressed to {args.address}.")
        return EXIT_OK
 
    entries = _unread(log, args.address)
    _emit(args, entries, command="inbox", title=f"Inbox for {args.address}, unread",
          empty=f"Nothing new for {args.address}.")
    if entries:
        _advance(log, args.address, entries[-1].seq)
        _dbg("OC_CHAT.MAIN", func,
             f"{args.address} cursor -> {entries[-1].seq}", 2)
    return EXIT_OK
 
 
def _cmd_asks(args: argparse.Namespace, log: Any) -> int:
    """Open questions, or every question ever asked."""
    _ingest(args, log)
    recipient = parse_address(args.address) if args.address else None
    scope = f" for {recipient}" if recipient else ""
    if args.open:
        entries = log.open_asks(recipient=recipient)
        title, empty = f"Open asks{scope}", "No open asks."
    else:
        entries = log.entries(kind=KIND_ASK, recipient=recipient)
        title, empty = f"All asks{scope}", "No asks recorded."
    _emit(args, entries, command="asks", title=title, empty=empty)
    return EXIT_OK
 
 
def _cmd_fleet(args: argparse.Namespace, log: Any) -> int:
    """P7: one line per open instance, saying whether it is waiting."""
    _ingest(args, log)
    registry_path = (
        Path(args.agent_registry) if args.agent_registry else default_agent_registry()
    )
    if not Path(registry_path).is_file():
        print(f"## Fleet\n\nNo instance registry at {registry_path}.")
        return EXIT_OK
    try:
        instances = AgentInstanceRegistry(Path(registry_path)).list(active_only=True)
    except (AgentRegistryError, ValueError) as exc:
        # EXTERNAL_ERROR (L2): "unreadable" and "none" are different
        # answers to the operator's question, so say which one this is.
        print(f"## Fleet\n\nThe instance registry is unreadable: {exc}")
        return EXIT_OK
    print(format_fleet_report(instances, log.entries()))
    return EXIT_OK
 
 
def _cmd_ingest(args: argparse.Namespace, log: Any) -> int:
    count = _ingest(args, log)
    print(
        f"Ingested {count} lane event(s)." if count else "Nothing new on the lanes."
    )
    return EXIT_OK
 
 
def _cmd_history(args: argparse.Namespace, log: Any) -> int:
    _ingest(args, log)
    entries = log.entries(subject=args.subject)
    if args.json:
        print(entries_as_json(entries, command="history", source=_source_label(args)))
        return EXIT_OK
    # Dense on purpose: history is the audit view (operator ruling, 2026-09-13).
    print(format_entries(entries, empty=f"No recorded messages about {args.subject}."))
    return EXIT_OK
 
 
def _cmd_relay_check(args: argparse.Namespace, log: Any) -> int:
    """Print the relay line an operator would run. Never run it.
 
    Relaying holds a credential and reaches the transport; the autonomy
    boundary keeps both in the operator's own process. This module does
    not import ``subprocess``, so this could not spawn anything even by
    accident.
    """
    _ingest(args, log)
    pending = [
        entry
        for entry in log.entries(recipient="operator@remote")
    ]
    print(f"{len(pending)} hub message(s) addressed to operator@remote.")
    print(f"relay: {RELAY_LINE}")
    print("(this command prints the line; it does not run it)")
    return EXIT_OK
 
 
_COMMANDS = {
    "send": _cmd_send,
    "ask": _cmd_ask,
    "rule": _cmd_rule,
    "inbox": _cmd_inbox,
    "asks": _cmd_asks,
    "fleet": _cmd_fleet,
    "ingest": _cmd_ingest,
    "history": _cmd_history,
    "relay-check": _cmd_relay_check,
}
 
 
# ── Internals the commands share ─────────────────────────────────────────
 
 
def _deliver(hub: Hub, message: Any) -> None:
    """Send, tolerating a hub with no carrier for that recipient.
 
    The log is the truth and :meth:`Hub.send` records BEFORE it routes,
    so a message nobody could carry is still a recorded message. A CLI
    that exited non-zero here would be reporting the absence of a
    listener as a failure to send.
    """
    func = "_deliver"
    from oc_orchestrator.hub import CarrierError
 
    try:
        hub.send(message)
    except CarrierError as exc:
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {exc}", 2)
        print(f"note: {exc}", file=sys.stderr)
 
 
def _unread(log: Any, address: str) -> tuple[HubEntry, ...]:
    """Unread entries for *address*, or everything when the log has no cursor.
 
    The registry backend has no cursor file: its log is the operation
    log, which no reader may write to. A cursor for it is the follow-up
    the report names; until then ``inbox --backend registry`` shows
    everything and says so rather than pretending to track a position.
    """
    reader = getattr(log, "unread_for", None)
    if reader is None:
        return log.entries(recipient=address)
    return reader(address)
 
 
def _advance(log: Any, address: str, seq: int) -> None:
    """Move *address*'s cursor, where the backend has one."""
    advance = getattr(log, "advance_cursor", None)
    if advance is not None:
        advance(address, seq)
 
 
def _reply_from_inbox(args: argparse.Namespace) -> tuple[str, Provenance, str]:
    """Resolve a stored inbound reply into (ask id, provenance, answer).
 
    The ONE place this tool reaches the connector for a reply, and it is
    a lazy import because the correlation being walked is the
    connector's: a ``hub_manage asks`` must not load a package that
    handles credentials, and since plan 272 it must not even REQUIRE one
    to be installed. ``oc_connect`` is the ``[telegram]`` extra, so an
    ``ImportError`` here means the operator asked for a sub-command this
    venv was not built for, and it is a named refusal rather than a
    traceback.
 
    The provenance built here is ``telegram`` with the three correlation
    fields the hub's human rule demands -- and every one of them comes
    from the STORED message, never from the command line.
 
    Raises:
        HubError: VALIDATION_ERROR (L3) -- no comms bus, an uncorrelated
            reply, an answer outside the vocabulary, or the
            ``[telegram]`` extra absent.
    """
    func = "_reply_from_inbox"
    try:
        from oc_connect.audit import AuditStore
        from oc_connect.hub_lane import UncorrelatedReplyError, resolve_hub_reply
        from oc_connect.inbound.inbox import InboxStore
    except ImportError as exc:
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {NO_TELEGRAM_EXTRA}", 3)
        raise HubError(NO_TELEGRAM_EXTRA) from exc
 
    root = _comms_root(args)
    if root is None:
        raise HubError(
            "--from-inbox needs a comms bus, and none was found above the "
            "working directory; pass --comms-root"
        )
    audit = AuditStore(root / "audit" / "audit.jsonl")
    inbox = InboxStore(root / "inbound" / "telegram", audit=audit)
    try:
        reply = resolve_hub_reply(args.from_inbox, inbox=inbox, audit=audit)
    except UncorrelatedReplyError as exc:
        raise HubError(str(exc)) from exc
    if reply.answer not in RULING_ANSWERS:
        raise HubError(
            f"inbound message {args.from_inbox!r} carries {reply.answer!r}, which "
            f"is not one of {', '.join(RULING_ANSWERS)}; a ruling needs a "
            "vocabulary token, and free text is not one"
        )
    provenance = Provenance(
        actor=ADDRESS_OPERATOR_LOCAL,
        actor_type="human_operator",
        channel="telegram",
        correlation=dict(reply.correlation),
    )
    _dbg("OC_CHAT.MAIN", func,
         f"{args.from_inbox} answers {reply.hub_message_id}", 2)
    return reply.hub_message_id, provenance, reply.answer
 
 
def _consume_inbound(args: argparse.Namespace, message_id: str, decision: str) -> None:
    """Mark the inbound reply consumed by the ruling it produced.
 
    Best-effort: the ruling is already recorded, and failing to annotate
    the connector's copy of the operator's tap must not un-rule it. That
    covers the ``[telegram]`` extra being absent too -- unreachable in
    practice, because only ``--from-inbox`` calls this and that arm
    already refused without the extra, but a best-effort annotation that
    could raise on an import would be best-effort in name only.
    """
    func = "_consume_inbound"
    try:
        from oc_connect.audit import AuditStore
        from oc_connect.inbound.inbox import InboxStore
    except ImportError:
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {NO_TELEGRAM_EXTRA}", 2)
        return
 
    root = _comms_root(args)
    if root is None:
        return
    try:
        audit = AuditStore(root / "audit" / "audit.jsonl")
        InboxStore(root / "inbound" / "telegram", audit=audit).mark_acted(
            message_id, decision
        )
    except Exception as exc:  # noqa: BLE001 - EXTERNAL_ERROR (L2); the ruling stands
        _dbg("OC_CHAT.MAIN", func,
             f"EXTERNAL_ERROR: could not mark {message_id} acted: {exc}", 2)
        print(
            f"warning: inbound message {message_id} was not marked acted ({exc}); "
            f"the ruling {decision} is recorded regardless",
            file=sys.stderr,
        )
 
 
# ── The dispatch hooks (P6) ──────────────────────────────────────────────
 
 
class _HubDispatchHooks:
    """Record a dispatch, and dequeue the item only if it was pending."""
 
    def __init__(self, *, log: Any, queue: Any | None, comms_root: Path | None) -> None:
        self.log = log
        self.queue = queue
        self.comms_root = comms_root
 
    def dispatched(self, result: Any) -> None:
        """Record the ``dispatch`` message, then the queue operation."""
        func = "_HubDispatchHooks.dispatched"
        carriers = (
            (FileLaneCarrier(comms_root=self.comms_root),)
            if self.comms_root is not None
            else ()
        )
        message = new_message(
            kind=KIND_DISPATCH,
            sender=ADDRESS_ORCHESTRATOR,
            recipient=result.instance.agent_id,
            subject=result.plan_id,
            body=(
                f"Dispatched {result.plan_id} to {result.instance.agent_id} on "
                f"{result.branch}.\nLaunch: {result.launch_path}\n"
                f"Worktree: {result.worktree}\nBASE: {result.base_sha}"
            ),
            provenance=Provenance(
                actor=ADDRESS_ORCHESTRATOR,
                actor_type="orchestrator_agent",
                channel="session",
                correlation={},
            ),
        )
        _deliver(Hub(log=self.log, carriers=carriers), message)
        _dbg("OC_CHAT.MAIN", func,
             f"recorded dispatch of {result.plan_id} to {result.instance.agent_id}", 2)
        self._dequeue_if_pending(result)
 
    def _dequeue_if_pending(self, result: Any) -> None:
        """Take the item into progress -- only if the queue held it as pending.
 
        A plan that is NOT on the queue is logged as a dispatch and never
        enqueued: inventing queue membership at dispatch time is the
        silent projection plan 258 filed as a residual, and this is the
        side of it this plan controls.
        """
        func = "_HubDispatchHooks._dequeue_if_pending"
        if self.queue is None:
            return
        pending = [item.item_id for item in self.queue.status().pending]
        if result.plan_id not in pending:
            _dbg("OC_CHAT.MAIN", func,
                 f"{result.plan_id} is not pending; recorded, not enqueued", 2)
            return
        self.queue.dequeue(
            result.plan_id,
            instance=result.instance.agent_id,
            reason=DEQUEUE_REASON_DISPATCHED,
        )
        _dbg("OC_CHAT.MAIN", func,
             f"dequeued {result.plan_id} to {result.instance.agent_id}", 2)
 
 
def default_hooks(
    *,
    log: Any,
    queue: Any | None = None,
    comms_root: Path | None = None,
) -> Any:
    """The live :class:`~oc_orchestrator.dispatch.DispatchHooks` for P6.
 
    Registered by the CALLER -- ``plan_manage._cmd_dispatch`` -- exactly
    as ``plan_status`` registers its state observer, so ``dispatch.py``
    names neither the hub nor the queue. That direction survived the move
    unchanged: the hooks live here, beside the hub they announce into,
    and :mod:`oc_orchestrator.dispatch` still imports nothing from this
    module (plan 272 P3; ``test_270_standalone_dispatch.py`` holds it).
    """
    return _HubDispatchHooks(log=log, queue=queue, comms_root=comms_root)
 
 
def hooks_for_dispatch(
    *,
    queue_dir: Path,
    comms_root: Path | None = None,
) -> Any:
    """:func:`default_hooks` over the FILE backends under *queue_dir*.
 
    The convenience ``plan_manage`` calls, so the dispatch path names one
    directory rather than assembling two backends. The file mode is the
    default everywhere else and it is the default here.
    """
    paths = HubPaths(queue_dir=Path(queue_dir))
    return default_hooks(
        log=FileHubLog(paths),
        queue=FileQueueBackend(QueuePaths(queue_dir=Path(queue_dir))),
        comms_root=comms_root,
    )
 
 
# ── Parser ───────────────────────────────────────────────────────────────
 
 
def _listing_flags(parser: argparse.ArgumentParser) -> None:
    """``--full`` and ``--json`` on the two card-rendering listings."""
    parser.add_argument(
        "--full", action="store_true",
        help="Print every body in full instead of its first line and a count",
    )
    parser.add_argument(
        "--json", action="store_true", help="One JSON document instead of cards",
    )
 
 
#: How this CLI names itself in ``--help``. A consumer that exposes the
#: tool under its own module path passes its own (plan 272 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.hub_manage"
 
 
def build_parser(prog: str = DEFAULT_PROG) -> argparse.ArgumentParser:
    """The command shape, exposed so a witness can read the defaults.
 
    There is deliberately no ``--as`` and no way to set provenance from
    the command line. See the module docstring.
    """
    parser = argparse.ArgumentParser(
        prog=prog,
        description=(
            "Send, answer and read hub messages. This command dispatches "
            "nothing and relays nothing."
        ),
    )
    parser.add_argument(
        "--backend", choices=["file", "registry"], default="file",
        help="Hub log backend (default: file, the tested mode)",
    )
    parser.add_argument(
        "--queue-dir", default=str(DEFAULT_QUEUE_DIR),
        help=f"File-backend directory for the log and cursors (default: {DEFAULT_QUEUE_DIR})",
    )
    parser.add_argument(
        "--comms-root", help="Comms bus root (default: the workspace walk)"
    )
    parser.add_argument("--registry", help="Registry-backend SQLite path")
    parser.add_argument(
        "--actor-slot", help="Registry actor slot (default: the active operator)"
    )
    parser.add_argument(
        "--agent-registry", help="agents.sqlite path, read by fleet"
    )
    sub = parser.add_subparsers(dest="command", required=True)
 
    send = sub.add_parser("send", help="Record and route one message")
    send.add_argument("--kind", required=True, choices=sorted(SUPPORTED_KINDS))
    send.add_argument("--from", dest="sender", required=True, metavar="ADDRESS")
    send.add_argument("--to", dest="recipient", required=True, metavar="ADDRESS")
    send.add_argument("--subject", help="A work item id, such as 259-hub")
    send.add_argument(
        "--requires", choices=list(SUPPORTED_REQUIRES),
        help="For an ask: who may answer it",
    )
    send.add_argument("body")
 
    ask = sub.add_parser("ask", help="Record a question that expects a ruling")
    ask.add_argument("--from", dest="sender", required=True, metavar="ADDRESS")
    ask.add_argument("--to", dest="recipient", required=True, metavar="ADDRESS")
    ask.add_argument("--subject", help="A work item id, such as 259-hub")
    ask.add_argument("--requires", required=True, choices=list(SUPPORTED_REQUIRES))
    ask.add_argument("body")
 
    rule = sub.add_parser("rule", help="Answer an open ask")
    rule.add_argument("ask_id", nargs="?", help="The ask being answered")
    rule.add_argument(
        "--answer", choices=list(RULING_ANSWERS),
        help="Required unless --from-inbox, which reads the operator's own word",
    )
    rule.add_argument("--body", help="Free text; defaults to naming the answer")
    rule.add_argument(
        "--from-inbox", metavar="INBOUND_ID",
        help=(
            "Build the ruling from a stored operator reply, taking the answer "
            "and all three correlation fields from the message itself"
        ),
    )
 
    inbox = sub.add_parser("inbox", help="Read what is addressed to one party")
    inbox.add_argument("--for", dest="address", required=True, metavar="ADDRESS")
    inbox.add_argument(
        "--all", action="store_true",
        help="Show everything and leave the cursor where it is",
    )
    _listing_flags(inbox)
 
    asks = sub.add_parser("asks", help="List questions")
    asks.add_argument("--for", dest="address", metavar="ADDRESS")
    asks.add_argument("--open", action="store_true", help="Only unanswered ones")
    _listing_flags(asks)
 
    sub.add_parser("fleet", help="One line per open instance; says who is waiting")
    sub.add_parser("ingest", help="Fold the worker-report and session-alert lanes in")
 
    history = sub.add_parser("history", help="Every message about one work item")
    history.add_argument("subject")
    history.add_argument(
        "--json", action="store_true", help="One JSON document instead of text",
    )
 
    sub.add_parser(
        "relay-check",
        help="Print the oc_connect relay line for the hub lane. Never runs it.",
    )
    return parser
 
 
def main(
    argv: list[str] | None = None,
    *,
    prog: str = DEFAULT_PROG,
    log_factory: "Callable[[argparse.Namespace], Any] | None" = None,
) -> int:
    """Run one subcommand and return its exit code.
 
    *prog* and *log_factory* are the consumer seams (plan 272 P2). Both
    default to this package standing alone: it names itself, and
    ``--backend registry`` refuses because this package has no registry
    to build a log from. ``oc_chat``'s shim passes its own module path
    and a factory that builds ``RegistryHubLog``, which is what keeps
    every output and every recorded row 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)
    if args.command == "rule" and not args.from_inbox and not args.answer:
        print("rule needs --answer, or --from-inbox", file=sys.stderr)
        return EXIT_REFUSED
    try:
        log = _build_log(args, log_factory)
        return _COMMANDS[args.command](args, log)
    except CorruptHubLogError as exc:
        # The BACKEND refused: a corrupt log, a missing registry, a
        # storage constraint. 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 HubError as exc:
        print(f"refused: {exc}", file=sys.stderr)
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: {exc}", 3)
        return EXIT_REFUSED
    except QueueError as exc:
        print(f"backend refused: {exc}", file=sys.stderr)
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: {exc}", 1, override=True)
        return EXIT_BACKEND
    except (OSError, ValueError) as exc:
        print(f"backend refused: {exc}", file=sys.stderr)
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: {exc}", 1, override=True)
        return EXIT_BACKEND
 
 
if __name__ == "__main__":
    sys.exit(main())