#!/usr/bin/env python3
"""Own the durable human-facing obligations of one vault-ingress run.

Processing completion and run completion are different events. A queue claim
closes when ``persist-results.py`` merges a batch; that says nothing about the
speaker. This ledger records what the run still owes after the merge: the
downstream steps that turn persisted returns into rendered analyses, summary,
profile, and goal outcomes; the clarification offer and its disposition; the
clarification session when the offer is accepted; and the end report. A run is
complete only when every obligation carries an explicit disposition. Silence,
elapsed time, and an invitation merely sent never resolve an obligation, so an
interrupted run shows up in ``pending`` until the speaker answers or the report
is delivered.

Usage:
    run-obligations.py <tracking-database.json> adopt --now <ISO-8601>
    run-obligations.py <tracking-database.json> open \
        --run-id <id> --now <ISO-8601> (--talk <talk.md> ... | --talks-from <file>) \
        [--from-run <run id whose closed claim persisted the talks>]
    run-obligations.py <tracking-database.json> record-downstream \
        --run-id <id> --now <ISO-8601>
    run-obligations.py <tracking-database.json> record-offer \
        --run-id <id> --now <ISO-8601> [--topic <text> ...] [--topics-from <file>]
    run-obligations.py <tracking-database.json> record-disposition \
        --run-id <id> --now <ISO-8601> \
        --disposition accepted|declined|deferred \
        [--return-condition <text> | --return-condition-from <file>]
    run-obligations.py <tracking-database.json> claim-session \
        --run-id <id> --now <ISO-8601> [--claim-token <token>]
    run-obligations.py <tracking-database.json> release-session \
        --run-id <id> --now <ISO-8601> --claim-token <token> \
        (--reason <text> | --reason-from <file>)
    run-obligations.py <tracking-database.json> count-session \
        --run-id <id> --claim-token <token>
    run-obligations.py <tracking-database.json> reconcile-count \
        --run-id <id> --now <ISO-8601> --claim-token <token> --counted yes|no
    run-obligations.py <tracking-database.json> record-session \
        --run-id <id> --now <ISO-8601> --claim-token <token> \
        --profile-inputs changed|unchanged [--profile-refreshed]
    run-obligations.py <tracking-database.json> record-report \
        --run-id <id> --now <ISO-8601> --report-file <delivered-report.md>
    run-obligations.py <tracking-database.json> dismiss \
        --run-id <id> --now <ISO-8601> (--reason <text> | --reason-from <file>)
    run-obligations.py <tracking-database.json> pending
    run-obligations.py <tracking-database.json> status --run-id <id>
    run-obligations.py <tracking-database.json> session-agenda [--run-id <id>]

Every successful command emits one JSON object on stdout and exits 0. Known
input/state errors emit a JSON error object on stdout, an actionable diagnostic
on stderr, and exit 2. Mutating commands rewrite the ledger atomically under the
same sibling lock discipline as the tracking database, only when state changed.
Every recording command is replay-safe: repeating it with the inputs it already
recorded is an unchanged success, so a caller that lost the first response can
retry; only a conflicting answer is refused.

Ledger: ``{vault_root}/ingress-obligations.json`` (schema 1, run records
schema 3), owned by vault-ingress and created by ``adopt``, which records
every claim already closed at adoption as history by its exact identity —
never by time, which ``persist-results.py --run-date`` can stamp at midnight —
so everything closed afterwards is reconciled. Delivered reports are copied to
``{vault_root}/ingress-reports/{stem}.{sha256}.md`` and bound to the ledger by
that digest, recorded vault-relative so the binding never depends on how a
command spelled the vault root. Each recorded talk carries the run id and release time of the
closed claim that persisted it, so ``pending`` can reconcile the ledger against
closed claims and name every persisted fact after the boundary that the ledger
does not cover (a whole run, a later batch of a recorded run, or a talk merged
again under the same run, that crashed between the merge and ``open``) without
repeating talks recovered under a fresh run id. An uncovered run stays listed
until it is opened or explicitly dismissed with a reason; a later run's
existence never stands in for either. ``pending`` also lists every deferred
offer with the speaker's return condition, so it can be raised again.
``session-agenda`` is vault-clarification's read: the accepted session that
has not completed, with the topics recorded at the offer as its seed agenda,
selected here (see ``pending_sessions``) so the session never picks one by
hand. ``claim-session`` reserves that session for one host and returns the
claim token every later session command carries; ``count-session`` counts it
toward the tracking database's ``clarification_sessions_completed`` while
holding the ledger's lock through that database write; ``record-session``
closes it only once its run is counted; ``release-session`` is the operator's explicit
decision to free one named claim whose host died. Field
meanings, transitions, and the reader/writer contract live in
``skills/vault-ingress/references/schemas-obligations.md``.

Recency policy (the delivery-recency buckets the clarification handoff keys on)
is encoded once, here, in the constants below. ``open`` and ``record-offer``
read each talk's ``date`` from the tracking database and compute
``days_since_delivery`` from ``--now``; the stored recency is a snapshot
labeled ``recency_as_of``, refreshed by ``record-offer`` at the moment the offer
is made and frozen from then on. The handoff never recomputes ``today - date``
by hand and reads the offer mode from ``record-offer``'s own output.
"""

from __future__ import annotations

import argparse
from datetime import date, datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import stat
import sys
from typing import Any

from queue_claim_contract import (
    KNOWN_STATUSES,
    QueueClaimContractError,
    require_queue_identifier,
)
from cooperative_lock import CooperativeLockError, exclusive_file_lock
from tracking_database import (
    CLARIFICATION_COUNTED_RUNS_FIELD,
    CLARIFICATION_SESSIONS_FIELD,
    LEGACY_COUNT_COUNTED,
    LEGACY_COUNT_FIELD,
    LEGACY_COUNT_STATES,
    LEGACY_COUNT_UNCOUNTED,
    LEGACY_COUNT_UNKNOWN,
    OBLIGATIONS_LEDGER_FILENAME,
    TrackingDatabaseError,
    counted_clarification_runs,
    require_current_tracking_database,
)
from tracking_database_io import (
    DATABASE_LOCK_LABEL,
    DATABASE_READ_DIAGNOSTICS,
    DATABASE_READ_FALLBACK,
    TrackingDatabaseIOError,
    TrackingDatabaseSnapshot,
    commit_tracking_database,
    decode_json_object,
    initialize_tracking_database,
    render_json_object,
    snapshot_tracking_database,
    write_json_object,
)
from vault_root_authority import (
    VaultRootAuthorityError,
    materialize_native_authority,
    resolve_vault_root_authority,
)

LEDGER_FILENAME = OBLIGATIONS_LEDGER_FILENAME
REPORTS_DIRECTORY = "ingress-reports"
PROFILE_FILENAME = "speaker-profile.json"
LEDGER_SCHEMA_VERSION = 1
# v2 records ``end_report.report_path`` vault-relative. v1 recorded it absolute
# under whatever spelling of the vault root the delivering command was given, so
# a vault reached through a symlink refused its own copies (#486). v3 adds the
# session claim and counter target (#466). Older records are upgraded on read
# and the read that upgrades them rewrites the ledger.
RUN_RECORD_SCHEMA_VERSION = 3
PRE_SESSION_CLAIM_RUN_RECORD_SCHEMA_VERSION = 2
LEGACY_RUN_RECORD_SCHEMA_VERSION = 1
READABLE_RUN_RECORD_SCHEMA_VERSIONS = frozenset(
    {
        LEGACY_RUN_RECORD_SCHEMA_VERSION,
        PRE_SESSION_CLAIM_RUN_RECORD_SCHEMA_VERSION,
        RUN_RECORD_SCHEMA_VERSION,
    }
)
# The report copy is content-addressed: ``{stem}.{sha256}.md`` with the run id
# reduced to a safe filename stem, so two deliveries never overwrite each
# other and a ledger-edited run id can never name a path outside the directory.
_SAFE_STEM = re.compile(r"[^A-Za-z0-9._-]")
REPORT_STEM_MAX = 40
REPORT_STEM_PREFIX = 31
REPORT_STEM_HASH = 8
_SHA256_HEX = re.compile(r"^[0-9a-f]{64}$")
# What ``persist-results.py`` stamps on the claim it closes; a talk whose claim
# carries it persisted, and one the ledger does not cover crashed between the
# merge and ``open``.
PERSISTED_RELEASE_REASON = "return_persisted"

# Talk statuses whose results can feed a clarification session. Skipped talks
# are recorded for the end report's scope but never generate an offer.
ANALYZED_STATUSES = frozenset({"processed", "processed_partial"})

# Delivery-recency buckets, inclusive upper bounds in whole days since delivery.
# ``same_week`` gets the inline offer; ``recent`` a recommended full session;
# ``older`` a recommended compressed session. A talk with no usable delivery
# date is ``unknown`` and treated like ``recent``: recommend, never auto-invoke.
SAME_WEEK_MAX_DAYS = 7
RECENT_MAX_DAYS = 30
BUCKET_SAME_WEEK = "same_week"
BUCKET_RECENT = "recent"
BUCKET_OLDER = "older"
BUCKET_UNKNOWN = "unknown"
OFFER_MODE_INLINE = "inline"
OFFER_MODE_RECOMMEND_FULL = "recommend_full"
OFFER_MODE_RECOMMEND_COMPRESSED = "recommend_compressed"
OFFER_MODE_NONE = "none"
OFFER_MODE_BY_BUCKET = {
    BUCKET_SAME_WEEK: OFFER_MODE_INLINE,
    BUCKET_RECENT: OFFER_MODE_RECOMMEND_FULL,
    BUCKET_UNKNOWN: OFFER_MODE_RECOMMEND_FULL,
    BUCKET_OLDER: OFFER_MODE_RECOMMEND_COMPRESSED,
}
# Strongest offer wins when a run mixes buckets.
OFFER_MODE_PRECEDENCE = (
    OFFER_MODE_INLINE,
    OFFER_MODE_RECOMMEND_FULL,
    OFFER_MODE_RECOMMEND_COMPRESSED,
)
OFFER_MODES = frozenset({*OFFER_MODE_PRECEDENCE, OFFER_MODE_NONE})

STATE_OWED = "owed"
STATE_OFFERED = "offered"
STATE_ACCEPTED = "accepted"
STATE_DECLINED = "declined"
STATE_DEFERRED = "deferred"
STATE_NOT_APPLICABLE = "not_applicable"
CLARIFICATION_STATES = frozenset(
    {
        STATE_OWED,
        STATE_OFFERED,
        STATE_ACCEPTED,
        STATE_DECLINED,
        STATE_DEFERRED,
        STATE_NOT_APPLICABLE,
    }
)
DISPOSITIONS = (STATE_ACCEPTED, STATE_DECLINED, STATE_DEFERRED)
SESSION_PENDING = "pending"
SESSION_RUNNING = "running"
SESSION_COMPLETED = "completed"
SESSION_STATES = frozenset({SESSION_PENDING, SESSION_RUNNING, SESSION_COMPLETED})
# A claim token is this many random bytes, hex-encoded; it names one claim.
CLAIM_TOKEN_BYTES = 16
_CLAIM_TOKEN = re.compile(rf"[0-9a-f]{{{CLAIM_TOKEN_BYTES * 2}}}")

DOWNSTREAM_OWED = "owed"
DOWNSTREAM_COMPLETED = "completed"
REPORT_OWED = "owed"
REPORT_DELIVERED = "delivered"

NEXT_OPEN = "open_obligations"
NEXT_DOWNSTREAM = "complete_downstream_steps"
NEXT_OFFER = "offer_clarification"
NEXT_AWAIT = "await_disposition"
NEXT_SESSION = "complete_clarification_session"
NEXT_REPORT = "deliver_end_report"
NEXT_NONE = "none"
REASON_UNRECORDED = "unrecorded_run"
REASON_MISSING_TALKS = "missing_talks"
REASON_AFTER_COMPLETION = "talks_persisted_after_completion"
REASON_AFTER_ANSWER = "talks_persisted_after_answer"

