Add evidence store, intent store and the fluid CLI

Completes FLUID-WP-0003. The evidence store is append-only at the
database, not by convention in Go: a trigger blocks UPDATE and DELETE on
fluid_events in both SQLite and Postgres, so the guarantee binds a psql
session and the CLI equally, not just callers who go through the Go API.
Records are derived summary state and may be superseded; the event log
stays the authority.

The intent store makes a recorded version immutable and content
addressed, since reassigning what governed a revision after the fact
would break the one audit question Blueprint 27 exists to answer. The
unfilled InterfaceEvolutionIntent template is rejected rather than
defaulted, because a template still listing every FLUID-N mode has not
been completed and defaulting it would pick a permissive authority by
accident.

The CLI reads the evidence store directly rather than through the
control-plane API, so an operator can reconstruct history when the
control plane is down.

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/evidence/sink.go

@@ -0,0 +1,19 @@
+package evidence
+
+import (
+	"context"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// TelemetrySink adapts a Store to the runtime's telemetry sink.
+//
+// It is a separate type so the data plane depends on the narrow sink interface
+// rather than on the whole evidence store: the gateway should not be able to
+// read hypotheses.
+type TelemetrySink struct{ Store Store }
+
+// Write implements the runtime Sink interface.
+func (s TelemetrySink) Write(ctx context.Context, ev contract.FluidTelemetry) error {
+	return s.Store.WriteTelemetry(ctx, ev)
+}

internal/evidence/sqlstore.go

@@ -0,0 +1,399 @@
+package evidence
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// SQLStore is the relational evidence store.
+//
+// One schema serves both backends (ADR-0004): SQLite for development, CI and
+// single-node deployments, PostgreSQL for anything shared. Only portable SQL is
+// used, so the statements CI exercises are the statements production runs.
+type SQLStore struct {
+	db      *sql.DB
+	dialect dialect
+}
+
+type dialect int
+
+const (
+	dialectSQLite dialect = iota
+	dialectPostgres
+)
+
+// schemaStatements builds the DDL for a dialect.
+//
+// The immutability of fluid_events is enforced at the database, not by
+// convention in Go. Blueprint invariant 8 (every promotion is auditable) is
+// only as strong as the weakest writer, and a trigger binds every one of them.
+func schemaStatements(d dialect) []string {
+	stmts := []string{
+		`CREATE TABLE IF NOT EXISTS fluid_events (
+			id           TEXT PRIMARY KEY,
+			occurred_at  TEXT NOT NULL,
+			entity_type  TEXT NOT NULL,
+			entity_id    TEXT NOT NULL,
+			event_type   TEXT NOT NULL,
+			body         TEXT NOT NULL
+		)`,
+		`CREATE INDEX IF NOT EXISTS fluid_events_entity
+			ON fluid_events (entity_type, entity_id, occurred_at)`,
+		`CREATE INDEX IF NOT EXISTS fluid_events_time
+			ON fluid_events (occurred_at)`,
+
+		`CREATE TABLE IF NOT EXISTS fluid_telemetry (
+			id            TEXT PRIMARY KEY,
+			occurred_at   TEXT NOT NULL,
+			interface_id  TEXT NOT NULL,
+			kind          TEXT NOT NULL,
+			revision      TEXT,
+			experiment    TEXT,
+			cohort        TEXT,
+			body          TEXT NOT NULL
+		)`,
+		`CREATE INDEX IF NOT EXISTS fluid_telemetry_window
+			ON fluid_telemetry (interface_id, occurred_at)`,
+		`CREATE INDEX IF NOT EXISTS fluid_telemetry_revision
+			ON fluid_telemetry (revision, occurred_at)`,
+
+		// Records are derived summary state, rebuildable from events.
+		`CREATE TABLE IF NOT EXISTS fluid_records (
+			kind       TEXT NOT NULL,
+			id         TEXT NOT NULL,
+			updated_at TEXT NOT NULL,
+			body       TEXT NOT NULL,
+			PRIMARY KEY (kind, id)
+		)`,
+	}
+
+	switch d {
+	case dialectSQLite:
+		stmts = append(stmts,
+			`CREATE TRIGGER IF NOT EXISTS fluid_events_no_update
+				BEFORE UPDATE ON fluid_events
+				BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`,
+			`CREATE TRIGGER IF NOT EXISTS fluid_events_no_delete
+				BEFORE DELETE ON fluid_events
+				BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`,
+		)
+	case dialectPostgres:
+		stmts = append(stmts,
+			`CREATE OR REPLACE FUNCTION fluid_events_append_only()
+				RETURNS trigger AS $$
+				BEGIN RAISE EXCEPTION 'fluid_events is append-only'; END;
+				$$ LANGUAGE plpgsql`,
+			`DROP TRIGGER IF EXISTS fluid_events_no_change ON fluid_events`,
+			`CREATE TRIGGER fluid_events_no_change
+				BEFORE UPDATE OR DELETE ON fluid_events
+				FOR EACH ROW EXECUTE FUNCTION fluid_events_append_only()`,
+		)
+	}
+	return stmts
+}
+
+// OpenSQLite opens (and migrates) a SQLite-backed store.
+func OpenSQLite(ctx context.Context, path string) (*SQLStore, error) {
+	db, err := sql.Open("sqlite", path)
+	if err != nil {
+		return nil, fmt.Errorf("open sqlite %q: %w", path, err)
+	}
+	// SQLite serializes writers; more than one connection buys contention.
+	db.SetMaxOpenConns(1)
+
+	for _, pragma := range []string{
+		"PRAGMA journal_mode = WAL",
+		"PRAGMA foreign_keys = ON",
+		"PRAGMA busy_timeout = 5000",
+	} {
+		if _, err := db.ExecContext(ctx, pragma); err != nil {
+			_ = db.Close()
+			return nil, fmt.Errorf("%s: %w", pragma, err)
+		}
+	}
+
+	s := &SQLStore{db: db, dialect: dialectSQLite}
+	if err := s.migrate(ctx); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return s, nil
+}
+
+// NewSQLStore wraps an already-open database, for Postgres or for tests.
+func NewSQLStore(ctx context.Context, db *sql.DB, postgres bool) (*SQLStore, error) {
+	d := dialectSQLite
+	if postgres {
+		d = dialectPostgres
+	}
+	s := &SQLStore{db: db, dialect: d}
+	if err := s.migrate(ctx); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *SQLStore) migrate(ctx context.Context) error {
+	for _, stmt := range schemaStatements(s.dialect) {
+		if _, err := s.db.ExecContext(ctx, stmt); err != nil {
+			return fmt.Errorf("migrate: %w\nstatement: %s", err, stmt)
+		}
+	}
+	return nil
+}
+
+// arg renders the nth placeholder for the dialect.
+func (s *SQLStore) arg(n int) string {
+	if s.dialect == dialectPostgres {
+		return fmt.Sprintf("$%d", n)
+	}
+	return "?"
+}
+
+// rfc3339 normalizes timestamps so lexical ordering matches chronological
+// ordering in both backends.
+func rfc3339(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) }
+
+// AppendEvent records a lifecycle transition.
+func (s *SQLStore) AppendEvent(ctx context.Context, ev contract.FluidEvent) error {
+	if ev.ID == "" {
+		return errors.New("event has no id")
+	}
+	body, err := json.Marshal(ev)
+	if err != nil {
+		return fmt.Errorf("marshal event %s: %w", ev.ID, err)
+	}
+
+	q := fmt.Sprintf(
+		`INSERT INTO fluid_events (id, occurred_at, entity_type, entity_id, event_type, body)
+		 VALUES (%s, %s, %s, %s, %s, %s)`,
+		s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6))
+
+	_, err = s.db.ExecContext(ctx, q,
+		string(ev.ID), rfc3339(ev.OccurredAt), string(ev.EntityType),
+		ev.EntityID, ev.EventType, string(body))
+	if err != nil {
+		return fmt.Errorf("append event %s: %w", ev.ID, err)
+	}
+	return nil
+}
+
+// Events returns matching events in occurrence order.
+func (s *SQLStore) Events(ctx context.Context, f EventFilter) ([]contract.FluidEvent, error) {
+	var (
+		where []string
+		args  []any
+	)
+	add := func(clause string, v any) {
+		args = append(args, v)
+		where = append(where, fmt.Sprintf(clause, s.arg(len(args))))
+	}
+
+	if f.EntityType != "" {
+		add("entity_type = %s", string(f.EntityType))
+	}
+	if f.EntityID != "" {
+		add("entity_id = %s", f.EntityID)
+	}
+	if f.EventType != "" {
+		add("event_type = %s", f.EventType)
+	}
+	if !f.Since.IsZero() {
+		add("occurred_at >= %s", rfc3339(f.Since))
+	}
+	if !f.Until.IsZero() {
+		add("occurred_at <= %s", rfc3339(f.Until))
+	}
+
+	q := "SELECT body FROM fluid_events"
+	if len(where) > 0 {
+		q += " WHERE " + strings.Join(where, " AND ")
+	}
+	// id breaks ties so that two events in the same instant still order stably.
+	q += " ORDER BY occurred_at, id"
+	if f.Limit > 0 {
+		q += fmt.Sprintf(" LIMIT %d", f.Limit)
+	}
+
+	rows, err := s.db.QueryContext(ctx, q, args...)
+	if err != nil {
+		return nil, fmt.Errorf("query events: %w", err)
+	}
+	defer rows.Close()
+
+	var out []contract.FluidEvent
+	for rows.Next() {
+		var body string
+		if err := rows.Scan(&body); err != nil {
+			return nil, err
+		}
+		var ev contract.FluidEvent
+		if err := json.Unmarshal([]byte(body), &ev); err != nil {
+			return nil, fmt.Errorf("decode stored event: %w", err)
+		}
+		out = append(out, ev)
+	}
+	return out, rows.Err()
+}
+
+// WriteTelemetry records a normalized interaction event.
+func (s *SQLStore) WriteTelemetry(ctx context.Context, ev contract.FluidTelemetry) error {
+	body, err := json.Marshal(ev)
+	if err != nil {
+		return fmt.Errorf("marshal telemetry %s: %w", ev.ID, err)
+	}
+
+	var revision, experiment, cohort any
+	if ev.Revision != nil {
+		revision = string(*ev.Revision)
+	}
+	if ev.Experiment != nil {
+		experiment = string(*ev.Experiment)
+	}
+	if ev.Cohort != nil {
+		cohort = string(*ev.Cohort)
+	}
+
+	q := fmt.Sprintf(
+		`INSERT INTO fluid_telemetry (id, occurred_at, interface_id, kind, revision, experiment, cohort, body)
+		 VALUES (%s, %s, %s, %s, %s, %s, %s, %s)`,
+		s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6), s.arg(7), s.arg(8))
+
+	_, err = s.db.ExecContext(ctx, q,
+		ev.ID, rfc3339(ev.OccurredAt), string(ev.InterfaceID), string(ev.Kind),
+		revision, experiment, cohort, string(body))
+	if err != nil {
+		return fmt.Errorf("write telemetry %s: %w", ev.ID, err)
+	}
+	return nil
+}
+
+// Telemetry returns matching telemetry in occurrence order.
+func (s *SQLStore) Telemetry(ctx context.Context, f TelemetryFilter) ([]contract.FluidTelemetry, error) {
+	var (
+		where []string
+		args  []any
+	)
+	add := func(clause string, v any) {
+		args = append(args, v)
+		where = append(where, fmt.Sprintf(clause, s.arg(len(args))))
+	}
+
+	if f.InterfaceID != "" {
+		add("interface_id = %s", string(f.InterfaceID))
+	}
+	if f.Revision != "" {
+		add("revision = %s", string(f.Revision))
+	}
+	if f.Experiment != "" {
+		add("experiment = %s", string(f.Experiment))
+	}
+	if f.Cohort != "" {
+		add("cohort = %s", string(f.Cohort))
+	}
+	if f.Kind != "" {
+		add("kind = %s", string(f.Kind))
+	}
+	if !f.Since.IsZero() {
+		add("occurred_at >= %s", rfc3339(f.Since))
+	}
+	if !f.Until.IsZero() {
+		add("occurred_at <= %s", rfc3339(f.Until))
+	}
+
+	q := "SELECT body FROM fluid_telemetry"
+	if len(where) > 0 {
+		q += " WHERE " + strings.Join(where, " AND ")
+	}
+	q += " ORDER BY occurred_at, id"
+	if f.Limit > 0 {
+		q += fmt.Sprintf(" LIMIT %d", f.Limit)
+	}
+
+	rows, err := s.db.QueryContext(ctx, q, args...)
+	if err != nil {
+		return nil, fmt.Errorf("query telemetry: %w", err)
+	}
+	defer rows.Close()
+
+	var out []contract.FluidTelemetry
+	for rows.Next() {
+		var body string
+		if err := rows.Scan(&body); err != nil {
+			return nil, err
+		}
+		var ev contract.FluidTelemetry
+		if err := json.Unmarshal([]byte(body), &ev); err != nil {
+			return nil, fmt.Errorf("decode stored telemetry: %w", err)
+		}
+		out = append(out, ev)
+	}
+	return out, rows.Err()
+}
+
+// PutRecord stores or supersedes a derived record.
+func (s *SQLStore) PutRecord(ctx context.Context, kind contract.EntityKind, id string, body []byte) error {
+	if id == "" {
+		return errors.New("record has no id")
+	}
+	if !json.Valid(body) {
+		return fmt.Errorf("record %s/%s: body is not valid JSON", kind, id)
+	}
+
+	q := fmt.Sprintf(
+		`INSERT INTO fluid_records (kind, id, updated_at, body) VALUES (%s, %s, %s, %s)
+		 ON CONFLICT (kind, id) DO UPDATE SET updated_at = excluded.updated_at, body = excluded.body`,
+		s.arg(1), s.arg(2), s.arg(3), s.arg(4))
+
+	_, err := s.db.ExecContext(ctx, q, string(kind), id, rfc3339(time.Now()), string(body))
+	if err != nil {
+		return fmt.Errorf("put record %s/%s: %w", kind, id, err)
+	}
+	return nil
+}
+
+// Record returns one stored record.
+func (s *SQLStore) Record(ctx context.Context, kind contract.EntityKind, id string) ([]byte, error) {
+	q := fmt.Sprintf(`SELECT body FROM fluid_records WHERE kind = %s AND id = %s`, s.arg(1), s.arg(2))
+
+	var body string
+	err := s.db.QueryRowContext(ctx, q, string(kind), id).Scan(&body)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, fmt.Errorf("%w: %s %s", ErrNotFound, kind, id)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return []byte(body), nil
+}
+
+// Records returns every record of a kind.
+func (s *SQLStore) Records(ctx context.Context, kind contract.EntityKind) (map[string][]byte, error) {
+	q := fmt.Sprintf(`SELECT id, body FROM fluid_records WHERE kind = %s ORDER BY id`, s.arg(1))
+
+	rows, err := s.db.QueryContext(ctx, q, string(kind))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := map[string][]byte{}
+	for rows.Next() {
+		var id, body string
+		if err := rows.Scan(&id, &body); err != nil {
+			return nil, err
+		}
+		out[id] = []byte(body)
+	}
+	return out, rows.Err()
+}
+
+// Close releases the database.
+func (s *SQLStore) Close() error { return s.db.Close() }

