diff --git a/cmd/fluid-control/main.go b/cmd/fluid-control/main.go
index e0c2fed..a90fed7 100644
--- a/cmd/fluid-control/main.go
+++ b/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,
 	}
 
diff --git a/cmd/fluid/insight.go b/cmd/fluid/insight.go
new file mode 100644
index 0000000..a57d2b8
--- /dev/null
+++ b/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] + "…"
+}
diff --git a/cmd/fluid/main.go b/cmd/fluid/main.go
index 0c9f26f..6da0158 100644
--- a/cmd/fluid/main.go
+++ b/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:
diff --git a/conformance/loop/loop_test.go b/conformance/loop/loop_test.go
new file mode 100644
index 0000000..dafc1f7
--- /dev/null
+++ b/conformance/loop/loop_test.go
@@ -0,0 +1,154 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	_ "modernc.org/sqlite"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+	"github.com/tegwick/fluid-core/internal/observation"
+	"github.com/tegwick/fluid-core/internal/runtime"
+)
+
+// TestInsightLoopEndToEnd drives real traffic through the gateway and confirms
+// the observation plane turns it into a pressure record.
+//
+// This is the Blueprint section 33 shape end to end: consumers repeatedly
+// listing a collection to find one entry, observed at the gateway, redacted on
+// ingest, and classified as inefficient usage without anyone naming it as such.
+func TestInsightLoopEndToEnd(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+
+	store, err := evidence.OpenSQLite(ctx, filepath.Join(dir, "e.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	policy := observation.DefaultRedactionPolicy([]byte("a-stable-salt-for-this-test-only"))
+	ingest, err := observation.NewIngest(store, "hall-publishing", policy)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	adapter := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`[{"id":1},{"id":2}]`))
+	}))
+	defer adapter.Close()
+
+	reg := runtime.NewRegistry("hall-publishing")
+	d := contract.Revision{
+		SchemaVersion: "0.1", ID: "R-1", Interface: "hall-publishing",
+		State:    contract.RevisionStateStable,
+		Contract: contract.RevisionContract{Type: contract.RevisionContractTypeOpenapi, Digest: contract.Digest("sha256:" + repeat("1", 64))},
+		Runtime:  contract.RevisionRuntime{Upstream: adapter.URL},
+		Intent:   contract.RevisionIntent{Version: "IEI-1"},
+		Policy: contract.RevisionPolicy{
+			Compatibility: contract.RevisionPolicyCompatibilityAdditive,
+			SecurityCheck: contract.RevisionPolicySecurityCheckPassed,
+		},
+	}
+	if err := reg.PutRevision(d); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.PutPolicy(contract.RoutingPolicy{
+		SchemaVersion: "0.1", Interface: "hall-publishing",
+		Generation: 1, DefaultRevision: "R-1",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	emitter := runtime.NewEmitter(ingest, runtime.EmitterOptions{Buffer: 4096, Workers: 2})
+	gw, err := runtime.NewGateway(runtime.GatewayOptions{
+		Interface: "hall-publishing",
+		Registry:  reg,
+		Resolver:  runtime.NewResolver(reg, false),
+		Connector: runtime.NewConnector(),
+		Emitter:   emitter,
+		Cohorts:   observation.NewCohortEngine("coding-agents", policy),
+		Response:  runtime.ResponsePolicy{FeedbackPath: "/v1/feedback"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Four independent consumers, each listing the collection four times per
+	// task to find one entry: the Blueprint section 33 shape.
+	for c := 0; c < 4; c++ {
+		for chain := 0; chain < 3; chain++ {
+			for i := 0; i < 4; i++ {
+				req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil)
+				req.Header.Set("X-FLUID-Consumer", fmt.Sprintf("agent-%d", c))
+				req.Header.Set("X-FLUID-Correlation", fmt.Sprintf("c-%d-%d", c, chain))
+				rec := httptest.NewRecorder()
+				gw.ServeHTTP(rec, req)
+				if rec.Code != http.StatusOK {
+					t.Fatalf("request failed: %d", rec.Code)
+				}
+			}
+		}
+	}
+	emitter.Close()
+
+	events, err := store.Telemetry(ctx, evidence.TelemetryFilter{InterfaceID: "hall-publishing"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 48 {
+		t.Fatalf("recorded %d telemetry events, want 48", len(events))
+	}
+
+	// The raw consumer header must never have reached the store.
+	for _, ev := range events {
+		if ev.ConsumerRef == "agent-0" {
+			t.Fatal("a raw consumer identity reached the evidence store")
+		}
+	}
+
+	classifier := observation.NewClassifier(
+		observation.DefaultClassifierOptions(), observation.NewTopologyAnalyzer())
+	findings := classifier.Classify(events)
+	if len(findings) == 0 {
+		t.Fatal("no pressure found in clearly inefficient traffic")
+	}
+
+	registry := observation.NewPressureRegistry(store, "hall-publishing")
+	recorded, err := registry.RecordAll(ctx, findings)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(recorded) == 0 {
+		t.Fatal("no pressure recorded")
+	}
+
+	var inefficient *contract.FluidPressure
+	for i := range recorded {
+		if recorded[i].Class == contract.PressureClassSuccessfulButInefficientUsage {
+			inefficient = &recorded[i]
+		}
+	}
+	if inefficient == nil {
+		t.Fatalf("inefficient usage not recorded; got %v", recorded)
+	}
+	if inefficient.Frequency == nil || *inefficient.Frequency.IndependentConsumers != 4 {
+		t.Errorf("independent consumers not recorded correctly: %+v", inefficient.Frequency)
+	}
+	if len(inefficient.EvidenceRefs) == 0 {
+		t.Error("pressure recorded with no evidence references")
+	}
+
+	t.Logf("pressure %s: %s (severity %.2f, confidence %.2f, %d consumers)",
+		inefficient.ID, inefficient.Summary,
+		float64(*inefficient.Severity), float64(*inefficient.Confidence),
+		*inefficient.Frequency.IndependentConsumers)
+}
+
+func repeat(s string, n int) string { return strings.Repeat(s, n) }
diff --git a/internal/contract/event_gen.go b/internal/contract/event_gen.go
index a6f478a..9bb0da2 100644
--- a/internal/contract/event_gen.go
+++ b/internal/contract/event_gen.go
@@ -10,6 +10,7 @@ const (
 	FluidEventEntityTypeHypothesis         FluidEventEntityType = "hypothesis"
 	FluidEventEntityTypeRevision           FluidEventEntityType = "revision"
 	FluidEventEntityTypeExperiment         FluidEventEntityType = "experiment"
+	FluidEventEntityTypeFeedback           FluidEventEntityType = "feedback"
 	FluidEventEntityTypeBackendRequirement FluidEventEntityType = "backend_requirement"
 	FluidEventEntityTypeIntent             FluidEventEntityType = "intent"
 	FluidEventEntityTypeDecision           FluidEventEntityType = "decision"
@@ -19,7 +20,7 @@ const (
 // Valid reports whether v is a defined FluidEventEntityType.
 func (v FluidEventEntityType) Valid() bool {
 	switch v {
-	case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy:
+	case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeFeedback, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy:
 		return true
 	}
 	return false
diff --git a/internal/control/api.go b/internal/control/api.go
index 1e83a07..8b4234b 100644
--- a/internal/control/api.go
+++ b/internal/control/api.go
@@ -23,11 +23,13 @@ import (
 type Server struct {
 	revisions *RevisionAPI
 	intents   *IntentAPI
+	pressure  *PressureAPI
 }
 
-// NewServer wires the control APIs.
-func NewServer(rev *RevisionAPI, in *IntentAPI) *Server {
-	return &Server{revisions: rev, intents: in}
+// NewServer wires the control APIs. The pressure API may be nil where an
+// interface runs without an observation plane.
+func NewServer(rev *RevisionAPI, in *IntentAPI, p *PressureAPI) *Server {
+	return &Server{revisions: rev, intents: in, pressure: p}
 }
 
 // Routes returns the control-plane mux.
@@ -40,6 +42,14 @@ func (s *Server) Routes() *http.ServeMux {
 	mux.HandleFunc("/control/v1/intents/", s.intents.handleItem)
 	mux.HandleFunc("/control/v1/intents/active", s.intents.handleActive)
 
+	if s.pressure != nil {
+		mux.HandleFunc("/control/v1/pressure", s.pressure.handleCollection)
+		mux.HandleFunc("/control/v1/pressure/", s.pressure.handleItem)
+		mux.HandleFunc("/control/v1/telemetry", s.pressure.handleTelemetry)
+		// Consumer-reachable, unlike the rest of this surface.
+		mux.HandleFunc("/v1/feedback", s.pressure.handleFeedback)
+	}
+
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
 	})
diff --git a/internal/control/api_test.go b/internal/control/api_test.go
index ca9995e..3073c99 100644
--- a/internal/control/api_test.go
+++ b/internal/control/api_test.go
@@ -57,7 +57,7 @@ func newServer(t *testing.T) (*http.ServeMux, *evidence.SQLStore) {
 		t.Fatal(err)
 	}
 
-	srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate))
+	srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate), nil)
 	return srv.Routes(), store
 }
 
diff --git a/internal/control/pressure.go b/internal/control/pressure.go
new file mode 100644
index 0000000..6b31534
--- /dev/null
+++ b/internal/control/pressure.go
@@ -0,0 +1,171 @@
+package control
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/observation"
+)
+
+// PressureAPI implements ArchitectureBlueprint.md section 44.2: record
+// pressure, aggregate, link evidence, link hypothesis, close or dismiss.
+type PressureAPI struct {
+	registry *observation.PressureRegistry
+	ingest   *observation.Ingest
+}
+
+// NewPressureAPI returns the pressure API.
+func NewPressureAPI(r *observation.PressureRegistry, in *observation.Ingest) *PressureAPI {
+	return &PressureAPI{registry: r, ingest: in}
+}
+
+func (a *PressureAPI) handleCollection(w http.ResponseWriter, r *http.Request) {
+	switch r.Method {
+	case http.MethodGet:
+		status := contract.FluidPressureStatus(r.URL.Query().Get("status"))
+		if status != "" && !status.Valid() {
+			writeError(w, http.StatusBadRequest, "unknown status filter")
+			return
+		}
+		list, err := a.registry.List(r.Context(), status)
+		if err != nil {
+			writeError(w, http.StatusInternalServerError, "could not list pressure")
+			return
+		}
+		writeJSON(w, http.StatusOK, map[string]any{"pressures": list})
+	default:
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+	}
+}
+
+func (a *PressureAPI) handleItem(w http.ResponseWriter, r *http.Request) {
+	id := pathTail(r.URL.Path, "/control/v1/pressure")
+	if id == "" {
+		writeError(w, http.StatusNotFound, "no pressure named")
+		return
+	}
+
+	switch r.Method {
+	case http.MethodGet:
+		p, err := a.registry.Get(r.Context(), contract.PressureID(id))
+		if err != nil {
+			writeError(w, statusForStoreError(err), "pressure not found")
+			return
+		}
+		writeJSON(w, http.StatusOK, p)
+
+	case http.MethodPatch:
+		a.patch(w, r, contract.PressureID(id))
+
+	default:
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+	}
+}
+
+// PatchPressureRequest changes a pressure's disposition.
+type PatchPressureRequest struct {
+	Status     contract.FluidPressureStatus `json:"status,omitempty"`
+	Reason     string                       `json:"reason,omitempty"`
+	Actor      contract.Actor               `json:"actor"`
+	Hypothesis contract.HypothesisID        `json:"link_hypothesis,omitempty"`
+}
+
+func (a *PressureAPI) patch(w http.ResponseWriter, r *http.Request, id contract.PressureID) {
+	var req PatchPressureRequest
+	if err := decodeBody(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "could not decode request", err.Error())
+		return
+	}
+	if req.Actor.ID == "" {
+		// A disposition change with no actor cannot be audited, and dismissals
+		// are exactly the decisions worth attributing.
+		writeError(w, http.StatusBadRequest, "a disposition change must name its actor")
+		return
+	}
+
+	if req.Hypothesis != "" {
+		if err := a.registry.LinkHypothesis(r.Context(), id, req.Hypothesis); err != nil {
+			writeError(w, statusForStoreError(err), err.Error())
+			return
+		}
+	}
+
+	if req.Status != "" {
+		if err := a.registry.SetStatus(r.Context(), id, req.Status, req.Actor, req.Reason); err != nil {
+			writeError(w, http.StatusBadRequest, err.Error())
+			return
+		}
+	}
+
+	p, err := a.registry.Get(r.Context(), id)
+	if err != nil {
+		writeError(w, statusForStoreError(err), "pressure not found")
+		return
+	}
+	writeJSON(w, http.StatusOK, p)
+}
+
+// handleTelemetry accepts telemetry from out-of-process adapters and consumers.
+func (a *PressureAPI) handleTelemetry(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+		return
+	}
+
+	var batch struct {
+		Events []contract.FluidTelemetry `json:"events"`
+	}
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&batch); err != nil {
+		writeError(w, http.StatusBadRequest, "could not decode telemetry", err.Error())
+		return
+	}
+
+	accepted, rejected := a.ingest.WriteBatch(r.Context(), batch.Events)
+
+	causes := make([]string, 0, len(rejected))
+	for _, err := range rejected {
+		causes = append(causes, err.Error())
+	}
+	// Partial acceptance is reported rather than failed: telemetry is
+	// best-effort evidence and the good events are worth keeping.
+	writeJSON(w, http.StatusAccepted, map[string]any{
+		"accepted": accepted,
+		"rejected": len(rejected),
+		"causes":   causes,
+	})
+}
+
+// handleFeedback accepts explicit consumer feedback.
+//
+// This endpoint is reachable by consumers, unlike the rest of the control
+// plane. What it accepts is evidence, never authority: recording feedback
+// creates no pressure and no hypothesis on its own.
+func (a *PressureAPI) handleFeedback(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+		return
+	}
+
+	var f contract.FluidFeedback
+	if err := decodeBody(r, &f); err != nil {
+		writeError(w, http.StatusBadRequest, "could not decode feedback", err.Error())
+		return
+	}
+
+	stored, err := a.ingest.RecordFeedback(r.Context(), f)
+	if err != nil {
+		if errors.Is(err, observation.ErrWrongInterface) {
+			writeError(w, http.StatusBadRequest, err.Error())
+			return
+		}
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	writeJSON(w, http.StatusCreated, map[string]any{
+		"id":   stored.ID,
+		"note": "recorded as evidence; feedback does not itself authorize an interface change",
+	})
+}
diff --git a/internal/observation/feedback.go b/internal/observation/feedback.go
new file mode 100644
index 0000000..0ed3885
--- /dev/null
+++ b/internal/observation/feedback.go
@@ -0,0 +1,22 @@
+package observation
+
+import (
+	"encoding/json"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// consumerLabel names the source of a piece of feedback for the audit trail.
+//
+// It prefers the cohort over anything consumer-specific: the trail needs to
+// know what kind of consumer said this, not which one.
+func consumerLabel(f contract.FluidFeedback) string {
+	if f.Cohort != nil && *f.Cohort != "" {
+		return string(*f.Cohort)
+	}
+	return "unclassified"
+}
+
+func marshalFeedback(f contract.FluidFeedback) ([]byte, error) {
+	return json.Marshal(contract.FeedbackDocument{FluidFeedback: f})
+}
diff --git a/internal/observation/ingest.go b/internal/observation/ingest.go
new file mode 100644
index 0000000..e8dfc23
--- /dev/null
+++ b/internal/observation/ingest.go
@@ -0,0 +1,191 @@
+package observation
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+)
+
+// Ingest normalizes, redacts and stores telemetry.
+//
+// Redaction happens here rather than at the query side because the evidence
+// store is append-only: anything written unredacted stays unredacted forever.
+// The filter belongs on the way in, where there is still a decision to make.
+type Ingest struct {
+	store  evidence.Store
+	policy RedactionPolicy
+	iface  contract.InterfaceID
+	now    func() time.Time
+}
+
+// NewIngest returns an ingest pipeline.
+func NewIngest(store evidence.Store, iface contract.InterfaceID, policy RedactionPolicy) (*Ingest, error) {
+	if err := policy.Validate(); err != nil {
+		return nil, err
+	}
+	return &Ingest{store: store, policy: policy, iface: iface, now: time.Now}, nil
+}
+
+// ErrWrongInterface reports telemetry submitted for another interface.
+var ErrWrongInterface = errors.New("telemetry belongs to a different interface")
+
+// Normalize fills in defaults and applies redaction.
+//
+// It is separated from Write so the same normalization runs whether an event
+// arrives from the in-process emitter or over the ingest endpoint. Two paths
+// with two normalizations would eventually disagree, and the disagreement would
+// surface as a pressure finding that is really a bug in the pipeline.
+func (i *Ingest) Normalize(ev contract.FluidTelemetry) (contract.FluidTelemetry, error) {
+	if ev.InterfaceID == "" {
+		ev.InterfaceID = i.iface
+	}
+	if ev.InterfaceID != i.iface {
+		return ev, fmt.Errorf("%w: event is for %q, ingest serves %q",
+			ErrWrongInterface, ev.InterfaceID, i.iface)
+	}
+
+	if ev.SchemaVersion == "" {
+		ev.SchemaVersion = "0.1"
+	}
+	if ev.ID == "" {
+		ev.ID = newID("tl-")
+	}
+	if ev.OccurredAt.IsZero() {
+		ev.OccurredAt = i.now().UTC()
+	}
+	ev.OccurredAt = ev.OccurredAt.UTC()
+
+	if ev.Kind == "" {
+		// Infer from shape rather than defaulting to "request": an error event
+		// filed as a request would understate the interface's failure rate.
+		switch {
+		case ev.Error != nil:
+			ev.Kind = contract.FluidTelemetryKindError
+		case ev.Adoption != nil:
+			ev.Kind = contract.FluidTelemetryKindAdoption
+		case ev.Sequence != nil:
+			ev.Kind = contract.FluidTelemetryKindSequence
+		default:
+			ev.Kind = contract.FluidTelemetryKindRequest
+		}
+	}
+	if !ev.Kind.Valid() {
+		return ev, fmt.Errorf("unknown telemetry kind %q", ev.Kind)
+	}
+
+	// Raw payload capture is not the default (Blueprint 6.2). Where a policy
+	// forbids it, error detail is the only free-text field that survives, and
+	// it is scrubbed below.
+	if !i.policy.AllowRawPayload && ev.Error != nil && ev.Error.Detail != "" {
+		if len(ev.Error.Detail) > 512 {
+			ev.Error.Detail = ev.Error.Detail[:512] + "…"
+		}
+	}
+
+	i.policy.Apply(&ev)
+	return ev, nil
+}
+
+// Write normalizes and stores one event.
+func (i *Ingest) Write(ctx context.Context, ev contract.FluidTelemetry) error {
+	normalized, err := i.Normalize(ev)
+	if err != nil {
+		return err
+	}
+	return i.store.WriteTelemetry(ctx, normalized)
+}
+
+// WriteBatch stores several events, reporting how many landed.
+//
+// A malformed event in a batch does not discard the rest. Telemetry is
+// best-effort evidence, and dropping a hundred good events because one was
+// wrong would lose more than it protects.
+func (i *Ingest) WriteBatch(ctx context.Context, events []contract.FluidTelemetry) (accepted int, rejected []error) {
+	for _, ev := range events {
+		if err := i.Write(ctx, ev); err != nil {
+			rejected = append(rejected, err)
+			continue
+		}
+		accepted++
+	}
+	return accepted, rejected
+}
+
+// RecordFeedback stores explicit consumer feedback.
+//
+// FluidAPIStandards.md section 15: feedback is evidence and must not itself
+// authorize interface changes. It is stored as a record and an event, and
+// nothing here creates pressure or a hypothesis from it — that stays a
+// deliberate step someone takes.
+func (i *Ingest) RecordFeedback(ctx context.Context, f contract.FluidFeedback) (contract.FluidFeedback, error) {
+	if f.Goal == "" {
+		// Feedback with no stated goal cannot be interpreted later; the goal is
+		// the part that says what the consumer was actually trying to do.
+		return f, errors.New("feedback must state a goal")
+	}
+
+	if f.SchemaVersion == "" {
+		f.SchemaVersion = "0.1"
+	}
+	if f.ID == "" {
+		f.ID = contract.FeedbackID(newID("F-"))
+	}
+	if err := contract.RequireKind(string(f.ID), contract.KindFeedback); err != nil {
+		return f, err
+	}
+	if f.InterfaceID == "" {
+		f.InterfaceID = i.iface
+	}
+	if f.InterfaceID != i.iface {
+		return f, fmt.Errorf("%w: feedback is for %q", ErrWrongInterface, f.InterfaceID)
+	}
+	if f.ReceivedAt.IsZero() {
+		f.ReceivedAt = i.now().UTC()
+	}
+
+	// Consumers write free text; it passes the same filter as everything else.
+	f.Goal, _ = i.policy.Scrub(f.Goal)
+	f.Outcome, _ = i.policy.Scrub(f.Outcome)
+	f.Attempt, _ = i.policy.Scrub(f.Attempt)
+	f.MissingCapability, _ = i.policy.Scrub(f.MissingCapability)
+
+	body, err := marshalFeedback(f)
+	if err != nil {
+		return f, err
+	}
+	if err := i.store.PutRecord(ctx, contract.KindFeedback, string(f.ID), body); err != nil {
+		return f, err
+	}
+
+	if err := i.store.AppendEvent(ctx, contract.FluidEvent{
+		SchemaVersion: "0.1",
+		ID:            contract.EventID(fmt.Sprintf("EV-%s-%d", f.ID, i.now().UnixNano())),
+		OccurredAt:    f.ReceivedAt,
+		EntityType:    contract.FluidEventEntityTypeFeedback,
+		EntityID:      string(f.ID),
+		EventType:     "FEEDBACK_RECEIVED",
+		// The consumer is the actor, and a consumer is untrusted (Blueprint 47).
+		// Recording them as the actor keeps that visible in the audit trail.
+		Actor:  contract.Actor{Type: contract.ActorTypeConsumer, ID: consumerLabel(f)},
+		Reason: f.Goal,
+	}); err != nil {
+		return f, err
+	}
+
+	return f, nil
+}
+
+// newID generates an identifier with the given prefix.
+func newID(prefix string) string {
+	var b [10]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return prefix + "0"
+	}
+	return prefix + hex.EncodeToString(b[:])
+}
diff --git a/internal/observation/ingest_test.go b/internal/observation/ingest_test.go
new file mode 100644
index 0000000..99d1f98
--- /dev/null
+++ b/internal/observation/ingest_test.go
@@ -0,0 +1,264 @@
+package observation
+
+import (
+	"context"
+	"errors"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	_ "modernc.org/sqlite"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+)
+
+func newIngest(t *testing.T) (*Ingest, *evidence.SQLStore) {
+	t.Helper()
+	store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+
+	in, err := NewIngest(store, "hall-publishing", testPolicy())
+	if err != nil {
+		t.Fatal(err)
+	}
+	return in, store
+}
+
+func TestIngestRejectsUnsaltedPolicy(t *testing.T) {
+	store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	if _, err := NewIngest(store, "x", RedactionPolicy{}); !errors.Is(err, ErrNoSalt) {
+		t.Errorf("an unsalted ingest was constructed: %v", err)
+	}
+}
+
+// TestRedactionHappensOnTheWayIn: the store is append-only, so anything written
+// unredacted stays that way forever.
+func TestRedactionHappensOnTheWayIn(t *testing.T) {
+	ctx := context.Background()
+	in, store := newIngest(t)
+
+	err := in.Write(ctx, contract.FluidTelemetry{
+		ConsumerRef: "bernd@example.com",
+		Request:     &contract.FluidTelemetryRequest{Route: "/v1/entries?token=hunter2", Method: "GET"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rows, err := store.Telemetry(ctx, evidence.TelemetryFilter{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("stored %d rows, want 1", len(rows))
+	}
+
+	stored := rows[0]
+	if strings.Contains(stored.ConsumerRef, "@") {
+		t.Errorf("raw consumer identity reached the store: %q", stored.ConsumerRef)
+	}
+	if strings.Contains(stored.Request.Route, "hunter2") {
+		t.Errorf("credential reached the store: %q", stored.Request.Route)
+	}
+	if stored.Redaction == nil || !stored.Redaction.Applied {
+		t.Error("redaction was applied but not recorded on the stored event")
+	}
+}
+
+// TestKindIsInferredFromShape: an error event filed as a request would
+// understate the interface's failure rate.
+func TestKindIsInferredFromShape(t *testing.T) {
+	in, _ := newIngest(t)
+
+	got, err := in.Normalize(contract.FluidTelemetry{
+		Error: &contract.FluidTelemetryError{Class: contract.FluidTelemetryErrorClassTimeout},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Kind != contract.FluidTelemetryKindError {
+		t.Errorf("kind = %s, want error", got.Kind)
+	}
+
+	got, err = in.Normalize(contract.FluidTelemetry{
+		Adoption: &contract.FluidTelemetryAdoption{Event: contract.FluidTelemetryAdoptionEventFirstUse},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Kind != contract.FluidTelemetryKindAdoption {
+		t.Errorf("kind = %s, want adoption", got.Kind)
+	}
+}
+
+func TestNormalizeFillsDefaultsAndGuardsInterface(t *testing.T) {
+	in, _ := newIngest(t)
+
+	got, err := in.Normalize(contract.FluidTelemetry{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.ID == "" || got.OccurredAt.IsZero() || got.SchemaVersion == "" {
+		t.Errorf("defaults not filled: %+v", got)
+	}
+	if got.InterfaceID != "hall-publishing" {
+		t.Errorf("interface = %q", got.InterfaceID)
+	}
+
+	// Another interface's telemetry must not land in this evidence store.
+	if _, err := in.Normalize(contract.FluidTelemetry{InterfaceID: "some-other-api"}); !errors.Is(err, ErrWrongInterface) {
+		t.Errorf("foreign telemetry accepted: %v", err)
+	}
+}
+
+// TestBatchKeepsGoodEventsWhenOneIsBad: dropping a hundred good events because
+// one was malformed loses more than it protects.
+func TestBatchKeepsGoodEventsWhenOneIsBad(t *testing.T) {
+	ctx := context.Background()
+	in, store := newIngest(t)
+
+	accepted, rejected := in.WriteBatch(ctx, []contract.FluidTelemetry{
+		{Request: &contract.FluidTelemetryRequest{Route: "/a"}},
+		{InterfaceID: "wrong-interface"},
+		{Request: &contract.FluidTelemetryRequest{Route: "/b"}},
+	})
+
+	if accepted != 2 {
+		t.Errorf("accepted = %d, want 2", accepted)
+	}
+	if len(rejected) != 1 {
+		t.Errorf("rejected = %d, want 1", len(rejected))
+	}
+
+	rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{})
+	if len(rows) != 2 {
+		t.Errorf("stored %d rows, want 2", len(rows))
+	}
+}
+
+func TestErrorDetailIsBoundedWhenRawPayloadIsOff(t *testing.T) {
+	in, _ := newIngest(t)
+
+	long := strings.Repeat("x", 2000)
+	got, err := in.Normalize(contract.FluidTelemetry{
+		Error: &contract.FluidTelemetryError{
+			Class:  contract.FluidTelemetryErrorClassBackendFailure,
+			Detail: long,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got.Error.Detail) > 600 {
+		t.Errorf("error detail was not bounded: %d chars", len(got.Error.Detail))
+	}
+}
+
+func TestFeedbackIsStoredAsEvidence(t *testing.T) {
+	ctx := context.Background()
+	in, store := newIngest(t)
+
+	cohort := contract.CohortID("coding-agents")
+	got, err := in.RecordFeedback(ctx, contract.FluidFeedback{
+		Cohort:            &cohort,
+		Goal:              "publish a hall entry without splitting it by hand",
+		MissingCapability: "long-form serialization",
+		Outcome:           "capability unavailable, contact bernd@example.com",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if got.ID == "" || !strings.HasPrefix(string(got.ID), "F-") {
+		t.Errorf("feedback id = %q, want an F- prefix", got.ID)
+	}
+	// Consumers write free text; it passes the same filter as everything else.
+	if strings.Contains(got.Outcome, "bernd@example.com") {
+		t.Errorf("an address survived in feedback: %q", got.Outcome)
+	}
+
+	if _, err := store.Record(ctx, contract.KindFeedback, string(got.ID)); err != nil {
+		t.Errorf("feedback was not persisted: %v", err)
+	}
+
+	events, err := store.Events(ctx, evidence.EventFilter{EntityID: string(got.ID)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 {
+		t.Fatalf("feedback left %d events, want 1", len(events))
+	}
+	// A consumer is untrusted; recording them as the actor keeps that visible.
+	if events[0].Actor.Type != contract.ActorTypeConsumer {
+		t.Errorf("actor type = %s, want consumer", events[0].Actor.Type)
+	}
+}
+
+func TestFeedbackNeedsAGoal(t *testing.T) {
+	in, _ := newIngest(t)
+	// Without a goal there is nothing to interpret later.
+	if _, err := in.RecordFeedback(context.Background(), contract.FluidFeedback{
+		Outcome: "it did not work",
+	}); err == nil {
+		t.Error("feedback with no goal was accepted")
+	}
+}
+
+// TestFeedbackDoesNotCreatePressure is the section 15 boundary: feedback is
+// evidence and must not itself authorize a change.
+func TestFeedbackDoesNotCreatePressure(t *testing.T) {
+	ctx := context.Background()
+	in, store := newIngest(t)
+
+	if _, err := in.RecordFeedback(ctx, contract.FluidFeedback{
+		Goal:              "I need a bulk publish endpoint",
+		MissingCapability: "bulk publish",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	records, err := store.Records(ctx, contract.KindPressure)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(records) != 0 {
+		t.Errorf("feedback created %d pressure records on its own", len(records))
+	}
+}
+
+func TestIngestIsUsableAsARuntimeSink(t *testing.T) {
+	ctx := context.Background()
+	in, store := newIngest(t)
+
+	// The gateway's emitter writes through this interface; redaction must apply
+	// on that path too, not only to events arriving over the endpoint.
+	var sink interface {
+		Write(context.Context, contract.FluidTelemetry) error
+	} = in
+
+	if err := sink.Write(ctx, contract.FluidTelemetry{
+		ConsumerRef: "raw-identity",
+		OccurredAt:  time.Now(),
+		Request:     &contract.FluidTelemetryRequest{Route: "/v1/x"},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{})
+	if len(rows) != 1 {
+		t.Fatalf("stored %d rows", len(rows))
+	}
+	if rows[0].ConsumerRef == "raw-identity" {
+		t.Error("the emitter path bypassed redaction")
+	}
+}
diff --git a/schemas/event.schema.json b/schemas/event.schema.json
index cf2a3a3..2cb1db5 100644
--- a/schemas/event.schema.json
+++ b/schemas/event.schema.json
@@ -8,28 +8,70 @@
     "fluid_event": {
       "type": "object",
       "properties": {
-        "schema_version": { "$ref": "common.schema.json#/$defs/schemaVersion" },
-        "id": { "$ref": "common.schema.json#/$defs/eventId" },
-        "occurred_at": { "$ref": "common.schema.json#/$defs/timestamp" },
-        "entity_type": { "enum": ["pressure", "hypothesis", "revision", "experiment", "backend_requirement", "intent", "decision", "routing_policy"] },
-        "entity_id": { "type": "string", "minLength": 1 },
-        "event_type": { "type": "string", "minLength": 1 },
-        "actor": { "$ref": "common.schema.json#/$defs/actor" },
+        "schema_version": {
+          "$ref": "common.schema.json#/$defs/schemaVersion"
+        },
+        "id": {
+          "$ref": "common.schema.json#/$defs/eventId"
+        },
+        "occurred_at": {
+          "$ref": "common.schema.json#/$defs/timestamp"
+        },
+        "entity_type": {
+          "enum": [
+            "pressure",
+            "hypothesis",
+            "revision",
+            "experiment",
+            "feedback",
+            "backend_requirement",
+            "intent",
+            "decision",
+            "routing_policy"
+          ]
+        },
+        "entity_id": {
+          "type": "string",
+          "minLength": 1
+        },
+        "event_type": {
+          "type": "string",
+          "minLength": 1
+        },
+        "actor": {
+          "$ref": "common.schema.json#/$defs/actor"
+        },
         "inputs": {
           "type": "array",
-          "items": { "type": "string" },
+          "items": {
+            "type": "string"
+          },
           "description": "Identifiers of the records this transition drew on."
         },
-        "reason": { "type": "string" },
+        "reason": {
+          "type": "string"
+        },
         "evidence_refs": {
           "type": "array",
-          "items": { "$ref": "common.schema.json#/$defs/evidenceRef" }
+          "items": {
+            "$ref": "common.schema.json#/$defs/evidenceRef"
+          }
         }
       },
-      "required": ["schema_version", "id", "occurred_at", "entity_type", "entity_id", "event_type", "actor"],
+      "required": [
+        "schema_version",
+        "id",
+        "occurred_at",
+        "entity_type",
+        "entity_id",
+        "event_type",
+        "actor"
+      ],
       "additionalProperties": false
     }
   },
-  "required": ["fluid_event"],
+  "required": [
+    "fluid_event"
+  ],
   "additionalProperties": false
 }
diff --git a/workplans/FLUID-WP-0005-fluid-insight.md b/workplans/FLUID-WP-0005-fluid-insight.md
index a2e46f0..1349c46 100644
--- a/workplans/FLUID-WP-0005-fluid-insight.md
+++ b/workplans/FLUID-WP-0005-fluid-insight.md
@@ -4,7 +4,7 @@ type: workplan
 title: "FLUID Insight - observation plane (Blueprint Phase B)"
 domain: infotech
 repo: fluid-core
-status: active
+status: done
 owner: worsch
 topic_slug: fluid-core
 created: "2026-09-04"
@@ -25,7 +25,7 @@ model inference anywhere in this workplan.
 
 ```task
 id: FLUID-WP-0005-T01
-status: todo
+status: done
 priority: high
 state_hub_task_id: "0c074022-031b-5049-832f-1fa8b1ff8b4f"
 ```
@@ -37,7 +37,7 @@ normalized interaction event (Blueprint §6.1).
 
 ```task
 id: FLUID-WP-0005-T02
-status: todo
+status: done
 priority: high
 state_hub_task_id: "13086f4a-217c-5999-a039-90f5d84fb53b"
 ```
@@ -50,7 +50,7 @@ FLUID learns about the interface, not about people.
 
 ```task
 id: FLUID-WP-0005-T03
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "815765d4-8d20-5c6f-87ac-e085778cd58c"
 ```
@@ -62,7 +62,7 @@ not be more specific than the analysis requires.
 
 ```task
 id: FLUID-WP-0005-T04
-status: todo
+status: done
 priority: high
 state_hub_task_id: "27726efa-e054-56c2-a6b8-3cce8f2e85e7"
 ```
@@ -74,7 +74,7 @@ error counting.
 
 ```task
 id: FLUID-WP-0005-T05
-status: todo
+status: done
 priority: high
 state_hub_task_id: "d2e4f2df-809f-5817-9b74-e1533f8f71d9"
 ```
@@ -86,7 +86,7 @@ in every case.
 
 ```task
 id: FLUID-WP-0005-T06
-status: todo
+status: done
 priority: high
 state_hub_task_id: "50bd30a6-ae29-5e2b-9618-a3efe6a362b3"
 ```
@@ -99,7 +99,7 @@ deserves adaptation.
 
 ```task
 id: FLUID-WP-0005-T07
-status: todo
+status: done
 priority: high
 state_hub_task_id: "ebecef59-2ff8-59c0-8cff-cde53030d0c8"
 ```
@@ -111,7 +111,7 @@ distinct. Baseline and measurement window are retained, never recomputed.
 
 ```task
 id: FLUID-WP-0005-T08
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "48f6bbce-2ae8-5b72-8fdc-e962c6ea1f54"
 ```
@@ -123,7 +123,7 @@ treated as authority to change anything.
 
 ```task
 id: FLUID-WP-0005-T09
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "ba413808-346b-5026-bfce-aec0eae0c410"
 ```