"""Krineia Watcher — Service agent that observes the coordination bus and
creates Krineia cryptographic receipts for skill invocations.

This module is designed to support Krineia Invariant 5 (Trust-Root Separation)
when deployed as a structurally independent observer process that:
- Reads the coordination bus (read-only; never writes to it)
- Detects SKILL_INVOKE events posted by fleet agents
- Creates and signs receipts independently
- Writes receipts to its own append-only JSONL log

The observed agents NEVER:
- Generate their own receipts
- Hold signing keys for the receipt chain
- Control the receipt-generation process

Schema: spec/krineia_receipt.schema.json
"""

from __future__ import annotations

import argparse
import hashlib
import json
import logging
import os
import re
import signal
import sys
import time
import uuid
from collections import deque
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

# Make the repo root importable so we can delegate chain verification to the
# canonical tool implementation in tools/verify_chain.py. The daemon lives in
# daemon/ and the tool lives in tools/, so the repo root (this file's
# parent.parent) must be on sys.path for `import tools.verify_chain` to work.
_REPO_ROOT = str(Path(__file__).resolve().parent.parent)
if _REPO_ROOT not in sys.path:
    sys.path.insert(0, _REPO_ROOT)

from tools.verify_chain import (
    MAX_CHAIN_BYTES as MAX_RECEIPT_CHAIN_BYTES,
    MAX_KEY_REGISTRY_BYTES,
    MAX_LINE_BYTES as MAX_RECEIPT_LINE_BYTES,
    MAX_RECEIPTS,
    canonical_receipt,
    canonical_signed_body,
    parse_key_registry,
    sha256_hex,
    validate_receipt,
    verify_chain as _tool_verify_chain,
    verify_signature,
)

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

WATCHER_IDENTITY = "krineia-watcher"
RECEIPT_VERSION = "2.1.0"
EVENT_PREFIX = "skill.invoked"
DEFAULT_POLL_INTERVAL = 5  # seconds
DEFAULT_MAX_RECORDS_PER_POLL = 64
MAX_BUS_LINE_BYTES = 16 * 1024
MAX_BUS_BYTES_PER_POLL = 256 * 1024
MAX_OVERSIZED_RECORD_BYTES = 16 * 1024 * 1024
MAX_SKILL_NAME_BYTES = 256
MAX_MODE_BYTES = 64
MAX_SESSION_ID_BYTES = 256
MAX_BUS_TIMESTAMP_BYTES = 64
MAX_BUS_IDENTITY_BYTES = 128
MAX_BUS_TYPE_BYTES = 64
MAX_METADATA_TOKENS = 32
MAX_METADATA_KEY_BYTES = 64
MAX_METADATA_VALUE_BYTES = 256
MAX_TAIL_RECEIPTS = 1_000
MAX_REJECTION_MARKERS = 10_000
OVERSIZE_CONTINUITY_ANCHOR_BYTES = 4 * 1024
CURSOR_VERSION = 1
ZERO_DIGEST = "0" * 64
DEFAULT_BUS_PATH = os.environ.get(
    "HUMMBL_BUS_TSV",
    os.path.expanduser("~/projects/hummbl-governance/_state/coordination/messages.tsv"),
)
DEFAULT_RECEIPT_PATH = os.environ.get(
    "HUMMBL_RECEIPT_JSONL",
    os.path.expanduser("~/projects/hummbl-governance/_state/governance/skill_receipts.jsonl"),
)

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _utc_now() -> str:
    """Return current UTC timestamp in ISO 8601 Z format."""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


_REQUIRED_SKILL_TAGS = frozenset({"skill", "mode", "args_hash", "session"})
_TAG_RE = re.compile(r"\[([A-Za-z_][A-Za-z0-9_]*)=([^\]\r\n]+)\]")
_METADATA_RE = re.compile(
    r"([a-z][a-z0-9_.-]*)=([A-Za-z0-9][A-Za-z0-9_.:/\\@%+#,=-]*)"
)
_LOWER_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
_SKILL_NAME_RE = re.compile(r"^[A-Za-z0-9_.:/-]+$")
_MODE_RE = re.compile(r"^[A-Za-z0-9_.+-]+$")
_SESSION_RE = re.compile(r"^[A-Za-z0-9_.:/#-]+$")
_BUS_IDENTITY_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
_BUS_TYPE_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
_BUS_TIMESTAMP_RE = re.compile(
    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$"
)


def _utf8_len(value: str) -> int:
    return len(value.encode("utf-8"))


def _valid_bus_timestamp(value: str) -> bool:
    if _BUS_TIMESTAMP_RE.fullmatch(value) is None:
        return False
    try:
        datetime.fromisoformat(value[:-1] + "+00:00")
    except ValueError:
        return False
    return True


def _parse_skill_invoke(message: str) -> dict[str, str] | None:
    """Parse a SKILL_INVOKE bus message into structured fields.

    A canonical message begins with ``host=<safe-token>`` and may contain
    other unique, bounded ``key=value`` metadata tokens around exactly one of
    each required bracketed tag. Returns None if parsing fails.
    """
    if not message or _utf8_len(message) > MAX_BUS_LINE_BYTES:
        return None

    # Tokens are separated only by a single ASCII space. This makes residue,
    # embedded control characters, and alternate whitespace unambiguous.
    tokens = message.split(" ")
    if any(not token for token in tokens):
        return None
    first_metadata = _METADATA_RE.fullmatch(tokens[0])
    if first_metadata is None or first_metadata.group(1) != "host":
        return None

    result: dict[str, str] = {}
    metadata_keys: set[str] = set()
    metadata_count = 0
    for token in tokens:
        tag = _TAG_RE.fullmatch(token)
        if tag is not None:
            key = tag.group(1)
            value = tag.group(2)
            if key not in _REQUIRED_SKILL_TAGS or key in result:
                return None
            result[key] = value
            continue

        metadata = _METADATA_RE.fullmatch(token)
        if metadata is None:
            return None
        key, value = metadata.groups()
        # Required invocation fields are bracketed by contract; accepting an
        # unbracketed spelling would create a second, ambiguous grammar.
        if key in _REQUIRED_SKILL_TAGS or key in metadata_keys:
            return None
        metadata_count += 1
        if (
            metadata_count > MAX_METADATA_TOKENS
            or _utf8_len(key) > MAX_METADATA_KEY_BYTES
            or _utf8_len(value) > MAX_METADATA_VALUE_BYTES
        ):
            return None
        metadata_keys.add(key)

    if set(result) != _REQUIRED_SKILL_TAGS:
        return None
    if (
        not 1 <= _utf8_len(result["skill"]) <= MAX_SKILL_NAME_BYTES
        or _SKILL_NAME_RE.fullmatch(result["skill"]) is None
        or not 1 <= _utf8_len(result["mode"]) <= MAX_MODE_BYTES
        or _MODE_RE.fullmatch(result["mode"]) is None
        or _LOWER_SHA256_RE.fullmatch(result["args_hash"]) is None
        or not 1 <= _utf8_len(result["session"]) <= MAX_SESSION_ID_BYTES
        or _SESSION_RE.fullmatch(result["session"]) is None
    ):
        return None
    return result


class SourceIntegrityError(RuntimeError):
    """Raised when the append-only bus no longer matches its checkpoint."""


class SourceRecordAlreadyReceipted(RuntimeError):
    """Raised when replay reaches a source record already durably receipted."""


@dataclass(frozen=True)
class BusRecord:
    """One newline-terminated source record awaiting durable disposition."""

    start_offset: int
    end_offset: int
    record_sha256: str
    source_record_id: str
    entry: dict[str, str] | None
    rejection_reason: str | None = None