internal/evidence/store.go

@@ -0,0 +1,73 @@
+// Package evidence implements the FLUID evidence store.
+//
+// ArchitectureBlueprint.md section 26 requires append-only history, with
+// mutable summary views derived from immutable events. Auditability is a core
+// invariant (FluidAPIStandards.md section 25): the store must be able to answer
+// what changed, why, on what evidence, and how the prior state is restored.
+package evidence
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Store is the append-only evidence log plus the queries built on it.
+//
+// Nothing in this interface updates or deletes. That is deliberate: history
+// that can be rewritten is not evidence, and Blueprint section 28.1 gives the
+// evidence writer permission to append but not to revise.
+type Store interface {
+	// AppendEvent records a lifecycle transition.
+	AppendEvent(context.Context, contract.FluidEvent) error
+	// Events returns matching events in occurrence order.
+	Events(context.Context, EventFilter) ([]contract.FluidEvent, error)
+
+	// WriteTelemetry records a normalized interaction event.
+	WriteTelemetry(context.Context, contract.FluidTelemetry) error
+	// Telemetry returns matching telemetry in occurrence order.
+	Telemetry(context.Context, TelemetryFilter) ([]contract.FluidTelemetry, error)
+
+	// PutRecord stores or supersedes a FLUID record (pressure, hypothesis,
+	// revision, experiment, feedback, backend requirement).
+	//
+	// Records are summary state derived from events; the event log remains the
+	// authority. A record may be rewritten, an event may not.
+	PutRecord(ctx context.Context, kind contract.EntityKind, id string, body []byte) error
+	// Record returns one stored record.
+	Record(ctx context.Context, kind contract.EntityKind, id string) ([]byte, error)
+	// Records returns every record of a kind, ordered by id.
+	Records(ctx context.Context, kind contract.EntityKind) (map[string][]byte, error)
+
+	Close() error
+}
+
+// ErrNotFound reports a record or event that does not exist.
+var ErrNotFound = errors.New("not found")
+
+// ErrImmutable reports an attempt to rewrite history.
+var ErrImmutable = errors.New("evidence events are append-only")
+
+// EventFilter narrows an event query. Zero values mean "no constraint".
+type EventFilter struct {
+	EntityType contract.EntityKind
+	EntityID   string
+	EventType  string
+	Since      time.Time
+	Until      time.Time
+	Limit      int
+}
+
+// TelemetryFilter narrows a telemetry query.
+type TelemetryFilter struct {
+	InterfaceID contract.InterfaceID
+	Revision    contract.RevisionID
+	Experiment  contract.ExperimentID
+	Cohort      contract.CohortID
+	Kind        contract.FluidTelemetryKind
+	Since       time.Time
+	Until       time.Time
+	Limit       int
+}

