Add telemetry ingest, feedback collector, pressure API and insight CLI

Completes FLUID-WP-0005. Normalization and redaction live on one path,
shared by the in-process emitter and the ingest endpoint: two paths with
two normalizations would eventually disagree, and the disagreement would
surface as a pressure finding that is really a pipeline bug.

Telemetry kind is inferred from event shape rather than defaulting to
"request", since an error filed as a request understates the interface's
failure rate. A malformed event in a batch does not discard the rest.

Feedback is stored as evidence and creates no pressure and no hypothesis
on its own, per API Standards 15, with the consumer recorded as the
actor so their untrusted status stays visible in the audit trail. The
feedback endpoint is the only consumer-reachable part of the control
plane.

The observation endpoints are not served at all when no pseudonymization
salt is configured, rather than served with a generated one: a salt that
changed per run would make the same consumer look new every time and
every cohort count wrong.

Adds an end-to-end test driving real traffic through the gateway and
confirming it becomes a classified pressure record, with no raw consumer
identity reaching the store.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014KmVxhJ35tCo7rE7UnLwWu

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1116572@bnt-lap001
Assistant-Session: 8ba9bb93-a72a-4883-b189-2499cce5c400

cmd/fluid-control/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/tegwick/fluid-core/internal/control"
 	"github.com/tegwick/fluid-core/internal/evidence"
 	"github.com/tegwick/fluid-core/internal/intent"
+	"github.com/tegwick/fluid-core/internal/observation"
 	"github.com/tegwick/fluid-core/internal/policy"
 	"github.com/tegwick/fluid-core/internal/publish"
 	"github.com/tegwick/fluid-core/internal/signing"
@@ -49,6 +50,8 @@ func run() error {
 		keyID     = flag.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier")
 		keyFile   = flag.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 private key file")
 		ephemeral = flag.Bool("ephemeral-key", false, "generate a throwaway signing key (development only)")
+		saltFile  = flag.String("redaction-salt-file", os.Getenv("FLUID_REDACTION_SALT"),
+			"file holding the pseudonymization salt; required for the observation plane")
 	)
 	flag.Parse()
 
@@ -83,9 +86,33 @@ func run() error {
 		return err
 	}
 
+	// The observation plane is optional. Without a salt there is no safe way to
+	// pseudonymize consumer identities, so the endpoints that would record them
+	// are simply not served rather than served unsafely.
+	var pressureAPI *control.PressureAPI
+	if *saltFile != "" {
+		salt, err := os.ReadFile(*saltFile)
+		if err != nil {
+			return fmt.Errorf("read redaction salt: %w", err)
+		}
+		policy := observation.DefaultRedactionPolicy([]byte(trimSpace(string(salt))))
+		ingest, err := observation.NewIngest(ev, contract.InterfaceID(*iface), policy)
+		if err != nil {
+			return err
+		}
+		pressureAPI = control.NewPressureAPI(
+			observation.NewPressureRegistry(ev, contract.InterfaceID(*iface)), ingest)
+	} else {
+		log.Print("no redaction salt configured: telemetry, feedback and pressure endpoints are disabled")
+	}
+
 	srv := &http.Server{
-		Addr:              *addr,
-		Handler:           control.NewServer(control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate)).Routes(),
+		Addr: *addr,
+		Handler: control.NewServer(
+			control.NewRevisionAPI(ev, pipeline),
+			control.NewIntentAPI(intents, gate),
+			pressureAPI,
+		).Routes(),
 		ReadHeaderTimeout: 10 * time.Second,
 	}
 

cmd/fluid/insight.go

