doctor.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/doctor.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_operator_tools_and_guards.py"
"""
Operator doctor checks for the orchestration registry.
 
Purpose
-------
Provide the health checks and small repairs owned by M11.1c: local file
permissions, stray temp-file cleanup, expired-lock sweep auditing, and
credential fingerprint consistency.
 
Responsibilities
----------------
- Inspect and optionally repair `.option_c` and `agent_secrets.json` modes.
- Remove stray `.tmp-*` files below `.option_c/orchestration`.
- Sweep expired lock rows and emit a `doctor.repair` audit operation.
- Compare on-disk credential fingerprints with `agent_credentials`.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Audit rows go through `Registry.execute_domain_command`.
- Secret values are never returned in structured output.
- Windows permission inspection is a graceful v0.2.x skip.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.api (Registry facade)
- oc_chat.registry.backends.sqlite (SQLite backend)
- oc_chat.registry.domain (plan_reimport for the trust-local snapshot repair)
- oc_chat.registry.render (render_all for snapshot drift detection and trust-registry repair)
- oc_chat.registry.render._common (RenderResult and plan_for_worktree lookup)
"""
from __future__ import annotations
 
from dataclasses import dataclass
import base64
import hashlib
import json
import os
from pathlib import Path
import stat
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.backends.sqlite import SQLiteBackend
from oc_chat.registry.domain import plan_reimport
from oc_chat.registry.render import render_all
from oc_chat.registry.render._common import RenderResult, plan_for_worktree
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
OPTION_C_DIR = ".option_c"
ORCHESTRATION_DIR = "orchestration"
# OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
SECRETS_FILENAME = "agent_secrets.json"
EXPECTED_OPTION_C_MODE = 0o700
EXPECTED_SECRETS_MODE = 0o600
 
 
@dataclass(frozen=True)
class DoctorOptions:
    """Runtime flags for `doctor`."""
 
    fix_perms: bool = False
    skip_perms: bool = False
    sweep_temps: bool = True
    lock_sweep: bool = True
    credential_check: bool = True
    resolve_snapshot_conflict: bool = False
    yes_trust_registry: bool = False
    yes_trust_local: bool = False
 
 
def run_doctor(
    registry: Registry,
    backend: SQLiteBackend,
    workspace_root: Path,
    options: DoctorOptions,
) -> dict[str, object]:
    """Run all enabled doctor checks and return a JSON-safe report."""
 
    option_c = workspace_root / OPTION_C_DIR
    permissions = check_permissions(
        option_c,
        fix=options.fix_perms,
        skip=options.skip_perms,
    )
    temp_files = (
        sweep_stray_temp_files(option_c / ORCHESTRATION_DIR)
        if options.sweep_temps
        else _skipped("stray temp-file sweep skipped")
    )
    locks = (
        sweep_expired_locks(registry, backend)
        if options.lock_sweep
        else _skipped("lock-expiry sweep skipped")
    )
    credentials = (
        check_credential_consistency(backend, option_c / SECRETS_FILENAME)
        if options.credential_check
        else _skipped("credential consistency check skipped")
    )
    round_state = check_round_state(backend)
    sections = [permissions, temp_files, locks, credentials, round_state]
    report: dict[str, object] = {
        "ok": True,
        "workspace_root": str(workspace_root),
        "registry_path": str(backend.path),
        "permissions": permissions,
        "temp_files": temp_files,
        "locks": locks,
        "credentials": credentials,
        "round_state": round_state,
    }
    auth_events = auth_event_summary(backend)
    if auth_events["failures"]:
        report["auth_events"] = auth_events
    if options.resolve_snapshot_conflict:
        snapshots = check_snapshot_conflicts(
            registry,
            backend,
            workspace_root,
            options,
        )
        sections.append(snapshots)
        report["snapshot_conflicts"] = snapshots
    warnings = _warnings_from(*sections)
    report["ok"] = not warnings
    report["warnings"] = warnings
    return report
 
 
