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

"""S1 事实确认服务：多来源数值冲突的检测、阻断与裁定。

职责 owner（docs/v2/03-detailed-design.md §9.2「事实确认服务」）。

解决的真实缺口（claude_use/T00/04 §3）：
    既有 `assistant_materials.apply_candidate()` 对 `values[field]` / `facts[field]`
    是**后写覆盖**；`sources[field]` 虽累积多个来源，但**无人检测分歧、无人阻断**。
    这正是 BR-23 禁止的行为（"不按上传先后覆盖"）。本模块在**新路径**前置检测，
    **不修改** apply_candidate 的既有语义（调用方 supplement / operations_demo 不变）。

规则（BR-23 / E-17 / AC-24）：
1. 检测：同字段的新候选值与已确认值不同 → 建 FactConflict(unresolved)
2. 阻断：该字段存在 unresolved 冲突时，**不写** values / selected_sources / facts；
   证据仍按 BR-22 记入 sources（来源留痕，值不采用）
3. 裁定：resolve 显式选择候选并要求说明 + 操作人；此后该字段才可确认
4. 不覆盖：再次分歧新建对象并用 supersedes_id 指向旧对象，不改历史
"""

from __future__ import annotations

from contracts.prescreen import ErrorCode, NOTE_MAX
from services import assistant_store as store
from services.assistant_materials import FIELDS, apply_candidate, scalar
from pydantic import ValidationError

KIND = "fact_conflict"
STANDARD_FIELDS = frozenset(FIELDS.values())


class FactConflictUnresolved(ValueError):
    """该字段存在未解决的事实冲突，拒绝确认（E-17）。

    HTTP 语义：作为**确认动作**的拒绝返回 422（INVALID_REQUEST 同族但 code 保持
    FACT_CONFLICT_UNRESOLVED）。注意与**运行级**结果区分——运行结果里未解决的冲突
    保存在 degradations/finding 中，HTTP 仍为 200（01 §5 末）。同一 code 的两个面。
    """

    code = ErrorCode.FACT_CONFLICT_UNRESOLVED


def _check_field(field: str) -> None:
    """标准字段白名单校验留在 service 层（契约有意不查，避免 contracts→services）。"""
    if field not in STANDARD_FIELDS:
        raise ValueError(f"不支持的字段：{field}；请使用标准业务字段")


def _normalized(field: str, value):
    """用与 apply_candidate 相同的标量规范化，使比较与落库口径一致。"""
    return scalar(field, value)


def blocked_fields(db, project_id) -> set[str]:
    """该项目下所有 unresolved 冲突的字段。"""
    return {
        row["field"]
        for row in store.listing(db, KIND, project_id)
        if row["status"] == "unresolved"
    }


def assert_confirmable(db, project_id, fields) -> None:
    """确认前置检查：命中未解决冲突即拒绝（BR-23 阻断）。"""
    hit = sorted(set(fields) & blocked_fields(db, project_id))
    if hit:
        raise FactConflictUnresolved(
            "材料存在不一致，请核对并确认；不得静默选择。涉及字段：" + "、".join(hit)
        )


def _candidate_row(candidate: dict, evidence: dict) -> dict:
    return {
        "evidence_id": evidence["id"],
        "value": None if evidence.get("value") is None else str(evidence["value"]),
        "source": str(candidate.get("source") or evidence.get("source") or "未知来源"),
        "locator": str(candidate.get("locator") or evidence.get("locator") or "未定位"),
        "unit": candidate.get("unit"),
        "currency": candidate.get("currency"),
        "period": candidate.get("period"),
    }


def _validate(payload: dict) -> dict:
    from contracts.prescreen import FactConflict

    try:
        return FactConflict(**payload).model_dump(mode="json")
    except ValidationError as exc:
        first = exc.errors()[0]
        raise ValueError(f"事实冲突记录不合法：{first['msg']}") from exc


