From d52dcc92a92d928fe801868a9e4379b9dda9101c Mon Sep 17 00:00:00 2001 From: tegwick Date: Fri, 4 Sep 2026 02:26:23 +0200 Subject: [PATCH] Add evidence store, intent store and the fluid CLI Completes FLUID-WP-0003. The evidence store is append-only at the database, not by convention in Go: a trigger blocks UPDATE and DELETE on fluid_events in both SQLite and Postgres, so the guarantee binds a psql session and the CLI equally, not just callers who go through the Go API. Records are derived summary state and may be superseded; the event log stays the authority. The intent store makes a recorded version immutable and content addressed, since reassigning what governed a revision after the fact would break the one audit question Blueprint 27 exists to answer. The unfilled InterfaceEvolutionIntent template is rejected rather than defaulted, because a template still listing every FLUID-N mode has not been completed and defaulting it would pick a permissive authority by accident. The CLI reads the evidence store directly rather than through the control-plane API, so an operator can reconstruct history when the control plane is down. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014KmVxhJ35tCo7rE7UnLwWu Assistant: claude-code Assistant-Model: opus Assistant-Process: 1116572@bnt-lap001 Assistant-Session: 8ba9bb93-a72a-4883-b189-2499cce5c400 --- cmd/fluid/commands.go | 510 ++++++++++++++++++ cmd/fluid/flags.go | 23 + cmd/fluid/main.go | 144 +++++ internal/contract/ids.go | 2 + internal/evidence/sink.go | 19 + internal/evidence/sqlstore.go | 399 ++++++++++++++ internal/evidence/store.go | 73 +++ internal/evidence/store_test.go | 214 ++++++++ internal/intent/store.go | 283 ++++++++++ internal/intent/store_test.go | 166 ++++++ .../FLUID-WP-0003-deterministic-data-plane.md | 6 +- 11 files changed, 1836 insertions(+), 3 deletions(-) create mode 100644 cmd/fluid/commands.go create mode 100644 cmd/fluid/flags.go create mode 100644 cmd/fluid/main.go create mode 100644 internal/evidence/sink.go create mode 100644 internal/evidence/sqlstore.go create mode 100644 internal/evidence/store.go create mode 100644 internal/evidence/store_test.go create mode 100644 internal/intent/store.go create mode 100644 internal/intent/store_test.go diff --git a/cmd/fluid/commands.go b/cmd/fluid/commands.go new file mode 100644 index 0000000..fc16d1a --- /dev/null +++ b/cmd/fluid/commands.go @@ -0,0 +1,510 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "sort" + "strings" + "text/tabwriter" + "time" + + "gopkg.in/yaml.v3" + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/intent" +) + +// open connects to the evidence store for this invocation. +func (g globals) open(ctx context.Context) (*evidence.SQLStore, error) { + return evidence.OpenSQLite(ctx, g.storePath()) +} + +func out() *tabwriter.Writer { + return tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) +} + +// ---------- intent ---------- + +func runIntent(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("intent needs a subcommand: put, activate, show") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + is := intent.New(store, contract.InterfaceID(iface)) + + switch args[0] { + case "put": + fs := newFlagSet("intent put") + version := fs.String("version", "", "intent version label, such as IEI-1") + file := fs.String("file", "", "path to the InterfaceEvolutionIntent document") + activate := fs.Bool("activate", false, "make this version the governing intent") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *version == "" || *file == "" { + return errors.New("intent put needs --version and --file") + } + + doc, err := os.ReadFile(*file) + if err != nil { + return err + } + v, err := is.Put(ctx, *version, string(doc)) + if err != nil { + return err + } + fmt.Printf("recorded %s\n digest %s\n mode %s\n", v.Version, v.Digest, v.Mode) + + if *activate { + if err := is.SetActive(ctx, v.Version); err != nil { + return err + } + fmt.Printf(" active yes\n") + } + return nil + + case "activate": + if len(args) < 2 { + return errors.New("intent activate needs a version") + } + if err := is.SetActive(ctx, args[1]); err != nil { + return err + } + fmt.Printf("%s is now the governing intent\n", args[1]) + return nil + + case "show": + var v intent.Version + if len(args) > 1 { + v, err = is.Get(ctx, args[1]) + } else { + v, err = is.Active(ctx) + } + if err != nil { + return err + } + fmt.Printf("version %s\ndigest %s\nmode %s\nrecorded %s\n\n", + v.Version, v.Digest, v.Mode, v.RecordedAt.Format(time.RFC3339)) + fmt.Println(v.Document) + return nil + + default: + return fmt.Errorf("unknown intent subcommand %q", args[0]) + } +} + +// ---------- revision ---------- + +func runRevision(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("revision needs a subcommand: publish, list, show") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + switch args[0] { + case "publish": + fs := newFlagSet("revision publish") + file := fs.String("file", "", "path to the revision descriptor (YAML or JSON)") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *file == "" { + return errors.New("revision publish needs --file") + } + + raw, err := os.ReadFile(*file) + if err != nil { + return err + } + var doc contract.RevisionDescriptorDocument + if err := yaml.Unmarshal(raw, &doc); err != nil { + return fmt.Errorf("parse descriptor: %w", err) + } + d := doc.Revision + + if string(d.Interface) != iface { + return fmt.Errorf("descriptor is for interface %q, but --interface is %q", d.Interface, iface) + } + if d.ID == "" { + return errors.New("descriptor has no revision id") + } + // The router refuses unsigned descriptors at runtime; refusing them here + // as well means an operator finds out at publish time rather than when + // traffic starts failing. + if d.Signature == nil { + fmt.Fprintln(os.Stderr, + "warning: descriptor is unsigned; the router will refuse it once signature enforcement is enabled") + } + + body, err := json.Marshal(d) + if err != nil { + return err + } + if err := store.PutRecord(ctx, contract.KindRevision, string(d.ID), body); err != nil { + return err + } + + if err := store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-pub-%s-%d", d.ID, time.Now().UnixNano())), + OccurredAt: time.Now().UTC(), + EntityType: contract.FluidEventEntityTypeRevision, + EntityID: string(d.ID), + EventType: "REVISION_PUBLISHED", + Actor: contract.Actor{Type: contract.ActorTypeHuman, ID: operator()}, + Reason: fmt.Sprintf("published %s in state %s", d.ID, d.State), + }); err != nil { + return err + } + + // Binding the revision to its governing intent is what makes the later + // audit question answerable (Blueprint 27). + is := intent.New(store, contract.InterfaceID(iface)) + if err := is.Bind(ctx, d.ID, d.Intent.Version); err != nil { + return fmt.Errorf("published, but binding to intent %s failed: %w", d.Intent.Version, err) + } + + fmt.Printf("published %s (%s), governed by %s\n", d.ID, d.State, d.Intent.Version) + return nil + + case "list": + records, err := store.Records(ctx, contract.KindRevision) + if err != nil { + return err + } + if len(records) == 0 { + fmt.Println("no revisions published") + return nil + } + + ids := make([]string, 0, len(records)) + for id := range records { + ids = append(ids, id) + } + sort.Strings(ids) + + w := out() + fmt.Fprintln(w, "REVISION\tSTATE\tINTENT\tCOMPATIBILITY\tUPSTREAM") + for _, id := range ids { + var d contract.Revision + if err := json.Unmarshal(records[id], &d); err != nil { + fmt.Fprintf(w, "%s\t(unreadable)\t\t\t\n", id) + continue + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", + d.ID, d.State, d.Intent.Version, d.Policy.Compatibility, d.Runtime.Upstream) + } + return w.Flush() + + case "show": + if len(args) < 2 { + return errors.New("revision show needs a revision id") + } + body, err := store.Record(ctx, contract.KindRevision, args[1]) + if err != nil { + return err + } + var pretty any + _ = json.Unmarshal(body, &pretty) + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(pretty) + + default: + return fmt.Errorf("unknown revision subcommand %q", args[0]) + } +} + +// ---------- policy ---------- + +const policyRecordID = "__routing_policy__" + +func runPolicy(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("policy needs a subcommand: put, show") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + switch args[0] { + case "put": + fs := newFlagSet("policy put") + file := fs.String("file", "", "path to the routing policy") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *file == "" { + return errors.New("policy put needs --file") + } + + raw, err := os.ReadFile(*file) + if err != nil { + return err + } + var doc contract.RoutingPolicyDocument + if err := yaml.Unmarshal(raw, &doc); err != nil { + return fmt.Errorf("parse policy: %w", err) + } + p := doc.RoutingPolicy + + if string(p.Interface) != iface { + return fmt.Errorf("policy is for interface %q, but --interface is %q", p.Interface, iface) + } + + // Generations are monotonic in the registry; catching a regression here + // avoids shipping a policy the gateway will silently refuse. + if existing, err := store.Record(ctx, contract.KindRoutingPolicy, policyRecordID); err == nil { + var prev contract.RoutingPolicy + if json.Unmarshal(existing, &prev) == nil && p.Generation <= prev.Generation { + return fmt.Errorf("policy generation %d is not newer than the installed %d", + p.Generation, prev.Generation) + } + } + + body, err := json.Marshal(p) + if err != nil { + return err + } + if err := store.PutRecord(ctx, contract.KindRoutingPolicy, policyRecordID, body); err != nil { + return err + } + + if err := store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-policy-%d", time.Now().UnixNano())), + OccurredAt: time.Now().UTC(), + EntityType: contract.FluidEventEntityTypeRoutingPolicy, + EntityID: fmt.Sprintf("generation-%d", p.Generation), + EventType: "ROUTING_POLICY_INSTALLED", + Actor: contract.Actor{Type: contract.ActorTypeHuman, ID: operator()}, + Reason: fmt.Sprintf("default %s, %d rule(s)", p.DefaultRevision, len(p.Rules)), + }); err != nil { + return err + } + + fmt.Printf("installed routing policy generation %d (default %s, %d rules)\n", + p.Generation, p.DefaultRevision, len(p.Rules)) + return nil + + case "show": + body, err := store.Record(ctx, contract.KindRoutingPolicy, policyRecordID) + if err != nil { + if errors.Is(err, evidence.ErrNotFound) { + fmt.Println("no routing policy installed") + return nil + } + return err + } + var pretty any + _ = json.Unmarshal(body, &pretty) + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(pretty) + + default: + return fmt.Errorf("unknown policy subcommand %q", args[0]) + } +} + +// ---------- events and telemetry ---------- + +func runEvents(ctx context.Context, g globals, args []string) error { + fs := newFlagSet("events") + entity := fs.String("entity", "", "filter by entity id, such as R-1") + kind := fs.String("type", "", "filter by event type") + limit := fs.Int("limit", 50, "maximum events to show") + if err := fs.Parse(args); err != nil { + return err + } + + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + events, err := store.Events(ctx, evidence.EventFilter{ + EntityID: *entity, + EventType: *kind, + Limit: *limit, + }) + if err != nil { + return err + } + if len(events) == 0 { + fmt.Println("no events") + return nil + } + + w := out() + fmt.Fprintln(w, "WHEN\tENTITY\tTYPE\tACTOR\tREASON") + for _, ev := range events { + fmt.Fprintf(w, "%s\t%s %s\t%s\t%s:%s\t%s\n", + ev.OccurredAt.Format(time.RFC3339), ev.EntityType, ev.EntityID, + ev.EventType, ev.Actor.Type, ev.Actor.ID, oneLine(ev.Reason)) + } + return w.Flush() +} + +func runTelemetry(ctx context.Context, g globals, args []string) error { + fs := newFlagSet("telemetry") + revision := fs.String("revision", "", "filter by revision") + kind := fs.String("kind", "", "filter by event kind") + limit := fs.Int("limit", 50, "maximum rows to show") + if err := fs.Parse(args); err != nil { + return err + } + + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + rows, err := store.Telemetry(ctx, evidence.TelemetryFilter{ + InterfaceID: contract.InterfaceID(g.iface), + Revision: contract.RevisionID(*revision), + Kind: contract.FluidTelemetryKind(*kind), + Limit: *limit, + }) + if err != nil { + return err + } + if len(rows) == 0 { + fmt.Println("no telemetry") + return nil + } + + w := out() + fmt.Fprintln(w, "WHEN\tKIND\tREVISION\tRESOLVED BY\tROUTE\tSTATUS") + for _, ev := range rows { + route, status := "", "" + if ev.Request != nil { + route = ev.Request.Route + if ev.Request.Status != nil { + status = fmt.Sprint(*ev.Request.Status) + } + } + reason := "" + if ev.Resolution != nil { + reason = string(ev.Resolution.Reason) + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\n", + ev.OccurredAt.Format(time.RFC3339), ev.Kind, deref(ev.Revision), reason, route, status) + } + return w.Flush() +} + +// ---------- audit ---------- + +// runAudit reconstructs the chain behind a revision. +// +// This is the command that has to answer the eleven questions in +// FluidAPIStandards.md section 25. It is deliberately built from events rather +// than from summary records: the records say what is true now, the events say +// how it came to be. +func runAudit(ctx context.Context, g globals, args []string) error { + if len(args) < 2 || args[0] != "trace" { + return errors.New("usage: fluid audit trace ") + } + target := args[1] + + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + events, err := store.Events(ctx, evidence.EventFilter{EntityID: target}) + if err != nil { + return err + } + if len(events) == 0 { + return fmt.Errorf("no history recorded for %s", target) + } + + fmt.Printf("Audit trace for %s\n\n", target) + + is := intent.New(store, contract.InterfaceID(iface)) + if v, err := is.GoverningIntent(ctx, contract.RevisionID(target)); err == nil { + fmt.Printf("Governed by %s (%s), authority mode %s\n\n", v.Version, v.Digest, v.Mode) + } else { + fmt.Printf("Governed by: UNKNOWN — no intent binding recorded\n\n") + } + + w := out() + fmt.Fprintln(w, "WHEN\tEVENT\tACTOR\tINPUTS\tREASON") + for _, ev := range events { + fmt.Fprintf(w, "%s\t%s\t%s:%s\t%s\t%s\n", + ev.OccurredAt.Format(time.RFC3339), ev.EventType, + ev.Actor.Type, ev.Actor.ID, + strings.Join(ev.Inputs, ","), oneLine(ev.Reason)) + } + if err := w.Flush(); err != nil { + return err + } + + count, err := store.Telemetry(ctx, evidence.TelemetryFilter{Revision: contract.RevisionID(target)}) + if err == nil { + fmt.Printf("\n%d telemetry event(s) recorded against %s\n", len(count), target) + } + return nil +} + +// ---------- helpers ---------- + +func newFlagSet(name string) *flagSet { return &flagSet{fs: newStdFlagSet(name)} } + +func oneLine(s string) string { + return strings.Join(strings.Fields(s), " ") +} + +func deref(s *contract.RevisionID) string { + if s == nil { + return "" + } + return string(*s) +} + +func operator() string { + if v := os.Getenv("FLUID_OPERATOR"); v != "" { + return v + } + if v := os.Getenv("USER"); v != "" { + return v + } + return "unknown" +} diff --git a/cmd/fluid/flags.go b/cmd/fluid/flags.go new file mode 100644 index 0000000..c137373 --- /dev/null +++ b/cmd/fluid/flags.go @@ -0,0 +1,23 @@ +package main + +import ( + "flag" + "os" +) + +// flagSet wraps flag.FlagSet so subcommands report errors instead of exiting, +// which keeps error handling in one place in run(). +type flagSet struct{ fs *flag.FlagSet } + +func newStdFlagSet(name string) *flag.FlagSet { + fs := flag.NewFlagSet(name, flag.ContinueOnError) + fs.SetOutput(os.Stderr) + return fs +} + +func (f *flagSet) String(name, value, usage string) *string { return f.fs.String(name, value, usage) } +func (f *flagSet) Bool(name string, value bool, usage string) *bool { + return f.fs.Bool(name, value, usage) +} +func (f *flagSet) Int(name string, value int, usage string) *int { return f.fs.Int(name, value, usage) } +func (f *flagSet) Parse(args []string) error { return f.fs.Parse(args) } diff --git a/cmd/fluid/main.go b/cmd/fluid/main.go new file mode 100644 index 0000000..0c9f26f --- /dev/null +++ b/cmd/fluid/main.go @@ -0,0 +1,144 @@ +// Command fluid is the FLUID operator CLI. +// +// It reads and writes the evidence store directly rather than going through the +// control-plane API, so that an operator can still inspect an interface's +// history when the control plane is down. Auditability must not depend on the +// availability of the thing being audited. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "os" + "path/filepath" + "strings" +) + +const usage = `fluid — FLUID operator CLI + +Usage: + fluid [--store PATH] [--interface ID] [arguments] + +Commands: + intent put Record an interface evolution intent version + intent activate Make a recorded version the governing intent + intent show Show the active or a named intent version + revision publish Publish a signed revision descriptor + revision list List published revisions + revision show Show one revision descriptor + policy put Install a routing policy + policy show Show the current routing policy + events Show audit events + telemetry Show recorded telemetry + audit trace Reconstruct the history behind a revision + +Global flags: + --store PATH Evidence store path (default $FLUID_STORE or ./fluid.db) + --interface ID Interface identifier (default $FLUID_INTERFACE) +` + +func main() { + if err := run(os.Args[1:]); err != nil { + if errors.Is(err, flag.ErrHelp) { + os.Exit(0) + } + fmt.Fprintln(os.Stderr, "fluid:", err) + os.Exit(1) + } +} + +type globals struct { + store string + iface string + remaining []string +} + +func parseGlobals(args []string) (globals, error) { + g := globals{ + store: envOr("FLUID_STORE", "fluid.db"), + iface: os.Getenv("FLUID_INTERFACE"), + } + + i := 0 + for i < len(args) { + switch args[i] { + case "--store": + if i+1 >= len(args) { + return g, errors.New("--store needs a value") + } + g.store, i = args[i+1], i+2 + case "--interface": + if i+1 >= len(args) { + return g, errors.New("--interface needs a value") + } + g.iface, i = args[i+1], i+2 + case "-h", "--help", "help": + fmt.Print(usage) + return g, flag.ErrHelp + default: + if strings.HasPrefix(args[i], "--") { + return g, fmt.Errorf("unknown global flag %q", args[i]) + } + g.remaining = args[i:] + return g, nil + } + } + g.remaining = nil + return g, nil +} + +func run(args []string) error { + g, err := parseGlobals(args) + if err != nil { + return err + } + if len(g.remaining) == 0 { + fmt.Print(usage) + return nil + } + + ctx := context.Background() + cmd, rest := g.remaining[0], g.remaining[1:] + + switch cmd { + case "intent": + return runIntent(ctx, g, rest) + case "revision": + return runRevision(ctx, g, rest) + case "policy": + return runPolicy(ctx, g, rest) + case "events": + return runEvents(ctx, g, rest) + case "telemetry": + return runTelemetry(ctx, g, rest) + case "audit": + return runAudit(ctx, g, rest) + default: + return fmt.Errorf("unknown command %q (try `fluid help`)", cmd) + } +} + +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +// requireInterface fails clearly rather than silently operating on the wrong +// interface's history. +func (g globals) requireInterface() (string, error) { + if g.iface == "" { + return "", errors.New("no interface selected: pass --interface or set FLUID_INTERFACE") + } + return g.iface, nil +} + +func (g globals) storePath() string { + if abs, err := filepath.Abs(g.store); err == nil { + return abs + } + return g.store +} diff --git a/internal/contract/ids.go b/internal/contract/ids.go index 93d6cbc..5a15340 100644 --- a/internal/contract/ids.go +++ b/internal/contract/ids.go @@ -34,6 +34,8 @@ const ( KindEvent EntityKind = "event" KindFeedback EntityKind = "feedback" KindCohort EntityKind = "cohort" + KindIntent EntityKind = "intent" + KindRoutingPolicy EntityKind = "routing_policy" ) // prefixOrder matters: "BR-" must be tested before "B"-less single letters diff --git a/internal/evidence/sink.go b/internal/evidence/sink.go new file mode 100644 index 0000000..e505988 --- /dev/null +++ b/internal/evidence/sink.go @@ -0,0 +1,19 @@ +package evidence + +import ( + "context" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// TelemetrySink adapts a Store to the runtime's telemetry sink. +// +// It is a separate type so the data plane depends on the narrow sink interface +// rather than on the whole evidence store: the gateway should not be able to +// read hypotheses. +type TelemetrySink struct{ Store Store } + +// Write implements the runtime Sink interface. +func (s TelemetrySink) Write(ctx context.Context, ev contract.FluidTelemetry) error { + return s.Store.WriteTelemetry(ctx, ev) +} diff --git a/internal/evidence/sqlstore.go b/internal/evidence/sqlstore.go new file mode 100644 index 0000000..c7bf45a --- /dev/null +++ b/internal/evidence/sqlstore.go @@ -0,0 +1,399 @@ +package evidence + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// SQLStore is the relational evidence store. +// +// One schema serves both backends (ADR-0004): SQLite for development, CI and +// single-node deployments, PostgreSQL for anything shared. Only portable SQL is +// used, so the statements CI exercises are the statements production runs. +type SQLStore struct { + db *sql.DB + dialect dialect +} + +type dialect int + +const ( + dialectSQLite dialect = iota + dialectPostgres +) + +// schemaStatements builds the DDL for a dialect. +// +// The immutability of fluid_events is enforced at the database, not by +// convention in Go. Blueprint invariant 8 (every promotion is auditable) is +// only as strong as the weakest writer, and a trigger binds every one of them. +func schemaStatements(d dialect) []string { + stmts := []string{ + `CREATE TABLE IF NOT EXISTS fluid_events ( + id TEXT PRIMARY KEY, + occurred_at TEXT NOT NULL, + entity_type TEXT NOT NULL, + entity_id TEXT NOT NULL, + event_type TEXT NOT NULL, + body TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS fluid_events_entity + ON fluid_events (entity_type, entity_id, occurred_at)`, + `CREATE INDEX IF NOT EXISTS fluid_events_time + ON fluid_events (occurred_at)`, + + `CREATE TABLE IF NOT EXISTS fluid_telemetry ( + id TEXT PRIMARY KEY, + occurred_at TEXT NOT NULL, + interface_id TEXT NOT NULL, + kind TEXT NOT NULL, + revision TEXT, + experiment TEXT, + cohort TEXT, + body TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS fluid_telemetry_window + ON fluid_telemetry (interface_id, occurred_at)`, + `CREATE INDEX IF NOT EXISTS fluid_telemetry_revision + ON fluid_telemetry (revision, occurred_at)`, + + // Records are derived summary state, rebuildable from events. + `CREATE TABLE IF NOT EXISTS fluid_records ( + kind TEXT NOT NULL, + id TEXT NOT NULL, + updated_at TEXT NOT NULL, + body TEXT NOT NULL, + PRIMARY KEY (kind, id) + )`, + } + + switch d { + case dialectSQLite: + stmts = append(stmts, + `CREATE TRIGGER IF NOT EXISTS fluid_events_no_update + BEFORE UPDATE ON fluid_events + BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`, + `CREATE TRIGGER IF NOT EXISTS fluid_events_no_delete + BEFORE DELETE ON fluid_events + BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`, + ) + case dialectPostgres: + stmts = append(stmts, + `CREATE OR REPLACE FUNCTION fluid_events_append_only() + RETURNS trigger AS $$ + BEGIN RAISE EXCEPTION 'fluid_events is append-only'; END; + $$ LANGUAGE plpgsql`, + `DROP TRIGGER IF EXISTS fluid_events_no_change ON fluid_events`, + `CREATE TRIGGER fluid_events_no_change + BEFORE UPDATE OR DELETE ON fluid_events + FOR EACH ROW EXECUTE FUNCTION fluid_events_append_only()`, + ) + } + return stmts +} + +// OpenSQLite opens (and migrates) a SQLite-backed store. +func OpenSQLite(ctx context.Context, path string) (*SQLStore, error) { + db, err := sql.Open("sqlite", path) + if err != nil { + return nil, fmt.Errorf("open sqlite %q: %w", path, err) + } + // SQLite serializes writers; more than one connection buys contention. + db.SetMaxOpenConns(1) + + for _, pragma := range []string{ + "PRAGMA journal_mode = WAL", + "PRAGMA foreign_keys = ON", + "PRAGMA busy_timeout = 5000", + } { + if _, err := db.ExecContext(ctx, pragma); err != nil { + _ = db.Close() + return nil, fmt.Errorf("%s: %w", pragma, err) + } + } + + s := &SQLStore{db: db, dialect: dialectSQLite} + if err := s.migrate(ctx); err != nil { + _ = db.Close() + return nil, err + } + return s, nil +} + +// NewSQLStore wraps an already-open database, for Postgres or for tests. +func NewSQLStore(ctx context.Context, db *sql.DB, postgres bool) (*SQLStore, error) { + d := dialectSQLite + if postgres { + d = dialectPostgres + } + s := &SQLStore{db: db, dialect: d} + if err := s.migrate(ctx); err != nil { + return nil, err + } + return s, nil +} + +func (s *SQLStore) migrate(ctx context.Context) error { + for _, stmt := range schemaStatements(s.dialect) { + if _, err := s.db.ExecContext(ctx, stmt); err != nil { + return fmt.Errorf("migrate: %w\nstatement: %s", err, stmt) + } + } + return nil +} + +// arg renders the nth placeholder for the dialect. +func (s *SQLStore) arg(n int) string { + if s.dialect == dialectPostgres { + return fmt.Sprintf("$%d", n) + } + return "?" +} + +// rfc3339 normalizes timestamps so lexical ordering matches chronological +// ordering in both backends. +func rfc3339(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) } + +// AppendEvent records a lifecycle transition. +func (s *SQLStore) AppendEvent(ctx context.Context, ev contract.FluidEvent) error { + if ev.ID == "" { + return errors.New("event has no id") + } + body, err := json.Marshal(ev) + if err != nil { + return fmt.Errorf("marshal event %s: %w", ev.ID, err) + } + + q := fmt.Sprintf( + `INSERT INTO fluid_events (id, occurred_at, entity_type, entity_id, event_type, body) + VALUES (%s, %s, %s, %s, %s, %s)`, + s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6)) + + _, err = s.db.ExecContext(ctx, q, + string(ev.ID), rfc3339(ev.OccurredAt), string(ev.EntityType), + ev.EntityID, ev.EventType, string(body)) + if err != nil { + return fmt.Errorf("append event %s: %w", ev.ID, err) + } + return nil +} + +// Events returns matching events in occurrence order. +func (s *SQLStore) Events(ctx context.Context, f EventFilter) ([]contract.FluidEvent, error) { + var ( + where []string + args []any + ) + add := func(clause string, v any) { + args = append(args, v) + where = append(where, fmt.Sprintf(clause, s.arg(len(args)))) + } + + if f.EntityType != "" { + add("entity_type = %s", string(f.EntityType)) + } + if f.EntityID != "" { + add("entity_id = %s", f.EntityID) + } + if f.EventType != "" { + add("event_type = %s", f.EventType) + } + if !f.Since.IsZero() { + add("occurred_at >= %s", rfc3339(f.Since)) + } + if !f.Until.IsZero() { + add("occurred_at <= %s", rfc3339(f.Until)) + } + + q := "SELECT body FROM fluid_events" + if len(where) > 0 { + q += " WHERE " + strings.Join(where, " AND ") + } + // id breaks ties so that two events in the same instant still order stably. + q += " ORDER BY occurred_at, id" + if f.Limit > 0 { + q += fmt.Sprintf(" LIMIT %d", f.Limit) + } + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("query events: %w", err) + } + defer rows.Close() + + var out []contract.FluidEvent + for rows.Next() { + var body string + if err := rows.Scan(&body); err != nil { + return nil, err + } + var ev contract.FluidEvent + if err := json.Unmarshal([]byte(body), &ev); err != nil { + return nil, fmt.Errorf("decode stored event: %w", err) + } + out = append(out, ev) + } + return out, rows.Err() +} + +// WriteTelemetry records a normalized interaction event. +func (s *SQLStore) WriteTelemetry(ctx context.Context, ev contract.FluidTelemetry) error { + body, err := json.Marshal(ev) + if err != nil { + return fmt.Errorf("marshal telemetry %s: %w", ev.ID, err) + } + + var revision, experiment, cohort any + if ev.Revision != nil { + revision = string(*ev.Revision) + } + if ev.Experiment != nil { + experiment = string(*ev.Experiment) + } + if ev.Cohort != nil { + cohort = string(*ev.Cohort) + } + + q := fmt.Sprintf( + `INSERT INTO fluid_telemetry (id, occurred_at, interface_id, kind, revision, experiment, cohort, body) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s)`, + s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6), s.arg(7), s.arg(8)) + + _, err = s.db.ExecContext(ctx, q, + ev.ID, rfc3339(ev.OccurredAt), string(ev.InterfaceID), string(ev.Kind), + revision, experiment, cohort, string(body)) + if err != nil { + return fmt.Errorf("write telemetry %s: %w", ev.ID, err) + } + return nil +} + +// Telemetry returns matching telemetry in occurrence order. +func (s *SQLStore) Telemetry(ctx context.Context, f TelemetryFilter) ([]contract.FluidTelemetry, error) { + var ( + where []string + args []any + ) + add := func(clause string, v any) { + args = append(args, v) + where = append(where, fmt.Sprintf(clause, s.arg(len(args)))) + } + + if f.InterfaceID != "" { + add("interface_id = %s", string(f.InterfaceID)) + } + if f.Revision != "" { + add("revision = %s", string(f.Revision)) + } + if f.Experiment != "" { + add("experiment = %s", string(f.Experiment)) + } + if f.Cohort != "" { + add("cohort = %s", string(f.Cohort)) + } + if f.Kind != "" { + add("kind = %s", string(f.Kind)) + } + if !f.Since.IsZero() { + add("occurred_at >= %s", rfc3339(f.Since)) + } + if !f.Until.IsZero() { + add("occurred_at <= %s", rfc3339(f.Until)) + } + + q := "SELECT body FROM fluid_telemetry" + if len(where) > 0 { + q += " WHERE " + strings.Join(where, " AND ") + } + q += " ORDER BY occurred_at, id" + if f.Limit > 0 { + q += fmt.Sprintf(" LIMIT %d", f.Limit) + } + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("query telemetry: %w", err) + } + defer rows.Close() + + var out []contract.FluidTelemetry + for rows.Next() { + var body string + if err := rows.Scan(&body); err != nil { + return nil, err + } + var ev contract.FluidTelemetry + if err := json.Unmarshal([]byte(body), &ev); err != nil { + return nil, fmt.Errorf("decode stored telemetry: %w", err) + } + out = append(out, ev) + } + return out, rows.Err() +} + +// PutRecord stores or supersedes a derived record. +func (s *SQLStore) PutRecord(ctx context.Context, kind contract.EntityKind, id string, body []byte) error { + if id == "" { + return errors.New("record has no id") + } + if !json.Valid(body) { + return fmt.Errorf("record %s/%s: body is not valid JSON", kind, id) + } + + q := fmt.Sprintf( + `INSERT INTO fluid_records (kind, id, updated_at, body) VALUES (%s, %s, %s, %s) + ON CONFLICT (kind, id) DO UPDATE SET updated_at = excluded.updated_at, body = excluded.body`, + s.arg(1), s.arg(2), s.arg(3), s.arg(4)) + + _, err := s.db.ExecContext(ctx, q, string(kind), id, rfc3339(time.Now()), string(body)) + if err != nil { + return fmt.Errorf("put record %s/%s: %w", kind, id, err) + } + return nil +} + +// Record returns one stored record. +func (s *SQLStore) Record(ctx context.Context, kind contract.EntityKind, id string) ([]byte, error) { + q := fmt.Sprintf(`SELECT body FROM fluid_records WHERE kind = %s AND id = %s`, s.arg(1), s.arg(2)) + + var body string + err := s.db.QueryRowContext(ctx, q, string(kind), id).Scan(&body) + if errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("%w: %s %s", ErrNotFound, kind, id) + } + if err != nil { + return nil, err + } + return []byte(body), nil +} + +// Records returns every record of a kind. +func (s *SQLStore) Records(ctx context.Context, kind contract.EntityKind) (map[string][]byte, error) { + q := fmt.Sprintf(`SELECT id, body FROM fluid_records WHERE kind = %s ORDER BY id`, s.arg(1)) + + rows, err := s.db.QueryContext(ctx, q, string(kind)) + if err != nil { + return nil, err + } + defer rows.Close() + + out := map[string][]byte{} + for rows.Next() { + var id, body string + if err := rows.Scan(&id, &body); err != nil { + return nil, err + } + out[id] = []byte(body) + } + return out, rows.Err() +} + +// Close releases the database. +func (s *SQLStore) Close() error { return s.db.Close() } diff --git a/internal/evidence/store.go b/internal/evidence/store.go new file mode 100644 index 0000000..e8b2722 --- /dev/null +++ b/internal/evidence/store.go @@ -0,0 +1,73 @@ +// Package evidence implements the FLUID evidence store. +// +// ArchitectureBlueprint.md section 26 requires append-only history, with +// mutable summary views derived from immutable events. Auditability is a core +// invariant (FluidAPIStandards.md section 25): the store must be able to answer +// what changed, why, on what evidence, and how the prior state is restored. +package evidence + +import ( + "context" + "errors" + "time" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Store is the append-only evidence log plus the queries built on it. +// +// Nothing in this interface updates or deletes. That is deliberate: history +// that can be rewritten is not evidence, and Blueprint section 28.1 gives the +// evidence writer permission to append but not to revise. +type Store interface { + // AppendEvent records a lifecycle transition. + AppendEvent(context.Context, contract.FluidEvent) error + // Events returns matching events in occurrence order. + Events(context.Context, EventFilter) ([]contract.FluidEvent, error) + + // WriteTelemetry records a normalized interaction event. + WriteTelemetry(context.Context, contract.FluidTelemetry) error + // Telemetry returns matching telemetry in occurrence order. + Telemetry(context.Context, TelemetryFilter) ([]contract.FluidTelemetry, error) + + // PutRecord stores or supersedes a FLUID record (pressure, hypothesis, + // revision, experiment, feedback, backend requirement). + // + // Records are summary state derived from events; the event log remains the + // authority. A record may be rewritten, an event may not. + PutRecord(ctx context.Context, kind contract.EntityKind, id string, body []byte) error + // Record returns one stored record. + Record(ctx context.Context, kind contract.EntityKind, id string) ([]byte, error) + // Records returns every record of a kind, ordered by id. + Records(ctx context.Context, kind contract.EntityKind) (map[string][]byte, error) + + Close() error +} + +// ErrNotFound reports a record or event that does not exist. +var ErrNotFound = errors.New("not found") + +// ErrImmutable reports an attempt to rewrite history. +var ErrImmutable = errors.New("evidence events are append-only") + +// EventFilter narrows an event query. Zero values mean "no constraint". +type EventFilter struct { + EntityType contract.EntityKind + EntityID string + EventType string + Since time.Time + Until time.Time + Limit int +} + +// TelemetryFilter narrows a telemetry query. +type TelemetryFilter struct { + InterfaceID contract.InterfaceID + Revision contract.RevisionID + Experiment contract.ExperimentID + Cohort contract.CohortID + Kind contract.FluidTelemetryKind + Since time.Time + Until time.Time + Limit int +} diff --git a/internal/evidence/store_test.go b/internal/evidence/store_test.go new file mode 100644 index 0000000..a25843a --- /dev/null +++ b/internal/evidence/store_test.go @@ -0,0 +1,214 @@ +package evidence + +import ( + "context" + "encoding/json" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" +) + +func newStore(t *testing.T) *SQLStore { + t.Helper() + path := filepath.Join(t.TempDir(), "evidence.db") + s, err := OpenSQLite(context.Background(), path) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + return s +} + +func event(id string, at time.Time, entity contract.EntityKind, entityID, kind string) contract.FluidEvent { + return contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(id), + OccurredAt: at, + EntityType: contract.FluidEventEntityType(entity), + EntityID: entityID, + EventType: kind, + Actor: contract.Actor{Type: contract.ActorTypeSystem, ID: "test"}, + } +} + +func TestAppendAndQueryEvents(t *testing.T) { + s := newStore(t) + ctx := context.Background() + base := time.Date(2026, 9, 4, 12, 0, 0, 0, time.UTC) + + events := []contract.FluidEvent{ + event("EV-3", base.Add(2*time.Minute), contract.KindRevision, "R-1", "PUBLISHED"), + event("EV-1", base, contract.KindRevision, "R-1", "CREATED"), + event("EV-2", base.Add(time.Minute), contract.KindRevision, "R-1", "VERIFIED"), + event("EV-4", base.Add(3*time.Minute), contract.KindHypothesis, "H-1", "CREATED"), + } + for _, ev := range events { + if err := s.AppendEvent(ctx, ev); err != nil { + t.Fatalf("append %s: %v", ev.ID, err) + } + } + + got, err := s.Events(ctx, EventFilter{EntityType: contract.KindRevision, EntityID: "R-1"}) + if err != nil { + t.Fatal(err) + } + if len(got) != 3 { + t.Fatalf("got %d events, want 3", len(got)) + } + // Insertion order was scrambled; occurrence order must come back sorted, + // because an audit trail read out of order is worse than none. + want := []string{"CREATED", "VERIFIED", "PUBLISHED"} + for i, ev := range got { + if ev.EventType != want[i] { + t.Errorf("position %d = %s, want %s", i, ev.EventType, want[i]) + } + } +} + +// TestEventsAreAppendOnly checks the invariant at the database, not in Go. +// Blueprint 28.1 gives the evidence writer permission to append and not to +// revise; a Go-level convention would not bind a CLI or a psql session. +func TestEventsAreAppendOnly(t *testing.T) { + s := newStore(t) + ctx := context.Background() + + ev := event("EV-1", time.Now(), contract.KindRevision, "R-1", "CREATED") + if err := s.AppendEvent(ctx, ev); err != nil { + t.Fatal(err) + } + + if _, err := s.db.ExecContext(ctx, `UPDATE fluid_events SET event_type = 'REWRITTEN'`); err == nil { + t.Error("UPDATE on fluid_events succeeded; history is rewritable") + } else if !strings.Contains(err.Error(), "append-only") { + t.Errorf("UPDATE failed for the wrong reason: %v", err) + } + + if _, err := s.db.ExecContext(ctx, `DELETE FROM fluid_events`); err == nil { + t.Error("DELETE on fluid_events succeeded; history is erasable") + } else if !strings.Contains(err.Error(), "append-only") { + t.Errorf("DELETE failed for the wrong reason: %v", err) + } + + // The original event must still be there and unchanged. + got, err := s.Events(ctx, EventFilter{EntityID: "R-1"}) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].EventType != "CREATED" { + t.Errorf("event was altered: %+v", got) + } +} + +func TestDuplicateEventIDRefused(t *testing.T) { + s := newStore(t) + ctx := context.Background() + ev := event("EV-1", time.Now(), contract.KindRevision, "R-1", "CREATED") + if err := s.AppendEvent(ctx, ev); err != nil { + t.Fatal(err) + } + if err := s.AppendEvent(ctx, ev); err == nil { + t.Error("duplicate event id accepted; an event was silently replayed") + } +} + +func TestTelemetryRoundTripAndFilters(t *testing.T) { + s := newStore(t) + ctx := context.Background() + base := time.Date(2026, 9, 5, 8, 0, 0, 0, time.UTC) + + r1 := contract.RevisionID("R-1") + r2 := contract.RevisionID("R-2") + exp := contract.ExperimentID("E-1") + cohort := contract.CohortID("coding-agents") + + for i, rev := range []*contract.RevisionID{&r1, &r1, &r2} { + ev := contract.FluidTelemetry{ + SchemaVersion: "0.1", + ID: "tl-" + string(rune('a'+i)), + OccurredAt: base.Add(time.Duration(i) * time.Minute), + InterfaceID: "hall-publishing", + Kind: contract.FluidTelemetryKindRequest, + Revision: rev, + Experiment: &exp, + Cohort: &cohort, + } + if err := s.WriteTelemetry(ctx, ev); err != nil { + t.Fatalf("write %d: %v", i, err) + } + } + + all, err := s.Telemetry(ctx, TelemetryFilter{InterfaceID: "hall-publishing"}) + if err != nil { + t.Fatal(err) + } + if len(all) != 3 { + t.Fatalf("got %d telemetry rows, want 3", len(all)) + } + + only1, err := s.Telemetry(ctx, TelemetryFilter{Revision: "R-1"}) + if err != nil { + t.Fatal(err) + } + if len(only1) != 2 { + t.Errorf("revision filter returned %d rows, want 2", len(only1)) + } + + // The measurement window is what makes a fitness comparison honest, so + // bounded queries have to be exact. + windowed, err := s.Telemetry(ctx, TelemetryFilter{ + InterfaceID: "hall-publishing", + Since: base.Add(time.Minute), + }) + if err != nil { + t.Fatal(err) + } + if len(windowed) != 2 { + t.Errorf("windowed query returned %d rows, want 2", len(windowed)) + } +} + +func TestRecordsAreSupersedable(t *testing.T) { + s := newStore(t) + ctx := context.Background() + + first, _ := json.Marshal(map[string]string{"status": "OPEN"}) + if err := s.PutRecord(ctx, contract.KindPressure, "P-1", first); err != nil { + t.Fatal(err) + } + + // Records are derived summary state and may be rewritten; the event log + // remains the authority for how they got there. + second, _ := json.Marshal(map[string]string{"status": "ADDRESSED"}) + if err := s.PutRecord(ctx, contract.KindPressure, "P-1", second); err != nil { + t.Fatal(err) + } + + got, err := s.Record(ctx, contract.KindPressure, "P-1") + if err != nil { + t.Fatal(err) + } + var decoded map[string]string + if err := json.Unmarshal(got, &decoded); err != nil { + t.Fatal(err) + } + if decoded["status"] != "ADDRESSED" { + t.Errorf("record not superseded: %v", decoded) + } + + if _, err := s.Record(ctx, contract.KindPressure, "P-missing"); !errors.Is(err, ErrNotFound) { + t.Errorf("missing record returned %v, want ErrNotFound", err) + } +} + +func TestPutRecordRejectsNonJSON(t *testing.T) { + s := newStore(t) + if err := s.PutRecord(context.Background(), contract.KindPressure, "P-1", []byte("not json")); err == nil { + t.Error("non-JSON record body accepted") + } +} diff --git a/internal/intent/store.go b/internal/intent/store.go new file mode 100644 index 0000000..923a219 --- /dev/null +++ b/internal/intent/store.go @@ -0,0 +1,283 @@ +// Package intent implements the FLUID intent store. +// +// ArchitectureBlueprint.md section 27: the system must retain the exact +// interface evolution intent governing each revision, so that a later audit can +// answer whether a change was valid under the intent that existed when it was +// made. An intent document is therefore versioned, content-addressed and +// immutable once recorded. +package intent + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "regexp" + "strings" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +// AuthorityMode is the operational authority granted to the Daimon. +// +// FluidAPIStandards.md section 26 is explicit that these describe authority, +// not maturity: a high-assurance system may deliberately remain at FLUID-2. +type AuthorityMode int + +const ( + ModeInstrumented AuthorityMode = iota // FLUID-0 + ModeAnalytical // FLUID-1 + ModeAdvisory // FLUID-2 + ModeConstructive // FLUID-3 + ModeExperimental // FLUID-4 + ModeBoundedAutonomous // FLUID-5 + ModeEvolutionary // FLUID-6 +) + +// String renders the canonical FLUID-N form. +func (m AuthorityMode) String() string { return fmt.Sprintf("FLUID-%d", int(m)) } + +// Valid reports whether m is a defined mode. +func (m AuthorityMode) Valid() bool { return m >= ModeInstrumented && m <= ModeEvolutionary } + +// Allows reports whether this mode permits at least the authority of want. +func (m AuthorityMode) Allows(want AuthorityMode) bool { return m >= want } + +var modePattern = regexp.MustCompile(`FLUID-([0-6])\b`) + +// Version is one recorded interface evolution intent. +type Version struct { + // Version is the human-facing label, such as "IEI-7". + Version string `json:"version"` + // Digest content-addresses the document. Two intent versions with the same + // digest are the same intent, whatever they were called. + Digest contract.Digest `json:"digest"` + // Document is the full text, retained verbatim. A summary would not settle + // an audit question about what the intent actually said. + Document string `json:"document"` + // Mode is the operational authority the document declares. + Mode AuthorityMode `json:"mode"` + // RecordedAt is when this version entered the store. + RecordedAt time.Time `json:"recorded_at"` +} + +// Store holds intent versions and which one is currently active. +type Store struct { + ev evidence.Store + iface contract.InterfaceID +} + +// New returns a store backed by ev. +func New(ev evidence.Store, iface contract.InterfaceID) *Store { + return &Store{ev: ev, iface: iface} +} + +var ( + // ErrNotFound reports an unknown intent version. + ErrNotFound = evidence.ErrNotFound + // ErrImmutable reports an attempt to change a recorded intent version. + ErrImmutable = errors.New("a recorded intent version cannot be changed") + // ErrNoActive reports that no intent has been made active yet. + ErrNoActive = errors.New("no active interface evolution intent") +) + +// activeKey is the record id holding the active-version pointer. +const activeKey = "__active__" + +// Digest computes the content address of an intent document. +func Digest(document string) contract.Digest { + sum := sha256.Sum256([]byte(document)) + return contract.Digest("sha256:" + hex.EncodeToString(sum[:])) +} + +// ParseMode extracts the declared authority mode from an intent document. +// +// The template writes the mode as a fenced list of alternatives before it is +// filled in; a document that still contains every mode has not been completed, +// and defaulting it to something permissive would be exactly the wrong failure. +func ParseMode(document string) (AuthorityMode, error) { + matches := modePattern.FindAllStringSubmatch(document, -1) + if len(matches) == 0 { + return 0, errors.New("intent document declares no FLUID-N authority mode") + } + + seen := map[string]bool{} + for _, m := range matches { + seen[m[1]] = true + } + if len(seen) > 1 { + return 0, fmt.Errorf( + "intent document declares %d different authority modes; the template placeholder has not been resolved", + len(seen)) + } + + return AuthorityMode(matches[0][1][0] - '0'), nil +} + +// Put records an intent version. +// +// Recording the same version twice with identical content is a no-op, which +// makes startup idempotent. Recording it with different content is refused: +// changing what a revision was governed by, after the fact, would break the +// audit question the store exists to answer. +func (s *Store) Put(ctx context.Context, version, document string) (Version, error) { + if version == "" { + return Version{}, errors.New("intent version label is required") + } + + mode, err := ParseMode(document) + if err != nil { + return Version{}, fmt.Errorf("intent %s: %w", version, err) + } + + v := Version{ + Version: version, + Digest: Digest(document), + Document: document, + Mode: mode, + RecordedAt: time.Now().UTC(), + } + + if existing, err := s.Get(ctx, version); err == nil { + if existing.Digest != v.Digest { + return Version{}, fmt.Errorf("%w: %s already recorded with digest %s", + ErrImmutable, version, existing.Digest) + } + return existing, nil + } else if !errors.Is(err, ErrNotFound) { + return Version{}, err + } + + body, err := json.Marshal(v) + if err != nil { + return Version{}, err + } + if err := s.ev.PutRecord(ctx, contract.KindIntent, s.key(version), body); err != nil { + return Version{}, err + } + + if err := s.ev.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID("EV-intent-" + string(v.Digest[7:19])), + OccurredAt: v.RecordedAt, + EntityType: contract.FluidEventEntityTypeIntent, + EntityID: version, + EventType: "INTENT_RECORDED", + Actor: contract.Actor{Type: contract.ActorTypeSystem, ID: "fluid-intent-store"}, + Reason: fmt.Sprintf("recorded %s at authority mode %s", version, mode), + }); err != nil { + return Version{}, err + } + + return v, nil +} + +// Get returns a recorded intent version. +func (s *Store) Get(ctx context.Context, version string) (Version, error) { + body, err := s.ev.Record(ctx, contract.KindIntent, s.key(version)) + if err != nil { + return Version{}, err + } + var v Version + if err := json.Unmarshal(body, &v); err != nil { + return Version{}, fmt.Errorf("decode intent %s: %w", version, err) + } + return v, nil +} + +// SetActive makes a recorded version the governing intent. +func (s *Store) SetActive(ctx context.Context, version string) error { + v, err := s.Get(ctx, version) + if err != nil { + return fmt.Errorf("cannot activate %s: %w", version, err) + } + + body, _ := json.Marshal(map[string]string{"version": v.Version, "digest": string(v.Digest)}) + if err := s.ev.PutRecord(ctx, contract.KindIntent, s.key(activeKey), body); err != nil { + return err + } + + return s.ev.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID("EV-intent-active-" + string(v.Digest[7:19])), + OccurredAt: time.Now().UTC(), + EntityType: contract.FluidEventEntityTypeIntent, + EntityID: version, + EventType: "INTENT_ACTIVATED", + Actor: contract.Actor{Type: contract.ActorTypeSystem, ID: "fluid-intent-store"}, + Reason: fmt.Sprintf("%s is now the governing intent", version), + }) +} + +// Active returns the currently governing intent version. +func (s *Store) Active(ctx context.Context) (Version, error) { + body, err := s.ev.Record(ctx, contract.KindIntent, s.key(activeKey)) + if errors.Is(err, ErrNotFound) { + return Version{}, ErrNoActive + } + if err != nil { + return Version{}, err + } + + var ptr struct { + Version string `json:"version"` + } + if err := json.Unmarshal(body, &ptr); err != nil { + return Version{}, err + } + return s.Get(ctx, ptr.Version) +} + +// Bind records that a revision is governed by an intent version. +// +// This is the link Blueprint section 27 requires. It is stored as an event +// rather than as a mutable field so that a revision's governing intent cannot +// be quietly reassigned later. +func (s *Store) Bind(ctx context.Context, rev contract.RevisionID, version string) error { + v, err := s.Get(ctx, version) + if err != nil { + return fmt.Errorf("cannot bind %s: %w", rev, err) + } + + return s.ev.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID("EV-bind-" + string(rev) + "-" + version), + OccurredAt: time.Now().UTC(), + EntityType: contract.FluidEventEntityTypeRevision, + EntityID: string(rev), + EventType: "INTENT_BOUND", + Actor: contract.Actor{Type: contract.ActorTypeSystem, ID: "fluid-intent-store"}, + Inputs: []string{version}, + Reason: fmt.Sprintf("%s is governed by %s (%s)", rev, version, v.Digest), + }) +} + +// GoverningIntent returns the intent version a revision was bound to. +func (s *Store) GoverningIntent(ctx context.Context, rev contract.RevisionID) (Version, error) { + events, err := s.ev.Events(ctx, evidence.EventFilter{ + EntityType: contract.KindRevision, + EntityID: string(rev), + EventType: "INTENT_BOUND", + }) + if err != nil { + return Version{}, err + } + if len(events) == 0 { + return Version{}, fmt.Errorf("%w: no intent bound to %s", ErrNotFound, rev) + } + // The first binding governs. A later one would be a reassignment, which the + // audit trail records but must not silently win. + first := events[0] + if len(first.Inputs) == 0 { + return Version{}, fmt.Errorf("binding event %s names no intent version", first.ID) + } + return s.Get(ctx, first.Inputs[0]) +} + +func (s *Store) key(id string) string { + return string(s.iface) + "/" + strings.TrimSpace(id) +} diff --git a/internal/intent/store_test.go b/internal/intent/store_test.go new file mode 100644 index 0000000..393877f --- /dev/null +++ b/internal/intent/store_test.go @@ -0,0 +1,166 @@ +package intent + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +func newStore(t *testing.T) *Store { + t.Helper() + ev, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ev.Close() }) + return New(ev, "hall-publishing") +} + +const filledIntent = `# Interface Evolution Intent + +**Current operational authority mode:** +FLUID-2 + +## Mission +Publish hall-of-helix entries to a Telegram channel. +` + +func TestPutIsIdempotentAndImmutable(t *testing.T) { + s := newStore(t) + ctx := context.Background() + + first, err := s.Put(ctx, "IEI-1", filledIntent) + if err != nil { + t.Fatal(err) + } + if first.Mode != ModeAdvisory { + t.Errorf("mode = %s, want FLUID-2", first.Mode) + } + + // Recording the same content again must be a no-op, so startup is safe to + // repeat. + again, err := s.Put(ctx, "IEI-1", filledIntent) + if err != nil { + t.Fatalf("re-recording identical intent failed: %v", err) + } + if again.Digest != first.Digest { + t.Error("identical documents produced different digests") + } + + // Changing what a version says, after revisions may already be bound to it, + // would break the audit question the store exists to answer. + changed := strings.Replace(filledIntent, "FLUID-2", "FLUID-5", 1) + if _, err := s.Put(ctx, "IEI-1", changed); !errors.Is(err, ErrImmutable) { + t.Errorf("mutating a recorded version returned %v, want ErrImmutable", err) + } +} + +func TestParseModeRejectsUnresolvedTemplate(t *testing.T) { + // The shipped template lists every mode as alternatives. Accepting that and + // defaulting to something permissive is exactly the wrong failure. + template, err := os.ReadFile(filepath.Join("..", "..", "spec", "InterfaceEvolutionIntent.md")) + if err != nil { + t.Skipf("template not readable: %v", err) + } + if _, err := ParseMode(string(template)); err == nil { + t.Error("the unfilled template was accepted as a governing intent") + } + + if _, err := ParseMode("no mode declared here"); err == nil { + t.Error("a document with no authority mode was accepted") + } +} + +func TestAuthorityModeOrdering(t *testing.T) { + if !ModeExperimental.Allows(ModeAdvisory) { + t.Error("FLUID-4 should permit advisory authority") + } + if ModeAdvisory.Allows(ModeExperimental) { + t.Error("FLUID-2 must not permit experimental authority") + } + if got := ModeBoundedAutonomous.String(); got != "FLUID-5" { + t.Errorf("String() = %s", got) + } +} + +func TestActiveIntent(t *testing.T) { + s := newStore(t) + ctx := context.Background() + + if _, err := s.Active(ctx); !errors.Is(err, ErrNoActive) { + t.Errorf("Active with nothing recorded returned %v, want ErrNoActive", err) + } + + if _, err := s.Put(ctx, "IEI-1", filledIntent); err != nil { + t.Fatal(err) + } + if err := s.SetActive(ctx, "IEI-1"); err != nil { + t.Fatal(err) + } + + active, err := s.Active(ctx) + if err != nil { + t.Fatal(err) + } + if active.Version != "IEI-1" { + t.Errorf("active = %s, want IEI-1", active.Version) + } + + if err := s.SetActive(ctx, "IEI-missing"); err == nil { + t.Error("activating an unrecorded version succeeded") + } +} + +func TestBindingIsRecordedAndFirstWins(t *testing.T) { + s := newStore(t) + ctx := context.Background() + + if _, err := s.Put(ctx, "IEI-1", filledIntent); err != nil { + t.Fatal(err) + } + v2 := strings.Replace(filledIntent, "FLUID-2", "FLUID-4", 1) + if _, err := s.Put(ctx, "IEI-2", v2); err != nil { + t.Fatal(err) + } + + rev := contract.RevisionID("R-1") + if err := s.Bind(ctx, rev, "IEI-1"); err != nil { + t.Fatal(err) + } + + got, err := s.GoverningIntent(ctx, rev) + if err != nil { + t.Fatal(err) + } + if got.Version != "IEI-1" { + t.Errorf("governing intent = %s, want IEI-1", got.Version) + } + + // A later reassignment is recorded in the audit trail but must not silently + // become the answer to "what governed this revision". + if err := s.Bind(ctx, rev, "IEI-2"); err != nil { + t.Fatal(err) + } + got, err = s.GoverningIntent(ctx, rev) + if err != nil { + t.Fatal(err) + } + if got.Version != "IEI-1" { + t.Errorf("a later binding overrode the original: got %s", got.Version) + } +} + +func TestUnboundRevisionHasNoIntent(t *testing.T) { + s := newStore(t) + if _, err := s.GoverningIntent(context.Background(), "R-unbound"); !errors.Is(err, ErrNotFound) { + t.Errorf("got %v, want ErrNotFound", err) + } +} diff --git a/workplans/FLUID-WP-0003-deterministic-data-plane.md b/workplans/FLUID-WP-0003-deterministic-data-plane.md index ddc1002..8938a50 100644 --- a/workplans/FLUID-WP-0003-deterministic-data-plane.md +++ b/workplans/FLUID-WP-0003-deterministic-data-plane.md @@ -113,7 +113,7 @@ backpressure must never block or slow a request. Test it under a stalled sink. ```task id: FLUID-WP-0003-T08 -status: todo +status: done priority: high state_hub_task_id: "1773c7bf-9400-576e-9e74-aa210f08e0c4" ``` @@ -125,7 +125,7 @@ summaries exist only as derived views (§26). ```task id: FLUID-WP-0003-T09 -status: todo +status: done priority: high state_hub_task_id: "23d70a60-14bf-525e-a712-1d3a14111f62" ``` @@ -138,7 +138,7 @@ was made. ```task id: FLUID-WP-0003-T10 -status: todo +status: done priority: medium state_hub_task_id: "ad7d83d7-a088-5ea4-94e7-fa5be49897b9" ```