queue_backend.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/queue_backend.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_258_queue_backend.py"
"""
The registry backend: the queue's operations over the registry's own.
 
Purpose
-------
Plan 258 P3. The queue has one operation set and two backends. This is
the one oc_chat plugs in, and it lives on the REGISTRY side on purpose:
signing, audit and checkpoints already live here, so the queue inherits
them instead of growing a second copy. It implements the same
:class:`~oc_chat.core.work_queue.QueueBackend` protocol over the four
typed operations that ``registry/domain.py`` already exposes --
``queue_enqueue``, ``queue_move``, ``queue_dequeue``, ``queue_complete``
-- and writes nothing else: every mutation is one operation-log row with
``related_plan_id`` set, and the registry's own projection updates the
``plans`` table.
 
The import direction is registry -> dispatch, which is the permitted
one and the one ``registry/mirror_observer.py`` already takes. Nothing
in the dispatch layer imports this module.
 
**The vocabulary mismatch, named rather than papered over.** The
registry's operations are slot-shaped and reason-closed; the contract's
are lane-shaped and reason-free. Three things do not fit, and each is
handled in the open:
 
``lane`` does not map onto ``slot``
    MEASURED against a temporary registry, not assumed: ``plans``
    declares ``assigned_slot TEXT REFERENCES agents(slot)`` and
    ``PRAGMA foreign_keys`` is on, so a free-text slot is refused with
    ``sqlite3.IntegrityError: FOREIGN KEY constraint failed``. The
    adapter therefore takes ONE registered slot at construction and
    refuses a work item whose ``lane`` names anything else, by name.
    No roster was invented to make the mapping work.
 
``instance`` and the contract's own reason have no column
    The queue payloads are fixed and adding a field is a registry
    change this plan does not make. They ride in ``command``, the
    operation log's free-text provenance column, in a deterministic
    form this module also parses back.
 
``complete`` needs a round and a merge commit
    ``queue_complete`` requires ``round_no`` and ``merge_commit_sha``.
    They are taken as keyword arguments -- which the file backend
    accepts and ignores -- rather than smuggled through ``reason``,
    because a round number and a sha packed into free text would be a
    parser in disguise.
 
Responsibilities
----------------
- :class:`RegistryQueueBackend` -- the six contract operations.
- :data:`ENQUEUE_REASON` / :data:`DEQUEUE_REASON_DEFAULT` -- the two
  closed-set members this adapter maps onto, pinned as constants so the
  witness can assert them.
- :class:`LaneIsNotASlotError`, :class:`MissingRegistryArgumentError`,
  :class:`UnknownPlanRowError`, :class:`UnsupportedItemKindError` --
  the adapter's own named refusals, all under the contract's
  :class:`~oc_chat.core.work_queue.QueueBackendError`.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (an operation emitted)
    L3 - semantic details (a mapping, a refusal)
    L4 - deep tracing (module load)
 
Contracts
---------
- This module MUTATES only through ``registry.domain``: no SQL write,
  no direct table update. Its reads are SELECTs through
  ``registry.readers`` and one connection for the ``plans`` projection.
- ``status()`` takes the pending ORDER from the ``plans`` table's
  ``assigned_slot`` / ``queue_position`` projection -- that projection
  is what the registry maintains -- and the PHASE of each item from its
  latest queue operation in the log, because the projection models
  pending and not-pending and cannot by itself tell "in progress" from
  "completed" from "never queued".
- ``kind`` other than ``"plan"`` is refused BY NAME: ``related_plan_id``
  is plan-shaped, which the design note §5 names as the seam a second
  kind will find.
- Nothing here is edited in ``registry/domain.py``,
  ``registry/backends/**`` or ``registry/cli/**``; this adapter is the
  only registry-side file plan 258 adds.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.work_queue (the contract; moved there by plan 271)
- oc_chat.registry.api (the Registry facade)
- oc_chat.registry.domain (the four typed queue operations)
- oc_chat.registry.readers (the queue history reader)
"""
 
