Add connector, response policy, telemetry emitter and gateway

Completes the request path for FLUID-WP-0003 T01-T03 and T05-T07. The
gateway resolves a revision, routes to the adapter process, and records
what happened, without ever depending on the control plane to serve.

Three behaviours carry tests because the architecture rests on them:
Emit never blocks against a stalled sink (Blueprint 34.2), the gateway
keeps serving after control-plane loss (invariant 2), and backend
internals do not leak into error responses (5.7).

The connector retries only idempotent methods, so a slow adapter cannot
cause a hall-of-helix entry to be published twice, and breakers are
per-revision so a broken candidate does not take the stable revision
down with 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

go.mod

@@ -1,3 +1,21 @@
 module github.com/tegwick/fluid-core
 
 go 1.22.2
+
+require (
+	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	golang.org/x/sys v0.19.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+	modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
+	modernc.org/libc v1.49.3 // indirect
+	modernc.org/mathutil v1.6.0 // indirect
+	modernc.org/memory v1.8.0 // indirect
+	modernc.org/sqlite v1.29.10 // indirect
+	modernc.org/strutil v1.2.0 // indirect
+	modernc.org/token v1.1.0 // indirect
+)

go.sum

@@ -0,0 +1,32 @@
+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
+github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
+golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 h1:5D53IMaUuA5InSeMu9eJtlQXS2NxAhyWQvkKEgXZhHI=
+modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6/go.mod h1:Qz0X07sNOR1jWYCrJMEnbW/X55x206Q7Vt4mz6/wHp4=
+modernc.org/libc v1.49.3 h1:j2MRCRdwJI2ls/sGbeSk0t2bypOG/uvPZUsGQFDulqg=
+modernc.org/libc v1.49.3/go.mod h1:yMZuGkn7pXbKfoT/M35gFJOAEdSKdxL0q64sF7KqCDo=
+modernc.org/mathutil v1.6.0 h1:fRe9+AmYlaej+64JsEEhoWuAYBkOtQiMEU7n/XgfYi4=
+modernc.org/mathutil v1.6.0/go.mod h1:Ui5Q9q1TR2gFm0AQRqQUaBWFLAhQpCwNcuhBOSedWPo=
+modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E=
+modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU=
+modernc.org/sqlite v1.29.10 h1:3u93dz83myFnMilBGCOLbr+HjklS6+5rJLx4q86RDAg=
+modernc.org/sqlite v1.29.10/go.mod h1:ItX2a1OVGgNsFh6Dv60JQvGfJfTPHPVpV6DF59akYOA=
+modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA=
+modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0=
+modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
+modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=

internal/runtime/connector.go

@@ -0,0 +1,248 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Connector calls the adapter process serving a revision.
+//
+// ArchitectureBlueprint.md section 5.6 puts protocol and dependency detail
+// behind this boundary. Under out-of-process attachment (ADR-0002) the adapter
+// is reached over HTTP, which is what lets it be written in any language.
+type Connector struct {
+	client   *http.Client
+	breakers sync.Map // contract.RevisionID -> *breaker
+	now      func() time.Time
+}
+
+// NewConnector returns a connector with sane transport defaults.
+func NewConnector() *Connector {
+	return &Connector{
+		client: &http.Client{
+			// No client-level timeout: per-revision timeouts come from the
+			// descriptor and are applied through the request context, so one
+			// slow revision cannot impose its budget on another.
+			Transport: &http.Transport{
+				MaxIdleConns:          100,
+				MaxIdleConnsPerHost:   16,
+				IdleConnTimeout:       90 * time.Second,
+				TLSHandshakeTimeout:   5 * time.Second,
+				ExpectContinueTimeout: time.Second,
+			},
+		},
+		now: time.Now,
+	}
+}
+
+// UpstreamError describes a failed adapter call, classified for the response
+// policy and for telemetry.
+type UpstreamError struct {
+	Kind ErrorKind
+	Err  error
+}
+
+func (e *UpstreamError) Error() string { return fmt.Sprintf("%s: %v", e.Kind, e.Err) }
+func (e *UpstreamError) Unwrap() error { return e.Err }
+
+// ErrCircuitOpen is returned while a revision's breaker is open.
+var ErrCircuitOpen = errors.New("circuit open")
+
+// Call forwards a request to the adapter serving rev.
+func (c *Connector) Call(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader) (*http.Response, error) {
+	b := c.breakerFor(rev)
+	if !b.allow(c.now()) {
+		return nil, &UpstreamError{Kind: ErrorUnavailable, Err: ErrCircuitOpen}
+	}
+
+	timeout := 5 * time.Second
+	if rev.Runtime.TimeoutMS != nil && *rev.Runtime.TimeoutMS > 0 {
+		timeout = time.Duration(*rev.Runtime.TimeoutMS) * time.Millisecond
+	}
+
+	attempts := 1
+	var backoff time.Duration
+	if rev.Runtime.Retry != nil {
+		if rev.Runtime.Retry.MaxAttempts != nil && *rev.Runtime.Retry.MaxAttempts > 1 {
+			attempts = int(*rev.Runtime.Retry.MaxAttempts)
+		}
+		if rev.Runtime.Retry.BackoffMS != nil {
+			backoff = time.Duration(*rev.Runtime.Retry.BackoffMS) * time.Millisecond
+		}
+	}
+
+	// Only idempotent methods are retried. Replaying a POST because an adapter
+	// was slow would publish the same hall-of-helix entry twice.
+	if !idempotent(r.Method) {
+		attempts = 1
+	}
+
+	var lastErr error
+	for attempt := 0; attempt < attempts; attempt++ {
+		if attempt > 0 {
+			select {
+			case <-ctx.Done():
+				return nil, &UpstreamError{Kind: ErrorTimeout, Err: ctx.Err()}
+			case <-time.After(backoff):
+			}
+		}
+
+		resp, err := c.attempt(ctx, rev, r, body, timeout)
+		if err == nil {
+			b.record(true, c.now(), rev)
+			return resp, nil
+		}
+		lastErr = err
+
+		// A retry needs a rewindable body; without one, one attempt is all
+		// there is.
+		if body != nil {
+			if seeker, ok := body.(io.Seeker); ok {
+				if _, serr := seeker.Seek(0, io.SeekStart); serr != nil {
+					break
+				}
+			} else {
+				break
+			}
+		}
+	}
+
+	b.record(false, c.now(), rev)
+	return nil, lastErr
+}
+
+func (c *Connector) attempt(ctx context.Context, rev contract.Revision, r *http.Request, body io.Reader, timeout time.Duration) (*http.Response, error) {
+	callCtx, cancel := context.WithTimeout(ctx, timeout)
+
+	target := strings.TrimSuffix(rev.Runtime.Upstream, "/") + r.URL.RequestURI()
+	req, err := http.NewRequestWithContext(callCtx, r.Method, target, body)
+	if err != nil {
+		cancel()
+		return nil, &UpstreamError{Kind: ErrorUnavailable, Err: err}
+	}
+
+	// Forward consumer headers, minus hop-by-hop ones.
+	for k, vs := range r.Header {
+		if hopByHop[strings.ToLower(k)] {
+			continue
+		}
+		for _, v := range vs {
+			req.Header.Add(k, v)
+		}
+	}
+	req.Header.Set("X-FLUID-Revision", string(rev.ID))
+	req.Header.Set("X-FLUID-Interface", string(rev.Interface))
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		cancel()
+		kind := ErrorUnavailable
+		var netErr net.Error
+		if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) {
+			kind = ErrorTimeout
+		}
+		return nil, &UpstreamError{Kind: kind, Err: err}
+	}
+
+	// The body is still streaming, so cancellation has to outlive this call.
+	resp.Body = &cancelOnClose{ReadCloser: resp.Body, cancel: cancel}
+	return resp, nil
+}
+
+func idempotent(method string) bool {
+	switch method {
+	case http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodPut, http.MethodDelete:
+		return true
+	}
+	return false
+}
+
+var hopByHop = map[string]bool{
+	"connection": true, "keep-alive": true, "proxy-authenticate": true,
+	"proxy-authorization": true, "te": true, "trailer": true,
+	"transfer-encoding": true, "upgrade": true,
+}
+
+type cancelOnClose struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+	once   sync.Once
+}
+
+func (c *cancelOnClose) Close() error {
+	err := c.ReadCloser.Close()
+	c.once.Do(c.cancel)
+	return err
+}
+
+// breaker is a per-revision circuit breaker.
+//
+// It is per-revision rather than per-host because two revisions frequently
+// share a host: an adapter that is broken at R-3 should not take R-2 down
+// with it.
+type breaker struct {
+	mu           sync.Mutex
+	failures     int64
+	threshold    int64
+	openUntil    time.Time
+	resetAfter   time.Duration
+	probeAllowed bool
+}
+
+func (c *Connector) breakerFor(rev contract.Revision) *breaker {
+	if v, ok := c.breakers.Load(rev.ID); ok {
+		return v.(*breaker)
+	}
+	b := &breaker{threshold: 5, resetAfter: 30 * time.Second}
+	if cb := rev.Runtime.CircuitBreaker; cb != nil {
+		if cb.FailureThreshold != nil && *cb.FailureThreshold > 0 {
+			b.threshold = *cb.FailureThreshold
+		}
+		if cb.ResetAfterMS != nil && *cb.ResetAfterMS > 0 {
+			b.resetAfter = time.Duration(*cb.ResetAfterMS) * time.Millisecond
+		}
+	}
+	actual, _ := c.breakers.LoadOrStore(rev.ID, b)
+	return actual.(*breaker)
+}
+
+func (b *breaker) allow(now time.Time) bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.openUntil.IsZero() || now.After(b.openUntil) {
+		if !b.openUntil.IsZero() {
+			// Half-open: let exactly one request through to test recovery.
+			if b.probeAllowed {
+				return false
+			}
+			b.probeAllowed = true
+		}
+		return true
+	}
+	return false
+}
+
+func (b *breaker) record(success bool, now time.Time, rev contract.Revision) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if success {
+		b.failures = 0
+		b.openUntil = time.Time{}
+		b.probeAllowed = false
+		return
+	}
+	b.failures++
+	if b.failures >= b.threshold {
+		b.openUntil = now.Add(b.resetAfter)
+		b.probeAllowed = false
+	}
+}

