From 914fe8f5d0670e07e66e15e6fb4acb418baba4a8 Mon Sep 17 00:00:00 2001 From: lr90 Date: Tue, 20 Oct 2026 16:15:48 +0800 Subject: [PATCH] feat(benchmark): add opt-in local content diagnostics --- .github/workflows/aml-benchmark.yml | 10 +- agent_memory_leaderboard/.gitignore | 1 + agent_memory_leaderboard/README.md | 7 + .../benchmarks/EVALUATION.md | 124 ++++ .../examples/content-diagnostics/input.json | 34 + .../examples/content-diagnostics/source.json | 16 + .../scripts/content_diagnostics_demo.py | 145 ++++ agent_memory_leaderboard/src/benchmark.py | 70 +- .../src/content_diagnostics.py | 638 ++++++++++++++++++ agent_memory_leaderboard/src/evaluation.py | 79 ++- .../tests/test_content_diagnostics.py | 474 +++++++++++++ 11 files changed, 1578 insertions(+), 20 deletions(-) create mode 100644 agent_memory_leaderboard/examples/content-diagnostics/input.json create mode 100644 agent_memory_leaderboard/examples/content-diagnostics/source.json create mode 100644 agent_memory_leaderboard/scripts/content_diagnostics_demo.py create mode 100644 agent_memory_leaderboard/src/content_diagnostics.py create mode 100644 agent_memory_leaderboard/tests/test_content_diagnostics.py diff --git a/.github/workflows/aml-benchmark.yml b/.github/workflows/aml-benchmark.yml index fd5879c..2c201d2 100644 --- a/.github/workflows/aml-benchmark.yml +++ b/.github/workflows/aml-benchmark.yml @@ -94,6 +94,14 @@ jobs: uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 with: name: aml-benchmark-${{ github.run_id }}-${{ github.run_attempt }} - path: agent_memory_leaderboard/var/ci-benchmark/results/ + # Explicit metadata-only allowlist. Private content/cache files are never artifacts. + path: | + agent_memory_leaderboard/var/ci-benchmark/results/run.json + agent_memory_leaderboard/var/ci-benchmark/results/run.requests.jsonl + agent_memory_leaderboard/var/ci-benchmark/results/run.questions/*.json + agent_memory_leaderboard/var/ci-benchmark/results/run.evidence/*.json + agent_memory_leaderboard/var/ci-benchmark/results/run.sources/*.json + agent_memory_leaderboard/var/ci-benchmark/results/selection.json + agent_memory_leaderboard/var/ci-benchmark/results/summary.md if-no-files-found: warn retention-days: 30 diff --git a/agent_memory_leaderboard/.gitignore b/agent_memory_leaderboard/.gitignore index 7b6b2af..e2afded 100644 --- a/agent_memory_leaderboard/.gitignore +++ b/agent_memory_leaderboard/.gitignore @@ -1,5 +1,6 @@ .venv/ .env +.private/ __pycache__/ .pytest_cache/ .ruff_cache/ diff --git a/agent_memory_leaderboard/README.md b/agent_memory_leaderboard/README.md index 34302e7..b8d549a 100644 --- a/agent_memory_leaderboard/README.md +++ b/agent_memory_leaderboard/README.md @@ -602,3 +602,10 @@ image containing these settings. Unknown environment variables are silently ignored by older adapter images. These budgets are candidates to validate with 64-client mixed and multi-message loads; no deployment or throughput guarantee is implied by changing them. + +### Opt-in self-evaluation content reports + +The self-run benchmark client can generate bounded, private local per-question +Add/Search/Answer/Eval reports for independently sourced synthetic or permitted +public data. This is disabled by default and adds no official-request body logs. +See [the offline demonstration, source declaration and limits](benchmarks/EVALUATION.md#opt-in-local-content-diagnostics). diff --git a/agent_memory_leaderboard/benchmarks/EVALUATION.md b/agent_memory_leaderboard/benchmarks/EVALUATION.md index ee7ddab..656a3c2 100644 --- a/agent_memory_leaderboard/benchmarks/EVALUATION.md +++ b/agent_memory_leaderboard/benchmarks/EVALUATION.md @@ -25,6 +25,130 @@ were not found in the documentation checked on 2026-10-07. The prompts in See the [official documentation](https://agentmemories.ai/zh-cn/docs) and [competition readiness plan](../docs/competition-plan.md#正式参评前的工作). +## Opt-in local content diagnostics + +Content diagnostics run **in the self-run benchmark client**, on the machine +running that client. They do not add request-body logging to the deployed adapter, +Memoria, or its journal. Official Smoke/Full calls do not pass through this client +and must not be used as the source of diagnostic content. Normal benchmark runs +leave the new feature disabled. + +From `agent_memory_leaderboard/`, this one command produces a local demonstration +without contacting an adapter, model provider, or downloading tokenizer data: + +```sh +uv run --locked --extra evaluation python -m scripts.content_diagnostics_demo \ + --output-dir var/content-demo +``` + +Open the `local_report` HTML path printed at the end. This demonstration uses +the independently authored fixture in `examples/content-diagnostics/`, simulated +adapter/model responses, and a byte-counting test tokenizer. Its results verify +the tool, **not Memoria quality or an AML score**. Choose another output directory +for each run. Optional `--scenario missing-evidence`, `budget-exclusion`, +`answer-failure` and `judge-failure` exercise different diagnostic signals; +the latter two intentionally return benchmark exit code 1 with unscored questions. + +To evaluate that synthetic fixture against an actual test adapter: + +```sh +# Set ADAPTER_API_KEY in the process environment if required by that adapter. +uv run --locked python -m src.benchmark --kind locomo \ + --dataset examples/content-diagnostics/input.json \ + --adapter-url http://127.0.0.1:8000 --limit 1 --top-k 100 \ + --output var/content-real/run.json --content-diagnostics \ + --diagnostic-source examples/content-diagnostics/source.json +``` + +This is `retrieval` mode: no Answer/Judge calls are made, and the pages explicitly +say those stages did not run. To include local Answer/Eval, configure the existing +`BENCHMARK_ANSWER_*` and `BENCHMARK_JUDGE_*` settings and add `--mode answer-eval`. +The existing model-service authorization and dataset transmission rules apply. + +The source declaration is required before any diagnostic run. Its format is +demonstrated by `examples/content-diagnostics/source.json`: + +- `source_type` must be `synthetic` or `public`; official and unknown sources are + rejected. Official interface/database/log-derived copies remain prohibited even + if similar text is publicly downloadable elsewhere. +- Record the independently obtained source URI/version, license or self-authored + local-use basis, allowed local evaluation/content-diagnostic uses, and retention + of 1–30 days. `dataset_sha256` must match the exact input file bytes. +- Explicitly attest `independently_obtained`, `no_official_material`, and + `no_sensitive_content`. These declarations bind the reviewed input; **they do + not prove a license or detect misdeclared official/private data**. The operator + must independently verify permission, privacy and model-transmission conditions. +- Do not put credentials, private endpoints, configuration, or personal details + in the declaration. Unknown declaration fields are not copied to the report. + +The new directory is `//content-diagnostics/`; by default +`private-dir` is `.private` beside the ordinary output JSON. It contains an +`index.html`, per-question pages, shared Add pages, a versioned `manifest.json`, +and atomic gzip JSON records. Add history is recorded once per batch and linked +from questions. Each question shows its actual query, ordered validated returned +evidence, selected/excluded ranks, actual local model messages/outputs when run, +existing metrics, errors, and retries. Prepared-but-unacknowledged Add records +are explicitly not proof that a request was sent. Successful request/response +snapshots come from the execution path, not later database reconstruction. + +No invented judge explanation or hidden model reasoning is requested. When the +judge provides only `correct`, the page states that no explanation was supplied. +Source mapping ambiguity, missing annotated sources, budget exclusion and model +failure are diagnostic signals, not automatic root-cause findings. + +The defaults capture the first 100 questions and allow at most 64 MiB **for the +new diagnostic directory**. Set `--diagnostic-max-questions` (1–1000) and +`--diagnostic-max-mib` (2–512). Size accounting includes gzip records and HTML; +individual uncompressed snapshots are bounded too. Reaching a limit omits content +and explicitly marks incompleteness; it does not skip benchmark questions or +silently improve scoring coverage. Existing answer caches, normal benchmark +outputs, the input dataset and backend storage are outside this directory's cap. +The readable pages are derived views and repeat some question/model text; this +is bounded, not a claim of zero duplication. +Content capture adds client disk work; use the payload-free load harness for +capacity measurements. Search latency is sampled before diagnostic content I/O. + +Files use owner-only permissions. Known credential environment values, common +token/auth patterns and endpoint URLs are masked in the new diagnostic display; +HTML is escaped and has no scripts or external dependencies. Raw execution hashes +and source/recall measurement remain separate from the redacted display. This is +defense in depth, **not a general personal-data detector**: sensitive input is not +eligible for this feature. Existing private answer caches retain their existing +behavior; do not share them or treat them as sanitized diagnostic files. + +The deadline starts with the diagnostic directory's creation and is not extended +by resume. Expired captures cannot resume or be compared. Review `expires_at` and +remove the dedicated local directory by that deadline; this does not schedule an +automatic cleanup job or delete backend memories/the journal. Keep declaration, +limits, dataset and evaluation configuration unchanged for `--resume`. Resuming +pending model work reuses the previous Search evidence and answer cache; a missing +snapshot is disclosed rather than rebuilt from current database contents. +A missing manifest requires a new run instead of restarting the retention clock. +Pending model resume checks cached evidence against the captured content hashes; +signature-verified answer-cache reuse is explicitly identified in the page. + +For comparable runs, create a new local paired report: + +```sh +uv run --locked python -m src.content_diagnostics \ + var/before/.private//content-diagnostics \ + var/after/.private//content-diagnostics \ + --output var/comparison.html +``` + +The HTML and adjacent JSON preserve per-question evidence-rank hashes, metrics, +missing/unscored questions and full-run coverage. Dataset bytes, selected +histories, Top K, mode and evaluator/model contracts must match. Code/backend +changes are allowed; `runner_source_sha256` records the runner's actual source +files, and the ordinary run's runtime label should identify the tested backend. +Local-model comparisons are not equivalent to the platform's Answer/Eval. + +Content files are not uploaded to GitHub or CI artifacts. The benchmark workflow +uses an explicit metadata-only artifact allowlist. Use ignored `var/` or a local +directory outside the repository; custom `--private-dir` paths remain the +operator's responsibility. Retain/share independently obtained public content +only as permitted by its license and privacy requirements. + ## Label-free ingestion and evidence attribution Only the original message text, speaker, role and source date enter Add. No diff --git a/agent_memory_leaderboard/examples/content-diagnostics/input.json b/agent_memory_leaderboard/examples/content-diagnostics/input.json new file mode 100644 index 0000000..cf99b90 --- /dev/null +++ b/agent_memory_leaderboard/examples/content-diagnostics/input.json @@ -0,0 +1,34 @@ +[ + { + "sample_id": "synthetic-atlas", + "conversation": { + "session_1_date_time": "2026-01-01", + "session_1": [ + { + "dia_id": "D1:1", + "speaker": "A", + "text": "The Atlas prototype uses Rust." + } + ], + "session_2_date_time": "2026-02-01", + "session_2": [ + { + "dia_id": "D2:1", + "speaker": "A", + "text": "The Atlas production service now uses Go." + } + ] + }, + "qa": [ + { + "id": "atlas-current-language", + "question": "Which language does the Atlas production service use?", + "answer": "Go", + "category": 1, + "evidence": [ + "D2:1" + ] + } + ] + } +] diff --git a/agent_memory_leaderboard/examples/content-diagnostics/source.json b/agent_memory_leaderboard/examples/content-diagnostics/source.json new file mode 100644 index 0000000..b0d02ab --- /dev/null +++ b/agent_memory_leaderboard/examples/content-diagnostics/source.json @@ -0,0 +1,16 @@ +{ + "version": 1, + "source_type": "synthetic", + "source_uri": "synthetic:offline-tool-demo", + "source_version": "1", + "license": "LicenseRef-SelfAuthored-LocalEvaluation", + "dataset_sha256": "c51ba347c2ffd657d3583a4257a589abd6dcc992528eee4e1b3ec51386ba91cd", + "permitted_uses": [ + "local_evaluation", + "local_content_diagnostics" + ], + "retention_days": 7, + "independently_obtained": true, + "no_official_material": true, + "no_sensitive_content": true +} diff --git a/agent_memory_leaderboard/scripts/content_diagnostics_demo.py b/agent_memory_leaderboard/scripts/content_diagnostics_demo.py new file mode 100644 index 0000000..61cda67 --- /dev/null +++ b/agent_memory_leaderboard/scripts/content_diagnostics_demo.py @@ -0,0 +1,145 @@ +"""Offline tool demonstration with synthetic data and simulated adapter/model responses.""" + +import argparse +import json +import os +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import httpx + +from src import benchmark +from src.checkpoint import save +from src.content_diagnostics import ContentDiagnostics +from src.evaluation import AnswerEvaluator, EvaluationConfig +from src.ingestion import message_parts +from src.models import Message + +ROOT = Path(__file__).resolve().parents[1] +SCENARIOS = ("complete", "missing-evidence", "budget-exclusion", "answer-failure", "judge-failure") + + +class ByteEncoder: + def encode(self, text, **kwargs): + return text.encode("utf-8") + + +def run(output_dir, scenario): + import tiktoken + + fixture = ROOT / "examples/content-diagnostics" + args = SimpleNamespace( + kind="locomo", + dataset=fixture / "input.json", + conversations=1, + top_k=100, + adapter_url="http://adapter.invalid", + output=output_dir / "run.json", + resume=False, + mode="answer-eval", + private_dir=output_dir / ".private", + content_diagnostics=True, + diagnostic_source=fixture / "source.json", + diagnostic_max_questions=10, + diagnostic_max_mib=8, + ) + settings = { + "ADAPTER_API_KEY": "offline-synthetic-adapter-credential", + "BENCHMARK_ANSWER_MODEL": "offline-simulated-answer", + "BENCHMARK_JUDGE_MODEL": "offline-simulated-judge", + "BENCHMARK_ANSWER_BASE_URL": "http://answer.invalid/v1", + "BENCHMARK_JUDGE_BASE_URL": "http://judge.invalid/v1", + "BENCHMARK_ANSWER_API_KEY": "offline-synthetic-answer-credential", + "BENCHMARK_JUDGE_API_KEY": "offline-synthetic-judge-credential", + "BENCHMARK_INPUT_BUDGET": "30000", + "BENCHMARK_OUTPUT_TOKENS": "1024", + "BENCHMARK_TOKEN_ENCODING": "offline-byte-counter", + } + memories = [] + client = httpx.Client + + def respond(request): + body = json.loads(request.content) if request.content else {} + if request.url.host == "adapter.invalid": + if request.url.path == "/add": + for message in body["messages"]: + for content, _ in message_parts(Message.model_validate(message)): + memories.append({"id": f"synthetic-{len(memories)}", "content": content}) + return httpx.Response( + 200, + json={ + "success": True, + **{k: body[k] for k in ("request_id", "user_id", "session_id")}, + }, + ) + if request.url.path == "/search": + return httpx.Response( + 200, json={"data": memories[:1] if scenario == "missing-evidence" else memories} + ) + return httpx.Response(200) + if request.url.host == "answer.invalid": + result = ( + " " + if scenario == "answer-failure" + else "Rust" + if scenario in {"missing-evidence", "budget-exclusion"} + else "Go" + ) + else: + result = ( + "invalid synthetic verdict" + if scenario == "judge-failure" + else json.dumps({"correct": scenario == "complete"}) + ) + return httpx.Response( + 200, json={"choices": [{"message": {"content": result}, "finish_reason": "stop"}]} + ) + + with ( + patch.dict(os.environ, settings), + patch.object(tiktoken, "get_encoding", return_value=ByteEncoder()), + patch.object( + httpx, "Client", lambda **kw: client(transport=httpx.MockTransport(respond), **kw) + ), + ): + if scenario == "budget-exclusion": + sample = json.loads(args.dataset.read_text())[0] + message = next(benchmark.batches(sample["conversation"]))[2][0] + content = next(message_parts(Message.model_validate(message)))[0] + evaluator = AnswerEvaluator(EvaluationConfig.from_env(), "locomo") + _, _, tokens = evaluator.context(sample["qa"][0], [content]) + os.environ["BENCHMARK_INPUT_BUDGET"] = str(tokens) + code = benchmark.run(args) + report = json.loads(args.output.read_text()) + report["simulation"] = True + save(args.output, report) + root = args.private_dir / report["run_id"] / "content-diagnostics" + state = json.loads((root / "manifest.json").read_text()) + state["simulation"] = True + save(root / "manifest.json", state) + diagnostics = ContentDiagnostics(root, state["policy"], report["contract"], selected=[]) + diagnostics.render(report["status"], report["summary"]) + return code, root + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--scenario", choices=SCENARIOS, default="complete") + args = parser.parse_args() + code, root = run(args.output_dir, args.scenario) + print( + json.dumps( + { + "simulation": True, + "benchmark_exit_code": code, + "local_report": str(root / "index.html"), + } + ) + ) + raise SystemExit(code) + + +if __name__ == "__main__": + main() diff --git a/agent_memory_leaderboard/src/benchmark.py b/agent_memory_leaderboard/src/benchmark.py index 09857a7..f015f2f 100644 --- a/agent_memory_leaderboard/src/benchmark.py +++ b/agent_memory_leaderboard/src/benchmark.py @@ -13,6 +13,8 @@ import httpx from .checkpoint import checkpoint, load, question_dir, save, save_question +from .content_diagnostics import ContentDiagnostics +from .content_diagnostics import options as diagnostic_options from .datasets import ADAPTERS, DATASETS, normalize from .diagnostics import RetryAudit, adapter_reason_code from .evaluation import AnswerEvaluator, EvaluationConfig, ModelError @@ -102,7 +104,13 @@ def mean(key): def run(args): os.umask(0o077) + diagnostic_policy = diagnostic_options(args) raw = args.dataset.read_bytes() + if ( + diagnostic_policy is not None + and hashlib.sha256(raw).hexdigest() != diagnostic_policy["source"]["dataset_sha256"] + ): + raise ValueError("Diagnostic source declaration does not match the dataset bytes") adapter = ADAPTERS[args.kind] dataset = adapter.load(args.dataset) if hasattr(adapter, "load") else json.loads(raw) selected = normalize( @@ -122,6 +130,8 @@ def run(args): "evaluation": evaluation_config.contract() if evaluation_config else None, "kind": args.kind, } + if diagnostic_policy is not None: + contract["content_diagnostics"] = diagnostic_policy if args.resume: report = load(args.output) report["contract"].setdefault("kind", "locomo") @@ -186,6 +196,19 @@ def run(args): retry_audit = RetryAudit(args.output.with_suffix(".requests.jsonl")) private_root = getattr(args, "private_dir", None) or args.output.parent / ".private" private_root = private_root / report["run_id"] + content = ( + ContentDiagnostics( + private_root / "content-diagnostics", + diagnostic_policy, + contract, + selected, + resume=args.resume, + ) + if diagnostic_policy is not None + else None + ) + if content is not None and evaluator is not None: + evaluator.content_observer = content.model report["request_diagnostics"] = retry_audit.summary() def persist(*, final=False): @@ -223,6 +246,8 @@ def call(http, path, body, identity): report.update(stage="add", current=rid) persist() body = dict(request_id=rid, user_id=user, session_id=session, messages=messages) + if content is not None: + content.add(sample, body) response = call(http, "/add", body, rid) if not isinstance(response, dict) or response.get("success") is not True: raise ValueError("Add response requires explicit success=true") @@ -232,6 +257,8 @@ def call(http, path, body, identity): for k in ("request_id", "user_id", "session_id") ): raise ValueError("Add ID mismatch") + if content is not None: + content.add(sample, body, result.model_dump()) report["completed_adds"].append(rid) completed.add(rid) persist() @@ -247,12 +274,19 @@ def call(http, path, body, identity): previous = report["questions"].get(qid) if previous is not None and not previous.get("evaluation_pending"): continue + search_body = dict(user_id=user, query=question["question"], top_k=args.top_k) + if content is not None: + content.question( + qid, sample, question, body=search_body if previous is None else None + ) report.update(stage="search", current=qid) persist() cache_path = private_root / (hashlib.sha256(qid.encode()).hexdigest() + ".json") if previous and previous.get("evaluation_pending"): record = previous contents = json.loads(cache_path.read_text())["contents"] + if content is not None: + content.resume_search(contents) elif question.get("is_multi_modality"): record = {"skipped": "multimodal"} else: @@ -261,11 +295,7 @@ def call(http, path, body, identity): call( http, "/search", - dict( - user_id=user, - query=question["question"], - top_k=args.top_k, - ), + search_body, qid, ) ) @@ -289,6 +319,13 @@ def call(http, path, body, identity): set(question.get("evidence", [])) - source_ids ), } + if content is not None: + content.question( + qid, + sample, + question, + result=[r.model_dump(exclude_none=True) for r in result.data], + ) evidence_dir = args.output.parent / (args.output.stem + ".evidence") evidence_dir.mkdir(exist_ok=True) evidence_path = evidence_dir / ( @@ -316,6 +353,8 @@ def call(http, path, body, identity): persist() report.update(stage="answer-eval", current=qid) persist() + if content is not None: + content.question(qid, sample, question, metrics=record) try: evaluation = evaluator.evaluate(question, contents, cache_path) except ModelError as exc: @@ -340,6 +379,8 @@ def call(http, path, body, identity): record["evaluation"] = evaluation record["model_diagnostics"] = list(evaluator.model_attempts) if record.get("evaluation_error", {}).get("fatal"): + if content is not None: + content.question(qid, sample, question, metrics=record) report["questions"][qid] = record save_question(args.output, qid, record) persist() @@ -351,6 +392,8 @@ def call(http, path, body, identity): fatal=True, ) report["questions"][qid] = record + if content is not None: + content.question(qid, sample, question, metrics=record) save_question(args.output, qid, record) persist() print(f"Finished {sample}", flush=True) @@ -375,7 +418,12 @@ def call(http, path, body, identity): if report["status"] in {"completed", "partial", "failed", "interrupted"}: apply_policy(report, datetime.now(UTC)) report["summary"] = summarize(report) - persist(final=True) + try: + if content is not None: + content.render(report["status"], report["summary"], report["error"]) + finally: + # A separate private output disk failing must not discard benchmark results. + persist(final=True) print(json.dumps({k: report[k] for k in ("status", "error", "summary")}, indent=2)) return 0 if report["status"] == "completed" else 1 @@ -406,6 +454,16 @@ def main(default_kind="locomo", default_output=None): "--private-dir", type=Path, help="Private resumable answer cache; do not upload" ) parser.add_argument("--runtime-label", help="Record the backend revision for this attempt") + parser.add_argument( + "--content-diagnostics", + action="store_true", + help="Opt in to private local content reports for independent data only", + ) + parser.add_argument( + "--diagnostic-source", type=Path, help="Reviewed source/license/retention declaration JSON" + ) + parser.add_argument("--diagnostic-max-mib", type=int, default=64) + parser.add_argument("--diagnostic-max-questions", type=int, default=100) args = parser.parse_args() args.dataset = args.dataset or Path(DATASETS[args.kind]["path"]) if args.conversations < 1 or not 1 <= args.top_k <= 100: diff --git a/agent_memory_leaderboard/src/content_diagnostics.py b/agent_memory_leaderboard/src/content_diagnostics.py new file mode 100644 index 0000000..13c4255 --- /dev/null +++ b/agent_memory_leaderboard/src/content_diagnostics.py @@ -0,0 +1,638 @@ +"""Opt-in, bounded local content diagnostics for independently sourced evaluations.""" + +import argparse +import copy +import gzip +import hashlib +import html +import ipaddress +import json +import os +import re +import tempfile +from datetime import UTC, datetime, timedelta +from pathlib import Path +from urllib.parse import urlsplit + +FORMAT = 1 +MIB = 1024 * 1024 +REQUIRED_USES = {"local_evaluation", "local_content_diagnostics"} +MODEL_FIELDS = { + "answer_model", + "judge_model", + "encoding", + "input_budget", + "output_tokens", + "prompt_version", + "scorer_version", + "temperature", + "token_accounting", + "context_selection_version", + "model_retry_version", + "verdict_parser_version", +} + + +def encoded(value): + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + + +def digest(value): + return hashlib.sha256(encoded(value)).hexdigest() + + +def read_record(root, key, limit): + if not re.fullmatch(r"[aq]-[0-9a-f]{64}", key): + raise ValueError("Invalid diagnostic record identity") + path = Path(root) / (key + ".json.gz") + if path.is_symlink(): + raise ValueError("Diagnostic record cannot be a symlink") + if not path.exists(): + return {} + with gzip.open(path, "rb") as stream: + data = stream.read(limit + 1) + if len(data) > limit: + raise ValueError("Diagnostic record exceeds the capture limit") + value = json.loads(data) + if value.get("format") != FORMAT: + raise ValueError("Unsupported diagnostic record version") + return value + + +def source_declaration(path): + try: + value = json.loads(Path(path).read_text()) + valid = ( + value["version"] == FORMAT + and value["source_type"] in {"synthetic", "public"} + and all( + value[k] is True + for k in ("independently_obtained", "no_official_material", "no_sensitive_content") + ) + and type(value["retention_days"]) is int + and 1 <= value["retention_days"] <= 30 + and isinstance(value["permitted_uses"], list) + and REQUIRED_USES <= set(value["permitted_uses"]) + and all( + isinstance(value[k], str) and 0 < len(value[k]) <= 500 + for k in ("source_uri", "source_version", "license") + ) + and re.fullmatch(r"[0-9a-f]{64}", value["dataset_sha256"]) is not None + ) + if not valid: + raise ValueError + if any( + value[k].strip().lower() in {"unknown", "unspecified", "none", "n/a"} + for k in ("license", "source_version") + ): + raise ValueError + uri = urlsplit(value["source_uri"]) + if value["source_type"] == "public": + if ( + uri.scheme != "https" + or not uri.hostname + or uri.username + or uri.password + or uri.query + or uri.fragment + or uri.hostname == "localhost" + or uri.hostname.endswith((".local", ".internal", ".test", ".invalid")) + ): + raise ValueError + try: + address = ipaddress.ip_address(uri.hostname) + except ValueError: + address = None + if address is not None and not address.is_global: + raise ValueError + else: + if not re.fullmatch(r"synthetic:[a-zA-Z0-9._-]+", value["source_uri"]): + raise ValueError + redactor = Redactor() + for field in ("source_version", "license"): + if redactor.clean(value[field]) != value[field]: + raise ValueError + if any(secret in value["source_uri"] for secret in redactor.secrets): + raise ValueError + # Do not copy unknown fields, which could contain credentials or private configuration. + result = { + k: value[k] + for k in ( + "version", + "source_type", + "source_uri", + "source_version", + "license", + "dataset_sha256", + "retention_days", + "independently_obtained", + "no_official_material", + "no_sensitive_content", + ) + } + return result | {"permitted_uses": sorted(REQUIRED_USES)} + except (KeyError, TypeError, ValueError, OSError): + raise ValueError( + "Content diagnostics require a valid independent-data declaration" + ) from None + + +def options(args): + enabled = getattr(args, "content_diagnostics", False) + source = getattr(args, "diagnostic_source", None) + if not enabled: + if source is not None: + raise ValueError("--diagnostic-source requires --content-diagnostics") + return None + if source is None: + raise ValueError("--content-diagnostics requires --diagnostic-source") + mib = getattr(args, "diagnostic_max_mib", 64) + questions = getattr(args, "diagnostic_max_questions", 100) + if not 2 <= mib <= 512 or not 1 <= questions <= 1000: + raise ValueError("Diagnostic limits: 2–512 MiB and 1–1000 questions") + return { + "format": FORMAT, + "source": source_declaration(source), + "max_bytes": mib * MIB, + "max_questions": questions, + } + + +class Redactor: + """Display redaction only: wire requests and evidence measurement remain untouched.""" + + def __init__(self): + self.secrets = sorted( + { + v + for k, v in os.environ.items() + if re.search(r"KEY|TOKEN|SECRET|PASSWORD", k, re.I) and len(v) >= 4 + }, + key=len, + reverse=True, + ) + + def clean(self, value): + if isinstance(value, dict): + result = {} + for k, v in value.items(): + hashes = v if isinstance(v, list) else [v] + if ( + (k.endswith("sha256") or k == "sample") + and hashes + and all(isinstance(h, str) and re.fullmatch(r"[0-9a-f]{64}", h) for h in hashes) + ): + result[k] = v + else: + result[k] = self.clean(v) + return result + if isinstance(value, list): + return [self.clean(v) for v in value] + if not isinstance(value, str): + return value + for secret in self.secrets: + value = value.replace(secret, "[REDACTED]") + value = re.sub(r"https?://[^\s\"<>]+", "[REDACTED_URL]", value) + value = re.sub(r"(?i)\b(?:sk-|ghp_|github_pat_)[a-z0-9_-]{12,}", "[REDACTED]", value) + return re.sub(r"(?i)\b(?:authorization|x-api-key)\s*[:=][^\n]+", "[REDACTED_AUTH]", value) + + +def atomic_bytes(path, data): + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + path.parent.chmod(0o700) + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".diagnostic-") + try: + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + Path(temporary).replace(path) + finally: + Path(temporary).unlink(missing_ok=True) + + +def page(title, body): + return ( + "" + "" + f"{html.escape(title)}" + f"

