agent-management/backend/hermes-bridge/internal_context_tools.py

"""
工作流节点上下文查询工具（Python 端实现）。

3 个工具供 Hermes Agent 在执行工作流节点任务时调用，通过 HTTP 反向查询 Java 端的
`/internal/context/{path,output,meta}` 接口，获取前驱节点的路径图 / 输出 / 中间数据。

设计要点：
- 工具永久注册到全局 registry（启动时一次性，幂等）
- check_fn 检查 contextvar `_CONTEXT_QUERY_CTX`：
  - 已设置（含 context_query 的请求）→ 工具对 agent 可见
  - 未设置（普通对话请求）→ 工具不可见，避免污染
- handler 内通过 contextvar 读取 base_url/token/run_id/visit_uuid，
  避免在工具签名中暴露这些敏感字段给 LLM
- 返回格式（用户决策 Q3）：自然语言摘要 + JSON 代码块

协议契约：参见 `docs/specs/internal-context-query-spec.md`

日志要点：
- 入口：每个 handler 被调用时打印工具名 + 关键参数（不含 token）
- HTTP：_http_post 打印 URL + 状态码 + 耗时 + 响应字节数
- 出口：返回内容的长度与摘要预览（前 100 字符）
- token 永远不打印；visit_uuid 仅打印前 8 位
"""

import json
import logging
import time
import urllib.error
import urllib.request
from typing import Optional

logger = logging.getLogger("internal-context-tools")

# Java 端点路径（前缀 /internal/context）
_PATH_ENDPOINT = "/internal/context/path"
_OUTPUT_ENDPOINT = "/internal/context/output"
_META_ENDPOINT = "/internal/context/meta"

# HTTP 超时（秒）：内存查询亚秒级，DB 回退 < 300ms，5s 兜底足够
_HTTP_TIMEOUT = 5.0

# 工具所属 toolset，独立于 hermes-agent 内置 toolset，避免命名冲突
_TOOLSET = "workflow_context"

# 日志预览截断长度（字符）
_LOG_PREVIEW_LEN = 100


def _short_visit(uuid_str) -> str:
    """visit_uuid 截断为前 8 位用于日志（避免完整 UUID 噪声）。"""
    if not uuid_str:
        return "<none>"
    return str(uuid_str)[:8]


def _get_cfg():
    """从 hermes_bridge 模块的 contextvar 读取当前请求的 context_query 配置。

    返回 None 表示未启用（普通对话场景），handler 与 check_fn 都依赖此函数。
    延迟 import 避免循环依赖（hermes_bridge 启动时会 import 本模块注册工具）。
    """
    try:
        from hermes_bridge import _CONTEXT_QUERY_CTX
        return _CONTEXT_QUERY_CTX.get(None)
    except Exception:
        return None


def _http_post(base_url: str, token: str, path: str, payload: dict) -> tuple:
    """调用 Java 内部接口。

    返回 (status_code, data_or_error)：
    - 成功：(200, <dict from Result.data>)
    - 业务失败：(non-200, {"message": "...", "code": ...})
    - 网络/异常：(-1, {"message": "..."})
    """
    url = base_url.rstrip("/") + path
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "X-Context-Token": token,
        },
    )
    start = time.time()
    try:
        with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp:
            body = resp.read().decode("utf-8")
            elapsed_ms = int((time.time() - start) * 1000)
            # Java Result<T>: {code: 200, message: ..., data: T}
            parsed = json.loads(body) if body else {}
            code = parsed.get("code", resp.status)
            if code == 200:
                logger.info(
                    "[CtxQuery] POST %s -> 200, %d bytes, %dms, payload=%s",
                    url, len(body), elapsed_ms, _truncate_payload_for_log(payload),
                )
                return 200, parsed.get("data") or {}
            logger.warning(
                "[CtxQuery] POST %s -> code=%d, %d bytes, %dms, msg=%s, payload=%s",
                url, code, len(body), elapsed_ms,
                parsed.get("message", ""), _truncate_payload_for_log(payload),
            )
            return code, {"message": parsed.get("message", "未知错误"), "code": code}
    except urllib.error.HTTPError as e:
        elapsed_ms = int((time.time() - start) * 1000)
        try:
            err_body = e.read().decode("utf-8")
            parsed = json.loads(err_body) if err_body else {}
            msg = parsed.get("message", err_body)
        except Exception:
            parsed = {}
            msg = str(e)
        logger.warning(
            "[CtxQuery] POST %s -> HTTPError %d, %dms, msg=%s, payload=%s",
            url, e.code, elapsed_ms, msg, _truncate_payload_for_log(payload),
        )
        return e.code, {"message": msg, "code": e.code}
    except Exception as e:
        elapsed_ms = int((time.time() - start) * 1000)
        logger.warning(
            "[CtxQuery] POST %s -> EXC %dms, err=%s, payload=%s",
            url, elapsed_ms, e, _truncate_payload_for_log(payload),
        )
        return -1, {"message": f"请求失败: {e}", "code": -1}


