agent.py

"""OpenTag's general-purpose knowledge-work Deep Agent."""

import logging
import os
from pathlib import Path
from uuid import uuid4

from copilotkit import CopilotKitMiddleware
from deepagents import (
    GeneralPurposeSubagentProfile,
    HarnessProfile,
    create_deep_agent,
    register_harness_profile,
)
from deepagents.backends import StateBackend
from dotenv import load_dotenv
from langchain.agents.middleware import AgentMiddleware
from langchain_core.messages import ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import GraphRecursionError

from coding.config import (
    coding_enabled,
    github_providers,
    log_configuration_warnings,
)
from coding.subagent import build_coder_subagent
from ag_ui_langgraph import CustomEventNames
from langchain_core.callbacks.manager import adispatch_custom_event
from langchain_core.runnables.config import ensure_config
from composio_tools.config import DEFAULT_WORKSPACE_USER_ID
from composio_tools.runtime import composio_runtime
from composio_tools.state import ComposioAgentState
from composio_tools.tools import build_composio_tools
from internal_sources import internal_source_toolsets
from prompts import (
    BASE_SYSTEM_PROMPT,
    DEFAULT_AGENT_DISPLAY_NAME,
    NO_WEB_SEARCH_TOOL_ADDENDUM,
    WEB_SEARCH_TOOL_ADDENDUM,
    CODING_OFF_ADDENDUM,
    CODING_ON_ADDENDUM,
    current_date_prompt,
    build_base_system_prompt,
    composio_addendum,
)
from tools import web_search

logger = logging.getLogger(__name__)

load_dotenv(Path(__file__).resolve().parent.parent / ".env")


class LogToolCalls(AgentMiddleware):
    """Print each tool start/fail so a Slack turn can be diagnosed from logs."""

    def _name(self, request) -> str:
        tool = getattr(request, "tool", None)
        return getattr(tool, "name", None) or getattr(request, "name", "?")

    def _tool_call_id(self, request) -> str:
        tool_call = getattr(request, "tool_call", None) or {}
        if isinstance(tool_call, dict):
            return str(tool_call.get("id") or "unknown")
        return str(getattr(tool_call, "id", None) or "unknown")

    def _recursion_error_message(self, request, error: GraphRecursionError) -> ToolMessage:
        return ToolMessage(
            content=(
                "Coder stopped: recursion limit reached. "
                "The sandbox looped without finishing. "
                f"({error})"
            ),
            tool_call_id=self._tool_call_id(request),
            status="error",
        )

    def wrap_tool_call(self, request, handler):
        name = self._name(request)
        print(f"[TOOL] start {name}")
        try:
            result = handler(request)
        except GraphRecursionError as error:
            print(f"[TOOL] fail {name}: {type(error).__name__}: {error}")
            return self._recursion_error_message(request, error)
        except Exception as error:
            print(f"[TOOL] fail {name}: {type(error).__name__}: {error}")
            raise
        print(f"[TOOL] done {name}")
        return result

    async def awrap_tool_call(self, request, handler):
        name = self._name(request)
        print(f"[TOOL] start {name}")
        if name == "task":
            try:
                # The adapter turns this event into a complete text message.
                # It requires message_id/message; a role/content payload makes
                # the event consumer fail after dispatch has already returned.
                await adispatch_custom_event(
                    CustomEventNames.ManuallyEmitMessage.value,
                    {
                        "message_id": str(uuid4()),
                        "message": (
                            "Starting the coder in a Daytona sandbox. "
                            "This can take a few minutes."
                        ),
                    },
                    config=ensure_config(),
                )
            except Exception:
                # A note that did not arrive must not take the coder run with
                # it. Logged rather than swallowed: silence here is what let the
                # broken dispatch above go unnoticed.
                logger.warning(
                    "[TOOL] could not tell the thread the coder was starting",
                    exc_info=True,
                )
        try:
            result = await handler(request)
        except GraphRecursionError as error:
            print(f"[TOOL] fail {name}: {type(error).__name__}: {error}")
            return self._recursion_error_message(request, error)
        except Exception as error:
            print(f"[TOOL] fail {name}: {type(error).__name__}: {error}")
            raise
        print(f"[TOOL] done {name}")
        return result

VALID_REASONING_EFFORTS = frozenset(
    {"none", "minimal", "low", "medium", "high", "xhigh", "max"}
)
VALID_VERBOSITY_LEVELS = frozenset({"low", "medium", "high"})

# Deep Agents adds shell execution and a general-purpose delegation tool by
# default. execute is not globally banned: the coder subagent needs it. The
# main agent allowlists filesystem tools without execute via
# FilesystemMiddleware (StateBackend has no sandbox). The general-purpose
# subagent stays off so routine turns do not pay extra delegation latency.
register_harness_profile(
    "openai",
    HarnessProfile(
        excluded_tools=frozenset(),  # execute restricted via FilesystemMiddleware
        general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False),
    ),
)


def _validated_openai_setting(
    name: str,
    *,
    default: str,
    allowed: frozenset[str],
) -> str:
    """Read and validate a non-secret OpenAI tuning setting."""
    value = os.environ.get(name, default).strip().lower()
    if value not in allowed:
        choices = ", ".join(sorted(allowed))
        raise RuntimeError(f"Invalid {name}: expected one of {choices}")
    return value


def graph_recursion_limit(coding_on: bool | None = None) -> int:
    """Steps the main graph may take in one Slack turn."""
    return 80 if (coding_enabled() if coding_on is None else coding_on) else 25