def _source_record_id(start_offset: int, record_sha256: str) -> str:
    return f"bus-tsv:{start_offset}:{record_sha256}"


def _source_file_identity(path: Path) -> str:
    stat = path.stat()
    return f"{stat.st_dev}:{stat.st_ino}"


def _atomic_write_json(path: Path, value: dict[str, Any]) -> None:
    """Atomically replace a small JSON state file after fsyncing its bytes."""
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    encoded = json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n"
    try:
        with open(temporary, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(encoded)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        _fsync_directory(path.parent)
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass


def _fsync_directory(path: Path) -> None:
    """Best-effort directory-entry durability on supporting filesystems."""
    try:
        directory_fd = os.open(path, os.O_RDONLY)
    except OSError:
        return
    try:
        os.fsync(directory_fd)
    except OSError:
        # Windows does not permit fsync on directory handles.
        pass
    finally:
        os.close(directory_fd)


def _bounded_tail(path: Path, n: int) -> list[dict[str, Any]]:
    """Read a convenience tail without unbounded count, line, or file work."""
    if isinstance(n, bool) or not isinstance(n, int) or n < 0:
        raise ValueError("tail count must be a non-negative integer")
    if n > MAX_TAIL_RECEIPTS:
        raise ValueError(f"tail count exceeds {MAX_TAIL_RECEIPTS}")
    if n == 0 or not path.exists():
        return []
    lines: deque[str] = deque(maxlen=n)
    total_bytes = 0
    line_number = 0
    with path.open("rb") as handle:
        while True:
            raw = handle.readline(MAX_RECEIPT_LINE_BYTES + 1)
            if not raw:
                break
            line_number += 1
            total_bytes += len(raw)
            if total_bytes > MAX_RECEIPT_CHAIN_BYTES:
                raise SourceIntegrityError("receipt chain exceeds tail read limit")
            if len(raw) > MAX_RECEIPT_LINE_BYTES:
                raise SourceIntegrityError(
                    f"receipt line {line_number} exceeds tail read limit"
                )
            try:
                text = raw.decode("utf-8", errors="strict")
            except UnicodeDecodeError as exc:
                raise SourceIntegrityError(
                    f"receipt line {line_number} is not valid UTF-8"
                ) from exc
            if text.strip():
                lines.append(text)
    result: list[dict[str, Any]] = []
    for text in lines:
        try:
            value = json.loads(text)
        except (json.JSONDecodeError, RecursionError) as exc:
            raise SourceIntegrityError(f"invalid receipt in tail: {exc}") from exc
        if not isinstance(value, dict):
            raise SourceIntegrityError("tail receipt is not a JSON object")
        result.append(value)
    return result


@contextmanager
def _exclusive_append_lock(lock_path: Path) -> Iterator[None]:
    """Hold a cross-platform advisory lock for one append transaction."""
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    with open(lock_path, "a+b") as handle:
        handle.seek(0, os.SEEK_END)
        if handle.tell() == 0:
            handle.write(b"\0")
            handle.flush()
        handle.seek(0)
        if os.name == "nt":
            import msvcrt

            msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
            try:
                yield
            finally:
                handle.seek(0)
                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
        else:
            import fcntl

            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)

# ---------------------------------------------------------------------------
# Receipt creation
# ---------------------------------------------------------------------------


