state_machine.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/state_machine.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_close_e2e.py"
"""
Seven-phase round-close state machine and step driver.
 
Purpose
-------
Sequence the round-close phases (open -> merge xN -> test -> lint ->
consolidate -> ff_main -> cleanup), record durable per-step state in
round_jobs / round_job_steps, and make every step resumable.
 
Responsibilities
----------------
- Build the ordered step plan for a round from its plan set.
- Drive steps idempotently; skip steps already recorded ``done``.
- Persist step status, post-step HEAD, and input hashes for resume.
- Surface RoundEscalation as a paused/failed job, not a crash.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - phase transitions
    L3 - step bookkeeping
    L4 - deep tracing
 
Contracts
---------
- The driver writes round_job_steps; executors never do.
- A step recorded ``done`` is never re-executed on resume.
- Interactive abort runs round.abort and stops cleanly.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.round.executors (the per-phase run_* executors and the PlanSpec, RoundContext and StepOutcome types)
- oc_chat.registry.round.escalation (RoundEscalation caught and recorded as a failed step and job, and the Escalation record on the result)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
import json
import uuid
from typing import Callable
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.round import executors as ex
from oc_chat.registry.round.escalation import Escalation, RoundEscalation
from oc_chat.registry.round.executors import PlanSpec, RoundContext, StepOutcome
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
PHASE_ORDER = ("open", "merge", "test", "lint", "consolidate", "ff_main", "cleanup")
TERMINAL_PHASE = "cleanup"
 
 
@dataclass(frozen=True)
class StepDef:
    """A single position in the round-close step plan."""
 
    step_no: int
    step_name: str
    phase: str
    plan_spec: PlanSpec | None = None
 
 
@dataclass
class RoundCloseResult:
    """The outcome of driving a round-close job to a stopping point."""
 
    job_id: str
    status: str
    completed_steps: list[str] = field(default_factory=list)
    escalation: Escalation | None = None
 
 
Prompter = Callable[[StepDef, RoundContext], str]
 
 
def build_step_plan(ctx: RoundContext) -> list[StepDef]:
    """Build the ordered step plan for ``ctx``'s round."""
 
    func = "build_step_plan"
    steps: list[StepDef] = []
    step_no = 1
    steps.append(StepDef(step_no, "open", "open"))
    step_no += 1
    for spec in sorted(ctx.plan_specs, key=lambda s: s.merge_order):
        steps.append(StepDef(step_no, f"merge:{spec.plan_id}", "merge", spec))
        step_no += 1
    for phase in ("test", "lint", "consolidate", "ff_main", "cleanup"):
        steps.append(StepDef(step_no, phase, phase))
        step_no += 1
    _dbg("OC_CHAT.MAIN", func, f"{len(steps)} steps", 4)
    return steps
 
 
