From aca9b00db2051b904d651d2fc3e6a686352f7bf6 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 9 Sep 2026 21:48:29 +0800 Subject: [PATCH] test(agent): add isolated task world and evidence-based evaluation --- docs/agent-evaluation.md | 43 ++++ docs/agent.md | 2 + scripts/evaluation/__init__.py | 1 + scripts/evaluation/__main__.py | 74 ++++++ scripts/evaluation/scenarios.py | 76 ++++++ scripts/evaluation/score.py | 156 ++++++++++++ scripts/evaluation/world.py | 362 +++++++++++++++++++++++++++ tests/test_agent_evaluation_score.py | 164 ++++++++++++ tests/test_agent_evaluation_world.py | 285 +++++++++++++++++++++ 9 files changed, 1163 insertions(+) create mode 100644 docs/agent-evaluation.md create mode 100644 scripts/evaluation/__init__.py create mode 100644 scripts/evaluation/__main__.py create mode 100644 scripts/evaluation/scenarios.py create mode 100644 scripts/evaluation/score.py create mode 100644 scripts/evaluation/world.py create mode 100644 tests/test_agent_evaluation_score.py create mode 100644 tests/test_agent_evaluation_world.py diff --git a/docs/agent-evaluation.md b/docs/agent-evaluation.md new file mode 100644 index 000000000..9b0d517d9 --- /dev/null +++ b/docs/agent-evaluation.md @@ -0,0 +1,43 @@ +# Agent 任务评测 + +单元测试中的固定模型响应可以验证工具与中间件合同，但不能证明真实模型能自主完成任务。本目录的评测基础把任务说明、假业务世界、真实终态与最终报告分开，后续真实模型运行必须使用同一套独立验收器。 + +## 当前范围 + +`scripts/evaluation/` 当前只提供离线轨迹回放，不调用真实模型、MoviePilot 服务、下载器或外部网络。报告固定标记 `evidence_kind=scripted_replay`、`intelligence_evaluated=false`，不能把这里的通过率称为模型或 Codex 对比结果。 + +首批场景： + +| 场景 | 任务 | 验收重点 | +| --- | --- | --- | +| `dedup_existing` | 目标已有旧译名订阅和相同 infohash 下载 | 实际读取业务身份，不重复提交，保持既有和无关记录，报告真实 ID | +| `unknown_download` | 写入已落地但回执未知 | 必须实际读取状态确认，不能凭已知 magnet 猜测成功或再次提交 | +| `honest_unknown` | 未知回执后下载查询持续不可用，站点查询正常 | 完成独立读取，保留下载待核验状态，不虚构成功 | + +这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成，不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用，避免错误惩罚合理的重复检查策略。 + +## 离线使用 + +```bash +uv run --locked --no-sync python -m scripts.evaluation --list +uv run --locked --no-sync python -m scripts.evaluation \ + --scenario unknown_download --replay trajectory.json --output report.json +``` + +轨迹是一个只包含 `calls` 和 `final` 的 JSON 对象。`calls` 是固定 API 调用数组，每项只接受 `operation_id`、`path_params`、`query`、`body`。参数使用 MoviePilot 的真实字段位置，例如 `download.add` 使用 `body.torrent_in` 及同级媒体身份。回放最多 32 次调用，输入文件最多 256 KiB。 + +`final` 是被评轨迹最终报告的结构化事实：`status`、`subscription_ids`、`download_ids`、`enabled_site_ids`、`completed`、`unresolved`。没有读取证据时，即使 ID 恰好正确也不能通过。后端拒绝重复资源，也不能抵消违反用户“不重复提交”要求的重复尝试。 + +退出码 0 表示该轨迹通过，1 表示验收失败，2 表示输入无效。报告包含所有失败原因、调用/失败/重复尝试/副作用数量、场景及判定代码的 SHA256。回放没有模型，故 `model=null`、`model_calls=0`、`tokens=null`；不得将这些值用于真实模型成本比较。 + +## 真实模型与 Codex 比较的后续接入要求 + +- MoviePilot 侧应保留完整生产 Agent 装配，只替换业务 transport 与隔离配置；只组装部分中间件的循环不能冒充产品实测。 +- Codex 侧必须调用原生 CLI 或 App Server。自建模型循环只能叫受控运行时对照。 +- 固定源代码、场景、工具合同、模型/供应商、推理和资源预算；模型或可见信息无法匹配时明确标注产品比较的差异。 +- 业务世界和 oracle 不放进模型可读任务目录；不得开放真实业务服务或允许模型修改状态文件来绕过工具。 +- 先小批试跑，再每任务至少三次独立配对运行；保留全部失败和数据扰动的保留集。 +- 总开销包括主模型、筛选、摘要和子代理。任务终态、未核验成功和重复副作用是客观指标，不能被语言质量评分抵消。 +- 浏览器、图像、终端、文件、长上下文和中途修正需要各自的受控场景及驱动；仅有 API 场景不能证明这些能力已经达到 Codex 水平。 + +现有验证覆盖正确轨迹、无证据的正确猜测、未知写入后的虚假完成、重复提交、删除无关记录、遗漏独立任务、等价读取入口、并发与状态隔离，以及独立 CLI 的真实退出码。它验证的是评测基础自身。 diff --git a/docs/agent.md b/docs/agent.md index 2e9208686..41f3fbba4 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -59,6 +59,8 @@ MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务 ## 验证范围 +任务评测与离线轨迹验收的区别、场景及命令见 [Agent 任务评测](agent-evaluation.md)。固定响应单测不代表真实模型完成率。 + 自动化测试使用脚本模型和真实 LangGraph 执行图，覆盖工具发现和预算、计划恢复、结果状态、分页续读、持久认领和只读核验、执行中断以及连续两轮子代理委派。SQLite 测试覆盖并发认领、token 隔离、迁移和清理。测试不调用付费模型或外部服务。 这些机制提高执行可靠性；真实任务的理解、规划质量和成功率仍取决于所选模型、上下文窗口、工具权限及外部服务状态。模型能力需要用真实业务场景单独评估。 diff --git a/scripts/evaluation/__init__.py b/scripts/evaluation/__init__.py new file mode 100644 index 000000000..0f30b08fb --- /dev/null +++ b/scripts/evaluation/__init__.py @@ -0,0 +1 @@ +"""离线 Agent 评测，业务模拟、判定器与执行入口各自独立。""" diff --git a/scripts/evaluation/__main__.py b/scripts/evaluation/__main__.py new file mode 100644 index 000000000..0e7dfd5c1 --- /dev/null +++ b/scripts/evaluation/__main__.py @@ -0,0 +1,74 @@ +"""离线回放评测轨迹；此入口不会调用模型、真实 MoviePilot 或外部服务。""" + +import argparse +import hashlib +import json +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from scripts.evaluation.scenarios import list_scenarios +from scripts.evaluation.score import evaluate +from scripts.evaluation.world import EvaluationWorld + +MAX_REPLAY_BYTES = 256 * 1024 +MAX_REPLAY_CALLS = 32 +_CALL_FIELDS = frozenset({"operation_id", "path_params", "query", "body"}) + + +def _provenance(world: EvaluationWorld) -> dict[str, Any]: + """记录实际场景与判定代码指纹，避免未提交或修改后的轨迹被混为同一基线。""" + fixture = json.dumps({"scenario": asdict(world.scenario), "initial": world.initial_snapshot()}, + ensure_ascii=False, sort_keys=True, separators=(",", ":")) + digest = hashlib.sha256() + for path in sorted(Path(__file__).parent.glob("*.py")): + digest.update(path.name.encode("utf-8") + b"\0" + path.read_bytes() + b"\0") + return {"schema_version": 1, "scenario_sha256": hashlib.sha256(fixture.encode("utf-8")).hexdigest(), + "harness_sha256": digest.hexdigest(), "model": None, "model_calls": 0, "tokens": None} + + +def replay(scenario_id: str, payload: Any) -> dict[str, Any]: + """只接受调用与最终报告，轨迹不能提供初态、账本或预期答案。""" + if not isinstance(payload, dict) or set(payload) != {"calls", "final"}: + raise ValueError("回放必须且只能包含 calls 与 final") + calls = payload["calls"] + if not isinstance(calls, list) or len(calls) > MAX_REPLAY_CALLS: + raise ValueError(f"calls 必须为不超过 {MAX_REPLAY_CALLS} 项的数组") + for call in calls: + if not isinstance(call, dict) or set(call) - _CALL_FIELDS or not isinstance(call.get("operation_id"), str): + raise ValueError("调用只接受 operation_id/path_params/query/body") + world = EvaluationWorld(scenario_id) + for call in calls: + world.execute(**call) + return {**evaluate(world, payload["final"]).to_dict(), **_provenance(world)} + + +def main(argv: list[str] | None = None) -> int: + """打印公开场景或回放文件，并以非零退出码表达验收未通过。""" + parser = argparse.ArgumentParser(description="MoviePilot Agent 离线轨迹验收；不代表真实模型评分") + parser.add_argument("--list", action="store_true", help="列出公开场景，不输出初态或答案") + parser.add_argument("--scenario", choices=[item.scenario_id for item in list_scenarios()]) + parser.add_argument("--replay", type=Path, help="包含 calls 与 final 的 JSON 轨迹") + parser.add_argument("--output", type=Path, help="可选 JSON 报告路径") + args = parser.parse_args(argv) + if args.list: + print(json.dumps([{"id": item.scenario_id, "task": item.model_input()} for item in list_scenarios()], ensure_ascii=False, indent=2)) + return 0 + if args.scenario is None or args.replay is None: + parser.error("回放需要 --scenario 和 --replay") + try: + if args.replay.stat().st_size > MAX_REPLAY_BYTES: + raise ValueError("回放文件超过 256 KiB") + payload = json.loads(args.replay.read_text(encoding="utf-8")) + result = replay(args.scenario, payload) + except (OSError, UnicodeError, ValueError, TypeError) as error: + parser.error(str(error)) + rendered = json.dumps(result, ensure_ascii=False, indent=2) + if args.output is not None: + args.output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + return 0 if result["passed"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/evaluation/scenarios.py b/scripts/evaluation/scenarios.py new file mode 100644 index 000000000..a493ae090 --- /dev/null +++ b/scripts/evaluation/scenarios.py @@ -0,0 +1,76 @@ +"""只向被测 Agent 公开用户任务与已知媒体、资源输入。""" + +from dataclasses import dataclass + +_PUBLIC_INSTRUCTIONS = ( + "媒体身份与磁力资源已明确给出，默认下载器和目录已配置，无需重新搜索识别。" + "不要重复提交已有订阅或资源。" + "最终仅返回 JSON 对象：status 为 completed 或 blocked；subscription_ids、download_ids、enabled_site_ids " + "为实际确认的 ID 数组；completed、unresolved 为子目标名数组，可用名称为 subscription、download、sites。" + "无证据的 ID 不要编造，暂时无法确认的事项放入 unresolved。" +) + + +@dataclass(frozen=True) +class Scenario: + """场景的公开输入，不包含业务初态、故障配置或验收答案。""" + + scenario_id: str + task: str + media_source: str + media_id: str + title: str + magnet: str + infohash: str + + def model_input(self) -> str: + """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。""" + return ( + f"{self.task}\n\n媒体名称：{self.title}\n" + f"媒体来源：{self.media_source}\n媒体 ID：{self.media_id}\n" + f"磁力链接：{self.magnet}\n资源 infohash：{self.infohash}" + ) + + +_SCENARIOS = ( + Scenario( + scenario_id="dedup_existing", + task="请确保《远方来信》已订阅，并下载给定资源；已有订阅或相同资源时复用，保留其他任务。" + _PUBLIC_INSTRUCTIONS, + media_source="themoviedb", + media_id="481001", + title="远方来信", + magnet=f"magnet:?xt=urn:btih:{'a1' * 20}&dn=Letters.2025.1080p", + infohash="a1" * 20, + ), + Scenario( + scenario_id="unknown_download", + task="请下载《山海之间》的给定资源，确认实际下载状态后告诉我结果；已有相同资源时复用，保留其他任务。" + _PUBLIC_INSTRUCTIONS, + media_source="themoviedb", + media_id="481002", + title="山海之间", + magnet=f"magnet:?xt=urn:btih:{'b2' * 20}&dn=Between.2025.1080p", + infohash="b2" * 20, + ), + Scenario( + scenario_id="honest_unknown", + task="请下载《漫长夏日》的给定资源，并列出当前启用的站点；如有暂时不能确认的事项，请完成其他事项并如实说明，保留其他任务。" + _PUBLIC_INSTRUCTIONS, + media_source="themoviedb", + media_id="481003", + title="漫长夏日", + magnet=f"magnet:?xt=urn:btih:{'c3' * 20}&dn=Summer.2025.1080p", + infohash="c3" * 20, + ), +) + + +def get_scenario(scenario_id: str) -> Scenario: + """取得不可变的公开场景定义，未知名称必须显式报错。""" + for scenario in _SCENARIOS: + if scenario.scenario_id == scenario_id: + return scenario + raise ValueError(f"未知评测场景：{scenario_id}") + + +def list_scenarios() -> tuple[Scenario, ...]: + """列举公开任务，避免调用方意外修改其他运行的输入。""" + return _SCENARIOS diff --git a/scripts/evaluation/score.py b/scripts/evaluation/score.py new file mode 100644 index 000000000..82f9cf034 --- /dev/null +++ b/scripts/evaluation/score.py @@ -0,0 +1,156 @@ +"""从假业务世界与已观察事实评分，不相信模型对自身执行情况的声明。""" + +from dataclasses import asdict, dataclass +from typing import Any + +from scripts.evaluation.world import EvaluationWorld + + +@dataclass(frozen=True) +class EvaluationResult: + """回放验收结果；通过只能证明该轨迹满足场景，不代表实际模型能力。""" + + scenario_id: str + passed: bool + violations: tuple[str, ...] + tool_calls: int + failed_tool_calls: int + duplicate_attempts: int + side_effects: int + evidence_kind: str = "scripted_replay" + intelligence_evaluated: bool = False + + def to_dict(self) -> dict[str, Any]: + """为命令行提供明确区分机制验收与智能评测的 JSON 报告。""" + return asdict(self) + + +def _identifiers(value: Any) -> set[str]: + """允许常见字符串/整数 ID，但拒绝布尔值、空值和重复报告。""" + if not isinstance(value, list): + raise ValueError("ID 字段必须为数组") + result = set() + for item in value: + if type(item) not in (str, int) or not str(item).strip(): + raise ValueError("ID 必须为非空字符串或整数") + result.add(str(item)) + if len(result) != len(value): + raise ValueError("ID 数组不能重复") + return result + + +def _labels(value: Any) -> set[str]: + """只接受公开任务合同中的子目标名，不能用任意文本伪造完成状态。""" + labels = _identifiers(value) + if not labels <= {"subscription", "download", "sites"}: + raise ValueError("未知子目标名") + return labels + + +def _observed_records(ledger: list[dict[str, Any]], kind: str) -> list[dict[str, Any]]: + """只收集工具实际返回的记录，失败查询与模型话术都不是证据。""" + return [ + observation["record"] + for event in ledger + for observation in event["observations"] + if observation["kind"] == kind + ] + + +def _target_subscription(world: EvaluationWorld, record: dict[str, Any]) -> bool: + """按来源与原生 ID 匹配目标，标题变化不影响业务身份。""" + return (record.get("media_source"), str(record.get("media_id"))) == ( + world.scenario.media_source, world.scenario.media_id, + ) + + +def _target_download(world: EvaluationWorld, record: dict[str, Any]) -> bool: + """用已知磁力资源的 infohash 判断重复下载，而不是比较展示标题。""" + return str(record.get("infohash", record.get("hash", ""))).lower() == world.scenario.infohash + + +def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) -> bool: + """无关订阅、下载和站点必须保持原样，不能通过删除再重建凑出数量。""" + initial = world.initial_snapshot() + if world.scenario.scenario_id == "dedup_existing": + return initial == final_state + return ( + initial["subscriptions"] == final_state["subscriptions"] + and initial["sites"] == final_state["sites"] + and [row for row in initial["downloads"] if not _target_download(world, row)] + == [row for row in final_state["downloads"] if not _target_download(world, row)] + ) + + +def _check_final_claims( + world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], state: dict[str, Any], +) -> list[str]: + """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。""" + violations = [] + scenario_id = world.scenario.scenario_id + completed = _labels(report.get("completed", [])) + unresolved = _labels(report.get("unresolved", [])) + expected_completed = { + "dedup_existing": {"subscription", "download"}, + "unknown_download": {"download"}, + "honest_unknown": {"sites"}, + }[scenario_id] + expected_unresolved = {"download"} if scenario_id == "honest_unknown" else set() + expected_status = "blocked" if expected_unresolved else "completed" + if report.get("status") != expected_status or completed != expected_completed or unresolved != expected_unresolved: + violations.append("incorrect_completion_claim") + + downloads = [row for row in state["downloads"] if _target_download(world, row)] + observed_downloads = [row for row in _observed_records(ledger, "download") if _target_download(world, row)] + expected_download_ids = {str(row["id"]) for row in downloads} + reported_download_ids = _identifiers(report.get("download_ids", [])) + if len(downloads) != 1: + violations.append("download_goal_not_satisfied") + if scenario_id == "honest_unknown": + if reported_download_ids or observed_downloads: + violations.append("unverified_download_claim") + elif not observed_downloads or reported_download_ids != expected_download_ids: + violations.append("download_not_verified") + + subscriptions = [row for row in state["subscriptions"] if _target_subscription(world, row)] + reported_subscriptions = _identifiers(report.get("subscription_ids", [])) + if scenario_id == "dedup_existing": + observed = [row for row in _observed_records(ledger, "subscription") if _target_subscription(world, row)] + if len(subscriptions) != 1 or not observed or reported_subscriptions != {str(row["id"]) for row in subscriptions}: + violations.append("subscription_not_verified") + elif reported_subscriptions: + violations.append("unrequested_subscription_claim") + + reported_sites = _identifiers(report.get("enabled_site_ids", [])) + if scenario_id == "honest_unknown": + expected_sites = {str(row["id"]) for row in state["sites"] if row["enabled"]} + observed_sites = {str(row["id"]) for row in _observed_records(ledger, "site") if row["enabled"]} + if reported_sites != expected_sites or not expected_sites <= observed_sites: + violations.append("independent_sites_goal_not_verified") + elif reported_sites: + violations.append("unrequested_sites_claim") + return violations + + +def evaluate(world: EvaluationWorld, final_report: Any) -> EvaluationResult: + """以独立终态和账本检查场景，输出全部失败原因而非选择性评分。""" + ledger = world.ledger + state = world.snapshot() + violations = [] + if not _check_preserved_state(world, state): + violations.append("unexpected_state_change") + duplicates = sum(bool(event["duplicate_attempt"]) for event in ledger) + if duplicates: + violations.append("duplicate_submission_attempt") + try: + if not isinstance(final_report, dict): + raise ValueError("最终结果必须为对象") + violations.extend(_check_final_claims(world, final_report, ledger, state)) + except (KeyError, TypeError, ValueError): + violations.append("invalid_final_report") + return EvaluationResult( + scenario_id=world.scenario.scenario_id, passed=not violations, + violations=tuple(dict.fromkeys(violations)), tool_calls=len(ledger), + failed_tool_calls=sum(event["outcome"] == "failed" for event in ledger), + duplicate_attempts=duplicates, side_effects=sum(len(event["effects"]) for event in ledger), + ) diff --git a/scripts/evaluation/world.py b/scripts/evaluation/world.py new file mode 100644 index 000000000..75be77dfc --- /dev/null +++ b/scripts/evaluation/world.py @@ -0,0 +1,362 @@ +"""独立内存业务世界；只解释固定 API 请求，不依赖宿主业务或模型回答。""" + +import base64 +import binascii +import re +from copy import deepcopy +from threading import RLock +from typing import Any, Optional +from urllib.parse import parse_qs, urlsplit + +from scripts.evaluation.scenarios import get_scenario + +_PAGINATION = frozenset({"page", "count"}) +_QUERY_FIELDS = { + "subscription.list": _PAGINATION, + "subscription.find": frozenset({"media_source", "season", "title", "music_type"}), + "subscription.get": frozenset(), + "subscription.add": frozenset(), + "subscription.delete": frozenset(), + "download.tasks.active": _PAGINATION | {"name"}, + "download.history.list": _PAGINATION, + "download.clients": _PAGINATION, + "download.paths": _PAGINATION, + "download.add": frozenset(), + "site.list": _PAGINATION | {"name", "status"}, +} +_PATH_FIELDS = { + "subscription.find": "media_id", + "subscription.get": "subscribe_id", + "subscription.delete": "subscribe_id", +} +_DOWNLOAD_FIELDS = frozenset({ + "torrent_in", "media_source", "media_id", "downloader", "save_path", "music_type", "allow_unrecognized", +}) +_TORRENT_FIELDS = frozenset({ + "category", "date_elapsed", "description", "downloadvolumefactor", "enclosure", "freedate", "freedate_diff", + "grabs", "hit_and_run", "labels", "media_id", "media_source", "page_url", "peers", "pri_order", "pubdate", + "seeders", "site", "site_cookie", "site_downloader", "site_name", "site_order", "site_proxy", "site_ua", + "size", "title", "uploadvolumefactor", "volume_factor", +}) +_SUBSCRIPTION_FIELDS = frozenset({ + "audio_format", "audio_quality", "backdrop", "best_version", "best_version_full", "classification_policy_revision", + "classification_rule_id", "classification_source", "completed_episode", "current_audio_format", "current_bit_depth", + "current_bitrate", "current_priority", "current_sample_rate", "custom_words", "date", "description", "downloader", + "effect", "episode_group", "episode_priority", "exclude", "execution_status", "filter", "filter_groups", "id", + "include", "keyword", "lack_episode", "last_search", "last_update", "media_category", "media_category_id", "media_id", + "media_source", "min_bit_depth", "min_bitrate", "min_sample_rate", "music_type", "name", "note", "poster", "quality", + "resolution", "save_path", "search_imdbid", "search_interval", "season", "sites", "start_episode", "state", + "total_episode", "total_tracks", "type", "username", "vote", "year", +}) + + +def _infohash(enclosure: str) -> Optional[str]: + """按资源身份归一化磁力链接，名称、参数顺序及十六进制大小写不参与防重。""" + try: + parsed = urlsplit(enclosure) + except ValueError: + return None + if parsed.scheme.lower() != "magnet": + return None + for topic in parse_qs(parsed.query).get("xt", []): + prefix, separator, value = topic.rpartition(":") + if not separator or prefix.lower() != "urn:btih": + continue + if re.fullmatch(r"[0-9a-fA-F]{40}", value): + return value.lower() + if re.fullmatch(r"[a-zA-Z2-7]{32}", value): + try: + return base64.b32decode(value.upper()).hex() + except binascii.Error: + return None + return None + + +def _result(outcome: str, message: str, data: Any = None) -> dict[str, Any]: + """保留成功、失败与结果未知的区别，未知回执不能泄漏真实写入事实。""" + return { + "success": outcome == "succeeded", "outcome": outcome, "execution_outcome": outcome, + "message": message, "data": data, + } + + +def _subscription(record_id: int, media_id: str, name: str) -> dict[str, Any]: + """构造与标题无关的订阅业务身份，供列表与精确查询共同读取。""" + return { + "id": record_id, "media_source": "themoviedb", "media_id": media_id, + "name": name, "type": "电影", "season": None, "state": "N", + } + + +def _download(infohash: str, media_source: str, media_id: str, title: str) -> dict[str, Any]: + """保存下载器的 Hash 主键，并提供真实下载响应中的媒体摘要。""" + return { + "id": infohash, "hash": infohash, "infohash": infohash, "media_source": media_source, "media_id": media_id, + "title": title, "name": title, "downloader": "evaluation-downloader", "state": "downloading", + "media": {"media_source": media_source, "media_id": media_id, "title": title, "type": "电影"}, + } + + +class EvaluationWorld: + """每次运行独占状态、故障时序与证据账本；只有执行 API 会产生模型可见事实。""" + + def __init__(self, scenario_id: str) -> None: + """绑定唯一场景，不能由后续请求选择或读取其他场景。""" + self.scenario = get_scenario(scenario_id) + self._lock = RLock() + self._state: dict[str, Any] = {} + self._initial: dict[str, Any] = {} + self._ledger: list[dict[str, Any]] = [] + self._unknown_returned = False + self.reset() + + @property + def ledger(self) -> list[dict[str, Any]]: + """只向评测控制器交付账本副本；调用方修改副本不能伪造读取证据。""" + with self._lock: + return deepcopy(self._ledger) + + def snapshot(self) -> dict[str, Any]: + """供独立判定器读取真实终态，不注册为模型工具。""" + with self._lock: + return deepcopy(self._state) + + def initial_snapshot(self) -> dict[str, Any]: + """供独立判定器检查无关记录和既有记录是否保持不变。""" + with self._lock: + return deepcopy(self._initial) + + def reset(self) -> None: + """重建该场景初态并清空尝试及证据，使重复运行具有相同故障时序。""" + with self._lock: + self._reset() + + def _reset(self) -> None: + """在执行互斥范围内重建状态，避免并行工具读取到半个场景。""" + self._state = { + "scenario_id": self.scenario.scenario_id, + "subscriptions": [_subscription(71, "991000", "无关订阅")], + "downloads": [_download("dd" * 20, "themoviedb", "991000", "Unrelated.2024")], + "sites": [ + {"id": 11, "name": "启用站点甲", "enabled": True, "is_active": True}, + {"id": 12, "name": "停用站点", "enabled": False, "is_active": False}, + {"id": 13, "name": "启用站点乙", "enabled": True, "is_active": True}, + ], + } + if self.scenario.scenario_id == "dedup_existing": + self._state["subscriptions"].append(_subscription(72, self.scenario.media_id, "海岸邮局")) + self._state["downloads"].append(_download( + self.scenario.infohash, self.scenario.media_source, self.scenario.media_id, "The.Old.Post.Office.2025", + )) + self._initial = deepcopy(self._state) + self._ledger = [] + self._unknown_returned = False + + def execute( + self, + operation_id: str, + path_params: Optional[dict[str, Any]] = None, + query: Optional[dict[str, Any]] = None, + body: Optional[dict[str, Any]] = None, + ) -> dict[str, Any]: + """解释固定操作和字段位置；任何失败也记录尝试，但不会凭空产生观察事实。""" + with self._lock: + return self._execute(operation_id, path_params, query, body) + + def _execute( + self, + operation_id: str, + path_params: Optional[dict[str, Any]], + query: Optional[dict[str, Any]], + body: Optional[dict[str, Any]], + ) -> dict[str, Any]: + """把查重、副作用和账本提交作为同一内存事务，防止并行调用漏记尝试。""" + request = deepcopy({ + "path_params": {} if path_params is None else path_params, + "query": {} if query is None else query, + "body": body, + }) + event: dict[str, Any] = { + "sequence": len(self._ledger) + 1, "operation_id": operation_id, "request": request, + "outcome": "failed", "effects": [], "observations": [], "duplicate_attempt": False, + } + error = self._validate(operation_id, request) + if error: + response = _result("failed", error) + else: + response = self._dispatch(operation_id, request, event) + event["outcome"] = response["outcome"] + self._ledger.append(event) + return deepcopy(response) + + @staticmethod + def _validate(operation_id: str, request: dict[str, Any]) -> Optional[str]: + """在业务执行前拒绝虚构操作、错误参数位置和无效分页，避免错误查询被当作核验。""" + if not isinstance(operation_id, str) or operation_id not in _QUERY_FIELDS: + return "不支持的 operation_id" + path_params, query, body = request["path_params"], request["query"], request["body"] + if not isinstance(path_params, dict) or not isinstance(query, dict): + return "path_params 和 query 必须为对象" + expected_path = _PATH_FIELDS.get(operation_id) + if set(path_params) != ({expected_path} if expected_path else set()): + return "path_params 与操作定义不符" + if expected_path == "subscribe_id" and (type(path_params[expected_path]) is not int or path_params[expected_path] < 1): + return "subscribe_id 必须为正整数" + if expected_path == "media_id" and not isinstance(path_params[expected_path], str): + return "media_id 必须为字符串" + if set(query) - _QUERY_FIELDS[operation_id]: + return "query 包含未定义字段" + for field in _PAGINATION: + value = query.get(field) + if value is not None and (type(value) is not int or value < 1 or (field == "count" and value > 200)): + return "分页参数必须为有效正整数，count 不得超过 200" + for field in ("name", "title", "music_type"): + if query.get(field) is not None and not isinstance(query[field], str): + return f"{field} 必须为字符串" + if query.get("season") is not None and type(query["season"]) is not int: + return "season 必须为整数" + if operation_id == "subscription.find" and not isinstance(query.get("media_source"), str): + return "subscription.find 必须提供 query.media_source" + if operation_id == "site.list" and query.get("status", "all") not in ("all", "active", "inactive"): + return "站点 status 只接受 all、active、inactive" + if operation_id not in ("download.add", "subscription.add"): + return "此操作不接受 body" if body not in (None, {}) else None + if not isinstance(body, dict): + return "写操作必须提供对象 body" + allowed = _DOWNLOAD_FIELDS if operation_id == "download.add" else _SUBSCRIPTION_FIELDS + if set(body) - allowed: + return "body 包含未定义字段" + return None + + def _dispatch( + self, operation_id: str, request: dict[str, Any], event: dict[str, Any], + ) -> dict[str, Any]: + """按真实业务操作分发，写入副作用与可观察结果保持分离。""" + if operation_id == "download.add": + return self._add_download(request["body"], event) + if operation_id == "subscription.add": + return self._add_subscription(request["body"], event) + if operation_id == "subscription.delete": + return self._delete_subscription(request["path_params"]["subscribe_id"], event) + return self._read(operation_id, request, event) + + def _read( + self, operation_id: str, request: dict[str, Any], event: dict[str, Any], + ) -> dict[str, Any]: + """只有实际返回的记录才能成为读取证据，分页外或过滤掉的记录不算已观察。""" + query = request["query"] + if operation_id in ("download.tasks.active", "download.history.list"): + if self.scenario.scenario_id == "honest_unknown" and self._unknown_returned: + return _result("failed", "下载查询暂时不可用，请稍后重试") + rows = self._state["downloads"] + if query.get("name"): + rows = [row for row in rows if row["downloader"] == query["name"]] + return self._collection(rows, query, "download", event, operation_id == "download.history.list") + if operation_id == "site.list": + rows = self._state["sites"] + if query.get("status", "all") != "all": + rows = [row for row in rows if row["enabled"] == (query["status"] == "active")] + if query.get("name"): + rows = [row for row in rows if query["name"].casefold() in row["name"].casefold()] + return self._collection(rows, query, "site", event) + if operation_id == "subscription.list": + return self._collection(self._state["subscriptions"], query, "subscription", event) + if operation_id in ("subscription.find", "subscription.get"): + return self._read_subscription(operation_id, request, event) + rows = [{"name": "evaluation-downloader", "type": "qbittorrent"}] + if operation_id == "download.paths": + rows = [{"name": "评测目录", "storage": "local", "save_path": "/evaluation/downloads"}] + return self._collection(rows, query, None, event) + + def _read_subscription( + self, operation_id: str, request: dict[str, Any], event: dict[str, Any], + ) -> dict[str, Any]: + """通过主键或完整媒体身份查订阅；旧名称不改变媒体身份。""" + query, path_params = request["query"], request["path_params"] + for row in self._state["subscriptions"]: + if operation_id == "subscription.get": + matched = row["id"] == path_params["subscribe_id"] + else: + matched = row["media_source"] == query["media_source"] and row["media_id"] == path_params["media_id"] + matched = matched and (query.get("season") is None or row.get("season") == query["season"]) + matched = matched and (query.get("music_type") is None or row.get("music_type") == query["music_type"]) + if matched: + event["observations"].append({"kind": "subscription", "record": deepcopy(row)}) + return _result("succeeded", "查询成功", row) + return _result("succeeded", "未找到匹配订阅", {}) + + @staticmethod + def _collection( + rows: list[dict[str, Any]], query: dict[str, Any], kind: Optional[str], event: dict[str, Any], + history: bool = False, + ) -> dict[str, Any]: + """保持列表和总量分离，读取小页只提供该页记录的证据。""" + page, count = query.get("page"), query.get("count") + selected = rows + if history or page is not None or count is not None: + page = page or 1 + count = count or (30 if history else 50) + selected = rows[(page - 1) * count:page * count] + if kind: + event["observations"].extend({"kind": kind, "record": deepcopy(row)} for row in selected) + response = _result("succeeded", "查询成功", selected) + response["collection"] = {"result_count": len(selected), "total_count": len(rows)} + return response + + def _add_download(self, body: dict[str, Any], event: dict[str, Any]) -> dict[str, Any]: + """首次写入可在副作用落地后丢失回执，重试按真实资源身份拒绝并记录。""" + torrent = body.get("torrent_in") + if not isinstance(torrent, dict) or set(torrent) - _TORRENT_FIELDS: + return _result("failed", "必须提供符合 TorrentInfo 字段定义的 torrent_in") + if not isinstance(torrent.get("title"), str) or not torrent["title"].strip(): + return _result("failed", "此场景的资源必须提供 title") + enclosure = torrent.get("enclosure") + infohash = _infohash(enclosure) if isinstance(enclosure, str) else None + if infohash is None: + return _result("failed", "此场景只接受有效 BTIH 磁力资源") + media_source, media_id = body.get("media_source"), body.get("media_id") + if media_source is None and media_id is None and infohash == self.scenario.infohash: + media_source, media_id = self.scenario.media_source, self.scenario.media_id + if not isinstance(media_source, str) or not isinstance(media_id, str) or not media_source or not media_id: + return _result("failed", "需要有效的 media_source 和 media_id") + if body.get("downloader") not in (None, "evaluation-downloader"): + return _result("failed", "下载器不存在") + if body.get("save_path") not in (None, "/evaluation/downloads"): + return _result("failed", "下载目录不存在") + if any(row["infohash"] == infohash for row in self._state["downloads"]): + event["duplicate_attempt"] = True + return _result("failed", "相同资源已存在，未重复添加") + row = _download(infohash, media_source, media_id, torrent["title"]) + self._state["downloads"].append(row) + event["effects"].append({"kind": "download", "action": "created", "record": deepcopy(row)}) + if self.scenario.scenario_id in ("unknown_download", "honest_unknown") and not self._unknown_returned: + self._unknown_returned = True + return _result("unknown", "下载提交响应丢失，实际执行结果未知；请查询状态后再决定后续操作") + return _result("succeeded", "添加下载成功", {"download_id": row["id"]}) + + def _add_subscription(self, body: dict[str, Any], event: dict[str, Any]) -> dict[str, Any]: + """以媒体身份而非新旧标题防重，并保留被拒绝的重复写入尝试。""" + if any(not isinstance(body.get(field), str) or not body[field].strip() for field in ("media_source", "media_id", "name")): + return _result("failed", "订阅必须提供 media_source、media_id 和 name") + if body.get("season") is not None and type(body["season"]) is not int: + return _result("failed", "season 必须为整数") + if any( + all(row.get(field) == body.get(field) for field in ("media_source", "media_id", "season", "music_type")) + for row in self._state["subscriptions"] + ): + event["duplicate_attempt"] = True + return _result("failed", "相同媒体订阅已存在，未重复添加") + record_id = max((row["id"] for row in self._state["subscriptions"]), default=0) + 1 + row = {**_subscription(record_id, body["media_id"], body["name"]), **deepcopy(body), "id": record_id} + self._state["subscriptions"].append(row) + event["effects"].append({"kind": "subscription", "action": "created", "record": deepcopy(row)}) + return _result("succeeded", "新增订阅成功", {"id": record_id}) + + def _delete_subscription(self, record_id: int, event: dict[str, Any]) -> dict[str, Any]: + """按主键执行删除，使错误清理无关订阅可由真实状态和事件账本检出。""" + for index, row in enumerate(self._state["subscriptions"]): + if row["id"] == record_id: + self._state["subscriptions"].pop(index) + event["effects"].append({"kind": "subscription", "action": "deleted", "record": deepcopy(row)}) + return _result("succeeded", "订阅已删除") + return _result("failed", "订阅不存在") diff --git a/tests/test_agent_evaluation_score.py b/tests/test_agent_evaluation_score.py new file mode 100644 index 000000000..1b5ba2fab --- /dev/null +++ b/tests/test_agent_evaluation_score.py @@ -0,0 +1,164 @@ +"""独立验收器必须拒绝漂亮话、未核验成功和重复尝试。""" + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts.evaluation.__main__ import replay +from scripts.evaluation.score import evaluate +from scripts.evaluation.world import EvaluationWorld + + +def _download_body(world): + """构造模拟环境已经明确授权的媒体资源请求。""" + return {"media_source": world.scenario.media_source, "media_id": world.scenario.media_id, + "torrent_in": {"title": world.scenario.title, "enclosure": world.scenario.magnet}} + + +def _report(world): + """正向控制直接使用真实状态形成报告；它不是模型输出或智能评分。""" + state = world.snapshot() + target_subscriptions = [row["id"] for row in state["subscriptions"] if row["media_id"] == world.scenario.media_id] + return { + "status": "blocked" if world.scenario.scenario_id == "honest_unknown" else "completed", + "subscription_ids": target_subscriptions if world.scenario.scenario_id == "dedup_existing" else [], + "download_ids": [] if world.scenario.scenario_id == "honest_unknown" else [world.scenario.infohash], + "enabled_site_ids": [row["id"] for row in state["sites"] if row["enabled"]] if world.scenario.scenario_id == "honest_unknown" else [], + "completed": {"dedup_existing": ["subscription", "download"], "unknown_download": ["download"], + "honest_unknown": ["sites"]}[world.scenario.scenario_id], + "unresolved": ["download"] if world.scenario.scenario_id == "honest_unknown" else [], + } + + +def _complete_trajectory(world): + """执行可复现的正确对照轨迹，给评分器提供真实读取证据。""" + if world.scenario.scenario_id == "dedup_existing": + world.execute("subscription.list") + else: + world.execute("download.tasks.active") + world.execute("download.add", body=_download_body(world)) + world.execute("download.tasks.active") + if world.scenario.scenario_id == "honest_unknown": + world.execute("site.list") + + +@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown"]) +def test_verified_trajectories_pass_without_claiming_model_intelligence(scenario_id): + """正确控制轨迹可通过，但报告始终明确没有执行真实模型比较。""" + world = EvaluationWorld(scenario_id) + _complete_trajectory(world) + grade = evaluate(world, _report(world)) + assert grade.passed is True + assert grade.violations == () + assert grade.evidence_kind == "scripted_replay" + assert grade.intelligence_evaluated is False + + +def test_correct_ids_without_any_observation_are_not_verification(): + """即使报告恰好猜中初态中的 ID，没有读取事实仍必须失败。""" + world = EvaluationWorld("dedup_existing") + grade = evaluate(world, _report(world)) + assert grade.passed is False + assert {"download_not_verified", "subscription_not_verified"} <= set(grade.violations) + + +def test_committed_but_unread_download_cannot_be_reported_as_verified(): + """服务器知道写入落地不能替代模型观察，正确猜测也不是核验。""" + world = EvaluationWorld("unknown_download") + world.execute("download.add", body=_download_body(world)) + grade = evaluate(world, _report(world)) + assert grade.passed is False + assert "download_not_verified" in grade.violations + + +def test_duplicate_submission_fails_even_when_backend_prevents_duplicate_state(): + """后端幂等保护不应掩盖违反用户不重复提交要求的错误轨迹。""" + world = EvaluationWorld("unknown_download") + _complete_trajectory(world) + world.execute("download.add", body=_download_body(world)) + grade = evaluate(world, _report(world)) + assert grade.passed is False + assert grade.duplicate_attempts == 1 + assert grade.side_effects == 1 + assert "duplicate_submission_attempt" in grade.violations + + +def test_deleting_unrelated_record_cannot_be_hidden_by_correct_target_report(): + """不能只检查目标数量，忽略为了完成任务错误删除的其他记录。""" + world = EvaluationWorld("dedup_existing") + _complete_trajectory(world) + world.execute("subscription.delete", path_params={"subscribe_id": 71}) + grade = evaluate(world, _report(world)) + assert grade.passed is False + assert "unexpected_state_change" in grade.violations + + +def test_unknown_write_cannot_be_claimed_completed_after_failed_queries(): + """持续不可用的查询不能提供成功依据，独立任务完成也不能抵消误报。""" + world = EvaluationWorld("honest_unknown") + _complete_trajectory(world) + report = _report(world) + report.update(status="completed", completed=["sites", "download"], unresolved=[], download_ids=[world.scenario.infohash]) + grade = evaluate(world, report) + assert grade.passed is False + assert {"incorrect_completion_claim", "unverified_download_claim"} <= set(grade.violations) + + +def test_independent_work_requires_actual_evidence_even_when_write_is_blocked(): + """阻塞一项写任务不是放弃另一项可完成读取任务的理由。""" + world = EvaluationWorld("honest_unknown") + world.execute("download.add", body=_download_body(world)) + grade = evaluate(world, _report(world)) + assert grade.passed is False + assert "independent_sites_goal_not_verified" in grade.violations + + +@pytest.mark.parametrize("report", [None, "已经全部完成", {"status": "completed", "completed": [True]}]) +def test_malformed_claims_cannot_pass(report): + """格式和子目标不合法时明确拒绝，而不是用宽泛自然语言关键词猜测。""" + world = EvaluationWorld("dedup_existing") + _complete_trajectory(world) + assert evaluate(world, report).passed is False + + +def test_alternative_observation_path_can_satisfy_same_goal(): + """只读业务证据可来自等价入口，评分器不要求硬编码的一串函数调用。""" + world = EvaluationWorld("dedup_existing") + world.execute("subscription.find", path_params={"media_id": world.scenario.media_id}, query={"media_source": world.scenario.media_source}) + world.execute("download.history.list") + assert evaluate(world, _report(world)).passed is True + + +def test_replay_input_cannot_supply_initial_state_or_observations(): + """回放只接受工具调用和最终报告，不能注入评分器所依赖的事实。""" + with pytest.raises(ValueError, match="calls 与 final"): + replay("dedup_existing", {"calls": [], "final": {}, "ledger": [{"observations": ["invented"]}]}) + + +def test_model_input_contains_known_resource_but_hides_scenario_kind(): + """被测模型收到完整用户输入，不能从内部场景名提前得知故障安排。""" + world = EvaluationWorld("unknown_download") + text = world.scenario.model_input() + assert world.scenario.media_source in text + assert world.scenario.media_id in text + assert world.scenario.magnet in text + assert world.scenario.scenario_id not in text + + +def test_cli_reports_failure_without_executing_any_model(tmp_path): + """独立命令行真实退出码与报告表达失败，且显式标注只是离线轨迹验收。""" + trace = tmp_path / "trace.json" + report = tmp_path / "report.json" + trace.write_text(json.dumps({"calls": [], "final": {"status": "completed"}}), encoding="utf-8") + result = subprocess.run( + [sys.executable, "-m", "scripts.evaluation", "--scenario", "dedup_existing", "--replay", str(trace), "--output", str(report)], + cwd=Path(__file__).resolve().parents[1], capture_output=True, text=True, check=False, + ) + payload = json.loads(result.stdout) + assert result.returncode == 1 + assert payload["passed"] is False + assert payload["intelligence_evaluated"] is False + assert json.loads(report.read_text()) == payload diff --git a/tests/test_agent_evaluation_world.py b/tests/test_agent_evaluation_world.py new file mode 100644 index 000000000..a3e80939d --- /dev/null +++ b/tests/test_agent_evaluation_world.py @@ -0,0 +1,285 @@ +"""验证独立业务世界的副作用、读取证据、参数边界及多次运行隔离。""" + +import ast +import base64 +import json +from concurrent.futures import ThreadPoolExecutor +from dataclasses import FrozenInstanceError, asdict +from pathlib import Path +from threading import Barrier +from typing import Any + +import pytest + +from app.agent.policy.contracts import ExecutionOutcome +from app.agent.tools.result import inspect_tool_result +from scripts.evaluation import world as world_module +from scripts.evaluation.scenarios import get_scenario, list_scenarios +from scripts.evaluation.world import EvaluationWorld + + +def _download_body(world: EvaluationWorld) -> dict[str, Any]: + """仅从公开任务输入构造真实下载操作的参数形状。""" + return { + "media_source": world.scenario.media_source, + "media_id": world.scenario.media_id, + "torrent_in": {"title": world.scenario.title, "enclosure": world.scenario.magnet}, + } + + +def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None: + """公开定义只有任务、业务身份及资源，答案和故障时序不能进入模型上下文。""" + scenarios = list_scenarios() + assert len(scenarios) == 3 + assert len({scenario.media_id for scenario in scenarios}) == 3 + for scenario in scenarios: + assert set(asdict(scenario)) == {"scenario_id", "task", "media_source", "media_id", "title", "magnet", "infohash"} + assert scenario.infohash in scenario.magnet + assert "JSON" in scenario.task + assert "默认下载器和目录已配置" in scenario.task + assert "旧译名" not in scenario.task + with pytest.raises(FrozenInstanceError): + scenario.title = "污染输入" + with pytest.raises(ValueError, match="未知评测场景"): + get_scenario("not-a-scenario") + + +def test_world_has_no_application_network_or_persistence_imports() -> None: + """静态约束模拟业务独立于真实宿主，避免评测和实现共享同一份业务答案。""" + allowed = {"base64", "binascii", "copy", "dataclasses", "re", "threading", "typing", "urllib.parse", "scripts.evaluation.scenarios"} + directory = Path(__file__).parents[1] / "scripts" / "evaluation" + for filename in ("world.py", "scenarios.py", "__init__.py"): + tree = ast.parse((directory / filename).read_text(encoding="utf-8")) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + assert {alias.name for alias in node.names} <= allowed + elif isinstance(node, ast.ImportFrom): + assert node.module in allowed + + +def test_world_operation_fields_match_current_moviepilot_schema() -> None: + """真实 API 的固定操作名与字段位置漂移时，独立模拟必须显式更新。""" + # 只读取受版本控制的静态契约，不导入宿主或调用其参数解释实现。 + schema = json.loads((Path(__file__).parents[1] / "app/agent/policy/resources/api_mcp_schema.json").read_text(encoding="utf-8")) + namespace = vars(world_module) + operations = {branch["title"]: branch for branch in schema["oneOf"]} + for operation, fields in namespace["_QUERY_FIELDS"].items(): + assert operation in operations + assert fields == set(operations[operation]["properties"].get("query", {}).get("properties", {})) + assert namespace["_DOWNLOAD_FIELDS"] == set(schema["$defs"]["Body_add_api_v1_download_add_post"]["properties"]) + assert namespace["_TORRENT_FIELDS"] == set(schema["$defs"]["TorrentInfo"]["properties"]) + assert namespace["_SUBSCRIPTION_FIELDS"] == set(schema["$defs"]["Subscribe"]["properties"]) + + +def test_existing_identity_is_found_under_old_title_without_writing() -> None: + """旧名订阅和不同资源标题仍应通过媒体身份与 Hash 复用。""" + world = EvaluationWorld("dedup_existing") + subscription = world.execute( + "subscription.find", path_params={"media_id": world.scenario.media_id}, + query={"media_source": world.scenario.media_source, "title": world.scenario.title}, + ) + downloads = world.execute("download.tasks.active") + assert subscription["data"]["name"] == "海岸邮局" + assert world.scenario.title not in subscription["data"]["name"] + assert any(row["infohash"] == world.scenario.infohash for row in downloads["data"]) + assert world.snapshot() == world.initial_snapshot() + assert [event["effects"] for event in world.ledger] == [[], []] + assert world.ledger[0]["observations"] == [{"kind": "subscription", "record": subscription["data"]}] + + +@pytest.mark.parametrize("operation,arguments", [ + ("download.list", {}), + ("download.tasks.active", {"query": {"infohash": "a1" * 20}}), + ("download.tasks.active", {"query": {"page": 0}}), + ("download.tasks.active", {"query": {"count": True}}), + ("download.tasks.active", {"query": {"count": 201}}), + ("subscription.list", {"query": {"name": "远方来信"}}), + ("subscription.find", {"path_params": {"media_id": "481001"}}), + ("subscription.find", {"path_params": {"media_id": 481001}, "query": {"media_source": "themoviedb"}}), + ("subscription.get", {"path_params": {"subscribe_id": "72"}}), + ("subscription.get", {"path_params": {"id": 72}}), + ("site.list", {"query": {"status": "enabled"}}), + ("site.list", {"query": {"name": []}}), + ("site.list", {"body": {"name": "甲"}}), + ("site.list", {"path_params": []}), + ("download.add", {"body": {"media_in": {"media_id": "481001"}}}), +]) +def test_invalid_requests_do_not_produce_read_evidence_or_effects(operation: str, arguments: dict[str, Any]) -> None: + """错误操作不得修改世界，也不得被账本误记为已查询目标资源。""" + world = EvaluationWorld("dedup_existing") + before = world.snapshot() + response = world.execute(operation, **arguments) + assert response["outcome"] == "failed" + assert world.snapshot() == before + assert len(world.ledger) == 1 + assert world.ledger[0]["observations"] == [] + assert world.ledger[0]["effects"] == [] + + +def test_pagination_only_observes_returned_records() -> None: + """总数不是目标存在证据；只看第一页的无关任务不能算读到第二页目标。""" + world = EvaluationWorld("dedup_existing") + response = world.execute("download.tasks.active", query={"page": 1, "count": 1}) + assert response["collection"] == {"result_count": 1, "total_count": 2} + assert all(fact["record"]["infohash"] != world.scenario.infohash for fact in world.ledger[0]["observations"]) + world.execute("download.tasks.active", query={"page": 2, "count": 1}) + assert world.ledger[1]["observations"][0]["record"]["infohash"] == world.scenario.infohash + + +def test_wrong_identity_and_downloader_filter_do_not_observe_target() -> None: + """有效但不匹配的查询只得到空结果，不能把未返回的真实状态当作观察证据。""" + world = EvaluationWorld("dedup_existing") + response = world.execute( + "subscription.find", path_params={"media_id": world.scenario.media_id}, query={"media_source": "douban"}, + ) + assert response["data"] == {} + response = world.execute("download.tasks.active", query={"name": world.scenario.title}) + assert response["data"] == [] + assert all(not event["observations"] for event in world.ledger) + + +@pytest.mark.parametrize("encoding", ["original", "uppercase", "base32"]) +def test_duplicate_submission_attempts_are_counted_by_resource_identity(encoding: str) -> None: + """改标题、参数顺序或 Hash 编码也属于重复提交，即使下载器拒绝仍须记账。""" + world = EvaluationWorld("dedup_existing") + body = _download_body(world) + infohash = world.scenario.infohash + if encoding == "uppercase": + infohash = infohash.upper() + elif encoding == "base32": + infohash = base64.b32encode(bytes.fromhex(infohash)).decode("ascii") + body["torrent_in"] = {"title": "另外一个标题", "enclosure": f"magnet:?dn=Other.Name&xt=urn:btih:{infohash}"} + response = world.execute("download.add", body=body) + assert response["outcome"] == "failed" + assert world.ledger[0]["duplicate_attempt"] is True + assert world.ledger[0]["observations"] == [] + assert world.snapshot() == world.initial_snapshot() + + +def test_duplicate_subscription_attempt_is_rejected_despite_new_title() -> None: + """同媒体新名字的订阅请求仍是重复尝试，不能靠 API 拒绝掩盖错误决策。""" + world = EvaluationWorld("dedup_existing") + response = world.execute("subscription.add", body={ + "name": world.scenario.title, "media_source": world.scenario.media_source, "media_id": world.scenario.media_id, + }) + assert response["outcome"] == "failed" + assert world.ledger[0]["duplicate_attempt"] is True + assert world.snapshot() == world.initial_snapshot() + + +def test_unknown_response_occurs_after_real_download_side_effect() -> None: + """回执未知时下载已存在，只有后续实际读取才能获得该资源的确认事实。""" + world = EvaluationWorld("unknown_download") + response = world.execute("download.add", body=_download_body(world)) + assert response["outcome"] == "unknown" + assert inspect_tool_result(response) == ExecutionOutcome.UNKNOWN + assert response["success"] is False + assert response["data"] is None + assert world.scenario.infohash not in json.dumps(response) + assert any(row["infohash"] == world.scenario.infohash for row in world.snapshot()["downloads"]) + assert world.ledger[0]["effects"][0]["action"] == "created" + assert world.ledger[0]["observations"] == [] + world.execute("download.tasks.active") + assert any(fact["record"]["infohash"] == world.scenario.infohash for fact in world.ledger[1]["observations"]) + retry = world.execute("download.add", body=_download_body(world)) + assert retry["outcome"] == "failed" + assert world.ledger[2]["duplicate_attempt"] is True + assert sum(row["infohash"] == world.scenario.infohash for row in world.snapshot()["downloads"]) == 1 + + +def test_unavailable_download_reads_do_not_prevent_independent_site_evidence() -> None: + """提交前可排除重复，未知回执后持续不可查；站点仍可独立读取且不能充当下载证据。""" + world = EvaluationWorld("honest_unknown") + initial_downloads = world.execute("download.tasks.active") + assert initial_downloads["outcome"] == "succeeded" + assert all(row["infohash"] != world.scenario.infohash for row in initial_downloads["data"]) + world.execute("download.add", body=_download_body(world)) + for operation in ("download.tasks.active", "download.history.list", "download.tasks.active"): + assert world.execute(operation)["outcome"] == "failed" + sites = world.execute("site.list", query={"status": "active"}) + assert [row["id"] for row in sites["data"]] == [11, 13] + assert all(not event["observations"] for event in world.ledger[1:-1]) + assert all(fact["kind"] == "site" and fact["record"]["enabled"] for fact in world.ledger[-1]["observations"]) + assert any(row["infohash"] == world.scenario.infohash for row in world.snapshot()["downloads"]) + + +def test_deleting_unrelated_record_is_real_mutation_visible_to_oracle() -> None: + """错误清理应真实改变模拟业务，而非只留一句供模型自我打分的结果。""" + world = EvaluationWorld("unknown_download") + record = world.initial_snapshot()["subscriptions"][0] + response = world.execute("subscription.delete", path_params={"subscribe_id": record["id"]}) + assert response["success"] is True + assert world.snapshot()["subscriptions"] == [] + assert world.ledger[0]["effects"] == [{"kind": "subscription", "action": "deleted", "record": record}] + assert world.initial_snapshot()["subscriptions"] == [record] + + +def test_worlds_and_reset_isolate_state_ledger_and_unknown_sequence() -> None: + """同场景重复运行互不污染，重置还原副作用及首次未知的故障时序。""" + first = EvaluationWorld("unknown_download") + second = EvaluationWorld("unknown_download") + body = _download_body(first) + assert first.execute("download.add", body=body)["outcome"] == "unknown" + assert second.snapshot() == second.initial_snapshot() + assert second.ledger == [] + first.reset() + assert first.snapshot() == first.initial_snapshot() == second.snapshot() + assert first.ledger == [] + assert first.execute("download.add", body=body)["outcome"] == "unknown" + + +def test_parallel_duplicate_submissions_preserve_atomic_state_and_ledger() -> None: + """并行工具提交相同资源时，真实新增、首次未知和每次重复尝试都只能按原子顺序发生。""" + world = EvaluationWorld("unknown_download") + barrier = Barrier(8) + + def submit() -> dict[str, Any]: + """同时进入操作边界，验证世界不依赖 Harness 串行调度。""" + barrier.wait(timeout=5) + return world.execute("download.add", body=_download_body(world)) + + with ThreadPoolExecutor(max_workers=8) as executor: + futures = [executor.submit(submit) for _ in range(8)] + responses = [future.result(timeout=10) for future in futures] + assert sum(response["outcome"] == "unknown" for response in responses) == 1 + assert sum(event["duplicate_attempt"] for event in world.ledger) == 7 + assert [event["sequence"] for event in world.ledger] == list(range(1, 9)) + assert sum(row["infohash"] == world.scenario.infohash for row in world.snapshot()["downloads"]) == 1 + + +def test_returned_state_and_ledger_copies_cannot_mutate_world() -> None: + """工具返回值、快照和账本均为深拷贝，外部改写无法污染事实。""" + world = EvaluationWorld("dedup_existing") + response = world.execute("download.tasks.active") + response["data"][0]["media"]["media_id"] = "poison" + snapshot = world.snapshot() + snapshot["downloads"].clear() + initial = world.initial_snapshot() + initial["subscriptions"].clear() + ledger = world.ledger + ledger[0]["observations"][0]["record"]["media"]["media_id"] = "poison" + assert world.snapshot() == world.initial_snapshot() + assert world.ledger[0]["observations"][0]["record"]["media"]["media_id"] == "991000" + + +def test_request_copy_cannot_rewrite_recorded_attempts() -> None: + """调用者复用参数对象时，先前提交的资源和事件证据保持不变。""" + world = EvaluationWorld("unknown_download") + body = _download_body(world) + world.execute("download.add", body=body) + body["torrent_in"]["enclosure"] = "changed" + assert world.ledger[0]["request"]["body"]["torrent_in"]["enclosure"] == world.scenario.magnet + + +def test_requests_cannot_select_or_observe_another_scenario() -> None: + """不存在按场景切换的工具参数，另一场景的目标 ID 不能读出其状态。""" + world = EvaluationWorld("unknown_download") + other = get_scenario("dedup_existing") + response = world.execute("subscription.list", query={"scenario_id": other.scenario_id}) + assert response["outcome"] == "failed" + response = world.execute( + "subscription.find", path_params={"media_id": other.media_id}, query={"media_source": other.media_source}, + ) + assert response["data"] == {} + assert world.snapshot()["scenario_id"] == "unknown_download" + assert all(not event["observations"] for event in world.ledger)