from __future__ import annotations
 
import re
from dataclasses import replace
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.work_queue import (
    KIND_PLAN,
    OP_COMPLETE,
    OP_DEQUEUE,
    OP_ENQUEUE,
    OP_MOVE,
    BadPositionError,
    DuplicateItemError,
    QueueBackendError,
    QueueOp,
    QueueState,
    UnknownItemError,
    WorkItem,
)
from oc_chat.registry.api import Registry
from oc_chat.registry.domain import (
    QUEUE_DEQUEUE_REASONS,
    queue_complete,
    queue_dequeue,
    queue_enqueue,
    queue_move,
)
from oc_chat.registry.readers import queue_history
 
_dbg("OC_CHAT.MAIN", "module", "queue_backend loaded", 4)
 
 
#: The ``QUEUE_ENQUEUE_REASONS`` member an enqueue maps onto. The
#: contract's enqueue carries no reason, and this is the member that
#: means what it does.
ENQUEUE_REASON = "initial enqueue"
 
#: The ``QUEUE_DEQUEUE_REASONS`` member a dequeue maps onto when the
#: contract's own reason is not already a member. The closed set has NO
#: member meaning "dispatched to an instance"; of the three it has,
#: this is the only one whose projection leaves the plan's status
#: alone, which is what a dispatch needs -- ``transition_to_failed``
#: and ``transition_to_integrating`` both rewrite it, and a dispatch is
#: neither. That the set needs a member is reported, not invented.
DEQUEUE_REASON_DEFAULT = "operator_dequeue"
 
#: The registry operation types this adapter emits and folds, mapped to
#: the contract's op names so a history read through either backend
#: reads the same.
_OP_BY_TYPE: dict[str, str] = {
    "queue.enqueue": OP_ENQUEUE,
    "queue.move": OP_MOVE,
    "queue.dequeue": OP_DEQUEUE,
    "queue.complete": OP_COMPLETE,
}
 
_COMMAND_PREFIX = "work_queue"
_INSTANCE_RE = re.compile(r"instance=(?P<instance>\S+)")
 
 
class LaneIsNotASlotError(QueueBackendError):
    """A work item's lane is not the registered slot this adapter writes to."""
 
 
class MissingRegistryArgumentError(QueueBackendError):
    """An operation needs an argument only the registry backend demands."""
 
 
class UnknownPlanRowError(QueueBackendError):
    """The registry holds no ``plans`` row for this item."""
 
 
class UnsupportedItemKindError(QueueBackendError):
    """This adapter is plan-shaped, because ``related_plan_id`` is."""
 
 
