"""T09 推进沙盘契约:假设动作与情景结果。 Owner:本文件(docs/v2/03-detailed-design.md §9.3 Simulation、 docs/v2/04-ui-spec.md §9、13-probability-prototype.md)。 三条不可让渡的规则: 1. **模拟不改真实运行**:情景是独立对象,基线以 `baseline_probability_view` 副本随行携带;任何情况下都不写回 `ScreeningRun.probability`(BR-11 / AC-11)。 2. **演示值只属于 demo 域**:`probability_view.demo_estimate` 在 production 域直接构造失败(E-21 / AC-30);真实估计在 P0 恒为空。 3. **组合不是加和**:结果由完整动作组合选出(`selection_hash`), 不按单个动作的增量累加(BR-30 / AC-31)。 """ from __future__ import annotations import re from typing import Annotated, Literal from pydantic import Field, model_validator from contracts.prescreen import ( DESC_MAX, ID_MAX, AwareDatetime, Contract, DataDomain, ErrorCode, ProbabilityView, ) SHA256 = re.compile(r"^[0-9a-f]{64}$") Identifier = Annotated[str, Field(min_length=1, max_length=ID_MAX)] MAX_ASSUMPTIONS = 8 class SimulationAssumption(Contract): """一条假设动作。必须写明它改变哪个维度、改到什么阶段、前提是什么。 `target_dimension_id` 只能指向本次冻结方案里真实存在的维度:假设不能凭空 创造维度,也不能顺手改动别的轴(13 §P-17)。 """ key: Identifier label: str = Field(min_length=1, max_length=DESC_MAX) target_dimension_id: Identifier assumed_stage: int = Field(ge=0, le=4) premise: str = Field(min_length=1, max_length=DESC_MAX) class ScenarioDimensionState(Contract): """情景下的维度阶段。基线值与情景值同时给出,界面据此画虚线与实线。""" id: Identifier baseline_stage: int | None = Field(default=None, ge=0, le=4) scenario_stage: int | None = Field(default=None, ge=0, le=4) assumed: bool = False class Simulation(Contract): """一次情景推演的结果视图。**不是**对真实运行的修改。""" id: Identifier run_id: Identifier project_id: Identifier data_domain: DataDomain revision: int = Field(ge=1) # 完整动作组合的摘要:结果必须对应当前选择,而不是"最新到达的响应" selection_hash: str = Field(pattern=SHA256.pattern) status: Literal["queued", "computing", "ready", "failed"] assumptions: list[SimulationAssumption] = Field(default_factory=list, max_length=MAX_ASSUMPTIONS) # 基线副本:模拟结束后界面据此恢复,真实运行未被触碰 baseline_probability_view: ProbabilityView probability_view: ProbabilityView | None = None dimensions: list[ScenarioDimensionState] = Field(default_factory=list, max_length=32) remaining_inputs: list[str] = Field(default_factory=list, max_length=200) reason: str = Field(default="", max_length=DESC_MAX) error_code: ErrorCode | None = None frozen_versions: dict = Field(default_factory=dict) created_at: AwareDatetime @model_validator(mode="after") def _check_simulation(self) -> Simulation: if self.status == "ready" and self.probability_view is None: raise ValueError("就绪的情景必须给出概率视图") if self.status == "failed": if self.error_code is None: raise ValueError("失败的情景必须带错误码,不能只留空结果") if self.probability_view is not None: raise ValueError("失败的情景不得显示任何结果") if self.status in ("queued", "computing") and self.probability_view is not None: raise ValueError("情景尚未算完时不得先给一个乐观增幅") # 演示值只在 demo 域:越域直接构造失败,而不是靠调用方记得不传(AC-30) for view in (self.probability_view, self.baseline_probability_view): demo = (view.demo_estimate if view else None) or None if demo is not None and self.data_domain != DataDomain.demo: raise ValueError("演示估计只允许出现在演示数据域(E-21)") keys = [item.key for item in self.assumptions] if len(set(keys)) != len(keys): raise ValueError("同一假设动作不得重复计入组合") targets = [item.target_dimension_id for item in self.assumptions] if len(set(targets)) != len(targets): raise ValueError("一次情景只允许每个维度有一个假设动作,冲突假设应被拒绝") return self def unchanged_dimensions(self) -> list[str]: """未被假设触碰的维度 id:界面必须显示它们没有变化,而不是消失。""" return [item.id for item in self.dimensions if not item.assumed]