migrate.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/migrate.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_migration.py"
"""
Legacy queue/status migration into the M11 registry.
 
Purpose
-------
Project the Markdown-era orchestration state into the SQLite registry
without making the registry depend on those legacy files afterward.
 
Responsibilities
----------------
- Parse the legacy agent queue, plan headers, status files, and git evidence.
- Map legacy lifecycle names onto the ADR-00Z registry statuses.
- Build a deterministic projection for dry-run, import, verify, and rollback.
- Delegate all current-state SQL and operation-log writes to SQLiteBackend.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- No raw wall-clock imports are used here; timestamps come from the backend Clock.
- Import uses one SQLite transaction for projected rows and audit rows.
- Unknown legacy statuses fail the import before writing audit rows.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry._legacy_queue (read-only legacy queue parser; plan 223
  moved it here when the queue machinery retired)
- oc_chat.scripts.plan_status (legacy status file parser)
- oc_chat.registry.backends.sqlite (SQL owner for migration writes)
- oc_chat.registry (Registry facade passed to the import, verify and rollback entry points)
- yaml (fallback parse of legacy status-file frontmatter)
- git (not imported; run as a subprocess for worktree list, the first-parent log of main and rev-parse main, the legacy status-file and round evidence)
"""
from __future__ import annotations
 
from dataclasses import dataclass
import hashlib
import json
from pathlib import Path
import re
import subprocess
from typing import Mapping, Sequence
 
import yaml
 
from oc_diagnostics import _dbg
 
from oc_chat.registry import Registry
from oc_chat.registry import _legacy_queue
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.scripts import plan_status
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Regexes and model-family literals are parser fixtures, not secrets"
PLAN_ID_RE = re.compile(r"^(?P<number>\d{3})-(?P<slug>.+)$")
LEGACY_PLAN_ID_RE = re.compile(r"\b\d{3}-[A-Za-z0-9][A-Za-z0-9_.-]*\b")
ROUND_RE = re.compile(r"\bround[-\s]*(?P<round>\d+)\b", re.IGNORECASE)
HEADER_PATTERNS = {
    "branch": re.compile(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Regexes and model-family literals are parser fixtures, not secrets"
        r"^\*\*Branch:\*\*\s*(?:`(?P<ticks>[^`\n]+)`|(?P<plain>\S+))",
        re.IGNORECASE | re.MULTILINE,
    ),
    "worktree": re.compile(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Regexes and model-family literals are parser fixtures, not secrets"
        r"^\*\*Worktree:\*\*\s*(?:`(?P<ticks>[^`\n]+)`|(?P<plain>\S+))",
        re.IGNORECASE | re.MULTILINE,
    ),
    "plan_type": re.compile(
        r"^\*\*Plan Type:\*\*\s*(?P<value>.+?)\s*$",
        re.IGNORECASE | re.MULTILINE,
    ),
    "status": re.compile(
        r"^\*\*Status:\*\*\s*(?P<value>.+?)\s*$",
        re.IGNORECASE | re.MULTILINE,
    ),
    "suggested_model": re.compile(
        r"^\*\*Suggested model:\*\*\s*(?P<value>.+?)\s*$",
        re.IGNORECASE | re.MULTILINE,
    ),
}
H1_RE = re.compile(r"^#\s+(?P<value>.+?)\s*$", re.MULTILINE)
VALID_LEGACY_STATUSES = {
    "Completed",
    "In Progress",
    "Blocked",
    "Failed",
    "Partial",
    "Pending",
}
STATUS_NORMALISATIONS = {
    "queued": "Pending",
    "queue": "Pending",
    "pending": "Pending",
    "implementing": "In Progress",
    "in progress": "In Progress",
    "in-progress": "In Progress",
    "completed": "Completed",
    "partially completed": "Partial",
    "partial": "Partial",
    "failed": "Failed",
    "blocked": "Blocked",
}
PRE_ROUND_SYSTEM_PLAN_CEILING = 50
 
 
@dataclass(frozen=True)
class RoundEvidence:
    """Evidence that a plan was integrated by a closed mainline round."""
 
    plan_id: str
    round_no: int
    merge_commit_sha: str
    closed_at: str
 
 
@dataclass(frozen=True)
class MigrationOptions:
    """Filesystem inputs used by migration commands."""
 
    workspace_root: Path
    queue_path: Path | None = None
    plan_dir: Path | None = None
    status_paths: tuple[Path, ...] | None = None
    round_evidence: tuple[RoundEvidence, ...] | None = None
    discovered_in_main_at_sha: str | None = None
 
 
@dataclass(frozen=True)
class ImportResult:
    ok: bool
    dry_run: bool
    summary: dict[str, object]
    errors: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
    backup_path: str | None = None
 
    def to_dict(self) -> dict[str, object]:
        return _result_dict(self)
 
 
@dataclass(frozen=True)
class VerifyResult:
    ok: bool
    summary: dict[str, object]
    errors: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
    diff: dict[str, object] | None = None
 
    def to_dict(self) -> dict[str, object]:
        return _result_dict(self)
 
 
