ci: add focused test guidance signal (#4982)

* ci: add focused test guidance signal

* ci: diff focused guidance from merge base

.github/scripts/focused_test_guidance.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Report focused pytest guidance for changed paths under tests/."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import shlex
+import subprocess
+import sys
+from collections.abc import Iterable
+from pathlib import PurePosixPath
+
+
+def parse_paths(raw_paths: bytes) -> list[str]:
+    """Decode the NUL-delimited output of ``git diff --name-only -z``."""
+    return [os.fsdecode(path) for path in raw_paths.split(b"\0") if path]
+
+
+def changed_paths_from_merge_base(base_sha: str, head_sha: str) -> list[str]:
+    """Return changed ``tests/`` paths using GitHub PR three-dot semantics.
+
+    GitHub PR changed files are based on the merge base and the PR head, not a
+    direct endpoint diff between the current base branch tip and the PR head.
+    Using the direct endpoint diff can include files changed only on the base
+    branch when the PR branch is stale.
+    """
+    merge_base = subprocess.check_output(
+        ["git", "merge-base", base_sha, head_sha],
+        stderr=subprocess.DEVNULL,
+    ).strip()
+    raw_paths = subprocess.check_output(
+        [
+            "git",
+            "diff",
+            "--name-only",
+            "--diff-filter=ACMRT",
+            "-z",
+            os.fsdecode(merge_base),
+            head_sha,
+            "--",
+            "tests/",
+        ],
+    )
+    return parse_paths(raw_paths)
+
+
+def select_test_paths(paths: Iterable[str]) -> list[str]:
+    """Return unique, repository-relative paths contained by tests/."""
+    selected: set[str] = set()
+    for raw_path in paths:
+        path = PurePosixPath(raw_path)
+        if path.is_absolute() or ".." in path.parts:
+            continue
+        parts = tuple(part for part in path.parts if part != ".")
+        if len(parts) >= 2 and parts[0] == "tests":
+            selected.add(PurePosixPath(*parts).as_posix())
+    return sorted(selected)
+
+
+def is_pytest_file(path: str) -> bool:
+    """Return whether a changed path follows this repository's pytest naming."""
+    name = PurePosixPath(path).name
+    return name.endswith(".py") and (
+        name.startswith("test_") or name.endswith("_test.py")
+    )
+
+
+def pytest_command(paths: Iterable[str]) -> str:
+    """Build a copyable pytest command for changed runnable test files."""
+    command = ["python3", "-m", "pytest", "-q", *paths]
+    return shlex.join(command)
+
+
+def format_report(paths: Iterable[str]) -> str:
+    """Format focused guidance for CI logs and the workflow summary."""
+    changed_paths = select_test_paths(paths)
+    runnable_paths = [path for path in changed_paths if is_pytest_file(path)]
+    lines = ["## Focused test guidance (report-only)", ""]
+    if not changed_paths:
+        lines.append("No changed paths under `tests/`.")
+    else:
+        lines.extend(["Changed paths under `tests/`:", ""])
+        lines.extend(f"- `{path}`" for path in changed_paths)
+    lines.extend(["", "Suggested focused validation:", ""])
+    if runnable_paths:
+        lines.append(f"```sh\n{pytest_command(runnable_paths)}\n```")
+    else:
+        lines.append("No directly runnable pytest files changed.")
+    lines.extend(
+        [
+            "",
+            "This guidance does not infer tests from source changes. "
+            "Existing blocking CI remains the source of truth.",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def _parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Report focused pytest guidance for changed tests/ paths.",
+    )
+    parser.add_argument("--base-sha", help="Pull request base commit SHA.")
+    parser.add_argument("--head-sha", help="Pull request head commit SHA.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(sys.argv[1:] if argv is None else argv)
+    if bool(args.base_sha) != bool(args.head_sha):
+        raise SystemExit("--base-sha and --head-sha must be provided together")
+
+    if args.base_sha and args.head_sha:
+        paths = changed_paths_from_merge_base(args.base_sha, args.head_sha)
+    else:
+        paths = parse_paths(sys.stdin.buffer.read())
+
+    print(format_report(paths))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/workflows/ci.yml

@@ -15,6 +15,60 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  focused-test-guidance:
+    name: Focused test guidance (report-only)
+    if: github.event_name == 'pull_request'
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+      - name: Report changed test paths
+        env:
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: |
+          report_file="$RUNNER_TEMP/focused-test-guidance.md"
+          publish_report() {
+            cat "$report_file"
+            if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then
+              cat "$report_file" >> "$GITHUB_STEP_SUMMARY" || true
+            fi
+            return 0
+          }
+
+          report_unavailable() {
+            {
+              printf '%s\n\n' '## Focused test guidance unavailable (report-only)'
+              printf '%s\n\n' "$1"
+              printf '%s\n' 'Existing blocking CI remains the source of truth.'
+            } > "$report_file"
+            publish_report
+            exit 0
+          }
+
+          if [ -z "$BASE_SHA" ] || [ -z "$HEAD_SHA" ]; then
+            report_unavailable "Pull request base/head metadata is missing."
+          fi
+
+          if ! git cat-file -e "${BASE_SHA}^{commit}" 2>/dev/null; then
+            report_unavailable "The pull request base commit is unavailable locally."
+          fi
+
+          if ! git cat-file -e "${HEAD_SHA}^{commit}" 2>/dev/null; then
+            report_unavailable "The pull request head commit is unavailable locally."
+          fi
+
+          if ! python3 .github/scripts/focused_test_guidance.py \
+            --base-sha "$BASE_SHA" \
+            --head-sha "$HEAD_SHA" > "$report_file"; then
+            report_unavailable "The focused test guidance helper could not produce a report."
+          fi
+
+          publish_report
+
   python-syntax:
     name: Python syntax (compileall)
     runs-on: ubuntu-latest

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/email_helpers.py

@@ -1273,10 +1273,15 @@ def _imap_move(uid, dest, src="INBOX", account_id: str | None = None, owner: str
     try:
         c = _imap_connect(account_id, owner=owner)
         c.select(_q(src))
-        status, _ = c.copy(uid, _q(dest))
+        # Callers pass a real IMAP UID (from conn.uid("SEARCH", ...)). copy()
+        # and store() operate on message SEQUENCE NUMBERS, so addressing them
+        # with a UID moved/deleted the wrong message (or silently no-oped when
+        # the UID exceeded the message count). Use the UID commands, matching
+        # the move/delete path in email_routes.py.
+        status, _ = c.uid("COPY", uid, _q(dest))
         if status != "OK":
             return False
-        c.store(uid, "+FLAGS", "\\Deleted")
+        c.uid("STORE", uid, "+FLAGS", "\\Deleted")
         c.expunge()
         return True
     except Exception as e:

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_tools/filesystem_tools.py

@@ -281,7 +281,13 @@ class LsTool:
 
 class GlobTool:
     async def execute(self, content: str, ctx: dict) -> dict:
-        from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate
+        from src.tool_execution import (
+            _SENSITIVE_BASENAMES,
+            _is_sensitive_path,
+            _resolve_tool_path,
+            _resolve_search_root,
+            _truncate,
+        )
         args = {}
         _s = (content or "").strip()
         if _s.startswith("{"):
@@ -322,7 +328,11 @@ class GlobTool:
                     ) == nbase
                 except ValueError:
                     inside = False
-                if inside and os.path.exists(cand):
+                # A literal that names a deny-listed sensitive file (.env,
+                # .ssh/id_rsa, …) falls through to the walk, which skips it —
+                # otherwise glob would surface secret paths that read_file /
+                # grep already refuse to touch.
+                if inside and os.path.exists(cand) and not _is_sensitive_path(cand):
                     return [cand], None
                 # Literal not at exact path — fall through to walk so
                 # e.g. "foo.py" still matches at any depth (like rglob).
@@ -334,11 +344,20 @@ class GlobTool:
                 for dp, dns, fns in os.walk(base):
                     # Prune skipped dirs before descending (unlike rglob which
                     # descends first then filters — fatal on large node_modules).
-                    dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS]
+                    # Sensitive dirs (.ssh, .gnupg, …) are pruned too so glob
+                    # never enumerates the keys/tokens inside them.
+                    dns[:] = [
+                        d for d in dns
+                        if d not in _CODENAV_SKIP_DIRS and d not in _SENSITIVE_BASENAMES
+                    ]
                     for name in fns + dns:
                         full = os.path.join(dp, name)
                         rel = os.path.relpath(full, base).replace(os.sep, "/")
                         if regex.fullmatch(rel) or regex.fullmatch(name):
+                            # Skip deny-listed sensitive files (.env, id_rsa,
+                            # known_hosts, …) the same way grep does.
+                            if _is_sensitive_path(os.path.realpath(full)):
+                                continue
                             try:
                                 mtime = os.stat(full).st_mtime
                             except OSError:

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/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/run_focus.py

@@ -50,6 +50,14 @@ AREAS: tuple[str, ...] = (
 # Backward-compatible aggregate selectors for focused runs whose original
 # monolithic files were split into more specific taxonomy sub-areas.
 SUB_AREA_ALIASES: dict[str, tuple[str, ...]] = {
+    "service_health": (
+        "service_health_chromadb",
+        "service_health_search",
+        "service_health_ntfy",
+        "service_health_email",
+        "service_health_providers",
+        "service_health_collect",
+    ),
     "embedding": ("embedding", "embedding_memory"),
 }
 
@@ -214,6 +222,7 @@ def build_parser(
     """Build the argument parser for the focused runner."""
     if valid_sub_areas is None:
         valid_sub_areas = discover_sub_areas()
+    valid_sub_areas = frozenset(valid_sub_areas) | frozenset(SUB_AREA_ALIASES)
     parser = argparse.ArgumentParser(
         prog="run_focus.py",
         description=(

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_focused_test_guidance.py

@@ -0,0 +1,148 @@
+"""Tests for the report-only changed-test guidance helper."""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+SCRIPT_PATH = ROOT / ".github" / "scripts" / "focused_test_guidance.py"
+
+
+def _load_helper():
+    spec = importlib.util.spec_from_file_location("focused_test_guidance", SCRIPT_PATH)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+guidance = _load_helper()
+
+
+def test_parse_paths_supports_nul_delimited_git_output():
+    raw_paths = b"tests/test_alpha.py\0tests/path with spaces/test_beta.py\0"
+
+    assert guidance.parse_paths(raw_paths) == [
+        "tests/test_alpha.py",
+        "tests/path with spaces/test_beta.py",
+    ]
+
+
+def test_select_test_paths_ignores_paths_outside_tests():
+    paths = [
+        "src/test_alpha.py",
+        "tests/test_beta.py",
+        "./tests/unit/example_test.py",
+        "tests/../src/test_gamma.py",
+    ]
+
+    assert guidance.select_test_paths(paths) == [
+        "tests/test_beta.py",
+        "tests/unit/example_test.py",
+    ]
+
+
+def test_select_test_paths_deduplicates_paths():
+    paths = ["tests/test_alpha.py", "./tests/test_alpha.py"]
+
+    assert guidance.select_test_paths(paths) == ["tests/test_alpha.py"]
+
+
+def test_format_report_builds_command_for_changed_python_test():
+    report = guidance.format_report(["tests/test_beta.py"])
+
+    assert "- `tests/test_beta.py`" in report
+    assert "python3 -m pytest -q tests/test_beta.py" in report
+    assert "does not infer tests from source changes" in report
+    assert "Existing blocking CI remains the source of truth" in report
+
+
+def test_format_report_lists_changed_non_python_test_path_without_command():
+    report = guidance.format_report(["tests/README.md"])
+
+    assert "- `tests/README.md`" in report
+    assert "No directly runnable pytest files changed." in report
+    assert "python3 -m pytest" not in report
+
+
+def test_format_report_ignores_src_path():
+    report = guidance.format_report(["src/test_ignored.py"])
+
+    assert "src/test_ignored.py" not in report
+    assert "No changed paths under `tests/`." in report
+
+
+def test_format_report_shell_quotes_path_with_spaces():
+    report = guidance.format_report(["tests/path with spaces/test_beta.py"])
+
+    assert "- `tests/path with spaces/test_beta.py`" in report
+    assert (
+        "python3 -m pytest -q 'tests/path with spaces/test_beta.py'"
+    ) in report
+
+
+def test_format_report_handles_no_changed_test_paths():
+    report = guidance.format_report([])
+
+    assert "No changed paths under `tests/`." in report
+    assert "No directly runnable pytest files changed." in report
+
+def _git(repo: Path, *args: str) -> str:
+    return subprocess.check_output(["git", *args], cwd=repo, text=True).strip()
+
+
+def _write(path: Path, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+
+
+def test_changed_paths_from_merge_base_excludes_base_only_test_changes(tmp_path, monkeypatch):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+
+    _git(repo, "init")
+    _git(repo, "config", "user.email", "ci@example.test")
+    _git(repo, "config", "user.name", "CI Test")
+
+    _write(repo / "tests/test_shared.py", "def test_shared():\n    assert True\n")
+    _git(repo, "add", "tests/test_shared.py")
+    _git(repo, "commit", "-m", "base")
+    ancestor = _git(repo, "rev-parse", "HEAD")
+
+    _git(repo, "checkout", "-b", "feature")
+    _write(repo / "tests/test_pr_delta.py", "def test_pr_delta():\n    assert True\n")
+    _git(repo, "add", "tests/test_pr_delta.py")
+    _git(repo, "commit", "-m", "add pr test")
+    head_sha = _git(repo, "rev-parse", "HEAD")
+
+    _git(repo, "checkout", "-b", "dev", ancestor)
+    _write(repo / "tests/test_shared.py", "def test_shared():\n    assert 1 == 1\n")
+    _git(repo, "add", "tests/test_shared.py")
+    _git(repo, "commit", "-m", "base-only test change")
+    base_sha = _git(repo, "rev-parse", "HEAD")
+
+    endpoint_paths = guidance.parse_paths(
+        subprocess.check_output(
+            [
+                "git",
+                "diff",
+                "--name-only",
+                "--diff-filter=ACMRT",
+                "-z",
+                base_sha,
+                head_sha,
+                "--",
+                "tests/",
+            ],
+            cwd=repo,
+        )
+    )
+    assert "tests/test_shared.py" in endpoint_paths
+
+    monkeypatch.chdir(repo)
+    assert guidance.changed_paths_from_merge_base(base_sha, head_sha) == [
+        "tests/test_pr_delta.py"
+    ]

tests/test_imap_move_uid.py

@@ -0,0 +1,56 @@
+"""_imap_move must address messages by UID, not sequence number.
+
+The auto-spam poller passes a real IMAP UID (from conn.uid("SEARCH", ...))
+to _imap_move, but the function used conn.copy()/conn.store(), which operate
+on message SEQUENCE NUMBERS. So a UID like 90521 was interpreted as sequence
+number 90521 — moving/deleting the wrong message or silently no-oping. It
+must use the UID commands.
+"""
+import sys
+import types
+
+import pytest
+
+
+@pytest.fixture
+def email_helpers(monkeypatch, tmp_path):
+    # Keep _init_scheduled_db (run at import) off the real data dir.
+    monkeypatch.setenv("ODYSSEUS_DATA_DIR", str(tmp_path))
+    import routes.email_helpers as eh
+    return eh
+
+
+class _FakeIMAP:
+    def __init__(self):
+        self.calls = []
+
+    def select(self, mbox):
+        self.calls.append(("select", mbox)); return ("OK", [b""])
+
+    def copy(self, *a):
+        self.calls.append(("copy",) + a); return ("OK", [b""])
+
+    def store(self, *a):
+        self.calls.append(("store",) + a); return ("OK", [b""])
+
+    def uid(self, *a):
+        self.calls.append(("uid",) + a); return ("OK", [b""])
+
+    def expunge(self):
+        self.calls.append(("expunge",)); return ("OK", [b""])
+
+    def logout(self):
+        pass
+
+
+def test_move_uses_uid_commands_not_seqnum(email_helpers, monkeypatch):
+    fake = _FakeIMAP()
+    monkeypatch.setattr(email_helpers, "_imap_connect", lambda *a, **k: fake)
+    ok = email_helpers._imap_move(b"90521", "Spam", src="INBOX")
+    assert ok is True
+    verbs = [c[0] for c in fake.calls]
+    uid_ops = [c[1] for c in fake.calls if c[0] == "uid"]
+    assert "COPY" in uid_ops and "STORE" in uid_ops
+    # the sequence-number commands must NOT be used to address a UID
+    assert "copy" not in verbs
+    assert "store" not in verbs

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_run_focus.py

@@ -448,3 +448,45 @@ def test_fast_lane_collects_only_unmarked_auth_concurrency_test():
     assert _FAST_AUTH_CONCURRENCY_TEST in collected
     for slow_test in _SLOW_AUTH_CONCURRENCY_TESTS:
         assert slow_test not in collected, f"slow test was not deselected: {slow_test}"
+
+def test_service_health_sub_area_command_includes_split_files():
+    assert _cmd(sub_area="service_health") == [
+        PY,
+        "-m",
+        "pytest",
+        "-m",
+        (
+            "(sub_service_health_chromadb or "
+            "sub_service_health_search or "
+            "sub_service_health_ntfy or "
+            "sub_service_health_email or "
+            "sub_service_health_providers or "
+            "sub_service_health_collect)"
+        ),
+    ]
+
+
+def test_service_health_alias_is_accepted_by_run():
+    seen = []
+
+    def executor(cmd):
+        seen.append(cmd)
+        return 0
+
+    result = run(["--sub-area", "service_health"], executor=executor)
+
+    assert result == 0
+    assert len(seen) == 1
+    assert seen[0][1:] == [
+        "-m",
+        "pytest",
+        "-m",
+        (
+            "(sub_service_health_chromadb or "
+            "sub_service_health_search or "
+            "sub_service_health_ntfy or "
+            "sub_service_health_email or "
+            "sub_service_health_providers or "
+            "sub_service_health_collect)"
+        ),
+    ]

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_service_health.py

@@ -1,472 +0,0 @@
-"""Tests for src.service_health — the consolidated degraded-state report.
-
-Imports the real module (conftest.py stubs the heavy deps). Network is never
-touched: HTTP probes take an injected `http_get`, and the email/provider probes
-take an injected `connect` / `probe`. Asserts the ok/degraded/down/disabled
-mapping per subsystem, the overall rollup, and that no secrets leak into meta.
-"""
-import types
-
-import pytest
-
-from src import service_health as sh
-
-
-def _resp(status_code):
-    return types.SimpleNamespace(status_code=status_code)
-
-
-def _raise(*_a, **_k):
-    raise RuntimeError("connection refused")
-
-
-# ── chromadb_health ──
-
-class _Store:
-    def __init__(self, healthy):
-        self.healthy = healthy
-
-
-def test_chromadb_both_healthy_ok():
-    s = sh.chromadb_health(_Store(True), _Store(True))
-    assert s["status"] == sh.OK
-    assert s["meta"] == {"rag": True, "memory": True}
-
-
-def test_chromadb_one_down_degraded():
-    s = sh.chromadb_health(_Store(True), _Store(False))
-    assert s["status"] == sh.DEGRADED
-
-
-def test_chromadb_both_unhealthy_down():
-    s = sh.chromadb_health(_Store(False), _Store(False))
-    assert s["status"] == sh.DOWN
-
-
-def test_chromadb_both_absent_disabled():
-    s = sh.chromadb_health(None, None)
-    assert s["status"] == sh.DISABLED
-
-
-def test_chromadb_one_absent_one_healthy_ok():
-    # An absent store is not a failure; the present one being healthy is ok.
-    s = sh.chromadb_health(_Store(True), None)
-    assert s["status"] == sh.OK
-    assert s["meta"]["memory"] is None
-
-
-# ── searxng_health ──
-
-def test_searxng_disabled_when_other_provider():
-    s = sh.searxng_health({"search_provider": "brave"})
-    assert s["status"] == sh.DISABLED
-
-
-def test_searxng_ok_on_healthz():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=lambda url, timeout: _resp(200),
-    )
-    assert s["status"] == sh.OK
-    assert s["meta"]["probed"] == "/healthz"
-
-
-def test_searxng_ok_on_root_fallback():
-    def getter(url, timeout):
-        return _resp(404) if url.endswith("/healthz") else _resp(200)
-
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=getter,
-    )
-    assert s["status"] == sh.OK
-    assert s["meta"]["probed"] == "/"
-
-
-def test_searxng_down_on_exception():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=_raise,
-    )
-    assert s["status"] == sh.DOWN
-
-
-def test_searxng_down_on_5xx():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=lambda url, timeout: _resp(502),
-    )
-    assert s["status"] == sh.DOWN
-
-
-# ── ntfy_health ──
-
-def _ntfy_intg():
-    return [{"preset": "ntfy", "enabled": True, "base_url": "http://ntfy:80"}]
-
-
-def test_ntfy_disabled_without_integration():
-    s = sh.ntfy_health([], {"reminder_channel": "ntfy"})
-    assert s["status"] == sh.DISABLED
-
-
-def test_ntfy_ok():
-    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
-                       http_get=lambda url, timeout: _resp(200))
-    assert s["status"] == sh.OK
-    assert s["meta"]["base"] == "http://ntfy:80"
-
-
-def test_ntfy_probes_v1_health_not_a_topic():
-    seen = {}
-
-    def getter(url, timeout):
-        seen["url"] = url
-        return _resp(200)
-
-    sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"}, http_get=getter)
-    # Non-intrusive: hits /v1/health, never publishes to a topic.
-    assert seen["url"].endswith("/v1/health")
-
-
-def test_ntfy_down_on_exception():
-    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
-                       http_get=_raise)
-    assert s["status"] == sh.DOWN
-
-
-# ── email_health ──
-
-def _acct(name, host="imap.example.com"):
-    return {"account_id": name, "account_name": name, "imap_host": host,
-            "imap_password": "hunter2"}
-
-
-class _Conn:
-    def logout(self):
-        pass
-
-
-def test_email_disabled_without_accounts():
-    assert sh.email_health([])["status"] == sh.DISABLED
-
-
-def test_email_ok_all_connect():
-    s = sh.email_health([_acct("a"), _acct("b")], connect=lambda _id: _Conn())
-    assert s["status"] == sh.OK
-
-
-def test_email_degraded_some_fail():
-    def connect(account_id):
-        if account_id == "bad":
-            raise RuntimeError("auth failed")
-        return _Conn()
-
-    s = sh.email_health([_acct("good"), _acct("bad")], connect=connect)
-    assert s["status"] == sh.DEGRADED
-
-
-def test_email_down_all_fail():
-    s = sh.email_health([_acct("a")], connect=_raise)
-    assert s["status"] == sh.DOWN
-
-
-def test_email_account_without_host_marked_failed():
-    s = sh.email_health([_acct("a", host="")], connect=lambda _id: _Conn())
-    assert s["status"] == sh.DOWN
-
-
-def test_email_meta_never_leaks_password():
-    s = sh.email_health([_acct("a")], connect=lambda _id: _Conn())
-    assert "hunter2" not in repr(s)
-
-
-# ── providers_health ──
-
-def _ep(name):
-    return {"name": name, "base_url": f"http://{name}:8000/v1", "api_key": "sk-secret"}
-
-
-def test_providers_disabled_without_endpoints():
-    assert sh.providers_health([])["status"] == sh.DISABLED
-
-
-def test_providers_ok_all_reachable():
-    s = sh.providers_health([_ep("a")],
-                            probe=lambda base, key, timeout: ["m1", "m2"])
-    assert s["status"] == sh.OK
-    assert s["meta"]["endpoints"][0]["model_count"] == 2
-
-
-def test_providers_degraded_some_empty():
-    def probe(base, key, timeout):
-        return ["m1"] if "good" in base else []
-
-    s = sh.providers_health([_ep("good"), _ep("bad")], probe=probe)
-    assert s["status"] == sh.DEGRADED
-
-
-def test_providers_down_all_fail():
-    s = sh.providers_health([_ep("a")], probe=_raise)
-    assert s["status"] == sh.DOWN
-
-
-def test_providers_meta_never_leaks_api_key():
-    s = sh.providers_health([_ep("a")],
-                            probe=lambda base, key, timeout: ["m1"])
-    assert "sk-secret" not in repr(s)
-
-
-# ── rollup ──
-
-def test_rollup_picks_worst_non_disabled():
-    services = [
-        {"status": sh.OK}, {"status": sh.DISABLED},
-        {"status": sh.DEGRADED}, {"status": sh.OK},
-    ]
-    assert sh._rollup(services) == sh.DEGRADED
-
-
-def test_rollup_down_beats_degraded():
-    assert sh._rollup([{"status": sh.DEGRADED}, {"status": sh.DOWN}]) == sh.DOWN
-
-
-def test_rollup_all_disabled_is_ok():
-    assert sh._rollup([{"status": sh.DISABLED}, {"status": sh.DISABLED}]) == sh.OK
-
-
-# ── collect_service_health (async aggregate) ──
-
-def test_collect_service_health_shape(monkeypatch):
-    import asyncio
-
-    # Avoid touching real data sources / network.
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {"search_provider": "disabled"},
-        "integrations": [],
-        "accounts": [],
-        "endpoints": [],
-    })
-    out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
-    assert set(out) == {"overall", "services", "timestamp"}
-    names = {s["name"] for s in out["services"]}
-    assert names == {"chromadb", "searxng", "ntfy", "email", "providers"}
-    # Chroma healthy, everything else disabled → overall ok.
-    assert out["overall"] == sh.OK
-
-
-# ── _safe_url: strip userinfo / query / fragment ──
-
-@pytest.mark.parametrize("raw,expected", [
-    ("http://user:pass@host:8080/path?api_key=secret#frag", "http://host:8080/path"),
-    ("https://admin:hunter2@searx.example.com/", "https://searx.example.com"),
-    ("http://ntfy.local:80?token=abc", "http://ntfy.local:80"),
-    ("host:8080", "host:8080"),
-    ("", ""),
-    (None, ""),
-])
-def test_safe_url_strips_secrets(raw, expected):
-    out = sh._safe_url(raw)
-    assert out == expected
-    for bad in ("pass", "secret", "hunter2", "abc", "token", "@"):
-        if raw and bad in raw and bad not in expected:
-            assert bad not in out
-
-
-# ── _classify_error: controlled categories, never raw text ──
-
-def test_classify_error_categories():
-    import socket
-    assert sh._classify_error(TimeoutError()) == "timeout"
-    assert sh._classify_error(socket.timeout()) == "timeout"
-    assert sh._classify_error(socket.gaierror()) == "dns_error"
-    assert sh._classify_error(ConnectionRefusedError()) == "connection_refused"
-    assert sh._classify_error(OSError("boom")) == "network_error"
-    assert sh._classify_error(ValueError("x")) == "error"
-
-
-# ── Sanitization in subsystem output (blocker #2) ──
-
-def test_searxng_meta_redacts_instance_url():
-    s = sh.searxng_health(
-        {"search_provider": "searxng",
-         "search_url": "http://user:s3cr3t@searx.local:8080/?token=zzz"},
-        http_get=lambda url, timeout: _resp(200),
-    )
-    blob = repr(s)
-    assert "s3cr3t" not in blob and "zzz" not in blob and "user:" not in blob
-    assert s["meta"]["instance"] == "http://searx.local:8080"
-
-
-def test_searxng_down_uses_error_category_not_raw_exception():
-    def boom(url, timeout):
-        raise RuntimeError("failed connecting to http://user:pw@searx.local secret-token")
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://searx.local"},
-        http_get=boom,
-    )
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["error"] == "error"           # controlled category token
-    assert "secret-token" not in repr(s) and "pw@" not in repr(s)
-
-
-def test_ntfy_meta_redacts_userinfo_in_base():
-    intg = [{"preset": "ntfy", "enabled": True,
-             "base_url": "https://user:topsecret@ntfy.example.com"}]
-    seen = {}
-
-    def getter(url, timeout):
-        seen["url"] = url          # the probe itself may keep credentials
-        return _resp(200)
-
-    s = sh.ntfy_health(intg, {"reminder_channel": "ntfy"}, http_get=getter)
-    assert s["meta"]["base"] == "https://ntfy.example.com"
-    assert "topsecret" not in repr(s)
-
-
-def test_providers_name_fallback_is_sanitized():
-    # No display name → falls back to the base_url, which must be sanitized.
-    ep = {"base_url": "http://user:k3y@prov.local:9000/v1?api_key=zzz", "api_key": "sk-x"}
-    s = sh.providers_health([ep], probe=lambda b, k, t: ["m1"])
-    entry = s["meta"]["endpoints"][0]
-    assert entry["name"] == "http://prov.local:9000/v1"
-    assert "k3y" not in repr(s) and "zzz" not in repr(s) and "sk-x" not in repr(s)
-
-
-def test_providers_probe_exception_maps_to_category():
-    def boom(base, key, timeout):
-        raise RuntimeError(f"500 from {base} with key {key}")  # would leak base+key
-    s = sh.providers_health([_ep("a")], probe=boom)
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["endpoints"][0]["error"] == "error"
-    assert "sk-secret" not in repr(s) and "http://a" not in repr(s)
-
-
-def test_email_connect_exception_maps_to_category():
-    def boom(account_id):
-        raise RuntimeError("login failed for user bob with password hunter2")
-    s = sh.email_health([_acct("a")], connect=boom)
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["accounts"][0]["error"] == "error"
-    assert "hunter2" not in repr(s)
-
-
-# ── Bounded wall-clock (blocker #1) ──
-
-def test_providers_bounded_marks_slow_as_timeout(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def probe(base, key, timeout):
-        if "slow" in base:
-            time.sleep(10)          # would blow the budget if unbounded
-        return ["m1"]
-
-    eps = [{"name": "fast", "base_url": "http://fast", "api_key": "k"},
-           {"name": "slow", "base_url": "http://slow", "api_key": "k"}]
-    t0 = time.monotonic()
-    out = sh.providers_health(eps, probe=probe)
-    elapsed = time.monotonic() - t0
-    assert elapsed < 4, f"providers_health not bounded: took {elapsed:.1f}s"
-    by = {e["name"]: e for e in out["meta"]["endpoints"]}
-    assert by["fast"]["ok"] is True
-    assert by["slow"]["ok"] is False and by["slow"]["error"] == "timeout"
-    assert out["status"] == sh.DEGRADED
-
-
-def test_providers_bounded_with_many_slow_endpoints(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def probe(base, key, timeout):
-        time.sleep(10)
-        return ["m1"]
-
-    eps = [{"name": f"ep{i}", "base_url": f"http://ep{i}", "api_key": "k"}
-           for i in range(25)]
-    t0 = time.monotonic()
-    out = sh.providers_health(eps, probe=probe)
-    elapsed = time.monotonic() - t0
-    # 25 endpoints * sleep would be huge if sequential; bounded keeps it ~budget.
-    assert elapsed < 4, f"not bounded with many endpoints: {elapsed:.1f}s"
-    assert out["status"] == sh.DOWN
-    assert all(e["error"] == "timeout" for e in out["meta"]["endpoints"])
-
-
-def test_email_bounded_marks_slow_as_timeout(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def connect(account_id):
-        if account_id == "slow":
-            time.sleep(10)
-        return _Conn()
-
-    accts = [_acct("fast"), _acct("slow")]
-    accts[1]["account_id"] = "slow"
-    t0 = time.monotonic()
-    out = sh.email_health(accts, connect=connect)
-    elapsed = time.monotonic() - t0
-    assert elapsed < 4, f"email_health not bounded: took {elapsed:.1f}s"
-    by = {a["name"]: a for a in out["meta"]["accounts"]}
-    assert by["slow"]["error"] == "timeout"
-
-
-def test_collect_runs_subsystems_concurrently(monkeypatch):
-    # The aggregate is bounded by running the (internally-bounded) subsystems
-    # concurrently, so total wall-clock ≈ max(subsystem), not the sum. Each of
-    # the four network subsystems here sleeps ~0.6s; sequential would be ~2.4s.
-    import asyncio
-    import time
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
-    })
-
-    def slow(name):
-        def _fn(*_a, **_k):
-            time.sleep(0.6)
-            return {"name": name, "status": sh.OK, "detail": "", "meta": {}}
-        return _fn
-
-    monkeypatch.setattr(sh, "searxng_health", slow("searxng"))
-    monkeypatch.setattr(sh, "ntfy_health", slow("ntfy"))
-    monkeypatch.setattr(sh, "email_health", slow("email"))
-    monkeypatch.setattr(sh, "providers_health", slow("providers"))
-
-    t0 = time.monotonic()
-    out = asyncio.run(sh.collect_service_health(None, None))
-    elapsed = time.monotonic() - t0
-    assert elapsed < 1.5, f"subsystems not concurrent: took {elapsed:.1f}s"
-    assert {s["name"] for s in out["services"]} == {
-        "chromadb", "searxng", "ntfy", "email", "providers"}
-
-
-def test_collect_aggregate_deadline_yields_controlled_result(monkeypatch):
-    # If the gather overruns the aggregate ceiling, the response is still a
-    # controlled {overall, services, timestamp} with each network subsystem
-    # marked down/timeout — never a hang or a raised exception.
-    import asyncio
-    import time
-    monkeypatch.setattr(sh, "_AGGREGATE_DEADLINE", 0.5)
-    monkeypatch.setattr(sh, "_SUBSYSTEM_DEADLINE", 0.4)
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
-    })
-
-    async def _slow_gather(*coros, **_k):
-        for c in coros:                 # close unawaited coros to avoid warnings
-            close = getattr(c, "close", None)
-            if close:
-                close()
-        await asyncio.sleep(5)
-
-    # Force the outer wait_for to trip by making gather itself slow.
-    monkeypatch.setattr(sh.asyncio, "gather", _slow_gather)
-    t0 = time.monotonic()
-    out = asyncio.run(sh.collect_service_health(None, None))
-    elapsed = time.monotonic() - t0
-    assert elapsed < 2, f"aggregate deadline did not bound: {elapsed:.1f}s"
-    assert set(out) == {"overall", "services", "timestamp"}
-    net = [s for s in out["services"] if s["name"] != "chromadb"]
-    assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
-               for s in net)

