hyperguild/ingestion/cmd/server/main.go

// ingestion/cmd/server/main.go
package main

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	chassisauth "git.d-ma.be/mathias/mcp-chassis/auth"

	"github.com/mathiasbq/hyperguild/ingestion/internal/api"
	"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"
	"github.com/mathiasbq/hyperguild/ingestion/internal/graphsync"
	"github.com/mathiasbq/hyperguild/ingestion/internal/llm"
	"github.com/mathiasbq/hyperguild/ingestion/internal/mcp"
	"github.com/mathiasbq/hyperguild/ingestion/internal/metrics"
	"github.com/mathiasbq/hyperguild/ingestion/internal/oauth"
	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
	"github.com/mathiasbq/hyperguild/ingestion/internal/reranker"
	"github.com/mathiasbq/hyperguild/ingestion/internal/search"
	"github.com/mathiasbq/hyperguild/ingestion/internal/symlinkguard"
	"github.com/mathiasbq/hyperguild/ingestion/internal/vectorstore"
	"github.com/mathiasbq/hyperguild/ingestion/internal/watcher"
	"github.com/mathiasbq/hyperguild/ingestion/internal/webhook"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
)

// kubeClient builds an in-cluster Kubernetes client (falls back to
// $KUBECONFIG for local dev/testing against a real cluster).
func kubeClient() (kubernetes.Interface, error) {
	if cfg, err := rest.InClusterConfig(); err == nil {
		return kubernetes.NewForConfig(cfg)
	}
	rules := clientcmd.NewDefaultClientConfigLoadingRules()
	cc := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(rules, &clientcmd.ConfigOverrides{})
	cfg, err := cc.ClientConfig()
	if err != nil {
		return nil, fmt.Errorf("kube config (no in-cluster, no kubeconfig): %w", err)
	}
	return kubernetes.NewForConfig(cfg)
}

// claudeSink converts each claudewatcher.Batch into a raw session dump
// under brain/archive/claude-sessions/<host>/. Deliberately NOT a wiki
// note (api.WriteNote / brain/wiki/) — raw full transcripts out-ranked
// curated ai-sessions summaries in BM25 (177,818 vs 45,335 on the same
// query) and duplicated content already summarized elsewhere. Kept for
// deep lookups, never indexed. See ai-sessions#10.
type claudeSink struct {
	brainDir string
	logger   *slog.Logger
}

func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error {
	if len(b.Turns) == 0 {
		return nil
	}
	var sb strings.Builder
	fmt.Fprintf(&sb, "# Claude session %s (%s)\n\n", b.SessionID, b.Host)
	fmt.Fprintf(&sb, "_Project: `%s`. File: `%s`. Turns: %d._\n\n", b.ProjectID, b.FilePath, len(b.Turns))
	for _, t := range b.Turns {
		fmt.Fprintf(&sb, "## %s — %s\n\n", t.Type, t.Timestamp.UTC().Format(time.RFC3339))
		if t.ToolName != "" {
			fmt.Fprintf(&sb, "_tool: `%s`_\n\n", t.ToolName)
		}
		// Cap per-turn excerpt to keep page size bounded; the full
		// transcript lives on disk under ~/.claude/projects/ already.
		content := t.Content
		if len(content) > 2000 {
			content = content[:2000] + "…"
		}
		sb.WriteString(content)
		sb.WriteString("\n\n")
	}
	slug := "session-" + b.Host + "-" + b.SessionID
	dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md")
	if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
		return fmt.Errorf("create claude-sessions archive dir: %w", err)
	}
	if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil {
		return fmt.Errorf("write claude session archive: %w", err)
	}
	s.logger.Debug("claude session archived (non-indexed)", "path", dest)
	return nil
}

// redactDSN parses a Postgres URL and replaces its password with `***`
// for safe inclusion in logs. Falls back to a non-leaking placeholder
// if parsing fails — we never log a raw DSN.
func redactDSN(dsn string) string {
	u, err := url.Parse(dsn)
	if err != nil || u.User == nil {
		return "postgres://***"
	}
	return u.Redacted()
}

// vectorAdapter bridges *vectorstore.PGStore (returns []vectorstore.Hit)
// to the search.VectorSearcher interface (which uses []search.VectorHit).
// Kept here, not in either package, so neither has to import the other.
type vectorAdapter struct{ s *vectorstore.PGStore }

