diff --git a/.claude/skills/replay-eval/scripts/claude_runtime.py b/.claude/skills/replay-eval/scripts/claude_runtime.py index 07cd8da..c11457b 100644 --- a/.claude/skills/replay-eval/scripts/claude_runtime.py +++ b/.claude/skills/replay-eval/scripts/claude_runtime.py @@ -54,6 +54,17 @@ MODEL_ID_PATTERN = re.compile( r"^(?=.{6,128}$)[A-Za-z0-9][A-Za-z0-9._:-]{5,127}(?:\[[A-Za-z0-9._:-]+\])?$" ) MONEY_QUANTUM = Decimal("0.000001") +USAGE_REQUIRED_COUNT_FIELDS = frozenset({"input_tokens", "output_tokens"}) +USAGE_OPTIONAL_COUNT_FIELDS = frozenset( + {"cache_creation_input_tokens", "cache_read_input_tokens"} +) +USAGE_COUNT_MAP_FIELDS = { + "server_tool_use": frozenset({"web_search_requests", "web_fetch_requests"}), + "cache_creation": frozenset( + {"ephemeral_5m_input_tokens", "ephemeral_1h_input_tokens"} + ), +} +USAGE_STRING_FIELDS = frozenset({"service_tier", "speed", "inference_geo"}) class ClaudeRuntimeError(RuntimeError): @@ -503,7 +514,7 @@ def _minimal_environment( def _validate_number_tree(value: Any, path: str) -> None: - """递归校验 usage 中的数值均有限且非负。""" + """递归校验严格数字树，用于 modelUsage 的计数核账。""" if isinstance(value, Mapping): if not value: @@ -520,6 +531,89 @@ def _validate_number_tree(value: Any, path: str) -> None: raise ValueError(f"{path} 数值非法") +def _validate_non_negative_finite_number(value: Any, path: str) -> None: + """校验计数字段是非负有限数，不接受 bool、NaN 或 Infinity。""" + + if isinstance(value, bool) or not isinstance(value, (int, float, Decimal)): + raise ValueError(f"{path} 必须是非负有限数") + try: + decimal_value = Decimal(str(value)) + except (InvalidOperation, ValueError) as exc: + raise ValueError(f"{path} 必须是非负有限数") from exc + if not decimal_value.is_finite() or decimal_value < 0: + raise ValueError(f"{path} 必须是非负有限数") + + +def _validate_safe_json(value: Any, path: str) -> None: + """递归校验前向兼容字段仍然只包含安全 JSON 值。""" + + if value is None or isinstance(value, (str, bool, int)): + return + if isinstance(value, (float, Decimal)): + try: + decimal_value = Decimal(str(value)) + except (InvalidOperation, ValueError) as exc: + raise ValueError(f"{path} 不是安全 JSON") from exc + if not decimal_value.is_finite(): + raise ValueError(f"{path} 不是安全 JSON") + return + if isinstance(value, list): + for index, item in enumerate(value): + _validate_safe_json(item, f"{path}[{index}]") + return + if isinstance(value, Mapping): + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError(f"{path} 对象键不是字符串") + _validate_safe_json(item, f"{path}.{key}") + return + raise ValueError(f"{path} 不是安全 JSON") + + +def _validate_usage_count_mapping( + value: Any, path: str, known_count_fields: frozenset[str] +) -> None: + """校验已知计数映射，并允许未来字段继续使用安全 JSON。""" + + if not isinstance(value, Mapping): + raise ValueError(f"{path} 必须是对象") + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError(f"{path} 对象键不是字符串") + if key in known_count_fields: + _validate_non_negative_finite_number(item, f"{path}.{key}") + else: + _validate_safe_json(item, f"{path}.{key}") + + +def _validate_usage(value: Any, path: str = "usage") -> None: + """精确校验 usage 的权威计数，同时兼容安全的未来 metadata。""" + + if not isinstance(value, Mapping) or not value: + raise ValueError(f"{path} 必须是非空对象") + missing_fields = USAGE_REQUIRED_COUNT_FIELDS - set(value) + if missing_fields: + raise ValueError(f"{path} 缺少必需计数字段") + + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError(f"{path} 对象键不是字符串") + field_path = f"{path}.{key}" + if key in USAGE_REQUIRED_COUNT_FIELDS or key in USAGE_OPTIONAL_COUNT_FIELDS: + _validate_non_negative_finite_number(item, field_path) + elif key in USAGE_COUNT_MAP_FIELDS: + _validate_usage_count_mapping(item, field_path, USAGE_COUNT_MAP_FIELDS[key]) + elif key in USAGE_STRING_FIELDS: + if not isinstance(item, str) or (key != "inference_geo" and not item): + raise ValueError(f"{field_path} 必须是字符串") + elif key == "iterations": + if not isinstance(item, list): + raise ValueError(f"{field_path} 必须是数组") + _validate_safe_json(item, field_path) + else: + _validate_safe_json(item, field_path) + + def _schema_type_matches(value: Any, expected: str) -> bool: """按 JSON 类型语义判断 Python 值，显式排除 bool 伪装整数。""" @@ -775,7 +869,7 @@ def run_claude( issues.add(f"{prefix}_BUDGET_EXCEEDED") try: - _validate_number_tree(envelope.get("usage"), "usage") + _validate_usage(envelope.get("usage")) except (ValueError, InvalidOperation): issues.add(f"{prefix}_RECEIPT_INVALID") diff --git a/.claude/skills/replay-eval/scripts/test_claude_runtime.py b/.claude/skills/replay-eval/scripts/test_claude_runtime.py index 5c214f0..a87733f 100644 --- a/.claude/skills/replay-eval/scripts/test_claude_runtime.py +++ b/.claude/skills/replay-eval/scripts/test_claude_runtime.py @@ -3,6 +3,7 @@ from __future__ import annotations +import copy import hashlib import json import os @@ -20,6 +21,7 @@ from claude_runtime import ( # noqa: E402 ClaudeRuntimeError, ExecutionProfile, _minimal_environment, + _validate_usage, build_sandbox_command, canonical_json, run_claude, @@ -95,6 +97,27 @@ def success_envelope() -> dict[str, object]: } +def realistic_usage() -> dict[str, object]: + """构造包含 Claude metadata、已知计数映射和迭代数组的真实 usage 形状。""" + + return { + "input_tokens": 10, + "output_tokens": 2, + "cache_creation_input_tokens": 3, + "cache_read_input_tokens": 4, + "server_tool_use": {"web_search_requests": 1, "web_fetch_requests": 0}, + "cache_creation": {"ephemeral_5m_input_tokens": 5, "ephemeral_1h_input_tokens": 6}, + "service_tier": "standard", + "speed": "standard", + "inference_geo": "", + "iterations": [ + {"type": "tool_use", "duration_ms": 12}, + {"type": "metadata", "values": [True, None, "safe"]}, + ], + "future_metadata": {"trace_id": "safe", "enabled": True}, + } + + class FakeRunner: """记录 subprocess 参数并返回预设结果，测试绝不调用真实模型。""" @@ -210,6 +233,15 @@ class ClaudeRuntimeTest(unittest.TestCase): self.assertEqual(pathlib.Path(str(kwargs["cwd"])), pathlib.Path(kwargs["env"]["HOME"])) self.assertFalse(pathlib.Path(str(kwargs["cwd"])).exists()) + with tempfile.TemporaryDirectory(dir="/private/tmp") as directory: + real_auth_environment = _minimal_environment( + {"ANTHROPIC_AUTH_TOKEN": auth_token, "ANTHROPIC_BASE_URL": base_url}, + pathlib.Path(directory), + require_authentication=True, + ) + self.assertEqual(real_auth_environment["ANTHROPIC_AUTH_TOKEN"], auth_token) + self.assertEqual(real_auth_environment["ANTHROPIC_BASE_URL"], base_url) + def test_auth_token_requires_a_safe_base_url(self): """AUTH_TOKEN 必须绑定无凭据、无查询和无片段的 HTTP(S) 网关地址。""" @@ -242,6 +274,109 @@ class ClaudeRuntimeTest(unittest.TestCase): self.assertNotIn(base_url, str(raised.exception)) self.assertNotIn(base_url, canonical_json(raised.exception.details)) + def test_realistic_usage_metadata_is_accepted_without_changing_cost_authority(self): + """真实 usage metadata 可通过，但模型成本仍只由 modelUsage 对账。""" + + envelope = success_envelope() + envelope["usage"] = realistic_usage() + result = run_claude( + execution_profile(), + {"request": "x"}, + runner=runner_for(envelope), + binding_verifier=lambda _profile: None, + ) + + self.assertEqual(result.structured_output, {"value": "ok"}) + self.assertEqual(result.receipt.as_dict()["totalCostUsd"], "0.120000") + self.assertEqual(result.receipt.as_dict()["usage"], realistic_usage()) + + def test_usage_rejects_bad_counts_metadata_and_non_json_iterations(self): + """usage 的必需计数、已知映射和 metadata 破坏时必须失败关闭。""" + + cases: list[tuple[str, dict[str, object]]] = [] + + missing_required = realistic_usage() + del missing_required["input_tokens"] + cases.append(("missing input_tokens", missing_required)) + + negative_token = realistic_usage() + negative_token["output_tokens"] = -1 + cases.append(("negative output_tokens", negative_token)) + + negative_server_count = realistic_usage() + negative_server_count["server_tool_use"] = {"web_search_requests": -1} + cases.append(("negative server_tool_use count", negative_server_count)) + + negative_cache_count = realistic_usage() + negative_cache_count["cache_creation"] = {"ephemeral_5m_input_tokens": -1} + cases.append(("negative cache_creation count", negative_cache_count)) + + invalid_service_tier = realistic_usage() + invalid_service_tier["service_tier"] = 1 + cases.append(("non-string service_tier", invalid_service_tier)) + + invalid_inference_geo = realistic_usage() + invalid_inference_geo["inference_geo"] = None + cases.append(("non-string inference_geo", invalid_inference_geo)) + + invalid_unknown_metadata = realistic_usage() + invalid_unknown_metadata["future_metadata"] = {"ratio": float("inf")} + cases.append(("infinite unknown metadata", invalid_unknown_metadata)) + + invalid_iteration_number = realistic_usage() + invalid_iteration_number["iterations"] = [{"duration_ms": float("nan")}] + cases.append(("non-finite iteration metadata", invalid_iteration_number)) + + for label, usage in cases: + envelope = success_envelope() + envelope["usage"] = usage + with self.subTest(label=label), self.assertRaises(ClaudeRuntimeError) as raised: + run_claude( + execution_profile(), + {"request": "x"}, + runner=runner_for(envelope), + binding_verifier=lambda _profile: None, + ) + self.assertEqual(raised.exception.primary_code, "WRITER_RECEIPT_INVALID") + + invalid_iteration_json = copy.deepcopy(realistic_usage()) + invalid_iteration_json["iterations"] = [object()] + with self.assertRaises(ValueError): + _validate_usage(invalid_iteration_json) + + def test_realistic_usage_cannot_bypass_model_cost_reconciliation(self): + """丰富 usage metadata 不能掩盖 modelUsage 与 total cost 对账失败。""" + + cases = [] + mismatched_model = success_envelope() + mismatched_model["usage"] = realistic_usage() + mismatched_model["modelUsage"] = { + "claude-sonnet-4-20250514": { + "inputTokens": 10, + "outputTokens": 2, + "costUSD": "0.120000", + } + } + cases.append(mismatched_model) + + mismatched_total = success_envelope() + mismatched_total["usage"] = realistic_usage() + mismatched_total["total_cost_usd"] = "0.130000" + cases.append(mismatched_total) + + for envelope in cases: + with self.assertRaises(ClaudeRuntimeError) as raised: + run_claude( + execution_profile(), + {"request": "x"}, + runner=runner_for(envelope), + binding_verifier=lambda _profile: None, + ) + self.assertIn( + raised.exception.primary_code, + {"WRITER_MODEL_MISMATCH", "WRITER_RECEIPT_INVALID"}, + ) + def test_api_key_and_oauth_remain_supported_and_exclusive(self): """API key/OAuth 仍可进入最小环境，多个认证字段仍然失败关闭。"""