import os
import re
import regex
import json
import time
import logging
import base64
import uuid
import asyncio
from datetime import datetime
from fastapi import HTTPException
from Memories import Memories
from Websearch import Websearch
from Extensions import Extensions
from Memories import extract_keywords
from ApiClient import (
    Agent,
    Prompts,
    Chain,
    Conversations,
    AGIXT_URI,
)
from MagicalAuth import (
    MagicalAuth,
    convert_time,
    get_current_user_time,
    impersonate_user,
)
from Globals import getenv, DEFAULT_USER, get_tokens
from WebhookManager import WebhookEventEmitter
from middleware import log_silenced_exception

logging.basicConfig(
    level=getenv("LOG_LEVEL"),
    format=getenv("LOG_FORMAT"),
)

# Initialize webhook event emitter
webhook_emitter = WebhookEventEmitter()

# Pre-compiled regex patterns for hot-path tag processing
# NOTE: All thinking-related patterns match BOTH <think> and <thinking> forms.
# Models like DeepSeek-R1 and Qwen QwQ use <think>, while AGiXT's prompts use <thinking>.
# We preserve the model's native form to avoid confusing it during continuation prompts.
_RE_ANSWER_OPEN = re.compile(r"<answer>", re.IGNORECASE)
_RE_ANSWER_CLOSE = re.compile(r"</answer>", re.IGNORECASE)
_RE_THINKING_OPEN = re.compile(r"<think(?:ing)?>", re.IGNORECASE)
_RE_THINKING_CLOSE = re.compile(r"</think(?:ing)?>", re.IGNORECASE)
_RE_REFLECTION_OPEN = re.compile(r"<reflection>", re.IGNORECASE)
_RE_REFLECTION_CLOSE = re.compile(r"</reflection>", re.IGNORECASE)
_RE_PROTOCOL_ANSWER_BOUNDARY = re.compile(
    r"(</(?:think(?:ing)?|reflection)>)\s*(<answer>)", re.IGNORECASE
)
_RE_CLOSING_TAG = re.compile(
    r"</(think(?:ing)?|reflection|step|execute|output)>\s*$", re.IGNORECASE
)
_RE_TOOL_CALL = re.compile(
    r"<name>\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*</name>", re.IGNORECASE
)
_RE_STEP_TAG = re.compile(r"<step>(.*?)</step>", re.DOTALL | re.IGNORECASE)
_RE_REWARD_TAG = re.compile(r"<reward>(.*?)</reward>", re.DOTALL | re.IGNORECASE)
_RE_COUNT_TAG = re.compile(r"<count>(.*?)</count>", re.DOTALL | re.IGNORECASE)
_RE_RATE_TAG = re.compile(r"<rate>.*?</rate>", re.DOTALL)
_RE_EXECUTE_TAG = re.compile(r"<execute>.*?</execute>", re.DOTALL | re.IGNORECASE)
_RE_INTERACTION_TAG = re.compile(
    r"<interaction\b[^>]*>.*?</interaction>", re.DOTALL | re.IGNORECASE
)
_RE_OUTPUT_TAG = re.compile(r"<output>.*?</output>", re.DOTALL)
_RE_NAME_TAG = re.compile(r"<name>.*?</name>", re.DOTALL)
_RE_THINKING_BLOCK = re.compile(
    r"<think(?:ing)?>.*?</think(?:ing)?>", re.DOTALL | re.IGNORECASE
)
_RE_REFLECTION_BLOCK = re.compile(
    r"<reflection>.*?</reflection>", re.DOTALL | re.IGNORECASE
)
_RE_MULTI_NEWLINE = re.compile(r"\n\s*\n\s*\n")
_RE_TRIPLE_NEWLINE = re.compile(r"\n{3,}")
_RE_CUSTOM_FORMAT = re.compile(r"(?<!{){([^{}\n]+)}(?!})")
_RE_MARKDOWN_FENCE_LINE = re.compile(r"(?m)^[ \t]*(`{3,}|~{3,})")
_RE_PROTOCOL_CONTAINER_TAG = re.compile(
    r"</?(think(?:ing)?|reflection)>", re.IGNORECASE
)


def _get_unclosed_markdown_fence_marker(text: str) -> str:
    """
    Return the currently open markdown fence marker, if the text ends inside one.
    """
    open_marker = ""
    for match in _RE_MARKDOWN_FENCE_LINE.finditer(text or ""):
        marker = match.group(1)
        if not open_marker:
            open_marker = marker
        elif marker[0] == open_marker[0] and len(marker) >= len(open_marker):
            open_marker = ""
    return open_marker


def _get_unclosed_protocol_container_closers(text: str) -> list[str]:
    """
    Return closing tags needed to leave dangling thinking/reflection containers.

    Recovery can happen after the stream parser has shown answer text to the UI
    while the raw provider text still contains an unclosed <think>/<thinking> or
    <reflection> block. A recovered <answer> must be appended outside those
    containers or has_complete_answer() will keep treating it as hidden work.
    """
    stack = []
    for match in _RE_PROTOCOL_CONTAINER_TAG.finditer(text or ""):
        raw_tag = match.group(0).lower()
        tag_name = match.group(1).lower()
        canonical_name = "thinking" if tag_name in ("think", "thinking") else tag_name

        if raw_tag.startswith("</"):
            for idx in range(len(stack) - 1, -1, -1):
                if stack[idx][0] == canonical_name:
                    del stack[idx:]
                    break
            continue

        close_tag = "</think>" if tag_name == "think" else f"</{canonical_name}>"
        stack.append((canonical_name, close_tag))

    return [close_tag for _, close_tag in reversed(stack)]


def _append_recovered_answer_block(response: str, answer_text: str) -> str:
    """
    Append a synthetic protocol answer outside markdown/code/protocol boundaries.
    """
    response = response or ""
    answer_text = (answer_text or "").strip()
    if not answer_text:
        return response

    additions = []
    if response and not response.endswith(("\n", "\r")):
        additions.append("\n")
    open_fence = _get_unclosed_markdown_fence_marker(response)
    if open_fence:
        additions.append(f"\n{open_fence}\n")

    for close_tag in _get_unclosed_protocol_container_closers(
        response + "".join(additions)
    ):
        additions.append(f"{close_tag}\n")

    additions.append(f"<answer>{answer_text}</answer>")
    return response + "".join(additions)


def _close_recovered_answer_block(response: str) -> str:
    """
    Close an already-open top-level answer after streamed content was emitted.
    """
    response = response or ""
    additions = []
    open_fence = _get_unclosed_markdown_fence_marker(response)
    if open_fence:
        if response and not response.endswith(("\n", "\r")):
            additions.append("\n")
        additions.append(f"{open_fence}\n")
    additions.append("</answer>")
    return response + "".join(additions)


def _convert_interaction_to_execute(response: str) -> str:
    """
    Detect leaked <interaction> XML in the agent's response and convert it
    to a proper <execute> call for 'Interact with Webpage'.

    The web browsing extension uses <interaction><step>...</step></interaction>
    as an internal browser control protocol. When the outer agent outputs this
    format directly (instead of using <execute>), it means the LLM confused
    the internal protocol with the command API.

    - 'respond' operations: the value IS the agent's answer — extract it
    - 'done' operations: task complete signal — remove the block
    - browser actions (click, fill, etc.): convert to <execute> so the
      web browsing command can handle them properly
    """
    import xml.etree.ElementTree as ET

    for match in _RE_INTERACTION_TAG.finditer(response):
        block = match.group(0)
        try:
            root = ET.fromstring(block)
            steps = root.findall(".//step")
            if not steps:
                # Empty interaction block — remove it
                response = response.replace(block, "", 1)
                continue

            step = steps[0]
            operation = (step.findtext("operation") or "").strip().lower()
            value = (step.findtext("value") or "").strip()
            description = (step.findtext("description") or "").strip()

            if operation == "respond":
                # Agent wants to report findings — the value IS the answer
                logging.info(
                    "[interaction_convert] Extracted 'respond' value "
                    f"({len(value)} chars) from leaked <interaction> XML"
                )
                response = response.replace(block, value, 1)
            elif operation == "done":
                logging.info("[interaction_convert] Removed 'done' <interaction> block")
                response = response.replace(block, "", 1)
            else:
                # Browser action — convert to proper <execute> call
                task = description if description else f"{operation}: {value}"
                execute_block = (
                    f"<execute>\n"
                    f"<name>Interact with Webpage</name>\n"
                    f"<url>search.brave.com</url>\n"
                    f"<task>{task}</task>\n"
                    f"</execute>"
                )
                logging.info(
                    f"[interaction_convert] Converted '{operation}' <interaction> "
                    f"to <execute> for 'Interact with Webpage': {task[:100]}"
                )
                response = response.replace(block, execute_block, 1)
        except ET.ParseError as e:
            logging.warning(
                f"[interaction_convert] Could not parse <interaction> XML: {e}"
            )
            continue

    return response


_RE_DOT_SPACE = re.compile(r"\. ")
_RE_OUTPUT_BLOCK = re.compile(r"<output>(.*?)</output>", re.DOTALL | re.IGNORECASE)
_RE_THINKING_CONTENT = re.compile(
    r"<think(?:ing)?>(.*?)</think(?:ing)?>", re.DOTALL | re.IGNORECASE
)
_RE_REFLECTION_CONTENT = re.compile(
    r"<reflection>(.*?)</reflection>", re.DOTALL | re.IGNORECASE
)
_RE_REWARD_CONTENT = re.compile(r"<reward>(.*?)</reward>", re.DOTALL | re.IGNORECASE)
_RE_COUNT_CONTENT = re.compile(r"<count>(.*?)</count>", re.DOTALL | re.IGNORECASE)
_RE_FILE_BACKTICK = re.compile(r"`([^`]+\.[a-zA-Z0-9]+)`")
_RE_THINKING_REFLECTION = re.compile(
    r"<(think(?:ing)?|reflection)>(.*?)(?=<(?:think(?:ing)?|reflection|answer)|$)",
    re.DOTALL,
)


async def _ability_selection_inference(server_url: str, model: str, prompt: str) -> str:
    """
    Make a direct inference call to a dedicated ability selection server.

    Uses the OpenAI-compatible chat completions API to call a fast, small model
    for command/ability selection. This avoids tying up the main LLM provider
    for the lightweight selection step.

    Args:
        server_url: Base URL of the ezlocalai/OpenAI-compatible server
        model: Model name to use for inference
        prompt: The selection prompt

    Returns:
        str: The model's response text
    """
    import requests
    from functools import partial

    api_url = server_url.rstrip("/")
    if "/v1" not in api_url:
        api_url += "/v1"
    api_url = api_url.rstrip("/") + "/chat/completions"

    api_key = getenv("EZLOCALAI_API_KEY", "none")
    if not api_key:
        api_key = "none"

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.3,
        "top_p": 0.9,
        "stream": False,
    }

    loop = asyncio.get_event_loop()
    try:
        resp = await loop.run_in_executor(
            None,
            partial(
                requests.post,
                api_url,
                headers=headers,
                json=payload,
                timeout=(600, 120),
            ),
        )
        resp.raise_for_status()
        data = resp.json()
        response = data["choices"][0]["message"]["content"]
        # Strip thinking tags if present
        response = _RE_THINKING_REFLECTION.sub("", response).strip()
        if "<answer>" in response:
            response = response.split("<answer>")[-1]
        if "</answer>" in response:
            response = response.split("</answer>")[0]
        return response.strip()
    except Exception as e:
        logging.error(f"[_ability_selection_inference] Error: {e}")
        return ""


def _small_model_inference_sync(
    prompt: str,
    temperature: float = 0.3,
    max_tokens: int = 900,
) -> str:
    """
    Synchronous call to the small (0.8B) model for utility tasks like summarization.
    Uses the same server/model as ability selection.
    """
    import requests as _requests

    server_url = getenv("ABILITY_SELECTION_SERVER", "")
    model = getenv("ABILITY_SELECTION_MODEL", "unsloth/Qwen3.5-4B-GGUF")
    if not server_url:
        return ""

    api_url = server_url.rstrip("/")
    if "/v1" not in api_url:
        api_url += "/v1"
    api_url = api_url.rstrip("/") + "/chat/completions"

    api_key = getenv("EZLOCALAI_API_KEY", "none") or "none"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "top_p": 0.9,
        "max_tokens": max_tokens,
        "stream": False,
    }

    try:
        resp = _requests.post(api_url, headers=headers, json=payload, timeout=(10, 120))
        resp.raise_for_status()
        data = resp.json()
        response = data["choices"][0]["message"]["content"]
        response = _RE_THINKING_REFLECTION.sub("", response).strip()
        if "<answer>" in response:
            response = response.split("<answer>")[-1]
        if "</answer>" in response:
            response = response.split("</answer>")[0]
        return response.strip()
    except Exception as e:
        logging.error(f"[_small_model_inference_sync] Error: {e}")
        return ""


def generate_conversation_summary(
    messages: list,
    existing_summary: str = "",
    conversation_name: str = "",
) -> str:
    """
    Generate or update a conversation summary using the small model.

    Args:
        messages: List of dicts with 'role' and 'message' keys
        existing_summary: Current summary to update (empty for new)
        conversation_name: Name of the conversation for context

    Returns:
        Updated summary string
    """
    if not messages:
        return existing_summary
    logging.info(
        f"[generate_conversation_summary] Starting for '{conversation_name}' ({len(messages)} messages)"
    )

    # Build conversation text, chunk through 100k at a time
    CHUNK_SIZE = 100000
    all_text = ""
    for msg in messages:
        role = msg.get("role", "unknown")
        content = msg.get("message", "")
        if (
            not content
            or content.startswith("[ACTIVITY]")
            or content.startswith("[SUBACTIVITY]")
        ):
            continue
        all_text += f"{role}: {content}\n\n"

    if not all_text.strip():
        return existing_summary

    # Process in chunks
    running_summary = existing_summary
    chunks = []
    for i in range(0, len(all_text), CHUNK_SIZE):
        chunks.append(all_text[i : i + CHUNK_SIZE])

    for chunk_idx, chunk in enumerate(chunks):
        if running_summary:
            prompt = f"""You are analyzing a conversation to maintain a living summary. Below is the current summary followed by the next portion of conversation.

## Current Summary
{running_summary}

## Conversation Chunk ({chunk_idx + 1}/{len(chunks)})
{chunk}

## Instructions
Update the summary to incorporate new information from this conversation chunk. The summary should capture:

1. **Topics Discussed**: Key subjects and themes covered
2. **User Preferences & Style**: Communication preferences, technical level, preferred approaches, what they like/dislike
3. **Lessons Learned**: What worked well and what didn't. Successful approaches the AI used that the user responded positively to
4. **Key Decisions**: Important choices or conclusions reached
5. **User Context**: Personal/professional details shared that help understand the user better
6. **AI Behavior Notes**: Things the AI did well (approaches that resolved the user's needs effectively) and things to avoid

Keep the summary concise but comprehensive. Focus on durable insights that would be useful in future conversations. Remove outdated information if superseded. Write in third person about the user."""
        else:
            prompt = f"""You are analyzing a conversation to create a summary. Below is a portion of the conversation.

## Conversation: {conversation_name}
## Chunk ({chunk_idx + 1}/{len(chunks)})
{chunk}

## Instructions
Create a summary of this conversation. The summary should capture:

1. **Topics Discussed**: Key subjects and themes covered
2. **User Preferences & Style**: Communication preferences, technical level, preferred approaches, what they like/dislike
3. **Lessons Learned**: What worked well and what didn't. Successful approaches the AI used that the user responded positively to
4. **Key Decisions**: Important choices or conclusions reached
5. **User Context**: Personal/professional details shared that help understand the user better
6. **AI Behavior Notes**: Things the AI did well (approaches that resolved the user's needs effectively) and things to avoid

Keep the summary concise but comprehensive. Focus on durable insights that would be useful in future conversations. Write in third person about the user."""

        result = _small_model_inference_sync(prompt, temperature=0.3)
        if result:
            running_summary = result
            logging.info(
                f"[generate_conversation_summary] Chunk {chunk_idx + 1}/{len(chunks)} processed ({len(result)} chars)"
            )

    logging.info(
        f"[generate_conversation_summary] Complete for '{conversation_name}' ({len(running_summary)} chars)"
    )
    return running_summary


async def update_conversation_summary_after_interaction(
    conversation: "Conversations",
    user_input: str,
    agent_response: str,
    agent_name: str,
    user_id: str = "",
):
    """
    Update the conversation summary after an interaction using the small model.
    Also updates the persistent user knowledge profile.
    Runs in a background thread to avoid blocking the response.
    """
    try:
        conversation_name = conversation.conversation_name
        logging.info(
            f"[update_conversation_summary] Starting background update for '{conversation_name}'"
        )

        def _do_update():
            existing_summary = conversation.get_conversation_summary()
            if existing_summary:
                prompt = f"""You are maintaining a living conversation summary. A new interaction just occurred.

## Current Summary
{existing_summary}

## Latest Interaction
User: {user_input[:12000]}

{agent_name}: {agent_response[:12000]}

## Instructions
Update the summary to incorporate this latest interaction. Focus on:

1. **New topics or context** introduced in this exchange
2. **User preferences** revealed (communication style, what they want, how they respond)
3. **What worked well** in the AI's response (if the user seemed satisfied or the request was resolved)
4. **Lessons learned** about how to better serve this user
5. **Key decisions or outcomes** from this exchange

Keep previous summary content that's still relevant. Be concise but comprehensive. Write in third person about the user."""
            else:
                prompt = f"""You are creating an initial conversation summary based on an interaction.

## Conversation: {conversation_name}

## Interaction
User: {user_input[:12000]}

{agent_name}: {agent_response[:12000]}

## Instructions
Create an initial summary capturing:

1. **Topics Discussed**: What this conversation is about
2. **User Preferences & Style**: Any communication preferences or context about the user
3. **What Worked Well**: Effective approaches used in the response
4. **Key Outcomes**: Results or decisions from this exchange
5. **User Context**: Any personal/professional details that help understand the user

Be concise but comprehensive. Write in third person about the user."""

            result = _small_model_inference_sync(
                prompt,
                temperature=0.3,
                max_tokens=800,
            )
            if result:
                conversation.set_conversation_summary(result)
                logging.info(
                    f"[update_conversation_summary] Saved summary for '{conversation_name}' ({len(result)} chars)"
                )

            # Also update persistent user knowledge profile
            if user_id:
                try:
                    update_user_knowledge_after_interaction(
                        user_id=user_id,
                        user_input=user_input,
                        agent_response=agent_response,
                        agent_name=agent_name,
                        conversation_name=conversation_name,
                    )
                except Exception as e:
                    logging.warning(
                        f"[update_user_knowledge] Error in background thread: {e}"
                    )

        import threading

        thread = threading.Thread(target=_do_update, daemon=True)
        thread.start()
    except Exception as e:
        logging.warning(f"[update_conversation_summary] Error: {e}")


def get_user_knowledge(user_id: str) -> str:
    """Get the agent-maintained knowledge about a user."""
    from DB import get_session, User

    session = get_session()
    try:
        user = session.query(User).filter(User.id == user_id).first()
        if user and user.knowledge:
            return user.knowledge
        return ""
    except Exception as e:
        logging.warning(f"[get_user_knowledge] Error: {e}")
        return ""
    finally:
        session.close()


def set_user_knowledge(user_id: str, knowledge: str) -> str:
    """Set the agent-maintained knowledge about a user."""
    from DB import get_session, User

    session = get_session()
    try:
        user = session.query(User).filter(User.id == user_id).first()
        if user:
            user.knowledge = knowledge
            session.commit()
            return knowledge
        return ""
    except Exception as e:
        logging.warning(f"[set_user_knowledge] Error: {e}")
        return ""
    finally:
        session.close()


def update_user_knowledge_after_interaction(
    user_id: str,
    user_input: str,
    agent_response: str,
    agent_name: str,
    conversation_name: str = "",
):
    """
    Extract and update persistent user knowledge from an interaction.
    Called synchronously from a background thread.
    Focuses on personal details, preferences, and behavioral patterns — NOT code or technical content.
    """
    existing_knowledge = get_user_knowledge(user_id)

    if existing_knowledge:
        prompt = f"""You are maintaining a persistent knowledge profile about a user based on their interactions with an AI assistant. A new interaction just occurred.

## Current User Knowledge
{existing_knowledge}

## Latest Interaction (Conversation: {conversation_name})
User: {user_input[:8000]}

{agent_name}: {agent_response[:8000]}

## Instructions
Update the user knowledge profile to incorporate any new personal insights from this interaction. Focus ONLY on:

1. **Personal Details**: Names (family, pets, friends), locations, age, occupation, company, relationships
2. **Preferences & Tastes**: Favorite things, communication style preferences, how they like to work, tools they prefer
3. **Lifestyle**: Diet, exercise routine, hobbies, interests, daily habits, routines
4. **Professional Context**: Role, industry, projects, team, goals, challenges
5. **Communication Patterns**: How they ask questions, level of detail they want, tone preferences
6. **Important Dates**: Birthdays, anniversaries, deadlines they've mentioned
7. **Values & Priorities**: What matters to them, what frustrates them, what delights them

DO NOT include:
- Code snippets, technical implementations, or project-specific details
- Conversation-specific task context (that belongs in conversation summaries)
- Anything the user hasn't explicitly shared or strongly implied

Keep previous knowledge that's still relevant. Remove anything that's been contradicted or corrected. Be concise — use bullet points. If there's nothing new to add about the user personally, return the existing knowledge unchanged."""
    else:
        prompt = f"""You are creating an initial knowledge profile about a user based on their interaction with an AI assistant.

## Interaction (Conversation: {conversation_name})
User: {user_input[:8000]}

{agent_name}: {agent_response[:8000]}

## Instructions
Extract any personal insights about the user from this interaction. Focus ONLY on:

1. **Personal Details**: Names (family, pets, friends), locations, age, occupation
2. **Preferences & Tastes**: Communication style, how they like to work, tools they prefer  
3. **Lifestyle**: Diet, exercise, hobbies, interests, daily habits
4. **Professional Context**: Role, industry, projects, team, goals
5. **Communication Patterns**: How they ask questions, level of detail they want
6. **Values & Priorities**: What matters to them, what frustrates them

DO NOT include code, technical details, or task-specific context.
If there's nothing personal to extract from this interaction, respond with exactly: NO_UPDATE

Be concise — use bullet points."""

    result = _small_model_inference_sync(
        prompt,
        temperature=0.3,
        max_tokens=600,
    )
    if result and result.strip().upper() != "NO_UPDATE":
        set_user_knowledge(user_id, result)
        preview = " ".join(result.split())[:220]
        logging.info(
            f"[update_user_knowledge] Saved knowledge for user {user_id[:8]}... ({len(result)} chars) preview='{preview}'"
        )
    elif result:
        logging.info(
            f"[update_user_knowledge] Model returned NO_UPDATE for user {user_id[:8]}..."
        )


async def stream_inference_to_string(agent, prompt: str, **kwargs) -> str:
    """
    Run inference with stream=True and collect the full response as a string.

    This is faster than non-streaming inference because:
    1. Tokens start arriving immediately instead of waiting for full generation
    2. The server can begin sending data as soon as the first token is ready
    3. For thinking models, we get the response without blocking on think tokens

    All internal inference in AGiXT should use streaming. When a caller needs
    the full string response (e.g., command selection, context reduction), use
    this function instead of agent.inference(stream=False).

    Args:
        agent: The Agent instance to call inference on
        prompt: The prompt to send
        **kwargs: Additional kwargs passed to agent.inference (e.g., use_smartest)

    Returns:
        str: The full collected response text
    """
    import queue
    import threading

    stream_obj = await agent.inference(prompt=prompt, stream=True, **kwargs)
    if stream_obj is None:
        return ""

    # If for some reason inference returned a string (fallback), just return it
    if isinstance(stream_obj, str):
        return stream_obj

    collected = []

    if hasattr(stream_obj, "__aiter__"):
        # Async iterator
        async for chunk in stream_obj:
            token = _extract_token(chunk)
            if token:
                collected.append(token)
    else:
        # Sync iterator (e.g., requests-based SSE stream from ezlocalai)
        chunk_queue = queue.Queue()
        done_event = threading.Event()

        def _sync_iter():
            try:
                for chunk in stream_obj:
                    chunk_queue.put(("chunk", chunk))
                chunk_queue.put(("done", None))
            except Exception as e:
                chunk_queue.put(("error", e))
            finally:
                done_event.set()

        thread = threading.Thread(target=_sync_iter, daemon=True)
        thread.start()

        while True:
            try:
                msg_type, msg_data = chunk_queue.get_nowait()
                if msg_type == "chunk":
                    token = _extract_token(msg_data)
                    if token:
                        collected.append(token)
                elif msg_type == "done":
                    break
                elif msg_type == "error":
                    logging.error(
                        f"[stream_inference_to_string] Stream error: {msg_data}"
                    )
                    break
            except queue.Empty:
                if done_event.is_set():
                    break
                await asyncio.sleep(0.005)

    result = "".join(collected)

    # Track token usage (non-streaming path does this in Agent.inference)
    input_tokens = get_tokens(prompt)
    output_tokens = get_tokens(result)
    try:
        agent.auth.increase_token_counts(
            input_tokens=input_tokens, output_tokens=output_tokens
        )
    except Exception:
        pass  # Don't fail on billing tracking errors

    # Clean up like the non-streaming path does
    result = result.replace("\\_", "_")
    if result.endswith("\n\n"):
        result = result[:-2]

    return result


def _extract_token(chunk) -> str:
    """Extract text token from a stream chunk (handles multiple formats)."""
    if isinstance(chunk, str):
        return chunk
    if hasattr(chunk, "choices") and len(chunk.choices) > 0:
        delta = chunk.choices[0].delta
        if hasattr(delta, "content") and delta.content:
            return delta.content
        if hasattr(delta, "reasoning_content") and delta.reasoning_content:
            return delta.reasoning_content
    return ""


def extract_top_level_answer(response: str) -> str:
    """
    Extract the content from a top-level <answer>...</answer> block.

    This properly handles cases where <answer> appears inside <thinking> blocks:
    - <thinking>The <answer> block format...</thinking>
      <answer>Real answer</answer>
      → Returns "Real answer"
    - <answer>Simple answer</answer>
      → Returns "Simple answer"
    - <thinking>I'll use <answer>example</answer> format</thinking>
      → Returns "" (no top-level answer)

    Args:
        response: The full response text

    Returns:
        str: The extracted answer content, or empty string if no top-level answer
    """
    # Find all <answer> tags
    answer_opens = []
    for match in _RE_ANSWER_OPEN.finditer(response):
        open_pos = match.start()

        # Check if this is a real tag (not just mentioned in text)
        if not is_real_answer_tag(response, open_pos):
            continue

        # Check if this answer is at top level (not inside thinking/reflection)
        text_before = response[:open_pos]
        thinking_depth = len(_RE_THINKING_OPEN.findall(text_before)) - len(
            _RE_THINKING_CLOSE.findall(text_before)
        )
        reflection_depth = len(_RE_REFLECTION_OPEN.findall(text_before)) - len(
            _RE_REFLECTION_CLOSE.findall(text_before)
        )

        if thinking_depth == 0 and reflection_depth == 0:
            answer_opens.append(match)

    if not answer_opens:
        return ""

    # Use the LAST top-level answer (in case model restarts its answer)
    last_answer = answer_opens[-1]
    answer_start = last_answer.end()

    # Find the matching </answer>
    text_after = response[answer_start:]
    answer_depth = 1
    pos = 0

    while pos < len(text_after):
        next_open_lower = text_after.lower().find("<answer>", pos)
        next_close_lower = text_after.lower().find("</answer>", pos)

        next_open = float("inf") if next_open_lower == -1 else next_open_lower
        next_close = float("inf") if next_close_lower == -1 else next_close_lower

        if next_open == float("inf") and next_close == float("inf"):
            # No closing tag found, return everything after <answer>
            return text_after.strip()

        if next_open < next_close:
            answer_depth += 1
            pos = next_open + len("<answer>")
        else:
            answer_depth -= 1
            if answer_depth == 0:
                return text_after[:next_close].strip()
            pos = next_close + len("</answer>")

    # No closing found, return everything
    return text_after.strip()


def is_real_answer_tag(response: str, match_start: int) -> bool:
    """
    Determine if an <answer> tag at the given position is a real XML tag
    or just a mention in natural language (e.g., "I'll respond in the <answer> block").

    A real answer tag must be:
    1. At the start of the string, OR
    2. The first thing on a new line

    It must not be inside a fenced markdown code block. This allows AGiXT to
    safely discuss or write literal <answer> tags while coding without
    accidentally ending the reasoning/acting phase.

    This is much stricter than trying to detect fake tags by context after the tag.

    Args:
        response: The full response text
        match_start: The position where <answer> starts

    Returns:
        bool: True if this appears to be a real XML tag, False if it's just a mention
    """
    text_before = response[:match_start]

    # Reject answer tags inside fenced markdown code blocks. Count opening and
    # closing fences before the candidate tag; an odd count means the tag is
    # code/content, not the protocol boundary.
    fence_count = len(re.findall(r"(?m)^[ \t]*(?:```|~~~)", text_before))
    if fence_count % 2 != 0:
        return False

    # At start of string - real tag.
    if match_start == 0:
        return True

    # The opening answer tag must be the first thing on a new line. Do not
    # allow inline tags after sentences, closing tags, indentation, or code.
    previous_char = response[match_start - 1]
    return previous_char in ("\n", "\r")


def find_real_answer_tags(response: str, tag_type: str = "open") -> list:
    """
    Find all real <answer> or </answer> tags in the response,
    filtering out mentions in natural language.

    Args:
        response: The full response text
        tag_type: "open" for <answer>, "close" for </answer>

    Returns:
        list: List of match positions for real tags
    """
    if tag_type == "open":
        pat = _RE_ANSWER_OPEN
    else:
        pat = _RE_ANSWER_CLOSE

    positions = []
    for match in pat.finditer(response):
        if tag_type == "close" or is_real_answer_tag(response, match.start()):
            positions.append(match.start())

    return positions


def has_complete_answer(response: str) -> bool:
    """
    Check if the response contains a complete <answer>...</answer> block at the top level
    with meaningful content (not just step/thinking tags).

    A complete answer means:
    1. There is an <answer> tag (that's a real XML tag, not just mentioned in text)
    2. There is a matching </answer> tag
    3. The </answer> is NOT inside a <thinking> or <reflection> block
    4. The content inside has actual text after removing step/reward/count tags
    5. There are no pending tool calls (execute blocks, name tags) that haven't been executed

    This handles edge cases like:
    - <answer>Some text <thinking>thoughts</thinking> more text</answer> - COMPLETE (thinking is inside answer)
    - <answer>Some text</answer> - COMPLETE
    - <answer>Some text <thinking>thoughts</thinking> - INCOMPLETE (no closing answer after thinking)
    - <thinking><answer>fake</answer></thinking> - NOT a valid top-level answer
    - <answer><step>plan step</step></answer> - NOT COMPLETE (only contains step tags)
    - <thinking>I'll put my response in the <answer> block</thinking> - NOT an answer (just mentioned in text)
    - <thinking><name>tool_name</name></thinking>
      <answer>...</answer> - NOT COMPLETE (has unexecuted tool call)

    Returns:
        bool: True if there's a complete top-level answer block with meaningful content
    """
    # Check for pending tool calls - if there's a <name>...</name> that looks like a tool call
    # and no corresponding execute output, don't consider the answer complete
    # Pattern: <name>word</name> where word looks like a command name (no spaces, alphanumeric + underscore)
    tool_call_pattern = r"<name>\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*</name>"
    if _RE_TOOL_CALL.search(response):
        # Check if this tool call has been executed (would have <output> after it)
        # or is inside an <execute>...</execute> block that has output
        if "<output>" not in response.lower():
            return False  # Tool call pending, not complete

    # First, quick check - if no </answer> at all, definitely incomplete
    if "</answer>" not in response.lower():
        return False

    # If no <answer> at all, definitely incomplete
    if "<answer>" not in response.lower():
        return False

    # Find real answer tags (not just mentions in natural language)
    answer_opens = find_real_answer_tags(response, "open")
    answer_closes = find_real_answer_tags(response, "close")

    if not answer_opens or not answer_closes:
        return False

    # For each answer open, check if it has a valid close
    # A valid close is one that:
    # 1. Comes after the open
    # 2. Is at the top level (not inside a thinking/reflection that started after the answer open)

    for answer_open_pos in answer_opens:
        # Get text before this answer tag to check if it's inside thinking/reflection
        text_before = response[:answer_open_pos]

        # Count open/close tags before this position
        thinking_depth = len(_RE_THINKING_OPEN.findall(text_before)) - len(
            _RE_THINKING_CLOSE.findall(text_before)
        )
        reflection_depth = len(_RE_REFLECTION_OPEN.findall(text_before)) - len(
            _RE_REFLECTION_CLOSE.findall(text_before)
        )

        # If this answer open is inside thinking/reflection, skip it
        if thinking_depth > 0 or reflection_depth > 0:
            continue

        # This is a top-level answer open - now find its matching close
        # The close should be at the same nesting level
        # We need to track answer nesting within the answer block
        text_after_open = response[answer_open_pos + len("<answer>") :]

        # Track nesting - we start inside the answer (depth 1)
        answer_depth = 1
        pos = 0

        while pos < len(text_after_open):
            # Look for next tag
            next_open = text_after_open.find("<answer>", pos)
            next_close = text_after_open.find("</answer>", pos)

            # Case insensitive search
            next_open_lower = text_after_open.lower().find("<answer>", pos)
            next_close_lower = text_after_open.lower().find("</answer>", pos)

            if next_open_lower == -1:
                next_open = float("inf")
            else:
                next_open = next_open_lower
            if next_close_lower == -1:
                next_close = float("inf")
            else:
                next_close = next_close_lower

            if next_open == float("inf") and next_close == float("inf"):
                # No more answer tags found
                break

            if next_open < next_close:
                # Found nested answer open
                answer_depth += 1
                pos = next_open + len("<answer>")
            else:
                # Found answer close
                answer_depth -= 1
                if answer_depth == 0:
                    # This is the matching close for our top-level answer
                    # Extract the answer content and check if it has meaningful text
                    answer_content = text_after_open[:next_close]

                    # Clean out step/reward/count/thinking/reflection tags
                    cleaned_content = _RE_STEP_TAG.sub(
                        "",
                        answer_content,
                    )
                    cleaned_content = _RE_REWARD_TAG.sub(
                        "",
                        cleaned_content,
                    )
                    cleaned_content = _RE_COUNT_TAG.sub(
                        "",
                        cleaned_content,
                    )
                    cleaned_content = _RE_THINKING_BLOCK.sub(
                        "",
                        cleaned_content,
                    )
                    cleaned_content = _RE_REFLECTION_BLOCK.sub(
                        "",
                        cleaned_content,
                    )
                    cleaned_content = cleaned_content.strip()

                    # Only consider it complete if there's actual content
                    if cleaned_content:
                        return True
                    else:
                        # Answer only contained step/thinking tags, not a real answer
                        continue
                pos = next_close + len("</answer>")

    return False


