feat(v2): T07-b 专项 disabled 路径与工具策略守卫

T07 第二块。本批**只做 disabled 路径**（T07 卡片要求"先 disabled 专项路径"），
真正的 Jev/本地适配器属 T08，不预造。

新增契约：JudgmentAnswer（可辨识联合，score 必须有限数值）、NativeMetric、
Judgment、JudgmentConfig。契约层硬约束：disabled 不得产出 completed；
关闭专项必须记录原因且不得冻结任何配置修订。

ErrorCode 补齐 E-06/E-13/E-14/E-15/E-08/E-10，并**区分两类编码**：
- 请求级（有 HTTP 语义，进 S1_ERROR_HTTP）：含新增 JUDGMENT_CONFIG_INVALID 422、
  EGRESS_NOT_APPROVED 403
- 运行级（RUN_LEVEL_CODES，只出现在运行体 degradations/judgment 里）：
  JUDGMENT_UNAVAILABLE / CAPABILITY_UNSUPPORTED / OUTPUT_INVALID / RUN_BUDGET_EXCEEDED
不再把假的 HTTP 200 塞进错误映射；两者互斥且并集等于枚举，有测试锁定。

新增 services/judgment_provider.py：
- resolve_config()：关闭时记原因、不冻结配置；开关 false 却传 ID 属 E-07；
  无可用配置时明确报 E-13 而非悄悄关闭；未验证/提供方不符均拒绝
- judge()：disabled → skipped+disabled；缺证**不发起调用**直接 completed+unknown；
  超时/异常 → failed+E-06；题型不符 → skipped+unsupported（E-14）；
  题型对但内容非法 → failed+E-15。失败一律 answer 为空，**不补默认分值**
- allowed_tools()：工具 = Skill 声明 ∩ 注册表白名单，**取交不取并**；材料正文里
  写什么都无法扩展权限（AC-10，有反例测试）

测试：新增 test_judgment_provider.py 22 项。全量 390 项 OK（368 → 390）；
verify.py 10/10 PASS。

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/contracts/prescreen.py

@@ -103,6 +103,14 @@ class ErrorCode(StrEnum):
     MATERIAL_PARSE_FAILED = "MATERIAL_PARSE_FAILED"
     IMPORT_BATCH_INVALID = "IMPORT_BATCH_INVALID"
     FACT_CONFLICT_UNRESOLVED = "FACT_CONFLICT_UNRESOLVED"
+    JUDGMENT_CONFIG_INVALID = "JUDGMENT_CONFIG_INVALID"
+    EGRESS_NOT_APPROVED = "EGRESS_NOT_APPROVED"
+    # 运行级编码：记录在运行体（degradations / judgment）里，**不构成 HTTP 错误**，
+    # 因此刻意不进 S1_ERROR_HTTP；见下方 RUN_LEVEL_CODES。
+    JUDGMENT_UNAVAILABLE = "JUDGMENT_UNAVAILABLE"
+    JUDGMENT_CAPABILITY_UNSUPPORTED = "JUDGMENT_CAPABILITY_UNSUPPORTED"
+    JUDGMENT_OUTPUT_INVALID = "JUDGMENT_OUTPUT_INVALID"
+    RUN_BUDGET_EXCEEDED = "RUN_BUDGET_EXCEEDED"
 
 
 # HTTP 语义沿用既有 translate()（backend/api/project_assistant.py:36-46），
@@ -121,10 +129,24 @@ S1_ERROR_HTTP: dict[ErrorCode, int] = {
     ErrorCode.MATERIAL_PARSE_FAILED: 422,
     ErrorCode.IMPORT_BATCH_INVALID: 422,
     ErrorCode.FACT_CONFLICT_UNRESOLVED: 422,
+    ErrorCode.JUDGMENT_CONFIG_INVALID: 422,
+    ErrorCode.EGRESS_NOT_APPROVED: 403,
 }
 