internal/runtime/gateway.go

@@ -0,0 +1,284 @@
+package runtime
+
+import (
+	"bytes"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// CohortResolver assigns an inbound request to a consumer cohort.
+//
+// Cohorts should be coarse (FluidAPIStandards.md section 14): granular enough
+// to compare populations, never more specific than the analysis requires.
+type CohortResolver interface {
+	Cohort(*http.Request) (contract.CohortID, string)
+}
+
+// StaticCohort assigns everything to one cohort. Useful before cohort analysis
+// exists, and for interfaces with a single kind of consumer.
+type StaticCohort contract.CohortID
+
+// Cohort implements CohortResolver.
+func (s StaticCohort) Cohort(r *http.Request) (contract.CohortID, string) {
+	return contract.CohortID(s), consumerRef(r)
+}
+
+// HeaderCohort reads the cohort from a request header, falling back to a
+// default when absent or unrecognized.
+type HeaderCohort struct {
+	Header  string
+	Known   map[string]contract.CohortID
+	Default contract.CohortID
+}
+
+// Cohort implements CohortResolver.
+func (h HeaderCohort) Cohort(r *http.Request) (contract.CohortID, string) {
+	if v := r.Header.Get(h.Header); v != "" {
+		if c, ok := h.Known[v]; ok {
+			return c, consumerRef(r)
+		}
+	}
+	return h.Default, consumerRef(r)
+}
+
+// consumerRef extracts a stable, pseudonymous consumer identity.
+//
+// It must never be a raw end-user identifier: the telemetry envelope carries
+// this value into the evidence store, and Blueprint section 6.2 requires
+// pseudonymization there.
+func consumerRef(r *http.Request) string {
+	if v := r.Header.Get("X-FLUID-Consumer"); v != "" {
+		return v
+	}
+	return ""
+}
+
+// GatewayOptions configures the data plane.
+type GatewayOptions struct {
+	Interface contract.InterfaceID
+	Registry  *Registry
+	Resolver  *Resolver
+	Connector *Connector
+	Emitter   *Emitter
+	Cohorts   CohortResolver
+	Response  ResponsePolicy
+
+	// MaxBodyBytes bounds request size. Zero applies a 1 MiB default.
+	MaxBodyBytes int64
+	// Validator, when set, checks requests against the revision contract.
+	Validator ContractValidator
+}
+
+// ContractValidator checks a request against a revision's declared contract.
+//
+// It is an interface rather than a concrete OpenAPI implementation because
+// FluidAPIStandards.md section 4 admits several contract forms, and the gateway
+// should not know which one an interface chose.
+type ContractValidator interface {
+	Validate(rev contract.Revision, r *http.Request, body []byte) error
+}
+
+// ValidationError reports a contract violation.
+type ValidationError struct {
+	Field   string
+	Message string
+}
+
+func (e *ValidationError) Error() string { return e.Message }
+
+// Gateway is the deterministic entry point for interface traffic.
+//
+// It terminates transport, assigns correlation, resolves and routes a revision,
+// calls the adapter, and emits telemetry. It does not interpret semantics:
+// Blueprint section 5.1 forbids the gateway from inventing them, and section
+// 48.1 names an LLM in the request path as an anti-pattern.
+type Gateway struct {
+	opts GatewayOptions
+	now  func() time.Time
+}
+
+// NewGateway returns a gateway. Registry, Resolver and Connector are required.
+func NewGateway(opts GatewayOptions) (*Gateway, error) {
+	if opts.Registry == nil || opts.Resolver == nil || opts.Connector == nil {
+		return nil, errors.New("gateway requires a registry, resolver and connector")
+	}
+	if opts.Cohorts == nil {
+		opts.Cohorts = StaticCohort("unclassified")
+	}
+	if opts.MaxBodyBytes <= 0 {
+		opts.MaxBodyBytes = 1 << 20
+	}
+	return &Gateway{opts: opts, now: time.Now}, nil
+}
+
+func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	started := g.now()
+	correlation := correlationID(r)
+	w.Header().Set("X-FLUID-Correlation", correlation)
+
+	cohort, consumer := g.opts.Cohorts.Cohort(r)
+
+	body, err := io.ReadAll(io.LimitReader(r.Body, g.opts.MaxBodyBytes+1))
+	if err != nil {
+		g.fail(w, r, correlation, cohort, "", ErrorValidation, "could not read request body", "", started)
+		return
+	}
+	if int64(len(body)) > g.opts.MaxBodyBytes {
+		g.fail(w, r, correlation, cohort, "", ErrorValidation, "request body exceeds the configured limit", "", started)
+		return
+	}
+
+	req := Request{
+		ExplicitRevision: contract.RevisionID(r.Header.Get("X-FLUID-Revision")),
+		BoundRevision:    contract.RevisionID(r.Header.Get("X-FLUID-Bound-Revision")),
+		Cohort:           cohort,
+		Tenant:           r.Header.Get("X-FLUID-Tenant"),
+		ConsumerRef:      consumer,
+		CorrelationID:    correlation,
+	}
+
+	resolution, err := g.opts.Resolver.Resolve(req)
+	if err != nil {
+		kind := ErrorUnavailable
+		message := "no revision is currently able to serve this request"
+		if errors.Is(err, ErrRevisionNotRoutable) || errors.Is(err, ErrUnknownRevision) {
+			kind = ErrorValidation
+			message = "the requested revision is not available"
+		}
+		g.fail(w, r, correlation, cohort, "", kind, message, "", started)
+		return
+	}
+
+	rev, err := g.opts.Registry.Revision(resolution.Revision)
+	if err != nil {
+		g.fail(w, r, correlation, cohort, resolution.Revision, ErrorUnavailable,
+			"the resolved revision is not published", "", started)
+		return
+	}
+
+	if g.opts.Validator != nil {
+		if verr := g.opts.Validator.Validate(rev, r, body); verr != nil {
+			field := ""
+			var ve *ValidationError
+			if errors.As(verr, &ve) {
+				field = ve.Field
+			}
+			g.failWith(w, r, correlation, cohort, rev.ID, ErrorValidation, verr.Error(), field, started, &resolution)
+			return
+		}
+	}
+
+	w.Header().Set("X-FLUID-Revision", string(rev.ID))
+
+	resp, err := g.opts.Connector.Call(r.Context(), rev, r, bytes.NewReader(body))
+	if err != nil {
+		kind := ErrorUnavailable
+		var ue *UpstreamError
+		if errors.As(err, &ue) {
+			kind = ue.Kind
+		}
+		g.failWith(w, r, correlation, cohort, rev.ID, kind,
+			"the interface could not complete this request", "", started, &resolution)
+		return
+	}
+	defer resp.Body.Close()
+
+	for k, vs := range resp.Header {
+		if hopByHop[strings.ToLower(k)] {
+			continue
+		}
+		for _, v := range vs {
+			w.Header().Add(k, v)
+		}
+	}
+	w.WriteHeader(resp.StatusCode)
+	written, _ := io.Copy(w, resp.Body)
+
+	g.emitRequest(r, correlation, cohort, rev.ID, &resolution, resp.StatusCode, written, int64(len(body)), started, nil)
+}
+
+func (g *Gateway) fail(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time) {
+	g.failWith(w, r, correlation, cohort, rev, kind, msg, field, started, nil)
+}
+
+func (g *Gateway) failWith(w http.ResponseWriter, r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, kind ErrorKind, msg, field string, started time.Time, res *Resolution) {
+	g.opts.Response.WriteError(w, kind, correlation, msg, rev, field)
+	status := statusFor[kind]
+	if status == 0 {
+		status = http.StatusInternalServerError
+	}
+	g.emitRequest(r, correlation, cohort, rev, res, status, 0, 0, started, &kind)
+}
+
+// emitRequest records what happened. Errors are evidence, not noise:
+// FluidAPIStandards.md principle 3 treats them as product signals.
+func (g *Gateway) emitRequest(r *http.Request, correlation string, cohort contract.CohortID, rev contract.RevisionID, res *Resolution, status int, respBytes, reqBytes int64, started time.Time, errKind *ErrorKind) {
+	if g.opts.Emitter == nil {
+		return
+	}
+
+	latency := float64(g.now().Sub(started).Microseconds()) / 1000.0
+	statusCode := int64(status)
+	ev := contract.FluidTelemetry{
+		SchemaVersion: "0.1",
+		ID:            newID("tl-"),
+		OccurredAt:    g.now().UTC(),
+		InterfaceID:   g.opts.Interface,
+		Kind:          contract.FluidTelemetryKindRequest,
+		CorrelationID: correlation,
+		ConsumerRef:   r.Header.Get("X-FLUID-Consumer"),
+		Cohort:        &cohort,
+		Revision:      &rev,
+		Request: &contract.FluidTelemetryRequest{
+			Route:         r.URL.Path,
+			Method:        r.Method,
+			Status:        &statusCode,
+			LatencyMS:     &latency,
+			RequestBytes:  &reqBytes,
+			ResponseBytes: &respBytes,
+		},
+		Redaction: &contract.FluidTelemetryRedaction{Applied: false},
+	}
+
+	if res != nil {
+		ev.Resolution = &contract.FluidTelemetryResolution{
+			Reason:           res.Reason,
+			PolicyGeneration: &res.PolicyGeneration,
+		}
+		ev.Experiment = res.Experiment
+	}
+
+	if errKind != nil {
+		ev.Kind = contract.FluidTelemetryKindError
+		class := contract.FluidTelemetryErrorClass(*errKind)
+		if class.Valid() {
+			ev.Error = &contract.FluidTelemetryError{Class: class}
+		}
+	}
+
+	g.opts.Emitter.Emit(ev)
+}
+
+// correlationID reuses an inbound correlation reference when the consumer
+// supplied one, so a call chain stays linked across services.
+func correlationID(r *http.Request) string {
+	if v := r.Header.Get("X-FLUID-Correlation"); v != "" {
+		return v
+	}
+	return newID("c-")
+}
+
+func newID(prefix string) string {
+	var b [12]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return prefix + "0"
+	}
+	return prefix + hex.EncodeToString(b[:])
+}