@dataclass(frozen=True)
class RollbackResult:
    ok: bool
    summary: dict[str, object]
    errors: tuple[str, ...] = ()
    restored_from: str | None = None
 
    def to_dict(self) -> dict[str, object]:
        return _result_dict(self)
 
 
@dataclass(frozen=True)
class CutoverResult:
    ok: bool
    exit_code: int
    message: str
 
    def to_dict(self) -> dict[str, object]:
        return _result_dict(self)
 
 
@dataclass(frozen=True)
class StatusSource:
    plan_id: str
    legacy_status: str
    source_path: Path
    source_file: str
    branch: str | None
    worktree: str | None
    agent_label: str | None
    model: str | None
    thinking: str | None
    session: str | None
    body_observed_at: str
    body: str
    blocked_reason: str | None
 
 
@dataclass(frozen=True)
class PlanHeader:
    plan_id: str
    plan_number: int
    title: str
    path: Path
    source_file: str
    plan_type: str | None
    branch: str | None
    worktree: str | None
    model: str | None
    thinking: str | None
    session: str | None
    legacy_status: str | None
    body_sha256: str
 
 
@dataclass(frozen=True)
class QueueEntry:
    plan_id: str
    slot: str
    position: int
    source_file: str
    model: str | None
    thinking: str | None
    session: str | None
 
 
@dataclass(frozen=True)
class MigrationProjection:
    agents: tuple[dict[str, object], ...]
    plans: tuple[dict[str, object], ...]
    queue_entries: tuple[dict[str, object], ...]
    rounds: tuple[dict[str, object], ...]
    round_plans: tuple[dict[str, object], ...]
    errors: tuple[str, ...]
    warnings: tuple[str, ...]
 
    def to_backend_payload(self) -> dict[str, object]:
        return {
            "agents": list(self.agents),
            "plans": list(self.plans),
            "queue_entries": list(self.queue_entries),
            "rounds": list(self.rounds),
            "round_plans": list(self.round_plans),
        }
 
    def expected_snapshot(self) -> dict[str, object]:
        return {
            "agents": _comparable_agents(self.agents),
            "plans": _comparable_plans(self.plans),
            "queue_entries": _comparable_queue_entries(self.queue_entries),
            "rounds": _comparable_rounds(self.rounds),
            "round_plans": _comparable_round_plans(self.round_plans),
        }
 
    def summary(self) -> dict[str, object]:
        status_counts: dict[str, int] = {}
        for plan in self.plans:
            status = str(plan["status"])
            status_counts[status] = status_counts.get(status, 0) + 1
        return {
            "agents": len(self.agents),
            "plans": len(self.plans),
            "queue_entries": len(self.queue_entries),
            "rounds": len(self.rounds),
            "round_plans": len(self.round_plans),
            "status_counts": dict(sorted(status_counts.items())),
        }
 
 