+# 这些编码只出现在运行结果里（"HTTP 200 仅表示结果可读取"），永远不会作为
+# 错误信封返回，因此与 S1_ERROR_HTTP 互不相交。
+RUN_LEVEL_CODES = frozenset({
+    ErrorCode.JUDGMENT_UNAVAILABLE,
+    ErrorCode.JUDGMENT_CAPABILITY_UNSUPPORTED,
+    ErrorCode.JUDGMENT_OUTPUT_INVALID,
+    ErrorCode.RUN_BUDGET_EXCEEDED,
+})
+
 # 403/422 不自动重试；409 需客户端改载荷或刷新。见 03 §5。
-S1_ERROR_RETRYABLE: dict[ErrorCode, bool] = {code: False for code in ErrorCode}
+S1_ERROR_RETRYABLE: dict[ErrorCode, bool] = {
+    code: code in (ErrorCode.JUDGMENT_UNAVAILABLE, ErrorCode.RUN_BUDGET_EXCEEDED)
+    for code in ErrorCode
+}
 
 
 class ErrorBody(Contract):
@@ -616,3 +638,120 @@ class ReviewPlan(Contract):
         if self.status == GovernanceStatus.published and self.reviewer_id is None:
             raise ValueError("已发布方案必须有审核人")
         return self
+
+
+# ---------------------------------------------------------------- 专项判断（T07-b）
+class JudgmentAnswer(Contract):
+    """可辨识联合：正常答案的 kind 必须与题型一致（11 §5）。"""
+
+    kind: Literal["choice", "score", "binary", "unknown"]
+    value: str | float | bool | None = None
+    reason_code: Literal["missing_evidence", "inconclusive"] | None = None
+
+    @model_validator(mode="after")
+    def _check_answer(self) -> JudgmentAnswer:
+        if self.kind == "unknown":
+            if self.reason_code is None:
+                raise ValueError("unknown 必须给出 reason_code")
+            if self.value is not None:
+                raise ValueError("unknown 不得带取值")
+            return self
+        if self.reason_code is not None:
+            raise ValueError("非 unknown 的答案不得带 reason_code")
+        if self.kind == "choice" and not isinstance(self.value, str):
+            raise ValueError("choice 的取值必须是字符串")
+        if self.kind == "binary" and not isinstance(self.value, bool):
+            raise ValueError("binary 的取值必须是布尔值")
+        if self.kind == "score":
+            if isinstance(self.value, bool) or not isinstance(self.value, (int, float)):
+                raise ValueError("score 的取值必须是数值")
+            if self.value != self.value or self.value in (float("inf"), float("-inf")):
+                raise ValueError("score 必须是有限数值")
+        return self
+
+
+class NativeMetric(Contract):
+    """提供方原生指标。保留语义，**不得**互换阈值或补造概率（BR-18）。"""
+
+    name: str = Field(min_length=1, max_length=80)
+    value: float
+    semantics: str = Field(min_length=1, max_length=DESC_MAX)
+    source: Literal["provider_native", "self_reported"]
+
+    @model_validator(mode="after")
+    def _check_finite(self) -> NativeMetric:
+        if self.value != self.value or self.value in (float("inf"), float("-inf")):
+            raise ValueError("原生指标必须是有限数值；缺失应留空而不是填 0/1")
+        return self
+
+
+class Judgment(Contract):
+    """单题专项判断结果。`completed` 只表示完成该题处理与校验，不表示支持结论。"""
+
+    id: Identifier
+    run_id: Identifier
+    question_id: Identifier
+    question_version: str = Field(min_length=1, max_length=70)
+    type: Literal["choice", "score", "binary"]
+    status: Literal["completed", "skipped", "failed"]
+    answer: JudgmentAnswer | None = None
+    evidence_ids: list[Identifier] = Field(default_factory=list, max_length=20)
+    provider: Literal["jev", "local", "disabled"]
+    profile_id: Identifier | None = None
+    profile_revision: int | None = Field(default=None, ge=1)
+    model_version: str | None = Field(default=None, max_length=ID_MAX)
+    native_metrics: list[NativeMetric] = Field(default_factory=list, max_length=32)
+    error_code: ErrorCode | None = None
+    skip_reason: Literal["disabled", "not_configured", "not_authorized", "unsupported"] | None = None
+
+    @model_validator(mode="after")
+    def _check_status_consistency(self) -> Judgment:
+        if self.status == "completed":
+            if self.answer is None:
+                raise ValueError("completed 必须有合法 answer")
+            if self.error_code or self.skip_reason:
+                raise ValueError("completed 不得带 error_code 或 skip_reason")
+        if self.status == "skipped":
+            if self.skip_reason is None:
+                raise ValueError("skipped 必须有 skip_reason")
+            if self.answer is not None:
+                raise ValueError("skipped 的 answer 必须为空")
+        if self.status == "failed":
+            if self.error_code is None:
+                raise ValueError("failed 必须有 error_code")
+            if self.answer is not None:
+                raise ValueError("failed 的 answer 必须为空")
+        # 关闭专项时不得声称有提供方配置（避免"关了却像在跑"）
+        if self.provider == "disabled" and self.status == "completed":
+            raise ValueError("专项已关闭时不得产出 completed 判断")
+        return self
+
+
+class JudgmentConfig(Contract):
+    """运行创建时冻结的专项配置（BR-20：切换只影响新运行）。"""
+
+    enabled: bool
+    provider: Literal["jev", "local", "disabled"]
+    profile_id: Identifier | None = None
+    profile_revision: int | None = Field(default=None, ge=1)
+    model_version: str | None = Field(default=None, max_length=ID_MAX)
+    adapter_version: str | None = Field(default=None, max_length=ID_MAX)
+    prompt_version: str | None = Field(default=None, max_length=ID_MAX)
+    mapping_version: str | None = Field(default=None, max_length=ID_MAX)
+    disabled_reason: str | None = Field(default=None, max_length=DESC_MAX)
+
+    @model_validator(mode="after")
+    def _check_freeze(self) -> JudgmentConfig:
+        if self.enabled:
+            if self.provider == "disabled":
+                raise ValueError("启用专项时提供方不得为 disabled")
+            if not self.profile_id or self.profile_revision is None:
+                raise ValueError("启用专项时必须冻结配置 ID 与修订")
+        else:
+            if self.provider != "disabled":
+                raise ValueError("关闭专项时提供方必须是 disabled")
+            if self.profile_id or self.profile_revision is not None:
+                raise ValueError("关闭专项时不得冻结任何配置修订")
+            if not self.disabled_reason:
+                raise ValueError("关闭专项必须记录关闭原因，不能静默")
+        return self