def detect(project, candidate: dict, actor: str, reason: str) -> dict | None:
    """检测单个候选是否与已确认值冲突，并就地更新 `project`。

    **本函数不落库**，调用方负责：
        1) `payload = detect(project, candidate, actor, reason)`
        2) `store.put(db, "project", project)` —— 保存证据与（可能的）已采用值
        3) 若 payload 非 None：`create(db, project["id"], payload, actor)`

    返回待落库的 FactConflict 载荷；无冲突返回 None。
    证据在两种情况下都记入 `sources`：冲突时以 `adopt=False` 记录 —— 来源留痕，
    但值**不采用**（BR-22 与 BR-23 同时满足）。
    """
    field = candidate["field"]
    _check_field(field)
    incoming = _normalized(field, candidate.get("value"))
    confirmed = project.get("values", {}).get(field)

    if confirmed is None or incoming is None or str(incoming) == str(confirmed):
        # 首次确认、或双方都未知、或值相同 → 不构成分歧
        apply_candidate(project, candidate, actor, reason)
        return None

    evidence = apply_candidate(project, candidate, actor, reason, adopt=False)
    old_evidence_id = project.get("selected_sources", {}).get(field)
    if old_evidence_id is None:
        # 已确认值却没有来源记录：属数据异常，不静默生成冲突对象
        raise ValueError(f"字段 {field} 已有确认值但缺少来源记录，无法判定冲突")

    payload = {
        "id": store.uid("fc"),
        "project_id": project["id"],
        "data_domain": project.get("data_domain", "demo"),
        "field": field,
        "candidate_evidence_ids": [old_evidence_id, evidence["id"]],
        "candidates": [
            {"evidence_id": old_evidence_id, "value": str(confirmed),
             "source": "已确认值", "locator": "项目事实"},
            _candidate_row(candidate, evidence),
        ],
        "status": "unresolved",
        "created_at": store.now(),
    }
    # revision 由 put() 维护
    return payload


def create(db, project_id, payload: dict, actor: str, supersedes_id: str | None = None) -> dict:
    # revision 由 store.put() 维护；新对象的初值为 1，这里显式带上以通过契约校验
    # （FactConflict.revision 为必填，线格式里必须可见）。
    body = _validate(dict(payload, revision=1, supersedes_id=supersedes_id))
    body = store.put(db, KIND, body)
    store.event(db, project_id, "fact_conflict.created", actor,
                {"conflict_id": body["id"], "field": body["field"]})
    return body


def load(db, conflict_id, project_id=None) -> dict:
    row = store.get(db, KIND, conflict_id)
    if project_id is not None and row["project_id"] != project_id:
        raise KeyError("事实冲突不存在")
    return row


def listing(db, project_id, status=None, limit=20) -> dict:
    rows = store.listing(db, KIND, project_id)
    if status:
        rows = [row for row in rows if row["status"] == status]
    return {"items": rows[:limit], "next_cursor": None if len(rows) <= limit else str(limit)}


def resolve(db, conflict_id, project, selected_evidence_ids, note, actor, expected):
    """显式裁定：写入所选值，并把冲突置为 resolved。

    只允许 unresolved → resolved；已 resolved 再次调用即版本冲突（409）。
    """
    if not isinstance(note, str) or not note.strip():
        raise ValueError("请填写裁定说明")
    if len(note) > NOTE_MAX:
        raise ValueError(f"裁定说明过长（上限 {NOTE_MAX} 字）")

    row = store.get(db, KIND, conflict_id)
    if row["status"] != "unresolved":
        raise store.Conflict("该冲突已裁定，不能重复裁定")

    selected = list(selected_evidence_ids)
    if not selected:
        raise ValueError("请选择采纳的来源")
    unknown = sorted(set(selected) - set(row["candidate_evidence_ids"]))
    if unknown:
        raise ValueError(f"所选的来源不属于该冲突的候选：{unknown}")

    chosen = next(c for c in row["candidates"] if c["evidence_id"] == selected[0])
    field = row["field"]
    _check_field(field)
    value = _normalized(field, chosen["value"])
    project.setdefault("values", {})[field] = value
    project.setdefault("selected_sources", {})[field] = chosen["evidence_id"]
    old_key = next((k for k, v in FIELDS.items() if v == field), field)
    project.setdefault("facts", {})[old_key] = value

    row.update(status="resolved", resolution_note=note,
               selected_evidence_ids=selected, resolved_by=actor, resolved_at=store.now())
    row = store.put(db, KIND, row, expected=expected)
    store.event(db, row["project_id"], "fact_conflict.resolved", actor,
                {"conflict_id": row["id"], "field": field, "selected": selected})
    return row