agent.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/agent.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_storage_foundation.py"
"""
Argparse command surface for the registry operator tools.
 
Purpose
-------
Expose registry operator commands through
``python -m oc_chat.registry`` before the future ``oct agent`` rename.
 
Responsibilities
----------------
- Parse registry bootstrap command arguments.
- Resolve the target registry and workspace metadata.
- Generate seed identifiers and the initial operator credential fingerprint.
- Call :class:`SQLiteBackend` to apply schema and seed singleton rows.
- Dispatch doctor and db integrity-check commands.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- `init` creates the SQLite registry only; doctor/db commands inspect or
  repair existing registries.
- Raw secret bytes are used only to derive the stored fingerprint.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (registry initialization backend)
- oc_chat.registry.cli.db (database maintenance command wiring)
- oc_chat.registry.cli.inspect (read-only inspect command wiring)
- oc_chat.registry.cli.migrate (legacy migration command wiring)
- oc_chat.registry.cli.plan (plan domain command wiring)
- oc_chat.registry.cli.queue (queue domain command wiring)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.core._version (resolve_version stamps the app version into the init seed)
- oc_chat.registry (Registry facade for the doctor, rotation, handover and auth commands)
- oc_chat.registry.auth.bootstrap (bootstrap-machine install and export-secrets bundle)
- oc_chat.registry.auth.handover (initiate and accept orchestrator handover)
- oc_chat.registry.auth.rotation (start and finalise credential rotation)
- oc_chat.registry.auth.secrets_store (writes the operator's initial secret block at init)
- oc_chat.registry.cli.checkpoint (checkpoint export and import command wiring)
- oc_chat.registry.cli.regenerate (top-level regenerate command wiring)
- oc_chat.registry.cli.roadmap (roadmap command group wiring)
- oc_chat.registry.cli.round (round-close command wiring)
- oc_chat.registry.cli.sync (sync command wiring for registry-backed artefacts)
- oc_chat.registry.doctor (doctor checks and report formatting)
"""
from __future__ import annotations
 
import argparse
import hashlib
import json
from pathlib import Path
import secrets
import socket
import sys
import uuid
 
from oc_diagnostics import _dbg
 
