From 7c1a3c8e721a22e625678eb4e8bd958dd89fa54a Mon Sep 17 00:00:00 2001 From: cjimti Date: Thu, 16 Jul 2026 01:53:16 -0700 Subject: [PATCH] =?UTF-8?q?bench(cold-start):=20pre-run=20hardening=20?= =?UTF-8?q?=E2=80=94=20promoted-page=20summary=20delivery,=20run=20isolati?= =?UTF-8?q?on,=20baseline=20preflight,=20cache=20settle=20(Closes=20#975)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deliver promoted knowledge-page facts through search by sending the page summary on the apply_knowledge payload (required by curriculum and protocol validation, shared with the a2 seed text, and used as the page-sink surfaced-needle). Isolate every cold-start run in its own timestamped results dir with benchrun refusing to overwrite existing results or transcripts. Bound capture verification to the current run with a since filter on the insights read. Add a baseline-integrity preflight that refuses a contaminated platform (entity descriptions/tags/deprecation, any leftover insight, any knowledge page) before any episode is spent, with explicit remediation. Pace datahub-sink promotes with a -settle window (default 5m, recorded on the manifest, skipped for page sinks and unpromoted lessons) and flush each lesson record before the pause. Classify pre-audit platform refusals in the reviewer promote path as harness errors via a shared mcpc.PreAuditRefusal. Forbid evaluator memory writes in the eval scaffold and surface an audit-side MemoryWrites validity signal in the results and summary. Close the regression-blindness test gaps (identity math, flush ordering, teachAndPromote branches, credential rotation, refusal classification, benchrun wiring) and update the cold-start runbook with the fresh-quickstart requirement, pass criteria, adapter costs, and the new run-dir/report/settle workflow. --- Makefile | 22 +- bench/README.md | 57 +- bench/benchrun/main.go | 30 + bench/benchrun/main_test.go | 66 +++ bench/curriculum/cs-traps.yaml | 3 + bench/internal/coldstart/episode.go | 80 ++- bench/internal/coldstart/episode_test.go | 71 +++ bench/internal/coldstart/preflight.go | 166 ++++++ bench/internal/coldstart/report.go | 32 +- bench/internal/coldstart/report_test.go | 20 + bench/internal/coldstart/runner.go | 83 ++- bench/internal/coldstart/runner_test.go | 546 +++++++++++++++++- bench/internal/curriculum/curriculum.go | 7 + bench/internal/curriculum/curriculum_test.go | 12 +- bench/internal/gen/emit_curriculum.go | 16 +- bench/internal/gen/emit_kp.go | 84 ++- bench/internal/gen/emit_protocols.go | 10 +- bench/internal/gen/gen_test.go | 57 ++ bench/internal/lifecycle/episode.go | 14 +- bench/internal/lifecycle/instrument.go | 9 +- bench/internal/lifecycle/instrument_test.go | 9 +- bench/internal/lifecycle/promote.go | 9 +- bench/internal/lifecycle/runner.go | 5 +- bench/internal/lifecycle/runner_test.go | 2 +- bench/internal/lifecycleapi/lifecycleapi.go | 43 ++ .../lifecycleapi/lifecycleapi_test.go | 50 +- bench/internal/mcpc/mcpc.go | 16 + bench/internal/mcpc/mcpc_test.go | 19 + bench/internal/pipeline/pipeline.go | 15 +- bench/internal/promote/promote.go | 38 +- bench/internal/promote/promote_test.go | 107 +++- bench/internal/protocol/protocol.go | 18 +- bench/internal/protocol/protocol_test.go | 11 +- bench/protocols/lc-active-customers.yaml | 1 + bench/protocols/lc-anchor-region.yaml | 1 + bench/protocols/lc-charter-cohort.yaml | 1 + bench/protocols/lc-clearance-set.yaml | 1 + bench/protocols/lc-core-market.yaml | 1 + bench/protocols/lc-focus-region-net.yaml | 1 + bench/protocols/lc-headline-tier.yaml | 1 + bench/protocols/lc-holiday-total.yaml | 1 + bench/protocols/lc-house-tier.yaml | 1 + bench/protocols/lc-peak-region.yaml | 1 + bench/protocols/lc-premium-orders.yaml | 1 + bench/protocols/lc-priority-orders.yaml | 1 + bench/protocols/lc-settlement-total.yaml | 1 + bench/protocols/lc-standard-order.yaml | 1 + bench/protocols/lc-value-tier.yaml | 1 + 48 files changed, 1596 insertions(+), 146 deletions(-) create mode 100644 bench/internal/coldstart/episode_test.go create mode 100644 bench/internal/coldstart/preflight.go diff --git a/Makefile b/Makefile index cdb95e7c..e05441a7 100644 --- a/Makefile +++ b/Makefile @@ -964,7 +964,7 @@ bench-supersede-report: @cd bench && $(GO) build -o ../$(BUILD_DIR)/benchrun ./benchrun $(BUILD_DIR)/benchrun -supersede -summarize build/bench-results/supersede-a3.json -## bench-cold-start: Run the cold-start knowledge-growth curriculum (issue #963; needs an empty-seeded a3: bench-up BENCH_ARM=a3 BENCH_SEED_PAGES=0 + bench-seed-datahub-empty; LLM=anthropic|scripted|claude-cli, K=, MODEL=) +## bench-cold-start: Run the cold-start knowledge-growth curriculum into a fresh per-run dir under build/bench-results/ (issue #963; needs an empty-seeded a3: bench-up BENCH_ARM=a3 BENCH_SEED_PAGES=0 + bench-seed-datahub-empty on a FRESH DataHub quickstart; LLM=anthropic|scripted|claude-cli, K=, MODEL=, SETTLE=) bench-cold-start: @mkdir -p build/bench-results @cd bench && $(GO) build -o ../$(BUILD_DIR)/benchrun ./benchrun @@ -972,6 +972,9 @@ bench-cold-start: @echo " (CASCADE also clears portal_threads, which FK-references knowledge pages; the bench stack is disposable scratch state.)" @$(BENCH_COMPOSE) exec -T postgres psql -q -U platform -d mcp_platform -v ON_ERROR_STOP=1 \ -c "TRUNCATE search_gate_discovery, memory_records, knowledge_changesets, portal_knowledge_pages CASCADE" + @out_dir="build/bench-results/cold-start-a3-$$(date +%Y%m%d-%H%M%S)"; \ + mkdir -p "$$out_dir"; \ + echo "Cold-start results dir: $$out_dir (each run gets its own dir; nothing is ever overwritten)"; \ $(BUILD_DIR)/benchrun \ -cold-start \ -arm a3 \ @@ -980,20 +983,25 @@ bench-cold-start: -curriculum bench/curriculum \ -tasks bench/tasks \ -git-commit $$(git rev-parse HEAD) \ - -out build/bench-results/cold-start-a3.json \ + -out "$$out_dir/results.json" \ $(if $(LLM),-llm $(LLM),) \ $(if $(SCRIPT),-script $(SCRIPT),) \ $(if $(K),-k $(K),) \ - $(if $(MODEL),-model $(MODEL),) + $(if $(MODEL),-model $(MODEL),) \ + $(if $(SETTLE),-settle $(SETTLE),) -## bench-cold-start-smoke: Run the scripted (no-API-key) cold-start smoke against the running a3 platform +## bench-cold-start-smoke: Run the scripted (no-API-key) cold-start smoke against the running a3 platform (no cache-settle pause) bench-cold-start-smoke: - @$(MAKE) bench-cold-start LLM=scripted SCRIPT=bench/curriculum/scripted-cold-start-smoke.json K=1 + @$(MAKE) bench-cold-start LLM=scripted SCRIPT=bench/curriculum/scripted-cold-start-smoke.json K=1 SETTLE=0s -## bench-cold-start-report: Print the human summary (learning curve) of the last cold-start run +## bench-cold-start-report: Print the human summary (learning curve) of a cold-start run (RESULTS=/results.json) bench-cold-start-report: + @if [ -z "$(RESULTS)" ]; then \ + echo "ERROR: set RESULTS=. Available run dirs:"; \ + ls -d build/bench-results/cold-start-a3-*/ 2>/dev/null || echo " (none under build/bench-results/)"; \ + exit 1; fi @cd bench && $(GO) build -o ../$(BUILD_DIR)/benchrun ./benchrun - $(BUILD_DIR)/benchrun -cold-start -summarize build/bench-results/cold-start-a3.json + $(BUILD_DIR)/benchrun -cold-start -summarize $(RESULTS) ## bench-report: Print the human summary of the last run for BENCH_ARM bench-report: diff --git a/bench/README.md b/bench/README.md index eebdcf2b..39b36bac 100644 --- a/bench/README.md +++ b/bench/README.md @@ -406,6 +406,36 @@ suite reuses one identity pool (a distinct teacher per lesson, fresh evaluators per checkpoint), and a run refuses to start when the lessons plus per-checkpoint evaluators exceed the pool. +**A cold-start run requires a FRESH DataHub quickstart**, not just re-ingesting +the empty seed. `apply_knowledge` description promotions write the +`editableDatasetProperties` aspect, and a prior a2 seed leaves +`editableSchemaMetadata` column docs, tags, and deprecation; the empty seed +(`bench_mces_empty.json`) upserts only `datasetProperties`, so re-ingesting it +cannot clear any of that, and the read path prefers the editable description +when non-empty. A **baseline-integrity preflight** enforces this: before any +episode is spent, the run reads every lesson entity through the platform, +lists insights for every (teacher, lesson URN) pair, and scans knowledge pages +for the curriculum slugs, refusing to start if anything is already there. +Postgres state (search gate, memory records, changesets, knowledge pages) is +reset by the `bench-cold-start` target's TRUNCATE; DataHub state requires +`datahub docker nuke`, a re-quickstart, then `make bench-seed-datahub-empty`. + +Between a successful datahub-sink promote and the next eval checkpoint the +runner pauses for the `-settle` window (default `5m`, matching the a3 semantic +cache TTL, `SETTLE=` on the make target) so a table-context cache entry +populated by the previous checkpoint's evaluators can never serve the stale +pre-promotion description to the next ones. Page-sink promotes skip the pause +(page hits are served live from the portal store, nothing is cached), as do +lessons that did not promote. The window is recorded on the results manifest, +and the scripted smoke runs with `SETTLE=0s`. + +**Pass criteria.** A zero exit code is NOT a pass signal: capture misses and +promote refusals are measured outcomes by design, so a run can exit 0 with a +flat curve. A valid full run's summary must read `lessons 6 (captured 6, +promoted 6)` with `harness failures 0`, enrichment coverage climbing from the +baseline, and no evaluator memory-write warning (evaluators are forbidden from +saving memories; the summary warns if the audit-side count is non-zero). + ## Running From the repository root: @@ -439,19 +469,38 @@ the insight/changeset APIs, supersede, grading, and the metrics against the live platform with no API key and no model variance. The cold-start suite (#963) boots the same `a3` arm but with the empty baseline -(no knowledge pages, undocumented DataHub), then teaches the curriculum: +(no knowledge pages, undocumented DataHub, on a FRESH quickstart; the preflight +refuses leftovers from a prior run or an a2 seed, see the cold-start section +above), then teaches the curriculum: ```bash -# Boot a3 with an empty enrichment layer: no knowledge pages, empty DataHub. +# Fresh DataHub quickstart first (datahub docker nuke + re-quickstart if reused), +# then boot a3 with an empty enrichment layer: no knowledge pages, empty DataHub. make bench-up BENCH_ARM=a3 BENCH_SEED_PAGES=0 make bench-seed-datahub-empty # entities present, undocumented -make bench-cold-start-smoke # scripted no-API-key loop validation +make bench-cold-start-smoke # scripted no-API-key loop validation (SETTLE=0s) make bench-cold-start K=1 # real learning-curve run (needs a model) make bench-cold-start LLM=claude-cli MODEL=sonnet K=1 # subscription run -make bench-cold-start-report # print the learning curve +make bench-cold-start-report RESULTS=build/bench-results/cold-start-a3-/results.json ``` +Every `bench-cold-start` invocation writes into its own timestamped directory +(`build/bench-results/cold-start-a3-/results.json` plus a +`results.json.transcripts/` directory beside it) and `benchrun` refuses an +`-out` that already exists, so a re-run can never overwrite a prior run's +paid-for results; `bench-cold-start-report` therefore takes the run to +summarize via `RESULTS=` (it lists the available run dirs when unset). + +Two model paths, two cost profiles. `LLM=claude-cli` runs each episode through +a real `claude -p` client and is subscription-funded (the runner strips +`ANTHROPIC_API_KEY` from the child environment): a k=1 full run is 181 episodes +(6 teaches + 7 checkpoints x 25 eval tasks) at roughly 50s each, so plan for +2.5-3 hours of wall clock. The `anthropic` adapter bills the API: k=1 is +estimated around USD 20 at claude-sonnet-5 pricing, extrapolated from the +phase-2 per-attempt token data. Either way a full run exceeds an interactive +session: launch it in the background and read the summary from the run dir. + The **scripted cold-start smoke** (`-llm scripted`) plays `curriculum/scripted-cold-start-smoke.json` (generated): each lesson captures its fact and the harness drives the real promotion; each eval task answers with its diff --git a/bench/benchrun/main.go b/bench/benchrun/main.go index edecd3d6..7848edd5 100644 --- a/bench/benchrun/main.go +++ b/bench/benchrun/main.go @@ -57,6 +57,7 @@ type config struct { merge string coldStart bool curriculumDir string + settle time.Duration supersede bool teachBudget int } @@ -102,6 +103,7 @@ func parseFlags() config { flag.StringVar(&cfg.merge, "merge", "", "comma-separated per-pass lifecycle result JSONs (with -lifecycle): merge independent k=1 passes into one k=N result and exit") flag.BoolVar(&cfg.coldStart, "cold-start", false, "run the cold-start knowledge-growth curriculum (issue #963) instead of the task suites") flag.StringVar(&cfg.curriculumDir, "curriculum", "curriculum", "curriculum YAML directory (with -cold-start)") + flag.DurationVar(&cfg.settle, "settle", 5*time.Minute, "pause between a successful promote and the next eval checkpoint (with -cold-start), matching the a3 semantic-cache TTL so a stale pre-promotion cache entry cannot attenuate the lift; 0 disables (scripted smoke)") flag.BoolVar(&cfg.supersede, "supersede", false, "run the isolated supersede sub-benchmark (issue #964: the supersede protocols only) instead of the full S5 lifecycle") flag.IntVar(&cfg.teachBudget, "teach-budget", 0, "override the per-episode tool-call budget for the capture-bearing stages (teach, update); 0 = protocol budget (issue #964 capture-budget lever)") flag.Parse() @@ -325,6 +327,9 @@ func runColdStart(cfg config) error { if cfg.baseline != "" { return errors.New("-baseline is not supported with -cold-start (the regression gate scores S1-S3 task suites, not the learning curve)") } + if err := refuseExistingRunArtifacts(cfg.out); err != nil { + return err + } log := slog.New(slog.NewTextHandler(os.Stderr, nil)) opts := coldstart.Options{ Target: target.Target{BaseURL: cfg.url, Credential: cfg.credential}, @@ -338,6 +343,7 @@ func runColdStart(cfg config) error { GitCommit: cfg.gitCommit, AuditTimeout: cfg.auditTimeout, IdentityKeys: cfg.identityKeys, + Settle: cfg.settle, OnCheckpoint: func(r *coldstart.Results) { if err := r.WriteJSON(cfg.out); err != nil { log.Warn("checkpoint write", "error", err) @@ -706,6 +712,30 @@ func transcriptDir(out string) string { return filepath.Join(filepath.Dir(out), filepath.Base(out)+".transcripts") } +// refuseExistingRunArtifacts errors when out or its sibling transcript +// directory already exists. A cold-start run's results and transcripts are +// paid-for evidence that can never be regenerated; an existing -out would be +// silently clobbered by the per-checkpoint flush, and the transcript dir by +// the deterministic per-episode filenames — a run interrupted before its first +// flush leaves only transcripts, which a rerun to the same -out would destroy. +// There is deliberately no override: the operator picks a new path, every +// run's data is kept. +func refuseExistingRunArtifacts(out string) error { + if _, err := os.Stat(out); err == nil { + return fmt.Errorf("-out %s already exists; refusing to overwrite a prior run's results; choose a new -out path (all run data is kept)", out) + } + if dir := transcriptDir(out); dirExists(dir) { + return fmt.Errorf("transcript directory %s already exists (an earlier run's episode transcripts); refusing to overwrite; choose a new -out path (all run data is kept)", dir) + } + return nil +} + +// dirExists reports whether path exists and is a directory. +func dirExists(path string) bool { + info, err := os.Stat(path) + return err == nil && info.IsDir() +} + // claudeCLIProvider is the -llm value selecting the real Claude Code client // path (see package claudecli). const claudeCLIProvider = "claude-cli" diff --git a/bench/benchrun/main_test.go b/bench/benchrun/main_test.go index e1271c83..a19a3645 100644 --- a/bench/benchrun/main_test.go +++ b/bench/benchrun/main_test.go @@ -3,6 +3,7 @@ package main import ( "os" "path/filepath" + "strings" "testing" "github.com/txn2/mcp-data-platform/bench/internal/lifecycle" @@ -245,6 +246,71 @@ func TestSupersedeRejectsBaseline(t *testing.T) { } } +// TestColdStartRequiresArm proves the cold-start run refuses to start without +// an arm rather than launching an unattributable run. +func TestColdStartRequiresArm(t *testing.T) { + if err := runColdStart(config{coldStart: true}); err == nil { + t.Fatal("cold-start run accepted an empty arm") + } +} + +// TestColdStartRejectsBaseline proves -baseline is refused for cold-start runs: +// the regression gate scores the S1-S3 task shape, not the learning curve, so a +// silently-ignored -baseline would give a false sense of gating. +func TestColdStartRejectsBaseline(t *testing.T) { + if err := runColdStart(config{coldStart: true, arm: "a3", baseline: "b.json"}); err == nil { + t.Fatal("cold-start run accepted -baseline") + } +} + +// TestColdStartRefusesToOverwriteResults proves a cold-start run whose -out +// already exists is refused before anything is spent: a prior run's results and +// transcripts are paid-for evidence that must never be overwritten, and there +// is deliberately no override flag. +func TestColdStartRefusesToOverwriteResults(t *testing.T) { + out := filepath.Join(t.TempDir(), "results.json") + if err := os.WriteFile(out, []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + err := runColdStart(config{coldStart: true, arm: "a3", out: out}) + if err == nil { + t.Fatal("cold-start run accepted an -out that already exists") + } + if got := err.Error(); !strings.Contains(got, "refusing to overwrite") { + t.Errorf("error should explain the overwrite refusal, got: %s", got) + } +} + +// TestColdStartRefusesToOverwriteTranscripts proves the refusal also covers the +// sibling transcript directory: a run interrupted before its first checkpoint +// flush leaves transcripts but no results file, and a rerun to the same -out +// would overwrite them episode by episode (deterministic filenames). +func TestColdStartRefusesToOverwriteTranscripts(t *testing.T) { + out := filepath.Join(t.TempDir(), "results.json") + if err := os.MkdirAll(transcriptDir(out), 0o750); err != nil { + t.Fatal(err) + } + err := runColdStart(config{coldStart: true, arm: "a3", out: out}) + if err == nil { + t.Fatal("cold-start run accepted an -out whose transcript directory already exists") + } + if got := err.Error(); !strings.Contains(got, "transcript directory") { + t.Errorf("error should name the transcript directory, got: %s", got) + } +} + +// TestBuildColdStartFactoryErrors proves the factory wiring fails fast on a +// scripted run without a script and on an unknown provider, mirroring the +// lifecycle factory contract. +func TestBuildColdStartFactoryErrors(t *testing.T) { + if _, err := buildColdStartFactory(config{llmProvider: "scripted"}); err == nil { + t.Error("scripted provider without -script was accepted") + } + if _, err := buildColdStartFactory(config{llmProvider: "nope"}); err == nil { + t.Error("unknown provider was accepted") + } +} + // baselineFile writes a results JSON to a temp path and returns it, standing in // for a committed baseline that gateOnBaseline loads from disk. func baselineFile(t *testing.T, r *report.Results) string { diff --git a/bench/curriculum/cs-traps.yaml b/bench/curriculum/cs-traps.yaml index 97b0b2da..f4c8f54d 100644 --- a/bench/curriculum/cs-traps.yaml +++ b/bench/curriculum/cs-traps.yaml @@ -22,6 +22,7 @@ lessons: page: slug: revenue-reporting-policy title: Revenue Reporting Policy + summary: Revenue = amount - discount over completed orders only; amounts are stored in US cents. The authoritative definition for all bench warehouse reporting. body: |- # Revenue Reporting Policy @@ -51,6 +52,7 @@ lessons: page: slug: fiscal-calendar-policy title: Fiscal Calendar Policy + summary: 'The fiscal year starts February 1: fiscal year 2025 runs 2025-02-01 through 2026-01-31. Fiscal figures must not be computed over the calendar year.' body: |- # Fiscal Calendar Policy @@ -91,6 +93,7 @@ lessons: page: slug: customer-tier-definitions title: Customer Tier Definitions + summary: A 'key account' is any customer on the plus or enterprise tier, a derived grouping not stored in any column and broader than the enterprise tier alone. body: |- # Customer Tier Definitions diff --git a/bench/internal/coldstart/episode.go b/bench/internal/coldstart/episode.go index 8d8d0ade..b973a4c1 100644 --- a/bench/internal/coldstart/episode.go +++ b/bench/internal/coldstart/episode.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "time" "github.com/txn2/mcp-data-platform/bench/internal/agent" @@ -37,11 +38,16 @@ Rules: // evalScaffold frames an evaluation session. The evaluator was never taught the // facts, so its only knowledge source is what the platform surfaces — search // results, catalog descriptions, and knowledge pages — which is exactly the -// promoted-knowledge delivery channel the curve measures. +// promoted-knowledge delivery channel the curve measures. The no-self-teach +// rule keeps it that way: an evaluator that saved a memory could surface it to +// a LATER checkpoint's evaluators through shared records and confound the +// curve. The rule is a steer, not a guarantee — MemoryWrites on each attempt is +// the audit-side validity signal that catches a violation. const evalScaffold = `You are a data analyst agent connected to a data platform over MCP. Rules: - Ground every answer in tool results and in the knowledge the platform surfaces (search results, catalog descriptions, knowledge pages); do not answer from prior knowledge about any specific dataset. - Use the search tool to discover context and data before querying. +- This is an evaluation of EXISTING platform knowledge: do not save, capture, or update memories or knowledge during this session. - When you have the answer, end your reply with a single line: "FINAL ANSWER: ".` // Per-grading-kind answer format rules, matching the deterministic graders' @@ -98,6 +104,7 @@ type episodeResult struct { toolCalls int toolErrors int searchCalled bool + memoryWrites int wallMS int64 usage llm.Usage audit auditapi.Metrics @@ -105,6 +112,61 @@ type episodeResult struct { err string } +// isMemoryWriteCall reports whether a tool call is a memory WRITE an eval +// session must never make: an evaluator that saves a memory could surface it +// to later checkpoints' evaluators through shared records, contaminating the +// curve with self-taught knowledge. memory_capture always writes; memory_manage +// writes only for its mutating commands (update, forget, consolidate) — its +// list/review commands are read-only and permitted by the eval scaffold, so +// counting them would falsely flag a clean run. The claude-cli transcript +// records the namespaced form (mcp____memory_capture), so the name is +// matched on its final "__"-separated segment. +func isMemoryWriteCall(c llm.ToolCall) bool { + name := c.Name + if i := strings.LastIndex(name, "__"); i >= 0 { + name = name[i+2:] + } + switch name { + case "memory_capture": + return true + case "memory_manage": + cmd, _ := c.Args["command"].(string) + return cmd == "update" || cmd == "forget" || cmd == "consolidate" + } + return false +} + +// countMemoryWrites derives the evaluator no-self-teach validity signal from +// the transcript: the number of EXECUTED memory-write tool calls. Deriving it +// from the transcript (rather than hooking each execution path) keeps the loop +// and claude-cli paths on one definition, mirroring the lifecycle +// instrumentation. A write that never landed does not count: a budget-refused +// call never ran (its paired result is the refusal sentinel), and an error +// result (a platform refusal or a handler failure) wrote no record that a +// later checkpoint could read. +func countMemoryWrites(msgs []llm.Message) int { + writeIDs := map[string]bool{} + for _, m := range msgs { + for _, c := range m.ToolCalls { + if isMemoryWriteCall(c) { + writeIDs[c.ID] = true + } + } + } + if len(writeIDs) == 0 { + return 0 + } + writes := 0 + for _, m := range msgs { + for _, r := range m.ToolResults { + if writeIDs[r.CallID] && !r.IsError && r.Text != agent.BudgetRefusalText { + writes++ + } + } + } + return writes +} + // runEpisode drives one fresh MCP session end to end: authenticate as the pool // identity, mint the handle, run the agent loop against the a3 tool surface, and // read the audit trail back (best effort — lesson state comes from the knowledge @@ -155,7 +217,7 @@ func (e *runEnv) runEpisode(ctx context.Context, spec episodeSpec) episodeResult indeterminate++ return llm.ToolResult{Text: "transport error: " + r.TransportErr.Error(), IsError: true} } - if !preAuditRefusal(r.ErrorCode) { + if !mcpc.PreAuditRefusal(r.ErrorCode) { audited++ } return llm.ToolResult{Text: r.Text, IsError: r.ToolErr} @@ -170,6 +232,7 @@ func (e *runEnv) runEpisode(ctx context.Context, spec episodeSpec) episodeResult res.toolErrors = result.ToolErrors res.usage = result.Usage res.finalAnswer = result.FinalAnswer + res.memoryWrites = countMemoryWrites(result.Transcript) e.writeTranscript(spec, result.Transcript) if runErr != nil { res.err = fmt.Sprintf("agent loop: %v", runErr) @@ -203,6 +266,7 @@ func (e *runEnv) runClaudeCLIEpisode(ctx context.Context, spec episodeSpec) epis res.toolCalls = cres.MCPCalls res.toolErrors = cres.ToolErrors res.searchCalled = cres.SearchCalled + res.memoryWrites = countMemoryWrites(cres.Transcript) res.usage = cres.Usage res.finalAnswer = cres.FinalText e.recordPlatformVersion(cres.PlatformVersion) @@ -274,15 +338,3 @@ type transcriptFile struct { Email string `json:"email"` Transcript []llm.Message `json:"transcript"` } - -// preAuditRefusal reports whether a structured error code marks a platform -// refusal issued outer to the audit middleware (so it leaves no audit row). -// Mirrors the S1-S3 pipeline and S5 lifecycle classification. -func preAuditRefusal(code string) bool { - switch code { - case "unauthenticated", "unauthorized", "session_required", "session_expired", - "search_required", "setup_required", "rate_limited": - return true - } - return false -} diff --git a/bench/internal/coldstart/episode_test.go b/bench/internal/coldstart/episode_test.go new file mode 100644 index 00000000..a0c516b4 --- /dev/null +++ b/bench/internal/coldstart/episode_test.go @@ -0,0 +1,71 @@ +package coldstart + +import ( + "testing" + + "github.com/txn2/mcp-data-platform/bench/internal/agent" + "github.com/txn2/mcp-data-platform/bench/internal/llm" +) + +// TestCountMemoryWrites pins the evaluator no-self-teach validity signal: +// executed memory writes count (including the claude-cli path's namespaced +// form), while budget-refused writes, error results (a refused write landed +// nothing), read-only memory_manage commands, and non-memory tools do not. +func TestCountMemoryWrites(t *testing.T) { + transcript := []llm.Message{ + {Role: "assistant", ToolCalls: []llm.ToolCall{ + {ID: "1", Name: "search"}, + {ID: "2", Name: "memory_capture"}, + {ID: "3", Name: "mcp__bench__memory_manage", Args: map[string]any{"command": "update"}}, + {ID: "4", Name: "memory_capture"}, // budget-refused, never ran + {ID: "5", Name: "memory_capture"}, // platform-refused (error result), wrote nothing + {ID: "6", Name: "memory_manage", Args: map[string]any{"command": "list"}}, // read-only + }}, + {Role: "user", ToolResults: []llm.ToolResult{ + {CallID: "1", Text: "results"}, + {CallID: "2", Text: "captured in-1"}, + {CallID: "3", Text: "updated"}, + {CallID: "4", Text: agent.BudgetRefusalText}, + {CallID: "5", Text: "SESSION_EXPIRED: mint a new handle", IsError: true}, + {CallID: "6", Text: "3 memories"}, + }}, + } + if got := countMemoryWrites(transcript); got != 2 { + t.Errorf("countMemoryWrites = %d, want 2 (executed capture + namespaced manage update)", got) + } + if got := countMemoryWrites(nil); got != 0 { + t.Errorf("countMemoryWrites(nil) = %d, want 0", got) + } + clean := []llm.Message{{Role: "assistant", ToolCalls: []llm.ToolCall{{ID: "1", Name: "search"}}}, + {Role: "user", ToolResults: []llm.ToolResult{{CallID: "1", Text: "ok"}}}} + if got := countMemoryWrites(clean); got != 0 { + t.Errorf("countMemoryWrites(clean) = %d, want 0", got) + } +} + +// TestIsMemoryWriteCall covers the plain and namespaced tool-name forms and +// the memory_manage command split: only mutating commands are writes. +func TestIsMemoryWriteCall(t *testing.T) { + cases := []struct { + call llm.ToolCall + want bool + }{ + {llm.ToolCall{Name: "memory_capture"}, true}, + {llm.ToolCall{Name: "mcp__bench__memory_capture"}, true}, + {llm.ToolCall{Name: "memory_manage", Args: map[string]any{"command": "update"}}, true}, + {llm.ToolCall{Name: "memory_manage", Args: map[string]any{"command": "forget"}}, true}, + {llm.ToolCall{Name: "mcp__bench__memory_manage", Args: map[string]any{"command": "consolidate"}}, true}, + {llm.ToolCall{Name: "memory_manage", Args: map[string]any{"command": "list"}}, false}, + {llm.ToolCall{Name: "memory_manage", Args: map[string]any{"command": "review_stale"}}, false}, + {llm.ToolCall{Name: "memory_manage", Args: map[string]any{"command": "review_duplicates"}}, false}, + {llm.ToolCall{Name: "memory_manage"}, false}, // no command = help, read-only + {llm.ToolCall{Name: "search"}, false}, + {llm.ToolCall{Name: "mcp__bench__search"}, false}, + {llm.ToolCall{Name: "apply_knowledge"}, false}, + } + for _, tc := range cases { + if got := isMemoryWriteCall(tc.call); got != tc.want { + t.Errorf("isMemoryWriteCall(%s %v) = %v, want %v", tc.call.Name, tc.call.Args, got, tc.want) + } + } +} diff --git a/bench/internal/coldstart/preflight.go b/bench/internal/coldstart/preflight.go new file mode 100644 index 00000000..1fd0511f --- /dev/null +++ b/bench/internal/coldstart/preflight.go @@ -0,0 +1,166 @@ +package coldstart + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + "github.com/txn2/mcp-data-platform/bench/internal/curriculum" + "github.com/txn2/mcp-data-platform/bench/internal/lifecycleapi" + "github.com/txn2/mcp-data-platform/bench/internal/mcpc" +) + +// getEntityToolName reads an entity's effective metadata through the platform +// (the same read path evaluators see), so the description check covers both +// contamination sources: the read prefers the editable description when +// non-empty, and apply_knowledge promotions write exactly that editable aspect. +const getEntityToolName = "datahub_get_entity" + +// preflightRemediation tells the operator how to restore a clean baseline. The +// two states reset differently, and the DataHub half CANNOT be fixed by +// re-seeding: promotions write the editableDatasetProperties aspect (and a full +// a2 seed leaves editableSchemaMetadata column docs, tags, and deprecation), +// while bench_mces_empty.json upserts only datasetProperties, so re-ingesting +// the empty seed leaves the contamination in place. +const preflightRemediation = "remediation: (a) Postgres state is reset by the bench-cold-start make target's TRUNCATE (search gate, memory records, changesets, knowledge pages); (b) DataHub state requires a FRESH quickstart (`datahub docker nuke`, re-quickstart, then `make bench-seed-datahub-empty`), because re-ingesting the empty seed cannot clear editable aspects left by prior promotions or an a2 seed" + +// preflightEntity is the subset of the datahub_get_entity result the preflight +// inspects: the effective description plus the tag and deprecation aspects an +// a2 seed would have left behind. +type preflightEntity struct { + Description string `json:"description"` + Tags []json.RawMessage `json:"tags"` + Deprecation *struct { + Deprecated bool `json:"deprecated"` + } `json:"deprecation"` +} + +// preflight refuses to run against a contaminated platform, before any LLM +// episode is spent. An empty baseline cannot be restored by re-seeding (see +// preflightRemediation), and nothing downstream would catch the contamination — +// it would just publish a silently wrong curve (an inflated baseline, an +// attenuated lift). Checked, in order: every lesson entity is undocumented +// (empty effective description, no tags, no deprecation), no insight of ANY +// status exists at all, and no knowledge page exists at all — the cold-start +// contract is an EMPTY enrichment layer, so the store-level checks assert +// emptiness outright rather than hunting curriculum-shaped leftovers (an S5 +// run's lc-* pages or evaluator-identity insights would otherwise slip through). +func (e *runEnv) preflight(ctx context.Context, cur curriculum.Curriculum) error { + if err := e.preflightEntities(ctx, cur); err != nil { + return err + } + if err := e.preflightInsights(ctx); err != nil { + return err + } + return e.preflightPages(ctx) +} + +// preflightEntities checks every distinct lesson entity reads back +// undocumented. The per-entity reads are independent, so they run concurrently. +func (e *runEnv) preflightEntities(ctx context.Context, cur curriculum.Curriculum) error { + session, handle, err := e.adminSession(ctx) + if err != nil { + return fmt.Errorf("preflight admin session: %w", err) + } + var urns []string + seen := map[string]bool{} + for _, l := range cur.Lessons { + if !seen[l.EntityURN] { + seen[l.EntityURN] = true + urns = append(urns, l.EntityURN) + } + } + var wg sync.WaitGroup + errs := make([]error, len(urns)) + for i, urn := range urns { + wg.Go(func() { + errs[i] = e.checkEntityClean(ctx, session, handle, urn) + }) + } + wg.Wait() + return errors.Join(errs...) +} + +// checkEntityClean reads one entity through the platform and reports the first +// contamination it carries. +func (e *runEnv) checkEntityClean(ctx context.Context, session *mcp.ClientSession, handle, urn string) error { + res := mcpc.Call(ctx, session, getEntityToolName, map[string]any{"urn": urn}, handle) + if res.TransportErr != nil { + return fmt.Errorf("preflight %s(%s): %w", getEntityToolName, urn, res.TransportErr) + } + if res.ToolErr { + return fmt.Errorf("preflight %s(%s) failed: %.300s (an unreadable baseline cannot be verified clean)", getEntityToolName, urn, res.Text) + } + entity, err := parsePreflightEntity(res.Text) + if err != nil { + return fmt.Errorf("preflight %s(%s): %w", getEntityToolName, urn, err) + } + if reason := entityContamination(entity); reason != "" { + return fmt.Errorf("baseline contaminated: entity %s %s; %s", urn, reason, preflightRemediation) + } + return nil +} + +// entityContamination reports why an entity is not an empty-baseline entity, or +// "" when it is clean. +func entityContamination(entity preflightEntity) string { + switch { + case strings.TrimSpace(entity.Description) != "": + return fmt.Sprintf("already carries a description (%.120q), left by a prior run's promotion or an a2 seed", entity.Description) + case len(entity.Tags) > 0: + return "carries tags left by an a2 seed" + case entity.Deprecation != nil && entity.Deprecation.Deprecated: + return "carries a deprecation left by an a2 seed" + } + return "" +} + +// parsePreflightEntity decodes the first JSON value from a tool result's text. +// Enrichment middleware may append context after the entity JSON, so the +// decoder reads one value and ignores the rest. +func parsePreflightEntity(text string) (preflightEntity, error) { + var entity preflightEntity + dec := json.NewDecoder(strings.NewReader(text)) + if err := dec.Decode(&entity); err != nil { + return preflightEntity{}, fmt.Errorf("parse entity result: %w (text: %.200s)", err, text) + } + return entity, nil +} + +// preflightInsights checks the insight store is EMPTY (any capturer, any +// status, any entity). The baseline is an empty enrichment layer, so any +// insight at all — this suite's deterministic teacher identities, an S5 run's +// teachers or learners — is a leftover that would distort the measured state. +// One unfiltered list is both one round-trip and strictly stronger than +// per-pair queries. +func (e *runEnv) preflightInsights(ctx context.Context) error { + insights, err := e.life.ListInsights(ctx, lifecycleapi.InsightFilter{}) + if err != nil { + return fmt.Errorf("preflight list insights: %w", err) + } + if len(insights) > 0 { + return fmt.Errorf("baseline contaminated: %d insight(s) already exist (the cold-start baseline requires an empty insight store); %s", + len(insights), preflightRemediation) + } + return nil +} + +// preflightPages checks no knowledge page exists at all — the run boots with +// BENCH_SEED_PAGES=0, so any page (a curriculum slug, an S5 lc-* definition +// page) is a leftover the eval could read. +func (e *runEnv) preflightPages(ctx context.Context) error { + pages, err := e.life.ListKnowledgePages(ctx) + if err != nil { + return fmt.Errorf("preflight list knowledge pages: %w", err) + } + if len(pages) > 0 { + return fmt.Errorf("baseline contaminated: %d knowledge page(s) already exist (first: %q; the cold-start baseline requires none); %s", + len(pages), pages[0].Slug, preflightRemediation) + } + return nil +} diff --git a/bench/internal/coldstart/report.go b/bench/internal/coldstart/report.go index 34460d31..3a323829 100644 --- a/bench/internal/coldstart/report.go +++ b/bench/internal/coldstart/report.go @@ -39,6 +39,11 @@ type Manifest struct { // K is the number of fresh evaluator identities per checkpoint; each answers // the whole eval set, so a checkpoint's accuracy averages over K x eval-tasks. K int `json:"k"` + // Settle is the promote-to-eval cache-settle window the run was paced with + // (Duration string, e.g. "5m0s"; empty when disabled). Settle pacing affects + // how much of a datahub-sink lesson's lift the next checkpoint can see, so a + // kept result must record it to stay comparable with other runs. + Settle string `json:"settle,omitempty"` } // EpisodeRecord is one teach session's telemetry (the lesson's capture episode). @@ -78,13 +83,18 @@ type LessonRecord struct { // which is excluded from accuracy and reported separately, mirroring the S1-S3 // and S5 pipelines. type EvalAttempt struct { - TaskID string `json:"task_id"` - TrapClasses []string `json:"trap_classes,omitempty"` - Email string `json:"email"` - SessionID string `json:"session_id,omitempty"` - Repeat int `json:"repeat"` - Graded bool `json:"graded"` - Correct bool `json:"correct"` + TaskID string `json:"task_id"` + TrapClasses []string `json:"trap_classes,omitempty"` + Email string `json:"email"` + SessionID string `json:"session_id,omitempty"` + Repeat int `json:"repeat"` + Graded bool `json:"graded"` + Correct bool `json:"correct"` + // MemoryWrites counts memory_capture/memory_manage calls the evaluator made, + // derived from the transcript. Evaluators are instructed never to write + // (self-taught knowledge could surface to later checkpoints and confound the + // curve), so any non-zero value is a validity warning on the run. + MemoryWrites int `json:"memory_writes,omitempty"` FinalAnswer string `json:"final_answer,omitempty"` WallMS int64 `json:"wall_ms"` InputTokens int64 `json:"input_tokens"` @@ -173,6 +183,10 @@ type Metrics struct { Checkpoints int `json:"checkpoints"` EvalTasks int `json:"eval_tasks"` HarnessFailures int `json:"harness_failures"` + // EvalMemoryWrites totals evaluator memory writes across every attempt. Any + // non-zero value flags the curve's validity: an evaluator taught itself + // something a later checkpoint's evaluators may have read. + EvalMemoryWrites int `json:"eval_memory_writes,omitempty"` // Token totals across every episode and attempt, so a run self-reports its // cost basis. The cache split lets a cached run's cost be computed from @@ -220,6 +234,7 @@ func (res *Results) Aggregate() { res.Checkpoints[i].aggregate() m.HarnessFailures += res.Checkpoints[i].HarnessFailures for _, a := range res.Checkpoints[i].Attempts { + m.EvalMemoryWrites += a.MemoryWrites m.TotalInputTokens += a.InputTokens m.TotalOutputTokens += a.OutputTokens m.TotalCacheReadTokens += a.CacheReadTokens @@ -298,6 +313,9 @@ func (res *Results) HumanSummary() string { mt := res.Metrics fmt.Fprintf(&b, "lessons %d (captured %d, promoted %d) eval tasks %d checkpoints %d harness failures %d\n", mt.Lessons, mt.LessonsCaptured, mt.LessonsPromoted, mt.EvalTasks, mt.Checkpoints, mt.HarnessFailures) + if mt.EvalMemoryWrites > 0 { + fmt.Fprintf(&b, "WARNING: evaluators performed %d memory write(s); an evaluator taught itself knowledge that later checkpoints may have read, so the curve's validity is suspect\n", mt.EvalMemoryWrites) + } fmt.Fprintf(&b, "tokens: input %d output %d cache read %d cache write %d (apply current model pricing for cost)\n\n", mt.TotalInputTokens, mt.TotalOutputTokens, mt.TotalCacheReadTokens, mt.TotalCacheCreationTokens) diff --git a/bench/internal/coldstart/report_test.go b/bench/internal/coldstart/report_test.go index 9b3c1d04..be6f6c11 100644 --- a/bench/internal/coldstart/report_test.go +++ b/bench/internal/coldstart/report_test.go @@ -123,6 +123,26 @@ func TestHumanSummaryRendersCurve(t *testing.T) { t.Errorf("summary missing %q\n%s", want, out) } } + // No evaluator wrote a memory, so no validity warning appears. + if strings.Contains(out, "memory write") { + t.Errorf("summary warns about memory writes on a clean run\n%s", out) + } +} + +// TestEvalMemoryWritesWarn proves an evaluator memory write is totaled in +// Metrics and surfaces as a validity warning in the human summary: the +// no-self-teach rule is prompt-level, so the report must carry the audit-side +// signal when it is violated. +func TestEvalMemoryWritesWarn(t *testing.T) { + res := sampleResults() + res.Checkpoints[1].Attempts[0].MemoryWrites = 2 + res.Aggregate() + if res.Metrics.EvalMemoryWrites != 2 { + t.Errorf("EvalMemoryWrites = %d, want 2", res.Metrics.EvalMemoryWrites) + } + if !strings.Contains(res.HumanSummary(), "WARNING: evaluators performed 2 memory write(s)") { + t.Errorf("summary missing the memory-write validity warning\n%s", res.HumanSummary()) + } } func TestAggregateEmptyIsSafe(t *testing.T) { diff --git a/bench/internal/coldstart/runner.go b/bench/internal/coldstart/runner.go index 22cb6910..856ab8ec 100644 --- a/bench/internal/coldstart/runner.go +++ b/bench/internal/coldstart/runner.go @@ -18,6 +18,7 @@ import ( "github.com/txn2/mcp-data-platform/bench/internal/mcpc" "github.com/txn2/mcp-data-platform/bench/internal/pool" "github.com/txn2/mcp-data-platform/bench/internal/promote" + "github.com/txn2/mcp-data-platform/bench/internal/protocol" "github.com/txn2/mcp-data-platform/bench/internal/target" "github.com/txn2/mcp-data-platform/bench/internal/task" ) @@ -53,6 +54,16 @@ type Options struct { // distinct identities so the curve measures promoted (shared) knowledge, never // an evaluator's own capture. IdentityKeys int + // Settle is the pause between a successful promote and the following eval + // checkpoint. The a3 arm caches table context with a 5m TTL, so a cache entry + // populated by the PREVIOUS checkpoint's evaluators can serve the stale + // pre-promotion description to the next checkpoint's evaluators, + // nondeterministically attenuating the datahub-sink lift; waiting out the TTL + // removes that pacing dependence. The scripted smoke sets it to zero. + Settle time.Duration + // SettleSleep overrides the settle pause's sleeper. Tests inject a recorder + // so no test real-sleeps; nil uses a context-aware real sleep. + SettleSleep func(ctx context.Context, d time.Duration) error // OnCheckpoint, if set, is called after every checkpoint with the aggregated // results so far. benchrun wires it to flush the results file, so a run that // spends real API budget always leaves every completed checkpoint on disk. @@ -79,6 +90,7 @@ func Run(ctx context.Context, opts Options) (*Results, error) { Arm: opts.Arm, LLMProvider: opts.LLMProvider, Seed: gen.Seed, CurriculumID: cur.ID, CurriculumHash: curriculum.Hash([]curriculum.Curriculum{cur}), EvalSuite: cur.EvalSuite, TaskSetHash: task.Hash(evalTasks), K: opts.K, + Settle: settleLabel(opts.Settle), }} life := lifecycleapi.New(opts.Target.BaseURL, opts.Target.HTTPClient(opts.HTTPTimeout)) env := &runEnv{ @@ -90,6 +102,13 @@ func Run(ctx context.Context, opts Options) (*Results, error) { } defer env.closeAdmin() + // Refuse a contaminated baseline before any LLM episode is spent: a + // non-empty starting state cannot be restored by re-seeding and would + // publish a silently wrong curve (see preflight). + if err := env.preflight(ctx, cur); err != nil { + return nil, err + } + failures := env.run(ctx, cur, evalTasks, res) res.Manifest.FinishedAt = time.Now().UTC() res.Manifest.PlatformVersion = env.platformVersion @@ -212,6 +231,11 @@ func (e *runEnv) run(ctx context.Context, cur curriculum.Curriculum, evalTasks [ promoted++ } res.Lessons = append(res.Lessons, lr) + // Flush the paid-for lesson record BEFORE the settle pause: an + // interruption during the settle (or the following checkpoint) must never + // discard a completed teach episode and its promote outcome. + e.flush(res) + e.settleAfterPromote(ctx, lr) cp := e.evalCheckpoint(ctx, i+1, lesson, evalTasks, n, promoted) res.Checkpoints = append(res.Checkpoints, cp) @@ -221,12 +245,66 @@ func (e *runEnv) run(ctx context.Context, cur curriculum.Curriculum, evalTasks [ return failures } +// settleAfterPromote waits out the semantic-cache TTL between a successful +// promote and the following eval checkpoint, so a table-context entry cached by +// the previous checkpoint's evaluators can never serve the stale pre-promotion +// description to the next ones. Only a datahub-sink promote needs it: the cache +// holds DataHub table context, while knowledge-page hits are served live from +// the portal store with no TTL. A lesson that did not promote changed nothing. +// Every skip is logged so a paced run's timeline stays auditable. +func (e *runEnv) settleAfterPromote(ctx context.Context, lr LessonRecord) { + if e.opts.Settle <= 0 { + return + } + if !boolTrue(lr.Promoted) { + e.log.Info("cold-start settle skipped: lesson did not promote, the enrichment layer is unchanged", + "lesson", lr.LessonID, "settle", e.opts.Settle) + return + } + if lr.Sink != protocol.SinkDataHub { + e.log.Info("cold-start settle skipped: page-sink promotes are served live from the portal store, nothing is cached", + "lesson", lr.LessonID, "settle", e.opts.Settle) + return + } + e.log.Info("cold-start settle: waiting out the semantic-cache TTL before the next eval checkpoint", + "lesson", lr.LessonID, "settle", e.opts.Settle) + if err := e.sleep(ctx, e.opts.Settle); err != nil { + e.log.Warn("cold-start settle interrupted", "error", err) + } +} + +// settleLabel renders the settle window for the manifest ("" when disabled). +func settleLabel(d time.Duration) string { + if d <= 0 { + return "" + } + return d.String() +} + +// sleep pauses for d respecting ctx cancellation, via the injected test sleeper +// when set. +func (e *runEnv) sleep(ctx context.Context, d time.Duration) error { + if e.opts.SettleSleep != nil { + return e.opts.SettleSleep(ctx, d) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} + // teachAndPromote runs the lesson's teach episode, verifies capture through the // insights API, and promotes the insight to its sink. A harness failure lands in // Error; a missed capture or refused apply is a measured miss (Captured/Promoted // false, no error). func (e *runEnv) teachAndPromote(ctx context.Context, lesson curriculum.Lesson, seq int) LessonRecord { lr := LessonRecord{LessonID: lesson.ID, Title: lesson.Title, TrapClass: lesson.TrapClass, Sink: lesson.Sink} + // The teach start (minus a skew margin) bounds capture verification to THIS + // run: teacher identities and URNs are deterministic, so an unbounded read + // could match a pending insight left by an interrupted prior run. + teachStart := time.Now() rec := e.runEpisode(ctx, episodeSpec{ stage: StageTeach, unitID: lesson.ID, seq: seq, prompt: lesson.Teach.Prompt, system: teachScaffold, budget: lesson.BudgetToolCalls, @@ -236,7 +314,8 @@ func (e *runEnv) teachAndPromote(ctx context.Context, lesson curriculum.Lesson, lr.Error = rec.err return lr } - insight, err := promote.WaitForInsight(ctx, e.life, pool.Email(seq), lesson.EntityURN, e.opts.AuditTimeout, insightPollInterval) + since := teachStart.Add(-promote.CaptureSkewMargin) + insight, err := promote.WaitForInsight(ctx, e.life, pool.Email(seq), lesson.EntityURN, since, e.opts.AuditTimeout, insightPollInterval) if err != nil { lr.Error = "capture verify: " + err.Error() return lr @@ -302,7 +381,7 @@ func (e *runEnv) evalAttempt(ctx context.Context, t task.Task, seq, repeat int) }) att := EvalAttempt{ TaskID: t.ID, TrapClasses: t.TrapClasses, Email: rec.email, SessionID: rec.sessionID, - Repeat: repeat, FinalAnswer: rec.finalAnswer, WallMS: rec.wallMS, + Repeat: repeat, MemoryWrites: rec.memoryWrites, FinalAnswer: rec.finalAnswer, WallMS: rec.wallMS, InputTokens: rec.usage.InputTokens, OutputTokens: rec.usage.OutputTokens, CacheReadTokens: rec.usage.CacheReadInputTokens, CacheCreationTokens: rec.usage.CacheCreationInputTokens, Audit: rec.audit, diff --git a/bench/internal/coldstart/runner_test.go b/bench/internal/coldstart/runner_test.go index 181c7690..a824f2b9 100644 --- a/bench/internal/coldstart/runner_test.go +++ b/bench/internal/coldstart/runner_test.go @@ -3,6 +3,7 @@ package coldstart import ( "context" "encoding/json" + "errors" "fmt" "io" "log/slog" @@ -28,6 +29,8 @@ import ( "github.com/txn2/mcp-data-platform/bench/internal/curriculum" "github.com/txn2/mcp-data-platform/bench/internal/lifecycleapi" "github.com/txn2/mcp-data-platform/bench/internal/llm" + "github.com/txn2/mcp-data-platform/bench/internal/pool" + "github.com/txn2/mcp-data-platform/bench/internal/promote" "github.com/txn2/mcp-data-platform/bench/internal/protocol" "github.com/txn2/mcp-data-platform/bench/internal/target" "github.com/txn2/mcp-data-platform/bench/internal/task" @@ -51,17 +54,27 @@ type fakePlatform struct { changesets []lifecycleapi.Changeset events []auditapi.Event applied map[string]bool // trap class -> promoted - httpSrv *httptest.Server + // Preflight contamination knobs (all empty = clean baseline). + entityDesc map[string]string // urn -> pre-existing description + pages []lifecycleapi.KnowledgePage // pre-existing knowledge pages + // Failure knobs, set before the run starts (all zero = healthy platform). + infoFailAfter int64 // platform_info mints fail after this many (0 = never) + insightsFail bool // insights list returns 500 + approveFail bool // insight status PUT returns 500 + applyRefuse bool // apply_knowledge returns a plain tool error + httpSrv *httptest.Server } func newFakePlatform(t *testing.T) *fakePlatform { t.Helper() - fp := &fakePlatform{applied: map[string]bool{}} + fp := &fakePlatform{applied: map[string]bool{}, entityDesc: map[string]string{}} server := mcp.NewServer(&mcp.Implementation{Name: "fake-coldstart", Version: "fake-1.0.0"}, nil) fp.addPlatformInfo(server) fp.addMemoryCapture(server) fp.addSearch(server) fp.addApplyKnowledge(server) + fp.addGetEntity(server) + fp.addGatedQuery(server) mcpHandler := mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return server }, nil) mux := http.NewServeMux() @@ -70,6 +83,7 @@ func newFakePlatform(t *testing.T) *fakePlatform { mux.HandleFunc("PUT /api/v1/admin/knowledge/insights/{id}/status", fp.putStatus) mux.HandleFunc("GET /api/v1/admin/knowledge/changesets", fp.listChangesets) mux.HandleFunc("GET /api/v1/admin/knowledge/changesets/{id}", fp.getChangeset) + mux.HandleFunc("GET /api/v1/portal/knowledge-pages", fp.listPages) mux.HandleFunc("/api/v1/admin/audit/events", fp.serveAudit) mux.Handle("/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { ctx := context.WithValue(r.Context(), authCtxKey{}, r.Header.Get("Authorization")) @@ -105,12 +119,30 @@ func sessionSchema(extra map[string]*jsonschema.Schema) *jsonschema.Schema { func (fp *fakePlatform) addPlatformInfo(server *mcp.Server) { mcp.AddTool(server, &mcp.Tool{Name: "platform_info", Description: "orientation"}, func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { - payload := map[string]any{"session_id": fmt.Sprintf("dps_%d", fp.minted.Add(1)), "version": "fake-1.0.0"} + n := fp.minted.Add(1) + if fp.infoFailAfter > 0 && n > fp.infoFailAfter { + return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "mint refused"}}}, nil, nil + } + payload := map[string]any{"session_id": fmt.Sprintf("dps_%d", n), "version": "fake-1.0.0"} raw, _ := json.Marshal(payload) return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: string(raw)}}, StructuredContent: payload}, nil, nil }) } +// addGatedQuery models the platform's search-first gate: a query tool that +// always refuses with the structured pre-audit error contract (no audit row). +func (fp *fakePlatform) addGatedQuery(server *mcp.Server) { + schema := sessionSchema(map[string]*jsonschema.Schema{"sql": {Type: "string"}}) + mcp.AddTool(server, &mcp.Tool{Name: "trino_query", Description: "query", InputSchema: schema}, + func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { + return &mcp.CallToolResult{ + IsError: true, + Content: []mcp.Content{&mcp.TextContent{Text: "SEARCH_REQUIRED: call search first"}}, + StructuredContent: map[string]any{"error": map[string]any{"code": "search_required"}}, + }, nil, nil + }) +} + // addMemoryCapture records a pending insight owned by the caller. The insight // text is the fact's trap class, so apply can mark that class promoted. func (fp *fakePlatform) addMemoryCapture(server *mcp.Server) { @@ -160,6 +192,9 @@ func (fp *fakePlatform) addApplyKnowledge(server *mcp.Server) { }) mcp.AddTool(server, &mcp.Tool{Name: "apply_knowledge", Description: "promote", InputSchema: schema}, func(ctx context.Context, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) { + if fp.applyRefuse { + return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "change declined"}}}, nil, nil + } fp.mu.Lock() defer fp.mu.Unlock() ids := stringSlice(args["insight_ids"]) @@ -181,6 +216,30 @@ func (fp *fakePlatform) addApplyKnowledge(server *mcp.Server) { }) } +// addGetEntity serves the preflight's baseline read: an entity's effective +// description (the entityDesc knob models a prior run's promotion or an a2 +// seed; empty is the clean baseline). +func (fp *fakePlatform) addGetEntity(server *mcp.Server) { + schema := sessionSchema(map[string]*jsonschema.Schema{"urn": {Type: "string"}}) + mcp.AddTool(server, &mcp.Tool{Name: "datahub_get_entity", Description: "entity metadata", InputSchema: schema}, + func(_ context.Context, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) { + urn, _ := args["urn"].(string) + fp.mu.Lock() + desc := fp.entityDesc[urn] + fp.mu.Unlock() + raw, _ := json.Marshal(map[string]any{"urn": urn, "type": "DATASET", "description": desc}) + return okResult(string(raw)), nil, nil + }) +} + +// listPages serves the portal knowledge-pages list the preflight scans. +func (fp *fakePlatform) listPages(w http.ResponseWriter, _ *http.Request) { + fp.mu.Lock() + pages := append([]lifecycleapi.KnowledgePage{}, fp.pages...) + fp.mu.Unlock() + writeJSON(w, map[string]any{"pages": pages, "total": len(pages)}) +} + // recordLocked appends an audit row; a call carries enrichment once any // knowledge has been promoted. The caller holds fp.mu. func (fp *fakePlatform) recordLocked(args map[string]any, tool string) { @@ -195,6 +254,10 @@ func (fp *fakePlatform) recordLocked(args map[string]any, tool string) { } func (fp *fakePlatform) listInsights(w http.ResponseWriter, r *http.Request) { + if fp.insightsFail { + http.Error(w, "insights unavailable", http.StatusInternalServerError) + return + } q := r.URL.Query() fp.mu.Lock() defer fp.mu.Unlock() @@ -227,6 +290,10 @@ func (fp *fakePlatform) getInsight(w http.ResponseWriter, r *http.Request) { } func (fp *fakePlatform) putStatus(w http.ResponseWriter, r *http.Request) { + if fp.approveFail { + http.Error(w, "status update unavailable", http.StatusInternalServerError) + return + } var body struct { Status string `json:"status"` } @@ -360,7 +427,7 @@ func testCurriculum() curriculum.Curriculum { Sink: protocol.SinkDataHub, BudgetToolCalls: 5, Teach: protocol.TeachStage{Prompt: "remember cents"}}, {ID: "cs-net", Title: "net", TrapClass: "net_revenue", Fact: "net", EntityURN: ordersURN, Sink: protocol.SinkKnowledgePage, BudgetToolCalls: 5, - Page: &protocol.PagePayload{Slug: "net", Title: "Net", Body: "net policy"}, + Page: &protocol.PagePayload{Slug: "net", Title: "Net", Summary: "net = amount - discount", Body: "net policy"}, Teach: protocol.TeachStage{Prompt: "remember net"}}, }, } @@ -446,15 +513,22 @@ func TestColdStartCurveClimbs(t *testing.T) { fp := newFakePlatform(t) opts := testOptions(fp, curDir, tasksDir, testFactory(cur, tasks)) opts.TranscriptDir = t.TempDir() - flushes := 0 - opts.OnCheckpoint = func(*Results) { flushes++ } + // Every append is flushed before the next long-running step (settle, eval), + // so an interruption never discards a completed checkpoint OR a completed + // teach+promote lesson. For 2 lessons the (lessons, checkpoints) snapshot + // sequence is pinned: baseline cp, lesson 1, cp 1, lesson 2, cp 2. + var flushed [][2]int + opts.OnCheckpoint = func(r *Results) { + flushed = append(flushed, [2]int{len(r.Lessons), len(r.Checkpoints)}) + } res, err := Run(context.Background(), opts) if err != nil { t.Fatalf("run: %v", err) } - if flushes != 3 { - t.Errorf("OnCheckpoint fired %d times, want one per checkpoint (3)", flushes) + wantFlushed := [][2]int{{0, 1}, {1, 1}, {1, 2}, {2, 2}, {2, 3}} + if fmt.Sprint(flushed) != fmt.Sprint(wantFlushed) { + t.Errorf("flush snapshots (lessons, checkpoints) = %v, want %v (flush-after-append ordering broken)", flushed, wantFlushed) } if entries, _ := os.ReadDir(opts.TranscriptDir); len(entries) == 0 { t.Error("expected per-episode transcripts to be written") @@ -483,21 +557,455 @@ func TestColdStartCurveClimbs(t *testing.T) { assertDistinctIdentities(t, res) } +// assertDistinctIdentities checks the run's identity discipline: no evaluator +// reuses a teacher identity, and every evaluator identity is globally unique +// across ALL checkpoints (scoped to its one checkpoint+repeat) — a reused +// evaluator would carry its own discovery scope and memories forward, +// contaminating a later checkpoint's measurement. func assertDistinctIdentities(t *testing.T, res *Results) { t.Helper() teachers := map[string]bool{} for _, l := range res.Lessons { teachers[l.Episode.Email] = true } + type slot struct{ checkpoint, repeat int } + owner := map[string]slot{} for _, cp := range res.Checkpoints { for _, a := range cp.Attempts { if teachers[a.Email] { t.Errorf("evaluator %s reused a teacher identity", a.Email) } + s := slot{cp.Index, a.Repeat} + if prev, ok := owner[a.Email]; ok && prev != s { + t.Errorf("evaluator %s used at checkpoint %d repeat %d AND checkpoint %d repeat %d; evaluator identities must be globally unique", + a.Email, prev.checkpoint, prev.repeat, s.checkpoint, s.repeat) + } + owner[a.Email] = s + } + } +} + +// TestIdentitySequences pins the identity math: teachers occupy 1..n, and every +// checkpoint's evaluators start above them and never collide across +// checkpoints or repeats. The exact values guard the arithmetic a refactor +// could silently shift (which would reuse identities and contaminate a run). +func TestIdentitySequences(t *testing.T) { + cases := []struct { + name string + got int + want int + }{ + {"teacherSeq(0)", teacherSeq(0), 1}, + {"teacherSeq(5)", teacherSeq(5), 6}, + {"evaluatorSeq(0,1,6,1)", evaluatorSeq(0, 1, 6, 1), 7}, + {"evaluatorSeq(6,1,6,1)", evaluatorSeq(6, 1, 6, 1), 13}, + {"evaluatorSeq(6,3,6,3)", evaluatorSeq(6, 3, 6, 3), 27}, + {"maxIdentitySeq(6,1)", maxIdentitySeq(6, 1), 13}, + {"maxIdentitySeq(6,3)", maxIdentitySeq(6, 3), 27}, + } + for _, tc := range cases { + if tc.got != tc.want { + t.Errorf("%s = %d, want %d", tc.name, tc.got, tc.want) } } } +// countingFactory wraps a factory and counts adapter builds, so a test can +// assert a refused run spent no episode. +func countingFactory(inner AdapterFactory, calls *atomic.Int64) AdapterFactory { + return func(unitID, stage string) (llm.Adapter, error) { + calls.Add(1) + return inner(unitID, stage) + } +} + +// TestPreflightRefusesContaminatedBaseline proves each contamination source — +// a pre-existing entity description, a leftover insight on a (teacher, URN) +// pair, a knowledge page with a curriculum slug — aborts the run before any +// adapter (LLM episode) is built, with the remediation in the error. +func TestPreflightRefusesContaminatedBaseline(t *testing.T) { + cur := testCurriculum() + tasks := []task.Task{numericTask("s3-units-a", "units_cents", 100)} + cases := map[string]func(fp *fakePlatform){ + "entity description": func(fp *fakePlatform) { + fp.entityDesc[ordersURN] = "Amounts are integer cents." + }, + "leftover insight": func(fp *fakePlatform) { + // The baseline requires an EMPTY insight store: any leftover trips the + // preflight, even one captured by a non-curriculum identity (an S5 + // run's teacher) on a non-curriculum anchor. + fp.insights = append(fp.insights, lifecycleapi.Insight{ + ID: "in-old", CreatedAt: time.Unix(1, 0), CapturedBy: "bench-agent-042@apikey.local", + InsightText: "old", Status: "applied", EntityURNs: []string{"urn:li:dataset:other"}, + }) + }, + "leftover knowledge page": func(fp *fakePlatform) { + // Any page trips it, curriculum slug or not (e.g. an S5 lc-* page). + fp.pages = append(fp.pages, lifecycleapi.KnowledgePage{ID: "kp-1", Slug: "focus-region-definition"}) + }, + } + for name, contaminate := range cases { + t.Run(name, func(t *testing.T) { + curDir, tasksDir := writeFixtures(t, cur, tasks) + fp := newFakePlatform(t) + contaminate(fp) + var adapterCalls atomic.Int64 + opts := testOptions(fp, curDir, tasksDir, countingFactory(testFactory(cur, tasks), &adapterCalls)) + res, err := Run(context.Background(), opts) + if err == nil { + t.Fatal("run accepted a contaminated baseline") + } + if !strings.Contains(err.Error(), "remediation") { + t.Errorf("preflight error missing remediation guidance: %v", err) + } + if res != nil { + t.Errorf("a refused run must not return results, got %+v", res) + } + if n := adapterCalls.Load(); n != 0 { + t.Errorf("preflight refusal built %d adapter(s); it must abort before any episode is spent", n) + } + }) + } +} + +// TestEntityContamination covers the aspect checks the integration variants do +// not exercise: tags and deprecation left by an a2 seed, and a clean entity. +func TestEntityContamination(t *testing.T) { + if got := entityContamination(preflightEntity{}); got != "" { + t.Errorf("clean entity flagged: %s", got) + } + if got := entityContamination(preflightEntity{Description: " "}); got != "" { + t.Errorf("whitespace-only description flagged: %s", got) + } + if got := entityContamination(preflightEntity{Description: "docs"}); got == "" { + t.Error("description contamination not flagged") + } + if got := entityContamination(preflightEntity{Tags: []json.RawMessage{[]byte(`{"name":"pii"}`)}}); got == "" { + t.Error("tag contamination not flagged") + } + deprecated := preflightEntity{} + deprecated.Deprecation = &struct { + Deprecated bool `json:"deprecated"` + }{Deprecated: true} + if got := entityContamination(deprecated); got == "" { + t.Error("deprecation contamination not flagged") + } +} + +// TestParsePreflightEntityIgnoresTrailingText proves the entity JSON is parsed +// even when enrichment middleware appends context after it. +func TestParsePreflightEntityIgnoresTrailingText(t *testing.T) { + got, err := parsePreflightEntity(`{"description":"d"}` + "\n--- Semantic Context ---\nmore text") + if err != nil { + t.Fatalf("parse: %v", err) + } + if got.Description != "d" { + t.Errorf("description = %q, want d", got.Description) + } + if _, err := parsePreflightEntity("not json"); err == nil { + t.Error("non-JSON result must error") + } +} + +// TestSettleRunsBetweenPromoteAndEval proves the settle pause runs after a +// successful DATAHUB-sink promote and before the following eval checkpoint, +// with the configured duration — and is skipped for a page-sink promote (page +// hits are served live from the portal store; only DataHub table context is +// cached). The injected sleeper records instead of sleeping, so no test +// real-sleeps. +func TestSettleRunsBetweenPromoteAndEval(t *testing.T) { + cur := testCurriculum() // lesson 1 datahub sink, lesson 2 page sink + tasks := []task.Task{ + numericTask("s3-units-a", "units_cents", 100), + numericTask("s3-net-a", "net_revenue", 50), + } + curDir, tasksDir := writeFixtures(t, cur, tasks) + fp := newFakePlatform(t) + + // Sequence log: the factory records episode starts, the sleeper records + // settles, so ordering (promote -> settle -> eval) is assertable. + var mu sync.Mutex + var events []string + factory := func(unitID, stage string) (llm.Adapter, error) { + mu.Lock() + events = append(events, stage+":"+unitID) + mu.Unlock() + return testFactory(cur, tasks)(unitID, stage) + } + opts := testOptions(fp, curDir, tasksDir, factory) + opts.Settle = 5 * time.Minute + var slept []time.Duration + opts.SettleSleep = func(_ context.Context, d time.Duration) error { + mu.Lock() + events = append(events, "settle") + slept = append(slept, d) + mu.Unlock() + return nil + } + res, err := Run(context.Background(), opts) + if err != nil { + t.Fatalf("run: %v", err) + } + // Both lessons promote, but only the datahub-sink lesson settles. + if len(slept) != 1 || slept[0] != 5*time.Minute { + t.Fatalf("settle sleeps = %v, want one of 5m (datahub-sink lesson only)", slept) + } + if res.Manifest.Settle != "5m0s" { + t.Errorf("manifest settle = %q, want 5m0s (pacing must be recorded on kept results)", res.Manifest.Settle) + } + assertSettleOrdering(t, events) +} + +// assertSettleOrdering checks every settle event lands after a teach episode +// and before the next eval episode. +func assertSettleOrdering(t *testing.T, events []string) { + t.Helper() + for i, ev := range events { + if ev != "settle" { + continue + } + if i == 0 || !strings.HasPrefix(events[i-1], StageTeach+":") { + t.Errorf("settle at %d not immediately after a teach episode: %v", i, events) + } + if i+1 >= len(events) || !strings.HasPrefix(events[i+1], StageEval+":") { + t.Errorf("settle at %d not immediately before an eval episode: %v", i, events) + } + } +} + +// TestSleepRespectsContext covers the real (non-injected) sleeper: it completes +// a tiny pause and aborts immediately on a canceled context, so a settle +// window never blocks an interrupted run. +func TestSleepRespectsContext(t *testing.T) { + e := &runEnv{opts: Options{}} + if err := e.sleep(context.Background(), time.Millisecond); err != nil { + t.Fatalf("sleep: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := e.sleep(ctx, time.Hour); err == nil { + t.Fatal("a canceled context must interrupt the settle sleep") + } +} + +// TestSettleSkippedForUnpromotedLesson proves an unpromoted lesson triggers no +// settle pause (nothing changed in the enrichment layer). +func TestSettleSkippedForUnpromotedLesson(t *testing.T) { + cur := testCurriculum() + tasks := []task.Task{numericTask("s3-units-a", "units_cents", 100)} + curDir, tasksDir := writeFixtures(t, cur, tasks) + fp := newFakePlatform(t) + + // A factory whose teach episodes never capture: WaitForInsight misses, so no + // lesson promotes (Captured=false is a measured outcome, not an error). + factory := func(unitID, stage string) (llm.Adapter, error) { + if stage == StageTeach { + return llm.NewScripted([]llm.Step{{FinalText: "noted, but not saved"}}), nil + } + return testFactory(cur, tasks)(unitID, stage) + } + opts := testOptions(fp, curDir, tasksDir, factory) + opts.AuditTimeout = 50 * time.Millisecond // capture-verify miss window + opts.Settle = 5 * time.Minute + settles := 0 + opts.SettleSleep = func(context.Context, time.Duration) error { + settles++ + return nil + } + res, err := Run(context.Background(), opts) + if err != nil { + t.Fatalf("run: %v", err) + } + if res.Metrics.LessonsPromoted != 0 { + t.Fatalf("expected no promotions, got %d", res.Metrics.LessonsPromoted) + } + if settles != 0 { + t.Errorf("settle fired %d time(s) for unpromoted lessons; it must be skipped", settles) + } +} + +// newTestEnv builds a runEnv over the fake platform for driving teachAndPromote +// and runEpisode directly (the branch tests below), mirroring Run's wiring. +func newTestEnv(fp *fakePlatform, factory AdapterFactory) *runEnv { + tgt := target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"} + life := lifecycleapi.New(fp.httpSrv.URL, tgt.HTTPClient(5*time.Second)) + return &runEnv{ + opts: Options{ + Target: tgt, HTTPTimeout: 5 * time.Second, Arm: "a3", K: 1, + Factory: factory, AuditTimeout: 300 * time.Millisecond, IdentityKeys: 64, + Log: testLogger(), + }, + log: testLogger(), + audit: auditapi.New(fp.httpSrv.URL, tgt.HTTPClient(5*time.Second)), + life: life, + reviewer: promote.Reviewer{Life: life, Log: testLogger()}, + } +} + +// capturingTeachFactory returns a factory whose teach episodes capture the +// lesson's fact, sufficient for driving teachAndPromote through its +// post-capture branches. +func capturingTeachFactory(l curriculum.Lesson) AdapterFactory { + return func(string, string) (llm.Adapter, error) { + return &knowledgeAdapter{mode: StageTeach, class: l.TrapClass, urn: l.EntityURN}, nil + } +} + +// TestTeachAndPromoteBranches covers every teachAndPromote outcome: harness +// errors carry their stage prefix, while a capture miss and an apply refusal +// are measured outcomes (no error). Each error string is what a paid run's +// lessons[].error would show, so the prefixes are pinned. +func TestTeachAndPromoteBranches(t *testing.T) { + lesson := testCurriculum().Lessons[0] + cases := []struct { + name string + configure func(fp *fakePlatform) AdapterFactory + wantErrPrefix string // "" = measured outcome, no harness error + wantCaptured *bool + wantPromoted *bool + }{ + { + name: "teach harness error", + configure: func(*fakePlatform) AdapterFactory { + return func(string, string) (llm.Adapter, error) { return nil, errors.New("adapter down") } + }, + wantErrPrefix: "build adapter", + }, + { + name: "capture verify error", + configure: func(fp *fakePlatform) AdapterFactory { + fp.insightsFail = true + return capturingTeachFactory(lesson) + }, + wantErrPrefix: "capture verify: ", + }, + { + name: "capture miss is measured", + configure: func(*fakePlatform) AdapterFactory { + // The teach episode never reaches the capture tool. + return func(string, string) (llm.Adapter, error) { + return llm.NewScripted([]llm.Step{{FinalText: "noted"}}), nil + } + }, + wantCaptured: new(false), + }, + { + name: "admin session error", + configure: func(fp *fakePlatform) AdapterFactory { + fp.infoFailAfter = 1 // the teach mint succeeds; the admin mint fails + return capturingTeachFactory(lesson) + }, + wantErrPrefix: "admin session: ", + wantCaptured: new(true), + }, + { + name: "apply error", + configure: func(fp *fakePlatform) AdapterFactory { + fp.approveFail = true + return capturingTeachFactory(lesson) + }, + wantErrPrefix: "promote: ", + wantCaptured: new(true), + }, + { + name: "apply refusal is measured", + configure: func(fp *fakePlatform) AdapterFactory { + fp.applyRefuse = true + return capturingTeachFactory(lesson) + }, + wantCaptured: new(true), + wantPromoted: new(false), + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fp := newFakePlatform(t) + env := newTestEnv(fp, tc.configure(fp)) + defer env.closeAdmin() + lr := env.teachAndPromote(context.Background(), lesson, 1) + if tc.wantErrPrefix == "" && lr.Error != "" { + t.Fatalf("measured outcome produced a harness error: %s", lr.Error) + } + if tc.wantErrPrefix != "" && !strings.Contains(lr.Error, tc.wantErrPrefix) { + t.Fatalf("error = %q, want it to contain %q", lr.Error, tc.wantErrPrefix) + } + assertBoolPtr(t, "captured", lr.Captured, tc.wantCaptured) + assertBoolPtr(t, "promoted", lr.Promoted, tc.wantPromoted) + }) + } +} + +func assertBoolPtr(t *testing.T, field string, got, want *bool) { + t.Helper() + if want == nil { + return + } + if got == nil { + t.Errorf("%s = nil, want %v", field, *want) + return + } + if *got != *want { + t.Errorf("%s = %v, want %v", field, *got, *want) + } +} + +// TestRunSurfacesHarnessFailures proves a run with any harness-level episode +// failure returns a non-nil error (nonzero exit), so a paid run can never +// silently publish a curve with unaccounted episodes. +func TestRunSurfacesHarnessFailures(t *testing.T) { + cur := testCurriculum() + tasks := []task.Task{numericTask("s3-units-a", "units_cents", 100)} + curDir, tasksDir := writeFixtures(t, cur, tasks) + fp := newFakePlatform(t) + inner := testFactory(cur, tasks) + factory := func(unitID, stage string) (llm.Adapter, error) { + if stage == StageEval && unitID == "s3-units-a" { + return nil, errors.New("adapter down") + } + return inner(unitID, stage) + } + res, err := Run(context.Background(), testOptions(fp, curDir, tasksDir, factory)) + if err == nil { + t.Fatal("run with harness failures exited clean") + } + if res == nil || res.Metrics.HarnessFailures == 0 { + t.Fatalf("results must still carry the failed attempts, got %+v", res) + } +} + +// TestEpisodeClassifiesRefusalsAndTransportErrors drives one episode whose +// tools hit all three classifications: an audited call (search), a pre-audit +// structured refusal (the gated trino_query, no audit row expected), and a +// transport-level error (an unknown tool name, indeterminate). The audit +// read-back bounds only converge when the classification is right: counting +// the refusal as audited would demand a row the platform never wrote. +func TestEpisodeClassifiesRefusalsAndTransportErrors(t *testing.T) { + fp := newFakePlatform(t) + steps := []llm.Step{ + {ToolCalls: []llm.ToolCall{{Name: "search", Args: map[string]any{"query": "q"}}}}, + {ToolCalls: []llm.ToolCall{{Name: "trino_query", Args: map[string]any{"sql": "SELECT 1"}}}}, + {ToolCalls: []llm.ToolCall{{Name: "no_such_tool", Args: map[string]any{}}}}, + {FinalText: "FINAL ANSWER: done"}, + } + env := newTestEnv(fp, func(string, string) (llm.Adapter, error) { return llm.NewScripted(steps), nil }) + rec := env.runEpisode(context.Background(), episodeSpec{stage: StageEval, unitID: "s3-x", seq: 9, prompt: "q", system: "sys", budget: 10}) + if rec.err != "" { + t.Fatalf("episode error: %s", rec.err) + } + if !rec.searchCalled { + t.Error("search-called signal not recorded") + } + // Two of the three calls errored (refusal + transport), one succeeded. + if rec.toolErrors != 2 { + t.Errorf("tool errors = %d, want 2", rec.toolErrors) + } + // Only the search is audited: the refusal never reached the audit layer and + // the transport error is indeterminate (bounded above, not below). + if rec.audit.AuditedCalls != 1 { + t.Errorf("audited calls = %d, want 1 (refusal/transport misclassified)", rec.audit.AuditedCalls) + } +} + func TestGuardPoolRejectsSmallPool(t *testing.T) { cur := testCurriculum() if err := guardPool(cur, Options{K: 3, IdentityKeys: 2}); err == nil { @@ -525,13 +1033,30 @@ const claudeEvalStream = `{"type":"system","subtype":"init","mcp_servers":[{"nam // TestClaudeCLIEpisode isolates the claude-cli episode path: a stubbed client // returns a canned eval transcript, and the harness maps the client result, // reads audit best effort by the threaded handle, and writes the transcript. +// The Exec stub also pins credential rotation: the per-episode MCP config must +// authenticate as the episode's pool identity (seq 7 -> testkey-007), because a +// wrong credential would silently collapse every evaluator onto one identity. func TestClaudeCLIEpisode(t *testing.T) { fp := newFakePlatform(t) fp.events = append(fp.events, auditapi.Event{DurationMS: 5, SessionID: "dps_cc_1", ToolName: "search", Success: true}) + var gotCredential string runner, err := claudecli.New(claudecli.Options{ Model: "claude-sonnet-5", - Exec: func(context.Context, claudecli.CommandSpec) ([]byte, []byte, error) { + Exec: func(_ context.Context, spec claudecli.CommandSpec) ([]byte, []byte, error) { + raw, err := os.ReadFile(filepath.Join(spec.Dir, "mcp-config.json")) + if err != nil { + t.Errorf("read per-episode mcp config: %v", err) + } + var cfg struct { + MCPServers map[string]struct { + Headers map[string]string `json:"headers"` + } `json:"mcpServers"` + } + if err := json.Unmarshal(raw, &cfg); err != nil { + t.Errorf("parse per-episode mcp config: %v", err) + } + gotCredential = strings.TrimPrefix(cfg.MCPServers["bench"].Headers["Authorization"], "Bearer ") return []byte(claudeEvalStream), nil, nil }, }) @@ -551,6 +1076,9 @@ func TestClaudeCLIEpisode(t *testing.T) { if rec.err != "" { t.Fatalf("claude-cli episode error: %s", rec.err) } + if want := pool.Credential("testkey", 7, 32); gotCredential != want { + t.Errorf("episode credential = %q, want %q (pool rotation broken)", gotCredential, want) + } if !gradeEval(rec.finalAnswer, task.Grading{Kind: task.GradeNumeric, Value: new(100.0), AbsTolerance: 0.01}) { t.Errorf("claude-cli answer %q did not grade correct", rec.finalAnswer) } diff --git a/bench/internal/curriculum/curriculum.go b/bench/internal/curriculum/curriculum.go index ab604d8b..71f045a7 100644 --- a/bench/internal/curriculum/curriculum.go +++ b/bench/internal/curriculum/curriculum.go @@ -136,6 +136,13 @@ func (l Lesson) validateSink(curriculumID string) error { if l.Page == nil || l.Page.Slug == "" || l.Page.Title == "" || l.Page.Body == "" { return fmt.Errorf("curriculum %s: lesson %s knowledge_page sink requires a complete page payload", curriculumID, l.ID) } + // The summary is required, not optional: search renders a page hit as + // title plus summary, and the a3 tool surface has no page-body fetch, so + // a promoted page with an empty summary is a title-only hit whose fact + // never reaches an evaluator. + if l.Page.Summary == "" { + return fmt.Errorf("curriculum %s: lesson %s knowledge_page sink requires a non-empty page summary (search delivers the fact through it)", curriculumID, l.ID) + } return nil default: return fmt.Errorf("curriculum %s: lesson %s unknown sink %q", curriculumID, l.ID, l.Sink) diff --git a/bench/internal/curriculum/curriculum_test.go b/bench/internal/curriculum/curriculum_test.go index 5e842004..c82b73f5 100644 --- a/bench/internal/curriculum/curriculum_test.go +++ b/bench/internal/curriculum/curriculum_test.go @@ -25,10 +25,10 @@ func TestValidateAcceptsWellFormed(t *testing.T) { if err := validCurriculum().Validate(); err != nil { t.Fatalf("valid curriculum rejected: %v", err) } - // A page-sink lesson with a complete payload is valid. + // A page-sink lesson with a complete payload (summary included) is valid. c := validCurriculum() c.Lessons[0].Sink = protocol.SinkKnowledgePage - c.Lessons[0].Page = &protocol.PagePayload{Slug: "s", Title: "T", Body: "B"} + c.Lessons[0].Page = &protocol.PagePayload{Slug: "s", Title: "T", Summary: "the fact", Body: "B"} if err := c.Validate(); err != nil { t.Fatalf("valid page-sink curriculum rejected: %v", err) } @@ -55,7 +55,13 @@ func TestValidateRejectsMalformed(t *testing.T) { }, "page sink partial payload": func(c *Curriculum) { c.Lessons[0].Sink = protocol.SinkKnowledgePage - c.Lessons[0].Page = &protocol.PagePayload{Slug: "s", Title: "", Body: "B"} + c.Lessons[0].Page = &protocol.PagePayload{Slug: "s", Title: "", Summary: "f", Body: "B"} + }, + // A page without a summary is a title-only search hit on the a3 tool + // surface (no page-body fetch), so the fact never reaches an evaluator. + "page sink empty summary": func(c *Curriculum) { + c.Lessons[0].Sink = protocol.SinkKnowledgePage + c.Lessons[0].Page = &protocol.PagePayload{Slug: "s", Title: "T", Body: "B"} }, "duplicate lesson id": func(c *Curriculum) { c.Lessons = append(c.Lessons, c.Lessons[0]) diff --git a/bench/internal/gen/emit_curriculum.go b/bench/internal/gen/emit_curriculum.go index 82456117..99ad51cb 100644 --- a/bench/internal/gen/emit_curriculum.go +++ b/bench/internal/gen/emit_curriculum.go @@ -65,12 +65,12 @@ func (d *Dataset) Curriculum() curriculum.Curriculum { Lessons: []curriculum.Lesson{ datahubLesson("cs-units-cents", "Monetary columns are integer cents", "units_cents", unitsCentsFact, orders), pageLesson("cs-net-revenue", "Net-revenue reporting policy", "net_revenue", netRevenueFact, orders, - "revenue-reporting-policy", "Revenue Reporting Policy", revenuePolicyBody), + "revenue-reporting-policy", "Revenue Reporting Policy", revenuePolicySummary, revenuePolicyBody), pageLesson("cs-fiscal-calendar", "Fiscal calendar boundaries", "fiscal_calendar", fiscalCalendarFact, orders, - "fiscal-calendar-policy", "Fiscal Calendar Policy", fiscalCalendarBody), + "fiscal-calendar-policy", "Fiscal Calendar Policy", fiscalCalendarSummary, fiscalCalendarBody), datahubLesson("cs-freshness-cutoff", "Daily index freshness cutoff", "freshness_cutoff", dailyDescription, daily), pageLesson("cs-tier-boundary", "Key-account tier definition", "tier_boundary", tierBoundaryFact, customers, - "customer-tier-definitions", "Customer Tier Definitions", tierDefinitionsBody), + "customer-tier-definitions", "Customer Tier Definitions", tierDefinitionsSummary, tierDefinitionsBody), datahubLesson("cs-deprecated-table", "legacy_orders is deprecated", "deprecated_table", legacyDescription, legacy), }, } @@ -87,13 +87,15 @@ func datahubLesson(id, title, trapClass, fact, urn string) curriculum.Lesson { } // pageLesson builds a lesson that promotes its fact to a portal knowledge page -// (delivered to any identity via the search tool). The page reuses the A2 seed's -// slug/title/body so the promoted page is identical to the documented baseline. -func pageLesson(id, title, trapClass, fact, urn, slug, pageTitle, body string) curriculum.Lesson { +// (delivered to any identity via the search tool). The page reuses the A2 +// seed's slug/title/summary/body so the promoted page is identical to the +// documented baseline — the summary especially, because search renders a page +// hit as title plus summary and the a3 surface has no page-body fetch tool. +func pageLesson(id, title, trapClass, fact, urn, slug, pageTitle, summary, body string) curriculum.Lesson { return curriculum.Lesson{ ID: id, Title: title, TrapClass: trapClass, Fact: fact, EntityURN: urn, Sink: protocol.SinkKnowledgePage, BudgetToolCalls: coldStartBudget, - Page: &protocol.PagePayload{Slug: slug, Title: pageTitle, Body: body}, + Page: &protocol.PagePayload{Slug: slug, Title: pageTitle, Summary: summary, Body: body}, Teach: protocol.TeachStage{Prompt: teachPrompt(fact)}, } } diff --git a/bench/internal/gen/emit_kp.go b/bench/internal/gen/emit_kp.go index dfd28520..ecb4023d 100644 --- a/bench/internal/gen/emit_kp.go +++ b/bench/internal/gen/emit_kp.go @@ -10,6 +10,22 @@ import ( // pattern (embeddings NULL, search falls back to lexical until the indexer // fills them) and is idempotent via ON CONFLICT. +// Page summaries, shared between the a2 seed rows and the cold-start +// curriculum's page-sink lessons (one source of truth). Search renders a page +// hit as title plus summary — the a3 tool surface has no page-body fetch — so +// the summary must carry the page's answer-bearing fact on its own; a promoted +// page delivers exactly what the a2 seed delivers for the same page. +const ( + revenuePolicySummary = "Revenue = amount - discount over completed orders only; amounts are stored in US cents. " + + "The authoritative definition for all bench warehouse reporting." + warehouseGuideSummary = "Which bench table to use: orders is current, legacy_orders is deprecated, " + + "daily_region_revenue is gross-only pre-aggregation refreshed through 2025-11-30." + fiscalCalendarSummary = "The fiscal year starts February 1: fiscal year 2025 runs 2025-02-01 through " + + "2026-01-31. Fiscal figures must not be computed over the calendar year." + tierDefinitionsSummary = "A 'key account' is any customer on the plus or enterprise tier, a derived " + + "grouping not stored in any column and broader than the enterprise tier alone." +) + const revenuePolicyBody = `# Revenue Reporting Policy This is the authoritative definition of "revenue" for all bench warehouse reporting. @@ -86,47 +102,51 @@ type kpRow struct { tags string } -// KnowledgePagesSQL emits idempotent inserts for the benchmark's knowledge -// pages, applied via psql after platform migrations have run. -func (d *Dataset) KnowledgePagesSQL() string { - rows := []kpRow{ +// knowledgePageRows returns the a2 seed's page rows. It is the slug -> content +// source of truth the curriculum's page-sink lessons must agree with (a +// generator test diffs the two), so the promoted page is identical to the +// documented baseline. +func knowledgePageRows() []kpRow { + return []kpRow{ { - id: "kp-bench-1", - slug: "revenue-reporting-policy", - title: "Revenue Reporting Policy", - summary: "Revenue = amount - discount over completed orders only; amounts are stored in US cents. " + - "The authoritative definition for all bench warehouse reporting.", - body: revenuePolicyBody, - tags: `["finance","policy","bench"]`, + id: "kp-bench-1", + slug: "revenue-reporting-policy", + title: "Revenue Reporting Policy", + summary: revenuePolicySummary, + body: revenuePolicyBody, + tags: `["finance","policy","bench"]`, }, { - id: "kp-bench-2", - slug: "bench-warehouse-guide", - title: "Bench Warehouse Guide", - summary: "Which bench table to use: orders is current, legacy_orders is deprecated, " + - "daily_region_revenue is gross-only pre-aggregation refreshed through 2025-11-30.", - body: warehouseGuideBody, - tags: `["warehouse","bench"]`, + id: "kp-bench-2", + slug: "bench-warehouse-guide", + title: "Bench Warehouse Guide", + summary: warehouseGuideSummary, + body: warehouseGuideBody, + tags: `["warehouse","bench"]`, }, { - id: "kp-bench-3", - slug: "fiscal-calendar-policy", - title: "Fiscal Calendar Policy", - summary: "The fiscal year starts February 1: fiscal year 2025 runs 2025-02-01 through " + - "2026-01-31. Fiscal figures must not be computed over the calendar year.", - body: fiscalCalendarBody, - tags: `["finance","policy","bench"]`, + id: "kp-bench-3", + slug: "fiscal-calendar-policy", + title: "Fiscal Calendar Policy", + summary: fiscalCalendarSummary, + body: fiscalCalendarBody, + tags: `["finance","policy","bench"]`, }, { - id: "kp-bench-4", - slug: "customer-tier-definitions", - title: "Customer Tier Definitions", - summary: "A 'key account' is any customer on the plus or enterprise tier, a derived " + - "grouping not stored in any column and broader than the enterprise tier alone.", - body: tierDefinitionsBody, - tags: `["reporting","policy","bench"]`, + id: "kp-bench-4", + slug: "customer-tier-definitions", + title: "Customer Tier Definitions", + summary: tierDefinitionsSummary, + body: tierDefinitionsBody, + tags: `["reporting","policy","bench"]`, }, } +} + +// KnowledgePagesSQL emits idempotent inserts for the benchmark's knowledge +// pages, applied via psql after platform migrations have run. +func (d *Dataset) KnowledgePagesSQL() string { + rows := knowledgePageRows() var b strings.Builder fmt.Fprintf(&b, "-- Generated by bench/seedgen (seed %d). Do not edit; regenerate with `make bench-gen`.\n", Seed) b.WriteString("-- Requires platform migrations (table portal_knowledge_pages); apply after the platform has booted.\n") diff --git a/bench/internal/gen/emit_protocols.go b/bench/internal/gen/emit_protocols.go index ba560d7a..5ae00f34 100644 --- a/bench/internal/gen/emit_protocols.go +++ b/bench/internal/gen/emit_protocols.go @@ -58,10 +58,14 @@ func correctionPrompt(fact string) string { " Please update your saved knowledge so future answers use this corrected definition." } -// pagePayload builds a knowledge page for the knowledge_page sink from the fact. -// The slug is unique to the protocol and distinct from the seeded pages. +// pagePayload builds a knowledge page for the knowledge_page sink from the +// fact. The slug is unique to the protocol and distinct from the seeded pages +// (so no shared a2 summary applies), and the fact — a self-contained +// one-sentence definition — doubles as the summary: search renders a page hit +// as title plus summary, and the a3 tool surface has no page-body fetch, so +// the summary is the only channel the promoted fact reaches a learner through. func pagePayload(slug, title, fact string) *protocol.PagePayload { - return &protocol.PagePayload{Slug: slug, Title: title, Body: "# " + title + "\n\n" + fact} + return &protocol.PagePayload{Slug: slug, Title: title, Summary: fact, Body: "# " + title + "\n\n" + fact} } // otherRegions returns every region except one, for entity wrong-answer sets. diff --git a/bench/internal/gen/gen_test.go b/bench/internal/gen/gen_test.go index e55f6622..b0e2b9c2 100644 --- a/bench/internal/gen/gen_test.go +++ b/bench/internal/gen/gen_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/txn2/mcp-data-platform/bench/internal/curriculum" + "github.com/txn2/mcp-data-platform/bench/internal/protocol" "github.com/txn2/mcp-data-platform/bench/internal/task" ) @@ -202,6 +203,62 @@ func TestScriptedColdStartSmokeCoversUnits(t *testing.T) { } } +// TestCurriculumPageLessonsMatchSeedPages asserts every page-sink lesson's page +// payload (slug, title, summary, body) equals the a2 seed row for the same +// slug, so a promoted page is byte-identical to the documented baseline — the +// summary especially, because search renders a page hit as title plus summary +// and the a3 tool surface has no page-body fetch tool. +func TestCurriculumPageLessonsMatchSeedPages(t *testing.T) { + bySlug := map[string]kpRow{} + for _, r := range knowledgePageRows() { + bySlug[r.slug] = r + } + pageLessons := 0 + for _, l := range Generate().Curriculum().Lessons { + if l.Sink != protocol.SinkKnowledgePage { + continue + } + pageLessons++ + row, ok := bySlug[l.Page.Slug] + if !ok { + t.Errorf("lesson %s promotes page %q, which is not an a2 seed page", l.ID, l.Page.Slug) + continue + } + if l.Page.Title != row.title || l.Page.Summary != row.summary || l.Page.Body != row.body { + t.Errorf("lesson %s page payload diverges from the a2 seed row for slug %q", l.ID, l.Page.Slug) + } + if l.Page.Summary == "" { + t.Errorf("lesson %s page has an empty summary", l.ID) + } + } + if pageLessons != 3 { + t.Errorf("expected 3 page-sink lessons, found %d", pageLessons) + } +} + +// TestProtocolPagePayloadsCarrySummaries asserts every page-sink protocol sends +// a fact-bearing summary, so the promoted page's fact is deliverable through +// search on the a3 tool surface (the #958 page-sink transfer gap). +func TestProtocolPagePayloadsCarrySummaries(t *testing.T) { + pageSinks := 0 + for _, p := range Generate().Protocols() { + if p.Sink != protocol.SinkKnowledgePage { + continue + } + pageSinks++ + if p.Page == nil || p.Page.Summary == "" { + t.Errorf("protocol %s page payload has no summary", p.ID) + continue + } + if p.Page.Summary != p.Fact { + t.Errorf("protocol %s page summary %q is not its one-sentence fact", p.ID, p.Page.Summary) + } + } + if pageSinks == 0 { + t.Fatal("no page-sink protocols found") + } +} + // TestCommittedArtifactsMatch is the reproducibility gate: the committed seed // artifacts and task set must regenerate byte-identically from the fixed seed. func TestCommittedArtifactsMatch(t *testing.T) { diff --git a/bench/internal/lifecycle/episode.go b/bench/internal/lifecycle/episode.go index 8ce0cf3f..578ed5f0 100644 --- a/bench/internal/lifecycle/episode.go +++ b/bench/internal/lifecycle/episode.go @@ -125,7 +125,7 @@ func (e *runEnv) runEpisode(ctx context.Context, spec episodeSpec) (EpisodeRecor indeterminate++ return llm.ToolResult{Text: "transport error: " + r.TransportErr.Error(), IsError: true} } - if !preAuditRefusal(r.ErrorCode) { + if !mcpc.PreAuditRefusal(r.ErrorCode) { audited++ } return llm.ToolResult{Text: r.Text, IsError: r.ToolErr} @@ -248,15 +248,3 @@ func (e *runEnv) writeClaudeTranscript(spec episodeSpec, msgs []llm.Message) { e.log.Warn("write transcript", "error", err) } } - -// preAuditRefusal reports whether a structured error code marks a platform -// refusal issued outer to the audit middleware (so it leaves no audit row). -// Mirrors the S1-S3 pipeline's classification. -func preAuditRefusal(code string) bool { - switch code { - case "unauthenticated", "unauthorized", "session_required", "session_expired", - "search_required", "setup_required", "rate_limited": - return true - } - return false -} diff --git a/bench/internal/lifecycle/instrument.go b/bench/internal/lifecycle/instrument.go index eb626549..9a19b366 100644 --- a/bench/internal/lifecycle/instrument.go +++ b/bench/internal/lifecycle/instrument.go @@ -127,11 +127,14 @@ func isAlphaNum(r rune) bool { // surfacedTarget returns the promoted content whose presence in a tool result // marks the transfer fact as surfaced: the datahub sink applies the fact as the -// entity description, and the knowledge-page sink applies the page body. Both -// are what cross-enrichment / search deliver to the second identity. +// entity description (returned verbatim by cross-enrichment), and the +// knowledge-page sink is matched on the page SUMMARY — search renders a page +// hit as title plus summary, and the a3 tool surface has no page-body fetch +// tool, so the body can never appear in a tool result there; a body needle +// would report ~100% "not surfaced" even when delivery worked. func surfacedTarget(p protocol.Protocol) string { if p.Sink == protocol.SinkKnowledgePage && p.Page != nil { - return p.Page.Body + return p.Page.Summary } return p.Fact } diff --git a/bench/internal/lifecycle/instrument_test.go b/bench/internal/lifecycle/instrument_test.go index b99003d2..48c7edbd 100644 --- a/bench/internal/lifecycle/instrument_test.go +++ b/bench/internal/lifecycle/instrument_test.go @@ -102,10 +102,13 @@ func TestSurfacedTarget(t *testing.T) { if got := surfacedTarget(datahub); got != "the fact" { t.Errorf("datahub sink target = %q, want the fact", got) } + // The page-sink needle is the SUMMARY, not the body: search renders a page + // hit as title plus summary and the a3 surface has no page-body fetch, so a + // body needle could never appear in a tool result there. page := protocol.Protocol{Sink: protocol.SinkKnowledgePage, Fact: "the fact", - Page: &protocol.PagePayload{Body: "the page body"}} - if got := surfacedTarget(page); got != "the page body" { - t.Errorf("page sink target = %q, want the page body", got) + Page: &protocol.PagePayload{Summary: "the page summary", Body: "the page body"}} + if got := surfacedTarget(page); got != "the page summary" { + t.Errorf("page sink target = %q, want the page summary", got) } // A page sink with no page payload falls back to the fact rather than panicking. if got := surfacedTarget(protocol.Protocol{Sink: protocol.SinkKnowledgePage, Fact: "the fact"}); got != "the fact" { diff --git a/bench/internal/lifecycle/promote.go b/bench/internal/lifecycle/promote.go index 1c620112..0eef828c 100644 --- a/bench/internal/lifecycle/promote.go +++ b/bench/internal/lifecycle/promote.go @@ -19,9 +19,12 @@ import ( const insightPollInterval = 250 * time.Millisecond // waitForInsight polls for a pending insight captured by the identity and -// anchored to the entity, using the shared promote path (see promote.WaitForInsight). -func (e *runEnv) waitForInsight(ctx context.Context, email, urn string) (*lifecycleapi.Insight, error) { - return promote.WaitForInsight(ctx, e.life, email, urn, e.opts.AuditTimeout, insightPollInterval) +// anchored to the entity, using the shared promote path (see +// promote.WaitForInsight). since is the teach episode's start time; it bounds +// the match to this run so an interrupted prior run's leftover pending insight +// can never fake this episode's capture. +func (e *runEnv) waitForInsight(ctx context.Context, email, urn string, since time.Time) (*lifecycleapi.Insight, error) { + return promote.WaitForInsight(ctx, e.life, email, urn, since.Add(-promote.CaptureSkewMargin), e.opts.AuditTimeout, insightPollInterval) } // promoteInsight plays the reviewer: it approves the insight and applies it to diff --git a/bench/internal/lifecycle/runner.go b/bench/internal/lifecycle/runner.go index 47fac2f7..5fa9a79d 100644 --- a/bench/internal/lifecycle/runner.go +++ b/bench/internal/lifecycle/runner.go @@ -263,6 +263,9 @@ func (e *runEnv) teachAndRecall(ctx context.Context, p protocol.Protocol, teache // capture tool and whether it exhausted its budget) so a capture miss can be // attributed. Returns true when a harness failure aborts the run. func (e *runEnv) teachAndCapture(ctx context.Context, p protocol.Protocol, teacherSeq int, run *ProtocolRun) bool { + // The teach start bounds capture verification to this run (identities and + // URNs repeat deterministically across runs; see waitForInsight). + teachStart := time.Now() teach, _ := e.runEpisode(ctx, episodeSpec{ stage: StageTeach, identity: "teacher", seq: teacherSeq, prompt: p.Teach.Prompt, system: teachSystem, budget: e.teachBudget(p), @@ -283,7 +286,7 @@ func (e *runEnv) teachAndCapture(ctx context.Context, p protocol.Protocol, teach return true } - insight, err := e.waitForInsight(ctx, poolEmail(teacherSeq), p.EntityURN) + insight, err := e.waitForInsight(ctx, poolEmail(teacherSeq), p.EntityURN, teachStart) if err != nil { run.Error = "capture verify: " + err.Error() return true diff --git a/bench/internal/lifecycle/runner_test.go b/bench/internal/lifecycle/runner_test.go index cccfa23e..e2527915 100644 --- a/bench/internal/lifecycle/runner_test.go +++ b/bench/internal/lifecycle/runner_test.go @@ -653,7 +653,7 @@ func pageProtocol() protocol.Protocol { p := okProtocol() p.ID = "lc-page" p.Sink = protocol.SinkKnowledgePage - p.Page = &protocol.PagePayload{Slug: "revenue-policy", Title: "Revenue Policy", Body: "net revenue is amount minus discount"} + p.Page = &protocol.PagePayload{Slug: "revenue-policy", Title: "Revenue Policy", Summary: "net revenue is amount minus discount", Body: "net revenue is amount minus discount"} return p } diff --git a/bench/internal/lifecycleapi/lifecycleapi.go b/bench/internal/lifecycleapi/lifecycleapi.go index bb7b0ab3..dc69c5cb 100644 --- a/bench/internal/lifecycleapi/lifecycleapi.go +++ b/bench/internal/lifecycleapi/lifecycleapi.go @@ -68,6 +68,12 @@ type InsightFilter struct { CapturedBy string EntityURN string Status string + // Since bounds the listing to insights created at or after this time + // (the admin API's `since` param, RFC 3339). Capture verification passes the + // episode's start time so a pending insight left behind by an interrupted + // earlier run — same deterministic teacher identity, same curriculum URN — + // can never fake this episode's capture. + Since time.Time } // ChangesetFilter selects changesets to list. Zero-valued fields are omitted. @@ -114,6 +120,9 @@ func (c *Client) ListInsights(ctx context.Context, f InsightFilter) ([]Insight, setNonEmpty(q, "captured_by", f.CapturedBy) setNonEmpty(q, "entity_urn", f.EntityURN) setNonEmpty(q, "status", f.Status) + if !f.Since.IsZero() { + q.Set("since", f.Since.UTC().Format(time.RFC3339)) + } q.Set("page", strconv.Itoa(page)) q.Set("per_page", strconv.Itoa(pageSize)) var env insightEnvelope @@ -156,6 +165,40 @@ func (c *Client) ListChangesets(ctx context.Context, f ChangesetFilter) ([]Chang } } +// KnowledgePage is the subset of the portal knowledge-page record the harness +// reads (the cold-start preflight checks no curriculum slug already exists). +type KnowledgePage struct { + ID string `json:"id"` + Slug string `json:"slug"` +} + +// knowledgePageEnvelope is the portal list response for knowledge pages. +type knowledgePageEnvelope struct { + Pages []KnowledgePage `json:"pages"` + Total int `json:"total"` +} + +// ListKnowledgePages returns every live knowledge page, following pagination. +// It reads the portal REST list (the same authenticated client the admin +// knowledge reads use); the cold-start preflight scans it for curriculum slugs +// left by a prior run. +func (c *Client) ListKnowledgePages(ctx context.Context) ([]KnowledgePage, error) { + var all []KnowledgePage + for offset := 0; ; offset += pageSize { + q := url.Values{} + q.Set("limit", strconv.Itoa(pageSize)) + q.Set("offset", strconv.Itoa(offset)) + var env knowledgePageEnvelope + if err := c.getJSON(ctx, "/api/v1/portal/knowledge-pages", q, &env); err != nil { + return nil, err + } + all = append(all, env.Pages...) + if len(all) >= env.Total || len(env.Pages) == 0 { + return all, nil + } + } +} + // GetChangeset returns one changeset by ID. func (c *Client) GetChangeset(ctx context.Context, id string) (*Changeset, error) { var cs Changeset diff --git a/bench/internal/lifecycleapi/lifecycleapi_test.go b/bench/internal/lifecycleapi/lifecycleapi_test.go index d72294cc..d3cab197 100644 --- a/bench/internal/lifecycleapi/lifecycleapi_test.go +++ b/bench/internal/lifecycleapi/lifecycleapi_test.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" "testing" + "time" ) // fakeKnowledge serves the admin insights + changesets endpoints over httptest. @@ -35,7 +36,7 @@ func (f *fakeKnowledge) handler() http.Handler { func (f *fakeKnowledge) listInsights(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() - for _, k := range []string{"captured_by", "entity_urn", "status"} { + for _, k := range []string{"captured_by", "entity_urn", "status", "since"} { if v := q.Get(k); v != "" { f.lastQuery[k] = v } @@ -177,6 +178,53 @@ func TestListInsightsFiltersAndPaginates(t *testing.T) { if f.lastQuery["captured_by"] != "teacher@apikey.local" || f.lastQuery["entity_urn"] != "urn:orders" { t.Fatalf("filters not forwarded: %v", f.lastQuery) } + // A zero Since must not emit the param. + if _, present := f.lastQuery["since"]; present { + t.Errorf("zero Since emitted a since param: %v", f.lastQuery) + } +} + +// TestListInsightsEmitsSince proves a non-zero Since is forwarded as the admin +// API's RFC 3339 `since` param, the bound capture verification relies on to +// exclude an interrupted prior run's leftovers. +func TestListInsightsEmitsSince(t *testing.T) { + f := newFake() + f.insights = []Insight{{ID: "in-1", Status: "pending"}} + since := time.Date(2026, 7, 16, 10, 30, 0, 0, time.UTC) + if _, err := newClient(t, f).ListInsights(context.Background(), InsightFilter{Since: since}); err != nil { + t.Fatalf("list: %v", err) + } + if got, want := f.lastQuery["since"], since.Format(time.RFC3339); got != want { + t.Errorf("since param = %q, want %q", got, want) + } +} + +// TestListKnowledgePagesPaginates proves the portal knowledge-pages read the +// cold-start preflight scans follows pagination to completion. +func TestListKnowledgePagesPaginates(t *testing.T) { + pages := make([]KnowledgePage, 0, 150) + for i := range 150 { + pages = append(pages, KnowledgePage{ID: "kp-" + strconv.Itoa(i), Slug: "slug-" + strconv.Itoa(i)}) + } + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/portal/knowledge-pages" { + http.Error(w, "not found", http.StatusNotFound) + return + } + limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) + offset, _ := strconv.Atoi(r.URL.Query().Get("offset")) + lo := min(offset, len(pages)) + hi := min(lo+limit, len(pages)) + writeJSON(w, map[string]any{"pages": pages[lo:hi], "total": len(pages)}) + })) + t.Cleanup(srv.Close) + got, err := New(srv.URL, srv.Client()).ListKnowledgePages(context.Background()) + if err != nil { + t.Fatalf("list pages: %v", err) + } + if len(got) != 150 || got[149].Slug != "slug-149" { + t.Fatalf("got %d pages, want 150 across two pages", len(got)) + } } func TestGetInsight(t *testing.T) { diff --git a/bench/internal/mcpc/mcpc.go b/bench/internal/mcpc/mcpc.go index d9f3ea45..eebae40d 100644 --- a/bench/internal/mcpc/mcpc.go +++ b/bench/internal/mcpc/mcpc.go @@ -196,6 +196,22 @@ func Call(ctx context.Context, s *mcp.ClientSession, name string, args map[strin return CallResult{ToolErr: res.IsError, ErrorCode: errorCode(res), Text: allText(res)} } +// PreAuditRefusal reports whether a structured error code marks a platform +// refusal issued outer to the audit middleware (so it leaves no audit row): +// authentication, session, workflow-gate, setup, and rate-limit refusals. The +// episode loops use it to bound the audit read-back (a refused call cannot have +// a row), and the reviewer promote path uses it to classify a refused +// apply_knowledge as a harness failure rather than a measured miss. One shared +// copy keeps every path's classification identical. +func PreAuditRefusal(code string) bool { + switch code { + case "unauthenticated", "unauthorized", "session_required", "session_expired", + "search_required", "setup_required", "rate_limited": + return true + } + return false +} + // errorCode extracts the structured error code from a tool result, if any. func errorCode(res *mcp.CallToolResult) string { if res == nil || !res.IsError || res.StructuredContent == nil { diff --git a/bench/internal/mcpc/mcpc_test.go b/bench/internal/mcpc/mcpc_test.go index 4375a851..704087e6 100644 --- a/bench/internal/mcpc/mcpc_test.go +++ b/bench/internal/mcpc/mcpc_test.go @@ -105,3 +105,22 @@ func TestFirstText(t *testing.T) { t.Errorf("FirstText(nil) = %q, want empty", got) } } + +// TestPreAuditRefusal pins the shared refusal-code set: every code the platform +// error contract issues outer to the audit middleware classifies true, and a +// handler-level tool error (or an empty code) classifies false. +func TestPreAuditRefusal(t *testing.T) { + for _, code := range []string{ + "unauthenticated", "unauthorized", "session_required", "session_expired", + "search_required", "setup_required", "rate_limited", + } { + if !PreAuditRefusal(code) { + t.Errorf("code %q must classify as a pre-audit refusal", code) + } + } + for _, code := range []string{"", "query_failed", "invalid_arguments"} { + if PreAuditRefusal(code) { + t.Errorf("code %q must not classify as a pre-audit refusal", code) + } + } +} diff --git a/bench/internal/pipeline/pipeline.go b/bench/internal/pipeline/pipeline.go index 8c8dd80b..579a10e2 100644 --- a/bench/internal/pipeline/pipeline.go +++ b/bench/internal/pipeline/pipeline.go @@ -259,7 +259,7 @@ func (e *runEnv) runLoopAttempt(ctx context.Context, t task.Task, attempt, seq i indeterminate++ return llm.ToolResult{Text: "transport error: " + r.TransportErr.Error(), IsError: true} } - if !preAuditRefusal(r.ErrorCode) { + if !mcpc.PreAuditRefusal(r.ErrorCode) { audited++ } return llm.ToolResult{Text: r.Text, IsError: r.ToolErr} @@ -453,19 +453,6 @@ func (e *runEnv) referenceRows(ctx context.Context, exec SQLExecutor, t task.Tas return rows, nil } -// preAuditRefusal reports whether a structured error code marks a platform -// refusal issued outer to the audit middleware (pkg/middleware error contract: -// auth, authz, session gate, search-first gate). Such calls complete from the -// client's view but leave no audit row. -func preAuditRefusal(code string) bool { - switch code { - case "unauthenticated", "unauthorized", "session_required", "session_expired", - "search_required", "setup_required", "rate_limited": - return true - } - return false -} - // formatInstruction returns the per-grading-kind answer format rule. func formatInstruction(t task.Task) string { switch t.Grading.Kind { diff --git a/bench/internal/promote/promote.go b/bench/internal/promote/promote.go index faee96bc..c27b3103 100644 --- a/bench/internal/promote/promote.go +++ b/bench/internal/promote/promote.go @@ -25,6 +25,13 @@ import ( // ApplyToolName is the platform tool that applies an approved insight to a sink. const ApplyToolName = "apply_knowledge" +// CaptureSkewMargin is subtracted from a teach episode's start time before it +// is passed to WaitForInsight as the since bound. Harness and platform run on +// the same host in the bench stack, but the margin means a modest clock skew +// can never exclude a genuinely fresh capture; it is far shorter than the gap +// to any prior run's leftovers, so those stay excluded. +const CaptureSkewMargin = 30 * time.Second + // Insight lifecycle statuses the harness reads back through the knowledge API. const ( // StatusPending is a freshly captured, unreviewed insight; capture @@ -59,11 +66,17 @@ type Target struct { // WaitForInsight polls the insights API until a pending insight captured by the // given identity and anchored to the entity appears, returning the newest, or // nil when none lands within the timeout (a missed capture, not a harness -// error). The pending filter scopes the read to this episode's fresh capture. -func WaitForInsight(ctx context.Context, life *lifecycleapi.Client, email, urn string, timeout, poll time.Duration) (*lifecycleapi.Insight, error) { +// error). The pending filter scopes the read to this episode's fresh capture, +// and since bounds the match to insights created in THIS run: teacher +// identities are deterministic per lesson index and URNs are fixed by the +// curriculum, so without it a pending insight left by an interrupted prior run +// would fake the capture (and then be promoted). Callers pass the teach +// episode's start time minus a clock-skew margin; a zero since disables the +// bound. +func WaitForInsight(ctx context.Context, life *lifecycleapi.Client, email, urn string, since time.Time, timeout, poll time.Duration) (*lifecycleapi.Insight, error) { deadline := time.Now().Add(timeout) for { - insights, err := life.ListInsights(ctx, lifecycleapi.InsightFilter{CapturedBy: email, EntityURN: urn, Status: StatusPending}) + insights, err := life.ListInsights(ctx, lifecycleapi.InsightFilter{CapturedBy: email, EntityURN: urn, Status: StatusPending, Since: since}) if err != nil { return nil, err } @@ -103,9 +116,13 @@ type Reviewer struct { // Apply approves the insight (pending -> approved) and applies it via // apply_knowledge to the target's sink over the given admin session, then // verifies through the knowledge API that the insight is applied and a live -// changeset links it. A transport-level failure is a harness error (returned); -// an apply the platform refuses, or a promotion the API cannot confirm, is a -// measured miss (false, nil). +// changeset links it. A transport-level failure or a pre-audit platform refusal +// (an expired admin session, a rate limit — see mcpc.PreAuditRefusal) is a +// harness error (returned): the reviewer never legitimately loses its session, +// so scoring such a refusal as a miss would flatline the promote metric on a +// harness defect. A genuine tool-level refusal (apply_knowledge declining the +// change), or a promotion the API cannot confirm, is a measured miss +// (false, nil). func (r Reviewer) Apply(ctx context.Context, session *mcp.ClientSession, handle string, t Target, insightID string) (bool, error) { if err := r.Life.Approve(ctx, insightID, t.Notes); err != nil { return false, fmt.Errorf("approve insight: %w", err) @@ -115,6 +132,9 @@ func (r Reviewer) Apply(ctx context.Context, session *mcp.ClientSession, handle return false, fmt.Errorf("apply transport: %w", res.TransportErr) } if res.ToolErr { + if mcpc.PreAuditRefusal(res.ErrorCode) { + return false, fmt.Errorf("apply refused pre-audit (%s): %.300s", res.ErrorCode, res.Text) + } if r.Log != nil { r.Log.Warn("apply_knowledge returned an error", "target", t.Label, "text", res.Text) } @@ -169,7 +189,11 @@ func BuildApplyArgs(t Target, insightID string) map[string]any { args["page"] = map[string]any{ "slug": t.Page.Slug, "title": t.Page.Title, - "body": t.Page.Body, + // The summary is what search renders next to the title; on tool + // surfaces without a page-body fetch tool it is the only channel the + // page's fact reaches an agent through, so it must always be sent. + "summary": t.Page.Summary, + "body": t.Page.Body, } return args } diff --git a/bench/internal/promote/promote_test.go b/bench/internal/promote/promote_test.go index 167c8426..184a2eb4 100644 --- a/bench/internal/promote/promote_test.go +++ b/bench/internal/promote/promote_test.go @@ -40,7 +40,7 @@ func TestBuildApplyArgsDataHub(t *testing.T) { func TestBuildApplyArgsKnowledgePage(t *testing.T) { args := BuildApplyArgs(Target{ Label: "lc-y", EntityURN: "urn:y", Sink: protocol.SinkKnowledgePage, - Page: &protocol.PagePayload{Slug: "s", Title: "T", Body: "B"}, + Page: &protocol.PagePayload{Slug: "s", Title: "T", Summary: "the fact in one line", Body: "B"}, }, "ins-2") if args["sink"] != "knowledge_page" { t.Fatalf("page sink wrong: %+v", args) @@ -49,6 +49,12 @@ func TestBuildApplyArgsKnowledgePage(t *testing.T) { if !ok || page["slug"] != "s" || page["title"] != "T" || page["body"] != "B" { t.Fatalf("page payload wrong: %+v", args["page"]) } + // The summary must be sent: search renders a page hit as title plus summary, + // and on tool surfaces without a page-body fetch it is the only channel the + // promoted fact reaches an agent through. + if page["summary"] != "the fact in one line" { + t.Errorf("page summary not sent: %+v", args["page"]) + } if _, hasChanges := args["changes"]; hasChanges { t.Error("page sink must not carry datahub changes") } @@ -73,7 +79,7 @@ func TestWaitForInsight(t *testing.T) { defer srv.Close() life := lifecycleapi.New(srv.URL, srv.Client()) - got, err := WaitForInsight(context.Background(), life, "a@b", "urn:x", time.Second, time.Millisecond) + got, err := WaitForInsight(context.Background(), life, "a@b", "urn:x", time.Time{}, time.Second, time.Millisecond) if err != nil { t.Fatalf("wait: %v", err) } @@ -85,6 +91,52 @@ func TestWaitForInsight(t *testing.T) { } } +// TestWaitForInsightSinceBoundsToThisRun proves the since bound excludes a +// pending insight left by an earlier run (same deterministic identity and URN) +// while a fresh capture is matched. The fake applies the `since` param exactly +// as the admin API does (created_at >=), so the test exercises both the param +// emission and the resulting match window. +func TestWaitForInsightSinceBoundsToThisRun(t *testing.T) { + stale := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + fresh := stale.Add(2 * time.Hour) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var out []lifecycleapi.Insight + since, err := time.Parse(time.RFC3339, r.URL.Query().Get("since")) + if err != nil { + t.Errorf("since param missing or not RFC 3339: %v", err) + } + for _, in := range []lifecycleapi.Insight{ + {ID: "stale", CreatedAt: stale, Status: "pending"}, + {ID: "fresh", CreatedAt: fresh, Status: "pending"}, + } { + if !in.CreatedAt.Before(since) { + out = append(out, in) + } + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": out, "total": len(out)}) + })) + defer srv.Close() + life := lifecycleapi.New(srv.URL, srv.Client()) + + got, err := WaitForInsight(context.Background(), life, "a@b", "urn:x", stale.Add(time.Hour), time.Second, time.Millisecond) + if err != nil { + t.Fatalf("wait: %v", err) + } + if got == nil || got.ID != "fresh" { + t.Fatalf("expected only the fresh insight to match, got %+v", got) + } + + // A window after both insights matches nothing: the stale leftover cannot + // fake a capture (nil, nil is the measured-miss contract). + got, err = WaitForInsight(context.Background(), life, "a@b", "urn:x", fresh.Add(time.Hour), 20*time.Millisecond, time.Millisecond) + if err != nil { + t.Fatalf("a missed capture must not be an error: %v", err) + } + if got != nil { + t.Fatalf("stale insight matched despite the since bound: %+v", got) + } +} + func TestWaitForInsightTimeoutIsMiss(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _ = json.NewEncoder(w).Encode(map[string]any{"data": []lifecycleapi.Insight{}, "total": 0}) @@ -92,7 +144,7 @@ func TestWaitForInsightTimeoutIsMiss(t *testing.T) { defer srv.Close() life := lifecycleapi.New(srv.URL, srv.Client()) - got, err := WaitForInsight(context.Background(), life, "a@b", "urn:x", 20*time.Millisecond, time.Millisecond) + got, err := WaitForInsight(context.Background(), life, "a@b", "urn:x", time.Time{}, 20*time.Millisecond, time.Millisecond) if err != nil { t.Fatalf("a missed capture must not be an error: %v", err) } @@ -162,10 +214,14 @@ func TestReviewerVerify(t *testing.T) { // the Reviewer verifies against. It lets Apply be tested end to end (approve -> // apply_knowledge -> API verify) through genuine protocol wiring. type applyFake struct { - mu sync.Mutex - approved map[string]bool - applyError bool - srv *httptest.Server + mu sync.Mutex + approved map[string]bool + // applyError makes apply_knowledge return a plain tool error (a measured + // refusal); applyRefusalCode additionally attaches the platform error + // contract's structured code (a pre-audit refusal, e.g. session_expired). + applyError bool + applyRefusalCode string + srv *httptest.Server } func newApplyFake(t *testing.T, applyError bool) *applyFake { @@ -183,6 +239,13 @@ func newApplyFake(t *testing.T, applyError bool) *applyFake { }, Required: []string{"session_id"}} mcp.AddTool(server, &mcp.Tool{Name: ApplyToolName, Description: "promote", InputSchema: schema}, func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { + if f.applyRefusalCode != "" { + return &mcp.CallToolResult{ + IsError: true, + Content: []mcp.Content{&mcp.TextContent{Text: "SESSION_EXPIRED: mint a new handle"}}, + StructuredContent: map[string]any{"error": map[string]any{"code": f.applyRefusalCode}}, + }, nil, nil + } if f.applyError { return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "write refused"}}}, nil, nil } @@ -253,6 +316,36 @@ func TestReviewerApply(t *testing.T) { } } +// TestReviewerApplyPreAuditRefusalIsHarnessError proves a platform refusal +// issued outer to the audit middleware (session_expired, rate_limited, ...) on +// the admin session is a harness error, not a measured miss: scoring it as a +// refusal would silently flatline the promote metric on a harness defect. +func TestReviewerApplyPreAuditRefusalIsHarnessError(t *testing.T) { + f := newApplyFake(t, false) + f.applyRefusalCode = "session_expired" + ctx := context.Background() + session, err := mcpc.New(f.srv.URL, f.srv.Client()).Connect(ctx) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer func() { _ = session.Close() }() + info, err := mcpc.Mint(ctx, session) + if err != nil { + t.Fatalf("mint: %v", err) + } + r := Reviewer{Life: lifecycleapi.New(f.srv.URL, f.srv.Client())} + got, err := r.Apply(ctx, session, info.Handle, Target{Label: "cs-x", EntityURN: "urn:x", Sink: protocol.SinkDataHub, Fact: "f"}, "ins-1") + if err == nil { + t.Fatal("a pre-audit platform refusal must be a harness error, not a measured miss") + } + if got { + t.Error("a refused apply must not report promoted") + } + if !strings.Contains(err.Error(), "session_expired") { + t.Errorf("error should carry the refusal code, got: %v", err) + } +} + // TestReviewerVerifyRolledBackFails ensures a rolled-back changeset does not // count as a live promotion. func TestReviewerVerifyRolledBackFails(t *testing.T) { diff --git a/bench/internal/protocol/protocol.go b/bench/internal/protocol/protocol.go index b4c6c2bf..b22040ad 100644 --- a/bench/internal/protocol/protocol.go +++ b/bench/internal/protocol/protocol.go @@ -88,10 +88,15 @@ type Protocol struct { } // PagePayload is the knowledge_page promotion target (sink=knowledge_page). +// Summary is the fact-bearing one-liner search renders next to the title: on +// tool surfaces without a page-body fetch tool (the a3 arm), the summary is the +// ONLY channel through which a promoted page's fact reaches an agent, so a +// page payload without one is a title-only search hit that delivers nothing. type PagePayload struct { - Slug string `yaml:"slug" json:"slug"` - Title string `yaml:"title" json:"title"` - Body string `yaml:"body" json:"body"` + Slug string `yaml:"slug" json:"slug"` + Title string `yaml:"title" json:"title"` + Summary string `yaml:"summary" json:"summary"` + Body string `yaml:"body" json:"body"` } // TeachStage is the fact-capture episode. The prompt states the fact @@ -191,6 +196,13 @@ func (p Protocol) validateSink() error { if p.Page == nil || p.Page.Slug == "" || p.Page.Title == "" || p.Page.Body == "" { return fmt.Errorf("protocol %s: knowledge_page sink requires a complete page payload", p.ID) } + // The summary is required, not optional: search renders a page hit as + // title plus summary and the a3 tool surface has no page-body fetch, so a + // promoted page with an empty summary structurally cannot deliver its + // fact — the run would spend its budget measuring an impossible channel. + if p.Page.Summary == "" { + return fmt.Errorf("protocol %s: knowledge_page sink requires a non-empty page summary (search delivers the fact through it)", p.ID) + } return nil default: return fmt.Errorf("protocol %s: unknown sink %q", p.ID, p.Sink) diff --git a/bench/internal/protocol/protocol_test.go b/bench/internal/protocol/protocol_test.go index 887bc493..1776da49 100644 --- a/bench/internal/protocol/protocol_test.go +++ b/bench/internal/protocol/protocol_test.go @@ -77,7 +77,14 @@ func TestValidateRejects(t *testing.T) { }, "page sink partial payload": func(p *Protocol) { p.Sink = SinkKnowledgePage - p.Page = &PagePayload{Slug: "s", Title: "", Body: "b"} + p.Page = &PagePayload{Slug: "s", Title: "", Summary: "f", Body: "b"} + }, + // A page without a summary is a title-only search hit on tool surfaces + // with no page-body fetch, so its fact is structurally undeliverable; the + // set must refuse at load rather than spend a run measuring it. + "page sink empty summary": func(p *Protocol) { + p.Sink = SinkKnowledgePage + p.Page = &PagePayload{Slug: "s", Title: "T", Body: "b"} }, "transfer bad grade": func(p *Protocol) { p.Transfer.Grading = task.Grading{Kind: "bogus"} }, "abstain empty prompt": func(p *Protocol) { p.Abstain.Prompt = "" }, @@ -109,7 +116,7 @@ func runRejectCases(t *testing.T, group string, base func() Protocol, cases map[ func TestValidatePageSinkAccepted(t *testing.T) { p := validProtocol() p.Sink = SinkKnowledgePage - p.Page = &PagePayload{Slug: "revenue", Title: "Revenue", Body: "net revenue"} + p.Page = &PagePayload{Slug: "revenue", Title: "Revenue", Summary: "net revenue is amount minus discount", Body: "net revenue"} if err := p.Validate(); err != nil { t.Fatalf("page-sink protocol rejected: %v", err) } diff --git a/bench/protocols/lc-active-customers.yaml b/bench/protocols/lc-active-customers.yaml index e6d8da8e..326161d6 100644 --- a/bench/protocols/lc-active-customers.yaml +++ b/bench/protocols/lc-active-customers.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: active-customer-definition title: Active Customer Definition + summary: An 'active customer' is any customer who has placed at least one order. body: |- # Active Customer Definition diff --git a/bench/protocols/lc-anchor-region.yaml b/bench/protocols/lc-anchor-region.yaml index 527c8734..2785ee04 100644 --- a/bench/protocols/lc-anchor-region.yaml +++ b/bench/protocols/lc-anchor-region.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: anchor-region-definition title: Anchor Region Definition + summary: The 'anchor region' for this study is North. body: |- # Anchor Region Definition diff --git a/bench/protocols/lc-charter-cohort.yaml b/bench/protocols/lc-charter-cohort.yaml index 50fe9dec..3aaf48fc 100644 --- a/bench/protocols/lc-charter-cohort.yaml +++ b/bench/protocols/lc-charter-cohort.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: charter-cohort-definition title: Charter Cohort Definition + summary: The 'charter cohort' is the set of customers on the plus tier. body: |- # Charter Cohort Definition diff --git a/bench/protocols/lc-clearance-set.yaml b/bench/protocols/lc-clearance-set.yaml index ba4a7310..8365a112 100644 --- a/bench/protocols/lc-clearance-set.yaml +++ b/bench/protocols/lc-clearance-set.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: clearance-set-definition title: Clearance Set Definition + summary: The 'clearance set' is the set of completed orders placed by customers in the West region. body: |- # Clearance Set Definition diff --git a/bench/protocols/lc-core-market.yaml b/bench/protocols/lc-core-market.yaml index f5471e23..d57002af 100644 --- a/bench/protocols/lc-core-market.yaml +++ b/bench/protocols/lc-core-market.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: core-market-definition title: Core Market Definition + summary: The 'core market' is the North region. body: |- # Core Market Definition diff --git a/bench/protocols/lc-focus-region-net.yaml b/bench/protocols/lc-focus-region-net.yaml index 6294719c..0966e82d 100644 --- a/bench/protocols/lc-focus-region-net.yaml +++ b/bench/protocols/lc-focus-region-net.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: focus-region-definition title: Focus Region Definition + summary: The 'focus region' for this study is West. body: |- # Focus Region Definition diff --git a/bench/protocols/lc-headline-tier.yaml b/bench/protocols/lc-headline-tier.yaml index 50891d87..e790f261 100644 --- a/bench/protocols/lc-headline-tier.yaml +++ b/bench/protocols/lc-headline-tier.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: headline-tier-definition title: Headline Tier Definition + summary: The 'headline tier' for this study is the plus tier. body: |- # Headline Tier Definition diff --git a/bench/protocols/lc-holiday-total.yaml b/bench/protocols/lc-holiday-total.yaml index 7a2fc6e5..7476271f 100644 --- a/bench/protocols/lc-holiday-total.yaml +++ b/bench/protocols/lc-holiday-total.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: holiday-total-definition title: Holiday Total Definition + summary: The 'holiday total' is the gross revenue (sum of amount over completed orders) for November and December 2025. body: |- # Holiday Total Definition diff --git a/bench/protocols/lc-house-tier.yaml b/bench/protocols/lc-house-tier.yaml index 9a90a8e8..c9015f96 100644 --- a/bench/protocols/lc-house-tier.yaml +++ b/bench/protocols/lc-house-tier.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: house-tier-definition title: House Tier Definition + summary: The 'house tier' is the customer tier that placed the most orders. body: |- # House Tier Definition diff --git a/bench/protocols/lc-peak-region.yaml b/bench/protocols/lc-peak-region.yaml index d7d37997..55069d0d 100644 --- a/bench/protocols/lc-peak-region.yaml +++ b/bench/protocols/lc-peak-region.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: peak-region-definition title: Peak Region Definition + summary: The 'peak region' is the region with the most customers. body: |- # Peak Region Definition diff --git a/bench/protocols/lc-premium-orders.yaml b/bench/protocols/lc-premium-orders.yaml index 4a20a4da..1b2193d2 100644 --- a/bench/protocols/lc-premium-orders.yaml +++ b/bench/protocols/lc-premium-orders.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: premium-order-definition title: Premium Order Definition + summary: A 'premium order' is any order placed by a customer on the enterprise tier. body: |- # Premium Order Definition diff --git a/bench/protocols/lc-priority-orders.yaml b/bench/protocols/lc-priority-orders.yaml index c23fc78b..e1891fd4 100644 --- a/bench/protocols/lc-priority-orders.yaml +++ b/bench/protocols/lc-priority-orders.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: priority-order-definition title: Priority Order Definition + summary: A 'priority order' is any order whose amount is at least $1,000.00 (100000 cents). body: |- # Priority Order Definition diff --git a/bench/protocols/lc-settlement-total.yaml b/bench/protocols/lc-settlement-total.yaml index 6a7f3b92..ccb77007 100644 --- a/bench/protocols/lc-settlement-total.yaml +++ b/bench/protocols/lc-settlement-total.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: settlement-total-definition title: Settlement Total Definition + summary: The 'settlement total' is the gross revenue (sum of amount over completed orders) for calendar quarter 4 of 2025 (October through December). body: |- # Settlement Total Definition diff --git a/bench/protocols/lc-standard-order.yaml b/bench/protocols/lc-standard-order.yaml index 679f5215..4ab87e55 100644 --- a/bench/protocols/lc-standard-order.yaml +++ b/bench/protocols/lc-standard-order.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: standard-order-definition title: Standard Order Definition + summary: A 'standard order' is any order whose amount is below $1,000.00 (100000 cents). body: |- # Standard Order Definition diff --git a/bench/protocols/lc-value-tier.yaml b/bench/protocols/lc-value-tier.yaml index ef08ec89..405c29ef 100644 --- a/bench/protocols/lc-value-tier.yaml +++ b/bench/protocols/lc-value-tier.yaml @@ -8,6 +8,7 @@ sink: knowledge_page page: slug: value-tier-definition title: Value Tier Definition + summary: The 'value tier' is the customer tier that placed the fewest orders. body: |- # Value Tier Definition