Add the conformance suite, echo fixture and integration guide
Some checks failed
ci / build (push) Has been cancelled

Completes FLUID-WP-0007. The seven minimal-conformance requirements and
the mechanically checkable architectural invariants are asserted as
tests rather than claimed in a README, because a conformance claim
nobody re-checks is one that quietly stops being true. Only the
checkable subset of the invariants is asserted; pretending a test can
settle the rest would be worse than leaving them to review.

TestFirstVerticalSlice runs all eleven steps of Blueprint 50 with no
human steps: two revisions, explicit routing, telemetry, a cohort
dimension, detected pressure, a hypothesis, a candidate, a 90/10
experiment, fitness comparison, promotion, and a complete audit trail.
Requests per completed task fall from 5.65 to 1.00 against a 1.20
target. A companion test runs the loop twice and requires the same
verdict, since a loop whose conclusion depended on run order would be
measuring the harness rather than the interface.

The failure-containment matrix covers Blueprint 34 directly: the data
plane keeps serving with the evidence store closed, with telemetry
wedged against a sink that never returns, after a failed build, after an
experiment rollback, and with the adaptive concurrency limit saturated.

Fixes a real bug the suite exposed. Drain closed the emitter outright,
so every request after the first flush emitted into a dead emitter and
was silently lost -- the kind of fault that makes a later measurement
quietly wrong rather than loudly broken. Emitter.Flush now waits for
delivery without stopping it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014KmVxhJ35tCo7rE7UnLwWu

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1116572@bnt-lap001
Assistant-Session: 8ba9bb93-a72a-4883-b189-2499cce5c400
This commit is contained in:
tegwick 2026-09-04 08:21:49 +02:00
parent 61d8d8cabe
commit 55363905bc
16 changed files with 1885 additions and 23 deletions

View file

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

View file

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

View file

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

View file

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

View file

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

321
conformance/suite/loop.go Normal file
View file

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

35
conformance/suite/repo.go Normal file
View file

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

View file

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

137
docs/integration-guide.md Normal file
View file

@ -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:<digest of your OpenAPI document>"
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.

View file

@ -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
revisionexperimentfitness loop is proven mechanically before a real workload
depends on it.

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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