From daae8dea0b474a4c133ec4dbc5d672b141b202f7 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Sun, 26 Jul 2026 17:17:27 -0400 Subject: [PATCH] chore(receipts): audit and canonicalize receipt schemas Co-Authored-By: Codex --- docs/receipt-schema-audit.md | 184 +++++++++++ docs/receipt-schemas.md | 467 +++++++++++++++++++++++++++ src/brigade/aboyeur.py | 61 ++-- src/brigade/outcome_cmd.py | 4 +- src/brigade/receipt_schema.py | 87 +++++ src/brigade/run_resume.py | 26 +- src/brigade/runguard.py | 4 +- src/brigade/work_cmd/verification.py | 22 +- tests/test_receipt_schema_audit.py | 457 ++++++++++++++++++++++++++ 9 files changed, 1256 insertions(+), 56 deletions(-) create mode 100644 docs/receipt-schema-audit.md create mode 100644 docs/receipt-schemas.md create mode 100644 src/brigade/receipt_schema.py create mode 100644 tests/test_receipt_schema_audit.py diff --git a/docs/receipt-schema-audit.md b/docs/receipt-schema-audit.md new file mode 100644 index 00000000..92c98417 --- /dev/null +++ b/docs/receipt-schema-audit.md @@ -0,0 +1,184 @@ +# Receipt JSON SIEM-readiness audit (#506) + +Audit date: 2026-07-26. Scope: the three sanctioned receipt families on `main`: +work verification (`src/brigade/work_cmd/verification.py`), Brigade run receipts +(`src/brigade/run_receipts.py` serializers plus `run.json` and run sidecars in +`aboyeur.py` / `run_resume.py`), and outcome records (`src/brigade/outcome_cmd.py`). +Out of scope: route-decision, runbook, tool-call, daily, skills, and other +receipt producers. + +Criteria: + +1. **Deterministic field names**: stable string keys, no positional-only meaning. +2. **Stable key ordering**: canonical sorted-key JSON on disk. +3. **Append-only semantics**: new evidence is appended or written once. No silent + rewrite of historical facts. +4. **Schema version**: machine-readable `schema_version` (and/or documented + `schema` string) for evolution. + +Patch identity from **#485** is present on the audited `main`. Lane **#491** +(telemetry projection) may add fields concurrently. Within a given +`schema_version`, **additive fields only** are permitted. Consumers must ignore +unknown keys. + +Tag legend: `compliant`, `fixed-here`, `needs-follow-up-issue`. + +## Writer-site inventory + +### Work verification (`src/brigade/work_cmd/verification.py`) + +| Function | Receipt | Behavior | +| --- | --- | --- | +| `_run_verify_commands` | `receipt.json` | Builds in-memory receipt. Disk write deferred to finalize | +| `_finalize_verify_receipt` | `receipt.json`, `summary.md` | **Single** `helpers._write_json` for `receipt.json`. Markdown is best-effort. Retention pruning follows | +| `_safe_finalize_verify_receipt` | `receipt.json` | Exception wrapper. Emergency write only when finalize itself raises | +| `_write_reused_receipt` | `receipt.json` | New verify-run directory. Copies prior commands. Retention pruning follows | +| `_prune_verify_runs` | verify-run directories | Deletes directories older than the newest 50, including their written receipts | +| `_work_closeout_payload` | `closeout.json` | One write per closeout via `helpers._write_json` | +| `verify_run` | (dispatch) | Entry point for `_run_verify_commands` | + +### Run lifecycle `run.json` + +| Module / function | Behavior | +| --- | --- | +| `aboyeur.record_run_start` | Initial `run.json` via `_run_payload` + `_write_json`. Detached and regular paths may run before the main loop writes the latest snapshot | +| `aboyeur` run loop (`run`, `plan`, dispatch/synthesis paths) | **In-place** `run.json` rewrites on status transitions (`planning`, `dispatching`, `synthesizing`, terminal states) | +| `aboyeur.record_run_termination` | Merge terminal failure/success into existing `run.json` | +| `aboyeur.record_artifact_collection` | Merge `artifact_collection` block into `run.json` | +| `aboyeur.record_dispatch_stage` | Merge dispatch stage metadata | +| `aboyeur.record_result_processing` | Merge result-processing seat metadata | +| `aboyeur.terminal_sigterm_handler` | SIGTERM path calls `record_run_termination` | +| `run_resume._resume_locked` | Rewrites `run.json` after resume synthesis | +| `runguard._recover_run_artifact` | Stale-lock recovery rewrites `run.json` via `localio.write_json` | +| `cli/run.py` | Error/exit paths call `record_run_termination` | + +Shared serializer: `aboyeur._run_payload` (schema + `schema_version`). Shared writer: +`aboyeur._write_json` (`sort_keys=True`, atomic). + +### Run sidecars (`.brigade/runs//`) + +| Module / function | File | Builder | +| --- | --- | --- | +| `aboyeur.record_run_start` | `roster.json` | `aboyeur._roster_payload` | +| `aboyeur` planning phase | `plan.json` | `receipt_schema.run_plan_document` | +| `aboyeur` post-dispatch | `worker-results.json` | `receipt_schema.worker_results_document` | +| `aboyeur` post-synthesis | `synthesis.json` | `receipt_schema.synthesis_document` | +| `aboyeur.set_artifact_patch_ref` | `worker-results.json`, `synthesis.json` | Rewrites `ground_truth.patch_ref` on existing sidecars | +| `run_resume._resume_locked` | `worker-results.json`, `synthesis.json` | Same builders after resume | + +Entry serialization for worker rows: `run_receipts.worker_payload`, +`run_receipts.assignment_payload`, `run_receipts.agent_result_payload`. + +The run directory also contains support artifacts that are not receipts: + +| Artifact | Classification | +| --- | --- | +| `pre-run-snapshot.json` | Run-guard input snapshot, not an event receipt | +| `plan-attempts.json` | Planner retry trace, not the accepted `plan.json` receipt | +| `read-only-enforcement.json` | Enforcement evidence sidecar, not a run-state receipt | +| GraphTrail before/after/delta JSON | Code-graph evidence referenced by receipts, not a Brigade receipt type | + +### Outcome (`src/brigade/outcome_cmd.py`) + +| Function | Receipt | Behavior | +| --- | --- | --- | +| `append_records` | `memory/outcome/records.jsonl` | JSONL append, one sorted object per line | +| `record` | (dispatch) | Builds `OutcomeRecord`, calls `append_records` | +| `_record_payload` | row shape | Adds `schema_version` before append | +| `reconcile` | `memory/outcome/decisions/-.json` | One decision file per applied transition via `localio.write_json` | +| `_decision_path` | decision filename | Second-resolution timestamp plus a lossy artifact slug can collide | + +Readers: `outcome_cmd._read_run_receipt` (run.json for capture), `load_records` +(legacy rows without `schema_version` accepted). + +## Verification family + +| Receipt type | Path | Names | Ordering | Append-only | Schema version | Tag | +| --- | --- | --- | --- | --- | --- | --- | +| Work verify run | `.brigade/work/verify-runs//receipt.json` | Stable snake_case, command objects use fixed keys and a fixed identity tuple | `helpers._write_json` (`sort_keys=True`) | Finalization now writes once, but `_prune_verify_runs` deletes receipts beyond the newest 50 | `schema_version: 2` | names **compliant**, ordering **compliant**, write-once **fixed-here**, retention **needs-follow-up-issue**, version **compliant** | +| Work verify reuse | same layout | Copies prior `commands`, same key set and a fresh identity tuple | same | New directory per reuse, followed by the same retention pruning | `schema_version: 2` | names **compliant**, ordering **compliant**, append-only **needs-follow-up-issue**, version **compliant** | +| Work closeout | `.brigade/work/closeouts//closeout.json` | Stable keys, nested session/verification summaries | `helpers._write_json` | Written once per closeout | `schema_version: 1` | names **compliant**, ordering **compliant**, append-only **compliant**, version **fixed-here** | + +Notes: + +- Verify `commands[].env` is a sorted list of `KEY=value` strings (**compliant**). +- `digests.receipt_sha256` uses `localio.canonical_json_digest` (sorted keys) + (**compliant**). +- The 50-run retention cap is operationally useful, but deletion is not + append-only evidence storage (**needs-follow-up-issue**). + +## Brigade run family + +| Receipt type | Path | Names | Ordering | Append-only | Schema version | Tag | +| --- | --- | --- | --- | --- | --- | --- | +| Run lifecycle | `.brigade/runs//run.json` | Stable keys, status values enumerated | `aboyeur._write_json` (`sort_keys=True`) | **In-place lifecycle mutation** across statuses | `schema` + `schema_version: 1` | names **compliant**, ordering **fixed-here**, append-only **needs-follow-up-issue**, version **fixed-here** | +| Roster snapshot | `roster.json` | `schema` + roster fields | sorted writer | Write-once at run start | `schema` + `schema_version: 1` | names **compliant**, ordering **fixed-here**, append-only **compliant**, version **fixed-here** | +| Run plan | `plan.json` | `schema` + `assignments` | sorted writer | Write-once per planning phase | `schema` + `schema_version: 1` | names **compliant**, ordering **fixed-here**, append-only **compliant**, version **fixed-here** | +| Worker results | `worker-results.json` | `schema` + `results` from `run_receipts.py` | sorted writer | `run_resume._resume_locked` and `aboyeur.set_artifact_patch_ref` rewrite the sidecar | `schema` + `schema_version: 1` | names **compliant**, ordering **fixed-here**, append-only **needs-follow-up-issue**, version **fixed-here** | +| Synthesis | `synthesis.json` | `schema` + orchestrator/result | sorted writer | `run_resume._resume_locked` and `aboyeur.set_artifact_patch_ref` rewrite the sidecar | `schema` + `schema_version: 1` | names **compliant**, ordering **fixed-here**, append-only **needs-follow-up-issue**, version **fixed-here** | + +Notes: + +- `run.json` omits explicit JSON `null` for absent `cwd` (**fixed-here**). +- `record_run_termination` / `runguard._recover_run_artifact` merge into existing + `run.json`. The latest-snapshot file supports `brigade runs watch` polling. Append-only lifecycle events + deferred (**needs-follow-up-issue**). +- Resume salvage and patch-ref binding both overwrite `worker-results.json` and + `synthesis.json`. Preserving each attempt as a new sidecar is deferred + (**needs-follow-up-issue**). +- Patch identity from #485 uses verify `schema_version: 2`. Future #491 telemetry + cross-references must remain additive within the version used by each receipt + family. + +### Null and absent-field evidence + +These cases support the deterministic-names findings in the family tables: + +| Shape | Policy | +| --- | --- | +| Verify `baseline_commit`, `tree_fingerprint`, `changes_patch_sha256` | Explicit `null` tuple when identity capture is unavailable in version 2 | +| Verify `reused_from` | Omitted when no source run id is available | +| Verify command `exit_code` | Explicit `null` means no child exit status exists | +| Run `cwd` | Omitted when unavailable | +| Roster agent `env`, model metadata | Fixed snapshot rows retain explicit `null` | +| Synthesis `orchestrator` | Explicit `null` identifies direct-worker mode | +| Work closeout `task`, `verification` | Explicit `null` records that no item was available | +| Outcome `prev_digest` | Explicit `null` identifies the first chain row | + +## Outcome family + +| Receipt type | Path | Names | Ordering | Append-only | Schema version | Tag | +| --- | --- | --- | --- | --- | --- | --- | +| Outcome ledger row | `memory/outcome/records.jsonl` | Stable snake_case from `OutcomeRecord` | `json.dumps(..., sort_keys=True)` per line | JSONL append with digest chain. Concurrent writers can select the same predecessor | `schema_version: 1` | names **compliant**, ordering **compliant**, append-only **needs-follow-up-issue**, version **fixed-here** | +| Reconcile decision | `memory/outcome/decisions/-.json` | Stable keys, nested `score` | `localio.write_json` | Second-resolution, lossy-slug filenames can overwrite a prior decision | `schema_version: 1` | names **compliant**, ordering **compliant**, append-only **needs-follow-up-issue**, version **fixed-here** | +| Status cache | `memory/outcome/status.json` | `version` + `artifacts` map | sorted write | Regenerated from decisions (derived cache) | `version: 1` (file format, not receipt `schema_version`) | names **compliant**, ordering **compliant**, append-only **compliant**, version **compliant** (documented derived cache) | + +Notes: + +- Legacy rows without `schema_version` still load via `_record_from_dict` and + `_read_run_receipt` (**fixed-here** backward-compat tests). +- `append_records` does not lock the last-digest read and append as one + transaction (**needs-follow-up-issue**). + +## Follow-up issues (draft titles) + +1. **`receipts: replace mutable run.json snapshots with append-only lifecycle events`** +2. **`receipts: preserve worker and synthesis sidecars across resume and patch-ref updates`** +3. **`receipts: archive verification evidence before retention pruning`** +4. **`outcome: make decision receipt filenames collision-safe and write-exclusive`** +5. **`outcome: serialize digest-chain appends across concurrent captures`** + +## Finding counts + +Counts are tag occurrences across the four criterion findings for each receipt +type. The work-verify append-only cell has both a fixed write-once defect and a +remaining retention follow-up. + +| Tag | Count | +| --- | --- | +| **compliant** | 24 | +| **fixed-here** | 14 | +| **needs-follow-up-issue** | 7 | + +See [`receipt-schemas.md`](receipt-schemas.md) for field-level contracts and +evolution rules. diff --git a/docs/receipt-schemas.md b/docs/receipt-schemas.md new file mode 100644 index 00000000..bd401acd --- /dev/null +++ b/docs/receipt-schemas.md @@ -0,0 +1,467 @@ +# Brigade receipt schema reference + +Machine-readable contracts for the three sanctioned receipt families audited in +[#506](receipt-schema-audit.md). These are documentation contracts, not runtime +JSON Schema files. + +## Evolution rules (all families) + +1. **`schema_version` is an integer.** Bump only for breaking changes (rename, + remove, or change the type/meaning of an existing field). +2. **Within a `schema_version`, evolution is additive only.** New optional fields + may appear (including #491 telemetry projection). Patch identity from #485 + is already represented by verify `schema_version: 2`. Consumers **must + ignore unknown keys**. +3. **Absent vs null:** optional fields are **omitted** when unset. Producers should + not write JSON `null` for optional top-level fields unless the field's presence + carries semantic meaning. Intentional nullable shapes: + - Verify version 2 always includes `baseline_commit`, `tree_fingerprint`, and + `changes_patch_sha256`. All three are `null` when identity capture is + unavailable, keeping the binding tuple structurally complete. + - Verify `commands[].exit_code` may be `null` when an interrupted or timed-out + child has no exit status. A command rejected before execution uses exit code `2`. + - Run `scheduler.used` may be `null` while a run is in flight. + - Run `roster.json` agent rows may include `"env": null` when the seat has no + env overrides (snapshot preserves the roster table shape). + - Synthesis `orchestrator` may be explicit JSON `null` in direct-worker mode. + - Outcome `prev_digest` may be `null` on the first ledger row. + - Work closeout `task` and `verification` may be `null` when absent. + - Reused verify receipts **omit** `reused_from` when the source receipt lacks + a `run_id`. +4. **Serialization:** on-disk JSON uses UTF-8, indent 2, trailing newline, and + `sort_keys=True` at every object level unless noted (JSONL: one sorted object + per line). +5. **Readers** must accept records **without** `schema_version` (pre-#506 storage). + +--- + +## `brigade.work_verify_receipt`: `schema_version: 2` + +**Path:** `.brigade/work/verify-runs//receipt.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema_version` | integer | yes (new writes) | Always `2` for this contract | +| `run_id` | string | yes | Timestamp-prefixed id | +| `target` | string | yes | Absolute workspace path | +| `status` | string | yes | `running`, `completed`, `failed`, `rejected`, `canceled` | +| `started_at` | string (ISO-8601) | yes | UTC timestamp | +| `completed_at` | string (ISO-8601) | no | Set at finalize | +| `duration_seconds` | number | no | Wall time | +| `timeout` | integer | no | Per-command timeout seconds | +| `path` | string | yes | Run directory path | +| `commands` | array of object | yes | See command object below | +| `planned_commands` | array of string | no | Display argv joined | +| `evidence` | object | no | Workspace evidence snapshot | +| `baseline_commit` | string \| null | yes | Verified Git baseline, or `null` when identity capture is unavailable | +| `tree_fingerprint` | string \| null | yes | Verified Git tree hash, or `null` with the unavailable identity tuple | +| `changes_patch_sha256` | string \| null | yes | SHA-256 of `changes.patch`, or `null` with the unavailable identity tuple | +| `git` | object | no | `{head, branch, dirty_files}` | +| `code_graph_delta` | object | no | GraphTrail summary | +| `harness_session` | object | no | `{harness, fingerprint}` | +| `digests` | object | no | `{algorithm, logs, receipt_sha256, signature?, key_id?}` | +| `reused_from` | string | no | Prior run id when reused | +| `interruption` | object | no | Cancel metadata | + +**Command object** + +| Field | Type | Notes | +| --- | --- | --- | +| `command` | string | Display command | +| `argv` | array of string | no | Resolved argv | +| `env` | array of string | Sorted `KEY=value` pairs | +| `status` | string | `completed`, `failed`, `timed_out`, `rejected`, `interrupted` | +| `exit_code` | integer \| null | Child exit status. `null` when interrupted or timed out without one. Rejected commands use `2` | +| `started_at`, `completed_at` | string | ISO-8601 | +| `duration_seconds` | number | | +| `stdout_summary`, `stderr_summary` | string | | +| `stdout_log_path`, `stderr_log_path` | string | Paths under run dir | + +--- + +## `brigade.work_closeout`: `schema_version: 1` + +**Path:** `.brigade/work/closeouts//closeout.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema_version` | integer | yes (new writes) | `1` | +| `closeout_id` | string | yes | | +| `target` | string | yes | | +| `status` | string | yes | `ready` or `blocked` | +| `ready` | boolean | yes | | +| `created_at` | string | yes | ISO-8601 | +| `session` | object | yes | Session summary | +| `session_path` | string | yes | | +| `task` | object \| null | | Task summary | +| `acceptance_criteria` | array | yes | | +| `verification` | object \| null | | Latest verify receipt ref | +| `scanner_sweep` | object | yes | | +| `code_review` | object | yes | | +| `handoff_drafts` | object | yes | | +| `blockers` | array of string | yes | | + +**Session summary** (`session` object) + +| Field | Type | Notes | +| --- | --- | --- | +| `path` | string | Session directory | +| `id` | string | Session id | +| `status` | string | Session status | +| `title` | string \| null | | +| `started_at`, `ended_at` | string \| null | ISO-8601 | +| `note`, `latest_note` | string \| null | | +| `handoff` | object \| null | Handoff metadata when present | +| `branch` | string \| null | Git branch from snapshot | +| `dirty_files` | integer | Count from snapshot | +| `next` | string \| null | Suggested next step | + +**Verification summary** (`verification` object, when present) + +| Field | Type | Notes | +| --- | --- | --- | +| `run_id` | string | Latest verify run id | +| `status` | string | Verify receipt status | +| `path` | string | Verify run directory | +| `command_count` | integer | Number of command records | + +--- + +## `brigade.run.v1`: `schema_version: 1` + +**Path:** `.brigade/runs//run.json` + +The required column below describes normal run creation. Stale-lock recovery may +create the partial recovery variant documented after the main table when the +original file is missing, corrupt, or not an object. + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | Always `brigade.run.v1` | +| `schema_version` | integer | yes (new writes) | `1` | +| `task` | string | yes | | +| `cwd` | string | no | Omitted when unknown | +| `orchestrator` | string | yes | Seat name | +| `dry_run`, `read_only` | boolean | yes | | +| `status` | string | yes | Lifecycle status | +| `started_at`, `status_started_at` | string | yes | ISO-8601 UTC (`Z`) | +| `finished_at` | string | no | | +| `duration_seconds` | number | no | | +| `suspected_noop` | boolean | yes | | +| `code_graph_brief`, `drift_impact_brief`, `evidence_brief`, `brief_budget` | object | yes | Brief attachment summaries | +| `scheduler` | object | no | `{requested, used, fallback_reason}` | +| `roster` | object | no | Resolution metadata | +| `lock_workspace` | string | no | | +| `route` | object | no | Routing brief | +| `worker` | string | no | Direct-worker seat | +| `git` | object | no | | +| `pre_run_snapshot` | object | no | Run-guard snapshot | +| `code_graph_delta` | object | no | | +| `context_eval` | object | no | | +| `artifacts` | string | no | Output directory | +| `handoff` | string | no | Handoff path | +| `error` | string | no | | +| `failure_phase`, `failure_kind` | string | no | | +| `failure` | object | no | `{phase, kind, detail, seat?}` | +| `transport_warning` | object | no | | +| `codex_transport` | string | no | | +| `control_transport`, `control_socket` | object / string | no | | +| `active_stage` | integer | no | Current dispatch stage | +| `active_seats` | array of string | no | Seats active in the current dispatch stage | +| `phase_owner` | string | no | Seat responsible for result processing | +| `artifact_collection` | object | no | Artifact-retention result | +| `resumed_at` | array of string | no | ISO-8601 resume timestamps | +| `recovery_history` | array of object | no | Prior failure objects retained after a successful resume | + +**Partial stale-recovery variant** + +When no valid original `run.json` object survives, `runguard._recover_run_artifact` +writes this smaller receipt: + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | `brigade.run.v1` | +| `schema_version` | integer | yes | `1` | +| `artifacts` | string | yes | Recovered run directory | +| `recovery_preserved_artifact` | string | no | Renamed corrupt source, when one existed | +| `cwd`, `lock_workspace` | string | no | Recovered from lock metadata | +| `started_at` | string | no | Recovered lock acquisition time | +| `status`, `status_started_at`, `finished_at` | string | yes | Terminal recovery state and timestamps | +| `error`, `failure_phase` | string | yes | Recovery summary | +| `failure` | object | yes | Stale-lock failure variant documented below | + +**Brief attachment objects** (`code_graph_brief`, `drift_impact_brief`, `evidence_brief`) + +| Field | Type | Notes | +| --- | --- | --- | +| `attached` | boolean | Whether the brief was attached | +| `bytes` | integer | Serialized brief size | + +`drift_impact_brief` also includes `pending_count` (integer). + +**`brief_budget` object** + +| Field | Type | Notes | +| --- | --- | --- | +| `bytes` | integer | Budget ceiling | +| `attached` | array of object | Rows shaped `{name: string, bytes: integer, truncated: boolean}` | + +**`scheduler` object** + +| Field | Type | Notes | +| --- | --- | --- | +| `requested` | string | Requested scheduler name | +| `used` | string \| null | Resolved scheduler. `null` while unresolved | +| `fallback_reason` | string \| null | `null` unless a fallback scheduler was used | + +**`roster` object** (resolution metadata on `run.json`) + +| Field | Type | Notes | +| --- | --- | --- | +| `path` | string | Resolved roster file path | +| `source` | string | Resolution source label | +| `shadowed` | array of string | Shadowed roster paths | + +**`failure` object** + +| Field | Type | Notes | +| --- | --- | --- | +| `phase` | string | Failure phase | +| `kind` | string | Failure kind | +| `detail` | string | Human-readable detail | +| `seat` | string | Optional single seat attribution | +| `seats` | array of string | Optional multi-seat attribution | +| `owner_pid` | integer | Stale-lock recovery only | +| `prior_status` | string | Stale-lock recovery only | +| `recovered_at` | string | Stale-lock recovery only, ISO-8601 | + +**`artifact_collection` object** + +| Field | Type | Notes | +| --- | --- | --- | +| `status` | string | `ok` or `failed` | +| `patch_ref` | string | Relative patch path when collected | +| `changed` | boolean | Whether the worktree changed | +| `tracked_count`, `untracked_count` | integer | Change counts | +| `worktree` | string | Detached worktree path when used | +| `failure` | object | Same shape as `failure` when collection failed | + +**Lifecycle:** this file is **updated in place** during a run. Treat each write as +the latest snapshot, not an append-only log. Sidecars (`roster.json`, `plan.json`, +`worker-results.json`, `synthesis.json`) are write-once per phase (resume salvage +and patch-ref binding may rewrite worker/synthesis artifacts). + +--- + +## `brigade.roster_snapshot.v1`: `schema_version: 1` + +**Path:** `.brigade/runs//roster.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | Always `brigade.roster_snapshot.v1` | +| `schema_version` | integer | yes (new writes) | `1` | +| `orchestrator` | string | yes | Seat name | +| `max_workers` | integer | yes | | +| `timeout_seconds` | integer \| null | yes | | +| `allow_models` | array of string | yes | | +| `sandbox` | string \| null | yes | | +| `agents` | object | yes | Seat name → agent row (below) | + +**Agent row** (values in `agents`) + +| Field | Type | Notes | +| --- | --- | --- | +| `cli` | string \| null | CLI adapter name for direct seats | +| `model` | string \| null | Model id | +| `reasoning` | string \| null | Reasoning effort tier | +| `transport` | string | `direct`, `acpx`, `app-server`, etc. | +| `transport_version` | string \| null | Transport adapter version | +| `role` | string | `orchestrator` or `worker` | +| `timeout_seconds` | number \| null | Per-seat timeout override | +| `invalid_final_fallback` | string \| null | Fallback seat for invalid finals | +| `read_only_capable` | boolean | Whether the seat may run read-only | +| `env` | object \| null | Env override table (names/refs only) | + +--- + +## `brigade.run_plan.v1`: `schema_version: 1` + +**Path:** `.brigade/runs//plan.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | Always `brigade.run_plan.v1` | +| `schema_version` | integer | yes (new writes) | `1` | +| `assignments` | array of object | yes | See assignment object below | + +**Assignment object** (`run_receipts.assignment_payload`) + +| Field | Type | Notes | +| --- | --- | --- | +| `stage` | integer | Dispatch stage number | +| `worker` | string | Assigned seat name | +| `task` | string | Task text for the worker | +| `covers` | array of string | Optional covered artifact ids | + +--- + +## `brigade.worker_results.v1`: `schema_version: 1` + +**Path:** `.brigade/runs//worker-results.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | Always `brigade.worker_results.v1` | +| `schema_version` | integer | yes (new writes) | `1` | +| `results` | array of object | yes | Worker result entries (below) | +| `ground_truth` | object | no | No-op / ground-truth metadata when present | + +**Worker result entry** (`run_receipts.worker_payload`) + +| Field | Type | Notes | +| --- | --- | --- | +| `worker` | string | Seat name | +| `task` | string | Assigned task text | +| `ok` | boolean | Whether the worker succeeded | +| `detail` | string | Failure or status detail | +| `text` | string | Worker output text | +| `transport` | string | Transport used | +| `failure_phase` | string | Optional failure phase | +| `failure_kind` | string | Optional failure kind | +| `transport_warning` | object | Optional transport warning metadata | +| `thread_id` | string | App-server thread id when resumable | +| `status` | string | App-server turn status (with `thread_id`) | +| `exit_code` | integer | Optional child exit code | +| `timed_out` | boolean | Present when exit metadata or timeout applies | +| `stdout_log`, `stderr_log` | string | Optional log paths under the run dir | +| `duration_seconds` | number | Optional wall time | +| `requested_model` | string | Optional requested model | +| `effective_model` | string | Optional resolved model | +| `reasoning` | string | Optional reasoning tier | +| `stop_reason` | string | Optional terminal reason | +| `protocol_version` | integer | Optional protocol version | +| `session_id` | string | Optional session id | +| `request_id` | string | Optional request id | +| `acpx_version` | string | Optional ACPX adapter version | +| `events` | array of object | Optional redacted transport events | +| `env_overrides` | array of string | Sorted env override key names | +| `endpoint_host` | string | Comma-joined endpoint hosts from env | +| `attempts` | array of object | Optional retry log (below) | + +**Attempt object** (`run_receipts._attempt_payload`) + +| Field | Type | Notes | +| --- | --- | --- | +| `kind` | string | Attempt kind label | +| `worker` | string | Seat name | +| `task` | string | Task text | +| `transport` | string | Transport used | +| `model` | string \| null | Model id | +| `reasoning` | string \| null | Reasoning tier | +| `started_at`, `finished_at` | string | ISO-8601 timestamps | +| `exit_code` | integer \| null | Child exit code | +| `terminal_reason` | string | Terminal status label | +| `failure_phase` | string \| null | Failure phase when applicable | +| `failure_kind` | string \| null | Failure kind when applicable | +| `session_id` | string \| null | Session id when applicable | +| `selected` | boolean | Whether this attempt was selected | +| `stdout_log`, `stderr_log` | string | Optional log paths | + +--- + +## `brigade.synthesis.v1`: `schema_version: 1` + +**Path:** `.brigade/runs//synthesis.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema` | string | yes | Always `brigade.synthesis.v1` | +| `schema_version` | integer | yes (new writes) | `1` | +| `orchestrator` | string \| null | no | Orchestrator seat. `null` in direct-worker mode | +| `worker` | string | no | Direct-worker seat name when `mode` is `direct-worker` | +| `mode` | string | no | `direct-worker` when dispatch skipped planning/synthesis | +| `result` | object | yes | `{ok, detail, text}` from `run_receipts.agent_result_payload` | +| `ground_truth` | object | no | Copied from worker-results when present | + +**Synthesis `result` object** (`run_receipts.agent_result_payload`) + +| Field | Type | Notes | +| --- | --- | --- | +| `ok` | boolean | Whether synthesis succeeded | +| `detail` | string | Failure or status detail | +| `text` | string | Synthesis output text | +| `transport` | string | Transport used | +| `failure_phase` | string | Optional failure phase | +| `failure_kind` | string | Optional failure kind | +| `transport_warning` | object | Optional transport warning metadata | +| `exit_code` | integer | Optional child exit code | +| `timed_out` | boolean | Present when exit metadata or timeout applies | +| `stdout_log`, `stderr_log` | string | Optional log paths | +| `duration_seconds` | number | Optional wall time | +| `requested_model` | string | Optional requested model | +| `effective_model` | string | Optional resolved model | +| `reasoning` | string | Optional reasoning tier | +| `stop_reason` | string | Optional terminal reason | +| `protocol_version` | integer | Optional protocol version | +| `session_id` | string | Optional session id | +| `request_id` | string | Optional request id | +| `acpx_version` | string | Optional ACPX adapter version | +| `events` | array of object | Optional redacted transport events | + +--- + +## `brigade.outcome_record`: `schema_version: 1` + +**Path:** `memory/outcome/records.jsonl` (one object per line) + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema_version` | integer | yes (new writes) | `1` | +| `artifact_id` | string | yes | Skill or card id | +| `artifact_kind` | string | yes | `skill` or `card` | +| `task_id` | string | yes | May be empty | +| `source` | string | yes | `verify`, `run`, `friction`, … | +| `signal_value` | integer | yes | `-1`, `0`, or `+1` | +| `evidence_ref` | string | yes | Path to receipt | +| `ts` | string | yes | ISO-8601 | +| `prev_digest` | string \| null | yes | Chain link | +| `digest` | string | yes | Row digest | +| `code_graph_delta` | object | no | Compact delta | +| `context_eval` | object | no | | +| `content_fingerprint` | string | no | Artifact bytes hash | +| `context` | object | no | Harness manifest | +| `capability_fingerprint` | string | no | | +| `route` | object | no | Route manifest | +| `route_fingerprint` | string | no | | + +--- + +## `brigade.outcome_decision`: `schema_version: 1` + +**Path:** `memory/outcome/decisions/-.json` + +| Field | Type | Required | Notes | +| --- | --- | --- | --- | +| `schema_version` | integer | yes (new writes) | `1` | +| `artifact_id` | string | yes | | +| `action` | string | yes | `install`, `rollback`, `hold`, … | +| `prior_status`, `new_status`, `decided_status` | string | yes | | +| `reason` | string | yes | | +| `score` | object | yes | Scoring breakdown | +| `execution` | string | yes | Physical side-effect result | +| `created_at` | string | yes | ISO-8601 | +| `content_fingerprint` | string | no | Current content fingerprint when stale evidence was excluded | +| `lifetime_score` | number | no | Lifetime score before fingerprint filtering | +| `lifetime_helped` | integer | no | Lifetime positive-signal count | +| `lifetime_hurt` | integer | no | Lifetime negative-signal count | +| `stale_records` | integer | no | Records excluded as stale | +| `legacy_records` | integer | no | Records without a content fingerprint | + +--- + +## Related commands + +- `brigade receipts verify`: digest chain checks for verify receipts and outcome rows +- `brigade receipts export miseledger`: adapter export (separate `miseledger.adapter.v1` envelope) +- `brigade outcome rebuild-status`: prove `status.json` matches decision receipts diff --git a/src/brigade/aboyeur.py b/src/brigade/aboyeur.py index 1fc00c17..4afcd809 100644 --- a/src/brigade/aboyeur.py +++ b/src/brigade/aboyeur.py @@ -25,7 +25,7 @@ from . import evidence_brief as evidence_brief_mod from . import graphtrail_delta from . import localio -from . import proc, runguard +from . import proc, receipt_schema, runguard from . import run_control from .result_integrity import validate_final_output from .run_receipts import ( @@ -475,7 +475,7 @@ def _write_json(path: Path, payload: object) -> None: # run.json is polled by `brigade runs watch/steer/interrupt` while the run # rewrites it, so the write must be atomic or a concurrent reader can # observe a truncated file. - localio.write_text_atomic(path, json.dumps(payload, indent=2) + "\n") + localio.write_text_atomic(path, json.dumps(payload, indent=2, sort_keys=True) + "\n") def _utc_iso(value: datetime) -> str: @@ -1707,6 +1707,10 @@ def set_artifact_patch_ref(output_dir: Path, patch_ref: str = "changes.patch") - if not isinstance(payload, dict) or "ground_truth" not in payload: raise runguard.RunGuardError(f"{filename} is missing ground_truth while recording artifact patch reference") payload["ground_truth"] = _with_patch_ref(payload.get("ground_truth"), patch_ref) + if filename == "worker-results.json": + receipt_schema.stamp_worker_results_document(payload) + else: + receipt_schema.stamp_synthesis_document(payload) try: _write_json(path, payload) except OSError as exc: @@ -1799,7 +1803,7 @@ def record_artifact_collection( ) payload["artifact_collection"] = collection try: - _write_json(run_path, payload) + _write_json(run_path, receipt_schema.stamp_run_receipt(payload)) except OSError as exc: raise runguard.RetainRunLockError(f"failed to update run receipt after artifact collection: {exc}") from exc @@ -1872,7 +1876,7 @@ def record_run_termination( round((finished_at - started).total_seconds(), 3), ) try: - _write_json(run_path, payload) + _write_json(run_path, receipt_schema.stamp_run_receipt(payload)) except OSError as exc: raise runguard.RetainRunLockError(f"failed to write terminal run receipt: {exc}") from exc @@ -1900,7 +1904,7 @@ def record_dispatch_stage(output_dir: Path, *, stage: int, seats: tuple[str, ... } ) try: - _write_json(run_path, payload) + _write_json(run_path, receipt_schema.stamp_run_receipt(payload)) except OSError as exc: raise runguard.RetainRunLockError(f"failed to write dispatch stage receipt: {exc}") from exc @@ -1929,7 +1933,7 @@ def record_result_processing(output_dir: Path, *, seat: str) -> None: payload.pop("active_stage", None) payload.pop("active_seats", None) try: - _write_json(run_path, payload) + _write_json(run_path, receipt_schema.stamp_run_receipt(payload)) except OSError as exc: raise runguard.RetainRunLockError(f"failed to record result-processing phase: {exc}") from exc @@ -1946,7 +1950,8 @@ def _roster_resolution_payload(roster: Roster) -> dict[str, object] | None: def _roster_payload(roster: Roster) -> dict[str, object]: payload: dict[str, object] = { - "schema": "brigade.roster_snapshot.v1", + "schema": receipt_schema.ROSTER_SNAPSHOT_SCHEMA, + "schema_version": receipt_schema.ROSTER_SNAPSHOT_SCHEMA_VERSION, "orchestrator": roster.orchestrator, "max_workers": roster.max_workers, "timeout_seconds": roster.timeout_seconds, @@ -2011,9 +2016,9 @@ def _run_payload( scheduler: dict[str, object] | None = None, ) -> dict[str, object]: payload: dict[str, object] = { - "schema": "brigade.run.v1", + "schema": receipt_schema.RUN_RECEIPT_SCHEMA, + "schema_version": receipt_schema.RUN_RECEIPT_SCHEMA_VERSION, "task": task, - "cwd": str(cwd) if cwd is not None else None, "orchestrator": roster.orchestrator, "dry_run": dry_run, "read_only": read_only, @@ -2039,6 +2044,8 @@ def _run_payload( "attached": list(brief_set.attached) if brief_set is not None else [], }, } + if cwd is not None: + payload["cwd"] = str(cwd) if scheduler is not None: payload["scheduler"] = scheduler if resolution := _roster_resolution_payload(roster): @@ -2575,7 +2582,7 @@ def _drift_failure_rc() -> int | None: _write_json(output_dir / "plan-attempts.json", attempts_payload) _write_json( output_dir / "plan.json", - {"schema": "brigade.run_plan.v1", "assignments": _assignment_payload(assignments)}, + receipt_schema.run_plan_document(_assignment_payload(assignments)), ) if dry_run: @@ -2851,11 +2858,10 @@ def dispatch_interrupted() -> None: worker_results = _write_worker_logs(output_dir, worker_results) _write_json( output_dir / "worker-results.json", - { - "schema": "brigade.worker_results.v1", - "results": _worker_payload(worker_results), - "ground_truth": ground_truth, - }, + receipt_schema.worker_results_document( + _worker_payload(worker_results), + ground_truth=ground_truth, + ), ) if verbose: _print_worker_status(worker_results) @@ -2933,21 +2939,18 @@ def dispatch_interrupted() -> None: if not direct_worker: final = _write_agent_logs(output_dir, "synthesis", final) synthesis_payload = ( - { - "schema": "brigade.synthesis.v1", - "mode": "direct-worker", - "worker": worker, - "orchestrator": None, - "result": _agent_result_payload(final), - "ground_truth": ground_truth, - } + receipt_schema.synthesis_document( + mode="direct-worker", + worker=worker, + result=_agent_result_payload(final), + ground_truth=ground_truth, + ) if direct_worker - else { - "schema": "brigade.synthesis.v1", - "orchestrator": roster.orchestrator, - "result": _agent_result_payload(final), - "ground_truth": ground_truth, - } + else receipt_schema.synthesis_document( + orchestrator=roster.orchestrator, + result=_agent_result_payload(final), + ground_truth=ground_truth, + ) ) _write_json(output_dir / "synthesis.json", synthesis_payload) if not final.ok: diff --git a/src/brigade/outcome_cmd.py b/src/brigade/outcome_cmd.py index 3974030e..db615a07 100644 --- a/src/brigade/outcome_cmd.py +++ b/src/brigade/outcome_cmd.py @@ -19,7 +19,7 @@ from pathlib import Path from typing import Any -from . import localio, outcome as core +from . import localio, outcome as core, receipt_schema def _records_path(target: Path) -> Path: @@ -453,6 +453,7 @@ def _last_record_digest(path: Path) -> str | None: def _record_payload(record: core.OutcomeRecord) -> dict: row = dataclasses.asdict(record) + row["schema_version"] = receipt_schema.OUTCOME_RECORD_SCHEMA_VERSION if row.get("code_graph_delta") is None: row.pop("code_graph_delta", None) if row.get("context_eval") is None: @@ -1286,6 +1287,7 @@ def reconcile( new_status = prior_status if install_failed else decision.new_status effective_status[decision.artifact_id] = new_status receipt = { + "schema_version": receipt_schema.OUTCOME_DECISION_SCHEMA_VERSION, "artifact_id": decision.artifact_id, "action": decision.action, "prior_status": prior_status, diff --git a/src/brigade/receipt_schema.py b/src/brigade/receipt_schema.py new file mode 100644 index 00000000..73664332 --- /dev/null +++ b/src/brigade/receipt_schema.py @@ -0,0 +1,87 @@ +"""Schema version constants and SIEM-oriented receipt serialization helpers.""" + +from __future__ import annotations + +VERIFY_RECEIPT_SCHEMA_VERSION = 2 +WORK_CLOSEOUT_SCHEMA_VERSION = 1 +RUN_RECEIPT_SCHEMA_VERSION = 1 +OUTCOME_RECORD_SCHEMA_VERSION = 1 +OUTCOME_DECISION_SCHEMA_VERSION = 1 + +RUN_RECEIPT_SCHEMA = "brigade.run.v1" + +ROSTER_SNAPSHOT_SCHEMA = "brigade.roster_snapshot.v1" +ROSTER_SNAPSHOT_SCHEMA_VERSION = 1 +RUN_PLAN_SCHEMA = "brigade.run_plan.v1" +RUN_PLAN_SCHEMA_VERSION = 1 +WORKER_RESULTS_SCHEMA = "brigade.worker_results.v1" +WORKER_RESULTS_SCHEMA_VERSION = 1 +SYNTHESIS_SCHEMA = "brigade.synthesis.v1" +SYNTHESIS_SCHEMA_VERSION = 1 + + +def stamp_run_receipt(payload: dict[str, object]) -> dict[str, object]: + payload.setdefault("schema", RUN_RECEIPT_SCHEMA) + payload.setdefault("schema_version", RUN_RECEIPT_SCHEMA_VERSION) + return payload + + +def stamp_worker_results_document(payload: dict[str, object]) -> dict[str, object]: + payload.setdefault("schema", WORKER_RESULTS_SCHEMA) + payload.setdefault("schema_version", WORKER_RESULTS_SCHEMA_VERSION) + return payload + + +def stamp_synthesis_document(payload: dict[str, object]) -> dict[str, object]: + payload.setdefault("schema", SYNTHESIS_SCHEMA) + payload.setdefault("schema_version", SYNTHESIS_SCHEMA_VERSION) + return payload + + +def run_plan_document(assignments: list[dict[str, object]]) -> dict[str, object]: + return { + "schema": RUN_PLAN_SCHEMA, + "schema_version": RUN_PLAN_SCHEMA_VERSION, + "assignments": assignments, + } + + +def worker_results_document( + results: list[dict[str, object]], + *, + ground_truth: dict[str, object] | None = None, +) -> dict[str, object]: + doc: dict[str, object] = { + "schema": WORKER_RESULTS_SCHEMA, + "schema_version": WORKER_RESULTS_SCHEMA_VERSION, + "results": results, + } + if ground_truth is not None: + doc["ground_truth"] = ground_truth + return doc + + +def synthesis_document( + *, + result: dict[str, object], + orchestrator: str | None = None, + worker: str | None = None, + mode: str | None = None, + ground_truth: dict[str, object] | None = None, +) -> dict[str, object]: + doc: dict[str, object] = { + "schema": SYNTHESIS_SCHEMA, + "schema_version": SYNTHESIS_SCHEMA_VERSION, + "result": result, + } + if mode is not None: + doc["mode"] = mode + if worker is not None: + doc["worker"] = worker + if orchestrator is not None: + doc["orchestrator"] = orchestrator + elif mode == "direct-worker": + doc["orchestrator"] = None + if ground_truth is not None: + doc["ground_truth"] = ground_truth + return doc diff --git a/src/brigade/run_resume.py b/src/brigade/run_resume.py index 8a0cf84a..eccbf56c 100644 --- a/src/brigade/run_resume.py +++ b/src/brigade/run_resume.py @@ -12,7 +12,7 @@ from datetime import datetime, timezone from pathlib import Path -from . import aboyeur, agents, codex_appserver, runguard +from . import aboyeur, agents, codex_appserver, receipt_schema, runguard from .roster import Agent, Roster, _as_bool, _as_env _RESUMABLE_STATUSES = ("interrupted", "failed") @@ -201,11 +201,10 @@ def _resume_locked(run_dir: Path) -> int: ground_truth = worker_data.get("ground_truth") or {} aboyeur._write_json( run_dir / "worker-results.json", - { - "schema": "brigade.worker_results.v1", - "results": aboyeur._worker_payload(worker_results), - "ground_truth": ground_truth, - }, + receipt_schema.worker_results_document( + aboyeur._worker_payload(worker_results), + ground_truth=ground_truth, + ), ) task = run_meta.get("task", "") @@ -229,19 +228,18 @@ def _resume_locked(run_dir: Path) -> int: ) aboyeur._write_json( run_dir / "synthesis.json", - { - "schema": "brigade.synthesis.v1", - "orchestrator": roster.orchestrator, - "result": {"ok": final.ok, "detail": final.detail, "text": final.text}, - "ground_truth": ground_truth, - }, + receipt_schema.synthesis_document( + orchestrator=roster.orchestrator, + result={"ok": final.ok, "detail": final.detail, "text": final.text}, + ground_truth=ground_truth, + ), ) now = datetime.now(timezone.utc).isoformat() run_meta.setdefault("resumed_at", []).append(now) if not final.ok: run_meta["status"] = "failed" run_meta["error"] = final.detail - aboyeur._write_json(run_dir / "run.json", run_meta) + aboyeur._write_json(run_dir / "run.json", receipt_schema.stamp_run_receipt(run_meta)) print(f"error: orchestrator failed during synthesis: {final.detail}", file=sys.stderr) return 2 (run_dir / "final.txt").write_text(final.text + "\n") @@ -255,6 +253,6 @@ def _resume_locked(run_dir: Path) -> int: history = [] run_meta["recovery_history"] = history history.append(recovered_failure) - aboyeur._write_json(run_dir / "run.json", run_meta) + aboyeur._write_json(run_dir / "run.json", receipt_schema.stamp_run_receipt(run_meta)) print(final.text) return 0 diff --git a/src/brigade/runguard.py b/src/brigade/runguard.py index 47bca172..ecf67ab4 100644 --- a/src/brigade/runguard.py +++ b/src/brigade/runguard.py @@ -16,7 +16,7 @@ from time import sleep as _sleep from uuid import uuid4 -from . import localio, proc +from . import localio, proc, receipt_schema _NONTERMINAL_RUN_STATUSES = frozenset( { @@ -366,7 +366,7 @@ def _recover_run_artifact(owner: dict[str, object] | None) -> str: } ) try: - localio.write_json(run_json, payload) + localio.write_json(run_json, receipt_schema.stamp_run_receipt(payload)) except OSError: return "write-failed" return "recovered" diff --git a/src/brigade/work_cmd/verification.py b/src/brigade/work_cmd/verification.py index 54b2989d..c36bf119 100644 --- a/src/brigade/work_cmd/verification.py +++ b/src/brigade/work_cmd/verification.py @@ -14,7 +14,7 @@ from pathlib import Path from typing import Any from uuid import uuid4 -from .. import config, graphtrail_delta, localio, proc, receipt_signing, runguard +from .. import config, graphtrail_delta, localio, proc, receipt_schema, receipt_signing, runguard from . import constants, helpers, ledger as ledger_mod from . import reviews as reviews_mod from . import scanners as scanners_mod @@ -133,7 +133,6 @@ def _verify_execution_argv(argv: list[str], target: Path) -> list[str]: _VERIFY_CANCELED_RC = 130 _VERIFY_INTERRUPTED_COMMAND_STATUS = "interrupted" _VERIFY_CANCELED_RECEIPT_STATUS = "canceled" -_VERIFY_RECEIPT_SCHEMA_VERSION = 2 def _verify_child_popen_kwargs() -> dict[str, Any]: @@ -281,15 +280,15 @@ def _finalize_verify_receipt( receipt["digests"]["key_id"] = key_id except Exception: receipt.pop("digests", None) + helpers._write_json(run_dir / "receipt.json", receipt) try: - helpers._write_json(run_dir / "receipt.json", receipt) _write_verify_markdown(run_dir, receipt) + except Exception: + pass + try: _prune_verify_runs(target) except Exception: - try: - helpers._write_json(run_dir / "receipt.json", receipt) - except OSError: - pass + pass return receipt, rc @@ -687,7 +686,7 @@ def _tree_fingerprint(target: Path) -> str | None: def _capture_verify_identity(target: Path, run_dir: Path) -> dict[str, Any]: unavailable: dict[str, Any] = { - "schema_version": _VERIFY_RECEIPT_SCHEMA_VERSION, + "schema_version": receipt_schema.VERIFY_RECEIPT_SCHEMA_VERSION, "baseline_commit": None, "tree_fingerprint": None, "changes_patch_sha256": None, @@ -710,7 +709,7 @@ def _capture_verify_identity(target: Path, run_dir: Path) -> dict[str, Any]: patch_path.unlink(missing_ok=True) return unavailable return { - "schema_version": _VERIFY_RECEIPT_SCHEMA_VERSION, + "schema_version": receipt_schema.VERIFY_RECEIPT_SCHEMA_VERSION, "baseline_commit": baseline_commit, "tree_fingerprint": tree_fingerprint, "changes_patch_sha256": hashlib.sha256(patch_bytes).hexdigest(), @@ -897,11 +896,13 @@ def _write_reused_receipt( "timeout": timeout, "path": str(run_dir), "commands": copy.deepcopy(latest.get("commands", [])), - "reused_from": latest.get("run_id"), "planned_commands": planned_display, } receipt.update(identity) _stamp_harness_session(receipt) + reused_from = latest.get("run_id") + if isinstance(reused_from, str) and reused_from: + receipt["reused_from"] = reused_from git = _receipt_git_snapshot(target) if git is not None: receipt["git"] = git @@ -1026,6 +1027,7 @@ def _work_closeout_payload(target: Path, session_id: str, *, write: bool = False now = helpers._now() closeout_id = f"{now.strftime('%Y%m%d-%H%M%S')}-work-closeout-{uuid4().hex[:6]}" closeout = { + "schema_version": receipt_schema.WORK_CLOSEOUT_SCHEMA_VERSION, "closeout_id": closeout_id, "target": str(target), "status": "ready" if not blockers else "blocked", diff --git a/tests/test_receipt_schema_audit.py b/tests/test_receipt_schema_audit.py new file mode 100644 index 00000000..cfa77a19 --- /dev/null +++ b/tests/test_receipt_schema_audit.py @@ -0,0 +1,457 @@ +"""SIEM-readiness schema audit fixes for sanctioned receipt families (#506).""" + +from __future__ import annotations + +import json +import re +import sys +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from brigade import aboyeur, outcome, outcome_cmd, receipt_schema, runguard +from brigade.work_cmd import helpers, verification as verify_mod + + +def _init_git_repo(path: Path) -> None: + import subprocess + + subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=path, check=True, capture_output=True) + subprocess.run(["git", "config", "user.name", "Test"], cwd=path, check=True, capture_output=True) + (path / "README.md").write_text("ok\n") + subprocess.run(["git", "add", "README.md"], cwd=path, check=True, capture_output=True) + subprocess.run(["git", "commit", "-m", "init"], cwd=path, check=True, capture_output=True) + + +def test_verify_receipt_emits_schema_version(tmp_path, capsys): + _init_git_repo(tmp_path) + rc = verify_mod.verify_run( + target=tmp_path, + commands=[f"{sys.executable} -c \"print('ok')\""], + reuse=False, + json_output=True, + ) + assert rc == 0 + receipt = json.loads(capsys.readouterr().out) + assert receipt["schema_version"] == receipt_schema.VERIFY_RECEIPT_SCHEMA_VERSION + stored = json.loads((Path(receipt["path"]) / "receipt.json").read_text()) + assert stored["schema_version"] == receipt_schema.VERIFY_RECEIPT_SCHEMA_VERSION + + +def test_verify_receipt_legacy_without_schema_version_still_loads(tmp_path): + run_dir = tmp_path / ".brigade" / "work" / "verify-runs" / "legacy-run" + run_dir.mkdir(parents=True) + legacy = { + "run_id": "legacy-run", + "target": str(tmp_path), + "status": "completed", + "started_at": "2026-01-01T00:00:00+00:00", + "commands": [], + } + (run_dir / "receipt.json").write_text(json.dumps(legacy) + "\n") + loaded = verify_mod._verify_read_receipt(run_dir) + assert loaded is not None + assert loaded["run_id"] == "legacy-run" + assert "schema_version" not in loaded + + +def test_finalize_verify_receipt_writes_once_when_summary_fails(tmp_path, monkeypatch): + run_dir = tmp_path / "verify-run" + run_dir.mkdir() + started = datetime(2026, 1, 1, tzinfo=timezone.utc) + receipt = { + "schema_version": receipt_schema.VERIFY_RECEIPT_SCHEMA_VERSION, + "run_id": "verify-run", + "target": str(tmp_path), + "status": "running", + "started_at": started.isoformat(), + "path": str(run_dir), + "commands": [ + { + "command": "true", + "status": "completed", + "exit_code": 0, + } + ], + } + writes: list[Path] = [] + original_write = helpers._write_json + + def counting_write(path: Path, payload: object) -> None: + if path.name == "receipt.json": + writes.append(path) + original_write(path, payload) + + monkeypatch.setattr(helpers, "_write_json", counting_write) + monkeypatch.setattr( + verify_mod, + "_write_verify_markdown", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("summary failed")), + ) + monkeypatch.setattr(verify_mod, "_prune_verify_runs", lambda *_args, **_kwargs: None) + + finalized, rc = verify_mod._finalize_verify_receipt( + tmp_path, + run_dir, + receipt, + started=started, + rc=0, + canceled=False, + ) + + assert rc == 0 + assert finalized["status"] == "completed" + assert len(writes) == 1 + assert json.loads((run_dir / "receipt.json").read_text())["status"] == "completed" + + +def test_run_receipt_emits_schema_version_and_sorted_keys(tmp_path): + payload = aboyeur._run_payload( + task="audit", + cwd=tmp_path, + lock_workspace=tmp_path, + roster=_minimal_roster(), + dry_run=True, + read_only=True, + status="started", + started_at=aboyeur.datetime.now(aboyeur.timezone.utc), + include_git=False, + ) + assert payload["schema_version"] == receipt_schema.RUN_RECEIPT_SCHEMA_VERSION + rendered = json.dumps(payload, indent=2, sort_keys=True) + key_order = re.findall(r'^ "([^"]+)":', rendered, flags=re.MULTILINE) + assert key_order == sorted(key_order) + + +def test_run_receipt_legacy_without_schema_version_still_loads(tmp_path): + run_dir = tmp_path / ".brigade" / "runs" / "legacy-run" + run_dir.mkdir(parents=True) + legacy = { + "schema": receipt_schema.RUN_RECEIPT_SCHEMA, + "task": "legacy", + "cwd": str(tmp_path), + "orchestrator": "chef", + "dry_run": False, + "read_only": True, + "status": "ok", + "started_at": "2026-01-01T00:00:00Z", + "status_started_at": "2026-01-01T00:00:00Z", + "suspected_noop": False, + "code_graph_brief": {"attached": False}, + "drift_impact_brief": {"attached": False}, + "evidence_brief": {"attached": False}, + "brief_budget": {"attached": []}, + } + (run_dir / "run.json").write_text(json.dumps(legacy, sort_keys=True) + "\n") + payload, run_json = outcome_cmd._read_run_receipt(run_dir) + assert payload is not None + assert run_json.name == "run.json" + assert payload["schema"] == receipt_schema.RUN_RECEIPT_SCHEMA + assert "schema_version" not in payload + + +def test_run_receipt_omits_null_cwd(tmp_path): + payload = aboyeur._run_payload( + task="audit", + cwd=None, + lock_workspace=tmp_path, + roster=_minimal_roster(), + dry_run=True, + read_only=True, + status="started", + started_at=aboyeur.datetime.now(aboyeur.timezone.utc), + include_git=False, + ) + assert "cwd" not in payload + + +def test_run_json_writer_uses_sorted_keys(tmp_path): + path = tmp_path / "run.json" + aboyeur._write_json(path, {"z": 1, "a": 2, "m": {"y": 1, "b": 2}}) + text = path.read_text() + assert text.index('"a"') < text.index('"m"') < text.index('"z"') + nested = re.search(r'"m": \{\n(.*?)\n \}', text, flags=re.DOTALL) + assert nested is not None + assert nested.group(1).index('"b"') < nested.group(1).index('"y"') + + +@pytest.mark.parametrize( + ("document", "schema", "schema_version"), + [ + ( + lambda: aboyeur._roster_payload(_minimal_roster()), + receipt_schema.ROSTER_SNAPSHOT_SCHEMA, + receipt_schema.ROSTER_SNAPSHOT_SCHEMA_VERSION, + ), + ( + lambda: receipt_schema.run_plan_document([]), + receipt_schema.RUN_PLAN_SCHEMA, + receipt_schema.RUN_PLAN_SCHEMA_VERSION, + ), + ( + lambda: receipt_schema.worker_results_document([], ground_truth={}), + receipt_schema.WORKER_RESULTS_SCHEMA, + receipt_schema.WORKER_RESULTS_SCHEMA_VERSION, + ), + ( + lambda: receipt_schema.synthesis_document( + orchestrator="chef", + result={"ok": True, "detail": "", "text": "done"}, + ground_truth={}, + ), + receipt_schema.SYNTHESIS_SCHEMA, + receipt_schema.SYNTHESIS_SCHEMA_VERSION, + ), + ], +) +def test_run_sidecar_emits_schema_version(document, schema, schema_version): + payload = document() + assert payload["schema"] == schema + assert payload["schema_version"] == schema_version + + +def test_outcome_record_emits_schema_version(tmp_path): + outcome_cmd.record( + target=tmp_path, + artifact_id="brigade-work", + source="friction", + status="cleared", + json_output=True, + ) + row = json.loads((tmp_path / "memory" / "outcome" / "records.jsonl").read_text().strip()) + assert row["schema_version"] == receipt_schema.OUTCOME_RECORD_SCHEMA_VERSION + + +def test_outcome_record_legacy_without_schema_version_still_loads(tmp_path): + path = tmp_path / "memory" / "outcome" / "records.jsonl" + path.parent.mkdir(parents=True) + legacy = { + "artifact_id": "brigade-work", + "artifact_kind": "skill", + "task_id": "", + "source": "friction", + "signal_value": 1, + "evidence_ref": "", + "ts": "2026-01-01T00:00:00+00:00", + "prev_digest": None, + "digest": "abc", + } + path.write_text(json.dumps(legacy, sort_keys=True) + "\n") + records = outcome_cmd.load_records(tmp_path) + assert len(records) == 1 + assert records[0].artifact_id == "brigade-work" + + +def test_recover_run_artifact_stamps_schema_on_missing_run_json(tmp_path): + run_dir = tmp_path / "run" + run_dir.mkdir() + result = runguard._recover_run_artifact({"run_dir": str(run_dir), "pid": 4321}) + assert result == "recovered" + payload = json.loads((run_dir / "run.json").read_text()) + assert payload["schema"] == receipt_schema.RUN_RECEIPT_SCHEMA + assert payload["schema_version"] == receipt_schema.RUN_RECEIPT_SCHEMA_VERSION + assert payload["status"] == "failed" + assert payload["failure_phase"] == "stale-lock-recovery" + assert payload["failure"]["owner_pid"] == 4321 + assert payload["failure"]["prior_status"] == "artifact-unavailable" + + +def test_set_artifact_patch_ref_stamps_schema_on_legacy_sidecars(tmp_path): + output_dir = tmp_path / "run" + output_dir.mkdir() + (output_dir / "worker-results.json").write_text( + json.dumps( + { + "schema": receipt_schema.WORKER_RESULTS_SCHEMA, + "results": [], + "ground_truth": {}, + } + ) + + "\n" + ) + (output_dir / "synthesis.json").write_text( + json.dumps( + { + "schema": receipt_schema.SYNTHESIS_SCHEMA, + "orchestrator": "chef", + "result": {"ok": True, "detail": "", "text": "done"}, + "ground_truth": {}, + } + ) + + "\n" + ) + aboyeur.set_artifact_patch_ref(output_dir, "changes.patch") + worker = json.loads((output_dir / "worker-results.json").read_text()) + synthesis = json.loads((output_dir / "synthesis.json").read_text()) + assert worker["schema_version"] == receipt_schema.WORKER_RESULTS_SCHEMA_VERSION + assert synthesis["schema_version"] == receipt_schema.SYNTHESIS_SCHEMA_VERSION + assert worker["ground_truth"]["patch_ref"] == "changes.patch" + + +def test_record_dispatch_stage_stamps_schema_on_legacy_run_json(tmp_path): + output_dir = tmp_path / "run" + output_dir.mkdir() + (output_dir / "run.json").write_text( + json.dumps( + { + "schema": receipt_schema.RUN_RECEIPT_SCHEMA, + "status": "planning", + "started_at": "2026-01-01T00:00:00Z", + } + ) + + "\n" + ) + aboyeur.record_dispatch_stage(output_dir, stage=1, seats=("coder",)) + payload = json.loads((output_dir / "run.json").read_text()) + assert payload["schema_version"] == receipt_schema.RUN_RECEIPT_SCHEMA_VERSION + assert payload["status"] == "dispatching" + + +def test_verify_receipt_emits_null_identity_tuple_outside_git(tmp_path, capsys, monkeypatch): + monkeypatch.setenv("GRAPHTRAIL_BIN", str(tmp_path / "missing-graphtrail")) + rc = verify_mod.verify_run( + target=tmp_path, + commands=[f"{sys.executable} -c \"print('ok')\""], + reuse=False, + json_output=True, + ) + assert rc == 0 + receipt = json.loads(capsys.readouterr().out) + assert receipt["baseline_commit"] is None + assert receipt["tree_fingerprint"] is None + assert receipt["changes_patch_sha256"] is None + stored = json.loads((Path(receipt["path"]) / "receipt.json").read_text()) + assert stored["baseline_commit"] is None + assert stored["tree_fingerprint"] is None + assert stored["changes_patch_sha256"] is None + + +def test_write_reused_receipt_omits_reused_from_without_source_run_id(tmp_path): + receipt, rc = verify_mod._write_reused_receipt( + tmp_path, + {"commands": [], "status": "completed"}, + ["true"], + 60, + ) + assert rc == 0 + assert "reused_from" not in receipt + assert receipt["baseline_commit"] is None + assert receipt["tree_fingerprint"] is None + assert receipt["changes_patch_sha256"] is None + + +def test_work_closeout_emits_schema_version(tmp_path, monkeypatch): + session_path = tmp_path / ".brigade" / "work" / "20260101-session" + session_path.mkdir(parents=True) + session_payload = { + "id": "20260101-session", + "status": "ended", + "title": "Audit", + "started_at": "2026-01-01T00:00:00+00:00", + "ended_at": "2026-01-01T01:00:00+00:00", + } + (session_path / "session.json").write_text(json.dumps(session_payload) + "\n") + monkeypatch.setattr( + verify_mod, + "_verification_evidence_payload", + lambda _target, _session: { + "latest_verify": { + "run_id": "verify-run", + "status": "completed", + "path": str(tmp_path / "verify"), + "commands": [{"command": "true"}], + }, + "task": {"id": "task-one", "text": "Audit"}, + "task_acceptance": ["Done"], + "scanner_sweep": {}, + "code_review": {}, + "handoff_drafts": {}, + }, + ) + closeout, rc = verify_mod._work_closeout_payload(tmp_path, "20260101-session", write=True) + assert rc == 0 + assert closeout["schema_version"] == receipt_schema.WORK_CLOSEOUT_SCHEMA_VERSION + stored = json.loads(Path(closeout["path"]).read_text()) + assert stored["schema_version"] == receipt_schema.WORK_CLOSEOUT_SCHEMA_VERSION + + +def test_outcome_decision_emits_schema_version(tmp_path, monkeypatch): + def _fake_execute(_target, _artifact_id, _action): + return "installed" + + monkeypatch.setattr(outcome_cmd, "_execute_skill_decision", _fake_execute) + records_path = tmp_path / "memory" / "outcome" / "records.jsonl" + records_path.parent.mkdir(parents=True) + row = outcome_cmd._record_payload( + outcome.OutcomeRecord( + "skill-x", + "skill", + "", + "verify", + 1, + "", + "2026-01-01T00:00:00+00:00", + ) + ) + records_path.write_text(json.dumps(row, sort_keys=True) + "\n") + records_path.write_text( + records_path.read_text() + + json.dumps( + outcome_cmd._record_payload( + outcome.OutcomeRecord( + "skill-x", + "skill", + "", + "verify", + 1, + "", + "2026-01-01T01:00:00+00:00", + ) + ), + sort_keys=True, + ) + + "\n" + ) + assert outcome_cmd.reconcile(target=tmp_path, apply=True, json_output=True) == 0 + decision_path = next((tmp_path / "memory" / "outcome" / "decisions").glob("*.json")) + decision = json.loads(decision_path.read_text()) + assert decision["schema_version"] == receipt_schema.OUTCOME_DECISION_SCHEMA_VERSION + + +def test_sorted_receipt_writer_is_byte_deterministic(tmp_path): + payload = { + "schema": receipt_schema.RUN_RECEIPT_SCHEMA, + "schema_version": receipt_schema.RUN_RECEIPT_SCHEMA_VERSION, + "task": "audit", + "orchestrator": "chef", + "dry_run": False, + "read_only": True, + "status": "ok", + "started_at": "2026-01-01T00:00:00Z", + "nested": {"z": 1, "a": 2}, + } + first = tmp_path / "first.json" + second = tmp_path / "second.json" + aboyeur._write_json(first, payload) + aboyeur._write_json(second, payload) + assert first.read_bytes() == second.read_bytes() + + +def _minimal_roster(): + from brigade.roster import Agent, Roster + + return Roster( + orchestrator="chef", + max_workers=1, + agents={ + "chef": Agent( + name="chef", + cli="echo", + role="orchestrator", + model="test", + transport="local", + ) + }, + )