vent_agent_2.0/runtime/interrupts.py

"""中断展示与恢复共用一个载荷和标识，批准范围等于卡片展示范围。"""
import hashlib
import json

from dataclasses import dataclass
from typing import Any
from obs import get_logger

logger = get_logger(__name__)


@dataclass(frozen=True)
class ResumeDecision:
    interrupt_token: str = ""
    approved: bool | None = None
    status: str | None = None
    answers: list[str] | None = None
    error: str | None = None


class ResumeError(ValueError):
    def __init__(self, code: str, message: str):
        self.code = code
        super().__init__(message)



def pack(item):
    value = getattr(item, "value", item)
    payload = dict(value) if isinstance(value, dict) else {"message": str(value)}
    if getattr(item, "id", None):
        payload["_interrupt_id"] = item.id
    return payload


def token_for(payload):
    return hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode()).hexdigest()


def interrupt_event(writer, payload):
    payload = payload if isinstance(payload, dict) else {"message": str(payload)}
    token = token_for(payload)
    if payload.get("kind") == "ask_user":
        return writer.event("interrupt", interrupt_token=token,
                            ask_user={"questions": payload["questions"]})
    actions = payload.get("action_requests") or []
    if actions:
        first = actions[0]
        return writer.interrupt_tool_approval(
            tool=first.get("name", ""), args=first.get("args"),
            reason=first.get("description", ""), side_effect="批准后执行卡片中的全部操作",
            actions=actions, pending_count=len(actions), interrupt_token=token,
        )
    return writer.event("interrupt", message=payload.get("message", "智能体暂停，等待输入"),
                        interrupt_token=token)


def resume_value(request, state):
    items = [item for task in getattr(state, "tasks", ()) for item in getattr(task, "interrupts", ())]
    if not items:
        raise ResumeError(**{"code": "no_pending_interrupt", "message": "没有待确认请求"})
    selected = next((item for item in items if token_for(pack(item)) == request.interrupt_token), None)
    if selected is None:
        raise ResumeError(**{"code": "stale_interrupt", "message": "确认请求已变化，请刷新后重试"})
    payload = pack(selected)
    if payload.get("kind") == "ask_user":
        if request.status == "cancelled":
            value = {"cancelled": True}
        else:
            if not request.answers or len(request.answers) != len(payload["questions"]) or any(not a.strip() for a in request.answers):
                raise ResumeError(**{"code": "invalid_answers", "message": "请回答全部问题"})
            value = {"answers": request.answers}
    else:
        if request.approved is None and request.status not in {"cancelled", "error"}:
            raise ResumeError(**{"code": "invalid_decision", "message": "请选择批准或拒绝"})
        value = official_resume(request, len(payload.get("action_requests") or []))
    interrupt_id = getattr(selected, "id", None)
    return {interrupt_id: value} if interrupt_id else value


def official_resume(payload: ResumeDecision, pending_count: int) -> dict[str, Any]:
    """把领域层的 ResumeDecision 翻译成官方 HITL 的 ``{"decisions": [...]}``。

    官方要求 decisions 与 action_requests **数量严格一一对应**，数量不符会
    抛错，所以这里按 ``pending_count`` 复制同一决策。

    翻译表：
    - ``approved=True`` → ``approve``
    - ``approved=False`` → ``reject``（带拒绝理由）
    - ``status=cancelled`` → ``reject``（用户取消等价于不执行）
    - ``status=error`` → ``reject``（前端恢复出错，不执行更安全）
    - 有 ``answers`` → ``respond``（把人的回答作为工具结果）
    """
    count = max(pending_count, 1)
    if count > 1 and payload.approved is True and not payload.interrupt_token:
        raise ResumeError(**{"code": "approval_refresh_required", "message": "请刷新并确认完整操作列表"})

    if payload.approved is True:
        decision: dict[str, Any] = {"type": "approve"}
    elif payload.approved is False:
        decision = {"type": "reject", "message": "用户拒绝了该操作，请勿重试，改为说明情况"}
    elif payload.status == "cancelled":
        decision = {"type": "reject", "message": "用户取消了该操作"}
    elif payload.status == "error":
        decision = {
            "type": "reject",
            "message": f"前端恢复出错，该操作未执行：{payload.error or '未提供详情'}",
        }
    elif payload.answers:
        decision = {"type": "respond", "message": "\n".join(payload.answers)}
    else:
        decision = {"type": "reject", "message": "未收到明确答复，该操作未执行"}

    return {"decisions": [dict(decision) for _ in range(count)]}


async def pending_interrupts(agent: Any, config: dict[str, Any]) -> list[Any]:
    """读取当前挂起的中断载荷列表。

    供两处复用：``GET /chat/interrupt``（刷新页面后补渲染卡片）与
    ``POST /chat/resume``（在构造 decisions 前先知道有几个待审批动作——
    官方 HITL 要求 decisions 与 action_requests **数量严格一一对应**，
    数量对不上会直接抛错）。
    """
    get_state = getattr(agent, "aget_state", None)
    if get_state is None:
        return []
    try:
        state = await get_state(config)
    except Exception as exc:
        logger.warning("读取图状态失败：%s: %s", type(exc).__name__, exc)
        raise
    found: list[Any] = []
    for task in getattr(state, "tasks", None) or ():
        for item in getattr(task, "interrupts", None) or ():
            found.append(pack(item))
    return found