PROFILE_INPUTS_CHANGED = "changed"
PROFILE_INPUTS_UNCHANGED = "unchanged"
PROFILE_INPUTS = (PROFILE_INPUTS_CHANGED, PROFILE_INPUTS_UNCHANGED)

_ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


class RunObligationsError(Exception):
    """A known input or state error, reported on stdout as JSON and exit 2."""

    def __init__(self, message: str, *, reason_code: str) -> None:
        super().__init__(message)
        self.reason_code = reason_code


def _io_failure(
    exc: TrackingDatabaseIOError, subject: str, reason_code: str
) -> RunObligationsError:
    """Report an io-layer failure through its closed vocabulary, never verbatim.

    Decoder messages carry the rejected key or value; only the typed reason
    code travels, mapped to the shared diagnostic text with the subject named.
    """
    _code, message = DATABASE_READ_DIAGNOSTICS.get(
        exc.reason_code, DATABASE_READ_FALLBACK
    )
    return RunObligationsError(
        f"{message.replace('tracking database', subject)} (io reason: {exc.reason_code})",
        reason_code=reason_code,
    )


class JsonArgumentParser(argparse.ArgumentParser):
    """Convert argparse failures into the script's JSON error contract."""

    def error(self, message: str) -> Any:
        raise RunObligationsError(
            f"invalid arguments: {message}", reason_code="invalid_arguments"
        )


def parse_timestamp(value: object, label: str) -> datetime:
    """Parse a timezone-aware ISO-8601 timestamp and normalize it to UTC."""
    if not isinstance(value, str) or not value:
        raise RunObligationsError(
            f"{label} must be a non-empty ISO-8601 timestamp",
            reason_code="invalid_timestamp",
        )
    candidate = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        moment = datetime.fromisoformat(candidate)
    except ValueError as exc:
        raise RunObligationsError(
            f"{label} {value!r} is malformed — use a timezone-aware ISO-8601 "
            "timestamp such as 2026-09-14T18:00:00+00:00",
            reason_code="invalid_timestamp",
        ) from exc
    if moment.tzinfo is None:
        raise RunObligationsError(
            f"{label} {value!r} has no timezone — append an explicit UTC offset",
            reason_code="invalid_timestamp",
        )
    try:
        return moment.astimezone(timezone.utc).replace(microsecond=0)
    except (OverflowError, ValueError) as exc:
        raise RunObligationsError(
            f"{label} {value!r} is out of range once normalized to UTC — use a "
            "timestamp between years 1 and 9999 in UTC",
            reason_code="invalid_timestamp",
        ) from exc


def render_timestamp(moment: datetime) -> str:
    return moment.isoformat()


def parse_delivery_date(value: object) -> date | None:
    """Return the talk's delivery date, or None when it is absent or unusable."""
    if not isinstance(value, str) or not _ISO_DATE.match(value):
        return None
    try:
        return date.fromisoformat(value)
    except ValueError:
        return None


def recency_bucket(days_since_delivery: int | None) -> str:
    """Bucket a delivery age; a missing or future date is ``unknown``."""
    if days_since_delivery is None or days_since_delivery < 0:
        return BUCKET_UNKNOWN
    if days_since_delivery <= SAME_WEEK_MAX_DAYS:
        return BUCKET_SAME_WEEK
    if days_since_delivery <= RECENT_MAX_DAYS:
        return BUCKET_RECENT
    return BUCKET_OLDER


def offer_mode_for(talks: list[dict[str, Any]]) -> str:
    """Pick the strongest offer among the run's analyzed talks."""
    modes = {
        OFFER_MODE_BY_BUCKET[str(talk["recency_bucket"])]
        for talk in talks
        if talk["status"] in ANALYZED_STATUSES
    }
    for mode in OFFER_MODE_PRECEDENCE:
        if mode in modes:
            return mode
    return OFFER_MODE_NONE


def next_action(run: dict[str, Any]) -> str:
    if run["downstream"]["state"] == DOWNSTREAM_OWED:
        return NEXT_DOWNSTREAM
    clarification = run["clarification"]
    state = clarification["state"]
    if state == STATE_OWED:
        return NEXT_OFFER
    if state == STATE_OFFERED:
        return NEXT_AWAIT
    if state == STATE_ACCEPTED and clarification["session"]["state"] != (
        SESSION_COMPLETED
    ):
        return NEXT_SESSION
    if run["end_report"]["state"] == REPORT_OWED:
        return NEXT_REPORT
    return NEXT_NONE


def safe_report_stem(run_id: str) -> str:
    """A filename stem for the run id: sanitized, and bounded for long ids.

    A run id longer than ``REPORT_STEM_MAX`` characters keeps its first
    ``REPORT_STEM_PREFIX`` characters plus a short hash of the whole id, so the
    copy name stays under filesystem limits while distinct ids stay distinct.
    """
    stem = _SAFE_STEM.sub("_", run_id) or "run"
    if len(stem) <= REPORT_STEM_MAX:
        return stem
    tag = hashlib.sha256(run_id.encode("utf-8")).hexdigest()[:REPORT_STEM_HASH]
    return f"{stem[:REPORT_STEM_PREFIX]}-{tag}"


def bound_report_path(run_id: str, digest: str) -> str:
    """The vault-relative path a delivery of ``digest`` is bound to."""
    return f"{REPORTS_DIRECTORY}/{safe_report_stem(run_id)}.{digest}.md"


def clarification_resolved(run: dict[str, Any]) -> bool:
    clarification = run["clarification"]
    state = clarification["state"]
    if state in {STATE_DECLINED, STATE_DEFERRED, STATE_NOT_APPLICABLE}:
        return True
    if state == STATE_ACCEPTED:
        return clarification["session"]["state"] == SESSION_COMPLETED
    return False


# ── Ledger validation ─────────────────────────────────────────────────


def empty_ledger(
    adopted_at: str | None = None, adopted_facts: list[list[Any]] | None = None
) -> dict[str, Any]:
    return {
        "schema_version": LEDGER_SCHEMA_VERSION,
        "adopted_at": adopted_at,
        "adopted_facts": adopted_facts or [],
        "dismissed_runs": [],
        "runs": [],
    }


def fact_identity(fact: dict[str, Any], filename: str) -> list[Any]:
    """The exact identity of one persisted fact, as the ledger stores it."""
    return [
        str(fact["run_id"]),
        filename,
        str(fact["batch_id"]),
        int(fact["generation"]),
        str(fact["released_at"]),
    ]


def _validate_fact_identity(value: object, label: str) -> tuple[Any, ...]:
    if not isinstance(value, list) or len(value) != 5:
        raise _invalid(
            label, "must be [run_id, filename, batch_id, generation, released_at]"
        )
    run_id, filename, batch_id, generation, released_at = value
    _require_identifier(run_id, f"{label}[0]")
    if not isinstance(filename, str) or not filename:
        raise _invalid(label, "[1] filename must be a non-empty string")
    _require_identifier(batch_id, f"{label}[2]")
    if type(generation) is not int or generation < 0:
        raise _invalid(label, "[3] generation must be a non-negative integer")
    _require_timestamp(released_at, f"{label}[4]")
    return (run_id, filename, batch_id, generation, released_at)


def _validate_fact_list(value: object, label: str) -> set[tuple[Any, ...]]:
    if not isinstance(value, list):
        raise _invalid(label, "must be an array of fact identities")
    facts: set[tuple[Any, ...]] = set()
    for index, entry in enumerate(value):
        identity = _validate_fact_identity(entry, f"{label}[{index}]")
        if identity in facts:
            raise _invalid(label, f"[{index}] duplicates an identity")
        facts.add(identity)
    return facts


def require_adopted(ledger: dict[str, Any], path: Path) -> str:
    adopted_at = ledger.get("adopted_at")
    if not isinstance(adopted_at, str):
        raise RunObligationsError(
            f"obligations ledger {path} does not exist yet; run `adopt --now` "
            "at Step 1, before any batch, to set the reconciliation boundary",
            reason_code="ledger_not_adopted",
        )
    return adopted_at


def _invalid(label: str, detail: str) -> RunObligationsError:
    return RunObligationsError(f"{label} {detail}", reason_code="ledger_invalid")


def _require_keys(record: Any, keys: tuple[str, ...], label: str) -> dict[str, Any]:
    if not isinstance(record, dict):
        raise _invalid(label, "must be an object")
    for key in keys:
        if key not in record:
            raise _invalid(label, f"lacks {key}")
    return record


def _require_choice(value: object, choices: Any, label: str) -> str:
    """Type before membership: a list or object must not raise TypeError."""
    if not isinstance(value, str):
        raise _invalid(label, "must be a string")
    if value not in choices:
        raise _invalid(label, f"{value!r} is not one of the known values")
    return value


def _require_identifier(value: object, label: str) -> str:
    """Run ids share the queue claim's identifier contract."""
    try:
        return require_queue_identifier(value, label)
    except QueueClaimContractError as exc:
        raise _invalid(label, str(exc).removeprefix(label).strip()) from exc


def _require_timestamp(value: object, label: str) -> None:
    """Stored stamps are canonical UTC whole seconds, so they compare as text."""
    try:
        moment = parse_timestamp(value, label)
    except RunObligationsError as exc:
        raise _invalid(label, str(exc).removeprefix(label).strip()) from exc
    if render_timestamp(moment) != value:
        raise _invalid(
            label,
            f"{value!r} must be the canonical UTC form {render_timestamp(moment)!r}",
        )


def _require_timestamp_or_null(value: object, label: str, *, present: bool) -> None:
    """A stamp is required exactly when its state says the event happened."""
    if present:
        if value is None:
            raise _invalid(label, "must be set in this state")
        _require_timestamp(value, label)
    elif value is not None:
        raise _invalid(label, "must be null in this state")


def _require_null(value: object, label: str) -> None:
    if value is not None:
        raise _invalid(label, "must be null in this state")


def _require_bool(value: object, label: str) -> None:
    if not isinstance(value, bool):
        raise _invalid(label, "must be a boolean")


def _validate_talk(entry: Any, label: str) -> None:
    talk = _require_keys(
        entry,
        (
            "filename",
            "status",
            "delivery_date",
            "days_since_delivery",
            "recency_bucket",
            "claim_run_id",
            "claim_batch_id",
            "claim_generation",
            "claim_released_at",
        ),
        label,
    )
    if not isinstance(talk["filename"], str) or not talk["filename"]:
        raise _invalid(label, "filename must be a non-empty string")
    _require_choice(talk["status"], KNOWN_STATUSES, f"{label}.status")
    delivered = talk["delivery_date"]
    if delivered is not None and (
        not isinstance(delivered, str) or parse_delivery_date(delivered) is None
    ):
        raise _invalid(label, "delivery_date must be a YYYY-MM-DD string or null")
    days = talk["days_since_delivery"]
    if days is not None and (type(days) is not int):
        raise _invalid(label, "days_since_delivery must be an integer or null")
    _require_choice(
        talk["recency_bucket"], OFFER_MODE_BY_BUCKET, f"{label}.recency_bucket"
    )
    _require_identifier(talk["claim_run_id"], f"{label}.claim_run_id")
    _require_identifier(talk["claim_batch_id"], f"{label}.claim_batch_id")
    generation = talk["claim_generation"]
    if type(generation) is not int or generation < 0:
        raise _invalid(label, "claim_generation must be a non-negative integer")
    _require_timestamp(talk["claim_released_at"], f"{label}.claim_released_at")


def _validate_session_claim(session: dict[str, Any], state: str, label: str) -> None:
    """A running session holds a claim; a pending one holds none.

    A completed session keeps the claim it completed under, or none when it
    completed before claims existed.
    """
    token = session["claim_token"]
    claimed = state == SESSION_RUNNING or (
        state == SESSION_COMPLETED and token is not None
    )
    if claimed:
        if not isinstance(token, str) or not _CLAIM_TOKEN.fullmatch(token):
            raise _invalid(
                label,
                f"claim_token must be {CLAIM_TOKEN_BYTES * 2} lowercase hex digits",
            )
        _require_timestamp(session["claimed_at"], f"{label}.claimed_at")
    else:
        _require_null(token, f"{label}.claim_token")
        _require_null(session["claimed_at"], f"{label}.claimed_at")