from oc_chat.core._version import resolve_version
from oc_chat.registry import Registry
from oc_chat.registry.auth.bootstrap import bootstrap_machine, export_bundle
from oc_chat.registry.auth.handover import accept_handover, initiate_handover
from oc_chat.registry.auth.rotation import finalize_rotation, rotate_secret
from oc_chat.registry.auth.secrets_store import block_from_secret, write_secrets
from oc_chat.registry.backends.sqlite import RegistrySeed, SQLiteBackend
from oc_chat.registry.cli.checkpoint import add_checkpoint_subcommands
from oc_chat.registry.cli.db import add_db_subcommands
from oc_chat.registry.cli.inspect import add_inspect_subcommands
from oc_chat.registry.cli.migrate import add_migrate_subcommands
from oc_chat.registry.cli.plan import add_plan_subcommands
from oc_chat.registry.cli.queue import add_queue_subcommands
from oc_chat.registry.cli.regenerate import add_regenerate_subcommand
from oc_chat.registry.cli.roadmap import add_roadmap_subcommands
from oc_chat.registry.cli.round import add_round_subcommands
from oc_chat.registry.cli.sync import add_sync_subcommand
from oc_chat.registry.discovery import discover_registry_location
from oc_chat.registry.doctor import (
    DoctorOptions,
    format_doctor_report,
    run_doctor,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    handler = getattr(args, "handler", None)
    if handler is None:
        parser.print_help()
        return 1
    return int(handler(args))
 
 
def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="python -m oc_chat.registry",
        description="Orchestration registry operator commands.",
    )
    subcommands = parser.add_subparsers(dest="command")
 
    init_parser = subcommands.add_parser(
        "init",
        help="Create and seed a registry SQLite database.",
    )
    init_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    init_parser.add_argument(
        "--workspace-root",
        help="Workspace root to record in the initial workspaces row.",
    )
    init_parser.add_argument(
        "--operator-slot",
        help="Initial human operator slot. Defaults to the local hostname.",
    )
    init_parser.add_argument(
        "--quiet",
        action="store_true",
        help="Suppress success output.",
    )
    init_parser.set_defaults(handler=_cmd_init)
 
    doctor_parser = subcommands.add_parser(
        "doctor",
        help="Run registry health checks and safe repairs.",
    )
    doctor_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    doctor_parser.add_argument(
        "--fix-perms",
        action="store_true",
        help="Repair .option_c and agent_secrets.json modes where supported.",
    )
    doctor_parser.add_argument(
        "--skip-perms",
        action="store_true",
        help="Skip local filesystem permission inspection.",
    )
    doctor_parser.add_argument(
        "--no-sweep-temps",
        action="store_true",
        help="Do not remove stray .tmp-* files below .option_c/orchestration.",
    )
    doctor_parser.add_argument(
        "--no-lock-sweep",
        action="store_true",
        help="Do not sweep expired volatile lock rows.",
    )
    doctor_parser.add_argument(
        "--no-credential-check",
        action="store_true",
        help="Do not compare agent_secrets.json fingerprints to the registry.",
    )
    doctor_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the structured doctor report as JSON.",
    )
    doctor_parser.add_argument(
        "--resolve-snapshot-conflict",
        action="store_true",
        help="Resolve drifted rendered artifacts.",
    )
    doctor_parser.add_argument(
        "--yes-trust-registry",
        action="store_true",
        help="Resolve snapshot drift by overwriting from the registry.",
    )
    doctor_parser.add_argument(
        "--yes-trust-local",
        action="store_true",
        help="Resolve snapshot drift by keeping supported local edits.",
    )
    doctor_parser.set_defaults(handler=_cmd_doctor)
 
    rotate_parser = subcommands.add_parser(
        "rotate-secrets",
        help="Start or finalize HMAC credential rotation.",
    )
    rotate_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    rotate_scope = rotate_parser.add_mutually_exclusive_group()
    rotate_scope.add_argument("--slot", help="Slot to rotate.")
    rotate_scope.add_argument(
        "--all",
        action="store_true",
        help="Rotate every active slot.",
    )
    rotate_parser.add_argument(
        "--grace-seconds",
        type=int,
        help="Rotation grace window override.",
    )
    rotate_parser.add_argument(
        "--finalize",
        action="store_true",
        help="Finalize an in-flight rotation instead of starting one.",
    )
    rotate_parser.set_defaults(handler=_cmd_rotate_secrets)
 
    handover_parser = subcommands.add_parser(
        "handover",
        help="Initiate or accept orchestrator handover.",
    )
    handover_parser.add_argument(
        "registry",
        nargs="?",
        help="Registry SQLite path. Defaults to the discovery algorithm.",
    )
    handover_action = handover_parser.add_mutually_exclusive_group(required=True)
    handover_action.add_argument("--to", help="Incoming orchestrator slot.")
    handover_action.add_argument(
        "--accept",
        type=int,
        metavar="HANDOVER_ID",
        help="Accept a handover-start operation id.",
    )
    handover_parser.add_argument("--from-slot", help="Outgoing orchestrator slot.")
    handover_parser.add_argument("--reason", default="operator handover")
    handover_parser.set_defaults(handler=_cmd_handover)
 
    auth_parser = subcommands.add_parser(
        "auth",
        help="Actor-auth inspection and bootstrap commands.",
    )
    add_auth_subcommands(auth_parser)
 
    db_parser = subcommands.add_parser(
        "db",
        help="Database maintenance and inspection commands.",
    )
    add_db_subcommands(db_parser)
 
    migrate_parser = subcommands.add_parser(
        "migrate",
        help="Legacy registry migration commands.",
    )
    add_migrate_subcommands(migrate_parser)
 
    queue_parser = subcommands.add_parser(
        "queue",
        help="Registry-native queue commands.",
    )
    add_queue_subcommands(queue_parser)
 
    plan_parser = subcommands.add_parser(
        "plan",
        help="Registry-native plan commands.",
    )
    add_plan_subcommands(plan_parser)
 
    round_parser = subcommands.add_parser(
        "round",
        help="Registry-native round-close commands.",
    )
    add_round_subcommands(round_parser)
 
    add_sync_subcommand(subcommands)
    add_roadmap_subcommands(subcommands)
    add_regenerate_subcommand(subcommands)  # Plan 142 P1.4 (top-level, §13)
 
    checkpoint_parser = subcommands.add_parser(
        "checkpoint",
        help="Export and import registry JSON checkpoints.",
    )
    add_checkpoint_subcommands(checkpoint_parser)
 
    add_inspect_subcommands(subcommands)
    return parser
 
 