internal/runtime/gateway_test.go

@@ -0,0 +1,215 @@
+package runtime
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// recordingSink keeps every event for assertions.
+type recordingSink struct {
+	mu     sync.Mutex
+	events []contract.FluidTelemetry
+}
+
+func (s *recordingSink) Write(_ context.Context, ev contract.FluidTelemetry) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.events = append(s.events, ev)
+	return nil
+}
+
+func (s *recordingSink) all() []contract.FluidTelemetry {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]contract.FluidTelemetry, len(s.events))
+	copy(out, s.events)
+	return out
+}
+
+// harness wires a gateway in front of a stub adapter.
+type harness struct {
+	gateway *Gateway
+	sink    *recordingSink
+	emitter *Emitter
+	reg     *Registry
+	adapter *httptest.Server
+}
+
+func newHarness(t *testing.T, handler http.HandlerFunc) *harness {
+	t.Helper()
+
+	adapter := httptest.NewServer(handler)
+	t.Cleanup(adapter.Close)
+
+	reg := NewRegistry(testInterface)
+	d := descriptor("R-1", contract.RevisionStateStable)
+	d.Runtime.Upstream = adapter.URL
+	if err := reg.PutRevision(d); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+
+	sink := &recordingSink{}
+	emitter := NewEmitter(sink, EmitterOptions{Buffer: 64, Workers: 1})
+	t.Cleanup(emitter.Close)
+
+	gw, err := NewGateway(GatewayOptions{
+		Interface: testInterface,
+		Registry:  reg,
+		Resolver:  NewResolver(reg, true),
+		Connector: NewConnector(),
+		Emitter:   emitter,
+		Cohorts:   StaticCohort("publishing-jobs"),
+		Response:  ResponsePolicy{FeedbackPath: "/v1/feedback"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return &harness{gateway: gw, sink: sink, emitter: emitter, reg: reg, adapter: adapter}
+}
+
+func TestGatewayForwardsAndRecords(t *testing.T) {
+	h := newHarness(t, func(w http.ResponseWriter, r *http.Request) {
+		if got := r.Header.Get("X-FLUID-Revision"); got != "R-1" {
+			t.Errorf("adapter saw revision %q, want R-1", got)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_, _ = w.Write([]byte(`{"published":true}`))
+	})
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/v1/hall-entries", strings.NewReader(`{"id":"e-1"}`))
+	req.Header.Set("X-FLUID-Consumer", "hall-publisher")
+	h.gateway.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("status = %d, want 201; body %s", rec.Code, rec.Body.String())
+	}
+	if rec.Header().Get("X-FLUID-Revision") != "R-1" {
+		t.Error("response does not name the revision that served it")
+	}
+	if rec.Header().Get("X-FLUID-Correlation") == "" {
+		t.Error("response carries no correlation reference")
+	}
+
+	h.emitter.Close()
+	events := h.sink.all()
+	if len(events) != 1 {
+		t.Fatalf("emitted %d events, want 1", len(events))
+	}
+	ev := events[0]
+	if ev.Kind != contract.FluidTelemetryKindRequest {
+		t.Errorf("kind = %s", ev.Kind)
+	}
+	if ev.Resolution == nil || ev.Resolution.Reason != contract.FluidTelemetryResolutionReasonStableDefault {
+		t.Errorf("resolution reason not recorded: %+v", ev.Resolution)
+	}
+	if ev.Revision == nil || *ev.Revision != "R-1" {
+		t.Error("revision not recorded on the event")
+	}
+	if ev.Request == nil || ev.Request.Route != "/v1/hall-entries" {
+		t.Errorf("request detail missing: %+v", ev.Request)
+	}
+}
+
+func TestGatewaySurfacesBackendFailureWithoutLeaking(t *testing.T) {
+	h := newHarness(t, func(w http.ResponseWriter, r *http.Request) {
+		panic("adapter exploded: dsn=postgres://user:hunter2@db.internal/prod")
+	})
+	// The stub's panic is handled by httptest's server, which closes the
+	// connection; the connector sees a transport failure.
+	h.adapter.Config.ErrorLog = nil
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries/e-1", nil)
+	h.gateway.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusBadGateway && rec.Code != http.StatusGatewayTimeout {
+		t.Fatalf("status = %d, want a gateway error", rec.Code)
+	}
+
+	var body ErrorBody
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("error body is not JSON: %v", err)
+	}
+	if strings.Contains(rec.Body.String(), "hunter2") || strings.Contains(rec.Body.String(), "postgres://") {
+		t.Error("backend internals leaked into the error response")
+	}
+	if body.Correlation == "" {
+		t.Error("error response carries no correlation reference")
+	}
+	if body.Feedback != "/v1/feedback" {
+		t.Error("error response does not point at the feedback endpoint")
+	}
+
+	h.emitter.Close()
+	events := h.sink.all()
+	if len(events) != 1 || events[0].Kind != contract.FluidTelemetryKindError {
+		t.Fatalf("failure was not recorded as error telemetry: %+v", events)
+	}
+}
+
+// TestGatewayServesWithoutControlPlane is the Blueprint invariant 2 check:
+// evolution can stop without stopping the API. Here the control plane is
+// represented by the registry's ability to accept updates; the gateway must
+// keep serving from what it already holds.
+func TestGatewayServesWithoutControlPlane(t *testing.T) {
+	h := newHarness(t, func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`ok`))
+	})
+
+	// Simulate control-plane loss: no further policies or descriptors arrive,
+	// and telemetry delivery is dead.
+	h.emitter.Close()
+
+	for i := 0; i < 5; i++ {
+		rec := httptest.NewRecorder()
+		h.gateway.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("request %d failed with %d after control-plane loss", i, rec.Code)
+		}
+	}
+}
+
+func TestGatewayRejectsOversizedBody(t *testing.T) {
+	h := newHarness(t, func(w http.ResponseWriter, r *http.Request) {
+		t.Error("adapter should not have been reached")
+	})
+	h.gateway.opts.MaxBodyBytes = 16
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodPost, "/v1/hall-entries", strings.NewReader(strings.Repeat("x", 64)))
+	h.gateway.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusBadRequest {
+		t.Errorf("status = %d, want 400", rec.Code)
+	}
+}
+
+func TestGatewayPropagatesCorrelation(t *testing.T) {
+	h := newHarness(t, func(w http.ResponseWriter, r *http.Request) {
+		if got := r.Header.Get("X-FLUID-Correlation"); got != "c-upstream" {
+			t.Errorf("adapter saw correlation %q, want it forwarded", got)
+		}
+	})
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/v1/hall-entries", nil)
+	req.Header.Set("X-FLUID-Correlation", "c-upstream")
+	h.gateway.ServeHTTP(rec, req)
+
+	if rec.Header().Get("X-FLUID-Correlation") != "c-upstream" {
+		t.Error("inbound correlation was not reused")
+	}
+}

