work_queue.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/work_queue.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_orchestrator/tests/test_258_work_queue.py"
"""
The work queue's contract: one work item, one operation set, two backends.
 
Purpose
-------
Plan 258 P1, from the factory-decomposition design note §5. The unit of
work is a work item WITH A KIND -- ``plan`` is the first kind and the
only one this release accepts -- and the queue has one set of
operations that two backends implement: a file backend that is the
default and the mode the package is tested in, and a registry backend
that inherits signing, audit and checkpoints from the registry because
that is where those already live. Neither backend is a special case,
so the operations, the item, the state and the operation record are
declared exactly once, here, and neither backend is the place where
the shape gets decided.
 
This module is deliberately a LEAF: it imports nothing from any
consumer -- no registry, no chat client -- and nothing outside the
standard library besides the diagnostics runtime. A witness measures
that with the committed import-graph extractor rather than asserting
it by hand. Plan 271 moved it here from ``oc_chat/core/work_queue.py``,
which is now a rebinding shim over this module.
 
Responsibilities
----------------
- :class:`WorkItem` -- the unit of work: a kind, an id, and three
  optional free-text fields.
- :func:`check_kind` -- the ONE place a kind is checked, so a second
  kind is one edit to :data:`SUPPORTED_KINDS`.
- :class:`QueueBackend` -- the six operations, as a runtime-checkable
  :class:`typing.Protocol`, so a backend is a plain class.
- :class:`QueueState` -- three ordered tuples: pending, in progress,
  completed.
- :class:`QueueOp` -- one operation, as the log records it.
- :class:`QueueError` and its named subclasses -- every refusal.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (module load)
    L3 - semantic details (a refusal, with what it refused)
    L4 - deep tracing
 
Contracts
---------
- ``kind`` is a plain string checked by name against
  :data:`SUPPORTED_KINDS`, read at call time. Widening that set is the
  whole edit a second kind needs in this module. The design note names
  the registry's plan-shaped ``related_plan_id`` column as the seam a
  second kind will also find; that is named, not built.
- ``item_id`` for a ``plan`` is the plan id, ``NNN-stem``. ``lane`` is
  a free-text preference such as a model family or a harness name and
  is NEVER a roster letter -- the letter model was retired by plan 223
  and is not revived. ``instance`` is the ``agent-NNN`` identity
  allocated at dispatch; identity is plan-AND-instance, so an item may
  be re-dispatched and the log then shows both instances.
- Positions are 1-BASED, matching the registry's ``queue_position``.
- :class:`QueueOp.at` is ISO 8601 with an explicit offset.
  :class:`QueueOp.actor` is free text and is NEVER signed here: a
  backend that signs (the registry one) inherits signing from the
  registry, and a backend that does not (the file one) must not
  pretend to.
- Every refusal is a named exception under :class:`QueueError`.
  :class:`QueueBackendError` is the separate arm a caller maps to its
  own "the backend said no" exit code; the other three are semantic.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
 
from __future__ import annotations
 
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "work_queue loaded", 4)
 
 
# ── Kinds ────────────────────────────────────────────────────────────────
 
 
#: The first kind, and the only one this plan accepts.
KIND_PLAN = "plan"
 
#: The kinds a :class:`WorkItem` may carry. Widening this set is the
#: whole edit a second kind needs here -- :func:`check_kind` reads it at
#: call time and every construction path goes through that function.
SUPPORTED_KINDS: frozenset[str] = frozenset({KIND_PLAN})
 
 
# ── Operation names ──────────────────────────────────────────────────────
 
 
OP_ENQUEUE = "enqueue"
OP_MOVE = "move"
OP_DEQUEUE = "dequeue"
OP_COMPLETE = "complete"
 
#: The four op names, in lifecycle order. Both backends report these, so
#: a history read through one is comparable with a history read through
#: the other -- which is what the plan-258 equivalence witness asserts.
OP_NAMES: tuple[str, ...] = (OP_ENQUEUE, OP_MOVE, OP_DEQUEUE, OP_COMPLETE)
 
 
# ── Refusals ─────────────────────────────────────────────────────────────
 
 
class QueueError(Exception):
    """Base class for every refusal the queue makes."""
 
 
class UnknownKindError(QueueError):
    """A work item named a kind this release does not accept."""
 
 
class UnknownItemError(QueueError):
    """An operation named an item the queue does not hold."""
 
 
class DuplicateItemError(QueueError):
    """An enqueue named an item the queue already holds."""
 
 
class BadPositionError(QueueError):
    """A position is not a 1-based index into the pending list."""
 
 
class QueueBackendError(QueueError):
    """The BACKEND refused, rather than the operation being meaningless.
 
    A corrupt log line, a storage constraint, a missing argument the
    storage demands: things a different backend would not have refused.
    Callers that map refusals onto exit codes give this its own code.
    """
 
 
def check_kind(kind: str) -> None:
    """Refuse *kind* by name unless it is in :data:`SUPPORTED_KINDS`.
 
    The single check site. :data:`SUPPORTED_KINDS` is read here, at call
    time, so widening it is the whole edit a second kind needs.
 
    Raises:
        UnknownKindError: VALIDATION_ERROR (L3) -- *kind* is not
            accepted; the message names it and the accepted set.
    """
    func = "check_kind"
    if kind not in SUPPORTED_KINDS:
        accepted = ", ".join(sorted(SUPPORTED_KINDS))
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: kind {kind!r}", 3)
        raise UnknownKindError(
            f"unknown work-item kind {kind!r}; this release accepts: {accepted}"
        )
 
 
# ── Value types ──────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class WorkItem:
    """One unit of work: a kind, an id, and three free-text fields."""
 
    #: ``"plan"`` today; checked by name in :func:`check_kind`.
    kind: str
    #: For a plan, the plan id ``NNN-stem``.
    item_id: str
    #: A free-text preference -- a model family, a harness name. NEVER a
    #: roster letter; the letter model is retired and is not revived.
    lane: str | None = None
    #: The ``agent-NNN`` identity, once dispatched. Identity is
    #: plan-AND-instance: a re-dispatch sets a second instance and the
    #: log shows both.
    instance: str | None = None
    #: Free text for the operator.
    note: str | None = None
 
    def __post_init__(self) -> None:
        """Refuse an unaccepted kind at construction, by name."""
        check_kind(self.kind)
 
 
@dataclass(frozen=True)
class QueueState:
    """The queue's three buckets, each in its own order."""
 
    pending: tuple[WorkItem, ...] = ()
    in_progress: tuple[WorkItem, ...] = ()
    completed: tuple[WorkItem, ...] = ()
 
 
