roadmap.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/roadmap.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_cli.py"
"""
CLI commands for registry-backed ROADMAP data.
 
Purpose
-------
Expose roadmap import, export, audit, bump, and status commands.
 
Responsibilities
----------------
- Resolve registry and workspace paths through standard discovery.
- Dispatch roadmap corpus ingest, artifact export, and drift audit helpers.
- Dispatch the Plan 205 reconciliation check, which sits beside ``render
  --check`` and answers a different question from it.
- Emit compact human or JSON command output.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - CLI lifecycle
    L3 - command argument summaries
    L4 - subcommand dispatch details
 
Contracts
---------
- Mutating bump operations route through the registry seam.
- Export writes only to the requested output directory.
 
Dependencies
------------
- argparse (CLI parsing)
- oc_chat.registry.roadmap.reconcile (the Plan 205 claims-vs-state check)
- oc_chat.registry.api (Registry facade over the live registry backend)
- oc_chat.registry.backends.sqlite (SQLiteBackend for the live registry and the status counts)
- oc_chat.registry.cli.source (source and hash-preimage subcommand wiring)
- oc_chat.registry.discovery (registry and workspace path discovery)
- oc_chat.registry.regenerate.envelope (error envelope and exit code for render failures)
- oc_chat.registry.render.roadmap (render check and living set writer)
- oc_chat.registry.roadmap.audit (audit_corpus for the audit command)
- oc_chat.registry.roadmap.ingest (ingest_corpus for the import command)
- oc_chat.registry.roadmap.operations (roadmap_version_bump for the bump command)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import argparse
import json
from pathlib import Path
import sys
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.roadmap.audit import audit_corpus
from oc_chat.registry.roadmap.ingest import ingest_corpus
from oc_chat.registry.roadmap.operations import roadmap_version_bump
from oc_chat.registry.roadmap.reconcile import (
    NOT_DRIFT_BANNER,
    JoinWriteResult,
    ReconcileConfigError,
    reconcile_workspace,
    write_plan_links,
)
from oc_chat.registry.render.roadmap import render_check, write_living_set
from oc_chat.registry.regenerate.envelope import (
    RegenerateError,
    SourceError,
    build_envelope,
    exit_code_for,
)
from oc_chat.registry.cli.source import add_source_subcommands
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: The round-close discipline — a declared ``advances:`` field on every plan and a
#: status flip (or explicit ``none``) at the gate — landed 2026-08-14 with plan 205.
#: The reconcile check grandfathers everything before it: plans 014-204 legitimately
#: predate the ``advances:`` mechanism, and changelog records written before the rule
#: cannot be held to it. These are the DEFAULTS for the two grandfather knobs, chosen
#: so a bare ``reconcile`` reports today's actionable state, not ~525 historical
#: not-applicables. Override (earlier date / lower plan id / empty) to audit history.
_DISCIPLINE_LANDED = "2026-08-14"
_ADVANCES_REQUIRED_FROM = "205"
 
 
def add_roadmap_subcommands(subcommands: argparse._SubParsersAction) -> None:
    """Attach the ``roadmap`` command group."""
 
    parser = subcommands.add_parser(
        "roadmap",
        help="Import, export, audit, and version ROADMAP registry data.",
    )
    roadmap_commands = parser.add_subparsers(dest="roadmap_command")
 
    import_parser = roadmap_commands.add_parser("import", help="Import ROADMAP markdown.")
    import_parser.add_argument("registry", nargs="?")
    import_parser.add_argument("--dir", dest="roadmap_dir")
    import_parser.add_argument("--dry-run", action="store_true")
    import_parser.add_argument("--actor-slot")
    import_parser.add_argument("--json", action="store_true")
    import_parser.set_defaults(handler=_cmd_import)
 
    render_parser = roadmap_commands.add_parser(
        "render",
        # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
        help="Render (or --check) the living ROADMAP GeneratedView set from the "
        "verified authored projection + source/.",
    )
    render_parser.add_argument("registry", nargs="?")
    render_parser.add_argument("--db", dest="db", help="Authored-projection store (default: workspace).")
    render_parser.add_argument("--source", dest="source", help="source/ tree (default: workspace).")
    # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
    render_parser.add_argument("--out", dest="out", help="Roadmap dir for ROADMAP*.md (default: workspace).")
    # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
    render_parser.add_argument("--root", dest="root", help="Root dir for CHANGELOG.md (default: workspace root).")
    render_parser.add_argument("--published", help="Stamp this 'Updated' date into the header.")
    render_parser.add_argument("--check", action="store_true", help="Audit on-disk views for drift; do not write.")
    render_parser.add_argument("--json", action="store_true")
    render_parser.set_defaults(handler=_cmd_render)
 
    # Plan 205: deliberately attached HERE, next to `render --check`, because the
    # two are read together at round close and answer DIFFERENT questions. Drift
    # compares the rendered views against the source records; reconcile compares
    # the source records against what actually landed. Conflating them is what
    # let M8.0 read `todo` for three weeks after sign-off.
    reconcile_parser = roadmap_commands.add_parser(
        "reconcile",
        # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
        help="Check roadmap CLAIMS against roadmap STATE (NOT the drift check).",
    )
    reconcile_parser.add_argument("registry", nargs="?")
    reconcile_parser.add_argument("--source", dest="source", help="source/ tree (default: workspace).")
    reconcile_parser.add_argument("--plans", dest="plans", help="Completed-plans dir (default: workspace).")
    reconcile_parser.add_argument("--repo", dest="repo", help="Repository root (default: workspace).")
    reconcile_parser.add_argument(
        "--since", default=_DISCIPLINE_LANDED,
        help="Only inspect commits newer than this git date. Defaults to the day "
             "the round-close discipline landed (%s, plan 205); pre-discipline "
             "changelog records predate the advances/status-flip rule and are "
             "grandfathered. Pass an earlier date (or an empty value) to audit "
             "the full history." % _DISCIPLINE_LANDED,
    )
    reconcile_parser.add_argument(
        "--advances-required-from", dest="advances_required_from",
        default=_ADVANCES_REQUIRED_FROM,
        # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
        help="Lowest plan id the `advances:` field is REQUIRED on; older plans "
             "missing it are advisory (the field landed 2026-08-14 with plan 205, "
             "so this defaults to %s — plans 014-204 predate the mechanism and are "
             "grandfathered)." % _ADVANCES_REQUIRED_FROM,
    )
    reconcile_parser.add_argument(
        "--fail-on-violations",
        action="store_true",
        help="Exit 1 when violations are present (default: advisory, always exit 0).",
    )
    reconcile_parser.add_argument(
        "--all", dest="show_all", action="store_true",
        help="List advisory findings too, not just violations.",
    )
    reconcile_parser.add_argument(
        "--limit", type=int, default=25, help="Max findings to list in human output (0 = all).",
    )
    reconcile_parser.add_argument(
        "--write-links", action="store_true",
        help="Persist the plan->item join into roadmap_plan_links (live store only).",
    )
    reconcile_parser.add_argument("--actor-slot")
    reconcile_parser.add_argument("--json", action="store_true")
    reconcile_parser.set_defaults(handler=_cmd_reconcile)
 
    audit_parser = roadmap_commands.add_parser("audit", help="Audit ROADMAP drift.")
    audit_parser.add_argument("registry", nargs="?")
    audit_parser.add_argument("--dir", dest="roadmap_dir")
    audit_parser.add_argument("--version")
    audit_parser.add_argument("--json", action="store_true")
    audit_parser.set_defaults(handler=_cmd_audit)
 
    bump_parser = roadmap_commands.add_parser("bump", help="Create a draft roadmap version.")
    bump_parser.add_argument("registry", nargs="?")
    bump_parser.add_argument("--from", dest="from_version")
    bump_parser.add_argument("--to", dest="to_version", required=True)
    bump_parser.add_argument("--kind", default="patch")
    bump_parser.add_argument("--actor-slot")
    bump_parser.set_defaults(handler=_cmd_bump)
 
    status_parser = roadmap_commands.add_parser("status", help="Show roadmap store status.")
    status_parser.add_argument("registry", nargs="?")
    status_parser.add_argument("--json", action="store_true")
    status_parser.set_defaults(handler=_cmd_status)
 
    # Plan 142 P1.3: attach the AuthoredSource `source`/`hash-preimage` commands
    # (implemented in registry/cli/source.py; one additive wiring line here).
    add_source_subcommands(roadmap_commands)
 
 
def _cmd_import(args: argparse.Namespace) -> int:
    location, backend, registry = _open(args.registry)
    result = ingest_corpus(
        registry,
        args.roadmap_dir or Path(location.workspace_root) / "oc_chat" / "roadmap",
        actor_slot=args.actor_slot,
        dry_run=args.dry_run,
    )
    _print_result(result.to_dict(), args.json)
    return 0 if result.ok else 1
 
 
def _cmd_render(args: argparse.Namespace) -> int:
    """Render (or --check) the living GeneratedView set (§6).
 
    Resolves the worktree-local **authored-projection** store + the ``source/``
    tree — NEVER the shared live registry (the living set renders from the
    verified projection, not the runtime store).
    """
    location = discover_registry_location(registry_path=args.registry)
    workspace = Path(location.workspace_root)
    db_path = (
        Path(args.db) if args.db
        else workspace / ".option_c" / "orchestration" / "registry_authored.sqlite"
    )
    source_dir = Path(args.source) if args.source else workspace / "oc_chat" / "roadmap" / "source"
    roadmap_dir = Path(args.out) if args.out else workspace / "oc_chat" / "roadmap"
    root_dir = Path(args.root) if args.root else workspace
    # N1 (R2): the render CLI has no envelope wrapper and the dispatcher does not
    # wrap handlers, so a raised SourceError/RegenerateError (e.g. the R2(a)
    # REGEN_UNVERIFIED_PROJECTION, the R2(b) source-recompute failures) would
    # traceback through the CLI. Wrap the render body in the §10 envelope +
    # exit_code_for (copying _cmd_source_format's SHAPE, but mapping the exit via
    # exit_code_for — SourceScan/Validation -> 1 — not its inline scan->2 choice).
    try:
        if args.check:
            results = render_check(db_path, source_dir, roadmap_dir=roadmap_dir, root_dir=root_dir)
            drift = [result.to_dict() for result in results if result.drift]
            payload = {"ok": not drift, "checked": len(results), "drift": drift}
            _print_result(payload, args.json)
            return 0 if not drift else 1
        written = write_living_set(
            db_path, source_dir, roadmap_dir=roadmap_dir, root_dir=root_dir,
            published=args.published,
        )
        payload = {"ok": True, "files": [str(path) for path in written]}
        _print_result(payload, args.json)
        return 0
    except (SourceError, RegenerateError) as error:
        _emit_render_envelope(error, args.json)
        return exit_code_for(error)
 
 
def _emit_render_envelope(error: Exception, as_json: bool) -> None:
    """Emit the §10 JSON envelope (or a compact line) for a render failure (N1)."""
    envelope = build_envelope(error)
    if as_json:
        print(json.dumps(envelope, indent=2, sort_keys=True))
    else:
        print(f"roadmap render failed [{envelope['code']}]: {envelope['message']}")
 
 
def _cmd_reconcile(args: argparse.Namespace) -> int:
    """Compare roadmap CLAIMS against roadmap STATE (Plan 205).
 
    Advisory by default: it exits 0 with violations present. That is deliberate.
    The first full-tree run reports the whole pre-existing backlog, and a check
    that must be green before it can land is a check that gets weakened until it
    is. ``--fail-on-violations`` is the opt-in for a gate that has been cleaned.
    """
    func = "_cmd_reconcile"
    location = discover_registry_location(registry_path=args.registry)
    workspace = Path(location.workspace_root)
    repo_root = Path(args.repo) if args.repo else workspace
    try:
        report = reconcile_workspace(
            repo_root,
            source_dir=args.source,
            plans_dir=args.plans,
            since=args.since,
            advances_required_from=args.advances_required_from,
        )
    except ReconcileConfigError as error:
        # Exit 2, never 0 or 1: a scope that cannot be read is neither "clean"
        # nor "violations found", and must not be mistaken for either.
        print(f"roadmap reconcile: configuration error\n{error}", file=sys.stderr)
        print(f"roadmap reconcile: configuration error\n{error}")
        return 2
    _dbg("OC_CHAT.MAIN", func, f"reconciled {repo_root}", 2)
 
    payload = report.to_dict()
    if args.write_links:
        payload["write_links"] = _write_links(args, report).to_dict()
    if args.json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        _print_reconcile(report, payload, show_all=args.show_all, limit=args.limit)
    return 1 if (args.fail_on_violations and report.would_fail) else 0
 
 
def _write_links(args: argparse.Namespace, report: object) -> JoinWriteResult:
    """Persist the plan-link join into the LIVE registry, never the projection."""
 
    _location, backend, registry = _open(args.registry)
    actor_slot = args.actor_slot or backend.active_operator_slot()
    if actor_slot is None:
        return JoinWriteResult(False, 0, (), (), "no actor slot: pass --actor-slot")
    return write_plan_links(registry, report.links, actor_slot=actor_slot)  # type: ignore[attr-defined]
 
 
def _print_reconcile(
    report: object, payload: dict[str, object], *, show_all: bool, limit: int
) -> None:
    """Render the human view, banner first so it can never read as a drift line."""
 
    print(NOT_DRIFT_BANNER)
    print()
    # Said as loudly as a violation, and ABOVE the counts: a blind spot that
    # renders as a quiet zero is the exact disease this whole plan treats.
    if not payload["changelog_class_evaluated"]:
        print("  " + "!" * 68)
        print("  !! class 1 (changelog-vs-status) was NOT EVALUATED")
        print(f"  !! {payload['changelog_not_evaluated_reason']}")
        print("  !! its 0 below means 'NOT CHECKED', not 'clean'.")
        print("  " + "!" * 68)
        print()
    print(
        f"violations: {payload['violations']}  advisories: {payload['advisories']}  "
        f"advances-none: {payload['advances_none']}  (advisory mode — exit 0)"
    )
    print(f"  {'kind':<30} {'violation':>9} {'advisory':>9}")
    for kind, split in sorted(payload["counts_by_severity"].items()):  # type: ignore[union-attr]
        print(f"  {kind:<30} {split['violation']:>9} {split['advisory']:>9}")
    checked = payload["checked"]
    print(
        f"checked: {checked['plans']} plans, {checked['changelog_records']} changelog "  # type: ignore[index]
        f"records, {checked['roadmap_records']} roadmap records"  # type: ignore[index]
    )
    findings = [
        finding for finding in payload["findings"]  # type: ignore[union-attr]
        if show_all or finding["severity"] == "violation"
    ]
    shown = findings if limit <= 0 else findings[:limit]
    print()
    for finding in shown:
        print(f"  [{finding['severity']:<9}] {finding['kind']}: {finding['detail']}")
    if len(shown) < len(findings):
        print(f"  ... {len(findings) - len(shown)} more (raise --limit, or use --json)")
    if not show_all and payload["advisories"]:
        print(f"  ({payload['advisories']} advisory finding(s) hidden — pass --all)")
    unscannable = payload["unscannable_ids"]
    if unscannable:
        print(
            f"\nnot scanned in prose (degenerate ids collide with slot handles): "
            f"{', '.join(unscannable)}"  # type: ignore[arg-type]
        )
 
 
def _cmd_audit(args: argparse.Namespace) -> int:
    location, backend, registry = _open(args.registry)
    result = audit_corpus(
        registry,
        args.roadmap_dir or Path(location.workspace_root) / "oc_chat" / "roadmap",
        version_id=args.version,
    )
    _print_result(result.to_dict(), args.json)
    return 0 if result.ok else 1
 
 
def _cmd_bump(args: argparse.Namespace) -> int:
    _location, backend, registry = _open(args.registry)
    actor_slot = args.actor_slot or backend.active_operator_slot()
    if actor_slot is None:
        print("no active operator slot found", file=sys.stderr)
        return 1
    roadmap_version_bump(
        registry,
        from_version_id=args.from_version,
        to_version_id=args.to_version,
        bump_kind=args.kind,
        actor_slot=actor_slot,
        actor_type="human_operator",
        command="roadmap bump",
        system_actor=True,
    )
    print(f"created roadmap draft version {args.to_version}")
    return 0
 
 
def _cmd_status(args: argparse.Namespace) -> int:
    _location, backend, _registry = _open(args.registry)
    with backend.connect() as conn:
        payload = {
            "ok": True,
            "user_version": conn.execute("PRAGMA user_version").fetchone()[0],
            # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
            "versions": conn.execute("SELECT COUNT(*) FROM roadmap_versions").fetchone()[0],
            # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
            "items": conn.execute("SELECT COUNT(*) FROM roadmap_items_v3").fetchone()[0],
            # OCT-LINT: disable=no_hardcoded_secrets reason="CLI SQL projections and help strings are not secrets"
            "events": conn.execute("SELECT COUNT(*) FROM roadmap_item_events").fetchone()[0],
        }
    _print_result(payload, args.json)
    return 0
 
 
def _open(registry_path: str | None) -> tuple[object, SQLiteBackend, Registry]:
    location = discover_registry_location(registry_path=registry_path)
    backend = SQLiteBackend(location.registry_path)
    return location, backend, Registry(backend)
 
 
def _print_result(payload: dict[str, object], as_json: bool) -> None:
    if as_json:
        print(json.dumps(payload, indent=2, sort_keys=True))
        return
    status = "ok" if payload.get("ok") else "failed"
    print(f"roadmap {status}")
    for key, value in payload.items():
        if key == "ok":
            continue
        print(f"  {key}: {value}")