informed-decision/informed_decision/evidence.py

"""Commitment records and the in-memory outbox used by domain tests.

The persistent transactional implementation is ``store.Store``; this module
constructs the commitments used by both implementations.

Payload is **commitment-only**, granted for Stage 1 by `GH-DEC-2026-014`:
hashes, principal, timestamps, acks, the co-referenced approval id. Never the
brief, never the packet.

What that does and does not establish is not a detail — see
`docs/specs/EvidenceModel.md` §8d:

- It satisfies **non-alteration**. It does **not** satisfy
  **reconstructability**, and must never be described as doing so.
- It moves *integrity* out of our control and leaves *availability* entirely
  inside it. The party that can withhold the content is the party the evidence
  is about.

Which is why every record carries the §4 existence assertion.
"""

from __future__ import annotations

import json
import re
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum

from .disposition import Disposition
from .presentation import Presentation


class EventClass(str, Enum):
    """One source, distinct type values per class (`audit-core`, `AUDIT-IN-0003`)."""

    PRESENTATION = "informed-decision.presentation"
    DISPOSITION = "informed-decision.disposition"
    STANCE_APPLICATION = "informed-decision.stance_application"
    HEARTBEAT = "audit-core.heartbeat"


#: Heartbeat gap per class, seconds. Declared per class rather than per source:
#: a per-source heartbeat from a mixed-volume emitter is satisfied by its
#: chattiest class and says nothing about the quiet, security-relevant one.
HEARTBEAT_CLASSES: dict[str, int] = {
    EventClass.PRESENTATION.value: 86400,
    EventClass.DISPOSITION.value: 86400,
    EventClass.STANCE_APPLICATION.value: 86400,
}


class CustodyLocatorRejected(Exception):
    """The custody locator looked secret-shaped (PR-12).

    `audit-core` applies ``secret_policy: redact``, which scans ``data``. A
    credentialed URL is redacted out and the existence declaration arrives
    without its pointer — visibly (``details.redaction.paths`` records it), but
    the declaration is then useless while *looking* complete. Build-breaking,
    not a warning.
    """


#: Anything carrying userinfo, a query string, or a token-shaped segment.
_SECRET_SHAPED = re.compile(
    r"(://[^/@\s]*@)|([?&](token|key|secret|sig|password|access[_-]?token)=)|(\bBearer\b)",
    re.IGNORECASE,
)


def assert_custody_locator_safe(locator: str) -> None:
    if not locator:
        raise CustodyLocatorRejected("custody locator must be non-empty")
    if _SECRET_SHAPED.search(locator):
        raise CustodyLocatorRejected(
            f"custody locator {locator!r} is secret-shaped; use a stable "
            "identifier the custodian resolves, never a credentialed URL (PR-12)"
        )


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


@dataclass(frozen=True)
class Commitment:
    """One emitted record. `data` is stored verbatim by `audit-core` and chained."""

    id: str
    event_class: EventClass
    at: str
    data: dict

    def as_envelope(self) -> dict:
        return {"type": self.event_class.value, "data": self.data}


def _existence(custody: str) -> dict:
    """The `GH-DEC-2026-014` §4 assertion.

    A commitment with no assertion that something is being committed to is
    indistinguishable from a commitment to nothing. Failure to produce at
    retrieval is then a conformance failure attributable to the custodian —
    which is us.
    """
    assert_custody_locator_safe(custody)
    return {"content_exists": True, "custody": custody}


def commit_presentation(p: Presentation, *, custody: str) -> Commitment:
    data = {
        "memo_id": p.memo_id,
        "memo_version": p.memo_version,
        "presentation_id": p.id,
        "principal_sub": p.principal_sub,
        "locale": p.locale,
        "ui_release": p.ui_release,
        "rendered_at": p.rendered_at,
        "view_hash": p.view_hash,
        "awareness_hash": p.awareness_hash,
        "phase": p.phase.value,
        "acked_highlight_ids": sorted(p.acked_highlight_ids),
        "approval_id": p.approval_id,
    }
    # Claims travel with their route (PR-09/PR-11), never as bare strings.
    if p.tenant is not None:
        data["tenant"] = {"value": p.tenant.value, "route": p.tenant.route.value}
    if p.principal_type is not None:
        data["principal_type"] = {
            "value": p.principal_type.value,
            "route": p.principal_type.route.value,
        }
    data.update(_existence(custody))
    return Commitment(f"ev-{uuid.uuid4()}", EventClass.PRESENTATION, _now(), data)