def build_agent():
    """Build the OpenTag knowledge-work graph."""
    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        raise RuntimeError("Missing OPENAI_API_KEY environment variable")

    reasoning_effort = _validated_openai_setting(
        "OPENAI_REASONING_EFFORT",
        default="low",
        allowed=VALID_REASONING_EFFORTS,
    )
    verbosity = _validated_openai_setting(
        "OPENAI_VERBOSITY",
        default="low",
        allowed=VALID_VERBOSITY_LEVELS,
    )
    has_web_search = bool(os.environ.get("TAVILY_API_KEY"))
    model_name = os.environ.get("OPENAI_MODEL", "gpt-5.5")
    llm = ChatOpenAI(
        model=model_name,
        api_key=api_key,
        reasoning_effort=reasoning_effort,
        verbosity=verbosity,
        use_responses_api=True,
    )

    providers = github_providers()
    log_configuration_warnings(providers)
    coding_on = coding_enabled(selection=providers)
    source_toolsets = internal_source_toolsets(providers.search)
    internal_tools = [
        tool for tools in source_toolsets.values() for tool in tools
    ]
    # The same runtime the connect route uses, built once per process. Two
    # session caches would mean two sessions per identity, and one process
    # holding one session is the reason this moved into the agent at all.
    # `DEFAULT_WORKSPACE_USER_ID`, not a second spelling of it. The runtime
    # caches on this id, and `main.py` passes the constant for the connect
    # route: two spellings would build two caches, so the account an operator
    # connected through the route is not the one a turn runs in.
    composio = composio_runtime(
        default_user_id=os.environ.get(
            "INTELLIGENCE_CHANNEL_NAME", DEFAULT_WORKSPACE_USER_ID
        ),
    )
    composio_tools: list = (
        []
        if composio is None
        else build_composio_tools(composio.config, composio.cache, composio.effects)
    )

    main_tools = (
        [web_search, *internal_tools, *composio_tools]
        if has_web_search
        else [*internal_tools, *composio_tools]
    )

    agent_display_name = (
        os.environ.get("AGENT_DISPLAY_NAME", DEFAULT_AGENT_DISPLAY_NAME).strip()
        or DEFAULT_AGENT_DISPLAY_NAME
    )
    # Only claim the internal-source tools that were actually registered. The
    # prompt used to describe Notion, Linear and GitHub tools unconditionally,
    # so an agent holding only the Composio search believed it already had them
    # and answered without looking.
    system_prompt = build_base_system_prompt(
        agent_display_name,
        internal_sources=tuple(name for name, tools in source_toolsets.items() if tools),
    ) + (
        WEB_SEARCH_TOOL_ADDENDUM
        if has_web_search
        else NO_WEB_SEARCH_TOOL_ADDENDUM
    )
    system_prompt = system_prompt + (
        CODING_ON_ADDENDUM if coding_on else CODING_OFF_ADDENDUM
    )
    # Which apps exist is known here and was never passed on, so the model
    # answered questions about its own reach by guessing. It names apps only;
    # `search_my_tools` still owns which actions each one has.
    system_prompt = system_prompt + composio_addendum(
        composio.config if composio is not None else None
    )

    checkpointer = MemorySaver()
    create_kwargs = {
        "model": llm,
        "system_prompt": system_prompt,
        "tools": main_tools,
        "middleware": [
            CopilotKitMiddleware(),
            current_date_prompt,
            LogToolCalls(),
        ],
        # StateBackend has no sandbox, so the built-in FilesystemMiddleware
        # filters execute. Do not pass a second FilesystemMiddleware:
        # create_agent rejects duplicate middleware names.
        "backend": StateBackend(),
        "checkpointer": checkpointer,
        # Declared whether or not Composio is configured. The Channel forwards
        # the actor on every run and the AG-UI adapter drops a forwarded key the
        # state schema does not name, so leaving it out would make "who spoke"
        # depend on an unrelated feature flag.
        "state_schema": ComposioAgentState,
    }
    if coding_on:
        assert providers.coding is not None
        create_kwargs["subagents"] = [
            build_coder_subagent(
                model=llm,
                checkpointer=checkpointer,
                provider=providers.coding,
                github_tools=source_toolsets.get("github", []),
            )
        ]

    agent_graph = create_deep_agent(**create_kwargs)

    print(
        "[AGENT] OpenTag Agent created "
        f"with model={model_name}, reasoning={reasoning_effort}, verbosity={verbosity}"
    )
    print(f"[AGENT] web search: {'enabled' if has_web_search else 'disabled'}")
    print(f"[AGENT] coding: {'enabled' if coding_on else 'disabled'}")
    print(f"[AGENT] internal-source tools: {len(internal_tools)}")
    print(
        "[AGENT] composio: "
        + (
            "disabled"
            if composio is None
            else "shared="
            + (",".join(composio.config.workspace_toolkits) or "none")
            + " personal="
            + (",".join(composio.config.user_toolkits) or "none")
            + f" approvals={composio.config.approvals}"
        )
    )
    print(f"[AGENT] Main tools: {[t.name for t in main_tools]}")

    # A coding turn uses many GitHub MCP reads before task(). 25 steps is
    # enough for chat and too low for "read this PR, then code".
    # graph.with_config is for direct invoke. Slack/AG-UI must also get
    # this value on LangGraphAGUIAgent(config=...) in main.py.
    recursion_limit = graph_recursion_limit(coding_on)
    print(f"[AGENT] recursion_limit: {recursion_limit}")
    return agent_graph.with_config({"recursion_limit": recursion_limit})