fix(chat): sanitize web search query to strip markdown and code blocks (#4863)

Layer a defensive cleanup on top of the generated-query web-search flow so the final selected query is sanitized before reaching comprehensive_web_search.

- remove fenced code blocks from the final search query
- preserve inline code as plain text
- collapse whitespace and cap query length
- cover generated-query success plus LLM failure/empty fallback paths

Partially addresses #4547.

mcp_servers/image_gen_server.py

@@ -73,7 +73,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
         if not model_spec:
             for candidate in ("gpt-image-1.5", "gpt-image-1", "dall-e-3"):
                 try:
-                    _resolve_model(candidate)
+                    await asyncio.to_thread(_resolve_model, candidate)
                     model_spec = candidate
                     break
                 except ValueError:
@@ -81,7 +81,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
             if not model_spec:
                 return [TextContent(type="text", text="Error: No image model found. Configure one in Admin.")]
 
-        url, model_id, headers = _resolve_model(model_spec)
+        url, model_id, headers = await asyncio.to_thread(_resolve_model, model_spec)
 
         is_gpt_image = "gpt-image" in model_id.lower()
         base_url = url.replace("/chat/completions", "").replace("/v1/messages", "").rstrip("/")

routes/email_helpers.py

@@ -40,6 +40,16 @@ from src.secret_storage import decrypt as _decrypt
 logger = logging.getLogger(__name__)
 
 
+class EmailNotConfiguredError(RuntimeError):
+    """Raised when an IMAP operation is attempted on an account that has no
+    inbox configured (e.g. a send-only / SMTP-only account).
+
+    Subclasses RuntimeError so existing broad ``except Exception`` handlers
+    keep working; callers that want to treat "no inbox" as an empty result
+    rather than a failure can catch this type specifically.
+    """
+
+
 def _xoauth2_raw(user: str, access_token: str) -> str:
     """The SASL XOAUTH2 initial-response string (unencoded).
 
@@ -929,6 +939,14 @@ def _imap_connect(account_id: str | None = None, owner: str = "",
     # `timeout` is overridable so short-lived callers (e.g. the service-health
     # probe) can impose a tighter budget than the default IMAP timeout.
     cfg = _get_email_config(account_id, owner=owner)
+    # Send-only (SMTP-only) account: no IMAP host means there is no inbox to
+    # read. Bail out with a clear, typed error instead of handing an empty
+    # host to imaplib — IMAP4("", 993) silently dials localhost:993 and fails
+    # with a confusing "[Errno 111] Connection refused" on every inbox poll.
+    if not cfg.get("imap_host"):
+        raise EmailNotConfiguredError(
+            f"IMAP is not configured for account {cfg.get('account_name') or 'default'!r}"
+        )
     # Connection mode:
     #   STARTTLS on → plain + upgrade
     #   STARTTLS off + port 993 → implicit SSL (IMAPS)

routes/email_routes.py

@@ -46,6 +46,7 @@ from routes.email_helpers import (
     _send_smtp_message, _smtp_security_mode,
     _IMAP_TIMEOUT_SECONDS, _open_imap_connection,
     make_oauth_state, verify_oauth_state,
+    EmailNotConfiguredError,
     _imap_connect, _imap, _decode_header, _detect_sent_folder, _detect_drafts_folder,
     _extract_attachment_text, _list_attachments_from_msg, _has_visible_attachments, _is_likely_signature_image_attachment,
     _extract_attachment_to_disk, _extract_html, _extract_text,
@@ -1029,6 +1030,11 @@ def setup_email_routes():
                 logger.debug(f"Bulk summary attach skipped: {_summary_err}")
 
             return {"emails": emails, "total": total, "folder": folder, "offset": offset}
+        except EmailNotConfiguredError:
+            # Send-only (SMTP-only) account: there is no inbox to read, so the
+            # poll returns an empty list instead of a per-minute error. SMTP
+            # send is unaffected.
+            return {"emails": [], "total": 0, "folder": folder, "offset": offset}
         except Exception as e:
             logger.error(f"Failed to list emails: {e}")
             detail = str(e).strip()

routes/preset_routes.py

@@ -1,5 +1,6 @@
 """Preset routes — /api/presets GET, /api/presets/custom POST, user templates CRUD."""
 
+import asyncio
 import logging
 import uuid
 from typing import Dict, Any, List
@@ -102,7 +103,7 @@ def setup_preset_routes(preset_manager) -> APIRouter:
         try:
             model_spec = data.get("model") or ""
             user = effective_user(request)
-            url, model, headers = _resolve_model(model_spec, owner=user)
+            url, model, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=user)
             result = await llm_call_async(url, model, messages, temperature=0.8, max_tokens=500, headers=headers)
             return {"success": True, "prompt": result.strip()}
         except Exception as e:

routes/shell_routes.py

@@ -1063,8 +1063,19 @@ def setup_shell_routes() -> APIRouter:
         importlib.invalidate_caches()
         try:
             user_site = site.getusersitepackages()
-            if user_site and os.path.isdir(user_site) and user_site not in sys.path:
+            if user_site and os.path.isdir(user_site):
-                sys.path.append(user_site)
+                # Use addsitedir(), NOT a bare sys.path.append(). When a package
+                # is `pip install --user`'d at runtime (Cookbook → Install) the
+                # long-lived server process started before the user-site existed,
+                # so site never processed it — including its `.pth` hooks. On
+                # Python 3.12+ `distutils` is gone from stdlib and is only
+                # restored by setuptools' `distutils-precedence.pth`, which ships
+                # in user-site. basicsr (a realesrgan dep) does `import distutils`
+                # at import time, so a plain append left the package importable
+                # but `import distutils` failing → realesrgan probed as
+                # not-installed until a full process restart. addsitedir() replays
+                # the `.pth` files so the shim is active.
+                site.addsitedir(user_site)
         except Exception:
             pass
         if ssh_port and str(ssh_port).strip() not in ("", "22"):

src/agent_loop.py

@@ -1648,6 +1648,7 @@ def _build_base_prompt(
 def _resolve_tool_blocks(round_response: str, native_tool_calls: list, round_num: int, is_api_model: bool = False):
     """Choose native function calls or fenced code block parsing. Returns (tool_blocks, used_native)."""
     used_native = False
+    converted_calls = []  # native calls that converted, ALIGNED with tool_blocks
     if native_tool_calls:
         tool_blocks = []
         for tc in native_tool_calls:
@@ -1656,6 +1657,7 @@ def _resolve_tool_blocks(round_response: str, native_tool_calls: list, round_num
             block = function_call_to_tool_block(tc_name, tc_args)
             if block:
                 tool_blocks.append(block)
+                converted_calls.append(tc)
                 logger.info(f"  -> converted: {tc_name} -> {block.tool_type}")
             else:
                 logger.warning(f"  -> FAILED to convert native call: {tc_name} args={tc_args[:200]}")
@@ -1685,7 +1687,7 @@ def _resolve_tool_blocks(round_response: str, native_tool_calls: list, round_num
                 f"{len(native_tool_calls)} native calls, "
                 f"{len(tool_blocks)} tool blocks. Preview: {resp_preview}")
 
-    return tool_blocks, used_native
+    return tool_blocks, used_native, converted_calls
 
 
 def _append_tool_results(
@@ -2868,7 +2870,7 @@ async def stream_agent_loop(
             _round_first_event_logged,
             _round_first_token_logged,
         )
-        tool_blocks, used_native = _resolve_tool_blocks(
+        tool_blocks, used_native, converted_calls = _resolve_tool_blocks(
             round_response,
             native_tool_calls,
             round_num,
@@ -3500,7 +3502,12 @@ async def stream_agent_loop(
             break
 
         # Feed results back to LLM for next round
-        _append_tool_results(messages, round_response, native_tool_calls,
+        # Pass the CONVERTED calls (aligned 1:1 with tool_result_texts), not the
+        # raw native_tool_calls: a call that failed to convert is dropped from
+        # tool_blocks but stayed in native_tool_calls, so indexing results by
+        # native position mis-attached each result to the wrong tool_call_id
+        # (and left the real call answered empty).
+        _append_tool_results(messages, round_response, converted_calls,
                              tool_results, tool_result_texts, used_native, round_num,
                              round_reasoning=round_reasoning)
 

src/agent_tools/model_interaction_tools.py

@@ -10,6 +10,7 @@ Shared helpers that still live in ``src.ai_interaction`` and are used by tools
 not yet migrated (``_resolve_model``, ``AI_CHAT_TIMEOUT``) are imported lazily
 inside the functions to avoid an import cycle at module load.
 """
+import asyncio
 import logging
 from typing import Dict, Optional
 
@@ -46,7 +47,7 @@ async def chat_with_model(content: str, session_id: Optional[str] = None, owner:
         return {"error": "No message provided (line 2+ is the message)"}
 
     try:
-        url, model, headers = _resolve_model(model_spec, owner=owner)
+        url, model, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
     except ValueError as e:
         return {"error": str(e)}
 
@@ -90,7 +91,7 @@ async def ask_teacher(content: str, session_id: Optional[str] = None, owner: Opt
             return {"error": "No teacher model configured. Specify a model name or set teacher_model in settings."}
 
     try:
-        url, model, headers = _resolve_model(model_spec, owner=owner)
+        url, model, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
     except ValueError as e:
         return {"error": str(e)}
 

src/agent_tools/session_tools.py

@@ -8,6 +8,7 @@ The session manager is a runtime-set singleton in src.ai_interaction, so each
 function fetches it via get_session_manager() (imported here); _resolve_model and
 AI_CHAT_TIMEOUT are reused from there too.
 """
+import asyncio
 import json
 import logging
 import uuid
@@ -40,7 +41,7 @@ async def create_session(content: str, session_id: Optional[str] = None, owner:
         return {"error": "Session name cannot be empty"}
 
     try:
-        url, model, headers = _resolve_model(model_spec, owner=owner)
+        url, model, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
     except ValueError as e:
         return {"error": str(e)}
 

src/ai_interaction.py

@@ -14,6 +14,7 @@ These are agent tools — the LLM writes fenced code blocks and they execute
 through the standard agent_tools.py pipeline.
 """
 
+import asyncio
 import json
 import logging
 import uuid
@@ -229,7 +230,7 @@ async def do_pipeline(content: str, session_id: Optional[str] = None, owner: Opt
         if not model_spec or not instruction:
             return {"error": f"Step {i + 1}: both 'model' and 'instruction' are required"}
         try:
-            url, model, headers = _resolve_model(model_spec, owner=owner)
+            url, model, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
             resolved.append((url, model, headers, instruction))
         except ValueError as e:
             return {"error": f"Step {i + 1}: {e}"}
@@ -624,7 +625,7 @@ async def do_ui_control(content: str, session_id: Optional[str] = None, owner: O
 
         # Resolve the model to validate it exists
         try:
-            url, model_id, headers = _resolve_model(model_spec, owner=owner)
+            url, model_id, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
         except ValueError as e:
             return {"error": str(e)}
 
@@ -914,7 +915,7 @@ async def do_generate_image(content: str, session_id: Optional[str] = None, owne
     if not model_spec:
         for candidate in ("gpt-image-1.5", "gpt-image-1", "dall-e-3"):
             try:
-                _resolve_model(candidate, owner=owner)
+                await asyncio.to_thread(_resolve_model, candidate, owner=owner)
                 model_spec = candidate
                 break
             except ValueError:
@@ -958,7 +959,7 @@ async def do_generate_image(content: str, session_id: Optional[str] = None, owne
 
     # Resolve the model to find the right endpoint
     try:
-        url, model_id, headers = _resolve_model(model_spec, owner=owner)
+        url, model_id, headers = await asyncio.to_thread(_resolve_model, model_spec, owner=owner)
     except ValueError:
         return {"error": f"No endpoint found with image model '{model_spec}'. "
                 "Configure an OpenAI-compatible endpoint with image generation support."}

src/app_initializer.py

@@ -68,8 +68,10 @@ def initialize_managers(base_dir: str, rag_manager=None) -> Dict[str, Any]:
                     logger.info(f"Rebuilt memory vector index from {len(existing)} existing entries")
             logger.info("MemoryVectorStore initialized")
         else:
+            # Keep the unhealthy object (do NOT reset to None): consumers gate on
+            # `.healthy`, and service_health.chromadb_health() needs a present
+            # object to report DEGRADED/DOWN instead of DISABLED ("not configured").
             logger.warning("MemoryVectorStore DEGRADED: ChromaDB vector memory unavailable")
-            memory_vector = None
     except Exception as e:
         logger.warning(f"MemoryVectorStore DEGRADED: {e}")
         memory_vector = None

src/builtin_mcp.py

@@ -89,6 +89,21 @@ _BUILTIN_NPX_SERVERS = {
 MCP_DISABLED = os.environ.get("ODYSSEUS_DISABLE_MCP", "").lower() in ("1", "true", "yes")
 
 
+# Strong references to the fire-and-forget startup tasks scheduled below.
+# asyncio only keeps weak references to tasks created via create_task, so
+# without this the GC can collect a task mid-execution and the server
+# registration silently never runs. Mirrors _spawn_bg in routes/chat_helpers.py.
+_BG_TASKS: set[asyncio.Task] = set()
+
+
+def _spawn_bg(coro) -> asyncio.Task:
+    """Schedule a background task and hold a strong reference until it finishes."""
+    task = asyncio.create_task(coro)
+    _BG_TASKS.add(task)
+    task.add_done_callback(_BG_TASKS.discard)
+    return task
+
+
 async def register_builtin_servers(mcp_manager):
     """Connect all built-in MCP servers to the manager."""
     if MCP_DISABLED:
@@ -123,7 +138,7 @@ async def register_builtin_servers(mcp_manager):
         if not os.path.exists(script_path):
             logger.warning(f"Built-in MCP server script not found: {script_path}")
             continue
-        asyncio.create_task(_connect_python_server(server_id, script_path, name))
+        _spawn_bg(_connect_python_server(server_id, script_path, name))
 
     # Register NPX-based servers in the background (they take longer to start)
     npx_path = _find_npx()
@@ -175,7 +190,7 @@ async def register_builtin_servers(mcp_manager):
             except BaseException as e:
                 logger.warning(f"Built-in NPX server {cfg['name']} error: {type(e).__name__}: {e}")
 
-    asyncio.create_task(_start_npx_servers())
+    _spawn_bg(_start_npx_servers())
 
 
 def _npx_package_from_args(args):
@@ -233,6 +248,15 @@ async def _is_npx_package_cached(npx_path, package_spec, timeout_s=5):
         except Exception:
             pass
         return False
+    except asyncio.CancelledError:
+        # The probe was cancelled (e.g. app shutdown). Reap the child so it
+        # isn't orphaned, then propagate the cancellation.
+        try:
+            proc.kill()
+            await proc.wait()
+        except Exception:
+            pass
+        raise
     return proc.returncode == 0 and bool(stdout.strip())
 
 

src/chat_processor.py

@@ -12,6 +12,45 @@ from src.prompt_security import UNTRUSTED_CONTEXT_POLICY, untrusted_context_mess
 
 logger = logging.getLogger(__name__)
 
+
+def _clean_search_query(query: str, max_len: int = 200) -> str:
+    """Strip fenced code blocks from a search query while preserving inline
+    code text.
+
+    This is a focused, defensive cleanup for the *final* web-search query
+    selected in ``build_context_preface`` (issue #4547): regardless of whether
+    the query came from the LLM-generated path (#4557) or the first-line
+    fallback, residual fenced / inline markdown should not leak into the search
+    call. Rather than using regex (which is brittle and strips inline code
+    text like ``git reset`` from the query), we render the query to HTML via
+    ``markdown`` and parse it with ``BeautifulSoup`` so that:
+
+    * ``<pre>`` blocks (fenced / indented code) are removed entirely.
+    * ``<code>`` elements (inline code) are preserved as plain text.
+
+    Both libraries are already project dependencies. The result is whitespace
+    collapsed and truncated to ``max_len``; an all-code input collapses to an
+    empty string, which the caller treats as "no query".
+    """
+    import markdown as _md
+    from bs4 import BeautifulSoup as _BS
+
+    html = _md.markdown(query, extensions=["fenced_code"])
+    soup = _BS(html, "html.parser")
+
+    # Remove fenced / indented code blocks.
+    for pre in soup.find_all("pre"):
+        pre.decompose()
+
+    # Preserve inline code by unwrapping <code> to text.
+    for code in soup.find_all("code"):
+        code.replace_with(code.get_text())
+
+    text = soup.get_text(" ", strip=True)
+    text = re.sub(r"\s+", " ", text)
+    return text[:max_len]
+
+
 # ── Stopwords & tokenizer ──
 
 _STOPWORDS = frozenset(
@@ -322,6 +361,13 @@ class ChatProcessor:
                 if len(search_query) > 150:
                     search_query = search_query[:150].strip()
 
+                # Defensive cleanup of the final selected query (interim fix
+                # for #4547): strip any residual fenced/inline markdown so that
+                # neither the generated query nor the first-line fallback leaks
+                # fences or backticks into the search call. No-op on clean
+                # generated queries; collapses to "" when the query is all code.
+                search_query = _clean_search_query(search_query, max_len=150)
+
                 if search_query:
                     # Execute web search using the final selected query
                     web_context, web_sources = comprehensive_web_search(

src/task_scheduler.py

@@ -1450,19 +1450,18 @@ class TaskScheduler:
                     system_prompt = f"{char_prompt}\n\n{system_prompt}"
             except Exception:
                 pass
-        # Inject current time so the model knows what's past vs upcoming
+        # Provide current date/time as a user-role message so the system prompt
+        # stays byte-identical across runs and doesn't bust the Anthropic prompt
+        # cache on every scheduled tick (see issue #2927 and the identical fix on
+        # the interactive-chat path in src/agent_loop.py).  The message is built
+        # once here and shared by both execution paths below (agent loop and the
+        # direct fallback) so time grounding is never lost on either path.
         tz_name = _resolve_task_timezone(db, task)
         try:
-            if tz_name:
+            from src.user_time import current_datetime_context_message_for_tz
-                from zoneinfo import ZoneInfo
+            _dt_msg: dict | None = current_datetime_context_message_for_tz(tz_name)
-                from datetime import timezone
-                now_local = _utcnow().replace(tzinfo=timezone.utc).astimezone(ZoneInfo(tz_name))
-                time_str = now_local.strftime("%A, %B %d %Y, %H:%M %Z")
-            else:
-                time_str = _utcnow().strftime("%A, %B %d %Y, %H:%M UTC")
         except Exception:
-            time_str = _utcnow().strftime("%A, %B %d %Y, %H:%M UTC")
+            _dt_msg = None
-        system_prompt = f"Current time: {time_str}\n\n{system_prompt}"
 
         # Compute the disabled-tools set: the crew's enabled_tools allowlist
         # (inverted) plus the operator's global disabled_tools setting. The
@@ -1510,14 +1509,15 @@ class TaskScheduler:
                 endpoint_url, model, task, session_id,
                 system_prompt=system_prompt, disabled_tools=disabled_tools or None,
                 relevant_tools=relevant_tools,
+                datetime_context_msg=_dt_msg,
             )
         except Exception as e:
             logger.warning(f"Agent loop failed for task '{task.name}', falling back to simple call: {e}")
             from src.task_endpoint import task_llm_call_async
-            messages = [
+            messages: list = [{"role": "system", "content": system_prompt}]
-                {"role": "system", "content": system_prompt},
+            if _dt_msg:
-                {"role": "user", "content": task.prompt},
+                messages.append(_dt_msg)
-            ]
+            messages.append({"role": "user", "content": task.prompt})
             result = await task_llm_call_async(
                 messages,
                 fallback_url=endpoint_url,
@@ -1715,16 +1715,20 @@ class TaskScheduler:
                               system_prompt: str | None = None,
                               disabled_tools: set | None = None,
                               relevant_tools: set | None = None,
-                              override_user_message: str | None = None) -> str:
+                              override_user_message: str | None = None,
+                              datetime_context_msg: dict | None = None) -> str:
         """Run the full agent loop with tool access, collecting the final text."""
         from src.agent_loop import stream_agent_loop
 
         system_content = system_prompt or "You are a helpful assistant executing a scheduled task. Use available tools to complete the task thoroughly."
         user_content = override_user_message or task.prompt
-        messages = [
+        # Build the message list. The datetime context message (user-role) is
-            {"role": "system", "content": system_content},
+        # inserted immediately before the task prompt so the system prefix stays
-            {"role": "user", "content": user_content},
+        # byte-identical and cacheable across runs (see issue #2927).
-        ]
+        messages: list = [{"role": "system", "content": system_content}]
+        if datetime_context_msg:
+            messages.append(datetime_context_msg)
+        messages.append({"role": "user", "content": user_content})
 
         # Resolve headers from the endpoint's API key
         headers = {}

src/teacher_escalation.py

@@ -235,7 +235,7 @@ async def _call_teacher(teacher_model_spec: str, prompt: str,
     from src.llm_core import llm_call_async
     from src.ai_interaction import _resolve_model, _TEACHER_SYSTEM_PROMPT
     try:
-        url, model, headers = _resolve_model(teacher_model_spec, owner=owner)
+        url, model, headers = await asyncio.to_thread(_resolve_model, teacher_model_spec, owner=owner)
     except Exception as e:
         logger.warning(f"teacher endpoint not resolvable ({teacher_model_spec!r}): {e}")
         return None
@@ -619,7 +619,7 @@ async def run_teacher_inline(
     # Resolve teacher endpoint
     try:
         from src.ai_interaction import _resolve_model
-        teacher_url, teacher_model, teacher_headers = _resolve_model(teacher_spec, owner=owner)
+        teacher_url, teacher_model, teacher_headers = await asyncio.to_thread(_resolve_model, teacher_spec, owner=owner)
     except Exception as e:
         logger.warning(f"teacher endpoint not resolvable ({teacher_spec!r}): {e}")
         yield (

src/tool_parsing.py

@@ -6,6 +6,7 @@ Supports fenced code blocks, [TOOL_CALL] blocks, and XML-style <invoke> blocks.
 """
 
 import ast
+import bisect
 import json
 import logging
 import re
@@ -70,6 +71,27 @@ _XML_DIRECT_TOOL_RE = re.compile(
     r"<\s*([A-Za-z_][\w-]*)\s*>([\s\S]*?)</\s*\1\s*>",
     re.IGNORECASE,
 )
+# Forward-only delimiters for the lazy XML patterns above, so untrusted "many
+# openers, no closer" model output can't drive finditer's O(n^2) lazy rescan
+# (CodeQL py/polynomial-redos). Consumed by _iter_xml_invoke / _iter_xml_direct.
+_XML_INVOKE_OPEN_RE = re.compile(r'<invoke\s+name=["\'](\w+)["\']>\s*', re.IGNORECASE)
+_XML_INVOKE_CLOSE_RE = re.compile(r'</invoke>', re.IGNORECASE)
+_XML_DIRECT_OPEN_RE = re.compile(r"<\s*([A-Za-z_][\w-]*)\s*>", re.IGNORECASE)
+# Split <parameter ...>...</parameter> delimiters: the parameter scan inside an
+# invoke body is forward-only too, so a closed invoke stuffed with unclosed
+# parameter openers can't drive finditer's O(n^2) rescan. See _iter_named_blocks.
+_XML_PARAM_OPEN_RE = re.compile(r'<parameter\s+name=["\'](\w+)["\']>', re.IGNORECASE)
+_XML_PARAM_CLOSE_RE = re.compile(r'</parameter>', re.IGNORECASE)
+# Closer tokens (any tag name) for the backref scanners, pre-indexed by name so a
+# flood of distinct unclosed tag names stays near-linear. See _iter_backref_blocks.
+_XML_DIRECT_CLOSE_ANY_RE = re.compile(r"</\s*([A-Za-z_][\w-]*)\s*>", re.IGNORECASE)
+# `args => { ... }` opener (its closer is the last `}`, found with rfind) and the
+# `<tag>` opener for tool_code XML params — both split out of greedy/backref
+# patterns that finditer would otherwise rescan from every opener. See
+# _parse_tool_call_block / _parse_tool_code_block.
+_ARGS_BRACE_OPEN_RE = re.compile(r'args\s*(?:=>|:|=)\s*\{')
+_TOOL_CODE_PARAM_OPEN_RE = re.compile(r"<(\w+)>")
+_TOOL_CODE_PARAM_CLOSE_ANY_RE = re.compile(r"</(\w+)>")
 
 # Pattern 3b: StepFun Step-3.x native tool-call tokens. The tokenizer defines:
 #   <｜tool▁calls▁begin｜> ... <｜tool▁calls▁end｜>
@@ -507,11 +529,15 @@ def _parse_tool_call_block(raw: str) -> Optional[ToolBlock]:
         if cmd_match:
             content = cmd_match.group(1)
 
-    # Pattern: args => {content} — extract everything inside the nested braces
+    # Pattern: args => {content} — extract everything inside the nested braces.
+    # Find the opener, then take through the LAST `}` (rfind). Equivalent to the
+    # greedy `\{([\s\S]*)\}` capture, but the bounded opener + rfind avoids
+    # finditer rescanning from every `args:{` opener (CodeQL py/polynomial-redos).
     if not content:
-        args_match = re.search(r'args\s*(?:=>|:|=)\s*\{([\s\S]*)\}', raw, re.DOTALL)
+        am = _ARGS_BRACE_OPEN_RE.search(raw)
-        if args_match:
+        close = raw.rfind('}')
-            inner = args_match.group(1).strip()
+        if am and close >= am.end():
+            inner = raw[am.end():close].strip()
             # Strip quotes and key prefixes
             inner = re.sub(r'^--?\w+\s+', '', inner)
             inner = inner.strip('\'"')
@@ -539,8 +565,8 @@ def _parse_tool_call_block(raw: str) -> Optional[ToolBlock]:
     return None
 
 
-def _parse_xml_invoke(inv_match) -> Optional[ToolBlock]:
+def _parse_xml_invoke(name, body) -> Optional[ToolBlock]:
-    """Parse an <invoke name="tool"><parameter ...>...</parameter></invoke> match.
+    """Parse an <invoke name="tool"><parameter ...>...</parameter></invoke> call.
 
     Delegates content-shaping to function_call_to_tool_block — the SAME
     converter used for native function calls — so the full tool set (every
@@ -555,17 +581,16 @@ def _parse_xml_invoke(inv_match) -> Optional[ToolBlock]:
     # (e.g. <invoke name="Bash">) and function_call_to_tool_block matches
     # case-sensitively against the lowercase _TOOL_NAME_MAP / TOOL_TAGS, so a
     # raw capitalized name would be silently dropped.
-    tool_name = inv_match.group(1).lower()
+    tool_name = name.lower()
-    body = inv_match.group(2)
     params = {}
-    for pm in _XML_PARAM_RE.finditer(body):
+    for pname, pval in _iter_named_blocks(body, _XML_PARAM_OPEN_RE, _XML_PARAM_CLOSE_RE):
-        params[pm.group(1)] = pm.group(2).strip()
+        params[pname] = pval.strip()
     # Local import to avoid a circular import at module load.
     from src.tool_schemas import function_call_to_tool_block
     return function_call_to_tool_block(tool_name, json.dumps(params))
 
 
-def _parse_xml_direct_tool(tool_match) -> Optional[ToolBlock]:
+def _parse_xml_direct_tool(name, body) -> Optional[ToolBlock]:
     """Parse direct XML tool tags inside <tool_call>.
 
     Some local models emit:
@@ -575,13 +600,13 @@ def _parse_xml_direct_tool(tool_match) -> Optional[ToolBlock]:
     Keep this as an adapter to the canonical function-call converter so aliases
     and per-tool argument formatting stay in one place.
     """
-    tool_name = tool_match.group(1).lower().replace("-", "_")
+    tool_name = name.lower().replace("-", "_")
     if tool_name in {"invoke", "parameter", "tool_call", "function_call"}:
         return None
     mapped = _TOOL_NAME_MAP.get(tool_name) or (tool_name if tool_name in TOOL_TAGS else None)
     if not mapped:
         return None
-    body = tool_match.group(2).strip()
+    body = body.strip()
     if not body:
         return None
     try:
@@ -716,10 +741,12 @@ def _parse_tool_code_block(raw: str) -> Optional[ToolBlock]:
     args_match = re.search(r"args\s*=>\s*['\"]?\s*([\s\S]*?)\s*['\"]?\s*$", raw, re.DOTALL)
     args_body = args_match.group(1).strip().strip("'\"") if args_match else ""
 
-    # Parse XML params inside args (e.g. <command>ls</command>)
+    # Parse XML params inside args (e.g. <command>ls</command>). Forward-only
+    # backref scan so a `<x><x>...` opener flood can't drive the O(n^2) lazy
+    # rescan (CodeQL py/polynomial-redos); see _iter_backref_blocks.
     xml_params = {}
-    for pm in re.finditer(r"<(\w+)>([\s\S]*?)</\1>", args_body):
+    for pname, pval in _iter_backref_blocks(args_body, _TOOL_CODE_PARAM_OPEN_RE, _TOOL_CODE_PARAM_CLOSE_ANY_RE):
-        xml_params[pm.group(1)] = pm.group(2).strip()
+        xml_params[pname] = pval.strip()
 
     # When the model gave structured params, hand them to the canonical
     # converter (same as native calls + <invoke>) so the full tool set and
@@ -800,6 +827,69 @@ def _strip_delimited(text: str, open_re, close_re) -> str:
     return "".join(out)
 
 
+def _iter_named_blocks(text, open_re, close_re):
+    """Forward-only equivalent of ``open_re([\\s\\S]*?)close_re`` finditer where
+    open_re captures a name in group 1: yield ``(name, body)``, pairing each
+    opener with the first ``close_re`` after it. O(n) once no closer is reachable
+    from an opener, no later opener has one either (see _iter_delimited), so
+    untrusted opener floods can't drive the lazy O(n^2) rescan."""
+    pos = 0
+    while True:
+        om = open_re.search(text, pos)
+        if om is None:
+            return
+        cm = close_re.search(text, om.end())
+        if cm is None:
+            return
+        yield om.group(1), text[om.end():cm.start()]
+        pos = cm.end()
+
+
+def _iter_xml_invoke(text):
+    """Forward-only ``<invoke name="..">...</invoke>`` scan (see _iter_named_blocks)."""
+    return _iter_named_blocks(text, _XML_INVOKE_OPEN_RE, _XML_INVOKE_CLOSE_RE)
+
+
+def _iter_backref_blocks(text, open_re, close_any_re, ci=False):
+    """Forward-only equivalent of an ``<tag>([\\s\\S]*?)</tag>`` backreference
+    finditer (same-name open/close): yield ``(name, body)``, pairing each opener
+    with the nearest following matching closer and skipping an opener whose
+    closer is unreachable.
+
+    Every closer is indexed by tag name in one linear pass, then each opener
+    binary-searches its own name's closer positions. A flood of distinct unclosed
+    tag names therefore stays O(n log n) rather than the lazy backref's O(n^2)
+    suffix rescan (CodeQL py/polynomial-redos); per-name memoization alone left
+    that distinct-name case quadratic. ``close_any_re`` matches ANY closer and
+    captures its tag name in group 1; ``ci`` lowercases names for matching, since
+    the original backref closer is case-insensitive under re.IGNORECASE."""
+    norm = (lambda s: s.lower()) if ci else (lambda s: s)
+    closer_starts = {}
+    closer_ends = {}
+    for cm in close_any_re.finditer(text):
+        k = norm(cm.group(1))
+        closer_starts.setdefault(k, []).append(cm.start())
+        closer_ends.setdefault(k, []).append(cm.end())
+    om = open_re.search(text)
+    while om is not None:
+        name = om.group(1)
+        k = norm(name)
+        resume = om.end()
+        starts = closer_starts.get(k)
+        if starts:
+            i = bisect.bisect_left(starts, om.end())
+            if i < len(starts):
+                yield name, text[om.end():starts[i]]
+                resume = closer_ends[k][i]
+        om = open_re.search(text, resume)
+
+
+def _iter_xml_direct(text):
+    """Forward-only equivalent of ``_XML_DIRECT_TOOL_RE.finditer`` (see
+    _iter_backref_blocks)."""
+    return _iter_backref_blocks(text, _XML_DIRECT_OPEN_RE, _XML_DIRECT_CLOSE_ANY_RE, ci=True)
+
+
 def parse_tool_blocks(text: str, skip_fenced: bool = False) -> List[ToolBlock]:
     """Extract executable tool blocks from LLM response text.
 
@@ -840,8 +930,8 @@ def parse_tool_blocks(text: str, skip_fenced: bool = False) -> List[ToolBlock]:
             # If a code block's content is an <invoke> XML call (some models wrap
             # tool calls in ```python or ```xml fences), parse the invoke instead.
             if '<invoke' in content:
-                for inv in _XML_INVOKE_RE.finditer(content):
+                for inv_name, inv_body in _iter_xml_invoke(content):
-                    block = _parse_xml_invoke(inv)
+                    block = _parse_xml_invoke(inv_name, inv_body)
                     if block:
                         blocks.append(block)
                 # This fenced block is <invoke> markup, not literal code. Whether or
@@ -882,13 +972,13 @@ def parse_tool_blocks(text: str, skip_fenced: bool = False) -> List[ToolBlock]:
             text, _XML_TOOL_CALL_OPEN_RE, _XML_TOOL_CALL_CLOSE_RE
         ):
             body = text[inner_start:inner_end]
-            for inv in _XML_INVOKE_RE.finditer(body):
+            for inv_name, inv_body in _iter_xml_invoke(body):
-                block = _parse_xml_invoke(inv)
+                block = _parse_xml_invoke(inv_name, inv_body)
                 if block:
                     blocks.append(block)
             if not blocks:
-                for direct in _XML_DIRECT_TOOL_RE.finditer(body):
+                for d_name, d_body in _iter_xml_direct(body):
-                    block = _parse_xml_direct_tool(direct)
+                    block = _parse_xml_direct_tool(d_name, d_body)
                     if block:
                         blocks.append(block)
         # Some local models stream an opening <tool_call> wrapper and a
@@ -896,20 +986,20 @@ def parse_tool_blocks(text: str, skip_fenced: bool = False) -> List[ToolBlock]:
         if not blocks:
             for m in _XML_OPEN_TOOL_CALL_RE.finditer(text):
                 body = m.group(1)
-                for inv in _XML_INVOKE_RE.finditer(body):
+                for inv_name, inv_body in _iter_xml_invoke(body):
-                    block = _parse_xml_invoke(inv)
+                    block = _parse_xml_invoke(inv_name, inv_body)
                     if block:
                         blocks.append(block)
                 if blocks:
                     break
-                for direct in _XML_DIRECT_TOOL_RE.finditer(body):
+                for d_name, d_body in _iter_xml_direct(body):
-                    block = _parse_xml_direct_tool(direct)
+                    block = _parse_xml_direct_tool(d_name, d_body)
                     if block:
                         blocks.append(block)
         # Try bare <invoke> without wrapper
         if not blocks:
-            for inv in _XML_INVOKE_RE.finditer(text):
+            for inv_name, inv_body in _iter_xml_invoke(text):
-                block = _parse_xml_invoke(inv)
+                block = _parse_xml_invoke(inv_name, inv_body)
                 if block:
                     blocks.append(block)
 

src/user_time.py

@@ -138,6 +138,69 @@ def current_datetime_prompt(now_utc: Optional[datetime] = None) -> str:
     )
 
 
+def current_datetime_context_message_for_tz(
+    iana_tz_name: Optional[str],
+    now_utc: Optional[datetime] = None,
+) -> Dict[str, str]:
+    """Build the current-date/time context as a user-role message, resolved
+    against an explicit IANA timezone name rather than browser ContextVars.
+
+    Unlike ``current_datetime_context_message()``, this function does not read
+    or write any ContextVar and leaves no per-request state behind — it is safe
+    to call from background tasks that have no browser request context.
+
+    Timezone resolution:
+    * ``iana_tz_name`` is a valid IANA name (e.g. ``"Europe/Berlin"``) → uses that zone.
+    * ``iana_tz_name`` is ``None`` OR resolves to an invalid zone → falls back to UTC.
+      This matches the existing scheduler behaviour: tasks without a linked crew
+      timezone render in UTC, not server-local time.
+    """
+    if now_utc is None:
+        utc_now = datetime.now(timezone.utc)
+    elif now_utc.tzinfo is None:
+        utc_now = now_utc.replace(tzinfo=timezone.utc)
+    else:
+        utc_now = now_utc.astimezone(timezone.utc)
+
+    # Resolve the display timezone — UTC fallback on any failure.
+    tz = timezone.utc
+    resolved_name: Optional[str] = None
+    if iana_tz_name:
+        try:
+            from zoneinfo import ZoneInfo
+            tz = ZoneInfo(iana_tz_name)
+            resolved_name = iana_tz_name
+        except Exception:
+            tz = timezone.utc  # invalid zone → UTC, no ContextVar touched
+
+    local_now = utc_now.astimezone(tz)
+    tomorrow = local_now + timedelta(days=1)
+
+    _utc_offset = local_now.utcoffset()
+    offset_min = int(_utc_offset.total_seconds() // 60) if _utc_offset is not None else 0
+    offset_label = f"UTC{format_utc_offset(offset_min)}"
+    tz_label = f"{resolved_name}, {offset_label}" if resolved_name else offset_label
+
+    prompt = (
+        "## Current date and time\n"
+        f"Today is {_date_label(local_now)} ({local_now.strftime('%Y-%m-%d')}). "
+        f"Local time is {_clock_label(local_now)} ({tz_label}); "
+        f"current UTC time is {utc_now.strftime('%H:%M')}.\n"
+        f"Tomorrow is {_date_label(tomorrow)} ({tomorrow.strftime('%Y-%m-%d')}) "
+        "in this timezone.\n"
+        "Use this for any 'today', 'tomorrow', 'tonight', 'this week', or other "
+        "relative-date reasoning. Do not ask for an exact date just because the "
+        "user used a relative date.\n\n"
+    )
+    return {
+        "role": "user",
+        "content": (
+            "[Context — current date/time, refreshed each turn; not part of "
+            "your instructions]\n" + prompt
+        ),
+    }
+
+
 def current_datetime_context_message(now_utc: Optional[datetime] = None) -> Dict[str, str]:
     """Build the current-date/time context as a standalone chat message.
 

tests/test_ai_interaction_owner_scope.py

@@ -25,9 +25,10 @@ def test_model_listing_and_image_fallback_are_owner_scoped():
 
     assert "owner: Optional[str] = None" in list_body
     assert "owner_filter(query, ModelEndpoint, owner)" in list_body
-    assert "_resolve_model(candidate, owner=owner)" in image_body
+    # _resolve_model is offloaded to a worker thread (#4589) but stays owner-scoped.
+    assert "asyncio.to_thread(_resolve_model, candidate, owner=owner)" in image_body
     assert "owner_filter(_img_q, ModelEndpoint, owner)" in image_body
-    assert "_resolve_model(model_spec, owner=owner)" in image_body
+    assert "asyncio.to_thread(_resolve_model, model_spec, owner=owner)" in image_body
 
 
 # chat_with_model, list_models and ask_teacher moved to the registry (#3629)

tests/test_app_initializer_memory_vector_degraded.py

@@ -0,0 +1,71 @@
+"""Regression: a present-but-unhealthy MemoryVectorStore must survive initialization.
+
+When MemoryVectorStore._initialize() fails (ChromaDB unavailable / embeddings not
+installed) it swallows the exception and leaves `.healthy == False` — the object
+exists but is unhealthy. app_initializer.initialize_managers() previously reset that
+object to ``None`` in the ``else`` branch, so service_health.chromadb_health() saw
+``memory_vector is None`` and reported the vector memory as DISABLED ("not
+configured") instead of DEGRADED/DOWN ("initialization failed") — losing the
+diagnostic distinction the /api/diagnostics/services probe is built to surface.
+
+This test fails before the fix (memory_vector is None) and passes after it.
+"""
+from unittest.mock import MagicMock
+
+import src.app_initializer as app_init
+import src.memory_vector as memory_vector_mod
+import src.service_health as sh
+
+
+class _UnhealthyVectorStore:
+    """Stand-in for a MemoryVectorStore whose init failed: present but inert."""
+    healthy = False
+
+    def count(self):
+        return 0
+
+    def search(self, *a, **k):
+        return []
+
+
+def _neutralize_collaborators(monkeypatch):
+    """Stub out everything initialize_managers() builds except the vector store,
+    so the test isolates the memory_vector health-handling branch."""
+    for name in [
+        "MemoryManager", "SkillsManager", "SessionManager", "UploadHandler",
+        "PersonalDocsManager", "APIKeyManager", "PresetManager",
+        "MemoryProviderRegistry", "NativeMemoryProvider", "ChatProcessor",
+        "ResearchHandler", "ChatHandler", "ModelDiscovery",
+    ]:
+        monkeypatch.setattr(app_init, name, lambda *a, **k: MagicMock())
+    monkeypatch.setattr(app_init, "set_session_manager", lambda *a, **k: None)
+    monkeypatch.setattr(app_init, "update_search_config", lambda *a, **k: None)
+    monkeypatch.setattr(app_init, "create_directories", lambda: None)
+
+
+def test_failed_memory_vector_init_is_kept_not_discarded(monkeypatch, tmp_path):
+    _neutralize_collaborators(monkeypatch)
+    # initialize_managers does `from src.memory_vector import MemoryVectorStore`
+    # at call time, so patch it on the source module.
+    monkeypatch.setattr(
+        memory_vector_mod, "MemoryVectorStore",
+        lambda *a, **k: _UnhealthyVectorStore(),
+    )
+
+    result = app_init.initialize_managers(str(tmp_path), rag_manager=None)
+
+    mv = result["memory_vector"]
+    assert mv is not None, "unhealthy MemoryVectorStore was discarded (reported as DISABLED, not DEGRADED/DOWN)"
+    assert mv.healthy is False
+
+
+def test_chromadb_health_reports_down_for_unhealthy_vector_store():
+    # Pins the downstream taxonomy the fix feeds: a present-but-unhealthy vector
+    # store (rag absent) is DOWN, not DISABLED; with a healthy rag it is DEGRADED;
+    # only when both are absent is it DISABLED.
+    store = _UnhealthyVectorStore()
+    healthy_rag = MagicMock(healthy=True)
+
+    assert sh.chromadb_health(None, None)["status"] == sh.DISABLED
+    assert sh.chromadb_health(None, store)["status"] == sh.DOWN
+    assert sh.chromadb_health(healthy_rag, store)["status"] == sh.DEGRADED

tests/test_builtin_mcp_bg_tasks.py

@@ -0,0 +1,108 @@
+"""Issue #4592 — built-in MCP startup must not leak tasks or subprocesses.
+
+Two defects in src/builtin_mcp.py:
+  * `register_builtin_servers` scheduled its python/npx connect coroutines with
+    a bare `asyncio.create_task(...)` whose return value was dropped. asyncio
+    keeps only a weak reference to such tasks, so the GC can collect one
+    mid-flight and the server silently never registers.
+  * `_is_npx_package_cached` killed its `npx --version` probe subprocess on
+    `TimeoutError` but not on `CancelledError`, so a cancellation (e.g. app
+    shutdown) orphaned the child.
+
+Both are exercised here with the module loaded in isolation (the same loader
+the existing npx-cache tests use), so no real servers or npx are involved.
+"""
+
+import asyncio
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _load_builtin_mcp(monkeypatch):
+    core = types.ModuleType("core")
+    core.__path__ = []
+    platform_compat = types.ModuleType("core.platform_compat")
+    platform_compat.IS_WINDOWS = False
+    platform_compat.which_tool = lambda name: None
+    monkeypatch.setitem(sys.modules, "core", core)
+    monkeypatch.setitem(sys.modules, "core.platform_compat", platform_compat)
+
+    spec = importlib.util.spec_from_file_location(
+        "builtin_mcp_under_test",
+        ROOT / "src" / "builtin_mcp.py",
+    )
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+async def test_spawn_bg_holds_strong_ref_until_task_finishes(monkeypatch):
+    builtin_mcp = _load_builtin_mcp(monkeypatch)
+
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def work():
+        started.set()
+        await release.wait()
+
+    task = builtin_mcp._spawn_bg(work())
+    await started.wait()
+    # While the task is in flight it must be reachable from the module-level
+    # set — that strong reference is what keeps the GC from collecting it.
+    assert task in builtin_mcp._BG_TASKS
+
+    release.set()
+    await task
+    await asyncio.sleep(0)  # let the done-callback run
+    # Once finished it is discarded so the set doesn't grow without bound.
+    assert task not in builtin_mcp._BG_TASKS
+
+
+async def test_npx_probe_reaps_subprocess_on_cancel(monkeypatch):
+    builtin_mcp = _load_builtin_mcp(monkeypatch)
+
+    # Force the code past the fast cache hit so it spawns the probe subprocess.
+    monkeypatch.setattr(builtin_mcp, "_is_package_in_npx_cache", lambda spec: False)
+
+    state = {"killed": False, "waited": False}
+    started = asyncio.Event()
+
+    class FakeProc:
+        returncode = None
+
+        async def communicate(self):
+            started.set()
+            await asyncio.sleep(3600)  # block until the probe is cancelled
+
+        def kill(self):
+            state["killed"] = True
+
+        async def wait(self):
+            state["waited"] = True
+
+    async def fake_create(*args, **kwargs):
+        return FakeProc()
+
+    monkeypatch.setattr(builtin_mcp.asyncio, "create_subprocess_exec", fake_create)
+
+    task = asyncio.create_task(
+        builtin_mcp._is_npx_package_cached("npx", "some-pkg@1.0.0", timeout_s=3600)
+    )
+    await started.wait()
+    task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    # The child was killed and reaped rather than orphaned.
+    assert state["killed"] is True
+    assert state["waited"] is True

tests/test_cookbook_dependency_completion_regression.py

@@ -106,4 +106,9 @@ def test_local_dependency_probe_refreshes_user_site_visibility():
 
     assert "importlib.invalidate_caches()" in source
     assert "user_site = site.getusersitepackages()" in source
-    assert "if user_site and os.path.isdir(user_site) and user_site not in sys.path:" in source
+    # addsitedir (not a bare sys.path.append) so user-site `.pth` hooks are
+    # replayed when a package is installed into an already-running process —
+    # otherwise setuptools' distutils shim never activates and basicsr-based
+    # deps (realesrgan) probe as not-installed until a restart. See #4810.
+    assert "if user_site and os.path.isdir(user_site):" in source
+    assert "site.addsitedir(user_site)" in source

tests/test_email_send_only_no_inbox.py

@@ -0,0 +1,75 @@
+"""A send-only (SMTP-only) account has no inbox to read.
+
+`_imap_connect` must fail fast with a clear, typed error instead of handing an
+empty host to imaplib — `imaplib.IMAP4("", 993)` silently dials localhost:993
+and surfaces a confusing "[Errno 111] Connection refused" on every inbox poll.
+"""
+import os
+import tempfile
+from pathlib import Path
+
+import pytest
+
+_tmp_data = Path(tempfile.mkdtemp(prefix="odysseus-email-send-only-test-"))
+os.environ.setdefault("DATA_DIR", str(_tmp_data))
+os.environ.setdefault("DATABASE_URL", f"sqlite:///{_tmp_data / 'app.db'}")
+
+import routes.email_helpers as helpers
+from routes.email_helpers import EmailNotConfiguredError, _imap_connect
+
+
+_SEND_ONLY_CFG = {
+    "account_id": "acct-send-only",
+    "account_name": "send-only",
+    "smtp_host": "smtp.example.org",
+    "smtp_port": 465,
+    "smtp_user": "noreply@example.org",
+    "smtp_password": "secret",
+    "imap_host": "",          # <- the send-only marker
+    "imap_port": 993,
+    "imap_user": "",
+    "imap_password": "",
+    "imap_starttls": True,
+    "from_address": "noreply@example.org",
+}
+
+
+def test_not_configured_error_is_runtime_error():
+    # Subclassing RuntimeError keeps existing broad `except Exception` handlers
+    # working while letting the inbox poll catch this case specifically.
+    assert issubclass(EmailNotConfiguredError, RuntimeError)
+
+
+def test_imap_connect_send_only_raises_and_never_dials(monkeypatch):
+    monkeypatch.setattr(helpers, "_get_email_config", lambda *a, **k: dict(_SEND_ONLY_CFG))
+
+    def _boom(*a, **k):  # opening a connection means we dialed an empty host
+        raise AssertionError("send-only account must not open an IMAP connection")
+
+    monkeypatch.setattr(helpers, "_open_imap_connection", _boom)
+
+    with pytest.raises(EmailNotConfiguredError):
+        _imap_connect("acct-send-only")
+
+
+def test_imap_connect_with_host_still_connects(monkeypatch):
+    # Guard must not regress normal accounts: a configured imap_host still
+    # reaches _open_imap_connection.
+    cfg = dict(_SEND_ONLY_CFG, imap_host="imap.example.org", imap_user="u", imap_password="p")
+    monkeypatch.setattr(helpers, "_get_email_config", lambda *a, **k: cfg)
+
+    opened = {}
+
+    class _FakeConn:
+        def login(self, user, password):
+            opened["login"] = (user, password)
+
+    def _fake_open(host, port, *, starttls, timeout):
+        opened["host"] = host
+        return _FakeConn()
+
+    monkeypatch.setattr(helpers, "_open_imap_connection", _fake_open)
+
+    conn = _imap_connect("acct-with-imap")
+    assert opened["host"] == "imap.example.org"
+    assert isinstance(conn, _FakeConn)

tests/test_endpoint_resolver_headers.py

@@ -0,0 +1,16 @@
+"""Tests for endpoint_resolver — request header construction."""
+from src.endpoint_resolver import build_headers
+
+
+class TestBuildHeaders:
+    def test_no_key(self):
+        assert build_headers(None, "https://api.openai.com/v1") == {}
+
+    def test_openai_bearer(self):
+        assert build_headers("sk-abc", "https://api.openai.com/v1") == {"Authorization": "Bearer sk-abc"}
+
+    def test_anthropic_headers(self):
+        assert build_headers("sk-ant-abc", "https://api.anthropic.com") == {"x-api-key": "sk-ant-abc", "anthropic-version": "2023-06-01"}
+
+    def test_empty_key(self):
+        assert build_headers("", "https://api.openai.com/v1") == {}

tests/test_endpoint_resolver_models.py

@@ -0,0 +1,67 @@
+"""Tests for endpoint_resolver — endpoint/model selection and enabled-model filtering."""
+import json
+
+from src.endpoint_resolver import (
+    _first_chat_model,
+    _endpoint_hidden_models,
+    _endpoint_enabled_models,
+)
+
+
+class _Ep:
+    """Minimal ModelEndpoint stand-in for the model-picking helpers."""
+    def __init__(self, cached=None, hidden=None):
+        self.cached_models = json.dumps(cached) if cached is not None else None
+        self.hidden_models = json.dumps(hidden) if hidden is not None else None
+
+
+class TestFirstChatModel:
+    def test_skips_embedding_and_tts(self):
+        models = ["text-embedding-ada-002", "whisper-large-v3", "gpt-4o"]
+        assert _first_chat_model(models) == "gpt-4o"
+
+    def test_falls_back_to_first_when_all_non_chat(self):
+        assert _first_chat_model(["whisper-large-v3"]) == "whisper-large-v3"
+
+    def test_empty(self):
+        assert _first_chat_model([]) is None
+
+
+class TestEnabledModels:
+    def test_excludes_hidden(self):
+        # The Groq repro: 16 models, only gpt-oss-120b enabled.
+        cached = [
+            "openai/gpt-oss-safeguard-20b", "canopylabs/orpheus-arabic-saudi",
+            "whisper-large-v3", "openai/gpt-oss-120b",
+        ]
+        hidden = [
+            "openai/gpt-oss-safeguard-20b", "canopylabs/orpheus-arabic-saudi",
+            "whisper-large-v3",
+        ]
+        ep = _Ep(cached=cached, hidden=hidden)
+        assert _endpoint_enabled_models(ep) == ["openai/gpt-oss-120b"]
+
+    def test_no_hidden_returns_all(self):
+        ep = _Ep(cached=["a", "b"], hidden=None)
+        assert _endpoint_enabled_models(ep) == ["a", "b"]
+
+    def test_picker_never_selects_disabled_model(self):
+        # Regression: a disabled model listed first must not be auto-picked.
+        cached = ["canopylabs/orpheus-arabic-saudi", "openai/gpt-oss-120b"]
+        hidden = ["canopylabs/orpheus-arabic-saudi"]
+        ep = _Ep(cached=cached, hidden=hidden)
+        assert _first_chat_model(_endpoint_enabled_models(ep)) == "openai/gpt-oss-120b"
+
+    def test_stale_configured_model_is_discarded(self):
+        # A configured model that's been disabled is dropped, falling through
+        # to the first enabled chat model.
+        ep = _Ep(
+            cached=["canopylabs/orpheus-arabic-saudi", "openai/gpt-oss-120b"],
+            hidden=["canopylabs/orpheus-arabic-saudi"],
+        )
+        configured = "canopylabs/orpheus-arabic-saudi"
+        if configured in _endpoint_hidden_models(ep):
+            configured = ""
+        if not configured:
+            configured = _first_chat_model(_endpoint_enabled_models(ep))
+        assert configured == "openai/gpt-oss-120b"

tests/test_endpoint_resolver_urls.py

@@ -1,16 +1,10 @@
-"""Tests for endpoint_resolver — pure functions tested directly."""
+"""Tests for endpoint_resolver — URL normalization and URL construction."""
-import json
-
 import pytest
 
 from src.endpoint_resolver import (
-    _first_chat_model,
-    _endpoint_hidden_models,
-    _endpoint_enabled_models,
     normalize_base,
     build_chat_url,
     build_models_url,
-    build_headers,
 )
 
 
@@ -99,76 +93,3 @@ class TestBuildModelsUrl:
     def test_rejects_query_or_fragment_base(self, bad_base):
         with pytest.raises(ValueError, match="query or fragment"):
             build_models_url(bad_base)
-
-
-class TestBuildHeaders:
-    def test_no_key(self):
-        assert build_headers(None, "https://api.openai.com/v1") == {}
-
-    def test_openai_bearer(self):
-        assert build_headers("sk-abc", "https://api.openai.com/v1") == {"Authorization": "Bearer sk-abc"}
-
-    def test_anthropic_headers(self):
-        assert build_headers("sk-ant-abc", "https://api.anthropic.com") == {"x-api-key": "sk-ant-abc", "anthropic-version": "2023-06-01"}
-
-    def test_empty_key(self):
-        assert build_headers("", "https://api.openai.com/v1") == {}
-
-
-class _Ep:
-    """Minimal ModelEndpoint stand-in for the model-picking helpers."""
-    def __init__(self, cached=None, hidden=None):
-        self.cached_models = json.dumps(cached) if cached is not None else None
-        self.hidden_models = json.dumps(hidden) if hidden is not None else None
-
-
-class TestFirstChatModel:
-    def test_skips_embedding_and_tts(self):
-        models = ["text-embedding-ada-002", "whisper-large-v3", "gpt-4o"]
-        assert _first_chat_model(models) == "gpt-4o"
-
-    def test_falls_back_to_first_when_all_non_chat(self):
-        assert _first_chat_model(["whisper-large-v3"]) == "whisper-large-v3"
-
-    def test_empty(self):
-        assert _first_chat_model([]) is None
-
-
-class TestEnabledModels:
-    def test_excludes_hidden(self):
-        # The Groq repro: 16 models, only gpt-oss-120b enabled.
-        cached = [
-            "openai/gpt-oss-safeguard-20b", "canopylabs/orpheus-arabic-saudi",
-            "whisper-large-v3", "openai/gpt-oss-120b",
-        ]
-        hidden = [
-            "openai/gpt-oss-safeguard-20b", "canopylabs/orpheus-arabic-saudi",
-            "whisper-large-v3",
-        ]
-        ep = _Ep(cached=cached, hidden=hidden)
-        assert _endpoint_enabled_models(ep) == ["openai/gpt-oss-120b"]
-
-    def test_no_hidden_returns_all(self):
-        ep = _Ep(cached=["a", "b"], hidden=None)
-        assert _endpoint_enabled_models(ep) == ["a", "b"]
-
-    def test_picker_never_selects_disabled_model(self):
-        # Regression: a disabled model listed first must not be auto-picked.
-        cached = ["canopylabs/orpheus-arabic-saudi", "openai/gpt-oss-120b"]
-        hidden = ["canopylabs/orpheus-arabic-saudi"]
-        ep = _Ep(cached=cached, hidden=hidden)
-        assert _first_chat_model(_endpoint_enabled_models(ep)) == "openai/gpt-oss-120b"
-
-    def test_stale_configured_model_is_discarded(self):
-        # A configured model that's been disabled is dropped, falling through
-        # to the first enabled chat model.
-        ep = _Ep(
-            cached=["canopylabs/orpheus-arabic-saudi", "openai/gpt-oss-120b"],
-            hidden=["canopylabs/orpheus-arabic-saudi"],
-        )
-        configured = "canopylabs/orpheus-arabic-saudi"
-        if configured in _endpoint_hidden_models(ep):
-            configured = ""
-        if not configured:
-            configured = _first_chat_model(_endpoint_enabled_models(ep))
-        assert configured == "openai/gpt-oss-120b"

tests/test_fenced_example_not_executed_for_native_models.py

@@ -178,14 +178,14 @@ def test_issue_3222_repro_guide_only_response_resolves_no_tool_actions(monkeypat
 # ---------------------------------------------------------------------------
 def test_resolve_tool_blocks_skips_textual_fallback_for_native_models_with_no_native_calls():
     guide_only = "```bash\nnpm run plan:articles\n```\n```json\n{\"a\": 1}\n```"
-    blocks, used_native = al._resolve_tool_blocks(guide_only, [], round_num=1, is_api_model=True)
+    blocks, used_native, _ = al._resolve_tool_blocks(guide_only, [], round_num=1, is_api_model=True)
     assert blocks == []
     assert used_native is False
 
 
 def test_resolve_tool_blocks_keeps_textual_fallback_for_non_native_models():
     text = "```bash\necho hi\n```"
-    blocks, used_native = al._resolve_tool_blocks(text, [], round_num=1, is_api_model=False)
+    blocks, used_native, _ = al._resolve_tool_blocks(text, [], round_num=1, is_api_model=False)
     assert len(blocks) == 1
     assert blocks[0].tool_type == "bash"
     assert used_native is False
@@ -193,7 +193,7 @@ def test_resolve_tool_blocks_keeps_textual_fallback_for_non_native_models():
 
 def test_resolve_tool_blocks_native_path_untouched_when_native_calls_present():
     native_calls = [{"name": "bash", "arguments": json.dumps({"command": "echo hi"})}]
-    blocks, used_native = al._resolve_tool_blocks("some prose", native_calls, round_num=1, is_api_model=True)
+    blocks, used_native, _ = al._resolve_tool_blocks("some prose", native_calls, round_num=1, is_api_model=True)
     assert used_native is True
     assert len(blocks) == 1
     assert blocks[0].tool_type == "bash"
@@ -305,7 +305,7 @@ def test_resolve_tool_blocks_recovers_invoke_markup_for_native_model_with_no_nat
         "I'll search for that now.\n"
         '<invoke name="web_search"><parameter name="query">odysseus changelog</parameter></invoke>'
     )
-    blocks, used_native = al._resolve_tool_blocks(leaked, [], round_num=1, is_api_model=True)
+    blocks, used_native, _ = al._resolve_tool_blocks(leaked, [], round_num=1, is_api_model=True)
     assert used_native is False
     assert len(blocks) == 1
     assert blocks[0].tool_type == "web_search"

tests/test_llm_core_temperature_anthropic.py

@@ -0,0 +1,34 @@
+"""Regression tests: Anthropic temperature clamping.
+
+Anthropic rejects temperature values outside [0.0, 1.0]. The payload builder
+must clamp the value to that range before sending rather than letting the API
+return HTTP 400.
+"""
+from src import llm_core
+
+
+def _anthropic_payload(temperature):
+    return llm_core._build_anthropic_payload(
+        "claude-3-5-sonnet",
+        [{"role": "user", "content": "Hi"}],
+        temperature,
+        max_tokens=5,
+    )
+
+
+def test_anthropic_payload_clamps_above_one():
+    # Anthropic rejects temperature > 1.0 (e.g. the Nietzsche preset's 1.2).
+    assert _anthropic_payload(1.2)["temperature"] == 1.0
+
+
+def test_anthropic_payload_keeps_in_range():
+    assert _anthropic_payload(0.7)["temperature"] == 0.7
+
+
+def test_anthropic_payload_clamps_negative():
+    assert _anthropic_payload(-0.5)["temperature"] == 0.0
+
+
+def test_anthropic_payload_none_temperature_does_not_crash():
+    payload = _anthropic_payload(None)
+    assert payload["temperature"] is None

tests/test_llm_core_temperature_moonshot.py

@@ -0,0 +1,100 @@
+"""Regression tests: Moonshot/Kimi temperature detection and payload behavior.
+
+Moonshot kimi-k2.5+ models reject custom temperature values; the payload
+builder must detect the Moonshot provider and omit temperature for the affected
+model family. Self-hosted Kimi deployments (non-Moonshot URL) must keep the
+caller-specified temperature unchanged.
+"""
+import httpx
+import pytest
+
+from src import llm_core
+
+
+@pytest.mark.parametrize(
+    "model",
+    [
+        "kimi-k2.5",
+        "kimi-k2.6",
+        "moonshot/kimi-k2.6",
+        "kimi-k2.6-preview",
+    ],
+)
+def test_moonshot_k2_5_plus_uses_fixed_temperature(model):
+    assert llm_core._moonshot_rejects_custom_temperature("moonshot", model)
+
+
+@pytest.mark.parametrize(
+    "provider,model",
+    [
+        ("openai", "kimi-k2.6"),
+        ("moonshot", "kimi-k2-0905-preview"),
+        ("moonshot", "kimi-k2-thinking"),
+        ("moonshot", "kimi-k2.50"),
+        ("moonshot", None),
+    ],
+)
+def test_other_models_keep_temperature(provider, model):
+    assert not llm_core._moonshot_rejects_custom_temperature(provider, model)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://api.moonshot.ai/v1/chat/completions",
+        "https://api.moonshot.cn/v1/chat/completions",
+    ],
+)
+def test_moonshot_provider_detection(url):
+    assert llm_core._detect_provider(url) == "moonshot"
+
+
+def _capture_openai_payload(
+    monkeypatch,
+    model,
+    temperature,
+    url="https://api.openai.com/v1/chat/completions",
+):
+    """Run a synchronous OpenAI-compatible call and return the posted JSON body."""
+    llm_core._response_cache.clear()
+    seen = {}
+
+    def fake_post(url, headers=None, json=None, timeout=None):
+        seen["json"] = json
+        request = httpx.Request("POST", url)
+        return httpx.Response(
+            200,
+            request=request,
+            json={"choices": [{"message": {"content": "OK"}}]},
+        )
+
+    monkeypatch.setattr(llm_core.httpx, "post", fake_post)
+    result = llm_core.llm_call(
+        url,
+        model,
+        [{"role": "user", "content": "Say OK"}],
+        temperature=temperature,
+        max_tokens=5,
+    )
+    assert result == "OK"
+    return seen["json"]
+
+
+def test_moonshot_k2_6_payload_omits_temperature(monkeypatch):
+    payload = _capture_openai_payload(
+        monkeypatch,
+        "kimi-k2.6",
+        0.7,
+        url="https://api.moonshot.ai/v1/chat/completions",
+    )
+    assert "temperature" not in payload
+
+
+def test_self_hosted_kimi_k2_6_payload_keeps_temperature(monkeypatch):
+    payload = _capture_openai_payload(
+        monkeypatch,
+        "kimi-k2.6",
+        0.7,
+        url="http://localhost:8000/v1/chat/completions",
+    )
+    assert payload["temperature"] == 0.7

tests/test_llm_core_temperature_reasoning.py

@@ -109,88 +109,3 @@ def test_chatgpt_subscription_payload_omits_max_output_tokens_when_zero():
     )
 
     assert "max_output_tokens" not in payload
-
-
-def _anthropic_payload(temperature):
-    return llm_core._build_anthropic_payload(
-        "claude-3-5-sonnet",
-        [{"role": "user", "content": "Hi"}],
-        temperature,
-        max_tokens=5,
-    )
-
-
-def test_anthropic_payload_clamps_above_one():
-    # Anthropic rejects temperature > 1.0 (e.g. the Nietzsche preset's 1.2).
-    assert _anthropic_payload(1.2)["temperature"] == 1.0
-
-
-def test_anthropic_payload_keeps_in_range():
-    assert _anthropic_payload(0.7)["temperature"] == 0.7
-
-
-def test_anthropic_payload_clamps_negative():
-    assert _anthropic_payload(-0.5)["temperature"] == 0.0
-
-
-def test_anthropic_payload_none_temperature_does_not_crash():
-    payload = _anthropic_payload(None)
-    assert payload["temperature"] is None
-
-
-@pytest.mark.parametrize(
-    "model",
-    [
-        "kimi-k2.5",
-        "kimi-k2.6",
-        "moonshot/kimi-k2.6",
-        "kimi-k2.6-preview",
-    ],
-)
-def test_moonshot_k2_5_plus_uses_fixed_temperature(model):
-    assert llm_core._moonshot_rejects_custom_temperature("moonshot", model)
-
-
-@pytest.mark.parametrize(
-    "provider,model",
-    [
-        ("openai", "kimi-k2.6"),
-        ("moonshot", "kimi-k2-0905-preview"),
-        ("moonshot", "kimi-k2-thinking"),
-        ("moonshot", "kimi-k2.50"),
-        ("moonshot", None),
-    ],
-)
-def test_other_models_keep_temperature(provider, model):
-    assert not llm_core._moonshot_rejects_custom_temperature(provider, model)
-
-
-@pytest.mark.parametrize(
-    "url",
-    [
-        "https://api.moonshot.ai/v1/chat/completions",
-        "https://api.moonshot.cn/v1/chat/completions",
-    ],
-)
-def test_moonshot_provider_detection(url):
-    assert llm_core._detect_provider(url) == "moonshot"
-
-
-def test_moonshot_k2_6_payload_omits_temperature(monkeypatch):
-    payload = _capture_openai_payload(
-        monkeypatch,
-        "kimi-k2.6",
-        0.7,
-        url="https://api.moonshot.ai/v1/chat/completions",
-    )
-    assert "temperature" not in payload
-
-
-def test_self_hosted_kimi_k2_6_payload_keeps_temperature(monkeypatch):
-    payload = _capture_openai_payload(
-        monkeypatch,
-        "kimi-k2.6",
-        0.7,
-        url="http://localhost:8000/v1/chat/completions",
-    )
-    assert payload["temperature"] == 0.7

tests/test_native_tool_result_threading.py

@@ -0,0 +1,39 @@
+"""Native tool-call results must be threaded by CONVERTED-call position.
+
+When an OpenAI/Anthropic model emits several tool_calls in one round and one
+fails to convert (hallucinated name or bad-JSON args), it is dropped from
+tool_blocks (so it produces no result) but used to stay in native_tool_calls.
+_append_tool_results indexed tool_result_texts by native-call position, so the
+surviving result was attached to the wrong tool_call_id and the real call was
+answered with an empty string. _resolve_tool_blocks now returns the converted
+calls aligned 1:1 with tool_blocks/tool_result_texts, and that aligned list is
+what is threaded back.
+"""
+import src.agent_loop as al
+
+
+def test_resolve_returns_converted_calls_aligned():
+    native = [
+        {"name": "bogus_unknown_tool", "arguments": "{}", "id": "A"},
+        {"name": "web_search", "arguments": '{"query": "hello"}', "id": "B"},
+    ]
+    tool_blocks, used_native, converted = al._resolve_tool_blocks("", native, 1)
+    assert used_native is True
+    assert len(tool_blocks) == 1           # only web_search converted
+    assert [c["name"] for c in converted] == ["web_search"]
+    assert len(converted) == len(tool_blocks)  # aligned 1:1
+
+
+def test_append_threads_result_to_correct_tool_call_id():
+    messages = []
+    converted = [{"id": "B", "name": "web_search", "arguments": "{}"}]
+    al._append_tool_results(
+        messages, "some response", converted,
+        ["RESULT"], ["RESULT"], True, 1,
+    )
+    tool_msgs = [m for m in messages if m.get("role") == "tool"]
+    assert len(tool_msgs) == 1
+    assert tool_msgs[0]["tool_call_id"] == "B"
+    assert tool_msgs[0]["content"] == "RESULT"
+    asst = next(m for m in messages if m.get("role") == "assistant")
+    assert [tc["id"] for tc in asst["tool_calls"]] == ["B"]

tests/test_provider_detection_builders.py

@@ -1,4 +1,4 @@
-"""Provider detection tests (re: #768).
+"""Provider detection tests — build_chat_url / build_models_url routing (re: #768).
 
 These import the *real* helpers from ``src.llm_core`` (not local copies) so a
 regression in hostname matching is actually caught. The point of the change
@@ -13,72 +13,6 @@ from src import endpoint_resolver
 from src.endpoint_resolver import build_chat_url, build_models_url
 
 
-class TestHostMatch:
-    def test_exact_host(self):
-        assert llm_core._host_match("https://anthropic.com/v1", "anthropic.com")
-
-    def test_subdomain(self):
-        assert llm_core._host_match("https://api.anthropic.com/v1", "anthropic.com")
-
-    def test_multiple_domains(self):
-        assert llm_core._host_match("https://api.together.ai/v1", "together.xyz", "together.ai")
-
-    def test_trailing_dot_fqdn(self):
-        # A fully-qualified host with a trailing dot is legal and resolvable.
-        assert llm_core._host_match("https://api.anthropic.com./v1", "anthropic.com")
-
-    def test_domain_in_path_does_not_match(self):
-        assert not llm_core._host_match("https://myproxy.internal/anthropic.com/v1", "anthropic.com")
-
-    def test_domain_in_query_does_not_match(self):
-        assert not llm_core._host_match("https://example.com/v1?ref=anthropic.com", "anthropic.com")
-
-    def test_lookalike_host_does_not_match(self):
-        assert not llm_core._host_match("https://anthropic.com.example/v1", "anthropic.com")
-
-    def test_none_and_empty_safe(self):
-        assert not llm_core._host_match(None, "anthropic.com")
-        assert not llm_core._host_match("", "anthropic.com")
-
-
-class TestDetectProviderRealHosts:
-    def test_chatgpt_subscription_codex_backend(self):
-        assert llm_core._detect_provider("https://chatgpt.com/backend-api/codex") == "chatgpt-subscription"
-        assert llm_core._detect_provider("https://chatgpt.com/backend-api/codex/responses") == "chatgpt-subscription"
-
-    def test_anthropic(self):
-        assert llm_core._detect_provider("https://api.anthropic.com") == "anthropic"
-
-    def test_openrouter(self):
-        assert llm_core._detect_provider("https://openrouter.ai/api/v1") == "openrouter"
-
-    def test_groq_openai_compat_path(self):
-        # Groq's base carries an /openai/v1 path; detection must still see the host.
-        assert llm_core._detect_provider("https://api.groq.com/openai/v1") == "groq"
-
-    def test_ollama_native_unchanged(self):
-        assert llm_core._detect_provider("https://ollama.com/api") == "ollama"
-
-    def test_unknown_host_defaults_to_openai(self):
-        assert llm_core._detect_provider("https://api.example.com/v1") == "openai"
-
-
-class TestDetectProviderRejectsSubstringFalsePositives:
-    """The regression that motivated #768: substring matching mislabeled these."""
-
-    def test_provider_domain_in_path(self):
-        assert llm_core._detect_provider("https://myproxy.internal/anthropic.com/v1") == "openai"
-
-    def test_provider_domain_in_query(self):
-        assert llm_core._detect_provider("https://example.com/v1?ref=anthropic.com") == "openai"
-
-    def test_lookalike_host(self):
-        assert llm_core._detect_provider("https://anthropic.com.example/v1") == "openai"
-
-    def test_none_safe(self):
-        assert llm_core._detect_provider(None) == "openai"
-
-
 class TestBuildersRejectLookalikeHosts:
     """build_chat_url / build_models_url must route look-alike and
     domain-in-path hosts to the OpenAI-compatible default, not the

tests/test_provider_detection_detect.py

@@ -0,0 +1,47 @@
+"""Provider detection tests — _detect_provider real hosts and false-positive rejection (re: #768).
+
+These import the *real* helpers from ``src.llm_core`` (not local copies) so a
+regression in hostname matching is actually caught. The point of the change
+under test is that provider detection keys off the URL's *hostname*, not a
+substring of the whole URL — so a domain appearing in a path/query, or a
+look-alike host, must not be misclassified.
+"""
+from src import llm_core
+
+
+class TestDetectProviderRealHosts:
+    def test_chatgpt_subscription_codex_backend(self):
+        assert llm_core._detect_provider("https://chatgpt.com/backend-api/codex") == "chatgpt-subscription"
+        assert llm_core._detect_provider("https://chatgpt.com/backend-api/codex/responses") == "chatgpt-subscription"
+
+    def test_anthropic(self):
+        assert llm_core._detect_provider("https://api.anthropic.com") == "anthropic"
+
+    def test_openrouter(self):
+        assert llm_core._detect_provider("https://openrouter.ai/api/v1") == "openrouter"
+
+    def test_groq_openai_compat_path(self):
+        # Groq's base carries an /openai/v1 path; detection must still see the host.
+        assert llm_core._detect_provider("https://api.groq.com/openai/v1") == "groq"
+
+    def test_ollama_native_unchanged(self):
+        assert llm_core._detect_provider("https://ollama.com/api") == "ollama"
+
+    def test_unknown_host_defaults_to_openai(self):
+        assert llm_core._detect_provider("https://api.example.com/v1") == "openai"
+
+
+class TestDetectProviderRejectsSubstringFalsePositives:
+    """The regression that motivated #768: substring matching mislabeled these."""
+
+    def test_provider_domain_in_path(self):
+        assert llm_core._detect_provider("https://myproxy.internal/anthropic.com/v1") == "openai"
+
+    def test_provider_domain_in_query(self):
+        assert llm_core._detect_provider("https://example.com/v1?ref=anthropic.com") == "openai"
+
+    def test_lookalike_host(self):
+        assert llm_core._detect_provider("https://anthropic.com.example/v1") == "openai"
+
+    def test_none_safe(self):
+        assert llm_core._detect_provider(None) == "openai"

tests/test_provider_detection_host_match.py

@@ -0,0 +1,37 @@
+"""Provider detection tests — hostname matching helpers (re: #768).
+
+These import the *real* helpers from ``src.llm_core`` (not local copies) so a
+regression in hostname matching is actually caught. The point of the change
+under test is that provider detection keys off the URL's *hostname*, not a
+substring of the whole URL — so a domain appearing in a path/query, or a
+look-alike host, must not be misclassified.
+"""
+from src import llm_core
+
+
+class TestHostMatch:
+    def test_exact_host(self):
+        assert llm_core._host_match("https://anthropic.com/v1", "anthropic.com")
+
+    def test_subdomain(self):
+        assert llm_core._host_match("https://api.anthropic.com/v1", "anthropic.com")
+
+    def test_multiple_domains(self):
+        assert llm_core._host_match("https://api.together.ai/v1", "together.xyz", "together.ai")
+
+    def test_trailing_dot_fqdn(self):
+        # A fully-qualified host with a trailing dot is legal and resolvable.
+        assert llm_core._host_match("https://api.anthropic.com./v1", "anthropic.com")
+
+    def test_domain_in_path_does_not_match(self):
+        assert not llm_core._host_match("https://myproxy.internal/anthropic.com/v1", "anthropic.com")
+
+    def test_domain_in_query_does_not_match(self):
+        assert not llm_core._host_match("https://example.com/v1?ref=anthropic.com", "anthropic.com")
+
+    def test_lookalike_host_does_not_match(self):
+        assert not llm_core._host_match("https://anthropic.com.example/v1", "anthropic.com")
+
+    def test_none_and_empty_safe(self):
+        assert not llm_core._host_match(None, "anthropic.com")
+        assert not llm_core._host_match("", "anthropic.com")

tests/test_redos_xml_tool_parsers.py

@@ -0,0 +1,197 @@
+"""Regression tests for the remaining ReDoS sinks in tool_parsing.py.
+
+A previous fix (test_redos_llm_parsers.py) hardened the delimiter-bounded
+[TOOL_CALL]/<tool_call>/<tool_code> scanners but explicitly left four patterns
+that CodeQL (py/polynomial-redos) flagged on the next rescan:
+
+  * `args => { ... }` in `_parse_tool_call_block` — greedy `\\{([\\s\\S]*)\\}`
+    that `re.search` restarts from every `args:{` opener -> O(n^2).
+  * `_XML_INVOKE_RE` — lazy `<invoke ...>([\\s\\S]*?)</invoke>` that rescans to
+    end-of-string from every opener when no `</invoke>` follows.
+  * `_XML_DIRECT_TOOL_RE` and the `<tag>([\\s\\S]*?)</\\1>` param scan in
+    `_parse_tool_code_block` — lazy *backreference* patterns with the same
+    opener-flood blowup.
+
+These run over untrusted model output (tool-call markup is attacker-influenced
+via prompt injection), so each is now a forward-only scan. The tests pin:
+  * correctness is unchanged for legitimate tool-call markup, and
+  * pathological "many openers, no closer" inputs complete promptly.
+
+The timing bound is loose (seconds) so it never flakes on a slow CI box; the
+unguarded patterns took 2-15s on these inputs, so the margin is ~100x.
+"""
+
+import time
+
+import pytest
+
+import src.agent_tools  # noqa: F401  (break agent_tools<->tool_parsing import cycle)
+from src.tool_parsing import (
+    parse_tool_blocks,
+    strip_tool_blocks,
+    _parse_tool_call_block,
+    _parse_tool_code_block,
+)
+
+_BUDGET_S = 4.0
+
+
+def _timed(fn, *args):
+    start = time.perf_counter()
+    result = fn(*args)
+    return result, time.perf_counter() - start
+
+
+# ── correctness is preserved ────────────────────────────────────────────────
+
+def test_xml_invoke_call_still_parsed():
+    blocks = parse_tool_blocks(
+        '<tool_call><invoke name="bash"><parameter name="command">ls -la</parameter></invoke></tool_call>'
+    )
+    assert [(b.tool_type, b.content) for b in blocks] == [("bash", "ls -la")]
+
+
+def test_xml_direct_tool_still_parsed():
+    blocks = parse_tool_blocks('<tool_call><web_search>weather today</web_search></tool_call>')
+    assert [(b.tool_type, b.content) for b in blocks] == [("web_search", "weather today")]
+
+
+def test_xml_direct_tool_backref_is_case_insensitive():
+    # `</\\1>` matched case-insensitively under re.IGNORECASE; the forward-only
+    # scanner preserves that (mixed-case closer still pairs with its opener).
+    blocks = parse_tool_blocks('<tool_call><Web_Search>q</WEB_SEARCH></tool_call>')
+    assert [(b.tool_type, b.content) for b in blocks] == [("web_search", "q")]
+
+
+def test_tool_code_xml_params_still_parsed():
+    blocks = parse_tool_blocks("<tool_code>{tool => 'bash', args => '<command>ls -la</command>'}</tool_code>")
+    assert [(b.tool_type, b.content) for b in blocks] == [("bash", "ls -la")]
+
+
+def test_xml_invoke_multiple_parameters_still_parsed():
+    # The invoke parameter scan is forward-only; a well-formed invoke with more
+    # than one <parameter> must still yield every name/value pair.
+    blocks = parse_tool_blocks(
+        '<tool_call><invoke name="web_search">'
+        '<parameter name="query">rust traits</parameter>'
+        '<parameter name="time_filter">week</parameter>'
+        '</invoke></tool_call>'
+    )
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert '"query": "rust traits"' in blocks[0].content
+    assert '"time_filter": "week"' in blocks[0].content
+
+
+def test_xml_direct_distinct_tag_names_still_parsed():
+    # Distinct sibling tags inside <tool_call> each pair with their own closer;
+    # the forward-only direct scan must keep matching after the first block.
+    blocks = parse_tool_blocks(
+        '<tool_call><web_search>weather</web_search><read_file>notes.txt</read_file></tool_call>'
+    )
+    assert [(b.tool_type, b.content) for b in blocks] == [
+        ("web_search", "weather"),
+        ("read_file", "notes.txt"),
+    ]
+
+
+def test_tool_call_args_brace_still_parsed():
+    blocks = parse_tool_blocks('[TOOL_CALL]{tool => "shell", args => {--command "ls"}}[/TOOL_CALL]')
+    assert [(b.tool_type, b.content) for b in blocks] == [("bash", "ls")]
+
+
+def test_args_brace_takes_through_last_close_brace():
+    # `\\{([\\s\\S]*)\\}` is greedy to the LAST `}`; the rfind-based rewrite must
+    # match that (keep the nested object intact, not stop at the first `}`).
+    block = _parse_tool_call_block('tool => "bash", args => {--command "echo {x} done"}')
+    assert block is not None and block.tool_type == "bash"
+    assert block.content == "echo {x} done"
+
+
+def test_fenced_invoke_still_parsed():
+    blocks = parse_tool_blocks(
+        '```python\n<invoke name="bash"><parameter name="command">whoami</parameter></invoke>\n```'
+    )
+    assert [(b.tool_type, b.content) for b in blocks] == [("bash", "whoami")]
+
+
+# ── pathological inputs no longer blow up ───────────────────────────────────
+
+def test_args_brace_opener_flood_is_fast():
+    # Many `args:{` openers, no closing `}` — old greedy capture restarted from
+    # every opener (>10s); the bounded opener + rfind is O(n).
+    evil = "args:{{a" * 14000
+    block, dt = _timed(_parse_tool_call_block, evil)
+    assert dt < _BUDGET_S, f"_parse_tool_call_block took {dt:.2f}s"
+    assert block is None
+    # And through the public path, wrapped in a [TOOL_CALL] block.
+    _, dt2 = _timed(parse_tool_blocks, "[TOOL_CALL]{" + evil + "}[/TOOL_CALL]")
+    assert dt2 < _BUDGET_S, f"parse_tool_blocks took {dt2:.2f}s"
+
+
+def test_xml_invoke_opener_flood_is_fast():
+    # Bare <invoke> opener flood, no </invoke> closer.
+    evil = ('<invoke name="x">' + "a" * 10) * 6000
+    blocks, dt = _timed(parse_tool_blocks, evil)
+    assert dt < _BUDGET_S, f"parse_tool_blocks took {dt:.2f}s"
+    assert blocks == []
+
+
+def test_xml_invoke_stale_closer_before_opener_flood_is_fast():
+    # A lone leading </invoke> satisfies a substring guard, but no opener after
+    # it has a reachable closer.
+    evil = "</invoke>" + ('<invoke name="x">' + "a" * 10) * 6000
+    _, dt = _timed(parse_tool_blocks, evil)
+    assert dt < _BUDGET_S, f"parse_tool_blocks took {dt:.2f}s"
+
+
+def test_xml_direct_backref_opener_flood_is_fast():
+    # <tool_call> wrapper (no </tool_call>) routes into the open-wrapper path,
+    # which reaches the _XML_DIRECT_TOOL_RE backreference scan: a `<a><a>...`
+    # flood with no `</a>` closer.
+    evil = "<tool_call>" + "<a><a>b" * 6000
+    blocks, dt = _timed(parse_tool_blocks, evil)
+    assert dt < _BUDGET_S, f"parse_tool_blocks took {dt:.2f}s"
+    assert blocks == []
+
+
+def test_tool_code_param_backref_flood_is_fast():
+    # `<x><x>...` param flood inside tool_code args, no `</x>` closer — exercises
+    # the `<tag>([\\s\\S]*?)</\\1>` backreference scan in _parse_tool_code_block.
+    args_flood = "tool => 'bash', args => " + "<x><x>a" * 6000
+    block, dt = _timed(_parse_tool_code_block, args_flood)
+    assert dt < _BUDGET_S, f"_parse_tool_code_block took {dt:.2f}s"
+    # Through the public path, inside a closed <tool_code> block.
+    _, dt2 = _timed(parse_tool_blocks, "<tool_code>{" + args_flood + "}</tool_code>")
+    assert dt2 < _BUDGET_S, f"parse_tool_blocks took {dt2:.2f}s"
+
+
+def test_xml_invoke_closed_with_parameter_opener_flood_is_fast():
+    # A CLOSED <invoke> whose body is a flood of `<parameter name=..>` openers
+    # with no `</parameter>` closer: the invoke delimiter pairs fine, but the
+    # inner parameter scan must not rescan the body from every opener (O(n^2)).
+    evil = ('<tool_call><invoke name="bash">'
+            + '<parameter name="x">' * 6000
+            + '</invoke></tool_call>')
+    blocks, dt = _timed(parse_tool_blocks, evil)
+    assert dt < _BUDGET_S, f"parse_tool_blocks took {dt:.2f}s"
+    # No `</parameter>` ever closes, so no params are captured.
+    assert len(blocks) == 1 and blocks[0].tool_type == "bash"
+
+
+def test_xml_direct_distinct_name_opener_flood_is_fast():
+    # Distinct unclosed tag names (`<t0><t1>...`) defeat per-name memoization;
+    # the scan must still stay near-linear instead of searching the suffix once
+    # per new name.
+    evil = "<tool_call>" + "".join(f"<t{i}>" for i in range(45000))
+    blocks, dt = _timed(parse_tool_blocks, evil)
+    assert dt < _BUDGET_S, f"parse_tool_blocks took {dt:.2f}s"
+    assert blocks == []
+
+
+def test_tool_code_param_distinct_name_flood_is_fast():
+    # Same distinct-name flood inside tool_code args, reaching the param backref
+    # scan in _parse_tool_code_block.
+    args_flood = "tool => 'bash', args => " + "".join(f"<t{i}>" for i in range(45000))
+    _, dt = _timed(_parse_tool_code_block, args_flood)
+    assert dt < _BUDGET_S, f"_parse_tool_code_block took {dt:.2f}s"

tests/test_resolve_model_offloaded.py

@@ -0,0 +1,61 @@
+"""Issue #4589 — _resolve_model does a blocking httpx.get, so calling it
+directly from an async handler stalls the whole event loop for the duration of
+the probe. The async call sites now wrap it in asyncio.to_thread.
+
+do_pipeline is used as the representative handler: _resolve_model is the first
+real work it does, and a ValueError returns early before any LLM call, so these
+tests drive the offload path without a live model endpoint.
+"""
+
+import asyncio
+import threading
+import time
+
+import src.ai_interaction as ai
+
+
+async def test_do_pipeline_resolves_model_off_the_event_loop(monkeypatch):
+    # A deliberately blocking _resolve_model that records how many copies run
+    # at once. If it ran on the event loop, the first call would block the loop
+    # and the second could not start — peak concurrency would be 1.
+    state = {"active": 0, "peak": 0}
+    lock = threading.Lock()
+
+    def slow_resolve(spec, owner=None):
+        with lock:
+            state["active"] += 1
+            state["peak"] = max(state["peak"], state["active"])
+        time.sleep(0.2)
+        with lock:
+            state["active"] -= 1
+        raise ValueError("no such model")  # early-return path, no LLM call
+
+    monkeypatch.setattr(ai, "_resolve_model", slow_resolve)
+
+    content = '[{"model": "m", "instruction": "go"}]'
+    results = await asyncio.gather(
+        ai.do_pipeline(content, owner="u"),
+        ai.do_pipeline(content, owner="u"),
+    )
+
+    assert all("error" in r for r in results)
+    assert state["peak"] == 2, "resolutions did not overlap — call still blocks the loop"
+
+
+async def test_do_pipeline_uses_offloaded_resolution_result(monkeypatch):
+    # The offload must also return the resolved tuple, not just propagate errors.
+    monkeypatch.setattr(
+        ai, "_resolve_model",
+        lambda spec, owner=None: ("http://x/v1/chat/completions", "resolved-model", {}),
+    )
+
+    async def fake_llm(url, model, messages, **kwargs):
+        return f"output from {model}"
+
+    monkeypatch.setattr("src.llm_core.llm_call_async", fake_llm)
+
+    result = await ai.do_pipeline('[{"model": "m", "instruction": "go"}]', owner="u")
+
+    assert "error" not in result, result
+    # The model the offloaded _resolve_model returned made it through to the call.
+    assert "resolved-model" in str(result)

tests/test_scheduler_prompt_cache_time.py

@@ -0,0 +1,136 @@
+"""Regression tests for #4850 — scheduled-task system prompt must not embed
+a minute-level timestamp that busts the Anthropic prompt cache.
+
+Three focused tests:
+1. End-to-end: system prompt is clean; message ordering is [system, datetime
+   user-context, task user-prompt] through the real _run_agent_loop.
+2. Fallback: same ordering when the agent loop raises and task_llm_call_async
+   is used directly.
+3. Helper: current_datetime_context_message_for_tz() renders the correct local
+   time for an explicit IANA timezone, and falls back to UTC for None or invalid.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+
+def _make_task(prompt="run the digest"):
+    return SimpleNamespace(
+        crew_member_id=None, endpoint_url="http://ep/v1", model="m",
+        session_id="s", owner="admin", prompt=prompt,
+        name="job", max_steps=5, character_id=None,
+    )
+
+
+def _patch_scheduler_deps(monkeypatch):
+    monkeypatch.setattr(
+        "src.settings.get_setting",
+        lambda key, default=None: [] if key == "disabled_tools" else default,
+    )
+    monkeypatch.setattr("src.tool_index.get_tool_index", lambda: None)
+
+
+# ---------------------------------------------------------------------------
+# Test 1 — end-to-end: system is clean; agent-loop message ordering is correct
+# ---------------------------------------------------------------------------
+
+async def test_scheduler_agent_loop_path(monkeypatch):
+    """Drive _execute_llm_task end-to-end (real _run_agent_loop, stubbed
+    stream_agent_loop).  Asserts:
+      - system message contains no 'Current time:' prefix
+      - messages[1] is a user-role date/time context block
+      - messages[2] is the task prompt
+    """
+    _patch_scheduler_deps(monkeypatch)
+
+    captured = {}
+
+    async def _stub_stream(**kwargs):
+        captured["messages"] = list(kwargs.get("messages", []))
+        return
+        yield  # async generator
+
+    monkeypatch.setattr("src.agent_loop.stream_agent_loop", _stub_stream)
+    monkeypatch.setattr("src.task_endpoint.resolve_task_candidates", lambda **kw: [])
+
+    from src.task_scheduler import TaskScheduler
+    await TaskScheduler(session_manager=None)._execute_llm_task(_make_task(), db=None)
+
+    msgs = captured.get("messages", [])
+    assert len(msgs) == 3, f"expected 3 messages, got {len(msgs)}"
+    assert msgs[0]["role"] == "system"
+    assert "Current time:" not in msgs[0]["content"]
+    assert msgs[1]["role"] == "user"
+    assert "## Current date and time" in msgs[1]["content"]
+    assert msgs[2]["role"] == "user"
+    assert msgs[2]["content"] == "run the digest"
+
+
+# ---------------------------------------------------------------------------
+# Test 2 — fallback path receives the same datetime context
+# ---------------------------------------------------------------------------
+
+async def test_scheduler_fallback_path(monkeypatch):
+    """When _run_agent_loop raises, task_llm_call_async must receive
+    [system, datetime user-context, task user-prompt] — the same ordering."""
+    _patch_scheduler_deps(monkeypatch)
+
+    captured = {}
+
+    async def _fail(*args, **kwargs):
+        raise RuntimeError("simulated failure")
+
+    async def _capture_call(messages, **kw):
+        captured["messages"] = list(messages)
+        return "fallback"
+
+    import src.task_endpoint as _te
+    monkeypatch.setattr(_te, "task_llm_call_async", _capture_call)
+
+    from src.task_scheduler import TaskScheduler
+    sched = TaskScheduler(session_manager=None)
+    sched._run_agent_loop = _fail
+    await sched._execute_llm_task(_make_task(prompt="send the digest"), db=None)
+
+    msgs = captured.get("messages", [])
+    assert len(msgs) == 3, f"expected 3 messages, got {len(msgs)}"
+    assert msgs[0]["role"] == "system"
+    assert "Current time:" not in msgs[0]["content"]
+    assert msgs[1]["role"] == "user"
+    assert "## Current date and time" in msgs[1]["content"]
+    assert msgs[2]["role"] == "user"
+    assert msgs[2]["content"] == "send the digest"
+
+
+# ---------------------------------------------------------------------------
+# Test 3 — current_datetime_context_message_for_tz() timezone resolution
+# ---------------------------------------------------------------------------
+
+def test_datetime_context_message_for_tz(monkeypatch):
+    """Three cases with a fixed UTC timestamp (2026-06-25 18:00 UTC):
+      - explicit 'America/New_York' → 2:00 PM EDT, UTC-04:00
+      - None                        → UTC fallback: 6:00 PM, UTC+00:00
+      - invalid IANA name           → UTC fallback: same
+    """
+    from src.user_time import current_datetime_context_message_for_tz
+
+    fixed = datetime(2026, 6, 25, 18, 0, tzinfo=timezone.utc)
+
+    # Explicit IANA timezone
+    msg = current_datetime_context_message_for_tz("America/New_York", fixed)
+    assert msg["role"] == "user"
+    assert "America/New_York" in msg["content"]
+    assert "UTC-04:00" in msg["content"]
+    assert "2:00 PM" in msg["content"]
+
+    # None → UTC (preserves old scheduler behaviour for tasks without a crew tz)
+    msg = current_datetime_context_message_for_tz(None, fixed)
+    assert "UTC+00:00" in msg["content"]
+    assert "6:00 PM" in msg["content"]
+
+    # Invalid IANA name → UTC fallback, no exception raised
+    msg = current_datetime_context_message_for_tz("Not/A_Real_Zone", fixed)
+    assert "UTC+00:00" in msg["content"]
+    assert "6:00 PM" in msg["content"]

tests/test_task_shell_tools.py

@@ -111,7 +111,8 @@ async def test_scheduled_task_honors_global_disabled_tools(monkeypatch):
     captured = {}
 
     async def _capture(endpoint_url, model, task, session_id, *,
-                       system_prompt=None, disabled_tools=None, relevant_tools=None):
+                       system_prompt=None, disabled_tools=None, relevant_tools=None,
+                       datetime_context_msg=None):
         captured["disabled_tools"] = disabled_tools
         captured["relevant_tools"] = relevant_tools
         return "done"

tests/test_web_search_query_sanitization.py

@@ -0,0 +1,209 @@
+"""Regression tests for #4547 — chat-mode web search query sanitization.
+
+Chat-mode web search (``use_web``) selects a search query via the
+generated-query flow added in #4557: an LLM extracts a concise query, falling
+back to the first non-empty line of the user message when the LLM fails or
+returns an empty result. PR #4863 layers a focused, *defensive* cleanup on top
+of that flow: whatever query is finally selected (generated or fallback) is
+passed through ``_clean_search_query()`` before reaching
+``comprehensive_web_search()``, so residual fenced/inline markdown never leaks
+into the search call.
+
+``_clean_search_query()`` renders the query to HTML via ``markdown``
+(``fenced_code`` extension), drops ``<pre>`` blocks entirely, unwraps inline
+``<code>`` to its text (so ``git reset`` survives), collapses whitespace, and
+truncates.
+
+The first four tests pin the helper directly; the last three prove it is
+wired into the production path and that the combined generated-query +
+sanitization behaviour holds for all three selection outcomes (generated
+success, LLM exception, empty LLM result).
+
+This is intentionally a narrow interim/defensive fix for #4547; it does not
+replace the generated-query flow from #4557.
+"""
+from src.chat_processor import ChatProcessor, _clean_search_query
+
+
+# ── Unit tests: _clean_search_query ──
+
+
+def test_clean_search_query_removes_fenced_code_blocks():
+    """A fenced code block must be dropped entirely, including the code body
+    and the fences — only the surrounding prose survives."""
+    message = '```python\nprint("hello")\n```\nWhat is the capital of France?'
+
+    result = _clean_search_query(message)
+
+    assert result == "What is the capital of France?"
+    # Guards against the original leak: no fences, no code body.
+    assert "```" not in result
+    assert "print" not in result
+
+
+def test_clean_search_query_preserves_inline_code():
+    """Inline code text is search-relevant and must survive unwrapped; only the
+    backticks are removed. This is the ``git reset`` case the reviewer flagged
+    against the earlier regex approach (which dropped the word entirely)."""
+    message = "Is it a good idea to use `git reset` to undo my changes?"
+
+    result = _clean_search_query(message)
+
+    assert result == "Is it a good idea to use git reset to undo my changes?"
+    assert "git reset" in result
+    assert "`" not in result
+
+
+def test_clean_search_query_collapses_whitespace():
+    """Runs of whitespace (tabs, multiple spaces, newlines) collapse to a single
+    space so the query is a single clean line."""
+    message = "hello\tworld   foo\n\n   bar"
+
+    result = _clean_search_query(message)
+
+    assert result == "hello world foo bar"
+    assert "  " not in result
+    assert "\n" not in result
+    assert "\t" not in result
+
+
+def test_clean_search_query_truncates_long_input():
+    """Long queries are capped at ``max_len`` (default 200) to stay within search
+    API limits; truncation is a strict prefix of the cleaned text."""
+    long_message = "x" * 300
+
+    result_default = _clean_search_query(long_message)
+    result_custom = _clean_search_query(long_message, max_len=10)
+
+    assert len(result_default) == 200
+    assert result_default == "x" * 200
+    assert result_custom == "x" * 10
+
+
+# ── Integration tests: the generated-query + sanitization flow ──
+#
+# These cover the combined behaviour requested in review of #4863 after #4557
+# landed: the LLM-generated query is used on success, the first-line fallback is
+# used when the LLM fails or returns empty, and in every case the *final* query
+# handed to comprehensive_web_search() is sanitized.
+
+# A messy user message whose first non-empty line (the #4557 fallback) is
+# inline-code prose followed by a fenced block. After sanitization the fallback
+# collapses to plain prose.
+_MESSY = 'Is `git reset` safe?\n```python\nprint("leaked body")\n```'
+_SANITIZED_FALLBACK = "Is git reset safe?"
+
+
+class _Session:
+    """Minimal stand-in for the session object read by the generated-query
+    flow (endpoint_url / model / headers)."""
+
+    endpoint_url = "http://example.local/v1"
+    model = "test-model"
+    headers = {"Authorization": "Bearer test"}
+
+
+class _Memory:
+    def load(self, owner=None):
+        return []
+
+
+class _Docs:
+    rag_manager = None
+
+
+def _patch_flow(monkeypatch, llm_behaviour, captured):
+    """Wire both seams of the generated-query flow: the LLM call and the
+    search call. ``llm_behaviour`` is either a string to return or an Exception
+    instance to raise."""
+
+    def _fake_search(query, *args, **kwargs):
+        captured["query"] = query
+        captured["kwargs"] = kwargs
+        return ("web context", [{"title": "src"}])
+
+    def _fake_llm(*args, **kwargs):
+        if isinstance(llm_behaviour, Exception):
+            raise llm_behaviour
+        return llm_behaviour
+
+    monkeypatch.setattr("src.chat_processor.comprehensive_web_search", _fake_search)
+    monkeypatch.setattr("src.llm_core.llm_call", _fake_llm)
+
+
+def test_generated_query_is_used_and_sanitized(monkeypatch):
+    """Requirement: on LLM success the generated query wins, and the *final*
+    query handed to comprehensive_web_search() is sanitized.
+
+    The fake LLM returns a query containing inline-code markdown so we can also
+    prove the sanitizer runs on the generated path (not just the fallback)."""
+    captured = {}
+    _patch_flow(monkeypatch, "capital of `France`", captured)
+
+    processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+    preface, _, _ = processor.build_context_preface(
+        message=_MESSY,
+        session=_Session(),
+        use_web=True,
+        use_memory=False,
+        use_rag=False,
+    )
+
+    assert "query" in captured, "comprehensive_web_search was not called"
+
+    # The generated query won (not the sanitized first-line fallback) ...
+    assert captured["query"] == "capital of France"
+    assert captured["query"] != _SANITIZED_FALLBACK
+    # ... and it was sanitized: no residual markdown fences/backticks.
+    assert "`" not in captured["query"]
+    assert "```" not in captured["query"]
+
+    # The other call-site kwargs (return_sources) are still forwarded.
+    assert captured["kwargs"].get("return_sources") is True
+    # And the retrieved context was still appended to the preface.
+    assert any("web context" in (msg.get("content") or "") for msg in preface)
+
+
+def test_falls_back_to_sanitized_first_line_when_llm_raises(monkeypatch):
+    """Requirement: when the LLM call raises, #4557's fallback (first non-empty
+    line) is used — and that fallback is sanitized before the search call."""
+    captured = {}
+    _patch_flow(monkeypatch, RuntimeError("LLM endpoint down"), captured)
+
+    processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+    processor.build_context_preface(
+        message=_MESSY,
+        session=_Session(),
+        use_web=True,
+        use_memory=False,
+        use_rag=False,
+    )
+
+    assert "query" in captured, "comprehensive_web_search was not called"
+    # Fallback was the first line ("Is `git reset` safe?"), sanitized.
+    assert captured["query"] == _SANITIZED_FALLBACK
+    assert "git reset" in captured["query"]  # inline code preserved
+    assert "`" not in captured["query"]  # backticks stripped
+    # The fenced body from later lines never reached the query.
+    assert "leaked body" not in captured["query"]
+
+
+def test_falls_back_to_sanitized_first_line_when_llm_returns_empty(monkeypatch):
+    """Requirement: when the LLM returns an empty/whitespace-only query, #4557
+    falls back — and that fallback is sanitized before the search call."""
+    captured = {}
+    _patch_flow(monkeypatch, "   ", captured)
+
+    processor = ChatProcessor(memory_manager=_Memory(), personal_docs_manager=_Docs())
+    processor.build_context_preface(
+        message=_MESSY,
+        session=_Session(),
+        use_web=True,
+        use_memory=False,
+        use_rag=False,
+    )
+
+    assert "query" in captured, "comprehensive_web_search was not called"
+    assert captured["query"] == _SANITIZED_FALLBACK
+    assert "git reset" in captured["query"]
+    assert "`" not in captured["query"]