backend/services/judgment_provider.py

@@ -0,0 +1,191 @@
+"""T07-b 专项判断的统一入口：disabled 路径与工具策略守卫。
+
+依据：docs/v2/11-judgment-providers.md §3–§6、03-detailed-design.md §9.3；
+BR-09/10/17/18/20、E-06/08/10/13/14/15、AC-09/10/21。
+
+本批**只做 disabled 路径**（T07 卡片要求"先 disabled 专项路径"）：真正的 Jev /
+本地适配器属 T08，这里不预造。
+
+三条不可让渡的规则：
+1. **关闭就明确关闭**：disabled 时已计划题目记 `skipped + disabled`，运行记录关闭
+   原因，绝不静默产生判断，也绝不因为"关了"就伪造高分（BR-09 / AC-21）。
+2. **失败不补分**：服务失效/超时/畸形输出记 `failed + error_code` 与降级记录，
+   `answer` 必须为空；不生成默认分值（AC-09 / E-06 / E-15）。
+3. **工具白名单来自冻结的 Skill**：材料正文里写什么都**不能**扩展工具集合
+   （BR-10 / AC-10）。工具 = Skill 声明 ∩ 注册表白名单，二者取交，不取并。
+"""
+
+from __future__ import annotations
+
+from contracts.prescreen import (
+    ErrorCode,
+    Judgment,
+    JudgmentAnswer,
+    JudgmentConfig,
+    NativeMetric,
+)
+from services import assistant_store as store
+from services.component_registry import TOOLS
+
+DISABLED_REASON_DEFAULT = "本次运行未启用专项判断；主审仍按规则、工具与经验形成意见"
+
+# 仅这些错误码表示"提供方不可用"，可按预算重试一次（03 §5）。
+RETRYABLE_PROVIDER_ERRORS = (ErrorCode.JUDGMENT_UNAVAILABLE,)
+
+
+def resolve_config(enabled: bool, profile_id: str | None = None,
+                   profile_revision: int | None = None, *, known_profiles=(),
+                   provider: str = "local") -> JudgmentConfig:
+    """解析并冻结本次运行的专项配置（BR-20：切换只影响新运行）。
+
+    `known_profiles` 是管理员在后端配置里登记的可用配置；P0 为空，因此在
+    enabled=True 且未给 ID 时会明确报错（E-13），而不是悄悄关闭专项。
+    """
+    if not enabled:
+        if profile_id:
+            # 开关 false 同时传 ID 属 E-07，避免静默忽略用户选择（11 §3）
+            raise ValueError("专项已关闭却指定了配置 ID；请明确启用或清空选择")
+        return JudgmentConfig(enabled=False, provider="disabled",
+                              disabled_reason=DISABLED_REASON_DEFAULT)
+
+    if not profile_id:
+        default = next((item for item in known_profiles if item.get("is_default")), None)
+        if default is None:
+            raise ValueError("当前没有可用的专项模型配置，请选择可用配置或关闭专项核查")
+        profile_id, profile_revision = default["id"], default.get("revision", 1)
+
+    profile = next((item for item in known_profiles if item.get("id") == profile_id), None)
+    if profile is None:
+        raise ValueError("当前专项模型配置不可用，请选择可用配置或关闭专项核查")
+    if profile.get("readiness") != "ready":
+        raise ValueError("该专项模型配置尚未通过验证，暂不可用")
+    if provider not in ("jev", "local"):
+        raise ValueError("专项提供方只能是 jev 或 local")
+    if profile.get("provider") != provider:
+        raise ValueError("所选配置的提供方与请求不一致")
+    return JudgmentConfig(
+        enabled=True, provider=provider, profile_id=profile_id,
+        profile_revision=profile_revision or profile.get("revision", 1),
+        model_version=profile.get("model_version") or "unresolved",
+        adapter_version=profile.get("adapter_version") or "not_applicable",
+        prompt_version=profile.get("prompt_version") or "not_applicable",
+        mapping_version=profile.get("mapping_version") or "not_applicable")
+
+
+def allowed_tools(skill_body: dict, registry_tools=None) -> set[str]:
+    """工具集合 = Skill 声明 ∩ 注册表白名单（取交，不取并）。
+
+    材料正文、运行补充说明、专项模型输出都**不参与**这个计算，因此无法借它们
+    扩展权限（BR-10 / AC-10）。
+    """
+    declared = set(skill_body.get("allowed_tools") or ())
+    return declared & set(registry_tools or TOOLS)
+
+
+def assert_tools_within_policy(requested, skill_body, registry_tools=None) -> None:
+    """任何越界工具请求都必须被拒，而不是"尽量满足"。"""
+    permitted = allowed_tools(skill_body, registry_tools)
+    outside = sorted(set(requested or ()) - permitted)
+    if outside:
+        raise ValueError(f"工具超出当前 Skill 授权范围，已拒绝：{outside}")
+
+
+def _skipped(run_id: str, question: dict, config: JudgmentConfig, reason: str) -> dict:
+    return {"id": store.uid("judgment"), "run_id": run_id,
+            "question_id": question["id"], "question_version": question["version"],
+            "type": question["type"], "status": "skipped", "answer": None,
+            "evidence_ids": list(question.get("evidence_ids", [])),
+            "provider": "disabled", "profile_id": None, "profile_revision": None,
+            "model_version": None, "native_metrics": [], "error_code": None,
+            "skip_reason": reason}
+
+
+def judge(config: JudgmentConfig, run_id: str, question: dict,
+          provider_call=None) -> dict:
+    """执行一道题的专项判断。
+
+    `provider_call` 是 T08 才接的适配器；本批只覆盖 disabled 与失败/畸形路径。
+    """
+    if config.provider == "disabled":
+        return _skipped(run_id, question, config, "disabled")
+
+    if not question.get("evidence_ids"):
+        # 无证据不请求模型：按缺证处理并明确返回 unknown（11 §4）
+        return {"id": store.uid("judgment"), "run_id": run_id,
+                "question_id": question["id"], "question_version": question["version"],
+                "type": question["type"], "status": "completed",
+                "answer": {"kind": "unknown", "value": None, "reason_code": "missing_evidence"},
+                "evidence_ids": [], "provider": config.provider,
+                "profile_id": config.profile_id, "profile_revision": config.profile_revision,
+                "model_version": config.model_version, "native_metrics": [],
+                "error_code": None, "skip_reason": None}
+
+    if provider_call is None:
+        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)
+
+    try:
+        raw = provider_call(config, question)
+    except TimeoutError:
+        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)
+    except Exception:
+        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)
+
+    return _normalise(run_id, question, config, raw)
+
+
+def _failed(run_id: str, question: dict, config: JudgmentConfig, code: ErrorCode) -> dict:
+    """失败必须留痕且**不补分**：answer 为空，不生成任何默认取值。"""
+    return {"id": store.uid("judgment"), "run_id": run_id,
+            "question_id": question["id"], "question_version": question["version"],
+            "type": question["type"], "status": "failed", "answer": None,
+            "evidence_ids": list(question.get("evidence_ids", [])),
+            "provider": config.provider, "profile_id": config.profile_id,
+            "profile_revision": config.profile_revision,
+            "model_version": config.model_version, "native_metrics": [],
+            "error_code": code, "skip_reason": None}
+
+
+def _normalise(run_id: str, question: dict, config: JudgmentConfig, raw: dict) -> dict:
+    """校验并归一化提供方输出；畸形即 E-15，不猜一个分值补齐。"""
+    if not isinstance(raw, dict):
+        return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
+    answer_kind = raw.get("kind")
+    if answer_kind == "unknown":
+        if raw.get("reason_code") not in ("missing_evidence", "inconclusive"):
+            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
+        answer = JudgmentAnswer(kind="unknown", reason_code=raw["reason_code"])
+    else:
+        if answer_kind != question["type"]:
+            # 题型不符（含"把连续输出硬当 binary"）属能力不支持，不是失败
+            return {**_failed(run_id, question, config, ErrorCode.JUDGMENT_CAPABILITY_UNSUPPORTED),
+                    "status": "skipped", "error_code": None, "skip_reason": "unsupported"}
+        try:
+            answer = JudgmentAnswer(kind=answer_kind, value=raw.get("value"))
+        except Exception:
+            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
+
+    metrics = []
+    for item in raw.get("native_metrics") or ():
+        try:
+            metrics.append(NativeMetric(**item).model_dump(mode="json"))
+        except Exception:
+            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
+
+    body = {"id": store.uid("judgment"), "run_id": run_id,
+            "question_id": question["id"], "question_version": question["version"],
+            "type": question["type"], "status": "completed",
+            "answer": answer.model_dump(mode="json"),
+            "evidence_ids": list(question.get("evidence_ids", [])),
+            "provider": config.provider, "profile_id": config.profile_id,
+            "profile_revision": config.profile_revision,
+            "model_version": config.model_version, "native_metrics": metrics,
+            "error_code": None, "skip_reason": None}
+    return Judgment(**body).model_dump(mode="json")
+
+
+def plan_questions(config: JudgmentConfig, questions) -> dict:
+    """关闭专项时把已计划题目统一记 skipped，并给出运行级关闭原因。"""
+    if config.provider != "disabled":
+        return {"judgments": [], "disabled_reason": None}
+    return {"judgments": [_skipped("", question, config, "disabled") for question in questions or ()],
+            "disabled_reason": config.disabled_reason}