def _validate_clarification(record: Any, label: str) -> None:
    clarification = _require_keys(
        record,
        (
            "state",
            "offer_mode",
            "topics",
            "recency_as_of",
            "offered_at",
            "resolved_at",
            "return_condition",
            "session",
        ),
        label,
    )
    state = _require_choice(
        clarification["state"], CLARIFICATION_STATES, f"{label}.state"
    )
    _require_choice(clarification["offer_mode"], OFFER_MODES, f"{label}.offer_mode")
    if not isinstance(clarification["topics"], list) or not all(
        isinstance(topic, str) for topic in clarification["topics"]
    ):
        raise _invalid(label, "topics must be an array of strings")
    if clarification["recency_as_of"] is not None:
        _require_timestamp(clarification["recency_as_of"], f"{label}.recency_as_of")
    offered = state not in {STATE_OWED, STATE_NOT_APPLICABLE}
    resolved = state in {STATE_ACCEPTED, STATE_DECLINED, STATE_DEFERRED}
    _require_timestamp_or_null(
        clarification["offered_at"], f"{label}.offered_at", present=offered
    )
    _require_timestamp_or_null(
        clarification["resolved_at"], f"{label}.resolved_at", present=resolved
    )
    condition = clarification["return_condition"]
    if state == STATE_DEFERRED:
        if not isinstance(condition, str) or not condition.strip():
            raise _invalid(
                label, "return_condition must name when to raise a deferred offer"
            )
    else:
        _require_null(condition, f"{label}.return_condition")
    session = clarification["session"]
    if state != STATE_ACCEPTED:
        _require_null(session, f"{label}.session")
        return
    session = _require_keys(
        session,
        (
            "state",
            "completed_at",
            "profile_inputs",
            "profile_refreshed",
            "claim_token",
            "claimed_at",
        ),
        f"{label}.session",
    )
    session_state = _require_choice(
        session["state"], SESSION_STATES, f"{label}.session.state"
    )
    _validate_session_claim(session, session_state, f"{label}.session")
    if LEGACY_COUNT_FIELD in session:
        _require_choice(
            session[LEGACY_COUNT_FIELD],
            LEGACY_COUNT_STATES,
            f"{label}.session.{LEGACY_COUNT_FIELD}",
        )
    completed = session_state == SESSION_COMPLETED
    _require_timestamp_or_null(
        session["completed_at"], f"{label}.session.completed_at", present=completed
    )
    if completed:
        _require_choice(
            session["profile_inputs"], PROFILE_INPUTS, f"{label}.session.profile_inputs"
        )
        _require_bool(
            session["profile_refreshed"], f"{label}.session.profile_refreshed"
        )
    else:
        _require_null(session["profile_inputs"], f"{label}.session.profile_inputs")
        _require_null(
            session["profile_refreshed"], f"{label}.session.profile_refreshed"
        )


def _validate_downstream(record: Any, label: str) -> None:
    downstream = _require_keys(record, ("state", "completed_at"), label)
    state = _require_choice(
        downstream["state"], {DOWNSTREAM_OWED, DOWNSTREAM_COMPLETED}, f"{label}.state"
    )
    _require_timestamp_or_null(
        downstream["completed_at"],
        f"{label}.completed_at",
        present=state == DOWNSTREAM_COMPLETED,
    )


def _validate_report(record: Any, label: str, expected_path: Any) -> bool:
    """``expected_path`` maps a digest to the copy path this vault binds."""
    report = _require_keys(
        record,
        ("state", "delivered_at", "report_path", "report_sha256", "reopened_at"),
        label,
    )
    delivered = (
        _require_choice(
            report["state"], {REPORT_OWED, REPORT_DELIVERED}, f"{label}.state"
        )
        == REPORT_DELIVERED
    )
    _require_timestamp_or_null(
        report["delivered_at"], f"{label}.delivered_at", present=delivered
    )
    if delivered:
        if not isinstance(report["report_sha256"], str) or not _SHA256_HEX.match(
            report["report_sha256"]
        ):
            raise _invalid(label, "report_sha256 must be a hex SHA-256 once delivered")
        bound = expected_path(report["report_sha256"])
        if report["report_path"] != bound:
            raise _invalid(
                label,
                f"report_path must be the bound copy {bound!r} for its digest",
            )
    else:
        _require_null(report["report_path"], f"{label}.report_path")
        _require_null(report["report_sha256"], f"{label}.report_sha256")
    if report["reopened_at"] is not None:
        _require_timestamp(report["reopened_at"], f"{label}.reopened_at")
    return delivered


def _upgrade_legacy_report_path(
    run: dict[str, Any], label: str, vault_root: Path
) -> None:
    """Read a v1 run record as v2 by making its absolute report path relative.

    The recorded path is stripped of the vault root as spelled and, with both
    sides resolved, of the real vault root; the candidate naming the bound copy
    wins, so a copy recorded under the symlink, its target, or an alias nested
    inside the vault reads the same. A path outside the vault stays refused, and
    the bound-copy check that follows refuses any other name.
    """
    run["schema_version"] = PRE_SESSION_CLAIM_RUN_RECORD_SCHEMA_VERSION
    report = run.get("end_report")
    if not isinstance(report, dict) or not isinstance(report.get("report_path"), str):
        return
    recorded = Path(report["report_path"])
    if not recorded.is_absolute():
        raise _invalid(
            label,
            "end_report report_path must be absolute in a schema-1 run record; "
            "run-obligations.py never wrote this value, so restore the ledger "
            "from a copy the owner wrote and re-run the command",
        )
    digest = report.get("report_sha256")
    bound = (
        bound_report_path(str(run.get("run_id")), digest)
        if isinstance(digest, str)
        else None
    )
    candidates: list[str] = []
    if recorded.is_relative_to(vault_root):
        candidates.append(recorded.relative_to(vault_root).as_posix())
    if bound not in candidates:
        # The persisted path is untrusted ledger content, so neither it nor the
        # resolver's exception text reaches the diagnostic. ValueError is an
        # embedded NUL, which JSON can carry and `resolve` refuses.
        try:
            resolved, real_root = recorded.resolve(), vault_root.resolve()
        except (OSError, RuntimeError, ValueError) as exc:
            raise _invalid(
                label,
                "end_report report_path cannot be resolved against this vault; "
                "make the vault path readable, or restore the ledger from a copy "
                "the owner wrote, and re-run the command",
            ) from exc
        if resolved.is_relative_to(real_root):
            candidates.append(resolved.relative_to(real_root).as_posix())
    if not candidates:
        raise _invalid(
            label,
            "end_report report_path is not inside this vault; "
            "pass the tracking database at the vault location this report was "
            "delivered under (or a link to it) so the owner can rewrite the path "
            "vault-relative, or restore the ledger from a copy the owner wrote",
        )
    report["report_path"] = bound if bound in candidates else candidates[0]


def _require_derived_offer_mode(run: dict[str, Any], label: str) -> None:
    """Refuse an offer mode its recorded talks do not produce.

    ``open`` and ``record-offer`` derive the mode from the talks they record
    in the same write, so a mismatch is a hand edit, and ``not_applicable`` is
    the state exactly when no analyzed talk is recorded.
    """
    clarification = run["clarification"]
    derived = offer_mode_for(run["talks"])
    if clarification["offer_mode"] != derived:
        raise _invalid(
            label,
            f"offer_mode {clarification['offer_mode']!r} is not the "
            f"{derived!r} its recorded talks give",
        )
    if (clarification["state"] == STATE_NOT_APPLICABLE) != (derived == OFFER_MODE_NONE):
        raise _invalid(
            label,
            f"state {clarification['state']!r} disagrees with offer_mode "
            f"{derived!r}; not_applicable is the state exactly when no analyzed "
            "talk is recorded",
        )


def _upgrade_session_claim(run: dict[str, Any], label: str) -> None:
    """Read a v2 run record as v3: a session recorded before claims has none.

    A v2 session is pending or completed and carries no claim fields; one that
    claims otherwise was not written by the owner, so it is refused rather than
    upgraded into a claim nobody holds.
    """
    run["schema_version"] = RUN_RECORD_SCHEMA_VERSION
    clarification = run.get("clarification")
    session = clarification.get("session") if isinstance(clarification, dict) else None
    if not isinstance(session, dict):
        return
    state = session.get("state")
    if not isinstance(state, str):
        raise _invalid(
            label,
            f"is a schema-2 record whose session state {state!r} is not a string; "
            "restore the ledger from a copy the owner wrote",
        )
    if state not in (SESSION_PENDING, SESSION_COMPLETED) or any(
        key in session for key in ("claim_token", "claimed_at")
    ):
        raise _invalid(
            label,
            "is a schema-2 record whose session carries a claim, which schema 2 "
            "never recorded; restore the ledger from a copy the owner wrote",
        )
    session["claim_token"] = None
    session["claimed_at"] = None
    if state == SESSION_PENDING:
        # Schema 2's Step 9 advanced the counter before recording the session,
        # so a pending one may already be counted; reconcile-count decides.
        session[LEGACY_COUNT_FIELD] = LEGACY_COUNT_UNKNOWN


def _validate_run(run: dict[str, Any], label: str, vault_root: Path) -> None:
    """Refuse a run record whose shape a command would otherwise trip over."""
    _require_keys(
        run,
        (
            "schema_version",
            "opened_at",
            "updated_at",
            "talks",
            "downstream",
            "clarification",
            "end_report",
            "completed_at",
        ),
        label,
    )
    version = run["schema_version"]
    if type(version) is not int or version not in READABLE_RUN_RECORD_SCHEMA_VERSIONS:
        newer = type(version) is int and version > RUN_RECORD_SCHEMA_VERSION
        recovery = (
            "A newer speaker-toolkit wrote it: update speaker-toolkit and re-run"
            if newer
            else "No speaker-toolkit writes that value: inspect the ledger, or "
            "restore it from a copy the owner wrote, and re-run"
        )
        raise _invalid(
            label,
            f"has run-record schema_version {version!r}; this script reads "
            f"{', '.join(map(str, sorted(READABLE_RUN_RECORD_SCHEMA_VERSIONS)))} "
            f"only. {recovery}; the ledger is left as it is",
        )
    if version == LEGACY_RUN_RECORD_SCHEMA_VERSION:
        _upgrade_legacy_report_path(run, label, vault_root)
    if run["schema_version"] == PRE_SESSION_CLAIM_RUN_RECORD_SCHEMA_VERSION:
        _upgrade_session_claim(run, label)
    for key in ("opened_at", "updated_at"):
        _require_timestamp(run[key], f"{label} {key}")
    if not isinstance(run["talks"], list):
        raise _invalid(label, "talks must be an array")
    for position, talk in enumerate(run["talks"]):
        _validate_talk(talk, f"{label} talks[{position}]")
    _validate_downstream(run["downstream"], f"{label} downstream")
    _validate_clarification(run["clarification"], f"{label} clarification")
    _require_derived_offer_mode(run, f"{label} clarification")
    delivered = _validate_report(
        run["end_report"],
        f"{label} end_report",
        lambda digest: bound_report_path(str(run["run_id"]), digest),
    )
    _require_timestamp_or_null(
        run["completed_at"], f"{label} completed_at", present=delivered
    )
    if delivered:
        # A report goes out only after the downstream steps and the offer's
        # answer; a record claiming otherwise is not one this script wrote.
        if run["downstream"]["state"] != DOWNSTREAM_COMPLETED:
            raise _invalid(label, "end_report is delivered while downstream is owed")
        if run["clarification"]["state"] in {STATE_OWED, STATE_OFFERED}:
            raise _invalid(
                label, "end_report is delivered while the offer has no answer"
            )
        if run["completed_at"] != run["end_report"]["delivered_at"]:
            raise _invalid(label, "completed_at must equal end_report.delivered_at")