def build_projection(
    registry: Registry,
    options: MigrationOptions,
) -> MigrationProjection:
    """Build the deterministic legacy-to-registry projection."""
 
    backend = _sqlite_backend(registry)
    root = options.workspace_root.resolve()
    queue_path = (options.queue_path or root / _legacy_queue.DEFAULT_AGENT_QUEUE).resolve()
    plan_dir = (options.plan_dir or root / "oc_chat" / "plans").resolve()
    warnings: list[str] = []
    errors: list[str] = []
 
    agents = _build_agent_rows()
    headers = _read_plan_headers(plan_dir, root)
    pending, completed = _read_queue_entries(queue_path, root)
    status_sources = _read_status_sources(
        options.status_paths,
        root,
        backend.clock.now_wall(),
        warnings,
    )
    evidence = _round_evidence_by_plan(options, root)
    discovered_sha = options.discovered_in_main_at_sha or _main_sha(root)
 
    plan_ids = (
        set(headers)
        | set(pending)
        | set(status_sources)
        | (set(completed) & set(headers))
    )
    plan_ids = _dedupe_plan_ids_by_number(
        plan_ids,
        headers,
        pending,
        completed,
        status_sources,
        warnings,
    )
    plan_rows: list[dict[str, object]] = []
    queue_rows: list[dict[str, object]] = []
    rounds_by_no: dict[int, dict[str, object]] = {}
    round_plans: list[dict[str, object]] = []
    round_orders: dict[int, int] = {}
 
    for plan_id in sorted(plan_ids, key=_plan_sort_key):
        header = headers.get(plan_id) or _placeholder_header(plan_id, plan_dir, root)
        source = status_sources.get(plan_id)
        queue_entry = pending.get(plan_id)
        legacy_status = _select_legacy_status(plan_id, header, source, pending, completed)
        mapped = _map_legacy_status(
            plan_id,
            header.plan_number,
            legacy_status,
            source,
            evidence.get(plan_id),
            discovered_sha,
            errors,
        )
        if mapped is None:
            continue
 
        assigned_slot = _assigned_slot(queue_entry, source, completed.get(plan_id))
        queue_position = (
            queue_entry.position
            if queue_entry and mapped["status"] in {"pending", "implementing", "ready_for_merge"}
            else None
        )
        body_observed_at = source.body_observed_at if source is not None else None
        model, thinking, session = _select_overrides(header, source, queue_entry)
        branch = _normalise_branch(
            _first_text(source.branch if source else None, header.branch)
        )
        worktree = _first_text(source.worktree if source else None, header.worktree)
        row = {
            "plan_id": plan_id,
            "plan_number": header.plan_number,
            "title": header.title,
            "plan_type": header.plan_type,
            "status": mapped["status"],
            "assigned_slot": assigned_slot,
            "queue_position": queue_position,
            "branch": branch,
            "worktree": worktree,
            "body_path": header.source_file,
            "body_sha256": header.body_sha256,
            "body_observed_at": body_observed_at,
            "model_override": model,
            "thinking_override": thinking,
            "session_override": session,
            "ready_for_merge_at": mapped["ready_for_merge_at"],
            "completed_at": mapped["completed_at"],
            "closed_round": mapped["closed_round"],
            "legacy_status": legacy_status,
            "payload": _plan_payload(
                plan_id,
                header,
                legacy_status,
                mapped,
                source.source_file if source else header.source_file,
            ),
        }
        if mapped["backfill_payload"] is not None:
            row["backfill_payload"] = mapped["backfill_payload"]
        if mapped["blocked_payload"] is not None:
            row["blocked_payload"] = mapped["blocked_payload"]
        plan_rows.append(row)
 
        plan_evidence = evidence.get(plan_id)
        if plan_evidence is not None and mapped["closed_round"] is not None:
            rounds_by_no[plan_evidence.round_no] = _round_row(plan_evidence)
            round_orders[plan_evidence.round_no] = round_orders.get(
                plan_evidence.round_no,
                0,
            ) + 1
            round_plans.append(
                {
                    "round_no": plan_evidence.round_no,
                    "plan_id": plan_id,
                    "merge_order": round_orders[plan_evidence.round_no],
                    "merge_commit_sha": plan_evidence.merge_commit_sha,
                    "test_result": "pass",
                }
            )
 
        if queue_entry is not None:
            queue_rows.append(
                {
                    "plan_id": plan_id,
                    "slot": queue_entry.slot,
                    "position": queue_entry.position,
                    "source_file": queue_entry.source_file,
                    "payload": {
                        "version": 1,
                        "plan_id": plan_id,
                        "slot": queue_entry.slot,
                        "position": queue_entry.position,
                        "source_file": queue_entry.source_file,
                    },
                }
            )
 
    if queue_path.is_file() is False:
        warnings.append(f"legacy queue not found: {queue_path}")
    if not headers:
        warnings.append(f"no plan headers found in {plan_dir}")
    _clear_duplicate_branches(plan_rows, warnings)
 
    return MigrationProjection(
        agents=tuple(agents),
        plans=tuple(plan_rows),
        queue_entries=tuple(sorted(queue_rows, key=lambda row: (row["slot"], row["position"]))),
        rounds=tuple(rounds_by_no[key] for key in sorted(rounds_by_no)),
        round_plans=tuple(sorted(round_plans, key=lambda row: (row["round_no"], row["merge_order"]))),
        errors=tuple(errors),
        warnings=tuple(warnings),
    )
 
 
def run_import(
    registry: Registry,
    options: MigrationOptions,
    *,
    dry_run: bool = False,
) -> ImportResult:
    """Dry-run or apply the legacy migration projection."""
 
    projection = build_projection(registry, options)
    if projection.errors:
        return ImportResult(
            ok=False,
            dry_run=dry_run,
            summary=projection.summary(),
            errors=projection.errors,
            warnings=projection.warnings,
        )
    if dry_run:
        return ImportResult(
            ok=True,
            dry_run=True,
            summary=projection.summary(),
            warnings=projection.warnings,
        )
 
    backend = _sqlite_backend(registry)
    actor_slot = backend.active_operator_slot()
    if actor_slot is None:
        return ImportResult(
            ok=False,
            dry_run=False,
            summary=projection.summary(),
            errors=("no active human operator slot exists",),
            warnings=projection.warnings,
        )
 
    backup_path = _backup_path(options.workspace_root.resolve(), backend.clock.now_wall())
    backend.backup(backup_path)
    registry.acquire_lock(
        "migrate-import",
        actor_slot,
        3600,
        {"command": "migrate.import", "version": 1},
    )
    try:
        backend.apply_migration_projection(
            projection.to_backend_payload(),
            actor_slot=actor_slot,
            actor_type="human_operator",
        )
    finally:
        registry.release_lock("migrate-import", actor_slot)
 
    return ImportResult(
        ok=True,
        dry_run=False,
        summary=projection.summary(),
        warnings=projection.warnings,
        backup_path=str(backup_path),
    )
 
 