@dataclass(frozen=True)
class QueueOp:
    """One recorded operation, in the shape the log stores."""
 
    #: 1-based, monotonic within one backend's log.
    seq: int
    #: ISO 8601 with an explicit offset.
    at: str
    #: One of :data:`OP_NAMES`.
    op: str
    #: The item the operation was about.
    item_id: str
    #: Whatever the operation carried; never interpreted here.
    payload: Mapping[str, object] = field(default_factory=dict)
    #: Free text. NEVER signed here -- see the module contract.
    actor: str = ""
 
 
# ── The operation set ────────────────────────────────────────────────────
 
 
@runtime_checkable
class QueueBackend(Protocol):
    """The six operations, defined once, implemented by both backends."""
 
    def enqueue(self, item: WorkItem, position: int | None = None) -> QueueState:
        """Put *item* into pending at *position*, or at the end."""
 
    def move(self, item_id: str, position: int, *, reason: str) -> QueueState:
        """Move a pending item to *position*, for a stated *reason*."""
 
    def dequeue(self, item_id: str, *, instance: str, reason: str) -> QueueState:
        """Take an item out of pending into progress, naming *instance*."""
 
    def complete(self, item_id: str, *, reason: str) -> QueueState:
        """Finish an item, for a stated *reason*."""
 
    def status(self) -> QueueState:
        """The three buckets as they stand."""
 
    def history(self, item_id: str) -> tuple[QueueOp, ...]:
        """Every recorded operation about *item_id*, oldest first."""
 
 
__all__ = [
    "KIND_PLAN",
    "OP_COMPLETE",
    "OP_DEQUEUE",
    "OP_ENQUEUE",
    "OP_MOVE",
    "OP_NAMES",
    "SUPPORTED_KINDS",
    "BadPositionError",
    "DuplicateItemError",
    "QueueBackend",
    "QueueBackendError",
    "QueueError",
    "QueueOp",
    "QueueState",
    "UnknownItemError",
    "UnknownKindError",
    "WorkItem",
    "check_kind",
]