internal/evidence/store_test.go

@@ -0,0 +1,214 @@
+package evidence
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	_ "modernc.org/sqlite"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+func newStore(t *testing.T) *SQLStore {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "evidence.db")
+	s, err := OpenSQLite(context.Background(), path)
+	if err != nil {
+		t.Fatalf("open store: %v", err)
+	}
+	t.Cleanup(func() { _ = s.Close() })
+	return s
+}
+
+func event(id string, at time.Time, entity contract.EntityKind, entityID, kind string) contract.FluidEvent {
+	return contract.FluidEvent{
+		SchemaVersion: "0.1",
+		ID:            contract.EventID(id),
+		OccurredAt:    at,
+		EntityType:    contract.FluidEventEntityType(entity),
+		EntityID:      entityID,
+		EventType:     kind,
+		Actor:         contract.Actor{Type: contract.ActorTypeSystem, ID: "test"},
+	}
+}
+
+func TestAppendAndQueryEvents(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+	base := time.Date(2026, 9, 4, 12, 0, 0, 0, time.UTC)
+
+	events := []contract.FluidEvent{
+		event("EV-3", base.Add(2*time.Minute), contract.KindRevision, "R-1", "PUBLISHED"),
+		event("EV-1", base, contract.KindRevision, "R-1", "CREATED"),
+		event("EV-2", base.Add(time.Minute), contract.KindRevision, "R-1", "VERIFIED"),
+		event("EV-4", base.Add(3*time.Minute), contract.KindHypothesis, "H-1", "CREATED"),
+	}
+	for _, ev := range events {
+		if err := s.AppendEvent(ctx, ev); err != nil {
+			t.Fatalf("append %s: %v", ev.ID, err)
+		}
+	}
+
+	got, err := s.Events(ctx, EventFilter{EntityType: contract.KindRevision, EntityID: "R-1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 3 {
+		t.Fatalf("got %d events, want 3", len(got))
+	}
+	// Insertion order was scrambled; occurrence order must come back sorted,
+	// because an audit trail read out of order is worse than none.
+	want := []string{"CREATED", "VERIFIED", "PUBLISHED"}
+	for i, ev := range got {
+		if ev.EventType != want[i] {
+			t.Errorf("position %d = %s, want %s", i, ev.EventType, want[i])
+		}
+	}
+}
+
+// TestEventsAreAppendOnly checks the invariant at the database, not in Go.
+// Blueprint 28.1 gives the evidence writer permission to append and not to
+// revise; a Go-level convention would not bind a CLI or a psql session.
+func TestEventsAreAppendOnly(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+
+	ev := event("EV-1", time.Now(), contract.KindRevision, "R-1", "CREATED")
+	if err := s.AppendEvent(ctx, ev); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := s.db.ExecContext(ctx, `UPDATE fluid_events SET event_type = 'REWRITTEN'`); err == nil {
+		t.Error("UPDATE on fluid_events succeeded; history is rewritable")
+	} else if !strings.Contains(err.Error(), "append-only") {
+		t.Errorf("UPDATE failed for the wrong reason: %v", err)
+	}
+
+	if _, err := s.db.ExecContext(ctx, `DELETE FROM fluid_events`); err == nil {
+		t.Error("DELETE on fluid_events succeeded; history is erasable")
+	} else if !strings.Contains(err.Error(), "append-only") {
+		t.Errorf("DELETE failed for the wrong reason: %v", err)
+	}
+
+	// The original event must still be there and unchanged.
+	got, err := s.Events(ctx, EventFilter{EntityID: "R-1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].EventType != "CREATED" {
+		t.Errorf("event was altered: %+v", got)
+	}
+}
+
+func TestDuplicateEventIDRefused(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+	ev := event("EV-1", time.Now(), contract.KindRevision, "R-1", "CREATED")
+	if err := s.AppendEvent(ctx, ev); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.AppendEvent(ctx, ev); err == nil {
+		t.Error("duplicate event id accepted; an event was silently replayed")
+	}
+}
+
+func TestTelemetryRoundTripAndFilters(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+	base := time.Date(2026, 9, 5, 8, 0, 0, 0, time.UTC)
+
+	r1 := contract.RevisionID("R-1")
+	r2 := contract.RevisionID("R-2")
+	exp := contract.ExperimentID("E-1")
+	cohort := contract.CohortID("coding-agents")
+
+	for i, rev := range []*contract.RevisionID{&r1, &r1, &r2} {
+		ev := contract.FluidTelemetry{
+			SchemaVersion: "0.1",
+			ID:            "tl-" + string(rune('a'+i)),
+			OccurredAt:    base.Add(time.Duration(i) * time.Minute),
+			InterfaceID:   "hall-publishing",
+			Kind:          contract.FluidTelemetryKindRequest,
+			Revision:      rev,
+			Experiment:    &exp,
+			Cohort:        &cohort,
+		}
+		if err := s.WriteTelemetry(ctx, ev); err != nil {
+			t.Fatalf("write %d: %v", i, err)
+		}
+	}
+
+	all, err := s.Telemetry(ctx, TelemetryFilter{InterfaceID: "hall-publishing"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(all) != 3 {
+		t.Fatalf("got %d telemetry rows, want 3", len(all))
+	}
+
+	only1, err := s.Telemetry(ctx, TelemetryFilter{Revision: "R-1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(only1) != 2 {
+		t.Errorf("revision filter returned %d rows, want 2", len(only1))
+	}
+
+	// The measurement window is what makes a fitness comparison honest, so
+	// bounded queries have to be exact.
+	windowed, err := s.Telemetry(ctx, TelemetryFilter{
+		InterfaceID: "hall-publishing",
+		Since:       base.Add(time.Minute),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(windowed) != 2 {
+		t.Errorf("windowed query returned %d rows, want 2", len(windowed))
+	}
+}
+
+func TestRecordsAreSupersedable(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+
+	first, _ := json.Marshal(map[string]string{"status": "OPEN"})
+	if err := s.PutRecord(ctx, contract.KindPressure, "P-1", first); err != nil {
+		t.Fatal(err)
+	}
+
+	// Records are derived summary state and may be rewritten; the event log
+	// remains the authority for how they got there.
+	second, _ := json.Marshal(map[string]string{"status": "ADDRESSED"})
+	if err := s.PutRecord(ctx, contract.KindPressure, "P-1", second); err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := s.Record(ctx, contract.KindPressure, "P-1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var decoded map[string]string
+	if err := json.Unmarshal(got, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	if decoded["status"] != "ADDRESSED" {
+		t.Errorf("record not superseded: %v", decoded)
+	}
+
+	if _, err := s.Record(ctx, contract.KindPressure, "P-missing"); !errors.Is(err, ErrNotFound) {
+		t.Errorf("missing record returned %v, want ErrNotFound", err)
+	}
+}
+
+func TestPutRecordRejectsNonJSON(t *testing.T) {
+	s := newStore(t)
+	if err := s.PutRecord(context.Background(), contract.KindPressure, "P-1", []byte("not json")); err == nil {
+		t.Error("non-JSON record body accepted")
+	}
+}