preflight.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/preflight.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_open_executor.py"
"""
Preflight checks for round open and resume.
 
Purpose
-------
Validate that a round-close can safely begin (or resume) before any
git or operation-log side effect happens.
 
Responsibilities
----------------
- Confirm the plan set is non-empty and every plan is merge-ready.
- Confirm there is no orphan staging branch for a fresh open.
- Confirm a resume target job and round both exist.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - preflight pass/fail
    L3 - per-check details
    L4 - deep tracing
 
Contracts
---------
- Preflight performs reads only; it never mutates the registry or git.
- A failing preflight returns reasons; the caller raises PreflightFailedError.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.round.git_shim (GitShim seam)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from typing import Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.round.git_shim import GitShim
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
MERGE_READY_STATUSES = frozenset({"ready_for_merge", "integrating"})
 
 
@dataclass(frozen=True)
class PreflightResult:
    """Outcome of a preflight check sweep."""
 
    ok: bool
    reasons: tuple[str, ...] = field(default_factory=tuple)
 
 
def preflight_open(
    registry: Registry,
    *,
    round_no: int,
    plan_ids: Sequence[str],
    git: GitShim,
    staging_branch: str,
) -> PreflightResult:
    """Validate a fresh round-open request."""
 
    func = "preflight_open"
    reasons: list[str] = []
    plan_ids = list(plan_ids)
    if not plan_ids:
        reasons.append("plan set is empty")
    statuses = _plan_statuses(registry, plan_ids)
    for plan_id in plan_ids:
        status = statuses.get(plan_id)
        if status is None:
            reasons.append(f"plan {plan_id!r} not found in registry")
        elif status not in MERGE_READY_STATUSES:
            reasons.append(
                f"plan {plan_id!r} status {status!r} is not merge-ready"
            )
    existing = registry.backend.get_round(round_no)
    if existing is None and git.branch_exists(staging_branch):
        reasons.append(
            f"orphan staging branch {staging_branch!r} with no round row"
        )
    ok = not reasons
    _dbg("OC_CHAT.MAIN", func, f"round={round_no} ok={ok}", 2)
    return PreflightResult(ok=ok, reasons=tuple(reasons))
 
 
def preflight_resume(
    registry: Registry,
    *,
    round_no: int,
    job_id: str | None,
) -> PreflightResult:
    """Validate that a resume target job and round exist."""
 
    func = "preflight_resume"
    reasons: list[str] = []
    if registry.backend.get_round(round_no) is None:
        reasons.append(f"round {round_no} not found")
    if job_id is None:
        reasons.append(f"no resumable round-job for round {round_no}")
    elif registry.backend.get_round_job(job_id) is None:
        reasons.append(f"round-job {job_id!r} not found")
    ok = not reasons
    _dbg("OC_CHAT.MAIN", func, f"round={round_no} ok={ok}", 2)
    return PreflightResult(ok=ok, reasons=tuple(reasons))
 
 
def _plan_statuses(
    registry: Registry,
    plan_ids: Sequence[str],
) -> dict[str, str]:
    func = "_plan_statuses"
    if not plan_ids:
        return {}
    placeholders = ", ".join("?" for _ in plan_ids)
    connect_func = getattr(registry.backend, "connect")
    with connect_func() as conn:
        rows = conn.execute(
            f"SELECT plan_id, status FROM plans WHERE plan_id IN ({placeholders})",
            tuple(plan_ids),
        ).fetchall()
    result = {str(row["plan_id"]): str(row["status"]) for row in rows}
    _dbg("OC_CHAT.MAIN", func, f"{len(result)} plan statuses", 4)
    return result