def create_close_job(ctx: RoundContext, *, create_staging: bool = True) -> str:
    """Create the round-job row and staging branch; return the job id."""
 
    func = "create_close_job"
    # The round_jobs row carries an FK to rounds(round_no); ensure the round
    # row exists first. round.open is idempotent (ON CONFLICT DO NOTHING).
    ctx.registry.round_open(
        round_no=ctx.round_no,
        orchestrator_slot=ctx.actor_slot,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    if create_staging and not ctx.git.branch_exists(ctx.staging_branch):
        ctx.git.create_branch(ctx.staging_branch, ctx.base_ref)
    head_sha = (
        ctx.git.rev_parse(ctx.staging_branch)
        if ctx.git.branch_exists(ctx.staging_branch)
        else None
    )
    ctx.registry.backend.create_round_job(
        job_id=ctx.job_id,
        round_no=ctx.round_no,
        created_by=ctx.actor_slot,
        staging_branch_head_sha=head_sha,
    )
    _dbg("OC_CHAT.MAIN", func, f"job {ctx.job_id} head={head_sha}", 2)
    return ctx.job_id
 
 
def new_job_id() -> str:
    """Return a fresh round-close job id (UUIDv4)."""
 
    return str(uuid.uuid4())
 
 
def run_round_close(
    ctx: RoundContext,
    *,
    interactive: bool = False,
    prompter: Prompter | None = None,
    create_staging: bool = True,
) -> RoundCloseResult:
    """Drive a fresh round-close from the first step to a stopping point."""
 
    func = "run_round_close"
    create_close_job(ctx, create_staging=create_staging)
    _dbg("OC_CHAT.MAIN", func, f"begin round {ctx.round_no}", 2)
    return drive_steps(
        ctx,
        start_step=1,
        interactive=interactive,
        prompter=prompter,
    )
 
 
def drive_steps(
    ctx: RoundContext,
    *,
    start_step: int,
    interactive: bool = False,
    prompter: Prompter | None = None,
) -> RoundCloseResult:
    """Execute the step plan from ``start_step`` onward, persisting state."""
 
    func = "drive_steps"
    backend = ctx.registry.backend
    steps = build_step_plan(ctx)
    existing = {int(row["step_no"]): row for row in backend.list_round_job_steps(ctx.job_id)}
    completed: list[str] = []
    for step in steps:
        prior = existing.get(step.step_no)
        if step.step_no < start_step or (prior and str(prior["status"]) == "done"):
            if prior and str(prior["status"]) == "done":
                completed.append(step.step_name)
            continue
        if interactive and prompter is not None:
            decision = prompter(step, ctx)
            if decision == "abort":
                ex.run_abort(ctx, reason="operator abort at phase prompt")
                backend.set_round_job_status(
                    ctx.job_id, status="rolled_back", current_step=step.step_name
                )
                _dbg("OC_CHAT.MAIN", func, "operator aborted", 2)
                return RoundCloseResult(ctx.job_id, "aborted", completed)
        result = _run_one_step(ctx, step, prior)
        if result is not None:
            return result
        completed.append(step.step_name)
    backend.set_round_job_status(
        ctx.job_id, status="done", current_step=TERMINAL_PHASE
    )
    _dbg("OC_CHAT.MAIN", func, f"round {ctx.round_no} closed", 2)
    return RoundCloseResult(ctx.job_id, "done", completed)
 
 
def _run_one_step(
    ctx: RoundContext,
    step: StepDef,
    prior: dict[str, object] | None,
) -> RoundCloseResult | None:
    func = "_run_one_step"
    backend = ctx.registry.backend
    now = ctx.registry.clock.now_wall()
    backend.upsert_round_job_step(
        job_id=ctx.job_id,
        step_no=step.step_no,
        step_name=step.step_name,
        status="running",
        started_at=now,
    )
    backend.set_round_job_status(
        ctx.job_id, status="running", current_step=step.step_name
    )
    prior_hash = None if prior is None else _optional_text(prior.get("input_hash"))
    try:
        outcome = _execute_step(ctx, step, prior_hash)
    except RoundEscalation as esc:
        backend.upsert_round_job_step(
            job_id=ctx.job_id,
            step_no=step.step_no,
            step_name=step.step_name,
            status="failed",
            error_summary=esc.escalation.message,
            finished_at=ctx.registry.clock.now_wall(),
        )
        backend.set_round_job_status(
            ctx.job_id, status="failed", current_step=step.step_name
        )
        _dbg("OC_CHAT.MAIN", func, f"escalation {esc.escalation.code.value}", 2)
        return RoundCloseResult(
            ctx.job_id,
            "failed",
            [],
            escalation=esc.escalation,
        )
    artifacts_json = (
        json.dumps(list(outcome.artifact_paths)) if outcome.artifact_paths else None
    )
    backend.upsert_round_job_step(
        job_id=ctx.job_id,
        step_no=step.step_no,
        step_name=step.step_name,
        status="done",
        input_hash=outcome.input_hash,
        post_step_head_sha=outcome.post_step_head_sha,
        result_summary=outcome.result_summary,
        artifact_paths_json=artifacts_json,
        finished_at=ctx.registry.clock.now_wall(),
    )
    if outcome.post_step_head_sha:
        backend.set_round_job_head_sha(ctx.job_id, outcome.post_step_head_sha)
    _dbg("OC_CHAT.MAIN", func, f"step {step.step_name} {outcome.status}", 3)
    return None
 
 
def _execute_step(
    ctx: RoundContext,
    step: StepDef,
    prior_input_hash: str | None,
) -> StepOutcome:
    func = "_execute_step"
    _dbg("OC_CHAT.MAIN", func, f"exec {step.phase}", 4)
    if step.phase == "open":
        return ex.run_open(ctx)
    if step.phase == "merge":
        assert step.plan_spec is not None
        return ex.run_merge(ctx, step.plan_spec)
    if step.phase == "test":
        return ex.run_test(ctx, prior_input_hash)
    if step.phase == "lint":
        return ex.run_lint(ctx, prior_input_hash)
    if step.phase == "consolidate":
        return ex.run_consolidate(ctx)
    if step.phase == "ff_main":
        return ex.run_ff_main(ctx)
    if step.phase == "cleanup":
        return ex.run_cleanup(ctx)
    raise ValueError(f"unknown phase: {step.phase}")
 
 
def _optional_text(value: object) -> str | None:
    if value is None:
        return None
    return str(value)