Add fitness evaluator and telemetry measurer

FLUID-WP-0005 T07. Primary metrics, guardrails, secondary observations
and learning signals stay in separate roles, and there is no universal
scalar: the standards decline to define one, and inventing one would
make the interface's most important trade-offs on the operator's behalf.

Three verdict rules carry the weight. A guardrail breach dominates
primary success, because a candidate that hit its target while
regressing a guardrail traded something it was told not to. Apparent
success on too few samples is inconclusive rather than successful, since
promoting on three requests is promoting on noise. A metric missing from
one side is reported absent rather than defaulted to zero, which would
read as a dramatic change that never happened.

Specs are an input, never derived from the data, so that changing
success criteria after seeing results requires a recorded amendment
rather than being indistinguishable from normal operation.

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/fitness/measure.go

@@ -0,0 +1,164 @@
+package fitness
+
+import (
+	"sort"
+	"time"
+
+	"github.com/tegwick/fluid-core/internal/contract"
+)
+
+// Standard metric names the measurer derives from telemetry.
+//
+// These are the metrics the Blueprint's worked example turns on, and having
+// them named in one place stops a hypothesis and an evaluation from measuring
+// subtly different things under the same word.
+const (
+	MetricRequestsPerTask = "requests_per_completed_task"
+	MetricP95LatencyMS    = "p95_latency_ms"
+	MetricErrorRate       = "error_rate"
+	MetricSuccessRate     = "success_rate"
+)
+
+// Measurer derives metric observations from raw telemetry.
+type Measurer struct {
+	// ChainGap bounds one completed task when the consumer supplies no chain
+	// id, matching the topology analyzer's grouping.
+	ChainGap time.Duration
+}
+
+// NewMeasurer returns a measurer with the default grouping window.
+func NewMeasurer() *Measurer { return &Measurer{ChainGap: 30 * time.Second} }
+
+// Measure computes the standard metrics per revision over a window.
+//
+// Only events inside the window count. Blueprint section 18 requires the
+// measurement window to be retained, and quietly including events outside it
+// would make a comparison irreproducible.
+func (m *Measurer) Measure(events []contract.FluidTelemetry, window Window) []Observation {
+	type acc struct {
+		requests int
+		errors   int
+		tasks    map[string]int
+		latency  []float64
+	}
+	byRevision := map[contract.RevisionID]*acc{}
+
+	for _, ev := range events {
+		if !inWindow(ev.OccurredAt, window) {
+			continue
+		}
+		if ev.Revision == nil {
+			continue
+		}
+		rev := *ev.Revision
+
+		a, ok := byRevision[rev]
+		if !ok {
+			a = &acc{tasks: map[string]int{}}
+			byRevision[rev] = a
+		}
+
+		a.requests++
+		if ev.Error != nil {
+			a.errors++
+		}
+		if ev.Request != nil && ev.Request.LatencyMS != nil {
+			a.latency = append(a.latency, *ev.Request.LatencyMS)
+		}
+		a.tasks[taskKey(ev, m.ChainGap)]++
+	}
+
+	revisions := make([]contract.RevisionID, 0, len(byRevision))
+	for r := range byRevision {
+		revisions = append(revisions, r)
+	}
+	sort.Slice(revisions, func(i, j int) bool { return revisions[i] < revisions[j] })
+
+	var out []Observation
+	for _, rev := range revisions {
+		a := byRevision[rev]
+		if a.requests == 0 {
+			continue
+		}
+
+		// Requests per completed task is the metric that catches an interface
+		// making consumers assemble what it could have handed them.
+		tasks := len(a.tasks)
+		if tasks > 0 {
+			out = append(out, Observation{
+				Metric:   MetricRequestsPerTask,
+				Revision: rev,
+				Value:    round4(float64(a.requests) / float64(tasks)),
+				Samples:  tasks,
+			})
+		}
+
+		out = append(out, Observation{
+			Metric:   MetricErrorRate,
+			Revision: rev,
+			Value:    round4(float64(a.errors) / float64(a.requests)),
+			Samples:  a.requests,
+		})
+		out = append(out, Observation{
+			Metric:   MetricSuccessRate,
+			Revision: rev,
+			Value:    round4(float64(a.requests-a.errors) / float64(a.requests)),
+			Samples:  a.requests,
+		})
+
+		if len(a.latency) > 0 {
+			out = append(out, Observation{
+				Metric:   MetricP95LatencyMS,
+				Revision: rev,
+				Value:    round4(percentile(a.latency, 0.95)),
+				Samples:  len(a.latency),
+			})
+		}
+	}
+	return out
+}
+
+// taskKey groups events into completed tasks.
+func taskKey(ev contract.FluidTelemetry, gap time.Duration) string {
+	consumer := ev.ConsumerRef
+	if consumer == "" {
+		consumer = ev.CorrelationID
+	}
+	if ev.Sequence != nil && ev.Sequence.ChainID != "" {
+		return consumer + "/" + ev.Sequence.ChainID
+	}
+	// Without a chain id, bucket by consumer and elapsed gap. This is a
+	// heuristic, which is the reason to prefer chain ids from agentic
+	// consumers where they can supply them.
+	bucket := ev.OccurredAt.Truncate(gap).UTC().Format(time.RFC3339)
+	return consumer + "/" + bucket
+}
+
+func inWindow(t time.Time, w Window) bool {
+	if !w.Start.IsZero() && t.Before(w.Start) {
+		return false
+	}
+	if w.End != nil && t.After(*w.End) {
+		return false
+	}
+	return true
+}
+
+// percentile returns the nearest-rank percentile of values.
+func percentile(values []float64, p float64) float64 {
+	sorted := make([]float64, len(values))
+	copy(sorted, values)
+	sort.Float64s(sorted)
+
+	if len(sorted) == 1 {
+		return sorted[0]
+	}
+	rank := int(p*float64(len(sorted)-1) + 0.5)
+	if rank < 0 {
+		rank = 0
+	}
+	if rank >= len(sorted) {
+		rank = len(sorted) - 1
+	}
+	return sorted[rank]
+}