tests/test_service_health_chromadb.py

@@ -0,0 +1,37 @@
+"""Tests for chromadb_health — ok/degraded/down/disabled classification."""
+import pytest
+
+from src import service_health as sh
+
+
+class _Store:
+    def __init__(self, healthy):
+        self.healthy = healthy
+
+
+def test_chromadb_both_healthy_ok():
+    s = sh.chromadb_health(_Store(True), _Store(True))
+    assert s["status"] == sh.OK
+    assert s["meta"] == {"rag": True, "memory": True}
+
+
+def test_chromadb_one_down_degraded():
+    s = sh.chromadb_health(_Store(True), _Store(False))
+    assert s["status"] == sh.DEGRADED
+
+
+def test_chromadb_both_unhealthy_down():
+    s = sh.chromadb_health(_Store(False), _Store(False))
+    assert s["status"] == sh.DOWN
+
+
+def test_chromadb_both_absent_disabled():
+    s = sh.chromadb_health(None, None)
+    assert s["status"] == sh.DISABLED
+
+
+def test_chromadb_one_absent_one_healthy_ok():
+    # An absent store is not a failure; the present one being healthy is ok.
+    s = sh.chromadb_health(_Store(True), None)
+    assert s["status"] == sh.OK
+    assert s["meta"]["memory"] is None

