script-build-and-framework-tmp/agent/agent/tools/builtin/context.py

"""
上下文工具 - 获取当前执行上下文

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

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

from agent.orchestration.models import CompletionPolicy
from agent.tools import tool, ToolResult, ToolContext


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

    Returns:
        ToolResult: 包含当前模式的计划、焦点提醒和协作者状态
    """
    runner = context.get("runner")
    goal_tree = context.get("goal_tree")
    trace_id = context.get("trace_id")
    policy = CompletionPolicy(
        context.get("completion_policy", CompletionPolicy.LEGACY_AUTO.value)
    )

    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)

    task_context = ""
    if policy == CompletionPolicy.EXPLICIT_VALIDATION:
        if hasattr(runner, "_build_task_context"):
            task_context = await runner._build_task_context(trace)
        else:
            coordinator = context.get("coordinator")
            root_trace_id = context.get("root_trace_id")
            if coordinator and root_trace_id:
                task_context = await coordinator.task_context(root_trace_id)

    # 构建上下文内容（复用 runner 的周期注入格式）
    if hasattr(runner, '_build_context_injection'):
        context_content = runner._build_context_injection(
            trace,
            goal_tree,
            task_context=task_context,
        )
    else:
        if task_context:
            context_content = task_context
        elif goal_tree and goal_tree.goals:
            context_content = f"## Current Plan\n\n{goal_tree.to_prompt()}"
        else:
            context_content = "暂无计划信息"

        # Fallback 也检查 IM 通知
        im_config = trace.context.get("im_config") if trace else None
        if im_config:
            contact_id = im_config.get("contact_id")
            chat_id = im_config.get("chat_id")
            if contact_id and chat_id:
                try:
                    from agent.tools.builtin.im import chat as im_chat
                    notification = im_chat._notifications.get((contact_id, chat_id))
                    if notification:
                        count = notification.get("count", 0)
                        senders = notification.get("from", [])
                        senders_str = ", ".join(senders)
                        context_content += (
                            f"\n\n## IM 消息通知\n\n"
                            f"你有 {count} 条新消息，来自: {senders_str}\n"
                            f"使用 `im_receive_messages(contact_id=\"{contact_id}\", chat_id=\"{chat_id}\")` 查看消息内容。"
                        )
                    else:
                        context_content += "\n\n## IM 消息通知\n\n暂无新消息"
                except (ImportError, AttributeError):
                    pass

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

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