round.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/round.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_round_close_e2e.py"
"""
Registry-native round-close commands.
 
Purpose
-------
Expose the seven-phase round-close orchestration beneath
``python -m oc_chat.registry round``: close, resume, abort, and status.
 
Responsibilities
----------------
- Parse round-close command arguments and plan specs.
- Resolve the registry, orchestrator actor, and the git seam.
- Drive run_round_close / resume_round / run_abort and render results.
- Render the versioned escalation document when a job stops.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- ``status`` is read-only; it never mutates the registry or git.
- Interactive prompting is the default; ``--non-interactive`` disables it.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.registry (Registry facade carried in the round context)
- oc_chat.registry.errors (PreflightFailedError reported by the resume command)
- oc_chat.registry.round.executors (round context, plan specs, test and lint results and run_abort)
- oc_chat.registry.round.git_shim (GitShim git seam over the workspace repository)
- oc_chat.registry.round.resume (resume_round for the resume command)
- oc_chat.registry.round.state_machine (run_round_close, step definitions and job ids)
- pytest (not imported; python -m pytest subprocess run by the default round-close test runner)
- oct_lint (not imported; python -m oct_lint subprocess run by the default round-close lint runner)
"""
from __future__ import annotations
 
import argparse
import json
from pathlib import Path
import subprocess
import sys
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.errors import PreflightFailedError
from oc_chat.registry.round.executors import (
    LintRunResult,
    PlanSpec,
    RoundContext,
    TestRunResult,
    run_abort,
)
from oc_chat.registry.round.git_shim import GitShim
from oc_chat.registry.round.state_machine import (
    RoundCloseResult,
    StepDef,
    new_job_id,
    run_round_close,
)
from oc_chat.registry.round.resume import resume_round
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_round_subcommands(round_parser: argparse.ArgumentParser) -> None:
    """Attach registry-native round-close subcommands."""
 
    round_parser.set_defaults(handler=_cmd_round_group, round_parser=round_parser)
    subcommands = round_parser.add_subparsers(dest="round_command")
 
    close = subcommands.add_parser("close", help="Drive a fresh round-close")
    _add_common(close)
    close.add_argument("round_no", type=int)
    close.add_argument(
        "--plan",
        action="append",
        default=[],
        metavar="PLAN_ID:BRANCH:WORKTREE[:ORDER]",
        help="A plan to merge; repeatable. Order defaults to argument order.",
    )
    close.add_argument("--staging-branch", help="Override the staging branch name.")
    close.add_argument("--base-ref", default="main")
    close.add_argument("--main-branch", default="main")
    close.add_argument("--app-version")
    close.add_argument("--roadmap-dir", help="ROADMAP artifact directory.")
    close.add_argument(
        "--allow-untracked-roadmap-edits",
        action="store_true",
        help="Proceed through cleanup after recording a ROADMAP audit override.",
    )
    close.add_argument(
        "--non-interactive",
        action="store_true",
        help="Do not prompt at phase boundaries (default is interactive).",
    )
    close.set_defaults(handler=_cmd_close)
 
    resume = subcommands.add_parser("resume", help="Resume a crashed round-close")
    _add_common(resume)
    resume.add_argument("round_no", type=int)
    resume.add_argument("--plan", action="append", default=[])
    resume.add_argument("--staging-branch")
    resume.add_argument("--base-ref", default="main")
    resume.add_argument("--main-branch", default="main")
    resume.add_argument("--roadmap-dir", help="ROADMAP artifact directory.")
    resume.add_argument(
        "--allow-untracked-roadmap-edits",
        action="store_true",
        help="Proceed through cleanup after recording a ROADMAP audit override.",
    )
    resume.add_argument("--non-interactive", action="store_true")
    resume.set_defaults(handler=_cmd_resume)
 
    abort = subcommands.add_parser("abort", help="Abort a round and reset its plans")
    _add_common(abort)
    abort.add_argument("round_no", type=int)
    abort.add_argument("--plan", action="append", default=[])
    abort.add_argument("--staging-branch")
    abort.add_argument("--base-ref", default="main")
    abort.add_argument("--reason", required=True)
    abort.set_defaults(handler=_cmd_abort)
 
    status = subcommands.add_parser("status", help="Show round and job status")
    status.add_argument("round_no", type=int)
    status.add_argument("--registry", help="Registry SQLite path.")
    status.add_argument("--json", action="store_true")
    status.set_defaults(handler=_cmd_status)
 
 
