diff --git a/cmd/fluid-control/main.go b/cmd/fluid-control/main.go index e0c2fed..a90fed7 100644 --- a/cmd/fluid-control/main.go +++ b/cmd/fluid-control/main.go @@ -30,6 +30,7 @@ import ( "github.com/tegwick/fluid-core/internal/control" "github.com/tegwick/fluid-core/internal/evidence" "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/observation" "github.com/tegwick/fluid-core/internal/policy" "github.com/tegwick/fluid-core/internal/publish" "github.com/tegwick/fluid-core/internal/signing" @@ -49,6 +50,8 @@ func run() error { keyID = flag.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier") keyFile = flag.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 private key file") ephemeral = flag.Bool("ephemeral-key", false, "generate a throwaway signing key (development only)") + saltFile = flag.String("redaction-salt-file", os.Getenv("FLUID_REDACTION_SALT"), + "file holding the pseudonymization salt; required for the observation plane") ) flag.Parse() @@ -83,9 +86,33 @@ func run() error { return err } + // The observation plane is optional. Without a salt there is no safe way to + // pseudonymize consumer identities, so the endpoints that would record them + // are simply not served rather than served unsafely. + var pressureAPI *control.PressureAPI + if *saltFile != "" { + salt, err := os.ReadFile(*saltFile) + if err != nil { + return fmt.Errorf("read redaction salt: %w", err) + } + policy := observation.DefaultRedactionPolicy([]byte(trimSpace(string(salt)))) + ingest, err := observation.NewIngest(ev, contract.InterfaceID(*iface), policy) + if err != nil { + return err + } + pressureAPI = control.NewPressureAPI( + observation.NewPressureRegistry(ev, contract.InterfaceID(*iface)), ingest) + } else { + log.Print("no redaction salt configured: telemetry, feedback and pressure endpoints are disabled") + } + srv := &http.Server{ - Addr: *addr, - Handler: control.NewServer(control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate)).Routes(), + Addr: *addr, + Handler: control.NewServer( + control.NewRevisionAPI(ev, pipeline), + control.NewIntentAPI(intents, gate), + pressureAPI, + ).Routes(), ReadHeaderTimeout: 10 * time.Second, } diff --git a/cmd/fluid/insight.go b/cmd/fluid/insight.go new file mode 100644 index 0000000..a57d2b8 --- /dev/null +++ b/cmd/fluid/insight.go @@ -0,0 +1,385 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/fitness" + "github.com/tegwick/fluid-core/internal/observation" +) + +// ---------- pressure ---------- + +func runPressure(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("pressure needs a subcommand: list, show, analyze, dismiss") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + reg := observation.NewPressureRegistry(store, contract.InterfaceID(iface)) + + switch args[0] { + case "list": + fs := newFlagSet("pressure list") + status := fs.String("status", "", "filter by status, such as OPEN or DISMISSED") + if err := fs.Parse(args[1:]); err != nil { + return err + } + + list, err := reg.List(ctx, contract.FluidPressureStatus(*status)) + if err != nil { + return err + } + if len(list) == 0 { + fmt.Println("no pressure recorded") + return nil + } + + w := out() + fmt.Fprintln(w, "ID\tCLASS\tSEV\tCONF\tCONSUMERS\tSTATUS\tSUMMARY") + for _, p := range list { + consumers := int64(0) + if p.Frequency != nil && p.Frequency.IndependentConsumers != nil { + consumers = *p.Frequency.IndependentConsumers + } + fmt.Fprintf(w, "%s\t%s\t%.2f\t%.2f\t%d\t%s\t%s\n", + p.ID, p.Class, unitOf(p.Severity), unitOf(p.Confidence), + consumers, p.Status, truncate(oneLine(p.Summary), 60)) + } + return w.Flush() + + case "show": + if len(args) < 2 { + return errors.New("pressure show needs an id") + } + p, err := reg.Get(ctx, contract.PressureID(args[1])) + if err != nil { + return err + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(p) + + case "analyze": + return analyzePressure(ctx, g, store, reg, args[1:]) + + case "dismiss": + fs := newFlagSet("pressure dismiss") + reason := fs.String("reason", "", "why this pressure will not be acted on") + if err := fs.Parse(args[1:]); err != nil { + return err + } + rest := fs.fs.Args() + if len(rest) == 0 { + return errors.New("pressure dismiss needs an id") + } + if *reason == "" { + return errors.New("dismissal requires --reason; an unexplained dismissal is not auditable") + } + + actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()} + if err := reg.SetStatus(ctx, contract.PressureID(rest[0]), + contract.FluidPressureStatusDISMISSED, actor, *reason); err != nil { + return err + } + fmt.Printf("dismissed %s\n", rest[0]) + return nil + + default: + return fmt.Errorf("unknown pressure subcommand %q", args[0]) + } +} + +// analyzePressure runs the classifier over recorded telemetry. +// +// Analysis is an explicit command rather than something that happens on ingest. +// Blueprint section 30 wants adaptive work to be budgeted and deferrable, and a +// classifier that ran on every event would be neither. +func analyzePressure(ctx context.Context, g globals, store *evidence.SQLStore, reg *observation.PressureRegistry, args []string) error { + fs := newFlagSet("pressure analyze") + since := fs.String("since", "168h", "how far back to analyze, as a Go duration") + dryRun := fs.Bool("dry-run", false, "report findings without recording them") + if err := fs.Parse(args); err != nil { + return err + } + + window, err := time.ParseDuration(*since) + if err != nil { + return fmt.Errorf("invalid --since: %w", err) + } + + events, err := store.Telemetry(ctx, evidence.TelemetryFilter{ + InterfaceID: contract.InterfaceID(g.iface), + Since: time.Now().Add(-window), + }) + if err != nil { + return err + } + if len(events) == 0 { + fmt.Println("no telemetry in the window; nothing to analyze") + return nil + } + + classifier := observation.NewClassifier( + observation.DefaultClassifierOptions(), observation.NewTopologyAnalyzer()) + findings := classifier.Classify(events) + + if len(findings) == 0 { + fmt.Printf("analyzed %d events; no material pressure found\n", len(events)) + return nil + } + + w := out() + fmt.Fprintln(w, "CLASS\tSEV\tCONF\tCONSUMERS\tSUMMARY") + for _, f := range findings { + fmt.Fprintf(w, "%s\t%.2f\t%.2f\t%d\t%s\n", + f.Class, f.Severity, f.Confidence, f.Consumers, truncate(oneLine(f.Summary), 70)) + } + if err := w.Flush(); err != nil { + return err + } + + if *dryRun { + fmt.Printf("\n%d finding(s) from %d events; not recorded (--dry-run)\n", len(findings), len(events)) + return nil + } + + recorded, err := reg.RecordAll(ctx, findings) + if err != nil { + return err + } + fmt.Printf("\nrecorded %d pressure record(s) from %d events\n", len(recorded), len(events)) + return nil +} + +// ---------- cohorts ---------- + +func runCohort(ctx context.Context, g globals, args []string) error { + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + fs := newFlagSet("cohort list") + since := fs.String("since", "168h", "how far back to summarize") + saltFile := fs.String("salt-file", os.Getenv("FLUID_REDACTION_SALT"), "pseudonymization salt file") + if err := fs.Parse(args); err != nil { + return err + } + + window, err := time.ParseDuration(*since) + if err != nil { + return fmt.Errorf("invalid --since: %w", err) + } + + policy, err := loadPolicy(*saltFile) + if err != nil { + return err + } + + events, err := store.Telemetry(ctx, evidence.TelemetryFilter{ + InterfaceID: contract.InterfaceID(iface), + Since: time.Now().Add(-window), + }) + if err != nil { + return err + } + + pops := observation.NewCohortEngine("unclassified", policy).Populations(events) + if len(pops) == 0 { + fmt.Println("no cohort activity in the window") + return nil + } + + w := out() + fmt.Fprintln(w, "COHORT\tCONSUMERS\tEVENTS") + for _, p := range pops { + consumers := fmt.Sprint(p.Consumers) + if p.Suppressed { + // Reporting the exact count of a tiny cohort identifies individuals. + consumers = fmt.Sprintf("<%d (suppressed)", policy.CohortMinimumSize) + } + fmt.Fprintf(w, "%s\t%s\t%d\n", p.Cohort, consumers, p.Events) + } + return w.Flush() +} + +// ---------- fitness ---------- + +func runFitness(ctx context.Context, g globals, args []string) error { + if len(args) == 0 || args[0] != "compare" { + return errors.New("usage: fluid fitness compare --control R-1 --candidate R-2") + } + + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + fs := newFlagSet("fitness compare") + control := fs.String("control", "", "control revision") + candidate := fs.String("candidate", "", "candidate revision") + since := fs.String("since", "168h", "measurement window") + target := fs.Float64("target-requests-per-task", 0, "primary target for requests per completed task") + latencyGuard := fs.Float64("guard-p95-latency-ms", 0, "p95 latency guardrail") + errorGuard := fs.Float64("guard-error-rate", 0, "error rate guardrail") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *control == "" || *candidate == "" { + return errors.New("fitness compare needs --control and --candidate") + } + + window, err := time.ParseDuration(*since) + if err != nil { + return fmt.Errorf("invalid --since: %w", err) + } + start := time.Now().Add(-window) + + events, err := store.Telemetry(ctx, evidence.TelemetryFilter{ + InterfaceID: contract.InterfaceID(iface), + Since: start, + }) + if err != nil { + return err + } + + measureWindow := fitness.Window{Start: start} + observations := fitness.NewMeasurer().Measure(events, measureWindow) + + // Specs are declared here rather than inferred from the data, so that the + // criteria a comparison was judged against are visible in the command that + // ran it. + var specs []fitness.MetricSpec + if *target > 0 { + specs = append(specs, fitness.MetricSpec{ + Name: fitness.MetricRequestsPerTask, Role: fitness.RolePrimary, + Direction: fitness.Lower, Target: target, + }) + } + if *latencyGuard > 0 { + specs = append(specs, fitness.MetricSpec{ + Name: fitness.MetricP95LatencyMS, Role: fitness.RoleGuardrail, + Direction: fitness.Lower, Threshold: latencyGuard, + }) + } + if *errorGuard > 0 { + specs = append(specs, fitness.MetricSpec{ + Name: fitness.MetricErrorRate, Role: fitness.RoleGuardrail, + Direction: fitness.Lower, Threshold: errorGuard, + }) + } + if len(specs) == 0 { + return errors.New("no criteria given: pass at least --target-requests-per-task") + } + + eval := fitness.NewEvaluator().Evaluate( + contract.RevisionID(*control), contract.RevisionID(*candidate), + measureWindow, specs, observations) + + fmt.Printf("Fitness: %s vs %s\nWindow: since %s\nVerdict: %s\n\n", + eval.Control, eval.Candidate, start.Format(time.RFC3339), eval.Verdict) + + w := out() + fmt.Fprintln(w, "METRIC\tROLE\tBASELINE\tCURRENT\tDELTA\tOUTCOME") + for _, m := range eval.Metrics { + outcome := "" + switch { + case m.Underpowered: + outcome = "underpowered" + case m.TargetMet != nil && *m.TargetMet: + outcome = "target met" + case m.TargetMet != nil: + outcome = "target missed" + case m.GuardrailBreached != nil && *m.GuardrailBreached: + outcome = "BREACHED" + case m.GuardrailBreached != nil: + outcome = "within guardrail" + } + fmt.Fprintf(w, "%s\t%s\t%.4g\t%.4g\t%+.4g\t%s\n", + m.Name, m.Role, m.Baseline, m.Current, m.Delta, outcome) + } + if err := w.Flush(); err != nil { + return err + } + + if len(eval.Reasons) > 0 { + fmt.Println() + for _, r := range eval.Reasons { + fmt.Printf(" %s\n", r) + } + } + return nil +} + +// ---------- helpers ---------- + +// loadPolicy builds a redaction policy from a salt file. +// +// It refuses to invent a salt. A generated one would change on every +// invocation, so the same consumer would look like a new consumer each time and +// every cohort count would be wrong. +func loadPolicy(saltFile string) (observation.RedactionPolicy, error) { + if saltFile == "" { + return observation.RedactionPolicy{}, errors.New( + "no redaction salt: pass --salt-file or set FLUID_REDACTION_SALT") + } + salt, err := os.ReadFile(saltFile) + if err != nil { + return observation.RedactionPolicy{}, fmt.Errorf("read salt: %w", err) + } + policy := observation.DefaultRedactionPolicy([]byte(trimSpaceStr(string(salt)))) + if err := policy.Validate(); err != nil { + return observation.RedactionPolicy{}, err + } + return policy, nil +} + +func trimSpaceStr(s string) string { + start, end := 0, len(s) + for start < end && (s[start] == ' ' || s[start] == '\n' || s[start] == '\t' || s[start] == '\r') { + start++ + } + for end > start && (s[end-1] == ' ' || s[end-1] == '\n' || s[end-1] == '\t' || s[end-1] == '\r') { + end-- + } + return s[start:end] +} + +func unitOf(v *contract.UnitInterval) float64 { + if v == nil { + return 0 + } + return float64(*v) +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n-1] + "…" +} diff --git a/cmd/fluid/main.go b/cmd/fluid/main.go index 0c9f26f..6da0158 100644 --- a/cmd/fluid/main.go +++ b/cmd/fluid/main.go @@ -30,6 +30,11 @@ Commands: revision show Show one revision descriptor policy put Install a routing policy policy show Show the current routing policy + pressure list List recorded interface pressure + pressure analyze Classify recent telemetry into pressure records + pressure dismiss Record that a pressure will not be acted on + cohort Summarize cohort populations + fitness compare Compare a candidate revision against its control events Show audit events telemetry Show recorded telemetry audit trace Reconstruct the history behind a revision @@ -113,6 +118,12 @@ func run(args []string) error { return runEvents(ctx, g, rest) case "telemetry": return runTelemetry(ctx, g, rest) + case "pressure": + return runPressure(ctx, g, rest) + case "cohort": + return runCohort(ctx, g, rest) + case "fitness": + return runFitness(ctx, g, rest) case "audit": return runAudit(ctx, g, rest) default: diff --git a/conformance/loop/loop_test.go b/conformance/loop/loop_test.go new file mode 100644 index 0000000..dafc1f7 --- /dev/null +++ b/conformance/loop/loop_test.go @@ -0,0 +1,154 @@ +package main + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/observation" + "github.com/tegwick/fluid-core/internal/runtime" +) + +// TestInsightLoopEndToEnd drives real traffic through the gateway and confirms +// the observation plane turns it into a pressure record. +// +// This is the Blueprint section 33 shape end to end: consumers repeatedly +// listing a collection to find one entry, observed at the gateway, redacted on +// ingest, and classified as inefficient usage without anyone naming it as such. +func TestInsightLoopEndToEnd(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + + store, err := evidence.OpenSQLite(ctx, filepath.Join(dir, "e.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + policy := observation.DefaultRedactionPolicy([]byte("a-stable-salt-for-this-test-only")) + ingest, err := observation.NewIngest(store, "hall-publishing", policy) + if err != nil { + t.Fatal(err) + } + + adapter := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`[{"id":1},{"id":2}]`)) + })) + defer adapter.Close() + + reg := runtime.NewRegistry("hall-publishing") + d := contract.Revision{ + SchemaVersion: "0.1", ID: "R-1", Interface: "hall-publishing", + State: contract.RevisionStateStable, + Contract: contract.RevisionContract{Type: contract.RevisionContractTypeOpenapi, Digest: contract.Digest("sha256:" + repeat("1", 64))}, + Runtime: contract.RevisionRuntime{Upstream: adapter.URL}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + }, + } + if err := reg.PutRevision(d); err != nil { + t.Fatal(err) + } + if err := reg.PutPolicy(contract.RoutingPolicy{ + SchemaVersion: "0.1", Interface: "hall-publishing", + Generation: 1, DefaultRevision: "R-1", + }); err != nil { + t.Fatal(err) + } + + emitter := runtime.NewEmitter(ingest, runtime.EmitterOptions{Buffer: 4096, Workers: 2}) + gw, err := runtime.NewGateway(runtime.GatewayOptions{ + Interface: "hall-publishing", + Registry: reg, + Resolver: runtime.NewResolver(reg, false), + Connector: runtime.NewConnector(), + Emitter: emitter, + Cohorts: observation.NewCohortEngine("coding-agents", policy), + Response: runtime.ResponsePolicy{FeedbackPath: "/v1/feedback"}, + }) + if err != nil { + t.Fatal(err) + } + + // Four independent consumers, each listing the collection four times per + // task to find one entry: the Blueprint section 33 shape. + for c := 0; c < 4; c++ { + for chain := 0; chain < 3; chain++ { + for i := 0; i < 4; i++ { + req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil) + req.Header.Set("X-FLUID-Consumer", fmt.Sprintf("agent-%d", c)) + req.Header.Set("X-FLUID-Correlation", fmt.Sprintf("c-%d-%d", c, chain)) + rec := httptest.NewRecorder() + gw.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("request failed: %d", rec.Code) + } + } + } + } + emitter.Close() + + events, err := store.Telemetry(ctx, evidence.TelemetryFilter{InterfaceID: "hall-publishing"}) + if err != nil { + t.Fatal(err) + } + if len(events) != 48 { + t.Fatalf("recorded %d telemetry events, want 48", len(events)) + } + + // The raw consumer header must never have reached the store. + for _, ev := range events { + if ev.ConsumerRef == "agent-0" { + t.Fatal("a raw consumer identity reached the evidence store") + } + } + + classifier := observation.NewClassifier( + observation.DefaultClassifierOptions(), observation.NewTopologyAnalyzer()) + findings := classifier.Classify(events) + if len(findings) == 0 { + t.Fatal("no pressure found in clearly inefficient traffic") + } + + registry := observation.NewPressureRegistry(store, "hall-publishing") + recorded, err := registry.RecordAll(ctx, findings) + if err != nil { + t.Fatal(err) + } + if len(recorded) == 0 { + t.Fatal("no pressure recorded") + } + + var inefficient *contract.FluidPressure + for i := range recorded { + if recorded[i].Class == contract.PressureClassSuccessfulButInefficientUsage { + inefficient = &recorded[i] + } + } + if inefficient == nil { + t.Fatalf("inefficient usage not recorded; got %v", recorded) + } + if inefficient.Frequency == nil || *inefficient.Frequency.IndependentConsumers != 4 { + t.Errorf("independent consumers not recorded correctly: %+v", inefficient.Frequency) + } + if len(inefficient.EvidenceRefs) == 0 { + t.Error("pressure recorded with no evidence references") + } + + t.Logf("pressure %s: %s (severity %.2f, confidence %.2f, %d consumers)", + inefficient.ID, inefficient.Summary, + float64(*inefficient.Severity), float64(*inefficient.Confidence), + *inefficient.Frequency.IndependentConsumers) +} + +func repeat(s string, n int) string { return strings.Repeat(s, n) } diff --git a/internal/contract/event_gen.go b/internal/contract/event_gen.go index a6f478a..9bb0da2 100644 --- a/internal/contract/event_gen.go +++ b/internal/contract/event_gen.go @@ -10,6 +10,7 @@ const ( FluidEventEntityTypeHypothesis FluidEventEntityType = "hypothesis" FluidEventEntityTypeRevision FluidEventEntityType = "revision" FluidEventEntityTypeExperiment FluidEventEntityType = "experiment" + FluidEventEntityTypeFeedback FluidEventEntityType = "feedback" FluidEventEntityTypeBackendRequirement FluidEventEntityType = "backend_requirement" FluidEventEntityTypeIntent FluidEventEntityType = "intent" FluidEventEntityTypeDecision FluidEventEntityType = "decision" @@ -19,7 +20,7 @@ const ( // Valid reports whether v is a defined FluidEventEntityType. func (v FluidEventEntityType) Valid() bool { switch v { - case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy: + case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeFeedback, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy: return true } return false diff --git a/internal/control/api.go b/internal/control/api.go index 1e83a07..8b4234b 100644 --- a/internal/control/api.go +++ b/internal/control/api.go @@ -23,11 +23,13 @@ import ( type Server struct { revisions *RevisionAPI intents *IntentAPI + pressure *PressureAPI } -// NewServer wires the control APIs. -func NewServer(rev *RevisionAPI, in *IntentAPI) *Server { - return &Server{revisions: rev, intents: in} +// NewServer wires the control APIs. The pressure API may be nil where an +// interface runs without an observation plane. +func NewServer(rev *RevisionAPI, in *IntentAPI, p *PressureAPI) *Server { + return &Server{revisions: rev, intents: in, pressure: p} } // Routes returns the control-plane mux. @@ -40,6 +42,14 @@ func (s *Server) Routes() *http.ServeMux { mux.HandleFunc("/control/v1/intents/", s.intents.handleItem) mux.HandleFunc("/control/v1/intents/active", s.intents.handleActive) + if s.pressure != nil { + mux.HandleFunc("/control/v1/pressure", s.pressure.handleCollection) + mux.HandleFunc("/control/v1/pressure/", s.pressure.handleItem) + mux.HandleFunc("/control/v1/telemetry", s.pressure.handleTelemetry) + // Consumer-reachable, unlike the rest of this surface. + mux.HandleFunc("/v1/feedback", s.pressure.handleFeedback) + } + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) }) diff --git a/internal/control/api_test.go b/internal/control/api_test.go index ca9995e..3073c99 100644 --- a/internal/control/api_test.go +++ b/internal/control/api_test.go @@ -57,7 +57,7 @@ func newServer(t *testing.T) (*http.ServeMux, *evidence.SQLStore) { t.Fatal(err) } - srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate)) + srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate), nil) return srv.Routes(), store } diff --git a/internal/control/pressure.go b/internal/control/pressure.go new file mode 100644 index 0000000..6b31534 --- /dev/null +++ b/internal/control/pressure.go @@ -0,0 +1,171 @@ +package control + +import ( + "encoding/json" + "errors" + "net/http" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/observation" +) + +// PressureAPI implements ArchitectureBlueprint.md section 44.2: record +// pressure, aggregate, link evidence, link hypothesis, close or dismiss. +type PressureAPI struct { + registry *observation.PressureRegistry + ingest *observation.Ingest +} + +// NewPressureAPI returns the pressure API. +func NewPressureAPI(r *observation.PressureRegistry, in *observation.Ingest) *PressureAPI { + return &PressureAPI{registry: r, ingest: in} +} + +func (a *PressureAPI) handleCollection(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + status := contract.FluidPressureStatus(r.URL.Query().Get("status")) + if status != "" && !status.Valid() { + writeError(w, http.StatusBadRequest, "unknown status filter") + return + } + list, err := a.registry.List(r.Context(), status) + if err != nil { + writeError(w, http.StatusInternalServerError, "could not list pressure") + return + } + writeJSON(w, http.StatusOK, map[string]any{"pressures": list}) + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *PressureAPI) handleItem(w http.ResponseWriter, r *http.Request) { + id := pathTail(r.URL.Path, "/control/v1/pressure") + if id == "" { + writeError(w, http.StatusNotFound, "no pressure named") + return + } + + switch r.Method { + case http.MethodGet: + p, err := a.registry.Get(r.Context(), contract.PressureID(id)) + if err != nil { + writeError(w, statusForStoreError(err), "pressure not found") + return + } + writeJSON(w, http.StatusOK, p) + + case http.MethodPatch: + a.patch(w, r, contract.PressureID(id)) + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// PatchPressureRequest changes a pressure's disposition. +type PatchPressureRequest struct { + Status contract.FluidPressureStatus `json:"status,omitempty"` + Reason string `json:"reason,omitempty"` + Actor contract.Actor `json:"actor"` + Hypothesis contract.HypothesisID `json:"link_hypothesis,omitempty"` +} + +func (a *PressureAPI) patch(w http.ResponseWriter, r *http.Request, id contract.PressureID) { + var req PatchPressureRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + // A disposition change with no actor cannot be audited, and dismissals + // are exactly the decisions worth attributing. + writeError(w, http.StatusBadRequest, "a disposition change must name its actor") + return + } + + if req.Hypothesis != "" { + if err := a.registry.LinkHypothesis(r.Context(), id, req.Hypothesis); err != nil { + writeError(w, statusForStoreError(err), err.Error()) + return + } + } + + if req.Status != "" { + if err := a.registry.SetStatus(r.Context(), id, req.Status, req.Actor, req.Reason); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + } + + p, err := a.registry.Get(r.Context(), id) + if err != nil { + writeError(w, statusForStoreError(err), "pressure not found") + return + } + writeJSON(w, http.StatusOK, p) +} + +// handleTelemetry accepts telemetry from out-of-process adapters and consumers. +func (a *PressureAPI) handleTelemetry(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + var batch struct { + Events []contract.FluidTelemetry `json:"events"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&batch); err != nil { + writeError(w, http.StatusBadRequest, "could not decode telemetry", err.Error()) + return + } + + accepted, rejected := a.ingest.WriteBatch(r.Context(), batch.Events) + + causes := make([]string, 0, len(rejected)) + for _, err := range rejected { + causes = append(causes, err.Error()) + } + // Partial acceptance is reported rather than failed: telemetry is + // best-effort evidence and the good events are worth keeping. + writeJSON(w, http.StatusAccepted, map[string]any{ + "accepted": accepted, + "rejected": len(rejected), + "causes": causes, + }) +} + +// handleFeedback accepts explicit consumer feedback. +// +// This endpoint is reachable by consumers, unlike the rest of the control +// plane. What it accepts is evidence, never authority: recording feedback +// creates no pressure and no hypothesis on its own. +func (a *PressureAPI) handleFeedback(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + var f contract.FluidFeedback + if err := decodeBody(r, &f); err != nil { + writeError(w, http.StatusBadRequest, "could not decode feedback", err.Error()) + return + } + + stored, err := a.ingest.RecordFeedback(r.Context(), f) + if err != nil { + if errors.Is(err, observation.ErrWrongInterface) { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + writeJSON(w, http.StatusCreated, map[string]any{ + "id": stored.ID, + "note": "recorded as evidence; feedback does not itself authorize an interface change", + }) +} diff --git a/internal/fitness/fitness.go b/internal/fitness/fitness.go new file mode 100644 index 0000000..c75e052 --- /dev/null +++ b/internal/fitness/fitness.go @@ -0,0 +1,357 @@ +// Package fitness evaluates how well an interface revision fulfils its purpose. +// +// ArchitectureBlueprint.md section 18 requires four kinds of metric to stay +// distinct: primary metrics the hypothesis predicts, guardrails that must not +// regress, secondary observations, and learning signals. Section 48.5 names the +// failure mode this prevents — a single fitness number that hides the +// dimensions and guardrails underneath it. +// +// There is deliberately no universal scalar. FluidAPIStandards.md section 20 +// declines to define one, and a framework that invented one anyway would be +// making the interface's most important trade-offs on the operator's behalf. +package fitness + +import ( + "fmt" + "math" + "sort" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// MetricRole distinguishes what a measurement is for. +type MetricRole string + +const ( + // RolePrimary metrics are the outcomes the hypothesis predicted. + RolePrimary MetricRole = "primary" + // RoleGuardrail metrics must not regress beyond a threshold, whatever the + // primary metrics do. + RoleGuardrail MetricRole = "guardrail" + // RoleSecondary metrics are useful context, not decision inputs. + RoleSecondary MetricRole = "secondary" + // RoleLearning metrics improve future hypothesis formation. + RoleLearning MetricRole = "learning" +) + +// Direction says which way is better for a metric. +type Direction string + +const ( + Lower Direction = "lower" + Higher Direction = "higher" + Unchanged Direction = "unchanged" +) + +// MetricSpec declares how one metric is judged. +type MetricSpec struct { + Name string `json:"name"` + Role MetricRole `json:"role"` + Direction Direction `json:"direction"` + // Target is the value a primary metric must reach. + Target *float64 `json:"target,omitempty"` + // Threshold is the limit a guardrail must not cross. + Threshold *float64 `json:"threshold,omitempty"` +} + +// Window is a measurement period. +// +// Retaining it is not bookkeeping: Blueprint section 18 requires the evaluator +// to keep the baseline and the window, because a comparison whose period is +// unknown cannot be reproduced or challenged. +type Window struct { + Start time.Time `json:"start"` + End *time.Time `json:"end,omitempty"` +} + +// Observation is a metric measured over one revision. +type Observation struct { + Metric string `json:"metric"` + Revision contract.RevisionID `json:"revision"` + Value float64 `json:"value"` + Samples int `json:"samples"` +} + +// MetricResult is one metric compared between control and candidate. +type MetricResult struct { + Name string `json:"name"` + Role MetricRole `json:"role"` + Direction Direction `json:"direction"` + + Baseline float64 `json:"baseline"` + Current float64 `json:"current"` + Delta float64 `json:"delta"` + Target *float64 `json:"target,omitempty"` + Threshold *float64 `json:"threshold,omitempty"` + + // TargetMet applies to primary metrics only. + TargetMet *bool `json:"target_met,omitempty"` + // GuardrailBreached applies to guardrails only. + GuardrailBreached *bool `json:"guardrail_breached,omitempty"` + + BaselineSamples int `json:"baseline_samples"` + CurrentSamples int `json:"current_samples"` + // Underpowered marks a comparison with too little data to lean on. + Underpowered bool `json:"underpowered"` +} + +// Verdict is the evaluator's overall reading. +type Verdict string + +const ( + // VerdictSucceeded: every primary target met, no guardrail breached. + VerdictSucceeded Verdict = "SUCCEEDED" + // VerdictFailed: a primary target was missed without a guardrail breach. + VerdictFailed Verdict = "FAILED" + // VerdictGuardrailBreached: a guardrail regressed, whatever else happened. + VerdictGuardrailBreached Verdict = "GUARDRAIL_BREACHED" + // VerdictInconclusive: not enough evidence to say either way. + VerdictInconclusive Verdict = "INCONCLUSIVE" +) + +// Evaluation is a complete fitness comparison. +type Evaluation struct { + Experiment contract.ExperimentID `json:"experiment,omitempty"` + Control contract.RevisionID `json:"control"` + Candidate contract.RevisionID `json:"candidate"` + Window Window `json:"window"` + + Verdict Verdict `json:"verdict"` + Reasons []string `json:"reasons,omitempty"` + Metrics []MetricResult `json:"metrics"` +} + +// PrimaryResults returns just the primary metrics. +func (e Evaluation) PrimaryResults() []MetricResult { return e.byRole(RolePrimary) } + +// GuardrailResults returns just the guardrails. +func (e Evaluation) GuardrailResults() []MetricResult { return e.byRole(RoleGuardrail) } + +func (e Evaluation) byRole(role MetricRole) []MetricResult { + var out []MetricResult + for _, m := range e.Metrics { + if m.Role == role { + out = append(out, m) + } + } + return out +} + +// Evaluator compares a candidate revision against its control. +type Evaluator struct { + // MinSamples is the per-side sample count below which a comparison is + // marked underpowered. A difference measured on three requests is noise + // wearing a result's clothes. + MinSamples int +} + +// NewEvaluator returns an evaluator with a workable default. +func NewEvaluator() *Evaluator { return &Evaluator{MinSamples: 30} } + +// Evaluate compares observations against the declared specs. +// +// Specs are an input, not something derived from the data. Blueprint section 18 +// says success criteria must not be changed after results are visible without +// recording the amendment, and an evaluator that inferred its own criteria +// would make that impossible to enforce. +func (e *Evaluator) Evaluate( + control, candidate contract.RevisionID, + window Window, + specs []MetricSpec, + observations []Observation, +) Evaluation { + byMetric := map[string]map[contract.RevisionID]Observation{} + for _, o := range observations { + if byMetric[o.Metric] == nil { + byMetric[o.Metric] = map[contract.RevisionID]Observation{} + } + byMetric[o.Metric][o.Revision] = o + } + + eval := Evaluation{Control: control, Candidate: candidate, Window: window} + + ordered := make([]MetricSpec, len(specs)) + copy(ordered, specs) + sort.SliceStable(ordered, func(i, j int) bool { + if ordered[i].Role != ordered[j].Role { + return roleRank(ordered[i].Role) < roleRank(ordered[j].Role) + } + return ordered[i].Name < ordered[j].Name + }) + + var ( + missingPrimary []string + breached []string + underpowered []string + primaryCount int + primaryMetCount int + haveAnyPrimary bool + ) + + for _, spec := range ordered { + base, hasBase := byMetric[spec.Name][control] + cur, hasCur := byMetric[spec.Name][candidate] + + if !hasBase || !hasCur { + // A metric with no measurement on one side is reported as absent + // rather than defaulted to zero, which would read as a dramatic + // improvement or regression that never happened. + eval.Metrics = append(eval.Metrics, MetricResult{ + Name: spec.Name, + Role: spec.Role, + Direction: spec.Direction, + Target: spec.Target, + Threshold: spec.Threshold, + Underpowered: true, + }) + if spec.Role == RolePrimary { + primaryCount++ + underpowered = append(underpowered, + fmt.Sprintf("%s has no measurement on both sides", spec.Name)) + } + continue + } + + r := MetricResult{ + Name: spec.Name, + Role: spec.Role, + Direction: spec.Direction, + Baseline: base.Value, + Current: cur.Value, + Delta: round4(cur.Value - base.Value), + Target: spec.Target, + Threshold: spec.Threshold, + BaselineSamples: base.Samples, + CurrentSamples: cur.Samples, + } + + if base.Samples < e.MinSamples || cur.Samples < e.MinSamples { + r.Underpowered = true + underpowered = append(underpowered, fmt.Sprintf( + "%s has %d control and %d candidate samples, below the %d needed", + spec.Name, base.Samples, cur.Samples, e.MinSamples)) + } + + switch spec.Role { + case RolePrimary: + primaryCount++ + haveAnyPrimary = true + met := meetsTarget(spec, cur.Value) + r.TargetMet = &met + if met { + primaryMetCount++ + } else { + missingPrimary = append(missingPrimary, describeMiss(spec, cur.Value)) + } + + case RoleGuardrail: + crossed := breachesGuardrail(spec, cur.Value) + r.GuardrailBreached = &crossed + if crossed { + breached = append(breached, describeBreach(spec, cur.Value)) + } + } + + eval.Metrics = append(eval.Metrics, r) + } + + // Guardrails dominate. A candidate that hit every target while regressing a + // guardrail has not succeeded; it has traded something it was told not to. + switch { + case len(breached) > 0: + eval.Verdict = VerdictGuardrailBreached + eval.Reasons = breached + + case !haveAnyPrimary || primaryCount == 0: + eval.Verdict = VerdictInconclusive + eval.Reasons = []string{"no primary metric was declared, so there is nothing to conclude"} + + case len(underpowered) > 0 && len(missingPrimary) == 0: + // Every target appears met, but on too little data to act on. Reporting + // success here is how an experiment gets promoted on noise. + eval.Verdict = VerdictInconclusive + eval.Reasons = underpowered + + case len(missingPrimary) > 0: + eval.Verdict = VerdictFailed + eval.Reasons = missingPrimary + + default: + eval.Verdict = VerdictSucceeded + eval.Reasons = []string{fmt.Sprintf("%d of %d primary targets met with no guardrail breach", + primaryMetCount, primaryCount)} + } + + sort.Strings(eval.Reasons) + return eval +} + +func meetsTarget(spec MetricSpec, value float64) bool { + if spec.Target == nil { + // A primary metric with no target cannot be judged, so it is not met. + return false + } + switch spec.Direction { + case Lower: + return value <= *spec.Target + case Higher: + return value >= *spec.Target + case Unchanged: + return math.Abs(value-*spec.Target) < 1e-9 + } + return false +} + +func breachesGuardrail(spec MetricSpec, value float64) bool { + if spec.Threshold == nil { + return false + } + switch spec.Direction { + case Lower: + // Lower is better, so exceeding the threshold is the breach. + return value > *spec.Threshold + case Higher: + return value < *spec.Threshold + case Unchanged: + return math.Abs(value-*spec.Threshold) > 1e-9 + } + return false +} + +func describeMiss(spec MetricSpec, value float64) string { + if spec.Target == nil { + return fmt.Sprintf("%s is a primary metric with no declared target", spec.Name) + } + return fmt.Sprintf("%s reached %.4g, target was %s %.4g", + spec.Name, value, comparator(spec.Direction), *spec.Target) +} + +func describeBreach(spec MetricSpec, value float64) string { + return fmt.Sprintf("guardrail %s at %.4g breached its threshold of %.4g", + spec.Name, value, *spec.Threshold) +} + +func comparator(d Direction) string { + switch d { + case Lower: + return "at most" + case Higher: + return "at least" + } + return "exactly" +} + +func roleRank(r MetricRole) int { + switch r { + case RolePrimary: + return 0 + case RoleGuardrail: + return 1 + case RoleSecondary: + return 2 + } + return 3 +} + +func round4(v float64) float64 { return math.Round(v*10000) / 10000 } diff --git a/internal/fitness/fitness_test.go b/internal/fitness/fitness_test.go new file mode 100644 index 0000000..d10cec0 --- /dev/null +++ b/internal/fitness/fitness_test.go @@ -0,0 +1,325 @@ +package fitness + +import ( + "fmt" + "testing" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +func f64(v float64) *float64 { return &v } + +func obs(metric string, rev contract.RevisionID, value float64, samples int) Observation { + return Observation{Metric: metric, Revision: rev, Value: value, Samples: samples} +} + +// specsFromBlueprint mirrors the section 33 worked example: requests per task +// must drop, latency must not regress past its guardrail. +func specsFromBlueprint() []MetricSpec { + return []MetricSpec{ + {Name: MetricRequestsPerTask, Role: RolePrimary, Direction: Lower, Target: f64(1.2)}, + {Name: MetricP95LatencyMS, Role: RoleGuardrail, Direction: Lower, Threshold: f64(315)}, + {Name: MetricErrorRate, Role: RoleGuardrail, Direction: Lower, Threshold: f64(0.01)}, + } +} + +func TestSucceedsWhenTargetsMetAndGuardrailsHold(t *testing.T) { + e := NewEvaluator() + + eval := e.Evaluate("R-1", "R-2", Window{Start: time.Now().Add(-time.Hour)}, specsFromBlueprint(), + []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + obs(MetricRequestsPerTask, "R-2", 1.15, 400), + obs(MetricP95LatencyMS, "R-1", 300, 400), + obs(MetricP95LatencyMS, "R-2", 302, 400), + obs(MetricErrorRate, "R-1", 0.008, 400), + obs(MetricErrorRate, "R-2", 0.007, 400), + }) + + if eval.Verdict != VerdictSucceeded { + t.Fatalf("verdict = %s, reasons %v", eval.Verdict, eval.Reasons) + } + + primary := eval.PrimaryResults() + if len(primary) != 1 { + t.Fatalf("primary metrics = %d, want 1", len(primary)) + } + if primary[0].TargetMet == nil || !*primary[0].TargetMet { + t.Error("primary target not marked as met") + } + if primary[0].Delta >= 0 { + t.Errorf("delta = %v, expected a reduction", primary[0].Delta) + } +} + +// TestGuardrailDominatesPrimarySuccess is the section 48.5 protection: a +// candidate that hit its target while regressing a guardrail has traded +// something it was told not to. +func TestGuardrailDominatesPrimarySuccess(t *testing.T) { + e := NewEvaluator() + + eval := e.Evaluate("R-1", "R-2", Window{}, specsFromBlueprint(), []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + obs(MetricRequestsPerTask, "R-2", 1.05, 400), // target smashed + obs(MetricP95LatencyMS, "R-1", 300, 400), + obs(MetricP95LatencyMS, "R-2", 980, 400), // and latency ruined + obs(MetricErrorRate, "R-1", 0.008, 400), + obs(MetricErrorRate, "R-2", 0.009, 400), + }) + + if eval.Verdict != VerdictGuardrailBreached { + t.Fatalf("verdict = %s, want GUARDRAIL_BREACHED; reasons %v", eval.Verdict, eval.Reasons) + } + if len(eval.Reasons) == 0 { + t.Error("a breach was reported with no reason") + } + + var latency *MetricResult + for i, m := range eval.GuardrailResults() { + if m.Name == MetricP95LatencyMS { + latency = &eval.GuardrailResults()[i] + } + } + if latency == nil || latency.GuardrailBreached == nil || !*latency.GuardrailBreached { + t.Error("the breached guardrail is not marked as breached") + } +} + +func TestFailsWhenPrimaryTargetMissed(t *testing.T) { + e := NewEvaluator() + + eval := e.Evaluate("R-1", "R-2", Window{}, specsFromBlueprint(), []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + obs(MetricRequestsPerTask, "R-2", 2.6, 400), // barely moved + obs(MetricP95LatencyMS, "R-1", 300, 400), + obs(MetricP95LatencyMS, "R-2", 301, 400), + obs(MetricErrorRate, "R-1", 0.008, 400), + obs(MetricErrorRate, "R-2", 0.008, 400), + }) + + if eval.Verdict != VerdictFailed { + t.Fatalf("verdict = %s, want FAILED", eval.Verdict) + } + if len(eval.Reasons) == 0 || eval.Reasons[0] == "" { + t.Error("failure reported without saying which target was missed") + } +} + +// TestUnderpoweredIsInconclusiveNotSuccessful: promoting on three requests is +// promoting on noise. +func TestUnderpoweredIsInconclusiveNotSuccessful(t *testing.T) { + e := NewEvaluator() + + eval := e.Evaluate("R-1", "R-2", Window{}, specsFromBlueprint(), []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 3), + obs(MetricRequestsPerTask, "R-2", 1.0, 2), + obs(MetricP95LatencyMS, "R-1", 300, 3), + obs(MetricP95LatencyMS, "R-2", 290, 2), + obs(MetricErrorRate, "R-1", 0.0, 3), + obs(MetricErrorRate, "R-2", 0.0, 2), + }) + + if eval.Verdict != VerdictInconclusive { + t.Fatalf("verdict = %s, want INCONCLUSIVE on tiny samples", eval.Verdict) + } + for _, m := range eval.PrimaryResults() { + if !m.Underpowered { + t.Error("a two-sample comparison was not marked underpowered") + } + } +} + +// TestMissingMeasurementIsNotZero: defaulting an absent metric to zero would +// read as a dramatic change that never happened. +func TestMissingMeasurementIsNotZero(t *testing.T) { + e := NewEvaluator() + + eval := e.Evaluate("R-1", "R-2", Window{}, specsFromBlueprint(), []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + // no candidate measurement at all + obs(MetricP95LatencyMS, "R-1", 300, 400), + obs(MetricP95LatencyMS, "R-2", 300, 400), + obs(MetricErrorRate, "R-1", 0.008, 400), + obs(MetricErrorRate, "R-2", 0.008, 400), + }) + + if eval.Verdict == VerdictSucceeded { + t.Fatal("a missing primary measurement produced a success verdict") + } + for _, m := range eval.PrimaryResults() { + if m.Current != 0 && m.Baseline != 0 { + continue + } + if !m.Underpowered { + t.Error("an absent measurement was not flagged") + } + } +} + +func TestNoPrimaryMetricIsInconclusive(t *testing.T) { + e := NewEvaluator() + eval := e.Evaluate("R-1", "R-2", Window{}, []MetricSpec{ + {Name: MetricErrorRate, Role: RoleGuardrail, Direction: Lower, Threshold: f64(0.05)}, + }, []Observation{ + obs(MetricErrorRate, "R-1", 0.01, 400), + obs(MetricErrorRate, "R-2", 0.01, 400), + }) + + if eval.Verdict != VerdictInconclusive { + t.Errorf("verdict = %s, want INCONCLUSIVE with no primary metric", eval.Verdict) + } +} + +func TestPrimaryWithoutTargetCannotPass(t *testing.T) { + e := NewEvaluator() + eval := e.Evaluate("R-1", "R-2", Window{}, []MetricSpec{ + {Name: MetricRequestsPerTask, Role: RolePrimary, Direction: Lower}, // no target + }, []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + obs(MetricRequestsPerTask, "R-2", 1.0, 400), + }) + + if eval.Verdict == VerdictSucceeded { + t.Error("a primary metric with no declared target was treated as met") + } +} + +func TestHigherIsBetterDirection(t *testing.T) { + e := NewEvaluator() + eval := e.Evaluate("R-1", "R-2", Window{}, []MetricSpec{ + {Name: MetricSuccessRate, Role: RolePrimary, Direction: Higher, Target: f64(0.99)}, + {Name: MetricSuccessRate + "_guard", Role: RoleGuardrail, Direction: Higher, Threshold: f64(0.95)}, + }, []Observation{ + obs(MetricSuccessRate, "R-1", 0.97, 400), + obs(MetricSuccessRate, "R-2", 0.995, 400), + obs(MetricSuccessRate+"_guard", "R-1", 0.97, 400), + obs(MetricSuccessRate+"_guard", "R-2", 0.96, 400), + }) + + if eval.Verdict != VerdictSucceeded { + t.Errorf("verdict = %s, reasons %v", eval.Verdict, eval.Reasons) + } +} + +func TestEvaluationIsDeterministic(t *testing.T) { + e := NewEvaluator() + specs := specsFromBlueprint() + observations := []Observation{ + obs(MetricRequestsPerTask, "R-1", 2.7, 400), + obs(MetricRequestsPerTask, "R-2", 2.6, 400), + obs(MetricP95LatencyMS, "R-1", 300, 400), + obs(MetricP95LatencyMS, "R-2", 999, 400), + obs(MetricErrorRate, "R-1", 0.008, 400), + obs(MetricErrorRate, "R-2", 0.5, 400), + } + + first := e.Evaluate("R-1", "R-2", Window{}, specs, observations) + for i := 0; i < 50; i++ { + again := e.Evaluate("R-1", "R-2", Window{}, specs, observations) + if again.Verdict != first.Verdict || len(again.Reasons) != len(first.Reasons) { + t.Fatal("evaluation varied between runs") + } + for j := range first.Reasons { + if again.Reasons[j] != first.Reasons[j] { + t.Fatalf("reason order varied: %q vs %q", first.Reasons[j], again.Reasons[j]) + } + } + for j := range first.Metrics { + if again.Metrics[j].Name != first.Metrics[j].Name { + t.Fatal("metric order varied between runs") + } + } + } +} + +func TestMeasurerDerivesRequestsPerTask(t *testing.T) { + m := NewMeasurer() + now := time.Date(2026, 9, 5, 8, 0, 0, 0, time.UTC) + + var events []contract.FluidTelemetry + // R-1: three calls per task. R-2: one. + for c := 0; c < 10; c++ { + chain := fmt.Sprintf("chain-%d", c) + for i := 0; i < 3; i++ { + events = append(events, telemetry("R-1", fmt.Sprintf("c-%d", c), chain, now, 120, false)) + } + events = append(events, telemetry("R-2", fmt.Sprintf("c-%d", c), chain+"-b", now, 130, false)) + } + + observations := m.Measure(events, Window{Start: now.Add(-time.Hour)}) + + got := map[contract.RevisionID]float64{} + for _, o := range observations { + if o.Metric == MetricRequestsPerTask { + got[o.Revision] = o.Value + } + } + if got["R-1"] != 3 { + t.Errorf("R-1 requests per task = %v, want 3", got["R-1"]) + } + if got["R-2"] != 1 { + t.Errorf("R-2 requests per task = %v, want 1", got["R-2"]) + } +} + +func TestMeasurerRespectsWindow(t *testing.T) { + m := NewMeasurer() + now := time.Date(2026, 9, 5, 8, 0, 0, 0, time.UTC) + + events := []contract.FluidTelemetry{ + telemetry("R-1", "c-1", "old", now.AddDate(0, 0, -10), 100, false), + telemetry("R-1", "c-1", "new", now, 100, false), + } + + // Only events inside the window may count, or the comparison is not + // reproducible from its recorded period. + observations := m.Measure(events, Window{Start: now.Add(-time.Hour)}) + for _, o := range observations { + if o.Metric == MetricErrorRate && o.Samples != 1 { + t.Errorf("samples = %d, want 1; events outside the window were counted", o.Samples) + } + } +} + +func TestMeasurerComputesErrorRate(t *testing.T) { + m := NewMeasurer() + now := time.Date(2026, 9, 5, 8, 0, 0, 0, time.UTC) + + var events []contract.FluidTelemetry + for i := 0; i < 8; i++ { + events = append(events, telemetry("R-1", "c-1", fmt.Sprintf("ch-%d", i), now, 100, false)) + } + for i := 0; i < 2; i++ { + events = append(events, telemetry("R-1", "c-1", fmt.Sprintf("er-%d", i), now, 100, true)) + } + + for _, o := range m.Measure(events, Window{}) { + if o.Metric == MetricErrorRate && o.Value != 0.2 { + t.Errorf("error rate = %v, want 0.2", o.Value) + } + if o.Metric == MetricSuccessRate && o.Value != 0.8 { + t.Errorf("success rate = %v, want 0.8", o.Value) + } + } +} + +func telemetry(rev contract.RevisionID, consumer, chain string, at time.Time, latency float64, isError bool) contract.FluidTelemetry { + r := rev + l := latency + status := int64(200) + ev := contract.FluidTelemetry{ + ID: fmt.Sprintf("tl-%s-%s", consumer, chain), + OccurredAt: at, + Kind: contract.FluidTelemetryKindRequest, + ConsumerRef: consumer, + Revision: &r, + Sequence: &contract.FluidTelemetrySequence{ChainID: chain}, + Request: &contract.FluidTelemetryRequest{Route: "/v1/x", Method: "GET", Status: &status, LatencyMS: &l}, + } + if isError { + ev.Kind = contract.FluidTelemetryKindError + ev.Error = &contract.FluidTelemetryError{Class: contract.FluidTelemetryErrorClassBackendFailure} + } + return ev +} diff --git a/internal/fitness/measure.go b/internal/fitness/measure.go new file mode 100644 index 0000000..bc39537 --- /dev/null +++ b/internal/fitness/measure.go @@ -0,0 +1,164 @@ +package fitness + +import ( + "sort" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Standard metric names the measurer derives from telemetry. +// +// These are the metrics the Blueprint's worked example turns on, and having +// them named in one place stops a hypothesis and an evaluation from measuring +// subtly different things under the same word. +const ( + MetricRequestsPerTask = "requests_per_completed_task" + MetricP95LatencyMS = "p95_latency_ms" + MetricErrorRate = "error_rate" + MetricSuccessRate = "success_rate" +) + +// Measurer derives metric observations from raw telemetry. +type Measurer struct { + // ChainGap bounds one completed task when the consumer supplies no chain + // id, matching the topology analyzer's grouping. + ChainGap time.Duration +} + +// NewMeasurer returns a measurer with the default grouping window. +func NewMeasurer() *Measurer { return &Measurer{ChainGap: 30 * time.Second} } + +// Measure computes the standard metrics per revision over a window. +// +// Only events inside the window count. Blueprint section 18 requires the +// measurement window to be retained, and quietly including events outside it +// would make a comparison irreproducible. +func (m *Measurer) Measure(events []contract.FluidTelemetry, window Window) []Observation { + type acc struct { + requests int + errors int + tasks map[string]int + latency []float64 + } + byRevision := map[contract.RevisionID]*acc{} + + for _, ev := range events { + if !inWindow(ev.OccurredAt, window) { + continue + } + if ev.Revision == nil { + continue + } + rev := *ev.Revision + + a, ok := byRevision[rev] + if !ok { + a = &acc{tasks: map[string]int{}} + byRevision[rev] = a + } + + a.requests++ + if ev.Error != nil { + a.errors++ + } + if ev.Request != nil && ev.Request.LatencyMS != nil { + a.latency = append(a.latency, *ev.Request.LatencyMS) + } + a.tasks[taskKey(ev, m.ChainGap)]++ + } + + revisions := make([]contract.RevisionID, 0, len(byRevision)) + for r := range byRevision { + revisions = append(revisions, r) + } + sort.Slice(revisions, func(i, j int) bool { return revisions[i] < revisions[j] }) + + var out []Observation + for _, rev := range revisions { + a := byRevision[rev] + if a.requests == 0 { + continue + } + + // Requests per completed task is the metric that catches an interface + // making consumers assemble what it could have handed them. + tasks := len(a.tasks) + if tasks > 0 { + out = append(out, Observation{ + Metric: MetricRequestsPerTask, + Revision: rev, + Value: round4(float64(a.requests) / float64(tasks)), + Samples: tasks, + }) + } + + out = append(out, Observation{ + Metric: MetricErrorRate, + Revision: rev, + Value: round4(float64(a.errors) / float64(a.requests)), + Samples: a.requests, + }) + out = append(out, Observation{ + Metric: MetricSuccessRate, + Revision: rev, + Value: round4(float64(a.requests-a.errors) / float64(a.requests)), + Samples: a.requests, + }) + + if len(a.latency) > 0 { + out = append(out, Observation{ + Metric: MetricP95LatencyMS, + Revision: rev, + Value: round4(percentile(a.latency, 0.95)), + Samples: len(a.latency), + }) + } + } + return out +} + +// taskKey groups events into completed tasks. +func taskKey(ev contract.FluidTelemetry, gap time.Duration) string { + consumer := ev.ConsumerRef + if consumer == "" { + consumer = ev.CorrelationID + } + if ev.Sequence != nil && ev.Sequence.ChainID != "" { + return consumer + "/" + ev.Sequence.ChainID + } + // Without a chain id, bucket by consumer and elapsed gap. This is a + // heuristic, which is the reason to prefer chain ids from agentic + // consumers where they can supply them. + bucket := ev.OccurredAt.Truncate(gap).UTC().Format(time.RFC3339) + return consumer + "/" + bucket +} + +func inWindow(t time.Time, w Window) bool { + if !w.Start.IsZero() && t.Before(w.Start) { + return false + } + if w.End != nil && t.After(*w.End) { + return false + } + return true +} + +// percentile returns the nearest-rank percentile of values. +func percentile(values []float64, p float64) float64 { + sorted := make([]float64, len(values)) + copy(sorted, values) + sort.Float64s(sorted) + + if len(sorted) == 1 { + return sorted[0] + } + rank := int(p*float64(len(sorted)-1) + 0.5) + if rank < 0 { + rank = 0 + } + if rank >= len(sorted) { + rank = len(sorted) - 1 + } + return sorted[rank] +} 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/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/feedback.go b/internal/observation/feedback.go new file mode 100644 index 0000000..0ed3885 --- /dev/null +++ b/internal/observation/feedback.go @@ -0,0 +1,22 @@ +package observation + +import ( + "encoding/json" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// consumerLabel names the source of a piece of feedback for the audit trail. +// +// It prefers the cohort over anything consumer-specific: the trail needs to +// know what kind of consumer said this, not which one. +func consumerLabel(f contract.FluidFeedback) string { + if f.Cohort != nil && *f.Cohort != "" { + return string(*f.Cohort) + } + return "unclassified" +} + +func marshalFeedback(f contract.FluidFeedback) ([]byte, error) { + return json.Marshal(contract.FeedbackDocument{FluidFeedback: f}) +} diff --git a/internal/observation/ingest.go b/internal/observation/ingest.go new file mode 100644 index 0000000..e8dfc23 --- /dev/null +++ b/internal/observation/ingest.go @@ -0,0 +1,191 @@ +package observation + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +// Ingest normalizes, redacts and stores telemetry. +// +// Redaction happens here rather than at the query side because the evidence +// store is append-only: anything written unredacted stays unredacted forever. +// The filter belongs on the way in, where there is still a decision to make. +type Ingest struct { + store evidence.Store + policy RedactionPolicy + iface contract.InterfaceID + now func() time.Time +} + +// NewIngest returns an ingest pipeline. +func NewIngest(store evidence.Store, iface contract.InterfaceID, policy RedactionPolicy) (*Ingest, error) { + if err := policy.Validate(); err != nil { + return nil, err + } + return &Ingest{store: store, policy: policy, iface: iface, now: time.Now}, nil +} + +// ErrWrongInterface reports telemetry submitted for another interface. +var ErrWrongInterface = errors.New("telemetry belongs to a different interface") + +// Normalize fills in defaults and applies redaction. +// +// It is separated from Write so the same normalization runs whether an event +// arrives from the in-process emitter or over the ingest endpoint. Two paths +// with two normalizations would eventually disagree, and the disagreement would +// surface as a pressure finding that is really a bug in the pipeline. +func (i *Ingest) Normalize(ev contract.FluidTelemetry) (contract.FluidTelemetry, error) { + if ev.InterfaceID == "" { + ev.InterfaceID = i.iface + } + if ev.InterfaceID != i.iface { + return ev, fmt.Errorf("%w: event is for %q, ingest serves %q", + ErrWrongInterface, ev.InterfaceID, i.iface) + } + + if ev.SchemaVersion == "" { + ev.SchemaVersion = "0.1" + } + if ev.ID == "" { + ev.ID = newID("tl-") + } + if ev.OccurredAt.IsZero() { + ev.OccurredAt = i.now().UTC() + } + ev.OccurredAt = ev.OccurredAt.UTC() + + if ev.Kind == "" { + // Infer from shape rather than defaulting to "request": an error event + // filed as a request would understate the interface's failure rate. + switch { + case ev.Error != nil: + ev.Kind = contract.FluidTelemetryKindError + case ev.Adoption != nil: + ev.Kind = contract.FluidTelemetryKindAdoption + case ev.Sequence != nil: + ev.Kind = contract.FluidTelemetryKindSequence + default: + ev.Kind = contract.FluidTelemetryKindRequest + } + } + if !ev.Kind.Valid() { + return ev, fmt.Errorf("unknown telemetry kind %q", ev.Kind) + } + + // Raw payload capture is not the default (Blueprint 6.2). Where a policy + // forbids it, error detail is the only free-text field that survives, and + // it is scrubbed below. + if !i.policy.AllowRawPayload && ev.Error != nil && ev.Error.Detail != "" { + if len(ev.Error.Detail) > 512 { + ev.Error.Detail = ev.Error.Detail[:512] + "…" + } + } + + i.policy.Apply(&ev) + return ev, nil +} + +// Write normalizes and stores one event. +func (i *Ingest) Write(ctx context.Context, ev contract.FluidTelemetry) error { + normalized, err := i.Normalize(ev) + if err != nil { + return err + } + return i.store.WriteTelemetry(ctx, normalized) +} + +// WriteBatch stores several events, reporting how many landed. +// +// A malformed event in a batch does not discard the rest. Telemetry is +// best-effort evidence, and dropping a hundred good events because one was +// wrong would lose more than it protects. +func (i *Ingest) WriteBatch(ctx context.Context, events []contract.FluidTelemetry) (accepted int, rejected []error) { + for _, ev := range events { + if err := i.Write(ctx, ev); err != nil { + rejected = append(rejected, err) + continue + } + accepted++ + } + return accepted, rejected +} + +// RecordFeedback stores explicit consumer feedback. +// +// FluidAPIStandards.md section 15: feedback is evidence and must not itself +// authorize interface changes. It is stored as a record and an event, and +// nothing here creates pressure or a hypothesis from it — that stays a +// deliberate step someone takes. +func (i *Ingest) RecordFeedback(ctx context.Context, f contract.FluidFeedback) (contract.FluidFeedback, error) { + if f.Goal == "" { + // Feedback with no stated goal cannot be interpreted later; the goal is + // the part that says what the consumer was actually trying to do. + return f, errors.New("feedback must state a goal") + } + + if f.SchemaVersion == "" { + f.SchemaVersion = "0.1" + } + if f.ID == "" { + f.ID = contract.FeedbackID(newID("F-")) + } + if err := contract.RequireKind(string(f.ID), contract.KindFeedback); err != nil { + return f, err + } + if f.InterfaceID == "" { + f.InterfaceID = i.iface + } + if f.InterfaceID != i.iface { + return f, fmt.Errorf("%w: feedback is for %q", ErrWrongInterface, f.InterfaceID) + } + if f.ReceivedAt.IsZero() { + f.ReceivedAt = i.now().UTC() + } + + // Consumers write free text; it passes the same filter as everything else. + f.Goal, _ = i.policy.Scrub(f.Goal) + f.Outcome, _ = i.policy.Scrub(f.Outcome) + f.Attempt, _ = i.policy.Scrub(f.Attempt) + f.MissingCapability, _ = i.policy.Scrub(f.MissingCapability) + + body, err := marshalFeedback(f) + if err != nil { + return f, err + } + if err := i.store.PutRecord(ctx, contract.KindFeedback, string(f.ID), body); err != nil { + return f, err + } + + if err := i.store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-%s-%d", f.ID, i.now().UnixNano())), + OccurredAt: f.ReceivedAt, + EntityType: contract.FluidEventEntityTypeFeedback, + EntityID: string(f.ID), + EventType: "FEEDBACK_RECEIVED", + // The consumer is the actor, and a consumer is untrusted (Blueprint 47). + // Recording them as the actor keeps that visible in the audit trail. + Actor: contract.Actor{Type: contract.ActorTypeConsumer, ID: consumerLabel(f)}, + Reason: f.Goal, + }); err != nil { + return f, err + } + + return f, nil +} + +// newID generates an identifier with the given prefix. +func newID(prefix string) string { + var b [10]byte + if _, err := rand.Read(b[:]); err != nil { + return prefix + "0" + } + return prefix + hex.EncodeToString(b[:]) +} diff --git a/internal/observation/ingest_test.go b/internal/observation/ingest_test.go new file mode 100644 index 0000000..99d1f98 --- /dev/null +++ b/internal/observation/ingest_test.go @@ -0,0 +1,264 @@ +package observation + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +func newIngest(t *testing.T) (*Ingest, *evidence.SQLStore) { + t.Helper() + store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + + in, err := NewIngest(store, "hall-publishing", testPolicy()) + if err != nil { + t.Fatal(err) + } + return in, store +} + +func TestIngestRejectsUnsaltedPolicy(t *testing.T) { + store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db")) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + if _, err := NewIngest(store, "x", RedactionPolicy{}); !errors.Is(err, ErrNoSalt) { + t.Errorf("an unsalted ingest was constructed: %v", err) + } +} + +// TestRedactionHappensOnTheWayIn: the store is append-only, so anything written +// unredacted stays that way forever. +func TestRedactionHappensOnTheWayIn(t *testing.T) { + ctx := context.Background() + in, store := newIngest(t) + + err := in.Write(ctx, contract.FluidTelemetry{ + ConsumerRef: "bernd@example.com", + Request: &contract.FluidTelemetryRequest{Route: "/v1/entries?token=hunter2", Method: "GET"}, + }) + if err != nil { + t.Fatal(err) + } + + rows, err := store.Telemetry(ctx, evidence.TelemetryFilter{}) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 { + t.Fatalf("stored %d rows, want 1", len(rows)) + } + + stored := rows[0] + if strings.Contains(stored.ConsumerRef, "@") { + t.Errorf("raw consumer identity reached the store: %q", stored.ConsumerRef) + } + if strings.Contains(stored.Request.Route, "hunter2") { + t.Errorf("credential reached the store: %q", stored.Request.Route) + } + if stored.Redaction == nil || !stored.Redaction.Applied { + t.Error("redaction was applied but not recorded on the stored event") + } +} + +// TestKindIsInferredFromShape: an error event filed as a request would +// understate the interface's failure rate. +func TestKindIsInferredFromShape(t *testing.T) { + in, _ := newIngest(t) + + got, err := in.Normalize(contract.FluidTelemetry{ + Error: &contract.FluidTelemetryError{Class: contract.FluidTelemetryErrorClassTimeout}, + }) + if err != nil { + t.Fatal(err) + } + if got.Kind != contract.FluidTelemetryKindError { + t.Errorf("kind = %s, want error", got.Kind) + } + + got, err = in.Normalize(contract.FluidTelemetry{ + Adoption: &contract.FluidTelemetryAdoption{Event: contract.FluidTelemetryAdoptionEventFirstUse}, + }) + if err != nil { + t.Fatal(err) + } + if got.Kind != contract.FluidTelemetryKindAdoption { + t.Errorf("kind = %s, want adoption", got.Kind) + } +} + +func TestNormalizeFillsDefaultsAndGuardsInterface(t *testing.T) { + in, _ := newIngest(t) + + got, err := in.Normalize(contract.FluidTelemetry{}) + if err != nil { + t.Fatal(err) + } + if got.ID == "" || got.OccurredAt.IsZero() || got.SchemaVersion == "" { + t.Errorf("defaults not filled: %+v", got) + } + if got.InterfaceID != "hall-publishing" { + t.Errorf("interface = %q", got.InterfaceID) + } + + // Another interface's telemetry must not land in this evidence store. + if _, err := in.Normalize(contract.FluidTelemetry{InterfaceID: "some-other-api"}); !errors.Is(err, ErrWrongInterface) { + t.Errorf("foreign telemetry accepted: %v", err) + } +} + +// TestBatchKeepsGoodEventsWhenOneIsBad: dropping a hundred good events because +// one was malformed loses more than it protects. +func TestBatchKeepsGoodEventsWhenOneIsBad(t *testing.T) { + ctx := context.Background() + in, store := newIngest(t) + + accepted, rejected := in.WriteBatch(ctx, []contract.FluidTelemetry{ + {Request: &contract.FluidTelemetryRequest{Route: "/a"}}, + {InterfaceID: "wrong-interface"}, + {Request: &contract.FluidTelemetryRequest{Route: "/b"}}, + }) + + if accepted != 2 { + t.Errorf("accepted = %d, want 2", accepted) + } + if len(rejected) != 1 { + t.Errorf("rejected = %d, want 1", len(rejected)) + } + + rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{}) + if len(rows) != 2 { + t.Errorf("stored %d rows, want 2", len(rows)) + } +} + +func TestErrorDetailIsBoundedWhenRawPayloadIsOff(t *testing.T) { + in, _ := newIngest(t) + + long := strings.Repeat("x", 2000) + got, err := in.Normalize(contract.FluidTelemetry{ + Error: &contract.FluidTelemetryError{ + Class: contract.FluidTelemetryErrorClassBackendFailure, + Detail: long, + }, + }) + if err != nil { + t.Fatal(err) + } + if len(got.Error.Detail) > 600 { + t.Errorf("error detail was not bounded: %d chars", len(got.Error.Detail)) + } +} + +func TestFeedbackIsStoredAsEvidence(t *testing.T) { + ctx := context.Background() + in, store := newIngest(t) + + cohort := contract.CohortID("coding-agents") + got, err := in.RecordFeedback(ctx, contract.FluidFeedback{ + Cohort: &cohort, + Goal: "publish a hall entry without splitting it by hand", + MissingCapability: "long-form serialization", + Outcome: "capability unavailable, contact bernd@example.com", + }) + if err != nil { + t.Fatal(err) + } + + if got.ID == "" || !strings.HasPrefix(string(got.ID), "F-") { + t.Errorf("feedback id = %q, want an F- prefix", got.ID) + } + // Consumers write free text; it passes the same filter as everything else. + if strings.Contains(got.Outcome, "bernd@example.com") { + t.Errorf("an address survived in feedback: %q", got.Outcome) + } + + if _, err := store.Record(ctx, contract.KindFeedback, string(got.ID)); err != nil { + t.Errorf("feedback was not persisted: %v", err) + } + + events, err := store.Events(ctx, evidence.EventFilter{EntityID: string(got.ID)}) + if err != nil { + t.Fatal(err) + } + if len(events) != 1 { + t.Fatalf("feedback left %d events, want 1", len(events)) + } + // A consumer is untrusted; recording them as the actor keeps that visible. + if events[0].Actor.Type != contract.ActorTypeConsumer { + t.Errorf("actor type = %s, want consumer", events[0].Actor.Type) + } +} + +func TestFeedbackNeedsAGoal(t *testing.T) { + in, _ := newIngest(t) + // Without a goal there is nothing to interpret later. + if _, err := in.RecordFeedback(context.Background(), contract.FluidFeedback{ + Outcome: "it did not work", + }); err == nil { + t.Error("feedback with no goal was accepted") + } +} + +// TestFeedbackDoesNotCreatePressure is the section 15 boundary: feedback is +// evidence and must not itself authorize a change. +func TestFeedbackDoesNotCreatePressure(t *testing.T) { + ctx := context.Background() + in, store := newIngest(t) + + if _, err := in.RecordFeedback(ctx, contract.FluidFeedback{ + Goal: "I need a bulk publish endpoint", + MissingCapability: "bulk publish", + }); err != nil { + t.Fatal(err) + } + + records, err := store.Records(ctx, contract.KindPressure) + if err != nil { + t.Fatal(err) + } + if len(records) != 0 { + t.Errorf("feedback created %d pressure records on its own", len(records)) + } +} + +func TestIngestIsUsableAsARuntimeSink(t *testing.T) { + ctx := context.Background() + in, store := newIngest(t) + + // The gateway's emitter writes through this interface; redaction must apply + // on that path too, not only to events arriving over the endpoint. + var sink interface { + Write(context.Context, contract.FluidTelemetry) error + } = in + + if err := sink.Write(ctx, contract.FluidTelemetry{ + ConsumerRef: "raw-identity", + OccurredAt: time.Now(), + Request: &contract.FluidTelemetryRequest{Route: "/v1/x"}, + }); err != nil { + t.Fatal(err) + } + + rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{}) + if len(rows) != 1 { + t.Fatalf("stored %d rows", len(rows)) + } + if rows[0].ConsumerRef == "raw-identity" { + t.Error("the emitter path bypassed redaction") + } +} diff --git a/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/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/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 +} 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") + } +} diff --git a/schemas/event.schema.json b/schemas/event.schema.json index cf2a3a3..2cb1db5 100644 --- a/schemas/event.schema.json +++ b/schemas/event.schema.json @@ -8,28 +8,70 @@ "fluid_event": { "type": "object", "properties": { - "schema_version": { "$ref": "common.schema.json#/$defs/schemaVersion" }, - "id": { "$ref": "common.schema.json#/$defs/eventId" }, - "occurred_at": { "$ref": "common.schema.json#/$defs/timestamp" }, - "entity_type": { "enum": ["pressure", "hypothesis", "revision", "experiment", "backend_requirement", "intent", "decision", "routing_policy"] }, - "entity_id": { "type": "string", "minLength": 1 }, - "event_type": { "type": "string", "minLength": 1 }, - "actor": { "$ref": "common.schema.json#/$defs/actor" }, + "schema_version": { + "$ref": "common.schema.json#/$defs/schemaVersion" + }, + "id": { + "$ref": "common.schema.json#/$defs/eventId" + }, + "occurred_at": { + "$ref": "common.schema.json#/$defs/timestamp" + }, + "entity_type": { + "enum": [ + "pressure", + "hypothesis", + "revision", + "experiment", + "feedback", + "backend_requirement", + "intent", + "decision", + "routing_policy" + ] + }, + "entity_id": { + "type": "string", + "minLength": 1 + }, + "event_type": { + "type": "string", + "minLength": 1 + }, + "actor": { + "$ref": "common.schema.json#/$defs/actor" + }, "inputs": { "type": "array", - "items": { "type": "string" }, + "items": { + "type": "string" + }, "description": "Identifiers of the records this transition drew on." }, - "reason": { "type": "string" }, + "reason": { + "type": "string" + }, "evidence_refs": { "type": "array", - "items": { "$ref": "common.schema.json#/$defs/evidenceRef" } + "items": { + "$ref": "common.schema.json#/$defs/evidenceRef" + } } }, - "required": ["schema_version", "id", "occurred_at", "entity_type", "entity_id", "event_type", "actor"], + "required": [ + "schema_version", + "id", + "occurred_at", + "entity_type", + "entity_id", + "event_type", + "actor" + ], "additionalProperties": false } }, - "required": ["fluid_event"], + "required": [ + "fluid_event" + ], "additionalProperties": false } diff --git a/workplans/FLUID-WP-0005-fluid-insight.md b/workplans/FLUID-WP-0005-fluid-insight.md index a2e46f0..1349c46 100644 --- a/workplans/FLUID-WP-0005-fluid-insight.md +++ b/workplans/FLUID-WP-0005-fluid-insight.md @@ -4,7 +4,7 @@ type: workplan title: "FLUID Insight - observation plane (Blueprint Phase B)" domain: infotech repo: fluid-core -status: active +status: done owner: worsch topic_slug: fluid-core created: "2026-09-04" @@ -25,7 +25,7 @@ model inference anywhere in this workplan. ```task id: FLUID-WP-0005-T01 -status: todo +status: done priority: high state_hub_task_id: "0c074022-031b-5049-832f-1fa8b1ff8b4f" ``` @@ -37,7 +37,7 @@ normalized interaction event (Blueprint §6.1). ```task id: FLUID-WP-0005-T02 -status: todo +status: done priority: high state_hub_task_id: "13086f4a-217c-5999-a039-90f5d84fb53b" ``` @@ -50,7 +50,7 @@ FLUID learns about the interface, not about people. ```task id: FLUID-WP-0005-T03 -status: todo +status: done priority: medium state_hub_task_id: "815765d4-8d20-5c6f-87ac-e085778cd58c" ``` @@ -62,7 +62,7 @@ not be more specific than the analysis requires. ```task id: FLUID-WP-0005-T04 -status: todo +status: done priority: high state_hub_task_id: "27726efa-e054-56c2-a6b8-3cce8f2e85e7" ``` @@ -74,7 +74,7 @@ error counting. ```task id: FLUID-WP-0005-T05 -status: todo +status: done priority: high state_hub_task_id: "d2e4f2df-809f-5817-9b74-e1533f8f71d9" ``` @@ -86,7 +86,7 @@ in every case. ```task id: FLUID-WP-0005-T06 -status: todo +status: done priority: high state_hub_task_id: "50bd30a6-ae29-5e2b-9618-a3efe6a362b3" ``` @@ -99,7 +99,7 @@ deserves adaptation. ```task id: FLUID-WP-0005-T07 -status: todo +status: done priority: high state_hub_task_id: "ebecef59-2ff8-59c0-8cff-cde53030d0c8" ``` @@ -111,7 +111,7 @@ distinct. Baseline and measurement window are retained, never recomputed. ```task id: FLUID-WP-0005-T08 -status: todo +status: done priority: medium state_hub_task_id: "48f6bbce-2ae8-5b72-8fdc-e962c6ea1f54" ``` @@ -123,7 +123,7 @@ treated as authority to change anything. ```task id: FLUID-WP-0005-T09 -status: todo +status: done priority: medium state_hub_task_id: "ba413808-346b-5026-bfce-aec0eae0c410" ```