queue.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/queue.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_domain_queue.py"
"""
Registry-native queue commands.
 
Purpose
-------
Expose queue mutation and history commands beneath
``python -m oc_chat.registry queue``.
 
Responsibilities
----------------
- Parse queue domain command arguments.
- Resolve the registry and actor slot.
- Dispatch to the domain API and reader library.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Mutating commands emit exactly one operation-log row through the seam.
- ``history`` is read-only and returns decoded typed events.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.registry.domain (queue mutation API)
- oc_chat.registry.readers (queue history reader)
- oc_chat.registry (Registry facade for queue mutations and the history reader)
"""
from __future__ import annotations
 
import argparse
import json
from typing import Callable
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.domain import (
    queue_complete,
    queue_dequeue,
    queue_enqueue,
    queue_move,
)
from oc_chat.registry.readers import queue_history
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_queue_subcommands(queue_parser: argparse.ArgumentParser) -> None:
    """Attach registry-native queue subcommands."""
 
    queue_parser.set_defaults(handler=_cmd_queue_group, queue_parser=queue_parser)
    subcommands = queue_parser.add_subparsers(dest="queue_command")
 
    enqueue = subcommands.add_parser("enqueue", help="Emit queue.enqueue")
    _add_common(enqueue)
    enqueue.add_argument("plan_id")
    enqueue.add_argument("--slot", required=True)
    enqueue.add_argument("--position", required=True, type=int)
    enqueue.add_argument(
        "--reason",
        required=True,
        choices=[
            "initial enqueue",
            "promoted_from_partial",
            "restored_from_aborted_round",
        ],
    )
    enqueue.set_defaults(handler=_cmd_enqueue)
 
    move = subcommands.add_parser("move", help="Emit queue.move")
    _add_common(move)
    move.add_argument("plan_id")
    move.add_argument("--from-slot", required=True)
    move.add_argument("--to-slot", required=True)
    move.add_argument("--from-position", required=True, type=int)
    move.add_argument("--to-position", required=True, type=int)
    move.add_argument("--reason", required=True)
    move.set_defaults(handler=_cmd_move)
 
    dequeue = subcommands.add_parser("dequeue", help="Emit queue.dequeue")
    _add_common(dequeue)
    dequeue.add_argument("plan_id")
    dequeue.add_argument("--slot", required=True)
    dequeue.add_argument("--position", required=True, type=int)
    dequeue.add_argument(
        "--reason",
        required=True,
        choices=[
            "transition_to_failed",
            "transition_to_integrating",
            "operator_dequeue",
        ],
    )
    dequeue.set_defaults(handler=_cmd_dequeue)
 
    complete = subcommands.add_parser("complete", help="Emit queue.complete")
    _add_common(complete)
    complete.add_argument("plan_id")
    complete.add_argument("--round-no", required=True, type=int)
    complete.add_argument("--merge-commit-sha", required=True)
    complete.set_defaults(handler=_cmd_complete)
 
    history = subcommands.add_parser("history", help="Show queue history")
    history.add_argument("plan_id")
    history.add_argument("--registry", help="Registry SQLite path.")
    history.add_argument("--json", action="store_true")
    history.set_defaults(handler=_cmd_history)
 
 
def _cmd_queue_group(args: argparse.Namespace) -> int:
    parser = args.queue_parser
    parser.print_help()
    return 1
 
 
def _cmd_enqueue(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: queue_enqueue(
            registry,
            plan_id=args.plan_id,
            slot=args.slot,
            position=args.position,
            reason=args.reason,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="queue enqueue",
        ),
    )
 
 
def _cmd_move(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: queue_move(
            registry,
            plan_id=args.plan_id,
            from_slot=args.from_slot,
            to_slot=args.to_slot,
            from_position=args.from_position,
            to_position=args.to_position,
            reason=args.reason,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="queue move",
        ),
    )
 
 
def _cmd_dequeue(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: queue_dequeue(
            registry,
            plan_id=args.plan_id,
            slot=args.slot,
            position=args.position,
            reason=args.reason,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="queue dequeue",
        ),
    )
 
 
def _cmd_complete(args: argparse.Namespace) -> int:
    return _emit(
        args,
        lambda registry, actor_slot: queue_complete(
            registry,
            plan_id=args.plan_id,
            round_no=args.round_no,
            merge_commit_sha=args.merge_commit_sha,
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="queue complete",
        ),
    )
 
 
def _cmd_history(args: argparse.Namespace) -> int:
    registry, _backend = _registry(args.registry)
    events = queue_history(registry, args.plan_id)
    if args.json:
        print(json.dumps([event.__dict__ for event in events], indent=2, sort_keys=True))
    else:
        for event in events:
            print(
                f"{event.state_version}: {event.operation_type} "
                f"{dict(event.payload)}"
            )
    return 0
 
 
def _emit(
    args: argparse.Namespace,
    emit: Callable[[Registry, str], object],
) -> int:
    registry, backend = _registry(args.registry)
    actor_slot = args.actor_slot or backend.active_operator_slot()
    if actor_slot is None:
        print("no active operator slot found")
        return 1
    row = emit(registry, actor_slot)
    print(f"operation {getattr(row, 'id')} state_version={getattr(row, 'state_version')}")
    return 0
 
 
def _registry(registry_path: str | None) -> tuple[Registry, SQLiteBackend]:
    location = discover_registry_location(registry_path=registry_path)
    backend = SQLiteBackend(location.registry_path)
    return Registry(backend), backend
 
 
def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--registry", help="Registry SQLite path.")
    parser.add_argument("--actor-slot", help="Actor slot. Defaults to active operator.")