From 0ac35892a56d0addf51f1b60ad84d60ca5e92b2b Mon Sep 17 00:00:00 2001 From: tegwick Date: Fri, 4 Sep 2026 03:11:49 +0200 Subject: [PATCH] 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 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 | 328 +++++++++++++++++++++++ internal/observation/pressure_test.go | 365 ++++++++++++++++++++++++++ internal/observation/registry.go | 326 +++++++++++++++++++++++ 3 files changed, 1019 insertions(+) create mode 100644 internal/observation/classify.go create mode 100644 internal/observation/pressure_test.go create mode 100644 internal/observation/registry.go diff --git a/internal/observation/classify.go b/internal/observation/classify.go new file mode 100644 index 0000000..36e2a40 --- /dev/null +++ b/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 +} diff --git a/internal/observation/pressure_test.go b/internal/observation/pressure_test.go new file mode 100644 index 0000000..96579c9 --- /dev/null +++ b/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 +} diff --git a/internal/observation/registry.go b/internal/observation/registry.go new file mode 100644 index 0000000..fa034fc --- /dev/null +++ b/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 +}