def run_verify(registry: Registry, options: MigrationOptions) -> VerifyResult:
    """Compare the projected legacy state with the registry tables."""
 
    projection = build_projection(registry, options)
    if projection.errors:
        return VerifyResult(
            ok=False,
            summary=projection.summary(),
            errors=projection.errors,
            warnings=projection.warnings,
        )
    backend = _sqlite_backend(registry)
    expected = projection.expected_snapshot()
    actual = backend.migration_snapshot()
    diff = _snapshot_diff(expected, actual)
    backfill_violations = backend.migration_null_round_backfill_violations()
    if backfill_violations:
        diff["null_round_backfill_violations"] = backfill_violations
    ok = not diff
    return VerifyResult(
        ok=ok,
        summary=projection.summary(),
        warnings=projection.warnings,
        diff=None if ok else diff,
    )
 
 
def run_rollback(registry: Registry, options: MigrationOptions) -> RollbackResult:
    """Restore the most recent pre-migration backup."""
 
    backend = _sqlite_backend(registry)
    backups = sorted(_backup_dir(options.workspace_root.resolve()).glob("pre-migrate-*.sqlite"))
    if not backups:
        return RollbackResult(
            ok=False,
            summary={"backups": 0},
            errors=("no pre-migration backup found",),
        )
    latest = backups[-1]
    backend.restore(latest)
    return RollbackResult(
        ok=True,
        summary={"restored": str(latest)},
        restored_from=str(latest),
    )
 
 
def run_cutover_stub() -> CutoverResult:
    """Return the v0.2.x cutover placeholder."""
 
    return CutoverResult(
        ok=False,
        exit_code=2,
        message="migrate cutover not implemented in v0.2.x; see M11.8",
    )
 
 
def audit_null_closed_round_backfills(registry: Registry) -> dict[str, object]:
    """Check the migration backfill marker post-condition."""
 
    violations = _sqlite_backend(registry).migration_null_round_backfill_violations()
    return {"ok": not violations, "violations": violations}
 
 
def _build_agent_rows() -> list[dict[str, object]]:
    rows: list[dict[str, object]] = []
    for label, info in _legacy_queue.LEGACY_AGENT_POOLS.items():
        slot = _agent_label_to_slot(label)
        role = str(info.get("role", "Worker"))
        slot_kind = "orchestrator" if "orchestrator" in role.lower() else "worker"
        model = _first_model(info.get("models"))
        model_class = "frontier"
        if slot_kind == "worker" and _looks_small_model(model):
            model_class = "small"
        orchestrator_eligible = slot_kind == "orchestrator"
        row = {
            "slot": slot,
            "slot_kind": slot_kind,
            "identity_kind": "agent",
            "model_class": model_class,
            "orchestrator_eligible": 1 if orchestrator_eligible else 0,
            "provider": _provider_from_model(model),
            "profile": str(info.get("profile", "")) or None,
        }
        row["payload"] = {
            "version": 1,
            "slot": slot,
            "slot_kind": slot_kind,
            "identity_kind": "agent",
            "model_class": model_class,
            "orchestrator_eligible": orchestrator_eligible,
            "provider": row["provider"],
            "profile": row["profile"],
            # The gitignored ``.oc_chat/agent_pool.json`` override sidecar
            # retired with the queue machinery (plan 223 P3): it had no
            # readers left and no copy on disk, so the committed roster is
            # now the only provenance this projection can name.
            # OCT-LINT: disable=no_hardcoded_secrets reason="module-path provenance label recorded in the migration payload, not a secret"
            "source_file": "oc_chat.registry._legacy_queue.LEGACY_AGENT_POOLS",
        }
        rows.append(row)
    return sorted(rows, key=lambda row: str(row["slot"]))
 
 
def _read_queue_entries(
    queue_path: Path,
    root: Path,
) -> tuple[dict[str, QueueEntry], dict[str, str]]:
    pending: dict[str, QueueEntry] = {}
    completed: dict[str, str] = {}
    queue = _legacy_queue.load_legacy_queue(queue_path)
    source_file = _display_path(queue_path, root)
    for label, section in queue.items():
        slot_id = _agent_label_to_slot(label)
        for index, entry in enumerate(section["pending"], start=1):
            plan_id = _legacy_queue.entry_plan_id(entry)
            model, thinking, session = _legacy_queue.parse_entry_overrides(entry)
            pending[plan_id] = QueueEntry(
                plan_id=plan_id,
                slot=slot_id,
                position=index,
                source_file=source_file,
                model=model,
                thinking=thinking,
                session=session,
            )
        for plan_id in section["completed"]:
            completed.setdefault(plan_id, slot_id)
    return pending, completed
 
 
def _read_plan_headers(plan_dir: Path, root: Path) -> dict[str, PlanHeader]:
    headers: dict[str, PlanHeader] = {}
    if not plan_dir.is_dir():
        return headers
    for path in sorted(plan_dir.glob("[0-9][0-9][0-9]-*.md")):
        plan_id = path.stem
        match = PLAN_ID_RE.match(plan_id)
        if match is None:
            continue
        text = path.read_text(encoding="utf-8")
        model, thinking, session = _parse_suggested_model(text)
        legacy_status = _normalise_header_status(_header_value(text, "status"))
        headers[plan_id] = PlanHeader(
            plan_id=plan_id,
            plan_number=int(match.group("number")),
            title=_parse_title(text, plan_id),
            path=path,
            source_file=_display_path(path, root),
            plan_type=_header_value(text, "plan_type"),
            branch=_header_value(text, "branch"),
            worktree=_header_value(text, "worktree"),
            model=model,
            thinking=thinking,
            session=session,
            legacy_status=legacy_status,
            body_sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
        )
    return headers
 
 
