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

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