Agent/cyber_agent/tools/builtin/context.py

"""
上下文工具 - 刷新当前执行状态并读取授权协议快照

提供 get_current_context 工具，让 Agent 可以主动获取：
- 当前计划（GoalTree）
- 焦点提醒
- 协作者状态

框架也会在特定轮次自动调用此工具进行周期性上下文刷新。
"""

import json

from cyber_agent.tools import tool, ToolResult, ToolContext
from cyber_agent.core.agent_mode import policy_from_context
from cyber_agent.core.context_policy import (
    ContextPolicyError,
    get_authorized_context_snapshot,
    require_matching_root_task_anchor,
)


@tool(
    description="获取当前执行上下文，包括计划状态、焦点提醒、协作者信息等。当你感到困惑或需要回顾当前任务状态时调用。",
    hidden_params=["context"],
    groups=["core"],
)
async def get_current_context(
    context: ToolContext,
) -> ToolResult:
    """
    获取当前执行上下文

    Returns:
        ToolResult: 包含 GoalTree、焦点提醒、协作者状态等信息
    """
    runner = context.get("runner")
    goal_tree = context.get("goal_tree")
    trace_id = context.get("trace_id")

    if not runner:
        return ToolResult(
            title="❌ 无法获取上下文",
            output="Runner 未初始化",
            error="Runner not available"
        )

    # 获取 trace 对象
    trace = None
    if runner.trace_store and trace_id:
        trace = await runner.trace_store.get_trace(trace_id)

    # Runner 是工具的必需上下文，统一由它渲染 Goal、通知和 Recursive 摘要。
    context_content = runner._build_context_injection(trace, goal_tree)

    if not context_content:
        context_content = "当前无需要刷新的上下文信息"

    return ToolResult(
        title="📋 当前执行上下文",
        output=context_content,
        long_term_memory="已刷新执行上下文",
    )


@tool(
    description="读取当前 Recursive Trace 已获授权的一份不可变协议上下文引用。",
    hidden_params=["context"],
    groups=["core"],
)
async def read_context_ref(
    ref_id: str,
    version: str,
    context: ToolContext,
) -> ToolResult:
    """读取当前 Trace本地持有的精确 ContextRef版本。

    Agent按需调用；工具重新加载持久化 Trace并校验模式、所有者、树归属和内容哈希，
    不接受任意 Trace ID，也不遍历父级、祖先或兄弟。
    """
    store = context.get("store")
    trace_id = context.get("trace_id")
    if not store or not trace_id:
        return ToolResult(
            title="上下文引用读取失败",
            output="",
            error="store and trace_id are required",
        )
    trace = await store.get_trace(trace_id)
    if not trace:
        return ToolResult(
            title="上下文引用读取失败",
            output="",
            error=f"Trace not found: {trace_id}",
        )
    policy = policy_from_context(trace.context)
    if (
        not policy.requires_task_protocol
        or trace.agent_type == "validator"
        or trace.context.get("created_by_tool") == "validator"
    ):
        return ToolResult(
            title="上下文引用读取失败",
            output="",
            error="read_context_ref requires a writable Recursive business Agent",
        )
    root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
    root = trace if root_trace_id == trace.trace_id else await store.get_trace(
        root_trace_id
    )
    if not root:
        return ToolResult(
            title="上下文引用读取失败",
            output="",
            error=f"Recursive root Trace not found: {root_trace_id}",
        )
    try:
        require_matching_root_task_anchor(root.context, trace.context)
        snapshot = get_authorized_context_snapshot(
            trace.context,
            ref_id=ref_id,
            version=version,
            root_trace_id=root_trace_id,
            uid=trace.uid,
        )
    except ContextPolicyError as exc:
        return ToolResult(
            title="上下文引用读取失败",
            output="",
            error=str(exc),
        )
    payload = {
        "ref_id": snapshot.ref_id,
        "version": snapshot.version,
        "kind": snapshot.kind,
        "summary": snapshot.summary,
        "source_trace_id": snapshot.source_trace_id,
        "content": snapshot.content,
    }
    return ToolResult(
        title="授权上下文引用",
        output=json.dumps(
            payload,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ),
        long_term_memory=(
            f"已读取 {snapshot.kind} 引用 {snapshot.ref_id}@{snapshot.version[:12]}"
        ),
    )