@@ -0,0 +1,385 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+	"github.com/tegwick/fluid-core/internal/fitness"
+	"github.com/tegwick/fluid-core/internal/observation"
+)
+
+// ---------- pressure ----------
+
+func runPressure(ctx context.Context, g globals, args []string) error {
+	if len(args) == 0 {
+		return errors.New("pressure needs a subcommand: list, show, analyze, dismiss")
+	}
+	iface, err := g.requireInterface()
+	if err != nil {
+		return err
+	}
+	store, err := g.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer store.Close()
+
+	reg := observation.NewPressureRegistry(store, contract.InterfaceID(iface))
+
+	switch args[0] {
+	case "list":
+		fs := newFlagSet("pressure list")
+		status := fs.String("status", "", "filter by status, such as OPEN or DISMISSED")
+		if err := fs.Parse(args[1:]); err != nil {
+			return err
+		}
+
+		list, err := reg.List(ctx, contract.FluidPressureStatus(*status))
+		if err != nil {
+			return err
+		}
+		if len(list) == 0 {
+			fmt.Println("no pressure recorded")
+			return nil
+		}
+
+		w := out()
+		fmt.Fprintln(w, "ID\tCLASS\tSEV\tCONF\tCONSUMERS\tSTATUS\tSUMMARY")
+		for _, p := range list {
+			consumers := int64(0)
+			if p.Frequency != nil && p.Frequency.IndependentConsumers != nil {
+				consumers = *p.Frequency.IndependentConsumers
+			}
+			fmt.Fprintf(w, "%s\t%s\t%.2f\t%.2f\t%d\t%s\t%s\n",
+				p.ID, p.Class, unitOf(p.Severity), unitOf(p.Confidence),
+				consumers, p.Status, truncate(oneLine(p.Summary), 60))
+		}
+		return w.Flush()
+
+	case "show":
+		if len(args) < 2 {
+			return errors.New("pressure show needs an id")
+		}
+		p, err := reg.Get(ctx, contract.PressureID(args[1]))
+		if err != nil {
+			return err
+		}
+		enc := json.NewEncoder(os.Stdout)
+		enc.SetIndent("", "  ")
+		return enc.Encode(p)
+
+	case "analyze":
+		return analyzePressure(ctx, g, store, reg, args[1:])
+
+	case "dismiss":
+		fs := newFlagSet("pressure dismiss")
+		reason := fs.String("reason", "", "why this pressure will not be acted on")
+		if err := fs.Parse(args[1:]); err != nil {
+			return err
+		}
+		rest := fs.fs.Args()
+		if len(rest) == 0 {
+			return errors.New("pressure dismiss needs an id")
+		}
+		if *reason == "" {
+			return errors.New("dismissal requires --reason; an unexplained dismissal is not auditable")
+		}
+
+		actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()}
+		if err := reg.SetStatus(ctx, contract.PressureID(rest[0]),
+			contract.FluidPressureStatusDISMISSED, actor, *reason); err != nil {
+			return err
+		}
+		fmt.Printf("dismissed %s\n", rest[0])
+		return nil
+
+	default:
+		return fmt.Errorf("unknown pressure subcommand %q", args[0])
+	}
+}
+
+// analyzePressure runs the classifier over recorded telemetry.
+//
+// Analysis is an explicit command rather than something that happens on ingest.
+// Blueprint section 30 wants adaptive work to be budgeted and deferrable, and a
+// classifier that ran on every event would be neither.
+func analyzePressure(ctx context.Context, g globals, store *evidence.SQLStore, reg *observation.PressureRegistry, args []string) error {
+	fs := newFlagSet("pressure analyze")
+	since := fs.String("since", "168h", "how far back to analyze, as a Go duration")
+	dryRun := fs.Bool("dry-run", false, "report findings without recording them")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	window, err := time.ParseDuration(*since)
+	if err != nil {
+		return fmt.Errorf("invalid --since: %w", err)
+	}
+
+	events, err := store.Telemetry(ctx, evidence.TelemetryFilter{
+		InterfaceID: contract.InterfaceID(g.iface),
+		Since:       time.Now().Add(-window),
+	})
+	if err != nil {
+		return err
+	}
+	if len(events) == 0 {
+		fmt.Println("no telemetry in the window; nothing to analyze")
+		return nil
+	}
+
+	classifier := observation.NewClassifier(
+		observation.DefaultClassifierOptions(), observation.NewTopologyAnalyzer())
+	findings := classifier.Classify(events)
+
+	if len(findings) == 0 {
+		fmt.Printf("analyzed %d events; no material pressure found\n", len(events))
+		return nil
+	}
+
+	w := out()
+	fmt.Fprintln(w, "CLASS\tSEV\tCONF\tCONSUMERS\tSUMMARY")
+	for _, f := range findings {
+		fmt.Fprintf(w, "%s\t%.2f\t%.2f\t%d\t%s\n",
+			f.Class, f.Severity, f.Confidence, f.Consumers, truncate(oneLine(f.Summary), 70))
+	}
+	if err := w.Flush(); err != nil {
+		return err
+	}
+
+	if *dryRun {
+		fmt.Printf("\n%d finding(s) from %d events; not recorded (--dry-run)\n", len(findings), len(events))
+		return nil
+	}
+
+	recorded, err := reg.RecordAll(ctx, findings)
+	if err != nil {
+		return err
+	}
+	fmt.Printf("\nrecorded %d pressure record(s) from %d events\n", len(recorded), len(events))
+	return nil
+}
+
+// ---------- cohorts ----------
+
+func runCohort(ctx context.Context, g globals, args []string) error {
+	iface, err := g.requireInterface()
+	if err != nil {
+		return err
+	}
+	store, err := g.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer store.Close()
+
+	fs := newFlagSet("cohort list")
+	since := fs.String("since", "168h", "how far back to summarize")
+	saltFile := fs.String("salt-file", os.Getenv("FLUID_REDACTION_SALT"), "pseudonymization salt file")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	window, err := time.ParseDuration(*since)
+	if err != nil {
+		return fmt.Errorf("invalid --since: %w", err)
+	}
+
+	policy, err := loadPolicy(*saltFile)
+	if err != nil {
+		return err
+	}
+
+	events, err := store.Telemetry(ctx, evidence.TelemetryFilter{
+		InterfaceID: contract.InterfaceID(iface),
+		Since:       time.Now().Add(-window),
+	})
+	if err != nil {
+		return err
+	}
+
+	pops := observation.NewCohortEngine("unclassified", policy).Populations(events)
+	if len(pops) == 0 {
+		fmt.Println("no cohort activity in the window")
+		return nil
+	}
+
+	w := out()
+	fmt.Fprintln(w, "COHORT\tCONSUMERS\tEVENTS")
+	for _, p := range pops {
+		consumers := fmt.Sprint(p.Consumers)
+		if p.Suppressed {
+			// Reporting the exact count of a tiny cohort identifies individuals.
+			consumers = fmt.Sprintf("<%d (suppressed)", policy.CohortMinimumSize)
+		}
+		fmt.Fprintf(w, "%s\t%s\t%d\n", p.Cohort, consumers, p.Events)
+	}
+	return w.Flush()
+}
+
+// ---------- fitness ----------
+
+func runFitness(ctx context.Context, g globals, args []string) error {
+	if len(args) == 0 || args[0] != "compare" {
+		return errors.New("usage: fluid fitness compare --control R-1 --candidate R-2")
+	}
+
+	iface, err := g.requireInterface()
+	if err != nil {
+		return err
+	}
+	store, err := g.open(ctx)
+	if err != nil {
+		return err
+	}
+	defer store.Close()
+
+	fs := newFlagSet("fitness compare")
+	control := fs.String("control", "", "control revision")
+	candidate := fs.String("candidate", "", "candidate revision")
+	since := fs.String("since", "168h", "measurement window")
+	target := fs.Float64("target-requests-per-task", 0, "primary target for requests per completed task")
+	latencyGuard := fs.Float64("guard-p95-latency-ms", 0, "p95 latency guardrail")
+	errorGuard := fs.Float64("guard-error-rate", 0, "error rate guardrail")
+	if err := fs.Parse(args[1:]); err != nil {
+		return err
+	}
+	if *control == "" || *candidate == "" {
+		return errors.New("fitness compare needs --control and --candidate")
+	}
+
+	window, err := time.ParseDuration(*since)
+	if err != nil {
+		return fmt.Errorf("invalid --since: %w", err)
+	}
+	start := time.Now().Add(-window)
+
+	events, err := store.Telemetry(ctx, evidence.TelemetryFilter{
+		InterfaceID: contract.InterfaceID(iface),
+		Since:       start,
+	})
+	if err != nil {
+		return err
+	}
+
+	measureWindow := fitness.Window{Start: start}
+	observations := fitness.NewMeasurer().Measure(events, measureWindow)
+
+	// Specs are declared here rather than inferred from the data, so that the
+	// criteria a comparison was judged against are visible in the command that
+	// ran it.
+	var specs []fitness.MetricSpec
+	if *target > 0 {
+		specs = append(specs, fitness.MetricSpec{
+			Name: fitness.MetricRequestsPerTask, Role: fitness.RolePrimary,
+			Direction: fitness.Lower, Target: target,
+		})
+	}
+	if *latencyGuard > 0 {
+		specs = append(specs, fitness.MetricSpec{
+			Name: fitness.MetricP95LatencyMS, Role: fitness.RoleGuardrail,
+			Direction: fitness.Lower, Threshold: latencyGuard,
+		})
+	}
+	if *errorGuard > 0 {
+		specs = append(specs, fitness.MetricSpec{
+			Name: fitness.MetricErrorRate, Role: fitness.RoleGuardrail,
+			Direction: fitness.Lower, Threshold: errorGuard,
+		})
+	}
+	if len(specs) == 0 {
+		return errors.New("no criteria given: pass at least --target-requests-per-task")
+	}
+
+	eval := fitness.NewEvaluator().Evaluate(
+		contract.RevisionID(*control), contract.RevisionID(*candidate),
+		measureWindow, specs, observations)
+
+	fmt.Printf("Fitness: %s vs %s\nWindow:  since %s\nVerdict: %s\n\n",
+		eval.Control, eval.Candidate, start.Format(time.RFC3339), eval.Verdict)
+
+	w := out()
+	fmt.Fprintln(w, "METRIC\tROLE\tBASELINE\tCURRENT\tDELTA\tOUTCOME")
+	for _, m := range eval.Metrics {
+		outcome := ""
+		switch {
+		case m.Underpowered:
+			outcome = "underpowered"
+		case m.TargetMet != nil && *m.TargetMet:
+			outcome = "target met"
+		case m.TargetMet != nil:
+			outcome = "target missed"
+		case m.GuardrailBreached != nil && *m.GuardrailBreached:
+			outcome = "BREACHED"
+		case m.GuardrailBreached != nil:
+			outcome = "within guardrail"
+		}
+		fmt.Fprintf(w, "%s\t%s\t%.4g\t%.4g\t%+.4g\t%s\n",
+			m.Name, m.Role, m.Baseline, m.Current, m.Delta, outcome)
+	}
+	if err := w.Flush(); err != nil {
+		return err
+	}
+
+	if len(eval.Reasons) > 0 {
+		fmt.Println()
+		for _, r := range eval.Reasons {
+			fmt.Printf("  %s\n", r)
+		}
+	}
+	return nil
+}
+
+// ---------- helpers ----------
+
+// loadPolicy builds a redaction policy from a salt file.
+//
+// It refuses to invent a salt. A generated one would change on every
+// invocation, so the same consumer would look like a new consumer each time and
+// every cohort count would be wrong.
+func loadPolicy(saltFile string) (observation.RedactionPolicy, error) {
+	if saltFile == "" {
+		return observation.RedactionPolicy{}, errors.New(
+			"no redaction salt: pass --salt-file or set FLUID_REDACTION_SALT")
+	}
+	salt, err := os.ReadFile(saltFile)
+	if err != nil {
+		return observation.RedactionPolicy{}, fmt.Errorf("read salt: %w", err)
+	}
+	policy := observation.DefaultRedactionPolicy([]byte(trimSpaceStr(string(salt))))
+	if err := policy.Validate(); err != nil {
+		return observation.RedactionPolicy{}, err
+	}
+	return policy, nil
+}
+
+func trimSpaceStr(s string) string {
+	start, end := 0, len(s)
+	for start < end && (s[start] == ' ' || s[start] == '\n' || s[start] == '\t' || s[start] == '\r') {
+		start++
+	}
+	for end > start && (s[end-1] == ' ' || s[end-1] == '\n' || s[end-1] == '\t' || s[end-1] == '\r') {
+		end--
+	}
+	return s[start:end]
+}
+
+func unitOf(v *contract.UnitInterval) float64 {
+	if v == nil {
+		return 0
+	}
+	return float64(*v)
+}
+
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n-1] + "…"
+}

cmd/fluid/main.go

@@ -30,6 +30,11 @@ Commands:
   revision show    Show one revision descriptor
   policy put       Install a routing policy
   policy show      Show the current routing policy
+  pressure list    List recorded interface pressure
+  pressure analyze Classify recent telemetry into pressure records
+  pressure dismiss Record that a pressure will not be acted on
+  cohort           Summarize cohort populations
+  fitness compare  Compare a candidate revision against its control
   events           Show audit events
   telemetry        Show recorded telemetry
   audit trace      Reconstruct the history behind a revision
@@ -113,6 +118,12 @@ func run(args []string) error {
 		return runEvents(ctx, g, rest)
 	case "telemetry":
 		return runTelemetry(ctx, g, rest)
+	case "pressure":
+		return runPressure(ctx, g, rest)
+	case "cohort":
+		return runCohort(ctx, g, rest)
+	case "fitness":
+		return runFitness(ctx, g, rest)
 	case "audit":
 		return runAudit(ctx, g, rest)
 	default: