fix(agent): preserve loop guard stream behavior

* fix: resolve RAG manager search signature TypeError and adjust similarity threshold

* fix: revert similarity threshold change to keep PR focused

* test(rag): remove trailing whitespace

---------

Co-authored-by: Alexandre Teixeira <alexandremagteixeira@gmail.com>

requirements.txt

@@ -3,6 +3,7 @@ uvicorn
 python-multipart
 python-dotenv
 httpx
+httpcore>=1.0,<2.0
 pydantic>=2.13.4
 pydantic-settings>=2.14.1
 SQLAlchemy

routes/chat_routes.py

@@ -1327,6 +1327,8 @@ def setup_chat_routes(
                                     "doc_stream_open", "doc_stream_delta",
                                     "doc_update", "doc_suggestions", "ui_control",
                                     "rounds_exhausted",
+                                    "loop_breaker_triggered",
+                                    "intent_nudge_exhausted",
                                     "ask_user",
                                     "plan_update",
                                 ):

routes/research/research_routes.py

@@ -20,6 +20,26 @@ from src.constants import DEEP_RESEARCH_DIR
 
 _SESSION_ID_RE = re.compile(r"^[a-zA-Z0-9-]{1,128}$")
 
+
+def _confine_research_path(session_id: str) -> Path:
+    """Return the resolved Path for session_id's JSON inside DEEP_RESEARCH_DIR.
+
+    Validates the session ID format and asserts containment after symlink
+    expansion. Raises HTTPException(400) on format failures, traversal
+    attempts, absolute-path injection, and symlink escape so every caller
+    gets a safe, confined path with no extra validation needed.
+    """
+    if not _SESSION_ID_RE.fullmatch(session_id):
+        raise HTTPException(400, "Invalid session ID")
+    root = Path(DEEP_RESEARCH_DIR).resolve()
+    candidate = (root / f"{session_id}.json").resolve()
+    try:
+        candidate.relative_to(root)
+    except ValueError:
+        raise HTTPException(400, "Invalid session ID")
+    return candidate
+
+
 logger = logging.getLogger(__name__)
 
 # Model-name substrings that are NOT chat/generation models — research must
@@ -183,7 +203,10 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         if entry is not None:
             return entry.get("owner", "") == user
         # Task no longer in memory — check the persisted JSON.
-        path = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+        try:
+            path = _confine_research_path(session_id)
+        except HTTPException:
+            return False
         if not path.exists():
             return False
         try:
@@ -247,7 +270,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
     def _assert_owns_research(session_id: str, user: str) -> None:
         """404-not-403 ownership gate for a research session's on-disk JSON.
         Use BEFORE returning any data or mutating the file."""
-        path = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+        path = _confine_research_path(session_id)
         if not path.exists():
             raise HTTPException(404, "Research not found")
         try:
@@ -361,7 +384,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         summary, stats — used by the Library preview panel."""
         user = _require_user(request)
         _validate_session_id(session_id)
-        path = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+        path = _confine_research_path(session_id)
         if not path.exists():
             raise HTTPException(404, "Research not found")
         try:
@@ -378,7 +401,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         """Soft-archive / restore a research report (sets `archived` in its JSON)."""
         user = _require_user(request)
         _validate_session_id(session_id)
-        path = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+        path = _confine_research_path(session_id)
         if not path.exists():
             raise HTTPException(404, "Research not found")
         try:
@@ -398,8 +421,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         """Delete a research result from disk."""
         user = _require_user(request)
         _validate_session_id(session_id)
-        data_dir = Path(DEEP_RESEARCH_DIR)
-        json_path = data_dir / f"{session_id}.json"
+        json_path = _confine_research_path(session_id)
         deleted = False
         if json_path.exists():
             # SECURITY: verify ownership before letting the caller delete it.
@@ -561,7 +583,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
             raise HTTPException(404, "No research found for this session")
         result = research_handler.get_result(session_id)
         if result is None:
-            p = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+            p = _confine_research_path(session_id)
             if p.exists():
                 d = json.loads(p.read_text(encoding="utf-8"))
                 return {
@@ -601,7 +623,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         sources = research_handler.get_sources(session_id) or []
         query = ""
 
-        path = Path(DEEP_RESEARCH_DIR) / f"{session_id}.json"
+        path = _confine_research_path(session_id)
         if path.exists():
             try:
                 disk = json.loads(path.read_text(encoding="utf-8"))

services/search/content.py

@@ -8,11 +8,13 @@ import os
 import re
 import logging
 import socket
+import ssl
 from datetime import datetime, timedelta
-from typing import List
+from typing import Iterable, List, cast
 from urllib.parse import urljoin, urlparse
 
 import httpx
+import httpcore
 from bs4 import BeautifulSoup
 
 from src.constants import WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES, WEB_FETCH_USER_AGENT
@@ -91,6 +93,148 @@ def _public_http_url(url: str) -> bool:
         return False
 
 
+def _resolve_public_ips(url: str) -> list[ipaddress._BaseAddress]:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https") or not parsed.hostname:
+        raise httpx.RequestError(f"Blocked non-public URL: {url}")
+    host = (parsed.hostname or "").strip().lower()
+    if host in ("localhost", "metadata", "metadata.google.internal"):
+        raise httpx.RequestError(f"Blocked non-public hostname: {host}")
+    try:
+        ip = ipaddress.ip_address(host)
+        if _is_private_address(ip):
+            raise httpx.RequestError(f"Blocked non-public IP literal: {host}")
+        return [ip]
+    except httpx.RequestError:
+        raise
+    except ValueError:
+        pass
+    addrs = _resolve_hostname_ips(host)
+    if not addrs or any(_is_private_address(a) for a in addrs):
+        raise httpx.RequestError(f"Blocked non-public URL: {url}")
+    return addrs
+
+
+class _PinnedBackend(httpcore.NetworkBackend):
+    """Network backend that connects to a pre-resolved IP.
+
+    httpcore derives the TLS SNI and the ``Host`` header from the URL's
+    origin, not from the host argument passed to ``connect_tcp``. So
+    routing the TCP connect to a resolved IP while leaving the URL
+    untouched keeps SNI / vhost behaviour correct and closes the
+    DNS-rebinding TOCTOU between the SSRF check and the connect.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress):
+        self._ip = str(ip)
+        self._real = httpcore.SyncBackend()
+
+    def connect_tcp(
+        self,
+        host: str,
+        port: int,
+        timeout: float | None = None,
+        local_address: str | None = None,
+        socket_options=None,
+    ):
+        return self._real.connect_tcp(
+            self._ip, port, timeout, local_address, socket_options
+        )
+
+    def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return self._real.connect_unix_socket(path, timeout, socket_options)
+
+    def sleep(self, seconds: float) -> None:
+        return self._real.sleep(seconds)
+
+
+# Map httpcore exception classes to their httpx equivalents. Built
+# once at import time from the public exception classes; avoids any
+# import of httpx's private transport machinery. httpcore's
+# ``ConnectionNotAvailable`` is a pool-internal signal (the pool will
+# close and retry on its own) — we never expect to see it surface to
+# a transport caller, so it has no httpx counterpart here.
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.LocalProtocolError: httpx.LocalProtocolError,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ProxyError: httpx.ProxyError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.UnsupportedProtocol: httpx.UnsupportedProtocol,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedTransport(httpx.BaseTransport):
+    """Transport that pins every TCP connect to a pre-resolved IP.
+
+    Uses only the public ``httpcore`` and ``httpx`` APIs — no
+    subclassing of ``httpx.HTTPTransport``, no reads of private
+    ``httpcore.ConnectionPool`` attributes, no imports from
+    ``httpx private transport internals``. The URL is passed through unchanged so SNI
+    / vhost work as if httpx had been given the hostname directly;
+    only the TCP destination is pinned, closing the DNS-rebinding
+    TOCTOU between the SSRF check and the connect.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress, *, http2: bool = False):
+        self._pool = httpcore.ConnectionPool(
+            ssl_context=ssl.create_default_context(),
+            http1=True,
+            http2=http2,
+            network_backend=_PinnedBackend(ip),
+        )
+
+    def __enter__(self):
+        self._pool.__enter__()
+        return self
+
+    def __exit__(self, exc_type=None, exc_value=None, traceback=None) -> None:
+        self._pool.__exit__(exc_type, exc_value, traceback)
+
+    def handle_request(self, request: httpx.Request) -> httpx.Response:
+        httpcore_req = httpcore.Request(
+            method=request.method,
+            url=httpcore.URL(
+                scheme=request.url.raw_scheme,
+                host=request.url.raw_host,
+                port=request.url.port,
+                target=request.url.raw_path,
+            ),
+            headers=request.headers.raw,
+            content=request.stream,
+            extensions=request.extensions,
+        )
+        try:
+            httpcore_resp = self._pool.handle_request(httpcore_req)
+            # Eager materialisation matches the original
+            # ``response.text`` usage in fetch_webpage_content. The
+            # sync pool's stream is a plain Iterable[bytes] despite
+            # the httpcore type hint unioning the async variant.
+            content = b"".join(cast(Iterable[bytes], httpcore_resp.stream))
+        except Exception as exc:
+            mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc))
+            if mapped is not None:
+                raise mapped(str(exc)) from exc
+            raise
+
+        return httpx.Response(
+            status_code=httpcore_resp.status,
+            headers=httpcore_resp.headers,
+            content=content,
+            extensions=httpcore_resp.extensions,
+        )
+
+    def close(self) -> None:
+        self._pool.close()
+
 class BodyTooLargeError(Exception):
     """The server declared a body larger than the hard fetch ceiling."""
 
@@ -141,78 +285,78 @@ class _CappedFetch:
 
 def _get_public_url(url: str, headers: dict, timeout: int, max_redirects: int = 5,
                     max_bytes: int = None) -> "_CappedFetch":
-    """Capped streaming GET with SSRF-guarded manual redirects.
+    """Capped streaming GET with SSRF-guarded, DNS-pinned manual redirects.
 
-    The body is streamed and buffering stops at ``max_bytes`` (default: the
-    soft cap), so an oversized resource cannot be pulled into memory or the
-    content cache in full. When Content-Length already declares a body over
-    the hard ceiling, the fetch is refused before any body bytes are read.
+    Each hop is resolved once, validated as public, and then the actual TCP
+    connection is pinned to that resolved IP. The request URL is left unchanged
+    so Host and TLS SNI keep the original hostname.
     """
     cap = min(max_bytes or WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES)
     current = url
     for _ in range(max_redirects + 1):
-        if not _public_http_url(current):
-            raise httpx.RequestError("Blocked private/internal URL", request=httpx.Request("GET", current))
+        ips = _resolve_public_ips(current)
+
         # Force identity transfer-encoding. With gzip/deflate the wire bytes
-        # (and Content-Length) can be a small fraction of the decoded body, so
-        # a tiny compressed response could pass the hard-cap preflight and then
-        # expand past the ceiling in a single decoded chunk before the streamed
-        # cap below can slice it. Identity makes Content-Length the true body
-        # size and keeps each streamed chunk bounded by the network read.
+        # and Content-Length can be a small fraction of the decoded body, so a
+        # tiny compressed response could pass the hard-cap preflight and then
+        # expand past the ceiling in one decoded chunk before the streamed cap
+        # below can slice it.
         req_headers = dict(headers or {})
         req_headers["Accept-Encoding"] = "identity"
-        with httpx.stream("GET", current, headers=req_headers, timeout=timeout,
-                          follow_redirects=False) as response:
-            if response.status_code in (301, 302, 303, 307, 308):
-                location = response.headers.get("location")
-                if not location:
-                    return _CappedFetch(response.status_code, response.headers, b"",
-                                        False, None, response.encoding, str(response.url))
-                current = urljoin(str(response.url), location)
-                continue
 
-            # A server can ignore the identity request and still return a
-            # compressed body; httpx.iter_bytes would then decode it, and a tiny
-            # gzip can balloon into one decoded chunk far past the cap before we
-            # slice. Refuse a compressed Content-Encoding so the streamed cap
-            # stays a real memory bound (Content-Length is the compressed wire
-            # length here, so the preflight and size metadata are unreliable too).
-            enc = (response.headers.get("content-encoding") or "").strip().lower()
-            if enc and enc != "identity":
-                raise httpx.RequestError(
-                    f"Refusing compressed response (Content-Encoding: {enc}) after "
-                    "requesting identity: cannot bound decoded body size",
-                    request=httpx.Request("GET", current),
-                )
+        with httpx.Client(
+            headers=req_headers,
+            timeout=timeout,
+            follow_redirects=False,
+            transport=_PinnedTransport(ips[0]),
+        ) as client:
+            with client.stream("GET", current) as response:
+                if response.status_code in (301, 302, 303, 307, 308):
+                    location = response.headers.get("location")
+                    if not location:
+                        return _CappedFetch(response.status_code, response.headers, b"",
+                                            False, None, response.encoding, str(response.url))
+                    current = urljoin(str(response.url), location)
+                    continue
 
-            declared = None
-            raw_len = response.headers.get("content-length")
-            if raw_len and raw_len.isdigit():
-                declared = int(raw_len)
-            # Refuse before buffering anything when the server already tells
-            # us the body exceeds the absolute ceiling (Content-Length is wire
-            # bytes; the decompressed body can only be larger).
-            if declared is not None and declared > WEB_FETCH_HARD_MAX_BYTES:
-                raise BodyTooLargeError(current, declared)
+                # A server can ignore the identity request and still return a
+                # compressed body; httpx.iter_bytes would then decode it, and a
+                # tiny gzip can balloon into one decoded chunk far past the cap.
+                # Refuse compressed Content-Encoding so the streamed cap stays
+                # a real memory bound.
+                enc = (response.headers.get("content-encoding") or "").strip().lower()
+                if enc and enc != "identity":
+                    raise httpx.RequestError(
+                        f"Refusing compressed response (Content-Encoding: {enc}) after "
+                        "requesting identity: cannot bound decoded body size",
+                        request=httpx.Request("GET", current),
+                    )
+
+                declared = None
+                raw_len = response.headers.get("content-length")
+                if raw_len and raw_len.isdigit():
+                    declared = int(raw_len)
+
+                if declared is not None and declared > WEB_FETCH_HARD_MAX_BYTES:
+                    raise BodyTooLargeError(current, declared)
+
+                chunks = []
+                read = 0
+                truncated = False
+                for chunk in response.iter_bytes():
+                    read += len(chunk)
+                    if read > cap:
+                        keep = cap - (read - len(chunk))
+                        if keep > 0:
+                            chunks.append(chunk[:keep])
+                        truncated = True
+                        break
+                    chunks.append(chunk)
+
+                return _CappedFetch(response.status_code, response.headers,
+                                    b"".join(chunks), truncated, declared,
+                                    response.encoding, str(response.url))
 
-            chunks = []
-            read = 0
-            truncated = False
-            # We requested identity above, so iter_bytes yields the raw body in
-            # network-read-sized chunks (no decompression expansion); the cap
-            # therefore bounds what we actually buffer.
-            for chunk in response.iter_bytes():
-                read += len(chunk)
-                if read > cap:
-                    keep = cap - (read - len(chunk))
-                    if keep > 0:
-                        chunks.append(chunk[:keep])
-                    truncated = True
-                    break
-                chunks.append(chunk)
-            return _CappedFetch(response.status_code, response.headers,
-                                b"".join(chunks), truncated, declared,
-                                response.encoding, str(response.url))
     raise httpx.RequestError("Too many redirects", request=httpx.Request("GET", current))
 
 # PDF extraction (optional dependency)

src/agent_loop.py