def _truncate_payload_for_log(payload: dict) -> str:
    """日志中打印 payload 的安全摘要（不含 token，visit_uuid 截断）。"""
    if not payload:
        return "{}"
    safe = {}
    for k, v in payload.items():
        if k in ("visitUuid", "visit_uuid"):
            safe[k] = _short_visit(v)
        else:
            safe[k] = v
    s = json.dumps(safe, ensure_ascii=False)
    return s if len(s) <= 200 else s[:200] + "..."


def _format_success(summary: str, data: dict) -> str:
    """格式化为「自然语言摘要 + JSON 代码块」。"""
    if not data:
        return summary
    return "{summary}\n\n```json\n{json}\n```".format(
        summary=summary,
        json=json.dumps(data, ensure_ascii=False, indent=2),
    )


def _format_error(message: str) -> str:
    """错误场景只返回自然语言描述（不带 JSON），避免 LLM 误以为有数据可引用。"""
    return message


def _log_handler_return(tool_name: str, result: str, extra: str = ""):
    """打印 handler 返回值的长度与预览（前 100 字符）。"""
    preview = result[:_LOG_PREVIEW_LEN].replace("\n", "\\n")
    if len(result) > _LOG_PREVIEW_LEN:
        preview += "..."
    logger.info(
        "[CtxQuery] %s 返回 %d 字符%s: %s",
        tool_name, len(result), f" ({extra})" if extra else "", preview,
    )


# ---------------------------------------------------------------------------
# Tool handlers
# ---------------------------------------------------------------------------

def _handle_get_predecessor_paths(args, **kw):
    tool_name = "get_predecessor_paths"
    cfg = _get_cfg()
    if cfg is None:
        logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name)
        return _format_error("上下文查询未启用：本次运行未注入 context_query 配置")
    logger.info(
        "[CtxQuery] %s 被调用: run_id=%s, visit_uuid=%s",
        tool_name, cfg.get("run_id"), _short_visit(cfg.get("visit_uuid")),
    )
    status, data = _http_post(
        cfg["base_url"], cfg["token"], _PATH_ENDPOINT,
        {"runId": cfg["run_id"], "visitUuid": cfg["visit_uuid"]},
    )
    if status == 401:
        _log_handler_return(tool_name, "token_invalid", "401")
        return _format_error("token 无效或已过期，无法查询前驱路径")
    if status == 403:
        _log_handler_return(tool_name, "forbidden", "403")
        return _format_error("runId 与 token 不匹配，禁止查询")
    if status == 404:
        _log_handler_return(tool_name, "not_found", "404")
        return _format_error(
            "未找到当前 visit 的前驱路径（runId={}, visitUuid={}）".format(
                cfg["run_id"], _short_visit(cfg["visit_uuid"]),
            )
        )
    if status != 200:
        _log_handler_return(tool_name, "error:" + str(status), str(status))
        return _format_error("查询前驱路径失败：{}".format(data.get("message", "未知错误")))
    node_count = len(data.get("nodeIds") or [])
    path_count = len(data.get("paths") or [])
    backtrack = data.get("backtrackEdgeKeys") or []
    summary = "当前节点的前驱访问路径图：共 {} 个节点、{} 条路径".format(
        node_count, path_count,
    )
    if backtrack:
        summary += "（含 {} 条智能回边）".format(len(backtrack))
    summary += "。"
    result = _format_success(summary, data)
    _log_handler_return(tool_name, result, "nodes={},paths={}".format(node_count, path_count))
    return result


