chatgpt-on-wechat/bridge/agent_bridge.py

"""
Agent Bridge - Integrates Agent system with existing COW bridge
"""

import os
import re
import threading
import uuid
from typing import Dict, Iterator, Optional, List, Tuple

from agent.protocol import (
    Agent,
    LLMModel,
    LLMRequest,
    get_cancel_registry,
    get_steer_registry,
)
from agent.protocol.step_writer import StepWriter
from bridge.agent_event_handler import AgentEventHandler
from bridge.agent_initializer import AgentInitializer
from bridge.bridge import Bridge
from bridge.context import Context
from bridge.reply import Reply, ReplyType
from common import const
from common.log import logger
from common.utils import expand_path
from config import conf
from models.openai_compatible_bot import OpenAICompatibleBot


def add_openai_compatible_support(bot_instance):
    """
    Dynamically add OpenAI-compatible tool calling support to a bot instance.
    
    This allows any bot to gain tool calling capability without modifying its code,
    as long as it uses OpenAI-compatible API format.
    
    Note: Some bots like ZHIPUAIBot have native tool calling support and don't need enhancement.
    """
    if hasattr(bot_instance, 'call_with_tools'):
        # Bot already has tool calling support (e.g., ZHIPUAIBot)
        logger.debug(f"[AgentBridge] {type(bot_instance).__name__} already has native tool calling support")
        return bot_instance

    # Create a temporary mixin class that combines the bot with OpenAI compatibility
    class EnhancedBot(bot_instance.__class__, OpenAICompatibleBot):
        """Dynamically enhanced bot with OpenAI-compatible tool calling"""

        def get_api_config(self):
            """
            Infer API config from common configuration patterns.
            Most OpenAI-compatible bots use similar configuration.
            """
            from config import conf

            return {
                'api_key': conf().get("open_ai_api_key"),
                'api_base': conf().get("open_ai_api_base"),
                'model': conf().get("model") or const.DEFAULT_MODEL,
                'default_temperature': conf().get("temperature", 0.9),
                'default_top_p': conf().get("top_p", 1.0),
                'default_frequency_penalty': conf().get("frequency_penalty", 0.0),
                'default_presence_penalty': conf().get("presence_penalty", 0.0),
            }

    # Change the bot's class to the enhanced version
    bot_instance.__class__ = EnhancedBot
    logger.info(
        f"[AgentBridge] Enhanced {bot_instance.__class__.__bases__[0].__name__} with OpenAI-compatible tool calling")

    return bot_instance


