feat(audit): DegradingSink + durable buffer + loki/ntfy + reconcile (#54)

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) <noreply@anthropic.com>
2026-06-22 23:54:24 +02:00
parent 202212e8d5
commit 77f5e06d6b
9 changed files with 773 additions and 6 deletions
@@ -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]
+}
@@ -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)
+	}
+}
@@ -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)
+}
@@ -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
+}
@@ -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"))
+}
@@ -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
+}
@@ -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)
+				}
+			}
+		}
+	}()
+}
@@ -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,
@@ -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)
 	})
 }