148 lines
7.1 KiB
Python
148 lines
7.1 KiB
Python
 
|
"""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 与修订")
|