def check_permissions(option_c: Path, *, fix: bool, skip: bool) -> dict[str, object]:
    """Check POSIX modes for `.option_c` and `agent_secrets.json`."""
 
    if skip:
        return _skipped("perms check skipped by flag")
    if os.name == "nt":
        return _skipped("perms check skipped on this platform")
    targets = [
        (option_c, EXPECTED_OPTION_C_MODE),
        (option_c / SECRETS_FILENAME, EXPECTED_SECRETS_MODE),
    ]
    checked: list[dict[str, object]] = []
    fixed: list[str] = []
    warnings: list[str] = []
    for path, expected_mode in targets:
        if not path.exists():
            warnings.append(f"{path} is missing")
            continue
        actual_mode = stat.S_IMODE(path.stat().st_mode)
        checked.append(
            {
                "path": str(path),
                "actual": oct(actual_mode),
                "expected": oct(expected_mode),
            }
        )
        if actual_mode == expected_mode:
            continue
        message = (
            f"{path} mode is {oct(actual_mode)}; expected {oct(expected_mode)}"
        )
        if fix:
            path.chmod(expected_mode)
            fixed.append(str(path))
        else:
            warnings.append(message)
    status = "ok"
    if warnings:
        status = "warn"
    elif fixed:
        status = "fixed"
    return {
        "status": status,
        "checked": checked,
        "fixed": fixed,
        "warnings": warnings,
        "messages": [],
    }
 
 
def sweep_stray_temp_files(orchestration_dir: Path) -> dict[str, object]:
    """Remove stray `.tmp-*` files below the orchestration directory."""
 
    removed: list[str] = []
    warnings: list[str] = []
    if not orchestration_dir.exists():
        return {
            "status": "ok",
            "removed": removed,
            "warnings": warnings,
            "messages": [],
        }
    for path in sorted(orchestration_dir.rglob(".tmp-*")):
        if not path.is_file() and not path.is_symlink():
            warnings.append(f"skipped non-file temp path: {path}")
            continue
        try:
            path.unlink()
            removed.append(str(path))
        except OSError as exc:
            warnings.append(f"failed to remove {path}: {exc}")
    return {
        "status": "warn" if warnings else "ok",
        "removed": removed,
        "warnings": warnings,
        "messages": [],
    }
 
 
def sweep_expired_locks(
    registry: Registry,
    backend: SQLiteBackend,
) -> dict[str, object]:
    """Sweep expired locks and audit the repair when rows were removed."""
 
    swept = backend.startup_sweep_with_names()
    result: dict[str, object] = {
        "status": "ok",
        "swept": swept,
        "count": len(swept),
        "audit_operation_id": None,
        "warnings": [],
        "messages": [],
    }
    if not swept:
        return result
    operator_slot = backend.active_operator_slot()
    if operator_slot is None:
        result["status"] = "warn"
        result["warnings"] = ["no active operator slot available for doctor audit"]
        return result
    row = registry.execute_domain_command(
        "doctor.repair",
        operator_slot,
        {
            "version": 1,
            "repair_kind": "lock_expiry_sweep",
            "items_repaired": swept,
            "items_warned": [],
        },
        actor_type="human_operator",
    )
    result["audit_operation_id"] = row.id
    return result
 
 
def check_credential_consistency(
    backend: SQLiteBackend,
    secrets_path: Path,
) -> dict[str, object]:
    """Compare non-revoked DB credential fingerprints with the secrets file."""
 
    credentials = [
        row for row in backend.credential_fingerprints() if row.get("revoked_at") is None
    ]
    if not secrets_path.is_file():
        return {
            "status": "missing",
            "checked": 0,
            "warnings": [f"{secrets_path} is missing"],
            "messages": [],
        }
    try:
        data = json.loads(secrets_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        return {
            "status": "unreadable",
            "checked": 0,
            "warnings": [f"{secrets_path} could not be read as JSON: {exc}"],
            "messages": [],
        }
    warnings: list[str] = []
    checked = 0
    for credential in credentials:
        checked += 1
        slot = str(credential["slot"])
        phase = str(credential["rotation_phase"])
        block = _secret_block(data, slot, phase)
        if block is None:
            warnings.append(f"missing secret block for {slot}/{phase}")
            continue
        expected_key = str(credential["hmac_key_id"])
        actual_key = block.get("hmac_key_id")
        if isinstance(actual_key, str) and actual_key != expected_key:
            warnings.append(f"hmac_key_id mismatch for {slot}/{phase}")
        actual_fingerprint = _fingerprint_for_block(block)
        if actual_fingerprint != credential["secret_fingerprint"]:
            warnings.append(f"secret_fingerprint mismatch for {slot}/{phase}")
    return {
        "status": "warn" if warnings else "ok",
        "checked": checked,
        "warnings": warnings,
        "messages": [],
    }
 
 
def check_round_state(backend: SQLiteBackend) -> dict[str, object]:
    """Audit round/round_job invariants (read-only).
 
    Flags the OI-324 invariant (a ``closed`` round must be ``clean``) and any
    round-job left in ``failed`` status that needs an operator resume or abort.
    """
 
    func = "check_round_state"
    warnings: list[str] = []
    with backend.connect() as conn:
        bad_close = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
            "SELECT round_no, cleanup_status FROM rounds "
            "WHERE status = 'closed' AND cleanup_status != 'clean'"
        ).fetchall()
        failed_jobs = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
            "SELECT job_id, round_no, current_step FROM round_jobs "
            "WHERE status = 'failed' ORDER BY round_no"
        ).fetchall()
        in_flight = int(
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
                "SELECT COUNT(*) FROM round_jobs WHERE status IN ('running', 'paused')"
            ).fetchone()[0]
        )
    for row in bad_close:
        warnings.append(
            f"round {row['round_no']} is closed but cleanup_status="
            f"{row['cleanup_status']!r} (OI-324 invariant)"
        )
    for row in failed_jobs:
        warnings.append(
            f"round {row['round_no']} job {str(row['job_id'])[:8]} failed at "
            f"{row['current_step'] or '-'}; resume or abort required"
        )
    _dbg("OC_CHAT.MAIN", func, f"warnings={len(warnings)} in_flight={in_flight}", 2)
    return {
        "status": "warn" if warnings else "ok",
        "checked": len(bad_close) + len(failed_jobs),
        "in_flight": in_flight,
        "failed_jobs": len(failed_jobs),
        "warnings": warnings,
        "messages": [],
    }
 
 
