| 1 | """Append-only in-memory audit trail for the deterministic local path.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from collections.abc import Callable |
| 6 | from datetime import UTC, datetime |
| 7 | from typing import Any |
| 8 | |
| 9 | from .models import AuditEvent, AuditEventType |
| 10 | |
| 11 | Clock = Callable[[], datetime] |
| 12 | |
| 13 | |
| 14 | def utc_now() -> datetime: |
| 15 | return datetime.now(UTC) |
| 16 | |
| 17 | |
| 18 | class AuditTrail: |
| 19 | """Record ordered events without exposing mutable internal state.""" |
| 20 | |
| 21 | def __init__(self, *, clock: Clock = utc_now) -> None: |
| 22 | self._clock = clock |
| 23 | self._events: list[AuditEvent] = [] |
| 24 | |
| 25 | def append( |
| 26 | self, |
| 27 | *, |
| 28 | request_id: str, |
| 29 | event_type: AuditEventType, |
| 30 | detail: dict[str, Any] | None = None, |
| 31 | ) -> AuditEvent: |
| 32 | event = AuditEvent( |
| 33 | sequence=len(self._events) + 1, |
| 34 | occurred_at=self._clock(), |
| 35 | request_id=request_id, |
| 36 | event_type=event_type, |
| 37 | detail=detail or {}, |
| 38 | ) |
| 39 | self._events.append(event) |
| 40 | return event |
| 41 | |
| 42 | def for_request(self, request_id: str) -> tuple[AuditEvent, ...]: |
| 43 | return tuple(event for event in self._events if event.request_id == request_id) |
| 44 | |
| 45 | @property |
| 46 | def events(self) -> tuple[AuditEvent, ...]: |
| 47 | return tuple(self._events) |