diff --git a/internal/observation/cohort.go b/internal/observation/cohort.go
new file mode 100644
index 0000000..4342914
--- /dev/null
+++ b/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
+}
diff --git a/internal/observation/redact.go b/internal/observation/redact.go
new file mode 100644
index 0000000..67a1954
--- /dev/null
+++ b/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
+}
diff --git a/internal/observation/redact_test.go b/internal/observation/redact_test.go
new file mode 100644
index 0000000..6a58d83
--- /dev/null
+++ b/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")
+	}
+}
diff --git a/internal/observation/topology.go b/internal/observation/topology.go
new file mode 100644
index 0000000..9d7febf
--- /dev/null
+++ b/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
+}
diff --git a/internal/observation/topology_test.go b/internal/observation/topology_test.go
new file mode 100644
index 0000000..217437b
--- /dev/null
+++ b/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")
+	}
+}