diff --git a/evaluation/README.md b/evaluation/README.md index 89aa7a1502..01e29633f1 100644 --- a/evaluation/README.md +++ b/evaluation/README.md @@ -7,7 +7,7 @@ Python environment and build configuration. | Directory | Purpose | Suites | | --- | --- | --- | | [`coding/`](coding/README.md) | Coding-task completion and continuation quality | [SWE-bench Pro](coding/swebench_pro/README.md), [Work continuity](coding/work_continuity/README.md) | -| [`memory/`](memory/README.md) | Memory retrieval and answer quality | [LoCoMo](memory/locomo/README.md), [LoCoMo-Plus](memory/locomo_plus/README.md), [LongMemEval-V2](memory/longmemeval_v2/README.md) | +| [`memory/`](memory/README.md) | Memory retrieval, answer quality, and write-gate evidence | [LoCoMo](memory/locomo/README.md), [LoCoMo-Plus](memory/locomo_plus/README.md), [LongMemEval-V2](memory/longmemeval_v2/README.md), [Memory write gate](memory/write_gate/README.md) | | [`performance/`](performance/README.md) | Load, latency, storage growth, and capacity limits | None yet | | [`skills/`](skills/README.md) | Agent instruction routing and authorization regressions | [Claude Code skill-up](skills/skill-up/README.md) | diff --git a/evaluation/memory/README.md b/evaluation/memory/README.md index c938ba80b4..603338ed1b 100644 --- a/evaluation/memory/README.md +++ b/evaluation/memory/README.md @@ -9,6 +9,7 @@ the shared `evaluation/` environment. | [LoCoMo](locomo/README.md) | End-to-end Source ingestion, Memory retrieval, answer generation, and judging on LoCoMo. | | [LongMemEval-V2](longmemeval_v2/README.md) | Long-horizon memory retrieval, prepared context, reader answers, and official scoring. | | [LoCoMo-Plus](locomo_plus/README.md) | Factual and cognitive memory evaluation with bounded smoke and full profiles. | +| [Memory write gate](write_gate/README.md) | Offline evidence-sufficiency calibration, held-out checks, and promotion decision. | Add dataset-specific Memory quality suites here. Put latency, throughput, storage, and capacity measurements in [performance](../performance/README.md). Keep each suite's execution and grading tools with the suite; see diff --git a/evaluation/memory/write_gate/.gitignore b/evaluation/memory/write_gate/.gitignore new file mode 100644 index 0000000000..929ec078ce --- /dev/null +++ b/evaluation/memory/write_gate/.gitignore @@ -0,0 +1,6 @@ +# Generated inputs and run evidence stay outside Git. +/data/xnli/calibration.json +/data/xnli/held_out.json +/results/jev-1.13/* +!/results/jev-1.13/RESULTS.md +/data/xnli/validation.json diff --git a/evaluation/memory/write_gate/PROTOCOL.md b/evaluation/memory/write_gate/PROTOCOL.md new file mode 100644 index 0000000000..4c3b0594fe --- /dev/null +++ b/evaluation/memory/write_gate/PROTOCOL.md @@ -0,0 +1,85 @@ +# Memory 写入门评测规则 + +本规则针对 #1930 的离线评测。模型的实际回答是被测对象，数据集标签是评分依据。模型不得读取标签、另一套题的结果或已有模拟回答。 + +## 项目规定与本次选择 + +| 项目依据 | 本次执行规则 | +| --- | --- | +| [#1930](https://github.com/oceanbase/powercontext/issues/1930)：可复现、授权感知，缺失输入与负面判断分开 | 生产引用通过当前 Scope 权限读取精确版本；公开题集通过固定归档、题目身份和摘要读取。两种输入分别标明来源。 | +| #1930、[RFC #1770](https://github.com/oceanbase/powercontext/pull/1770)：带标签的 calibration 与独立 held-out，中英文 | 公开 XNLI 1.0 的 dev 只用于 calibration，test 只用于 held-out；同一道题的中英文版本保持在同一组。 | +| #1930：比较 outcome mapping、confidence threshold | 比较 yes/no 的 hold 方向和 none/0.5/0.7/0.9。这个网格是实验选择，不是官方及格线。先看 calibration，冻结配置后才运行 held-out。 | +| #1930：误拦、误放、fallback、延迟、请求成本、hold/review friction | 给出计数、分母、分语言结果、95% Wilson 区间、按原因统计的 fallback。实际调用才记录延迟和 usage；金额必须有价格依据，人工时间必须有操作测量。 | +| RFC #1770：已知高严重度错误动作、摩擦预算、回滚 | 正向启用建议必须说明严重错误检查结果及部署接受的耗时、金额、拦截和审核预算；未确定预算时标为未确定，不冒充达标。 | +| #1930：不可变 revision、精确引用、Scope 隔离、fallback 不算已裁决 | 离线评测不写 Memory 和观察记录；授权、缺失引用、版本变化、故障由行为测试检查。 | + +## 公开题集与标签转换 + +使用 [XNLI 1.0 官方数据](https://github.com/facebookresearch/XNLI)。它有现成的人类标签、中英文对应题和独立 dev/test。 +本次固定抽取 calibration 每种原始标签 100 个题组、held-out 每种原始标签 300 个题组，每组保留英文与中文。 +因此目标规模是 calibration 600 条、held-out 1,800 条。这个规模是本次有界实验选择，不是 issue 要求。 + +- `sentence1` / premise 作为证据，`sentence2` / hypothesis 作为候选记忆。 +- `entailment` 映射为 `accept`：证据支持候选。 +- `contradiction` 映射为 `hold`：证据与候选矛盾。 +- `neutral` 映射为 `hold`：在给定证据下无法推出候选，不等于候选在现实中一定为假。 +- 模型的 `abstain`、故障或隐私拒绝是未裁决，不能把它们改成数据集的 `neutral` 标签。 + +按稳定 SHA-256 排序抽样，不按模型成绩选题。同一套题也只保留无重复证据或候选的题组。 +不确定、缺失或不一致标签不进入题集，并报告排除数量。 +相同题目、相同证据或相同候选不得跨两套题；翻译对应版本不能分到两侧。 +公开数据标签转换是本评测的声明，不是 XNLI 作者对 PowerContext 写入行为的背书。 + +XNLI 的正文留在外部归档中；题目清单按固定规则生成到 Git 忽略目录，只保存题目身份、字段、标签和内容摘要。 +仓库提交生成程序、规则及清单摘要，不提交展开的 2,400 条清单。 +[数据许可](https://github.com/facebookresearch/XNLI/blob/main/LICENSE)是 CC BY-NC 4.0；该研究题集不作为产品数据分发。 + +## 结果能说明什么 + +XNLI 衡量句子级证据支持能力，不能单独证明真实 Memory 写入的工作负担、多条互补证据处理、历史引用恢复率或生产准确率。 +这些覆盖缺口必须在结论中列明；公开题集的错误率不能直接外推为生产每百次写入的错误或审核次数。 +已被模型训练使用的题集可能产生偏高成绩，报告必须声明无法核实训练重叠。 + +官方没有指定模型品牌、参数规模、最低准确率、价格或耗时的统一及格线。 +本规则确定数据、评分和测量方法；部署的数值预算需要作为具体策略提案声明，并在看到 held-out 结果前固定。 +95% 区间用于表达样本不确定性，不是官方晋级门槛，也不把两种语言的对应题当成彼此独立的 1,800 个场景。 + +## 本次模型选择与调用范围 + +首个被测候选为 `typesafe/jev-1.13`，通过 OpenRouter 的 Decisions API 调用。它是本次实验选择， +不是 #1930 或 RFC 指定的后端。选择依据是它直接返回选择题答案与定义明确的 confidence，适合比较阈值， +且仓库已有可复用的 System One 适配器；这些理由不构成准确率证据。 +本轮不部署本地模型，不要求增加第二个对照模型。 + +本轮公开题集的配置选择规则在完整 calibration 调用前固定：最小化误拦率与包含 fallback 的误放率的 +算术平均值；持平时依次选择误拦率更低、阈值更高、`hold_on=yes` 的配置。每类错误以本类标签数为分母， +避免题集中 `hold` 占两倍数量时支配选择。这是有界研究实验的等权选择，不是项目规定的损失排序或上线及格线。 +不根据 held-out 成绩重新选择配置，不尝试不同模型或问题版本来挑选最高成绩。 + +只向该接口发送固定官方 XNLI 归档恢复的公开句对，不发送生产 Memory、Source、Scope 凭据或标签。 +程序先经过现有写入门的输入检查并取得原有问题，再调用 Jev；答案在本地复用现有规则重新评分。 +它衡量公开题目上的模型表现与假设执行效果，不声称已完成 Runtime 的云端传输接入。 +运行报告明确记录这一区别；正式云端写入门的依赖由 #1973 / #1987 处理。 + +先跑完整 calibration，冻结配置后才运行 held-out。 +记录模型版本标识、适配器问题版本、执行时间、代码摘要，以及实际用量和返回的美元费用。 +云端权重无法核实时明确标注；接口返回的 confidence 不解释为实际正确率。 +错误码有界，失败详情和原文不写入报告。若任何调用缺少费用，报告已知小计，完整金额保持未知。 + +## 程序检查 + +私有题集跨 calibration 和 held-out 检查候选及完整证据组合的重复：同时比较 Scope 内的固定版本引用和内容摘要， +不依赖题号、场景编号、标签或证据排列顺序。相同内容复制为其他记录或放在其他 Scope 中也不能跨集合重复。 +仅共用证据但候选文本不同，或候选文本相同但实际证据投影不同，不按重复输入拒绝。 +上述引用与记录摘要检查不读取或保存正文。 + +私有输入还需在授权解析后比较实际候选文本和完整证据投影的摘要，防止仅修改 `kind` 等记录字段后重复评测相同文本。 +该检查逐题处理临时正文，只保留摘要；缺失、未授权和不支持的引用仍由评测单独报告覆盖状态。 +`SplitCheckedResolver` 在每次授权读取后检查本次实际输入，再交给 gate，不额外读取或缓存正文。 +Python 调用方必须为两套题共用同一个 `SplitCheckedResolver`，不能分别创建后各自运行。 + +手写小例子仅用于验证读不到输入、未授权、引用不匹配和 fallback 等程序行为，不进入正式公开题集成绩。 +它们是普通测试素材，不是 issue 要求的官方题库，也不支持启用或不启用的质量结论。 + +依据：#1930、RFC #1770；[#1943](https://github.com/oceanbase/powercontext/issues/1943)的相关评测约定； +[scikit-learn 对独立测试集和分组切分的说明](https://scikit-learn.org/stable/modules/cross_validation.html)。 diff --git a/evaluation/memory/write_gate/README.md b/evaluation/memory/write_gate/README.md new file mode 100644 index 0000000000..c70cea1da7 --- /dev/null +++ b/evaluation/memory/write_gate/README.md @@ -0,0 +1,129 @@ +# Memory write-gate promotion evaluation + +Offline, read-only evaluation for RFC #1770 and issue #1930. It never changes Memory revisions, policy modes or the +private observation ledger. The existing Runtime owns the gate, its question and outcome mapping. + +## Prepare the fixed public dataset + +[PROTOCOL.md](PROTOCOL.md) defines sampling, label mapping and calibration selection. Download the official +[XNLI 1.0 archive](https://dl.fbaipublicfiles.com/XNLI/XNLI-1.0.zip) outside Git. Its pinned SHA-256 is +`4ba1d5e1afdb7161f0f23c66dc787802ccfa8a25a3ddd3b165a35e50df346ab1`. + +Choose your input and output locations before running the commands. Set `MEMORY_GATE_ARCHIVE` to the downloaded +archive, `MEMORY_GATE_DATA_DIR` to a new generated-input directory, and `MEMORY_GATE_OUTPUT_DIR` to an output directory. +Keep generated inputs and outputs outside version control. For cloud calls, set `MEMORY_GATE_KEY_FILE` to your credential +file, or use the supported API-key environment variable. + +From a clean checkout, generate 600 calibration and 1,800 held-out reference-only cases: + +```bash +uv sync --locked --group dev +mkdir -p "$MEMORY_GATE_OUTPUT_DIR" +uv run python -m evaluation.memory.write_gate.prepare_public \ + --archive "$MEMORY_GATE_ARCHIVE" --output-dir "$MEMORY_GATE_DATA_DIR" +uv run python -m evaluation.memory.write_gate \ + --calibration "$MEMORY_GATE_DATA_DIR/calibration.json" \ + --held-out "$MEMORY_GATE_DATA_DIR/held_out.json" \ + --public-archive "$MEMORY_GATE_ARCHIVE" --backend no_call \ + --output "$MEMORY_GATE_OUTPUT_DIR/input-coverage.json" +``` + +Preparation refuses an existing output directory. The generated manifest hashes must match +[data/xnli/preparation.json](data/xnli/preparation.json). Git contains the preparation rules and small summaries; +expanded manifests, the raw corpus and new run output stay outside version control. XNLI has a CC BY-NC 4.0 license. +Labels never enter model requests. Translation groups and duplicate text do not cross splits. + +`no_call` resolves inputs without model inference. Its `input_coverage_audit` report preserves local preflight HOLD, +with no selected model configuration. Local-rule judgments are not model-quality evidence. + +## Live Jev evaluation + +The public backend calls `typesafe/jev-1.13` through OpenRouter's Decisions API using the existing System One adapter. +It accepts only the pinned public archive, never private Server references or arbitrary text. It captures the gate's +question, makes the cloud request outside Runtime, and rescores the stored answer locally. The cloud model is not +marked local. This does not implement Runtime hosted-redacted transport or qualify a hosted deployment. + +Supply an OpenRouter key from a file outside the repository, `POWERCONTEXT_EVAL_JEV_API_KEY` or `OPENROUTER_API_KEY`. +An ordinary OpenAI credential is not reused. Keys and raw error responses are not written to reports. + +Run calibration first: + +```bash +uv run python -m evaluation.memory.write_gate \ + --backend jev --calibration-only \ + --calibration "$MEMORY_GATE_DATA_DIR/calibration.json" \ + --held-out "$MEMORY_GATE_DATA_DIR/held_out.json" \ + --public-archive "$MEMORY_GATE_ARCHIVE" --jev-key-file "$MEMORY_GATE_KEY_FILE" \ + --output "$MEMORY_GATE_OUTPUT_DIR/jev-calibration.json" +``` + +Compare the eight calibration configurations, freeze a selection, then use the following held-out command. +The example configuration must be replaced if calibration selects another: + +```bash +uv run python -m evaluation.memory.write_gate \ + --backend jev --configuration 'hold_on=yes;threshold=none' \ + --calibration-report "$MEMORY_GATE_OUTPUT_DIR/jev-calibration.json" \ + --calibration "$MEMORY_GATE_DATA_DIR/calibration.json" \ + --held-out "$MEMORY_GATE_DATA_DIR/held_out.json" \ + --public-archive "$MEMORY_GATE_ARCHIVE" --jev-key-file "$MEMORY_GATE_KEY_FILE" \ + --output "$MEMORY_GATE_OUTPUT_DIR/jev-held-out.json" +``` + +Each input makes at most one request, without automatic retry; all configurations reuse the answer. Calibration reuse +requires matching data, model, question and code. A partial or mismatched report cannot serve as a current cache. +Without `--calibration-report`, the held-out command also reruns calibration. A sibling `.cases.jsonl` checkpoint +flushes each bounded result. Credit, authentication or rate-limit refusal stops new calls; skipped cases are +`not_attempted`. Interruptions can leave checkpoints without an aggregate report. Output paths must be new. + +## Report identity + +New reports expose a path-to-SHA-256 code manifest including the shared assessor, question, Memory gate protocols and +inference result/error contracts. Its canonical digest invalidates old calibration after any covered code change. +Historical answers stay historical evidence and are not accepted as current-code calibration caches. + +## Authorized private inputs + +Private manifests retain references and labels, not production text. `resolver.py` uses current Scope authorization +and exact Artifact revisions through `PowerContextClient`. Provide `--server-url` and `POWERCONTEXT_CLIENT_API_TOKEN`. +Source reads require a pinned `gate_input_digest` matching the reconstructed original projection. Pending +`candidate:1` references cannot be recovered as immutable Memory. Missing, unauthorized or unsupported input is +reported separately from a negative judgment. These limits are not silently replaced with current content. + +Use `--backend no_call --server-url "$MEMORY_GATE_SERVER_URL"` with private manifests to audit input coverage. +For authorized re-evaluation with an existing local-only `DecisionModel`, use the Python interface: + +```python +from evaluation.memory.write_gate.dataset import load_dataset +from evaluation.memory.write_gate.resolver import AuthorizedClientResolver, SplitCheckedResolver + + +def prepare_private_evaluation(calibration_path, held_out_path, client): + calibration, _ = load_dataset(calibration_path) + held_out, _ = load_dataset(held_out_path) + resolver = SplitCheckedResolver(calibration, held_out, AuthorizedClientResolver(client)) + return calibration, held_out, resolver +``` + +The caller supplies a configured `PowerContextClient` and local-only model. The Runtime privacy boundary still applies. +Share this resolver between both splits: it compares each authorized candidate/evidence projection before scoring, +retaining only hashes and performing no additional reads. Missing, unauthorized or unsupported inputs remain separate +coverage results. +Use `evaluate_dataset(dataset, resolver, model=local_model)` from `runner` to run calibration before choosing a +configuration, then evaluate held-out and pass both result sets to `report(..., selected_configuration=...)`. +Input labels are used only for scoring. Controlled models in tests verify program behavior; they do not measure +model quality. + +## Read the measurements + +`input_coverage` counts resolved cases; `adjudication_coverage` counts substantive local/model judgments. +Unresolved cases do not enter error denominators. Abstention and fallback are unadjudicated; fail-open cases count +in the separate end-to-end false-pass effect. `used_fallback=true` always implies `coverage=unadjudicated`. +FLAG permits writing, so it counts as a pass. Missing confidence does not downgrade HOLD under current Runtime rules. + +Latency and cost describe the recorded attempt; retries require accounting for earlier attempts separately. +Backend attempts and provider-returned request counts differ. Missing usage leaves full token/currency totals unknown, +with a separately named known subtotal. HOLD/FLAG rates describe this sample; human review time and production +friction are unmeasured. Jev confidence is not a calibrated probability of correctness; the provider ID does not +attest immutable weights. XNLI does not establish production, multi-citation or training-overlap performance. +See [the measured results](results/jev-1.13/RESULTS.md) for the measured recommendation and rollback procedure. diff --git a/evaluation/memory/write_gate/__init__.py b/evaluation/memory/write_gate/__init__.py new file mode 100644 index 0000000000..0807b9a75e --- /dev/null +++ b/evaluation/memory/write_gate/__init__.py @@ -0,0 +1,15 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Offline, read-only evaluation of the governed Memory write gate.""" diff --git a/evaluation/memory/write_gate/__main__.py b/evaluation/memory/write_gate/__main__.py new file mode 100644 index 0000000000..3448caa78d --- /dev/null +++ b/evaluation/memory/write_gate/__main__.py @@ -0,0 +1,19 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Run the standalone Memory write-gate evaluation command.""" + +from .cli import main + +raise SystemExit(main()) diff --git a/evaluation/memory/write_gate/artifacts.py b/evaluation/memory/write_gate/artifacts.py new file mode 100644 index 0000000000..55e80e04b3 --- /dev/null +++ b/evaluation/memory/write_gate/artifacts.py @@ -0,0 +1,71 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Read measured calibration answers without paying to evaluate them again.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from pydantic import TypeAdapter + +from powercontext.builtin.decision_observations import DecisionCoverage + +from .dataset import Dataset +from .runner import HOLD_DIRECTIONS, THRESHOLDS, CaseResult, EvaluationStatus, configuration_key + + +def load_calibration_results( + path: Path, + dataset: Dataset, + *, + dataset_hashes: dict[str, str], + code_digest: str, + backend_policy_id: str, + model_id: str, +) -> tuple[CaseResult, ...]: + """Require the full calibration run from the same code, model, policy and data.""" + + value = json.loads(path.read_text(encoding="utf-8")) + expected = { + "schema_version": 1, + "phase": "calibration_only", + "run_scope": "evaluation", + "execution": "hosted_public_benchmark", + "dataset_id": dataset.dataset_id, + "dataset_hashes": dataset_hashes, + "evaluation_code_digest": code_digest, + "backend_model_id": model_id, + "decision_instructions_version": backend_policy_id, + } + if not isinstance(value, dict) or any(value.get(key) != item for key, item in expected.items()): + raise ValueError("calibration report does not match this full evaluation") # noqa: TRY003 + rows = TypeAdapter(tuple[CaseResult, ...]).validate_python(value.get("cases")) + if len(rows) != len(dataset.cases): + raise ValueError("calibration report has incomplete case coverage") # noqa: TRY003 + configurations = { + configuration_key(direction, threshold) for direction in HOLD_DIRECTIONS for threshold in THRESHOLDS + } + for case, row in zip(dataset.cases, rows, strict=True): + if ( + (row.case_id, row.group_id, row.language, row.label, row.original_label) + != (case.case_id, case.group_id, case.language, case.label, case.original_label) + or row.status is EvaluationStatus.NOT_ATTEMPTED + or (row.used_fallback and row.coverage is not DecisionCoverage.UNADJUDICATED) + or row.actions is None + or set(row.actions) != configurations + ): + raise ValueError("calibration report has invalid case identity or coverage") # noqa: TRY003 + return rows diff --git a/evaluation/memory/write_gate/cli.py b/evaluation/memory/write_gate/cli.py new file mode 100644 index 0000000000..779158c95f --- /dev/null +++ b/evaluation/memory/write_gate/cli.py @@ -0,0 +1,255 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Offline Memory write-gate evaluation; output contains no source bodies.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import subprocess +import sys +from datetime import UTC, datetime +from hashlib import sha256 +from pathlib import Path +from typing import Any + +import httpx +from pydantic import SecretStr + +from powercontext.client.client import PowerContextClient + +from .artifacts import load_calibration_results +from .dataset import load_dataset, validate_split_pair +from .jev import JEV_ENDPOINT, JEV_MODEL, PublicJevModel +from .report import calibration_report, case_record, report +from .resolver import AuthorizedClientResolver, PublicDatasetResolver, SplitCheckedResolver +from .runner import HOLD_DIRECTIONS, THRESHOLDS, configuration_key, evaluate_dataset, evaluate_public_dataset + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m evaluation.memory.write_gate") + parser.add_argument("--calibration", type=Path) + parser.add_argument("--held-out", type=Path) + parser.add_argument("--public-archive", type=Path, help="external official XNLI archive pinned by the manifests") + parser.add_argument("--backend", choices=("no_call", "jev"), default="no_call") + parser.add_argument("--jev-key-file", type=Path, help="file containing only an OpenRouter API key") + parser.add_argument( + "--calibration-report", type=Path, help="reuse a full measured Jev calibration run for held-out" + ) + parser.add_argument( + "--calibration-only", action="store_true", help="compare candidates without evaluating held-out" + ) + parser.add_argument("--server-url", help="use current Server authorization for exact production references") + parser.add_argument( + "--configuration", + choices=[configuration_key(direction, threshold) for direction in HOLD_DIRECTIONS for threshold in THRESHOLDS], + help="configuration frozen from calibration before scoring held-out", + ) + parser.add_argument("--output", type=Path, help="write a content-free JSON report") + args = parser.parse_args(argv) + if args.output and ( + args.output.exists() or (args.backend == "jev" and args.output.with_suffix(".cases.jsonl").exists()) + ): + parser.error("output or case checkpoint already exists; choose a new report path") + _validate_arguments(parser, args) + + calibration, calibration_hash = load_dataset(args.calibration) + held_out, held_out_hash = load_dataset(args.held_out) + validate_split_pair(calibration, held_out) + args.dataset_hashes = {"calibration": calibration_hash, "held_out": held_out_hash} + if bool(calibration.public_source) != bool(args.public_archive): + parser.error("public manifests require --public-archive; private manifests cannot use it") + if args.server_url: + token = os.environ.get("POWERCONTEXT_CLIENT_API_TOKEN") + if not token: + parser.error("POWERCONTEXT_CLIENT_API_TOKEN is required for authorized Server reads") + result = asyncio.run(_evaluate_remote(args, calibration, held_out, token)) + else: + resolver = PublicDatasetResolver(args.public_archive) + result = asyncio.run(_evaluate(calibration, held_out, resolver, args)) + hashes = {"calibration": calibration_hash, "held_out": held_out_hash} + calibration_results, held_out_results, selected = result + metadata: dict[str, Any] = { + "dataset_id": calibration.dataset_id, + "dataset_hashes": hashes, + "code_commit": _commit(), + "evaluation_code_digest": _code_digest(), + "evaluation_code_files": _code_identity(), + "execution": "hosted_public_benchmark" if args.backend == "jev" else "no_call", + "backend_model_id": JEV_MODEL if args.backend == "jev" else None, + "input_source": "authorized_references" if args.server_url else "public_xnli", + } + if args.calibration_only: + value = calibration_report(calibration_results, **metadata) + else: + value = report(calibration_results, held_out_results, selected_configuration=selected, **metadata) + if args.backend == "jev": + value.update({ + "completed_at_utc": datetime.now(UTC).isoformat(), + "backend_endpoint": JEV_ENDPOINT, + "decision_instructions_version": args.jev_policy_id, + "model_revision_guarantee": "versioned provider ID; immutable server weights are not verifiable", + "runtime_hosted_egress_validated": False, + "evaluation_path": "public benchmark model call followed by local Runtime outcome rescoring", + "run_scope": "evaluation", + "calibration_report_sha256": sha256(args.calibration_report.read_bytes()).hexdigest() + if args.calibration_report + else None, + "calibration_calls_reused": bool(args.calibration_report), + "case_checkpoint": args.output.with_suffix(".cases.jsonl").name, + }) + output = json.dumps(value, ensure_ascii=False, indent=2) + "\n" + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.open("x", encoding="utf-8") as destination: + destination.write(output) + else: + print(output, end="") + return 0 + + +def _validate_arguments(parser: argparse.ArgumentParser, args: argparse.Namespace) -> None: + if args.calibration is None or args.held_out is None: + parser.error("provide explicit --calibration and --held-out manifests") + _validate_jev_arguments(parser, args) + if bool(args.public_archive) == bool(args.server_url): + parser.error("provide either --public-archive or --server-url") + if not args.calibration_only and args.backend != "no_call" and args.configuration is None: + parser.error("declare --configuration before scoring held-out cases") + if (args.backend == "no_call" or args.calibration_only) and args.configuration is not None: + parser.error("this run cannot select a decision configuration") + + +def _validate_jev_arguments(parser: argparse.ArgumentParser, args: argparse.Namespace) -> None: + if args.calibration_report and (args.backend != "jev" or args.calibration_only): + parser.error("--calibration-report requires a Jev held-out run") + if args.jev_key_file and args.backend != "jev": + parser.error("--jev-key-file requires --backend jev") + if args.backend != "jev": + return + if not args.public_archive or args.server_url: + parser.error("Jev is available only for the pinned public XNLI archive, without Server reads") + if not args.output: + parser.error("Jev evaluation requires --output to preserve the paid run's report") + try: + key = ( + args.jev_key_file.read_text(encoding="utf-8") + if args.jev_key_file + else os.environ.get("POWERCONTEXT_EVAL_JEV_API_KEY") or os.environ.get("OPENROUTER_API_KEY", "") + ).strip() + except (OSError, UnicodeError): + parser.error("cannot read --jev-key-file") + if not key: + parser.error("configure POWERCONTEXT_EVAL_JEV_API_KEY, OPENROUTER_API_KEY, or --jev-key-file") + args.jev_key = SecretStr(key) + + +async def _evaluate_remote(args, calibration, held_out, token): + async with PowerContextClient(args.server_url, token=token) as client: + return await _evaluate(calibration, held_out, AuthorizedClientResolver(client), args) + + +async def _evaluate(calibration, held_out, resolver, args): + if args.backend == "jev": + return await _evaluate_jev(calibration, held_out, resolver, args) + if calibration.public_source is None: + resolver = SplitCheckedResolver(calibration, held_out, resolver) + calibration_results = await evaluate_dataset(calibration, resolver) + if args.calibration_only: + return calibration_results, (), None + held_out_results = await evaluate_dataset(held_out, resolver) + return calibration_results, held_out_results, args.configuration + + +async def _evaluate_jev(calibration, held_out, resolver, args): + async with httpx.AsyncClient() as client: + model = PublicJevModel(client, args.jev_key) + args.jev_policy_id = model.policy_id + cached = None + if args.calibration_report: + cached = load_calibration_results( + args.calibration_report, + calibration, + dataset_hashes=args.dataset_hashes, + code_digest=_code_digest(), + backend_policy_id=model.policy_id, + model_id=JEV_MODEL, + ) + args.output.parent.mkdir(parents=True, exist_ok=True) + with args.output.with_suffix(".cases.jsonl").open("x", encoding="utf-8") as checkpoint: + + async def evaluate(dataset): + count = 0 + + def save(row): + nonlocal count + count += 1 + checkpoint.write( + json.dumps({"split": dataset.split, **case_record(row)}, ensure_ascii=False) + "\n" + ) + checkpoint.flush() + if count % 25 == 0 or count == len(dataset.cases): + print( + f"{dataset.split}: {count}/{len(dataset.cases)} records saved", file=sys.stderr, flush=True + ) + + return await evaluate_public_dataset(dataset, resolver, model=model, on_result=save) + + calibration_results = cached if cached is not None else await evaluate(calibration) + if args.calibration_only: + return calibration_results, (), None + held_out_results = await evaluate(held_out) + return calibration_results, held_out_results, args.configuration + + +def _commit() -> str: + return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() # noqa: S607 + + +def _code_identity() -> dict[str, str]: + root = Path(__file__).resolve().parents[3] + paths = ( + *( + Path("evaluation/memory/write_gate") / name + for name in ( + "artifacts.py", + "cli.py", + "dataset.py", + "jev.py", + "prepare_public.py", + "report.py", + "resolver.py", + "runner.py", + ) + ), + Path("examples/systemone/adapter.py"), + Path("examples/systemone/laya.py"), + Path("src/powercontext/builtin/runtime/memory_write_gate.py"), + Path("src/powercontext/builtin/runtime/decision_model.py"), + Path("src/powercontext/builtin/runtime/decision_policy.py"), + Path("src/powercontext/builtin/artifacts/memory/protocols.py"), + Path("src/powercontext/builtin/inference/models.py"), + Path("src/powercontext/builtin/inference/errors.py"), + Path("src/powercontext/builtin/decision_observations.py"), + ) + return {path.as_posix(): sha256((root / path).read_bytes()).hexdigest() for path in paths} + + +def _code_digest() -> str: + """Bind saved measurements to the evaluator and its assessment contracts.""" + + return sha256(json.dumps(_code_identity(), sort_keys=True, separators=(",", ":")).encode()).hexdigest() diff --git a/evaluation/memory/write_gate/data/xnli/preparation.json b/evaluation/memory/write_gate/data/xnli/preparation.json new file mode 100644 index 0000000000..5dc6b7c18f --- /dev/null +++ b/evaluation/memory/write_gate/data/xnli/preparation.json @@ -0,0 +1,57 @@ +{ + "preparation_version": "xnli-reference-v1", + "source": { + "dataset": "xnli-1.0", + "archive_sha256": "4ba1d5e1afdb7161f0f23c66dc787802ccfa8a25a3ddd3b165a35e50df346ab1", + "source_url": "https://dl.fbaipublicfiles.com/XNLI/XNLI-1.0.zip", + "license": "CC-BY-NC-4.0", + "preparation_version": "xnli-reference-v1" + }, + "manifest_hashes": { + "calibration": "d52407d61be9847c568dce63eaeba778ed41b51e81dd1904f84df5dc0c39aad5", + "held_out": "ea3735a786c9e5e17be4f7fed2c5602f71e608bd0244987ba6275cef00bcd0ab" + }, + "excluded_groups": { + "within_split_repeated_text": 304 + }, + "splits": { + "calibration": { + "cases": 600, + "source_pairs": 300, + "labels_by_language": { + "en": { + "neutral": 100, + "contradiction": 100, + "entailment": 100 + }, + "zh": { + "neutral": 100, + "contradiction": 100, + "entailment": 100 + } + } + }, + "held_out": { + "cases": 1800, + "source_pairs": 900, + "labels_by_language": { + "en": { + "contradiction": 300, + "entailment": 300, + "neutral": 300 + }, + "zh": { + "contradiction": 300, + "entailment": 300, + "neutral": 300 + } + } + } + }, + "measurement_status": "not_run", + "limitations": [ + "sentence_pair_benchmark", + "training_overlap_unknown", + "not_production_write_distribution" + ] +} diff --git a/evaluation/memory/write_gate/dataset.py b/evaluation/memory/write_gate/dataset.py new file mode 100644 index 0000000000..e831065de7 --- /dev/null +++ b/evaluation/memory/write_gate/dataset.py @@ -0,0 +1,178 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Versioned, reference-only labels for Memory write-gate evaluation.""" + +from __future__ import annotations + +from enum import StrEnum +from hashlib import sha256 +from pathlib import Path +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class Label(StrEnum): + ACCEPT = "accept" + HOLD = "hold" + + +class PublicDatasetRef(_StrictModel): + kind: Literal["public_dataset"] = "public_dataset" + dataset: Literal["xnli-1.0"] = "xnli-1.0" + archive_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + split: Literal["dev", "test"] + pair_id: str = Field(min_length=1, max_length=256) + language: Literal["en", "zh"] + field: Literal["premise", "hypothesis"] + content_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + + +class PublicDatasetSource(_StrictModel): + dataset: Literal["xnli-1.0"] = "xnli-1.0" + archive_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + source_url: Literal["https://dl.fbaipublicfiles.com/XNLI/XNLI-1.0.zip"] = ( + "https://dl.fbaipublicfiles.com/XNLI/XNLI-1.0.zip" + ) + license: Literal["CC-BY-NC-4.0"] = "CC-BY-NC-4.0" + preparation_version: Literal["xnli-reference-v1"] = "xnli-reference-v1" + + +class SourceRef(_StrictModel): + kind: Literal["source"] = "source" + source_type: Literal["content"] = "content" + source_id: str = Field(min_length=1) + content_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + gate_input_digest: str | None = Field(default=None, pattern=r"^sha256:[0-9a-f]{64}$") + + +class ArtifactRef(_StrictModel): + kind: Literal["artifact"] = "artifact" + family: Literal["atomic-memory"] = "atomic-memory" + artifact_id: str = Field(min_length=1) + revision: int = Field(ge=1) + content_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + + +Reference = Annotated[PublicDatasetRef | SourceRef | ArtifactRef, Field(discriminator="kind")] + + +class Case(_StrictModel): + case_id: str = Field(min_length=1, max_length=128, pattern=r"^[a-z0-9][a-z0-9_-]*$") + group_id: str = Field(min_length=1, max_length=128) + scope_id: str = Field(min_length=1, max_length=256) + language: Literal["en", "zh"] + label: Label + label_reason: str = Field(min_length=1, max_length=128, pattern=r"^[a-z][a-z0-9_]*$") + original_label: Literal["entailment", "contradiction", "neutral"] | None = None + subject: Reference + evidence: tuple[Reference, ...] = Field(min_length=1) + + @model_validator(mode="after") + def validate_label_mapping(self) -> Case: + if isinstance(self.subject, PublicDatasetRef): + expected = Label.ACCEPT if self.original_label == "entailment" else Label.HOLD + if self.original_label is None or self.label != expected or self.label_reason != self.original_label: + raise ValueError("public dataset labels must preserve the declared NLI mapping") # noqa: TRY003 + return self + + +class Dataset(_StrictModel): + schema_version: Literal[1] = 1 + dataset_id: str = Field(min_length=1, max_length=128) + split: Literal["calibration", "held_out"] + public_source: PublicDatasetSource | None = None + cases: tuple[Case, ...] = Field(min_length=1) + + @model_validator(mode="after") + def validate_cases(self) -> Dataset: + ids = [case.case_id for case in self.cases] + if len(ids) != len(set(ids)): + raise ValueError("case IDs must be unique") # noqa: TRY003 + public = [ + ref for case in self.cases for ref in (case.subject, *case.evidence) if isinstance(ref, PublicDatasetRef) + ] + if bool(self.public_source) != bool(public) or ( + public and len(public) != sum(1 + len(case.evidence) for case in self.cases) + ): + raise ValueError("public manifests cannot mix private references") # noqa: TRY003 + if public: + expected_split = "dev" if self.split == "calibration" else "test" + if self.public_source is None or any( + ref.archive_sha256 != self.public_source.archive_sha256 or ref.split != expected_split for ref in public + ): + raise ValueError("public references must match the pinned source and original partition") # noqa: TRY003 + return self + + +def load_dataset(path: Path) -> tuple[Dataset, str]: + raw = path.read_bytes() + return Dataset.model_validate_json(raw), sha256(raw).hexdigest() + + +def _private_reference_identity(ref: Reference) -> tuple[str | int, ...]: + if isinstance(ref, ArtifactRef): + return ref.kind, ref.family, ref.artifact_id, ref.revision + if isinstance(ref, SourceRef): + return ref.kind, ref.source_type, ref.source_id, ref.content_digest + raise ValueError("private inputs cannot contain public references") # noqa: TRY003 + + +def _validate_private_input_overlap(calibration: Dataset, held_out: Dataset) -> None: + reference_keys = [ + { + ( + case.scope_id, + _private_reference_identity(case.subject), + tuple(sorted(_private_reference_identity(ref) for ref in case.evidence)), + ) + for case in dataset.cases + } + for dataset in (calibration, held_out) + ] + # Digests also catch copied content under new identities or another Scope. + # Evidence order and caller-assigned labels do not make a new input. + content_keys = [ + { + (case.subject.content_digest, tuple(sorted(ref.content_digest for ref in case.evidence))) + for case in dataset.cases + } + for dataset in (calibration, held_out) + ] + if reference_keys[0] & reference_keys[1] or content_keys[0] & content_keys[1]: + raise ValueError("private inputs must not cross splits") # noqa: TRY003 + + +def validate_split_pair(calibration: Dataset, held_out: Dataset) -> None: + if calibration.split != "calibration" or held_out.split != "held_out": + raise ValueError("expected calibration and held-out datasets") # noqa: TRY003 + if calibration.dataset_id != held_out.dataset_id: + raise ValueError("dataset IDs must match") # noqa: TRY003 + if calibration.public_source != held_out.public_source: + raise ValueError("dataset sources must match") # noqa: TRY003 + if {case.case_id for case in calibration.cases} & {case.case_id for case in held_out.cases}: + raise ValueError("case IDs must not cross splits") # noqa: TRY003 + if {case.group_id for case in calibration.cases} & {case.group_id for case in held_out.cases}: + raise ValueError("scenario groups must not cross splits") # noqa: TRY003 + if calibration.public_source is None: + _validate_private_input_overlap(calibration, held_out) + for dataset in (calibration, held_out): + for language in ("en", "zh"): + if {case.label for case in dataset.cases if case.language == language} != {Label.ACCEPT, Label.HOLD}: + raise ValueError(f"{dataset.split} needs accept and hold cases in {language}") # noqa: TRY003 diff --git a/evaluation/memory/write_gate/jev.py b/evaluation/memory/write_gate/jev.py new file mode 100644 index 0000000000..2e9aaf27c9 --- /dev/null +++ b/evaluation/memory/write_gate/jev.py @@ -0,0 +1,93 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Pinned Jev transport for public benchmark inputs, never a local runtime backend.""" + +from __future__ import annotations + +import math +from typing import Any + +import httpx +from pydantic import SecretStr +from typing_extensions import override + +from examples.systemone.adapter import SystemOneConfig, SystemOneDecisionModel +from powercontext.builtin.inference import ( + InferenceConfigurationError, + InferenceTimeoutError, + InferenceUnavailableError, + InvalidInferenceOutputError, +) +from powercontext.builtin.runtime.decision_model import DecisionRequest, DecisionResult + +JEV_MODEL = "typesafe/jev-1.13" +JEV_ENDPOINT = "https://openrouter.ai/api/alpha/decisions" + + +class PublicJevModel(SystemOneDecisionModel): + """Reuse the example's choice protocol and expose bounded measurement facts. + + A versioned provider ID does not guarantee immutable server-side weights. + Missing/invalid provider cost remains unknown; raw responses are not retained. + """ + + is_local_only = False + provider_id = "openrouter-typesafe" + backend_model_id = JEV_MODEL + + def __init__(self, client: httpx.AsyncClient, api_key: SecretStr) -> None: + super().__init__( + SystemOneConfig(provider="jev", endpoint=JEV_ENDPOINT, model=JEV_MODEL, api_key=api_key), client + ) + self.last_cost_usd: float | None = None + self.failure_reason: str | None = None + self._http_status: int | None = None + client.event_hooks["response"].append(self._record_status) + + async def _record_status(self, response: httpx.Response) -> None: + self._http_status = response.status_code + + @override + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + self.last_cost_usd = None + self.failure_reason = None + self._http_status = None + try: + return await super().evaluate(request) + except InferenceTimeoutError: + self.failure_reason = "backend_timeout" + raise + except InferenceUnavailableError: + self.failure_reason = { + 401: "credential_rejected", + 402: "insufficient_credit", + 403: "credential_rejected", + 429: "rate_limited", + }.get(self._http_status, "backend_unavailable") + raise + except InvalidInferenceOutputError: + self.failure_reason = "invalid_backend_output" + raise + except InferenceConfigurationError: + self.failure_reason = "backend_input_rejected" + raise + + @override + def _result(self, payload: Any) -> DecisionResult: + result = super()._result(payload) + cost = payload.get("usage", {}).get("cost") + if type(cost) in (int, float) and math.isfinite(cost) and cost >= 0: + self.last_cost_usd = float(cost) + return result diff --git a/evaluation/memory/write_gate/prepare_public.py b/evaluation/memory/write_gate/prepare_public.py new file mode 100644 index 0000000000..ad989b4353 --- /dev/null +++ b/evaluation/memory/write_gate/prepare_public.py @@ -0,0 +1,225 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Prepare reference-only bilingual XNLI manifests from the official external archive.""" + +from __future__ import annotations + +import argparse +import json +from collections import Counter +from dataclasses import dataclass +from hashlib import sha256 +from io import TextIOWrapper +from pathlib import Path +from typing import Any +from zipfile import ZipFile + +from .dataset import Case, Dataset, Label, PublicDatasetSource, validate_split_pair + +LABELS = ("entailment", "contradiction", "neutral") +LANGUAGES = ("en", "zh") +ARCHIVE_SHA256 = "4ba1d5e1afdb7161f0f23c66dc787802ccfa8a25a3ddd3b165a35e50df346ab1" + + +@dataclass(frozen=True, slots=True) +class NliPair: + pair_id: str + language: str + label: str + premise: str + hypothesis: str + + +def text_digest(text: str) -> str: + return "sha256:" + sha256(text.encode("utf-8")).hexdigest() + + +def load_xnli(path: Path) -> tuple[dict[str, dict[tuple[str, str], NliPair]], str]: + """Read only the named official members; no archive extraction or model calls.""" + + digest = sha256(path.read_bytes()).hexdigest() + splits: dict[str, dict[tuple[str, str], NliPair]] = {} + with ZipFile(path) as archive: + for split in ("dev", "test"): + info = archive.getinfo(f"XNLI-1.0/xnli.{split}.jsonl") + if info.file_size > 100_000_000: + raise ValueError("XNLI member exceeds the supported input size") # noqa: TRY003 + rows: dict[tuple[str, str], NliPair] = {} + with archive.open(info) as binary, TextIOWrapper(binary, encoding="utf-8") as handle: + for line in handle: + row = json.loads(line) + language = row.get("language") + if language not in LANGUAGES: + continue + values = (row.get("pairID"), row.get("gold_label"), row.get("sentence1"), row.get("sentence2")) + if not all(isinstance(value, str) and value for value in values): + raise ValueError("XNLI row has incomplete identity, label, or text") # noqa: TRY003 + pair = NliPair(values[0], language, values[1], values[2], values[3]) + key = (pair.pair_id, pair.language) + if key in rows: + raise ValueError("XNLI contains a duplicate pair/language identity") # noqa: TRY003 + rows[key] = pair + splits[split] = rows + return splits, digest + + +def _groups(rows: dict[tuple[str, str], NliPair], excluded: Counter[str]) -> list[tuple[NliPair, NliPair]]: + result = [] + for pair_id in sorted({key[0] for key in rows}): + english = rows.get((pair_id, "en")) + chinese = rows.get((pair_id, "zh")) + if english is None or chinese is None: + excluded["missing_translation"] += 1 + elif english.label not in LABELS or chinese.label != english.label: + excluded["invalid_or_inconsistent_label"] += 1 + else: + result.append((english, chinese)) + return sorted(result, key=lambda group: sha256(group[0].pair_id.encode()).hexdigest()) + + +def _identities(group: tuple[NliPair, NliPair]) -> set[str]: + return {text_digest(text) for pair in group for text in (pair.premise, pair.hypothesis)} + + +def _select( + groups: list[tuple[NliPair, NliPair]], + count: int, + blocked_text: set[str], + blocked_ids: set[str], + excluded: Counter[str], +) -> list[tuple[NliPair, NliPair]]: + selected: list[tuple[NliPair, NliPair]] = [] + counts: Counter[str] = Counter() + selected_text: set[str] = set() + for group in groups: + if group[0].pair_id in blocked_ids or _identities(group) & blocked_text: + excluded["cross_split_identity_or_text"] += 1 + continue + if counts[group[0].label] < count: + if _identities(group) & selected_text: + excluded["within_split_repeated_text"] += 1 + continue + selected.append(group) + counts[group[0].label] += 1 + selected_text.update(_identities(group)) + if any(counts[label] != count for label in LABELS): + raise ValueError("not enough eligible bilingual groups for the declared sample") # noqa: TRY003 + return selected + + +def _dataset(groups: list[tuple[NliPair, NliPair]], split: str, digest: str) -> Dataset: + cases = [] + for group in groups: + group_id = "xnli-" + sha256(group[0].premise.encode()).hexdigest() + for pair in group: + reference = { + "archive_sha256": digest, + "split": split, + "pair_id": pair.pair_id, + "language": pair.language, + } + cases.append( + Case.model_validate({ + "case_id": f"xnli-{split}-{sha256(pair.pair_id.encode()).hexdigest()[:24]}-{pair.language}", + "group_id": group_id, + "scope_id": "public-xnli", + "language": pair.language, + "label": Label.ACCEPT if pair.label == "entailment" else Label.HOLD, + "label_reason": pair.label, + "original_label": pair.label, + "subject": { + **reference, + "kind": "public_dataset", + "field": "hypothesis", + "content_digest": text_digest(pair.hypothesis), + }, + "evidence": [ + { + **reference, + "kind": "public_dataset", + "field": "premise", + "content_digest": text_digest(pair.premise), + } + ], + }) + ) + return Dataset( + dataset_id="memory-write-gate-xnli-v1", + split="calibration" if split == "dev" else "held_out", + public_source=PublicDatasetSource(archive_sha256=digest), + cases=tuple(cases), + ) + + +def prepare(path: Path, output_dir: Path) -> dict[str, Any]: + """Use fixed dev/test partitions and stable, label-balanced group selection.""" + + rows, digest = load_xnli(path) + if digest != ARCHIVE_SHA256: + raise ValueError("archive differs from the pinned official XNLI 1.0 download") # noqa: TRY003 + excluded: Counter[str] = Counter() + calibration_groups = _select(_groups(rows["dev"], excluded), 100, set(), set(), excluded) + blocked_text = set().union(*(_identities(group) for group in calibration_groups)) + held_groups = _select( + _groups(rows["test"], excluded), + 300, + blocked_text, + {group[0].pair_id for group in calibration_groups}, + excluded, + ) + calibration = _dataset(calibration_groups, "dev", digest) + held_out = _dataset(held_groups, "test", digest) + validate_split_pair(calibration, held_out) + output_dir.mkdir(parents=True, exist_ok=False) + manifest_hashes = {} + for dataset in (calibration, held_out): + raw = dataset.model_dump_json(indent=2) + "\n" + (output_dir / f"{dataset.split}.json").write_text(raw, encoding="utf-8") + manifest_hashes[dataset.split] = sha256(raw.encode()).hexdigest() + summary = { + "preparation_version": "xnli-reference-v1", + "source": PublicDatasetSource(archive_sha256=digest).model_dump(mode="json"), + "manifest_hashes": manifest_hashes, + "excluded_groups": dict(sorted(excluded.items())), + "splits": { + dataset.split: { + "cases": len(dataset.cases), + "source_pairs": len(dataset.cases) // 2, + "labels_by_language": { + language: dict(Counter(case.original_label for case in dataset.cases if case.language == language)) + for language in LANGUAGES + }, + } + for dataset in (calibration, held_out) + }, + "measurement_status": "not_run", + "limitations": ["sentence_pair_benchmark", "training_overlap_unknown", "not_production_write_distribution"], + } + (output_dir / "preparation.json").write_text( + json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + return summary + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--archive", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + args = parser.parse_args() + print(json.dumps(prepare(args.archive, args.output_dir), ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/evaluation/memory/write_gate/report.py b/evaluation/memory/write_gate/report.py new file mode 100644 index 0000000000..f5acbf0dd0 --- /dev/null +++ b/evaluation/memory/write_gate/report.py @@ -0,0 +1,290 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Content-free calibration and held-out measurements.""" + +from __future__ import annotations + +from collections import Counter +from math import sqrt +from statistics import median +from typing import Any + +from powercontext.builtin.artifacts.memory.protocols import MemoryWriteVerdict +from powercontext.builtin.runtime.decision_model import DECISION_INSTRUCTIONS_VERSION, DecisionOutcome +from powercontext.builtin.runtime.memory_write_gate import memory_write_policy + +from .runner import ( + HOLD_DIRECTIONS, + THRESHOLDS, + CaseResult, + EvaluationStatus, + configuration_key, +) + + +def _percentile(values: list[float], percentile: float) -> float | None: + if not values: + return None + ordered = sorted(values) + rank = (len(ordered) - 1) * percentile + lower = int(rank) + upper = min(lower + 1, len(ordered) - 1) + return ordered[lower] + (ordered[upper] - ordered[lower]) * (rank - lower) + + +def _rate(count: int, total: int) -> float | None: + return None if total == 0 else count / total + + +def _interval(count: int, rows: tuple[CaseResult, ...]) -> dict[str, float] | None: + if not rows or len({row.group_id for row in rows}) != len(rows): + return None + n = len(rows) + p = count / n + z = 1.959963984540054 + divisor = 1 + z * z / n + center = (p + z * z / (2 * n)) / divisor + radius = z * sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / divisor + return {"lower": max(0.0, center - radius), "upper": min(1.0, center + radius)} + + +def _cost(rows: tuple[CaseResult, ...]) -> dict[str, Any]: + evaluated = [ + row for row in rows if row.status in {EvaluationStatus.EVALUATED, EvaluationStatus.EVALUATION_FALLBACK} + ] + latencies = [row.latency_ms for row in evaluated if row.latency_ms is not None] + attempts = sum(row.outcome is not None for row in evaluated) + billed = [row for row in evaluated if row.outcome is not None] + costs = [row.currency_cost_usd for row in billed if row.currency_cost_usd is not None] + return { + "backend_attempts": attempts, + "model_requests": sum(row.requests for row in evaluated), + "input_tokens": 0 + if attempts == 0 + else None + if any(row.input_tokens is None for row in evaluated) + else sum(row.input_tokens or 0 for row in evaluated), + "output_tokens": 0 + if attempts == 0 + else None + if any(row.output_tokens is None for row in evaluated) + else sum(row.output_tokens or 0 for row in evaluated), + "latency_samples": len(latencies), + "latency_p50_ms": median(latencies) if latencies else None, + "latency_p95_ms": _percentile(latencies, 0.95), + "currency_cost": sum(costs) if billed and len(costs) == len(billed) else None, + "currency": "USD" if costs else None, + "currency_cost_source": "provider_reported" if costs else None, + "currency_cost_known_subtotal": sum(costs) if costs else None, + "currency_cost_known_cases": len(costs), + "human_review_minutes": None, + } + + +def score(rows: tuple[CaseResult, ...], configuration: str) -> dict[str, Any]: + """Keep missing inputs and non-adjudication out of classifier-error denominators.""" + + statuses = Counter(row.status.value for row in rows) + complete = tuple( + row for row in rows if row.status in {EvaluationStatus.EVALUATED, EvaluationStatus.EVALUATION_FALLBACK} + ) + judged = tuple(row for row in rows if row.status is EvaluationStatus.EVALUATED) + expected_accept = tuple(row for row in judged if row.label == "accept") + expected_hold = tuple(row for row in judged if row.label == "hold") + false_holds = sum(row.actions[configuration] is MemoryWriteVerdict.HOLD for row in expected_accept if row.actions) + false_passes = sum( + row.actions[configuration] is not MemoryWriteVerdict.HOLD for row in expected_hold if row.actions + ) + all_expected_hold = tuple(row for row in complete if row.label == "hold") + effective_false_passes = sum( + row.actions[configuration] is not MemoryWriteVerdict.HOLD for row in all_expected_hold if row.actions + ) + holds = sum(row.actions[configuration] is MemoryWriteVerdict.HOLD for row in complete if row.actions) + flags = sum(row.actions[configuration] is MemoryWriteVerdict.FLAG for row in complete if row.actions) + fallback = Counter( + row.fallback_reason or "unknown" for row in rows if row.status is EvaluationStatus.EVALUATION_FALLBACK + ) + return { + "cases": len(rows), + "status_counts": dict(sorted(statuses.items())), + "input_coverage": _rate(len(complete), len(rows)), + "adjudication_coverage": _rate(len(judged), len(rows)), + "judged": len(judged), + "false_holds": false_holds, + "false_hold_denominator": len(expected_accept), + "false_hold_rate": _rate(false_holds, len(expected_accept)), + "false_hold_wilson_95": _interval(false_holds, expected_accept), + "false_passes": false_passes, + "false_pass_denominator": len(expected_hold), + "false_pass_rate": _rate(false_passes, len(expected_hold)), + "false_pass_wilson_95": _interval(false_passes, expected_hold), + "false_passes_by_original_label": dict( + Counter( + row.original_label + for row in expected_hold + if row.original_label is not None + and row.actions + and row.actions[configuration] is not MemoryWriteVerdict.HOLD + ) + ), + "effective_false_passes_including_fallback": effective_false_passes, + "effective_false_pass_denominator": len(all_expected_hold), + "fallback_by_reason": dict(sorted(fallback.items())), + "hold_count": holds, + "flag_count": flags, + "action_denominator": len(complete), + "hold_rate_in_sample": _rate(holds, len(complete)), + "flag_rate_in_sample": _rate(flags, len(complete)), + "missing_confidence": sum(row.confidence is None for row in judged if row.outcome is not None), + "cost": _cost(rows), + } + + +def configuration_grid(rows: tuple[CaseResult, ...]) -> dict[str, dict[str, Any]]: + return { + configuration_key(direction, threshold): score(rows, configuration_key(direction, threshold)) + for direction in HOLD_DIRECTIONS + for threshold in THRESHOLDS + } + + +def calibration_report( + calibration: tuple[CaseResult, ...], + *, + dataset_id: str, + dataset_hashes: dict[str, str], + code_commit: str, + evaluation_code_digest: str | None = None, + evaluation_code_files: dict[str, str] | None = None, + execution: str, + backend_model_id: str | None = None, + backend_model_revision: str | None = None, + input_source: str = "unspecified", +) -> dict[str, Any]: + """Report calibration without opening or scoring held-out outcomes.""" + + return { + "schema_version": 1, + "phase": "calibration_only", + "dataset_id": dataset_id, + "dataset_hashes": dataset_hashes, + "code_commit": code_commit, + "evaluation_code_digest": evaluation_code_digest, + "evaluation_code_files": evaluation_code_files, + "execution": execution, + "input_source": input_source, + "backend_model_id": backend_model_id, + "backend_model_revision": backend_model_revision, + "decision_instructions_version": DECISION_INSTRUCTIONS_VERSION, + "policy_id": "memory.write.evidence_sufficiency.v1", + "calibration_grid": configuration_grid(calibration), + "calibration_by_language": { + language: configuration_grid( + tuple(row for row in calibration if row.language == language), + ) + for language in ("en", "zh") + }, + "cases": [case_record(row) for row in calibration], + } + + +def report( + calibration: tuple[CaseResult, ...], + held_out: tuple[CaseResult, ...], + *, + dataset_id: str, + dataset_hashes: dict[str, str], + code_commit: str, + evaluation_code_digest: str | None = None, + evaluation_code_files: dict[str, str] | None = None, + execution: str, + backend_model_id: str | None = None, + backend_model_revision: str | None = None, + input_source: str = "unspecified", + selected_configuration: str | None = None, +) -> dict[str, Any]: + if execution == "no_call" and selected_configuration is not None: + raise ValueError("no-call coverage audits cannot select a model configuration") # noqa: TRY003 + grid = configuration_grid(calibration) + if selected_configuration is not None and selected_configuration not in grid: + raise ValueError("selected configuration is outside the declared calibration grid") # noqa: TRY003 + if selected_configuration is not None and not grid[selected_configuration]["judged"]: + raise ValueError("cannot select a configuration without adjudicated calibration cases") # noqa: TRY003 + if execution != "no_call" and selected_configuration is None and any(result["judged"] for result in grid.values()): + raise ValueError("declare a calibration configuration before scoring held-out cases") # noqa: TRY003 + selected = selected_configuration + selected_policy_version = None + if selected is not None: + hold_part, threshold_part = selected.split(";") + hold_on = DecisionOutcome(hold_part.removeprefix("hold_on=")) + threshold_text = threshold_part.removeprefix("threshold=") + threshold = None if threshold_text == "none" else float(threshold_text) + selected_policy_version = memory_write_policy(hold_on=hold_on, threshold=threshold).version + scoring_key = selected or configuration_key(DecisionOutcome.YES, None) + return { + "schema_version": 1, + "phase": "input_coverage_audit" if execution == "no_call" else "held_out_evaluation", + "dataset_id": dataset_id, + "dataset_hashes": dataset_hashes, + "code_commit": code_commit, + "evaluation_code_digest": evaluation_code_digest, + "evaluation_code_files": evaluation_code_files, + "execution": execution, + "input_source": input_source, + "backend_model_id": backend_model_id, + "backend_model_revision": backend_model_revision, + "decision_instructions_version": DECISION_INSTRUCTIONS_VERSION, + "policy_id": "memory.write.evidence_sufficiency.v1", + "selected_configuration": selected, + "selected_policy_version": selected_policy_version, + "calibration_grid": grid, + "held_out": { + "overall": score(held_out, scoring_key), + "en": score(tuple(row for row in held_out if row.language == "en"), scoring_key), + "zh": score(tuple(row for row in held_out if row.language == "zh"), scoring_key), + }, + "cases": { + "calibration": [case_record(row) for row in calibration], + "held_out": [case_record(row) for row in held_out], + }, + "promotion_decision": "not_assessed", + "promotion_reason": ( + "no model inference was attempted; local preflight rules may adjudicate" + if execution == "no_call" + else "human hold/review friction and promotion budgets have not been measured" + ), + } + + +def case_record(row: CaseResult) -> dict[str, Any]: + return { + "case_id": row.case_id, + "group_id": row.group_id, + "language": row.language, + "label": row.label.value, + "original_label": row.original_label, + "status": row.status.value, + "coverage": None if row.coverage is None else row.coverage.value, + "used_fallback": row.used_fallback, + "outcome": None if row.outcome is None else row.outcome.value, + "confidence": row.confidence, + "fallback_reason": row.fallback_reason, + "latency_ms": row.latency_ms, + "requests": row.requests, + "input_tokens": row.input_tokens, + "output_tokens": row.output_tokens, + "currency_cost_usd": row.currency_cost_usd, + "actions": None if row.actions is None else {key: value.value for key, value in row.actions.items()}, + } diff --git a/evaluation/memory/write_gate/resolver.py b/evaluation/memory/write_gate/resolver.py new file mode 100644 index 0000000000..9723852000 --- /dev/null +++ b/evaluation/memory/write_gate/resolver.py @@ -0,0 +1,212 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Recover evaluation inputs through exact, Scope-authorized read paths.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from hashlib import sha256 +from pathlib import Path +from typing import Protocol + +from powercontext.client.client import PowerContextClient +from powercontext.client.errors import ( + ForbiddenResponseError, + ServerResponseError, + UnauthorizedResponseError, +) + +from .dataset import ArtifactRef, Case, Dataset, PublicDatasetRef, SourceRef, validate_split_pair +from .prepare_public import load_xnli, text_digest + + +class MissingInput(Exception): + """An exact input reference no longer resolves.""" + + +class Unauthorized(Exception): + """Current credentials cannot read an input in the selected Scope.""" + + +class UnsupportedReference(Exception): + """A reference cannot reconstruct the exact gate input.""" + + +@dataclass(frozen=True, slots=True) +class ResolvedCase: + candidate: str + evidence: tuple[str, ...] + + +class CaseResolver(Protocol): + async def resolve(self, case: Case) -> ResolvedCase: ... + + +class SplitCheckedResolver: + """Check cross-split gate inputs on each authorized read, retaining only hashes.""" + + def __init__(self, calibration: Dataset, held_out: Dataset, resolver: CaseResolver) -> None: + validate_split_pair(calibration, held_out) + self._resolver = resolver + self._splits = {case.case_id: dataset.split for dataset in (calibration, held_out) for case in dataset.cases} + self._inputs: dict[str, set[bytes]] = {"calibration": set(), "held_out": set()} + + async def resolve(self, case: Case) -> ResolvedCase: + split = self._splits[case.case_id] + resolved = await self._resolver.resolve(case) + # Record metadata and evidence order do not distinguish the evaluated input. + payload = json.dumps([resolved.candidate, sorted(resolved.evidence)], ensure_ascii=False, separators=(",", ":")) + digest = sha256(payload.encode("utf-8")).digest() + other_split = "held_out" if split == "calibration" else "calibration" + if digest in self._inputs[other_split]: + raise ValueError("private gate inputs must not cross splits") # noqa: TRY003 + self._inputs[split].add(digest) + return resolved + + +class PublicDatasetResolver: + """Resolve pinned public research inputs; no private Scope or credentials are used.""" + + def __init__(self, archive: Path) -> None: + self._rows, self.archive_sha256 = load_xnli(archive) + + async def resolve(self, case: Case) -> ResolvedCase: + subject = case.subject + if case.scope_id != "public-xnli" or not isinstance(subject, PublicDatasetRef) or len(case.evidence) != 1: + raise UnsupportedReference + evidence = case.evidence[0] + if not isinstance(evidence, PublicDatasetRef): + raise UnsupportedReference + if ( + subject.field != "hypothesis" + or evidence.field != "premise" + or subject.language != case.language + or evidence.language != subject.language + or evidence.pair_id != subject.pair_id + or evidence.split != subject.split + or subject.archive_sha256 != self.archive_sha256 + or evidence.archive_sha256 != self.archive_sha256 + ): + raise UnsupportedReference + pair = self._rows[subject.split].get((subject.pair_id, subject.language)) + if pair is None: + raise MissingInput + if ( + text_digest(pair.hypothesis) != subject.content_digest + or text_digest(pair.premise) != evidence.content_digest + or pair.label != case.original_label + ): + raise MissingInput + return ResolvedCase(candidate=pair.hypothesis, evidence=(pair.premise,)) + + +class AuthorizedClientResolver: + """Read an existing candidate revision and its evidence with the caller's Server token. + + Pending candidates have no immutable Artifact revision and are deliberately unsupported. + Every source and Artifact is read through the same Scope and checked against the + content digest pinned by the dataset. A changed or unavailable projection is not replayed. + """ + + def __init__(self, client: PowerContextClient) -> None: + self._client = client + + async def resolve(self, case: Case) -> ResolvedCase: + if not isinstance(case.subject, ArtifactRef): + raise UnsupportedReference + record = await self._artifact_record(case.scope_id, case.subject) + cited = ( + *(("source", ref.source_type, ref.source_id) for ref in record.sources), + *(("artifact", ref.family, ref.artifact_id, ref.revision) for ref in record.artifacts), + ) + supplied = [] + for ref in case.evidence: + if isinstance(ref, SourceRef): + supplied.append(("source", ref.source_type, ref.source_id)) + elif isinstance(ref, ArtifactRef): + supplied.append(("artifact", ref.family, ref.artifact_id, ref.revision)) + else: + raise UnsupportedReference + if cited != tuple(supplied): + raise UnsupportedReference + candidate = record.content.get("text") + if not isinstance(candidate, str): + raise UnsupportedReference + evidence: list[str] = [] + for ref in case.evidence: + if isinstance(ref, SourceRef): + evidence.append(await self._source_text(case.scope_id, ref)) + elif isinstance(ref, ArtifactRef): + evidence.append(await self._artifact_text(case.scope_id, ref)) + else: + raise UnsupportedReference + return ResolvedCase(candidate=candidate, evidence=tuple(evidence)) + + async def _artifact_record(self, scope_id: str, ref: ArtifactRef): + try: + record = await self._client.get_artifact_revision(scope_id, ref.family, ref.artifact_id, ref.revision) + except (ForbiddenResponseError, UnauthorizedResponseError) as error: + raise Unauthorized from error + except ServerResponseError as error: + if error.status_code == 404: + raise MissingInput from error + if error.status_code in {400, 422}: + raise UnsupportedReference from error + raise + if ( + record.scope_id != scope_id + or record.family != ref.family + or record.artifact_id != ref.artifact_id + or record.revision != ref.revision + or record.content_digest != ref.content_digest + ): + raise MissingInput + return record + + async def _artifact_text(self, scope_id: str, ref: ArtifactRef) -> str: + record = await self._artifact_record(scope_id, ref) + value = record.content.get("text") + if not isinstance(value, str): + raise UnsupportedReference + return value + + async def _source_text(self, scope_id: str, ref: SourceRef) -> str: + try: + record = await self._client.get_source(scope_id, ref.source_type, ref.source_id) + except (ForbiddenResponseError, UnauthorizedResponseError) as error: + raise Unauthorized from error + except ServerResponseError as error: + if error.status_code == 404: + raise MissingInput from error + if error.status_code in {400, 422}: + raise UnsupportedReference from error + raise + if ( + record.scope_id != scope_id + or record.source_type != ref.source_type + or record.source_id != ref.source_id + or record.content_digest != ref.content_digest + ): + raise MissingInput + # The public Source read may differ from the original gate projection. + # A digest captured from that exact projection proves when canonical content + # reproduces it; without the digest, the input remains incomplete. + if ref.gate_input_digest is None: + raise UnsupportedReference + text = json.dumps(record.content, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + if f"sha256:{sha256(text.encode('utf-8')).hexdigest()}" != ref.gate_input_digest: + raise UnsupportedReference + return text diff --git a/evaluation/memory/write_gate/results/jev-1.13/RESULTS.md b/evaluation/memory/write_gate/results/jev-1.13/RESULTS.md new file mode 100644 index 0000000000..8a37495033 --- /dev/null +++ b/evaluation/memory/write_gate/results/jev-1.13/RESULTS.md @@ -0,0 +1,76 @@ +# Jev 1.13 Memory 写入门实测结果 + +2026-10-14，通过 OpenRouter 实际调用 `typesafe/jev-1.13`。公开 XNLI 题集包含 600 条 calibration、 +1,800 条 held-out；固定配置为 `hold_on=yes;threshold=none`。 + +问题适配器版本为 `powercontext.decision.systemone.jev.v2:typesafe/jev-1.13`，策略版本为 +`1;hold_on=yes;threshold=none`。实测记录的评测代码摘要为 +`8edd5373032f3cca59c607445083ac6b85b9353e79b4ed621639852ecf4e1fe8`。 +该摘要标识调用时的实验实现，不表示当前代码重新执行过云端测量。 + +**建议：本次模型、问题和配置继续保持 shadow/advisory。** +独立题集误拦了 107 条有支持证据的候选，其中中文 82/299（27.42%）；包含未裁决后的放行， +另有 176/1200 条本应暂缓的候选被放行。这是具体配置的实测建议，不是官方统一及格线。 + +## 独立评估 + +| 语言 | 题目数 | 已裁决 | 误拦／应接受（已裁决） | 误放／应暂缓（已裁决） | 含未裁决的误放／应暂缓 | +| --- | ---: | ---: | --- | --- | --- | +| 英文 | 900 | 891 | 25/296（8.45%） | 84/595（14.12%） | 89/600 | +| 中文 | 900 | 889 | 82/299（27.42%） | 77/590（13.05%） | 87/600 | +| 总体 | 1800 | 1780 | 107/595（17.98%） | 161/1185（13.59%） | 176/1200 | + +输入恢复率 100%，实质判断覆盖率 98.89%。20 条未裁决分别为 10 条 abstain 和 10 条响应校验失败。 +它们不进入已裁决质量分母；涉及应暂缓题时，进入包含 fail-open 的放行效果。 +所有 `used_fallback=true` 记录保持 `coverage=unadjudicated`。 +中英文是对应题，不把总体当作 1,800 个独立场景。 + +最后一次尝试 P50/P95 调用耗时为 0.706/1.380 秒。正式 calibration + 最终 held-out 的已知费用小计为 +**USD 0.049940184**。13 条响应校验失败及早期连接失败缺少可靠 usage，完整费用和 token 总量未知。 +最终用量不包含旧失败尝试的未知费用，重试可能重复计费。六题 smoke 和诊断不参与正式成绩。 + +样本 HOLD 为 1131/1800（62.83%），FLAG 为 0。HOLD 不写入，FLAG 允许写入，按实际效果评分。 +这些比例不外推生产负担；人工处理时间和部署摩擦预算未测量。 + +## Calibration 比较与冻结 + +按已声明规则最小化误拦率与包含 fallback 的误放率的平均值；冻结配置后才调用 held-out。 + +| 配置 | 误拦率 | 误放率（含 fallback） | +| --- | ---: | ---: | +| `hold_on=yes;threshold=none` | 15.08% | 14.00% | +| `hold_on=yes;threshold=0.5` | 8.04% | 28.50% | +| `hold_on=yes;threshold=0.7` | 4.52% | 42.00% | +| `hold_on=yes;threshold=0.9` | 1.51% | 77.50% | +| `hold_on=no;threshold=none` | 84.92% | 86.75% | +| `hold_on=no;threshold=0.5` | 65.83% | 94.00% | +| `hold_on=no;threshold=0.7` | 54.27% | 95.75% | +| `hold_on=no;threshold=0.9` | 31.66% | 98.50% | + +## 初次故障与两轮补测 + +首轮仅 1074/1800 已裁决：711 条连接不可用、3 条超时、7 条响应校验失败、5 条 abstain。 +首轮含失败后的误放为 597/1200（49.75%）；这不是已回答题的模型误判率。 +新连接可用，故障原因未确定。先补测 714 条网络失败，再补测仍失败的 7 条。 +没有替换已裁决回答、主动 abstain 或格式异常结果，没有改模型、问题或配置。 +正式后端尝试为 600 + 1800 + 714 + 7 = 3121 次。 + +逐题最后一次耗时不能冒充全流程等待时间。 + +## 评估范围 + +本实验仅把公开句对发送到云端，随后本地复用 Runtime 规则，不验证正式 hosted-redacted 传输。 +XNLI 不证明生产 Memory 分布、多条证据、业务严重错误、真实历史引用覆盖或训练数据无重叠。 +私有授权和精确版本行为由 Server/SQLite 受控推理测试验证，不冒充生产部署验收。 +本实验不会修改部署模式。问题或模型发生变化后，需要重新校准，并使用未参与修改决策的独立题集。 + +## 回滚步骤 + +如果部署之后启用了 `enforcing`，需要撤回时： + +1. 将 `POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_MODE=shadow`，重启受影响的 Runtime 并核对生效配置。 +2. 执行一次有界写入探针，确认原本会被 gate 暂缓的写入正常完成，且 Scope 私有观察记录仍保存评估。 +3. 如需同时停止模型调用，设置 `POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_PRIVACY_BOUNDARY=no_external_call`。 + 如需完全关闭 gate，设置 `POWERCONTEXT_SERVER_RUNTIME_MEMORY_WRITE_GATE_ENABLED=false`；这也会停止观察记录。 +4. 比较变更前后的暂缓数量和 fallback 原因，不重写历史 Memory revision、citation 或观察记录。 + enforcing 期间已写入内容如需修正，使用正常的 Memory 审核与生命周期操作。 diff --git a/evaluation/memory/write_gate/runner.py b/evaluation/memory/write_gate/runner.py new file mode 100644 index 0000000000..036b7891a2 --- /dev/null +++ b/evaluation/memory/write_gate/runner.py @@ -0,0 +1,292 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Evaluate exact Memory write inputs without changing authoritative state.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, replace +from enum import StrEnum +from time import perf_counter + +from powercontext.builtin.artifacts.memory.protocols import MemoryWriteGateRequest, MemoryWriteVerdict +from powercontext.builtin.decision_observations import ( + DecisionCoverage, + DecisionObservation, + DecisionPolicyMode, + DecisionPrivacyBoundary, +) +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import ( + DecisionModel, + DecisionOutcome, + DecisionRequest, + DecisionResult, + FailOpenDecisionModel, +) +from powercontext.builtin.runtime.memory_write_gate import DecisionMemoryWriteGate, memory_write_policy + +from .dataset import Case, Dataset, Label +from .prepare_public import ARCHIVE_SHA256 +from .resolver import CaseResolver, MissingInput, PublicDatasetResolver, Unauthorized, UnsupportedReference + +THRESHOLDS: tuple[float | None, ...] = (None, 0.5, 0.7, 0.9) +HOLD_DIRECTIONS = (DecisionOutcome.YES, DecisionOutcome.NO) + + +class EvaluationStatus(StrEnum): + EVALUATED = "evaluated" + MISSING_INPUT = "missing_input" + UNAUTHORIZED = "unauthorized" + UNSUPPORTED_REFERENCE = "unsupported_reference" + EVALUATION_FALLBACK = "evaluation_fallback" + NOT_ATTEMPTED = "not_attempted" + + +@dataclass(slots=True) +class _Collector: + observation: DecisionObservation | None = None + + async def record(self, observation: DecisionObservation, /) -> None: + self.observation = observation + + +class _CapturingModel: + def __init__(self, delegate: DecisionModel) -> None: + self._delegate = FailOpenDecisionModel(delegate) + self.policy_id = self._delegate.policy_id + self.is_local_only = bool(getattr(self._delegate, "is_local_only", False)) + self.provider_id = getattr(self._delegate, "provider_id", None) + self.backend_model_id = getattr(self._delegate, "backend_model_id", None) + self.result: DecisionResult | None = None + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + self.result = await self._delegate.evaluate(request) + return self.result + + +@dataclass(frozen=True, slots=True) +class CaseResult: + case_id: str + group_id: str + language: str + label: Label + status: EvaluationStatus + original_label: str | None = None + coverage: DecisionCoverage | None = None + outcome: DecisionOutcome | None = None + confidence: float | None = None + used_fallback: bool = False + fallback_reason: str | None = None + latency_ms: float | None = None + requests: int = 0 + input_tokens: int | None = None + output_tokens: int | None = None + currency_cost_usd: float | None = None + actions: dict[str, MemoryWriteVerdict] | None = None + + +def configuration_key(hold_on: DecisionOutcome, threshold: float | None) -> str: + return f"hold_on={hold_on.value};threshold={'none' if threshold is None else format(threshold, '.1f')}" + + +async def evaluate_dataset( + dataset: Dataset, + resolver: CaseResolver, + *, + model: DecisionModel | None = None, +) -> tuple[CaseResult, ...]: + """Run each case once and apply candidate policies to its ephemeral model result.""" + + return tuple([await _evaluate_case(case, resolver, model=model) for case in dataset.cases]) + + +async def evaluate_public_dataset( + dataset: Dataset, + resolver: PublicDatasetResolver, + *, + model: DecisionModel, + on_result: Callable[[CaseResult], None] | None = None, +) -> tuple[CaseResult, ...]: + """Call a hosted model only on pinned public data, then rescore its answer locally. + + This measures a candidate backend; it does not exercise hosted Runtime egress. + Private references and arbitrary archives cannot enter this path. + """ + + if ( + not isinstance(resolver, PublicDatasetResolver) + or resolver.archive_sha256 != ARCHIVE_SHA256 + or dataset.public_source is None + or dataset.public_source.archive_sha256 != ARCHIVE_SHA256 + ): + raise ValueError("hosted evaluation requires the pinned official public archive and manifests") # noqa: TRY003 + results = [] + stop_reason = None + for case in dataset.cases: + if stop_reason: + result = replace(_incomplete(case, EvaluationStatus.NOT_ATTEMPTED), fallback_reason=stop_reason) + else: + result = await _evaluate_case(case, resolver, model=None, public_model=model) + if result.fallback_reason in {"insufficient_credit", "credential_rejected", "rate_limited"}: + stop_reason = result.fallback_reason + results.append(result) + if on_result is not None: + on_result(result) + return tuple(results) + + +async def _evaluate_case( + case: Case, + resolver: CaseResolver, + *, + model: DecisionModel | None, + public_model: DecisionModel | None = None, +) -> CaseResult: + try: + resolved = await resolver.resolve(case) + except MissingInput: + return _incomplete(case, EvaluationStatus.MISSING_INPUT) + except Unauthorized: + return _incomplete(case, EvaluationStatus.UNAUTHORIZED) + except UnsupportedReference: + return _incomplete(case, EvaluationStatus.UNSUPPORTED_REFERENCE) + + chosen_model = model + request_capture = _RequestCapture() if public_model is not None else None + if request_capture is not None: + chosen_model = request_capture + capturing = None if chosen_model is None else _CapturingModel(chosen_model) + privacy = DecisionPrivacyBoundary.LOCAL_ONLY if capturing is not None else DecisionPrivacyBoundary.NO_EXTERNAL_CALL + collector = _Collector() + request = MemoryWriteGateRequest( + candidates=(resolved.candidate,), + evidence=resolved.evidence, + scope_id=case.scope_id, + operation_id=f"offline-evaluation:{case.case_id}", + observation_sink=collector, + defer_observation_emission=True, + ) + gate = DecisionMemoryWriteGate( + capturing, + hold_on=DecisionOutcome.YES, + policy=memory_write_policy(mode=DecisionPolicyMode.ENFORCING, privacy_boundary=privacy), + ) + await gate.assess(request) + public_latency_ms = None + public_result = None + if request_capture is not None and request_capture.request is not None and public_model is not None: + started = perf_counter() + public_result = await FailOpenDecisionModel(public_model).evaluate(request_capture.request) + public_latency_ms = (perf_counter() - started) * 1000 + # Only an already obtained answer enters the gate. No hosted model is + # marked local or invoked by the Runtime's local_only path. + capturing = _CapturingModel(_ResultModel(public_result)) + await DecisionMemoryWriteGate(capturing).assess(request) + observation = collector.observation + if observation is None: + raise RuntimeError("write gate produced no observation") # noqa: TRY003 + assessment = observation.assessment + status = ( + EvaluationStatus.EVALUATED + if assessment.coverage is DecisionCoverage.ADJUDICATED + else EvaluationStatus.EVALUATION_FALLBACK + ) + result = None if capturing is None else capturing.result + actions: dict[str, MemoryWriteVerdict] = {} + rescore_request = replace(request, observation_sink=None) + for hold_on in HOLD_DIRECTIONS: + for threshold in THRESHOLDS: + mapped = DecisionMemoryWriteGate( + _ResultModel(result) if result is not None else None, + hold_on=hold_on, + threshold=threshold, + policy=memory_write_policy( + hold_on=hold_on, + threshold=threshold, + mode=DecisionPolicyMode.ENFORCING, + privacy_boundary=DecisionPrivacyBoundary.LOCAL_ONLY + if result is not None + else DecisionPrivacyBoundary.NO_EXTERNAL_CALL, + ), + ) + mapped_assessment = await mapped.assess(rescore_request) + actions[configuration_key(hold_on, threshold)] = mapped_assessment.verdict + return CaseResult( + case_id=case.case_id, + group_id=case.group_id, + language=case.language, + label=case.label, + status=status, + original_label=case.original_label, + coverage=assessment.coverage, + outcome=None if result is None else result.outcome, + confidence=None if result is None else result.confidence, + used_fallback=assessment.used_fallback, + fallback_reason=(getattr(public_model, "failure_reason", None) if public_result is not None else None) + or observation.fallback_reason + or (assessment.reason if status is EvaluationStatus.EVALUATION_FALLBACK else None), + latency_ms=public_latency_ms if public_latency_ms is not None else assessment.latency_ms, + requests=assessment.usage.requests, + input_tokens=assessment.usage.input_tokens, + output_tokens=assessment.usage.output_tokens, + currency_cost_usd=getattr(public_model, "last_cost_usd", None) if public_result is not None else None, + actions=actions, + ) + + +class _RequestCapture: + """Capture the gate's exact question/projection without performing inference. + + Runtime preflight can reject an input without calling this local recorder. + Its placeholder abstention is never scored as a hosted model response. + """ + + policy_id = "evaluation.capture.v1" + is_local_only = True + + def __init__(self) -> None: + self.request: DecisionRequest | None = None + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + self.request = request + return DecisionResult( + outcome=DecisionOutcome.ABSTAIN, + policy_id=self.policy_id, + usage=InferenceUsage(requests=0), + used_fallback=True, + ) + + +class _ResultModel: + policy_id = "evaluation.rescore.v1" + is_local_only = True + + def __init__(self, result: DecisionResult) -> None: + self._result = result + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + return self._result + + +def _incomplete(case: Case, status: EvaluationStatus) -> CaseResult: + return CaseResult( + case_id=case.case_id, + group_id=case.group_id, + language=case.language, + label=case.label, + status=status, + original_label=case.original_label, + ) diff --git a/tests/e2e/test_memory_write_gate_evaluation.py b/tests/e2e/test_memory_write_gate_evaluation.py new file mode 100644 index 0000000000..701fe38bef --- /dev/null +++ b/tests/e2e/test_memory_write_gate_evaluation.py @@ -0,0 +1,141 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The offline evaluator reads exact, authorized records through the real Server API.""" + +from __future__ import annotations + +import asyncio +import json +from hashlib import sha256 +from pathlib import Path + +import httpx +from pydantic import SecretStr + +from evaluation.memory.write_gate.dataset import ArtifactRef, Case, Dataset, Label, SourceRef +from evaluation.memory.write_gate.resolver import AuthorizedClientResolver +from evaluation.memory.write_gate.runner import EvaluationStatus, evaluate_dataset +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionRequest, DecisionResult +from powercontext.client import PowerContextClient +from powercontext.http import CreateArtifactRequest, ReplaceArtifactRequest +from powercontext.server.factory import create_server_app +from powercontext.server.settings import BearerAuthConfig, McpConfig, MetricsConfig, ServerSettings + + +class _AcceptingModel: + policy_id = "evaluation.live-read-test.v1" + is_local_only = True + + async def evaluate(self, request: DecisionRequest, /) -> DecisionResult: + return DecisionResult(outcome=DecisionOutcome.NO, policy_id=self.policy_id, usage=InferenceUsage(requests=1)) + + +def test_exact_source_projection_and_memory_revision_re_evaluate_through_server(tmp_path: Path) -> None: + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'write-gate-evaluation.db'}"), + auth=BearerAuthConfig(enabled=True, token=SecretStr("evaluation-test-token")), + mcp=McpConfig(enabled=False), + metrics=MetricsConfig(enabled=False), + ), + scheduler_path=tmp_path / "scheduler.db", + ) + + async def scenario() -> None: + async with ( + app.router.lifespan_context(app), + httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as transport, + ): + client = PowerContextClient( + "http://testserver", + token="evaluation-test-token", # noqa: S106 + http_client=transport, + trust_transport_security=True, + ) + scope_id = (await client.get_default_scope()).scope_id + created = await client.create_artifact( + scope_id, + CreateArtifactRequest.model_validate({ + "family": "atomic-memory", + "content": { + "schema": "powercontext.atomic-memory.v1", + "kind": "fact", + "text": "The API is available.", + }, + }), + ) + exact = await client.get_artifact_revision(scope_id, "atomic-memory", created.artifact_id, 1) + citation = created.sources[0] + source = await client.get_source(scope_id, citation.source_type, citation.source_id) + gate_input = json.dumps(source.content, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + case = Case( + case_id="server-read", + group_id="server-read", + scope_id=scope_id, + language="en", + label=Label.ACCEPT, + label_reason="direct_support", + subject=ArtifactRef( + artifact_id=created.artifact_id, + revision=1, + content_digest=exact.content_digest, + ), + evidence=( + SourceRef( + source_id=citation.source_id, + content_digest=source.content_digest, + gate_input_digest=f"sha256:{sha256(gate_input.encode()).hexdigest()}", + ), + ), + ) + dataset = Dataset(dataset_id="server-read", split="calibration", cases=(case,)) + resolver = AuthorizedClientResolver(client) + first = await evaluate_dataset(dataset, resolver, model=_AcceptingModel()) + assert first[0].status is EvaluationStatus.EVALUATED + assert first[0].outcome is DecisionOutcome.NO + + replaced = await client.replace_artifact( + scope_id, + "atomic-memory", + created.artifact_id, + ReplaceArtifactRequest.model_validate({ + "content": {"kind": "fact", "text": "The API is available on the new endpoint."} + }), + expected_etag='"revision:1"', + ) + assert replaced.revision == 2 + again = await evaluate_dataset(dataset, resolver, model=_AcceptingModel()) + assert again[0].status is EvaluationStatus.EVALUATED + + changed_digest = case.model_copy( + update={"evidence": (case.evidence[0].model_copy(update={"gate_input_digest": "sha256:" + "0" * 64}),)} + ) + unsupported = await evaluate_dataset( + dataset.model_copy(update={"cases": (changed_digest,)}), resolver, model=_AcceptingModel() + ) + assert unsupported[0].status is EvaluationStatus.UNSUPPORTED_REFERENCE + + unauthorized_client = PowerContextClient( + "http://testserver", + token="revoked-test-token", # noqa: S106 + http_client=transport, + trust_transport_security=True, + ) + denied = await evaluate_dataset(dataset, AuthorizedClientResolver(unauthorized_client)) + assert denied[0].status is EvaluationStatus.UNAUTHORIZED + + asyncio.run(scenario()) diff --git a/tests/evaluation/test_memory_write_gate_evaluation.py b/tests/evaluation/test_memory_write_gate_evaluation.py new file mode 100644 index 0000000000..f84a7a304c --- /dev/null +++ b/tests/evaluation/test_memory_write_gate_evaluation.py @@ -0,0 +1,601 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Observable offline evaluation behavior for Memory write-gate promotion evidence.""" + +from __future__ import annotations + +import asyncio +import json +from hashlib import sha256 +from pathlib import Path +from types import SimpleNamespace +from typing import cast + +import pytest +import rfc8785 + +from evaluation.memory.write_gate.dataset import ( + ArtifactRef, + Case, + Dataset, + Label, + SourceRef, + validate_split_pair, +) +from evaluation.memory.write_gate.report import calibration_report, report +from evaluation.memory.write_gate.resolver import ( + AuthorizedClientResolver, + MissingInput, + ResolvedCase, + SplitCheckedResolver, + Unauthorized, + UnsupportedReference, +) +from evaluation.memory.write_gate.runner import EvaluationStatus, evaluate_dataset +from powercontext.builtin.decision_observations import DecisionCoverage +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionResult +from powercontext.client.client import PowerContextClient +from powercontext.client.errors import ForbiddenResponseError + +DIGEST = "sha256:" + "a" * 64 + + +class _Resolver: + async def resolve(self, case: Case) -> ResolvedCase: + return ResolvedCase(candidate=f"Candidate {case.case_id}", evidence=("会议决定在周五发布。",)) + + +class _Model: + policy_id = "test.write-gate" + is_local_only = True + + def __init__(self, *, used_fallback: bool = False): + self.used_fallback = used_fallback + self.outcomes = iter(("no", "yes", "yes", "yes", "no", "yes", "no", "abstain")) + + async def evaluate(self, request): + return DecisionResult( + outcome=DecisionOutcome(next(self.outcomes)), + confidence=0.8, + policy_id=self.policy_id, + used_fallback=self.used_fallback, + usage=InferenceUsage(requests=1), + ) + + +def _datasets(): + datasets = [] + for split in ("calibration", "held_out"): + cases = tuple( + Case( + case_id=f"{split}-{index}", + group_id=f"{split}-{index}", + scope_id="test-scope", + language="en" if index < 4 else "zh", + label=Label.ACCEPT if index % 2 == 0 else Label.HOLD, + label_reason="test_support", + subject=ArtifactRef( + artifact_id=f"{split}-{index}", + revision=1, + content_digest="sha256:" + sha256(f"{split}-{index}".encode()).hexdigest(), + ), + evidence=(ArtifactRef(artifact_id="evidence", revision=1, content_digest=DIGEST),), + ) + for index in range(8) + ) + datasets.append(Dataset(dataset_id="test", split=split, cases=cases)) + validate_split_pair(*datasets) + return *datasets, {} + + +def test_calibration_and_held_out_report_fallback_separately() -> None: + calibration, held_out, hashes = _datasets() + resolver = _Resolver() + + async def scenario(): + return ( + await evaluate_dataset(calibration, resolver, model=_Model()), + await evaluate_dataset(held_out, resolver, model=_Model()), + ) + + calibration_results, held_results = asyncio.run(scenario()) + result = report( + calibration_results, + held_results, + dataset_id=calibration.dataset_id, + dataset_hashes=hashes, + code_commit="test", + execution="controlled_model", + selected_configuration="hold_on=yes;threshold=none", + ) + + assert result["selected_configuration"] == "hold_on=yes;threshold=none" + assert result["held_out"]["overall"]["status_counts"] == {"evaluated": 7, "evaluation_fallback": 1} + assert result["held_out"]["overall"]["false_holds"] == 1 + assert result["held_out"]["overall"]["effective_false_passes_including_fallback"] == 1 + assert result["held_out"]["overall"]["cost"]["model_requests"] == 8 + assert result["held_out"]["en"]["false_holds"] == 1 + assert result["held_out"]["zh"]["false_holds"] == 0 + assert result["promotion_decision"] == "not_assessed" + assert result["held_out"]["overall"]["hold_count"] == 4 + assert result["held_out"]["overall"]["action_denominator"] == 8 + serialized = json.dumps(result, ensure_ascii=False) + assert "Candidate held_out-0" not in serialized + assert "会议决定" not in serialized + + +def test_calibration_requires_an_explicit_configuration_before_held_out_scoring() -> None: + calibration, held_out, hashes = _datasets() + resolver = _Resolver() + + async def scenario(): + return ( + await evaluate_dataset(calibration, resolver, model=_Model()), + await evaluate_dataset(held_out, resolver, model=_Model()), + ) + + calibration_results, held_results = asyncio.run(scenario()) + only = calibration_report( + calibration_results, + dataset_id=calibration.dataset_id, + dataset_hashes=hashes, + code_commit="test", + execution="controlled_model", + ) + assert only["phase"] == "calibration_only" + assert "held_out" not in only + with pytest.raises(ValueError, match="declare a calibration configuration"): + report( + calibration_results, + held_results, + dataset_id=calibration.dataset_id, + dataset_hashes=hashes, + code_commit="test", + execution="controlled_model", + ) + + +def test_missing_input_is_not_scored_as_negative() -> None: + class MissingResolver: + async def resolve(self, case): + raise MissingInput + + calibration, _, _ = _datasets() + result = asyncio.run(evaluate_dataset(calibration, MissingResolver(), model=_Model())) + assert {row.status for row in result} == {EvaluationStatus.MISSING_INPUT} + assert all(row.actions is None for row in result) + + +def test_no_call_and_backend_fallback_are_not_counted_as_adjudication() -> None: + calibration, _, _ = _datasets() + resolver = _Resolver() + no_call = asyncio.run(evaluate_dataset(calibration, resolver)) + assert {row.status for row in no_call} == {EvaluationStatus.EVALUATION_FALLBACK} + assert all(row.coverage is DecisionCoverage.UNADJUDICATED and row.used_fallback for row in no_call) + no_call_report = report( + no_call, + no_call, + dataset_id=calibration.dataset_id, + dataset_hashes={}, + code_commit="test", + execution="no_call", + ) + assert no_call_report["selected_configuration"] is None + assert no_call_report["selected_policy_version"] is None + assert no_call_report["held_out"]["overall"]["adjudication_coverage"] == 0 + + backend_fallback = asyncio.run(evaluate_dataset(calibration, resolver, model=_Model(used_fallback=True))) + assert backend_fallback[0].status is EvaluationStatus.EVALUATION_FALLBACK + assert backend_fallback[0].coverage is DecisionCoverage.UNADJUDICATED + assert backend_fallback[0].used_fallback is True + + +def test_rejects_scenario_group_leakage_between_splits() -> None: + calibration, held_out, _ = _datasets() + copied = held_out.cases[0].model_copy(update={"group_id": calibration.cases[0].group_id}) + bad = held_out.model_copy(update={"cases": (copied, *held_out.cases[1:])}) + with pytest.raises(ValueError, match="scenario groups"): + validate_split_pair(calibration, bad) + + +def test_rejects_copied_private_inputs_with_distinct_case_and_group_ids() -> None: + calibration, held_out, _ = _datasets() + value = held_out.model_dump() + for case, original in zip(value["cases"], calibration.cases, strict=True): + case["subject"] = original.subject.model_dump() + case["evidence"] = tuple(ref.model_dump() for ref in original.evidence) + copied = Dataset.model_validate(value) + with pytest.raises(ValueError, match="private inputs must not cross splits"): + validate_split_pair(calibration, copied) + + +def test_rejects_duplicate_immutable_references_even_with_changed_artifact_digests() -> None: + calibration, held_out, _ = _datasets() + original = calibration.cases[0] + copied = held_out.cases[0].model_copy( + update={ + "subject": original.subject.model_copy(update={"content_digest": DIGEST}), + "evidence": tuple( + ref.model_copy(update={"content_digest": "sha256:" + "b" * 64}) for ref in original.evidence + ), + } + ) + duplicate = Dataset.model_validate({**held_out.model_dump(), "cases": (copied, *held_out.cases[1:])}) + with pytest.raises(ValueError, match="private inputs must not cross splits"): + validate_split_pair(calibration, duplicate) + + +@pytest.mark.parametrize("evidence_kind", ["artifact", "source"]) +def test_rejects_duplicate_content_under_different_references_and_evidence_order(evidence_kind: str) -> None: + calibration, held_out, _ = _datasets() + if evidence_kind == "source": + evidence = ( + SourceRef(source_id="source-a", content_digest=DIGEST, gate_input_digest=DIGEST), + SourceRef(source_id="source-b", content_digest="sha256:" + "b" * 64), + ) + renamed = tuple(ref.model_copy(update={"source_id": f"copy-{index}"}) for index, ref in enumerate(evidence)) + else: + evidence = ( + ArtifactRef(artifact_id="evidence-a", revision=1, content_digest=DIGEST), + ArtifactRef(artifact_id="evidence-b", revision=1, content_digest="sha256:" + "b" * 64), + ) + renamed = tuple( + ref.model_copy(update={"artifact_id": f"copy-{index}", "revision": 2}) for index, ref in enumerate(evidence) + ) + original = calibration.cases[0].model_copy(update={"evidence": evidence}) + calibration = calibration.model_copy(update={"cases": (original, *calibration.cases[1:])}) + copied = held_out.cases[0].model_copy( + update={ + "scope_id": "another-scope", + "subject": original.subject.model_copy(update={"artifact_id": "copied-memory", "revision": 2}), + "evidence": tuple(reversed(renamed)), + } + ) + held_out = held_out.model_copy(update={"cases": (copied, *held_out.cases[1:])}) + with pytest.raises(ValueError, match="private inputs must not cross splits"): + validate_split_pair(calibration, held_out) + + +def test_allows_shared_evidence_for_distinct_private_candidates() -> None: + calibration, held_out, _ = _datasets() + assert calibration.cases[0].evidence == held_out.cases[0].evidence + validate_split_pair(calibration, held_out) + + +def test_allows_same_private_candidate_with_different_evidence() -> None: + calibration, held_out, _ = _datasets() + changed = held_out.cases[0].model_copy( + update={ + "subject": calibration.cases[0].subject, + "evidence": (ArtifactRef(artifact_id="other-evidence", revision=1, content_digest="sha256:" + "b" * 64),), + } + ) + validate_split_pair(calibration, held_out.model_copy(update={"cases": (changed, *held_out.cases[1:])})) + + +def test_cli_rejects_duplicate_private_inputs_before_server_reads(tmp_path: Path, monkeypatch) -> None: + from evaluation.memory.write_gate import cli + + calibration, held_out, _ = _datasets() + copied = held_out.cases[0].model_copy( + update={ + "subject": calibration.cases[0].subject, + "evidence": calibration.cases[0].evidence, + } + ) + held_out = held_out.model_copy(update={"cases": (copied, *held_out.cases[1:])}) + for dataset in (calibration, held_out): + (tmp_path / f"{dataset.split}.json").write_text(dataset.model_dump_json(), encoding="utf-8") + + def unexpected_read(*args): + pytest.fail("overlapping inputs must be rejected before Server reads") + + monkeypatch.setenv("POWERCONTEXT_CLIENT_API_TOKEN", "test-token") + monkeypatch.setattr(cli, "_evaluate_remote", unexpected_read) + with pytest.raises(ValueError, match="private inputs must not cross splits"): + cli.main([ + "--calibration", + str(tmp_path / "calibration.json"), + "--held-out", + str(tmp_path / "held_out.json"), + "--server-url", + "http://localhost:8000", + ]) + + +@pytest.mark.parametrize("scenario", ["duplicate", "reordered_duplicate", "different_candidate", "different_evidence"]) +def test_resolved_split_validation_uses_gate_text_instead_of_artifact_metadata(scenario: str) -> None: + records = {} + datasets = [] + + def reference(scope_id, artifact_id, kind, text, citations=()): + content = {"schema": "powercontext.atomic-memory.v1", "kind": kind, "text": text, "creation": None} + digest = "sha256:" + sha256(rfc8785.dumps(content)).hexdigest() + records[artifact_id] = SimpleNamespace( + scope_id=scope_id, + family="atomic-memory", + artifact_id=artifact_id, + revision=1, + content_digest=digest, + content=content, + sources=(), + artifacts=citations, + ) + return ArtifactRef(artifact_id=artifact_id, revision=1, content_digest=digest) + + for dataset, kind in zip(_datasets()[:2], ("fact", "preference"), strict=True): + held_out = dataset.split == "held_out" + cases = [] + for index, case in enumerate(dataset.cases): + scope = f"scope-{dataset.split}" + evidence_text = f"evidence {index}" + if held_out and scenario == "different_evidence": + evidence_text += " changed" + evidence = ( + reference(scope, f"{case.case_id}-evidence-a", kind, evidence_text), + reference(scope, f"{case.case_id}-evidence-b", kind, "additional support"), + ) + if held_out and scenario == "reordered_duplicate": + evidence = tuple(reversed(evidence)) + citations = tuple( + SimpleNamespace(family=ref.family, artifact_id=ref.artifact_id, revision=ref.revision) + for ref in evidence + ) + candidate = f"candidate {index}" + if held_out and scenario == "different_candidate": + candidate += " changed" + subject = reference(scope, case.case_id, kind, candidate, citations) + cases.append(case.model_copy(update={"scope_id": scope, "subject": subject, "evidence": evidence})) + datasets.append(Dataset.model_validate({**dataset.model_dump(), "cases": cases})) + + class ExactReadClient: + async def get_artifact_revision(self, scope_id, family, artifact_id, revision): + return records[artifact_id] + + calibration, held_out = datasets + validate_split_pair(calibration, held_out) + assert calibration.cases[0].subject.content_digest != held_out.cases[0].subject.content_digest + resolver = SplitCheckedResolver( + calibration, held_out, AuthorizedClientResolver(cast(PowerContextClient, ExactReadClient())) + ) + + async def evaluate_splits(): + await evaluate_dataset(calibration, resolver) + return await evaluate_dataset(held_out, resolver) + + if scenario in {"duplicate", "reordered_duplicate"}: + with pytest.raises(ValueError, match="private gate inputs must not cross splits"): + asyncio.run(evaluate_splits()) + else: + asyncio.run(evaluate_splits()) + + +def test_cli_rejects_duplicate_gate_text_without_writing_report(tmp_path: Path, monkeypatch) -> None: + from evaluation.memory.write_gate import cli + + calibration, held_out, _ = _datasets() + for dataset in (calibration, held_out): + (tmp_path / f"{dataset.split}.json").write_text(dataset.model_dump_json(), encoding="utf-8") + + async def resolve(self, case): + return ResolvedCase(candidate="same candidate", evidence=("same evidence",)) + + monkeypatch.setenv("POWERCONTEXT_CLIENT_API_TOKEN", "test-token") + monkeypatch.setattr(AuthorizedClientResolver, "resolve", resolve) + output = tmp_path / "coverage.json" + with pytest.raises(ValueError, match="private gate inputs must not cross splits"): + cli.main([ + "--calibration", + str(tmp_path / "calibration.json"), + "--held-out", + str(tmp_path / "held_out.json"), + "--server-url", + "http://localhost:8000", + "--output", + str(output), + ]) + assert not output.exists() + + +@pytest.mark.parametrize( + ("error", "status"), + [ + (MissingInput, EvaluationStatus.MISSING_INPUT), + (Unauthorized, EvaluationStatus.UNAUTHORIZED), + (UnsupportedReference, EvaluationStatus.UNSUPPORTED_REFERENCE), + ], +) +def test_resolved_split_validation_preserves_incomplete_coverage(error, status) -> None: + class IncompleteResolver: + async def resolve(self, case): + raise error + + calibration, held_out, _ = _datasets() + + async def scenario(): + resolver = SplitCheckedResolver(calibration, held_out, IncompleteResolver()) + return await evaluate_dataset(held_out, resolver) + + rows = asyncio.run(scenario()) + assert all(row.status is status and row.actions is None for row in rows) + + +def _remote_case(*, evidence: tuple[SourceRef | ArtifactRef, ...]) -> Case: + return Case( + case_id="remote-1", + group_id="remote-group", + scope_id="scope-a", + language="en", + label=Label.ACCEPT, + label_reason="direct_support", + subject=ArtifactRef(artifact_id="memory-1", revision=1, content_digest=DIGEST), + evidence=evidence, + ) + + +def test_current_authorization_failure_is_reported_separately() -> None: + class DenyingClient: + async def get_artifact_revision(self, scope_id, family, artifact_id, revision): + assert scope_id == "scope-a" + raise ForbiddenResponseError(status_code=403, request_id=None) + + case = _remote_case(evidence=(ArtifactRef(artifact_id="evidence", revision=1, content_digest=DIGEST),)) + dataset = Dataset(dataset_id="remote", split="calibration", cases=(case,)) + result = asyncio.run(evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, DenyingClient())))) + assert result[0].status is EvaluationStatus.UNAUTHORIZED + assert result[0].actions is None + + +def test_exact_revision_and_citations_are_required_before_model_call() -> None: + class ChangedClient: + async def get_artifact_revision(self, scope_id, family, artifact_id, revision): + assert scope_id == "scope-a" + return SimpleNamespace( + scope_id=scope_id, + family=family, + artifact_id=artifact_id, + revision=revision, + content_digest="sha256:" + "b" * 64, + content={"text": "changed"}, + sources=[], + artifacts=[], + ) + + case = _remote_case(evidence=(ArtifactRef(artifact_id="evidence", revision=1, content_digest=DIGEST),)) + dataset = Dataset(dataset_id="remote", split="calibration", cases=(case,)) + result = asyncio.run(evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, ChangedClient())))) + assert result[0].status is EvaluationStatus.MISSING_INPUT + + +def test_current_scope_and_exact_citations_cannot_be_substituted() -> None: + class IncorrectClient: + def __init__(self, scope_id: str) -> None: + self.scope_id = scope_id + + async def get_artifact_revision(self, scope_id, family, artifact_id, revision): + return SimpleNamespace( + scope_id=self.scope_id, + family=family, + artifact_id=artifact_id, + revision=revision, + content_digest=DIGEST, + content={"text": "candidate"}, + sources=[], + artifacts=[SimpleNamespace(family="atomic-memory", artifact_id="other", revision=1)], + ) + + case = _remote_case(evidence=(ArtifactRef(artifact_id="evidence", revision=1, content_digest=DIGEST),)) + dataset = Dataset(dataset_id="remote", split="calibration", cases=(case,)) + wrong_scope = asyncio.run( + evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, IncorrectClient("scope-b")))) + ) + wrong_citation = asyncio.run( + evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, IncorrectClient("scope-a")))) + ) + assert wrong_scope[0].status is EvaluationStatus.MISSING_INPUT + assert wrong_citation[0].status is EvaluationStatus.UNSUPPORTED_REFERENCE + + +def test_source_read_without_exact_gate_projection_is_incomplete() -> None: + source = SourceRef(source_id="source-1", content_digest=DIGEST) + + class SourceClient: + async def get_artifact_revision(self, scope_id, family, artifact_id, revision): + return SimpleNamespace( + scope_id=scope_id, + family=family, + artifact_id=artifact_id, + revision=revision, + content_digest=DIGEST, + content={"text": "candidate"}, + sources=[SimpleNamespace(source_type="content", source_id="source-1")], + artifacts=[], + ) + + async def get_source(self, scope_id, source_type, source_id): + return SimpleNamespace( + scope_id=scope_id, + source_type=source_type, + source_id=source_id, + content_digest=DIGEST, + content="source text", + ) + + case = _remote_case(evidence=(source,)) + dataset = Dataset(dataset_id="remote", split="calibration", cases=(case,)) + result = asyncio.run(evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, SourceClient())))) + assert result[0].status is EvaluationStatus.UNSUPPORTED_REFERENCE + + +def test_uncommitted_candidate_reference_is_incomplete() -> None: + case = Case( + case_id="pending", + group_id="pending", + scope_id="scope-a", + language="en", + label=Label.HOLD, + label_reason="unsupported", + subject=SourceRef(source_id="candidate:1", content_digest=DIGEST), + evidence=(SourceRef(source_id="source-1", content_digest=DIGEST),), + ) + dataset = Dataset(dataset_id="remote", split="calibration", cases=(case,)) + result = asyncio.run(evaluate_dataset(dataset, AuthorizedClientResolver(cast(PowerContextClient, object())))) + assert result[0].status is EvaluationStatus.UNSUPPORTED_REFERENCE + + +@pytest.mark.parametrize("split", ["calibration", "held_out"]) +def test_no_call_cli_reports_local_preflight_hold_without_model_configuration( + tmp_path: Path, split: str, monkeypatch +) -> None: + from evaluation.memory.write_gate.cli import main + + calibration, held_out, _ = _datasets() + for dataset in (calibration, held_out): + (tmp_path / f"{dataset.split}.json").write_text(dataset.model_dump_json(), encoding="utf-8") + + async def resolve(self, case): + candidate = "x" * 4000 if case.case_id == f"{split}-0" else f"candidate {case.case_id}" + return ResolvedCase(candidate=candidate, evidence=("evidence",)) + + monkeypatch.setattr(AuthorizedClientResolver, "resolve", resolve) + monkeypatch.setenv("POWERCONTEXT_CLIENT_API_TOKEN", "unit-test-token") + output = tmp_path / "coverage.json" + assert ( + main([ + "--backend", + "no_call", + "--calibration", + str(tmp_path / "calibration.json"), + "--held-out", + str(tmp_path / "held_out.json"), + "--server-url", + "http://localhost:8000", + "--output", + str(output), + ]) + == 0 + ) + value = json.loads(output.read_text()) + assert value["phase"] == "input_coverage_audit" + assert value["selected_configuration"] is None + assert value["selected_policy_version"] is None + row = value["cases"][split][0] + assert row["coverage"] == "adjudicated" + assert row["used_fallback"] is False + assert row["outcome"] is None + assert set(row["actions"].values()) == {"hold"} + assert all(item["requests"] == 0 for rows in value["cases"].values() for item in rows) + assert "x" * 4000 not in output.read_text() diff --git a/tests/evaluation/test_memory_write_gate_public_dataset.py b/tests/evaluation/test_memory_write_gate_public_dataset.py new file mode 100644 index 0000000000..af8c70f777 --- /dev/null +++ b/tests/evaluation/test_memory_write_gate_public_dataset.py @@ -0,0 +1,549 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Public data identity, label isolation, and bilingual uncertainty behavior.""" + +import asyncio +import json +from pathlib import Path +from zipfile import ZipFile + +import httpx +import pytest +from pydantic import SecretStr + +from evaluation.memory.write_gate import cli, runner +from evaluation.memory.write_gate.artifacts import load_calibration_results +from evaluation.memory.write_gate.cli import main +from evaluation.memory.write_gate.dataset import Case, Dataset, Label, PublicDatasetRef, PublicDatasetSource +from evaluation.memory.write_gate.jev import JEV_ENDPOINT, JEV_MODEL, PublicJevModel +from evaluation.memory.write_gate.prepare_public import load_xnli, prepare, text_digest +from evaluation.memory.write_gate.report import calibration_report, score +from evaluation.memory.write_gate.resolver import MissingInput, PublicDatasetResolver +from evaluation.memory.write_gate.runner import CaseResult, EvaluationStatus, evaluate_dataset, evaluate_public_dataset +from powercontext.builtin.artifacts.memory.protocols import MemoryWriteVerdict +from powercontext.builtin.inference import InferenceUsage +from powercontext.builtin.runtime.decision_model import DecisionOutcome, DecisionResult + + +@pytest.fixture +def public_archive(tmp_path: Path) -> Path: + path = tmp_path / "corpus.zip" + with ZipFile(path, "w") as archive: + for split in ("dev", "test"): + rows = [ + { + "pairID": "1", + "language": language, + "gold_label": "entailment", + "sentence1": text, + "sentence2": text, + } + for language, text in (("en", "The meeting is on Friday."), ("zh", "会议定在周五。")) + ] + archive.writestr( + f"XNLI-1.0/xnli.{split}.jsonl", "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + ) + return path + + +def _dataset(path: Path) -> Dataset: + rows, digest = load_xnli(path) + pair = rows["dev"][("1", "en")] + ref = {"archive_sha256": digest, "split": "dev", "pair_id": pair.pair_id, "language": "en"} + case = Case( + case_id="public-1", + group_id="public-1", + scope_id="public-xnli", + language="en", + label=Label.ACCEPT, + label_reason="entailment", + original_label="entailment", + subject=PublicDatasetRef.model_validate({ + **ref, + "field": "hypothesis", + "content_digest": text_digest(pair.hypothesis), + }), + evidence=( + PublicDatasetRef.model_validate({**ref, "field": "premise", "content_digest": text_digest(pair.premise)}), + ), + ) + return Dataset( + dataset_id="public-test", + split="calibration", + public_source=PublicDatasetSource(archive_sha256=digest), + cases=(case,), + ) + + +def test_public_inputs_recover_without_exposing_gold_labels_to_model(public_archive: Path) -> None: + class CheckingModel: + policy_id = "test.public-label-isolation" + is_local_only = True + + async def evaluate(self, request): + assert request.evidence == ("The meeting is on Friday.",) + assert "The meeting is on Friday." in request.subject + assert "entailment" not in request.subject + " ".join(request.evidence) + return DecisionResult( + outcome=DecisionOutcome.NO, policy_id=self.policy_id, usage=InferenceUsage(requests=1) + ) + + result = asyncio.run( + evaluate_dataset(_dataset(public_archive), PublicDatasetResolver(public_archive), model=CheckingModel()) + ) + assert result[0].status is EvaluationStatus.EVALUATED + assert result[0].actions is not None + assert result[0].actions["hold_on=yes;threshold=none"] is MemoryWriteVerdict.ACCEPT + + +def test_public_reference_requires_matching_text_and_original_label(public_archive: Path) -> None: + case = _dataset(public_archive).cases[0] + changed = case.model_copy(update={"original_label": "neutral"}) + with pytest.raises(MissingInput): + asyncio.run(PublicDatasetResolver(public_archive).resolve(changed)) + changed = case.model_copy( + update={"subject": case.subject.model_copy(update={"content_digest": "sha256:" + "0" * 64})} + ) + with pytest.raises(MissingInput): + asyncio.run(PublicDatasetResolver(public_archive).resolve(changed)) + + +def test_public_source_rejects_wrong_partition_and_label_mapping(public_archive: Path) -> None: + dataset = _dataset(public_archive) + with pytest.raises(ValueError, match="original partition"): + Dataset.model_validate({**dataset.model_dump(), "split": "held_out"}) + with pytest.raises(ValueError, match="NLI mapping"): + Case.model_validate({**dataset.cases[0].model_dump(), "label": "hold"}) + + +def test_preparation_rejects_unpinned_archive_without_writing_outputs(public_archive: Path, tmp_path: Path) -> None: + output = tmp_path / "output" + with pytest.raises(ValueError, match="pinned official"): + prepare(public_archive, output) + assert not output.exists() + + +def test_translation_pairs_do_not_double_the_independent_sample_count() -> None: + rows = tuple( + CaseResult( + case_id=f"pair-{number}-{language}", + group_id=f"pair-{number}", + language=language, + label=Label.ACCEPT, + status=EvaluationStatus.EVALUATED, + actions={"hold_on=yes;threshold=none": MemoryWriteVerdict.ACCEPT}, + ) + for number in range(300) + for language in ("en", "zh") + ) + assert score(rows, "hold_on=yes;threshold=none")["false_hold_wilson_95"] is None + english = score(tuple(row for row in rows if row.language == "en"), "hold_on=yes;threshold=none") + assert 0.01 < english["false_hold_wilson_95"]["upper"] < 0.02 + + +def test_cli_requires_explicit_input_manifests() -> None: + with pytest.raises(SystemExit) as error: + main([]) + assert error.value.code == 2 + + +def test_hosted_benchmark_rejects_an_unpinned_archive_before_any_call(public_archive: Path) -> None: + class ForbiddenModel: + policy_id = "test.must-not-call" + + async def evaluate(self, request): + pytest.fail("untrusted inputs reached a hosted model") + + with pytest.raises(ValueError, match="pinned official"): + asyncio.run( + evaluate_public_dataset( + _dataset(public_archive), PublicDatasetResolver(public_archive), model=ForbiddenModel() + ) + ) + + +def test_jev_public_call_scores_real_response_fields_without_labels_or_local_attestation( + public_archive: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + dataset = _dataset(public_archive) + # Only this unit test trusts its small archive; real runs require the published checksum. + assert dataset.public_source is not None + monkeypatch.setattr(runner, "ARCHIVE_SHA256", dataset.public_source.archive_sha256) + sent = [] + + def handler(request: httpx.Request) -> httpx.Response: + assert str(request.url) == JEV_ENDPOINT + assert request.headers["Authorization"] == "Bearer unit-test-key" + payload = json.loads(request.content) + assert payload["model"] == JEV_MODEL + assert "The meeting is on Friday." in payload["state"] + assert "entailment" not in request.content.decode() + assert "public-1" not in request.content.decode() + sent.append(payload) + return httpx.Response( + 200, + json={ + "answers": { + "decision": { + "type": "choice", + "choice": "no", + "probabilities": {"yes": 0.02, "no": 0.96, "abstain": 0.02}, + "confidence": 0.94, + } + }, + "usage": {"input_tokens": 50, "output_tokens": 0, "cost": 0.0000021}, + }, + ) + + async def scenario(): + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = PublicJevModel(client, SecretStr("unit-test-key")) + resolver = PublicDatasetResolver(public_archive) + # The real gate still refuses a cloud backend, even for public inputs. + refused = await evaluate_dataset(dataset, resolver, model=model) + assert refused[0].fallback_reason == "privacy_boundary" + assert not sent + result = await evaluate_public_dataset(dataset, resolver, model=model) + assert model.is_local_only is False + return result + + rows = asyncio.run(scenario()) + assert len(sent) == 1 # Eight policy configurations rescore this one paid answer. + assert rows[0].actions["hold_on=yes;threshold=none"] is MemoryWriteVerdict.ACCEPT + assert rows[0].actions["hold_on=no;threshold=0.9"] is MemoryWriteVerdict.HOLD + assert rows[0].status is EvaluationStatus.EVALUATED + assert rows[0].requests == 1 + assert rows[0].latency_ms is not None + summary = calibration_report( + rows, dataset_id=dataset.dataset_id, dataset_hashes={}, code_commit="test", execution="mock_transport" + ) + cost = summary["calibration_grid"]["hold_on=yes;threshold=none"]["cost"] + assert cost["currency_cost"] == 0.0000021 + assert cost["input_tokens"] == 50 + assert cost["currency_cost_source"] == "provider_reported" + serialized = json.dumps(summary) + assert "The meeting is on Friday." not in serialized + assert "unit-test-key" not in serialized + + +def test_jev_failure_remains_unadjudicated_and_does_not_reuse_previous_cost( + public_archive: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + dataset = _dataset(public_archive) + assert dataset.public_source is not None + monkeypatch.setattr(runner, "ARCHIVE_SHA256", dataset.public_source.archive_sha256) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, text="sensitive provider error details") + + async def scenario(): + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + model = PublicJevModel(client, SecretStr("unit-test-key")) + model.last_cost_usd = 99.0 + return await evaluate_public_dataset(dataset, PublicDatasetResolver(public_archive), model=model) + + row = asyncio.run(scenario())[0] + assert row.used_fallback + assert row.coverage.value == "unadjudicated" + assert row.status is EvaluationStatus.EVALUATION_FALLBACK + assert row.fallback_reason == "backend_unavailable" + assert row.currency_cost_usd is None + assert all(action is MemoryWriteVerdict.ACCEPT for action in row.actions.values()) + summary = score((row,), "hold_on=yes;threshold=none") + assert summary["cost"]["backend_attempts"] == 1 + assert summary["cost"]["currency_cost"] is None + assert "sensitive provider" not in json.dumps(summary) + + +def test_jev_cli_rejects_private_inputs_and_missing_credentials(tmp_path: Path, monkeypatch, capsys) -> None: + for name in ("POWERCONTEXT_EVAL_JEV_API_KEY", "OPENROUTER_API_KEY"): + monkeypatch.delenv(name, raising=False) + args = [ + "--backend", + "jev", + "--calibration-only", + "--calibration", + "unused.json", + "--held-out", + "unused.json", + "--output", + str(tmp_path / "report.json"), + ] + with pytest.raises(SystemExit) as error: + main([*args, "--server-url", "https://private.example"]) + assert error.value.code == 2 + assert "only for the pinned public" in capsys.readouterr().err + with pytest.raises(SystemExit) as error: + main([*args, "--public-archive", "unused.zip"]) + assert error.value.code == 2 + assert "configure POWERCONTEXT_EVAL_JEV_API_KEY" in capsys.readouterr().err + assert not (tmp_path / "report.json").exists() + + +def test_calibration_command_never_calls_held_out_and_preserves_measurement_metadata( + tmp_path: Path, monkeypatch +) -> None: + archive_path = tmp_path / "public.zip" + with ZipFile(archive_path, "w") as archive: + for split in ("dev", "test"): + rows = [ + { + "pairID": f"{split}-{label}", + "language": language, + "gold_label": label, + "sentence1": f"{split} evidence {language} {index}", + "sentence2": f"{split} candidate {language} {index}", + } + for index, label in enumerate(("entailment", "contradiction", "neutral")) + for language in ("en", "zh") + ] + archive.writestr(f"XNLI-1.0/xnli.{split}.jsonl", "\n".join(json.dumps(row) for row in rows)) + splits, digest = load_xnli(archive_path) + monkeypatch.setattr(runner, "ARCHIVE_SHA256", digest) + for split, name in (("dev", "calibration"), ("test", "held_out")): + cases = [] + for pair in splits[split].values(): + ref = {"archive_sha256": digest, "split": split, "pair_id": pair.pair_id, "language": pair.language} + cases.append( + Case.model_validate({ + "case_id": f"{pair.pair_id}-{pair.language}", + "group_id": pair.pair_id, + "scope_id": "public-xnli", + "language": pair.language, + "label": Label.ACCEPT if pair.label == "entailment" else Label.HOLD, + "label_reason": pair.label, + "original_label": pair.label, + "subject": PublicDatasetRef.model_validate({ + **ref, + "field": "hypothesis", + "content_digest": text_digest(pair.hypothesis), + }), + "evidence": ( + PublicDatasetRef.model_validate({ + **ref, + "field": "premise", + "content_digest": text_digest(pair.premise), + }), + ), + }) + ) + dataset = Dataset.model_validate({ + "dataset_id": "mock-public-corpus", + "split": name, + "public_source": {"archive_sha256": digest}, + "cases": cases, + }) + (tmp_path / f"{name}.json").write_text(dataset.model_dump_json(), encoding="utf-8") + + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + payload = json.loads(request.content) + assert "test candidate" not in payload["state"] + assert "dev candidate" in payload["state"] + requests.append(payload) + return httpx.Response( + 200, + json={ + "answers": { + "decision": { + "type": "choice", + "choice": "yes", + "probabilities": {"yes": 0.8, "no": 0.1, "abstain": 0.1}, + "confidence": 0.7, + } + }, + "usage": {"input_tokens": 100, "output_tokens": 0, "cost": 0.0000042}, + }, + ) + + client_type = httpx.AsyncClient + monkeypatch.setattr(httpx, "AsyncClient", lambda: client_type(transport=httpx.MockTransport(handler))) + monkeypatch.setenv("POWERCONTEXT_EVAL_JEV_API_KEY", "unit-test-key") + output = tmp_path / "report.json" + assert ( + main([ + "--backend", + "jev", + "--calibration-only", + "--calibration", + str(tmp_path / "calibration.json"), + "--held-out", + str(tmp_path / "held_out.json"), + "--public-archive", + str(archive_path), + "--output", + str(output), + ]) + == 0 + ) + value = json.loads(output.read_text()) + assert len(requests) == 6 + assert len(value["cases"]) == 6 + assert value["runtime_hosted_egress_validated"] is False + assert value["run_scope"] == "evaluation" + assert value["backend_model_id"] == JEV_MODEL + assert value["decision_instructions_version"] == f"powercontext.decision.systemone.jev.v2:{JEV_MODEL}" + assert "held_out" not in value + assert set(value["calibration_by_language"]) == {"en", "zh"} + assert "unit-test-key" not in output.read_text() + assert "dev candidate" not in output.read_text() + + +def test_partial_provider_cost_is_not_reported_as_a_complete_total() -> None: + rows = tuple( + CaseResult( + case_id=f"case-{index}", + group_id=f"group-{index}", + language="en", + label=Label.ACCEPT, + status=EvaluationStatus.EVALUATED, + outcome=DecisionOutcome.NO, + currency_cost_usd=cost, + actions={"hold_on=yes;threshold=none": MemoryWriteVerdict.ACCEPT}, + ) + for index, cost in enumerate((0.001, None)) + ) + cost = score(rows, "hold_on=yes;threshold=none")["cost"] + assert cost["currency_cost"] is None + assert cost["currency_cost_known_subtotal"] == 0.001 + assert cost["currency_cost_known_cases"] == 1 + + +def test_calibration_reuse_requires_full_matching_data_code_and_backend(public_archive: Path, tmp_path: Path) -> None: + dataset = _dataset(public_archive) + rows = asyncio.run(evaluate_dataset(dataset, PublicDatasetResolver(public_archive))) + hashes = {"calibration": "calibration-digest", "held_out": "held-digest"} + value = calibration_report( + rows, + dataset_id=dataset.dataset_id, + dataset_hashes=hashes, + code_commit="test", + evaluation_code_digest="test-code", + execution="hosted_public_benchmark", + backend_model_id=JEV_MODEL, + ) + value.update(run_scope="evaluation", decision_instructions_version="test-policy") + path = tmp_path / "calibration.json" + path.write_text(json.dumps(value)) + restored = load_calibration_results( + path, + dataset, + dataset_hashes=hashes, + code_digest="test-code", + backend_policy_id="test-policy", + model_id=JEV_MODEL, + ) + assert restored == rows + with pytest.raises(ValueError, match="does not match"): + load_calibration_results( + path, + dataset, + dataset_hashes=hashes, + code_digest="changed-code", + backend_policy_id="test-policy", + model_id=JEV_MODEL, + ) + value["cases"] = [] + path.write_text(json.dumps(value)) + with pytest.raises(ValueError, match="incomplete case coverage"): + load_calibration_results( + path, + dataset, + dataset_hashes=hashes, + code_digest="test-code", + backend_policy_id="test-policy", + model_id=JEV_MODEL, + ) + + +def test_jev_credit_exhaustion_preserves_attempted_and_unattempted_cases( + public_archive: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + dataset = _dataset(public_archive) + assert dataset.public_source is not None + monkeypatch.setattr(runner, "ARCHIVE_SHA256", dataset.public_source.archive_sha256) + case = dataset.cases[0] + dataset = dataset.model_copy(update={"cases": (case, case.model_copy(update={"case_id": "public-2"}))}) + sent = [] + saved = [] + + def handler(request: httpx.Request) -> httpx.Response: + sent.append(request) + return httpx.Response(402, json={"error": {"code": 402, "message": "private provider details"}}) + + async def scenario(): + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + return await evaluate_public_dataset( + dataset, + PublicDatasetResolver(public_archive), + model=PublicJevModel(client, SecretStr("unit-test-key")), + on_result=saved.append, + ) + + rows = asyncio.run(scenario()) + assert len(sent) == 1 + assert saved == list(rows) + assert rows[0].status is EvaluationStatus.EVALUATION_FALLBACK + assert rows[0].used_fallback + assert rows[0].coverage.value == "unadjudicated" + assert rows[0].fallback_reason == "insufficient_credit" + assert rows[1].status is EvaluationStatus.NOT_ATTEMPTED + summary = score(rows, "hold_on=yes;threshold=none") + assert summary["input_coverage"] == 0.5 + assert summary["adjudication_coverage"] == 0 + assert summary["false_hold_denominator"] == 0 + assert summary["cost"]["backend_attempts"] == 1 + + +def test_cached_calibration_rejected_when_shared_assessor_changes( + public_archive: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + dataset = _dataset(public_archive) + rows = asyncio.run(evaluate_dataset(dataset, PublicDatasetResolver(public_archive))) + hashes = {"calibration": "calibration-digest", "held_out": "held-digest"} + root = Path(cli.__file__).resolve().parents[3] + copied = tmp_path / "checkout" + for relative in cli._code_identity(): + target = copied / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes((root / relative).read_bytes()) + monkeypatch.setattr(cli, "__file__", str(copied / "evaluation/memory/write_gate/cli.py")) + value = calibration_report( + rows, + dataset_id=dataset.dataset_id, + dataset_hashes=hashes, + code_commit="test", + evaluation_code_digest=cli._code_digest(), + evaluation_code_files=cli._code_identity(), + execution="hosted_public_benchmark", + backend_model_id=JEV_MODEL, + ) + value.update(run_scope="evaluation", decision_instructions_version="test-policy") + path = tmp_path / "calibration.json" + path.write_text(json.dumps(value)) + assessor = copied / "src/powercontext/builtin/runtime/decision_policy.py" + original = assessor.read_text() + assert "verdict=mapping[result.outcome]" in original + assessor.write_text(original.replace("verdict=mapping[result.outcome]", "verdict=DecisionVerdict.ALLOW")) + with pytest.raises(ValueError, match="does not match"): + load_calibration_results( + path, + dataset, + dataset_hashes=hashes, + code_digest=cli._code_digest(), + backend_policy_id="test-policy", + model_id=JEV_MODEL, + )