From 202212e8d513897fbf29ec42e2ad973dfd98c4c4 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 22 Jun 2026 23:54:24 +0200 Subject: [PATCH 1/3] feat(capture): two-phase classification-aware AuditSink port (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Splits the audit port into Reserve (before any write) + Record (after), so "confidential + sink-down → refuse before any write" is literally true even though the audit record — which lists what landed — can only be written afterwards. - AuditSink.Reserve(ctx, level) → AuditOutcome | error. The error path refuses the capture before writing: confidential + central sink down, or the all-tiers floor (nothing can record). - AuditSink.Record(ctx, entry, outcome) persists per the reserved outcome. - Service: I5 gate runs after the I1 gate and after the dry-run short-circuit (dry-run never probes the sink). AuditBuffered surfaces on the receipt. New ErrAuditUnavailable sentinel (→ HTTP 503). The tier→behaviour decision lives in the sink impl (#54's DegradingSink), not the service — the service just honours Reserve's verdict. Co-Authored-By: Claude Opus 4.8 (1M context) --- ingestion/internal/capture/entities.go | 5 ++ ingestion/internal/capture/ports.go | 30 ++++++++-- ingestion/internal/capture/service.go | 29 ++++++++-- ingestion/internal/capture/service_test.go | 67 +++++++++++++++++++++- 4 files changed, 119 insertions(+), 12 deletions(-) diff --git a/ingestion/internal/capture/entities.go b/ingestion/internal/capture/entities.go index c9aa227..1c33583 100644 --- a/ingestion/internal/capture/entities.go +++ b/ingestion/internal/capture/entities.go @@ -140,4 +140,9 @@ type CaptureReceipt struct { Errors []ItemError `json:"errors"` EffectiveClassification string `json:"effective_classification,omitempty"` DryRun bool `json:"dry_run"` + // AuditBuffered is true when the central audit sink was unreachable and + // this capture's audit record was written to the durable local buffer + // instead (internal/public tier). Surfaces the degraded state to the + // caller per §4.4. + AuditBuffered bool `json:"audit_buffered,omitempty"` } diff --git a/ingestion/internal/capture/ports.go b/ingestion/internal/capture/ports.go index b2c312e..843f17e 100644 --- a/ingestion/internal/capture/ports.go +++ b/ingestion/internal/capture/ports.go @@ -96,9 +96,31 @@ type AuditEntry struct { SecurityEvents []string } -// AuditSink records the audit entry. The classification-aware -// degradation/refusal policy (confidential fails closed, internal -// degrades) is the caller's concern in #54; this port just records. +// AuditOutcome is how a capture's audit record was (or will be) persisted. +type AuditOutcome int + +const ( + // AuditCentral means the record goes to the central sink (loki). + AuditCentral AuditOutcome = iota + // AuditBuffered means the central sink was unreachable and the record + // is written to a durable local buffer for later reconciliation + // (internal/public tier only). + AuditBuffered +) + +// AuditSink is the two-phase, classification-aware audit port (I5, §4.4). +// +// Reserve runs BEFORE any write and decides whether the capture can be +// audited at its effective classification: it returns the outcome to use, +// or an error to refuse the capture before anything is written +// (confidential + central sink down → refuse; the all-tiers floor when +// nothing can record → refuse). Record runs AFTER the writes and persists +// the final entry per the reserved outcome. +// +// Splitting reserve from record is what lets "confidential + sink-down → +// refuse before any write" be literally true while the record itself +// (which lists what landed) is necessarily written afterwards. type AuditSink interface { - Record(ctx context.Context, e AuditEntry) error + Reserve(ctx context.Context, level classification.Level) (AuditOutcome, error) + Record(ctx context.Context, e AuditEntry, outcome AuditOutcome) error } diff --git a/ingestion/internal/capture/service.go b/ingestion/internal/capture/service.go index 98a83c9..840ac27 100644 --- a/ingestion/internal/capture/service.go +++ b/ingestion/internal/capture/service.go @@ -39,6 +39,12 @@ var validActions = map[string]bool{"create": true, "close": true, "comment": tru // REST adapter maps it to HTTP 403. Callers test with errors.Is. var ErrSovereigntyRefused = fmt.Errorf("capture refused by I1 sovereignty gate") +// ErrAuditUnavailable is returned when the I5 audit gate refuses a capture +// before any write: a confidential capture whose central audit sink is +// unreachable, or the all-tiers floor where nothing can record the audit. +// The REST adapter maps it to HTTP 503. Callers test with errors.Is. +var ErrAuditUnavailable = fmt.Errorf("capture refused: audit substrate unavailable") + // assertedZoneMismatch returns a security-event string when the caller's // harness label asserts a trust zone that contradicts the server-derived // origin. A harness label that names no zone (the normal case, e.g. @@ -105,7 +111,7 @@ func (s *Service) Capture(ctx context.Context, in CaptureInput) (CaptureReceipt, EffectiveClassification: effective.String(), Items: nil, // refused before any write SecurityEvents: append(securityEvents, "I1 refusal: confidential capture via us-nexus origin"), - }) + }, AuditCentral) return CaptureReceipt{}, fmt.Errorf("%w: effective classification confidential through %s origin", ErrSovereigntyRefused, in.Context.Origin) } @@ -131,6 +137,16 @@ func (s *Service) Capture(ctx context.Context, in CaptureInput) (CaptureReceipt, return receipt, nil } + // I5 audit gate: decide BEFORE any write whether this capture can be + // audited at its effective classification. Confidential + central sink + // down → refuse here, before writing anything; the all-tiers floor + // (nothing can record) likewise refuses. Internal/public degrade to the + // durable local buffer (signalled by AuditBuffered). + outcome, err := s.audit.Reserve(ctx, effective) + if err != nil { + return CaptureReceipt{}, fmt.Errorf("%w: %v", ErrAuditUnavailable, err) + } + var landed []string for i, ins := range in.Insights { @@ -163,9 +179,9 @@ func (s *Service) Capture(ctx context.Context, in CaptureInput) (CaptureReceipt, } } - // I5: emit a request-level audit record of exactly what landed. - // Best-effort here; the classification-aware refusal/degradation - // policy is #54. + // I5: persist the request-level audit record of exactly what landed, + // using the outcome reserved before the writes. AuditBuffered surfaces + // the degraded (locally-buffered) state on the receipt. if err := s.audit.Record(ctx, AuditEntry{ Timestamp: s.now().UTC(), Principal: in.Context.Principal, @@ -175,9 +191,12 @@ func (s *Service) Capture(ctx context.Context, in CaptureInput) (CaptureReceipt, EffectiveClassification: effective.String(), Items: landed, SecurityEvents: securityEvents, - }); err != nil { + }, outcome); err != nil { receipt.Errors = append(receipt.Errors, ItemError{Item: "audit", Error: err.Error()}) } + if outcome == AuditBuffered { + receipt.AuditBuffered = true + } return receipt, nil } diff --git a/ingestion/internal/capture/service_test.go b/ingestion/internal/capture/service_test.go index 2201513..38b16f7 100644 --- a/ingestion/internal/capture/service_test.go +++ b/ingestion/internal/capture/service_test.go @@ -112,11 +112,20 @@ func (p fakePolicy) Derive(t classification.Target) classification.Level { } type fakeAudit struct { - entries []AuditEntry - err error + entries []AuditEntry + err error // Record error + reserveErr error // Reserve error (refuse before write) + reserveMode AuditOutcome } -func (f *fakeAudit) Record(_ context.Context, e AuditEntry) error { +func (f *fakeAudit) Reserve(_ context.Context, _ classification.Level) (AuditOutcome, error) { + if f.reserveErr != nil { + return 0, f.reserveErr + } + return f.reserveMode, nil +} + +func (f *fakeAudit) Record(_ context.Context, e AuditEntry, _ AuditOutcome) error { if f.err != nil { return f.err } @@ -408,3 +417,55 @@ func TestCaptureInternalViaUSNexusAllowed(t *testing.T) { require.NoError(t, err) assert.True(t, rec.Insights[0].OK) } + +// --- I5 audit gate (#54) --- + +func TestCaptureRefusesWhenAuditReserveFails(t *testing.T) { + // Reserve refusing (e.g. confidential + central sink down, or the floor) + // aborts the capture before any write. + b := &fakeBrain{} + tr := &fakeTracker{} + au := &fakeAudit{reserveErr: errors.New("central sink unreachable")} + svc := newSvc(b, tr, nil, fakePolicy{}, au) + + _, err := svc.Capture(context.Background(), CaptureInput{ + Context: baseCtx(), + Insights: []Insight{{Text: "x", Wing: "hyperguild", Hall: "facts"}}, + }) + require.Error(t, err) + assert.ErrorIs(t, err, ErrAuditUnavailable) + assert.Empty(t, b.writes, "nothing written when audit unavailable") + assert.Empty(t, tr.created) +} + +func TestCaptureFlagsLocallyBufferedAudit(t *testing.T) { + // Reserve returns AuditBuffered (internal/public, central down) → capture + // proceeds and the receipt flags the degraded audit state. + b := &fakeBrain{} + au := &fakeAudit{reserveMode: AuditBuffered} + svc := newSvc(b, &fakeTracker{}, nil, fakePolicy{}, au) + + rec, err := svc.Capture(context.Background(), CaptureInput{ + Context: baseCtx(), + Insights: []Insight{{Text: "x", Wing: "hyperguild", Hall: "facts"}}, + }) + require.NoError(t, err) + assert.True(t, rec.Insights[0].OK, "capture proceeds on degraded audit") + assert.True(t, rec.AuditBuffered, "receipt flags locally-buffered audit") + require.Len(t, au.entries, 1) +} + +func TestCaptureDryRunSkipsAuditGate(t *testing.T) { + // dry_run must not even probe the audit sink (writes nothing anywhere). + au := &fakeAudit{reserveErr: errors.New("would refuse")} + svc := newSvc(&fakeBrain{}, &fakeTracker{}, nil, fakePolicy{}, au) + + rec, err := svc.Capture(context.Background(), CaptureInput{ + Context: baseCtx(), + DryRun: true, + Insights: []Insight{{Text: "x", Wing: "hyperguild", Hall: "facts"}}, + }) + require.NoError(t, err, "dry-run does not hit the audit gate") + assert.True(t, rec.DryRun) + assert.Empty(t, au.entries) +} From 77f5e06d6b2aaa53470bf60d2c408397c1ccd972 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 22 Jun 2026 23:54:24 +0200 Subject: [PATCH 2/3] feat(audit): DegradingSink + durable buffer + loki/ntfy + reconcile (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The classification-aware I5 audit path (§4.4): - DegradingSink.Reserve: central up → AuditCentral; central down + confidential → refuse (no buffer); central down + internal/public + buffer writable → AuditBuffered; central down + buffer unwritable → floor refuse. Record executes the reserved outcome and, when buffered, fires an ntfy alert. - FileBuffer: durable JSONL buffer that survives process restart; Confirm rewrites the file without a record, so a buffered record is cleared ONLY after its central write is confirmed. - LokiCentral: /ready probe + /loki/api/v1/push (full audit entry as the structured line). NtfyNotifier: degraded-state alerts; token only in the auth header, never logged (regression-tested). - Reconcile + StartReconcile: replay buffered records to central on recovery, confirm-then-clear per record; a failed push keeps the record buffered (no loss). SlogSink updated to the two-phase shape (always central, never fails) — the default when no loki endpoint is set. Co-Authored-By: Claude Opus 4.8 (1M context) --- ingestion/internal/audit/buffer.go | 167 ++++++++++++++++++ ingestion/internal/audit/degrading.go | 96 +++++++++++ ingestion/internal/audit/degrading_test.go | 191 +++++++++++++++++++++ ingestion/internal/audit/loki.go | 99 +++++++++++ ingestion/internal/audit/loki_ntfy_test.go | 86 ++++++++++ ingestion/internal/audit/ntfy.go | 55 ++++++ ingestion/internal/audit/reconcile.go | 65 +++++++ ingestion/internal/audit/slog.go | 16 +- ingestion/internal/audit/slog_test.go | 4 +- 9 files changed, 773 insertions(+), 6 deletions(-) create mode 100644 ingestion/internal/audit/buffer.go create mode 100644 ingestion/internal/audit/degrading.go create mode 100644 ingestion/internal/audit/degrading_test.go create mode 100644 ingestion/internal/audit/loki.go create mode 100644 ingestion/internal/audit/loki_ntfy_test.go create mode 100644 ingestion/internal/audit/ntfy.go create mode 100644 ingestion/internal/audit/reconcile.go diff --git a/ingestion/internal/audit/buffer.go b/ingestion/internal/audit/buffer.go new file mode 100644 index 0000000..aa3846d --- /dev/null +++ b/ingestion/internal/audit/buffer.go @@ -0,0 +1,167 @@ +package audit + +import ( + "bufio" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" +) + +// FileBuffer is a durable, restart-surviving audit buffer backed by a +// JSONL file: one {id, entry} record per line. It is the internal/public +// tier fallback when loki is unreachable. Confirm rewrites the file +// without the confirmed record, so a record is cleared only after its +// central write is confirmed. +// +// Access is serialised by a mutex; the buffer is low-throughput (only +// written during a loki outage), so a whole-file rewrite on Confirm is +// acceptable and keeps the on-disk format trivially correct. +type FileBuffer struct { + path string + mu sync.Mutex +} + +type bufferLine struct { + ID string `json:"id"` + Entry capture.AuditEntry `json:"entry"` +} + +// NewFileBuffer returns a buffer backed by path. The parent directory is +// created if needed. The file itself is created lazily on first Append. +func NewFileBuffer(path string) (*FileBuffer, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return nil, fmt.Errorf("create buffer dir: %w", err) + } + return &FileBuffer{path: path}, nil +} + +// Writable reports whether the buffer file can be appended to. It probes +// by opening the file for append (creating it if absent) — the same +// operation Append performs — so Reserve's check matches Append's reality. +func (b *FileBuffer) Writable() error { + b.mu.Lock() + defer b.mu.Unlock() + f, err := os.OpenFile(b.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + return f.Close() +} + +// Append durably writes one audit record. The ID is derived from the +// content + timestamp so it is stable and unique per record. +func (b *FileBuffer) Append(e capture.AuditEntry) error { + b.mu.Lock() + defer b.mu.Unlock() + + line := bufferLine{ID: recordID(e), Entry: e} + data, err := json.Marshal(line) + if err != nil { + return fmt.Errorf("marshal buffer line: %w", err) + } + f, err := os.OpenFile(b.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + defer func() { _ = f.Close() }() + if _, err := f.Write(append(data, '\n')); err != nil { + return err + } + return f.Sync() +} + +// Pending reads all buffered records. A missing file means none. +func (b *FileBuffer) Pending() ([]Buffered, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.readAllLocked() +} + +func (b *FileBuffer) readAllLocked() ([]Buffered, error) { + f, err := os.Open(b.path) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + defer func() { _ = f.Close() }() + + var out []Buffered + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + raw := sc.Bytes() + if len(raw) == 0 { + continue + } + var l bufferLine + if err := json.Unmarshal(raw, &l); err != nil { + return nil, fmt.Errorf("parse buffer line: %w", err) + } + out = append(out, Buffered(l)) + } + return out, sc.Err() +} + +// Confirm removes a single record after its central write is confirmed, by +// rewriting the file without it. Unknown IDs are a no-op. +func (b *FileBuffer) Confirm(id string) error { + b.mu.Lock() + defer b.mu.Unlock() + + all, err := b.readAllLocked() + if err != nil { + return err + } + tmp := b.path + ".tmp" + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return err + } + w := bufio.NewWriter(f) + kept := 0 + for _, rec := range all { + if rec.ID == id { + continue + } + data, _ := json.Marshal(bufferLine(rec)) + if _, err := w.Write(append(data, '\n')); err != nil { + _ = f.Close() + return err + } + kept++ + } + if err := w.Flush(); err != nil { + _ = f.Close() + return err + } + if err := f.Sync(); err != nil { + _ = f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + // Empty buffer → remove the file entirely so Pending sees nothing. + if kept == 0 { + _ = os.Remove(tmp) + return os.Remove(b.path) + } + return os.Rename(tmp, b.path) +} + +// recordID is a stable per-record identifier: sha256 of the principal, +// timestamp, and item list. Distinct captures never collide; the same +// buffered record always hashes the same. +func recordID(e capture.AuditEntry) string { + h := sha256.New() + _, _ = fmt.Fprintf(h, "%s|%s|%v|%s", e.Principal, e.Timestamp.UTC().Format("2006-01-02T15:04:05.000000000Z07:00"), e.Items, e.SessionRef) + return hex.EncodeToString(h.Sum(nil))[:16] +} diff --git a/ingestion/internal/audit/degrading.go b/ingestion/internal/audit/degrading.go new file mode 100644 index 0000000..017d6ab --- /dev/null +++ b/ingestion/internal/audit/degrading.go @@ -0,0 +1,96 @@ +package audit + +import ( + "context" + "fmt" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" +) + +// Central is the central audit substrate (loki). Ready is a cheap +// reachability probe used by the pre-write reserve; Push writes a record. +type Central interface { + Ready(ctx context.Context) error + Push(ctx context.Context, e capture.AuditEntry) error +} + +// Buffer is the durable local fallback for internal/public-tier records +// when the central sink is unreachable. It must survive process restart. +type Buffer interface { + // Writable reports whether the buffer can currently be appended to. + Writable() error + Append(e capture.AuditEntry) error + // Pending returns buffered records awaiting reconciliation, each with a + // stable ID used to Confirm (delete) it after a confirmed central write. + Pending() ([]Buffered, error) + Confirm(id string) error +} + +// Buffered is a buffered audit record plus its stable buffer ID. +type Buffered struct { + ID string + Entry capture.AuditEntry +} + +// Notifier raises an out-of-band alert (ntfy) about a degraded state. +type Notifier interface { + Notify(ctx context.Context, msg string) error +} + +// DegradingSink is the classification-aware AuditSink (§4.4): +// +// - central reachable → AuditCentral (all tiers). +// - central down + confidential → refuse (no buffer): confidential must +// be centrally auditable at write time. +// - central down + internal/public + buffer writable → AuditBuffered. +// - central down + (confidential, or buffer not writable) → refuse (floor). +// +// The decision is made in Reserve, before any write; Record then executes it. +type DegradingSink struct { + central Central + buffer Buffer + notifier Notifier +} + +// NewDegradingSink wires the central sink, durable buffer, and notifier. +func NewDegradingSink(central Central, buffer Buffer, notifier Notifier) *DegradingSink { + return &DegradingSink{central: central, buffer: buffer, notifier: notifier} +} + +// Reserve decides, before any write, how the capture will be audited — or +// returns an error to refuse it. +func (d *DegradingSink) Reserve(ctx context.Context, level classification.Level) (capture.AuditOutcome, error) { + if err := d.central.Ready(ctx); err == nil { + return capture.AuditCentral, nil + } + // Central sink is down. + if level == classification.Confidential { + return 0, fmt.Errorf("confidential capture requires the central audit sink, which is unreachable") + } + if err := d.buffer.Writable(); err != nil { + // Floor: neither central nor local buffer can record the audit. + return 0, fmt.Errorf("audit floor: central sink down and local buffer unwritable: %w", err) + } + return capture.AuditBuffered, nil +} + +// Record persists the entry per the reserved outcome. For AuditBuffered it +// also fires the degraded-state alert. +func (d *DegradingSink) Record(ctx context.Context, e capture.AuditEntry, outcome capture.AuditOutcome) error { + switch outcome { + case capture.AuditBuffered: + if err := d.buffer.Append(e); err != nil { + return fmt.Errorf("buffer audit record: %w", err) + } + // Best-effort alert; the record is already durably buffered. + if d.notifier != nil { + _ = d.notifier.Notify(ctx, fmt.Sprintf( + "capture audit BUFFERED LOCALLY (loki unreachable) — principal=%s class=%s items=%d", + e.Principal, e.EffectiveClassification, len(e.Items))) + } + return nil + default: + return d.central.Push(ctx, e) + } +} diff --git a/ingestion/internal/audit/degrading_test.go b/ingestion/internal/audit/degrading_test.go new file mode 100644 index 0000000..09703e6 --- /dev/null +++ b/ingestion/internal/audit/degrading_test.go @@ -0,0 +1,191 @@ +package audit_test + +import ( + "context" + "errors" + "path/filepath" + "testing" + + "github.com/mathiasbq/hyperguild/ingestion/internal/audit" + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- fakes --- + +type fakeCentral struct { + down bool + pushed []capture.AuditEntry + pushErr error +} + +func (f *fakeCentral) Ready(context.Context) error { + if f.down { + return errors.New("loki down") + } + return nil +} + +func (f *fakeCentral) Push(_ context.Context, e capture.AuditEntry) error { + if f.pushErr != nil { + return f.pushErr + } + f.pushed = append(f.pushed, e) + return nil +} + +type fakeNotifier struct{ msgs []string } + +func (f *fakeNotifier) Notify(_ context.Context, msg string) error { + f.msgs = append(f.msgs, msg) + return nil +} + +// unwritableBuffer always reports it cannot be written (floor condition). +type unwritableBuffer struct{} + +func (unwritableBuffer) Writable() error { return errors.New("disk full") } +func (unwritableBuffer) Append(capture.AuditEntry) error { return errors.New("disk full") } +func (unwritableBuffer) Pending() ([]audit.Buffered, error) { return nil, nil } +func (unwritableBuffer) Confirm(string) error { return nil } + +func newFileBuffer(t *testing.T) *audit.FileBuffer { + t.Helper() + b, err := audit.NewFileBuffer(filepath.Join(t.TempDir(), "audit-buffer.jsonl")) + require.NoError(t, err) + return b +} + +func entry(principal string) capture.AuditEntry { + return capture.AuditEntry{Principal: principal, EffectiveClassification: "internal", Items: []string{"insight:x"}} +} + +// --- Reserve: classification-aware decision --- + +func TestReserveCentralUpGrantsCentral(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{}, newFileBuffer(t), &fakeNotifier{}) + for _, lvl := range []classification.Level{classification.Public, classification.Internal, classification.Confidential} { + out, err := d.Reserve(context.Background(), lvl) + require.NoError(t, err) + assert.Equal(t, capture.AuditCentral, out) + } +} + +func TestReserveConfidentialSinkDownRefuses(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, newFileBuffer(t), &fakeNotifier{}) + _, err := d.Reserve(context.Background(), classification.Confidential) + require.Error(t, err, "confidential + sink down → refuse, no buffer") +} + +func TestReserveInternalSinkDownBuffers(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, newFileBuffer(t), &fakeNotifier{}) + out, err := d.Reserve(context.Background(), classification.Internal) + require.NoError(t, err) + assert.Equal(t, capture.AuditBuffered, out) +} + +func TestReserveFloorRefusesWhenNothingCanRecord(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, unwritableBuffer{}, &fakeNotifier{}) + _, err := d.Reserve(context.Background(), classification.Internal) + require.Error(t, err, "central down AND buffer unwritable → floor refuse") +} + +// --- Record: executes the reserved outcome --- + +func TestRecordCentralPushes(t *testing.T) { + c := &fakeCentral{} + d := audit.NewDegradingSink(c, newFileBuffer(t), &fakeNotifier{}) + require.NoError(t, d.Record(context.Background(), entry("p"), capture.AuditCentral)) + assert.Len(t, c.pushed, 1) +} + +func TestRecordBufferedAppendsAndNotifies(t *testing.T) { + buf := newFileBuffer(t) + nt := &fakeNotifier{} + d := audit.NewDegradingSink(&fakeCentral{down: true}, buf, nt) + require.NoError(t, d.Record(context.Background(), entry("p"), capture.AuditBuffered)) + + pending, err := buf.Pending() + require.NoError(t, err) + assert.Len(t, pending, 1) + assert.NotEmpty(t, nt.msgs, "degraded state alerts via ntfy") +} + +// --- FileBuffer durability + Confirm --- + +func TestFileBufferSurvivesRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "buf.jsonl") + b1, err := audit.NewFileBuffer(path) + require.NoError(t, err) + require.NoError(t, b1.Append(entry("p1"))) + require.NoError(t, b1.Append(entry("p2"))) + + // "restart": a fresh FileBuffer over the same file sees the records. + b2, err := audit.NewFileBuffer(path) + require.NoError(t, err) + pending, err := b2.Pending() + require.NoError(t, err) + assert.Len(t, pending, 2) +} + +func TestFileBufferConfirmRemovesOnlyThatRecord(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("keep"))) + require.NoError(t, buf.Append(entry("drop"))) + + pending, _ := buf.Pending() + require.Len(t, pending, 2) + var dropID string + for _, p := range pending { + if p.Entry.Principal == "drop" { + dropID = p.ID + } + } + require.NoError(t, buf.Confirm(dropID)) + + after, _ := buf.Pending() + require.Len(t, after, 1) + assert.Equal(t, "keep", after[0].Entry.Principal) +} + +// --- Reconcile --- + +func TestReconcileReplaysAndClearsOnlyAfterConfirmedWrite(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + require.NoError(t, buf.Append(entry("b"))) + c := &fakeCentral{} // up + nt := &fakeNotifier{} + + n, err := audit.Reconcile(context.Background(), c, buf, nt) + require.NoError(t, err) + assert.Equal(t, 2, n) + assert.Len(t, c.pushed, 2, "buffered records replayed to central") + + pending, _ := buf.Pending() + assert.Empty(t, pending, "buffer cleared after confirmed central writes") +} + +func TestReconcileNoopWhenCentralDown(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + n, err := audit.Reconcile(context.Background(), &fakeCentral{down: true}, buf, &fakeNotifier{}) + require.NoError(t, err) + assert.Equal(t, 0, n) + pending, _ := buf.Pending() + assert.Len(t, pending, 1, "records stay buffered while central is down") +} + +func TestReconcileKeepsRecordWhenPushFails(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + // Ready ok but Push fails → record must remain buffered (not lost). + c := &fakeCentral{pushErr: errors.New("push rejected")} + n, err := audit.Reconcile(context.Background(), c, buf, &fakeNotifier{}) + require.NoError(t, err) + assert.Equal(t, 0, n) + pending, _ := buf.Pending() + assert.Len(t, pending, 1) +} diff --git a/ingestion/internal/audit/loki.go b/ingestion/internal/audit/loki.go new file mode 100644 index 0000000..176a369 --- /dev/null +++ b/ingestion/internal/audit/loki.go @@ -0,0 +1,99 @@ +package audit + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" +) + +// LokiCentral pushes capture audit records to a Grafana Loki instance via +// its push API, and probes readiness via /ready. It is the central audit +// substrate behind DegradingSink. +type LokiCentral struct { + baseURL string + labels map[string]string + http *http.Client +} + +// NewLokiCentral constructs a LokiCentral for the given base URL (e.g. +// http://loki:3100). Returns nil when baseURL is empty so callers can +// treat missing config as "no central sink" with a single nil check. +func NewLokiCentral(baseURL string) *LokiCentral { + if baseURL == "" { + return nil + } + return &LokiCentral{ + baseURL: strings.TrimRight(baseURL, "/"), + labels: map[string]string{"service": "brain-capture", "kind": "audit"}, + http: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Ready probes Loki's readiness endpoint. +func (l *LokiCentral) Ready(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, l.baseURL+"/ready", nil) + if err != nil { + return err + } + resp, err := l.http.Do(req) + if err != nil { + return fmt.Errorf("loki not ready: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("loki not ready: status %d", resp.StatusCode) + } + return nil +} + +// pushPayload is the Loki push API body: one stream, one entry whose line +// is the JSON-encoded audit record. +type pushPayload struct { + Streams []lokiStream `json:"streams"` +} + +type lokiStream struct { + Stream map[string]string `json:"stream"` + Values [][2]string `json:"values"` +} + +// Push writes one audit record to Loki as a structured log line. +func (l *LokiCentral) Push(ctx context.Context, e capture.AuditEntry) error { + line, err := json.Marshal(e) + if err != nil { + return fmt.Errorf("marshal audit entry: %w", err) + } + ts := e.Timestamp + if ts.IsZero() { + ts = time.Now() + } + body, err := json.Marshal(pushPayload{Streams: []lokiStream{{ + Stream: l.labels, + Values: [][2]string{{strconv.FormatInt(ts.UTC().UnixNano(), 10), string(line)}}, + }}}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + l.baseURL+"/loki/api/v1/push", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := l.http.Do(req) + if err != nil { + return fmt.Errorf("loki push: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("loki push: status %d", resp.StatusCode) + } + return nil +} diff --git a/ingestion/internal/audit/loki_ntfy_test.go b/ingestion/internal/audit/loki_ntfy_test.go new file mode 100644 index 0000000..1c8d2f0 --- /dev/null +++ b/ingestion/internal/audit/loki_ntfy_test.go @@ -0,0 +1,86 @@ +package audit_test + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/mathiasbq/hyperguild/ingestion/internal/audit" + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLokiReadyAndPush(t *testing.T) { + var pushBody string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/ready": + w.WriteHeader(http.StatusOK) + case "/loki/api/v1/push": + b, _ := io.ReadAll(r.Body) + pushBody = string(b) + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer srv.Close() + + c := audit.NewLokiCentral(srv.URL) + require.NotNil(t, c) + require.NoError(t, c.Ready(context.Background())) + + err := c.Push(context.Background(), capture.AuditEntry{ + Principal: "koala-cli", EffectiveClassification: "internal", Items: []string{"insight:x"}, + }) + require.NoError(t, err) + assert.Contains(t, pushBody, "streams") + assert.Contains(t, pushBody, "koala-cli", "audit entry serialised into the loki line") +} + +func TestLokiReadyFailsWhenDown(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + require.Error(t, audit.NewLokiCentral(srv.URL).Ready(context.Background())) +} + +func TestLokiNilWhenUnconfigured(t *testing.T) { + assert.Nil(t, audit.NewLokiCentral("")) +} + +func TestNtfyNotify(t *testing.T) { + var gotBody, gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + gotBody = string(b) + gotAuth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + n := audit.NewNtfyNotifier(srv.URL, "ntfy-token") + require.NotNil(t, n) + require.NoError(t, n.Notify(context.Background(), "audit buffered locally")) + assert.Contains(t, gotBody, "audit buffered locally") + assert.Equal(t, "Bearer ntfy-token", gotAuth) +} + +func TestNtfyDoesNotLeakTokenOnError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + err := audit.NewNtfyNotifier(srv.URL, "secret-token").Notify(context.Background(), "x") + require.Error(t, err) + assert.False(t, strings.Contains(err.Error(), "secret-token"), "token must not leak into errors") +} + +func TestNtfyNilWhenUnconfigured(t *testing.T) { + assert.Nil(t, audit.NewNtfyNotifier("", "tok")) +} diff --git a/ingestion/internal/audit/ntfy.go b/ingestion/internal/audit/ntfy.go new file mode 100644 index 0000000..7aea375 --- /dev/null +++ b/ingestion/internal/audit/ntfy.go @@ -0,0 +1,55 @@ +package audit + +import ( + "context" + "fmt" + "net/http" + "strings" + "time" +) + +// NtfyNotifier posts alerts to an ntfy topic URL. Used to surface a +// degraded audit state (records buffered locally during a loki outage). +type NtfyNotifier struct { + topicURL string + token string + http *http.Client +} + +// NewNtfyNotifier constructs a notifier for the given ntfy topic URL +// (e.g. https://ntfy.sh/my-topic). token is an optional bearer for +// protected ntfy instances; it is held here and only sent in the +// Authorization header, never logged. Returns nil when topicURL is empty. +func NewNtfyNotifier(topicURL, token string) *NtfyNotifier { + if topicURL == "" { + return nil + } + return &NtfyNotifier{ + topicURL: strings.TrimRight(topicURL, "/"), + token: token, + http: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Notify posts a message to the ntfy topic. +func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.topicURL, strings.NewReader(msg)) + if err != nil { + return err + } + req.Header.Set("Title", "brain-capture audit degraded") + req.Header.Set("Priority", "high") + req.Header.Set("Tags", "warning,brain") + if n.token != "" { + req.Header.Set("Authorization", "Bearer "+n.token) + } + resp, err := n.http.Do(req) + if err != nil { + return fmt.Errorf("ntfy notify: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("ntfy notify: status %d", resp.StatusCode) + } + return nil +} diff --git a/ingestion/internal/audit/reconcile.go b/ingestion/internal/audit/reconcile.go new file mode 100644 index 0000000..845286e --- /dev/null +++ b/ingestion/internal/audit/reconcile.go @@ -0,0 +1,65 @@ +package audit + +import ( + "context" + "fmt" + "log/slog" + "time" +) + +// Reconcile replays locally-buffered audit records to the central sink +// when it is reachable again. A record is removed from the buffer ONLY +// after its central write is confirmed, so a crash mid-reconcile re-plays +// rather than loses. Returns the number of records reconciled. +// +// A no-op (0, nil) when the central sink is still unreachable or the +// buffer is empty. +func Reconcile(ctx context.Context, central Central, buffer Buffer, notifier Notifier) (int, error) { + if err := central.Ready(ctx); err != nil { + return 0, nil // still down; try again next tick + } + pending, err := buffer.Pending() + if err != nil { + return 0, fmt.Errorf("read buffer: %w", err) + } + reconciled := 0 + for _, rec := range pending { + if err := central.Push(ctx, rec.Entry); err != nil { + // Central went away mid-drain; stop and keep the rest buffered. + break + } + if err := buffer.Confirm(rec.ID); err != nil { + return reconciled, fmt.Errorf("confirm buffered record %s: %w", rec.ID, err) + } + reconciled++ + } + if reconciled > 0 && notifier != nil { + _ = notifier.Notify(ctx, fmt.Sprintf("reconciled %d buffered capture audit record(s) to loki", reconciled)) + } + return reconciled, nil +} + +// StartReconcile runs Reconcile on a ticker until ctx is cancelled. It is +// the recovery half of the degrade-and-buffer path; pair it with a +// DegradingSink sharing the same buffer + central. +func StartReconcile(ctx context.Context, central Central, buffer Buffer, notifier Notifier, interval time.Duration) { + if interval <= 0 { + interval = time.Minute + } + go func() { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if n, err := Reconcile(ctx, central, buffer, notifier); err != nil { + slog.Warn("audit reconcile failed", "err", err) + } else if n > 0 { + slog.Info("audit reconcile", "reconciled", n) + } + } + } + }() +} diff --git a/ingestion/internal/audit/slog.go b/ingestion/internal/audit/slog.go index b7d57d2..dda0f4b 100644 --- a/ingestion/internal/audit/slog.go +++ b/ingestion/internal/audit/slog.go @@ -11,11 +11,14 @@ import ( "log/slog" "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" ) -// SlogSink records audit entries to an slog.Logger. It never fails, so it -// does not exercise the I5 floor (refuse-if-unauditable) — that is #54's -// loki+buffer sink. A nil logger falls back to slog.Default(). +// SlogSink records audit entries to an slog.Logger. It never fails and is +// always centrally available, so its Reserve always grants AuditCentral — +// it does not exercise the I5 degradation/floor. That is DegradingSink's +// job (loki + durable buffer). SlogSink is the default for deployments +// without a loki endpoint configured. A nil logger ⇒ slog.Default(). type SlogSink struct { logger *slog.Logger } @@ -28,10 +31,15 @@ func NewSlogSink(logger *slog.Logger) *SlogSink { return &SlogSink{logger: logger} } +// Reserve always grants central recording — slog is always available. +func (s *SlogSink) Reserve(_ context.Context, _ classification.Level) (capture.AuditOutcome, error) { + return capture.AuditCentral, nil +} + // Record emits the audit entry at info level. Security events, when // present, are logged at warn level so they surface independently of the // routine audit stream. -func (s *SlogSink) Record(_ context.Context, e capture.AuditEntry) error { +func (s *SlogSink) Record(_ context.Context, e capture.AuditEntry, _ capture.AuditOutcome) error { s.logger.Info("capture audit", "principal", e.Principal, "actor", e.Actor, diff --git a/ingestion/internal/audit/slog_test.go b/ingestion/internal/audit/slog_test.go index 687528d..b1ab29a 100644 --- a/ingestion/internal/audit/slog_test.go +++ b/ingestion/internal/audit/slog_test.go @@ -22,7 +22,7 @@ func TestSlogSinkRecordsEntryAndSecurityEvents(t *testing.T) { EffectiveClassification: "confidential", Items: []string{"insight:wiki/a/facts/x.md"}, SecurityEvents: []string{"asserted-vs-derived origin mismatch"}, - }) + }, capture.AuditCentral) require.NoError(t, err) out := buf.String() @@ -36,6 +36,6 @@ func TestSlogSinkRecordsEntryAndSecurityEvents(t *testing.T) { func TestSlogSinkNilLoggerDefaults(t *testing.T) { // nil logger must not panic. require.NotPanics(t, func() { - _ = audit.NewSlogSink(nil).Record(context.Background(), capture.AuditEntry{}) + _ = audit.NewSlogSink(nil).Record(context.Background(), capture.AuditEntry{}, capture.AuditCentral) }) } From 38a2e91002144d88ee84904622f7d5106a13443a Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 22 Jun 2026 23:54:24 +0200 Subject: [PATCH 3/3] feat(capturehttp): 503 on audit-unavailable; wire degrading sink (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Map capture.ErrAuditUnavailable → HTTP 503 (audit substrate down / confidential unauditable / floor). - main: buildAuditSink selects the DegradingSink (loki central + durable file buffer under brainDir + optional ntfy) when BRAIN_LOKI_URL is set and starts the reconcile loop; else the plain slog sink. Notifier kept as a nil interface (not typed-nil) when unconfigured so the sink and reconcile skip it cleanly. Env: BRAIN_LOKI_URL, BRAIN_NTFY_URL, BRAIN_NTFY_TOKEN, BRAIN_AUDIT_RECONCILE_INTERVAL (default 60s). Buffer at /.audit-buffer/capture.jsonl. Co-Authored-By: Claude Opus 4.8 (1M context) --- ingestion/cmd/server/main.go | 42 ++++++++++++++++--- ingestion/internal/capturehttp/handler.go | 4 ++ .../internal/capturehttp/handler_test.go | 20 +++++++++ 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index f513463..9e1de43 100644 --- a/ingestion/cmd/server/main.go +++ b/ingestion/cmd/server/main.go @@ -8,6 +8,7 @@ import ( "net/http" "net/url" "os" + "path/filepath" "strconv" "strings" "time" @@ -15,11 +16,11 @@ import ( chassisauth "gitea.d-ma.be/mathias/mcp-chassis/auth" "github.com/mathiasbq/hyperguild/ingestion/internal/api" - "github.com/mathiasbq/hyperguild/ingestion/internal/claudewatcher" "github.com/mathiasbq/hyperguild/ingestion/internal/audit" "github.com/mathiasbq/hyperguild/ingestion/internal/capture" "github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp" "github.com/mathiasbq/hyperguild/ingestion/internal/classification" + "github.com/mathiasbq/hyperguild/ingestion/internal/claudewatcher" "github.com/mathiasbq/hyperguild/ingestion/internal/embed" "github.com/mathiasbq/hyperguild/ingestion/internal/gitea" "github.com/mathiasbq/hyperguild/ingestion/internal/graphstore" @@ -123,6 +124,35 @@ func envInt(key string, fallback int) int { return fallback } +// buildAuditSink selects the capture audit sink. When BRAIN_LOKI_URL is +// set it builds the classification-aware DegradingSink (loki central + +// durable file buffer + optional ntfy) and starts the reconcile loop; +// otherwise it falls back to a plain slog sink. The buffer lives under the +// brain dir so it survives process restarts. +func buildAuditSink(ctx context.Context, brainDir string, logger *slog.Logger) capture.AuditSink { + lokiURL := os.Getenv("BRAIN_LOKI_URL") + central := audit.NewLokiCentral(lokiURL) + if central == nil { + logger.Info("capture audit: slog sink (BRAIN_LOKI_URL unset)") + return audit.NewSlogSink(logger) + } + buffer, err := audit.NewFileBuffer(filepath.Join(brainDir, ".audit-buffer", "capture.jsonl")) + if err != nil { + logger.Error("capture audit buffer init", "err", err) + os.Exit(1) + } + // Keep notifier as a nil interface (not a typed-nil) when unconfigured + // so DegradingSink/Reconcile skip it cleanly. + var notifier audit.Notifier + if n := audit.NewNtfyNotifier(os.Getenv("BRAIN_NTFY_URL"), os.Getenv("BRAIN_NTFY_TOKEN")); n != nil { + notifier = n + } + reconcileInterval := time.Duration(envInt("BRAIN_AUDIT_RECONCILE_INTERVAL", 60)) * time.Second + audit.StartReconcile(ctx, central, buffer, notifier, reconcileInterval) + logger.Info("capture audit: loki+buffer sink", "loki", lokiURL, "reconcile_s", int(reconcileInterval.Seconds())) + return audit.NewDegradingSink(central, buffer, notifier) +} + // splitList parses a comma-separated env value into a trimmed, // empty-free slice. Used for the capture sovereign-principal allowlist. func splitList(v string) []string { @@ -365,11 +395,12 @@ func main() { mux.Handle("/mcp", chassisauth.BearerMiddleware(mcpToken, jwtValidator, "brain", resourceMetadataURL, mcpSrv)) - // POST /capture (#53): the uniform capture REST door. Needs a ticket + // POST /capture (#53/#54): the uniform capture REST door. Needs a ticket // tracker to file action items, so it only mounts when Gitea is // configured. It reuses the MCP server's graph-wired brain store (one - // implementation), the classification tags for the I1 gate, and a slog - // audit sink (the loki+buffer sink lands in #54). The handler does its + // implementation), the classification tags for the I1 gate, and a + // classification-aware audit sink (loki + durable buffer + ntfy when + // BRAIN_LOKI_URL is set, else a plain slog sink). The handler does its // own auth (static + JWT) because it needs the principal to derive the // trust-zone origin — the chassis middleware hides it. if tracker := mcpSrv.IssueTracker(); tracker != nil { @@ -378,8 +409,9 @@ func main() { logger.Error("load classification config", "err", cerr) os.Exit(1) } + auditSink := buildAuditSink(ctx, brainDir, logger) captureSvc := capture.NewService( - mcpSrv.BrainStore(), tracker, nil, classCfg, audit.NewSlogSink(logger)) + mcpSrv.BrainStore(), tracker, nil, classCfg, auditSink) sovereign := splitList(os.Getenv("BRAIN_CAPTURE_SOVEREIGN_PRINCIPALS")) captureH := capturehttp.New(captureSvc, jwtValidator, mcpToken, "local-cli", capturehttp.NewOriginResolver(sovereign)) diff --git a/ingestion/internal/capturehttp/handler.go b/ingestion/internal/capturehttp/handler.go index 8bc73f0..2fd1c73 100644 --- a/ingestion/internal/capturehttp/handler.go +++ b/ingestion/internal/capturehttp/handler.go @@ -113,6 +113,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { case errors.Is(err, capture.ErrSovereigntyRefused): writeJSON(w, http.StatusForbidden, map[string]string{"error": err.Error()}) return + case errors.Is(err, capture.ErrAuditUnavailable): + // I5 refusal: confidential + audit sink down, or the all-tiers floor. + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) + return case err != nil: // Pre-write validation failure (fail-closed). writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) diff --git a/ingestion/internal/capturehttp/handler_test.go b/ingestion/internal/capturehttp/handler_test.go index a2247cc..e33342c 100644 --- a/ingestion/internal/capturehttp/handler_test.go +++ b/ingestion/internal/capturehttp/handler_test.go @@ -176,3 +176,23 @@ func TestCallerCannotForgeOrigin(t *testing.T) { }) assert.Equal(t, http.StatusForbidden, rr.Code) } + +// refusingAudit refuses at Reserve (e.g. confidential + loki down, or floor). +type refusingAudit struct{} + +func (refusingAudit) Reserve(context.Context, classification.Level) (capture.AuditOutcome, error) { + return 0, errors.New("central audit sink unreachable") +} +func (refusingAudit) Record(context.Context, capture.AuditEntry, capture.AuditOutcome) error { + return nil +} + +func TestAuditUnavailableIs503(t *testing.T) { + cfg, err := classification.Load(t.TempDir()) + require.NoError(t, err) + svc := capture.NewService(brainstore.New(t.TempDir()), fakeTracker{}, nil, cfg, refusingAudit{}) + h := capturehttp.New(svc, nil, staticTok, "local-cli", capturehttp.NewOriginResolver(nil)) + + rr := do(t, h, "Bearer "+staticTok, internalReq()) + assert.Equal(t, http.StatusServiceUnavailable, rr.Code) +}