diff --git a/cmd/fluid-control/main.go b/cmd/fluid-control/main.go index a90fed7..e0c2fed 100644 --- a/cmd/fluid-control/main.go +++ b/cmd/fluid-control/main.go @@ -30,7 +30,6 @@ 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" @@ -50,8 +49,6 @@ 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() @@ -86,33 +83,9 @@ 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), - pressureAPI, - ).Routes(), + Addr: *addr, + Handler: control.NewServer(control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate)).Routes(), ReadHeaderTimeout: 10 * time.Second, } diff --git a/cmd/fluid/insight.go b/cmd/fluid/insight.go deleted file mode 100644 index a57d2b8..0000000 --- a/cmd/fluid/insight.go +++ /dev/null @@ -1,385 +0,0 @@ -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 6da0158..0c9f26f 100644 --- a/cmd/fluid/main.go +++ b/cmd/fluid/main.go @@ -30,11 +30,6 @@ 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 @@ -118,12 +113,6 @@ 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 deleted file mode 100644 index dafc1f7..0000000 --- a/conformance/loop/loop_test.go +++ /dev/null @@ -1,154 +0,0 @@ -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 9bb0da2..a6f478a 100644 --- a/internal/contract/event_gen.go +++ b/internal/contract/event_gen.go @@ -10,7 +10,6 @@ const ( FluidEventEntityTypeHypothesis FluidEventEntityType = "hypothesis" FluidEventEntityTypeRevision FluidEventEntityType = "revision" FluidEventEntityTypeExperiment FluidEventEntityType = "experiment" - FluidEventEntityTypeFeedback FluidEventEntityType = "feedback" FluidEventEntityTypeBackendRequirement FluidEventEntityType = "backend_requirement" FluidEventEntityTypeIntent FluidEventEntityType = "intent" FluidEventEntityTypeDecision FluidEventEntityType = "decision" @@ -20,7 +19,7 @@ const ( // Valid reports whether v is a defined FluidEventEntityType. func (v FluidEventEntityType) Valid() bool { switch v { - case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeFeedback, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy: + case FluidEventEntityTypePressure, FluidEventEntityTypeHypothesis, FluidEventEntityTypeRevision, FluidEventEntityTypeExperiment, FluidEventEntityTypeBackendRequirement, FluidEventEntityTypeIntent, FluidEventEntityTypeDecision, FluidEventEntityTypeRoutingPolicy: return true } return false diff --git a/internal/control/api.go b/internal/control/api.go index 8b4234b..1e83a07 100644 --- a/internal/control/api.go +++ b/internal/control/api.go @@ -23,13 +23,11 @@ import ( type Server struct { revisions *RevisionAPI intents *IntentAPI - pressure *PressureAPI } -// 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} +// NewServer wires the control APIs. +func NewServer(rev *RevisionAPI, in *IntentAPI) *Server { + return &Server{revisions: rev, intents: in} } // Routes returns the control-plane mux. @@ -42,14 +40,6 @@ 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 3073c99..ca9995e 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), nil) + srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate)) return srv.Routes(), store } diff --git a/internal/control/pressure.go b/internal/control/pressure.go deleted file mode 100644 index 6b31534..0000000 --- a/internal/control/pressure.go +++ /dev/null @@ -1,171 +0,0 @@ -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 deleted file mode 100644 index c75e052..0000000 --- a/internal/fitness/fitness.go +++ /dev/null @@ -1,357 +0,0 @@ -// 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 deleted file mode 100644 index d10cec0..0000000 --- a/internal/fitness/fitness_test.go +++ /dev/null @@ -1,325 +0,0 @@ -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 deleted file mode 100644 index bc39537..0000000 --- a/internal/fitness/measure.go +++ /dev/null @@ -1,164 +0,0 @@ -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 deleted file mode 100644 index 36e2a40..0000000 --- a/internal/observation/classify.go +++ /dev/null @@ -1,328 +0,0 @@ -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 deleted file mode 100644 index 4342914..0000000 --- a/internal/observation/cohort.go +++ /dev/null @@ -1,148 +0,0 @@ -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 deleted file mode 100644 index 0ed3885..0000000 --- a/internal/observation/feedback.go +++ /dev/null @@ -1,22 +0,0 @@ -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 deleted file mode 100644 index e8dfc23..0000000 --- a/internal/observation/ingest.go +++ /dev/null @@ -1,191 +0,0 @@ -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 deleted file mode 100644 index 99d1f98..0000000 --- a/internal/observation/ingest_test.go +++ /dev/null @@ -1,264 +0,0 @@ -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 deleted file mode 100644 index 96579c9..0000000 --- a/internal/observation/pressure_test.go +++ /dev/null @@ -1,365 +0,0 @@ -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 deleted file mode 100644 index 67a1954..0000000 --- a/internal/observation/redact.go +++ /dev/null @@ -1,235 +0,0 @@ -// 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 deleted file mode 100644 index 6a58d83..0000000 --- a/internal/observation/redact_test.go +++ /dev/null @@ -1,204 +0,0 @@ -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 deleted file mode 100644 index fa034fc..0000000 --- a/internal/observation/registry.go +++ /dev/null @@ -1,326 +0,0 @@ -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 deleted file mode 100644 index 9d7febf..0000000 --- a/internal/observation/topology.go +++ /dev/null @@ -1,306 +0,0 @@ -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 deleted file mode 100644 index 217437b..0000000 --- a/internal/observation/topology_test.go +++ /dev/null @@ -1,223 +0,0 @@ -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 2cb1db5..cf2a3a3 100644 --- a/schemas/event.schema.json +++ b/schemas/event.schema.json @@ -8,70 +8,28 @@ "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", - "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" - }, + "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" }, "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 1349c46..a2e46f0 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: done +status: active 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: done +status: todo 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: done +status: todo 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: done +status: todo 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: done +status: todo priority: high state_hub_task_id: "27726efa-e054-56c2-a6b8-3cce8f2e85e7" ``` @@ -74,7 +74,7 @@ error counting. ```task id: FLUID-WP-0005-T05 -status: done +status: todo 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: done +status: todo priority: high state_hub_task_id: "50bd30a6-ae29-5e2b-9618-a3efe6a362b3" ``` @@ -99,7 +99,7 @@ deserves adaptation. ```task id: FLUID-WP-0005-T07 -status: done +status: todo 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: done +status: todo 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: done +status: todo priority: medium state_hub_task_id: "ba413808-346b-5026-bfce-aec0eae0c410" ```