feat(capture): two-phase classification-aware AuditSink port (#54)

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) <noreply@anthropic.com>

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"`
 }

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
 }

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
 }

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)
+}