operator

#!/usr/bin/env python3
"""
Operator Control Plane CLI (MVP) - Isolated Witness Edition
A local, file-backed control plane for coordinating multiple AI coding harnesses.
"""

from __future__ import annotations

import argparse
import datetime
import fcntl
import glob
import hashlib
import json
import os
import pwd
import re
import shutil
import socket
import sqlite3
import stat
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

import authority_client
import authority_projection
import crystal_parse

LEDGER_DB_NAME = "ledger.sqlite3"
LEDGER_HASH_FORMAT = "operator-ledger-event-v1"
LEDGER_SCHEMA_VERSION = 1
VALID_TASK_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
VALID_CLAIM_ID = re.compile(r"^claim-[0-9]{4,}$")
VALID_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$")
TRUST_RECORD_DIRS = {
    "tasks": "task",
    "claims": "claim",
    "evidence": "evidence",
    "handoffs": "handoff",
    "usage": "usage",
}
MUTATING_COMMANDS = {
    "init",
    "task-create",
    "claim-add",
    "evidence-attach",
    "crystal-attach",
    "crystal-import",
    "crystal-bridge",
    "usage-add",
    "usage-import",
    "usage-annotate",
    "handoff-add",
    "session-start",
    "session-end",
}

EVIDENCE_TYPES = [
    "run_log",
    "manifest",
    "database_query",
    "test_output",
    "git_commit",
    "screenshot",
    "transcript",
    "paper_section",
    "external_doc",
    "session_crystal",
]


class LedgerWriteError(RuntimeError):
    pass


class EvidenceFingerprintError(RuntimeError):
    pass


class IdentityPolicyError(RuntimeError):
    pass


class IdentityPolicyLoader(yaml.SafeLoader):
    pass


def construct_unique_identity_mapping(loader, node, deep=False):
    loader.flatten_mapping(node)
    mapping = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        try:
            duplicate = key in mapping
        except TypeError as exc:
            raise yaml.constructor.ConstructorError(
                "while constructing identity mapping",
                node.start_mark,
                "found an unhashable mapping key",
                key_node.start_mark,
            ) from exc
        if duplicate:
            raise yaml.constructor.ConstructorError(
                "while constructing identity mapping",
                node.start_mark,
                f"found duplicate key {key!r}",
                key_node.start_mark,
            )
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


IdentityPolicyLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    construct_unique_identity_mapping,
)


def acquire_command_lock(command: str):
    operator_dir = find_operator_dir()
    if not operator_dir:
        return None
    lock_path = os.path.join(operator_dir, "ledger.lock")
    is_mutating = command in MUTATING_COMMANDS
    if not is_mutating and not os.path.exists(lock_path):
        return None
    handle = None
    fd = None
    try:
        flags = os.O_RDONLY | (os.O_CREAT if is_mutating else 0)
        fd = os.open(lock_path, flags, 0o666)
        handle = os.fdopen(fd, "r")
        fd = None
        lock_mode = fcntl.LOCK_EX if is_mutating else fcntl.LOCK_SH
        fcntl.flock(handle.fileno(), lock_mode)
        return handle
    except OSError as exc:
        if fd is not None:
            os.close(fd)
        if handle is not None:
            handle.close()
        raise LedgerWriteError(f"Could not acquire ledger command lock: {exc}") from exc


def release_command_lock(handle) -> None:
    if handle is None:
        return
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
    finally:
        handle.close()


def canonical_json(data: any) -> str:
    return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=True, default=str)


def ledger_db_path(operator_dir: str) -> str:
    return os.path.join(operator_dir, LEDGER_DB_NAME)


def connect_event_store(operator_dir: str, read_only: bool = False) -> sqlite3.Connection:
    db_path = Path(ledger_db_path(operator_dir)).resolve()
    if read_only:
        conn = sqlite3.connect(f"{db_path.as_uri()}?mode=ro", uri=True)
    else:
        conn = sqlite3.connect(str(db_path), timeout=5.0)
        conn.execute("PRAGMA busy_timeout = 5000")
    conn.row_factory = sqlite3.Row
    return conn


def initialize_event_store(operator_dir: str) -> None:
    os.makedirs(operator_dir, exist_ok=True)
    conn = None
    try:
        conn = connect_event_store(operator_dir)
        schema_version = conn.execute("PRAGMA user_version").fetchone()[0]
        if schema_version > LEDGER_SCHEMA_VERSION:
            raise LedgerWriteError(
                f"Durable ledger schema version {schema_version} is newer than this CLI supports "
                f"(maximum {LEDGER_SCHEMA_VERSION})"
            )
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA synchronous = FULL")
        conn.executescript(
            """
                CREATE TABLE IF NOT EXISTS ledger_events (
                    event_id TEXT PRIMARY KEY,
                    record_type TEXT NOT NULL,
                    record_id TEXT NOT NULL,
                    version INTEGER NOT NULL CHECK (version > 0),
                    event_type TEXT NOT NULL,
                    payload_json TEXT NOT NULL,
                    actor_uid INTEGER,
                    actor_name TEXT,
                    created_at TEXT NOT NULL,
                    source_command TEXT,
                    previous_event_hash TEXT,
                    event_hash TEXT NOT NULL,
                    UNIQUE(record_type, record_id, version)
                );

                CREATE INDEX IF NOT EXISTS idx_ledger_events_record
                    ON ledger_events(record_type, record_id, version);

                CREATE TRIGGER IF NOT EXISTS ledger_events_no_update
                BEFORE UPDATE ON ledger_events
                BEGIN
                    SELECT RAISE(ABORT, 'ledger_events is append-only');
                END;

                CREATE TRIGGER IF NOT EXISTS ledger_events_no_delete
                BEFORE DELETE ON ledger_events
                BEGIN
                    SELECT RAISE(ABORT, 'ledger_events is append-only');
                END;
            """
        )
        conn.execute(f"PRAGMA user_version = {LEDGER_SCHEMA_VERSION}")
        conn.commit()
    except LedgerWriteError:
        raise
    except sqlite3.Error as exc:
        raise LedgerWriteError(f"Could not initialize durable ledger store: {exc}") from exc
    finally:
        if conn is not None:
            conn.close()


def operator_dir_for_path(filepath: str) -> str | None:
    path = Path(os.path.abspath(filepath))
    for parent in [path.parent, *path.parents]:
        if parent.name == ".operator":
            return str(parent)
    return None


def logical_record_id(record_type: str, record: dict) -> str:
    id_fields = {
        "task": "task_id",
        "claim": "claim_id",
        "evidence": "evidence_id",
        "handoff": "handoff_id",
        "usage": "usage_id",
    }
    field = id_fields[record_type]
    record_id = record.get(field)
    if not record_id:
        raise LedgerWriteError(f"{record_type} record is missing required field '{field}'")
    if record_type in {"evidence", "handoff"}:
        task_id = record.get("task_id")
        if not task_id:
            raise LedgerWriteError(
                f"{record_type} record '{record_id}' is missing required field 'task_id'"
            )
        return f"{task_id}/{record_id}"
    return str(record_id)


def record_items_for_data(filepath: str, data: any) -> list[tuple[str, str, dict]]:
    operator_dir = operator_dir_for_path(filepath)
    if not operator_dir:
        return []
    try:
        absolute_path = Path(os.path.abspath(filepath))
        absolute_operator_dir = Path(os.path.abspath(operator_dir))
        relative_parts = absolute_path.relative_to(absolute_operator_dir).parts
    except ValueError:
        return []
    if not relative_parts or relative_parts[0] not in TRUST_RECORD_DIRS:
        return []

    record_type = TRUST_RECORD_DIRS[relative_parts[0]]
    if record_type == "usage":
        if len(relative_parts) != 2 or not relative_parts[1].endswith(".yaml"):
            return []
        if not isinstance(data, list):
            raise LedgerWriteError(f"Usage ledger '{filepath}' must contain a YAML list")
        records = data
    else:
        expected_depth = 3 if record_type in {"evidence", "handoff"} else 2
        if len(relative_parts) != expected_depth or not relative_parts[-1].endswith(".yaml"):
            return []
        if not isinstance(data, dict):
            raise LedgerWriteError(f"{record_type} ledger '{filepath}' must contain a YAML mapping")
        records = [data]

    items = []
    seen_ids = set()
    for record in records:
        if not isinstance(record, dict):
            raise LedgerWriteError(
                f"{record_type} ledger '{filepath}' contains a non-mapping record"
            )
        record_id = logical_record_id(record_type, record)
        if record_id in seen_ids:
            raise LedgerWriteError(
                f"{record_type} ledger '{filepath}' contains duplicate record ID '{record_id}'"
            )
        seen_ids.add(record_id)
        items.append((record_type, record_id, record))
    return items


def visible_record_files(operator_dir: str) -> list[str]:
    patterns = [
        os.path.join(operator_dir, "tasks", "*.yaml"),
        os.path.join(operator_dir, "claims", "*.yaml"),
        os.path.join(operator_dir, "evidence", "*", "evidence-*.yaml"),
        os.path.join(operator_dir, "handoffs", "*", "handoff-*.yaml"),
        os.path.join(operator_dir, "usage", "*.yaml"),
    ]
    return sorted(path for pattern in patterns for path in glob.glob(pattern))


def load_visible_records(
    operator_dir: str,
) -> tuple[dict[tuple[str, str], tuple[dict, str]], list[str]]:
    records = {}
    errors = []
    for filepath in visible_record_files(operator_dir):
        relative_path = os.path.relpath(filepath, operator_dir)
        try:
            with open(filepath, "r") as handle:
                data = yaml.safe_load(handle)
            items = record_items_for_data(filepath, data)
        except Exception as exc:
            errors.append(
                f"[Error] Durable ledger cannot parse YAML projection {relative_path}: {exc}"
            )
            continue
        for record_type, record_id, record in items:
            key = (record_type, record_id)
            if key in records:
                errors.append(
                    f"[Error] Duplicate YAML projection for {record_type} {record_id}: "
                    f"{records[key][1]} and {relative_path}"
                )
                continue
            records[key] = (record, relative_path)
    return records, errors


def event_hash_for_fields(
    record_type: str,
    record_id: str,
    version: int,
    event_type: str,
    payload_json: str,
    actor_uid: int | None,
    actor_name: str | None,
    created_at: str,
    source_command: str | None,
    previous_event_hash: str | None,
) -> str:
    hash_fields = {
        "hash_format": LEDGER_HASH_FORMAT,
        "record_type": record_type,
        "record_id": record_id,
        "version": version,
        "event_type": event_type,
        "payload_json": payload_json,
        "actor_uid": actor_uid,
        "actor_name": actor_name,
        "created_at": created_at,
        "source_command": source_command,
        "previous_event_hash": previous_event_hash,
    }
    return hashlib.sha256(canonical_json(hash_fields).encode("utf-8")).hexdigest()


def event_type_for(record_type: str, version: int, source_command: str) -> str:
    if source_command == "legacy-bootstrap":
        return "baseline_imported"
    if record_type == "usage" and source_command == "session-start":
        return "session_started"
    if record_type == "usage" and source_command == "session-end":
        return "session_ended"
    return "record_created" if version == 1 else "record_updated"


