diff --git a/Makefile b/Makefile index e05441a72..88c394335 100644 --- a/Makefile +++ b/Makefile @@ -448,7 +448,7 @@ verify-release: verify mutate ## verify: Run the CI-equivalent per-commit suite (test, lint, security, SAST, coverage, release) ## NOTE: mutation testing is intentionally excluded — it lives in verify-release. ## Do not add `mutate` back to this per-commit target. -verify: tools-check fmt swagger-check embed-clean test migrate-check test-realdb frontend-test frontend-lint frontend-e2e lint security semgrep codeql coverage-report patch-coverage doc-check emdash-check dead-code release-check +verify: tools-check fmt swagger-check embed-clean test migrate-check test-realdb frontend-test frontend-lint frontend-e2e lint bench-test bench-lint security semgrep codeql coverage-report patch-coverage doc-check emdash-check dead-code release-check @echo "" @echo "=== All checks passed ===" @# Write the gate sentinel: the short SHA-256 of the working-tree diff @@ -905,13 +905,16 @@ bench-run: bench-smoke: @$(MAKE) bench-run LLM=scripted SCRIPT=bench/tasks/scripted-smoke.json K=1 -## bench-lifecycle: Run the S5 memory-insight-knowledge lifecycle protocols (needs bench-up BENCH_ARM=a3; LLM=anthropic|scripted|claude-cli, K=, MODEL=) +## bench-lifecycle: Run the S5 memory-insight-knowledge lifecycle protocols into a fresh per-run dir under build/bench-results/ (needs bench-up BENCH_ARM=a3; LLM=anthropic|scripted|claude-cli, K=, MODEL=, BASELINE=) bench-lifecycle: @mkdir -p build/bench-results @cd bench && $(GO) build -o ../$(BUILD_DIR)/benchrun ./benchrun @echo "Resetting search-first gate state (discovery scopes persist in Postgres across runs)..." @$(BENCH_COMPOSE) exec -T postgres psql -q -U platform -d mcp_platform -v ON_ERROR_STOP=1 \ -c "TRUNCATE search_gate_discovery" + @out_dir="build/bench-results/lifecycle-a3-$$(date +%Y%m%d-%H%M%S)-$$$$"; \ + mkdir "$$out_dir"; \ + echo "Lifecycle results dir: $$out_dir (each run gets its own dir; nothing is ever overwritten)"; \ $(BUILD_DIR)/benchrun \ -lifecycle \ -arm a3 \ @@ -919,28 +922,36 @@ bench-lifecycle: -credential $(BENCH_KEY) \ -protocols bench/protocols \ -git-commit $$(git rev-parse HEAD) \ - -out build/bench-results/lifecycle-a3.json \ + -out "$$out_dir/lifecycle-a3.json" \ $(if $(LLM),-llm $(LLM),) \ $(if $(SCRIPT),-script $(SCRIPT),) \ $(if $(K),-k $(K),) \ - $(if $(MODEL),-model $(MODEL),) + $(if $(MODEL),-model $(MODEL),) \ + $(if $(BASELINE),-baseline $(BASELINE),) ## bench-lifecycle-smoke: Run the scripted (no-API-key) lifecycle smoke against the running a3 platform bench-lifecycle-smoke: @$(MAKE) bench-lifecycle LLM=scripted SCRIPT=bench/protocols/scripted-lifecycle-smoke.json K=1 -## bench-lifecycle-report: Print the human summary of the last lifecycle run +## bench-lifecycle-report: Print the human summary of a lifecycle run (RESULTS=/lifecycle-a3.json) bench-lifecycle-report: + @if [ -z "$(RESULTS)" ]; then \ + echo "ERROR: set RESULTS=. Available run dirs:"; \ + ls -d build/bench-results/lifecycle-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 -lifecycle -summarize build/bench-results/lifecycle-a3.json + $(BUILD_DIR)/benchrun -lifecycle -summarize $(RESULTS) -## bench-supersede: Run the isolated supersede sub-benchmark (issue #964; needs bench-up BENCH_ARM=a3; LLM=anthropic|scripted|claude-cli, K=, MODEL=, TEACH_BUDGET=) +## bench-supersede: Run the isolated supersede sub-benchmark into a fresh per-run dir under build/bench-results/ (issue #964; needs bench-up BENCH_ARM=a3; LLM=anthropic|scripted|claude-cli, K=, MODEL=, TEACH_BUDGET=) bench-supersede: @mkdir -p build/bench-results @cd bench && $(GO) build -o ../$(BUILD_DIR)/benchrun ./benchrun @echo "Resetting search-first gate state (discovery scopes persist in Postgres across runs)..." @$(BENCH_COMPOSE) exec -T postgres psql -q -U platform -d mcp_platform -v ON_ERROR_STOP=1 \ -c "TRUNCATE search_gate_discovery" + @out_dir="build/bench-results/supersede-a3-$$(date +%Y%m%d-%H%M%S)-$$$$"; \ + mkdir "$$out_dir"; \ + echo "Supersede results dir: $$out_dir (each run gets its own dir; nothing is ever overwritten)"; \ $(BUILD_DIR)/benchrun \ -supersede \ -arm a3 \ @@ -948,7 +959,7 @@ bench-supersede: -credential $(BENCH_KEY) \ -protocols bench/protocols \ -git-commit $$(git rev-parse HEAD) \ - -out build/bench-results/supersede-a3.json \ + -out "$$out_dir/supersede-a3.json" \ $(if $(LLM),-llm $(LLM),) \ $(if $(SCRIPT),-script $(SCRIPT),) \ $(if $(K),-k $(K),) \ @@ -959,10 +970,14 @@ bench-supersede: bench-supersede-smoke: @$(MAKE) bench-supersede LLM=scripted SCRIPT=bench/protocols/scripted-lifecycle-smoke.json K=1 -## bench-supersede-report: Print the human summary of the last supersede sub-benchmark run +## bench-supersede-report: Print the human summary of a supersede run (RESULTS=/supersede-a3.json) bench-supersede-report: + @if [ -z "$(RESULTS)" ]; then \ + echo "ERROR: set RESULTS=. Available run dirs:"; \ + ls -d build/bench-results/supersede-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 -supersede -summarize build/bench-results/supersede-a3.json + $(BUILD_DIR)/benchrun -supersede -summarize $(RESULTS) ## 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: @@ -972,8 +987,8 @@ 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"; \ + @out_dir="build/bench-results/cold-start-a3-$$(date +%Y%m%d-%H%M%S)-$$$$"; \ + mkdir "$$out_dir"; \ echo "Cold-start results dir: $$out_dir (each run gets its own dir; nothing is ever overwritten)"; \ $(BUILD_DIR)/benchrun \ -cold-start \ @@ -1034,3 +1049,16 @@ bench-down: bench-test: @echo "Testing the benchmark module..." @cd bench && $(GO) build ./... && $(GO) vet ./... && $(GO) test ./... + +## bench-lint: Full-module lint of the bench/ harness +## +## Mirrors CI's "Harness module checks" job, which runs golangci-lint over the +## whole bench module with NO only-new-issues scoping — unlike the main-module +## `lint` target, whose --new-from-patch scoping cannot see a finding that +## anchors on an unchanged line (e.g. a gocognit report on a func declaration +## whose body grew). bench/ is a separate Go module, so the root `lint` never +## reaches it at all; without this target a bench lint finding surfaces only +## in CI (PR #978's gocognit failure was exactly this gap). +bench-lint: + @echo "Linting the benchmark module (full module, matching CI's harness job)..." + @cd bench && $(GOLINT) run ./... diff --git a/bench/README.md b/bench/README.md index 39b36bace..7b457eefb 100644 --- a/bench/README.md +++ b/bench/README.md @@ -25,11 +25,21 @@ go build ./... && go vet ./... && go test ./... golangci-lint run ./... ``` -Or from the repo root: `make bench-test`. - -Like mutation and load testing, benchmarking is **deliberately not part of -`make verify`** — it stands up Docker services, a real server binary, and (for -real runs) a model API. Do not add `bench-*` to the `verify` target. +Or from the repo root: `make bench-test` and `make bench-lint`. Both are part +of `make verify`: they are pure module checks (build, vet, test, full-module +lint) that mirror CI's "Harness module checks" job, and because this module is +outside the root `./...`, the root `lint`/`test` targets never reach it — a +bench-only finding would otherwise surface first in CI (the `bench-lint` full- +module scope also matters: the root lint's --new-from-patch scoping cannot see +a finding anchored on an unchanged line, such as a gocognit report on a func +whose body grew). + +Like mutation and load testing, benchmark **runs** are deliberately not part of +`make verify` — they stand up Docker services, a real server binary, and (for +real runs) a model API. Do not add the stack-dependent `bench-*` run targets +(`bench-up`, `bench-run`, `bench-smoke`, the lifecycle/supersede/cold-start +runs) to the `verify` target; `bench-test` and `bench-lint` are the only +exceptions because they touch nothing outside the module. ## Arms @@ -219,9 +229,13 @@ that reuses pool identities does not cross-contaminate the metrics. Metrics (each a numerator/denominator over the applicable, non-harness-failed runs): **capture rate**, **personal recall**, **unprompted surface**, **transfer -rate**, **update correctness**, **duplicate rate** (lower is better), and -**abstention rate**, plus **pass^k** over protocols (all k attempts pass the full -applicable lifecycle). Harness-level failures (connect, adapter, API read-back) +rate**, **update correctness**, **update capture rate**, **duplicate rate** +(lower is better), and **abstention rate**, plus **pass^k** over protocols (all +k attempts pass the full applicable lifecycle). The duplicate rate counts only +attempts whose correction capture actually executed: an update episode that +never called capture left one live insight and the supersede gate never ran, so +it is an update-capture miss (measured on its own rate, and a pass^k failure), +not a duplicate. Harness-level failures (connect, adapter, API read-back) are excluded from the metrics and reported separately, mirroring the S1–S3 pipeline. @@ -266,7 +280,7 @@ protocol set. - **Regression gate.** `-baseline ` with `-lifecycle` gates the run against a committed S5 baseline and exits nonzero when a headline lifecycle metric regresses, so CI catches a lifecycle capability loss the same way it already catches an S1-S3 one. It applies to both a single-process run and a `-merge`d k=N scorecard (the canonical multi-pass artifact CI gates). The gated metrics are capture rate, personal recall, transfer rate, update correctness, abstention rate, duplicate rate (an increase past tolerance is the regression), and pass^k; a metric that either run did not exercise (zero denominator) is skipped as a coverage gap, not scored as a drop. The #964 diagnostic decompositions are deliberately not gated — their small denominators would trip the gate on noise; they exist to explain a regression, not define one. The gate refuses a cross-arm comparison, and a cross-client-path one (anthropic vs claude-cli), rather than producing a meaningless verdict; it compares the client path, not the exact CLI version, so a benign `claude` bump does not disable it. Default tolerances are loose (5 points) to absorb run-to-run variance. - **Output isolation.** The transcript directory is keyed on the full `-out` filename (`results.json` -> `results.json.transcripts/`), so several passes written into the same directory under different output names — even ones sharing a stem but differing by extension — never overwrite one another's raw transcripts. The `-merge` step refuses an `-out` that is the same on-disk file as one of its input passes (compared by device+inode, so a case-variant or symlinked alias is caught too), so a merged scorecard never clobbers the raw per-pass evidence it was built from. Together these mean multi-pass orchestration cannot silently discard paid-for data. -- **claude-cli cache tokens.** A cached `claude -p` run's `cache_read_input_tokens` / `cache_creation_input_tokens` flow through the stream parser into each lifecycle `EpisodeRecord` and are summed into the run's `total_cache_read_tokens` / `total_cache_creation_tokens`, so a cached run self-reports its true cost basis (cache reads bill far below fresh input). The full parser -> `EpisodeRecord` -> aggregate path is covered by a test that drives the real parser on a canned cached stream; that a real `claude` process emits those fields is confirmed by one real cached run. +- **claude-cli cache tokens.** A cached `claude -p` run's `cache_read_input_tokens` / `cache_creation_input_tokens` flow through the stream parser into each lifecycle `EpisodeRecord` and are summed into the run's `total_cache_read_tokens` / `total_cache_creation_tokens`, so a cached run self-reports its true cost basis (cache reads bill far below fresh input). The full parser -> `EpisodeRecord` -> aggregate path is covered by a test that drives the real parser on a canned cached stream; confirming that a real `claude` process emits those fields end to end is pending the first real claude-cli run on this parser (every kept claude-cli results file predates it), and the runner warns loudly when an episode with tool calls reports zero usage so a silent field move cannot go unnoticed. ### Statistical power and identity-pool sizing (#965) @@ -342,10 +356,12 @@ metric (0% vs 42.9% between identical runs); measuring supersede on its own, wit a per-protocol stability breakdown, makes that instability visible per protocol instead of hidden in one blended range. -Metrics: **capture rate**, **supersede rate** (original superseded, higher is -better), **duplicate rate** (its complement, lower is better), **update -correctness**, **pass^k**, and a per-protocol `superseded`/`duplicated` count -across the k attempts. Because the supersede gate is embedding-similarity based +Metrics: **capture rate**, **update capture rate** (the correction capture +executed; a miss is excluded from the supersede/duplicate denominator — with no +correction on the platform the gate never ran), **supersede rate** (original +superseded, higher is better), **duplicate rate** (its complement, lower is +better), **update correctness**, **pass^k**, and a per-protocol +`superseded`/`duplicated`/`update_capture_missed` count across the k attempts. Because the supersede gate is embedding-similarity based (nomic-embed-text), the threshold / embedding-model / deterministic-fallback evaluation is done by re-running this sub-benchmark against platforms configured differently and comparing the supersede rate — the sub-benchmark is the measuring @@ -356,7 +372,7 @@ make bench-up BENCH_ARM=a3 make bench-supersede-smoke # scripted no-API-key validation make bench-supersede K=3 # real run (needs ANTHROPIC_API_KEY) make bench-supersede K=3 TEACH_BUDGET=20 # same, with the larger teach budget lever -make bench-supersede-report # human summary of the last supersede run +make bench-supersede-report RESULTS=build/bench-results/supersede-a3-/supersede-a3.json ``` ## Cold-start knowledge growth (#963) @@ -384,7 +400,12 @@ trap class — the harness: aspects — tags, the structured deprecation flag, column docs — are not, but the S3 traps read the fact text, not those). Capture and promotion are verified through the admin insights and changesets APIs, reusing the same reviewer-promotion - path (`internal/promote`) the S5 lifecycle uses. + path (`internal/promote`) the S5 lifecycle uses; after the API verify, the + reviewer also reads the promoted content back from its sink (the entity's + effective description via `datahub_get_entity`, the page's summary via the + portal list) and fails the run when an API-confirmed apply is not readable + there — a silent sink-write loss must abort before episodes are spent, not + surface later as an unexplained flat curve. 2. **Evaluates** at every checkpoint (the empty baseline and after each lesson) by re-running the fixed S3 trap suite with a **fresh, never-taught evaluator identity**. Its only knowledge source is what the platform surfaces — @@ -433,8 +454,11 @@ and the scripted smoke runs with `SETTLE=0s`. 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). +baseline, no evaluator memory-write warning (evaluators are forbidden from +saving memories; the summary warns if the audit-side count is non-zero), and no +audit-read-back warning (a lost audit read contributes zero to coverage through +signal loss, recorded per attempt as `audit_read_error` and totaled on the +summary). ## Running @@ -458,7 +482,7 @@ auto-enable): make bench-up BENCH_ARM=a3 # boot the lifecycle arm make bench-lifecycle-smoke # scripted no-API-key lifecycle validation make bench-lifecycle K=3 # real run (needs ANTHROPIC_API_KEY) -make bench-lifecycle-report # human summary of the last lifecycle run +make bench-lifecycle-report RESULTS=build/bench-results/lifecycle-a3-/lifecycle-a3.json ``` The **scripted lifecycle smoke** (`-llm scripted`) plays diff --git a/bench/benchrun/main.go b/bench/benchrun/main.go index 7848edd53..3ddc9e5a1 100644 --- a/bench/benchrun/main.go +++ b/bench/benchrun/main.go @@ -44,6 +44,7 @@ type config struct { httpTimeout time.Duration llmTimeout time.Duration auditTimeout time.Duration + sinkTimeout time.Duration identityKeys int summarize string compare string @@ -90,6 +91,7 @@ func parseFlags() config { flag.DurationVar(&cfg.httpTimeout, "http-timeout", 120*time.Second, "platform HTTP timeout") flag.DurationVar(&cfg.llmTimeout, "llm-timeout", 5*time.Minute, "model API request timeout") flag.DurationVar(&cfg.auditTimeout, "audit-timeout", 15*time.Second, "audit read-back timeout per session") + flag.DurationVar(&cfg.sinkTimeout, "sink-timeout", 0, "post-apply sink read-back window for promotions (0 = the promote default, 15s); raise for a store that serves reads slowly") flag.IntVar(&cfg.identityKeys, "identity-keys", 264, "per-attempt identity pool size matching the arm config (0 = single identity)") flag.StringVar(&cfg.summarize, "summarize", "", "print the human summary of an existing results JSON and exit") flag.StringVar(&cfg.compare, "compare", "", "comma-separated per-arm results JSON files: render the cross-arm comparison and exit") @@ -189,6 +191,9 @@ func runLifecycle(cfg config) error { if cfg.arm == "" { return errors.New("-arm is required") } + if err := refuseExistingRunArtifacts(cfg.out); err != nil { + return err + } log := slog.New(slog.NewTextHandler(os.Stderr, nil)) opts := lifecycle.Options{ Target: target.Target{BaseURL: cfg.url, Credential: cfg.credential}, @@ -200,6 +205,7 @@ func runLifecycle(cfg config) error { LLMProvider: cfg.llmProvider, GitCommit: cfg.gitCommit, AuditTimeout: cfg.auditTimeout, + SinkTimeout: cfg.sinkTimeout, IdentityKeys: cfg.identityKeys, TeachBudget: cfg.teachBudget, // Flush the results file after every protocol so an interruption (timeout, @@ -283,6 +289,9 @@ func runSupersede(cfg config) error { if cfg.baseline != "" { return errors.New("-baseline is not supported with -supersede (the regression gate scores S1-S3 task suites, not supersede metrics)") } + if err := refuseExistingRunArtifacts(cfg.out); err != nil { + return err + } log := slog.New(slog.NewTextHandler(os.Stderr, nil)) opts := lifecycle.Options{ Target: target.Target{BaseURL: cfg.url, Credential: cfg.credential}, @@ -294,6 +303,7 @@ func runSupersede(cfg config) error { LLMProvider: cfg.llmProvider, GitCommit: cfg.gitCommit, AuditTimeout: cfg.auditTimeout, + SinkTimeout: cfg.sinkTimeout, IdentityKeys: cfg.identityKeys, TeachBudget: cfg.teachBudget, OnSupersede: func(r *lifecycle.SupersedeResults) { @@ -342,6 +352,7 @@ func runColdStart(cfg config) error { LLMProvider: cfg.llmProvider, GitCommit: cfg.gitCommit, AuditTimeout: cfg.auditTimeout, + SinkTimeout: cfg.sinkTimeout, IdentityKeys: cfg.identityKeys, Settle: cfg.settle, OnCheckpoint: func(r *coldstart.Results) { @@ -555,8 +566,11 @@ func foldPass(merged *lifecycle.Results, path string, pass int) error { } // sameConfig refuses to merge passes that were not produced under the same arm, -// protocol set, model, and seed — merging across configurations would publish a -// scorecard mislabeled with pass 1's manifest. +// protocol set, model, client path, and seed — merging across configurations +// would publish a scorecard mislabeled with pass 1's manifest. The client path +// is compared by provider (anthropic vs claude-cli vs scripted), not exact CLI +// version, mirroring BaselineCompatible: the paths produce incomparable numbers, +// while a benign `claude` patch bump must not split a multi-pass run. func sameConfig(a, b lifecycle.Manifest, path string) error { switch { case a.Arm != b.Arm: @@ -565,6 +579,8 @@ func sameConfig(a, b lifecycle.Manifest, path string) error { return fmt.Errorf("merge: %s protocol-set hash %q != %q", path, b.ProtocolSetHash, a.ProtocolSetHash) case a.Model != b.Model: return fmt.Errorf("merge: %s model %q != %q", path, b.Model, a.Model) + case a.LLMProvider != b.LLMProvider: + return fmt.Errorf("merge: %s llm provider %q != %q — anthropic and claude-cli passes are not comparable", path, b.LLMProvider, a.LLMProvider) case a.Seed != b.Seed: return fmt.Errorf("merge: %s seed %d != %d", path, b.Seed, a.Seed) } @@ -713,13 +729,18 @@ func transcriptDir(out string) string { } // 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 +// directory already exists. A 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. +// The timestamped-run modes (-cold-start, -lifecycle, -supersede) enforce this; +// the default S1-S3 task mode deliberately does NOT: its -out is the fixed +// per-arm slot (build/bench-results/results-.json) that bench-compare and +// the CI regression gate consume, so overwriting the slot on each run is that +// mode's documented contract — copy a slot result aside (as bench/results/ +// does) before rerunning an arm. There is deliberately no override on the +// guarded modes: 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) diff --git a/bench/benchrun/main_test.go b/bench/benchrun/main_test.go index a19a36455..5a7efde41 100644 --- a/bench/benchrun/main_test.go +++ b/bench/benchrun/main_test.go @@ -47,6 +47,22 @@ func TestMergeRejectsConfigMismatch(t *testing.T) { } } +// TestMergeRejectsClientPathMismatch proves the merge refuses to fold an +// anthropic pass and a claude-cli pass into one scorecard: the two client paths +// produce incomparable numbers (BaselineCompatible refuses to gate across them, +// so a merge must not silently blend them either). +func TestMergeRejectsClientPathMismatch(t *testing.T) { + base := lifecycle.Manifest{Arm: "a3", ProtocolSetHash: "h1", Model: "claude-sonnet-5", Seed: 930, LLMProvider: "anthropic"} + other := base + other.LLMProvider = "claude-cli" + if err := sameConfig(base, other, "p2.json"); err == nil { + t.Fatal("sameConfig accepted an anthropic pass merged with a claude-cli pass") + } + if err := sameConfig(base, base, "p2.json"); err != nil { + t.Fatalf("sameConfig rejected identical configs: %v", err) + } +} + // TestMergeSucceedsAndComputesPassK proves three valid k=1 passes merge to k=3 // with pass^k computed over the passes (each protocol has one run per pass). func TestMergeSucceedsAndComputesPassK(t *testing.T) { @@ -178,6 +194,10 @@ func healthyLifecycleMetrics() lifecycle.Metrics { PersonalRecall: lifecycleRate(9, 10), TransferRate: lifecycleRate(8, 10), UpdateCorrectness: lifecycleRate(5, 5), + // A current-harness run that scores duplicates always carries + // update-capture data; without it BaselineCompatible refuses the + // baseline as a pre-redefinition artifact. + UpdateCaptureRate: lifecycleRate(5, 5), AbstentionRate: lifecycleRate(9, 10), DuplicateRate: lifecycleRate(0, 5), PassK: lifecycleRate(7, 10), diff --git a/bench/internal/claudecli/claudecli.go b/bench/internal/claudecli/claudecli.go index 79efbbd97..02d111446 100644 --- a/bench/internal/claudecli/claudecli.go +++ b/bench/internal/claudecli/claudecli.go @@ -22,10 +22,13 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "os" "os/exec" "path/filepath" "strings" + + "github.com/txn2/mcp-data-platform/bench/internal/llm" ) // defaultDisallowedTools are the built-in Claude Code tools the bench run @@ -65,6 +68,9 @@ type Options struct { // and lifecycle paths that drive it — are exercisable without the real // claude binary. Exec CommandRunner + // Log receives runner warnings (e.g. zero-usage detection); nil uses + // slog.Default(). + Log *slog.Logger } // CommandSpec is one child-process invocation. @@ -106,6 +112,9 @@ func New(opts Options) (*Runner, error) { if opts.Exec == nil { opts.Exec = execCommand } + if opts.Log == nil { + opts.Log = slog.Default() + } return &Runner{opts: opts}, nil } @@ -171,6 +180,13 @@ func (r *Runner) Run(ctx context.Context, req Request) (Result, error) { } return Result{}, fmt.Errorf("%w; claude exit: %s; stderr: %.500s", parseErr, exit, strings.TrimSpace(string(stderr))) } + // An episode that drove tools but reports no token usage means the stream's + // usage field moved (a claude release change): the run still grades, but its + // self-reported cost basis is silently zero. Warn loudly so an expensive run + // is not published with a garbage token total. + if res.MCPCalls > 0 && res.Usage == (llm.Usage{}) { + r.opts.Log.Warn("claude-cli stream carried no token usage for an episode with tool calls; the stream shape may have changed and the run's token totals will under-report") + } return res, nil } @@ -234,13 +250,14 @@ func writeMCPConfig(path, server, endpoint, credential string) error { // execCommand is the default commandRunner. It captures stdout and stderr // separately and returns the process error (if any) alongside both, so the // caller can parse a partial stream even on a non-zero exit. The child env -// strips ANTHROPIC_API_KEY so the run is always subscription-funded: a key -// sourced for the metered anthropic adapter must never silently make a -// claude-cli run bill the API (issue #949's whole point is the keyless path). +// strips every metered-billing credential so the run is always +// subscription-funded: a key sourced for the metered anthropic adapter must +// never silently make a claude-cli run bill the API (issue #949's whole point +// is the keyless path). func execCommand(ctx context.Context, spec CommandSpec) ([]byte, []byte, error) { cmd := exec.CommandContext(ctx, spec.Bin, spec.Args...) // #nosec G204 -- bin is operator-configured, args are harness-built cmd.Dir = spec.Dir - cmd.Env = envWithoutAPIKey(os.Environ()) + cmd.Env = envWithoutMeteredCreds(os.Environ()) if spec.Stdin != nil { cmd.Stdin = bytes.NewReader(spec.Stdin) } @@ -251,16 +268,37 @@ func execCommand(ctx context.Context, spec CommandSpec) ([]byte, []byte, error) return stdout.Bytes(), stderr.Bytes(), err } -// envWithoutAPIKey returns env with any ANTHROPIC_API_KEY entry removed, so the -// child claude authenticates against the logged-in subscription rather than a -// metered API key. -func envWithoutAPIKey(env []string) []string { +// meteredCredVars are the environment variables that would route the child +// claude onto metered billing: the raw API key, the bearer-token override, and +// the Bedrock/Vertex provider switches. All are stripped so the child always +// authenticates against the logged-in subscription. +var meteredCredVars = []string{ + "ANTHROPIC_API_KEY", + "ANTHROPIC_AUTH_TOKEN", + "CLAUDE_CODE_USE_BEDROCK", + "CLAUDE_CODE_USE_VERTEX", +} + +// envWithoutMeteredCreds returns env with every metered-billing credential +// entry removed, so the child claude authenticates against the logged-in +// subscription rather than a metered path. +func envWithoutMeteredCreds(env []string) []string { out := make([]string, 0, len(env)) for _, kv := range env { - if strings.HasPrefix(kv, "ANTHROPIC_API_KEY=") { + if hasMeteredCred(kv) { continue } out = append(out, kv) } return out } + +// hasMeteredCred reports whether one KEY=value entry names a metered credential. +func hasMeteredCred(kv string) bool { + for _, name := range meteredCredVars { + if strings.HasPrefix(kv, name+"=") { + return true + } + } + return false +} diff --git a/bench/internal/claudecli/claudecli_test.go b/bench/internal/claudecli/claudecli_test.go index 2c0b62c42..63e3f576f 100644 --- a/bench/internal/claudecli/claudecli_test.go +++ b/bench/internal/claudecli/claudecli_test.go @@ -165,12 +165,22 @@ func TestExecCommandRealProcess(t *testing.T) { } } -func TestEnvWithoutAPIKey(t *testing.T) { - in := []string{"PATH=/bin", "ANTHROPIC_API_KEY=secret", "HOME=/home/x", "ANTHROPIC_API_KEY_OTHER=keep"} - out := envWithoutAPIKey(in) +func TestEnvWithoutMeteredCreds(t *testing.T) { + in := []string{ + "PATH=/bin", + "ANTHROPIC_API_KEY=secret", + "ANTHROPIC_AUTH_TOKEN=bearer", + "CLAUDE_CODE_USE_BEDROCK=1", + "CLAUDE_CODE_USE_VERTEX=1", + "HOME=/home/x", + "ANTHROPIC_API_KEY_OTHER=keep", + } + out := envWithoutMeteredCreds(in) for _, kv := range out { - if strings.HasPrefix(kv, "ANTHROPIC_API_KEY=") { - t.Errorf("ANTHROPIC_API_KEY not stripped: %v", out) + for _, name := range meteredCredVars { + if strings.HasPrefix(kv, name+"=") { + t.Errorf("%s not stripped: %v", name, out) + } } } // A similarly-named var must not be stripped. @@ -184,7 +194,7 @@ func TestEnvWithoutAPIKey(t *testing.T) { t.Errorf("stripped an unrelated var: %v", out) } if len(out) != 3 { - t.Errorf("got %d entries, want 3", len(out)) + t.Errorf("got %d entries, want 3 (PATH, HOME, the unrelated var): %v", len(out), out) } } diff --git a/bench/internal/claudecli/parse.go b/bench/internal/claudecli/parse.go index 9e89375af..471f82e71 100644 --- a/bench/internal/claudecli/parse.go +++ b/bench/internal/claudecli/parse.go @@ -248,7 +248,9 @@ func (p *parser) applyAssistant(raw json.RawMessage) { // recordToolUse classifies one tool_use block by name: search flips the // surfaced flag, platform_info is tracked for handle extraction but not counted, -// and any other bench tool is counted toward the audit lower/upper bounds. +// and any other bench tool enters the deduplicated set the audit lower/upper +// bounds are derived from (MCPCalls is computed in tallyErrors, so a replayed +// assistant event carrying the same tool_use id cannot inflate it). func (p *parser) recordToolUse(id, name string) { if !strings.HasPrefix(name, p.prefix) { return @@ -257,12 +259,10 @@ func (p *parser) recordToolUse(id, name string) { case searchToolName: p.res.SearchCalled = true p.toolIsBench[id] = true - p.res.MCPCalls++ case infoToolName: p.infoIDs[id] = true default: p.toolIsBench[id] = true - p.res.MCPCalls++ } } @@ -334,12 +334,14 @@ func (p *parser) applyResult(ev streamLine) { // tallyErrors folds the per-call error flags into success and error counts: // every non-error bench call is a confirmed audit row (the lower bound), and // error calls are only the upper bound (a gate refusal leaves no row, a handler -// error does). +// error does). MCPCalls is the deduplicated bench tool_use count, so all three +// counters come from one id set and stay mutually consistent. func (p *parser) tallyErrors() { + p.res.MCPCalls = len(p.toolIsBench) for id := range p.toolIsBench { if !p.toolDone[id] { - // No paired result: indeterminate. Already counted in MCPCalls (the - // upper bound); must not enter the confirmed-success lower bound. + // No paired result: indeterminate. Counted in MCPCalls (the upper + // bound); must not enter the confirmed-success lower bound. continue } if p.toolErr[id] { diff --git a/bench/internal/claudecli/parse_test.go b/bench/internal/claudecli/parse_test.go index 3f0a01f55..0f7b10d49 100644 --- a/bench/internal/claudecli/parse_test.go +++ b/bench/internal/claudecli/parse_test.go @@ -196,6 +196,28 @@ func TestParseUnpairedToolUseNotCountedSuccessful(t *testing.T) { } } +func TestParseReplayedToolUseNotDoubleCounted(t *testing.T) { + // A replayed assistant event carrying an already-seen tool_use id must not + // inflate MCPCalls: the count is derived from the deduplicated id set, so it + // stays consistent with SuccessfulMCPCalls/ToolErrors and the audit + // read-back's overcount detection keeps its teeth. + stream := `{"type":"system","subtype":"init","mcp_servers":[{"name":"bench","status":"connected"}]} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"q1","name":"mcp__bench__trino_query","input":{}}]}} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"q1","name":"mcp__bench__trino_query","input":{}}]}} +{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"q1","content":"ok","is_error":false}]}} +{"type":"result","subtype":"success","is_error":false,"result":"FINAL ANSWER: x","usage":{"input_tokens":1,"output_tokens":1}}` + res, err := Parse("bench", "q", []byte(stream)) + if err != nil { + t.Fatalf("Parse: %v", err) + } + if res.MCPCalls != 1 { + t.Errorf("MCPCalls = %d, want 1 (replayed tool_use id deduplicated)", res.MCPCalls) + } + if res.SuccessfulMCPCalls != 1 { + t.Errorf("SuccessfulMCPCalls = %d, want 1", res.SuccessfulMCPCalls) + } +} + func TestParseNonStringResult(t *testing.T) { // A terminal result event whose `result` is not a bare string must still be // recognized as the result event (not dropped, which would report a spurious diff --git a/bench/internal/coldstart/episode.go b/bench/internal/coldstart/episode.go index b973a4c11..53efa6375 100644 --- a/bench/internal/coldstart/episode.go +++ b/bench/internal/coldstart/episode.go @@ -108,6 +108,11 @@ type episodeResult struct { wallMS int64 usage llm.Usage audit auditapi.Metrics + // auditReadErr records a failed audit read-back on an otherwise-successful + // episode. The attempt still grades, but its zero audit metrics must not + // pass for "no enrichment": pooled coverage is a documented pass criterion, + // so audit-signal loss is carried on the record instead of only logged. + auditReadErr string finalAnswer string err string } @@ -238,7 +243,7 @@ func (e *runEnv) runEpisode(ctx context.Context, spec episodeSpec) episodeResult res.err = fmt.Sprintf("agent loop: %v", runErr) return res } - res.audit = e.readAudit(ctx, info.Handle, audited, audited+indeterminate) + res.audit, res.auditReadErr = e.readAudit(ctx, info.Handle, audited, audited+indeterminate) return res } @@ -280,23 +285,34 @@ func (e *runEnv) runClaudeCLIEpisode(ctx context.Context, spec episodeSpec) epis res.err = fmt.Sprintf("bench MCP server did not connect (status %q)", cres.ServerStatus) return res } - if cres.Handle != "" { - res.audit = e.readAudit(ctx, cres.Handle, cres.SuccessfulMCPCalls, cres.MCPCalls) + if cres.Handle == "" { + // No handle means claude never minted one via platform_info. With no + // handle it cannot have threaded a successful data call (the session-gate + // middleware refuses un-threaded calls), so a positive success count with + // no handle is a harness inconsistency to surface, not audit loss + // (mirrors the S1-S3 pipeline's contract). + if cres.SuccessfulMCPCalls > 0 { + res.err = fmt.Sprintf("claude-cli reported %d successful tool call(s) but surfaced no dps_ handle to correlate audit (platform_info result missing or unparseable)", cres.SuccessfulMCPCalls) + } + return res } + res.audit, res.auditReadErr = e.readAudit(ctx, cres.Handle, cres.SuccessfulMCPCalls, cres.MCPCalls) return res } // readAudit reads the session's audit trail back best effort. A missing row does // not fail an episode: lesson state comes from the knowledge API and eval // correctness from grading; audit only enriches the efficiency and coverage -// picture. A read failure yields zero metrics and is logged. -func (e *runEnv) readAudit(ctx context.Context, handle string, minAudited, maxAudited int) auditapi.Metrics { +// picture. A read failure yields zero metrics plus the error, which callers +// record on the attempt so audit-signal loss is visible in results.json (the +// coverage curve must never silently degrade to zeros). +func (e *runEnv) readAudit(ctx context.Context, handle string, minAudited, maxAudited int) (auditapi.Metrics, string) { events, err := e.audit.WaitForSession(ctx, handle, minAudited, maxAudited, e.opts.AuditTimeout) if err != nil { e.log.Warn("cold-start audit read-back", "handle", handle, "error", err) - return auditapi.Metrics{} + return auditapi.Metrics{}, err.Error() } - return auditapi.Summarize(events) + return auditapi.Summarize(events), "" } // writeTranscript persists a loop episode's transcript for manual audit. diff --git a/bench/internal/coldstart/report.go b/bench/internal/coldstart/report.go index 3a323829e..c2a4a52e4 100644 --- a/bench/internal/coldstart/report.go +++ b/bench/internal/coldstart/report.go @@ -58,7 +58,10 @@ type EpisodeRecord struct { CacheReadTokens int64 `json:"cache_read_tokens,omitempty"` CacheCreationTokens int64 `json:"cache_creation_tokens,omitempty"` Audit auditapi.Metrics `json:"audit"` - Error string `json:"error,omitempty"` + // AuditReadError records a failed audit read-back on an otherwise-successful + // episode: its zero audit metrics mean signal loss, not zero enrichment. + AuditReadError string `json:"audit_read_error,omitempty"` + Error string `json:"error,omitempty"` } // LessonRecord captures one lesson's teach-and-promote outcome. Captured and @@ -102,7 +105,12 @@ type EvalAttempt struct { CacheReadTokens int64 `json:"cache_read_tokens,omitempty"` CacheCreationTokens int64 `json:"cache_creation_tokens,omitempty"` Audit auditapi.Metrics `json:"audit"` - Error string `json:"error,omitempty"` + // AuditReadError records a failed audit read-back on a graded attempt: the + // attempt contributes nothing to enrichment coverage because the signal was + // lost, not because no call was enriched. Any non-zero count across a run is + // a coverage-integrity warning on the summary. + AuditReadError string `json:"audit_read_error,omitempty"` + Error string `json:"error,omitempty"` } // ClassScore is one trap class's accuracy at a checkpoint. @@ -187,6 +195,11 @@ type Metrics struct { // 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"` + // AuditReadFailures counts graded episodes (lessons and eval attempts) whose + // audit read-back failed. Each contributes zero to enrichment coverage + // through signal loss, so any non-zero value flags the coverage curve's + // integrity. + AuditReadFailures int `json:"audit_read_failures,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 @@ -216,29 +229,13 @@ type Results struct { func (res *Results) Aggregate() { m := Metrics{Lessons: len(res.Lessons), Checkpoints: len(res.Checkpoints)} for _, l := range res.Lessons { - if boolTrue(l.Captured) { - m.LessonsCaptured++ - } - if boolTrue(l.Promoted) { - m.LessonsPromoted++ - } - if l.Error != "" { - m.HarnessFailures++ - } - m.TotalInputTokens += l.Episode.InputTokens - m.TotalOutputTokens += l.Episode.OutputTokens - m.TotalCacheReadTokens += l.Episode.CacheReadTokens - m.TotalCacheCreationTokens += l.Episode.CacheCreationTokens + m.foldLesson(l) } for i := range res.Checkpoints { 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 - m.TotalCacheCreationTokens += a.CacheCreationTokens + m.foldAttempt(a) } } if len(res.Checkpoints) > 0 { @@ -251,6 +248,40 @@ func (res *Results) Aggregate() { res.Metrics = m } +// foldLesson accumulates one lesson's outcome, validity signals, and token +// spend into the scorecard. +func (m *Metrics) foldLesson(l LessonRecord) { + if boolTrue(l.Captured) { + m.LessonsCaptured++ + } + if boolTrue(l.Promoted) { + m.LessonsPromoted++ + } + if l.Error != "" { + m.HarnessFailures++ + } + if l.Episode.AuditReadError != "" { + m.AuditReadFailures++ + } + m.TotalInputTokens += l.Episode.InputTokens + m.TotalOutputTokens += l.Episode.OutputTokens + m.TotalCacheReadTokens += l.Episode.CacheReadTokens + m.TotalCacheCreationTokens += l.Episode.CacheCreationTokens +} + +// foldAttempt accumulates one eval attempt's validity signals and token spend +// into the scorecard. +func (m *Metrics) foldAttempt(a EvalAttempt) { + m.EvalMemoryWrites += a.MemoryWrites + if a.AuditReadError != "" { + m.AuditReadFailures++ + } + m.TotalInputTokens += a.InputTokens + m.TotalOutputTokens += a.OutputTokens + m.TotalCacheReadTokens += a.CacheReadTokens + m.TotalCacheCreationTokens += a.CacheCreationTokens +} + // distinctTasks counts the unique task IDs in a checkpoint's attempts (the eval // set size, independent of k). func distinctTasks(attempts []EvalAttempt) int { @@ -316,6 +347,9 @@ func (res *Results) HumanSummary() string { 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) } + if mt.AuditReadFailures > 0 { + fmt.Fprintf(&b, "WARNING: %d episode(s) lost their audit read-back; each contributes zero to enrichment coverage through signal loss, so the coverage curve under-reports\n", mt.AuditReadFailures) + } 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 be6f6c119..5600bb34f 100644 --- a/bench/internal/coldstart/report_test.go +++ b/bench/internal/coldstart/report_test.go @@ -145,6 +145,23 @@ func TestEvalMemoryWritesWarn(t *testing.T) { } } +// TestAuditReadFailuresWarn proves a lost audit read-back (on a lesson episode +// or an eval attempt) is totaled in Metrics and surfaces as a coverage-integrity +// warning in the human summary: the attempt contributes zero to enrichment +// coverage through signal loss, which must never read as "nothing was enriched". +func TestAuditReadFailuresWarn(t *testing.T) { + res := sampleResults() + res.Lessons[0].Episode.AuditReadError = "audit rows below minimum" + res.Checkpoints[1].Attempts[0].AuditReadError = "audit rows below minimum" + res.Aggregate() + if res.Metrics.AuditReadFailures != 2 { + t.Errorf("AuditReadFailures = %d, want 2", res.Metrics.AuditReadFailures) + } + if !strings.Contains(res.HumanSummary(), "WARNING: 2 episode(s) lost their audit read-back") { + t.Errorf("summary missing the audit-read-back coverage warning\n%s", res.HumanSummary()) + } +} + func TestAggregateEmptyIsSafe(t *testing.T) { res := &Results{} res.Aggregate() // must not panic on no lessons/checkpoints diff --git a/bench/internal/coldstart/runner.go b/bench/internal/coldstart/runner.go index 856ab8ec9..6b1ff946d 100644 --- a/bench/internal/coldstart/runner.go +++ b/bench/internal/coldstart/runner.go @@ -48,6 +48,9 @@ type Options struct { LLMProvider string GitCommit string AuditTimeout time.Duration + // SinkTimeout bounds the reviewer's post-apply sink read-back (zero uses the + // promote package default); raise it for a store that serves reads slowly. + SinkTimeout time.Duration // IdentityKeys is the identity-pool size the arm config defines. A run refuses // to start when the lessons + per-checkpoint evaluators exceed the pool. It // must be positive: the teacher and every checkpoint's evaluators must be @@ -98,7 +101,7 @@ func Run(ctx context.Context, opts Options) (*Results, error) { log: opts.Log, audit: auditapi.New(opts.Target.BaseURL, opts.Target.HTTPClient(opts.HTTPTimeout)), life: life, - reviewer: promote.Reviewer{Life: life, Log: opts.Log}, + reviewer: promote.Reviewer{Life: life, Log: opts.Log, SinkTimeout: opts.SinkTimeout}, } defer env.closeAdmin() @@ -352,7 +355,7 @@ func teachEpisode(rec episodeResult) EpisodeRecord { Email: rec.email, SessionID: rec.sessionID, ToolCalls: rec.toolCalls, ToolErrors: rec.toolErrors, WallMS: rec.wallMS, InputTokens: rec.usage.InputTokens, OutputTokens: rec.usage.OutputTokens, CacheReadTokens: rec.usage.CacheReadInputTokens, CacheCreationTokens: rec.usage.CacheCreationInputTokens, - Audit: rec.audit, Error: rec.err, + Audit: rec.audit, AuditReadError: rec.auditReadErr, Error: rec.err, } } @@ -384,7 +387,7 @@ func (e *runEnv) evalAttempt(ctx context.Context, t task.Task, seq, repeat int) 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, + Audit: rec.audit, AuditReadError: rec.auditReadErr, } if rec.err != "" { att.Error = rec.err diff --git a/bench/internal/coldstart/runner_test.go b/bench/internal/coldstart/runner_test.go index a824f2b98..52e1138e6 100644 --- a/bench/internal/coldstart/runner_test.go +++ b/bench/internal/coldstart/runner_test.go @@ -211,14 +211,33 @@ func (fp *fakePlatform) addApplyKnowledge(server *mcp.Server) { } } } + fp.applySinkLocked(urn, args) fp.recordLocked(args, "apply_knowledge") return okResult("applied " + csID), nil, nil }) } -// 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). +// applySinkLocked models the platform landing an applied change in its sink, +// which the Reviewer's post-apply sink read-back verifies: a datahub change +// becomes the entity's effective description, a page payload becomes a live +// knowledge page. The caller holds fp.mu. +func (fp *fakePlatform) applySinkLocked(urn string, args map[string]any) { + if changes, ok := args["changes"].([]any); ok && len(changes) > 0 { + if change, ok := changes[0].(map[string]any); ok { + detail, _ := change["detail"].(string) + fp.entityDesc[urn] = detail + } + } + if page, ok := args["page"].(map[string]any); ok { + slug, _ := page["slug"].(string) + summary, _ := page["summary"].(string) + fp.pages = append(fp.pages, lifecycleapi.KnowledgePage{ID: "kp-" + slug, Slug: slug, Summary: summary}) + } +} + +// addGetEntity serves the preflight's baseline read and the promote sink +// read-back: 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}, @@ -1090,6 +1109,78 @@ func TestClaudeCLIEpisode(t *testing.T) { } } +// claudeNoHandleStream is a canned transcript where claude drove a successful +// bench call but never minted a dps_ handle via platform_info. +const claudeNoHandleStream = `{"type":"system","subtype":"init","mcp_servers":[{"name":"bench","status":"connected"}]} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"s1","name":"mcp__bench__search","input":{"query":"units"}}]}} +{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"s1","is_error":false,"content":"ok"}]}} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"FINAL ANSWER: 100.00"}]}} +{"type":"result","subtype":"success","is_error":false,"result":"FINAL ANSWER: 100.00","session_id":"cc-2","usage":{"input_tokens":30,"output_tokens":8}}` + +// TestClaudeCLIEpisodeNoHandleWithSuccesses proves a claude-cli episode that +// reports successful tool calls but no dps_ handle is a harness error, not a +// silent zero-coverage attempt: with no handle a successful data call is +// impossible (the session gate refuses un-threaded calls), so the inconsistency +// must surface rather than degrade the coverage curve. +func TestClaudeCLIEpisodeNoHandleWithSuccesses(t *testing.T) { + fp := newFakePlatform(t) + runner, err := claudecli.New(claudecli.Options{ + Model: "claude-sonnet-5", + Exec: func(_ context.Context, _ claudecli.CommandSpec) ([]byte, []byte, error) { + return []byte(claudeNoHandleStream), nil, nil + }, + }) + if err != nil { + t.Fatalf("claudecli.New: %v", err) + } + env := &runEnv{ + opts: Options{ + Target: target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"}, + HTTPTimeout: 10 * time.Second, Arm: "a3", ClaudeCLI: runner, + IdentityKeys: 32, AuditTimeout: time.Second, + }, + log: testLogger(), + audit: auditapi.New(fp.httpSrv.URL, target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"}.HTTPClient(10*time.Second)), + } + rec := env.runEpisode(context.Background(), episodeSpec{stage: StageEval, unitID: "s3-units-a", seq: 7, prompt: "q", budget: 5}) + if rec.err == "" || !strings.Contains(rec.err, "no dps_ handle") { + t.Fatalf("episode err = %q, want a no-handle harness error", rec.err) + } +} + +// TestClaudeCLIEpisodeAuditReadFailureRecorded proves a failed audit read-back +// lands on the episode result instead of silently zeroing the audit metrics: +// the fake platform has no audit rows for the handle, so WaitForSession times +// out and the loss must be visible on the record. +func TestClaudeCLIEpisodeAuditReadFailureRecorded(t *testing.T) { + fp := newFakePlatform(t) // no audit events appended for dps_cc_1 + runner, err := claudecli.New(claudecli.Options{ + Model: "claude-sonnet-5", + Exec: func(_ context.Context, _ claudecli.CommandSpec) ([]byte, []byte, error) { + return []byte(claudeEvalStream), nil, nil + }, + }) + if err != nil { + t.Fatalf("claudecli.New: %v", err) + } + env := &runEnv{ + opts: Options{ + Target: target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"}, + HTTPTimeout: 10 * time.Second, Arm: "a3", ClaudeCLI: runner, + IdentityKeys: 32, AuditTimeout: 200 * time.Millisecond, + }, + log: testLogger(), + audit: auditapi.New(fp.httpSrv.URL, target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"}.HTTPClient(10*time.Second)), + } + rec := env.runEpisode(context.Background(), episodeSpec{stage: StageEval, unitID: "s3-units-a", seq: 7, prompt: "q", budget: 5}) + if rec.err != "" { + t.Fatalf("episode err = %q, want graded episode (audit loss is not a harness failure)", rec.err) + } + if rec.auditReadErr == "" { + t.Fatal("auditReadErr empty: a lost audit read-back must be recorded, not silently zeroed") + } +} + func TestGradeEval(t *testing.T) { numeric := task.Grading{Kind: task.GradeNumeric, Value: new(42.0), AbsTolerance: 0.5} if !gradeEval("FINAL ANSWER: 42.1", numeric) { diff --git a/bench/internal/grade/grade.go b/bench/internal/grade/grade.go index 1987aa698..3fb248988 100644 --- a/bench/internal/grade/grade.go +++ b/bench/internal/grade/grade.go @@ -14,6 +14,8 @@ import ( "regexp" "strconv" "strings" + "unicode" + "unicode/utf8" ) // finalMarker matches the answer convention the system prompt mandates. The @@ -72,17 +74,67 @@ func Numeric(final string, expected, absTolerance float64) (got float64, ok, cor // AND no wrong alias does: the wrong-alias list enumerates the task's known // trap answers (the deprecated table, the gross-revenue region), so a verbose // answer that names the trap while mentioning the truth is not credited. +// Aliases match on word boundaries, not bare substrings: "East" must not veto +// "at least", and "West" must not match "southwest". Letters, digits, and +// underscores are word characters; dots are boundaries, so a schema-qualified +// alias still matches inside a longer qualified name. The boundary rule also +// narrows the veto: an answer naming a suffixed variant of a trap identifier +// ("legacy_orders_v9") is not vetoed by the "legacy_orders" wrong alias — the +// wrong-alias list enumerates the warehouse's ACTUAL trap answers, and a +// nonexistent variant neither names the trap nor matches a correct alias, so +// it cannot be credited either. func Entity(final string, aliases, wrongAliases []string) (matched string, correct bool) { line := strings.ToLower(firstLine(final)) for _, w := range wrongAliases { - if w != "" && strings.Contains(line, strings.ToLower(w)) { + if w != "" && containsWord(line, strings.ToLower(w)) { return "", false } } for _, a := range aliases { - if a != "" && strings.Contains(line, strings.ToLower(a)) { + if a != "" && containsWord(line, strings.ToLower(a)) { return a, true } } return "", false } + +// containsWord reports whether needle occurs in line with no word character +// (letter, digit, underscore) immediately adjacent on either side. +func containsWord(line, needle string) bool { + if needle == "" { + return false + } + for start := 0; ; start++ { + i := strings.Index(line[start:], needle) + if i < 0 { + return false + } + start += i + if boundaryBefore(line, start) && boundaryAfter(line, start+len(needle)) { + return true + } + } +} + +// boundaryBefore reports whether position i starts at a word boundary. +func boundaryBefore(s string, i int) bool { + if i == 0 { + return true + } + r, _ := utf8.DecodeLastRuneInString(s[:i]) + return !isWordRune(r) +} + +// boundaryAfter reports whether position i ends at a word boundary. +func boundaryAfter(s string, i int) bool { + if i >= len(s) { + return true + } + r, _ := utf8.DecodeRuneInString(s[i:]) + return !isWordRune(r) +} + +// isWordRune reports whether r is a word character for alias matching. +func isWordRune(r rune) bool { + return r == '_' || unicode.IsLetter(r) || unicode.IsDigit(r) +} diff --git a/bench/internal/grade/grade_test.go b/bench/internal/grade/grade_test.go index 607a8fc07..05244f678 100644 --- a/bench/internal/grade/grade_test.go +++ b/bench/internal/grade/grade_test.go @@ -81,6 +81,29 @@ func TestEntity(t *testing.T) { } } +func TestEntityWordBoundary(t *testing.T) { + aliases := []string{"West"} + wrong := []string{"North", "South", "East"} + cases := []struct { + name string + final string + correct bool + }{ + {"east inside at least does not veto", "West, at least by net revenue", true}, + {"east as a word vetoes", "West beats East on gross", false}, + {"west inside southwest does not match", "the southwest region", false}, + {"punctuation is a boundary", "West.", true}, + {"digit adjacency is not a boundary", "West2", false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if _, correct := Entity(c.final, aliases, wrong); correct != c.correct { + t.Errorf("Entity(%q) correct = %v, want %v", c.final, correct, c.correct) + } + }) + } +} + func TestEntityRegionTrap(t *testing.T) { // The top-region trap: naming a losing region anywhere on the answer line // is incorrect even when the winner is also mentioned. diff --git a/bench/internal/lifecycle/claudecli_test.go b/bench/internal/lifecycle/claudecli_test.go index 90ff434f4..b2b624453 100644 --- a/bench/internal/lifecycle/claudecli_test.go +++ b/bench/internal/lifecycle/claudecli_test.go @@ -37,7 +37,8 @@ const claudeCachedStream = `{"type":"system","subtype":"init","mcp_servers":[{"n // EpisodeRecord, so a cached run self-reports its true cost basis (cache reads // bill far below fresh input) rather than being estimated from input/output // totals. The one remaining gap — that a real `claude` process actually emits -// these fields — is confirmed by a single real cached run (see bench/README.md). +// these fields — awaits the first real claude-cli run on this parser (see +// bench/README.md); the runner warns loudly on zero usage in the meantime. func TestClaudeCLIEpisodeRecordsCacheTokens(t *testing.T) { fp := newFakePlatform(t) runner, err := claudecli.New(claudecli.Options{ @@ -152,6 +153,82 @@ func TestClaudeCLIEpisode(t *testing.T) { } } +// claudeNoHandleRecallStream drives a successful bench call but never mints a +// dps_ handle via platform_info. +const claudeNoHandleRecallStream = `{"type":"system","subtype":"init","mcp_servers":[{"name":"bench","status":"connected"}]} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"s1","name":"mcp__bench__search","input":{"query":"threshold"}}]}} +{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"s1","is_error":false,"content":"saved: threshold is 500"}]}} +{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"FINAL ANSWER: 500"}]}} +{"type":"result","subtype":"success","is_error":false,"result":"FINAL ANSWER: 500","session_id":"cc-2","usage":{"input_tokens":30,"output_tokens":8}}` + +// TestLifecycleClaudeCLIEpisodeNoHandleWithSuccesses proves a lifecycle +// claude-cli episode reporting successful tool calls but no dps_ handle is a +// harness error, not a silently zeroed audit record: the session gate refuses +// un-threaded calls, so the combination is an impossibility to surface +// (mirrors the S1-S3 pipeline and cold-start contracts). +func TestLifecycleClaudeCLIEpisodeNoHandleWithSuccesses(t *testing.T) { + fp := newFakePlatform(t) + runner, err := claudecli.New(claudecli.Options{ + Model: "claude-sonnet-5", + Exec: func(context.Context, claudecli.CommandSpec) ([]byte, []byte, error) { + return []byte(claudeNoHandleRecallStream), nil, nil + }, + }) + if err != nil { + t.Fatalf("claudecli.New: %v", err) + } + tgt := target.Target{BaseURL: fp.httpSrv.URL, Credential: fp.base} + env := &runEnv{ + opts: Options{Target: tgt, HTTPTimeout: 10 * time.Second, Arm: "a3", ClaudeCLI: runner, IdentityKeys: 32, AuditTimeout: time.Second}, + log: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelWarn})), + audit: auditapi.New(fp.httpSrv.URL, tgt.HTTPClient(10*time.Second)), + } + env.currentProtocolID = "p-nohandle" + rec, _ := env.runClaudeCLIEpisode(context.Background(), episodeSpec{stage: StageRecall, identity: "teacher", seq: 3, prompt: "q"}) + if rec.Error == "" || !strings.Contains(rec.Error, "no dps_ handle") { + t.Fatalf("episode error = %q, want a no-handle harness error", rec.Error) + } +} + +// TestLifecycleClaudeCLIEpisodeAuditReadFailureRecorded proves a failed audit +// read-back lands on the episode record instead of silently zeroing the audit +// metrics, and is totaled as a summary warning: the fake has no audit rows for +// the threaded handle, so WaitForSession times out. +func TestLifecycleClaudeCLIEpisodeAuditReadFailureRecorded(t *testing.T) { + fp := newFakePlatform(t) // no audit events for dps_cc_1 + runner, err := claudecli.New(claudecli.Options{ + Model: "claude-sonnet-5", + Exec: func(context.Context, claudecli.CommandSpec) ([]byte, []byte, error) { + return []byte(claudeRecallStream), nil, nil + }, + }) + if err != nil { + t.Fatalf("claudecli.New: %v", err) + } + tgt := target.Target{BaseURL: fp.httpSrv.URL, Credential: fp.base} + env := &runEnv{ + opts: Options{Target: tgt, HTTPTimeout: 10 * time.Second, Arm: "a3", ClaudeCLI: runner, IdentityKeys: 32, AuditTimeout: 200 * time.Millisecond}, + log: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelWarn})), + audit: auditapi.New(fp.httpSrv.URL, tgt.HTTPClient(10*time.Second)), + } + env.currentProtocolID = "p-auditloss" + rec, _ := env.runClaudeCLIEpisode(context.Background(), episodeSpec{stage: StageRecall, identity: "teacher", seq: 3, prompt: "q"}) + if rec.Error != "" { + t.Fatalf("episode error = %q, want graded episode (audit loss is not a harness failure)", rec.Error) + } + if rec.AuditReadError == "" { + t.Fatal("AuditReadError empty: a lost audit read-back must be recorded, not silently zeroed") + } + res := &Results{Runs: []ProtocolRun{{Episodes: []EpisodeRecord{rec}}}} + res.Aggregate() + if res.Metrics.AuditReadFailures != 1 { + t.Fatalf("AuditReadFailures = %d, want 1", res.Metrics.AuditReadFailures) + } + if !strings.Contains(res.HumanSummary(), "lost their audit read-back") { + t.Error("summary missing the audit-read-back warning") + } +} + // TestClaudeCLIEpisodeServerNotConnected records a harness failure when the // bench MCP server never connects (claude reached no platform tools). func TestClaudeCLIEpisodeServerNotConnected(t *testing.T) { diff --git a/bench/internal/lifecycle/episode.go b/bench/internal/lifecycle/episode.go index 578ed5f0e..e3099dfca 100644 --- a/bench/internal/lifecycle/episode.go +++ b/bench/internal/lifecycle/episode.go @@ -153,7 +153,7 @@ func (e *runEnv) runEpisode(ctx context.Context, spec episodeSpec) (EpisodeRecor rec.Error = fmt.Sprintf("agent loop: %v", runErr) return rec, final } - rec.Audit = e.readAudit(ctx, info.Handle, audited, audited+indeterminate) + rec.Audit, rec.AuditReadError = e.readAudit(ctx, info.Handle, audited, audited+indeterminate) return rec, final } @@ -201,26 +201,38 @@ func (e *runEnv) runClaudeCLIEpisode(ctx context.Context, spec episodeSpec) (Epi rec.Error = fmt.Sprintf("bench MCP server did not connect (status %q)", cres.ServerStatus) return rec, cres.FinalText } + if cres.Handle == "" { + // No handle means no parseable platform_info result. With no handle a + // successful data call is impossible (the session-gate middleware refuses + // un-threaded calls), so a positive success count is a harness + // inconsistency to surface — either the stream shape drifted or the + // platform_info result did not parse — not audit loss (mirrors the S1-S3 + // pipeline and cold-start contracts). + if cres.SuccessfulMCPCalls > 0 { + rec.Error = fmt.Sprintf("claude-cli reported %d successful tool call(s) but surfaced no dps_ handle to correlate audit (platform_info result missing or unparseable)", cres.SuccessfulMCPCalls) + } + return rec, cres.FinalText + } // Correlate by the dps_ handle claude threaded (each episode mints its own), // so consecutive same-identity stages never fold into one another. Best // effort like the loop path: S5 correctness comes from the knowledge API. - if cres.Handle != "" { - rec.Audit = e.readAudit(ctx, cres.Handle, cres.SuccessfulMCPCalls, cres.MCPCalls) - } + rec.Audit, rec.AuditReadError = e.readAudit(ctx, cres.Handle, cres.SuccessfulMCPCalls, cres.MCPCalls) return rec, cres.FinalText } // readAudit reads the session's audit trail back best effort. Unlike the S1-S3 // pipeline, a missing audit row does not fail an S5 run: the lifecycle state is // verified through the knowledge API, and audit here only enriches the -// efficiency picture. A read failure yields zero metrics and is logged. -func (e *runEnv) readAudit(ctx context.Context, handle string, minAudited, maxAudited int) auditapi.Metrics { +// efficiency picture. A read failure yields zero metrics plus the error, which +// is recorded on the episode so audit-signal loss is visible in results.json +// rather than only in a log line. +func (e *runEnv) readAudit(ctx context.Context, handle string, minAudited, maxAudited int) (auditapi.Metrics, string) { events, err := e.audit.WaitForSession(ctx, handle, minAudited, maxAudited, e.opts.AuditTimeout) if err != nil { e.log.Warn("lifecycle audit read-back", "handle", handle, "error", err) - return auditapi.Metrics{} + return auditapi.Metrics{}, err.Error() } - return auditapi.Summarize(events) + return auditapi.Summarize(events), "" } // writeClaudeTranscript persists a claude-cli episode's reconstructed transcript diff --git a/bench/internal/lifecycle/regression.go b/bench/internal/lifecycle/regression.go index 047d9e6ea..5ab9cdbae 100644 --- a/bench/internal/lifecycle/regression.go +++ b/bench/internal/lifecycle/regression.go @@ -80,6 +80,17 @@ func BaselineCompatible(candidate, baseline *Results) error { if baseline.Metrics.Attempts == 0 { return errors.New("baseline graded no attempts — nothing to gate against") } + // A baseline produced before the update-capture gating of the duplicate rate + // counted update-capture misses as duplicates, so its duplicate_rate is + // definitionally higher than the current harness reports on identical + // behavior — gating against it can absorb a real supersede regression up to + // that inflation. Detection: a current-harness run that scored any duplicate + // necessarily carries update_capture_rate data (every attempt entering the + // duplicate denominator passed the capture gate), so a duplicate denominator + // with no update-capture denominator marks a pre-redefinition artifact. + if baseline.Metrics.DuplicateRate.Den > 0 && baseline.Metrics.UpdateCaptureRate.Den == 0 { + return errors.New("baseline predates the update-capture gating of the duplicate rate (duplicate_rate has a denominator but update_capture_rate has none), so its duplicate numbers are not comparable — regenerate the baseline with the current harness") + } return nil } diff --git a/bench/internal/lifecycle/regression_test.go b/bench/internal/lifecycle/regression_test.go index 0692a7621..96e36b706 100644 --- a/bench/internal/lifecycle/regression_test.go +++ b/bench/internal/lifecycle/regression_test.go @@ -22,6 +22,9 @@ func baseMetrics() Metrics { PersonalRecall: rate(9, 10), TransferRate: rate(8, 10), UpdateCorrectness: rate(5, 5), + // A current-harness run that scores duplicates always carries + // update-capture data (the gate feeds the duplicate denominator). + UpdateCaptureRate: rate(5, 5), AbstentionRate: rate(9, 10), DuplicateRate: rate(0, 5), PassK: rate(7, 10), @@ -53,6 +56,22 @@ func TestBaselineCompatible(t *testing.T) { if err := BaselineCompatible(lifecycleResults("a3", "", Metrics{}), lifecycleResults("a3", "", Metrics{})); err == nil { t.Fatal("an empty baseline should be refused") } + // A baseline from before the update-capture gating of the duplicate rate + // (duplicate denominator present, no update-capture data) counted capture + // misses as duplicates; its duplicate numbers are definitionally inflated, + // so gating against it must be refused, not silently absorbed. + legacy := baseMetrics() + legacy.UpdateCaptureRate = Rate{} + if err := BaselineCompatible(lifecycleResults("a3", "", baseMetrics()), lifecycleResults("a3", "", legacy)); err == nil { + t.Fatal("a pre-redefinition baseline (duplicates scored, no update-capture data) should be refused") + } + // A baseline with no supersede coverage at all (both denominators zero) is + // not a legacy artifact — nothing about its duplicate numbers is stale. + noSupersede := baseMetrics() + noSupersede.DuplicateRate, noSupersede.UpdateCaptureRate = Rate{}, Rate{} + if err := BaselineCompatible(lifecycleResults("a3", "", baseMetrics()), lifecycleResults("a3", "", noSupersede)); err != nil { + t.Fatalf("a baseline without supersede coverage should be compatible: %v", err) + } } func TestCheckRegressionClean(t *testing.T) { diff --git a/bench/internal/lifecycle/report.go b/bench/internal/lifecycle/report.go index 27e871e66..2d75964a6 100644 --- a/bench/internal/lifecycle/report.go +++ b/bench/internal/lifecycle/report.go @@ -88,7 +88,10 @@ type EpisodeRecord struct { CacheReadTokens int64 `json:"cache_read_tokens,omitempty"` CacheCreationTokens int64 `json:"cache_creation_tokens,omitempty"` Audit auditapi.Metrics `json:"audit"` - Error string `json:"error,omitempty"` + // AuditReadError records a failed audit read-back on an otherwise-successful + // episode: its zero audit metrics mean signal loss, not zero activity. + AuditReadError string `json:"audit_read_error,omitempty"` + Error string `json:"error,omitempty"` } // ProtocolRun records one protocol attempt (1..k). Each stage outcome is a @@ -111,8 +114,13 @@ type ProtocolRun struct { TransferCorrect *bool `json:"transfer_correct,omitempty"` // cross-identity recall correct TransferSurfaced *bool `json:"transfer_surfaced,omitempty"` // promoted fact appeared in a tool result the learner saw UpdateCorrect *bool `json:"update_correct,omitempty"` // recall flipped to the corrected value - Duplicated *bool `json:"duplicated,omitempty"` // supersede left more than one live insight - AbstainCorrect *bool `json:"abstain_correct,omitempty"` // abstained on a never-taught fact + // UpdateCaptured reports whether the update episode actually executed a + // correction capture call. When false the platform never received the + // correction, so its supersede gate never ran: Duplicated stays nil (the + // attempt is a capture miss, not a duplicate) and the run cannot pass. + UpdateCaptured *bool `json:"update_captured,omitempty"` + Duplicated *bool `json:"duplicated,omitempty"` // supersede left more than one live insight (nil when the correction capture never executed) + AbstainCorrect *bool `json:"abstain_correct,omitempty"` // abstained on a never-taught fact // Capture-budget diagnosis (issue #964), read from the teach episode. Nil // when the teach episode never ran (harness abort before teach). @@ -140,9 +148,17 @@ func (r ProtocolRun) Passed() bool { r.updatePassed() && optPass(r.AbstainCorrect) } -// updatePassed reports whether the supersede stage passed, or was not run. It -// requires both a flipped recall and no duplicate. +// updatePassed reports whether the supersede stage passed, or was not run. A +// missed correction capture fails the stage outright (the lifecycle never +// received the correction; its recall is skipped, so UpdateCorrect is nil and +// this check must come first). Otherwise the stage requires a flipped recall +// and no duplicate; a nil UpdateCorrect then means the stage was not run. A +// nil UpdateCaptured (results from before the field existed) falls back to the +// recall-and-duplicate check alone. func (r ProtocolRun) updatePassed() bool { + if r.UpdateCaptured != nil && !*r.UpdateCaptured { + return false + } if r.UpdateCorrect == nil { return true } @@ -247,6 +263,9 @@ type Metrics struct { Protocols int `json:"protocols"` Attempts int `json:"attempts"` // graded protocol runs (harness failures excluded) HarnessFailures int `json:"harness_failures"` // runs aborted by a harness error + // AuditReadFailures counts episodes whose audit read-back failed: each + // contributes zero audit metrics through signal loss, not zero activity. + AuditReadFailures int `json:"audit_read_failures,omitempty"` // Token totals across every episode of every run (including harness-failed // runs — a failed episode still spent tokens), so a run self-reports its cost @@ -263,7 +282,12 @@ type Metrics struct { UnpromptedSurface Rate `json:"unprompted_surface"` // among captured runs, search surfaced the memory TransferRate Rate `json:"transfer_rate"` UpdateCorrectness Rate `json:"update_correctness"` - DuplicateRate Rate `json:"duplicate_rate"` // fraction of supersedes that duplicated (lower is better) + // UpdateCaptureRate is, among update stages that ran, the fraction whose + // correction capture actually executed. Its misses are excluded from + // DuplicateRate (no correction reached the platform, so the supersede gate + // never ran) and reported here instead of inflating the duplicate count. + UpdateCaptureRate Rate `json:"update_capture_rate"` + DuplicateRate Rate `json:"duplicate_rate"` // fraction of executed supersedes that duplicated (lower is better) AbstentionRate Rate `json:"abstention_rate"` // Transfer-gap decomposition (issue #964). TransferSurfaced is the fraction @@ -304,6 +328,9 @@ func (res *Results) Aggregate() { } byProtocol[r.ProtocolID] = append(byProtocol[r.ProtocolID], r) for _, e := range r.Episodes { + if e.AuditReadError != "" { + m.AuditReadFailures++ + } m.TotalInputTokens += e.InputTokens m.TotalOutputTokens += e.OutputTokens m.TotalCacheReadTokens += e.CacheReadTokens @@ -321,6 +348,7 @@ func (res *Results) Aggregate() { m.TransferSurfaced.add(r.TransferSurfaced) m.TransferUsedGivenSurfaced.addConditional(boolTrue(r.TransferSurfaced), boolTrue(r.TransferCorrect)) m.UpdateCorrectness.add(r.UpdateCorrect) + m.UpdateCaptureRate.add(r.UpdateCaptured) m.DuplicateRate.add(r.Duplicated) m.AbstentionRate.add(r.AbstainCorrect) m.CaptureBudgetStarved.addConditional(captureBudgetObservable(r), budgetStarved(r)) @@ -340,7 +368,7 @@ func (m *Metrics) fillCIs(rng *rand.Rand) { for _, r := range []*Rate{ &m.CaptureRate, &m.PersonalRecall, &m.UnpromptedSurface, &m.TransferRate, &m.TransferSurfaced, &m.TransferUsedGivenSurfaced, - &m.UpdateCorrectness, &m.DuplicateRate, &m.AbstentionRate, + &m.UpdateCorrectness, &m.UpdateCaptureRate, &m.DuplicateRate, &m.AbstentionRate, &m.CaptureBudgetStarved, &m.PassK, } { r.fillCI(rng) @@ -412,6 +440,9 @@ func (res *Results) HumanSummary() string { fmt.Fprintf(&b, " %s .. %s\n\n", m.StartedAt.Format(time.RFC3339), m.FinishedAt.Format(time.RFC3339)) mt := res.Metrics fmt.Fprintf(&b, "protocols %d attempts %d harness failures %d\n", mt.Protocols, mt.Attempts, mt.HarnessFailures) + if mt.AuditReadFailures > 0 { + fmt.Fprintf(&b, "WARNING: %d episode(s) lost their audit read-back; their audit-derived metrics are zero through signal loss, not zero activity\n", mt.AuditReadFailures) + } 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) writeMetric(&b, "capture rate", mt.CaptureRate) @@ -421,6 +452,7 @@ func (res *Results) HumanSummary() string { writeMetric(&b, " transfer surfaced", mt.TransferSurfaced) writeMetric(&b, " used given surfaced", mt.TransferUsedGivenSurfaced) writeMetric(&b, "update correctness", mt.UpdateCorrectness) + writeMetric(&b, " update capture rate", mt.UpdateCaptureRate) writeMetric(&b, "duplicate rate", mt.DuplicateRate) writeMetric(&b, "abstention rate", mt.AbstentionRate) writeMetric(&b, "capture budget-starved", mt.CaptureBudgetStarved) diff --git a/bench/internal/lifecycle/runner.go b/bench/internal/lifecycle/runner.go index 5fa9a79d1..6acec566c 100644 --- a/bench/internal/lifecycle/runner.go +++ b/bench/internal/lifecycle/runner.go @@ -56,6 +56,9 @@ type Options struct { LLMProvider string GitCommit string AuditTimeout time.Duration + // SinkTimeout bounds the reviewer's post-apply sink read-back (zero uses the + // promote package default); raise it for a store that serves reads slowly. + SinkTimeout time.Duration // IdentityKeys is the identity-pool size the arm config defines. Each // protocol attempt consumes identitiesPerRun identities, so the run refuses // to start when protocols x k x 2 exceeds the pool (attempts would share a @@ -136,7 +139,7 @@ func newRunEnv(opts Options) *runEnv { log: opts.Log, audit: auditapi.New(opts.Target.BaseURL, opts.Target.HTTPClient(opts.HTTPTimeout)), life: life, - reviewer: promote.Reviewer{Life: life, Log: opts.Log}, + reviewer: promote.Reviewer{Life: life, Log: opts.Log, SinkTimeout: opts.SinkTimeout}, } } @@ -348,13 +351,17 @@ func (e *runEnv) promoteAndTransfer(ctx context.Context, p protocol.Protocol, le return false } -// supersede runs the correction episode and its post-update recall, then checks -// via the insights API that the fact flipped and no duplicate remains. Returns -// true on a harness abort. +// supersede runs the correction episode, verifies the correction reached the +// platform, then runs the post-update recall and checks via the insights API +// that the fact flipped and no duplicate remains. Returns true on a harness +// abort. func (e *runEnv) supersede(ctx context.Context, p protocol.Protocol, teacherSeq int, run *ProtocolRun) bool { if p.Update == nil { return false } + // The update start (minus the shared skew margin) bounds the capture-verify + // fallback to THIS episode, excluding the teach-stage insight. + updStart := time.Now() upd, _ := e.runEpisode(ctx, episodeSpec{ stage: StageUpdate, identity: "teacher", seq: teacherSeq, prompt: p.Update.Prompt, system: teachSystem, budget: e.teachBudget(p), @@ -364,6 +371,25 @@ func (e *runEnv) supersede(ctx context.Context, p protocol.Protocol, teacherSeq run.Error = "update: " + upd.Error return true } + + // The duplicate check is meaningful only when the correction actually + // reached the platform: with no correction captured, exactly one live + // insight exists and the supersede gate never ran, so scoring the attempt + // as "duplicated" would inflate the duplicate rate with capture noise — the + // very noise the isolated sub-benchmark exists to remove. A miss is + // recorded on the update-capture rate (and fails the attempt), the recall + // episode is skipped (it would grade staleness, not the lifecycle, and + // dilute update correctness), and Duplicated stays nil. + captured, err := e.correctionCaptured(ctx, upd, p, teacherSeq, updStart, run.InsightID) + if err != nil { + run.Error = "update capture verify: " + err.Error() + return true + } + run.UpdateCaptured = &captured + if !captured { + return false + } + recall, ans := e.runEpisode(ctx, episodeSpec{ stage: StageUpdateRecall, identity: "teacher", seq: teacherSeq, prompt: p.Update.Recall.Prompt, system: recallSystem(p.Update.Recall.Grading.Kind), budget: p.BudgetToolCalls, @@ -384,6 +410,34 @@ func (e *runEnv) supersede(ctx context.Context, p protocol.Protocol, teacherSeq return false } +// correctionCaptured reports whether the update episode's correction reached +// the platform. The transcript's executed-capture signal is the fast path; when +// it is inconclusive — a claude-cli stream can drop the paired tool_result of a +// call that really executed, which the parser deliberately models as +// indeterminate — the insights API is the arbiter: an insight captured by the +// teacher on the protocol's entity since the update episode started proves the +// capture landed, regardless of transcript shape. The teach-stage insight is +// excluded by ID (the since bound alone cannot: the clock-skew margin it +// carries can reach back past a teach capture made moments earlier). +func (e *runEnv) correctionCaptured(ctx context.Context, upd EpisodeRecord, p protocol.Protocol, teacherSeq int, updStart time.Time, teachInsightID string) (bool, error) { + if upd.CaptureAttempted { + return true, nil + } + insights, err := e.life.ListInsights(ctx, lifecycleapi.InsightFilter{ + CapturedBy: poolEmail(teacherSeq), EntityURN: p.EntityURN, + Since: updStart.Add(-promote.CaptureSkewMargin), + }) + if err != nil { + return false, err + } + for _, in := range insights { + if in.ID != teachInsightID { + return true, nil + } + } + return false, nil +} + // duplicated reports whether the supersede left the taught fact duplicated: a // clean supersede transitions the taught insight to superseded (so the recall // path surfaces only the correction), while a failure to detect the restatement diff --git a/bench/internal/lifecycle/runner_test.go b/bench/internal/lifecycle/runner_test.go index e25279155..443ec5d4c 100644 --- a/bench/internal/lifecycle/runner_test.go +++ b/bench/internal/lifecycle/runner_test.go @@ -24,6 +24,7 @@ import ( "github.com/txn2/mcp-data-platform/bench/internal/auditapi" "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/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" @@ -46,6 +47,8 @@ type fakePlatform struct { insights []lifecycleapi.Insight changesets []lifecycleapi.Changeset events []auditapi.Event + entityDesc map[string]string // urn -> effective description (set by apply, read by the sink read-back) + pages []lifecycleapi.KnowledgePage // live knowledge pages (appended by apply, listed for the sink read-back) httpSrv *httptest.Server applyFails bool // apply_knowledge returns a tool error (measured miss) noChangesetRef bool // apply records the changeset but leaves insight.changeset_ref empty @@ -54,12 +57,13 @@ type fakePlatform struct { func newFakePlatform(t *testing.T) *fakePlatform { t.Helper() - fp := &fakePlatform{base: "testkey"} + fp := &fakePlatform{base: "testkey", entityDesc: map[string]string{}} server := mcp.NewServer(&mcp.Implementation{Name: "fake-lifecycle", Version: "1.0.0"}, nil) fp.addPlatformInfo(server) fp.addMemoryCapture(server) fp.addSearch(server) fp.addApplyKnowledge(server) + fp.addGetEntity(server) mcpHandler := mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return server }, nil) mux := http.NewServeMux() @@ -68,6 +72,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")) @@ -147,7 +152,7 @@ func (fp *fakePlatform) addMemoryCapture(server *mcp.Server) { fp.seq++ id := "in-" + strconv.FormatInt(fp.seq, 10) fp.insights = append(fp.insights, lifecycleapi.Insight{ - ID: id, CreatedAt: time.Unix(fp.seq, 0).UTC(), CapturedBy: email, + ID: id, CreatedAt: time.Now().UTC(), CapturedBy: email, Category: category, InsightText: text, Status: "pending", EntityURNs: urnSlice(urn), }) @@ -212,10 +217,52 @@ func (fp *fakePlatform) addApplyKnowledge(server *mcp.Server) { } } } + fp.applySinkLocked(urn, args) return okResult("applied " + csID), nil, nil }) } +// applySinkLocked models the platform landing an applied change in its sink, +// which the Reviewer's post-apply sink read-back verifies: a datahub change +// becomes the entity's effective description, a page payload becomes a live +// knowledge page. The caller holds fp.mu. +func (fp *fakePlatform) applySinkLocked(urn string, args map[string]any) { + if changes, ok := args["changes"].([]any); ok && len(changes) > 0 { + if change, ok := changes[0].(map[string]any); ok { + detail, _ := change["detail"].(string) + fp.entityDesc[urn] = detail + } + } + if page, ok := args["page"].(map[string]any); ok { + slug, _ := page["slug"].(string) + summary, _ := page["summary"].(string) + fp.pages = append(fp.pages, lifecycleapi.KnowledgePage{ID: "kp-" + slug, Slug: slug, Summary: summary}) + } +} + +// addGetEntity serves the promote sink read-back: an entity's effective +// description, set by a prior apply_knowledge. +func (fp *fakePlatform) addGetEntity(server *mcp.Server) { + schema := sessionSchema(map[string]*jsonschema.Schema{"urn": {Type: "string"}}) + mcp.AddTool(server, &mcp.Tool{Name: promote.EntityToolName, 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 page-sink read-back 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 for a call that carried a session handle. // The caller holds fp.mu. func (fp *fakePlatform) recordLocked(args map[string]any, tool string) { @@ -244,6 +291,16 @@ func (fp *fakePlatform) listInsights(w http.ResponseWriter, r *http.Request) { if v := q.Get("entity_urn"); v != "" && !in.LinksEntity(v) { continue } + // Honor the since bound exactly as the admin API does (created_at >=): + // the runner's capture verification depends on it to exclude an earlier + // stage's (or run's) insight, so the fake must not silently pass what + // the real platform would filter. + if v := q.Get("since"); v != "" { + since, err := time.Parse(time.RFC3339, v) + if err != nil || in.CreatedAt.Before(since) { + continue + } + } out = append(out, in) } writeJSON(w, map[string]any{"data": out, "total": len(out), "page": 1, "per_page": 100}) diff --git a/bench/internal/lifecycle/supersede_report.go b/bench/internal/lifecycle/supersede_report.go index 1ccdd7f4e..420b7c029 100644 --- a/bench/internal/lifecycle/supersede_report.go +++ b/bench/internal/lifecycle/supersede_report.go @@ -39,9 +39,15 @@ type SupersedeMetrics struct { TotalCacheCreationTokens int64 `json:"total_cache_creation_tokens"` CaptureRate Rate `json:"capture_rate"` - SupersedeRate Rate `json:"supersede_rate"` // among captured attempts, the original was superseded (higher is better) - DuplicateRate Rate `json:"duplicate_rate"` // among captured attempts, a duplicate was left (lower is better) + SupersedeRate Rate `json:"supersede_rate"` // among executed supersedes, the original was superseded (higher is better) + DuplicateRate Rate `json:"duplicate_rate"` // among executed supersedes, a duplicate was left (lower is better) UpdateCorrectness Rate `json:"update_correctness"` // post-correction recall flipped to the new value + // UpdateCaptureRate is, among update stages that ran, the fraction whose + // correction capture executed. Its misses never enter the supersede/duplicate + // denominator: with no correction on the platform the supersede gate never + // ran, so counting the attempt as duplicated would re-introduce exactly the + // capture noise this sub-benchmark isolates away. + UpdateCaptureRate Rate `json:"update_capture_rate"` PassK Rate `json:"pass_k"` // protocols cleanly superseding on all k attempts PerProtocol []SupersedeProtocolStat `json:"per_protocol"` // per-protocol stability across the k attempts @@ -58,6 +64,9 @@ type SupersedeProtocolStat struct { Captured int `json:"captured"` Superseded int `json:"superseded"` Duplicated int `json:"duplicated"` + // UpdateCaptureMissed counts captured attempts whose correction capture never + // executed — the gap between Captured and Superseded+Duplicated. + UpdateCaptureMissed int `json:"update_capture_missed,omitempty"` } // Aggregate computes the supersede metrics from the runs. @@ -86,6 +95,7 @@ func (res *SupersedeResults) Aggregate() { m.Attempts++ foldSupersedeStat(stat, r) m.CaptureRate.add(r.Captured) + m.UpdateCaptureRate.add(r.UpdateCaptured) m.DuplicateRate.add(r.Duplicated) m.UpdateCorrectness.add(r.UpdateCorrect) } @@ -114,7 +124,7 @@ func (res *SupersedeResults) Aggregate() { // zero-width default rather than reflecting into a spurious [1, 1]. func (m *SupersedeMetrics) fillCIs(rng *rand.Rand) { for _, r := range []*Rate{ - &m.CaptureRate, &m.DuplicateRate, &m.UpdateCorrectness, &m.PassK, + &m.CaptureRate, &m.UpdateCaptureRate, &m.DuplicateRate, &m.UpdateCorrectness, &m.PassK, } { r.fillCI(rng) } @@ -131,6 +141,9 @@ func foldSupersedeStat(stat *SupersedeProtocolStat, r ProtocolRun) { if boolTrue(r.Captured) { stat.Captured++ } + if r.UpdateCaptured != nil && !*r.UpdateCaptured { + stat.UpdateCaptureMissed++ + } if r.Duplicated != nil { if *r.Duplicated { stat.Duplicated++ @@ -171,8 +184,13 @@ func supersedePassKRate(order []string, runs []ProtocolRun, k int) Rate { } // supersedePassed reports whether one attempt cleanly superseded: captured the -// fact, left no duplicate, and flipped the post-correction recall. +// fact, executed the correction capture (when observed; results from before the +// field existed leave it nil), left no duplicate, and flipped the +// post-correction recall. func supersedePassed(r ProtocolRun) bool { + if r.UpdateCaptured != nil && !*r.UpdateCaptured { + return false + } return r.Error == "" && boolTrue(r.Captured) && !boolTrue(r.Duplicated) && boolTrue(r.UpdateCorrect) } @@ -218,15 +236,16 @@ func (res *SupersedeResults) HumanSummary() string { 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) writeMetric(&b, "capture rate", mt.CaptureRate) + writeMetric(&b, "update capture rate", mt.UpdateCaptureRate) writeMetric(&b, "supersede rate", mt.SupersedeRate) writeMetric(&b, "duplicate rate", mt.DuplicateRate) writeMetric(&b, "update correctness", mt.UpdateCorrectness) writeMetric(&b, "pass^k (protocols)", mt.PassK) if len(mt.PerProtocol) > 0 { - b.WriteString("\nper-protocol stability (superseded/duplicated of captured attempts):\n") + b.WriteString("\nper-protocol stability (superseded/duplicated of executed supersedes):\n") for _, s := range mt.PerProtocol { - fmt.Fprintf(&b, " %-24s cap %d/%d superseded %d duplicated %d\n", - s.ProtocolID, s.Captured, s.Attempts, s.Superseded, s.Duplicated) + fmt.Fprintf(&b, " %-24s cap %d/%d superseded %d duplicated %d update-capture missed %d\n", + s.ProtocolID, s.Captured, s.Attempts, s.Superseded, s.Duplicated, s.UpdateCaptureMissed) } } return b.String() diff --git a/bench/internal/lifecycle/supersede_test.go b/bench/internal/lifecycle/supersede_test.go index 620fe1e9c..8bfa26a3e 100644 --- a/bench/internal/lifecycle/supersede_test.go +++ b/bench/internal/lifecycle/supersede_test.go @@ -5,8 +5,11 @@ import ( "path/filepath" "strings" "testing" + "time" + "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/target" ) // TestSupersedeCIComplement guards the exact-complement relationship the @@ -130,6 +133,114 @@ func TestRunSupersedeCaptureMiss(t *testing.T) { } } +// TestRunSupersedeUpdateCaptureMiss proves an attempt whose UPDATE episode never +// executes the correction capture is excluded from the supersede/duplicate +// denominators rather than misclassified as a duplicate: the platform never +// received the correction, so its supersede gate never ran and the taught +// insight legitimately stays pending. The miss is measured on its own rate and +// still fails pass^k even when the recall answer happens to be correct. +func TestRunSupersedeUpdateCaptureMiss(t *testing.T) { + fp := newFakePlatform(t) + dir := t.TempDir() + p := updateProtocol() + writeProtocols(t, dir, p) + scripts := map[string]llm.Script{ + p.ID: { + StageTeach: {captureStep("definition"), {FinalText: "saved"}}, + StageUpdate: {{FinalText: "noted, but I did not save the correction"}}, // no capture call + // No StageUpdateRecall script: the recall must be SKIPPED on a + // capture miss (it would grade staleness and dilute update + // correctness), so the runner must never request this stage. + }, + } + res, err := RunSupersede(context.Background(), runOptions(fp, dir, scriptFactory(scripts))) + if err != nil { + t.Fatalf("run supersede: %v", err) + } + m := res.Metrics + if m.CaptureRate.Num != 1 || m.CaptureRate.Den != 1 { + t.Fatalf("capture rate = %d/%d, want 1/1 (teach captured)", m.CaptureRate.Num, m.CaptureRate.Den) + } + if m.UpdateCaptureRate.Num != 0 || m.UpdateCaptureRate.Den != 1 { + t.Fatalf("update capture rate = %d/%d, want 0/1", m.UpdateCaptureRate.Num, m.UpdateCaptureRate.Den) + } + if m.SupersedeRate.Den != 0 || m.DuplicateRate.Den != 0 { + t.Fatalf("supersede/duplicate denominators = %d/%d, want 0/0 (no executed supersede to score)", + m.SupersedeRate.Den, m.DuplicateRate.Den) + } + if m.UpdateCorrectness.Den != 0 { + t.Fatalf("update correctness denominator = %d, want 0 (recall skipped, no staleness dilution)", m.UpdateCorrectness.Den) + } + if m.PassK.Rate != 0 { + t.Fatalf("pass^k = %v, want 0 (a missed correction capture fails the lifecycle)", m.PassK.Rate) + } + if s := m.PerProtocol[0]; s.UpdateCaptureMissed != 1 || s.Superseded != 0 || s.Duplicated != 0 { + t.Fatalf("per-protocol = update-capture missed %d superseded %d dup %d, want 1/0/0", + s.UpdateCaptureMissed, s.Superseded, s.Duplicated) + } + run := res.Runs[0] + if run.Duplicated != nil { + t.Fatalf("Duplicated = %v, want nil (excluded, not scored)", *run.Duplicated) + } + if got := len(run.Episodes); got != 2 { + t.Fatalf("episodes = %d, want 2 (teach + update; the recall episode must not be spent on a capture miss)", got) + } +} + +// TestCorrectionCapturedAPIArbiter pins the platform-truth fallback: when the +// transcript carries no executed-capture signal (a claude-cli stream can drop +// the paired tool_result of a call that really ran), the insights API decides — +// and the teach-stage insight, which the skew-widened since window can reach +// back to, is excluded by ID rather than trusted by time. +func TestCorrectionCapturedAPIArbiter(t *testing.T) { + fp := newFakePlatform(t) + env := &runEnv{ + opts: Options{Target: target.Target{BaseURL: fp.httpSrv.URL, Credential: "testkey"}, HTTPTimeout: 10 * time.Second}, + life: lifecycleapi.New(fp.httpSrv.URL, fp.httpSrv.Client()), + } + p := updateProtocol() + teacherSeq := 1 + upd := EpisodeRecord{CaptureAttempted: false} + updStart := time.Now() + + got, err := env.correctionCaptured(context.Background(), upd, p, teacherSeq, updStart, "in-teach") + if err != nil || got { + t.Fatalf("no insights: captured = %v err = %v, want false", got, err) + } + + // The teach insight alone — inside the skew window but excluded by ID — + // must not count as the correction. + fp.mu.Lock() + fp.insights = append(fp.insights, lifecycleapi.Insight{ + ID: "in-teach", CreatedAt: time.Now().UTC(), CapturedBy: poolEmail(teacherSeq), + Status: "pending", EntityURNs: []string{p.EntityURN}, + }) + fp.mu.Unlock() + got, err = env.correctionCaptured(context.Background(), upd, p, teacherSeq, updStart, "in-teach") + if err != nil || got { + t.Fatalf("teach insight only: captured = %v err = %v, want false (excluded by ID)", got, err) + } + + // A distinct fresh insight proves the correction landed even though the + // transcript never showed it. + fp.mu.Lock() + fp.insights = append(fp.insights, lifecycleapi.Insight{ + ID: "in-corr", CreatedAt: time.Now().UTC(), CapturedBy: poolEmail(teacherSeq), + Status: "pending", EntityURNs: []string{p.EntityURN}, + }) + fp.mu.Unlock() + got, err = env.correctionCaptured(context.Background(), upd, p, teacherSeq, updStart, "in-teach") + if err != nil || !got { + t.Fatalf("fresh correction insight: captured = %v err = %v, want true", got, err) + } + + // The transcript fast path never consults the API. + got, err = env.correctionCaptured(context.Background(), EpisodeRecord{CaptureAttempted: true}, p, teacherSeq, updStart, "in-teach") + if err != nil || !got { + t.Fatalf("transcript-attempted: captured = %v err = %v, want true", got, err) + } +} + func TestRunSupersedeRejectsNonSupersedeProtocols(t *testing.T) { fp := newFakePlatform(t) dir := t.TempDir() diff --git a/bench/internal/lifecycleapi/lifecycleapi.go b/bench/internal/lifecycleapi/lifecycleapi.go index dc69c5cb3..b0b379f6a 100644 --- a/bench/internal/lifecycleapi/lifecycleapi.go +++ b/bench/internal/lifecycleapi/lifecycleapi.go @@ -168,8 +168,13 @@ 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"` + ID string `json:"id"` + Slug string `json:"slug"` + Title string `json:"title,omitempty"` + // Summary is what search renders next to the title — on tool surfaces + // without a page-body fetch it is the only channel a promoted fact reaches + // an agent through, so the promote sink read-back asserts on it. + Summary string `json:"summary,omitempty"` } // knowledgePageEnvelope is the portal list response for knowledge pages. diff --git a/bench/internal/promote/promote.go b/bench/internal/promote/promote.go index c27b31037..6c72bad98 100644 --- a/bench/internal/promote/promote.go +++ b/bench/internal/promote/promote.go @@ -11,8 +11,10 @@ package promote import ( "context" + "encoding/json" "fmt" "log/slog" + "strings" "time" "github.com/modelcontextprotocol/go-sdk/mcp" @@ -25,6 +27,16 @@ import ( // ApplyToolName is the platform tool that applies an approved insight to a sink. const ApplyToolName = "apply_knowledge" +// EntityToolName is the platform tool the sink read-back reads a promoted +// entity description through — the same effective-metadata path evaluators see. +const EntityToolName = "datahub_get_entity" + +// Sink read-back pacing defaults (see Reviewer.SinkTimeout/SinkPoll). +const ( + defaultSinkTimeout = 15 * time.Second + defaultSinkPoll = 500 * time.Millisecond +) + // 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 @@ -107,10 +119,16 @@ func newestInsight(insights []lifecycleapi.Insight) *lifecycleapi.Insight { } // Reviewer plays the platform reviewer: it approves an insight and applies it, -// then verifies the result through the knowledge API. +// then verifies the result through the knowledge API and reads the promoted +// content back from its sink. type Reviewer struct { Life *lifecycleapi.Client Log *slog.Logger + // SinkTimeout bounds the post-apply sink read-back (zero = 15s): how long + // the reviewer polls for the promoted content to become readable before + // declaring the write lost. SinkPoll is the poll interval (zero = 500ms). + SinkTimeout time.Duration + SinkPoll time.Duration } // Apply approves the insight (pending -> approved) and applies it via @@ -140,7 +158,20 @@ func (r Reviewer) Apply(ctx context.Context, session *mcp.ClientSession, handle } return false, nil } - return r.verify(ctx, t, insightID) + ok, err := r.verify(ctx, t, insightID) + if err != nil || !ok { + return ok, err + } + // The API verify above proves the platform recorded the promotion; the sink + // read-back proves the promoted content is actually readable where agents + // read it. An API-confirmed apply whose sink does not show the change is a + // silent write loss (the platform's own records claim success), which would + // otherwise surface only as an unexplained flat metric downstream — so it is + // a harness error, never a measured miss. + if err := r.verifySink(ctx, session, handle, t); err != nil { + return false, fmt.Errorf("sink read-back after apply (%s, insight %s): %w — NOTE: the promotion may be live despite this failure (approve and apply already succeeded), so platform state no longer matches the recorded outcome; treat the run as contaminated (a cold-start rerun needs a fresh baseline reset), and if the store is merely slow to serve reads, raise -sink-timeout", t.Label, insightID, err) + } + return true, nil } // verify confirms the insight is applied and a non-rolled-back changeset lists @@ -176,6 +207,97 @@ func (r Reviewer) verify(ctx context.Context, t Target, insightID string) (bool, return false, nil } +// verifySink polls the target's sink until the promoted content is readable or +// the window closes. A short window absorbs store-side write propagation; the +// scripted smokes run this against the live platform, so a real delivery +// regression fails there before any paid run. +func (r Reviewer) verifySink(ctx context.Context, session *mcp.ClientSession, handle string, t Target) error { + timeout, poll := r.SinkTimeout, r.SinkPoll + if timeout <= 0 { + timeout = defaultSinkTimeout + } + if poll <= 0 { + poll = defaultSinkPoll + } + deadline := time.Now().Add(timeout) + for { + ok, state, err := r.sinkHolds(ctx, session, handle, t) + if ok { + return nil + } + if time.Now().After(deadline) { + if err != nil { + return err + } + return fmt.Errorf("the applied change is not readable in its sink: %s", state) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(poll): + } + } +} + +// sinkHolds reports whether the sink currently shows the promoted content, +// with a state description for the timeout error. A read error is returned for +// the deadline path but retried until then (a transient API blip must not fail +// a verified promotion). +func (r Reviewer) sinkHolds(ctx context.Context, session *mcp.ClientSession, handle string, t Target) (bool, string, error) { + if t.Sink == protocol.SinkKnowledgePage { + return r.pageHolds(ctx, t) + } + return entityHolds(ctx, session, handle, t) +} + +// sinkEntity is the subset of the datahub_get_entity result the sink read-back +// inspects. +type sinkEntity struct { + Description string `json:"description"` +} + +// entityHolds reads the promoted entity through the platform and reports +// whether its effective description carries the applied fact. +func entityHolds(ctx context.Context, session *mcp.ClientSession, handle string, t Target) (bool, string, error) { + res := mcpc.Call(ctx, session, EntityToolName, map[string]any{"urn": t.EntityURN}, handle) + if res.TransportErr != nil { + return false, "", fmt.Errorf("%s(%s): %w", EntityToolName, t.EntityURN, res.TransportErr) + } + if res.ToolErr { + return false, "", fmt.Errorf("%s(%s) failed: %.300s", EntityToolName, t.EntityURN, res.Text) + } + var entity sinkEntity + // Enrichment middleware may append context after the entity JSON, so the + // decoder reads one value and ignores the rest. + if err := json.NewDecoder(strings.NewReader(res.Text)).Decode(&entity); err != nil { + return false, "", fmt.Errorf("parse %s result: %w (text: %.200s)", EntityToolName, err, res.Text) + } + if !strings.Contains(entity.Description, t.Fact) { + return false, fmt.Sprintf("entity %s description does not carry the applied fact (description: %.200q)", t.EntityURN, entity.Description), nil + } + return true, "", nil +} + +// pageHolds reports whether the promoted knowledge page exists with the applied +// summary — the summary is what search renders, so it is the content that must +// have landed for the fact to be deliverable. +func (r Reviewer) pageHolds(ctx context.Context, t Target) (bool, string, error) { + pages, err := r.Life.ListKnowledgePages(ctx) + if err != nil { + return false, "", fmt.Errorf("list knowledge pages: %w", err) + } + for _, p := range pages { + if p.Slug != t.Page.Slug { + continue + } + if strings.TrimSpace(p.Summary) == strings.TrimSpace(t.Page.Summary) { + return true, "", nil + } + return false, fmt.Sprintf("page %q exists but its summary does not match the applied summary (got %.200q)", t.Page.Slug, p.Summary), nil + } + return false, fmt.Sprintf("no knowledge page with slug %q", t.Page.Slug), nil +} + // BuildApplyArgs builds the apply_knowledge arguments for the target's sink. func BuildApplyArgs(t Target, insightID string) map[string]any { args := map[string]any{ diff --git a/bench/internal/promote/promote_test.go b/bench/internal/promote/promote_test.go index 184a2eb48..c508ecf71 100644 --- a/bench/internal/promote/promote_test.go +++ b/bench/internal/promote/promote_test.go @@ -221,12 +221,18 @@ type applyFake struct { // contract's structured code (a pre-audit refusal, e.g. session_expired). applyError bool applyRefusalCode string - srv *httptest.Server + // sinkLoses models the silent-write-loss defect the sink read-back exists to + // catch: apply_knowledge succeeds and the API records the changeset, but the + // sink (entity description / knowledge page) never shows the change. + sinkLoses bool + entityDesc map[string]string + pages []lifecycleapi.KnowledgePage + srv *httptest.Server } func newApplyFake(t *testing.T, applyError bool) *applyFake { t.Helper() - f := &applyFake{approved: map[string]bool{}, applyError: applyError} + f := &applyFake{approved: map[string]bool{}, applyError: applyError, entityDesc: map[string]string{}} server := mcp.NewServer(&mcp.Implementation{Name: "apply-fake", Version: "1.0.0"}, nil) mcp.AddTool(server, &mcp.Tool{Name: "platform_info", Description: "orientation"}, func(context.Context, *mcp.CallToolRequest, map[string]any) (*mcp.CallToolResult, any, error) { @@ -238,7 +244,7 @@ func newApplyFake(t *testing.T, applyError bool) *applyFake { "session_id": {Type: "string"}, "action": {Type: "string"}, "entity_urn": {Type: "string"}, }, 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) { + func(_ context.Context, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) { if f.applyRefusalCode != "" { return &mcp.CallToolResult{ IsError: true, @@ -249,10 +255,31 @@ func newApplyFake(t *testing.T, applyError bool) *applyFake { if f.applyError { return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "write refused"}}}, nil, nil } + if !f.sinkLoses { + f.applySink(args) + } return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: "applied"}}}, nil, nil }) + entitySchema := &jsonschema.Schema{Type: "object", Properties: map[string]*jsonschema.Schema{ + "session_id": {Type: "string"}, "urn": {Type: "string"}, + }, Required: []string{"session_id"}} + mcp.AddTool(server, &mcp.Tool{Name: EntityToolName, Description: "entity metadata", InputSchema: entitySchema}, + func(_ context.Context, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) { + urn, _ := args["urn"].(string) + f.mu.Lock() + desc := f.entityDesc[urn] + f.mu.Unlock() + raw, _ := json.Marshal(map[string]any{"urn": urn, "description": desc}) + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: string(raw)}}}, nil, nil + }) mcpHandler := mcp.NewStreamableHTTPHandler(func(*http.Request) *mcp.Server { return server }, nil) mux := http.NewServeMux() + mux.HandleFunc("GET /api/v1/portal/knowledge-pages", func(w http.ResponseWriter, _ *http.Request) { + f.mu.Lock() + pages := append([]lifecycleapi.KnowledgePage{}, f.pages...) + f.mu.Unlock() + _ = json.NewEncoder(w).Encode(map[string]any{"pages": pages, "total": len(pages)}) + }) mux.HandleFunc("PUT /api/v1/admin/knowledge/insights/{id}/status", func(w http.ResponseWriter, r *http.Request) { f.mu.Lock() f.approved[r.PathValue("id")] = true @@ -279,6 +306,26 @@ func newApplyFake(t *testing.T, applyError bool) *applyFake { return f } +// applySink models the platform landing the change in its sink: a datahub +// change becomes the entity's description, a page payload becomes a live +// knowledge page. +func (f *applyFake) applySink(args map[string]any) { + f.mu.Lock() + defer f.mu.Unlock() + urn, _ := args["entity_urn"].(string) + if changes, ok := args["changes"].([]any); ok && len(changes) > 0 { + if change, ok := changes[0].(map[string]any); ok { + detail, _ := change["detail"].(string) + f.entityDesc[urn] = detail + } + } + if page, ok := args["page"].(map[string]any); ok { + slug, _ := page["slug"].(string) + summary, _ := page["summary"].(string) + f.pages = append(f.pages, lifecycleapi.KnowledgePage{ID: "kp-" + slug, Slug: slug, Summary: summary}) + } +} + func TestReviewerApply(t *testing.T) { cases := []struct { name string @@ -316,6 +363,67 @@ func TestReviewerApply(t *testing.T) { } } +// TestReviewerApplySinkReadBack proves Apply reads the promoted content back +// from the actual sink, not only from the platform's own promotion records: an +// apply the insights/changesets API confirms but whose sink write was silently +// lost (the resulting_state defect class) must be a harness error, never a +// reported promotion — otherwise a run publishes a flat curve with the failure +// misattributed to agent behavior. +func TestReviewerApplySinkReadBack(t *testing.T) { + pageTarget := Target{ + Label: "cs-y", EntityURN: "urn:y", Sink: protocol.SinkKnowledgePage, + Page: &protocol.PagePayload{Slug: "rev-policy", Title: "T", Summary: "revenue = amount - discount", Body: "B"}, + } + cases := []struct { + name string + target Target + sinkLoses bool + want bool + wantErr bool + }{ + {name: "datahub sink holds the fact", target: Target{Label: "cs-x", EntityURN: "urn:x", Sink: protocol.SinkDataHub, Fact: "amounts are cents"}, want: true}, + {name: "datahub sink lost the write", target: Target{Label: "cs-x", EntityURN: "urn:x", Sink: protocol.SinkDataHub, Fact: "amounts are cents"}, sinkLoses: true, wantErr: true}, + {name: "page sink holds the summary", target: pageTarget, want: true}, + {name: "page sink lost the write", target: pageTarget, sinkLoses: true, wantErr: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newApplyFake(t, false) + f.sinkLoses = tc.sinkLoses + 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()), + SinkTimeout: 100 * time.Millisecond, SinkPoll: 10 * time.Millisecond, + } + got, err := r.Apply(ctx, session, info.Handle, tc.target, "ins-1") + if tc.wantErr { + if err == nil || !strings.Contains(err.Error(), "sink read-back") { + t.Fatalf("expected a sink read-back harness error, got promoted=%v err=%v", got, err) + } + if got { + t.Error("a lost sink write must not report promoted") + } + return + } + if err != nil { + t.Fatalf("apply: %v", err) + } + if got != tc.want { + t.Errorf("Apply = %v, want %v", got, tc.want) + } + }) + } +} + // 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 diff --git a/docs/reference/benchmarks.md b/docs/reference/benchmarks.md index b022b556c..eda14dd09 100644 --- a/docs/reference/benchmarks.md +++ b/docs/reference/benchmarks.md @@ -461,9 +461,15 @@ avoid the DataHub Kafka :9092 clash, and with `ollama serve` + ```bash make bench-up BENCH_ARM=a3 BENCH_METRICS_ADDR=:9095 make bench-lifecycle LLM=anthropic MODEL=claude-sonnet-5 K=3 -make bench-lifecycle-report +make bench-lifecycle-report RESULTS=build/bench-results/lifecycle-a3-/lifecycle-a3.json ``` +Every `bench-lifecycle` (and `bench-supersede`) invocation writes into its own +timestamped run directory under `build/bench-results/` and refuses to overwrite +an existing results file, so no run's data is ever lost; the report targets +therefore take the run to summarize via `RESULTS=` (they list the available run +directories when it is unset). + Token spend for the shared-store S5 run, from the committed records: **2,578 fresh input, 356,015 output, 23,913,219 cache-read, and 1,502,311 cache-write tokens** (≈ 25.8M total) across the 45 attempts, bounded during the run by a