diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index 430369b..a80a8af 100644 --- a/backend/contracts/prescreen.py +++ b/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 diff --git a/backend/services/judgment_provider.py b/backend/services/judgment_provider.py new file mode 100644 index 0000000..6528985 --- /dev/null +++ b/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} diff --git a/backend/test_judgment_provider.py b/backend/test_judgment_provider.py new file mode 100644 index 0000000..da3481b --- /dev/null +++ b/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() diff --git a/backend/test_prescreen_contracts.py b/backend/test_prescreen_contracts.py index 97bebe6..fdfe93e 100644 --- a/backend/test_prescreen_contracts.py +++ b/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": {