refactor(routes): move contacts domain into routes/contacts/ subpackage (#5227)

Slice 2e of the route-domain reorganization (#4082/#4071, per
specs/architecture-runtime-inventory.md §6.3). Moves contacts_routes.py into
routes/contacts/, leaving a backward-compat sys.modules shim at the old path.
Pure file reorganization, no behavior change.

The shim uses sys.modules replacement (same pattern as the merged gallery
#4903, research #4975, memory #5007, and history #5090 slices) so that
`import routes.contacts_routes`, `from routes.contacts_routes import X`,
`importlib.import_module(...)`, the string-targeted
`monkeypatch.setattr("routes.contacts_routes.SETTINGS_FILE", ...)` used by
test_carddav_password_encryption.py, and the `import ... as cr` +
`setattr(cr, ...)` pattern in test_contacts_add_null_name.py all operate on
the same module object the application uses. This also keeps the mutable
module state `_contact_cache` identical across import paths.

The canonical module does NOT depend on the shim — routes/contacts/
contacts_routes.py imports only from core/, src/, and stdlib (zero internal
routes/ coupling). The inbound edge from routes/email_helpers.py (imports
_fetch_contacts) keeps working through the shim.

Zero source-introspection landmines — no test reads this file by path.

Adds tests/test_contacts_routes_shim.py to pin the sys.modules shim contract
(same-object + string-targeted monkeypatch reach-through).

Verified: compileall clean; full suite 4485 passed, 3 skipped.

.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

app.py

@@ -645,7 +645,7 @@ from routes.research.research_routes import setup_research_routes
 app.include_router(setup_research_routes(research_handler, session_manager=session_manager))
 
 # History
-from routes.history_routes import setup_history_routes
+from routes.history.history_routes import setup_history_routes
 app.include_router(setup_history_routes(session_manager))
 
 # Search
@@ -813,7 +813,7 @@ from routes.vault_routes import setup_vault_routes
 app.include_router(setup_vault_routes())
 
 # Contacts (CardDAV)
-from routes.contacts_routes import setup_contacts_routes
+from routes.contacts.contacts_routes import setup_contacts_routes
 app.include_router(setup_contacts_routes())
 
 from companion import setup_companion_routes

routes/chat_routes.py

@@ -781,11 +781,9 @@ def setup_chat_routes(
         # by default without having to send allow_bash in every request.
         if allow_bash is not None and str(allow_bash).lower() != "true":
             disabled_tools.add("bash")
-        _explicit_web_intent = bool(_tool_intent and _tool_intent.category == "web")
         if (
             allow_web_search is not None
             and str(allow_web_search).lower() != "true"
-            and not _explicit_web_intent
         ):
             disabled_tools.add("web_search")
             disabled_tools.add("web_fetch")

routes/contacts/__init__.py

@@ -0,0 +1,5 @@
+"""Contacts route domain package (slice 2e, #4082/#4071).
+
+Contains contacts_routes.py, migrated from the flat routes/ directory.
+Backward-compat shim at routes/contacts_routes.py re-exports from here.
+"""

routes/contacts/contacts_routes.py

@@ -0,0 +1,900 @@
+"""
+contacts_routes.py
+
+CardDAV contacts integration. Reads from local Radicale, supports
+search and adding new contacts.
+"""
+
+import re
+import logging
+import uuid
+import json
+import csv
+import io
+import os
+import inspect
+import httpx
+from pathlib import Path
+from datetime import datetime
+from urllib.parse import urljoin, urlparse, urlunparse
+
+from core.log_safety import redact_url
+from fastapi import APIRouter, Query, Depends, Response, HTTPException
+from typing import List, Dict, Optional
+
+from core.middleware import require_admin
+from src.url_safety import check_outbound_url
+
+logger = logging.getLogger(__name__)
+
+from src.constants import DATA_DIR as _DATA_DIR, SETTINGS_FILE as _SETTINGS_FILE, CONTACTS_FILE as _CONTACTS_FILE
+DATA_DIR = Path(_DATA_DIR)
+SETTINGS_FILE = Path(_SETTINGS_FILE)
+LOCAL_CONTACTS_FILE = Path(_CONTACTS_FILE)
+
+
+def _load_settings():
+    if SETTINGS_FILE.exists():
+        return json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
+    return {}
+
+
+def _save_settings(settings):
+    from core.atomic_io import atomic_write_json
+    atomic_write_json(str(SETTINGS_FILE), settings, indent=2)
+
+
+def _get_carddav_config():
+    import os
+    settings = _load_settings()
+    password = settings.get("carddav_password", os.environ.get("CARDDAV_PASSWORD", ""))
+    if password and "carddav_password" in settings:
+        from src.secret_storage import decrypt
+        password = decrypt(password)
+    return {
+        "url": settings.get("carddav_url", os.environ.get("CARDDAV_URL", "")),
+        "username": settings.get("carddav_username", os.environ.get("CARDDAV_USERNAME", "")),
+        "password": password,
+    }
+
+
+def _carddav_configured(cfg: Optional[Dict] = None) -> bool:
+    cfg = cfg or _get_carddav_config()
+    return bool((cfg.get("url") or "").strip())
+
+
+def _validate_carddav_url(url: str) -> str:
+    cleaned = (url if isinstance(url, str) else "").strip().rstrip("/")
+    ok, reason = check_outbound_url(
+        cleaned,
+        block_private=os.getenv("CARDDAV_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+    )
+    if not ok:
+        raise ValueError(f"Rejected CardDAV URL: {reason}")
+    return cleaned
+
+
+def _carddav_base_url(cfg: Dict) -> str:
+    return _validate_carddav_url(cfg.get("url") or "")
+
+
+def _normalize_contact(contact: Dict) -> Dict:
+    emails = []
+    for e in contact.get("emails") or ([] if not contact.get("email") else [contact.get("email")]):
+        e = str(e or "").strip()
+        if e and e not in emails:
+            emails.append(e)
+    phones = []
+    for p in contact.get("phones") or ([] if not contact.get("phone") else [contact.get("phone")]):
+        p = str(p or "").strip()
+        if p and p not in phones:
+            phones.append(p)
+    name = str(contact.get("name") or "").strip()
+    if not name and emails:
+        name = emails[0].split("@")[0]
+    address = str(contact.get("address") or "").strip()
+    return {
+        "uid": str(contact.get("uid") or uuid.uuid4()),
+        "name": name,
+        "emails": emails,
+        "phones": phones,
+        "address": address,
+    }
+
+
+def _load_local_contacts() -> List[Dict]:
+    try:
+        if not LOCAL_CONTACTS_FILE.exists():
+            return []
+        data = json.loads(LOCAL_CONTACTS_FILE.read_text(encoding="utf-8"))
+        rows = data.get("contacts", data) if isinstance(data, dict) else data
+        return [_normalize_contact(c) for c in (rows or []) if isinstance(c, dict)]
+    except Exception as e:
+        logger.error(f"Failed to load local contacts: {e}")
+        return []
+
+
+def _save_local_contacts(contacts: List[Dict]) -> None:
+    from core.atomic_io import atomic_write_json
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    atomic_write_json(str(LOCAL_CONTACTS_FILE), {"contacts": [_normalize_contact(c) for c in contacts]}, indent=2)
+    _contact_cache["contacts"] = [_normalize_contact(c) for c in contacts]
+    _contact_cache["fetched_at"] = datetime.utcnow()
+
+
+# ── vCard parsing ──
+
+def _vunesc(value: str) -> str:
+    """Reverse _vesc() — turn escaped vCard text back into the raw value.
+    Order matters: handle \\n/\\, /\\; first, backslash-unescape last."""
+    if not value:
+        return value
+    out = []
+    i = 0
+    while i < len(value):
+        ch = value[i]
+        if ch == "\\" and i + 1 < len(value):
+            nxt = value[i + 1]
+            if nxt in ("n", "N"):
+                out.append("\n")
+            elif nxt in (",", ";", "\\"):
+                out.append(nxt)
+            else:
+                out.append(nxt)
+            i += 2
+        else:
+            out.append(ch)
+            i += 1
+    return "".join(out)
+
+
+def _parse_vcards(text: str) -> List[Dict]:
+    """Parse a stream of vCards into dicts with name, email, phone."""
+    # Unfold RFC 6350 3.2 line folding first: a CRLF/LF followed by a single
+    # space or tab is a continuation of the previous logical line. Real
+    # CardDAV servers (Radicale, iCloud, Apple/Google) fold long EMAIL / FN /
+    # PHOTO lines, and splitting on raw newlines without unfolding dropped the
+    # continuation (e.g. "...@example\n .com" lost the ".com"), truncating the
+    # email/name.
+    text = re.sub(r"\r\n[ \t]", "", text or "")
+    text = re.sub(r"\n[ \t]", "", text)
+    contacts = []
+    for block in re.split(r"BEGIN:VCARD", text):
+        if not block.strip():
+            continue
+        contact = {"name": "", "emails": [], "phones": [], "uid": "", "address": ""}
+        for line in block.split("\n"):
+            line = line.strip()
+            # Strip an optional RFC 6350 group prefix (e.g. "item1.EMAIL;...")
+            # that Apple Contacts / iCloud / many CardDAV servers emit by
+            # default — without this the property-name checks below miss those
+            # lines and silently drop the email / phone. The group token only
+            # precedes the property name, so it is safe to strip for matching
+            # and value extraction, and a no-op for non-grouped lines.
+            name_part = re.sub(r"^[A-Za-z0-9-]+\.", "", line, count=1)
+            if name_part.startswith("FN:") or name_part.startswith("FN;"):
+                contact["name"] = _vunesc(name_part.split(":", 1)[1]) if ":" in name_part else ""
+            elif name_part.startswith("EMAIL"):
+                # Handle EMAIL:foo@bar OR EMAIL;TYPE=...:foo@bar OR EMAIL;PREF=1:foo@bar
+                if ":" in name_part:
+                    email_addr = _vunesc(name_part.split(":", 1)[1])
+                    if email_addr and email_addr not in contact["emails"]:
+                        contact["emails"].append(email_addr)
+            elif name_part.startswith("TEL"):
+                if ":" in name_part:
+                    phone = _vunesc(name_part.split(":", 1)[1])
+                    if phone and phone not in contact["phones"]:
+                        contact["phones"].append(phone)
+            elif name_part.startswith("ADR"):
+                # vCard ADR is 7 semicolon-separated components:
+                # post-office-box;extended-address;street;locality;region;postal-code;country.
+                # Recover a human-readable string by joining non-empty
+                # components with ", ".
+                if ":" in name_part:
+                    raw = name_part.split(":", 1)[1]
+                    parts = [_vunesc(p).strip() for p in raw.split(";")]
+                    contact["address"] = ", ".join(p for p in parts if p)
+            elif name_part.startswith("UID:"):
+                contact["uid"] = _vunesc(name_part[4:])
+        if contact["name"] or contact["emails"]:
+            contacts.append(contact)
+    return contacts
+
+
+def _vesc(value: str) -> str:
+    """Escape a vCard property VALUE per RFC 6350 §3.4: backslash, comma,
+    semicolon, and newlines. Without this, a name like 'Sekisui House,Ltd'
+    or any value containing a newline produces a malformed vCard (broken
+    N/FN fields) or could inject arbitrary properties."""
+    return (
+        (value or "")
+        .replace("\\", "\\\\")
+        .replace("\n", "\\n")
+        .replace("\r", "")
+        .replace(",", "\\,")
+        .replace(";", "\\;")
+    )
+
+
+def _build_vcard(name: str, email: str, uid: Optional[str] = None,
+                 emails: Optional[List[str]] = None,
+                 phones: Optional[List[str]] = None,
+                 address: Optional[str] = None) -> str:
+    """Build a vCard. Accepts either a single `email` (legacy callers) or
+    full `emails`/`phones` lists (edit path). The first email is marked
+    PREF=1. All values are RFC-6350-escaped."""
+    if not uid:
+        uid = str(uuid.uuid4())
+    # Normalize email lists — `email` arg is a convenience for single-email
+    # creation; `emails` (if given) is authoritative.
+    email_list = [e.strip() for e in (emails if emails is not None else ([email] if email else [])) if e and e.strip()]
+    phone_list = [p.strip() for p in (phones or []) if p and p.strip()]
+    # Try to split name into first/last
+    parts = name.strip().split()
+    if len(parts) >= 2:
+        first = parts[0]
+        last = " ".join(parts[1:])
+    else:
+        first = name
+        last = ""
+    # N field is structured (5 components separated by ';') — escape each
+    # component individually so a comma in the name doesn't split it.
+    n_field = f"{_vesc(last)};{_vesc(first)};;;"
+    lines = [
+        "BEGIN:VCARD",
+        "VERSION:4.0",
+        f"UID:{_vesc(uid)}",
+        f"FN:{_vesc(name)}",
+        f"N:{n_field}",
+    ]
+    for i, em in enumerate(email_list):
+        # First email is the preferred one.
+        lines.append(f"EMAIL;PREF=1:{_vesc(em)}" if i == 0 else f"EMAIL:{_vesc(em)}")
+    for ph in phone_list:
+        lines.append(f"TEL:{_vesc(ph)}")
+    # Address: stuff the whole human-readable string into the street
+    # component of ADR. vCard ADR has 7 semicolon-separated components:
+    # post-office-box;extended-address;street;locality;region;postal-code;country.
+    addr = (address or "").strip()
+    if addr:
+        lines.append(f"ADR:;;{_vesc(addr)};;;;")
+    lines.append("END:VCARD")
+    return "\r\n".join(lines) + "\r\n"
+
+
+# ── In-memory cache ──
+
+_contact_cache = {"contacts": [], "fetched_at": None}
+
+
+def _abs_url(href: str) -> str:
+    """Combine a multistatus <href> (an absolute path like
+    /user/contacts/x.vcf) with the configured CardDAV server origin so we
+    get a fully-qualified URL to PUT/DELETE. Absolute hrefs are accepted only
+    for the configured origin; a cross-origin href is treated as a path on the
+    configured server so a malicious CardDAV response cannot redirect later
+    writes/deletes to cloud metadata or another host."""
+    cfg = _get_carddav_config()
+    base = _carddav_base_url(cfg)
+    base_p = urlparse(base)
+    joined = urljoin(base.rstrip("/") + "/", href or "")
+    joined_p = urlparse(joined)
+    if (joined_p.scheme, joined_p.netloc) != (base_p.scheme, base_p.netloc):
+        joined = urlunparse((base_p.scheme, base_p.netloc, joined_p.path or "/", "", joined_p.query, ""))
+    return _validate_carddav_url(joined)
+
+
+# CardDAV REPORT body — pull every card's etag + raw vCard in ONE request,
+# alongside the resource href. Lets us map each contact's UID to the real
+# server resource path (which is NOT always <uid>.vcf for contacts created
+# by other clients).
+_ADDRESSBOOK_QUERY = (
+    '<?xml version="1.0" encoding="utf-8"?>'
+    '<C:addressbook-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:carddav">'
+    '<D:prop><D:getetag/><C:address-data/></D:prop>'
+    '<C:filter/>'
+    '</C:addressbook-query>'
+)
+
+
+def _fetch_via_report(cfg, auth):
+    """Try a CardDAV REPORT addressbook-query — returns contacts WITH an
+    `href` field, or None if the server doesn't support it / errors."""
+    from defusedxml import ElementTree as ET
+    try:
+        r = httpx.request(
+            "REPORT", cfg["url"],
+            content=_ADDRESSBOOK_QUERY.encode("utf-8"),
+            headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "1"},
+            auth=auth, timeout=10,
+        )
+        if r.status_code not in (207, 200):
+            return None
+        root = ET.fromstring(r.text)
+        ns = {"D": "DAV:", "C": "urn:ietf:params:xml:ns:carddav"}
+        out = []
+        for resp in root.findall("D:response", ns):
+            href_el = resp.find("D:href", ns)
+            data_el = resp.find(".//C:address-data", ns)
+            if href_el is None or data_el is None or not (data_el.text or "").strip():
+                continue
+            parsed = _parse_vcards(data_el.text)
+            if not parsed:
+                continue
+            c = parsed[0]
+            c["href"] = href_el.text.strip()
+            out.append(c)
+        # If the REPORT parsed to ZERO contacts, don't trust it — some
+        # CardDAV servers treat an empty <filter/> as "match nothing" and
+        # return a valid-but-empty 207. Return None so the caller falls
+        # back to the plain GET (which lists everything). A genuinely empty
+        # address book just costs one extra GET that also returns nothing.
+        if not out:
+            return None
+        return out
+    except Exception as e:
+        logger.warning(f"CardDAV REPORT failed, falling back to GET: {e}")
+        return None
+
+
+def _fetch_contacts(force=False):
+    """Fetch all contacts. Uses CardDAV when configured, otherwise local JSON."""
+    if not force and _contact_cache["fetched_at"]:
+        age = (datetime.utcnow() - _contact_cache["fetched_at"]).total_seconds()
+        if age < 60:
+            return _contact_cache["contacts"]
+
+    cfg = _get_carddav_config()
+    if not _carddav_configured(cfg):
+        contacts = _load_local_contacts()
+        _contact_cache["contacts"] = contacts
+        _contact_cache["fetched_at"] = datetime.utcnow()
+        return contacts
+
+    try:
+        cfg["url"] = _carddav_base_url(cfg)
+        auth = None
+        if cfg["username"]:
+            auth = (cfg["username"], cfg["password"])
+        # Preferred path: REPORT gives us hrefs for reliable edit/delete.
+        contacts = _fetch_via_report(cfg, auth)
+        if contacts is None:
+            # Fallback: plain GET, concatenated vCards, no hrefs.
+            r = httpx.get(cfg["url"], auth=auth, timeout=10)
+            if r.status_code != 200:
+                logger.warning(f"CardDAV returned {r.status_code}")
+                return _contact_cache["contacts"]
+            contacts = _parse_vcards(r.text)
+        _contact_cache["contacts"] = contacts
+        _contact_cache["fetched_at"] = datetime.utcnow()
+        return contacts
+    except Exception as e:
+        logger.error(f"Failed to fetch contacts: {e}")
+        return _contact_cache["contacts"]
+
+
+def _resolve_resource_url(uid: str) -> str:
+    """Map a contact UID to its real CardDAV resource URL. Uses the href
+    captured during fetch when available (handles contacts whose filename
+    != UID); falls back to the <uid>.vcf guess for app-created contacts or
+    when no href is known."""
+    def _lookup():
+        for c in _contact_cache.get("contacts", []):
+            if c.get("uid") == uid and c.get("href"):
+                return _abs_url(c["href"])
+        return None
+    found = _lookup()
+    if found:
+        return found
+    # Not in cache (or no href) — refresh once and retry before guessing.
+    try:
+        _fetch_contacts(force=True)
+    except Exception:
+        pass
+    return _lookup() or _vcard_url(uid)
+
+
+def _create_contact(name: str, email: str, address: str = "") -> bool:
+    """Add a new contact via CardDAV or local contacts."""
+    cfg = _get_carddav_config()
+    if not _carddav_configured(cfg):
+        contacts = _load_local_contacts()
+        email_l = (email or "").strip().lower()
+        for c in contacts:
+            if email_l and email_l in [e.lower() for e in c.get("emails", [])]:
+                return True
+        contacts.append(_normalize_contact({"name": name, "emails": [email], "address": address}))
+        _save_local_contacts(contacts)
+        return True
+
+    contact_uid = str(uuid.uuid4())
+    vcard = _build_vcard(name, email, contact_uid, address=address)
+    try:
+        url = _carddav_base_url(cfg) + "/" + contact_uid + ".vcf"
+        auth = None
+        if cfg["username"]:
+            auth = (cfg["username"], cfg["password"])
+        r = httpx.put(
+            url,
+            data=vcard.encode("utf-8"),
+            headers={"Content-Type": "text/vcard; charset=utf-8"},
+            auth=auth,
+            timeout=10,
+        )
+        if r.status_code in (200, 201, 204):
+            # Invalidate cache
+            _contact_cache["fetched_at"] = None
+            return True
+        logger.warning(f"CardDAV PUT returned {r.status_code}: {r.text[:200]}")
+        return False
+    except Exception as e:
+        logger.error(f"Failed to create contact: {e}")
+        return False
+
+
+def _vcard_url(uid: str) -> str:
+    """The CardDAV resource URL for a given contact UID. The uid is URL-
+    encoded so a value containing '/', '..' or other path chars can't
+    escape the collection and target an arbitrary CardDAV resource."""
+    from urllib.parse import quote
+    cfg = _get_carddav_config()
+    return _carddav_base_url(cfg) + "/" + quote(uid, safe="") + ".vcf"
+
+
+def _import_vcards(text: str) -> Dict:
+    """Import a (possibly multi-card) .vcf blob. Each card is PUT to the
+    CardDAV server PRESERVING its full original content (ADR/ORG/photo/
+    etc.) — we don't rebuild it, just ensure it has VERSION + UID and
+    normalize line endings. Returns {imported, failed, total}."""
+    from urllib.parse import quote
+    cfg = _get_carddav_config()
+    if not cfg.get("url"):
+        parsed = _parse_vcards(text)
+        contacts = _load_local_contacts()
+        existing = {
+            e.lower()
+            for c in contacts
+            for e in (c.get("emails") or [])
+            if e
+        }
+        imported = 0
+        for c in parsed:
+            emails = [e for e in (c.get("emails") or []) if e]
+            if emails and any(e.lower() in existing for e in emails):
+                continue
+            contacts.append(_normalize_contact(c))
+            for e in emails:
+                existing.add(e.lower())
+            imported += 1
+        if imported:
+            _save_local_contacts(contacts)
+        return {"imported": imported, "failed": 0, "total": len(parsed)}
+    try:
+        base_url = _carddav_base_url(cfg)
+    except ValueError as e:
+        logger.warning("CardDAV import URL rejected: %s", e)
+        return {"imported": 0, "failed": 0, "total": 0, "error": str(e)}
+    auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
+    # Split into individual cards. re.split drops the BEGIN line, so we
+    # re-add it. Normalize CRLF.
+    raw = (text or "").replace("\r\n", "\n").replace("\r", "\n")
+    blocks = []
+    for chunk in raw.split("BEGIN:VCARD"):
+        chunk = chunk.strip()
+        if not chunk:
+            continue
+        # Trim anything after END:VCARD (defensive).
+        end = chunk.upper().find("END:VCARD")
+        body = chunk[: end + len("END:VCARD")] if end != -1 else chunk
+        blocks.append("BEGIN:VCARD\n" + body)
+    imported = 0
+    failed = 0
+    for block in blocks:
+        # Extract or assign a UID.
+        m = re.search(r"^UID:(.+)$", block, re.MULTILINE)
+        uid = (m.group(1).strip() if m else "") or str(uuid.uuid4())
+        if not m:
+            # Inject a UID right after the VERSION line (or after BEGIN).
+            if re.search(r"^VERSION:", block, re.MULTILINE):
+                block = re.sub(r"(^VERSION:.*$)", r"\1\nUID:" + uid, block, count=1, flags=re.MULTILINE)
+            else:
+                block = block.replace("BEGIN:VCARD", f"BEGIN:VCARD\nVERSION:4.0\nUID:{uid}", 1)
+        elif not re.search(r"^VERSION:", block, re.MULTILINE):
+            block = block.replace("BEGIN:VCARD", "BEGIN:VCARD\nVERSION:4.0", 1)
+        vcard = block.replace("\n", "\r\n") + "\r\n"
+        url = base_url + "/" + quote(uid, safe="") + ".vcf"
+        try:
+            r = httpx.put(
+                url, data=vcard.encode("utf-8"),
+                headers={"Content-Type": "text/vcard; charset=utf-8"},
+                auth=auth, timeout=15,
+            )
+            if r.status_code in (200, 201, 204):
+                imported += 1
+            else:
+                failed += 1
+                logger.warning(f"Import PUT {uid} returned {r.status_code}: {r.text[:120]}")
+        except Exception as e:
+            failed += 1
+            logger.error(f"Import PUT {uid} failed: {e}")
+    if imported:
+        _contact_cache["fetched_at"] = None
+    return {"imported": imported, "failed": failed, "total": len(blocks)}
+
+
+def _import_csv_contacts(text: str) -> Dict:
+    """Import contacts from CSV. Supports common headers:
+    name/full_name/display_name, email/email_address/e-mail, phone/tel.
+    Falls back to first columns as name,email,phone when no headers exist."""
+    raw = (text or "").strip()
+    if not raw:
+        return {"imported": 0, "failed": 0, "total": 0, "error": "No CSV data found"}
+
+    try:
+        sample = raw[:2048]
+        dialect = csv.Sniffer().sniff(sample)
+    except Exception:
+        dialect = csv.excel
+
+    stream = io.StringIO(raw)
+    try:
+        has_header = csv.Sniffer().has_header(raw[:2048])
+    except Exception:
+        has_header = True
+
+    rows = []
+    if has_header:
+        reader = csv.DictReader(stream, dialect=dialect)
+        for row in reader:
+            lowered = {str(k or "").strip().lower(): (v or "").strip() for k, v in row.items()}
+            name = (
+                lowered.get("name") or lowered.get("full name") or lowered.get("full_name")
+                or lowered.get("display name") or lowered.get("display_name")
+                or lowered.get("fn") or ""
+            )
+            email = (
+                lowered.get("email") or lowered.get("email address")
+                or lowered.get("email_address") or lowered.get("e-mail")
+                or lowered.get("mail") or ""
+            )
+            phone = lowered.get("phone") or lowered.get("telephone") or lowered.get("tel") or ""
+            rows.append((name, email, phone))
+    else:
+        stream.seek(0)
+        reader = csv.reader(stream, dialect=dialect)
+        for row in reader:
+            cols = [(c or "").strip() for c in row]
+            if not any(cols):
+                continue
+            rows.append((
+                cols[0] if len(cols) > 0 else "",
+                cols[1] if len(cols) > 1 else "",
+                cols[2] if len(cols) > 2 else "",
+            ))
+
+    imported = 0
+    failed = 0
+    total = 0
+    existing_emails = {
+        e.lower()
+        for c in _fetch_contacts()
+        for e in (c.get("emails") or [])
+        if e
+    }
+    for name, email, phone in rows:
+        email = (email or "").strip()
+        name = (name or "").strip() or (email.split("@")[0] if email else "")
+        if not email:
+            continue
+        total += 1
+        if email.lower() in existing_emails:
+            continue
+        ok = _create_contact(name, email)
+        if ok:
+            imported += 1
+            existing_emails.add(email.lower())
+            # If the CSV had a phone number, rewrite the just-created row
+            # through the richer update path so phone lands in CardDAV too.
+            if phone:
+                try:
+                    contacts = _fetch_contacts(force=True)
+                    created = next((c for c in contacts if email.lower() in [e.lower() for e in c.get("emails", [])]), None)
+                    if created and created.get("uid"):
+                        _update_contact(created["uid"], name, [email], [phone])
+                except Exception:
+                    pass
+        else:
+            failed += 1
+
+    if imported:
+        _contact_cache["fetched_at"] = None
+    return {"imported": imported, "failed": failed, "total": total}
+
+
+def _contacts_to_vcf(contacts: List[Dict]) -> str:
+    return "".join(
+        _build_vcard(
+            c.get("name") or ((c.get("emails") or [""])[0].split("@")[0] if c.get("emails") else "Contact"),
+            "",
+            uid=c.get("uid") or str(uuid.uuid4()),
+            emails=c.get("emails") or [],
+            phones=c.get("phones") or [],
+        )
+        for c in contacts
+    )
+
+
+def _contacts_to_csv(contacts: List[Dict]) -> str:
+    out = io.StringIO()
+    writer = csv.writer(out)
+    writer.writerow(["name", "email", "phone"])
+    for c in contacts:
+        emails = c.get("emails") or [""]
+        phones = c.get("phones") or [""]
+        max_len = max(len(emails), len(phones), 1)
+        for i in range(max_len):
+            writer.writerow([
+                c.get("name") or "",
+                emails[i] if i < len(emails) else "",
+                phones[i] if i < len(phones) else "",
+            ])
+    return out.getvalue()
+
+
+def _update_contact(uid: str, name: str, emails: List[str], phones: List[str], address: str = "") -> bool:
+    """Rewrite an existing contact via CardDAV or local contacts."""
+    cfg = _get_carddav_config()
+    if not _carddav_configured(cfg):
+        contacts = _load_local_contacts()
+        found = False
+        out = []
+        for c in contacts:
+            if c.get("uid") == uid:
+                # Preserve existing address when caller passes "" (only
+                # updating name/emails/phones, not touching address).
+                addr = address if address else c.get("address", "")
+                out.append(_normalize_contact({"uid": uid, "name": name, "emails": emails, "phones": phones, "address": addr}))
+                found = True
+            else:
+                out.append(c)
+        if not found:
+            out.append(_normalize_contact({"uid": uid, "name": name, "emails": emails, "phones": phones, "address": address}))
+        _save_local_contacts(out)
+        return True
+
+    vcard = _build_vcard(name, "", uid=uid, emails=emails, phones=phones, address=address)
+    # Use the real resource href (handles externally-created contacts whose
+    # filename != UID); falls back to the <uid>.vcf guess.
+    try:
+        url = _resolve_resource_url(uid)
+        auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
+        r = httpx.put(
+            url,
+            data=vcard.encode("utf-8"),
+            headers={"Content-Type": "text/vcard; charset=utf-8"},
+            auth=auth,
+            timeout=10,
+        )
+        if r.status_code in (200, 201, 204):
+            _contact_cache["fetched_at"] = None
+            return True
+        logger.warning(f"CardDAV update PUT returned {r.status_code}: {r.text[:200]}")
+        return False
+    except Exception as e:
+        logger.error(f"Failed to update contact: {e}")
+        return False
+
+
+def _delete_contact(uid: str) -> bool:
+    """Delete a contact via CardDAV or local contacts."""
+    cfg = _get_carddav_config()
+    if not _carddav_configured(cfg):
+        contacts = _load_local_contacts()
+        remaining = [c for c in contacts if c.get("uid") != uid]
+        _save_local_contacts(remaining)
+        return True
+
+    try:
+        url = _resolve_resource_url(uid)
+        auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
+        r = httpx.delete(url, auth=auth, timeout=10)
+        if r.status_code in (200, 204, 404):
+            # Invalidate cache so the next fetch sees the server truth.
+            _contact_cache["fetched_at"] = None
+            # Verify: force a fresh fetch and check the UID is actually gone.
+            # A 404 on the guessed URL ({uid}.vcf) can mean the contact
+            # lives at a different resource URL — the DELETE missed it but
+            # we'd silently report success. This check catches that.
+            fresh = _fetch_contacts(force=True)
+            still_there = any(c.get("uid") == uid for c in fresh)
+            if still_there:
+                logger.warning(
+                    f"CardDAV DELETE reported success for {uid} "
+                    f"but UID still present after re-fetch — "
+                    f"resource URL may differ from {redact_url(url)}"
+                )
+                return False
+            if r.status_code == 404:
+                logger.info(f"CardDAV DELETE 404 for {uid} — already gone")
+            return True
+        logger.warning(f"CardDAV DELETE returned {r.status_code}: {r.text[:200]}")
+        return False
+    except Exception as e:
+        logger.error(f"Failed to delete contact: {e}")
+        return False
+
+
+# ── Routes ──
+
+def setup_contacts_routes():
+    router = APIRouter(prefix="/api/contacts", tags=["contacts"])
+
+    @router.get("/list")
+    async def list_contacts(_admin: str = Depends(require_admin)):
+        """List all contacts."""
+        contacts = _fetch_contacts()
+        return {"contacts": contacts, "count": len(contacts)}
+
+    @router.get("/search")
+    async def search_contacts(q: str = Query(""), _admin: str = Depends(require_admin)):
+        """Search contacts by name or email. Returns up to 10 matches."""
+        contacts = _fetch_contacts()
+        if not q:
+            return {"results": []}
+        q_lower = q.lower()
+        results = []
+        for c in contacts:
+            if q_lower in c["name"].lower():
+                results.append(c)
+                continue
+            for em in c["emails"]:
+                if q_lower in em.lower():
+                    results.append(c)
+                    break
+        return {"results": results[:10]}
+
+    @router.post("/add")
+    async def add_contact(data: dict, _admin: str = Depends(require_admin)):
+        """Add a new contact."""
+        name = (data.get("name") or "").strip()
+        email = (data.get("email") or "").strip()
+        phone = (data.get("phone") or "").strip()
+        address = (data.get("address") or "").strip()
+        if not email:
+            return {"success": False, "error": "Email required"}
+        # Check if already exists by email
+        if email:
+            contacts = _fetch_contacts()
+            for c in contacts:
+                if email.lower() in [e.lower() for e in c["emails"]]:
+                    return {"success": True, "message": "Already exists", "contact": c}
+        if not name:
+            name = email.split("@")[0]
+        create_params = inspect.signature(_create_contact).parameters
+        if len(create_params) >= 3:
+            ok = _create_contact(name, email, address)
+        else:
+            ok = _create_contact(name, email)
+        # If a phone was provided, do an immediate update to thread it
+        # through (the simple _create_contact signature only takes name +
+        # email + address; phones happen via update).
+        if ok and phone:
+            try:
+                fresh = _fetch_contacts(force=True)
+                created = next((c for c in fresh if name == c.get("name") and (not email or email in c.get("emails", []))), None)
+                if created:
+                    _update_contact(
+                        created["uid"], name,
+                        created.get("emails", []),
+                        [phone],
+                        address,
+                    )
+            except Exception:
+                pass
+        return {"success": ok}
+
+    @router.post("/import")
+    async def import_vcf(data: dict, _admin: str = Depends(require_admin)):
+        """Import contacts from .vcf or CSV. Body: {"vcf": "..."} or {"csv": "..."}."""
+        # Coerce defensively: a non-string vcf/text/csv (e.g. a number or list
+        # in the JSON body) would otherwise reach .strip() and 500 with an
+        # AttributeError instead of degrading to a clean "no data" response.
+        text = str(data.get("vcf") or data.get("text") or "")
+        csv_text = str(data.get("csv") or "")
+        if text.strip():
+            if "BEGIN:VCARD" not in text.upper():
+                return {"success": False, "error": "No vCard data found"}
+            result = _import_vcards(text)
+        elif csv_text.strip():
+            result = _import_csv_contacts(csv_text)
+        else:
+            return {"success": False, "error": "No contact data found"}
+        result["success"] = result.get("imported", 0) > 0
+        return result
+
+    @router.get("/export")
+    async def export_contacts(
+        format: str = Query("vcf", pattern="^(vcf|csv)$"),
+        _admin: str = Depends(require_admin),
+    ):
+        """Export all contacts as vCard or CSV."""
+        contacts = _fetch_contacts(force=True)
+        if format == "csv":
+            content = _contacts_to_csv(contacts)
+            media_type = "text/csv; charset=utf-8"
+            filename = "odysseus-contacts.csv"
+        else:
+            content = _contacts_to_vcf(contacts)
+            media_type = "text/vcard; charset=utf-8"
+            filename = "odysseus-contacts.vcf"
+        return Response(
+            content=content,
+            media_type=media_type,
+            headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+        )
+
+    @router.get("/config")
+    async def get_config(_admin: str = Depends(require_admin)):
+        cfg = _get_carddav_config()
+        # Mask password
+        if cfg["password"]:
+            cfg["password"] = "***"
+        return cfg
+
+    @router.put("/config")
+    async def update_config(data: dict, _admin: str = Depends(require_admin)):
+        settings = _load_settings()
+        for key in ("carddav_url", "carddav_username", "carddav_password"):
+            if key in data:
+                if key == "carddav_url" and str(data[key] or "").strip():
+                    try:
+                        settings[key] = _validate_carddav_url(data[key])
+                    except ValueError as e:
+                        raise HTTPException(400, str(e))
+                else:
+                    value = data[key]
+                    if key == "carddav_password" and value:
+                        from src.secret_storage import encrypt
+                        value = encrypt(value)
+                    settings[key] = value
+        _save_settings(settings)
+        # Force re-fetch
+        _contact_cache["fetched_at"] = None
+        return {"success": True}
+
+    @router.delete("/clear")
+    async def clear_contacts(_admin: str = Depends(require_admin)):
+        """Clear all local contacts. If CardDAV is configured, only clears the local fallback cache."""
+        _save_local_contacts([])
+        return {"success": True}
+
+    # NOTE: the /{uid} routes are declared LAST so the literal paths above
+    # (/list, /search, /add, /config) win — otherwise PUT /config would
+    # match PUT /{uid} with uid="config".
+    @router.put("/{uid}")
+    async def edit_contact(uid: str, data: dict, _admin: str = Depends(require_admin)):
+        """Edit an existing contact — name / emails / phones / address."""
+        name = (data.get("name") or "").strip()
+        emails = data.get("emails")
+        phones = data.get("phones")
+        if emails is None and data.get("email"):
+            emails = [data["email"]]
+        emails = [e.strip() for e in (emails or []) if e and e.strip()]
+        phones = [p.strip() for p in (phones or []) if p and p.strip()]
+        address = (data.get("address") or "").strip()
+        if not name and not emails and not address:
+            return {"success": False, "error": "Name, email, or address required"}
+        if not name and emails:
+            name = emails[0].split("@")[0]
+        ok = _update_contact(uid, name, emails, phones, address)
+        return {"success": ok}
+
+    @router.delete("/{uid}")
+    async def delete_contact(uid: str, _admin: str = Depends(require_admin)):
+        """Delete a contact by UID."""
+        if not uid:
+            return {"success": False, "error": "UID required"}
+        ok = _delete_contact(uid)
+        return {"success": ok}
+
+    return router

routes/contacts_routes.py

@@ -1,900 +1,20 @@
-"""
-contacts_routes.py
+"""Backward-compat shim — canonical location is routes/contacts/contacts_routes.py.
 
-CardDAV contacts integration. Reads from local Radicale, supports
-search and adding new contacts.
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.contacts_routes``, ``from routes.contacts_routes import X``,
+``importlib.import_module("routes.contacts_routes")``, and the string-targeted
+``monkeypatch.setattr("routes.contacts_routes.SETTINGS_FILE", ...)`` pattern
+used by test_carddav_password_encryption.py / test_contacts_carddav_security.py
+— plus the ``import ... as contacts_routes`` + ``setattr(...)`` pattern in
+test_contacts_add_null_name.py — all operate on the *same* object the
+application actually uses. This also keeps ``_contact_cache`` (mutable module
+state) identical across import paths. Keeps existing import paths working
+after slice 2e (#4082/#4071). No source-introspection tests read this file
+by path.
 """
 
-import re
-import logging
-import uuid
-import json
-import csv
-import io
-import os
-import inspect
-import httpx
-from pathlib import Path
-from datetime import datetime
-from urllib.parse import urljoin, urlparse, urlunparse
+import sys as _sys
 
-from core.log_safety import redact_url
-from fastapi import APIRouter, Query, Depends, Response, HTTPException
-from typing import List, Dict, Optional
+from routes.contacts import contacts_routes as _canonical  # noqa: F401
 
-from core.middleware import require_admin
-from src.url_safety import check_outbound_url
-
-logger = logging.getLogger(__name__)
-
-from src.constants import DATA_DIR as _DATA_DIR, SETTINGS_FILE as _SETTINGS_FILE, CONTACTS_FILE as _CONTACTS_FILE
-DATA_DIR = Path(_DATA_DIR)
-SETTINGS_FILE = Path(_SETTINGS_FILE)
-LOCAL_CONTACTS_FILE = Path(_CONTACTS_FILE)
-
-
-def _load_settings():
-    if SETTINGS_FILE.exists():
-        return json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
-    return {}
-
-
-def _save_settings(settings):
-    from core.atomic_io import atomic_write_json
-    atomic_write_json(str(SETTINGS_FILE), settings, indent=2)
-
-
-def _get_carddav_config():
-    import os
-    settings = _load_settings()
-    password = settings.get("carddav_password", os.environ.get("CARDDAV_PASSWORD", ""))
-    if password and "carddav_password" in settings:
-        from src.secret_storage import decrypt
-        password = decrypt(password)
-    return {
-        "url": settings.get("carddav_url", os.environ.get("CARDDAV_URL", "")),
-        "username": settings.get("carddav_username", os.environ.get("CARDDAV_USERNAME", "")),
-        "password": password,
-    }
-
-
-def _carddav_configured(cfg: Optional[Dict] = None) -> bool:
-    cfg = cfg or _get_carddav_config()
-    return bool((cfg.get("url") or "").strip())
-
-
-def _validate_carddav_url(url: str) -> str:
-    cleaned = (url if isinstance(url, str) else "").strip().rstrip("/")
-    ok, reason = check_outbound_url(
-        cleaned,
-        block_private=os.getenv("CARDDAV_BLOCK_PRIVATE_IPS", "false").lower() == "true",
-    )
-    if not ok:
-        raise ValueError(f"Rejected CardDAV URL: {reason}")
-    return cleaned
-
-
-def _carddav_base_url(cfg: Dict) -> str:
-    return _validate_carddav_url(cfg.get("url") or "")
-
-
-def _normalize_contact(contact: Dict) -> Dict:
-    emails = []
-    for e in contact.get("emails") or ([] if not contact.get("email") else [contact.get("email")]):
-        e = str(e or "").strip()
-        if e and e not in emails:
-            emails.append(e)
-    phones = []
-    for p in contact.get("phones") or ([] if not contact.get("phone") else [contact.get("phone")]):
-        p = str(p or "").strip()
-        if p and p not in phones:
-            phones.append(p)
-    name = str(contact.get("name") or "").strip()
-    if not name and emails:
-        name = emails[0].split("@")[0]
-    address = str(contact.get("address") or "").strip()
-    return {
-        "uid": str(contact.get("uid") or uuid.uuid4()),
-        "name": name,
-        "emails": emails,
-        "phones": phones,
-        "address": address,
-    }
-
-
-def _load_local_contacts() -> List[Dict]:
-    try:
-        if not LOCAL_CONTACTS_FILE.exists():
-            return []
-        data = json.loads(LOCAL_CONTACTS_FILE.read_text(encoding="utf-8"))
-        rows = data.get("contacts", data) if isinstance(data, dict) else data
-        return [_normalize_contact(c) for c in (rows or []) if isinstance(c, dict)]
-    except Exception as e:
-        logger.error(f"Failed to load local contacts: {e}")
-        return []
-
-
-def _save_local_contacts(contacts: List[Dict]) -> None:
-    from core.atomic_io import atomic_write_json
-    DATA_DIR.mkdir(parents=True, exist_ok=True)
-    atomic_write_json(str(LOCAL_CONTACTS_FILE), {"contacts": [_normalize_contact(c) for c in contacts]}, indent=2)
-    _contact_cache["contacts"] = [_normalize_contact(c) for c in contacts]
-    _contact_cache["fetched_at"] = datetime.utcnow()
-
-
-# ── vCard parsing ──
-
-def _vunesc(value: str) -> str:
-    """Reverse _vesc() — turn escaped vCard text back into the raw value.
-    Order matters: handle \\n/\\, /\\; first, backslash-unescape last."""
-    if not value:
-        return value
-    out = []
-    i = 0
-    while i < len(value):
-        ch = value[i]
-        if ch == "\\" and i + 1 < len(value):
-            nxt = value[i + 1]
-            if nxt in ("n", "N"):
-                out.append("\n")
-            elif nxt in (",", ";", "\\"):
-                out.append(nxt)
-            else:
-                out.append(nxt)
-            i += 2
-        else:
-            out.append(ch)
-            i += 1
-    return "".join(out)
-
-
-def _parse_vcards(text: str) -> List[Dict]:
-    """Parse a stream of vCards into dicts with name, email, phone."""
-    # Unfold RFC 6350 3.2 line folding first: a CRLF/LF followed by a single
-    # space or tab is a continuation of the previous logical line. Real
-    # CardDAV servers (Radicale, iCloud, Apple/Google) fold long EMAIL / FN /
-    # PHOTO lines, and splitting on raw newlines without unfolding dropped the
-    # continuation (e.g. "...@example\n .com" lost the ".com"), truncating the
-    # email/name.
-    text = re.sub(r"\r\n[ \t]", "", text or "")
-    text = re.sub(r"\n[ \t]", "", text)
-    contacts = []
-    for block in re.split(r"BEGIN:VCARD", text):
-        if not block.strip():
-            continue
-        contact = {"name": "", "emails": [], "phones": [], "uid": "", "address": ""}
-        for line in block.split("\n"):
-            line = line.strip()
-            # Strip an optional RFC 6350 group prefix (e.g. "item1.EMAIL;...")
-            # that Apple Contacts / iCloud / many CardDAV servers emit by
-            # default — without this the property-name checks below miss those
-            # lines and silently drop the email / phone. The group token only
-            # precedes the property name, so it is safe to strip for matching
-            # and value extraction, and a no-op for non-grouped lines.
-            name_part = re.sub(r"^[A-Za-z0-9-]+\.", "", line, count=1)
-            if name_part.startswith("FN:") or name_part.startswith("FN;"):
-                contact["name"] = _vunesc(name_part.split(":", 1)[1]) if ":" in name_part else ""
-            elif name_part.startswith("EMAIL"):
-                # Handle EMAIL:foo@bar OR EMAIL;TYPE=...:foo@bar OR EMAIL;PREF=1:foo@bar
-                if ":" in name_part:
-                    email_addr = _vunesc(name_part.split(":", 1)[1])
-                    if email_addr and email_addr not in contact["emails"]:
-                        contact["emails"].append(email_addr)
-            elif name_part.startswith("TEL"):
-                if ":" in name_part:
-                    phone = _vunesc(name_part.split(":", 1)[1])
-                    if phone and phone not in contact["phones"]:
-                        contact["phones"].append(phone)
-            elif name_part.startswith("ADR"):
-                # vCard ADR is 7 semicolon-separated components:
-                # post-office-box;extended-address;street;locality;region;postal-code;country.
-                # Recover a human-readable string by joining non-empty
-                # components with ", ".
-                if ":" in name_part:
-                    raw = name_part.split(":", 1)[1]
-                    parts = [_vunesc(p).strip() for p in raw.split(";")]
-                    contact["address"] = ", ".join(p for p in parts if p)
-            elif name_part.startswith("UID:"):
-                contact["uid"] = _vunesc(name_part[4:])
-        if contact["name"] or contact["emails"]:
-            contacts.append(contact)
-    return contacts
-
-
-def _vesc(value: str) -> str:
-    """Escape a vCard property VALUE per RFC 6350 §3.4: backslash, comma,
-    semicolon, and newlines. Without this, a name like 'Sekisui House,Ltd'
-    or any value containing a newline produces a malformed vCard (broken
-    N/FN fields) or could inject arbitrary properties."""
-    return (
-        (value or "")
-        .replace("\\", "\\\\")
-        .replace("\n", "\\n")
-        .replace("\r", "")
-        .replace(",", "\\,")
-        .replace(";", "\\;")
-    )
-
-
-def _build_vcard(name: str, email: str, uid: Optional[str] = None,
-                 emails: Optional[List[str]] = None,
-                 phones: Optional[List[str]] = None,
-                 address: Optional[str] = None) -> str:
-    """Build a vCard. Accepts either a single `email` (legacy callers) or
-    full `emails`/`phones` lists (edit path). The first email is marked
-    PREF=1. All values are RFC-6350-escaped."""
-    if not uid:
-        uid = str(uuid.uuid4())
-    # Normalize email lists — `email` arg is a convenience for single-email
-    # creation; `emails` (if given) is authoritative.
-    email_list = [e.strip() for e in (emails if emails is not None else ([email] if email else [])) if e and e.strip()]
-    phone_list = [p.strip() for p in (phones or []) if p and p.strip()]
-    # Try to split name into first/last
-    parts = name.strip().split()
-    if len(parts) >= 2:
-        first = parts[0]
-        last = " ".join(parts[1:])
-    else:
-        first = name
-        last = ""
-    # N field is structured (5 components separated by ';') — escape each
-    # component individually so a comma in the name doesn't split it.
-    n_field = f"{_vesc(last)};{_vesc(first)};;;"
-    lines = [
-        "BEGIN:VCARD",
-        "VERSION:4.0",
-        f"UID:{_vesc(uid)}",
-        f"FN:{_vesc(name)}",
-        f"N:{n_field}",
-    ]
-    for i, em in enumerate(email_list):
-        # First email is the preferred one.
-        lines.append(f"EMAIL;PREF=1:{_vesc(em)}" if i == 0 else f"EMAIL:{_vesc(em)}")
-    for ph in phone_list:
-        lines.append(f"TEL:{_vesc(ph)}")
-    # Address: stuff the whole human-readable string into the street
-    # component of ADR. vCard ADR has 7 semicolon-separated components:
-    # post-office-box;extended-address;street;locality;region;postal-code;country.
-    addr = (address or "").strip()
-    if addr:
-        lines.append(f"ADR:;;{_vesc(addr)};;;;")
-    lines.append("END:VCARD")
-    return "\r\n".join(lines) + "\r\n"
-
-
-# ── In-memory cache ──
-
-_contact_cache = {"contacts": [], "fetched_at": None}
-
-
-def _abs_url(href: str) -> str:
-    """Combine a multistatus <href> (an absolute path like
-    /user/contacts/x.vcf) with the configured CardDAV server origin so we
-    get a fully-qualified URL to PUT/DELETE. Absolute hrefs are accepted only
-    for the configured origin; a cross-origin href is treated as a path on the
-    configured server so a malicious CardDAV response cannot redirect later
-    writes/deletes to cloud metadata or another host."""
-    cfg = _get_carddav_config()
-    base = _carddav_base_url(cfg)
-    base_p = urlparse(base)
-    joined = urljoin(base.rstrip("/") + "/", href or "")
-    joined_p = urlparse(joined)
-    if (joined_p.scheme, joined_p.netloc) != (base_p.scheme, base_p.netloc):
-        joined = urlunparse((base_p.scheme, base_p.netloc, joined_p.path or "/", "", joined_p.query, ""))
-    return _validate_carddav_url(joined)
-
-
-# CardDAV REPORT body — pull every card's etag + raw vCard in ONE request,
-# alongside the resource href. Lets us map each contact's UID to the real
-# server resource path (which is NOT always <uid>.vcf for contacts created
-# by other clients).
-_ADDRESSBOOK_QUERY = (
-    '<?xml version="1.0" encoding="utf-8"?>'
-    '<C:addressbook-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:carddav">'
-    '<D:prop><D:getetag/><C:address-data/></D:prop>'
-    '<C:filter/>'
-    '</C:addressbook-query>'
-)
-
-
-def _fetch_via_report(cfg, auth):
-    """Try a CardDAV REPORT addressbook-query — returns contacts WITH an
-    `href` field, or None if the server doesn't support it / errors."""
-    from defusedxml import ElementTree as ET
-    try:
-        r = httpx.request(
-            "REPORT", cfg["url"],
-            content=_ADDRESSBOOK_QUERY.encode("utf-8"),
-            headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "1"},
-            auth=auth, timeout=10,
-        )
-        if r.status_code not in (207, 200):
-            return None
-        root = ET.fromstring(r.text)
-        ns = {"D": "DAV:", "C": "urn:ietf:params:xml:ns:carddav"}
-        out = []
-        for resp in root.findall("D:response", ns):
-            href_el = resp.find("D:href", ns)
-            data_el = resp.find(".//C:address-data", ns)
-            if href_el is None or data_el is None or not (data_el.text or "").strip():
-                continue
-            parsed = _parse_vcards(data_el.text)
-            if not parsed:
-                continue
-            c = parsed[0]
-            c["href"] = href_el.text.strip()
-            out.append(c)
-        # If the REPORT parsed to ZERO contacts, don't trust it — some
-        # CardDAV servers treat an empty <filter/> as "match nothing" and
-        # return a valid-but-empty 207. Return None so the caller falls
-        # back to the plain GET (which lists everything). A genuinely empty
-        # address book just costs one extra GET that also returns nothing.
-        if not out:
-            return None
-        return out
-    except Exception as e:
-        logger.warning(f"CardDAV REPORT failed, falling back to GET: {e}")
-        return None
-
-
-def _fetch_contacts(force=False):
-    """Fetch all contacts. Uses CardDAV when configured, otherwise local JSON."""
-    if not force and _contact_cache["fetched_at"]:
-        age = (datetime.utcnow() - _contact_cache["fetched_at"]).total_seconds()
-        if age < 60:
-            return _contact_cache["contacts"]
-
-    cfg = _get_carddav_config()
-    if not _carddav_configured(cfg):
-        contacts = _load_local_contacts()
-        _contact_cache["contacts"] = contacts
-        _contact_cache["fetched_at"] = datetime.utcnow()
-        return contacts
-
-    try:
-        cfg["url"] = _carddav_base_url(cfg)
-        auth = None
-        if cfg["username"]:
-            auth = (cfg["username"], cfg["password"])
-        # Preferred path: REPORT gives us hrefs for reliable edit/delete.
-        contacts = _fetch_via_report(cfg, auth)
-        if contacts is None:
-            # Fallback: plain GET, concatenated vCards, no hrefs.
-            r = httpx.get(cfg["url"], auth=auth, timeout=10)
-            if r.status_code != 200:
-                logger.warning(f"CardDAV returned {r.status_code}")
-                return _contact_cache["contacts"]
-            contacts = _parse_vcards(r.text)
-        _contact_cache["contacts"] = contacts
-        _contact_cache["fetched_at"] = datetime.utcnow()
-        return contacts
-    except Exception as e:
-        logger.error(f"Failed to fetch contacts: {e}")
-        return _contact_cache["contacts"]
-
-
-def _resolve_resource_url(uid: str) -> str:
-    """Map a contact UID to its real CardDAV resource URL. Uses the href
-    captured during fetch when available (handles contacts whose filename
-    != UID); falls back to the <uid>.vcf guess for app-created contacts or
-    when no href is known."""
-    def _lookup():
-        for c in _contact_cache.get("contacts", []):
-            if c.get("uid") == uid and c.get("href"):
-                return _abs_url(c["href"])
-        return None
-    found = _lookup()
-    if found:
-        return found
-    # Not in cache (or no href) — refresh once and retry before guessing.
-    try:
-        _fetch_contacts(force=True)
-    except Exception:
-        pass
-    return _lookup() or _vcard_url(uid)
-
-
-def _create_contact(name: str, email: str, address: str = "") -> bool:
-    """Add a new contact via CardDAV or local contacts."""
-    cfg = _get_carddav_config()
-    if not _carddav_configured(cfg):
-        contacts = _load_local_contacts()
-        email_l = (email or "").strip().lower()
-        for c in contacts:
-            if email_l and email_l in [e.lower() for e in c.get("emails", [])]:
-                return True
-        contacts.append(_normalize_contact({"name": name, "emails": [email], "address": address}))
-        _save_local_contacts(contacts)
-        return True
-
-    contact_uid = str(uuid.uuid4())
-    vcard = _build_vcard(name, email, contact_uid, address=address)
-    try:
-        url = _carddav_base_url(cfg) + "/" + contact_uid + ".vcf"
-        auth = None
-        if cfg["username"]:
-            auth = (cfg["username"], cfg["password"])
-        r = httpx.put(
-            url,
-            data=vcard.encode("utf-8"),
-            headers={"Content-Type": "text/vcard; charset=utf-8"},
-            auth=auth,
-            timeout=10,
-        )
-        if r.status_code in (200, 201, 204):
-            # Invalidate cache
-            _contact_cache["fetched_at"] = None
-            return True
-        logger.warning(f"CardDAV PUT returned {r.status_code}: {r.text[:200]}")
-        return False
-    except Exception as e:
-        logger.error(f"Failed to create contact: {e}")
-        return False
-
-
-def _vcard_url(uid: str) -> str:
-    """The CardDAV resource URL for a given contact UID. The uid is URL-
-    encoded so a value containing '/', '..' or other path chars can't
-    escape the collection and target an arbitrary CardDAV resource."""
-    from urllib.parse import quote
-    cfg = _get_carddav_config()
-    return _carddav_base_url(cfg) + "/" + quote(uid, safe="") + ".vcf"
-
-
-def _import_vcards(text: str) -> Dict:
-    """Import a (possibly multi-card) .vcf blob. Each card is PUT to the
-    CardDAV server PRESERVING its full original content (ADR/ORG/photo/
-    etc.) — we don't rebuild it, just ensure it has VERSION + UID and
-    normalize line endings. Returns {imported, failed, total}."""
-    from urllib.parse import quote
-    cfg = _get_carddav_config()
-    if not cfg.get("url"):
-        parsed = _parse_vcards(text)
-        contacts = _load_local_contacts()
-        existing = {
-            e.lower()
-            for c in contacts
-            for e in (c.get("emails") or [])
-            if e
-        }
-        imported = 0
-        for c in parsed:
-            emails = [e for e in (c.get("emails") or []) if e]
-            if emails and any(e.lower() in existing for e in emails):
-                continue
-            contacts.append(_normalize_contact(c))
-            for e in emails:
-                existing.add(e.lower())
-            imported += 1
-        if imported:
-            _save_local_contacts(contacts)
-        return {"imported": imported, "failed": 0, "total": len(parsed)}
-    try:
-        base_url = _carddav_base_url(cfg)
-    except ValueError as e:
-        logger.warning("CardDAV import URL rejected: %s", e)
-        return {"imported": 0, "failed": 0, "total": 0, "error": str(e)}
-    auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
-    # Split into individual cards. re.split drops the BEGIN line, so we
-    # re-add it. Normalize CRLF.
-    raw = (text or "").replace("\r\n", "\n").replace("\r", "\n")
-    blocks = []
-    for chunk in raw.split("BEGIN:VCARD"):
-        chunk = chunk.strip()
-        if not chunk:
-            continue
-        # Trim anything after END:VCARD (defensive).
-        end = chunk.upper().find("END:VCARD")
-        body = chunk[: end + len("END:VCARD")] if end != -1 else chunk
-        blocks.append("BEGIN:VCARD\n" + body)
-    imported = 0
-    failed = 0
-    for block in blocks:
-        # Extract or assign a UID.
-        m = re.search(r"^UID:(.+)$", block, re.MULTILINE)
-        uid = (m.group(1).strip() if m else "") or str(uuid.uuid4())
-        if not m:
-            # Inject a UID right after the VERSION line (or after BEGIN).
-            if re.search(r"^VERSION:", block, re.MULTILINE):
-                block = re.sub(r"(^VERSION:.*$)", r"\1\nUID:" + uid, block, count=1, flags=re.MULTILINE)
-            else:
-                block = block.replace("BEGIN:VCARD", f"BEGIN:VCARD\nVERSION:4.0\nUID:{uid}", 1)
-        elif not re.search(r"^VERSION:", block, re.MULTILINE):
-            block = block.replace("BEGIN:VCARD", "BEGIN:VCARD\nVERSION:4.0", 1)
-        vcard = block.replace("\n", "\r\n") + "\r\n"
-        url = base_url + "/" + quote(uid, safe="") + ".vcf"
-        try:
-            r = httpx.put(
-                url, data=vcard.encode("utf-8"),
-                headers={"Content-Type": "text/vcard; charset=utf-8"},
-                auth=auth, timeout=15,
-            )
-            if r.status_code in (200, 201, 204):
-                imported += 1
-            else:
-                failed += 1
-                logger.warning(f"Import PUT {uid} returned {r.status_code}: {r.text[:120]}")
-        except Exception as e:
-            failed += 1
-            logger.error(f"Import PUT {uid} failed: {e}")
-    if imported:
-        _contact_cache["fetched_at"] = None
-    return {"imported": imported, "failed": failed, "total": len(blocks)}
-
-
-def _import_csv_contacts(text: str) -> Dict:
-    """Import contacts from CSV. Supports common headers:
-    name/full_name/display_name, email/email_address/e-mail, phone/tel.
-    Falls back to first columns as name,email,phone when no headers exist."""
-    raw = (text or "").strip()
-    if not raw:
-        return {"imported": 0, "failed": 0, "total": 0, "error": "No CSV data found"}
-
-    try:
-        sample = raw[:2048]
-        dialect = csv.Sniffer().sniff(sample)
-    except Exception:
-        dialect = csv.excel
-
-    stream = io.StringIO(raw)
-    try:
-        has_header = csv.Sniffer().has_header(raw[:2048])
-    except Exception:
-        has_header = True
-
-    rows = []
-    if has_header:
-        reader = csv.DictReader(stream, dialect=dialect)
-        for row in reader:
-            lowered = {str(k or "").strip().lower(): (v or "").strip() for k, v in row.items()}
-            name = (
-                lowered.get("name") or lowered.get("full name") or lowered.get("full_name")
-                or lowered.get("display name") or lowered.get("display_name")
-                or lowered.get("fn") or ""
-            )
-            email = (
-                lowered.get("email") or lowered.get("email address")
-                or lowered.get("email_address") or lowered.get("e-mail")
-                or lowered.get("mail") or ""
-            )
-            phone = lowered.get("phone") or lowered.get("telephone") or lowered.get("tel") or ""
-            rows.append((name, email, phone))
-    else:
-        stream.seek(0)
-        reader = csv.reader(stream, dialect=dialect)
-        for row in reader:
-            cols = [(c or "").strip() for c in row]
-            if not any(cols):
-                continue
-            rows.append((
-                cols[0] if len(cols) > 0 else "",
-                cols[1] if len(cols) > 1 else "",
-                cols[2] if len(cols) > 2 else "",
-            ))
-
-    imported = 0
-    failed = 0
-    total = 0
-    existing_emails = {
-        e.lower()
-        for c in _fetch_contacts()
-        for e in (c.get("emails") or [])
-        if e
-    }
-    for name, email, phone in rows:
-        email = (email or "").strip()
-        name = (name or "").strip() or (email.split("@")[0] if email else "")
-        if not email:
-            continue
-        total += 1
-        if email.lower() in existing_emails:
-            continue
-        ok = _create_contact(name, email)
-        if ok:
-            imported += 1
-            existing_emails.add(email.lower())
-            # If the CSV had a phone number, rewrite the just-created row
-            # through the richer update path so phone lands in CardDAV too.
-            if phone:
-                try:
-                    contacts = _fetch_contacts(force=True)
-                    created = next((c for c in contacts if email.lower() in [e.lower() for e in c.get("emails", [])]), None)
-                    if created and created.get("uid"):
-                        _update_contact(created["uid"], name, [email], [phone])
-                except Exception:
-                    pass
-        else:
-            failed += 1
-
-    if imported:
-        _contact_cache["fetched_at"] = None
-    return {"imported": imported, "failed": failed, "total": total}
-
-
-def _contacts_to_vcf(contacts: List[Dict]) -> str:
-    return "".join(
-        _build_vcard(
-            c.get("name") or ((c.get("emails") or [""])[0].split("@")[0] if c.get("emails") else "Contact"),
-            "",
-            uid=c.get("uid") or str(uuid.uuid4()),
-            emails=c.get("emails") or [],
-            phones=c.get("phones") or [],
-        )
-        for c in contacts
-    )
-
-
-def _contacts_to_csv(contacts: List[Dict]) -> str:
-    out = io.StringIO()
-    writer = csv.writer(out)
-    writer.writerow(["name", "email", "phone"])
-    for c in contacts:
-        emails = c.get("emails") or [""]
-        phones = c.get("phones") or [""]
-        max_len = max(len(emails), len(phones), 1)
-        for i in range(max_len):
-            writer.writerow([
-                c.get("name") or "",
-                emails[i] if i < len(emails) else "",
-                phones[i] if i < len(phones) else "",
-            ])
-    return out.getvalue()
-
-
-def _update_contact(uid: str, name: str, emails: List[str], phones: List[str], address: str = "") -> bool:
-    """Rewrite an existing contact via CardDAV or local contacts."""
-    cfg = _get_carddav_config()
-    if not _carddav_configured(cfg):
-        contacts = _load_local_contacts()
-        found = False
-        out = []
-        for c in contacts:
-            if c.get("uid") == uid:
-                # Preserve existing address when caller passes "" (only
-                # updating name/emails/phones, not touching address).
-                addr = address if address else c.get("address", "")
-                out.append(_normalize_contact({"uid": uid, "name": name, "emails": emails, "phones": phones, "address": addr}))
-                found = True
-            else:
-                out.append(c)
-        if not found:
-            out.append(_normalize_contact({"uid": uid, "name": name, "emails": emails, "phones": phones, "address": address}))
-        _save_local_contacts(out)
-        return True
-
-    vcard = _build_vcard(name, "", uid=uid, emails=emails, phones=phones, address=address)
-    # Use the real resource href (handles externally-created contacts whose
-    # filename != UID); falls back to the <uid>.vcf guess.
-    try:
-        url = _resolve_resource_url(uid)
-        auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
-        r = httpx.put(
-            url,
-            data=vcard.encode("utf-8"),
-            headers={"Content-Type": "text/vcard; charset=utf-8"},
-            auth=auth,
-            timeout=10,
-        )
-        if r.status_code in (200, 201, 204):
-            _contact_cache["fetched_at"] = None
-            return True
-        logger.warning(f"CardDAV update PUT returned {r.status_code}: {r.text[:200]}")
-        return False
-    except Exception as e:
-        logger.error(f"Failed to update contact: {e}")
-        return False
-
-
-def _delete_contact(uid: str) -> bool:
-    """Delete a contact via CardDAV or local contacts."""
-    cfg = _get_carddav_config()
-    if not _carddav_configured(cfg):
-        contacts = _load_local_contacts()
-        remaining = [c for c in contacts if c.get("uid") != uid]
-        _save_local_contacts(remaining)
-        return True
-
-    try:
-        url = _resolve_resource_url(uid)
-        auth = (cfg["username"], cfg["password"]) if cfg["username"] else None
-        r = httpx.delete(url, auth=auth, timeout=10)
-        if r.status_code in (200, 204, 404):
-            # Invalidate cache so the next fetch sees the server truth.
-            _contact_cache["fetched_at"] = None
-            # Verify: force a fresh fetch and check the UID is actually gone.
-            # A 404 on the guessed URL ({uid}.vcf) can mean the contact
-            # lives at a different resource URL — the DELETE missed it but
-            # we'd silently report success. This check catches that.
-            fresh = _fetch_contacts(force=True)
-            still_there = any(c.get("uid") == uid for c in fresh)
-            if still_there:
-                logger.warning(
-                    f"CardDAV DELETE reported success for {uid} "
-                    f"but UID still present after re-fetch — "
-                    f"resource URL may differ from {redact_url(url)}"
-                )
-                return False
-            if r.status_code == 404:
-                logger.info(f"CardDAV DELETE 404 for {uid} — already gone")
-            return True
-        logger.warning(f"CardDAV DELETE returned {r.status_code}: {r.text[:200]}")
-        return False
-    except Exception as e:
-        logger.error(f"Failed to delete contact: {e}")
-        return False
-
-
-# ── Routes ──
-
-def setup_contacts_routes():
-    router = APIRouter(prefix="/api/contacts", tags=["contacts"])
-
-    @router.get("/list")
-    async def list_contacts(_admin: str = Depends(require_admin)):
-        """List all contacts."""
-        contacts = _fetch_contacts()
-        return {"contacts": contacts, "count": len(contacts)}
-
-    @router.get("/search")
-    async def search_contacts(q: str = Query(""), _admin: str = Depends(require_admin)):
-        """Search contacts by name or email. Returns up to 10 matches."""
-        contacts = _fetch_contacts()
-        if not q:
-            return {"results": []}
-        q_lower = q.lower()
-        results = []
-        for c in contacts:
-            if q_lower in c["name"].lower():
-                results.append(c)
-                continue
-            for em in c["emails"]:
-                if q_lower in em.lower():
-                    results.append(c)
-                    break
-        return {"results": results[:10]}
-
-    @router.post("/add")
-    async def add_contact(data: dict, _admin: str = Depends(require_admin)):
-        """Add a new contact."""
-        name = (data.get("name") or "").strip()
-        email = (data.get("email") or "").strip()
-        phone = (data.get("phone") or "").strip()
-        address = (data.get("address") or "").strip()
-        if not email:
-            return {"success": False, "error": "Email required"}
-        # Check if already exists by email
-        if email:
-            contacts = _fetch_contacts()
-            for c in contacts:
-                if email.lower() in [e.lower() for e in c["emails"]]:
-                    return {"success": True, "message": "Already exists", "contact": c}
-        if not name:
-            name = email.split("@")[0]
-        create_params = inspect.signature(_create_contact).parameters
-        if len(create_params) >= 3:
-            ok = _create_contact(name, email, address)
-        else:
-            ok = _create_contact(name, email)
-        # If a phone was provided, do an immediate update to thread it
-        # through (the simple _create_contact signature only takes name +
-        # email + address; phones happen via update).
-        if ok and phone:
-            try:
-                fresh = _fetch_contacts(force=True)
-                created = next((c for c in fresh if name == c.get("name") and (not email or email in c.get("emails", []))), None)
-                if created:
-                    _update_contact(
-                        created["uid"], name,
-                        created.get("emails", []),
-                        [phone],
-                        address,
-                    )
-            except Exception:
-                pass
-        return {"success": ok}
-
-    @router.post("/import")
-    async def import_vcf(data: dict, _admin: str = Depends(require_admin)):
-        """Import contacts from .vcf or CSV. Body: {"vcf": "..."} or {"csv": "..."}."""
-        # Coerce defensively: a non-string vcf/text/csv (e.g. a number or list
-        # in the JSON body) would otherwise reach .strip() and 500 with an
-        # AttributeError instead of degrading to a clean "no data" response.
-        text = str(data.get("vcf") or data.get("text") or "")
-        csv_text = str(data.get("csv") or "")
-        if text.strip():
-            if "BEGIN:VCARD" not in text.upper():
-                return {"success": False, "error": "No vCard data found"}
-            result = _import_vcards(text)
-        elif csv_text.strip():
-            result = _import_csv_contacts(csv_text)
-        else:
-            return {"success": False, "error": "No contact data found"}
-        result["success"] = result.get("imported", 0) > 0
-        return result
-
-    @router.get("/export")
-    async def export_contacts(
-        format: str = Query("vcf", pattern="^(vcf|csv)$"),
-        _admin: str = Depends(require_admin),
-    ):
-        """Export all contacts as vCard or CSV."""
-        contacts = _fetch_contacts(force=True)
-        if format == "csv":
-            content = _contacts_to_csv(contacts)
-            media_type = "text/csv; charset=utf-8"
-            filename = "odysseus-contacts.csv"
-        else:
-            content = _contacts_to_vcf(contacts)
-            media_type = "text/vcard; charset=utf-8"
-            filename = "odysseus-contacts.vcf"
-        return Response(
-            content=content,
-            media_type=media_type,
-            headers={"Content-Disposition": f'attachment; filename="{filename}"'},
-        )
-
-    @router.get("/config")
-    async def get_config(_admin: str = Depends(require_admin)):
-        cfg = _get_carddav_config()
-        # Mask password
-        if cfg["password"]:
-            cfg["password"] = "***"
-        return cfg
-
-    @router.put("/config")
-    async def update_config(data: dict, _admin: str = Depends(require_admin)):
-        settings = _load_settings()
-        for key in ("carddav_url", "carddav_username", "carddav_password"):
-            if key in data:
-                if key == "carddav_url" and str(data[key] or "").strip():
-                    try:
-                        settings[key] = _validate_carddav_url(data[key])
-                    except ValueError as e:
-                        raise HTTPException(400, str(e))
-                else:
-                    value = data[key]
-                    if key == "carddav_password" and value:
-                        from src.secret_storage import encrypt
-                        value = encrypt(value)
-                    settings[key] = value
-        _save_settings(settings)
-        # Force re-fetch
-        _contact_cache["fetched_at"] = None
-        return {"success": True}
-
-    @router.delete("/clear")
-    async def clear_contacts(_admin: str = Depends(require_admin)):
-        """Clear all local contacts. If CardDAV is configured, only clears the local fallback cache."""
-        _save_local_contacts([])
-        return {"success": True}
-
-    # NOTE: the /{uid} routes are declared LAST so the literal paths above
-    # (/list, /search, /add, /config) win — otherwise PUT /config would
-    # match PUT /{uid} with uid="config".
-    @router.put("/{uid}")
-    async def edit_contact(uid: str, data: dict, _admin: str = Depends(require_admin)):
-        """Edit an existing contact — name / emails / phones / address."""
-        name = (data.get("name") or "").strip()
-        emails = data.get("emails")
-        phones = data.get("phones")
-        if emails is None and data.get("email"):
-            emails = [data["email"]]
-        emails = [e.strip() for e in (emails or []) if e and e.strip()]
-        phones = [p.strip() for p in (phones or []) if p and p.strip()]
-        address = (data.get("address") or "").strip()
-        if not name and not emails and not address:
-            return {"success": False, "error": "Name, email, or address required"}
-        if not name and emails:
-            name = emails[0].split("@")[0]
-        ok = _update_contact(uid, name, emails, phones, address)
-        return {"success": ok}
-
-    @router.delete("/{uid}")
-    async def delete_contact(uid: str, _admin: str = Depends(require_admin)):
-        """Delete a contact by UID."""
-        if not uid:
-            return {"success": False, "error": "UID required"}
-        ok = _delete_contact(uid)
-        return {"success": ok}
-
-    return router
+_sys.modules[__name__] = _canonical

routes/history/__init__.py

@@ -0,0 +1,5 @@
+"""History route domain package (slice 2d, #4082/#4071).
+
+Contains history_routes.py, migrated from the flat routes/ directory.
+Backward-compat shim at routes/history_routes.py re-exports from here.
+"""

routes/history/history_routes.py

@@ -0,0 +1,768 @@
+"""History routes — session history, truncation, fork, conversation topics."""
+
+import json
+import uuid
+import logging
+import re
+from typing import Dict, Any, Optional
+
+from fastapi import APIRouter, Request, HTTPException
+
+from core.models import ChatMessage
+from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession
+from src.topic_analyzer import analyze_topics
+from routes.session_routes import (
+    _message_role,
+    _message_text,
+    _reject_compact_during_active_run,
+    _verify_session_owner,
+)
+
+logger = logging.getLogger(__name__)
+
+_HISTORY_INLINE_MEDIA_THRESHOLD = 200_000
+_DATA_IMAGE_RE = re.compile(r"data:image/[^;,\"]+;base64,[A-Za-z0-9+/=\s]+")
+
+
+def _history_display_content(content: Any) -> Any:
+    """Return a lightweight browser-display copy of stored message content.
+
+    Older multimodal user messages may be persisted as a JSON *string*
+    containing image_url blocks with inline base64 image bytes. Those bytes are
+    needed for model calls when the turn is first sent, but they should not be
+    sent back through /api/history every time the user opens the chat. The
+    attachment metadata already carries file ids/names for the UI cards.
+    """
+    if isinstance(content, list):
+        text_parts = []
+        omitted_media = 0
+        for block in content:
+            if not isinstance(block, dict):
+                continue
+            if block.get("type") == "text":
+                text = block.get("text")
+                if isinstance(text, str) and text:
+                    text_parts.append(text)
+            elif block.get("type") in {"image_url", "input_image", "audio", "input_audio"}:
+                omitted_media += 1
+        text = "\n".join(text_parts).strip()
+        if omitted_media and not text:
+            return f"[{omitted_media} media attachment{'s' if omitted_media != 1 else ''} omitted from history view]"
+        return text
+
+    if not isinstance(content, str):
+        return content
+    if len(content) < _HISTORY_INLINE_MEDIA_THRESHOLD and "data:image/" not in content:
+        return content
+
+    stripped = content.lstrip()
+    if stripped.startswith("["):
+        try:
+            blocks = json.loads(content)
+        except (json.JSONDecodeError, TypeError, ValueError):
+            blocks = None
+        if isinstance(blocks, list):
+            text_parts = []
+            for block in blocks:
+                if not isinstance(block, dict):
+                    continue
+                if block.get("type") == "text":
+                    text = block.get("text")
+                    if isinstance(text, str) and text:
+                        text_parts.append(text)
+            if text_parts:
+                return "\n".join(text_parts).strip()
+
+    if "data:image/" in content:
+        return _DATA_IMAGE_RE.sub("[inline image omitted from history view]", content)
+    return content
+
+
+def _merge_continue_rows_to_delete(db_messages, db1, db2):
+    """DB rows to delete when merging the last two assistant messages.
+
+    Always the second assistant message (db2), plus ONLY the single
+    intervening "continue" user message (the one carrying "previous response
+    was interrupted") — matching the in-memory merge. The previous code
+    deleted the whole index range between the two assistant rows, destroying
+    any tool/system/user messages in between and desyncing the DB from the
+    in-memory history.
+    """
+    to_delete = [db2]
+    i1 = next((i for i, m in enumerate(db_messages) if m is db1), None)
+    i2 = next((i for i, m in enumerate(db_messages) if m is db2), None)
+    if i1 is not None and i2 is not None and i2 - 1 > i1:
+        between = db_messages[i2 - 1]
+        if getattr(between, "role", "") == "user" and            "previous response was interrupted" in (getattr(between, "content", "") or ""):
+            to_delete.append(between)
+    return to_delete
+
+
+def setup_history_routes(session_manager) -> APIRouter:
+    router = APIRouter(tags=["history"])
+
+    def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]:
+        entry = {"role": m.role, "content": _history_display_content(m.content)}
+        meta = {}
+        if m.meta_data:
+            try:
+                meta = json.loads(m.meta_data) or {}
+            except (json.JSONDecodeError, ValueError):
+                meta = {}
+        if m.timestamp and "timestamp" not in meta:
+            meta["timestamp"] = m.timestamp.isoformat() + "Z"
+        if meta:
+            entry["metadata"] = meta
+        return entry
+
+    @router.get("/api/history/{session_id}")
+    async def get_session_history(
+        request: Request,
+        session_id: str,
+        limit: Optional[int] = None,
+        offset: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        _verify_session_owner(request, session_id)
+        if limit is not None:
+            page_limit = max(1, min(int(limit), 100))
+            db = SessionLocal()
+            try:
+                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
+                if db_session is None:
+                    raise HTTPException(404, f"Session '{session_id}' not found")
+
+                total = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id)
+                    .count()
+                )
+                page_offset = int(offset) if offset is not None else max(total - page_limit, 0)
+                page_offset = max(0, min(page_offset, total))
+                rows = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id)
+                    .order_by(DbChatMessage.timestamp)
+                    .offset(page_offset)
+                    .limit(page_limit)
+                    .all()
+                )
+                history_dict = [
+                    entry for entry in (_db_history_entry(m) for m in rows)
+                    if not (entry.get("metadata") or {}).get("hidden")
+                ]
+                return {
+                    "history": history_dict,
+                    "model": db_session.model,
+                    "endpoint_url": db_session.endpoint_url,
+                    "name": db_session.name,
+                    "offset": page_offset,
+                    "limit": page_limit,
+                    "total": total,
+                    "has_more_before": page_offset > 0,
+                    "has_more_after": page_offset + len(rows) < total,
+                }
+            finally:
+                db.close()
+
+        try:
+            session = session_manager.get_session(session_id)
+        except KeyError:
+            raise HTTPException(404, f"Session '{session_id}' not found")
+
+        history_dict = []
+        for msg in session.history:
+            if isinstance(msg, ChatMessage):
+                # Skip hidden messages (e.g. compaction summaries for AI context)
+                if msg.metadata and msg.metadata.get("hidden"):
+                    continue
+                entry = {"role": msg.role, "content": _history_display_content(msg.content)}
+                if msg.metadata:
+                    entry["metadata"] = msg.metadata
+                history_dict.append(entry)
+            elif isinstance(msg, dict):
+                if msg.get("metadata", {}).get("hidden"):
+                    continue
+                entry = {
+                    "role": msg.get("role", ""),
+                    "content": _history_display_content(msg.get("content", "")),
+                }
+                if msg.get("metadata"):
+                    entry["metadata"] = msg["metadata"]
+                history_dict.append(entry)
+
+        # Fallback: load from DB if in-memory is empty
+        if not history_dict:
+            db = SessionLocal()
+            try:
+                db_messages = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id)
+                    .order_by(DbChatMessage.timestamp)
+                    .all()
+                )
+                db_history = []
+                for m in db_messages:
+                    db_history.append(_db_history_entry(m))
+                if db_history:
+                    # Rebuild in-memory history from the full set so hidden
+                    # messages (e.g. compaction summaries) are kept for AI context.
+                    session.history = [
+                        ChatMessage(role=m["role"], content=m["content"], metadata=m.get("metadata"))
+                        for m in db_history
+                    ]
+                # Response excludes hidden messages, matching the in-memory path.
+                history_dict = [
+                    m for m in db_history
+                    if not (m.get("metadata") or {}).get("hidden")
+                ]
+            except Exception as e:
+                logger.error(f"DB fallback failed for {session_id}: {e}")
+            finally:
+                db.close()
+
+        return {
+            "history": history_dict,
+            "model": session.model,
+            "endpoint_url": session.endpoint_url,
+            "name": session.name,
+        }
+
+    @router.post("/api/session/{session_id}/truncate")
+    async def truncate_session(request: Request, session_id: str):
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            keep_count = body.get("keep_count", 0)
+            result = session_manager.truncate_messages(session_id, keep_count)
+            return {"status": "ok", "kept": keep_count, "truncated": result}
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except Exception as e:
+            logger.error(f"Truncate error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/message")
+    async def add_message(request: Request, session_id: str):
+        """Add a message to a session (for slash command persistence)."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            role = body.get("role", "assistant")
+            content = body.get("content", "")
+            if not content:
+                raise HTTPException(400, "content is required")
+            msg = ChatMessage(role=role, content=content, metadata=body.get("metadata"))
+            session_manager.add_message(session_id, msg)
+            return {"status": "ok"}
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+
+    @router.post("/api/session/{session_id}/delete-messages")
+    async def delete_messages(request: Request, session_id: str):
+        """Delete specific messages by DB ID (or legacy index)."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            msg_ids = body.get("msg_ids", [])
+            indices = body.get("indices")  # legacy fallback
+
+            session = session_manager.get_session(session_id)
+            db = SessionLocal()
+            try:
+                if msg_ids:
+                    # New ID-based delete
+                    deleted = 0
+                    for mid in msg_ids:
+                        db_msg = db.query(DbChatMessage).filter(
+                            DbChatMessage.id == mid,
+                            DbChatMessage.session_id == session_id,
+                        ).first()
+                        if db_msg:
+                            db.delete(db_msg)
+                            deleted += 1
+
+                    # Remove from in-memory history by matching _db_id
+                    def _get_db_id(m):
+                        meta = m.metadata if isinstance(m, ChatMessage) else (m.get('metadata') if isinstance(m, dict) else None)
+                        return meta.get('_db_id') if isinstance(meta, dict) else None
+                    session.history = [m for m in session.history if _get_db_id(m) not in msg_ids]
+                elif indices:
+                    # Legacy index-based delete
+                    indices = sorted(indices, reverse=True)
+                    db_messages = db.query(DbChatMessage).filter(
+                        DbChatMessage.session_id == session_id
+                    ).order_by(DbChatMessage.timestamp).all()
+
+                    deleted = 0
+                    for idx in indices:
+                        if 0 <= idx < len(db_messages):
+                            db.delete(db_messages[idx])
+                            deleted += 1
+                        if 0 <= idx < len(session.history):
+                            session.history.pop(idx)
+                else:
+                    return {"status": "ok", "deleted": 0}
+
+                session.message_count = len(session.history)
+                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
+                if db_session:
+                    db_session.message_count = len(session.history)
+                    from datetime import datetime, timezone
+                    db_session.updated_at = datetime.now(timezone.utc)
+
+                db.commit()
+                return {"status": "ok", "deleted": deleted}
+            finally:
+                db.close()
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except Exception as e:
+            logger.error(f"Delete messages error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/edit-message")
+    async def edit_message(request: Request, session_id: str):
+        """Edit the content of a message by its database ID."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            msg_id = body.get("msg_id")
+            content = body.get("content")
+            if not msg_id or content is None:
+                raise HTTPException(400, "msg_id and content are required")
+
+            session = session_manager.get_session(session_id)
+            db = SessionLocal()
+            try:
+                db_msg = db.query(DbChatMessage).filter(
+                    DbChatMessage.id == msg_id,
+                    DbChatMessage.session_id == session_id,
+                ).first()
+                if not db_msg:
+                    raise HTTPException(404, "Message not found")
+
+                db_msg.content = content
+                meta = {}
+                if db_msg.meta_data:
+                    try: meta = json.loads(db_msg.meta_data)
+                    except (json.JSONDecodeError, ValueError): pass
+                meta['edited'] = True
+                db_msg.meta_data = json.dumps(meta)
+
+                # Update in-memory history by matching _db_id
+                for hmsg in session.history:
+                    hmeta = hmsg.metadata if isinstance(hmsg, ChatMessage) else hmsg.get('metadata')
+                    if isinstance(hmeta, dict) and hmeta.get('_db_id') == msg_id:
+                        if isinstance(hmsg, ChatMessage):
+                            hmsg.content = content
+                            hmsg.metadata['edited'] = True
+                        elif isinstance(hmsg, dict):
+                            hmsg['content'] = content
+                            hmsg['metadata']['edited'] = True
+                        break
+
+                db.commit()
+                return {"status": "ok"}
+            finally:
+                db.close()
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error(f"Edit message error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/mark-stopped")
+    async def mark_stopped(request: Request, session_id: str):
+        """Mark the last assistant message as stopped by user."""
+        _verify_session_owner(request, session_id)
+        try:
+            session = session_manager.get_session(session_id)
+            # Find last assistant message and add stopped metadata
+            for msg in reversed(session.history):
+                if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
+                   (isinstance(msg, dict) and msg.get('role') == 'assistant'):
+                    if isinstance(msg, ChatMessage):
+                        if not msg.metadata:
+                            msg.metadata = {}
+                        msg.metadata['stopped'] = True
+                        if not msg.metadata.get('model'):
+                            msg.metadata['model'] = session.model
+                    else:
+                        if 'metadata' not in msg:
+                            msg['metadata'] = {}
+                        msg['metadata']['stopped'] = True
+                        if not msg['metadata'].get('model'):
+                            msg['metadata']['model'] = session.model
+                    break
+            # Also update in DB
+            db = SessionLocal()
+            try:
+                import json as _json
+                db_messages = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
+                    .order_by(DbChatMessage.timestamp.desc())
+                    .first()
+                )
+                if db_messages:
+                    meta = {}
+                    if db_messages.meta_data:
+                        try:
+                            meta = _json.loads(db_messages.meta_data)
+                        except (json.JSONDecodeError, ValueError):
+                            pass
+                    meta['stopped'] = True
+                    if not meta.get('model'):
+                        meta['model'] = session.model
+                    db_messages.meta_data = _json.dumps(meta)
+                    db.commit()
+            finally:
+                db.close()
+            session_manager.save_sessions()
+            return {"status": "ok"}
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except Exception as e:
+            logger.error(f"Mark stopped error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/update-last-meta")
+    async def update_last_meta(request: Request, session_id: str):
+        """Merge metadata into the last assistant message (e.g. save variants)."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            meta_update = body.get("metadata", {})
+            session = session_manager.get_session(session_id)
+
+            # Update in-memory
+            for msg in reversed(session.history):
+                if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
+                   (isinstance(msg, dict) and msg.get('role') == 'assistant'):
+                    if isinstance(msg, ChatMessage):
+                        if not msg.metadata:
+                            msg.metadata = {}
+                        msg.metadata.update(meta_update)
+                    else:
+                        if 'metadata' not in msg:
+                            msg['metadata'] = {}
+                        msg['metadata'].update(meta_update)
+                    break
+
+            # Update in DB
+            db = SessionLocal()
+            try:
+                import json as _json
+                db_msg = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
+                    .order_by(DbChatMessage.timestamp.desc())
+                    .first()
+                )
+                if db_msg:
+                    meta = {}
+                    if db_msg.meta_data:
+                        try: meta = _json.loads(db_msg.meta_data)
+                        except (json.JSONDecodeError, ValueError): pass
+                    meta.update(meta_update)
+                    db_msg.meta_data = _json.dumps(meta)
+                    db.commit()
+            finally:
+                db.close()
+            session_manager.save_sessions()
+            return {"status": "ok"}
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except Exception as e:
+            logger.error(f"Update last meta error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/merge-last-assistant")
+    async def merge_last_assistant(request: Request, session_id: str):
+        """Merge the last two assistant messages into one (for continue)."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            separator = body.get("separator", "\n\n")
+            session = session_manager.get_session(session_id)
+
+            # Find last two assistant messages in-memory
+            ai_indices = []
+            for i, msg in enumerate(session.history):
+                role = msg.role if isinstance(msg, ChatMessage) else msg.get('role', '')
+                if role == 'assistant':
+                    ai_indices.append(i)
+
+            if len(ai_indices) < 2:
+                return {"status": "ok", "merged": False}
+
+            idx1, idx2 = ai_indices[-2], ai_indices[-1]
+            msg1, msg2 = session.history[idx1], session.history[idx2]
+
+            content1 = msg1.content if isinstance(msg1, ChatMessage) else msg1.get('content', '')
+            content2 = msg2.content if isinstance(msg2, ChatMessage) else msg2.get('content', '')
+            merged_content = content1 + separator + content2
+
+            # Merge metadata
+            meta1 = (msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')) or {}
+            meta2 = (msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')) or {}
+            merged_meta = {**meta1, **meta2}
+            merged_meta.pop('stopped', None)  # no longer stopped after continue
+
+            # Update first message, remove second
+            if isinstance(msg1, ChatMessage):
+                msg1.content = merged_content
+                msg1.metadata = merged_meta
+            else:
+                msg1['content'] = merged_content
+                msg1['metadata'] = merged_meta
+
+            # Also remove the hidden "continue" user message between them if present
+            # It's the message at idx2-1 if it's a user message with continue text
+            remove_indices = [idx2]
+            if idx2 - 1 > idx1:
+                between = session.history[idx2 - 1]
+                between_role = between.role if isinstance(between, ChatMessage) else between.get('role', '')
+                between_content = between.content if isinstance(between, ChatMessage) else between.get('content', '')
+                if between_role == 'user' and 'previous response was interrupted' in between_content:
+                    remove_indices.insert(0, idx2 - 1)
+
+            for ri in sorted(remove_indices, reverse=True):
+                session.history.pop(ri)
+
+            # Update DB
+            db = SessionLocal()
+            try:
+                import json as _json
+                db_messages = (
+                    db.query(DbChatMessage)
+                    .filter(DbChatMessage.session_id == session_id)
+                    .order_by(DbChatMessage.timestamp)
+                    .all()
+                )
+                # Find last two assistant messages in DB
+                ai_db = [(i, m) for i, m in enumerate(db_messages) if m.role == 'assistant']
+                if len(ai_db) >= 2:
+                    (_, db1), (_, db2) = ai_db[-2], ai_db[-1]
+                    db1.content = merged_content
+                    db1.meta_data = _json.dumps(merged_meta)
+
+                    # Mirror the in-memory deletion: remove the second assistant
+                    # message and ONLY the "continue" user message between them
+                    # (not arbitrary tool/system/user rows). The old
+                    # range-delete destroyed every row between the two assistant
+                    # messages, desyncing the DB from the in-memory history.
+                    for _row in _merge_continue_rows_to_delete(db_messages, db1, db2):
+                        db.delete(_row)
+
+                    db.commit()
+            finally:
+                db.close()
+            session_manager.save_sessions()
+            return {"status": "ok", "merged": True}
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        except Exception as e:
+            logger.error(f"Merge assistant error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.post("/api/session/{session_id}/fork")
+    async def fork_session(request: Request, session_id: str):
+        """Create a new session with messages copied up to keep_count."""
+        _verify_session_owner(request, session_id)
+        try:
+            body = await request.json()
+            keep_count = body.get("keep_count", 0)
+
+            # Get the source session
+            source = session_manager.sessions.get(session_id)
+            if not source:
+                raise HTTPException(404, "Session not found")
+
+            # Create new session
+            new_id = str(uuid.uuid4())
+            fork_name = f"\u2ADD {source.name}"
+            new_session = session_manager.create_session(
+                session_id=new_id,
+                name=fork_name,
+                endpoint_url=source.endpoint_url,
+                model=source.model,
+                rag=False,
+                owner=getattr(source, 'owner', None),
+            )
+
+            # Copy messages up to keep_count
+            msgs_to_copy = source.history[:keep_count]
+            for msg in msgs_to_copy:
+                # Copy the metadata dict. Sharing it would let the fork's
+                # persistence (add_message -> _persist_message stamps
+                # _db_id/timestamp onto the dict) mutate the SOURCE session's
+                # in-memory messages, corrupting their _db_id and breaking
+                # edit/delete-by-id on the original conversation.
+                meta = dict(msg.metadata) if isinstance(msg.metadata, dict) else None
+                new_session.add_message(ChatMessage(msg.role, msg.content, meta))
+            try:
+                from src.event_bus import fire_event
+                fire_event("session_created", getattr(source, 'owner', None))
+            except Exception:
+                logger.debug("session_created event dispatch failed", exc_info=True)
+
+            return {
+                "status": "ok",
+                "id": new_id,
+                "name": fork_name,
+                "kept": len(msgs_to_copy),
+            }
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error(f"Fork error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    @router.get("/api/conversations/topics")
+    async def get_conversation_topics(request: Request) -> Dict[str, Any]:
+        from src.auth_helpers import require_user
+        user = require_user(request)
+        try:
+            return analyze_topics(session_manager, owner=user or None)
+        except Exception as e:
+            raise HTTPException(500, f"Topic analysis failed: {e}")
+
+    @router.post("/api/session/{session_id}/compact")
+    async def compact_session(request: Request, session_id: str):
+        """Manually trigger context compaction for a session."""
+        _verify_session_owner(request, session_id)
+        from src.auth_helpers import effective_user
+        owner = effective_user(request)
+        try:
+            session = session_manager.get_session(session_id)
+        except KeyError:
+            raise HTTPException(404, "Session not found")
+        _reject_compact_during_active_run(session_id)
+
+        try:
+            from src.model_context import estimate_tokens, get_context_length
+            from src.llm_core import llm_call_async
+            from src.endpoint_resolver import resolve_endpoint
+
+            if len(session.history) < 6:
+                return {"status": "ok", "message": "Not enough messages to compact"}
+
+            ctx_len = get_context_length(session.endpoint_url, session.model)
+            messages_before = session.get_context_messages()
+            used_before = estimate_tokens(messages_before)
+            pct_before = round((used_before / ctx_len) * 100, 1) if ctx_len else 0
+            msg_count_before = len(session.history)
+
+            # Keep only last 4 messages, summarize the rest
+            keep_count = 4
+            older = session.history[:-keep_count]
+            recent = session.history[-keep_count:]
+
+            # Build text to summarize
+            convo_text = "\n".join(
+                f"{_message_role(m).upper()}: "
+                f"{_message_text(m)[:2000]}"
+                for m in older
+            )
+
+            # Use utility model if available
+            util_url, util_model, util_headers = resolve_endpoint("utility", owner=owner or None)
+            compact_url = util_url or session.endpoint_url
+            compact_model = util_model or session.model
+            compact_headers = util_headers if util_url else session.headers
+
+            from src.context_compactor import SELF_SUMMARY_SYSTEM_PROMPT
+            compaction_count = sum(1 for m in session.history if isinstance(m, ChatMessage) and "[Conversation summary" in (m.content or ""))
+            sys_prompt = SELF_SUMMARY_SYSTEM_PROMPT.replace("{count}", str(len(older))).replace("{n}", str(compaction_count + 1))
+            summary = await llm_call_async(
+                compact_url, compact_model,
+                [
+                    {"role": "system", "content": sys_prompt},
+                    {"role": "user", "content": convo_text},
+                ],
+                temperature=0.2, max_tokens=1024,
+                headers=compact_headers, timeout=30,
+            )
+
+            # Replace session history: summary as system message + recent messages
+            # System message holds the full summary for AI context
+            system_summary = ChatMessage(
+                role="system",
+                content=f"[Conversation summary — {len(older)} earlier messages were compacted]\n\n{summary}",
+                metadata={"compacted": True, "hidden": True},
+            )
+            # Visible assistant message just shows stats
+            summary_msg = ChatMessage(
+                role="assistant",
+                content=f"**Conversation compacted** — {len(older)} messages summarized, {len(recent)} kept.",
+                metadata={"compacted": True, "messages_removed": len(older)},
+            )
+            new_history = [system_summary, summary_msg] + list(recent)
+            session.history = new_history
+            session.message_count = len(session.history)
+            logger.info(f"Compact: session {session_id} history now has {len(session.history)} messages (was {msg_count_before})")
+
+            # Update DB: delete old messages, insert summary
+            db = SessionLocal()
+            try:
+                db_msgs = db.query(DbChatMessage).filter(
+                    DbChatMessage.session_id == session_id
+                ).order_by(DbChatMessage.timestamp).all()
+
+                # Delete all but the last keep_count
+                for m in db_msgs[:-keep_count]:
+                    db.delete(m)
+
+                # Insert system summary (hidden, for AI context) and visible summary
+                import json as _json
+                import uuid
+                from datetime import datetime, timezone
+                now = datetime.now(timezone.utc)
+                db_sys_summary = DbChatMessage(
+                    id=str(uuid.uuid4()),
+                    session_id=session_id,
+                    role="system",
+                    content=system_summary.content,
+                    meta_data=_json.dumps(system_summary.metadata),
+                    timestamp=now,
+                )
+                db.add(db_sys_summary)
+                db_summary = DbChatMessage(
+                    id=str(uuid.uuid4()),
+                    session_id=session_id,
+                    role="assistant",
+                    content=summary_msg.content,
+                    meta_data=_json.dumps(summary_msg.metadata),
+                    timestamp=now,
+                )
+                db.add(db_summary)
+
+                # Update session record
+                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
+                if db_session:
+                    db_session.message_count = len(session.history)
+                    db_session.updated_at = datetime.now(timezone.utc)
+                db.commit()
+            finally:
+                db.close()
+
+            session_manager.save_sessions()
+
+            used_after = estimate_tokens(session.get_context_messages())
+            pct_after = round((used_after / ctx_len) * 100, 1) if ctx_len else 0
+
+            return {
+                "status": "ok",
+                "message": f"Compacted: {msg_count_before} msgs → {len(session.history)} msgs ({pct_before}% → {pct_after}%)",
+                "before": pct_before,
+                "after": pct_after,
+            }
+
+        except Exception as e:
+            logger.error(f"Manual compact error {session_id}: {e}")
+            raise HTTPException(500, str(e))
+
+    return router

routes/history_routes.py

@@ -1,768 +1,17 @@
-"""History routes — session history, truncation, fork, conversation topics."""
+"""Backward-compat shim — canonical location is routes/history/history_routes.py.
 
