Add redaction filter, cohort engine and topology analyzer

FLUID-WP-0005 T02-T04. Redaction is deterministic configuration rather
than a heuristic: the evidence store is append-only, so anything it
accepts cannot be taken back out. Consumer identities are pseudonymized
with HMAC rather than a bare hash, since a plain digest of a short
identifier is reversible by enumeration.

Applied rules are recorded on each event, so later analysis knows what
it cannot see instead of mistaking an absence of evidence for evidence
of absence. Redacted query parameters keep their key: which parameters a
consumer sent is itself interface evidence.

The topology analyzer requires several independent consumers before
calling a shape a pattern. One client looping is a client bug; several
converging on the same workaround is interface pressure. It also
distinguishes a recovered error from a hard failure, because those two
deserve different hypotheses.

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/cohort.go

@@ -0,0 +1,148 @@
+package observation
+
+import (
+	"net/http"
+	"sort"
+	"strings"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// CohortRule assigns a request to a cohort when every stated condition holds.
+//
+// Rules are deterministic and declarative rather than learned. Blueprint 6.3
+// wants cohorts stable enough to compare over time, and a classifier that
+// drifts makes last month's measurement incomparable with this month's.
+type CohortRule struct {
+	// Cohort is the assignment this rule produces.
+	Cohort contract.CohortID
+
+	// Header matches a header value exactly, when both are set.
+	Header      string
+	HeaderValue string
+
+	// HeaderPrefix matches a header by prefix, for SDK version families.
+	HeaderPrefix string
+
+	// PathPrefix matches the request path.
+	PathPrefix string
+
+	// Description explains the population, for the operator reading a report.
+	Description string
+}
+
+func (r CohortRule) matches(req *http.Request) bool {
+	if r.Header != "" {
+		got := req.Header.Get(r.Header)
+		switch {
+		case r.HeaderValue != "":
+			if !strings.EqualFold(got, r.HeaderValue) {
+				return false
+			}
+		case r.HeaderPrefix != "":
+			if !strings.HasPrefix(strings.ToLower(got), strings.ToLower(r.HeaderPrefix)) {
+				return false
+			}
+		default:
+			if got == "" {
+				return false
+			}
+		}
+	}
+	if r.PathPrefix != "" && !strings.HasPrefix(req.URL.Path, r.PathPrefix) {
+		return false
+	}
+	return true
+}
+
+// CohortEngine groups consumers into analytically useful populations.
+type CohortEngine struct {
+	rules    []CohortRule
+	fallback contract.CohortID
+	policy   RedactionPolicy
+}
+
+// NewCohortEngine returns an engine. Rules are evaluated in order, first match
+// wins, so ordering is how an operator expresses precedence.
+func NewCohortEngine(fallback contract.CohortID, policy RedactionPolicy, rules ...CohortRule) *CohortEngine {
+	return &CohortEngine{rules: rules, fallback: fallback, policy: policy}
+}
+
+// Cohort implements the runtime's CohortResolver.
+//
+// It returns the pseudonymous consumer reference alongside the cohort, so the
+// identity never reaches the data plane in raw form: redaction happens at
+// assignment rather than later in the pipeline, where an intervening component
+// could have logged it.
+func (e *CohortEngine) Cohort(r *http.Request) (contract.CohortID, string) {
+	consumer := e.policy.Pseudonymize(consumerIdentity(r))
+
+	for _, rule := range e.rules {
+		if rule.matches(r) {
+			return rule.Cohort, consumer
+		}
+	}
+	return e.fallback, consumer
+}
+
+// consumerIdentity extracts the raw identity a request claims.
+func consumerIdentity(r *http.Request) string {
+	for _, header := range []string{"X-FLUID-Consumer", "X-Consumer-ID"} {
+		if v := r.Header.Get(header); v != "" {
+			return v
+		}
+	}
+	return ""
+}
+
+// Describe lists the configured cohorts, for operator display.
+func (e *CohortEngine) Describe() []CohortRule {
+	out := make([]CohortRule, len(e.rules))
+	copy(out, e.rules)
+	sort.Slice(out, func(i, j int) bool { return out[i].Cohort < out[j].Cohort })
+	return out
+}
+
+// Population counts distinct consumers per cohort over a set of events.
+//
+// Counts below the policy's minimum are reported as suppressed rather than as
+// a number, so a report cannot accidentally single out an individual.
+type Population struct {
+	Cohort     contract.CohortID `json:"cohort"`
+	Consumers  int               `json:"consumers"`
+	Events     int               `json:"events"`
+	Suppressed bool              `json:"suppressed"`
+}
+
+// Populations summarizes cohort sizes across events.
+func (e *CohortEngine) Populations(events []contract.FluidTelemetry) []Population {
+	consumers := map[contract.CohortID]map[string]struct{}{}
+	counts := map[contract.CohortID]int{}
+
+	for _, ev := range events {
+		if ev.Cohort == nil {
+			continue
+		}
+		c := *ev.Cohort
+		counts[c]++
+		if consumers[c] == nil {
+			consumers[c] = map[string]struct{}{}
+		}
+		if ev.ConsumerRef != "" {
+			consumers[c][ev.ConsumerRef] = struct{}{}
+		}
+	}
+
+	out := make([]Population, 0, len(counts))
+	for c, n := range counts {
+		distinct := len(consumers[c])
+		out = append(out, Population{
+			Cohort:     c,
+			Consumers:  distinct,
+			Events:     n,
+			Suppressed: e.policy.SuppressSmallCohort(distinct),
+		})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Cohort < out[j].Cohort })
+	return out
+}