class AgentLLMModel(LLMModel):
    """
    LLM Model adapter that uses COW's existing bot infrastructure
    """

    _MODEL_BOT_TYPE_MAP = {
        "wenxin": const.BAIDU, "wenxin-4": const.BAIDU,
        "xunfei": const.XUNFEI, const.QWEN: const.QWEN_DASHSCOPE,
        const.QIANFAN: const.QIANFAN,
        const.MODELSCOPE: const.MODELSCOPE,
    }
    _MODEL_PREFIX_MAP = [
        ("qwen", const.QWEN_DASHSCOPE), ("qwq", const.QWEN_DASHSCOPE), ("qvq", const.QWEN_DASHSCOPE),
        ("gemini", const.GEMINI), ("glm", const.ZHIPU_AI), ("claude", const.CLAUDEAPI),
        ("moonshot", const.MOONSHOT), ("kimi", const.MOONSHOT),
        ("doubao", const.DOUBAO), ("deepseek", const.DEEPSEEK),
        ("ernie", const.QIANFAN),
        ("mimo-", const.MIMO),
    ]

    # Which model answers, most specific first. All default to None, meaning
    # "follow the next one down" and ultimately the global config. Declared on
    # the class so that resolving a model never depends on __init__ having run:
    # these are read on every call, including from doubles built with __new__.
    #
    # Per-conversation, set from agent.workspace.session_prefs when the user
    # picks a model for one conversation.
    _session_model = None
    _session_provider = None
    # Per Agent, from its profile. Under the conversation's choice but above
    # the global one: an Agent picked for its judgement should not answer on
    # whatever the console was last set to. The default Agent never has one —
    # it *is* the global choice.
    _agent_model = None
    _agent_provider = None
    # Fallback routing, engaged by use_fallback() after the primary model
    # failed a turn for good. Declared here (not in __init__) for the same
    # reason as the fields above: `model` is read on every call, including on
    # instances built with __new__.
    #
    # The fallback is a chain: `_fallback_depth` is the index of the link the
    # run currently sits on, so a run that has burned through link 0 and is on
    # link 1 can still advance to link 2 when that one fails too.
    _fallback_model = None
    _fallback_provider = None
    _fallback_depth = 0

    def __init__(self, bridge: Bridge, bot_type: str = "chat"):
        super().__init__(model=conf().get("model") or const.DEFAULT_MODEL)
        self.bridge = bridge
        self.bot_type = bot_type
        self._bot = None
        self._bot_model = None

    @property
    def model(self):
        # A fallback that has been engaged outranks every normal choice: once
        # the primary model has failed the turn, the whole point is to answer
        # on something else.
        if self._fallback_model:
            return self._fallback_model
        return (
            self._session_model
            or self._agent_model
            or conf().get("model")
            or const.DEFAULT_MODEL
        )

    @model.setter
    def model(self, value):
        pass

    def fallback_config(self) -> dict:
        """Return the configured fallback chain, normalized.

        A non-dict or disabled entry yields an empty chain so callers can treat
        "not usable" as a single check. Links missing a provider or a model are
        dropped — half a link could route the turn nowhere — as are duplicates
        of an earlier link, which would only re-probe a model the run has
        already proven is down. So is a link naming the primary's own route,
        matched on provider *and* model: the same model name behind another
        provider — a second account or endpoint of it — is a legitimate
        backup, not a duplicate.

        The chain is unbounded: however many links the user configured is how
        many switches a turn gets. There is no separate cap.
        """
        raw = conf().get("chat_fallback")
        empty = {"chain": []}
        if not isinstance(raw, dict) or not raw.get("enabled"):
            return empty
        raw_chain = raw.get("chain")
        if not isinstance(raw_chain, list):
            # Pre-chain shape ({provider, model}): config._migrate_chat_fallback
            # normally upgrades it at load time, so reaching here means a
            # caller handed us the raw dict. Honor it rather than dropping the
            # user's backup model.
            raw_chain = [raw]
        primary = (self._session_model or self._agent_model
                   or conf().get("model") or const.DEFAULT_MODEL)
        chain = []
        # Seed the dedup with the primary's own route: its provider is
        # resolved first (the global model config carries none), so a link
        # is matched on the (provider, model) pair — not the model name
        # alone, which would drop same-model links behind other providers.
        seen = {(self._primary_provider_id(primary), (primary or "").strip())}
        for item in raw_chain:
            if not isinstance(item, dict):
                continue
            provider = (item.get("provider") or "").strip()
            model = (item.get("model") or "").strip()
            if not provider or not model:
                continue
            key = (provider, model)
            if key in seen:
                continue
            seen.add(key)
            chain.append({"provider": provider, "model": model})
        return {"chain": chain}

    def _primary_provider_id(self, primary_model: str) -> str:
        """The provider the primary model routes through, as a provider id.

        Resolved in the id space the fallback chain's links are written in
        ("openai", "custom:<id>", ...), so the chain's dedup seed compares a
        link against the route the primary actually takes rather than against
        a provider-less model name. Follows the routing precedence — a session
        or Agent pin, then use_linkai, the configured bot_type, and finally
        the model-name inference _primary_bot_type applies when nothing is
        configured ("chatGPT", the persisted spelling of "openai", maps back
        onto it).
        """
        if self._session_provider:
            return self._session_provider
        if self._agent_provider:
            return self._agent_provider
        bot_type = self._primary_bot_type(primary_model)
        return "openai" if bot_type == const.CHATGPT else bot_type

    # How many times one turn may walk the whole chain before the failure is
    # reported. Two passes rather than one because a pass takes real time: by
    # the time the walk comes back around to a link that was rate limited, the
    # window may well have cleared. A third pass would mostly re-probe an
    # outage that is not going to clear inside a single turn.
    _FALLBACK_MAX_PASSES = 2

    def fallback_available(self) -> bool:
        """Whether this run can still advance along the fallback chain.

        True while link attempts remain inside the pass budget. Unlike the old
        single-model fallback, sitting on a link is not the end: a backup that
        fails for good earns the next one, which is the point of having a
        chain — and reaching the last link wraps back to the first instead of
        ending the turn.
        """
        chain = self.fallback_config()["chain"]
        return self._fallback_depth < len(chain) * self._FALLBACK_MAX_PASSES

    def use_fallback(self) -> bool:
        """Advance the rest of this run onto the next fallback model.

        Returns True when the switch happened. Called after the *current*
        model has failed a turn for good (retries exhausted), never mid-retry —
        for the primary that is the end of its own retries, and for a fallback
        link the single attempt it is granted.

        The switch is sticky in the sense that the run stays on whichever link
        answered, so a sustained outage isn't re-probed once per step; but a
        link that fails advances to the next one rather than giving up.
        reset_fallback() returns the run to the primary at the start of the
        next one.
        """
        if not self.fallback_available():
            return False
        chain = self.fallback_config()["chain"]
        # Wrap around: `_fallback_depth` counts attempts, not links, so the
        # second pass re-tries link 0. A turn that reaches the end of the chain
        # has not run out of options — the walk took long enough that a rate
        # limit hit on the first pass may have cleared by now.
        link = chain[self._fallback_depth % len(chain)]
        self._fallback_provider = link["provider"]
        self._fallback_model = link["model"]
        self._fallback_depth += 1
        # Drop the cached primary bot; `bot` rebuilds it for the new routing.
        self._bot = None
        self._bot_model = None
        self._bot_type = None
        total = len(chain) * self._FALLBACK_MAX_PASSES
        logger.warning(
            "[AgentLLMModel] current model failed; falling back to "
            f"{link['provider']}/{link['model']} "
            f"(link {self._fallback_depth}/{total})"
        )
        return True

    def reset_fallback(self) -> None:
        """Return to the primary model — call once at the start of a run.

        A new user message always starts fresh on the primary; within a run the
        fallback stays engaged on whichever link answered (see use_fallback).
        Rewinding the chain index here — not mid-run — is what lets the *next*
        run walk the chain again from the front.
        """
        if self._fallback_model is None:
            return
        self._fallback_model = None
        self._fallback_provider = None
        # Back to the front of the chain: the next run starts on the primary
        # model, and if it fails again it is a new failure that earns a new
        # walk through the links.
        self._fallback_depth = 0
        self._bot = None
        self._bot_model = None
        self._bot_type = None

    def set_agent_default(self, provider: Optional[str], model: Optional[str]) -> None:
        """Pin the Agent's own model, under any per-conversation choice."""
        provider = (provider or "").strip() or None
        model = (model or "").strip() or None
        if provider == self._agent_provider and model == self._agent_model:
            return
        self._agent_provider = provider
        self._agent_model = model
        self._bot = None
        self._bot_model = None
        self._bot_type = None

    def set_session_override(self, provider: Optional[str], model: Optional[str]) -> None:
        """Pin this session to one model/provider, or clear it with None/None.

        The provider matters as much as the model: without it a session that
        switches from DeepSeek to Claude would keep routing through the globally
        configured bot type and ask DeepSeek for a Claude model.
        """
        provider = (provider or "").strip() or None
        model = (model or "").strip() or None
        if provider == self._session_provider and model == self._session_model:
            return
        self._session_provider = provider
        self._session_model = model
        # Force the lazy bot to be rebuilt for the new routing on the next call.
        self._bot = None
        self._bot_model = None
        self._bot_type = None

    def catalog_model_meta(self) -> dict:
        """Model-catalog metadata for the effective provider+model, or {}.

        The provider mirrors ``_resolve_bot_type`` precedence (session
        override, then use_linkai, then the configured bot type), mapped back
        onto the UI provider ids the catalog is keyed by."""
        from models import model_catalog
        if self._session_provider:
            provider = self._session_provider
        elif conf().get("use_linkai", False) and conf().get("linkai_api_key"):
            provider = "linkai"
        else:
            bot_type = conf().get("bot_type") or ""
            provider = "openai" if bot_type == const.CHATGPT else bot_type
        return model_catalog.resolve_model_meta(provider, self.model)

    @staticmethod
    def provider_to_bot_type(provider_id: str) -> str:
        """Map a UI provider id onto a bot type, as the models console does."""
        if not provider_id:
            return ""
        # Same mapping the models console persists: "openai" routes through the
        # OpenAI-compatible bot, not the legacy completion one.
        if provider_id == "openai":
            return const.CHATGPT
        return provider_id

    def _resolve_bot_type(self, model_name: str) -> str:
        """Resolve bot type from model name, matching Bridge.__init__ logic."""
        # An engaged fallback outranks every normal choice: the whole point of
        # the fallback is to leave whichever provider just failed, and the
        # model being requested (`self.model`) is already the fallback's own.
        if self._fallback_provider:
            return self.provider_to_bot_type(self._fallback_provider)
        return self._primary_bot_type(model_name)

    def _primary_bot_type(self, model_name: str) -> str:
        """Bot type the primary model routes through, ignoring any engaged fallback.

        The tail _resolve_bot_type delegates to; split out so the fallback
        chain can seed its dedup with the primary's route without an engaged
        fallback's provider leaking in.
        """
        # A session override wins over every global routing switch, including
        # use_linkai: the user picked this provider for this conversation.
        if self._session_provider:
            return self.provider_to_bot_type(self._session_provider)
        if self._agent_provider:
            return self.provider_to_bot_type(self._agent_provider)

        if conf().get("use_linkai", False) and conf().get("linkai_api_key"):
            return const.LINKAI
        # Support custom bot type configuration
        configured_bot_type = conf().get("bot_type")
        if configured_bot_type:
            return configured_bot_type

        if not model_name or not isinstance(model_name, str):
            return const.OPENAI
        if model_name in self._MODEL_BOT_TYPE_MAP:
            return self._MODEL_BOT_TYPE_MAP[model_name]
        if model_name.lower().startswith("minimax") or model_name in ["abab6.5-chat"]:
            return const.MiniMax
        if model_name in [const.QWEN_TURBO, const.QWEN_PLUS, const.QWEN_MAX]:
            return const.QWEN_DASHSCOPE
        if model_name in [const.MOONSHOT, "moonshot-v1-8k", "moonshot-v1-32k", "moonshot-v1-128k"]:
            return const.MOONSHOT
        if conf().get("bot_type") == "modelscope":
            return const.MODELSCOPE
        lowered_model = model_name.lower()
        for prefix, btype in self._MODEL_PREFIX_MAP:
            if lowered_model.startswith(prefix):
                return btype
        return const.OPENAI

    def _normalized_reasoning_effort(self):
        """Return the active model's effort value after config resolution."""
        from models.reasoning_capabilities import resolve_reasoning_effort

        return resolve_reasoning_effort(
            self._resolve_bot_type(self.model),
            self.model,
            conf().get("reasoning_effort_by_model", {}),
            conf().get("reasoning_effort", "high"),
        )

    def _is_thinking_only_model(self) -> bool:
        """Return True for models that require reasoning to stay enabled."""
        from models.reasoning_capabilities import get_reasoning_capability

        capability = get_reasoning_capability(
            self._resolve_bot_type(self.model),
            self.model,
        )
        return bool(capability.get("thinking_only"))

    @property
    def bot(self):
        """Lazy load the bot, re-create when model or bot_type changes"""
        from models.bot_factory import create_bot
        cur_model = self.model
        cur_bot_type = self._resolve_bot_type(cur_model)
        if self._bot is None or self._bot_model != cur_model or getattr(self, '_bot_type', None) != cur_bot_type:
            # Hand the resolved type to create_bot as the credential provider
            # too. cur_bot_type already encodes the engaged fallback / session
            # override, and the bot must resolve api_key+api_base from *that*
            # provider — reading the global bot_type instead would pair this
            # link's model id with the primary provider's endpoint.
            self._bot = create_bot(cur_bot_type, credential_bot_type=cur_bot_type)
            self._bot = add_openai_compatible_support(self._bot)
            self._bot_model = cur_model
            self._bot_type = cur_bot_type
        return self._bot

    def call(self, request: LLMRequest):
        """
        Call the model using COW's bot infrastructure
        """
        try:
            # For non-streaming calls, we'll use the existing reply method
            # This is a simplified implementation
            if hasattr(self.bot, 'call_with_tools'):
                # Use tool-enabled call if available
                kwargs = {
                    'messages': request.messages,
                    'tools': getattr(request, 'tools', None),
                    'stream': False,
                    'model': self.model  # Pass model parameter
                }
                # Only pass max_tokens if it's explicitly set
                if request.max_tokens is not None:
                    kwargs['max_tokens'] = request.max_tokens

                # Extract system prompt if present
                system_prompt = getattr(request, 'system', None)
                if system_prompt:
                    kwargs['system'] = system_prompt

                # Pass context metadata to bot
                channel_type = getattr(self, 'channel_type', None) or ''
                if channel_type:
                    kwargs['channel_type'] = channel_type
                session_id = getattr(self, 'session_id', None)
                if session_id:
                    kwargs['session_id'] = session_id
                # Only bots that declare it get the agent id: others may pass
                # unknown kwargs straight through to their provider.
                agent_id = getattr(self, 'agent_id', None)
                if agent_id and getattr(self.bot, 'accepts_agent_id', False):
                    kwargs['agent_id'] = agent_id

                # Thinking mode is a global toggle independent of the channel.
                # IM channels (WeChat/WeCom/DingTalk/Feishu) won't render the
                # reasoning trace, but still benefit from the higher answer
                # quality the thinking pass produces.
                from config import conf
                thinking_enabled = bool(conf().get("enable_thinking", False))
                # Some native-reasoning models reject disabled thinking or use
                # effort as their only control, so they cannot follow the UI
                # toggle literally.
                if self._is_thinking_only_model():
                    thinking_enabled = True
                kwargs['thinking'] = (
                    {"type": "enabled"} if thinking_enabled
                    else {"type": "disabled"}
                )
                # Effort only shapes a thinking pass. Thinking-only models keep
                # receiving it because they force thinking_enabled above.
                if thinking_enabled:
                    effort = self._normalized_reasoning_effort()
                    if effort:
                        kwargs['reasoning_effort'] = effort

                response = self.bot.call_with_tools(**kwargs)
                return self._format_response(response)
            else:
                # Fallback to regular call
                # This would need to be implemented based on your specific needs
                raise NotImplementedError("Regular call not implemented yet")
                
        except Exception as e:
            logger.error(f"AgentLLMModel call error: {e}")
            raise
    
    def call_stream(self, request: LLMRequest):
        """
        Call the model with streaming using COW's bot infrastructure
        """
        try:
            if hasattr(self.bot, 'call_with_tools'):
                # Use tool-enabled streaming call if available
                # Extract system prompt if present
                system_prompt = getattr(request, 'system', None)

                # Build kwargs for call_with_tools
                kwargs = {
                    'messages': request.messages,
                    'tools': getattr(request, 'tools', None),
                    'stream': True,
                    'model': self.model  # Pass model parameter
                }

                # Only pass max_tokens if explicitly set, let the bot use its default
                if request.max_tokens is not None:
                    kwargs['max_tokens'] = request.max_tokens

                # Add system prompt if present
                if system_prompt:
                    kwargs['system'] = system_prompt

                # Pass context metadata to bot
                channel_type = getattr(self, 'channel_type', None) or ''
                if channel_type:
                    kwargs['channel_type'] = channel_type
                session_id = getattr(self, 'session_id', None)
                if session_id:
                    kwargs['session_id'] = session_id
                # Only bots that declare it get the agent id: others may pass
                # unknown kwargs straight through to their provider.
                agent_id = getattr(self, 'agent_id', None)
                if agent_id and getattr(self.bot, 'accepts_agent_id', False):
                    kwargs['agent_id'] = agent_id

                # Thinking mode is a global toggle independent of the channel.
                # IM channels (WeChat/WeCom/DingTalk/Feishu) won't render the
                # reasoning trace, but still benefit from the higher answer
                # quality the thinking pass produces.
                from config import conf
                thinking_enabled = bool(conf().get("enable_thinking", False))
                # Keep streaming and non-streaming calls on the same provider
                # contract for always-thinking models.
                if self._is_thinking_only_model():
                    thinking_enabled = True
                kwargs['thinking'] = (
                    {"type": "enabled"} if thinking_enabled
                    else {"type": "disabled"}
                )
                # Effort only shapes a thinking pass. Thinking-only models keep
                # receiving it because they force thinking_enabled above.
                if thinking_enabled:
                    effort = self._normalized_reasoning_effort()
                    if effort:
                        kwargs['reasoning_effort'] = effort

                stream = self.bot.call_with_tools(**kwargs)
                
                # Convert stream format to our expected format
                for chunk in stream:
                    yield self._format_stream_chunk(chunk)
            else:
                bot_type = type(self.bot).__name__
                raise NotImplementedError(f"Bot {bot_type} does not support call_with_tools. Please add the method.")
                
        except Exception as e:
            logger.error(f"AgentLLMModel call_stream error: {e}", exc_info=True)
            raise
    
    def _format_response(self, response):
        """Format Claude response to our expected format"""
        # This would need to be implemented based on Claude's response format
        return response
    
    def _format_stream_chunk(self, chunk):
        """Format Claude stream chunk to our expected format"""
        # This would need to be implemented based on Claude's stream format
        return chunk