func (a vectorAdapter) Search(ctx context.Context, q []float32, limit int) ([]search.VectorHit, error) {
	hits, err := a.s.Search(ctx, q, limit)
	if err != nil {
		return nil, err
	}
	out := make([]search.VectorHit, len(hits))
	for i, h := range hits {
		out[i] = search.VectorHit{Path: h.Path, Distance: h.Distance}
	}
	return out, nil
}

func envOr(key, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}

func envInt(key string, fallback int) int {
	if v := os.Getenv(key); v != "" {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}
	return fallback
}

// buildNotifier returns the ntfy notifier, or a nil interface when
// BRAIN_NTFY_URL is unset. A nil *interface* (not a typed-nil) is required so
// DegradingSink/Reconcile and the governance observer skip it cleanly.
// alertingWarnings names every alert that will not fire under the current
// configuration (#107).
//
// buildNotifier returns a nil interface when BRAIN_NTFY_URL is unset, and every
// consumer skips a nil notifier silently — right for an unconfigured homelab,
// wrong for an alert whose entire value is that someone finds out. The cost of
// the silence is that "configured and working" and "never configured" are
// byte-identical in behaviour: no error, no warning, only a message that never
// arrives, which nobody notices because nobody is waiting for an alert they
// have never seen. Two alerts were dark in production for exactly that reason
// (#100), found by firing a real event rather than by reading anything.
//
// WARN rather than fatal, deliberately. The brain sibling rule
// (`empty-credential-must-refuse-to-start-not-switch-modes`) says an empty
// credential must be a startup error, never a silent mode switch — but that is
// about a control being removed. Here observability is degraded, not a gate
// bypassed, and refusing to boot would break the unconfigured-homelab case the
// nil notifier exists to support. What the rule does demand, and what this
// provides, is that the mode switch stops being silent.
//
// Pure and env-injected so the message is testable without a running server:
// the reason this gap survived is that nothing about it was observable.
func alertingWarnings(getenv func(string) string) []string {
	var out []string
	if getenv("BRAIN_NTFY_URL") == "" {
		out = append(out, "alerting DISABLED: BRAIN_NTFY_URL unset — "+
			"brain_edit_root governance rewrites, capture audit degradation and the "+
			"brain symlink tripwire (hyperguild#157) will not notify")
	}
	return out
}

func buildNotifier() audit.Notifier {
	if n := audit.NewNtfyNotifier(os.Getenv("BRAIN_NTFY_URL"), os.Getenv("BRAIN_NTFY_TOKEN")); n != nil {
		return n
	}
	return nil
}

// 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)
	}
	notifier := buildNotifier()
	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 {
	var out []string
	for _, p := range strings.Split(v, ",") {
		if p = strings.TrimSpace(p); p != "" {
			out = append(out, p)
		}
	}
	return out
}

