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

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

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