mini-openclaw-v2/backend/services/prescreen_sandbox.py

"""T09 推进沙盘服务：把"假设补件"算成独立情景，不触碰真实运行。

依据：docs/v2/13-probability-prototype.md §2/§4、04-ui-spec.md §5/§9、
03-detailed-design.md §9.3；BR-11/29/30、E-21/22、AC-11/12/30/31/44。

设计要点：
1. **演示值查表，不累加**。结果由完整动作组合选出（`selection_hash`）。表里
   00/10/01/11 是四条独立条目，不是 42+16+9=67 的算术和；`assert_not_additive`
   把这条约束写成可执行断言，防止后来者把表换成"每项加分"。
2. **真实运行只读**。`simulate()` 接受运行视图 dict，返回新对象；基线的演示值
   以副本随行携带，任何时候都不写回运行的 `probability`。
3. **越域即拒绝**。production 域不得有演示估计（E-21）；假设指向不存在的维度、
   同一维度两个互斥假设，都按 E-22 拒绝，而不是忽略掉一个。
"""

from __future__ import annotations

import hashlib
import json

from contracts.prescreen import DataDomain, ErrorCode, ProbabilityView
from contracts.simulation import Simulation, SimulationAssumption
from services import assistant_store as store

DEMO_TABLE_VERSION = "DEMO-TABLE-v1"
DEMO_SOURCE = "synthetic_demo"
# 说明文字固定：界面必须紧邻数值展示，避免被当作真实概率（DESIGN.md §38）
DEMO_LABEL = "人工设定，仅演示交互；未经历史数据校准"

STAGE_NAMES = ("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核")


class DemoEstimateForbidden(ValueError):
    """当前数据域不允许使用演示概率（E-21）。"""


class AssumptionConflict(ValueError):
    """所选推进条件冲突、缺少前提或指向不存在的维度（E-22）。"""


# 假设动作登记表。`dimension_id` 必须与冻结方案里的维度 id 一致，否则拒绝。
ASSUMPTIONS: dict[str, dict] = {
    "assume_bank_flow": {
        "label": "假设银行流水逐笔匹配",
        "dimension_id": "DC-CASH",
        "assumed_stage": 3,
        "premise": "假设核验支持账期变化解释；只改变回款核验维度",
    },
    "assume_title": {
        "label": "假设设备权属证明一致",
        "dimension_id": "DC-TITLE",
        "assumed_stage": 3,
        "premise": "假设合同、发票与设备清单核验一致；只改变权属维度",
    },
}

# 演示情景表（13 §3）。key = 流水假设 + 权属假设的组合位。
DEMO_SCENARIOS: dict[str, dict] = {
    "00": {"value": 42, "remaining": ["回款核验仍待核实", "设备权属仍待核实"],
           "reason": "未选择假设。回款与设备权属证据仍有缺口，保持演示基线。"},
    "10": {"value": 58, "remaining": ["设备权属仍待核实"],
           "reason": "假设银行流水逐笔核验后支持账期变化解释，回款不确定性减小；"
                     "设备权属缺口仍在。演示变化 +16 个百分点。"},
    "01": {"value": 51, "remaining": ["实际回款仍待核实"],
           "reason": "假设合同、发票与清单一致且经核验，权属争议得到回应；"
                     "实际回款仍缺银行流水。演示变化 +9 个百分点。"},
    "11": {"value": 64, "remaining": ["仍需整体人工复核"],
           "reason": "假设回款与权属核验均支持项目解释。组合情景为 64%，变化 +22 "
                     "个百分点；组合影响不简单相加，仍需综合复核其他风险。"},
}


def scenario_key(keys) -> str:
    """把动作组合压成表键：按登记顺序取位，未选为 0。"""
    order = list(ASSUMPTIONS)
    unknown = sorted(set(keys or ()) - set(order))
    if unknown:
        raise AssumptionConflict(f"未登记的假设动作：{unknown}")
    return "".join("1" if item in set(keys or ()) else "0" for item in order)


