fix(bench): modern DataHub seeding, enrichment-volume reporting, and the first published pilot results (#951)

Standing the a2 arm up on a current DataHub (v1.6.0 quickstart on
OpenSearch) surfaced two harness defects, both exercised live after the fix:

- The MCE emitter wrote the legacy datahub put --file bulk format, which
  current datahub CLIs no longer accept. Aspects are now emitted in the MCP
  file format (GenericAspect {"json": ...}), and bench-seed-datahub drives
  datahub ingest with a generated file-source recipe (BENCH_DATAHUB_GMS
  overrides the sink). Seed artifacts regenerated.
- Suite reports summed enrichment_tokens_dedup, which is zero when
  enrichment runs in full mode, so the a2 arm's enrichment volume read as
  zero. Metrics and reports now carry enrichment_tokens_full (the delivered
  volume) alongside dedup, and the suite table prints the full volume.

docs/reference/benchmarks.md is the new living results page (mkdocs nav +
llms pair updated): phase 1 pilot manifest, arm-by-suite and trap-class
tables, honest caveats, and a result-history table that future phases
append to. Raw pilot run JSON (claude-sonnet-5, k=3, 60 episodes, zero
harness failures) is committed under bench/results/v1.102.0-pilot/ as the
first history entry.

Pilot headline recorded on #930: knowledge-trap accuracy 60% (a0) vs 100%
(a2) with the policy-revenue tasks going 0/6 vs 6/6, median knowledge-suite
tool calls 16 vs 8, wall clock 67s vs 25s.

Author: cjimti

Makefile

@@ -857,9 +857,12 @@ bench-up: e2e-up
 	@echo "Platform ready (pid $$(cat $(BENCH_PID)), arm $(BENCH_ARM))."
 
 ## bench-seed-datahub: Push bench metadata into a running DataHub quickstart (a2 arm)
+BENCH_DATAHUB_GMS ?= http://localhost:8080
 bench-seed-datahub:
 	@command -v datahub >/dev/null 2>&1 || { echo "ERROR: datahub CLI not found (pip install acryl-datahub)"; exit 1; }
-	datahub put --file bench/seed/datahub/bench_mces.json
+	@mkdir -p $(BUILD_DIR)
+	@printf 'source:\n  type: file\n  config:\n    path: %s/bench/seed/datahub/bench_mces.json\nsink:\n  type: datahub-rest\n  config:\n    server: %s\n' "$$(pwd)" "$(BENCH_DATAHUB_GMS)" > $(BUILD_DIR)/bench-datahub-recipe.yml
+	datahub ingest -c $(BUILD_DIR)/bench-datahub-recipe.yml
 
 ## bench-run: Run the benchmark (ARM must match bench-up; LLM=anthropic|scripted, SUITE=, K=, MODEL=)
 bench-run:

bench/internal/auditapi/auditapi.go

@@ -174,7 +174,10 @@ type Metrics struct {
 	TotalDurationMS int64 `json:"total_duration_ms"`
 	// EnrichedCalls counts calls where cross-enrichment was applied.
 	EnrichedCalls int `json:"enriched_calls"`
-	// EnrichmentTokensDedup sums the deduplicated enrichment token volume.
+	// EnrichmentTokensFull sums the full enrichment token volume delivered.
+	EnrichmentTokensFull int `json:"enrichment_tokens_full"`
+	// EnrichmentTokensDedup sums the session-deduplicated enrichment token
+	// volume (zero when enrichment ran in full mode).
 	EnrichmentTokensDedup int `json:"enrichment_tokens_dedup"`
 }
 
@@ -189,6 +192,7 @@ func Summarize(events []Event) Metrics {
 		m.TotalDurationMS += e.DurationMS
 		if e.EnrichmentApplied {
 			m.EnrichedCalls++
+			m.EnrichmentTokensFull += e.EnrichmentTokensFull
 			m.EnrichmentTokensDedup += e.EnrichmentTokensDedup
 		}
 	}

bench/internal/auditapi/auditapi_test.go

@@ -163,12 +163,12 @@ func TestClientErrorStatus(t *testing.T) {
 
 func TestSummarize(t *testing.T) {
 	events := []Event{
-		{Success: true, DurationMS: 10, EnrichmentApplied: true, EnrichmentTokensDedup: 100},
+		{Success: true, DurationMS: 10, EnrichmentApplied: true, EnrichmentTokensFull: 300, EnrichmentTokensDedup: 100},
 		{Success: false, DurationMS: 5},
-		{Success: true, DurationMS: 20, EnrichmentApplied: true, EnrichmentTokensDedup: 50},
+		{Success: true, DurationMS: 20, EnrichmentApplied: true, EnrichmentTokensFull: 200, EnrichmentTokensDedup: 50},
 	}
 	m := Summarize(events)
-	want := Metrics{AuditedCalls: 3, Errors: 1, TotalDurationMS: 35, EnrichedCalls: 2, EnrichmentTokensDedup: 150}
+	want := Metrics{AuditedCalls: 3, Errors: 1, TotalDurationMS: 35, EnrichedCalls: 2, EnrichmentTokensFull: 500, EnrichmentTokensDedup: 150}
 	if m != want {
 		t.Errorf("Summarize = %+v, want %+v", m, want)
 	}

bench/internal/gen/emit_datahub.go

@@ -10,14 +10,26 @@ func benchURN(table string) string {
 	return fmt.Sprintf("urn:li:dataset:(urn:li:dataPlatform:trino,memory.bench.%s,PROD)", table)
 }
 
-// mcp is one metadata change proposal in the `datahub put --file` format the
-// e2e testdata uses (test/e2e/testdata/datahub/datasets.json).
+// mcp is one metadata change proposal in the MCP file format `datahub ingest`
+// consumes (file source): the aspect payload is a GenericAspect wrapped as
+// {"json": {...}}. The legacy `datahub put --file` bulk mode that the e2e
+// testdata targets no longer exists in current CLIs.
 type mcp struct {
-	EntityURN  string `json:"entityUrn"`
-	EntityType string `json:"entityType"`
-	AspectName string `json:"aspectName"`
-	ChangeType string `json:"changeType"`
-	Aspect     any    `json:"aspect"`
+	EntityURN  string        `json:"entityUrn"`
+	EntityType string        `json:"entityType"`
+	AspectName string        `json:"aspectName"`
+	ChangeType string        `json:"changeType"`
+	Aspect     genericAspect `json:"aspect"`
+}
+
+// genericAspect is the MCP file serialization of an aspect payload.
+type genericAspect struct {
+	JSON any `json:"json"`
+}
+
+// wrap boxes an aspect payload for the MCP file format.
+func wrap(aspect any) genericAspect {
+	return genericAspect{JSON: aspect}
 }
 
 // Dataset descriptions are the A2 knowledge channel: they carry the facts the
@@ -64,11 +76,11 @@ func datasetProps(table, description string, custom map[string]string) mcp {
 		EntityType: "dataset",
 		AspectName: "datasetProperties",
 		ChangeType: "UPSERT",
-		Aspect: map[string]any{
+		Aspect: wrap(map[string]any{
 			"name":             table,
 			"description":      description,
 			"customProperties": custom,
-		},
+		}),
 	}
 }
 
@@ -85,7 +97,7 @@ func editableSchema(table string, fields map[string]string) mcp {
 		EntityType: "dataset",
 		AspectName: "editableSchemaMetadata",
 		ChangeType: "UPSERT",
-		Aspect:     map[string]any{"editableSchemaFieldInfo": infos},
+		Aspect:     wrap(map[string]any{"editableSchemaFieldInfo": infos}),
 	}
 }
 
@@ -96,11 +108,11 @@ func deprecation(table, note string) mcp {
 		EntityType: "dataset",
 		AspectName: "deprecation",
 		ChangeType: "UPSERT",
-		Aspect: map[string]any{
+		Aspect: wrap(map[string]any{
 			"deprecated": true,
 			"note":       note,
 			"actor":      "urn:li:corpuser:bench-seed",
-		},
+		}),
 	}
 }
 