def commit_disposition(d: Disposition, *, custody: str) -> Commitment:
    data = {
        "memo_id": d.memo_id,
        "memo_version": d.memo_version,
        "presentation_id": d.presentation_id,
        "disposition_id": d.id,
        "verb": d.verb.value,
        "actor_sub": d.actor.sub,
        "actor_kind": d.actor.kind.value,
        "at": d.at,
        "reasons": list(d.reasons),
        "approval_id": d.approval_id,
    }
    data.update(_existence(custody))
    return Commitment(f"ev-{uuid.uuid4()}", EventClass.DISPOSITION, _now(), data)


def commit_stance_application(
    *, memo_id: str, memo_version: int, binding_level: str | None,
    binding_level_state: str, stance: str, dependency: str, custody: str,
) -> Commitment:
    """A fail-closed outcome is recorded as a STANCE APPLICATION.

    Never as a decline: no disposition exists, because the human did not make
    one. Conflating them would put a refusal in the record that no person
    authored.
    """
    data = {
        "memo_id": memo_id,
        "memo_version": memo_version,
        "binding_level": binding_level,
        "binding_level_state": binding_level_state,
        "stance_applied": stance,
        "unreachable_dependency": dependency,
        # GH-DEC-2026-010: we can show a decision was obtained and what it said;
        # we cannot show access-engine said it. Pinned false until FLEX-WP-0024.
        "decision_attributable": False,
        "at": _now(),
    }
    data.update(_existence(custody))
    return Commitment(f"ev-{uuid.uuid4()}", EventClass.STANCE_APPLICATION, _now(), data)


def heartbeat(event_class: EventClass) -> Commitment:
    """An ORDINARY event — same envelope, same chain.

    Deliberately so: a heartbeat stored outside the chain would be the one
    record that could be back-dated.
    """
    return Commitment(
        f"ev-{uuid.uuid4()}",
        EventClass.HEARTBEAT,
        _now(),
        {"class": event_class.value, "assertion": "nothing-to-report"},
    )


class Outbox:
    """In-memory domain-test double; use ``store.Store`` for durable atomicity.

    Emit-after-commit is a defect. The queue is local so an `audit-core` outage
    never blocks a binding act — the same reasoning that keeps it from blocking
    a revocation upstream.

    This in-memory implementation stands in for the transactional store; the
    property it must preserve is that :meth:`append` cannot succeed while the
    state change fails, or vice versa.
    """

    def __init__(self) -> None:
        self._pending: list[Commitment] = []
        self._drained: list[Commitment] = []

    def append(self, commitment: Commitment) -> None:
        self._pending.append(commitment)

    @property
    def pending(self) -> tuple[Commitment, ...]:
        return tuple(self._pending)

    @property
    def drained(self) -> tuple[Commitment, ...]:
        return tuple(self._drained)

    def drain(self, sink) -> int:
        """Deliver to `audit-core`. Failure leaves the record pending, never lost."""
        sent = 0
        while self._pending:
            c = self._pending[0]
            sink(c.as_envelope())
            self._drained.append(self._pending.pop(0))
            sent += 1
        return sent

    def counts_by_class(self) -> dict[str, int]:
        """Our side of reconciliation, keyed by class.

        Bounded: where the emitter is compromised this count is suppressed
        alongside the event, and reconciliation agrees with it. Covers loss,
        outage, drain failure and accident — not adversarial omission by us.
        """
        counts: dict[str, int] = {}
        for c in [*self._drained, *self._pending]:
            counts[c.event_class.value] = counts.get(c.event_class.value, 0) + 1
        return counts