inspect.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/cli/inspect.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_domain_inspect.py"
"""
Registry-native inspect commands.
 
Purpose
-------
Expose read-only registry inspection commands at the top-level registry CLI.
 
Responsibilities
----------------
- Wire ``show``, ``status``, and ``whoami`` commands.
- Resolve registry paths through standard discovery.
- Format read-model dataclasses for humans or JSON.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Every command in this module is read-only.
- Missing explicit plan lookups exit non-zero with a clear message.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.discovery (registry path discovery)
- oc_chat.registry.readers (typed read projections)
- oc_chat.registry (Registry facade the read projections query)
"""
from __future__ import annotations
 
import argparse
import json
 
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.readers import current_status, show_plan, whoami
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
def add_inspect_subcommands(subcommands: argparse._SubParsersAction) -> None:
    """Attach top-level read-only inspect commands."""
 
    show = subcommands.add_parser("show", help="Show a projected plan row")
    show.add_argument("plan_id")
    show.add_argument("--registry", help="Registry SQLite path.")
    show.add_argument("--json", action="store_true")
    show.set_defaults(handler=_cmd_show)
 
    status = subcommands.add_parser("status", help="Show registry status")
    status.add_argument("--registry", help="Registry SQLite path.")
    status.add_argument("--json", action="store_true")
    status.set_defaults(handler=_cmd_status)
 
    identity = subcommands.add_parser("whoami", help="Show actor metadata")
    identity.add_argument("actor_slot")
    identity.add_argument("--registry", help="Registry SQLite path.")
    identity.add_argument("--json", action="store_true")
    identity.set_defaults(handler=_cmd_whoami)
 
 
def _cmd_show(args: argparse.Namespace) -> int:
    registry = _registry(args.registry)
    view = show_plan(registry, args.plan_id)
    if view is None:
        print(f"plan not found: {args.plan_id}")
        return 1
    _print(view.__dict__, args.json)
    return 0
 
 
def _cmd_status(args: argparse.Namespace) -> int:
    registry = _registry(args.registry)
    status = current_status(registry)
    _print(status.__dict__, args.json)
    return 0
 
 
def _cmd_whoami(args: argparse.Namespace) -> int:
    registry = _registry(args.registry)
    actor = whoami(registry, args.actor_slot)
    _print(actor.__dict__, args.json)
    return 0
 
 
def _registry(registry_path: str | None) -> Registry:
    location = discover_registry_location(registry_path=registry_path)
    return Registry(SQLiteBackend(location.registry_path))
 
 
def _print(payload: dict[str, object], as_json: bool) -> None:
    if as_json:
        print(json.dumps(payload, indent=2, sort_keys=True))
        return
    for key, value in payload.items():
        print(f"{key}: {value}")