@@ -111,6 +123,6 @@ func tag(table string) mcp {
 		EntityType: "dataset",
 		AspectName: "globalTags",
 		ChangeType: "UPSERT",
-		Aspect:     map[string]any{"tags": []map[string]string{{"tag": "urn:li:tag:bench"}}},
+		Aspect:     wrap(map[string]any{"tags": []map[string]string{{"tag": "urn:li:tag:bench"}}}),
 	}
 }

bench/internal/report/report.go

@@ -80,6 +80,7 @@ type SuiteSummary struct {
 	P90ToolCalls          float64 `json:"p90_tool_calls"`
 	MedianWallMS          float64 `json:"median_wall_ms"`
 	ToolErrors            int     `json:"tool_errors"`
+	EnrichmentTokensFull  int     `json:"enrichment_tokens_full"`
 	EnrichmentTokensDedup int     `json:"enrichment_tokens_dedup"`
 }
 
@@ -165,6 +166,7 @@ func summarizeSuite(suite string, attempts []Attempt, tasks []TaskSummary) Suite
 		toolCalls = append(toolCalls, float64(a.ToolCalls))
 		wall = append(wall, float64(a.WallMS))
 		s.ToolErrors += a.ToolErrors
+		s.EnrichmentTokensFull += a.Audit.EnrichmentTokensFull
 		s.EnrichmentTokensDedup += a.Audit.EnrichmentTokensDedup
 	}
 	if s.Graded > 0 {
@@ -237,7 +239,7 @@ func (r *Results) HumanSummary() string {
 	for _, s := range r.Suites {
 		fmt.Fprintf(&b, "%-7s %5d  %6d  %7.1f%%  %5.1f%%  %9.1f  %9.1f  %11.0f  %9d  %10d  %13d\n",
 			s.Suite, s.Tasks, s.Graded, s.Accuracy*100, s.PassKRate*100,
-			s.MedianToolCalls, s.P90ToolCalls, s.MedianWallMS, s.ToolErrors, s.EnrichmentTokensDedup, s.HarnessFailures)
+			s.MedianToolCalls, s.P90ToolCalls, s.MedianWallMS, s.ToolErrors, s.EnrichmentTokensFull, s.HarnessFailures)
 	}
 	b.WriteString("\ntask                            graded  correct  pass^k\n")
 	for _, t := range r.Tasks {