regenerate.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/regenerate.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_pipeline.py"
"""
The ``regenerate`` CLI command (Plan 142 P1.4, §13).
 
Purpose
-------
Expose ``oct agent regenerate`` / ``python -m oc_chat.registry regenerate`` — the
deterministic source -> worktree-local authored-projection store rebuild. Emits
the §10 JSON envelope and the §5.3 exit codes.
 
Responsibilities
----------------
- Resolve the source tree + the worktree-local authored-projection DB path
  (NEVER the shared live registry).
- Invoke the pipeline and map the result to the envelope + exit code.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - CLI lifecycle
    L3 - argument resolution
    L4 - envelope emission
 
Contracts
---------
- The default ``--db`` is a separate authored-projection file, never the live
  ``registry.sqlite`` — regenerate never reads/writes the shared registry.
 
Dependencies
------------
- argparse (CLI parsing)
- oc_chat.registry.regenerate.pipeline (regenerate runs the source to authored-projection store rebuild)
- oc_chat.registry.regenerate.envelope (exit_code_for maps a failed result to its exit code)
- oc_chat.registry.discovery (workspace resolution for path defaults)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import argparse
import json
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.regenerate.envelope import exit_code_for
from oc_chat.registry.regenerate.pipeline import regenerate
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_regenerate_subcommand(subcommands: argparse._SubParsersAction) -> None:
    """Attach the top-level ``regenerate`` command (§13)."""
 
    parser = subcommands.add_parser(
        "regenerate",
        help="Rebuild the worktree-local authored-projection store from source/.",
    )
    parser.add_argument("registry", nargs="?")
    parser.add_argument("--source", dest="source")
    parser.add_argument("--db", dest="db")
    parser.add_argument("--backend", default="sqlite")
    parser.add_argument("--scope", default="authored", choices=["authored", "runtime", "all"])
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument(
        "--force",
        action="store_true",
        help="Reclaim a stale regenerate lock (logged; never skips validation).",
    )
    parser.add_argument("--json", action="store_true")
    parser.set_defaults(handler=_cmd_regenerate)
 
 
def _cmd_regenerate(args: argparse.Namespace) -> int:
    func = "_cmd_regenerate"
    location = discover_registry_location(registry_path=args.registry)
    workspace = Path(location.workspace_root)
    source_dir = Path(args.source) if args.source else workspace / "oc_chat" / "roadmap" / "source"
    db_path = (
        Path(args.db)
        if args.db
        else workspace / ".option_c" / "orchestration" / "registry_authored.sqlite"
    )
    db_path.parent.mkdir(parents=True, exist_ok=True)
    if args.dry_run:
        db_path = db_path.with_name("regen_dryrun.sqlite")
    # R5: the CLI always locks at the workspace lock path (independent of --db);
    # a --dry-run serializes with real runs by design (N-a). lock.py mkdirs the
    # lock parent (H5), so a non-default --db in a fresh workspace still works.
    lock_path = workspace / ".option_c" / "orchestration" / "regenerate.lock"
    result = regenerate(
        source_dir, db_path, backend=args.backend, scope=args.scope,
        lock_path=lock_path, force=args.force,
    )
    _dbg("OC_CHAT.MAIN", func, f"regenerate ok={result.ok}", 3)
    if args.json:
        print(json.dumps(result.envelope, indent=2, sort_keys=True))
    elif result.ok:
        print(
            f"regenerate ok ({result.records} records, "
            f"logical_state_hash={result.logical_state_hash})"
        )
    else:
        print(f"regenerate failed [{result.envelope.get('code')}]: {result.envelope.get('message')}")
    return 0 if result.ok else exit_code_for(result.error)