From a823d7ee2d6594a4cd053a4c56ddeb623568ae66 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 23:37:29 -0400 Subject: [PATCH 1/2] feat(runs): add durable cursors and idempotent live control Wire issue #604 on the lifecycle journal: opaque event cursors for `runs events`, closed control.* event types, and request-id idempotency for app-server steer/interrupt without storing steering text. Co-authored-by: Cursor --- src/brigade/cli/runs.py | 120 +++- src/brigade/run_control_journal.py | 463 +++++++++++++ src/brigade/run_event_cursor.py | 311 +++++++++ src/brigade/run_events.py | 6 + src/brigade/run_projector.py | 4 + src/brigade/runs_cmd.py | 126 ++++ tests/test_run_event_cursor.py | 825 ++++++++++++++++++++++++ tests/test_run_events.py | 56 ++ tests/test_run_events_cursor_control.py | 573 ++++++++++++++++ tests/test_run_projector.py | 37 ++ 10 files changed, 2515 insertions(+), 6 deletions(-) create mode 100644 src/brigade/run_control_journal.py create mode 100644 src/brigade/run_event_cursor.py create mode 100644 tests/test_run_event_cursor.py create mode 100644 tests/test_run_events_cursor_control.py diff --git a/src/brigade/cli/runs.py b/src/brigade/cli/runs.py index 475ef973..ccc5c205 100644 --- a/src/brigade/cli/runs.py +++ b/src/brigade/cli/runs.py @@ -62,6 +62,45 @@ def register(sub: argparse._SubParsersAction) -> None: default=1.0, help="Polling interval in seconds.", ) + p_runs_events = runs_sub.add_parser( + "events", + help="Read verified lifecycle journal events with durable cursors.", + ) + p_runs_events.add_argument("run", help="Run directory path, run id under --runs-dir, or 'latest'.") + p_runs_events.add_argument( + "--cwd", + type=Path, + default=Path("."), + help="Workspace whose default .brigade/runs directory should be used for run ids.", + ) + p_runs_events.add_argument( + "--runs-dir", + type=Path, + default=None, + help="Explicit runs directory for run ids. Defaults to .brigade/runs under --cwd.", + ) + p_runs_events.add_argument( + "--after", + default=None, + help="Opaque cursor; emit only committed lifecycle events after this position.", + ) + p_runs_events.add_argument( + "--follow", + action="store_true", + help="Wait for appended lifecycle events and exit after a terminal event.", + ) + p_runs_events.add_argument( + "--json", + action="store_true", + default=True, + help="Emit newline-delimited JSON records (default).", + ) + p_runs_events.add_argument( + "--interval", + type=float, + default=1.0, + help="Polling interval in seconds when --follow is set.", + ) p_runs_steer = runs_sub.add_parser("steer", help="Send steering text to an active app-server worker turn.") p_runs_steer.add_argument("run", help="Run directory path, run id under --runs-dir, or 'latest'.") p_runs_steer.add_argument("worker", help="Worker name to steer.") @@ -78,6 +117,11 @@ def register(sub: argparse._SubParsersAction) -> None: default=None, help="Explicit runs directory for run ids. Defaults to .brigade/runs under --cwd.", ) + p_runs_steer.add_argument( + "--request-id", + default=None, + help="Caller request identity for idempotent live control (generated when omitted).", + ) p_runs_interrupt = runs_sub.add_parser("interrupt", help="Interrupt active app-server worker turns.") p_runs_interrupt.add_argument("run", help="Run directory path, run id under --runs-dir, or 'latest'.") p_runs_interrupt.add_argument("worker", nargs="?", default=None, help="Optional worker name to interrupt.") @@ -93,6 +137,11 @@ def register(sub: argparse._SubParsersAction) -> None: default=None, help="Explicit runs directory for run ids. Defaults to .brigade/runs under --cwd.", ) + p_runs_interrupt.add_argument( + "--request-id", + default=None, + help="Caller request identity for idempotent live control (generated when omitted).", + ) p_runs_recover = runs_sub.add_parser( "recover", help="Recover a nonterminal run whose recorded owner process has exited.", @@ -170,18 +219,34 @@ def dispatch(args) -> int: json_output=args.json, interval=args.interval, ) + if args.runs_command == "events": + return runs_cmd.events( + args.run, + cwd=args.cwd, + runs_dir=args.runs_dir, + after=args.after, + follow=args.follow, + interval=args.interval, + ) if args.runs_command == "steer": return _control_request( args.run, cwd=args.cwd, runs_dir=args.runs_dir, payload={"op": "steer", "worker": args.worker, "text": " ".join(args.text)}, + request_id=args.request_id, ) if args.runs_command == "interrupt": payload = {"op": "interrupt"} if args.worker is not None: payload["worker"] = args.worker - return _control_request(args.run, cwd=args.cwd, runs_dir=args.runs_dir, payload=payload) + return _control_request( + args.run, + cwd=args.cwd, + runs_dir=args.runs_dir, + payload=payload, + request_id=args.request_id, + ) if args.runs_command == "recover": return runs_cmd.recover(args.run, cwd=args.cwd, runs_dir=args.runs_dir) if args.runs_command == "redact": @@ -201,20 +266,63 @@ def dispatch(args) -> int: return 2 -def _control_request(run: str, *, cwd: Path, runs_dir: Path | None, payload: Mapping[str, object]) -> int: +def _control_request( + run: str, + *, + cwd: Path, + runs_dir: Path | None, + payload: Mapping[str, object], + request_id: str | None = None, +) -> int: import sys - from .. import run_control, runs_cmd + from .. import run_control, run_control_journal, runs_cmd run_dir, error = runs_cmd._resolve_run_dir(run, cwd=cwd, runs_dir=runs_dir) if error is not None: print(error, file=sys.stderr) return 2 assert run_dir is not None + + # Legacy runs without a lifecycle journal keep the pre-#604 transport path. + # An explicit --request-id cannot be honored without a journal. + if not run_control_journal.journal_present(run_dir): + if request_id is not None: + print("error: legacy-no-journal", file=sys.stderr) + return 2 + try: + transport = run_control.control_transport_from_run(run_dir) + response = run_control.send_request_with_retry(run_dir, transport, dict(payload)) + except run_control.ControlError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + return run_control.print_control_response(response, op=str(payload["op"])) + + op = str(payload["op"]) + worker = payload.get("worker") + text = payload.get("text") try: - socket_path = run_control.control_transport_from_run(run_dir) - response = run_control.send_request_with_retry(run_dir, socket_path, dict(payload)) + transport = run_control.control_transport_from_run(run_dir) + + def _send(transport_payload: dict[str, object]) -> dict[str, object]: + return run_control.send_request_with_retry(run_dir, transport, transport_payload) + + result = run_control_journal.execute_control_request( + run_dir, + op=op, + request_id=request_id, + worker=str(worker) if isinstance(worker, str) else None, + text=str(text) if isinstance(text, str) else None, + send=_send, + ) + except run_control_journal.ControlJournalError as exc: + print(f"error: {exc.code}: {exc}", file=sys.stderr) + return 2 except run_control.ControlError as exc: print(f"error: {exc}", file=sys.stderr) return 2 - return run_control.print_control_response(response, op=str(payload["op"])) + + print(f"request_id: {result.request_id}") + if result.replayed: + print("control: replay") + return run_control.print_control_response(result.response, op=op) diff --git a/src/brigade/run_control_journal.py b/src/brigade/run_control_journal.py new file mode 100644 index 00000000..f0b89137 --- /dev/null +++ b/src/brigade/run_control_journal.py @@ -0,0 +1,463 @@ +"""Idempotent live-control requests backed by the lifecycle journal (issue #604). + +Wraps app-server steer/interrupt with a caller request identity: + +1. Append and fsync ``control.requested`` before contacting the transport. +2. Append ``control.observed`` or ``control.failed`` after the transport returns. +3. Reuse of the same request id with the same fingerprint returns the committed + terminal result without a second transport call. +4. Reuse with a different fingerprint is a conflict. +5. ``control.requested`` without a terminal observation is ``indeterminate``; + the transport is not reissued automatically. + +Steering text never enters the journal — only a SHA-256 digest. External CLI +callers do not hold the run lock, so this module uses ``append_event`` directly +(status-neutral; no checkpoint pairing) and retries on ``StaleSequenceError``. +""" + +from __future__ import annotations + +import hashlib +import json +import secrets +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Mapping + +from brigade import run_events, run_journal, run_lifecycle, run_shadow +from brigade.run_events import MAX_DIAGNOSTIC_LEN, MAX_PAYLOAD_STR_LEN + +CONTROL_REQUESTED = "control.requested" +CONTROL_OBSERVED = "control.observed" +CONTROL_FAILED = "control.failed" +_TERMINAL_CONTROL_TYPES = frozenset({CONTROL_OBSERVED, CONTROL_FAILED}) +_MAX_STALE_RETRIES = 32 +_REQUEST_ID_HEX_BYTES = 8 + + +class ControlJournalError(RuntimeError): + """Bounded control-journal failure with a stable ``code``.""" + + def __init__(self, message: str, *, code: str) -> None: + super().__init__(message) + self.code = code + self.diagnostic = ( + message if len(message) <= MAX_DIAGNOSTIC_LEN else message[: MAX_DIAGNOSTIC_LEN - 1] + "\u2026" + ) + + +@dataclass(frozen=True) +class ControlResult: + """Outcome of an idempotent control request.""" + + request_id: str + response: dict[str, Any] + replayed: bool + requested_event: run_journal.RunEvent | None + terminal_event: run_journal.RunEvent | None + + +def generate_request_id() -> str: + """Return a new opaque request id suitable for ``--request-id``.""" + return secrets.token_hex(_REQUEST_ID_HEX_BYTES) + + +def text_digest(text: str) -> str: + """SHA-256 hex digest of steering text (never store the text itself).""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def control_fingerprint_payload( + *, + op: str, + request_id: str, + worker: str | None = None, + text: str | None = None, + turn_id: str | None = None, +) -> dict[str, Any]: + """Build the closed ``control.requested`` payload (reference fields only).""" + if op not in ("steer", "interrupt"): + raise ControlJournalError("unsupported control op", code="control_op_invalid") + if not isinstance(request_id, str) or not request_id: + raise ControlJournalError("request_id must be a non-empty string", code="request_id_invalid") + if len(request_id) > MAX_PAYLOAD_STR_LEN: + raise ControlJournalError("request_id exceeds payload string bound", code="request_id_invalid") + payload: dict[str, Any] = { + "op": op, + "request_id": request_id, + "text_digest": text_digest(text) if text is not None else "", + } + if worker is not None: + payload["worker"] = worker + if turn_id is not None: + payload["turn_id"] = turn_id + return payload + + +def requested_idempotency_key(request_id: str) -> str: + return f"control.requested:{request_id}" + + +def observed_idempotency_key(request_id: str) -> str: + return f"control.observed:{request_id}" + + +def failed_idempotency_key(request_id: str) -> str: + return f"control.failed:{request_id}" + + +def _bound(msg: str) -> str: + if len(msg) <= MAX_DIAGNOSTIC_LEN: + return msg + return msg[: MAX_DIAGNOSTIC_LEN - 1] + "\u2026" + + +def _journal_path(run_dir: Path) -> Path: + return run_lifecycle._journal_path(run_dir) + + +def _run_id(run_dir: Path) -> str: + return run_lifecycle._run_id_from_dir(run_dir) + + +def journal_present(run_dir: Path) -> bool: + """True when the run has a regular lifecycle journal file.""" + path = _journal_path(run_dir) + try: + return path.is_file() and not path.is_symlink() + except OSError: + return False + + +def _read_verified(journal_path: Path) -> list[run_journal.RunEvent]: + report = run_journal.read_journal_bounded(journal_path) + if report.partial_tail is not None: + raise ControlJournalError( + _bound("lifecycle journal ends in a partial line"), + code="journal_partial_tail", + ) + if report.chain_errors: + raise ControlJournalError( + _bound(f"lifecycle journal chain error: {report.chain_errors[0]}"), + code="journal_chain_error", + ) + return list(report.events) + + +def _find_control_pair( + events: list[run_journal.RunEvent], + *, + request_id: str, +) -> tuple[run_journal.RunEvent | None, run_journal.RunEvent | None]: + requested: run_journal.RunEvent | None = None + terminal: run_journal.RunEvent | None = None + for event in events: + payload = event.payload + if not isinstance(payload, Mapping): + continue + if payload.get("request_id") != request_id: + continue + if event.event_type == CONTROL_REQUESTED: + requested = event + elif event.event_type in _TERMINAL_CONTROL_TYPES: + terminal = event + return requested, terminal + + +def _response_from_terminal(event: run_journal.RunEvent) -> dict[str, Any]: + payload = event.payload + if event.event_type == CONTROL_OBSERVED: + response: dict[str, Any] = {"ok": True, "op": payload.get("op")} + if isinstance(payload.get("worker"), str): + response["worker"] = payload["worker"] + if isinstance(payload.get("turn_id"), str): + response["turn_id"] = payload["turn_id"] + detail = payload.get("detail") + if isinstance(detail, str) and detail.startswith("interrupted="): + try: + response["interrupted"] = int(detail.split("=", 1)[1]) + except ValueError: + response["detail"] = detail + return response + response = { + "ok": False, + "op": payload.get("op"), + "error": payload.get("detail") or "control request failed", + } + if isinstance(payload.get("code"), str): + response["code"] = payload["code"] + if isinstance(payload.get("worker"), str): + response["worker"] = payload["worker"] + return response + + +def _load_run_snapshot(run_dir: Path) -> dict[str, Any] | None: + try: + raw = (run_dir / "run.json").read_bytes() + except OSError: + return None + try: + payload = json.loads(raw) + except (ValueError, UnicodeDecodeError): + return None + return payload if isinstance(payload, dict) else None + + +def _sync_shadow_after_control(run_dir: Path) -> None: + """Advance shadow evidence to the new journal tail after a control append. + + Authoritative runs gate on shadow ``last_compared_*`` matching the journal + tail. Status-neutral control events must not leave ``journal-ahead-of-evidence`` + for the run owner. Mirrors the post-append ``record_shadow_comparison`` call + used by dispatch facts. + """ + snapshot = _load_run_snapshot(run_dir) + if snapshot is None: + return + run_shadow.record_shadow_comparison(run_dir, snapshot) + + +def _append_control_event( + journal_path: Path, + *, + run_dir: Path, + run_id: str, + event_type: str, + payload: dict[str, Any], + idempotency_key: str, +) -> run_journal.RunEvent: + """Append a control event, retrying on concurrent tail races. + + Holds ``checkpoint_event_pair`` around the append + shadow sync so a control + line cannot land between an owner's checkpoint and its paired event. + """ + last_error: Exception | None = None + for _ in range(_MAX_STALE_RETRIES): + try: + with run_lifecycle.checkpoint_event_pair(): + events = _read_verified(journal_path) + expected = events[-1].sequence if events else 0 + try: + event = run_journal.append_event( + journal_path, + run_id=run_id, + event_type=event_type, + payload=payload, + idempotency_key=idempotency_key, + expected_previous_sequence=expected, + ) + except run_journal.StaleSequenceError as exc: + last_error = exc + continue + except run_journal.IdempotencyConflict as exc: + raise ControlJournalError( + _bound(exc.diagnostic), + code="control_fingerprint_conflict", + ) from exc + except run_events.CanonicalizationError as exc: + raise ControlJournalError(_bound(str(exc)), code="control_payload_invalid") from exc + _sync_shadow_after_control(run_dir) + return event + except run_lifecycle.LifecycleJournalError as exc: + raise ControlJournalError(_bound(str(exc)), code="control_append_failed") from exc + raise ControlJournalError( + _bound(f"control append raced the journal tail: {last_error}"), + code="control_append_race", + ) + + +def inspect_control_request( + run_dir: Path, + *, + request_id: str, + fingerprint: Mapping[str, Any], +) -> tuple[str, run_journal.RunEvent | None, run_journal.RunEvent | None]: + """Classify a prior control request: ``absent``, ``replay``, ``indeterminate``, or raise conflict.""" + journal_path = _journal_path(run_dir) + events = _read_verified(journal_path) + requested, terminal = _find_control_pair(events, request_id=request_id) + if requested is None: + return "absent", None, None + expected_digest = run_events.request_digest( + event_type=CONTROL_REQUESTED, + payload=dict(fingerprint), + idempotency_key=requested_idempotency_key(request_id), + ) + if requested.request_digest != expected_digest: + raise ControlJournalError( + _bound("request_id reused with a different control fingerprint"), + code="control_fingerprint_conflict", + ) + if terminal is None: + return "indeterminate", requested, None + return "replay", requested, terminal + + +def execute_control_request( + run_dir: Path, + *, + op: str, + request_id: str | None = None, + worker: str | None = None, + text: str | None = None, + turn_id: str | None = None, + send: Callable[[dict[str, object]], dict[str, Any]], +) -> ControlResult: + """Execute an idempotent control request against a journal-backed run. + + ``send`` is the transport callable (typically a lambda around + ``send_request_with_retry``). It is not invoked on replay or indeterminate. + """ + run_dir = Path(run_dir) + if not journal_present(run_dir): + raise ControlJournalError("legacy-no-journal", code="legacy-no-journal") + + resolved_id = request_id if request_id is not None else generate_request_id() + fingerprint = control_fingerprint_payload( + op=op, + request_id=resolved_id, + worker=worker, + text=text, + turn_id=turn_id, + ) + journal_path = _journal_path(run_dir) + run_id = _run_id(run_dir) + + state, prior_requested, prior_terminal = inspect_control_request( + run_dir, + request_id=resolved_id, + fingerprint=fingerprint, + ) + if state == "indeterminate": + raise ControlJournalError( + _bound(f"control request {resolved_id!r} is indeterminate"), + code="indeterminate", + ) + if state == "replay": + assert prior_terminal is not None + return ControlResult( + request_id=resolved_id, + response=_response_from_terminal(prior_terminal), + replayed=True, + requested_event=prior_requested, + terminal_event=prior_terminal, + ) + + requested_event = _append_control_event( + journal_path, + run_dir=run_dir, + run_id=run_id, + event_type=CONTROL_REQUESTED, + payload=fingerprint, + idempotency_key=requested_idempotency_key(resolved_id), + ) + + transport_payload: dict[str, object] = {"op": op} + if worker is not None: + transport_payload["worker"] = worker + if text is not None: + transport_payload["text"] = text + + try: + response = send(transport_payload) + except Exception as exc: + detail = _bound(str(exc)) + failed_payload: dict[str, Any] = { + "op": op, + "request_id": resolved_id, + "code": "transport-error", + "detail": detail[:MAX_PAYLOAD_STR_LEN], + } + if worker is not None: + failed_payload["worker"] = worker + terminal = _append_control_event( + journal_path, + run_dir=run_dir, + run_id=run_id, + event_type=CONTROL_FAILED, + payload=failed_payload, + idempotency_key=failed_idempotency_key(resolved_id), + ) + return ControlResult( + request_id=resolved_id, + response={"ok": False, "error": detail, "code": "transport-error"}, + replayed=False, + requested_event=requested_event, + terminal_event=terminal, + ) + + if response.get("ok") is True: + observed_payload: dict[str, Any] = { + "op": op, + "request_id": resolved_id, + "detail": "ok", + } + if worker is not None: + observed_payload["worker"] = worker + elif isinstance(response.get("worker"), str): + observed_payload["worker"] = response["worker"] + resp_turn = response.get("turn_id") + if isinstance(resp_turn, str) and resp_turn: + observed_payload["turn_id"] = resp_turn + elif turn_id is not None: + observed_payload["turn_id"] = turn_id + interrupted = response.get("interrupted") + if isinstance(interrupted, int) and not isinstance(interrupted, bool): + observed_payload["detail"] = f"interrupted={interrupted}" + terminal = _append_control_event( + journal_path, + run_dir=run_dir, + run_id=run_id, + event_type=CONTROL_OBSERVED, + payload=observed_payload, + idempotency_key=observed_idempotency_key(resolved_id), + ) + return ControlResult( + request_id=resolved_id, + response=dict(response), + replayed=False, + requested_event=requested_event, + terminal_event=terminal, + ) + + failed_payload = { + "op": op, + "request_id": resolved_id, + "code": str(response.get("code") or "control-failed")[:MAX_PAYLOAD_STR_LEN], + "detail": str(response.get("error") or "control request failed")[:MAX_PAYLOAD_STR_LEN], + } + if worker is not None: + failed_payload["worker"] = worker + terminal = _append_control_event( + journal_path, + run_dir=run_dir, + run_id=run_id, + event_type=CONTROL_FAILED, + payload=failed_payload, + idempotency_key=failed_idempotency_key(resolved_id), + ) + return ControlResult( + request_id=resolved_id, + response=dict(response), + replayed=False, + requested_event=requested_event, + terminal_event=terminal, + ) + + +__all__ = [ + "CONTROL_FAILED", + "CONTROL_OBSERVED", + "CONTROL_REQUESTED", + "ControlJournalError", + "ControlResult", + "control_fingerprint_payload", + "execute_control_request", + "failed_idempotency_key", + "generate_request_id", + "inspect_control_request", + "journal_present", + "observed_idempotency_key", + "requested_idempotency_key", + "text_digest", +] diff --git a/src/brigade/run_event_cursor.py b/src/brigade/run_event_cursor.py new file mode 100644 index 00000000..c77375dd --- /dev/null +++ b/src/brigade/run_event_cursor.py @@ -0,0 +1,311 @@ +"""Opaque, dependency-free run-event cursor for Brigade issue #604. + +A cursor binds four coordinates from a verified journal event: + + - the cursor schema/version (``SUPPORTED_SCHEMA``), + - the owning ``run_id`` (non-empty string), + - a positive 1-based ``sequence`` index inside that run's event stream, and + - a lowercase 64-hex ``digest`` of the event (a SHA-256 style fingerprint). + +The cursor is **opaque**, not a MAC. It carries no secret and proves nothing +on its own. Integrity comes from the consumer matching the cursor's +coordinates against a verified #568 journal record: the consumer looks up the +journal entry for ``run_id`` at ``sequence`` and confirms that record's +``digest`` equals the cursor's ``digest``. ``validate`` is the helper that +performs that comparison against caller-supplied verified coordinates and +reports mismatches with stable, bounded error codes suitable for CLI +diagnostics. + +This module: + + - uses only the Python standard library, + - is deterministic (canonical compact JSON, sorted keys, urlsafe base64 + without padding, canonical pad bits), + - fails closed on every malformed input (bad base64, non-ASCII or + surrogate-containing cursor strings, bad UTF-8, non-object JSON, + duplicate JSON object keys, missing or extra fields, unsupported schema, + empty or invalid ``run_id``, bool / non-int / non-positive ``sequence``, + invalid ``digest``, non-canonical representation, and oversized input), + - validates every field on both :func:`encode` and :func:`decode` with the + same stable :class:`CursorError` codes, so a hand-constructed + :class:`DecodedCursor` cannot smuggle invalid values through ``encode``, + - never persists subscriber state and never carries event payloads, and + - is owned by the journal consumer; encoding and validation live here while + ``runs events`` and control idempotency live in the CLI / control layers. + +The cursor is intentionally not a MAC. Do not add HMAC material here. + +Canonical representation contract (enforced by :func:`decode`): + + - non-empty ASCII base64url alphabet ``[A-Za-z0-9_-]``, + - no ``=`` padding and no whitespace, + - canonical (zero) pad bits in the final base64 character, and + - canonical compact JSON (sorted keys, ``","``/``":"`` separators, no + whitespace, no duplicate object keys). + +After parsing and field validation, :func:`decode` re-encodes the decoded +cursor and requires byte-for-byte equality with the input, which rejects +any non-canonical representation that decodes to otherwise-valid data. + +Error messages are bounded and never echo untrusted schema values or +missing/extra field names; the stable ``code`` attribute is the public +diagnostic contract. +""" + +from __future__ import annotations + +import base64 +import binascii +import json +import re +from dataclasses import dataclass +from typing import Any + +SUPPORTED_SCHEMA = "brigade.run_event_cursor.v1" + +_MAX_CURSOR_BYTES = 8192 +_MAX_RUN_ID_LEN = 256 +_FIELDS = ("schema", "run_id", "sequence", "digest") +_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") +# Match the lifecycle journal run_id alphabet (run_events._RUN_ID_RE) so a +# cursor cannot bind a run_id the journal would refuse. +_RUN_ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,256}$") +# Canonical base64url alphabet: no padding, no whitespace, ASCII only. +_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+$") + + +class CursorError(RuntimeError): + """Cursor encode/decode/validation failure with a stable ``code``. + + ``code`` is a stable, bounded string suitable for CLI diagnostics and + test assertions. Error messages are bounded and never echo untrusted + cursor values or field names. The bounded set of codes is: + + - ``cursor_malformed`` (non-string / non-ASCII / surrogate / + empty / bad base64 / bad UTF-8 / bad JSON / duplicate JSON object + keys / non-canonical representation) + - ``cursor_non_object`` (valid JSON but not an object) + - ``cursor_oversized`` (raw input exceeds the size cap) + - ``cursor_schema_unsupported`` (missing or unknown schema field) + - ``cursor_field_missing`` (a required field is absent) + - ``cursor_field_extra`` (an unknown field is present) + - ``cursor_run_id_invalid`` (empty / non-string / oversized / control chars) + - ``cursor_sequence_invalid`` (bool / non-int / non-positive) + - ``cursor_digest_invalid`` (non-string / wrong length / non-lowercase-hex) + - ``cursor_run_id_mismatch`` (validate: run_id differs) + - ``cursor_sequence_mismatch`` (validate: sequence differs) + - ``cursor_digest_mismatch`` (validate: digest differs) + """ + + def __init__(self, message: str, *, code: str) -> None: + super().__init__(message) + self.code = code + + +@dataclass(frozen=True) +class DecodedCursor: + """A decoded, validated cursor. + + Field invariants are enforced by :func:`decode` and :func:`encode`; a + manually constructed instance is validated by :func:`encode` before it + is ever emitted. + """ + + schema: str + run_id: str + sequence: int + digest: str + + +def encode(cursor: DecodedCursor) -> str: + """Encode a cursor to an opaque, URL-safe, deterministic string. + + Validates every field with the same stable :class:`CursorError` codes as + :func:`decode`, so a hand-constructed :class:`DecodedCursor` with invalid + fields fails closed instead of producing an opaque string that decodes + to different values. The payload is canonical compact JSON (sorted keys, + no whitespace) wrapped in urlsafe base64 without padding. + """ + _validate_fields(cursor.schema, cursor.run_id, cursor.sequence, cursor.digest) + payload = { + "schema": cursor.schema, + "run_id": cursor.run_id, + "sequence": cursor.sequence, + "digest": cursor.digest, + } + raw = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + +def decode(cursor: str) -> DecodedCursor: + """Decode and fully validate an opaque cursor string. + + Accepts exactly one canonical representation: non-empty ASCII base64url + alphabet ``[A-Za-z0-9_-]`` with no ``=`` padding, no whitespace, and + canonical (zero) pad bits. Duplicate JSON object keys are rejected + (instead of ``json.loads`` last-key-wins). After parsing and field + validation, the decoded cursor is re-encoded and required to equal the + input byte-for-byte, which also rejects non-canonical JSON key order or + whitespace. Fails closed with :class:`CursorError` on any malformed or + invalid input. + """ + if not isinstance(cursor, str): + raise CursorError("cursor must be a string", code="cursor_malformed") + # Reject non-ASCII and surrogate-containing cursor strings up front. + if not cursor.isascii(): + raise CursorError("cursor must be ASCII", code="cursor_malformed") + if not cursor: + raise CursorError("cursor must not be empty", code="cursor_malformed") + if len(cursor.encode("utf-8")) > _MAX_CURSOR_BYTES: + raise CursorError("cursor input exceeds size limit", code="cursor_oversized") + # Canonical alphabet, no padding, no whitespace. fullmatch (not match) + # so a trailing newline cannot satisfy the ``$`` anchor. + if not _B64URL_RE.fullmatch(cursor): + raise CursorError("cursor base64 is malformed", code="cursor_malformed") + try: + raw = base64.urlsafe_b64decode(_add_b64_padding(cursor)) + except (ValueError, binascii.Error): + raise CursorError("cursor base64 is malformed", code="cursor_malformed") from None + try: + text = raw.decode("utf-8") + except UnicodeDecodeError: + raise CursorError("cursor payload is not valid UTF-8", code="cursor_malformed") from None + try: + payload = json.loads(text, object_pairs_hook=_reject_duplicate_keys) + except (json.JSONDecodeError, _DuplicateKeyError): + raise CursorError("cursor payload is not valid JSON", code="cursor_malformed") from None + if not isinstance(payload, dict): + raise CursorError("cursor payload is not a JSON object", code="cursor_non_object") + + schema, run_id, sequence, digest = _extract_fields(payload) + _validate_fields(schema, run_id, sequence, digest) + + decoded = DecodedCursor(schema=schema, run_id=run_id, sequence=sequence, digest=digest) + # Canonical round-trip: re-encode and require byte-for-byte equality. + # Rejects non-canonical pad bits, non-canonical JSON key order/whitespace, + # and any other non-canonical representation that decodes to the same data. + if encode(decoded) != cursor: + raise CursorError("cursor is not in canonical form", code="cursor_malformed") + return decoded + + +def validate( + cursor: DecodedCursor, + *, + expected_run_id: str, + event_sequence: int, + event_digest: str, +) -> None: + """Compare a decoded cursor against a verified journal event's coordinates. + + ``expected_run_id`` / ``event_sequence`` / ``event_digest`` must come from a + journal record the caller has already verified. On success returns + ``None``; on any mismatch raises :class:`CursorError` with a stable + ``code`` (``cursor_run_id_mismatch`` / ``cursor_sequence_mismatch`` / + ``cursor_digest_mismatch``). The caller-supplied coordinates are also + sanity-checked and rejected with the corresponding ``*_invalid`` code so + callers cannot silently compare against garbage. + + :class:`DecodedCursor` is public and manually constructible, so the + cursor's own fields are validated with the shared field validator before + any comparison; an invalid cursor schema/run_id/sequence/digest fails + with the corresponding ``*_invalid`` / ``cursor_schema_unsupported`` + code rather than being silently compared. + """ + # Validate the cursor itself first: a hand-constructed DecodedCursor may + # carry invalid fields, and we must not compare garbage against garbage. + _validate_fields(cursor.schema, cursor.run_id, cursor.sequence, cursor.digest) + + if not _is_valid_run_id(expected_run_id): + raise CursorError("invalid expected run_id", code="cursor_run_id_invalid") + if not _is_valid_sequence(event_sequence): + raise CursorError("invalid event sequence", code="cursor_sequence_invalid") + if not _is_valid_digest(event_digest): + raise CursorError("invalid event digest", code="cursor_digest_invalid") + + if cursor.run_id != expected_run_id: + raise CursorError( + "cursor run_id does not match the verified journal record", + code="cursor_run_id_mismatch", + ) + if cursor.sequence != event_sequence: + raise CursorError( + "cursor sequence does not match the verified journal record", + code="cursor_sequence_mismatch", + ) + if cursor.digest != event_digest: + raise CursorError( + "cursor digest does not match the verified journal record", + code="cursor_digest_mismatch", + ) + + +class _DuplicateKeyError(ValueError): + """Raised by the JSON object_pairs_hook on duplicate object keys.""" + + +def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + seen: dict[str, Any] = {} + for key, value in pairs: + if key in seen: + raise _DuplicateKeyError("duplicate key") + seen[key] = value + return seen + + +def _extract_fields(payload: dict[str, Any]) -> tuple[Any, Any, Any, Any]: + keys = set(payload.keys()) + expected = set(_FIELDS) + missing = expected - keys + if missing: + raise CursorError("cursor missing required field(s)", code="cursor_field_missing") + extra = keys - expected + if extra: + raise CursorError("cursor has extra field(s)", code="cursor_field_extra") + return payload["schema"], payload["run_id"], payload["sequence"], payload["digest"] + + +def _validate_fields(schema: Any, run_id: Any, sequence: Any, digest: Any) -> None: + if not isinstance(schema, str) or schema != SUPPORTED_SCHEMA: + raise CursorError("unsupported cursor schema", code="cursor_schema_unsupported") + if not _is_valid_run_id(run_id): + raise CursorError("invalid run_id", code="cursor_run_id_invalid") + if not _is_valid_sequence(sequence): + raise CursorError("invalid sequence", code="cursor_sequence_invalid") + if not _is_valid_digest(digest): + raise CursorError("invalid digest", code="cursor_digest_invalid") + + +def _add_b64_padding(value: str) -> str: + # urlsafe_b64decode tolerates missing padding in modern Python, but + # restoring it explicitly keeps the failure mode narrow and predictable. + pad = (-len(value)) % 4 + return value + ("=" * pad) + + +def _is_valid_run_id(value: Any) -> bool: + # fullmatch (not match) so the trailing ``$`` anchor cannot be satisfied + # before a final newline - ``"run-id\n"`` must NOT validate. + return isinstance(value, str) and bool(_RUN_ID_RE.fullmatch(value)) and len(value) <= _MAX_RUN_ID_LEN + + +def _is_valid_sequence(value: Any) -> bool: + # bool is a subclass of int; reject it explicitly so True/False are not + # silently coerced to 1/0. Positive non-bool integer is the shared + # requirement; #568 owns any sequence upper bound. + return isinstance(value, int) and not isinstance(value, bool) and value >= 1 + + +def _is_valid_digest(value: Any) -> bool: + # fullmatch (not match) so a trailing newline cannot slip past ``$``. + return isinstance(value, str) and bool(_DIGEST_RE.fullmatch(value)) + + +__all__ = [ + "SUPPORTED_SCHEMA", + "CursorError", + "DecodedCursor", + "decode", + "encode", + "validate", +] diff --git a/src/brigade/run_events.py b/src/brigade/run_events.py index c5a6d629..f2f32313 100644 --- a/src/brigade/run_events.py +++ b/src/brigade/run_events.py @@ -116,6 +116,12 @@ "record_sha256", } ), + # Live-control request/observation pairs (issue #604). Reference-only: + # digests, op codes, worker names, turn ids, and request ids — never + # steering text, model output, or provider bodies. + "control.requested": frozenset({"op", "worker", "text_digest", "turn_id", "request_id"}), + "control.observed": frozenset({"op", "worker", "turn_id", "request_id", "detail"}), + "control.failed": frozenset({"op", "worker", "turn_id", "request_id", "code", "detail"}), } APPROVAL_DECISION_STATES = frozenset({"pending", "approved", "rejected", "held", "consumed"}) APPROVAL_DECISION_EVENT_STATES = { diff --git a/src/brigade/run_projector.py b/src/brigade/run_projector.py index 3a3f9b93..f450280e 100644 --- a/src/brigade/run_projector.py +++ b/src/brigade/run_projector.py @@ -145,6 +145,10 @@ def _has_dispatch_identity(payload: Any) -> bool: "approval.held", "approval.consumed", "run.redaction.recorded", + # Live-control pairs advance the journal cursor only (issue #604). + "control.requested", + "control.observed", + "control.failed", } ) diff --git a/src/brigade/runs_cmd.py b/src/brigade/runs_cmd.py index 9bf55dad..0164a979 100644 --- a/src/brigade/runs_cmd.py +++ b/src/brigade/runs_cmd.py @@ -1826,6 +1826,132 @@ def redact( return 0 +_TERMINAL_LIFECYCLE_EVENT_TYPES = frozenset( + { + "run.completed", + "run.failed", + "run.interrupted", + } +) + + +def _lifecycle_event_record(event: Any) -> dict[str, object]: + from . import run_event_cursor + + payload = event.to_dict() + payload["cursor"] = run_event_cursor.encode( + run_event_cursor.DecodedCursor( + schema=run_event_cursor.SUPPORTED_SCHEMA, + run_id=event.run_id, + sequence=event.sequence, + digest=event.event_digest, + ) + ) + return payload + + +def _events_return_code(event: Any) -> int: + if event.event_type == "run.completed": + status = event.payload.get("status") if isinstance(event.payload, Mapping) else None + return 0 if status in _SUCCESS_STATUSES else 1 + return 1 + + +def events( + run: str | Path, + *, + cwd: Path, + runs_dir: Path | None = None, + after: str | None = None, + follow: bool = False, + interval: float = 1.0, +) -> int: + """Emit verified lifecycle journal records as newline-delimited JSON (issue #604).""" + from . import run_event_cursor, run_journal, run_lifecycle + + if interval < 0: + print("error: --interval must be non-negative", file=sys.stderr) + return 2 + run_dir, error = _resolve_run_dir(run, cwd=cwd, runs_dir=runs_dir) + if error is not None: + print(error, file=sys.stderr) + return 2 + assert run_dir is not None + + journal_path = run_lifecycle._journal_path(run_dir) + if not journal_path.is_file(): + print("error: legacy-no-journal", file=sys.stderr) + return 2 + + expected_run_id = run_lifecycle._run_id_from_dir(run_dir) + after_sequence = 0 + if after is not None: + try: + cursor = run_event_cursor.decode(after) + except run_event_cursor.CursorError as exc: + print(f"error: {exc.code}: {exc}", file=sys.stderr) + return 2 + if cursor.run_id != expected_run_id: + print("error: cursor_run_id_mismatch: cursor run_id does not match the run", file=sys.stderr) + return 2 + try: + report = run_journal.read_journal_bounded(journal_path) + except run_journal.RunJournalError as exc: + print(f"error: {exc.diagnostic}", file=sys.stderr) + return 2 + if report.partial_tail is not None: + print("error: lifecycle journal ends in a partial line", file=sys.stderr) + return 2 + if report.chain_errors: + print(f"error: {report.chain_errors[0]}", file=sys.stderr) + return 2 + match = next((event for event in report.events if event.sequence == cursor.sequence), None) + if match is None: + print("error: cursor_sequence_mismatch: cursor sequence is not in the journal", file=sys.stderr) + return 2 + try: + run_event_cursor.validate( + cursor, + expected_run_id=expected_run_id, + event_sequence=match.sequence, + event_digest=match.event_digest, + ) + except run_event_cursor.CursorError as exc: + print(f"error: {exc.code}: {exc}", file=sys.stderr) + return 2 + after_sequence = cursor.sequence + + last_emitted = after_sequence + while True: + try: + report = run_journal.read_journal_bounded(journal_path) + except run_journal.RunJournalError as exc: + print(f"error: {exc.diagnostic}", file=sys.stderr) + return 2 + if report.partial_tail is not None: + print("error: lifecycle journal ends in a partial line", file=sys.stderr) + return 2 + if report.chain_errors: + print(f"error: {report.chain_errors[0]}", file=sys.stderr) + return 2 + + terminal_event = None + for event in report.events: + if event.sequence <= last_emitted: + continue + _emit_json(_lifecycle_event_record(event)) + last_emitted = event.sequence + if event.event_type in _TERMINAL_LIFECYCLE_EVENT_TYPES: + terminal_event = event + break + + if terminal_event is not None: + return _events_return_code(terminal_event) + if not follow: + return 0 + time.sleep(interval) + + def watch( run: str | Path, *, diff --git a/tests/test_run_event_cursor.py b/tests/test_run_event_cursor.py new file mode 100644 index 00000000..44acf0d7 --- /dev/null +++ b/tests/test_run_event_cursor.py @@ -0,0 +1,825 @@ +"""Failing-first tests for the dependency-free run-event cursor (issue #604). + +These tests pin the hardened contract from the independent Ollama review: +canonical-only decode, duplicate-key rejection, bounded/redacted error +messages, encode-side field validation, and non-ASCII / surrogate rejection. +They use only the standard library. +""" + +from __future__ import annotations + +import base64 +import json + +import pytest + +from brigade import run_event_cursor +from brigade.run_event_cursor import ( + CursorError, + DecodedCursor, + decode, + encode, + validate, +) + +_SCHEMA = "brigade.run_event_cursor.v1" +_GOOD_RUN_ID = "run-20260728-aa74565d" +_GOOD_SEQUENCE = 7 +_GOOD_DIGEST = "a" * 64 + + +def _good_cursor() -> DecodedCursor: + return DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + + +def _encode_payload(payload: dict) -> str: + """Canonical encoding helper: sorted keys, compact separators, no padding.""" + raw = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + +def _encode_raw_bytes(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + +# --- encode / decode round trip ------------------------------------------------- + + +def test_encode_is_deterministic_and_url_safe(): + c = _good_cursor() + a = encode(c) + b = encode(c) + assert a == b + assert all(ch not in a for ch in "+/=\n\r\t ") + assert a.isascii() + + +def test_encode_decode_round_trip_preserves_fields(): + c = _good_cursor() + decoded = decode(encode(c)) + assert decoded == c + + +def test_encoded_payload_is_canonical_compact_sorted_json(): + c = _good_cursor() + raw = base64.urlsafe_b64decode(encode(c).encode("ascii") + b"==") + payload = json.loads(raw.decode("utf-8")) + assert payload == { + "schema": _SCHEMA, + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + } + assert raw.decode("utf-8") == json.dumps( + { + "schema": _SCHEMA, + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + }, + sort_keys=True, + separators=(",", ":"), + ) + + +# --- decode: input shape rejections -------------------------------------------- + + +def test_decode_rejects_non_string_input(): + for bad in (123, None, b"not-a-str"): + with pytest.raises(CursorError) as exc: + decode(bad) # type: ignore[arg-type] + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_ascii_cursor_string(): + with pytest.raises(CursorError) as exc: + decode("caf\xc3\xa9") # type: ignore[arg-type] + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_surrogate_containing_cursor_string(): + with pytest.raises(CursorError) as exc: + decode("ab\ud800cd") + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_empty_cursor_string(): + with pytest.raises(CursorError) as exc: + decode("") + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_malformed_base64(): + with pytest.raises(CursorError) as exc: + decode("!!!not-base64!!!") + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_padded_base64(): + c = _good_cursor() + padded = encode(c) + "=" + with pytest.raises(CursorError) as exc: + decode(padded) + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_whitespace_in_cursor(): + c = _good_cursor() + with pytest.raises(CursorError) as exc: + decode(" " + encode(c) + " ") + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_newline_in_cursor(): + c = _good_cursor() + with pytest.raises(CursorError) as exc: + decode(encode(c) + "\n") + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_url_safe_alphabet(): + # '+' and '/' are standard base64, not base64url. Substituting one into a + # otherwise-valid cursor must be rejected at the alphabet check. + c = _good_cursor() + bad = "+" + encode(c)[1:] + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + bad2 = "/" + encode(c)[1:] + with pytest.raises(CursorError) as exc2: + decode(bad2) + assert exc2.value.code == "cursor_malformed" + + +def test_decode_rejects_oversized_input(): + c = _good_cursor() + bloated = encode(c) + ("A" * 8192) + with pytest.raises(CursorError) as exc: + decode(bloated) + assert exc.value.code == "cursor_oversized" + + +def test_decode_rejects_non_utf8_payload(): + bad = _encode_raw_bytes(b"\xff\xfe\x00\x01") + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_object_json(): + arr = _encode_raw_bytes(b"[1,2,3]") + with pytest.raises(CursorError) as exc: + decode(arr) + assert exc.value.code == "cursor_non_object" + scalar = _encode_raw_bytes(b'"hello"') + with pytest.raises(CursorError) as exc2: + decode(scalar) + assert exc2.value.code == "cursor_non_object" + + +# --- decode: canonical representation rejections -------------------------------- + + +def test_decode_rejects_duplicate_json_keys(): + raw = ( + '{"schema": "' + + _SCHEMA + + '", "run_id": "' + + _GOOD_RUN_ID + + '", "sequence": ' + + str(_GOOD_SEQUENCE) + + ', "digest": "' + + _GOOD_DIGEST + + '", "digest": "' + + ("b" * 64) + + '"}' + ).encode("utf-8") + bad = _encode_raw_bytes(raw) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_canonical_json_key_order(): + # Same fields, valid values, but keys emitted in non-sorted order. + raw = ( + '{"run_id": "' + + _GOOD_RUN_ID + + '", "schema": "' + + _SCHEMA + + '", "digest": "' + + _GOOD_DIGEST + + '", "sequence": ' + + str(_GOOD_SEQUENCE) + + "}" + ).encode("utf-8") + bad = _encode_raw_bytes(raw) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_canonical_json_whitespace(): + raw = ( + '{ "schema": "' + + _SCHEMA + + '", "run_id": "' + + _GOOD_RUN_ID + + '", "sequence": ' + + str(_GOOD_SEQUENCE) + + ', "digest": "' + + _GOOD_DIGEST + + '" }' + ).encode("utf-8") + bad = _encode_raw_bytes(raw) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + + +def test_decode_rejects_non_canonical_pad_bits(): + # Build a valid cursor whose JSON payload length is NOT a multiple of 3, + # so the final base64 char carries non-zero-width pad bits we can flip + # without changing the decoded bytes. + cursor = DecodedCursor( + schema=_SCHEMA, + run_id="run-20260728-aa74565d-x", # length tuned so payload % 3 != 0 + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + encoded = encode(cursor) + raw = base64.urlsafe_b64decode(_pad(encoded)) + rem = len(raw) % 3 + assert rem != 0, "test payload must have pad bits to corrupt" + alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" + # Pad bits are the LOW bits of the final char. Canonical form has them + # zero; flipping the lowest pad bit yields a non-canonical string that + # decodes to the same bytes. + increment = 1 + last_val = alphabet.index(encoded[-1]) + new_val = last_val + increment + assert new_val < 64 + bad = encoded[:-1] + alphabet[new_val] + # Sanity: decodes to the same raw bytes (pad bits ignored by decoder). + assert base64.urlsafe_b64decode(_pad(bad)) == raw + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_malformed" + + +def _pad(value: str) -> str: + pad = (-len(value)) % 4 + return value + ("=" * pad) + + +# --- decode: field validation rejections --------------------------------------- + + +def _bad_cursor(payload: dict) -> str: + return _encode_payload(payload) + + +def test_decode_rejects_unsupported_schema(): + bad = _bad_cursor( + { + "schema": "brigade.run_event_cursor.v999", + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + } + ) + with pytest.raises(CursorError, match="schema") as exc: + decode(bad) + assert exc.value.code == "cursor_schema_unsupported" + + +def test_decode_rejects_missing_schema_field(): + bad = _bad_cursor({"run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="missing") as exc: + decode(bad) + assert exc.value.code == "cursor_field_missing" + + +def test_decode_rejects_missing_run_id_field(): + bad = _bad_cursor({"schema": _SCHEMA, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="missing") as exc: + decode(bad) + assert exc.value.code == "cursor_field_missing" + + +def test_decode_rejects_missing_sequence_field(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="missing") as exc: + decode(bad) + assert exc.value.code == "cursor_field_missing" + + +def test_decode_rejects_missing_digest_field(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE}) + with pytest.raises(CursorError, match="missing") as exc: + decode(bad) + assert exc.value.code == "cursor_field_missing" + + +def test_decode_rejects_extra_field(): + bad = _bad_cursor( + { + "schema": _SCHEMA, + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + "payload": "sneaky", + } + ) + with pytest.raises(CursorError, match="extra") as exc: + decode(bad) + assert exc.value.code == "cursor_field_extra" + + +def test_decode_rejects_empty_run_id(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": "", "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="run_id") as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_decode_rejects_non_string_run_id(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": 42, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="run_id") as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_decode_rejects_run_id_with_control_characters(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": "bad\nid", "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="run_id") as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +@pytest.mark.parametrize("run_id", ["has space", "has/slash", "has@at", "emoji-🙂"]) +def test_decode_rejects_run_id_outside_journal_alphabet(run_id): + """Cursor run_id must match the lifecycle journal alphabet (issue #604).""" + bad = _bad_cursor({"schema": _SCHEMA, "run_id": run_id, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="run_id") as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_decode_rejects_oversized_run_id(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": "x" * 1024, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="run_id") as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_decode_rejects_bool_sequence(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": True, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="sequence") as exc: + decode(bad) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_decode_rejects_non_int_sequence(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": "7", "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="sequence") as exc: + decode(bad) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_decode_rejects_zero_sequence(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": 0, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="sequence") as exc: + decode(bad) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_decode_rejects_negative_sequence(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": -1, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError, match="sequence") as exc: + decode(bad) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_decode_rejects_non_string_digest(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": 64}) + with pytest.raises(CursorError, match="digest") as exc: + decode(bad) + assert exc.value.code == "cursor_digest_invalid" + + +def test_decode_rejects_uppercase_digest(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": "A" * 64}) + with pytest.raises(CursorError, match="digest") as exc: + decode(bad) + assert exc.value.code == "cursor_digest_invalid" + + +def test_decode_rejects_wrong_length_digest(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": "a" * 63}) + with pytest.raises(CursorError, match="digest") as exc: + decode(bad) + assert exc.value.code == "cursor_digest_invalid" + + +def test_decode_rejects_non_hex_digest(): + bad = _bad_cursor({"schema": _SCHEMA, "run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": "g" * 64}) + with pytest.raises(CursorError, match="digest") as exc: + decode(bad) + assert exc.value.code == "cursor_digest_invalid" + + +def test_decode_error_carries_stable_code(): + with pytest.raises(CursorError) as exc: + decode("!!!not-base64!!!") + assert exc.value.code == "cursor_malformed" + + +# --- encode: manual DecodedCursor field validation ----------------------------- + + +def test_encode_rejects_unsupported_schema(): + c = DecodedCursor( + schema="brigade.run_event_cursor.v999", + run_id=_GOOD_RUN_ID, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_schema_unsupported" + + +def test_encode_rejects_empty_run_id(): + c = DecodedCursor(schema=_SCHEMA, run_id="", sequence=_GOOD_SEQUENCE, digest=_GOOD_DIGEST) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_encode_rejects_non_string_run_id(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=42, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, # type: ignore[arg-type] + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_encode_rejects_bool_sequence(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence=True, + digest=_GOOD_DIGEST, # type: ignore[arg-type] + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_encode_rejects_zero_sequence(): + c = DecodedCursor(schema=_SCHEMA, run_id=_GOOD_RUN_ID, sequence=0, digest=_GOOD_DIGEST) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_encode_rejects_non_int_sequence(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence="7", + digest=_GOOD_DIGEST, # type: ignore[arg-type] + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_sequence_invalid" + + +def test_encode_rejects_uppercase_digest(): + c = DecodedCursor(schema=_SCHEMA, run_id=_GOOD_RUN_ID, sequence=_GOOD_SEQUENCE, digest="A" * 64) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_digest_invalid" + + +def test_encode_rejects_wrong_length_digest(): + c = DecodedCursor(schema=_SCHEMA, run_id=_GOOD_RUN_ID, sequence=_GOOD_SEQUENCE, digest="a" * 63) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_digest_invalid" + + +def test_encode_rejects_non_hex_digest(): + c = DecodedCursor(schema=_SCHEMA, run_id=_GOOD_RUN_ID, sequence=_GOOD_SEQUENCE, digest="g" * 64) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_digest_invalid" + + +def test_encode_then_decode_round_trips_for_valid_manual_cursor(): + c = _good_cursor() + assert decode(encode(c)) == c + + +# --- bounded / redacted error messages ----------------------------------------- + + +def test_error_message_does_not_echo_untrusted_schema_value(): + bad = _bad_cursor( + { + "schema": "UNTRUSTED-SECRET-SCHEMA-VALUE", + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + } + ) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_schema_unsupported" + assert "UNTRUSTED-SECRET-SCHEMA-VALUE" not in str(exc.value) + + +def test_error_message_does_not_echo_missing_field_names(): + bad = _bad_cursor({"run_id": _GOOD_RUN_ID, "sequence": _GOOD_SEQUENCE, "digest": _GOOD_DIGEST}) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_field_missing" + msg = str(exc.value) + assert "schema" not in msg + assert "run_id" not in msg + assert "sequence" not in msg + assert "digest" not in msg + + +def test_error_message_does_not_echo_extra_field_names(): + bad = _bad_cursor( + { + "schema": _SCHEMA, + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + "secret_extra": "leak", + } + ) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_field_extra" + msg = str(exc.value) + assert "secret_extra" not in msg + assert "leak" not in msg + + +def test_error_messages_are_bounded(): + # Every CursorError message is a short, fixed string with no interpolated + # untrusted content. Assert a hard upper bound to catch future regressions. + cases = [ + ("cursor_malformed", "!!!not-base64!!!"), + ("cursor_malformed", "caf\xc3\xa9"), + ("cursor_malformed", "ab\ud800cd"), + ("cursor_malformed", ""), + ("cursor_oversized", encode(_good_cursor()) + ("A" * 8192)), + ("cursor_non_object", _encode_raw_bytes(b"[1,2,3]")), + ("cursor_schema_unsupported", _bad_cursor({**_good_payload(), "schema": "x"})), + ("cursor_field_missing", _bad_cursor({"run_id": _GOOD_RUN_ID})), + ("cursor_field_extra", _bad_cursor({**_good_payload(), "extra": 1})), + ("cursor_run_id_invalid", _bad_cursor({**_good_payload(), "run_id": ""})), + ("cursor_sequence_invalid", _bad_cursor({**_good_payload(), "sequence": True})), + ("cursor_digest_invalid", _bad_cursor({**_good_payload(), "digest": "A" * 64})), + ] + for expected_code, bad_input in cases: + with pytest.raises(CursorError) as exc: + decode(bad_input) + assert exc.value.code == expected_code + assert len(str(exc.value)) <= 80, (expected_code, str(exc.value)) + + +def _good_payload() -> dict: + return { + "schema": _SCHEMA, + "run_id": _GOOD_RUN_ID, + "sequence": _GOOD_SEQUENCE, + "digest": _GOOD_DIGEST, + } + + +# --- validate helper ----------------------------------------------------------- + + +def test_validate_accepts_matching_coordinates(): + c = _good_cursor() + validate( + c, + expected_run_id=_GOOD_RUN_ID, + event_sequence=_GOOD_SEQUENCE, + event_digest=_GOOD_DIGEST, + ) + + +def test_validate_rejects_run_id_mismatch_with_stable_code(): + c = _good_cursor() + with pytest.raises(CursorError, match="run_id") as exc: + validate(c, expected_run_id="other-run", event_sequence=_GOOD_SEQUENCE, event_digest=_GOOD_DIGEST) + assert exc.value.code == "cursor_run_id_mismatch" + + +def test_validate_rejects_sequence_mismatch_with_stable_code(): + c = _good_cursor() + with pytest.raises(CursorError, match="sequence") as exc: + validate(c, expected_run_id=_GOOD_RUN_ID, event_sequence=_GOOD_SEQUENCE + 1, event_digest=_GOOD_DIGEST) + assert exc.value.code == "cursor_sequence_mismatch" + + +def test_validate_rejects_digest_mismatch_with_stable_code(): + c = _good_cursor() + with pytest.raises(CursorError, match="digest") as exc: + validate(c, expected_run_id=_GOOD_RUN_ID, event_sequence=_GOOD_SEQUENCE, event_digest="b" * 64) + assert exc.value.code == "cursor_digest_mismatch" + + +def test_validate_rejects_empty_expected_run_id(): + c = _good_cursor() + with pytest.raises(CursorError, match="run_id"): + validate(c, expected_run_id="", event_sequence=_GOOD_SEQUENCE, event_digest=_GOOD_DIGEST) + + +def test_validate_rejects_non_positive_event_sequence(): + c = _good_cursor() + with pytest.raises(CursorError, match="sequence"): + validate(c, expected_run_id=_GOOD_RUN_ID, event_sequence=0, event_digest=_GOOD_DIGEST) + + +def test_validate_rejects_invalid_event_digest(): + c = _good_cursor() + with pytest.raises(CursorError, match="digest"): + validate(c, expected_run_id=_GOOD_RUN_ID, event_sequence=_GOOD_SEQUENCE, event_digest="bad") + + +# --- regression: trailing-newline rejection through fullmatch ------------------- +# +# ``$`` in a Python regex can match before a final newline, so a run_id or +# digest ending in ``\n`` would slip past ``_RUN_ID_RE`` / ``_DIGEST_RE`` when +# used with ``.match()``. These tests pin the stricter ``fullmatch`` behavior +# across encode(), decode(), and direct validate(). + + +def test_encode_rejects_run_id_with_trailing_newline(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID + "\n", + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_decode_rejects_run_id_with_trailing_newline(): + # Canonical JSON with a run_id value that ends in a newline. The shared + # field validator (called from decode) must reject it before any + # canonical round-trip comparison runs. + raw = ( + '{"digest": "' + + _GOOD_DIGEST + + '", "run_id": "' + + _GOOD_RUN_ID + + '\\n", "schema": "' + + _SCHEMA + + '", "sequence": ' + + str(_GOOD_SEQUENCE) + + "}" + ).encode("utf-8") + bad = _encode_raw_bytes(raw) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_validate_rejects_cursor_run_id_with_trailing_newline(): + # DecodedCursor is public and manually constructible; validate() must + # validate the cursor's own fields, not just the caller-supplied coords. + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID + "\n", + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + with pytest.raises(CursorError) as exc: + validate( + c, + expected_run_id=_GOOD_RUN_ID, + event_sequence=_GOOD_SEQUENCE, + event_digest=_GOOD_DIGEST, + ) + assert exc.value.code == "cursor_run_id_invalid" + + +def test_encode_rejects_digest_with_trailing_newline(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST + "\n", + ) + with pytest.raises(CursorError) as exc: + encode(c) + assert exc.value.code == "cursor_digest_invalid" + + +def test_decode_rejects_digest_with_trailing_newline(): + raw = ( + '{"digest": "' + + _GOOD_DIGEST + + '\\n", "run_id": "' + + _GOOD_RUN_ID + + '", "schema": "' + + _SCHEMA + + '", "sequence": ' + + str(_GOOD_SEQUENCE) + + "}" + ).encode("utf-8") + bad = _encode_raw_bytes(raw) + with pytest.raises(CursorError) as exc: + decode(bad) + assert exc.value.code == "cursor_digest_invalid" + + +def test_validate_rejects_cursor_digest_with_trailing_newline(): + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST + "\n", + ) + with pytest.raises(CursorError) as exc: + validate( + c, + expected_run_id=_GOOD_RUN_ID, + event_sequence=_GOOD_SEQUENCE, + event_digest=_GOOD_DIGEST, + ) + assert exc.value.code == "cursor_digest_invalid" + + +def test_validate_rejects_manually_constructed_unsupported_schema(): + # A hand-built DecodedCursor with an unsupported schema must be rejected + # by validate() with cursor_schema_unsupported, not silently compared. + c = DecodedCursor( + schema="brigade.run_event_cursor.v999", + run_id=_GOOD_RUN_ID, + sequence=_GOOD_SEQUENCE, + digest=_GOOD_DIGEST, + ) + with pytest.raises(CursorError) as exc: + validate( + c, + expected_run_id=_GOOD_RUN_ID, + event_sequence=_GOOD_SEQUENCE, + event_digest=_GOOD_DIGEST, + ) + assert exc.value.code == "cursor_schema_unsupported" + + +def test_validate_rejects_manually_constructed_invalid_sequence(): + # Symmetric coverage: a hand-built cursor with a bool sequence must be + # rejected by validate() via the shared field validator. + c = DecodedCursor( + schema=_SCHEMA, + run_id=_GOOD_RUN_ID, + sequence=True, # type: ignore[arg-type] + digest=_GOOD_DIGEST, + ) + with pytest.raises(CursorError) as exc: + validate( + c, + expected_run_id=_GOOD_RUN_ID, + event_sequence=_GOOD_SEQUENCE, + event_digest=_GOOD_DIGEST, + ) + assert exc.value.code == "cursor_sequence_invalid" + + +# --- module surface ------------------------------------------------------------ + + +def test_module_exposes_supported_schema_constant(): + assert run_event_cursor.SUPPORTED_SCHEMA == _SCHEMA + + +def test_module_docstring_documents_mac_boundary(): + assert "not a MAC" in run_event_cursor.__doc__ or "not a mac" in (run_event_cursor.__doc__ or "").lower() + + +def test_module_docstring_documents_canonical_representation(): + assert "canonical" in (run_event_cursor.__doc__ or "").lower() diff --git a/tests/test_run_events.py b/tests/test_run_events.py index b4ea3f82..78f03652 100644 --- a/tests/test_run_events.py +++ b/tests/test_run_events.py @@ -341,3 +341,59 @@ def test_validate_event_limits_unknown_key_diagnostics(): errors = run_events.validate_event(env) assert errors assert len(errors[0]) <= 240 + + +@pytest.mark.parametrize( + ("event_type", "allowed"), + [ + ( + "control.requested", + frozenset({"op", "worker", "text_digest", "turn_id", "request_id"}), + ), + ( + "control.observed", + frozenset({"op", "worker", "turn_id", "request_id", "detail"}), + ), + ( + "control.failed", + frozenset({"op", "worker", "turn_id", "request_id", "code", "detail"}), + ), + ], +) +def test_control_event_types_registered_with_closed_payload_keys(event_type, allowed): + assert event_type in run_events.EVENT_TYPES + assert run_events.EVENT_TYPES[event_type] == allowed + + +def test_control_requested_rejects_unknown_payload_key_with_bounded_diagnostic(): + with pytest.raises(run_events.CanonicalizationError) as exc: + run_events.build_event( + run_id=RUN_ID, + sequence=1, + event_type="control.requested", + payload={ + "op": "steer", + "worker": "coder", + "text_digest": "a" * 64, + "request_id": "req-1", + "steering_text": "do not store me", + }, + idempotency_key="control.requested:req-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + assert len(str(exc.value)) <= 240 + + +def test_control_requested_rejects_forbidden_private_payload_keys(): + with pytest.raises(run_events.CanonicalizationError) as exc: + run_events.build_event( + run_id=RUN_ID, + sequence=1, + event_type="control.requested", + payload={"op": "steer", "request_id": "req-1", "prompt": "secret"}, + idempotency_key="control.requested:req-1", + recorded_at=RECORDED_AT, + previous_digest=None, + ) + assert len(str(exc.value)) <= 240 diff --git a/tests/test_run_events_cursor_control.py b/tests/test_run_events_cursor_control.py new file mode 100644 index 00000000..ef93ffa3 --- /dev/null +++ b/tests/test_run_events_cursor_control.py @@ -0,0 +1,573 @@ +"""Acceptance tests for durable cursors and idempotent live control (issue #604).""" + +from __future__ import annotations + +import json +import threading +import time +from pathlib import Path + +import pytest + +from brigade import cli, run_control, run_control_journal, run_event_cursor, run_journal +from brigade.run_control_journal import ControlJournalError + +RUN_ID = "20260731-604000-cursors01" +RECORDED_AT = "2026-07-31T12:00:00.000000Z" + + +def _run_dir(tmp_path: Path, run_id: str = RUN_ID) -> 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, + *, + run_id: str, + 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 _seed_lifecycle(run_dir: Path, *, terminal: bool = False) -> list[run_journal.RunEvent]: + journal = _journal_path(run_dir) + run_id = run_dir.name + events = [] + events.append( + _append( + journal, + run_id=run_id, + event_type="run.created", + payload={"status": "started"}, + idempotency_key="create-1", + expected_previous_sequence=0, + recorded_at=RECORDED_AT, + ) + ) + events.append( + _append( + journal, + run_id=run_id, + event_type="run.dispatching.started", + payload={"detail": "dispatching"}, + idempotency_key="dispatching-1", + expected_previous_sequence=1, + recorded_at="2026-07-31T12:00:01.000000Z", + ) + ) + if terminal: + events.append( + _append( + journal, + run_id=run_id, + event_type="run.completed", + payload={"status": "ok", "detail": "done"}, + idempotency_key="complete-1", + expected_previous_sequence=2, + recorded_at="2026-07-31T12:00:02.000000Z", + ) + ) + return events + + +def _cursor_for(event: run_journal.RunEvent) -> str: + return run_event_cursor.encode( + run_event_cursor.DecodedCursor( + schema=run_event_cursor.SUPPORTED_SCHEMA, + run_id=event.run_id, + sequence=event.sequence, + digest=event.event_digest, + ) + ) + + +def _parse_ndjson(stdout: str) -> list[dict]: + rows = [] + for line in stdout.splitlines(): + if not line.strip(): + continue + rows.append(json.loads(line)) + return rows + + +def test_events_follow_restart_with_cursor_emits_each_later_event_once(tmp_path, capsys): + """AC: stop follow, restart with last cursor, receive later events once in order.""" + run_dir = _run_dir(tmp_path) + seeded = _seed_lifecycle(run_dir) + assert cli.main(["runs", "events", str(run_dir)]) == 0 + first = _parse_ndjson(capsys.readouterr().out) + assert [row["sequence"] for row in first] == [1, 2] + cursor = first[-1]["cursor"] + + _append( + _journal_path(run_dir), + run_id=run_dir.name, + event_type="run.completed", + payload={"status": "ok", "detail": "done"}, + idempotency_key="complete-1", + expected_previous_sequence=2, + recorded_at="2026-07-31T12:00:02.000000Z", + ) + assert cli.main(["runs", "events", str(run_dir), "--after", cursor]) == 0 + second = _parse_ndjson(capsys.readouterr().out) + assert [row["sequence"] for row in second] == [3] + assert second[0]["event_type"] == "run.completed" + assert second[0]["cursor"] + # Restart with the same cursor again yields the same suffix once. + assert cli.main(["runs", "events", str(run_dir), "--after", cursor]) == 0 + again = _parse_ndjson(capsys.readouterr().out) + assert [row["event_digest"] for row in again] == [row["event_digest"] for row in second] + assert seeded[0].event_type == "run.created" + + +def test_events_completed_run_same_cursor_produces_same_suffix(tmp_path, capsys): + """AC: reading a completed run with the same cursor produces the same suffix.""" + run_dir = _run_dir(tmp_path) + events = _seed_lifecycle(run_dir, terminal=True) + cursor = _cursor_for(events[0]) + assert cli.main(["runs", "events", str(run_dir), "--after", cursor]) == 0 + first = _parse_ndjson(capsys.readouterr().out) + assert cli.main(["runs", "events", str(run_dir), "--after", cursor]) == 0 + second = _parse_ndjson(capsys.readouterr().out) + assert first == second + assert [row["sequence"] for row in first] == [2, 3] + + +@pytest.mark.parametrize( + "case", + ["other_run", "unknown_sequence", "digest_mismatch", "unsupported_schema"], +) +def test_events_cursor_failures_emit_no_unverified_suffix(tmp_path, capsys, case): + """AC: bad cursors fail closed without emitting an unverified suffix.""" + run_dir = _run_dir(tmp_path) + events = _seed_lifecycle(run_dir, terminal=True) + if case == "other_run": + other = _run_dir(tmp_path, "20260731-604000-other99") + other_events = _seed_lifecycle(other, terminal=True) + cursor = _cursor_for(other_events[0]) + elif case == "unknown_sequence": + cursor = run_event_cursor.encode( + run_event_cursor.DecodedCursor( + schema=run_event_cursor.SUPPORTED_SCHEMA, + run_id=run_dir.name, + sequence=99, + digest="c" * 64, + ) + ) + elif case == "digest_mismatch": + cursor = run_event_cursor.encode( + run_event_cursor.DecodedCursor( + schema=run_event_cursor.SUPPORTED_SCHEMA, + run_id=run_dir.name, + sequence=events[0].sequence, + digest="d" * 64, + ) + ) + else: + # Unsupported schema fails at decode before any emission. + raw = json.dumps( + { + "digest": "e" * 64, + "run_id": run_dir.name, + "schema": "brigade.run_event_cursor.v0", + "sequence": 1, + }, + sort_keys=True, + separators=(",", ":"), + ).encode() + import base64 + + cursor = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + rc = cli.main(["runs", "events", str(run_dir), "--after", cursor]) + captured = capsys.readouterr() + assert rc == 2 + assert captured.out == "" + assert "error:" in captured.err + + +def test_events_legacy_run_reports_legacy_no_journal(tmp_path, capsys): + run_dir = _run_dir(tmp_path) + (run_dir / "run.json").write_text(json.dumps({"status": "ok"}) + "\n") + rc = cli.main(["runs", "events", str(run_dir)]) + assert rc == 2 + assert "legacy-no-journal" in capsys.readouterr().err + + +def test_idempotent_request_id_replays_without_second_transport_call(tmp_path): + """AC: same request id + fingerprint returns first terminal result; no second transport call.""" + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + calls: list[dict] = [] + + def send(payload: dict[str, object]) -> dict[str, object]: + calls.append(dict(payload)) + return {"ok": True, "worker": "coder", "turn_id": "turn-1"} + + first = run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id="req-idem-1", + worker="coder", + text="keep going", + send=send, + ) + second = run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id="req-idem-1", + worker="coder", + text="keep going", + send=send, + ) + assert len(calls) == 1 + assert first.replayed is False + assert second.replayed is True + assert second.response["ok"] is True + assert second.response["turn_id"] == "turn-1" + report = run_journal.read_journal_bounded(_journal_path(run_dir)) + control_types = [event.event_type for event in report.events if event.event_type.startswith("control.")] + assert control_types == ["control.requested", "control.observed"] + for event in report.events: + assert "keep going" not in json.dumps(event.payload) + + +@pytest.mark.parametrize( + ("changed", "kwargs"), + [ + ("worker", {"worker": "reviewer", "text": "keep going", "op": "steer"}), + ("text", {"worker": "coder", "text": "different text", "op": "steer"}), + ("op", {"worker": "coder", "text": None, "op": "interrupt"}), + ("turn_id", {"worker": "coder", "text": "keep going", "op": "steer", "turn_id": "turn-9"}), + ], +) +def test_request_id_fingerprint_conflict(tmp_path, changed, kwargs): + """AC: same request id with changed worker/text/op/turn identity conflicts.""" + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + + def send(_payload: dict[str, object]) -> dict[str, object]: + return {"ok": True, "worker": "coder", "turn_id": "turn-1"} + + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id="req-conflict-1", + worker="coder", + text="keep going", + send=send, + ) + with pytest.raises(ControlJournalError) as exc: + run_control_journal.execute_control_request( + run_dir, + op=kwargs["op"], + request_id="req-conflict-1", + worker=kwargs.get("worker"), + text=kwargs.get("text"), + turn_id=kwargs.get("turn_id"), + send=send, + ) + assert exc.value.code == "control_fingerprint_conflict" + assert changed # parametrize label retained for AC mapping + + +def test_indeterminate_control_request_is_not_reissued(tmp_path): + """AC: requested without terminal observation reports indeterminate; no reissue.""" + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + fingerprint = run_control_journal.control_fingerprint_payload( + op="steer", + request_id="req-indet-1", + worker="coder", + text="steer me", + ) + run_journal.append_event( + _journal_path(run_dir), + run_id=run_dir.name, + event_type="control.requested", + payload=fingerprint, + idempotency_key=run_control_journal.requested_idempotency_key("req-indet-1"), + expected_previous_sequence=2, + recorded_at="2026-07-31T12:00:03.000000Z", + ) + calls: list[dict] = [] + + def send(payload: dict[str, object]) -> dict[str, object]: + calls.append(dict(payload)) + return {"ok": True, "worker": "coder", "turn_id": "turn-1"} + + with pytest.raises(ControlJournalError) as exc: + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id="req-indet-1", + worker="coder", + text="steer me", + send=send, + ) + assert exc.value.code == "indeterminate" + assert calls == [] + + +@pytest.mark.parametrize( + "crash_point", + ["before_requested", "after_requested", "after_transport", "after_terminal"], +) +def test_control_crash_points(tmp_path, monkeypatch, crash_point): + """AC: crash coverage before requested, after sync, after transport, after terminal.""" + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + request_id = f"req-crash-{crash_point}" + transport_calls: list[dict] = [] + + def send(payload: dict[str, object]) -> dict[str, object]: + transport_calls.append(dict(payload)) + return {"ok": True, "worker": "coder", "turn_id": "turn-1"} + + if crash_point == "before_requested": + # No prior journal mutation; a fresh attempt proceeds and calls transport once. + result = run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert result.replayed is False + assert len(transport_calls) == 1 + return + + if crash_point == "after_requested": + fingerprint = run_control_journal.control_fingerprint_payload( + op="steer", + request_id=request_id, + worker="coder", + text="go", + ) + run_journal.append_event( + _journal_path(run_dir), + run_id=run_dir.name, + event_type="control.requested", + payload=fingerprint, + idempotency_key=run_control_journal.requested_idempotency_key(request_id), + expected_previous_sequence=2, + recorded_at="2026-07-31T12:00:03.000000Z", + ) + with pytest.raises(ControlJournalError) as exc: + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert exc.value.code == "indeterminate" + assert transport_calls == [] + return + + if crash_point == "after_transport": + original_append = run_control_journal.run_journal.append_event + + def crash_on_terminal(*args, **kwargs): + event_type = kwargs.get("event_type") + if event_type in {"control.observed", "control.failed"}: + raise RuntimeError("simulated crash after transport before terminal append") + return original_append(*args, **kwargs) + + monkeypatch.setattr(run_control_journal.run_journal, "append_event", crash_on_terminal) + with pytest.raises(RuntimeError, match="simulated crash"): + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert len(transport_calls) == 1 + monkeypatch.undo() + # Retry after the crash must report indeterminate and must not reissue. + with pytest.raises(ControlJournalError) as exc: + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert exc.value.code == "indeterminate" + assert len(transport_calls) == 1 + return + + # after_terminal: completed request replays without transport. + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert len(transport_calls) == 1 + replay = run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id=request_id, + worker="coder", + text="go", + send=send, + ) + assert replay.replayed is True + assert len(transport_calls) == 1 + + +def test_cli_steer_with_request_id_on_journal_run(tmp_path, capsys): + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + registry = run_control.LiveTurnRegistry() + + class _Thread: + def steer(self, text, turn_id): + assert text == "please finish" + assert turn_id == "turn-cli-1" + + def interrupt(self, turn_id): + raise AssertionError("interrupt not expected") + + registry.register("coder", _Thread(), "turn-cli-1") + socket_path = run_dir / "control.sock" + server = run_control.ControlServer(socket_path, registry) + transport = server.start() + (run_dir / "run.json").write_text( + json.dumps( + { + "status": "dispatching", + "codex_transport": "app-server", + "control_transport": transport.to_metadata(), + "control_socket": str(socket_path), + } + ) + + "\n" + ) + try: + rc = cli.main( + [ + "runs", + "steer", + str(run_dir), + "coder", + "please", + "finish", + "--request-id", + "req-cli-1", + ] + ) + finally: + server.close() + captured = capsys.readouterr() + assert rc == 0 + assert "request_id: req-cli-1" in captured.out + assert "steer: coder" in captured.out + # Replay + rc = cli.main( + [ + "runs", + "steer", + str(run_dir), + "coder", + "please", + "finish", + "--request-id", + "req-cli-1", + ] + ) + captured = capsys.readouterr() + assert rc == 0 + assert "control: replay" in captured.out + + +def test_cli_explicit_request_id_on_legacy_run_fails(tmp_path, capsys): + run_dir = _run_dir(tmp_path) + (run_dir / "run.json").write_text( + json.dumps({"status": "started", "codex_transport": "app-server", "control_socket": str(run_dir / "x.sock")}) + + "\n" + ) + rc = cli.main(["runs", "steer", str(run_dir), "coder", "go", "--request-id", "req-legacy"]) + assert rc == 2 + assert "legacy-no-journal" in capsys.readouterr().err + + +def test_events_follow_exits_after_terminal_event(tmp_path, capsys): + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + result: dict[str, int] = {} + + def reader(): + result["rc"] = cli.main(["runs", "events", str(run_dir), "--follow", "--interval", "0.05"]) + + thread = threading.Thread(target=reader, daemon=True) + thread.start() + time.sleep(0.1) + _append( + _journal_path(run_dir), + run_id=run_dir.name, + event_type="run.completed", + payload={"status": "ok", "detail": "done"}, + idempotency_key="complete-1", + expected_previous_sequence=2, + recorded_at="2026-07-31T12:00:02.000000Z", + ) + thread.join(timeout=5.0) + assert result.get("rc") == 0 + rows = _parse_ndjson(capsys.readouterr().out) + assert [row["event_type"] for row in rows] == [ + "run.created", + "run.dispatching.started", + "run.completed", + ] + + +def test_control_payload_never_contains_steering_text(tmp_path): + """AC: lifecycle control events carry digests, not steering text.""" + run_dir = _run_dir(tmp_path) + _seed_lifecycle(run_dir) + secret = "super-secret-steering-instructions" + + def send(_payload: dict[str, object]) -> dict[str, object]: + return {"ok": True, "worker": "coder", "turn_id": "turn-1"} + + run_control_journal.execute_control_request( + run_dir, + op="steer", + request_id="req-privacy-1", + worker="coder", + text=secret, + send=send, + ) + raw = _journal_path(run_dir).read_text() + assert secret not in raw + report = run_journal.read_journal_bounded(_journal_path(run_dir)) + requested = next(event for event in report.events if event.event_type == "control.requested") + assert requested.payload["text_digest"] == run_control_journal.text_digest(secret) + assert "text" not in requested.payload diff --git a/tests/test_run_projector.py b/tests/test_run_projector.py index 6dcb11c7..1d4ca7e0 100644 --- a/tests/test_run_projector.py +++ b/tests/test_run_projector.py @@ -747,3 +747,40 @@ def test_artifact_collection_started_derives_artifact_collection(): ) assert projection.status == "artifact-collection" assert projection.snapshot["status"] == "artifact-collection" + + +@pytest.mark.parametrize( + "event_type", + ["control.requested", "control.observed", "control.failed"], +) +def test_control_events_are_status_neutral_and_advance_chain_cursor(event_type): + created = _build_event(1, "run.created", {"status": "started"}, "create-1", RECORDED_AT, None) + if event_type == "control.requested": + payload = { + "op": "steer", + "worker": "coder", + "text_digest": "b" * 64, + "request_id": "req-1", + } + elif event_type == "control.observed": + payload = {"op": "steer", "worker": "coder", "request_id": "req-1", "detail": "ok"} + else: + payload = { + "op": "steer", + "worker": "coder", + "request_id": "req-1", + "code": "no-active-turn", + "detail": "no active turn", + } + control = _build_event( + 2, + event_type, + payload, + f"{event_type}:req-1", + "2026-07-27T15:30:46.000000Z", + created["event_digest"], + ) + projection = project_run_snapshot(_minimal_base_snapshot(), [created, control], journal_present=True) + assert projection.status == "started" + assert projection.last_sequence == 2 + assert projection.last_event_digest == control["event_digest"] From 58a3ed75e82411c4c9009df3d1394aeea4926c83 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Fri, 31 Jul 2026 23:52:37 -0400 Subject: [PATCH 2/2] docs(roadmap): refresh command inventory for runs events Co-authored-by: Cursor --- docs/command-inventory.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/command-inventory.md b/docs/command-inventory.md index 189e29b5..ce8f990f 100644 --- a/docs/command-inventory.md +++ b/docs/command-inventory.md @@ -56,7 +56,7 @@ enabled: run `brigade extras on` once, or set `BRIGADE_EXTRAS=1`. - `brigade route`: 1 command path(s) - `brigade run`: 1 command path(s) - `brigade runbook` (extras): 5 command path(s) -- `brigade runs`: 9 command path(s) +- `brigade runs`: 10 command path(s) - `brigade scrub`: 1 command path(s) - `brigade search`: 6 command path(s) - `brigade security`: 15 command path(s) @@ -439,6 +439,7 @@ enabled: run `brigade extras on` once, or set `BRIGADE_EXTRAS=1`. - `brigade runbook plan` (extras) - `brigade runbook resume` (extras) - `brigade runbook run` (extras) +- `brigade runs events` - `brigade runs interrupt` - `brigade runs latest` - `brigade runs list`