Add pressure classifier and pressure registry

FLUID-WP-0005 T05-T06. The classifier is deterministic: its output is
what a hypothesis will later cite as its observation, and a first
implementation that reached for a model could not be audited.

Two mappings are deliberate. An authorization failure is not pressure --
the interface working as designed is not a defect, and classifying it as
one would send the Daimon after a change that must not happen. Backend
failures and timeouts are reported at any rate, while consumer-side
errors must be a meaningful share of traffic first, since a handful says
more about one confused client than about the interface.

Pressure ids are derived from a finding fingerprint rather than
allocated, which makes ingest idempotent: re-analysing a window updates
one record instead of minting another. first_seen is never overwritten,
because how long the interface has had a problem is the evidence that
makes an old pressure worth prioritizing. A dismissed pressure is not
silently reopened, or the operator's decision would mean nothing.

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

internal/observation/classify.go

@@ -0,0 +1,328 @@
+package observation
+
+import (
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Finding is a classified observation, before it becomes a pressure record.
+//
+// A finding carries its evidence and its reasoning. FluidAPIStandards.md
+// section 13 is explicit that pressure is evidence and not truth, so a finding
+// that cannot show its working is not usable: the whole point of separating
+// observation from explanation (schema doc section 18) is that a later reader
+// can disagree with the interpretation while still trusting the observation.
+type Finding struct {
+	Class       contract.PressureClass `json:"class"`
+	Summary     string                 `json:"summary"`
+	Cohorts     []contract.CohortID    `json:"cohorts"`
+	Occurrences int                    `json:"occurrences"`
+	Consumers   int                    `json:"independent_consumers"`
+	Severity    float64                `json:"severity"`
+	Confidence  float64                `json:"confidence"`
+	Evidence    []contract.EvidenceRef `json:"evidence_refs"`
+	FirstSeen   time.Time              `json:"first_seen"`
+	LastSeen    time.Time              `json:"last_seen"`
+	// Fingerprint identifies the same finding across runs, so the registry can
+	// deduplicate rather than accumulating one record per analysis pass.
+	Fingerprint string `json:"fingerprint"`
+}
+
+// ClassifierOptions tunes the deterministic thresholds.
+//
+// They are explicit configuration because Blueprint section 54 lists pressure
+// detection as an open question: which signals produce the best improvement
+// candidates is meant to be learned by operating the system, not fixed now.
+type ClassifierOptions struct {
+	// RepeatThreshold is how many identical calls in one chain count as
+	// compensating for a missing capability rather than ordinary paging.
+	RepeatThreshold int
+	// MinConsumers is the independent-consumer floor for any finding.
+	MinConsumers int
+	// ErrorRateThreshold is the share of a route's calls that must fail before
+	// implementation failure is claimed.
+	ErrorRateThreshold float64
+}
+
+// DefaultClassifierOptions returns workable starting thresholds.
+func DefaultClassifierOptions() ClassifierOptions {
+	return ClassifierOptions{
+		RepeatThreshold:    3,
+		MinConsumers:       2,
+		ErrorRateThreshold: 0.20,
+	}
+}
+
+// Classifier maps evidence onto the ten standard pressure classes.
+//
+// It is deterministic on purpose. FluidAPIStandards.md section 13 allows
+// heuristics, statistics or agentic reasoning here, but a first implementation
+// that reaches for a model cannot be audited, and the classifier's output is
+// what a hypothesis will later cite as its observation.
+type Classifier struct {
+	opts     ClassifierOptions
+	analyzer *TopologyAnalyzer
+}
+
+// NewClassifier returns a classifier.
+func NewClassifier(opts ClassifierOptions, analyzer *TopologyAnalyzer) *Classifier {
+	if analyzer == nil {
+		analyzer = NewTopologyAnalyzer()
+	}
+	return &Classifier{opts: opts, analyzer: analyzer}
+}
+
+// Classify examines telemetry and returns findings.
+func (c *Classifier) Classify(events []contract.FluidTelemetry) []Finding {
+	var findings []Finding
+
+	findings = append(findings, c.fromPatterns(events)...)
+	findings = append(findings, c.fromErrors(events)...)
+
+	// Highest severity first, with a stable tiebreak so output is diffable.
+	sort.Slice(findings, func(i, j int) bool {
+		if findings[i].Severity != findings[j].Severity {
+			return findings[i].Severity > findings[j].Severity
+		}
+		return findings[i].Fingerprint < findings[j].Fingerprint
+	})
+	return findings
+}
+
+// fromPatterns derives findings from interaction topology.
+func (c *Classifier) fromPatterns(events []contract.FluidTelemetry) []Finding {
+	var out []Finding
+
+	for _, p := range c.analyzer.Patterns(events) {
+		if p.Consumers < c.opts.MinConsumers {
+			continue
+		}
+
+		switch {
+		case p.MaxRepeats >= c.opts.RepeatThreshold:
+			// Independent consumers calling one route repeatedly inside a
+			// single task is the classic shape of an interface that makes them
+			// assemble something it could have handed them.
+			out = append(out, Finding{
+				Class: contract.PressureClassSuccessfulButInefficientUsage,
+				Summary: fmt.Sprintf(
+					"%d independent consumers call %s up to %d times within one interaction",
+					p.Consumers, p.RepeatedStep, p.MaxRepeats),
+				Cohorts:     p.Cohorts,
+				Occurrences: p.Count,
+				Consumers:   p.Consumers,
+				Severity:    severityFrom(p.Consumers, p.Count, float64(p.MaxRepeats)/10),
+				Confidence:  confidenceFrom(p.Consumers, p.Count),
+				Evidence:    []contract.EvidenceRef{contract.EvidenceRef("topology:" + p.Signature)},
+				FirstSeen:   p.FirstSeen,
+				LastSeen:    p.LastSeen,
+				Fingerprint: fingerprint("inefficient", p.RepeatedStep),
+			})
+
+		case p.RecoveredError == contract.FluidTelemetryErrorClassValidation,
+			p.RecoveredError == contract.FluidTelemetryErrorClassUnknownField,
+			p.RecoveredError == contract.FluidTelemetryErrorClassUnsupportedParameter:
+			// Consumers who fail, correct themselves, and succeed have
+			// understood the interface eventually. That is a documentation and
+			// discoverability problem, not a capability gap.
+			out = append(out, Finding{
+				Class: contract.PressureClassRecoverableMisunderstanding,
+				Summary: fmt.Sprintf(
+					"%d independent consumers recover from a %s error within the same interaction",
+					p.Consumers, p.RecoveredError),
+				Cohorts:     p.Cohorts,
+				Occurrences: p.Count,
+				Consumers:   p.Consumers,
+				Severity:    severityFrom(p.Consumers, p.Count, 0.1),
+				Confidence:  confidenceFrom(p.Consumers, p.Count),
+				Evidence:    []contract.EvidenceRef{contract.EvidenceRef("topology:" + p.Signature)},
+				FirstSeen:   p.FirstSeen,
+				LastSeen:    p.LastSeen,
+				Fingerprint: fingerprint("recoverable", string(p.RecoveredError)),
+			})
+		}
+	}
+	return out
+}
+
+// routeStats accumulates per-route outcomes.
+type routeStats struct {
+	total     int
+	errors    map[contract.FluidTelemetryErrorClass]int
+	consumers map[string]struct{}
+	cohorts   map[contract.CohortID]struct{}
+	first     time.Time
+	last      time.Time
+}
+
+// fromErrors derives findings from error classes on individual routes.
+func (c *Classifier) fromErrors(events []contract.FluidTelemetry) []Finding {
+	byRoute := map[string]*routeStats{}
+
+	for _, ev := range events {
+		route := "(unknown)"
+		if ev.Request != nil && ev.Request.Route != "" {
+			route = ev.Request.Method + " " + ev.Request.Route
+		}
+
+		s, ok := byRoute[route]
+		if !ok {
+			s = &routeStats{
+				errors:    map[contract.FluidTelemetryErrorClass]int{},
+				consumers: map[string]struct{}{},
+				cohorts:   map[contract.CohortID]struct{}{},
+				first:     ev.OccurredAt,
+				last:      ev.OccurredAt,
+			}
+			byRoute[route] = s
+		}
+
+		s.total++
+		if ev.ConsumerRef != "" {
+			s.consumers[ev.ConsumerRef] = struct{}{}
+		}
+		if ev.Cohort != nil {
+			s.cohorts[*ev.Cohort] = struct{}{}
+		}
+		if ev.OccurredAt.Before(s.first) {
+			s.first = ev.OccurredAt
+		}
+		if ev.OccurredAt.After(s.last) {
+			s.last = ev.OccurredAt
+		}
+		if ev.Error != nil {
+			s.errors[ev.Error.Class]++
+		}
+	}
+
+	routes := make([]string, 0, len(byRoute))
+	for r := range byRoute {
+		routes = append(routes, r)
+	}
+	sort.Strings(routes)
+
+	var out []Finding
+	for _, route := range routes {
+		s := byRoute[route]
+		consumers := len(s.consumers)
+		if consumers < c.opts.MinConsumers {
+			continue
+		}
+
+		classes := make([]contract.FluidTelemetryErrorClass, 0, len(s.errors))
+		for cl := range s.errors {
+			classes = append(classes, cl)
+		}
+		sort.Slice(classes, func(i, j int) bool { return classes[i] < classes[j] })
+
+		for _, class := range classes {
+			count := s.errors[class]
+			rate := float64(count) / float64(s.total)
+
+			pressure, ok := pressureForError(class)
+			if !ok {
+				continue
+			}
+			// Backend failures and timeouts are reported on any recurrence;
+			// consumer-side errors need to be a meaningful share of traffic
+			// before they say something about the interface rather than about
+			// one confused client.
+			if !alwaysReport(class) && rate < c.opts.ErrorRateThreshold {
+				continue
+			}
+
+			cohorts := make([]contract.CohortID, 0, len(s.cohorts))
+			for co := range s.cohorts {
+				cohorts = append(cohorts, co)
+			}
+			sort.Slice(cohorts, func(i, j int) bool { return cohorts[i] < cohorts[j] })
+
+			out = append(out, Finding{
+				Class: pressure,
+				Summary: fmt.Sprintf("%s returns %s for %.0f%% of calls across %d consumers",
+					route, class, rate*100, consumers),
+				Cohorts:     cohorts,
+				Occurrences: count,
+				Consumers:   consumers,
+				Severity:    severityFrom(consumers, count, rate),
+				Confidence:  confidenceFrom(consumers, count),
+				Evidence: []contract.EvidenceRef{
+					contract.EvidenceRef("route:" + route),
+					contract.EvidenceRef("error:" + string(class)),
+				},
+				FirstSeen:   s.first,
+				LastSeen:    s.last,
+				Fingerprint: fingerprint(string(pressure), route+"/"+string(class)),
+			})
+		}
+	}
+	return out
+}
+
+// pressureForError maps a telemetry error class onto a pressure class.
+//
+// Not every error is pressure. An authorization failure usually means the
+// interface worked exactly as designed, so it is left unmapped rather than
+// dressed up as a defect.
+func pressureForError(class contract.FluidTelemetryErrorClass) (contract.PressureClass, bool) {
+	switch class {
+	case contract.FluidTelemetryErrorClassUnknownPath:
+		return contract.PressureClassMissingInterfaceCapability, true
+	case contract.FluidTelemetryErrorClassUnknownField,
+		contract.FluidTelemetryErrorClassUnsupportedParameter:
+		return contract.PressureClassRepeatedExpectationMismatch, true
+	case contract.FluidTelemetryErrorClassValidation:
+		return contract.PressureClassPoorDiscoverability, true
+	case contract.FluidTelemetryErrorClassBackendFailure,
+		contract.FluidTelemetryErrorClassTimeout:
+		return contract.PressureClassImplementationFailure, true
+	case contract.FluidTelemetryErrorClassPolicyRejection:
+		return contract.PressureClassProhibitedDemand, true
+	}
+	return "", false
+}
+
+// alwaysReport marks classes worth surfacing at any rate.
+func alwaysReport(class contract.FluidTelemetryErrorClass) bool {
+	switch class {
+	case contract.FluidTelemetryErrorClassBackendFailure,
+		contract.FluidTelemetryErrorClassTimeout:
+		return true
+	}
+	return false
+}
+
+// severityFrom combines breadth, volume and intensity into a bounded score.
+//
+// The weighting favours breadth: a problem five consumers hit occasionally
+// says more about the interface than one consumer hitting it constantly.
+func severityFrom(consumers, occurrences int, intensity float64) float64 {
+	breadth := saturate(float64(consumers) / 10)
+	volume := saturate(float64(occurrences) / 100)
+	return round2(saturate(0.5*breadth + 0.2*volume + 0.3*saturate(intensity)))
+}
+
+// confidenceFrom expresses how much the evidence supports any conclusion.
+func confidenceFrom(consumers, occurrences int) float64 {
+	return round2(saturate(0.6*saturate(float64(consumers)/5) + 0.4*saturate(float64(occurrences)/50)))
+}
+
+func saturate(v float64) float64 {
+	switch {
+	case v < 0:
+		return 0
+	case v > 1:
+		return 1
+	}
+	return v
+}
+
+func round2(v float64) float64 { return float64(int(v*100+0.5)) / 100 }
+
+// fingerprint identifies a finding stably across analysis runs.
+func fingerprint(kind, subject string) string {
+	return kind + ":" + subject
+}

internal/observation/pressure_test.go

@@ -0,0 +1,365 @@
+package observation
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path/filepath"
+	"testing"
+	"time"
+
+	_ "modernc.org/sqlite"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+)
+
+func newRegistry(t *testing.T) (*PressureRegistry, *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() })
+	return NewPressureRegistry(store, "hall-publishing"), store
+}
+
+// inefficientTraffic reproduces the Blueprint section 33 shape: several
+// consumers listing a collection repeatedly to find one record.
+func inefficientTraffic() []contract.FluidTelemetry {
+	var events []contract.FluidTelemetry
+	for _, consumer := range []string{"c-1", "c-2", "c-3", "c-4"} {
+		for chain := 0; chain < 3; chain++ {
+			start := time.Duration(chain) * time.Hour
+			for i := 0; i < 4; i++ {
+				events = append(events,
+					req(consumer, start+time.Duration(i)*time.Second, "GET", "/customers/{id}/invoices", 200))
+			}
+		}
+	}
+	return events
+}
+
+func TestClassifierFindsInefficientUsage(t *testing.T) {
+	c := NewClassifier(DefaultClassifierOptions(), NewTopologyAnalyzer())
+
+	findings := c.Classify(inefficientTraffic())
+	if len(findings) == 0 {
+		t.Fatal("no findings from clearly inefficient traffic")
+	}
+
+	var found *Finding
+	for i := range findings {
+		if findings[i].Class == contract.PressureClassSuccessfulButInefficientUsage {
+			found = &findings[i]
+		}
+	}
+	if found == nil {
+		t.Fatalf("inefficient usage not classified; got %v", classesOf(findings))
+	}
+	if found.Consumers != 4 {
+		t.Errorf("consumers = %d, want 4", found.Consumers)
+	}
+	if len(found.Evidence) == 0 {
+		t.Error("a finding with no evidence references is not auditable")
+	}
+	if found.Confidence <= 0 || found.Confidence > 1 {
+		t.Errorf("confidence out of range: %v", found.Confidence)
+	}
+}
+
+func TestClassifierMapsErrorsToPressureClasses(t *testing.T) {
+	c := NewClassifier(DefaultClassifierOptions(), NewTopologyAnalyzer())
+
+	var events []contract.FluidTelemetry
+	for i, consumer := range []string{"c-1", "c-2", "c-3"} {
+		for n := 0; n < 5; n++ {
+			ev := errEv(consumer, time.Duration(i*10+n)*time.Minute, "/v1/latest",
+				contract.FluidTelemetryErrorClassUnknownPath)
+			events = append(events, ev)
+		}
+	}
+
+	findings := c.Classify(events)
+	if !hasClass(findings, contract.PressureClassMissingInterfaceCapability) {
+		t.Errorf("repeated unknown-path errors did not yield missing capability; got %v", classesOf(findings))
+	}
+}
+
+// TestAuthorizationFailureIsNotPressure: the interface working as designed is
+// not a defect, and dressing it up as one would send the Daimon chasing a
+// change that must not happen.
+func TestAuthorizationFailureIsNotPressure(t *testing.T) {
+	c := NewClassifier(DefaultClassifierOptions(), NewTopologyAnalyzer())
+
+	var events []contract.FluidTelemetry
+	for i, consumer := range []string{"c-1", "c-2", "c-3"} {
+		for n := 0; n < 10; n++ {
+			events = append(events, errEv(consumer, time.Duration(i*10+n)*time.Minute, "/v1/admin",
+				contract.FluidTelemetryErrorClassAuthorization))
+		}
+	}
+
+	for _, f := range c.Classify(events) {
+		if len(f.Evidence) > 0 && f.Class == contract.PressureClassMissingInterfaceCapability {
+			t.Errorf("authorization failures were classified as pressure: %+v", f)
+		}
+	}
+}
+
+func TestClassifierNeedsIndependentConsumers(t *testing.T) {
+	c := NewClassifier(DefaultClassifierOptions(), NewTopologyAnalyzer())
+
+	var events []contract.FluidTelemetry
+	for n := 0; n < 50; n++ {
+		events = append(events, errEv("only-one", time.Duration(n)*time.Minute, "/v1/x",
+			contract.FluidTelemetryErrorClassUnknownPath))
+	}
+
+	if findings := c.Classify(events); len(findings) != 0 {
+		t.Errorf("one consumer's behaviour became interface pressure: %v", classesOf(findings))
+	}
+}
+
+// TestRecordIsIdempotent is what stops re-analysing a window from minting a new
+// pressure record on every pass.
+func TestRecordIsIdempotent(t *testing.T) {
+	ctx := context.Background()
+	reg, _ := newRegistry(t)
+	c := NewClassifier(DefaultClassifierOptions(), NewTopologyAnalyzer())
+
+	findings := c.Classify(inefficientTraffic())
+	if len(findings) == 0 {
+		t.Fatal("no findings")
+	}
+
+	first, err := reg.RecordAll(ctx, findings)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := reg.RecordAll(ctx, findings)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(first) != len(second) {
+		t.Fatalf("record counts differ between runs: %d then %d", len(first), len(second))
+	}
+	for i := range first {
+		if first[i].ID != second[i].ID {
+			t.Errorf("re-recording minted a new id: %s then %s", first[i].ID, second[i].ID)
+		}
+	}
+
+	all, err := reg.List(ctx, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(all) != len(first) {
+		t.Errorf("registry holds %d records after two identical runs, want %d", len(all), len(first))
+	}
+}
+
+// TestFirstSeenIsNotOverwritten: how long the interface has had a problem is
+// evidence, and it is what makes an old pressure worth prioritizing.
+func TestFirstSeenIsNotOverwritten(t *testing.T) {
+	ctx := context.Background()
+	reg, _ := newRegistry(t)
+
+	old := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	f := Finding{
+		Class:     contract.PressureClassPoorDiscoverability,
+		Summary:   "consumers repeatedly send an unsupported filter",
+		Consumers: 3, Occurrences: 30,
+		Severity: 0.4, Confidence: 0.6,
+		Evidence:    []contract.EvidenceRef{"route:GET /entries"},
+		FirstSeen:   old,
+		LastSeen:    old.Add(time.Hour),
+		Fingerprint: "poor:GET /entries",
+	}
+
+	if _, err := reg.Record(ctx, f); err != nil {
+		t.Fatal(err)
+	}
+
+	later := f
+	later.FirstSeen = old.AddDate(0, 6, 0)
+	later.LastSeen = old.AddDate(0, 6, 1)
+	got, err := reg.Record(ctx, later)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !got.FirstSeen.Equal(old) {
+		t.Errorf("first_seen was overwritten: %s, want %s", got.FirstSeen, old)
+	}
+	if !got.LastSeen.Equal(later.LastSeen) {
+		t.Errorf("last_seen was not extended: %s", got.LastSeen)
+	}
+}
+
+// TestDismissedPressureStaysDismissed: an operator's decision that something is
+// not worth acting on must survive the next analysis run.
+func TestDismissedPressureStaysDismissed(t *testing.T) {
+	ctx := context.Background()
+	reg, _ := newRegistry(t)
+
+	f := Finding{
+		Class:     contract.PressureClassOutOfScopeDemand,
+		Summary:   "consumers ask for a capability outside the interface boundary",
+		Consumers: 5, Occurrences: 40,
+		Severity: 0.5, Confidence: 0.7,
+		Evidence:    []contract.EvidenceRef{"route:GET /billing"},
+		FirstSeen:   base,
+		LastSeen:    base.Add(time.Hour),
+		Fingerprint: "out-of-scope:GET /billing",
+	}
+
+	p, err := reg.Record(ctx, f)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	actor := contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}
+	if err := reg.SetStatus(ctx, p.ID, contract.FluidPressureStatusDISMISSED, actor,
+		"billing belongs to the payments interface"); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := reg.Record(ctx, f); !errors.Is(err, ErrDismissed) {
+		t.Errorf("a dismissed pressure was silently reopened: %v", err)
+	}
+
+	// A batch run must skip it rather than fail.
+	if _, err := reg.RecordAll(ctx, []Finding{f}); err != nil {
+		t.Errorf("RecordAll failed on a dismissed pressure: %v", err)
+	}
+
+	got, err := reg.Get(ctx, p.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != contract.FluidPressureStatusDISMISSED {
+		t.Errorf("status = %s, want DISMISSED", got.Status)
+	}
+}
+
+func TestStatusChangeRequiresAReason(t *testing.T) {
+	ctx := context.Background()
+	reg, _ := newRegistry(t)
+
+	p, err := reg.Record(ctx, Finding{
+		Class: contract.PressureClassPoorDiscoverability, Summary: "s",
+		Consumers: 2, Occurrences: 5, Evidence: []contract.EvidenceRef{"route:x"},
+		FirstSeen: base, LastSeen: base, Fingerprint: "fp",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	actor := contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"}
+	if err := reg.SetStatus(ctx, p.ID, contract.FluidPressureStatusDISMISSED, actor, ""); err == nil {
+		t.Error("a dismissal with no reason was accepted")
+	}
+}
+
+func TestLinkHypothesisMovesToAnalyzing(t *testing.T) {
+	ctx := context.Background()
+	reg, store := newRegistry(t)
+
+	p, err := reg.Record(ctx, Finding{
+		Class: contract.PressureClassSuccessfulButInefficientUsage, Summary: "s",
+		Consumers: 3, Occurrences: 12, Evidence: []contract.EvidenceRef{"route:x"},
+		FirstSeen: base, LastSeen: base, Fingerprint: "fp",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := reg.LinkHypothesis(ctx, p.ID, "H-1"); err != nil {
+		t.Fatal(err)
+	}
+	// Linking must reject a mis-prefixed id rather than store nonsense.
+	if err := reg.LinkHypothesis(ctx, p.ID, "R-1"); err == nil {
+		t.Error("a revision id was accepted as a hypothesis link")
+	}
+
+	got, err := reg.Get(ctx, p.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != contract.FluidPressureStatusANALYZING {
+		t.Errorf("status = %s, want ANALYZING", got.Status)
+	}
+	if len(got.LinkedHypotheses) != 1 {
+		t.Errorf("linked hypotheses = %v", got.LinkedHypotheses)
+	}
+
+	// Linking twice must not duplicate.
+	if err := reg.LinkHypothesis(ctx, p.ID, "H-1"); err != nil {
+		t.Fatal(err)
+	}
+	got, _ = reg.Get(ctx, p.ID)
+	if len(got.LinkedHypotheses) != 1 {
+		t.Errorf("duplicate link recorded: %v", got.LinkedHypotheses)
+	}
+
+	events, err := store.Events(ctx, evidence.EventFilter{EntityID: string(p.ID)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) < 2 {
+		t.Errorf("lifecycle transitions left too few events: %d", len(events))
+	}
+}
+
+func TestListOrdersBySeverity(t *testing.T) {
+	ctx := context.Background()
+	reg, _ := newRegistry(t)
+
+	for i, sev := range []float64{0.2, 0.9, 0.5} {
+		_, err := reg.Record(ctx, Finding{
+			Class:     contract.PressureClassPoorDiscoverability,
+			Summary:   fmt.Sprintf("finding %d", i),
+			Consumers: 3, Occurrences: 10,
+			Severity: sev, Confidence: 0.5,
+			Evidence:    []contract.EvidenceRef{contract.EvidenceRef(fmt.Sprintf("route:%d", i))},
+			FirstSeen:   base,
+			LastSeen:    base,
+			Fingerprint: fmt.Sprintf("fp-%d", i),
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	got, err := reg.List(ctx, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 3 {
+		t.Fatalf("got %d records", len(got))
+	}
+	for i := 1; i < len(got); i++ {
+		if unit(got[i-1].Severity) < unit(got[i].Severity) {
+			t.Error("records are not ordered by severity")
+		}
+	}
+}
+
+func classesOf(fs []Finding) []contract.PressureClass {
+	out := make([]contract.PressureClass, len(fs))
+	for i, f := range fs {
+		out[i] = f.Class
+	}
+	return out
+}
+
+func hasClass(fs []Finding, want contract.PressureClass) bool {
+	for _, f := range fs {
+		if f.Class == want {
+			return true
+		}
+	}
+	return false
+}

internal/observation/registry.go

@@ -0,0 +1,326 @@
+package observation
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+	"github.com/tegwick/fluid-core/internal/evidence"
+)
+
+// PressureRegistry is the durable inventory of material interface pressure.
+//
+// ArchitectureBlueprint.md section 9 requires deduplication, aggregation,
+// cohort segmentation, frequency tracking, severity, confidence, linked
+// hypotheses and disposition — and says plainly that pressure may remain
+// unresolved on purpose. Not every observed mismatch deserves adaptation.
+type PressureRegistry struct {
+	store evidence.Store
+	iface contract.InterfaceID
+	now   func() time.Time
+}
+
+// NewPressureRegistry returns a registry backed by the evidence store.
+func NewPressureRegistry(store evidence.Store, iface contract.InterfaceID) *PressureRegistry {
+	return &PressureRegistry{store: store, iface: iface, now: time.Now}
+}
+
+// ErrDismissed reports an attempt to reopen a deliberately closed pressure.
+var ErrDismissed = errors.New("pressure was dismissed and will not be reopened automatically")
+
+// pressureID derives a stable identifier from a finding's fingerprint.
+//
+// Deriving rather than allocating is what makes ingest idempotent: re-running
+// analysis over the same window updates one record instead of minting a new one
+// on every pass.
+func pressureID(iface contract.InterfaceID, fingerprint string) contract.PressureID {
+	sum := sha256.Sum256([]byte(string(iface) + "\x00" + fingerprint))
+	return contract.PressureID("P-" + hex.EncodeToString(sum[:6]))
+}
+
+// Record folds a finding into the registry, creating or updating one pressure.
+//
+// A finding that matches an existing record extends its window and refreshes
+// its counts rather than replacing it: first_seen is evidence about how long
+// the interface has had this problem, and overwriting it would erase the age
+// that makes a pressure worth prioritizing.
+func (r *PressureRegistry) Record(ctx context.Context, f Finding) (contract.FluidPressure, error) {
+	id := pressureID(r.iface, f.Fingerprint)
+
+	existing, err := r.Get(ctx, id)
+	switch {
+	case err == nil:
+		if existing.Status == contract.FluidPressureStatusDISMISSED {
+			// An operator decided this is not worth acting on. Silently
+			// resurrecting it would make the dismissal meaningless.
+			return existing, fmt.Errorf("%w: %s", ErrDismissed, id)
+		}
+		return r.update(ctx, existing, f)
+	case errors.Is(err, evidence.ErrNotFound):
+		return r.create(ctx, id, f)
+	default:
+		return contract.FluidPressure{}, err
+	}
+}
+
+func (r *PressureRegistry) create(ctx context.Context, id contract.PressureID, f Finding) (contract.FluidPressure, error) {
+	observations := int64(f.Occurrences)
+	consumers := int64(f.Consumers)
+	severity := contract.UnitInterval(f.Severity)
+	confidence := contract.UnitInterval(f.Confidence)
+
+	p := contract.FluidPressure{
+		SchemaVersion:   "0.1",
+		ID:              id,
+		InterfaceID:     r.iface,
+		Class:           f.Class,
+		FirstSeen:       f.FirstSeen,
+		LastSeen:        f.LastSeen,
+		AffectedCohorts: f.Cohorts,
+		Frequency: &contract.FluidPressureFrequency{
+			Observations:         &observations,
+			IndependentConsumers: &consumers,
+		},
+		Severity:     &severity,
+		Confidence:   &confidence,
+		Summary:      f.Summary,
+		EvidenceRefs: f.Evidence,
+		Status:       contract.FluidPressureStatusOPEN,
+	}
+
+	if err := r.put(ctx, p); err != nil {
+		return contract.FluidPressure{}, err
+	}
+	if err := r.event(ctx, p, "PRESSURE_OPENED", f.Summary); err != nil {
+		return contract.FluidPressure{}, err
+	}
+	return p, nil
+}
+
+func (r *PressureRegistry) update(ctx context.Context, p contract.FluidPressure, f Finding) (contract.FluidPressure, error) {
+	if f.FirstSeen.Before(p.FirstSeen) {
+		p.FirstSeen = f.FirstSeen
+	}
+	if f.LastSeen.After(p.LastSeen) {
+		p.LastSeen = f.LastSeen
+	}
+
+	observations := int64(f.Occurrences)
+	consumers := int64(f.Consumers)
+	p.Frequency = &contract.FluidPressureFrequency{
+		Observations:         &observations,
+		IndependentConsumers: &consumers,
+	}
+
+	severity := contract.UnitInterval(f.Severity)
+	confidence := contract.UnitInterval(f.Confidence)
+	p.Severity = &severity
+	p.Confidence = &confidence
+	p.Summary = f.Summary
+	p.AffectedCohorts = mergeCohorts(p.AffectedCohorts, f.Cohorts)
+	p.EvidenceRefs = mergeRefs(p.EvidenceRefs, f.Evidence)
+
+	// Fresh evidence for something previously explained means it is back.
+	if p.Status == contract.FluidPressureStatusADDRESSED {
+		p.Status = contract.FluidPressureStatusOPEN
+		if err := r.put(ctx, p); err != nil {
+			return contract.FluidPressure{}, err
+		}
+		return p, r.event(ctx, p, "PRESSURE_REOPENED",
+			"new evidence observed after the pressure was marked addressed")
+	}
+
+	if err := r.put(ctx, p); err != nil {
+		return contract.FluidPressure{}, err
+	}
+	return p, r.event(ctx, p, "PRESSURE_OBSERVED", f.Summary)
+}
+
+// RecordAll folds a batch of findings into the registry.
+//
+// Dismissed pressures are skipped rather than treated as errors: hitting one
+// is the expected outcome of re-analysing a window an operator has already
+// triaged.
+func (r *PressureRegistry) RecordAll(ctx context.Context, findings []Finding) ([]contract.FluidPressure, error) {
+	var out []contract.FluidPressure
+	for _, f := range findings {
+		p, err := r.Record(ctx, f)
+		if errors.Is(err, ErrDismissed) {
+			continue
+		}
+		if err != nil {
+			return out, err
+		}
+		out = append(out, p)
+	}
+	return out, nil
+}
+
+// Get returns one pressure record.
+func (r *PressureRegistry) Get(ctx context.Context, id contract.PressureID) (contract.FluidPressure, error) {
+	body, err := r.store.Record(ctx, contract.KindPressure, string(id))
+	if err != nil {
+		return contract.FluidPressure{}, err
+	}
+	var doc contract.PressureDocument
+	if err := json.Unmarshal(body, &doc); err != nil {
+		return contract.FluidPressure{}, fmt.Errorf("decode pressure %s: %w", id, err)
+	}
+	return doc.FluidPressure, nil
+}
+
+// List returns pressures, optionally filtered by status.
+func (r *PressureRegistry) List(ctx context.Context, status contract.FluidPressureStatus) ([]contract.FluidPressure, error) {
+	records, err := r.store.Records(ctx, contract.KindPressure)
+	if err != nil {
+		return nil, err
+	}
+
+	out := make([]contract.FluidPressure, 0, len(records))
+	for _, body := range records {
+		var doc contract.PressureDocument
+		if err := json.Unmarshal(body, &doc); err != nil {
+			continue
+		}
+		if status != "" && doc.FluidPressure.Status != status {
+			continue
+		}
+		out = append(out, doc.FluidPressure)
+	}
+
+	// Most severe first: the registry is a work queue as much as an inventory.
+	sort.Slice(out, func(i, j int) bool {
+		si, sj := unit(out[i].Severity), unit(out[j].Severity)
+		if si != sj {
+			return si > sj
+		}
+		return out[i].ID < out[j].ID
+	})
+	return out, nil
+}
+
+// SetStatus moves a pressure through its lifecycle.
+//
+// Transitions are recorded as events rather than only as a field, so a
+// dismissal can be traced to whoever made it and why.
+func (r *PressureRegistry) SetStatus(ctx context.Context, id contract.PressureID, status contract.FluidPressureStatus, actor contract.Actor, reason string) error {
+	if !status.Valid() {
+		return fmt.Errorf("unknown pressure status %q", status)
+	}
+	if reason == "" {
+		// A status change with no reason is not auditable, and dismissals
+		// without a reason are how a registry quietly loses its evidence.
+		return errors.New("a status change requires a reason")
+	}
+
+	p, err := r.Get(ctx, id)
+	if err != nil {
+		return err
+	}
+	previous := p.Status
+	p.Status = status
+
+	if err := r.put(ctx, p); err != nil {
+		return err
+	}
+	return r.eventBy(ctx, p, "PRESSURE_STATUS_CHANGED", actor,
+		fmt.Sprintf("%s -> %s: %s", previous, status, reason))
+}
+
+// LinkHypothesis records that a hypothesis addresses this pressure.
+func (r *PressureRegistry) LinkHypothesis(ctx context.Context, id contract.PressureID, h contract.HypothesisID) error {
+	if err := contract.RequireKind(string(h), contract.KindHypothesis); err != nil {
+		return err
+	}
+
+	p, err := r.Get(ctx, id)
+	if err != nil {
+		return err
+	}
+	for _, existing := range p.LinkedHypotheses {
+		if existing == h {
+			return nil
+		}
+	}
+	p.LinkedHypotheses = append(p.LinkedHypotheses, h)
+	if p.Status == contract.FluidPressureStatusOPEN {
+		p.Status = contract.FluidPressureStatusANALYZING
+	}
+
+	if err := r.put(ctx, p); err != nil {
+		return err
+	}
+	return r.event(ctx, p, "PRESSURE_HYPOTHESIS_LINKED", fmt.Sprintf("linked %s", h))
+}
+
+func (r *PressureRegistry) put(ctx context.Context, p contract.FluidPressure) error {
+	body, err := json.Marshal(contract.PressureDocument{FluidPressure: p})
+	if err != nil {
+		return err
+	}
+	return r.store.PutRecord(ctx, contract.KindPressure, string(p.ID), body)
+}
+
+func (r *PressureRegistry) event(ctx context.Context, p contract.FluidPressure, kind, reason string) error {
+	return r.eventBy(ctx, p, kind,
+		contract.Actor{Type: contract.ActorTypeSystem, ID: "fluid-pressure-engine"}, reason)
+}
+
+func (r *PressureRegistry) eventBy(ctx context.Context, p contract.FluidPressure, kind string, actor contract.Actor, reason string) error {
+	return r.store.AppendEvent(ctx, contract.FluidEvent{
+		SchemaVersion: "0.1",
+		ID:            contract.EventID(fmt.Sprintf("EV-%s-%d", p.ID, r.now().UnixNano())),
+		OccurredAt:    r.now().UTC(),
+		EntityType:    contract.FluidEventEntityTypePressure,
+		EntityID:      string(p.ID),
+		EventType:     kind,
+		Actor:         actor,
+		Reason:        reason,
+		EvidenceRefs:  p.EvidenceRefs,
+	})
+}
+
+func unit(v *contract.UnitInterval) float64 {
+	if v == nil {
+		return 0
+	}
+	return float64(*v)
+}
+
+func mergeCohorts(a, b []contract.CohortID) []contract.CohortID {
+	seen := map[contract.CohortID]struct{}{}
+	var out []contract.CohortID
+	for _, list := range [][]contract.CohortID{a, b} {
+		for _, c := range list {
+			if _, ok := seen[c]; ok {
+				continue
+			}
+			seen[c] = struct{}{}
+			out = append(out, c)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
+	return out
+}
+
+func mergeRefs(a, b []contract.EvidenceRef) []contract.EvidenceRef {
+	seen := map[contract.EvidenceRef]struct{}{}
+	var out []contract.EvidenceRef
+	for _, list := range [][]contract.EvidenceRef{a, b} {
+		for _, s := range list {
+			if _, ok := seen[s]; ok {
+				continue
+			}
+			seen[s] = struct{}{}
+			out = append(out, s)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
+	return out
+}