openswarm/backend/apps/agents/manager/Messaging.py

"""Turn-producing message operations for AgentManager (send + queue), the ones that append a
user Message and spawn the agent loop. Editing a prior message lives in EditMessage; session-control
ops (stop / approve / branch / update) live in SessionControl. Pure relocation: self.* resolves across the MRO as before."""

import asyncio
import logging
from typing import List, Optional

from pydantic import BaseModel, ConfigDict
from typeguard import typechecked

from backend.apps.agents.core.models import AgentSession, Message
from backend.apps.agents.core.ws_manager import ws_manager
from backend.apps.settings.settings import load_settings
from backend.apps.agents.manager.run_browser_fast_path import run_browser_fast_path
from backend.apps.agents.manager.session.session_store import snapshot_session_now, load_session_data
from backend.apps.agents.manager.session.apply_context_window import apply_context_window
from backend.apps.agents.manager.prompt.tool_catalog import get_all_tool_names
from backend.apps.agents.manager.prompt.prompt_context import resolve_mode

logger = logging.getLogger(__name__)



from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol


class QueuedMessage(BaseModel):
    """A user message that arrived while a turn was live, held until that turn ends.
    Carries the full send_message argument set so delivery is a faithful replay."""

    model_config = ConfigDict(validate_assignment=True)

    prompt: str
    mode: Optional[str] = None
    model: Optional[str] = None
    provider: Optional[str] = None
    images: Optional[List] = None
    context_paths: Optional[List] = None
    forced_tools: Optional[List[str]] = None
    attached_skills: Optional[List] = None
    hidden: bool = False
    selected_browser_ids: Optional[List[str]] = None
    selected_app_output_ids: Optional[List[str]] = None
    selected_setting_ids: Optional[List[str]] = None
    client_message_id: Optional[str] = None