tests/test_service_health_collect.py

@@ -0,0 +1,139 @@
+"""Tests for rollup logic, aggregate collection, and shared utility helpers (_safe_url, _classify_error)."""
+import pytest
+
+from src import service_health as sh
+
+
+class _Store:
+    def __init__(self, healthy):
+        self.healthy = healthy
+
+
+# ── rollup ──
+
+def test_rollup_picks_worst_non_disabled():
+    services = [
+        {"status": sh.OK}, {"status": sh.DISABLED},
+        {"status": sh.DEGRADED}, {"status": sh.OK},
+    ]
+    assert sh._rollup(services) == sh.DEGRADED
+
+
+def test_rollup_down_beats_degraded():
+    assert sh._rollup([{"status": sh.DEGRADED}, {"status": sh.DOWN}]) == sh.DOWN
+
+
+def test_rollup_all_disabled_is_ok():
+    assert sh._rollup([{"status": sh.DISABLED}, {"status": sh.DISABLED}]) == sh.OK
+
+
+# ── collect_service_health (async aggregate) ──
+
+def test_collect_service_health_shape(monkeypatch):
+    import asyncio
+
+    # Avoid touching real data sources / network.
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {"search_provider": "disabled"},
+        "integrations": [],
+        "accounts": [],
+        "endpoints": [],
+    })
+    out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
+    assert set(out) == {"overall", "services", "timestamp"}
+    names = {s["name"] for s in out["services"]}
+    assert names == {"chromadb", "searxng", "ntfy", "email", "providers"}
+    # Chroma healthy, everything else disabled → overall ok.
+    assert out["overall"] == sh.OK
+
+
+# ── _safe_url: strip userinfo / query / fragment ──
+
+@pytest.mark.parametrize("raw,expected", [
+    ("http://user:pass@host:8080/path?api_key=secret#frag", "http://host:8080/path"),
+    ("https://admin:hunter2@searx.example.com/", "https://searx.example.com"),
+    ("http://ntfy.local:80?token=abc", "http://ntfy.local:80"),
+    ("host:8080", "host:8080"),
+    ("", ""),
+    (None, ""),
+])
+def test_safe_url_strips_secrets(raw, expected):
+    out = sh._safe_url(raw)
+    assert out == expected
+    for bad in ("pass", "secret", "hunter2", "abc", "token", "@"):
+        if raw and bad in raw and bad not in expected:
+            assert bad not in out
+
+
+# ── _classify_error: controlled categories, never raw text ──
+
+def test_classify_error_categories():
+    import socket
+    assert sh._classify_error(TimeoutError()) == "timeout"
+    assert sh._classify_error(socket.timeout()) == "timeout"
+    assert sh._classify_error(socket.gaierror()) == "dns_error"
+    assert sh._classify_error(ConnectionRefusedError()) == "connection_refused"
+    assert sh._classify_error(OSError("boom")) == "network_error"
+    assert sh._classify_error(ValueError("x")) == "error"
+
+
+# ── Concurrent collection and aggregate deadline ──
+
+def test_collect_runs_subsystems_concurrently(monkeypatch):
+    # The aggregate is bounded by running the (internally-bounded) subsystems
+    # concurrently, so total wall-clock ≈ max(subsystem), not the sum. Each of
+    # the four network subsystems here sleeps ~0.6s; sequential would be ~2.4s.
+    import asyncio
+    import time
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
+    })
+
+    def slow(name):
+        def _fn(*_a, **_k):
+            time.sleep(0.6)
+            return {"name": name, "status": sh.OK, "detail": "", "meta": {}}
+        return _fn
+
+    monkeypatch.setattr(sh, "searxng_health", slow("searxng"))
+    monkeypatch.setattr(sh, "ntfy_health", slow("ntfy"))
+    monkeypatch.setattr(sh, "email_health", slow("email"))
+    monkeypatch.setattr(sh, "providers_health", slow("providers"))
+
+    t0 = time.monotonic()
+    out = asyncio.run(sh.collect_service_health(None, None))
+    elapsed = time.monotonic() - t0
+    assert elapsed < 1.5, f"subsystems not concurrent: took {elapsed:.1f}s"
+    assert {s["name"] for s in out["services"]} == {
+        "chromadb", "searxng", "ntfy", "email", "providers"}
+
+
+def test_collect_aggregate_deadline_yields_controlled_result(monkeypatch):
+    # If the gather overruns the aggregate ceiling, the response is still a
+    # controlled {overall, services, timestamp} with each network subsystem
+    # marked down/timeout — never a hang or a raised exception.
+    import asyncio
+    import time
+    monkeypatch.setattr(sh, "_AGGREGATE_DEADLINE", 0.5)
+    monkeypatch.setattr(sh, "_SUBSYSTEM_DEADLINE", 0.4)
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
+    })
+
+    async def _slow_gather(*coros, **_k):
+        for c in coros:                 # close unawaited coros to avoid warnings
+            close = getattr(c, "close", None)
+            if close:
+                close()
+        await asyncio.sleep(5)
+
+    # Force the outer wait_for to trip by making gather itself slow.
+    monkeypatch.setattr(sh.asyncio, "gather", _slow_gather)
+    t0 = time.monotonic()
+    out = asyncio.run(sh.collect_service_health(None, None))
+    elapsed = time.monotonic() - t0
+    assert elapsed < 2, f"aggregate deadline did not bound: {elapsed:.1f}s"
+    assert set(out) == {"overall", "services", "timestamp"}
+    net = [s for s in out["services"] if s["name"] != "chromadb"]
+    assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
+               for s in net)