internal/observation/redact.go

@@ -0,0 +1,235 @@
+// Package observation implements the FLUID observation plane: telemetry
+// normalization, redaction, cohorts, interaction topology, and interface
+// pressure classification.
+//
+// ArchitectureBlueprint.md section 6.2 sets the boundary this package works
+// inside: telemetry should be designed for interface learning without becoming
+// an unrestricted behavioural capture layer. Raw payload capture is never the
+// default, and semantic learning relies on minimized evidence where it can.
+package observation
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"net/url"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// RedactionPolicy governs what may reach the evidence store.
+//
+// It is deterministic configuration, not a heuristic. A privacy filter that
+// guesses is a privacy filter that will one day guess wrong, and the evidence
+// store is append-only: anything it accepts cannot be taken back out.
+type RedactionPolicy struct {
+	// Salt keys the pseudonymization of consumer identities. It must be stable
+	// for the life of the interface: a rotated salt breaks every longitudinal
+	// comparison, because the same consumer starts looking like a new one.
+	Salt []byte
+
+	// AllowRawPayload permits request and response bodies into telemetry.
+	// Off by default, and Blueprint 6.2 says it should stay that way.
+	AllowRawPayload bool
+
+	// DropQueryParams removes named query parameters from recorded routes.
+	DropQueryParams []string
+
+	// DropHeaders removes named headers from recorded evidence.
+	DropHeaders []string
+
+	// SensitivePatterns match values that must never be stored, wherever they
+	// appear. Anything matching is replaced rather than dropped, so the shape
+	// of the evidence survives while the content does not.
+	SensitivePatterns []*regexp.Regexp
+
+	// RetentionDays bounds how long telemetry is kept. Zero means unbounded,
+	// which should be a deliberate choice rather than an oversight.
+	RetentionDays int
+
+	// CohortMinimumSize is the smallest population that may be reported
+	// separately. Below it, a "cohort" identifies individuals.
+	CohortMinimumSize int
+}
+
+// DefaultRedactionPolicy returns a conservative policy.
+//
+// The defaults assume the interface handles something worth protecting. An
+// operator who knows otherwise can loosen them explicitly; an operator who has
+// not thought about it gets the safe behaviour.
+func DefaultRedactionPolicy(salt []byte) RedactionPolicy {
+	return RedactionPolicy{
+		Salt:            salt,
+		AllowRawPayload: false,
+		DropQueryParams: []string{"token", "api_key", "apikey", "access_token", "signature", "password"},
+		DropHeaders:     []string{"authorization", "cookie", "set-cookie", "proxy-authorization", "x-api-key"},
+		SensitivePatterns: []*regexp.Regexp{
+			// Bearer tokens and basic credentials appearing in free text.
+			regexp.MustCompile(`(?i)bearer\s+[A-Za-z0-9._~+/-]+=*`),
+			// Anything that looks like an email address.
+			regexp.MustCompile(`[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}`),
+			// Connection strings with embedded credentials.
+			regexp.MustCompile(`[a-z][a-z0-9+.-]*://[^\s:@/]+:[^\s@/]+@`),
+		},
+		RetentionDays:     90,
+		CohortMinimumSize: 5,
+	}
+}
+
+// ErrNoSalt reports a policy that would pseudonymize with an empty key.
+var ErrNoSalt = errors.New("redaction policy has no salt; consumer identities would be trivially reversible")
+
+// Validate checks a policy is usable.
+func (p RedactionPolicy) Validate() error {
+	if len(p.Salt) < 16 {
+		return ErrNoSalt
+	}
+	return nil
+}
+
+// Pseudonymize maps a consumer identity to a stable opaque reference.
+//
+// HMAC rather than a plain hash: a bare SHA-256 of a short identifier — an
+// account id, an email — is reversible by anyone willing to enumerate the
+// input space, which for most identifier schemes is cheap.
+func (p RedactionPolicy) Pseudonymize(identity string) string {
+	if identity == "" {
+		return ""
+	}
+	mac := hmac.New(sha256.New, p.Salt)
+	_, _ = mac.Write([]byte(identity))
+	// Twelve bytes is ample to keep collisions negligible at interface scale
+	// while keeping the value short enough to read in a terminal.
+	return "psu-" + hex.EncodeToString(mac.Sum(nil)[:12])
+}
+
+// Scrub removes sensitive substrings from free text.
+func (p RedactionPolicy) Scrub(s string) (string, bool) {
+	redacted := false
+	for _, pattern := range p.SensitivePatterns {
+		if pattern.MatchString(s) {
+			s = pattern.ReplaceAllString(s, "[redacted]")
+			redacted = true
+		}
+	}
+	return s, redacted
+}
+
+// CleanRoute strips sensitive query parameters from a recorded route.
+//
+// The parameter is kept with an emptied value rather than removed. Which
+// parameters a consumer sent is itself interface evidence — it tells you what
+// they were trying to do — and deleting the key loses that.
+func (p RedactionPolicy) CleanRoute(route string) (string, bool) {
+	idx := strings.IndexByte(route, '?')
+	if idx < 0 {
+		return route, false
+	}
+
+	path, rawQuery := route[:idx], route[idx+1:]
+	values, err := url.ParseQuery(rawQuery)
+	if err != nil {
+		// An unparseable query is dropped entirely: it cannot be inspected, so
+		// it cannot be shown to be safe.
+		return path, true
+	}
+
+	drop := map[string]bool{}
+	for _, k := range p.DropQueryParams {
+		drop[strings.ToLower(k)] = true
+	}
+
+	redacted := false
+	keys := make([]string, 0, len(values))
+	for k := range values {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	cleaned := url.Values{}
+	for _, k := range keys {
+		if drop[strings.ToLower(k)] {
+			cleaned.Set(k, "[redacted]")
+			redacted = true
+			continue
+		}
+		for _, v := range values[k] {
+			scrubbed, hit := p.Scrub(v)
+			if hit {
+				redacted = true
+			}
+			cleaned.Add(k, scrubbed)
+		}
+	}
+
+	if len(cleaned) == 0 {
+		return path, redacted
+	}
+	return path + "?" + cleaned.Encode(), redacted
+}
+
+// Apply redacts a telemetry event in place and records what it did.
+//
+// The applied rules are recorded on the event so that later analysis knows what
+// it cannot see. Silent redaction would let an analyst mistake an absence of
+// evidence for evidence of absence.
+func (p RedactionPolicy) Apply(ev *contract.FluidTelemetry) {
+	var rules []string
+
+	if ev.ConsumerRef != "" && !strings.HasPrefix(ev.ConsumerRef, "psu-") {
+		ev.ConsumerRef = p.Pseudonymize(ev.ConsumerRef)
+		rules = append(rules, "pseudonymize-consumer")
+	}
+
+	if ev.Request != nil {
+		if cleaned, hit := p.CleanRoute(ev.Request.Route); hit {
+			ev.Request.Route = cleaned
+			rules = append(rules, "clean-route")
+		}
+	}
+
+	if ev.Error != nil && ev.Error.Detail != "" {
+		if scrubbed, hit := p.Scrub(ev.Error.Detail); hit {
+			ev.Error.Detail = scrubbed
+			rules = append(rules, "scrub-error-detail")
+		}
+	}
+
+	if ev.Sequence != nil && ev.Sequence.Pattern != "" {
+		if scrubbed, hit := p.Scrub(ev.Sequence.Pattern); hit {
+			ev.Sequence.Pattern = scrubbed
+			rules = append(rules, "scrub-sequence-pattern")
+		}
+	}
+
+	sort.Strings(rules)
+	ev.Redaction = &contract.FluidTelemetryRedaction{
+		Applied: len(rules) > 0,
+		Rules:   rules,
+	}
+}
+
+// Expired reports whether an event has outlived the retention policy.
+func (p RedactionPolicy) Expired(ev contract.FluidTelemetry, now time.Time) bool {
+	if p.RetentionDays <= 0 {
+		return false
+	}
+	return ev.OccurredAt.Before(now.AddDate(0, 0, -p.RetentionDays))
+}
+
+// SuppressSmallCohort reports whether a population is too small to report on
+// separately.
+//
+// Blueprint 6.2 requires cohort minimum sizes because a cohort of one is not a
+// cohort; it is a named individual with extra steps.
+func (p RedactionPolicy) SuppressSmallCohort(size int) bool {
+	if p.CohortMinimumSize <= 0 {
+		return false
+	}
+	return size < p.CohortMinimumSize
+}

internal/observation/redact_test.go

@@ -0,0 +1,204 @@
+package observation
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+func testPolicy() RedactionPolicy {
+	return DefaultRedactionPolicy([]byte("a-stable-salt-of-sufficient-length"))
+}
+
+func TestPolicyRequiresASalt(t *testing.T) {
+	if err := (RedactionPolicy{}).Validate(); !errors.Is(err, ErrNoSalt) {
+		t.Errorf("an unsalted policy validated: %v", err)
+	}
+	if err := testPolicy().Validate(); err != nil {
+		t.Errorf("a salted policy was rejected: %v", err)
+	}
+}
+
+// TestPseudonymIsStableAndOpaque covers both halves of the requirement: the
+// same consumer must look the same over time, and the value must not give the
+// identity back.
+func TestPseudonymIsStableAndOpaque(t *testing.T) {
+	p := testPolicy()
+
+	first := p.Pseudonymize("bernd@example.com")
+	if first == "" {
+		t.Fatal("pseudonymizing a real identity produced nothing")
+	}
+	if strings.Contains(first, "bernd") || strings.Contains(first, "example.com") {
+		t.Errorf("pseudonym leaks the identity: %q", first)
+	}
+
+	for i := 0; i < 20; i++ {
+		if again := p.Pseudonymize("bernd@example.com"); again != first {
+			t.Fatalf("pseudonym is unstable: %q then %q", first, again)
+		}
+	}
+
+	if p.Pseudonymize("someone-else@example.com") == first {
+		t.Error("two identities collided")
+	}
+
+	// A different salt must produce a different value, or the mapping would be
+	// portable between deployments.
+	other := DefaultRedactionPolicy([]byte("a-completely-different-salt-value"))
+	if other.Pseudonymize("bernd@example.com") == first {
+		t.Error("pseudonym does not depend on the salt")
+	}
+
+	if p.Pseudonymize("") != "" {
+		t.Error("an empty identity should stay empty rather than become a pseudonym")
+	}
+}
+
+func TestCleanRouteRedactsCredentialsButKeepsShape(t *testing.T) {
+	p := testPolicy()
+
+	got, redacted := p.CleanRoute("/v1/entries?token=hunter2&limit=10")
+	if !redacted {
+		t.Fatal("a route carrying a token was not flagged as redacted")
+	}
+	if strings.Contains(got, "hunter2") {
+		t.Errorf("token survived redaction: %q", got)
+	}
+	// Which parameters were sent is interface evidence in itself.
+	if !strings.Contains(got, "token=") {
+		t.Errorf("the parameter name was dropped, losing the evidence: %q", got)
+	}
+	if !strings.Contains(got, "limit=10") {
+		t.Errorf("a harmless parameter was removed: %q", got)
+	}
+
+	plain, redacted := p.CleanRoute("/v1/entries")
+	if redacted || plain != "/v1/entries" {
+		t.Errorf("a clean route was altered: %q", plain)
+	}
+}
+
+func TestUnparseableQueryIsDroppedEntirely(t *testing.T) {
+	p := testPolicy()
+	// A query that cannot be inspected cannot be shown to be safe.
+	got, redacted := p.CleanRoute("/v1/entries?%zz")
+	if !redacted {
+		t.Error("an unparseable query was not flagged")
+	}
+	if strings.Contains(got, "%zz") {
+		t.Errorf("unparseable query survived: %q", got)
+	}
+}
+
+func TestScrubRemovesSensitivePatterns(t *testing.T) {
+	p := testPolicy()
+
+	for _, tc := range []struct{ name, in, mustNotContain string }{
+		{"bearer token", "upstream rejected: Bearer eyJhbGciOiJIUzI1NiJ9.abc", "eyJhbGciOiJIUzI1NiJ9"},
+		{"email", "no account for bernd@example.com", "bernd@example.com"},
+		{"connection string", "dial postgres://user:hunter2@db.internal/prod", "hunter2"},
+	} {
+		got, hit := p.Scrub(tc.in)
+		if !hit {
+			t.Errorf("%s: not flagged as redacted", tc.name)
+		}
+		if strings.Contains(got, tc.mustNotContain) {
+			t.Errorf("%s: sensitive value survived: %q", tc.name, got)
+		}
+	}
+
+	if got, hit := p.Scrub("timeout after 5000ms"); hit || got != "timeout after 5000ms" {
+		t.Errorf("harmless text was altered: %q", got)
+	}
+}
+
+// TestApplyRecordsWhatItDid: silent redaction would let an analyst mistake an
+// absence of evidence for evidence of absence.
+func TestApplyRecordsWhatItDid(t *testing.T) {
+	p := testPolicy()
+
+	ev := contract.FluidTelemetry{
+		ConsumerRef: "bernd@example.com",
+		Request:     &contract.FluidTelemetryRequest{Route: "/v1/entries?api_key=secret"},
+		Error:       &contract.FluidTelemetryError{Class: contract.FluidTelemetryErrorClassBackendFailure, Detail: "dial postgres://u:p@db/x"},
+	}
+	p.Apply(&ev)
+
+	if ev.Redaction == nil || !ev.Redaction.Applied {
+		t.Fatal("redaction was applied but not recorded")
+	}
+	for _, want := range []string{"pseudonymize-consumer", "clean-route", "scrub-error-detail"} {
+		found := false
+		for _, r := range ev.Redaction.Rules {
+			if r == want {
+				found = true
+			}
+		}
+		if !found {
+			t.Errorf("rule %q not recorded; recorded: %v", want, ev.Redaction.Rules)
+		}
+	}
+	if strings.Contains(ev.ConsumerRef, "@") {
+		t.Error("consumer identity survived")
+	}
+}
+
+func TestApplyIsIdempotent(t *testing.T) {
+	// Events may pass through the filter more than once on their way to the
+	// store; pseudonymizing a pseudonym would break consumer continuity.
+	p := testPolicy()
+	ev := contract.FluidTelemetry{ConsumerRef: "consumer-1"}
+
+	p.Apply(&ev)
+	once := ev.ConsumerRef
+	p.Apply(&ev)
+
+	if ev.ConsumerRef != once {
+		t.Errorf("re-applying redaction changed the pseudonym: %q then %q", once, ev.ConsumerRef)
+	}
+}
+
+func TestApplyOnCleanEventRecordsNoRedaction(t *testing.T) {
+	p := testPolicy()
+	ev := contract.FluidTelemetry{Request: &contract.FluidTelemetryRequest{Route: "/v1/entries"}}
+	p.Apply(&ev)
+
+	if ev.Redaction == nil {
+		t.Fatal("redaction status not recorded at all")
+	}
+	if ev.Redaction.Applied {
+		t.Errorf("a clean event was marked redacted: %v", ev.Redaction.Rules)
+	}
+}
+
+func TestRetentionAndCohortFloor(t *testing.T) {
+	p := testPolicy()
+	now := time.Date(2026, 9, 4, 0, 0, 0, 0, time.UTC)
+
+	old := contract.FluidTelemetry{OccurredAt: now.AddDate(0, 0, -91)}
+	if !p.Expired(old, now) {
+		t.Error("an event past retention was not expired")
+	}
+	recent := contract.FluidTelemetry{OccurredAt: now.AddDate(0, 0, -1)}
+	if p.Expired(recent, now) {
+		t.Error("a recent event was expired")
+	}
+
+	unbounded := p
+	unbounded.RetentionDays = 0
+	if unbounded.Expired(old, now) {
+		t.Error("unbounded retention expired an event")
+	}
+
+	// A cohort of one is a named individual with extra steps.
+	if !p.SuppressSmallCohort(1) {
+		t.Error("a cohort of one was reportable")
+	}
+	if p.SuppressSmallCohort(50) {
+		t.Error("a large cohort was suppressed")
+	}
+}

internal/observation/topology.go

@@ -0,0 +1,306 @@
+package observation
+
+import (
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Interaction is one consumer's ordered call chain.
+//
+// ArchitectureBlueprint.md section 6.4: the analyzer looks beyond individual
+// requests, because interaction topologies are often more informative than
+// simple error counts. A single 200 tells you nothing; the same 200 fetched
+// forty times to find one record tells you the interface is missing a concept.
+type Interaction struct {
+	ConsumerRef string
+	Cohort      contract.CohortID
+	Started     time.Time
+	Ended       time.Time
+	Steps       []Step
+}
+
+// Step is one call within an interaction.
+type Step struct {
+	Route  string
+	Method string
+	Status int64
+	Error  contract.FluidTelemetryErrorClass
+}
+
+// Signature renders an interaction as a comparable shape.
+//
+// Routes are used rather than concrete URLs so that two consumers doing the
+// same thing to different resources produce the same signature. Without that
+// normalization every chain is unique and no pattern is ever detected twice.
+func (i Interaction) Signature() string {
+	parts := make([]string, 0, len(i.Steps))
+	for _, s := range i.Steps {
+		part := s.Method + " " + s.Route
+		if s.Error != "" {
+			part += " !" + string(s.Error)
+		}
+		parts = append(parts, part)
+	}
+	return strings.Join(parts, " -> ")
+}
+
+// Pattern is a recurring interaction shape observed across consumers.
+type Pattern struct {
+	Signature string              `json:"signature"`
+	Steps     int                 `json:"steps"`
+	Count     int                 `json:"occurrences"`
+	Consumers int                 `json:"independent_consumers"`
+	Cohorts   []contract.CohortID `json:"cohorts"`
+	// RepeatedStep names a route called more than once in the same chain, which
+	// is the usual shape of a consumer compensating for a missing capability.
+	RepeatedStep string `json:"repeated_step,omitempty"`
+	// MaxRepeats is how many times that route appeared in the worst chain.
+	MaxRepeats int `json:"max_repeats,omitempty"`
+	// RecoveredError names an error the consumer hit and then worked past,
+	// which distinguishes a recoverable misunderstanding from a hard failure.
+	RecoveredError contract.FluidTelemetryErrorClass `json:"recovered_error,omitempty"`
+	FirstSeen      time.Time                         `json:"first_seen"`
+	LastSeen       time.Time                         `json:"last_seen"`
+}
+
+// TopologyAnalyzer groups telemetry into interactions and finds patterns.
+type TopologyAnalyzer struct {
+	// ChainGap is the idle time after which a consumer's next call starts a new
+	// interaction rather than continuing the previous one.
+	ChainGap time.Duration
+	// MinOccurrences is how often a shape must appear before it is a pattern.
+	MinOccurrences int
+	// MinConsumers is how many independent consumers must show the shape.
+	// One consumer repeating itself is a client bug; several independent
+	// consumers converging on the same workaround is interface pressure.
+	MinConsumers int
+}
+
+// NewTopologyAnalyzer returns an analyzer with workable defaults.
+func NewTopologyAnalyzer() *TopologyAnalyzer {
+	return &TopologyAnalyzer{
+		ChainGap:       30 * time.Second,
+		MinOccurrences: 3,
+		MinConsumers:   2,
+	}
+}
+
+// Interactions groups events into per-consumer call chains.
+//
+// Grouping prefers an explicit chain id when the consumer supplied one, and
+// falls back to time-bounded sessions per consumer. The fallback is a heuristic
+// and is why chain ids are worth asking agentic consumers for.
+func (a *TopologyAnalyzer) Interactions(events []contract.FluidTelemetry) []Interaction {
+	ordered := make([]contract.FluidTelemetry, len(events))
+	copy(ordered, events)
+	sort.SliceStable(ordered, func(i, j int) bool {
+		return ordered[i].OccurredAt.Before(ordered[j].OccurredAt)
+	})
+
+	type key struct{ consumer, chain string }
+	open := map[key]*Interaction{}
+	var done []Interaction
+
+	for _, ev := range ordered {
+		if ev.Request == nil && ev.Error == nil {
+			continue
+		}
+
+		consumer := ev.ConsumerRef
+		if consumer == "" {
+			consumer = ev.CorrelationID
+		}
+		if consumer == "" {
+			continue
+		}
+
+		chain := ""
+		if ev.Sequence != nil {
+			chain = ev.Sequence.ChainID
+		}
+		k := key{consumer: consumer, chain: chain}
+
+		current, ok := open[k]
+		// With no explicit chain id, an idle gap ends the interaction.
+		if ok && chain == "" && ev.OccurredAt.Sub(current.Ended) > a.ChainGap {
+			done = append(done, *current)
+			ok = false
+		}
+		if !ok {
+			cohort := contract.CohortID("")
+			if ev.Cohort != nil {
+				cohort = *ev.Cohort
+			}
+			current = &Interaction{
+				ConsumerRef: consumer,
+				Cohort:      cohort,
+				Started:     ev.OccurredAt,
+			}
+			open[k] = current
+		}
+
+		current.Ended = ev.OccurredAt
+		current.Steps = append(current.Steps, stepOf(ev))
+	}
+
+	for _, in := range open {
+		done = append(done, *in)
+	}
+	sort.Slice(done, func(i, j int) bool { return done[i].Started.Before(done[j].Started) })
+	return done
+}
+
+func stepOf(ev contract.FluidTelemetry) Step {
+	var s Step
+	if ev.Request != nil {
+		s.Route = ev.Request.Route
+		s.Method = ev.Request.Method
+		if ev.Request.Status != nil {
+			s.Status = *ev.Request.Status
+		}
+	}
+	if ev.Error != nil {
+		s.Error = ev.Error.Class
+	}
+	if s.Route == "" {
+		s.Route = "(unknown)"
+	}
+	return s
+}
+
+// Patterns reports recurring interaction shapes.
+func (a *TopologyAnalyzer) Patterns(events []contract.FluidTelemetry) []Pattern {
+	interactions := a.Interactions(events)
+
+	type acc struct {
+		count      int
+		consumers  map[string]struct{}
+		cohorts    map[contract.CohortID]struct{}
+		steps      int
+		repeated   string
+		maxRepeats int
+		recovered  contract.FluidTelemetryErrorClass
+		first      time.Time
+		last       time.Time
+	}
+	groups := map[string]*acc{}
+
+	for _, in := range interactions {
+		if len(in.Steps) == 0 {
+			continue
+		}
+		sig := in.Signature()
+
+		g, ok := groups[sig]
+		if !ok {
+			g = &acc{
+				consumers: map[string]struct{}{},
+				cohorts:   map[contract.CohortID]struct{}{},
+				steps:     len(in.Steps),
+				first:     in.Started,
+				last:      in.Ended,
+			}
+			groups[sig] = g
+		}
+
+		g.count++
+		g.consumers[in.ConsumerRef] = struct{}{}
+		if in.Cohort != "" {
+			g.cohorts[in.Cohort] = struct{}{}
+		}
+		if in.Started.Before(g.first) {
+			g.first = in.Started
+		}
+		if in.Ended.After(g.last) {
+			g.last = in.Ended
+		}
+
+		if route, n := repeatedRoute(in); n > g.maxRepeats {
+			g.repeated, g.maxRepeats = route, n
+		}
+		if class, ok := recoveredError(in); ok {
+			g.recovered = class
+		}
+	}
+
+	out := make([]Pattern, 0, len(groups))
+	for sig, g := range groups {
+		if g.count < a.MinOccurrences || len(g.consumers) < a.MinConsumers {
+			continue
+		}
+		cohorts := make([]contract.CohortID, 0, len(g.cohorts))
+		for c := range g.cohorts {
+			cohorts = append(cohorts, c)
+		}
+		sort.Slice(cohorts, func(i, j int) bool { return cohorts[i] < cohorts[j] })
+
+		p := Pattern{
+			Signature: sig,
+			Steps:     g.steps,
+			Count:     g.count,
+			Consumers: len(g.consumers),
+			Cohorts:   cohorts,
+			FirstSeen: g.first,
+			LastSeen:  g.last,
+		}
+		if g.maxRepeats > 1 {
+			p.RepeatedStep, p.MaxRepeats = g.repeated, g.maxRepeats
+		}
+		p.RecoveredError = g.recovered
+		out = append(out, p)
+	}
+
+	// Most frequent first: an analyst reading this wants the biggest signal at
+	// the top, and a stable tiebreak keeps the output diffable.
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Count != out[j].Count {
+			return out[i].Count > out[j].Count
+		}
+		return out[i].Signature < out[j].Signature
+	})
+	return out
+}
+
+// repeatedRoute finds the most-repeated route within one interaction.
+func repeatedRoute(in Interaction) (string, int) {
+	counts := map[string]int{}
+	for _, s := range in.Steps {
+		counts[s.Method+" "+s.Route]++
+	}
+
+	best, bestN := "", 0
+	routes := make([]string, 0, len(counts))
+	for r := range counts {
+		routes = append(routes, r)
+	}
+	sort.Strings(routes)
+	for _, r := range routes {
+		if counts[r] > bestN {
+			best, bestN = r, counts[r]
+		}
+	}
+	return best, bestN
+}
+
+// recoveredError reports an error the consumer hit and then got past.
+//
+// This is the shape Blueprint 6.4 calls out as a recoverable misunderstanding:
+// invalid request, schema lookup, retry with a corrected request. It is a
+// different problem from a chain that simply fails, and conflating the two
+// would send the wrong hypothesis to the Daimon.
+func recoveredError(in Interaction) (contract.FluidTelemetryErrorClass, bool) {
+	var seen contract.FluidTelemetryErrorClass
+	for _, s := range in.Steps {
+		if s.Error != "" {
+			seen = s.Error
+			continue
+		}
+		if seen != "" && s.Status >= 200 && s.Status < 300 {
+			return seen, true
+		}
+	}
+	return "", false
+}