def _cmd_init(args: argparse.Namespace) -> int:
    location = discover_registry_location(
        registry_path=args.registry,
        workspace_root=args.workspace_root,
    )
    raw_secret = secrets.token_bytes(32)
    operator_slot = args.operator_slot or _default_operator_slot()
    seed = RegistrySeed(
        workspace_id=str(uuid.uuid4()),
        registry_id=str(uuid.uuid4()),
        workspace_root=str(location.workspace_root),
        git_common_dir=str(location.git_common_dir),
        registry_path=str(location.registry_path),
        checkpoint_path=str(location.checkpoint_path),
        operator_slot=operator_slot,
        hmac_key_id=str(uuid.uuid4()),
        secret_fingerprint=hashlib.sha256(raw_secret).hexdigest(),
        app_version=resolve_version(),
    )
    backend = SQLiteBackend(location.registry_path)
    backend.initialize(seed)
    secret_block = block_from_secret(
        slot=operator_slot,
        phase="active",
        hmac_key_id=seed.hmac_key_id,
        raw_secret=raw_secret,
        created_at=backend.clock.now_wall(),
    )
    write_secrets({operator_slot: {"active": secret_block}}, _secrets_path(location))
    backend.startup_sweep()
    if not args.quiet:
        print(f"initialised registry: {location.registry_path}")
        print(f"operator slot: {operator_slot}")
    return 0
 
 