// systemHostname returns os.Hostname() with a "unknown" fallback so the
// caller never has to handle the rare error path.
func systemHostname() string {
	h, err := os.Hostname()
	if err != nil || h == "" {
		return "unknown"
	}
	return h
}

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))

	for _, w := range alertingWarnings(os.Getenv) {
		logger.Warn(w)
	}

	brainDir := envOr("INGEST_BRAIN_DIR", "../brain")
	if err := startupSymlinkCheck(brainDir); err != nil {
		logger.Error("brain symlink tripwire", "err", err)
		os.Exit(1)
	}
	port := envOr("INGEST_PORT", "3300")

	llmURL := envOr("INGEST_LLM_URL", "http://iguana:4000/v1")
	llmKey := os.Getenv("INGEST_LLM_KEY")
	llmModel := envOr("INGEST_LLM_MODEL", "koala/qwen35-9b-fast")
	llmTimeoutMins := envInt("INGEST_LLM_TIMEOUT", 15)
	chunkSize := envInt("INGEST_CHUNK_SIZE", 6000)
	watchInterval := envInt("INGEST_WATCH_INTERVAL", 30)

	llmClient := llm.New(llmURL, llmKey, llmModel, time.Duration(llmTimeoutMins)*time.Minute)

	pipelineCfg := pipeline.Config{
		Complete:  llmClient.Complete,
		ChunkSize: chunkSize,
	}

	h := api.NewHandler(brainDir, logger, pipelineCfg)

	var answerComplete pipeline.CompleteFunc
	if primaryURL := os.Getenv("BRAIN_LLM_PRIMARY_URL"); primaryURL != "" {
		primaryModel := envOr("BRAIN_LLM_PRIMARY_MODEL", "gemma4:31b")
		primaryKey := os.Getenv("BERGET_API_KEY")
		timeoutMS := envInt("BRAIN_LLM_TIMEOUT_MS", 10000)
		timeout := time.Duration(timeoutMS) * time.Millisecond

		primary := llm.New(primaryURL, primaryKey, primaryModel, timeout)
		router := &llm.Router{Primary: primary}

		if fallbackURL := os.Getenv("BRAIN_LLM_FALLBACK_URL"); fallbackURL != "" {
			fallbackModel := envOr("BRAIN_LLM_FALLBACK_MODEL", "gemma4:31b")
			router.Fallback = llm.New(fallbackURL, "", fallbackModel, timeout)
		}
		if tertiaryURL := os.Getenv("BRAIN_LLM_TERTIARY_URL"); tertiaryURL != "" {
			tertiaryModel := envOr("BRAIN_LLM_TERTIARY_MODEL", "gemini-2.5-flash")
			tertiaryKey := os.Getenv("GEMINI_API_KEY")
			router.Tertiary = llm.New(tertiaryURL, tertiaryKey, tertiaryModel, timeout)
		}
		answerComplete = router.Complete
		logger.Info("brain answer LLM configured", "primary", primaryURL, "model", primaryModel)
	}

	mcpSrv := mcp.NewServer(brainDir, &pipelineCfg, llmClient.Complete, answerComplete)
	if rerankURL := os.Getenv("BRAIN_RERANKER_URL"); rerankURL != "" {
		rerankModel := envOr("BRAIN_RERANKER_MODEL", "dengcao/Qwen3-Reranker-0.6B:F16")
		mcpSrv = mcpSrv.WithReranker(reranker.New(rerankURL, rerankModel))
		logger.Info("brain reranker configured", "url", rerankURL, "model", rerankModel)
	}

	// Gitea ticket tracker for the capture capability (#52). Token via env
	// only — never logged or in argv. Both vars must be set to enable it;
	// gitea.New returns nil otherwise, leaving ticket integration off.
	giteaURL := envOr("BRAIN_GITEA_URL", "https://git.d-ma.be")
	if tracker := gitea.New(giteaURL, os.Getenv("BRAIN_GITEA_TOKEN")); tracker != nil {
		mcpSrv = mcpSrv.WithIssueTracker(tracker)
		logger.Info("brain gitea tracker configured", "url", giteaURL)
	}

	// Hybrid retrieval (pgvector + nomic-embed-text). Both env vars must
	// be set together for the path to wire on; otherwise BM25-only.
	var vectorStore *vectorstore.PGStore
	pgDSN := os.Getenv("BRAIN_PG_DSN")
	embedURL := os.Getenv("BRAIN_EMBED_URL")
	switch {
	case pgDSN != "" && embedURL != "":
		embedModel := envOr("BRAIN_EMBED_MODEL", "nomic-embed-text:latest")
		store, err := vectorstore.New(context.Background(), pgDSN)
		if err != nil {
			logger.Error("vector store init", "err", err)
			os.Exit(1)
		}
		if err := store.Init(context.Background()); err != nil {
			logger.Error("vector store migrate", "err", err)
			os.Exit(1)
		}
		vectorStore = store
		embedder := embed.New(embedURL, embedModel)
		mcpSrv = mcpSrv.WithHybridRetrieval(vectorAdapter{s: store}, embedder)
		h.WithEmbedSync(store, embedder)
		logger.Info("brain hybrid retrieval enabled",
			"pg", redactDSN(pgDSN),
			"embed_url", embedURL, "embed_model", embedModel)

		// Graph store shares the same postgres18 DSN as the vector
		// store and is opt-in via BRAIN_GRAPH_ENABLED=true. Defaults
		// to off so first rollout doesn't surprise — flip on after
		// the migration completes and the backfill finishes.
		if envOr("BRAIN_GRAPH_ENABLED", "false") == "true" {
			gstore, gerr := graphstore.New(context.Background(), pgDSN)
			if gerr != nil {
				logger.Error("graph store init", "err", gerr)
				os.Exit(1)
			}
			if gerr := gstore.Init(context.Background()); gerr != nil {
				logger.Error("graph store migrate", "err", gerr)
				os.Exit(1)
			}
			mcpSrv = mcpSrv.WithGraph(gstore)
			if envOr("BRAIN_GRAPH_BACKFILL", "false") == "true" {
				n, berr := graphsync.BackfillFromBrainDir(context.Background(), gstore, brainDir)
				if berr != nil {
					logger.Warn("graph backfill incomplete", "indexed", n, "err", berr)
				} else {
					logger.Info("graph backfill complete", "indexed", n)
				}
			}
			logger.Info("brain graph enabled", "pg", redactDSN(pgDSN))
		}
	case pgDSN == "" && embedURL == "":
		// disabled — fine
	default:
		logger.Error("BRAIN_PG_DSN and BRAIN_EMBED_URL must be set together")
		os.Exit(1)
	}

	mcpToken := os.Getenv("BRAIN_MCP_TOKEN")
	if mcpToken == "" {
		logger.Error("BRAIN_MCP_TOKEN not set")
		os.Exit(1)
	}

	ctx := context.Background()
	// hyperguild#157: report a symlink that appears in the brain while running.
	go symlinkguard.Watch(ctx, brainDir, symlinkScanInterval(os.Getenv), buildNotifier())

	// Never fatal on an unreachable issuer (#83): on a cold cluster boot this
	// pod can beat dex to serving and get a 502 from ingress, and exiting meant
	// CrashLoopBackOff redoing the backfill. Since mcp-chassis v0.7 the
	// validator resolves the issuer lazily and answers JWTs 503-and-retry until
	// it does (hyperguild#155), so no boot retry is needed. A misconfigured
	// issuer no longer exits the process: it is logged at boot and every JWT
	// gets 503; the static bearer keeps working.
	jwtValidator, err := buildJWTValidator(ctx, logger,
		os.Getenv("DEX_ISSUER_URL"), os.Getenv("MCP_AUDIENCE"))
	if err != nil {
		logger.Error("build jwt validator", "err", err)
		os.Exit(1)
	}
	if jwtValidator != nil {
		logger.Info("jwt auth enabled", "issuer", os.Getenv("DEX_ISSUER_URL"))
	}

	if watchInterval > 0 {
		watcher.Start(ctx, watcher.Config{
			BrainDir: brainDir,
			Interval: time.Duration(watchInterval) * time.Second,
			Pipeline: pipelineCfg,
		})
	}

	// Claude Code session ingestion (hyperguild#27 / infra#73 Track E.1).
	// Off by default — explicitly opt in by setting CLAUDE_SESSIONS_DIR
	// to the ~/.claude/projects path. Requires BRAIN_PG_DSN for the
	// cursor table (resumable offsets across restarts).
	if claudeDir := os.Getenv("CLAUDE_SESSIONS_DIR"); claudeDir != "" {
		if pgDSN == "" {
			logger.Error("CLAUDE_SESSIONS_DIR set but BRAIN_PG_DSN missing — claudewatcher needs the cursor table")
			os.Exit(1)
		}
		// Client-name guard. The env value is a regex alternation
		// (e.g. "SEB|Mastercard"); we wrap it with word boundaries
		// and case-insensitive flag so substrings inside longer
		// identifiers don't false-match. Sourced from a SOPS secret
		// so client identities never live in source.
		if clientBlock := os.Getenv("CLAUDE_INGEST_CLIENT_BLOCK"); clientBlock != "" {
			pattern := `(?i)\b(` + clientBlock + `)\b`
			if err := claudewatcher.RegisterRule("client-name", pattern); err != nil {
				logger.Error("claudewatcher client-block rule invalid", "err", err)
				os.Exit(1)
			}
			logger.Info("claudewatcher client-block guard registered")
		}
		cursorStore, cerr := claudewatcher.NewCursorStore(ctx, pgDSN)
		if cerr != nil {
			logger.Error("claudewatcher cursor init", "err", cerr)
			os.Exit(1)
		}
		if cerr := cursorStore.Init(ctx); cerr != nil {
			logger.Error("claudewatcher cursor migrate", "err", cerr)
			os.Exit(1)
		}
		host := envOr("CLAUDE_INGEST_HOST", systemHostname())
		interval := time.Duration(envInt("CLAUDE_INGEST_INTERVAL", 60)) * time.Second

		// Project directories excluded from ingestion entirely, as globs
		// against the project directory name — e.g.
		// "*cobalt-dingo*,*claude-worktrees-bridge-cse*".
		//
		// This is a veto list, deliberately named as one. It complements the
		// client-name scrubber rather than replacing it: the scrubber drops
		// individual TURNS whose content matches, while this never opens a
		// project's files at all. Sessions working on live company financials
		// are excluded here because their content is confidential in ways no
		// pattern reliably recognises, and their learnings are captured
		// deliberately through the session-close ritual instead.
		//
		// Caveat worth knowing: a project directory is derived from the
		// session's WORKING DIRECTORY, not from the repositories it touches.
		// A session run from ~/dev that happens to read a skipped repo will
		// NOT be excluded by this. The scrubber remains the control for
		// content; this is a control for whole workspaces.
		skipProjects := splitList(os.Getenv("CLAUDE_INGEST_SKIP_PROJECTS"))
		if len(skipProjects) > 0 {
			logger.Info("claudewatcher: projects excluded from ingestion", "patterns", skipProjects)
		}

		sink := &claudeSink{brainDir: brainDir, logger: logger}
		go func() {
			if err := claudewatcher.Watch(ctx, claudewatcher.Config{
				SessionsDir:  claudeDir,
				Host:         host,
				Interval:     interval,
				Sink:         sink,
				Cursors:      cursorStore,
				Logger:       logger,
				SkipProjects: skipProjects,
			}); err != nil && err != context.Canceled {
				logger.Error("claudewatcher exited", "err", err)
			}
		}()
		logger.Info("claudewatcher started",
			"sessions_dir", claudeDir, "host", host, "interval", interval)
	}

	// Gitea push webhook -> on-demand brain-sync Job, instead of waiting up
	// to 15 minutes for the next CronJob poll. Off by default (opt in via
	// GITEA_WEBHOOK_SECRET) since it needs Job-create RBAC in the "brain"
	// namespace that a fresh deploy won't have granted yet.
	var webhookHandler *webhook.Handler
	if webhookSecret := os.Getenv("GITEA_WEBHOOK_SECRET"); webhookSecret != "" {
		kc, err := kubeClient()
		if err != nil {
			logger.Error("brain-sync webhook: kube client", "err", err)
			os.Exit(1)
		}
		webhookHandler = &webhook.Handler{
			Secret:      webhookSecret,
			Clientset:   kc,
			Namespace:   envOr("BRAIN_SYNC_NAMESPACE", "brain"),
			CronJobName: envOr("BRAIN_SYNC_CRONJOB", "brain-sync"),
			WatchRepo:   envOr("BRAIN_SYNC_WATCH_REPO", "mathias/brain"),
			Logger:      logger,
		}
		logger.Info("brain-sync webhook enabled", "namespace", webhookHandler.Namespace, "cronjob", webhookHandler.CronJobName)
	}

	if vectorStore != nil {
		embedSyncInterval := envInt("BRAIN_EMBED_SYNC_INTERVAL", 300)
		vectorstore.StartSync(ctx, brainDir, vectorStore,
			embed.New(os.Getenv("BRAIN_EMBED_URL"),
				envOr("BRAIN_EMBED_MODEL", "nomic-embed-text:latest")),
			time.Duration(embedSyncInterval)*time.Second)
		logger.Info("embed sync started", "interval_s", embedSyncInterval)
	}

	// Public routes only; buildMux mounts the authenticated REST API (#149).
	public := map[string]http.Handler{}
	if webhookHandler != nil {
		public["POST /webhooks/brain-sync"] = webhookHandler
	}
	// Resource-metadata URL is only emitted on 401 when Dex OAuth is
	// configured. Static-Bearer-only deployments leave this empty so
	// clients never see an OAuth challenge.
	var resourceMetadataURL string
	if dexURL := os.Getenv("DEX_ISSUER_URL"); dexURL != "" {
		resourceURL := os.Getenv("MCP_RESOURCE_URL")
		public["GET /.well-known/oauth-protected-resource"] = http.HandlerFunc(
			chassisauth.ProtectedResourceHandler(resourceURL, dexURL))
		if resourceURL != "" {
			resourceMetadataURL = strings.TrimRight(resourceURL, "/") + "/.well-known/oauth-protected-resource"
		}
	}

	// 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
	// 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 {
		classCfg, cerr := classification.Load(brainDir)
		if cerr != nil {
			logger.Error("load classification config", "err", cerr)
			os.Exit(1)
		}
		auditSink := buildAuditSink(ctx, brainDir, logger)
		captureSvc := capture.NewService(
			mcpSrv.BrainStore(), tracker, classCfg, auditSink)
		sovereign := splitList(os.Getenv("BRAIN_CAPTURE_SOVEREIGN_PRINCIPALS"))
		resolver := capturehttp.NewOriginResolver(sovereign)
		captureH := capturehttp.New(captureSvc, jwtValidator, mcpToken, "local-cli", resolver)
		public["POST /capture"] = captureH
		// Same use-case behind the MCP `capture` tool (#55 relay) so MCP-native
		// harnesses (claude.ai, Crush, Pi, LLM Council) reach capture through
		// the existing /mcp OAuth connector. buildMux wraps this same mcpSrv;
		// WithCapture mutates the same instance, so the tool appears live.
		mcpSrv.WithCapture(captureSvc, jwtValidator, mcpToken, "local-cli", resolver)
		// hyperguild#97: a governance-file rewrite gets the same trail a
		// capture already gets, plus an out-of-band alert. Wired here because
		// brain_edit_root needs the capture plumbing anyway — that is what
		// re-derives the static principal its gate checks.
		mcpSrv.WithGovernanceObserver(auditSink, buildNotifier())
		logger.Info("capture enabled (REST + MCP tool)", "sovereign_principals", len(sovereign),
			"governance_observer", true)
	} else {
		logger.Info("capture endpoint disabled (BRAIN_GITEA_TOKEN unset)")
	}

	// Opt-in OAuth 2.0 client_credentials flow for claude.ai's custom-MCP
	// integration UI, which has no static-Bearer field. Setting both
	// OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET enables the token exchange;
	// setting only one is misconfiguration → fail fast.
	oauthID := os.Getenv("OAUTH_CLIENT_ID")
	oauthSecret := os.Getenv("OAUTH_CLIENT_SECRET")
	switch {
	case oauthID != "" && oauthSecret != "":
		issuer := os.Getenv("MCP_RESOURCE_URL")
		if issuer == "" {
			logger.Error("OAUTH_CLIENT_ID/SECRET set but MCP_RESOURCE_URL is empty; cannot derive issuer")
			os.Exit(1)
		}
		public["GET /.well-known/oauth-authorization-server"] = http.HandlerFunc(
			oauth.MetadataHandler(issuer))
		public["POST /oauth/token"] = http.HandlerFunc(oauth.TokenHandler(oauth.TokenConfig{
			ClientID:     oauthID,
			ClientSecret: oauthSecret,
			AccessToken:  mcpToken,
		}))
		logger.Info("oauth client_credentials enabled", "issuer", strings.TrimRight(issuer, "/"))
	case oauthID == "" && oauthSecret == "":
		// disabled — that's fine
	default:
		logger.Error("OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET must be set together")
		os.Exit(1)
	}

	// /metrics — unauthenticated Prometheus endpoint, scraped via the
	// ServiceMonitor in k3s/apps/supervisor/. buildMux registers it and wraps
	// the whole router in the registry's middleware.
	//
	// The whole router, /mcp's bearer included, is built in buildMux so its
	// tests are tests of production wiring (PR #152 review, hyperguild#154).
	// The REST API's credential is the same as /mcp's. ListenAndServe below
	// must serve its result unchanged: that one line is the untested part.
	handler, err := buildMux(restAuth{token: mcpToken, validator: jwtValidator, realm: "brain",
		resourceMetadataURL: resourceMetadataURL}, h, mcpSrv, metrics.New(), public)
	if err != nil {
		logger.Error("build router", "err", err)
		os.Exit(1)
	}
	logger.Info("metrics endpoint registered", "path", "/metrics")

	addr := ":" + port
	watchIntervalLog := "disabled"
	if watchInterval > 0 {
		watchIntervalLog = fmt.Sprintf("%ds", watchInterval)
	}
	logger.Info("ingestion server starting",
		"addr", addr,
		"brain_dir", brainDir,
		"llm_url", llmURL,
		"llm_model", llmModel,
		"chunk_size", chunkSize,
		"watch_interval", watchIntervalLog,
		"mcp_enabled", true,
	)
	if err := http.ListenAndServe(addr, handler); err != nil {
		logger.Error("server stopped", "err", err)
		os.Exit(1)
	}
}