backend/test_judgment_provider.py

@@ -0,0 +1,187 @@
+"""T07-b 专项 disabled 路径与工具策略测试：AC-09/10/21。
+
+关键反例：
+- 关闭专项**不得**产出任何 completed 判断，也不得伪造高分
+- 提供方失效/超时/畸形输出记 failed（或 unsupported 记 skipped），answer 必须为空
+- 材料或运行要求里写什么都**不能**扩展工具集合（工具 = Skill 声明 ∩ 白名单）
+"""
+
+import unittest
+
+from contracts.prescreen import ErrorCode
+from services import judgment_provider as jp
+
+PROFILE = {"id": "p1", "provider": "local", "readiness": "ready", "revision": 2,
+           "model_version": "laya@abc123"}
+
+QUESTION = {"id": "q1", "version": "v1", "type": "choice", "evidence_ids": ["ev-1"]}
+
+
+class DisabledPathTests(unittest.TestCase):
+    def setUp(self):
+        self.config = jp.resolve_config(False)
+
+    def test_disabled_config_records_a_reason(self):
+        self.assertEqual(self.config.provider, "disabled")
+        self.assertFalse(self.config.enabled)
+        self.assertTrue(self.config.disabled_reason)
+        self.assertIsNone(self.config.profile_id)
+        self.assertIsNone(self.config.profile_revision)
+
+    def test_disabled_never_freezes_a_profile_revision(self):
+        frozen = self.config.model_dump()
+        self.assertIsNone(frozen["profile_id"])
+        self.assertIsNone(frozen["profile_revision"])
+
+    def test_disabled_question_is_skipped_with_reason(self):
+        # AC-21：已计划题目记 skipped + disabled，明确关闭原因
+        result = jp.judge(self.config, "run-1", QUESTION)
+        self.assertEqual(result["status"], "skipped")
+        self.assertEqual(result["skip_reason"], "disabled")
+        self.assertIsNone(result["answer"], "关闭时不得产出任何答案")
+        self.assertIsNone(result["error_code"])
+
+    def test_disabled_never_produces_a_high_score(self):
+        result = jp.judge(self.config, "run-1", QUESTION)
+        self.assertNotEqual(result["status"], "completed")
+        self.assertEqual(result["native_metrics"], [], "不得伪造任何指标")
+
+    def test_plan_questions_marks_all_as_skipped(self):
+        planned = jp.plan_questions(self.config, [QUESTION, dict(QUESTION, id="q2")])
+        self.assertEqual([item["skip_reason"] for item in planned["judgments"]],
+                         ["disabled", "disabled"])
+        self.assertTrue(planned["disabled_reason"])
+
+    def test_switching_off_with_an_id_is_rejected(self):
+        # 开关 false 却传 ID 属 E-07，避免静默忽略用户选择
+        with self.assertRaises(ValueError) as ctx:
+            jp.resolve_config(False, "p1")
+        self.assertIn("却指定了配置 ID", str(ctx.exception))
+
+    def test_enabling_without_a_usable_profile_fails_loudly(self):
+        # 没有可用配置时明确报 E-13，而不是悄悄关闭专项
+        with self.assertRaises(ValueError) as ctx:
+            jp.resolve_config(True, None, None, known_profiles=[])
+        self.assertIn("没有可用的专项模型配置", str(ctx.exception))
+
+
+class EnabledConfigTests(unittest.TestCase):
+    def test_enabled_freezes_identity_and_versions(self):
+        config = jp.resolve_config(True, "p1", 2, known_profiles=[PROFILE])
+        self.assertEqual(config.provider, "local")
+        self.assertEqual(config.profile_id, "p1")
+        self.assertEqual(config.profile_revision, 2)
+        self.assertEqual(config.model_version, "laya@abc123")
+
+    def test_unready_profile_is_refused(self):
+        with self.assertRaises(ValueError) as ctx:
+            jp.resolve_config(True, "p1", 1,
+                              known_profiles=[dict(PROFILE, readiness="unavailable")])
+        self.assertIn("尚未通过验证", str(ctx.exception))
+
+    def test_provider_mismatch_is_refused(self):
+        with self.assertRaises(ValueError) as ctx:
+            jp.resolve_config(True, "p1", 1, known_profiles=[PROFILE], provider="jev")
+        self.assertIn("提供方与请求不一致", str(ctx.exception))
+
+    def test_unknown_profile_is_refused(self):
+        with self.assertRaises(ValueError):
+            jp.resolve_config(True, "nope", 1, known_profiles=[PROFILE])
+
+
+class FailurePathTests(unittest.TestCase):
+    def setUp(self):
+        self.config = jp.resolve_config(True, "p1", 2, known_profiles=[PROFILE])
+
+    def test_missing_evidence_returns_unknown_without_calling_the_provider(self):
+        called = []
+        result = jp.judge(self.config, "run-1",
+                          {"id": "q1", "version": "v1", "type": "choice", "evidence_ids": []},
+                          provider_call=lambda *a: called.append(1))
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["answer"]["reason_code"], "missing_evidence")
+        self.assertEqual(called, [], "缺证时不得发起模型调用")
+
+    def test_provider_timeout_is_failed_not_scored(self):
+        def boom(*_):
+            raise TimeoutError("slow")
+        result = jp.judge(self.config, "run-1", QUESTION, provider_call=boom)
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(result["error_code"], "JUDGMENT_UNAVAILABLE")
+        self.assertIsNone(result["answer"], "失败不得补一个默认分值")
+
+    def test_provider_exception_is_failed(self):
+        def boom(*_):
+            raise RuntimeError("connection reset")
+        result = jp.judge(self.config, "run-1", QUESTION, provider_call=boom)
+        self.assertEqual(result["status"], "failed")
+        self.assertIsNone(result["answer"])
+
+    def test_malformed_output_is_rejected_not_guessed(self):
+        # 注意区分：**题型不符**是 E-14（能力不支持，见下一个用例）；
+        # 这里是**题型正确但内容非法** → E-15。
+        for bad in ({"kind": "choice", "value": 123},              # 值不是字符串
+                    {"kind": "unknown"},                            # 缺 reason_code
+                    {"kind": "choice", "value": "supported",
+                     "native_metrics": [{"name": "x", "value": float("nan"),
+                                         "semantics": "s", "source": "provider_native"}]}):
+            with self.subTest(bad=bad):
+                result = jp.judge(self.config, "run-1", QUESTION, provider_call=lambda *a: bad)
+                self.assertEqual(result["status"], "failed")
+                self.assertEqual(result["error_code"], "JUDGMENT_OUTPUT_INVALID")
+                self.assertIsNone(result["answer"])
+
+    def test_wrong_answer_kind_is_unsupported_not_a_failure_score(self):
+        result = jp.judge(self.config, "run-1", QUESTION,
+                          provider_call=lambda *a: {"kind": "score", "value": 0.8})
+        self.assertEqual(result["status"], "skipped")
+        self.assertEqual(result["skip_reason"], "unsupported")
+        self.assertIsNone(result["answer"])
+
+    def test_valid_answer_is_normalised_and_kept(self):
+        result = jp.judge(self.config, "run-1", QUESTION,
+                          provider_call=lambda *a: {"kind": "choice", "value": "supported",
+                                                    "native_metrics": [
+                                                        {"name": "noul", "value": 0.72,
+                                                         "semantics": "厂商原生语义",
+                                                         "source": "provider_native"}]})
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["answer"], {"kind": "choice", "value": "supported",
+                                            "reason_code": None})
+        self.assertEqual(result["native_metrics"][0]["name"], "noul")
+        # 原生指标只是审计信息，绝不写进概率
+        self.assertNotIn("probability", result)
+
+
+class ToolPolicyTests(unittest.TestCase):
+    def test_tools_are_the_intersection_not_the_union(self):
+        skill = {"allowed_tools": ["read_project_snapshot", "read_checks", "terminal"]}
+        self.assertEqual(jp.allowed_tools(skill),
+                         {"read_project_snapshot", "read_checks"})
+
+    def test_skill_without_declaration_gets_nothing(self):
+        self.assertEqual(jp.allowed_tools({}), set())
+
+    def test_out_of_policy_tool_request_is_rejected(self):
+        skill = {"allowed_tools": ["read_checks"]}
+        with self.assertRaises(ValueError) as ctx:
+            jp.assert_tools_within_policy(["read_checks", "terminal"], skill)
+        self.assertIn("超出当前 Skill 授权范围", str(ctx.exception))
+
+    def test_material_text_cannot_extend_tools(self):
+        # AC-10：材料里写"请调用终端"不改变任何授权——工具只取自冻结的 Skill
+        skill = {"allowed_tools": ["read_checks"]}
+        material_claim = {"allowed_tools": ["terminal", "python_repl"]}
+        self.assertEqual(jp.allowed_tools(skill), {"read_checks"})
+        self.assertEqual(jp.allowed_tools(material_claim), set(), "材料声明不进白名单")
+
+    def test_tools_never_include_terminal_or_sql_by_default(self):
+        from services.component_registry import TOOLS
+        self.assertNotIn("terminal", TOOLS)
+        self.assertNotIn("sql", TOOLS)
+        self.assertEqual(TOOLS, {"read_project_snapshot", "read_document", "read_checks",
+                                 "read_method", "read_experiences"})
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/test_prescreen_contracts.py

