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.

app.py

@@ -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

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

@@ -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/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]]:

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/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_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_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_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") == []