internal/runtime/registry.go

@@ -0,0 +1,170 @@
+// Package runtime implements the FLUID deterministic data plane: the gateway,
+// revision resolution, routing, backend connection and telemetry emission.
+//
+// Nothing here may depend on the evolution control plane to serve a request.
+// ArchitectureBlueprint.md section 2 makes this an invariant: the interface
+// runtime must continue to function when the Daimon, the model provider, the
+// hypothesis store, the experiment controller and the AI budget are all
+// unavailable.
+package runtime
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Registry is the gateway's cached view of published control-plane state.
+//
+// It is deliberately a snapshot rather than a client. When the control plane
+// dies the registry keeps answering from what it last held, which is what
+// ArchitectureBlueprint.md section 34.6 requires: the runtime continues using
+// cached published configuration, and no new promotions occur until
+// control-plane consistency is restored.
+type Registry struct {
+	mu        sync.RWMutex
+	iface     contract.InterfaceID
+	revisions map[contract.RevisionID]contract.Revision
+	policy    contract.RoutingPolicy
+	hasPolicy bool
+}
+
+// NewRegistry returns an empty registry for one interface.
+func NewRegistry(iface contract.InterfaceID) *Registry {
+	return &Registry{
+		iface:     iface,
+		revisions: make(map[contract.RevisionID]contract.Revision),
+	}
+}
+
+var (
+	// ErrUnknownRevision is returned for a revision the registry has never seen.
+	ErrUnknownRevision = errors.New("unknown revision")
+	// ErrRevisionNotRoutable is returned for a revision that exists but must not
+	// receive traffic.
+	ErrRevisionNotRoutable = errors.New("revision not routable")
+	// ErrNoPolicy is returned before any routing policy has been loaded.
+	ErrNoPolicy = errors.New("no routing policy loaded")
+	// ErrWrongInterface guards against loading another interface's artifacts.
+	ErrWrongInterface = errors.New("artifact belongs to a different interface")
+	// ErrStalePolicy is returned when an older policy generation is offered.
+	ErrStalePolicy = errors.New("routing policy generation is not newer")
+)
+
+// PutRevision publishes a revision descriptor into the registry.
+//
+// The descriptor is expected to have been signature-verified already; this
+// method enforces only the structural conditions the router depends on.
+func (r *Registry) PutRevision(d contract.Revision) error {
+	if d.Interface != r.iface {
+		return fmt.Errorf("%w: descriptor is for %q, registry serves %q",
+			ErrWrongInterface, d.Interface, r.iface)
+	}
+	if d.Runtime.Upstream == "" {
+		return fmt.Errorf("revision %s: descriptor has no runtime upstream", d.ID)
+	}
+	if !d.State.Valid() {
+		return fmt.Errorf("revision %s: unknown state %q", d.ID, d.State)
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.revisions[d.ID] = d
+	return nil
+}
+
+// Revision returns a published descriptor.
+func (r *Registry) Revision(id contract.RevisionID) (contract.Revision, error) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	d, ok := r.revisions[id]
+	if !ok {
+		return contract.Revision{}, fmt.Errorf("%w: %s", ErrUnknownRevision, id)
+	}
+	return d, nil
+}
+
+// PutPolicy installs a routing policy.
+//
+// Generations are monotonic: an older policy is refused rather than applied.
+// Without this a delayed delivery could silently roll traffic back to a
+// superseded allocation, which would corrupt an in-flight experiment's
+// measurement window.
+func (r *Registry) PutPolicy(p contract.RoutingPolicy) error {
+	if p.Interface != r.iface {
+		return fmt.Errorf("%w: policy is for %q, registry serves %q",
+			ErrWrongInterface, p.Interface, r.iface)
+	}
+
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.hasPolicy && p.Generation <= r.policy.Generation {
+		return fmt.Errorf("%w: offered %d, holding %d",
+			ErrStalePolicy, p.Generation, r.policy.Generation)
+	}
+	r.policy = p
+	r.hasPolicy = true
+	return nil
+}
+
+// Policy returns the current routing policy.
+func (r *Registry) Policy() (contract.RoutingPolicy, error) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	if !r.hasPolicy {
+		return contract.RoutingPolicy{}, ErrNoPolicy
+	}
+	return r.policy, nil
+}
+
+// routableStates lists the descriptor states the router may send traffic to.
+//
+// ArchitectureBlueprint.md section 5.3 requires rejecting unpublished, failed
+// and retired revisions. "created" is unpublished; "verified" has passed tests
+// but has not been exposed; "retired" is finished.
+var routableStates = map[contract.RevisionState]bool{
+	contract.RevisionStateExperiment: true,
+	contract.RevisionStateCandidate:  true,
+	contract.RevisionStateStable:     true,
+	contract.RevisionStateDeprecated: true,
+}
+
+// CheckRoutable reports whether a revision may currently receive traffic from
+// the given cohort.
+func (r *Registry) CheckRoutable(id contract.RevisionID, cohort contract.CohortID) error {
+	d, err := r.Revision(id)
+	if err != nil {
+		return err
+	}
+
+	if !routableStates[d.State] {
+		return fmt.Errorf("%w: %s is %s", ErrRevisionNotRoutable, id, d.State)
+	}
+	if d.Policy.SecurityCheck != contract.RevisionPolicySecurityCheckPassed {
+		return fmt.Errorf("%w: %s has security_check=%s",
+			ErrRevisionNotRoutable, id, d.Policy.SecurityCheck)
+	}
+	if d.Policy.PolicyCheck != nil && *d.Policy.PolicyCheck == contract.RevisionPolicyPolicyCheckFailed {
+		return fmt.Errorf("%w: %s failed its policy check", ErrRevisionNotRoutable, id)
+	}
+
+	if d.Routing != nil && len(d.Routing.EligibleCohorts) > 0 {
+		if !containsCohort(d.Routing.EligibleCohorts, cohort) {
+			return fmt.Errorf("%w: cohort %q is not eligible for %s",
+				ErrRevisionNotRoutable, cohort, id)
+		}
+	}
+	return nil
+}
+
+func containsCohort(list []contract.CohortID, want contract.CohortID) bool {
+	for _, c := range list {
+		if c == want {
+			return true
+		}
+	}
+	return false
+}

