handover.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/handover.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_handover.py"
"""
Orchestrator handover state machine.
 
Purpose
-------
Provide the M11.5 audited handover operations that prevent two
orchestrators from writing round state at the same time.
 
Responsibilities
----------------
- Emit ``role.handover.start`` and ``role.handover.complete`` operations.
- Recover the start payload needed by accept calls.
- Keep durable state projection inside the SQLite backend.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- ``role.handover.start`` enters the requested/draining state.
- ``role.handover.complete`` makes the incoming orchestrator the holder.
- Round writes are rejected unless the durable state permits the actor.
 
Dependencies
------------
- json (decode the stored handover-start arguments)
- oc_chat.registry.api (Registry that executes the handover domain commands)
- oc_chat.registry.errors (HandoverStateError for an unknown handover)
- oc_chat.registry.backends.base (OperationLogRow returned by both operations)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import json
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.base import OperationLogRow
from oc_chat.registry.errors import HandoverStateError
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def initiate_handover(
    registry: Registry,
    *,
    from_slot: str,
    to_slot: str,
    reason: str,
    actor_slot: str | None = None,
    actor_type: str = "orchestrator_agent",
    command: str | None = None,
) -> OperationLogRow:
    """Emit a handover-start operation."""
 
    return registry.execute_domain_command(
        "role.handover.start",
        actor_slot or from_slot,
        {
            "version": 1,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "reason": reason,
        },
        actor_type=actor_type,
        command=command or "handover.initiate",
    )
 
 
def accept_handover(
    registry: Registry,
    *,
    handover_id: int,
    actor_slot: str,
    actor_type: str = "orchestrator_agent",
    command: str | None = None,
) -> OperationLogRow:
    """Emit a handover-complete operation for ``handover_id``."""
 
    start = _handover_start(registry, handover_id)
    if start is None:
        raise HandoverStateError(
            state="unknown_handover",
            actor_slot=actor_slot,
            operation_type="role.handover.complete",
            handover_id=handover_id,
        )
    from_slot = str(start["from_slot"])
    to_slot = str(start["to_slot"])
    started_at = str(start["started_at"])
    return registry.execute_domain_command(
        "role.handover.complete",
        actor_slot,
        {
            "version": 1,
            "from_slot": from_slot,
            "to_slot": to_slot,
            "started_at": started_at,
            "completed_at": registry.clock.now_wall(),
            "parent_operation_id": handover_id,
        },
        actor_type=actor_type,
        command=command or "handover.accept",
        parent_operation_id=handover_id,
    )
 
 
def _handover_start(
    registry: Registry,
    handover_id: int,
) -> Mapping[str, object] | None:
    connect_func = getattr(registry.backend, "connect", None)
    if connect_func is None:
        return None
    with connect_func() as conn:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection and operation names are not secrets"
            "SELECT ts, args_json FROM operation_log "
            "WHERE id = ? AND operation_type = 'role.handover.start'",
            (handover_id,),
        ).fetchone()
    if row is None:
        return None
    payload = json.loads(str(row["args_json"]))
    if not isinstance(payload, dict):
        return None
    payload["started_at"] = row["ts"]
    return payload