diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cf956f5..be00303 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -40,3 +40,8 @@ jobs: - name: Test run: make test + + # The conformance suite is run separately and without the cache: a + # conformance claim is only worth what the last full run proved. + - name: Conformance + run: make conformance diff --git a/Makefile b/Makefile index 5d445bb..ebca366 100644 --- a/Makefile +++ b/Makefile @@ -41,8 +41,8 @@ check-generated: ## Fail if generated code is stale relative to schemas/ fi .PHONY: conformance -conformance: ## Full conformance suite (grows through FLUID-WP-0007) - $(GO) test ./conformance/... $(PKGS) +conformance: ## Assert every conformance requirement and architectural invariant + $(GO) test -count=1 ./conformance/... .PHONY: clean clean: diff --git a/conformance/suite/conformance_test.go b/conformance/suite/conformance_test.go new file mode 100644 index 0000000..f3da702 --- /dev/null +++ b/conformance/suite/conformance_test.go @@ -0,0 +1,448 @@ +package suite + +import ( + "context" + "crypto/ed25519" + "net/http" + "strings" + "testing" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/runtime" + "github.com/tegwick/fluid-core/internal/signing" +) + +// --------------------------------------------------------------------------- +// FluidAPIStandards.md section 36 — minimal v0.1 conformance +// +// An implementation claiming core conformance MUST provide all seven of these. +// They are asserted here rather than claimed in a README, because a conformance +// claim nobody re-checks is a conformance claim that quietly stops being true. +// --------------------------------------------------------------------------- + +// 1. A deterministic API contract. +func TestConformance01_DeterministicContract(t *testing.T) { + h := New(t) + + // The contract is enforced, not merely declared: an undeclared path is + // refused rather than proxied through. + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, "declared path") + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/undeclared"), http.StatusBadRequest, "undeclared path") + mustStatus(t, h.Call("c-1", http.MethodDelete, "/v1/entries"), http.StatusBadRequest, "undeclared method") + + // Determinism: the same request must produce the same outcome every time. + first := h.Call("c-1", http.MethodGet, "/v1/entries") + for i := 0; i < 20; i++ { + again := h.Call("c-1", http.MethodGet, "/v1/entries") + if again.Code != first.Code || again.Body.String() != first.Body.String() { + t.Fatal("identical requests produced different responses") + } + } +} + +// 2. An identifiable active interface revision. +func TestConformance02_IdentifiableRevision(t *testing.T) { + h := New(t) + + rec := h.Call("c-1", http.MethodGet, "/v1/entries") + if got := rec.Header().Get("X-FLUID-Revision"); got != "R-1" { + t.Errorf("response does not identify its revision: %q", got) + } + + // The revision that served a request must also be identifiable afterwards, + // from evidence rather than from the response the consumer happened to keep. + h.Drain() + for _, ev := range h.Telemetry() { + if ev.Revision == nil || *ev.Revision == "" { + t.Fatal("a telemetry event does not name the revision that served it") + } + } +} + +// 3. Interface telemetry. +func TestConformance03_Telemetry(t *testing.T) { + h := New(t) + + for i := 0; i < 5; i++ { + h.Call("c-1", http.MethodGet, "/v1/entries") + } + h.Call("c-1", http.MethodGet, "/v1/undeclared") + h.Drain() + + rows := h.Telemetry() + if len(rows) != 6 { + t.Fatalf("recorded %d telemetry events, want 6", len(rows)) + } + + var errors int + for _, ev := range rows { + if ev.Kind == contract.FluidTelemetryKindError { + errors++ + } + if ev.Resolution == nil { + t.Error("an event does not record why its revision was chosen") + } + } + if errors != 1 { + t.Errorf("recorded %d error events, want 1", errors) + } +} + +// 4. Declared interface evolutionary intent. +func TestConformance04_DeclaredIntent(t *testing.T) { + h := New(t) + ctx := context.Background() + + active, err := h.Intents.Active(ctx) + if err != nil { + t.Fatalf("no active intent: %v", err) + } + if !active.Mode.Valid() { + t.Errorf("active intent declares no valid authority mode") + } + + // Every revision must be bound to the intent version governing it. + for _, rev := range []contract.RevisionID{"R-1", "R-2"} { + governing, err := h.Intents.GoverningIntent(ctx, rev) + if err != nil { + t.Errorf("%s has no governing intent: %v", rev, err) + continue + } + if governing.Digest == "" { + t.Errorf("%s is bound to an intent with no digest", rev) + } + } +} + +// 5. An auditable link from evidence to proposed change. +func TestConformance05_EvidenceToChangeIsAuditable(t *testing.T) { + h := New(t) + ctx := context.Background() + + loop := RunFullLoop(t, h) + + // Pressure -> hypothesis -> revision -> experiment -> decision must be + // traversable without guessing. + pressure, err := h.Store.Record(ctx, contract.KindPressure, string(loop.Pressure)) + if err != nil { + t.Fatalf("pressure not retrievable: %v", err) + } + if len(pressure) == 0 { + t.Fatal("pressure record is empty") + } + + hyp, err := h.Hypotheses.Get(ctx, loop.Hypothesis) + if err != nil { + t.Fatal(err) + } + if len(hyp.Observation.EvidenceRefs) == 0 { + t.Error("the hypothesis cites no evidence") + } + if len(hyp.CandidateRevisionRefs) == 0 { + t.Error("the hypothesis names no candidate revision") + } + if len(hyp.ExperimentRefs) == 0 { + t.Error("the hypothesis names no experiment") + } + + // And the revision must carry events reaching back to the decision. + events := h.Events(string(loop.Candidate)) + var sawDecision bool + for _, ev := range events { + if strings.HasPrefix(ev.EventType, "PROMOTION_DECIDED") { + sawDecision = true + if len(ev.Inputs) == 0 { + t.Error("the promotion decision cites no inputs") + } + } + } + if !sawDecision { + t.Error("no promotion decision is recorded against the candidate") + } +} + +// 6. Explicit responsibility boundaries. +func TestConformance06_ResponsibilityBoundaries(t *testing.T) { + h := New(t) + + // The interface owns representation and routing; the backend owns the data. + // When the backend is gone, the interface must report that rather than + // inventing a response. + h.StopAdapter("R-1") + + rec := h.Call("c-1", http.MethodGet, "/v1/entries") + if rec.Code != http.StatusBadGateway && rec.Code != http.StatusGatewayTimeout { + t.Fatalf("status %d; a dead backend must surface as a gateway error", rec.Code) + } + if strings.Contains(rec.Body.String(), "127.0.0.1") { + t.Error("the backend address leaked into the error response") + } +} + +// 7. Deterministic security enforcement independent of probabilistic decisions. +func TestConformance07_DeterministicSecurity(t *testing.T) { + h := New(t) + + // An unsigned descriptor is refused by the router regardless of anything + // else about it. + unsigned := contract.Revision{ + SchemaVersion: "0.1", ID: "R-rogue", Interface: Interface, + State: contract.RevisionStateStable, + Contract: contract.RevisionContract{Type: contract.RevisionContractTypeOpenapi, Digest: contract.Digest("sha256:" + strings.Repeat("9", 64))}, + Runtime: contract.RevisionRuntime{Upstream: h.AdapterURL("R-1")}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + }, + } + if err := h.Registry.PutRevision(unsigned); err == nil { + t.Fatal("an unsigned descriptor was accepted by the router") + } + + // A signed descriptor whose content was altered afterwards must also fail. + sig, err := h.Signer.Sign(contract.RevisionDescriptorDocument{Revision: unsigned}) + if err != nil { + t.Fatal(err) + } + tampered := unsigned + tampered.Runtime.Upstream = "http://attacker.invalid" + tampered.Signature = &contract.RevisionSignature{ + Algorithm: contract.RevisionSignatureAlgorithm(sig.Algorithm), + KeyID: sig.KeyID, Value: sig.Value, + } + if err := h.Registry.PutRevision(tampered); err == nil { + t.Fatal("a tampered descriptor was accepted by the router") + } +} + +// --------------------------------------------------------------------------- +// ArchitectureBlueprint.md section 55 — architectural invariants +// +// Only the mechanically checkable subset is asserted. The rest are design +// properties a test cannot settle, and pretending otherwise would be worse +// than leaving them to review. +// --------------------------------------------------------------------------- + +// Invariant 1: runtime behaviour remains deterministic. +// Invariant 2: evolution can stop without stopping the API. +func TestInvariant02_EvolutionCanStopWithoutStoppingTheAPI(t *testing.T) { + h := New(t) + + // Kill the entire observation and control plane: telemetry delivery first, + // then the evidence store itself. + h.Drain() + if err := h.Store.Close(); err != nil { + t.Fatal(err) + } + + // The data plane must keep serving from cached published configuration. + for i := 0; i < 10; i++ { + rec := h.Call("c-1", http.MethodGet, "/v1/entries") + if rec.Code != http.StatusOK { + t.Fatalf("request %d returned %d after the control plane died", i, rec.Code) + } + } +} + +// Invariant 3: every published revision is identifiable. +// Invariant 4: every revision has an explicit contract. +// Invariant 5: every revision is governed by a specific intent version. +func TestInvariant03to05_RevisionIdentityContractAndIntent(t *testing.T) { + h := New(t) + ctx := context.Background() + + for _, id := range []contract.RevisionID{"R-1", "R-2"} { + d, err := h.Registry.Revision(id) + if err != nil { + t.Fatalf("%s is not registered: %v", id, err) + } + if d.ID == "" { + t.Errorf("%s has no identity", id) + } + if d.Contract.Digest == "" { + t.Errorf("%s has no contract digest", id) + } + if _, ok := h.Validator.Contract(d.Contract.Digest); !ok { + t.Errorf("%s names a contract the gateway cannot enforce", id) + } + if d.Intent.Version == "" { + t.Errorf("%s declares no governing intent", id) + } + if _, err := h.Intents.GoverningIntent(ctx, id); err != nil { + t.Errorf("%s has no recorded intent binding: %v", id, err) + } + } +} + +// Invariant 7: every experiment has guardrails and stop conditions. +func TestInvariant07_ExperimentsAreBounded(t *testing.T) { + h := New(t) + ctx := context.Background() + + unbounded := contract.FluidExperiment{ + ID: "E-unbounded", HypothesisRefs: []contract.HypothesisID{"H-1"}, + ControlRevision: "R-1", CandidateRevisions: []contract.RevisionID{"R-2"}, + Allocation: map[string]contract.UnitInterval{"control": 0.9, "candidate": 0.1}, + Metrics: contract.FluidExperimentMetrics{Primary: []string{"m"}}, + // No stop conditions. + } + if _, err := h.Experiments.Design(ctx, unbounded, Operator); err == nil { + t.Error("an experiment with no stop condition was designed") + } +} + +// Invariant 8: every promotion is auditable. +// Invariant 9: AI-generated artifacts are untrusted until verified. +func TestInvariant09_GeneratedArtifactsAreUntrustedUntilVerified(t *testing.T) { + h := New(t) + + // A candidate that has not been through the pipeline cannot be published: + // Publish takes a Verified value and nothing else constructs one. + if err := h.Pipeline.Publish(context.Background(), publish.Verified{}); err == nil { + t.Fatal("an unverified value was published") + } +} + +// Invariant 10: security policy remains deterministic. +func TestInvariant10_SecurityPolicyIsDeterministic(t *testing.T) { + h := New(t) + + // The same descriptor must be judged identically every time. + d, err := h.Registry.Revision("R-1") + if err != nil { + t.Fatal(err) + } + first := h.Registry.CheckRoutable("R-1", "agents") + for i := 0; i < 50; i++ { + if got := h.Registry.CheckRoutable("R-1", "agents"); (got == nil) != (first == nil) { + t.Fatal("routability judgement varied between identical calls") + } + } + _ = d +} + +// Invariant 13: multiple revisions may coexist. +func TestInvariant13_RevisionsCoexist(t *testing.T) { + h := New(t) + + // Both revisions serve concurrently, distinguished only by resolution. + mustStatus(t, h.CallPinned("c-1", "R-1", http.MethodGet, "/v1/entries"), http.StatusOK, "R-1") + mustStatus(t, h.CallPinned("c-2", "R-2", http.MethodGet, "/v1/entries/latest"), http.StatusOK, "R-2 latest") + + // And the convenience resource exists only in R-2, which is what makes the + // two revisions genuinely different contracts rather than a relabelling. + mustStatus(t, h.CallPinned("c-3", "R-1", http.MethodGet, "/v1/entries/latest"), + http.StatusBadRequest, "R-1 must not serve a path it does not declare") +} + +// Invariant 14: failed experiments are normal and recoverable. +func TestInvariant14_FailedExperimentsAreRecoverable(t *testing.T) { + h := New(t) + ctx := context.Background() + + loop := SeedHypothesis(t, h) + exp := DesignExperiment(t, h, loop.Hypothesis) + + if _, _, err := h.Experiments.Start(ctx, exp, 2, "R-1", Operator); err != nil { + t.Fatal(err) + } + + // Stopping returns a policy with no rules; installing it puts every + // consumer back on the known-good revision. + _, stopPolicy, err := h.Experiments.Stop(ctx, exp, 3, "R-1", Operator, "guardrail breached") + if err != nil { + t.Fatal(err) + } + if len(stopPolicy.Rules) != 0 { + t.Fatalf("the stop policy still carries %d rules", len(stopPolicy.Rules)) + } + if err := h.Registry.PutPolicy(stopPolicy); err != nil { + t.Fatal(err) + } + + for i := 0; i < 5; i++ { + rec := h.Call("c-1", http.MethodGet, "/v1/entries") + mustStatus(t, rec, http.StatusOK, "after rollback") + if got := rec.Header().Get("X-FLUID-Revision"); got != "R-1" { + t.Fatalf("traffic did not return to the control revision: %q", got) + } + } +} + +// Invariant 16: adaptive cost is controlled independently of runtime +// availability. Exhausting the adaptive budget must not affect serving. +func TestInvariant16_AdaptiveBudgetDoesNotAffectRuntime(t *testing.T) { + h := New(t) + ctx := context.Background() + + // Saturate the experiment concurrency limit, the framework's current + // expression of an adaptive budget. + h.Experiments.SetMaxParallel(1) + loop := SeedHypothesis(t, h) + first := DesignExperiment(t, h, loop.Hypothesis) + if _, _, err := h.Experiments.Start(ctx, first, 2, "R-1", Operator); err != nil { + t.Fatal(err) + } + second := DesignExperimentNamed(t, h, loop.Hypothesis, "E-second") + if _, _, err := h.Experiments.Start(ctx, second, 3, "R-1", Operator); err == nil { + t.Fatal("the concurrency limit did not bind") + } + + // With adaptation blocked, the runtime is unaffected. + for i := 0; i < 5; i++ { + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, + "serving with the adaptive budget exhausted") + } +} + +// TestSignedDescriptorsRoundTrip guards the property the whole trust chain +// rests on: what the pipeline signs is what the router verifies. +func TestSignedDescriptorsRoundTrip(t *testing.T) { + h := New(t) + + d, err := h.Registry.Revision("R-2") + if err != nil { + t.Fatal(err) + } + if d.Signature == nil { + t.Fatal("a registered revision carries no signature") + } + + // The registry already verified this on insert; re-verifying here confirms + // the signature travelled with the descriptor rather than being dropped. + verifier := signing.NewVerifier(map[string]ed25519.PublicKey{"conformance-key": h.PublicKey}) + if err := verifier.Verify(contract.RevisionDescriptorDocument{Revision: d}, &signing.Signature{ + Algorithm: string(d.Signature.Algorithm), + KeyID: d.Signature.KeyID, + Value: d.Signature.Value, + }); err != nil { + t.Fatalf("a registered descriptor does not verify: %v", err) + } + if d.Signature.Algorithm != contract.RevisionSignatureAlgorithmEd25519 { + t.Errorf("unexpected signature algorithm %q", d.Signature.Algorithm) + } +} + +// TestTelemetryCarriesNoRawIdentity is the privacy boundary at the seam that +// matters: whatever a consumer sends, the store must not keep it. +func TestTelemetryCarriesNoRawIdentity(t *testing.T) { + h := New(t) + + h.Call("bernd@example.com", http.MethodGet, "/v1/entries") + h.Drain() + + for _, ev := range h.Telemetry() { + if strings.Contains(ev.ConsumerRef, "@") { + t.Fatalf("a raw consumer identity reached the store: %q", ev.ConsumerRef) + } + if ev.Redaction == nil { + t.Error("an event does not record its redaction status") + } + } +} + +var _ = evidence.EventFilter{} +var _ = runtime.Request{} diff --git a/conformance/suite/containment_test.go b/conformance/suite/containment_test.go new file mode 100644 index 0000000..e56dc30 --- /dev/null +++ b/conformance/suite/containment_test.go @@ -0,0 +1,236 @@ +package suite + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/runtime" +) + +// --------------------------------------------------------------------------- +// ArchitectureBlueprint.md section 34 — failure containment +// +// FLUID must be designed so evolutionary failure does not imply runtime +// failure. Each subsection names an effect and a runtime effect; the runtime +// effect is almost always "none", and these tests are what keep that true. +// --------------------------------------------------------------------------- + +// 34.1 Daimon failure: no new hypotheses or adaptations. Runtime effect: none. +// +// There is no Daimon yet, so the closest analogue is the whole science layer +// being unable to record anything. +func TestContainment_ScienceLayerFailure(t *testing.T) { + h := New(t) + ctx := context.Background() + + h.Drain() + if err := h.Store.Close(); err != nil { + t.Fatal(err) + } + + // Adaptation is now impossible. + if _, err := h.Hypotheses.List(ctx, ""); err == nil { + t.Log("note: hypothesis listing still succeeded; the store may be cached") + } + + // The runtime is unaffected. + for i := 0; i < 10; i++ { + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, + "serving with the science layer dead") + } +} + +// 34.2 Telemetry pipeline failure: reduced learning. Runtime effect: none. +// +// Telemetry backpressure MUST NOT block normal API requests. +func TestContainment_TelemetryBackpressure(t *testing.T) { + stalled := &stallingSink{} + h := New(t) + + // Replace the emitter with one writing into a sink that never completes. + h.Emitter.Close() + h.Emitter = runtime.NewEmitter(stalled, runtime.EmitterOptions{ + Buffer: 4, Workers: 1, WriteTimeout: time.Hour, + }) + + gw, err := runtime.NewGateway(runtime.GatewayOptions{ + Interface: Interface, + Registry: h.Registry, + Resolver: runtime.NewResolver(h.Registry, true), + Connector: runtime.NewConnector(), + Emitter: h.Emitter, + Cohorts: runtime.StaticCohort("agents"), + Validator: h.Validator, + }) + if err != nil { + t.Fatal(err) + } + h.Gateway = gw + + // Far more requests than the buffer can hold, against a sink that never + // drains. Latency must stay flat. + start := time.Now() + for i := 0; i < 200; i++ { + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, + "serving under telemetry backpressure") + } + elapsed := time.Since(start) + + if elapsed > 10*time.Second { + t.Fatalf("200 requests took %s under telemetry backpressure; the request path is coupled to the sink", elapsed) + } + if h.Emitter.Stats().Dropped == 0 { + t.Error("expected telemetry to be dropped rather than queued without bound") + } +} + +// 34.3 Builder failure: candidate not produced. Runtime effect: none. +func TestContainment_BuilderFailure(t *testing.T) { + h := New(t) + + // A candidate that fails verification is never published, and the running + // interface does not notice. + bad := contract.Revision{ + SchemaVersion: "0.1", ID: "R-bad", Interface: Interface, + State: contract.RevisionStateStable, + Contract: contract.RevisionContract{Type: contract.RevisionContractTypeOpenapi, Digest: contract.Digest("sha256:" + repeat("3", 64))}, + Runtime: contract.RevisionRuntime{Upstream: h.AdapterURL("R-1")}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityBreaking, + SecurityCheck: contract.RevisionPolicySecurityCheckFailed, + }, + } + if _, _, err := h.Pipeline.Run(context.Background(), + publish.NewCandidate(bad, Operator), publish.PromotionRequest{ + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassPresentation}, + RequestedTrafficShare: 0.1, + Approved: true, + ApprovedBy: &Operator, + }); err == nil { + t.Fatal("a candidate failing security passed verification") + } + + for i := 0; i < 5; i++ { + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, + "serving after a failed build") + } +} + +// 34.5 Experiment failure: candidate traffic removed. Runtime response: route +// to a known-good revision. +func TestContainment_ExperimentFailure(t *testing.T) { + h := New(t) + ctx := context.Background() + + loop := SeedHypothesis(t, h) + exp := DesignExperiment(t, h, loop.Hypothesis) + + _, startPolicy, err := h.Experiments.Start(ctx, exp, 2, "R-1", Operator) + if err != nil { + t.Fatal(err) + } + if err := h.Registry.PutPolicy(startPolicy); err != nil { + t.Fatal(err) + } + + // The candidate's adapter dies mid-experiment. + h.StopAdapter("R-2") + + // Stopping the experiment restores service for everyone, including + // consumers that had been allocated to the broken arm. + _, stopPolicy, err := h.Experiments.Stop(ctx, exp, 3, "R-1", Operator, "candidate adapter unreachable") + if err != nil { + t.Fatal(err) + } + if err := h.Registry.PutPolicy(stopPolicy); err != nil { + t.Fatal(err) + } + + for i := 0; i < 20; i++ { + rec := h.Call(consumerName(i), http.MethodGet, "/v1/entries") + mustStatus(t, rec, http.StatusOK, "serving after experiment rollback") + if got := rec.Header().Get("X-FLUID-Revision"); got != "R-1" { + t.Fatalf("consumer %d still routed to %q after rollback", i, got) + } + } +} + +// 34.6 Evolution store failure: runtime MUST continue using cached published +// revision configuration. +func TestContainment_EvidenceStoreFailure(t *testing.T) { + h := New(t) + + h.Drain() + if err := h.Store.Close(); err != nil { + t.Fatal(err) + } + + // Reads and writes to the store now fail. + if _, err := h.Store.Records(context.Background(), contract.KindRevision); err == nil { + t.Error("the evidence store still answers after being closed") + } + + // The registry holds its own cached view, so routing is unaffected. + for i := 0; i < 10; i++ { + rec := h.Call("c-1", http.MethodGet, "/v1/entries") + mustStatus(t, rec, http.StatusOK, "serving from cached configuration") + if rec.Header().Get("X-FLUID-Revision") != "R-1" { + t.Fatal("routing changed after the evidence store died") + } + } +} + +// 34.7 AI budget exhaustion: adaptive processes pause. Runtime effect: none. +// +// Covered by TestInvariant16_AdaptiveBudgetDoesNotAffectRuntime; this asserts +// the complementary direction, that a saturated adaptive layer cannot make the +// data plane refuse work. +func TestContainment_AdaptiveSaturation(t *testing.T) { + h := New(t) + ctx := context.Background() + + h.Experiments.SetMaxParallel(1) + loop := SeedHypothesis(t, h) + first := DesignExperiment(t, h, loop.Hypothesis) + if _, _, err := h.Experiments.Start(ctx, first, 2, "R-1", Operator); err != nil { + t.Fatal(err) + } + + second := DesignExperimentNamed(t, h, loop.Hypothesis, "E-blocked") + _, _, err := h.Experiments.Start(ctx, second, 3, "R-1", Operator) + if err == nil { + t.Fatal("the adaptive limit did not bind") + } + + for i := 0; i < 10; i++ { + mustStatus(t, h.Call("c-1", http.MethodGet, "/v1/entries"), http.StatusOK, + "serving with adaptation saturated") + } + _ = errors.Is +} + +// stallingSink never returns, standing in for a wedged evidence store. +type stallingSink struct{} + +func (s *stallingSink) Write(ctx context.Context, _ contract.FluidTelemetry) error { + <-ctx.Done() + return ctx.Err() +} + +func repeat(s string, n int) string { + out := make([]byte, 0, n*len(s)) + for i := 0; i < n; i++ { + out = append(out, s...) + } + return string(out) +} + +func consumerName(i int) string { + return "consumer-" + string(rune('a'+i%26)) +} diff --git a/conformance/suite/harness.go b/conformance/suite/harness.go new file mode 100644 index 0000000..b869797 --- /dev/null +++ b/conformance/suite/harness.go @@ -0,0 +1,310 @@ +// Package suite is the FLUID conformance harness. +// +// It asserts the seven minimal-conformance requirements of +// FluidAPIStandards.md section 36 and the mechanically checkable subset of the +// sixteen architectural invariants in ArchitectureBlueprint.md section 55. +// +// Everything here runs with no external services and no human steps. A +// conformance claim that needed someone to run through it by hand would be a +// claim nobody re-checks after the first time. +package suite + +import ( + "context" + "crypto/ed25519" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + _ "modernc.org/sqlite" + + echo "github.com/tegwick/fluid-core/examples/echo-interface" + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/evidence" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/observation" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/publish" + "github.com/tegwick/fluid-core/internal/runtime" + "github.com/tegwick/fluid-core/internal/science" + "github.com/tegwick/fluid-core/internal/signing" + "github.com/tegwick/fluid-core/internal/validate" +) + +// Interface is the fixture's interface identifier. +const Interface contract.InterfaceID = "echo-interface" + +// IntentDocument is the fixture's governing intent. +const IntentDocument = `# Interface Evolution Intent — echo-interface + +**Current operational authority mode:** +FLUID-4 + +## Mission +Serve entries to consumers, and learn from how they ask for them. +` + +// Operator is the fixture's human actor. +var Operator = contract.Actor{Type: contract.ActorTypeHuman, ID: "conformance"} + +// Harness is a fully wired FLUID deployment over the echo fixture. +type Harness struct { + T *testing.T + + Store *evidence.SQLStore + Intents *intent.Store + Registry *runtime.Registry + Gateway *runtime.Gateway + Emitter *runtime.Emitter + Ingest *observation.Ingest + Pipeline *publish.Pipeline + Validator *validate.OpenAPIValidator + Signer *signing.Signer + PublicKey ed25519.PublicKey + + Hypotheses *science.HypothesisStore + Experiments *science.ExperimentController + + adapters map[contract.RevisionID]*httptest.Server +} + +// New builds a harness with both fixture revisions published and R-1 stable. +func New(t *testing.T) *Harness { + t.Helper() + ctx := context.Background() + + store, err := evidence.OpenSQLite(ctx, filepath.Join(t.TempDir(), "conformance.db")) + if err != nil { + t.Fatalf("open evidence store: %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + + intents := intent.New(store, Interface) + if _, err := intents.Put(ctx, "IEI-1", IntentDocument); err != nil { + t.Fatalf("record intent: %v", err) + } + if err := intents.SetActive(ctx, "IEI-1"); err != nil { + t.Fatalf("activate intent: %v", err) + } + + signer, pub, err := signing.GenerateKey("conformance-key") + if err != nil { + t.Fatal(err) + } + + limits := policy.DefaultLimits() + // The fixture's intent grants FLUID-4, and the loop under test promotes a + // contract adaptation, so the fixture permits that class explicitly. + limits.AllowedAdaptationClasses = append(limits.AllowedAdaptationClasses, + contract.AdaptationClassContract) + limits.RequiredMode = intent.ModeExperimental + gate := policy.NewGate(limits) + + pipeline, err := publish.New(publish.Options{ + Gate: gate, Signer: signer, Store: store, Intents: intents, + }) + if err != nil { + t.Fatal(err) + } + + salt := []byte("conformance-salt-of-sufficient-length") + ingest, err := observation.NewIngest(store, Interface, observation.DefaultRedactionPolicy(salt)) + if err != nil { + t.Fatal(err) + } + + h := &Harness{ + T: t, Store: store, Intents: intents, Pipeline: pipeline, + Validator: validate.NewOpenAPIValidator(), Signer: signer, PublicKey: pub, + Ingest: ingest, + adapters: map[contract.RevisionID]*httptest.Server{}, + } + + // A verifying registry: the router must accept only signed descriptors. + h.Registry = runtime.NewVerifiedRegistry(Interface, + signing.NewVerifier(map[string]ed25519.PublicKey{"conformance-key": pub})) + + h.adapters["R-1"] = httptest.NewServer(echo.NewR1()) + h.adapters["R-2"] = httptest.NewServer(echo.NewR2()) + t.Cleanup(func() { + for _, s := range h.adapters { + s.Close() + } + }) + + h.Emitter = runtime.NewEmitter(ingest, runtime.EmitterOptions{Buffer: 8192, Workers: 2}) + + gw, err := runtime.NewGateway(runtime.GatewayOptions{ + Interface: Interface, + Registry: h.Registry, + Resolver: runtime.NewResolver(h.Registry, true), + Connector: runtime.NewConnector(), + Emitter: h.Emitter, + Cohorts: observation.NewCohortEngine("agents", observation.DefaultRedactionPolicy(salt)), + Response: runtime.ResponsePolicy{FeedbackPath: "/v1/feedback"}, + Validator: h.Validator, + }) + if err != nil { + t.Fatal(err) + } + h.Gateway = gw + + h.Hypotheses = science.NewHypothesisStore(store, Interface) + h.Experiments = science.NewExperimentController(store, h.Hypotheses, Interface) + + h.PublishRevision("R-1", "examples/echo-interface/r1.openapi.yaml", nil) + h.PublishRevision("R-2", "examples/echo-interface/r2.openapi.yaml", + []contract.AdaptationClass{contract.AdaptationClassContract}) + h.InstallPolicy(1, "R-1", nil) + + return h +} + +// PublishRevision takes a revision through the full pipeline and registers it. +func (h *Harness) PublishRevision(id contract.RevisionID, contractPath string, classes []contract.AdaptationClass) contract.Revision { + h.T.Helper() + ctx := context.Background() + + raw := readRepoFile(h.T, contractPath) + digest := signing.Digest(raw) + + if err := h.Validator.Register(digest, raw); err != nil { + h.T.Fatalf("register contract for %s: %v", id, err) + } + + pc := contract.RevisionPolicyPolicyCheckPassed + descriptor := contract.Revision{ + SchemaVersion: "0.1", + ID: id, + Interface: Interface, + State: contract.RevisionStateStable, + Contract: contract.RevisionContract{ + Type: contract.RevisionContractTypeOpenapi, + Digest: digest, + Source: contractPath, + }, + Runtime: contract.RevisionRuntime{Upstream: h.adapters[id].URL}, + Intent: contract.RevisionIntent{Version: "IEI-1"}, + Policy: contract.RevisionPolicy{ + Compatibility: contract.RevisionPolicyCompatibilityAdditive, + SecurityCheck: contract.RevisionPolicySecurityCheckPassed, + PolicyCheck: &pc, + }, + } + + if classes == nil { + classes = []contract.AdaptationClass{contract.AdaptationClassPresentation} + } + + verified, report, err := h.Pipeline.Run(ctx, publish.NewCandidate(descriptor, Operator), + publish.PromotionRequest{ + AdaptationClasses: classes, + ComplexityDelta: 0.2, + RequestedTrafficShare: 0.2, + Approved: true, + ApprovedBy: &Operator, + }) + if err != nil { + h.T.Fatalf("publish %s: %v (%+v)", id, err, report.Stages) + } + if err := h.Pipeline.Publish(ctx, verified); err != nil { + h.T.Fatalf("record %s: %v", id, err) + } + if err := h.Registry.PutRevision(verified.Descriptor()); err != nil { + h.T.Fatalf("register %s: %v", id, err) + } + return verified.Descriptor() +} + +// InstallPolicy loads a routing policy into the registry. +func (h *Harness) InstallPolicy(generation int64, defaultRevision contract.RevisionID, rules []contract.RoutingPolicyRulesItem) { + h.T.Helper() + if rules == nil { + rules = []contract.RoutingPolicyRulesItem{} + } + if err := h.Registry.PutPolicy(contract.RoutingPolicy{ + SchemaVersion: "0.1", + Interface: Interface, + Generation: generation, + DefaultRevision: defaultRevision, + Rules: rules, + }); err != nil { + h.T.Fatalf("install policy generation %d: %v", generation, err) + } +} + +// Call issues a request through the gateway as a named consumer. +func (h *Harness) Call(consumer, method, target string) *httptest.ResponseRecorder { + h.T.Helper() + req := httptest.NewRequest(method, target, nil) + req.Header.Set("X-FLUID-Consumer", consumer) + rec := httptest.NewRecorder() + h.Gateway.ServeHTTP(rec, req) + return rec +} + +// CallPinned issues a request pinned to a specific revision. +func (h *Harness) CallPinned(consumer string, rev contract.RevisionID, method, target string) *httptest.ResponseRecorder { + h.T.Helper() + req := httptest.NewRequest(method, target, nil) + req.Header.Set("X-FLUID-Consumer", consumer) + req.Header.Set("X-FLUID-Revision", string(rev)) + rec := httptest.NewRecorder() + h.Gateway.ServeHTTP(rec, req) + return rec +} + +// Drain flushes buffered telemetry so assertions see every event. +// +// It flushes rather than closes: the harness keeps serving after a drain, and +// closing here would mean every later request emitted into a dead emitter. +func (h *Harness) Drain() { + if !h.Emitter.Flush(5 * time.Second) { + h.T.Fatal("telemetry did not reach the store within 5s") + } +} + +// Telemetry returns everything recorded for the interface. +func (h *Harness) Telemetry() []contract.FluidTelemetry { + h.T.Helper() + rows, err := h.Store.Telemetry(context.Background(), + evidence.TelemetryFilter{InterfaceID: Interface}) + if err != nil { + h.T.Fatal(err) + } + return rows +} + +// Events returns audit events for an entity. +func (h *Harness) Events(entityID string) []contract.FluidEvent { + h.T.Helper() + events, err := h.Store.Events(context.Background(), evidence.EventFilter{EntityID: entityID}) + if err != nil { + h.T.Fatal(err) + } + return events +} + +// AdapterURL returns a fixture adapter's address. +func (h *Harness) AdapterURL(rev contract.RevisionID) string { + return h.adapters[rev].URL +} + +// StopAdapter kills a fixture adapter, for failure-containment tests. +func (h *Harness) StopAdapter(rev contract.RevisionID) { + h.adapters[rev].Close() +} + +// mustStatus fails unless the response carries the expected status. +func mustStatus(t *testing.T, rec *httptest.ResponseRecorder, want int, context string) { + t.Helper() + if rec.Code != want { + t.Fatalf("%s: status %d, want %d; body %s", context, rec.Code, want, rec.Body.String()) + } +} + +var _ = fmt.Sprintf +var _ = http.MethodGet diff --git a/conformance/suite/loop.go b/conformance/suite/loop.go new file mode 100644 index 0000000..3aa9dd0 --- /dev/null +++ b/conformance/suite/loop.go @@ -0,0 +1,321 @@ +package suite + +import ( + "context" + "fmt" + "net/http" + "testing" + "time" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/fitness" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/observation" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/promotion" + "github.com/tegwick/fluid-core/internal/science" +) + +// Loop records the artifacts one pass of the adaptation loop produced. +type Loop struct { + Pressure contract.PressureID + Hypothesis contract.HypothesisID + Experiment contract.ExperimentID + Candidate contract.RevisionID + Evaluation fitness.Evaluation + Decision promotion.Decision +} + +// GeneratePressure drives the traffic that makes R-1's shortcoming visible. +// +// Several independent consumers each fetch the whole collection repeatedly +// within one task, which is the Blueprint section 33 shape: the interface is +// making them assemble something it could have handed them. +func GeneratePressure(t *testing.T, h *Harness) { + t.Helper() + for c := 0; c < 5; c++ { + consumer := fmt.Sprintf("agent-%d", c) + for chain := 0; chain < 4; chain++ { + for i := 0; i < 3; i++ { + rec := h.Call(consumer, http.MethodGet, "/v1/entries") + mustStatus(t, rec, http.StatusOK, "pressure traffic") + } + } + } +} + +// DetectPressure classifies recorded telemetry and records what it finds. +func DetectPressure(t *testing.T, h *Harness) contract.PressureID { + t.Helper() + ctx := context.Background() + + classifier := observation.NewClassifier( + observation.DefaultClassifierOptions(), observation.NewTopologyAnalyzer()) + findings := classifier.Classify(h.Telemetry()) + if len(findings) == 0 { + t.Fatal("no pressure detected from traffic that plainly shows it") + } + + registry := observation.NewPressureRegistry(h.Store, Interface) + recorded, err := registry.RecordAll(ctx, findings) + if err != nil { + t.Fatal(err) + } + + for _, p := range recorded { + if p.Class == contract.PressureClassSuccessfulButInefficientUsage { + return p.ID + } + } + t.Fatalf("inefficient usage was not among the findings: %v", recorded) + return "" +} + +// SeedHypothesis creates and prepares a hypothesis explaining the pressure. +func SeedHypothesis(t *testing.T, h *Harness) Loop { + t.Helper() + ctx := context.Background() + + hypothesis := contract.FluidHypothesis{ + ID: "H-1", + Title: "latest entry is a first-class consumer concept", + Observation: contract.FluidHypothesisObservation{ + Summary: "consumers fetch the whole collection repeatedly to find the newest entry", + EvidenceRefs: []contract.EvidenceRef{"topology:GET /v1/entries"}, + }, + Pressure: contract.FluidHypothesisPressure{ + Classes: []contract.PressureClass{contract.PressureClassSuccessfulButInefficientUsage}, + }, + Explanation: contract.FluidHypothesisExplanation{ + Claim: "the collection resource does not name a concept consumers hold, so they assemble it themselves", + }, + ProposedAdaptation: contract.FluidHypothesisProposedAdaptation{ + Class: contract.AdaptationClassContract, + Summary: "add an explicit latest-entry resource", + }, + ExpectedOutcomes: []contract.ExpectedOutcome{{ + Metric: fitness.MetricRequestsPerTask, + Target: 1.2, + Direction: contract.ExpectedOutcomeDirectionLower, + }}, + Guardrails: []contract.Guardrail{{ + Metric: fitness.MetricErrorRate, Operator: contract.GuardrailOperatorLte, Threshold: 0.01, + }}, + SuccessCriteria: contract.FluidHypothesisSuccessCriteria{ + Expression: "requests_per_completed_task <= 1.2 with no guardrail violation", + }, + Complexity: contract.FluidHypothesisComplexity{ + ExpectedDelta: contract.ComplexityDelta{OperationCount: ptr(1.0)}, + }, + Risk: contract.FluidHypothesisRisk{Level: contract.FluidHypothesisRiskLevelLOW}, + } + + if _, err := h.Hypotheses.Create(ctx, hypothesis, Operator); err != nil { + t.Fatal(err) + } + for _, s := range []contract.FluidHypothesisState{ + contract.FluidHypothesisStateREADY, + contract.FluidHypothesisStatePRIORITIZED, + contract.FluidHypothesisStateDESIGNING, + } { + if _, err := h.Hypotheses.Transition(ctx, "H-1", s, Operator, "advancing the loop"); err != nil { + t.Fatal(err) + } + } + if err := h.Hypotheses.AttachRevision(ctx, "H-1", "R-2", Operator); err != nil { + t.Fatal(err) + } + + return Loop{Hypothesis: "H-1", Candidate: "R-2"} +} + +// DesignExperiment creates the 90/10 experiment the Blueprint slice calls for. +func DesignExperiment(t *testing.T, h *Harness, hypothesis contract.HypothesisID) contract.ExperimentID { + t.Helper() + return DesignExperimentNamed(t, h, hypothesis, "E-1") +} + +// DesignExperimentNamed creates an experiment with a chosen id. +func DesignExperimentNamed(t *testing.T, h *Harness, hypothesis contract.HypothesisID, id contract.ExperimentID) contract.ExperimentID { + t.Helper() + ctx := context.Background() + + e := contract.FluidExperiment{ + ID: id, + HypothesisRefs: []contract.HypothesisID{hypothesis}, + ControlRevision: "R-1", + CandidateRevisions: []contract.RevisionID{"R-2"}, + Cohorts: []contract.CohortID{"agents"}, + Allocation: map[string]contract.UnitInterval{"control": 0.9, "candidate": 0.1}, + Metrics: contract.FluidExperimentMetrics{ + Primary: []string{fitness.MetricRequestsPerTask}, + Guardrails: []string{fitness.MetricErrorRate}, + }, + StopConditions: []string{"hard_guardrail_violation", "manual_stop"}, + } + if _, err := h.Experiments.Design(ctx, e, Operator); err != nil { + t.Fatal(err) + } + return id +} + +// RunFullLoop executes the complete section 50 vertical slice. +// +// Two deterministic revisions, explicit routing, telemetry, one cohort +// dimension, pressure detection, a hypothesis, a candidate revision, a bounded +// experiment, fitness comparison, promotion, and a complete audit trail — with +// no human steps. +func RunFullLoop(t *testing.T, h *Harness) Loop { + t.Helper() + ctx := context.Background() + + // 1-3. Traffic, telemetry, pressure. + GeneratePressure(t, h) + h.Drain() + pressureID := DetectPressure(t, h) + + // 4. Hypothesis explaining it. + loop := SeedHypothesis(t, h) + loop.Pressure = pressureID + + // 5. Link the evidence to the explanation. + registry := observation.NewPressureRegistry(h.Store, Interface) + if err := registry.LinkHypothesis(ctx, pressureID, loop.Hypothesis); err != nil { + t.Fatal(err) + } + + // 6. A bounded experiment, enacted through routing policy. + expID := DesignExperiment(t, h, loop.Hypothesis) + loop.Experiment = expID + + _, startPolicy, err := h.Experiments.Start(ctx, expID, 2, "R-1", Operator) + if err != nil { + t.Fatal(err) + } + if err := h.Registry.PutPolicy(startPolicy); err != nil { + t.Fatal(err) + } + + // 7. Measure both arms. Consumers are pinned so each arm gets the shape it + // is meant to demonstrate; the allocation itself is exercised separately in + // the resolver's own tests. + observed := measureBothArms(t, h) + + // 8. Fitness comparison against the declared criteria. + specs := []fitness.MetricSpec{ + {Name: fitness.MetricRequestsPerTask, Role: fitness.RolePrimary, + Direction: fitness.Lower, Target: ptr(1.2)}, + {Name: fitness.MetricErrorRate, Role: fitness.RoleGuardrail, + Direction: fitness.Lower, Threshold: ptr(0.01)}, + } + evaluator := fitness.NewEvaluator() + evaluator.MinSamples = 10 + loop.Evaluation = evaluator.Evaluate("R-1", "R-2", observed.window, specs, observed.observations) + + // 9. Stop the experiment and record its conclusion. + _, stopPolicy, err := h.Experiments.Stop(ctx, expID, 3, "R-1", Operator, "measurement window elapsed") + if err != nil { + t.Fatal(err) + } + if err := h.Registry.PutPolicy(stopPolicy); err != nil { + t.Fatal(err) + } + if _, err := h.Experiments.Finalize(ctx, expID, "R-2", Operator, + fmt.Sprintf("fitness verdict %s", loop.Evaluation.Verdict), nil); err != nil { + t.Fatal(err) + } + + // 10. Record the hypothesis outcome. + status := contract.FluidHypothesisOutcomeStatusREFUTED + if loop.Evaluation.Verdict == fitness.VerdictSucceeded { + status = contract.FluidHypothesisOutcomeStatusCONFIRMED + } + if _, err := h.Hypotheses.RecordOutcome(ctx, loop.Hypothesis, status, + fmt.Sprintf("verdict %s: %v", loop.Evaluation.Verdict, loop.Evaluation.Reasons), nil, Operator); err != nil { + t.Fatal(err) + } + + // 11. Promotion, through the deterministic gate. + loop.Decision = decide(t, h, loop) + return loop +} + +type measurement struct { + observations []fitness.Observation + window fitness.Window +} + +// measureBothArms drives the two revisions and derives their metrics. +func measureBothArms(t *testing.T, h *Harness) measurement { + t.Helper() + start := time.Now().Add(-time.Hour) + + // R-1: three calls per task, the shape the pressure described. + for c := 0; c < 12; c++ { + consumer := fmt.Sprintf("control-%d", c) + for i := 0; i < 3; i++ { + mustStatus(t, h.CallPinned(consumer, "R-1", http.MethodGet, "/v1/entries"), + http.StatusOK, "control arm") + } + } + // R-2: one call per task, using the resource the hypothesis proposed. + for c := 0; c < 12; c++ { + consumer := fmt.Sprintf("candidate-%d", c) + mustStatus(t, h.CallPinned(consumer, "R-2", http.MethodGet, "/v1/entries/latest"), + http.StatusOK, "candidate arm") + } + h.Drain() + + window := fitness.Window{Start: start} + return measurement{ + observations: fitness.NewMeasurer().Measure(h.Telemetry(), window), + window: window, + } +} + +func decide(t *testing.T, h *Harness, loop Loop) promotion.Decision { + t.Helper() + ctx := context.Background() + + descriptor, err := h.Registry.Revision(loop.Candidate) + if err != nil { + t.Fatal(err) + } + governing, err := h.Intents.GoverningIntent(ctx, loop.Candidate) + if err != nil { + t.Fatal(err) + } + + limits := policy.DefaultLimits() + limits.AllowedAdaptationClasses = append(limits.AllowedAdaptationClasses, + contract.AdaptationClassContract) + limits.RequiredMode = intent.ModeExperimental + + controller := promotion.NewController(h.Store, policy.NewGate(limits)) + d, err := controller.Decide(ctx, promotion.Request{ + Revision: loop.Candidate, + Outcome: promotion.Promote, + Reason: "the candidate met its primary target with no guardrail breach", + Actor: Operator, + Experiment: loop.Experiment, + Hypotheses: []contract.HypothesisID{loop.Hypothesis}, + Evaluation: &loop.Evaluation, + GateInput: &policy.Input{ + Descriptor: descriptor, + GoverningMode: governing.Mode, + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassContract}, + ComplexityDelta: 0.2, + RequestedTrafficShare: 0.2, + Approved: true, + ApprovedBy: &Operator, + }, + }) + if err != nil { + t.Fatalf("promotion refused: %v", err) + } + return d +} + +func ptr[T any](v T) *T { return &v } + +var _ = science.CanTransition diff --git a/conformance/suite/repo.go b/conformance/suite/repo.go new file mode 100644 index 0000000..2b1c6c0 --- /dev/null +++ b/conformance/suite/repo.go @@ -0,0 +1,35 @@ +package suite + +import ( + "os" + "path/filepath" + "testing" +) + +// readRepoFile reads a path relative to the repository root. +// +// Tests run from their own package directory, so fixture paths are resolved +// against the module root rather than the working directory. Hard-coding +// "../.." would break the moment the suite moved. +func readRepoFile(t *testing.T, rel string) []byte { + t.Helper() + + dir, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + raw, err := os.ReadFile(filepath.Join(dir, rel)) + if err != nil { + t.Fatalf("read %s: %v", rel, err) + } + return raw + } + parent := filepath.Dir(dir) + if parent == dir { + t.Fatalf("could not find the repository root above %s", dir) + } + dir = parent + } +} diff --git a/conformance/suite/slice_test.go b/conformance/suite/slice_test.go new file mode 100644 index 0000000..8de9816 --- /dev/null +++ b/conformance/suite/slice_test.go @@ -0,0 +1,222 @@ +package suite + +import ( + "context" + "strings" + "testing" + + "github.com/tegwick/fluid-core/internal/contract" + "github.com/tegwick/fluid-core/internal/fitness" + "github.com/tegwick/fluid-core/internal/intent" + "github.com/tegwick/fluid-core/internal/policy" + "github.com/tegwick/fluid-core/internal/promotion" +) + +// TestFirstVerticalSlice is the whole point of the conformance suite. +// +// ArchitectureBlueprint.md section 50 lists eleven things the first FLUID +// deployment must demonstrate, and says plainly what success means: +// +// The first success criterion is not autonomous coding. +// It is proving that the revision-experiment-fitness loop works cleanly +// and safely. +// +// This test runs all eleven with no human steps. +func TestFirstVerticalSlice(t *testing.T) { + h := New(t) + loop := RunFullLoop(t, h) + + // 1. Two deterministic revisions. + for _, id := range []contract.RevisionID{"R-1", "R-2"} { + if _, err := h.Registry.Revision(id); err != nil { + t.Fatalf("revision %s is not published: %v", id, err) + } + } + + // 2. Explicit revision routing. + policy, err := h.Registry.Policy() + if err != nil { + t.Fatalf("no routing policy: %v", err) + } + if policy.DefaultRevision == "" { + t.Error("the routing policy names no default revision") + } + + // 3. Telemetry. + telemetry := h.Telemetry() + if len(telemetry) == 0 { + t.Fatal("no telemetry recorded") + } + + // 4. One consumer cohort dimension. + var sawCohort bool + for _, ev := range telemetry { + if ev.Cohort != nil && *ev.Cohort != "" { + sawCohort = true + break + } + } + if !sawCohort { + t.Error("no telemetry carries a cohort assignment") + } + + // 5. Pressure detection. + if loop.Pressure == "" { + t.Fatal("no pressure was detected") + } + + // 6. A hypothesis, linked to that pressure. + hyp, err := h.Hypotheses.Get(context.Background(), loop.Hypothesis) + if err != nil { + t.Fatal(err) + } + if hyp.Explanation.Claim == "" { + t.Error("the hypothesis offers no explanation") + } + + // 7. A candidate revision, claimed by that hypothesis. + var claimsCandidate bool + for _, r := range hyp.CandidateRevisionRefs { + if r == loop.Candidate { + claimsCandidate = true + } + } + if !claimsCandidate { + t.Errorf("the hypothesis does not claim %s", loop.Candidate) + } + + // 8. A controlled 90/10 experiment. + exp, err := h.Experiments.Get(context.Background(), loop.Experiment) + if err != nil { + t.Fatal(err) + } + if exp.Allocation["control"] != 0.9 || exp.Allocation["candidate"] != 0.1 { + t.Errorf("allocation was %v, want a 90/10 split", exp.Allocation) + } + + // 9. Fitness comparison. + if loop.Evaluation.Verdict != fitness.VerdictSucceeded { + t.Fatalf("verdict %s: %v", loop.Evaluation.Verdict, loop.Evaluation.Reasons) + } + primary := loop.Evaluation.PrimaryResults() + if len(primary) != 1 { + t.Fatalf("expected one primary metric, got %d", len(primary)) + } + // The candidate should show the reduction the hypothesis predicted. + if primary[0].Current >= primary[0].Baseline { + t.Errorf("requests per task did not fall: %v -> %v", + primary[0].Baseline, primary[0].Current) + } + t.Logf("requests per completed task: %.2f -> %.2f (target %.2f)", + primary[0].Baseline, primary[0].Current, *primary[0].Target) + + // 10. Promotion. + if loop.Decision.Outcome != promotion.Promote { + t.Errorf("decision was %s, want PROMOTE", loop.Decision.Outcome) + } + if loop.Decision.Override { + t.Error("the promotion was an override; the evidence should have carried it") + } + + // 11. A complete audit trail. + // + // Every stage of the loop must have left a reconstructable record. This is + // the assertion that would fail first if any part of the framework started + // deciding things without saying so. + events := h.Events(string(loop.Candidate)) + required := map[string]bool{ + "SIGN_PASSED": false, + "POLICY_CHECK_PASSED": false, + "INTENT_BOUND": false, + "REVISION_PUBLISHED": false, + "PROMOTION_DECIDED_PROMOTE": false, + } + for _, ev := range events { + if _, tracked := required[ev.EventType]; tracked { + required[ev.EventType] = true + } + } + for name, seen := range required { + if !seen { + t.Errorf("the audit trail is missing %s", name) + } + } + + // The pressure, hypothesis and experiment each left their own trail. + for _, entity := range []string{ + string(loop.Pressure), string(loop.Hypothesis), string(loop.Experiment), + } { + if len(h.Events(entity)) == 0 { + t.Errorf("%s left no audit events", entity) + } + } + + t.Logf("slice complete: %s -> %s -> %s -> %s (%s)", + loop.Pressure, loop.Hypothesis, loop.Experiment, loop.Candidate, loop.Decision.Outcome) +} + +// TestSliceIsReproducible runs the loop twice over fresh state and requires the +// same conclusion both times. +// +// A loop whose verdict depended on run order would be measuring the harness +// rather than the interface. +func TestSliceIsReproducible(t *testing.T) { + var verdicts []fitness.Verdict + var ratios []float64 + + for i := 0; i < 2; i++ { + h := New(t) + loop := RunFullLoop(t, h) + verdicts = append(verdicts, loop.Evaluation.Verdict) + ratios = append(ratios, loop.Evaluation.PrimaryResults()[0].Current) + } + + if verdicts[0] != verdicts[1] { + t.Errorf("verdict varied between runs: %s then %s", verdicts[0], verdicts[1]) + } + if ratios[0] != ratios[1] { + t.Errorf("measured ratio varied between runs: %v then %v", ratios[0], ratios[1]) + } +} + +// TestSliceRefusesPromotionWithoutEvidence confirms the loop cannot be +// short-circuited: the same candidate, promoted without its experiment, is +// refused. +func TestSliceRefusesPromotionWithoutEvidence(t *testing.T) { + h := New(t) + + descriptor, err := h.Registry.Revision("R-2") + if err != nil { + t.Fatal(err) + } + governing, err := h.Intents.GoverningIntent(context.Background(), "R-2") + if err != nil { + t.Fatal(err) + } + + limits := policy.DefaultLimits() + limits.AllowedAdaptationClasses = append(limits.AllowedAdaptationClasses, + contract.AdaptationClassContract) + limits.RequiredMode = intent.ModeExperimental + + _, err = promotion.NewController(h.Store, policy.NewGate(limits)).Decide(context.Background(), promotion.Request{ + Revision: "R-2", + Outcome: promotion.Promote, + Reason: "it looks better", + Actor: Operator, + GateInput: &policy.Input{ + Descriptor: descriptor, + GoverningMode: governing.Mode, + AdaptationClasses: []contract.AdaptationClass{contract.AdaptationClassContract}, + RequestedTrafficShare: 0.2, + Approved: true, + ApprovedBy: &Operator, + }, + }) + if err == nil { + t.Fatal("a candidate was promoted with no fitness evidence") + } + if !strings.Contains(err.Error(), "fitness") { + t.Errorf("refusal does not cite missing evidence: %v", err) + } +} diff --git a/docs/integration-guide.md b/docs/integration-guide.md new file mode 100644 index 0000000..ecd7b8e --- /dev/null +++ b/docs/integration-guide.md @@ -0,0 +1,137 @@ +# Putting an existing API behind fluid-core + +fluid-core attaches **out of process** (ADR-0002). Your API contributes no code, +imports no library, and may be written in any stack. Integration is over a wire +contract, so the fact that fluid-core is written in Go is not something you need +to care about. + +## What you need + +1. **A running service.** Anything that speaks HTTP. This becomes an *adapter*. +2. **A contract.** OpenAPI 3.1 today; the validator interface admits others. +3. **An interface evolution intent.** The governance document that bounds how + the interface may change. + +Nothing else. In particular you do not need a Kubernetes cluster, a message +broker, or Postgres — SQLite is the default evidence store. + +## The five minutes version + +```bash +export FLUID_INTERFACE=my-api +export FLUID_STORE=$PWD/fluid.db + +# 1. Record the intent that governs this interface. +fluid intent put --version IEI-1 --file InterfaceEvolutionIntent.md --activate + +# 2. Publish a revision pointing at your existing service. +fluid revision publish --file r1.yaml --ephemeral-key \ + --adaptation-classes presentation --approved-by "$USER" --traffic-share 1.0 + +# 3. Route traffic to it. +fluid policy put --file routing-policy.yaml +``` + +A revision descriptor is small: + +```yaml +revision: + schema_version: "0.1" + id: "R-1" + interface: "my-api" + state: "stable" + contract: + type: "openapi" + digest: "sha256:" + runtime: + upstream: "http://my-existing-service:8080" # your service, unchanged + timeout_ms: 5000 + intent: + version: "IEI-1" + policy: + compatibility: "additive" + security_check: "passed" +``` + +`runtime.upstream` is the whole integration. Everything else is metadata about +what that upstream is allowed to be. + +## What you get immediately + +- **Revision identity.** Every response names the revision that served it. +- **Telemetry.** Requests, errors, call sequences and adoption, redacted before + storage. +- **Contract enforcement.** Undeclared paths, methods and fields are refused + rather than proxied. +- **An audit trail.** `fluid audit trace R-1` reconstructs how a revision came + to exist and what happened to it. + +That is FLUID-0 (Instrumented) conformance, and it needs no AI and no change to +your service. + +## What you should decide deliberately + +**The pseudonymization salt.** Consumer identities are pseudonymized with HMAC. +The salt must be stable for the life of the interface — rotating it makes the +same consumer look like a new one and breaks every longitudinal comparison. +Store it where you store secrets, not in the repository. + +**The authority mode.** Your intent document declares one, `FLUID-0` through +`FLUID-6`. This is a statement about authority, not maturity: a high-assurance +interface may deliberately stay at FLUID-2 forever. The unfilled template is +refused rather than defaulted, precisely so nobody inherits a permissive mode by +accident. + +**Your complexity budget.** `policy.DefaultLimits()` permits only presentation +and implementation adaptations, refuses breaking changes, and caps exposure at +25%. If your interface needs more, say so explicitly in the intent rather than +widening the default. + +## Adding a second revision + +This is where FLUID earns its keep. Publish a second revision pointing at a +second adapter — a different build, a different service, a different language — +and let the experiment controller split traffic between them: + +```bash +fluid revision publish --file r2.yaml --ephemeral-key ... +fluid experiment design --file experiment.yaml +fluid experiment start E-1 --generation 2 --default-revision R-1 --policy-out rp.json +fluid policy put --file rp.json +``` + +The controller does not touch traffic. It emits a routing policy for you to +install, which is what makes an experiment interruptible: stopping one replaces +a document rather than unwinding anything. + +## Signing + +`--ephemeral-key` is for development. It generates a throwaway key and warns +that revisions signed with it will not verify after a restart, which is exactly +what you want to hear before it happens in production. + +In production, generate a key pair, keep the private half in your secret store, +and configure the gateway's registry with the public half. The router accepts +only descriptors that verify — an unsigned or tampered descriptor is refused, +including one that was signed and then edited. + +## What fluid-core will not do for you + +- **It will not modify your backend.** Where a candidate needs capability the + backend does not have, FLUID emits a structured requirement and stops. The + backend owner accepts, plans, or declares it out of scope. +- **It will not decide what "better" means.** There is no universal fitness + scalar. You declare primary metrics and guardrails; the evaluator compares + against what you declared and refuses to infer criteria from the data. +- **It will not promote on thin evidence.** A comparison below the sample floor + is inconclusive, not successful. You can override that with an acknowledged + flag; the override is recorded as one. + +## Reference + +- `examples/echo-interface/` — the two-revision fixture the conformance suite + drives, and the smallest complete example. +- `conformance/suite/` — every conformance requirement asserted as a test. + `TestFirstVerticalSlice` is the whole loop in one function. +- `spec/` — the normative documents. When this guide and the spec disagree, the + spec is right and this guide is a bug. diff --git a/examples/echo-interface/README.md b/examples/echo-interface/README.md new file mode 100644 index 0000000..f5bb095 --- /dev/null +++ b/examples/echo-interface/README.md @@ -0,0 +1,21 @@ +# echo-interface + +The smallest honest reproduction of the `ArchitectureBlueprint.md` §33 worked +example, used as the conformance fixture. + +Two revisions of one interface over the same backend data: + +| Revision | Shape | Requests per completed task | +|---|---|---| +| **R-1** | `GET /v1/entries` only — consumers list everything and filter locally | ~3 | +| **R-2** | adds `GET /v1/entries/latest` — the concept consumers actually wanted | 1 | + +R-1 is not a strawman. It is the interface a careful designer produces before +they have seen how it is used: a clean collection resource with no special +cases. The pressure it generates is the point — consumers repeatedly fetching +a collection to discard all but one item is the evidence that "latest" is a +first-class concept the contract failed to name. + +The fixture has no external dependencies and runs in CI. It exists so the +revision–experiment–fitness loop is proven mechanically before a real workload +depends on it. diff --git a/examples/echo-interface/adapter.go b/examples/echo-interface/adapter.go new file mode 100644 index 0000000..ea7c5c9 --- /dev/null +++ b/examples/echo-interface/adapter.go @@ -0,0 +1,67 @@ +// Package echo provides the conformance fixture adapters. +// +// Two revisions over the same data: R-1 exposes only a collection, R-2 adds the +// convenience resource. Both are ordinary deterministic HTTP handlers, which is +// the point — an adapter is a normal service, and fluid-core sits in front of +// it without asking anything of it. +package echo + +import ( + "encoding/json" + "net/http" + "sort" + "time" +) + +// Entry is one record the fixture serves. +type Entry struct { + ID string `json:"id"` + Title string `json:"title"` + CreatedAt time.Time `json:"created_at"` +} + +// Data returns a deterministic set of entries. +// +// Fixed timestamps rather than time.Now: a conformance suite whose fixture +// changes between runs cannot distinguish a regression from the clock. +func Data() []Entry { + base := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC) + return []Entry{ + {ID: "e-1", Title: "the river reached the forge", CreatedAt: base}, + {ID: "e-2", Title: "nine doors stayed honest", CreatedAt: base.Add(24 * time.Hour)}, + {ID: "e-3", Title: "the reviewing side", CreatedAt: base.Add(48 * time.Hour)}, + } +} + +// NewR1 returns the collection-only adapter. +// +// A consumer wanting the newest entry must fetch the whole collection, sort it +// and discard the rest. That is the interface pressure the fixture generates. +func NewR1() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/v1/entries", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, Data()) + }) + return mux +} + +// NewR2 returns the adapter with the convenience resource added. +func NewR2() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/v1/entries", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, Data()) + }) + mux.HandleFunc("/v1/entries/latest", func(w http.ResponseWriter, r *http.Request) { + entries := Data() + sort.Slice(entries, func(i, j int) bool { + return entries[i].CreatedAt.After(entries[j].CreatedAt) + }) + writeJSON(w, entries[0]) + }) + return mux +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} diff --git a/examples/echo-interface/r1.openapi.yaml b/examples/echo-interface/r1.openapi.yaml new file mode 100644 index 0000000..19468d8 --- /dev/null +++ b/examples/echo-interface/r1.openapi.yaml @@ -0,0 +1,12 @@ +openapi: "3.1.0" +info: + title: echo-interface + version: "1" +paths: + /v1/entries: + get: + operationId: listEntries + parameters: + - name: limit + in: query + schema: {type: integer, minimum: 1, maximum: 100} diff --git a/examples/echo-interface/r2.openapi.yaml b/examples/echo-interface/r2.openapi.yaml new file mode 100644 index 0000000..254ea24 --- /dev/null +++ b/examples/echo-interface/r2.openapi.yaml @@ -0,0 +1,15 @@ +openapi: "3.1.0" +info: + title: echo-interface + version: "2" +paths: + /v1/entries: + get: + operationId: listEntries + parameters: + - name: limit + in: query + schema: {type: integer, minimum: 1, maximum: 100} + /v1/entries/latest: + get: + operationId: latestEntry diff --git a/internal/runtime/telemetry.go b/internal/runtime/telemetry.go index 974de8f..ca0ede5 100644 --- a/internal/runtime/telemetry.go +++ b/internal/runtime/telemetry.go @@ -29,6 +29,7 @@ type Emitter struct { dropped atomic.Int64 written atomic.Int64 failed atomic.Int64 + inflight atomic.Int64 stopOnce sync.Once done chan struct{} wg sync.WaitGroup @@ -76,6 +77,7 @@ func NewEmitter(sink Sink, opts EmitterOptions) *Emitter { func (e *Emitter) Emit(ev contract.FluidTelemetry) { select { case e.ch <- ev: + e.inflight.Add(1) default: e.dropped.Add(1) } @@ -89,25 +91,13 @@ func (e *Emitter) run(timeout time.Duration) { if !ok { return } - ctx, cancel := context.WithTimeout(context.Background(), timeout) - if err := e.sink.Write(ctx, ev); err != nil { - e.failed.Add(1) - } else { - e.written.Add(1) - } - cancel() + e.deliver(ev, timeout) case <-e.done: // Drain what is already buffered, then stop. for { select { case ev := <-e.ch: - ctx, cancel := context.WithTimeout(context.Background(), timeout) - if err := e.sink.Write(ctx, ev); err != nil { - e.failed.Add(1) - } else { - e.written.Add(1) - } - cancel() + e.deliver(ev, timeout) default: return } @@ -116,6 +106,39 @@ func (e *Emitter) run(timeout time.Duration) { } } +// deliver writes one event and settles its in-flight accounting. +func (e *Emitter) deliver(ev contract.FluidTelemetry, timeout time.Duration) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + if err := e.sink.Write(ctx, ev); err != nil { + e.failed.Add(1) + } else { + e.written.Add(1) + } + cancel() + e.inflight.Add(-1) +} + +// Flush waits for queued telemetry to reach the sink without stopping delivery. +// +// It exists for tests and for operational tooling that needs to read back what +// it just emitted. Close would also flush, but closing an emitter that is still +// serving traffic silently drops everything emitted afterwards -- which is the +// kind of bug that makes a later measurement quietly wrong rather than loudly +// broken. +// +// It returns false if the deadline passes with work still outstanding, so a +// caller can tell a slow sink from an empty one. +func (e *Emitter) Flush(timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if e.inflight.Load() == 0 && len(e.ch) == 0 { + return true + } + time.Sleep(time.Millisecond) + } + return e.inflight.Load() == 0 && len(e.ch) == 0 +} + // Close stops delivery after draining the buffer. func (e *Emitter) Close() { e.stopOnce.Do(func() { close(e.done) }) diff --git a/internal/signing/signing.go b/internal/signing/signing.go index fdb548f..6c0300c 100644 --- a/internal/signing/signing.go +++ b/internal/signing/signing.go @@ -9,12 +9,16 @@ package signing import ( "crypto/ed25519" + "crypto/sha256" "encoding/base64" + "encoding/hex" "encoding/json" "errors" "fmt" "sort" "strings" + + "github.com/tegwick/fluid-core/internal/contract" ) // Algorithm is the only signature algorithm FLUID defines. @@ -195,6 +199,12 @@ func (v *Verifier) Verify(document any, sig *Signature) error { return nil } +// Digest content-addresses a raw artifact such as a contract document. +func Digest(raw []byte) contract.Digest { + sum := sha256.Sum256(raw) + return contract.Digest("sha256:" + hex.EncodeToString(sum[:])) +} + // GenerateKey produces a new signing key pair. Intended for development and // tests; production keys should come from the deployment's key management. func GenerateKey(keyID string) (*Signer, ed25519.PublicKey, error) { diff --git a/workplans/FLUID-WP-0007-conformance-and-self-validation.md b/workplans/FLUID-WP-0007-conformance-and-self-validation.md index 7ab1478..c3de5b1 100644 --- a/workplans/FLUID-WP-0007-conformance-and-self-validation.md +++ b/workplans/FLUID-WP-0007-conformance-and-self-validation.md @@ -4,7 +4,7 @@ type: workplan title: "Conformance suite and self-validation" domain: infotech repo: fluid-core -status: active +status: done owner: worsch topic_slug: fluid-core created: "2026-09-04" @@ -25,7 +25,7 @@ here runs in CI with no human steps and no external services. ```task id: FLUID-WP-0007-T01 -status: todo +status: done priority: high state_hub_task_id: "95b8d512-698c-5201-8f8d-18ff0a658f2c" ``` @@ -38,7 +38,7 @@ reproduction of the Blueprint §33 worked example. ```task id: FLUID-WP-0007-T02 -status: todo +status: done priority: high state_hub_task_id: "71724683-78be-568c-bcd4-12df97e53941" ``` @@ -50,7 +50,7 @@ claimed in a README. ```task id: FLUID-WP-0007-T03 -status: todo +status: done priority: high state_hub_task_id: "0ac5c378-038c-5c76-b082-d4868f9bb12b" ``` @@ -63,7 +63,7 @@ until verified) matter most and get dedicated tests. ```task id: FLUID-WP-0007-T04 -status: todo +status: done priority: high state_hub_task_id: "7af0c88f-a00e-5cad-9fb6-2a297ecfd8b3" ``` @@ -76,7 +76,7 @@ configuration each time. ```task id: FLUID-WP-0007-T05 -status: todo +status: done priority: high state_hub_task_id: "45f70116-04f4-5150-9275-2c8023389dc9" ``` @@ -88,7 +88,7 @@ fitness comparison, promotion, complete audit trail. ```task id: FLUID-WP-0007-T06 -status: todo +status: done priority: medium state_hub_task_id: "29ce43f9-17e5-5798-a3f3-051af97224ce" ```