def _cmd_doctor(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    report = run_doctor(
        registry,
        backend,
        location.workspace_root,
        DoctorOptions(
            fix_perms=args.fix_perms,
            skip_perms=args.skip_perms,
            sweep_temps=not args.no_sweep_temps,
            lock_sweep=not args.no_lock_sweep,
            credential_check=not args.no_credential_check,
            resolve_snapshot_conflict=args.resolve_snapshot_conflict,
            yes_trust_registry=args.yes_trust_registry,
            yes_trust_local=args.yes_trust_local,
        ),
    )
    if args.json:
        print(json.dumps(report, indent=2, sort_keys=True))
    else:
        print(format_doctor_report(report))
    return 0
 
 
def add_auth_subcommands(parser: argparse.ArgumentParser) -> None:
    """Register actor-auth subcommands."""
 
    subcommands = parser.add_subparsers(dest="auth_command")
 
    show = subcommands.add_parser("show", help="Show credential state.")
    show.add_argument("registry", nargs="?")
    show.add_argument("--slot")
    show.set_defaults(handler=_cmd_auth_show)
 
    verify = subcommands.add_parser("verify", help="Verify recent auth metadata.")
    verify.add_argument("registry", nargs="?")
    verify.add_argument("--slot")
    verify.add_argument("--all", action="store_true")
    verify.set_defaults(handler=_cmd_auth_verify)
 
    export = subcommands.add_parser("export-secrets", help="Export local secrets.")
    export.add_argument("registry", nargs="?")
    export.add_argument("--slot")
    export.add_argument("--out", required=True)
    export.add_argument("--encrypt-command")
    export.set_defaults(handler=_cmd_auth_export)
 
    bootstrap = subcommands.add_parser(
        "bootstrap-machine",
        help="Install local secrets from a bundle.",
    )
    bootstrap.add_argument("registry", nargs="?")
    bootstrap.add_argument("--secrets", required=True)
    bootstrap.add_argument("--decrypt-command")
    bootstrap.set_defaults(handler=_cmd_auth_bootstrap)
 
 
def _cmd_rotate_secrets(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    actor_slot = backend.active_operator_slot()
    if actor_slot is None:
        print("no active operator slot", file=sys.stderr)
        return 1
    slots = _credential_slots(backend) if args.all else [args.slot or actor_slot]
    failures = 0
    for slot in slots:
        try:
            if args.finalize:
                result = finalize_rotation(
                    registry,
                    slot=slot,
                    actor_slot=actor_slot,
                    secrets_path=_secrets_path(location),
                )
                print(
                    f"finalized {slot}: {result.old_key_id[:8]} -> "
                    f"{result.new_key_id[:8]}"
                )
            else:
                result = rotate_secret(
                    registry,
                    slot=slot,
                    actor_slot=actor_slot,
                    grace_seconds=args.grace_seconds,
                    secrets_path=_secrets_path(location),
                )
                print(
                    f"rotating {slot}: {result.old_key_id[:8]} -> "
                    f"{result.new_key_id[:8]}"
                )
        except Exception as exc:
            failures += 1
            print(f"{slot}: {exc}", file=sys.stderr)
    if failures and failures != len(slots):
        return 2
    return 1 if failures else 0
 
 
def _cmd_handover(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    if args.accept is not None:
        actor_slot = args.from_slot or _current_orchestrator_or_operator(backend)
        row = accept_handover(
            registry,
            handover_id=args.accept,
            actor_slot=actor_slot,
            actor_type=_actor_type_for_slot(backend, actor_slot),
        )
        print(f"accepted handover: operation {row.id}")
        return 0
    from_slot = args.from_slot or _current_orchestrator_or_operator(backend)
    row = initiate_handover(
        registry,
        from_slot=from_slot,
        to_slot=args.to,
        reason=args.reason,
        actor_slot=from_slot,
        actor_type=_actor_type_for_slot(backend, from_slot),
    )
    print(f"started handover: operation {row.id}")
    return 0
 
 
def _cmd_auth_show(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    for row in _credential_rows(backend, slot=args.slot):
        fingerprint = str(row["secret_fingerprint"])
        print(
            f"{row['slot']} {row['rotation_phase']} "
            f"key={str(row['hmac_key_id'])[:8]} "
            f"fingerprint={fingerprint[:8]} "
            f"expires_at={row['expires_at'] or '-'} "
            f"revoked_at={row['revoked_at'] or '-'}"
        )
    return 0
 
 
def _cmd_auth_verify(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    rows = _operation_auth_rows(backend, slot=args.slot)
    failures = [
        row for row in rows
        if row["actor_slot"] != "system" and row["hmac_key_id"] is None
    ]
    print(f"checked={len(rows)} failures={len(failures)}")
    return 1 if failures else 0
 
 
def _cmd_auth_export(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    result = export_bundle(
        registry,
        out=args.out,
        secrets_path=_secrets_path(location),
        slot=args.slot,
        encrypt_command=args.encrypt_command,
    )
    print(f"exported {result.slot_count} slot(s): {result.path}")
    return 0
 
 
def _cmd_auth_bootstrap(args: argparse.Namespace) -> int:
    location = discover_registry_location(registry_path=args.registry)
    backend = SQLiteBackend(location.registry_path)
    registry = Registry(backend)
    result = bootstrap_machine(
        registry,
        bundle_path=args.secrets,
        secrets_path=_secrets_path(location),
        decrypt_command=args.decrypt_command,
    )
    print(f"installed {result.slot_count} slot(s): {result.path}")
    return 0
 
 
def _default_operator_slot() -> str:
    hostname = socket.gethostname().lower()
    safe = "".join(ch if ch.isalnum() or ch in "-_" else "-" for ch in hostname)
    safe = "-".join(part for part in safe.split("-") if part) or "local"
    return f"human-operator-{safe[:48]}"
 
 
def _secrets_path(location: object) -> Path:
    return Path(location.workspace_root) / ".option_c" / "agent_secrets.json"
 
 
def _credential_slots(backend: SQLiteBackend) -> list[str]:
    with backend.connect() as conn:
        return [
            str(row["slot"])
            for row in conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help, SQL projections, and JSON field names are not secrets"
                "SELECT DISTINCT slot FROM agent_credentials "
                "WHERE revoked_at IS NULL ORDER BY slot"
            ).fetchall()
        ]
 
 
def _credential_rows(
    backend: SQLiteBackend,
    *,
    slot: str | None = None,
) -> list[object]:
    with backend.connect() as conn:
        if slot is None:
            return conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help, SQL projections, and JSON field names are not secrets"
                "SELECT hmac_key_id, slot, secret_fingerprint, rotation_phase, "
                "expires_at, revoked_at FROM agent_credentials "
                "ORDER BY slot, rotation_phase"
            ).fetchall()
        return conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help, SQL projections, and JSON field names are not secrets"
            "SELECT hmac_key_id, slot, secret_fingerprint, rotation_phase, "
            "expires_at, revoked_at FROM agent_credentials "
            "WHERE slot = ? ORDER BY rotation_phase",
            (slot,),
        ).fetchall()
 
 
def _operation_auth_rows(
    backend: SQLiteBackend,
    *,
    slot: str | None = None,
) -> list[object]:
    with backend.connect() as conn:
        if slot is None:
            return conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help, SQL projections, and JSON field names are not secrets"
                "SELECT actor_slot, hmac_key_id FROM operation_log ORDER BY id"
            ).fetchall()
        return conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="CLI help, SQL projections, and JSON field names are not secrets"
            "SELECT actor_slot, hmac_key_id FROM operation_log "
            "WHERE actor_slot = ? ORDER BY id",
            (slot,),
        ).fetchall()
 
 
def _current_orchestrator_or_operator(backend: SQLiteBackend) -> str:
    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]
 
 
if __name__ == "__main__":
    sys.exit(main())