internal/runtime/resolver.go

@@ -0,0 +1,212 @@
+package runtime
+
+import (
+	"fmt"
+	"hash/fnv"
+	"sort"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Request is the subset of an inbound request that revision resolution may
+// consider. Nothing else is allowed to influence the decision: resolution must
+// be a pure function of these fields and the loaded policy, or it stops being
+// auditable (ArchitectureBlueprint.md section 5.2).
+type Request struct {
+	// ExplicitRevision is a revision the consumer asked for by name.
+	ExplicitRevision contract.RevisionID
+	// BoundRevision comes from a client contract binding.
+	BoundRevision contract.RevisionID
+	// Cohort is the consumer's cohort assignment.
+	Cohort contract.CohortID
+	// Tenant identifies the calling tenant, where the interface is multi-tenant.
+	Tenant string
+	// ConsumerRef is the pseudonymous, stable consumer identity used to keep a
+	// long-lived consumer on one side of an experiment.
+	ConsumerRef string
+	// CorrelationID ties this request to its telemetry.
+	CorrelationID string
+}
+
+// Resolution is the outcome of revision resolution, including why.
+//
+// The reason is not decoration. Blueprint section 5.2 requires resolution to be
+// auditable, and "which revision served this request" is unanswerable later
+// without recording how it was chosen.
+type Resolution struct {
+	Revision   contract.RevisionID
+	Reason     contract.FluidTelemetryResolutionReason
+	Experiment *contract.ExperimentID
+	// PolicyGeneration records which policy produced this decision.
+	PolicyGeneration int64
+}
+
+// Resolver implements the deterministic precedence chain.
+type Resolver struct {
+	registry *Registry
+	// allowExplicit controls whether consumers may pin a revision by name. Some
+	// interfaces want this for migration testing; others must not expose it.
+	allowExplicit bool
+}
+
+// NewResolver returns a resolver over reg.
+func NewResolver(reg *Registry, allowExplicit bool) *Resolver {
+	return &Resolver{registry: reg, allowExplicit: allowExplicit}
+}
+
+// Resolve selects the revision that will serve req.
+//
+// The order is fixed by ArchitectureBlueprint.md section 5.2:
+//
+//	explicit revision -> bound client contract -> experiment assignment -> stable default
+//
+// Each step is skipped rather than failed when the candidate is not routable,
+// so a retired pin or an ineligible cohort degrades to the default instead of
+// erroring the request.
+func (r *Resolver) Resolve(req Request) (Resolution, error) {
+	policy, err := r.registry.Policy()
+	if err != nil {
+		return Resolution{}, err
+	}
+
+	if r.allowExplicit && req.ExplicitRevision != "" {
+		if err := r.registry.CheckRoutable(req.ExplicitRevision, req.Cohort); err != nil {
+			// An explicit request for something unroutable is a consumer error
+			// worth surfacing, not something to silently reinterpret.
+			return Resolution{}, fmt.Errorf("explicit revision %s: %w", req.ExplicitRevision, err)
+		}
+		return Resolution{
+			Revision:         req.ExplicitRevision,
+			Reason:           contract.FluidTelemetryResolutionReasonExplicitRevision,
+			PolicyGeneration: policy.Generation,
+		}, nil
+	}
+
+	if req.BoundRevision != "" {
+		if err := r.registry.CheckRoutable(req.BoundRevision, req.Cohort); err == nil {
+			return Resolution{
+				Revision:         req.BoundRevision,
+				Reason:           contract.FluidTelemetryResolutionReasonBoundContract,
+				PolicyGeneration: policy.Generation,
+			}, nil
+		}
+	}
+
+	if rule, ok := matchRule(policy.Rules, req); ok {
+		chosen, ok := allocate(rule, req, policy.DefaultRevision)
+		if ok {
+			if err := r.registry.CheckRoutable(chosen, req.Cohort); err == nil {
+				reason := contract.FluidTelemetryResolutionReasonCohortRule
+				if rule.Experiment != nil {
+					reason = contract.FluidTelemetryResolutionReasonExperimentAssignment
+				}
+				return Resolution{
+					Revision:         chosen,
+					Reason:           reason,
+					Experiment:       rule.Experiment,
+					PolicyGeneration: policy.Generation,
+				}, nil
+			}
+		}
+	}
+
+	if err := r.registry.CheckRoutable(policy.DefaultRevision, req.Cohort); err != nil {
+		return Resolution{}, fmt.Errorf("default revision %s: %w", policy.DefaultRevision, err)
+	}
+	return Resolution{
+		Revision:         policy.DefaultRevision,
+		Reason:           contract.FluidTelemetryResolutionReasonStableDefault,
+		PolicyGeneration: policy.Generation,
+	}, nil
+}
+
+// matchRule returns the first rule matching the request. Rules are evaluated in
+// document order and the first match wins, so policy authors control precedence
+// by ordering rather than by scoring.
+func matchRule(rules []contract.RoutingPolicyRulesItem, req Request) (contract.RoutingPolicyRulesItem, bool) {
+	for _, rule := range rules {
+		if rule.Cohort != nil && *rule.Cohort != req.Cohort {
+			continue
+		}
+		if rule.Tenant != "" && rule.Tenant != req.Tenant {
+			continue
+		}
+		return rule, true
+	}
+	return contract.RoutingPolicyRulesItem{}, false
+}
+
+// allocate picks a revision from a rule's traffic shares.
+//
+// Assignment is a deterministic function of the sticky key, so a given consumer
+// lands on the same side of an experiment for its whole duration. Random
+// per-request assignment would make within-consumer comparisons meaningless and
+// would let a client observe both revisions at once.
+func allocate(rule contract.RoutingPolicyRulesItem, req Request, fallback contract.RevisionID) (contract.RevisionID, bool) {
+	if len(rule.Allocation) == 0 {
+		return "", false
+	}
+
+	// Sorting makes the traversal order independent of Go's map iteration, which
+	// is what turns a hash bucket into a stable assignment.
+	ids := make([]string, 0, len(rule.Allocation))
+	var total float64
+	for id, share := range rule.Allocation {
+		ids = append(ids, id)
+		total += float64(share)
+	}
+	sort.Strings(ids)
+
+	if total <= 0 {
+		return "", false
+	}
+
+	key := stickyKey(rule, req)
+	position := bucket(key) * total
+
+	var cumulative float64
+	for _, id := range ids {
+		cumulative += float64(rule.Allocation[id])
+		if position < cumulative {
+			return contract.RevisionID(id), true
+		}
+	}
+	// Floating-point drift at the top of the range.
+	return contract.RevisionID(ids[len(ids)-1]), true
+}
+
+// stickyKey chooses what keeps a consumer on one side of an experiment.
+func stickyKey(rule contract.RoutingPolicyRulesItem, req Request) string {
+	mode := contract.RoutingPolicyRulesItemStickyByConsumerID
+	if rule.StickyBy != nil {
+		mode = *rule.StickyBy
+	}
+
+	var subject string
+	switch mode {
+	case contract.RoutingPolicyRulesItemStickyByTenant:
+		subject = req.Tenant
+	case contract.RoutingPolicyRulesItemStickyByCorrelationID:
+		subject = req.CorrelationID
+	case contract.RoutingPolicyRulesItemStickyByNone:
+		subject = req.CorrelationID
+	default:
+		subject = req.ConsumerRef
+	}
+
+	// Namespacing by experiment stops one consumer from landing in the same
+	// arm of every concurrent experiment, which would confound their results.
+	if rule.Experiment != nil {
+		return string(*rule.Experiment) + "\x00" + subject
+	}
+	return subject
+}
+
+// bucket maps a key into [0, 1).
+func bucket(key string) float64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte(key))
+	// 53 bits keeps the result exactly representable as a float64.
+	const mask = 1<<53 - 1
+	return float64(h.Sum64()&mask) / float64(mask+1)
+}