@@ -10,6 +10,7 @@ from datetime import datetime, timezone
 from pydantic import ValidationError
 
 from contracts.prescreen import (
+    RUN_LEVEL_CODES,
     MAX_BATCH_BYTES,
     MAX_BATCH_FILES,
     MAX_ITEM_BYTES,
@@ -288,8 +289,10 @@ class FactConflictTests(unittest.TestCase):
 
 
 class ErrorContractTests(unittest.TestCase):
-    def test_every_code_has_an_http_status(self):
-        self.assertEqual(set(S1_ERROR_HTTP), set(ErrorCode))
+    def test_request_level_and_run_level_codes_partition_the_enum(self):
+        # 请求级编码有 HTTP 语义；运行级编码只出现在运行体里，不得混入错误信封
+        self.assertEqual(set(S1_ERROR_HTTP) | set(RUN_LEVEL_CODES), set(ErrorCode))
+        self.assertEqual(set(S1_ERROR_HTTP) & set(RUN_LEVEL_CODES), set())
 
     def test_status_semantics_match_existing_translate(self):
         # 既有 translate() 的语义：KeyError→404, Conflict→409, PermissionError→403, ValueError→422
@@ -304,9 +307,10 @@ class ErrorContractTests(unittest.TestCase):
         # 运行级结果 = HTTP 200 且冲突保存在 degradations（那个面不走错误信封）。
         self.assertEqual(S1_ERROR_HTTP[ErrorCode.FACT_CONFLICT_UNRESOLVED], 422)
 
-    def test_all_codes_are_non_retryable(self):
-        # 03 §5：403/422 不自动重试；409 需客户端改载荷或刷新
-        self.assertFalse(any(S1_ERROR_RETRYABLE.values()))
+    def test_only_provider_and_budget_errors_are_retryable(self):
+        # 03 §5：403/422 不自动重试；仅提供方不可用与预算类可按预算重试一次
+        retryable = {code for code, flag in S1_ERROR_RETRYABLE.items() if flag}
+        self.assertEqual(retryable, {ErrorCode.JUDGMENT_UNAVAILABLE, ErrorCode.RUN_BUDGET_EXCEEDED})
 
     def test_envelope_shape(self):
         body = ErrorResponse.model_validate({"error": {