-import json
-import uuid
-import logging
-import re
-from typing import Dict, Any, Optional
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.history_routes``, ``from routes.history_routes import X``,
+``importlib.import_module("routes.history_routes")``, and the
+``import ... as history_routes`` + ``monkeypatch.setattr(history_routes, ...)``
+pattern used by test_history_compact_tool_calls.py / test_fork_session_metadata.py
+all operate on the *same* object the application actually uses. Keeps existing
+import paths working after slice 2d (#4082/#4071). Source-introspection tests
+read the canonical file by path.
+"""
 
-from fastapi import APIRouter, Request, HTTPException
+import sys as _sys
 
-from core.models import ChatMessage
-from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession
-from src.topic_analyzer import analyze_topics
-from routes.session_routes import (
-    _message_role,
-    _message_text,
-    _reject_compact_during_active_run,
-    _verify_session_owner,
-)
+from routes.history import history_routes as _canonical  # noqa: F401
 
-logger = logging.getLogger(__name__)
-
-_HISTORY_INLINE_MEDIA_THRESHOLD = 200_000
-_DATA_IMAGE_RE = re.compile(r"data:image/[^;,\"]+;base64,[A-Za-z0-9+/=\s]+")
-
-
-def _history_display_content(content: Any) -> Any:
-    """Return a lightweight browser-display copy of stored message content.
-
-    Older multimodal user messages may be persisted as a JSON *string*
-    containing image_url blocks with inline base64 image bytes. Those bytes are
-    needed for model calls when the turn is first sent, but they should not be
-    sent back through /api/history every time the user opens the chat. The
-    attachment metadata already carries file ids/names for the UI cards.
-    """
-    if isinstance(content, list):
-        text_parts = []
-        omitted_media = 0
-        for block in content:
-            if not isinstance(block, dict):
-                continue
-            if block.get("type") == "text":
-                text = block.get("text")
-                if isinstance(text, str) and text:
-                    text_parts.append(text)
-            elif block.get("type") in {"image_url", "input_image", "audio", "input_audio"}:
-                omitted_media += 1
-        text = "\n".join(text_parts).strip()
-        if omitted_media and not text:
-            return f"[{omitted_media} media attachment{'s' if omitted_media != 1 else ''} omitted from history view]"
-        return text
-
-    if not isinstance(content, str):
-        return content
-    if len(content) < _HISTORY_INLINE_MEDIA_THRESHOLD and "data:image/" not in content:
-        return content
-
-    stripped = content.lstrip()
-    if stripped.startswith("["):
-        try:
-            blocks = json.loads(content)
-        except (json.JSONDecodeError, TypeError, ValueError):
-            blocks = None
-        if isinstance(blocks, list):
-            text_parts = []
-            for block in blocks:
-                if not isinstance(block, dict):
-                    continue
-                if block.get("type") == "text":
-                    text = block.get("text")
-                    if isinstance(text, str) and text:
-                        text_parts.append(text)
-            if text_parts:
-                return "\n".join(text_parts).strip()
-
-    if "data:image/" in content:
-        return _DATA_IMAGE_RE.sub("[inline image omitted from history view]", content)
-    return content
-
-
-def _merge_continue_rows_to_delete(db_messages, db1, db2):
-    """DB rows to delete when merging the last two assistant messages.
-
-    Always the second assistant message (db2), plus ONLY the single
-    intervening "continue" user message (the one carrying "previous response
-    was interrupted") — matching the in-memory merge. The previous code
-    deleted the whole index range between the two assistant rows, destroying
-    any tool/system/user messages in between and desyncing the DB from the
-    in-memory history.
-    """
-    to_delete = [db2]
-    i1 = next((i for i, m in enumerate(db_messages) if m is db1), None)
-    i2 = next((i for i, m in enumerate(db_messages) if m is db2), None)
-    if i1 is not None and i2 is not None and i2 - 1 > i1:
-        between = db_messages[i2 - 1]
-        if getattr(between, "role", "") == "user" and            "previous response was interrupted" in (getattr(between, "content", "") or ""):
-            to_delete.append(between)
-    return to_delete
-
-
-def setup_history_routes(session_manager) -> APIRouter:
-    router = APIRouter(tags=["history"])
-
-    def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]:
-        entry = {"role": m.role, "content": _history_display_content(m.content)}
-        meta = {}
-        if m.meta_data:
-            try:
-                meta = json.loads(m.meta_data) or {}
-            except (json.JSONDecodeError, ValueError):
-                meta = {}
-        if m.timestamp and "timestamp" not in meta:
-            meta["timestamp"] = m.timestamp.isoformat() + "Z"
-        if meta:
-            entry["metadata"] = meta
-        return entry
-
-    @router.get("/api/history/{session_id}")
-    async def get_session_history(
-        request: Request,
-        session_id: str,
-        limit: Optional[int] = None,
-        offset: Optional[int] = None,
-    ) -> Dict[str, Any]:
-        _verify_session_owner(request, session_id)
-        if limit is not None:
-            page_limit = max(1, min(int(limit), 100))
-            db = SessionLocal()
-            try:
-                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
-                if db_session is None:
-                    raise HTTPException(404, f"Session '{session_id}' not found")
-
-                total = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id)
-                    .count()
-                )
-                page_offset = int(offset) if offset is not None else max(total - page_limit, 0)
-                page_offset = max(0, min(page_offset, total))
-                rows = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id)
-                    .order_by(DbChatMessage.timestamp)
-                    .offset(page_offset)
-                    .limit(page_limit)
-                    .all()
-                )
-                history_dict = [
-                    entry for entry in (_db_history_entry(m) for m in rows)
-                    if not (entry.get("metadata") or {}).get("hidden")
-                ]
-                return {
-                    "history": history_dict,
-                    "model": db_session.model,
-                    "endpoint_url": db_session.endpoint_url,
-                    "name": db_session.name,
-                    "offset": page_offset,
-                    "limit": page_limit,
-                    "total": total,
-                    "has_more_before": page_offset > 0,
-                    "has_more_after": page_offset + len(rows) < total,
-                }
-            finally:
-                db.close()
-
-        try:
-            session = session_manager.get_session(session_id)
-        except KeyError:
-            raise HTTPException(404, f"Session '{session_id}' not found")
-
-        history_dict = []
-        for msg in session.history:
-            if isinstance(msg, ChatMessage):
-                # Skip hidden messages (e.g. compaction summaries for AI context)
-                if msg.metadata and msg.metadata.get("hidden"):
-                    continue
-                entry = {"role": msg.role, "content": _history_display_content(msg.content)}
-                if msg.metadata:
-                    entry["metadata"] = msg.metadata
-                history_dict.append(entry)
-            elif isinstance(msg, dict):
-                if msg.get("metadata", {}).get("hidden"):
-                    continue
-                entry = {
-                    "role": msg.get("role", ""),
-                    "content": _history_display_content(msg.get("content", "")),
-                }
-                if msg.get("metadata"):
-                    entry["metadata"] = msg["metadata"]
-                history_dict.append(entry)
-
-        # Fallback: load from DB if in-memory is empty
-        if not history_dict:
-            db = SessionLocal()
-            try:
-                db_messages = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id)
-                    .order_by(DbChatMessage.timestamp)
-                    .all()
-                )
-                db_history = []
-                for m in db_messages:
-                    db_history.append(_db_history_entry(m))
-                if db_history:
-                    # Rebuild in-memory history from the full set so hidden
-                    # messages (e.g. compaction summaries) are kept for AI context.
-                    session.history = [
-                        ChatMessage(role=m["role"], content=m["content"], metadata=m.get("metadata"))
-                        for m in db_history
-                    ]
-                # Response excludes hidden messages, matching the in-memory path.
-                history_dict = [
-                    m for m in db_history
-                    if not (m.get("metadata") or {}).get("hidden")
-                ]
-            except Exception as e:
-                logger.error(f"DB fallback failed for {session_id}: {e}")
-            finally:
-                db.close()
-
-        return {
-            "history": history_dict,
-            "model": session.model,
-            "endpoint_url": session.endpoint_url,
-            "name": session.name,
-        }
-
-    @router.post("/api/session/{session_id}/truncate")
-    async def truncate_session(request: Request, session_id: str):
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            keep_count = body.get("keep_count", 0)
-            result = session_manager.truncate_messages(session_id, keep_count)
-            return {"status": "ok", "kept": keep_count, "truncated": result}
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except Exception as e:
-            logger.error(f"Truncate error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/message")
-    async def add_message(request: Request, session_id: str):
-        """Add a message to a session (for slash command persistence)."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            role = body.get("role", "assistant")
-            content = body.get("content", "")
-            if not content:
-                raise HTTPException(400, "content is required")
-            msg = ChatMessage(role=role, content=content, metadata=body.get("metadata"))
-            session_manager.add_message(session_id, msg)
-            return {"status": "ok"}
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-
-    @router.post("/api/session/{session_id}/delete-messages")
-    async def delete_messages(request: Request, session_id: str):
-        """Delete specific messages by DB ID (or legacy index)."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            msg_ids = body.get("msg_ids", [])
-            indices = body.get("indices")  # legacy fallback
-
-            session = session_manager.get_session(session_id)
-            db = SessionLocal()
-            try:
-                if msg_ids:
-                    # New ID-based delete
-                    deleted = 0
-                    for mid in msg_ids:
-                        db_msg = db.query(DbChatMessage).filter(
-                            DbChatMessage.id == mid,
-                            DbChatMessage.session_id == session_id,
-                        ).first()
-                        if db_msg:
-                            db.delete(db_msg)
-                            deleted += 1
-
-                    # Remove from in-memory history by matching _db_id
-                    def _get_db_id(m):
-                        meta = m.metadata if isinstance(m, ChatMessage) else (m.get('metadata') if isinstance(m, dict) else None)
-                        return meta.get('_db_id') if isinstance(meta, dict) else None
-                    session.history = [m for m in session.history if _get_db_id(m) not in msg_ids]
-                elif indices:
-                    # Legacy index-based delete
-                    indices = sorted(indices, reverse=True)
-                    db_messages = db.query(DbChatMessage).filter(
-                        DbChatMessage.session_id == session_id
-                    ).order_by(DbChatMessage.timestamp).all()
-
-                    deleted = 0
-                    for idx in indices:
-                        if 0 <= idx < len(db_messages):
-                            db.delete(db_messages[idx])
-                            deleted += 1
-                        if 0 <= idx < len(session.history):
-                            session.history.pop(idx)
-                else:
-                    return {"status": "ok", "deleted": 0}
-
-                session.message_count = len(session.history)
-                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
-                if db_session:
-                    db_session.message_count = len(session.history)
-                    from datetime import datetime, timezone
-                    db_session.updated_at = datetime.now(timezone.utc)
-
-                db.commit()
-                return {"status": "ok", "deleted": deleted}
-            finally:
-                db.close()
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except Exception as e:
-            logger.error(f"Delete messages error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/edit-message")
-    async def edit_message(request: Request, session_id: str):
-        """Edit the content of a message by its database ID."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            msg_id = body.get("msg_id")
-            content = body.get("content")
-            if not msg_id or content is None:
-                raise HTTPException(400, "msg_id and content are required")
-
-            session = session_manager.get_session(session_id)
-            db = SessionLocal()
-            try:
-                db_msg = db.query(DbChatMessage).filter(
-                    DbChatMessage.id == msg_id,
-                    DbChatMessage.session_id == session_id,
-                ).first()
-                if not db_msg:
-                    raise HTTPException(404, "Message not found")
-
-                db_msg.content = content
-                meta = {}
-                if db_msg.meta_data:
-                    try: meta = json.loads(db_msg.meta_data)
-                    except (json.JSONDecodeError, ValueError): pass
-                meta['edited'] = True
-                db_msg.meta_data = json.dumps(meta)
-
-                # Update in-memory history by matching _db_id
-                for hmsg in session.history:
-                    hmeta = hmsg.metadata if isinstance(hmsg, ChatMessage) else hmsg.get('metadata')
-                    if isinstance(hmeta, dict) and hmeta.get('_db_id') == msg_id:
-                        if isinstance(hmsg, ChatMessage):
-                            hmsg.content = content
-                            hmsg.metadata['edited'] = True
-                        elif isinstance(hmsg, dict):
-                            hmsg['content'] = content
-                            hmsg['metadata']['edited'] = True
-                        break
-
-                db.commit()
-                return {"status": "ok"}
-            finally:
-                db.close()
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except HTTPException:
-            raise
-        except Exception as e:
-            logger.error(f"Edit message error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/mark-stopped")
-    async def mark_stopped(request: Request, session_id: str):
-        """Mark the last assistant message as stopped by user."""
-        _verify_session_owner(request, session_id)
-        try:
-            session = session_manager.get_session(session_id)
-            # Find last assistant message and add stopped metadata
-            for msg in reversed(session.history):
-                if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
-                   (isinstance(msg, dict) and msg.get('role') == 'assistant'):
-                    if isinstance(msg, ChatMessage):
-                        if not msg.metadata:
-                            msg.metadata = {}
-                        msg.metadata['stopped'] = True
-                        if not msg.metadata.get('model'):
-                            msg.metadata['model'] = session.model
-                    else:
-                        if 'metadata' not in msg:
-                            msg['metadata'] = {}
-                        msg['metadata']['stopped'] = True
-                        if not msg['metadata'].get('model'):
-                            msg['metadata']['model'] = session.model
-                    break
-            # Also update in DB
-            db = SessionLocal()
-            try:
-                import json as _json
-                db_messages = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
-                    .order_by(DbChatMessage.timestamp.desc())
-                    .first()
-                )
-                if db_messages:
-                    meta = {}
-                    if db_messages.meta_data:
-                        try:
-                            meta = _json.loads(db_messages.meta_data)
-                        except (json.JSONDecodeError, ValueError):
-                            pass
-                    meta['stopped'] = True
-                    if not meta.get('model'):
-                        meta['model'] = session.model
-                    db_messages.meta_data = _json.dumps(meta)
-                    db.commit()
-            finally:
-                db.close()
-            session_manager.save_sessions()
-            return {"status": "ok"}
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except Exception as e:
-            logger.error(f"Mark stopped error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/update-last-meta")
-    async def update_last_meta(request: Request, session_id: str):
-        """Merge metadata into the last assistant message (e.g. save variants)."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            meta_update = body.get("metadata", {})
-            session = session_manager.get_session(session_id)
-
-            # Update in-memory
-            for msg in reversed(session.history):
-                if (isinstance(msg, ChatMessage) and msg.role == 'assistant') or \
-                   (isinstance(msg, dict) and msg.get('role') == 'assistant'):
-                    if isinstance(msg, ChatMessage):
-                        if not msg.metadata:
-                            msg.metadata = {}
-                        msg.metadata.update(meta_update)
-                    else:
-                        if 'metadata' not in msg:
-                            msg['metadata'] = {}
-                        msg['metadata'].update(meta_update)
-                    break
-
-            # Update in DB
-            db = SessionLocal()
-            try:
-                import json as _json
-                db_msg = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id, DbChatMessage.role == 'assistant')
-                    .order_by(DbChatMessage.timestamp.desc())
-                    .first()
-                )
-                if db_msg:
-                    meta = {}
-                    if db_msg.meta_data:
-                        try: meta = _json.loads(db_msg.meta_data)
-                        except (json.JSONDecodeError, ValueError): pass
-                    meta.update(meta_update)
-                    db_msg.meta_data = _json.dumps(meta)
-                    db.commit()
-            finally:
-                db.close()
-            session_manager.save_sessions()
-            return {"status": "ok"}
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except Exception as e:
-            logger.error(f"Update last meta error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/merge-last-assistant")
-    async def merge_last_assistant(request: Request, session_id: str):
-        """Merge the last two assistant messages into one (for continue)."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            separator = body.get("separator", "\n\n")
-            session = session_manager.get_session(session_id)
-
-            # Find last two assistant messages in-memory
-            ai_indices = []
-            for i, msg in enumerate(session.history):
-                role = msg.role if isinstance(msg, ChatMessage) else msg.get('role', '')
-                if role == 'assistant':
-                    ai_indices.append(i)
-
-            if len(ai_indices) < 2:
-                return {"status": "ok", "merged": False}
-
-            idx1, idx2 = ai_indices[-2], ai_indices[-1]
-            msg1, msg2 = session.history[idx1], session.history[idx2]
-
-            content1 = msg1.content if isinstance(msg1, ChatMessage) else msg1.get('content', '')
-            content2 = msg2.content if isinstance(msg2, ChatMessage) else msg2.get('content', '')
-            merged_content = content1 + separator + content2
-
-            # Merge metadata
-            meta1 = (msg1.metadata if isinstance(msg1, ChatMessage) else msg1.get('metadata')) or {}
-            meta2 = (msg2.metadata if isinstance(msg2, ChatMessage) else msg2.get('metadata')) or {}
-            merged_meta = {**meta1, **meta2}
-            merged_meta.pop('stopped', None)  # no longer stopped after continue
-
-            # Update first message, remove second
-            if isinstance(msg1, ChatMessage):
-                msg1.content = merged_content
-                msg1.metadata = merged_meta
-            else:
-                msg1['content'] = merged_content
-                msg1['metadata'] = merged_meta
-
-            # Also remove the hidden "continue" user message between them if present
-            # It's the message at idx2-1 if it's a user message with continue text
-            remove_indices = [idx2]
-            if idx2 - 1 > idx1:
-                between = session.history[idx2 - 1]
-                between_role = between.role if isinstance(between, ChatMessage) else between.get('role', '')
-                between_content = between.content if isinstance(between, ChatMessage) else between.get('content', '')
-                if between_role == 'user' and 'previous response was interrupted' in between_content:
-                    remove_indices.insert(0, idx2 - 1)
-
-            for ri in sorted(remove_indices, reverse=True):
-                session.history.pop(ri)
-
-            # Update DB
-            db = SessionLocal()
-            try:
-                import json as _json
-                db_messages = (
-                    db.query(DbChatMessage)
-                    .filter(DbChatMessage.session_id == session_id)
-                    .order_by(DbChatMessage.timestamp)
-                    .all()
-                )
-                # Find last two assistant messages in DB
-                ai_db = [(i, m) for i, m in enumerate(db_messages) if m.role == 'assistant']
-                if len(ai_db) >= 2:
-                    (_, db1), (_, db2) = ai_db[-2], ai_db[-1]
-                    db1.content = merged_content
-                    db1.meta_data = _json.dumps(merged_meta)
-
-                    # Mirror the in-memory deletion: remove the second assistant
-                    # message and ONLY the "continue" user message between them
-                    # (not arbitrary tool/system/user rows). The old
-                    # range-delete destroyed every row between the two assistant
-                    # messages, desyncing the DB from the in-memory history.
-                    for _row in _merge_continue_rows_to_delete(db_messages, db1, db2):
-                        db.delete(_row)
-
-                    db.commit()
-            finally:
-                db.close()
-            session_manager.save_sessions()
-            return {"status": "ok", "merged": True}
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        except Exception as e:
-            logger.error(f"Merge assistant error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.post("/api/session/{session_id}/fork")
-    async def fork_session(request: Request, session_id: str):
-        """Create a new session with messages copied up to keep_count."""
-        _verify_session_owner(request, session_id)
-        try:
-            body = await request.json()
-            keep_count = body.get("keep_count", 0)
-
-            # Get the source session
-            source = session_manager.sessions.get(session_id)
-            if not source:
-                raise HTTPException(404, "Session not found")
-
-            # Create new session
-            new_id = str(uuid.uuid4())
-            fork_name = f"\u2ADD {source.name}"
-            new_session = session_manager.create_session(
-                session_id=new_id,
-                name=fork_name,
-                endpoint_url=source.endpoint_url,
-                model=source.model,
-                rag=False,
-                owner=getattr(source, 'owner', None),
-            )
-
-            # Copy messages up to keep_count
-            msgs_to_copy = source.history[:keep_count]
-            for msg in msgs_to_copy:
-                # Copy the metadata dict. Sharing it would let the fork's
-                # persistence (add_message -> _persist_message stamps
-                # _db_id/timestamp onto the dict) mutate the SOURCE session's
-                # in-memory messages, corrupting their _db_id and breaking
-                # edit/delete-by-id on the original conversation.
-                meta = dict(msg.metadata) if isinstance(msg.metadata, dict) else None
-                new_session.add_message(ChatMessage(msg.role, msg.content, meta))
-            try:
-                from src.event_bus import fire_event
-                fire_event("session_created", getattr(source, 'owner', None))
-            except Exception:
-                logger.debug("session_created event dispatch failed", exc_info=True)
-
-            return {
-                "status": "ok",
-                "id": new_id,
-                "name": fork_name,
-                "kept": len(msgs_to_copy),
-            }
-        except HTTPException:
-            raise
-        except Exception as e:
-            logger.error(f"Fork error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    @router.get("/api/conversations/topics")
-    async def get_conversation_topics(request: Request) -> Dict[str, Any]:
-        from src.auth_helpers import require_user
-        user = require_user(request)
-        try:
-            return analyze_topics(session_manager, owner=user or None)
-        except Exception as e:
-            raise HTTPException(500, f"Topic analysis failed: {e}")
-
-    @router.post("/api/session/{session_id}/compact")
-    async def compact_session(request: Request, session_id: str):
-        """Manually trigger context compaction for a session."""
-        _verify_session_owner(request, session_id)
-        from src.auth_helpers import effective_user
-        owner = effective_user(request)
-        try:
-            session = session_manager.get_session(session_id)
-        except KeyError:
-            raise HTTPException(404, "Session not found")
-        _reject_compact_during_active_run(session_id)
-
-        try:
-            from src.model_context import estimate_tokens, get_context_length
-            from src.llm_core import llm_call_async
-            from src.endpoint_resolver import resolve_endpoint
-
-            if len(session.history) < 6:
-                return {"status": "ok", "message": "Not enough messages to compact"}
-
-            ctx_len = get_context_length(session.endpoint_url, session.model)
-            messages_before = session.get_context_messages()
-            used_before = estimate_tokens(messages_before)
-            pct_before = round((used_before / ctx_len) * 100, 1) if ctx_len else 0
-            msg_count_before = len(session.history)
-
-            # Keep only last 4 messages, summarize the rest
-            keep_count = 4
-            older = session.history[:-keep_count]
-            recent = session.history[-keep_count:]
-
-            # Build text to summarize
-            convo_text = "\n".join(
-                f"{_message_role(m).upper()}: "
-                f"{_message_text(m)[:2000]}"
-                for m in older
-            )
-
-            # Use utility model if available
-            util_url, util_model, util_headers = resolve_endpoint("utility", owner=owner or None)
-            compact_url = util_url or session.endpoint_url
-            compact_model = util_model or session.model
-            compact_headers = util_headers if util_url else session.headers
-
-            from src.context_compactor import SELF_SUMMARY_SYSTEM_PROMPT
-            compaction_count = sum(1 for m in session.history if isinstance(m, ChatMessage) and "[Conversation summary" in (m.content or ""))
-            sys_prompt = SELF_SUMMARY_SYSTEM_PROMPT.replace("{count}", str(len(older))).replace("{n}", str(compaction_count + 1))
-            summary = await llm_call_async(
-                compact_url, compact_model,
-                [
-                    {"role": "system", "content": sys_prompt},
-                    {"role": "user", "content": convo_text},
-                ],
-                temperature=0.2, max_tokens=1024,
-                headers=compact_headers, timeout=30,
-            )
-
-            # Replace session history: summary as system message + recent messages
-            # System message holds the full summary for AI context
-            system_summary = ChatMessage(
-                role="system",
-                content=f"[Conversation summary — {len(older)} earlier messages were compacted]\n\n{summary}",
-                metadata={"compacted": True, "hidden": True},
-            )
-            # Visible assistant message just shows stats
-            summary_msg = ChatMessage(
-                role="assistant",
-                content=f"**Conversation compacted** — {len(older)} messages summarized, {len(recent)} kept.",
-                metadata={"compacted": True, "messages_removed": len(older)},
-            )
-            new_history = [system_summary, summary_msg] + list(recent)
-            session.history = new_history
-            session.message_count = len(session.history)
-            logger.info(f"Compact: session {session_id} history now has {len(session.history)} messages (was {msg_count_before})")
-
-            # Update DB: delete old messages, insert summary
-            db = SessionLocal()
-            try:
-                db_msgs = db.query(DbChatMessage).filter(
-                    DbChatMessage.session_id == session_id
-                ).order_by(DbChatMessage.timestamp).all()
-
-                # Delete all but the last keep_count
-                for m in db_msgs[:-keep_count]:
-                    db.delete(m)
-
-                # Insert system summary (hidden, for AI context) and visible summary
-                import json as _json
-                import uuid
-                from datetime import datetime, timezone
-                now = datetime.now(timezone.utc)
-                db_sys_summary = DbChatMessage(
-                    id=str(uuid.uuid4()),
-                    session_id=session_id,
-                    role="system",
-                    content=system_summary.content,
-                    meta_data=_json.dumps(system_summary.metadata),
-                    timestamp=now,
-                )
-                db.add(db_sys_summary)
-                db_summary = DbChatMessage(
-                    id=str(uuid.uuid4()),
-                    session_id=session_id,
-                    role="assistant",
-                    content=summary_msg.content,
-                    meta_data=_json.dumps(summary_msg.metadata),
-                    timestamp=now,
-                )
-                db.add(db_summary)
-
-                # Update session record
-                db_session = db.query(DbSession).filter(DbSession.id == session_id).first()
-                if db_session:
-                    db_session.message_count = len(session.history)
-                    db_session.updated_at = datetime.now(timezone.utc)
-                db.commit()
-            finally:
-                db.close()
-
-            session_manager.save_sessions()
-
-            used_after = estimate_tokens(session.get_context_messages())
-            pct_after = round((used_after / ctx_len) * 100, 1) if ctx_len else 0
-
-            return {
-                "status": "ok",
-                "message": f"Compacted: {msg_count_before} msgs → {len(session.history)} msgs ({pct_before}% → {pct_after}%)",
-                "before": pct_before,
-                "after": pct_after,
-            }
-
-        except Exception as e:
-            logger.error(f"Manual compact error {session_id}: {e}")
-            raise HTTPException(500, str(e))
-
-    return router
+_sys.modules[__name__] = _canonical

routes/note_routes.py

@@ -483,11 +483,22 @@ async def dispatch_reminder(
                 api_key = intg.get("api_key", "")
                 if api_key:
                     hdrs["Authorization"] = f"Bearer {api_key}"
-                async with httpx.AsyncClient(timeout=10.0) as client:
-                    resp = await client.post(f"{base}/{topic}", content=ntfy_body, headers=hdrs)
-                    ntfy_sent = resp.is_success
-                    if not ntfy_sent:
-                        ntfy_error = f"ntfy returned HTTP {resp.status_code}"
+                # SSRF guard — same check (and env knob) as the webhook branch
+                # above: link-local / metadata addresses are always rejected;
+                # REMINDER_WEBHOOK_BLOCK_PRIVATE_IPS=true also blocks RFC-1918
+                # so a ntfy base_url can't be pointed at internal services.
+                import os as _os
+                from src.url_safety import check_outbound_url as _chk
+                _block = _os.getenv("REMINDER_WEBHOOK_BLOCK_PRIVATE_IPS", "false").lower() == "true"
+                _ok, _reason = _chk(f"{base}/{topic}", block_private=_block)
+                if not _ok:
+                    ntfy_error = f"ntfy URL rejected: {_reason}"
+                else:
+                    async with httpx.AsyncClient(timeout=10.0) as client:
+                        resp = await client.post(f"{base}/{topic}", content=ntfy_body, headers=hdrs)
+                        ntfy_sent = resp.is_success
+                        if not ntfy_sent:
+                            ntfy_error = f"ntfy returned HTTP {resp.status_code}"
             else:
                 ntfy_error = "No enabled ntfy integration"
         except Exception as e:

routes/research/research_routes.py

@@ -21,23 +21,52 @@ 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.
-    """
+def _validate_session_id(session_id: str) -> str:
     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()
+        raise HTTPException(400, "Invalid session ID format")
+    return session_id
+
+
+def _research_storage_root() -> Path:
+    return Path(DEEP_RESEARCH_DIR).resolve()
+
+
+def _find_research_path(session_id: str) -> Path | None:
+    """Find a persisted research file without deriving its path from input."""
+    expected_name = f"{_validate_session_id(session_id)}.json"
+    root = _research_storage_root()
+    for stored_path in root.glob("*.json"):
+        if stored_path.name != expected_name:
+            continue
+        resolved = stored_path.resolve()
+        try:
+            resolved.relative_to(root)
+        except ValueError:
+            return None
+        if not resolved.is_file():
+            return None
+        return resolved
+    return None
+
+
+def _require_research_path(session_id: str) -> Path:
+    path = _find_research_path(session_id)
+    if path is None:
+        raise HTTPException(404, "Research not found")
+    return path
+
+
+def _find_owned_research_path(session_id: str, user: str) -> Path | None:
+    path = _find_research_path(session_id)
+    if path is None:
+        return None
     try:
-        candidate.relative_to(root)
-    except ValueError:
-        raise HTTPException(400, "Invalid session ID")
-    return candidate
+        owner = json.loads(path.read_text(encoding="utf-8")).get("owner")
+    except Exception:
+        return None
+    if owner != user:
+        return None
+    return path
 
 
 logger = logging.getLogger(__name__)
@@ -192,10 +221,6 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
             raise HTTPException(401, "Not authenticated")
         return user
 
-    def _validate_session_id(session_id: str) -> None:
-        if not _SESSION_ID_RE.fullmatch(session_id):
-            raise HTTPException(400, "Invalid session ID format")
-
     def _owns_in_memory(session_id: str, user: str) -> bool:
         """Ownership check for an in-flight (in-memory) research task.
         Falls back to the on-disk JSON if the task has already finished."""
@@ -204,15 +229,32 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
             return entry.get("owner", "") == user
         # Task no longer in memory — check the persisted JSON.
         try:
-            path = _confine_research_path(session_id)
+            return _find_owned_research_path(session_id, user) is not None
         except HTTPException:
             return False
-        if not path.exists():
-            return False
-        try:
-            return json.loads(path.read_text(encoding="utf-8")).get("owner") == user
-        except Exception:
-            return False
+
+    def _require_owned_or_active_research_path(session_id: str, user: str) -> Path | None:
+        """Validate ownership once and return the completed on-disk path.
+
+        Active running research has no completed disk path yet. Completed
+        tasks can remain in _active_tasks after persistence, so prefer their
+        owned disk path when available. Completed disk lookups still reuse the
+        path after the ownership gate.
+        """
+        entry = research_handler._active_tasks.get(session_id)
+        if entry is not None:
+            if entry.get("owner", "") != user:
+                raise HTTPException(404, "No research found for this session")
+            if entry.get("status") != "running":
+                path = _find_owned_research_path(session_id, user)
+                if path is not None:
+                    return path
+            return None
+
+        path = _find_owned_research_path(session_id, user)
+        if path is None:
+            raise HTTPException(404, "No research found for this session")
+        return path
 
     @router.get("/api/research/active")
     async def research_active(request: Request):
@@ -270,9 +312,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 = _confine_research_path(session_id)
-        if not path.exists():
-            raise HTTPException(404, "Research not found")
+        path = _require_research_path(session_id)
         try:
             owner = json.loads(path.read_text(encoding="utf-8")).get("owner")
         except Exception:
@@ -384,9 +424,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 = _confine_research_path(session_id)
-        if not path.exists():
-            raise HTTPException(404, "Research not found")
+        path = _require_research_path(session_id)
         try:
             data = json.loads(path.read_text(encoding="utf-8"))
         except Exception as e:
@@ -401,9 +439,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 = _confine_research_path(session_id)
-        if not path.exists():
-            raise HTTPException(404, "Research not found")
+        path = _require_research_path(session_id)
         try:
             data = json.loads(path.read_text(encoding="utf-8"))
             if data.get("owner") != user:
@@ -421,9 +457,9 @@ 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)
-        json_path = _confine_research_path(session_id)
+        json_path = _find_research_path(session_id)
         deleted = False
-        if json_path.exists():
+        if json_path is not None:
             # SECURITY: verify ownership before letting the caller delete it.
             try:
                 data = json.loads(json_path.read_text(encoding="utf-8"))
@@ -579,12 +615,11 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         """Get research result without clearing it (for panel use)."""
         user = _require_user(request)
         _validate_session_id(session_id)
-        if not _owns_in_memory(session_id, user):
-            raise HTTPException(404, "No research found for this session")
+        owned_disk_path = _require_owned_or_active_research_path(session_id, user)
         result = research_handler.get_result(session_id)
         if result is None:
-            p = _confine_research_path(session_id)
-            if p.exists():
+            p = owned_disk_path
+            if p is not None:
                 d = json.loads(p.read_text(encoding="utf-8"))
                 return {
                     "result": d.get("result", ""),
@@ -613,8 +648,7 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         # otherwise any authenticated user could spin off (and thereby read)
         # another user's report by guessing its session ID. Mirrors every other
         # endpoint in this file (see result_peek above).
-        if not _owns_in_memory(session_id, user):
-            raise HTTPException(404, "No research found for this session")
+        owned_disk_path = _require_owned_or_active_research_path(session_id, user)
         if session_manager is None:
             raise HTTPException(500, "session_manager not configured")
 
@@ -623,8 +657,8 @@ def setup_research_routes(research_handler, session_manager=None) -> APIRouter:
         sources = research_handler.get_sources(session_id) or []
         query = ""
 
-        path = _confine_research_path(session_id)
-        if path.exists():
+        path = owned_disk_path
+        if path is not None:
             try:
                 disk = json.loads(path.read_text(encoding="utf-8"))
                 if not result:

services/search/query.py

@@ -34,8 +34,14 @@ def _extract_entities(query: str) -> Dict[str, List[str]]:
     cleaned = query
     if qtype:
         cleaned = re.sub(rf"^{qtype}\b", "", cleaned, flags=re.I).strip()
-    for token in re.findall(r"\b[A-Z][a-zA-Z]+\b", cleaned):
-        entities["names"].append(token)
+    # Unicode-aware capitalized-word (name) detection. The old [A-Z][a-zA-Z]+
+    # class missed non-ASCII names like "İstanbul"/"Zürich" (dropped) and
+    # "São" (shredded). Keep the ASCII behaviour — the word boundary already
+    # excludes camelCase mid-word capitals — by requiring an all-alphabetic
+    # token of length > 1 whose first character is uppercase.
+    for token in re.findall(r"\b\w+\b", cleaned):
+        if len(token) > 1 and token[0].isupper() and token.isalpha():
+            entities["names"].append(token)
     for year in re.findall(r"\b(?:19|20)\d{2}\b", cleaned):
         entities["dates"].append(year)
     month_day_year = re.findall(

src/agent_loop.py

@@ -469,7 +469,7 @@ Bulk delete/archive/mark emails. Use this for "delete all those" after listing e
 {"action": "create_event", "summary": "<event title>", "dtstart": "<natural language or ISO datetime>"}
 ```
 Calendar event management (CalDAV). Actions: `list_events`, `create_event`, `update_event`, `delete_event`, `list_calendars`. \
-For `list_events`: {start?, end?, calendar?}; prefer `start`/`end` for the range, though start_date/end_date and from/to aliases are accepted. \
+For `list_events`: {action: "list_events", start: "YYYY-MM-DDT00:00:00", end: "YYYY-MM-DDT00:00:00", calendar?}; resolve month/week phrases yourself from the Current date and time context and do not pass a loose `query` field. Prefer `start`/`end`; start_time/end_time, start_date/end_date, and from/to aliases are accepted. \
 For `create_event`: {summary, dtstart, dtend?, duration?, calendar?, location?, description?, reminder_minutes?, rrule?}. \
 For `update_event`: {uid, summary?, dtstart?, dtend?, all_day?, location?, description?, event_type?, importance?, rrule?}. Pass `rrule: ""` to remove recurrence and make a repeating event a single event. \
 `dtstart` accepts natural language ("tomorrow at 1pm", "in 2 hours", "next monday 9am") or ISO ("2026-05-12T13:00:00"). \
@@ -2532,11 +2532,17 @@ async def stream_agent_loop(
                         )
                         logger.info(f"[tool-rag] Retrieved tools for query: {sorted(_relevant_tools - ALWAYS_AVAILABLE)}")
                     except asyncio.TimeoutError:
+                        # Leave _relevant_tools unset so the keyword fallback
+                        # below still runs. Hard-coding ALWAYS_AVAILABLE here
+                        # skipped the deterministic keyword hints whenever the
+                        # embedding backend was slow (e.g. a remote endpoint
+                        # cold-loading its model), silently stripping email/
+                        # calendar tools from queries that named them outright.
                         logger.warning(
-                            "[tool-rag] Retrieval exceeded %.1fs; falling back to always-available tools",
+                            "[tool-rag] Retrieval exceeded %.1fs; falling back to keyword tool selection",
                             _TOOL_SELECTION_TIMEOUT_SECONDS,
                         )
-                        _relevant_tools = set(ALWAYS_AVAILABLE)
+                        _relevant_tools = None
         except Exception as e:
             logger.warning(f"[tool-rag] Retrieval failed, using keyword fallback: {e}")
             _relevant_tools = None

src/agent_tools/filesystem_tools.py

@@ -424,8 +424,12 @@ class GrepTool:
                     cmd.append("--ignore-case")
                 if glob_pat:
                     cmd += ["--glob", glob_pat]
+                # --iglob (not --glob) so the exclusion is case-insensitive:
+                # on a case-insensitive filesystem "ID_RSA"/"Known_Hosts"
+                # resolve to the same secret as their lowercase forms, and the
+                # Python fallback below already folds case via _is_sensitive_path.
                 for _pat in _SENSITIVE_FILE_PATTERNS:
-                    cmd += ["--glob", f"!*{_pat}*"]
+                    cmd += ["--iglob", f"!*{_pat}*"]
                 for _d in _CODENAV_SKIP_DIRS:
                     cmd += ["--glob", f"!**/{_d}/**"]
                 cmd += ["--regexp", pattern, root]

src/agent_tools/session_tools.py

@@ -184,8 +184,12 @@ async def send_to_session(content: str, session_id: Optional[str] = None, owner:
     if not sess:
         return {"error": f"Session '{target_sid}' not found"}
 
-    # Owner-scope: reject access to another user's session
-    if owner and getattr(sess, "owner", None) and sess.owner != owner:
+    # Owner-scope: reject access to another user's session. When the caller is
+    # authenticated, a null-owner (legacy / auth-was-off) session is not theirs
+    # either — list_sessions (get_sessions_for_user) and manage_session already
+    # exclude those, so treating it as reachable here let an authenticated agent
+    # read/write a session the other tools hide. Require an exact owner match.
+    if owner and getattr(sess, "owner", None) != owner:
         return {"error": f"Session '{target_sid}' not found"}
 
     if not message:

src/integrations.py

@@ -216,7 +216,14 @@ def _normalize_integration_base_url(base_url: Any) -> str:
 
 
 def _join_integration_url(base_url: str, path: str) -> str:
-    return urljoin(base_url.rstrip("/") + "/", path.lstrip("/"))
+    base = base_url.rstrip("/")
+    rel = path.lstrip("/")
+    if not rel:
+        # A bare "/" must resolve to the base URL itself, not base + "/".
+        # POST-to-base integrations (e.g. Discord webhooks) 404 on the
+        # trailing-slash variant of their URL.
+        return base
+    return urljoin(base + "/", rel)
 
 
 def load_integrations() -> List[Dict[str, Any]]:
@@ -394,6 +401,22 @@ async def execute_api_call(
         return {"error": "Path must not contain a fragment", "exit_code": 1}
 
     url = _join_integration_url(base_url, path)
+
+    # SSRF guard — same check used by the gallery endpoint, embeddings,
+    # CardDAV, and the reminder webhook sender. Link-local / metadata
+    # addresses (169.254.x.x — the cloud credential-exfil vector) are always
+    # rejected; INTEGRATION_API_BLOCK_PRIVATE_IPS=true also blocks RFC-1918 /
+    # loopback for locked-down deployments. Private stays allowed by default
+    # because LAN integrations (Home Assistant, Miniflux, ntfy) are the
+    # primary use case.
+    from src.url_safety import check_outbound_url
+    block_private = os.getenv(
+        "INTEGRATION_API_BLOCK_PRIVATE_IPS", "false"
+    ).lower() == "true"
+    ok, reason = check_outbound_url(url, block_private=block_private)
+    if not ok:
+        return {"error": f"URL rejected: {reason}", "exit_code": 1}
+
     method = method.upper()
 
     # Build headers

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

@@ -556,8 +556,8 @@ FUNCTION_TOOL_SCHEMAS = [
                     "uid": {"type": "string", "description": "Event UID (for update/delete)"},
                     "calendar_href": {"type": "string", "description": "Specific calendar URL (optional; defaults to first calendar)"},
                     "calendar": {"type": "string", "description": "Filter list_events by calendar name or href"},
-                    "start": {"type": "string", "description": "list_events range start (ISO datetime); defaults to today. Prefer start; backend also accepts start_date, range_start, from, dtstart, since."},
-                    "end": {"type": "string", "description": "list_events range end (ISO datetime); defaults to +14 days. Prefer end; backend also accepts end_date, range_end, to, dtend, until."},
+                    "start": {"type": "string", "description": "list_events range start (ISO datetime). Use this for month/week requests after resolving the date range; do not pass a loose query string. Prefer start; backend also accepts start_time, start_date, range_start, from, dtstart, since."},
+                    "end": {"type": "string", "description": "list_events range end (ISO datetime). Use this for month/week requests after resolving the date range; defaults to +14 days only when no range is requested. Prefer end; backend also accepts end_time, end_date, range_end, to, dtend, until."},
                     "event_type": {"type": "string", "description": "Tag / category for the event. Common values: work, personal, health, travel, meal, social, admin, other. Aliases accepted: tag, category, type."},
                     "importance": {"type": "string", "enum": ["low", "normal", "high", "critical"], "description": "Priority level (defaults to 'normal')"},
                     "reminder_minutes": {"type": "integer", "description": "For create_event: create an Odysseus reminder this many minutes before the event, e.g. 5 for 'reminder 5 min before'."},

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):

src/tools/calendar.py

@@ -208,11 +208,20 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
         elif action == "list_events":
             try:
                 start_raw = _first_nonempty_arg(
-                    "start", "start_date", "range_start", "from", "dtstart", "since"
+                    "start", "start_time", "start_date", "range_start", "from", "dtstart", "since"
                 )
                 end_raw = _first_nonempty_arg(
-                    "end", "end_date", "range_end", "to", "dtend", "until"
+                    "end", "end_time", "end_date", "range_end", "to", "dtend", "until"
                 )
+                query_raw = args.get("query") or args.get("date_range") or args.get("range")
+                if query_raw and (not start_raw or not end_raw):
+                    return {
+                        "error": (
+                            "list_events needs explicit start/end ISO datetimes; "
+                            f"resolve the requested range ({query_raw!r}) and call manage_calendar again."
+                        ),
+                        "exit_code": 1,
+                    }
                 if start_raw:
                     start_dt = _parse_dt(start_raw)
                 else:

src/webhook_manager.py

@@ -7,10 +7,12 @@ import ipaddress
 import json
 import logging
 import re
+import ssl
 from datetime import datetime, timezone
 from typing import Optional
 from urllib.parse import urlparse
 
+import httpcore
 import httpx
 
 from src.database import SessionLocal, Webhook
@@ -125,6 +127,128 @@ def validate_webhook_url(url: str) -> str:
     return url
 
 
+def _validated_public_ips(url: str) -> list:
+    """Resolve *url*'s host and return its IPs, raising ValueError if any is
+    private/internal.
+
+    ``validate_webhook_url`` resolves the host to decide accept/reject, but the
+    subsequent ``httpx`` connect re-resolves independently — so a DNS record
+    that flips between the two lookups (rebinding) can slip an internal IP past
+    the check. Callers pin the delivery connection to the IP this function
+    returns, closing that TOCTOU. Fail closed: an unresolvable or partly-private
+    result raises rather than returning a usable IP.
+    """
+    parsed = urlparse(url)
+    hostname = (parsed.hostname or "").strip()
+    if not hostname:
+        raise ValueError("URL must have a hostname")
+    try:
+        literal = ipaddress.ip_address(hostname)
+    except ValueError:
+        literal = None
+    if literal is not None:
+        if _ip_is_private(literal):
+            raise ValueError("URL must not point to private/internal addresses")
+        return [literal]
+    addrs = _resolve_hostname_ips(hostname)
+    if not addrs or any(_ip_is_private(a) for a in addrs):
+        raise ValueError("URL must not point to private/internal addresses")
+    return addrs
+
+
+# httpcore raises its own exception hierarchy; map the ones a simple POST can
+# surface back to their httpx equivalents so callers' `except httpx.*` blocks
+# (and sanitize_error) behave exactly as they did with the default transport.
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedAsyncBackend(httpcore.AsyncNetworkBackend):
+    """Async network backend that routes every TCP connect to a fixed IP.
+
+    httpcore derives TLS SNI and the ``Host`` header from the request URL, not
+    from the connect host, so pinning only the socket destination keeps
+    certificate validation and vhost routing pointed at the original hostname.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress):
+        self._ip = str(ip)
+        self._real = httpcore.AnyIOBackend()
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None,
+                          socket_options=None):
+        return await self._real.connect_tcp(
+            self._ip, port, timeout, local_address, socket_options
+        )
+
+    async def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return await self._real.connect_unix_socket(path, timeout, socket_options)
+
+    async def sleep(self, seconds: float) -> None:
+        return await self._real.sleep(seconds)
+
+
+class _PinnedAsyncTransport(httpx.AsyncBaseTransport):
+    """httpx transport that pins the TCP connect to a pre-resolved public IP.
+
+    Uses only public ``httpcore`` / ``httpx`` APIs. The request URL is passed
+    through unchanged (Host + SNI stay the original hostname); only the socket
+    destination is pinned, closing the DNS-rebinding TOCTOU between the SSRF
+    check and the connect. HTTP/1.1 only — webhook deliveries are small POSTs.
+    """
+
+    def __init__(self, ip: ipaddress._BaseAddress):
+        self._pool = httpcore.AsyncConnectionPool(
+            ssl_context=ssl.create_default_context(),
+            http1=True,
+            http2=False,
+            network_backend=_PinnedAsyncBackend(ip),
+        )
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        core_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:
+            core_resp = await self._pool.handle_async_request(core_req)
+            content = b"".join([chunk async for chunk in core_resp.aiter_stream()])
+            await core_resp.aclose()
+        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=core_resp.status,
+            headers=core_resp.headers,
+            content=content,
+            extensions=core_resp.extensions,
+        )
+
+    async def aclose(self) -> None:
+        await self._pool.aclose()
+
+
 def validate_events(events_str: str) -> str:
     """Validate comma-separated event names. Returns cleaned string."""
     events = [e.strip() for e in events_str.split(",") if e.strip()]
@@ -198,8 +322,11 @@ def sanitize_error(error: str, max_len: int = 200) -> str:
 
 class WebhookManager:
     def __init__(self, api_key_manager=None):
-        # Disable redirects to prevent SSRF via redirect chains
-        self._client = httpx.AsyncClient(timeout=10, follow_redirects=False)
+        # No shared client: each delivery builds a short-lived client whose
+        # transport is pinned to the SSRF-approved IP (see _deliver /
+        # _send_request), so a single reusable client can't be pointed at
+        # different pinned hosts. Redirects stay disabled on every delivery
+        # client to prevent SSRF via redirect chains.
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._api_key_manager = api_key_manager
         # Strong references to in-flight fire-and-forget tasks. asyncio only
@@ -262,11 +389,28 @@ class WebhookManager:
         decrypted = self._decrypt_secret(encrypted_secret)
         await self._deliver(webhook_id, url, decrypted, "webhook.test", {"message": "Test ping from Odysseus"})
 
+    async def _send_request(self, url: str, body: str, headers: dict,
+                            ip: ipaddress._BaseAddress) -> httpx.Response:
+        """POST *body* to *url* with the TCP connect pinned to *ip*.
+
+        Overridable seam: tests replace this to avoid real sockets. Redirects
+        are disabled so a 3xx can't bounce the delivery to another host.
+        """
+        transport = _PinnedAsyncTransport(ip)
+        async with httpx.AsyncClient(
+            timeout=10, follow_redirects=False, transport=transport,
+        ) as client:
+            return await client.post(url, content=body, headers=headers)
+
     async def _deliver(self, webhook_id: str, url: str, secret: Optional[str], event: str, payload: dict):
         """Internal delivery. Never call directly from outside this class (use deliver_test)."""
-        # Re-validate URL at delivery time in case DB was tampered with
+        # Re-validate URL at delivery time in case DB was tampered with, and
+        # capture the exact IPs that passed the check so the connect can be
+        # pinned to one of them (closes the DNS-rebinding TOCTOU: the check
+        # below and the socket connect no longer resolve independently).
         try:
             validate_webhook_url(url)
+            pinned_ips = _validated_public_ips(url)
         except ValueError as e:
             logger.warning(f"Webhook {webhook_id} has invalid URL, skipping: {e}")
             return
@@ -283,7 +427,7 @@ class WebhookManager:
 
         db = SessionLocal()
         try:
-            resp = await self._client.post(url, content=body, headers=headers)
+            resp = await self._send_request(url, body, headers, pinned_ips[0])
             db.query(Webhook).filter(Webhook.id == webhook_id).update({
                 "last_triggered_at": _utcnow(),
                 "last_status_code": resp.status_code,
@@ -305,4 +449,7 @@ class WebhookManager:
             db.close()
 
     async def close(self):
-        await self._client.aclose()
+        # Delivery clients are per-request and closed via their async context
+        # manager, so there is no long-lived client to tear down here. Kept for
+        # API compatibility with callers (e.g. app shutdown).
+        return None

static/js/document.js

@@ -2378,20 +2378,27 @@ import { bindMenuDismiss, dismissOrRemove } from './escMenuStack.js';
   }
 
   // ── WYSIWYG email body helpers ──
+  function _emailPlainTextToHtml(text) {
+    const d = document.createElement('div');
+    d.textContent = text == null ? '' : String(text);
+    return d.innerHTML.replace(/\n/g, '<br>');
+  }
+
   function _emailBodyToHtml(text) {
     const t = (text || '').trim();
     if (!t) return '';
     // If it already contains a formatting/structural HTML tag, it's a saved
-    // WYSIWYG body — use it verbatim. (Checking a leading '<' isn't enough: a
+    // WYSIWYG body — sanitize it before rendering. (Checking a leading '<' isn't enough: a
     // rich body often starts with plain text, e.g. "Hi <b>there</b>".)
-    if (/<\/?(b|i|u|s|strong|em|del|strike|a|p|div|br|ul|ol|li|h[1-3]|blockquote|span|code|pre)\b[^>]*>/i.test(t)) return t;
+    if (/<\/?(b|i|u|s|strong|em|del|strike|a|p|div|br|ul|ol|li|h[1-3]|blockquote|span|code|pre)\b[^>]*>/i.test(t)) {
+      return markdownModule.sanitizeAllowedHtml
+        ? markdownModule.sanitizeAllowedHtml(t)
+        : _emailPlainTextToHtml(t);
+    }
     // Email body: keep author-typed `:shortcode:` text literal. Issue #345
     // (shortcode → emoji) is scoped to chat; do not rewrite colons in mail.
     try { return markdownModule.mdToHtml(text, { shortcodes: false }); }
-    catch (_) {
-      const d = document.createElement('div'); d.textContent = text;
-      return d.innerHTML.replace(/\n/g, '<br>');
-    }
+    catch (_) { return _emailPlainTextToHtml(text); }
   }
   // Mirror the rich body's plain text into the hidden textarea so the existing
   // send / draft / change-detection plumbing (which reads the textarea) stays

static/js/markdown.js

@@ -133,7 +133,7 @@ function _cleanAllowedHtmlOnce(htmlString) {
   return tpl.innerHTML;
 }
 
-function sanitizeAllowedHtml(html) {
+export function sanitizeAllowedHtml(html) {
   const raw = String(html == null ? '' : html);
   // Non-browser context (e.g. a future SSR/Node import): fail closed by
   // escaping rather than trusting the markup.
@@ -824,6 +824,7 @@ export function renderMermaid(container) {
 const markdownModule = {
   escapeHtml,
   mdToHtml,
+  sanitizeAllowedHtml,
   squashOutsideCode,
   renderContent,
   processWithThinking,

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

@@ -78,3 +78,29 @@ async def test_list_events_honors_range_aliases(start_key, end_key):
     summaries = [event["summary"] for event in res["events"]]
     assert summaries == ["Late June planning"]
     assert "between 2126-06-01 and 2126-07-01" in res["response"]
+
+async def test_list_events_rejects_partial_loose_range():
+    from src.tool_implementations import do_manage_calendar
+
+    owner = "calendar-partial-" + uuid.uuid4().hex[:8]
+
+    # Partial: has query and start, but no end
+    res = await do_manage_calendar(json.dumps({
+        "action": "list_events",
+        "query": "July",
+        "start_time": "2126-07-01T00:00:00Z",
+    }), owner=owner)
+
+    assert res.get("exit_code", 1) == 1, res
+    assert "list_events needs explicit start/end" in res.get("error", "")
+
+    # Partial: has query and end, but no start
+    res2 = await do_manage_calendar(json.dumps({
+        "action": "list_events",
+        "query": "July",
+        "end_time": "2126-07-31T23:59:59Z",
+    }), owner=owner)
+
+    assert res2.get("exit_code", 1) == 1, res2
+    assert "list_events needs explicit start/end" in res2.get("error", "")
+

tests/test_calendar_update_event_tz.py

@@ -78,3 +78,36 @@ async def test_update_event_dtstart_anchored_to_user_tz(tokyo_offset):
         assert bool(ev.is_utc) is True
     finally:
         db.close()
+
+
+async def test_list_events_accepts_start_time_end_time_aliases():
+    from src.tool_implementations import do_manage_calendar
+
+    owner = "list-" + uuid.uuid4().hex[:6]
+    created = await do_manage_calendar(json.dumps({
+        "action": "create_event",
+        "summary": "July planning",
+        "dtstart": "2026-07-15T12:00:00Z",
+        "dtend": "2026-07-15T13:00:00Z",
+    }), owner=owner)
+    assert created.get("exit_code", 0) == 0, created
+
+    listed = await do_manage_calendar(json.dumps({
+        "action": "list_events",
+        "start_time": "2026-07-01T00:00:00Z",
+        "end_time": "2026-08-01T00:00:00Z",
+    }), owner=owner)
+    assert listed.get("exit_code", 0) == 0, listed
+    assert "between 2026-07-01 and 2026-08-01" in listed["response"]
+    assert [event["summary"] for event in listed["events"]] == ["July planning"]
+
+
+async def test_list_events_query_without_range_does_not_default_to_two_weeks():
+    from src.tool_implementations import do_manage_calendar
+
+    listed = await do_manage_calendar(json.dumps({
+        "action": "list_events",
+        "query": "July",
+    }), owner="list-" + uuid.uuid4().hex[:6])
+    assert listed.get("exit_code") == 1, listed
+    assert "explicit start/end" in listed["error"]

tests/test_chat_route_tool_policy.py

@@ -14,6 +14,8 @@ from pathlib import Path
 
 import pytest
 
+from src.action_intents import classify_tool_intent
+
 _CHAT_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "chat_routes.py"
 
 
@@ -73,7 +75,7 @@ def test_allow_web_search_reads_from_body_as_fallback():
     )
 
 
-def test_disabled_tools_does_not_bash_when_allow_bash_is_none():
+def test_disabled_tools_respects_missing_vs_explicit_toggles():
     """When allow_bash is not set (None), bash must NOT be unconditionally
     added to disabled_tools.  The per-user privilege check handles it.
     """
@@ -89,8 +91,8 @@ def test_disabled_tools_does_not_bash_when_allow_bash_is_none():
     assert "allow_web_search is not None" in source, (
         "disabled_tools check must guard against allow_web_search being None"
     )
-    assert "_explicit_web_intent" in source and "not _explicit_web_intent" in source, (
-        "explicit web-search requests must override an off web toggle for that turn"
+    assert "and not _explicit_web_intent" not in source, (
+        "explicit allow_web_search=false must not be overridden by prompt web intent"
     )
 
 
@@ -116,7 +118,6 @@ def _build_disabled_tools(
     if (
         allow_web_search is not None
         and str(allow_web_search).lower() != "true"
-        and not explicit_web_intent
     ):
         disabled_tools.add("web_search")
         disabled_tools.add("web_fetch")
@@ -156,15 +157,27 @@ def test_json_body_allow_web_search_false_disables_web():
     assert "web_fetch" in disabled
 
 
-def test_explicit_web_intent_overrides_false_web_toggle_for_turn():
-    """A stale/off web toggle must not remove web tools when the message
-    explicitly asks to use web search."""
+@pytest.mark.parametrize(
+    "message",
+    [
+        "please use web search for current CVEs",
+        "search the web for current CVEs",
+        "can you look up the latest docs",
+    ],
+)
+def test_explicit_false_disables_web_despite_prompt_web_intent(message):
+    """Explicit allow_web_search=false is a hard deny even when the prompt
+    asks for web search."""
+    intent = classify_tool_intent(message)
+    assert intent is not None
+    assert intent.category == "web"
+
     disabled = _build_disabled_tools(
         allow_web_search="false",
         explicit_web_intent=True,
     )
-    assert "web_search" not in disabled
-    assert "web_fetch" not in disabled
+    assert "web_search" in disabled
+    assert "web_fetch" in disabled
 
 
 def test_admin_user_gets_bash_enabled_by_default():

tests/test_code_nav_tools.py

@@ -91,6 +91,30 @@ def test_grep_python_fallback_when_no_rg(repo, monkeypatch):
     assert ".git/config" not in r["output"]
 
 
+@pytest.mark.skipif(shutil.which("rg") is None, reason="targets the ripgrep fast-path")
+def test_grep_skips_case_variant_sensitive_files_rg(repo):
+    """The rg fast-path must exclude deny-listed key files case-insensitively.
+
+    A file whose name is a case variant of a sensitive pattern (e.g. ID_RSA vs
+    id_rsa, Known_Hosts vs known_hosts) points at the same secret on a
+    case-insensitive filesystem, so grep must not return its contents. The
+    Python fallback already folds case via _is_sensitive_path; a plain --glob
+    exclusion is case-sensitive, so it would leak these — this pins the rg path.
+    """
+    token = "GREPSECRET_TOKEN_ZZZ"
+    with open(os.path.join(repo, "notes.txt"), "w") as f:
+        f.write(f"see {token}\n")
+    with open(os.path.join(repo, "ID_RSA"), "w") as f:
+        f.write(f"PRIVATE {token}\n")
+    with open(os.path.join(repo, "Known_Hosts"), "w") as f:
+        f.write(f"host {token}\n")
+    r = _run("grep", f'{{"pattern": "{token}", "path": "{repo}"}}')
+    assert r["exit_code"] == 0
+    assert "notes.txt" in r["output"]        # ordinary matches still returned
+    assert "ID_RSA" not in r["output"]        # case-variant key excluded
+    assert "Known_Hosts" not in r["output"]
+
+
 # ── glob ──────────────────────────────────────────────────────────────────
 
 def test_glob_py(repo):

tests/test_contacts_routes_shim.py

@@ -0,0 +1,46 @@
+"""Regression test for the contacts route shim (slice 2e, #4082/#4071).
+
+The backward-compat shim at ``routes/contacts_routes.py`` uses ``sys.modules``
+replacement so the legacy import path and the canonical ``routes.contacts.*``
+path resolve to the *same* module object. This is required because:
+
+* ``test_carddav_password_encryption.py`` uses string-targeted
+  ``monkeypatch.setattr("routes.contacts_routes.SETTINGS_FILE", ...)`` which
+  must reach the canonical module to take effect;
+* ``test_contacts_add_null_name.py`` / ``test_contacts_carddav_security.py``
+  use ``import routes.contacts_routes as cr`` + ``setattr(cr, ...)``;
+* the module owns mutable state (``_contact_cache``) that must be shared
+  across import paths.
+"""
+
+import importlib
+
+import routes.contacts_routes as _shim_contacts  # noqa: F401
+
+
+def test_legacy_and_canonical_contacts_module_are_same_object():
+    """``import routes.contacts_routes`` must alias the canonical module."""
+    legacy = importlib.import_module("routes.contacts_routes")
+    canonical = importlib.import_module("routes.contacts.contacts_routes")
+    assert legacy is canonical, (
+        "routes.contacts_routes shim must resolve to the canonical "
+        "routes.contacts.contacts_routes module object"
+    )
+
+
+def test_string_targeted_monkeypatch_reaches_canonical(monkeypatch):
+    """String-targeted ``monkeypatch.setattr`` via the legacy path must reach
+    the canonical module.
+
+    ``test_carddav_password_encryption.py`` patches
+    ``"routes.contacts_routes.SETTINGS_FILE"`` as a fixture setup; for that
+    to take effect at runtime, the legacy module name and the canonical
+    module must be identical.
+    """
+    canonical = importlib.import_module("routes.contacts.contacts_routes")
+
+    sentinel = object()
+    monkeypatch.setattr("routes.contacts_routes.setup_contacts_routes", sentinel)
+    assert canonical.setup_contacts_routes is sentinel, (
+        "string-targeted monkeypatch via legacy path did not reach the canonical module"
+    )

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

@@ -16,7 +16,7 @@ in this repo do).
 import ast
 from pathlib import Path
 
-SRC = Path(__file__).resolve().parent.parent / "routes" / "history_routes.py"
+SRC = Path(__file__).resolve().parent.parent / "routes" / "history" / "history_routes.py"
 
 
 def _function_source(src_text, name):

tests/test_history_order_by_timestamp_regression.py

@@ -25,7 +25,7 @@ from sqlalchemy.orm import sessionmaker
 from core.database import Base, ChatMessage as DbChatMessage, Session as DbSession
 
 
-HISTORY_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "history_routes.py"
+HISTORY_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "history" / "history_routes.py"
 
 
 def test_chatmessage_model_has_timestamp_not_created_at():

tests/test_history_routes_shim.py

@@ -0,0 +1,43 @@
+"""Regression test for the history route shim (slice 2d, #4082/#4071).
+
+The backward-compat shim at ``routes/history_routes.py`` uses ``sys.modules``
+replacement so the legacy import path and the canonical ``routes.history.*``
+path resolve to the *same* module object. This is required because
+``test_history_compact_tool_calls.py`` and ``test_fork_session_metadata.py``
+do ``import routes.history_routes as history_routes`` followed by
+``monkeypatch.setattr(history_routes, "_verify_session_owner", ...)`` — for
+those patches to take effect at runtime, the legacy module object and the
+canonical one must be identical. This test pins that contract.
+"""
+
+import importlib
+
+import routes.history_routes as _shim_history  # noqa: F401
+
+
+def test_legacy_and_canonical_history_module_are_same_object():
+    """``import routes.history_routes`` must alias the canonical module."""
+    legacy = importlib.import_module("routes.history_routes")
+    canonical = importlib.import_module("routes.history.history_routes")
+    assert legacy is canonical, (
+        "routes.history_routes shim must resolve to the canonical "
+        "routes.history.history_routes module object"
+    )
+
+
+def test_monkeypatch_via_legacy_alias_reaches_canonical(monkeypatch):
+    """Patching through the legacy alias must reach the canonical module.
+
+    Several history tests do ``import routes.history_routes as history_routes``
+    followed by ``monkeypatch.setattr(history_routes, "_verify_session_owner",
+    ...)``. For that to take effect at runtime, the legacy module object and the
+    canonical one must be identical.
+    """
+    legacy = importlib.import_module("routes.history_routes")
+    canonical = importlib.import_module("routes.history.history_routes")
+
+    sentinel = object()
+    monkeypatch.setattr(legacy, "setup_history_routes", sentinel)
+    assert canonical.setup_history_routes is sentinel, (
+        "monkeypatch via legacy alias did not reach the canonical module"
+    )

tests/test_integration_api_call_ssrf.py

@@ -0,0 +1,99 @@
+"""Regression: execute_api_call must run the outbound SSRF guard.
+
+The api_call agent tool lets the LLM drive HTTP requests against a
+user-configured integration base_url. Before this guard, a base_url (or a
+hostname resolving) to the cloud metadata range was requested server-side
+with the integration's auth headers attached. execute_api_call now validates
+the joined URL with src.url_safety.check_outbound_url before connecting:
+link-local/metadata is always rejected; RFC-1918/loopback only when
+INTEGRATION_API_BLOCK_PRIVATE_IPS=true (LAN integrations are the primary
+use case, so private stays allowed by default).
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src import integrations
+
+
+def _integration(base_url):
+    return {
+        "id": "test_integ",
+        "name": "TestInteg",
+        "enabled": True,
+        "base_url": base_url,
+        "auth_type": "bearer",
+        "api_key": "secret-token",
+        "auth_header": "",
+        "auth_param": "",
+        "description": "",
+        "preset": "",
+    }
+
+
+async def _call(base_url, path="/items"):
+    resp = MagicMock()
+    resp.status_code = 200
+    resp.headers = {"content-type": "application/json"}
+    resp.json.return_value = {"ok": True}
+    resp.text = '{"ok": true}'
+
+    client = AsyncMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+    client.request = AsyncMock(return_value=resp)
+
+    with (
+        patch.object(integrations, "_find_integration",
+                     return_value=_integration(base_url)),
+        patch("httpx.AsyncClient", return_value=client),
+    ):
+        result = await integrations.execute_api_call("test_integ", "GET", path)
+    return result, client
+
+
+@pytest.mark.asyncio
+async def test_metadata_ip_base_url_is_rejected_without_requesting():
+    result, client = await _call("http://169.254.169.254")
+
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_hostname_resolving_to_metadata_ip_is_rejected(monkeypatch):
+    """DNS-based variant: an innocuous-looking hostname that resolves into
+    the link-local range must be caught by the resolver check."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["169.254.169.254"])
+    result, client = await _call("http://internal.attacker.example")
+
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_public_ip_base_url_still_requests():
+    # Public literal — no DNS involved.
+    result, client = await _call("http://93.184.216.34")
+
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_private_base_url_allowed_by_default_blocked_with_knob(monkeypatch):
+    # Local-first default: LAN integrations (Home Assistant etc.) must work.
+    monkeypatch.delenv("INTEGRATION_API_BLOCK_PRIVATE_IPS", raising=False)
+    result, client = await _call("http://192.168.1.50")
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+
+    # Locked-down deployments opt in to a full private/loopback block.
+    monkeypatch.setenv("INTEGRATION_API_BLOCK_PRIVATE_IPS", "true")
+    result, client = await _call("http://192.168.1.50")
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()

tests/test_integrations_api_call_truncation.py

@@ -83,6 +83,9 @@ async def _call(json_data, status=200):
     with (
         patch.object(integrations, "_find_integration", return_value=DUMMY_INTEGRATION),
         patch("httpx.AsyncClient", return_value=mock_client),
+        # api.example.com doesn't resolve; the SSRF guard would fail closed.
+        # These tests are about truncation, so stub the guard open.
+        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
     ):
         return await integrations.execute_api_call("test_integ", "GET", "/items")
 
@@ -98,6 +101,9 @@ async def _call_with_integration(integration, path="/items"):
     with (
         patch.object(integrations, "_find_integration", return_value=integration),
         patch("httpx.AsyncClient", return_value=mock_client),
+        # api.example.com doesn't resolve; the SSRF guard would fail closed.
+        # These tests are about URL joining, so stub the guard open.
+        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
     ):
         result = await integrations.execute_api_call("test_integ", "GET", path)
     return result, mock_client

tests/test_integrations_url_join.py

@@ -0,0 +1,116 @@
+"""Tests for integration URL construction in execute_api_call.
+
+Covers the trailing-slash regression from #5138: a bare "/" path must
+resolve to the base URL itself, not base + "/". Discord webhook URLs
+404 on the trailing-slash variant, so api_call against a
+POST-to-base integration silently failed.
+"""
+import sys
+import types
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Minimal stubs so src.integrations can be imported without heavy deps
+# ---------------------------------------------------------------------------
+
+for mod_name in ("core", "core.atomic_io", "core.platform_compat"):
+    if mod_name not in sys.modules:
+        sys.modules[mod_name] = types.ModuleType(mod_name)
+
+core_atomic = sys.modules["core.atomic_io"]
+if not hasattr(core_atomic, "atomic_write_json"):
+    core_atomic.atomic_write_json = lambda *a, **kw: None  # type: ignore
+
+core_compat = sys.modules["core.platform_compat"]
+if not hasattr(core_compat, "safe_chmod"):
+    core_compat.safe_chmod = lambda *a, **kw: None  # type: ignore
+
+if "src.secret_storage" not in sys.modules:
+    stub = types.ModuleType("src.secret_storage")
+    stub.encrypt = lambda s: s  # type: ignore
+    stub.decrypt = lambda s: s  # type: ignore
+    stub.is_encrypted = lambda s: False  # type: ignore
+    sys.modules["src.secret_storage"] = stub
+
+if "src.constants" not in sys.modules:
+    stub_c = types.ModuleType("src.constants")
+    stub_c.DATA_DIR = "/tmp"  # type: ignore
+    stub_c.INTEGRATIONS_FILE = "/tmp/integrations_test.json"  # type: ignore
+    stub_c.SETTINGS_FILE = "/tmp/settings_test.json"  # type: ignore
+    sys.modules["src.constants"] = stub_c
+
+from src import integrations  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# _join_integration_url unit tests
+# ---------------------------------------------------------------------------
+
+WEBHOOK_BASE = "https://discord.com/api/webhooks/123/tokentokentoken"
+
+
+@pytest.mark.parametrize(
+    "base,path,expected",
+    [
+        # Bare "/" (the minimum path execute_api_call accepts) must not
+        # grow a trailing slash — Discord webhooks 404 on it (#5138).
+        (WEBHOOK_BASE, "/", WEBHOOK_BASE),
+        (WEBHOOK_BASE + "/", "/", WEBHOOK_BASE),
+        (WEBHOOK_BASE, "", WEBHOOK_BASE),
+        # Normal paths keep joining exactly as before.
+        ("http://api.example.com", "/items", "http://api.example.com/items"),
+        ("http://api.example.com/", "/items", "http://api.example.com/items"),
+        ("http://host/base", "/v1/me", "http://host/base/v1/me"),
+        # A deliberate trailing slash inside a non-empty path is preserved
+        # (e.g. linkding's /api/tags/, Home Assistant's /api/).
+        ("http://host", "/api/tags/", "http://host/api/tags/"),
+        ("http://host", "/api/", "http://host/api/"),
+    ],
+)
+def test_join_integration_url(base, path, expected):
+    assert integrations._join_integration_url(base, path) == expected
+
+
+# ---------------------------------------------------------------------------
+# Behavioral test through execute_api_call
+# ---------------------------------------------------------------------------
+
+DISCORD_INTEGRATION = {
+    "id": "discord_test",
+    "name": "Discord Webhook",
+    "enabled": True,
+    "base_url": WEBHOOK_BASE,
+    "auth_type": "none",
+    "api_key": "",
+    "auth_header": "",
+    "auth_param": "",
+    "description": "",
+    "preset": "discord_webhook",
+}
+
+
+@pytest.mark.asyncio
+async def test_api_call_root_path_has_no_trailing_slash():
+    mock_resp = MagicMock()
+    mock_resp.status_code = 204
+    mock_resp.headers = {"content-type": "text/plain"}
+    mock_resp.text = ""
+
+    mock_client = AsyncMock()
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=None)
+    mock_client.request = AsyncMock(return_value=mock_resp)
+
+    with (
+        patch.object(integrations, "_find_integration", return_value=DISCORD_INTEGRATION),
+        patch("httpx.AsyncClient", return_value=mock_client),
+    ):
+        result = await integrations.execute_api_call(
+            "discord_test", "POST", "/", body={"content": "test"}
+        )
+
+    assert result.get("exit_code") == 0
+    requested_url = mock_client.request.call_args.args[1]
+    assert requested_url == WEBHOOK_BASE

tests/test_markdown_dom_xss_helpers.py

@@ -23,3 +23,16 @@ def test_markdown_raw_html_sanitizer_strips_scriptable_css():
     assert "if (name === 'style')" in src
     assert r"javascript:|vbscript:|data:|expression\(" in src
     assert "el.removeAttribute(attr.name);" in src
+
+
+def test_email_rich_body_render_path_reuses_raw_html_sanitizer():
+    markdown_src = (_REPO / "static" / "js" / "markdown.js").read_text(encoding="utf-8")
+    document_src = (_REPO / "static" / "js" / "document.js").read_text(encoding="utf-8")
+    email_body_helper = document_src.split("function _emailBodyToHtml(text)", 1)[1].split(
+        "  // Mirror the rich body's plain text", 1
+    )[0]
+
+    assert "export function sanitizeAllowedHtml(html)" in markdown_src
+    assert "sanitizeAllowedHtml," in markdown_src
+    assert "markdownModule.sanitizeAllowedHtml(t)" in email_body_helper
+    assert "return t;" not in email_body_helper

tests/test_model_helper_owner_scope.py

@@ -34,7 +34,7 @@ def test_task_parse_resolves_with_owner_scope():
 
 
 def test_history_compact_resolves_with_owner_scope():
-    body = _function_source("routes/history_routes.py", "compact_session")
+    body = _function_source("routes/history/history_routes.py", "compact_session")
     assert "owner = effective_user(request)" in body
     assert 'resolve_endpoint("utility", owner=owner or None)' in body
 

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

@@ -0,0 +1,117 @@
+"""Regression: the reminder ntfy sender must run the same SSRF guard as the
+webhook sender.
+
+The webhook branch of dispatch_reminder validates its target with
+src.url_safety.check_outbound_url before posting; the ntfy branch posted to
+the integration's base_url with no check, so a base_url pointing at the cloud
+metadata range (169.254.169.254) was fetched server-side — with the
+integration's Authorization header attached — every time a reminder fired.
+"""
+import asyncio
+from unittest.mock import MagicMock, patch
+
+import httpx
+
+from routes.note_routes import dispatch_reminder
+
+
+def _ntfy_integration(base_url):
+    return [{
+        "preset": "ntfy",
+        "enabled": True,
+        "base_url": base_url,
+        "api_key": "secret-token",
+        "name": "ntfy",
+    }]
+
+
+def _settings(**extra):
+    return {
+        "reminder_channel": "ntfy",
+        "reminder_llm_synthesis": False,
+        "reminder_ntfy_topic": "reminders",
+        **extra,
+    }
+
+
+class _SpyAsyncClient:
+    """Stands in for httpx.AsyncClient; records posts, returns success."""
+    calls = []
+
+    def __init__(self, **kwargs):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *a):
+        pass
+
+    async def post(self, url, **kw):
+        _SpyAsyncClient.calls.append(url)
+        resp = MagicMock()
+        resp.is_success = True
+        resp.status_code = 200
+        return resp
+
+
+def _dispatch():
+    return asyncio.run(dispatch_reminder(
+        "Title", "Body", note_id="", queue_browser=True,
+        settings_override=_settings(),
+    ))
+
+
+def test_metadata_ip_ntfy_base_url_is_rejected_and_not_fetched():
+    _SpyAsyncClient.calls = []
+    with (
+        patch("src.integrations.load_integrations",
+              return_value=_ntfy_integration("http://169.254.169.254")),
+        patch.object(httpx, "AsyncClient", _SpyAsyncClient),
+    ):
+        result = _dispatch()
+
+    assert _SpyAsyncClient.calls == [], "metadata address must never be fetched"
+    assert result["ntfy_sent"] is False
+    assert "rejected" in result["ntfy_error"].lower()
+
+
+def test_public_ntfy_base_url_still_sends():
+    _SpyAsyncClient.calls = []
+    with (
+        # 93.184.216.34 is a public literal — no DNS resolution involved.
+        patch("src.integrations.load_integrations",
+              return_value=_ntfy_integration("http://93.184.216.34")),
+        patch.object(httpx, "AsyncClient", _SpyAsyncClient),
+    ):
+        result = _dispatch()
+
+    assert _SpyAsyncClient.calls == ["http://93.184.216.34/reminders"]
+    assert result["ntfy_sent"] is True
+    assert result["ntfy_error"] == ""
+
+
+def test_private_ntfy_base_url_blocked_only_with_env_knob(monkeypatch):
+    # Default (local-first): a LAN ntfy server is a normal setup and must work.
+    _SpyAsyncClient.calls = []
+    monkeypatch.delenv("REMINDER_WEBHOOK_BLOCK_PRIVATE_IPS", raising=False)
+    with (
+        patch("src.integrations.load_integrations",
+              return_value=_ntfy_integration("http://192.168.1.50")),
+        patch.object(httpx, "AsyncClient", _SpyAsyncClient),
+    ):
+        result = _dispatch()
+    assert result["ntfy_sent"] is True
+
+    # Locked-down deployments: the same knob the webhook branch honors.
+    _SpyAsyncClient.calls = []
+    monkeypatch.setenv("REMINDER_WEBHOOK_BLOCK_PRIVATE_IPS", "true")
+    with (
+        patch("src.integrations.load_integrations",
+              return_value=_ntfy_integration("http://192.168.1.50")),
+        patch.object(httpx, "AsyncClient", _SpyAsyncClient),
+    ):
+        result = _dispatch()
+    assert _SpyAsyncClient.calls == []
+    assert result["ntfy_sent"] is False
+    assert "rejected" in result["ntfy_error"].lower()

tests/test_research_routes_path_confinement.py

@@ -1,6 +1,6 @@
 """Path-confinement regression tests for research routes.
 
-Covers the CodeQL py/path-injection alert cluster (#552-#567) in
+Covers the CodeQL py/path-injection alert cluster (#552-#567 and #594) in
 routes/research/research_routes.py:
   - _owns_in_memory disk fallback (alerts #552, #553)
   - _assert_owns_research (alerts #554, #555)
@@ -20,7 +20,11 @@ import pytest
 from fastapi import HTTPException
 
 from routes.research_routes import setup_research_routes
-from routes.research.research_routes import _confine_research_path
+from routes.research.research_routes import (
+    _find_owned_research_path,
+    _find_research_path,
+    _require_research_path,
+)
 
 
 @pytest.fixture(autouse=True)
@@ -58,15 +62,30 @@ def _research_handler():
 
 
 # ---------------------------------------------------------------------------
-# Helper-level tests — _confine_research_path
+# Helper-level tests
 # ---------------------------------------------------------------------------
 
-def test_confine_allows_valid_session_id(tmp_path, monkeypatch):
+def test_find_returns_existing_trusted_research_path(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    expected = _write_research(data_dir, "rp-abc123de4567", owner="alice")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+    assert _find_research_path("rp-abc123de4567") == expected.resolve()
+
+
+def test_find_returns_none_for_missing_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()
+    assert _find_research_path("rp-missing12345") is None
+
+
+def test_require_returns_404_for_missing_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))
+    with pytest.raises(HTTPException) as exc:
+        _require_research_path("rp-missing12345")
+    assert exc.value.status_code == 404
 
 
 @pytest.mark.parametrize("bad_id", [
@@ -79,17 +98,45 @@ def test_confine_allows_valid_session_id(tmp_path, monkeypatch):
     "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))
+def test_find_rejects_bad_session_ids_before_enumeration(monkeypatch, bad_id):
+    storage_root = MagicMock()
+    monkeypatch.setattr(
+        "routes.research.research_routes._research_storage_root",
+        MagicMock(return_value=storage_root),
+    )
     with pytest.raises(HTTPException) as exc:
-        _confine_research_path(bad_id)
+        _find_research_path(bad_id)
     assert exc.value.status_code == 400
+    storage_root.glob.assert_not_called()
 
 
-def test_confine_rejects_symlink_escape(tmp_path, monkeypatch):
-    """A symlink inside DEEP_RESEARCH_DIR that resolves outside is rejected."""
+def test_find_matches_names_from_trusted_enumeration_without_joining_input(
+    tmp_path, monkeypatch
+):
+    """Pin the CodeQL-friendly lookup: match a glob result, never root / input."""
+    data_dir = tmp_path / "deep_research"
+    expected = _write_research(data_dir, "rp-abc123de4567", owner="alice").resolve()
+
+    class EnumeratedRoot:
+        def glob(self, pattern):
+            assert pattern == "*.json"
+            return [expected]
+
+        def __fspath__(self):
+            return str(data_dir.resolve())
+
+        def __truediv__(self, _other):
+            raise AssertionError("user-derived path segment was joined to root")
+
+    monkeypatch.setattr(
+        "routes.research.research_routes._research_storage_root",
+        lambda: EnumeratedRoot(),
+    )
+    assert _find_research_path("rp-abc123de4567") == expected
+
+
+def test_find_ignores_symlink_escape(tmp_path, monkeypatch):
+    """A matching symlink that resolves outside is not a trusted file."""
     data_dir = tmp_path / "deep_research"
     outside = tmp_path / "outside"
     data_dir.mkdir()
@@ -102,9 +149,24 @@ def test_confine_rejects_symlink_escape(tmp_path, monkeypatch):
         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
+    assert _find_research_path("rp-linktest1234") is None
+
+
+
+def test_find_owned_returns_path_for_matching_owner(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    expected = _write_research(data_dir, "rp-ownedalice1", owner="alice")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    assert _find_owned_research_path("rp-ownedalice1", "alice") == expected.resolve()
+
+
+def test_find_owned_returns_none_for_other_owner(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    _write_research(data_dir, "rp-ownedbybob12", owner="bob")
+    monkeypatch.setattr("routes.research_routes.DEEP_RESEARCH_DIR", str(data_dir))
+
+    assert _find_owned_research_path("rp-ownedbybob12", "alice") is None
 
 
 # ---------------------------------------------------------------------------
@@ -120,6 +182,24 @@ def test_detail_returns_data_for_owner(tmp_path):
     assert out["query"] == "valid query"
 
 
+def test_detail_returns_404_for_missing_valid_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="rp-missing12345", request=_request("alice")))
+    assert exc.value.status_code == 404
+
+
+def test_detail_hides_other_owners_research_with_404(tmp_path):
+    data_dir = tmp_path / "deep_research"
+    _write_research(data_dir, "rp-ownedbybob12", owner="bob")
+    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-ownedbybob12", request=_request("alice")))
+    assert exc.value.status_code == 404
+
+
 # ---------------------------------------------------------------------------
 # Route-level tests — traversal and injection rejected
 # ---------------------------------------------------------------------------
@@ -214,28 +294,76 @@ def test_detail_traversal_does_not_read_outside_file(tmp_path, monkeypatch):
 # 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."""
+def _write_outside_symlink(tmp_path, session_id: str, data: dict):
     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"
+    outside_file = outside_dir / f"{session_id}.json"
+    outside_file.write_text(json.dumps(data), encoding="utf-8")
+    link = data_dir / f"{session_id}.json"
     try:
         link.symlink_to(outside_file)
     except (AttributeError, NotImplementedError, OSError) as e:
         pytest.skip(f"symlinks unavailable: {e}")
+    return data_dir, outside_file
+
+
+def test_detail_rejects_symlink_escape(tmp_path, monkeypatch):
+    """research_detail never reads a matching symlink outside the root."""
+    data_dir, _ = _write_outside_symlink(
+        tmp_path,
+        "rp-linktest5678",
+        {"owner": "alice", "result": "secret"},
+    )
     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
+    assert exc.value.status_code == 404
+
+
+def test_archive_does_not_write_through_symlink_escape(tmp_path, monkeypatch):
+    data_dir, outside_file = _write_outside_symlink(
+        tmp_path,
+        "rp-linkarchive1",
+        {"owner": "alice", "archived": False},
+    )
+    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="rp-linkarchive1",
+                request=_request("alice"),
+                archived=True,
+            )
+        )
+    assert exc.value.status_code == 404
+    assert json.loads(outside_file.read_text(encoding="utf-8"))["archived"] is False
+
+
+def test_delete_does_not_unlink_symlink_escape(tmp_path, monkeypatch):
+    data_dir, outside_file = _write_outside_symlink(
+        tmp_path,
+        "rp-linkdelete12",
+        {"owner": "alice"},
+    )
+    link = data_dir / "rp-linkdelete12.json"
+    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")
+    out = asyncio.run(
+        target(session_id="rp-linkdelete12", request=_request("alice"))
+    )
+    assert out == {"deleted": False}
+    assert link.is_symlink()
+    assert outside_file.exists()
 
 
 # ---------------------------------------------------------------------------
@@ -243,14 +371,15 @@ def test_detail_rejects_symlink_escape(tmp_path, monkeypatch):
 # ---------------------------------------------------------------------------
 
 def test_owner_scoped_paths_stay_within_research_root(tmp_path, monkeypatch):
-    """Owner-scoped session IDs never produce paths outside DEEP_RESEARCH_DIR."""
+    """Owner-scoped persisted files resolve within 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)
+        _write_research(data_dir, session_id, owner="alice")
+        path = _require_research_path(session_id)
         assert path.resolve().is_relative_to(root), (
             f"{session_id!r} produced path outside research root: {path}"
         )
@@ -271,3 +400,166 @@ def test_spinoff_rejects_traversal(bad_id):
     with pytest.raises(HTTPException) as exc:
         asyncio.run(target(session_id=bad_id, request=_request("alice")))
     assert exc.value.status_code == 400
+
+def test_result_peek_uses_single_disk_lookup_for_completed_result(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    path = _write_research(
+        data_dir,
+        "rp-peeksingle1",
+        owner="alice",
+        result="saved result",
+        sources=["s1"],
+        raw_findings=["f1"],
+        category="security",
+    ).resolve()
+
+    calls = []
+
+    def fake_find_owned(session_id, user):
+        calls.append((session_id, user))
+        return path
+
+    monkeypatch.setattr(
+        "routes.research.research_routes._find_owned_research_path",
+        fake_find_owned,
+    )
+
+    handler = _research_handler()
+    handler.get_result.return_value = None
+    router = setup_research_routes(handler)
+    target = _route(router, "/api/research/result-peek/{session_id}", "POST")
+
+    out = asyncio.run(target(session_id="rp-peeksingle1", request=_request("alice")))
+
+    assert out["result"] == "saved result"
+    assert out["sources"] == ["s1"]
+    assert out["raw_findings"] == ["f1"]
+    assert out["category"] == "security"
+    assert calls == [("rp-peeksingle1", "alice")]
+
+
+def test_spinoff_uses_single_disk_lookup_for_completed_result(tmp_path, monkeypatch):
+    data_dir = tmp_path / "deep_research"
+    path = _write_research(
+        data_dir,
+        "rp-spinsingle1",
+        owner="alice",
+        result="saved report",
+        sources=["s1", "s2"],
+        query="original query",
+    ).resolve()
+
+    calls = []
+
+    def fake_find_owned(session_id, user):
+        calls.append((session_id, user))
+        return path
+
+    class FakeSession:
+        endpoint_url = ""
+        model = ""
+        headers = {}
+
+        def __init__(self):
+            self.messages = []
+
+        def add_message(self, message):
+            self.messages.append(message)
+
+    class FakeSessionManager:
+        def __init__(self):
+            self.created = None
+
+        def get_session(self, session_id):
+            raise KeyError(session_id)
+
+        def create_session(self, **kwargs):
+            self.created = FakeSession()
+            return self.created
+
+        def save_sessions(self):
+            pass
+
+    monkeypatch.setattr(
+        "routes.research.research_routes._find_owned_research_path",
+        fake_find_owned,
+    )
+    monkeypatch.setattr(
+        "routes.research.research_routes.resolve_endpoint",
+        lambda *_args, **_kwargs: ("http://endpoint/v1", "model", {}),
+    )
+
+    handler = _research_handler()
+    handler.get_result.return_value = None
+    handler.get_sources.return_value = []
+    session_manager = FakeSessionManager()
+    router = setup_research_routes(handler, session_manager=session_manager)
+    target = _route(router, "/api/research/spinoff/{session_id}", "POST")
+
+    out = asyncio.run(target(session_id="rp-spinsingle1", request=_request("alice")))
+
+    assert out["name"] == "Follow-up: original query"
+    assert out["source_count"] == 2
+    assert calls == [("rp-spinsingle1", "alice")]
+    assert session_manager.created is not None
+    assert session_manager.created.messages
+
+def test_spinoff_reads_saved_query_for_done_active_task(tmp_path, monkeypatch):
+    session_id = "rp-activedone1"
+    data_dir = tmp_path / "deep_research"
+    _write_research(
+        data_dir,
+        session_id,
+        owner="alice",
+        result="saved report",
+        sources=["s1"],
+        query="completed query",
+    )
+
+    class FakeSession:
+        endpoint_url = ""
+        model = ""
+        headers = {}
+
+        def __init__(self):
+            self.messages = []
+
+        def add_message(self, message):
+            self.messages.append(message)
+
+    class FakeSessionManager:
+        def __init__(self):
+            self.created = None
+
+        def get_session(self, session_id):
+            raise KeyError(session_id)
+
+        def create_session(self, **kwargs):
+            self.created = FakeSession()
+            return self.created
+
+        def save_sessions(self):
+            pass
+
+    monkeypatch.setattr(
+        "routes.research.research_routes.resolve_endpoint",
+        lambda *_args, **_kwargs: ("http://endpoint/v1", "model", {}),
+    )
+
+    handler = _research_handler()
+    handler._active_tasks[session_id] = {"owner": "alice", "status": "done"}
+    handler.get_result.return_value = None
+    handler.get_sources.return_value = []
+
+    session_manager = FakeSessionManager()
+    router = setup_research_routes(handler, session_manager=session_manager)
+    target = _route(router, "/api/research/spinoff/{session_id}", "POST")
+
+    out = asyncio.run(target(session_id=session_id, request=_request("alice")))
+
+    assert out["name"] == "Follow-up: completed query"
+    assert out["source_count"] == 1
+    assert session_manager.created is not None
+    primer = session_manager.created.messages[0].content
+    assert "completed query" in primer
+    assert "(not recorded)" not in primer

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

@@ -0,0 +1,31 @@
+"""Regression: _extract_entities must find non-ASCII capitalized names.
+
+The name extractor used the ASCII-only class [A-Z][a-zA-Z]+, so a query like
+"İstanbul weather" or "Zürich hotels" yielded no name entities at all, and
+"São Paulo" lost "São" — non-English/accented place and proper names were
+silently dropped from query enhancement. Detection is now Unicode-aware;
+ASCII behaviour (including camelCase mid-word capitals not counting as names)
+is preserved.
+"""
+from services.search.query import _extract_entities
+
+
+def _names(q):
+    return _extract_entities(q)["names"]
+
+
+def test_non_ascii_names_are_extracted():
+    assert "İstanbul" in _names("İstanbul weather")
+    assert "Zürich" in _names("Zürich hotels")
+    assert set(_names("trip to São Paulo")) >= {"São", "Paulo"}
+
+
+def test_ascii_names_unchanged():
+    assert _names("What did Alice do in 2024") == ["Alice"]
+    assert _names("news about OpenAI and Google") == ["OpenAI", "Google"]
+
+
+def test_lowercase_camelcase_and_numbers_are_not_names():
+    assert _names("the iphone price") == []
+    assert _names("iPhone price") == []       # mid-word capital is not a name
+    assert _names("top 50 albums") == []

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

@@ -137,6 +137,55 @@ def test_no_session_manager_is_handled(monkeypatch):
     assert "error" in res or "results" in res
 
 
+class _FakeSession:
+    def __init__(self, owner, name, history):
+        self.owner = owner
+        self.name = name
+        self.endpoint_url = "http://x"
+        self.model = "fixture-tool-model"  # offline path: returns transcript, no network
+        self._history = history
+        self.added = []
+
+    def get_context_messages(self):
+        return list(self._history)
+
+    def add_message(self, m):
+        self.added.append(m)
+
+
+class _FakeMgr:
+    def __init__(self, sessions):
+        self._s = sessions
+
+    def get_session(self, sid):
+        return self._s.get(sid)
+
+
+def test_send_to_session_blocks_null_owner_for_authenticated_caller(monkeypatch):
+    # An authenticated caller must not reach a null-owner (legacy / auth-was-off)
+    # session: list_sessions and manage_session already hide those, so this path
+    # was the inconsistency — it let an agent read/write a session the other
+    # tools exclude. Mirrors the calendar owner=None hardening.
+    null_sess = _FakeSession(None, "Secret", [{"role": "user", "content": "PIN 4321"}])
+    bob_sess = _FakeSession("bob", "Bob", [{"role": "user", "content": "bob secret"}])
+    monkeypatch.setattr(st, "get_session_manager",
+                        lambda: _FakeMgr({"nsid": null_sess, "bsid": bob_sess}))
+
+    # authenticated alice: null-owner session is not-found and its history is not leaked
+    r = asyncio.run(st.send_to_session("nsid\nhello", owner="alice"))
+    assert r.get("error", "").endswith("not found")
+    assert "4321" not in str(r)
+    assert null_sess.added == []  # nothing written into it either
+
+    # authenticated alice still cannot reach another real user's session
+    r2 = asyncio.run(st.send_to_session("bsid\nhello", owner="alice"))
+    assert r2.get("error", "").endswith("not found")
+
+    # auth disabled (no owner): single-user still reaches the null-owner session
+    r3 = asyncio.run(st.send_to_session("nsid\nhello", owner=None))
+    assert r3.get("offline_transcript") is True
+
+
 def test_dispatched_via_registry_not_dispatch_ai_tool():
     source = (Path(__file__).resolve().parent.parent / "src" / "tool_execution.py").read_text(encoding="utf-8")
     assert 'elif tool in ("create_session", "list_sessions", "send_to_session", "manage_session"):' in source

tests/test_webhook_dns_rebinding_pin.py

@@ -0,0 +1,156 @@
+"""Regression: webhook delivery must pin the TCP connect to the SSRF-approved IP.
+
+validate_webhook_url resolves the host to accept/reject, but the delivery
+connect previously re-resolved independently — a DNS record flipping between
+the two lookups (rebinding) could slip an internal IP past the check. _deliver
+now resolves+validates once via _validated_public_ips and pins the connect to
+that IP through _PinnedAsyncTransport. These tests drive the real transport
+against local servers so the pin is exercised end-to-end, not mocked away.
+"""
+import asyncio
+import http.server
+import ipaddress
+import socketserver
+import threading
+
+import pytest
+
+from tests.helpers.import_state import clear_module, preserve_import_state
+
+import os
+import sys
+from unittest.mock import patch
+
+with patch.dict(os.environ, {"DATABASE_URL": "sqlite:///:memory:"}), \
+        preserve_import_state("src.database", "core.database"):
+    clear_module("src.database")
+    _core_database = sys.modules.get("core.database")
+    if _core_database is not None and not getattr(_core_database, "__file__", None):
+        del sys.modules["core.database"]
+    import src.webhook_manager as wm
+
+
+# ---------------------------------------------------------------------------
+# _validated_public_ips
+# ---------------------------------------------------------------------------
+
+def test_validated_public_ips_rejects_metadata_literal():
+    with pytest.raises(ValueError):
+        wm._validated_public_ips("http://169.254.169.254/")
+
+
+def test_validated_public_ips_rejects_loopback_literal():
+    with pytest.raises(ValueError):
+        wm._validated_public_ips("http://127.0.0.1/")
+
+
+def test_validated_public_ips_returns_public_literal():
+    ips = wm._validated_public_ips("http://93.184.216.34/")
+    assert ips == [ipaddress.ip_address("93.184.216.34")]
+
+
+def test_validated_public_ips_rejects_hostname_resolving_private(monkeypatch):
+    # Rebinding shape: a hostname that (now) resolves into loopback space.
+    monkeypatch.setattr(wm, "_resolve_hostname_ips",
+                        lambda h: [ipaddress.ip_address("127.0.0.1")])
+    with pytest.raises(ValueError):
+        wm._validated_public_ips("http://evil.rebind.example/")
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: the pinned transport actually routes to the pinned IP
+# ---------------------------------------------------------------------------
+
+def _serve(handler):
+    srv = socketserver.TCPServer(("127.0.0.1", 0), handler)
+    port = srv.server_address[1]
+    threading.Thread(target=srv.serve_forever, daemon=True).start()
+    return srv, port
+
+
+def test_pinned_transport_connects_to_pinned_ip():
+    """A request whose URL host is a throwaway hostname is still delivered to
+    the pinned loopback IP — proving the socket destination comes from the pin,
+    not from resolving the URL host."""
+    hits = []
+
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def do_POST(self):  # noqa: N802
+            length = int(self.headers.get("Content-Length", 0))
+            self.rfile.read(length)
+            hits.append(self.path)
+            self.send_response(204)
+            self.end_headers()
+
+        def log_message(self, *a):
+            pass
+
+    srv, port = _serve(_Handler)
+    try:
+        ip = ipaddress.ip_address("127.0.0.1")
+        transport = wm._PinnedAsyncTransport(ip)
+
+        async def go():
+            async with __import__("httpx").AsyncClient(
+                transport=transport, timeout=5, follow_redirects=False,
+            ) as client:
+                # Host "unresolvable.invalid" would never resolve; the pin is
+                # what makes this reach the loopback server on `port`.
+                return await client.post(
+                    f"http://unresolvable.invalid:{port}/hook", content=b"{}",
+                )
+
+        resp = asyncio.run(go())
+        assert resp.status_code == 204
+        assert hits == ["/hook"]
+    finally:
+        srv.shutdown()
+
+
+def test_deliver_pins_to_validated_ip_end_to_end(monkeypatch):
+    """Full _deliver path: a hostname that validation resolves to loopback is
+    pinned to loopback and the local server receives the signed POST."""
+    received = {}
+
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def do_POST(self):  # noqa: N802
+            length = int(self.headers.get("Content-Length", 0))
+            received["body"] = self.rfile.read(length)
+            received["event"] = self.headers.get("X-Odysseus-Event")
+            self.send_response(200)
+            self.end_headers()
+
+        def log_message(self, *a):
+            pass
+
+    srv, port = _serve(_Handler)
+
+    class _Query:
+        def filter(self, *a, **k): return self
+        def update(self, values): return None
+
+    class _Db:
+        def query(self, _m): return _Query()
+        def commit(self): pass
+        def rollback(self): pass
+        def close(self): pass
+
+    # Make both the validation resolve and the pin target loopback, and treat
+    # loopback as allowed for this test (production blocks it — here we only
+    # want to prove the pin routes to the validated IP).
+    monkeypatch.setattr(wm, "SessionLocal", lambda: _Db())
+    monkeypatch.setattr(wm, "_is_private_url", lambda url: False)
+    monkeypatch.setattr(wm, "_resolve_hostname_ips",
+                        lambda h: [ipaddress.ip_address("127.0.0.1")])
+    monkeypatch.setattr(wm, "_ip_is_private", lambda a: False)
+
+    manager = wm.WebhookManager()
+    try:
+        asyncio.run(manager._deliver(
+            "hook-1", f"http://webhook.test:{port}/cb", "s3cret",
+            "webhook.test", {"ok": True},
+        ))
+        assert received.get("event") == "webhook.test"
+        assert b'"ok": true' in received["body"]
+    finally:
+        srv.shutdown()

tests/test_webhook_ssrf_resilience.py

@@ -96,26 +96,29 @@ async def test_webhook_delivery_uses_naive_utc_timestamps(monkeypatch):
     class _Response:
         status_code = 204
 
-    class _Client:
-        def __init__(self):
-            self.content = ""
-
-        async def post(self, _url, content, headers):
-            self.content = content
-            assert headers["X-Odysseus-Event"] == "webhook.test"
-            return _Response()
-
     db = _Db()
-    client = _Client()
     monkeypatch.setattr(wm, "SessionLocal", lambda: db)
 
     manager = wm.WebhookManager()
-    await manager._client.aclose()
-    manager._client = client
+
+    # Replace the pinned-transport send seam so no real socket is opened. The
+    # public-IP literal below still exercises _validated_public_ips (which pins
+    # the connect); the captured content proves the body/headers are built.
+    captured = {}
+
+    async def _fake_send(url, body, headers, ip):
+        captured["content"] = body
+        captured["ip"] = str(ip)
+        assert headers["X-Odysseus-Event"] == "webhook.test"
+        return _Response()
+
+    monkeypatch.setattr(manager, "_send_request", _fake_send)
 
     await manager._deliver("hook-1", "http://93.184.216.34/", None, "webhook.test", {"ok": True})
 
-    body = json.loads(client.content)
+    # The delivery must have pinned to the literal public IP from the URL.
+    assert captured["ip"] == "93.184.216.34"
+    body = json.loads(captured["content"])
     payload_timestamp = datetime.fromisoformat(body["timestamp"])
     assert payload_timestamp.tzinfo is None
     assert db.updates[0]["last_triggered_at"].tzinfo is None