class AgentBridge:
    """
    Bridge class that integrates super Agent with COW
    Manages multiple agent instances per session for conversation isolation
    """
    
    def __init__(self, bridge: Bridge):
        self.bridge = bridge
        from agent.registry import get_agent_registry
        from agent.routing import get_agent_router

        self.agent_registry = get_agent_registry()
        self.agent_router = get_agent_router(self.agent_registry)
        # Canonical runtime map. A session identifier is only unique inside an
        # agent workspace, so the agent id is part of every live key.
        self._agent_instances: Dict[Tuple[str, str], Agent] = {}
        self._default_agents: Dict[str, Agent] = {}
        self._agents_lock = threading.RLock()
        # Backward-compatible view for integrations that inspect sessions of
        # the configured default agent directly.
        self.agents: Dict[str, Agent] = {}
        self.default_agent = None
        self.agent: Optional[Agent] = None
        self.scheduler_initialized = False
        self.scheduler_agent_ids = set()
        
        # Create helper instances
        self.initializer = AgentInitializer(bridge, self)

        # Eager-start the scheduler so cron tasks fire without waiting
        # for the first user message. init_scheduler is idempotent.
        try:
            from agent.tools.scheduler.integration import init_scheduler
            for profile in self.agent_registry.list(include_disabled=False):
                if init_scheduler(self, profile.workspace, profile.id):
                    self.scheduler_agent_ids.add(profile.id)
            self.scheduler_initialized = bool(self.scheduler_agent_ids)
        except Exception as e:
            logger.warning(f"[AgentBridge] Eager scheduler init failed: {e}")

        # Start the self-evolution idle trigger (idempotent, daemon thread).
        try:
            from agent.evolution.trigger import start_evolution_trigger
            start_evolution_trigger(self)
        except Exception as e:
            logger.warning(f"[AgentBridge] Evolution trigger init failed: {e}")

    def create_agent(self, system_prompt: str, tools: List = None, **kwargs) -> Agent:
        """
        Create the super agent with COW integration
        
        Args:
            system_prompt: System prompt
            tools: List of tools (optional)
            **kwargs: Additional agent parameters
            
        Returns:
            Agent instance
        """
        # Create LLM model that uses COW's bot infrastructure
        model = AgentLLMModel(self.bridge)
        
        # Default tools if none provided
        if tools is None:
            # Use ToolManager to load all available tools
            from agent.tools import ToolManager
            tool_manager = ToolManager()
            tool_manager.load_tools()
            
            tools = []
            workspace_dir = kwargs.get("workspace_dir")
            for tool_name in tool_manager.tool_classes.keys():
                try:
                    tool = tool_manager.create_tool(tool_name)
                    if tool:
                        if workspace_dir:
                            tool.cwd = workspace_dir
                        tools.append(tool)
                except Exception as e:
                    logger.warning(f"[AgentBridge] Failed to load tool {tool_name}: {e}")
        
        # Create agent instance
        agent = Agent(
            system_prompt=system_prompt,
            description=kwargs.get("description", "AI Super Agent"),
            model=model,
            tools=tools,
            max_steps=kwargs.get("max_steps", 100),
            output_mode=kwargs.get("output_mode", "logger"),
            workspace_dir=kwargs.get("workspace_dir"),
            skill_manager=kwargs.get("skill_manager"),
            enable_skills=kwargs.get("enable_skills", True),
            memory_manager=kwargs.get("memory_manager"),
            max_context_tokens=kwargs.get("max_context_tokens"),
            context_reserve_tokens=kwargs.get("context_reserve_tokens"),
            runtime_info=kwargs.get("runtime_info"),
        )

        # Log skill loading details
        if agent.skill_manager:
            logger.debug(f"[AgentBridge] SkillManager initialized with {len(agent.skill_manager.skills)} skills")

        return agent
    
    def steer_session(self, session_id: str, instruction: str, agent_id: str = None):
        """Inject an explicit instruction into one active session."""
        logger.info(f"[AgentBridge] steer new instruction: session={session_id}, content={instruction}")
        return get_steer_registry().submit(
            self.scoped_session_key(session_id, agent_id), instruction
        )

    def _resolve_agent_id(self, agent_id: str = None) -> str:
        return self.agent_registry.get(agent_id).id

    def scoped_session_key(self, session_id: str, agent_id: str = None) -> str:
        """Namespace a session id by its agent.

        Session ids are only unique within one Agent, so cancel and steer
        lookups must be scoped or a /cancel in one Agent's chat would abort a
        different Agent's run that happens to share the id.
        """
        return self._cancel_key(
            self._resolve_agent_id(agent_id),
            session_id,
            self.agent_registry.default_agent_id,
        )

    def route_context(self, context: Context) -> str:
        """Resolve and attach the workspace selected for an inbound context."""
        return self.agent_router.resolve_context(context)

    def get_conversation_store(self, agent_id: str = None):
        """Return the session store owned by one agent workspace."""
        profile = self.agent_registry.get(agent_id)
        from agent.memory import get_conversation_store
        return get_conversation_store(profile.workspace)

    def _seed_team_members(self, session_id: str, host_agent_id: str, context: Context = None) -> None:
        """Project a team bot's roster onto the session.

        A channel instance configured with ``members`` is a fixed team: its
        owner (``host_agent_id``) plus teammates it may delegate to. The rest of
        the stack learns a conversation is a team from
        ``session_prefs.members``, so the instance roster is mirrored there.

        Two sources, two policies:

        - **Channel instance** (message carries an ``instance_id``): the instance
          roster is *authoritative* and is reconciled onto the session every
          time — including shrinking it, or clearing it when the instance was
          switched back to a single Agent. Without this, a session seeded once
          when the instance was a team keeps injecting the team prompt forever
          even after the roster is emptied in the console.

        - **Delegation** (a delegated turn runs in its own private session and
          carries ``delegation_members``): seed-once, never overwrite, so a
          teammate can delegate onward to the same team.

        Only enabled teammates other than the owner are kept, matching how a Web
        team is stored.
        """
        if not session_id or not context:
            return
        # A roster may name teammates that live in another process. Learn how to
        # reach them before resolving the roster below, so a member reached
        # through the transport is kept rather than dropped as unknown. Empty or
        # no transport (a stand-alone install) makes this a no-op.
        peers = context.get("peers")
        if peers is None:
            peers = context.kwargs.get("peers")
        if peers:
            try:
                from agent.multiagent import get_transport

                transport = get_transport()
                if transport is not None:
                    transport.register_peers(peers)
            except Exception as e:
                logger.debug(f"[AgentBridge] register_peers failed: {e}")
        # The channel path carries the roster under ``members`` and is
        # authoritative (it mirrors the instance's live team.json roster). A
        # delegated turn instead carries ``delegation_members`` and is seed-once.
        channel_members = context.get("members")
        if channel_members is None:
            channel_members = context.kwargs.get("members")
        from_channel = channel_members is not None
        delegation_members = context.get("delegation_members") or context.kwargs.get("delegation_members")

        try:
            from agent.workspace import session_prefs

            existing = session_prefs.get_prefs(session_id, host_agent_id).get("members")

            if from_channel:
                # Authoritative reconcile against the instance's current roster,
                # even when it is now empty (single-Agent instance).
                cleaned = self._clean_team_members(channel_members or [], host_agent_id)
                if list(existing or []) == cleaned:
                    return  # already in sync — nothing to write
                if cleaned:
                    session_prefs.set_prefs(session_id, host_agent_id, members=cleaned)
                    logger.info(
                        f"[AgentBridge] Reconciled team roster {cleaned} onto session "
                        f"'{session_id}' owned by {host_agent_id}"
                    )
                elif existing:
                    # Instance is no longer a team: drop the stale session roster
                    # so the team prompt stops being injected.
                    session_prefs.set_prefs(session_id, host_agent_id, members=None)
                    logger.info(
                        f"[AgentBridge] Cleared stale team roster from session "
                        f"'{session_id}' owned by {host_agent_id} (instance is single-Agent)"
                    )
                self._retire_session_runtimes(session_id)
                return

            # Delegation path: seed once, never clobber an existing roster.
            if existing:
                return
            cleaned = self._clean_team_members(delegation_members or [], host_agent_id)
            if cleaned:
                session_prefs.set_prefs(session_id, host_agent_id, members=cleaned)
                logger.info(
                    f"[AgentBridge] Seeded team roster {cleaned} onto session "
                    f"'{session_id}' owned by {host_agent_id}"
                )
                self._retire_session_runtimes(session_id)
        except Exception as e:
            logger.debug(f"[AgentBridge] _seed_team_members failed: {e}")

    def _retire_session_runtimes(self, session_id: str) -> None:
        """Drop the runtimes of *session_id* so the next turn rebuilds them.

        A runtime fixes its tool list when it is built, and delegation is only
        offered to a conversation that has teammates. A roster that changes
        while the conversation is already live therefore has to retire what was
        built under the old one: otherwise the team gains a member and the tool
        to reach them only appears after a restart. Retiring costs a rebuild on
        the next turn; the transcript is reloaded from the store either way.
        """
        with self._agents_lock:
            retired = [key for key in self._agent_instances if key[1] == session_id]
            for key in retired:
                self._agent_instances.pop(key, None)
            if retired:
                self.agents.pop(session_id, None)
        if retired:
            logger.info(
                f"[AgentBridge] Retired {len(retired)} runtime(s) of session "
                f"'{session_id}' after its roster changed"
            )

    def _clean_team_members(self, members, host_agent_id: str) -> list:
        """Normalize a roster: drop the owner, blanks, dupes and unknown/disabled
        Agents, preserving order. Returns the teammates to store on a session.

        Members arrive as ids a caller may address an Agent by, so the reserved
        "default" alias is resolved here too; storing the resolved id keeps the
        roster comparable with the owner and with an id sent the other way.

        A member hosted in another process is kept under the id the installed
        transport knows it by; without a transport such ids are unknown here
        and dropped like any other stranger.
        """
        from agent.multiagent import peer as peer_of

        cleaned = []
        for mid in members or []:
            mid = str(mid or "").strip()
            if not mid:
                continue
            try:
                resolved = self.agent_registry.get_addressed(mid, require_enabled=True).id
            except Exception:
                found = peer_of(mid)
                if found is None:
                    continue  # skip unknown/disabled teammates
                resolved = found.id
            if resolved == host_agent_id or resolved in cleaned:
                continue
            cleaned.append(resolved)
        return cleaned

    def _resolve_speaker(self, host_agent_id: str, context: Context = None) -> str:
        """Pick who answers this turn: the conversation's owner, or a teammate
        the user addressed by name.

        Naming someone is an instruction about who should answer, so it is
        honoured literally. Anything unrecognised, disabled, or already the
        owner falls back to the owner, which is the single-Agent behaviour.
        """
        named = (context.get("speaker_agent_id") if context else "") or ""
        if not named or named == host_agent_id:
            return host_agent_id
        try:
            profile = self.agent_registry.get(named, require_enabled=True)
        except Exception:
            logger.warning(
                f"[AgentBridge] Ignoring unknown addressee '{named}', "
                f"answering as {host_agent_id}"
            )
            return host_agent_id
        logger.info(
            f"[AgentBridge] Turn addressed to {profile.id}; "
            f"answering in {host_agent_id}'s conversation"
        )
        return profile.id

    def _peer_speaker(self, named: str, host_agent_id: str):
        """The teammate a turn names when it is hosted in another process.

        None for an id that is empty, the owner's, a local Agent's, or that no
        transport knows — every one of which the ordinary path already handles.
        """
        named = str(named or "").strip()
        if not named or named == host_agent_id:
            return None
        try:
            self.agent_registry.get_addressed(named, require_enabled=False)
            return None  # local: answered here, as always
        except Exception:
            pass
        try:
            from agent.multiagent import get_transport, peer as peer_of

            if get_transport() is None:
                return None
            return peer_of(named)
        except Exception:
            return None

    def _speak_on_peer(self, query: str, session_id: str, host_agent_id: str, peer,
                       channel_type: str = "") -> Reply:
        """Let a teammate elsewhere answer this turn, as itself.

        The conversation stays the owner's — same session, same transcript — and
        only the voice changes, which is what addressing someone by name asks
        for. The teammate is given the conversation so far so it answers in
        context rather than cold.
        """
        from agent.multiagent import MODE_SPEAK, InvokeRequest, PeerAgent, get_transport, resolve_teammate

        owner = self.agent_registry.get(host_agent_id, require_enabled=False)

        members = [owner.id]
        for member_id in self._session_members(session_id, host_agent_id):
            if member_id not in (owner.id, peer.id) and member_id not in members:
                members.append(member_id)
        peers = []
        for member_id in members:
            found = resolve_teammate(member_id)
            profile = PeerAgent.from_any(found) if found else None
            if profile is not None:
                peers.append(profile)

        request = InvokeRequest(
            request_id=uuid.uuid4().hex,
            target_id=peer.id,
            task=self._strip_peer_address(query, peer),
            source_id=owner.id,
            source_name=owner.name,
            root_session_id=session_id,
            trace=(owner.id,),
            depth=0,
            members=tuple(members),
            peers=tuple(peers),
            timeout_seconds=self._delegation_timeout(),
            mode=MODE_SPEAK,
            history=tuple(self._shared_history(session_id, owner.id)),
        )
        logger.info(
            f"[AgentBridge] Turn addressed to peer {peer.id}; "
            f"answering in {owner.id}'s conversation, session={session_id}"
        )
        result = get_transport().invoke(request)
        if not result.ok:
            return Reply(ReplyType.ERROR, f"{peer.name} could not answer: {result.error}")

        content = result.content or ""
        # The turn happened in this conversation, so it belongs in its record;
        # nothing local ran to write it down.
        turn = [
            {"role": "user", "content": [{"type": "text", "text": query}]},
            {"role": "assistant", "content": [{"type": "text", "text": content}]},
        ]
        self._persist_messages(
            session_id, self._attribute_to_speaker(turn, peer.id), channel_type, owner.id
        )
        return Reply(ReplyType.TEXT, content)

    @staticmethod
    def _session_members(session_id: str, host_agent_id: str) -> list:
        """Teammate ids recorded on the session, as stored."""
        if not session_id:
            return []
        try:
            from agent.workspace import session_prefs

            return list(session_prefs.get_prefs(session_id, host_agent_id).get("members") or [])
        except Exception:
            return []

    @staticmethod
    def _strip_peer_address(query: str, peer) -> str:
        """Drop the leading "@name" aimed at *peer*; see ``_strip_address``."""
        from agent.team_addressing import strip_address

        return strip_address(query, [peer.name, peer.id])

    @staticmethod
    def _delegation_timeout() -> float:
        try:
            return float((conf().get("agent_delegation") or {}).get("timeout_seconds") or 600)
        except Exception:
            return 600.0

    def _shared_history(self, session_id: str, owner_agent_id: str) -> list:
        """Text-only history with authors, oldest first."""
        try:
            if not conf().get("conversation_persistence", True):
                return []
            max_turns = conf().get("agent_max_context_turns", 30)
            saved = self.get_conversation_store(owner_agent_id).load_messages(
                session_id, max_turns=max(3, max_turns // 2), with_authors=True
            )
        except Exception as e:
            logger.warning(f"[AgentBridge] shared history unavailable for {session_id}: {e}")
            return []
        history = []
        for message in AgentInitializer._filter_text_only_messages(saved or []):
            blocks = message.get("content") or []
            text = blocks[0].get("text", "") if blocks and isinstance(blocks[0], dict) else ""
            if not text:
                continue
            entry = {"role": message["role"], "text": text}
            if message["role"] == "assistant":
                entry["agent_id"] = message.get("agent_id") or owner_agent_id
            history.append(entry)
        return history

    def _strip_address(self, query: str, speaker_agent_id: str) -> str:
        """Drop the leading "@name" now that it has been acted on.

        Routing already answered the question the mention was asking, so
        leaving it in makes the Agent read its own name as someone else's and
        reply about that person instead of as itself. It only ever comes off
        the front, and only for the Agent it named.

        The transcript keeps the original: the mention is what the user wrote,
        and it records who the turn was aimed at.
        """
        if not query:
            return query
        try:
            profile = self.agent_registry.get(speaker_agent_id, require_enabled=False)
        except Exception:
            return query
        from agent.team_addressing import strip_address

        return strip_address(query, [profile.name, profile.id])

    @staticmethod
    def _attribute_to_speaker(messages: list, speaker_agent_id: str) -> list:
        """Tag a guest's turns with who wrote them, for the transcript.

        A shared conversation that records no author replays as one voice, so a
        reload loses track of who said what.

        Returns copies. The dicts handed in are the Agent's live context, and
        ``extras`` is a column of ours: annotating them in place would put an
        unknown key on every later request and the model rejects the call.
        """
        return [
            {
                **message,
                "extras": {
                    **(message.get("extras") or {}),
                    "agent_id": speaker_agent_id,
                },
            }
            for message in messages or []
        ]

    def _roster_speaker_labels(self) -> list:
        """Names and ids a model might copy from a shared transcript."""
        labels = []
        try:
            for profile in self.agent_registry.list(include_disabled=True):
                if profile.name:
                    labels.append(profile.name)
                if profile.id:
                    labels.append(profile.id)
        except Exception:
            return []
        return labels

    @staticmethod
    def _strip_speaker_prefix(text: str, labels: list) -> str:
        """Drop a leading speaker label the model copied from history.

        Covers the forms a shared transcript can show: ``[Name]``,
        ``Name：`` and ``Name(@id)：`` (with either colon).
        """
        if not text or not labels:
            return text
        escaped = "|".join(
            re.escape(label)
            for label in sorted({label for label in labels if label}, key=len, reverse=True)
        )
        if not escaped:
            return text
        name = rf"(?:{escaped})"
        return re.sub(
            rf"^\s*(?:\[{name}\]|{name}\s*(?:\(@{name}\))?\s*[:：])\s*",
            "",
            text,
            count=1,
        )

    def _strip_speaker_prefix_from_messages(self, messages: list) -> list:
        labels = self._roster_speaker_labels()
        if not labels:
            return messages
        cleaned = []
        for message in messages or []:
            if message.get("role") != "assistant":
                cleaned.append(message)
                continue
            content = message.get("content")
            if isinstance(content, list) and content and content[0].get("type") == "text":
                text = self._strip_speaker_prefix(content[0].get("text", ""), labels)
                cleaned.append({
                    **message,
                    "content": [{**content[0], "text": text}, *content[1:]],
                })
            elif isinstance(content, str):
                cleaned.append({
                    **message,
                    "content": self._strip_speaker_prefix(content, labels),
                })
            else:
                cleaned.append(message)
        return cleaned

    def _begin_run(self, session_id: str, agent_id: str, context: Context = None):
        """Open a run for this turn and make its id the ambient one.

        Returns ``(run_id, token, store)``; every element is None when the run
        could not be recorded. Bookkeeping must never break a reply, so any
        failure here degrades to "no run row" rather than raising: the turn
        still runs, it just is not addressable afterwards.

        A run id already in scope means this turn was started by another run
        (a delegation or a spawn), so that one becomes the parent and the tree
        stays walkable from either end. A caller that hands work to another
        thread cannot rely on that ambient id, since context variables do not
        cross threads, so it may instead name the run and its parent through
        the context and keep the tree intact.
        """
        from common.utils import current_agent_run_id, set_agent_run_id

        try:
            parent_run_id = str(
                (context.get("parent_run_id") if context else "")
                or current_agent_run_id()
                or ""
            )
            run_id = str((context.get("run_id") if context else "") or "") or uuid.uuid4().hex
            store = self.get_conversation_store(agent_id)
            # An external system driving this work passes its own handle
            # through the context; a native turn leaves both empty.
            task_id = str((context.get("task_id") if context else "") or "")
            task_source = str((context.get("task_source") if context else "") or "")
            store.create_run(
                run_id,
                agent_id=agent_id or "",
                session_id=session_id or "",
                parent_run_id=parent_run_id,
                task_id=task_id,
                task_source=task_source,
            )
            # Set last: once the ambient id changes, the caller owes us a reset.
            token = set_agent_run_id(run_id)
            return run_id, token, store
        except Exception as e:
            logger.warning(f"[AgentBridge] Could not open run: {e}")
            return None, None, None

    def _end_run(self, store, run_id: str, token, status: str, error: str = "") -> None:
        """Close a run and restore the previous ambient run id.

        The reset happens even when the status update fails, or the ambient id
        would leak into whatever this thread handles next.
        """
        from common.utils import clear_agent_run_id

        try:
            if store is not None and run_id:
                store.finish_run(run_id, status=status, error=error)
        except Exception as e:
            logger.warning(f"[AgentBridge] Could not close run {run_id}: {e}")
        finally:
            if token is not None:
                clear_agent_run_id(token)

    def peek_agent(self, session_id: str, agent_id: str = None) -> Optional[Agent]:
        """Return the session's live agent, or None if it has not been built.

        The read-only counterpart to `get_agent`, which initializes an agent on
        miss — spinning up MCP connections and skills. Callers that only want to
        inspect existing state (the context-usage endpoint hovers on this) must
        use this instead, and must tolerate None. Deliberately skips
        `_apply_session_project` / `apply_session_prefs`: both mutate the agent.

        :param session_id: Session identifier
        :param agent_id: Agent profile identifier. Omit for the configured default.
        :return: The existing Agent instance, or None.
        """
        if not session_id:
            return None
        resolved_agent_id = self._resolve_agent_id(agent_id)
        with self._agents_lock:
            return self._agent_instances.get(
                self._runtime_key(resolved_agent_id, session_id)
            )

    @staticmethod
    def _runtime_key(agent_id: str, session_id: str) -> Tuple[str, str]:
        return agent_id, session_id

    @staticmethod
    def _cancel_key(agent_id: str, token: str, default_agent_id: str) -> str:
        """Keep legacy token keys for the default agent, namespace the rest."""
        return token if agent_id == default_agent_id else f"{agent_id}::{token}"

    def _has_runtime(self, agent_id: str, session_id: str) -> bool:
        """Whether ``get_agent`` would return a cached runtime rather than build one."""
        if not session_id:
            return False
        with self._agents_lock:
            key = self._runtime_key(self._resolve_agent_id(agent_id), session_id)
            return key in self._agent_instances

    def get_agent(
        self,
        session_id: str = None,
        agent_id: str = None,
        host_agent_id: str = None,
        permission_mode: str = None,
    ) -> Optional[Agent]:
        """
        Get agent instance for the given session
        
        Args:
            session_id: Session identifier (e.g., user_id). If None, returns
                the workspace's default runtime instance.
            agent_id: Agent profile identifier. Omit for the configured default.
            host_agent_id: Agent that owns this conversation, when it is not
                ``agent_id``. Set when the user addressed a teammate directly:
                the teammate answers as itself, but reads and continues the
                host's transcript instead of starting a private one.
        
        Returns:
            Agent instance for this session
        """
        resolved_agent_id = self._resolve_agent_id(agent_id)
        with self._agents_lock:
            if session_id is None:
                agent = self._default_agents.get(resolved_agent_id)
                if agent is None:
                    agent = self.initializer.initialize_agent(
                        session_id=None, agent_id=resolved_agent_id
                    )
                    self._default_agents[resolved_agent_id] = agent
                if resolved_agent_id == self.agent_registry.default_agent_id:
                    self.default_agent = agent
                return agent

            host_id = self._resolve_agent_id(host_agent_id or resolved_agent_id)
            key = self._runtime_key(resolved_agent_id, session_id)
            agent = self._agent_instances.get(key)
            if agent is None:
                agent = self.initializer.initialize_agent(
                    session_id=session_id,
                    agent_id=resolved_agent_id,
                    host_agent_id=host_id,
                )
                self._agent_instances[key] = agent
                if resolved_agent_id == self.agent_registry.default_agent_id:
                    self.agents[session_id] = agent
            # Point the working directory at the session's project (if any).
            # Applied on every fetch, so switching projects — or back to the
            # default — takes effect on the next message without rebuilding the
            # agent. Memory/skills stay anchored to the workspace regardless.
            # Project and per-session settings belong to the conversation, so a
            # guest follows the host's, not its own unrelated ones.
            self._apply_session_project(agent, session_id, host_id)
            # Same idea for the session's permission mode, a per-conversation
            # override that falls back to the global config. The model is not
            # shared with a guest — see apply_session_prefs.
            self.apply_session_prefs(
                agent,
                session_id,
                host_id,
                owns_conversation=resolved_agent_id == host_id,
                permission_mode=permission_mode,
            )
            return agent

    def _sync_shared_transcript(self, agent, session_id: str, host_agent_id: str) -> None:
        """Reload the host's transcript so every teammate sees the same history.

        Solo conversations keep their live in-memory list (including tool
        chains). A team conversation is reread from the host store: this
        speaker's own turns keep their tool chains, and colleagues' replies are
        replayed as ``Name：`` user turns so ``assistant`` stays its own voice.
        """
        if not session_id or not AgentInitializer._is_shared_conversation(
            session_id, host_agent_id
        ):
            return
        restore = getattr(self.initializer, "_restore_conversation_history", None)
        if restore is None:
            return
        try:
            host = self.agent_registry.get(host_agent_id, require_enabled=False)
        except Exception:
            return
        restore(agent, session_id, host.workspace, host_agent_id)

    def _apply_session_project(self, agent, session_id: str, agent_id: str) -> None:
        """Retarget the agent's working directory to the session's project dir.

        A no-op when the session has no project selected (clears any previous
        override). Failures are swallowed: a bad project setting must not break
        the chat, it just falls back to the default workspace.
        """
        try:
            from agent.workspace import project_store
            project_dir = project_store.get_project_dir(session_id, agent_id)
            if getattr(agent, "apply_project_dir", None):
                agent.apply_project_dir(project_dir)
        except Exception as e:
            logger.debug(f"[AgentBridge] apply_session_project failed: {e}")

    def apply_session_prefs(
        self,
        agent,
        session_id: str,
        agent_id: str = None,
        owns_conversation: bool = True,
        permission_mode: str = None,
    ) -> None:
        """Apply a session's model / permission overrides to its agent.

        Called on every agent fetch and right after the user changes a setting,
        so a switch takes effect on the next message without rebuilding the
        agent. An empty override resets the agent to the global config, which is
        what makes "follow global" work after a session had pinned something.

        The model is the exception on two counts. A conversation's pinned model
        belongs to the Agent that owns it, so an invited teammate is never
        forced onto it — that would throw away the model it was configured with,
        often the reason it was invited. And in a group chat nobody follows the
        pin, the owner included: a team is a set of Agents each answering on its
        own model, so the pinned model (a single-chat notion) is ignored and
        every speaker uses its own default. Permission stays shared either way,
        because it bounds what this conversation may change regardless of who is
        speaking.
        """
        if agent is None or not session_id:
            return
        try:
            from agent.workspace import session_prefs

            prefs = session_prefs.get_prefs(session_id, agent_id)
            model = getattr(agent, "model", None)
            if model is not None and hasattr(model, "set_session_override"):
                # A conversation with members is a group: each Agent answers on
                # its own configured model, so the session pin never applies.
                is_group = bool(prefs.get("members"))
                if owns_conversation and not is_group:
                    model.set_session_override(prefs.get("provider"), prefs.get("model"))
                else:
                    model.set_session_override(None, None)
            if hasattr(agent, "apply_permission_mode"):
                # A delegated turn's session has no prefs; it carries the
                # delegating conversation's mode instead.
                agent.apply_permission_mode(permission_mode or prefs.get("permission"))
        except Exception as e:
            logger.debug(f"[AgentBridge] apply_session_prefs failed: {e}")

    def get_cached_agent(self, session_id: str, agent_id: str = None) -> Optional[Agent]:
        """Return an existing session agent without creating one."""
        resolved_agent_id = self._resolve_agent_id(agent_id)
        with self._agents_lock:
            return self._agent_instances.get(
                self._runtime_key(resolved_agent_id, session_id)
            )

    def iter_agent_instances(
        self, include_defaults: bool = True
    ) -> Iterator[Tuple[str, Optional[str], Agent]]:
        """Snapshot all live instances as ``(agent_id, session_id, agent)``."""
        with self._agents_lock:
            defaults = list(self._default_agents.items()) if include_defaults else []
            sessions = list(self._agent_instances.items())
        for agent_id, agent in defaults:
            yield agent_id, None, agent
        for (agent_id, session_id), agent in sessions:
            yield agent_id, session_id, agent

    def clear_all_model_fallbacks(self) -> int:
        """Drop any engaged fallback routing on every live agent's model.

        Fallback state lives on the long-lived ``AgentLLMModel`` and is normally
        cleared at the top of the next run. Disabling the fallback in the UI only
        rewrites config, so a model that had already switched would stay on the
        backup until that next run happens to reset it. Called when the user
        turns the fallback off so the change takes effect immediately, on the
        very next message, without waiting for a run boundary or a restart.

        Returns the number of models that were actually on a fallback.
        """
        cleared = 0
        for _agent_id, _session_id, agent in self.iter_agent_instances():
            model = getattr(agent, "model", None)
            reset = getattr(model, "reset_fallback", None)
            if not callable(reset):
                continue
            if getattr(model, "_fallback_model", None) is None:
                continue
            try:
                reset()
                cleared += 1
            except Exception as e:
                logger.debug(f"[AgentBridge] clear fallback skipped: {e}")
        if cleared:
            logger.info(f"[AgentBridge] cleared engaged fallback on {cleared} model(s)")
        return cleared

    def apply_runtime_limits(self) -> int:
        """Push the configured step cap and context budget onto live agents.

        Both are read once when an agent is built. Updating them in place keeps
        each session's working context and prompt cache; a smaller budget is
        enforced by the regular trim (with summary) at the start of the next
        run. Evicting the agents instead would rebuild them from the store's
        short, clipped restore window.

        Returns the number of agents updated.
        """
        from bridge.agent_initializer import configured_runtime_limits

        max_steps, max_context_tokens = configured_runtime_limits()
        updated = 0
        for _agent_id, _session_id, agent in self.iter_agent_instances():
            agent.max_steps = max_steps
            agent.max_context_tokens = max_context_tokens
            updated += 1
        logger.info(
            f"[AgentBridge] Applied runtime limits to {updated} agent(s): "
            f"max_steps={max_steps}, max_context_tokens={max_context_tokens}"
        )
        return updated

    def sync_session_messages_from_store(
        self, session_id: str, agent_id: str = None
    ) -> int:
        """Reload an agent's in-memory ``messages`` list from the persistent
        conversation store.

        Used after an external mutation (e.g. user edits / deletes a message
        via the web console) so the agent's next turn sees the same history
        as the database. The operation is a no-op when the agent has not been
        instantiated yet for the session.

        History is rebuilt exactly as on session restore. Deleting a message
        can orphan a tool_use from its tool_result; a turn left like that is
        replayed as text, since the provider would reject the broken pair.

        Returns:
            Number of messages now held in the agent's memory. Returns -1 if
            the agent does not exist or has no compatible ``messages`` attr.
        """
        if not session_id:
            return -1
        agent = self.get_cached_agent(session_id, agent_id=agent_id)
        if agent is None:
            return -1
        if not (hasattr(agent, "messages") and hasattr(agent, "messages_lock")):
            return -1
        try:
            store = self.get_conversation_store(agent_id)
            # No turn cap here: we want a faithful mirror of what the store
            # has for this session after deletion.
            remaining = store.load_messages(session_id, max_turns=10**6)
        except Exception as e:
            logger.warning(
                f"[AgentBridge] Failed to load messages for sync (session={session_id}): {e}"
            )
            return -1
        try:
            remaining = AgentInitializer._restored_history(remaining)
        except Exception as e:
            logger.warning(
                f"[AgentBridge] Replaying tool calls failed for session={session_id}, "
                f"syncing text only: {e}"
            )
            remaining = AgentInitializer._filter_text_only_messages(remaining)
        with agent.messages_lock:
            agent.messages.clear()
            for msg in remaining:
                agent.messages.append({
                    "role": msg["role"],
                    "content": msg["content"],
                })
            count = len(agent.messages)
        logger.info(
            f"[AgentBridge] Synced agent memory for session={session_id}, messages={count}"
        )
        return count

    def agent_reply(self, query: str, context: Context = None, 
                   on_event=None, clear_history: bool = False) -> Reply:
        """
        Use super agent to reply to a query
        
        Args:
            query: User query
            context: COW context (optional, contains session_id for user isolation)
            on_event: Event callback (optional)
            clear_history: Whether to clear conversation history
            
        Returns:
            Reply object
        """
        session_id = None
        agent = None
        request_id = None
        cancel_event = None
        token_key = None
        # Bound before the try so the error paths below can read it even when
        # the failure happened before it was computed.
        scoped_session_id = None
        steer_inbox = None
        run_id = None
        run_token = None
        run_store = None
        run_status = "done"
        run_error = ""
        try:
            # Extract session_id from context for user isolation
            if context:
                session_id = context.kwargs.get("session_id") or context.get("session_id")
                request_id = context.kwargs.get("request_id") or context.get("request_id")

            resolved_agent_id = (
                self.route_context(context)
                if context is not None
                else self.agent_registry.default_agent_id
            )

            # A team channel bot (e.g. a Feishu instance with members) carries
            # its roster on every inbound message. Materialize it into the
            # session the first time we see the conversation so the shared
            # delegate/@mention machinery — which reads session_prefs.members —
            # treats it as a team, exactly like a Web team conversation.
            self._seed_team_members(session_id, resolved_agent_id, context)

            # Addressing a teammate by name hands the turn to that teammate
            # directly. The conversation still belongs to `resolved_agent_id`,
            # so the transcript, the run and the queue all stay in one place —
            # only the voice answering this turn changes.
            #
            # A teammate on the roster may be hosted in another process; it
            # answers over the transport instead of here, and the turn is
            # recorded in this conversation either way.
            addressed = (context.get("speaker_agent_id") if context else "") or ""
            remote_speaker = self._peer_speaker(addressed, resolved_agent_id)
            if remote_speaker is not None:
                return self._speak_on_peer(
                    query, session_id, resolved_agent_id, remote_speaker,
                    channel_type=(context.get("channel_type") or "") if context else "",
                )
            speaker_agent_id = self._resolve_speaker(resolved_agent_id, context)
            # With multiple Agents (and especially several bound channel
            # instances) it isn't obvious from the logs which Agent a message
            # was routed to. Emit one line naming the target Agent and, when
            # present, the channel instance it arrived on. Skipped for
            # single-Agent setups to avoid noise.
            try:
                if len(self.agent_registry.list()) > 1:
                    instance_id = (
                        context.get("instance_id")
                        or context.kwargs.get("instance_id")
                        if context is not None else ""
                    )
                    via = f" | {instance_id}" if instance_id else ""
                    # The Agent that actually answers this turn is the speaker,
                    # which differs from the owner when the user addressed a
                    # teammate by name. Log the speaker so the line matches who
                    # replies; note the owner's conversation it runs in when
                    # they differ, so routing + addressing read consistently.
                    speaker = self.agent_registry.get(speaker_agent_id)
                    if speaker_agent_id != resolved_agent_id:
                        owner = self.agent_registry.get(resolved_agent_id)
                        logger.info(
                            f"[Routing] → 🤖 {speaker.name}({speaker.id}) "
                            f"in {owner.name}({owner.id})'s conversation{via}"
                        )
                    else:
                        logger.info(
                            f"[Routing] → 🤖 {speaker.name}({speaker.id}){via}"
                        )
            except Exception:
                pass
            # What the Agent is asked, once the addressing has been acted on.
            # Kept apart from `query`, which stays verbatim for the transcript.
            model_query = (
                self._strip_address(query, speaker_agent_id)
                if speaker_agent_id != resolved_agent_id
                else query
            )

            # Register a cancel token. Prefer per-turn request_id (web),
            # fall back to session_id (IM channels). The Event is polled by
            # AgentStreamExecutor at safe checkpoints.
            registry = get_cancel_registry()
            # Both the cancel token and the session grouping are namespaced:
            # session ids are only unique within one Agent, so two Agents
            # serving the same id must not cancel or steer each other. The
            # steer registry is keyed by the same scoped id.
            scoped_session_id = (
                self._cancel_key(
                    resolved_agent_id,
                    session_id,
                    self.agent_registry.default_agent_id,
                )
                if session_id
                else None
            )
            token_key = request_id or session_id
            if token_key:
                token_key = self._cancel_key(
                    resolved_agent_id,
                    token_key,
                    self.agent_registry.default_agent_id,
                )
                cancel_event = registry.register(
                    token_key, session_id=scoped_session_id
                )

            # Get agent for this session (will auto-initialize if needed)
            cached = self._has_runtime(speaker_agent_id, session_id)
            agent = self.get_agent(
                session_id=session_id,
                agent_id=speaker_agent_id,
                host_agent_id=resolved_agent_id,
                permission_mode=context.get("delegated_permission_mode"),
            )
            if not agent:
                return Reply(ReplyType.ERROR, "Failed to initialize super agent")

            # A team conversation is one transcript. Each Agent caches its own
            # in-memory list and only restores it on first init, so a teammate
            # that already joined would miss later turns spoken by someone else
            # (and the host would miss guest replies). Reload the shared
            # transcript with author labels before this turn is appended; a
            # runtime built for this turn has only just restored it.
            if cached:
                self._sync_shared_transcript(agent, session_id, resolved_agent_id)
            
            # Create event handler for logging and channel communication
            event_handler = AgentEventHandler(context=context, original_callback=on_event)
            
            # Filter tools based on context
            original_tools = agent.tools
            filtered_tools = original_tools
            
            # If this is a scheduled task execution, exclude scheduler tool to prevent recursion
            if context and context.get("is_scheduled_task"):
                filtered_tools = [tool for tool in agent.tools if tool.name != "scheduler"]
                agent.tools = filtered_tools
                logger.info(f"[AgentBridge] Scheduled task execution: excluded scheduler tool ({len(filtered_tools)}/{len(original_tools)} tools)")

            if context and agent.tools:
                for tool in agent.tools:
                    if tool.name == "scheduler" and not context.get("is_scheduled_task"):
                        try:
                            from agent.tools.scheduler.integration import attach_scheduler_to_tool
                            attach_scheduler_to_tool(tool, context)
                        except Exception as e:
                            logger.warning(f"[AgentBridge] Failed to attach context to scheduler: {e}")
                    elif tool.name == "agent_delegate":
                        try:
                            from agent.tools.agent_delegate.agent_delegate import attach_agent_delegate_to_tool
                            attach_agent_delegate_to_tool(tool, self, context)
                        except Exception as e:
                            logger.warning(f"[AgentBridge] Failed to attach delegation context: {e}")
            
            # Pass context metadata to model for downstream API requests
            if context and hasattr(agent, 'model'):
                agent.model.channel_type = context.get("channel_type", "")
                agent.model.session_id = session_id or ""
                agent.model.agent_id = speaker_agent_id

            # Store session_id on agent so executor can clear DB on fatal errors.
            # The conversation's owner is what identifies the transcript, so a
            # guest speaker still reads and writes the shared one.
            agent._current_session_id = session_id
            agent._current_agent_id = resolved_agent_id

            # Bound the in-memory context for scheduler sessions before each run.
            # Scheduler sessions are stable per-task and append every trigger,
            # so without trimming they would grow unbounded across runs and
            # blow up prompt cost. Regular user chats are not touched here —
            # the agent's own context manager handles that path.
            if session_id and session_id.startswith("scheduler_"):
                from config import conf
                scheduler_keep_turns = max(
                    1, int(conf().get("agent_max_context_turns", 30)) // 5
                )
                self._trim_in_memory_to_turns(agent, scheduler_keep_turns)

            # Open the run before anything is persisted, so both the user turn
            # and the reply are attributed to it and the work is addressable
            # while it is still in flight.
            run_id, run_token, run_store = self._begin_run(
                session_id, resolved_agent_id, context
            )

            # Eagerly persist the user message BEFORE running the agent so the
            # session and the user's bubble are immediately visible — even if
            # the user switches away or refreshes before the reply finishes.
            # The reply (assistant/tool messages) is appended step by step as
            # the run goes; later writes skip this already-stored user turn.
            pre_persisted = self._pre_persist_user_message(
                session_id, query, context, clear_history, resolved_agent_id
            )

            channel_type = (context.get("channel_type") or "") if context else ""

            def write_reply(messages: list):
                # Stamp every reply with its author, the owner's included. In a
                # shared conversation a guest reconstructs "who said what" from
                # this stamp; if the owner's turns went unstamped they would read
                # as unattributed, and a guest would mistake the owner's persona
                # ("I am Gray…") for its own and answer in that voice.
                messages = self._attribute_to_speaker(messages, speaker_agent_id)
                messages = self._strip_speaker_prefix_from_messages(messages)
                if messages:
                    persisted = self._persist_messages(
                        session_id,
                        list(messages),
                        channel_type,
                        resolved_agent_id,
                        create_if_missing=not pre_persisted,
                    )
                    if persisted is False:
                        raise RuntimeError("conversation write failed; step remains pending")

            writer = StepWriter(write_reply, skip_query=pre_persisted) if session_id else None

            def on_run_event(event):
                # Store the step before announcing it: a listener hearing
                # turn_end may rely on the step being in the transcript.
                if writer is not None and event.get("type") == "turn_end":
                    writer.step()
                event_handler.handle_event(event)

            # Mark this session as mid-run so the self-evolution idle scan does
            # not fire concurrently when a single turn runs longer than
            # idle_minutes.
            try:
                from agent.evolution.trigger import mark_run_active
                mark_run_active(agent, True)
            except Exception:
                pass

            try:
                if scoped_session_id:
                    steer_inbox = get_steer_registry().register(scoped_session_id)
                # Use agent's run_stream method with event handler
                response = agent.run_stream(
                    user_message=model_query,
                    on_event=on_run_event,
                    clear_history=clear_history,
                    cancel_event=cancel_event,
                    steer_inbox=steer_inbox,
                    # A scheduled task may legitimately have nothing to report
                    # (e.g. "notify me only if the price drops"). Nobody is
                    # waiting on this run, so an empty answer stays empty and
                    # the scheduler sends no message at all.
                    allow_empty_response=bool(context and context.get("is_scheduled_task")),
                    on_executor=writer.bind if writer is not None else None,
                )
            except Exception:
                # Keep the steps finished before the failure.
                if writer is not None:
                    writer.step()
                raise
            finally:
                # Clear the mid-run flag so idle scans can review this session.
                try:
                    from agent.evolution.trigger import mark_run_active
                    mark_run_active(agent, False)
                except Exception:
                    pass

                # Restore original tools
                if context and context.get("is_scheduled_task"):
                    agent.tools = original_tools

                # Log execution summary
                event_handler.log_summary()

                # Release cancel token; keep registry bounded.
                if token_key:
                    try:
                        registry.unregister(token_key)
                    except Exception:
                        pass
                if scoped_session_id and steer_inbox is not None:
                    get_steer_registry().unregister(scoped_session_id, steer_inbox)

            # A cancelled turn is not a failure, but it is not a completed run
            # either: the distinction is what tells a reader whether the result
            # is trustworthy or simply absent.
            if cancel_event is not None and cancel_event.is_set():
                run_status = "cancelled"

            # Persist what this run added beyond the steps already stored
            if writer is not None:
                new_messages = list(getattr(agent, '_last_run_new_messages', []))
                # The leading user turn was already persisted eagerly above;
                # drop it here so it isn't stored twice.
                if pre_persisted and new_messages and new_messages[0].get("role") == "user":
                    new_messages = new_messages[1:]
                writer.finish(new_messages)

            self.remember_context_usage(agent, session_id, resolved_agent_id)
            
            # Record this user turn for the self-evolution idle trigger. Skip
            # scheduler-injected / scheduled-task sessions so internal runs do
            # not count as user activity.
            if session_id and not session_id.startswith("scheduler_") and not (
                context and (
                    context.get("is_scheduled_task")
                    or context.get("is_delegated_task")
                )
            ):
                try:
                    from agent.evolution.trigger import note_user_turn
                    ch = (context.get("channel_type") or "") if context else ""
                    rcv = (context.get("receiver") or "") if context else ""
                    is_group = bool(context.get("isgroup")) if context else False
                    # Only enable proactive push for single chats (group push is
                    # noisy); group sessions still evolve, just without notify.
                    note_user_turn(agent, channel_type=ch, receiver=(rcv if not is_group else ""))
                except Exception:
                    pass

            # Post-message hot-reload: detect edits to ~/cow/mcp.json and
            # sync any new/removed MCP tools into the live agent in the
            # background. Off the critical path so user latency is unaffected;
            # changes take effect on the user's next message.
            self._schedule_mcp_hot_reload(agent)

            if isinstance(response, str):
                response = self._strip_speaker_prefix(
                    response, self._roster_speaker_labels()
                )

            # Check if there are files to send (from send/read tool)
            if hasattr(agent, 'stream_executor') and hasattr(agent.stream_executor, 'files_to_send'):
                files_to_send = agent.stream_executor.files_to_send
                if files_to_send:
                    logger.info(
                        f"[AgentBridge] Sending {len(files_to_send)} file(s), "
                        f"first={files_to_send[0].get('path')}"
                    )

                    # Clear files_to_send for next request
                    agent.stream_executor.files_to_send = []

                    # The reply pipeline carries one Reply, so the remaining
                    # files ride along and the channel sends them afterwards.
                    # Only the first carries the text, or it would repeat.
                    reply = self._create_file_reply(files_to_send[0], response, context)
                    extras = [
                        self._create_file_reply(f, "", context)
                        for f in files_to_send[1:]
                    ]
                    if extras:
                        reply.extra_replies = extras
                    return reply
            
            return Reply(ReplyType.TEXT, response)
            
        except Exception as e:
            logger.error(f"Agent reply error: {e}")
            run_status = "failed"
            run_error = str(e)
            # The in-memory context may have been reset to recover from a format
            # error or overflow, but the stored history is deliberately left
            # intact: it is irreplaceable and is never reloaded with tool blocks.
            # Release cancel token on error path too (idempotent).
            if cancel_event is not None and token_key:
                try:
                    get_cancel_registry().unregister(token_key)
                except Exception:
                    pass
            if scoped_session_id and steer_inbox is not None:
                try:
                    get_steer_registry().unregister(scoped_session_id, steer_inbox)
                except Exception:
                    pass
            return Reply(ReplyType.ERROR, f"Agent error: {str(e)}")

        finally:
            self._end_run(run_store, run_id, run_token, run_status, run_error)
    
    def _schedule_mcp_hot_reload(self, agent):
        """
        Fire-and-forget: detect mcp.json edits and reconcile the agent's
        tool dict in the background. Runs after the user's reply is sent,
        so any cost (file stat, hash, server boot) never adds to user latency.
        Failures are isolated and never raise into the message pipeline.
        """
        import threading
        from agent.tools import ToolManager
        from common.runtime_identity import wrap

        def _run():
            try:
                tm = ToolManager()
                tm.refresh_mcp_if_changed()
                added, removed = tm.sync_mcp_into_agent(agent)
                if added or removed:
                    logger.info(
                        f"[AgentBridge] Agent tools synced — "
                        f"added={added}, removed={removed}"
                    )
            except Exception as e:
                logger.warning(f"[AgentBridge] MCP hot-reload failed (non-fatal): {e}")

        # wrap carries the routed identity into the thread; without it this
        # would reload the default Agent's mcp.json and sync its tools into
        # whichever Agent actually served the message.
        threading.Thread(
            target=wrap(_run), daemon=True, name="mcp-hot-reload"
        ).start()

    def _create_file_reply(self, file_info: dict, text_response: str, context: Context = None) -> Reply:
        """
        Create a reply for sending files
        
        Args:
            file_info: File metadata from read tool
            text_response: Text response from agent
            context: Context object
            
        Returns:
            Reply object for file sending
        """
        file_type = file_info.get("file_type", "file")
        file_path = file_info.get("path")
        # Remote URLs are passed through as-is; local paths get a file:// prefix
        # so the channel can read them from disk.
        remote_url = file_info.get("url", "")
        is_remote = bool(remote_url) and remote_url.lower().startswith(("http://", "https://"))

        def _to_channel_url(p: str) -> str:
            if is_remote:
                return remote_url
            if p and p.lower().startswith(("http://", "https://")):
                return p
            return f"file://{p}"

        # For images, use IMAGE_URL type (channel will handle upload)
        if file_type == "image":
            file_url = _to_channel_url(file_path)
            logger.info(f"[AgentBridge] Sending image: {file_url}")
            reply = Reply(ReplyType.IMAGE_URL, file_url)
            # Attach text message if present (for channels that support text+image)
            if text_response:
                reply.text_content = text_response  # Store accompanying text
            return reply
        
        # For all file types (document, video, audio), use FILE type
        if file_type in ["document", "video", "audio"]:
            file_url = _to_channel_url(file_path)
            logger.info(f"[AgentBridge] Sending {file_type}: {file_url}")
            reply = Reply(ReplyType.FILE, file_url)
            reply.file_type = file_type
            reply.file_name = file_info.get("file_name", os.path.basename(file_path))
            # Attach text message if present
            if text_response:
                reply.text_content = text_response
            return reply
        
        # For all other file types (tar.gz, zip, etc.), also use FILE type
        file_url = _to_channel_url(file_path)
        logger.info(f"[AgentBridge] Sending generic file: {file_url}")
        reply = Reply(ReplyType.FILE, file_url)
        reply.file_type = file_type
        reply.file_name = file_info.get("file_name", os.path.basename(file_path))
        if text_response:
            reply.text_content = text_response
        return reply
    
    def _migrate_config_to_env(self, workspace_root: str):
        """
        Sync API keys from config.json to .env file.
        Adds new keys and updates changed values on each startup.

        Args:
            workspace_root: Workspace directory path (not used, kept for compatibility)
        """
        from config import conf
        import os
        
        key_mapping = {
            "open_ai_api_key": "OPENAI_API_KEY",
            "open_ai_api_base": "OPENAI_API_BASE",
            "gemini_api_key": "GEMINI_API_KEY",
            "claude_api_key": "CLAUDE_API_KEY",
            "linkai_api_key": "LINKAI_API_KEY",
        }
        
        env_file = expand_path("~/.cow/.env")
        
        # Read existing env vars (key -> value)
        existing_env_vars = {}
        if os.path.exists(env_file):
            try:
                with open(env_file, 'r', encoding='utf-8') as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith('#') and '=' in line:
                            key, val = line.split('=', 1)
                            existing_env_vars[key.strip()] = val.strip()
            except Exception as e:
                logger.warning(f"[AgentBridge] Failed to read .env file: {e}")
        
        # Sync config.json values into .env (add/update/remove)
        updated = False
        for config_key, env_key in key_mapping.items():
            raw = conf().get(config_key, "")
            value = raw.strip() if raw else ""
            old_value = existing_env_vars.get(env_key)

            if value:
                if old_value == value:
                    continue
                existing_env_vars[env_key] = value
                os.environ[env_key] = value
                updated = True
            else:
                if old_value is None:
                    continue
                existing_env_vars.pop(env_key, None)
                os.environ.pop(env_key, None)
                updated = True
            updated = True

        if updated:
            try:
                env_dir = os.path.dirname(env_file)
                os.makedirs(env_dir, exist_ok=True)

                with open(env_file, 'w', encoding='utf-8') as f:
                    f.write('# Environment variables for agent\n')
                    f.write('# Auto-managed - synced from config.json on startup\n\n')
                    for key, value in sorted(existing_env_vars.items()):
                        f.write(f'{key}={value}\n')

                logger.info("[AgentBridge] Synced API keys from config.json to .env")
            except Exception as e:
                logger.warning(f"[AgentBridge] Failed to sync API keys: {e}")
    
    def _pre_persist_user_message(
        self,
        session_id: str,
        query: str,
        context: Context,
        clear_history: bool,
        agent_id: str = None,
    ) -> bool:
        """Persist the user's message before the agent runs.

        This makes a brand-new session (and the user's bubble) visible even if
        the reply hasn't finished — switching away or refreshing no longer
        loses the in-flight session. Returns True when the user turn was
        stored, so the caller can skip it in the post-run persist.

        Best-effort: any failure is swallowed and reported as not-persisted.
        """
        if not session_id or not query:
            return False
        # Only real user turns: skip scheduler-injected / scheduled-task runs.
        if session_id.startswith("scheduler_") or (
            context and context.get("is_scheduled_task")
        ):
            return False
        try:
            from config import conf
            if not conf().get("conversation_persistence", True):
                return False
            store = self.get_conversation_store(agent_id)
            # clear_history starts a fresh transcript: wipe the store first so
            # the eager user turn becomes seq 0, matching in-memory state.
            if clear_history:
                store.clear_session(session_id)
            channel_type = (context.get("channel_type") or "") if context else ""
            user_msg = {
                "role": "user",
                "content": [{"type": "text", "text": query}],
            }
            display = context.get("display") if context else None
            if display:
                user_msg["extras"] = {"display": display}
            store.append_messages(session_id, [user_msg], channel_type=channel_type)
            return True
        except Exception as e:
            logger.warning(
                f"[AgentBridge] Failed to pre-persist user message for session={session_id}: {e}"
            )
            return False

    def _persist_messages(
        self,
        session_id: str,
        new_messages: list,
        channel_type: str = "",
        agent_id: str = None,
        create_if_missing: bool = True,
    ) -> bool:
        """
        Persist new messages to the conversation store after each agent run.

        Failures are logged but never propagate — they must not interrupt replies.

        ``create_if_missing=False`` is used once the user turn is known to be
        stored already: a missing session row then means the user deleted the
        session while the reply was still running, so the reply is dropped
        instead of resurrecting the session without its question.
        """
        if not new_messages:
            return True
        try:
            from config import conf
            if not conf().get("conversation_persistence", True):
                return True
            # When deep-thinking display is disabled, strip "thinking" content
            # blocks before persisting so they don't resurface on history reload.
            # The in-memory message list keeps them intact for this run's
            # multi-turn LLM context.
            thinking_enabled = bool(conf().get("enable_thinking", False))
            if not thinking_enabled:
                from models.reasoning_capabilities import get_reasoning_capability

                # Thinking-only models need their reasoning trace in stored
                # history so the next tool-calling turn can echo it back.
                capability = get_reasoning_capability(
                    AgentLLMModel(None)._resolve_bot_type(conf().get("model", "")),
                    conf().get("model", ""),
                )
                thinking_enabled = bool(capability.get("thinking_only"))
        except Exception:
            thinking_enabled = False

        messages_to_store = new_messages
        if not thinking_enabled:
            messages_to_store = self._strip_thinking_blocks(new_messages)

        stored = False
        try:
            store = self.get_conversation_store(agent_id)
            stored = store.append_messages(
                session_id, messages_to_store, channel_type=channel_type,
                create_if_missing=create_if_missing
            )
            if stored:
                from agent.protocol.artifact import index_message_artifacts
                index_message_artifacts(store, session_id, messages_to_store, agent_id=agent_id)
            elif not create_if_missing:
                logger.info(
                    f"[AgentBridge] Session {session_id} was deleted mid-run, "
                    f"dropped {len(messages_to_store)} reply message(s)"
                )
            # A deleted session is an intentional discard, not a retryable failure.
            return bool(stored) or not create_if_missing
        except Exception as e:
            logger.warning(
                f"[AgentBridge] Failed to persist messages for session={session_id}: {e}"
            )
            # Artifact indexing happens after the append committed; do not replay it.
            return bool(stored)

    def remember_context_usage(
        self, agent, session_id: str, agent_id: str = None
    ) -> None:
        """Store the context usage a run ended on, so the chart survives a restart.
        Best-effort: never fails the turn."""
        if not session_id:
            return
        try:
            store = self.get_conversation_store(agent_id)
            store.set_context_usage(session_id, agent.get_context_usage())
        except Exception as e:
            logger.debug(f"[AgentBridge] Context usage snapshot skipped: {e}")

    # Marker used to identify scheduler-injected user messages so we can apply
    # a sliding window without touching real user turns. The legacy prefix
    # "Scheduled task" (written by the v2 PR) is also recognised when pruning,
    # so old data can be aged out instead of leaking forever.
    _SCHEDULED_MARKER = "[SCHEDULED]"
    _SCHEDULED_LEGACY_MARKERS = ("Scheduled task",)

    def remember_scheduled_output(
        self,
        session_id: str,
        content: str,
        channel_type: str = "",
        task_description: str = "",
        agent_id: str = None,
        run_id: str = "",
    ) -> None:
        """Add the visible output of a scheduled task to the receiver's session.

        Scheduled task execution uses an isolated session so internal planning and
        tool calls do not leak into the user's chat. The final message is still
        part of the conversation from the user's point of view, so keep a small
        visible turn in the receiver session for follow-up questions.

        Configuration:
            scheduler_inject_to_session (bool, default True):
                Master switch. When False, this method is a no-op.
            scheduler_inject_max_per_session (int, default 3):
                Maximum scheduler-injected user/assistant pairs retained per
                session. Older injections are pruned automatically.

        ``run_id`` associates the stored pair with its scheduler execution;
        callers without an id remain supported without guessing association.

        Content is truncated to 4000 chars to prevent a single high-volume task
        from bloating one entry, while staying long enough that the history
        detail view (which recovers this copy) shows the full message for the
        vast majority of tasks. An ellipsis marks the rare over-limit case.
        """
        from config import conf
        if not conf().get("scheduler_inject_to_session", True):
            return
        if not session_id or not content:
            return

        max_len = 4000
        if len(content) > max_len:
            content = content[:max_len].rstrip() + "…"

        user_text = self._SCHEDULED_MARKER
        if task_description:
            user_text = f"{self._SCHEDULED_MARKER} {task_description}"

        messages = [
            {"role": "user", "content": [{"type": "text", "text": user_text}]},
            {"role": "assistant", "content": [{"type": "text", "text": content}]},
        ]

        # Association is storage metadata, not a field in model messages.
        persisted_messages = (
            [dict(message, run_id=run_id) for message in messages] if run_id else messages
        )

        # Persist first so the new pair gets a stable seq, then prune old
        # scheduler pairs in DB, then sync the in-memory agent.messages buffer.
        self._persist_messages(session_id, persisted_messages, channel_type, agent_id)

        keep_last_n = max(int(conf().get("scheduler_inject_max_per_session", 3) or 0), 0)
        try:
            deleted = self.get_conversation_store(agent_id).prune_scheduled_messages(
                session_id, keep_last_n=keep_last_n
            )
            if deleted:
                logger.debug(
                    f"[AgentBridge] Pruned {deleted} old scheduler messages "
                    f"for session={session_id} (keep_last_n={keep_last_n})"
                )
        except Exception as e:
            logger.warning(
                f"[AgentBridge] Failed to prune scheduled messages "
                f"for session={session_id}: {e}"
            )

        agent = self.get_cached_agent(session_id, agent_id=agent_id)
        if agent:
            try:
                with agent.messages_lock:
                    agent.messages.extend(messages)
                    self._prune_scheduled_in_memory(agent, keep_last_n)
            except Exception as e:
                logger.warning(
                    f"[AgentBridge] Failed to update in-memory scheduled output "
                    f"for session={session_id}: {e}"
                )

    @staticmethod
    def _trim_in_memory_to_turns(agent, keep_turns: int) -> None:
        """Bound ``agent.messages`` to the most recent ``keep_turns`` real
        user/assistant turns, dropping older history together with any
        intermediate tool_use/tool_result blocks that belonged to it.

        A "real" user message is any user message whose content is not solely a
        tool_result block — matches the heuristic used elsewhere when filtering
        history (see ``AgentInitializer._filter_text_only_messages``).

        No-op when the session is already within budget. Caller does not need
        to hold the lock; this method acquires it itself.
        """
        if keep_turns <= 0:
            return

        def _is_real_user(msg) -> bool:
            if not isinstance(msg, dict) or msg.get("role") != "user":
                return False
            content = msg.get("content")
            if isinstance(content, list):
                if any(
                    isinstance(b, dict) and b.get("type") == "tool_result"
                    for b in content
                ):
                    return False
                return any(
                    isinstance(b, dict) and b.get("type") == "text" and b.get("text")
                    for b in content
                )
            if isinstance(content, str):
                return bool(content.strip())
            return False

        with agent.messages_lock:
            msgs = agent.messages
            real_user_indices = [i for i, m in enumerate(msgs) if _is_real_user(m)]
            if len(real_user_indices) <= keep_turns:
                return

            # Cut at the (k-th from the end) real user message; keep everything
            # from there onwards so the surviving slice is still a valid
            # user/assistant sequence.
            cut_idx = real_user_indices[-keep_turns]
            if cut_idx == 0:
                return

            kept = msgs[cut_idx:]
            msgs.clear()
            msgs.extend(kept)
            logger.debug(
                f"[AgentBridge] Trimmed in-memory messages to last "
                f"{keep_turns} turns ({len(kept)} messages remain)"
            )

    @classmethod
    def _prune_scheduled_in_memory(cls, agent, keep_last_n: int) -> None:
        """Mirror conversation_store.prune_scheduled_messages on agent.messages.

        Caller must hold ``agent.messages_lock``.
        """
        if keep_last_n < 0:
            keep_last_n = 0

        markers = (cls._SCHEDULED_MARKER,) + cls._SCHEDULED_LEGACY_MARKERS

        def _is_marker_user(msg) -> bool:
            if not isinstance(msg, dict) or msg.get("role") != "user":
                return False
            content = msg.get("content")
            text = ""
            if isinstance(content, str):
                text = content
            elif isinstance(content, list):
                for block in content:
                    if isinstance(block, dict) and block.get("type") == "text":
                        text = block.get("text", "")
                        break
            return any(text.startswith(m) for m in markers)

        msgs = agent.messages
        pair_indices = []  # list of (user_idx, assistant_idx_or_None)
        for idx, msg in enumerate(msgs):
            if not _is_marker_user(msg):
                continue
            assistant_idx = None
            if idx + 1 < len(msgs):
                nxt = msgs[idx + 1]
                if isinstance(nxt, dict) and nxt.get("role") == "assistant":
                    assistant_idx = idx + 1
            pair_indices.append((idx, assistant_idx))

        if len(pair_indices) <= keep_last_n:
            return

        to_drop = pair_indices[: len(pair_indices) - keep_last_n]
        drop_set = set()
        for u_idx, a_idx in to_drop:
            drop_set.add(u_idx)
            if a_idx is not None:
                drop_set.add(a_idx)

        # Rebuild the list in place to keep external references stable.
        kept = [m for i, m in enumerate(msgs) if i not in drop_set]
        msgs.clear()
        msgs.extend(kept)

    @staticmethod
    def _strip_thinking_blocks(messages: list) -> list:
        """Return a shallow copy of messages with assistant "thinking" blocks removed."""
        cleaned = []
        for msg in messages:
            if not isinstance(msg, dict):
                cleaned.append(msg)
                continue
            if msg.get("role") != "assistant":
                cleaned.append(msg)
                continue
            content = msg.get("content")
            if not isinstance(content, list):
                cleaned.append(msg)
                continue
            filtered_blocks = [
                b for b in content
                if not (isinstance(b, dict) and b.get("type") == "thinking")
            ]
            if len(filtered_blocks) == len(content):
                cleaned.append(msg)
            else:
                new_msg = dict(msg)
                new_msg["content"] = filtered_blocks
                cleaned.append(new_msg)
        return cleaned

    def clear_session(self, session_id: str, agent_id: str = None):
        """
        Drop the cached agent for a session. Persisted history is untouched;
        the next request rebuilds the agent and restores from the store.

        Args:
            session_id: Session identifier to clear
        """
        if not session_id:
            return
        resolved_agent_id = self._resolve_agent_id(agent_id)
        key = self._runtime_key(resolved_agent_id, session_id)
        with self._agents_lock:
            removed = self._agent_instances.pop(key, None)
            if resolved_agent_id == self.agent_registry.default_agent_id:
                self.agents.pop(session_id, None)
        if removed is not None:
            logger.info(
                f"[AgentBridge] Clearing session: agent={resolved_agent_id}, "
                f"session={session_id}"
            )

    def clear_agent(self, agent_id: str) -> int:
        """Evict every live runtime instance for one agent workspace."""
        resolved_agent_id = self._resolve_agent_id(agent_id)
        with self._agents_lock:
            keys = [key for key in self._agent_instances if key[0] == resolved_agent_id]
            for key in keys:
                self._agent_instances.pop(key, None)
            self._default_agents.pop(resolved_agent_id, None)
            if resolved_agent_id == self.agent_registry.default_agent_id:
                self.agents.clear()
                self.default_agent = None
        return len(keys)
    
    def clear_all_sessions(self):
        """Clear all agent sessions"""
        with self._agents_lock:
            count = len(self._agent_instances)
            logger.info(f"[AgentBridge] Clearing all sessions ({count} total)")
            self._agent_instances.clear()
            self._default_agents.clear()
            self.agents.clear()
            self.default_agent = None
    
    def refresh_all_skills(self) -> int:
        """
        Refresh skills and conditional tools in all agent instances after
        environment variable changes. This allows hot-reload without restarting.

        Returns:
            Number of agent instances refreshed
        """
        import os
        from dotenv import load_dotenv
        from common.state_dir import env_file as workspace_env_file

        refreshed_count = 0
        loaded_env_files = set()

        # Each live agent reloads its own workspace .env, not just the one the
        # caller happens to be routed to.
        for agent_id, session_id, agent in self.iter_agent_instances():
            workspace_root = getattr(agent, "workspace_dir", None)
            env_file = str(workspace_env_file(base=workspace_root)) if workspace_root else None
            if env_file and env_file not in loaded_env_files and os.path.exists(env_file):
                load_dotenv(env_file, override=True)
                loaded_env_files.add(env_file)
                logger.info(
                    f"[AgentBridge] Reloaded environment variables from {env_file}"
                )
            # Refresh skills
            if hasattr(agent, 'skill_manager') and agent.skill_manager:
                agent.skill_manager.refresh_skills()

            # Refresh conditional tools (e.g. web_search depends on API keys)
            self._refresh_conditional_tools(agent)

            refreshed_count += 1

        if refreshed_count > 0:
            logger.info(f"[AgentBridge] Refreshed skills & tools in {refreshed_count} agent instance(s)")

        return refreshed_count

    @staticmethod
    def _refresh_conditional_tools(agent):
        """
        Add or remove conditional tools based on current environment variables.
        For example, web_search should only be present when BOCHA_API_KEY or
        LINKAI_API_KEY is set.
        """
        try:
            from agent.tools.web_search.web_search import WebSearch

            has_tool = any(t.name == "web_search" for t in agent.tools)
            available = WebSearch.is_available()

            if available and not has_tool:
                # API key was added - inject the tool
                tool = WebSearch()
                tool.model = agent.model
                agent.tools.append(tool)
                logger.info("[AgentBridge] web_search tool added (API key now available)")
            elif not available and has_tool:
                # API key was removed - remove the tool
                agent.tools = [t for t in agent.tools if t.name != "web_search"]
                logger.info("[AgentBridge] web_search tool removed (API key no longer available)")
        except Exception as e:
            logger.debug(f"[AgentBridge] Failed to refresh conditional tools: {e}")