internal/observation/topology_test.go

@@ -0,0 +1,223 @@
+package observation
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+var base = time.Date(2026, 9, 4, 10, 0, 0, 0, time.UTC)
+
+func req(consumer string, offset time.Duration, method, route string, status int64) contract.FluidTelemetry {
+	cohort := contract.CohortID("coding-agents")
+	s := status
+	return contract.FluidTelemetry{
+		ID:          fmt.Sprintf("tl-%s-%d", consumer, offset),
+		OccurredAt:  base.Add(offset),
+		Kind:        contract.FluidTelemetryKindRequest,
+		ConsumerRef: consumer,
+		Cohort:      &cohort,
+		Request:     &contract.FluidTelemetryRequest{Route: route, Method: method, Status: &s},
+	}
+}
+
+func errEv(consumer string, offset time.Duration, route string, class contract.FluidTelemetryErrorClass) contract.FluidTelemetry {
+	ev := req(consumer, offset, "GET", route, 400)
+	ev.Kind = contract.FluidTelemetryKindError
+	ev.Error = &contract.FluidTelemetryError{Class: class}
+	return ev
+}
+
+// TestDetectsInefficientUsagePattern reproduces the Blueprint section 33
+// worked example: consumers listing everything to find one record.
+func TestDetectsInefficientUsagePattern(t *testing.T) {
+	a := NewTopologyAnalyzer()
+
+	var events []contract.FluidTelemetry
+	for _, consumer := range []string{"c-1", "c-2", "c-3"} {
+		for chain := 0; chain < 2; chain++ {
+			start := time.Duration(chain) * time.Hour
+			events = append(events,
+				req(consumer, start, "GET", "/customers/{id}/invoices", 200),
+				req(consumer, start+time.Second, "GET", "/customers/{id}/invoices", 200),
+				req(consumer, start+2*time.Second, "GET", "/customers/{id}/invoices", 200),
+			)
+		}
+	}
+
+	patterns := a.Patterns(events)
+	if len(patterns) == 0 {
+		t.Fatal("no pattern detected in a clearly repeated shape")
+	}
+
+	p := patterns[0]
+	if p.Consumers != 3 {
+		t.Errorf("independent consumers = %d, want 3", p.Consumers)
+	}
+	if p.Count != 6 {
+		t.Errorf("occurrences = %d, want 6", p.Count)
+	}
+	// The repeated route is the signal that the consumer is compensating.
+	if p.MaxRepeats != 3 {
+		t.Errorf("max repeats = %d, want 3", p.MaxRepeats)
+	}
+	if p.RepeatedStep != "GET /customers/{id}/invoices" {
+		t.Errorf("repeated step = %q", p.RepeatedStep)
+	}
+}
+
+// TestOneConsumerRepeatingIsNotAPattern: a single client looping is a client
+// bug; several independent consumers converging is interface pressure.
+func TestOneConsumerRepeatingIsNotAPattern(t *testing.T) {
+	a := NewTopologyAnalyzer()
+
+	var events []contract.FluidTelemetry
+	for chain := 0; chain < 10; chain++ {
+		start := time.Duration(chain) * time.Hour
+		events = append(events,
+			req("c-1", start, "GET", "/entries", 200),
+			req("c-1", start+time.Second, "GET", "/entries", 200),
+		)
+	}
+
+	if patterns := a.Patterns(events); len(patterns) != 0 {
+		t.Errorf("a single consumer's loop was reported as a pattern: %+v", patterns)
+	}
+}
+
+func TestChainGapSplitsInteractions(t *testing.T) {
+	a := NewTopologyAnalyzer()
+	a.ChainGap = 10 * time.Second
+
+	events := []contract.FluidTelemetry{
+		req("c-1", 0, "GET", "/a", 200),
+		req("c-1", 2*time.Second, "GET", "/b", 200),
+		// Well past the gap: a new interaction, not a continuation.
+		req("c-1", time.Minute, "GET", "/c", 200),
+	}
+
+	interactions := a.Interactions(events)
+	if len(interactions) != 2 {
+		t.Fatalf("got %d interactions, want 2", len(interactions))
+	}
+	if len(interactions[0].Steps) != 2 || len(interactions[1].Steps) != 1 {
+		t.Errorf("steps split wrongly: %d and %d",
+			len(interactions[0].Steps), len(interactions[1].Steps))
+	}
+}
+
+func TestExplicitChainIDOverridesTiming(t *testing.T) {
+	a := NewTopologyAnalyzer()
+	a.ChainGap = time.Second
+
+	withChain := func(ev contract.FluidTelemetry, id string) contract.FluidTelemetry {
+		ev.Sequence = &contract.FluidTelemetrySequence{ChainID: id}
+		return ev
+	}
+
+	// Two calls an hour apart, but the consumer says they are one task.
+	events := []contract.FluidTelemetry{
+		withChain(req("c-1", 0, "GET", "/a", 200), "chain-1"),
+		withChain(req("c-1", time.Hour, "GET", "/b", 200), "chain-1"),
+	}
+
+	interactions := a.Interactions(events)
+	if len(interactions) != 1 {
+		t.Fatalf("an explicit chain id was split by timing: got %d interactions", len(interactions))
+	}
+}
+
+// TestRecoveredErrorIsDistinguished: invalid request, then a corrected retry,
+// is a recoverable misunderstanding rather than a hard failure, and the two
+// deserve different hypotheses.
+func TestRecoveredErrorIsDistinguished(t *testing.T) {
+	a := NewTopologyAnalyzer()
+
+	var events []contract.FluidTelemetry
+	for _, consumer := range []string{"c-1", "c-2", "c-3"} {
+		for chain := 0; chain < 2; chain++ {
+			start := time.Duration(chain) * time.Hour
+			events = append(events,
+				errEv(consumer, start, "/entries", contract.FluidTelemetryErrorClassValidation),
+				req(consumer, start+time.Second, "GET", "/entries", 200),
+			)
+		}
+	}
+
+	patterns := a.Patterns(events)
+	if len(patterns) == 0 {
+		t.Fatal("no pattern detected")
+	}
+	if patterns[0].RecoveredError != contract.FluidTelemetryErrorClassValidation {
+		t.Errorf("recovered error = %q, want validation", patterns[0].RecoveredError)
+	}
+}
+
+func TestPatternsAreDeterministicallyOrdered(t *testing.T) {
+	a := NewTopologyAnalyzer()
+
+	var events []contract.FluidTelemetry
+	for _, consumer := range []string{"c-1", "c-2", "c-3"} {
+		for chain := 0; chain < 5; chain++ {
+			start := time.Duration(chain) * time.Hour
+			events = append(events, req(consumer, start, "GET", "/frequent", 200))
+		}
+		for chain := 0; chain < 2; chain++ {
+			start := time.Duration(chain+10) * time.Hour
+			events = append(events, req(consumer, start, "GET", "/rare", 200))
+		}
+	}
+
+	first := a.Patterns(events)
+	if len(first) < 2 {
+		t.Fatalf("expected two patterns, got %d", len(first))
+	}
+	// Biggest signal first, so an analyst reads the important thing first.
+	if first[0].Count < first[1].Count {
+		t.Error("patterns are not ordered by frequency")
+	}
+
+	for i := 0; i < 20; i++ {
+		again := a.Patterns(events)
+		for j := range first {
+			if again[j].Signature != first[j].Signature {
+				t.Fatal("pattern ordering varied between runs")
+			}
+		}
+	}
+}
+
+func TestCohortPopulationsRespectMinimumSize(t *testing.T) {
+	policy := testPolicy()
+	e := NewCohortEngine("unclassified", policy)
+
+	var events []contract.FluidTelemetry
+	// One cohort with a single consumer, one with plenty.
+	small := contract.CohortID("lone-partner")
+	for i := 0; i < 20; i++ {
+		ev := req("only-one", time.Duration(i)*time.Second, "GET", "/a", 200)
+		ev.Cohort = &small
+		events = append(events, ev)
+	}
+	big := contract.CohortID("agents")
+	for i := 0; i < 20; i++ {
+		ev := req(fmt.Sprintf("c-%d", i), time.Duration(i)*time.Second, "GET", "/a", 200)
+		ev.Cohort = &big
+		events = append(events, ev)
+	}
+
+	pops := e.Populations(events)
+	byCohort := map[contract.CohortID]Population{}
+	for _, p := range pops {
+		byCohort[p.Cohort] = p
+	}
+
+	if !byCohort[small].Suppressed {
+		t.Error("a cohort of one consumer was reportable")
+	}
+	if byCohort[big].Suppressed {
+		t.Error("a cohort of twenty consumers was suppressed")
+	}
+}