readers.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/readers.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_readers.py"
"""
Read-only registry projection helpers.
 
Purpose
-------
Expose typed reader libraries for queue history, plan lifecycle, plan
inspection, registry status, and actor identity.
 
Responsibilities
----------------
- Read current-state and operation-log projections without mutating state.
- Decode canonical JSON payloads into typed dataclass records.
- Keep SQL limited to SELECT statements.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Reader functions never call ``execute_domain_command``.
- Missing plan rows return ``None`` for explicit plan views.
- Unknown actor lookups raise ``UnknownActorError``.
 
Dependencies
------------
- json (payload decoding)
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.dataclasses (typed read models)
- oc_chat.registry.errors (typed unknown-actor error)
- oc_chat.registry.api (Registry facade whose backend connection the readers query)
"""
from __future__ import annotations
 
import json
from typing import Iterable, Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.dataclasses import (
    ActorView,
    CurrentStatus,
    PlanAssignmentEvent,
    PlanStatusEvent,
    PlanTimeline,
    PlanView,
    QueueEvent,
)
from oc_chat.registry.errors import UnknownActorError
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
QUEUE_HISTORY_TYPES = {
    "queue.enqueue",
    "queue.move",
    "queue.dequeue",
    "queue.complete",
    "plan.assign",
    "plan.set_status",
    "migrate.import.queue_entry",
}
 
 
def queue_history(registry: Registry, plan_id: str) -> list[QueueEvent]:
    """Return typed queue history events for ``plan_id``."""
 
    placeholders = ", ".join("?" for _ in QUEUE_HISTORY_TYPES)
    sql = (
        # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
        "SELECT ts, operation_type, args_json, state_version, actor_slot, "
        "actor_type, related_plan_id, related_round FROM operation_log "
        f"WHERE related_plan_id = ? AND operation_type IN ({placeholders}) "
        "ORDER BY state_version"
    )
    with _connect(registry) as conn:
        rows = conn.execute(sql, (plan_id, *sorted(QUEUE_HISTORY_TYPES))).fetchall()
    return [
        QueueEvent(
            ts=str(row["ts"]),
            operation_type=str(row["operation_type"]),
            payload=_payload(row["args_json"]),
            state_version=int(row["state_version"]),
            actor_slot=str(row["actor_slot"]),
            actor_type=str(row["actor_type"]),
            related_plan_id=row["related_plan_id"],
            related_round=row["related_round"],
            is_migration_synthetic=str(row["operation_type"]).startswith("migrate."),
        )
        for row in rows
    ]
 
 
def plan_lifecycle(registry: Registry, plan_id: str) -> PlanTimeline:
    """Return typed status and assignment events for ``plan_id``."""
 
    with _connect(registry) as conn:
        status_rows = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
            "SELECT plan_id, from_status, to_status, actor_slot, source, "
            "created_at, note FROM plan_status_events "
            "WHERE plan_id = ? ORDER BY created_at, id",
            (plan_id,),
        ).fetchall()
        op_rows = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
            "SELECT ts, operation_type, args_json, state_version FROM operation_log "
            "WHERE related_plan_id = ? AND operation_type = 'plan.assign' "
            "ORDER BY state_version",
            (plan_id,),
        ).fetchall()
 
    status_events = [
        PlanStatusEvent(
            ts=str(row["created_at"]),
            plan_id=str(row["plan_id"]),
            from_status=row["from_status"],
            to_status=str(row["to_status"]),
            actor_slot=row["actor_slot"],
            source=row["source"],
            note=row["note"],
        )
        for row in status_rows
    ]
    assignment_events: list[PlanAssignmentEvent] = []
    for row in op_rows:
        payload = _payload(row["args_json"])
        if row["operation_type"] == "plan.assign":
            assignment_events.append(
                PlanAssignmentEvent(
                    ts=str(row["ts"]),
                    plan_id=str(payload["plan_id"]),
                    from_slot=_optional_str(payload.get("from_slot")),
                    to_slot=_optional_str(payload.get("to_slot")),
                    queue_position=_optional_int(payload.get("queue_position")),
                    state_version=int(row["state_version"]),
                )
            )
    return PlanTimeline(
        plan_id=plan_id,
        status_events=tuple(status_events),
        assignment_events=tuple(assignment_events),
    )
 
 
