coulomb/fluid-core
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: coulomb:7e0de9e5b73a39f5e22956fb0e98d0213ad6e69f
Branches Tags
coulomb:main
..
compare: coulomb:03ff7a8ad72f7f385d5ea9f13e4c68cc7582cbf7
Branches Tags
coulomb:main

No commits in common. "7e0de9e5b73a39f5e22956fb0e98d0213ad6e69f" and "03ff7a8ad72f7f385d5ea9f13e4c68cc7582cbf7" have entirely different histories.
7e0de9e5b7 ... 03ff7a8ad7

24 changed files with 29 additions and 4288 deletions
Download patch file Download diff file Expand all files Collapse all files

31
cmd/fluid-control/main.go
View file

@ -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,
}

385
cmd/fluid/insight.go
View file

@ -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] + "…"
}

11
cmd/fluid/main.go
View file

@ -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:

154
conformance/loop/loop_test.go
View file

@ -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) }

3
internal/contract/event_gen.go
View file

@ -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

16
internal/control/api.go
View file

@ -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"})
})

2
internal/control/api_test.go
View file

@ -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
}

171
internal/control/pressure.go
View file

@ -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",
})
}

357
internal/fitness/fitness.go
View file

@ -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 }

325
internal/fitness/fitness_test.go
View file

@ -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
}

164
internal/fitness/measure.go
View file

@ -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]
}

328
internal/observation/classify.go
View file

@ -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
}

148
internal/observation/cohort.go
View file

@ -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
}

22
internal/observation/feedback.go
View file

@ -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})
}

191
internal/observation/ingest.go
View file

@ -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[:])
}

264
internal/observation/ingest_test.go
View file

@ -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")
}
}

365
internal/observation/pressure_test.go
View file

@ -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
}

235
internal/observation/redact.go
View file

@ -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
}

204
internal/observation/redact_test.go
View file

@ -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")
}
}

326
internal/observation/registry.go
View file

@ -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
}

306
internal/observation/topology.go
View file

@ -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
}

223
internal/observation/topology_test.go
View file

@ -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")
}
}

66
schemas/event.schema.json
View file

@ -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
}

20
workplans/FLUID-WP-0005-fluid-insight.md
View file

@ -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"
```