def is_inside_top_level_answer(response: str, position: int = None) -> bool:
    """
    Check if the current position (or end of string) is inside a top-level <answer> block.

    This handles cases where <thinking> appears inside <answer>:
    - <answer>Text <thinking>thought</thinking> more</answer> - position after <thinking> IS inside answer
    - <thinking>thoughts</thinking>
      <answer>text - position at end IS inside answer
    - <thinking><answer>text</answer></thinking> - the answer is NOT top-level
    - <thinking>I'll put response in the <answer> block</thinking> - NOT an answer (just mentioned)

    Args:
        response: The full response text
        position: The position to check (default: end of string)

    Returns:
        bool: True if the position is inside a top-level answer block
    """
    if position is None:
        position = len(response)

    text_to_check = response[:position]

    # Find all top-level answer opens before this position (using real tag detection)
    answer_open_positions = []
    for match in _RE_ANSWER_OPEN.finditer(text_to_check):
        open_pos = match.start()

        # First check if this is a real tag (not just mentioned in text)
        if not is_real_answer_tag(response, open_pos):
            continue

        # Check if this answer open is at top level (not inside thinking/reflection)
        text_before = text_to_check[:open_pos]
        thinking_depth = len(_RE_THINKING_OPEN.findall(text_before)) - len(
            _RE_THINKING_CLOSE.findall(text_before)
        )
        reflection_depth = len(_RE_REFLECTION_OPEN.findall(text_before)) - len(
            _RE_REFLECTION_CLOSE.findall(text_before)
        )
        if thinking_depth == 0 and reflection_depth == 0:
            answer_open_positions.append(open_pos)

    if not answer_open_positions:
        return False

    # For each top-level answer open, check if it's been closed before our position
    for answer_open_pos in answer_open_positions:
        # Look for the matching close after this open but before our position
        text_after_open = text_to_check[answer_open_pos + len("<answer>") :]

        # Count answer opens and closes to find the matching close
        answer_depth = 1
        pos = 0
        found_close = False

        while pos < len(text_after_open):
            next_open_lower = text_after_open.lower().find("<answer>", pos)
            next_close_lower = text_after_open.lower().find("</answer>", pos)

            next_open = float("inf") if next_open_lower == -1 else next_open_lower
            next_close = float("inf") if next_close_lower == -1 else next_close_lower

            if next_open == float("inf") and next_close == float("inf"):
                break
            if next_open < next_close:
                answer_depth += 1
                pos = next_open + len("<answer>")
            else:
                answer_depth -= 1
                if answer_depth == 0:
                    found_close = True
                    break
                pos = next_close + len("</answer>")

        if not found_close:
            # This top-level answer hasn't been closed yet - we're inside it
            return True

    return False