class SkillReceiptWriter:
    """Creates and persists Krineia receipts for skill invocations.

    Each receipt is:
    - Cryptographically signed with Ed25519 via an externally held key
    - Hash-chained to the prior receipt (genesis uses zeros)
    - Append-only (never edited in place)

    Schema v2.0 is verification-only. This writer emits only strict v2.1
    receipts and refuses to generate one without an Ed25519 private key.
    """

    def __init__(
        self,
        receipt_path: str | Path,
        signing_private_key: str | bytes | None = None,
        signing_key_id: str = "krineia-assurance-observer-v1",
        verification_keys: dict[str, Any] | None = None,
        expected_existing_head: str | None = None,
        expected_existing_count: int | None = None,
        initialize_new_chain: bool = False,
    ) -> None:
        self.receipt_path = Path(receipt_path)
        self.receipt_path.parent.mkdir(parents=True, exist_ok=True)
        self.signing_key_id = signing_key_id
        key_material = (
            signing_private_key
            if signing_private_key is not None
            else os.environ.get("KRINEIA_ED25519_PRIVATE_KEY")
        )
        self._private_key = self._load_private_key(key_material)
        self._verification_keys = dict(verification_keys or {})
        self._last_hash: str | None = None
        self._chain_length: int | None = None
        if (expected_existing_head is None) != (expected_existing_count is None):
            raise ValueError(
                "expected_existing_head and expected_existing_count must be supplied together"
            )
        if initialize_new_chain and expected_existing_head is not None:
            raise ValueError(
                "initialize_new_chain cannot be combined with an expected existing checkpoint"
            )
        if expected_existing_head is not None:
            if _LOWER_SHA256_RE.fullmatch(expected_existing_head) is None:
                raise ValueError(
                    "expected_existing_head must be 64 lowercase hex characters"
                )
            if (
                isinstance(expected_existing_count, bool)
                or not isinstance(expected_existing_count, int)
                or expected_existing_count < 0
            ):
                raise ValueError("expected_existing_count must be non-negative")
            if expected_existing_count == 0 and expected_existing_head != ZERO_DIGEST:
                raise ValueError("an empty chain checkpoint must use the zero digest")
            self._checkpoint_head = expected_existing_head
            self._checkpoint_count = expected_existing_count
        elif initialize_new_chain:
            if self.receipt_path.exists() and self.receipt_path.stat().st_size != 0:
                raise RuntimeError(
                    "refusing to initialize a new chain over existing non-empty bytes"
                )
            self._checkpoint_head = ZERO_DIGEST
            self._checkpoint_count = 0
        else:
            self._checkpoint_head = None
            self._checkpoint_count = None
        if not self.receipt_path.exists() or self.receipt_path.stat().st_size == 0:
            self._last_hash = ZERO_DIGEST
            self._chain_length = 0

    @staticmethod
    def _load_private_key(key_material: str | bytes | None) -> Any | None:
        """Load a raw 32-byte Ed25519 private key without logging its value."""
        if key_material is None:
            return None
        try:
            from cryptography.hazmat.primitives.asymmetric.ed25519 import (
                Ed25519PrivateKey,
            )
        except ImportError as exc:
            raise RuntimeError(
                "schema 2.1 receipt emission requires the cryptography package"
            ) from exc

        if isinstance(key_material, str):
            if len(key_material) != 64 or any(
                char not in "0123456789abcdef" for char in key_material
            ):
                raise ValueError(
                    "KRINEIA_ED25519_PRIVATE_KEY must be 64 lowercase hex characters"
                )
            try:
                raw = bytes.fromhex(key_material)
            except ValueError as exc:
                raise ValueError(
                    "KRINEIA_ED25519_PRIVATE_KEY must be 64 lowercase hex characters"
                ) from exc
        elif isinstance(key_material, bytes):
            raw = key_material
        else:
            raise TypeError("Ed25519 private key material must be raw bytes or hex text")
        if len(raw) != 32:
            raise ValueError("Ed25519 private key material must contain exactly 32 bytes")
        return Ed25519PrivateKey.from_private_bytes(raw)

    @property
    def public_key_hex(self) -> str | None:
        """Return the raw public verification key for the configured signer."""
        if self._private_key is None:
            return None
        from cryptography.hazmat.primitives import serialization

        return self._private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        ).hex()

    def _keys_for_verification(self) -> dict[str, Any]:
        keys = dict(self._verification_keys)
        if self.public_key_hex is not None:
            keys.setdefault(
                self.signing_key_id,
                {
                    "algorithm": "Ed25519",
                    "public_key": self.public_key_hex,
                    "signer_identity": WATCHER_IDENTITY,
                },
            )
        return keys

    def _assert_existing_chain_valid(self) -> None:
        """Fail closed before extension if any prior receipt is untrusted."""
        if not self.receipt_path.exists() or self.receipt_path.stat().st_size == 0:
            self._last_hash = ZERO_DIGEST
            self._chain_length = 0
            if (
                self._checkpoint_head != ZERO_DIGEST
                or self._checkpoint_count != 0
            ):
                raise RuntimeError(
                    "refusing an empty or missing chain without explicit new-chain "
                    "initialization or a trusted zero-head/zero-count checkpoint"
                )
            return
        if self._checkpoint_head is None or self._checkpoint_count is None:
            raise RuntimeError(
                "refusing to extend an existing chain without a trusted expected head and count"
            )
        count, errors, warnings = _tool_verify_chain(
            self.receipt_path,
            self._keys_for_verification(),
            expected_head=self._checkpoint_head,
            expected_count=self._checkpoint_count,
        )
        if errors or warnings:
            findings = errors + warnings
            raise RuntimeError(
                f"refusing to extend unverified chain ({count} receipt(s)): "
                + "; ".join(findings)
            )
        self._chain_length = count
        self._last_hash = self._checkpoint_head

    def _source_record_already_receipted(
        self,
        source_record_id: str,
        source_byte_offset: int,
        source_record_sha256: str,
    ) -> bool:
        """Return whether a verified chain already contains this source record.

        A byte offset may be admitted only once. Seeing the same offset with a
        different digest after a crash indicates that the source changed
        before its durable cursor acknowledgement; appending a second receipt
        would turn that source race into ambiguous provenance.
        """
        if not self.receipt_path.exists():
            return False
        with open(self.receipt_path, "r", encoding="utf-8") as handle:
            for line in handle:
                if not line.strip():
                    continue
                receipt = json.loads(line)
                state = receipt.get("state")
                payload = state.get("payload") if isinstance(state, dict) else None
                if not isinstance(payload, dict):
                    continue
                if payload.get("source_byte_offset") != source_byte_offset:
                    continue
                if (
                    payload.get("source_record_id") != source_record_id
                    or payload.get("source_record_sha256") != source_record_sha256
                ):
                    raise RuntimeError(
                        "source byte offset is already bound to a different record digest"
                    )
                return True
        return False

    def generate(
        self,
        actor_identity: str,
        skill_name: str,
        execution_mode: str,
        args_hash: str,
        session_id: str,
        bus_timestamp: str,
        source_byte_offset: int,
        source_record_sha256: str,
    ) -> dict[str, Any]:
        """Create a Krineia receipt for a skill invocation.

        Args:
            actor_identity: Bare canonical identity of the invoking agent.
            skill_name: Name of the invoked skill.
            execution_mode: The skill's execution-mode value.
            args_hash: SHA-256 hash of the skill arguments.
            session_id: Opaque session identifier from the SKILL_INVOKE event.
            bus_timestamp: Timestamp from the bus message.
            source_byte_offset: Starting byte offset of the exact TSV record.
            source_record_sha256: SHA-256 of the exact raw TSV bytes, including
                their terminating newline.
        """
        if self._private_key is None:
            raise RuntimeError(
                "schema 2.1 receipt emission requires KRINEIA_ED25519_PRIVATE_KEY"
            )
        if (
            not 1 <= _utf8_len(actor_identity) <= MAX_BUS_IDENTITY_BYTES
            or _BUS_IDENTITY_RE.fullmatch(actor_identity) is None
            or not 1 <= _utf8_len(skill_name) <= MAX_SKILL_NAME_BYTES
            or _SKILL_NAME_RE.fullmatch(skill_name) is None
            or not 1 <= _utf8_len(execution_mode) <= MAX_MODE_BYTES
            or _MODE_RE.fullmatch(execution_mode) is None
            or _LOWER_SHA256_RE.fullmatch(args_hash) is None
            or not 1 <= _utf8_len(session_id) <= MAX_SESSION_ID_BYTES
            or _SESSION_RE.fullmatch(session_id) is None
            or not 1 <= _utf8_len(bus_timestamp) <= MAX_BUS_TIMESTAMP_BYTES
            or not _valid_bus_timestamp(bus_timestamp)
        ):
            raise ValueError("invalid or oversized skill-invocation source field")
        if isinstance(source_byte_offset, bool) or not isinstance(source_byte_offset, int):
            raise ValueError("source_byte_offset must be a non-negative integer")
        if source_byte_offset < 0:
            raise ValueError("source_byte_offset must be a non-negative integer")
        if _LOWER_SHA256_RE.fullmatch(source_record_sha256) is None:
            raise ValueError("source_record_sha256 must be 64 lowercase hex characters")
        source_id = _source_record_id(source_byte_offset, source_record_sha256)
        self._assert_existing_chain_valid()
        if self._last_hash is None or self._chain_length is None:
            raise RuntimeError("verified chain state was not initialized")
        if self._source_record_already_receipted(
            source_id, source_byte_offset, source_record_sha256
        ):
            raise SourceRecordAlreadyReceipted(source_id)
        receipt_id = str(uuid.uuid4())
        timestamp = _utc_now()

        payload = {
            "skill": skill_name,
            "mode": execution_mode,
            "args_hash": args_hash,
            "session_id": session_id,
            "bus_timestamp": bus_timestamp,
            "source_byte_offset": source_byte_offset,
            "source_record_sha256": source_record_sha256,
            "source_record_id": source_id,
        }

        receipt: dict[str, Any] = {
            "schema_version": RECEIPT_VERSION,
            "id": receipt_id,
            "session_id": session_id,
            "actor_identity": actor_identity,
            "time": timestamp,
            "state": {
                "event": EVENT_PREFIX,
                "payload": payload,
            },
            "drift": 0.0,
            "prev_hash": self._last_hash,
            "signature_envelope": {
                "key_id": self.signing_key_id,
                "signer_identity": WATCHER_IDENTITY,
                "algorithm": "Ed25519",
            },
            "hash": "",
        }

        # v2.1 signs the canonical body, including signature metadata but
        # excluding the signature value and top-level hash. The final hash then
        # commits to that body plus the signature.
        signed_body = canonical_signed_body(receipt)
        receipt["signature_envelope"]["signature"] = self._private_key.sign(
            signed_body.encode("utf-8")
        ).hex()
        receipt["hash"] = sha256_hex(canonical_receipt(receipt))

        return receipt

    def append(self, receipt: dict[str, Any]) -> None:
        """Append receipt to the chain log."""
        if receipt.get("schema_version") != RECEIPT_VERSION:
            raise ValueError("the reference writer emits only schema 2.1.0 receipts")
        lock_path = self.receipt_path.with_suffix(self.receipt_path.suffix + ".lock")
        with _exclusive_append_lock(lock_path):
            # Re-read and verify while holding the lock. This prevents two
            # writers that generated from the same tail from silently creating
            # a fork in one JSONL file; the stale writer is rejected.
            self._assert_existing_chain_valid()
            if self._last_hash is None or self._chain_length is None:
                raise RuntimeError("verified chain state was not initialized")
            previous_hash = None if self._chain_length == 0 else self._last_hash
            errors = validate_receipt(
                receipt, self._chain_length + 1, previous_hash
            )
            signature_errors, _ = verify_signature(
                receipt, self._keys_for_verification(), self._chain_length + 1
            )
            errors.extend(signature_errors)
            if errors:
                raise ValueError(
                    "refusing to append invalid receipt: " + "; ".join(errors)
                )
            line = json.dumps(
                receipt,
                sort_keys=True,
                separators=(",", ":"),
                ensure_ascii=False,
                allow_nan=False,
            )
            encoded_line = (line + "\n").encode("utf-8")
            if len(encoded_line) > MAX_RECEIPT_LINE_BYTES:
                raise RuntimeError(
                    "refusing to append a receipt that exceeds the verifier line limit"
                )
            if self._chain_length >= MAX_RECEIPTS:
                raise RuntimeError(
                    "refusing to append beyond the verifier receipt-count limit"
                )
            current_size = (
                self.receipt_path.stat().st_size
                if self.receipt_path.exists()
                else 0
            )
            if current_size + len(encoded_line) > MAX_RECEIPT_CHAIN_BYTES:
                raise RuntimeError(
                    "refusing to append beyond the verifier chain-size limit"
                )
            created = not self.receipt_path.exists()
            with open(self.receipt_path, "ab") as handle:
                handle.write(encoded_line)
                handle.flush()
                os.fsync(handle.fileno())
            if created:
                _fsync_directory(self.receipt_path.parent)

            self._last_hash = receipt["hash"]
            self._chain_length += 1
            self._checkpoint_head = receipt["hash"]
            self._checkpoint_count = self._chain_length

    @property
    def chain_length(self) -> int:
        if self._chain_length is None:
            raise RuntimeError("chain length is unavailable until anchored verification")
        return self._chain_length

    @property
    def checkpoint_head(self) -> str | None:
        return self._checkpoint_head

    @property
    def checkpoint_count(self) -> int | None:
        return self._checkpoint_count