@@ -42,6 +42,167 @@ from src.agent_tools import (
 
 logger = logging.getLogger(__name__)
 
+# Redaction patterns for common secret-bearing shapes. Explicit and tested
+# (see tests/test_loop_guard_signals.py) rather than one clever broad regex —
+# safety first, but we try not to mangle harmless prose. Applied in order.
+_REDACTED = "[redacted]"
+
+# Cookie: ... / Set-Cookie: ... — redact the rest of the line (cookies hold spaces).
+_SENSITIVE_COOKIE_RE = re.compile(
+    r"(?i)\b((?:set-)?cookie\s*[:=]\s*)[^\r\n]+"
+)
+# URL credentials, e.g. postgres://user:pass@host/db. The password half allows
+# inner colons (postgres://user:pa:ss@host/db) but still stops at / and @.
+_SENSITIVE_URL_CRED_RE = re.compile(
+    r"(?i)\b([a-z][a-z0-9+.\-]*://)[^\s:/@]+:[^\s/@]+@"
+)
+# Prefix-only discovery regexes. Each matches the key and its separator (the part
+# we KEEP); the value that follows is found by a linear scanner rather than by a
+# regex, so there is no backtracking-prone quantifier over uncontrolled input.
+#
+# Authorization: Bearer <tok> / Authorization: Basic "two word secret"
+_AUTH_PREFIX_RE = re.compile(
+    r"(?i)authorization\s*[:=]\s*(?:bearer|basic)\s+"
+)
+# Provider-prefixed env names, e.g. OPENAI_API_KEY=..., AWS_SECRET_ACCESS_KEY=...,
+# GITHUB_TOKEN=... — require a sensitive suffix preceded by `_` so benign names
+# that merely end in KEY (MONKEY, TURKEY) are left alone.
+_ENV_PREFIX_RE = re.compile(
+    r"(?:export\s+)?\b[A-Z][A-Z0-9_]*"
+    r"_(?:KEY|TOKEN|SECRET|PASSWORD|PASSWD|PWD|CREDENTIALS?)\s*=\s*"
+)
+# Generic sensitive key, e.g. password=..., api_key: ..., client_secret=...
+_KEY_PREFIX_RE = re.compile(
+    r"(?i)\b(?:password|passwd|pwd|token|api[_-]?key|client_secret|secret)\b\s*[:=]\s*"
+)
+# Obvious provider-shaped bare tokens (no surrounding key needed).
+_SENSITIVE_BARE_TOKEN_RE = re.compile(
+    r"\b("
+    r"sk-[A-Za-z0-9_\-]{16,}"          # OpenAI / Anthropic style
+    r"|gh[pousr]_[A-Za-z0-9]{20,}"     # GitHub PAT
+    r"|xox[baprs]-[A-Za-z0-9\-]{10,}"  # Slack
+    r"|AKIA[0-9A-Z]{16}"               # AWS access key id
+    r"|hf_[A-Za-z0-9]{16,}"            # Hugging Face token
+    r"|AIza[0-9A-Za-z_\-]{20,}"        # Google API key
+    r")\b"
+)
+
+
+def _consume_secret_value_end(text: str, start: int) -> int:
+    """Return the exclusive end index of the secret value beginning at ``start``.
+
+    If the value is quoted, scan to the matching unescaped quote (backslash
+    escapes are skipped two chars at a time). Otherwise scan to the first
+    whitespace, comma, or semicolon. The scan is linear in the length of the
+    input, so it cannot exhibit catastrophic backtracking.
+    """
+    n = len(text)
+    if start >= n:
+        return start
+    quote = text[start]
+    if quote in ("'", '"'):
+        i = start + 1
+        while i < n:
+            ch = text[i]
+            if ch == "\\":
+                i += 2
+                continue
+            if ch == quote:
+                return i + 1
+            i += 1
+        return n  # unterminated quote: redact to the end
+    i = start
+    while i < n and not text[i].isspace() and text[i] not in (",", ";"):
+        i += 1
+    return i
+
+
+def _redact_after_prefix(text: str, prefix_re: "re.Pattern") -> str:
+    """Redact the value following each ``prefix_re`` match using a linear scan."""
+    result = []
+    pos = 0
+    n = len(text)
+    while pos < n:
+        match = prefix_re.search(text, pos)
+        if match is None:
+            result.append(text[pos:])
+            break
+        result.append(text[pos:match.end()])
+        value_end = _consume_secret_value_end(text, match.end())
+        if value_end > match.end():
+            result.append(_REDACTED)
+            pos = value_end
+        else:
+            # Empty value: nothing to redact; step past the prefix and continue.
+            pos = match.end()
+            if pos < n:
+                result.append(text[pos])
+                pos += 1
+    return "".join(result)
+
+
+def _redact_private_keys(text: str) -> str:
+    """Replace PEM private-key blocks with a placeholder via linear scanning.
+
+    Finds ``-----BEGIN `` markers, verifies the header names a PRIVATE KEY,
+    locates the matching ``-----END `` marker, and collapses the whole block.
+    No regex is used, so the (multi-line, uncontrolled) body cannot trigger
+    polynomial matching.
+    """
+    begin_marker = "-----BEGIN "
+    end_marker = "-----END "
+    dash = "-----"
+    max_header = 64  # generous bound on "[TYPE ]PRIVATE KEY"
+    result = []
+    pos = 0
+    while True:
+        begin = text.find(begin_marker, pos)
+        if begin == -1:
+            result.append(text[pos:])
+            return "".join(result)
+        header_start = begin + len(begin_marker)
+        header_close = text.find(dash, header_start)
+        if (
+            header_close == -1
+            or header_close - header_start > max_header
+            or not text[header_start:header_close].endswith("PRIVATE KEY")
+        ):
+            result.append(text[pos:header_start])
+            pos = header_start
+            continue
+        end = text.find(end_marker, header_close)
+        if end == -1:
+            result.append(text[pos:])
+            return "".join(result)
+        end_header_start = end + len(end_marker)
+        end_close = text.find(dash, end_header_start)
+        if (
+            end_close == -1
+            or end_close - end_header_start > max_header
+            or not text[end_header_start:end_close].endswith("PRIVATE KEY")
+        ):
+            result.append(text[pos:header_start])
+            pos = header_start
+            continue
+        result.append(text[pos:begin])
+        result.append("[redacted private key]")
+        pos = end_close + len(dash)
+
+
+def _redact_sensitive_text(value: object) -> str:
+    """Redact obvious credential values before surfacing tool output."""
+    if value is None:
+        return ""
+
+    text = str(value)
+    text = _redact_private_keys(text)
+    text = _redact_after_prefix(text, _AUTH_PREFIX_RE)
+    text = _SENSITIVE_COOKIE_RE.sub(r"\1" + _REDACTED, text)
+    text = _SENSITIVE_URL_CRED_RE.sub(r"\1" + _REDACTED + "@", text)
+    text = _redact_after_prefix(text, _ENV_PREFIX_RE)
+    text = _redact_after_prefix(text, _KEY_PREFIX_RE)
+    return _SENSITIVE_BARE_TOKEN_RE.sub(_REDACTED, text)
+
 
 def _load_mcp_disabled_map() -> Dict[str, set]:
     """Load per-server disabled tool sets from the database."""
@@ -2882,6 +3043,7 @@ async def stream_agent_loop(
     # signatures + consecutive no-text tool rounds to bail early.
     _recent_call_sigs = collections.deque(maxlen=6)
     _stuck_rounds = 0
+    _MAX_STUCK_ROUNDS = 4  # consecutive no-progress rounds before loop-breaker bails
     # Frequency of each exact call signature (tool + args), for the runaway
     # backstop. Counting identical repeats — not distinct same-tool calls —
     # lets a legit batch (e.g. 18 calendar events at once) through.
@@ -3411,17 +3573,22 @@ async def stream_agent_loop(
             # promise: short response (<400 chars), no fenced code/answer,
             # and an action-intent phrase was matched. Long answers that
             # happen to contain "let me know" are not stalls.
-            _looks_like_promise = (
+            _promise_shape = (
                 not guide_only
                 and _intent_match is not None
                 and len(_intent_text) < 400
                 and "```" not in _intent_text
-                and _intent_nudge_count < _MAX_INTENT_NUDGES
             )
+            _looks_like_promise = _promise_shape and _intent_nudge_count < _MAX_INTENT_NUDGES
             if _looks_like_promise:
                 _intent_nudge_count += 1
                 _matched_phrase = _intent_match.group(0).strip()
-                logger.info(f"[agent] intent-without-action nudge #{_intent_nudge_count} on round {round_num}: {_matched_phrase!r}")
+                # Don't log the matched phrase — it's raw model text that may
+                # carry credentials. Structural metadata only.
+                logger.info(
+                    "[agent] intent-without-action nudge #%d on round %d",
+                    _intent_nudge_count, round_num,
+                )
                 _lower_phrase = _matched_phrase.lower()
                 _cookbook_log_hint = ""
                 if any(_word in _lower_phrase for _word in ("log", "logs", "output", "tail", "status")):
@@ -3447,6 +3614,24 @@ async def stream_agent_loop(
                 # Visible signal in the stream so the user knows we caught it.
                 yield f'data: {json.dumps({"type": "agent_step", "round": round_num + 1})}\n\n'
                 continue
+            # The model keeps announcing actions it never takes and we've spent
+            # every nudge — surface why the turn is ending instead of letting it
+            # look like a clean completion.
+            if _promise_shape and _intent_nudge_count >= _MAX_INTENT_NUDGES:
+                _matched_phrase = _intent_match.group(0).strip()
+                _matched_phrase_safe = _redact_sensitive_text(_matched_phrase)
+                _in_message = (
+                    f"Intent-nudge cap reached on round {round_num}: the model "
+                    f"announced an action ({_matched_phrase_safe!r}) without a tool call "
+                    f"after {_intent_nudge_count} nudge(s); ending the turn."
+                )
+                # Do not log the matched phrase, even redacted. It is raw model
+                # text and may contain credentials; keep logs structural only.
+                logger.warning(
+                    "[agent] intent-nudge cap exhausted on round %d (%d/%d)",
+                    round_num, _intent_nudge_count, _MAX_INTENT_NUDGES,
+                )
+                yield f'data: {json.dumps({"type": "intent_nudge_exhausted", "round": round_num, "nudges": _intent_nudge_count, "max_nudges": _MAX_INTENT_NUDGES, "message": _in_message})}\n\n'
             break  # no tools — done
 
         # ── Loop-breaker (Terminus-style stall detector) ──────────────
@@ -3479,10 +3664,23 @@ async def stream_agent_loop(
         # Distinct calls to one tool (a real batch) are legitimate work, so we
         # count identical call signatures, not raw per-tool-type totals.
         _runaway = _detect_runaway_call(_call_freq)
-        if _stuck_rounds >= 4 or _runaway:
+        if _stuck_rounds >= _MAX_STUCK_ROUNDS or _runaway:
             reason = (f"calling {_runaway} with identical arguments over and over" if _runaway
                       else "repeating the same tool calls without new progress")
-            logger.warning(f"[agent] loop-breaker tripped on round {round_num} ({reason}); sig={_sig[:80]!r}")
+            _lb_message = (
+                f"Loop-breaker stopped the agent on round {round_num}: {reason}. "
+                "Forced one tool-free round to converge on an answer or state what's blocked."
+            )
+            # Log structural metadata only — `_sig` is raw tool-call content
+            # that may carry credentials.
+            logger.warning(
+                "[agent] loop-breaker tripped on round %d (%s); "
+                "stuck_rounds=%d/%d runaway=%r",
+                round_num, reason, _stuck_rounds, _MAX_STUCK_ROUNDS, _runaway,
+            )
+            # Surface the stop cause to the stream so the user (and journalctl)
+            # can tell a guard fired, not a clean completion.
+            yield f'data: {json.dumps({"type": "loop_breaker_triggered", "round": round_num, "reason": reason, "stuck_rounds": _stuck_rounds, "max_stuck_rounds": _MAX_STUCK_ROUNDS, "runaway": _runaway, "message": _lb_message})}\n\n'
             # The model has been executing tools, so its results are already
             # in context. Force ONE tool-free round to converge: write the
             # answer from what it has, or state plainly what's blocking it.
@@ -3562,6 +3760,10 @@ async def stream_agent_loop(
                 cmd_display = block.content.split("\n")[0].strip()[:80]
             else:
                 cmd_display = full_command
+            # The display string is streamed (tool_start/tool_output) and persisted;
+            # redact any secrets in it. The executed block content is left untouched
+            # so tool execution still sees the real command.
+            cmd_display = _redact_sensitive_text(cmd_display)
 
             if tool_policy and tool_policy.blocks(block.tool_type):
                 desc = f"{block.tool_type}: BLOCKED"
@@ -3607,8 +3809,15 @@ async def stream_agent_loop(
                     evt = await _progress_q.get()
                     if evt is None:
                         break
+                    # Redact secrets in the live tail before streaming — the
+                    # final tool_output is redacted, so the progress tail must
+                    # be too, or a secret could flash by mid-run. Copy so we
+                    # don't mutate the tool's own event payload.
+                    _evt = dict(evt)
+                    if isinstance(_evt.get("tail"), str):
+                        _evt["tail"] = _redact_sensitive_text(_evt["tail"])
                     yield (
-                        f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **evt})}\n\n'
+                        f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **_evt})}\n\n'
                     )
                 desc, result = await _tool_task
 
@@ -3674,7 +3883,7 @@ async def stream_agent_loop(
                                 result["results"] = _clean
                             elif "stdout" in result:
                                 result["stdout"] = _clean
-                        except (json.JSONDecodeError, Exception):
+                        except Exception:
                             pass
 
             # Emit doc-specific event for document tools — the frontend
@@ -3744,29 +3953,29 @@ async def stream_agent_loop(
                 # empty) stdout/stderr; fall back to the error so the "timed
                 # out" reason reaches the UI instead of a blank result.
                 raw = result["stdout"] or result["stderr"] or result.get("error", "")
-                output_text = _truncate(raw)
+                output_text = _truncate(_redact_sensitive_text(raw))
             elif "output" in result:
                 # bash / python canonical result: {"output": ..., "exit_code": ...}
                 raw = result["output"] or ""
-                output_text = _truncate(raw)
+                output_text = _truncate(_redact_sensitive_text(raw))
             elif "response" in result:
                 # AI interaction tools (chat_with_model, send_to_session)
                 label = result.get("model", result.get("session_name", "AI"))
-                output_text = _truncate(f"{label}: {result['response']}")
+                output_text = _truncate(_redact_sensitive_text(f"{label}: {result['response']}"))
             elif "content" in result:
-                output_text = _truncate(result["content"])
+                output_text = _truncate(_redact_sensitive_text(result["content"]))
             elif "results" in result:
-                output_text = _truncate(result["results"])
+                output_text = _truncate(_redact_sensitive_text(result["results"]))
             elif "session_id" in result and "name" in result:
                 output_text = f"Session created: {result['name']} (id: {result['session_id']})"
             elif "success" in result:
                 output_text = (
                     f"Written: {result.get('path', '')}"
                     if result["success"]
-                    else f"Error: {result.get('error', '')}"
+                    else f"Error: {_redact_sensitive_text(result.get('error', ''))}"
                 )
             elif "error" in result:
-                output_text = _truncate(result["error"])
+                output_text = _truncate(_redact_sensitive_text(result["error"]))
 
             # Emit tool_output (include ui_event data if present)
             tool_output_data = {"type": "tool_output", "tool": block.tool_type, "command": cmd_display, "output": output_text, "exit_code": result.get("exit_code")}

src/rag_manager.py

@@ -32,9 +32,9 @@ class RAGManager:
         logger.info("RAGManager initialized as wrapper for VectorRAG")
     
     # Delegate all methods to VectorRAG
-    def search(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
+    def search(self, query: str, k: int = 5, owner: Optional[str] = None) -> List[Dict[str, Any]]:
         """Search for documents - delegates to VectorRAG."""
-        return self.vector_rag.search(query, k)
+        return self.vector_rag.search(query, k, owner=owner)
     
     def index_personal_documents(
         self,

src/tool_utils.py

@@ -54,6 +54,8 @@ def _parse_tool_args(content):
     if isinstance(content, str):
         try:
             args = json.loads(content) if content.strip() else {}
+            if not isinstance(args, dict):
+                args = {}
         except (json.JSONDecodeError, TypeError) as e:
             raise ValueError(str(e))
     elif isinstance(content, dict):

static/js/chat.js

@@ -2201,6 +2201,23 @@ import { wireArrowUpRecall, getLastUserMessageFromChatHistory } from './composer
                   _chatBox.appendChild(note);
                   try { note.scrollIntoView({ block: 'end', behavior: 'smooth' }); } catch (_) { uiModule.scrollHistory && uiModule.scrollHistory(); }
                 }
+              } else if (json.type === 'loop_breaker_triggered' || json.type === 'intent_nudge_exhausted') {
+                // A loop guard ended the turn — surface why so it isn't mistaken
+                // for a clean completion or a silent stall.
+                const _chatBox = document.getElementById('chat-history');
+                if (!_isBg && _chatBox) {
+                  const note = document.createElement('div');
+                  note.className = 'stopped-indicator loop-guard-stop';
+                  const label = document.createElement('span');
+                  label.className = 'rounds-exhausted-label';
+                  label.textContent = json.message ||
+                    (json.type === 'loop_breaker_triggered'
+                      ? 'Stopped by the loop-breaker (no new progress).'
+                      : 'Stopped: announced an action but never called the tool.');
+                  note.appendChild(label);
+                  _chatBox.appendChild(note);
+                  try { note.scrollIntoView({ block: 'end', behavior: 'smooth' }); } catch (_) { uiModule.scrollHistory && uiModule.scrollHistory(); }
+                }
               } else if (json.type === 'model_actual') {
                 if (!_isBg && holder) {
                   holder._requestedModel = json.requested_model || holder._requestedModel || modelName;

static/js/settings.js

@@ -542,6 +542,9 @@ async function initDefaultChat() {
     renderFallbacks();
   } catch (e) { console.warn('Failed to load default chat settings', e); }
 
+  epSel.addEventListener('change', function() { refreshModels(''); saveDefault(); });
+  modelSel.addEventListener('change', saveDefault);
+
   async function saveDefault() {
     try {
       var clean = _fallbacks.filter(function(f) { return f.endpoint_id && f.model; });
@@ -558,8 +561,6 @@ async function initDefaultChat() {
     } catch (e) { msg.textContent = 'Failed to save'; msg.style.color = 'var(--red)'; }
   }
 
-  epSel.addEventListener('change', function() { refreshModels(''); saveDefault(); });
-  modelSel.addEventListener('change', saveDefault);
   if (addFbBtn) addFbBtn.addEventListener('click', function() {
     var first = enabledEndpoints()[0];
     _fallbacks.push({ endpoint_id: first ? first.id : '', model: '' });

static/style.css

@@ -39695,3 +39695,13 @@ body.theme-frosted .modal {
 .log-line-default {
   color: var(--fg, #9cdef2);
 }
+
+/* The model-comparison grid hard-codes 2-4 equal columns with no phone
+   breakpoint that stacks them, so at 390px two models render ~178px columns and
+   four render ~88px columns. Each column is a full scrolling chat (code blocks,
+   tool output, vote footer), so the text is unreadably over-wrapped and clipped.
+   On phones, stack the panes into a single scrollable column instead. */
+@media (max-width: 768px) {
+  .compare-grid[data-cols] { grid-template-columns: 1fr !important; overflow-y: auto; }
+  .compare-pane { min-height: 60dvh; }
+}

tests/test_admin_tools_registry.py

@@ -72,3 +72,8 @@ def test_parse_tool_args_lives_in_tool_utils_single_source():
     assert _parse_tool_args('{"action":"add"}') == {"action": "add"}
     # body-envelope unwrap still works
     assert _parse_tool_args('{"body":{"action":"x"}}') == {"action": "x"}
+
+    # non-dict JSON values should return {}
+    assert _parse_tool_args('[1, 2]') == {}
+    assert _parse_tool_args('42') == {}
+    assert _parse_tool_args('"hello"') == {}

tests/test_loop_guard_signals.py

@@ -0,0 +1,350 @@
+"""Regression: stream_agent_loop surfaces *why* a guard ended the turn.
+
+Two internal guards used to stop the agent in ways that looked like a clean
+completion or a vague blocked message:
+
+  * the loop-breaker stall detector -> now emits `loop_breaker_triggered`
+  * the intent-without-action nudge cap -> now emits `intent_nudge_exhausted`
+
+These tests run the real loop body against a fake LLM stream (no model calls,
+no sleeps) and assert the structured stop event is emitted.
+"""
+
+import asyncio
+import json
+import logging
+
+import pytest
+
+import src.agent_loop as al
+
+
+def _collect(gen):
+    async def _run():
+        return [c async for c in gen]
+    return asyncio.run(_run())
+
+
+def _types(chunks):
+    out = []
+    for c in chunks:
+        if c.startswith("data: ") and not c.startswith("data: [DONE]"):
+            try:
+                out.append(json.loads(c[6:]))
+            except Exception:
+                pass
+    return out
+
+
+def _patch_common(monkeypatch):
+    monkeypatch.setattr(al, "get_setting", lambda key, default=None: default, raising=False)
+    monkeypatch.setattr(al, "get_mcp_manager", lambda: None, raising=False)
+    monkeypatch.setattr(al, "estimate_tokens", lambda *a, **k: 10, raising=False)
+
+    async def _fake_exec(block, *a, **k):
+        return ("bash", {"output": "ok", "exit_code": 0})
+    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
+
+
+def _run_loop(monkeypatch, round_text, max_rounds, relevant_tools={"bash"}):
+    async def _fake_stream(_candidates, messages, **kwargs):
+        yield f'data: {json.dumps({"delta": round_text})}\n\n'
+        yield "data: [DONE]\n\n"
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    gen = al.stream_agent_loop(
+        "http://x/v1", "m",
+        [{"role": "user", "content": "do a long multi-step task"}],
+        max_rounds=max_rounds,
+        relevant_tools=relevant_tools,
+    )
+    return _types(_collect(gen))
+
+
+def test_emits_loop_breaker_triggered_on_repeated_no_progress(monkeypatch):
+    _patch_common(monkeypatch)
+    # Same exact tool call every round, no answer text -> stuck-round streak
+    # trips the loop-breaker once the cap is reached.
+    events = _run_loop(monkeypatch, "```bash\necho hi\n```", max_rounds=8)
+    lb = [e for e in events if e.get("type") == "loop_breaker_triggered"]
+    assert lb, events
+    e = lb[0]
+    assert e["reason"]
+    assert e["max_stuck_rounds"] == 4
+    assert e["stuck_rounds"] >= 4
+    assert "message" in e
+
+
+def test_no_loop_breaker_on_normal_finish(monkeypatch):
+    _patch_common(monkeypatch)
+    events = _run_loop(monkeypatch, "All done, here is your answer.", max_rounds=8)
+    assert not any(e.get("type") == "loop_breaker_triggered" for e in events), events
+
+
+def test_emits_intent_nudge_exhausted_when_cap_reached(monkeypatch):
+    _patch_common(monkeypatch)
+    # The model keeps announcing an action with no tool call. After the nudge
+    # cap is spent, the turn ends with an explicit intent_nudge_exhausted event.
+    events = _run_loop(monkeypatch, "Let me check the logs now", max_rounds=5)
+    inx = [e for e in events if e.get("type") == "intent_nudge_exhausted"]
+    assert inx, events
+    e = inx[0]
+    assert e["max_nudges"] == 2
+    assert e["nudges"] >= 2
+    assert "message" in e
+
+
+def test_no_intent_nudge_exhausted_on_normal_finish(monkeypatch):
+    _patch_common(monkeypatch)
+    events = _run_loop(monkeypatch, "Here is the complete answer to your question.", max_rounds=5)
+    assert not any(e.get("type") == "intent_nudge_exhausted" for e in events), events
+
+
+def _assert_guard_log_safe(caplog, *, structural, secret="secret123"):
+    """The guard's own structural log line fired, and that record carries no raw
+    secret. Scoped to the guard's records on purpose: an unrelated, pre-existing
+    round-summary log echoes raw model text and is out of scope for this PR."""
+    records = [r for r in caplog.records if structural in r.getMessage()]
+    assert records, caplog.text
+    for r in records:
+        assert secret not in r.getMessage(), r.getMessage()
+
+
+def test_intent_nudge_logging_does_not_leak_secret(monkeypatch, caplog):
+    # The model announces an action (no tool call) with a secret in the text.
+    # The nudge logger must record only structural metadata, never the matched
+    # phrase — so the credential never lands in journalctl.
+    _patch_common(monkeypatch)
+    with caplog.at_level(logging.INFO, logger="src.agent_loop"):
+        events = _run_loop(monkeypatch, "Let me check api_key=secret123 now", max_rounds=5)
+    assert any(e.get("type") == "intent_nudge_exhausted" for e in events), events
+    _assert_guard_log_safe(caplog, structural="intent-without-action nudge")
+
+
+def test_loop_breaker_logging_does_not_leak_secret(monkeypatch, caplog):
+    # A repeated tool command carrying a secret trips the loop-breaker. The
+    # structural log must not contain `_sig` / raw tool-call content.
+    _patch_common(monkeypatch)
+    with caplog.at_level(logging.INFO, logger="src.agent_loop"):
+        events = _run_loop(monkeypatch, "```bash\necho api_key=secret123\n```", max_rounds=8)
+    assert any(e.get("type") == "loop_breaker_triggered" for e in events), events
+    _assert_guard_log_safe(caplog, structural="loop-breaker tripped")
+
+
+def test_redacts_sensitive_tool_output_before_surfacing():
+    text = al._redact_sensitive_text(
+        "password: private-value\n"
+        "api_key=private-key\n"
+        "Authorization: Bearer private-token\n"
+        "normal output"
+    )
+
+    assert "private-value" not in text
+    assert "private-key" not in text
+    assert "private-token" not in text
+    assert "password: [redacted]" in text
+    assert "api_key=[redacted]" in text
+    assert "Authorization: Bearer [redacted]" in text
+    assert "normal output" in text
+
+
+_GCP_API_KEY_SAMPLE = "AI" + "za" + ("A" * 35)
+
+# (input, secret substring that must be gone, expected substring that must remain)
+_REDACTION_CASES = [
+    ("Authorization: Bearer abc123tok", "abc123tok", "Authorization: Bearer [redacted]"),
+    ("Authorization: Basic dXNlcjpwYXNz", "dXNlcjpwYXNz", "Authorization: Basic [redacted]"),
+    # Quoted Authorization value (spaces) must be redacted whole.
+    ('Authorization: Bearer "two word secret"', "two word secret", "Authorization: Bearer [redacted]"),
+    # Escaped quote inside a quoted secret must not leak the tail.
+    (r'password="abc\"def secret"', "def secret", "password=[redacted]"),
+    # URL password containing a colon must still be redacted whole.
+    ("postgres://user:pa:ss@host/db", "pa:ss", "postgres://[redacted]@host/db"),
+    # Provider-shaped bare tokens.
+    ("token is hf_abcdefghij1234567890XYZ", "hf_abcdefghij1234567890XYZ", "[redacted]"),
+    ("key " + _GCP_API_KEY_SAMPLE, _GCP_API_KEY_SAMPLE, "[redacted]"),
+    ("Cookie: session=abc123secret", "abc123secret", "Cookie: [redacted]"),
+    ("Set-Cookie: sid=xyz789; HttpOnly", "xyz789", "Set-Cookie: [redacted]"),
+    ("postgres://user:pa55word@host/db", "pa55word", "postgres://[redacted]@host/db"),
+    ("client_secret=supersecretvalue", "supersecretvalue", "client_secret=[redacted]"),
+    ("OPENAI_API_KEY=abcd1234deadbeef", "abcd1234deadbeef", "OPENAI_API_KEY=[redacted]"),
+    # Quoted multi-word env value must be fully redacted, not clipped at the space.
+    ('OPENAI_API_KEY="two word secret"', "two word secret", "OPENAI_API_KEY=[redacted]"),
+    ('password: "my secret value"', "my secret value", "password: [redacted]"),
+    ("here is sk-abcdefghij1234567890", "sk-abcdefghij1234567890", "[redacted]"),
+    (
+        "-----BEGIN PRIVATE KEY-----\nMIIfakeKEYbody\n-----END PRIVATE KEY-----",
+        "MIIfakeKEYbody",
+        "[redacted private key]",
+    ),
+]
+
+
+@pytest.mark.parametrize("raw, secret, expected", _REDACTION_CASES)
+def test_redaction_covers_requested_secret_shapes(raw, secret, expected):
+    out = al._redact_sensitive_text(raw)
+    assert secret not in out, out
+    assert expected in out, out
+
+
+@pytest.mark.parametrize("raw", [
+    "the build completed in 3.2s with 0 errors",
+    "password reset email sent to the user",
+    "Listing 5 files: a.py b.py c.py d.py e.py",
+    "https://example.com/path?page=2",
+    # Benign uppercase names that merely end in KEY must not be redacted.
+    "MONKEY=banana",
+    "TURKEY=dinner",
+])
+def test_redaction_keeps_normal_output_readable(raw):
+    assert al._redact_sensitive_text(raw) == raw
+
+
+def test_redacts_before_truncating():
+    # A secret near the start must be gone even if truncation would otherwise
+    # only clip the tail — redaction runs first.
+    raw = "api_key=topsecretvalue " + ("x" * 50_000)
+    out = al._truncate(al._redact_sensitive_text(raw))
+    assert "topsecretvalue" not in out
+    assert "api_key=[redacted]" in out
+
+
+def _run_tool_result(monkeypatch, tool, exec_result, max_rounds=2):
+    """Drive one tool round whose execution returns `exec_result`, and collect
+    the streamed events. Used to assert restored per-tool-result emissions."""
+    _patch_common(monkeypatch)
+
+    async def _fake_exec(block, *a, **k):
+        return (tool, exec_result)
+    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
+
+    round_text = f"```{tool}\n{{}}\n```"
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        yield f'data: {json.dumps({"delta": round_text})}\n\n'
+        yield "data: [DONE]\n\n"
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    gen = al.stream_agent_loop(
+        "http://x/v1", "m",
+        [{"role": "user", "content": "do something"}],
+        max_rounds=max_rounds,
+        relevant_tools={tool},
+    )
+    return _types(_collect(gen))
+
+
+def test_restores_doc_suggestions_event(monkeypatch):
+    events = _run_tool_result(
+        monkeypatch, "suggest_document",
+        {"action": "suggest", "doc_id": "d1", "suggestions": [{"text": "x"}], "exit_code": 0},
+    )
+    assert any(e.get("type") == "doc_suggestions" for e in events), events
+
+
+def test_restores_doc_update_event(monkeypatch):
+    events = _run_tool_result(
+        monkeypatch, "edit_document",
+        {"action": "edit", "doc_id": "d1", "content": "body", "version": 2,
+         "title": "T", "language": "md", "exit_code": 0},
+    )
+    # A native document block also emits doc_update AFTER tool_output, so a plain
+    # "any doc_update" check would pass even if the restored generic block were
+    # gone. Prove the restored block fires BEFORE the first tool_output.
+    types = [e.get("type") for e in events]
+    assert "doc_update" in types, events
+    assert "tool_output" in types, events
+    assert types.index("doc_update") < types.index("tool_output"), types
+
+
+def test_restores_ui_control_event(monkeypatch):
+    events = _run_tool_result(
+        monkeypatch, "ui_control",
+        {"ui_event": "toggle", "toggle_name": "bash", "state": "off", "exit_code": 0},
+    )
+    assert any(e.get("type") == "ui_control" for e in events), events
+
+
+def test_restores_plan_update_event(monkeypatch):
+    events = _run_tool_result(
+        monkeypatch, "update_plan",
+        {"plan_update": {"steps": [{"text": "step", "done": True}]}, "exit_code": 0},
+    )
+    assert any(e.get("type") == "plan_update" for e in events), events
+
+
+def test_restores_ask_user_event_and_persists_question(monkeypatch):
+    events = _run_tool_result(
+        monkeypatch, "ask_user",
+        {"ask_user": {"question": "Which option?", "options": [{"label": "A"}, {"label": "B"}]},
+         "exit_code": 0},
+    )
+    # Exactly one ask_user event — not re-emitted on a follow-up round.
+    _ask_events = [e for e in events if e.get("type") == "ask_user"]
+    assert len(_ask_events) == 1, events
+    # The question is streamed as assistant text so it persists for replay.
+    # Upstream prepends "\n\n" when full_response already holds streamed text,
+    # so match on containment — and it must be streamed exactly once.
+    _q_deltas = [e for e in events if "Which option?" in (e.get("delta") or "")]
+    assert len(_q_deltas) == 1, events
+    # Setting `_awaiting_user` breaks the loop, so the turn does NOT advance into
+    # another agent round (which would emit an agent_step event) after the ask.
+    assert not any(e.get("type") == "agent_step" for e in events), events
+
+
+def test_redacts_command_display_in_streamed_events(monkeypatch):
+    # A tool command line can carry a secret. The streamed command display
+    # (tool_start / tool_output) must be redacted, even though the real command
+    # passed to execution is left untouched.
+    _patch_common(monkeypatch)
+
+    round_text = "```bash\necho api_key=secret123\n```"
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        yield f'data: {json.dumps({"delta": round_text})}\n\n'
+        yield "data: [DONE]\n\n"
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    gen = al.stream_agent_loop(
+        "http://x/v1", "m",
+        [{"role": "user", "content": "run it"}],
+        max_rounds=2,
+        relevant_tools={"bash"},
+    )
+    events = _types(_collect(gen))
+    cmds = [e for e in events if e.get("type") in ("tool_start", "tool_output")]
+    assert cmds, events
+    assert all("secret123" not in (e.get("command") or "") for e in cmds), cmds
+    assert any("api_key=[redacted]" in (e.get("command") or "") for e in cmds), cmds
+
+
+def test_redacts_live_tool_progress_tail(monkeypatch):
+    # A secret in the live progress tail must be redacted before streaming —
+    # otherwise it flashes by before the (already redacted) final tool_output.
+    _patch_common(monkeypatch)
+
+    async def _fake_exec(block, *a, **k):
+        await k["progress_cb"]({"tail": "api_key=secret123", "elapsed_s": 1})
+        return ("bash", {"output": "done", "exit_code": 0})
+    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
+
+    round_text = "```bash\necho hi\n```"
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        yield f'data: {json.dumps({"delta": round_text})}\n\n'
+        yield "data: [DONE]\n\n"
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    gen = al.stream_agent_loop(
+        "http://x/v1", "m",
+        [{"role": "user", "content": "run it"}],
+        max_rounds=2,
+        relevant_tools={"bash"},
+    )
+    events = _types(_collect(gen))
+    prog = [e for e in events if e.get("type") == "tool_progress"]
+    assert prog, events
+    assert all("secret123" not in (e.get("tail") or "") for e in prog), prog
+    assert any("api_key=[redacted]" in (e.get("tail") or "") for e in prog), prog
+    # Other fields are preserved.
+    assert any(e.get("elapsed_s") == 1 for e in prog), prog

tests/test_rag_search_signature.py

@@ -0,0 +1,22 @@
+import unittest
+from unittest.mock import MagicMock, patch
+from src.rag_manager import RAGManager
+
+class TestRAGManagerSearchSignature(unittest.TestCase):
+    @patch('src.rag_manager.VectorRAG')
+    def test_search_signature_accepts_owner(self, mock_vector_rag_class):
+        # Create a mock instance for VectorRAG
+        mock_vector_rag = MagicMock()
+        mock_vector_rag_class.return_value = mock_vector_rag
+
+        # Initialize RAGManager
+        manager = RAGManager()
+
+        # Test call with owner parameter
+        manager.search("test query", k=3, owner="user1")
+
+        # Verify that search was called on the underlying vector_rag with the correct parameters
+        mock_vector_rag.search.assert_called_once_with("test query", 3, owner="user1")
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_research_routes_path_confinement.py

@@ -0,0 +1,273 @@
+"""Path-confinement regression tests for research routes.
+
+Covers the CodeQL py/path-injection alert cluster (#552-#567) in
+routes/research/research_routes.py:
+  - _owns_in_memory disk fallback (alerts #552, #553)
+  - _assert_owns_research (alerts #554, #555)
+  - research_detail (alerts #556, #557)
+  - research_archive (alerts #558, #559, #560)
+  - research_delete (alerts #561, #562, #563)
+  - research_result_peek (alerts #564, #565)
+  - research_spinoff (alerts #566, #567)
+"""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+
+from routes.research_routes import setup_research_routes
+from routes.research.research_routes import _confine_research_path
+
+
+@pytest.fixture(autouse=True)
+def _redirect_research_dir(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        "routes.research_routes.DEEP_RESEARCH_DIR",
+        str(tmp_path / "deep_research"),
+    )
+
+
+def _request(user: str):
+    return SimpleNamespace(state=SimpleNamespace(current_user=user))
+
+
+def _route(router, path: str, method: str):
+    for route in router.routes:
+        if getattr(route, "path", "") != path:
+            continue
+        if method in getattr(route, "methods", set()):
+            return route.endpoint
+    raise AssertionError(f"{method} {path} route not registered")
+
+
+def _write_research(data_dir, session_id: str, **data):
+    data_dir.mkdir(parents=True, exist_ok=True)
+    path = data_dir / f"{session_id}.json"
+    path.write_text(json.dumps(data), encoding="utf-8")
+    return path
+
+
+def _research_handler():
+    handler = MagicMock()
+    handler._active_tasks = {}
+    return handler
+
+
+# ---------------------------------------------------------------------------
+# Helper-level tests — _confine_research_path
+# ---------------------------------------------------------------------------
+
+def test_confine_allows_valid_session_id(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir()
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+    path = _confine_research_path("rp-abc123de4567")
+    assert path == (data_dir / "rp-abc123de4567.json").resolve()
+
+
+@pytest.mark.parametrize("bad_id", [
+    "../escape",
+    "../../etc/passwd",
+    "/etc/passwd",
+    "safe/../../x",
+    "",
+    "rp_bad",          # underscore not in allowed charset
+    "rp-bad.json",     # dot not in allowed charset
+    "a" * 129,         # exceeds length limit
+])
+def test_confine_rejects_bad_session_ids(tmp_path, monkeypatch, bad_id):
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir()
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+    with pytest.raises(HTTPException) as exc:
+        _confine_research_path(bad_id)
+    assert exc.value.status_code == 400
+
+
+def test_confine_rejects_symlink_escape(tmp_path, monkeypatch):
+    """A symlink inside DEEP_RESEARCH_DIR that resolves outside is rejected."""
+    data_dir = tmp_path / "deep_research"
+    outside = tmp_path / "outside"
+    data_dir.mkdir()
+    outside.mkdir()
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+    target = outside / "rp-linktest1234.json"
+    target.write_text("{}", encoding="utf-8")
+    link = data_dir / "rp-linktest1234.json"
+    try:
+        link.symlink_to(target)
+    except (AttributeError, NotImplementedError, OSError) as e:
+        pytest.skip(f"symlinks unavailable: {e}")
+    with pytest.raises(HTTPException) as exc:
+        _confine_research_path("rp-linktest1234")
+    assert exc.value.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Route-level tests — valid paths work
+# ---------------------------------------------------------------------------
+
+def test_detail_returns_data_for_owner(tmp_path):
+    data_dir = tmp_path / "deep_research"
+    _write_research(data_dir, "rp-validid12345", owner="alice", query="valid query")
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/detail/{session_id}", "GET")
+    out = asyncio.run(target(session_id="rp-validid12345", request=_request("alice")))
+    assert out["query"] == "valid query"
+
+
+# ---------------------------------------------------------------------------
+# Route-level tests — traversal and injection rejected
+# ---------------------------------------------------------------------------
+
+_TRAVERSAL_IDS = [
+    "../escape",
+    "../../etc/passwd",
+    "/etc/passwd",
+    "safe/../../x",
+    "rp_under",
+    "a" * 129,
+]
+
+
+@pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+def test_detail_rejects_traversal(bad_id):
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/detail/{session_id}", "GET")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id=bad_id, request=_request("alice")))
+    assert exc.value.status_code == 400
+
+
+@pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+def test_archive_rejects_traversal(bad_id):
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/{session_id}/archive", "POST")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id=bad_id, request=_request("alice"), archived=True))
+    assert exc.value.status_code == 400
+
+
+@pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+def test_delete_rejects_traversal(bad_id):
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/{session_id}", "DELETE")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id=bad_id, request=_request("alice")))
+    assert exc.value.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Route-level tests — traversal does not touch files outside DEEP_RESEARCH_DIR
+# ---------------------------------------------------------------------------
+
+def test_delete_traversal_does_not_delete_outside_file(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir(parents=True)
+    outside = tmp_path / "sensitive.json"
+    outside.write_text('{"secret": true}', encoding="utf-8")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/{session_id}", "DELETE")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id="../sensitive", request=_request("alice")))
+    assert exc.value.status_code == 400
+    assert outside.exists(), "file outside DEEP_RESEARCH_DIR must not be deleted"
+
+
+def test_archive_traversal_does_not_mutate_outside_file(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir(parents=True)
+    outside = tmp_path / "sensitive.json"
+    outside.write_text('{"owner": "alice", "archived": false}', encoding="utf-8")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/{session_id}/archive", "POST")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id="../sensitive", request=_request("alice"), archived=True))
+    assert exc.value.status_code == 400
+    data = json.loads(outside.read_text(encoding="utf-8"))
+    assert data["archived"] is False, "file outside DEEP_RESEARCH_DIR must not be mutated"
+
+
+def test_detail_traversal_does_not_read_outside_file(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir(parents=True)
+    outside = tmp_path / "sensitive.json"
+    outside.write_text('{"owner": "alice", "result": "secret data"}', encoding="utf-8")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/detail/{session_id}", "GET")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id="../sensitive", request=_request("alice")))
+    assert exc.value.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Route-level symlink escape test
+# ---------------------------------------------------------------------------
+
+def test_detail_rejects_symlink_escape(tmp_path, monkeypatch):
+    """research_detail rejects a confined-format ID whose JSON is a symlink to outside."""
+    data_dir = tmp_path / "deep_research"
+    outside_dir = tmp_path / "outside"
+    data_dir.mkdir(parents=True)
+    outside_dir.mkdir()
+    outside_file = outside_dir / "rp-linktest5678.json"
+    outside_file.write_text(
+        json.dumps({"owner": "alice", "result": "secret"}), encoding="utf-8"
+    )
+    link = data_dir / "rp-linktest5678.json"
+    try:
+        link.symlink_to(outside_file)
+    except (AttributeError, NotImplementedError, OSError) as e:
+        pytest.skip(f"symlinks unavailable: {e}")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/detail/{session_id}", "GET")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id="rp-linktest5678", request=_request("alice")))
+    assert exc.value.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# Owner/session scoping cannot escape root
+# ---------------------------------------------------------------------------
+
+def test_owner_scoped_paths_stay_within_research_root(tmp_path, monkeypatch):
+    """Owner-scoped session IDs never produce paths outside DEEP_RESEARCH_DIR."""
+    data_dir = tmp_path / "deep_research"
+    data_dir.mkdir(parents=True)
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    root = data_dir.resolve()
+    for session_id in ("rp-abc123456789", "rp-000000000001", "abc-xyz-123"):
+        path = _confine_research_path(session_id)
+        assert path.resolve().is_relative_to(root), (
+            f"{session_id!r} produced path outside research root: {path}"
+        )
+
+@pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+def test_result_peek_rejects_traversal(bad_id):
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/result-peek/{session_id}", "POST")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id=bad_id, request=_request("alice")))
+    assert exc.value.status_code == 400
+
+
+@pytest.mark.parametrize("bad_id", _TRAVERSAL_IDS)
+def test_spinoff_rejects_traversal(bad_id):
+    router = setup_research_routes(_research_handler())
+    target = _route(router, "/api/research/spinoff/{session_id}", "POST")
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(target(session_id=bad_id, request=_request("alice")))
+    assert exc.value.status_code == 400

tests/test_security_regressions.py

@@ -894,7 +894,8 @@ def test_web_fetch_guard_fails_closed_on_empty_resolution(monkeypatch):
 
 def test_web_fetch_guard_blocks_redirect_into_private(monkeypatch):
     # A public URL that 302-redirects to an internal address must be blocked
-    # at the redirect hop, not followed.
+    # at the redirect hop, not followed. _get_public_url now uses
+    # httpx.Client(...).stream(...) so the test must mock that path.
     import httpx
     from src.search import content
 
@@ -905,14 +906,31 @@ def test_web_fetch_guard_blocks_redirect_into_private(monkeypatch):
         status_code = 302
         url = "http://public.example/start"
         headers = {"location": "http://169.254.169.254/latest/meta-data/"}
+        encoding = "utf-8"
 
-    from contextlib import contextmanager
+    class _FakeStream:
+        def __enter__(self):
+            return _Resp()
 
-    @contextmanager
-    def _fake_stream(method, url, **kwargs):
-        yield _Resp()
+        def __exit__(self, *args):
+            return False
 
-    monkeypatch.setattr(httpx, "stream", _fake_stream)
+    class _FakeClient:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+        def stream(self, method, url):
+            assert method == "GET"
+            assert url == "http://public.example/start"
+            return _FakeStream()
+
+    monkeypatch.setattr(httpx, "Client", _FakeClient)
 
     with _pytest.raises(httpx.RequestError) as exc:
         content._get_public_url("http://public.example/start", headers={}, timeout=5)
@@ -1224,3 +1242,274 @@ def test_visual_report_escapes_request_category():
     # value must coerce rather than crash the render (html.escape needs a str).
     out = generate_visual_report(question="q", report_markdown="## H", category=12345)
     assert "category-12345" in out
+
+
+# ── DNS rebinding (audit finding 8.1) ────────────────────────────────
+# _resolve_public_ips resolves a URL's hostname once per hop and rejects
+# private / metadata targets, but httpx would then re-resolve the
+# hostname at connect time. The fix: the actual TCP connect is pinned
+# to the resolved IP via a custom httpcore.NetworkBackend, while the
+# URL / Host header / SNI stay on the original hostname.
+
+import ipaddress as _ipaddr
+import socket as _socket
+import threading as _threading
+
+import httpx as _httpx
+
+
+def test_dns_rebinding_blocked_by_resolve_gate(monkeypatch):
+    from src.search import content
+
+    monkeypatch.setattr(content, "_resolve_hostname_ips",
+                        lambda host: [_ipaddr.ip_address("10.0.0.5")])
+
+    with _pytest.raises(_httpx.RequestError) as exc:
+        content._resolve_public_ips("https://attacker.example/")
+    assert "non-public" in str(exc.value).lower()
+
+
+def test_dns_rebinding_pinned_backend_connects_to_resolved_ip(monkeypatch):
+    """``_PinnedBackend.connect_tcp`` must ignore the URL's host and
+    dial the pinned IP at the original port. This is the core of the
+    fix: httpcore's NetworkBackend contract lets us intercept the
+    connect before DNS lookup happens.
+    """
+    from src.search import content
+
+    pinned_ip = _ipaddr.ip_address("93.184.216.34")
+    captured = {}
+
+    class _StubStream:
+        def close(self):
+            pass
+
+    class _StubBackend:
+        def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
+            captured["host"] = host
+            captured["port"] = port
+            return _StubStream()
+
+        def connect_unix_socket(self, path, timeout=None, socket_options=None):
+            raise OSError("not used")
+
+        def sleep(self, seconds):
+            pass
+
+    backend = content._PinnedBackend(pinned_ip)
+    monkeypatch.setattr(backend, "_real", _StubBackend())
+
+    backend.connect_tcp("attacker.example", 443)
+
+    assert captured["host"] == "93.184.216.34", captured
+    assert captured["port"] == 443, captured
+
+
+def test_dns_rebinding_pinned_transport_dials_pinned_ip(monkeypatch):
+    """End-to-end: ``_PinnedTransport`` actually dials the pinned IP
+    when given a hostname, with the original URL's Host header
+    preserved. We stand up a local socket server on a free port and
+    make the transport connect there via the pinned backend.
+    """
+    from src.search import content
+    import httpcore
+
+    # Stand up a TCP server that accepts one connection and records
+    # the request bytes it received, then returns a minimal HTTP/1.1
+    # response.
+    captured = {"request": b""}
+    server_sock = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
+    server_sock.bind(("127.0.0.1", 0))
+    server_sock.listen(1)
+    port = server_sock.getsockname()[1]
+
+    def serve_once():
+        conn, _ = server_sock.accept()
+        with conn:
+            conn.settimeout(2.0)
+            buf = b""
+            try:
+                while b"\r\n\r\n" not in buf:
+                    chunk = conn.recv(4096)
+                    if not chunk:
+                        break
+                    buf += chunk
+            except _socket.timeout:
+                pass
+            captured["request"] = buf
+            conn.sendall(
+                b"HTTP/1.1 200 OK\r\n"
+                b"Content-Length: 2\r\n"
+                b"Connection: close\r\n"
+                b"\r\n"
+                b"OK"
+            )
+
+    t = _threading.Thread(target=serve_once, daemon=True)
+    t.start()
+
+    # Pin the transport to 127.0.0.1:<port>. The caller hands it a URL
+    # with a fake hostname so we can verify the host header is sent
+    # while the TCP connect goes to the pinned IP.
+    pinned_ip = _ipaddr.ip_address("127.0.0.1")
+    transport = content._PinnedTransport(pinned_ip)
+
+    req = _httpx.Request(
+        "GET",
+        f"http://attacker.test:{port}/path?q=1",
+        headers={"host": "attacker.test"},
+    )
+    try:
+        with _httpx.Client(transport=transport, timeout=5) as client:
+            response = client.send(req)
+        assert response.status_code == 200, response.text
+    finally:
+        server_sock.close()
+
+    t.join(timeout=2)
+
+    request_bytes = captured["request"]
+    assert request_bytes, "server never received a request"
+    # Host header is the original hostname, not the IP. (httpx
+    # lowercases header names; compare case-insensitively.)
+    headers_blob = request_bytes.lower()
+    assert b"host: attacker.test" in headers_blob, request_bytes
+    # The path was preserved.
+    assert b"/path?q=1" in request_bytes, request_bytes
+
+
+def test_dns_rebinding_pinned_transport_preserves_url_netloc(monkeypatch):
+    """The URL the transport hands to the underlying httpcore layer
+    must still be the original ``https://example.com/...`` — never
+    rewritten to the pinned IP. SNI / vhost depend on this.
+    """
+    from src.search import content
+
+    seen_url = {}
+
+    class _RecordingPool:
+        def handle_request(self, req):
+            seen_url["host"] = req.url.host.decode() if isinstance(req.url.host, bytes) else req.url.host
+            seen_url["scheme"] = req.url.scheme.decode() if isinstance(req.url.scheme, bytes) else req.url.scheme
+            seen_url["target"] = req.url.target.decode() if isinstance(req.url.target, bytes) else req.url.target
+            raise _httpx.ConnectError("intercepted")
+
+        def close(self):
+            pass
+
+    pinned_ip = _ipaddr.ip_address("93.184.216.34")
+    transport = content._PinnedTransport(pinned_ip)
+    transport._pool = _RecordingPool()
+
+    req = _httpx.Request("GET", "https://example.com/some/path?q=1")
+    with _pytest.raises(_httpx.ConnectError):
+        transport.handle_request(req)
+
+    assert seen_url["host"] == "example.com", seen_url
+    assert seen_url["scheme"] == "https", seen_url
+    assert seen_url["target"] == "/some/path?q=1", seen_url
+
+
+def test_dns_rebinding_redirect_re_resolves_per_hop(monkeypatch):
+    """Every redirect hop must call ``_resolve_public_ips`` again.
+    A redirect to a private-IP target must be blocked even when the
+    first hop was public.
+    """
+    from src.search import content
+
+    seen = []
+
+    def fake_resolve(url):
+        seen.append(url)
+        if "private" in url:
+            raise _httpx.RequestError(f"Blocked non-public URL: {url}")
+        return [_ipaddr.ip_address("93.184.216.34")]
+
+    monkeypatch.setattr(content, "_resolve_public_ips", fake_resolve)
+
+    class _Resp:
+        status_code = 302
+        headers = {"location": "http://private.example/secret"}
+        encoding = "utf-8"
+
+        def __init__(self, url):
+            self.url = url
+
+    class _FakeStream:
+        def __init__(self, response):
+            self.response = response
+
+        def __enter__(self):
+            return self.response
+
+        def __exit__(self, *args):
+            return False
+
+    class _FakeClient:
+        def __init__(self, *a, **k):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *a):
+            return False
+
+        def stream(self, method, url):
+            assert method == "GET"
+            return _FakeStream(_Resp(url))
+
+    monkeypatch.setattr(_httpx, "Client", _FakeClient)
+
+    with _pytest.raises(_httpx.RequestError) as exc:
+        content._get_public_url("http://public.example/start", headers={}, timeout=5)
+    assert "non-public" in str(exc.value).lower()
+    # Both hops were validated.
+    assert seen == ["http://public.example/start", "http://private.example/secret"], seen
+
+
+def test_dns_rebinding_transport_uses_public_apis(monkeypatch):
+    """Static guard: ``_PinnedTransport`` must use only the public
+    ``httpx.BaseTransport`` / ``httpcore`` APIs. No subclassing of
+    ``httpx.HTTPTransport`` (whose ``_pool`` slot we'd have to
+    overwrite), no reads of private ``httpcore.ConnectionPool``
+    attributes, and no imports from ``httpx._transports``.
+    """
+    from src.search import content
+
+    import inspect
+
+    # 1) Subclass check: must be BaseTransport, not HTTPTransport.
+    mro_names = [c.__name__ for c in content._PinnedTransport.__mro__]
+    assert "BaseTransport" in mro_names, mro_names
+    assert "HTTPTransport" not in mro_names, (
+        "_PinnedTransport subclasses httpx.HTTPTransport. Subclass "
+        "httpx.BaseTransport instead and build the pool from scratch "
+        "with the public httpcore.ConnectionPool API."
+    )
+
+    # 2) No reads of private httpcore.ConnectionPool attrs.
+    src = inspect.getsource(content._PinnedTransport)
+    forbidden = (
+        "_ssl_context",
+        "_max_connections",
+        "_max_keepalive_connections",
+        "_keepalive_expiry",
+        "_http1",
+        "_http2",
+        "_network_backend",
+    )
+    leaked = [name for name in forbidden if name in src]
+    assert not leaked, (
+        f"_PinnedTransport reads private httpcore.ConnectionPool attrs: {leaked}. "
+        "Build the pool from the public httpcore.ConnectionPool API instead."
+    )
+
+    # 3) No imports from httpx's private transport module.
+    module_src = inspect.getsource(content)
+    forbidden_imports = ("from httpx._transports", "import httpx._transports")
+    leaked_imports = [s for s in forbidden_imports if s in module_src]
+    assert not leaked_imports, (
+        f"content.py imports from httpx's private transport module: {leaked_imports}. "
+        "Use only the public httpx and httpcore APIs."
+    )

tests/test_web_fetch_size_caps.py

@@ -13,6 +13,57 @@ import pytest
 from src.constants import WEB_FETCH_SOFT_MAX_BYTES, WEB_FETCH_HARD_MAX_BYTES
 from services.search import content as content_mod
 
+import pytest as _pytest_for_client_stream_compat
+
+
+@_pytest_for_client_stream_compat.fixture(autouse=True)
+def _client_stream_compat_for_pinned_fetch(monkeypatch):
+    """Adapt old size-cap tests to the current pinned Client.stream path.
+
+    These tests monkeypatch httpx.stream(...) to return fake responses. The
+    production fetcher now uses httpx.Client(...).stream(...) so it can pass a
+    pinned transport. When a test has replaced httpx.stream, route Client.stream
+    through that fake. When it has not, fall back to a real Client so unrelated
+    behavior in this file is not changed.
+    """
+    import httpx
+
+    real_client_cls = httpx.Client
+    original_stream = httpx.stream
+
+    class _ClientProxy:
+        def __init__(self, *args, **kwargs):
+            self._args = args
+            self._kwargs = kwargs
+            self._real_cm = None
+            self._real_client = None
+
+        def __enter__(self):
+            if httpx.stream is original_stream:
+                self._real_cm = real_client_cls(*self._args, **self._kwargs)
+                self._real_client = self._real_cm.__enter__()
+                return self._real_client
+            return self
+
+        def __exit__(self, *args):
+            if self._real_cm is not None:
+                return self._real_cm.__exit__(*args)
+            return False
+
+        def stream(self, method, url):
+            if self._real_client is not None:
+                return self._real_client.stream(method, url)
+
+            kwargs = {
+                "headers": self._kwargs.get("headers"),
+                "timeout": self._kwargs.get("timeout"),
+                "follow_redirects": self._kwargs.get("follow_redirects"),
+            }
+            return httpx.stream(method, url, **kwargs)
+
+    monkeypatch.setattr(httpx, "Client", _ClientProxy)
+
+
 
 class _FakeStream:
     """Stands in for the httpx.stream(...) context manager."""