class Interactions:
    def __init__(
        self,
        agent_name: str = "",
        user=DEFAULT_USER,
        ApiClient=None,
        collection_id: str = "0",
    ):
        self.ApiClient = ApiClient
        self.user = user
        self.uri = getenv("AGIXT_URI")
        if agent_name != "":
            self.agent_name = agent_name
            self.agent = Agent(self.agent_name, user=user, ApiClient=self.ApiClient)
            # Reuse auth from Agent to avoid redundant JWT encode/decode
            self.auth = self.agent.auth
            self.user_id = self.agent.user_id
            self.websearch = Websearch(
                collection_number=collection_id,
                agent=self.agent,
                user=self.user,
                ApiClient=self.ApiClient,
            )
            self.agent_memory = Memories(
                agent_name=self.agent_name,
                agent_config=self.agent.AGENT_CONFIG,
                collection_number="0",
                ApiClient=self.ApiClient,
                user=self.user,
            )
            self.outputs = f"{self.uri}/outputs/{self.agent.agent_id}"
        else:
            self.agent_name = ""
            self.agent = None
            # Fallback: create auth when no agent
            self.auth = MagicalAuth(token=impersonate_user(email=self.user))
            self.user_id = self.auth.user_id
            self.websearch = None
            self.agent_memory = None
            self.outputs = f"{self.uri}/outputs"
        self.response = ""
        self.failures = 0
        self.chain = Chain(user=user)
        self.cp = Prompts(user=user)
        self._processed_commands = set()
        # Cumulative count of commands skipped because they were already-executed.
        # Used by run_stream to detect a model stuck regenerating commands across
        # many iterations even when one new variant slips through each round.
        self._total_dedup_count = 0
        self._last_iteration_dedup_count = 0

    def _check_cancelled(self):
        """
        Check if the current asyncio task has been cancelled.
        Raises asyncio.CancelledError if the task was cancelled.
        This allows graceful stopping of long-running operations.
        """
        task = asyncio.current_task()
        if task and task.cancelled():
            raise asyncio.CancelledError("Task was cancelled by user")

    # Directories that are noisy / huge and should be collapsed in the
    # workspace file tree shown to the agent. The directory itself is still
    # listed (so the agent knows it exists) but its contents are replaced
    # with a "..." marker. The agent can use shell commands to inspect them
    # if it really needs to.
    _COLLAPSED_TREE_DIRS = frozenset(
        {
            "node_modules",
            ".git",
            ".venv",
            "venv",
            "env",
            ".env",
            "__pycache__",
            ".mypy_cache",
            ".pytest_cache",
            ".ruff_cache",
            ".tox",
            ".next",
            ".nuxt",
            ".cache",
            ".turbo",
            ".parcel-cache",
            "dist",
            "build",
            "out",
            "target",
            ".gradle",
            ".idea",
            ".vscode",
            ".DS_Store",
            "coverage",
            ".nyc_output",
            "vendor",
            "bower_components",
            ".terraform",
            ".serverless",
            "site-packages",
            "Pods",
            "DerivedData",
        }
    )

    # Hard cap on number of nodes (files + dirs) rendered in the tree.
    # Prevents pathological workspaces from blowing up context.
    _FILE_TREE_MAX_NODES = 500

    @staticmethod
    def _build_file_tree(file_paths: list) -> str:
        """Build an indented file tree string from a list of relative file paths.

        Collapses noisy directories (node_modules, .git, .venv, etc.) to a
        "..." marker and truncates trees larger than ``_FILE_TREE_MAX_NODES``
        nodes so it can never blow out the context window.

        Example output:
          ├── report.csv
          ├── data/
          │   ├── input.xlsx
          │   └── output.json
          ├── node_modules/
          │   └── ... (collapsed, use terminal to inspect)
          └── notes.txt
        """
        if not file_paths:
            return "(empty)"
        collapsed = Interactions._COLLAPSED_TREE_DIRS
        max_nodes = Interactions._FILE_TREE_MAX_NODES

        # Build a nested dict representing the directory structure.
        # When we encounter a path component whose name is in the collapsed
        # set, we mark that node as collapsed and stop descending.
        tree = {}
        collapsed_dirs = set()  # ids of dict nodes that are collapsed
        for path in sorted(file_paths):
            parts = path.replace("\\", "/").split("/")
            node = tree
            for idx, part in enumerate(parts):
                if part not in node:
                    node[part] = {}
                child = node[part]
                # If this component is a noisy dir AND it's not the leaf file,
                # mark it collapsed and skip its descendants.
                is_dir = idx < len(parts) - 1
                if is_dir and part in collapsed:
                    collapsed_dirs.add(id(child))
                    break
                node = child

        lines = []
        node_count = [0]
        truncated = [False]

        def _render(node, prefix=""):
            entries = sorted(node.keys(), key=lambda k: (not bool(node[k]), k.lower()))
            for i, name in enumerate(entries):
                if node_count[0] >= max_nodes:
                    truncated[0] = True
                    return
                is_last = i == len(entries) - 1
                connector = "└── " if is_last else "├── "
                child = node[name]
                if child or id(child) in collapsed_dirs:  # directory
                    lines.append(f"{prefix}{connector}{name}/")
                    node_count[0] += 1
                    extension = "    " if is_last else "│   "
                    if id(child) in collapsed_dirs:
                        lines.append(
                            f"{prefix}{extension}└── ... (collapsed, use terminal to inspect)"
                        )
                        node_count[0] += 1
                    else:
                        _render(child, prefix + extension)
                else:  # file
                    lines.append(f"{prefix}{connector}{name}")
                    node_count[0] += 1

        _render(tree)
        if truncated[0]:
            lines.append(
                f"... (tree truncated at {max_nodes} entries, use terminal to inspect more)"
            )
        return "\n".join(lines)

    def custom_format(self, string, **kwargs):
        if "fp" in kwargs:
            return kwargs["user_input"]
        if isinstance(string, list):
            string = "".join(str(x) for x in string)

        def replace(match):
            key = match.group(1)
            value = kwargs.get(key, match.group(0))
            if isinstance(value, list):
                return "".join(str(x) for x in value)
            else:
                return str(value)

        result = _RE_CUSTOM_FORMAT.sub(replace, string)
        return result

    async def format_prompt(
        self,
        user_input: str = "",
        top_results: int = 5,
        prompt="",
        conversation_name="",
        vision_response: str = "",
        selected_commands: list = None,
        max_context_tokens: int = None,
        **kwargs,
    ):
        # Use agent's max_input_tokens as default for context limit
        # Reserve some tokens for the response (about 25% of max)
        if max_context_tokens is None:
            max_context_tokens = int(self.agent.max_input_tokens * 0.75)
        # Apply a hard ceiling so we never blow past the model server's real
        # context window (e.g. ezlocalai n_ctx). Even when an agent is
        # configured with a very large MAX_TOKENS, model coherence degrades
        # well before 1M tokens, so we cap the *input* portion at 200k by
        # default. Configurable via MAX_CONTEXT_TOKENS_HARD_CAP env var.
        try:
            hard_cap = int(getenv("MAX_CONTEXT_TOKENS_HARD_CAP", "200000"))
        except (TypeError, ValueError):
            hard_cap = 200000
        if hard_cap > 0 and max_context_tokens > hard_cap:
            max_context_tokens = hard_cap
        if "user_input" in kwargs and user_input == "":
            user_input = kwargs["user_input"]
        prompt_name = prompt if prompt != "" else "Custom Input"
        prompt_category = (
            "Default" if "prompt_category" not in kwargs else kwargs["prompt_category"]
        )
        try:
            prompt = self.cp.get_prompt(
                prompt_name=prompt_name, prompt_category=prompt_category
            )
            prompt_args = self.cp.get_prompt_args(prompt_text=prompt)
        except Exception as e:
            logging.error(
                f"Error: {self.agent_name} failed to get prompt {prompt_name} from prompt category {prompt_category}. {e}"
            )
            prompt = prompt_name
            prompt_args = []
        if "conversation_name" in kwargs:
            conversation_name = kwargs["conversation_name"]
        if conversation_name == "":
            conversation_name = "-"
        # Use conversation_id if provided - it's more stable than name during renames
        conversation_id = kwargs.get("conversation_id")
        c = Conversations(
            conversation_name=conversation_name,
            user=self.user,
            conversation_id=conversation_id,
        )
        conversation_id = c.get_conversation_id()
        conversation_outputs = (
            f"http://localhost:7437/outputs/{self.agent.agent_id}/{conversation_id}/"
        )
        context = []
        if int(top_results) > 0:
            if user_input:
                min_relevance_score = 0.2
                if "min_relevance_score" in kwargs:
                    try:
                        min_relevance_score = float(kwargs["min_relevance_score"])
                    except:
                        min_relevance_score = 0.2
                context += await self.agent_memory.get_memories(
                    user_input=user_input,
                    limit=top_results,
                    min_relevance_score=min_relevance_score,
                )
                # Default to injecting from collection 0 if no specific collection is specified
                # This provides additional memories from collection 0 beyond what agent_memory provides
                collection_id = kwargs.get(
                    "inject_memories_from_collection_number", "0"
                )

                # Always inject additional memories from the specified collection
                # Even if it's collection 0, as this may provide different or additional results
                try:
                    additional_memories = await Memories(
                        agent_name=self.agent_name,
                        agent_config=self.agent.AGENT_CONFIG,
                        collection_number=collection_id,
                        ApiClient=self.ApiClient,
                        user=self.user,
                    ).get_memories(
                        user_input=user_input,
                        limit=top_results,
                        min_relevance_score=min_relevance_score,
                    )
                    # Only add if we got different memories to avoid complete duplicates
                    if additional_memories:
                        context += additional_memories
                except Exception as e:
                    logging.error(
                        f"Error: {self.agent_name} failed to get memories from collection {collection_id}. {e}"
                    )
                conversation_context = await self.websearch.agent_memory.get_memories(
                    user_input=user_input,
                    limit=top_results,
                    min_relevance_score=min_relevance_score,
                )
                if len(conversation_context) == int(top_results):
                    conversational_context_tokens = get_tokens(
                        " ".join(conversation_context)
                    )
                    if int(conversational_context_tokens) < 4000:
                        conversational_results = top_results * 2
                        conversation_context = (
                            await self.websearch.agent_memory.get_memories(
                                user_input=user_input,
                                limit=conversational_results,
                                min_relevance_score=min_relevance_score,
                            )
                        )
                        conversational_context_tokens = get_tokens(
                            " ".join(conversation_context)
                        )
                        if int(conversational_context_tokens) < 4000:
                            conversational_results = conversational_results * 2
                            conversation_context = (
                                await self.websearch.agent_memory.get_memories(
                                    user_input=user_input,
                                    limit=conversational_results,
                                    min_relevance_score=min_relevance_score,
                                )
                            )
                context += conversation_context
        if "context" in kwargs:
            context.append(kwargs["context"])
        include_sources = (
            str(kwargs["include_sources"]).lower() == "true"
            if "include_sources" in kwargs
            else False
        )
        if include_sources:
            sources = []
            for line in context:
                if "Content from" in line:
                    source = line.split("Content from ")[1].split("\n")[0]
                    if f"Content from {source}" not in sources:
                        sources.append(f"Content from {source}")
            if sources != []:
                joined_sources = "\n".join(sources)
                thinking_id = c.get_thinking_id(agent_name=self.agent_name)
                source_count = len(sources)
                c.log_interaction(
                    role=self.agent_name,
                    message=f"[SUBACTIVITY][{thinking_id}] Referencing {source_count} sources from content.\n{joined_sources}.",
                )
        working_directory = f"{self.agent.working_directory}/{conversation_id}"
        helper_agent_name = self.agent_name
        if "helper_agent_name" not in kwargs:
            if "helper_agent_name" in self.agent.AGENT_CONFIG["settings"]:
                helper_agent_name = self.agent.AGENT_CONFIG["settings"][
                    "helper_agent_name"
                ]
        if "conversation_results" in kwargs:
            try:
                conversation_results = int(kwargs["conversation_results"])
            except:
                conversation_results = 5
        else:
            try:
                conversation_results = int(top_results) if top_results > 0 else 5
            except:
                conversation_results = 5
        agent_tasks = self.agent.get_conversation_tasks(conversation_id=conversation_id)
        if agent_tasks != "":
            context.append(agent_tasks)
        conversation_history = ""
        # Fetch a wider window than the default to avoid silently dropping
        # relevant turns in longer chats before we apply our own slicing.
        history_fetch_limit = max(200, min(2000, conversation_results * 20))
        conversation = c.get_conversation(limit=history_fetch_limit, page=1)
        if "interactions" in conversation:
            if conversation["interactions"] != []:
                activity_history = [
                    interaction
                    for interaction in conversation["interactions"]
                    if str(interaction["message"]).startswith("[ACTIVITY]")
                ]
                activities = []
                for activity in activity_history:
                    if "audio response" not in activity["message"]:
                        activities.append(activity)
                if len(activity_history) > 5:
                    activity_history = activity_history[-5:]
                interactions = []
                for interaction in conversation["interactions"]:
                    if (
                        not str(interaction["message"]).startswith("<audio controls>")
                        and not str(interaction["message"]).startswith("[ACTIVITY]")
                        and not str(interaction["message"]).startswith("[SUBACTIVITY]")
                    ):
                        raw_timestamp = (
                            interaction["timestamp"]
                            if "timestamp" in interaction
                            else None
                        )
                        # Format timestamp in user's timezone for readability
                        # Timestamps are stored in UTC, so convert to user's local time
                        if raw_timestamp:
                            if hasattr(raw_timestamp, "strftime"):
                                # Convert from UTC to user's timezone
                                timestamp = convert_time(
                                    raw_timestamp, user_id=self.user_id
                                ).strftime("%B %d, %Y %I:%M %p")
                            else:
                                timestamp = str(raw_timestamp)
                        else:
                            timestamp = ""
                        role = interaction["role"] if "role" in interaction else ""
                        message = (
                            interaction["message"] if "message" in interaction else ""
                        )
                        message = regex.sub(r"(```.*?```)", "", message)
                        interactions.append(f"{timestamp} {role}: {message} \n ")
                if len(interactions) > 0:
                    interactions = interactions[-conversation_results:]
                    conversation_history = "\n".join(interactions)
                conversation_history += "\n## The assistant's recent activities:\n"
                activity_window = max(12, min(80, conversation_results * 4))
                subactivity_window = max(6, min(20, conversation_results * 2))
                conversation_history += c.get_activities_with_subactivities(
                    max_activities=activity_window,
                    max_subactivities_per_activity=subactivity_window,
                )
        if conversation_history != "":
            context.append(
                f"### Recent Activities and Conversation History\n{conversation_history}\n"
            )
        # Inject conversation summary for long-term memory / alignment
        try:
            conversation_summary = c.get_conversation_summary()
            if conversation_summary:
                context.append(
                    f"### Conversation Summary\nThe following is a living summary of this conversation capturing key topics, user preferences, lessons learned, and important context:\n{conversation_summary}\n"
                )
        except Exception as e:
            log_silenced_exception(e, "format_prompt: getting conversation summary")
        # Inject persistent user knowledge profile
        try:
            user_id = getattr(self, "user_id", "")
            if user_id:
                user_knowledge = get_user_knowledge(user_id)
                if user_knowledge:
                    context.append(
                        f"### User Knowledge\nThe following are observations about this user gathered over time across conversations. Use this to personalize responses and remember important details about them:\n{user_knowledge}\n"
                    )
        except Exception as e:
            log_silenced_exception(e, "format_prompt: getting user knowledge")
        persona = ""
        if "PERSONA" in self.agent.AGENT_CONFIG["settings"]:
            persona = self.agent.AGENT_CONFIG["settings"]["PERSONA"]
        if "persona" in self.agent.AGENT_CONFIG["settings"]:
            persona = self.agent.AGENT_CONFIG["settings"]["persona"]
        try:
            company_id = self.auth.company_id
            if "company_id" in kwargs:
                company_id = kwargs["company_id"]
            if company_id:
                company_training = self.auth.get_training_data(company_id=company_id)
                persona += f"\n\n**Guidelines as they pertain to the company:**\n{company_training}"
                # Get company memories using the agent's company_agent if available
                try:
                    company_agent = self.agent.company_agent if self.agent else None
                    if company_agent:
                        company_memories_obj = Memories(
                            agent_name=company_agent.agent_name,
                            agent_config=company_agent.AGENT_CONFIG,
                            collection_number="0",
                            ApiClient=self.ApiClient,
                            user=self.user,
                        )
                        company_memories = await company_memories_obj.get_memories(
                            user_input=user_input,
                            limit=5,
                            min_relevance_score=0.3,
                        )
                        if company_memories:
                            for metadata in company_memories:
                                if metadata not in context and metadata != "":
                                    context.append(metadata)
                except Exception as mem_e:
                    log_silenced_exception(
                        mem_e, "format_prompt: getting company agent memories"
                    )
        except Exception as e:
            log_silenced_exception(e, "format_prompt: getting company training data")
        context.append(self.auth.get_markdown_companies())
        if persona != "":
            context.append(
                f"## Persona\n**The assistant follows a persona and uses the following guidelines and information to remain in character.**\n{persona}\nThe assistant is {self.agent_name} and is an AGiXT agent created by DevXT, empowered with AGiXT abilities."
            )
        APP_URI = getenv("APP_URI")
        if "localhost:" not in APP_URI:
            context.append(
                f"The assistant is an AGiXT agent named `{self.agent_name}` running on {APP_URI}. The assistant can access the documentation about the website at {AGIXT_URI}/docs as well as information about the open source AGiXT back end repository at https://github.com/Josh-XT/AGiXT if necessary."
            )
        if "72" in kwargs and "42" in kwargs:
            if kwargs["72"] == True and kwargs["42"] == True:
                kwargs["fp"] = context
        if "uploaded_file_data" in kwargs:
            context.append(
                f"The user uploaded these files for the assistant to analyze:\n{kwargs['uploaded_file_data']}\n"
            )
        # Always include workspace file tree so the agent knows what files exist
        # This applies to both initial uploads AND follow-up requests
        if "workspace_file_context" in kwargs and kwargs["workspace_file_context"]:
            workspace_file_info = kwargs["workspace_file_context"]
            context.append(
                f"## Files in Workspace\nThe following files are available in the assistant's workspace directory and can be accessed using file operation commands (Read File, Write to File, Modify File, Delete File, List Directory, Search Files, Search File Content, Grep Search, Execute Python File, Run Data Analysis, Execute Python Code):\n{workspace_file_info}\n\n**IMPORTANT: When using file commands, use ONLY the relative path from the tree above.** For example, if the tree shows `CandleLaunchGame/src/App.tsx`, use `CandleLaunchGame/src/App.tsx` as the filename — do NOT prepend `/agixt/`, `/workspace/`, `WORKSPACE/`, or any absolute path. The system automatically resolves relative paths to the correct location.\n"
            )
        if vision_response != "":
            context.append(
                f"The assistant's visual description from viewing uploaded images by user in this interaction:\n{vision_response}\n"
            )
        if "data_analysis" in kwargs:
            context.append(
                f"The assistant's data analysis from the user's input and file uploads:\n{kwargs['data_analysis']}\n"
            )
        if context != [] and context != "":
            if isinstance(context, list):
                context = "\n".join(context)
            context = f"The user's input causes the assistant to recall these memories from activities:\n{context}\n\n**If referencing a file or image from context to the user, link to it with a url at `{conversation_outputs}the_file_name` - The URL is accessible to the user. If the file has not been referenced in context or from activities, do not attempt to link to it as it may not exist. Use exact file names and links from context only.** If linking an image, use the format `![alt_text](URL). The assistant can render HTML in chat including using javascript and threejs just by using an HTML code block.`\n"
        else:
            context = ""
        file_contents = ""
        if "import_files" in prompt_args:
            # import_files should be formatted like [{"file_name": "file_content"}]
            files = []
            if "import_files" in kwargs:
                if kwargs["import_files"] != "":
                    try:
                        files = json.loads(kwargs["import_files"])
                    except:
                        files = []
            all_files_content = ""
            file_list = []
            for file in files:
                file_name = file["file_name"]
                file_list.append(file_name)
                file_name = regex.sub(r"(\[.*?\])", "", file_name)
                file_path = os.path.normpath(os.path.join(working_directory, file_name))
                if not file_path.startswith(os.path.normpath(working_directory)):
                    logging.warning(f"Path traversal attempt blocked: {file_name}")
                    continue
                if not os.path.exists(file_path):
                    # Create it with the content if it doesn't exist.
                    with open(file_path, "w") as f:
                        f.write(file["file_content"])
                    file_content = file["file_content"]
                else:
                    with open(file_path, "r") as f:
                        file_content = f.read()
                    file_contents += f"\n`{file_path}` content:\n{file_content}\n\n"
                if file_name != "" and file_content != "":
                    all_files_content += file_content
            if files != []:
                the_files = (
                    f"these files: {', '.join(file_list)}."
                    if len(file_list) > 1
                    else f"the file {file_list[0]}."
                )
                c.log_interaction(
                    role=self.agent_name,
                    message=f"I have read the file contents of {the_files}.",
                )
            else:
                the_files = "files."
            tokens_used = get_tokens(
                f"{prompt}{user_input}{all_files_content}{context}"
            )
            agent_max_tokens = int(
                self.agent.AGENT_CONFIG["settings"]["MAX_TOKENS"]
                if "MAX_TOKENS" in self.agent.AGENT_CONFIG["settings"]
                else 8192
            )
            if tokens_used > agent_max_tokens or files == []:
                memories = Memories(
                    agent_name=self.agent_name,
                    agent_config=self.agent.AGENT_CONFIG,
                    collection_number="1",
                    ApiClient=self.ApiClient,
                    user=self.user,
                )
                fragmented_content = await memories.get_memories(
                    user_input=f"{user_input} {file_list}",
                    min_relevance_score=0.3,
                    limit=top_results if top_results > 0 else 5,
                )
                if fragmented_content != "":
                    file_contents = f"Here is some potentially relevant information from {the_files}\n{fragmented_content}\n\n"
        skip_args = [
            "user_input",
            "agent_name",
            "COMMANDS",
            "context",
            "command_list",
            "date",
            "working_directory",
            "helper_agent_name",
            "conversation_history",
            "persona",
            "import_files",
            "output_url",
        ]
        args = kwargs.copy()
        for arg in kwargs:
            if arg in skip_args:
                del args[arg]
        agent_commands = ""
        if "disable_commands" not in kwargs:
            # Cache the commands prompt to avoid re-calling extension contexts
            # (which may make HTTP requests) on every continuation iteration.
            # The cache is invalidated when selected_commands changes (e.g. via
            # Optimize Command Selection).
            _cache_key = tuple(sorted(selected_commands)) if selected_commands else None
            if (
                hasattr(self, "_cached_commands_prompt")
                and self._cached_commands_prompt is not None
                and hasattr(self, "_cached_commands_key")
                and self._cached_commands_key == _cache_key
            ):
                agent_commands = self._cached_commands_prompt
            else:
                agent_commands = self.agent.get_commands_prompt(
                    conversation_id=conversation_id,
                    running_command=kwargs.get("running_command", None),
                    selected_commands=selected_commands,
                    workspace_file_tree=kwargs.get("workspace_file_context", None),
                )
                self._cached_commands_prompt = agent_commands
                self._cached_commands_key = _cache_key
            # Safety check: if commands prompt is too large, it will crowd out
            # user context, conversation history, and model thinking space.
            # Log a warning so we can identify bloated command sets.
            commands_token_count = get_tokens(agent_commands)
            if commands_token_count > 20000:
                logging.warning(
                    f"[format_prompt] Commands prompt is {commands_token_count} tokens "
                    f"({len(selected_commands) if selected_commands else 'all'} commands). "
                    f"This may exceed context budget."
                )

        # Check if context needs reduction before building final prompt
        # Estimate tokens from variable-size context components
        context_str = "\n".join(context) if isinstance(context, list) else str(context)
        estimated_context_tokens = get_tokens(
            f"{prompt}{user_input}{context_str}{conversation_history}{agent_commands}{file_contents}"
        )

        if estimated_context_tokens > max_context_tokens:
            # Build context sections dict for reduce_context
            context_sections = {
                "memories": context_str,  # Already retrieved memories as string
                "conversation_history": conversation_history,
                "file_contents": file_contents,
            }

            # Reduce context using intelligent selection
            reduced = await self.reduce_context(
                user_input=user_input,
                context_sections=context_sections,
                target_tokens=max_context_tokens,
                conversation_name=conversation_name,
            )

            # Apply reduced context
            if "memories" in reduced:
                old_context_len = len(context_str)
                context = [reduced["memories"]] if reduced["memories"] else []
                new_context_len = len(reduced["memories"]) if reduced["memories"] else 0
            if "conversation_history" in reduced:
                conversation_history = reduced["conversation_history"]
            if "file_contents" in reduced:
                file_contents = reduced["file_contents"]

            new_tokens = get_tokens(
                f"{prompt}{user_input}{context_str}{conversation_history}{agent_commands}{file_contents}"
            )
        user_datetime = get_current_user_time(user_id=self.user_id).strftime(
            "%B %d, %Y %I:%M %p"
        )
        formatted_prompt = self.custom_format(
            string=prompt,
            user_input=user_input,
            agent_name=self.agent_name,
            COMMANDS=agent_commands,
            context=context,
            command_list=agent_commands,
            date=f"{user_datetime} (This and other timestamps are in the users local timezone)",
            working_directory=working_directory,
            helper_agent_name=helper_agent_name,
            conversation_history=conversation_history,
            persona=persona,
            import_files=file_contents,
            output_url=conversation_outputs,
            **args,
        )
        tokens = get_tokens(formatted_prompt)

        # Add context management guidance when approaching high token counts
        # Ideal context is under 32k tokens for best quality responses
        OPTIMAL_TOKEN_THRESHOLD = 30000
        HIGH_TOKEN_WARNING_THRESHOLD = 32000

        if tokens >= OPTIMAL_TOKEN_THRESHOLD:
            context_guidance = f"""

## ⚠️ Context Management Advisory
Current context size: **{tokens:,} tokens** (target: under 32,000 for optimal quality)

You have access to context management commands to reduce token usage:
- **Discard Context**: After reading files or noting important information, discard activities you no longer need in full detail. Provide the message_id and a brief reason (e.g., "noted key functions", "empty file", "not relevant to current task").
- **Retrieve Context**: If you need full details from something you previously discarded, retrieve it by message_id.
- **List Discarded Context**: See what you've discarded and their reasons.

**Guidelines:**
- Look for activities in your recent history that you've already processed and taken notes on
- File reads, searches, and informational outputs are good candidates for discarding once noted
- Keep conversation messages and important decision points
- Each discarded activity shows a brief summary instead of full content
"""
            if tokens >= HIGH_TOKEN_WARNING_THRESHOLD:
                context_guidance += f"""
**CRITICAL:** Context is at {tokens:,} tokens. Response quality may degrade. Consider discarding older activities to reduce context size.
"""
            formatted_prompt = formatted_prompt + context_guidance

        # Add TTS filler speech instructions if provided
        # This enables the model to use <speak> tags for voice feedback during thinking
        tts_filler = args.get("tts_filler_instructions", "")
        if tts_filler:
            formatted_prompt = formatted_prompt + "\n" + tts_filler

        return formatted_prompt, prompt, tokens

    def process_thinking_tags(
        self, response: str, thinking_id: str, c: Conversations
    ) -> str:
        """
        Process thinking and reflection tags in the response and log them as subactivities.
        Also processes <step>, <reward>, and <count> tags that appear outside of thinking tags.
        Only processes new, unprocessed tags.

        Args:
            response: The response text containing thinking/reflection tags
            thinking_id: ID for the thinking activity
            c: Conversation object for logging interactions

        Returns:
            Updated response with processed tags
        """
        # Pattern to match thinking and reflection tags and their content
        tag_pattern = (
            r"<(thinking|reflection)>(.*?)(?=<(?:thinking|reflection|answer)|$)"
        )

        matches = list(_RE_THINKING_REFLECTION.finditer(response))

        # Keep track of processed tags using content as key to avoid duplicates
        if not hasattr(self, "_processed_tags"):
            self._processed_tags = set()

        # Create a dict to store unique thoughts by their cleaned content
        unique_thoughts = {}

        for match in matches:
            tag_name = match.group(1)  # think, thinking, or reflection
            # Canonicalize "think" to "thinking" for internal processing
            if tag_name.lower() == "think":
                tag_name = "thinking"
            tag_content = match.group(2).strip()

            # Clean the content
            cleaned_content = tag_content
            cleaned_content = _RE_EXECUTE_TAG.sub("", cleaned_content)
            cleaned_content = _RE_OUTPUT_TAG.sub("", cleaned_content)
            cleaned_content = _RE_RATE_TAG.sub("", cleaned_content)
            cleaned_content = _RE_REWARD_TAG.sub("", cleaned_content)
            cleaned_content = _RE_COUNT_TAG.sub("", cleaned_content)
            cleaned_content = cleaned_content.replace("\n\\n", "\n")
            cleaned_content = cleaned_content.replace("</reflection>", "")
            cleaned_content = cleaned_content.replace("</thinking>", "")
            cleaned_content = cleaned_content.replace("</think>", "")
            cleaned_content = cleaned_content.replace("<output>", "")
            cleaned_content = cleaned_content.replace("</output>", "")
            cleaned_content = cleaned_content.replace("<step>", "")
            cleaned_content = cleaned_content.replace("</step>", "")
            cleaned_content = _RE_NAME_TAG.sub("", cleaned_content)
            cleaned_content = _RE_MULTI_NEWLINE.sub("\n\n", cleaned_content)
            cleaned_content = cleaned_content.strip()

            # Use cleaned content as key to prevent duplicates
            if cleaned_content and cleaned_content not in unique_thoughts:
                unique_thoughts[cleaned_content] = {
                    "tag_name": tag_name,
                    "content": cleaned_content,
                }

        # Process <step> tags that appear outside of thinking/reflection tags
        # These should be treated as thinking steps
        for step_match in _RE_STEP_TAG.finditer(response):
            step_content = step_match.group(1).strip()
            step_start = step_match.start()

            # Check if this step is inside a thinking or reflection tag
            is_inside_thinking = False
            for thinking_match in re.finditer(
                r"<(think(?:ing)?|reflection)>.*?</\1>",
                response,
                re.DOTALL | re.IGNORECASE,
            ):
                if thinking_match.start() < step_start < thinking_match.end():
                    is_inside_thinking = True
                    break

            # Also check for unclosed thinking tags
            if not is_inside_thinking:
                text_before = response[:step_start]
                thinking_opens = len(_RE_THINKING_OPEN.findall(text_before))
                thinking_closes = len(_RE_THINKING_CLOSE.findall(text_before))
                reflection_opens = len(_RE_REFLECTION_OPEN.findall(text_before))
                reflection_closes = len(_RE_REFLECTION_CLOSE.findall(text_before))
                if (
                    thinking_opens > thinking_closes
                    or reflection_opens > reflection_closes
                ):
                    is_inside_thinking = True

            # Only process if outside thinking tags and not inside answer
            if not is_inside_thinking:
                # Check if inside answer block
                answer_opens = len(_RE_ANSWER_OPEN.findall(response[:step_start]))
                answer_closes = len(_RE_ANSWER_CLOSE.findall(response[:step_start]))
                if answer_opens > answer_closes:
                    continue  # Skip steps inside answer blocks

                # Clean the step content
                cleaned_step = _RE_REWARD_TAG.sub("", step_content)
                cleaned_step = _RE_COUNT_TAG.sub("", cleaned_step)
                cleaned_step = _RE_MULTI_NEWLINE.sub("\n\n", cleaned_step).strip()

                if cleaned_step and cleaned_step not in unique_thoughts:
                    unique_thoughts[cleaned_step] = {
                        "tag_name": "step",
                        "content": cleaned_step,
                    }

        # Process standalone <reward> tags outside thinking/reflection (log as reflection score)
        for reward_match in _RE_REWARD_TAG.finditer(response):
            reward_content = reward_match.group(1).strip()
            reward_start = reward_match.start()

            # Check if inside thinking, reflection, or step tag
            is_inside_container = False
            for container_match in re.finditer(
                r"<(thinking|reflection|step)>.*?</\1>",
                response,
                re.DOTALL | re.IGNORECASE,
            ):
                if container_match.start() < reward_start < container_match.end():
                    is_inside_container = True
                    break

            # Check for unclosed tags
            if not is_inside_container:
                text_before = response[:reward_start]
                for tag in ["think(?:ing)?", "reflection", "step"]:
                    opens = len(re.findall(f"<{tag}>", text_before, re.IGNORECASE))
                    closes = len(re.findall(f"</{tag}>", text_before, re.IGNORECASE))
                    if opens > closes:
                        is_inside_container = True
                        break

            if not is_inside_container:
                # Check if inside answer block
                answer_opens = len(_RE_ANSWER_OPEN.findall(response[:reward_start]))
                answer_closes = len(_RE_ANSWER_CLOSE.findall(response[:reward_start]))
                if answer_opens > answer_closes:
                    continue

                reward_key = f"reward:{reward_content}"
                if reward_key not in unique_thoughts:
                    unique_thoughts[reward_key] = {
                        "tag_name": "reward",
                        "content": reward_content,
                    }

        # Combine all step tags into a single thought
        step_contents = []
        non_step_thoughts = {}
        for key, thought in unique_thoughts.items():
            if thought["tag_name"] == "step":
                step_contents.append(thought["content"])
            else:
                non_step_thoughts[key] = thought

        # If we have steps, combine them into a single thinking entry
        if step_contents:
            combined_steps = "\n".join(f"- {step}" for step in step_contents)
            non_step_thoughts["combined_steps"] = {
                "tag_name": "thinking",
                "content": combined_steps,
            }

        # Log only unique thoughts (now with combined steps)
        for thought in non_step_thoughts.values():
            tag_name = str(thought["tag_name"]).lower()
            content = thought["content"]
            content = re.sub(r"\. ", ".\n", content, count=1)
            if content.startswith("1."):
                # Remove the first 3 character of the string
                content = content[3:]
                content = re.sub(r"\. ", ".\n", content, count=1)

            # Create a unique identifier for this tag
            tag_identifier = f"{tag_name}:{content}"

            # Only log if we haven't seen this exact thought before
            if tag_identifier not in self._processed_tags:
                if tag_name == "thinking":
                    log_message = f"[SUBACTIVITY][{thinking_id}][THOUGHT] {content}"
                elif tag_name == "reflection":
                    log_message = f"[SUBACTIVITY][{thinking_id}][REFLECTION] {content}"
                elif tag_name == "step":
                    log_message = f"[SUBACTIVITY][{thinking_id}][STEP] {content}"
                elif tag_name == "reward":
                    log_message = f"[SUBACTIVITY][{thinking_id}][SCORE] {content}"
                else:
                    log_message = f"[SUBACTIVITY][{thinking_id}] {content}"
                c.log_interaction(role=self.agent_name, message=log_message)
                self._processed_tags.add(tag_identifier)

        return response

    def compress_response_for_continuation(
        self,
        response: str,
        max_output_lines: int = 80,
        max_thinking_chars: int = 2000,
    ) -> str:
        """
        Compress a response for continuation prompts to prevent context explosion
        while preserving critical execution context.

        CRITICAL: The most important thing to preserve is the <execute>...<output>
        pairs — these tell the model what commands were already run and their results.
        Without this context, the model will regenerate the same commands, which get
        silently deduplicated, resulting in "swallowed" activities.

        This function:
        1. Preserves ALL <execute>...<output> pairs intact (most important context)
        2. Compresses standalone <output> blocks (not paired with execute)
        3. Compresses verbose thinking/reflection blocks (keep enough for reasoning continuity)
        4. Preserves answer content intact
        5. Removes internal-only tags (<step>, <reward>, <count>)

        Args:
            response: The full response to compress
            max_output_lines: Maximum lines to keep per standalone output block
            max_thinking_chars: Maximum characters to keep per thinking block

        Returns:
            Compressed response suitable for continuation context
        """
        compressed = response

        # 1. FIRST: Extract and protect all <execute>...<output>...</output> pairs
        # These are the MOST CRITICAL context — they tell the model what was already done.
        # We replace them with placeholders, compress everything else, then restore them.
        execute_output_pairs = []
        pair_pattern = re.compile(
            r"(<execute>.*?</execute>\s*<output>.*?</output>)",
            re.DOTALL | re.IGNORECASE,
        )
        for i, match in enumerate(pair_pattern.finditer(compressed)):
            placeholder = f"__EXEC_OUTPUT_PAIR_{i}__"
            execute_output_pairs.append((placeholder, match.group(1)))

        # Replace pairs with placeholders (reverse order to preserve positions)
        for placeholder, pair_content in reversed(execute_output_pairs):
            compressed = compressed.replace(pair_content, placeholder, 1)

        # 2. Compress STANDALONE <output> blocks (not paired with execute)
        def compress_output(match):
            content = match.group(1).strip()
            lines = content.split("\n")

            if len(lines) <= max_output_lines:
                return match.group(0)  # Keep as-is if short enough

            # Keep first portion and last portion with a summary in between
            keep_start = max_output_lines * 2 // 3
            keep_end = max_output_lines // 3

            compressed_lines = lines[:keep_start]
            compressed_lines.append(
                f"\n... [{len(lines) - max_output_lines} lines omitted for brevity] ...\n"
            )
            compressed_lines.extend(lines[-keep_end:])

            return f"<output>{chr(10).join(compressed_lines)}</output>"

        compressed = _RE_OUTPUT_BLOCK.sub(compress_output, compressed)

        # 3. Compress <thinking>/<think> blocks — keep enough for reasoning continuity
        # but remove verbose internal deliberation

        def compress_thinking(match):
            content = match.group(1).strip()

            if len(content) <= max_thinking_chars:
                return match.group(0)

            # Keep first part and note that it was truncated
            truncated = content[:max_thinking_chars]
            # Try to cut at a sentence boundary
            last_period = truncated.rfind(".")
            if last_period > max_thinking_chars * 0.7:
                truncated = truncated[: last_period + 1]

            # Preserve the model's native tag form (e.g., <think> vs <thinking>)
            original_open = match.group(0)[: match.group(0).index(">") + 1]
            # Derive closing tag from opening tag
            original_close = original_open.replace("<", "</")
            return f"{original_open}{truncated} [thinking truncated]{original_close}"

        compressed = _RE_THINKING_CONTENT.sub(
            compress_thinking,
            compressed,
        )

        # 4. Compress <reflection> blocks similarly

        def compress_reflection(match):
            content = match.group(1).strip()

            if len(content) <= max_thinking_chars:
                return match.group(0)

            truncated = content[:max_thinking_chars]
            last_period = truncated.rfind(".")
            if last_period > max_thinking_chars * 0.7:
                truncated = truncated[: last_period + 1]

            return f"<reflection>{truncated} [reflection truncated]</reflection>"

        compressed = _RE_REFLECTION_CONTENT.sub(
            compress_reflection,
            compressed,
        )

        # 5. Remove <step> tags entirely from continuation context - they're internal
        compressed = _RE_STEP_TAG.sub("", compressed)

        # 6. Remove <reward> and <count> tags
        compressed = _RE_REWARD_TAG.sub("", compressed)
        compressed = _RE_COUNT_TAG.sub("", compressed)

        # 7. Restore protected execute+output pairs
        for placeholder, pair_content in execute_output_pairs:
            compressed = compressed.replace(placeholder, pair_content)

        # Now compress the output blocks WITHIN the restored pairs more gently
        # (keep more lines since these are critical execution results)
        exec_output_max_lines = 50  # More generous for execution outputs

        def compress_exec_output(match):
            content = match.group(1).strip()
            lines = content.split("\n")

            if len(lines) <= exec_output_max_lines:
                return match.group(0)

            keep_start = exec_output_max_lines * 2 // 3
            keep_end = exec_output_max_lines // 3

            compressed_lines = lines[:keep_start]
            compressed_lines.append(
                f"\n... [{len(lines) - exec_output_max_lines} lines omitted] ...\n"
            )
            compressed_lines.extend(lines[-keep_end:])

            return f"<output>{chr(10).join(compressed_lines)}</output>"

        compressed = _RE_OUTPUT_BLOCK.sub(compress_exec_output, compressed)

        # Clean up any excessive whitespace left behind
        compressed = _RE_TRIPLE_NEWLINE.sub("\n\n", compressed)

        return compressed.strip()

    def deduplicate_repeated_content(
        self, content: str, min_paragraph_length: int = 50
    ) -> str:
        """
        Detect and remove repeated content blocks from the model's response.

        Sometimes models get stuck in a loop and repeat the same content multiple times.
        This function detects such patterns and keeps only the first occurrence.

        Uses paragraph-based matching which is more robust than substring matching,
        as it handles minor formatting differences between repetitions.

        Args:
            content: The content to deduplicate
            min_paragraph_length: Minimum length of a paragraph to consider for deduplication

        Returns:
            Deduplicated content with only the first occurrence of repeated blocks
        """
        if len(content) < min_paragraph_length * 2:
            return content  # Too short to have meaningful repetition

        # Split into paragraphs/sections (split on one or more blank lines)
        paragraphs = re.split(r"\n\s*\n", content)

        if len(paragraphs) < 2:
            return content  # Not enough paragraphs to have repetition

        # Track seen paragraphs (normalized for comparison)
        seen = {}

        for i, para in enumerate(paragraphs):
            # Normalize paragraph for comparison (strip whitespace, normalize internal spaces)
            normalized = " ".join(para.split())

            if len(normalized) < min_paragraph_length:
                continue  # Skip very short paragraphs

            if normalized in seen:
                # Found repetition - keep only paragraphs before this one
                result_paras = paragraphs[:i]
                result = "\n\n".join(result_paras).strip()

                # Log that we found and removed repetition
                logging.debug(
                    f"Deduplicated response: removed {len(paragraphs) - i} repeated paragraphs"
                )
                return result
            else:
                seen[normalized] = i

        return content

    async def reduce_context(
        self,
        user_input: str,
        context_sections: dict,
        target_tokens: int = 16000,
        conversation_name: str = "",
    ) -> dict:
        """
        Intelligently reduce context by having the LLM select which sections are needed.
        This runs multiple lightweight inferences to prune unnecessary context.

        Args:
            user_input: The user's current request
            context_sections: Dict of context sections with their content:
                - "memories": List of memory items
                - "activities": List of recent activities
                - "conversation": List of conversation messages
                - "files": List of file metadata
                - "persona": Persona string
                - "tasks": Current tasks
            target_tokens: Target token count to reduce to
            conversation_name: Name of conversation for logging

        Returns:
            dict: Reduced context sections
        """
        # Calculate current token counts per section
        section_tokens = {}
        total_tokens = 0
        for section_name, content in context_sections.items():
            if isinstance(content, list):
                section_text = "\n".join(str(item) for item in content)
            else:
                section_text = str(content) if content else ""
            tokens = get_tokens(section_text)
            section_tokens[section_name] = tokens
            total_tokens += tokens

        logging.info(
            f"[reduce_context] Total context tokens: {total_tokens}, target: {target_tokens}"
        )

        # If already under target, no reduction needed
        if total_tokens <= target_tokens:
            logging.info(
                f"[reduce_context] Context already under target, no reduction needed"
            )
            return context_sections

        # Build a summary of sections for the selector
        section_summaries = []
        for section_name, content in context_sections.items():
            tokens = section_tokens[section_name]
            if tokens == 0:
                continue

            # Create a brief summary/preview of each section
            if isinstance(content, list) and len(content) > 0:
                preview = (
                    str(content[0])[:200] + "..."
                    if len(str(content[0])) > 200
                    else str(content[0])
                )
                item_count = len(content)
                section_summaries.append(
                    f"**{section_name}** ({tokens} tokens, {item_count} items)\nPreview: {preview}"
                )
            elif content:
                preview = (
                    str(content)[:200] + "..."
                    if len(str(content)) > 200
                    else str(content)
                )
                section_summaries.append(
                    f"**{section_name}** ({tokens} tokens)\nPreview: {preview}"
                )

        # Step 1: Ask which sections are needed
        section_selection_prompt = f"""You are helping optimize context for an AI assistant. The user's request is:

"{user_input}"

The following context sections are available. Select which ones are ESSENTIAL for answering this request.
Total tokens: {total_tokens}, Target: {target_tokens} tokens.

## Available Sections:
{chr(10).join(section_summaries)}

## Instructions:
- Select ONLY sections that are directly relevant to the user's request
- The user's input is always included (not optional)
- Persona is usually needed for consistent responses
- Activities/conversation history may not be needed for simple questions
- Memories are useful for context but can be pruned for simple requests

Respond with ONLY a comma-separated list of section names to KEEP, or "all" if all are needed.
Example: memories, persona, files"""

        # Use dedicated ability selection server if configured (fast small model)
        ability_selection_server = getenv("ABILITY_SELECTION_SERVER")
        ability_selection_model = getenv(
            "ABILITY_SELECTION_MODEL", "unsloth/Qwen3.5-4B-GGUF"
        )

        try:
            if ability_selection_server:
                selection_response = await _ability_selection_inference(
                    server_url=ability_selection_server,
                    model=ability_selection_model,
                    prompt=section_selection_prompt,
                )
            else:
                selection_response = await stream_inference_to_string(
                    self.agent, prompt=section_selection_prompt
                )

            # Robust parser: strip code fences, markdown bold/italics, quotes,
            # backticks, and any leading "answer:"/"keep:" prefixes that small
            # models commonly produce. Without this, a single stray ``` or
            # whitespace can cause the keep-list to be empty, dropping ALL
            # context to 0 tokens (observed in production logs).
            raw = (selection_response or "").strip()
            # Remove fenced code blocks of any language
            raw = re.sub(r"```[a-zA-Z0-9_-]*\n?", "", raw)
            raw = raw.replace("```", "")
            # Remove common preambles
            raw = re.sub(
                r"^(?:answer|keep|sections?|response)\s*[:\-]\s*",
                "",
                raw,
                flags=re.IGNORECASE,
            )
            # Strip wrapping quotes/backticks/markdown emphasis
            raw = raw.strip().strip("`'\"*_ \n\t")

            valid_section_names = {k.lower() for k in context_sections.keys()}

            if raw.lower() in ("all", "*", "everything", "keep all"):
                sections_to_keep = list(context_sections.keys())
            else:
                # Split on commas/newlines/semicolons and clean each token
                candidates = [
                    re.sub(r"[`'\"*_]", "", s).strip().lower()
                    for s in re.split(r"[,\n;]+", raw)
                ]
                candidates = [c for c in candidates if c]
                # Keep only candidates that are actually valid section names
                sections_to_keep = [c for c in candidates if c in valid_section_names]
                # SAFETY NET: if parser produced nothing valid, the selector
                # output was malformed — keep ALL sections rather than drop
                # everything to 0 tokens. The summarization step below will
                # still bring us under target.
                if not sections_to_keep:
                    logging.warning(
                        f"[reduce_context] Selector returned unparseable response "
                        f"({selection_response!r}); keeping all sections and "
                        f"relying on summarization."
                    )
                    sections_to_keep = list(context_sections.keys())

            logging.info(f"[reduce_context] Sections to keep: {sections_to_keep}")

        except Exception as e:
            logging.error(f"[reduce_context] Error in section selection: {e}")
            sections_to_keep = list(context_sections.keys())

        # Step 2: Build reduced context, pruning unneeded sections
        reduced_context = {}
        reduced_tokens = 0

        for section_name, content in context_sections.items():
            if section_name.lower() in sections_to_keep:
                reduced_context[section_name] = content
                reduced_tokens += section_tokens[section_name]
            else:
                reduced_context[section_name] = [] if isinstance(content, list) else ""

        logging.info(f"[reduce_context] After section pruning: {reduced_tokens} tokens")

        # Step 3: If still over target, summarize large sections using the fast model
        # Inspired by Claude Code's block compaction: rather than simply dropping items,
        # compress older/less-relevant blocks into concise summaries that preserve
        # key facts while dramatically reducing token count.
        if reduced_tokens > target_tokens:
            for section_name in [
                "memories",
                "activities",
                "conversation",
                "conversation_history",
                "file_contents",
            ]:
                if section_name not in reduced_context:
                    continue
                content = reduced_context[section_name]
                section_text = (
                    "\n".join(str(item) for item in content)
                    if isinstance(content, list)
                    else str(content) if content else ""
                )
                section_tok = get_tokens(section_text)
                if section_tok < 2000:
                    continue  # Not worth summarizing small sections

                # Calculate how much this section needs to shrink
                overage = reduced_tokens - target_tokens
                target_section_tokens = max(1000, section_tok - overage)

                summary_prompt = f"""Summarize the following {section_name} context concisely for an AI assistant.
Preserve key facts, recent actions, decisions, and any information that would be needed to continue the current task.
Drop redundant details, verbose tool outputs, and repetitive entries.
Target approximately {target_section_tokens} tokens.

## Current User Request
{user_input[:300]}

## {section_name.replace('_', ' ').title()} to Summarize
{section_text[:80000]}

Respond with ONLY the condensed summary, no preamble."""

                try:
                    if ability_selection_server:
                        summary = await _ability_selection_inference(
                            server_url=ability_selection_server,
                            model=ability_selection_model,
                            prompt=summary_prompt,
                        )
                    else:
                        summary = await stream_inference_to_string(
                            self.agent, prompt=summary_prompt
                        )

                    if summary and len(summary) > 50:
                        new_tokens = get_tokens(summary)
                        # Quality guard: if the summarizer collapsed the
                        # section to under 25% of the requested target, the
                        # output is almost certainly lossy beyond usefulness
                        # (observed: 17,438 -> 422 tokens, 97.6% loss). Reject
                        # and fall through to bounded truncation instead so we
                        # preserve actionable detail (e.g. flux numbers,
                        # persona-driven analysis rules).
                        min_acceptable = max(500, int(target_section_tokens * 0.25))
                        if (
                            new_tokens < min_acceptable
                            and section_tok > min_acceptable * 4
                        ):
                            logging.warning(
                                f"[reduce_context] Rejecting over-aggressive "
                                f"summary for {section_name}: {section_tok} -> "
                                f"{new_tokens} tokens (target ~{target_section_tokens}, "
                                f"min acceptable {min_acceptable}). Falling back "
                                f"to bounded truncation."
                            )
                            # Force the fallback path below
                            summary = ""
                            new_tokens = section_tok
                        saved = section_tok - new_tokens
                        if summary and saved > 0:
                            reduced_context[section_name] = summary
                            reduced_tokens -= saved
                            logging.info(
                                f"[reduce_context] Summarized {section_name}: {section_tok} -> {new_tokens} tokens (saved {saved})"
                            )
                    if not summary or len(summary) <= 50:
                        # Summarization failed, fall back to truncation
                        if isinstance(content, list) and len(content) > 3:
                            if section_name in [
                                "activities",
                                "conversation",
                                "conversation_history",
                            ]:
                                reduced_context[section_name] = content[-5:]
                            else:
                                reduced_context[section_name] = content[:5]
                            new_tokens = get_tokens(
                                "\n".join(
                                    str(item) for item in reduced_context[section_name]
                                )
                            )
                            reduced_tokens -= section_tok - new_tokens
                except Exception as e:
                    logging.error(
                        f"[reduce_context] Error summarizing {section_name}: {e}"
                    )
                    # Fall back to simple truncation
                    if isinstance(content, list) and len(content) > 3:
                        if section_name in [
                            "activities",
                            "conversation",
                            "conversation_history",
                        ]:
                            reduced_context[section_name] = content[-5:]
                        else:
                            reduced_context[section_name] = content[:5]
                        new_tokens = get_tokens(
                            "\n".join(
                                str(item) for item in reduced_context[section_name]
                            )
                        )
                        reduced_tokens -= section_tok - new_tokens

                # Stop iterating sections if we're under target
                if reduced_tokens <= target_tokens:
                    break

        # Step 4: Hard-truncation safety net. If summarization could not bring
        # context under target (e.g. summarizer model failed, returned bloated
        # output, or sections are too large to summarize in one pass), force
        # the largest remaining sections down by raw token slicing. This
        # guarantees we never hand the inference server a payload larger than
        # its context window, which would otherwise produce a hard failure
        # like "request (X tokens) exceeds the available context size".
        if reduced_tokens > target_tokens:
            # Order: drop file_contents first, then conversation/activities,
            # then memories. Persona is preserved.
            truncation_order = [
                "file_contents",
                "activities",
                "conversation",
                "conversation_history",
                "memories",
            ]
            for section_name in truncation_order:
                if reduced_tokens <= target_tokens:
                    break
                if section_name not in reduced_context:
                    continue
                content = reduced_context[section_name]
                section_text = (
                    "\n".join(str(item) for item in content)
                    if isinstance(content, list)
                    else str(content) if content else ""
                )
                section_tok = get_tokens(section_text)
                if section_tok <= 0:
                    continue
                overage = reduced_tokens - target_tokens
                # Leave at least 500 tokens of this section if possible.
                keep_tokens = max(500, section_tok - overage)
                if keep_tokens >= section_tok:
                    continue
                # Approximate token->char ratio (~4 chars/token is typical for
                # English; we use the actual ratio for this section to be safe).
                ratio = len(section_text) / max(section_tok, 1)
                keep_chars = max(int(keep_tokens * ratio), 200)
                # Prefer to keep the *tail* for conversational sections (most
                # recent context), and the *head* for files/memories.
                if section_name in (
                    "conversation",
                    "conversation_history",
                    "activities",
                ):
                    truncated_text = (
                        "[... older entries truncated for context size ...]\n"
                        + section_text[-keep_chars:]
                    )
                else:
                    truncated_text = (
                        section_text[:keep_chars]
                        + "\n[... remainder truncated for context size ...]"
                    )
                new_tok = get_tokens(truncated_text)
                reduced_context[section_name] = truncated_text
                reduced_tokens -= section_tok - new_tok
                logging.warning(
                    f"[reduce_context] Hard-truncated {section_name}: "
                    f"{section_tok} -> {new_tok} tokens (target {target_tokens})"
                )

        logging.info(
            f"[reduce_context] Final context: {reduced_tokens} tokens (target: {target_tokens})"
        )
        return reduced_context

    async def select_commands_for_task(
        self,
        user_input: str,
        conversation_name: str,
        file_context: str = "",
        has_uploaded_files: bool = False,
        log_output: bool = True,
        thinking_id: str = "",
        conversation_history: str = "",
    ) -> list:
        """
        Intelligently select which commands should be available for this task.
        This is called at the start of inference to optimize the command set.
        Splits commands into two batches to reduce token count per selection call.

        Args:
            user_input: The user's input/request
            conversation_name: Name of the conversation
            file_context: Description of files in workspace (not content, just names/types)
            has_uploaded_files: Whether the user uploaded files with this request
            log_output: Whether to log the selection as a subactivity
            thinking_id: Optional thinking_id for logging subactivities
            conversation_history: Recent conversation messages for follow-up context

        Returns:
            list: List of command friendly names that should be enabled
        """
        # Get all available commands with descriptions
        result = self.agent.get_commands_for_selection()
        if len(result) == 3:
            commands_prompt, all_command_names, command_descriptions = result
        else:
            # Backward compatibility
            commands_prompt, all_command_names = result
            command_descriptions = {}

        if not all_command_names:
            return []

        # If total enabled commands fit within a reasonable token budget, skip LLM selection
        # and just return all of them. LLM-based selection often misses relevant commands
        # (especially from extension hubs) because user intent doesn't keyword-match command names.
        # NOTE: The selection prompt is a compact format (~65 tokens/cmd), but the full
        # execution prompt with XML examples is ~3x larger. Keep this threshold low
        # so the full execution prompt stays within context budget (~20k tokens).
        commands_tokens = get_tokens(commands_prompt)
        logging.info(
            f"[select_commands_for_task] {len(all_command_names)} commands, "
            f"selection prompt: {commands_tokens} tokens"
        )
        if commands_tokens <= 12000:
            if log_output and thinking_id:
                c = Conversations(
                    conversation_name=conversation_name,
                    user=self.user,
                )
                c.log_interaction(
                    role=self.agent_name,
                    message=f"[SUBACTIVITY][{thinking_id}] All {len(all_command_names)} abilities available",
                )
            return all_command_names

        # Get extension context to help with command selection
        # This provides crucial info like machine names, connected accounts, etc.
        # that helps the LLM understand user intent (e.g. "xthh01" is a machine)
        extension_context = ""
        try:
            agent_extensions = self.agent.get_agent_extensions()
            extension_context = self.agent._get_extension_contexts(
                agent_extensions, all_command_names
            )
            if extension_context:
                extension_context = extension_context.strip()
        except Exception as e:
            logging.debug(
                f"[select_commands_for_task] Failed to get extension context: {e}"
            )
            extension_context = ""

        # If user literally mentions a command by its exact name, always include it.
        # This is the only heuristic — everything else is left to the LLM with good context.
        user_input_lower = user_input.lower()
        explicitly_requested_commands = []
        for cmd_name in all_command_names:
            if cmd_name.lower() in user_input_lower:
                explicitly_requested_commands.append(cmd_name)

        # Codex is a delegation-oriented coding ability. Users naturally ask for
        # "Codex" or repo/codebase work without spelling the exact command name,
        # so make the ability available whenever those signals appear.
        codex_trigger_pattern = re.compile(
            r"\b(codex|openai\s+codex|codebase|repo|repository|github|"
            r"git\s+clone|clone|branch|pull\s+request|pr|commit|refactor)\b"
        )
        if codex_trigger_pattern.search(user_input_lower):
            codex_command = "Ask OpenAI Codex"
            if (
                codex_command in all_command_names
                and codex_command not in explicitly_requested_commands
            ):
                explicitly_requested_commands.append(codex_command)

        # Keyword-based command boosting: when user mentions specific media types,
        # always include the relevant generation commands so the LLM doesn't have to
        # guess across hundreds of commands split into batches.
        KEYWORD_COMMAND_MAP = {
            "video": ["Generate Video", "Image to Video", "Video to Video"],
            "image": ["Generate Image", "Edit Image"],
            "picture": ["Generate Image", "Edit Image"],
            "photo": ["Generate Image", "Edit Image"],
            "speech": ["Text to Speech"],
            "transcri": ["Transcribe Audio"],
        }
        for keyword, prefixes in KEYWORD_COMMAND_MAP.items():
            if keyword in user_input_lower:
                for cmd_name in all_command_names:
                    cmd_lower = cmd_name.lower()
                    if any(p.lower() in cmd_lower for p in prefixes):
                        if cmd_name not in explicitly_requested_commands:
                            explicitly_requested_commands.append(cmd_name)

        # Build context about files, extensions, and conversation history
        context_parts = []
        if conversation_history:
            context_parts.append(
                f"## Recent Conversation (for understanding follow-up requests)\n{conversation_history}"
            )
        if extension_context:
            context_parts.append(extension_context)
        if file_context:
            context_parts.append(f"Files in workspace: {file_context}")
        if has_uploaded_files:
            context_parts.append("The user has uploaded file(s) with this request.")

        context = (
            "\n".join(context_parts) if context_parts else "No additional context."
        )

        # Token-aware batching: The ability selection model can handle large context
        # (200k tokens), so we use generous batch sizes to keep related commands together
        # which helps the model understand command relationships and prerequisites.
        MAX_BATCH_TOKENS = 100000

        # Calculate base prompt template tokens (without the batch content)
        base_selection_prompt = f"""You are an expert assistant that selects relevant commands/tools for a user request. You manage IT infrastructure, remote machines, tickets, desktop support, code execution, web browsing, media generation, and more. Your job is to select exactly the right set of commands the agent will need.

## User's Request
{user_input}

## Context
{context}

{{BATCH_COMMANDS}}

## Your Task
Select the commands needed to fulfill this request. Think carefully about:
1. **What the user is ACTUALLY asking to accomplish** — read between the lines. If they mention a hostname, device name, or machine name, they want to interact with that machine.
2. **What commands are PREREQUISITES** — many commands require setup steps first. For example:
   - Remote machine work requires: Open Remote Terminal → Execute in Terminal → Get Terminal Output
   - File transfers require machine identification first
   - Desktop control requires: Vision Desktop Control (which handles screenshots + mouse + keyboard)
3. **What the Context section tells you about available resources** — device names, hostnames, connected accounts, active sessions, etc. Match user references to these resources.
4. **The recent conversation history** — follow-up messages like "pause it", "do that again", "now close it", "check on that machine" refer to services/tools/devices used in recent messages. Select commands from the SAME extension/service that was just used.
5. **The full scope of each extension** — if the user needs ANY command from an extension, include ALL commands from that extension that could be relevant to the workflow.

**Selection guidance:**
- Most tasks need 3-10 commands. Select all that may be needed, including prerequisites and related commands.
- For follow-up requests, ALWAYS include commands from the extension/service used in the previous turn.
- If the user references a device/machine by hostname, IP, name, or any identifier from Context, include ALL machine interaction commands (terminal, desktop control, file transfer, machine details).
- If the user mentions tickets, issues, problems, or support requests, include ticket management commands.
- If the user wants to interact with files, include file operation commands.
- If the user wants web information, include web/search commands.
- Include both the "ideal" command and reasonable fallbacks (e.g. both desktop control AND terminal for opening an app).
- When in doubt, INCLUDE the command — it's better to have an unused command available than to miss one the agent needs.

**When NOT to select:**
- Commands from completely unrelated domains (e.g. don't select Spotify commands for a machine terminal task)
- Commands for services with no connection to the user's request or recent conversation

Respond with ONLY a comma-separated list of exact command names, or "None" if no commands are needed.
No explanations. Maximum 25 commands from this batch.

Example: Open Remote Terminal, Execute in Terminal, Get Terminal Output, Vision Desktop Control, Get Machine Details"""

        # Calculate the base prompt overhead (tokens used by template, user input, context)
        base_overhead_tokens = get_tokens(
            base_selection_prompt.replace("{BATCH_COMMANDS}", "")
        )
        available_tokens_per_batch = MAX_BATCH_TOKENS - base_overhead_tokens

        # Parse commands_prompt into individual command blocks (by extension sections)
        # Format: ### ExtensionName\nDescription\n- **CmdName**: CmdDesc\n...
        command_blocks = []
        current_block = []
        current_block_tokens = 0

        for line in commands_prompt.strip().split("\n"):
            line_tokens = get_tokens(line + "\n")

            # If adding this line would exceed the batch limit, save current block and start new
            if (
                current_block
                and (current_block_tokens + line_tokens) > available_tokens_per_batch
            ):
                command_blocks.append("\n".join(current_block))
                current_block = [line]
                current_block_tokens = line_tokens
            else:
                current_block.append(line)
                current_block_tokens += line_tokens

        # Don't forget the last block
        if current_block:
            command_blocks.append("\n".join(current_block))

        # Log individual batch sizes for debugging
        for i, block in enumerate(command_blocks):
            block_tokens = get_tokens(block)

        # Get conversation for logging
        c = Conversations(
            conversation_name=conversation_name,
            user=self.user,
        )

        if not thinking_id and log_output:
            thinking_id = c.get_thinking_id(agent_name=self.agent_name)

        valid_commands = []

        # Check if a dedicated ability selection server is configured
        ability_selection_server = getenv("ABILITY_SELECTION_SERVER")
        ability_selection_model = getenv(
            "ABILITY_SELECTION_MODEL", "unsloth/Qwen3.5-4B-GGUF"
        )

        # Process batches in parallel for speed using DIRECT inference (not run())
        # This avoids pulling in all memories/context which bloats token count
        async def select_from_batch(batch_prompt: str, batch_num: int) -> list:
            try:
                # Build a lightweight prompt directly without format_prompt overhead
                selection_prompt = base_selection_prompt.replace(
                    "{BATCH_COMMANDS}", batch_prompt
                )

                if ability_selection_server:
                    # Use the dedicated ability selection server for fast inference
                    # This allows offloading command selection to a smaller/faster model
                    selection_response = await _ability_selection_inference(
                        server_url=ability_selection_server,
                        model=ability_selection_model,
                        prompt=selection_prompt,
                    )
                else:
                    # Direct streaming inference - bypasses format_prompt and all its context loading
                    # Uses streaming internally for speed (non-streaming blocks until full response)
                    selection_response = await stream_inference_to_string(
                        self.agent, prompt=selection_prompt
                    )

                # Handle "None" or empty responses
                if (
                    not selection_response
                    or selection_response.strip().lower() == "none"
                ):
                    return []

                # Parse the response - should be comma-separated command names
                selected = [
                    cmd.strip()
                    for cmd in selection_response.split(",")
                    if cmd.strip() and cmd.strip().lower() != "none"
                ]

                # Validate against actual command names
                return [cmd for cmd in selected if cmd in all_command_names]

            except Exception as e:
                logging.error(
                    f"[select_commands_for_task] Error in batch {batch_num}: {e}"
                )
                return []

        # Run all batches in parallel and combine results
        try:
            batch_results = await asyncio.gather(
                *[
                    select_from_batch(block, i + 1)
                    for i, block in enumerate(command_blocks)
                ]
            )
            for result in batch_results:
                valid_commands.extend(result)
        except Exception as e:
            logging.error(
                f"[select_commands_for_task] Error in parallel selection: {e}"
            )
            # Fallback to all commands on error
            return all_command_names

        # Always include commands that user explicitly mentioned by exact name
        for cmd in explicitly_requested_commands:
            if cmd not in valid_commands:
                valid_commands.append(cmd)

        # Essential commands that must always be available when commands are enabled.
        # These are the foundational abilities for file operations, code execution,
        # and web interaction — without them the agent cannot perform basic tasks.
        # Remote terminal commands are included because they are prerequisites for
        # any machine interaction and the selection model may miss them.
        ESSENTIAL_COMMANDS = [
            "Read File",
            "Modify File",
            "Write to File",
            "List Directory",
            "Search Files",
            "Execute Python Code",
            "Use Terminal in Workspace",
            "Interact with Webpage",
            "Web Search",
            "Open Remote Terminal",
            "Execute in Terminal",
            "Get Terminal Output",
        ]
        for cmd in ESSENTIAL_COMMANDS:
            if cmd in all_command_names and cmd not in valid_commands:
                valid_commands.append(cmd)

        # Remove duplicates while preserving order
        seen = set()
        unique_commands = []
        for cmd in valid_commands:
            if cmd not in seen:
                seen.add(cmd)
                unique_commands.append(cmd)
        valid_commands = unique_commands

        # Cap at 35 commands to keep execution prompt within context budget
        # while allowing enough room for multi-extension workflows
        MAX_COMMANDS = 35
        if len(valid_commands) > MAX_COMMANDS:
            # Keep explicitly requested first, then LLM-selected in order
            prioritized = [
                c for c in valid_commands if c in explicitly_requested_commands
            ]
            for cmd in valid_commands:
                if cmd not in prioritized and len(prioritized) < MAX_COMMANDS:
                    prioritized.append(cmd)
            valid_commands = prioritized
            logging.info(
                f"[select_commands_for_task] Capped commands from {len(unique_commands)} to {len(valid_commands)}"
            )

        # Log the selection
        logging.info(
            f"[select_commands_for_task] Selected {len(valid_commands)} commands via LLM: {', '.join(valid_commands)}"
        )
        if log_output and thinking_id:
            if valid_commands:
                c.log_interaction(
                    role=self.agent_name,
                    message=f"[SUBACTIVITY][{thinking_id}] Selected {len(valid_commands)} abilities: {', '.join(valid_commands)}",
                )
            else:
                c.log_interaction(
                    role=self.agent_name,
                    message=f"[SUBACTIVITY][{thinking_id}] No additional abilities needed for this request.",
                )
        return valid_commands

    async def run(
        self,
        user_input: str = "",
        context_results: int = 100,
        shots: int = 1,
        disable_memory: bool = True,
        conversation_name: str = "",
        conversation_id: str = None,
        browse_links: bool = False,
        persist_context_in_history: bool = False,
        images: list = [],
        searching: bool = False,
        log_user_input: bool = True,
        log_output: bool = True,
        command_overrides: list = None,
        parent_activity_id: str = None,
        **kwargs,
    ):
        """
        Non-streaming version of run that internally uses run_stream() for all inference.
        This ensures consistent behavior between streaming and non-streaming paths.

        Returns the final answer as a string after all inference and command execution completes.
        """
        global AGIXT_URI

        # Store conversation_id in kwargs for downstream use
        if conversation_id:
            kwargs["conversation_id"] = conversation_id
        # Store parent_activity_id in kwargs for downstream use (e.g., nested prompt_agent calls)
        if parent_activity_id:
            kwargs["parent_activity_id"] = parent_activity_id

        # Handle shots parameter (multiple response generations)
        if shots == 0:
            shots = 1
        shots = int(shots)

        # Process prompt parameters
        prompt = "Think About It"
        prompt_category = "Default"
        if "prompt_category" in kwargs:
            prompt_category = kwargs["prompt_category"]
        if "prompt_name" in kwargs:
            prompt = kwargs["prompt_name"]
        if "prompt" in kwargs:
            prompt = kwargs["prompt"]

        # Handle disable_memory
        disable_memory = False if str(disable_memory).lower() == "false" else True
        if "disable_memory" in kwargs:
            disable_memory = (
                False if str(kwargs["disable_memory"]).lower() == "false" else True
            )
            kwargs["disable_memory"] = disable_memory

        # Handle browse_links
        browse_links = True if str(browse_links).lower() == "true" else False
        if "browse_links" in kwargs:
            browse_links = (
                True if str(kwargs["browse_links"]).lower() == "true" else False
            )

        # Handle collection_number for websearch
        if "collection_number" in kwargs:
            collection_number = str(kwargs["collection_number"])
            self.websearch = Websearch(
                collection_number=collection_number,
                agent=self.agent,
                user=self.user,
                ApiClient=self.ApiClient,
            )

        # Handle use_smartest
        use_smartest = False
        if "use_smartest" in kwargs:
            use_smartest = (
                True if str(kwargs["use_smartest"]).lower() == "true" else False
            )

        # Handle websearch
        websearch = False
        if "websearch" in self.agent.AGENT_CONFIG["settings"]:
            websearch = (
                str(self.agent.AGENT_CONFIG["settings"]["websearch"]).lower() == "true"
            )
        if "websearch" in kwargs:
            websearch = True if str(kwargs["websearch"]).lower() == "true" else False

        # Clean up kwargs to remove keys that are explicitly passed to run_stream
        # to avoid "got multiple values for keyword argument" errors
        stream_explicit_keys = [
            # Explicitly passed positional/keyword args to run_stream
            "user_input",
            "context_results",
            "conversation_name",
            "conversation_id",
            "browse_links",
            "websearch",
            "images",
            "log_user_input",
            "log_output",
            "use_smartest",
            "thinking_id",
            "searching",
            "command_overrides",
            # Additional keys processed in run()
            "disable_memory",
            "prompt_category",
            "prompt_name",
            "prompt",
            "collection_number",
            "parent_activity_id",
        ]
        stream_kwargs = {
            k: v for k, v in kwargs.items() if k not in stream_explicit_keys
        }

        # Run single shot using run_stream and collect final answer
        final_answer = ""
        # Capture any remote command requests for client-defined tools
        self._pending_remote_commands = []
        async for chunk in self.run_stream(
            user_input=user_input,
            context_results=context_results,
            conversation_name=conversation_name,
            conversation_id=conversation_id,
            browse_links=browse_links,
            websearch=websearch,
            images=images,
            log_user_input=log_user_input,
            log_output=log_output,
            use_smartest=use_smartest,
            searching=searching,
            command_overrides=command_overrides,
            **stream_kwargs,
        ):
            # Collect only complete answer chunks
            if chunk.get("type") == "answer" and chunk.get("complete"):
                final_answer = chunk.get("content", "")
            # Also capture remote command requests for client-defined tools
            elif chunk.get("type") == "remote_command_request":
                self._pending_remote_commands.append(chunk.get("content", {}))

        self.response = final_answer

        # Handle multiple shots if requested
        if shots > 1:
            responses = [final_answer]
            conversation_results = kwargs.get("conversation_results", 5)
            for shot in range(shots - 1):
                shot_kwargs = {
                    "user_input": user_input,
                    "context_results": context_results,
                    "conversation_results": conversation_results,
                    "conversation_name": conversation_name,
                    "disable_memory": disable_memory,
                }
                # Copy non-conflicting kwargs
                for k, v in kwargs.items():
                    if k not in [
                        "images",
                        "searching",
                        "tts",
                        "websearch",
                        "websearch_depth",
                        "browse_links",
                    ]:
                        shot_kwargs[k] = v

                shot_response = await self.run(
                    agent_name=self.agent_name,
                    prompt_name=prompt,
                    prompt_category=prompt_category,
                    log_user_input=False,
                    log_output=False,
                    websearch=False,
                    browse_links=False,
                    searching=True,
                    tts=False,
                    **shot_kwargs,
                )
                time.sleep(1)
                responses.append(shot_response)
            return "\n".join(
                [
                    f"Response {shot + 1}:\n{response}"
                    for shot, response in enumerate(responses)
                ]
            )
        return final_answer

    async def run_stream(
        self,
        user_input: str = "",
        context_results: int = 100,
        conversation_name: str = "",
        conversation_id: str = None,
        browse_links: bool = False,
        websearch: bool = False,
        images: list = [],
        log_user_input: bool = True,
        log_output: bool = True,
        use_smartest: bool = False,
        thinking_id: str = None,
        searching: bool = False,
        command_overrides: list = None,
        **kwargs,
    ):
        """
        Streaming version of run() that yields tokens as they come from the LLM.
        This allows real-time streaming of thinking activities to the frontend.

        Yields:
            dict with keys:
                - 'type': 'thinking', 'reflection', 'answer', 'activity'
                - 'content': the text content
                - 'complete': whether the tag is fully received
        """
        global AGIXT_URI
        import time as _time

        _stream_t0 = _time.monotonic()

        # Store conversation_id in kwargs for downstream use
        if conversation_id:
            kwargs["conversation_id"] = conversation_id
        for setting in self.agent.AGENT_CONFIG["settings"]:
            if setting not in kwargs:
                kwargs[setting] = self.agent.AGENT_CONFIG["settings"][setting]

        context_results = 5 if not context_results else int(context_results)
        prompt = "Think About It"
        prompt_category = "Default"
        if "prompt_category" in kwargs:
            prompt_category = kwargs["prompt_category"]
            del kwargs["prompt_category"]
        if "prompt_name" in kwargs:
            prompt = kwargs["prompt_name"]
            del kwargs["prompt_name"]
        if "prompt" in kwargs:
            prompt = kwargs["prompt"]
            del kwargs["prompt"]

        disable_memory = True
        if "disable_memory" in kwargs:
            disable_memory = (
                False if str(kwargs["disable_memory"]).lower() == "false" else True
            )
            del kwargs["disable_memory"]

        # Handle browse_links from kwargs or agent settings
        browse_links = True if str(browse_links).lower() == "true" else False
        if "browse_links" in kwargs:
            browse_links = (
                True if str(kwargs["browse_links"]).lower() == "true" else False
            )
            del kwargs["browse_links"]

        # Handle websearch from kwargs
        websearch = True if str(websearch).lower() == "true" else False
        if "websearch" in kwargs:
            websearch = True if str(kwargs["websearch"]).lower() == "true" else False
            del kwargs["websearch"]

        # Handle collection_number - reinitialize websearch with specific collection
        if "collection_number" in kwargs:
            collection_number = str(kwargs["collection_number"])
            self.websearch = Websearch(
                collection_number=collection_number,
                agent=self.agent,
                user=self.user,
                ApiClient=self.ApiClient,
            )
            del kwargs["collection_number"]

        # Get websearch settings from agent config
        websearch_depth = 3
        if "websearch" in self.agent.AGENT_CONFIG["settings"]:
            if not websearch:  # Only override if not explicitly set
                websearch = (
                    str(self.agent.AGENT_CONFIG["settings"]["websearch"]).lower()
                    == "true"
                )
        if "websearch_depth" in self.agent.AGENT_CONFIG["settings"]:
            websearch_depth = int(
                self.agent.AGENT_CONFIG["settings"]["websearch_depth"]
            )
        if "browse_links" in self.agent.AGENT_CONFIG["settings"]:
            if not browse_links:  # Only override if not explicitly set
                browse_links = (
                    str(self.agent.AGENT_CONFIG["settings"]["browse_links"]).lower()
                    == "true"
                )
        if "websearch_depth" in kwargs:
            try:
                websearch_depth = int(kwargs["websearch_depth"])
            except:
                websearch_depth = 3
            del kwargs["websearch_depth"]
        if "WEBSEARCH_TIMEOUT" in kwargs:
            try:
                websearch_timeout = int(kwargs["WEBSEARCH_TIMEOUT"])
            except:
                websearch_timeout = 0
        else:
            websearch_timeout = 0

        conversation_results = 5
        if "conversation_results" in kwargs:
            try:
                conversation_results = int(kwargs["conversation_results"])
            except:
                conversation_results = 5
            del kwargs["conversation_results"]

        if "conversation_name" in kwargs:
            conversation_name = kwargs["conversation_name"]
        if conversation_name == "":
            conversation_name = "-"

        # Use conversation_id if provided
        conversation_id = kwargs.get("conversation_id")
        c = Conversations(
            conversation_name=conversation_name,
            user=self.user,
            conversation_id=conversation_id,
        )

        # Handle vision if needed
        vision_response = ""
        if "vision_provider" in self.agent.AGENT_CONFIG["settings"]:
            if (
                images != []
                and self.agent.VISION_PROVIDER != "None"
                and self.agent.VISION_PROVIDER != ""
                and self.agent.VISION_PROVIDER != None
            ):
                c.log_interaction(
                    role=self.agent_name,
                    message=f"[ACTIVITY] Viewing images.",
                )
                try:
                    vision_response = await self.agent.vision_inference(
                        prompt=user_input, images=images, use_smartest=use_smartest
                    )
                except Exception as e:
                    c.log_interaction(
                        role=self.agent_name,
                        message=f"[ACTIVITY][ERROR] Unable to view image.",
                    )
                    logging.error(f"Error getting vision response: {e}")

        # Handle browse_links and websearch async tasks
        async_tasks = []
        if browse_links != False and websearch == False and searching == False:
            task = asyncio.create_task(
                self.websearch.scrape_websites(
                    user_input=user_input,
                    summarize_content=False,
                    conversation_name=conversation_name,
                )
            )
            async_tasks.append(task)

        if websearch and searching == False:
            if browse_links != False:
                task = asyncio.create_task(
                    self.websearch.scrape_websites(
                        user_input=user_input,
                        summarize_content=False,
                        conversation_name=conversation_name,
                    )
                )
                async_tasks.append(task)
            if user_input == "":
                if "primary_objective" in kwargs and "task" in kwargs:
                    user_input = f"Primary Objective: {kwargs['primary_objective']}\n\nTask: {kwargs['task']}"
            if user_input != "":
                thinking_id = c.get_thinking_id(agent_name=self.agent_name)
                searching_activity_id = c.log_interaction(
                    role=self.agent_name,
                    message=f"[SUBACTIVITY][{thinking_id}] Searching for information.",
                )
                ability_selection_server = getenv("ABILITY_SELECTION_SERVER")
                if ability_selection_server:
                    # Use fast 0.8B model for websearch yes/no decision
                    websearch_decision_prompt = (
                        f"Today's date is {datetime.now().strftime('%B %d, %Y')}.\n\n"
                        f"User's input: {user_input}\n\n"
                        "Decide if the user's input requires searching the web to answer.\n"
                        "- Search if the user asks about recent events, news, or needs current factual data.\n"
                        "- Do NOT search if the user is asking you to do a task, write code, or make something.\n"
                        "- Do NOT search if enough context is available to answer directly.\n\n"
                        "Respond ONLY with Yes or No."
                    )
                    to_search_or_not_to_search = await _ability_selection_inference(
                        server_url=ability_selection_server,
                        model=getenv(
                            "ABILITY_SELECTION_MODEL",
                            "unsloth/Qwen3.5-4B-GGUF",
                        ),
                        prompt=websearch_decision_prompt,
                    )
                else:
                    to_search_or_not_to_search = await self.run(
                        prompt_name="WebSearch Decision",
                        prompt_category="Default",
                        user_input=user_input,
                        context_results=context_results,
                        conversation_results=4,
                        conversation_name=conversation_name,
                        log_user_input=False,
                        log_output=False,
                        browse_links=False,
                        websearch=False,
                        tts=False,
                        searching=True,
                    )
                to_search = re.search(
                    r"\byes\b", str(to_search_or_not_to_search).lower()
                )
                if to_search:
                    if ability_selection_server:
                        # Use fast 0.8B model for search query generation
                        websearch_query_prompt = (
                            f"Today's date is {datetime.now().strftime('%B %d, %Y')}.\n\n"
                            f"User's input: {user_input}\n\n"
                            "Generate optimal web search strings for the user's input.\n"
                            "- Include today's date in searches when recency matters.\n"
                            "- For scientific queries, add 'ar5iv' or 'PubMed'.\n"
                            "- For product info, add 'review'.\n"
                            "- For stock/finance, add 'Yahoo Finance' or 'Bloomberg'.\n\n"
                            f"Provide up to {websearch_depth} search suggestions.\n\n"
                            "Respond ONLY with JSON, no explanation:\n"
                            "```json\n"
                            '{\n    "search_string_suggestion_1": "example search query",\n'
                            '    "search_string_suggestion_2": "another query"\n}\n'
                            "```"
                        )
                        search_strings = await _ability_selection_inference(
                            server_url=ability_selection_server,
                            model=getenv(
                                "ABILITY_SELECTION_MODEL",
                                "unsloth/Qwen3.5-4B-GGUF",
                            ),
                            prompt=websearch_query_prompt,
                        )
                    else:
                        search_strings = await self.run(
                            prompt_name="WebSearch",
                            prompt_category="Default",
                            user_input=user_input,
                            context_results=context_results,
                            conversation_results=10,
                            conversation_name=conversation_name,
                            log_user_input=False,
                            log_output=False,
                            browse_links=False,
                            websearch=False,
                            tts=False,
                            searching=True,
                        )
                    if "```json" in search_strings:
                        search_strings = (
                            search_strings.split("```json")[1].split("```")[0].strip()
                        )
                    elif "```" in search_strings:
                        search_strings = search_strings.split("```")[1].strip()
                    try:
                        search_suggestions = json.loads(search_strings)
                    except:
                        keywords = extract_keywords(text=str(search_strings), limit=5)
                        if keywords:
                            search_string = " ".join(keywords)
                            search_string += f" {datetime.now().strftime('%B %Y')}"
                        search_suggestions = [
                            {"search_string_suggestion_1": search_string}
                        ]
                    search_strings = []
                    if search_suggestions != []:
                        for i in range(1, int(websearch_depth) + 1):
                            if f"search_string_suggestion_{i}" in search_suggestions:
                                search_string = search_suggestions[
                                    f"search_string_suggestion_{i}"
                                ]
                                search_strings.append(search_string)
                                search_task = asyncio.create_task(
                                    self.websearch.websearch_agent(
                                        user_input=user_input,
                                        search_string=search_string,
                                        websearch_depth=websearch_depth,
                                        websearch_timeout=websearch_timeout,
                                        conversation_name=conversation_name,
                                        activity_id=searching_activity_id,
                                    )
                                )
                                async_tasks.append(search_task)

        # Wait for all async tasks to complete before formatting prompt
        await asyncio.gather(*async_tasks)

        # Process command_overrides BEFORE format_prompt so enabled state is correct
        # Initialize client_tools to store OpenAI-format function definitions for client-side execution
        self._client_tools = {}

        if command_overrides:
            # Check if any tool requests exclusive mode (disable all other commands)
            exclusive_mode = any(
                tool.get("exclusive", False) for tool in command_overrides
            )
            requested_tools = [tool.get("type") for tool in command_overrides]

            if exclusive_mode:
                # Disable ALL commands first, then enable only the requested ones
                for cmd in self.agent.available_commands:
                    cmd["enabled"] = False

            for tool in command_overrides:
                # Handle OpenAI-format tools (type: "function" with function object)
                # These are client-defined tools that should be routed back to the client for execution
                if tool.get("type") == "function" and "function" in tool:
                    func_def = tool["function"]
                    func_name = func_def.get("name", "")
                    if func_name:
                        # Store the full function definition for later use in execution_agent
                        self._client_tools[func_name] = func_def

                        # Add a pseudo-command to available_commands so the agent knows about it
                        # This allows the command to appear in the prompt and be called
                        description = func_def.get(
                            "description", f"Client-defined tool: {func_name}"
                        )
                        parameters = func_def.get("parameters", {})

                        # Build args list from parameters
                        args = {}
                        props = parameters.get("properties", {})
                        for param_name, param_def in props.items():
                            args[param_name] = param_def.get(
                                "description", f"Parameter: {param_name}"
                            )

                        # Create a pseudo-command entry
                        client_command = {
                            "friendly_name": func_name,
                            "name": func_name,
                            "description": description,
                            "enabled": True,
                            "args": args,
                            "extension_name": "__client__",  # Special marker for client-defined tools
                        }
                        self.agent.available_commands.append(client_command)
                    continue

                # Handle legacy format (type is the command name)
                tool_type = tool.get("type")
                # Find the command in available_commands list and enable it
                # This allows CLI/API to enable specific commands for this request
                for available_command in self.agent.available_commands:
                    if available_command["friendly_name"] == tool_type:
                        # Always enable when specified in tools (not toggle)
                        available_command["enabled"] = True
                        # If enabling Execute Terminal Command, disable Execute Shell
                        # to ensure the agent uses the remote terminal instead
                        if tool_type == "Execute Terminal Command":
                            for cmd in self.agent.available_commands:
                                if cmd["friendly_name"] == "Execute Shell":
                                    cmd["enabled"] = False
                                    break
                        break

        # Intelligent command selection - select relevant commands before building prompt
        selected_commands = None
        prompt_content = self.cp.get_prompt(
            prompt_name=prompt, prompt_category=prompt_category
        )
        has_commands_placeholder = (
            "{COMMANDS}" in prompt_content if prompt_content else False
        )

        # Enable command selection if commands are available and this is the main user interaction
        enable_command_selection = kwargs.get("enable_command_selection", log_output)

        file_context = ""
        if (
            "disable_commands" not in kwargs
            and not searching
            and enable_command_selection
            and has_commands_placeholder
        ):
            # Build file context for selection
            has_uploaded_files = False
            if "uploaded_file_data" in kwargs:
                has_uploaded_files = True
                uploaded_data = kwargs.get("uploaded_file_data", "")
                if (
                    "file uploaded named" in uploaded_data.lower()
                    or "Content from" in uploaded_data
                ):
                    file_matches = re.findall(r"`([^`]+\.[a-zA-Z0-9]+)`", uploaded_data)
                    if file_matches:
                        file_context = f"Uploaded files: {', '.join(file_matches)}"

            # Auto-discover existing files in the conversation workspace
            try:
                workspace_dir = (
                    f"{self.agent.working_directory}/{c.get_conversation_id()}"
                )
                if os.path.isdir(workspace_dir):
                    existing_files = []
                    collapsed_dirs_seen = set()
                    for root, dirs, files in os.walk(workspace_dir):
                        # Prune noisy directories in-place so os.walk skips
                        # descending into them (avoids enumerating millions
                        # of node_modules/.git files). We still record one
                        # placeholder path so the directory shows up in the
                        # rendered tree as collapsed.
                        pruned = [d for d in dirs if d in self._COLLAPSED_TREE_DIRS]
                        for d in pruned:
                            collapsed_path = os.path.relpath(
                                os.path.join(root, d), workspace_dir
                            )
                            if collapsed_path not in collapsed_dirs_seen:
                                collapsed_dirs_seen.add(collapsed_path)
                                # Add a sentinel child so _build_file_tree
                                # registers this as a collapsed directory.
                                existing_files.append(f"{collapsed_path}/__collapsed__")
                        dirs[:] = [
                            d for d in dirs if d not in self._COLLAPSED_TREE_DIRS
                        ]
                        for f in files:
                            rel_path = os.path.relpath(
                                os.path.join(root, f), workspace_dir
                            )
                            existing_files.append(rel_path)
                    if existing_files:
                        # Build an indented file tree for agent context
                        tree_lines = self._build_file_tree(existing_files)
                        existing_context = f"Workspace file tree:\n{tree_lines}"
                        if file_context:
                            file_context = f"{file_context}\n{existing_context}"
                        else:
                            file_context = existing_context
                        has_uploaded_files = True
                        logging.info(
                            f"[run_stream] Workspace discovery found {len(existing_files)} files in {workspace_dir}"
                        )
                else:
                    logging.info(
                        f"[run_stream] Workspace directory does not exist: {workspace_dir}"
                    )
            except Exception as e:
                logging.warning(
                    f"[run_stream] Error during workspace file discovery: {e}"
                )

            # Do intelligent command selection with conversation history
            # so follow-ups like "pause it" correctly map to Spotify, etc.
            yield {
                "type": "thinking_stream",
                "content": "Analyzing request...",
                "complete": False,
            }
            _t_cmd_sel = _time.monotonic()
            try:
                # Build recent conversation summary for command selection context
                _recent_history = ""
                _conv_data = c.get_conversation(limit=1000, page=1)
                if "interactions" in _conv_data and _conv_data["interactions"]:
                    _recent_msgs = [
                        i
                        for i in _conv_data["interactions"]
                        if not str(i.get("message", "")).startswith("[ACTIVITY]")
                        and not str(i.get("message", "")).startswith("[SUBACTIVITY]")
                        and not str(i.get("message", "")).startswith("<audio")
                    ]
                    # Last 6 messages (3 turns) is enough context for follow-ups
                    _recent_msgs = _recent_msgs[-6:]
                    if _recent_msgs:
                        _history_lines = []
                        for msg in _recent_msgs:
                            role = msg.get("role", "")
                            content = msg.get("message", "")
                            # Truncate long messages - we just need the gist
                            if len(content) > 300:
                                content = content[:300] + "..."
                            _history_lines.append(f"{role}: {content}")
                        _recent_history = "\n".join(_history_lines)

                selected_commands = await self.select_commands_for_task(
                    user_input=user_input,
                    conversation_name=conversation_name,
                    file_context=file_context,
                    has_uploaded_files=has_uploaded_files,
                    log_output=log_output,
                    thinking_id=thinking_id,
                    conversation_history=_recent_history,
                )
            except Exception as e:
                logging.error(f"[run_stream] select_commands_for_task failed: {e}")
                selected_commands = None
            logging.info(
                f"[run_stream] select_commands_for_task took {_time.monotonic() - _t_cmd_sel:.1f}s, "
                f"selected {len(selected_commands) if selected_commands is not None else 'None'} commands"
            )

        # Always include client-defined tools regardless of command selection.
        # When command selection is intentionally skipped for a tool-result
        # continuation, constrain the command prompt to the client tools
        # instead of falling back to every server-side ability.
        if self._client_tools:
            client_tool_names = list(self._client_tools.keys())
            if selected_commands is None and not enable_command_selection:
                selected_commands = client_tool_names
            elif selected_commands is not None:
                for client_tool_name in client_tool_names:
                    if client_tool_name not in selected_commands:
                        selected_commands.append(client_tool_name)

        # Store selected_commands as instance variable to persist across continuation loops
        self._selected_commands = selected_commands

        # Pass workspace file context to format_prompt so the agent knows what files exist
        if file_context and "workspace_file_context" not in kwargs:
            kwargs["workspace_file_context"] = file_context

        # Remove selected_commands from kwargs if present to avoid duplicate parameter
        kwargs.pop("selected_commands", None)

        # Add TTS filler instructions if TTS mode is enabled
        # This allows the model to generate <speak> tags with filler phrases during thinking
        tts_enabled = str(kwargs.get("tts", "false")).lower() == "true"

        if not tts_enabled:
            # Also check agent settings for tts_provider
            agent_settings = self.agent.AGENT_CONFIG.get("settings", {})
            tts_provider = agent_settings.get("tts_provider", "")
            tts_enabled = tts_provider and tts_provider not in ("None", "", None)

        if tts_enabled:
            kwargs[
                "tts_filler_instructions"
            ] = """
**VOICE MODE - IMMEDIATE RESPONSE REQUIRED**

The user is speaking to you and will hear your response aloud. You MUST begin with TWO `<speak>` tags IMMEDIATELY - before ANY thinking, commands, or processing:

1. **FIRST <speak>** (within your first 10 tokens): A brief acknowledgment (2-4 words)
   Examples: `<speak>Got it.</speak>` or `<speak>Sure thing.</speak>` or `<speak>One moment.</speak>`

2. **SECOND <speak>** (before <thinking>): Tell them what you're doing (4-8 words)
   Examples: `<speak>Let me check the time for you.</speak>` or `<speak>Looking that up now.</speak>`

Then proceed with your `<thinking>` tag.

Example response start:
`<speak>Sure.</speak><speak>Let me check on that for you.</speak><thinking>...`

This prevents awkward silence - the user hears feedback within 1 second while you process.
"""
        else:
            kwargs["tts_filler_instructions"] = ""

        # Format the prompt
        _t_fmt = _time.monotonic()
        formatted_prompt, unformatted_prompt, tokens = await self.format_prompt(
            user_input=user_input,
            top_results=int(context_results),
            conversation_results=conversation_results,
            prompt=prompt,
            prompt_category=prompt_category,
            conversation_name=conversation_name,
            websearch=websearch,
            vision_response=vision_response,
            selected_commands=selected_commands,
            **kwargs,
        )
        logging.info(
            f"[run_stream] format_prompt took {_time.monotonic() - _t_fmt:.1f}s, "
            f"prompt tokens: {tokens}"
        )

        # Inject special CLI instructions if terminal command tools were requested
        if command_overrides:
            # Check for both legacy "Execute Terminal Command" and new "execute_terminal_command"
            has_terminal_command = any(
                tool.get("type") == "Execute Terminal Command"
                or (
                    tool.get("type") == "function"
                    and tool.get("function", {}).get("name")
                    == "execute_terminal_command"
                )
                for tool in command_overrides
            )
            if has_terminal_command:
                # Get the correct command name based on what's available
                terminal_cmd_name = (
                    "execute_terminal_command"
                    if "execute_terminal_command" in self._client_tools
                    else "Execute Terminal Command"
                )
                cli_instructions = f"""
## IMPORTANT: CLI Mode Instructions
You are interacting with a user through a command-line interface (CLI). The user is on their local machine.

**You MUST use the `{terminal_cmd_name}` command for ANY of these requests:**
- Listing files or directories (e.g., "list files", "show me what's here", "what files do I have")
- Creating, moving, copying, or deleting files/folders
- Checking system information (pwd, whoami, uname, etc.)
- Running build commands (npm, cargo, make, pip, etc.)
- Executing scripts or programs
- Git operations (status, log, diff, etc.)
- Package installation
- ANY terminal or shell operation

**Do NOT use workspace file commands like "Search Files" or "Read File" - those only work on the server, not the user's machine.**
**The `{terminal_cmd_name}` is the ONLY way to interact with the user's local filesystem.**

Example: If user says "list my files", use:
<execute>
<name>{terminal_cmd_name}</name>
<command>ls -la</command>
</execute>
"""
                formatted_prompt = f"{formatted_prompt}\n{cli_instructions}"

        # Anonymize AGiXT server URL
        if self.outputs in formatted_prompt:
            formatted_prompt = formatted_prompt.replace(
                self.outputs, f"http://localhost:7437/outputs/{self.agent.agent_id}"
            )

        # Log user input
        log_message = user_input if user_input != "" else formatted_prompt
        if log_user_input:
            c.log_interaction(role="USER", message=log_message)
            # Emit webhook event for user message
            await webhook_emitter.emit_event(
                event_type="conversation.message.received",
                data={
                    "conversation_id": c.get_conversation_id(),
                    "conversation_name": conversation_name,
                    "agent_name": self.agent_name,
                    "user": self.user,
                    "message": log_message,
                    "role": "USER",
                    "timestamp": datetime.now().isoformat(),
                },
                user_id=self.user,
                agent_id=self.agent.agent_id,
                agent_name=self.agent_name,
                company_id=self.agent.company_id,
            )

        # Get streaming response from the LLM
        _t_inf = _time.monotonic()
        logging.info(
            f"[run_stream] Pre-processing complete in {_t_inf - _stream_t0:.1f}s total. "
            f"Starting inference with {get_tokens(formatted_prompt)} token prompt."
        )
        try:
            stream = await self.agent.inference(
                prompt=formatted_prompt, use_smartest=use_smartest, stream=True
            )
        except HTTPException as e:
            # Re-raise HTTP exceptions (like 402 Payment Required) to propagate to endpoint
            logging.error(
                f"HTTP error during streaming inference: {e.status_code} - {e.detail}"
            )
            raise
        except Exception as e:
            # agent.inference() already attempted provider rotation internally.
            # If we reach here, all providers have been exhausted.
            logging.error(
                f"Error starting streaming inference (all providers exhausted): {e}"
            )
            yield {"type": "error", "content": str(e), "complete": True}
            return

        # Process the stream
        full_response = ""
        processed_thinking_ids = set()
        processed_speak_ids = set()  # Track processed speak tags for TTS filler speech
        in_answer = False
        answer_content = ""
        thinking_content = ""  # Track streamed thinking content
        reflection_content = ""  # Track streamed reflection content
        is_executing = False  # Flag to pause streaming during command execution
        remote_command_yielded = (
            False  # Flag to track if remote command was yielded (skip continuation)
        )
        # Track standalone steps message ID for updating in place
        standalone_steps_message_id = None
        standalone_steps_logged_ids = set()  # Track which step IDs have been logged
        # Incremental tag depth counters — updated per-token instead of re-scanning full response
        _thinking_depth = 0
        _reflection_depth = 0
        # Track reasoning_content field state (DeepSeek, OpenAI reasoning models)
        _in_reasoning_content = False
        # Track position where the currently open thinking block's content starts
        # Avoids O(n²) re-scanning of full_response on every token
        _thinking_content_start = -1
        # Track placeholder activity IDs for progressive streaming display
        # Created when thinking/reflection starts so the front-end has an activity
        # item to attach streaming content to (avoids content buffering invisibly)
        _thinking_placeholder_id = None
        _reflection_placeholder_id = None

        # Helper to iterate over stream (handles sync iterators from OpenAI library)
        async def iterate_stream(stream_obj, idle_timeout=180):
            """Iterate over stream, wrapping sync iteration if needed.

            Args:
                stream_obj: The stream object to iterate over.
                idle_timeout: Maximum seconds to wait between chunks before
                    raising a TimeoutError. Prevents infinite hangs when the
                    LLM inference stream stalls silently. Default: 180s.
            """
            import time as _iter_time

            # OpenAI library returns sync iterators, check if it's async or sync
            if hasattr(stream_obj, "__aiter__"):
                # Async iterator
                async for chunk in stream_obj:
                    yield chunk
            else:
                # Sync iterator - use asyncio.to_thread to run iteration without blocking
                import queue
                import threading

                chunk_queue = queue.Queue()
                done_event = threading.Event()
                error_holder = [None]

                def sync_iterator():
                    """Run the sync iterator in a separate thread."""
                    try:
                        chunk_count = 0
                        iterator = iter(stream_obj)
                        while True:
                            try:
                                chunk = next(iterator)
                                chunk_count += 1
                                chunk_queue.put(("chunk", chunk))
                            except StopIteration:
                                break
                        chunk_queue.put(("done", None))
                    except Exception as e:
                        logging.error(f"Stream iteration error: {e}", exc_info=True)
                        error_holder[0] = e
                        chunk_queue.put(("error", e))
                    finally:
                        done_event.set()

                # Start the sync iteration in a thread
                thread = threading.Thread(target=sync_iterator, daemon=True)
                thread.start()

                # Yield chunks as they arrive — use non-blocking get to avoid
                # blocking the event loop (which would stall SSE delivery)
                _last_chunk_time = _iter_time.monotonic()
                while True:
                    try:
                        msg_type, msg_data = chunk_queue.get_nowait()
                        _last_chunk_time = _iter_time.monotonic()
                        if msg_type == "chunk":
                            yield msg_data
                        elif msg_type == "done":
                            break
                        elif msg_type == "error":
                            raise msg_data
                    except queue.Empty:
                        # Short non-blocking sleep lets other coroutines run
                        # 1ms keeps latency low for smooth streaming
                        await asyncio.sleep(0.001)
                        # Check if thread died unexpectedly
                        if done_event.is_set() and chunk_queue.empty():
                            if error_holder[0]:
                                raise error_holder[0]
                            break
                        # Check for idle timeout — if no chunk arrives within
                        # idle_timeout seconds, the stream is likely stalled
                        if _iter_time.monotonic() - _last_chunk_time > idle_timeout:
                            logging.error(
                                f"[iterate_stream] Idle timeout after {idle_timeout}s "
                                f"with no chunks received. Stream appears stalled."
                            )
                            # Try to close the underlying stream to free resources
                            try:
                                if hasattr(stream_obj, "close"):
                                    stream_obj.close()
                                elif hasattr(stream_obj, "response") and hasattr(
                                    stream_obj.response, "close"
                                ):
                                    stream_obj.response.close()
                            except Exception:
                                pass
                            raise TimeoutError(
                                f"Stream idle timeout: no data received for {idle_timeout}s"
                            )

        try:
            chunk_count = 0
            # Track last processed position for incremental tag detection
            last_tag_check_pos = 0
            _first_token_logged = False
            _break_for_continuation = False

            async for chunk in iterate_stream(stream):
                # Check for cancellation periodically during streaming
                if chunk_count % 10 == 0:  # Check every 10 chunks to avoid overhead
                    self._check_cancelled()

                chunk_count += 1

                # Extract content from the chunk - handle different formats
                token = None
                if isinstance(chunk, str):
                    # Some providers return raw strings
                    token = chunk
                elif hasattr(chunk, "choices") and len(chunk.choices) > 0:
                    delta = chunk.choices[0].delta
                    if hasattr(delta, "content") and delta.content:
                        token = delta.content
                        # If we were receiving reasoning_content, close the thinking block
                        if _in_reasoning_content:
                            _in_reasoning_content = False
                            token = "</think>" + token
                    elif (
                        hasattr(delta, "reasoning_content") and delta.reasoning_content
                    ):
                        # Providers like DeepSeek/OpenAI send thinking in a separate field
                        token = delta.reasoning_content
                        if not _in_reasoning_content:
                            _in_reasoning_content = True
                            token = "<think>" + token

                if not token:
                    continue

                if not _first_token_logged:
                    _first_token_logged = True
                    logging.info(
                        f"[run_stream] First token arrived {_time.monotonic() - _t_inf:.1f}s after inference start, "
                        f"{_time.monotonic() - _stream_t0:.1f}s total from run_stream entry"
                    )

                full_response += token
                if "<answer>" in token.lower():
                    normalized_full_response = _RE_PROTOCOL_ANSWER_BOUNDARY.sub(
                        r"\1\n\2", full_response
                    )
                    if normalized_full_response != full_response:
                        full_response = normalized_full_response

                # Incremental tag depth tracking — only scan the new token
                # Count both <think> and <thinking> forms for model compatibility
                token_lower = token.lower()
                _prev_thinking_depth = _thinking_depth
                _thinking_depth += len(_RE_THINKING_OPEN.findall(token)) - len(
                    _RE_THINKING_CLOSE.findall(token)
                )
                # Track where the current thinking block content starts (O(1) per token)
                if _thinking_depth > 0 and _prev_thinking_depth == 0:
                    # Just entered thinking — find the end of the opening tag
                    last_open = None
                    for m in _RE_THINKING_OPEN.finditer(full_response):
                        last_open = m
                    _thinking_content_start = (
                        last_open.end() if last_open else len(full_response)
                    )
                    # Create placeholder activity so the front-end has an item
                    # to attach progressive streaming content to immediately
                    if not is_executing and _thinking_placeholder_id is None:
                        _thinking_placeholder_id = c.log_interaction(
                            role=self.agent_name,
                            message="[SUBACTIVITY][THOUGHT] Thinking...",
                        )
                elif _thinking_depth == 0 and _prev_thinking_depth > 0:
                    _thinking_content_start = -1
                _reflection_depth += token_lower.count(
                    "<reflection>"
                ) - token_lower.count("</reflection>")
                in_thinking_or_reflection = _thinking_depth > 0 or _reflection_depth > 0

                # Use is_inside_top_level_answer for proper answer detection
                # This handles cases where <thinking> appears INSIDE <answer> blocks
                in_answer = is_inside_top_level_answer(full_response)

                # Detect leaked <interaction> XML and convert to proper <execute>
                # The web browsing extension uses <interaction> as an internal
                # browser control protocol.  When the outer agent outputs this
                # format directly, convert it so the execute machinery handles it.
                if "</interaction>" in full_response.lower():
                    converted = _convert_interaction_to_execute(full_response)
                    if converted != full_response:
                        full_response = converted

                # Check for execute tag completion - allow commands inside thinking, reflection, and answer blocks
                # Execute tags should be processed regardless of nesting to support agentic workflows
                for match in re.finditer(_RE_EXECUTE_TAG, full_response):
                    execute_end = match.end()
                    # Note: We no longer skip execute tags inside thinking/reflection blocks
                    # The agent may legitimately execute commands while thinking through a problem

                    # Check if inside answer block - if so, strip answer tags first
                    is_inside_answer = is_inside_top_level_answer(
                        full_response, match.start()
                    )
                    if is_inside_answer:
                        # Strip answer tags to allow command execution within answer phase
                        full_response = full_response.replace("</answer>", "").replace(
                            "<answer>", ""
                        )
                        in_answer = False
                        answer_content = ""

                    # This is a top-level execute - check if we've processed it
                    execute_content = match.group(0)
                    execute_id = f"execute:{hash(execute_content)}"
                    if execute_id in processed_thinking_ids:
                        continue

                    processed_thinking_ids.add(execute_id)

                    # Execute commands and STOP inference
                    if (
                        "{COMMANDS}" in unformatted_prompt
                        and "disable_commands" not in kwargs
                    ):
                        is_executing = True

                        # Truncate at </execute> to discard hallucinated output
                        truncated_response = full_response[:execute_end]
                        full_response = truncated_response
                        self.response = full_response

                        # Create a queue to receive remote command requests
                        remote_command_queue = asyncio.Queue()

                        async def remote_command_callback(remote_cmd):
                            """Callback that yields remote command request to SSE stream."""
                            # Put the remote command request in the queue
                            await remote_command_queue.put(remote_cmd)
                            # Return a placeholder - the actual result will come from CLI
                            return f"[REMOTE COMMAND QUEUED] Waiting for client-side execution.\nRequest ID: {remote_cmd.get('request_id', 'unknown')}"

                        # Execute the command with callback
                        # Run execution_agent concurrently with keepalive yields
                        # to prevent SSE stream timeout during long-running commands
                        execution_task = asyncio.create_task(
                            self.execution_agent(
                                conversation_name=conversation_name,
                                conversation_id=conversation_id,
                                thinking_id=thinking_id,
                                remote_command_callback=remote_command_callback,
                            )
                        )
                        # Yield keepalive events every 10 seconds while waiting
                        while not execution_task.done():
                            try:
                                await asyncio.wait_for(
                                    asyncio.shield(execution_task),
                                    timeout=10.0,
                                )
                            except asyncio.TimeoutError:
                                # Task still running, yield keepalive
                                yield {
                                    "type": "keepalive",
                                    "content": "",
                                    "complete": False,
                                }
                            except Exception:
                                break
                        # Ensure task result is retrieved (re-raises exceptions)
                        if execution_task.done():
                            execution_task.result()

                        # Yield any remote command requests that were queued
                        while not remote_command_queue.empty():
                            remote_cmd = await remote_command_queue.get()
                            remote_command_yielded = (
                                True  # Mark that we yielded a remote command
                            )
                            yield {
                                "type": "remote_command_request",
                                "content": remote_cmd,
                                "complete": True,
                            }

                        full_response = self.response
                        is_executing = False
                        _break_for_continuation = True
                        break  # Break inner for-match loop

                # After code execution, break the outer streaming loop immediately.
                # The model's remaining stream tokens are generated without knowledge
                # of the actual execution output, so they're stale/hallucinated.
                # Continuation logic will make a fresh inference with real output.
                if _break_for_continuation:
                    break

                # Process completed thinking/reflection tags for logging
                # This also consolidates any adjacent <step>, <count>, <reward> tags
                # that appear between thinking/reflection blocks
                for tag_name in ["thinking", "reflection"]:
                    # Use regex that matches both <think> and <thinking> forms
                    if tag_name == "thinking":
                        tag_pattern = r"<think(?:ing)?>(.*?)</think(?:ing)?>"
                    else:
                        tag_pattern = f"<{tag_name}>(.*?)</{tag_name}>"
                    for match in re.finditer(
                        tag_pattern, full_response, re.DOTALL | re.IGNORECASE
                    ):
                        content = match.group(1).strip()
                        tag_end_pos = match.end()
                        tag_id = f"{tag_name}:{hash(content)}"
                        if tag_id in processed_thinking_ids or not content:
                            continue

                        processed_thinking_ids.add(tag_id)

                        # Clean content - remove any tag mentions
                        content = re.sub(
                            r"</?(?:answer|execute|output|step|reward|count)>",
                            "",
                            content,
                        )
                        content = re.sub(
                            r"<reward>.*?</reward>", "", content, flags=re.DOTALL
                        )
                        content = re.sub(
                            r"<count>.*?</count>", "", content, flags=re.DOTALL
                        )
                        content = re.sub(r"\n\s*\n\s*\n", "\n\n", content).strip()

                        # Now look for any <step>, <reward>, <count> tags that come AFTER this
                        # thinking/reflection block but BEFORE the next thought, reflection, execute, or answer
                        trailing_content = full_response[tag_end_pos:]

                        # Find where the next major tag starts (thinking, reflection, execute, answer)
                        next_major_tag = re.search(
                            r"<(thinking|reflection|execute|answer)>",
                            trailing_content,
                            re.IGNORECASE,
                        )
                        trailing_section = (
                            trailing_content[: next_major_tag.start()]
                            if next_major_tag
                            else trailing_content
                        )

                        # Collect any standalone step content from the trailing section
                        step_additions = []
                        for step_match in re.finditer(
                            r"<step>(.*?)</step>",
                            trailing_section,
                            re.DOTALL | re.IGNORECASE,
                        ):
                            step_content = step_match.group(1).strip()
                            step_id = f"step:{hash(step_content)}"
                            if step_id not in processed_thinking_ids and step_content:
                                processed_thinking_ids.add(step_id)
                                # Clean step content
                                cleaned_step = re.sub(
                                    r"<reward>.*?</reward>",
                                    "",
                                    step_content,
                                    flags=re.DOTALL,
                                )
                                cleaned_step = re.sub(
                                    r"<count>.*?</count>",
                                    "",
                                    cleaned_step,
                                    flags=re.DOTALL,
                                )
                                cleaned_step = cleaned_step.strip()
                                if cleaned_step:
                                    step_additions.append(cleaned_step)

                        # Collect any standalone reward/count content (just mark as processed, don't add to content)
                        for reward_match in re.finditer(
                            r"<reward>(.*?)</reward>",
                            trailing_section,
                            re.DOTALL | re.IGNORECASE,
                        ):
                            reward_id = f"reward:{hash(reward_match.group(1).strip())}"
                            processed_thinking_ids.add(reward_id)

                        for count_match in re.finditer(
                            r"<count>(.*?)</count>",
                            trailing_section,
                            re.DOTALL | re.IGNORECASE,
                        ):
                            count_id = f"count:{hash(count_match.group(1).strip())}"
                            processed_thinking_ids.add(count_id)

                        # Append step content to the thinking/reflection content
                        if step_additions:
                            step_text = "\n".join(f"- {s}" for s in step_additions)
                            content = f"{content}\n\n**Steps:**\n{step_text}"

                        if content:
                            if tag_name == "thinking":
                                log_msg = f"[SUBACTIVITY][THOUGHT] {content}"
                            else:
                                log_msg = f"[SUBACTIVITY][REFLECTION] {content}"

                            if not is_executing:
                                # Update placeholder if we created one, otherwise create new
                                if tag_name == "thinking" and _thinking_placeholder_id:
                                    c.update_message_by_id(
                                        _thinking_placeholder_id, log_msg
                                    )
                                    _thinking_placeholder_id = (
                                        None  # Reset for next block
                                    )
                                elif (
                                    tag_name == "reflection"
                                    and _reflection_placeholder_id
                                ):
                                    c.update_message_by_id(
                                        _reflection_placeholder_id, log_msg
                                    )
                                    _reflection_placeholder_id = None
                                else:
                                    c.log_interaction(
                                        role=self.agent_name, message=log_msg
                                    )

                            yield {
                                "type": tag_name,
                                "content": content,
                                "complete": True,
                            }

                # Process standalone <step> tags that appear OUTSIDE any thinking/reflection block
                # These should be accumulated and logged as a SINGLE combined message
                # Use update-in-place to consolidate as steps arrive
                step_pattern = r"<step>(.*?)</step>"
                all_standalone_steps = []
                for match in re.finditer(
                    step_pattern, full_response, re.DOTALL | re.IGNORECASE
                ):
                    step_content = match.group(1).strip()
                    step_start = match.start()
                    step_id = f"step:{hash(step_content)}"

                    if not step_content:
                        continue

                    # Check if this step is inside thinking/reflection
                    text_before = full_response[:step_start]
                    if (
                        len(_RE_THINKING_OPEN.findall(text_before))
                        - len(_RE_THINKING_CLOSE.findall(text_before))
                        > 0
                        or text_before.lower().count("<reflection>")
                        - text_before.lower().count("</reflection>")
                        > 0
                    ):
                        continue  # Skip steps inside thinking/reflection

                    # Check if inside answer block
                    if is_inside_top_level_answer(full_response, step_start):
                        continue  # Skip steps inside answer

                    # Clean content
                    cleaned_step = re.sub(
                        r"<reward>.*?</reward>", "", step_content, flags=re.DOTALL
                    )
                    cleaned_step = re.sub(
                        r"<count>.*?</count>", "", cleaned_step, flags=re.DOTALL
                    )
                    cleaned_step = re.sub(r"\n\s*\n\s*\n", "\n\n", cleaned_step).strip()

                    if cleaned_step:
                        all_standalone_steps.append((step_id, cleaned_step))

                # If we have standalone steps, log/update as a single combined message
                if all_standalone_steps and not is_executing:
                    # Check if we have NEW steps to add (not already in logged set)
                    new_step_ids = set(s[0] for s in all_standalone_steps)
                    if new_step_ids != standalone_steps_logged_ids:
                        # Build combined message from ALL steps
                        combined_steps = "\n".join(
                            f"- {step[1]}" for step in all_standalone_steps
                        )
                        log_msg = f"[SUBACTIVITY][THOUGHT] {combined_steps}"

                        if standalone_steps_message_id is None:
                            # Create new message
                            standalone_steps_message_id = c.log_interaction(
                                role=self.agent_name, message=log_msg
                            )
                        else:
                            # Update existing message
                            c.update_message_by_id(standalone_steps_message_id, log_msg)

                        # Update tracking set
                        standalone_steps_logged_ids = new_step_ids.copy()

                        # Mark all step IDs as processed
                        for step_id, _ in all_standalone_steps:
                            processed_thinking_ids.add(step_id)

                        yield {
                            "type": "thinking",
                            "content": combined_steps,
                            "complete": True,
                        }

                # Mark any remaining standalone <reward> and <count> tags as processed
                # but don't create separate subactivities for them
                reward_pattern = r"<reward>(.*?)</reward>"
                for match in re.finditer(
                    reward_pattern, full_response, re.DOTALL | re.IGNORECASE
                ):
                    reward_content = match.group(1).strip()
                    reward_id = f"reward:{hash(reward_content)}"
                    if reward_id not in processed_thinking_ids:
                        processed_thinking_ids.add(reward_id)
                        # Don't yield or log - these are metadata, not separate activities

                count_pattern = r"<count>(.*?)</count>"
                for match in re.finditer(
                    count_pattern, full_response, re.DOTALL | re.IGNORECASE
                ):
                    count_content = match.group(1).strip()
                    count_id = f"count:{hash(count_content)}"
                    if count_id not in processed_thinking_ids:
                        processed_thinking_ids.add(count_id)

                # Process <speak> tags for TTS filler speech - yield immediately for speech
                # These are short filler phrases like "Let me check on that" that should be
                # spoken immediately while the model continues thinking
                # LIMIT: Only allow 2 speak tags max (quick ack + description) to prevent
                # the model from generating excessive filler speech
                MAX_SPEAK_TAGS = 2
                speak_pattern = r"<speak>(.*?)</speak>"
                for match in re.finditer(
                    speak_pattern, full_response, re.DOTALL | re.IGNORECASE
                ):
                    speak_content = match.group(1).strip()
                    speak_id = f"speak:{hash(speak_content)}"
                    if speak_id not in processed_speak_ids and speak_content:
                        # Stop processing speak tags after we've hit the limit
                        if len(processed_speak_ids) >= MAX_SPEAK_TAGS:
                            break
                        processed_speak_ids.add(speak_id)
                        # Yield speak event for immediate TTS playback
                        # Don't log to conversation - these are transient filler speech
                        yield {
                            "type": "speak",
                            "content": speak_content,
                            "complete": True,
                        }

                # Progressive streaming of thinking content (stream as it's generated)
                if (
                    _thinking_depth > 0
                    and not is_executing
                    and _thinking_content_start >= 0
                ):
                    new_thinking = full_response[_thinking_content_start:]
                    # Handle partial closing tags at the end of the buffer
                    partial_close = re.search(
                        r"</?t?h?i?n?k?i?n?g?>?$", new_thinking, re.IGNORECASE
                    )
                    if partial_close and partial_close.group().startswith("<"):
                        new_thinking = new_thinking[: partial_close.start()]

                    # Stream the delta
                    if len(new_thinking) > len(thinking_content):
                        delta = new_thinking[len(thinking_content) :]
                        if delta and not re.match(r"^\s*<[a-zA-Z]", delta):
                            yield {
                                "type": "thinking_stream",
                                "content": delta,
                                "complete": False,
                            }
                        thinking_content = new_thinking
                elif _thinking_depth == 0 and thinking_content:
                    # Reset tracking when we exit thinking
                    thinking_content = ""

                # Progressive streaming of reflection content (stream as it's generated)
                if _reflection_depth > 0 and not is_executing:
                    # Find the currently open (incomplete) reflection tag
                    reflection_start = None
                    for match in re.finditer(
                        r"<reflection>", full_response, re.IGNORECASE
                    ):
                        open_pos = match.end()
                        text_after = full_response[open_pos:]
                        opens_after = len(
                            re.findall(r"<reflection>", text_after, re.IGNORECASE)
                        )
                        closes_after = len(
                            re.findall(r"</reflection>", text_after, re.IGNORECASE)
                        )
                        if closes_after <= opens_after:
                            reflection_start = open_pos

                    if reflection_start is not None:
                        new_reflection = full_response[reflection_start:]
                        # Truncate at </reflection> if present
                        close_match = re.search(
                            r"</reflection>", new_reflection, re.IGNORECASE
                        )
                        if close_match:
                            new_reflection = new_reflection[: close_match.start()]
                        # Handle partial closing tags
                        partial_close = re.search(
                            r"</?r?e?f?l?e?c?t?i?o?n?>?$", new_reflection, re.IGNORECASE
                        )
                        if partial_close and partial_close.group().startswith("<"):
                            new_reflection = new_reflection[: partial_close.start()]
                        # Clean up
                        if new_reflection.startswith(">"):
                            new_reflection = new_reflection[1:]

                        # Stream the delta
                        if len(new_reflection) > len(reflection_content):
                            delta = new_reflection[len(reflection_content) :]
                            if delta and not re.match(r"^\s*<[a-zA-Z]", delta):
                                yield {
                                    "type": "reflection_stream",
                                    "content": delta,
                                    "complete": False,
                                }
                            reflection_content = new_reflection
                    else:
                        # Reset tracking when we exit reflection
                        reflection_content = ""

                # Reset thinking/reflection content tracking when tags close
                if _thinking_depth == 0 and thinking_content:
                    thinking_content = ""
                if _reflection_depth == 0 and reflection_content:
                    reflection_content = ""

                # Yield answer tokens for streaming
                if in_answer and not is_executing:
                    # Find the TOP-LEVEL answer tag (not inside thinking/reflection)
                    # Use real tag detection to avoid false positives from mentions in text
                    answer_start = None
                    for match in re.finditer(r"<answer>", full_response, re.IGNORECASE):
                        open_pos = match.start()

                        # Skip if this is not a real tag (just mentioned in text)
                        if not is_real_answer_tag(full_response, open_pos):
                            continue

                        text_before = full_response[:open_pos]
                        thinking_depth = len(
                            _RE_THINKING_OPEN.findall(text_before)
                        ) - len(_RE_THINKING_CLOSE.findall(text_before))
                        reflection_depth = len(
                            re.findall(r"<reflection>", text_before, re.IGNORECASE)
                        ) - len(
                            re.findall(r"</reflection>", text_before, re.IGNORECASE)
                        )
                        if thinking_depth == 0 and reflection_depth == 0:
                            answer_start = match.end()
                            break  # Use the first top-level answer

                    if answer_start is not None:
                        new_answer = full_response[answer_start:]

                        # Debug: log what we're extracting
                        if len(new_answer) < 100:
                            logging.debug(
                                f"[answer_extract] raw new_answer: {repr(new_answer)}"
                            )

                        # Check if </answer> appears - if so, truncate before it
                        close_tag_match = re.search(
                            r"</answer>", new_answer, re.IGNORECASE
                        )
                        if close_tag_match:
                            new_answer = new_answer[: close_tag_match.start()]
                        else:
                            # Also check for partial closing tags at the end (e.g., "</", "</ans", etc.)
                            # Match any partial </answer> pattern
                            partial_close_match = re.search(
                                r"</?a?n?s?w?e?r?>?$", new_answer, re.IGNORECASE
                            )
                            if (
                                partial_close_match
                                and partial_close_match.group().startswith("<")
                            ):
                                new_answer = new_answer[: partial_close_match.start()]

                        # Clean any leading ">" that might be from the opening <answer> tag
                        # This can happen if we start extracting mid-tag
                        if new_answer.startswith(">"):
                            new_answer = new_answer[1:]
                        # Also strip leading whitespace after tag cleanup
                        new_answer = new_answer.lstrip()

                        # Clean any trailing partial tag fragments like "answer>" or just ">"
                        new_answer = re.sub(
                            r"a?n?s?w?e?r?>$", "", new_answer, flags=re.IGNORECASE
                        )
                        # Also strip trailing ">" that might be from partial tag
                        if new_answer.endswith(">"):
                            new_answer = new_answer[:-1]

                        # Clean all internal tags from answer content before yielding
                        # Complete tags
                        cleaned_new_answer = re.sub(
                            r"<think(?:ing)?>.*?</think(?:ing)?>",
                            "",
                            new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        cleaned_new_answer = re.sub(
                            r"<reflection>.*?</reflection>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        cleaned_new_answer = re.sub(
                            r"<step>.*?</step>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        cleaned_new_answer = re.sub(
                            r"<reward>.*?</reward>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        cleaned_new_answer = re.sub(
                            r"<count>.*?</count>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        # Remove <final> tags - model-generated internal markers
                        cleaned_new_answer = re.sub(
                            r"<final>.*?</final>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        # Remove <speak> tags - TTS filler that shouldn't appear in text answer
                        cleaned_new_answer = re.sub(
                            r"<speak>.*?</speak>",
                            "",
                            cleaned_new_answer,
                            flags=re.DOTALL | re.IGNORECASE,
                        )
                        # Partial/unclosed opening tags at end (tag started but not closed)
                        cleaned_new_answer = re.sub(
                            r"<(thinking|reflection|step|reward|count|final)>[^<]*$",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Partial opening tags at end (tag not fully written yet)
                        # Match "<" followed by at least ONE character that looks like the start of a tag
                        # This handles partial tags like "<th", "<ref", "</thi", etc.
                        # NOTE: Do NOT match bare "<" alone - that would strip legitimate angle brackets
                        cleaned_new_answer = re.sub(
                            r"</?[a-zA-Z][a-zA-Z]*$",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Orphaned closing tags (closing tag without opening)
                        cleaned_new_answer = re.sub(
                            r"</(thinking|reflection|step|reward|count|final|speak)>",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Also clean orphaned tag fragments WITHOUT the leading "<"
                        # This handles cases where chunk boundaries split tags like "<reflection>"
                        # into "<" and "reflection>" with the "<" getting cleaned separately
                        cleaned_new_answer = re.sub(
                            r"(?<![<a-zA-Z])/?(?:thinking|reflection|step|reward|count|final|speak)>",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Partial closing tags at start (leftover from previous chunk)
                        cleaned_new_answer = re.sub(
                            r"^[^<]*</(thinking|reflection|step|reward|count|final)>",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Clean any remaining partial tag fragments at start
                        cleaned_new_answer = re.sub(
                            r"^(thinking|reflection|step|reward|count|final)?>",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Clean answer tag fragments at start (e.g., ">" or "nswer>" leftover)
                        cleaned_new_answer = re.sub(
                            r"^a?n?s?w?e?r?>[\s]*",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Clean lone ">" at start (leftover from answer tag)
                        if cleaned_new_answer.startswith(">"):
                            cleaned_new_answer = cleaned_new_answer[1:].lstrip()
                        # Clean answer tag fragments at end (e.g., "</ans" or "answer>")
                        cleaned_new_answer = re.sub(
                            r"</?a?n?s?w?e?r?>?[\s]*$",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Also clean just "answer>" at end (without the </)
                        cleaned_new_answer = re.sub(
                            r"a?n?s?w?e?r?>[\s]*$",
                            "",
                            cleaned_new_answer,
                            flags=re.IGNORECASE,
                        )
                        # Strip any remaining whitespace
                        cleaned_new_answer = cleaned_new_answer.strip()

                        if len(cleaned_new_answer) > len(answer_content):
                            delta = cleaned_new_answer[len(answer_content) :]
                            # Skip if it looks like an opening tag pattern (thinking, reflection, etc.)
                            # Also skip orphaned tag fragments like "reflection>" without the leading "<"
                            skip_delta = re.match(r"^\s*<[a-zA-Z]", delta) or re.match(
                                r"^\s*/?(?:thinking|reflection|step|reward|count)>",
                                delta,
                                re.IGNORECASE,
                            )
                            if not skip_delta:
                                if delta:
                                    yield {
                                        "type": "answer",
                                        "content": delta,
                                        "complete": False,
                                    }
                            answer_content = cleaned_new_answer

                # Check for complete answer - stop streaming early if we have a full answer
                # This prevents the model from continuing to think after providing the answer
                # IMPORTANT: This check must be AFTER the streaming logic so the final answer content is yielded
                if has_complete_answer(full_response):
                    break

        except Exception as e:
            logging.error(f"Error during streaming: {e}")
            import traceback

            logging.error(traceback.format_exc())

        # Cancel/close the old LLM stream so the inference slot is freed immediately.
        # Without this, the iterate_stream daemon thread keeps consuming tokens from
        # the old generation, blocking the single ezlocalai slot.  The continuation
        # logic below needs that slot for a fresh inference call.
        try:
            if hasattr(stream, "close"):
                stream.close()
            elif hasattr(stream, "response") and hasattr(stream.response, "close"):
                stream.response.close()
        except Exception:
            pass  # Best-effort cleanup

        # Store the full response
        self.response = full_response

        # If the stream was stopped by a stop sequence (e.g., "</execute>"),
        # the closing tag won't be in the output. Detect unclosed <execute> tags
        # and append the closing tag so the execution logic can process them.
        _lower_resp = full_response.lower()
        _last_exec_open = _lower_resp.rfind("<execute>")
        _stop_sequence_execute_appended = False
        if _last_exec_open != -1:
            _last_exec_close = _lower_resp.rfind("</execute>")
            if _last_exec_close < _last_exec_open:
                # Unclosed <execute> tag — stop sequence likely fired
                full_response += "</execute>"
                self.response = full_response
                _stop_sequence_execute_appended = True
                logging.info(
                    "[run_stream] Appended </execute> closing tag (stop sequence detected)"
                )

        if (
            _stop_sequence_execute_appended
            and "{COMMANDS}" in unformatted_prompt
            and "disable_commands" not in kwargs
        ):
            logging.info(
                "[run_stream] Executing initial command completed by stop sequence"
            )
            stop_remote_queue = asyncio.Queue()

            async def stop_remote_callback(remote_cmd):
                await stop_remote_queue.put(remote_cmd)
                return (
                    "[REMOTE COMMAND QUEUED] Waiting for client-side execution.\n"
                    f"Request ID: {remote_cmd.get('request_id', 'unknown')}"
                )

            stop_execution_task = asyncio.create_task(
                self.execution_agent(
                    conversation_name=conversation_name,
                    conversation_id=conversation_id,
                    thinking_id=thinking_id,
                    remote_command_callback=stop_remote_callback,
                )
            )
            while not stop_execution_task.done():
                try:
                    await asyncio.wait_for(
                        asyncio.shield(stop_execution_task),
                        timeout=10.0,
                    )
                except asyncio.TimeoutError:
                    yield {
                        "type": "keepalive",
                        "content": "",
                        "complete": False,
                    }
                except Exception:
                    break
            if stop_execution_task.done():
                stop_execution_task.result()

            while not stop_remote_queue.empty():
                remote_cmd = await stop_remote_queue.get()
                remote_command_yielded = True
                yield {
                    "type": "remote_command_request",
                    "content": remote_cmd,
                    "complete": True,
                }

            full_response = self.response
            _break_for_continuation = True

        # Skip continuation logic if a remote command was yielded
        # The CLI will handle the command execution and submit the result
        # The next user interaction will pick up from there
        if remote_command_yielded:
            # Yield a final completion indicator
            yield {
                "type": "remote_command_pending",
                "content": "Waiting for remote command execution...",
                "complete": True,
            }
            return

        # Continuation logic: Handle execution outputs and incomplete answers
        # This matches the non-streaming behavior where we inject output and run inference again
        # No max iteration limit — the agent should work through problems taking as many
        # steps and as much time as needed, just like a human would.
        continuation_count = 0
        _continuation_iter_lengths = (
            []
        )  # Track per-iteration response lengths for stuck detection (non-execution only)
        _previous_iteration_executed = (
            _break_for_continuation  # Carry forward from initial stream execution
        )
        _no_answer_non_exec_streak = (
            0  # Track consecutive non-exec iterations without <answer> tags
        )
        _consecutive_dedup_iterations = (
            0  # Track consecutive iterations where ALL commands were deduplicated
        )
        _incomplete_answer_non_exec_count = (
            0  # Track consecutive non-exec iterations with incomplete answer
        )
        _final_answer_review_enabled = str(
            getenv("FINAL_ANSWER_REVIEW_ENABLED", "false")
        ).lower() in ("1", "true", "yes", "on")
        _final_answer_review_server = (
            getenv("ABILITY_SELECTION_SERVER") if _final_answer_review_enabled else None
        )
        _final_answer_review_model = getenv(
            "ABILITY_SELECTION_MODEL", "unsloth/Qwen3.5-4B-GGUF"
        )
        try:
            _final_answer_review_max_attempts = int(
                getenv("FINAL_ANSWER_REVIEW_MAX_ATTEMPTS", "3")
            )
        except (TypeError, ValueError):
            _final_answer_review_max_attempts = 3
        _final_answer_review_max_attempts = max(0, _final_answer_review_max_attempts)
        if not _final_answer_review_enabled:
            _final_answer_review_max_attempts = 0
        try:
            _final_answer_review_max_chars = int(
                getenv("FINAL_ANSWER_REVIEW_MAX_CHARS", "600000")
            )
        except (TypeError, ValueError):
            _final_answer_review_max_chars = 600000
        _final_answer_review_max_chars = max(0, _final_answer_review_max_chars)
        _answer_review_attempts = 0
        _answer_review_approved_hash = None
        _answer_review_rejected = False
        _answer_review_rejected_hash = None
        _answer_review_feedback = ""
        _continuation_recovery_feedback = ""
        _continuation_recovery_count = 0
        _last_recovered_answer_hash = None
        _last_recovered_answer_text = ""
        _non_exec_continuation_count = 0
        try:
            _max_non_exec_continuations = int(
                getenv("CONTINUATION_MAX_NON_EXEC_ITERATIONS", "40")
            )
        except (TypeError, ValueError):
            _max_non_exec_continuations = 40
        _max_non_exec_continuations = max(8, _max_non_exec_continuations)
        try:
            _max_recovery_attempts = int(
                getenv("CONTINUATION_MAX_RECOVERY_ATTEMPTS", "12")
            )
        except (TypeError, ValueError):
            _max_recovery_attempts = 12
        _max_recovery_attempts = max(3, _max_recovery_attempts)
        if not _final_answer_review_enabled:
            logging.info(
                "[run_stream] Final answer review gate disabled; complete answers "
                "will finalize without ability-model rejection."
            )

        # Track the length of processed content to detect new executions
        processed_length = len(self.response)

        # Snapshot current user message IDs so we can detect new ones mid-task
        _known_user_msg_ids = set()
        try:
            _snap_conv = Conversations(
                conversation_name=conversation_name,
                user=self.user,
                conversation_id=conversation_id,
            )
            _snap_data = _snap_conv.get_conversation(limit=1000, page=1)
            for _msg in _snap_data.get("interactions", []):
                if str(_msg.get("role", "")).lower() == "user":
                    _known_user_msg_ids.add(str(_msg.get("id", "")))
        except Exception:
            pass

        logging.info(
            f"[run_stream] Entering continuation logic. Response length: {len(self.response)}. "
            f"has_complete_answer: {has_complete_answer(self.response)}. "
            f"remote_command_yielded: {remote_command_yielded}"
        )

        # Use has_complete_answer() to properly check for complete answer blocks
        # This handles edge cases like <thinking> inside <answer> tags
        while True:
            _response_has_complete_answer = has_complete_answer(self.response)
            _current_answer_hash = None
            if _response_has_complete_answer:
                _current_answer_for_review = extract_top_level_answer(self.response)
                if not _current_answer_for_review:
                    _current_answer_for_review = self.response
                _current_answer_for_review = _current_answer_for_review.strip()
                _current_answer_hash = str(hash(_current_answer_for_review))

                if (
                    _answer_review_rejected
                    and _current_answer_hash != _answer_review_rejected_hash
                ):
                    _answer_review_rejected = False
                    _answer_review_feedback = ""

                if not _answer_review_rejected:
                    if _answer_review_approved_hash == _current_answer_hash:
                        break
                    if (
                        not _final_answer_review_server
                        or _final_answer_review_max_attempts <= 0
                    ):
                        break
                    if _answer_review_attempts >= _final_answer_review_max_attempts:
                        logging.warning(
                            "[run_stream] Final answer review attempt cap reached; "
                            "continuing with self-healing feedback instead of "
                            "allowing an unapproved answer."
                        )
                        _answer_review_rejected = True
                        _answer_review_rejected_hash = _current_answer_hash
                        _answer_review_feedback = (
                            "The final answer review has not approved the answer "
                            f"after {_final_answer_review_max_attempts} review "
                            "attempts. Continue working from the review feedback "
                            "and prior outputs; do not end the request until a new "
                            "answer satisfies the completion criteria."
                        )
                        _answer_review_attempts = 0
                        continue

                    try:
                        _answer_review_attempts += 1
                        _review_context = self.compress_response_for_continuation(
                            self.response,
                            max_output_lines=120,
                            max_thinking_chars=1000,
                        )
                        _review_prompt_base, _, _ = await self.format_prompt(
                            user_input=user_input,
                            top_results=int(context_results),
                            conversation_results=conversation_results,
                            prompt=prompt,
                            prompt_category=prompt_category,
                            conversation_name=conversation_name,
                            websearch=websearch,
                            vision_response=vision_response,
                            selected_commands=self._selected_commands,
                            **kwargs,
                        )
                        if self.outputs in _review_prompt_base:
                            _review_prompt_base = _review_prompt_base.replace(
                                self.outputs,
                                f"http://localhost:7437/outputs/{self.agent.agent_id}",
                            )
                        if (
                            _final_answer_review_max_chars > 0
                            and len(_review_prompt_base)
                            > _final_answer_review_max_chars
                        ):
                            _review_half = _final_answer_review_max_chars // 2
                            _review_prompt_base = (
                                _review_prompt_base[:_review_half]
                                + "\n\n...[mandatory context truncated for review]...\n\n"
                                + _review_prompt_base[-_review_half:]
                            )

                        _answer_review_prompt = f"""You are the final-answer completion gate for an autonomous AGiXT agent.

Review whether the candidate <answer> is allowed to end the current inference request.
The system has already confirmed the candidate came from a complete top-level <answer>...</answer> block.
Do NOT reject solely because of answer tag formatting; judge whether the content actually completes the task.
Brief answers can be correct for simple requests. Do not require hidden thinking, tool use, extra explanation, or length unless the user's request requires it.
Before approving, actively extract the mandatory-context rules that apply to this user request. Treat explicit MUST, NEVER, DO NOT, REQUIRED, mandatory, persona, quality, safety, and scoring rules as hard approval criteria.
Reject if the candidate answer violates any applicable hard rule, overstates evidence the context says to caveat, ignores a required caveat, fabricates certainty, skips required verification, or presents a result the mandatory context says must be excluded or de-emphasized.

Approve only if ALL success criteria are met:
1. The answer follows the mandatory context, system instructions, persona, and user-specific rules.
2. The answer directly resolves the user's latest request, not merely a status update or promise to keep working.
3. The answer is complete, specific, well reasoned, and includes concrete results from any work already performed.
4. The answer does not skip required actions that the agent could still complete autonomously.
5. For coding or tool tasks, when applicable, the answer reflects actual completed work and verification, or clearly states exact blockers.

If any criterion is not met, reject the answer and explain what the agent must do next.
Return ONLY compact JSON with this shape:
{{"approved": true, "reason": "short reason", "missing": [], "instructions": ""}}
or
{{"approved": false, "reason": "short reason", "missing": ["specific missing item"], "instructions": "what the agent should do next"}}

## Mandatory Context And User Request
{_review_prompt_base}

## Assistant Work So Far
{_review_context}

## Candidate Final Answer Block
<answer>
{_current_answer_for_review}
</answer>
"""
                        _review_response = await _ability_selection_inference(
                            server_url=_final_answer_review_server,
                            model=_final_answer_review_model,
                            prompt=_answer_review_prompt,
                        )
                        _review_text = str(_review_response or "").strip()
                        _review_approved = True if not _review_text else False
                        _review_feedback_parts = []

                        if _review_text:
                            _review_json = {}
                            _review_json_match = re.search(
                                r"\{.*\}", _review_text, re.DOTALL
                            )
                            if _review_json_match:
                                try:
                                    _review_json = json.loads(
                                        _review_json_match.group(0)
                                    )
                                except Exception:
                                    _review_json = {}

                            if _review_json:
                                _approved_value = _review_json.get("approved", False)
                                if isinstance(_approved_value, bool):
                                    _review_approved = _approved_value
                                else:
                                    _review_approved = str(
                                        _approved_value
                                    ).strip().lower() in (
                                        "true",
                                        "yes",
                                        "approved",
                                        "approve",
                                        "pass",
                                    )
                                _reason = str(_review_json.get("reason", "")).strip()
                                _instructions = str(
                                    _review_json.get("instructions", "")
                                ).strip()
                                _missing = _review_json.get("missing", [])
                                if _reason:
                                    _review_feedback_parts.append(_reason)
                                if isinstance(_missing, list):
                                    _missing_text = "; ".join(
                                        str(item).strip()
                                        for item in _missing
                                        if str(item).strip()
                                    )
                                else:
                                    _missing_text = str(_missing).strip()
                                if _missing_text:
                                    _review_feedback_parts.append(
                                        f"Missing: {_missing_text}"
                                    )
                                if _instructions:
                                    _review_feedback_parts.append(
                                        f"Next: {_instructions}"
                                    )
                            else:
                                _review_lower = _review_text.lower()
                                _review_approved = (
                                    "approved" in _review_lower
                                    and "not approved" not in _review_lower
                                    and "reject" not in _review_lower
                                    and "false" not in _review_lower[:120]
                                )
                                _review_feedback_parts.append(_review_text)

                        if _review_approved:
                            _answer_review_approved_hash = _current_answer_hash
                            logging.info(
                                f"[run_stream] Final answer review approved attempt "
                                f"{_answer_review_attempts}."
                            )
                            break

                        _answer_review_rejected = True
                        _answer_review_rejected_hash = _current_answer_hash
                        _answer_review_feedback = "\n".join(
                            _review_feedback_parts
                        ).strip()
                        if not _answer_review_feedback:
                            _answer_review_feedback = (
                                "The candidate final answer did not satisfy the "
                                "completion criteria. Continue thinking and acting "
                                "until the user's request is fully completed."
                            )
                        _answer_review_feedback = _answer_review_feedback[:2500]
                        logging.info(
                            f"[run_stream] Final answer review rejected attempt "
                            f"{_answer_review_attempts}: {_answer_review_feedback[:500]}"
                        )
                    except Exception as e:
                        logging.warning(
                            f"[run_stream] Final answer review failed; allowing answer. Error: {e}"
                        )
                        break

            # Check if there was a NEW execution in the unprocessed portion, or incomplete answer
            # Always use processed_length to avoid re-detecting already-executed commands
            unprocessed_response = self.response[processed_length:]
            has_new_execution = "</execute>" in unprocessed_response
            # If the previous iteration broke for execution, treat this as
            # a post-execution iteration so we use the execution-aware prompt
            # that shows the model its error output instead of just "provide answer"
            if _previous_iteration_executed and not has_new_execution:
                has_new_execution = True
                _previous_iteration_executed = False
            # Use has_complete_answer for proper detection instead of simple string check
            has_real_answer_open = bool(find_real_answer_tags(self.response, "open"))
            has_incomplete_answer = has_real_answer_open and not has_complete_answer(
                self.response
            )
            # Also check if there's NO real answer at all - we need to prompt for one
            has_no_answer = not has_real_answer_open
            has_rejected_answer = _answer_review_rejected

            # A fresh continuation can start with <answer> and stream user-facing
            # answer content correctly, but if the prior buffer did not end with
            # a newline, concatenation can make that tag look inline in the
            # accumulated response. Preserve the stricter parser rule by
            # re-wrapping the already streamed answer on its own line.
            if has_no_answer and not has_new_execution and answer_content.strip():
                _recovered_answer_text = answer_content.strip()
                _last_recovered_answer_text = _recovered_answer_text
                _recovered_answer_hash = str(hash(_recovered_answer_text))
                if _recovered_answer_hash == _last_recovered_answer_hash:
                    _continuation_recovery_count += 1
                    _continuation_recovery_feedback = (
                        "The same streamed answer content was already recovered, "
                        "but the loop still did not recognize a complete final "
                        "answer. Stop repeating the same response. Produce a fresh, "
                        "complete top-level <answer> on its own line, or take the "
                        "next distinct action required to complete the user's request."
                    )
                    answer_content = ""
                    logging.warning(
                        "[run_stream] Skipping duplicate streamed-answer recovery; "
                        "injecting self-healing feedback instead."
                    )
                else:
                    self.response = _append_recovered_answer_block(
                        self.response, _recovered_answer_text
                    )
                    processed_length = len(self.response)
                    _last_recovered_answer_hash = _recovered_answer_hash
                    has_no_answer = False
                    has_incomplete_answer = False
                    logging.info(
                        "[run_stream] Recovered streamed answer content into a real "
                        "<answer> block after no-answer boundary detection."
                    )
                continue

            # If answer text has already streamed to the user but the raw transcript
            # still has an incomplete answer boundary, recover the transcript instead
            # of asking the model to keep "completing" visible text forever.
            if (
                has_incomplete_answer
                and not has_new_execution
                and answer_content.strip()
            ):
                _recovered_answer_text = answer_content.strip()
                _last_recovered_answer_text = _recovered_answer_text
                _before_recovery = self.response

                if is_inside_top_level_answer(self.response):
                    self.response = _close_recovered_answer_block(self.response)
                    logging.info(
                        "[run_stream] Closed streamed top-level <answer> after "
                        "incomplete-answer boundary detection."
                    )
                else:
                    self.response = _append_recovered_answer_block(
                        self.response, _recovered_answer_text
                    )
                    logging.info(
                        "[run_stream] Recovered streamed answer content into a "
                        "top-level <answer> block after incomplete-answer boundary "
                        "detection."
                    )

                processed_length = len(self.response)
                if has_complete_answer(self.response):
                    has_incomplete_answer = False
                    continue

                # Recovery did not produce a parser-visible final answer. Avoid
                # appending the same recovered text repeatedly; let the next
                # continuation receive explicit repair instructions instead.
                if self.response == _before_recovery:
                    logging.warning(
                        "[run_stream] Incomplete-answer recovery made no transcript "
                        "change; injecting self-healing feedback."
                    )
                else:
                    logging.warning(
                        "[run_stream] Incomplete-answer recovery did not produce a "
                        "complete parser-visible answer; injecting self-healing "
                        "feedback."
                    )
                _continuation_recovery_count += 1
                _continuation_recovery_feedback = (
                    "A user-visible answer was already streamed, but the raw "
                    "transcript still has invalid or nested answer boundaries. "
                    "Do not repeat the same answer text. Produce one fresh, "
                    "complete top-level <answer></answer> block on its own line, "
                    "outside any thinking, reflection, markdown code fence, or "
                    "other protocol block."
                )
                answer_content = ""

            # After many consecutive non-exec no-answer iterations, self-heal
            # by steering the next continuation instead of forcing a final
            # answer boundary into the transcript.
            if _no_answer_non_exec_streak >= 25 and has_no_answer:
                _continuation_recovery_count += 1
                _continuation_recovery_feedback = (
                    f"The agent has produced {_no_answer_non_exec_streak} "
                    "non-execution continuations without a real top-level "
                    "<answer> block. Reassess the current state. If the user's "
                    "request is fully satisfied by the existing work, provide a "
                    "complete final <answer> now. If it is not complete, take the "
                    "next distinct action needed to complete it. Do not repeat "
                    "prior thinking, prior command outputs, or already executed "
                    "commands."
                )
                _no_answer_non_exec_streak = 0
                _continuation_iter_lengths = []
                logging.info(
                    "[run_stream] No-answer continuation recovery triggered; "
                    "injecting self-healing feedback instead of forcing finalization."
                )

            # Continue if: new execution, incomplete answer, OR no answer at all (need to prompt for one)
            should_continue = (
                has_new_execution
                or has_incomplete_answer
                or has_no_answer
                or has_rejected_answer
            )

            if not should_continue:
                # Has a complete answer or nothing more to do
                break

            continuation_count += 1
            if has_new_execution:
                _non_exec_continuation_count = 0
            else:
                _non_exec_continuation_count += 1
            # Send keepalive at the start of each iteration to prevent
            # proxy timeouts (e.g. Cloudflare 100s idle timeout) during
            # prompt building and LLM inference startup.
            yield {
                "type": "keepalive",
                "content": "",
                "complete": False,
            }
            _cont_iter_t0 = _time.monotonic()
            logging.info(
                f"[run_stream] Continuation loop iteration {continuation_count}. "
                f"has_new_execution: {has_new_execution}, has_incomplete_answer: {has_incomplete_answer}, "
                f"has_no_answer: {has_no_answer}. response_len: {len(self.response)}"
            )

            if not has_new_execution and (
                _non_exec_continuation_count >= _max_non_exec_continuations
                or _continuation_recovery_count >= _max_recovery_attempts
            ):
                logging.warning(
                    "[run_stream] Non-execution continuation recovery cap reached "
                    f"at iteration {continuation_count}: "
                    f"non_exec={_non_exec_continuation_count}/"
                    f"{_max_non_exec_continuations}, recovery="
                    f"{_continuation_recovery_count}/{_max_recovery_attempts}. "
                    "Finalizing from recovered answer text or recovery pass."
                )
                if _last_recovered_answer_text:
                    self.response = _append_recovered_answer_block(
                        self.response, _last_recovered_answer_text
                    )
                    if has_complete_answer(self.response):
                        break
                c.log_interaction(
                    role=self.agent_name,
                    message="[SUBACTIVITY][CONTINUATION] Recovery cap reached; finalizing response...",
                )
                break

            # --- Stuck-loop detection ---
            # Only check for stuck loops using NON-EXECUTION iterations.
            # Execution iterations (where the model writes code) are productive
            # even if they fail — the model may need several tries to fix errors.
            # We only track iterations where the model just generated thinking/answer
            # without executing code.
            #
            # For incomplete answers, use a shorter window (8 iterations) since
            # the model is clearly stuck re-opening <answer> without closing it.
            # For no-answer cases, use the original 15-iteration window since
            # the model may be doing useful thinking/analysis.
            _stuck_window = 8 if has_incomplete_answer else 15
            if (
                len(_continuation_iter_lengths) >= _stuck_window
                and not has_new_execution
            ):
                _last4 = _continuation_iter_lengths[-_stuck_window:]
                _avg = sum(_last4) / len(_last4)
                _all_small = all(l < 5000 for l in _last4)
                _all_similar = all(abs(l - _avg) < 500 for l in _last4)
                if _all_small and _all_similar:
                    _continuation_recovery_count += 1
                    _continuation_iter_lengths = []
                    if has_incomplete_answer:
                        logging.warning(
                            f"[run_stream] Stuck loop detected (incomplete answer) at iteration {continuation_count}. "
                            f"Last {_stuck_window} non-exec lengths: {_last4}. "
                            "Injecting self-healing feedback."
                        )
                        _continuation_recovery_feedback = (
                            "The answer block is open but the continuation loop is "
                            "repeating similarly sized non-execution text without "
                            "closing it. Continue from the existing answer content, "
                            "include any missing concrete results, and close with "
                            "</answer> only after the answer is complete. If more "
                            "work is still required, leave the answer phase and take "
                            "the next distinct action instead of repeating text."
                        )
                        _incomplete_answer_non_exec_count = 0
                        c.log_interaction(
                            role=self.agent_name,
                            message="[SUBACTIVITY][CONTINUATION] Recovering incomplete answer loop...",
                        )
                    elif has_no_answer:
                        logging.warning(
                            f"[run_stream] Stuck loop detected (no answer) at iteration {continuation_count}. "
                            f"Last 15 non-exec lengths: {_last4}. Injecting self-healing feedback."
                        )
                        _continuation_recovery_feedback = (
                            "The agent appears to be repeating analysis without "
                            "opening a real top-level <answer> block. Do not stop "
                            "early and do not repeat the same analysis. Decide what "
                            "is missing: either perform the next distinct action "
                            "needed for the user request, or provide the complete "
                            "final answer if the task is already satisfied."
                        )
                        c.log_interaction(
                            role=self.agent_name,
                            message="[SUBACTIVITY][CONTINUATION] Recovering no-answer loop...",
                        )
                    elif has_rejected_answer:
                        logging.warning(
                            f"[run_stream] Stuck loop detected after final answer review rejection "
                            f"at iteration {continuation_count}. Last {_stuck_window} non-exec "
                            f"lengths: {_last4}. Injecting stronger review feedback."
                        )
                        _continuation_recovery_feedback = (
                            "The prior final answer was rejected and subsequent "
                            "continuations are repeating. Use the review feedback "
                            "as mandatory completion criteria, take a different "
                            "action if needed, and only produce a new <answer> when "
                            "those criteria are satisfied."
                        )
                        c.log_interaction(
                            role=self.agent_name,
                            message="[SUBACTIVITY][CONTINUATION] Recovering rejected-answer loop...",
                        )

            # --- Self-healing for repeated incomplete answer iterations ---
            # If the model stays in has_incomplete_answer mode for many
            # consecutive non-exec iterations, redirect the next continuation
            # instead of force-closing the answer and ending early.
            if has_incomplete_answer and not has_new_execution:
                _incomplete_answer_non_exec_count += 1
                if _incomplete_answer_non_exec_count >= 10:
                    _continuation_recovery_count += 1
                    logging.warning(
                        f"[run_stream] Incomplete answer recovery triggered at iteration "
                        f"{continuation_count}. {_incomplete_answer_non_exec_count} consecutive "
                        "non-exec iterations with incomplete answer."
                    )
                    _continuation_recovery_feedback = (
                        "The answer has remained incomplete across many continuations. "
                        "Continue from the current answer content without restarting or "
                        "repeating it. Add the missing substance, then close </answer> "
                        "only when the response fully completes the user's request. "
                        "If the task is not actually complete, perform the next distinct "
                        "action needed before closing the answer."
                    )
                    _incomplete_answer_non_exec_count = 0
                    _continuation_iter_lengths = []
                    c.log_interaction(
                        role=self.agent_name,
                        message="[SUBACTIVITY][CONTINUATION] Recovering incomplete answer...",
                    )
            else:
                if has_new_execution:
                    _incomplete_answer_non_exec_count = 0

            # Log a visible subactivity so the user sees continuation progress
            # Build a descriptive status message so the user knows what's happening
            if has_new_execution:
                _cont_status = "Analyzing command execution results"
            elif has_rejected_answer:
                _cont_status = "Improving final answer"
            elif has_incomplete_answer:
                _cont_status = "Completing response"
            elif has_no_answer:
                if continuation_count <= 1:
                    _cont_status = "Formulating response from analysis"
                else:
                    _cont_status = "Refining analysis before responding"
            else:
                _cont_status = "Continuing analysis"
            # Add token context so user sees scale of work
            _resp_tokens = get_tokens(self.response)
            _cont_detail = f"{_cont_status} — {_resp_tokens:,} tokens processed so far"
            c.log_interaction(
                role=self.agent_name,
                message=f"[SUBACTIVITY][CONTINUATION] {_cont_detail}",
            )
            # Yield a streaming event so the front end knows we're still alive
            yield {
                "type": "thinking_stream",
                "content": f"*{_cont_detail}*\n",
                "complete": False,
            }
            # Compress the response to prevent context explosion
            # This summarizes long outputs and truncates verbose thinking
            # CRITICAL: Preserve execute+output pairs intact so the model
            # knows what was already done and doesn't regenerate the same commands
            compressed_response = self.compress_response_for_continuation(
                self.response,
                max_output_lines=80,  # Keep 80 lines max per standalone output
                max_thinking_chars=2000,  # Keep 2000 chars max per thinking block
            )

            # Log compression stats
            original_tokens = get_tokens(self.response)
            compressed_tokens = get_tokens(compressed_response)
            logging.info(
                f"[run_stream] Continuation {continuation_count}: response {original_tokens} tokens "
                f"-> compressed {compressed_tokens} tokens ({compressed_tokens*100//max(original_tokens,1)}%)"
            )

            if has_new_execution:

                # Re-run format_prompt to get FRESH context that includes:
                # - Command execution results logged to conversation
                # - Updated conversation history
                # - Any new memories or context
                # This mirrors how run() would work if we started a new inference
                # IMPORTANT: Pass selected_commands to maintain the filtered command set
                fresh_formatted_prompt, _, _ = await self.format_prompt(
                    user_input=user_input,
                    top_results=int(context_results),
                    conversation_results=conversation_results,
                    prompt=prompt,
                    prompt_category=prompt_category,
                    conversation_name=conversation_name,
                    websearch=websearch,
                    vision_response=vision_response,
                    selected_commands=self._selected_commands,
                    **kwargs,
                )

                # Anonymize AGiXT server URL
                if self.outputs in fresh_formatted_prompt:
                    fresh_formatted_prompt = fresh_formatted_prompt.replace(
                        self.outputs,
                        f"http://localhost:7437/outputs/{self.agent.agent_id}",
                    )

                # The response now has real command output in <output> tags
                # CRITICAL: Tell the model this is a NEW inference turn where it should
                # analyze the actual command output, NOT continue generating from before
                # Use COMPRESSED response to prevent context explosion
                continuation_prompt = f"""{fresh_formatted_prompt}

## Command Execution Complete

The assistant previously executed commands. The ACTUAL outputs are shown below in <output> tags.
DO NOT hallucinate or make up what the command output should be - the real output is already provided.
Analyze the actual output shown and continue with your response.

### Previous Assistant Response (with real command outputs):
{compressed_response}

### Instructions:
1. READ the actual <output> content above - this is the REAL command result
2. DO NOT repeat or fabricate command outputs - they are already shown
3. Based on the ACTUAL output, continue thinking and provide your <answer>
4. If you need to execute more commands, you may do so
5. When you are ready to answer, the opening <answer> tag MUST be the first thing on a new line, not inline in a sentence and not inside a code block
"""
                if has_rejected_answer:
                    continuation_prompt += f"""

## Final Answer Review Still Applies

The previous <answer> block was not accepted as the final response.
Review feedback:
{_answer_review_feedback}

Use the command output above plus this feedback to continue the task until it is complete.
"""
            else:
                # Incomplete answer or no answer - prompt to continue/provide answer
                # IMPORTANT: Pass selected_commands to maintain the filtered command set
                fresh_formatted_prompt, _, _ = await self.format_prompt(
                    user_input=user_input,
                    top_results=int(context_results),
                    conversation_results=conversation_results,
                    prompt=prompt,
                    prompt_category=prompt_category,
                    conversation_name=conversation_name,
                    websearch=websearch,
                    vision_response=vision_response,
                    selected_commands=self._selected_commands,
                    **kwargs,
                )
                if self.outputs in fresh_formatted_prompt:
                    fresh_formatted_prompt = fresh_formatted_prompt.replace(
                        self.outputs,
                        f"http://localhost:7437/outputs/{self.agent.agent_id}",
                    )

                if has_no_answer:
                    # No answer block at all - prompt to provide one
                    # Use compressed response to prevent context explosion
                    if _consecutive_dedup_iterations >= 3:
                        # DEDUP RECOVERY: Model keeps regenerating same commands.
                        # Use a strong prompt telling it to STOP executing and answer.
                        continuation_prompt = (
                            f"{fresh_formatted_prompt}\n\n{self.agent_name}: {compressed_response}\n\n"
                            f"IMPORTANT: All commands you have generated were ALREADY executed in previous "
                            f"iterations and their outputs are shown above in <output> tags. Do NOT generate "
                            f"any more <execute> blocks — regenerating the same commands will be ignored. "
                            f"Analyze the command outputs already provided and give your complete response "
                            f"to the user inside <answer></answer> tags now. The opening <answer> tag "
                            f"MUST be the first thing on a new line, not inline in a sentence and not inside a code block."
                        )
                    else:
                        continuation_prompt = f"{fresh_formatted_prompt}\n\n{self.agent_name}: {compressed_response}\n\nContinue working on the user's request. You may continue thinking, analyzing, and executing commands as needed — take as much time as you need to do thorough work. When you have fully completed all necessary work and are ready to respond, provide your response inside <answer></answer> tags. The opening <answer> tag MUST be the first thing on a new line, not inline in a sentence and not inside a code block. Do not repeat previous thinking or command outputs."
                elif has_rejected_answer:
                    continuation_prompt = f"""{fresh_formatted_prompt}

{self.agent_name}: {compressed_response}

## Final Answer Review

The previous <answer> block was not accepted as the final response.
It should NOT end the task yet.

Review feedback:
{_answer_review_feedback}

Continue thinking and acting autonomously until the user's request is actually complete.
Do not repeat the rejected answer. If more tool use, verification, reasoning, or concrete details are needed, do that now.
Only provide a new top-level <answer></answer> block when all review criteria are satisfied. The opening <answer> tag MUST be the first thing on a new line, not inline in a sentence and not inside a code block.
"""
                else:
                    # Incomplete answer - prompt to continue
                    # Use compressed response to prevent context explosion
                    continuation_prompt = f"{fresh_formatted_prompt}\n\n{self.agent_name}: {compressed_response}\n\nThe assistant started providing an answer but didn't complete it. Continue from where you left off without repeating anything. If the response is complete, simply close the answer block with </answer>."

            if _continuation_recovery_feedback:
                continuation_prompt += f"""

## Continuation Self-Healing Feedback

The continuation loop detected a recoverable failure mode. Do not treat this as a reason to stop early.
Recovery attempt: {_continuation_recovery_count}

{_continuation_recovery_feedback}

Use the available context and outputs above to repair the trajectory. If the user's request is already fully satisfied, provide one complete top-level <answer></answer> block. If it is not complete, continue with the next distinct action required to complete it. Do not repeat already executed commands or previously generated analysis unless you are correcting it.
"""
                _continuation_recovery_feedback = ""

            # Check for new user messages sent during processing (mid-task steering)
            try:
                _iter_conv = Conversations(
                    conversation_name=conversation_name,
                    user=self.user,
                    conversation_id=conversation_id,
                )
                _iter_data = _iter_conv.get_conversation(limit=1000, page=1)
                _new_user_msgs = []
                for _msg in _iter_data.get("interactions", []):
                    if (
                        str(_msg.get("role", "")).lower() == "user"
                        and str(_msg.get("id", "")) not in _known_user_msg_ids
                    ):
                        _new_user_msgs.append(str(_msg.get("message", "")))
                        _known_user_msg_ids.add(str(_msg.get("id", "")))
                if _new_user_msgs:
                    _steering = "\n".join(_new_user_msgs)
                    continuation_prompt += (
                        f"\n\n## New User Message\n"
                        f"The user sent a follow-up message while you were working. "
                        f"Incorporate this feedback into your current task:\n\n"
                        f"{_steering}\n"
                    )
                    logging.info(
                        f"[run_stream] Detected {len(_new_user_msgs)} new user message(s) "
                        f"during continuation iteration {continuation_count}"
                    )
            except Exception as e:
                logging.debug(
                    f"[run_stream] Failed to check for new user messages: {e}"
                )

            try:
                # Send keepalive before starting new inference to prevent
                # client-side read timeouts during prompt building + slot wait
                yield {
                    "type": "keepalive",
                    "content": "",
                    "complete": False,
                }
                # Run FRESH inference with stream=True
                logging.info(
                    f"[run_stream] Starting continuation inference. Prompt length: {len(continuation_prompt)} chars"
                )
                continuation_stream = await self.agent.inference(
                    prompt=continuation_prompt, use_smartest=use_smartest, stream=True
                )

                # Process the continuation stream with FULL tag processing
                # This is similar to the main stream processing but for continuation
                continuation_response = ""
                continuation_in_answer = False
                continuation_answer_content = (
                    "" if _answer_review_rejected else answer_content
                )  # Rejected answers must be replaced, not continued.
                continuation_current_tag = None
                continuation_current_tag_content = ""
                continuation_current_tag_message_id = None  # Track message ID
                continuation_processed_thinking_ids = set()
                broke_for_execution = False  # Track if we broke early for execution
                continuation_detected_tags = (
                    set()
                )  # Track which opening tags we've already detected
                _cont_in_reasoning = False  # Track reasoning_content field state
                _cont_commands_executed = (
                    0  # Track if execution_agent actually ran any commands
                )

                async for chunk_data in iterate_stream(continuation_stream):
                    # Extract token from chunk — mirrors main stream logic
                    # including reasoning_content support for DeepSeek/OpenAI/Qwen
                    token = None
                    if isinstance(chunk_data, str):
                        token = chunk_data
                    elif hasattr(chunk_data, "choices") and len(chunk_data.choices) > 0:
                        delta = chunk_data.choices[0].delta
                        if hasattr(delta, "content") and delta.content:
                            token = delta.content
                            # If we were receiving reasoning_content, close the thinking block
                            if _cont_in_reasoning:
                                _cont_in_reasoning = False
                                token = "</think>" + token
                        elif (
                            hasattr(delta, "reasoning_content")
                            and delta.reasoning_content
                        ):
                            # Providers like DeepSeek/OpenAI/Qwen send thinking in a separate field
                            token = delta.reasoning_content
                            if not _cont_in_reasoning:
                                _cont_in_reasoning = True
                                token = "<think>" + token

                    if not token:
                        continue

                    prev_len = len(continuation_response)
                    continuation_response += token
                    if "<answer>" in token.lower():
                        normalized_continuation_response = (
                            _RE_PROTOCOL_ANSWER_BOUNDARY.sub(
                                r"\1\n\2", continuation_response
                            )
                        )
                        if normalized_continuation_response != continuation_response:
                            continuation_response = normalized_continuation_response

                    # Detect leaked <interaction> XML in continuation and convert
                    if "</interaction>" in continuation_response.lower():
                        converted = _convert_interaction_to_execute(
                            continuation_response
                        )
                        if converted != continuation_response:
                            continuation_response = converted

                    # Process tags in continuation (thinking, reflection, execute, answer)
                    # Check a sliding window of the last 20 chars for tags (enough for </reflection>)
                    # This handles tags that are split across multiple tokens
                    tag_check_window = (
                        continuation_response[-20:]
                        if len(continuation_response) > 20
                        else continuation_response
                    )

                    # Check for CLOSING tags FIRST, then opening tags.
                    # This ensures that when </thinking><answer> appear in the
                    # same window, </thinking> is processed before <answer> opens,
                    # so the thinking content is properly finalized in the DB.
                    # Support both </think> and </thinking> forms
                    _CONTINUATION_CLOSE_TAGS = [
                        ("thinking", "</thinking>"),
                        ("thinking", "</think>"),
                        ("reflection", "</reflection>"),
                        ("execute", "</execute>"),
                        ("answer", "</answer>"),
                    ]
                    for canonical_name, close_tag in _CONTINUATION_CLOSE_TAGS:
                        if (
                            close_tag in tag_check_window.lower()
                            and continuation_current_tag == canonical_name
                        ):
                            # Tag closed - process it
                            if canonical_name == "execute":
                                # Execute commands and stop the continuation stream
                                # Truncate at </execute> to discard hallucinated output
                                execute_end_match = re.search(
                                    r"</execute>", continuation_response, re.IGNORECASE
                                )
                                if execute_end_match:
                                    truncated_continuation = continuation_response[
                                        : execute_end_match.end()
                                    ]
                                    continuation_response = truncated_continuation

                                self.response += continuation_response
                                broke_for_execution = True

                                # Update processed_length before execution so we can detect new executions
                                processed_length = len(self.response)

                                # Close/cancel the continuation stream NOW to free the
                                # inference slot BEFORE running execution_agent.  Without
                                # this, the iterate_stream daemon thread keeps consuming
                                # from the old generation, occupying the single ezlocalai
                                # slot while code executes (which doesn't need the LLM).
                                try:
                                    if hasattr(continuation_stream, "close"):
                                        continuation_stream.close()
                                    elif hasattr(
                                        continuation_stream, "response"
                                    ) and hasattr(
                                        continuation_stream.response, "close"
                                    ):
                                        continuation_stream.response.close()
                                except Exception:
                                    pass  # Best-effort cleanup

                                # Create queue for remote command requests in continuation
                                cont_remote_queue = asyncio.Queue()

                                async def cont_remote_callback(remote_cmd):
                                    await cont_remote_queue.put(remote_cmd)
                                    return f"[REMOTE COMMAND QUEUED] Waiting for client-side execution.\nRequest ID: {remote_cmd.get('request_id', 'unknown')}"

                                # Run execution_agent with keepalive during long-running commands
                                cont_execution_task = asyncio.create_task(
                                    self.execution_agent(
                                        conversation_name=conversation_name,
                                        conversation_id=conversation_id,
                                        remote_command_callback=cont_remote_callback,
                                    )
                                )
                                while not cont_execution_task.done():
                                    try:
                                        await asyncio.wait_for(
                                            asyncio.shield(cont_execution_task),
                                            timeout=10.0,
                                        )
                                    except asyncio.TimeoutError:
                                        yield {
                                            "type": "keepalive",
                                            "content": "",
                                            "complete": False,
                                        }
                                    except Exception:
                                        break
                                if cont_execution_task.done():
                                    _cont_commands_executed = (
                                        cont_execution_task.result() or 0
                                    )
                                else:
                                    _cont_commands_executed = 0

                                # If ALL commands were deduplicated (nothing new ran),
                                # tell the model explicitly so it doesn't regenerate them
                                if _cont_commands_executed == 0:
                                    logging.warning(
                                        f"[run_stream] All commands in iteration {continuation_count} "
                                        f"were already processed. Injecting dedup notice."
                                    )
                                    self.response += (
                                        "\n<output>ALL COMMANDS ABOVE WERE ALREADY EXECUTED IN PREVIOUS ITERATIONS. "
                                        "Their outputs are already shown above. Do NOT regenerate these same commands. "
                                        "If you need to run DIFFERENT commands, do so. "
                                        "If the task is complete, provide your answer in <answer></answer> tags. "
                                        "The opening <answer> tag MUST be the first thing on a new line, "
                                        "not inline in a sentence and not inside a code block.</output>"
                                    )

                                # Yield any remote command requests
                                while not cont_remote_queue.empty():
                                    remote_cmd = await cont_remote_queue.get()
                                    yield {
                                        "type": "remote_command_request",
                                        "content": remote_cmd,
                                        "complete": True,
                                    }

                                # Update processed_length again after execution adds output
                                processed_length = len(self.response)

                                # Break to start new continuation with execution output
                                break
                            elif canonical_name in ("thinking", "reflection"):
                                # Extract and log the complete tag content
                                # Use regex that matches both <think> and <thinking> forms
                                tag_pattern = (
                                    r"<think(?:ing)?>(.*?)</think(?:ing)?>"
                                    if canonical_name == "thinking"
                                    else r"<reflection>(.*?)</reflection>"
                                )
                                matches = list(
                                    re.finditer(
                                        tag_pattern,
                                        continuation_response,
                                        re.DOTALL | re.IGNORECASE,
                                    )
                                )
                                for m in matches:
                                    content = m.group(1).strip()
                                    tag_id = f"{canonical_name}:{hash(content)}"
                                    if (
                                        tag_id
                                        not in continuation_processed_thinking_ids
                                        and content
                                    ):
                                        continuation_processed_thinking_ids.add(tag_id)

                                        if canonical_name == "thinking":
                                            log_msg = (
                                                f"[SUBACTIVITY][THOUGHT] {content}"
                                            )
                                        else:
                                            log_msg = (
                                                f"[SUBACTIVITY][REFLECTION] {content}"
                                            )

                                        # Update existing or create new message
                                        if continuation_current_tag_message_id:
                                            c.update_message_by_id(
                                                continuation_current_tag_message_id,
                                                log_msg,
                                            )
                                        else:
                                            c.log_interaction(
                                                role=self.agent_name, message=log_msg
                                            )

                                        yield {
                                            "type": canonical_name,
                                            "content": content,
                                            "complete": True,
                                        }

                            # Reset answer flag when </answer> is detected
                            if canonical_name == "answer":
                                continuation_in_answer = False
                            continuation_current_tag = None
                            continuation_current_tag_content = ""
                            continuation_current_tag_message_id = None  # Reset
                            # Allow the same tag type to be detected again
                            # (model may produce multiple <thinking> blocks per continuation)
                            # Clear both <think> and <thinking> forms when either closes
                            if canonical_name == "thinking":
                                continuation_detected_tags.discard("<thinking>")
                                continuation_detected_tags.discard("<think>")
                            else:
                                open_form = close_tag.replace("/", "")
                                continuation_detected_tags.discard(open_form)

                    # Check for opening tags AFTER closing tags
                    # This ensures closing tags are processed first when both appear
                    # in the same tag_check_window (e.g., </thinking><answer>)
                    _CONTINUATION_OPEN_TAGS = [
                        ("thinking", "<thinking>"),
                        ("thinking", "<think>"),  # DeepSeek-R1, Qwen QwQ short form
                        ("reflection", "<reflection>"),
                        ("execute", "<execute>"),
                        ("answer", "<answer>"),
                    ]
                    for canonical_name, open_tag in _CONTINUATION_OPEN_TAGS:
                        if (
                            open_tag in tag_check_window.lower()
                            and open_tag not in continuation_detected_tags
                        ):
                            if canonical_name == "answer":
                                answer_tag_pos = continuation_response.lower().rfind(
                                    open_tag
                                )
                                if answer_tag_pos < 0 or not is_real_answer_tag(
                                    continuation_response, answer_tag_pos
                                ):
                                    continue

                            # Finalize the current tag before switching to a new one.
                            # This prevents thought truncation when a new tag opens
                            # before the current tag's close is detected (e.g., model
                            # generates <thinking>content<answer> without </thinking>).
                            if (
                                continuation_current_tag
                                and continuation_current_tag != canonical_name
                                and continuation_current_tag
                                in ("thinking", "reflection")
                            ):
                                # Save the current tag content to DB before switching
                                if continuation_current_tag_content:
                                    tag_type = (
                                        "THOUGHT"
                                        if continuation_current_tag == "thinking"
                                        else "REFLECTION"
                                    )
                                    final_msg = f"[SUBACTIVITY][{tag_type}] {continuation_current_tag_content}"
                                    if continuation_current_tag_message_id:
                                        c.update_message_by_id(
                                            continuation_current_tag_message_id,
                                            final_msg,
                                        )
                                    else:
                                        c.log_interaction(
                                            role=self.agent_name,
                                            message=final_msg,
                                        )
                                    # Yield the complete thinking/reflection block
                                    yield {
                                        "type": continuation_current_tag,
                                        "content": continuation_current_tag_content,
                                        "complete": True,
                                    }
                                    logging.info(
                                        f"[run_stream] Finalized unclosed {continuation_current_tag} "
                                        f"tag ({len(continuation_current_tag_content)} chars) "
                                        f"before opening {canonical_name}"
                                    )

                            continuation_detected_tags.add(open_tag)
                            continuation_current_tag = canonical_name
                            continuation_current_tag_content = ""
                            continuation_current_tag_message_id = (
                                None  # Reset for new tag
                            )
                            if canonical_name == "answer":
                                continuation_in_answer = True
                            break

                    # Break out of stream loop if we're breaking for execution
                    if broke_for_execution:
                        break

                    # Stream progressive content for current tag
                    if continuation_current_tag and continuation_current_tag in (
                        "thinking",
                        "reflection",
                    ):
                        # Find the opening tag - handle both <think> and <thinking> forms
                        resp_lower = continuation_response.lower()
                        if continuation_current_tag == "thinking":
                            # Find last occurrence of either <think> or <thinking>
                            pos_thinking = resp_lower.rfind("<thinking>")
                            pos_think = resp_lower.rfind("<think>")
                            # Avoid matching <thinking> as <think> — only use <think> if it's not part of <thinking>
                            if (
                                pos_think >= 0
                                and pos_thinking >= 0
                                and pos_thinking == pos_think
                            ):
                                # They overlap, use <thinking>
                                last_start = pos_thinking
                                tag_len = len("<thinking>")
                            elif pos_thinking >= 0 and (
                                pos_think < 0 or pos_thinking > pos_think
                            ):
                                last_start = pos_thinking
                                tag_len = len("<thinking>")
                            elif pos_think >= 0:
                                last_start = pos_think
                                tag_len = len("<think>")
                            else:
                                last_start = -1
                                tag_len = 0
                        else:
                            tag_start_pattern = f"<{continuation_current_tag}>"
                            last_start = resp_lower.rfind(tag_start_pattern)
                            tag_len = len(tag_start_pattern)

                        if last_start >= 0:
                            partial = continuation_response[last_start + tag_len :]
                            # Handle partial closing tags at the end of the buffer
                            # (matches the main stream's approach — only strip partial
                            # tags at the END, not every '<' in the content)
                            if continuation_current_tag == "thinking":
                                partial_close = re.search(
                                    r"</?t?h?i?n?k?i?n?g?>?$",
                                    partial,
                                    re.IGNORECASE,
                                )
                            else:
                                partial_close = re.search(
                                    r"</?r?e?f?l?e?c?t?i?o?n?>?$",
                                    partial,
                                    re.IGNORECASE,
                                )
                            if partial_close and partial_close.group().startswith("<"):
                                partial = partial[: partial_close.start()]
                            # Also strip any other tag-like patterns at the very end
                            # (e.g., partial <answer>, <execute>, etc.)
                            partial_other = re.search(
                                r"</?[a-zA-Z][a-zA-Z]*>?$", partial
                            )
                            if partial_other and partial_other.group().startswith("<"):
                                partial = partial[: partial_other.start()]
                            if len(partial) > len(continuation_current_tag_content):
                                delta = partial[len(continuation_current_tag_content) :]
                                continuation_current_tag_content = partial

                                # Create/update message
                                tag_type = (
                                    "THOUGHT"
                                    if continuation_current_tag == "thinking"
                                    else "REFLECTION"
                                )
                                updated_msg = f"[SUBACTIVITY][{tag_type}] {continuation_current_tag_content}"

                                if not continuation_current_tag_message_id:
                                    # Create message once
                                    continuation_current_tag_message_id = (
                                        c.log_interaction(
                                            role=self.agent_name, message=updated_msg
                                        )
                                    )
                                else:
                                    # Update existing message
                                    c.update_message_by_id(
                                        continuation_current_tag_message_id,
                                        updated_msg,
                                    )

                                # Yield for streaming API
                                yield {
                                    "type": f"{continuation_current_tag}_stream",
                                    "content": delta,
                                    "complete": False,
                                }

                    # Yield answer tokens - use extract_top_level_answer to get the LAST answer
                    # (same logic as final_answer extraction to ensure consistency)
                    # IMPORTANT: Suppress answer yields while inside a nested tag
                    # (thinking, reflection, execute). This prevents thinking/execute
                    # content from bleeding through to the answer stream in the UI.
                    if continuation_in_answer and continuation_current_tag not in (
                        "thinking",
                        "reflection",
                        "execute",
                    ):
                        # Use the same extraction as final answer - gets LAST top-level answer
                        new_answer = extract_top_level_answer(continuation_response)
                        if new_answer:
                            # Clean internal tags from answer content
                            cleaned_new_answer = re.sub(
                                r"<think(?:ing)?>.*?</think(?:ing)?>",
                                "",
                                new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            cleaned_new_answer = re.sub(
                                r"<reflection>.*?</reflection>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            # Remove <final> tags - model-generated internal markers
                            cleaned_new_answer = re.sub(
                                r"<final>.*?</final>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            # Remove <step>, <reward>, <count>, <execute>, <output> tags
                            cleaned_new_answer = re.sub(
                                r"<step>.*?</step>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            cleaned_new_answer = re.sub(
                                r"<reward>.*?</reward>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            cleaned_new_answer = re.sub(
                                r"<count>.*?</count>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            cleaned_new_answer = re.sub(
                                r"<execute>.*?</execute>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            cleaned_new_answer = re.sub(
                                r"<output>.*?</output>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            # Remove <speak> tags - TTS filler that shouldn't appear in text answer
                            cleaned_new_answer = re.sub(
                                r"<speak>.*?</speak>",
                                "",
                                cleaned_new_answer,
                                flags=re.DOTALL | re.IGNORECASE,
                            )
                            # Remove orphaned closing tags
                            cleaned_new_answer = re.sub(
                                r"</(?:thinking|reflection|step|reward|count|answer|final|execute|output|speak)>",
                                "",
                                cleaned_new_answer,
                                flags=re.IGNORECASE,
                            )
                            # Remove plain text LLM reasoning artifacts that leak into answers
                            cleaned_new_answer = re.sub(
                                r"^\s*(?:Quality (?:check|score)|Reward)\s*:.*$",
                                "",
                                cleaned_new_answer,
                                flags=re.MULTILINE | re.IGNORECASE,
                            )
                            # Remove partial closing tag at end
                            cleaned_new_answer = re.sub(
                                r"</?a?n?s?w?e?r?$",
                                "",
                                cleaned_new_answer,
                                flags=re.IGNORECASE,
                            )
                            # Remove orphaned angle brackets at boundaries (remnants of stripped tags)
                            cleaned_new_answer = re.sub(
                                r"^[>\s]+", "", cleaned_new_answer
                            )
                            cleaned_new_answer = re.sub(
                                r"[<\s]+$", "", cleaned_new_answer
                            )
                            cleaned_new_answer = cleaned_new_answer.strip()

                            # Yield delta if we have new content
                            if len(cleaned_new_answer) > len(
                                continuation_answer_content
                            ):
                                delta = cleaned_new_answer[
                                    len(continuation_answer_content) :
                                ]
                                # Skip if it looks like an opening tag
                                if delta and not re.match(r"^\s*<[a-zA-Z]", delta):
                                    yield {
                                        "type": "answer",
                                        "content": delta,
                                        "complete": False,
                                    }
                                continuation_answer_content = cleaned_new_answer

                    # Stop streaming answer once </answer> is found
                    if "</answer>" in tag_check_window.lower():
                        continuation_in_answer = False

                    # Break early if we have a complete answer - don't keep consuming
                    # potentially very long post-answer thinking tokens from the model
                    if has_complete_answer(continuation_response):
                        break
                    continuation_response_for_check = continuation_response
                    if (
                        self.response
                        and not self.response.endswith(("\n", "\r"))
                        and continuation_response.lstrip()
                        .lower()
                        .startswith("<answer>")
                    ):
                        continuation_response_for_check = (
                            "\n" + continuation_response.lstrip()
                        )
                    if has_complete_answer(
                        continuation_response_for_check
                        if _answer_review_rejected
                        else self.response + continuation_response_for_check
                    ):
                        break

                # Close/cancel the continuation stream to free the inference slot
                # for the next iteration's LLM call (same rationale as main stream cleanup)
                try:
                    if hasattr(continuation_stream, "close"):
                        continuation_stream.close()
                    elif hasattr(continuation_stream, "response") and hasattr(
                        continuation_stream.response, "close"
                    ):
                        continuation_stream.response.close()
                except Exception:
                    pass  # Best-effort cleanup

                logging.info(
                    f"[run_stream] Continuation iteration {continuation_count} complete. "
                    f"broke_for_execution: {broke_for_execution}, "
                    f"continuation_response_len: {len(continuation_response)}, "
                    f"elapsed: {_time.monotonic() - _cont_iter_t0:.1f}s"
                )

                # Reset retry counter on successful iteration
                self._continuation_retry_count = 0

                # Propagate accumulated answer content back so next iteration
                # doesn't re-yield already-streamed content (answer_content may
                # have been reset to "" when execute tags appeared inside answer blocks)
                if continuation_answer_content:
                    answer_content = continuation_answer_content

                # Track this iteration's response length for stuck detection
                # Only track NON-execution iterations — execution iterations are
                # productive (model is writing/fixing code) and shouldn't count as "stuck"
                # BUT: if broke_for_execution but ALL commands were deduplicated,
                # that's NOT productive — treat it like a non-exec iteration
                # Repeated dedupes mean the model is re-requesting commands that
                # already ran. Treat that as recoverable steering feedback rather
                # than a reason to finalize before the task is complete.
                if self._total_dedup_count >= 15:
                    _continuation_recovery_count += 1
                    logging.warning(
                        f"[run_stream] Excessive dedup loop detected at iteration {continuation_count}. "
                        f"Total cumulative deduplicated commands={self._total_dedup_count}. "
                        "Injecting self-healing feedback."
                    )
                    _continuation_recovery_feedback = (
                        "Many generated commands were skipped because they were "
                        "duplicates of commands that already ran. Do not regenerate "
                        "those commands. Read the existing <output> blocks and decide "
                        "whether the task is complete. If more work is required, call "
                        "a different command with materially different arguments. If "
                        "the task is complete, provide the final answer."
                    )
                    self._total_dedup_count = 0
                    _consecutive_dedup_iterations = 0
                    _continuation_iter_lengths = []
                    c.log_interaction(
                        role=self.agent_name,
                        message="[SUBACTIVITY][CONTINUATION] Recovering duplicate-command loop...",
                    )
                if broke_for_execution and _cont_commands_executed > 0:
                    _previous_iteration_executed = True
                    _no_answer_non_exec_streak = 0  # Reset streak on real execution
                    # Only reset dedup streak if this iteration had ZERO dedupes —
                    # if some new commands ran but old ones were also regenerated,
                    # the model is still partially stuck.
                    if getattr(self, "_last_iteration_dedup_count", 0) == 0:
                        _consecutive_dedup_iterations = 0
                    else:
                        _consecutive_dedup_iterations += 1
                        logging.info(
                            f"[run_stream] Iteration {continuation_count}: mixed iteration — "
                            f"{_cont_commands_executed} new commands ran but "
                            f"{self._last_iteration_dedup_count} were deduplicated. "
                            f"Consecutive partial-dedup iterations: {_consecutive_dedup_iterations}."
                        )
                        if _consecutive_dedup_iterations >= 5:
                            _continuation_recovery_count += 1
                            logging.warning(
                                f"[run_stream] Mixed dedup loop detected at iteration {continuation_count}. "
                                f"{_consecutive_dedup_iterations} consecutive iterations with partial dedupes. "
                                "Injecting self-healing feedback."
                            )
                            _continuation_recovery_feedback = (
                                "Some new commands ran, but the model is also "
                                "regenerating commands that were already processed. "
                                "Use the new outputs and the prior outputs together. "
                                "Do not repeat any skipped command. Continue only with "
                                "new actions that materially advance the user's request, "
                                "or provide the final answer if the work is complete."
                            )
                            _consecutive_dedup_iterations = 0
                            _continuation_iter_lengths = []
                            c.log_interaction(
                                role=self.agent_name,
                                message="[SUBACTIVITY][CONTINUATION] Recovering mixed duplicate-command loop...",
                            )
                else:
                    if broke_for_execution and _cont_commands_executed == 0:
                        # Broke for execution but everything was deduplicated
                        # This IS a stuck iteration — the model keeps regenerating same commands
                        _consecutive_dedup_iterations += 1
                        logging.info(
                            f"[run_stream] Iteration {continuation_count}: broke_for_execution=True "
                            f"but 0 commands actually ran (all deduplicated). "
                            f"Consecutive dedup iterations: {_consecutive_dedup_iterations}. "
                            f"Treating as non-exec iteration."
                        )
                        if _consecutive_dedup_iterations >= 4:
                            _continuation_recovery_count += 1
                            logging.warning(
                                f"[run_stream] Dedup loop detected at iteration {continuation_count}. "
                                f"{_consecutive_dedup_iterations} consecutive iterations with all commands "
                                "deduplicated. Injecting self-healing feedback."
                            )
                            _continuation_recovery_feedback = (
                                "Every command in the last continuation was skipped "
                                "as a duplicate. Stop requesting those commands. Read "
                                "the existing outputs already in context. If they are "
                                "enough, synthesize the final answer. If not, choose a "
                                "new command or different arguments that directly address "
                                "what is still missing."
                            )
                            _consecutive_dedup_iterations = 0
                            _continuation_iter_lengths = []
                            c.log_interaction(
                                role=self.agent_name,
                                message="[SUBACTIVITY][CONTINUATION] Recovering duplicate-command loop...",
                            )
                    else:
                        _consecutive_dedup_iterations = (
                            0  # Reset on normal non-exec iteration
                        )
                    _continuation_iter_lengths.append(len(continuation_response))
                    if has_no_answer:
                        _no_answer_non_exec_streak += 1
                    else:
                        _no_answer_non_exec_streak = 0  # Reset if we have answer tags

                # Append continuation to main response (only if we didn't already do it when breaking for execution)
                if not broke_for_execution:
                    # Check for unclosed <execute> tag from stop sequence in continuation
                    _cont_lower = continuation_response.lower()
                    _cont_last_open = _cont_lower.rfind("<execute>")
                    _cont_stop_sequence_execute_appended = False
                    if _cont_last_open != -1:
                        _cont_last_close = _cont_lower.rfind("</execute>")
                        if _cont_last_close < _cont_last_open:
                            continuation_response += "</execute>"
                            _cont_stop_sequence_execute_appended = True
                            logging.info(
                                "[run_stream] Appended </execute> in continuation (stop sequence detected)"
                            )
                    if (
                        self.response
                        and not self.response.endswith(("\n", "\r"))
                        and continuation_response.lstrip()
                        .lower()
                        .startswith("<answer>")
                    ):
                        continuation_response = "\n" + continuation_response.lstrip()
                    self.response += continuation_response

                    if (
                        _cont_stop_sequence_execute_appended
                        and "{COMMANDS}" in unformatted_prompt
                        and "disable_commands" not in kwargs
                    ):
                        logging.info(
                            "[run_stream] Executing continuation command completed by stop sequence"
                        )
                        cont_stop_remote_queue = asyncio.Queue()

                        async def cont_stop_remote_callback(remote_cmd):
                            await cont_stop_remote_queue.put(remote_cmd)
                            return (
                                "[REMOTE COMMAND QUEUED] Waiting for client-side execution.\n"
                                f"Request ID: {remote_cmd.get('request_id', 'unknown')}"
                            )

                        cont_stop_execution_task = asyncio.create_task(
                            self.execution_agent(
                                conversation_name=conversation_name,
                                conversation_id=conversation_id,
                                remote_command_callback=cont_stop_remote_callback,
                            )
                        )
                        while not cont_stop_execution_task.done():
                            try:
                                await asyncio.wait_for(
                                    asyncio.shield(cont_stop_execution_task),
                                    timeout=10.0,
                                )
                            except asyncio.TimeoutError:
                                yield {
                                    "type": "keepalive",
                                    "content": "",
                                    "complete": False,
                                }
                            except Exception:
                                break
                        if cont_stop_execution_task.done():
                            _cont_commands_executed = (
                                cont_stop_execution_task.result() or 0
                            )
                        else:
                            _cont_commands_executed = 0

                        while not cont_stop_remote_queue.empty():
                            remote_cmd = await cont_stop_remote_queue.get()
                            yield {
                                "type": "remote_command_request",
                                "content": remote_cmd,
                                "complete": True,
                            }

                        broke_for_execution = True
                        if _cont_commands_executed:
                            _previous_iteration_executed = True
                            _no_answer_non_exec_streak = 0
                            if _continuation_iter_lengths:
                                _continuation_iter_lengths.pop()

                # Update processed_length to track what we've handled
                processed_length = len(self.response)

                # If we got a COMPLETE answer, loop back so complete-answer
                # handling can finalize it.
                if has_complete_answer(self.response):
                    continue

                # If we hit an execute tag, continue loop to handle it
                if "</execute>" in continuation_response:
                    continue

                has_real_answer_open = bool(
                    find_real_answer_tags(self.response, "open")
                )

                # If we still don't have a real answer, continue to prompt for one
                if not has_real_answer_open:
                    continue

                # If <answer> is open but not closed and this inference didn't
                # produce a command to execute, let the model continue in the
                # next iteration rather than force-closing. The model may have
                # hit its output token limit while writing a long answer and
                # should be given a chance to finish it properly.
                if (
                    has_real_answer_open
                    and not has_complete_answer(self.response)
                    and "</execute>" not in continuation_response
                ):
                    logging.info(
                        f"[run_stream] Inference opened <answer> without closing it "
                        f"at iteration {continuation_count}. Continuing to let agent finish."
                    )
                    continue

            except asyncio.CancelledError:
                logging.warning(
                    f"[run_stream] Continuation cancelled (CancelledError) at iteration {continuation_count}. "
                    f"Saving partial response ({len(self.response)} chars)."
                )
                # Save whatever response we've accumulated so far
                # so the user sees partial work instead of nothing
                partial_answer = self.response
                if find_real_answer_tags(partial_answer, "open"):
                    partial_answer = extract_top_level_answer(partial_answer)
                if partial_answer:
                    # Strip thinking/execute/output/speak content blocks
                    partial_answer = re.sub(
                        r"<think(?:ing)?>.*?</think(?:ing)?>",
                        "",
                        partial_answer,
                        flags=re.DOTALL | re.IGNORECASE,
                    )
                    partial_answer = re.sub(
                        r"<execute>.*?</execute>",
                        "",
                        partial_answer,
                        flags=re.DOTALL | re.IGNORECASE,
                    )
                    partial_answer = re.sub(
                        r"<output>.*?</output>",
                        "",
                        partial_answer,
                        flags=re.DOTALL | re.IGNORECASE,
                    )
                    # Strip <speak> content blocks — TTS filler that shouldn't appear in text
                    partial_answer = re.sub(
                        r"<speak>.*?</speak>",
                        "",
                        partial_answer,
                        flags=re.DOTALL | re.IGNORECASE,
                    )
                    partial_answer = re.sub(
                        r"</?(?:answer|think(?:ing)?|reflection|step|execute|output|speak|reward|count|final)>",
                        "",
                        partial_answer,
                        flags=re.IGNORECASE,
                    )
                    partial_answer = partial_answer.strip()
                if partial_answer and log_output:
                    try:
                        c.log_interaction(
                            role=self.agent_name,
                            message=partial_answer,
                        )
                        logging.info(
                            f"[run_stream] Saved partial response ({len(partial_answer)} chars) on cancellation"
                        )
                    except Exception:
                        pass
                raise
            except GeneratorExit:
                logging.warning(
                    f"[run_stream] Continuation generator closed (GeneratorExit) at iteration {continuation_count}"
                )
                raise
            except Exception as e:
                # Check if this is a retryable error (timeout, connection error)
                error_str = str(e).lower()
                is_retryable = any(
                    keyword in error_str
                    for keyword in [
                        "timed out",
                        "timeout",
                        "connection",
                        "reset by peer",
                        "broken pipe",
                        "server disconnected",
                        "503",
                        "502",
                        "429",
                    ]
                )

                if is_retryable:
                    # Mark current provider as failed so rotation picks a different one
                    if hasattr(self.agent, "ai_provider_manager"):
                        try:
                            provider_mgr = self.agent.ai_provider_manager
                            # Clear failed providers to allow fresh rotation
                            provider_mgr.failed_providers.clear()
                        except Exception:
                            pass

                    # Track retry attempts for this iteration
                    if not hasattr(self, "_continuation_retry_count"):
                        self._continuation_retry_count = 0
                    self._continuation_retry_count += 1

                    if self._continuation_retry_count <= 3:
                        # Wait with exponential backoff before retrying
                        wait_time = min(30, 5 * self._continuation_retry_count)
                        logging.warning(
                            f"[run_stream] Retryable error in continuation iteration {continuation_count} "
                            f"(retry {self._continuation_retry_count}/3): {e}. "
                            f"Waiting {wait_time}s before retry."
                        )
                        await asyncio.sleep(wait_time)
                        # Don't increment continuation_count for retries
                        continuation_count -= 1
                        continue
                    else:
                        logging.error(
                            f"[run_stream] Max retries (3) reached for continuation. "
                            f"Attempting self-healing continuation. Error: {e}"
                        )
                        self._continuation_retry_count = 0
                        _continuation_recovery_count += 1
                        _continuation_recovery_feedback = (
                            "The previous continuation inference hit repeated "
                            "retryable provider or connection errors. Resume from "
                            "the available context without repeating work. If the "
                            "task can be completed from existing outputs, provide "
                            "the final answer. If more work is required, take the "
                            "next distinct action."
                        )
                        continue
                else:
                    logging.error(f"Error during continuation: {e}")
                    import traceback

                    logging.error(traceback.format_exc())
                    break

        logging.info(
            f"[run_stream] Continuation loop finished. Total iterations: {continuation_count}. "
            f"has_complete_answer: {has_complete_answer(self.response)}. "
            f"Response length: {len(self.response)}"
        )

        # Recovery pass: if continuation ended without a complete answer,
        # run one non-streaming synthesis inference to produce a clean
        # user-facing answer from the already collected outputs.
        if not has_complete_answer(self.response):
            try:
                c.log_interaction(
                    role=self.agent_name,
                    message="[SUBACTIVITY][CONTINUATION] Recovering response from prior execution context...",
                )
                yield {
                    "type": "thinking_stream",
                    "content": "*Recovering final response from prior execution context...*\n",
                    "complete": False,
                }

                recovery_context = self.compress_response_for_continuation(
                    self.response,
                    max_output_lines=120,
                    max_thinking_chars=1000,
                )
                recovery_prompt_base, _, _ = await self.format_prompt(
                    user_input=user_input,
                    top_results=int(context_results),
                    conversation_results=conversation_results,
                    prompt=prompt,
                    prompt_category=prompt_category,
                    conversation_name=conversation_name,
                    websearch=websearch,
                    vision_response=vision_response,
                    selected_commands=self._selected_commands,
                    **kwargs,
                )
                if self.outputs in recovery_prompt_base:
                    recovery_prompt_base = recovery_prompt_base.replace(
                        self.outputs,
                        f"http://localhost:7437/outputs/{self.agent.agent_id}",
                    )

                recovery_prompt = f"""{recovery_prompt_base}

## Recovery Mode

The prior continuation loop did not produce a complete, clean <answer> block.
Your job now is to recover gracefully from the existing work:

1. Analyze the already available execution outputs and reasoning below.
2. Do NOT generate any <execute> blocks in this recovery step.
3. If prior attempts failed, explicitly state what failed and why.
4. Provide the best possible final user-facing response, including concrete findings.
5. If information is still missing, state the exact missing info and next best action.

Return only one top-level <answer>...</answer> block. The opening <answer> tag MUST be the first thing on a new line, not inline in a sentence and not inside a code block.

### Prior Assistant State
{recovery_context}
"""

                recovery_result = await self.agent.inference(
                    prompt=recovery_prompt,
                    use_smartest=use_smartest,
                    stream=False,
                )
                recovery_text = str(recovery_result or "").strip()
                recovery_answer = extract_top_level_answer(recovery_text)
                if not recovery_answer and "<execute>" not in recovery_text.lower():
                    recovery_answer = recovery_text

                if recovery_answer and recovery_answer.strip():
                    self.response += f"\n<answer>{recovery_answer.strip()}</answer>"
                    logging.info(
                        "[run_stream] Recovery inference produced a synthetic final answer"
                    )
            except Exception as recovery_error:
                logging.warning(
                    f"[run_stream] Recovery inference failed, using fallback extraction: {recovery_error}"
                )

        # Extract final answer using proper top-level detection
        # This handles cases where <answer> appears inside <thinking> blocks
        final_answer = ""
        final_answer = extract_top_level_answer(self.response)

        if not final_answer:
            # No top-level answer found, use full response
            final_answer = self.response

        # Clean final answer - ALSO REMOVE <answer> tags that might be left over
        final_answer = re.sub(
            r"<answer>|</answer>",
            "",
            final_answer,
            flags=re.IGNORECASE,
        )
        final_answer = re.sub(
            r"<think(?:ing)?>.*?</think(?:ing)?>",
            "",
            final_answer,
            flags=re.DOTALL | re.IGNORECASE,
        )
        final_answer = re.sub(
            r"<reflection>.*?</reflection>",
            "",
            final_answer,
            flags=re.DOTALL | re.IGNORECASE,
        )
        final_answer = re.sub(
            r"<execute>.*?</execute>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        final_answer = re.sub(
            r"<output>.*?</output>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        # Remove <step> tags from answer - these are thinking artifacts
        final_answer = re.sub(
            r"<step>.*?</step>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        # Remove <reward> and <count> tags
        final_answer = re.sub(
            r"<reward>.*?</reward>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        final_answer = re.sub(
            r"<count>.*?</count>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        # Remove <speak> tags - these are TTS filler speech and shouldn't appear in final response
        final_answer = re.sub(
            r"<speak>.*?</speak>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        # Remove <final> tags - these are model-generated internal markers that shouldn't be visible
        final_answer = re.sub(
            r"<final>.*?</final>", "", final_answer, flags=re.DOTALL | re.IGNORECASE
        )
        # Remove orphaned closing tags (closing tags without matching opening tags)
        # This handles malformed model output with stray </thinking>, </think>, </reflection>, etc.
        final_answer = re.sub(
            r"</think(?:ing)?>|</reflection>|</step>|</execute>|</output>|</speak>|</answer>|</count>|</reward>|</final>",
            "",
            final_answer,
            flags=re.IGNORECASE,
        )
        # Remove orphaned tag fragments WITHOUT the leading "<"
        # This handles cases where chunk boundaries split tags, leaving "reflection>", "/thinking>", etc.
        final_answer = re.sub(
            r"(?<![<a-zA-Z])/?(?:think(?:ing)?|reflection|step|reward|count|execute|output|speak|answer)>",
            "",
            final_answer,
            flags=re.IGNORECASE,
        )
        # Remove orphaned opening tag fragments (opening tags without closing, not inside other content)
        final_answer = re.sub(
            r"<(?:think(?:ing)?|reflection|step|reward|count)>(?![^<]*</(?:think(?:ing)?|reflection|step|reward|count)>)",
            "",
            final_answer,
            flags=re.IGNORECASE,
        )
        # Remove plain text LLM reasoning artifacts that leak into answers
        # e.g. "Quality check: The code correctly..." or "Reward: 1.0" or "The reward score is high..."
        final_answer = re.sub(
            r"^\s*(?:Quality (?:check|score)|Reward)\s*:.*$",
            "",
            final_answer,
            flags=re.MULTILINE | re.IGNORECASE,
        )
        # Remove sentences about reward scores (model meta-commentary)
        final_answer = re.sub(
            r"(?:^|\n)\s*The reward score\b[^\n]*",
            "",
            final_answer,
            flags=re.IGNORECASE,
        )
        # Remove orphaned angle brackets at boundaries (remnants of stripped tags)
        final_answer = re.sub(r"^[>\s]+", "", final_answer)
        final_answer = re.sub(r"[<\s]+$", "", final_answer)
        final_answer = final_answer.strip()

        # Fallback: if final_answer is empty after cleanup but response had output blocks,
        # extract the last non-empty <output> block's content as a usable answer.
        # This handles cases where the model executed code successfully but never
        # wrapped results in <answer> tags (e.g., hit max continuation loops).
        if not final_answer and "<output>" in self.response.lower():
            output_matches = re.findall(
                r"<output>(.*?)</output>",
                self.response,
                re.DOTALL | re.IGNORECASE,
            )
            for match in reversed(output_matches):
                stripped = match.strip()
                stripped_lower = stripped.lower()
                # Skip internal/system fallback outputs that should never be
                # shown to the user as the final answer.
                if any(
                    marker in stripped_lower
                    for marker in [
                        "all commands above were already executed",
                        "[remote command queued]",
                        "[remote command pending]",
                        "do not regenerate these same commands",
                    ]
                ):
                    continue
                if stripped:
                    final_answer = stripped
                    logging.info(
                        "[run_stream] No <answer> block found; using last code output as fallback answer"
                    )
                    break

        # Deduplicate repeated content - sometimes the model repeats the same answer multiple times
        # This detects if the answer contains repeated blocks and keeps only the first occurrence
        final_answer = self.deduplicate_repeated_content(final_answer)

        # Deanonymize AGiXT server URL
        final_answer = final_answer.replace(
            f"http://localhost:7437/outputs/{self.agent.agent_id}", self.outputs
        )
        self.response = self.response.replace(
            f"http://localhost:7437/outputs/{self.agent.agent_id}", self.outputs
        )

        # Handle TTS if enabled
        agent_settings = self.agent.AGENT_CONFIG["settings"]
        tts = False
        if "tts" in kwargs:
            tts = str(kwargs["tts"]).lower() == "true"
        if "tts_provider" in agent_settings and tts == True:
            if (
                agent_settings["tts_provider"] != "None"
                and agent_settings["tts_provider"] != ""
                and agent_settings["tts_provider"] != None
            ):
                try:
                    if thinking_id:
                        c.log_interaction(
                            role=self.agent_name,
                            message=f"[SUBACTIVITY][{thinking_id}][EXECUTION] Generating audio response.",
                        )
                    tts_response = await self.agent.text_to_speech(text=final_answer)
                    if str(tts_response).startswith("http"):
                        tts_response = f'<audio controls><source src="{tts_response}" type="audio/wav"></audio>'
                    elif not str(tts_response).startswith("<audio"):
                        file_type = "wav"
                        file_name = f"{uuid.uuid4().hex}.{file_type}"
                        audio_path = os.path.join(
                            self.agent.working_directory, file_name
                        )
                        audio_data = base64.b64decode(tts_response)
                        with open(audio_path, "wb") as f:
                            f.write(audio_data)
                        tts_response = f'<audio controls><source src="{AGIXT_URI}/outputs/{self.agent.agent_id}/{c.get_conversation_id()}/{file_name}" type="audio/wav"></audio>'
                    final_answer = f"{final_answer}\n\n{tts_response}"
                except Exception as e:
                    logging.warning(f"Failed to get TTS response: {e}")

        # Write to memory if enabled
        if disable_memory != True:
            try:
                await self.agent_memory.write_text_to_memory(
                    user_input=user_input,
                    text=final_answer,
                    external_source="user input",
                )
            except Exception as e:
                log_silenced_exception(e, "chat: writing to memory")

        # Log the final output
        if log_output and final_answer:
            c.log_interaction(role=self.agent_name, message=final_answer)

            # Emit webhook event for agent response
            await webhook_emitter.emit_event(
                event_type="conversation.message.sent",
                data={
                    "conversation_id": c.get_conversation_id(),
                    "conversation_name": conversation_name,
                    "agent_name": self.agent_name,
                    "user": self.user,
                    "message": final_answer,
                    "role": self.agent_name,
                    "timestamp": datetime.now().isoformat(),
                    "prompt_tokens": tokens if "tokens" in locals() else 0,
                },
                user_id=self.user,
                agent_id=self.agent.agent_id,
                agent_name=self.agent_name,
                company_id=self.agent.company_id,
            )

            # Also emit chat completion event
            await webhook_emitter.emit_event(
                event_type="chat.completion.completed",
                data={
                    "conversation_id": c.get_conversation_id(),
                    "conversation_name": conversation_name,
                    "agent_name": self.agent_name,
                    "user": self.user,
                    "user_input": user_input,
                    "response": final_answer,
                    "timestamp": datetime.now().isoformat(),
                    "prompt_tokens": tokens if "tokens" in locals() else 0,
                },
                user_id=self.user,
                agent_id=self.agent.agent_id,
                agent_name=self.agent_name,
                company_id=self.agent.company_id,
            )

        # Update conversation summary in the background
        if log_output and final_answer and user_input:
            try:
                from MagicalAuth import get_user_id as _get_uid

                _resolved_user_id = str(_get_uid(self.user)) if self.user else ""
                asyncio.create_task(
                    update_conversation_summary_after_interaction(
                        conversation=c,
                        user_input=user_input,
                        agent_response=final_answer,
                        agent_name=self.agent_name,
                        user_id=_resolved_user_id,
                    )
                )
            except Exception as e:
                log_silenced_exception(e, "chat: updating conversation summary")

        # Yield the complete answer
        yield {"type": "answer", "content": final_answer, "complete": True}

    def extract_commands_from_response(self, response):
        # Extract all <execute>...</execute> blocks
        command_blocks = re.findall(r"(<execute>.*?</execute>)", response, re.DOTALL)
        extracted_commands = []
        for command_block in command_blocks:
            # Extract the content inside <execute>...</execute>
            command_content = re.search(
                r"<execute>(.*?)</execute>", command_block, re.DOTALL
            ).group(1)
            # Extract the command name
            name_match = re.search(r"<name>(.*?)</name>", command_content, re.DOTALL)
            if name_match:
                command_name = name_match.group(1).strip()
                # Remove the <name> tag from the command_content
                command_content_without_name = re.sub(
                    r"<name>.*?</name>", "", command_content, flags=re.DOTALL
                )
                # Extract arguments
                arg_matches = re.findall(
                    r"<(.*?)>(.*?)</\1>", command_content_without_name, re.DOTALL
                )
                args = {}
                for arg_name, arg_value in arg_matches:
                    args[arg_name] = arg_value.strip()
                extracted_commands.append((command_block, command_name, args))

        # Also extract standalone <name>...</name> tool calls not wrapped in <execute>
        # This handles cases where the model outputs tool calls in thinking blocks
        # Pattern: <name>command_name</name> followed by optional argument tags
        standalone_pattern = r"<name>\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*</name>"
        for match in re.finditer(standalone_pattern, response, re.IGNORECASE):
            command_name = match.group(1).strip()
            # Check if this name tag is inside an <execute> block (already handled above)
            start_pos = match.start()
            text_before = response[:start_pos]
            # Count open <execute> and </execute> to see if we're inside one
            execute_depth = len(
                re.findall(r"<execute>", text_before, re.IGNORECASE)
            ) - len(re.findall(r"</execute>", text_before, re.IGNORECASE))
            if execute_depth > 0:
                continue  # Skip, already handled by <execute> block extraction

            # This is a standalone tool call - wrap it as an execute block for consistency
            # Extract any adjacent argument-like tags after the name tag
            remaining = response[match.end() :]
            args = {}
            # Look for tags that look like arguments (up to next </thinking>, </answer>, or <name>)
            arg_section_match = re.match(
                r"(.*?)(?=<(?:/thinking|/answer|name|execute))",
                remaining,
                re.DOTALL | re.IGNORECASE,
            )
            if arg_section_match:
                arg_section = arg_section_match.group(1)
                arg_matches = re.findall(
                    r"<([a-zA-Z_][a-zA-Z0-9_]*)>(.*?)</\1>", arg_section, re.DOTALL
                )
                for arg_name, arg_value in arg_matches:
                    args[arg_name.lower()] = arg_value.strip()

            # Create a synthetic execute block for tracking
            synthetic_block = f"<execute><name>{command_name}</name></execute>"
            extracted_commands.append((synthetic_block, command_name, args))

        return extracted_commands

    def _is_remote_command(self, command_output: str) -> dict:
        """
        Check if a command output is a remote command request.

        Remote commands are executed on the client side (e.g., CLI) instead of the server.
        They return a JSON object with __remote_command__: true

        Returns:
            dict with remote command details if it's a remote command, None otherwise
        """
        if not command_output:
            return None
        try:
            # Try to parse as JSON
            parsed = json.loads(command_output)
            if isinstance(parsed, dict) and parsed.get("__remote_command__") == True:
                return parsed
        except (json.JSONDecodeError, TypeError):
            pass
        return None

    def _get_executable_command_names(self) -> list:
        """
        Return the command names that execution_agent is allowed to run.

        Prompt construction and command selection both include commands enabled
        on the user's agent plus commands enabled through the company agent. The
        execution validator must use that same effective command set, otherwise a
        command can be shown in the prompt and then rejected as unknown.
        """
        command_names = []
        seen = set()

        def add_command(command_name):
            if not command_name:
                return
            normalized = str(command_name).strip().lower()
            if not normalized or normalized in seen:
                return
            seen.add(normalized)
            command_names.append(str(command_name).strip())

        def collect_available_commands(available_commands):
            for available_command in available_commands or []:
                try:
                    enabled = (
                        str(available_command.get("enabled", "")).lower() == "true"
                    )
                    if enabled:
                        add_command(available_command.get("friendly_name"))
                except Exception as e:
                    logging.debug(
                        f"[execution_agent] Could not inspect available command: {e}"
                    )

        if self.agent:
            collect_available_commands(getattr(self.agent, "available_commands", []))

            try:
                company_agent = getattr(self.agent, "company_agent", None)
            except Exception as e:
                logging.debug(
                    f"[execution_agent] Could not load company agent commands: {e}"
                )
                company_agent = None

            if company_agent:
                collect_available_commands(
                    getattr(company_agent, "available_commands", [])
                )

        for client_tool_name in (getattr(self, "_client_tools", {}) or {}).keys():
            add_command(client_tool_name)

        return command_names

    async def execution_agent(
        self,
        conversation_name,
        conversation_id=None,
        thinking_id=None,
        remote_command_callback=None,
    ):
        """
        Execute commands found in the agent's response.

        Returns:
            int: Number of commands actually executed (0 means all were deduplicated/skipped).

        Args:
            conversation_name: Name of the conversation
            conversation_id: ID of the conversation
            thinking_id: ID for tracking thinking activities
            remote_command_callback: Optional async callback for handling remote commands.
                                   If provided and a remote command is detected, this callback
                                   will be called with the remote command request dict.
                                   The callback should return the command output string.
        """
        c = Conversations(
            conversation_name=conversation_name,
            user=self.user,
            conversation_id=conversation_id,
        )
        command_list = self._get_executable_command_names()
        # Use provided thinking_id if available, otherwise get a new one
        if not thinking_id:
            thinking_id = c.get_thinking_id(agent_name=self.agent_name)

        # Check for cancellation before starting command execution
        self._check_cancelled()

        # Extract commands from the response
        commands_to_execute = self.extract_commands_from_response(self.response)
        reformatted_response = self.response

        # Get client-defined tools if available
        client_tools = getattr(self, "_client_tools", {})

        # Debug logging for client tools
        if commands_to_execute:
            logging.debug(
                f"[execution_agent] Commands to execute: {[(c[1], c[2]) for c in commands_to_execute]}"
            )

        if commands_to_execute:
            commands_actually_executed = 0
            commands_deduplicated_this_call = 0
            for command_block, command_name, command_args in commands_to_execute:
                # Check for cancellation before each command
                self._check_cancelled()

                # Save original args before any runtime modifications
                # This is used for reformatting the response without runtime metadata
                original_command_args = {k: v for k, v in command_args.items()}

                position = self.response.index(command_block)
                command_id = f"{position}:{command_name}:{json.dumps(command_args, sort_keys=True)}"
                # Skip if we've already processed this exact command
                if command_id in self._processed_commands:
                    commands_deduplicated_this_call += 1
                    self._total_dedup_count += 1
                    logging.warning(
                        f"[execution_agent] Skipping already-processed command: {command_name} "
                        f"(args: {json.dumps(command_args, sort_keys=True)[:200]}). "
                        f"This usually means the model lost context about previous executions "
                        f"and regenerated the same command. "
                        f"(iteration dedupes={commands_deduplicated_this_call}, "
                        f"total dedupes={self._total_dedup_count})"
                    )
                    # Replace the offending command block in the response with an
                    # explicit per-command rejection notice so the model sees on
                    # the NEXT inference exactly which call was skipped and why.
                    # Without this the model only sees its own raw <execute> tag
                    # and has no signal that the call was rejected, which is a
                    # primary driver of the regenerate-same-command loop.
                    rejection_notice = (
                        f"<output>[REJECTED — DUPLICATE] The command `{command_name}` "
                        f"with the exact same arguments was already executed earlier in this "
                        f"task. Its output is shown above. The system rejected this "
                        f"re-execution to avoid redundant work. "
                        f"Either: (a) call a DIFFERENT command, (b) call `{command_name}` "
                        f"with DIFFERENT arguments, or (c) if you have enough information, "
                        f"finalize your answer in <answer></answer> tags. The opening "
                        f"<answer> tag MUST be the first thing on a new line, not inline "
                        f"in a sentence and not inside a code block.</output>"
                    )
                    reformatted_response = reformatted_response.replace(
                        command_block, rejection_notice, 1
                    )
                    continue

                # Mark this command as processed
                self._processed_commands.add(command_id)
                commands_actually_executed += 1

                command_output = ""

                # Check if this is a client-defined tool
                if command_name in client_tools:
                    # This is a client-defined tool - create a remote command request
                    logging.info(
                        f"[execution_agent] Client-defined tool called: {command_name}"
                    )
                    json_args = json.dumps(command_args, indent=2)
                    c.log_interaction(
                        role=self.agent_name,
                        message=f"[SUBACTIVITY][{thinking_id}][CLIENT_TOOL] Calling client tool `{command_name}`.\n```json\n{json_args}```",
                    )

                    # Create a remote command request for the client
                    remote_cmd = {
                        "__remote_command__": True,
                        "tool_name": command_name,
                        "tool_args": command_args,
                        "request_id": str(uuid.uuid4()),
                    }

                    # For execute_terminal_command, map the args to the expected format
                    if command_name == "execute_terminal_command":
                        remote_cmd["command"] = command_args.get("command", "")
                        remote_cmd["terminal_id"] = command_args.get(
                            "terminal_id", str(uuid.uuid4())
                        )
                        remote_cmd["working_directory"] = command_args.get(
                            "working_directory"
                        )
                        remote_cmd["is_background"] = command_args.get(
                            "is_background", False
                        )

                    if remote_command_callback:
                        c.log_interaction(
                            role=self.agent_name,
                            message=f"[SUBACTIVITY][{thinking_id}][REMOTE] Requesting remote execution of client tool...",
                        )
                        try:
                            command_output = await remote_command_callback(remote_cmd)
                            c.log_interaction(
                                role=self.agent_name,
                                message=f"[SUBACTIVITY][{thinking_id}][REMOTE] Remote execution completed.\n```\n{command_output}\n```",
                            )
                        except Exception as e:
                            command_output = f"Client tool execution failed: {str(e)}"
                            c.log_interaction(
                                role=self.agent_name,
                                message=f"[SUBACTIVITY][{thinking_id}][ERROR] Client tool execution failed: {str(e)}",
                            )
                    else:
                        # No callback - emit webhook for external handling
                        await webhook_emitter.emit_event(
                            event_type="command.remote.request",
                            data={
                                "conversation_id": c.get_conversation_id(),
                                "conversation_name": conversation_name,
                                "agent_name": self.agent_name,
                                "user": self.user,
                                "command_name": command_name,
                                "remote_request": remote_cmd,
                                "timestamp": datetime.now().isoformat(),
                            },
                            user_id=self.user,
                            agent_id=self.agent.agent_id,
                            agent_name=self.agent_name,
                            company_id=self.agent.company_id,
                        )
                        command_output = f"[CLIENT TOOL PENDING] This tool requires client-side execution.\nTool: {command_name}\nRequest ID: {remote_cmd.get('request_id', 'unknown')}"

                elif command_name.strip().lower() not in [
                    cmd.lower() for cmd in command_list
                ]:
                    command_output = f"Unknown command: {command_name}"
                    logging.warning(command_output)
                else:
                    try:
                        json_args = json.dumps(command_args, indent=2)
                        c.log_interaction(
                            role=self.agent_name,
                            message=f"[SUBACTIVITY][{thinking_id}][EXECUTION] Executing `{command_name}`.\n```json\n{json_args}```",
                        )
                        ext = Extensions(
                            agent_name=self.agent_name,
                            agent_id=self.agent.agent_id,
                            agent_config=self.agent.AGENT_CONFIG,
                            conversation_name=conversation_name,
                            conversation_id=c.get_conversation_id(),
                            ApiClient=self.ApiClient,
                            api_key=self.ApiClient.headers["Authorization"],
                            user=self.user,
                        )
                        command_args["activity_id"] = thinking_id
                        command_output = await ext.execute_command(
                            command_name=command_name,
                            command_args=command_args,
                        )

                        # Check if this is a remote command that needs client-side execution
                        remote_cmd = self._is_remote_command(command_output)
                        if remote_cmd:
                            # This is a remote command - needs to be executed on the client
                            if remote_command_callback:
                                # Use callback to handle remote execution
                                c.log_interaction(
                                    role=self.agent_name,
                                    message=f"[SUBACTIVITY][{thinking_id}][REMOTE] Requesting remote execution of terminal command...",
                                )
                                try:
                                    # Call the callback with the remote command request
                                    command_output = await remote_command_callback(
                                        remote_cmd
                                    )
                                    c.log_interaction(
                                        role=self.agent_name,
                                        message=f"[SUBACTIVITY][{thinking_id}][REMOTE] Remote execution completed.\n```\n{command_output}\n```",
                                    )
                                except Exception as e:
                                    command_output = (
                                        f"Remote command execution failed: {str(e)}"
                                    )
                                    c.log_interaction(
                                        role=self.agent_name,
                                        message=f"[SUBACTIVITY][{thinking_id}][ERROR] Remote execution failed: {str(e)}",
                                    )
                            else:
                                # No callback - emit webhook for external handling
                                await webhook_emitter.emit_event(
                                    event_type="command.remote.request",
                                    data={
                                        "conversation_id": c.get_conversation_id(),
                                        "conversation_name": conversation_name,
                                        "agent_name": self.agent_name,
                                        "user": self.user,
                                        "command_name": command_name,
                                        "remote_request": remote_cmd,
                                        "timestamp": datetime.now().isoformat(),
                                    },
                                    user_id=self.user,
                                    agent_id=self.agent.agent_id,
                                    agent_name=self.agent_name,
                                    company_id=self.agent.company_id,
                                )
                                # Format as pending remote execution for the response
                                command_output = f"[REMOTE COMMAND PENDING] This command requires execution on the client.\nRequest ID: {remote_cmd.get('request_id', 'unknown')}\nCommand: {remote_cmd.get('command', 'unknown')}"
                        # Handle OPTIMIZE_COMMANDS signal from optimize_command_selection
                        if isinstance(
                            command_output, str
                        ) and command_output.startswith("OPTIMIZE_COMMANDS:"):
                            task_desc = command_output[
                                len("OPTIMIZE_COMMANDS:") :
                            ].strip()
                            logging.info(
                                f"[execution_agent] OPTIMIZE_COMMANDS signal received: {task_desc}"
                            )
                            try:
                                # Build conversation history for context
                                _conv_data = c.get_conversation(limit=1000, page=1)
                                _opt_history = ""
                                if (
                                    "interactions" in _conv_data
                                    and _conv_data["interactions"]
                                ):
                                    _recent = [
                                        i
                                        for i in _conv_data["interactions"]
                                        if not str(i.get("message", "")).startswith(
                                            "[ACTIVITY]"
                                        )
                                        and not str(i.get("message", "")).startswith(
                                            "[SUBACTIVITY]"
                                        )
                                        and not str(i.get("message", "")).startswith(
                                            "<audio"
                                        )
                                    ][-6:]
                                    if _recent:
                                        _opt_history = "\n".join(
                                            f"{m.get('role', '')}: {str(m.get('message', ''))[:300]}"
                                            for m in _recent
                                        )
                                new_commands = await self.select_commands_for_task(
                                    user_input=task_desc,
                                    conversation_name=conversation_name,
                                    log_output=True,
                                    thinking_id=thinking_id,
                                    conversation_history=_opt_history,
                                )
                                if new_commands:
                                    self._selected_commands = new_commands
                                    self._cached_commands_prompt = (
                                        None  # Invalidate cache
                                    )
                                    command_output = (
                                        f"Command selection re-optimized for: {task_desc}\n"
                                        f"Now available: {', '.join(new_commands)}"
                                    )
                                else:
                                    command_output = "Command re-optimization returned no commands. All commands remain available."
                                    self._selected_commands = None
                                    self._cached_commands_prompt = (
                                        None  # Invalidate cache
                                    )
                            except Exception as e:
                                logging.error(
                                    f"[execution_agent] OPTIMIZE_COMMANDS failed: {e}"
                                )
                                command_output = f"Command re-optimization failed: {e}. All commands remain available."
                                self._selected_commands = None
                                self._cached_commands_prompt = None  # Invalidate cache

                        # Handle different types of command output
                        if isinstance(command_output, (dict, list)):
                            # Already a structured object, serialize directly to JSON
                            try:
                                command_output = json.dumps(
                                    command_output, indent=2, ensure_ascii=False
                                )
                                # Wrap in json code block for better formatting
                                command_output = "```json\n" + command_output + "\n```"
                            except Exception as e:
                                # Fallback to string representation
                                command_output = str(command_output)
                        else:
                            # Convert to string and handle any ``` characters
                            command_output = str(command_output)
                            command_output = command_output.replace("```", "``'")
                            try:
                                # Try to parse as JSON in case it's a JSON string
                                parsed = json.loads(command_output)
                                command_output = json.dumps(
                                    parsed, indent=2, ensure_ascii=False
                                )
                                # Wrap in json code block for better formatting
                                command_output = "```json\n" + command_output + "\n```"
                            except:
                                # Not valid JSON — wrap in a plain code block to prevent
                                # raw markdown (e.g. ## headings) from rendering in the UI
                                command_output = "```\n" + command_output + "\n```"

                        c.log_interaction(
                            role=self.agent_name,
                            message=f"[SUBACTIVITY][{thinking_id}][EXECUTION] `{command_name}` was executed successfully.\n{command_output}",
                        )

                        # Emit webhook event for successful command execution
                        await webhook_emitter.emit_event(
                            event_type="command.execution.completed",
                            data={
                                "conversation_id": c.get_conversation_id(),
                                "conversation_name": conversation_name,
                                "agent_name": self.agent_name,
                                "user": self.user,
                                "command_name": command_name,
                                "command_args": command_args,
                                "output": command_output,
                                "status": "success",
                                "timestamp": datetime.now().isoformat(),
                            },
                            user_id=self.user,
                            agent_id=self.agent.agent_id,
                            agent_name=self.agent_name,
                            company_id=self.agent.company_id,
                        )
                    except Exception as e:
                        error_message = f"Error: {self.agent_name} failed to execute command `{command_name}`. {e}"
                        logging.error(error_message)
                        c.log_interaction(
                            role=self.agent_name,
                            message=f"[SUBACTIVITY][{thinking_id}][ERROR] Failed to execute command `{command_name}`.\n{error_message}",
                        )
                        command_output = error_message

                        # Emit webhook event for failed command execution
                        await webhook_emitter.emit_event(
                            event_type="command.execution.failed",
                            data={
                                "conversation_id": c.get_conversation_id(),
                                "conversation_name": conversation_name,
                                "agent_name": self.agent_name,
                                "user": self.user,
                                "command_name": command_name,
                                "command_args": command_args,
                                "error": str(e),
                                "timestamp": datetime.now().isoformat(),
                            },
                            user_id=self.user,
                            agent_id=self.agent.agent_id,
                            agent_name=self.agent_name,
                            company_id=self.agent.company_id,
                        )
                # Format the command execution and output
                # Use original_command_args (without runtime metadata like activity_id)
                # to prevent re-extraction issues on subsequent passes
                formatted_execution = (
                    f"<execute>\n"
                    f"<name>{command_name}</name>\n"
                    f"{chr(10).join([f'<{k}>{v}</{k}>' for k, v in original_command_args.items()])}\n"
                    f"</execute>\n"
                    f"<output>{command_output}</output>"
                )

                # Replace the original command block with the formatted execution and output
                reformatted_response = reformatted_response.replace(
                    command_block, formatted_execution, 1
                )
        else:
            cmds = "\n".join(command_list)
            self.response += f"\nThe assistant tried to execute a command, but it was not recognized. Ensure that the correct naming of the commands is being used, they go off of the friendly name. Please choose from the list of available commands and try again:\n{cmds}"
            return 0
        if reformatted_response != self.response:
            self.response = reformatted_response
        # Expose dedup count for this call so run_stream can detect stuck loops
        # even when at least one new command slipped through each iteration.
        self._last_iteration_dedup_count = commands_deduplicated_this_call
        return commands_actually_executed
