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

"""T07-b 专项判断的统一入口：disabled 路径与工具策略守卫。

依据：docs/v2/11-judgment-providers.md §3–§6、03-detailed-design.md §9.3；
BR-09/10/17/18/20、E-06/08/10/13/14/15、AC-09/10/21。

本批**只做 disabled 路径**（T07 卡片要求"先 disabled 专项路径"）：真正的 Jev /
本地适配器属 T08，这里不预造。

三条不可让渡的规则：
1. **关闭就明确关闭**：disabled 时已计划题目记 `skipped + disabled`，运行记录关闭
   原因，绝不静默产生判断，也绝不因为"关了"就伪造高分（BR-09 / AC-21）。
2. **失败不补分**：服务失效/超时/畸形输出记 `failed + error_code` 与降级记录，
   `answer` 必须为空；不生成默认分值（AC-09 / E-06 / E-15）。
3. **工具白名单来自冻结的 Skill**：材料正文里写什么都**不能**扩展工具集合
   （BR-10 / AC-10）。工具 = Skill 声明 ∩ 注册表白名单，二者取交，不取并。
"""

from __future__ import annotations

from contracts.prescreen import (
    ErrorCode,
    Judgment,
    JudgmentAnswer,
    JudgmentConfig,
    NativeMetric,
)
from services import assistant_store as store
from services.component_registry import TOOLS
# T08：适配层的策略拒绝用异常表达，在这里统一映射成 Judgment 记录。
# judgment_adapters 只依赖 contracts，不反向依赖本模块，因此没有循环导入。
from services.judgment_adapters import (
    CheckpointMismatch,
    ContextBudgetExceeded,
    EgressNotApproved,
    TaskNotValidated,
)

DISABLED_REASON_DEFAULT = "本次运行未启用专项判断；主审仍按规则、工具与经验形成意见"

# 仅这些错误码表示"提供方不可用"，可按预算重试一次（03 §5）。
RETRYABLE_PROVIDER_ERRORS = (ErrorCode.JUDGMENT_UNAVAILABLE,)


def resolve_config(enabled: bool, profile_id: str | None = None,
                   profile_revision: int | None = None, *, known_profiles=(),
                   provider: str = "local") -> JudgmentConfig:
    """解析并冻结本次运行的专项配置（BR-20：切换只影响新运行）。

    `known_profiles` 是管理员在后端配置里登记的可用配置；P0 为空，因此在
    enabled=True 且未给 ID 时会明确报错（E-13），而不是悄悄关闭专项。
    """
    if not enabled:
        if profile_id:
            # 开关 false 同时传 ID 属 E-07，避免静默忽略用户选择（11 §3）
            raise ValueError("专项已关闭却指定了配置 ID；请明确启用或清空选择")
        return JudgmentConfig(enabled=False, provider="disabled",
                              disabled_reason=DISABLED_REASON_DEFAULT)

    if not profile_id:
        default = next((item for item in known_profiles if item.get("is_default")), None)
        if default is None:
            raise ValueError("当前没有可用的专项模型配置，请选择可用配置或关闭专项核查")
        profile_id, profile_revision = default["id"], default.get("revision", 1)

    profile = next((item for item in known_profiles if item.get("id") == profile_id), None)
    if profile is None:
        raise ValueError("当前专项模型配置不可用，请选择可用配置或关闭专项核查")
    if profile.get("readiness") != "ready":
        raise ValueError("该专项模型配置尚未通过验证，暂不可用")
    if provider not in ("jev", "local"):
        raise ValueError("专项提供方只能是 jev 或 local")
    if profile.get("provider") != provider:
        raise ValueError("所选配置的提供方与请求不一致")
    return JudgmentConfig(
        enabled=True, provider=provider, profile_id=profile_id,
        profile_revision=profile_revision or profile.get("revision", 1),
        model_version=profile.get("model_version") or "unresolved",
        adapter_version=profile.get("adapter_version") or "not_applicable",
        prompt_version=profile.get("prompt_version") or "not_applicable",
        mapping_version=profile.get("mapping_version") or "not_applicable",
        # T08：权重修订随配置一起冻结；缺失记 unresolved 而不是留空猜测（AC-39）
        checkpoint_ref=profile.get("checkpoint_ref") or "unresolved")


def allowed_tools(skill_body: dict, registry_tools=None) -> set[str]:
    """工具集合 = Skill 声明 ∩ 注册表白名单（取交，不取并）。

    材料正文、运行补充说明、专项模型输出都**不参与**这个计算，因此无法借它们
    扩展权限（BR-10 / AC-10）。
    """
    declared = set(skill_body.get("allowed_tools") or ())
    return declared & set(registry_tools or TOOLS)


def assert_tools_within_policy(requested, skill_body, registry_tools=None) -> None:
    """任何越界工具请求都必须被拒，而不是"尽量满足"。"""
    permitted = allowed_tools(skill_body, registry_tools)
    outside = sorted(set(requested or ()) - permitted)
    if outside:
        raise ValueError(f"工具超出当前 Skill 授权范围，已拒绝：{outside}")