def _handle_get_node_output(args, **kw):
    tool_name = "get_node_output"
    cfg = _get_cfg()
    if cfg is None:
        logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name)
        return _format_error("上下文查询未启用：本次运行未注入 context_query 配置")
    node_id = (args.get("node_id") or "").strip()
    field = args.get("field")
    iter_seq = args.get("iter_seq")
    logger.info(
        "[CtxQuery] %s 被调用: node_id=%s, field=%s, iter_seq=%s, run_id=%s",
        tool_name, node_id, field, iter_seq, cfg.get("run_id"),
    )
    if not node_id:
        _log_handler_return(tool_name, "missing_node_id", "缺参")
        return _format_error("查询失败：node_id 是必填参数")
    payload = {"runId": cfg["run_id"], "nodeId": node_id}
    if field:
        payload["field"] = field
    if iter_seq is not None:
        payload["iterSeq"] = int(iter_seq)
    status, data = _http_post(
        cfg["base_url"], cfg["token"], _OUTPUT_ENDPOINT, payload,
    )
    if status == 401:
        _log_handler_return(tool_name, "token_invalid", "401")
        return _format_error("token 无效或已过期，无法查询节点输出")
    if status == 403:
        _log_handler_return(tool_name, "forbidden", "403")
        return _format_error("runId 与 token 不匹配，禁止查询")
    if status == 404:
        _log_handler_return(tool_name, "not_found:" + node_id, "404")
        return _format_error(
            "节点 {} 尚未被访问（runId={}）".format(node_id, cfg["run_id"]),
        )
    if status != 200:
        _log_handler_return(tool_name, "error:" + str(status), str(status))
        return _format_error(
            "查询节点 {} 输出失败：{}".format(
                node_id, data.get("message", "未知错误"),
            ),
        )
    iter_str = data.get("iterSeq", 1)
    status_code = data.get("status", "未知")
    field_hint = "字段 {}".format(field) if field else "整包输出"
    summary = "节点 {}（第 {} 次访问）的 {}，状态={}".format(
        node_id, iter_str, field_hint, status_code,
    )
    if data.get("truncated"):
        summary += "（结果已截断）"
    summary += "。"
    result = _format_success(summary, data)
    _log_handler_return(tool_name, result, "node={},status={}".format(node_id, status_code))
    return result


def _handle_get_node_meta(args, **kw):
    tool_name = "get_node_meta"
    cfg = _get_cfg()
    if cfg is None:
        logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name)
        return _format_error("上下文查询未启用：本次运行未注入 context_query 配置")
    node_id = (args.get("node_id") or "").strip()
    visit_uuid = (args.get("visit_uuid") or "").strip()
    kind = (args.get("kind") or "ALL").strip().upper()
    iter_seq = args.get("iter_seq")
    logger.info(
        "[CtxQuery] %s 被调用: node_id=%s, visit_uuid=%s, kind=%s, iter_seq=%s, run_id=%s",
        tool_name, node_id or "<none>", _short_visit(visit_uuid) if visit_uuid else "<none>",
        kind, iter_seq, cfg.get("run_id"),
    )
    if not node_id and not visit_uuid:
        _log_handler_return(tool_name, "missing_target", "缺参")
        return _format_error("查询失败：node_id 和 visit_uuid 至少需要提供一个")
    payload = {"runId": cfg["run_id"], "kind": kind}
    if node_id:
        payload["nodeId"] = node_id
    if visit_uuid:
        payload["visitUuid"] = visit_uuid
    if iter_seq is not None:
        payload["iterSeq"] = int(iter_seq)
    status, data = _http_post(
        cfg["base_url"], cfg["token"], _META_ENDPOINT, payload,
    )
    if status == 401:
        _log_handler_return(tool_name, "token_invalid", "401")
        return _format_error("token 无效或已过期，无法查询中间数据")
    if status == 403:
        _log_handler_return(tool_name, "forbidden", "403")
        return _format_error("runId 与 token 不匹配，禁止查询")
    if status == 404:
        target = _short_visit(visit_uuid) if visit_uuid else node_id
        _log_handler_return(tool_name, "not_found:" + target, "404")
        return _format_error(
            "未找到 {} 的访问记录（runId={}）".format(target, cfg["run_id"]),
        )
    if status != 200:
        _log_handler_return(tool_name, "error:" + str(status), str(status))
        return _format_error(
            "查询中间数据失败：{}".format(data.get("message", "未知错误")),
        )
    target_id = data.get("nodeId") or node_id or _short_visit(visit_uuid)
    iter_str = data.get("iterSeq", 1)
    summary = "节点 {}（第 {} 次访问）的 {} 数据".format(target_id, iter_str, kind)
    if data.get("truncated"):
        summary += "（结果已截断）"
    summary += "。"
    result = _format_success(summary, data)
    _log_handler_return(tool_name, result, "target={},kind={}".format(target_id, kind))
    return result


