e36cd57010
【评测合同与提示词】 - writer/detector/judge 评测提示词去支架化：删评测身份/输出格式叮嘱/盲化反向叮嘱/机器字段 （改由运行时 --json-schema/--tools ""/--strict-mcp-config/输入设计强制），装回各角色本业纪律 - writer 加写作纪律：戏剧化节拍禁抄细纲概述句、具体压倒抽象、每场戏三件套、篇幅用场景写够不注水不缩水 - 篇幅自纠环修订指令区分偏短（加场景）/偏长（删冗余）；机械门要求清单（章末钩子/硬事件/角色/伏笔）注入写手硬约束 - 引文校验从「必须恰好1次」放宽为「0次失败关闭、≥1次绑首次」（检测+盲评同口径） - llm 立 chat_governed 为主入口（chat 降为仅调试），clean_detect 改用 chat_governed 弃自带降级链 - rewrite/expansion/polish 输出合同对齐 writer.md（返回文本、写手不读写工作区、主会话落工作区） - 修复 confirm/replay-eval 探针两组坏自测（夹具适配现行合同、探针测试自包含不硬编码漂移哈希）；补 clean_detect 离线测试 - AGENTS.md 新增 §10「Agent 提示词与 Skill 审查标准」 【skill 三层重组：单向依赖 底座→能力→编排，断两环+修生产倒挂】 - Phase 0: 新建 runtime 底座（claude_runtime/file_cas/raw_vault），断环 C1、修 continuation 生产倒挂 - Phase 1: 评分尺+门判（writer_rubric/fine_outline_rubric/writer_gate/gate_input_builder）收进 quality-gate，断环 C2、名实相符 - Phase 2: 升格管线从 parse-book 独立成 upgrade skill（备份审计契约键名稳定、仅改路径定位） - Phase 3: replay-eval 瘦成纯编排 - read-context 共用簇（build_snapshot/audit_leakage/check_snapshot/load_reference_work）下沉到新 snapshot skill，消除能力层向上引用 - Phase 4: run_writer_replay.py（130KB）拆成包（_common/budget/authorization/sample/blind/execute），__init__ 全量 re-export 测试零改动 【base 配置】三角色 effort 提 high；提示词更新；探针重测绑定 writer 合同；预算 writer 45 次/总 450 美元（含篇幅修订） 全量离线测试 36 个文件全绿；函数逻辑零改动（仅搬位置/改 import/改文档，capture_code_identity 仅改路径定位）。
586 lines
26 KiB
Python
586 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""refresh_runtime_probe 的纯离线测试：假 invoker，不连库、不调模型、不连网。
|
||
|
||
覆盖任务要求的 a-e 五组场景：
|
||
a. 成功合规产出 → 探针记录与刷新后配置的 receiptSha256 都能用现有 _verify_self_hash 自检通过；
|
||
b. 刷新后配置喂给加固门 _validate_execute_authorization（用 writer profile 构建适配）→ 通过；
|
||
c. 自包含构造的旧探针(身份哈希相对当前合同末位翻转)被替换为新身份哈希(== writer profile 身份哈希)，不再等于旧值；
|
||
d. 失败关闭：schema 非法 / 超预算 / 调用失败 / 超 deadline / 回执不可信 → 不写 successful 探针、
|
||
不产出成功配置、退出非零；
|
||
e. 重签算法一致性：对刷新后的 runtimeProbe 用现有 _verify_self_hash 校验通过（逐字节同源）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import pathlib
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from typing import Any, Mapping
|
||
from unittest import mock
|
||
|
||
SCRIPT_DIR = pathlib.Path(__file__).resolve().parent
|
||
RUNTIME_DIR = SCRIPT_DIR.parents[1] / "runtime" / "scripts"
|
||
QUALITY_GATE_DIR = SCRIPT_DIR.parents[1] / "quality-gate" / "scripts"
|
||
if str(SCRIPT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(SCRIPT_DIR))
|
||
if str(RUNTIME_DIR) not in sys.path:
|
||
sys.path.insert(0, str(RUNTIME_DIR))
|
||
if str(QUALITY_GATE_DIR) not in sys.path:
|
||
sys.path.insert(0, str(QUALITY_GATE_DIR))
|
||
|
||
import refresh_runtime_probe as refresh_module # noqa: E402
|
||
from refresh_runtime_probe import ( # noqa: E402
|
||
PROBE_BUSINESS_INPUT,
|
||
ProbeRefreshError,
|
||
apply_probe,
|
||
build_probe_record,
|
||
build_writer_profile,
|
||
make_dry_run_invoker,
|
||
refresh_runtime_probe,
|
||
verify_probe_result,
|
||
)
|
||
import run_writer_replay as replay_module # noqa: E402
|
||
from run_writer_replay import ( # noqa: E402
|
||
WriterReplayProductionAdapters,
|
||
_profile_from_mapping,
|
||
_validate_execute_authorization,
|
||
)
|
||
from gate_input_builder import ( # noqa: E402
|
||
GateInputBuildError,
|
||
_verify_self_hash,
|
||
canonical_sha256,
|
||
)
|
||
from claude_runtime import ( # noqa: E402
|
||
HASH_PATTERN,
|
||
ClaudeInvocationResult,
|
||
ClaudeRuntimeError,
|
||
ExecutionProfile,
|
||
ExecutionReceipt,
|
||
sha256_json,
|
||
)
|
||
|
||
CONFIG_PATH = SCRIPT_DIR.parent / "configs" / "writer-gate-a-deep-space-v1.json"
|
||
CHECKED_AT = "2026-07-25T00:00:00+00:00"
|
||
EXPECTED_PROBE_FIELDS = {
|
||
"status",
|
||
"checkedAt",
|
||
"claudeExecutablePath",
|
||
"claudeExecutableSha256",
|
||
"claudeCliVersion",
|
||
"modelAlias",
|
||
"resolvedModelId",
|
||
"executionProfileSha256",
|
||
"executionReceiptSha256",
|
||
"structuredOutputSha256",
|
||
"terminalReason",
|
||
"totalCostUsd",
|
||
"modelMatch",
|
||
"exitCode",
|
||
"apiErrorStatus",
|
||
"receiptSha256",
|
||
}
|
||
|
||
|
||
def _load_base_config() -> dict[str, Any]:
|
||
return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _current_writer_identity_hash(config: Mapping[str, Any]) -> str:
|
||
"""动态算出当前 writer profile 的真实身份哈希。
|
||
|
||
WHY: 测试不再硬编码任何具体历史哈希。无论正式配置的探针刷成什么值，这里都用门同款
|
||
build_writer_profile 现场算出当前合同的身份哈希，作为「刷新后探针应等于」的基准。
|
||
"""
|
||
|
||
return build_writer_profile(config).execution_profile_sha256
|
||
|
||
|
||
def _make_stale_profile_hash(current_hash: str) -> str:
|
||
"""构造一个**确定与 current_hash 不同**的合法格式身份哈希（末位十六进制翻转）。
|
||
|
||
WHY: stale 探针不能硬编码某个历史值——正式配置一刷新，硬编码值就可能与现实重合而失效。
|
||
这里相对「当前身份哈希」现场构造：翻转最后一个十六进制字符，无论 current_hash 是什么，
|
||
结果都确定 != current_hash，且仍是合法的 `sha256:<64hex>` 格式，可被门按身份失配处理。
|
||
"""
|
||
|
||
prefix, _, hex_part = current_hash.partition(":")
|
||
flipped_last = "0" if hex_part[-1] != "0" else "1"
|
||
return f"{prefix}:{hex_part[:-1]}{flipped_last}"
|
||
|
||
|
||
def _make_stale_config(config: Mapping[str, Any]) -> tuple[dict[str, Any], str]:
|
||
"""深拷贝 base 配置，把 runtimeProbe.executionProfileSha256 换成确定失配的 stale 值。
|
||
|
||
WHY: 加固门（_validate_authorization_records）会先复核探针自身的 receiptSha256 自哈希，
|
||
再在最后一步校验 executionProfileSha256 合同绑定。若只改身份哈希不重算自哈希，探针会因
|
||
自哈希无效被提前以 EXECUTE_AUTHORIZATION_HASH_INVALID 挡下，到不了合同绑定那一关。
|
||
所以这里在替换身份哈希后重算 receiptSha256，让 stale 探针内部自洽、只在合同绑定关失配，
|
||
从而精准验证「旧探针被 EXECUTE_PROBE_CONTRACT_MISMATCH 挡下」。budget / rawRetention /
|
||
profileSha256 一律不动（它们各自自哈希仍有效，且 profileSha256.writer 仍等于当前身份）。
|
||
返回 (stale_config, stale_hash)。
|
||
"""
|
||
|
||
stale_config = copy.deepcopy(dict(config))
|
||
stale_hash = _make_stale_profile_hash(_current_writer_identity_hash(config))
|
||
probe = stale_config["executionAuthorization"]["runtimeProbe"]
|
||
probe["executionProfileSha256"] = stale_hash
|
||
probe["receiptSha256"] = canonical_sha256(
|
||
{key: value for key, value in probe.items() if key != "receiptSha256"}
|
||
)
|
||
return stale_config, stale_hash
|
||
|
||
|
||
def _build_result(
|
||
profile: ExecutionProfile,
|
||
*,
|
||
candidate_body: str = "黎明前，守塔人划亮火柴，灯芯燃起一小团光。",
|
||
total_cost: str = "0.003000",
|
||
model_match: bool = True,
|
||
actual_model_id: str | None = None,
|
||
is_error: bool = False,
|
||
exit_code: int = 0,
|
||
terminal_reason: str | None = None,
|
||
api_error_status: Any = None,
|
||
structured_output: Mapping[str, Any] | None = None,
|
||
structured_output_sha256: str | None = None,
|
||
execution_profile_sha256: str | None = None,
|
||
) -> ClaudeInvocationResult:
|
||
"""按当前 profile 构造一份可控的假调用结果；默认是合规成功产出，覆盖参数用于触发失败分支。"""
|
||
|
||
if structured_output is None:
|
||
structured_output = {"candidateBody": candidate_body}
|
||
if structured_output_sha256 is None:
|
||
structured_output_sha256 = sha256_json(structured_output)
|
||
if execution_profile_sha256 is None:
|
||
execution_profile_sha256 = profile.execution_profile_sha256
|
||
if terminal_reason is None:
|
||
terminal_reason = profile.normal_terminal_reasons[0]
|
||
if actual_model_id is None:
|
||
actual_model_id = profile.resolved_model_id
|
||
receipt = ExecutionReceipt(
|
||
adapter_role=profile.adapter_role,
|
||
invocation_id="test-probe",
|
||
execution_profile_sha256=execution_profile_sha256,
|
||
requested_model_id=profile.resolved_model_id,
|
||
actual_model_id=actual_model_id,
|
||
model_match=model_match,
|
||
effort=profile.effort,
|
||
max_budget_usd_per_call=format(profile.max_budget_usd_per_call, "f"),
|
||
total_cost_usd=total_cost,
|
||
usage={"input_tokens": 1, "output_tokens": 1},
|
||
model_usage={profile.resolved_model_id: {"costUSD": total_cost}},
|
||
stop_reason="end_turn",
|
||
terminal_reason=terminal_reason,
|
||
is_error=is_error,
|
||
api_error_status=api_error_status,
|
||
exit_code=exit_code,
|
||
duration_ms=1,
|
||
input_sha256=sha256_json(PROBE_BUSINESS_INPUT),
|
||
structured_output_sha256=structured_output_sha256,
|
||
json_schema_sha256=profile.json_schema_sha256,
|
||
)
|
||
return ClaudeInvocationResult(structured_output=structured_output, receipt=receipt)
|
||
|
||
|
||
def _invoker_factory(**overrides: Any):
|
||
"""返回一个把假结果绑定到传入 profile 的 invoker，便于注入 refresh_runtime_probe。"""
|
||
|
||
def _invoker(profile: ExecutionProfile, business_input: Mapping[str, Any]):
|
||
return _build_result(profile, **overrides)
|
||
|
||
return _invoker
|
||
|
||
|
||
def _adapters_from_config(config: Mapping[str, Any]) -> WriterReplayProductionAdapters:
|
||
"""用 writer/semantic/judge 三个真实 profile 构建生产适配器；门校验只读 profile，不读 oracle。"""
|
||
|
||
profiles = config["executionProfiles"]
|
||
return WriterReplayProductionAdapters(
|
||
writer_runner=lambda *args, **kwargs: None,
|
||
writer_profile=_profile_from_mapping(profiles["writer"], role="writer"),
|
||
semantic_model_runner=object(),
|
||
semantic_profile=_profile_from_mapping(
|
||
profiles["semantic_detector"], role="semantic_detector"
|
||
),
|
||
judge_model_runner=object(),
|
||
judge_profile=_profile_from_mapping(profiles["blind_judge"], role="blind_judge"),
|
||
oracle_truth_packs={},
|
||
)
|
||
|
||
|
||
class ProbeSuccessPathTest(unittest.TestCase):
|
||
"""a / c / e：成功合规产出下的自哈希、身份绑定与重签一致性。"""
|
||
|
||
def test_a_probe_and_refreshed_config_self_hash_pass_existing_verifier(self):
|
||
"""(a) 探针记录与刷新后配置的 receiptSha256 都能用现有 _verify_self_hash 自检通过。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
probe = result.probe
|
||
# 用 gate_input_builder 现有校验器复核探针自哈希（不抛即通过）。
|
||
_verify_self_hash(probe, "receiptSha256", "runtimeProbe")
|
||
|
||
refreshed_probe = result.refreshed_config["executionAuthorization"]["runtimeProbe"]
|
||
# 刷新后配置里的探针自哈希同样通过现有校验器。
|
||
_verify_self_hash(refreshed_probe, "receiptSha256", "executionAuthorization.runtimeProbe")
|
||
# 且与门校验完全同源的等式成立。
|
||
self.assertEqual(
|
||
refreshed_probe["receiptSha256"],
|
||
canonical_sha256(
|
||
{key: value for key, value in refreshed_probe.items() if key != "receiptSha256"}
|
||
),
|
||
)
|
||
|
||
def test_c_stale_probe_replaced_by_new_identity_hash(self):
|
||
"""(c) 自包含 stale 探针被刷新为当前 writer profile 身份哈希，不再等于 stale 值。"""
|
||
|
||
config = _load_base_config()
|
||
current_identity = _current_writer_identity_hash(config)
|
||
# 自包含构造一个确定失配的旧探针（身份哈希末位翻转），不依赖任何历史硬编码值。
|
||
stale_config, stale_hash = _make_stale_config(config)
|
||
self.assertEqual(
|
||
stale_config["executionAuthorization"]["runtimeProbe"]["executionProfileSha256"],
|
||
stale_hash,
|
||
)
|
||
self.assertNotEqual(stale_hash, current_identity)
|
||
|
||
result = refresh_runtime_probe(
|
||
stale_config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
refreshed_probe = result.refreshed_config["executionAuthorization"]["runtimeProbe"]
|
||
# 刷新后探针身份哈希 == 当前 writer profile 身份哈希（动态算，不硬编码）。
|
||
self.assertEqual(refreshed_probe["executionProfileSha256"], current_identity)
|
||
# 且确定不再等于自包含构造的 stale 值。
|
||
self.assertNotEqual(refreshed_probe["executionProfileSha256"], stale_hash)
|
||
# 新身份哈希必须等于配置已绑定的 profileSha256.writer（门据此放行）。
|
||
self.assertEqual(
|
||
refreshed_probe["executionProfileSha256"],
|
||
config["executionAuthorization"]["profileSha256"]["writer"],
|
||
)
|
||
|
||
def test_e_resign_algorithm_byte_identical_to_existing_verifier(self):
|
||
"""(e) 重签算法与现有 _verify_self_hash 逐字节一致：刷新后 runtimeProbe 通过现有校验。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
refreshed_probe = result.refreshed_config["executionAuthorization"]["runtimeProbe"]
|
||
# 现有 _verify_self_hash 不抛 GateInputBuildError 即证明逐字节一致。
|
||
try:
|
||
_verify_self_hash(refreshed_probe, "receiptSha256", "executionAuthorization.runtimeProbe")
|
||
except GateInputBuildError as exc: # pragma: no cover - 失败时给出可读信息
|
||
self.fail(f"重签自哈希未通过现有校验器: {exc}")
|
||
# 反向确认：篡改任一字段后现有校验器必抛，证明校验真的在起作用。
|
||
tampered = dict(refreshed_probe)
|
||
tampered["totalCostUsd"] = "0.999999"
|
||
with self.assertRaises(GateInputBuildError):
|
||
_verify_self_hash(tampered, "receiptSha256", "executionAuthorization.runtimeProbe")
|
||
|
||
def test_probe_field_set_matches_existing_exactly(self):
|
||
"""新探针字段集与现有 runtimeProbe 完全一致（不缺不多）。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
self.assertEqual(set(result.probe), EXPECTED_PROBE_FIELDS)
|
||
self.assertEqual(result.probe["status"], "successful")
|
||
self.assertEqual(result.probe["checkedAt"], CHECKED_AT)
|
||
self.assertTrue(HASH_PATTERN.fullmatch(result.probe["structuredOutputSha256"]))
|
||
|
||
def test_refresh_does_not_mutate_input_config(self):
|
||
"""刷新是深拷贝：原配置对象的旧探针保持不动。"""
|
||
|
||
# 用自包含 stale 配置：刷新后输入对象的旧探针必须原样保留为 stale 值。
|
||
stale_config, stale_hash = _make_stale_config(_load_base_config())
|
||
before = json.dumps(stale_config, ensure_ascii=False, sort_keys=True)
|
||
refresh_runtime_probe(stale_config, invoker=_invoker_factory(), checked_at=CHECKED_AT)
|
||
after = json.dumps(stale_config, ensure_ascii=False, sort_keys=True)
|
||
self.assertEqual(before, after)
|
||
self.assertEqual(
|
||
stale_config["executionAuthorization"]["runtimeProbe"]["executionProfileSha256"],
|
||
stale_hash,
|
||
)
|
||
|
||
|
||
class ProbeGateIntegrationTest(unittest.TestCase):
|
||
"""b：刷新后配置必须通过加固门；旧配置必须被加固门挡下（前后对照）。"""
|
||
|
||
def test_refreshed_config_passes_execute_authorization_gate(self):
|
||
"""(b) 刷新后配置喂给 _validate_execute_authorization（writer profile 适配）→ 通过。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
adapters = _adapters_from_config(result.refreshed_config)
|
||
gate_result, blocked_status, blocked_code = _validate_execute_authorization(
|
||
result.refreshed_config, adapters
|
||
)
|
||
self.assertIsNone(blocked_code, f"加固门意外阻断: {blocked_status}/{blocked_code}")
|
||
self.assertIsNone(blocked_status)
|
||
self.assertIsNotNone(gate_result)
|
||
# 门放行后回传的计划/上限来自配置预算合同。
|
||
self.assertEqual(gate_result["plannedCalls"]["writer"], 45)
|
||
self.assertEqual(gate_result["maxCalls"]["writer"], 150)
|
||
|
||
def test_stale_config_still_blocked_by_gate_before_refresh(self):
|
||
"""对照：自包含 stale 配置（身份哈希失配但自哈希自洽）仍以 EXECUTE_PROBE_CONTRACT_MISMATCH 被加固门挡下。"""
|
||
|
||
# 自包含构造旧探针：身份哈希确定 != 当前合同，但 receiptSha256 已重算为自洽，
|
||
# 所以它会越过自哈希/profile/运行时/预算各关，精准卡在最后一关合同绑定。
|
||
stale_config, _stale_hash = _make_stale_config(_load_base_config())
|
||
adapters = _adapters_from_config(stale_config)
|
||
gate_result, blocked_status, blocked_code = _validate_execute_authorization(
|
||
stale_config, adapters
|
||
)
|
||
self.assertIsNone(gate_result)
|
||
self.assertEqual(blocked_status, "blocked_execute_probe_contract")
|
||
self.assertEqual(blocked_code, "EXECUTE_PROBE_CONTRACT_MISMATCH")
|
||
|
||
def test_budget_and_raw_self_hashes_untouched_and_valid_after_refresh(self):
|
||
"""刷新只动探针：budget / rawRetention 原样保留且各自自哈希仍通过门同款复核。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(), checked_at=CHECKED_AT
|
||
)
|
||
authorization = result.refreshed_config["executionAuthorization"]
|
||
for name in ("budget", "rawRetention"):
|
||
self.assertEqual(authorization[name], config["executionAuthorization"][name])
|
||
self.assertEqual(
|
||
authorization[name]["receiptSha256"],
|
||
canonical_sha256(
|
||
{k: v for k, v in authorization[name].items() if k != "receiptSha256"}
|
||
),
|
||
)
|
||
|
||
|
||
class ProbeFailClosedTest(unittest.TestCase):
|
||
"""d：失败关闭分支——不写 successful 探针、不产出成功配置。"""
|
||
|
||
def _assert_fail_closed(self, **overrides: Any) -> ProbeRefreshError:
|
||
config = _load_base_config()
|
||
with self.assertRaises(ProbeRefreshError) as ctx:
|
||
refresh_runtime_probe(
|
||
config, invoker=_invoker_factory(**overrides), checked_at=CHECKED_AT
|
||
)
|
||
return ctx.exception
|
||
|
||
def test_d_schema_invalid_empty_body_fails_closed(self):
|
||
"""(d) 结构化输出 candidateBody 为空 → schema 不过 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(structured_output={"candidateBody": ""})
|
||
self.assertEqual(exc.code, "PROBE_SCHEMA_INVALID")
|
||
|
||
def test_d_schema_invalid_extra_field_fails_closed(self):
|
||
"""(d) 结构化输出含额外字段（additionalProperties=false）→ 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(
|
||
structured_output={"candidateBody": "正文", "runId": "leak"}
|
||
)
|
||
self.assertEqual(exc.code, "PROBE_SCHEMA_INVALID")
|
||
|
||
def test_d_over_budget_cap_fails_closed(self):
|
||
"""(d) 单次成本 9.0 超过冻结 cap 5.0 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(total_cost="9.000000")
|
||
self.assertEqual(exc.code, "PROBE_BUDGET_EXCEEDED")
|
||
|
||
def test_d_missing_cost_fails_closed(self):
|
||
"""(d) 回执缺成本 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(total_cost=None)
|
||
self.assertEqual(exc.code, "PROBE_BUDGET_EXCEEDED")
|
||
|
||
def test_d_model_mismatch_fails_closed(self):
|
||
"""(d) 模型不匹配 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(model_match=False)
|
||
self.assertEqual(exc.code, "PROBE_MODEL_MISMATCH")
|
||
|
||
def test_d_receipt_error_fails_closed(self):
|
||
"""(d) 回执标记错误 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(is_error=True)
|
||
self.assertEqual(exc.code, "PROBE_RECEIPT_ERROR")
|
||
|
||
def test_d_nonzero_exit_fails_closed(self):
|
||
"""(d) 非零退出码 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(exit_code=1)
|
||
self.assertEqual(exc.code, "PROBE_NONZERO_EXIT")
|
||
|
||
def test_d_bad_terminal_reason_fails_closed(self):
|
||
"""(d) 终止原因不在正常集合 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(terminal_reason="api_error")
|
||
self.assertEqual(exc.code, "PROBE_TERMINAL_REASON_INVALID")
|
||
|
||
def test_d_api_error_status_fails_closed(self):
|
||
"""(d) 携带 API 错误状态 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(api_error_status=529)
|
||
self.assertEqual(exc.code, "PROBE_API_ERROR")
|
||
|
||
def test_d_profile_binding_mismatch_fails_closed(self):
|
||
"""(d) 回执身份哈希指向旧合同 → 失败关闭。"""
|
||
|
||
# 自包含构造一个确定 != 当前合同身份哈希的回执身份（末位翻转），不硬编码历史值。
|
||
stale_hash = _make_stale_profile_hash(_current_writer_identity_hash(_load_base_config()))
|
||
exc = self._assert_fail_closed(execution_profile_sha256=stale_hash)
|
||
self.assertEqual(exc.code, "PROBE_PROFILE_BINDING_MISMATCH")
|
||
|
||
def test_d_output_hash_mismatch_fails_closed(self):
|
||
"""(d) 输出哈希与产出不一致 → 失败关闭。"""
|
||
|
||
exc = self._assert_fail_closed(structured_output_sha256="sha256:" + "1" * 64)
|
||
self.assertEqual(exc.code, "PROBE_OUTPUT_HASH_INVALID")
|
||
|
||
def test_d_invocation_failure_propagates_runtime_error(self):
|
||
"""(d) invoker 抛 ClaudeRuntimeError（如调用失败）→ 原样向上抛，由 main 兜底失败关闭。"""
|
||
|
||
config = _load_base_config()
|
||
|
||
def _raising_invoker(profile, business_input):
|
||
raise ClaudeRuntimeError("WRITER_RECEIPT_INVALID", "Claude 调用失败")
|
||
|
||
with self.assertRaises(ClaudeRuntimeError):
|
||
refresh_runtime_probe(
|
||
config, invoker=_raising_invoker, checked_at=CHECKED_AT
|
||
)
|
||
|
||
def test_d_deadline_timeout_propagates_runtime_error(self):
|
||
"""(d) invoker 抛 WRITER_TIMEOUT（超 deadline）→ 原样向上抛。"""
|
||
|
||
config = _load_base_config()
|
||
|
||
def _timeout_invoker(profile, business_input):
|
||
raise ClaudeRuntimeError("WRITER_TIMEOUT", "超过冻结 deadline")
|
||
|
||
with self.assertRaises(ClaudeRuntimeError):
|
||
refresh_runtime_probe(
|
||
config, invoker=_timeout_invoker, checked_at=CHECKED_AT
|
||
)
|
||
|
||
|
||
class ProbeDryRunAndCliTest(unittest.TestCase):
|
||
"""dry-run 与 CLI：构建+重签+写文件链路、失败关闭退出码、原配置不被篡改。"""
|
||
|
||
def test_dry_run_invoker_produces_valid_probe(self):
|
||
"""--dry-run 假 invoker 产出绑定当前合同的合规探针。"""
|
||
|
||
config = _load_base_config()
|
||
result = refresh_runtime_probe(
|
||
config, invoker=make_dry_run_invoker(), checked_at=CHECKED_AT, dry_run=True
|
||
)
|
||
_verify_self_hash(result.probe, "receiptSha256", "runtimeProbe")
|
||
writer_profile = build_writer_profile(config)
|
||
self.assertEqual(
|
||
result.probe["executionProfileSha256"], writer_profile.execution_profile_sha256
|
||
)
|
||
self.assertTrue(result.summary["dryRun"])
|
||
|
||
def test_probe_input_is_synthetic_and_free_of_real_work(self):
|
||
"""探针输入固定、极小、全合成，不含原书人物/场景/raw 路径。"""
|
||
|
||
text = json.dumps(PROBE_BUSINESS_INPUT, ensure_ascii=False)
|
||
self.assertIn("运行探针合成任务", text)
|
||
# 深空之影的真实场景关键词不得出现在合成探针输入里。
|
||
for forbidden in ("林澈", "圣蒂曼", "深空", "raw", "/raw", "vault"):
|
||
self.assertNotIn(forbidden, text)
|
||
self.assertEqual(PROBE_BUSINESS_INPUT["proseExcerpts"], [])
|
||
self.assertEqual(PROBE_BUSINESS_INPUT["factConstraints"], [])
|
||
|
||
def test_cli_dry_run_writes_refreshed_file_and_leaves_source_untouched(self):
|
||
"""CLI --dry-run 写出刷新文件、返回 0；原配置文件逐字节不动。"""
|
||
|
||
config = _load_base_config()
|
||
# 动态基准：当前 writer 合同身份哈希；并自包含构造一个确定失配的 stale 值作对照。
|
||
current_identity = _current_writer_identity_hash(config)
|
||
stale_hash = _make_stale_profile_hash(current_identity)
|
||
original_probe_hash = config["executionAuthorization"]["runtimeProbe"][
|
||
"executionProfileSha256"
|
||
]
|
||
source_before = CONFIG_PATH.read_text(encoding="utf-8")
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
output_path = pathlib.Path(tmp) / "refreshed.json"
|
||
rc = refresh_module.main(
|
||
[
|
||
"--config", str(CONFIG_PATH),
|
||
"--output", str(output_path),
|
||
"--dry-run",
|
||
"--checked-at", CHECKED_AT,
|
||
]
|
||
)
|
||
self.assertEqual(rc, 0)
|
||
self.assertTrue(output_path.exists())
|
||
refreshed = json.loads(output_path.read_text(encoding="utf-8"))
|
||
refreshed_probe = refreshed["executionAuthorization"]["runtimeProbe"]
|
||
_verify_self_hash(refreshed_probe, "receiptSha256", "runtimeProbe")
|
||
# 刷新后探针身份哈希 == 当前合同身份哈希（动态算），且确定 != 自包含 stale 值。
|
||
self.assertEqual(refreshed_probe["executionProfileSha256"], current_identity)
|
||
self.assertNotEqual(refreshed_probe["executionProfileSha256"], stale_hash)
|
||
# 原配置文件逐字节未变。
|
||
self.assertEqual(CONFIG_PATH.read_text(encoding="utf-8"), source_before)
|
||
# 内存配置对象的探针在 CLI 刷新前后保持原值（CLI 重读文件，不动这个对象）。
|
||
self.assertEqual(
|
||
config["executionAuthorization"]["runtimeProbe"]["executionProfileSha256"],
|
||
original_probe_hash,
|
||
)
|
||
|
||
def test_cli_output_same_as_config_refused(self):
|
||
"""--output 与 --config 相同 → 返回 2，拒绝就地覆盖。"""
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
copy_path = pathlib.Path(tmp) / "config.json"
|
||
copy_path.write_text(CONFIG_PATH.read_text(encoding="utf-8"), encoding="utf-8")
|
||
rc = refresh_module.main(
|
||
["--config", str(copy_path), "--output", str(copy_path), "--dry-run"]
|
||
)
|
||
self.assertEqual(rc, 2)
|
||
|
||
def test_cli_invocation_failure_returns_nonzero_and_writes_nothing(self):
|
||
"""CLI 真实调用层失败（注入抛错的 run_claude）→ 返回 1，不产出输出文件。"""
|
||
|
||
def _raising_invoker(profile, business_input):
|
||
raise ClaudeRuntimeError("WRITER_TIMEOUT", "超过冻结 deadline")
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
output_path = pathlib.Path(tmp) / "refreshed.json"
|
||
with mock.patch.object(refresh_module, "run_claude", _raising_invoker):
|
||
rc = refresh_module.main(
|
||
["--config", str(CONFIG_PATH), "--output", str(output_path)]
|
||
)
|
||
self.assertEqual(rc, 1)
|
||
self.assertFalse(output_path.exists())
|
||
|
||
def test_cli_schema_failure_returns_nonzero_and_writes_nothing(self):
|
||
"""CLI 假 invoker 返回 schema 非法产出 → 返回 1，不产出输出文件。"""
|
||
|
||
config = _load_base_config()
|
||
writer_profile = build_writer_profile(config)
|
||
bad_result = _build_result(writer_profile, structured_output={"candidateBody": ""})
|
||
|
||
def _bad_invoker(profile, business_input):
|
||
return bad_result
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
output_path = pathlib.Path(tmp) / "refreshed.json"
|
||
with mock.patch.object(refresh_module, "run_claude", _bad_invoker):
|
||
rc = refresh_module.main(
|
||
["--config", str(CONFIG_PATH), "--output", str(output_path)]
|
||
)
|
||
self.assertEqual(rc, 1)
|
||
self.assertFalse(output_path.exists())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|