def auth_event_summary(backend: SQLiteBackend) -> dict[str, object]:
    """Return auth-event counts for doctor output."""
 
    with backend.connect() as conn:
        total = int(conn.execute("SELECT COUNT(*) FROM auth_events").fetchone()[0])
        failures = int(
            conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
                "SELECT COUNT(*) FROM auth_events WHERE auth_ok = 0"
            ).fetchone()[0]
        )
    return {
        "status": "warn" if failures else "ok",
        "checked": total,
        "failures": failures,
        "warnings": [],
        "messages": [f"failed_auth_events={failures}"] if failures else [],
    }
 
 
def check_snapshot_conflicts(
    registry: Registry,
    backend: SQLiteBackend,
    workspace_root: Path,
    options: DoctorOptions,
) -> dict[str, object]:
    """Detect and optionally resolve rendered artifact drift."""
 
    dry_report = render_all(registry, target_root=workspace_root, dry_run=True)
    drifted = [result for result in dry_report.results if result.drift_detected]
    if not drifted:
        return {
            "status": "ok",
            "checked": len(dry_report.results),
            "resolved": [],
            "warnings": [],
            "messages": [],
        }
    warnings = [f"snapshot drift: {result.artifact_path}" for result in drifted]
    if not options.resolve_snapshot_conflict:
        return {
            "status": "warn",
            "checked": len(dry_report.results),
            "resolved": [],
            "warnings": warnings,
            "messages": [],
        }
    actor_slot = backend.active_operator_slot()
    if actor_slot is None:
        return {
            "status": "warn",
            "checked": len(dry_report.results),
            "resolved": [],
            "warnings": ["no active operator slot available for snapshot repair"],
            "messages": [],
        }
    mode = _resolution_mode(options)
    if mode == "skip":
        return {
            "status": "warn",
            "checked": len(dry_report.results),
            "resolved": [],
            "warnings": warnings,
            "messages": ["snapshot repair skipped"],
        }
    lock = registry.acquire_lock(
        "legacy-compat-write",
        actor_slot,
        ttl_seconds=300,
        metadata={"command": "doctor --resolve-snapshot-conflict"},
    )
    try:
        if mode == "trust_registry":
            render_report = render_all(registry, target_root=workspace_root)
            resolved = [
                _audit_snapshot_repair(registry, actor_slot, result, mode)
                for result in render_report.results
                if result.drift_detected
            ]
        else:
            resolved = [
                _trust_local(registry, actor_slot, result)
                for result in drifted
            ]
    finally:
        lock.release()
    unresolved = [item for item in resolved if item.get("status") != "resolved"]
    return {
        "status": "fixed" if not unresolved else "warn",
        "checked": len(dry_report.results),
        "resolved": resolved,
        "warnings": [] if not unresolved else [str(item["message"]) for item in unresolved],
        "messages": [f"resolved {len(resolved) - len(unresolved)} snapshot conflict(s)"],
    }
 
 
def format_doctor_report(report: Mapping[str, object]) -> str:
    """Return a compact human-readable doctor summary."""
 
    lines = [
        f"registry: {report['registry_path']}",
        f"workspace: {report['workspace_root']}",
        _section_line("permissions", report["permissions"]),
        _section_line("temp files", report["temp_files"]),
        _section_line("locks", report["locks"]),
        _section_line("credentials", report["credentials"]),
        _section_line("round state", report["round_state"]),
    ]
    if "auth_events" in report:
        lines.append(_section_line("auth events", report["auth_events"]))
    if "snapshot_conflicts" in report:
        lines.append(_section_line("snapshot conflicts", report["snapshot_conflicts"]))
    warnings = report.get("warnings", [])
    if isinstance(warnings, list) and warnings:
        lines.append("warnings:")
        lines.extend(f"  - {warning}" for warning in warnings)
    else:
        lines.append("doctor ok")
    return "\n".join(lines)
 
 