def validate_ledger(payload: object, path: Path, vault_root: Path) -> dict[str, Any]:
    """Accept only the owner's current ledger shape, bound to this vault."""
    if not isinstance(payload, dict):
        raise RunObligationsError(
            f"obligations ledger {path} must be a JSON object",
            reason_code="ledger_invalid",
        )
    version = payload.get("schema_version")
    if type(version) is not int or version != LEDGER_SCHEMA_VERSION:
        raise RunObligationsError(
            f"obligations ledger {path} has schema_version {version!r}; this "
            f"script reads schema {LEDGER_SCHEMA_VERSION} only — update "
            "speaker-toolkit or inspect the ledger by hand",
            reason_code="ledger_schema_unsupported",
        )
    label = f"obligations ledger {path}"
    if "adopted_at" not in payload:
        raise _invalid(label, "lacks adopted_at")
    _require_timestamp(payload["adopted_at"], f"{label} adopted_at")
    if "adopted_facts" not in payload:
        raise _invalid(label, "lacks adopted_facts")
    _validate_fact_list(payload["adopted_facts"], f"{label} adopted_facts")
    dismissed = payload.get("dismissed_runs")
    if not isinstance(dismissed, list):
        raise _invalid(label, "must carry a dismissed_runs array")
    dismissed_ids: set[str] = set()
    for index, entry in enumerate(dismissed):
        record = _require_keys(
            entry,
            ("run_id", "dismissed_at", "reason", "facts"),
            f"{label} dismissed_runs[{index}]",
        )
        dismissed_id = _require_identifier(
            record["run_id"], f"{label} dismissed_runs[{index}].run_id"
        )
        if dismissed_id in dismissed_ids:
            raise _invalid(
                label, f"dismissed_runs[{index}] duplicates {dismissed_id!r}"
            )
        dismissed_ids.add(dismissed_id)
        _require_timestamp(
            record["dismissed_at"], f"{label} dismissed_runs[{index}].dismissed_at"
        )
        if not isinstance(record["reason"], str) or not record["reason"].strip():
            raise _invalid(label, f"dismissed_runs[{index}].reason must say why")
        for identity in _validate_fact_list(
            record["facts"], f"{label} dismissed_runs[{index}].facts"
        ):
            if identity[0] != dismissed_id:
                raise _invalid(label, f"dismissed_runs[{index}].facts name another run")
    runs = payload.get("runs")
    if not isinstance(runs, list):
        raise RunObligationsError(
            f"obligations ledger {path} must carry a runs array",
            reason_code="ledger_invalid",
        )
    seen: set[str] = set()
    for index, run in enumerate(runs):
        if not isinstance(run, dict):
            raise RunObligationsError(
                f"obligations ledger {path} runs[{index}] must be an object",
                reason_code="ledger_invalid",
            )
        run_id = _require_identifier(
            run.get("run_id"), f"obligations ledger {path} runs[{index}].run_id"
        )
        if run_id in seen:
            raise RunObligationsError(
                f"obligations ledger {path} runs[{index}] duplicates run_id {run_id!r}",
                reason_code="ledger_invalid",
            )
        seen.add(run_id)
        _validate_run(run, f"obligations ledger {path} run {run_id!r}", vault_root)
    return payload


# ── Ledger IO ─────────────────────────────────────────────────────────


def load_ledger(
    path: Path, vault_root: Path
) -> tuple[TrackingDatabaseSnapshot | None, dict[str, Any]]:
    """Read the ledger with its generation, or an empty ledger when absent."""
    if not path.exists():
        return None, empty_ledger()
    try:
        snapshot = snapshot_tracking_database(path)
        payload = decode_json_object(snapshot)
    except TrackingDatabaseIOError as exc:
        code = (
            "ledger_invalid"
            if exc.reason_code.startswith(("json_", "encoding_"))
            else "ledger_unreadable"
        )
        raise _io_failure(exc, f"obligations ledger {path}", code) from exc
    runs = payload.get("runs") if isinstance(payload, dict) else None
    legacy = isinstance(runs, list) and any(
        isinstance(run, dict) and run.get("schema_version") != RUN_RECORD_SCHEMA_VERSION
        for run in runs
    )
    ledger = validate_ledger(payload, path, vault_root)
    if not legacy:
        return snapshot, ledger
    # The owner rewrites what it upgrades on the read that upgrades it, so an
    # older record never outlives the first command to read it. The caller
    # gets the rewritten generation, so a transition's later write still binds.
    store_ledger(path, snapshot, ledger)
    return load_ledger(path, vault_root)


def store_ledger(
    path: Path, snapshot: TrackingDatabaseSnapshot | None, payload: dict[str, Any]
) -> dict[str, Any]:
    """Commit the ledger atomically against the generation that was read.

    The returned fields say what the write achieved. ``durability_state`` other
    than ``durable`` or ``unchanged`` means the bytes were installed but a
    verification or directory fsync failed; every such warning also goes to
    stderr so a caller reading only the exit code still sees it.
    """
    try:
        if snapshot is None:
            result = initialize_tracking_database(path, payload)
        else:
            result = write_json_object(snapshot, payload)
    except TrackingDatabaseIOError as exc:
        raise _io_failure(
            exc, f"obligations ledger {path}", "ledger_write_failed"
        ) from exc
    for warning in result.warnings:
        print(f"WARNING: obligations ledger {path}: {warning}", file=sys.stderr)
    return {
        "written": bool(result.installed),
        "durability_state": result.durability_state,
        "warnings": list(result.warnings),
    }


def _copy_failed(detail: str) -> RunObligationsError:
    return RunObligationsError(
        f"{detail}; the delivered text was not recorded",
        reason_code="report_copy_failed",
    )


def _open_directory_no_follow(directory: Path) -> int:
    """A descriptor on the real directory; a symlink at that path is refused."""
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        return os.open(directory, flags)
    except OSError as exc:
        raise _copy_failed(
            f"{directory} is not a real directory ({exc}); the report copy is "
            "installed only in a plain directory inside the vault — replace "
            "whatever occupies that path with a directory and re-run "
            "record-report"
        ) from exc


def _read_regular_file(
    descriptor_flags: int, name: str, *, dir_fd: int | None
) -> bytes | None:
    """Bytes of an existing regular file, None when nothing is there.

    Anything else at the name — a link, a FIFO, a directory — is refused, and
    a filesystem failure while inspecting or reading it is reported as the
    structured copy failure, so a caller never blocks on, follows, or
    tracebacks over what it did not write.
    """
    try:
        descriptor = os.open(name, descriptor_flags, dir_fd=dir_fd)
    except FileNotFoundError:
        return None
    except OSError as exc:
        raise _copy_failed(
            f"{name} cannot be opened without following links ({exc})"
        ) from exc
    try:
        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
            raise _copy_failed(
                f"{name} exists and is not a regular file; remove it and re-run "
                "record-report"
            )
        chunks: list[bytes] = []
        while True:
            chunk = os.read(descriptor, 1 << 20)
            if not chunk:
                return b"".join(chunks)
            chunks.append(chunk)
    except OSError as exc:
        raise _copy_failed(
            f"{name} could not be read ({exc}); check the filesystem and re-run "
            "record-report"
        ) from exc
    finally:
        try:
            os.close(descriptor)
        except OSError as exc:
            print(
                f"WARNING: descriptor for {name} did not close: {exc}", file=sys.stderr
            )


