resume.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/resume.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_resume_protocol.py"
"""
Resume protocol for crashed or paused round-close jobs.
 
Purpose
-------
Given a round number, find the most recent resumable round-job, verify
the on-disk staging branch matches recorded state, and dispatch the next
pending step through the same executors used on the first run (OI-315).
 
Responsibilities
----------------
- Resolve the resumable job and compute the highest completed step.
- Verify the staging branch HEAD against known-good recorded heads.
- Emit STAGING_BRANCH_DIVERGED (empty resume_command) on real divergence.
- Re-enter the state-machine driver at the next pending step.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - resume decisions
    L3 - head comparison details
    L4 - deep tracing
 
Contracts
---------
- Resume is itself idempotent; re-running after success is a no-op.
- A diverged staging branch is never auto-merged; rollback is the only path.
- Crash-after-merge (git done, DB pending) re-attempts the idempotent step.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.round.state_machine (build_step_plan and drive_steps - re-entering the driver at the next pending step)
- oc_chat.registry.round.executors (RoundContext - the shared executor inputs)
- oc_chat.registry.round.escalation (make_escalation - the STAGING_BRANCH_DIVERGED escalation)
- oc_chat.registry.round.preflight (preflight_resume - the resume preflight check)
- oc_chat.registry.errors (PreflightFailedError - raised when the resume preflight fails)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.errors import PreflightFailedError
from oc_chat.registry.round.escalation import (
    Escalation,
    EscalationCode,
    make_escalation,
)
from oc_chat.registry.round.executors import RoundContext
from oc_chat.registry.round.preflight import preflight_resume
from oc_chat.registry.round.state_machine import (
    RoundCloseResult,
    Prompter,
    build_step_plan,
    drive_steps,
)
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
RESUMABLE_STATUSES = ("running", "paused", "failed")
 
 
def resume_round(
    ctx: RoundContext,
    *,
    interactive: bool = False,
    prompter: Prompter | None = None,
) -> RoundCloseResult:
    """Resume the most recent resumable round-job for ``ctx.round_no``."""
 
    func = "resume_round"
    backend = ctx.registry.backend
    job = backend.latest_round_job(ctx.round_no, statuses=RESUMABLE_STATUSES)
    job_id = None if job is None else str(job["job_id"])
    preflight = preflight_resume(ctx.registry, round_no=ctx.round_no, job_id=job_id)
    if not preflight.ok:
        raise PreflightFailedError(round_no=ctx.round_no, reasons=preflight.reasons)
    assert job is not None
    ctx.job_id = str(job["job_id"])
    steps = build_step_plan(ctx)
    step_rows = backend.list_round_job_steps(ctx.job_id)
    done = [row for row in step_rows if str(row["status"]) == "done"]
    highest_done = max((int(row["step_no"]) for row in done), default=0)
    start_step = highest_done + 1
    if start_step > len(steps):
        backend.set_round_job_status(
            ctx.job_id, status="done", current_step="cleanup"
        )
        _dbg("OC_CHAT.MAIN", func, "nothing left to resume", 2)
        return RoundCloseResult(ctx.job_id, "done", [s.step_name for s in steps])
 
    divergence = _check_divergence(
        ctx,
        job=job,
        done_rows=done,
        next_phase=steps[start_step - 1].phase,
    )
    if divergence is not None:
        backend.set_round_job_status(
            ctx.job_id, status="failed", current_step=steps[start_step - 1].step_name
        )
        _dbg("OC_CHAT.MAIN", func, "staging branch diverged", 2)
        return RoundCloseResult(ctx.job_id, "failed", [], escalation=divergence)
 
    _dbg("OC_CHAT.MAIN", func, f"resume from step {start_step}", 2)
    return drive_steps(
        ctx,
        start_step=start_step,
        interactive=interactive,
        prompter=prompter,
    )
 
 
def _check_divergence(
    ctx: RoundContext,
    *,
    job: dict[str, object],
    done_rows: list[dict[str, object]],
    next_phase: str,
) -> Escalation | None:
    func = "_check_divergence"
    if not ctx.git.branch_exists(ctx.staging_branch):
        return _diverged(ctx, expected=_text(job.get("staging_branch_head_sha")), actual=None)
    actual_head = ctx.git.rev_parse(ctx.staging_branch)
    expected_head = _text(job.get("staging_branch_head_sha"))
    known_heads: set[str] = set()
    if expected_head:
        known_heads.add(expected_head)
    for row in done_rows:
        head = _text(row.get("post_step_head_sha"))
        if head:
            known_heads.add(head)
    if actual_head in known_heads:
        _dbg("OC_CHAT.MAIN", func, "head matches known state", 3)
        return None
    # Crash-after-merge: git advanced HEAD before the DB row was written.
    if (
        next_phase == "merge"
        and expected_head
        and ctx.git.is_ancestor(expected_head, actual_head)
    ):
        _dbg("OC_CHAT.MAIN", func, "crash-after-merge; re-attempt", 2)
        return None
    return _diverged(ctx, expected=expected_head, actual=actual_head)
 
 
def _diverged(
    ctx: RoundContext,
    *,
    expected: str | None,
    actual: str | None,
) -> Escalation:
    return make_escalation(
        EscalationCode.STAGING_BRANCH_DIVERGED,
        round_no=ctx.round_no,
        phase="resume",
        message=(
            f"staging branch {ctx.staging_branch!r} HEAD {actual!r} does not match "
            f"recorded state {expected!r}; rollback required"
        ),
        details={"expected_sha": expected, "actual_sha": actual},
    )
 
 
def _text(value: object) -> str | None:
    if value is None:
        return None
    text = str(value)
    return text or None