# ---------------------------------------------------------------------------
# Bus polling
# ---------------------------------------------------------------------------


class BusPoller:
    """Peek append-only bus records and durably acknowledge dispositions.

    The committed cursor is a byte offset, not a read-ahead position. ``poll``
    never advances it. Callers must either ``acknowledge`` a durable receipt or
    ``reject`` malformed input into the durable rejection directory.
    """

    def __init__(
        self,
        bus_path: str | Path,
        checkpoint_path: str | Path | None = None,
        *,
        start_at_end: bool = False,
        max_records_per_poll: int = DEFAULT_MAX_RECORDS_PER_POLL,
        max_line_bytes: int = MAX_BUS_LINE_BYTES,
        max_bytes_per_poll: int = MAX_BUS_BYTES_PER_POLL,
        max_oversized_record_bytes: int = MAX_OVERSIZED_RECORD_BYTES,
        max_rejection_markers: int = MAX_REJECTION_MARKERS,
    ) -> None:
        self.bus_path = Path(bus_path)
        self.checkpoint_path = (
            Path(checkpoint_path)
            if checkpoint_path is not None
            else self.bus_path.with_suffix(self.bus_path.suffix + ".cursor.json")
        )
        self.rejection_dir = self.checkpoint_path.with_suffix(
            self.checkpoint_path.suffix + ".rejections"
        )
        self.rejection_state_path = self.checkpoint_path.with_suffix(
            self.checkpoint_path.suffix + ".rejections.state.json"
        )
        if max_records_per_poll <= 0:
            raise ValueError("max_records_per_poll must be positive")
        if max_line_bytes <= 0:
            raise ValueError("max_line_bytes must be positive")
        if max_bytes_per_poll < max_line_bytes + 1:
            raise ValueError("max_bytes_per_poll must exceed max_line_bytes")
        if max_oversized_record_bytes <= max_line_bytes:
            raise ValueError("max_oversized_record_bytes must exceed max_line_bytes")
        if max_rejection_markers <= 0:
            raise ValueError("max_rejection_markers must be positive")
        self.max_records_per_poll = max_records_per_poll
        self.max_line_bytes = max_line_bytes
        self.max_bytes_per_poll = max_bytes_per_poll
        self.max_oversized_record_bytes = max_oversized_record_bytes
        self.max_rejection_markers = max_rejection_markers
        self._rejection_count = 0
        self._rejection_accumulator = ZERO_DIGEST
        self._last_rejection_marker: dict[str, Any] | None = None
        self._load_rejection_state()
        self._source_file_id: str | None = None
        self._offset = 0
        self._last_record_start = 0
        self._prior_record_sha256 = ZERO_DIGEST
        self._oversize_start: int | None = None
        self._oversize_scan_offset = 0
        self._oversize_hasher: Any | None = None
        self._oversize_anchor_start = 0
        self._oversize_anchor_sha256: str | None = None
        self._load_or_initialize_checkpoint(start_at_end=start_at_end)

    def _load_rejection_state(self) -> None:
        if not self.rejection_state_path.exists():
            if self.rejection_dir.exists() and any(self.rejection_dir.iterdir()):
                raise SourceIntegrityError(
                    "rejection markers exist without their accumulator state"
                )
            return
        try:
            value = json.loads(
                self.rejection_state_path.read_text(encoding="utf-8")
            )
        except (OSError, json.JSONDecodeError) as exc:
            raise SourceIntegrityError(f"invalid rejection state: {exc}") from exc
        expected = {
            "version",
            "ring_size",
            "rejection_count",
            "rejection_accumulator",
            "last_marker",
        }
        if not isinstance(value, dict) or set(value) != expected:
            raise SourceIntegrityError("rejection state has an invalid field set")
        count = value.get("rejection_count")
        accumulator = value.get("rejection_accumulator")
        last_marker = value.get("last_marker")
        if (
            value.get("version") != 1
            or value.get("ring_size") != self.max_rejection_markers
            or isinstance(count, bool)
            or not isinstance(count, int)
            or count < 0
            or not isinstance(accumulator, str)
            or _LOWER_SHA256_RE.fullmatch(accumulator) is None
            or (last_marker is not None and not isinstance(last_marker, dict))
        ):
            raise SourceIntegrityError("rejection state contains invalid values")
        if count == 0:
            if accumulator != ZERO_DIGEST or last_marker is not None:
                raise SourceIntegrityError("empty rejection state has a nonzero anchor")
        else:
            required_marker = {
                "sequence",
                "source_record_id",
                "source_record_sha256",
                "source_byte_offset",
                "source_end_offset",
                "reason",
                "previous_accumulator",
                "accumulator",
            }
            if (
                not isinstance(last_marker, dict)
                or set(last_marker) != required_marker
                or last_marker.get("sequence") != count
                or last_marker.get("accumulator") != accumulator
                or _LOWER_SHA256_RE.fullmatch(
                    str(last_marker.get("source_record_sha256", ""))
                )
                is None
            ):
                raise SourceIntegrityError("rejection state last marker is invalid")
        if self.rejection_dir.exists():
            entries = 0
            with os.scandir(self.rejection_dir) as directory_entries:
                for _entry in directory_entries:
                    entries += 1
                    if entries > self.max_rejection_markers:
                        raise SourceIntegrityError(
                            "rejection ring exceeds its configured bound"
                        )
        self._rejection_count = count
        self._rejection_accumulator = accumulator
        self._last_rejection_marker = last_marker

    def _rejection_state_value(
        self, marker: dict[str, Any]
    ) -> dict[str, Any]:
        return {
            "version": 1,
            "ring_size": self.max_rejection_markers,
            "rejection_count": marker["sequence"],
            "rejection_accumulator": marker["accumulator"],
            "last_marker": marker,
        }

    def _write_rejection_ring_marker(self, marker: dict[str, Any]) -> None:
        slot = (int(marker["sequence"]) - 1) % self.max_rejection_markers
        marker_path = self.rejection_dir / f"{slot:08d}.json"
        _atomic_write_json(marker_path, marker)

    @property
    def committed_offset(self) -> int:
        return self._offset

    def _checkpoint_value(self) -> dict[str, Any]:
        return {
            "version": CURSOR_VERSION,
            "source_file_id": self._source_file_id,
            "offset": self._offset,
            "last_record_start": self._last_record_start,
            "prior_record_sha256": self._prior_record_sha256,
        }

    def _persist_checkpoint(self) -> None:
        _atomic_write_json(self.checkpoint_path, self._checkpoint_value())

    def _load_or_initialize_checkpoint(self, *, start_at_end: bool) -> None:
        if self.checkpoint_path.exists():
            if start_at_end:
                raise ValueError("start_at_end is valid only before a checkpoint exists")
            try:
                value = json.loads(self.checkpoint_path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError) as exc:
                raise SourceIntegrityError(f"invalid bus checkpoint: {exc}") from exc
            expected = {
                "version",
                "source_file_id",
                "offset",
                "last_record_start",
                "prior_record_sha256",
            }
            if not isinstance(value, dict) or set(value) != expected:
                raise SourceIntegrityError("bus checkpoint has an invalid field set")
            if value.get("version") != CURSOR_VERSION:
                raise SourceIntegrityError("unsupported bus checkpoint version")
            offset = value.get("offset")
            last_start = value.get("last_record_start")
            digest = value.get("prior_record_sha256")
            source_file_id = value.get("source_file_id")
            if (
                isinstance(offset, bool)
                or not isinstance(offset, int)
                or offset < 0
                or isinstance(last_start, bool)
                or not isinstance(last_start, int)
                or not 0 <= last_start <= offset
                or not isinstance(digest, str)
                or _LOWER_SHA256_RE.fullmatch(digest) is None
                or (source_file_id is not None and not isinstance(source_file_id, str))
            ):
                raise SourceIntegrityError("bus checkpoint contains invalid values")
            if offset == 0 and (last_start != 0 or digest != ZERO_DIGEST):
                raise SourceIntegrityError("zero-offset checkpoint has a nonzero anchor")
            if offset > 0 and last_start == offset:
                raise SourceIntegrityError("checkpoint prior-record range is empty")
            self._source_file_id = source_file_id
            self._offset = offset
            self._last_record_start = last_start
            self._prior_record_sha256 = digest
            self._verify_checkpoint_against_source()
            return

        if start_at_end and not self.bus_path.exists():
            raise FileNotFoundError("cannot start at end before the bus exists")
        if self.bus_path.exists():
            self._source_file_id = _source_file_identity(self.bus_path)
            if start_at_end:
                self._initialize_at_end()
        self._persist_checkpoint()

    def _initialize_at_end(self) -> None:
        size = self.bus_path.stat().st_size
        if size == 0:
            self._offset = 0
            self._last_record_start = 0
            self._prior_record_sha256 = ZERO_DIGEST
            return
        with open(self.bus_path, "rb") as handle:
            handle.seek(size - 1)
            if handle.read(1) != b"\n":
                raise SourceIntegrityError(
                    "cannot start at end across an incomplete bus record"
                )
            window_size = min(size, self.max_line_bytes + 1)
            window_start = size - window_size
            handle.seek(window_start)
            window = handle.read(window_size)
        prior_newline = window[:-1].rfind(b"\n")
        if prior_newline >= 0:
            last_start = window_start + prior_newline + 1
            raw = window[prior_newline + 1 :]
        else:
            last_start = 0
            raw = window
        if len(raw) > self.max_line_bytes or (
            last_start == 0 and size > self.max_line_bytes
        ):
            raise SourceIntegrityError(
                "cannot start at end across an oversized final bus record"
            )
        self._offset = size
        self._last_record_start = last_start
        self._prior_record_sha256 = hashlib.sha256(raw).hexdigest()

    def _hash_source_range(self, start: int, end: int) -> tuple[str, bytes]:
        digest = hashlib.sha256()
        last_byte = b""
        remaining = end - start
        with open(self.bus_path, "rb") as handle:
            handle.seek(start)
            while remaining:
                chunk = handle.read(min(64 * 1024, remaining))
                if not chunk:
                    raise SourceIntegrityError("bus was truncated inside its checkpoint")
                digest.update(chunk)
                last_byte = chunk[-1:]
                remaining -= len(chunk)
        return digest.hexdigest(), last_byte

    def _verify_checkpoint_against_source(self) -> None:
        if not self.bus_path.exists():
            if self._source_file_id is None and self._offset == 0:
                return
            raise SourceIntegrityError("checkpointed bus source disappeared")
        actual_identity = _source_file_identity(self.bus_path)
        if self._source_file_id is None:
            if self._offset != 0:
                raise SourceIntegrityError("unbound source has a nonzero checkpoint")
            self._source_file_id = actual_identity
            self._persist_checkpoint()
            return
        if actual_identity != self._source_file_id:
            raise SourceIntegrityError("bus source was rotated or replaced")
        size = self.bus_path.stat().st_size
        if size < self._offset:
            raise SourceIntegrityError("bus source was truncated before its checkpoint")
        prior_size = self._offset - self._last_record_start
        if self._offset and prior_size <= self.max_oversized_record_bytes:
            digest, last_byte = self._hash_source_range(
                self._last_record_start, self._offset
            )
            if digest != self._prior_record_sha256 or last_byte != b"\n":
                raise SourceIntegrityError("bus checkpoint prior-record digest mismatch")
        elif self._offset:
            raise SourceIntegrityError(
                "checkpoint prior record exceeds the bounded recovery scan limit"
            )

    def _source_is_available(self) -> bool:
        if not self.bus_path.exists():
            if self._source_file_id is None and self._offset == 0:
                return False
            raise SourceIntegrityError("checkpointed bus source disappeared")
        actual_identity = _source_file_identity(self.bus_path)
        if self._source_file_id is None:
            self._source_file_id = actual_identity
            self._persist_checkpoint()
        elif actual_identity != self._source_file_id:
            raise SourceIntegrityError("bus source was rotated or replaced")
        size = self.bus_path.stat().st_size
        if size < self._offset:
            raise SourceIntegrityError("bus source was truncated before its checkpoint")
        if self._oversize_start is not None:
            if (
                self._oversize_start != self._offset
                or self._oversize_scan_offset <= self._oversize_start
                or self._oversize_anchor_sha256 is None
                or not self._oversize_start
                <= self._oversize_anchor_start
                < self._oversize_scan_offset
            ):
                raise SourceIntegrityError(
                    "oversized-record scan continuity state is invalid"
                )
            if size < self._oversize_scan_offset:
                raise SourceIntegrityError(
                    "bus source was truncated during an oversized-record scan"
                )
            anchor_digest, _ = self._hash_source_range(
                self._oversize_anchor_start, self._oversize_scan_offset
            )
            if anchor_digest != self._oversize_anchor_sha256:
                raise SourceIntegrityError(
                    "bus source changed during an oversized-record scan"
                )
        prior_size = self._offset - self._last_record_start
        if self._offset and prior_size <= self.max_oversized_record_bytes:
            digest, last_byte = self._hash_source_range(
                self._last_record_start, self._offset
            )
            if digest != self._prior_record_sha256 or last_byte != b"\n":
                raise SourceIntegrityError(
                    "bus source was rotated, replaced, or modified before its checkpoint"
                )
        elif self._offset:
            raise SourceIntegrityError(
                "checkpoint prior record exceeds the bounded recovery scan limit"
            )
        return True

    def _record_from_raw(self, start: int, end: int, raw: bytes) -> BusRecord:
        digest = hashlib.sha256(raw).hexdigest()
        source_id = _source_record_id(start, digest)
        if len(raw) > self.max_line_bytes:
            return BusRecord(
                start, end, digest, source_id, None,
                f"bus record exceeds {self.max_line_bytes} bytes",
            )
        try:
            text = raw[:-1].decode("utf-8", errors="strict")
        except UnicodeDecodeError:
            return BusRecord(start, end, digest, source_id, None, "invalid UTF-8")
        if text.endswith("\r"):
            text = text[:-1]
        parts = text.split("\t", 4)
        if len(parts) != 5:
            return BusRecord(start, end, digest, source_id, None, "invalid TSV field count")
        timestamp, actor, target, event_type, message = parts
        if (
            not 1 <= _utf8_len(timestamp) <= MAX_BUS_TIMESTAMP_BYTES
            or not _valid_bus_timestamp(timestamp)
            or not 1 <= _utf8_len(actor) <= MAX_BUS_IDENTITY_BYTES
            or _BUS_IDENTITY_RE.fullmatch(actor) is None
            or not 1 <= _utf8_len(target) <= MAX_BUS_IDENTITY_BYTES
            or _BUS_IDENTITY_RE.fullmatch(target) is None
            or not 1 <= _utf8_len(event_type) <= MAX_BUS_TYPE_BYTES
            or _BUS_TYPE_RE.fullmatch(event_type) is None
            or _utf8_len(message) > MAX_BUS_LINE_BYTES
        ):
            return BusRecord(start, end, digest, source_id, None, "invalid or oversized TSV field")
        return BusRecord(
            start,
            end,
            digest,
            source_id,
            {
                "timestamp": timestamp,
                "from": actor,
                "to": target,
                "type": event_type,
                "message": message,
            },
        )

    def _continue_oversized_record(
        self, handle: Any, start: int, initial: bytes | None = None
    ) -> BusRecord | None:
        if self._oversize_start is None:
            if initial is None:
                raise AssertionError("oversized scan requires initial bytes")
            self._oversize_start = start
            self._oversize_scan_offset = handle.tell()
            self._oversize_hasher = hashlib.sha256()
            self._oversize_hasher.update(initial)
            consumed = len(initial)
            if initial.endswith(b"\n"):
                digest = self._oversize_hasher.hexdigest()
                end = self._oversize_scan_offset
                self._reset_oversized_scan()
                return BusRecord(
                    start,
                    end,
                    digest,
                    _source_record_id(start, digest),
                    None,
                    f"bus record exceeds {self.max_line_bytes} bytes",
                )
        else:
            if self._oversize_start != start or self._oversize_hasher is None:
                raise SourceIntegrityError("oversized-record scan lost cursor alignment")
            handle.seek(self._oversize_scan_offset)
            consumed = 0

        budget = self.max_bytes_per_poll - consumed
        while budget > 0:
            record_bytes = self._oversize_scan_offset - start
            remaining_limit = self.max_oversized_record_bytes - record_bytes
            if remaining_limit <= 0:
                raise SourceIntegrityError(
                    "oversized bus record exceeds the bounded recovery scan limit"
                )
            chunk = handle.readline(min(64 * 1024, budget, remaining_limit))
            if not chunk:
                self._oversize_scan_offset = handle.tell()
                self._refresh_oversized_anchor()
                return None
            self._oversize_hasher.update(chunk)
            consumed += len(chunk)
            budget -= len(chunk)
            self._oversize_scan_offset = handle.tell()
            if chunk.endswith(b"\n"):
                digest = self._oversize_hasher.hexdigest()
                end = self._oversize_scan_offset
                self._reset_oversized_scan()
                return BusRecord(
                    start,
                    end,
                    digest,
                    _source_record_id(start, digest),
                    None,
                    f"bus record exceeds {self.max_line_bytes} bytes",
                )
            if self._oversize_scan_offset - start >= self.max_oversized_record_bytes:
                raise SourceIntegrityError(
                    "oversized bus record exceeds the bounded recovery scan limit"
                )
        self._refresh_oversized_anchor()
        return None

    def _refresh_oversized_anchor(self) -> None:
        """Bind the bounded suffix of an in-progress oversized source scan."""
        if self._oversize_start is None or self._oversize_scan_offset <= self._oversize_start:
            raise SourceIntegrityError("oversized-record scan lost anchor alignment")
        self._oversize_anchor_start = max(
            self._oversize_start,
            self._oversize_scan_offset - OVERSIZE_CONTINUITY_ANCHOR_BYTES,
        )
        self._oversize_anchor_sha256, _ = self._hash_source_range(
            self._oversize_anchor_start, self._oversize_scan_offset
        )

    def _reset_oversized_scan(self) -> None:
        self._oversize_start = None
        self._oversize_scan_offset = 0
        self._oversize_hasher = None
        self._oversize_anchor_start = 0
        self._oversize_anchor_sha256 = None

    def poll(self) -> list[BusRecord]:
        """Peek a bounded batch without advancing the durable checkpoint."""
        if not self._source_is_available():
            return []
        records: list[BusRecord] = []
        consumed = 0
        with open(self.bus_path, "rb") as handle:
            handle.seek(self._offset)
            if self._oversize_start is not None:
                record = self._continue_oversized_record(handle, self._offset)
                return [record] if record is not None else []
            while len(records) < self.max_records_per_poll:
                if records and self.max_bytes_per_poll - consumed < self.max_line_bytes + 1:
                    break
                start = handle.tell()
                raw = handle.readline(self.max_line_bytes + 1)
                if not raw:
                    break
                consumed += len(raw)
                if len(raw) > self.max_line_bytes:
                    if records:
                        break
                    record = self._continue_oversized_record(handle, start, raw)
                    return [record] if record is not None else []
                if not raw.endswith(b"\n"):
                    # A writer may still be appending this record. It remains
                    # unacknowledged and is retried from the same byte offset.
                    break
                records.append(self._record_from_raw(start, handle.tell(), raw))
                if consumed >= self.max_bytes_per_poll:
                    break
        return records

    def acknowledge(self, record: BusRecord) -> None:
        """Durably advance past a record whose disposition is already durable."""
        self._validate_record_against_source(record)
        previous_cursor = (
            self._offset,
            self._last_record_start,
            self._prior_record_sha256,
        )
        self._offset = record.end_offset
        self._last_record_start = record.start_offset
        self._prior_record_sha256 = record.record_sha256
        try:
            self._persist_checkpoint()
        except Exception:
            (
                self._offset,
                self._last_record_start,
                self._prior_record_sha256,
            ) = previous_cursor
            raise
        if self._oversize_start == record.start_offset:
            self._reset_oversized_scan()

    def _validate_record_against_source(self, record: BusRecord) -> None:
        """Revalidate exact source bytes immediately before durable disposition."""
        if record.start_offset != self._offset:
            raise SourceIntegrityError("record acknowledgement is out of order")
        if record.end_offset <= record.start_offset:
            raise SourceIntegrityError("record acknowledgement range is invalid")
        if not self._source_is_available():
            raise SourceIntegrityError("cannot acknowledge an unavailable source")
        if self.bus_path.stat().st_size < record.end_offset:
            raise SourceIntegrityError("bus was truncated before acknowledgement")
        # Oversized discovery is incremental and bounded per poll. One linear
        # exact-byte revalidation at disposition prevents a mixed old/new
        # incremental digest from advancing the cursor after a source rewrite.
        digest, last_byte = self._hash_source_range(
            record.start_offset, record.end_offset
        )
        if digest != record.record_sha256 or last_byte != b"\n":
            raise SourceIntegrityError(
                "source record changed before durable acknowledgement"
            )

    def reject(self, record: BusRecord, reason: str) -> None:
        """Commit a rejection into a bounded hash-accumulating ring, then ack."""
        # Do not persist a rejection for stale or mixed source bytes. The
        # acknowledge call below revalidates again after the disposition is
        # durable, closing the ordinary write-before-cursor crash window.
        self._validate_record_against_source(record)
        if self._last_rejection_marker is not None and (
            self._last_rejection_marker.get("source_record_id")
            == record.source_record_id
        ):
            if (
                self._last_rejection_marker.get("source_record_sha256")
                != record.record_sha256
            ):
                raise SourceIntegrityError("replayed rejection digest mismatch")
            self._write_rejection_ring_marker(self._last_rejection_marker)
            self.acknowledge(record)
            return

        body = {
            "sequence": self._rejection_count + 1,
            "source_record_id": record.source_record_id,
            "source_record_sha256": record.record_sha256,
            "source_byte_offset": record.start_offset,
            "source_end_offset": record.end_offset,
            "reason": reason[:512],
            "previous_accumulator": self._rejection_accumulator,
        }
        canonical_body = json.dumps(
            body, sort_keys=True, separators=(",", ":"), ensure_ascii=False
        ).encode("utf-8")
        marker = dict(body)
        marker["accumulator"] = hashlib.sha256(
            bytes.fromhex(self._rejection_accumulator) + canonical_body
        ).hexdigest()

        # State first makes a crash replay recoverable: the bounded ring slot
        # can be recreated from last_marker before the source cursor advances.
        _atomic_write_json(
            self.rejection_state_path, self._rejection_state_value(marker)
        )
        self._rejection_count = int(marker["sequence"])
        self._rejection_accumulator = str(marker["accumulator"])
        self._last_rejection_marker = marker
        self._write_rejection_ring_marker(marker)
        self.acknowledge(record)


