escalation.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/escalation.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_close_escalation.py"
"""
Versioned escalation contract for round-close failures.
 
Purpose
-------
Render a stable, machine-readable escalation document (OI-218, OI-316)
whenever a round-close step stops and needs operator attention.
 
Responsibilities
----------------
- Enumerate the stable error-code set for round-close stops.
- Carry a ``resume_command`` except where recovery is rollback-only.
- Serialize deterministically for logs and operator prompts.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - escalation raised
    L3 - code and resume-command details
    L4 - deep tracing
 
Contracts
---------
- ``STAGING_BRANCH_DIVERGED`` always renders an empty ``resume_command``.
- The schema is versioned; consumers branch on ``schema_version``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- dataclasses (the frozen Escalation record)
- enum (the EscalationCode enum)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from enum import Enum
from typing import Mapping
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
ESCALATION_SCHEMA_VERSION = "1.0"
 
 
class EscalationCode(str, Enum):
    """Stable round-close escalation codes."""
 
    STAGING_BRANCH_DIVERGED = "STAGING_BRANCH_DIVERGED"
    MERGE_CONFLICT = "MERGE_CONFLICT"
    MANUAL_MERGE_REQUIRED = "MANUAL_MERGE_REQUIRED"
    TEST_FAILED = "TEST_FAILED"
    LINT_FAILED = "LINT_FAILED"
    PREFLIGHT_FAILED = "PREFLIGHT_FAILED"
    LOCK_HELD = "LOCK_HELD"
    ROUND_STATE = "ROUND_STATE"
    CLEANUP_FAILED = "CLEANUP_FAILED"
    ROADMAP_BUMP_FAILED = "ROADMAP_BUMP_FAILED"
    ROADMAP_UNTRACKED_EDIT = "E_ROADMAP_UNTRACKED_EDIT"
 
 
# The one non-resumable class: the operator committed to the staging branch
# outside the job, so re-running merges on top would silently incorporate it.
NON_RESUMABLE_CODES = frozenset({EscalationCode.STAGING_BRANCH_DIVERGED})
 
 
@dataclass(frozen=True)
class Escalation:
    """A versioned, serializable round-close escalation document."""
 
    code: EscalationCode
    round_no: int
    phase: str
    message: str
    resume_command: str | None = None
    details: Mapping[str, object] = field(default_factory=dict)
    schema_version: str = ESCALATION_SCHEMA_VERSION
 
    def to_dict(self) -> dict[str, object]:
        """Return a deterministic, JSON-serializable mapping."""
 
        func = "to_dict"
        resume = "" if self.code in NON_RESUMABLE_CODES else (self.resume_command or "")
        _dbg("OC_CHAT.MAIN", func, f"render {self.code.value}", 4)
        return {
            "schema_version": self.schema_version,
            "kind": "round_close_escalation",
            "code": self.code.value,
            "round_no": self.round_no,
            "phase": self.phase,
            "message": self.message,
            "resume_command": resume,
            "details": dict(self.details),
        }
 
 
class RoundEscalation(RuntimeError):
    """Carries an :class:`Escalation` payload up to the orchestrator surface."""
 
    def __init__(self, escalation: Escalation) -> None:
        self.escalation = escalation
        super().__init__(f"{escalation.code.value}: {escalation.message}")
 
 
def make_escalation(
    code: EscalationCode,
    *,
    round_no: int,
    phase: str,
    message: str,
    resume_command: str | None = None,
    details: Mapping[str, object] | None = None,
) -> Escalation:
    """Build an :class:`Escalation`, enforcing the non-resumable policy."""
 
    func = "make_escalation"
    effective_resume = None if code in NON_RESUMABLE_CODES else resume_command
    _dbg("OC_CHAT.MAIN", func, f"build {code.value} round={round_no}", 2)
    return Escalation(
        code=code,
        round_no=round_no,
        phase=phase,
        message=message,
        resume_command=effective_resume,
        details=dict(details or {}),
    )
 
 
def resume_command_for(round_no: int, *, slot: str | None = None) -> str:
    """Return the canonical resume command string for a round."""
 
    func = "resume_command_for"
    _dbg("OC_CHAT.MAIN", func, f"resume cmd round={round_no}", 4)
    base = f"oct agent round resume {round_no}"
    return base if slot is None else f"{base} --as {slot}"