def _read_status_sources(
    explicit_paths: tuple[Path, ...] | None,
    root: Path,
    observed_at: str,
    warnings: list[str],
) -> dict[str, StatusSource]:
    paths = explicit_paths if explicit_paths is not None else tuple(_discover_status_paths(root))
    sources: dict[str, StatusSource] = {}
    for path in paths:
        if not path.is_file():
            continue
        try:
            source = _read_status_source(path, root, observed_at)
        except Exception as exc:
            warnings.append(f"could not parse {path}: {type(exc).__name__}: {exc}")
            continue
        sources[source.plan_id] = source
    return sources
 
 
def _read_status_source(path: Path, root: Path, observed_at: str) -> StatusSource:
    try:
        status, body = plan_status.read(path)
        data = status.model_dump(mode="json")
    except Exception:
        text = path.read_text(encoding="utf-8")
        frontmatter, body = plan_status._split_frontmatter(text)
        loaded = yaml.safe_load(frontmatter)
        if not isinstance(loaded, dict):
            raise ValueError("frontmatter must be a YAML mapping")
        data = loaded
 
    raw_agent = data.get("agent")
    agent_label = _agent_label(str(raw_agent)) if raw_agent else None
    legacy_status = _normalise_legacy_status(data.get("status"))
    blocked_reason = _first_text(
        data.get("blocked_reason"),
        data.get("blocker"),
        _first_nonempty_line(body) if legacy_status == "Blocked" else None,
    )
    body_observed_at = _first_text(
        _string_timestamp(data.get("finished_at")),
        _string_timestamp(data.get("started_at")),
        _string_timestamp(data.get("queued_at")),
        observed_at,
    )
    return StatusSource(
        plan_id=str(data["plan_id"]),
        legacy_status=legacy_status,
        source_path=path,
        source_file=_display_path(path, root),
        branch=_optional_text(data.get("branch")),
        worktree=_optional_text(data.get("worktree")),
        agent_label=agent_label,
        model=_optional_text(data.get("model")),
        thinking=_optional_text(data.get("thinking")),
        session=_optional_text(data.get("session")),
        body_observed_at=body_observed_at,
        body=body,
        blocked_reason=blocked_reason,
    )
 
 