def _skipped(run_id: str, question: dict, config: JudgmentConfig, reason: str) -> dict:
    return {"id": store.uid("judgment"), "run_id": run_id,
            "question_id": question["id"], "question_version": question["version"],
            "type": question["type"], "status": "skipped", "answer": None,
            "evidence_ids": list(question.get("evidence_ids", [])),
            "provider": "disabled", "profile_id": None, "profile_revision": None,
            "model_version": None, "native_metrics": [], "error_code": None,
            "skip_reason": reason}


def judge(config: JudgmentConfig, run_id: str, question: dict,
          provider_call=None) -> dict:
    """执行一道题的专项判断。

    `provider_call` 是 T08 才接的适配器；本批只覆盖 disabled 与失败/畸形路径。
    """
    if config.provider == "disabled":
        return _skipped(run_id, question, config, "disabled")

    if not question.get("evidence_ids"):
        # 无证据不请求模型：按缺证处理并明确返回 unknown（11 §4）
        return {"id": store.uid("judgment"), "run_id": run_id,
                "question_id": question["id"], "question_version": question["version"],
                "type": question["type"], "status": "completed",
                "answer": {"kind": "unknown", "value": None, "reason_code": "missing_evidence"},
                "evidence_ids": [], "provider": config.provider,
                "profile_id": config.profile_id, "profile_revision": config.profile_revision,
                "model_version": config.model_version, "native_metrics": [],
                "error_code": None, "skip_reason": None}

    if provider_call is None:
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)

    try:
        raw = provider_call(config, question)
    except TimeoutError:
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)
    except ContextBudgetExceeded:
        # E-25：超预算不外发、不静默截断；evidence_ids 仍保留原引用范围（AC-40）
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_CONTEXT_EXCEEDED)
    except TaskNotValidated:
        # 未获准的任务不给"未完成"以外的结论（AC-38）
        return {**_failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE),
                "status": "skipped", "error_code": None, "skip_reason": "unsupported"}
    except EgressNotApproved:
        # 未批准出口：不发送，记录策略拒绝（AC-16/19）
        return {**_failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE),
                "status": "skipped", "error_code": None, "skip_reason": "not_authorized"}
    except CheckpointMismatch:
        # 实际权重与冻结修订不符：拒绝并记失败，不得静默换模型（AC-39）
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
    except Exception:
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_UNAVAILABLE)

    return _normalise(run_id, question, config, raw)


def _failed(run_id: str, question: dict, config: JudgmentConfig, code: ErrorCode) -> dict:
    """失败必须留痕且**不补分**：answer 为空，不生成任何默认取值。"""
    return {"id": store.uid("judgment"), "run_id": run_id,
            "question_id": question["id"], "question_version": question["version"],
            "type": question["type"], "status": "failed", "answer": None,
            "evidence_ids": list(question.get("evidence_ids", [])),
            "provider": config.provider, "profile_id": config.profile_id,
            "profile_revision": config.profile_revision,
            "model_version": config.model_version, "native_metrics": [],
            "error_code": code, "skip_reason": None}


def _normalise(run_id: str, question: dict, config: JudgmentConfig, raw: dict) -> dict:
    """校验并归一化提供方输出；畸形即 E-15，不猜一个分值补齐。"""
    if not isinstance(raw, dict):
        return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
    answer_kind = raw.get("kind")
    if answer_kind == "unknown":
        if raw.get("reason_code") not in ("missing_evidence", "inconclusive"):
            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)
        answer = JudgmentAnswer(kind="unknown", reason_code=raw["reason_code"])
    else:
        if answer_kind != question["type"]:
            # 题型不符（含"把连续输出硬当 binary"）属能力不支持，不是失败
            return {**_failed(run_id, question, config, ErrorCode.JUDGMENT_CAPABILITY_UNSUPPORTED),
                    "status": "skipped", "error_code": None, "skip_reason": "unsupported"}
        try:
            answer = JudgmentAnswer(kind=answer_kind, value=raw.get("value"))
        except Exception:
            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)

    metrics = []
    for item in raw.get("native_metrics") or ():
        try:
            metrics.append(NativeMetric(**item).model_dump(mode="json"))
        except Exception:
            return _failed(run_id, question, config, ErrorCode.JUDGMENT_OUTPUT_INVALID)

    body = {"id": store.uid("judgment"), "run_id": run_id,
            "question_id": question["id"], "question_version": question["version"],
            "type": question["type"], "status": "completed",
            "answer": answer.model_dump(mode="json"),
            "evidence_ids": list(question.get("evidence_ids", [])),
            "provider": config.provider, "profile_id": config.profile_id,
            "profile_revision": config.profile_revision,
            "model_version": config.model_version, "native_metrics": metrics,
            "error_code": None, "skip_reason": None,
            "actual_checkpoint_ref": raw.get("actual_checkpoint_ref")}
    return Judgment(**body).model_dump(mode="json")


def plan_questions(config: JudgmentConfig, questions) -> dict:
    """关闭专项时把已计划题目统一记 skipped，并给出运行级关闭原因。"""
    if config.provider != "disabled":
        return {"judgments": [], "disabled_reason": None}
    return {"judgments": [_skipped("", question, config, "disabled") for question in questions or ()],
            "disabled_reason": config.disabled_reason}