def selection_hash(run_id: str, revision: int, keys) -> str:
    """完整动作组合的摘要。前端只接受与当前选择一致的结果（AC-31）。"""
    payload = json.dumps({"run_id": run_id, "revision": revision,
                          "key": scenario_key(keys)}, ensure_ascii=False,
                         sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def assert_demo_domain(data_domain) -> None:
    value = data_domain.value if hasattr(data_domain, "value") else str(data_domain)
    if value != DataDomain.demo.value:
        raise DemoEstimateForbidden("当前数据域不允许使用演示概率；真实概率需完成校准后另立验证")


def assert_not_additive() -> None:
    """组合值必须来自查表：任何"单项增量相加"的实现都会在这里失败。"""
    baseline = DEMO_SCENARIOS["00"]["value"]
    pair = DEMO_SCENARIOS["11"]["value"]
    added = baseline + (DEMO_SCENARIOS["10"]["value"] - baseline) \
        + (DEMO_SCENARIOS["01"]["value"] - baseline)
    if pair == added:
        raise AssertionError("演示情景表退化成单项加和，须回到按完整组合查表（BR-30）")


def _view(snapshot_id: str, key: str, as_of: str | None = None) -> ProbabilityView:
    row = DEMO_SCENARIOS[key]
    return ProbabilityView(
        snapshot_id=snapshot_id, as_of=as_of or store.now(),
        demo_estimate={"value": row["value"] / 100, "source": DEMO_SOURCE,
                       "scenario_table_version": DEMO_TABLE_VERSION,
                       "scenario_key": key, "label": DEMO_LABEL})


def _target(key, dimension_ids):
    aliases = {'assume_bank_flow': ('DC-CASH', 'corporate.cashflow'),
               'assume_title': ('DC-TITLE', 'corporate.title')}
    return next((name for name in aliases[key] if name in set(dimension_ids)), None)


def _assumptions_for(keys, dimension_ids) -> list[SimulationAssumption]:
    known = set(dimension_ids)
    chosen: list[SimulationAssumption] = []
    for key in ASSUMPTIONS:
        if key not in set(keys or ()):
            continue
        item = dict(ASSUMPTIONS[key], dimension_id=_target(key, known) or ASSUMPTIONS[key]["dimension_id"])
        if item["dimension_id"] not in known:
            raise AssumptionConflict(
                f"假设动作 {key} 指向本次方案不存在的维度 {item['dimension_id']}")
        chosen.append(SimulationAssumption(
            key=key, label=item["label"], target_dimension_id=item["dimension_id"],
            assumed_stage=item["assumed_stage"], premise=item["premise"]))
    return chosen


def _dimension_states(run_view: dict, assumptions) -> list[dict]:
    """只覆盖被假设的维度；其余维度原样保留并标为未变（13 §P-17）。"""
    assumed = {item.target_dimension_id: item for item in assumptions}
    rows = []
    for item in run_view.get("dimension_results") or ():
        hit = assumed.get(item["id"])
        rows.append({"id": item["id"],
                     "baseline_stage": item.get("stage"),
                     "scenario_stage": hit.assumed_stage if hit else item.get("stage"),
                     "assumed": hit is not None})
    return rows


def failed(run_view: dict, keys, message: str, code: ErrorCode,
           revision: int | None = None) -> dict:
    """情景失败：保留基线、标未完成，不得回退成"演示表默认值"（13 §44）。"""
    revision = revision or int(run_view.get("revision") or 1)
    baseline = _baseline_view(run_view)
    body = Simulation(
        id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"],
        data_domain=run_view["data_domain"], revision=revision,
        selection_hash=selection_hash(run_view["id"], revision, keys),
        status="failed", assumptions=[], baseline_probability_view=baseline,
        probability_view=None, dimensions=_dimension_states(run_view, []),
        remaining_inputs=list(run_view.get("missing_inputs") or ()),
        reason=message, error_code=code,
        frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now())
    return body.model_dump(mode="json")


def available_assumptions(run_view: dict) -> list[dict]:
    """列出本次方案可用的假设动作。

    只返回目标维度真实存在于该运行的动作：界面因此不会给出点了必然失败的选项，
    服务端也不靠"忽略无效动作"来掩饰不匹配（E-22）。
    """
    known = {item["id"] for item in run_view.get("dimension_results") or ()}
    return [{"key": key, "label": item["label"], "dimension_id": _target(key, known),
             "assumed_stage": item["assumed_stage"], "premise": item["premise"]}
            for key, item in ASSUMPTIONS.items() if _target(key, known)]


def _baseline_view(run_view: dict) -> ProbabilityView:
    snapshot_id = run_view.get("snapshot_id") or run_view.get("input_snapshot_id")
    if not snapshot_id:
        raise ValueError("运行缺少输入快照，无法建立情景基线")
    return _view(snapshot_id, "00")


def simulate(run_view: dict, keys, revision: int | None = None) -> dict:
    """按完整动作组合产出一条情景；真实运行只被读取，不被改写。"""
    assert_demo_domain(run_view["data_domain"])
    revision = revision or int(run_view.get("revision") or 1)
    dimension_ids = [item["id"] for item in run_view.get("dimension_results") or ()]
    assumptions = _assumptions_for(keys, dimension_ids)
    key = scenario_key(keys)
    row = DEMO_SCENARIOS[key]
    baseline = _baseline_view(run_view)

    body = Simulation(
        id=store.uid("simulation"), run_id=run_view["id"], project_id=run_view["project_id"],
        data_domain=run_view["data_domain"], revision=revision,
        selection_hash=selection_hash(run_view["id"], revision, keys),
        status="ready", assumptions=assumptions,
        baseline_probability_view=baseline,
        probability_view=_view(baseline.snapshot_id, key, baseline.as_of),
        dimensions=_dimension_states(run_view, assumptions),
        remaining_inputs=list(row["remaining"]), reason=row["reason"],
        frozen_versions=dict(run_view.get("versions") or ()), created_at=store.now())
    return body.model_dump(mode="json")


def assert_baseline_untouched(run_view: dict, simulation: dict) -> None:
    """装配出口的自检：情景不得改写运行的概率字段（BR-11 / AC-11）。"""
    if run_view.get("probability") is not None:
        raise AssertionError("情景推演不得给真实运行写入概率")
    if (run_view.get("probability_view") or {}).get("real_estimate") is not None:
        raise AssertionError("情景推演不得给真实运行写入真实估计")
    baseline = simulation.get("baseline_probability_view") or {}
    if baseline.get("real_estimate") is not None:
        raise AssertionError("情景基线副本的真实估计必须为空")