tests/test_service_health_email.py

@@ -0,0 +1,80 @@
+"""Tests for email_health — probe logic, status classification, sanitization, and bounded timeout."""
+import pytest
+
+from src import service_health as sh
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _acct(name, host="imap.example.com"):
+    return {"account_id": name, "account_name": name, "imap_host": host,
+            "imap_password": "hunter2"}
+
+
+class _Conn:
+    def logout(self):
+        pass
+
+
+def test_email_disabled_without_accounts():
+    assert sh.email_health([])["status"] == sh.DISABLED
+
+
+def test_email_ok_all_connect():
+    s = sh.email_health([_acct("a"), _acct("b")], connect=lambda _id: _Conn())
+    assert s["status"] == sh.OK
+
+
+def test_email_degraded_some_fail():
+    def connect(account_id):
+        if account_id == "bad":
+            raise RuntimeError("auth failed")
+        return _Conn()
+
+    s = sh.email_health([_acct("good"), _acct("bad")], connect=connect)
+    assert s["status"] == sh.DEGRADED
+
+
+def test_email_down_all_fail():
+    s = sh.email_health([_acct("a")], connect=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_email_account_without_host_marked_failed():
+    s = sh.email_health([_acct("a", host="")], connect=lambda _id: _Conn())
+    assert s["status"] == sh.DOWN
+
+
+def test_email_meta_never_leaks_password():
+    s = sh.email_health([_acct("a")], connect=lambda _id: _Conn())
+    assert "hunter2" not in repr(s)
+
+
+def test_email_connect_exception_maps_to_category():
+    def boom(account_id):
+        raise RuntimeError("login failed for user bob with password hunter2")
+    s = sh.email_health([_acct("a")], connect=boom)
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["accounts"][0]["error"] == "error"
+    assert "hunter2" not in repr(s)
+
+
+def test_email_bounded_marks_slow_as_timeout(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def connect(account_id):
+        if account_id == "slow":
+            time.sleep(10)
+        return _Conn()
+
+    accts = [_acct("fast"), _acct("slow")]
+    accts[1]["account_id"] = "slow"
+    t0 = time.monotonic()
+    out = sh.email_health(accts, connect=connect)
+    elapsed = time.monotonic() - t0
+    assert elapsed < 4, f"email_health not bounded: took {elapsed:.1f}s"
+    by = {a["name"]: a for a in out["meta"]["accounts"]}
+    assert by["slow"]["error"] == "timeout"

tests/test_service_health_ntfy.py

@@ -0,0 +1,62 @@
+"""Tests for ntfy_health — probe logic, status classification, and sanitization."""
+import types
+
+import pytest
+
+from src import service_health as sh
+
+
+def _resp(status_code):
+    return types.SimpleNamespace(status_code=status_code)
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _ntfy_intg():
+    return [{"preset": "ntfy", "enabled": True, "base_url": "http://ntfy:80"}]
+
+
+def test_ntfy_disabled_without_integration():
+    s = sh.ntfy_health([], {"reminder_channel": "ntfy"})
+    assert s["status"] == sh.DISABLED
+
+
+def test_ntfy_ok():
+    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
+                       http_get=lambda url, timeout: _resp(200))
+    assert s["status"] == sh.OK
+    assert s["meta"]["base"] == "http://ntfy:80"
+
+
+def test_ntfy_probes_v1_health_not_a_topic():
+    seen = {}
+
+    def getter(url, timeout):
+        seen["url"] = url
+        return _resp(200)
+
+    sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"}, http_get=getter)
+    # Non-intrusive: hits /v1/health, never publishes to a topic.
+    assert seen["url"].endswith("/v1/health")
+
+
+def test_ntfy_down_on_exception():
+    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
+                       http_get=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_ntfy_meta_redacts_userinfo_in_base():
+    intg = [{"preset": "ntfy", "enabled": True,
+             "base_url": "https://user:topsecret@ntfy.example.com"}]
+    seen = {}
+
+    def getter(url, timeout):
+        seen["url"] = url          # the probe itself may keep credentials
+        return _resp(200)
+
+    s = sh.ntfy_health(intg, {"reminder_channel": "ntfy"}, http_get=getter)
+    assert s["meta"]["base"] == "https://ntfy.example.com"
+    assert "topsecret" not in repr(s)

tests/test_service_health_providers.py

@@ -0,0 +1,100 @@
+"""Tests for providers_health — probe logic, status classification, sanitization, and bounded timeout."""
+import pytest
+
+from src import service_health as sh
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _ep(name):
+    return {"name": name, "base_url": f"http://{name}:8000/v1", "api_key": "sk-secret"}
+
+
+def test_providers_disabled_without_endpoints():
+    assert sh.providers_health([])["status"] == sh.DISABLED
+
+
+def test_providers_ok_all_reachable():
+    s = sh.providers_health([_ep("a")],
+                            probe=lambda base, key, timeout: ["m1", "m2"])
+    assert s["status"] == sh.OK
+    assert s["meta"]["endpoints"][0]["model_count"] == 2
+
+
+def test_providers_degraded_some_empty():
+    def probe(base, key, timeout):
+        return ["m1"] if "good" in base else []
+
+    s = sh.providers_health([_ep("good"), _ep("bad")], probe=probe)
+    assert s["status"] == sh.DEGRADED
+
+
+def test_providers_down_all_fail():
+    s = sh.providers_health([_ep("a")], probe=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_providers_meta_never_leaks_api_key():
+    s = sh.providers_health([_ep("a")],
+                            probe=lambda base, key, timeout: ["m1"])
+    assert "sk-secret" not in repr(s)
+
+
+def test_providers_name_fallback_is_sanitized():
+    # No display name → falls back to the base_url, which must be sanitized.
+    ep = {"base_url": "http://user:k3y@prov.local:9000/v1?api_key=zzz", "api_key": "sk-x"}
+    s = sh.providers_health([ep], probe=lambda b, k, t: ["m1"])
+    entry = s["meta"]["endpoints"][0]
+    assert entry["name"] == "http://prov.local:9000/v1"
+    assert "k3y" not in repr(s) and "zzz" not in repr(s) and "sk-x" not in repr(s)
+
+
+def test_providers_probe_exception_maps_to_category():
+    def boom(base, key, timeout):
+        raise RuntimeError(f"500 from {base} with key {key}")  # would leak base+key
+    s = sh.providers_health([_ep("a")], probe=boom)
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["endpoints"][0]["error"] == "error"
+    assert "sk-secret" not in repr(s) and "http://a" not in repr(s)
+
+
+def test_providers_bounded_marks_slow_as_timeout(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def probe(base, key, timeout):
+        if "slow" in base:
+            time.sleep(10)          # would blow the budget if unbounded
+        return ["m1"]
+
+    eps = [{"name": "fast", "base_url": "http://fast", "api_key": "k"},
+           {"name": "slow", "base_url": "http://slow", "api_key": "k"}]
+    t0 = time.monotonic()
+    out = sh.providers_health(eps, probe=probe)
+    elapsed = time.monotonic() - t0
+    assert elapsed < 4, f"providers_health not bounded: took {elapsed:.1f}s"
+    by = {e["name"]: e for e in out["meta"]["endpoints"]}
+    assert by["fast"]["ok"] is True
+    assert by["slow"]["ok"] is False and by["slow"]["error"] == "timeout"
+    assert out["status"] == sh.DEGRADED
+
+
+def test_providers_bounded_with_many_slow_endpoints(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def probe(base, key, timeout):
+        time.sleep(10)
+        return ["m1"]
+
+    eps = [{"name": f"ep{i}", "base_url": f"http://ep{i}", "api_key": "k"}
+           for i in range(25)]
+    t0 = time.monotonic()
+    out = sh.providers_health(eps, probe=probe)
+    elapsed = time.monotonic() - t0
+    # 25 endpoints * sleep would be huge if sequential; bounded keeps it ~budget.
+    assert elapsed < 4, f"not bounded with many endpoints: {elapsed:.1f}s"
+    assert out["status"] == sh.DOWN
+    assert all(e["error"] == "timeout" for e in out["meta"]["endpoints"])

tests/test_service_health_search.py

@@ -0,0 +1,79 @@
+"""Tests for searxng_health — probe logic, status classification, and sanitization."""
+import types
+
+import pytest
+
+from src import service_health as sh
+
+
+def _resp(status_code):
+    return types.SimpleNamespace(status_code=status_code)
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def test_searxng_disabled_when_other_provider():
+    s = sh.searxng_health({"search_provider": "brave"})
+    assert s["status"] == sh.DISABLED
+
+
+def test_searxng_ok_on_healthz():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=lambda url, timeout: _resp(200),
+    )
+    assert s["status"] == sh.OK
+    assert s["meta"]["probed"] == "/healthz"
+
+
+def test_searxng_ok_on_root_fallback():
+    def getter(url, timeout):
+        return _resp(404) if url.endswith("/healthz") else _resp(200)
+
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=getter,
+    )
+    assert s["status"] == sh.OK
+    assert s["meta"]["probed"] == "/"
+
+
+def test_searxng_down_on_exception():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=_raise,
+    )
+    assert s["status"] == sh.DOWN
+
+
+def test_searxng_down_on_5xx():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=lambda url, timeout: _resp(502),
+    )
+    assert s["status"] == sh.DOWN
+
+
+def test_searxng_meta_redacts_instance_url():
+    s = sh.searxng_health(
+        {"search_provider": "searxng",
+         "search_url": "http://user:s3cr3t@searx.local:8080/?token=zzz"},
+        http_get=lambda url, timeout: _resp(200),
+    )
+    blob = repr(s)
+    assert "s3cr3t" not in blob and "zzz" not in blob and "user:" not in blob
+    assert s["meta"]["instance"] == "http://searx.local:8080"
+
+
+def test_searxng_down_uses_error_category_not_raw_exception():
+    def boom(url, timeout):
+        raise RuntimeError("failed connecting to http://user:pw@searx.local secret-token")
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://searx.local"},
+        http_get=boom,
+    )
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["error"] == "error"           # controlled category token
+    assert "secret-token" not in repr(s) and "pw@" not in repr(s)

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."""

tests/test_workspace_confine.py

@@ -167,6 +167,38 @@ async def test_glob_confined_e2e(ws, admin):
     assert r["exit_code"] == 0 and "No files" in r["output"]
 
 
+@pytest.mark.asyncio
+async def test_glob_skips_sensitive_files_in_workspace(ws, admin):
+    """glob must not enumerate deny-listed sensitive files that live inside the
+    workspace. read_file/write_file/edit_file refuse them and grep skips them,
+    so glob surfacing their paths is an enumeration oracle for prompt-injection.
+    """
+    with open(os.path.join(ws, "keep.py"), "w") as f:
+        f.write("x")
+    with open(os.path.join(ws, ".env"), "w") as f:
+        f.write("AWS_SECRET=xxx")
+    with open(os.path.join(ws, "id_rsa"), "w") as f:  # non-dotfile key at root
+        f.write("KEY")
+    os.makedirs(os.path.join(ws, ".ssh"), exist_ok=True)
+    with open(os.path.join(ws, ".ssh", "authorized_keys"), "w") as f:
+        f.write("ssh-rsa AAAA")
+
+    # A recursive wildcard returns ordinary files but none of the sensitive
+    # ones. The pattern "**/*" contains no secret names, so a secret basename
+    # appearing in the output is a real leak (not the echoed not-found pattern).
+    _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": "**/*"})), owner="a", workspace=ws)
+    assert r["exit_code"] == 0
+    assert "keep.py" in r["output"]
+    for leak in (".env", "id_rsa", "authorized_keys"):
+        assert leak not in r["output"], f"glob leaked sensitive file: {leak}"
+
+    # Directly targeting a sensitive file (literal fast-path and wildcard) must
+    # come back as the not-found message, never a match with the file's path.
+    for pat in (".env", "**/id_rsa", "**/authorized_keys"):
+        _, r = await execute_tool_block(_block("glob", json.dumps({"pattern": pat})), owner="a", workspace=ws)
+        assert r["exit_code"] == 0 and "No files" in r["output"]
+
+
 @pytest.mark.asyncio
 async def test_subprocess_cwd_is_workspace_e2e(ws, admin):
     """python tool runs with cwd = workspace (OS-agnostic probe)."""