executors.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/round/executors.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_*_executor.py"
"""
Per-phase executors for the eight round.* operation types.
 
Purpose
-------
Implement the idempotent-first executors that drive each round-close
phase: open, merge, test, lint, consolidate, ff_main, cleanup, abort.
 
Responsibilities
----------------
- Perform the git side effect for a phase via the injectable GitShim.
- Check "already done" before doing work so every step is resumable.
- Write exactly one operation-log row per phase per plan via the seam.
- Surface operator-actionable stops as RoundEscalation.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - phase lifecycle
    L3 - idempotency and git details
    L4 - deep tracing
 
Contracts
---------
- Executors never write round_job_steps; the state machine owns that.
- Executors are pure with respect to the registry except for the audit row.
- A no-op (already-done) executor returns ``status='skipped'``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.round.git_shim (GitShim - the injectable git seam for each phase's side effect)
- oc_chat.registry.round.conflict (ConflictPolicy and ConflictStrategy - auto-resolve or abort on a merge conflict)
- oc_chat.registry.round.escalation (RoundEscalation, make_escalation and resume_command_for - operator-actionable stops)
- oc_chat.registry.canonical (canonical_json - the test and lint input hashes)
- oc_chat.registry.errors (ManualMergeRequiredError and RoundInvariantError)
- oc_chat.registry.roadmap.audit (verify_roadmap_files_unchanged - the roadmap hand-edit check at cleanup)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
import hashlib
from typing import Callable, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.canonical import canonical_json
from oc_chat.registry.errors import (
    ManualMergeRequiredError,
    RoundInvariantError,
)
from oc_chat.registry.round.conflict import ConflictPolicy, ConflictStrategy
from oc_chat.registry.round.escalation import (
    EscalationCode,
    RoundEscalation,
    make_escalation,
    resume_command_for,
)
from oc_chat.registry.round.git_shim import GitShim
from oc_chat.registry.roadmap.audit import verify_roadmap_files_unchanged
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class PlanSpec:
    """A plan to merge in a round, with its branch and worktree."""
 
    plan_id: str
    branch: str
    worktree: str
    merge_order: int
 
 
@dataclass(frozen=True)
class TestRunResult:
    """Outcome of a full-suite test run."""
 
    passed: bool
    count_before: int | None = None
    count_after: int | None = None
    duration_seconds: float | None = None
    artifact_paths: tuple[str, ...] = field(default_factory=tuple)
    plan_results: tuple[dict[str, object], ...] = field(default_factory=tuple)
 
 
@dataclass(frozen=True)
class LintRunResult:
    """Outcome of an ``oct lint --profile strict`` run."""
 
    passed: bool
    errors: int = 0
    warnings: int = 0
    artifact_paths: tuple[str, ...] = field(default_factory=tuple)
 
 
@dataclass(frozen=True)
class StepOutcome:
    """Result of executing one round-close phase."""
 
    step_name: str
    status: str
    post_step_head_sha: str | None = None
    input_hash: str | None = None
    result_summary: str | None = None
    error_summary: str | None = None
    artifact_paths: tuple[str, ...] = field(default_factory=tuple)
 
 
TestRunner = Callable[["RoundContext"], TestRunResult]
LintRunner = Callable[["RoundContext"], LintRunResult]
RoadmapBumper = Callable[["RoundContext"], str | None]
 
 
@dataclass
class RoundContext:
    """Shared inputs for the round-close executors."""
 
    registry: Registry
    git: GitShim
    round_no: int
    job_id: str
    actor_slot: str
    staging_branch: str
    plan_specs: list[PlanSpec]
    actor_type: str = "orchestrator_agent"
    main_branch: str = "main"
    base_ref: str = "main"
    conflict_policy: ConflictPolicy = field(default_factory=ConflictPolicy)
    test_runner: TestRunner | None = None
    lint_runner: LintRunner | None = None
    roadmap_bumper: RoadmapBumper | None = None
    roadmap_dir: str | None = None
    allow_untracked_roadmap_edits: bool = False
    app_version: str | None = None
    telemetry: dict[str, object] | None = None
 
 
def _staging_head(ctx: RoundContext) -> str:
    return ctx.git.rev_parse(ctx.staging_branch)
 
 
def _hash_inputs(*parts: object) -> str:
    return hashlib.sha256(canonical_json(list(parts)).encode("utf-8")).hexdigest()
 
 
def run_open(ctx: RoundContext) -> StepOutcome:
    """Open the round: one ``round.open`` audit row per plan (SF-001)."""
 
    func = "run_open"
    round_obj = ctx.registry.backend.get_round(ctx.round_no)
    if round_obj is not None and str(round_obj.get("status")) in {"closed", "aborted"}:
        raise RoundEscalation(
            make_escalation(
                EscalationCode.ROUND_STATE,
                round_no=ctx.round_no,
                phase="open",
                message=f"round {ctx.round_no} is {round_obj.get('status')}",
            )
        )
    for spec in sorted(ctx.plan_specs, key=lambda s: s.merge_order):
        existing = ctx.registry.backend.get_round_plan(ctx.round_no, spec.plan_id)
        if existing is not None:
            _dbg("OC_CHAT.MAIN", func, f"plan {spec.plan_id} already open", 3)
            continue
        ctx.registry.round_open(
            round_no=ctx.round_no,
            orchestrator_slot=ctx.actor_slot,
            plan_id=spec.plan_id,
            merge_order=spec.merge_order,
            actor_slot=ctx.actor_slot,
            actor_type=ctx.actor_type,
        )
    if not ctx.plan_specs and round_obj is None:
        ctx.registry.round_open(
            round_no=ctx.round_no,
            orchestrator_slot=ctx.actor_slot,
            actor_slot=ctx.actor_slot,
            actor_type=ctx.actor_type,
        )
    _dbg("OC_CHAT.MAIN", func, f"round {ctx.round_no} open", 2)
    return StepOutcome(
        step_name="open",
        status="done",
        result_summary=f"opened round {ctx.round_no} with {len(ctx.plan_specs)} plans",
    )
 
 
def run_merge(ctx: RoundContext, spec: PlanSpec) -> StepOutcome:
    """Merge one plan branch into staging (idempotent on merge_commit_sha)."""
 
    func = "run_merge"
    plan_row = ctx.registry.backend.get_round_plan(ctx.round_no, spec.plan_id)
    if plan_row is not None and plan_row.get("merge_commit_sha"):
        _dbg("OC_CHAT.MAIN", func, f"plan {spec.plan_id} merged already", 2)
        return StepOutcome(
            step_name=f"merge:{spec.plan_id}",
            status="skipped",
            post_step_head_sha=_staging_head(ctx),
            result_summary=f"{spec.plan_id} already merged",
        )
    ctx.git.checkout(ctx.staging_branch)
    base_sha = ctx.git.rev_parse(ctx.staging_branch)
    result = ctx.git.merge(
        spec.branch,
        no_ff=True,
        message=f"[round-{ctx.round_no}] merge {spec.plan_id}",
    )
    if not result.ok:
        self_resolved = _try_resolve_conflict(ctx, spec, result.conflicted_paths)
        if not self_resolved:
            ctx.git.merge_abort()
            raise RoundEscalation(
                make_escalation(
                    EscalationCode.MERGE_CONFLICT,
                    round_no=ctx.round_no,
                    phase="merge",
                    message=(
                        f"merge of {spec.plan_id} conflicted on "
                        f"{', '.join(result.conflicted_paths) or 'unknown'}"
                    ),
                    resume_command=resume_command_for(ctx.round_no),
                    details={
                        "plan_id": spec.plan_id,
                        "conflicted_paths": list(result.conflicted_paths),
                    },
                )
            )
        head_sha = _staging_head(ctx)
    else:
        head_sha = result.head_sha
    ctx.registry.round_merge(
        round_no=ctx.round_no,
        plan_id=spec.plan_id,
        merge_order=spec.merge_order,
        merge_commit_sha=head_sha,
        base_sha=base_sha,
        staging_branch=ctx.staging_branch,
        merge_base_sha=base_sha,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _dbg("OC_CHAT.MAIN", func, f"merged {spec.plan_id} -> {head_sha[:8]}", 2)
    return StepOutcome(
        step_name=f"merge:{spec.plan_id}",
        status="done",
        post_step_head_sha=head_sha,
        result_summary=f"merged {spec.plan_id}",
    )
 
 
def _try_resolve_conflict(
    ctx: RoundContext,
    spec: PlanSpec,
    conflicted_paths: Sequence[str],
) -> bool:
    func = "_try_resolve_conflict"
    decision = ctx.conflict_policy.classify(conflicted_paths)
    if decision.strategy is ConflictStrategy.ABORT:
        if decision.blocking_paths:
            _dbg("OC_CHAT.MAIN", func, "blocking conflict", 2)
        return False
    if not decision.resolvable_paths:
        return False
    for path in decision.resolvable_paths:
        ctx.git.checkout_path("current", path)
    try:
        ctx.git.run("commit", "--no-edit")
    except Exception as exc:  # pragma: no cover - defensive
        raise ManualMergeRequiredError(
            round_no=ctx.round_no,
            plan_id=spec.plan_id,
            conflicted_paths=tuple(conflicted_paths),
        ) from exc
    _dbg("OC_CHAT.MAIN", func, f"auto-resolved {len(decision.resolvable_paths)}", 2)
    return True
 
 
def run_test(ctx: RoundContext, prior_input_hash: str | None = None) -> StepOutcome:
    """Run the full suite; idempotent on the staging+args input hash."""
 
    func = "run_test"
    staging_sha = _staging_head(ctx)
    plan_ids = sorted(spec.plan_id for spec in ctx.plan_specs)
    input_hash = _hash_inputs("test", staging_sha, plan_ids)
    if prior_input_hash is not None and prior_input_hash == input_hash:
        _dbg("OC_CHAT.MAIN", func, "test input unchanged; skip", 2)
        return StepOutcome(
            step_name="test",
            status="skipped",
            post_step_head_sha=staging_sha,
            input_hash=input_hash,
            result_summary="tests already green for this staging head",
        )
    if ctx.test_runner is None:
        raise RoundInvariantError(
            round_no=ctx.round_no,
            message="no test_runner configured for round.test",
        )
    result = ctx.test_runner(ctx)
    if not result.passed:
        raise RoundEscalation(
            make_escalation(
                EscalationCode.TEST_FAILED,
                round_no=ctx.round_no,
                phase="test",
                message="full test suite failed on staging",
                resume_command=resume_command_for(ctx.round_no),
                details={"artifact_paths": list(result.artifact_paths)},
            )
        )
    ctx.registry.round_test(
        round_no=ctx.round_no,
        test_count_before=result.count_before,
        test_count_after=result.count_after,
        input_hash=input_hash,
        result="pass",
        duration_seconds=result.duration_seconds,
        plan_results=list(result.plan_results),
        artifact_paths=list(result.artifact_paths),
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _dbg("OC_CHAT.MAIN", func, "tests passed", 2)
    return StepOutcome(
        step_name="test",
        status="done",
        post_step_head_sha=staging_sha,
        input_hash=input_hash,
        result_summary=f"tests passed ({result.count_after})",
        artifact_paths=tuple(result.artifact_paths),
    )
 
 
def run_lint(ctx: RoundContext, prior_input_hash: str | None = None) -> StepOutcome:
    """Run strict lint over staging; idempotent on the staging+args hash."""
 
    func = "run_lint"
    staging_sha = _staging_head(ctx)
    input_hash = _hash_inputs("lint", staging_sha)
    if prior_input_hash is not None and prior_input_hash == input_hash:
        _dbg("OC_CHAT.MAIN", func, "lint input unchanged; skip", 2)
        return StepOutcome(
            step_name="lint",
            status="skipped",
            post_step_head_sha=staging_sha,
            input_hash=input_hash,
            result_summary="lint already clean for this staging head",
        )
    if ctx.lint_runner is None:
        raise RoundInvariantError(
            round_no=ctx.round_no,
            message="no lint_runner configured for round.lint",
        )
    result = ctx.lint_runner(ctx)
    if not result.passed:
        raise RoundEscalation(
            make_escalation(
                EscalationCode.LINT_FAILED,
                round_no=ctx.round_no,
                phase="lint",
                message=f"strict lint failed: {result.errors} errors",
                resume_command=resume_command_for(ctx.round_no),
                details={
                    "errors": result.errors,
                    "warnings": result.warnings,
                    "artifact_paths": list(result.artifact_paths),
                },
            )
        )
    ctx.registry.round_lint(
        round_no=ctx.round_no,
        staging_sha=staging_sha,
        errors=result.errors,
        warnings=result.warnings,
        input_hash=input_hash,
        result="pass",
        artifact_paths=list(result.artifact_paths),
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _dbg("OC_CHAT.MAIN", func, "lint passed", 2)
    return StepOutcome(
        step_name="lint",
        status="done",
        post_step_head_sha=staging_sha,
        input_hash=input_hash,
        result_summary=f"lint clean ({result.warnings} warnings)",
        artifact_paths=tuple(result.artifact_paths),
    )
 
 
def run_consolidate(ctx: RoundContext) -> StepOutcome:
    """Record the staging HEAD as final_commit_sha; flip to committed."""
 
    func = "run_consolidate"
    round_obj = ctx.registry.backend.get_round(ctx.round_no)
    final_sha = _staging_head(ctx)
    if round_obj is not None and str(round_obj.get("status")) == "committed":
        _dbg("OC_CHAT.MAIN", func, "already committed; skip", 2)
        return StepOutcome(
            step_name="consolidate",
            status="skipped",
            post_step_head_sha=final_sha,
            result_summary="round already committed",
        )
    ctx.registry.round_consolidate(
        round_no=ctx.round_no,
        final_commit_sha=final_sha,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _dbg("OC_CHAT.MAIN", func, f"committed {final_sha[:8]}", 2)
    return StepOutcome(
        step_name="consolidate",
        status="done",
        post_step_head_sha=final_sha,
        result_summary=f"consolidated at {final_sha[:8]}",
    )
 
 
def run_ff_main(ctx: RoundContext) -> StepOutcome:
    """Fast-forward main to the round's final commit (rejects non-FF)."""
 
    func = "run_ff_main"
    round_obj = ctx.registry.backend.get_round(ctx.round_no)
    if round_obj is None or not round_obj.get("final_commit_sha"):
        raise RoundInvariantError(
            round_no=ctx.round_no,
            message="ff_main requires a recorded final_commit_sha",
        )
    target = str(round_obj["final_commit_sha"])
    current_main = ctx.git.rev_parse(ctx.main_branch)
    if current_main == target:
        _dbg("OC_CHAT.MAIN", func, "main already at target; skip", 2)
        return StepOutcome(
            step_name="ff_main",
            status="skipped",
            post_step_head_sha=target,
            result_summary="main already fast-forwarded",
        )
    if not ctx.git.is_ancestor(current_main, target):
        raise RoundInvariantError(
            round_no=ctx.round_no,
            message=f"main {current_main[:8]} is not an ancestor of {target[:8]}",
        )
    new_head = ctx.git.fast_forward(ctx.main_branch, target)
    ctx.registry.round_ff_main(
        round_no=ctx.round_no,
        from_sha=current_main,
        to_sha=new_head,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    ctx.git.checkout(ctx.staging_branch)
    _dbg("OC_CHAT.MAIN", func, f"main -> {new_head[:8]}", 2)
    return StepOutcome(
        step_name="ff_main",
        status="done",
        post_step_head_sha=_staging_head(ctx),
        result_summary=f"main fast-forwarded to {new_head[:8]}",
    )
 
 
def run_cleanup(ctx: RoundContext) -> StepOutcome:
    """Delete plan branches/worktrees, bump roadmap, close round when clean."""
 
    func = "run_cleanup"
    round_obj = ctx.registry.backend.get_round(ctx.round_no)
    if round_obj is not None and str(round_obj.get("status")) == "closed":
        _dbg("OC_CHAT.MAIN", func, "round already closed; skip", 2)
        return StepOutcome(
            step_name="cleanup",
            status="skipped",
            post_step_head_sha=_staging_head(ctx),
            result_summary="round already closed",
        )
    _audit_roadmap_hand_edits(ctx)
    branches_deleted: list[str] = []
    worktrees_removed: list[str] = []
    failures: list[str] = []
    for spec in ctx.plan_specs:
        if spec.worktree:
            if ctx.git.worktree_remove(spec.worktree, force=True):
                worktrees_removed.append(spec.worktree)
            elif _worktree_present(ctx, spec.worktree):
                failures.append(f"worktree {spec.worktree}")
        if spec.branch and ctx.git.branch_exists(spec.branch):
            if ctx.git.branch_delete(spec.branch, force=True):
                branches_deleted.append(spec.branch)
            else:
                failures.append(f"branch {spec.branch}")
    roadmap_version: str | None = None
    roadmap_failed = False
    if not failures and ctx.roadmap_bumper is not None:
        try:
            roadmap_version = ctx.roadmap_bumper(ctx)
        except Exception as exc:  # noqa: BLE001 - surfaced as escalation
            roadmap_failed = True
            failures.append(f"roadmap bump: {exc}")
    if failures:
        ctx.registry.round_cleanup(
            round_no=ctx.round_no,
            outcome="failed",
            branches_deleted=branches_deleted,
            worktrees_removed=worktrees_removed,
            error_summary="; ".join(failures),
            actor_slot=ctx.actor_slot,
            actor_type=ctx.actor_type,
        )
        code = (
            EscalationCode.ROADMAP_BUMP_FAILED
            if roadmap_failed
            else EscalationCode.CLEANUP_FAILED
        )
        raise RoundEscalation(
            make_escalation(
                code,
                round_no=ctx.round_no,
                phase="cleanup",
                message="; ".join(failures),
                resume_command=f"oct agent round cleanup {ctx.round_no}",
                details={
                    "branches_deleted": branches_deleted,
                    "worktrees_removed": worktrees_removed,
                },
            )
        )
    ctx.registry.round_cleanup(
        round_no=ctx.round_no,
        outcome="clean",
        branches_deleted=branches_deleted,
        worktrees_removed=worktrees_removed,
        roadmap_version=roadmap_version,
        app_version=ctx.app_version,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _dbg("OC_CHAT.MAIN", func, f"closed round {ctx.round_no}", 2)
    return StepOutcome(
        step_name="cleanup",
        status="done",
        post_step_head_sha=_staging_head(ctx),
        result_summary=(
            f"deleted {len(branches_deleted)} branches, "
            f"{len(worktrees_removed)} worktrees"
        ),
    )
 
 
def _audit_roadmap_hand_edits(ctx: RoundContext) -> None:
    if ctx.roadmap_dir is None:
        return
    result = verify_roadmap_files_unchanged(ctx.registry, ctx.roadmap_dir)
    if result.ok:
        return
    drift = [item.to_dict() for item in result.drift]
    if ctx.allow_untracked_roadmap_edits:
        ctx.registry.execute_domain_command(
            "audit.override",
            ctx.actor_slot,
            {
                "version": 1,
                "kind": "roadmap_untracked_edit",
                "round_no": ctx.round_no,
                "roadmap_dir": ctx.roadmap_dir,
                "drift": drift,
            },
            actor_type=ctx.actor_type,
            related_round=ctx.round_no,
        )
        return
    first = result.drift[0]
    raise RoundEscalation(
        make_escalation(
            EscalationCode.ROADMAP_UNTRACKED_EDIT,
            round_no=ctx.round_no,
            phase="cleanup",
            message=(
                f"untracked ROADMAP edit in {first.source_file}: "
                f"{first.kind} {first.item_id}"
            ),
            resume_command=f"oct agent round cleanup {ctx.round_no}",
            details={
                "roadmap_dir": ctx.roadmap_dir,
                "files": result.files,
                "drift": drift,
            },
        )
    )
 
 
def _worktree_present(ctx: RoundContext, worktree: str) -> bool:
    result = ctx.git.run("worktree", "list", "--porcelain")
    return worktree.replace("\\", "/") in result.stdout.replace("\\", "/")
 
 
def run_abort(ctx: RoundContext, *, reason: str, drop_staging: bool = True) -> StepOutcome:
    """Abort the round: return plans to ready_for_merge and drop staging."""
 
    func = "run_abort"
    ctx.registry.round_abort(
        round_no=ctx.round_no,
        reason=reason,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    _reset_plans_to_ready(ctx)
    if drop_staging and ctx.git.branch_exists(ctx.staging_branch):
        ctx.git.checkout(ctx.base_ref)
        ctx.git.branch_delete(ctx.staging_branch, force=True)
    _dbg("OC_CHAT.MAIN", func, f"aborted round {ctx.round_no}", 2)
    return StepOutcome(
        step_name="abort",
        status="done",
        result_summary=f"aborted: {reason}",
    )
 
 
def run_rollback(ctx: RoundContext, *, reason: str) -> StepOutcome:
    """Roll back a diverged round: drop staging without touching plans."""
 
    func = "run_rollback"
    ctx.registry.round_abort(
        round_no=ctx.round_no,
        reason=reason,
        actor_slot=ctx.actor_slot,
        actor_type=ctx.actor_type,
    )
    if ctx.git.branch_exists(ctx.staging_branch):
        ctx.git.checkout(ctx.base_ref)
        ctx.git.branch_delete(ctx.staging_branch, force=True)
    _dbg("OC_CHAT.MAIN", func, f"rolled back round {ctx.round_no}", 2)
    return StepOutcome(
        step_name="rollback",
        status="done",
        result_summary=f"rolled back: {reason}",
    )
 
 
def _reset_plans_to_ready(ctx: RoundContext) -> None:
    func = "_reset_plans_to_ready"
    connect_func = getattr(ctx.registry.backend, "connect")
    with connect_func() as conn:
        rows = conn.execute(
            "SELECT p.plan_id AS plan_id, p.status AS status "
            "FROM round_plans rp JOIN plans p ON p.plan_id = rp.plan_id "
            "WHERE rp.round_no = ?",
            (ctx.round_no,),
        ).fetchall()
    for row in rows:
        current = str(row["status"])
        if current == "ready_for_merge":
            continue
        ctx.registry.plan_set_status(
            plan_id=str(row["plan_id"]),
            from_status=current,
            to_status="ready_for_merge",
            source="round_lifecycle",
            note=f"restored from aborted round {ctx.round_no}",
            actor_slot=ctx.actor_slot,
            actor_type=ctx.actor_type,
        )
    _dbg("OC_CHAT.MAIN", func, f"reset {len(rows)} plans", 3)