def write_report_copy(
    directory: Path, run_id: str, digest: str, content: bytes
) -> tuple[Path, bool]:
    """Install the delivered report beside the ledger, durably.

    Returns the content-addressed path and whether this call created it. Every
    step runs relative to a descriptor opened on the real directory without
    following links, so nothing that happens to the path between the check
    and the write can redirect the copy outside the vault. An existing
    regular-file copy with the same bytes is left alone, so a replayed or
    racing delivery of identical text never rewrites it; anything else at the
    target is refused.
    """
    name = f"{safe_report_stem(run_id)}.{digest}.md"
    target = directory / name
    try:
        created = not directory.exists()
        directory.mkdir(parents=True, exist_ok=True)
        if created:
            # A new directory entry is durable only once its parent is synced;
            # a replay that changes nothing else must not leave it in doubt.
            parent_fd = os.open(directory.parent, os.O_RDONLY)
            try:
                os.fsync(parent_fd)
            finally:
                os.close(parent_fd)
    except OSError as exc:
        raise _copy_failed(
            f"cannot create {directory}: {exc} — make its parent writable (or "
            "remove whatever occupies that path) and re-run record-report"
        ) from exc
    dir_fd = _open_directory_no_follow(directory)
    try:
        no_follow = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | os.O_NONBLOCK
        existing = _read_regular_file(no_follow, name, dir_fd=dir_fd)
        if existing == content:
            # The bytes may be on disk while their directory entry is not; a
            # replay that finds the copy still makes its name durable.
            try:
                os.fsync(dir_fd)
            except OSError as exc:
                raise _copy_failed(
                    f"cannot sync {directory} for the existing report copy "
                    f"{target}: {exc} — fix the volume and re-run record-report"
                ) from exc
            return target, False
        # Random, so an orphan a crashed process left can never collide with
        # a later process that reused its pid.
        staged = f".{name}.{secrets.token_hex(8)}.tmp"
        try:
            descriptor = os.open(
                staged,
                os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
                0o600,
                dir_fd=dir_fd,
            )
        except OSError as exc:
            raise _copy_failed(
                f"cannot stage the report copy under {directory}: {exc} — make "
                "it writable and re-run record-report"
            ) from exc
        try:
            with os.fdopen(descriptor, "wb") as handle:
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(staged, name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
            os.fsync(dir_fd)
        except OSError as exc:
            try:
                os.unlink(staged, dir_fd=dir_fd)
            except OSError as cleanup:
                # Best effort: the write failure is the error to report; a
                # staged file that cannot be removed is named, never allowed
                # to mask it.
                print(
                    f"WARNING: staged report copy {directory / staged} was not "
                    f"removed: {cleanup}",
                    file=sys.stderr,
                )
            raise _copy_failed(
                f"cannot write the report copy {target}: {exc} — free the space "
                "or fix the permissions and re-run record-report"
            ) from exc
    finally:
        os.close(dir_fd)
    return target, True


# ── Database context ──────────────────────────────────────────────────


def persisted_claims(record: dict[str, Any]) -> list[dict[str, Any]]:
    """Every closed claim that persisted this talk, by its full identity.

    ``persist-results.py`` stamps one ``released_at`` on a whole batch, so the
    identity of a persisted fact is the claim's run id, batch id, reprocess
    generation, and release time together.
    """
    claims = [record.get("_queue_claim")]
    history = record.get("_queue_claim_history")
    if isinstance(history, list):
        claims.extend(history)
    found: list[dict[str, Any]] = []
    for claim in claims:
        if not isinstance(claim, dict):
            continue
        generation = claim.get("reprocess_generation")
        if (
            claim.get("state") != "completed"
            or claim.get("release_reason") != PERSISTED_RELEASE_REASON
            or not isinstance(claim.get("run_id"), str)
            or not isinstance(claim.get("batch_id"), str)
            or type(generation) is not int
            or not isinstance(claim.get("released_at"), str)
        ):
            continue
        try:
            released = parse_timestamp(claim["released_at"], "released_at")
        except RunObligationsError:
            continue
        found.append(
            {
                "run_id": claim["run_id"],
                "batch_id": claim["batch_id"],
                "generation": generation,
                "released_at": render_timestamp(released),
            }
        )
    return found


def _fact_order(fact: dict[str, Any]) -> tuple[int, str]:
    """Newer means a later reprocess generation, then a later release.

    The queue assigns each closed claim of a talk its own generation, so two
    facts of one talk under one run never tie; the batch id is identity, never
    order.
    """
    return (int(fact["generation"]), str(fact["released_at"]))


class Context:
    """The database-bound paths every command resolves the same way."""

    def __init__(self, raw_database: str) -> None:
        self.database_path = materialize_native_authority(
            raw_database, authority="database_path"
        )
        self.database, self.vault_root = self._read()
        self.ledger_path = self.vault_root / LEDGER_FILENAME
        self.reports_directory = self.vault_root / REPORTS_DIRECTORY
        self.profile_path = self.vault_root / PROFILE_FILENAME

    def refresh(self) -> dict[str, Any]:
        """Re-read the current database generation right before it is used.

        The vault root is re-resolved from the fresh generation's config and
        must agree with the one the command started under; obligations are
        never written under an authority the database no longer asserts.
        """
        database, vault_root = self._read()
        if vault_root != self.vault_root:
            raise RunObligationsError(
                f"the tracking database now asserts vault root {vault_root} "
                f"but this command started under {self.vault_root}; re-run it",
                reason_code="vault_root_changed",
            )
        self.database = database
        return database

    def _read(self) -> tuple[dict[str, Any], Path]:
        """One current database generation and the vault root it asserts."""
        try:
            snapshot = snapshot_tracking_database(self.database_path)
            database = decode_json_object(snapshot)
        except TrackingDatabaseIOError as exc:
            raise _io_failure(
                exc, f"tracking database {self.database_path}", "database_unusable"
            ) from exc
        try:
            require_current_tracking_database(database)
        except TrackingDatabaseError as exc:
            # The owner's own generation diagnostics name reason codes and
            # record labels, never raw values.
            raise RunObligationsError(
                str(exc), reason_code="database_unusable"
            ) from exc
        vault_root = resolve_vault_root_authority(
            database_path=self.database_path, config=database.get("config")
        )
        return database, vault_root

    def persisted_runs(self) -> dict[str, dict[str, Any]]:
        """Runs whose closed claims say persist-results.py merged their talks.

        ``talks`` maps each filename to every persisted fact under that run,
        so a talk merged again under the same run id is a distinct fact.
        """
        runs: dict[str, dict[str, Any]] = {}
        for record in self.database["talks"]:
            if not isinstance(record, dict):
                continue
            filename = record.get("filename")
            if not isinstance(filename, str):
                continue
            for fact in persisted_claims(record):
                entry = runs.setdefault(
                    fact["run_id"],
                    {"run_id": fact["run_id"], "talks": {}, "latest_released_at": None},
                )
                entry["talks"].setdefault(filename, []).append(fact)
                if (
                    entry["latest_released_at"] is None
                    or fact["released_at"] > entry["latest_released_at"]
                ):
                    entry["latest_released_at"] = fact["released_at"]
        return runs

    def talk(self, filename: str) -> dict[str, Any]:
        for record in self.database["talks"]:
            if isinstance(record, dict) and record.get("filename") == filename:
                return record
        raise RunObligationsError(
            f"talk {filename!r} is not in {self.database_path}; pass the exact "
            "filenames persist-results.py merged for this batch",
            reason_code="talk_not_found",
        )


def find_run(ledger: dict[str, Any], run_id: str) -> dict[str, Any]:
    for run in ledger["runs"]:
        if run["run_id"] == run_id:
            return run
    raise RunObligationsError(
        f"run {run_id!r} has no obligations record; run `open` for it after "
        "persisting its first batch",
        reason_code="run_not_found",
    )


def require_run_id(value: str) -> str:
    try:
        return require_queue_identifier(value, "--run-id")
    except QueueClaimContractError as exc:
        raise RunObligationsError(str(exc), reason_code="invalid_arguments") from exc


def newest_claim(
    claims: list[dict[str, Any]], run_id: str | None
) -> dict[str, Any] | None:
    """The newest closed claim under ``run_id``, or of any run when None."""
    matching = [
        claim for claim in claims if run_id is None or claim["run_id"] == run_id
    ]
    return max(matching, key=_fact_order) if matching else None


def describe_talk(
    context: Context,
    filename: str,
    now: datetime,
    run_id: str,
    *,
    source_run: str | None = None,
    keep: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """The talk as the run records it, linked to the claim that persisted it.

    ``keep`` preserves an already-recorded link through a recency refresh.
    Otherwise the link is the newest closed claim of ``source_run`` when the
    caller names one (recovery of an exact persisted fact under a fresh run
    id), else this run's own newest claim, else the newest claim of any run.
    """
    record = context.talk(filename)
    claims = persisted_claims(record)
    if keep is not None:
        link: dict[str, Any] | None = keep
    elif source_run is not None:
        link = newest_claim(claims, source_run)
        if link is None:
            raise RunObligationsError(
                f"talk {filename!r} has no closed return_persisted claim under "
                f"run {source_run!r} in {context.database_path}; --from-run "
                "names the run whose claim persisted the talk",
                reason_code="talk_not_persisted",
            )
    else:
        link = newest_claim(claims, run_id) or newest_claim(claims, None)
        if link is None:
            raise RunObligationsError(
                f"talk {filename!r} has no closed return_persisted claim in "
                f"{context.database_path}; open records only talks "
                "persist-results.py merged",
                reason_code="talk_not_persisted",
            )
    delivered = parse_delivery_date(record.get("date"))
    days = (now.date() - delivered).days if delivered is not None else None
    return {
        "filename": filename,
        "status": str(record.get("status")),
        "delivery_date": delivered.isoformat() if delivered is not None else None,
        "days_since_delivery": days,
        "recency_bucket": recency_bucket(days),
        "claim_run_id": link["run_id"],
        "claim_batch_id": link["batch_id"],
        "claim_generation": link["generation"],
        "claim_released_at": link["released_at"],
    }


def _link(talk: dict[str, Any]) -> dict[str, Any]:
    return {
        "run_id": str(talk["claim_run_id"]),
        "batch_id": str(talk["claim_batch_id"]),
        "generation": int(talk["claim_generation"]),
        "released_at": str(talk["claim_released_at"]),
    }


def new_run(run_id: str, now: str) -> dict[str, Any]:
    return {
        "schema_version": RUN_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "opened_at": now,
        "updated_at": now,
        "talks": [],
        "downstream": {"state": DOWNSTREAM_OWED, "completed_at": None},
        "clarification": {
            "state": STATE_OWED,
            "offer_mode": OFFER_MODE_NONE,
            "topics": [],
            "recency_as_of": None,
            "offered_at": None,
            "resolved_at": None,
            "return_condition": None,
            "session": None,
        },
        "end_report": {
            "state": REPORT_OWED,
            "delivered_at": None,
            "report_path": None,
            "report_sha256": None,
            "reopened_at": None,
        },
        "completed_at": None,
    }


def summarize(run: dict[str, Any]) -> dict[str, Any]:
    return {
        "run_id": run["run_id"],
        "opened_at": run["opened_at"],
        "next_action": next_action(run),
        "downstream_state": run["downstream"]["state"],
        "clarification_state": run["clarification"]["state"],
        "offer_mode": run["clarification"]["offer_mode"],
        "recency_as_of": run["clarification"]["recency_as_of"],
        "end_report_state": run["end_report"]["state"],
        "talk_count": len(run["talks"]),
    }


# ── Commands ──────────────────────────────────────────────────────────


def _apply_recency(
    clarification: dict[str, Any], talks: list[dict[str, Any]], stamp: str
) -> None:
    mode = offer_mode_for(talks)
    clarification["offer_mode"] = mode
    clarification["recency_as_of"] = stamp
    clarification["state"] = (
        STATE_NOT_APPLICABLE if mode == OFFER_MODE_NONE else STATE_OWED
    )


def _require_downstream(run: dict[str, Any], command: str) -> None:
    if run["downstream"]["state"] != DOWNSTREAM_COMPLETED:
        raise RunObligationsError(
            f"run {run['run_id']!r} has not recorded its downstream steps "
            f"(rendering, summary, profile, goals); {command} follows "
            "record-downstream, which follows Step 8",
            reason_code="invalid_transition",
        )


def _conflict(run: dict[str, Any], command: str, detail: str) -> RunObligationsError:
    return RunObligationsError(
        f"run {run['run_id']!r} already recorded a different {command}: {detail}; "
        "a retry must repeat the recorded inputs, and a changed answer is a "
        "new transition, not a replay",
        reason_code="invalid_transition",
    )


def command_open(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    now = parse_timestamp(args.now, "--now")
    stamp = render_timestamp(now)
    run_id = require_run_id(args.run_id)
    source_run = require_run_id(args.from_run) if args.from_run else None
    filenames = list(
        dict.fromkeys(list(args.talk) + _lines_from(args.talks_from, "--talks-from"))
    )
    if not filenames:
        raise RunObligationsError(
            "open requires at least one talk, via --talk or a --talks-from file "
            "with one filename per line",
            reason_code="invalid_arguments",
        )
    context.refresh()
    snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    require_adopted(ledger, context.ledger_path)
    try:
        run = find_run(ledger, run_id)
    except RunObligationsError:
        run = new_run(run_id, stamp)
        ledger["runs"].append(run)
    before = json.dumps(run, sort_keys=True)
    clarification = run["clarification"]
    known = {str(talk["filename"]): talk for talk in run["talks"]}
    # A talk joins the run as a new persisted fact when its filename is
    # unrecorded, or when a newer claim persisted it again under the run the
    # record links it to; a known talk otherwise keeps its recorded link.
    joined: dict[str, dict[str, Any]] = {}
    for name in filenames:
        fresh = describe_talk(context, name, now, run_id, source_run=source_run)
        if name not in known:
            joined[name] = fresh
            continue
        recorded = known[name]
        newer = newest_claim(
            persisted_claims(context.talk(name)), str(recorded["claim_run_id"])
        )
        if newer is not None and newer != _link(recorded):
            joined[name] = describe_talk(context, name, now, run_id, keep=newer)
    if run["completed_at"] is not None:
        if joined:
            raise RunObligationsError(
                f"run {run_id!r} completed at {run['completed_at']}; open a new "
                f"run id for {', '.join(sorted(joined))}",
                reason_code="invalid_transition",
            )
        # An exact replay of facts a completed run already recorded.
        return {
            "ok": True,
            "ledger_path": str(context.ledger_path),
            **store_ledger(context.ledger_path, snapshot, ledger),
            "replayed": True,
            "run": run,
        }
    state = clarification["state"]
    if joined and state in DISPOSITIONS:
        raise RunObligationsError(
            f"run {run_id!r} clarification is already {state!r}; a talk joining "
            f"now would never be offered — open {', '.join(sorted(joined))} "
            "under a fresh run id",
            reason_code="invalid_transition",
        )
    if joined and state == STATE_OFFERED:
        # The standing offer covers less than the run now does: withdraw it
        # so Step 9 makes it again with the fuller topics and recency.
        clarification.update({"state": STATE_OWED, "offered_at": None, "topics": []})
    recency_open = clarification["state"] in {STATE_OWED, STATE_NOT_APPLICABLE}
    talks: list[dict[str, Any]] = []
    for name, talk in known.items():
        if name in joined:
            continue
        if recency_open:
            # Recency follows the newest ``--now`` until the offer is made;
            # the recorded claim link is preserved through the refresh.
            talks.append(describe_talk(context, name, now, run_id, keep=_link(talk)))
        else:
            talks.append(talk)
    talks.extend(joined.values())
    talks.sort(key=lambda talk: str(talk["filename"]))
    run["talks"] = talks
    if recency_open:
        _apply_recency(clarification, talks, stamp)
    if joined:
        run["downstream"] = {"state": DOWNSTREAM_OWED, "completed_at": None}
    if json.dumps(run, sort_keys=True) != before:
        run["updated_at"] = stamp
    outcome = store_ledger(context.ledger_path, snapshot, ledger)
    return {"ok": True, "ledger_path": str(context.ledger_path), **outcome, "run": run}


def _transition(
    context: Context,
    args: argparse.Namespace,
    mutate: Any,
) -> dict[str, Any]:
    now = parse_timestamp(args.now, "--now")
    stamp = render_timestamp(now)
    run_id = require_run_id(args.run_id)
    context.refresh()
    snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    require_adopted(ledger, context.ledger_path)
    run = find_run(ledger, run_id)
    before = json.dumps(run, sort_keys=True)
    extras = mutate(run, stamp, now) or {}
    if json.dumps(run, sort_keys=True) != before:
        run["updated_at"] = stamp
    outcome = store_ledger(context.ledger_path, snapshot, ledger)
    return {
        "ok": True,
        "ledger_path": str(context.ledger_path),
        **outcome,
        **extras,
        "run": run,
    }


def command_record_downstream(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    def mutate(run: dict[str, Any], stamp: str, _now: datetime) -> dict[str, Any]:
        # Replay-safe: recording twice after a crash is the same fact twice.
        if run["downstream"]["state"] == DOWNSTREAM_COMPLETED:
            return {"replayed": True}
        run["downstream"] = {"state": DOWNSTREAM_COMPLETED, "completed_at": stamp}
        return {}

    return _transition(context, args, mutate)


def _lines_from(path_text: str | None, option: str) -> list[str]:
    """Free text arrives through a file, one entry per line, never a shell."""
    if not path_text:
        return []
    try:
        content = Path(path_text).read_text(encoding="utf-8")
    except (OSError, ValueError) as exc:
        raise RunObligationsError(
            f"cannot read {option} {path_text!r}: {exc}",
            reason_code="invalid_arguments",
        ) from exc
    return [line.strip() for line in content.splitlines() if line.strip()]


def command_record_offer(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    topics = [topic.strip() for topic in (args.topic or []) if topic.strip()]
    topics.extend(_lines_from(args.topics_from, "--topics-from"))
    topics = list(dict.fromkeys(topics))

    def mutate(run: dict[str, Any], stamp: str, now: datetime) -> dict[str, Any]:
        _require_downstream(run, "record-offer")
        clarification = run["clarification"]
        state = clarification["state"]
        if state == STATE_OFFERED:
            # Replay of a recorded offer: the same topics are the same fact.
            if clarification["topics"] != topics:
                raise _conflict(
                    run, "offer", f"topics {clarification['topics']!r} stand"
                )
            return {"offered": True, "replayed": True}
        if state == STATE_NOT_APPLICABLE:
            return {
                "offered": False,
                "replayed": True,
                "reason": f"run {run['run_id']!r} has nothing to clarify",
            }
        if state != STATE_OWED:
            raise RunObligationsError(
                f"run {run['run_id']!r} clarification is {state!r}; "
                "record-offer applies only while the offer is owed",
                reason_code="invalid_transition",
            )
        # The stored recency is a snapshot from the last ``open``. The offer is
        # being made now, against the current database, so refresh before
        # freezing: a run resumed weeks later must not promise an inline
        # session for a talk that is no longer same-week.
        talks = [
            describe_talk(
                context, str(talk["filename"]), now, run["run_id"], keep=_link(talk)
            )
            for talk in run["talks"]
        ]
        run["talks"] = talks
        _apply_recency(clarification, talks, stamp)
        if clarification["state"] == STATE_NOT_APPLICABLE:
            # Persisted as not applicable: the database changed since the run
            # opened and no analyzed talk remains, so nothing is asked and the
            # report is no longer blocked on an offer.
            return {
                "offered": False,
                "reason": (
                    f"run {run['run_id']!r} no longer has an analyzed talk to "
                    "clarify; recorded as not applicable, nothing to ask"
                ),
            }
        clarification["state"] = STATE_OFFERED
        clarification["offered_at"] = stamp
        clarification["topics"] = topics
        return {"offered": True}

    return _transition(context, args, mutate)


def command_record_disposition(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    disposition = args.disposition
    return_condition = (
        " ".join(
            [(args.return_condition or "").strip()]
            + _lines_from(args.return_condition_from, "--return-condition-from")
        ).strip()
        or None
    )
    if disposition == STATE_DEFERRED and return_condition is None:
        raise RunObligationsError(
            "a deferred offer needs --return-condition or a "
            "--return-condition-from file naming, in the speaker's words, when "
            "it is raised again",
            reason_code="invalid_arguments",
        )

    def mutate(run: dict[str, Any], stamp: str, _now: datetime) -> dict[str, Any]:
        clarification = run["clarification"]
        state = clarification["state"]
        if state == disposition and (
            disposition != STATE_DEFERRED
            or clarification["return_condition"] == return_condition
        ):
            # Replay of the recorded answer: nothing moves, the session an
            # accepted answer opened is left exactly as it is.
            return {"replayed": True}
        if state not in {STATE_OFFERED, STATE_DEFERRED}:
            if state in DISPOSITIONS:
                raise _conflict(run, "disposition", f"{state!r} stands")
            raise RunObligationsError(
                f"run {run['run_id']!r} clarification is {state!r}; a "
                "disposition needs a recorded offer first",
                reason_code="invalid_transition",
            )
        clarification["state"] = disposition
        clarification["resolved_at"] = stamp
        clarification["return_condition"] = (
            return_condition if disposition == STATE_DEFERRED else None
        )
        clarification["session"] = (
            {
                "state": SESSION_PENDING,
                "completed_at": None,
                "profile_inputs": None,
                "profile_refreshed": None,
                "claim_token": None,
                "claimed_at": None,
            }
            if disposition == STATE_ACCEPTED
            else None
        )
        return {}

    return _transition(context, args, mutate)


def _accepted_session(run: dict[str, Any]) -> dict[str, Any]:
    clarification = run["clarification"]
    session = clarification["session"]
    if clarification["state"] != STATE_ACCEPTED or session is None:
        raise RunObligationsError(
            f"run {run['run_id']!r} has no accepted clarification session",
            reason_code="invalid_transition",
        )
    return session


def _require_claim(run: dict[str, Any], session: dict[str, Any], token: str) -> None:
    """The session is running under exactly this claim."""
    if session["state"] != SESSION_RUNNING:
        raise RunObligationsError(
            f"run {run['run_id']!r} session is {session['state']!r}; claim it "
            "with claim-session first",
            reason_code="invalid_transition",
        )
    if token != session["claim_token"]:
        raise RunObligationsError(
            f"run {run['run_id']!r} session is running under another claim "
            f"since {session['claimed_at']}; only that claim's holder records "
            "it. If that host died, release the claim with release-session "
            "(an operator decision) and claim it again",
            reason_code="session_claimed",
        )


def command_claim_session(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    """Reserve the accepted session for one host; the token names the claim."""

    def mutate(run: dict[str, Any], stamp: str, _now: datetime) -> dict[str, Any]:
        session = _accepted_session(run)
        state = session["state"]
        if state == SESSION_RUNNING:
            if (
                args.claim_token is not None
                and args.claim_token == (session["claim_token"])
            ):
                return {"claim_token": session["claim_token"], "replayed": True}
            raise RunObligationsError(
                f"run {run['run_id']!r} session is already claimed since "
                f"{session['claimed_at']}. If its host died, release the claim "
                "with release-session (an operator decision), then claim again",
                reason_code="session_claimed",
            )
        if state != SESSION_PENDING:
            raise RunObligationsError(
                f"run {run['run_id']!r} session is {state!r}; only a pending "
                "session can be claimed",
                reason_code="invalid_transition",
            )
        session["state"] = SESSION_RUNNING
        session["claim_token"] = secrets.token_hex(CLAIM_TOKEN_BYTES)
        session["claimed_at"] = stamp
        return {"claim_token": session["claim_token"]}

    return _transition(context, args, mutate)


def command_release_session(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    """Free one named claim: the operator's decision that its host died.

    The claim to free is named by its token, so a retry after a lost response
    can never free a claim another host took in between.
    """
    reason = " ".join(
        [(args.reason or "").strip()] + _lines_from(args.reason_from, "--reason-from")
    ).strip()
    if not reason:
        raise RunObligationsError(
            "release-session needs --reason or a --reason-from file saying why "
            "the claim's host will not finish the session",
            reason_code="invalid_arguments",
        )

    def mutate(run: dict[str, Any], _stamp: str, _now: datetime) -> dict[str, Any]:
        session = _accepted_session(run)
        if session["state"] == SESSION_PENDING:
            return {"released": None, "replayed": True}
        if session["state"] != SESSION_RUNNING:
            raise RunObligationsError(
                f"run {run['run_id']!r} session is {session['state']!r}; only a "
                "running session has a claim to release",
                reason_code="invalid_transition",
            )
        if args.claim_token != session["claim_token"]:
            raise RunObligationsError(
                f"run {run['run_id']!r} session is running under a different claim "
                f"since {session['claimed_at']}, not the one named; read its "
                "claim_token from status --run-id before releasing it",
                reason_code="session_claimed",
            )
        released = {
            "claim_token": session["claim_token"],
            "claimed_at": session["claimed_at"],
            "reason": reason,
        }
        session["state"] = SESSION_PENDING
        session["claim_token"] = None
        session["claimed_at"] = None
        return {"released": released}

    return _transition(context, args, mutate)


def _read_ledger_unwritten(
    context: Context,
) -> tuple[TrackingDatabaseSnapshot, dict[str, Any]]:
    """The current ledger, read strictly without writing; for use under its lock.

    ``load_ledger`` rewrites an older record it upgrades, which takes this same
    lock, so a caller already holding it reads here and refuses an older record.
    """
    try:
        snapshot = snapshot_tracking_database(context.ledger_path)
        payload = decode_json_object(snapshot)
    except TrackingDatabaseIOError as exc:
        raise _io_failure(
            exc, f"obligations ledger {context.ledger_path}", "ledger_unreadable"
        ) from exc
    runs = payload.get("runs") if isinstance(payload, dict) else None
    if isinstance(runs, list) and any(
        isinstance(run, dict) and run.get("schema_version") != RUN_RECORD_SCHEMA_VERSION
        for run in runs
    ):
        raise RunObligationsError(
            f"obligations ledger {context.ledger_path} gained an older record "
            "while this command ran; re-run it",
            reason_code="ledger_changed",
        )
    return snapshot, validate_ledger(payload, context.ledger_path, context.vault_root)


def _count_in_database(
    context: Context, run_id: str, *, advances: bool
) -> dict[str, Any]:
    """Advance the counter and record the run, against one database generation."""
    try:
        snapshot = snapshot_tracking_database(context.database_path)
        database = decode_json_object(snapshot)
        require_current_tracking_database(database)
    except TrackingDatabaseIOError as exc:
        raise _io_failure(
            exc, f"tracking database {context.database_path}", "database_unusable"
        ) from exc
    except TrackingDatabaseError as exc:
        raise RunObligationsError(str(exc), reason_code="database_unusable") from exc
    config = database["config"]
    counted = counted_clarification_runs(config)
    sessions = config.get(CLARIFICATION_SESSIONS_FIELD, 0)
    if run_id in counted:
        return {
            "counted": False,
            "replayed": True,
            CLARIFICATION_SESSIONS_FIELD: sessions,
        }
    candidate = json.loads(json.dumps(database))
    after = sessions + 1 if advances else sessions
    candidate["config"][CLARIFICATION_SESSIONS_FIELD] = after
    candidate["config"][CLARIFICATION_COUNTED_RUNS_FIELD] = [*counted, run_id]
    try:
        require_current_tracking_database(candidate)
        result = commit_tracking_database(snapshot, render_json_object(candidate))
    except TrackingDatabaseIOError as exc:
        raise _io_failure(
            exc, f"tracking database {context.database_path}", "database_write_failed"
        ) from exc
    except TrackingDatabaseError as exc:
        raise RunObligationsError(str(exc), reason_code="database_unusable") from exc
    for warning in result.warnings:
        print(f"WARNING: tracking database: {warning}", file=sys.stderr)
    return {
        "counted": True,
        "advanced": advances,
        CLARIFICATION_SESSIONS_FIELD: after,
        "durability_state": result.durability_state,
        "warnings": list(result.warnings),
    }


def command_count_session(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    """Count the claimed session once, holding the ledger's lock through the write.

    The claim check and the database commit happen under the ledger's own
    writer lock, so no release or record can land between them. Lock order is
    ledger, then database; nothing takes them the other way. The ledger itself
    is not written.
    """
    run_id = require_run_id(args.run_id)
    # Any older record is upgraded here, by the owner's read that rewrites it.
    load_ledger(context.ledger_path, context.vault_root)
    try:
        with exclusive_file_lock(context.ledger_path, label=DATABASE_LOCK_LABEL):
            _snapshot, ledger = _read_ledger_unwritten(context)
            require_adopted(ledger, context.ledger_path)
            run = find_run(ledger, run_id)
            session = _accepted_session(run)
            _require_claim(run, session, args.claim_token)
            legacy = session.get(LEGACY_COUNT_FIELD)
            if legacy == LEGACY_COUNT_UNKNOWN:
                raise RunObligationsError(
                    f"run {run_id!r} session began under run-record schema 2, "
                    "whose Step 9 counted before recording, so it may be "
                    "counted already; record which with reconcile-count, then "
                    "count it",
                    reason_code="session_count_unknown",
                )
            outcome = _count_in_database(
                context, run_id, advances=legacy != LEGACY_COUNT_COUNTED
            )
    except CooperativeLockError as exc:
        raise RunObligationsError(
            f"cannot hold the obligations ledger lock: {exc}; re-run",
            reason_code="ledger_unreadable",
        ) from exc
    return {"ok": True, "run_id": run_id, **outcome}


def command_reconcile_count(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    """Record whether a session upgraded from schema 2 was already counted."""
    answer = LEGACY_COUNT_COUNTED if args.counted == "yes" else LEGACY_COUNT_UNCOUNTED

    def mutate(run: dict[str, Any], _stamp: str, _now: datetime) -> dict[str, Any]:
        session = _accepted_session(run)
        _require_claim(run, session, args.claim_token)
        state = session.get(LEGACY_COUNT_FIELD)
        if state == answer:
            return {"replayed": True}
        if state is None:
            raise RunObligationsError(
                f"run {run['run_id']!r} session began under schema 3, so its count "
                "is never in doubt; count it with count-session",
                reason_code="invalid_transition",
            )
        if state != LEGACY_COUNT_UNKNOWN:
            raise _conflict(run, "legacy count", f"{LEGACY_COUNT_FIELD}={state!r}")
        session[LEGACY_COUNT_FIELD] = answer
        return {}

    return _transition(context, args, mutate)


def command_record_session(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    refreshed = bool(args.profile_refreshed)
    inputs = args.profile_inputs

    def mutate(run: dict[str, Any], stamp: str, _now: datetime) -> dict[str, Any]:
        session = _accepted_session(run)
        if session["state"] == SESSION_COMPLETED:
            # A session completed before claims existed migrated with a null
            # token; any claim token replays it, since none was ever held.
            same_claim = session["claim_token"] in (None, args.claim_token)
            if (
                session["profile_inputs"] == inputs
                and session["profile_refreshed"] == refreshed
                and same_claim
            ):
                return {"replayed": True}
            detail = (
                f"profile_inputs={session['profile_inputs']!r}, "
                f"profile_refreshed={session['profile_refreshed']!r} at "
                f"{session['completed_at']}"
            )
            if not same_claim:
                detail += " under another claim"
            raise _conflict(run, "session completion", detail)
        _require_claim(run, session, args.claim_token)
        if run["run_id"] not in counted_clarification_runs(context.database["config"]):
            raise RunObligationsError(
                f"run {run['run_id']!r} session has not been counted: run "
                "count-session for this run under its claim before recording "
                "the session",
                reason_code="session_not_counted",
            )
        if (
            inputs == PROFILE_INPUTS_CHANGED
            and not refreshed
            and context.profile_path.exists()
        ):
            raise RunObligationsError(
                f"the session changed profile inputs and {context.profile_path} "
                "exists; regenerate the profile (Step 7) first, then record the "
                "session with --profile-refreshed",
                reason_code="profile_refresh_required",
            )
        session["state"] = SESSION_COMPLETED
        session["completed_at"] = stamp
        session["profile_inputs"] = inputs
        session["profile_refreshed"] = refreshed
        report = run["end_report"]
        extras: dict[str, Any] = {}
        if inputs == PROFILE_INPUTS_CHANGED and report["state"] == REPORT_DELIVERED:
            # A session accepted after the report went out (a deferred offer
            # raised again) changed what the report described; the run owes a
            # fresh report that carries the refreshed profile.
            run["end_report"] = {
                "state": REPORT_OWED,
                "delivered_at": None,
                "report_path": None,
                "report_sha256": None,
                "reopened_at": stamp,
            }
            run["completed_at"] = None
            extras["report_reopened"] = True
        return extras

    return _transition(context, args, mutate)


def command_record_report(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    report_file = Path(args.report_file)
    try:
        content = _read_regular_file(
            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | os.O_NONBLOCK,
            str(report_file),
            dir_fd=None,
        )
    except (OSError, ValueError, RunObligationsError) as exc:
        raise RunObligationsError(
            f"cannot read the delivered report {args.report_file!r} as a regular "
            f"file: {exc}",
            reason_code="report_unreadable",
        ) from exc
    if content is None:
        raise RunObligationsError(
            f"delivered report {args.report_file!r} does not exist",
            reason_code="report_unreadable",
        )
    if not content.strip():
        raise RunObligationsError(
            f"delivered report {report_file} is empty; the end report is the "
            "speaker-facing text, not a placeholder",
            reason_code="report_empty",
        )
    digest = hashlib.sha256(content).hexdigest()

    def mutate(run: dict[str, Any], stamp: str, _now: datetime) -> dict[str, Any]:
        _require_downstream(run, "record-report")
        if not clarification_resolved(run):
            raise RunObligationsError(
                f"run {run['run_id']!r} still owes its clarification "
                f"disposition ({next_action(run)}); the end report is delivered "
                "after the speaker answers, so the report can carry any profile "
                "refresh the answers caused",
                reason_code="invalid_transition",
            )
        report = run["end_report"]
        # The copy is installed before the ledger commit and kept whatever the
        # commit does: content-addressed, it is shared by every delivery of
        # these bytes, so a copy left unbound by a lost generation race is
        # reused by the retry rather than removed from under a winner. A replay
        # of the recorded digest still goes through here, so a copy deleted
        # after delivery is recreated rather than reported as present.
        write_report_copy(context.reports_directory, run["run_id"], digest, content)
        bound = bound_report_path(run["run_id"], digest)
        if report["state"] == REPORT_DELIVERED and report["report_sha256"] == digest:
            report["report_path"] = bound
            return {"replayed": True}
        report["state"] = REPORT_DELIVERED
        report["delivered_at"] = stamp
        report["report_path"] = bound
        report["report_sha256"] = digest
        run["completed_at"] = stamp
        return {}

    return _transition(context, args, mutate)


def _excluded_facts(ledger: dict[str, Any]) -> set[tuple[Any, ...]]:
    """Facts adoption recorded as history plus facts each dismissal covered."""
    excluded = {tuple(identity) for identity in ledger["adopted_facts"]}
    for entry in ledger["dismissed_runs"]:
        excluded.update(tuple(identity) for identity in entry["facts"])
    return excluded


def uncovered_facts(
    persisted: dict[str, dict[str, Any]], ledger: dict[str, Any]
) -> dict[str, dict[str, list[dict[str, Any]]]]:
    """Persisted facts no record covers and no boundary or dismissal excludes.

    A fact is covered when a run record lists the talk with exactly that
    claim, or with a strictly newer claim of the same run (a later generation
    or release) — a run that merged a talk again superseded its earlier
    result — whichever run id recorded it.
    """
    excluded = _excluded_facts(ledger)
    exact: set[tuple[Any, ...]] = set()
    newest: dict[tuple[str, str], tuple[int, str]] = {}
    for run in ledger["runs"]:
        for talk in run["talks"]:
            link = _link(talk)
            name = str(talk["filename"])
            exact.add(tuple(fact_identity(link, name)))
            key = (link["run_id"], name)
            order = _fact_order(link)
            if key not in newest or order > newest[key]:
                newest[key] = order
    result: dict[str, dict[str, list[dict[str, Any]]]] = {}
    for run_id, entry in persisted.items():
        for name, facts in entry["talks"].items():
            open_facts = [
                fact
                for fact in facts
                if tuple(fact_identity(fact, name)) not in excluded
                and tuple(fact_identity(fact, name)) not in exact
                and not (
                    (run_id, name) in newest
                    and _fact_order(fact) < newest[(run_id, name)]
                )
            ]
            if open_facts:
                result.setdefault(run_id, {})[name] = open_facts
    return result


def open_required(
    persisted: dict[str, dict[str, Any]], ledger: dict[str, Any]
) -> list[dict[str, Any]]:
    """Persisted facts the ledger neither covers nor excludes, per run.

    A persisted fact is one closed claim: run id, filename, batch id,
    reprocess generation, and release time. Facts recorded at adoption are
    history and never listed; facts a dismissal covered are left out. Every
    other uncovered fact (see ``uncovered_facts``) lists its run, in run-id
    order, with exactly the talks concerned: a recorded run as
    ``missing_talks``, or ``talks_persisted_after_completion`` when its report
    is already delivered and ``talks_persisted_after_answer`` when its offer
    was answered — in both of those no fact joins the run, so they need a
    fresh run id — and a run with no record as ``unrecorded_run``. No later
    run's existence and no timestamp stands in for coverage.
    """
    records = {run["run_id"]: run for run in ledger["runs"]}
    required: list[dict[str, Any]] = []
    for run_id, talks in sorted(uncovered_facts(persisted, ledger).items()):
        run = records.get(run_id)
        if run is None:
            reason = REASON_UNRECORDED
        elif run["completed_at"] is not None:
            reason = REASON_AFTER_COMPLETION
        elif run["clarification"]["state"] in DISPOSITIONS:
            reason = REASON_AFTER_ANSWER
        else:
            reason = REASON_MISSING_TALKS
        required.append(
            {
                "run_id": run_id,
                "talks": sorted(talks),
                "latest_released_at": max(
                    fact["released_at"] for facts in talks.values() for fact in facts
                ),
                "reason": reason,
                "next_action": NEXT_OPEN,
            }
        )
    return required


def command_adopt(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    """Create the ledger and stamp the reconciliation boundary; replay-safe."""
    now = parse_timestamp(args.now, "--now")
    stamp = render_timestamp(now)
    context.refresh()
    snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    if ledger.get("adopted_at") is None:
        history = sorted(
            fact_identity(fact, name)
            for entry in context.persisted_runs().values()
            for name, facts in entry["talks"].items()
            for fact in facts
        )
        ledger = empty_ledger(stamp, history)
        outcome = store_ledger(context.ledger_path, snapshot, ledger)
        return {
            "ok": True,
            "ledger_path": str(context.ledger_path),
            **outcome,
            "adopted_at": stamp,
            "adopted_fact_count": len(history),
        }
    return {
        "ok": True,
        "ledger_path": str(context.ledger_path),
        "written": False,
        "durability_state": "unchanged",
        "warnings": [],
        "replayed": True,
        "adopted_at": ledger["adopted_at"],
    }


def command_dismiss(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    """Record that an uncovered run is deliberately not being opened."""
    now = parse_timestamp(args.now, "--now")
    stamp = render_timestamp(now)
    run_id = require_run_id(args.run_id)
    reason = " ".join(
        [(args.reason or "").strip()] + _lines_from(args.reason_from, "--reason-from")
    ).strip()
    if not reason:
        raise RunObligationsError(
            "dismiss needs --reason or a --reason-from file saying why this "
            "run's persisted talks are not being opened",
            reason_code="invalid_arguments",
        )
    context.refresh()
    snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    require_adopted(ledger, context.ledger_path)
    if any(run["run_id"] == run_id for run in ledger["runs"]):
        raise RunObligationsError(
            f"run {run_id!r} has an obligations record; it completes through "
            "its obligations, not through dismissal",
            reason_code="invalid_transition",
        )
    uncovered = uncovered_facts(context.persisted_runs(), ledger).get(run_id, {})
    covering = sorted(
        fact_identity(fact, name) for name, facts in uncovered.items() for fact in facts
    )
    existing = next(
        (entry for entry in ledger["dismissed_runs"] if entry["run_id"] == run_id),
        None,
    )
    if covering:
        # Listed now means facts no earlier dismissal covered exist; today's
        # dismissal names exactly those, renewing an older entry in place.
        if existing is None:
            entry = {
                "run_id": run_id,
                "dismissed_at": stamp,
                "reason": reason,
                "facts": covering,
            }
            ledger["dismissed_runs"].append(entry)
            extras: dict[str, Any] = {}
        else:
            known = {tuple(identity) for identity in existing["facts"]}
            existing["facts"] = sorted(
                [list(identity) for identity in known]
                + [identity for identity in covering if tuple(identity) not in known]
            )
            existing.update({"dismissed_at": stamp, "reason": reason})
            entry = existing
            extras = {"renewed": True}
        outcome = store_ledger(context.ledger_path, snapshot, ledger)
        return {
            "ok": True,
            "ledger_path": str(context.ledger_path),
            **outcome,
            **extras,
            "dismissed": entry,
        }
    if existing is None:
        raise RunObligationsError(
            f"run {run_id!r} is not an uncovered persisted run in `pending`; "
            "only a listed unrecorded_run can be dismissed",
            reason_code="invalid_transition",
        )
    if existing["reason"] == reason:
        return {
            "ok": True,
            "ledger_path": str(context.ledger_path),
            "written": False,
            "durability_state": "unchanged",
            "warnings": [],
            "replayed": True,
            "dismissed": existing,
        }
    raise RunObligationsError(
        f"run {run_id!r} was already dismissed at {existing['dismissed_at']} "
        f"for {existing['reason']!r} and nothing new is listed; a retry "
        "repeats that reason",
        reason_code="invalid_transition",
    )


def command_pending(context: Context, _args: argparse.Namespace) -> dict[str, Any]:
    _snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    adopted = isinstance(ledger.get("adopted_at"), str)
    pending = [
        summarize(run) for run in ledger["runs"] if next_action(run) != NEXT_NONE
    ]
    deferred = [
        {
            "run_id": run["run_id"],
            "return_condition": run["clarification"]["return_condition"],
            "topics": run["clarification"]["topics"],
            "resolved_at": run["clarification"]["resolved_at"],
        }
        for run in ledger["runs"]
        if run["clarification"]["state"] == STATE_DEFERRED
    ]
    return {
        "ok": True,
        "ledger_path": str(context.ledger_path),
        "ledger_present": context.ledger_path.exists(),
        "adopt_required": not adopted,
        "adopted_at": ledger.get("adopted_at"),
        "pending": pending,
        "count": len(pending),
        "deferred_offers": deferred,
        "open_required": (
            open_required(context.persisted_runs(), ledger) if adopted else []
        ),
    }


def command_status(context: Context, args: argparse.Namespace) -> dict[str, Any]:
    _snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    require_adopted(ledger, context.ledger_path)
    run = find_run(ledger, require_run_id(args.run_id))
    return {
        "ok": True,
        "ledger_path": str(context.ledger_path),
        "summary": summarize(run),
        "run": run,
    }


# ── CLI ───────────────────────────────────────────────────────────────


def pending_sessions(
    ledger: dict[str, Any], state: str = SESSION_PENDING
) -> list[dict[str, Any]]:
    """Unfinished accepted sessions in ``state``, earliest ``opened_at`` first.

    A running session is claimed by a host and is never selected for another;
    ``state=SESSION_RUNNING`` lists those. Stored timestamps are canonical, so
    text order is time order; two runs opened in the same second keep the
    ledger's opening order.
    """
    candidates = [
        (run["opened_at"], index, run)
        for index, run in enumerate(ledger["runs"])
        if next_action(run) == NEXT_SESSION
        and run["clarification"]["session"]["state"] == state
    ]
    return [run for _stamp, _index, run in sorted(candidates, key=lambda c: c[:2])]


def _session_view(run: dict[str, Any]) -> dict[str, Any]:
    clarification = run["clarification"]
    return {
        "run_id": run["run_id"],
        "opened_at": run["opened_at"],
        "offered_at": clarification["offered_at"],
        "resolved_at": clarification["resolved_at"],
        "offer_mode": clarification["offer_mode"],
        "topics": list(clarification["topics"]),
        "summary": summarize(run),
    }


def _session_recovery(action: str) -> str:
    """What to do when a named run has no pending session, by its next action."""
    if action in (NEXT_REPORT, NEXT_NONE):
        return (
            "its session is already recorded; run `session-agenda` without "
            "--run-id to select a session that is still pending"
        )
    return (
        f"resume vault-ingress at the step `{action}` names (see `status "
        "--run-id`) so the offer reaches `accepted`, or run `session-agenda` "
        "without --run-id to select a session that is already pending"
    )


def command_session_agenda(
    context: Context, args: argparse.Namespace
) -> dict[str, Any]:
    """The clarification session to run, with its seed agenda; read-only.

    Without ``--run-id`` the first of ``pending_sessions`` is selected, or
    ``session`` is null when none is pending. With ``--run-id`` the named run
    must exist and its session must be pending; a running one is refused as
    ``session_claimed`` and any other state as ``invalid_transition`` naming
    what stands. Running sessions are listed, never selected.
    """
    _snapshot, ledger = load_ledger(context.ledger_path, context.vault_root)
    adopted = isinstance(ledger.get("adopted_at"), str)
    sessions = pending_sessions(ledger)
    running = pending_sessions(ledger, SESSION_RUNNING)
    if args.run_id is None:
        selected = sessions[0] if sessions else None
    else:
        require_adopted(ledger, context.ledger_path)
        selected = find_run(ledger, require_run_id(args.run_id))
        action = next_action(selected)
        if selected in running:
            raise RunObligationsError(
                f"run {selected['run_id']!r} session is running under a claim "
                f"since {selected['clarification']['session']['claimed_at']}; "
                "another host is conducting it. If that host died, release the "
                "claim with release-session (an operator decision)",
                reason_code="session_claimed",
            )
        if action != NEXT_SESSION:
            raise RunObligationsError(
                f"run {selected['run_id']!r} has no pending clarification "
                f"session: its clarification is "
                f"{selected['clarification']['state']!r} and its next action is "
                f"{action!r}; {_session_recovery(action)}",
                reason_code="invalid_transition",
            )
    return {
        "ok": True,
        "ledger_path": str(context.ledger_path),
        "ledger_present": context.ledger_path.exists(),
        "adopt_required": not adopted,
        "adopted_at": ledger.get("adopted_at"),
        "session": _session_view(selected) if selected is not None else None,
        "pending_sessions": [run["run_id"] for run in sessions],
        "running_sessions": [
            {
                "run_id": run["run_id"],
                "claimed_at": run["clarification"]["session"]["claimed_at"],
            }
            for run in running
        ],
    }


def build_parser() -> JsonArgumentParser:
    parser = JsonArgumentParser(description=(__doc__ or "").split("\n")[0])
    parser.add_argument("database", help="tracking-database.json path")
    actions = parser.add_subparsers(
        dest="action", required=True, parser_class=JsonArgumentParser
    )

    def with_run(sub: argparse.ArgumentParser, *, now: bool = True) -> None:
        sub.add_argument("--run-id", required=True)
        if now:
            sub.add_argument(
                "--now", required=True, help="timezone-aware ISO-8601 event time"
            )

    adopt = actions.add_parser(
        "adopt", help="create the ledger and stamp the reconciliation boundary"
    )
    adopt.add_argument("--now", required=True, help="timezone-aware ISO-8601 time")

    opened = actions.add_parser(
        "open", help="record the run's obligations for a persisted batch"
    )
    with_run(opened)
    opened.add_argument(
        "--talk",
        action="append",
        default=[],
        help="filename persist-results.py merged; repeat per talk",
    )
    opened.add_argument(
        "--talks-from",
        help="a file naming persisted talks, one filename per line",
    )
    opened.add_argument(
        "--from-run",
        help="recovery: the run whose closed claim persisted these talks",
    )

    downstream = actions.add_parser(
        "record-downstream",
        help="rendering, summary, profile, and goal steps completed for the run",
    )
    with_run(downstream)

    offer = actions.add_parser("record-offer", help="the offer was put to the speaker")
    with_run(offer)
    offer.add_argument(
        "--topic", action="append", help="candidate clarification topic; repeat"
    )
    offer.add_argument(
        "--topics-from", help="a file naming candidate topics, one per line"
    )

    disposition = actions.add_parser(
        "record-disposition", help="the speaker's explicit answer to the offer"
    )
    with_run(disposition)
    disposition.add_argument("--disposition", choices=DISPOSITIONS, required=True)
    disposition.add_argument(
        "--return-condition", help="when a deferred offer is raised again"
    )
    disposition.add_argument(
        "--return-condition-from",
        help="a file holding the speaker's return condition in their words",
    )

    claim = actions.add_parser(
        "claim-session", help="reserve the accepted session for this host"
    )
    with_run(claim)
    claim.add_argument(
        "--claim-token", help="this host's token, to replay a claim it holds"
    )

    release = actions.add_parser(
        "release-session", help="free a claim whose host will not finish the session"
    )
    with_run(release)
    release.add_argument(
        "--claim-token", required=True, help="the claim to free, from status"
    )
    release.add_argument("--reason", help="why, in plain words")
    release.add_argument(
        "--reason-from", help="a file holding the reason in the operator's words"
    )

    count = actions.add_parser(
        "count-session",
        help="count the claimed session toward clarification_sessions_completed",
    )
    count.add_argument("--run-id", required=True)
    count.add_argument("--claim-token", required=True)

    reconcile = actions.add_parser(
        "reconcile-count",
        help="record whether a session begun under schema 2 was already counted",
    )
    with_run(reconcile)
    reconcile.add_argument("--claim-token", required=True)
    reconcile.add_argument(
        "--counted",
        choices=("yes", "no"),
        required=True,
        help="whether clarification_sessions_completed already includes it",
    )

    session = actions.add_parser(
        "record-session", help="the accepted clarification session completed"
    )
    with_run(session)
    session.add_argument("--claim-token", required=True)
    session.add_argument(
        "--profile-inputs",
        choices=PROFILE_INPUTS,
        required=True,
        help=(
            "whether the session changed confirmed intents, config fields, goals,"
            " or the summary"
        ),
    )
    session.add_argument(
        "--profile-refreshed",
        action="store_true",
        help="the speaker profile was regenerated from the session's answers",
    )

    report = actions.add_parser(
        "record-report", help="the end report was delivered to the speaker"
    )
    with_run(report)
    report.add_argument(
        "--report-file", required=True, help="the delivered report text"
    )

    dismiss = actions.add_parser(
        "dismiss", help="an uncovered run is deliberately not being opened"
    )
    with_run(dismiss)
    dismiss.add_argument("--reason", help="why, in plain words")
    dismiss.add_argument(
        "--reason-from", help="a file holding the speaker's reason in their words"
    )

    actions.add_parser("pending", help="list runs with unresolved obligations")

    status = actions.add_parser("status", help="show one run's obligations")
    with_run(status, now=False)

    agenda = actions.add_parser(
        "session-agenda",
        help="the accepted clarification session to run, with its seed agenda",
    )
    agenda.add_argument("--run-id", help="that run's session, or a refusal")
    return parser


COMMANDS = {
    "adopt": command_adopt,
    "open": command_open,
    "record-downstream": command_record_downstream,
    "record-offer": command_record_offer,
    "record-disposition": command_record_disposition,
    "claim-session": command_claim_session,
    "release-session": command_release_session,
    "count-session": command_count_session,
    "reconcile-count": command_reconcile_count,
    "record-session": command_record_session,
    "record-report": command_record_report,
    "dismiss": command_dismiss,
    "pending": command_pending,
    "status": command_status,
    "session-agenda": command_session_agenda,
}


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    try:
        args = parser.parse_args(argv)
        context = Context(args.database)
        payload = COMMANDS[args.action](context, args)
    except (RunObligationsError, VaultRootAuthorityError) as exc:
        payload: dict[str, Any] = {"ok": False, "error": str(exc)}
        reason_code = getattr(exc, "reason_code", None)
        if isinstance(reason_code, str):
            payload["reason_code"] = reason_code
        print(str(exc), file=sys.stderr)
        print(json.dumps(payload, ensure_ascii=False))
        return 2
    print(json.dumps(payload, ensure_ascii=False))
    return 0


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