def _cmd_round_group(args: argparse.Namespace) -> int:
    args.round_parser.print_help()
    return 1
 
 
def _interactive_prompter(step: StepDef, _ctx: RoundContext) -> str:
    """Prompt the operator at a phase boundary; default is to continue."""
 
    reply = input(f"[round] proceed with phase {step.step_name}? [Y/abort] ").strip()
    return "abort" if reply.lower() in {"abort", "a", "n", "no"} else "continue"
 
 
def _cmd_close(args: argparse.Namespace) -> int:
    ctx = _build_context(args)
    prompter = None if args.non_interactive else _interactive_prompter
    result = run_round_close(
        ctx,
        interactive=not args.non_interactive,
        prompter=prompter,
    )
    return _render_result(result)
 
 
def _cmd_resume(args: argparse.Namespace) -> int:
    ctx = _build_context(args)
    prompter = None if args.non_interactive else _interactive_prompter
    try:
        result = resume_round(
            ctx,
            interactive=not args.non_interactive,
            prompter=prompter,
        )
    except PreflightFailedError as exc:
        print(f"resume preflight failed: {'; '.join(exc.reasons)}", file=sys.stderr)
        return 1
    return _render_result(result)
 
 
def _cmd_abort(args: argparse.Namespace) -> int:
    ctx = _build_context(args)
    outcome = run_abort(ctx, reason=args.reason)
    print(f"aborted round {ctx.round_no}: {outcome.result_summary}")
    return 0
 
 
def _cmd_status(args: argparse.Namespace) -> int:
    registry, backend = _registry(args.registry)
    round_row = backend.get_round(args.round_no)
    if round_row is None:
        print(f"round not found: {args.round_no}")
        return 1
    job = backend.latest_round_job(args.round_no)
    steps = backend.list_round_job_steps(str(job["job_id"])) if job else []
    if args.json:
        payload = {"round": round_row, "job": job, "steps": steps}
        print(json.dumps(payload, indent=2, sort_keys=True, default=str))
        return 0
    print(f"round {args.round_no}: {round_row['status']} / {round_row['cleanup_status']}")
    if job is not None:
        print(f"job {job['job_id']}: {job['status']} @ {job['current_step'] or '-'}")
        for step in steps:
            print(f"  {step['step_no']:>2} {step['step_name']:<24} {step['status']}")
    return 0
 
 
def _build_context(args: argparse.Namespace) -> RoundContext:
    registry, backend = _registry(args.registry)
    actor_slot = _resolve_orchestrator(backend, getattr(args, "actor_slot", None))
    actor_type = _actor_type_for_slot(backend, actor_slot)
    location = discover_registry_location(registry_path=args.registry)
    repo_dir = Path(location.workspace_root)
    git = GitShim(repo_dir)
    staging = args.staging_branch or f"round/{args.round_no}/staging"
    specs = _parse_plan_specs(args.plan)
    roadmap_dir = getattr(args, "roadmap_dir", None)
    if roadmap_dir is None and hasattr(args, "allow_untracked_roadmap_edits"):
        roadmap_dir = str(repo_dir / "oc_chat" / "roadmap")
    return RoundContext(
        registry=registry,
        git=git,
        round_no=args.round_no,
        job_id=new_job_id(),
        actor_slot=actor_slot,
        staging_branch=staging,
        plan_specs=specs,
        actor_type=actor_type,
        main_branch=getattr(args, "main_branch", "main"),
        base_ref=getattr(args, "base_ref", "main"),
        test_runner=_default_test_runner(repo_dir),
        lint_runner=_default_lint_runner(repo_dir),
        roadmap_bumper=_default_roadmap_bumper(actor_slot, actor_type),
        roadmap_dir=roadmap_dir,
        allow_untracked_roadmap_edits=bool(
            getattr(args, "allow_untracked_roadmap_edits", False)
        ),
        app_version=getattr(args, "app_version", None),
    )
 
 