class RegistryQueueBackend:
    """The contract's six operations over the registry's four."""
 
    def __init__(
        self,
        registry: Registry,
        *,
        slot: str,
        actor_slot: str,
        actor_type: str = "human_operator",
    ) -> None:
        func = "RegistryQueueBackend.__init__"
        self.registry = registry
        self.slot = slot
        self.actor_slot = actor_slot
        self.actor_type = actor_type
        _dbg("OC_CHAT.MAIN", func, f"slot={slot} actor={actor_slot}", 2)
 
    # -- reads ------------------------------------------------------------
 
    def status(self) -> QueueState:
        """The three buckets: order from the projection, phase from the log."""
        func = "RegistryQueueBackend.status"
        phases = self._phases()
        instances = self._instances(phases)
        positions = self._pending_positions()
 
        pending = [
            WorkItem(
                kind=KIND_PLAN, item_id=plan_id, instance=instances.get(plan_id)
            )
            for plan_id in positions
            if phases.get(plan_id) in {OP_ENQUEUE, OP_MOVE}
        ]
        in_progress = [
            WorkItem(kind=KIND_PLAN, item_id=plan_id, instance=instances.get(plan_id))
            for plan_id, op_name in phases.items()
            if op_name == OP_DEQUEUE
        ]
        completed = [
            WorkItem(kind=KIND_PLAN, item_id=plan_id, instance=instances.get(plan_id))
            for plan_id, op_name in phases.items()
            if op_name == OP_COMPLETE
        ]
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"{len(pending)}p/{len(in_progress)}i/{len(completed)}c",
            3,
        )
        return QueueState(
            pending=tuple(pending),
            in_progress=tuple(in_progress),
            completed=tuple(completed),
        )
 
    def history(self, item_id: str) -> tuple[QueueOp, ...]:
        """Every queue operation the log holds for *item_id*, oldest first."""
        events = [
            event
            for event in queue_history(self.registry, item_id)
            if event.operation_type in _OP_BY_TYPE
        ]
        return tuple(
            QueueOp(
                seq=event.state_version,
                at=event.ts,
                op=_OP_BY_TYPE[event.operation_type],
                item_id=item_id,
                payload=dict(event.payload),
                actor=event.actor_slot,
            )
            for event in events
        )
 
    # -- writes -----------------------------------------------------------
 
    def enqueue(self, item: WorkItem, position: int | None = None) -> QueueState:
        """Emit ``queue.enqueue`` for *item* at *position*, or at the end.
 
        Raises:
            UnsupportedItemKindError: VALIDATION_ERROR (L3) -- the kind
                is not ``"plan"``.
            LaneIsNotASlotError: VALIDATION_ERROR (L3) -- the lane is
                not this adapter's registered slot.
            UnknownPlanRowError: EXTERNAL_ERROR (L2) -- the registry
                holds no ``plans`` row, so the projection would be
                silently skipped.
            DuplicateItemError / BadPositionError: as the contract says.
        """
        func = "RegistryQueueBackend.enqueue"
        self._check_kind(item.kind)
        self._check_lane(item.lane)
        self._require_plan_row(item.item_id)
        state = self.status()
        held = _bucket_of(state, item.item_id)
        if held is not None and held != "completed":
            raise DuplicateItemError(
                f"{item.item_id!r} is already {held} in this queue"
            )
        index = _checked_index(position, len(state.pending) + 1)
        _dbg("OC_CHAT.MAIN", func, f"{item.item_id} -> position {index}", 2)
        queue_enqueue(
            self.registry,
            plan_id=item.item_id,
            slot=self.slot,
            position=index,
            reason=ENQUEUE_REASON,
            actor_slot=self.actor_slot,
            actor_type=self.actor_type,
            command=self._command(OP_ENQUEUE, item.item_id, reason=ENQUEUE_REASON),
        )
        return self.status()
 
    def move(self, item_id: str, position: int, *, reason: str) -> QueueState:
        """Emit ``queue.move``; the registry's reason is free text, so verbatim."""
        func = "RegistryQueueBackend.move"
        state = self.status()
        current = _pending_index(state, item_id)
        index = _checked_index(position, len(state.pending))
        _dbg("OC_CHAT.MAIN", func, f"{item_id}: {current} -> {index}", 2)
        queue_move(
            self.registry,
            plan_id=item_id,
            from_slot=self.slot,
            to_slot=self.slot,
            from_position=current,
            to_position=index,
            reason=reason,
            actor_slot=self.actor_slot,
            actor_type=self.actor_type,
            command=self._command(OP_MOVE, item_id, reason=reason),
        )
        return self.status()
 
    def dequeue(self, item_id: str, *, instance: str, reason: str) -> QueueState:
        """Emit ``queue.dequeue``, mapping the reason onto the closed set.
 
        The contract's reason and the instance both survive in
        ``command``; see the module docstring for why neither has a
        column of its own.
        """
        func = "RegistryQueueBackend.dequeue"
        state = self.status()
        position = _pending_index(state, item_id)
        mapped = reason if reason in QUEUE_DEQUEUE_REASONS else DEQUEUE_REASON_DEFAULT
        _dbg("OC_CHAT.MAIN", func, f"reason {reason!r} -> {mapped!r}", 3)
        queue_dequeue(
            self.registry,
            plan_id=item_id,
            slot=self.slot,
            position=position,
            reason=mapped,
            actor_slot=self.actor_slot,
            actor_type=self.actor_type,
            command=self._command(
                OP_DEQUEUE, item_id, reason=reason, instance=instance
            ),
        )
        return self.status()
 
    def complete(
        self,
        item_id: str,
        *,
        reason: str,
        round_no: int | None = None,
        merge_commit_sha: str | None = None,
    ) -> QueueState:
        """Emit ``queue.complete``. Idempotent, like the file backend.
 
        Raises:
            MissingRegistryArgumentError: VALIDATION_ERROR (L3) --
                ``queue_complete`` requires both ``round_no`` and
                ``merge_commit_sha`` and this adapter will not invent
                either.
            UnknownItemError: VALIDATION_ERROR (L3) -- the queue holds
                no such item.
        """
        func = "RegistryQueueBackend.complete"
        state = self.status()
        held = _bucket_of(state, item_id)
        if held == "completed":
            _dbg("OC_CHAT.MAIN", func, f"{item_id} already completed; no-op", 2)
            return state
        if held is None:
            raise UnknownItemError(
                f"{item_id!r} is not pending or in progress in this queue"
            )
        if round_no is None or merge_commit_sha is None:
            raise MissingRegistryArgumentError(
                "the registry backend's complete requires round_no and "
                "merge_commit_sha; queue_complete demands both and they are not "
                "packed into reason. The file backend accepts and ignores them."
            )
        queue_complete(
            self.registry,
            plan_id=item_id,
            round_no=round_no,
            merge_commit_sha=merge_commit_sha,
            actor_slot=self.actor_slot,
            actor_type=self.actor_type,
            command=self._command(OP_COMPLETE, item_id, reason=reason),
        )
        return self.status()
 
    # -- internals --------------------------------------------------------
 
    def _check_kind(self, kind: str) -> None:
        """Refuse a non-plan kind by name; ``related_plan_id`` is plan-shaped."""
        if kind != KIND_PLAN:
            raise UnsupportedItemKindError(
                f"the registry backend accepts kind {KIND_PLAN!r} only; got "
                f"{kind!r}. The operation log's plan column is related_plan_id, "
                "so a second kind needs a registry change, not an adapter one."
            )
 
    def _check_lane(self, lane: str | None) -> None:
        """Refuse a lane that is not this adapter's registered slot."""
        if lane is not None and lane != self.slot:
            raise LaneIsNotASlotError(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Operator-facing refusal prose naming the schema constraint; no credential content"
                f"lane {lane!r} cannot be a registry slot: plans.assigned_slot "
                "REFERENCES agents(slot) with foreign keys on, so a free-text "
                "slot is refused with 'FOREIGN KEY constraint failed'. This "
                f"adapter writes to the registered slot {self.slot!r}; pass that "
                "lane or none."
            )
 
    def _require_plan_row(self, plan_id: str) -> None:
        """Refuse an item the ``plans`` table does not hold.
 
        Without this the operation would be written and the projection
        silently skipped -- ``_project_queue_enqueue`` returns early on
        a missing plan row -- leaving a log entry that changed nothing.
        """
        with self._connect() as conn:
            row = conn.execute(
                "SELECT 1 FROM plans WHERE plan_id = ?", (plan_id,)
            ).fetchone()
        if row is None:
            raise UnknownPlanRowError(
                f"the registry holds no plans row for {plan_id!r}; enqueue it "
                "through plan.create first, or the projection is skipped in "
                "silence."
            )
 
    def _command(self, op_name: str, item_id: str, **fields: object) -> str:
        """The deterministic ``command`` text for one operation."""
        parts = [f"{_COMMAND_PREFIX} {op_name}", f"item={item_id}"]
        if "instance" in fields and fields["instance"]:
            parts.append(f"instance={fields['instance']}")
        if fields.get("reason"):
            parts.append(f"reason={fields['reason']}")
        return " ".join(parts)
 
    def _connect(self) -> object:
        """A read connection through the storage backend."""
        connect = getattr(self.registry.backend, "connect", None)
        if connect is None:
            raise QueueBackendError(
                "the registry backend does not expose read connections"
            )
        return connect()
 
    def _queue_rows(self) -> list[tuple[str, str, str]]:
        """``(plan_id, operation_type, command)`` for every queue op, in order."""
        placeholders = ", ".join("?" for _ in _OP_BY_TYPE)
        with self._connect() as conn:
            rows = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
                "SELECT related_plan_id, operation_type, command FROM operation_log "
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
                f"WHERE operation_type IN ({placeholders}) "
                "AND related_plan_id IS NOT NULL ORDER BY state_version",
                tuple(sorted(_OP_BY_TYPE)),
            ).fetchall()
        return [
            (str(row["related_plan_id"]), str(row["operation_type"]), str(row["command"] or ""))
            for row in rows
        ]
 
    def _phases(self) -> dict[str, str]:
        """Each item's phase: the contract op name of its LATEST queue row."""
        phases: dict[str, str] = {}
        for plan_id, operation_type, _command in self._queue_rows():
            phases[plan_id] = _OP_BY_TYPE[operation_type]
        return phases
 
    def _instances(self, phases: dict[str, str]) -> dict[str, str]:
        """Each item's instance, read back out of its latest dequeue command."""
        instances: dict[str, str] = {}
        for plan_id, operation_type, command in self._queue_rows():
            if operation_type != "queue.dequeue":
                continue
            match = _INSTANCE_RE.search(command)
            if match:
                instances[plan_id] = match.group("instance")
            elif plan_id in instances:
                del instances[plan_id]
        # An item that came back into pending is no longer held by an
        # instance, so a stale name must not survive a re-enqueue.
        return {
            plan_id: instance
            for plan_id, instance in instances.items()
            if phases.get(plan_id) in {OP_DEQUEUE, OP_COMPLETE}
        }
 
    def _pending_positions(self) -> list[str]:
        """Plan ids with a queue position in this slot, in position order."""
        with self._connect() as conn:
            rows = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection query over the plans table; no credential content"
                "SELECT plan_id FROM plans WHERE assigned_slot = ? "
                "AND queue_position IS NOT NULL ORDER BY queue_position",
                (self.slot,),
            ).fetchall()
        return [str(row["plan_id"]) for row in rows]
 
 
