diff --git a/src/brigade/receipt_schema.py b/src/brigade/receipt_schema.py index 73664332..3d39fffe 100644 --- a/src/brigade/receipt_schema.py +++ b/src/brigade/receipt_schema.py @@ -19,6 +19,9 @@ SYNTHESIS_SCHEMA = "brigade.synthesis.v1" SYNTHESIS_SCHEMA_VERSION = 1 +RUN_EVENT_SCHEMA = "brigade.run_event.v1" +RUN_EVENT_SCHEMA_VERSION = 1 + def stamp_run_receipt(payload: dict[str, object]) -> dict[str, object]: payload.setdefault("schema", RUN_RECEIPT_SCHEMA) diff --git a/src/brigade/run_events.py b/src/brigade/run_events.py new file mode 100644 index 00000000..cb6dd7aa --- /dev/null +++ b/src/brigade/run_events.py @@ -0,0 +1,402 @@ +"""brigade.run_event.v1 envelope: canonicalization, digests, and validation. + +Implements the typed run-lifecycle event envelope for the append-only per-run +journal kernel (issue #568, slice 1). The envelope is a closed-key JSON object +serialized as canonical UTF-8 (sorted keys, compact separators, no ASCII +escaping) with exact six-digit UTC-Z timestamps, integer-only payload numbers, +explicit null handling, deterministic request/event digests, and deterministic +per-run event IDs. + +Digest binding (non-circular): ``event_digest`` is the SHA-256 of the canonical +envelope with both ``event_digest`` and ``event_id`` excluded, and +``event_id = f"{run_id}-{sequence:06d}-{event_digest[:12]}"``. Excluding +``event_id`` from the digest input is required because ``event_id`` embeds +``event_digest[:12]``; including it would create an infeasible SHA-256 fixed +point. ``request_digest`` is the SHA-256 of the canonical request object +``{event_type, payload, idempotency_key}`` and is computed by the append API, +never trusted from the caller. + +Standard library only. Brigade is zero-runtime-dependency. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import datetime, timezone +from typing import Any, Mapping + +from brigade.receipt_schema import RUN_EVENT_SCHEMA, RUN_EVENT_SCHEMA_VERSION + +SCHEMA = RUN_EVENT_SCHEMA +SCHEMA_VERSION = RUN_EVENT_SCHEMA_VERSION + +MAX_LINE_BYTES = 16384 +MAX_IDEMPOTENCY_KEY_LEN = 128 +MAX_PAYLOAD_STR_LEN = 512 +MAX_DIAGNOSTIC_LEN = 240 +# Canonical integers are bounded to the signed 64-bit range so every reader +# (including non-Python consumers) can represent them exactly. +MAX_CANONICAL_INT = (1 << 63) - 1 +MIN_CANONICAL_INT = -(1 << 63) + +_RUN_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") +_HEX64 = re.compile(r"^[0-9a-f]{64}$") +_RECORDED_AT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$") + +# Closed set of envelope keys. Any key outside this set fails closed. +ENVELOPE_KEYS = frozenset( + { + "schema", + "schema_version", + "event_id", + "run_id", + "sequence", + "event_type", + "recorded_at", + "idempotency_key", + "request_digest", + "previous_digest", + "event_digest", + "payload", + } +) + +# Allowlisted event-type registry -> closed per-type payload key set. +# Payloads are flat (no nested objects) so the allowlist is enforceable and +# private-data exclusions are structural: no key can carry raw prompts, model +# output, tool arguments, credentials, provider response bodies, or stack +# traces -- only bounded detail/status/seat-style strings and integer attempts. +EVENT_TYPES: dict[str, frozenset[str]] = { + "run.created": frozenset({"status"}), + "run.planning.started": frozenset({"detail"}), + "run.planning.completed": frozenset({"detail"}), + "run.planning.failed": frozenset({"detail"}), + "run.dispatch.requested": frozenset({"seat", "attempt", "detail"}), + "run.dispatch.observed": frozenset({"seat", "attempt", "detail"}), + "run.dispatch.completed": frozenset({"seat", "attempt", "detail"}), + "run.dispatch.failed": frozenset({"seat", "attempt", "detail"}), + "run.synthesis.started": frozenset({"detail"}), + "run.synthesis.completed": frozenset({"detail"}), + "run.synthesis.failed": frozenset({"detail"}), + "run.paused": frozenset({"approval_id", "reason"}), + "run.resumed": frozenset({"approval_id"}), + "approval.requested": frozenset({"approval_id", "source", "contract_fingerprint"}), + "approval.granted": frozenset({"approval_id", "decided_at", "decision_state"}), + "approval.rejected": frozenset({"approval_id", "decided_at", "decision_state"}), + "approval.held": frozenset({"approval_id", "decided_at", "decision_state"}), + "approval.consumed": frozenset({"approval_id", "consuming_run_id"}), + "run.recovery.started": frozenset({"detail"}), + "run.recovery.completed": frozenset({"detail"}), + "run.completed": frozenset({"status", "detail"}), + "run.failed": frozenset({"status", "detail"}), + "run.interrupted": frozenset({"status", "detail"}), +} + +# Private-data exclusion list: these payload key names are never part of any +# allowlist and are rejected explicitly if a caller attempts to set them. +FORBIDDEN_PAYLOAD_KEYS = frozenset( + { + "prompt", + "prompt_text", + "model_output", + "tool_args", + "credentials", + "provider_response", + "stack_trace", + } +) + + +class CanonicalizationError(ValueError): + """Raised when a value cannot be canonicalized under the strict rules.""" + + +def _bound(msg: str) -> str: + if len(msg) <= MAX_DIAGNOSTIC_LEN: + return msg + return msg[: MAX_DIAGNOSTIC_LEN - 1] + "\u2026" + + +def _validate_canonical_node(value: Any) -> None: + """Recursively enforce the strict canonical value rules.""" + if value is None: + return + if isinstance(value, bool): + raise CanonicalizationError("booleans are not allowed in canonical values") + if isinstance(value, int): + if value > MAX_CANONICAL_INT or value < MIN_CANONICAL_INT: + raise CanonicalizationError("integer exceeds the signed 64-bit canonical range") + return + if isinstance(value, str): + return + if isinstance(value, list): + for item in value: + _validate_canonical_node(item) + return + if isinstance(value, dict): + for key, child in value.items(): + if not isinstance(key, str): + raise CanonicalizationError(f"non-string key {key!r}") + _validate_canonical_node(child) + return + raise CanonicalizationError(f"unsupported canonical value type {type(value).__name__}") + + +def canonical_bytes(obj: Any) -> bytes: + """Canonical UTF-8 JSON: sorted keys, compact separators, no ASCII escaping. + + Rejects floats, booleans, oversized integers, and any non-JSON-native type + before serializing. Any residual ``json.dumps`` conversion failure (type + errors, circular references, recursion/overflow) is wrapped in a bounded + CanonicalizationError rather than escaping as a raw exception. + """ + try: + _validate_canonical_node(obj) + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + except CanonicalizationError: + raise + except (TypeError, ValueError, OverflowError, RecursionError) as exc: + raise CanonicalizationError(_bound(f"value cannot be canonicalized: {exc}")) from exc + + +def _sha256_hex(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def format_recorded_at(dt: datetime) -> str: + """Format a datetime as an exact six-digit UTC-Z timestamp.""" + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + dt = dt.astimezone(timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond:06d}Z" + + +def is_valid_recorded_at(value: object) -> bool: + return isinstance(value, str) and bool(_RECORDED_AT_RE.match(value)) + + +def request_digest(*, event_type: str, payload: Mapping[str, Any], idempotency_key: str) -> str: + """SHA-256 of the canonical request object ``{event_type, payload, idempotency_key}``. + + The request is canonicalized under the same strict rules; non-integer + numbers, booleans, and unknown types raise CanonicalizationError. + """ + request = { + "event_type": event_type, + "payload": dict(payload), + "idempotency_key": idempotency_key, + } + return _sha256_hex(canonical_bytes(request)) + + +def compute_event_digest(envelope: Mapping[str, Any]) -> str: + """SHA-256 of the canonical envelope excluding ``event_digest`` and ``event_id``. + + ``event_id`` is excluded because it embeds ``event_digest[:12]``; including + it would create an infeasible SHA-256 fixed point. This is the single + deviation from the "exclude only event_digest" phrasing in the slice-1 plan + and is required for the binding to be computable. + """ + subset = {k: v for k, v in envelope.items() if k not in ("event_digest", "event_id")} + return _sha256_hex(canonical_bytes(subset)) + + +def make_event_id(*, run_id: str, sequence: int, event_digest: str) -> str: + return f"{run_id}-{sequence:06d}-{event_digest[:12]}" + + +def build_event( + *, + run_id: str, + sequence: int, + event_type: str, + payload: Mapping[str, Any], + idempotency_key: str, + recorded_at: str, + previous_digest: str | None, + request_digest_value: str | None = None, +) -> dict[str, Any]: + """Build a fully validated, self-consistent run_event.v1 envelope. + + ``request_digest_value`` may be supplied by the journal layer (which has + already computed it for idempotency); if omitted it is derived here. + """ + if not isinstance(run_id, str) or not _RUN_ID_RE.match(run_id): + raise CanonicalizationError(_bound(f"invalid run_id {run_id!r}")) + if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 1: + raise CanonicalizationError(_bound(f"invalid sequence {sequence!r}")) + if event_type not in EVENT_TYPES: + raise CanonicalizationError(_bound(f"unknown event_type {event_type!r}")) + if not isinstance(idempotency_key, str) or not idempotency_key: + raise CanonicalizationError("idempotency_key must be a non-empty string") + if len(idempotency_key) > MAX_IDEMPOTENCY_KEY_LEN: + raise CanonicalizationError(_bound(f"idempotency_key exceeds {MAX_IDEMPOTENCY_KEY_LEN} chars")) + if not is_valid_recorded_at(recorded_at): + raise CanonicalizationError(_bound(f"invalid recorded_at {recorded_at!r}")) + if previous_digest is not None and not (isinstance(previous_digest, str) and bool(_HEX64.match(previous_digest))): + raise CanonicalizationError(_bound("invalid previous_digest")) + if sequence == 1 and previous_digest is not None: + raise CanonicalizationError("previous_digest must be null at sequence 1") + if sequence > 1 and previous_digest is None: + raise CanonicalizationError("previous_digest must not be null after sequence 1") + + _validate_payload(event_type, payload) + + rd = request_digest_value or request_digest(event_type=event_type, payload=payload, idempotency_key=idempotency_key) + if not (isinstance(rd, str) and bool(_HEX64.match(rd))): + raise CanonicalizationError(_bound("invalid request_digest")) + + base = { + "schema": SCHEMA, + "schema_version": SCHEMA_VERSION, + "run_id": run_id, + "sequence": sequence, + "event_type": event_type, + "recorded_at": recorded_at, + "idempotency_key": idempotency_key, + "request_digest": rd, + "previous_digest": previous_digest, + "payload": dict(payload), + } + event_digest = compute_event_digest(base) + event_id = make_event_id(run_id=run_id, sequence=sequence, event_digest=event_digest) + envelope = {**base, "event_id": event_id, "event_digest": event_digest} + errors = validate_event(envelope) + if errors: + raise CanonicalizationError(_bound("invalid built envelope: " + "; ".join(errors))) + return envelope + + +def _validate_payload(event_type: str, payload: Any) -> None: + if not isinstance(payload, Mapping): + raise CanonicalizationError("payload must be an object") + allowed = EVENT_TYPES[event_type] + extra = set(payload.keys()) - allowed + if extra: + forbidden = extra & FORBIDDEN_PAYLOAD_KEYS + if forbidden: + raise CanonicalizationError(_bound(f"forbidden private-data payload keys: {sorted(forbidden)}")) + raise CanonicalizationError(_bound(f"unknown payload keys for {event_type}: {sorted(extra)}")) + for key, value in payload.items(): + if value is None: + continue + if isinstance(value, bool): + raise CanonicalizationError(_bound(f"payload {key!r} must not be boolean")) + if isinstance(value, float): + raise CanonicalizationError(_bound(f"payload {key!r} must be integer, not float")) + if isinstance(value, int): + continue + if isinstance(value, str): + if len(value) > MAX_PAYLOAD_STR_LEN: + raise CanonicalizationError(_bound(f"payload {key!r} exceeds {MAX_PAYLOAD_STR_LEN} chars")) + continue + raise CanonicalizationError(_bound(f"payload {key!r} has unsupported type {type(value).__name__}")) + + +def validate_event(env: Any) -> list[str]: + """Return a list of bounded diagnostic strings for an envelope. Empty = valid.""" + errors: list[str] = [] + if not isinstance(env, Mapping): + return ["envelope must be a JSON object"] + + unknown = sorted(set(env.keys()) - ENVELOPE_KEYS) + if unknown: + shown = ", ".join(unknown[:8]) + errors.append(_bound(f"unknown envelope keys: {shown}")) + + if env.get("schema") != SCHEMA: + errors.append(_bound(f"schema must be {SCHEMA!r}")) + if env.get("schema_version") != SCHEMA_VERSION: + errors.append(_bound(f"schema_version must be {SCHEMA_VERSION}")) + + run_id = env.get("run_id") + if not isinstance(run_id, str) or not _RUN_ID_RE.match(run_id): + errors.append(_bound(f"invalid run_id {run_id!r}")) + + sequence = env.get("sequence") + if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 1: + errors.append(_bound(f"invalid sequence {sequence!r}")) + + event_type = env.get("event_type") + if event_type not in EVENT_TYPES: + errors.append(_bound(f"unknown event_type {event_type!r}")) + + recorded_at = env.get("recorded_at") + if not is_valid_recorded_at(recorded_at): + errors.append(_bound(f"invalid recorded_at {recorded_at!r}")) + + idempotency_key = env.get("idempotency_key") + if not isinstance(idempotency_key, str) or not idempotency_key: + errors.append("idempotency_key must be a non-empty string") + elif len(idempotency_key) > MAX_IDEMPOTENCY_KEY_LEN: + errors.append(_bound(f"idempotency_key exceeds {MAX_IDEMPOTENCY_KEY_LEN} chars")) + + request_digest_value = env.get("request_digest") + if not (isinstance(request_digest_value, str) and bool(_HEX64.match(request_digest_value))): + errors.append("request_digest must be a 64-char lowercase hex string") + + previous_digest = env.get("previous_digest") + if "previous_digest" not in env: + errors.append("previous_digest must be present (use null at sequence 1)") + elif previous_digest is None: + if isinstance(sequence, int) and not isinstance(sequence, bool) and sequence != 1: + errors.append("previous_digest must not be null after sequence 1") + elif not (isinstance(previous_digest, str) and bool(_HEX64.match(previous_digest))): + errors.append("previous_digest must be 64-char hex or null") + + event_digest = env.get("event_digest") + if not (isinstance(event_digest, str) and bool(_HEX64.match(event_digest))): + errors.append("event_digest must be a 64-char lowercase hex string") + event_digest = None + + event_id = env.get("event_id") + if not isinstance(event_id, str) or not event_id: + errors.append("event_id must be a non-empty string") + + payload = env.get("payload") + if not isinstance(payload, Mapping): + errors.append("payload must be an object") + elif event_type in EVENT_TYPES: + allowed = EVENT_TYPES[event_type] + extra = set(payload.keys()) - allowed + if extra: + forbidden = extra & FORBIDDEN_PAYLOAD_KEYS + if forbidden: + errors.append(_bound(f"forbidden private-data payload keys: {sorted(forbidden)}")) + else: + errors.append(_bound(f"unknown payload keys for {event_type}: {sorted(extra)}")) + for key, value in payload.items(): + if value is None: + continue + if isinstance(value, bool): + errors.append(_bound(f"payload {key!r} must not be boolean")) + elif isinstance(value, float): + errors.append(_bound(f"payload {key!r} must be integer, not float")) + elif isinstance(value, int): + continue + elif isinstance(value, str): + if len(value) > MAX_PAYLOAD_STR_LEN: + errors.append(_bound(f"payload {key!r} exceeds {MAX_PAYLOAD_STR_LEN} chars")) + else: + errors.append(_bound(f"payload {key!r} has unsupported type")) + + if errors: + return errors + + # Recompute digests and event_id for content binding (only when structurally valid). + try: + recomputed_digest = compute_event_digest(env) + except CanonicalizationError as exc: + errors.append(_bound(f"envelope is not canonicalizable: {exc}")) + return errors + if event_digest != recomputed_digest: + errors.append(_bound("event_digest does not match recomputed digest")) + assert isinstance(run_id, str) + assert isinstance(sequence, int) and not isinstance(sequence, bool) + expected_id = make_event_id(run_id=run_id, sequence=sequence, event_digest=recomputed_digest) + if event_id != expected_id: + errors.append(_bound("event_id does not match run_id/sequence/event_digest")) + + return errors diff --git a/src/brigade/run_journal.py b/src/brigade/run_journal.py new file mode 100644 index 00000000..4b41ca27 --- /dev/null +++ b/src/brigade/run_journal.py @@ -0,0 +1,683 @@ +"""Append-only per-run lifecycle journal kernel (issue #568, slice 1). + +Persists ``brigade.run_event.v1`` envelopes as canonical UTF-8 JSON lines at +``/events/lifecycle.jsonl``. Append is a single bounded ``os.write`` +to an ``O_APPEND`` descriptor with returned-byte-count verification and +``fsync`` before return. The append API requires ``expected_previous_sequence`` +and enforces contiguous sequence, previous-digest chaining, and idempotency by +key + request digest (same key + same digest returns the existing event; same +key + different digest raises a typed conflict without appending). Tail state +is derived fail-closed: every complete line must be a validated envelope whose +raw bytes exactly equal its canonical form, continuing a gap-free, +duplicate-free, digest-linked sequence; any deviation raises a bounded typed +error and no state is derived from it. Run-artifact permissions are private: +the ``events`` and quarantine directories are 0o700 and journal/quarantine +files are 0o600. On POSIX hosts with ``O_NOFOLLOW``, ``O_DIRECTORY``, and +``fchmod``, modes are enforced via ``fchmod`` on no-follow descriptors so a +permissive umask cannot widen them and a pre-placed symlink cannot redirect +reads, writes, or permission correction. On hosts where any of those APIs are +absent (notably Windows), the module still imports and journal operations use +a symlink-rejecting path-mode fallback: the final path component is rejected +via ``lstat`` before open, opened regular files are verified against the path +with ``lstat``/``fstat`` inode identity before mutation, and modes are applied +with ``chmod`` on the verified path when ``fchmod`` is unavailable. Normal +readers report a partial final line without mutating it; a separate +recovery-only API quarantines the incomplete suffix (write-once, collision-safe) +before truncating. + +Standard library only. Brigade is zero-runtime-dependency. +""" + +from __future__ import annotations + +import errno +import hashlib +import json +import os +import stat +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterator + +from brigade import run_events +from brigade.run_events import CanonicalizationError, canonical_bytes + +_DIR_MODE = 0o700 +_FILE_MODE = 0o600 +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) +_O_DIRECTORY = getattr(os, "O_DIRECTORY", 0) +_HAS_O_NOFOLLOW = _O_NOFOLLOW != 0 +_HAS_O_DIRECTORY = _O_DIRECTORY != 0 +_HAS_FCHMOD = hasattr(os, "fchmod") +# O_NOFOLLOW rejects symlinked targets so a pre-placed symlink cannot redirect +# journal writes or quarantine captures outside the private run-artifact tree. +_OPEN_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_APPEND +if _HAS_O_NOFOLLOW: + _OPEN_FLAGS |= _O_NOFOLLOW +_READ_CHUNK = 65536 +_MAX_QUARANTINE_ATTEMPTS = 16 + + +class RunJournalError(RuntimeError): + """Base class for run-journal failures. Carries a bounded ``diagnostic``.""" + + def __init__(self, diagnostic: str) -> None: + super().__init__(diagnostic) + self.diagnostic = diagnostic + + +class StaleSequenceError(RunJournalError): + """``expected_previous_sequence`` did not match the journal tail.""" + + +class IdempotencyConflict(RunJournalError): + """Same idempotency key recurred with a different request digest.""" + + def __init__( + self, + diagnostic: str, + *, + existing_event_id: str, + request_digest: str, + existing_request_digest: str, + ) -> None: + super().__init__(diagnostic) + self.existing_event_id = existing_event_id + self.request_digest = request_digest + self.existing_request_digest = existing_request_digest + + +class PartialWriteError(RunJournalError): + """``os.write`` returned fewer bytes than the canonical line.""" + + +class ChainIntegrityError(RunJournalError): + """Chain verification detected a gap, duplicate, or digest mismatch.""" + + +class UnknownFieldError(RunJournalError): + """An envelope or payload carried a key outside the closed allowlist.""" + + +class UnknownEventTypeError(RunJournalError): + """An envelope carried an event_type outside the registry.""" + + +class SchemaVersionError(RunJournalError): + """An envelope carried an unknown schema string or version.""" + + +class PartialTailError(RunJournalError): + """The journal ends in a partial (unterminated) line; recovery is required.""" + + +@dataclass(frozen=True) +class RunEvent: + """A validated, self-consistent run_event.v1 envelope read from the journal.""" + + schema: str + schema_version: int + event_id: str + run_id: str + sequence: int + event_type: str + recorded_at: str + idempotency_key: str + request_digest: str + previous_digest: str | None + event_digest: str + payload: dict[str, Any] + + def to_dict(self) -> dict[str, Any]: + return { + "schema": self.schema, + "schema_version": self.schema_version, + "event_id": self.event_id, + "run_id": self.run_id, + "sequence": self.sequence, + "event_type": self.event_type, + "recorded_at": self.recorded_at, + "idempotency_key": self.idempotency_key, + "request_digest": self.request_digest, + "previous_digest": self.previous_digest, + "event_digest": self.event_digest, + "payload": dict(self.payload), + } + + +@dataclass +class JournalReport: + """Result of a non-mutating journal read.""" + + events: list[RunEvent] = field(default_factory=list) + partial_tail: bytes | None = None + chain_errors: list[str] = field(default_factory=list) + + +@dataclass +class RecoveryReport: + """Result of a recovery-only partial-tail quarantine + truncate. + + ``quarantine_path`` is None when the journal had no partial tail to + capture (nothing was written). + """ + + partial_bytes: bytes + quarantine_path: Path | None + + +def _reject_symlink_final_component(path: Path) -> None: + """Reject a symlinked final path component via lstat (fallback no-follow guard).""" + try: + st = os.lstat(path) + except FileNotFoundError: + return + except OSError as exc: + raise RunJournalError(_bound(f"cannot stat path: {path.name}")) from exc + if stat.S_ISLNK(st.st_mode): + raise RunJournalError(_bound(f"refusing symlinked path: {path.name}")) + + +def _verify_fd_identity(path: Path, fd: int) -> None: + """Verify an opened descriptor still refers to the same inode as lstat(path).""" + try: + st_path = os.lstat(path) + st_fd = os.fstat(fd) + except OSError as exc: + raise RunJournalError(_bound(f"cannot verify opened path: {path.name}")) from exc + if st_path.st_ino != st_fd.st_ino or st_path.st_dev != st_fd.st_dev: + raise RunJournalError(_bound(f"opened path identity mismatch: {path.name}")) + + +def _chmod_fd_or_path(fd: int, path: Path, mode: int) -> None: + """Apply mode via fchmod when available, else chmod on a verified path.""" + if _HAS_FCHMOD: + os.fchmod(fd, mode) + return + _verify_fd_identity(path, fd) + os.chmod(path, mode) + + +def _open_nofollow(path: Path, flags: int, mode: int = 0o666) -> int: + """Open a path without following a symlinked final component. + + When ``O_NOFOLLOW`` is available, ``os.open`` rejects symlinked targets + directly (ELOOP or ENOTDIR for ``O_DIRECTORY`` on a symlinked directory). + Otherwise the final component is rejected via ``lstat`` before open and + inode identity is verified with ``lstat``/``fstat`` before the descriptor + is returned. + """ + wants_directory = bool(flags & _O_DIRECTORY) + open_flags = flags + if wants_directory and not _HAS_O_DIRECTORY: + open_flags &= ~_O_DIRECTORY + + if _HAS_O_NOFOLLOW: + try: + return os.open(path, open_flags | _O_NOFOLLOW, mode) + except OSError as exc: + if exc.errno == errno.ELOOP: + raise RunJournalError(_bound(f"refusing symlinked path: {path.name}")) from exc + if exc.errno == errno.ENOTDIR and wants_directory: + # Linux reports ENOTDIR (not ELOOP) for O_DIRECTORY|O_NOFOLLOW on + # a symlinked directory; it is the same rejection. + raise RunJournalError(_bound(f"refusing symlinked path: {path.name}")) from exc + raise + + _reject_symlink_final_component(path) + if wants_directory and not _HAS_O_DIRECTORY: + try: + st = os.lstat(path) + except FileNotFoundError as exc: + raise RunJournalError(_bound(f"directory path does not exist: {path.name}")) from exc + except OSError as exc: + raise RunJournalError(_bound(f"cannot stat path: {path.name}")) from exc + if not stat.S_ISDIR(st.st_mode): + raise RunJournalError(_bound(f"path is not a directory: {path.name}")) + + fd = os.open(path, open_flags, mode) + try: + _verify_fd_identity(path, fd) + except Exception: + os.close(fd) + raise + return fd + + +def _enforce_dir_mode(path: Path) -> None: + """Enforce 0o700 on a directory without following a symlinked final component. + + With ``O_NOFOLLOW`` and ``fchmod``, mode is corrected on the opened + directory descriptor. Without those APIs, the symlink guard and inode + identity check run first, then ``chmod`` is applied on the verified path. + """ + _reject_symlink_final_component(path) + try: + st = os.lstat(path) + except OSError as exc: + raise RunJournalError(_bound(f"cannot stat path: {path.name}")) from exc + if not stat.S_ISDIR(st.st_mode): + raise RunJournalError(_bound(f"path is not a directory: {path.name}")) + if not _HAS_O_DIRECTORY: + if stat.S_IMODE(st.st_mode) != _DIR_MODE: + os.chmod(path, _DIR_MODE) + return + dir_flags = os.O_RDONLY | _O_DIRECTORY + fd = _open_nofollow(path, dir_flags) + try: + if stat.S_IMODE(os.fstat(fd).st_mode) != _DIR_MODE: + _chmod_fd_or_path(fd, path, _DIR_MODE) + finally: + os.close(fd) + + +def _enforce_file_mode(path: Path) -> None: + """Enforce 0o600 on a regular file without following a symlinked final component. + + With ``O_NOFOLLOW`` and ``fchmod``, mode is corrected on the opened file + descriptor. Without those APIs, the symlink guard and inode identity check + run first, then ``chmod`` is applied on the verified path. + """ + fd = _open_nofollow(path, os.O_RDONLY) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode): + raise RunJournalError(_bound(f"journal path is not a regular file: {path.name}")) + if stat.S_IMODE(info.st_mode) != _FILE_MODE: + _chmod_fd_or_path(fd, path, _FILE_MODE) + finally: + os.close(fd) + + +def _mkdir_private(path: Path) -> None: + """Create a directory with mode 0o700 at mkdir time, then enforce it.""" + path.mkdir(parents=True, exist_ok=True, mode=_DIR_MODE) + _enforce_dir_mode(path) + + +def _read_bytes_nofollow(path: Path) -> bytes: + """Read a regular file without following a symlinked final component. + + Uses an ``O_NOFOLLOW`` descriptor when available; otherwise applies the + ``lstat`` symlink guard and inode identity verification before reading. + """ + fd = _open_nofollow(path, os.O_RDONLY) + try: + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise RunJournalError(_bound(f"journal path is not a regular file: {path.name}")) + chunks: list[bytes] = [] + while True: + chunk = os.read(fd, _READ_CHUNK) + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + finally: + os.close(fd) + + +def ensure_journal(journal_path: Path) -> None: + """Create the events directory (0o700) and journal file (0o600) if missing. + + Modes are applied at mkdir/open time and re-enforced without following a + symlinked final component: via ``fchmod`` on no-follow descriptors when + ``O_NOFOLLOW`` and ``fchmod`` exist, otherwise via ``lstat`` rejection, + inode identity verification, and path ``chmod``. + """ + journal_path = Path(journal_path) + _mkdir_private(journal_path.parent) + if not journal_path.exists(): + fd = _open_nofollow(journal_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, _FILE_MODE) + os.close(fd) + _enforce_file_mode(journal_path) + + +class _DuplicateKeyError(ValueError): + """Internal: a JSON object carried the same key twice.""" + + def __init__(self, key: str) -> None: + super().__init__(key) + self.key = key + + +def _object_pairs_no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + obj: dict[str, Any] = {} + for key, value in pairs: + if key in obj: + raise _DuplicateKeyError(key) + obj[key] = value + return obj + + +def _parse_canonical_line(line: bytes) -> dict[str, Any]: + """Parse one journal line, failing closed on any deviation. + + The line must be a validated run_event.v1 envelope whose raw bytes exactly + equal its canonical form. Raises ChainIntegrityError with a bounded + diagnostic on: invalid UTF-8 or JSON, duplicate JSON keys, non-object JSON, + uncanonicalizable values (floats, booleans, oversized integers), byte-level + differences from canonical form (whitespace, key order, ASCII escapes), or + failed envelope validation (bad fields, recomputed-digest or event_id + mismatch). + """ + try: + text = line.decode("utf-8") + except UnicodeDecodeError as exc: + raise ChainIntegrityError(_bound("journal line is not valid UTF-8")) from exc + try: + env = json.loads(text, object_pairs_hook=_object_pairs_no_duplicates) + except _DuplicateKeyError as exc: + raise ChainIntegrityError(_bound(f"journal line repeats JSON key {exc.key!r}")) from exc + except json.JSONDecodeError as exc: + raise ChainIntegrityError(_bound(f"journal line is not valid JSON: {exc}")) from exc + except ValueError as exc: + raise ChainIntegrityError(_bound("journal line integer exceeds parser limits")) from exc + except RecursionError as exc: + raise ChainIntegrityError(_bound("journal line is nested too deeply")) from exc + if not isinstance(env, dict): + raise ChainIntegrityError(_bound("journal line is not a JSON object")) + try: + canonical = canonical_bytes(env) + except CanonicalizationError as exc: + raise ChainIntegrityError(_bound(f"journal line is not canonicalizable: {exc}")) from exc + if canonical != line: + raise ChainIntegrityError(_bound("journal line bytes differ from canonical form")) + errors = run_events.validate_event(env) + if errors: + raise ChainIntegrityError(_bound("journal line failed validation: " + "; ".join(errors))) + return env + + +def _read_tail_state( + journal_path: Path, +) -> tuple[int, str | None, dict[str, dict[str, Any]], bytes | None]: + """Return (last_sequence, last_event_digest, idempotency_index, partial_tail). + + Fail closed: every complete line must be a validated canonical envelope + (see ``_parse_canonical_line``) continuing a strict, gap-free, + duplicate-free sequence whose previous_digest links to the prior + event_digest. Any deviation raises ChainIntegrityError and no state is + derived from it; last_sequence and last_digest come only from the final + verified line. The idempotency index maps idempotency_key -> validated + envelope dict, and a repeated key raises. last_sequence is 0 for an + empty/missing journal. A non-empty trailing segment without a terminating + newline is returned as ``partial_tail`` so the append path can refuse to + write over it (appending over a partial tail would glue the new line to + the partial bytes and corrupt the journal). + """ + last_sequence = 0 + last_digest: str | None = None + index: dict[str, dict[str, Any]] = {} + partial_tail: bytes | None = None + if os.path.lexists(journal_path) and not journal_path.exists(): + raise RunJournalError(_bound(f"journal path is a dangling symlink: {journal_path.name}")) + if not journal_path.exists(): + return last_sequence, last_digest, index, partial_tail + raw = _read_bytes_nofollow(journal_path) + segments = raw.split(b"\n") + # A non-empty final segment after the last newline is a partial tail. + if segments[-1]: + partial_tail = segments[-1] + for line in segments[:-1]: + env = _parse_canonical_line(line) + sequence = env["sequence"] + if sequence != last_sequence + 1: + raise ChainIntegrityError(_bound(f"journal sequence break: expected {last_sequence + 1}, got {sequence}")) + if sequence > 1 and env["previous_digest"] != last_digest: + raise ChainIntegrityError(_bound("journal previous_digest does not link to prior event_digest")) + key = env["idempotency_key"] + if key in index: + raise ChainIntegrityError(_bound(f"journal repeats idempotency key {key!r}")) + index[key] = env + last_sequence = sequence + last_digest = env["event_digest"] + return last_sequence, last_digest, index, partial_tail + + +def _envelope_to_event(env: dict[str, Any]) -> RunEvent: + """Materialize a validated envelope dict. + + Callers validate envelopes first; malformed fields here raise a bounded + typed error instead of KeyError (defense in depth). + """ + try: + return RunEvent( + schema=env["schema"], + schema_version=env["schema_version"], + event_id=env["event_id"], + run_id=env["run_id"], + sequence=env["sequence"], + event_type=env["event_type"], + recorded_at=env["recorded_at"], + idempotency_key=env["idempotency_key"], + request_digest=env["request_digest"], + previous_digest=env["previous_digest"], + event_digest=env["event_digest"], + payload=dict(env["payload"]), + ) + except (KeyError, TypeError) as exc: + raise ChainIntegrityError(_bound(f"envelope fields are malformed: {exc}")) from exc + + +def append_event( + journal_path: Path, + *, + run_id: str, + event_type: str, + payload: dict[str, Any], + idempotency_key: str, + expected_previous_sequence: int, + recorded_at: str | None = None, +) -> RunEvent: + """Append one event to the journal under the slice-1 contract. + + Idempotency: same key + same request digest returns the existing event with + no write; same key + different digest raises IdempotencyConflict with no + write. Concurrency: ``expected_previous_sequence`` must equal the current + tail sequence (0 for an empty journal) else StaleSequenceError, no write. + The write is a single bounded ``os.write`` to an ``O_APPEND`` descriptor + with returned-byte-count verification and ``fsync`` before return. + """ + journal_path = Path(journal_path) + ensure_journal(journal_path) + + if recorded_at is None: + from datetime import datetime, timezone + + recorded_at = run_events.format_recorded_at(datetime.now(timezone.utc)) + + rd = run_events.request_digest(event_type=event_type, payload=payload, idempotency_key=idempotency_key) + + last_sequence, last_digest, index, partial_tail = _read_tail_state(journal_path) + + if partial_tail is not None: + raise PartialTailError(_bound("journal ends in a partial line; run recover_partial_tail before appending")) + + existing = index.get(idempotency_key) + if existing is not None: + existing_rd = existing.get("request_digest") + if not isinstance(existing_rd, str): + raise ChainIntegrityError(_bound("indexed journal event is missing request_digest")) + if existing_rd == rd: + return _envelope_to_event(existing) + existing_event_id = existing.get("event_id") + if not isinstance(existing_event_id, str): + raise ChainIntegrityError(_bound("indexed journal event is missing event_id")) + raise IdempotencyConflict( + _bound(f"idempotency key {idempotency_key!r} conflict"), + existing_event_id=existing_event_id, + request_digest=rd, + existing_request_digest=existing_rd, + ) + + if expected_previous_sequence != last_sequence: + raise StaleSequenceError( + _bound(f"stale sequence: expected previous {expected_previous_sequence}, actual {last_sequence}") + ) + + sequence = last_sequence + 1 + envelope = run_events.build_event( + run_id=run_id, + sequence=sequence, + event_type=event_type, + payload=payload, + idempotency_key=idempotency_key, + recorded_at=recorded_at, + previous_digest=last_digest, + request_digest_value=rd, + ) + + line = canonical_bytes(envelope) + b"\n" + if len(line) > run_events.MAX_LINE_BYTES: + raise CanonicalizationError(_bound(f"canonical line exceeds {run_events.MAX_LINE_BYTES} bytes")) + + fd = _open_nofollow(journal_path, _OPEN_FLAGS, _FILE_MODE) + try: + _chmod_fd_or_path(fd, journal_path, _FILE_MODE) + written = os.write(fd, line) + if written != len(line): + raise PartialWriteError(_bound(f"partial write: wrote {written} of {len(line)} bytes")) + os.fsync(fd) + finally: + os.close(fd) + + return _envelope_to_event(envelope) + + +def _iter_lines(raw: bytes) -> Iterator[tuple[bytes | None, bytes | None]]: + """Yield (complete_line_without_newline, None) for each full line and a final + (None, partial_tail) if a non-empty trailing segment lacks a newline.""" + segments = raw.split(b"\n") + if len(segments) <= 1: + if raw: + yield None, raw + return + for seg in segments[:-1]: + yield seg, None + tail = segments[-1] + if tail: + yield None, tail + + +def read_journal(journal_path: Path) -> JournalReport: + """Read the journal without mutating it. + + Each complete line must be a validated canonical envelope; lines that are + malformed, carry duplicate JSON keys, or differ byte-wise from canonical + form are reported as bounded chain_errors. Verified-prefix semantics + apply: after the first invalid complete line, sequence mismatch, or + previous-digest mismatch, only the bounded first error is reported and + no later events are returned. A non-empty trailing segment without a + terminating newline is reported as ``partial_tail`` verbatim. The file + is never written. + """ + journal_path = Path(journal_path) + report = JournalReport() + if os.path.lexists(journal_path) and not journal_path.exists(): + raise RunJournalError(_bound(f"journal path is a dangling symlink: {journal_path.name}")) + if not journal_path.exists(): + return report + raw = _read_bytes_nofollow(journal_path) + + expected_sequence = 1 + expected_previous: str | None = None + for complete, partial in _iter_lines(raw): + if partial is not None: + report.partial_tail = partial + continue + if complete is None: + continue + try: + env = _parse_canonical_line(complete) + except RunJournalError as exc: + report.chain_errors.append(exc.diagnostic) + break + event = _envelope_to_event(env) + if event.sequence != expected_sequence: + report.chain_errors.append( + _bound(f"sequence gap/duplicate: expected {expected_sequence}, got {event.sequence}") + ) + break + if event.sequence == 1: + if event.previous_digest is not None: + report.chain_errors.append("sequence 1 previous_digest must be null") + break + elif event.previous_digest != expected_previous: + report.chain_errors.append(_bound("previous_digest does not link to prior event_digest")) + break + report.events.append(event) + expected_sequence = event.sequence + 1 + expected_previous = event.event_digest + return report + + +def recover_partial_tail(journal_path: Path, quarantine_dir: Path) -> RecoveryReport: + """Recovery-only: quarantine the partial suffix verbatim, then truncate. + + The incomplete trailing bytes are captured exactly once under + ``quarantine_dir`` (0o700) in a write-once (O_EXCL) 0o600 file named with + the complete-line count, the full SHA-256 of the partial bytes, and an + exclusive numeric suffix on collision, then the journal is truncated to + the last complete line (fsynced). With no partial tail nothing is written + and ``quarantine_path`` is None. Normal readers must never call this; it + is the only API that mutates the journal body. + """ + journal_path = Path(journal_path) + quarantine_dir = Path(quarantine_dir) + if os.path.lexists(journal_path) and not journal_path.exists(): + raise RunJournalError(_bound(f"journal path is a dangling symlink: {journal_path.name}")) + if not journal_path.exists(): + raise RunJournalError(_bound(f"journal path does not exist: {journal_path.name}")) + _mkdir_private(quarantine_dir) + + raw = _read_bytes_nofollow(journal_path) + last_newline = raw.rfind(b"\n") + if last_newline == -1: + partial = raw + complete = b"" + else: + partial = raw[last_newline + 1 :] + complete = raw[: last_newline + 1] + + if not partial: + return RecoveryReport(partial_bytes=b"", quarantine_path=None) + + digest = hashlib.sha256(partial).hexdigest() + context = complete.count(b"\n") + stem = f"lifecycle-partial-{context:06d}-{digest}" + quarantine_path: Path | None = None + for attempt in range(_MAX_QUARANTINE_ATTEMPTS): + suffix = "" if attempt == 0 else f"-{attempt}" + candidate = quarantine_dir / f"{stem}{suffix}.bin" + try: + qfd = _open_nofollow(candidate, os.O_WRONLY | os.O_CREAT | os.O_EXCL, _FILE_MODE) + except FileExistsError: + continue + try: + _chmod_fd_or_path(qfd, candidate, _FILE_MODE) + written = os.write(qfd, partial) + if written != len(partial): + raise PartialWriteError(_bound("quarantine write was partial")) + os.fsync(qfd) + finally: + os.close(qfd) + quarantine_path = candidate + break + if quarantine_path is None: + raise RunJournalError(_bound("no collision-free quarantine name available")) + + jfd = _open_nofollow(journal_path, os.O_RDWR) + try: + _chmod_fd_or_path(jfd, journal_path, _FILE_MODE) + os.ftruncate(jfd, len(complete)) + os.fsync(jfd) + finally: + os.close(jfd) + + return RecoveryReport(partial_bytes=partial, quarantine_path=quarantine_path) + + +def _bound(msg: str) -> str: + limit = run_events.MAX_DIAGNOSTIC_LEN + if len(msg) <= limit: + return msg + return msg[: limit - 1] + "…" diff --git a/tests/fixtures/run-lifecycle/golden-lifecycle.jsonl b/tests/fixtures/run-lifecycle/golden-lifecycle.jsonl new file mode 100644 index 00000000..c225dbd8 --- /dev/null +++ b/tests/fixtures/run-lifecycle/golden-lifecycle.jsonl @@ -0,0 +1,6 @@ +{"event_digest":"6d43c3e25e4d4c8627166ab5b565622f98b0999218889b65ac8c84dcfac9301b","event_id":"20260727-153045-a1b2c3d4-000001-6d43c3e25e4d","event_type":"run.created","idempotency_key":"create-1","payload":{"status":"started"},"previous_digest":null,"recorded_at":"2026-07-27T15:30:45.123456Z","request_digest":"3c8ed91a3781b4f7033bc1940ee95eaa50a88f33d453e2b9fb9142150b3f89cb","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":1} +{"event_digest":"d107ec22cdc9f63adf1ff841f6777457c1923139e033e62cbab8df332abf26e2","event_id":"20260727-153045-a1b2c3d4-000002-d107ec22cdc9","event_type":"run.planning.started","idempotency_key":"plan-start-1","payload":{"detail":"planning"},"previous_digest":"6d43c3e25e4d4c8627166ab5b565622f98b0999218889b65ac8c84dcfac9301b","recorded_at":"2026-07-27T15:30:46.000000Z","request_digest":"eb7f6239f58f0291e7d16c682af96a11c2a2d8aa46d729fe2d15395fb9bce070","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":2} +{"event_digest":"dfd4e84746245b5f54ab6bc3d417718e2ea6034139e9127632c99dcce3e58a3f","event_id":"20260727-153045-a1b2c3d4-000003-dfd4e8474624","event_type":"run.dispatch.requested","idempotency_key":"dispatch-req-1","payload":{"attempt":1,"seat":"coder"},"previous_digest":"d107ec22cdc9f63adf1ff841f6777457c1923139e033e62cbab8df332abf26e2","recorded_at":"2026-07-27T15:30:47.000000Z","request_digest":"e329e657466e02dd86a9318373905bfad1d87ed1a2bcf1d2c8fe471ae9711e88","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":3} +{"event_digest":"203f8fa371b3c02c631876e7a5c00cdf1918abffef82e2cb6a89154efaa49f5a","event_id":"20260727-153045-a1b2c3d4-000004-203f8fa371b3","event_type":"run.dispatch.completed","idempotency_key":"dispatch-done-1","payload":{"attempt":1,"detail":"ok","seat":"coder"},"previous_digest":"dfd4e84746245b5f54ab6bc3d417718e2ea6034139e9127632c99dcce3e58a3f","recorded_at":"2026-07-27T15:30:48.000000Z","request_digest":"e14447b164d6303f9e835b1c86dd2cbdc8ed10e84e9052d9c2535d6d8f7e370d","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":4} +{"event_digest":"df1a01d42479bc4a0563399c37c6316fddcdf763a624b2e0c607358642ce71a7","event_id":"20260727-153045-a1b2c3d4-000005-df1a01d42479","event_type":"run.synthesis.completed","idempotency_key":"synthesis-done-1","payload":{"detail":"synthesized"},"previous_digest":"203f8fa371b3c02c631876e7a5c00cdf1918abffef82e2cb6a89154efaa49f5a","recorded_at":"2026-07-27T15:30:49.000000Z","request_digest":"8279c964012ded2db849b01b6b27eadcd1c7b665082d4c9755180d4d8f03fd67","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":5} +{"event_digest":"176259cd00408b637f1c13496fcdfafea7dd96753fb8a9e76dbea58817442fc7","event_id":"20260727-153045-a1b2c3d4-000006-176259cd0040","event_type":"run.completed","idempotency_key":"complete-1","payload":{"detail":"done","status":"ok"},"previous_digest":"df1a01d42479bc4a0563399c37c6316fddcdf763a624b2e0c607358642ce71a7","recorded_at":"2026-07-27T15:30:50.000000Z","request_digest":"ea353ac696d7c73e8eab3128b548e4fad328a64d5e2c381fe640822749ea8c83","run_id":"20260727-153045-a1b2c3d4","schema":"brigade.run_event.v1","schema_version":1,"sequence":6} diff --git a/tests/test_run_events.py b/tests/test_run_events.py new file mode 100644 index 00000000..43557226 --- /dev/null +++ b/tests/test_run_events.py @@ -0,0 +1,279 @@ +"""Tests for brigade.run_events canonicalization, digests, and envelope validation.""" + +from __future__ import annotations + +import hashlib +from datetime import datetime, timezone + +import pytest + +from brigade import run_events +from brigade.receipt_schema import RUN_EVENT_SCHEMA, RUN_EVENT_SCHEMA_VERSION + +RUN_ID = "20260727-153045-a1b2c3d4" +RECORDED_AT = "2026-07-27T15:30:45.123456Z" +IDEMPOTENCY_KEY = "create-1" + +# UTF-8 bytes for "café" (é = 0xC3 0xA9). The slice-1 contract requires +# canonical UTF-8 JSON with no ASCII escaping; a Latin-1 byte literal would +# contradict that contract and produce a different digest. +_CANONICAL_CAFE = b'{"a":1,"b":"caf\xc3\xa9"}' +_CANONICAL_CAFE_DIGEST = "f7a52e662c92fc838cc3e18c060954d6b27190e0051d76e139b8bab26a2923bd" + +_REQUEST = { + "event_type": "run.created", + "idempotency_key": IDEMPOTENCY_KEY, + "payload": {"status": "started"}, +} +_REQUEST_DIGEST = "3c8ed91a3781b4f7033bc1940ee95eaa50a88f33d453e2b9fb9142150b3f89cb" +# event_digest is the SHA-256 of the canonical envelope with BOTH event_digest +# and event_id excluded. event_id embeds event_digest[:12], so including +# event_id in the digest input would create an infeasible SHA-256 fixed point. +# Excluding event_id is the only computable reading of the slice-1 binding. +_EVENT_DIGEST = "6d43c3e25e4d4c8627166ab5b565622f98b0999218889b65ac8c84dcfac9301b" +_EVENT_ID = f"{RUN_ID}-000001-{_EVENT_DIGEST[:12]}" + +_FORBIDDEN_PAYLOAD_KEYS = ( + "prompt", + "prompt_text", + "model_output", + "tool_args", + "credentials", + "provider_response", + "stack_trace", +) + + +def _sample_envelope(*, event_digest: str = _EVENT_DIGEST, event_id: str = _EVENT_ID) -> dict: + return { + "schema": RUN_EVENT_SCHEMA, + "schema_version": RUN_EVENT_SCHEMA_VERSION, + "event_id": event_id, + "run_id": RUN_ID, + "sequence": 1, + "event_type": "run.created", + "recorded_at": RECORDED_AT, + "idempotency_key": IDEMPOTENCY_KEY, + "request_digest": _REQUEST_DIGEST, + "previous_digest": None, + "event_digest": event_digest, + "payload": {"status": "started"}, + } + + +def test_schema_constants_are_registered_in_receipt_schema(): + assert RUN_EVENT_SCHEMA == "brigade.run_event.v1" + assert RUN_EVENT_SCHEMA_VERSION == 1 + assert run_events.SCHEMA == RUN_EVENT_SCHEMA + assert run_events.SCHEMA_VERSION == RUN_EVENT_SCHEMA_VERSION + + +def test_canonical_bytes_matches_exact_utf8_compact_sorted_form(): + assert run_events.canonical_bytes({"a": 1, "b": "café"}) == _CANONICAL_CAFE + assert hashlib.sha256(_CANONICAL_CAFE).hexdigest() == _CANONICAL_CAFE_DIGEST + + +def test_canonical_bytes_rejects_floats_and_bools(): + with pytest.raises(run_events.CanonicalizationError): + run_events.canonical_bytes({"value": 1.5}) + with pytest.raises(run_events.CanonicalizationError): + run_events.canonical_bytes({"value": True}) + with pytest.raises(run_events.CanonicalizationError): + run_events.canonical_bytes({"value": False}) + + +def test_canonical_bytes_rejects_oversized_integers(): + with pytest.raises(run_events.CanonicalizationError): + run_events.canonical_bytes({"value": 2**63}) + with pytest.raises(run_events.CanonicalizationError): + run_events.canonical_bytes({"value": -(2**63) - 1}) + # The signed 64-bit boundaries themselves remain canonical. + assert run_events.canonical_bytes({"value": 2**63 - 1}) == b'{"value":9223372036854775807}' + assert run_events.canonical_bytes({"value": -(2**63)}) == b'{"value":-9223372036854775808}' + + +def test_canonical_bytes_wraps_conversion_errors_in_bounded_typed_error(): + cyclic: dict = {} + cyclic["self"] = cyclic + with pytest.raises(run_events.CanonicalizationError) as excinfo: + run_events.canonical_bytes(cyclic) + assert len(str(excinfo.value)) <= 240 + + +def test_validate_event_reports_oversized_integer_as_bounded_error(): + env = _sample_envelope() + env["sequence"] = 2**63 + errors = run_events.validate_event(env) + assert errors + assert all(len(err) <= 240 for err in errors) + + +def test_canonical_bytes_preserves_explicit_null(): + assert run_events.canonical_bytes({"optional": None}) == b'{"optional":null}' + + +def test_format_recorded_at_emits_exact_six_digit_utc_z(): + dt = datetime(2026, 7, 27, 15, 30, 45, 123456, tzinfo=timezone.utc) + assert run_events.format_recorded_at(dt) == RECORDED_AT + assert run_events.is_valid_recorded_at(RECORDED_AT) is True + + +@pytest.mark.parametrize( + "bad_ts", + [ + "2026-07-27T15:30:45Z", + "2026-07-27T15:30:45.12345Z", + "2026-07-27T15:30:45.1234567Z", + "2026-07-27T15:30:45.123456+00:00", + "not-a-timestamp", + ], +) +def test_is_valid_recorded_at_rejects_nonconforming_timestamps(bad_ts): + assert run_events.is_valid_recorded_at(bad_ts) is False + + +def test_request_digest_is_deterministic_for_known_request(): + digest = run_events.request_digest( + event_type=_REQUEST["event_type"], + payload=_REQUEST["payload"], + idempotency_key=_REQUEST["idempotency_key"], + ) + assert digest == _REQUEST_DIGEST + assert len(digest) == 64 + assert digest == digest.lower() + + +def test_compute_event_digest_excludes_event_digest_and_event_id_for_noncircular_binding(): + # event_id embeds event_digest[:12], so event_digest must exclude event_id + # (and event_digest itself) to keep the binding computable. + envelope = _sample_envelope() + assert run_events.compute_event_digest(envelope) == _EVENT_DIGEST + + +def test_make_event_id_is_deterministic(): + assert ( + run_events.make_event_id( + run_id=RUN_ID, + sequence=1, + event_digest=_EVENT_DIGEST, + ) + == _EVENT_ID + ) + + +def test_build_event_produces_byte_exact_known_envelope(): + event = run_events.build_event( + run_id=RUN_ID, + sequence=1, + event_type="run.created", + payload={"status": "started"}, + idempotency_key=IDEMPOTENCY_KEY, + recorded_at=RECORDED_AT, + previous_digest=None, + ) + assert event["request_digest"] == _REQUEST_DIGEST + assert event["event_digest"] == _EVENT_DIGEST + assert event["event_id"] == _EVENT_ID + line = run_events.canonical_bytes(event) + b"\n" + expected = run_events.canonical_bytes(_sample_envelope()) + b"\n" + assert line == expected + + +def test_validate_event_accepts_known_good_envelope(): + assert run_events.validate_event(_sample_envelope()) == [] + + +def test_validate_event_rejects_unknown_envelope_key_with_bounded_diagnostic(): + env = _sample_envelope() + env["surprise"] = "nope" + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + assert "surprise" in errors[0] + + +def test_validate_event_rejects_unknown_payload_key_with_bounded_diagnostic(): + env = _sample_envelope() + env["payload"] = {"status": "started", "extra": "nope"} + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + assert "extra" in errors[0] + + +def test_validate_event_rejects_unknown_schema_version(): + env = _sample_envelope() + env["schema_version"] = 99 + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + assert "schema_version" in errors[0] + + +def test_validate_event_rejects_unknown_schema_string(): + env = _sample_envelope() + env["schema"] = "brigade.run_event.v99" + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + assert "schema" in errors[0] + + +def test_validate_event_rejects_unknown_event_type(): + env = _sample_envelope() + env["event_type"] = "run.not-a-real-event" + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + assert "event_type" in errors[0] + + +@pytest.mark.parametrize("forbidden_key", _FORBIDDEN_PAYLOAD_KEYS) +def test_validate_event_rejects_private_data_payload_keys(forbidden_key): + env = _sample_envelope() + env["payload"] = {forbidden_key: "secret"} + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 + + +def test_event_type_registry_includes_run_created_with_status_only(): + assert "run.created" in run_events.EVENT_TYPES + assert run_events.EVENT_TYPES["run.created"] == frozenset({"status"}) + + +def test_build_event_rejects_payload_with_non_integer_numbers(): + with pytest.raises(run_events.CanonicalizationError): + run_events.build_event( + run_id=RUN_ID, + sequence=1, + event_type="run.dispatch.requested", + payload={"seat": "coder", "attempt": 1.0}, + idempotency_key="dispatch-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + + +def test_validate_event_requires_explicit_null_previous_digest_at_sequence_one(): + env = _sample_envelope() + del env["previous_digest"] + errors = run_events.validate_event(env) + assert errors + assert any("previous_digest" in err for err in errors) + + +def test_validate_event_recomputes_event_digest_mismatch(): + env = _sample_envelope(event_digest="0" * 64) + errors = run_events.validate_event(env) + assert errors + assert any("event_digest" in err for err in errors) + + +def test_validate_event_limits_unknown_key_diagnostics(): + env = _sample_envelope() + for idx in range(12): + env[f"unknown_{idx}"] = "x" + errors = run_events.validate_event(env) + assert errors + assert len(errors[0]) <= 240 diff --git a/tests/test_run_journal.py b/tests/test_run_journal.py new file mode 100644 index 00000000..5d64a55e --- /dev/null +++ b/tests/test_run_journal.py @@ -0,0 +1,932 @@ +"""Tests for brigade.run_journal append-only lifecycle journal kernel.""" + +from __future__ import annotations + +import errno +import hashlib +import json +import os +import stat +from copy import deepcopy +from pathlib import Path + +import pytest + +from brigade import run_events, run_journal + +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "run-lifecycle" +GOLDEN_LIFECYCLE_PATH = FIXTURES / "golden-lifecycle.jsonl" + +RUN_ID = "20260727-153045-a1b2c3d4" +RECORDED_AT = "2026-07-27T15:30:45.123456Z" + +_GOLDEN_APPEND_PLAN = ( + { + "event_type": "run.created", + "payload": {"status": "started"}, + "idempotency_key": "create-1", + "recorded_at": "2026-07-27T15:30:45.123456Z", + }, + { + "event_type": "run.planning.started", + "payload": {"detail": "planning"}, + "idempotency_key": "plan-start-1", + "recorded_at": "2026-07-27T15:30:46.000000Z", + }, + { + "event_type": "run.dispatch.requested", + "payload": {"seat": "coder", "attempt": 1}, + "idempotency_key": "dispatch-req-1", + "recorded_at": "2026-07-27T15:30:47.000000Z", + }, + { + "event_type": "run.dispatch.completed", + "payload": {"seat": "coder", "attempt": 1, "detail": "ok"}, + "idempotency_key": "dispatch-done-1", + "recorded_at": "2026-07-27T15:30:48.000000Z", + }, + { + "event_type": "run.synthesis.completed", + "payload": {"detail": "synthesized"}, + "idempotency_key": "synthesis-done-1", + "recorded_at": "2026-07-27T15:30:49.000000Z", + }, + { + "event_type": "run.completed", + "payload": {"status": "ok", "detail": "done"}, + "idempotency_key": "complete-1", + "recorded_at": "2026-07-27T15:30:50.000000Z", + }, +) + + +def _run_dir(tmp_path: Path) -> Path: + run_dir = tmp_path / "runs" / RUN_ID + run_dir.mkdir(parents=True) + return run_dir + + +def _journal_path(run_dir: Path) -> Path: + return run_dir / "events" / "lifecycle.jsonl" + + +def _append( + journal_path: Path, + *, + event_type: str, + payload: dict, + idempotency_key: str, + expected_previous_sequence: int, + recorded_at: str, +) -> run_journal.RunEvent: + return run_journal.append_event( + journal_path, + run_id=RUN_ID, + event_type=event_type, + payload=payload, + idempotency_key=idempotency_key, + expected_previous_sequence=expected_previous_sequence, + recorded_at=recorded_at, + ) + + +def _append_first_event(journal_path: Path) -> run_journal.RunEvent: + return _append( + journal_path, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="create-1", + expected_previous_sequence=0, + recorded_at=RECORDED_AT, + ) + + +def test_ensure_journal_creates_private_directory_and_file(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + run_journal.ensure_journal(journal_path) + + assert journal_path.is_file() + assert journal_path.parent.is_dir() + assert stat.S_IMODE(journal_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(journal_path.parent.stat().st_mode) == 0o700 + + +def test_append_event_writes_canonical_line_with_fsync(tmp_path, monkeypatch): + journal_path = _journal_path(_run_dir(tmp_path)) + fsync_calls: list[int] = [] + + def _track_fsync(fd): + fsync_calls.append(fd) + + monkeypatch.setattr(os, "fsync", _track_fsync) + + event = _append_first_event(journal_path) + + lines = journal_path.read_bytes().splitlines(keepends=True) + assert len(lines) == 1 + assert lines[0] == run_events.canonical_bytes(event.to_dict()) + b"\n" + assert fsync_calls + + +def test_append_event_rejects_partial_os_write(tmp_path, monkeypatch): + journal_path = _journal_path(_run_dir(tmp_path)) + original_write = os.write + + def _short_write(fd, data): + return original_write(fd, data[: max(1, len(data) // 2)]) + + monkeypatch.setattr(os, "write", _short_write) + + with pytest.raises(run_journal.PartialWriteError): + _append_first_event(journal_path) + + +def test_idempotent_replay_returns_existing_event_without_second_append(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + first = _append_first_event(journal_path) + before = journal_path.read_bytes() + + replay = _append( + journal_path, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="create-1", + expected_previous_sequence=1, + recorded_at=RECORDED_AT, + ) + + assert replay.event_id == first.event_id + assert replay.request_digest == first.request_digest + assert replay.event_digest == first.event_digest + assert journal_path.read_bytes() == before + + +def test_same_idempotency_key_different_digest_raises_without_mutation(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + before = journal_path.read_bytes() + + with pytest.raises(run_journal.IdempotencyConflict) as excinfo: + _append( + journal_path, + event_type="run.created", + payload={"status": "changed"}, + idempotency_key="create-1", + expected_previous_sequence=1, + recorded_at=RECORDED_AT, + ) + + conflict = excinfo.value + assert conflict.existing_event_id + assert conflict.request_digest != conflict.existing_request_digest + assert len(conflict.diagnostic) <= 240 + assert journal_path.read_bytes() == before + + +def test_stale_expected_previous_sequence_raises_without_append(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + before = journal_path.read_bytes() + + with pytest.raises(run_journal.StaleSequenceError) as excinfo: + _append( + journal_path, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + expected_previous_sequence=0, + recorded_at="2026-07-27T15:30:46.000000Z", + ) + + assert len(excinfo.value.diagnostic) <= 240 + assert journal_path.read_bytes() == before + + +def _huge_json_integer_line() -> bytes: + return ("1" + "0" * 4999).encode() + b"\n" + + +def test_append_fails_closed_on_huge_json_integer_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + journal_path.write_bytes(journal_path.read_bytes() + _huge_json_integer_line()) + before = journal_path.read_bytes() + + with pytest.raises(run_journal.ChainIntegrityError) as excinfo: + _append_second_event(journal_path) + + assert len(excinfo.value.diagnostic) <= 240 + assert "5000" not in excinfo.value.diagnostic + assert journal_path.read_bytes() == before + + +def test_read_journal_stops_at_huge_json_integer_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + first = _append_first_event(journal_path) + journal_path.write_bytes(journal_path.read_bytes() + _huge_json_integer_line()) + + report = run_journal.read_journal(journal_path) + + assert len(report.events) == 1 + assert report.events[0].event_id == first.event_id + assert len(report.chain_errors) == 1 + assert len(report.chain_errors[0]) <= 240 + assert "5000" not in report.chain_errors[0] + + +def test_read_journal_verified_prefix_stops_after_tampered_middle_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + first = _append_first_event(journal_path) + _append_second_event(journal_path) + _append( + journal_path, + event_type="run.dispatch.requested", + payload={"seat": "coder", "attempt": 1}, + idempotency_key="dispatch-req-1", + expected_previous_sequence=2, + recorded_at="2026-07-27T15:30:47.000000Z", + ) + + lines = journal_path.read_text().splitlines() + broken = json.loads(lines[1]) + broken["previous_digest"] = "0" * 64 + lines[1] = run_events.canonical_bytes(broken).decode("utf-8") + journal_path.write_text("\n".join(lines) + "\n") + + report = run_journal.read_journal(journal_path) + + assert len(report.events) == 1 + assert report.events[0].event_id == first.event_id + assert report.chain_errors + assert any("previous_digest" in err or "digest" in err for err in report.chain_errors) + + +def test_recover_partial_tail_raises_on_missing_journal(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + quarantine_dir = tmp_path / "quarantine" + + with pytest.raises(run_journal.RunJournalError) as excinfo: + run_journal.recover_partial_tail(journal_path, quarantine_dir) + + assert len(excinfo.value.diagnostic) <= 240 + assert journal_path.name in excinfo.value.diagnostic + + +def test_read_journal_detects_sequence_gap_without_mutation(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + first = _append_first_event(journal_path) + second = _append( + journal_path, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + expected_previous_sequence=1, + recorded_at="2026-07-27T15:30:46.000000Z", + ) + + lines = journal_path.read_text().splitlines() + gap_line = deepcopy(json.loads(lines[1])) + gap_line["sequence"] = 3 + gap_line["event_id"] = gap_line["event_id"].replace("-000002-", "-000003-") + lines[1] = run_events.canonical_bytes(gap_line).decode("utf-8") + journal_path.write_text("\n".join(lines) + "\n") + before = journal_path.read_bytes() + + report = run_journal.read_journal(journal_path) + + assert report.partial_tail is None + assert len(report.events) == 1 + assert report.events[0].event_id == first.event_id + assert report.chain_errors + assert any("sequence" in err for err in report.chain_errors) + assert journal_path.read_bytes() == before + assert first.event_id != second.event_id + + +def test_read_journal_detects_previous_digest_mismatch_without_mutation(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + _append( + journal_path, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + expected_previous_sequence=1, + recorded_at="2026-07-27T15:30:46.000000Z", + ) + + lines = journal_path.read_text().splitlines() + broken = json.loads(lines[1]) + broken["previous_digest"] = "0" * 64 + lines[1] = run_events.canonical_bytes(broken).decode("utf-8") + journal_path.write_text("\n".join(lines) + "\n") + before = journal_path.read_bytes() + + report = run_journal.read_journal(journal_path) + + assert len(report.events) == 1 + assert report.chain_errors + assert any("previous_digest" in err or "digest" in err for err in report.chain_errors) + assert journal_path.read_bytes() == before + + +def test_read_journal_reports_partial_final_line_without_mutation(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + partial_suffix = b'{"schema":"brigade.run_event.v1","event_type":"run.plan' + journal_path.write_bytes(journal_path.read_bytes() + partial_suffix) + before = journal_path.read_bytes() + + report = run_journal.read_journal(journal_path) + + assert report.partial_tail == partial_suffix + assert len(report.events) == 1 + assert journal_path.read_bytes() == before + + +def test_append_refuses_to_write_over_partial_tail_without_mutation(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + partial_suffix = b'{"schema":"brigade.run_event.v1","event_type":"run.plan' + journal_path.write_bytes(journal_path.read_bytes() + partial_suffix) + before = journal_path.read_bytes() + + with pytest.raises(run_journal.PartialTailError) as excinfo: + _append( + journal_path, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + expected_previous_sequence=1, + recorded_at="2026-07-27T15:30:46.000000Z", + ) + + assert len(excinfo.value.diagnostic) <= 240 + assert journal_path.read_bytes() == before + + +def test_recover_partial_tail_quarantines_suffix_then_truncates(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + partial_suffix = b'{"schema":"brigade.run_event.v1","event_type":"run.plan' + journal_path.write_bytes(journal_path.read_bytes() + partial_suffix) + complete_bytes = journal_path.read_bytes()[: -len(partial_suffix)] + quarantine_dir = tmp_path / "quarantine" + + report = run_journal.recover_partial_tail(journal_path, quarantine_dir) + + assert report.partial_bytes == partial_suffix + assert report.quarantine_path.is_file() + assert report.quarantine_path.read_bytes() == partial_suffix + assert stat.S_IMODE(report.quarantine_path.stat().st_mode) == 0o600 + assert journal_path.read_bytes() == complete_bytes + + +def test_golden_lifecycle_journal_matches_fixture_bytes(tmp_path): + if not GOLDEN_LIFECYCLE_PATH.is_file(): + pytest.fail(f"missing golden fixture: {GOLDEN_LIFECYCLE_PATH}") + + golden_bytes = GOLDEN_LIFECYCLE_PATH.read_bytes() + journal_path = _journal_path(_run_dir(tmp_path)) + expected_previous_sequence = 0 + + for step in _GOLDEN_APPEND_PLAN: + _append( + journal_path, + event_type=step["event_type"], + payload=step["payload"], + idempotency_key=step["idempotency_key"], + expected_previous_sequence=expected_previous_sequence, + recorded_at=step["recorded_at"], + ) + expected_previous_sequence += 1 + + assert journal_path.read_bytes() == golden_bytes + + report = run_journal.read_journal(journal_path) + assert report.partial_tail is None + assert report.chain_errors == [] + assert len(report.events) == len(_GOLDEN_APPEND_PLAN) + + +def test_read_journal_rejects_unknown_envelope_fields_in_complete_lines(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + event = _append_first_event(journal_path) + tampered = event.to_dict() + tampered["unexpected"] = "field" + journal_path.write_text(run_events.canonical_bytes(tampered).decode("utf-8") + "\n") + + report = run_journal.read_journal(journal_path) + + assert report.chain_errors + assert len(report.chain_errors[0]) <= 240 + assert "unexpected" in report.chain_errors[0] + + +def _write_journal_lines(journal_path: Path, *envelopes: dict) -> None: + journal_path.parent.mkdir(parents=True, exist_ok=True) + journal_path.write_bytes(b"".join(run_events.canonical_bytes(env) + b"\n" for env in envelopes)) + + +def _build_envelope( + *, sequence: int, event_type: str, payload: dict, idempotency_key: str, recorded_at: str, previous_digest +) -> dict: + return run_events.build_event( + run_id=RUN_ID, + sequence=sequence, + event_type=event_type, + payload=payload, + idempotency_key=idempotency_key, + recorded_at=recorded_at, + previous_digest=previous_digest, + ) + + +def _append_second_event(journal_path: Path, *, expected_previous_sequence: int = 1) -> run_journal.RunEvent: + return _append( + journal_path, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + expected_previous_sequence=expected_previous_sequence, + recorded_at="2026-07-27T15:30:46.000000Z", + ) + + +def test_append_fails_closed_on_reordered_complete_lines(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + _append_second_event(journal_path) + lines = journal_path.read_bytes().splitlines() + journal_path.write_bytes(lines[1] + b"\n" + lines[0] + b"\n") + before = journal_path.read_bytes() + + with pytest.raises(run_journal.ChainIntegrityError): + _append( + journal_path, + event_type="run.dispatch.requested", + payload={"seat": "coder", "attempt": 1}, + idempotency_key="dispatch-req-1", + expected_previous_sequence=2, + recorded_at="2026-07-27T15:30:47.000000Z", + ) + assert journal_path.read_bytes() == before + + +def test_append_fails_closed_on_duplicated_complete_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + journal_path.write_bytes(line + b"\n" + line + b"\n") + before = journal_path.read_bytes() + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path, expected_previous_sequence=2) + assert journal_path.read_bytes() == before + + +def test_append_fails_closed_on_gapped_sequence(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + env1 = _build_envelope( + sequence=1, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="create-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + env3 = _build_envelope( + sequence=3, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + recorded_at="2026-07-27T15:30:46.000000Z", + previous_digest=env1["event_digest"], + ) + _write_journal_lines(journal_path, env1, env3) + before = journal_path.read_bytes() + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path, expected_previous_sequence=3) + assert journal_path.read_bytes() == before + + +def test_append_fails_closed_on_broken_previous_digest_link(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + env1 = _build_envelope( + sequence=1, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="create-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + env2 = _build_envelope( + sequence=2, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="plan-start-1", + recorded_at="2026-07-27T15:30:46.000000Z", + previous_digest="0" * 64, + ) + _write_journal_lines(journal_path, env1, env2) + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path, expected_previous_sequence=2) + + +def test_append_fails_closed_on_digest_invalid_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + env = json.loads(line.decode("utf-8")) + env["event_digest"] = "0" * 64 + journal_path.write_bytes(run_events.canonical_bytes(env) + b"\n") + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path) + + +def test_append_fails_closed_on_repeated_idempotency_key(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + env1 = _build_envelope( + sequence=1, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="dup-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + env2 = _build_envelope( + sequence=2, + event_type="run.planning.started", + payload={"detail": "planning"}, + idempotency_key="dup-1", + recorded_at="2026-07-27T15:30:46.000000Z", + previous_digest=env1["event_digest"], + ) + _write_journal_lines(journal_path, env1, env2) + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path, expected_previous_sequence=2) + + +@pytest.mark.parametrize( + "missing_field", + ["event_id", "event_digest", "idempotency_key", "request_digest", "sequence", "previous_digest"], +) +def test_append_fails_closed_with_typed_error_on_envelope_missing_field(tmp_path, missing_field): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + env = json.loads(line.decode("utf-8")) + del env[missing_field] + journal_path.write_bytes( + json.dumps(env, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + b"\n" + ) + before = journal_path.read_bytes() + + # RunJournalError (not KeyError) proves malformed fields become bounded + # typed errors rather than trusted RunEvent materialization. + with pytest.raises(run_journal.RunJournalError): + _append_second_event(journal_path) + assert journal_path.read_bytes() == before + + +def test_append_fails_closed_on_invalid_idempotency_entry(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + env = json.loads(line.decode("utf-8")) + env["idempotency_key"] = 42 + journal_path.write_bytes( + json.dumps(env, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + b"\n" + ) + + with pytest.raises(run_journal.RunJournalError): + _append_second_event(journal_path) + + +def test_append_fails_closed_on_duplicate_json_keys(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + journal_path.parent.mkdir(parents=True, exist_ok=True) + journal_path.write_bytes(b'{"schema":"brigade.run_event.v1","schema":"brigade.run_event.v1","sequence":1}\n') + + with pytest.raises(run_journal.ChainIntegrityError) as excinfo: + _append_first_event(journal_path) + assert "schema" in excinfo.value.diagnostic + + +def test_append_fails_closed_on_whitespace_noncanonical_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + spaced = json.dumps(json.loads(line.decode("utf-8")), separators=(", ", ": "), ensure_ascii=False).encode("utf-8") + assert spaced != line + journal_path.write_bytes(spaced + b"\n") + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path) + + +def test_append_fails_closed_on_ascii_escaped_noncanonical_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append( + journal_path, + event_type="run.created", + payload={"status": "café"}, + idempotency_key="create-1", + expected_previous_sequence=0, + recorded_at=RECORDED_AT, + ) + line = journal_path.read_bytes().splitlines()[0] + escaped = json.dumps( + json.loads(line.decode("utf-8")), sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + assert escaped != line + journal_path.write_bytes(escaped + b"\n") + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path) + + +def test_append_fails_closed_on_unsorted_key_order(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + env = json.loads(line.decode("utf-8")) + reversed_env = dict(reversed(list(env.items()))) + unsorted = json.dumps(reversed_env, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + assert unsorted != line + journal_path.write_bytes(unsorted + b"\n") + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path) + + +def test_append_fails_closed_on_oversized_integer_line(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + env = json.loads(line.decode("utf-8")) + env["sequence"] = 2**63 + journal_path.write_bytes( + json.dumps(env, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + b"\n" + ) + + with pytest.raises(run_journal.ChainIntegrityError): + _append_second_event(journal_path) + + +def test_read_journal_reports_noncanonical_lines_as_chain_errors(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + line = journal_path.read_bytes().splitlines()[0] + spaced = json.dumps(json.loads(line.decode("utf-8")), separators=(", ", ": "), ensure_ascii=False).encode("utf-8") + journal_path.write_bytes(spaced + b"\n") + + report = run_journal.read_journal(journal_path) + + assert report.chain_errors + assert all(len(err) <= 240 for err in report.chain_errors) + assert report.events == [] + + +def test_ensure_journal_modes_hold_under_permissive_umask(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + previous_umask = os.umask(0) + try: + run_journal.ensure_journal(journal_path) + finally: + os.umask(previous_umask) + + assert stat.S_IMODE(journal_path.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(journal_path.stat().st_mode) == 0o600 + + +def test_ensure_journal_corrects_permissive_preexisting_modes(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + parent = journal_path.parent + parent.mkdir(parents=True) + os.chmod(parent, 0o777) + fd = os.open(journal_path, os.O_CREAT | os.O_WRONLY, 0o644) + os.close(fd) + os.chmod(journal_path, 0o644) + + run_journal.ensure_journal(journal_path) + + assert stat.S_IMODE(parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(journal_path.stat().st_mode) == 0o600 + + +def test_ensure_journal_rejects_symlinked_events_directory(tmp_path): + run_dir = _run_dir(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (run_dir / "events").symlink_to(outside, target_is_directory=True) + journal_path = run_dir / "events" / "lifecycle.jsonl" + + with pytest.raises(run_journal.RunJournalError): + run_journal.ensure_journal(journal_path) + assert not (outside / "lifecycle.jsonl").exists() + + +def test_append_rejects_symlinked_journal_file(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + run_journal.ensure_journal(journal_path) + journal_path.unlink() + target = tmp_path / "target.jsonl" + target.write_bytes(b"") + journal_path.symlink_to(target) + + with pytest.raises(run_journal.RunJournalError): + _append_first_event(journal_path) + assert target.read_bytes() == b"" + + +def test_append_rejects_dangling_symlinked_journal_file(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + journal_path.parent.mkdir(parents=True) + journal_path.symlink_to(tmp_path / "missing-target") + + with pytest.raises(run_journal.RunJournalError): + _append_first_event(journal_path) + + +def test_recover_partial_tail_creates_private_quarantine_dir(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + journal_path.write_bytes(journal_path.read_bytes() + b"partial") + quarantine_dir = tmp_path / "quarantine" + + report = run_journal.recover_partial_tail(journal_path, quarantine_dir) + + assert stat.S_IMODE(quarantine_dir.stat().st_mode) == 0o700 + assert report.quarantine_path is not None + + +def test_recover_partial_tail_retries_on_quarantine_name_collision(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + partial_suffix = b'{"schema":"brigade.run_event.v1","event_type":"run.plan' + journal_path.write_bytes(journal_path.read_bytes() + partial_suffix) + quarantine_dir = tmp_path / "quarantine" + quarantine_dir.mkdir() + digest = hashlib.sha256(partial_suffix).hexdigest() + occupant = quarantine_dir / f"lifecycle-partial-000001-{digest}.bin" + occupant.write_bytes(b"occupant") + + report = run_journal.recover_partial_tail(journal_path, quarantine_dir) + + assert report.quarantine_path is not None + assert report.quarantine_path != occupant + assert report.quarantine_path.name.startswith(f"lifecycle-partial-000001-{digest}-") + assert report.quarantine_path.read_bytes() == partial_suffix + # Write-once: the pre-existing occupant was never overwritten. + assert occupant.read_bytes() == b"occupant" + + +def test_recover_partial_tail_returns_none_quarantine_path_without_partial(tmp_path): + journal_path = _journal_path(_run_dir(tmp_path)) + _append_first_event(journal_path) + before = journal_path.read_bytes() + + report = run_journal.recover_partial_tail(journal_path, tmp_path / "quarantine") + + assert report.partial_bytes == b"" + assert report.quarantine_path is None + assert journal_path.read_bytes() == before + + +def _disable_posix_open_guards(monkeypatch) -> None: + """Simulate hosts without O_NOFOLLOW, O_DIRECTORY, or fchmod (e.g. Windows).""" + monkeypatch.setattr(run_journal, "_HAS_O_NOFOLLOW", False) + monkeypatch.setattr(run_journal, "_HAS_O_DIRECTORY", False) + monkeypatch.setattr(run_journal, "_HAS_FCHMOD", False) + monkeypatch.setattr(run_journal, "_OPEN_FLAGS", os.O_WRONLY | os.O_CREAT | os.O_APPEND) + + +def test_module_constants_use_getattr_for_import_safe_access(): + """Module-level flags are derived via getattr so import succeeds without POSIX APIs.""" + assert isinstance(run_journal._HAS_O_NOFOLLOW, bool) + assert isinstance(run_journal._HAS_O_DIRECTORY, bool) + assert isinstance(run_journal._HAS_FCHMOD, bool) + assert isinstance(run_journal._O_NOFOLLOW, int) + assert isinstance(run_journal._O_DIRECTORY, int) + if run_journal._HAS_O_NOFOLLOW: + assert run_journal._OPEN_FLAGS & run_journal._O_NOFOLLOW + else: + assert not (run_journal._OPEN_FLAGS & getattr(os, "O_NOFOLLOW", 0)) + + +def test_fallback_ensure_journal_creates_private_directory_and_file(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + journal_path = _journal_path(_run_dir(tmp_path)) + + run_journal.ensure_journal(journal_path) + + assert journal_path.is_file() + assert journal_path.parent.is_dir() + assert stat.S_IMODE(journal_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(journal_path.parent.stat().st_mode) == 0o700 + + +def test_fallback_append_read_and_recover_partial_tail(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + journal_path = _journal_path(_run_dir(tmp_path)) + + event = _append_first_event(journal_path) + report = run_journal.read_journal(journal_path) + assert len(report.events) == 1 + assert report.events[0].event_id == event.event_id + + partial_suffix = b'{"schema":"brigade.run_event.v1","event_type":"run.plan' + journal_path.write_bytes(journal_path.read_bytes() + partial_suffix) + complete_bytes = journal_path.read_bytes()[: -len(partial_suffix)] + + recovery = run_journal.recover_partial_tail(journal_path, tmp_path / "quarantine") + assert recovery.partial_bytes == partial_suffix + assert recovery.quarantine_path is not None + assert recovery.quarantine_path.read_bytes() == partial_suffix + assert journal_path.read_bytes() == complete_bytes + + +def test_fallback_rejects_symlinked_events_directory(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + run_dir = _run_dir(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (run_dir / "events").symlink_to(outside, target_is_directory=True) + journal_path = run_dir / "events" / "lifecycle.jsonl" + + with pytest.raises(run_journal.RunJournalError) as excinfo: + run_journal.ensure_journal(journal_path) + + assert len(excinfo.value.diagnostic) <= 240 + assert not (outside / "lifecycle.jsonl").exists() + + +def test_fallback_rejects_symlinked_journal_file(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + journal_path = _journal_path(_run_dir(tmp_path)) + run_journal.ensure_journal(journal_path) + journal_path.unlink() + target = tmp_path / "target.jsonl" + target.write_bytes(b"") + journal_path.symlink_to(target) + + with pytest.raises(run_journal.RunJournalError) as excinfo: + _append_first_event(journal_path) + + assert len(excinfo.value.diagnostic) <= 240 + assert target.read_bytes() == b"" + + +def test_fallback_open_nofollow_closes_fd_when_verify_identity_raises(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + journal_path = _journal_path(_run_dir(tmp_path)) + opened_fds: list[int] = [] + real_open = os.open + + def tracking_open(path, flags, mode=0o777, *, dir_fd=None): + fd = real_open(path, flags, mode, dir_fd=dir_fd) if dir_fd is not None else real_open(path, flags, mode) + opened_fds.append(fd) + return fd + + monkeypatch.setattr(os, "open", tracking_open) + + def fail_verify(_path, _fd): + raise run_journal.RunJournalError("verify failed") + + monkeypatch.setattr(run_journal, "_verify_fd_identity", fail_verify) + + with pytest.raises(run_journal.RunJournalError, match="verify failed"): + run_journal.ensure_journal(journal_path) + + assert len(opened_fds) == 1 + with pytest.raises(OSError) as excinfo: + os.fstat(opened_fds[0]) + assert excinfo.value.errno == errno.EBADF + + +def test_fallback_enforce_dir_mode_skips_open_without_o_directory(tmp_path, monkeypatch): + _disable_posix_open_guards(monkeypatch) + journal_path = _journal_path(_run_dir(tmp_path)) + events_dir = journal_path.parent + real_open = os.open + + def guard_open(path, flags, mode=0o777, *, dir_fd=None): + if Path(path) == events_dir: + raise AssertionError("os.open must not be called for the events directory without O_DIRECTORY") + if dir_fd is not None: + return real_open(path, flags, mode, dir_fd=dir_fd) + return real_open(path, flags, mode) + + monkeypatch.setattr(os, "open", guard_open) + events_dir.mkdir(parents=True, exist_ok=True) + os.chmod(events_dir, 0o755) + + run_journal.ensure_journal(journal_path) + + assert journal_path.is_file() + assert stat.S_IMODE(events_dir.stat().st_mode) == 0o700