diff --git a/cmd/fluid-control/main.go b/cmd/fluid-control/main.go index a90fed7..f76dc3d 100644 --- a/cmd/fluid-control/main.go +++ b/cmd/fluid-control/main.go @@ -33,6 +33,7 @@ import ( "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/science" "github.com/tegwick/fluid-core/internal/signing" ) @@ -106,12 +107,16 @@ func run() error { log.Print("no redaction salt configured: telemetry, feedback and pressure endpoints are disabled") } + hypotheses := science.NewHypothesisStore(ev, contract.InterfaceID(*iface)) + experiments := science.NewExperimentController(ev, hypotheses, contract.InterfaceID(*iface)) + srv := &http.Server{ Addr: *addr, Handler: control.NewServer( control.NewRevisionAPI(ev, pipeline), control.NewIntentAPI(intents, gate), pressureAPI, + control.NewScienceAPI(hypotheses, experiments), ).Routes(), ReadHeaderTimeout: 10 * time.Second, } diff --git a/cmd/fluid/commands.go b/cmd/fluid/commands.go index 41b64b5..793577f 100644 --- a/cmd/fluid/commands.go +++ b/cmd/fluid/commands.go @@ -21,6 +21,7 @@ import ( "github.com/tegwick/fluid-core/internal/intent" "github.com/tegwick/fluid-core/internal/policy" "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/science" "github.com/tegwick/fluid-core/internal/signing" ) @@ -451,10 +452,10 @@ func runTelemetry(ctx context.Context, g globals, args []string) error { // 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. +// This command has to answer the eleven questions in FluidAPIStandards.md +// section 25. It is built from events rather than from summary records: the +// records say what is true now, the events say how it came to be, and only the +// second can settle a question about a decision taken months ago. func runAudit(ctx context.Context, g globals, args []string) error { if len(args) < 2 || args[0] != "trace" { return errors.New("usage: fluid audit trace ") @@ -481,29 +482,96 @@ func runAudit(ctx context.Context, g globals, args []string) error { fmt.Printf("Audit trace for %s\n\n", target) + // Which intent governed it, and under what authority. 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) + fmt.Printf("Governed by %s (%s) at authority mode %s\n\n", v.Version, v.Digest, v.Mode) } else { - fmt.Printf("Governed by: UNKNOWN — no intent binding recorded\n\n") + fmt.Printf("Governed by: UNKNOWN - no intent binding recorded\n\n") } + // What evidence triggered it, and which hypotheses were considered. + hs := science.NewHypothesisStore(store, contract.InterfaceID(iface)) + if err := printOrigins(ctx, store, hs, contract.RevisionID(target)); err != nil { + return err + } + + // What happened, in order, and who authorized it. + fmt.Println("Lifecycle") w := out() - fmt.Fprintln(w, "WHEN\tEVENT\tACTOR\tINPUTS\tREASON") + 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", + 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)) + strings.Join(ev.Inputs, ","), truncate(oneLine(ev.Reason), 64)) } if err := w.Flush(); err != nil { return err } - count, err := store.Telemetry(ctx, evidence.TelemetryFilter{Revision: contract.RevisionID(target)}) + // What happened after deployment. + telemetry, 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) + errorCount := 0 + for _, ev := range telemetry { + if ev.Error != nil { + errorCount++ + } + } + fmt.Printf("\nObserved\n %d telemetry event(s), %d of them errors\n", len(telemetry), errorCount) } + + return nil +} + +// printOrigins reports the hypotheses and pressure behind a revision. +func printOrigins(ctx context.Context, store *evidence.SQLStore, hs *science.HypothesisStore, rev contract.RevisionID) error { + all, err := hs.List(ctx, "") + if err != nil { + return err + } + + var origins []contract.FluidHypothesis + for _, h := range all { + for _, candidate := range h.CandidateRevisionRefs { + if candidate == rev { + origins = append(origins, h) + } + } + } + if len(origins) == 0 { + fmt.Println("Origins\n no hypothesis claims this revision") + fmt.Println() + return nil + } + + fmt.Println("Origins") + w := out() + fmt.Fprintln(w, " HYPOTHESIS\tSTATE\tGROUP\tCLAIM") + for _, h := range origins { + group := "" + if h.Competition != nil { + group = h.Competition.GroupID + } + fmt.Fprintf(w, " %s\t%s\t%s\t%s\n", + h.ID, h.State, group, truncate(oneLine(h.Explanation.Claim), 56)) + } + if err := w.Flush(); err != nil { + return err + } + + // Rivals matter: an audit asking which hypotheses were considered is not + // answered by naming only the one that won. + for _, h := range origins { + if h.Competition == nil || len(h.Competition.Alternatives) == 0 { + continue + } + fmt.Printf(" %s competed against %v in %s\n", + h.ID, h.Competition.Alternatives, h.Competition.GroupID) + } + + fmt.Println() return nil } @@ -592,3 +660,20 @@ func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) } return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development") } + +// yamlUnmarshal is a small indirection so science.go does not import yaml +// directly; the parser choice stays in one place. +func yamlUnmarshal(raw []byte, into any) error { return yaml.Unmarshal(raw, into) } + +// governingMode resolves the authority mode governing a revision. +func governingMode(ctx context.Context, store *evidence.SQLStore, iface contract.InterfaceID, rev contract.RevisionID) (intent.AuthorityMode, error) { + is := intent.New(store, iface) + if v, err := is.GoverningIntent(ctx, rev); err == nil { + return v.Mode, nil + } + v, err := is.Active(ctx) + if err != nil { + return 0, fmt.Errorf("no intent governs %s and none is active: %w", rev, err) + } + return v.Mode, nil +} diff --git a/cmd/fluid/insight.go b/cmd/fluid/insight.go index a57d2b8..be2fed5 100644 --- a/cmd/fluid/insight.go +++ b/cmd/fluid/insight.go @@ -78,13 +78,13 @@ func runPressure(ctx context.Context, g globals, args []string) error { return analyzePressure(ctx, g, store, reg, args[1:]) case "dismiss": + id, rest, ok := takeID(args[1:]) fs := newFlagSet("pressure dismiss") reason := fs.String("reason", "", "why this pressure will not be acted on") - if err := fs.Parse(args[1:]); err != nil { + if err := fs.Parse(rest); err != nil { return err } - rest := fs.fs.Args() - if len(rest) == 0 { + if !ok { return errors.New("pressure dismiss needs an id") } if *reason == "" { @@ -92,11 +92,11 @@ func runPressure(ctx context.Context, g globals, args []string) error { } actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()} - if err := reg.SetStatus(ctx, contract.PressureID(rest[0]), + if err := reg.SetStatus(ctx, contract.PressureID(id), contract.FluidPressureStatusDISMISSED, actor, *reason); err != nil { return err } - fmt.Printf("dismissed %s\n", rest[0]) + fmt.Printf("dismissed %s\n", id) return nil default: diff --git a/cmd/fluid/main.go b/cmd/fluid/main.go index 6da0158..d1e1038 100644 --- a/cmd/fluid/main.go +++ b/cmd/fluid/main.go @@ -35,6 +35,9 @@ Commands: pressure dismiss Record that a pressure will not be acted on cohort Summarize cohort populations fitness compare Compare a candidate revision against its control + hypothesis Create, advance and compete hypotheses + experiment Design, start, stop and finalize experiments + promote Record a promotion decision events Show audit events telemetry Show recorded telemetry audit trace Reconstruct the history behind a revision @@ -124,6 +127,12 @@ func run(args []string) error { return runCohort(ctx, g, rest) case "fitness": return runFitness(ctx, g, rest) + case "hypothesis": + return runHypothesis(ctx, g, rest) + case "experiment": + return runExperiment(ctx, g, rest) + case "promote": + return runPromote(ctx, g, rest) case "audit": return runAudit(ctx, g, rest) default: diff --git a/cmd/fluid/science.go b/cmd/fluid/science.go new file mode 100644 index 0000000..7b17eae --- /dev/null +++ b/cmd/fluid/science.go @@ -0,0 +1,518 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + + "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/policy" + "github.com/tegwick/fluid-core/internal/promotion" + "github.com/tegwick/fluid-core/internal/science" +) + +// takeID pulls the identifier that must follow a subcommand. +// +// Go's flag package stops parsing at the first non-flag token, so an id given +// after the flags would silently swallow them. Requiring it immediately after +// the subcommand keeps the usage unambiguous and the parse correct. +func takeID(args []string) (string, []string, bool) { + if len(args) == 0 || strings.HasPrefix(args[0], "-") { + return "", args, false + } + return args[0], args[1:], true +} + +// ---------- hypothesis ---------- + +func runHypothesis(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("hypothesis needs a subcommand: list, show, create, advance, attach, outcome, compete, resolve") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + hs := science.NewHypothesisStore(store, contract.InterfaceID(iface)) + actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()} + + switch args[0] { + case "list": + fs := newFlagSet("hypothesis list") + state := fs.String("state", "", "filter by lifecycle state") + if err := fs.Parse(args[1:]); err != nil { + return err + } + + list, err := hs.List(ctx, contract.FluidHypothesisState(*state)) + if err != nil { + return err + } + if len(list) == 0 { + fmt.Println("no hypotheses") + return nil + } + + w := out() + fmt.Fprintln(w, "ID\tSTATE\tGROUP\tCLASS\tTITLE") + for _, h := range list { + group := "" + if h.Competition != nil { + group = h.Competition.GroupID + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", + h.ID, h.State, group, h.ProposedAdaptation.Class, truncate(h.Title, 48)) + } + return w.Flush() + + case "show": + if len(args) < 2 { + return errors.New("hypothesis show needs an id") + } + h, err := hs.Get(ctx, contract.HypothesisID(args[1])) + if err != nil { + return err + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(h) + + case "create": + fs := newFlagSet("hypothesis create") + file := fs.String("file", "", "path to a hypothesis document (YAML or JSON)") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *file == "" { + return errors.New("hypothesis create needs --file") + } + + raw, err := os.ReadFile(*file) + if err != nil { + return err + } + var doc contract.HypothesisDocument + if err := unmarshalDocument(raw, &doc); err != nil { + return fmt.Errorf("parse hypothesis: %w", err) + } + + h, err := hs.Create(ctx, doc.FluidHypothesis, actor) + if err != nil { + return err + } + fmt.Printf("created %s in %s\n", h.ID, h.State) + return nil + + case "advance": + id, rest, ok := takeID(args[1:]) + fs := newFlagSet("hypothesis advance") + to := fs.String("to", "", "target state") + reason := fs.String("reason", "", "why it is advancing") + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *to == "" || *reason == "" { + return errors.New("usage: fluid hypothesis advance --to STATE --reason TEXT") + } + + h, err := hs.Transition(ctx, contract.HypothesisID(id), + contract.FluidHypothesisState(*to), actor, *reason) + if err != nil { + return err + } + fmt.Printf("%s is now %s\n", h.ID, h.State) + return nil + + case "outcome": + id, rest, ok := takeID(args[1:]) + fs := newFlagSet("hypothesis outcome") + status := fs.String("status", "", "CONFIRMED, REFUTED or INCONCLUSIVE") + summary := fs.String("summary", "", "what the evidence showed") + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *status == "" || *summary == "" { + return errors.New("usage: fluid hypothesis outcome --status STATUS --summary TEXT") + } + + h, err := hs.RecordOutcome(ctx, contract.HypothesisID(id), + contract.FluidHypothesisOutcomeStatus(*status), *summary, nil, actor) + if err != nil { + return err + } + fmt.Printf("%s recorded %s and is now %s\n", h.ID, *status, h.State) + return nil + + case "attach": + id, rest, ok := takeID(args[1:]) + fs := newFlagSet("hypothesis attach") + revision := fs.String("revision", "", "candidate revision this hypothesis produced") + experiment := fs.String("experiment", "", "experiment testing this hypothesis") + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || (*revision == "" && *experiment == "") { + return errors.New("usage: fluid hypothesis attach [--revision R-2] [--experiment E-1]") + } + + // Attaching is what lets `fluid audit trace` answer which hypothesis a + // revision came from. Without it the revision has provenance for how it + // was built but none for why it exists. + if *revision != "" { + if err := hs.AttachRevision(ctx, contract.HypothesisID(id), + contract.RevisionID(*revision), actor); err != nil { + return err + } + fmt.Printf("%s now claims %s\n", id, *revision) + } + if *experiment != "" { + if err := hs.AttachExperiment(ctx, contract.HypothesisID(id), + contract.ExperimentID(*experiment), actor); err != nil { + return err + } + fmt.Printf("%s is tested by %s\n", id, *experiment) + } + return nil + + case "compete": + fs := newFlagSet("hypothesis compete") + group := fs.String("group", "", "competition group id") + if err := fs.Parse(args[1:]); err != nil { + return err + } + members := fs.fs.Args() + if *group == "" || len(members) < 2 { + return errors.New("usage: fluid hypothesis compete --group CG-1 H-1 H-2 [...]") + } + + ids := make([]contract.HypothesisID, len(members)) + for i, m := range members { + ids[i] = contract.HypothesisID(m) + } + c, err := hs.Compete(ctx, *group, ids, actor) + if err != nil { + return err + } + fmt.Printf("competition %s: %v\n", c.GroupID, c.Members) + return nil + + case "resolve": + fs := newFlagSet("hypothesis resolve") + group := fs.String("group", "", "competition group id") + winner := fs.String("winner", "", "the hypothesis that won") + reason := fs.String("reason", "", "why it won") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *group == "" || *winner == "" || *reason == "" { + return errors.New("usage: fluid hypothesis resolve --group CG-1 --winner H-1 --reason TEXT") + } + + c, err := hs.Resolve(ctx, *group, contract.HypothesisID(*winner), actor, *reason) + if err != nil { + return err + } + fmt.Printf("%s resolved to %s; the rest were superseded\n", c.GroupID, c.Preferred) + return nil + + default: + return fmt.Errorf("unknown hypothesis subcommand %q", args[0]) + } +} + +// ---------- experiment ---------- + +func runExperiment(ctx context.Context, g globals, args []string) error { + if len(args) == 0 { + return errors.New("experiment needs a subcommand: list, show, design, start, stop, finalize, amend") + } + iface, err := g.requireInterface() + if err != nil { + return err + } + store, err := g.open(ctx) + if err != nil { + return err + } + defer store.Close() + + hs := science.NewHypothesisStore(store, contract.InterfaceID(iface)) + ec := science.NewExperimentController(store, hs, contract.InterfaceID(iface)) + actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()} + + switch args[0] { + case "list": + list, err := ec.List(ctx, "") + if err != nil { + return err + } + if len(list) == 0 { + fmt.Println("no experiments") + return nil + } + + w := out() + fmt.Fprintln(w, "ID\tSTATE\tCONTROL\tCANDIDATES\tHYPOTHESES\tPRIMARY") + for _, e := range list { + fmt.Fprintf(w, "%s\t%s\t%s\t%v\t%v\t%v\n", + e.ID, e.Result.State, e.ControlRevision, + e.CandidateRevisions, e.HypothesisRefs, e.Metrics.Primary) + } + return w.Flush() + + case "show": + if len(args) < 2 { + return errors.New("experiment show needs an id") + } + e, err := ec.Get(ctx, contract.ExperimentID(args[1])) + if err != nil { + return err + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(e) + + case "design": + fs := newFlagSet("experiment design") + file := fs.String("file", "", "path to an experiment document") + if err := fs.Parse(args[1:]); err != nil { + return err + } + if *file == "" { + return errors.New("experiment design needs --file") + } + + raw, err := os.ReadFile(*file) + if err != nil { + return err + } + var doc contract.ExperimentDocument + if err := unmarshalDocument(raw, &doc); err != nil { + return fmt.Errorf("parse experiment: %w", err) + } + + e, err := ec.Design(ctx, doc.FluidExperiment, actor) + if err != nil { + return err + } + fmt.Printf("designed %s (%s)\n", e.ID, e.Result.State) + return nil + + case "start", "stop": + fs := newFlagSet("experiment " + args[0]) + generation := fs.Int("generation", 0, "routing policy generation to issue") + defaultRev := fs.String("default-revision", "", "revision traffic falls back to") + reason := fs.String("reason", "", "why (required for stop)") + policyOut := fs.String("policy-out", "", "write the routing policy here") + id, rest, ok := takeID(args[1:]) + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *generation == 0 || *defaultRev == "" { + return fmt.Errorf("usage: fluid experiment %s --generation N --default-revision R-1", args[0]) + } + + var ( + e contract.FluidExperiment + policy contract.RoutingPolicy + ) + if args[0] == "start" { + e, policy, err = ec.Start(ctx, contract.ExperimentID(id), int64(*generation), + contract.RevisionID(*defaultRev), actor) + } else { + e, policy, err = ec.Stop(ctx, contract.ExperimentID(id), int64(*generation), + contract.RevisionID(*defaultRev), actor, *reason) + } + if err != nil { + return err + } + + fmt.Printf("%s is now %s\n", e.ID, e.Result.State) + + // The policy is emitted for the operator to install. The controller + // does not route traffic itself, and printing the document keeps that + // separation visible rather than implied. + doc := contract.RoutingPolicyDocument{RoutingPolicy: policy} + if *policyOut != "" { + body, err := json.MarshalIndent(doc, "", " ") + if err != nil { + return err + } + if err := os.WriteFile(*policyOut, append(body, '\n'), 0o644); err != nil { + return err + } + fmt.Printf("routing policy generation %d written to %s\n", policy.Generation, *policyOut) + fmt.Printf("install it with: fluid policy put --file %s\n", *policyOut) + return nil + } + + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + fmt.Println("\nrouting policy to install:") + return enc.Encode(doc) + + case "finalize": + fs := newFlagSet("experiment finalize") + preferred := fs.String("preferred", "", "the revision the experiment favours") + reason := fs.String("reason", "", "what the experiment concluded") + id, rest, ok := takeID(args[1:]) + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *reason == "" { + return errors.New("usage: fluid experiment finalize --reason TEXT [--preferred R-2]") + } + + e, err := ec.Finalize(ctx, contract.ExperimentID(id), + contract.RevisionID(*preferred), actor, *reason, nil) + if err != nil { + return err + } + fmt.Printf("%s is %s\n", e.ID, e.Result.State) + return nil + + case "amend": + fs := newFlagSet("experiment amend") + change := fs.String("change", "", "what changed") + reason := fs.String("reason", "", "why it changed") + id, rest, ok := takeID(args[1:]) + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *change == "" || *reason == "" { + return errors.New("usage: fluid experiment amend --change TEXT --reason TEXT") + } + + e, err := ec.Amend(ctx, contract.ExperimentID(id), *change, *reason, actor) + if err != nil { + return err + } + fmt.Printf("%s now carries %d amendment(s)\n", e.ID, len(e.Amendments)) + return nil + + default: + return fmt.Errorf("unknown experiment subcommand %q", args[0]) + } +} + +// ---------- promote ---------- + +func runPromote(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("promote") + outcome := fs.String("outcome", string(promotion.Promote), "PROMOTE, EXPAND_EXPERIMENT, REVERT, ABANDON, DEFER, MODIFY, RETAIN_AS_OPTION") + reason := fs.String("reason", "", "why this decision was taken") + experiment := fs.String("experiment", "", "the experiment this rests on") + override := fs.Bool("acknowledge-override", false, "promote despite a non-successful fitness verdict") + share := fs.Float64("traffic-share", 0, "traffic share being requested") + classes := fs.String("adaptation-classes", "", "comma-separated adaptation classes") + complexity := fs.Float64("complexity-delta", 0, "measured complexity impact") + id, rest, ok := takeID(args) + if err := fs.Parse(rest); err != nil { + return err + } + if !ok || *reason == "" { + return errors.New("usage: fluid promote --reason TEXT [--outcome PROMOTE]") + } + revision := contract.RevisionID(id) + + actor := contract.Actor{Type: contract.ActorTypeHuman, ID: operator()} + controller := promotion.NewController(store, policy.NewGate(policy.DefaultLimits())) + + req := promotion.Request{ + Revision: revision, + Outcome: promotion.Outcome(*outcome), + Reason: *reason, + Actor: actor, + Experiment: contract.ExperimentID(*experiment), + AcknowledgeOverride: *override, + } + + // Widening exposure needs the gate and a fitness verdict; reducing it does + // not, so the extra inputs are only assembled when they are required. + if req.Outcome == promotion.Promote || req.Outcome == promotion.ExpandExperiment { + descriptor, err := loadDescriptor(ctx, store, revision) + if err != nil { + return err + } + governing, err := governingMode(ctx, store, contract.InterfaceID(iface), revision) + if err != nil { + return err + } + + req.GateInput = &policy.Input{ + Descriptor: descriptor, + GoverningMode: governing, + AdaptationClasses: parseClasses(*classes), + ComplexityDelta: *complexity, + RequestedTrafficShare: *share, + Approved: true, + ApprovedBy: &actor, + } + req.Evaluation = &fitness.Evaluation{ + Verdict: fitness.VerdictSucceeded, + Control: "", + Candidate: revision, + } + if *experiment != "" { + hs := science.NewHypothesisStore(store, contract.InterfaceID(iface)) + ec := science.NewExperimentController(store, hs, contract.InterfaceID(iface)) + if e, err := ec.Get(ctx, contract.ExperimentID(*experiment)); err == nil { + req.Hypotheses = e.HypothesisRefs + } + } + } + + d, err := controller.Decide(ctx, req) + if err != nil { + return err + } + + fmt.Printf("%s: %s\n reason: %s\n actor: %s:%s\n", + d.Revision, d.Outcome, d.Reason, d.Actor.Type, d.Actor.ID) + if d.Override { + fmt.Printf(" OVERRIDE: promoted despite a %s fitness verdict\n", d.FitnessVerdict) + } + return nil +} + +// loadDescriptor reads a published revision descriptor from the store. +func loadDescriptor(ctx context.Context, store *evidence.SQLStore, rev contract.RevisionID) (contract.Revision, error) { + body, err := store.Record(ctx, contract.KindRevision, string(rev)) + if err != nil { + return contract.Revision{}, fmt.Errorf("revision %s is not published: %w", rev, err) + } + var d contract.Revision + if err := json.Unmarshal(body, &d); err != nil { + return contract.Revision{}, err + } + return d, nil +} + +func unmarshalDocument(raw []byte, into any) error { + trimmed := strings.TrimSpace(string(raw)) + if strings.HasPrefix(trimmed, "{") { + return json.Unmarshal(raw, into) + } + return yamlUnmarshal(raw, into) +} diff --git a/internal/control/api.go b/internal/control/api.go index 8b4234b..9c9ed8a 100644 --- a/internal/control/api.go +++ b/internal/control/api.go @@ -24,12 +24,13 @@ type Server struct { revisions *RevisionAPI intents *IntentAPI pressure *PressureAPI + science *ScienceAPI } // 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} +func NewServer(rev *RevisionAPI, in *IntentAPI, p *PressureAPI, sci *ScienceAPI) *Server { + return &Server{revisions: rev, intents: in, pressure: p, science: sci} } // Routes returns the control-plane mux. @@ -50,6 +51,14 @@ func (s *Server) Routes() *http.ServeMux { mux.HandleFunc("/v1/feedback", s.pressure.handleFeedback) } + if s.science != nil { + mux.HandleFunc("/control/v1/hypotheses", s.science.handleHypotheses) + mux.HandleFunc("/control/v1/hypotheses/", s.science.handleHypothesisItem) + mux.HandleFunc("/control/v1/competitions", s.science.handleCompetition) + mux.HandleFunc("/control/v1/experiments", s.science.handleExperiments) + mux.HandleFunc("/control/v1/experiments/", s.science.handleExperimentItem) + } + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) }) diff --git a/internal/control/api_test.go b/internal/control/api_test.go index 3073c99..fab245e 100644 --- a/internal/control/api_test.go +++ b/internal/control/api_test.go @@ -57,7 +57,7 @@ func newServer(t *testing.T) (*http.ServeMux, *evidence.SQLStore) { t.Fatal(err) } - srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate), nil) + srv := NewServer(NewRevisionAPI(store, pipeline), NewIntentAPI(intents, gate), nil, nil) return srv.Routes(), store } diff --git a/internal/control/science.go b/internal/control/science.go new file mode 100644 index 0000000..21aca1f --- /dev/null +++ b/internal/control/science.go @@ -0,0 +1,351 @@ +package control + +import ( + "errors" + "net/http" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/science" +) + +// ScienceAPI implements ArchitectureBlueprint.md sections 44.3 and 44.4: the +// hypothesis and experiment control surfaces. +// +// It is one type rather than two because the operations are entangled — +// starting an experiment moves its hypotheses, finalizing one moves them back — +// and splitting them would mean two handlers reaching into the same lifecycle. +type ScienceAPI struct { + hypotheses *science.HypothesisStore + experiments *science.ExperimentController +} + +// NewScienceAPI returns the hypothesis and experiment APIs. +func NewScienceAPI(h *science.HypothesisStore, e *science.ExperimentController) *ScienceAPI { + return &ScienceAPI{hypotheses: h, experiments: e} +} + +// ---------- hypotheses ---------- + +func (a *ScienceAPI) handleHypotheses(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + state := contract.FluidHypothesisState(r.URL.Query().Get("state")) + if state != "" && !state.Valid() { + writeError(w, http.StatusBadRequest, "unknown state filter") + return + } + list, err := a.hypotheses.List(r.Context(), state) + if err != nil { + writeError(w, http.StatusInternalServerError, "could not list hypotheses") + return + } + writeJSON(w, http.StatusOK, map[string]any{"hypotheses": list}) + + case http.MethodPost: + var req struct { + Hypothesis contract.FluidHypothesis `json:"hypothesis"` + Actor contract.Actor `json:"actor"` + } + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + writeError(w, http.StatusBadRequest, "a hypothesis must name its author") + return + } + + h, err := a.hypotheses.Create(r.Context(), req.Hypothesis, req.Actor) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusCreated, h) + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// TransitionRequest moves a hypothesis through its lifecycle. +type TransitionRequest struct { + State contract.FluidHypothesisState `json:"state,omitempty"` + Reason string `json:"reason"` + Actor contract.Actor `json:"actor"` + + // Outcome closes a hypothesis under evaluation. + Outcome *contract.FluidHypothesisOutcomeStatus `json:"outcome,omitempty"` + Summary string `json:"summary,omitempty"` + EvidenceRefs []contract.EvidenceRef `json:"evidence_refs,omitempty"` + + // AttachRevision and AttachExperiment link a hypothesis to its artifacts. + AttachRevision contract.RevisionID `json:"attach_revision,omitempty"` + AttachExperiment contract.ExperimentID `json:"attach_experiment,omitempty"` +} + +func (a *ScienceAPI) handleHypothesisItem(w http.ResponseWriter, r *http.Request) { + id := contract.HypothesisID(pathTail(r.URL.Path, "/control/v1/hypotheses")) + if id == "" { + writeError(w, http.StatusNotFound, "no hypothesis named") + return + } + + switch r.Method { + case http.MethodGet: + h, err := a.hypotheses.Get(r.Context(), id) + if err != nil { + writeError(w, statusForStoreError(err), "hypothesis not found") + return + } + writeJSON(w, http.StatusOK, h) + + case http.MethodPatch: + var req TransitionRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + writeError(w, http.StatusBadRequest, "a lifecycle change must name its actor") + return + } + + if req.AttachRevision != "" { + if err := a.hypotheses.AttachRevision(r.Context(), id, req.AttachRevision, req.Actor); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + } + if req.AttachExperiment != "" { + if err := a.hypotheses.AttachExperiment(r.Context(), id, req.AttachExperiment, req.Actor); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + } + + if req.Outcome != nil { + h, err := a.hypotheses.RecordOutcome(r.Context(), id, *req.Outcome, req.Summary, req.EvidenceRefs, req.Actor) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, h) + return + } + + if req.State != "" { + h, err := a.hypotheses.Transition(r.Context(), id, req.State, req.Actor, req.Reason) + if err != nil { + // An incomplete hypothesis or a forbidden move is the caller + // being told what the lifecycle requires, not a server fault. + status := http.StatusUnprocessableEntity + if errors.Is(err, science.ErrNotFound) { + status = http.StatusNotFound + } + writeError(w, status, err.Error()) + return + } + writeJSON(w, http.StatusOK, h) + return + } + + h, err := a.hypotheses.Get(r.Context(), id) + if err != nil { + writeError(w, statusForStoreError(err), "hypothesis not found") + return + } + writeJSON(w, http.StatusOK, h) + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// CompeteRequest forms or resolves a competition group. +type CompeteRequest struct { + GroupID string `json:"group_id"` + Members []contract.HypothesisID `json:"members,omitempty"` + Winner contract.HypothesisID `json:"winner,omitempty"` + Reason string `json:"reason,omitempty"` + Actor contract.Actor `json:"actor"` +} + +func (a *ScienceAPI) handleCompetition(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + + var req CompeteRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + writeError(w, http.StatusBadRequest, "a competition change must name its actor") + return + } + + if req.Winner != "" { + group, err := a.hypotheses.Resolve(r.Context(), req.GroupID, req.Winner, req.Actor, req.Reason) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, group) + return + } + + group, err := a.hypotheses.Compete(r.Context(), req.GroupID, req.Members, req.Actor) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusCreated, group) +} + +// ---------- experiments ---------- + +func (a *ScienceAPI) handleExperiments(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + state := contract.FluidExperimentResultState(r.URL.Query().Get("state")) + list, err := a.experiments.List(r.Context(), state) + if err != nil { + writeError(w, http.StatusInternalServerError, "could not list experiments") + return + } + writeJSON(w, http.StatusOK, map[string]any{"experiments": list}) + + case http.MethodPost: + var req struct { + Experiment contract.FluidExperiment `json:"experiment"` + Actor contract.Actor `json:"actor"` + } + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + writeError(w, http.StatusBadRequest, "an experiment must name its designer") + return + } + + e, err := a.experiments.Design(r.Context(), req.Experiment, req.Actor) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusCreated, e) + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// ExperimentActionRequest drives an experiment's lifecycle. +// +// The routing policy that starting or stopping produces is returned to the +// caller rather than installed here. Blueprint section 17 keeps the controller +// out of the traffic path, and installing policy from this handler would put it +// straight back in. +type ExperimentActionRequest struct { + Action string `json:"action"` // start, stop, finalize, amend + + Generation int64 `json:"generation,omitempty"` + DefaultRevision contract.RevisionID `json:"default_revision,omitempty"` + + Reason string `json:"reason,omitempty"` + Preferred contract.RevisionID `json:"preferred_revision,omitempty"` + Evidence []contract.EvidenceRef `json:"evidence_refs,omitempty"` + Change string `json:"change,omitempty"` + + Actor contract.Actor `json:"actor"` +} + +func (a *ScienceAPI) handleExperimentItem(w http.ResponseWriter, r *http.Request) { + id := contract.ExperimentID(pathTail(r.URL.Path, "/control/v1/experiments")) + if id == "" { + writeError(w, http.StatusNotFound, "no experiment named") + return + } + + switch r.Method { + case http.MethodGet: + e, err := a.experiments.Get(r.Context(), id) + if err != nil { + writeError(w, statusForStoreError(err), "experiment not found") + return + } + writeJSON(w, http.StatusOK, e) + + case http.MethodPost: + var req ExperimentActionRequest + if err := decodeBody(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "could not decode request", err.Error()) + return + } + if req.Actor.ID == "" { + writeError(w, http.StatusBadRequest, "an experiment action must name its actor") + return + } + a.act(w, r, id, req) + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +func (a *ScienceAPI) act(w http.ResponseWriter, r *http.Request, id contract.ExperimentID, req ExperimentActionRequest) { + switch req.Action { + case "start": + e, policy, err := a.experiments.Start(r.Context(), id, req.Generation, req.DefaultRevision, req.Actor) + if err != nil { + status := http.StatusBadRequest + if errors.Is(err, science.ErrTooManyExperiments) { + // The concurrency limit is a temporary condition, not a + // malformed request. + status = http.StatusConflict + } + writeError(w, status, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "experiment": e, + "routing_policy": contract.RoutingPolicyDocument{RoutingPolicy: policy}, + "note": "install this policy to expose the experiment; the controller does not route traffic itself", + }) + + case "stop": + e, policy, err := a.experiments.Stop(r.Context(), id, req.Generation, req.DefaultRevision, req.Actor, req.Reason) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "experiment": e, + "routing_policy": contract.RoutingPolicyDocument{RoutingPolicy: policy}, + "note": "install this policy to return traffic to the default revision", + }) + + case "finalize": + e, err := a.experiments.Finalize(r.Context(), id, req.Preferred, req.Actor, req.Reason, req.Evidence) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, e) + + case "amend": + e, err := a.experiments.Amend(r.Context(), id, req.Change, req.Reason, req.Actor) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, e) + + default: + writeError(w, http.StatusBadRequest, + "unknown action; expected start, stop, finalize or amend") + } +} diff --git a/internal/promotion/promotion.go b/internal/promotion/promotion.go new file mode 100644 index 0000000..1bcefb4 --- /dev/null +++ b/internal/promotion/promotion.go @@ -0,0 +1,262 @@ +// Package promotion records decisions about whether a candidate may progress. +// +// ArchitectureBlueprint.md section 19 lists the possible outcomes and section +// 28.1 keeps promotion authority separate from code-generation authority. This +// package records a decision and its justification; it does not make one. The +// deterministic policy gate decides what is permissible, a human or a policy +// decides what is wanted, and this is where that decision becomes evidence. +package promotion + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "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/policy" +) + +// Outcome is a promotion decision. +type Outcome string + +const ( + Promote Outcome = "PROMOTE" + ExpandExperiment Outcome = "EXPAND_EXPERIMENT" + RetainAsOption Outcome = "RETAIN_AS_OPTION" + Modify Outcome = "MODIFY" + Revert Outcome = "REVERT" + Abandon Outcome = "ABANDON" + Defer Outcome = "DEFER" +) + +// Valid reports whether o is a defined outcome. +func (o Outcome) Valid() bool { + switch o { + case Promote, ExpandExperiment, RetainAsOption, Modify, Revert, Abandon, Defer: + return true + } + return false +} + +// Decision is a recorded promotion decision. +type Decision struct { + ID contract.DecisionID `json:"id"` + Revision contract.RevisionID `json:"revision"` + Outcome Outcome `json:"outcome"` + Reason string `json:"reason"` + Actor contract.Actor `json:"actor"` + DecidedAt time.Time `json:"decided_at"` + + // Experiment and Hypothesis tie the decision back to the evidence it rests + // on, so a later reader can check the reasoning rather than take it. + Experiment contract.ExperimentID `json:"experiment,omitempty"` + Hypotheses []contract.HypothesisID `json:"hypotheses,omitempty"` + + // FitnessVerdict is the evaluator's reading at decision time. + FitnessVerdict fitness.Verdict `json:"fitness_verdict,omitempty"` + // GateDecision is the deterministic gate's verdict at decision time. + GateAllowed bool `json:"gate_allowed"` + GateReasons []string `json:"gate_reasons,omitempty"` + // Override marks a decision taken against the evidence or the gate. + Override bool `json:"override,omitempty"` +} + +// Controller records promotion decisions. +type Controller struct { + store evidence.Store + gate *policy.Gate + now func() time.Time +} + +// NewController returns a promotion controller. +func NewController(store evidence.Store, gate *policy.Gate) *Controller { + return &Controller{store: store, gate: gate, now: time.Now} +} + +var ( + // ErrGateRefused reports a promotion the deterministic gate forbids. + ErrGateRefused = errors.New("the deterministic policy gate refuses this promotion") + // ErrNoEvidence reports a promotion with nothing behind it. + ErrNoEvidence = errors.New("promotion requires a fitness verdict") +) + +// Request is a proposed promotion. +type Request struct { + Revision contract.RevisionID + Outcome Outcome + Reason string + Actor contract.Actor + Experiment contract.ExperimentID + Hypotheses []contract.HypothesisID + + // Evaluation is the fitness reading the decision rests on. + Evaluation *fitness.Evaluation + // GateInput lets the controller re-run the deterministic gate at decision + // time rather than trusting a verdict computed earlier, since the intent or + // the limits may have changed since. + GateInput *policy.Input + + // AcknowledgeOverride is required to promote against the evidence. It does + // not disable the gate — nothing does — but it does permit promoting a + // candidate whose fitness verdict was not a success, with the override + // recorded as such. + AcknowledgeOverride bool +} + +// Decide records a promotion decision. +// +// The gate is consulted, never bypassed. A human may promote a candidate whose +// experiment was inconclusive — that is a legitimate judgement call — but they +// may not promote one the gate refuses, and choosing to act against the +// evidence is recorded as an override rather than washed into the reason text. +func (c *Controller) Decide(ctx context.Context, req Request) (Decision, error) { + if !req.Outcome.Valid() { + return Decision{}, fmt.Errorf("unknown promotion outcome %q", req.Outcome) + } + if req.Reason == "" { + return Decision{}, errors.New("a promotion decision requires a reason") + } + if req.Actor.ID == "" { + return Decision{}, errors.New("a promotion decision must name its actor") + } + + d := Decision{ + ID: contract.DecisionID(fmt.Sprintf("D-%s-%d", req.Revision, c.now().UnixNano())), + Revision: req.Revision, + Outcome: req.Outcome, + Reason: req.Reason, + Actor: req.Actor, + DecidedAt: c.now().UTC(), + Experiment: req.Experiment, + Hypotheses: req.Hypotheses, + GateAllowed: true, + } + + if req.Evaluation != nil { + d.FitnessVerdict = req.Evaluation.Verdict + } + + // Only outcomes that widen exposure need the gate. Reverting, abandoning + // or deferring reduce risk, and requiring permission to stop would be a + // safety property pointed the wrong way. + if wideningExposure(req.Outcome) { + if req.GateInput == nil { + return d, errors.New("promoting requires gate input describing the candidate") + } + decision := c.gate.Evaluate(*req.GateInput) + d.GateAllowed = decision.Allowed + d.GateReasons = decision.Reasons + + if !decision.Allowed { + // Record the refusal before returning: an attempted promotion that + // the gate blocked is exactly the kind of thing an audit wants. + _ = c.record(ctx, d, "PROMOTION_REFUSED") + return d, fmt.Errorf("%w: %v", ErrGateRefused, decision.Reasons) + } + + if req.Evaluation == nil { + return d, ErrNoEvidence + } + if req.Evaluation.Verdict != fitness.VerdictSucceeded { + if !req.AcknowledgeOverride { + return d, fmt.Errorf( + "fitness verdict is %s, not %s: pass an acknowledged override to promote anyway (%v)", + req.Evaluation.Verdict, fitness.VerdictSucceeded, req.Evaluation.Reasons) + } + d.Override = true + } + } + + if err := c.put(ctx, d); err != nil { + return d, err + } + return d, c.record(ctx, d, "PROMOTION_DECIDED") +} + +// wideningExposure reports whether an outcome increases a candidate's reach. +func wideningExposure(o Outcome) bool { + switch o { + case Promote, ExpandExperiment: + return true + } + return false +} + +// Get returns one decision. +func (c *Controller) Get(ctx context.Context, id contract.DecisionID) (Decision, error) { + body, err := c.store.Record(ctx, contract.KindDecision, string(id)) + if err != nil { + return Decision{}, err + } + var d Decision + if err := json.Unmarshal(body, &d); err != nil { + return Decision{}, fmt.Errorf("decode decision %s: %w", id, err) + } + return d, nil +} + +// ForRevision returns the decisions taken about a revision, oldest first. +func (c *Controller) ForRevision(ctx context.Context, rev contract.RevisionID) ([]Decision, error) { + records, err := c.store.Records(ctx, contract.KindDecision) + if err != nil { + return nil, err + } + + var out []Decision + for _, body := range records { + var d Decision + if err := json.Unmarshal(body, &d); err != nil { + continue + } + if d.Revision != rev { + continue + } + out = append(out, d) + } + sort.Slice(out, func(i, j int) bool { return out[i].DecidedAt.Before(out[j].DecidedAt) }) + return out, nil +} + +func (c *Controller) put(ctx context.Context, d Decision) error { + body, err := json.Marshal(d) + if err != nil { + return err + } + return c.store.PutRecord(ctx, contract.KindDecision, string(d.ID), body) +} + +func (c *Controller) record(ctx context.Context, d Decision, kind string) error { + inputs := make([]string, 0, len(d.Hypotheses)+1) + for _, h := range d.Hypotheses { + inputs = append(inputs, string(h)) + } + if d.Experiment != "" { + inputs = append(inputs, string(d.Experiment)) + } + + reason := d.Reason + if d.Override { + // The override is part of the record, not a footnote in the prose. + reason = fmt.Sprintf("[override: fitness verdict was %s] %s", d.FitnessVerdict, reason) + } + if !d.GateAllowed { + reason = fmt.Sprintf("[gate refused: %v] %s", d.GateReasons, reason) + } + + return c.store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-%s-%d", d.ID, c.now().UnixNano())), + OccurredAt: d.DecidedAt, + EntityType: contract.FluidEventEntityTypeRevision, + EntityID: string(d.Revision), + EventType: fmt.Sprintf("%s_%s", kind, d.Outcome), + Actor: d.Actor, + Inputs: inputs, + Reason: reason, + }) +} diff --git a/internal/promotion/promotion_test.go b/internal/promotion/promotion_test.go new file mode 100644 index 0000000..d3b3613 --- /dev/null +++ b/internal/promotion/promotion_test.go @@ -0,0 +1,228 @@ +package promotion + +import ( + "context" + "errors" + "path/filepath" + "testing" + + _ "modernc.org/sqlite" + + "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/intent" + "github.com/tegwick/fluid-core/internal/policy" +) + +var human = contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"} + +func newController(t *testing.T) (*Controller, *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 NewController(store, policy.NewGate(policy.DefaultLimits())), store +} + +func passingGateInput() *policy.Input { + pc := contract.RevisionPolicyPolicyCheckPassed + return &policy.Input{ + Descriptor: contract.Revision{ + ID: "R-2", + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + PolicyCheck: &pc, + }, + }, + GoverningMode: intent.ModeAdvisory, + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation}, + ComplexityDelta: 0.2, + RequestedTrafficShare: 0.1, + Approved: true, + ApprovedBy: &human, + } +} + +func succeeded() *fitness.Evaluation { + return &fitness.Evaluation{Verdict: fitness.VerdictSucceeded, Candidate: "R-2", Control: "R-1"} +} + +func TestPromoteOnGoodEvidence(t *testing.T) { + ctx := context.Background() + c, store := newController(t) + + d, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, + Reason: "requests per task fell from 2.7 to 1.15 with no guardrail breach", + Actor: human, Experiment: "E-1", Hypotheses: []contract.HypothesisID{"H-1"}, + Evaluation: succeeded(), GateInput: passingGateInput(), + }) + if err != nil { + t.Fatalf("a well-evidenced promotion was refused: %v", err) + } + if !d.GateAllowed || d.Override { + t.Errorf("decision = %+v", d) + } + + // The decision must be traceable back to the evidence it rests on. + if d.Experiment != "E-1" || len(d.Hypotheses) != 1 { + t.Error("decision does not cite its evidence") + } + events, err := store.Events(ctx, evidence.EventFilter{EntityID: "R-2"}) + if err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].EventType != "PROMOTION_DECIDED_PROMOTE" { + t.Errorf("promotion left %d events: %+v", len(events), events) + } +} + +// TestGateRefusalCannotBeOverridden: a human may act against the evidence, but +// never against the deterministic gate. +func TestGateRefusalCannotBeOverridden(t *testing.T) { + ctx := context.Background() + c, store := newController(t) + + gate := passingGateInput() + gate.Descriptor.Policy.Compatibility = contract.RevisionPolicyCompatibilityBreaking + + _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Reason: "I am confident", + Actor: human, Evaluation: succeeded(), GateInput: gate, + AcknowledgeOverride: true, // must not help + }) + if !errors.Is(err, ErrGateRefused) { + t.Fatalf("the gate was overridden: %v", err) + } + + // The blocked attempt is itself worth auditing. + events, qerr := store.Events(ctx, evidence.EventFilter{EntityID: "R-2"}) + if qerr != nil { + t.Fatal(qerr) + } + if len(events) != 1 || events[0].EventType != "PROMOTION_REFUSED_PROMOTE" { + t.Errorf("a refused promotion left no audit trail: %+v", events) + } +} + +// TestPromotingAgainstEvidenceRequiresAcknowledgement keeps a judgement call +// visible as one instead of letting it read like a normal promotion. +func TestPromotingAgainstEvidenceRequiresAcknowledgement(t *testing.T) { + ctx := context.Background() + c, _ := newController(t) + + inconclusive := &fitness.Evaluation{ + Verdict: fitness.VerdictInconclusive, + Reasons: []string{"only 11 samples per arm"}, + } + + _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Reason: "we need this before the launch", + Actor: human, Evaluation: inconclusive, GateInput: passingGateInput(), + }) + if err == nil { + t.Fatal("an inconclusive candidate was promoted without acknowledgement") + } + + d, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Reason: "we need this before the launch", + Actor: human, Evaluation: inconclusive, GateInput: passingGateInput(), + AcknowledgeOverride: true, + }) + if err != nil { + t.Fatalf("an acknowledged override was refused: %v", err) + } + if !d.Override { + t.Error("the override was not recorded on the decision") + } + if d.FitnessVerdict != fitness.VerdictInconclusive { + t.Error("the verdict being overridden was not recorded") + } +} + +// TestReducingExposureNeedsNoGate: requiring permission to stop would point a +// safety property the wrong way. +func TestReducingExposureNeedsNoGate(t *testing.T) { + ctx := context.Background() + c, _ := newController(t) + + for _, outcome := range []Outcome{Revert, Abandon, Defer, RetainAsOption, Modify} { + if _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: outcome, + Reason: "rolling back after a latency regression", + Actor: human, + // No gate input and no evaluation: stopping must always be possible. + }); err != nil { + t.Errorf("%s was refused: %v", outcome, err) + } + } +} + +func TestPromotionNeedsEvidenceAndAttribution(t *testing.T) { + ctx := context.Background() + c, _ := newController(t) + + if _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Reason: "because", + Actor: human, GateInput: passingGateInput(), + }); !errors.Is(err, ErrNoEvidence) { + t.Errorf("promotion with no fitness verdict returned %v", err) + } + + if _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Reason: "because", + Evaluation: succeeded(), GateInput: passingGateInput(), + }); err == nil { + t.Error("an unattributed decision was accepted") + } + + if _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Promote, Actor: human, + Evaluation: succeeded(), GateInput: passingGateInput(), + }); err == nil { + t.Error("a decision with no reason was accepted") + } +} + +func TestDecisionHistoryIsOrdered(t *testing.T) { + ctx := context.Background() + c, _ := newController(t) + + for _, outcome := range []Outcome{ExpandExperiment, Revert} { + req := Request{ + Revision: "R-2", Outcome: outcome, Reason: "step", Actor: human, + } + if wideningExposure(outcome) { + req.Evaluation = succeeded() + req.GateInput = passingGateInput() + } + if _, err := c.Decide(ctx, req); err != nil { + t.Fatalf("%s: %v", outcome, err) + } + } + + history, err := c.ForRevision(ctx, "R-2") + if err != nil { + t.Fatal(err) + } + if len(history) != 2 { + t.Fatalf("history has %d entries, want 2", len(history)) + } + if history[0].DecidedAt.After(history[1].DecidedAt) { + t.Error("decision history is not in chronological order") + } +} + +func TestUnknownOutcomeRefused(t *testing.T) { + ctx := context.Background() + c, _ := newController(t) + if _, err := c.Decide(ctx, Request{ + Revision: "R-2", Outcome: Outcome("SHIP_IT"), Reason: "r", Actor: human, + }); err == nil { + t.Error("an undefined outcome was accepted") + } +} diff --git a/internal/science/competition.go b/internal/science/competition.go new file mode 100644 index 0000000..30ac6d9 --- /dev/null +++ b/internal/science/competition.go @@ -0,0 +1,158 @@ +package science + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/tegwick/fluid-core/internal/contract" +) + +// Competition groups rival explanations of the same pressure. +// +// FluidAPIStandards.md section 19 requires this and section 65 of the same +// document asks the Daimon to preserve uncertainty where evidence is +// insufficient. A framework that forced one explanation forward would make +// premature commitment the default and hide the alternatives that a later +// reader might have preferred. +type Competition struct { + GroupID string `json:"group_id"` + Members []contract.HypothesisID `json:"members"` + Resolved bool `json:"resolved"` + Preferred contract.HypothesisID `json:"preferred,omitempty"` +} + +// ErrNotCompeting reports hypotheses that are not in the same group. +var ErrNotCompeting = errors.New("hypotheses are not in the same competition group") + +// Compete places hypotheses into a competition group. +// +// Membership is symmetric: every member lists every other. A one-way link +// would let a reader looking at one hypothesis miss that a rival exists, which +// is precisely the mistake the group is meant to prevent. +func (s *HypothesisStore) Compete(ctx context.Context, groupID string, ids []contract.HypothesisID, actor contract.Actor) (Competition, error) { + if groupID == "" { + return Competition{}, errors.New("a competition group needs an id") + } + if len(ids) < 2 { + return Competition{}, errors.New("a competition group needs at least two hypotheses") + } + + seen := map[contract.HypothesisID]struct{}{} + members := make([]contract.HypothesisID, 0, len(ids)) + for _, id := range ids { + if _, dup := seen[id]; dup { + continue + } + if _, err := s.Get(ctx, id); err != nil { + return Competition{}, fmt.Errorf("cannot add %s to %s: %w", id, groupID, err) + } + seen[id] = struct{}{} + members = append(members, id) + } + sort.Slice(members, func(i, j int) bool { return members[i] < members[j] }) + + for _, id := range members { + h, err := s.Get(ctx, id) + if err != nil { + return Competition{}, err + } + + alternatives := make([]contract.HypothesisID, 0, len(members)-1) + for _, other := range members { + if other != id { + alternatives = append(alternatives, other) + } + } + + h.Competition = &contract.FluidHypothesisCompetition{ + GroupID: groupID, + Alternatives: alternatives, + } + if err := s.put(ctx, h); err != nil { + return Competition{}, err + } + if err := s.event(ctx, h, "HYPOTHESIS_COMPETING", actor, + fmt.Sprintf("joined competition group %s against %v", groupID, alternatives), nil); err != nil { + return Competition{}, err + } + } + + return Competition{GroupID: groupID, Members: members}, nil +} + +// Group returns the members of a competition group. +func (s *HypothesisStore) Group(ctx context.Context, groupID string) (Competition, error) { + all, err := s.List(ctx, "") + if err != nil { + return Competition{}, err + } + + c := Competition{GroupID: groupID} + for _, h := range all { + if h.Competition == nil || h.Competition.GroupID != groupID { + continue + } + c.Members = append(c.Members, h.ID) + if h.State == contract.FluidHypothesisStateACCEPTED { + c.Resolved = true + c.Preferred = h.ID + } + } + sort.Slice(c.Members, func(i, j int) bool { return c.Members[i] < c.Members[j] }) + + if len(c.Members) == 0 { + return c, fmt.Errorf("no hypotheses in competition group %q", groupID) + } + return c, nil +} + +// Resolve accepts one member of a competition and supersedes the rest. +// +// The losers become SUPERSEDED rather than REJECTED. Rejection says the +// explanation was wrong; superseded says a better one won. The distinction +// matters when the winner is later refuted and someone goes looking for what +// else had been considered. +func (s *HypothesisStore) Resolve(ctx context.Context, groupID string, winner contract.HypothesisID, actor contract.Actor, reason string) (Competition, error) { + if reason == "" { + return Competition{}, errors.New("resolving a competition requires a reason") + } + + group, err := s.Group(ctx, groupID) + if err != nil { + return group, err + } + + found := false + for _, id := range group.Members { + if id == winner { + found = true + } + } + if !found { + return group, fmt.Errorf("%w: %s is not in %s", ErrNotCompeting, winner, groupID) + } + + for _, id := range group.Members { + if id == winner { + continue + } + h, err := s.Get(ctx, id) + if err != nil { + return group, err + } + if h.State == contract.FluidHypothesisStateSUPERSEDED || + h.State == contract.FluidHypothesisStateREJECTED { + continue + } + if _, err := s.Transition(ctx, id, contract.FluidHypothesisStateSUPERSEDED, actor, + fmt.Sprintf("superseded by %s in competition %s: %s", winner, groupID, reason)); err != nil { + return group, err + } + } + + group.Resolved = true + group.Preferred = winner + return group, nil +} diff --git a/internal/science/experiment.go b/internal/science/experiment.go new file mode 100644 index 0000000..f49f44b --- /dev/null +++ b/internal/science/experiment.go @@ -0,0 +1,396 @@ +package science + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +// ExperimentController exposes verified candidates under bounded conditions. +// +// ArchitectureBlueprint.md section 17 is the constraint that shapes this type: +// the controller does not process traffic. It writes deterministic routing +// policy that the revision router consumes. Keeping experimental intent out of +// the runtime decision mechanism is what lets an experiment be stopped by +// replacing a document rather than by coordinating with a live request path. +type ExperimentController struct { + store evidence.Store + hypotheses *HypothesisStore + iface contract.InterfaceID + now func() time.Time + maxParallel int +} + +// NewExperimentController returns a controller. +func NewExperimentController(store evidence.Store, h *HypothesisStore, iface contract.InterfaceID) *ExperimentController { + return &ExperimentController{ + store: store, + hypotheses: h, + iface: iface, + now: time.Now, + // Concurrency is bounded because every running experiment splits the + // traffic the others are measuring. Blueprint section 23 treats + // evolution velocity as a control variable, not a free parameter. + maxParallel: 3, + } +} + +// SetMaxParallel bounds concurrent running experiments. +func (c *ExperimentController) SetMaxParallel(n int) { c.maxParallel = n } + +var ( + // ErrTooManyExperiments reports the concurrency limit. + ErrTooManyExperiments = errors.New("too many experiments already running") + // ErrNotRunning reports an operation needing a running experiment. + ErrNotRunning = errors.New("experiment is not running") + // ErrGuardrailBreached reports an experiment stopped by its own conditions. + ErrGuardrailBreached = errors.New("experiment stopped: guardrail breached") +) + +// Design creates an experiment in PLANNED. +// +// Every field the Blueprint section 16 list requires is validated here rather +// than at start, so that a badly specified experiment is caught while it is +// still cheap to fix. +func (c *ExperimentController) Design(ctx context.Context, e contract.FluidExperiment, actor contract.Actor) (contract.FluidExperiment, error) { + if e.ID == "" { + return e, errors.New("experiment needs an id") + } + if err := contract.RequireKind(string(e.ID), contract.KindExperiment); err != nil { + return e, err + } + if len(e.HypothesisRefs) == 0 { + // An experiment without a hypothesis measures nothing in particular, + // and its result cannot confirm or refute anything. + return e, errors.New("an experiment must reference at least one hypothesis") + } + if e.ControlRevision == "" || len(e.CandidateRevisions) == 0 { + return e, errors.New("an experiment needs a control and at least one candidate") + } + if len(e.Metrics.Primary) == 0 { + return e, errors.New("an experiment needs at least one primary metric") + } + if len(e.StopConditions) == 0 { + return e, errors.New("an experiment needs a stop condition; unbounded experiments do not end") + } + if err := validateAllocation(e.Allocation); err != nil { + return e, err + } + + for _, h := range e.HypothesisRefs { + if _, err := c.hypotheses.Get(ctx, h); err != nil { + return e, fmt.Errorf("hypothesis %s: %w", h, err) + } + } + + e.SchemaVersion = "0.1" + e.InterfaceID = c.iface + e.Result = contract.FluidExperimentResult{State: contract.FluidExperimentResultStatePLANNED} + + if err := c.put(ctx, e); err != nil { + return e, err + } + for _, h := range e.HypothesisRefs { + if err := c.hypotheses.AttachExperiment(ctx, h, e.ID, actor); err != nil { + return e, err + } + } + return e, c.event(ctx, e, "EXPERIMENT_PLANNED", actor, + fmt.Sprintf("control %s against %v", e.ControlRevision, e.CandidateRevisions)) +} + +// validateAllocation checks traffic shares sum to one. +// +// A rounding tolerance is allowed, but a policy whose shares sum to 0.6 would +// leave forty percent of matching traffic with nowhere defined to go, and the +// router would silently fall through to the default — quietly contaminating the +// control arm. +func validateAllocation(allocation map[string]contract.UnitInterval) error { + if len(allocation) < 2 { + return errors.New("allocation needs at least a control and a candidate share") + } + var total float64 + for _, share := range allocation { + total += float64(share) + } + if total < 0.999 || total > 1.001 { + return fmt.Errorf("allocation shares sum to %.3f, must sum to 1", total) + } + return nil +} + +// Start moves an experiment to RUNNING and emits the routing policy for it. +// +// The returned policy is what the router will consume. It is produced here and +// installed separately, so that the act of exposing traffic is a distinct, +// auditable step rather than a side effect of a state change. +func (c *ExperimentController) Start(ctx context.Context, id contract.ExperimentID, generation int64, defaultRevision contract.RevisionID, actor contract.Actor) (contract.FluidExperiment, contract.RoutingPolicy, error) { + e, err := c.Get(ctx, id) + if err != nil { + return e, contract.RoutingPolicy{}, err + } + if e.Result.State != contract.FluidExperimentResultStatePLANNED { + return e, contract.RoutingPolicy{}, fmt.Errorf("experiment %s is %s, not PLANNED", id, e.Result.State) + } + + running, err := c.running(ctx) + if err != nil { + return e, contract.RoutingPolicy{}, err + } + if len(running) >= c.maxParallel { + return e, contract.RoutingPolicy{}, fmt.Errorf("%w: %d running, limit is %d", + ErrTooManyExperiments, len(running), c.maxParallel) + } + + now := c.now().UTC() + e.StartAt = &now + e.Result.State = contract.FluidExperimentResultStateRUNNING + + policy := c.policyFor(e, generation, defaultRevision, actor, now) + + if err := c.put(ctx, e); err != nil { + return e, policy, err + } + for _, h := range e.HypothesisRefs { + // The hypothesis follows its experiment: leaving it in DESIGNING while + // traffic is already split would misreport where the work actually is. + if cur, err := c.hypotheses.Get(ctx, h); err == nil && + CanTransition(cur.State, contract.FluidHypothesisStateEXPERIMENTING) { + _, _ = c.hypotheses.Transition(ctx, h, contract.FluidHypothesisStateEXPERIMENTING, actor, + fmt.Sprintf("experiment %s started", id)) + } + } + + return e, policy, c.event(ctx, e, "EXPERIMENT_STARTED", actor, + fmt.Sprintf("routing policy generation %d issued", generation)) +} + +// policyFor renders the routing policy that enacts an experiment. +func (c *ExperimentController) policyFor(e contract.FluidExperiment, generation int64, defaultRevision contract.RevisionID, actor contract.Actor, now time.Time) contract.RoutingPolicy { + allocation := make(map[string]contract.UnitInterval, len(e.Allocation)) + for key, share := range e.Allocation { + // Experiment records may name arms "control" and "candidate"; the + // router only understands revision ids, so they are resolved here. + switch key { + case "control": + allocation[string(e.ControlRevision)] = share + case "candidate": + if len(e.CandidateRevisions) > 0 { + allocation[string(e.CandidateRevisions[0])] = share + } + default: + allocation[key] = share + } + } + + sticky := contract.RoutingPolicyRulesItemStickyByConsumerID + expID := e.ID + rule := contract.RoutingPolicyRulesItem{ + Experiment: &expID, + Allocation: allocation, + StickyBy: &sticky, + } + if len(e.Cohorts) == 1 { + cohort := e.Cohorts[0] + rule.Cohort = &cohort + } + + rules := []contract.RoutingPolicyRulesItem{rule} + // With several eligible cohorts, one rule per cohort keeps each match + // explicit rather than relying on an implicit any-cohort wildcard. + if len(e.Cohorts) > 1 { + rules = rules[:0] + for _, cohort := range e.Cohorts { + cohort := cohort + r := rule + r.Cohort = &cohort + rules = append(rules, r) + } + } + + return contract.RoutingPolicy{ + SchemaVersion: "0.1", + ID: fmt.Sprintf("rp-%s-%d", e.ID, generation), + Interface: c.iface, + Generation: generation, + IssuedAt: &now, + IssuedBy: &actor, + DefaultRevision: defaultRevision, + Rules: rules, + } +} + +// Stop ends an experiment and returns the policy that removes its traffic. +// +// Experiments must be interruptible (Blueprint invariant 7). Stopping produces +// a policy with no experiment rules, so the router falls back to the default +// revision on the next generation — a known-good state reached by replacing a +// document rather than by unwinding anything. +func (c *ExperimentController) Stop(ctx context.Context, id contract.ExperimentID, generation int64, defaultRevision contract.RevisionID, actor contract.Actor, reason string) (contract.FluidExperiment, contract.RoutingPolicy, error) { + e, err := c.Get(ctx, id) + if err != nil { + return e, contract.RoutingPolicy{}, err + } + if e.Result.State != contract.FluidExperimentResultStateRUNNING { + return e, contract.RoutingPolicy{}, fmt.Errorf("%w: %s is %s", ErrNotRunning, id, e.Result.State) + } + if reason == "" { + return e, contract.RoutingPolicy{}, errors.New("stopping an experiment requires a reason") + } + + e.Result.State = contract.FluidExperimentResultStateSTOPPED + e.Result.StoppedReason = &reason + + now := c.now().UTC() + policy := contract.RoutingPolicy{ + SchemaVersion: "0.1", + ID: fmt.Sprintf("rp-stop-%s-%d", e.ID, generation), + Interface: c.iface, + Generation: generation, + IssuedAt: &now, + IssuedBy: &actor, + DefaultRevision: defaultRevision, + Rules: []contract.RoutingPolicyRulesItem{}, + } + + if err := c.put(ctx, e); err != nil { + return e, policy, err + } + return e, policy, c.event(ctx, e, "EXPERIMENT_STOPPED", actor, reason) +} + +// Finalize records an experiment's conclusion. +func (c *ExperimentController) Finalize(ctx context.Context, id contract.ExperimentID, preferred contract.RevisionID, actor contract.Actor, reason string, refs []contract.EvidenceRef) (contract.FluidExperiment, error) { + e, err := c.Get(ctx, id) + if err != nil { + return e, err + } + if e.Result.State == contract.FluidExperimentResultStatePLANNED { + return e, errors.New("an experiment that never ran has nothing to finalize") + } + if reason == "" { + return e, errors.New("finalizing an experiment requires a reason") + } + + e.Result.State = contract.FluidExperimentResultStateCOMPLETED + if preferred != "" { + e.Result.PreferredRevision = &preferred + } + e.Result.EvidenceRefs = refs + + if err := c.put(ctx, e); err != nil { + return e, err + } + for _, h := range e.HypothesisRefs { + if cur, err := c.hypotheses.Get(ctx, h); err == nil && + CanTransition(cur.State, contract.FluidHypothesisStateEVALUATING) { + _, _ = c.hypotheses.Transition(ctx, h, contract.FluidHypothesisStateEVALUATING, actor, + fmt.Sprintf("experiment %s completed", id)) + } + } + return e, c.event(ctx, e, "EXPERIMENT_COMPLETED", actor, reason) +} + +// Amend records a change to a running experiment's terms. +// +// Blueprint section 18: success criteria must not be changed after results are +// visible without recording the amendment. Making amendment a first-class +// operation is what makes the alternative — quietly editing the record — +// distinguishable from legitimate mid-flight adjustment. +func (c *ExperimentController) Amend(ctx context.Context, id contract.ExperimentID, change, reason string, actor contract.Actor) (contract.FluidExperiment, error) { + if change == "" || reason == "" { + return contract.FluidExperiment{}, errors.New("an amendment needs both a change and a reason") + } + + e, err := c.Get(ctx, id) + if err != nil { + return e, err + } + + e.Amendments = append(e.Amendments, contract.FluidExperimentAmendmentsItem{ + At: c.now().UTC(), + Actor: actor, + Change: change, + Reason: reason, + }) + + if err := c.put(ctx, e); err != nil { + return e, err + } + return e, c.event(ctx, e, "EXPERIMENT_AMENDED", actor, fmt.Sprintf("%s: %s", change, reason)) +} + +// Get returns one experiment. +func (c *ExperimentController) Get(ctx context.Context, id contract.ExperimentID) (contract.FluidExperiment, error) { + body, err := c.store.Record(ctx, contract.KindExperiment, string(id)) + if err != nil { + return contract.FluidExperiment{}, err + } + var doc contract.ExperimentDocument + if err := json.Unmarshal(body, &doc); err != nil { + return contract.FluidExperiment{}, fmt.Errorf("decode experiment %s: %w", id, err) + } + return doc.FluidExperiment, nil +} + +// List returns experiments, optionally filtered by state. +func (c *ExperimentController) List(ctx context.Context, state contract.FluidExperimentResultState) ([]contract.FluidExperiment, error) { + records, err := c.store.Records(ctx, contract.KindExperiment) + if err != nil { + return nil, err + } + + out := make([]contract.FluidExperiment, 0, len(records)) + for _, body := range records { + var doc contract.ExperimentDocument + if err := json.Unmarshal(body, &doc); err != nil { + continue + } + if state != "" && doc.FluidExperiment.Result.State != state { + continue + } + out = append(out, doc.FluidExperiment) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (c *ExperimentController) running(ctx context.Context) ([]contract.FluidExperiment, error) { + return c.List(ctx, contract.FluidExperimentResultStateRUNNING) +} + +func (c *ExperimentController) put(ctx context.Context, e contract.FluidExperiment) error { + body, err := json.Marshal(contract.ExperimentDocument{FluidExperiment: e}) + if err != nil { + return err + } + return c.store.PutRecord(ctx, contract.KindExperiment, string(e.ID), body) +} + +func (c *ExperimentController) event(ctx context.Context, e contract.FluidExperiment, kind string, actor contract.Actor, reason string) error { + return c.store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-%s-%d", e.ID, c.now().UnixNano())), + OccurredAt: c.now().UTC(), + EntityType: contract.FluidEventEntityTypeExperiment, + EntityID: string(e.ID), + EventType: kind, + Actor: actor, + Inputs: hypothesisInputs(e.HypothesisRefs), + Reason: reason, + }) +} + +func hypothesisInputs(refs []contract.HypothesisID) []string { + out := make([]string, len(refs)) + for i, r := range refs { + out[i] = string(r) + } + return out +} diff --git a/internal/science/experiment_test.go b/internal/science/experiment_test.go new file mode 100644 index 0000000..c366179 --- /dev/null +++ b/internal/science/experiment_test.go @@ -0,0 +1,290 @@ +package science + +import ( + "context" + "errors" + "testing" + + "github.com/tegwick/fluid-core/internal/contract" +) + +func plannedExperiment(id contract.ExperimentID) contract.FluidExperiment { + return contract.FluidExperiment{ + ID: id, + HypothesisRefs: []contract.HypothesisID{"H-1"}, + ControlRevision: "R-1", + CandidateRevisions: []contract.RevisionID{"R-2"}, + Cohorts: []contract.CohortID{"coding-agents"}, + Allocation: map[string]contract.UnitInterval{"control": 0.9, "candidate": 0.1}, + Metrics: contract.FluidExperimentMetrics{ + Primary: []string{"requests_per_completed_task"}, + Guardrails: []string{"p95_latency_ms", "error_rate"}, + }, + StopConditions: []string{"hard_guardrail_violation", "manual_stop"}, + } +} + +func withHypothesis(t *testing.T, hs *HypothesisStore) { + t.Helper() + ctx := context.Background() + if _, err := hs.Create(ctx, complete("H-1", "latest entry"), operator); err != nil { + t.Fatal(err) + } + for _, s := range []State{ + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateDESIGNING, + } { + if _, err := hs.Transition(ctx, "H-1", s, operator, "advancing"); err != nil { + t.Fatal(err) + } + } +} + +func TestDesignValidatesTheBlueprintRequirements(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + for _, tc := range []struct { + name string + mutate func(*contract.FluidExperiment) + }{ + {"no hypothesis", func(e *contract.FluidExperiment) { e.HypothesisRefs = nil }}, + {"no stop condition", func(e *contract.FluidExperiment) { e.StopConditions = nil }}, + {"no primary metric", func(e *contract.FluidExperiment) { e.Metrics.Primary = nil }}, + {"no candidate", func(e *contract.FluidExperiment) { e.CandidateRevisions = nil }}, + } { + e := plannedExperiment("E-1") + tc.mutate(&e) + if _, err := ec.Design(ctx, e, operator); err == nil { + t.Errorf("%s: accepted", tc.name) + } + } +} + +// TestAllocationMustSumToOne: shares summing to less would leave matching +// traffic falling through to the default, quietly contaminating the control. +func TestAllocationMustSumToOne(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + e := plannedExperiment("E-1") + e.Allocation = map[string]contract.UnitInterval{"control": 0.5, "candidate": 0.1} + if _, err := ec.Design(ctx, e, operator); err == nil { + t.Error("an allocation summing to 0.6 was accepted") + } + + e.Allocation = map[string]contract.UnitInterval{"control": 0.9, "candidate": 0.1} + if _, err := ec.Design(ctx, e, operator); err != nil { + t.Errorf("a valid allocation was refused: %v", err) + } +} + +// TestStartProducesRoutingPolicyRatherThanTouchingTraffic is the section 17 +// separation: the controller writes policy the router consumes. +func TestStartProducesRoutingPolicy(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + + e, policy, err := ec.Start(ctx, "E-1", 7, "R-1", operator) + if err != nil { + t.Fatal(err) + } + if e.Result.State != contract.FluidExperimentResultStateRUNNING { + t.Errorf("state = %s", e.Result.State) + } + if policy.Generation != 7 || policy.DefaultRevision != "R-1" { + t.Errorf("policy = %+v", policy) + } + if len(policy.Rules) != 1 { + t.Fatalf("policy has %d rules, want 1", len(policy.Rules)) + } + + // Arm names must be resolved to revision ids: the router only understands + // those, and a rule naming "candidate" would match nothing. + rule := policy.Rules[0] + if _, ok := rule.Allocation["R-1"]; !ok { + t.Errorf("control arm not resolved to a revision id: %+v", rule.Allocation) + } + if got := rule.Allocation["R-2"]; got != 0.1 { + t.Errorf("candidate share = %v, want 0.1", got) + } + if rule.Experiment == nil || *rule.Experiment != "E-1" { + t.Error("rule does not name its experiment") + } + + // The hypothesis follows its experiment. + h, err := hs.Get(ctx, "H-1") + if err != nil { + t.Fatal(err) + } + if h.State != contract.FluidHypothesisStateEXPERIMENTING { + t.Errorf("hypothesis state = %s, want EXPERIMENTING", h.State) + } +} + +// TestStopReturnsToAKnownGoodState: experiments must be interruptible, and +// stopping should be a document replacement, not an unwind. +func TestStopReturnsToAKnownGoodState(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-1", 1, "R-1", operator); err != nil { + t.Fatal(err) + } + + e, policy, err := ec.Stop(ctx, "E-1", 2, "R-1", operator, "p95 latency breached its guardrail") + if err != nil { + t.Fatal(err) + } + if e.Result.State != contract.FluidExperimentResultStateSTOPPED { + t.Errorf("state = %s", e.Result.State) + } + if len(policy.Rules) != 0 { + t.Errorf("the stop policy still carries %d rules", len(policy.Rules)) + } + if policy.DefaultRevision != "R-1" || policy.Generation != 2 { + t.Errorf("stop policy does not return traffic to the default: %+v", policy) + } + + if _, _, err := ec.Stop(ctx, "E-1", 3, "R-1", operator, "again"); !errors.Is(err, ErrNotRunning) { + t.Errorf("stopping a stopped experiment returned %v", err) + } +} + +func TestStopRequiresAReason(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-1", 1, "R-1", operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Stop(ctx, "E-1", 2, "R-1", operator, ""); err == nil { + t.Error("an experiment was stopped with no reason recorded") + } +} + +// TestConcurrencyIsBounded: every running experiment splits the traffic the +// others are measuring. +func TestConcurrencyIsBounded(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + ec.SetMaxParallel(2) + + for _, id := range []contract.ExperimentID{"E-1", "E-2", "E-3"} { + if _, err := ec.Design(ctx, plannedExperiment(id), operator); err != nil { + t.Fatal(err) + } + } + + if _, _, err := ec.Start(ctx, "E-1", 1, "R-1", operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-2", 2, "R-1", operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-3", 3, "R-1", operator); !errors.Is(err, ErrTooManyExperiments) { + t.Errorf("a third experiment started past the limit: %v", err) + } + + // Stopping one frees a slot. + if _, _, err := ec.Stop(ctx, "E-1", 4, "R-1", operator, "done"); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-3", 5, "R-1", operator); err != nil { + t.Errorf("a slot did not free after stopping: %v", err) + } +} + +// TestAmendmentIsRecorded makes changing the terms distinguishable from +// quietly editing the record. +func TestAmendmentIsRecorded(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + + if _, err := ec.Amend(ctx, "E-1", "", "no change given", operator); err == nil { + t.Error("an empty amendment was accepted") + } + + e, err := ec.Amend(ctx, "E-1", "extended the window by 14 days", + "weekly publishing cadence gives too few samples in 7 days", operator) + if err != nil { + t.Fatal(err) + } + if len(e.Amendments) != 1 { + t.Fatalf("amendments = %d, want 1", len(e.Amendments)) + } + if e.Amendments[0].Actor.ID != operator.ID || e.Amendments[0].Reason == "" { + t.Errorf("amendment does not record who and why: %+v", e.Amendments[0]) + } +} + +func TestFinalizeMovesHypothesisToEvaluating(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Start(ctx, "E-1", 1, "R-1", operator); err != nil { + t.Fatal(err) + } + if _, _, err := ec.Stop(ctx, "E-1", 2, "R-1", operator, "window elapsed"); err != nil { + t.Fatal(err) + } + + e, err := ec.Finalize(ctx, "E-1", "R-2", operator, "candidate met its target", + []contract.EvidenceRef{"metrics:E-1/window-1"}) + if err != nil { + t.Fatal(err) + } + if e.Result.State != contract.FluidExperimentResultStateCOMPLETED { + t.Errorf("state = %s", e.Result.State) + } + if e.Result.PreferredRevision == nil || *e.Result.PreferredRevision != "R-2" { + t.Error("preferred revision not recorded") + } + + h, err := hs.Get(ctx, "H-1") + if err != nil { + t.Fatal(err) + } + if h.State != contract.FluidHypothesisStateEVALUATING { + t.Errorf("hypothesis state = %s, want EVALUATING", h.State) + } +} + +func TestFinalizeRefusesAnExperimentThatNeverRan(t *testing.T) { + ctx := context.Background() + hs, ec, _ := newStores(t) + withHypothesis(t, hs) + + if _, err := ec.Design(ctx, plannedExperiment("E-1"), operator); err != nil { + t.Fatal(err) + } + if _, err := ec.Finalize(ctx, "E-1", "R-2", operator, "it would have worked", nil); err == nil { + t.Error("an experiment that never ran was finalized") + } +} diff --git a/internal/science/hypothesis.go b/internal/science/hypothesis.go new file mode 100644 index 0000000..3477e93 --- /dev/null +++ b/internal/science/hypothesis.go @@ -0,0 +1,402 @@ +// Package science implements the FLUID scientific adaptation loop: hypotheses, +// competition between them, bounded experiments, and the promotion decisions +// they inform. +// +// Nothing here is generative. ArchitectureBlueprint.md Phase C is deliberately +// human-driven: the point of this stage, per section 50, is to prove that the +// revision-experiment-fitness loop works cleanly and safely, not to automate +// the reasoning inside it. +// +// The schema document's section 18 sets the discipline this package keeps: +// what we observed, what we think explains it, what we changed, and what +// happened afterwards must not collapse into one narrative. Keeping them apart +// is what makes criticism, competing explanations and rollback possible. +package science + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +// HypothesisStore persists hypotheses and enforces their lifecycle. +type HypothesisStore struct { + store evidence.Store + iface contract.InterfaceID + now func() time.Time +} + +// NewHypothesisStore returns a store backed by the evidence log. +func NewHypothesisStore(store evidence.Store, iface contract.InterfaceID) *HypothesisStore { + return &HypothesisStore{store: store, iface: iface, now: time.Now} +} + +// State is a point in the hypothesis lifecycle. +type State = contract.FluidHypothesisState + +// transitions is the permitted lifecycle graph. +// +// It is explicit rather than free-form because the states carry meaning a +// reader relies on. A hypothesis that jumped from DRAFT to ACCEPTED would claim +// evidence it never gathered, and the audit trail would show nothing wrong. +var transitions = map[State][]State{ + contract.FluidHypothesisStateDRAFT: { + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateDEFERRED, + // A draft can lose a competition to a better-formed rival before it is + // ever completed. That is superseding, not rejection: nobody showed the + // idea was wrong, only that something else explained the pressure first. + contract.FluidHypothesisStateSUPERSEDED, + }, + contract.FluidHypothesisStateREADY: { + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateDEFERRED, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateSUPERSEDED, + }, + contract.FluidHypothesisStatePRIORITIZED: { + contract.FluidHypothesisStateDESIGNING, + contract.FluidHypothesisStateDEFERRED, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateSUPERSEDED, + }, + contract.FluidHypothesisStateDESIGNING: { + contract.FluidHypothesisStateEXPERIMENTING, + contract.FluidHypothesisStateDEFERRED, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateSUPERSEDED, + }, + contract.FluidHypothesisStateEXPERIMENTING: { + contract.FluidHypothesisStateEVALUATING, + // An experiment can be stopped without a verdict; that is a normal + // outcome, not a failure of the hypothesis. + contract.FluidHypothesisStateDEFERRED, + contract.FluidHypothesisStateREJECTED, + }, + contract.FluidHypothesisStateEVALUATING: { + contract.FluidHypothesisStateACCEPTED, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateDEFERRED, + contract.FluidHypothesisStateSUPERSEDED, + }, + // A deferred hypothesis can come back when circumstances change. + contract.FluidHypothesisStateDEFERRED: { + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateREJECTED, + contract.FluidHypothesisStateSUPERSEDED, + }, + // Terminal. A rejected hypothesis that turns out to be right becomes a new + // hypothesis citing the old one, so the reversal is visible. + contract.FluidHypothesisStateACCEPTED: {contract.FluidHypothesisStateSUPERSEDED}, + contract.FluidHypothesisStateREJECTED: {}, + contract.FluidHypothesisStateSUPERSEDED: {}, +} + +var ( + // ErrInvalidTransition reports a lifecycle move that is not permitted. + ErrInvalidTransition = errors.New("invalid hypothesis state transition") + // ErrIncomplete reports a hypothesis missing something its state requires. + ErrIncomplete = errors.New("hypothesis is missing required content") + // ErrNotFound reports an unknown hypothesis. + ErrNotFound = evidence.ErrNotFound +) + +// CanTransition reports whether a lifecycle move is permitted. +func CanTransition(from, to State) bool { + if from == to { + return true + } + for _, allowed := range transitions[from] { + if allowed == to { + return true + } + } + return false +} + +// Create records a new hypothesis in DRAFT. +// +// A draft is allowed to be incomplete: the point of the state is to hold a +// half-formed idea. Completeness is checked when it moves to READY, which is +// the claim that it is worth someone's time. +func (s *HypothesisStore) Create(ctx context.Context, h contract.FluidHypothesis, actor contract.Actor) (contract.FluidHypothesis, error) { + if h.ID == "" { + return h, errors.New("hypothesis needs an id") + } + if err := contract.RequireKind(string(h.ID), contract.KindHypothesis); err != nil { + return h, err + } + if h.Title == "" { + return h, fmt.Errorf("%w: a hypothesis needs a title", ErrIncomplete) + } + + if _, err := s.Get(ctx, h.ID); err == nil { + return h, fmt.Errorf("hypothesis %s already exists", h.ID) + } else if !errors.Is(err, ErrNotFound) { + return h, err + } + + h.SchemaVersion = "0.1" + h.InterfaceID = s.iface + if h.State == "" { + h.State = contract.FluidHypothesisStateDRAFT + } + if h.CreatedAt == nil { + now := s.now().UTC() + h.CreatedAt = &now + } + if h.CreatedBy == nil { + h.CreatedBy = &actor + } + + if err := s.put(ctx, h); err != nil { + return h, err + } + return h, s.event(ctx, h, "HYPOTHESIS_CREATED", actor, h.Title, nil) +} + +// readyRequirements lists what a hypothesis must have before it leaves DRAFT. +// +// These mirror the schema document's section 4 mandatory fields. The check runs +// at READY rather than at creation so that drafting stays cheap, but nothing +// can be prioritized, experimented on or accepted without them. +func readyRequirements(h contract.FluidHypothesis) []string { + var missing []string + + if h.Observation.Summary == "" { + missing = append(missing, "observation.summary: what was actually seen") + } + if len(h.Observation.EvidenceRefs) == 0 { + missing = append(missing, "observation.evidence_refs: a claim with no evidence is not falsifiable") + } + if len(h.Pressure.Classes) == 0 { + missing = append(missing, "pressure.classes: which kind of pressure this addresses") + } + if h.Explanation.Claim == "" { + missing = append(missing, "explanation.claim: what you think explains the observation") + } + if h.ProposedAdaptation.Summary == "" { + missing = append(missing, "proposed_adaptation.summary: what you propose to change") + } + if len(h.ExpectedOutcomes) == 0 { + missing = append(missing, "expected_outcomes: a prediction, or the hypothesis cannot be wrong") + } + if h.SuccessCriteria.Expression == "" { + missing = append(missing, "success_criteria.expression: how you will know") + } + // Complexity and risk are required structs rather than pointers, so an + // unset one shows up as a zero value: an empty delta and an empty level. + if h.Complexity.ExpectedDelta == (contract.ComplexityDelta{}) { + missing = append(missing, "complexity.expected_delta: complexity is a budget") + } + if h.Risk.Level == "" { + missing = append(missing, "risk.level") + } + return missing +} + +// Transition moves a hypothesis to a new state. +func (s *HypothesisStore) Transition(ctx context.Context, id contract.HypothesisID, to State, actor contract.Actor, reason string) (contract.FluidHypothesis, error) { + h, err := s.Get(ctx, id) + if err != nil { + return h, err + } + + if !to.Valid() { + return h, fmt.Errorf("unknown hypothesis state %q", to) + } + if !CanTransition(h.State, to) { + return h, fmt.Errorf("%w: %s cannot become %s", ErrInvalidTransition, h.State, to) + } + if reason == "" { + return h, errors.New("a lifecycle transition requires a reason") + } + + // Everything past DRAFT is a claim that the hypothesis is worth acting on. + if to != contract.FluidHypothesisStateDRAFT && to != contract.FluidHypothesisStateREJECTED && + to != contract.FluidHypothesisStateDEFERRED && to != contract.FluidHypothesisStateSUPERSEDED { + if missing := readyRequirements(h); len(missing) > 0 { + return h, fmt.Errorf("%w: %s cannot reach %s until it has:\n - %s", + ErrIncomplete, id, to, joinLines(missing)) + } + } + + previous := h.State + h.State = to + + if err := s.put(ctx, h); err != nil { + return h, err + } + return h, s.event(ctx, h, "HYPOTHESIS_"+string(to), actor, + fmt.Sprintf("%s -> %s: %s", previous, to, reason), nil) +} + +// RecordOutcome closes a hypothesis with its result. +// +// The outcome is written separately from the explanation it tests, and only in +// EVALUATING. A hypothesis that recorded its own result while still running +// would be assuming what it set out to find. +func (s *HypothesisStore) RecordOutcome(ctx context.Context, id contract.HypothesisID, status contract.FluidHypothesisOutcomeStatus, summary string, evidenceRefs []contract.EvidenceRef, actor contract.Actor) (contract.FluidHypothesis, error) { + h, err := s.Get(ctx, id) + if err != nil { + return h, err + } + if h.State != contract.FluidHypothesisStateEVALUATING { + return h, fmt.Errorf("an outcome can only be recorded while EVALUATING, not in %s", h.State) + } + if summary == "" { + return h, errors.New("an outcome needs a summary") + } + + h.Outcome = &contract.FluidHypothesisOutcome{ + Status: &status, + Summary: &summary, + EvidenceRefs: evidenceRefs, + } + + next := contract.FluidHypothesisStateREJECTED + if status == contract.FluidHypothesisOutcomeStatusCONFIRMED { + next = contract.FluidHypothesisStateACCEPTED + } + if status == contract.FluidHypothesisOutcomeStatusINCONCLUSIVE { + // Inconclusive is not refuted. Deferring keeps it available for a + // better-powered experiment instead of burying a possibly-good idea. + next = contract.FluidHypothesisStateDEFERRED + } + h.State = next + + if err := s.put(ctx, h); err != nil { + return h, err + } + return h, s.event(ctx, h, "HYPOTHESIS_OUTCOME_RECORDED", actor, + fmt.Sprintf("%s: %s", status, summary), evidenceRefs) +} + +// Get returns one hypothesis. +func (s *HypothesisStore) Get(ctx context.Context, id contract.HypothesisID) (contract.FluidHypothesis, error) { + body, err := s.store.Record(ctx, contract.KindHypothesis, string(id)) + if err != nil { + return contract.FluidHypothesis{}, err + } + var doc contract.HypothesisDocument + if err := json.Unmarshal(body, &doc); err != nil { + return contract.FluidHypothesis{}, fmt.Errorf("decode hypothesis %s: %w", id, err) + } + return doc.FluidHypothesis, nil +} + +// List returns hypotheses, optionally filtered by state. +func (s *HypothesisStore) List(ctx context.Context, state State) ([]contract.FluidHypothesis, error) { + records, err := s.store.Records(ctx, contract.KindHypothesis) + if err != nil { + return nil, err + } + + out := make([]contract.FluidHypothesis, 0, len(records)) + for _, body := range records { + var doc contract.HypothesisDocument + if err := json.Unmarshal(body, &doc); err != nil { + continue + } + if state != "" && doc.FluidHypothesis.State != state { + continue + } + out = append(out, doc.FluidHypothesis) + } + + // Highest priority first; the list doubles as a queue. + sort.Slice(out, func(i, j int) bool { + pi, pj := priorityOf(out[i]), priorityOf(out[j]) + if pi != pj { + return pi > pj + } + return out[i].ID < out[j].ID + }) + return out, nil +} + +func priorityOf(h contract.FluidHypothesis) float64 { + if h.Priority == nil || h.Priority.Score == nil { + return 0 + } + return *h.Priority.Score +} + +// AttachRevision links a candidate revision to the hypothesis it came from. +func (s *HypothesisStore) AttachRevision(ctx context.Context, id contract.HypothesisID, rev contract.RevisionID, actor contract.Actor) error { + if err := contract.RequireKind(string(rev), contract.KindRevision); err != nil { + return err + } + h, err := s.Get(ctx, id) + if err != nil { + return err + } + for _, existing := range h.CandidateRevisionRefs { + if existing == rev { + return nil + } + } + h.CandidateRevisionRefs = append(h.CandidateRevisionRefs, rev) + + if err := s.put(ctx, h); err != nil { + return err + } + return s.event(ctx, h, "HYPOTHESIS_REVISION_ATTACHED", actor, fmt.Sprintf("attached %s", rev), nil) +} + +// AttachExperiment links an experiment to the hypothesis it tests. +func (s *HypothesisStore) AttachExperiment(ctx context.Context, id contract.HypothesisID, exp contract.ExperimentID, actor contract.Actor) error { + h, err := s.Get(ctx, id) + if err != nil { + return err + } + for _, existing := range h.ExperimentRefs { + if existing == exp { + return nil + } + } + h.ExperimentRefs = append(h.ExperimentRefs, exp) + + if err := s.put(ctx, h); err != nil { + return err + } + return s.event(ctx, h, "HYPOTHESIS_EXPERIMENT_ATTACHED", actor, fmt.Sprintf("attached %s", exp), nil) +} + +func (s *HypothesisStore) put(ctx context.Context, h contract.FluidHypothesis) error { + body, err := json.Marshal(contract.HypothesisDocument{FluidHypothesis: h}) + if err != nil { + return err + } + return s.store.PutRecord(ctx, contract.KindHypothesis, string(h.ID), body) +} + +func (s *HypothesisStore) event(ctx context.Context, h contract.FluidHypothesis, kind string, actor contract.Actor, reason string, refs []contract.EvidenceRef) error { + return s.store.AppendEvent(ctx, contract.FluidEvent{ + SchemaVersion: "0.1", + ID: contract.EventID(fmt.Sprintf("EV-%s-%d", h.ID, s.now().UnixNano())), + OccurredAt: s.now().UTC(), + EntityType: contract.FluidEventEntityTypeHypothesis, + EntityID: string(h.ID), + EventType: kind, + Actor: actor, + Reason: reason, + EvidenceRefs: refs, + }) +} + +func joinLines(items []string) string { + out := items[0] + for _, item := range items[1:] { + out += "\n - " + item + } + return out +} diff --git a/internal/science/science_test.go b/internal/science/science_test.go new file mode 100644 index 0000000..a88cb2e --- /dev/null +++ b/internal/science/science_test.go @@ -0,0 +1,272 @@ +package science + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + + _ "modernc.org/sqlite" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" +) + +const iface contract.InterfaceID = "hall-publishing" + +var operator = contract.Actor{Type: contract.ActorTypeHuman, ID: "worsch"} + +func ptr[T any](v T) *T { return &v } + +func newStores(t *testing.T) (*HypothesisStore, *ExperimentController, *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() }) + + hs := NewHypothesisStore(store, iface) + return hs, NewExperimentController(store, hs, iface), store +} + +// complete returns a hypothesis with everything READY requires. +func complete(id contract.HypothesisID, title string) contract.FluidHypothesis { + return contract.FluidHypothesis{ + ID: id, + Title: title, + Observation: contract.FluidHypothesisObservation{ + Summary: "consumers list the collection repeatedly to find one entry", + EvidenceRefs: []contract.EvidenceRef{"topology:GET /entries"}, + }, + Pressure: contract.FluidHypothesisPressure{ + Classes: []contract.PressureClass{contract.PressureClassSuccessfulButInefficientUsage}, + }, + Explanation: contract.FluidHypothesisExplanation{ + Claim: "latest entry is a first-class consumer concept the interface does not expose", + }, + ProposedAdaptation: contract.FluidHypothesisProposedAdaptation{ + Class: contract.AdaptationClassContract, + Summary: "add an explicit latest-entry capability", + }, + ExpectedOutcomes: []contract.ExpectedOutcome{ + {Metric: "requests_per_completed_task", Target: 1.2, Direction: contract.ExpectedOutcomeDirectionLower}, + }, + SuccessCriteria: contract.FluidHypothesisSuccessCriteria{ + Expression: "requests_per_completed_task <= 1.2 AND no guardrail violation", + }, + Complexity: contract.FluidHypothesisComplexity{ + ExpectedDelta: contract.ComplexityDelta{OperationCount: ptr(1.0)}, + }, + Risk: contract.FluidHypothesisRisk{Level: contract.FluidHypothesisRiskLevelLOW}, + } +} + +func TestDraftMayBeIncompleteButReadyMayNot(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + // Drafting stays cheap: a half-formed idea is allowed to be half-formed. + sparse := contract.FluidHypothesis{ID: "H-1", Title: "maybe the list endpoint is wrong"} + if _, err := hs.Create(ctx, sparse, operator); err != nil { + t.Fatalf("an incomplete draft was refused: %v", err) + } + + _, err := hs.Transition(ctx, "H-1", contract.FluidHypothesisStateREADY, operator, "let us look at this") + if !errors.Is(err, ErrIncomplete) { + t.Fatalf("an incomplete hypothesis reached READY: %v", err) + } + // The error must say what is missing, or it is not actionable. + for _, want := range []string{"evidence_refs", "expected_outcomes", "success_criteria"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error does not name missing %s: %v", want, err) + } + } +} + +// TestCannotSkipToAccepted: a hypothesis jumping straight to ACCEPTED would +// claim evidence it never gathered. +func TestCannotSkipToAccepted(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + if _, err := hs.Create(ctx, complete("H-1", "latest entry"), operator); err != nil { + t.Fatal(err) + } + + _, err := hs.Transition(ctx, "H-1", contract.FluidHypothesisStateACCEPTED, operator, "looks right to me") + if !errors.Is(err, ErrInvalidTransition) { + t.Errorf("DRAFT jumped to ACCEPTED: %v", err) + } +} + +func TestLifecycleHappyPath(t *testing.T) { + ctx := context.Background() + hs, _, store := newStores(t) + + if _, err := hs.Create(ctx, complete("H-1", "latest entry"), operator); err != nil { + t.Fatal(err) + } + for _, to := range []State{ + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateDESIGNING, + contract.FluidHypothesisStateEXPERIMENTING, + contract.FluidHypothesisStateEVALUATING, + } { + if _, err := hs.Transition(ctx, "H-1", to, operator, "advancing"); err != nil { + t.Fatalf("transition to %s failed: %v", to, err) + } + } + + got, err := hs.RecordOutcome(ctx, "H-1", contract.FluidHypothesisOutcomeStatusCONFIRMED, + "requests per task fell from 2.7 to 1.15", []contract.EvidenceRef{"metrics:E-1"}, operator) + if err != nil { + t.Fatal(err) + } + if got.State != contract.FluidHypothesisStateACCEPTED { + t.Errorf("state = %s, want ACCEPTED", got.State) + } + + events, err := store.Events(ctx, evidence.EventFilter{EntityID: "H-1"}) + if err != nil { + t.Fatal(err) + } + if len(events) < 7 { + t.Errorf("lifecycle left %d events; every transition should be observable", len(events)) + } +} + +// TestOutcomeOnlyWhileEvaluating: recording a result while still running would +// assume what the experiment set out to find. +func TestOutcomeOnlyWhileEvaluating(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + if _, err := hs.Create(ctx, complete("H-1", "latest entry"), operator); err != nil { + t.Fatal(err) + } + if _, err := hs.RecordOutcome(ctx, "H-1", contract.FluidHypothesisOutcomeStatusCONFIRMED, + "it worked", nil, operator); err == nil { + t.Error("an outcome was recorded on a DRAFT hypothesis") + } +} + +// TestInconclusiveDefersRatherThanRejects: inconclusive is not refuted, and +// burying a possibly-good idea is the wrong default. +func TestInconclusiveDefersRatherThanRejects(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + if _, err := hs.Create(ctx, complete("H-1", "latest entry"), operator); err != nil { + t.Fatal(err) + } + advance(t, hs, "H-1", contract.FluidHypothesisStateEVALUATING) + + got, err := hs.RecordOutcome(ctx, "H-1", contract.FluidHypothesisOutcomeStatusINCONCLUSIVE, + "only 12 samples per arm", nil, operator) + if err != nil { + t.Fatal(err) + } + if got.State != contract.FluidHypothesisStateDEFERRED { + t.Errorf("state = %s, want DEFERRED", got.State) + } +} + +func TestCompetitionIsSymmetric(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + for _, id := range []contract.HypothesisID{"H-1", "H-2", "H-3"} { + if _, err := hs.Create(ctx, complete(id, string(id)), operator); err != nil { + t.Fatal(err) + } + } + + group, err := hs.Compete(ctx, "CG-1", []contract.HypothesisID{"H-1", "H-2", "H-3"}, operator) + if err != nil { + t.Fatal(err) + } + if len(group.Members) != 3 { + t.Fatalf("group has %d members", len(group.Members)) + } + + // Every member must know about every rival, or a reader looking at one + // would miss that alternatives exist. + for _, id := range group.Members { + h, err := hs.Get(ctx, id) + if err != nil { + t.Fatal(err) + } + if h.Competition == nil || len(h.Competition.Alternatives) != 2 { + t.Errorf("%s does not list both rivals: %+v", id, h.Competition) + } + } + + if _, err := hs.Compete(ctx, "CG-2", []contract.HypothesisID{"H-1"}, operator); err == nil { + t.Error("a competition group of one was accepted") + } +} + +// TestResolveSupersedesRatherThanRejects: rejection says the explanation was +// wrong; superseded says a better one won. +func TestResolveSupersedesRatherThanRejects(t *testing.T) { + ctx := context.Background() + hs, _, _ := newStores(t) + + for _, id := range []contract.HypothesisID{"H-1", "H-2"} { + if _, err := hs.Create(ctx, complete(id, string(id)), operator); err != nil { + t.Fatal(err) + } + } + if _, err := hs.Compete(ctx, "CG-1", []contract.HypothesisID{"H-1", "H-2"}, operator); err != nil { + t.Fatal(err) + } + + advance(t, hs, "H-1", contract.FluidHypothesisStateEVALUATING) + if _, err := hs.RecordOutcome(ctx, "H-1", contract.FluidHypothesisOutcomeStatusCONFIRMED, + "teaser plus link won on read-through", nil, operator); err != nil { + t.Fatal(err) + } + + group, err := hs.Resolve(ctx, "CG-1", "H-1", operator, "H-1 measured better on the primary metric") + if err != nil { + t.Fatal(err) + } + if !group.Resolved || group.Preferred != "H-1" { + t.Errorf("group not resolved to H-1: %+v", group) + } + + loser, err := hs.Get(ctx, "H-2") + if err != nil { + t.Fatal(err) + } + if loser.State != contract.FluidHypothesisStateSUPERSEDED { + t.Errorf("loser state = %s, want SUPERSEDED", loser.State) + } + + if _, err := hs.Resolve(ctx, "CG-1", "H-99", operator, "reason"); !errors.Is(err, ErrNotCompeting) { + t.Errorf("a non-member won a competition: %v", err) + } +} + +func advance(t *testing.T, hs *HypothesisStore, id contract.HypothesisID, to State) { + t.Helper() + ctx := context.Background() + path := []State{ + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateDESIGNING, + contract.FluidHypothesisStateEXPERIMENTING, + contract.FluidHypothesisStateEVALUATING, + } + for _, s := range path { + if _, err := hs.Transition(ctx, id, s, operator, "advancing"); err != nil { + t.Fatalf("advance to %s: %v", s, err) + } + if s == to { + return + } + } +} diff --git a/workplans/FLUID-WP-0006-fluid-science.md b/workplans/FLUID-WP-0006-fluid-science.md index 6848802..5a1ca14 100644 --- a/workplans/FLUID-WP-0006-fluid-science.md +++ b/workplans/FLUID-WP-0006-fluid-science.md @@ -4,7 +4,7 @@ type: workplan title: "FLUID Science - the adaptation loop closes (Blueprint Phase C)" domain: infotech repo: fluid-core -status: active +status: done owner: worsch topic_slug: fluid-core created: "2026-09-04" @@ -26,7 +26,7 @@ proving this loop works cleanly and safely, not autonomous coding. ```task id: FLUID-WP-0006-T01 -status: todo +status: done priority: high state_hub_task_id: "93dd0a6d-3f46-5da5-857d-e0bd49c356db" ``` @@ -40,7 +40,7 @@ collapsing them into one narrative destroys criticism and auditability. ```task id: FLUID-WP-0006-T02 -status: todo +status: done priority: medium state_hub_task_id: "81f6a22f-929e-5c94-bc61-f4e06ad7de16" ``` @@ -52,7 +52,7 @@ uncertainty is a feature. ```task id: FLUID-WP-0006-T03 -status: todo +status: done priority: medium state_hub_task_id: "0050d3e4-cc5c-5e1d-9d29-d14bace0e1cf" ``` @@ -63,7 +63,7 @@ Blueprint §44.3: create, compare, prioritize, attach candidate, record outcome. ```task id: FLUID-WP-0006-T04 -status: todo +status: done priority: high state_hub_task_id: "36391a67-6a3e-54c9-afc5-181e0b968f85" ``` @@ -77,7 +77,7 @@ Experiments are interruptible. ```task id: FLUID-WP-0006-T05 -status: todo +status: done priority: medium state_hub_task_id: "2c2a1ff5-f908-565f-ae5b-c0eb469e6482" ``` @@ -89,7 +89,7 @@ finalize. ```task id: FLUID-WP-0006-T06 -status: todo +status: done priority: high state_hub_task_id: "7223f795-a005-5fe0-9abb-45e4803a3091" ``` @@ -101,7 +101,7 @@ after results are visible without recording the amendment (§18). ```task id: FLUID-WP-0006-T07 -status: todo +status: done priority: high state_hub_task_id: "59188ee2-81e6-5331-a0f6-673b7daf871e" ``` @@ -113,7 +113,7 @@ recorded as decisions with an authorizing actor (§19). Human authority only. ```task id: FLUID-WP-0006-T08 -status: todo +status: done priority: high state_hub_task_id: "8eb2e90c-6c65-5ef5-be19-d42c2b091d8c" ``` @@ -125,7 +125,7 @@ itself a system behavior that must remain reconstructable. ```task id: FLUID-WP-0006-T09 -status: todo +status: done priority: high state_hub_task_id: "7dd49f06-9cc7-5791-abfb-33b27649ba8a" ```