def _bucket_of(state: QueueState, item_id: str) -> str | None:
    """Which bucket holds *item_id*, in words, or ``None``."""
    for label, attribute in (
        ("pending", "pending"),
        ("in progress", "in_progress"),
        ("completed", "completed"),
    ):
        if any(item.item_id == item_id for item in getattr(state, attribute)):
            return label
    return None
 
 
def _pending_index(state: QueueState, item_id: str) -> int:
    """The 1-based position of *item_id* in pending, or a named refusal."""
    for index, item in enumerate(state.pending, start=1):
        if item.item_id == item_id:
            return index
    held = _bucket_of(state, item_id)
    where = f"it is {held}" if held else "the queue does not hold it"
    raise UnknownItemError(f"{item_id!r} is not pending: {where}")
 
 
def _checked_index(position: int | None, upper: int) -> int:
    """*position* as a 1-based index in ``1..upper``, or the end."""
    if position is None:
        return max(upper, 1)
    if not 1 <= position <= max(upper, 1):
        raise BadPositionError(f"position {position} is outside 1..{max(upper, 1)}")
    return position
 
 
__all__ = [
    "DEQUEUE_REASON_DEFAULT",
    "ENQUEUE_REASON",
    "LaneIsNotASlotError",
    "MissingRegistryArgumentError",
    "RegistryQueueBackend",
    "UnknownPlanRowError",
    "UnsupportedItemKindError",
]