internal/runtime/resolver_test.go

@@ -0,0 +1,275 @@
+package runtime
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+const testInterface contract.InterfaceID = "hall-publishing"
+
+// descriptor builds a minimal routable revision descriptor.
+func descriptor(id contract.RevisionID, state contract.RevisionState, cohorts ...contract.CohortID) contract.Revision {
+	d := contract.Revision{
+		SchemaVersion: "0.1",
+		ID:            id,
+		Interface:     testInterface,
+		State:         state,
+		Contract: contract.RevisionContract{
+			Type:   contract.RevisionContractTypeOpenapi,
+			Digest: contract.Digest("sha256:" + zeros(64)),
+		},
+		Runtime: contract.RevisionRuntime{Upstream: "http://adapter:8080"},
+		Intent:  contract.RevisionIntent{Version: "IEI-1"},
+		Policy: contract.RevisionPolicy{
+			Compatibility: contract.RevisionPolicyCompatibilityAdditive,
+			SecurityCheck: contract.RevisionPolicySecurityCheckPassed,
+		},
+	}
+	if len(cohorts) > 0 {
+		d.Routing = &contract.RevisionRouting{EligibleCohorts: cohorts}
+	}
+	return d
+}
+
+func zeros(n int) string {
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = '0'
+	}
+	return string(b)
+}
+
+func policyWith(defaultRev contract.RevisionID, gen int64, rules ...contract.RoutingPolicyRulesItem) contract.RoutingPolicy {
+	return contract.RoutingPolicy{
+		SchemaVersion:   "0.1",
+		Interface:       testInterface,
+		Generation:      gen,
+		DefaultRevision: defaultRev,
+		Rules:           rules,
+	}
+}
+
+func newFixture(t *testing.T, revs ...contract.Revision) *Registry {
+	t.Helper()
+	reg := NewRegistry(testInterface)
+	for _, r := range revs {
+		if err := reg.PutRevision(r); err != nil {
+			t.Fatalf("PutRevision(%s): %v", r.ID, err)
+		}
+	}
+	return reg
+}
+
+func TestResolvePrecedence(t *testing.T) {
+	reg := newFixture(t,
+		descriptor("R-1", contract.RevisionStateStable),
+		descriptor("R-2", contract.RevisionStateExperiment),
+		descriptor("R-3", contract.RevisionStateCandidate),
+	)
+	if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+
+	res := NewResolver(reg, true)
+
+	t.Run("explicit wins", func(t *testing.T) {
+		got, err := res.Resolve(Request{ExplicitRevision: "R-3", BoundRevision: "R-2"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Revision != "R-3" {
+			t.Errorf("got %s, want R-3", got.Revision)
+		}
+		if got.Reason != contract.FluidTelemetryResolutionReasonExplicitRevision {
+			t.Errorf("reason = %s", got.Reason)
+		}
+	})
+
+	t.Run("bound contract beats default", func(t *testing.T) {
+		got, err := res.Resolve(Request{BoundRevision: "R-2"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Revision != "R-2" || got.Reason != contract.FluidTelemetryResolutionReasonBoundContract {
+			t.Errorf("got %s via %s, want R-2 via bound_contract", got.Revision, got.Reason)
+		}
+	})
+
+	t.Run("falls through to stable default", func(t *testing.T) {
+		got, err := res.Resolve(Request{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Revision != "R-1" || got.Reason != contract.FluidTelemetryResolutionReasonStableDefault {
+			t.Errorf("got %s via %s, want R-1 via stable_default", got.Revision, got.Reason)
+		}
+	})
+
+	t.Run("explicit disabled is ignored", func(t *testing.T) {
+		strict := NewResolver(reg, false)
+		got, err := strict.Resolve(Request{ExplicitRevision: "R-3"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Revision != "R-1" {
+			t.Errorf("got %s, want the default R-1 when pinning is disabled", got.Revision)
+		}
+	})
+}
+
+func TestUnroutableRevisionsAreRefused(t *testing.T) {
+	reg := newFixture(t,
+		descriptor("R-1", contract.RevisionStateStable),
+		descriptor("R-created", contract.RevisionStateCreated),
+		descriptor("R-retired", contract.RevisionStateRetired),
+	)
+	failed := descriptor("R-insecure", contract.RevisionStateStable)
+	failed.Policy.SecurityCheck = contract.RevisionPolicySecurityCheckFailed
+	if err := reg.PutRevision(failed); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+	res := NewResolver(reg, true)
+
+	for _, id := range []contract.RevisionID{"R-created", "R-retired", "R-insecure"} {
+		if _, err := res.Resolve(Request{ExplicitRevision: id}); err == nil {
+			t.Errorf("resolving %s should have been refused", id)
+		} else if !errors.Is(err, ErrRevisionNotRoutable) {
+			t.Errorf("resolving %s: got %v, want ErrRevisionNotRoutable", id, err)
+		}
+	}
+}
+
+func TestBoundRevisionDegradesToDefault(t *testing.T) {
+	// A consumer pinned to a revision that has since retired should keep being
+	// served rather than start failing.
+	reg := newFixture(t,
+		descriptor("R-1", contract.RevisionStateStable),
+		descriptor("R-old", contract.RevisionStateRetired),
+	)
+	if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := NewResolver(reg, false).Resolve(Request{BoundRevision: "R-old"})
+	if err != nil {
+		t.Fatalf("a retired binding should degrade, not fail: %v", err)
+	}
+	if got.Revision != "R-1" {
+		t.Errorf("got %s, want R-1", got.Revision)
+	}
+}
+
+func TestCohortEligibility(t *testing.T) {
+	reg := newFixture(t,
+		descriptor("R-1", contract.RevisionStateStable),
+		descriptor("R-2", contract.RevisionStateExperiment, "coding-agents"),
+	)
+	if err := reg.PutPolicy(policyWith("R-1", 1)); err != nil {
+		t.Fatal(err)
+	}
+
+	if err := reg.CheckRoutable("R-2", "coding-agents"); err != nil {
+		t.Errorf("eligible cohort refused: %v", err)
+	}
+	if err := reg.CheckRoutable("R-2", "partner-integrations"); err == nil {
+		t.Error("ineligible cohort accepted")
+	}
+}
+
+func TestAllocationIsStickyAndProportional(t *testing.T) {
+	reg := newFixture(t,
+		descriptor("R-1", contract.RevisionStateStable),
+		descriptor("R-2", contract.RevisionStateExperiment),
+	)
+	exp := contract.ExperimentID("E-1")
+	rule := contract.RoutingPolicyRulesItem{
+		Experiment: &exp,
+		Allocation: map[string]contract.UnitInterval{"R-1": 0.9, "R-2": 0.1},
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 1, rule)); err != nil {
+		t.Fatal(err)
+	}
+	res := NewResolver(reg, false)
+
+	// Stickiness: the same consumer must resolve identically every time.
+	first, err := res.Resolve(Request{ConsumerRef: "consumer-42"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 50; i++ {
+		again, err := res.Resolve(Request{ConsumerRef: "consumer-42"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if again.Revision != first.Revision {
+			t.Fatalf("assignment drifted: %s then %s", first.Revision, again.Revision)
+		}
+	}
+
+	// Proportionality: roughly a tenth of consumers should see the candidate.
+	const n = 4000
+	candidate := 0
+	for i := 0; i < n; i++ {
+		got, err := res.Resolve(Request{ConsumerRef: consumerName(i)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Revision == "R-2" {
+			candidate++
+		}
+		if got.Experiment == nil || *got.Experiment != exp {
+			t.Fatalf("experiment not recorded on resolution for consumer %d", i)
+		}
+	}
+	share := float64(candidate) / n
+	if share < 0.07 || share > 0.13 {
+		t.Errorf("candidate share %.3f, want approximately 0.10", share)
+	}
+}
+
+func consumerName(i int) string {
+	digits := "0123456789"
+	out := []byte("consumer-")
+	if i == 0 {
+		return string(append(out, '0'))
+	}
+	var rev []byte
+	for i > 0 {
+		rev = append(rev, digits[i%10])
+		i /= 10
+	}
+	for j := len(rev) - 1; j >= 0; j-- {
+		out = append(out, rev[j])
+	}
+	return string(out)
+}
+
+func TestStalePolicyRefused(t *testing.T) {
+	reg := newFixture(t, descriptor("R-1", contract.RevisionStateStable))
+	if err := reg.PutPolicy(policyWith("R-1", 5)); err != nil {
+		t.Fatal(err)
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 4)); !errors.Is(err, ErrStalePolicy) {
+		t.Errorf("older generation accepted: %v", err)
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 5)); !errors.Is(err, ErrStalePolicy) {
+		t.Errorf("equal generation accepted: %v", err)
+	}
+	if err := reg.PutPolicy(policyWith("R-1", 6)); err != nil {
+		t.Errorf("newer generation refused: %v", err)
+	}
+}
+
+func TestWrongInterfaceRefused(t *testing.T) {
+	reg := NewRegistry(testInterface)
+	other := descriptor("R-1", contract.RevisionStateStable)
+	other.Interface = "some-other-api"
+	if err := reg.PutRevision(other); !errors.Is(err, ErrWrongInterface) {
+		t.Errorf("foreign descriptor accepted: %v", err)
+	}
+}

internal/runtime/response.go

@@ -0,0 +1,91 @@
+package runtime
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// ErrorKind classifies a response the interface produced rather than the
+// backend. These map onto the telemetry error classes so that a response and
+// the evidence it generates cannot disagree.
+type ErrorKind string
+
+const (
+	ErrorValidation    ErrorKind = "validation"
+	ErrorUnknownPath   ErrorKind = "unknown_path"
+	ErrorUnsupported   ErrorKind = "unsupported_parameter"
+	ErrorAuthorization ErrorKind = "authorization"
+	ErrorUnavailable   ErrorKind = "backend_failure"
+	ErrorTimeout       ErrorKind = "timeout"
+	ErrorPolicy        ErrorKind = "policy_rejection"
+	ErrorNoCapability  ErrorKind = "missing_capability"
+)
+
+// ErrorBody is the interface's error representation.
+//
+// ArchitectureBlueprint.md section 5.7: errors should carry enough structured
+// information to improve observability without leaking backend detail, and may
+// include a correlation reference that lets downstream analysis link the
+// response a consumer saw to the pressure it generated.
+type ErrorBody struct {
+	Kind ErrorKind `json:"kind"`
+	// Message is written for the consumer and must stay free of backend
+	// internals: hostnames, stack traces, driver errors, upstream payloads.
+	Message string `json:"message"`
+	// Correlation is the reference a consumer can quote back, and the key that
+	// ties this response to its telemetry.
+	Correlation string `json:"correlation"`
+	// Revision tells the consumer which contract answered. Without it a client
+	// debugging an unexpected response has no way to know what it was talking to.
+	Revision contract.RevisionID `json:"revision,omitempty"`
+	// Field names the offending input for validation failures.
+	Field string `json:"field,omitempty"`
+	// Feedback points at the explicit-feedback endpoint. Turning a dead end into
+	// an invitation is the cheapest pressure signal the interface can collect.
+	Feedback string `json:"feedback,omitempty"`
+}
+
+// statusFor maps an error kind to its HTTP status.
+var statusFor = map[ErrorKind]int{
+	ErrorValidation:    http.StatusBadRequest,
+	ErrorUnknownPath:   http.StatusNotFound,
+	ErrorUnsupported:   http.StatusBadRequest,
+	ErrorAuthorization: http.StatusForbidden,
+	ErrorUnavailable:   http.StatusBadGateway,
+	ErrorTimeout:       http.StatusGatewayTimeout,
+	ErrorPolicy:        http.StatusForbidden,
+	ErrorNoCapability:  http.StatusNotImplemented,
+}
+
+// ResponsePolicy renders interface errors consistently.
+type ResponsePolicy struct {
+	// FeedbackPath, when set, is advertised on every error.
+	FeedbackPath string
+}
+
+// WriteError renders an error response.
+func (p ResponsePolicy) WriteError(w http.ResponseWriter, kind ErrorKind, correlation, message string, rev contract.RevisionID, field string) {
+	status, ok := statusFor[kind]
+	if !ok {
+		status = http.StatusInternalServerError
+	}
+
+	body := ErrorBody{
+		Kind:        kind,
+		Message:     message,
+		Correlation: correlation,
+		Revision:    rev,
+		Field:       field,
+		Feedback:    p.FeedbackPath,
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("X-FLUID-Correlation", correlation)
+	if rev != "" {
+		w.Header().Set("X-FLUID-Revision", string(rev))
+	}
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(body)
+}

internal/runtime/telemetry.go

@@ -0,0 +1,143 @@
+package runtime
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Sink receives normalized telemetry. Implementations may block; the emitter
+// exists precisely so that blocking never reaches a request.
+type Sink interface {
+	Write(context.Context, contract.FluidTelemetry) error
+}
+
+// Emitter buffers telemetry and delivers it out of band.
+//
+// ArchitectureBlueprint.md section 34.2 is a hard invariant: telemetry
+// backpressure must not block normal API requests. The buffer is therefore
+// bounded and lossy by design. Losing evidence degrades learning; blocking a
+// request degrades the service, and the service is the thing that must not
+// degrade.
+type Emitter struct {
+	sink Sink
+	ch   chan contract.FluidTelemetry
+
+	dropped  atomic.Int64
+	written  atomic.Int64
+	failed   atomic.Int64
+	stopOnce sync.Once
+	done     chan struct{}
+	wg       sync.WaitGroup
+}
+
+// EmitterOptions configures buffering and delivery.
+type EmitterOptions struct {
+	// Buffer is the number of events held in memory. When it fills, new events
+	// are dropped rather than queued.
+	Buffer int
+	// Workers is the number of concurrent deliveries.
+	Workers int
+	// WriteTimeout bounds a single sink write.
+	WriteTimeout time.Duration
+}
+
+// NewEmitter starts an emitter delivering into sink.
+func NewEmitter(sink Sink, opts EmitterOptions) *Emitter {
+	if opts.Buffer <= 0 {
+		opts.Buffer = 4096
+	}
+	if opts.Workers <= 0 {
+		opts.Workers = 2
+	}
+	if opts.WriteTimeout <= 0 {
+		opts.WriteTimeout = 5 * time.Second
+	}
+
+	e := &Emitter{
+		sink: sink,
+		ch:   make(chan contract.FluidTelemetry, opts.Buffer),
+		done: make(chan struct{}),
+	}
+
+	for i := 0; i < opts.Workers; i++ {
+		e.wg.Add(1)
+		go e.run(opts.WriteTimeout)
+	}
+	return e
+}
+
+// Emit queues an event. It never blocks and never returns an error: a caller on
+// the request path has no useful response to a telemetry failure, and giving it
+// one invites handling that blocks.
+func (e *Emitter) Emit(ev contract.FluidTelemetry) {
+	select {
+	case e.ch <- ev:
+	default:
+		e.dropped.Add(1)
+	}
+}
+
+func (e *Emitter) run(timeout time.Duration) {
+	defer e.wg.Done()
+	for {
+		select {
+		case ev, ok := <-e.ch:
+			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()
+		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()
+				default:
+					return
+				}
+			}
+		}
+	}
+}
+
+// Close stops delivery after draining the buffer.
+func (e *Emitter) Close() {
+	e.stopOnce.Do(func() { close(e.done) })
+	e.wg.Wait()
+}
+
+// Stats reports emitter health.
+//
+// FLUID must observe itself (ArchitectureBlueprint.md section 40), and a
+// silently lossy telemetry path would make every downstream pressure count
+// quietly wrong. Dropped events are a first-class operational metric.
+type Stats struct {
+	Written int64
+	Dropped int64
+	Failed  int64
+}
+
+// Stats returns a snapshot of delivery counters.
+func (e *Emitter) Stats() Stats {
+	return Stats{
+		Written: e.written.Load(),
+		Dropped: e.dropped.Load(),
+		Failed:  e.failed.Load(),
+	}
+}

internal/runtime/telemetry_test.go

@@ -0,0 +1,101 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// stallingSink never returns, standing in for a wedged evidence store.
+type stallingSink struct{ entered chan struct{} }
+
+func (s *stallingSink) Write(ctx context.Context, _ contract.FluidTelemetry) error {
+	select {
+	case s.entered <- struct{}{}:
+	default:
+	}
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// TestEmitNeverBlocks is the test for Blueprint invariant 34.2. If it ever
+// starts failing, the data plane has acquired a dependency on the observation
+// plane and the isolation the whole architecture rests on is gone.
+func TestEmitNeverBlocks(t *testing.T) {
+	sink := &stallingSink{entered: make(chan struct{}, 1)}
+	e := NewEmitter(sink, EmitterOptions{Buffer: 8, Workers: 1, WriteTimeout: time.Hour})
+	defer func() {
+		// The stalled worker cannot drain, so do not wait on Close.
+		_ = e
+	}()
+
+	<-func() chan struct{} {
+		e.Emit(contract.FluidTelemetry{ID: "warm"})
+		return sink.entered
+	}()
+
+	done := make(chan struct{})
+	go func() {
+		// Far more than the buffer holds, against a sink that never completes.
+		for i := 0; i < 10_000; i++ {
+			e.Emit(contract.FluidTelemetry{ID: "ev"})
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Emit blocked while the sink was stalled; the request path is now coupled to telemetry")
+	}
+
+	if e.Stats().Dropped == 0 {
+		t.Error("expected drops against a stalled sink, got none")
+	}
+}
+
+// countingSink records deliveries.
+type countingSink struct {
+	ch  chan contract.FluidTelemetry
+	err error
+}
+
+func (s *countingSink) Write(_ context.Context, ev contract.FluidTelemetry) error {
+	if s.err != nil {
+		return s.err
+	}
+	s.ch <- ev
+	return nil
+}
+
+func TestEmitterDeliversAndDrains(t *testing.T) {
+	sink := &countingSink{ch: make(chan contract.FluidTelemetry, 32)}
+	e := NewEmitter(sink, EmitterOptions{Buffer: 32, Workers: 2})
+
+	for i := 0; i < 10; i++ {
+		e.Emit(contract.FluidTelemetry{ID: "ev", Kind: contract.FluidTelemetryKindRequest})
+	}
+	e.Close()
+
+	if got := len(sink.ch); got != 10 {
+		t.Errorf("delivered %d events, want 10", got)
+	}
+	if s := e.Stats(); s.Written != 10 || s.Dropped != 0 {
+		t.Errorf("stats = %+v, want 10 written and 0 dropped", s)
+	}
+}
+
+func TestEmitterCountsSinkFailures(t *testing.T) {
+	e := NewEmitter(&countingSink{err: errors.New("store down")}, EmitterOptions{Buffer: 4, Workers: 1})
+	for i := 0; i < 4; i++ {
+		e.Emit(contract.FluidTelemetry{ID: "ev"})
+	}
+	e.Close()
+
+	if s := e.Stats(); s.Failed == 0 {
+		t.Errorf("sink failures not counted: %+v", s)
+	}
+}

workplans/FLUID-WP-0003-deterministic-data-plane.md

@@ -26,7 +26,7 @@ above it is dead.
 
 ```task
 id: FLUID-WP-0003-T01
-status: todo
+status: done
 priority: high
 state_hub_task_id: "0dac6ab6-f3e4-5234-bb3c-2393558d8590"
 ```
@@ -38,7 +38,7 @@ shape limits. Blueprint §5.1 — the gateway must not invent interface semantic
 
 ```task
 id: FLUID-WP-0003-T02
-status: todo
+status: done
 priority: high
 state_hub_task_id: "6d8333ac-1bad-59a4-943d-b83819c8127a"
 ```
@@ -51,7 +51,7 @@ without an auditable reason is a defect.
 
 ```task
 id: FLUID-WP-0003-T03
-status: todo
+status: done
 priority: high
 state_hub_task_id: "1f042c4f-9e3f-50fa-985c-e4f2ebb3b38a"
 ```
@@ -76,7 +76,7 @@ of the revision artifact and is content-addressed.
 
 ```task
 id: FLUID-WP-0003-T05
-status: todo
+status: done
 priority: high
 state_hub_task_id: "0d62a681-3154-50af-894a-98b35db7b040"
 ```
@@ -89,7 +89,7 @@ language-agnostic promise is kept.
 
 ```task
 id: FLUID-WP-0003-T06
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "cc3c2a03-b24c-51de-b695-8aa9bfae85ab"
 ```
@@ -101,7 +101,7 @@ detail (§5.7).
 
 ```task
 id: FLUID-WP-0003-T07
-status: todo
+status: done
 priority: high
 state_hub_task_id: "e10a5011-cb7e-5f1c-9a0d-0cfd2d535d13"
 ```