def show_plan(registry: Registry, plan_id: str) -> PlanView | None:
    """Return the current projected plan row, or ``None`` if absent."""
 
    with _connect(registry) as conn:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
            "SELECT plan_id, plan_number, title, status, assigned_slot, "
            "queue_position, branch, worktree, body_path, body_sha256 "
            "FROM plans WHERE plan_id = ?",
            (plan_id,),
        ).fetchone()
    if row is None:
        return None
    return PlanView(
        plan_id=str(row["plan_id"]),
        plan_number=int(row["plan_number"]),
        title=str(row["title"]),
        status=str(row["status"]),
        assigned_slot=row["assigned_slot"],
        queue_position=row["queue_position"],
        branch=row["branch"],
        worktree=row["worktree"],
        body_path=str(row["body_path"]),
        body_sha256=row["body_sha256"],
    )
 
 
def current_status(registry: Registry) -> CurrentStatus:
    """Return compact registry status counters and active runtime rows."""
 
    with _connect(registry) as conn:
        plan_counts = {
            str(row["status"]): int(row["count"])
            for row in conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
                "SELECT status, COUNT(*) AS count FROM plans GROUP BY status"
            ).fetchall()
        }
        open_rounds = tuple(
            _row_dict(row)
            for row in conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
                "SELECT round_no, status, cleanup_status, orchestrator_slot "
                "FROM rounds WHERE status IN ('open', 'closing', 'committed') "
                "ORDER BY round_no"
            ).fetchall()
        )
        active_leases = tuple(
            _row_dict(row)
            for row in conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
                "SELECT id, holder_slot, lease_expires_at, last_heartbeat_at "
                "FROM leases WHERE holder_slot IS NOT NULL ORDER BY id"
            ).fetchall()
        )
        state_row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
            "SELECT id, orchestrator_slot, handover_status, ready_to_handover, "
            "updated_at, updated_by FROM orchestration_state WHERE id = 1"
        ).fetchone()
    return CurrentStatus(
        plan_counts=plan_counts,
        open_rounds=open_rounds,
        active_leases=active_leases,
        orchestration_state=None if state_row is None else _row_dict(state_row),
    )
 
 
def whoami(registry: Registry, actor_slot: str) -> ActorView:
    """Return current actor metadata for ``actor_slot``."""
 
    with _connect(registry) as conn:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection queries are not secrets"
            "SELECT slot, slot_kind, identity_kind, model_class, "
            "orchestrator_eligible, active, provider, profile "
            "FROM agents WHERE slot = ?",
            (actor_slot,),
        ).fetchone()
    if row is None:
        raise UnknownActorError(actor_slot)
    return ActorView(
        slot=str(row["slot"]),
        slot_kind=str(row["slot_kind"]),
        identity_kind=str(row["identity_kind"]),
        model_class=str(row["model_class"]),
        orchestrator_eligible=bool(row["orchestrator_eligible"]),
        active=bool(row["active"]),
        provider=row["provider"],
        profile=row["profile"],
    )
 
 
def _connect(registry: Registry) -> object:
    connect = getattr(registry.backend, "connect", None)
    if connect is None:
        raise TypeError("registry backend does not expose read connections")
    return connect()
 
 
def _payload(value: object) -> Mapping[str, object]:
    if value is None:
        return {}
    payload = json.loads(str(value))
    if not isinstance(payload, dict):
        return {}
    return payload
 
 
def _row_dict(row: object) -> dict[str, object]:
    keys: Iterable[str] = row.keys()
    return {key: row[key] for key in keys}
 
 
def _optional_str(value: object) -> str | None:
    return None if value is None else str(value)
 
 
def _optional_int(value: object) -> int | None:
    return None if value is None else int(value)