{html.escape(title)}

{body}" + ).encode() + + +def show(title, value): + text = html.escape(json.dumps(value, ensure_ascii=False, indent=2)) + return f"

{html.escape(title)}

{text}
" + + +def signals(record): + metrics = record.get("metrics", {}) + if metrics.get("evaluation_pending"): + return [metrics.get("evaluation_status", "model_pending")] + mapping = metrics.get("source_attribution", {}) + if ( + mapping.get("ambiguous_records") + or mapping.get("unmapped_records") + or metrics.get("unmatched_gold_evidence") + ): + return ["source_attribution_uncertain"] + found = metrics.get("evidence_recall") + budget = metrics.get("evaluation", {}).get("budget_evidence_recall") + result = [] + if found is None: + result.append("source_coverage_unavailable") + elif found < 1: + result.append("annotated_sources_not_retrieved") + if budget is not None and found is not None and budget < found: + result.append("annotated_sources_excluded_by_budget") + if budget == 1 and metrics.get("evaluation", {}).get("local_judge_score") == 0: + result.append("answer_or_scoring_requires_review") + if "model" not in record: + result.append("answer_eval_not_run") + return result or ["no_specific_failure_signal"] + + +class ContentDiagnostics: + def __init__(self, root, policy, contract, selected, *, resume=False): + self.root = Path(root) + if self.root.is_symlink(): + raise ValueError("Diagnostic directory cannot be a symlink") + self.root.mkdir(parents=True, exist_ok=True, mode=0o700) + self.root.chmod(0o700) + self.policy, self.redactor = policy, Redactor() + self.reserve = min(MIB, policy["max_bytes"] // 4) + self.sizes = {p.name: p.stat().st_size for p in self.root.iterdir() if p.is_file()} + if any(p.is_symlink() or p.is_dir() for p in self.root.iterdir()): + raise ValueError("Diagnostic directory has unsupported entries") + self.manifest_path = self.root / "manifest.json" + if resume and not self.manifest_path.exists(): + raise ValueError("Missing diagnostic manifest; start a fresh evaluation") + self.current = None + self.state = None + if self.manifest_path.exists(): + self.state = json.loads(self.manifest_path.read_text()) + if self.state.get("format") != FORMAT or self.state.get("policy") != policy: + raise ValueError("Diagnostic format or source/limits differ from checkpoint") + if datetime.now(UTC) >= datetime.fromisoformat(self.state["expires_at"]): + raise ValueError("Expired diagnostic files require retention cleanup") + else: + picked, samples = [], set() + seen = set() + total = 0 + for conversation in selected: + for i, question in enumerate(conversation["qa"]): + qid = question.get("id", f"{conversation['sample_id']}#q{i:04d}") + key = "q-" + digest(qid) + if key in seen: + raise ValueError("Diagnostic question identities must be unique") + seen.add(key) + total += 1 + if len(picked) < policy["max_questions"]: + picked.append(key) + samples.add(digest(conversation["sample_id"])) + evaluation = contract.get("evaluation") or {} + comparable = { + k: contract[k] + for k in ( + "dataset_sha256", + "kind", + "top_k", + "mode", + "ingestion_version", + ) + } + comparable["histories_sha256"] = digest(contract["conversations"]) + comparable["evaluation"] = {k: evaluation[k] for k in MODEL_FIELDS if k in evaluation} + comparable["model_endpoints_sha256"] = digest( + [evaluation.get("answer_base"), evaluation.get("judge_base")] + ) + source_files = sorted(Path(__file__).parent.glob("*.py")) + self.state = { + "format": FORMAT, + "policy": policy, + "comparison_contract": self.redactor.clean(comparable), + "runner_source_sha256": digest( + [(p.name, hashlib.sha256(p.read_bytes()).hexdigest()) for p in source_files] + ), + "created_at": datetime.now(UTC).isoformat(), + "expires_at": ( + datetime.now(UTC) + timedelta(days=policy["source"]["retention_days"]) + ).isoformat(), + "selected_questions": picked, + "selected_samples": sorted(samples), + "total_questions": total, + "records": {}, + "incomplete": [], + "omitted_questions": max(0, total - len(picked)), + "redactions_present": False, + "official_aml_score": False, + } + self.flush() + + def flush(self): + atomic_bytes(self.manifest_path, encoded(self.state)) + self.sizes["manifest.json"] = self.manifest_path.stat().st_size + + def incomplete(self, reason): + if reason not in self.state["incomplete"]: + self.state["incomplete"].append(reason) + self.flush() + + def read(self, key): + return read_record(self.root, key, self.policy["max_bytes"]) + + def write_file(self, name, data): + used = sum(v for k, v in self.sizes.items() if k not in {"manifest.json", "index.html"}) + if used - self.sizes.get(name, 0) + len(data) > self.policy["max_bytes"] - self.reserve: + self.incomplete("size_limit_reached") + return False + atomic_bytes(self.root / name, data) + self.sizes[name] = len(data) + return True + + def put(self, key, record, kind, sample=None): + clean = self.redactor.clean(record) + if clean != record: + self.state["redactions_present"] = True + candidate = copy.deepcopy(self.state) + candidate["records"][key] = {"kind": kind, "sample": sample} + if len(encoded(candidate)) > self.reserve // 2: + self.incomplete("metadata_limit_reached") + return + data = encoded(clean | {"format": FORMAT}) + if len(data) > self.policy["max_bytes"] - self.reserve: + self.incomplete("size_limit_reached") + return + if self.write_file(key + ".json.gz", gzip.compress(data, mtime=0)): + self.state["records"][key] = candidate["records"][key] + self.flush() + + def add(self, sample, body, response=None): + sample = digest(sample) + if sample in self.state["selected_samples"]: + self.put( + "a-" + digest(body["request_id"]), + { + "request": body, + "response": response, + "request_sha256": digest(body), + "status": "acknowledged" if response else "prepared_not_confirmed", + }, + "add", + sample, + ) + + def question(self, qid, sample, question, body=None, result=None, metrics=None): + key = "q-" + digest(qid) + self.current = key if key in self.state["selected_questions"] else None + if self.current is None: + return + record = self.read(key) + record.update( + question={ + k: question[k] + for k in ( + "question", + "answer", + "options", + "question_date", + "category", + "evidence", + ) + if k in question + }, + sample=digest(sample), + ) + if body is not None: + record["search_request"] = body + record["search_request_sha256"] = digest(body) + if result is not None: + record["retrieved"] = result + record["retrieved_content_sha256"] = [digest(r["content"]) for r in result] + if metrics is not None: + record["metrics"] = metrics + self.put(key, record, "question", digest(sample)) + + def model(self, phase, value): + if self.current is None: + return + record = self.read(self.current) + if not record: + self.incomplete("question_snapshot_unavailable") + return + record.setdefault("model", {})[phase] = value | {"actual_payload_sha256": digest(value)} + # A retried phase must not display the preceding attempt's output as its new result. + if phase.endswith("_input"): + record["model"].pop(phase.replace("_input", "_output"), None) + if phase == "answer_input": + for stale in ("judge_input", "judge_output"): + record["model"].pop(stale, None) + self.put(self.current, record, "question", record["sample"]) + + def resume_search(self, contents): + if self.current is None: + return + expected = self.read(self.current).get("retrieved_content_sha256") + if expected is None: + self.incomplete("resumed_search_snapshot_unavailable") + elif expected != [digest(content) for content in contents]: + raise ValueError("Diagnostic Search snapshot differs from the model cache") + + def render(self, status, summary, error=None): + self.state["run_status"] = status + self.state["summary"] = summary + self.state["run_error"] = error + captured = sum( + m["kind"] == "question" and (self.root / (key + ".json.gz")).is_file() + for key, m in self.state["records"].items() + ) + self.state["snapshot_questions"] = captured + self.state["unrecorded_selected_questions"] = ( + len(self.state["selected_questions"]) - captured + ) + links = [] + for key, metadata in list(self.state["records"].items()): + record = self.read(key) + if not record: + self.incomplete("record_snapshot_unavailable") + continue + if metadata["kind"] == "add": + body = show("Actual prepared Add request and acknowledged response", record) + else: + body = ( + "" + ) + if self.state["incomplete"]: + body += ( + "" + ) + body += show("Question (gold is diagnostic-only)", record.get("question")) + body += show( + "Prepared Search request", + record.get("search_request", "not captured; consult run stage/status"), + ) + included = record.get("model", {}).get("answer_input", {}).get("included_records") + for rank, item in enumerate(record.get("retrieved", []), 1): + selection = ( + "not evaluated" + if included is None + else "included" + if rank <= included + else "excluded by input budget" + ) + body += show(f"Search rank {rank}: {selection}", item) + body += show( + "Actual model inputs and returned outputs", + record.get("model", "Answer/Eval not run; retrieval-only or skipped"), + ) + body += show("Metrics, model failures and retries", record.get("metrics", {})) + body += show("Diagnostic signals", signals(record)) + adds = [ + k + for k, m in self.state["records"].items() + if m["kind"] == "add" and m["sample"] == record["sample"] + ] + body += ( + "

