errors.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/errors.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_seam_and_locks.py"
"""
Typed registry errors.
 
Purpose
-------
Expose precise exception types for the M11 registry mutating seam and
lock primitives.
 
Responsibilities
----------------
- Define typed pre-commit validation failures.
- Define lock contention and ownership failures with useful attributes.
- Keep error classes independent from SQLite implementation details.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Pre-commit failures are raised before the backend commits.
- Lock errors carry enough data for callers to render operator messages.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- dataclasses (dataclass decorator for the error types' attribute fields)
"""
from __future__ import annotations
 
from dataclasses import dataclass
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
class RegistryError(RuntimeError):
    """Base class for registry API errors."""
 
 
@dataclass
class ActorTypeMismatchError(RegistryError):
    """Raised when supplied actor_type disagrees with agents.slot_kind."""
 
    actor_slot: str
    expected_actor_type: str
    supplied_actor_type: str
 
    def __post_init__(self) -> None:
        super().__init__(
            "actor_type mismatch for "
            f"{self.actor_slot!r}: expected {self.expected_actor_type!r}, "
            f"got {self.supplied_actor_type!r}"
        )
 
 
@dataclass
class ActorRoleMismatchError(RegistryError):
    """Raised when a slot kind is not allowed to author an operation."""
 
    operation_type: str
    actor_slot: str
    slot_kind: str
    allowed_slot_kinds: tuple[str, ...]
 
    def __post_init__(self) -> None:
        super().__init__(
            f"{self.actor_slot!r} ({self.slot_kind}) cannot author "
            f"{self.operation_type!r}; allowed slot kinds: "
            f"{', '.join(self.allowed_slot_kinds) or 'none'}"
        )
 
 
@dataclass
class HmacVerifyError(RegistryError):
    """Raised when HMAC verification fails before the main write transaction."""
 
    attempted_slot: str | None
    hmac_key_id: str | None
    command: str
    failure_reason: str
 
    def __post_init__(self) -> None:
        super().__init__(
            f"HMAC verification failed for {self.command!r}: "
            f"{self.failure_reason}"
        )
 
 
@dataclass
class HandoverStateError(RegistryError):
    """Raised when an operation violates the orchestrator handover state."""
 
    state: str
    actor_slot: str
    operation_type: str
    handover_id: int | None = None
 
    def __post_init__(self) -> None:
        suffix = "" if self.handover_id is None else f" ({self.handover_id})"
        super().__init__(
            f"handover state {self.state!r}{suffix} rejects "
            f"{self.operation_type!r} from {self.actor_slot!r}"
        )
 
 
@dataclass
class RotationInFlightError(RegistryError):
    """Raised when a slot already has a pending key rotation."""
 
    slot: str
 
    def __post_init__(self) -> None:
        super().__init__(f"credential rotation is already in flight for {self.slot!r}")
 
 
@dataclass
class LockHeldError(RegistryError):
    """Raised when a named lock is already held."""
 
    name: str
    owner_slot: str
    expires_at: str | None
 
    def __post_init__(self) -> None:
        super().__init__(
            f"lock {self.name!r} is held by {self.owner_slot!r} "
            f"until {self.expires_at!r}"
        )
 
 
@dataclass
class LockOwnerError(RegistryError):
    """Raised when a caller tries to release or renew another slot's lock."""
 
    name: str
    owner_slot: str
    actual_owner_slot: str | None
 
    def __post_init__(self) -> None:
        super().__init__(
            f"lock {self.name!r} is owned by {self.actual_owner_slot!r}, "
            f"not {self.owner_slot!r}"
        )
 
 
@dataclass
class PayloadVersionMismatchError(RegistryError):
    """Raised when payload['version'] and operation payload_version differ."""
 
    operation_type: str
    payload_version: int
    supplied_version: object
 
    def __post_init__(self) -> None:
        super().__init__(
            f"{self.operation_type!r} payload version mismatch: "
            f"expected {self.payload_version!r}, got {self.supplied_version!r}"
        )
 
 
@dataclass
class PayloadSchemaViolationError(RegistryError):
    """Raised for the minimal payload-shape checks owned by M11.1b."""
 
    operation_type: str
    message: str
 
    def __post_init__(self) -> None:
        super().__init__(f"{self.operation_type!r}: {self.message}")
 
 
@dataclass
class ParentCycleError(RegistryError):
    """Raised when parent_operation_id would create or expose a cycle."""
 
    parent_operation_id: int
    operation_uuid: str
 
    def __post_init__(self) -> None:
        super().__init__(
            "parent_operation_id cycle detected for "
            f"{self.operation_uuid!r} via {self.parent_operation_id!r}"
        )
 
 
@dataclass
class UnknownActorError(RegistryError):
    """Raised when actor_slot is not present in agents."""
 
    actor_slot: str
 
    def __post_init__(self) -> None:
        super().__init__(f"unknown actor_slot: {self.actor_slot!r}")
 
 
@dataclass
class RoundStateError(RegistryError):
    """Raised when a round.* operation is illegal for the round's status."""
 
    round_no: int
    status: str
    operation_type: str
 
    def __post_init__(self) -> None:
        super().__init__(
            f"round {self.round_no} in status {self.status!r} rejects "
            f"{self.operation_type!r}"
        )
 
 
@dataclass
class RoundInvariantError(RegistryError):
    """Raised when a round-close invariant is violated (torn state, gaps)."""
 
    round_no: int
    message: str
 
    def __post_init__(self) -> None:
        super().__init__(f"round {self.round_no} invariant violated: {self.message}")
 
 
@dataclass
class PreflightFailedError(RegistryError):
    """Raised when a pre-open or pre-resume preflight check fails."""
 
    round_no: int | None
    reasons: tuple[str, ...]
 
    def __post_init__(self) -> None:
        joined = "; ".join(self.reasons) or "unspecified"
        scope = "round-close" if self.round_no is None else f"round {self.round_no}"
        super().__init__(f"{scope} preflight failed: {joined}")
 
 
@dataclass
class MergeConflictError(RegistryError):
    """Raised when a plan-branch merge into staging hits a conflict."""
 
    round_no: int
    plan_id: str
    conflicted_paths: tuple[str, ...]
 
    def __post_init__(self) -> None:
        paths = ", ".join(self.conflicted_paths) or "unknown"
        super().__init__(
            f"merge of {self.plan_id!r} into round {self.round_no} staging "
            f"conflicted on: {paths}"
        )
 
 
@dataclass
class ManualMergeRequiredError(RegistryError):
    """Raised when a conflict is outside the auto-resolve allowlist."""
 
    round_no: int
    plan_id: str
    conflicted_paths: tuple[str, ...]
 
    def __post_init__(self) -> None:
        paths = ", ".join(self.conflicted_paths) or "unknown"
        super().__init__(
            f"manual merge required for {self.plan_id!r} in round {self.round_no}; "
            f"non-allowlisted conflicts: {paths}"
        )
 
 
@dataclass
class StagingBranchDivergedError(RegistryError):
    """Raised when the staging branch HEAD no longer matches recorded state."""
 
    round_no: int
    expected_sha: str | None
    actual_sha: str | None
 
    def __post_init__(self) -> None:
        super().__init__(
            f"round {self.round_no} staging branch diverged: expected "
            f"{self.expected_sha!r}, found {self.actual_sha!r}; rollback required"
        )