sync.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/sync.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_sync_cli.py"
"""
Operator CLI for registry artifact rendering.
 
Purpose
-------
Expose `python -m oc_chat.registry sync` for re-rendering registry-backed
filesystem artifacts.
 
Responsibilities
----------------
- Resolve the registry and workspace root through standard discovery.
- Acquire the compatibility write lock while rendering artifacts.
- Emit one `sync.run` audit operation through the registry seam.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - sync command lifecycle events
    L3 - artifact action summaries
    L4 - argparse/dispatch tracing
 
Contracts
---------
- All registry mutations go through `Registry.execute_domain_command`.
- Drift warnings are reported to stderr before human output.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.discovery (path resolution)
- oc_chat.registry.render (artifact renderer)
- oc_chat.registry (Registry seam for the write lock and the sync.run audit)
- oc_chat.registry.backends.sqlite (SQLiteBackend over the discovered registry path)
- oc_chat.registry.domain (plan_reimport for kept local plan-status files)
- oc_chat.registry.render._common (RenderResult and plan_for_worktree lookup)
"""
from __future__ import annotations
 
import argparse
import hashlib
import json
import sys
from pathlib import Path
from typing import Mapping
 
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.domain import plan_reimport
from oc_chat.registry.render import ALL_ARTIFACTS, render_all
from oc_chat.registry.render._common import RenderResult, plan_for_worktree
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
LOCK_NAME = "legacy-compat-write"
 
 
def add_sync_subcommand(subcommands: argparse._SubParsersAction) -> None:
    """Attach the ``sync`` subcommand."""
 
    parser = subcommands.add_parser(
        "sync",
        help="Render registry-backed filesystem artifacts.",
    )
    parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    parser.add_argument(
        "--prefer-local",
        action="store_true",
        help="Keep drifted local files instead of overwriting them.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Report render actions without writing artifacts.",
    )
    parser.add_argument(
        "--artifact",
        action="append",
        choices=sorted(ALL_ARTIFACTS),
        help="Render only this artifact kind. May be passed multiple times.",
    )
    parser.add_argument("--actor-slot", help="Actor slot. Defaults to active operator.")
    parser.add_argument("--json", action="store_true")
    parser.set_defaults(handler=_cmd_sync)
 
 
def _cmd_sync(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    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
    lock = registry.acquire_lock(
        LOCK_NAME,
        actor_slot,
        ttl_seconds=300,
        metadata={"command": "sync"},
    )
    try:
        report = render_all(
            registry,
            target_root=Path(location.workspace_root),
            prefer_local=args.prefer_local,
            dry_run=args.dry_run,
            artifacts=args.artifact,
        )
        if args.prefer_local and not args.dry_run:
            _reimport_kept_plan_statuses(registry, actor_slot, report.results)
        _audit_sync(registry, actor_slot, report.to_dict())
    finally:
        lock.release()
    for result in report.results:
        if result.drift_detected:
            print(f"drift detected: {result.artifact_path}", file=sys.stderr)
        if (
            args.prefer_local
            and result.action_taken == "skipped_local_kept"
            and result.artifact_kind != "plan_status"
        ):
            print(
                f"prefer-local is unsupported for {result.artifact_kind}: "
                f"{result.artifact_path}",
                file=sys.stderr,
            )
    if args.json:
        print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
    else:
        print(format_sync_report(report.to_dict()))
    return 0 if report.ok else 1
 
 
def format_sync_report(report: Mapping[str, object]) -> str:
    """Return compact human output for sync results."""
 
    lines = ["sync ok" if report.get("ok") else "sync failed"]
    results = report.get("results", [])
    if isinstance(results, list):
        for result in results:
            if not isinstance(result, Mapping):
                continue
            lines.append(
                "  "
                + f"{result.get('artifact_kind')}: "
                + f"{result.get('action_taken')} "
                + f"{result.get('artifact_path')}"
            )
    return "\n".join(lines)
 
 
def _audit_sync(
    registry: Registry,
    actor_slot: str,
    report: Mapping[str, object],
) -> None:
    registry.execute_domain_command(
        "sync.run",
        actor_slot,
        {
            "version": 1,
            "artifacts": report.get("results", []),
        },
        actor_type="human_operator",
        command="sync",
    )
 
 
def _reimport_kept_plan_statuses(
    registry: Registry,
    actor_slot: str,
    results: tuple[RenderResult, ...],
) -> None:
    for result in results:
        if result.artifact_kind != "plan_status":
            continue
        if result.action_taken != "skipped_local_kept":
            continue
        row = plan_for_worktree(registry, result.artifact_path.parent)
        if row is None:
            continue
        previous_sha = str(row.get("body_sha256") or "")
        previous_observed = str(row.get("body_observed_at") or registry.clock.now_wall())
        new_sha = hashlib.sha256(result.artifact_path.read_bytes()).hexdigest()
        plan_reimport(
            registry,
            plan_id=str(row["plan_id"]),
            previous_body_sha256=previous_sha,
            new_body_sha256=new_sha,
            previous_body_observed_at=previous_observed,
            new_body_observed_at=registry.clock.now_wall(),
            actor_slot=actor_slot,
            actor_type="human_operator",
            command="sync --prefer-local plan-status",
        )