diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index a80a8af..8024dfa 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -755,3 +755,86 @@ class JudgmentConfig(Contract): if not self.disabled_reason: raise ValueError("关闭专项必须记录关闭原因,不能静默") return self + + +# ---------------------------------------------------------------- 预审运行(T07-c) +class Degradation(Contract): + """降级记录:未完成的检查必须留痕,不能静默(BR-09)。""" + + code: ErrorCode + message: str = Field(min_length=1, max_length=DESC_MAX) + + +class DimensionResult(Contract): + """维度在本次运行中的状态。`unknown` 不等于不成立(03 §9.1)。""" + + id: Identifier + stage: int | None = None + state: Literal["unknown", "checked", "blocked"] + basis_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + + +class ProbabilityView(Contract): + """概率呈现。P0 的 real_estimate **必须为空**,演示值只能走 demo_estimate(BR-29)。""" + + target_event: Literal["company_review_meeting_pass"] = "company_review_meeting_pass" + snapshot_id: Identifier + as_of: AwareDatetime + real_estimate: None = None + reason: Literal["not_calibrated"] = "not_calibrated" + demo_estimate: dict | None = None + + @model_validator(mode="after") + def _check_demo_isolated(self) -> ProbabilityView: + demo = self.demo_estimate + if demo is None: + return self + if demo.get("source") != "synthetic_demo": + raise ValueError("演示估计必须标明来源 synthetic_demo") + if not demo.get("label"): + raise ValueError("演示估计必须带固定的『人工设定』说明,避免被当作真实概率") + value = demo.get("value") + if not isinstance(value, (int, float)) or isinstance(value, bool) or not 0 <= value <= 1: + raise ValueError("演示估计须为 0–1 的数值") + return self + + +class ScreeningRun(Contract): + """预审运行视图(docs/v2/03-detailed-design.md §9.2 + §9.3 增量)。 + + 这是既有 run 对象的**受约束投影**,不是第二份运行存储:plan_ref、snapshot、 + judgment_config 与 versions 在创建时冻结(BR-13 / BR-20)。 + """ + + id: Identifier + project_id: Identifier + data_domain: DataDomain + revision: int = Field(ge=1) + snapshot_id: Identifier + plan_ref: Ref + status: Literal["queued", "running", "awaiting_input", "completed", "partial", "failed", "cancelled"] + recommendation: Literal["advance", "supplement", "stop", "insufficient_information"] | None = None + findings: list[dict] = Field(default_factory=list, max_length=500) + missing_inputs: list[str] = Field(default_factory=list, max_length=200) + applied_experiences: list[dict] = Field(default_factory=list, max_length=200) + # 未校准时恒为空——契约层直接禁止填入数值 + probability: None = None + probability_reason: Literal["not_calibrated"] = "not_calibrated" + probability_view: ProbabilityView | None = None + degradations: list[Degradation] = Field(default_factory=list, max_length=100) + judgment_config: JudgmentConfig + versions: dict = Field(default_factory=dict) + dimension_results: list[DimensionResult] = Field(default_factory=list, max_length=32) + run_instruction: str = Field(default="", max_length=2000) + created_at: AwareDatetime + finished_at: AwareDatetime | None = None + + @model_validator(mode="after") + def _check_run(self) -> ScreeningRun: + if self.status in ("queued", "running") and self.recommendation is not None: + raise ValueError("运行未结束时不得给出建议") + if self.status == "failed" and not self.degradations: + raise ValueError("failed 运行必须有降级记录,不能只留空结果") + if self.probability_view and self.status in ("queued", "running"): + raise ValueError("运行未结束时不得产出概率视图") + return self diff --git a/backend/services/prescreen.py b/backend/services/prescreen.py new file mode 100644 index 0000000..7199c41 --- /dev/null +++ b/backend/services/prescreen.py @@ -0,0 +1,147 @@ +"""T07-c 预审运行装配:把冻结方案、结论校验与专项配置收成一条可审计的运行记录。 + +依据:docs/v2/03-detailed-design.md §9.2/§9.3、02-architecture.md §4; +BR-13/14/20/31、AC-13/14/45。 + +**这是既有 run 对象的受约束投影,不是第二份运行存储。** 既有 +`ProjectAssistantService.enqueue()` 已经提供输入快照冻结、幂等键、revision 冲突与 +队列恢复(AC-13/AC-14 的实现),本模块**不重做**这些,只负责: +1. 把冻结方案与专项配置投影进 ScreeningRun 并校验 +2. 把政策结论、降级与维度状态落成契约合法的运行视图 +3. 保证任何路径都写不出未校准概率 +""" + +from __future__ import annotations + +from contracts.prescreen import ( + DataDomain, + DimensionResult, + ProbabilityView, + ScreeningRun, +) +from services import assistant_store as store +from services import prescreen_policy as policy + + +def versions_block(frozen_plan: dict, extra: dict | None = None) -> dict: + """冻结的运行版本块。缺哪一项就明确留空,不编造版本号。""" + plan = frozen_plan.get("plan") or {} + components = frozen_plan.get("component_versions") or [ + {"id": item["id"], "version": item.get("version")} + for item in [plan] + list(frozen_plan.get("skills") or ()) + + list(frozen_plan.get("rules") or ()) + + ([frozen_plan["template"]] if frozen_plan.get("template") else []) + ] + block = { + "schema": 1, + "policy": "prescreen_policy@1", + "skills": [item for item in components if item.get("id")], + "experiences": list(frozen_plan.get("experience_versions") or ()), + "model": None, # 主审模型配置由运行冻结,未配置即留空 + "questions": None, # 题库版本在 T08 统一适配后填入,P0 留空 + "plan": {"id": plan.get("id"), "version": plan.get("version")}, + "plan_hash": frozen_plan.get("content_hash"), + } + block.update(extra or {}) + return block + + +def dimension_results(frozen_plan: dict, findings) -> list[dict]: + """把维度与本次核查结果对上:命中即 checked,缺证即 unknown,永不默认成立。""" + result_by_code = {} + for item in findings or (): + result_by_code.setdefault(str(item.get("code")), []).append(item) + + rows = [] + for spec in frozen_plan.get("dimensions") or (): + refs = spec.get("component_refs") or () + bases, states, reported = [], set(), 0 + for ref in refs: + items = result_by_code.get(str(ref.get("id")), ()) + if not items: + continue # 该组件本次没有任何结果 = 未报告 + reported += 1 + for item in items: + states.add(item.get("result")) + bases.extend(item.get("evidence_ids") or []) + # 静默不等于通过(BR-07):任一被引用的组件没有结果,维度就不能算已核查。 + # issue 优先于 unknown —— 具体的阻断比"信息缺失"更该被看到。 + unreported = len(refs) - reported + if states & {"issue"}: + state = "blocked" + elif unreported or not states or (states & {"unknown", "error", "not_applicable"}): + state = "unknown" + else: + state = "checked" + stages = [item.get("stage") for ref in refs + for item in result_by_code.get(str(ref.get("id")), ()) + if isinstance(item.get("stage"), int)] + rows.append({"id": spec["id"], "state": state, + "stage": max(stages) if stages else None, + "basis_ids": sorted(set(bases))}) + return rows + + +def probability_view(snapshot_id: str, as_of: str | None = None, + demo_estimate: dict | None = None) -> dict: + """概率视图。真实估计恒为空;演示值只能走 demo_estimate 且必须带说明。""" + return ProbabilityView(snapshot_id=snapshot_id, as_of=as_of or store.now(), + demo_estimate=demo_estimate).model_dump(mode="json") + + +def assemble(run_body: dict, frozen_plan: dict, decision: dict, judgment_config, + data_domain: str = "demo", status: str = "completed", + trajectory_id: str | None = None) -> dict: + """装配 ScreeningRun 视图并校验。 + + 任何一条路径写不出未校准概率:`probability` 与 `probability_view.real_estimate` + 都是契约里的 `None` 类型,传别的值会直接被 pydantic 拒绝。 + """ + plan = frozen_plan.get("plan") or {} + snapshot_id = run_body.get("input_snapshot_id") or run_body.get("snapshot_id") + if not snapshot_id: + raise ValueError("运行缺少输入快照,无法装配预审视图") + + body = { + "id": run_body["id"], + "project_id": run_body["project_id"], + "data_domain": data_domain, + "revision": run_body.get("revision", 1), + "snapshot_id": snapshot_id, + "plan_ref": {"id": plan.get("id", ""), "version": str(plan.get("version", ""))}, + "status": status, + "recommendation": decision.get("recommendation"), + "findings": decision.get("findings") or [], + "missing_inputs": decision.get("missing_inputs") or [], + "applied_experiences": run_body.get("experience_applications") or [], + "degradations": decision.get("degradations") or [], + "judgment_config": judgment_config.model_dump(mode="json") + if hasattr(judgment_config, "model_dump") else dict(judgment_config), + "versions": versions_block(frozen_plan, {"trajectory_id": trajectory_id} + if trajectory_id else None), + "dimension_results": dimension_results(frozen_plan, decision.get("findings")), + "run_instruction": run_body.get("requirement") or "", + "created_at": run_body.get("created_at") or store.now(), + "finished_at": run_body.get("ended_at"), + } + if status in ("completed", "partial"): + body["probability_view"] = probability_view(snapshot_id) + result = ScreeningRun(**body).model_dump(mode="json") + # 最后一道闸:契约已把这两个字段的类型限死为 None,这里再显式确认一次, + # 使得"未校准就没有概率"在装配出口也是可读的断言而不是隐含约束。 + if result["probability"] is not None: + raise ValueError("未校准时不得产出审批概率") + if (result.get("probability_view") or {}).get("real_estimate") is not None: + raise ValueError("P0 的真实概率估计必须为空") + return result + + +def assert_versions_frozen(screening: dict) -> None: + """运行必须冻结方案与专项配置,否则不得进入执行(BR-13 / BR-20)。""" + reference = screening.get("plan_ref") or {} + if not reference.get("id") or not reference.get("version"): + raise ValueError("运行未冻结评审方案版本,不得执行") + config = screening.get("judgment_config") or {} + if config.get("enabled") and (not config.get("profile_id") + or config.get("profile_revision") is None): + raise ValueError("启用专项的运行必须冻结配置 ID 与修订") diff --git a/backend/test_prescreen_run.py b/backend/test_prescreen_run.py new file mode 100644 index 0000000..f4d6f0c --- /dev/null +++ b/backend/test_prescreen_run.py @@ -0,0 +1,217 @@ +"""T07-c 预审运行装配测试:AC-13/14/45 与冻结语义。 + +AC-13(新证据 → 新运行,旧运行留存)与 AC-14(幂等键)由既有 enqueue 实现, +本文件**验证它确实成立**而不是重新实现;其余断言针对装配与冻结。 +""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from services import assistant_store as store +from services import component_registry as registry +from services import judgment_provider as jp +from services import prescreen, prescreen_policy +from services.project_assistant import ProjectAssistantService + + +def frozen_plan(dimensions=None): + return { + "plan": {"id": "corporate.default", "version": "1.0.0"}, + "skills": [{"id": "corporate.cashflow", "version": "1.0.0", "code": "financial.cashflow.missing"}], + "rules": [{"id": "demo.coverage", "version": "1.0.0", + "conditions": [{"code": "lease.coverage.below_demo_threshold"}]}], + "template": {"id": "template.ai-pre-review", "version": "1.0.0"}, + "content_hash": "abc123", + "dimensions": dimensions or [], + } + + +def run_body(**over): + body = {"id": "run-1", "project_id": "DEMO-B", "created_at": store.now(), + "input_snapshot_id": "snap-1", "requirement": "重点核对现金流"} + body.update(over) + return body + + +class AssemblyTests(unittest.TestCase): + def setUp(self): + self.config = jp.resolve_config(False) + + def assemble(self, decision=None, plan=None, status="completed"): + decision = decision if decision is not None else prescreen_policy.decide([]) + return prescreen.assemble(run_body(), plan or frozen_plan(), decision, + self.config, status=status) + + def test_assembled_run_is_contract_valid(self): + screening = self.assemble() + self.assertEqual(screening["id"], "run-1") + self.assertEqual(screening["plan_ref"], {"id": "corporate.default", "version": "1.0.0"}) + self.assertEqual(screening["snapshot_id"], "snap-1") + + def test_probability_is_null_and_reason_recorded(self): + screening = self.assemble() + self.assertIsNone(screening["probability"]) + self.assertEqual(screening["probability_reason"], "not_calibrated") + self.assertIsNone(screening["probability_view"]["real_estimate"]) + self.assertEqual(screening["probability_view"]["reason"], "not_calibrated") + + def test_demo_estimate_must_be_labelled_and_cannot_become_real(self): + view = prescreen.probability_view("snap-1", demo_estimate={ + "value": 0.42, "source": "synthetic_demo", "scenario_table_version": "v1", + "scenario_key": "k", "label": "人工设定、未经真实历史校准"}) + self.assertEqual(view["demo_estimate"]["value"], 0.42) + self.assertIsNone(view["real_estimate"]) + for bad in ({"value": 0.42, "source": "model", "label": "x"}, # 来源不符 + {"value": 0.42, "source": "synthetic_demo"}, # 缺说明 + {"value": 1.5, "source": "synthetic_demo", "label": "x"}): # 越界 + with self.subTest(bad=bad): + with self.assertRaises(Exception): + prescreen.probability_view("snap-1", demo_estimate=bad) + + def test_judgment_config_is_frozen_into_the_run(self): + screening = self.assemble() + self.assertEqual(screening["judgment_config"]["provider"], "disabled") + self.assertFalse(screening["judgment_config"]["enabled"]) + self.assertTrue(screening["judgment_config"]["disabled_reason"]) + + def test_versions_block_records_plan_and_hash(self): + screening = self.assemble() + self.assertEqual(screening["versions"]["plan"]["id"], "corporate.default") + self.assertEqual(screening["versions"]["plan_hash"], "abc123") + self.assertIsNone(screening["versions"]["questions"], "题库版本未接入时留空,不编造") + + def test_hard_block_is_carried_into_the_run(self): + decision = prescreen_policy.decide( + [{"code": "lease.coverage.below_demo_threshold", "result": "issue", + "basis_version": "demo.coverage@1.0.0"}], + hard_rule_codes=["lease.coverage.below_demo_threshold"]) + screening = self.assemble(decision) + self.assertEqual(screening["recommendation"], "stop") + + def test_failed_run_must_carry_degradations(self): + with self.assertRaises(ValueError) as ctx: + self.assemble(prescreen_policy.decide([]), status="failed") + self.assertIn("降级记录", str(ctx.exception)) + + def test_unfinished_run_cannot_carry_a_recommendation(self): + decision = prescreen_policy.decide([]) + with self.assertRaises(ValueError) as ctx: + self.assemble(decision, status="running") + self.assertIn("未结束时不得给出建议", str(ctx.exception)) + + def test_missing_snapshot_is_refused(self): + with self.assertRaises(ValueError) as ctx: + prescreen.assemble(run_body(input_snapshot_id=None), frozen_plan(), + prescreen_policy.decide([]), self.config) + self.assertIn("输入快照", str(ctx.exception)) + + def test_junk_versions_are_not_invented(self): + block = prescreen.versions_block({"plan": {"id": "p", "version": "1"}}) + self.assertIsNone(block["model"], "主审模型未配置时留空,不编造版本") + self.assertIsNone(block["questions"]) + + +class DimensionResultTests(unittest.TestCase): + def test_unknown_when_no_component_reports(self): + rows = prescreen.dimension_results( + frozen_plan([{"id": "d1", "component_refs": [{"id": "corporate.cashflow"}]}]), + []) + self.assertEqual(rows[0]["state"], "unknown") + self.assertIsNone(rows[0]["stage"]) + + def test_issue_marks_the_dimension_blocked(self): + rows = prescreen.dimension_results( + frozen_plan([{"id": "d1", "component_refs": [{"id": "corporate.cashflow"}]}]), + [{"code": "corporate.cashflow", "result": "issue", "stage": 2, + "evidence_ids": ["ev-1"]}]) + self.assertEqual(rows[0]["state"], "blocked") + self.assertEqual(rows[0]["stage"], 2) + self.assertEqual(rows[0]["basis_ids"], ["ev-1"]) + + def test_pass_marks_checked_and_unknown_beats_pass(self): + plan = frozen_plan([{"id": "d1", "component_refs": [{"id": "a"}, {"id": "b"}]}]) + rows = prescreen.dimension_results(plan, [{"code": "a", "result": "pass"}]) + self.assertEqual(rows[0]["state"], "unknown", "部分无结果时不得算已核查") + rows = prescreen.dimension_results(plan, [{"code": "a", "result": "pass"}, + {"code": "b", "result": "pass"}]) + self.assertEqual(rows[0]["state"], "checked") + + def test_dimension_evidence_is_traceable(self): + rows = prescreen.dimension_results( + frozen_plan([{"id": "d1", "component_refs": [{"id": "a"}]}]), + [{"code": "a", "result": "pass", "evidence_ids": ["ev-9"]}]) + self.assertIn("ev-9", rows[0]["basis_ids"]) + + +class FrozenVersionsGuardTests(unittest.TestCase): + def test_run_without_a_plan_ref_is_refused(self): + with self.assertRaises(ValueError) as ctx: + prescreen.assert_versions_frozen({"plan_ref": {"id": "", "version": ""}, + "judgment_config": {"enabled": False}}) + self.assertIn("未冻结评审方案版本", str(ctx.exception)) + + def test_enabled_judgment_without_a_frozen_revision_is_refused(self): + with self.assertRaises(ValueError) as ctx: + prescreen.assert_versions_frozen({ + "plan_ref": {"id": "p", "version": "1"}, + "judgment_config": {"enabled": True, "profile_id": None, + "profile_revision": None}}) + self.assertIn("必须冻结配置 ID 与修订", str(ctx.exception)) + + +class ExistingRunGuaranteesTests(unittest.TestCase): + """AC-13/14 由既有 enqueue 提供——这里验证其成立,不重新实现。""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.path = Path(self._tmp.name) + self.service = ProjectAssistantService(self.path / "assistant.sqlite3") + + def tearDown(self): + self._tmp.cleanup() + + def identity(self): + return {"id": "demo-manager", "role": "manager", "projects": ["*"], "preview": False} + + def test_new_evidence_produces_a_new_snapshot_and_keeps_the_old_run(self): + # AC-13:新证据版本 → 新快照 + 新运行,旧运行留存 + first = self.service.get_project("DEMO-B") + run_one = self.service.enqueue("DEMO-B", self.identity(), "第一轮") + self.service.supplement("DEMO-B", "demo-manager", + {"collection_evidence": "已补充到账凭证"}, [], + self.service.get_project("DEMO-B")["revision"], "补件说明") + run_two = self.service.enqueue("DEMO-B", self.identity(), "第二轮") + self.assertNotEqual(run_one["input_snapshot_id"], run_two["input_snapshot_id"]) + with self.service.connect() as db: + snapshots = store.listing(db, "input_snapshot", "DEMO-B") + runs = store.listing(db, "run", "DEMO-B") + self.assertGreaterEqual(len(snapshots), 2, "两次运行各自冻结快照") + self.assertIn(run_one["id"], {item["id"] for item in runs}, "旧运行必须留存") + self.assertNotEqual(first["revision"], self.service.get_project("DEMO-B")["revision"]) + + def test_same_idempotency_key_replays_the_same_run(self): + # AC-14:同键同载荷返回同一运行 + one = self.service.enqueue("DEMO-B", self.identity(), "重复要求", idempotency_key="k1") + two = self.service.enqueue("DEMO-B", self.identity(), "重复要求", idempotency_key="k1") + self.assertEqual(one["id"], two["id"]) + + def test_same_key_with_different_payload_conflicts(self): + self.service.enqueue("DEMO-B", self.identity(), "要求甲", idempotency_key="k2") + with self.assertRaises(store.Conflict): + self.service.enqueue("DEMO-B", self.identity(), "要求乙", idempotency_key="k2") + + def test_stale_revision_is_rejected(self): + # AC-13 的另一半:依据旧 revision 更新会冲突 + project = self.service.get_project("DEMO-B") + self.service.supplement("DEMO-B", "demo-manager", + {"collection_evidence": "先改一次"}, [], project["revision"], "说明") + with self.assertRaises(store.Conflict): + self.service.supplement("DEMO-B", "demo-manager", + {"collection_evidence": "再用旧版本改"}, [], + project["revision"], "说明") + + +if __name__ == "__main__": + unittest.main()