# ---------------------------------------------------------------------------
# Main watcher loop
# ---------------------------------------------------------------------------


class KrineiaWatcher:
    """Main watcher daemon."""

    def __init__(
        self,
        bus_path: str | Path,
        receipt_path: str | Path,
        poll_interval: float,
        signing_private_key: str | bytes | None = None,
        verification_keys: dict[str, Any] | None = None,
        checkpoint_path: str | Path | None = None,
        start_at_end: bool = False,
        max_records_per_poll: int = DEFAULT_MAX_RECORDS_PER_POLL,
        max_oversized_record_bytes: int = MAX_OVERSIZED_RECORD_BYTES,
        expected_existing_head: str | None = None,
        expected_existing_count: int | None = None,
        initialize_new_chain: bool = False,
        max_rejection_markers: int = MAX_REJECTION_MARKERS,
    ) -> None:
        resolved_receipt_path = Path(receipt_path)
        resolved_checkpoint = (
            Path(checkpoint_path)
            if checkpoint_path is not None
            else resolved_receipt_path.with_suffix(
                resolved_receipt_path.suffix + ".source.cursor.json"
            )
        )
        self.poller = BusPoller(
            bus_path,
            resolved_checkpoint,
            start_at_end=start_at_end,
            max_records_per_poll=max_records_per_poll,
            max_oversized_record_bytes=max_oversized_record_bytes,
            max_rejection_markers=max_rejection_markers,
        )
        self.writer = SkillReceiptWriter(
            resolved_receipt_path,
            signing_private_key,
            verification_keys=verification_keys,
            expected_existing_head=expected_existing_head,
            expected_existing_count=expected_existing_count,
            initialize_new_chain=initialize_new_chain,
        )
        self.poll_interval = poll_interval
        self._running = False
        self._receipts_written = 0

    def _handle_signal(self, signum: int, _frame: Any) -> None:
        """Graceful shutdown on SIGTERM/SIGINT."""
        logger.info(
            "Krineia watcher shutting down (signal %d); receipts_written=%d",
            signum,
            self._receipts_written,
        )
        self._running = False

    def run(self) -> None:
        """Main loop: poll bus, create receipts, append to chain."""
        # Establish the trusted starting state before the daemon announces a
        # chain length or consumes source records. Existing chains require the
        # caller-supplied out-of-band checkpoint; a first genesis requires the
        # explicit initialization policy.
        self.writer._assert_existing_chain_valid()
        self._running = True
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)

        logger.info(
            "Krineia watcher started; chain_length=%d; poll_interval=%.1fs",
            self.writer.chain_length,
            self.poll_interval,
        )

        while self._running:
            try:
                self._tick()
            except Exception:
                logger.exception("Tick failed; continuing")
            time.sleep(self.poll_interval)

    def _tick(self) -> None:
        """Single poll-and-process cycle."""
        records = self.poller.poll()
        for record in records:
            if record.rejection_reason is not None:
                logger.warning(
                    "Rejected bus record %s: %s",
                    record.source_record_id,
                    record.rejection_reason,
                )
                self.poller.reject(record, record.rejection_reason)
                continue

            entry = record.entry
            if entry is None:
                self.poller.reject(record, "bus record has no parsed entry")
                continue
            if entry["type"] != "SKILL_INVOKE":
                self.poller.acknowledge(record)
                continue

            parsed = _parse_skill_invoke(entry["message"])
            if parsed is None:
                logger.warning(
                    "Rejected malformed SKILL_INVOKE source record %s",
                    record.source_record_id,
                )
                self.poller.reject(record, "invalid SKILL_INVOKE grammar or field")
                continue

            try:
                receipt = self.writer.generate(
                    actor_identity=entry["from"],
                    skill_name=parsed["skill"],
                    execution_mode=parsed["mode"],
                    args_hash=parsed["args_hash"],
                    session_id=parsed["session"],
                    bus_timestamp=entry["timestamp"],
                    source_byte_offset=record.start_offset,
                    source_record_sha256=record.record_sha256,
                )
            except SourceRecordAlreadyReceipted:
                # The receipt reached disk but a prior process died before its
                # cursor checkpoint. Replaying the source is therefore an ack,
                # not a second receipt.
                self.poller.acknowledge(record)
                logger.info(
                    "Acknowledged already-receipted source record %s",
                    record.source_record_id,
                )
                continue
            self.writer.append(receipt)
            # This is deliberately last: receipt append flushes and fsyncs
            # before the source checkpoint can advance.
            self.poller.acknowledge(record)
            self._receipts_written += 1
            logger.debug(
                "Receipt created: id=%s skill=%s actor=%s",
                receipt["id"],
                parsed["skill"],
                entry["from"],
            )

    def verify_chain(
        self,
        *,
        expected_head: str | None = None,
        expected_count: int | None = None,
        allow_unanchored: bool = False,
    ) -> bool:
        """Verify receipt chain integrity. Returns True if valid.

        Delegates to the canonical implementation in
        ``tools/verify_chain.py`` so the daemon and the CLI tool share one
        verifier. The tool's ``verify_chain`` returns
        ``(count, errors, warnings)``; this wrapper reports errors/warnings
        via the logger and returns True only when there are no errors.
        """
        if not self.writer.receipt_path.exists():
            logger.error("Chain does not exist: %s", self.writer.receipt_path)
            return False

        # Derive only the public half for verification. Private key material is
        # never passed into the verifier or written to the chain.
        effective_head = (
            expected_head if expected_head is not None else self.writer.checkpoint_head
        )
        effective_count = (
            expected_count if expected_count is not None else self.writer.checkpoint_count
        )
        count, errors, warnings = _tool_verify_chain(
            self.writer.receipt_path,
            self.writer._keys_for_verification(),
            expected_head=effective_head,
            expected_count=effective_count,
        )
        unanchored = effective_head is None or effective_count is None
        if unanchored:
            anchor_message = (
                "chain completeness was not checked against a trusted head and count"
            )
            if allow_unanchored:
                logger.warning(
                    "verify_chain: %s (explicit unanchored inspection policy)",
                    anchor_message,
                )
            else:
                errors.append(
                    anchor_message
                    + "; allow_unanchored is permitted only for non-assurance inspection"
                )
        for warning in warnings:
            logger.warning("verify_chain: %s", warning)
        if errors or warnings:
            for error in errors:
                logger.error("verify_chain: %s", error)
            logger.error(
                "Chain invalid: %d receipt(s), %d error(s), %d warning(s)",
                count,
                len(errors),
                len(warnings),
            )
            return False
        if unanchored:
            logger.info(
                "Chain internally valid under explicit unanchored inspection policy: %d receipts",
                count,
            )
        else:
            logger.info("Chain anchored and valid: %d receipts", count)
        return True

    def tail(self, n: int) -> list[dict[str, Any]]:
        """Return last N receipts."""
        return _bounded_tail(self.writer.receipt_path, n)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Krineia Watcher Service Agent")
    parser.add_argument(
        "--bus-path",
        default=DEFAULT_BUS_PATH,
        help=f"Path to coordination bus TSV (default: {DEFAULT_BUS_PATH})",
    )
    parser.add_argument(
        "--receipt-path",
        default=DEFAULT_RECEIPT_PATH,
        help=f"Path to receipt JSONL (default: {DEFAULT_RECEIPT_PATH})",
    )
    parser.add_argument(
        "--poll-interval",
        type=float,
        default=DEFAULT_POLL_INTERVAL,
        help=f"Poll interval in seconds (default: {DEFAULT_POLL_INTERVAL})",
    )
    parser.add_argument(
        "--cursor-path",
        type=Path,
        default=None,
        help="Durable bus-cursor checkpoint path (default: beside receipt chain)",
    )
    parser.add_argument(
        "--start-at-end",
        action="store_true",
        help="On first start only, explicitly skip existing complete bus records",
    )
    parser.add_argument(
        "--max-records-per-poll",
        type=int,
        default=DEFAULT_MAX_RECORDS_PER_POLL,
        help=(
            "Maximum source records considered per poll "
            f"(default: {DEFAULT_MAX_RECORDS_PER_POLL})"
        ),
    )
    parser.add_argument(
        "--max-oversized-record-bytes",
        type=int,
        default=MAX_OVERSIZED_RECORD_BYTES,
        help=(
            "Hard fail-closed bound for scanning one malformed source record "
            f"(default: {MAX_OVERSIZED_RECORD_BYTES})"
        ),
    )
    parser.add_argument(
        "--max-rejection-markers",
        type=int,
        default=MAX_REJECTION_MARKERS,
        help=(
            "Number of durable malformed-source rejection markers retained "
            "in the rotating ring; the cumulative accumulator is retained "
            f"across rotations (default: {MAX_REJECTION_MARKERS})"
        ),
    )
    parser.add_argument(
        "--daemon",
        action="store_true",
        help="Run continuous polling loop",
    )
    parser.add_argument(
        "--verify-chain",
        action="store_true",
        help="Verify receipt chain integrity and exit",
    )
    parser.add_argument(
        "--expected-head",
        default=None,
        help="Trusted expected existing/verified chain head hash",
    )
    parser.add_argument(
        "--expected-count",
        type=int,
        default=None,
        help="Trusted expected existing/verified receipt count",
    )
    parser.add_argument(
        "--initialize-new-chain",
        action="store_true",
        help=(
            "Explicitly authorize first genesis creation on a missing or zero-byte "
            "receipt path; never use this flag for a restart"
        ),
    )
    parser.add_argument(
        "--allow-legacy-structural",
        action="store_true",
        help="For --verify-chain only: inspect unsigned v0.1 structurally",
    )
    parser.add_argument(
        "--allow-unanchored",
        action="store_true",
        help="For --verify-chain only: permit internal checks without head/count",
    )
    parser.add_argument(
        "--tail",
        type=int,
        metavar="N",
        help="Print last N receipts and exit",
    )
    parser.add_argument(
        "--keys",
        type=Path,
        default=None,
        help="Public verification-key registry (key_id -> Ed25519 public key); "
        "private key material is never accepted on the command line",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Enable debug logging",
    )

    args = parser.parse_args(argv)

    if args.initialize_new_chain and (
        args.verify_chain or args.tail is not None
    ):
        parser.error("--initialize-new-chain is valid only for writer modes")

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    verification_keys: dict[str, Any] = {}
    if args.keys is not None:
        try:
            if args.keys.stat().st_size > MAX_KEY_REGISTRY_BYTES:
                raise ValueError(
                    f"key registry exceeds {MAX_KEY_REGISTRY_BYTES} bytes"
                )
            loaded_keys = parse_key_registry(
                args.keys.read_text(encoding="utf-8"),
                source=f"keys file {args.keys}",
            )
        except (OSError, UnicodeError, ValueError) as exc:
            parser.error(f"could not load public verification keys: {exc}")
        verification_keys.update(loaded_keys)
    env_keys = os.environ.get("KRINEIA_SIGNING_KEYS")
    if env_keys:
        try:
            loaded_env_keys = parse_key_registry(
                env_keys, source="KRINEIA_SIGNING_KEYS"
            )
        except (UnicodeError, ValueError) as exc:
            parser.error(str(exc))
        verification_keys.update(loaded_env_keys)

    if args.verify_chain:
        count, errors, warnings = _tool_verify_chain(
            Path(args.receipt_path),
            verification_keys,
            allow_legacy_structural=args.allow_legacy_structural,
            expected_head=args.expected_head,
            expected_count=args.expected_count,
        )
        unanchored = args.expected_head is None or args.expected_count is None
        if unanchored:
            anchor_message = (
                "chain completeness was not checked against a trusted head and count"
            )
            if args.allow_unanchored:
                logger.warning(
                    "verify_chain: %s (explicit unanchored inspection policy)",
                    anchor_message,
                )
            else:
                errors.append(
                    anchor_message
                    + "; --allow-unanchored is permitted only for non-assurance inspection"
                )
        for warning in warnings:
            logger.warning("verify_chain: %s", warning)
        for error in errors:
            logger.error("verify_chain: %s", error)
        if errors or warnings:
            logger.error(
                "Chain invalid: %d receipt(s), %d error(s), %d warning(s)",
                count,
                len(errors),
                len(warnings),
            )
            return 1
        if unanchored:
            logger.info(
                "Chain internally valid under explicit unanchored inspection policy: %d receipt(s)",
                count,
            )
        else:
            logger.info("Chain anchored and valid: %d receipt(s)", count)
        return 0

    if args.tail is not None:
        receipt_path = Path(args.receipt_path)
        try:
            receipts = _bounded_tail(receipt_path, args.tail)
        except (ValueError, SourceIntegrityError) as exc:
            logger.error("tail: %s", exc)
            return 1
        for receipt in receipts:
            print(json.dumps(receipt, indent=2))
        return 0

    watcher = KrineiaWatcher(
        bus_path=args.bus_path,
        receipt_path=args.receipt_path,
        poll_interval=args.poll_interval,
        verification_keys=verification_keys,
        checkpoint_path=args.cursor_path,
        start_at_end=args.start_at_end,
        max_records_per_poll=args.max_records_per_poll,
        max_oversized_record_bytes=args.max_oversized_record_bytes,
        max_rejection_markers=args.max_rejection_markers,
        expected_existing_head=args.expected_head,
        expected_existing_count=args.expected_count,
        initialize_new_chain=args.initialize_new_chain,
    )

    if args.daemon:
        watcher.run()
        return 0

    # One-shot poll
    watcher._tick()
    logger.info("One-shot poll complete; receipts_written=%d", watcher._receipts_written)
    return 0


if __name__ == "__main__":
    sys.exit(main())