def append_record_events(
    operator_dir: str,
    items: list[tuple[str, str, dict]],
    source_command: str,
) -> int:
    if not items:
        return 0
    actor = get_executor_identity()
    created_at = get_current_utc_iso()
    inserted = 0
    conn = connect_event_store(operator_dir)
    try:
        conn.execute("BEGIN IMMEDIATE")
        for record_type, record_id, record in items:
            payload_json = canonical_json(record)
            previous = conn.execute(
                """
                SELECT version, payload_json, event_hash
                FROM ledger_events
                WHERE record_type = ? AND record_id = ?
                ORDER BY version DESC
                LIMIT 1
                """,
                (record_type, record_id),
            ).fetchone()
            if previous and previous["payload_json"] == payload_json:
                continue
            version = (previous["version"] + 1) if previous else 1
            previous_hash = previous["event_hash"] if previous else None
            event_type = event_type_for(record_type, version, source_command)
            event_hash = event_hash_for_fields(
                record_type,
                record_id,
                version,
                event_type,
                payload_json,
                actor.get("uid"),
                actor.get("user"),
                created_at,
                source_command,
                previous_hash,
            )
            conn.execute(
                """
                INSERT INTO ledger_events (
                    event_id, record_type, record_id, version, event_type, payload_json,
                    actor_uid, actor_name, created_at, source_command, previous_event_hash,
                    event_hash
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    event_hash,
                    record_type,
                    record_id,
                    version,
                    event_type,
                    payload_json,
                    actor.get("uid"),
                    actor.get("user"),
                    created_at,
                    source_command,
                    previous_hash,
                    event_hash,
                ),
            )
            inserted += 1
        conn.commit()
    except Exception as exc:
        conn.rollback()
        raise LedgerWriteError(f"Could not append durable ledger event: {exc}") from exc
    finally:
        conn.close()
    return inserted


def ensure_event_store(operator_dir: str) -> None:
    initialize_event_store(operator_dir)
    try:
        conn = connect_event_store(operator_dir, read_only=True)
        try:
            event_count = conn.execute("SELECT COUNT(*) FROM ledger_events").fetchone()[0]
        finally:
            conn.close()
    except sqlite3.Error as exc:
        raise LedgerWriteError(f"Could not inspect durable ledger store: {exc}") from exc
    if event_count:
        return
    records, errors = load_visible_records(operator_dir)
    if errors:
        raise LedgerWriteError(errors[0].removeprefix("[Error] "))
    if records:
        items = [
            (record_type, record_id, payload)
            for (record_type, record_id), (payload, _) in records.items()
        ]
        append_record_events(operator_dir, items, "legacy-bootstrap")


def latest_event_payload(conn: sqlite3.Connection, record_type: str, record_id: str) -> str | None:
    row = conn.execute(
        """
        SELECT payload_json
        FROM ledger_events
        WHERE record_type = ? AND record_id = ?
        ORDER BY version DESC
        LIMIT 1
        """,
        (record_type, record_id),
    ).fetchone()
    return row["payload_json"] if row else None


def assert_projection_current(
    operator_dir: str,
    filepath: str,
    new_items: list[tuple[str, str, dict]],
) -> None:
    new_keys = {(record_type, record_id) for record_type, record_id, _ in new_items}
    old_items = []
    if os.path.exists(filepath):
        try:
            with open(filepath, "r") as handle:
                old_data = yaml.safe_load(handle)
            old_items = record_items_for_data(filepath, old_data)
        except Exception as exc:
            raise LedgerWriteError(
                f"YAML projection is not readable before write: {filepath}: {exc}"
            ) from exc
    old_keys = {(record_type, record_id) for record_type, record_id, _ in old_items}
    removed_keys = old_keys - new_keys
    if removed_keys:
        removed = ", ".join(f"{kind} {record_id}" for kind, record_id in sorted(removed_keys))
        raise LedgerWriteError(
            f"Refusing to remove append-only records from YAML projection: {removed}"
        )

    conn = connect_event_store(operator_dir, read_only=True)
    try:
        for record_type, record_id, record in old_items:
            stored_payload = latest_event_payload(conn, record_type, record_id)
            if stored_payload != canonical_json(record):
                raise LedgerWriteError(
                    f"YAML projection for {record_type} {record_id} differs from the durable ledger; "
                    "run 'operator doctor' before writing"
                )
        for record_type, record_id, _ in new_items:
            if (record_type, record_id) in old_keys:
                continue
            if latest_event_payload(conn, record_type, record_id) is not None:
                raise LedgerWriteError(
                    f"Refusing to reuse durable record ID for {record_type} {record_id}"
                )
    finally:
        conn.close()


def preflight_projection(operator_dir: str, filepath: str, data: any) -> None:
    ensure_event_store(operator_dir)
    items = record_items_for_data(filepath, data)
    assert_projection_current(operator_dir, filepath, items)


def atomic_write_text(filepath: str, content: str) -> None:
    parent = os.path.dirname(filepath)
    os.makedirs(parent, exist_ok=True)
    existing_stat = os.stat(filepath) if os.path.exists(filepath) else None
    if existing_stat:
        target_mode = stat.S_IMODE(existing_stat.st_mode)
    else:
        current_umask = os.umask(0)
        os.umask(current_umask)
        target_mode = 0o666 & ~current_umask
    fd, temp_path = tempfile.mkstemp(prefix=f".{os.path.basename(filepath)}.", dir=parent)
    try:
        if existing_stat:
            try:
                os.fchown(fd, -1, existing_stat.st_gid)
            except PermissionError:
                pass
            try:
                os.fchown(fd, existing_stat.st_uid, -1)
            except PermissionError:
                pass
        os.fchmod(fd, target_mode)
        with os.fdopen(fd, "w") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp_path, filepath)
        directory_fd = os.open(parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)
    except Exception:
        try:
            os.unlink(temp_path)
        except FileNotFoundError:
            pass
        raise


def audit_event_store(operator_dir: str) -> list[str]:
    db_path = ledger_db_path(operator_dir)
    if not os.path.exists(db_path):
        return [
            f"[Error] Durable ledger store is missing: {db_path}. Run 'operator init' to "
            "baseline this YAML ledger."
        ]

    visible_records, messages = load_visible_records(operator_dir)
    latest_events = {}
    chain_state = {}
    try:
        conn = connect_event_store(operator_dir, read_only=True)
        try:
            schema_version = conn.execute("PRAGMA user_version").fetchone()[0]
            if schema_version != LEDGER_SCHEMA_VERSION:
                messages.append(
                    f"[Error] Durable ledger schema version {schema_version} is unsupported; "
                    f"expected {LEDGER_SCHEMA_VERSION}"
                )
                return messages
            rows = conn.execute(
                """
                SELECT event_id, record_type, record_id, version, event_type, payload_json,
                       actor_uid, actor_name, created_at, source_command, previous_event_hash,
                       event_hash
                FROM ledger_events
                ORDER BY record_type, record_id, version
                """
            ).fetchall()
        finally:
            conn.close()
    except sqlite3.Error as exc:
        messages.append(f"[Error] Durable ledger store cannot be read: {exc}")
        return messages

    for row in rows:
        key = (row["record_type"], row["record_id"])
        previous = chain_state.get(key)
        expected_version = previous[0] + 1 if previous else 1
        expected_previous_hash = previous[1] if previous else None
        if row["version"] != expected_version:
            messages.append(
                f"[Error] Durable ledger version gap for {key[0]} {key[1]}: "
                f"expected {expected_version}, got {row['version']}"
            )
        if row["previous_event_hash"] != expected_previous_hash:
            messages.append(
                f"[Error] Durable ledger hash chain is broken for {key[0]} {key[1]} "
                f"at version {row['version']}"
            )

        try:
            decoded_payload = json.loads(row["payload_json"])
            if canonical_json(decoded_payload) != row["payload_json"]:
                messages.append(
                    f"[Error] Durable ledger payload is not canonical for {key[0]} {key[1]} "
                    f"version {row['version']}"
                )
        except (TypeError, json.JSONDecodeError):
            messages.append(
                f"[Error] Durable ledger payload is invalid JSON for {key[0]} {key[1]} "
                f"version {row['version']}"
            )

        expected_hash = event_hash_for_fields(
            row["record_type"],
            row["record_id"],
            row["version"],
            row["event_type"],
            row["payload_json"],
            row["actor_uid"],
            row["actor_name"],
            row["created_at"],
            row["source_command"],
            row["previous_event_hash"],
        )
        if row["event_hash"] != expected_hash or row["event_id"] != row["event_hash"]:
            messages.append(
                f"[Error] Durable ledger event hash mismatch for {key[0]} {key[1]} "
                f"version {row['version']}"
            )
        chain_state[key] = (row["version"], row["event_hash"])
        latest_events[key] = row["payload_json"]

    for key, (record, relative_path) in sorted(visible_records.items()):
        stored_payload = latest_events.get(key)
        if stored_payload is None:
            messages.append(
                f"[Error] Durable ledger has no event for YAML {key[0]} {key[1]} "
                f"({relative_path})"
            )
        elif stored_payload != canonical_json(record):
            messages.append(
                f"[Error] Durable ledger mismatch for {key[0]} {key[1]}: "
                f"YAML projection {relative_path} differs from latest event"
            )

    for key in sorted(set(latest_events) - set(visible_records)):
        messages.append(f"[Error] Durable ledger latest {key[0]} {key[1]} has no YAML projection")

    tasks = {
        record_id: record
        for (record_type, record_id), (record, _) in visible_records.items()
        if record_type == "task"
    }
    for (record_type, record_id), (record, relative_path) in sorted(visible_records.items()):
        if record_type == "task":
            continue
        task_id = record.get("task_id")
        task = tasks.get(task_id)
        if task is None:
            messages.append(
                f"[Error] YAML {record_type} {record_id} references missing task {task_id} "
                f"({relative_path})"
            )
            continue
        if record_type == "claim" and record_id not in (task.get("claims") or []):
            messages.append(f"[Error] Claim {record_id} is not referenced by owning task {task_id}")
        elif record_type == "evidence":
            if record.get("path_or_url") not in (task.get("evidence") or []):
                messages.append(
                    f"[Error] Evidence {record_id} is not referenced by owning task {task_id}"
                )
            claim_id = record.get("claim_id")
            if claim_id:
                claim_entry = visible_records.get(("claim", claim_id))
                evidence_ref = f"evidence/{task_id}/{record.get('evidence_id')}.yaml"
                if claim_entry is not None and claim_entry[0].get("task_id") != task_id:
                    messages.append(
                        f"[Error] Evidence {record_id} belongs to task {task_id} but claim "
                        f"{claim_id} belongs to task {claim_entry[0].get('task_id')}"
                    )
                elif claim_entry is None or evidence_ref not in (
                    claim_entry[0].get("evidence_refs") or []
                ):
                    messages.append(
                        f"[Error] Evidence {record_id} is not referenced by claim {claim_id}"
                    )
        elif record_type == "handoff":
            handoff_ref = f"handoffs/{task_id}/{record.get('handoff_id')}.yaml"
            if handoff_ref not in (task.get("handoffs") or []):
                messages.append(
                    f"[Error] Handoff {record_id} is not referenced by owning task {task_id}"
                )
    return messages


def max_durable_sequence(
    operator_dir: str, record_type: str, prefix: str, scope: str | None = None
) -> int:
    if not os.path.exists(ledger_db_path(operator_dir)):
        return 0
    conn = connect_event_store(operator_dir, read_only=True)
    try:
        if scope is None:
            rows = conn.execute(
                "SELECT DISTINCT record_id FROM ledger_events WHERE record_type = ?",
                (record_type,),
            ).fetchall()
        else:
            rows = conn.execute(
                """
                SELECT DISTINCT record_id
                FROM ledger_events
                WHERE record_type = ? AND record_id LIKE ?
                """,
                (record_type, f"{scope}/%"),
            ).fetchall()
    finally:
        conn.close()
    max_id = 0
    pattern = re.compile(rf"^{re.escape(prefix)}(\d+)$")
    for row in rows:
        leaf_id = row["record_id"].rsplit("/", 1)[-1]
        match = pattern.match(leaf_id)
        if match:
            max_id = max(max_id, int(match.group(1)))
    return max_id


def durable_record_exists(operator_dir: str, record_type: str, record_id: str) -> bool:
    if not os.path.exists(ledger_db_path(operator_dir)):
        return False
    conn = connect_event_store(operator_dir, read_only=True)
    try:
        row = conn.execute(
            "SELECT 1 FROM ledger_events WHERE record_type = ? AND record_id = ? LIMIT 1",
            (record_type, record_id),
        ).fetchone()
    finally:
        conn.close()
    return row is not None


def find_operator_dir(start_dir: str | None = None) -> str | None:
    """Search upwards from start_dir for a directory named '.operator'."""
    if start_dir is None:
        start_dir = os.getcwd()

    current = os.path.abspath(start_dir)
    while True:
        candidate = os.path.join(current, ".operator")
        if os.path.isdir(candidate):
            return candidate
        parent = os.path.dirname(current)
        if parent == current:
            break
        current = parent
    return None


def resolve_operator_context() -> tuple[authority_client.Enrollment | None, str | None]:
    enrollment = authority_client.resolve_enrollment()
    if enrollment is None:
        return None, find_operator_dir()
    operator_dir = enrollment.repository_path / ".operator"
    try:
        metadata = operator_dir.lstat()
    except FileNotFoundError:
        return enrollment, None
    if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode):
        raise authority_client.EnrollmentError(
            f"enrolled operator directory is not a real directory: {operator_dir}"
        )
    return enrollment, str(operator_dir)


def get_current_utc_iso() -> str:
    """Return the current time in UTC ISO format."""
    return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat()


def load_yaml(filepath: str) -> any:
    """Helper to load a YAML file safely."""
    try:
        with open(filepath, "r") as f:
            return yaml.safe_load(f)
    except Exception as e:
        print(f"Error loading YAML {filepath}: {e}", file=sys.stderr)
        return None


def save_yaml(data: any, filepath: str) -> bool:
    """Helper to save a python object to a YAML file."""
    try:
        rendered = yaml.safe_dump(data, default_flow_style=False, sort_keys=False)
        items = record_items_for_data(filepath, data)
        if items:
            operator_dir = operator_dir_for_path(filepath)
            ensure_event_store(operator_dir)
            assert_projection_current(operator_dir, filepath, items)
            source_command = sys.argv[1] if len(sys.argv) > 1 else "internal"
            append_record_events(operator_dir, items, source_command)
        atomic_write_text(filepath, rendered)
        return True
    except LedgerWriteError:
        raise
    except Exception as e:
        raise LedgerWriteError(f"Error saving YAML {filepath}: {e}") from e


def get_machine_identity() -> str:
    """Short machine name for record provenance.

    `OPERATOR_MACHINE` overrides detection (also the test hook); otherwise the
    short hostname. The ledger itself stays single-machine — this field only
    records WHERE a session/usage/evidence record was produced.
    """
    override = os.environ.get("OPERATOR_MACHINE")
    if override and override.strip():
        return override.strip()
    try:
        host = socket.gethostname()
    except Exception:
        return "unknown"
    short = host.split(".")[0].strip()
    return short or "unknown"


def get_executor_identity() -> dict:
    test_uid_env = os.environ.get("OPERATOR_TEST_UID")
    test_sentinel_env = os.environ.get("OPERATOR_TEST_SENTINEL")

    use_override = False
    spoof_attempt = False

    if test_uid_env:
        if test_sentinel_env == "true" or test_sentinel_env == "1":
            use_override = True
        else:
            spoof_attempt = True

    if use_override:
        uid = int(test_uid_env)
        try:
            user = pwd.getpwuid(uid).pw_name
        except Exception:
            user = f"uid-{uid}"
        return {
            "uid": uid,
            "user": user,
            "machine": get_machine_identity(),
            "test_override_active": True,
        }
    else:
        uid = os.getuid()
        try:
            user = pwd.getpwuid(uid).pw_name
        except Exception:
            user = f"uid-{uid}"
        res = {"uid": uid, "user": user, "machine": get_machine_identity()}
        if spoof_attempt:
            res["test_override_unauthorized"] = True
        return res


def load_identity_policy(operator_dir: str) -> dict:
    identity_path = os.path.join(operator_dir, "identity.yaml")
    if not os.path.exists(identity_path):
        return {"mode": "single_user", "uids": {}}

    try:
        with open(identity_path, "r") as handle:
            raw_policy = yaml.load(handle, Loader=IdentityPolicyLoader)
    except (OSError, yaml.YAMLError) as exc:
        raise IdentityPolicyError(f"could not read identity.yaml: {exc}") from exc

    if raw_policy is None:
        raw_policy = {}
    if not isinstance(raw_policy, dict):
        raise IdentityPolicyError("identity.yaml must contain a mapping")

    mode = raw_policy.get("mode", "single_user")
    if not isinstance(mode, str) or mode not in {"single_user", "enforced"}:
        raise IdentityPolicyError("identity.yaml mode must be either 'single_user' or 'enforced'")

    raw_uids = raw_policy.get("uids", {})
    if raw_uids is None:
        raw_uids = {}
    if not isinstance(raw_uids, dict):
        raise IdentityPolicyError("identity.yaml uids must contain a mapping")

    identities = {}
    for raw_uid, raw_identity in raw_uids.items():
        if isinstance(raw_uid, bool):
            raise IdentityPolicyError(f"identity UID key {raw_uid!r} is not a non-negative integer")
        if isinstance(raw_uid, int):
            uid = raw_uid
        elif isinstance(raw_uid, str) and re.fullmatch(r"[0-9]+", raw_uid):
            uid = int(raw_uid)
        else:
            raise IdentityPolicyError(f"identity UID key {raw_uid!r} is not a non-negative integer")
        if uid < 0:
            raise IdentityPolicyError(f"identity UID key {raw_uid!r} is not a non-negative integer")
        if uid in identities:
            raise IdentityPolicyError(f"identity UID {uid} is configured more than once")

        legacy_scalar = isinstance(raw_identity, str)
        if legacy_scalar:
            name = raw_identity.strip()
            roles = {"builder", "verifier"}
        elif isinstance(raw_identity, dict):
            name_value = raw_identity.get("name")
            if not isinstance(name_value, str):
                raise IdentityPolicyError(f"identity UID {uid} must have a string name")
            name = name_value.strip()
            raw_roles = raw_identity.get("roles")
            if not isinstance(raw_roles, list) or any(
                not isinstance(role, str) for role in raw_roles
            ):
                raise IdentityPolicyError(f"identity UID {uid} roles must be a list of role names")
            roles = set(raw_roles)
            unknown_roles = roles - {"builder", "verifier"}
            if unknown_roles:
                rendered = ", ".join(sorted(unknown_roles))
                raise IdentityPolicyError(f"identity UID {uid} has unknown role(s): {rendered}")
        else:
            raise IdentityPolicyError(
                f"identity UID {uid} must map to a name or a name/roles mapping"
            )

        if not name:
            raise IdentityPolicyError(f"identity UID {uid} must have a non-empty name")
        identities[uid] = {
            "name": name,
            "roles": frozenset(roles),
            "legacy_scalar": legacy_scalar,
        }

    return {"mode": mode, "uids": identities}


def identity_policy_for_write(operator_dir: str) -> dict | None:
    try:
        return load_identity_policy(operator_dir)
    except IdentityPolicyError as exc:
        print(f"Error: invalid identity policy: {exc}", file=sys.stderr)
        return None


def require_executor_role(policy: dict, executor: dict, role: str, action: str) -> dict | None:
    if policy["mode"] != "enforced":
        return None

    uid = executor.get("uid")
    identity = policy["uids"].get(uid)
    if identity is None:
        print(
            f"Error: refusing to {action}: executor uid {uid} is not a known identity",
            file=sys.stderr,
        )
        return None
    if role not in identity["roles"]:
        print(
            f"Error: refusing to {action}: executor uid {uid} does not have the {role} role",
            file=sys.stderr,
        )
        return None
    return identity


def slugify(text: str) -> str:
    """Slugify text to generate safe task IDs."""
    text = text.lower()
    text = re.sub(r"[^a-z0-9\-_]+", "-", text)
    text = re.sub(r"-+", "-", text).strip("-")
    return text


def resolve_task_record(
    operator_dir: str, requested_task_id: str | None
) -> tuple[str, str, dict] | None:
    task_id = requested_task_id
    if not task_id:
        config = load_yaml(os.path.join(operator_dir, "operator.yaml")) or {}
        task_id = config.get("current_task")
        if not task_id:
            print(
                "Error: No active task set. Specify task ID or create a task.",
                file=sys.stderr,
            )
            return None

    if not isinstance(task_id, str) or not VALID_TASK_ID.fullmatch(task_id):
        print(
            "Error: Task ID must start with an alphanumeric character and contain only "
            "letters, numbers, '.', '_', or '-'.",
            file=sys.stderr,
        )
        return None

    task_path = os.path.join(operator_dir, "tasks", f"{task_id}.yaml")
    if not os.path.isfile(task_path):
        print(f"Error: Task '{task_id}' not found.", file=sys.stderr)
        return None

    task = load_yaml(task_path)
    if not isinstance(task, dict):
        print(f"Error: Task '{task_id}' is not a YAML mapping.", file=sys.stderr)
        return None
    if task.get("task_id") != task_id:
        print(
            f"Error: Task file for '{task_id}' contains mismatched task_id "
            f"'{task.get('task_id')}'.",
            file=sys.stderr,
        )
        return None
    return task_id, task_path, task


def resolve_claim_record(operator_dir: str, claim_id: str) -> tuple[str, dict] | None:
    if not isinstance(claim_id, str) or not VALID_CLAIM_ID.fullmatch(claim_id):
        print(
            "Error: Claim ID must use the form claim-NNNN.",
            file=sys.stderr,
        )
        return None

    claim_path = os.path.join(operator_dir, "claims", f"{claim_id}.yaml")
    if not os.path.isfile(claim_path):
        print(f"Error: Claim '{claim_id}' not found.", file=sys.stderr)
        return None

    claim = load_yaml(claim_path)
    if not isinstance(claim, dict):
        print(f"Error: Claim '{claim_id}' is not a YAML mapping.", file=sys.stderr)
        return None
    if claim.get("claim_id") != claim_id:
        print(
            f"Error: Claim file for '{claim_id}' contains mismatched claim_id "
            f"'{claim.get('claim_id')}'.",
            file=sys.stderr,
        )
        return None
    return claim_path, claim


def normalize_sha256(value: str) -> str:
    if not isinstance(value, str) or not VALID_SHA256.fullmatch(value):
        raise EvidenceFingerprintError("SHA-256 must be exactly 64 hexadecimal characters")
    return value.lower()


def fingerprint_regular_file(filepath: str) -> dict:
    try:
        sha256 = hashlib.sha256()
        fd = os.open(filepath, os.O_RDONLY | os.O_NONBLOCK)
        with os.fdopen(fd, "rb") as f:
            before = os.fstat(f.fileno())
            if not stat.S_ISREG(before.st_mode):
                raise EvidenceFingerprintError(f"Evidence path is not a regular file: {filepath}")
            for chunk in iter(lambda: f.read(65536), b""):
                sha256.update(chunk)
            after = os.fstat(f.fileno())
        path_after = os.stat(filepath)
    except EvidenceFingerprintError:
        raise
    except OSError as exc:
        raise EvidenceFingerprintError(
            f"Could not fingerprint evidence file '{filepath}': {exc}"
        ) from exc

    before_identity = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns)
    after_identity = (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)
    if before_identity != after_identity:
        raise EvidenceFingerprintError(f"Evidence file changed while fingerprinting: {filepath}")
    path_identity = (
        path_after.st_dev,
        path_after.st_ino,
        path_after.st_size,
        path_after.st_mtime_ns,
    )
    if after_identity != path_identity:
        raise EvidenceFingerprintError(f"Evidence path changed while fingerprinting: {filepath}")

    return {
        "algorithm": "sha256",
        "value": sha256.hexdigest(),
        "size_bytes": after.st_size,
        "mtime_ns": after.st_mtime_ns,
    }


def parse_stored_fingerprint(
    fingerprint: object, legacy_hash: object = None
) -> tuple[dict | None, str | None]:
    if fingerprint is None:
        if legacy_hash is None:
            return None, "has no SHA-256 fingerprint"
        try:
            value = normalize_sha256(legacy_hash)
        except EvidenceFingerprintError as exc:
            return None, str(exc)
        return {
            "algorithm": "sha256",
            "value": value,
            "size_bytes": None,
            "mtime_ns": None,
        }, None

    if not isinstance(fingerprint, dict):
        return None, "fingerprint must be a mapping"
    if fingerprint.get("algorithm") != "sha256":
        return None, "fingerprint algorithm must be sha256"
    try:
        value = normalize_sha256(fingerprint.get("value"))
    except EvidenceFingerprintError as exc:
        return None, str(exc)

    size_bytes = fingerprint.get("size_bytes")
    mtime_ns = fingerprint.get("mtime_ns")
    if size_bytes is not None and (type(size_bytes) is not int or size_bytes < 0):
        return None, "fingerprint size_bytes must be a non-negative integer or null"
    if mtime_ns is not None and (type(mtime_ns) is not int or mtime_ns < 0):
        return None, "fingerprint mtime_ns must be a non-negative integer or null"

    if legacy_hash is not None:
        try:
            normalized_legacy = normalize_sha256(legacy_hash)
        except EvidenceFingerprintError as exc:
            return None, str(exc)
        if normalized_legacy != value:
            return None, "raw hash does not match structured fingerprint"

    return {
        "algorithm": "sha256",
        "value": value,
        "size_bytes": size_bytes,
        "mtime_ns": mtime_ns,
    }, None


def calculate_file_hash(filepath: str) -> str | None:
    """Calculate the SHA-256 hash of a local regular file."""
    try:
        return fingerprint_regular_file(filepath)["value"]
    except EvidenceFingerprintError:
        return None


def evidence_freshness_diagnostics(evidence: dict, verified: bool) -> tuple[list[str], list[str]]:
    issues = []
    infos = []
    evidence_id = evidence.get("evidence_id", "unknown")
    provenance = evidence.get("provenance")
    source = evidence.get("source")
    source_kind = source.get("kind") if isinstance(source, dict) else None
    is_local = (
        provenance
        in {
            "local_file",
            "living_document",
            "structural_verification",
        }
        or source_kind == "local_file"
    )

    stored, fingerprint_error = parse_stored_fingerprint(
        evidence.get("fingerprint"), evidence.get("hash")
    )
    if not is_local:
        if fingerprint_error and (
            evidence.get("fingerprint") is not None or evidence.get("hash") is not None
        ):
            prefix = "[Error]" if verified else "[Warning]"
            message = f"{prefix} Evidence {evidence_id} has invalid remote fingerprint: {fingerprint_error}"
            (issues if verified else infos).append(message)
        infos.append(
            f"[Warning] Evidence {evidence_id} remote freshness is uncheckable without a local "
            "snapshot; verification command was not executed"
        )
        return issues, infos

    snapshot_path = evidence.get("path_or_url")
    if not isinstance(snapshot_path, str) or not snapshot_path:
        issues.append(f"[Error] Evidence {evidence_id} retained snapshot path is missing")
        return issues, infos
    if not os.path.exists(snapshot_path):
        issues.append(
            f"[Error] Evidence {evidence_id} retained snapshot is missing: {snapshot_path}"
        )
        return issues, infos
    if fingerprint_error:
        issues.append(
            f"[Error] Evidence {evidence_id} retained snapshot fingerprint is invalid: "
            f"{fingerprint_error}"
        )
        return issues, infos

    try:
        current = fingerprint_regular_file(snapshot_path)
    except EvidenceFingerprintError as exc:
        issues.append(f"[Error] Evidence {evidence_id} retained snapshot is unreadable: {exc}")
        return issues, infos

    if current["value"] != stored["value"]:
        issues.append(
            f"[Error] Evidence {evidence_id} retained snapshot content changed "
            f"(expected {stored['value']}, got {current['value']}): {snapshot_path}"
        )
        return issues, infos
    if stored["size_bytes"] is not None and current["size_bytes"] != stored["size_bytes"]:
        issues.append(
            f"[Error] Evidence {evidence_id} retained snapshot size changed "
            f"(expected {stored['size_bytes']}, got {current['size_bytes']}): {snapshot_path}"
        )
        return issues, infos
    if stored["mtime_ns"] is not None and current["mtime_ns"] != stored["mtime_ns"]:
        infos.append(
            f"[Info] Evidence {evidence_id} retained snapshot metadata changed but content "
            "fingerprint remains current"
        )

    if source_kind != "local_file":
        return issues, infos
    source_locator = source.get("locator")
    if not isinstance(source_locator, str) or not source_locator:
        issues.append(f"[Error] Evidence {evidence_id} local source locator is invalid")
        return issues, infos
    if os.path.abspath(source_locator) == os.path.abspath(snapshot_path):
        return issues, infos
    if not os.path.exists(source_locator):
        infos.append(
            f"[Warning] Evidence {evidence_id} local source is unavailable; retained snapshot "
            "remains current"
        )
        return issues, infos

    source_stored, source_error = parse_stored_fingerprint(source.get("fingerprint"))
    if source_error:
        issues.append(
            f"[Error] Evidence {evidence_id} local source fingerprint is invalid: {source_error}"
        )
        return issues, infos
    try:
        source_current = fingerprint_regular_file(source_locator)
    except EvidenceFingerprintError as exc:
        prefix = "[Error]" if verified else "[Warning]"
        message = f"{prefix} Evidence {evidence_id} local source is unreadable: {exc}"
        (issues if verified else infos).append(message)
        return issues, infos

    if source_current["value"] != source_stored["value"]:
        state = "verified" if verified else "draft"
        prefix = "[Error]" if verified else "[Warning]"
        message = (
            f"{prefix} Evidence {evidence_id} {state} local source content changed since "
            f"attachment (expected {source_stored['value']}, got {source_current['value']}): "
            f"{source_locator}"
        )
        (issues if verified else infos).append(message)
    elif (
        source_stored["mtime_ns"] is not None
        and source_current["mtime_ns"] != source_stored["mtime_ns"]
    ):
        infos.append(
            f"[Info] Evidence {evidence_id} local source metadata changed but content fingerprint "
            "remains current"
        )
    return issues, infos


def get_next_claim_id(operator_dir: str) -> str:
    """Determine the next sequential claim-XXXX ID."""
    ensure_event_store(operator_dir)
    claims_dir = os.path.join(operator_dir, "claims")
    existing = glob.glob(os.path.join(claims_dir, "claim-*.yaml"))
    max_id = max_durable_sequence(operator_dir, "claim", "claim-")
    for path in existing:
        filename = os.path.basename(path)
        match = re.match(r"claim-(\d+)\.yaml", filename)
        if match:
            num = int(match.group(1))
            if num > max_id:
                max_id = num
    return f"claim-{max_id + 1:04d}"


def get_next_evidence_id(operator_dir: str, task_id: str) -> str:
    """Determine the next sequential evidence-XXXX ID for a specific task."""
    ensure_event_store(operator_dir)
    task_ev_dir = os.path.join(operator_dir, "evidence", task_id)
    os.makedirs(task_ev_dir, exist_ok=True)
    existing = glob.glob(os.path.join(task_ev_dir, "evidence-*.yaml"))
    max_id = max_durable_sequence(operator_dir, "evidence", "evidence-", task_id)
    for path in existing:
        filename = os.path.basename(path)
        match = re.match(r"evidence-(\d+)\.yaml", filename)
        if match:
            num = int(match.group(1))
            if num > max_id:
                max_id = num
    return f"evidence-{max_id + 1:04d}"


def get_next_usage_id(operator_dir: str) -> str:
    """Determine the next sequential usage-XXXX ID."""
    ensure_event_store(operator_dir)
    usage_dir = os.path.join(operator_dir, "usage")
    existing = glob.glob(os.path.join(usage_dir, "*.yaml"))
    max_id = max_durable_sequence(operator_dir, "usage", "usage-")
    for path in existing:
        try:
            records = load_yaml(path) or []
            for rec in records:
                uid = rec.get("usage_id", "")
                match = re.match(r"usage-(\d+)", uid)
                if match:
                    num = int(match.group(1))
                    if num > max_id:
                        max_id = num
        except Exception:
            pass
    return f"usage-{max_id + 1:04d}"


def get_next_handoff_id(operator_dir: str, task_id: str) -> str:
    """Determine the next sequential handoff-XXXX ID for a specific task."""
    ensure_event_store(operator_dir)
    task_ho_dir = os.path.join(operator_dir, "handoffs", task_id)
    os.makedirs(task_ho_dir, exist_ok=True)
    existing = glob.glob(os.path.join(task_ho_dir, "handoff-*.yaml"))
    max_id = max_durable_sequence(operator_dir, "handoff", "handoff-", task_id)
    for path in existing:
        filename = os.path.basename(path)
        match = re.match(r"handoff-(\d+)\.yaml", filename)
        if match:
            num = int(match.group(1))
            if num > max_id:
                max_id = num
    return f"handoff-{max_id + 1:04d}"


# --- Commands ---


def init_cmd(args) -> int:
    target_dir = os.path.abspath(os.path.join(os.getcwd(), ".operator"))
    if os.path.exists(target_dir):
        if not os.path.isdir(target_dir):
            print(
                f"Error: Operator path exists and is not a directory: {target_dir}",
                file=sys.stderr,
            )
            return 1
        ensure_event_store(target_dir)
        print(f"Operator directory already exists at: {target_dir}")
        print(f"Durable ledger store initialized at: {ledger_db_path(target_dir)}")
        return 0

    subdirs = [
        "harnesses",
        "tasks",
        "claims",
        "evidence",
        "handoffs",
        "usage",
        "briefs",
    ]
    for sd in subdirs:
        os.makedirs(os.path.join(target_dir, sd), exist_ok=True)

    ensure_event_store(target_dir)

    harnesses_data = {
        "codex": {
            "harness_id": "codex",
            "display_name": "Codex",
            "kind": "codex",
            "command": None,
            "working_directory": None,
            "model": "configured-codex-model",
            "permission_profile": "default",
            "usage_source": "status",
            "transcript_source": "local",
            "strengths": ["repo-local implementation", "local verification"],
            "known_failure_modes": [
                "narration overclaim without evidence gates",
                "context and usage budget pressure",
            ],
        },
        "claude": {
            "harness_id": "claude",
            "display_name": "Claude Code",
            "kind": "claude",
            "command": None,
            "working_directory": None,
            "model": "configured-claude-model",
            "permission_profile": "full",
            "usage_source": "usage",
            "transcript_source": "local",
            "strengths": ["architecture", "code review", "prose"],
            "known_failure_modes": [
                "recommendations can rely on unchecked data-source assumptions",
                "high-volume execution costs",
            ],
        },
        "gemini-agy": {
            "harness_id": "gemini-agy",
            "display_name": "Gemini / Antigravity",
            "kind": "gemini",
            "command": None,
            "working_directory": None,
            "model": "configured-gemini-model",
            "permission_profile": "default",
            "usage_source": "quota",
            "transcript_source": "cloud",
            "strengths": ["broad context", "alternate implementation", "high-volume work"],
            "known_failure_modes": [
                "execution narration can outrun evidence",
                "fine-grained local file manipulation without tool support",
            ],
        },
        "grok": {
            "harness_id": "grok",
            "display_name": "Grok",
            "kind": "grok",
            "command": None,
            "working_directory": None,
            "model": "configured-grok-model",
            "permission_profile": "default",
            "usage_source": "manual",
            "transcript_source": "cloud",
            "strengths": ["alternate implementation", "alternate review", "broad context"],
            "known_failure_modes": [
                "execution narration can outrun evidence",
                "unimplemented automatic usage import adapter",
            ],
        },
        "copilot": {
            "harness_id": "copilot",
            "display_name": "GitHub Copilot",
            "kind": "copilot",
            "command": None,
            "working_directory": None,
            "model": "default",
            "permission_profile": "default",
            "usage_source": "manual",
            "transcript_source": "cloud",
            "strengths": ["github-native async and SDK integration"],
            "known_failure_modes": ["offline operations"],
        },
        "gemma3_local": {
            "harness_id": "gemma3_local",
            "display_name": "Gemma 3 Local",
            "kind": "gemma3",
            "command": None,
            "working_directory": None,
            "model": "gemma3:27b",
            "permission_profile": "local",
            "usage_source": "local",
            "transcript_source": "local",
            "strengths": ["local execution", "offline privacy"],
            "known_failure_modes": ["structural regressions", "context limitations"],
        },
        "gemma4_local": {
            "harness_id": "gemma4_local",
            "display_name": "Gemma 4 Local",
            "kind": "gemma4",
            "command": None,
            "working_directory": None,
            "model": "gemma4:26b",
            "permission_profile": "local",
            "usage_source": "local",
            "transcript_source": "local",
            "strengths": [
                "local execution",
                "offline privacy",
                "reasoning capability",
            ],
            "known_failure_modes": ["structural regressions", "context limitations"],
        },
        "gpt-oss_local": {
            "harness_id": "gpt-oss_local",
            "display_name": "GPT-OSS Local",
            "kind": "gpt-oss",
            "command": None,
            "working_directory": None,
            "model": "gpt-oss:20b",
            "permission_profile": "local",
            "usage_source": "local",
            "transcript_source": "local",
            "strengths": [
                "local execution",
                "offline privacy",
                "reasoning capability",
            ],
            "known_failure_modes": ["structural regressions", "context limitations"],
        },
    }

    for hid, hdata in harnesses_data.items():
        save_yaml(hdata, os.path.join(target_dir, "harnesses", f"{hid}.yaml"))

    save_yaml({"current_task": None}, os.path.join(target_dir, "operator.yaml"))

    print(f"Initialized empty Operator Control Plane repository in {target_dir}")
    return 0


def task_create_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    if args.assign:
        h_path = os.path.join(op_dir, "harnesses", f"{args.assign}.yaml")
        if not os.path.exists(h_path):
            print(
                f"Error: Assigned harness '{args.assign}' does not exist in registry.",
                file=sys.stderr,
            )
            return 1
    if args.review:
        h_path = os.path.join(op_dir, "harnesses", f"{args.review}.yaml")
        if not os.path.exists(h_path):
            print(
                f"Error: Review harness '{args.review}' does not exist in registry.",
                file=sys.stderr,
            )
            return 1

    objective = args.objective
    task_id = args.task_id
    if not task_id:
        task_id = slugify(objective)[:50]
        if not task_id:
            print(
                "Error: Could not generate task ID from objective. Please provide --id.",
                file=sys.stderr,
            )
            return 1

    if not VALID_TASK_ID.fullmatch(task_id):
        print(
            "Error: Task ID must start with an alphanumeric character and contain only "
            "letters, numbers, '.', '_', or '-'.",
            file=sys.stderr,
        )
        return 1
    task_path = os.path.join(op_dir, "tasks", f"{task_id}.yaml")
    ensure_event_store(op_dir)
    if os.path.exists(task_path) or durable_record_exists(op_dir, "task", task_id):
        print(f"Error: Task '{task_id}' already exists.", file=sys.stderr)
        return 1

    repo = os.path.abspath(args.repo)
    utc_now = get_current_utc_iso()

    task_data = {
        "task_id": task_id,
        "repo": repo,
        "objective": objective,
        "status": "assigned" if args.assign else "new",
        "assigned_harness": args.assign,
        "review_harness": args.review,
        "created_at": utc_now,
        "updated_at": utc_now,
        "operator_decision": None,
        "next_action": None,
        "open_assumptions": args.assumptions,
        "claims": [],
        "evidence": [],
        "executor": get_executor_identity(),
    }

    save_yaml(task_data, task_path)

    config_path = os.path.join(op_dir, "operator.yaml")
    config = load_yaml(config_path) or {}
    config["current_task"] = task_id
    save_yaml(config, config_path)

    os.makedirs(os.path.join(op_dir, "evidence", task_id), exist_ok=True)
    os.makedirs(os.path.join(op_dir, "handoffs", task_id), exist_ok=True)

    print(f"Task '{task_id}' created successfully.")
    print(f"Active task set to '{task_id}'.")
    return 0


def task_show_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    print("=" * 80)
    print(f"Task ID:          {task.get('task_id')}")
    print(f"Objective:        {task.get('objective')}")
    print(f"Status:           {task.get('status')}")
    print(f"Repo:             {task.get('repo')}")
    print(f"Assigned Harness: {task.get('assigned_harness') or 'None'}")
    print(f"Review Harness:   {task.get('review_harness') or 'None'}")
    print(f"Created At:       {task.get('created_at')}")
    print(f"Updated At:       {task.get('updated_at')}")
    if task.get("operator_decision"):
        print(f"Operator Dec:     {task.get('operator_decision')}")
    if task.get("next_action"):
        print(f"Next Action:      {task.get('next_action')}")
    print("-" * 80)
    print("Open Assumptions:")
    assumptions = task.get("open_assumptions") or []
    if assumptions:
        for val in assumptions:
            print(f"  - {val}")
    else:
        print("  (None)")
    print("-" * 80)

    print("Claims:")
    claims_list = task.get("claims") or []
    if claims_list:
        for cid in claims_list:
            resolved_claim = resolve_claim_record(op_dir, cid)
            if not resolved_claim:
                print(f"  - {cid} (Record missing)")
                continue
            _, c_data = resolved_claim
            outcome = c_data.get("verification_outcome")
            if not outcome:
                outcome = "verified" if c_data.get("verification_status") else "unverified"
            status_str = str(outcome).upper()
            print(f"  - [{cid}] ({c_data.get('type')}) - {status_str}: {c_data.get('text')}")
            if c_data.get("verdict"):
                print(f"    Verdict: {c_data.get('verdict')}")
    else:
        print("  (None registered)")
    print("-" * 80)

    print("Evidence:")
    ev_list = task.get("evidence") or []
    if ev_list:
        for ev in ev_list:
            print(f"  - {ev}")
    else:
        print("  (None attached)")
    print("-" * 80)

    print("Handoffs:")
    ho_list = task.get("handoffs") or []
    if ho_list:
        for ho in ho_list:
            print(f"  - {ho}")
    else:
        print("  (None captured)")
    print("=" * 80)
    return 0


def prepare_journal_transaction(
    conn_journal, operation_key: str, request: dict, *, evidence_path: str | None = None
) -> int | None:
    """Wrap authority_projection.prepare_transaction; returns an exit code on failure, else None."""
    try:
        authority_projection.prepare_transaction(
            conn_journal, operation_key, request, evidence_path=evidence_path
        )
        return None
    except RuntimeError as e:
        print(
            f"Error: {e}. A prior attempt with this operation key may have already "
            "completed; check current state or run 'operator authority-reconcile' "
            "before retrying.",
            file=sys.stderr,
        )
        conn_journal.close()
        return 1


def claim_add_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    if enrollment:
        ledger_id = enrollment.ledger_id
        socket_path = enrollment.socket_path
        claim_id = get_next_claim_id(op_dir)

        # Prepare request
        operation_key = f"op-claim-add-{claim_id}"
        try:
            expected = authority_projection.compile_expected(
                op_dir, "claim.create", task_id, claim_id
            )
        except RuntimeError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

        request = {
            "action": "commit",
            "ledger_id": ledger_id,
            "operation_key": operation_key,
            "operation": {
                "kind": "claim.create",
                "task_id": task_id,
                "claim_id": claim_id,
                "claim_type": args.type,
                "text": args.text,
            },
            "expected": expected,
            "blob": None,
        }
        if args.gate:
            request["operation"]["required_gate"] = args.gate

        conn_journal = authority_projection.ensure_journal(op_dir)
        failure = prepare_journal_transaction(conn_journal, operation_key, request)
        if failure is not None:
            return failure

        try:
            client = authority_client.AuthorityClient(socket_path)
            res = client.send_request(request)
        except Exception as e:
            print(f"Error: Broker dispatch failed: {e}", file=sys.stderr)
            conn_journal.close()
            return 1

        if not res.get("ok"):
            authority_projection.discard_prepared_transaction(conn_journal, operation_key)
            print(f"Error: Claim creation rejected by broker: {res.get('error')}", file=sys.stderr)
            conn_journal.close()
            return 1

        receipt = res["receipt"]
        authority_projection.commit_transaction(conn_journal, operation_key, receipt)

        try:
            authority_projection.reconcile_projections(op_dir, ledger_id, client)
        except Exception as e:
            print(f"Warning: Projection reconciliation failed: {e}", file=sys.stderr)

        conn_journal.close()
        print(f"Registered claim '{claim_id}' on task '{task_id}' via authority broker.")
        return 0

    task_id, task_path, task = resolved_task

    exec_identity = get_executor_identity()
    identity_policy = identity_policy_for_write(op_dir)
    if identity_policy is None:
        return 1
    if identity_policy["mode"] == "enforced" and not require_executor_role(
        identity_policy, exec_identity, "builder", "create claim"
    ):
        return 1

    preflight_projection(op_dir, task_path, task)

    claim_id = get_next_claim_id(op_dir)
    made_by = args.by if args.by else task.get("assigned_harness") or "operator"
    utc_now = get_current_utc_iso()

    claim_data = {
        "claim_id": claim_id,
        "task_id": task_id,
        "type": args.type,
        "text": args.text,
        "made_by": made_by,
        "made_at": utc_now,
        "verification_status": False,
        "verdict": None,
        "verified_by": None,
        "evidence_refs": [],
        "required_gate": args.gate,
        "supervision_layer": args.layer,
        "executor": dict(exec_identity),
        "author_executor": dict(exec_identity),
    }

    claim_path = os.path.join(op_dir, "claims", f"{claim_id}.yaml")
    save_yaml(claim_data, claim_path)

    if args.type == "supervision_credit" and not args.layer:
        print(
            "Warning: supervision_credit claim has no --layer; doctor will flag "
            "it as an ambiguous credit (FR-12).",
            file=sys.stderr,
        )

    task["claims"] = task.get("claims") or []
    if claim_id not in task["claims"]:
        task["claims"].append(claim_id)
    task["updated_at"] = utc_now
    save_yaml(task, task_path)

    print(f"Registered claim '{claim_id}' on task '{task_id}'.")
    return 0


def evidence_attach_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    if enrollment:
        ledger_id = enrollment.ledger_id
        socket_path = enrollment.socket_path

        # Enrolled evidence requires claim
        if not args.claim:
            print(
                "Error: --claim is required for evidence-attach in enrolled repositories.",
                file=sys.stderr,
            )
            return 1

        resolved_claim = resolve_claim_record(op_dir, args.claim)
        if not resolved_claim:
            return 1
        claim_path, claim = resolved_claim
        if claim.get("task_id") != task_id:
            print(
                f"Error: Claim '{args.claim}' belongs to task '{claim.get('task_id')}', not '{task_id}'.",
                file=sys.stderr,
            )
            return 1

        # Get path_or_url
        original_locator = args.path_or_url
        if not os.path.exists(original_locator) or not os.path.isfile(original_locator):
            print(
                f"Error: Evidence path does not exist or is not a file: {original_locator}",
                file=sys.stderr,
            )
            return 1

        # Calculate file size and sha256
        size_bytes = os.path.getsize(original_locator)
        h = hashlib.sha256()
        with open(original_locator, "rb") as f:
            while True:
                chunk = f.read(65536)
                if not chunk:
                    break
                h.update(chunk)
        file_hash = h.hexdigest()

        if args.hash:
            try:
                expected_hash = normalize_sha256(args.hash)
                if expected_hash != file_hash:
                    print(
                        f"Error: SHA-256 hash mismatch. Expected: {expected_hash}, Actual: {file_hash}",
                        file=sys.stderr,
                    )
                    return 1
            except Exception as e:
                print(f"Error: Invalid --hash: {e}", file=sys.stderr)
                return 1

        # Generate evidence_id
        evidence_id = get_next_evidence_id(op_dir, task_id)

        # Prepare operation
        kind = "evidence.attach_status" if args.status else "evidence.attach_draft"
        operation = {
            "kind": kind,
            "task_id": task_id,
            "claim_id": args.claim,
            "evidence_id": evidence_id,
            "evidence_type": args.type,
        }
        if args.status:
            operation["verification_status"] = args.status

        try:
            expected = authority_projection.compile_expected(
                op_dir, kind, task_id, args.claim, evidence_id
            )
        except RuntimeError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

        operation_key = f"op-evidence-attach-{evidence_id}"
        request = {
            "action": "commit",
            "ledger_id": ledger_id,
            "operation_key": operation_key,
            "operation": operation,
            "expected": expected,
            "blob": {
                "sha256": file_hash,
                "size_bytes": size_bytes,
            },
        }

        conn_journal = authority_projection.ensure_journal(op_dir)
        failure = prepare_journal_transaction(
            conn_journal, operation_key, request, evidence_path=original_locator
        )
        if failure is not None:
            return failure

        try:
            client = authority_client.AuthorityClient(socket_path)
            res = client.send_request(request, evidence_path=original_locator)
        except Exception as e:
            print(f"Error: Broker dispatch failed: {e}", file=sys.stderr)
            conn_journal.close()
            return 1

        if not res.get("ok"):
            authority_projection.discard_prepared_transaction(conn_journal, operation_key)
            print(
                f"Error: Evidence attachment rejected by broker: {res.get('error')}",
                file=sys.stderr,
            )
            conn_journal.close()
            return 1

        receipt = res["receipt"]
        authority_projection.commit_transaction(conn_journal, operation_key, receipt)

        try:
            authority_projection.reconcile_projections(op_dir, ledger_id, client)
        except Exception as e:
            print(f"Warning: Projection reconciliation failed: {e}", file=sys.stderr)

        conn_journal.close()
        print(f"Registered evidence '{evidence_id}' on task '{task_id}' via authority broker.")
        return 0

    claim_id = args.claim
    claim = None
    exec_identity = get_executor_identity()
    identity_policy = identity_policy_for_write(op_dir)
    if identity_policy is None:
        return 1
    verified_by_value = args.verified_by
    verification_authority = None

    if claim_id:
        resolved_claim = resolve_claim_record(op_dir, claim_id)
        if not resolved_claim:
            return 1
        claim_path, claim = resolved_claim
        if claim.get("task_id") != task_id:
            print(
                f"Error: Claim '{claim_id}' belongs to task '{claim.get('task_id')}', not "
                f"'{task_id}'.",
                file=sys.stderr,
            )
            return 1

    # A status changes claim trust state; reject a bare write so it cannot silently succeed and drop.
    if args.status and not claim:
        print(
            "Error: --status requires --claim. A status changes trust state on a claim; bare evidence "
            "write cannot carry one (the status would be silently dropped).",
            file=sys.stderr,
        )
        return 1

    if args.status:
        if not getattr(args, "verified_by", None):
            print(
                "Error: --verified-by is required when setting --status (the verifier must identify themselves).",
                file=sys.stderr,
            )
            return 1

        if identity_policy["mode"] == "enforced":
            actual_verifier = require_executor_role(
                identity_policy, exec_identity, "verifier", "record verification"
            )
            if not actual_verifier:
                return 1
            if args.verified_by != actual_verifier["name"]:
                print(
                    f"Error: --verified-by '{args.verified_by}' does not match the executing "
                    f"identity '{actual_verifier['name']}'",
                    file=sys.stderr,
                )
                return 1

            author_executor = claim.get("author_executor")
            author_uid = author_executor.get("uid") if isinstance(author_executor, dict) else None
            if not isinstance(author_uid, int) or isinstance(author_uid, bool) or author_uid < 0:
                print(
                    "Error: refusing to record verification: claim has no valid author UID; "
                    "legacy authorship cannot be inferred",
                    file=sys.stderr,
                )
                return 1
            if author_uid == exec_identity["uid"]:
                print(
                    "Error: refusing to record verification: verifier UID must differ from the "
                    f"claim author UID ({author_uid})",
                    file=sys.stderr,
                )
                return 1

            verified_by_value = actual_verifier["name"]
            verification_authority = "uid_isolated"
        else:
            verified_by_value = args.verified_by
            verification_authority = "advisory"
    elif identity_policy["mode"] == "enforced" and not require_executor_role(
        identity_policy, exec_identity, "builder", "attach evidence"
    ):
        return 1

    expected_hash = None
    if args.hash is not None:
        try:
            expected_hash = normalize_sha256(args.hash)
        except EvidenceFingerprintError as exc:
            print(f"Error: {exc}", file=sys.stderr)
            return 1

    original_locator = args.path_or_url
    path_or_url = original_locator
    source_fingerprint = None
    source_path = None
    source_kind = None
    fingerprint = None
    source = None
    provenance = "url_or_external"

    if os.path.exists(original_locator):
        if not os.path.isfile(original_locator):
            print(
                f"Error: Evidence path is not a regular file: {original_locator}",
                file=sys.stderr,
            )
            return 1
        source_path = os.path.abspath(original_locator)
        try:
            source_fingerprint = fingerprint_regular_file(source_path)
        except EvidenceFingerprintError as exc:
            print(f"Error: {exc}", file=sys.stderr)
            return 1
        if expected_hash and expected_hash != source_fingerprint["value"]:
            print(
                "Error: Supplied SHA-256 does not match local evidence bytes "
                f"(expected {expected_hash}, got {source_fingerprint['value']})",
                file=sys.stderr,
            )
            return 1
    else:
        scheme_match = re.match(r"^([A-Za-z][A-Za-z0-9+.-]*):", original_locator)
        scheme = scheme_match.group(1).lower() if scheme_match else None
        if scheme in {"http", "https"}:
            source_kind = "remote_url"
        elif scheme and scheme != "file":
            source_kind = "external_reference"
        else:
            print(f"Error: Local evidence file does not exist: {original_locator}", file=sys.stderr)
            return 1

    preflight_projection(op_dir, task_path, task)
    if claim:
        preflight_projection(op_dir, claim_path, claim)

    evidence_id = get_next_evidence_id(op_dir, task_id)

    if source_path:

        dest_dir = os.path.join(op_dir, "evidence", task_id)
        os.makedirs(dest_dir, exist_ok=True)
        ext = os.path.splitext(source_path)[1]
        artifact_name = (
            f"artifact-{evidence_id}{ext}" if ext in {".yaml", ".yml"} else f"{evidence_id}{ext}"
        )
        dest_path = os.path.join(dest_dir, artifact_name)
        if os.path.lexists(dest_path):
            print(
                f"Error: Evidence artifact destination already exists: {dest_path}", file=sys.stderr
            )
            return 1

        temp_fd, temp_path = tempfile.mkstemp(prefix=f".{artifact_name}.", dir=dest_dir)
        os.close(temp_fd)
        try:
            shutil.copy2(source_path, temp_path)
            fingerprint = fingerprint_regular_file(temp_path)
            if fingerprint["value"] != source_fingerprint["value"]:
                raise EvidenceFingerprintError(
                    f"Evidence file changed while being persisted: {source_path}"
                )
            os.link(temp_path, dest_path)
        except (OSError, EvidenceFingerprintError) as exc:
            print(f"Error: Could not persist evidence file '{source_path}': {exc}", file=sys.stderr)
            return 1
        finally:
            try:
                os.unlink(temp_path)
            except FileNotFoundError:
                pass

        path_or_url = os.path.abspath(dest_path)
        provenance = "local_file"
        source = {
            "kind": "local_file",
            "locator": source_path,
            "fingerprint": source_fingerprint,
        }
    else:
        source = {"kind": source_kind, "locator": original_locator}
        if expected_hash:
            fingerprint = {
                "algorithm": "sha256",
                "value": expected_hash,
                "size_bytes": None,
                "mtime_ns": None,
            }

    file_hash = fingerprint["value"] if fingerprint else None
    made_by = args.by if args.by else task.get("assigned_harness") or "operator"
    utc_now = get_current_utc_iso()

    evidence_data = {
        "evidence_id": evidence_id,
        "task_id": task_id,
        "claim_id": claim_id,
        "path_or_url": path_or_url,
        "evidence_type": args.type,
        "produced_by": made_by,
        "produced_at": utc_now,
        "provenance": provenance,
        "hash": file_hash,
        "fingerprint": fingerprint,
        "source": source,
        "verification_command": args.verify_cmd,
        "notes": args.notes,
        "executor": exec_identity,
    }
    crystal_metadata = getattr(args, "crystal_metadata", None)
    if isinstance(crystal_metadata, dict):
        for key, value in crystal_metadata.items():
            if key and key not in evidence_data:
                evidence_data[key] = value

    ev_path = os.path.join(op_dir, "evidence", task_id, f"{evidence_id}.yaml")
    save_yaml(evidence_data, ev_path)

    task["evidence"] = task.get("evidence") or []
    if path_or_url not in task["evidence"]:
        task["evidence"].append(path_or_url)
    task["updated_at"] = utc_now
    save_yaml(task, task_path)

    if claim:
        claim["evidence_refs"] = claim.get("evidence_refs") or []
        ref_str = f"evidence/{task_id}/{evidence_id}.yaml"
        if ref_str not in claim["evidence_refs"]:
            claim["evidence_refs"].append(ref_str)

        if args.status:
            claim["verified_by"] = verified_by_value
            claim["verification_outcome"] = args.status
            claim["verification_executor"] = dict(exec_identity)
            claim["verification_authority"] = verification_authority
            claim["verification_mode"] = identity_policy["mode"]
            if args.status == "verified":
                claim["verification_status"] = True
            else:
                claim["verification_status"] = False

            claim["executor"] = exec_identity

        if args.verdict:
            claim["verdict"] = args.verdict

        save_yaml(claim, claim_path)

        if args.status == "quarantined":
            task["status"] = "quarantined"
            save_yaml(task, task_path)
        elif args.status == "verified" and task["status"] not in [
            "quarantined",
            "complete",
        ]:
            task["status"] = "verified"
            save_yaml(task, task_path)

    print(f"Attached evidence '{evidence_id}' to task '{task_id}'.")
    if claim_id:
        print(
            f"Associated with claim '{claim_id}' (Verification status: {claim.get('verification_status')})."
        )
    return 0


def crystal_attach_cmd(args) -> int:
    """Attach a session crystal as draft session_crystal evidence (Phase 1).

    Never accepts verification status (T2). Validates heading structure before
    any ledger write (T6). Does not execute crystal body content (T3).
    """
    # Fail closed if a future caller smuggles status onto the namespace.
    if getattr(args, "status", None) is not None:
        print(
            "Error: crystal-attach does not accept verification status "
            "(T2 — use evidence-attach --status as a distinct verifier on the "
            "already-attached crystal).",
            file=sys.stderr,
        )
        return 1
    if getattr(args, "verified_by", None) is not None or getattr(args, "verdict", None) is not None:
        print(
            "Error: crystal-attach does not accept --verified-by or --verdict "
            "(verification is a separate verifier action).",
            file=sys.stderr,
        )
        return 1

    original_locator = args.path
    if not os.path.exists(original_locator):
        print(f"Error: Crystal path does not exist: {original_locator}", file=sys.stderr)
        return 1
    if not os.path.isfile(original_locator):
        print(f"Error: Crystal path is not a regular file: {original_locator}", file=sys.stderr)
        return 1

    parsed = crystal_parse.load_and_parse_crystal(original_locator)
    for warning in parsed.warnings:
        print(f"[Warning] {warning}")
    if not parsed.ok:
        for err in parsed.errors:
            print(f"Error: crystal rejected (structural): {err}", file=sys.stderr)
        return 1

    crystal_metadata = {
        "crystal_kind": parsed.kind,
        "crystal_project": parsed.project,
        "crystal_observed_at": parsed.observed_at,
        "crystal_source_commit": parsed.git_commit,
        "crystal_source_branch": parsed.git_branch,
        "crystal_test_bullet_count": parsed.test_bullet_count,
        "crystal_upstream_pin": crystal_parse.CRYSTAL_UPSTREAM_PIN,
    }

    attach_args = argparse.Namespace(
        path_or_url=original_locator,
        claim=args.claim,
        type="session_crystal",
        task_id=args.task_id,
        by=args.by,
        hash=args.hash,
        notes=args.notes,
        verify_cmd=None,
        status=None,
        verdict=None,
        verified_by=None,
        crystal_metadata=crystal_metadata,
    )
    rc = evidence_attach_cmd(attach_args)
    if rc == 0:
        print(
            "Crystal metadata captured: "
            f"kind={parsed.kind}, project={parsed.project}, "
            f"commit={parsed.git_commit}, test_bullets={parsed.test_bullet_count}"
        )
    return rc


def find_session_crystal_by_fingerprint(
    operator_dir: str, task_id: str, sha256_value: str
) -> dict | None:
    """Return existing session_crystal evidence dict if fingerprint already on task."""
    ev_dir = os.path.join(operator_dir, "evidence", task_id)
    if not os.path.isdir(ev_dir):
        return None
    for path in sorted(glob.glob(os.path.join(ev_dir, "evidence-*.yaml"))):
        data = load_yaml(path)
        if not data or data.get("evidence_type") != "session_crystal":
            continue
        stored, _ = parse_stored_fingerprint(data.get("fingerprint"), data.get("hash"))
        if stored and stored.get("value") == sha256_value:
            return data
    return None


def crystal_import_cmd(args) -> int:
    """Extract draft claims from a crystal (Phase 2). Always draft-tier (T2).

    - Tests And Verification real bullets → draft test_passes claims prefixed
      ``[crystal-narrated]``, crystal auto-attached as status-less evidence.
    - Open Loops → new tasks only with ``--open-loops-as-tasks``.
    - Decisions / Findings / Memory Candidates / Resume Prompt / Continuity Tail
      are never extracted (T3 + Phase 2 scope).
    - Same fingerprint re-import is a no-op with a notice.
    """
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    enrollment, _ = resolve_operator_context()
    if enrollment:
        print(
            "Error: crystal-import Phase 2 supports local (non-enrolled) ledgers only.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    exec_identity = get_executor_identity()
    identity_policy = identity_policy_for_write(op_dir)
    if identity_policy is None:
        return 1
    if identity_policy["mode"] == "enforced" and not require_executor_role(
        identity_policy, exec_identity, "builder", "import crystal"
    ):
        return 1

    original_locator = args.path
    if not os.path.exists(original_locator):
        print(f"Error: Crystal path does not exist: {original_locator}", file=sys.stderr)
        return 1
    if not os.path.isfile(original_locator):
        print(f"Error: Crystal path is not a regular file: {original_locator}", file=sys.stderr)
        return 1

    try:
        source_fp = fingerprint_regular_file(os.path.abspath(original_locator))
    except EvidenceFingerprintError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    existing = find_session_crystal_by_fingerprint(op_dir, task_id, source_fp["value"])
    if existing:
        print(
            f"Notice: crystal fingerprint {source_fp['value'][:12]}… already imported as "
            f"{existing.get('evidence_id')} on task '{task_id}'; no-op (idempotent)."
        )
        return 0

    parsed = crystal_parse.load_and_parse_crystal(original_locator)
    for warning in parsed.warnings:
        print(f"[Warning] {warning}")
    if not parsed.ok:
        for err in parsed.errors:
            print(f"Error: crystal rejected (structural): {err}", file=sys.stderr)
        return 1

    # Attach once as shared draft evidence (no claim_id so multiple claims may cite it).
    attach_args = argparse.Namespace(
        path=original_locator,
        claim=None,
        task_id=task_id,
        by=args.by,
        hash=None,
        notes=args.notes or "crystal-import auto-attach",
        status=None,
        verified_by=None,
        verdict=None,
    )
    rc = crystal_attach_cmd(attach_args)
    if rc != 0:
        return rc

    attached = find_session_crystal_by_fingerprint(op_dir, task_id, source_fp["value"])
    if not attached:
        print(
            "Error: crystal-attach succeeded but evidence record was not found.",
            file=sys.stderr,
        )
        return 1
    evidence_id = attached["evidence_id"]
    ref_str = f"evidence/{task_id}/{evidence_id}.yaml"

    # Never pull Decisions / Findings / Resume Prompt / Continuity Tail into claims.
    created_claims: list[str] = []
    for bullet in parsed.test_bullets:
        claim_args = argparse.Namespace(
            type="test_passes",
            text=f"[crystal-narrated] {bullet}",
            task_id=task_id,
            by=args.by,
            gate=None,
            layer=None,
        )
        claim_rc = claim_add_cmd(claim_args)
        if claim_rc != 0:
            print(
                f"Error: failed creating draft claim for crystal bullet: {bullet!r}",
                file=sys.stderr,
            )
            return claim_rc
        # Reload task to pick up the claim id just registered.
        task = load_yaml(task_path) or {}
        claims_list = task.get("claims") or []
        if not claims_list:
            print("Error: claim-add did not register a claim on the task.", file=sys.stderr)
            return 1
        claim_id = claims_list[-1]
        resolved_claim = resolve_claim_record(op_dir, claim_id)
        if not resolved_claim:
            return 1
        claim_path, claim = resolved_claim
        claim["evidence_refs"] = claim.get("evidence_refs") or []
        if ref_str not in claim["evidence_refs"]:
            claim["evidence_refs"].append(ref_str)
        # Keep draft: never set verification fields from crystal text.
        claim["crystal_import_evidence"] = evidence_id
        save_yaml(claim, claim_path)
        created_claims.append(claim_id)

    created_tasks: list[str] = []
    if getattr(args, "open_loops_as_tasks", False):
        for loop_text in parsed.open_loop_bullets:
            loop_id = slugify(loop_text)[:50] or None
            create_args = argparse.Namespace(
                objective=f"[crystal-open-loop] {loop_text}",
                task_id=loop_id,
                repo=task.get("repo") or ".",
                assign=None,
                review=None,
                assumptions=[f"Extracted from crystal {evidence_id}; may be stale."],
            )
            # Avoid collisions if slug already exists.
            if loop_id:
                existing_path = os.path.join(op_dir, "tasks", f"{loop_id}.yaml")
                if os.path.exists(existing_path) or durable_record_exists(op_dir, "task", loop_id):
                    create_args.task_id = None
            create_rc = task_create_cmd(create_args)
            if create_rc != 0:
                print(
                    f"Warning: could not create task for open loop: {loop_text!r}",
                    file=sys.stderr,
                )
                continue
            # task-create sets current_task; recover id from operator.yaml
            config = load_yaml(os.path.join(op_dir, "operator.yaml")) or {}
            new_id = config.get("current_task")
            if new_id:
                created_tasks.append(new_id)
        # Restore original current task after optional open-loop task creation.
        config = load_yaml(os.path.join(op_dir, "operator.yaml")) or {}
        config["current_task"] = task_id
        save_yaml(config, os.path.join(op_dir, "operator.yaml"))

    print(
        f"Crystal import on task '{task_id}': evidence={evidence_id}, "
        f"draft_claims={len(created_claims)}"
        + (f", open_loop_tasks={len(created_tasks)}" if created_tasks else "")
    )
    if created_claims:
        print("  claims: " + ", ".join(created_claims))
    if created_tasks:
        print("  open-loop tasks: " + ", ".join(created_tasks))
    if not created_claims and not created_tasks:
        print(
            "  (no real test bullets to extract; crystal attached as draft evidence only)"
        )
    return 0


CRYSTAL_BRIDGE_EVENTS = (
    "SessionStart",
    "UserPromptSubmit",
    "PostToolUse",
    "PostToolBatch",
    "PreCompact",
    "PostCompact",
    "Stop",
)

CRYSTAL_HARNESS_ALIASES = {
    "claude-code": "claude",
    "claude": "claude",
    "codex": "codex",
    "gemini-agy": "gemini-agy",
    "gemini": "gemini-agy",
}


def map_crystal_harness(name: str | None) -> str | None:
    """Map crystallize/hook harness names to Operator registry ids."""
    if not name:
        return None
    key = name.strip().lower()
    if key in CRYSTAL_HARNESS_ALIASES:
        return CRYSTAL_HARNESS_ALIASES[key]
    return name.strip()


def find_latest_session_crystal(repo_path: str) -> str | None:
    """Newest session crystal under repo/.agent-crystals/sessions/*.md by mtime."""
    if not repo_path:
        return None
    sessions_dir = os.path.join(os.path.abspath(repo_path), ".agent-crystals", "sessions")
    if not os.path.isdir(sessions_dir):
        return None
    candidates = [
        p
        for p in glob.glob(os.path.join(sessions_dir, "*.md"))
        if os.path.isfile(p)
    ]
    if not candidates:
        return None
    candidates.sort(key=lambda p: os.path.getmtime(p), reverse=True)
    return candidates[0]


def resolve_attach_crystal_path(
    attach_crystal: str, repo_path: str | None
) -> tuple[str | None, str | None]:
    """Resolve auto|PATH for session crystal attach.

    Returns (path, error_message). error_message set when resolution fails hard.
    """
    if not attach_crystal or not str(attach_crystal).strip():
        return None, "empty --attach-crystal value"
    value = str(attach_crystal).strip()
    if value.lower() == "auto":
        if not repo_path:
            return None, "cannot resolve auto crystal: task has no repo path"
        path = find_latest_session_crystal(repo_path)
        if not path:
            return None, (
                f"no session crystal found under {os.path.join(repo_path, '.agent-crystals', 'sessions')}"
            )
        return path, None
    path = os.path.abspath(value)
    if not os.path.isfile(path):
        return None, f"crystal path is not a regular file: {path}"
    return path, None


def attach_session_crystal_for_task(
    op_dir: str,
    task_id: str,
    crystal_path: str,
    *,
    by: str | None = None,
    notes: str | None = None,
) -> tuple[int, str | None]:
    """Attach crystal as draft evidence; idempotent on fingerprint.

    Returns (exit_code, evidence_id_or_None).
    """
    try:
        source_fp = fingerprint_regular_file(os.path.abspath(crystal_path))
    except EvidenceFingerprintError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1, None

    existing = find_session_crystal_by_fingerprint(op_dir, task_id, source_fp["value"])
    if existing:
        print(
            f"Notice: crystal fingerprint {source_fp['value'][:12]}… already attached as "
            f"{existing.get('evidence_id')} on task '{task_id}'; skipping duplicate attach."
        )
        return 0, existing.get("evidence_id")

    attach_args = argparse.Namespace(
        path=crystal_path,
        claim=None,
        task_id=task_id,
        by=by,
        hash=None,
        notes=notes or "crystal-bridge session attach",
        status=None,
        verified_by=None,
        verdict=None,
    )
    rc = crystal_attach_cmd(attach_args)
    if rc != 0:
        return rc, None
    attached = find_session_crystal_by_fingerprint(op_dir, task_id, source_fp["value"])
    return 0, (attached or {}).get("evidence_id")


def tag_usage_crystal_bridge(op_dir: str, usage_id: str, bridge_info: dict) -> None:
    """Merge crystal_bridge metadata onto an existing usage record (best-effort)."""
    filepath, records, idx = find_usage_record(op_dir, usage_id)
    if not filepath or records is None or idx is None:
        return
    rec = records[idx]
    existing = rec.get("crystal_bridge")
    if isinstance(existing, dict):
        merged = dict(existing)
        merged.update(bridge_info)
        rec["crystal_bridge"] = merged
    else:
        rec["crystal_bridge"] = bridge_info
    rec["executor"] = get_executor_identity()
    save_yaml(records, filepath)


def crystal_bridge_cmd(args) -> int:
    """Phase 3 hook/session bridge (see CRYSTAL_SESSION_BRIDGE_SPEC.md).

    Soft-fails (exit 0) when no ledger is present so harness hooks stay safe.
    Never sets verification status (T2). Never writes .agent-crystals/ (T1).
    """
    event = getattr(args, "event", None) or "SessionStart"
    if event not in CRYSTAL_BRIDGE_EVENTS:
        print(
            f"Error: unknown bridge event '{event}'; expected one of "
            f"{', '.join(CRYSTAL_BRIDGE_EVENTS)}",
            file=sys.stderr,
        )
        return 1

    # Fail closed if status smuggled onto namespace.
    if getattr(args, "status", None) is not None:
        print(
            "Error: crystal-bridge does not accept verification status (T2).",
            file=sys.stderr,
        )
        return 1

    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Notice: crystal-bridge: no .operator ledger here; soft no-op "
            f"(event={event})."
        )
        return 0

    # Events other than SessionStart/Stop are crystallize-owned.
    if event not in {"SessionStart", "Stop"}:
        print(f"Notice: crystal-bridge: event {event} is a soft no-op (crystallize-owned).")
        return 0

    if event == "SessionStart":
        harness_id = map_crystal_harness(getattr(args, "harness", None))
        if not harness_id:
            print(
                "Error: crystal-bridge SessionStart requires --harness "
                "(e.g. claude, codex, claude-code).",
                file=sys.stderr,
            )
            return 1
        harness_path = os.path.join(op_dir, "harnesses", f"{harness_id}.yaml")
        if not os.path.exists(harness_path):
            print(
                f"Error: harness '{harness_id}' not registered under .operator/harnesses/.",
                file=sys.stderr,
            )
            return 1

        resolved_task = resolve_task_record(op_dir, getattr(args, "task_id", None))
        if not resolved_task:
            if getattr(args, "task_id", None):
                return 1  # explicit task missing → hard fail
            print(
                "Notice: crystal-bridge SessionStart: no current task; soft no-op."
            )
            return 0
        task_id, _task_path, task = resolved_task

        if task.get("status") == "running" and not getattr(args, "force", False):
            open_ids = get_open_usage_ids_for_task(op_dir, task_id)
            print(
                f"Notice: crystal-bridge SessionStart: task '{task_id}' already running "
                f"(open usage: {', '.join(open_ids) or 'none'}); soft no-op."
            )
            return 0

        start_args = argparse.Namespace(
            task_id=task_id,
            harness_id=harness_id,
            force=bool(getattr(args, "force", False)),
            lane=getattr(args, "lane", None),
            task_class=getattr(args, "task_class", None),
        )
        rc = session_start_cmd(start_args)
        if rc != 0:
            return rc

        open_ids = get_open_usage_ids_for_task(op_dir, task_id)
        usage_id = open_ids[-1] if open_ids else None
        if usage_id:
            tag_usage_crystal_bridge(
                op_dir,
                usage_id,
                {
                    "event": "SessionStart",
                    "harness": harness_id,
                    "bridged_at": get_current_utc_iso(),
                },
            )
            print(f"crystal-bridge: SessionStart linked usage '{usage_id}' on task '{task_id}'.")
        return 0

    # Stop — attach latest session crystal; do not close usage.
    resolved_task = resolve_task_record(op_dir, getattr(args, "task_id", None))
    if not resolved_task:
        if getattr(args, "task_id", None):
            return 1
        print("Notice: crystal-bridge Stop: no current task; soft no-op.")
        return 0
    task_id, _task_path, task = resolved_task
    repo_path = task.get("repo")

    crystal_arg = getattr(args, "crystal", None) or "auto"
    crystal_path, err = resolve_attach_crystal_path(crystal_arg, repo_path)
    if err or not crystal_path:
        msg = err or "crystal path unresolved"
        if getattr(args, "require_crystal", False):
            print(f"Error: crystal-bridge Stop: {msg}", file=sys.stderr)
            return 1
        print(f"Notice: crystal-bridge Stop: {msg}; soft no-op.")
        return 0

    rc, evidence_id = attach_session_crystal_for_task(
        op_dir,
        task_id,
        crystal_path,
        by=getattr(args, "by", None),
        notes="crystal-bridge Stop attach",
    )
    if rc != 0:
        return rc

    open_ids = get_open_usage_ids_for_task(op_dir, task_id)
    if open_ids and evidence_id:
        tag_usage_crystal_bridge(
            op_dir,
            open_ids[-1],
            {
                "event": "Stop",
                "attached_at": get_current_utc_iso(),
                "crystal_path": os.path.abspath(crystal_path),
                "evidence_id": evidence_id,
            },
        )
    print(
        f"crystal-bridge: Stop attached {os.path.basename(crystal_path)} "
        f"as {evidence_id or 'evidence'} on task '{task_id}'."
    )
    return 0


def git_commit_exists_in_repo(repo_path: str, commit: str) -> bool | None:
    """Return True/False if git can resolve commit; None if repo unusable.

    Uses fixed argv only — commit is a single argument, never shell-interpolated.
    """
    if not repo_path or not commit or not isinstance(commit, str):
        return None
    commit = commit.strip()
    if not commit or commit.lower() in {"unknown", "unavailable", "(none)", "none"}:
        return None
    if not os.path.isdir(repo_path):
        return None
    try:
        probe = subprocess.run(
            ["git", "-C", repo_path, "rev-parse", "--is-inside-work-tree"],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
        if probe.returncode != 0 or probe.stdout.strip() != "true":
            return None
        # ^{commit} ensures we resolve an object that is a commit.
        check = subprocess.run(
            ["git", "-C", repo_path, "cat-file", "-e", f"{commit}^{{commit}}"],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
        return check.returncode == 0
    except (OSError, subprocess.TimeoutExpired):
        return None


def usage_add_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task
    preflight_projection(op_dir, task_path, task)

    raw_payload = ""
    if args.file and args.file != "-":
        # Read the payload verbatim from a file. This is the fidelity-preserving
        # capture path: unlike piping through `echo "...$0..."`, file contents are
        # never subject to shell quoting/variable expansion (the "bash.00" trap).
        if not os.path.exists(args.file):
            print(f"Error: Payload file '{args.file}' not found.", file=sys.stderr)
            return 1
        try:
            with open(args.file, "r") as f:
                raw_payload = f.read()
        except Exception as e:
            print(f"Error reading payload file '{args.file}': {e}", file=sys.stderr)
            return 1
    else:
        if sys.stdin.isatty():
            print("Paste your usage snapshot / status / quota snippet (press Ctrl-D to save):")
        try:
            raw_payload = sys.stdin.read()
        except KeyboardInterrupt:
            print("\nAborted.")
            return 1

    usage_id = get_next_usage_id(op_dir)
    utc_now = get_current_utc_iso()
    date_str = utc_now.split("T")[0]

    usage_record = {
        "usage_id": usage_id,
        "task_id": task_id,
        "harness_id": args.harness,
        "model": args.model,
        "started_at": utc_now,
        "ended_at": utc_now,
        "cost_estimate_usd": args.cost,
        "outcome": args.outcome,
        "raw_payload": raw_payload.strip(),
        "executor": get_executor_identity(),
        "lane": getattr(args, "lane", None) or "unknown",
        "task_class": getattr(args, "task_class", None) or "unknown",
    }

    usage_file_path = os.path.join(op_dir, "usage", f"{date_str}.yaml")
    records = []
    if os.path.exists(usage_file_path):
        records = load_yaml(usage_file_path) or []

    records.append(usage_record)
    save_yaml(records, usage_file_path)

    task["updated_at"] = utc_now
    save_yaml(task, task_path)

    print(f"Usage record '{usage_id}' successfully added under {date_str}.yaml.")
    return 0


def parse_iso_to_datetime(ts_str: str) -> datetime.datetime | None:
    if not ts_str:
        return None
    ts_str = ts_str.strip()
    if ts_str.endswith("Z"):
        ts_str = ts_str[:-1] + "+00:00"
    try:
        return datetime.datetime.fromisoformat(ts_str)
    except Exception:
        pass
    m = re.match(
        r"^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2}):(\d{2})(?:\.(\d+))?(?:([+-]\d{2}):?(\d{2}))?$",
        ts_str,
    )
    if m:
        year, month, day, hour, minute, second, frac, tz_hour, tz_min = m.groups()
        microsecond = 0
        if frac:
            frac = frac[:6].ljust(6, "0")
            microsecond = int(frac)
        tz = datetime.timezone.utc
        if tz_hour:
            sign = -1 if tz_hour.startswith("-") else 1
            offset_mins = int(tz_hour) * 60 + sign * int(tz_min)
            tz = datetime.timezone(datetime.timedelta(minutes=offset_mins))
        return datetime.datetime(
            int(year),
            int(month),
            int(day),
            int(hour),
            int(minute),
            int(second),
            microsecond,
            tz,
        )
    try:
        return datetime.datetime.strptime(ts_str.split("+")[0].split("-")[0], "%Y-%m-%dT%H:%M:%S")
    except Exception:
        return None


def parse_glog_timestamp(line: str, default_year: int) -> datetime.datetime | None:
    m = re.match(r"^([IWEF])(\d{2})(\d{2}) (\d{2}):(\d{2}):(\d{2})\.(\d{6})", line)
    if m:
        _, month, day, hour, minute, second, micro = m.groups()
        dt = datetime.datetime(
            default_year,
            int(month),
            int(day),
            int(hour),
            int(minute),
            int(second),
            int(micro),
        )
        return dt.astimezone()
    return None


def parse_log_line_timestamp(line: str, default_year: int) -> datetime.datetime | None:
    dt = parse_glog_timestamp(line, default_year)
    if dt is not None:
        return dt
    parts = line.split()
    if parts:
        first_token = parts[0]
        dt = parse_iso_to_datetime(first_token)
        if dt is not None:
            return dt
        if len(parts) >= 2:
            dt = parse_iso_to_datetime(parts[0] + "T" + parts[1])
            if dt is not None:
                return dt
    return None


def load_pricing_table(op_dir: str) -> dict:
    pricing_path = os.path.join(op_dir, "pricing.yaml")
    if not os.path.exists(pricing_path):
        return {}
    return load_yaml(pricing_path) or {}


def calculate_cost(
    model: str | None,
    tokens_in: int,
    tokens_out: int,
    tokens_cache_read: int,
    tokens_cache_write: int,
    pricing_table: dict,
) -> float | None:
    if not model or model not in pricing_table:
        return None
    rates = pricing_table[model]
    cost = 0.0
    cost += tokens_in * rates.get("input", 0.0)
    cost += tokens_out * rates.get("output", 0.0)
    cost += tokens_cache_read * rates.get("cache_read", 0.0)
    cost += tokens_cache_write * rates.get("cache_write", 0.0)
    return cost / 1000000.0


def get_nested(d: dict, path: str, default: any = None) -> any:
    parts = path.split(".")
    for part in parts[:-1]:
        if not isinstance(d, dict) or part not in d:
            return default
        d = d[part]
    if isinstance(d, dict):
        return d.get(parts[-1], default)
    return default


def set_nested(d: dict, path: str, value: any) -> None:
    parts = path.split(".")
    for part in parts[:-1]:
        if part not in d or not isinstance(d[part], dict):
            d[part] = {}
        d = d[part]
    d[parts[-1]] = value


def set_record_field(record: dict, field_name: str, value: any, source: str = "auto") -> None:
    if "field_sources" not in record:
        record["field_sources"] = {}
    if "auto_values" not in record:
        record["auto_values"] = {}
    flat_key = field_name.split(".")[-1]
    if source == "auto":
        record["auto_values"][flat_key] = value
        if record["field_sources"].get(flat_key) != "manual":
            set_nested(record, field_name, value)
            record["field_sources"][flat_key] = "auto"
    else:
        set_nested(record, field_name, value)
        record["field_sources"][flat_key] = "manual"


def check_overlap(
    start1: datetime.datetime,
    end1: datetime.datetime | None,
    start2: datetime.datetime,
    end2: datetime.datetime | None,
) -> bool:
    t1_start = start1
    t1_end = (
        end1 if end1 is not None else datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
    )
    t2_start = start2
    t2_end = (
        end2 if end2 is not None else datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
    )

    # Standardise timezones to UTC for comparison
    if t1_start.tzinfo:
        t1_start = t1_start.astimezone(datetime.timezone.utc)
    if t1_end.tzinfo:
        t1_end = t1_end.astimezone(datetime.timezone.utc)
    if t2_start.tzinfo:
        t2_start = t2_start.astimezone(datetime.timezone.utc)
    if t2_end.tzinfo:
        t2_end = t2_end.astimezone(datetime.timezone.utc)

    return t1_start <= t2_end and t2_start <= t1_end


def is_within_window(
    dt: datetime.datetime, start: datetime.datetime, end: datetime.datetime
) -> bool:
    start_local = start.astimezone().replace(tzinfo=None) if start.tzinfo else start
    end_local = end.astimezone().replace(tzinfo=None) if end.tzinfo else end
    dt_naive = dt.replace(tzinfo=None) if dt.tzinfo else dt
    if start_local <= dt_naive <= end_local:
        return True
    start_naive = start.replace(tzinfo=None)
    end_naive = end.replace(tzinfo=None)
    if start_naive <= dt_naive <= end_naive:
        return True
    return False


def score_session(session: dict, abs_repo: str, task_id: str, harness_id: str) -> int:
    score = 0
    if harness_id in ("claude", "codex"):
        scwd = session.get("cwd")
        if scwd and os.path.abspath(scwd) == abs_repo:
            score += 5
    elif harness_id == "gemini-agy":
        sid = session.get("session_id")
        if sid and (task_id in sid or sid in task_id):
            score += 5
    return score


def parse_claude_session_file(filepath: str) -> dict | None:
    try:
        with open(filepath, "r") as f:
            lines = f.readlines()
    except Exception:
        return None
    timestamps = []
    model = None
    cwds = []
    turns = 0
    tool_calls = 0
    tokens_in = 0
    tokens_out = 0
    tokens_cache_read = 0
    tokens_cache_write = 0
    session_id = os.path.splitext(os.path.basename(filepath))[0]
    for line in lines:
        if not line.strip():
            continue
        try:
            data = json.loads(line)
            ts = data.get("timestamp")
            if ts:
                dt = parse_iso_to_datetime(ts)
                if dt:
                    timestamps.append(dt)
            scwd = data.get("cwd")
            if scwd:
                cwds.append(scwd)
            sid = data.get("sessionId")
            if sid:
                session_id = sid
            if data.get("type") == "assistant":
                turns += 1
                msg = data.get("message", {})
                if not model:
                    model = msg.get("model")
                usage = msg.get("usage", {})
                tokens_in += usage.get("input_tokens", 0)
                tokens_cache_write += usage.get("cache_creation_input_tokens", 0)
                tokens_cache_read += usage.get("cache_read_input_tokens", 0)
                tokens_out += usage.get("output_tokens", 0)
                content = msg.get("content", [])
                if isinstance(content, list):
                    for item in content:
                        if isinstance(item, dict) and item.get("type") == "tool_use":
                            tool_calls += 1
        except Exception:
            pass
    if not timestamps:
        return None
    timestamps.sort()
    start_time = timestamps[0]
    end_time = timestamps[-1]
    cwd = None
    if cwds:
        cwd = max(set(cwds), key=cwds.count)
    return {
        "session_id": session_id,
        "start_time": start_time,
        "end_time": end_time,
        "model": model or "unknown",
        "cwd": cwd,
        "turns": turns,
        "tool_calls": tool_calls,
        "tokens_in": tokens_in,
        "tokens_out": tokens_out,
        "tokens_cache_read": tokens_cache_read,
        "tokens_cache_write": tokens_cache_write,
        "source_session_ref": os.path.abspath(filepath),
        "metering": "tokens",
    }


def parse_codex_session_file(filepath: str) -> dict | None:
    try:
        with open(filepath, "r") as f:
            lines = f.readlines()
    except Exception:
        return None
    timestamps = []
    models = []
    cwds = []
    turns = 0
    tool_calls = 0
    last_token_event = None
    session_id = os.path.splitext(os.path.basename(filepath))[0]
    for line in lines:
        if not line.strip():
            continue
        try:
            data = json.loads(line)
            ts = data.get("timestamp")
            if ts:
                dt = parse_iso_to_datetime(ts)
                if dt:
                    timestamps.append(dt)
            ev_type = data.get("type")
            payload = data.get("payload", {})
            if ev_type == "session_meta":
                sid = payload.get("id")
                if sid:
                    session_id = sid
            elif ev_type == "turn_context":
                turns += 1
                model = payload.get("model")
                if model:
                    models.append(model)
                scwd = payload.get("cwd")
                if scwd:
                    cwds.append(scwd)
            elif ev_type == "response_item":
                ptype = payload.get("type")
                if ptype in {
                    "function_call",
                    "local_shell_call",
                    "custom_tool_call",
                    "mcp_tool_call",
                }:
                    tool_calls += 1
            elif ev_type == "event_msg" and payload.get("type") == "token_count":
                last_token_event = payload
        except Exception:
            pass
    if not timestamps:
        return None
    timestamps.sort()
    start_time = timestamps[0]
    end_time = timestamps[-1]
    tokens_in = 0
    tokens_out = 0
    tokens_cache_read = 0
    if last_token_event:
        info = last_token_event.get("info") or {}
        usage = info.get("total_token_usage") or last_token_event
        input_tokens = usage.get("input_tokens", 0)
        cached_input_tokens = usage.get("cached_input_tokens", 0)
        output_tokens = usage.get("output_tokens", 0)
        tokens_cache_read = cached_input_tokens
        tokens_in = input_tokens - cached_input_tokens
        tokens_out = output_tokens
    model = "unknown"
    if models:
        model = max(set(models), key=models.count)
    cwd = None
    if cwds:
        cwd = max(set(cwds), key=cwds.count)
    return {
        "session_id": session_id,
        "start_time": start_time,
        "end_time": end_time,
        "model": model,
        "cwd": cwd,
        "turns": turns,
        "tool_calls": tool_calls,
        "tokens_in": tokens_in,
        "tokens_out": tokens_out,
        "tokens_cache_read": tokens_cache_read,
        "tokens_cache_write": None,
        "source_session_ref": os.path.abspath(filepath),
        "metering": "tokens",
    }


def parse_gemini_session_file(filepath: str, base_dir: str) -> dict | None:
    try:
        with open(filepath, "r") as f:
            lines = f.readlines()
    except Exception:
        return None
    timestamps = []
    turns = 0
    tool_calls = 0
    session_id = "unknown"
    parts = filepath.split(os.sep)
    if "brain" in parts:
        idx = parts.index("brain")
        if idx + 1 < len(parts):
            session_id = parts[idx + 1]
    for line in lines:
        if not line.strip():
            continue
        try:
            data = json.loads(line)
            ts = data.get("created_at")
            if ts:
                dt = parse_iso_to_datetime(ts)
                if dt:
                    timestamps.append(dt)
            source = data.get("source", "")
            if source.lower() == "model":
                turns += 1
            calls = data.get("tool_calls", [])
            if isinstance(calls, list):
                tool_calls += len(calls)
        except Exception:
            pass
    if not timestamps:
        return None
    timestamps.sort()
    start_time = timestamps[0]
    end_time = timestamps[-1]
    quota_events = 0
    log_files = glob.glob(os.path.join(base_dir, "log", "cli-*.log")) + glob.glob(
        os.path.join(base_dir, "cli-*.log")
    )
    for lf in log_files:
        try:
            with open(lf, "r") as f:
                for line in f:
                    if "quota reached" in line.lower():
                        dt = parse_log_line_timestamp(line, start_time.year)
                        if dt and is_within_window(dt, start_time, end_time):
                            quota_events += 1
        except Exception:
            pass
    return {
        "session_id": session_id,
        "start_time": start_time,
        "end_time": end_time,
        "model": "configured-gemini-model",
        "cwd": None,
        "turns": turns,
        "tool_calls": tool_calls,
        "tokens_in": None,
        "tokens_out": None,
        "tokens_cache_read": None,
        "tokens_cache_write": None,
        "quota_events": quota_events,
        "source_session_ref": os.path.abspath(filepath),
        "metering": "activity",
    }


def find_usage_record_by_ref(
    op_dir: str, source_session_ref: str
) -> tuple[str, list, int] | tuple[None, None, None]:
    usage_dir = os.path.join(op_dir, "usage")
    if not os.path.isdir(usage_dir):
        return None, None, None
    for filepath in glob.glob(os.path.join(usage_dir, "*.yaml")):
        records = load_yaml(filepath) or []
        for idx, rec in enumerate(records):
            if rec.get("source_session_ref") == source_session_ref:
                return filepath, records, idx
    return None, None, None


def usage_import_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1
    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task
    preflight_projection(op_dir, task_path, task)
    abs_repo = os.path.abspath(task.get("repo", "."))
    harness_id = args.harness
    source_dir_override = getattr(args, "source_dir", None)
    if source_dir_override:
        source_dir_override = os.path.expanduser(source_dir_override)
        if not os.path.exists(source_dir_override):
            print(
                f"Error: --source-dir path does not exist: {source_dir_override}",
                file=sys.stderr,
            )
            return 1
    if harness_id == "claude":
        base_dir = source_dir_override or os.environ.get(
            "OPERATOR_TEST_CLAUDE_DIR", os.path.expanduser("~/.claude")
        )
        files = []
        if os.path.isfile(base_dir):
            if base_dir.endswith(".jsonl"):
                files.append(base_dir)
        elif os.path.isdir(base_dir):
            for root, _, filenames in os.walk(base_dir):
                for f in filenames:
                    if f.endswith(".jsonl"):
                        files.append(os.path.join(root, f))
        all_sessions = []
        for f in files:
            res = parse_claude_session_file(f)
            if res:
                all_sessions.append(res)
    elif harness_id == "codex":
        base_dir = source_dir_override or os.environ.get(
            "OPERATOR_TEST_CODEX_DIR", os.path.expanduser("~/.codex")
        )
        files = []
        if os.path.isfile(base_dir):
            if base_dir.endswith(".jsonl"):
                files.append(base_dir)
        elif os.path.isdir(base_dir):
            for root, _, filenames in os.walk(base_dir):
                for f in filenames:
                    if f.endswith(".jsonl"):
                        files.append(os.path.join(root, f))
        all_sessions = []
        for f in files:
            res = parse_codex_session_file(f)
            if res:
                all_sessions.append(res)
    elif harness_id == "gemini-agy":
        base_dir = source_dir_override or os.environ.get(
            "OPERATOR_TEST_GEMINI_DIR", os.path.expanduser("~/.gemini/antigravity-cli")
        )
        files = []
        if os.path.isfile(base_dir):
            if base_dir.endswith(".jsonl"):
                files.append(base_dir)
        elif os.path.isdir(base_dir):
            for root, _, filenames in os.walk(base_dir):
                for f in filenames:
                    if f.endswith("transcript.jsonl"):
                        files.append(os.path.join(root, f))
        all_sessions = []
        for f in files:
            res = parse_gemini_session_file(f, base_dir)
            if res:
                all_sessions.append(res)
    else:
        print(f"Error: Unknown harness '{harness_id}'", file=sys.stderr)
        return 1

    op_start = None
    op_end = None
    matching_records = []
    usage_dir = os.path.join(op_dir, "usage")
    if os.path.isdir(usage_dir):
        for usage_path in glob.glob(os.path.join(usage_dir, "*.yaml")):
            records = load_yaml(usage_path) or []
            for rec in records:
                if rec.get("task_id") == task_id and rec.get("harness_id") == harness_id:
                    matching_records.append(rec)
    if matching_records:
        matching_records.sort(key=lambda r: r.get("started_at", ""), reverse=True)
        latest_rec = matching_records[0]
        op_start = parse_iso_to_datetime(latest_rec.get("started_at"))
        op_end = parse_iso_to_datetime(latest_rec.get("ended_at"))
        if op_end is None:
            op_end = datetime.datetime.now(datetime.timezone.utc)

    selected_sessions = []
    if args.session_id:
        for s in all_sessions:
            if s["session_id"] == args.session_id or args.session_id in s.get(
                "source_session_ref", ""
            ):
                selected_sessions.append(s)
    elif args.since or args.until:
        since_dt = (
            parse_iso_to_datetime(args.since)
            if args.since
            else datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
        )
        until_dt = (
            parse_iso_to_datetime(args.until)
            if args.until
            else datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
        )
        for s in all_sessions:
            if since_dt <= s["start_time"] <= until_dt:
                selected_sessions.append(s)
    elif op_start is not None:
        for s in all_sessions:
            if check_overlap(s["start_time"], s["end_time"], op_start, op_end):
                selected_sessions.append(s)
    else:
        if all_sessions:
            scored_sessions = []
            for s in all_sessions:
                score = score_session(s, abs_repo, task_id, harness_id)
                scored_sessions.append((score, s["end_time"] or s["start_time"], s))
            scored_sessions.sort(key=lambda item: (item[0], item[1]), reverse=True)
            selected_sessions = [scored_sessions[0][2]]

    if not selected_sessions:
        print("No matching session found to import.")
        return 0

    pricing_table = load_pricing_table(op_dir)
    utc_now = get_current_utc_iso()

    for s in selected_sessions:
        ref = s["source_session_ref"]
        filepath, records, idx = find_usage_record_by_ref(op_dir, ref)
        rec = None
        if filepath is not None:
            rec = records[idx]
        else:
            if op_start is not None:
                # Find open placeholder record overlapping this session
                for u_path in glob.glob(os.path.join(usage_dir, "*.yaml")):
                    u_recs = load_yaml(u_path) or []
                    for u_idx, u_rec in enumerate(u_recs):
                        if (
                            u_rec.get("task_id") == task_id
                            and u_rec.get("harness_id") == harness_id
                            and u_rec.get("source_session_ref") is None
                            and u_rec.get("ended_at") is None
                        ):
                            u_start = parse_iso_to_datetime(u_rec.get("started_at"))
                            u_end = datetime.datetime.now(datetime.timezone.utc)
                            if check_overlap(s["start_time"], s["end_time"], u_start, u_end):
                                filepath = u_path
                                records = u_recs
                                idx = u_idx
                                rec = u_rec
                                break
                    if rec:
                        break
            if rec is None:
                usage_id = get_next_usage_id(op_dir)
                rec = {
                    "usage_id": usage_id,
                    "task_id": task_id,
                    "harness_id": harness_id,
                    "outcome": "unknown",
                    "raw_payload": f"Auto-imported from {ref}",
                }

        # Lane is an explicit economics tag, not a harness-brand inference.
        field_sources = rec.get("field_sources", {})
        if rec.get("lane") is None or (
            rec.get("lane") == "unknown" and "lane" not in field_sources
        ):
            set_record_field(rec, "lane", "unknown", "auto")
        elif "lane" not in field_sources:
            set_record_field(rec, "lane", rec.get("lane"), "manual")

        if rec.get("task_class") is None or rec.get("task_class") == "unknown":
            set_record_field(rec, "task_class", "unknown", "auto")
        elif "task_class" not in rec.get("field_sources", {}):
            set_record_field(rec, "task_class", rec.get("task_class"), "manual")

        cost = calculate_cost(
            s["model"],
            s["tokens_in"] or 0,
            s["tokens_out"] or 0,
            s["tokens_cache_read"] or 0,
            s["tokens_cache_write"] or 0,
            pricing_table,
        )

        rec["metering"] = s["metering"]
        rec["source_session_ref"] = ref

        start_iso = (
            s["start_time"].astimezone(datetime.timezone.utc).replace(microsecond=0).isoformat()
        )
        end_iso = (
            s["end_time"].astimezone(datetime.timezone.utc).replace(microsecond=0).isoformat()
            if s["end_time"]
            else None
        )

        set_record_field(rec, "started_at", start_iso, "auto")
        set_record_field(rec, "ended_at", end_iso, "auto")
        set_record_field(rec, "model", s["model"], "auto")

        if s["metering"] == "tokens":
            set_record_field(rec, "tokens_in", s["tokens_in"], "auto")
            set_record_field(rec, "tokens_out", s["tokens_out"], "auto")
            set_record_field(rec, "tokens_cache_read", s["tokens_cache_read"], "auto")
            set_record_field(rec, "tokens_cache_write", s["tokens_cache_write"], "auto")
            set_record_field(rec, "cost_estimate_usd", cost, "auto")
        else:
            rec["tokens_in"] = None
            rec["tokens_out"] = None
            rec["tokens_cache_read"] = None
            rec["tokens_cache_write"] = None
            rec["cost_estimate_usd"] = None

        set_record_field(rec, "activity.tool_calls", s["tool_calls"], "auto")
        set_record_field(rec, "activity.turns", s["turns"], "auto")

        wall_clock = 0
        if s["end_time"] and s["start_time"]:
            wall_clock = int((s["end_time"] - s["start_time"]).total_seconds())
        set_record_field(rec, "activity.wall_clock_s", wall_clock, "auto")

        if s["metering"] == "activity" and s.get("quota_events") is not None:
            set_record_field(rec, "activity.quota_events", s["quota_events"], "auto")
        else:
            if "quota_events" not in rec.get("activity", {}):
                set_nested(rec, "activity.quota_events", None)

        if "active_s" not in rec.get("activity", {}):
            set_nested(rec, "activity.active_s", None)

        rec["executor"] = get_executor_identity()
        if getattr(args, "machine", None):
            # Logs synced from another machine: provenance is the producer, not the importer.
            rec["executor"]["machine"] = args.machine
            rec["executor"]["machine_source"] = "manual"
        if args.dry_run:
            print("---")
            print(f"# Dry-run import for session: {ref}")
            print(yaml.safe_dump(rec, default_flow_style=False, sort_keys=False))
        else:
            if filepath:
                records[idx] = rec
                save_yaml(records, filepath)
            else:
                date_str = s["start_time"].astimezone(datetime.timezone.utc).strftime("%Y-%m-%d")
                u_path = os.path.join(op_dir, "usage", f"{date_str}.yaml")
                u_recs = []
                if os.path.exists(u_path):
                    u_recs = load_yaml(u_path) or []
                u_recs.append(rec)
                save_yaml(u_recs, u_path)

            if task:
                task["updated_at"] = utc_now
                save_yaml(task, task_path)
            print(f"Successfully imported usage record '{rec['usage_id']}' for task '{task_id}'.")

    return 0


def usage_annotate_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1
    usage_id = args.usage_id
    filepath, records, idx = find_usage_record(op_dir, usage_id)
    if not filepath:
        print(f"Error: Usage record '{usage_id}' not found.", file=sys.stderr)
        return 1
    usage_record = records[idx]
    if (args.active_start and not args.active_end) or (args.active_end and not args.active_start):
        print(
            "Error: Both --active-start and --active-end must be specified to calculate active_s.",
            file=sys.stderr,
        )
        return 1
    if args.active_start and args.active_end:
        start_dt = parse_iso_to_datetime(args.active_start)
        end_dt = parse_iso_to_datetime(args.active_end)
        if not start_dt or not end_dt:
            print("Error: Invalid ISO8601 format for active start/end.", file=sys.stderr)
            return 1
        active_s = int((end_dt - start_dt).total_seconds())
        set_record_field(usage_record, "activity.active_s", active_s, "manual")
    if args.cost is not None:
        set_record_field(usage_record, "cost_estimate_usd", args.cost, "manual")
    if args.quota_events is not None:
        set_record_field(usage_record, "activity.quota_events", args.quota_events, "manual")
    if args.note is not None:
        usage_record["raw_payload"] = args.note
    if getattr(args, "lane", None) is not None:
        set_record_field(usage_record, "lane", args.lane, "manual")
    if getattr(args, "task_class", None) is not None:
        set_record_field(usage_record, "task_class", args.task_class, "manual")

    usage_record["executor"] = get_executor_identity()
    save_yaml(records, filepath)
    print(f"Successfully annotated usage record '{usage_id}'.")
    return 0


def generate_brief_markdown(op_dir: str, task_id: str, harness_id: str) -> str | None:
    resolved_task = resolve_task_record(op_dir, task_id)
    if not resolved_task:
        return None
    task_id, _, task = resolved_task
    harness_path = os.path.join(op_dir, "harnesses", f"{harness_id}.yaml")

    if not os.path.exists(harness_path):
        return None

    harness = load_yaml(harness_path)
    if not harness:
        return None

    claims = []
    claims_list = task.get("claims") or []
    for cid in claims_list:
        resolved_claim = resolve_claim_record(op_dir, cid)
        if not resolved_claim:
            return None
        _, c_data = resolved_claim
        claims.append(c_data)

    latest_handoff = None
    handoffs_list = task.get("handoffs") or []
    if handoffs_list:
        ho_rel_path = handoffs_list[-1]
        expected_handoff = rf"handoffs/{re.escape(task_id)}/handoff-[0-9]{{4,}}\.yaml"
        if not isinstance(ho_rel_path, str) or not re.fullmatch(expected_handoff, ho_rel_path):
            print(
                f"Error: Task '{task_id}' contains an invalid handoff reference.",
                file=sys.stderr,
            )
            return None
        ho_path = os.path.join(op_dir, ho_rel_path)
        if os.path.exists(ho_path):
            latest_handoff = load_yaml(ho_path)

    md = []
    md.append(f"# Brief for Task: {task_id}")
    md.append(f"**Target Harness:** {harness.get('display_name')} (`{harness_id}`)")
    md.append(f"**Repository Path:** `{task.get('repo')}`")
    md.append(f"**Date Generated:** {get_current_utc_iso()}")
    md.append("")
    md.append("## 1. Objective")
    md.append(task.get("objective"))
    md.append("")
    md.append("## 2. Active Phase Rules & Constraints")
    md.append(f"- **Harness Kind:** {harness.get('kind')}")
    md.append(f"- **Model:** {harness.get('model')}")
    md.append("- **Harness Strengths:**")
    for s in harness.get("strengths") or []:
        md.append(f"  - {s}")
    md.append("- **Known Failure Modes / Warnings:**")
    for w in harness.get("known_failure_modes") or []:
        md.append(f"  - WARNING: {w}")
    md.append("")
    md.append("## 3. Current Task State & Recommended Next Action")
    md.append(f"- **Current Status:** `{task.get('status')}`")
    md.append(f"- **Recommended Next Action:** `{task.get('next_action') or 'Not specified'}`")
    md.append("- **Initial Assumptions:**")
    assumptions = task.get("open_assumptions") or []
    if assumptions:
        for val in assumptions:
            md.append(f"  - {val}")
    else:
        md.append("  - (None listed)")
    md.append("")
    md.append("## 4. Latest Harness Handoff Closeout")
    if latest_handoff:
        md.append(
            f"*(Consumed from `{latest_handoff.get('handoff_id')}` by `{latest_handoff.get('by')}` at `{latest_handoff.get('created_at')}`)*"
        )
        md.append("")
        md.append(f"- **What Changed:** {latest_handoff.get('what_changed') or 'Not documented'}")
        md.append(
            f"- **What Was Verified:** {latest_handoff.get('what_verified') or 'Not documented'}"
        )
        md.append(
            f"- **What Was Claimed (unverified):** {latest_handoff.get('what_claimed') or 'Not documented'}"
        )
        md.append(
            f"- **What Remains Open:** {latest_handoff.get('what_remains_open') or 'Not documented'}"
        )
        md.append(
            f"- **Unchecked Assumptions:** {latest_handoff.get('unchecked_assumptions') or 'Not documented'}"
        )
    else:
        md.append("No previous handoff captured for this task yet. Starting from initial setup.")
    md.append("")
    md.append("## 5. Claims & Verification Gates")
    if claims:
        md.append(
            "The following claims have been registered for this task. All load-bearing claims must have verification evidence."
        )
        md.append("")
        md.append("| Claim ID | Made By | Status | Type | Text | Required Gate |")
        md.append("| --- | --- | --- | --- | --- | --- |")
        for c in claims:
            outcome = c.get("verification_outcome")
            if not outcome:
                outcome = "verified" if c.get("verification_status") else "unverified"
            status_str = str(outcome)
            md.append(
                f"| `{c.get('claim_id')}` | {c.get('made_by')} | {status_str} | `{c.get('type')}` | {c.get('text')} | {c.get('required_gate') or 'N/A'} |"
            )
    else:
        md.append("No claims have been registered yet for this task.")
    md.append("")
    md.append("## 6. Required Evidence")
    md.append(
        "- **Builders attach evidence but do NOT set `--status`.** Register claims and attach evidence (`evidence-attach --claim ... --by <you>`). Leave verification to a distinct verifier identity."
    )
    md.append(
        "- **Only the verifier verifies**, with `evidence-attach --status verified --verified-by <reviewer>`."
    )
    md.append(
        "- **Trusted verification requires UID isolation.** `mode: enforced` must map the "
        "executing OS UID to a verifier role, and that UID must differ from the claim author's "
        "UID. `single_user` verification is advisory."
    )
    md.append("- **`file_exists`**: Verify by providing file path or `stat` output.")
    md.append("- **`test_passes`**: Verify by providing output text of the test run.")
    md.append("- **`real_data`**: Provide physical/provenance chain for the data sources.")
    md.append(
        "- **`supervision_credit`**: Name the layer supervised via `--layer` (`design`, `evidence`, `execution`, `release`, `end_to_end`); an unnamed layer is flagged ambiguous by doctor (FR-12)."
    )
    md.append("")
    md.append("## 7. Handoff Guidelines")
    md.append("Upon completion, output a handoff file summarizing:")
    md.append("1. **What changed:** modified files, new features.")
    md.append("2. **What was verified:** tests run, command outputs.")
    md.append("3. **What is claimed:** new claims made (not yet verified).")
    md.append("4. **Open items:** unchecked assumptions, remaining risks.")
    md.append("5. **Next action:** proposed next step for the subsequent harness.")
    md.append("")

    return "\n".join(md)


def brief_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, _, _ = resolved_task

    brief_content = generate_brief_markdown(op_dir, task_id, args.harness_id)
    if not brief_content:
        print(
            f"Error: Could not generate brief for task '{task_id}' and harness '{args.harness_id}'. Check if they exist.",
            file=sys.stderr,
        )
        return 1

    brief_filename = f"{task_id}.{args.harness_id}.md"
    brief_path = os.path.join(op_dir, "briefs", brief_filename)

    try:
        with open(brief_path, "w") as f:
            f.write(brief_content)
    except Exception as e:
        print(f"Error writing brief: {e}", file=sys.stderr)
        return 1

    print(brief_content)
    print(f"\nBrief written to: {brief_path}")
    return 0


def export_brief_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, _, _ = resolved_task

    brief_content = generate_brief_markdown(op_dir, task_id, args.harness_id)
    if not brief_content:
        print(
            f"Error: Could not generate brief for task '{task_id}' and harness '{args.harness_id}'. Check if they exist.",
            file=sys.stderr,
        )
        return 1

    harness_path = os.path.join(op_dir, "harnesses", f"{args.harness_id}.yaml")
    harness = load_yaml(harness_path) or {}
    harness_display_name = harness.get("display_name", args.harness_id)

    export_filename = f"{task_id}.{args.harness_id}.export.md"
    export_path = os.path.join(op_dir, "briefs", export_filename)

    export_header = (
        f"<!-- SYSTEM INSTRUCTIONS FOR {harness_display_name.upper()} -->\n"
        f"<!-- PASTE THIS ENTIRE BRIEF INTO THE CHAT SESSION TO INITIALIZE STATE -->\n"
        f"<!-- DO NOT INVENT OR ASSUME DETAILS NOT DOCUMENTED IN THIS BRIEF -->\n\n"
    )

    full_export_content = export_header + brief_content

    try:
        with open(export_path, "w") as f:
            f.write(full_export_content)
    except Exception as e:
        print(f"Error writing export brief: {e}", file=sys.stderr)
        return 1

    print("=" * 80)
    print(f"PASTE THE CONTENT BELOW INTO THE {harness_display_name.upper()} SESSION:")
    print("=" * 80)
    print(full_export_content)
    print("=" * 80)
    print(f"Export brief written to: {export_path}")
    print("=" * 80)
    return 0


def anchor_record_id_to_local(record_type: str, anchor_record_id: str) -> str:
    """Reverse validate_local_ledger's wire-safe ":" encoding for evidence/handoff anchor
    record IDs back to the local "<task_id>/<leaf_id>" form used in ledger_events.record_id.
    The broker's record_id character class disallows "/", so anchors sent to (and read back
    from) the broker use ":" instead; local storage never changed and still uses "/"."""
    if record_type in {"evidence", "handoff"} and ":" in anchor_record_id:
        return anchor_record_id.replace(":", "/", 1)
    return anchor_record_id


def doctor_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    if enrollment:
        ledger_id = enrollment.ledger_id
        socket_path = enrollment.socket_path
        print("=" * 80)
        print("OPERATOR DOCTOR - LEDGER CONSISTENCY REPORT")
        print("=" * 80)
        print("verification_authority: uid_isolated")
        print("policy_authority: external_broker")

        # Connect to broker and fetch snapshot using pagination
        client = None
        records = []
        broker_err = None
        broker_error_code = None
        broker_policy = None
        through_seq = None
        broker_incarnation = None
        try:
            client = authority_client.AuthorityClient(socket_path)
            after = None
            pinned_seq = None
            while True:
                res = client.send_request(
                    {
                        "action": "projection.snapshot",
                        "ledger_id": ledger_id,
                        "through_commit_sequence": pinned_seq,
                        "after": after,
                        "limit": 16,
                    }
                )
                if res.get("ok"):
                    snapshot = res["snapshot"]
                    if pinned_seq is None:
                        pinned_seq = snapshot["through_commit_sequence"]
                        through_seq = pinned_seq
                        broker_policy = snapshot["policy"]
                        broker_incarnation = snapshot.get("store_incarnation_id")
                    records.extend(snapshot["records"])
                    if not snapshot.get("has_more") or not snapshot.get("next_after"):
                        break
                    after = snapshot["next_after"]
                else:
                    err_obj = res.get("error")
                    if isinstance(err_obj, dict):
                        broker_error_code = err_obj.get("code")
                        broker_err = err_obj.get("message", "unknown error")
                    else:
                        broker_err = str(err_obj or "unknown error")
                    break
        except Exception as e:
            broker_err = str(e)

        if broker_err:
            if broker_error_code == "policy_revoked":
                print("status: policy_revoked")
                print(f"[Error] Broker policy has been revoked: {broker_err}")
            elif broker_error_code == "policy_rotated":
                print("status: policy_rotated")
                print(f"[Error] Broker policy has been rotated: {broker_err}")
            else:
                print("status: broker_unavailable")
                print(f"[Error] Authority broker is unreachable or returned error: {broker_err}")
            print("-" * 80)
            return 1

        if broker_policy != enrollment.policy_binding:
            print("status: policy_rotated")
            print(
                "[Error] Broker policy binding differs from the root-managed enrollment: "
                f"enrolled={canonical_json(enrollment.policy_binding)}, "
                f"broker={canonical_json(broker_policy)}"
            )
            print("-" * 80)
            return 1
        if through_seq < enrollment.first_broker_sequence:
            print("status: divergent_or_forged_local")
            print("[Error] Broker history predates the committed enrollment sequence.")
            print("-" * 80)
            return 1

        # Broker is available, check journal status read-only
        journal_path = os.path.join(op_dir, "client_journal.sqlite3")
        journal_error = None
        stored_policy = None
        local_incarnation = None
        if not os.path.exists(journal_path):
            last_applied = 0
            pending_count = 0
        else:
            try:
                conn_journal = sqlite3.connect(f"file:{journal_path}?mode=ro", uri=True)
                conn_journal.row_factory = sqlite3.Row
                last_applied = authority_projection.get_last_applied_sequence(conn_journal)
                pending_count = conn_journal.execute(
                    "SELECT COUNT(*) FROM transaction_journal WHERE state IN ('prepared', 'committed')"
                ).fetchone()[0]
                policy_row = conn_journal.execute(
                    "SELECT value FROM journal_metadata WHERE key = 'policy_binding'"
                ).fetchone()
                if policy_row:
                    stored_policy = json.loads(policy_row[0])
                local_incarnation = authority_projection.get_metadata(
                    conn_journal, authority_projection.STORE_INCARNATION_KEY
                )
                conn_journal.close()
            except Exception as exc:
                journal_error = str(exc)

        print(
            "policy_binding: "
            f"{broker_policy['id']}@{broker_policy['generation']} {broker_policy['sha256']}"
        )
        if broker_incarnation:
            print(f"store_incarnation_id: {broker_incarnation}")
        if journal_error:
            print("status: divergent_or_forged_local")
            print(f"[Error] Client journal cannot be read safely: {journal_error}")
            print("-" * 80)
            return 1
        if stored_policy is not None and stored_policy != broker_policy:
            print("status: policy_rotated")
            print(
                "[Error] Broker policy binding differs from the last projected binding: "
                f"local={canonical_json(stored_policy)}, broker={canonical_json(broker_policy)}"
            )
            print("-" * 80)
            return 1
        if (
            local_incarnation is not None
            and broker_incarnation is not None
            and local_incarnation != broker_incarnation
        ):
            print("status: divergent_or_forged_local")
            print(
                "[Error] Store incarnation discontinuity: "
                f"local journal={local_incarnation}, broker={broker_incarnation}. "
                "The authority store appears rebuilt/replaced. "
                "Run 'operator authority-reconcile --acknowledge-store-reset' only if intentional."
            )
            print("-" * 80)
            return 1

        effective_sequence = max(last_applied, enrollment.first_broker_sequence)
        if through_seq < effective_sequence:
            print("status: divergent_or_forged_local")
            print(
                f"[Error] Local sequence {effective_sequence} exceeds broker sequence {through_seq}."
            )
            print("-" * 80)
            return 1

        if effective_sequence < through_seq or pending_count > 0:
            print("status: projection_pending")
            print(
                f"[Warning] Projections are pending. Local sequence: {effective_sequence}, Broker sequence: {through_seq}"
            )
            print("-" * 80)
            return 1

        # last_applied == through_seq, check for divergence/forgeries
        diverged = False
        issues = []

        # Connect to local ledger
        db_path = os.path.join(op_dir, "ledger.sqlite3")
        if not os.path.exists(db_path):
            issues.append(f"[Error] Local database {db_path} is missing.")
            diverged = True
        else:
            conn_ledger = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
            conn_ledger.row_factory = sqlite3.Row

            # A repo enrolled before authority_projection_events existed has
            # ledger_events history but no rows in the new table yet, and
            # nothing backfills it retroactively. Report that plainly instead
            # of flagging every already-consistent record as diverged/forged.
            projection_table = conn_ledger.execute(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'authority_projection_events'"
            ).fetchone()
            if not projection_table and records:
                print("status: projection_schema_migration_required")
                print(
                    "[Warning] This ledger predates broker-projection-version tracking "
                    "(authority_projection_events table is missing). "
                    "Divergence cannot be checked until the ledger is re-synced."
                )
                print("-" * 80)
                conn_ledger.close()
                return 1

            try:
                # Compare snapshot records to local database and YAML projections
                for record in records:
                    rtype = record["record_type"]
                    rid = record["record_id"]
                    rversion = record["version"]
                    revent_hash = record["event_hash"]
                    rpayload = record["payload"]

                    # Get local db version and hash
                    local_ver, local_hash = authority_projection.get_local_record_version(
                        conn_ledger, rtype, rid
                    )
                    if local_ver != rversion or local_hash != revent_hash:
                        issues.append(
                            f"[Error] Mismatch on {rtype} {rid}: Local version={local_ver}, hash={local_hash}; Broker version={rversion}, hash={revent_hash}"
                        )
                        diverged = True
                        continue

                    # Get local YAML payload
                    if rtype == "task":
                        filepath = os.path.join(op_dir, "tasks", f"{rid}.yaml")
                    elif rtype == "claim":
                        filepath = os.path.join(op_dir, "claims", f"{rid}.yaml")
                    elif rtype == "evidence":
                        filepath = os.path.join(
                            op_dir, "evidence", rpayload["task_id"], f"{rid}.yaml"
                        )
                    elif rtype == "handoff":
                        filepath = os.path.join(
                            op_dir, "handoffs", rpayload["task_id"], f"{rid}.yaml"
                        )
                    else:
                        filepath = os.path.join(op_dir, f"{rtype}s", f"{rid}.yaml")

                    if not os.path.exists(filepath):
                        issues.append(
                            f"[Error] YAML projection missing for {rtype} {rid}: {filepath}"
                        )
                        diverged = True
                        continue

                    with open(filepath, "r") as f:
                        try:
                            local_payload = yaml.safe_load(f)
                        except Exception as e:
                            issues.append(f"[Error] Failed to load YAML for {rtype} {rid}: {e}")
                            diverged = True
                            continue

                    expected_payload, _, _ = authority_projection.normalize_record(
                        op_dir, record, read_only=True
                    )

                    if canonical_json(local_payload) != canonical_json(expected_payload):
                        issues.append(f"[Error] YAML payload differs from broker for {rtype} {rid}")
                        diverged = True

                # Check for local events not in snapshot (forgeries or legacy records)
                local_records = conn_ledger.execute(
                    """
                    SELECT record_type, record_id, MAX(version) AS ver
                    FROM ledger_events
                    GROUP BY record_type, record_id
                    """
                ).fetchall()

                snapshot_keys = {(r["record_type"], r["record_id"]) for r in records}
                anchor_map = {
                    (
                        r["record_type"],
                        anchor_record_id_to_local(r["record_type"], r["record_id"]),
                    ): (r["version"], r["event_hash"])
                    for r in enrollment.anchor_records
                }

                for lr in local_records:
                    # Strip task_id prefix for evidence/handoff types when checking snapshot
                    snapshot_rid = lr["record_id"]
                    if lr["record_type"] in {"evidence", "handoff"} and "/" in snapshot_rid:
                        snapshot_rid = snapshot_rid.split("/", 1)[1]

                    snapshot_key = (lr["record_type"], snapshot_rid)

                    # 1. If it exists in the broker snapshot
                    if snapshot_key in snapshot_keys:
                        continue

                    # 2. If it does not exist in the broker snapshot, it MUST be a legacy record in anchor_map
                    lr_key = (lr["record_type"], lr["record_id"])
                    if lr_key in anchor_map:
                        anchor_ver, anchor_hash = anchor_map[lr_key]
                        if lr["ver"] > anchor_ver:
                            issues.append(
                                f"[Error] Local forged record found: {lr['record_type']} {lr['record_id']} "
                                f"(exceeds pre-enrollment anchor version {anchor_ver})"
                            )
                            diverged = True
                            continue

                        row_hash = conn_ledger.execute(
                            "SELECT event_hash FROM ledger_events WHERE record_type = ? AND record_id = ? AND version = ?",
                            (lr["record_type"], lr["record_id"], anchor_ver),
                        ).fetchone()
                        if not row_hash or row_hash[0] != anchor_hash:
                            issues.append(
                                f"[Error] Legacy record hash mismatch for {lr['record_type']} {lr['record_id']} "
                                f"at version {anchor_ver}"
                            )
                            diverged = True
                    else:
                        issues.append(
                            f"[Error] Local forged record found: {lr['record_type']} {lr['record_id']} "
                            f"(not present in broker snapshot or pre-enrollment anchor)"
                        )
                        diverged = True

                # Check legacy-only records YAML projections match database
                for reg in enrollment.anchor_records:
                    rtype = reg["record_type"]
                    rid = anchor_record_id_to_local(rtype, reg["record_id"])
                    if (rtype, rid) not in snapshot_keys:
                        # Legacy-only record! Check its YAML projection
                        if rtype == "task":
                            filepath = os.path.join(op_dir, "tasks", f"{rid}.yaml")
                        elif rtype == "claim":
                            filepath = os.path.join(op_dir, "claims", f"{rid}.yaml")
                        elif rtype == "evidence":
                            row = conn_ledger.execute(
                                "SELECT payload_json FROM ledger_events WHERE record_type = ? AND record_id = ? AND version = ?",
                                (rtype, rid, reg["version"]),
                            ).fetchone()
                            if row:
                                try:
                                    payload = json.loads(row[0])
                                    task_id = payload["task_id"]
                                    leaf_id = rid.split("/", 1)[-1]
                                    filepath = os.path.join(
                                        op_dir, "evidence", task_id, f"{leaf_id}.yaml"
                                    )
                                except Exception:
                                    filepath = None
                            else:
                                filepath = None
                        elif rtype == "handoff":
                            row = conn_ledger.execute(
                                "SELECT payload_json FROM ledger_events WHERE record_type = ? AND record_id = ? AND version = ?",
                                (rtype, rid, reg["version"]),
                            ).fetchone()
                            if row:
                                try:
                                    payload = json.loads(row[0])
                                    task_id = payload["task_id"]
                                    leaf_id = rid.split("/", 1)[-1]
                                    filepath = os.path.join(
                                        op_dir, "handoffs", task_id, f"{leaf_id}.yaml"
                                    )
                                except Exception:
                                    filepath = None
                            else:
                                filepath = None
                        else:
                            filepath = os.path.join(op_dir, f"{rtype}s", f"{rid}.yaml")

                        if not filepath or not os.path.exists(filepath):
                            issues.append(
                                f"[Error] YAML projection missing for legacy record {rtype} {rid}"
                            )
                            diverged = True
                            continue

                        with open(filepath, "r") as f:
                            try:
                                local_payload = yaml.safe_load(f)
                            except Exception as e:
                                issues.append(
                                    f"[Error] Failed to load YAML for legacy record {rtype} {rid}: {e}"
                                )
                                diverged = True
                                continue

                        row = conn_ledger.execute(
                            "SELECT payload_json FROM ledger_events WHERE record_type = ? AND record_id = ? AND version = ?",
                            (rtype, rid, reg["version"]),
                        ).fetchone()
                        if not row:
                            issues.append(
                                f"[Error] Legacy record {rtype} {rid} event not found in database"
                            )
                            diverged = True
                            continue

                        try:
                            db_payload = json.loads(row[0])
                        except Exception:
                            issues.append(
                                f"[Error] Invalid JSON payload in database for legacy record {rtype} {rid}"
                            )
                            diverged = True
                            continue

                        # validate_local_ledger already requires, at enrollment time, that every
                        # record's full YAML payload canonically equals its database payload with
                        # no field subsetting (authority_admin.py: "YAML projection differs from
                        # database"). Re-derive that same full comparison here rather than
                        # reconstructing a hand-picked field subset, which can drift from the
                        # real local schema (evidence/claim/task each carry more fields locally --
                        # produced_by, provenance, executor, etc. -- than any fixed subset lists).
                        expected_payload = db_payload

                        if canonical_json(local_payload) != canonical_json(expected_payload):
                            issues.append(
                                f"[Error] YAML payload differs from database for legacy record {rtype} {rid}"
                            )
                            diverged = True

                # Validate the complete projected chain without treating broker hashes as P0 hashes.
                chain_rows = conn_ledger.execute(
                    """
                    SELECT event_id, record_type, record_id, version, payload_json,
                           previous_event_hash, event_hash
                    FROM ledger_events
                    ORDER BY record_type, record_id, version
                    """
                ).fetchall()
                chain_state = {}
                for row in chain_rows:
                    key = (row["record_type"], row["record_id"])
                    previous = chain_state.get(key)
                    expected_version = previous[0] + 1 if previous else 1
                    expected_hash = previous[1] if previous else None
                    if (
                        row["version"] != expected_version
                        or row["previous_event_hash"] != expected_hash
                    ):
                        issues.append(
                            f"[Error] Local broker projection chain is broken for "
                            f"{key[0]} {key[1]} at version {row['version']}"
                        )
                        diverged = True
                    try:
                        if canonical_json(json.loads(row["payload_json"])) != row["payload_json"]:
                            raise ValueError("payload is not canonical")
                    except (TypeError, ValueError, json.JSONDecodeError):
                        issues.append(
                            f"[Error] Local broker projection payload is invalid for "
                            f"{key[0]} {key[1]} version {row['version']}"
                        )
                        diverged = True
                    if row["event_id"] != row["event_hash"]:
                        issues.append(
                            f"[Error] Local broker projection event identity is invalid for "
                            f"{key[0]} {key[1]} version {row['version']}"
                        )
                        diverged = True
                    chain_state[key] = (row["version"], row["event_hash"])

                audit_issues = audit_event_store(op_dir)
                if audit_issues:
                    issues.extend(audit_issues)
                    diverged = True

                expected_triggers = {
                    "ledger_events_no_update": "create trigger ledger_events_no_update before update on ledger_events begin select raise(abort, 'ledger_events is append-only'); end",
                    "ledger_events_no_delete": "create trigger ledger_events_no_delete before delete on ledger_events begin select raise(abort, 'ledger_events is append-only'); end",
                    "authority_projection_events_no_update": "create trigger authority_projection_events_no_update before update on authority_projection_events begin select raise(abort, 'authority_projection_events is append-only'); end",
                    "authority_projection_events_no_delete": "create trigger authority_projection_events_no_delete before delete on authority_projection_events begin select raise(abort, 'authority_projection_events is append-only'); end",
                }
                trigger_rows = conn_ledger.execute(
                    "SELECT name, sql FROM sqlite_master WHERE type = 'trigger'"
                ).fetchall()
                actual_triggers = {
                    row["name"]: " ".join(row["sql"].lower().split()) for row in trigger_rows
                }
                for name, expected_sql in expected_triggers.items():
                    if name.startswith("authority_projection") and not projection_table:
                        continue
                    if actual_triggers.get(name) != expected_sql:
                        issues.append(
                            f"[Error] Local append-only trigger {name} is missing or modified."
                        )
                        diverged = True

            except Exception as e:
                issues.append(f"[Error] Local database verification failed: {e}")
                diverged = True
            finally:
                conn_ledger.close()

        if diverged:
            print("status: divergent_or_forged_local")
            for iss in issues:
                print(iss)
            print("-" * 80)
            return 1

        print("status: current")
        print("All records consistent with broker.")
        print("-" * 80)
        return 0

    # Non-enrolled logic
    try:
        identity_policy = load_identity_policy(op_dir)
        mode = identity_policy["mode"]
    except Exception:
        mode = "advisory"

    print("=" * 80)
    print("OPERATOR DOCTOR - LEDGER CONSISTENCY REPORT")
    print("=" * 80)
    auth_val = "uid_isolated" if mode == "enforced" else "advisory"
    print(f"verification_authority: {auth_val}")
    print("policy_authority: local_policy")
    print("status: legacy_local_policy")
    print("-" * 80)

    tasks_dir = os.path.join(op_dir, "tasks")
    task_files = glob.glob(os.path.join(tasks_dir, "*.yaml"))

    issues = []
    infos = []
    issues.extend(audit_event_store(op_dir))

    try:
        identity_policy = load_identity_policy(op_dir)
    except IdentityPolicyError as exc:
        issues.append(f"[Error] Invalid identity policy: {exc}")
        identity_policy = {"mode": "invalid", "uids": {}}
    identity_map = identity_policy["uids"]
    mode = identity_policy["mode"]

    def check_executor_block(item, source_name):
        contains_test_identity = False
        seen = set()
        for field in ("executor", "author_executor", "verification_executor"):
            executor = item.get(field)
            if not isinstance(executor, dict):
                continue
            marker = id(executor)
            if marker in seen:
                continue
            seen.add(marker)
            if executor.get("test_override_active"):
                contains_test_identity = True
                issues.append(
                    f"[Error] {source_name} {field} contains write made with test-override active"
                )
            if executor.get("test_override_unauthorized"):
                contains_test_identity = True
                issues.append(
                    f"[Error] {source_name} {field} contains write with unauthorized "
                    "test-override attempt"
                )
        return contains_test_identity

    for tf in task_files:
        task = load_yaml(tf)
        if not isinstance(task, dict):
            continue

        file_task_id = os.path.splitext(os.path.basename(tf))[0]
        task_id = task.get("task_id")
        if (
            not isinstance(task_id, str)
            or not VALID_TASK_ID.fullmatch(task_id)
            or task_id != file_task_id
        ):
            issues.append(
                f"[Error] Task projection {os.path.basename(tf)} has invalid or mismatched "
                f"task_id: {task_id}"
            )
            continue
        check_executor_block(task, f"Task {task_id}")
        task_status = task.get("status")

        # Repo existence check
        repo_path = task.get("repo")
        if repo_path and not os.path.exists(repo_path):
            if task_status in ["verified", "complete"]:
                issues.append(
                    f"[Error] Task {task_id} repo path does not exist on disk: {repo_path}"
                )
            else:
                infos.append(
                    f"[Warning] Task {task_id} repo path does not exist on disk: {repo_path}"
                )

        claims_list = task.get("claims") or []
        quarantined_claims = []
        verified_claims = []
        verified_evidence_paths = set()
        for cid in claims_list:
            if not isinstance(cid, str) or not VALID_CLAIM_ID.fullmatch(cid):
                issues.append(f"[Error] Task {task_id} references invalid claim ID: {cid}")
                continue
            c_path = os.path.join(op_dir, "claims", f"{cid}.yaml")
            if not os.path.exists(c_path):
                issues.append(f"[Error] Task {task_id} references missing claim: {cid}")
                continue

            c_data = load_yaml(c_path)
            if not c_data:
                issues.append(f"[Error] Claim {cid} could not be loaded.")
                continue
            if c_data.get("claim_id") != cid:
                issues.append(
                    f"[Error] Claim projection {cid}.yaml contains mismatched claim_id: "
                    f"{c_data.get('claim_id')}"
                )
                continue

            if c_data.get("task_id") != task_id:
                issues.append(
                    f"[Error] Task {task_id} references claim {cid} owned by task "
                    f"{c_data.get('task_id')}"
                )

            claim_has_test_identity = check_executor_block(c_data, f"Claim {cid}")

            outcome = c_data.get("verification_outcome")
            verified = (outcome == "verified") or bool(c_data.get("verification_status"))

            if not c_data.get("required_gate"):
                issues.append(
                    f"[Warning] Claim {cid} ('{c_data.get('text')}') has no required_gate defined."
                )

            if c_data.get("type") == "supervision_credit" and not c_data.get("supervision_layer"):
                issues.append(
                    f"[Warning] Claim {cid} is a supervision_credit with no named layer (ambiguous credit; FR-12)."
                )

            evidence_refs = c_data.get("evidence_refs") or []
            if verified and not evidence_refs:
                issues.append(
                    f"[Error] Claim {cid} is marked verified but has no evidence references."
                )

            for ev_ref in evidence_refs:
                ev_ref_path = os.path.join(op_dir, ev_ref)
                if verified:
                    verified_evidence_paths.add(os.path.abspath(ev_ref_path))
                if not os.path.exists(ev_ref_path):
                    issues.append(
                        f"[Error] Claim {cid} evidence reference points to missing file: {ev_ref}"
                    )
                else:
                    ev_ref_data = load_yaml(ev_ref_path)
                    if ev_ref_data:
                        check_executor_block(ev_ref_data, f"Evidence file {ev_ref}")
                        if ev_ref_data.get("task_id") != task_id:
                            issues.append(
                                f"[Error] Claim {cid} references evidence owned by task "
                                f"{ev_ref_data.get('task_id')}: {ev_ref}"
                            )
                        bound_claim_id = ev_ref_data.get("claim_id")
                        if bound_claim_id and bound_claim_id != cid:
                            issues.append(
                                f"[Error] Claim {cid} references evidence bound to claim "
                                f"{bound_claim_id}: {ev_ref}"
                            )
                        provenance = ev_ref_data.get("provenance")
                        stored_fingerprint, _ = parse_stored_fingerprint(
                            ev_ref_data.get("fingerprint"), ev_ref_data.get("hash")
                        )

                        if provenance == "local_file" and not stored_fingerprint:
                            issues.append(
                                f"[Warning] Claim {cid} evidence reference has no valid fingerprint: {ev_ref}"
                            )

                        is_recheckable = provenance in [
                            "local_file",
                            "living_document",
                            "structural_verification",
                        ]
                        if not is_recheckable:
                            v_cmd = ev_ref_data.get("verification_command")
                            if not v_cmd or not stored_fingerprint:
                                if verified:
                                    issues.append(
                                        f"[Error] Claim {cid} verified on unverifiable evidence: {ev_ref} (external/url provenance must have verification_command and hash)"
                                    )
                                else:
                                    infos.append(
                                        f"[Warning] Claim {cid} draft on unverifiable evidence: {ev_ref} (external/url provenance must have verification_command and hash)"
                                    )

                        if ev_ref_data.get("evidence_type") == "test_output":
                            ev_path = ev_ref_data.get("path_or_url")
                            if ev_path and not ev_path.startswith(("http://", "https://")):
                                # Resolve relative path relative to repo_path if available
                                if repo_path and not os.path.isabs(ev_path):
                                    ev_path_resolved = os.path.join(repo_path, ev_path)
                                else:
                                    ev_path_resolved = ev_path

                                if not os.path.exists(ev_path_resolved):
                                    if verified:
                                        issues.append(
                                            f"[Error] Evidence {ev_ref} test_output file does not exist: {ev_path_resolved}"
                                        )
                                    else:
                                        infos.append(
                                            f"[Warning] Evidence {ev_ref} test_output file does not exist: {ev_path_resolved}"
                                        )

            if outcome == "quarantined":
                quarantined_claims.append(cid)
            elif verified:
                verified_claims.append(cid)

            if c_data.get("type") == "test_passes":
                req_gate = c_data.get("required_gate")
                if req_gate:
                    gate_path = req_gate
                    if repo_path and not os.path.isabs(gate_path):
                        gate_path = os.path.join(repo_path, gate_path)
                    if not os.path.exists(gate_path):
                        if verified:
                            issues.append(
                                f"[Error] Claim {cid} verified test_passes required_gate file does not exist: {req_gate}"
                            )
                        else:
                            infos.append(
                                f"[Warning] Claim {cid} draft test_passes required_gate file does not exist: {req_gate}"
                            )

                # Check for verification_command in evidence
                has_verification_cmd = False
                for ev_ref in evidence_refs:
                    ev_ref_path = os.path.join(op_dir, ev_ref)
                    if os.path.exists(ev_ref_path):
                        ev_ref_data = load_yaml(ev_ref_path)
                        if ev_ref_data and ev_ref_data.get("verification_command"):
                            has_verification_cmd = True
                            break
                if not has_verification_cmd:
                    if verified:
                        issues.append(
                            f"[Error] Claim {cid} verified test_passes has no evidence reference with a verification_command."
                        )
                    else:
                        infos.append(
                            f"[Warning] Claim {cid} draft test_passes has no evidence reference with a verification_command."
                        )

            made = c_data.get("made_by")
            vby = c_data.get("verified_by")
            review = task.get("review_harness")
            authority = c_data.get("verification_authority")
            recorded_mode = c_data.get("verification_mode")

            if verified:
                if authority != "uid_isolated" and vby == made:
                    issues.append(
                        f"[Error] claim {cid} is self-verified by '{made}' (maker cannot verify own claim)"
                    )
                if vby and review and vby != review:
                    issues.append(
                        f"[Warning] claim {cid} verified by '{vby}', not the task's review harness '{review}'"
                    )

            if authority == "uid_isolated":
                authority_errors = []
                author_executor = c_data.get("author_executor")
                verification_executor = c_data.get("verification_executor")
                compatibility_executor = c_data.get("executor")
                author_uid = (
                    author_executor.get("uid") if isinstance(author_executor, dict) else None
                )
                verifier_uid = (
                    verification_executor.get("uid")
                    if isinstance(verification_executor, dict)
                    else None
                )
                compatibility_uid = (
                    compatibility_executor.get("uid")
                    if isinstance(compatibility_executor, dict)
                    else None
                )

                if recorded_mode != "enforced":
                    authority_errors.append(
                        f"uid_isolated verification was not recorded in enforced mode "
                        f"(found {recorded_mode!r})"
                    )
                if mode == "single_user":
                    infos.append(
                        f"[Warning] Claim {cid} was recorded as uid_isolated, but the "
                        "current identity policy is single_user; policy immutability is "
                        "outside P2"
                    )
                if (
                    not isinstance(author_uid, int)
                    or isinstance(author_uid, bool)
                    or author_uid < 0
                ):
                    authority_errors.append("uid_isolated verification has no valid author UID")
                if (
                    not isinstance(verifier_uid, int)
                    or isinstance(verifier_uid, bool)
                    or verifier_uid < 0
                ):
                    authority_errors.append("uid_isolated verification has no valid verifier UID")
                if (
                    not isinstance(compatibility_uid, int)
                    or isinstance(compatibility_uid, bool)
                    or compatibility_uid < 0
                ):
                    authority_errors.append(
                        "uid_isolated verification has no valid compatibility executor UID"
                    )
                elif (
                    isinstance(verifier_uid, int)
                    and not isinstance(verifier_uid, bool)
                    and verifier_uid >= 0
                    and compatibility_uid != verifier_uid
                ):
                    authority_errors.append(
                        f"uid_isolated compatibility executor UID {compatibility_uid} disagrees "
                        f"with verifier UID {verifier_uid}"
                    )
                if (
                    isinstance(author_uid, int)
                    and not isinstance(author_uid, bool)
                    and author_uid >= 0
                    and isinstance(verifier_uid, int)
                    and not isinstance(verifier_uid, bool)
                    and verifier_uid >= 0
                    and author_uid == verifier_uid
                ):
                    authority_errors.append(
                        f"uid_isolated verification uses the same author and verifier UID "
                        f"({author_uid})"
                    )

                verifier_identity = (
                    identity_map.get(verifier_uid)
                    if isinstance(verifier_uid, int)
                    and not isinstance(verifier_uid, bool)
                    and verifier_uid >= 0
                    else None
                )
                if verifier_uid is not None and verifier_identity is None:
                    authority_errors.append(
                        f"uid_isolated verification uses unregistered verifier UID {verifier_uid}"
                    )
                elif verifier_identity and "verifier" not in verifier_identity["roles"]:
                    authority_errors.append(
                        f"uid_isolated verification UID {verifier_uid} lacks the verifier role"
                    )
                elif verifier_identity and vby != verifier_identity["name"]:
                    authority_errors.append(
                        f"uid_isolated verification identity mismatch: verified_by is "
                        f"'{vby}', but verifier UID {verifier_uid} maps to "
                        f"'{verifier_identity['name']}'"
                    )

                if not vby:
                    authority_errors.append("uid_isolated verification has no verified_by identity")
                if claim_has_test_identity:
                    authority_errors.append(
                        "uid_isolated verification contains a guarded test identity and is not "
                        "proof of real UID isolation"
                    )

                if authority_errors:
                    issues.extend(f"[Error] Claim {cid} {message}" for message in authority_errors)
                else:
                    infos.append(
                        f"[Info] Claim {cid} verification is uid_isolated "
                        f"(author uid {author_uid}, verifier uid {verifier_uid})"
                    )
            elif authority == "advisory":
                if recorded_mode != "single_user":
                    issues.append(
                        f"[Error] Claim {cid} advisory verification has invalid recorded mode "
                        f"{recorded_mode!r}; expected 'single_user'"
                    )
                infos.append(
                    f"[Warning] claim {cid} verification is advisory and NOT "
                    "identity-enforced (single-user mode); see EXECUTOR_IDENTITY_SPEC.md"
                )

                verification_executor = c_data.get("verification_executor")
                exec_uid = (
                    verification_executor.get("uid")
                    if isinstance(verification_executor, dict)
                    else None
                )
                expected_verifier = identity_map.get(exec_uid)
                if expected_verifier and vby != expected_verifier["name"]:
                    infos.append(
                        f"[Warning] claim {cid} would be REJECTED under enforced mode: "
                        f"verified_by '{vby}' != identity '{expected_verifier['name']}' for "
                        f"uid {exec_uid}; enforcement appears relaxed to single_user"
                    )
            elif authority is None and verified:
                infos.append(
                    f"[Info] Claim {cid} has legacy/advisory verification authority; "
                    "it is not uid_isolated"
                )
                executor = c_data.get("executor")
                if not vby or not isinstance(executor, dict):
                    infos.append(f"[Info] Claim {cid} has unknown verifier (legacy claim)")
                else:
                    exec_uid = executor.get("uid")
                    expected_verifier = identity_map.get(exec_uid)
                    if mode == "enforced":
                        if not expected_verifier:
                            infos.append(
                                f"[Warning] legacy claim {cid} verifier UID {exec_uid} is not "
                                "registered; verification remains advisory"
                            )
                        elif vby != expected_verifier["name"]:
                            infos.append(
                                f"[Warning] legacy claim {cid} verification identity mismatch: "
                                f"verified_by is '{vby}', but executor maps to "
                                f"'{expected_verifier['name']}' (uid {exec_uid}); verification "
                                "remains advisory"
                            )
                    else:
                        infos.append(
                            f"[Warning] claim {cid} verification is NOT identity-enforced "
                            "(single-user mode); see EXECUTOR_IDENTITY_SPEC.md"
                        )
                        if expected_verifier and vby != expected_verifier["name"]:
                            infos.append(
                                f"[Warning] claim {cid} would be REJECTED under enforced mode: "
                                f"verified_by '{vby}' != identity '{expected_verifier['name']}' "
                                f"for uid {exec_uid}; enforcement appears relaxed to single_user"
                            )
            elif authority is None and recorded_mode is not None:
                issues.append(
                    f"[Error] Claim {cid} has verification_mode {recorded_mode!r} but no "
                    "verification_authority"
                )
            elif authority is not None:
                issues.append(
                    f"[Error] Claim {cid} has unknown verification_authority {authority!r}"
                )

        if task_status in ["verified", "complete"] and quarantined_claims:
            issues.append(
                f"[Error] Task {task_id} status is '{task_status}' while claims are quarantined: {', '.join(quarantined_claims)}"
            )

        if task_status == "quarantined" and verified_claims:
            issues.append(
                f"[Error] Task {task_id} is quarantined while claims are still marked verified: {', '.join(verified_claims)}"
            )

        ho_list = task.get("handoffs") or []
        for ho_ref in ho_list:
            ho_path = os.path.join(op_dir, ho_ref)
            if not os.path.exists(ho_path):
                issues.append(
                    f"[Error] Task {task_id} handoff reference points to missing file: {ho_ref}"
                )
            else:
                ho_data = load_yaml(ho_path)
                if ho_data:
                    ho_id = ho_data.get("handoff_id")
                    check_executor_block(ho_data, f"Handoff {ho_id} for task {task_id}")

        if ho_list:
            latest_ho_ref = ho_list[-1]
            latest_ho_path = os.path.join(op_dir, latest_ho_ref)
            if os.path.exists(latest_ho_path):
                latest_ho = load_yaml(latest_ho_path)
                if latest_ho:
                    ho_next = latest_ho.get("next_action")
                    task_next = task.get("next_action")
                    if (
                        task_status not in ["verified", "complete"]
                        and ho_next
                        and ho_next != task_next
                    ):
                        issues.append(
                            f"[Warning] Task {task_id} next_action differs from the latest handoff recommended next action. Task: '{task_next}', Handoff: '{ho_next}'"
                        )

        # Paths of evidence YAML cited by any claim on this task (for crystal gap reporting).
        cited_evidence_abs: set[str] = set()
        for cid in task.get("claims") or []:
            claim_path = os.path.join(op_dir, "claims", f"{cid}.yaml")
            if not os.path.exists(claim_path):
                continue
            c_data = load_yaml(claim_path)
            if not c_data:
                continue
            for ev_ref in c_data.get("evidence_refs") or []:
                cited_evidence_abs.add(os.path.abspath(os.path.join(op_dir, ev_ref)))

        ev_dir = os.path.join(op_dir, "evidence", task_id)
        if os.path.isdir(ev_dir):
            ev_files = glob.glob(os.path.join(ev_dir, "evidence-*.yaml"))
            for ef in ev_files:
                ev_data = load_yaml(ef)
                if not ev_data:
                    continue
                ev_id = ev_data.get("evidence_id")
                check_executor_block(ev_data, f"Evidence {ev_id} for task {task_id}")
                evidence_is_verified = (
                    os.path.abspath(ef) in verified_evidence_paths
                    or ev_data.get("claim_id") in verified_claims
                )
                freshness_issues, freshness_infos = evidence_freshness_diagnostics(
                    ev_data, evidence_is_verified
                )
                issues.extend(freshness_issues)
                infos.extend(freshness_infos)

                # Crystal↔Ledger Phase 1 doctor rules (advisory; foreign artifacts).
                # Spec §6: unknown-commit and narration-gap are non-fatal (exit 0).
                if ev_data.get("evidence_type") == "session_crystal":
                    crystal_commit = ev_data.get("crystal_source_commit")
                    if crystal_commit and repo_path:
                        known = git_commit_exists_in_repo(repo_path, str(crystal_commit))
                        if known is False:
                            infos.append(
                                f"[Warning] crystal cites unknown commit '{crystal_commit}' "
                                f"(evidence {ev_id})"
                            )
                    test_n = ev_data.get("crystal_test_bullet_count")
                    if isinstance(test_n, int) and test_n > 0:
                        if os.path.abspath(ef) not in cited_evidence_abs:
                            infos.append(
                                f"[Info] crystal narrates {test_n} verification claims; "
                                f"none registered (evidence {ev_id})"
                            )

        if args.audit and task_status == "verified":
            # Check for independent supervisor_verdict.md (Phase 3.2)
            # Assumption: run directories are under docs/domain_runs/DBB-001a/
            # We'll try to find a directory matching task_id (or similar)
            pass

    usage_by_task = {}
    pricing_table = load_pricing_table(op_dir)

    def check_divergence(val_manual, val_auto):
        try:
            m = float(val_manual)
            a = float(val_auto)
            if m == a:
                return False
            if a == 0 or m == 0:
                return True
            return m > 2 * a or a > 2 * m
        except Exception:
            return False

    usage_dir = os.path.join(op_dir, "usage")
    if os.path.isdir(usage_dir):
        for usage_path in glob.glob(os.path.join(usage_dir, "*.yaml")):
            records = load_yaml(usage_path) or []
            for rec in records:
                task_id = rec.get("task_id")
                usage_id = rec.get("usage_id", "unknown")
                check_executor_block(rec, f"Usage record {usage_id}")
                if task_id and rec.get("ended_at") is None:
                    usage_by_task.setdefault(task_id, []).append(usage_id)
                    infos.append(f"[Info] Task {task_id} has open usage session: {usage_id}")

                if rec.get("lane") == "frontier_driver" and rec.get("task_class") == "bounded":
                    infos.append(
                        f"[Info] Session {usage_id} is an offload candidate (tagged frontier_driver + bounded)."
                    )

                harness_id = rec.get("harness_id")
                metering = rec.get("metering")
                if not metering:
                    metering = "activity" if harness_id == "gemini-agy" else "tokens"

                if metering == "tokens" and rec.get("cost_estimate_usd") is None:
                    model = rec.get("model")
                    if model not in pricing_table:
                        issues.append(f"[Warning] price table missing model {model}")

                source_ref = rec.get("source_session_ref")
                if source_ref and not os.path.exists(source_ref):
                    issues.append(
                        f"[Warning] Usage record {usage_id} source session log file is missing: {source_ref}"
                    )

                cost_estimate = rec.get("cost_estimate_usd")
                field_sources = rec.get("field_sources", {})
                if metering == "activity" and cost_estimate is not None:
                    if field_sources.get("cost_estimate_usd") == "auto":
                        issues.append(
                            f"[Error] Usage record {usage_id} is activity-metered but has auto-calculated cost."
                        )

                auto_vals = rec.get("auto_values", {})

                for field in [
                    "tokens_in",
                    "tokens_out",
                    "tokens_cache_read",
                    "tokens_cache_write",
                    "cost_estimate_usd",
                ]:
                    if field_sources.get(field) == "manual" and field in auto_vals:
                        m_val = rec.get(field)
                        a_val = auto_vals[field]
                        if m_val is not None and a_val is not None:
                            if check_divergence(m_val, a_val):
                                issues.append(f"[Warning] manual/auto divergence on {field}")

                activity = rec.get("activity", {})
                for field in ["tool_calls", "turns", "wall_clock_s", "quota_events"]:
                    if field_sources.get(field) == "manual" and field in auto_vals:
                        m_val = activity.get(field)
                        a_val = auto_vals[field]
                        if m_val is not None and a_val is not None:
                            if check_divergence(m_val, a_val):
                                issues.append(f"[Warning] manual/auto divergence on {field}")
    infos.sort()

    for tf in task_files:
        task = load_yaml(tf)
        if not task:
            continue
        task_id = task.get("task_id", os.path.splitext(os.path.basename(tf))[0])
        status = task.get("status")
        open_usage_ids = usage_by_task.get(task_id, [])
        if status == "running" and not open_usage_ids:
            issues.append(f"[Warning] Task {task_id} is running but has no open usage session.")
        if status != "running" and open_usage_ids:
            issues.append(
                f"[Error] Task {task_id} is '{status}' but has open usage sessions: {', '.join(open_usage_ids)}"
            )
        if len(open_usage_ids) > 1:
            issues.append(
                f"[Warning] Task {task_id} has multiple open usage sessions: {', '.join(open_usage_ids)}"
            )

    print("=" * 80)
    print("OPERATOR DOCTOR - LEDGER CONSISTENCY REPORT")
    print("=" * 80)
    all_msgs = infos + issues
    if all_msgs:
        for msg in all_msgs:
            print(msg)
        print("-" * 80)

    if issues:
        print(f"Total consistency issues found: {len(issues)}")
        return 1
    else:
        has_warnings = any("Warning" in msg for msg in infos)
        if has_warnings:
            print("Records consistent, but non-fatal warnings exist. Review audit report above.")
        else:
            print("All records consistent. No issues found.")
        print("=" * 80)
        return 0


def verify_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print("Error: No .operator directory found.", file=sys.stderr)
        return 1

    run_dir = Path(args.run_dir).resolve()
    if not run_dir.is_dir():
        print(f"Error: {run_dir} is not a directory.", file=sys.stderr)
        return 1

    verdict_path = run_dir / "candidate_verdict.md"
    if not verdict_path.exists():
        print(f"Error: candidate_verdict.md missing in {run_dir}", file=sys.stderr)
        return 1

    print(f"Verifying run directory: {run_dir}")

    issues = []

    # Load candidate verdict
    try:
        with open(verdict_path, "r") as f:
            content = f.read()
            # Simple YAML extraction from markdown
            match = re.search(r"```yaml\n(.*?)\n```", content, re.DOTALL)
            if match:
                verdict_data = yaml.safe_load(match.group(1))
            else:
                # Try parsing raw if no code block
                verdict_data = yaml.safe_load(content)
    except Exception as e:
        print(f"Error parsing candidate_verdict.md: {e}", file=sys.stderr)
        return 1

    if not verdict_data:
        print("Error: Empty or invalid candidate_verdict.md", file=sys.stderr)
        return 1

    # Normalize structure (candidate_verdict might be nested or flat)
    v_info = verdict_data.get("candidate_verdict", verdict_data)

    # 1. Recalculate artifact hashes (Phase 3.1.1)
    stored_hashes = v_info.get("distinct_tools_hashes", {})
    if stored_hashes:
        for variation, stored_hash in stored_hashes.items():
            tools_path = run_dir / "saved_implementation" / "variations" / variation / "tools.py"
            if tools_path.exists():
                curr_hash = hashlib.md5(tools_path.read_bytes()).hexdigest()
                if curr_hash != stored_hash:
                    issues.append(
                        f"[Error] Hash mismatch for {variation}: stored {stored_hash}, recalculated {curr_hash}"
                    )
            else:
                issues.append(
                    f"[Warning] Referenced variation {variation} missing on disk at {tools_path}"
                )

    # 2. Verify verified_by != made_by (Phase 3.1.2)
    handoff_id = v_info.get("handoff_id")
    if handoff_id:
        # We need to find the claim in the operator ledger
        # This is a bit complex as we need to scan tasks
        found_claim = False
        tasks_dir = os.path.join(op_dir, "tasks")
        for tf in glob.glob(os.path.join(tasks_dir, "*.yaml")):
            task = load_yaml(tf)
            if not task:
                continue
            for cid in task.get("claims", []):
                if not isinstance(cid, str) or not VALID_CLAIM_ID.fullmatch(cid):
                    issues.append(f"[Error] Task references invalid claim ID: {cid}")
                    continue
                c_path = os.path.join(op_dir, "claims", f"{cid}.yaml")
                if os.path.exists(c_path):
                    c_data = load_yaml(c_path)
                    if c_data and c_data.get("text") and handoff_id in c_data.get("text", ""):
                        found_claim = True
                        made = c_data.get("made_by")
                        vby = c_data.get("verified_by")
                        if vby and made and vby == made:
                            issues.append(
                                f"[Error] Claim for {handoff_id} is self-verified by '{made}'"
                            )
        if not found_claim:
            # We skip if handoff isn't in ledger yet, but warn
            pass

    # 3. Check for Evidence Drift (Phase 3.1.3)
    # Check AGENT_HANDOFF_STATUS.md against current disk state
    status_path = Path(op_dir).parent / "AGENT_HANDOFF_STATUS.md"
    if status_path.exists():
        status_content = status_path.read_text()
        # Look for the row matching implementation_path
        impl_path = v_info.get("implementation_path")
        if impl_path:
            # Look for | Implementation Path | ... | hash | ...
            # Regex for table row
            pattern = rf"\|\s*{re.escape(impl_path)}.*\|\s*([a-f0-9]+)\s*\|"
            match = re.search(pattern, status_content, re.IGNORECASE)
            if match:
                status_hash = match.group(1)
                # Check against variation_1/tools.py hash (typical for DBB-001a)
                t1_path = (
                    run_dir / "saved_implementation" / "variations" / "variation_1" / "tools.py"
                )
                if t1_path.exists():
                    curr_t1_hash = hashlib.md5(t1_path.read_bytes()).hexdigest()[:8]
                    if curr_t1_hash != status_hash:
                        issues.append(
                            f"[Error] Evidence Drift: AGENT_HANDOFF_STATUS.md reports {status_hash}, but disk implementation hash is {curr_t1_hash}"
                        )

    if issues:
        print("--- VERIFICATION ISSUES FOUND ---")
        for msg in issues:
            print(msg)
        return 1
    else:
        print("All artifact hashes and metadata verified successfully.")
        return 0


def task_list_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    tasks_dir = os.path.join(op_dir, "tasks")
    task_files = glob.glob(os.path.join(tasks_dir, "*.yaml"))

    if not task_files:
        print("No tasks found in ledger.")
        return 0

    tasks_data = []
    for tf in sorted(task_files):
        task = load_yaml(tf)
        if not task:
            continue

        task_id = task.get("task_id")
        status = task.get("status")
        assign = task.get("assigned_harness") or "None"
        review = task.get("review_harness") or "None"
        next_act = task.get("next_action") or "None"

        claims_list = task.get("claims") or []
        verified_cnt = 0
        quarantined_cnt = 0
        open_cnt = 0

        for cid in claims_list:
            if not isinstance(cid, str) or not VALID_CLAIM_ID.fullmatch(cid):
                open_cnt += 1
                continue
            c_path = os.path.join(op_dir, "claims", f"{cid}.yaml")
            if os.path.exists(c_path):
                c_data = load_yaml(c_path)
                if c_data:
                    outcome = c_data.get("verification_outcome")
                    if outcome == "quarantined":
                        quarantined_cnt += 1
                    elif c_data.get("verification_status") or outcome == "verified":
                        verified_cnt += 1
                    else:
                        open_cnt += 1
            else:
                open_cnt += 1

        claims_summary = f"V:{verified_cnt} Q:{quarantined_cnt} O:{open_cnt}"

        tasks_data.append(
            {
                "task_id": task_id,
                "status": status,
                "assign": assign,
                "review": review,
                "claims": claims_summary,
                "next_action": next_act,
            }
        )

    print("=" * 100)
    print(
        f"{'TASK ID':<25} | {'STATUS':<12} | {'ASSIGNED':<10} | {'REVIEWER':<10} | {'CLAIMS':<10} | {'LATEST NEXT ACTION'}"
    )
    print("=" * 100)
    for td in tasks_data:
        na = td["next_action"]
        if len(na) > 30:
            na = na[:27] + "..."
        print(
            f"{td['task_id']:<25} | {td['status']:<12} | {td['assign']:<10} | {td['review']:<10} | {td['claims']:<10} | {na}"
        )
    print("=" * 100)
    return 0


def claim_list_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    task_id = args.task_id
    if not task_id:
        config = load_yaml(os.path.join(op_dir, "operator.yaml")) or {}
        task_id = config.get("current_task")

    claims_dir = os.path.join(op_dir, "claims")
    claim_files = glob.glob(os.path.join(claims_dir, "claim-*.yaml"))

    if not claim_files:
        print("No claims found in ledger.")
        return 0

    claims_data = []
    for cf in sorted(claim_files):
        c_data = load_yaml(cf)
        if not c_data:
            continue

        t_id = c_data.get("task_id")
        if task_id and task_id != "all" and t_id != task_id:
            continue

        cid = c_data.get("claim_id")
        c_type = c_data.get("type")
        text = c_data.get("text")
        outcome = c_data.get("verification_outcome")
        if not outcome:
            outcome = "verified" if c_data.get("verification_status") else "unverified"

        claims_data.append(
            {
                "claim_id": cid,
                "task_id": t_id,
                "type": c_type,
                "status": str(outcome).upper(),
                "text": text,
            }
        )

    if not claims_data:
        if task_id and task_id != "all":
            print(f"No claims found for task '{task_id}'.")
        else:
            print("No claims found.")
        return 0

    print("=" * 100)
    title_str = f"CLAIMS LEDGER (Task: {task_id or 'all'})"
    print(f"{title_str}")
    print("=" * 100)
    print(f"{'CLAIM ID':<10} | {'TASK ID':<20} | {'TYPE':<15} | {'STATUS':<12} | {'CLAIM TEXT'}")
    print("=" * 100)
    for cd in claims_data:
        text = cd["text"]
        if len(text) > 40:
            text = text[:37] + "..."
        print(
            f"{cd['claim_id']:<10} | {cd['task_id']:<20} | {cd['type']:<15} | {cd['status']:<12} | {text}"
        )
    print("=" * 100)
    return 0


def claim_show_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    claim_id = args.claim_id
    if not claim_id:
        print("Error: Specify a claim ID (e.g. claim-0001).", file=sys.stderr)
        return 1

    resolved_claim = resolve_claim_record(op_dir, claim_id)
    if not resolved_claim:
        return 1
    _, claim = resolved_claim

    print("=" * 80)
    print(f"Claim ID:            {claim.get('claim_id')}")
    print(f"Task ID:             {claim.get('task_id')}")
    print(f"Type:                {claim.get('type')}")
    print(f"Text:                {claim.get('text')}")
    print(f"Made By:             {claim.get('made_by')}")
    print(f"Made At:             {claim.get('made_at')}")

    outcome = claim.get("verification_outcome")
    if not outcome:
        outcome = "verified" if claim.get("verification_status") else "unverified"

    status_str = str(outcome).upper()
    print(f"Verification status: {status_str} (Status boolean: {claim.get('verification_status')})")
    print(f"Verified By:         {claim.get('verified_by') or 'N/A'}")
    print(f"Authority:           {claim.get('verification_authority') or 'legacy/advisory'}")
    print(f"Verification Mode:   {claim.get('verification_mode') or 'legacy'}")

    author_executor = claim.get("author_executor")
    if isinstance(author_executor, dict):
        print(
            f"Author Executor:      uid {author_executor.get('uid')} "
            f"({author_executor.get('user')})"
        )
    verification_executor = claim.get("verification_executor")
    if isinstance(verification_executor, dict):
        print(
            f"Verifier Executor:    uid {verification_executor.get('uid')} "
            f"({verification_executor.get('user')})"
        )

    if claim.get("verdict"):
        print(f"Verdict:             {claim.get('verdict')}")

    print(f"Required Gate:       {claim.get('required_gate') or 'N/A'}")
    if claim.get("type") == "supervision_credit":
        print(f"Supervision Layer:   {claim.get('supervision_layer') or 'AMBIGUOUS (no layer)'}")
    print("-" * 80)
    print("Evidence References:")
    refs = claim.get("evidence_refs") or []
    if refs:
        for r in refs:
            print(f"  - {r}")
    else:
        print("  (None)")
    print("=" * 80)
    return 0


def usage_summary_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    usage_dir = os.path.join(op_dir, "usage")
    usage_files = glob.glob(os.path.join(usage_dir, "*.yaml"))

    records = []
    for filepath in usage_files:
        data = load_yaml(filepath) or []
        if isinstance(data, list):
            records.extend(data)

    if not records:
        print("No usage records found in ledger.")
        return 0

    if getattr(args, "offload_audit", False):
        frontier_spend = 0.0
        avoidable_spend = 0.0
        for r in records:
            lane = r.get("lane") or "unknown"
            task_class = r.get("task_class") or "unknown"
            cost = (
                float(r.get("cost_estimate_usd") or 0.0)
                if r.get("cost_estimate_usd") is not None
                else 0.0
            )
            if lane.startswith("frontier_"):
                frontier_spend += cost
                if task_class == "bounded":
                    avoidable_spend += cost
        pct = (avoidable_spend / frontier_spend * 100.0) if frontier_spend > 0 else 0.0
        print(
            f"{pct:.1f}% of frontier spend went to bounded work (avoidable -> route to an explicit lower-cost lane)."
        )
        return 0

    def print_records_summary(recs, title, show_cost):
        total_cost = 0.0
        run_count = 0
        total_tool_calls = 0
        outcome_counts = {}

        task_groups = {}
        harness_groups = {}
        model_groups = {}

        for r in recs:
            cost = (
                float(r.get("cost_estimate_usd") or 0.0)
                if r.get("cost_estimate_usd") is not None
                else 0.0
            )
            task_id = r.get("task_id") or "unknown"
            harness_id = r.get("harness_id") or "unknown"
            model = r.get("model") or "unknown"
            outcome = r.get("outcome") or "unknown"

            activity = r.get("activity", {})
            tool_calls = int(activity.get("tool_calls") or 0)
            turns = int(activity.get("turns") or 0)
            wall_clock = int(activity.get("wall_clock_s") or 0)

            total_cost += cost
            run_count += 1
            total_tool_calls += tool_calls
            outcome_counts[outcome] = outcome_counts.get(outcome, 0) + 1

            if task_id not in task_groups:
                task_groups[task_id] = {
                    "cost": 0.0,
                    "runs": 0,
                    "outcomes": {},
                    "tool_calls": 0,
                }
            task_groups[task_id]["cost"] += cost
            task_groups[task_id]["runs"] += 1
            task_groups[task_id]["outcomes"][outcome] = (
                task_groups[task_id]["outcomes"].get(outcome, 0) + 1
            )
            task_groups[task_id]["tool_calls"] += tool_calls

            if harness_id not in harness_groups:
                harness_groups[harness_id] = {
                    "cost": 0.0,
                    "runs": 0,
                    "outcomes": {},
                    "tool_calls": 0,
                    "turns": 0,
                    "wall_clock_s": 0,
                }
            harness_groups[harness_id]["cost"] += cost
            harness_groups[harness_id]["runs"] += 1
            harness_groups[harness_id]["outcomes"][outcome] = (
                harness_groups[harness_id]["outcomes"].get(outcome, 0) + 1
            )
            harness_groups[harness_id]["tool_calls"] += tool_calls
            harness_groups[harness_id]["turns"] += turns
            harness_groups[harness_id]["wall_clock_s"] += wall_clock

            if model not in model_groups:
                model_groups[model] = {
                    "cost": 0.0,
                    "runs": 0,
                    "outcomes": {},
                    "tool_calls": 0,
                }
            model_groups[model]["cost"] += cost
            model_groups[model]["runs"] += 1
            model_groups[model]["outcomes"][outcome] = (
                model_groups[model]["outcomes"].get(outcome, 0) + 1
            )
            model_groups[model]["tool_calls"] += tool_calls

        print("=" * 80)
        print(title)
        print("=" * 80)
        print(f"Total Runs Captured: {run_count}")
        if show_cost:
            print(f"Total Estimated Cost: ${total_cost:.4f} USD")
        print(f"Total Tool Calls: {total_tool_calls}")
        print("-" * 80)
        print("Outcome Breakdown:")
        for out, cnt in sorted(outcome_counts.items()):
            print(f"  - {out}: {cnt}")
        print("=" * 80)

        if args.by_task:
            print("\n" + "=" * 80)
            print(f"SUMMARY BY TASK ({title})")
            print("=" * 80)
            for tid, data in sorted(task_groups.items()):
                print(f"Task ID:      {tid}")
                print(f"  Runs:       {data['runs']}")
                if show_cost:
                    print(f"  Est. Cost:  ${data['cost']:.4f} USD")
                print(f"  Tool Calls: {data['tool_calls']}")
                outs = ", ".join(f"{k}:{v}" for k, v in sorted(data["outcomes"].items()))
                print(f"  Outcomes:   {outs}")
                print("-" * 40)
            print("=" * 80)

        if args.by_harness:
            print("\n" + "=" * 80)
            print(f"SUMMARY BY HARNESS ({title})")
            print("=" * 80)
            for hid, data in sorted(harness_groups.items()):
                print(f"Harness ID:   {hid}")
                print(f"  Runs:       {data['runs']}")
                if show_cost:
                    print(f"  Est. Cost:  ${data['cost']:.4f} USD")
                print(f"  Tool Calls: {data['tool_calls']}")
                print(f"  Turns:      {data['turns']} (harness-internal; not cross-comparable)")
                print(
                    f"  Wall Clock: {data['wall_clock_s']}s (harness-internal; not cross-comparable)"
                )
                outs = ", ".join(f"{k}:{v}" for k, v in sorted(data["outcomes"].items()))
                print(f"  Outcomes:   {outs}")
                print("-" * 40)
            print("=" * 80)

        if args.by_model:
            print("\n" + "=" * 80)
            print(f"SUMMARY BY MODEL ({title})")
            print("=" * 80)
            for mid, data in sorted(model_groups.items()):
                print(f"Model ID:     {mid}")
                print(f"  Runs:       {data['runs']}")
                if show_cost:
                    print(f"  Est. Cost:  ${data['cost']:.4f} USD")
                print(f"  Tool Calls: {data['tool_calls']}")
                outs = ", ".join(f"{k}:{v}" for k, v in sorted(data["outcomes"].items()))
                print(f"  Outcomes:   {outs}")
                print("-" * 40)
            print("=" * 80)

        if getattr(args, "by_lane", False):
            print("\n" + "=" * 80)
            print(f"SUMMARY BY LANE × TASK_CLASS ({title})")
            print("=" * 80)
            lane_groups = {}
            for r in recs:
                ln = r.get("lane") or "unknown"
                tc = r.get("task_class") or "unknown"
                cost = (
                    float(r.get("cost_estimate_usd") or 0.0)
                    if r.get("cost_estimate_usd") is not None
                    else 0.0
                )
                t_in = int(r.get("tokens_in") or 0) if r.get("tokens_in") is not None else 0
                t_out = int(r.get("tokens_out") or 0) if r.get("tokens_out") is not None else 0

                key = (ln, tc)
                if key not in lane_groups:
                    lane_groups[key] = {
                        "cost": 0.0,
                        "runs": 0,
                        "tokens_in": 0,
                        "tokens_out": 0,
                    }
                lane_groups[key]["cost"] += cost
                lane_groups[key]["runs"] += 1
                lane_groups[key]["tokens_in"] += t_in
                lane_groups[key]["tokens_out"] += t_out

            for (ln, tc), data in sorted(lane_groups.items(), key=lambda item: item[0]):
                print(f"Lane:         {ln}")
                print(f"Task Class:   {tc}")
                print(f"  Runs:       {data['runs']}")
                if show_cost:
                    print(f"  Est. Cost:  ${data['cost']:.4f} USD")
                print(f"  Tokens In:  {data['tokens_in']}")
                print(f"  Tokens Out: {data['tokens_out']}")
                print("-" * 40)

            avoidable_cost = 0.0
            for (ln, tc), data in lane_groups.items():
                if ln.startswith("frontier_") and tc == "bounded":
                    avoidable_cost += data["cost"]
            if show_cost:
                print(f"Avoidable Rollup Cost: ${avoidable_cost:.4f} USD")
            print("=" * 80)

        if getattr(args, "by_machine", False):
            print("\n" + "=" * 80)
            print(f"SUMMARY BY MACHINE × HARNESS ({title})")
            print("=" * 80)
            machine_groups = {}
            for r in recs:
                executor = r.get("executor") or {}
                mc = executor.get("machine") or "unknown"
                hid = r.get("harness_id") or "unknown"
                cost = (
                    float(r.get("cost_estimate_usd") or 0.0)
                    if r.get("cost_estimate_usd") is not None
                    else 0.0
                )
                t_in = int(r.get("tokens_in") or 0) if r.get("tokens_in") is not None else 0
                t_out = int(r.get("tokens_out") or 0) if r.get("tokens_out") is not None else 0

                key = (mc, hid)
                if key not in machine_groups:
                    machine_groups[key] = {
                        "cost": 0.0,
                        "runs": 0,
                        "tokens_in": 0,
                        "tokens_out": 0,
                    }
                machine_groups[key]["cost"] += cost
                machine_groups[key]["runs"] += 1
                machine_groups[key]["tokens_in"] += t_in
                machine_groups[key]["tokens_out"] += t_out

            for (mc, hid), data in sorted(machine_groups.items(), key=lambda item: item[0]):
                print(f"Machine:      {mc}")
                print(f"Harness:      {hid}")
                print(f"  Runs:       {data['runs']}")
                if show_cost:
                    print(f"  Est. Cost:  ${data['cost']:.4f} USD")
                print(f"  Tokens In:  {data['tokens_in']}")
                print(f"  Tokens Out: {data['tokens_out']}")
                print("-" * 40)
            print("=" * 80)

    if args.metering:
        tokens_recs = []
        activity_recs = []
        for r in records:
            m = r.get("metering")
            if not m:
                m = "activity" if r.get("harness_id") == "gemini-agy" else "tokens"
            if m == "tokens":
                tokens_recs.append(r)
            else:
                activity_recs.append(r)

        if tokens_recs:
            print_records_summary(
                tokens_recs, "OPERATOR USAGE & COST SUMMARY (TOKENS-METERED)", True
            )
            if activity_recs:
                print()
        if activity_recs:
            print_records_summary(
                activity_recs,
                "OPERATOR USAGE & ACTIVITY SUMMARY (ACTIVITY-METERED)",
                False,
            )
    else:
        print_records_summary(records, "OPERATOR USAGE & COST SUMMARY", True)

    return 0


def handoff_add_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task
    preflight_projection(op_dir, task_path, task)

    handoff_id = get_next_handoff_id(op_dir, task_id)
    by = args.by if args.by else task.get("assigned_harness") or "operator"
    utc_now = get_current_utc_iso()

    handoff_data = {}
    if args.file:
        if args.file == "-":
            if sys.stdin.isatty():
                print("Paste your structured YAML handoff (press Ctrl-D to save):")
            try:
                raw = sys.stdin.read()
                handoff_data = yaml.safe_load(raw) or {}
            except Exception as e:
                print(f"Error parsing stdin YAML: {e}", file=sys.stderr)
                return 1
        else:
            if not os.path.exists(args.file):
                print(f"Error: Handoff file '{args.file}' not found.", file=sys.stderr)
                return 1
            handoff_data = load_yaml(args.file) or {}
    else:
        if (
            not (
                args.changed
                or args.verified
                or args.claimed
                or args.open
                or args.assumptions
                or args.next_action
            )
            and not sys.stdin.isatty()
        ):
            try:
                raw = sys.stdin.read()
                handoff_data = yaml.safe_load(raw) or {}
            except Exception as e:
                print(f"Error parsing piped YAML: {e}", file=sys.stderr)
                return 1
        else:
            handoff_data = {
                "what_changed": args.changed,
                "what_verified": args.verified,
                "what_claimed": args.claimed,
                "what_remains_open": args.open,
                "unchecked_assumptions": args.assumptions,
                "next_action": args.next_action,
            }

    if not isinstance(handoff_data, dict):
        print("Error: Handoff payload must be a YAML mapping/object.", file=sys.stderr)
        return 1

    handoff_fields = [
        "what_changed",
        "what_verified",
        "what_claimed",
        "what_remains_open",
        "unchecked_assumptions",
        "next_action",
    ]
    if not any(handoff_data.get(field) for field in handoff_fields):
        print(
            "Error: Handoff is empty. Provide at least one structured handoff field.",
            file=sys.stderr,
        )
        return 1

    final_data = {
        "handoff_id": handoff_id,
        "task_id": task_id,
        "by": by,
        "created_at": utc_now,
        "what_changed": handoff_data.get("what_changed"),
        "what_verified": handoff_data.get("what_verified"),
        "what_claimed": handoff_data.get("what_claimed"),
        "what_remains_open": handoff_data.get("what_remains_open"),
        "unchecked_assumptions": handoff_data.get("unchecked_assumptions"),
        "next_action": handoff_data.get("next_action"),
        "executor": get_executor_identity(),
    }

    ho_file_path = os.path.join(op_dir, "handoffs", task_id, f"{handoff_id}.yaml")
    save_yaml(final_data, ho_file_path)

    task["handoffs"] = task.get("handoffs") or []
    ref_str = f"handoffs/{task_id}/{handoff_id}.yaml"
    if ref_str not in task["handoffs"]:
        task["handoffs"].append(ref_str)

    task["updated_at"] = utc_now

    if final_data["next_action"]:
        task["next_action"] = final_data["next_action"]

    save_yaml(task, task_path)

    print(f"Successfully recorded handoff '{handoff_id}' for task '{task_id}'.")
    return 0


def session_start_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    harness_id = args.harness_id
    harness_path = os.path.join(op_dir, "harnesses", f"{harness_id}.yaml")
    if not os.path.exists(harness_path):
        print(
            f"Error: Harness '{harness_id}' not found in harnesses/ registry.",
            file=sys.stderr,
        )
        return 1

    harness = load_yaml(harness_path)
    if not harness:
        return 1
    preflight_projection(op_dir, task_path, task)

    if task.get("status") == "running" and not args.force:
        print(
            f"Error: Task '{task_id}' is already running. Use --force to override.",
            file=sys.stderr,
        )
        return 1

    brief_content = generate_brief_markdown(op_dir, task_id, harness_id)
    if not brief_content:
        print("Error: Brief generation failed.", file=sys.stderr)
        return 1

    harness_display_name = harness.get("display_name", harness_id)
    export_filename = f"{task_id}.{harness_id}.export.md"
    export_path = os.path.join(op_dir, "briefs", export_filename)

    export_header = (
        f"<!-- SYSTEM INSTRUCTIONS FOR {harness_display_name.upper()} -->\n"
        f"<!-- PASTE THIS ENTIRE BRIEF INTO THE CHAT SESSION TO INITIALIZE STATE -->\n"
        f"<!-- DO NOT INVENT OR ASSUME DETAILS NOT DOCUMENTED IN THIS BRIEF -->\n\n"
    )

    full_export_content = export_header + brief_content

    try:
        with open(export_path, "w") as f:
            f.write(full_export_content)
    except Exception as e:
        print(f"Error writing export brief: {e}", file=sys.stderr)
        return 1

    usage_id = get_next_usage_id(op_dir)
    utc_now = get_current_utc_iso()
    date_str = utc_now.split("T")[0]

    usage_record = {
        "usage_id": usage_id,
        "task_id": task_id,
        "harness_id": harness_id,
        "model": harness.get("model", "unknown"),
        "started_at": utc_now,
        "ended_at": None,
        "cost_estimate_usd": 0.0,
        "outcome": "unknown",
        "raw_payload": "Session started via operator session-start.",
        "executor": get_executor_identity(),
        "lane": getattr(args, "lane", None) or "unknown",
        "task_class": getattr(args, "task_class", None) or "unknown",
    }

    if getattr(args, "lane", None) is not None:
        usage_record.setdefault("field_sources", {})["lane"] = "manual"
    if getattr(args, "task_class", None) is not None:
        usage_record.setdefault("field_sources", {})["task_class"] = "manual"

    usage_file_path = os.path.join(op_dir, "usage", f"{date_str}.yaml")
    records = []
    if os.path.exists(usage_file_path):
        records = load_yaml(usage_file_path) or []
    records.append(usage_record)
    save_yaml(records, usage_file_path)

    task["status"] = "running"
    task["updated_at"] = utc_now
    save_yaml(task, task_path)

    print("=" * 80)
    print(f"SESSION INITIALIZED: {task_id} @ {harness_display_name.upper()}")
    print("=" * 80)
    print(f"1. Export brief written to: {export_path}")
    print(
        f"2. Usage placeholder recorded: {usage_id} (${usage_record['cost_estimate_usd']:.2f} USD)"
    )
    print("-" * 80)

    launch_cmd = harness.get("command")
    if launch_cmd:
        print(f"Launch command: {launch_cmd}")
    else:
        print(f"Launch instructions: Open your target environment for {harness_display_name}.")

    print("Paste instructions:")
    print(
        f"  Copy the entire content of {export_filename} and paste it into the start of your session."
    )
    print("=" * 80)

    return 0


def find_usage_record(
    op_dir: str, usage_id: str
) -> tuple[str, list, int] | tuple[None, None, None]:
    """Search all date YAML files in usage/ to find a usage_id."""
    usage_dir = os.path.join(op_dir, "usage")
    if not os.path.isdir(usage_dir):
        return None, None, None
    usage_files = glob.glob(os.path.join(usage_dir, "*.yaml"))
    for filepath in usage_files:
        records = load_yaml(filepath) or []
        for idx, rec in enumerate(records):
            if rec.get("usage_id") == usage_id:
                return filepath, records, idx
    return None, None, None


def get_open_usage_ids_for_task(op_dir: str, task_id: str) -> list[str]:
    """Return usage IDs for records on task_id that do not have ended_at set."""
    usage_dir = os.path.join(op_dir, "usage")
    if not os.path.isdir(usage_dir):
        return []

    open_ids = []
    for filepath in glob.glob(os.path.join(usage_dir, "*.yaml")):
        records = load_yaml(filepath) or []
        for rec in records:
            if rec.get("task_id") == task_id and rec.get("ended_at") is None:
                uid = rec.get("usage_id")
                if uid:
                    open_ids.append(uid)
    return open_ids


def session_end_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    if enrollment:
        if args.status in ("verified", "complete"):
            print(
                "Error: Task status transitions to 'verified' or 'complete' are restricted "
                "under P3 authority. Use 'operator task-transition' instead.",
                file=sys.stderr,
            )
            return 1

    usage_id = args.usage_id
    if not usage_id:
        print("Error: Specify a usage ID (e.g. usage-0001).", file=sys.stderr)
        return 1
    filepath, records, idx = find_usage_record(op_dir, usage_id)
    if not filepath:
        print(f"Error: Usage record '{usage_id}' not found.", file=sys.stderr)
        return 1

    usage_record = records[idx]

    if usage_record.get("ended_at") is not None and not args.force:
        print(
            f"Error: Usage record '{usage_id}' is already closed (ended_at: {usage_record.get('ended_at')}). Use --force to override.",
            file=sys.stderr,
        )
        return 1

    task_id = usage_record.get("task_id")
    task = None
    task_path = None
    if task_id:
        resolved_task = resolve_task_record(op_dir, task_id)
        if not resolved_task:
            return 1
        _, task_path, task = resolved_task
        preflight_projection(op_dir, task_path, task)

    # Phase 3: optional session crystal attach before closing usage (draft only).
    attach_crystal = getattr(args, "attach_crystal", None)
    if attach_crystal:
        if not task_id or not task:
            msg = "session-end --attach-crystal requires the usage record to reference a task"
            if getattr(args, "require_crystal", False):
                print(f"Error: {msg}.", file=sys.stderr)
                return 1
            print(f"Warning: {msg}; closing session without crystal attach.")
        else:
            crystal_path, err = resolve_attach_crystal_path(attach_crystal, task.get("repo"))
            if err or not crystal_path:
                msg = err or "crystal path unresolved"
                if getattr(args, "require_crystal", False):
                    print(f"Error: session-end --attach-crystal: {msg}", file=sys.stderr)
                    return 1
                print(f"Warning: session-end --attach-crystal: {msg}; closing without attach.")
            else:
                attach_rc, evidence_id = attach_session_crystal_for_task(
                    op_dir,
                    task_id,
                    crystal_path,
                    by=getattr(args, "by", None),
                    notes=f"session-end attach for {usage_id}",
                )
                if attach_rc != 0:
                    if getattr(args, "require_crystal", False):
                        return attach_rc
                    print(
                        "Warning: session-end crystal attach failed; closing session anyway.",
                        file=sys.stderr,
                    )
                else:
                    bridge = usage_record.get("crystal_bridge")
                    if not isinstance(bridge, dict):
                        bridge = {}
                    bridge = dict(bridge)
                    bridge.update(
                        {
                            "session_end_attach_at": get_current_utc_iso(),
                            "crystal_path": os.path.abspath(crystal_path),
                            "evidence_id": evidence_id,
                        }
                    )
                    usage_record["crystal_bridge"] = bridge
                    print(
                        f"session-end: attached crystal {os.path.basename(crystal_path)} "
                        f"as {evidence_id or 'evidence'}."
                    )

    utc_now = get_current_utc_iso()
    usage_record["ended_at"] = utc_now
    usage_record["outcome"] = args.outcome
    usage_record["cost_estimate_usd"] = args.cost

    raw_payload = ""
    if not sys.stdin.isatty():
        raw_payload = sys.stdin.read().strip()
    elif args.paste:
        print("Paste your final /status or notes (press Ctrl-D to save):")
        try:
            raw_payload = sys.stdin.read().strip()
        except KeyboardInterrupt:
            print("\nAborted.")
            return 1

    if raw_payload:
        usage_record["raw_payload"] = raw_payload

    usage_record["executor"] = get_executor_identity()
    save_yaml(records, filepath)

    if task:
        task["updated_at"] = utc_now
        open_usage_ids = get_open_usage_ids_for_task(op_dir, task_id)
        if args.status:
            task["status"] = args.status
        elif task.get("status") == "running" and not open_usage_ids:
            # Task is no longer running since all sessions for it are closed.
            # Default back to assigned if no explicit status is provided.
            task["status"] = "assigned"
        save_yaml(task, task_path)

    print(f"Successfully closed usage session '{usage_id}'.")
    return 0


def session_list_cmd(args) -> int:
    op_dir = find_operator_dir()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    usage_dir = os.path.join(op_dir, "usage")
    if not os.path.isdir(usage_dir):
        print("No sessions found in ledger.")
        return 0

    usage_files = glob.glob(os.path.join(usage_dir, "*.yaml"))

    records = []
    for filepath in usage_files:
        data = load_yaml(filepath) or []
        if isinstance(data, list):
            records.extend(data)

    if not records:
        print("No sessions found in ledger.")
        return 0

    records.sort(key=lambda r: r.get("usage_id", ""))

    filtered_records = []
    for r in records:
        if args.open and r.get("ended_at") is not None:
            continue
        if args.task_id and r.get("task_id") != args.task_id:
            continue
        if args.harness_id and r.get("harness_id") != args.harness_id:
            continue
        filtered_records.append(r)

    if not filtered_records:
        print("No sessions matched the filters.")
        return 0

    print("=" * 110)
    print(
        f"{'USAGE ID':<12} | {'TASK ID':<25} | {'HARNESS':<10} | {'STATUS':<8} | {'STARTED':<25} | {'ENDED'}"
    )
    print("=" * 110)
    for r in filtered_records:
        usage_id = r.get("usage_id", "unknown")
        task_id = r.get("task_id", "unknown")
        harness = r.get("harness_id", "unknown")
        status = "open" if r.get("ended_at") is None else "closed"
        started = r.get("started_at") or "-"
        ended = r.get("ended_at") or "-"
        print(
            f"{usage_id:<12} | {task_id:<25} | {harness:<10} | {status:<8} | {started:<25} | {ended}"
        )
    print("=" * 110)
    return 0


def task_transition_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print(
            "Error: No .operator directory found. Run 'operator init' first.",
            file=sys.stderr,
        )
        return 1

    if not enrollment:
        print(
            "Error: task-transition is only supported in enrolled repositories.",
            file=sys.stderr,
        )
        return 1
    ledger_id = enrollment.ledger_id
    socket_path = enrollment.socket_path

    # Resolve task record
    resolved_task = resolve_task_record(op_dir, args.task_id)
    if not resolved_task:
        return 1
    task_id, task_path, task = resolved_task

    # Validate status and claim
    if args.status == "verified":
        if not args.claim:
            print("Error: --claim is required for verified status transition.", file=sys.stderr)
            return 1
        resolved_claim = resolve_claim_record(op_dir, args.claim)
        if not resolved_claim:
            return 1
        claim_path, claim = resolved_claim
        if claim.get("task_id") != task_id:
            print(
                f"Error: Claim '{args.claim}' belongs to task '{claim.get('task_id')}', not '{task_id}'.",
                file=sys.stderr,
            )
            return 1
        claim_id = args.claim
    else:
        if args.claim:
            print("Error: complete transition does not accept claim_id.", file=sys.stderr)
            return 1
        claim_id = None

    # Prepare request
    operation_key = f"op-task-transition-{task_id}-{args.status}"
    try:
        expected = authority_projection.compile_expected(
            op_dir, "task.transition", task_id, claim_id
        )
    except RuntimeError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1

    request = {
        "action": "commit",
        "ledger_id": ledger_id,
        "operation_key": operation_key,
        "operation": {
            "kind": "task.transition",
            "task_id": task_id,
            "status": args.status,
        },
        "expected": expected,
        "blob": None,
    }
    if claim_id:
        request["operation"]["claim_id"] = claim_id

    # 1. Connect journal, prepare transaction
    conn_journal = authority_projection.ensure_journal(op_dir)
    failure = prepare_journal_transaction(conn_journal, operation_key, request)
    if failure is not None:
        return failure

    # 2. Connect client and send request
    try:
        client = authority_client.AuthorityClient(socket_path)
        res = client.send_request(request)
    except Exception as e:
        print(f"Error: Broker dispatch failed: {e}", file=sys.stderr)
        conn_journal.close()
        return 1

    if not res.get("ok"):
        authority_projection.discard_prepared_transaction(conn_journal, operation_key)
        print(f"Error: Transition rejected by broker: {res.get('error')}", file=sys.stderr)
        conn_journal.close()
        return 1

    # 3. Save receipt
    receipt = res["receipt"]
    authority_projection.commit_transaction(conn_journal, operation_key, receipt)

    # 4. Reconcile
    try:
        authority_projection.reconcile_projections(op_dir, ledger_id, client)
    except Exception as e:
        print(f"Warning: Projection reconciliation failed: {e}", file=sys.stderr)
        conn_journal.close()
        return 0

    conn_journal.close()
    print(f"Successfully transitioned task '{task_id}' to status '{args.status}'.")
    return 0


def authority_reconcile_cmd(args) -> int:
    enrollment, op_dir = resolve_operator_context()
    if not op_dir:
        print("Error: No .operator directory found.", file=sys.stderr)
        return 1

    if not enrollment:
        print(
            "Error: authority-reconcile is only supported in enrolled repositories.",
            file=sys.stderr,
        )
        return 1
    ledger_id = enrollment.ledger_id
    socket_path = enrollment.socket_path
    acknowledge = bool(getattr(args, "acknowledge_store_reset", False))

    try:
        client = authority_client.AuthorityClient(socket_path)
        last_digest, seq = authority_projection.reconcile_projections(
            op_dir,
            ledger_id,
            client,
            acknowledge_store_reset=acknowledge,
        )
        if acknowledge:
            print(
                "Acknowledged store reset; re-projected from the current broker incarnation "
                f"through commit sequence {seq}."
            )
        else:
            print(f"Successfully reconciled with broker through commit sequence {seq}.")
        return 0
    except Exception as e:
        print(f"Error: Reconciliation failed: {e}", file=sys.stderr)
        return 1


# --- CLI Setup ---


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Operator Control Plane CLI - Coordination and Ledger Tool for AI Coding Harnesses."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # operator init
    subparsers.add_parser(
        "init", help="Initialize .operator ledger directories in the current folder."
    )

    # operator task create
    create_parser = subparsers.add_parser("task-create", help="Create a new task in the ledger.")
    create_parser.add_argument("--objective", "-o", required=True, help="Task objective statement.")
    create_parser.add_argument(
        "--id",
        dest="task_id",
        default=None,
        help="Explicit task ID slug (auto-slugified if omitted).",
    )
    create_parser.add_argument(
        "--repo",
        "-r",
        default=".",
        help="Path to repository under task (defaults to current folder).",
    )
    create_parser.add_argument("--assign", "-a", default=None, help="Harness ID to assign.")
    create_parser.add_argument(
        "--review",
        dest="review",
        default=None,
        help="Harness ID to assign as reviewer.",
    )
    create_parser.add_argument(
        "--assumption",
        "-m",
        action="append",
        dest="assumptions",
        default=[],
        help="Open assumptions to document.",
    )

    # operator task show
    show_parser = subparsers.add_parser(
        "task-show", help="Show details of the current or specified task."
    )
    show_parser.add_argument(
        "task_id",
        nargs="?",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    show_parser.add_argument(
        "--id",
        dest="task_id_flag",
        default=None,
        help="Task ID (alias for positional task_id).",
    )

    # operator claim add
    claim_parser = subparsers.add_parser(
        "claim-add", help="Register a claim for the current or specified task."
    )
    claim_parser.add_argument(
        "--type",
        "-t",
        required=True,
        choices=[
            "file_exists",
            "test_passes",
            "numeric_measurement",
            "real_data",
            "model_output",
            "firmware_behavior",
            "deployment_state",
            "supervision_credit",
            "paper_or_report_claim",
        ],
        help="Type of claim.",
    )
    claim_parser.add_argument("--text", "-x", required=True, help="Claim description text.")
    claim_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    claim_parser.add_argument(
        "--by",
        default=None,
        help="Actor who made the claim (defaults to assigned harness).",
    )
    claim_parser.add_argument(
        "--gate", "-g", default=None, help="Required verification gate description."
    )
    claim_parser.add_argument(
        "--layer",
        "-l",
        default=None,
        choices=["design", "execution", "evidence", "release", "end_to_end"],
        help="Supervision layer (required for supervision_credit claims per FR-12).",
    )

    # operator claim show
    claim_show_parser = subparsers.add_parser(
        "claim-show", help="Show details of a registered claim."
    )
    claim_show_parser.add_argument(
        "claim_id", nargs="?", default=None, help="Claim ID (e.g. claim-0001)."
    )
    claim_show_parser.add_argument(
        "--id",
        dest="claim_id_flag",
        default=None,
        help="Claim ID (alias for positional claim_id).",
    )

    # operator evidence attach
    evidence_parser = subparsers.add_parser(
        "evidence-attach",
        help="Attach an evidence artifact to a task and optionally verify a claim.",
    )
    evidence_parser.add_argument(
        "path_or_url", help="Path to local file or URL of the evidence artifact."
    )
    evidence_parser.add_argument("--claim", "-c", default=None, help="Associated Claim ID.")
    evidence_parser.add_argument(
        "--type",
        "-t",
        default="run_log",
        choices=EVIDENCE_TYPES,
        help="Type of evidence.",
    )

    # operator crystal attach (Crystal↔Ledger Phase 1)
    crystal_parser = subparsers.add_parser(
        "crystal-attach",
        help=(
            "Attach an agent-crystallize checkpoint/session Markdown file as draft "
            "session_crystal evidence (never sets verification status)."
        ),
    )
    crystal_parser.add_argument(
        "path",
        help="Path to a local crystal Markdown file under .agent-crystals/ (or a fixture).",
    )
    crystal_parser.add_argument("--claim", "-c", default=None, help="Associated Claim ID.")
    crystal_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    crystal_parser.add_argument("--by", default=None, help="Actor who produced the crystal attach.")
    crystal_parser.add_argument(
        "--hash",
        default=None,
        help="Expected SHA-256 digest (validated against crystal file bytes before write).",
    )
    crystal_parser.add_argument(
        "--notes",
        "-n",
        default=None,
        help="Optional notes about the crystal attach.",
    )

    # operator crystal import (Crystal↔Ledger Phase 2)
    crystal_import_parser = subparsers.add_parser(
        "crystal-import",
        help=(
            "Import a crystal as draft session_crystal evidence and extract "
            "Tests And Verification bullets as draft test_passes claims "
            "(never sets verification status)."
        ),
    )
    crystal_import_parser.add_argument(
        "path",
        help="Path to a local crystal Markdown file.",
    )
    crystal_import_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    crystal_import_parser.add_argument(
        "--by", default=None, help="Actor who performed the crystal import."
    )
    crystal_import_parser.add_argument(
        "--notes",
        "-n",
        default=None,
        help="Optional notes stored on the auto-attached crystal evidence.",
    )
    crystal_import_parser.add_argument(
        "--open-loops-as-tasks",
        action="store_true",
        default=False,
        help=(
            "Also create new tasks from real Open Loops bullets "
            "(default: off; open loops are often stale)."
        ),
    )

    # operator crystal-bridge (Crystal↔Ledger Phase 3)
    crystal_bridge_parser = subparsers.add_parser(
        "crystal-bridge",
        help=(
            "Hook/session bridge: SessionStart opens operator session-start; "
            "Stop attaches the latest session crystal (draft only). "
            "Soft-no-op when no ledger is present."
        ),
    )
    crystal_bridge_parser.add_argument(
        "--event",
        default="SessionStart",
        choices=list(CRYSTAL_BRIDGE_EVENTS),
        help="Crystallize-compatible hook event name (default: SessionStart).",
    )
    crystal_bridge_parser.add_argument(
        "--harness",
        default=None,
        help="Harness id or crystallize alias (required for SessionStart: claude, codex, …).",
    )
    crystal_bridge_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    crystal_bridge_parser.add_argument(
        "--crystal",
        default=None,
        help="For Stop: crystal path or 'auto' (default auto under .agent-crystals/sessions/).",
    )
    crystal_bridge_parser.add_argument(
        "--by", default=None, help="Actor recorded on attach (Stop path)."
    )
    crystal_bridge_parser.add_argument(
        "--force",
        action="store_true",
        help="SessionStart: force session-start even if task is already running.",
    )
    crystal_bridge_parser.add_argument(
        "--require-crystal",
        action="store_true",
        default=False,
        help="Stop: exit non-zero if no session crystal can be attached.",
    )
    crystal_bridge_parser.add_argument(
        "--lane",
        choices=["local", "frontier_driver", "frontier_author"],
        default=None,
        help="Optional lane passed through to session-start.",
    )
    crystal_bridge_parser.add_argument(
        "--class",
        dest="task_class",
        choices=["bounded", "hard"],
        default=None,
        help="Optional task class passed through to session-start.",
    )
    evidence_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    evidence_parser.add_argument("--by", default=None, help="Actor who produced the evidence.")
    evidence_parser.add_argument(
        "--hash",
        default=None,
        help=(
            "Expected SHA-256 digest (validated against local file bytes before the evidence "
            "write; required as an assertion for remote evidence)."
        ),
    )
    evidence_parser.add_argument(
        "--notes",
        "-n",
        default=None,
        help="Optional description or details about the evidence.",
    )
    evidence_parser.add_argument(
        "--verify-cmd",
        dest="verify_cmd",
        default=None,
        help="Command recorded for later verification; evidence-attach and doctor do not run it.",
    )
    evidence_parser.add_argument(
        "--status",
        choices=["verified", "false", "quarantined"],
        default=None,
        help="Update associated claim status.",
    )
    evidence_parser.add_argument(
        "--verdict", "-v", default=None, help="Verification verdict text for the claim."
    )
    evidence_parser.add_argument(
        "--verified-by", default=None, help="Actor who verified the claim."
    )

    # operator usage add
    usage_parser = subparsers.add_parser(
        "usage-add", help="Capture a pasted usage snippet / quota log."
    )
    usage_parser.add_argument(
        "--harness", "-a", required=True, help="Harness ID associated with this usage."
    )
    usage_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    usage_parser.add_argument("--model", "-m", default=None, help="Model ID used.")
    usage_parser.add_argument(
        "--outcome",
        "-o",
        default="unknown",
        choices=["useful", "partial", "no_go", "quarantined", "reverted", "unknown"],
        help="Outcome classification.",
    )
    usage_parser.add_argument(
        "--cost", "-c", type=float, default=0.0, help="Estimated cost in USD."
    )
    usage_parser.add_argument(
        "--file",
        "-f",
        default=None,
        help="Read raw payload verbatim from a file (use '-' or omit for stdin). Avoids shell quoting/expansion corruption.",
    )
    usage_parser.add_argument(
        "--lane",
        choices=["local", "frontier_driver", "frontier_author"],
        default=None,
        help="Lane classification.",
    )
    usage_parser.add_argument(
        "--class",
        dest="task_class",
        choices=["bounded", "hard"],
        default=None,
        help="Task class classification.",
    )

    # operator usage summary
    usage_summary_parser = subparsers.add_parser(
        "usage-summary", help="Show summary of usage across tasks/harnesses."
    )
    usage_summary_parser.add_argument(
        "--by-task", action="store_true", help="Group summary by task."
    )
    usage_summary_parser.add_argument(
        "--by-harness", action="store_true", help="Group summary by harness."
    )
    usage_summary_parser.add_argument(
        "--by-model", action="store_true", help="Group summary by model."
    )
    usage_summary_parser.add_argument(
        "--metering",
        action="store_true",
        help="Split summary into token/cost and activity-only blocks.",
    )
    usage_summary_parser.add_argument(
        "--by-lane", action="store_true", help="Group summary by lane × task_class."
    )
    usage_summary_parser.add_argument(
        "--by-machine",
        action="store_true",
        help="Group summary by machine × harness (executor provenance).",
    )
    usage_summary_parser.add_argument(
        "--offload-audit",
        action="store_true",
        help="Produce one-line offload audit report.",
    )

    # operator usage-import
    usage_import_parser = subparsers.add_parser(
        "usage-import", help="Import usage automatically from session logs."
    )
    usage_import_parser.add_argument(
        "--harness",
        required=True,
        choices=["claude", "codex", "gemini-agy"],
        help="Target harness logs to parse.",
    )
    usage_import_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to active task).",
    )
    usage_import_parser.add_argument(
        "--session-id",
        default=None,
        help="Force selection of a specific source session ID.",
    )
    usage_import_parser.add_argument(
        "--since",
        default=None,
        help="Import sessions started on or after this ISO8601 timestamp.",
    )
    usage_import_parser.add_argument(
        "--until",
        default=None,
        help="Import sessions started on or before this ISO8601 timestamp.",
    )
    usage_import_parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print the record(s) that would be imported without writing them.",
    )
    usage_import_parser.add_argument(
        "--source-dir",
        default=None,
        help="Alternate harness log directory (e.g. synced logs from another machine).",
    )
    usage_import_parser.add_argument(
        "--machine",
        default=None,
        help="Machine label for imported records (the producer, not the importer).",
    )

    # operator usage-annotate
    usage_annotate_parser = subparsers.add_parser(
        "usage-annotate", help="Manually override fields in a usage record."
    )
    usage_annotate_parser.add_argument("usage_id", help="Usage ID to annotate (e.g. usage-0001).")
    usage_annotate_parser.add_argument(
        "--active-start", default=None, help="Start timestamp of active coding time."
    )
    usage_annotate_parser.add_argument(
        "--active-end", default=None, help="End timestamp of active coding time."
    )
    usage_annotate_parser.add_argument(
        "--cost", type=float, default=None, help="Set manual cost in USD."
    )
    usage_annotate_parser.add_argument(
        "--quota-events", type=int, default=None, help="Set manual quota events count."
    )
    usage_annotate_parser.add_argument(
        "--note",
        default=None,
        help="Add manual notes or paste a snippet to raw_payload.",
    )
    usage_annotate_parser.add_argument(
        "--lane",
        choices=["local", "frontier_driver", "frontier_author", "unknown"],
        default=None,
        help="Set manual lane.",
    )
    usage_annotate_parser.add_argument(
        "--class",
        dest="task_class",
        choices=["bounded", "hard", "unknown"],
        default=None,
        help="Set manual task class.",
    )

    # operator brief
    brief_parser = subparsers.add_parser("brief", help="Generate a harness-specific brief.")
    brief_parser.add_argument("--for", dest="harness_id", required=True, help="Target Harness ID.")
    brief_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )

    # operator export-brief
    export_brief_parser = subparsers.add_parser(
        "export-brief", help="Export a harness-specific brief formatted for copy-paste."
    )
    export_brief_parser.add_argument(
        "--for", dest="harness_id", required=True, help="Target Harness ID."
    )
    export_brief_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )

    # operator doctor
    doctor_parser = subparsers.add_parser(
        "doctor", help="Run a read-only consistency check across the operator ledger."
    )
    doctor_parser.add_argument(
        "--audit",
        action="store_true",
        help="Perform additional independent-supervision checks (Phase 3.2).",
    )

    # operator task-list
    subparsers.add_parser("task-list", help="List all tasks in the ledger with outcome summaries.")

    # operator claim-list
    claim_list_parser = subparsers.add_parser(
        "claim-list", help="List registered claims in the ledger."
    )
    claim_list_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Filter claims by Task ID (use 'all' for all).",
    )

    # operator handoff-add
    handoff_parser = subparsers.add_parser(
        "handoff-add", help="Record a handoff closeout for the task."
    )
    handoff_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    handoff_parser.add_argument("--by", default=None, help="Actor who produced the handoff.")
    handoff_parser.add_argument(
        "--changed", "-c", default=None, help="Description of what changed."
    )
    handoff_parser.add_argument(
        "--verified", "-v", default=None, help="Description of what was verified."
    )
    handoff_parser.add_argument(
        "--claimed",
        "-k",
        default=None,
        help="Description of what was claimed but unverified.",
    )
    handoff_parser.add_argument(
        "--open", "-o", default=None, help="Description of what remains open."
    )
    handoff_parser.add_argument(
        "--assumptions",
        "-u",
        default=None,
        help="Description of unchecked assumptions.",
    )
    handoff_parser.add_argument(
        "--next-action", "-n", default=None, help="First recommended next action."
    )
    handoff_parser.add_argument(
        "--file",
        "-f",
        default=None,
        help="YAML file path containing handoff fields (use '-' for stdin).",
    )

    # operator session-start
    session_start_parser = subparsers.add_parser(
        "session-start",
        help="Start a coding session for a task using a target harness.",
    )
    session_start_parser.add_argument(
        "--task",
        dest="task_id",
        default=None,
        help="Task ID (defaults to current active task).",
    )
    session_start_parser.add_argument(
        "--harness", dest="harness_id", required=True, help="Harness ID to start."
    )
    session_start_parser.add_argument(
        "--force",
        action="store_true",
        help="Force session start even if task is running.",
    )
    session_start_parser.add_argument(
        "--lane",
        choices=["local", "frontier_driver", "frontier_author"],
        default=None,
        help="Lane classification.",
    )
    session_start_parser.add_argument(
        "--class",
        dest="task_class",
        choices=["bounded", "hard"],
        default=None,
        help="Task class classification.",
    )

    # operator session-end
    session_end_parser = subparsers.add_parser("session-end", help="End session")
    session_end_parser.add_argument("usage_id", nargs="?", default=None, help="Usage ID to close.")
    session_end_parser.add_argument(
        "--usage", dest="usage_id_flag", default=None, help="Usage ID to close."
    )
    session_end_parser.add_argument(
        "--outcome",
        choices=["useful", "partial", "no_go", "quarantined", "reverted", "unknown"],
        required=True,
        help="Outcome status.",
    )
    session_end_parser.add_argument(
        "--cost", type=float, required=True, help="Total cost incurred."
    )
    session_end_parser.add_argument(
        "--status",
        choices=["assigned", "verified", "quarantined", "complete"],
        default=None,
        help="Optionally update task status.",
    )
    session_end_parser.add_argument(
        "--force", action="store_true", help="Force closing an already closed session."
    )
    session_end_parser.add_argument(
        "--paste",
        action="store_true",
        help="Interactively paste final status notes from stdin.",
    )
    session_end_parser.add_argument(
        "--attach-crystal",
        default=None,
        metavar="AUTO_OR_PATH",
        help=(
            "Before closing, attach a session crystal as draft session_crystal evidence: "
            "'auto' picks the newest file under <task.repo>/.agent-crystals/sessions/, "
            "or pass an explicit Markdown path (Crystal↔Ledger Phase 3)."
        ),
    )
    session_end_parser.add_argument(
        "--require-crystal",
        action="store_true",
        default=False,
        help="With --attach-crystal, abort session-end if the crystal cannot be attached.",
    )

    # operator session-list
    session_list_parser = subparsers.add_parser("session-list", help="List coding sessions.")
    session_list_parser.add_argument(
        "--open", action="store_true", help="Filter for open sessions only."
    )
    session_list_parser.add_argument(
        "--task", dest="task_id", default=None, help="Filter by task ID."
    )
    session_list_parser.add_argument(
        "--harness", dest="harness_id", default=None, help="Filter by harness ID."
    )

    # operator verify (Phase 3.1)
    verify_parser = subparsers.add_parser(
        "verify", help="Automated audit of a run directory artifacts (Phase 3)."
    )
    verify_parser.add_argument(
        "run_dir",
        help="Path to the agent run directory (e.g. docs/domain_runs/DBB-001a/gemini_cli_ladder).",
    )

    # operator task-transition
    transition_parser = subparsers.add_parser(
        "task-transition",
        help="Transition task state (verified or complete) via the authority broker.",
    )
    transition_parser.add_argument("--task", dest="task_id", required=True, help="Task ID.")
    transition_parser.add_argument(
        "--status", required=True, choices=["verified", "complete"], help="Target status."
    )
    transition_parser.add_argument(
        "--claim", default=None, help="Claim ID (required for status 'verified')."
    )

    # operator authority-reconcile
    reconcile_parser = subparsers.add_parser(
        "authority-reconcile", help="Reconcile local state with the authority broker."
    )
    reconcile_parser.add_argument(
        "--acknowledge-store-reset",
        action="store_true",
        help=(
            "Explicitly acknowledge that the authority store was rebuilt/replaced "
            "(store incarnation discontinuity). Resets local last_applied_sequence "
            "and re-projects from the new store. Never use this for ordinary outages."
        ),
    )

    args = parser.parse_args()

    if getattr(args, "command", None) == "task-show" and getattr(args, "task_id_flag", None):
        args.task_id = args.task_id_flag
    if getattr(args, "command", None) == "claim-show" and getattr(args, "claim_id_flag", None):
        args.claim_id = args.claim_id_flag
    if getattr(args, "command", None) == "session-end" and getattr(args, "usage_id_flag", None):
        args.usage_id = args.usage_id_flag

    # Mapping subcommand names to python functions.
    # Note: Using python command-group naming conventions to avoid bash collision.
    cmd_map = {
        "init": init_cmd,
        "task-create": task_create_cmd,
        "task-show": task_show_cmd,
        "task-list": task_list_cmd,
        "claim-add": claim_add_cmd,
        "claim-show": claim_show_cmd,
        "claim-list": claim_list_cmd,
        "evidence-attach": evidence_attach_cmd,
        "crystal-attach": crystal_attach_cmd,
        "crystal-import": crystal_import_cmd,
        "crystal-bridge": crystal_bridge_cmd,
        "usage-add": usage_add_cmd,
        "usage-import": usage_import_cmd,
        "usage-annotate": usage_annotate_cmd,
        "usage-summary": usage_summary_cmd,
        "handoff-add": handoff_add_cmd,
        "session-start": session_start_cmd,
        "session-end": session_end_cmd,
        "session-list": session_list_cmd,
        "brief": brief_cmd,
        "export-brief": export_brief_cmd,
        "doctor": doctor_cmd,
        "verify": verify_cmd,
        "task-transition": task_transition_cmd,
        "authority-reconcile": authority_reconcile_cmd,
    }

    lock_handle = None
    try:
        lock_handle = acquire_command_lock(args.command)
        return_code = cmd_map[args.command](args)
    except (LedgerWriteError, authority_client.EnrollmentError) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return_code = 1
    finally:
        release_command_lock(lock_handle)
    sys.exit(return_code)


if __name__ == "__main__":
    main()