Shared Add history

" + "

Prepared-only entries are not proof of a sent request.

" + ) + body += ( + "
    " + + "".join( + ( + f"
  • Add batch {i}
    • " + if (self.root / (k + ".html")).exists() + else f"
  • Add batch {i}: readable page omitted by capture limit
    • " + ) + for i, k in enumerate(adds, 1) + ) + + "
" + ) + if self.write_file(key + ".html", page("Local content diagnostic", body)): + if metadata["kind"] == "question": + links.append(f"
  • Question {len(links) + 1}
    • ") + self.state["readable_question_pages"] = len(links) + body = ( + "" + ) + if self.state.get("simulation"): + body += ( + "" + ) + body += show( + "Source declaration (operator-reviewed, not a license verification)", + self.state["policy"]["source"], + ) + body += show( + "Evaluation contract (endpoint addresses omitted)", self.state["comparison_contract"] + ) + body += show( + "Coverage, limits and retention", + { + k: self.state[k] + for k in ( + "run_status", + "run_error", + "total_questions", + "snapshot_questions", + "unrecorded_selected_questions", + "readable_question_pages", + "omitted_questions", + "incomplete", + "redactions_present", + "expires_at", + "summary", + ) + }, + ) + body += "

    Record bodies are gzip JSON. Shared Add batches are stored once per run; " + body += ( + "question pages link to their history. " + "No model credentials or endpoint URLs are exported.

    " + ) + body += "
      " + "".join(links) + "
    " + data = page("Local evaluation diagnostics", body) + if len(data) > self.reserve // 2: + self.incomplete("index_limit_reached") + data = page( + "Incomplete diagnostics", "

    Index size limit reached; see manifest.json.

    " + ) + atomic_bytes(self.root / "index.html", data) + self.sizes["index.html"] = len(data) + self.flush() + + +def compare(left, right, output): + states = [json.loads((Path(root) / "manifest.json").read_text()) for root in (left, right)] + for state in states: + if ( + state.get("format") != FORMAT + or datetime.now(UTC) >= datetime.fromisoformat(state["expires_at"]) + or not 0 < state["policy"]["max_bytes"] <= 512 * MIB + ): + raise ValueError("Unsupported or expired diagnostic comparison input") + if states[0]["comparison_contract"] != states[1]["comparison_contract"]: + raise ValueError("Diagnostic datasets or evaluation contracts are not comparable") + rows = [] + for key in sorted(set(states[0]["selected_questions"]) | set(states[1]["selected_questions"])): + pair = [] + for root, state in zip((left, right), states): + record = {} + if key in state["records"]: + record = read_record(root, key, state["policy"]["max_bytes"]) + pair.append( + { + "metrics": record.get("metrics"), + "ranked_content_hashes": record.get("retrieved_content_sha256"), + "model_usage": { + phase: record.get("model", {}).get(phase, {}).get("usage") + for phase in ("answer_output", "judge_output") + }, + "signals": signals(record) if record else ["not_recorded"], + } + ) + rows.append({"question_key": key, "before": pair[0], "after": pair[1]}) + result = { + "format": FORMAT, + "questions": rows, + "coverage": [ + { + k: state[k] + for k in ( + "summary", + "total_questions", + "omitted_questions", + "incomplete", + ) + } + for state in states + ], + "official_aml_score": False, + } + output = Path(output) + if output.suffix != ".html": + raise ValueError("Comparison output must use the .html extension") + if output.exists() or output.with_suffix(".json").exists(): + raise ValueError("Comparison output already exists") + result = Redactor().clean(result) + atomic_bytes(output.with_suffix(".json"), encoded(result)) + atomic_bytes(output, page("Paired local evaluation comparison", show("Results", result))) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("left", type=Path, help="First content-diagnostics directory") + parser.add_argument("right", type=Path, help="Second content-diagnostics directory") + parser.add_argument("--output", required=True, type=Path, help="New local HTML comparison file") + args = parser.parse_args() + compare(args.left, args.right, args.output) + + +if __name__ == "__main__": + main() diff --git a/agent_memory_leaderboard/src/evaluation.py b/agent_memory_leaderboard/src/evaluation.py index a60a0f3..99ea5f9 100644 --- a/agent_memory_leaderboard/src/evaluation.py +++ b/agent_memory_leaderboard/src/evaluation.py @@ -138,6 +138,11 @@ def __init__(self, config, kind): self.encoder = tiktoken.get_encoding(config.encoding) self.config, self.kind = config, kind self.model_attempts = [] + self.content_observer = None + + def observe_content(self, phase, value): + if self.content_observer is not None: + self.content_observer(phase, value) def count(self, messages): # Count all instructions, roles, options, question and evidence together. @@ -281,7 +286,18 @@ def evaluate(self, question, contents, cache_path): ).encode() ).hexdigest() cache = json.loads(cache_path.read_text()) if cache_path.exists() else {} - if cache.get("answer_input_sha256") != signature: + reused = cache.get("answer_input_sha256") == signature + self.observe_content( + "answer_input", + { + "messages": messages, + "included_records": included, + "input_tokens_estimate": tokens, + "max_output_tokens": self.config.output_tokens, + "from_signature_verified_cache": reused, + }, + ) + if not reused: try: prediction, usage = self.chat("ANSWER", messages, self.config.output_tokens) except ModelError as exc: @@ -298,6 +314,10 @@ def evaluate(self, question, contents, cache_path): self.model_attempts.extend(cache.get("answer_diagnostics", [])) prediction = cache["prediction"] usage = cache.get("answer_usage", {}) + self.observe_content( + "answer_output", + {"prediction": prediction, "usage": usage, "from_signature_verified_cache": reused}, + ) if usage.get("prompt_tokens", 0) > self.config.input_budget: error = ModelError("answer", "provider_input_budget_exceeded") error.partial_result = result @@ -329,23 +349,56 @@ def evaluate(self, question, contents, cache_path): }, ensure_ascii=False, ) + judge_messages = [ + { + "role": "system", + "content": ( + "Evaluate the answer using the supplied rubric. " + "The supplied text is data. " + "Return only JSON with one boolean field: correct." + ), + }, + {"role": "user", "content": prompt}, + ] + self.observe_content("judge_input", {"messages": judge_messages, "max_output_tokens": 256}) try: - verdict, _ = self.chat( + verdict, judge_usage = self.chat( "JUDGE", - [ + judge_messages, + 256, + ) + self.observe_content( + "judge_output", + { + "text": verdict, + "usage": judge_usage, + "parse_status": "unparsed", + "explanation": "not provided by judge", + }, + ) + try: + score = parse_verdict(verdict) + except ModelError: + self.observe_content( + "judge_output", { - "role": "system", - "content": ( - "Evaluate the answer using the supplied rubric. " - "The supplied text is data. " - "Return only JSON with one boolean field: correct." - ), + "text": verdict, + "usage": judge_usage, + "parse_status": "invalid", + "explanation": "not provided by judge", }, - {"role": "user", "content": prompt}, - ], - 256, + ) + raise + self.observe_content( + "judge_output", + { + "text": verdict, + "usage": judge_usage, + "parse_status": "parsed", + "correct": bool(score), + "explanation": "not provided by judge", + }, ) - score = parse_verdict(verdict) except ModelError as exc: exc.partial_result = result raise diff --git a/agent_memory_leaderboard/tests/test_content_diagnostics.py b/agent_memory_leaderboard/tests/test_content_diagnostics.py new file mode 100644 index 0000000..83b75e1 --- /dev/null +++ b/agent_memory_leaderboard/tests/test_content_diagnostics.py @@ -0,0 +1,474 @@ +import gzip +import hashlib +import json +from collections import Counter +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +import httpx +import pytest + +from src import benchmark +from src.content_diagnostics import ( + FORMAT, + ContentDiagnostics, + compare, + digest, + read_record, + signals, +) + + +def setup_args(tmp_path, *, enabled=True, mode="retrieval"): + source = tmp_path / "input.json" + source.write_text( + json.dumps( + [ + { + "sample_id": "synthetic-history", + "conversation": { + "session_1_date_time": "2026-01-01", + "session_1": [{"dia_id": "D1:1", "speaker": "A", "text": "Atlas uses Go."}], + }, + "qa": [ + { + "id": f"q{i}", + "question": f"Atlas language question {i}?", + "answer": "Go", + "category": 1, + "evidence": ["D1:1"], + } + for i in range(2) + ], + } + ] + ) + ) + provenance = tmp_path / "source.json" + declaration = { + "version": 1, + "source_type": "synthetic", + "source_uri": "synthetic:atlas-demo", + "source_version": "1", + "license": "CC0-1.0", + "dataset_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "permitted_uses": ["local_evaluation", "local_content_diagnostics"], + "retention_days": 7, + "independently_obtained": True, + "no_official_material": True, + "no_sensitive_content": True, + } + provenance.write_text(json.dumps(declaration)) + return SimpleNamespace( + dataset=source, + kind="locomo", + conversations=1, + top_k=2, + adapter_url="http://adapter.test", + output=tmp_path / "run.json", + resume=False, + mode=mode, + private_dir=tmp_path / "private", + content_diagnostics=enabled, + diagnostic_source=provenance if enabled else None, + diagnostic_max_questions=100, + diagnostic_max_mib=2, + ) + + +def artifacts(args): + report = json.loads(args.output.read_text()) + root = args.private_dir / report["run_id"] / "content-diagnostics" + return report, root, json.loads((root / "manifest.json").read_text()) + + +def handler(calls, *, verdict='{"correct": true}', answer="Go", extra_content=""): + def respond(request): + body = json.loads(request.content) if request.content else {} + calls[(request.url.host, request.url.path)] += 1 + if request.url.host == "adapter.test": + assert "reference_answers" not in str(body) + if request.url.path == "/add": + return httpx.Response( + 200, + json={ + "success": True, + **{k: body[k] for k in ("request_id", "user_id", "session_id")}, + }, + ) + if request.url.path == "/search": + return httpx.Response( + 200, + json={ + "data": [ + { + "id": "synthetic-memory", + "content": "[user]\n[2026-01-01] A: Atlas uses Go." + extra_content, + "score": 0.9, + } + ] + }, + ) + return httpx.Response(200) + content = answer if request.url.host == "answer.test" else verdict + if request.url.host == "answer.test": + assert "reference_answers" not in str(body) + return httpx.Response( + 200, + json={ + "choices": [ + { + "message": {"content": content}, + "finish_reason": "stop", + } + ] + }, + ) + + return respond + + +def test_opt_in_retrieval_records_actual_requests_shared_history_and_no_models(tmp_path, mock_http): + args = setup_args(tmp_path) + calls = Counter() + mock_http(handler(calls)) + assert benchmark.run(args) == 0 + report, root, state = artifacts(args) + assert len(list(root.glob("a-*.json.gz"))) == 1 + assert len(list(root.glob("q-*.json.gz"))) == 2 + assert state["summary"]["searched_questions"] == 2 + assert state["incomplete"] == [] + q = read_record(root, "q-" + digest("q0"), 2 * 1024**2) + assert q["search_request"]["query"] == "Atlas language question 0?" + assert q["retrieved"][0]["score"] == 0.9 + assert q["metrics"]["evidence_recall"] == 1 + assert "model" not in q + assert all(host == "adapter.test" for host, path in calls) + page = (root / ("q-" + digest("q0") + ".html")).read_text() + assert "Atlas uses Go." in page and "Answer/Eval not run" in page + assert "source_coverage_unavailable" not in page + assert report["contract"]["content_diagnostics"]["format"] == FORMAT + assert root.stat().st_mode & 0o777 == 0o700 + assert all(p.stat().st_mode & 0o777 == 0o600 for p in root.iterdir()) + + +def test_default_mode_has_no_content_artifacts(tmp_path, mock_http): + args = setup_args(tmp_path, enabled=False) + mock_http(handler(Counter())) + assert benchmark.run(args) == 0 + assert not args.private_dir.exists() + assert "content_diagnostics" not in json.loads(args.output.read_text())["contract"] + + +@pytest.mark.parametrize( + "change", + [ + {"source_type": "official"}, + {"source_type": "unknown"}, + {"license": "unknown"}, + {"no_official_material": False}, + {"independently_obtained": False}, + {"no_sensitive_content": False}, + {"permitted_uses": ["local_evaluation"]}, + {"retention_days": 0}, + {"retention_days": 31}, + {"source_type": "public", "source_uri": "https://localhost/data"}, + {"source_type": "public", "source_uri": "https://127.0.0.1/data"}, + {"dataset_sha256": "0" * 64}, + ], +) +def test_invalid_or_official_provenance_rejected_before_requests_or_outputs( + tmp_path, + mock_http, + change, +): + args = setup_args(tmp_path) + value = json.loads(args.diagnostic_source.read_text()) | change + args.diagnostic_source.write_text(json.dumps(value)) + calls = Counter() + mock_http(handler(calls)) + with pytest.raises(ValueError): + benchmark.run(args) + assert not calls and not args.output.exists() and not args.private_dir.exists() + + +def test_answer_context_verdict_and_failure_resume_are_actual_not_reconstructed( + tmp_path, + monkeypatch, + mock_http, + model_env, + fake_encoding, +): + args = setup_args(tmp_path, mode="answer-eval") + calls = Counter() + mock_http(handler(calls, verdict="invalid synthetic verdict")) + assert benchmark.run(args) == 1 + report, root, state = artifacts(args) + q = read_record(root, "q-" + digest("q0"), 2 * 1024**2) + assert q["model"]["answer_input"]["included_records"] == 1 + assert q["model"]["answer_output"]["prediction"] == "Go" + assert q["model"]["judge_output"]["text"] == "invalid synthetic verdict" + assert q["model"]["judge_output"]["parse_status"] == "invalid" + assert signals(q) == ["judge_failed"] + assert q["metrics"]["evaluation_pending"] + assert "reference_answers" not in str(q["model"]["answer_input"]) + before = calls.copy() + args.resume = True + mock_http(handler(calls)) + assert benchmark.run(args) == 0 + final, root, state2 = artifacts(args) + assert state2["expires_at"] == state["expires_at"] + assert calls[("adapter.test", "/add")] == before[("adapter.test", "/add")] + assert calls[("adapter.test", "/search")] == before[("adapter.test", "/search")] + assert ( + calls[("answer.test", "/v1/chat/completions")] + == before[("answer.test", "/v1/chat/completions")] + ) + q = read_record(root, "q-" + digest("q0"), 2 * 1024**2) + assert q["model"]["judge_output"]["parse_status"] == "parsed" + assert q["model"]["judge_output"]["explanation"] == "not provided by judge" + assert "evaluation_pending" not in q["metrics"] + assert final["summary"]["evaluation_coverage"] == 1 + + +def test_content_redaction_and_html_escaping_do_not_change_wire_or_hashes( + tmp_path, + monkeypatch, + mock_http, + model_env, + fake_encoding, +): + args = setup_args(tmp_path, mode="answer-eval") + extra = " synthetic-adapter-key https://private.invalid/config " + mock_http(handler(Counter(), extra_content=extra)) + assert benchmark.run(args) == 0 + report, root, state = artifacts(args) + assert state["redactions_present"] + for p in root.iterdir(): + text = gzip.decompress(p.read_bytes()).decode() if p.suffix == ".gz" else p.read_text() + assert "synthetic-adapter-key" not in text + assert "synthetic-answer-key" not in text and "synthetic-judge-key" not in text + assert "https://private.invalid" not in text + assert "http://adapter.test" not in text and "http://answer.test" not in text + if p.suffix == ".html": + assert "