def _parse_plan_specs(raw_specs: list[str]) -> list[PlanSpec]:
    specs: list[PlanSpec] = []
    for index, raw in enumerate(raw_specs, start=1):
        parts = raw.split(":")
        if len(parts) < 3:
            raise ValueError(
                # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help and SQL field names are not secrets"
                f"--plan must be PLAN_ID:BRANCH:WORKTREE[:ORDER], got {raw!r}"
            )
        order = int(parts[3]) if len(parts) >= 4 and parts[3] else index
        specs.append(
            PlanSpec(
                plan_id=parts[0],
                branch=parts[1],
                worktree=parts[2],
                merge_order=order,
            )
        )
    return specs
 
 
def _default_test_runner(repo_dir: Path):
    def _run(ctx: RoundContext) -> TestRunResult:
        clock = ctx.registry.clock
        started_ns = clock.now_monotonic_ns()
        completed = subprocess.run(
            [sys.executable, "-m", "pytest", "-q"],
            cwd=str(repo_dir),
            capture_output=True,
            text=True,
            check=False,
        )
        duration = (clock.now_monotonic_ns() - started_ns) / 1_000_000_000
        return TestRunResult(
            passed=completed.returncode == 0,
            duration_seconds=duration,
        )
 
    return _run
 
 
def _default_lint_runner(repo_dir: Path):
    def _run(_ctx: RoundContext) -> LintRunResult:
        completed = subprocess.run(
            [sys.executable, "-m", "oct_lint", "--profile", "strict"],
            cwd=str(repo_dir),
            capture_output=True,
            text=True,
            check=False,
        )
        return LintRunResult(passed=completed.returncode == 0)
 
    return _run
 
 
def _default_roadmap_bumper(actor_slot: str, actor_type: str):
    func = "_default_roadmap_bumper"
    _dbg("OC_CHAT.MAIN", func, "roadmap bump deferred to operator", 2)
 
    def _bump(_ctx: RoundContext) -> str | None:
        # Reuse the M11.6 roadmap-version bump in-process (ADR: no separate
        # round.bump_roadmap operation). Wired by the operator at call time.
        return None
 
    return _bump
 
 
def _render_result(result: RoundCloseResult) -> int:
    if result.escalation is not None:
        print(json.dumps(result.escalation.to_dict(), indent=2, sort_keys=True))
        return 2
    print(f"round-close {result.status}: {len(result.completed_steps)} steps")
    return 0 if result.status == "done" else 1
 
 
def _resolve_orchestrator(backend: SQLiteBackend, actor_slot: str | None) -> str:
    if actor_slot:
        return actor_slot
    with backend.connect() as conn:
        row = conn.execute(
            "SELECT orchestrator_slot FROM orchestration_state WHERE id = 1"
        ).fetchone()
    if row is not None and row["orchestrator_slot"]:
        return str(row["orchestrator_slot"])
    operator_slot = backend.active_operator_slot()
    if operator_slot is None:
        raise ValueError("no active orchestrator or operator slot")
    return operator_slot
 
 
def _actor_type_for_slot(backend: SQLiteBackend, slot: str) -> str:
    slot_kind = backend.actor_slot_kind(slot)
    return {
        "operator": "human_operator",
        "orchestrator": "orchestrator_agent",
        "worker": "worker_agent",
    }[slot_kind]
 
 
def _registry(registry_path: str | None) -> tuple[Registry, SQLiteBackend]:
    location = discover_registry_location(registry_path=registry_path)
    backend = SQLiteBackend(location.registry_path)
    return Registry(backend), backend
 
 
def _add_common(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--registry", help="Registry SQLite path.")
    parser.add_argument(
        "--actor-slot",
        help="Orchestrator slot. Defaults to the current orchestrator.",
    )