def _section_line(label: str, section: object) -> str:
    if not isinstance(section, Mapping):
        return f"{label}: unknown"
    status = section.get("status", "unknown")
    details: list[str] = []
    if "removed" in section:
        details.append(f"removed={len(section.get('removed', []))}")
    if "count" in section:
        details.append(f"swept={section.get('count')}")
    if "checked" in section:
        details.append(f"checked={section.get('checked')}")
    suffix = "" if not details else f" ({', '.join(details)})"
    messages = section.get("messages", [])
    message = ""
    if isinstance(messages, list) and messages:
        message = f" - {messages[0]}"
    return f"{label}: {status}{suffix}{message}"
 
 
def _secret_block(
    data: object,
    slot: str,
    phase: str,
) -> Mapping[str, object] | None:
    if not isinstance(data, Mapping):
        return None
    slot_block = data.get(slot)
    if not isinstance(slot_block, Mapping):
        return None
    phase_block = slot_block.get(phase)
    if not isinstance(phase_block, Mapping):
        return None
    return phase_block
 
 
def _fingerprint_for_block(block: Mapping[str, object]) -> object:
    secret_b64 = block.get("secret_b64")
    if isinstance(secret_b64, str):
        try:
            raw = base64.b64decode(secret_b64.encode("ascii"), validate=True)
        except (ValueError, UnicodeEncodeError):
            raw = b""
        if raw:
            return hashlib.sha256(raw).hexdigest()
    return block.get("secret_fingerprint")
 
 
def _warnings_from(*sections: Mapping[str, object]) -> list[str]:
    warnings: list[str] = []
    for section in sections:
        section_warnings = section.get("warnings", [])
        if not isinstance(section_warnings, list):
            continue
        warnings.extend(str(warning) for warning in section_warnings)
    return warnings
 
 
def _skipped(message: str) -> dict[str, object]:
    return {
        "status": "skipped",
        "warnings": [],
        "messages": [message],
    }
 
 
def _resolution_mode(options: DoctorOptions) -> str:
    if options.yes_trust_registry and options.yes_trust_local:
        raise ValueError("choose only one of --yes-trust-registry/--yes-trust-local")
    if options.yes_trust_registry:
        return "trust_registry"
    if options.yes_trust_local:
        return "trust_local"
    answer = input("Resolve snapshot drift? [r=trust-registry, l=trust-local, s=skip] ")
    if answer.strip().lower() in {"r", "registry", "trust-registry"}:
        return "trust_registry"
    if answer.strip().lower() in {"l", "local", "trust-local"}:
        return "trust_local"
    return "skip"
 
 
def _audit_snapshot_repair(
    registry: Registry,
    actor_slot: str,
    result: RenderResult,
    action: str,
) -> dict[str, object]:
    row = registry.execute_domain_command(
        "doctor.repair",
        actor_slot,
        {
            "version": 1,
            "repair_kind": "snapshot_conflict",
            "action": action,
            "artifact_kind": result.artifact_kind,
            "artifact_path": str(result.artifact_path),
            "content_sha256": result.content_sha256,
        },
        actor_type="human_operator",
        command="doctor --resolve-snapshot-conflict",
    )
    return {
        "status": "resolved",
        "action": action,
        "artifact_path": str(result.artifact_path),
        "audit_operation_id": row.id,
    }
 
 
def _trust_local(
    registry: Registry,
    actor_slot: str,
    result: RenderResult,
) -> dict[str, object]:
    if result.artifact_kind != "plan_status":
        return {
            "status": "unsupported",
            "action": "trust_local",
            "artifact_path": str(result.artifact_path),
            # OCT-LINT: disable=no_hardcoded_secrets reason="Secret-related names are local file and schema fields, not secret values"
            "message": "trust-local is only available for .PLAN_STATUS.md",
        }
    row = plan_for_worktree(registry, result.artifact_path.parent)
    if row is None:
        return {
            "status": "unsupported",
            "action": "trust_local",
            "artifact_path": str(result.artifact_path),
            "message": "no registry plan is assigned to this worktree",
        }
    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_row = 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="doctor trust-local plan-status",
    )
    repair = _audit_snapshot_repair(registry, actor_slot, result, "trust_local")
    repair["plan_reimport_operation_id"] = plan_row.id
    return repair