def _discover_status_paths(root: Path) -> list[Path]:
    paths = {root / plan_status.DEFAULT_FILENAME}
    try:
        result = subprocess.run(
            ["git", "worktree", "list", "--porcelain"],
            cwd=str(root),
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
    except (OSError, ValueError):
        result = None
    if result is not None and result.returncode == 0:
        for line in result.stdout.splitlines():
            if line.startswith("worktree "):
                paths.add(Path(line.removeprefix("worktree ")).resolve() / plan_status.DEFAULT_FILENAME)
    return sorted(path for path in paths if path.is_file())
 
 
def _round_evidence_by_plan(
    options: MigrationOptions,
    root: Path,
) -> dict[str, RoundEvidence]:
    evidence = list(options.round_evidence or ())
    if options.round_evidence is None:
        evidence.extend(_discover_round_evidence(root))
    by_plan: dict[str, RoundEvidence] = {}
    for item in evidence:
        by_plan[item.plan_id] = item
    return by_plan
 
 
def _discover_round_evidence(root: Path) -> list[RoundEvidence]:
    try:
        result = subprocess.run(
            [
                "git",
                "log",
                "--first-parent",
                "main",
                "--pretty=format:%H%x1f%aI%x1f%s%x1f%b%x1e",
            ],
            cwd=str(root),
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
    except (OSError, ValueError):
        return []
    if result.returncode != 0:
        return []
 
    evidence: list[RoundEvidence] = []
    for record in result.stdout.split("\x1e"):
        fields = record.strip().split("\x1f", 3)
        if len(fields) != 4:
            continue
        commit_sha, closed_at, subject, body = fields
        text = f"{subject}\n{body}"
        round_match = ROUND_RE.search(text)
        if round_match is None:
            continue
        round_no = int(round_match.group("round"))
        for plan_id in sorted(set(LEGACY_PLAN_ID_RE.findall(text))):
            evidence.append(
                RoundEvidence(
                    plan_id=plan_id,
                    round_no=round_no,
                    merge_commit_sha=commit_sha,
                    closed_at=closed_at,
                )
            )
    return evidence
 
 
def _main_sha(root: Path) -> str | None:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "main"],
            cwd=str(root),
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
    except (OSError, ValueError):
        return None
    if result.returncode != 0:
        return None
    return result.stdout.strip() or None
 
 
def _select_legacy_status(
    plan_id: str,
    header: PlanHeader,
    source: StatusSource | None,
    pending: Mapping[str, QueueEntry],
    completed: Mapping[str, str],
) -> str:
    if source is not None:
        return source.legacy_status
    if header.legacy_status is not None:
        return header.legacy_status
    if plan_id in pending:
        return "Pending"
    if plan_id in completed:
        return "Completed"
    return "Pending"
 
 
def _dedupe_plan_ids_by_number(
    plan_ids: set[str],
    headers: Mapping[str, PlanHeader],
    pending: Mapping[str, QueueEntry],
    completed: Mapping[str, str],
    status_sources: Mapping[str, StatusSource],
    warnings: list[str],
) -> set[str]:
    chosen: dict[int, str] = {}
    for plan_id in sorted(plan_ids, key=_plan_sort_key):
        number = _plan_number(plan_id, headers)
        previous = chosen.get(number)
        if previous is None:
            chosen[number] = plan_id
            continue
        winner = max(
            (previous, plan_id),
            key=lambda item: _plan_priority(item, pending, completed, status_sources),
        )
        loser = plan_id if winner == previous else previous
        chosen[number] = winner
        warnings.append(
            f"duplicate plan_number {number}: kept {winner}, skipped {loser}"
        )
    return set(chosen.values())
 
 
def _plan_number(plan_id: str, headers: Mapping[str, PlanHeader]) -> int:
    header = headers.get(plan_id)
    if header is not None:
        return header.plan_number
    match = PLAN_ID_RE.match(plan_id)
    return int(match.group("number")) if match else 0
 
 
def _plan_priority(
    plan_id: str,
    pending: Mapping[str, QueueEntry],
    completed: Mapping[str, str],
    status_sources: Mapping[str, StatusSource],
) -> tuple[int, str]:
    if plan_id in status_sources or plan_id in pending:
        return (3, plan_id)
    if plan_id in completed:
        return (2, plan_id)
    return (1, plan_id)
 
 
def _map_legacy_status(
    plan_id: str,
    plan_number: int,
    legacy_status: str,
    source: StatusSource | None,
    evidence: RoundEvidence | None,
    discovered_sha: str | None,
    errors: list[str],
) -> dict[str, object] | None:
    if legacy_status not in VALID_LEGACY_STATUSES:
        errors.append(f"{plan_id}: unknown legacy status {legacy_status!r}")
        return None
 
    observed_at = source.body_observed_at if source is not None else None
    base = {
        "status": "pending",
        "ready_for_merge_at": None,
        "completed_at": None,
        "closed_round": None,
        "backfill_payload": None,
        "blocked_payload": None,
    }
    if legacy_status == "Completed":
        if evidence is not None:
            base.update(
                {
                    "status": "completed",
                    "completed_at": evidence.closed_at,
                    "closed_round": evidence.round_no,
                }
            )
        elif plan_number < PRE_ROUND_SYSTEM_PLAN_CEILING:
            completed_at = observed_at
            base.update({"status": "completed", "completed_at": completed_at})
            base["backfill_payload"] = {
                "version": 1,
                "plan_id": plan_id,
                "reason": "pre-round-system completed plan",
                "body_observed_at": completed_at,
                "discovered_in_main_at_sha": discovered_sha,
            }
        else:
            base.update(
                {
                    "status": "ready_for_merge",
                    "ready_for_merge_at": observed_at,
                }
            )
    elif legacy_status == "In Progress":
        base["status"] = "implementing"
    elif legacy_status == "Blocked":
        reason = source.blocked_reason if source is not None else None
        base["blocked_payload"] = {
            "version": 1,
            "plan_id": plan_id,
            "original_blocked_reason": reason or "legacy status was Blocked",
        }
    elif legacy_status == "Failed":
        base["status"] = "failed"
    elif legacy_status == "Partial":
        base["status"] = "partial"
    return base
 
 
def _plan_payload(
    plan_id: str,
    header: PlanHeader,
    legacy_status: str,
    mapped: Mapping[str, object],
    source_file: str,
) -> dict[str, object]:
    return {
        "version": 1,
        "plan_id": plan_id,
        "plan_number": header.plan_number,
        "legacy_status": legacy_status,
        "mapped_status": mapped["status"],
        "source_file": source_file,
        "body_path": header.source_file,
        "body_sha256": header.body_sha256,
        "ready_for_merge_at": mapped["ready_for_merge_at"],
        "closed_round": mapped["closed_round"],
        "completed_at": mapped["completed_at"],
    }
 
 
def _round_row(evidence: RoundEvidence) -> dict[str, object]:
    return {
        "round_no": evidence.round_no,
        "status": "closed",
        "cleanup_status": "clean",
        "orchestrator_slot": "agent-a",
        "opened_at": evidence.closed_at,
        "closed_at": evidence.closed_at,
        "final_commit_sha": evidence.merge_commit_sha,
    }
 
 
def _placeholder_header(plan_id: str, plan_dir: Path, root: Path) -> PlanHeader:
    match = PLAN_ID_RE.match(plan_id)
    number = int(match.group("number")) if match else 0
    path = plan_dir / f"{plan_id}.md"
    return PlanHeader(
        plan_id=plan_id,
        plan_number=number,
        title=plan_id,
        path=path,
        source_file=_display_path(path, root),
        plan_type=None,
        branch=None,
        worktree=None,
        model=None,
        thinking=None,
        session=None,
        legacy_status=None,
        body_sha256="",
    )
 
 
def _assigned_slot(
    queue_entry: QueueEntry | None,
    source: StatusSource | None,
    completed_slot: str | None,
) -> str | None:
    if queue_entry is not None:
        return queue_entry.slot
    if source is not None and source.agent_label is not None:
        return _agent_label_to_slot(source.agent_label)
    return completed_slot
 
 
def _select_overrides(
    header: PlanHeader,
    source: StatusSource | None,
    queue_entry: QueueEntry | None,
) -> tuple[str | None, str | None, str | None]:
    return (
        _first_text(source.model if source else None, queue_entry.model if queue_entry else None, header.model),
        _first_text(source.thinking if source else None, queue_entry.thinking if queue_entry else None, header.thinking),
        _first_text(source.session if source else None, queue_entry.session if queue_entry else None, header.session),
    )
 
 
def _parse_title(text: str, plan_id: str) -> str:
    match = H1_RE.search(text)
    return plan_id if match is None else match.group("value").strip()
 
 
def _header_value(text: str, name: str) -> str | None:
    match = HEADER_PATTERNS[name].search(text)
    if match is None:
        return None
    if "value" in match.groupdict():
        return _clean_header_value(match.group("value"))
    return _clean_header_value(match.group("ticks") or match.group("plain"))
 
 
def _parse_suggested_model(text: str) -> tuple[str | None, str | None, str | None]:
    value = _header_value(text, "suggested_model")
    if value is None:
        return None, None, None
    # OCT-LINT: disable=no_hardcoded_secrets reason="Regexes and model-family literals are parser fixtures, not secrets"
    model_match = re.search(r"\b(?:gpt|claude|deepseek)[A-Za-z0-9_.-]*\b", value)
    # OCT-LINT: disable=no_hardcoded_secrets reason="Regexes and model-family literals are parser fixtures, not secrets"
    thinking_match = re.search(r"\bthinking\s*[=:]\s*(off|low|medium|high|max)\b", value)
    session_match = re.search(r"\bsession\s*[=:]\s*(fresh|continue|either)\b", value)
    return (
        None if model_match is None else model_match.group(0),
        None if thinking_match is None else thinking_match.group(1),
        None if session_match is None else session_match.group(1),
    )
 
 
def _clean_header_value(value: str | None) -> str | None:
    if value is None:
        return None
    text = value.strip().strip("`").strip()
    return text or None
 
 
def _normalise_branch(value: str | None) -> str | None:
    text = _optional_text(value)
    if text is None:
        return None
    if text.lower() in {"main", "none", "n/a", "na", "tbd"}:
        return None
    return text
 
 
def _clear_duplicate_branches(
    plan_rows: Sequence[dict[str, object]],
    warnings: list[str],
) -> None:
    by_branch: dict[str, list[dict[str, object]]] = {}
    for row in plan_rows:
        branch = _optional_text(row.get("branch"))
        if branch is not None:
            by_branch.setdefault(branch, []).append(row)
    for branch, rows in by_branch.items():
        if len(rows) < 2:
            continue
        plan_ids = ", ".join(str(row["plan_id"]) for row in rows)
        warnings.append(f"duplicate branch {branch}: cleared for {plan_ids}")
        for row in rows:
            row["branch"] = None
            payload = row.get("payload")
            if isinstance(payload, dict):
                payload["branch"] = None
 
 
def _normalise_legacy_status(value: object) -> str:
    text = _optional_text(value)
    if text is None:
        return "Pending"
    key = text.strip().lower()
    return STATUS_NORMALISATIONS.get(key, text.strip())
 
 
def _normalise_header_status(value: object) -> str | None:
    if value is None:
        return None
    status = _normalise_legacy_status(value)
    return status if status in VALID_LEGACY_STATUSES else None
 
 
def _agent_label(value: str) -> str:
    text = value.strip()
    if text.lower().startswith("agent "):
        return f"Agent {text.split()[-1].upper()}"
    if text.lower().startswith("agent-"):
        return f"Agent {text.rsplit('-', 1)[-1].upper()}"
    if len(text) == 1:
        return f"Agent {text.upper()}"
    return text
 
 
def _agent_label_to_slot(label: str) -> str:
    text = label.strip().lower()
    if text.startswith("agent-"):
        return text
    if text.startswith("agent "):
        return f"agent-{text.split()[-1]}"
    if len(text) == 1:
        return f"agent-{text}"
    return text.replace(" ", "-")
 
 
def _first_model(value: object) -> str | None:
    if isinstance(value, str):
        return value
    if isinstance(value, Sequence):
        for item in value:
            text = _optional_text(item)
            if text is not None:
                return text
    return None
 
 
def _looks_small_model(model: str | None) -> bool:
    if model is None:
        return False
    lowered = model.lower()
    return any(marker in lowered for marker in ("mini", "flash", "haiku", "small"))
 
 
def _provider_from_model(model: str | None) -> str | None:
    if model is None:
        return None
    lowered = model.lower()
    if lowered.startswith("gpt"):
        return "openai"
    if lowered.startswith("claude"):
        return "anthropic"
    if lowered.startswith("deepseek"):
        return "deepseek"
    return None
 
 
def _display_path(path: Path, root: Path) -> str:
    try:
        return path.resolve().relative_to(root.resolve()).as_posix()
    except ValueError:
        return str(path)
 
 
def _string_timestamp(value: object) -> str | None:
    if value is None:
        return None
    if hasattr(value, "isoformat"):
        return str(value.isoformat())
    text = str(value)
    return text if text else None
 
 
def _optional_text(value: object) -> str | None:
    if value is None:
        return None
    if hasattr(value, "value"):
        value = value.value
    text = str(value).strip()
    return text or None
 
 
def _first_text(*values: object) -> str | None:
    for value in values:
        text = _optional_text(value)
        if text is not None:
            return text
    return None
 
 
def _first_nonempty_line(text: str) -> str | None:
    for line in text.splitlines():
        stripped = line.strip()
        if stripped:
            return stripped
    return None
 
 
def _plan_sort_key(plan_id: str) -> tuple[int, str]:
    match = PLAN_ID_RE.match(plan_id)
    return (int(match.group("number")) if match else 999999, plan_id)
 
 
def _backup_dir(root: Path) -> Path:
    return root / ".option_c" / "orchestration" / "backups"
 
 
def _backup_path(root: Path, now: str) -> Path:
    safe = now.replace(":", "-").replace("+", "_").replace("/", "-")
    return _backup_dir(root) / f"pre-migrate-{safe}.sqlite"
 
 
def _sqlite_backend(registry: Registry) -> SQLiteBackend:
    backend = registry.backend
    if not isinstance(backend, SQLiteBackend):
        raise TypeError("migrate commands require SQLiteBackend")
    return backend
 
 
def _comparable_agents(rows: Sequence[Mapping[str, object]]) -> list[dict[str, object]]:
    return [
        {
            "slot": row["slot"],
            "slot_kind": row["slot_kind"],
            "identity_kind": row["identity_kind"],
            "model_class": row["model_class"],
            "orchestrator_eligible": int(row["orchestrator_eligible"]),
            "provider": row.get("provider"),
            "profile": row.get("profile"),
            "active": 1,
        }
        for row in sorted(rows, key=lambda item: str(item["slot"]))
    ]
 
 
def _comparable_plans(rows: Sequence[Mapping[str, object]]) -> list[dict[str, object]]:
    keys = (
        "plan_id",
        "plan_number",
        "title",
        "plan_type",
        "status",
        "assigned_slot",
        "queue_position",
        "branch",
        "worktree",
        "body_path",
        "body_sha256",
        "body_observed_at",
        "model_override",
        "thinking_override",
        "session_override",
        "ready_for_merge_at",
        "completed_at",
        "closed_round",
    )
    return [
        {key: row.get(key) for key in keys}
        for row in sorted(rows, key=lambda item: str(item["plan_id"]))
    ]
 
 
def _comparable_queue_entries(
    rows: Sequence[Mapping[str, object]],
) -> list[dict[str, object]]:
    return [
        {
            "plan_id": row["plan_id"],
            "slot": row["slot"],
            "position": row["position"],
        }
        for row in sorted(rows, key=lambda item: (str(item["slot"]), int(item["position"])))
    ]
 
 
def _comparable_rounds(rows: Sequence[Mapping[str, object]]) -> list[dict[str, object]]:
    keys = (
        "round_no",
        "status",
        "cleanup_status",
        "orchestrator_slot",
        "opened_at",
        "closed_at",
        "final_commit_sha",
    )
    return [
        {key: row.get(key) for key in keys}
        for row in sorted(rows, key=lambda item: int(item["round_no"]))
    ]
 
 
def _comparable_round_plans(
    rows: Sequence[Mapping[str, object]],
) -> list[dict[str, object]]:
    keys = ("round_no", "plan_id", "merge_order", "merge_commit_sha", "test_result")
    return [
        {key: row.get(key) for key in keys}
        for row in sorted(
            rows,
            key=lambda item: (int(item["round_no"]), int(item["merge_order"])),
        )
    ]
 
 
def _snapshot_diff(
    expected: Mapping[str, object],
    actual: Mapping[str, object],
) -> dict[str, object]:
    diff: dict[str, object] = {}
    for section in ("agents", "plans", "queue_entries", "rounds", "round_plans"):
        exp = expected.get(section)
        act = actual.get(section)
        if exp != act:
            diff[section] = {
                "expected_count": len(exp) if isinstance(exp, list) else None,
                "actual_count": len(act) if isinstance(act, list) else None,
                "expected": exp,
                "actual": act,
            }
    return diff
 
 
def _result_dict(value: object) -> dict[str, object]:
    result: dict[str, object] = {}
    for name in getattr(value, "__dataclass_fields__", {}):
        item = getattr(value, name)
        if isinstance(item, tuple):
            result[name] = list(item)
        else:
            result[name] = item
    return result