class Messaging(AgentManagerProtocol):
    @typechecked
    async def send_message(
        self,
        session_id: str,
        prompt: str,
        mode: Optional[str] = None,
        model: Optional[str] = None,
        provider: Optional[str] = None,
        images: Optional[List] = None,
        context_paths: Optional[List] = None,
        forced_tools: Optional[List[str]] = None,
        attached_skills: Optional[List] = None,
        hidden: bool = False,
        by_user: bool = False,
        selected_browser_ids: Optional[List[str]] = None,
        selected_app_output_ids: Optional[List[str]] = None,
        selected_setting_ids: Optional[List[str]] = None,
        client_message_id: Optional[str] = None,
    ):
        """Send a follow-up message to an existing session."""
        session = self.sessions.get(session_id)
        if not session:
            data = load_session_data(session_id)
            if data:
                session = AgentSession(**data)
                # This disk reload (and the closed_at wipe below) is how a late watchdog retry reopened a card the user had closed; a MACHINE send must not revive it (a human's own Resume click carries by_user and may).
                if hidden and not by_user and session.ended_by_user:
                    return
                apply_context_window(session)
                session.closed_at = None
                self.sessions[session_id] = session
            else:
                raise ValueError(f"Session {session_id} not found")
        # Every automatic resume arrives hidden; a human's Stop or close outranks all of them. `hidden` only means "do not render a user bubble", so the Resume chip's own click is hidden too and used to be swallowed here, leaving the chip to reappear forever (Eric, live, 2026-08-21). Authorship is what this guard cares about, so it asks by_user.
        if hidden and not by_user and session.ended_by_user:
            return
        if session.ended_by_user and (not hidden or by_user):
            session.ended_by_user = False

        existing = self.tasks.get(session_id)
        if existing and not existing.done():
            # A mid-turn message used to be silently dropped here (no bubble, no trace); queue it and the turn task's done callback replays it.
            self.pending_messages.setdefault(session_id, []).append(QueuedMessage(
                prompt=prompt, mode=mode, model=model, provider=provider, images=images,
                context_paths=context_paths, forced_tools=forced_tools,
                attached_skills=attached_skills, hidden=hidden,
                selected_browser_ids=selected_browser_ids,
                selected_app_output_ids=selected_app_output_ids,
                selected_setting_ids=selected_setting_ids,
                client_message_id=client_message_id,
            ))
            return

        session_changed = False
        if model and model != session.model:
            # Cross-provider model switches force a session fork. The CLI's resume transcript stores Anthropic-format content blocks with Anthropic tool_use_ids; replaying them on a non-Anthropic provider via 9Router's claude→openai translator corrupts history silently (fixMissingToolResponses stubs missing tool responses with placeholder text). Forking starts a new CLI session so history is re-sent fresh in whichever format the new provider expects.
            from backend.apps.agents.providers.registry import get_api_type as get_api_type_for_model
            if get_api_type_for_model(session.model) != get_api_type_for_model(model):
                session.needs_fork = True
                logger.info(f"[MCP-DEBUG] Forking session: api_type changed {session.model}→{model}")

            session.model = model
            apply_context_window(session)
            session_changed = True
        if mode and mode != session.mode:
            session.mode = mode
            mode_tools, _, _ = resolve_mode(mode, get_all_tool_names)
            session.allowed_tools = mode_tools
            session_changed = True
        if session_changed:
            await ws_manager.send_to_session(session_id, "agent:status", {
                "session_id": session_id,
                "status": session.status,
                "session": session.model_dump(mode="json"),
            })

        # Hidden messages (nudges, lost-step retries, auth heals) go out verbatim: the attribution prefix they used to carry (ENG-326) announced harness traffic on the subscription lane, whose provider filter blocks exactly that (Eric's call, 2026-08-21); honesty about who is speaking lives in the nudge texts themselves.
        skill_meta = [{"id": s["id"], "name": s["name"]} for s in (attached_skills or [])] or None
        image_meta = [{"data": img["data"], "media_type": img.get("media_type", "image/png")} for img in (images or [])] or None
        user_msg = Message(
            role="user",
            content=prompt,
            branch_id=session.active_branch_id,
            context_paths=context_paths if context_paths else None,
            attached_skills=skill_meta,
            forced_tools=forced_tools if forced_tools else None,
            images=image_meta,
            hidden=hidden,
            client_message_id=client_message_id,
        )
        session.messages.append(user_msg)
        # Status flips BEFORE the snapshot: the old order persisted a stale "completed" from the
        # prior turn, so a dirty death during a follow-up turn was invisible to the boot-time crash
        # detector (drilled live: round-2 kill -9 resumed nothing because disk said completed).
        session.status = "running"
        snapshot_session_now(session)
        await ws_manager.send_to_session(session_id, "agent:message", {
            "session_id": session_id,
            "message": user_msg.model_dump(mode="json"),
        })

        # A real user message opens a fresh silent-quit budget; the cap only guards within one ask.
        if not hidden:
            # A human actively driving the session forgives its crash history (breaker reset).
            session.crash_interrupt_count = 0
            session.empty_finish_nudges = 0
            session.empty_finish_progress_mark = 0
            session.empty_finish_surfaced = False
            session.auth_retry_used = False
            # The repeat-quit floor and the vanishing-quit rule key on this; one false positive used to arm both for the session's life (ENG-364).
            session.empty_finish_total = 0
            # A human is here and driving, so an earlier outage stops counting against the next one.
            session.reconnect_attempts = 0
            session.awaiting_reconnect = False
        # Fire a background aux LLM call to generate a 3-6 word verb-phrase describing this turn ("Auditing the pull request", "Drafting your email"). The narrator pill swaps from its heuristic verb to this label as soon as it lands, usually ~500ms-1s into the turn, which is exactly when "Thinking…" starts feeling generic. Provider-agnostic via resolve_aux_model. Non-blocking; failure is silent and the heuristic stays.
        if not hidden and prompt:
            try:
                asyncio.create_task(
                    self.generate_turn_label(session_id, user_msg.id, prompt)
                )
            except Exception:
                pass

        is_first_message = sum(1 for m in session.messages if m.role == "user") == 1

        session.status = "running"
        await ws_manager.send_to_session(session_id, "agent:status", {
            "session_id": session_id,
            "status": "running",
            "session": session.model_dump(mode="json"),
        })

        # Browser fast path: a plainly browser-only first message skips the orchestrator LLM entirely (it was ~2/3 of the token bill on these tasks, spent deciding "delegate to a browser" and restating the outcome). Conservative gates + a cheap aux classifier; any miss or error falls through to the normal loop.
        fast_verdict = "no"
        fast_brief = ""
        if not hidden:
            try:
                from backend.apps.agents.browser import browser_fast_path
                extras = bool(images or context_paths or forced_tools or attached_skills
                               or len(selected_browser_ids or []) > 1)
                if browser_fast_path.fast_path_eligible(
                    prompt, session.mode or "", session.dashboard_id, is_first_message, extras,
                ):
                    from backend.apps.agents.providers.registry import get_api_type
                    fast_verdict, fast_brief = await browser_fast_path.classify_and_brief(
                        prompt, load_settings(), get_api_type(session.model),
                    )
            except Exception as e:
                logger.warning(f"[browser-fast-path] gate error, normal path: {e}")

        if fast_verdict != "no":
            task = asyncio.create_task(run_browser_fast_path(session, session_id, prompt, selected_browser_ids, fast_brief, fast_verdict))
        else:
            task = asyncio.create_task(self.run_agent_loop(session_id, prompt, images=images, context_paths=context_paths, forced_tools=forced_tools, attached_skills=attached_skills, selected_browser_ids=selected_browser_ids, selected_app_output_ids=selected_app_output_ids, selected_setting_ids=selected_setting_ids))
        self.register_turn_task(session_id, task)

    @typechecked
    def register_turn_task(self, session_id: str, task: asyncio.Task) -> None:
        """One chokepoint for installing a turn task: every turn end (success, error, stop)
        fires the done callback, so a queued mid-turn message can never strand."""
        self.tasks[session_id] = task
        task.add_done_callback(lambda t: self.deliver_next_queued_message(session_id))

    @typechecked
    def deliver_next_queued_message(self, session_id: str) -> None:
        """Replay the oldest queued mid-turn message once no turn is live. One at a time:
        the delivered turn's own done callback drains the rest."""
        queue = self.pending_messages.get(session_id)
        if not queue:
            return
        live = self.tasks.get(session_id)
        if live and not live.done():
            return
        qm = queue.pop(0)
        if not queue:
            self.pending_messages.pop(session_id, None)
        asyncio.create_task(self.send_message(
            session_id, qm.prompt, mode=qm.mode, model=qm.model, provider=qm.provider,
            images=qm.images, context_paths=qm.context_paths, forced_tools=qm.forced_tools,
            attached_skills=qm.attached_skills, hidden=qm.hidden,
            selected_browser_ids=qm.selected_browser_ids,
            selected_app_output_ids=qm.selected_app_output_ids,
            selected_setting_ids=qm.selected_setting_ids,
            client_message_id=qm.client_message_id,
        ))