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