dataclasses.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/dataclasses.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_readers.py"
"""
Typed read models for the orchestration registry.
 
Purpose
-------
Provide stable dataclasses for registry reader APIs and native CLI output.
 
Responsibilities
----------------
- Represent queue history and plan lifecycle events without exposing raw JSON.
- Represent compact plan, actor, and orchestration status projections.
- Keep read-model types dependency-light and immutable.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Dataclasses are frozen and safe to return from reader functions.
- Payload mappings are already decoded from canonical JSON by readers.
 
Dependencies
------------
- dataclasses (immutable record types)
- typing (read-model payload annotations)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass
from typing import Mapping
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class QueueEvent:
    """One queue-related audit event for a plan."""
 
    ts: str
    operation_type: str
    payload: Mapping[str, object]
    state_version: int
    actor_slot: str
    actor_type: str
    related_plan_id: str | None
    related_round: int | None
    is_migration_synthetic: bool = False
 
 
@dataclass(frozen=True)
class PlanStatusEvent:
    """One status event from the projected status table or operation log."""
 
    ts: str
    plan_id: str
    from_status: str | None
    to_status: str
    actor_slot: str | None
    source: str | None
    note: str | None
    state_version: int | None = None
 
 
@dataclass(frozen=True)
class PlanAssignmentEvent:
    """One assignment-related audit event for a plan."""
 
    ts: str
    plan_id: str
    from_slot: str | None
    to_slot: str | None
    queue_position: int | None
    state_version: int
 
 
@dataclass(frozen=True)
class PlanTimeline:
    """Combined lifecycle timeline for a plan."""
 
    plan_id: str
    status_events: tuple[PlanStatusEvent, ...]
    assignment_events: tuple[PlanAssignmentEvent, ...]
 
 
@dataclass(frozen=True)
class PlanView:
    """Current projected plan row."""
 
    plan_id: str
    plan_number: int
    title: str
    status: str
    assigned_slot: str | None
    queue_position: int | None
    branch: str | None
    worktree: str | None
    body_path: str
    body_sha256: str | None
 
 
@dataclass(frozen=True)
class ActorView:
    """Current projected actor/agent row."""
 
    slot: str
    slot_kind: str
    identity_kind: str
    model_class: str
    orchestrator_eligible: bool
    active: bool
    provider: str | None
    profile: str | None
 
 
@dataclass(frozen=True)
class CurrentStatus:
    """Compact registry status projection."""
 
    plan_counts: Mapping[str, int]
    open_rounds: tuple[Mapping[str, object], ...]
    active_leases: tuple[Mapping[str, object], ...]
    orchestration_state: Mapping[str, object] | None