# ---------------------------------------------------------------------------
# Tool schemas (OpenAI Function Calling 格式，与 Java systemPrompt 描述对齐)
# ---------------------------------------------------------------------------

_GET_PREDECESSOR_PATHS_SCHEMA = {
    "name": "get_predecessor_paths",
    "description": (
        "查询当前节点的所有前驱访问路径（含 visit 链、回边标识）。"
        "无需参数，自动使用当前请求的工作流运行上下文。"
        "适用场景：需要了解「当前节点运行前发生了什么」时调用。"
    ),
    "parameters": {
        "type": "object",
        "properties": {},
        "required": [],
    },
}

_GET_NODE_OUTPUT_SCHEMA = {
    "name": "get_node_output",
    "description": (
        "查询某个前驱节点的输出（envelope.data 整包或单字段）。"
        "适用场景：需要精确引用某个前驱节点的输出变量时调用。"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "node_id": {
                "type": "string",
                "description": "目标节点 ID（必填）",
            },
            "field": {
                "type": "string",
                "description": "可选；指定后只返回 envelope.data 内该字段的值",
            },
            "iter_seq": {
                "type": "integer",
                "description": "可选；同节点多次访问时指定第几次（1-based），不填取最后一次",
            },
        },
        "required": ["node_id"],
    },
}

_GET_NODE_META_SCHEMA = {
    "name": "get_node_meta",
    "description": (
        "查询某个前驱节点的中间数据（思考过程 / 工具调用 / 待办清单 / 全部）。"
        "适用场景：需要理解前驱节点的思考过程或工具调用结果时调用。"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "node_id": {
                "type": "string",
                "description": "目标节点 ID（与 visit_uuid 二选一）",
            },
            "visit_uuid": {
                "type": "string",
                "description": "指定具体访问的 UUID（与 node_id 二选一，优先使用）",
            },
            "kind": {
                "type": "string",
                "enum": ["THINKING", "TOOL_CALLS", "TODO", "ALL"],
                "description": "返回内容类型，默认 ALL",
            },
            "iter_seq": {
                "type": "integer",
                "description": "可选；同节点多次访问时指定第几次，不填取最后一次",
            },
        },
        "required": [],
    },
}


# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------

def _check_context_active(*args, **kwargs) -> bool:
    """check_fn：仅当本次请求注入了 context_query 时工具对 agent 可见。"""
    return _get_cfg() is not None


def register_internal_context_tools():
    """一次性注册 3 个工具到全局 registry。

    幂等：重复调用不会报错（相同 name + toolset 会覆盖）。
    在 hermes_bridge 模块加载时调用一次即可。
    """
    try:
        from tools.registry import registry
    except ImportError:
        logger.warning("无法 import tools.registry，跳过工作流上下文工具注册")
        return

    try:
        registry.register(
            name="get_predecessor_paths",
            toolset=_TOOLSET,
            schema=_GET_PREDECESSOR_PATHS_SCHEMA,
            handler=_handle_get_predecessor_paths,
            check_fn=_check_context_active,
            emoji="🧭",
        )
        registry.register(
            name="get_node_output",
            toolset=_TOOLSET,
            schema=_GET_NODE_OUTPUT_SCHEMA,
            handler=_handle_get_node_output,
            check_fn=_check_context_active,
            emoji="📤",
        )
        registry.register(
            name="get_node_meta",
            toolset=_TOOLSET,
            schema=_GET_NODE_META_SCHEMA,
            handler=_handle_get_node_meta,
            check_fn=_check_context_active,
            emoji="🔍",
        )
        logger.info("已注册工作流上下文查询工具（toolset=%s）", _TOOLSET)
    except Exception as e:
        logger.warning("注册工作流上下文工具失败: %s", e)