fix(memory): don't let an unreadable store get overwritten with an empty one (#5831)

* fix(memory): don't let an unreadable store get overwritten with an empty one

load_all() answered a failed read the same way it answered an empty store:
with []. Every mutation path is a read-modify-write (load the whole file,
change it, save it back), so a failed read became

    load_all() -> []  ->  [].append(new)  ->  save([new])

and save() is atomic, so the replacement stuck.

The case that actually destroys data is a store that is READABLE but not
parseable - a truncated file, or one holding {} instead of []. Nothing
obstructs the write, so adding a memory returns HTTP 200 and every memory
already stored is gone. Verified end-to-end against a running instance: on the
current code a truncated memory.json plus one add leaves the file holding only
the new entry. Truncation is reachable - core/database.py rewrites memory.json
during migration with a plain open(.., "w") + json.dump, which is not atomic.

A live exclusive lock is not the dangerous case: it blocks the read and the
os.replace alike, so the save fails too and the store survives. That path
currently 500s and loses nothing.

_read_entries() now returns [] only when the file genuinely does not exist and
raises MemoryStoreUnreadable for every other failure, including a store that
parses but is not a JSON array. load_all() keeps the old lenient behaviour so
display, search and context injection still degrade quietly instead of
breaking chat. The read-modify-write callers switch to load_all_for_update(),
which propagates the error: the memory routes turn it into a 503 and change
nothing, backup import refuses rather than saving only the incoming rows, and
auto-extraction and the audit merge skip the write. The audit merge mattered
most - it rebuilds the whole file from one owner's slice plus everyone else's
rows, so an empty read there dropped every other tenant's memories.

The corrupt-JSON path still gets its one shot at the legacy memory.txt
migration before raising, so that recovery is unchanged.

The two updated fakes gained load_all_for_update because the real class has it;
MagicMock would otherwise hand the import path a Mock instead of the seeded list.

Fixes #5673

* fix(memory): fail closed on the remaining read-modify-write add paths

The strict loader landed with the routes, the backup import and the extractor
converted, but three read-modify-write sinks still called load_all(), which
degrades an unreadable store to []. Two of them are the paths users actually
reach, so the data loss in #5673 stayed reproducible:

- src/ai_interaction.py do_manage_memory, action "add" — reached from ordinary
  chat via src/tool_execution.py:793 -> dispatch_ai_tool. "Remember that I
  prefer X" against an unreadable store wrote a one-entry file over it and
  reported success.
- mcp_servers/memory_server.py, action "add" — the same shape through
  _scope_entries(), registered as a built-in in src/builtin_mcp.py.
- src/memory_provider.py NativeMemoryProvider.remember and .delete — wired
  into app state in src/app_initializer.py but not consumed outside tests yet,
  converted here so the pattern is uniform before it goes live.

The MCP server takes _scope_entries(for_update=True) so list keeps the lenient
read. The edit and delete branches on both tool paths were already fail-closed
by accident — an empty view matches nothing and returns before the save — so
they are left alone.

The three new tests drive the real entry points rather than replaying the
shape, and use a truncated store, which is the case that reads back fine so
nothing stops the save. Each asserts memory.json is byte-identical afterwards;
all three fail on the previous commit with the store overwritten.

.env.example

@@ -189,6 +189,7 @@ SEARXNG_INSTANCE=http://localhost:8080
 # ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=26214400       # email compose attachment (25 MB)
 # ODYSSEUS_STT_MAX_AUDIO_BYTES=26214400                  # speech-to-text audio (25 MB)
 # ODYSSEUS_ICS_MAX_BYTES=10485760                        # calendar .ics import (10 MB)
+# ODYSSEUS_TTS_CACHE_MAX_BYTES=524288000                 # TTS cache (500 MB)
 
 # ============================================================
 # Host Docker access (explicit opt-in)

.github/scripts/check-issue-description.js

@@ -153,6 +153,16 @@ module.exports = async ({ github, context, core }) => {
     }
   }
 
+  const LABEL_BAD  = 'needs more info';
+  const LABEL_GOOD = 'ready for review';
+
+  // Closed issues are no longer awaiting review.
+  // This also prevents later edits to closed issues from restoring the label.
+  if (issue.state === 'closed') {
+    await dropLabel(LABEL_GOOD);
+    return;
+  }
+
   // ── Find existing bot comment to update in-place ──────────────────────────
   const MARKER = '<!-- issue-description-check -->';
   const { data: comments } = await github.rest.issues.listComments({
@@ -160,9 +170,6 @@ module.exports = async ({ github, context, core }) => {
   });
   const existing = comments.find(c => c.user.type === 'Bot' && c.body.includes(MARKER));
 
-  const LABEL_BAD  = 'needs more info';
-  const LABEL_GOOD = 'ready for review';
-
   if (failures.length === 0) {
     if (existing) {
       await github.rest.issues.deleteComment({ owner, repo, comment_id: existing.id });

.github/workflows/issue-description-check.yml

@@ -2,7 +2,7 @@ name: ci / issue description check
 
 on:
   issues:
-    types: [opened, edited, reopened]
+    types: [opened, edited, reopened, closed]
 
 permissions:
   issues: write

app.py

@@ -692,7 +692,7 @@ from routes.history.history_routes import setup_history_routes
 app.include_router(setup_history_routes(session_manager, upload_handler=upload_handler))
 
 # Search
-from routes.search_routes import setup_search_routes
+from routes.search.search_routes import setup_search_routes
 app.include_router(setup_search_routes(config))
 
 # Presets
@@ -739,7 +739,7 @@ app.include_router(setup_stt_routes(stt_service))
 logger.info("STT service initialized (provider managed via settings)")
 
 # Documents (artifacts/canvas)
-from routes.document_routes import setup_document_routes
+from routes.document.document_routes import setup_document_routes
 document_router = setup_document_routes(session_manager, upload_handler)
 app.include_router(document_router)
 
@@ -820,7 +820,7 @@ set_ai_rag_manager(rag_manager, personal_docs_mgr)
 logger.info("AI interaction tools initialized (session, memory, RAG, UI control)")
 
 # Webhooks
-from routes.webhook_routes import setup_webhook_routes
+from routes.webhook.webhook_routes import setup_webhook_routes
 app.include_router(setup_webhook_routes(webhook_manager, auth_manager, session_manager, api_key_manager))
 
 # API Tokens
@@ -852,7 +852,7 @@ app.include_router(setup_codex_routes(
 ))
 app.include_router(setup_claude_routes())
 
-from routes.vault_routes import setup_vault_routes
+from routes.vault.vault_routes import setup_vault_routes
 app.include_router(setup_vault_routes())
 
 # Contacts (CardDAV)

docker-compose.gpu-amd.yml

@@ -67,6 +67,7 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
+      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

docker-compose.gpu-nvidia.yml

@@ -66,6 +66,7 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
+      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

docker-compose.yml

@@ -55,6 +55,7 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
+      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

mcp_servers/memory_server.py

@@ -17,6 +17,8 @@ from mcp.types import Tool, TextContent
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
+from src.memory import MemoryStoreUnreadable
+
 server = Server("memory")
 
 # Late-initialized managers (set during first tool call)
@@ -29,6 +31,10 @@ _OWNER_SCOPE_ERROR = (
     "Error: Memory MCP owner is not configured for an owner-scoped memory store. "
     "Set ODYSSEUS_MCP_MEMORY_OWNER for this server or use the owner-aware native memory tool."
 )
+_UNREADABLE_STORE_ERROR = (
+    "Error: Memory store is temporarily unreadable — nothing was saved. "
+    "Repair or restore memory.json, then retry."
+)
 
 
 def _configured_owner() -> str | None:
@@ -51,9 +57,21 @@ def _owner_scoped_store(entries: list[dict]) -> bool:
     return any(_entry_owner(entry) for entry in entries if isinstance(entry, dict))
 
 
-def _scope_entries() -> tuple[str | None, list[dict], list[dict], str | None]:
-    """Return configured owner, all entries, visible entries, and optional error."""
-    entries = _memory_manager.load_all()
+def _scope_entries(for_update: bool = False) -> tuple[str | None, list[dict], list[dict], str | None]:
+    """Return configured owner, all entries, visible entries, and optional error.
+
+    ``for_update=True`` is for read-modify-write callers. They save the ``all
+    entries`` list back, so an unreadable store must be reported as an error
+    instead of degrading to ``[]`` — otherwise the save writes their one new
+    entry over the whole store (issue #5673).
+    """
+    if for_update:
+        try:
+            entries = _memory_manager.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            return None, [], [], f"{_UNREADABLE_STORE_ERROR} ({e})"
+    else:
+        entries = _memory_manager.load_all()
     owner = _configured_owner()
     if owner is None and _owner_scoped_store(entries):
         return None, entries, [], _OWNER_SCOPE_ERROR
@@ -161,7 +179,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
         category = arguments.get("category", "fact")
         if not text:
             return _text_result("Error: Memory text cannot be empty")
-        owner, memories, _visible, scope_error = _scope_entries()
+        owner, memories, _visible, scope_error = _scope_entries(for_update=True)
         if scope_error:
             return _text_result(scope_error)
         entry = _memory_manager.add_entry(text, source="ai_agent", category=category, owner=owner)

requirements.txt

@@ -38,7 +38,10 @@ python-dateutil
 caldav
 cryptography
 bcrypt
-mcp
+# Built-in servers use the v1 low-level Server decorator API. MCP SDK v2 is a
+# breaking rewrite, so keep fresh installs on the maintained v1 line until the
+# servers are migrated together.
+mcp<2
 pyotp
 qrcode[pil]
 croniter

routes/backup_routes.py

@@ -6,6 +6,7 @@ from datetime import datetime
 
 from fastapi import APIRouter, HTTPException, Request, Response
 from core.middleware import require_admin
+from services.memory import MemoryStoreUnreadable
 from src.auth_helpers import get_current_user
 from src.settings import load_settings, save_settings, load_features, save_features
 
@@ -76,7 +77,15 @@ def setup_backup_routes(memory_manager, preset_manager, skills_manager) -> APIRo
 
         # ── Memories ──
         if "memories" in body and isinstance(body["memories"], list):
-            existing = memory_manager.load_all()
+            # Strict load: importing on top of an unreadable store would write
+            # only the incoming rows back and drop everything already saved.
+            try:
+                existing = memory_manager.load_all_for_update()
+            except MemoryStoreUnreadable as e:
+                logger.error("Refusing to import memories: %s", e)
+                raise HTTPException(
+                    503, "Memory store is temporarily unreadable — nothing was imported."
+                )
             # Dedup against THIS user's own memories only. Using every tenant's
             # rows (load_all) meant a memory whose text matched any other
             # user's was silently skipped, so the importing user lost their own

routes/document/__init__.py

@@ -0,0 +1,6 @@
+"""Document route domain package (slice 2m, #4082/#4071).
+
+Contains document_routes.py and document_helpers.py, migrated from the flat
+routes/ directory. Backward-compat shims at routes/document_routes.py and
+routes/document_helpers.py re-export from here.
+"""

routes/document/document_helpers.py

@@ -0,0 +1,243 @@
+"""document_helpers.py — Pydantic models, doc serializers, owner gating, file-locator helpers shared with document_routes.py."""
+
+"""Document routes — CRUD for living documents with version history."""
+
+import logging
+import os
+import re
+from typing import Any, Dict, Optional
+
+from fastapi import HTTPException, Request
+from pydantic import BaseModel
+
+from core.database import Document, DocumentVersion
+from core.database import Session as DbSession
+from src.auth_helpers import _auth_disabled
+from src.upload_handler import UploadHandler
+
+logger = logging.getLogger(__name__)
+
+
+# ---- Request schemas ----
+
+class DocumentCreate(BaseModel):
+    session_id: Optional[str] = None
+    title: str = "Untitled"
+    language: Optional[str] = None
+    content: str = ""
+
+class DocumentUpdate(BaseModel):
+    content: str
+    summary: Optional[str] = None
+    force_version: bool = False
+
+class DocumentPatch(BaseModel):
+    title: Optional[str] = None
+    language: Optional[str] = None
+    session_id: Optional[str] = None  # link/unlink document to a session
+
+
+# ---- Helpers ----
+
+def _doc_to_dict(doc: Document) -> Dict[str, Any]:
+    return {
+        "id": doc.id,
+        "session_id": doc.session_id,
+        "title": doc.title,
+        "language": doc.language,
+        "current_content": doc.current_content,
+        "version_count": doc.version_count,
+        "is_active": doc.is_active,
+        "archived": bool(getattr(doc, "archived", False)),
+        "created_at": (doc.created_at.isoformat() + "Z") if doc.created_at else None,
+        "updated_at": (doc.updated_at.isoformat() + "Z") if doc.updated_at else None,
+        # Source-email provenance (set when doc was created from an email
+        # attachment) — drives the "Send signed reply" menu item.
+        "source_email_uid":        getattr(doc, "source_email_uid", None),
+        "source_email_folder":     getattr(doc, "source_email_folder", None),
+        "source_email_account_id": getattr(doc, "source_email_account_id", None),
+        "source_email_message_id": getattr(doc, "source_email_message_id", None),
+    }
+
+def _version_to_dict(v: DocumentVersion) -> Dict[str, Any]:
+    return {
+        "id": v.id,
+        "document_id": v.document_id,
+        "version_number": v.version_number,
+        "content": v.content,
+        "summary": v.summary,
+        "source": v.source,
+        "created_at": v.created_at.isoformat() if v.created_at else None,
+    }
+
+
+def _verify_doc_owner(db, doc: Document, user: str):
+    """Verify `user` owns this document. Raise 404 if not.
+
+    Documents now carry their own `owner` column, so a doc whose session
+    was deleted (session_id → NULL) can still prove ownership and stay
+    openable / cloneable. We trust that column first and only fall back to
+    the session join for any not-yet-backfilled legacy row.
+    """
+    if user is None:
+        if _auth_disabled():
+            return  # Single-user / no-auth mode: allow access
+        raise HTTPException(403, "Authentication required")
+    if doc.owner is not None:
+        if doc.owner != user:
+            raise HTTPException(404, "Document not found")
+        return
+    # Legacy fallback: derive ownership from the linked session.
+    if not doc.session_id:
+        raise HTTPException(404, "Document not found")
+    session = db.query(DbSession).filter(DbSession.id == doc.session_id).first()
+    if not session or session.owner != user:
+        raise HTTPException(404, "Document not found")
+
+
+def _owner_session_filter(q, user):
+    """Restrict a documents query to those owned by `user`.
+
+    Documents now carry their own `owner` column (backfilled at boot from
+    the linked session, or assigned to the admin user for legacy/orphaned
+    docs). We filter on that directly rather than on a session join, so a
+    document whose session was deleted (session_id → NULL) still shows up
+    for its owner instead of silently vanishing from the Library + search.
+
+    The owner backfill runs in init_db before the app serves requests, so
+    by the time this filter is live there are no NULL-owner rows to leak;
+    we therefore match the owner strictly for authenticated callers."""
+    if not user:
+        if user == "" or _auth_disabled():
+            return q
+        return q.filter(False)
+    return q.filter(Document.owner == user)
+
+
+
+def _slug(name: str) -> str:
+    """Filesystem-friendly version of a document title.
+
+    Whitespace becomes underscores; other unsafe punctuation is dropped.
+    Preserves letters, digits, dot, hyphen, underscore. Idempotent.
+    """
+    import re as _re
+    s = (name or "").strip()
+    # Drop the trailing extension if the title happens to include one
+    s = _re.sub(r'\.pdf$', '', s, flags=_re.IGNORECASE)
+    s = _re.sub(r'\s+', '_', s)
+    s = _re.sub(r'[^A-Za-z0-9._-]', '', s)
+    s = _re.sub(r'_+', '_', s).strip('_')
+    return s or "form"
+
+
+# DPI scale for the interactive PDF view. ~150 DPI (2x of 72 PDF user-units).
+_PDF_RENDER_SCALE = 2.0
+
+
+def _upload_path_inside(upload_dir: str, path: str) -> bool:
+    base = os.path.realpath(upload_dir)
+    p = os.path.realpath(path)
+    try:
+        return os.path.commonpath([base, p]) == base
+    except Exception:
+        return False
+
+
+def _resolve_user_upload_path(
+    upload_handler: Any,
+    upload_id: str,
+    owner: Optional[str],
+    auth_manager=None,
+) -> Optional[str]:
+    """Resolve an upload id to a filesystem path the caller may read."""
+    if upload_handler is None:
+        return None
+    resolved = upload_handler.resolve_upload(
+        upload_id,
+        owner=owner,
+        auth_manager=auth_manager,
+    )
+    if not isinstance(resolved, dict) or not resolved:
+        return None
+    path = resolved.get("path")
+    upload_dir = getattr(upload_handler, "upload_dir", None)
+    if path and upload_dir and not _upload_path_inside(upload_dir, path):
+        logger.warning("Upload path outside upload directory: %s", path)
+        return None
+    return path
+
+
+def _locate_upload(
+    upload_dir: str,
+    file_id: str,
+    owner: Optional[str] = None,
+    auth_manager=None,
+    upload_handler: Any = None,
+):
+    """Find an upload by its filename ID via UploadHandler.resolve_upload."""
+    if upload_handler is None:
+        from src.upload_handler import UploadHandler
+
+        base_dir = os.path.dirname(os.path.abspath(upload_dir))
+        upload_handler = UploadHandler(base_dir, upload_dir)
+    return _resolve_user_upload_path(upload_handler, file_id, owner, auth_manager)
+
+
+def _assert_pdf_marker_upload_owned(
+    request: Request,
+    content: str,
+    user: Optional[str],
+    upload_handler: Any,
+) -> None:
+    """Reject document content whose pdf_source marker points at another user's upload."""
+    if upload_handler is None:
+        return
+    from src.pdf_form_doc import find_source_upload_id
+
+    upload_id = find_source_upload_id(content or "")
+    if not upload_id:
+        return
+    auth_manager = getattr(getattr(request.app, "state", None), "auth_manager", None)
+    if not _resolve_user_upload_path(upload_handler, upload_id, user, auth_manager):
+        raise HTTPException(
+            400,
+            "Document PDF marker references an upload you do not own",
+        )
+
+
+def _derive_title(content: str) -> str:
+    """Derive a title from document content."""
+    import re
+    if not isinstance(content, str):
+        return "Untitled"
+    text = content.strip()
+    if not text:
+        return "Untitled"
+
+    # Markdown header
+    md = re.match(r'^#{1,3}\s+(.+)', text, re.MULTILINE)
+    if md:
+        title = md.group(1).strip()
+        if len(title) > 50:
+            title = title[:48] + "…"
+        return title
+
+    # HTML heading
+    html = re.search(r'<h[1-3][^>]*>([^<]+)</h[1-3]>', text, re.IGNORECASE)
+    if html:
+        title = html.group(1).strip()
+        if len(title) > 50:
+            title = title[:48] + "…"
+        return title
+
+    # First non-empty line (if short enough)
+    for line in text.split('\n'):
+        line = line.strip()
+        if line and 2 <= len(line) <= 60:
+            title = re.sub(r'[:#*`]+$', '', line).strip()
+            if title and len(title) > 50:
+                title = title[:48] + "…"
+            return title or "Untitled"
+
+    return "Untitled"

routes/document_helpers.py

@@ -1,243 +1,14 @@
-"""document_helpers.py — Pydantic models, doc serializers, owner gating, file-locator helpers shared with document_routes.py."""
+"""Backward-compat shim — canonical location is routes/document/document_helpers.py.
 
-"""Document routes — CRUD for living documents with version history."""
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.document_helpers``, ``from routes.document_helpers import
+X``, and the ``sys.modules.pop("routes.document_helpers")`` + re-import
+pattern used by test_security_regressions.py all operate on the *same* object.
+Keeps existing import paths working after slice 2m (#4082/#4071).
+"""
 
-import logging
-import os
-import re
-from typing import Any, Dict, Optional
+import sys as _sys
 
-from fastapi import HTTPException, Request
-from pydantic import BaseModel
+from routes.document import document_helpers as _canonical  # noqa: F401
 
-from core.database import Document, DocumentVersion
-from core.database import Session as DbSession
-from src.auth_helpers import _auth_disabled
-from src.upload_handler import UploadHandler
-
-logger = logging.getLogger(__name__)
-
-
-# ---- Request schemas ----
-
-class DocumentCreate(BaseModel):
-    session_id: Optional[str] = None
-    title: str = "Untitled"
-    language: Optional[str] = None
-    content: str = ""
-
-class DocumentUpdate(BaseModel):
-    content: str
-    summary: Optional[str] = None
-    force_version: bool = False
-
-class DocumentPatch(BaseModel):
-    title: Optional[str] = None
-    language: Optional[str] = None
-    session_id: Optional[str] = None  # link/unlink document to a session
-
-
-# ---- Helpers ----
-
-def _doc_to_dict(doc: Document) -> Dict[str, Any]:
-    return {
-        "id": doc.id,
-        "session_id": doc.session_id,
-        "title": doc.title,
-        "language": doc.language,
-        "current_content": doc.current_content,
-        "version_count": doc.version_count,
-        "is_active": doc.is_active,
-        "archived": bool(getattr(doc, "archived", False)),
-        "created_at": (doc.created_at.isoformat() + "Z") if doc.created_at else None,
-        "updated_at": (doc.updated_at.isoformat() + "Z") if doc.updated_at else None,
-        # Source-email provenance (set when doc was created from an email
-        # attachment) — drives the "Send signed reply" menu item.
-        "source_email_uid":        getattr(doc, "source_email_uid", None),
-        "source_email_folder":     getattr(doc, "source_email_folder", None),
-        "source_email_account_id": getattr(doc, "source_email_account_id", None),
-        "source_email_message_id": getattr(doc, "source_email_message_id", None),
-    }
-
-def _version_to_dict(v: DocumentVersion) -> Dict[str, Any]:
-    return {
-        "id": v.id,
-        "document_id": v.document_id,
-        "version_number": v.version_number,
-        "content": v.content,
-        "summary": v.summary,
-        "source": v.source,
-        "created_at": v.created_at.isoformat() if v.created_at else None,
-    }
-
-
-def _verify_doc_owner(db, doc: Document, user: str):
-    """Verify `user` owns this document. Raise 404 if not.
-
-    Documents now carry their own `owner` column, so a doc whose session
-    was deleted (session_id → NULL) can still prove ownership and stay
-    openable / cloneable. We trust that column first and only fall back to
-    the session join for any not-yet-backfilled legacy row.
-    """
-    if user is None:
-        if _auth_disabled():
-            return  # Single-user / no-auth mode: allow access
-        raise HTTPException(403, "Authentication required")
-    if doc.owner is not None:
-        if doc.owner != user:
-            raise HTTPException(404, "Document not found")
-        return
-    # Legacy fallback: derive ownership from the linked session.
-    if not doc.session_id:
-        raise HTTPException(404, "Document not found")
-    session = db.query(DbSession).filter(DbSession.id == doc.session_id).first()
-    if not session or session.owner != user:
-        raise HTTPException(404, "Document not found")
-
-
-def _owner_session_filter(q, user):
-    """Restrict a documents query to those owned by `user`.
-
-    Documents now carry their own `owner` column (backfilled at boot from
-    the linked session, or assigned to the admin user for legacy/orphaned
-    docs). We filter on that directly rather than on a session join, so a
-    document whose session was deleted (session_id → NULL) still shows up
-    for its owner instead of silently vanishing from the Library + search.
-
-    The owner backfill runs in init_db before the app serves requests, so
-    by the time this filter is live there are no NULL-owner rows to leak;
-    we therefore match the owner strictly for authenticated callers."""
-    if not user:
-        if user == "" or _auth_disabled():
-            return q
-        return q.filter(False)
-    return q.filter(Document.owner == user)
-
-
-
-def _slug(name: str) -> str:
-    """Filesystem-friendly version of a document title.
-
-    Whitespace becomes underscores; other unsafe punctuation is dropped.
-    Preserves letters, digits, dot, hyphen, underscore. Idempotent.
-    """
-    import re as _re
-    s = (name or "").strip()
-    # Drop the trailing extension if the title happens to include one
-    s = _re.sub(r'\.pdf$', '', s, flags=_re.IGNORECASE)
-    s = _re.sub(r'\s+', '_', s)
-    s = _re.sub(r'[^A-Za-z0-9._-]', '', s)
-    s = _re.sub(r'_+', '_', s).strip('_')
-    return s or "form"
-
-
-# DPI scale for the interactive PDF view. ~150 DPI (2x of 72 PDF user-units).
-_PDF_RENDER_SCALE = 2.0
-
-
-def _upload_path_inside(upload_dir: str, path: str) -> bool:
-    base = os.path.realpath(upload_dir)
-    p = os.path.realpath(path)
-    try:
-        return os.path.commonpath([base, p]) == base
-    except Exception:
-        return False
-
-
-def _resolve_user_upload_path(
-    upload_handler: Any,
-    upload_id: str,
-    owner: Optional[str],
-    auth_manager=None,
-) -> Optional[str]:
-    """Resolve an upload id to a filesystem path the caller may read."""
-    if upload_handler is None:
-        return None
-    resolved = upload_handler.resolve_upload(
-        upload_id,
-        owner=owner,
-        auth_manager=auth_manager,
-    )
-    if not isinstance(resolved, dict) or not resolved:
-        return None
-    path = resolved.get("path")
-    upload_dir = getattr(upload_handler, "upload_dir", None)
-    if path and upload_dir and not _upload_path_inside(upload_dir, path):
-        logger.warning("Upload path outside upload directory: %s", path)
-        return None
-    return path
-
-
-def _locate_upload(
-    upload_dir: str,
-    file_id: str,
-    owner: Optional[str] = None,
-    auth_manager=None,
-    upload_handler: Any = None,
-):
-    """Find an upload by its filename ID via UploadHandler.resolve_upload."""
-    if upload_handler is None:
-        from src.upload_handler import UploadHandler
-
-        base_dir = os.path.dirname(os.path.abspath(upload_dir))
-        upload_handler = UploadHandler(base_dir, upload_dir)
-    return _resolve_user_upload_path(upload_handler, file_id, owner, auth_manager)
-
-
-def _assert_pdf_marker_upload_owned(
-    request: Request,
-    content: str,
-    user: Optional[str],
-    upload_handler: Any,
-) -> None:
-    """Reject document content whose pdf_source marker points at another user's upload."""
-    if upload_handler is None:
-        return
-    from src.pdf_form_doc import find_source_upload_id
-
-    upload_id = find_source_upload_id(content or "")
-    if not upload_id:
-        return
-    auth_manager = getattr(getattr(request.app, "state", None), "auth_manager", None)
-    if not _resolve_user_upload_path(upload_handler, upload_id, user, auth_manager):
-        raise HTTPException(
-            400,
-            "Document PDF marker references an upload you do not own",
-        )
-
-
-def _derive_title(content: str) -> str:
-    """Derive a title from document content."""
-    import re
-    if not isinstance(content, str):
-        return "Untitled"
-    text = content.strip()
-    if not text:
-        return "Untitled"
-
-    # Markdown header
-    md = re.match(r'^#{1,3}\s+(.+)', text, re.MULTILINE)
-    if md:
-        title = md.group(1).strip()
-        if len(title) > 50:
-            title = title[:48] + "…"
-        return title
-
-    # HTML heading
-    html = re.search(r'<h[1-3][^>]*>([^<]+)</h[1-3]>', text, re.IGNORECASE)
-    if html:
-        title = html.group(1).strip()
-        if len(title) > 50:
-            title = title[:48] + "…"
-        return title
-
-    # First non-empty line (if short enough)
-    for line in text.split('\n'):
-        line = line.strip()
-        if line and 2 <= len(line) <= 60:
-            title = re.sub(r'[:#*`]+$', '', line).strip()
-            if title and len(title) > 50:
-                title = title[:48] + "…"
-            return title or "Untitled"
-
-    return "Untitled"
+_sys.modules[__name__] = _canonical

routes/memory/memory_routes.py

@@ -21,7 +21,7 @@ def _strip_list_prefix(text: str) -> str:
         return text
     return _LIST_PREFIX_RE.sub("", text, count=1).strip()
 
-from services.memory import MemoryManager
+from services.memory import MemoryManager, MemoryStoreUnreadable
 from core.session_manager import SessionManager
 from src.request_models import MemoryAddRequest
 from core.database import SessionLocal
@@ -35,6 +35,22 @@ from src.upload_limits import read_upload_limited, MEMORY_IMPORT_MAX_BYTES
 logger = logging.getLogger(__name__)
 
 
+def _load_for_update(memory_manager) -> List[Dict[str, Any]]:
+    """Load the whole store for a read-modify-write cycle.
+
+    A transient read failure must not look like an empty store: the caller
+    would append to ``[]`` and save that back, atomically destroying every
+    existing memory (issue #5673). Surface it as a 503 and change nothing.
+    """
+    try:
+        return memory_manager.load_all_for_update()
+    except MemoryStoreUnreadable as e:
+        logger.error("Refusing to rewrite the memory store: %s", e)
+        raise HTTPException(
+            503, "Memory store is temporarily unreadable — no changes were made."
+        )
+
+
 def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionManager, memory_vector=None):
     """Set up memory-related routes."""
     router = APIRouter(prefix="/api/memory", tags=["memory"])
@@ -116,7 +132,7 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
         new_entry = memory_manager.add_entry(text, memory_data.source, memory_data.category, owner=user)
         if memory_data.session_id:
             new_entry["session_id"] = memory_data.session_id
-        all_mem = memory_manager.load_all()
+        all_mem = _load_for_update(memory_manager)
         all_mem.append(new_entry)
         memory_manager.save(all_mem)
         # Sync vector index
@@ -487,7 +503,7 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
     def pin_memory(request: Request, memory_id: str, pinned: bool = Form(True)):
         """Pin or unpin a memory. Pinned memories are always included in context."""
         user = _owner(request)
-        all_mem = memory_manager.load_all()
+        all_mem = _load_for_update(memory_manager)
         for i, memory in enumerate(all_mem):
             if memory["id"] == memory_id:
                 _verify_memory_owner(memory, user)
@@ -512,7 +528,7 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
     def update_memory(request: Request, memory_id: str, text: str = Form(...), category: str = Form(None)):
         """Update an existing memory item with new text and optional category."""
         user = _owner(request)
-        all_mem = memory_manager.load_all()
+        all_mem = _load_for_update(memory_manager)
         for i, memory in enumerate(all_mem):
             if memory["id"] == memory_id:
                 _verify_memory_owner(memory, user)
@@ -534,7 +550,7 @@ def setup_memory_routes(memory_manager: MemoryManager, session_manager: SessionM
     def delete_memory(request: Request, memory_id: str):
         """Delete a memory item by its ID."""
         user = _owner(request)
-        all_mem = memory_manager.load_all()
+        all_mem = _load_for_update(memory_manager)
 
         # Find and verify ownership before deleting
         target = next((m for m in all_mem if m["id"] == memory_id), None)

routes/personal_routes.py

@@ -1,13 +1,11 @@
 # routes/personal_routes.py
 """Routes for personal documents management."""
-import asyncio
 import os
 import logging
 import shutil
 import uuid
 from typing import Any, Dict, List, Tuple
 from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File, Depends
-from fastapi.concurrency import run_in_threadpool
 from src.request_models import DirectoryRequest
 from core.constants import BASE_DIR, PERSONAL_DIR, PERSONAL_UPLOADS_DIR
 from src.rag_singleton import get_rag_manager
@@ -20,6 +18,7 @@ UPLOADS_DIR = PERSONAL_UPLOADS_DIR
 
 logger = logging.getLogger(__name__)
 
+
 def _personal_upload_dir_for_owner(owner: str | None, *, create: bool = True) -> str:
     """Return the per-owner upload directory used for direct RAG uploads."""
     owner_segment = secure_filename((owner or "local").strip())[:80] or "local"
@@ -142,22 +141,6 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
     """
     router = APIRouter(prefix="/api/personal")
 
-    # Serializes directory index jobs across requests. Indexing runs in the
-    # threadpool (#5558), so concurrent requests would otherwise run in parallel
-    # and race PersonalDocsManager's unsynchronized list mutations and file
-    # writes; before the threadpool move they serialized on the blocked event
-    # loop, so one-at-a-time is behavior parity.
-    #
-    # An asyncio.Lock acquired in the async handler BEFORE offloading: a waiting
-    # request parks on the event loop instead of pinning a threadpool worker (an
-    # earlier threading.Lock taken INSIDE the worker meant queued jobs held pool
-    # tokens while blocked, starving every other run_in_threadpool caller).
-    # add/remove/reload all take this lock, so their mutations never interleave.
-    # Per-router (not module-global) so each app binds it to its own event loop.
-    # Scope is the single process: multi-worker deployments would need a shared
-    # lock (out of scope for #5558).
-    _index_job_lock = asyncio.Lock()
-
     def _rag():
         """Get the current RAG manager, retrying init if needed."""
         return get_rag_manager()
@@ -189,12 +172,8 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
         return {"files": files, "directories": directories}
     
     @router.post("/reload")
-    async def api_personal_reload(owner: str = Depends(require_user), _admin: None = Depends(require_admin)):
-        # refresh_index() re-extracts text across every tracked directory —
-        # blocking work. Take the shared job lock (so it cannot race an add /
-        # remove) and run it off the event loop.
-        async with _index_job_lock:
-            await run_in_threadpool(personal_docs_manager.refresh_index)
+    def api_personal_reload(owner: str = Depends(require_user), _admin: None = Depends(require_admin)):
+        personal_docs_manager.refresh_index()
         return {"ok": True, "count": len(personal_docs_manager.index)}
     
     @router.post("/add_directory")
@@ -228,26 +207,12 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
             # Use the RAGManager to index the directory
             rag = _rag()
             if rag:
-                def _index_directory():
-                    result = rag.index_personal_documents(directory, owner=owner)
-                    if result["success"]:
-                        # Also update the personal_docs_manager to track this
-                        # directory. Kept inside the offloaded call: it triggers
-                        # refresh_index(), which re-extracts text across tracked
-                        # directories.
-                        personal_docs_manager.add_directory(directory, index=False)
-                    return result
-
-                # Indexing walks, embeds, and stores the whole tree — minutes
-                # on a real directory. The handler is async, so calling it
-                # inline runs it on the event loop and every other request
-                # queues behind it until it finishes (#5558). Serialize on the
-                # async job lock BEFORE offloading so a queued request parks on
-                # the loop instead of pinning a threadpool worker.
-                async with _index_job_lock:
-                    result = await run_in_threadpool(_index_directory)
+                result = rag.index_personal_documents(directory, owner=owner)
                 
                 if result["success"]:
+                    # Also update the personal_docs_manager to track this directory
+                    personal_docs_manager.add_directory(directory, index=False)
+                    
                     return {
                         "success": True,
                         "message": f"Successfully indexed {result['indexed_count']} chunks from {directory}",
@@ -286,25 +251,17 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
 
             logger.info(f"Removing directory from RAG: {directory}")
 
+            # Always remove from personal_docs_manager tracking
+            if hasattr(personal_docs_manager, 'remove_directory'):
+                personal_docs_manager.remove_directory(directory)
+
+            # Remove from RAG vector store (best-effort)
             rag = _rag()
-
-            def _remove_directory():
-                # Always remove from personal_docs_manager tracking. This
-                # mutates the same unsynchronized list/index an add job touches
-                # and re-extracts text (refresh_index), so it is blocking work.
-                if hasattr(personal_docs_manager, 'remove_directory'):
-                    personal_docs_manager.remove_directory(directory)
-                # Remove from RAG vector store (best-effort).
-                if rag:
-                    try:
-                        rag.remove_directory(directory)
-                    except Exception as e:
-                        logger.warning(f"RAG removal failed for directory {directory}: {e}")
-
-            # Same job lock as add/reload so remove cannot interleave with an
-            # in-flight add; offloaded off the event loop.
-            async with _index_job_lock:
-                await run_in_threadpool(_remove_directory)
+            if rag:
+                try:
+                    rag.remove_directory(directory)
+                except Exception as e:
+                    logger.warning(f"RAG removal failed for directory {directory}: {e}")
 
             return {
                 "success": True,

routes/search/__init__.py

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

routes/search/search_routes.py

@@ -0,0 +1,111 @@
+"""Search routes — /api/search/config GET, /api/search POST."""
+
+import logging
+from typing import Dict, Any
+
+from fastapi import APIRouter, Request
+
+import time
+
+from services.search import get_search_config, comprehensive_web_search, PROVIDER_INFO
+from services.search.core import _call_provider
+from services.search.providers import _get_provider_key, _get_search_instance
+
+logger = logging.getLogger(__name__)
+
+
+async def _request_values(request: Request) -> Dict[str, Any]:
+    """Accept JSON, form data, or query params for search endpoints.
+
+    The browser UI posts FormData, while the agent's generic app_api tool
+    posts JSON. FastAPI Form(...) rejects JSON with a 422 before our handler
+    runs, which made the model think SearXNG was broken.
+    """
+    values: Dict[str, Any] = dict(request.query_params)
+    content_type = (request.headers.get("content-type") or "").lower()
+    try:
+        if "application/json" in content_type:
+            body = await request.json()
+            if isinstance(body, dict):
+                values.update(body)
+        else:
+            form = await request.form()
+            values.update(dict(form))
+    except Exception:
+        pass
+    return values
+
+
+def setup_search_routes(config) -> APIRouter:
+    router = APIRouter(tags=["search"])
+
+    @router.get("/api/search/config")
+    async def get_search_settings() -> Dict[str, Any]:
+        return get_search_config()
+
+    @router.post("/api/search")
+    async def do_web_search(request: Request) -> Dict[str, Any]:
+        """Standalone web search — returns context string + source list.
+
+        Used by Compare mode to pre-search once and share results across panes.
+        """
+        values = await _request_values(request)
+        query = str(values.get("query") or values.get("q") or "").strip()
+        if not query:
+            return {"context": "", "sources": [], "error": "query is required"}
+        time_filter = values.get("time_filter") or values.get("freshness")
+        if time_filter is not None:
+            time_filter = str(time_filter).strip() or None
+        try:
+            context, sources = comprehensive_web_search(
+                query, return_sources=True, time_filter=time_filter,
+            )
+            return {"context": context, "sources": sources}
+        except Exception as e:
+            logger.error(f"Standalone web search failed: {e}")
+            return {"context": "", "sources": [], "error": str(e)}
+
+    @router.get("/api/search/providers")
+    async def list_search_providers():
+        """Return available search providers with config status."""
+        providers = []
+        for pid, (label, needs_key, needs_url) in PROVIDER_INFO.items():
+            if pid == "disabled":
+                continue
+            available = True
+            if needs_key and not _get_provider_key(pid):
+                available = False
+            if needs_url and pid == "searxng" and not _get_search_instance():
+                available = False
+            providers.append({
+                "id": pid,
+                "label": label,
+                "available": available,
+            })
+        return providers
+
+    @router.post("/api/search/query")
+    async def search_with_provider(request: Request) -> Dict[str, Any]:
+        """Search using a specific provider. Used by compare search mode."""
+        values = await _request_values(request)
+        query = str(values.get("query") or values.get("q") or "").strip()
+        provider = str(values.get("provider") or "").strip()
+        try:
+            count = int(values.get("count") or values.get("limit") or 10)
+        except Exception:
+            count = 10
+        if not query:
+            return {"results": [], "provider": provider, "error": "query is required"}
+        if provider not in PROVIDER_INFO or provider == "disabled":
+            return {"results": [], "provider": provider, "error": "Unknown provider"}
+        t0 = time.time()
+        try:
+            results = _call_provider(provider, query, min(count, 20))
+            elapsed = round(time.time() - t0, 2)
+            return {"results": results, "provider": provider, "time": elapsed}
+        except Exception as e:
+            elapsed = round(time.time() - t0, 2)
+            logger.error(f"Search provider {provider} failed: {e}")
+            return {"results": [], "provider": provider, "time": elapsed, "error": str(e)}
+
+    return router

routes/search_routes.py

@@ -1,111 +1,13 @@
-"""Search routes — /api/search/config GET, /api/search POST."""
+"""Backward-compat shim — canonical location is routes/search/search_routes.py.
 
-import logging
-from typing import Dict, Any
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.search_routes`` and ``from routes.search_routes import X``
+keep resolving to the canonical module. Keeps existing import paths working
+after slice 2j (#4082/#4071).
+"""
 
-from fastapi import APIRouter, Request
+import sys as _sys
 
-import time
+from routes.search import search_routes as _canonical  # noqa: F401
 
-from services.search import get_search_config, comprehensive_web_search, PROVIDER_INFO
-from services.search.core import _call_provider
-from services.search.providers import _get_provider_key, _get_search_instance
-
-logger = logging.getLogger(__name__)
-
-
-async def _request_values(request: Request) -> Dict[str, Any]:
-    """Accept JSON, form data, or query params for search endpoints.
-
-    The browser UI posts FormData, while the agent's generic app_api tool
-    posts JSON. FastAPI Form(...) rejects JSON with a 422 before our handler
-    runs, which made the model think SearXNG was broken.
-    """
-    values: Dict[str, Any] = dict(request.query_params)
-    content_type = (request.headers.get("content-type") or "").lower()
-    try:
-        if "application/json" in content_type:
-            body = await request.json()
-            if isinstance(body, dict):
-                values.update(body)
-        else:
-            form = await request.form()
-            values.update(dict(form))
-    except Exception:
-        pass
-    return values
-
-
-def setup_search_routes(config) -> APIRouter:
-    router = APIRouter(tags=["search"])
-
-    @router.get("/api/search/config")
-    async def get_search_settings() -> Dict[str, Any]:
-        return get_search_config()
-
-    @router.post("/api/search")
-    async def do_web_search(request: Request) -> Dict[str, Any]:
-        """Standalone web search — returns context string + source list.
-
-        Used by Compare mode to pre-search once and share results across panes.
-        """
-        values = await _request_values(request)
-        query = str(values.get("query") or values.get("q") or "").strip()
-        if not query:
-            return {"context": "", "sources": [], "error": "query is required"}
-        time_filter = values.get("time_filter") or values.get("freshness")
-        if time_filter is not None:
-            time_filter = str(time_filter).strip() or None
-        try:
-            context, sources = comprehensive_web_search(
-                query, return_sources=True, time_filter=time_filter,
-            )
-            return {"context": context, "sources": sources}
-        except Exception as e:
-            logger.error(f"Standalone web search failed: {e}")
-            return {"context": "", "sources": [], "error": str(e)}
-
-    @router.get("/api/search/providers")
-    async def list_search_providers():
-        """Return available search providers with config status."""
-        providers = []
-        for pid, (label, needs_key, needs_url) in PROVIDER_INFO.items():
-            if pid == "disabled":
-                continue
-            available = True
-            if needs_key and not _get_provider_key(pid):
-                available = False
-            if needs_url and pid == "searxng" and not _get_search_instance():
-                available = False
-            providers.append({
-                "id": pid,
-                "label": label,
-                "available": available,
-            })
-        return providers
-
-    @router.post("/api/search/query")
-    async def search_with_provider(request: Request) -> Dict[str, Any]:
-        """Search using a specific provider. Used by compare search mode."""
-        values = await _request_values(request)
-        query = str(values.get("query") or values.get("q") or "").strip()
-        provider = str(values.get("provider") or "").strip()
-        try:
-            count = int(values.get("count") or values.get("limit") or 10)
-        except Exception:
-            count = 10
-        if not query:
-            return {"results": [], "provider": provider, "error": "query is required"}
-        if provider not in PROVIDER_INFO or provider == "disabled":
-            return {"results": [], "provider": provider, "error": "Unknown provider"}
-        t0 = time.time()
-        try:
-            results = _call_provider(provider, query, min(count, 20))
-            elapsed = round(time.time() - t0, 2)
-            return {"results": results, "provider": provider, "time": elapsed}
-        except Exception as e:
-            elapsed = round(time.time() - t0, 2)
-            logger.error(f"Search provider {provider} failed: {e}")
-            return {"results": [], "provider": provider, "time": elapsed, "error": str(e)}
-
-    return router
+_sys.modules[__name__] = _canonical

routes/skills_routes.py

@@ -1409,7 +1409,7 @@ def setup_skills_routes(skills_manager: SkillsManager) -> APIRouter:
 
         # Prefer the configured DEFAULT (→ Utility) model — not the current chat
         # session's model. Fall back to the caller's session model only if unset.
-        url, model, headers = resolve_endpoint("default", owner=user)
+        url, model, headers = resolve_endpoint("utility", owner=user)
         if not url or not model:
             url = url or ((body.get("endpoint_url") or "").strip() or None)
             model = model or ((body.get("model") or "").strip() or None)

routes/vault/__init__.py

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

routes/vault/vault_routes.py

@@ -0,0 +1,242 @@
+"""
+vault_routes.py
+
+Vaultwarden / Bitwarden CLI integration — config and unlock endpoints.
+Stores the BW_SESSION key in data/vault.json with restrictive permissions.
+"""
+
+import json
+import logging
+import os
+import shutil
+import asyncio
+from pathlib import Path
+from datetime import datetime
+from fastapi import APIRouter, Request
+from pydantic import BaseModel
+
+from core.middleware import require_admin
+from core.platform_compat import IS_WINDOWS, safe_chmod, which_tool
+from src.constants import VAULT_FILE as _VAULT_FILE
+
+logger = logging.getLogger(__name__)
+
+VAULT_FILE = Path(_VAULT_FILE)
+
+
+def _find_bw() -> str:
+    """Locate the bw binary, checking PATH and common npm-global locations.
+
+    On Windows the Bitwarden CLI shim is `bw.cmd`/`bw.exe`, resolved by
+    which_tool via PATHEXT.
+    """
+    p = which_tool("bw")
+    if p:
+        return p
+    if IS_WINDOWS:
+        appdata = os.environ.get("APPDATA", os.path.expanduser("~"))
+        for candidate in (
+            os.path.join(appdata, "npm", "bw.cmd"),
+            os.path.join(appdata, "npm", "bw.exe"),
+        ):
+            if os.path.isfile(candidate):
+                return candidate
+        return "bw"
+    home = os.path.expanduser("~")
+    for candidate in (
+        f"{home}/.npm-global/bin/bw",
+        f"{home}/.nvm/versions/node/*/bin/bw",
+        "/usr/local/bin/bw",
+        "/opt/homebrew/bin/bw",
+    ):
+        if "*" in candidate:
+            import glob
+            for m in glob.glob(candidate):
+                if os.path.isfile(m) and os.access(m, os.X_OK):
+                    return m
+        elif os.path.isfile(candidate) and os.access(candidate, os.X_OK):
+            return candidate
+    return "bw"  # fall back to PATH lookup (will FileNotFoundError, handled below)
+
+
+def _load_config() -> dict:
+    if VAULT_FILE.exists():
+        try:
+            data = json.loads(VAULT_FILE.read_text(encoding="utf-8"))
+            return data if isinstance(data, dict) else {}
+        except Exception:
+            pass
+    return {}
+
+
+def _save_config(cfg: dict):
+    VAULT_FILE.parent.mkdir(parents=True, exist_ok=True)
+    VAULT_FILE.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
+    # POSIX: restrict the BW_SESSION store to 0o600. Windows: no-op (profile dir
+    # is ACL-restricted already).
+    safe_chmod(str(VAULT_FILE), 0o600)
+
+
+async def _run_bw(args: list, session: str = None, input_text: str = None,
+                  bw_password: str = None) -> tuple:
+    env = {}
+    env.update(os.environ)
+    if session:
+        env["BW_SESSION"] = session
+    # Secrets must never be passed as argv — process arguments are world-readable
+    # via `ps` / `/proc/<pid>/cmdline` to any local user. Keep --passwordenv
+    # support for bw commands that need it; unlock/login callers should prefer
+    # stdin so the master password is not left in the child environment either.
+    if bw_password is not None:
+        env["BW_PASSWORD"] = bw_password
+    bw_path = _find_bw()
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            bw_path, *args,
+            stdin=asyncio.subprocess.PIPE if input_text else None,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            env=env,
+        )
+    except FileNotFoundError:
+        return "", "bw CLI not installed (install `nodejs-bitwarden-cli` or `bitwarden-cli`)", 127
+    except Exception as e:
+        return "", f"Failed to launch bw: {e}", 1
+    try:
+        stdout, stderr = await proc.communicate(input=input_text.encode() if input_text else None)
+    except Exception as e:
+        return "", f"bw subprocess error: {e}", 1
+    return stdout.decode(errors="replace").strip(), stderr.decode(errors="replace").strip(), proc.returncode
+
+
+class VaultConfig(BaseModel):
+    server_url: str = ""
+    email: str = ""
+
+
+class VaultUnlockRequest(BaseModel):
+    master_password: str
+
+
+class VaultLoginRequest(BaseModel):
+    email: str
+    master_password: str
+
+
+def setup_vault_routes():
+    router = APIRouter(prefix="/api/vault", tags=["vault"])
+
+    @router.get("/config")
+    async def get_config(request: Request):
+        """Return vault config (no sensitive fields)."""
+        require_admin(request)
+        cfg = _load_config()
+        return {
+            "server_url": cfg.get("server_url", ""),
+            "email": cfg.get("email", ""),
+            "unlocked": bool(cfg.get("session")),
+            "unlocked_at": cfg.get("unlocked_at", ""),
+            "bw_installed": await _check_bw_installed(),
+        }
+
+    @router.post("/config")
+    async def save_config(req: VaultConfig, request: Request):
+        """Save vault URL + email. Runs 'bw config server' to point at Vaultwarden."""
+        require_admin(request)
+        cfg = _load_config()
+        cfg["server_url"] = req.server_url.strip().rstrip("/")
+        cfg["email"] = req.email.strip()
+
+        if cfg["server_url"]:
+            _, stderr, rc = await _run_bw(["config", "server", cfg["server_url"]])
+            if rc != 0:
+                return {"ok": False, "error": f"bw config failed: {stderr[:300]}"}
+
+        _save_config(cfg)
+        return {"ok": True}
+
+    @router.post("/login")
+    async def login(req: VaultLoginRequest, request: Request):
+        """Log in to Vaultwarden (required once per account)."""
+        require_admin(request)
+        cfg = _load_config()
+        # Update email
+        cfg["email"] = req.email
+        _save_config(cfg)
+
+        stdout, stderr, rc = await _run_bw(
+            ["login", req.email, "--raw"],
+            input_text=req.master_password + "\n",
+        )
+        if rc != 0:
+            # Already logged in is OK
+            if "already logged in" in stderr.lower():
+                return {"ok": True, "already": True}
+            return {"ok": False, "error": f"Login failed: {stderr[:300]}"}
+        # bw login --raw prints session key on success (when 2FA disabled)
+        if stdout:
+            cfg["session"] = stdout
+            cfg["unlocked_at"] = datetime.utcnow().isoformat()
+            _save_config(cfg)
+        return {"ok": True}
+
+    @router.post("/unlock")
+    async def unlock(req: VaultUnlockRequest, request: Request):
+        """Unlock the vault and save the session key."""
+        require_admin(request)
+        # Pass the master password on stdin, not argv. argv is visible through
+        # `ps` / /proc/<pid>/cmdline; stdin also avoids leaving the secret in
+        # the child process environment.
+        stdout, stderr, rc = await _run_bw(
+            ["unlock", "--raw"],
+            input_text=req.master_password + "\n",
+        )
+        if rc != 0:
+            return {"ok": False, "error": f"Unlock failed: {stderr[:300]}"}
+        session = stdout.strip()
+        if not session:
+            return {"ok": False, "error": "bw returned empty session"}
+        cfg = _load_config()
+        cfg["session"] = session
+        cfg["unlocked_at"] = datetime.utcnow().isoformat()
+        _save_config(cfg)
+        return {"ok": True, "message": "Vault unlocked"}
+
+    @router.post("/lock")
+    async def lock(request: Request):
+        """Lock the vault (clear session from config)."""
+        require_admin(request)
+        cfg = _load_config()
+        cfg.pop("session", None)
+        cfg.pop("unlocked_at", None)
+        _save_config(cfg)
+        # Also tell bw to lock
+        await _run_bw(["lock"])
+        return {"ok": True, "message": "Vault locked"}
+
+    @router.post("/logout")
+    async def logout(request: Request):
+        """Log out of the Bitwarden CLI completely."""
+        require_admin(request)
+        await _run_bw(["logout"])
+        cfg = _load_config()
+        cfg.pop("session", None)
+        cfg.pop("email", None)
+        cfg.pop("unlocked_at", None)
+        _save_config(cfg)
+        return {"ok": True}
+
+    return router
+
+
+async def _check_bw_installed() -> bool:
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            _find_bw(), "--version",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        await proc.communicate()
+        return proc.returncode == 0
+    except Exception:
+        return False

routes/vault_routes.py

@@ -1,242 +1,14 @@
-"""
-vault_routes.py
+"""Backward-compat shim — canonical location is routes/vault/vault_routes.py.
 
-Vaultwarden / Bitwarden CLI integration — config and unlock endpoints.
-Stores the BW_SESSION key in data/vault.json with restrictive permissions.
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.vault_routes``, ``from routes.vault_routes import X``,
+and the ``import ... as vr`` + ``monkeypatch.setattr(vr, ...)`` pattern used
+by test_vault_password_not_in_argv.py all operate on the *same* object.
+Keeps existing import paths working after slice 2k (#4082/#4071).
 """
 
-import json
-import logging
-import os
-import shutil
-import asyncio
-from pathlib import Path
-from datetime import datetime
-from fastapi import APIRouter, Request
-from pydantic import BaseModel
+import sys as _sys
 
-from core.middleware import require_admin
-from core.platform_compat import IS_WINDOWS, safe_chmod, which_tool
-from src.constants import VAULT_FILE as _VAULT_FILE
+from routes.vault import vault_routes as _canonical  # noqa: F401
 
-logger = logging.getLogger(__name__)
-
-VAULT_FILE = Path(_VAULT_FILE)
-
-
-def _find_bw() -> str:
-    """Locate the bw binary, checking PATH and common npm-global locations.
-
-    On Windows the Bitwarden CLI shim is `bw.cmd`/`bw.exe`, resolved by
-    which_tool via PATHEXT.
-    """
-    p = which_tool("bw")
-    if p:
-        return p
-    if IS_WINDOWS:
-        appdata = os.environ.get("APPDATA", os.path.expanduser("~"))
-        for candidate in (
-            os.path.join(appdata, "npm", "bw.cmd"),
-            os.path.join(appdata, "npm", "bw.exe"),
-        ):
-            if os.path.isfile(candidate):
-                return candidate
-        return "bw"
-    home = os.path.expanduser("~")
-    for candidate in (
-        f"{home}/.npm-global/bin/bw",
-        f"{home}/.nvm/versions/node/*/bin/bw",
-        "/usr/local/bin/bw",
-        "/opt/homebrew/bin/bw",
-    ):
-        if "*" in candidate:
-            import glob
-            for m in glob.glob(candidate):
-                if os.path.isfile(m) and os.access(m, os.X_OK):
-                    return m
-        elif os.path.isfile(candidate) and os.access(candidate, os.X_OK):
-            return candidate
-    return "bw"  # fall back to PATH lookup (will FileNotFoundError, handled below)
-
-
-def _load_config() -> dict:
-    if VAULT_FILE.exists():
-        try:
-            data = json.loads(VAULT_FILE.read_text(encoding="utf-8"))
-            return data if isinstance(data, dict) else {}
-        except Exception:
-            pass
-    return {}
-
-
-def _save_config(cfg: dict):
-    VAULT_FILE.parent.mkdir(parents=True, exist_ok=True)
-    VAULT_FILE.write_text(json.dumps(cfg, indent=2), encoding="utf-8")
-    # POSIX: restrict the BW_SESSION store to 0o600. Windows: no-op (profile dir
-    # is ACL-restricted already).
-    safe_chmod(str(VAULT_FILE), 0o600)
-
-
-async def _run_bw(args: list, session: str = None, input_text: str = None,
-                  bw_password: str = None) -> tuple:
-    env = {}
-    env.update(os.environ)
-    if session:
-        env["BW_SESSION"] = session
-    # Secrets must never be passed as argv — process arguments are world-readable
-    # via `ps` / `/proc/<pid>/cmdline` to any local user. Keep --passwordenv
-    # support for bw commands that need it; unlock/login callers should prefer
-    # stdin so the master password is not left in the child environment either.
-    if bw_password is not None:
-        env["BW_PASSWORD"] = bw_password
-    bw_path = _find_bw()
-    try:
-        proc = await asyncio.create_subprocess_exec(
-            bw_path, *args,
-            stdin=asyncio.subprocess.PIPE if input_text else None,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-            env=env,
-        )
-    except FileNotFoundError:
-        return "", "bw CLI not installed (install `nodejs-bitwarden-cli` or `bitwarden-cli`)", 127
-    except Exception as e:
-        return "", f"Failed to launch bw: {e}", 1
-    try:
-        stdout, stderr = await proc.communicate(input=input_text.encode() if input_text else None)
-    except Exception as e:
-        return "", f"bw subprocess error: {e}", 1
-    return stdout.decode(errors="replace").strip(), stderr.decode(errors="replace").strip(), proc.returncode
-
-
-class VaultConfig(BaseModel):
-    server_url: str = ""
-    email: str = ""
-
-
-class VaultUnlockRequest(BaseModel):
-    master_password: str
-
-
-class VaultLoginRequest(BaseModel):
-    email: str
-    master_password: str
-
-
-def setup_vault_routes():
-    router = APIRouter(prefix="/api/vault", tags=["vault"])
-
-    @router.get("/config")
-    async def get_config(request: Request):
-        """Return vault config (no sensitive fields)."""
-        require_admin(request)
-        cfg = _load_config()
-        return {
-            "server_url": cfg.get("server_url", ""),
-            "email": cfg.get("email", ""),
-            "unlocked": bool(cfg.get("session")),
-            "unlocked_at": cfg.get("unlocked_at", ""),
-            "bw_installed": await _check_bw_installed(),
-        }
-
-    @router.post("/config")
-    async def save_config(req: VaultConfig, request: Request):
-        """Save vault URL + email. Runs 'bw config server' to point at Vaultwarden."""
-        require_admin(request)
-        cfg = _load_config()
-        cfg["server_url"] = req.server_url.strip().rstrip("/")
-        cfg["email"] = req.email.strip()
-
-        if cfg["server_url"]:
-            _, stderr, rc = await _run_bw(["config", "server", cfg["server_url"]])
-            if rc != 0:
-                return {"ok": False, "error": f"bw config failed: {stderr[:300]}"}
-
-        _save_config(cfg)
-        return {"ok": True}
-
-    @router.post("/login")
-    async def login(req: VaultLoginRequest, request: Request):
-        """Log in to Vaultwarden (required once per account)."""
-        require_admin(request)
-        cfg = _load_config()
-        # Update email
-        cfg["email"] = req.email
-        _save_config(cfg)
-
-        stdout, stderr, rc = await _run_bw(
-            ["login", req.email, "--raw"],
-            input_text=req.master_password + "\n",
-        )
-        if rc != 0:
-            # Already logged in is OK
-            if "already logged in" in stderr.lower():
-                return {"ok": True, "already": True}
-            return {"ok": False, "error": f"Login failed: {stderr[:300]}"}
-        # bw login --raw prints session key on success (when 2FA disabled)
-        if stdout:
-            cfg["session"] = stdout
-            cfg["unlocked_at"] = datetime.utcnow().isoformat()
-            _save_config(cfg)
-        return {"ok": True}
-
-    @router.post("/unlock")
-    async def unlock(req: VaultUnlockRequest, request: Request):
-        """Unlock the vault and save the session key."""
-        require_admin(request)
-        # Pass the master password on stdin, not argv. argv is visible through
-        # `ps` / /proc/<pid>/cmdline; stdin also avoids leaving the secret in
-        # the child process environment.
-        stdout, stderr, rc = await _run_bw(
-            ["unlock", "--raw"],
-            input_text=req.master_password + "\n",
-        )
-        if rc != 0:
-            return {"ok": False, "error": f"Unlock failed: {stderr[:300]}"}
-        session = stdout.strip()
-        if not session:
-            return {"ok": False, "error": "bw returned empty session"}
-        cfg = _load_config()
-        cfg["session"] = session
-        cfg["unlocked_at"] = datetime.utcnow().isoformat()
-        _save_config(cfg)
-        return {"ok": True, "message": "Vault unlocked"}
-
-    @router.post("/lock")
-    async def lock(request: Request):
-        """Lock the vault (clear session from config)."""
-        require_admin(request)
-        cfg = _load_config()
-        cfg.pop("session", None)
-        cfg.pop("unlocked_at", None)
-        _save_config(cfg)
-        # Also tell bw to lock
-        await _run_bw(["lock"])
-        return {"ok": True, "message": "Vault locked"}
-
-    @router.post("/logout")
-    async def logout(request: Request):
-        """Log out of the Bitwarden CLI completely."""
-        require_admin(request)
-        await _run_bw(["logout"])
-        cfg = _load_config()
-        cfg.pop("session", None)
-        cfg.pop("email", None)
-        cfg.pop("unlocked_at", None)
-        _save_config(cfg)
-        return {"ok": True}
-
-    return router
-
-
-async def _check_bw_installed() -> bool:
-    try:
-        proc = await asyncio.create_subprocess_exec(
-            _find_bw(), "--version",
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-        await proc.communicate()
-        return proc.returncode == 0
-    except Exception:
-        return False
+_sys.modules[__name__] = _canonical

routes/webhook/__init__.py

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

routes/webhook/webhook_routes.py

@@ -0,0 +1,395 @@
+"""Webhook, API Token, and sync chat routes."""
+
+import uuid
+import logging
+from typing import Optional
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request, Form
+from pydantic import BaseModel, Field
+
+from core.database import SessionLocal, Webhook, ModelEndpoint
+from src.auth_helpers import owner_filter
+from src.url_security import validate_public_http_url
+from src.webhook_manager import WebhookManager, validate_webhook_url, validate_events
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api", tags=["webhooks"])
+
+# Input limits
+MAX_NAME_LEN = 100
+MAX_URL_LEN = 2048
+MAX_SECRET_LEN = 256
+MAX_MESSAGE_LEN = 32_000
+
+
+from core.middleware import require_admin as _require_admin
+
+
+def _select_api_chat_fallback_endpoint(db, token_owner: Optional[str]):
+    """First enabled ModelEndpoint visible to token_owner — their own rows plus
+    legacy null-owner ("shared") rows. Owner-scoped: an unscoped .first() would
+    let a chat-scoped token fall back onto another user's private endpoint and
+    silently spend that owner's API key/quota. Prefer owner rows before shared
+    rows. Fails closed to null-owner rows only when token_owner is absent.
+    Does not validate base_url — admin-configured local/LAN endpoints remain allowed.
+    """
+    query = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True)  # noqa: E712
+    if token_owner:
+        query = owner_filter(query, ModelEndpoint, token_owner)
+        return query.order_by(ModelEndpoint.owner.desc(), ModelEndpoint.created_at).first()
+    return query.filter(ModelEndpoint.owner == None).order_by(ModelEndpoint.created_at).first()  # noqa: E711
+
+
+def _caller_owns_session(sess_owner, caller) -> bool:
+    """Strict session-ownership gate for the token-authenticated sync-chat
+    endpoint (`POST /api/v1/chat`).
+
+    Mirrors ``_verify_session_owner`` in session_routes.py and the null-owner
+    gates in notes/calendar/gallery: a caller may resume a session ONLY when
+    its owner matches them exactly. A null/empty session owner (legacy or
+    migrated rows) is deliberately NOT resumable by an arbitrary token — the
+    old ``sess_owner and sess_owner != caller`` form skipped the check whenever
+    ``sess_owner`` was falsy, so any chat-scoped token (e.g. a paired mobile
+    device) could resume such a session, inject a message, and read back its
+    history and reuse the owner's endpoint credentials. Fail closed: an
+    unresolvable caller also returns False.
+    """
+    if not caller:
+        return False
+    return sess_owner == caller
+
+
+def setup_webhook_routes(
+    webhook_manager: WebhookManager,
+    auth_manager,
+    session_manager=None,
+    api_key_manager=None,
+) -> APIRouter:
+
+    @router.get("/webhooks")
+    def list_webhooks(request: Request):
+        _require_admin(request)
+        db = SessionLocal()
+        try:
+            hooks = db.query(Webhook).all()
+            return [
+                {
+                    "id": w.id,
+                    "name": w.name,
+                    "url": w.url,
+                    "has_secret": bool(w.secret),
+                    "events": w.events.split(",") if w.events else [],
+                    "is_active": w.is_active,
+                    "last_triggered_at": w.last_triggered_at.isoformat() if w.last_triggered_at else None,
+                    "last_status_code": w.last_status_code,
+                    "last_error": w.last_error,
+                    "created_at": w.created_at.isoformat() if w.created_at else None,
+                }
+                for w in hooks
+            ]
+        finally:
+            db.close()
+
+    @router.post("/webhooks")
+    def create_webhook(
+        request: Request,
+        name: str = Form(""),
+        url: str = Form(""),
+        secret: str = Form(""),
+        events: str = Form(""),
+    ):
+        _require_admin(request)
+        name = name.strip()[:MAX_NAME_LEN]
+        if not name:
+            raise HTTPException(400, "Webhook name is required")
+        try:
+            url = validate_webhook_url(url)
+        except ValueError as e:
+            raise HTTPException(400, str(e))
+        try:
+            events = validate_events(events)
+        except ValueError as e:
+            raise HTTPException(400, str(e))
+
+        secret_val = secret.strip()[:MAX_SECRET_LEN] or None
+        # Encrypt the secret at rest using the same Fernet key as API keys
+        encrypted_secret = None
+        if secret_val and api_key_manager:
+            encrypted_secret = api_key_manager.encrypt_api_key(secret_val)
+        elif secret_val:
+            encrypted_secret = secret_val  # Fallback if no encryption available
+
+        webhook_id = str(uuid.uuid4())[:8]
+        db = SessionLocal()
+        try:
+            db.add(Webhook(
+                id=webhook_id,
+                name=name,
+                url=url,
+                secret=encrypted_secret,
+                events=events,
+                is_active=True,
+            ))
+            db.commit()
+        finally:
+            db.close()
+
+        return {"id": webhook_id, "name": name}
+
+    @router.post("/webhooks/{webhook_id}/test")
+    async def test_webhook(request: Request, webhook_id: str):
+        _require_admin(request)
+        db = SessionLocal()
+        try:
+            wh = db.query(Webhook).filter(Webhook.id == webhook_id).first()
+            if not wh:
+                raise HTTPException(404, "Webhook not found")
+            url, secret = wh.url, wh.secret
+        finally:
+            db.close()
+
+        await webhook_manager.deliver_test(webhook_id, url, secret)
+        return {"status": "sent"}
+
+    @router.patch("/webhooks/{webhook_id}")
+    def toggle_webhook(request: Request, webhook_id: str):
+        _require_admin(request)
+        db = SessionLocal()
+        try:
+            wh = db.query(Webhook).filter(Webhook.id == webhook_id).first()
+            if not wh:
+                raise HTTPException(404, "Webhook not found")
+            wh.is_active = not wh.is_active
+            db.commit()
+            return {"id": webhook_id, "is_active": wh.is_active}
+        finally:
+            db.close()
+
+    @router.delete("/webhooks/{webhook_id}")
+    def delete_webhook(request: Request, webhook_id: str):
+        _require_admin(request)
+        db = SessionLocal()
+        try:
+            deleted = db.query(Webhook).filter(Webhook.id == webhook_id).delete()
+            db.commit()
+            if not deleted:
+                raise HTTPException(404, "Webhook not found")
+        finally:
+            db.close()
+        return {"status": "deleted"}
+
+    # ================================================================
+    # Sync Chat Endpoint (for n8n / Make / Activepieces)
+    # ================================================================
+
+    # Known provider base URLs — auto-resolved from api_key prefix or model name
+    KNOWN_PROVIDERS = {
+        "deepseek": "https://api.deepseek.com/v1",
+        "openai": "https://api.openai.com/v1",
+        "mistral": "https://api.mistral.ai/v1",
+        "groq": "https://api.groq.com/openai/v1",
+        "together": "https://api.together.xyz/v1",
+        "openrouter": "https://openrouter.ai/api/v1",
+        "ollama": "https://ollama.com/api",
+        "opencode-zen": "https://opencode.ai/zen/v1",
+        "opencode-go": "https://opencode.ai/zen/go/v1",
+        "fireworks": "https://api.fireworks.ai/inference/v1",
+        "venice": "https://api.venice.ai/api/v1",
+        "kimi-code": "https://api.kimi.com/coding/v1",
+        "kimicode": "https://api.kimi.com/coding/v1",
+    }
+
+    # Model prefix → provider mapping for auto-detection
+    MODEL_PROVIDER_MAP = {
+        "deepseek": "deepseek",
+        "gpt-": "openai",
+        "o1": "openai",
+        "o3": "openai",
+        "o4": "openai",
+        "mistral": "mistral",
+        "llama": "groq",
+        "mixtral": "groq",
+        "kimi-for-coding": "kimi-code",
+        "kimi": "kimi-code",
+    }
+
+    def _resolve_base_url(model: Optional[str], provider: Optional[str]) -> Optional[str]:
+        """Try to auto-resolve a base URL from provider name or model prefix."""
+        if provider and provider.lower() in KNOWN_PROVIDERS:
+            return KNOWN_PROVIDERS[provider.lower()]
+        if model:
+            model_lower = model.lower()
+            for prefix, prov in MODEL_PROVIDER_MAP.items():
+                if model_lower.startswith(prefix):
+                    return KNOWN_PROVIDERS[prov]
+        return None
+
+    class SyncChatRequest(BaseModel):
+        message: str = Field(..., max_length=MAX_MESSAGE_LEN)
+        model: Optional[str] = Field(None, max_length=200)
+        session: Optional[str] = Field(None, max_length=100)
+        api_key: Optional[str] = Field(None, max_length=256)
+        base_url: Optional[str] = Field(None, max_length=MAX_URL_LEN)
+        provider: Optional[str] = Field(None, max_length=50)
+
+    @router.post("/v1/chat")
+    async def sync_chat(request: Request, body: SyncChatRequest):
+        if not getattr(request.state, "api_token", False):
+            raise HTTPException(403, "This endpoint requires an API token")
+        scopes = set(getattr(request.state, "api_token_scopes", []) or [])
+        if "chat" not in scopes:
+            raise HTTPException(403, "API token is not scoped for chat")
+        token_owner = getattr(request.state, "api_token_owner", None)
+
+        from core.models import ChatMessage
+        from src.llm_core import llm_call_async
+        from src.endpoint_resolver import build_chat_url, build_headers, build_models_url, normalize_base
+
+        message = body.message.strip()
+        if not message:
+            raise HTTPException(400, "Message is required")
+
+        session_id = body.session
+        sess = None
+
+        # --- Case 1: Resume an existing session ---
+        if session_id and session_manager:
+            try:
+                sess = session_manager.get_session(session_id)
+            except (KeyError, Exception):
+                raise HTTPException(404, "Session not found")
+            # SECURITY: verify the API-token's user owns this session — without
+            # this any token holder could resume any user's chat by passing its
+            # ID. The token's user is on request.state.user (set by API-token
+            # middleware); fall back to require_user if not present.
+            try:
+                from src.auth_helpers import get_current_user as _gcu
+                _tok_user = token_owner or getattr(request.state, "user", None) or _gcu(request)
+            except Exception:
+                _tok_user = None
+            # Strict ownership (see _caller_owns_session): fail closed so a
+            # null-owner / cross-owner session can't be resumed by an arbitrary
+            # chat-scoped token.
+            _sess_owner = getattr(sess, "owner", None)
+            if not _caller_owns_session(_sess_owner, _tok_user):
+                raise HTTPException(404, "Session not found")
+
+        # --- Case 2: Direct API key + model (no pre-configured endpoint needed) ---
+        if not sess and body.api_key:
+            api_key = body.api_key.strip()
+            model = body.model or "deepseek-chat"
+
+            # Validate only token-supplied direct base_url; auto-resolved known-provider
+            # URLs are not subject to extra local/LAN blocking beyond existing provider logic.
+            direct_base_url = body.base_url.strip().rstrip("/") if body.base_url else None
+            if direct_base_url:
+                try:
+                    base_url = validate_public_http_url(direct_base_url)
+                except ValueError as e:
+                    detail = str(e).replace("URL", "base_url", 1)
+                    raise HTTPException(400, detail)
+            else:
+                base_url = _resolve_base_url(model, body.provider)
+            if not base_url:
+                raise HTTPException(400,
+                    "Could not auto-detect provider. Pass base_url (e.g. 'https://api.deepseek.com/v1') "
+                    "or provider ('deepseek', 'openai', 'groq', etc.)")
+            base_url = normalize_base(base_url)
+            endpoint_url = build_chat_url(base_url)
+
+            if not session_manager:
+                raise HTTPException(500, "Session manager not available")
+
+            sid = str(uuid.uuid4())
+            sess = session_manager.create_session(
+                session_id=sid, name="API Chat", endpoint_url=endpoint_url,
+                model=model, owner=token_owner,
+            )
+            sess.headers = build_headers(api_key, base_url)
+            session_manager.save_sessions()
+            session_id = sid
+
+        # --- Case 3: Fall back to first configured ModelEndpoint ---
+        if not sess:
+            db = SessionLocal()
+            try:
+                ep = _select_api_chat_fallback_endpoint(db, token_owner)
+            finally:
+                db.close()
+
+            if not ep:
+                raise HTTPException(400,
+                    "No session, api_key, or configured endpoints. "
+                    "Pass api_key + model, or configure an endpoint in Admin.")
+
+            base_url = normalize_base(ep.base_url)
+            endpoint_url = build_chat_url(base_url)
+            model = body.model or "auto"
+            api_key = ep.api_key
+            if getattr(ep, "provider_auth_id", None):
+                try:
+                    from src.endpoint_resolver import resolve_endpoint_runtime
+                    base_url, api_key = resolve_endpoint_runtime(ep, owner=token_owner)
+                    endpoint_url = build_chat_url(base_url)
+                except Exception:
+                    raise HTTPException(500, "Could not resolve endpoint credentials")
+
+            if model == "auto":
+                try:
+                    async with httpx.AsyncClient(timeout=5) as client:
+                        models_url = build_models_url(base_url)
+                        hdrs = build_headers(api_key, base_url)
+                        if models_url:
+                            resp = await client.get(models_url, headers=hdrs)
+                            resp.raise_for_status()
+                            data = resp.json()
+                            items = data if isinstance(data, list) else (data.get("data") or [])
+                            ids = [m.get("id") for m in items if isinstance(m, dict) and m.get("id")]
+                            if not ids and isinstance(data, dict):
+                                ids = [
+                                    m.get("name") or m.get("model")
+                                    for m in (data.get("models") or [])
+                                    if m.get("name") or m.get("model")
+                                ]
+                        else:
+                            import json as _json
+                            ids = _json.loads(ep.cached_models or "[]")
+                        model = ids[0] if ids else "auto"
+                except Exception:
+                    raise HTTPException(500, "Could not discover models from endpoint")
+
+            if not session_manager:
+                raise HTTPException(500, "Session manager not available")
+
+            sid = str(uuid.uuid4())
+            sess = session_manager.create_session(
+                session_id=sid, name="API Chat", endpoint_url=endpoint_url,
+                model=model, owner=token_owner,
+            )
+            if api_key:
+                sess.headers = build_headers(api_key, base_url)
+                session_manager.save_sessions()
+            session_id = sid
+
+        # --- Send message and get response ---
+        sess.add_message(ChatMessage("user", message))
+
+        messages = [{"role": m.role, "content": m.content} for m in sess.history]
+
+        reply = await llm_call_async(
+            sess.endpoint_url, sess.model, messages,
+            headers=sess.headers, timeout=120,
+        )
+        sess.add_message(ChatMessage("assistant", reply))
+        session_manager.save_sessions()
+
+        webhook_manager.fire_and_forget("chat.completed", {
+            "session_id": session_id, "model": sess.model,
+            "user_message": message[:2000], "response": reply[:2000],
+        })
+
+        return {"response": reply, "session_id": session_id, "model": sess.model}
+
+    return router

routes/webhook_routes.py

@@ -1,395 +1,16 @@
-"""Webhook, API Token, and sync chat routes."""
+"""Backward-compat shim — canonical location is routes/webhook/webhook_routes.py.
 
-import uuid
-import logging
-from typing import Optional
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.webhook_routes``, ``from routes.webhook_routes import X``,
+``importlib.import_module("routes.webhook_routes")``, and the
+``__import__("routes.webhook_routes", fromlist=[...])`` + ``setattr(wh_mod,
+...)`` pattern used by test_null_owner_gates.py all operate on the *same*
+object. Keeps existing import paths working after slice 2l (#4082/#4071).
+Source-introspection tests read the canonical file by path.
+"""
 
-import httpx
-from fastapi import APIRouter, HTTPException, Request, Form
-from pydantic import BaseModel, Field
+import sys as _sys
 
-from core.database import SessionLocal, Webhook, ModelEndpoint
-from src.auth_helpers import owner_filter
-from src.url_security import validate_public_http_url
-from src.webhook_manager import WebhookManager, validate_webhook_url, validate_events
+from routes.webhook import webhook_routes as _canonical  # noqa: F401
 
-logger = logging.getLogger(__name__)
-
-router = APIRouter(prefix="/api", tags=["webhooks"])
-
-# Input limits
-MAX_NAME_LEN = 100
-MAX_URL_LEN = 2048
-MAX_SECRET_LEN = 256
-MAX_MESSAGE_LEN = 32_000
-
-
-from core.middleware import require_admin as _require_admin
-
-
-def _select_api_chat_fallback_endpoint(db, token_owner: Optional[str]):
-    """First enabled ModelEndpoint visible to token_owner — their own rows plus
-    legacy null-owner ("shared") rows. Owner-scoped: an unscoped .first() would
-    let a chat-scoped token fall back onto another user's private endpoint and
-    silently spend that owner's API key/quota. Prefer owner rows before shared
-    rows. Fails closed to null-owner rows only when token_owner is absent.
-    Does not validate base_url — admin-configured local/LAN endpoints remain allowed.
-    """
-    query = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True)  # noqa: E712
-    if token_owner:
-        query = owner_filter(query, ModelEndpoint, token_owner)
-        return query.order_by(ModelEndpoint.owner.desc(), ModelEndpoint.created_at).first()
-    return query.filter(ModelEndpoint.owner == None).order_by(ModelEndpoint.created_at).first()  # noqa: E711
-
-
-def _caller_owns_session(sess_owner, caller) -> bool:
-    """Strict session-ownership gate for the token-authenticated sync-chat
-    endpoint (`POST /api/v1/chat`).
-
-    Mirrors ``_verify_session_owner`` in session_routes.py and the null-owner
-    gates in notes/calendar/gallery: a caller may resume a session ONLY when
-    its owner matches them exactly. A null/empty session owner (legacy or
-    migrated rows) is deliberately NOT resumable by an arbitrary token — the
-    old ``sess_owner and sess_owner != caller`` form skipped the check whenever
-    ``sess_owner`` was falsy, so any chat-scoped token (e.g. a paired mobile
-    device) could resume such a session, inject a message, and read back its
-    history and reuse the owner's endpoint credentials. Fail closed: an
-    unresolvable caller also returns False.
-    """
-    if not caller:
-        return False
-    return sess_owner == caller
-
-
-def setup_webhook_routes(
-    webhook_manager: WebhookManager,
-    auth_manager,
-    session_manager=None,
-    api_key_manager=None,
-) -> APIRouter:
-
-    @router.get("/webhooks")
-    def list_webhooks(request: Request):
-        _require_admin(request)
-        db = SessionLocal()
-        try:
-            hooks = db.query(Webhook).all()
-            return [
-                {
-                    "id": w.id,
-                    "name": w.name,
-                    "url": w.url,
-                    "has_secret": bool(w.secret),
-                    "events": w.events.split(",") if w.events else [],
-                    "is_active": w.is_active,
-                    "last_triggered_at": w.last_triggered_at.isoformat() if w.last_triggered_at else None,
-                    "last_status_code": w.last_status_code,
-                    "last_error": w.last_error,
-                    "created_at": w.created_at.isoformat() if w.created_at else None,
-                }
-                for w in hooks
-            ]
-        finally:
-            db.close()
-
-    @router.post("/webhooks")
-    def create_webhook(
-        request: Request,
-        name: str = Form(""),
-        url: str = Form(""),
-        secret: str = Form(""),
-        events: str = Form(""),
-    ):
-        _require_admin(request)
-        name = name.strip()[:MAX_NAME_LEN]
-        if not name:
-            raise HTTPException(400, "Webhook name is required")
-        try:
-            url = validate_webhook_url(url)
-        except ValueError as e:
-            raise HTTPException(400, str(e))
-        try:
-            events = validate_events(events)
-        except ValueError as e:
-            raise HTTPException(400, str(e))
-
-        secret_val = secret.strip()[:MAX_SECRET_LEN] or None
-        # Encrypt the secret at rest using the same Fernet key as API keys
-        encrypted_secret = None
-        if secret_val and api_key_manager:
-            encrypted_secret = api_key_manager.encrypt_api_key(secret_val)
-        elif secret_val:
-            encrypted_secret = secret_val  # Fallback if no encryption available
-
-        webhook_id = str(uuid.uuid4())[:8]
-        db = SessionLocal()
-        try:
-            db.add(Webhook(
-                id=webhook_id,
-                name=name,
-                url=url,
-                secret=encrypted_secret,
-                events=events,
-                is_active=True,
-            ))
-            db.commit()
-        finally:
-            db.close()
-
-        return {"id": webhook_id, "name": name}
-
-    @router.post("/webhooks/{webhook_id}/test")
-    async def test_webhook(request: Request, webhook_id: str):
-        _require_admin(request)
-        db = SessionLocal()
-        try:
-            wh = db.query(Webhook).filter(Webhook.id == webhook_id).first()
-            if not wh:
-                raise HTTPException(404, "Webhook not found")
-            url, secret = wh.url, wh.secret
-        finally:
-            db.close()
-
-        await webhook_manager.deliver_test(webhook_id, url, secret)
-        return {"status": "sent"}
-
-    @router.patch("/webhooks/{webhook_id}")
-    def toggle_webhook(request: Request, webhook_id: str):
-        _require_admin(request)
-        db = SessionLocal()
-        try:
-            wh = db.query(Webhook).filter(Webhook.id == webhook_id).first()
-            if not wh:
-                raise HTTPException(404, "Webhook not found")
-            wh.is_active = not wh.is_active
-            db.commit()
-            return {"id": webhook_id, "is_active": wh.is_active}
-        finally:
-            db.close()
-
-    @router.delete("/webhooks/{webhook_id}")
-    def delete_webhook(request: Request, webhook_id: str):
-        _require_admin(request)
-        db = SessionLocal()
-        try:
-            deleted = db.query(Webhook).filter(Webhook.id == webhook_id).delete()
-            db.commit()
-            if not deleted:
-                raise HTTPException(404, "Webhook not found")
-        finally:
-            db.close()
-        return {"status": "deleted"}
-
-    # ================================================================
-    # Sync Chat Endpoint (for n8n / Make / Activepieces)
-    # ================================================================
-
-    # Known provider base URLs — auto-resolved from api_key prefix or model name
-    KNOWN_PROVIDERS = {
-        "deepseek": "https://api.deepseek.com/v1",
-        "openai": "https://api.openai.com/v1",
-        "mistral": "https://api.mistral.ai/v1",
-        "groq": "https://api.groq.com/openai/v1",
-        "together": "https://api.together.xyz/v1",
-        "openrouter": "https://openrouter.ai/api/v1",
-        "ollama": "https://ollama.com/api",
-        "opencode-zen": "https://opencode.ai/zen/v1",
-        "opencode-go": "https://opencode.ai/zen/go/v1",
-        "fireworks": "https://api.fireworks.ai/inference/v1",
-        "venice": "https://api.venice.ai/api/v1",
-        "kimi-code": "https://api.kimi.com/coding/v1",
-        "kimicode": "https://api.kimi.com/coding/v1",
-    }
-
-    # Model prefix → provider mapping for auto-detection
-    MODEL_PROVIDER_MAP = {
-        "deepseek": "deepseek",
-        "gpt-": "openai",
-        "o1": "openai",
-        "o3": "openai",
-        "o4": "openai",
-        "mistral": "mistral",
-        "llama": "groq",
-        "mixtral": "groq",
-        "kimi-for-coding": "kimi-code",
-        "kimi": "kimi-code",
-    }
-
-    def _resolve_base_url(model: Optional[str], provider: Optional[str]) -> Optional[str]:
-        """Try to auto-resolve a base URL from provider name or model prefix."""
-        if provider and provider.lower() in KNOWN_PROVIDERS:
-            return KNOWN_PROVIDERS[provider.lower()]
-        if model:
-            model_lower = model.lower()
-            for prefix, prov in MODEL_PROVIDER_MAP.items():
-                if model_lower.startswith(prefix):
-                    return KNOWN_PROVIDERS[prov]
-        return None
-
-    class SyncChatRequest(BaseModel):
-        message: str = Field(..., max_length=MAX_MESSAGE_LEN)
-        model: Optional[str] = Field(None, max_length=200)
-        session: Optional[str] = Field(None, max_length=100)
-        api_key: Optional[str] = Field(None, max_length=256)
-        base_url: Optional[str] = Field(None, max_length=MAX_URL_LEN)
-        provider: Optional[str] = Field(None, max_length=50)
-
-    @router.post("/v1/chat")
-    async def sync_chat(request: Request, body: SyncChatRequest):
-        if not getattr(request.state, "api_token", False):
-            raise HTTPException(403, "This endpoint requires an API token")
-        scopes = set(getattr(request.state, "api_token_scopes", []) or [])
-        if "chat" not in scopes:
-            raise HTTPException(403, "API token is not scoped for chat")
-        token_owner = getattr(request.state, "api_token_owner", None)
-
-        from core.models import ChatMessage
-        from src.llm_core import llm_call_async
-        from src.endpoint_resolver import build_chat_url, build_headers, build_models_url, normalize_base
-
-        message = body.message.strip()
-        if not message:
-            raise HTTPException(400, "Message is required")
-
-        session_id = body.session
-        sess = None
-
-        # --- Case 1: Resume an existing session ---
-        if session_id and session_manager:
-            try:
-                sess = session_manager.get_session(session_id)
-            except (KeyError, Exception):
-                raise HTTPException(404, "Session not found")
-            # SECURITY: verify the API-token's user owns this session — without
-            # this any token holder could resume any user's chat by passing its
-            # ID. The token's user is on request.state.user (set by API-token
-            # middleware); fall back to require_user if not present.
-            try:
-                from src.auth_helpers import get_current_user as _gcu
-                _tok_user = token_owner or getattr(request.state, "user", None) or _gcu(request)
-            except Exception:
-                _tok_user = None
-            # Strict ownership (see _caller_owns_session): fail closed so a
-            # null-owner / cross-owner session can't be resumed by an arbitrary
-            # chat-scoped token.
-            _sess_owner = getattr(sess, "owner", None)
-            if not _caller_owns_session(_sess_owner, _tok_user):
-                raise HTTPException(404, "Session not found")
-
-        # --- Case 2: Direct API key + model (no pre-configured endpoint needed) ---
-        if not sess and body.api_key:
-            api_key = body.api_key.strip()
-            model = body.model or "deepseek-chat"
-
-            # Validate only token-supplied direct base_url; auto-resolved known-provider
-            # URLs are not subject to extra local/LAN blocking beyond existing provider logic.
-            direct_base_url = body.base_url.strip().rstrip("/") if body.base_url else None
-            if direct_base_url:
-                try:
-                    base_url = validate_public_http_url(direct_base_url)
-                except ValueError as e:
-                    detail = str(e).replace("URL", "base_url", 1)
-                    raise HTTPException(400, detail)
-            else:
-                base_url = _resolve_base_url(model, body.provider)
-            if not base_url:
-                raise HTTPException(400,
-                    "Could not auto-detect provider. Pass base_url (e.g. 'https://api.deepseek.com/v1') "
-                    "or provider ('deepseek', 'openai', 'groq', etc.)")
-            base_url = normalize_base(base_url)
-            endpoint_url = build_chat_url(base_url)
-
-            if not session_manager:
-                raise HTTPException(500, "Session manager not available")
-
-            sid = str(uuid.uuid4())
-            sess = session_manager.create_session(
-                session_id=sid, name="API Chat", endpoint_url=endpoint_url,
-                model=model, owner=token_owner,
-            )
-            sess.headers = build_headers(api_key, base_url)
-            session_manager.save_sessions()
-            session_id = sid
-
-        # --- Case 3: Fall back to first configured ModelEndpoint ---
-        if not sess:
-            db = SessionLocal()
-            try:
-                ep = _select_api_chat_fallback_endpoint(db, token_owner)
-            finally:
-                db.close()
-
-            if not ep:
-                raise HTTPException(400,
-                    "No session, api_key, or configured endpoints. "
-                    "Pass api_key + model, or configure an endpoint in Admin.")
-
-            base_url = normalize_base(ep.base_url)
-            endpoint_url = build_chat_url(base_url)
-            model = body.model or "auto"
-            api_key = ep.api_key
-            if getattr(ep, "provider_auth_id", None):
-                try:
-                    from src.endpoint_resolver import resolve_endpoint_runtime
-                    base_url, api_key = resolve_endpoint_runtime(ep, owner=token_owner)
-                    endpoint_url = build_chat_url(base_url)
-                except Exception:
-                    raise HTTPException(500, "Could not resolve endpoint credentials")
-
-            if model == "auto":
-                try:
-                    async with httpx.AsyncClient(timeout=5) as client:
-                        models_url = build_models_url(base_url)
-                        hdrs = build_headers(api_key, base_url)
-                        if models_url:
-                            resp = await client.get(models_url, headers=hdrs)
-                            resp.raise_for_status()
-                            data = resp.json()
-                            items = data if isinstance(data, list) else (data.get("data") or [])
-                            ids = [m.get("id") for m in items if isinstance(m, dict) and m.get("id")]
-                            if not ids and isinstance(data, dict):
-                                ids = [
-                                    m.get("name") or m.get("model")
-                                    for m in (data.get("models") or [])
-                                    if m.get("name") or m.get("model")
-                                ]
-                        else:
-                            import json as _json
-                            ids = _json.loads(ep.cached_models or "[]")
-                        model = ids[0] if ids else "auto"
-                except Exception:
-                    raise HTTPException(500, "Could not discover models from endpoint")
-
-            if not session_manager:
-                raise HTTPException(500, "Session manager not available")
-
-            sid = str(uuid.uuid4())
-            sess = session_manager.create_session(
-                session_id=sid, name="API Chat", endpoint_url=endpoint_url,
-                model=model, owner=token_owner,
-            )
-            if api_key:
-                sess.headers = build_headers(api_key, base_url)
-                session_manager.save_sessions()
-            session_id = sid
-
-        # --- Send message and get response ---
-        sess.add_message(ChatMessage("user", message))
-
-        messages = [{"role": m.role, "content": m.content} for m in sess.history]
-
-        reply = await llm_call_async(
-            sess.endpoint_url, sess.model, messages,
-            headers=sess.headers, timeout=120,
-        )
-        sess.add_message(ChatMessage("assistant", reply))
-        session_manager.save_sessions()
-
-        webhook_manager.fire_and_forget("chat.completed", {
-            "session_id": session_id, "model": sess.model,
-            "user_message": message[:2000], "response": reply[:2000],
-        })
-
-        return {"response": reply, "session_id": session_id, "model": sess.model}
-
-    return router
+_sys.modules[__name__] = _canonical

services/memory/__init__.py

@@ -2,7 +2,7 @@
 """Memory service — persistent memory storage and retrieval."""
 
 from .service import MemoryService, Memory, MemorySearchResult
-from .memory import MemoryManager
+from .memory import MemoryManager, MemoryStoreUnreadable
 from .memory_vector import MemoryVectorStore
 
 __all__ = [
@@ -10,5 +10,6 @@ __all__ = [
     "Memory",
     "MemorySearchResult",
     "MemoryManager",
+    "MemoryStoreUnreadable",
     "MemoryVectorStore",
 ]

services/memory/memory.py

@@ -5,6 +5,16 @@ application runtime instantiates ``src.memory.MemoryManager``, so keeping a
 parallel implementation here risks silent drift between import paths.
 """
 
-from src.memory import MemoryManager, get_text_similarity, tokenize
+from src.memory import (
+    MemoryManager,
+    MemoryStoreUnreadable,
+    get_text_similarity,
+    tokenize,
+)
 
-__all__ = ["MemoryManager", "get_text_similarity", "tokenize"]
+__all__ = [
+    "MemoryManager",
+    "MemoryStoreUnreadable",
+    "get_text_similarity",
+    "tokenize",
+]

services/memory/memory_extractor.py

@@ -17,6 +17,8 @@ import os
 import re
 from typing import Optional
 
+from src.memory import MemoryStoreUnreadable
+
 logger = logging.getLogger(__name__)
 
 
@@ -387,7 +389,13 @@ async def extract_and_store(
         # Get owner from session
         _owner = getattr(session, 'owner', None)
 
-        existing = memory_manager.load_all()
+        # Strict load: this is a read-modify-write. Degrading to [] here would
+        # save only the newly extracted facts and drop the entire store.
+        try:
+            existing = memory_manager.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            logger.error("Skipping auto memory extraction, store unreadable: %s", e)
+            return
         added = 0
 
         for fact in facts:
@@ -626,7 +634,18 @@ async def audit_memories(
 
         # Merge audited entries back with other users' entries
         if owner:
-            all_entries = memory_manager.load_all()
+            # Strict load: the merge below reconstructs the whole file. If this
+            # degraded to [] we would save only this owner's audited slice and
+            # destroy every other tenant's memories.
+            try:
+                all_entries = memory_manager.load_all_for_update()
+            except MemoryStoreUnreadable as e:
+                logger.error("Aborting memory audit save, store unreadable: %s", e)
+                return {
+                    "before": before_count,
+                    "after": before_count,
+                    "error": "store_unreadable",
+                }
             audited_ids = {e["id"] for e in final_entries}
             other_entries = [e for e in all_entries if e.get("owner") != owner and (e.get("owner") is not None)]
             # Also keep legacy entries that weren't part of this audit

services/memory/skill_format.py

@@ -50,7 +50,7 @@ import json
 import logging
 import re
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
@@ -441,4 +441,4 @@ class Skill:
 
 
 def _now_iso() -> str:
-    return datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

services/tts/tts_service.py

@@ -2,6 +2,7 @@
 """Multi-provider TTS service — dispatches to local Kokoro, OpenAI-compatible API, or browser."""
 
 import io
+import os
 import wave
 import logging
 import hashlib
@@ -42,6 +43,11 @@ class TTSService:
         self.cache_dir.mkdir(parents=True, exist_ok=True)
         self._kokoro = None  # lazy-init
         
+        try:
+            self.max_cache_bytes = int(os.getenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", 500 * 1024 * 1024))
+        except ValueError:
+            self.max_cache_bytes = 500 * 1024 * 1024
+
     # ── Settings ──
 
     def _load_settings(self) -> dict:
@@ -89,6 +95,53 @@ class TTSService:
         ext = ".mp3" if (len(data) >= 3 and (data[:3] == b'ID3' or (data[0] == 0xff and (data[1] & 0xe0) == 0xe0))) else ".wav"
         (self.cache_dir / f"{key}{ext}").write_bytes(data)
 
+        self._enforce_cache_limit()
+
+    def _enforce_cache_limit(self):
+            """Evicts oldest files if the cache exceeds the configured byte limit."""
+            if self.max_cache_bytes <= 0:
+                return
+
+            try:
+                files = []
+                total_size = 0
+
+                # Safely scan files and sum sizes, ignoring files deleted mid-scan
+                for f in self.cache_dir.iterdir():
+                    try:
+                        if f.is_file() and f.suffix.lower() in (".mp3", ".wav"):
+                            files.append(f)
+                            total_size += f.stat().st_size
+                    except OSError:
+                        continue
+
+                if total_size > self.max_cache_bytes:
+                    logger.info(
+                        f"TTS cache ({total_size} bytes) exceeded limit ({self.max_cache_bytes} bytes). Evicting oldest files."
+                    )
+
+                    # Sort files by modification time (oldest first)
+                    try:
+                        files.sort(key=lambda f: f.stat().st_mtime)
+                    except OSError as e:
+                        logger.warning(f"Failed to sort cache files by mtime: {e}")
+
+                    # Trim down to 80% of max capacity
+                    target_size = self.max_cache_bytes * 0.8
+
+                    while files and total_size > target_size:
+                        f = files.pop(0)
+                        try:
+                            size = f.stat().st_size
+                            f.unlink()
+                            total_size -= size
+                        except OSError as e:
+                            logger.warning(f"Failed to evict cache file {f}: {e}")
+                            continue
+
+            except Exception as e:
+                logger.warning(f"Error enforcing TTS cache limit: {e}", exc_info=True)
+
     def clear_cache(self):
         count = 0
         for f in self.cache_dir.glob("*.*"):

src/ai_interaction.py

@@ -22,6 +22,7 @@ import time
 from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
 
 from src.constants import GENERATED_IMAGES_DIR
+from src.memory import MemoryStoreUnreadable
 
 logger = logging.getLogger(__name__)
 
@@ -384,7 +385,15 @@ async def do_manage_memory(content: str, session_id: Optional[str] = None, owner
             return {"error": "Memory text cannot be empty"}
 
         entry = _memory_manager.add_entry(text, source="ai_agent", category=category, owner=owner)
-        memories = _memory_manager.load_all()
+        # Strict load: this is a read-modify-write, and it is the path an
+        # ordinary "remember that I prefer X" takes. Degrading to [] here would
+        # save just this one entry over a store we only failed to read,
+        # atomically destroying every memory in it (issue #5673).
+        try:
+            memories = _memory_manager.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            logger.error("Refusing to add memory, store unreadable: %s", e)
+            return {"error": "Memory store is temporarily unreadable — nothing was saved."}
         memories.append(entry)
         _memory_manager.save(memories)
 

src/integrations.py

@@ -1,11 +1,14 @@
+import ipaddress
 import json
 import os
+import time
 import uuid
 import logging
 import re
 from typing import Dict, List, Optional, Any
 from urllib.parse import urljoin, urlparse, urlunparse
 
+import httpcore
 import httpx
 from fastapi import HTTPException
 
@@ -354,6 +357,152 @@ def _find_integration(identifier: str) -> Optional[Dict[str, Any]]:
     return None
 
 
+# httpcore raises its own exception hierarchy; map the ones a simple request can
+# surface back to their httpx equivalents so the caller's `except httpx.*` blocks
+# below behave exactly as they did with the default transport.
+_HTTPCORE_TO_HTTPX_EXC = {
+    httpcore.ConnectError: httpx.ConnectError,
+    httpcore.ConnectTimeout: httpx.ConnectTimeout,
+    httpcore.NetworkError: httpx.NetworkError,
+    httpcore.PoolTimeout: httpx.PoolTimeout,
+    httpcore.ProtocolError: httpx.ProtocolError,
+    httpcore.ReadError: httpx.ReadError,
+    httpcore.ReadTimeout: httpx.ReadTimeout,
+    httpcore.RemoteProtocolError: httpx.RemoteProtocolError,
+    httpcore.TimeoutException: httpx.TimeoutException,
+    httpcore.WriteError: httpx.WriteError,
+    httpcore.WriteTimeout: httpx.WriteTimeout,
+}
+
+
+class _PinnedAsyncBackend(httpcore.AsyncNetworkBackend):
+    """Network backend that connects only to the pre-validated IPs, in order.
+
+    Every address here came out of the single SSRF resolution, so moving to the
+    next one after a connect failure is not re-resolution — it's ordinary
+    multi-address fallback restricted to the set the guard already approved.
+    httpcore takes TLS SNI and the ``Host`` header from the request URL rather
+    than the connect host, so pinning the socket destination leaves certificate
+    validation and vhost routing pointed at the original hostname.
+    """
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._ips = [str(ip) for ip in ips]
+        self._real = httpcore.AnyIOBackend()
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None,
+                          socket_options=None):
+        # One shared connect budget: each attempt gets the time left until the
+        # original deadline, so N dead addresses can't stretch the connect
+        # phase to N * timeout.
+        deadline = None if timeout is None else time.monotonic() + timeout
+        last_exc: Optional[Exception] = None
+        for ip in self._ips:
+            remaining = None if deadline is None else max(0.0, deadline - time.monotonic())
+            try:
+                return await self._real.connect_tcp(
+                    ip, port, remaining, local_address, socket_options
+                )
+            except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc:
+                last_exc = exc
+                if deadline is not None and time.monotonic() >= deadline:
+                    break
+        raise last_exc
+
+    async def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        return await self._real.connect_unix_socket(path, timeout, socket_options)
+
+    async def sleep(self, seconds: float) -> None:
+        return await self._real.sleep(seconds)
+
+
+class _PinnedAsyncTransport(httpx.AsyncBaseTransport):
+    """httpx transport that pins the TCP connect to the pre-resolved IP(s).
+
+    Kept local, mirroring the per-module pinned transports web fetch and
+    webhook delivery already carry, rather than coupling api_call to the
+    webhook subsystem. The request URL passes through unchanged, so SNI and the
+    ``Host`` header stay the original hostname; only the socket destination is
+    pinned, which is what closes the rebinding window.
+    """
+
+    def __init__(self, ips: List[ipaddress._BaseAddress]):
+        self._pinned_ips = list(ips)
+        self._pool = httpcore.AsyncConnectionPool(
+            # Reuse the CA trust the default httpx client would build (certifi
+            # plus SSL_CERT_FILE / SSL_CERT_DIR when trust_env is set) so
+            # swapping in this transport doesn't quietly change which chains
+            # verify. ssl.create_default_context() would use system roots.
+            ssl_context=httpx.create_ssl_context(),
+            http1=True,
+            http2=False,
+            network_backend=_PinnedAsyncBackend(ips),
+        )
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        core_req = httpcore.Request(
+            method=request.method,
+            url=httpcore.URL(
+                scheme=request.url.raw_scheme,
+                host=request.url.raw_host,
+                port=request.url.port,
+                target=request.url.raw_path,
+            ),
+            headers=request.headers.raw,
+            content=request.stream,
+            extensions=request.extensions,
+        )
+        try:
+            core_resp = await self._pool.handle_async_request(core_req)
+            content = b"".join([chunk async for chunk in core_resp.aiter_stream()])
+            await core_resp.aclose()
+        except Exception as exc:
+            mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc))
+            if mapped is not None:
+                raise mapped(str(exc)) from exc
+            raise
+        return httpx.Response(
+            status_code=core_resp.status,
+            headers=core_resp.headers,
+            content=content,
+            extensions=core_resp.extensions,
+        )
+
+    async def aclose(self) -> None:
+        await self._pool.aclose()
+
+
+def _validated_ips(raw_ips: List[str]) -> List[ipaddress._BaseAddress]:
+    """Return every entry that parses as an IP address, de-duplicated, order
+    preserved.
+
+    check_outbound_url only reports ok when *all* of these classify as safe, so
+    the whole list is guard-approved and any of them is a legitimate connect
+    target. Skipping unparseable entries mirrors how the guard walks the same
+    resolver output.
+
+    De-duplication matters because the resolver is getaddrinfo(host, None) with
+    no socktype filter, so glibc reports the same address once per socktype
+    (SOCK_STREAM/SOCK_DGRAM/SOCK_RAW) — a single-homed host comes back three
+    times. Without this, the connect fallback would spend the shared deadline
+    retrying one dead address instead of moving on to a genuinely different one.
+    """
+    ips: List[ipaddress._BaseAddress] = []
+    seen = set()
+    for raw in raw_ips:
+        if not isinstance(raw, str):
+            continue
+        try:
+            ip = ipaddress.ip_address(raw.split("%")[0])  # strip IPv6 zone id
+        except ValueError:
+            continue
+        if ip in seen:
+            continue
+        seen.add(ip)
+        ips.append(ip)
+    return ips
+
+
 async def execute_api_call(
     integration_id: str,
     method: str,
@@ -409,13 +558,31 @@ async def execute_api_call(
     # loopback for locked-down deployments. Private stays allowed by default
     # because LAN integrations (Home Assistant, Miniflux, ntfy) are the
     # primary use case.
-    from src.url_safety import check_outbound_url
+    from src.url_safety import check_outbound_url, _default_resolver
     block_private = os.getenv(
         "INTEGRATION_API_BLOCK_PRIVATE_IPS", "false"
     ).lower() == "true"
-    ok, reason = check_outbound_url(url, block_private=block_private)
+    # Resolve the host exactly once and remember the IPs the guard validated so
+    # the request below can be pinned to them. check_outbound_url only reports
+    # (ok, reason); a plain httpx client re-resolves the host at connect time,
+    # which reopens a DNS-rebinding TOCTOU — a base_url host that answers with a
+    # public IP for the guard and then flips to 169.254.169.254 for the connect
+    # would reach cloud metadata with the integration's auth headers attached.
+    resolved_ips: List[str] = []
+
+    def _recording_resolver(host: str) -> List[str]:
+        ips = _default_resolver(host)
+        resolved_ips[:] = ips
+        return ips
+
+    ok, reason = check_outbound_url(
+        url, block_private=block_private, resolver=_recording_resolver
+    )
     if not ok:
         return {"error": f"URL rejected: {reason}", "exit_code": 1}
+    pinned_ips = _validated_ips(resolved_ips)
+    if not pinned_ips:
+        return {"error": "URL rejected: host did not resolve to a usable address", "exit_code": 1}
 
     method = method.upper()
 
@@ -455,7 +622,9 @@ async def execute_api_call(
             auth = httpx.BasicAuth(parts[0], parts[1])
 
     try:
-        async with httpx.AsyncClient(timeout=30.0) as client:
+        async with httpx.AsyncClient(
+            timeout=30.0, transport=_PinnedAsyncTransport(pinned_ips)
+        ) as client:
             response = await client.request(
                 method,
                 url,

src/llm_core.py

@@ -1237,15 +1237,27 @@ def _anthropic_rejects_temperature(model: str) -> bool:
         return False
     # `(?<![a-z])` anchors "opus" to a word boundary so a substring match like
     # `oct-opus`/`octopus-4-8` can't be read as Opus (it would otherwise strip
-    # temperature). Cap the minor at 1-2 digits and forbid a trailing digit so a
-    # dated id like `claude-opus-4-20250514` (Opus 4.0) parses as major-only (no
-    # minor match, kept) instead of reading the date `20250514` as a giant minor
-    # that would falsely test >= 4.7. Dated 4.7+ snapshots (`claude-opus-4-7-
-    # 20260201`) keep their explicit minor and are still matched.
-    match = re.search(r"(?<![a-z])opus[-_]?(\d+)[-_.](\d{1,2})(?!\d)", model.lower())
+    # temperature). Both version components are capped at 1-2 digits and forbid a
+    # trailing digit, so an 8-digit date can never be read as a version number:
+    # `claude-opus-4-20250514` (Opus 4.0) parses as major-only rather than reading
+    # `20250514` as a giant minor, and `claude-3-opus-20240229` (legacy Claude 3
+    # Opus, date directly after "opus-") fails to match at all rather than reading
+    # the date as a giant major. Dated 4.7+ snapshots (`claude-opus-4-7-20260201`)
+    # keep their explicit minor and are still matched.
+    #
+    # The minor is optional and a missing minor reads as `.0`, so major-only ids
+    # like `claude-opus-5` are correctly treated as >= 4.7 (issue #5753). Without
+    # this, every Opus 5 call kept `temperature` and failed with HTTP 400 — visible
+    # only on paths that pass a temperature, e.g. scheduled tasks inheriting
+    # `stream_agent_loop`'s 0.3 default, which returned empty responses.
+    match = re.search(
+        r"(?<![a-z])opus[-_]?(\d{1,2})(?!\d)(?:[-_.](\d{1,2})(?!\d))?", model.lower()
+    )
     if not match:
         return False
-    return (int(match.group(1)), int(match.group(2))) >= (4, 7)
+    major = int(match.group(1))
+    minor = int(match.group(2)) if match.group(2) else 0
+    return (major, minor) >= (4, 7)
 
 # Reasoning effort level sent to Mistral thinking-capable models. Mistral's
 # API accepts "high", "medium", "low", "none" — see

src/memory.py

@@ -10,6 +10,18 @@ from datetime import datetime
 
 logger = logging.getLogger(__name__)
 
+
+class MemoryStoreUnreadable(RuntimeError):
+    """memory.json exists on disk but could not be read or parsed.
+
+    "The contents are unknown" is categorically different from "there are no
+    memories". A read-modify-write caller that conflates the two appends to an
+    empty view and then persists it, destroying the whole store — the writes
+    are atomic, so the loss is durable. Raised by
+    :meth:`MemoryManager.load_all_for_update` so those callers fail closed.
+    """
+
+
 def tokenize(text: str) -> List[str]:
     """Simple tokenizer that splits on whitespace and removes punctuation."""
     return [word.strip('.,!?";') for word in text.split()]
@@ -110,21 +122,69 @@ class MemoryManager:
             with open(self.memory_file, 'w', encoding='utf-8') as f:
                 json.dump([], f, ensure_ascii=False, indent=2)
     
-    def load_all(self) -> List[Dict]:
-        """Load all memory entries from JSON file (unfiltered)."""
+    def _read_entries(self) -> List[Dict]:
+        """Parse the store, or raise :class:`MemoryStoreUnreadable`.
+
+        Returns ``[]`` only when the file genuinely does not exist. Every other
+        failure mode raises, so callers can tell "no memories" apart from
+        "couldn't read the memories".
+        """
         if not os.path.exists(self.memory_file):
             return []
 
         try:
             with open(self.memory_file, "r", encoding="utf-8") as f:
                 data = json.load(f)
-                if isinstance(data, list):
-                    return self._validate_entries(data)
-        except (json.JSONDecodeError, PermissionError) as e:
-            logger.error("Error loading memory.json: %s", e)
-            return self._migrate_from_legacy()
+        except OSError as e:
+            # PermissionError is an OSError (a scanner holding the file, a
+            # permissions problem, bad media).
+            raise MemoryStoreUnreadable(
+                f"cannot read {self.memory_file}: {e}"
+            ) from e
+        except json.JSONDecodeError as e:
+            # This is the branch that actually destroyed stores: the file reads
+            # back fine, so nothing stops the save that follows. A truncated
+            # memory.json is reachable because core/database.py rewrites it with
+            # a plain open(..,"w") + json.dump during migration.
+            #
+            # Preserved behaviour: a corrupt store still gets one shot at the
+            # pre-JSON memory.txt migration. Only raise when that finds nothing,
+            # so we never report "empty" for a store we simply failed to parse.
+            legacy = self._migrate_from_legacy()
+            if legacy:
+                return legacy
+            raise MemoryStoreUnreadable(
+                f"{self.memory_file} is not valid JSON: {e}"
+            ) from e
 
-        return []
+        if not isinstance(data, list):
+            raise MemoryStoreUnreadable(
+                f"{self.memory_file} is not a JSON array (got {type(data).__name__})"
+            )
+        return self._validate_entries(data)
+
+    def load_all(self) -> List[Dict]:
+        """Load all memory entries from JSON file (unfiltered).
+
+        Lenient by design: this feeds display, search, and context-injection
+        paths, so an unreadable store degrades to an empty list rather than
+        breaking chat. Never build a value from this that you intend to save
+        back — use :meth:`load_all_for_update` for that.
+        """
+        try:
+            return self._read_entries()
+        except MemoryStoreUnreadable as e:
+            logger.error("Error loading memory.json: %s", e)
+            return []
+
+    def load_all_for_update(self) -> List[Dict]:
+        """Load for a read-modify-write cycle.
+
+        Propagates :class:`MemoryStoreUnreadable` instead of degrading to ``[]``
+        so a caller can never append to an empty view and persist it over a
+        store that was only temporarily unreadable (issue #5673).
+        """
+        return self._read_entries()
 
     def load(self, owner: str = None) -> List[Dict]:
         """Load memory entries, optionally filtered by owner."""
@@ -135,7 +195,12 @@ class MemoryManager:
 
     def claim_ownerless(self, owner: str):
         """Assign all ownerless memory entries to the given owner."""
-        entries = self.load_all()
+        try:
+            entries = self.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            # Skip the sweep rather than rewrite the store from an unknown view.
+            logger.error("Skipping ownerless claim, memory store unreadable: %s", e)
+            return
         changed = False
         claimed = 0
         for entry in entries:
@@ -235,7 +300,12 @@ class MemoryManager:
         if not ids:
             return
         id_set = set(ids)
-        entries = self.load_all()
+        try:
+            entries = self.load_all_for_update()
+        except MemoryStoreUnreadable as e:
+            # Best-effort counter; never worth rewriting the store blind.
+            logger.error("Skipping uses bump, memory store unreadable: %s", e)
+            return
         changed = False
         for e in entries:
             if e.get("id") in id_set:

src/memory_provider.py

@@ -157,7 +157,11 @@ class NativeMemoryProvider(MemoryProvider):
         if metadata:
             entry["metadata"] = dict(metadata)
 
-        memories = self.memory_manager.load_all()
+        # Strict load: read-modify-write. `load_all` degrades an unreadable
+        # store to [], which would save this single entry over everything
+        # already stored (issue #5673). The provider API has no error channel,
+        # so MemoryStoreUnreadable propagates to the caller.
+        memories = self.memory_manager.load_all_for_update()
         memories.append(entry)
         self.memory_manager.save(memories)
 
@@ -223,7 +227,10 @@ class NativeMemoryProvider(MemoryProvider):
         ]
 
     async def delete(self, memory_id: str, *, owner: Optional[str] = None) -> bool:
-        memories = self.memory_manager.load_all()
+        # Strict load for the same reason: `remaining` is derived from this
+        # list and saved back, so it must never be built from a store we
+        # failed to read.
+        memories = self.memory_manager.load_all_for_update()
         remaining = []
         deleted_id = None
 

src/tools/system.py

@@ -46,7 +46,9 @@ async def do_manage_skills(content: str, owner: Optional[str] = None) -> Dict:
     except ValueError:
         return {"error": "Invalid JSON arguments", "exit_code": 1}
 
-    action = (args.get("action") or "").lower()
+    action = (args.get("action") or "").strip().lower()
+    if not action:
+        return {"error": "action is required (list|view|view_ref|add|edit|patch|publish|delete|search)", "exit_code": 1}
     from services.memory.skills import SkillsManager
     from services.memory.skill_format import Skill, slugify
     from src.constants import DATA_DIR
@@ -55,7 +57,7 @@ async def do_manage_skills(content: str, owner: Optional[str] = None) -> Dict:
     # Accept legacy `skill_id` as an alias for `name`.
     name = (args.get("name") or args.get("skill_id") or "").strip()
 
-    if action in ("list", "index", ""):
+    if action in ("list", "index"):
         all_skills = sm.load(owner=owner)
         if not all_skills:
             return {"results": "No skills yet. Create one with action='add'."}

static/js/markdown.js

@@ -758,30 +758,36 @@ export function mdToHtml(src, opts) {
   // Remove empty paragraphs
   s = s.replace(/<p><\/p>/g, '');
 
+  // Every restore below passes a function replacer rather than the block string
+  // itself. With a string replacement, `String.replace` reads `$&`, `` $` ``,
+  // `$'` and `$$` in the *replacement* as substitution patterns, so a restored
+  // block containing them is corrupted: `$&` re-inserts the placeholder, `` $` ``
+  // and `$'` splice in the surrounding document, and `$$` collapses to `$`. Those
+  // sequences are ordinary content in fenced code (`perl -pe 's/x/$& y/'`,
+  // `echo "$$USD"`). A function replacer inserts its return value verbatim.
+
   // CRITICAL: Restore allowed HTML blocks first
   allowedHtmlBlocks.forEach((block, index) => {
-    s = s.replace(`___ALLOWED_HTML_${index}___`, block);
+    s = s.replace(`___ALLOWED_HTML_${index}___`, () => block);
   });
 
   // Restore math blocks
   mathBlocks.forEach((block, index) => {
-    s = s.replace(`___MATH_BLOCK_${index}___`, block);
+    s = s.replace(`___MATH_BLOCK_${index}___`, () => block);
   });
 
   // Restore mermaid diagram blocks
   mermaidBlocks.forEach((block, index) => {
-    s = s.replace(`___MERMAID_BLOCK_${index}___`, block);
+    s = s.replace(`___MERMAID_BLOCK_${index}___`, () => block);
   });
 
   // CRITICAL: Restore code blocks at the end
   codeBlocks.forEach((block, index) => {
-    s = s.replace(`___CODE_BLOCK_${index}___`, block);
+    s = s.replace(`___CODE_BLOCK_${index}___`, () => block);
   });
 
   // Restore inline code spans last, so placeholders carried inside restored
-  // <a>/allowed-HTML blocks are resolved too. The function replacer keeps the
-  // escaped code literal — e.g. a shell snippet like `echo $1` is not treated
-  // as a regex back-reference.
+  // <a>/allowed-HTML blocks are resolved too.
   inlineCodeBlocks.forEach((block, index) => {
     s = s.replace(`___INLINE_CODE_${index}___`, () => block);
   });

static/js/settings.js

@@ -3031,12 +3031,14 @@ async function initEmailAccountsSettings() {
       const body = {
         name: el('eaf-name').value.trim() || el('eaf-from').value.trim(),
         from_address: el('eaf-from').value.trim(),
+        display_name: el('eaf-display-name').value.trim(),
         imap_host: el('eaf-imap-host').value.trim(),
         imap_port: parseInt(el('eaf-imap-port').value) || 993,
         imap_user: el('eaf-imap-user').value.trim(),
         imap_starttls: el('eaf-imap-starttls').checked,
         smtp_host: el('eaf-smtp-host').value.trim(),
         smtp_port: parseInt(el('eaf-smtp-port').value) || 587,
+        smtp_security: el('eaf-smtp-security').value,
         smtp_user: el('eaf-imap-user').value.trim(),
       };
       if (!body.name) { el('eaf-msg').textContent = 'Enter a Name or Email first'; el('eaf-msg').style.color = 'var(--red)'; return; }
@@ -5788,29 +5790,30 @@ export function close() {
   window.history.replaceState(null, '', clean);
   const success = sp.has('email_oauth_success');
   const errMsg = sp.get('email_oauth_error') || '';
-  // Open settings → integrations after the app has initialised.
-  function _tryOpen() {
-    if (window.settingsModule && typeof window.settingsModule.open === 'function') {
-      window.settingsModule.open('integrations');
-      // Brief toast-style banner.
-      const banner = document.createElement('div');
-      banner.textContent = success
-        ? '✓ Google account connected — email is ready'
-        : `Google OAuth failed: ${errMsg || 'unknown error'}`;
-      Object.assign(banner.style, {
-        position: 'fixed', bottom: '24px', left: '50%', transform: 'translateX(-50%)',
-        background: success ? 'var(--accent, #50fa7b)' : 'var(--red, #ff5555)',
-        color: '#000', padding: '8px 18px', borderRadius: '6px', fontSize: '12px',
-        fontWeight: '600', zIndex: '99999', pointerEvents: 'none',
-        boxShadow: '0 2px 12px rgba(0,0,0,0.3)',
-      });
-      document.body.appendChild(banner);
-      setTimeout(() => banner.remove(), 4000);
-    } else {
-      setTimeout(_tryOpen, 100);
-    }
+  // Open settings → integrations once the document is ready. This module owns
+  // the open() API, so it does not need to wait for a window-level alias.
+  function _showResult() {
+    open('integrations');
+    // Brief toast-style banner.
+    const banner = document.createElement('div');
+    banner.textContent = success
+      ? 'Google account connected — email is ready'
+      : `Google OAuth failed: ${errMsg || 'unknown error'}`;
+    Object.assign(banner.style, {
+      position: 'fixed', bottom: '24px', left: '50%', transform: 'translateX(-50%)',
+      background: success ? 'var(--accent, #50fa7b)' : 'var(--red, #ff5555)',
+      color: '#000', padding: '8px 18px', borderRadius: '6px', fontSize: '12px',
+      fontWeight: '600', zIndex: '99999', pointerEvents: 'none',
+      boxShadow: '0 2px 12px rgba(0,0,0,0.3)',
+    });
+    document.body.appendChild(banner);
+    setTimeout(() => banner.remove(), 4000);
+  }
+  if (document.readyState === 'loading') {
+    document.addEventListener('DOMContentLoaded', _showResult, { once: true });
+  } else {
+    _showResult();
   }
-  _tryOpen();
 })();
 
 const settingsModule = { open, close, initIntegrations, initUnifiedIntegrations, syncAdminVisibility, refreshAiModelEndpoints };

tests/test_add_directory_event_loop.py

@@ -1,253 +0,0 @@
-"""Regression guard for #5558 — POST /api/personal/add_directory must not run
-the indexing job on the event loop.
-
-The handler is ``async def`` but called ``rag.index_personal_documents``
-(os.walk + file reads + per-chunk embedding + Chroma inserts) inline, so
-FastAPI ran the whole job on the event loop and every other request queued
-behind it: indexing a real directory froze the UI and API for 25+ minutes.
-``personal_docs_manager.add_directory`` sits in the same blocking section — it
-triggers ``refresh_index()``, which re-extracts text across tracked dirs.
-
-These tests build the real router with fake managers and compare the thread
-the indexing work runs on against the event loop's thread.
-"""
-import asyncio
-import os
-import threading
-
-os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
-
-import httpx
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-
-def _serialization_probe():
-    """Shared counter proving two critical sections never overlap."""
-    state = {"active": 0, "max_active": 0}
-    lock = threading.Lock()
-
-    def enter():
-        with lock:
-            state["active"] += 1
-            state["max_active"] = max(state["max_active"], state["active"])
-
-    def leave():
-        with lock:
-            state["active"] -= 1
-
-    return state, enter, leave
-
-
-# Concurrency tests are `async def` (pyproject asyncio_mode="auto") and drive the
-# ASGI app through httpx.ASGITransport + AsyncClient + asyncio.gather, NOT starlette
-# TestClient + ThreadPoolExecutor: the job lock is an asyncio.Lock acquired in the
-# async handler, and TestClient's portal-thread dispatch deadlocks against it (same
-# reason test_notes_fail_closed_auth.py uses ASGITransport). asyncio.gather runs both
-# requests on the test's own loop.
-def _async_client(app):
-    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t")
-
-import routes.personal_routes as personal_routes
-from core.middleware import require_admin
-from src.auth_helpers import require_user
-
-
-class _FakeRag:
-    def __init__(self, record):
-        self._record = record
-
-    def index_personal_documents(self, directory, owner=None):
-        self._record["index_thread"] = threading.get_ident()
-        return {"success": True, "indexed_count": 3, "failed_count": 0}
-
-
-class _FakeDocsManager:
-    def __init__(self, record):
-        self._record = record
-        self.index = []
-
-    def add_directory(self, directory, *, index=True, owner=None):
-        self._record["bookkeeping_thread"] = threading.get_ident()
-        self._record["bookkeeping_index_flag"] = index
-
-
-def _build_app(tmp_path, monkeypatch, record):
-    monkeypatch.setattr(personal_routes, "PERSONAL_DIR", str(tmp_path))
-    monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: _FakeRag(record))
-
-    app = FastAPI()
-    app.include_router(
-        personal_routes.setup_personal_routes(_FakeDocsManager(record), None, True)
-    )
-    app.dependency_overrides[require_user] = lambda: "tester"
-    app.dependency_overrides[require_admin] = lambda: None
-
-    @app.get("/loop-thread")
-    async def loop_thread_probe():
-        return {"thread": threading.get_ident()}
-
-    return app
-
-
-def test_indexing_runs_off_the_event_loop(tmp_path, monkeypatch):
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    target = tmp_path / "docs"
-    target.mkdir()
-
-    # Context-manager client: one portal/event loop serves both requests, so
-    # the probe and the POST are guaranteed to see the same loop thread.
-    with TestClient(app) as client:
-        loop_thread = client.get("/loop-thread").json()["thread"]
-        resp = client.post(
-            "/api/personal/add_directory", json={"directory": str(target)}
-        )
-
-    assert resp.status_code == 200
-    assert record["index_thread"] != loop_thread, (
-        "index_personal_documents ran on the event loop thread — every other "
-        "request queues behind the indexing job (#5558)"
-    )
-    assert record["bookkeeping_thread"] != loop_thread, (
-        "personal_docs_manager.add_directory (refresh_index) ran on the event "
-        "loop thread"
-    )
-
-
-def test_response_and_bookkeeping_unchanged(tmp_path, monkeypatch):
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    target = tmp_path / "docs"
-    target.mkdir()
-
-    client = TestClient(app)
-    resp = client.post("/api/personal/add_directory", json={"directory": str(target)})
-
-    assert resp.status_code == 200
-    body = resp.json()
-    assert body["success"] is True
-    assert body["indexed_count"] == 3
-    assert body["failed_count"] == 0
-    assert body["directory"] == os.path.realpath(str(target))
-    assert record["bookkeeping_index_flag"] is False
-
-
-async def test_concurrent_add_directory_requests_serialize_indexing(tmp_path, monkeypatch):
-    """Off-loop execution must not mean parallel index jobs: concurrent
-    requests would race PersonalDocsManager's unsynchronized list mutations
-    and file writes (save_directories/_save_excluded are plain open('w'))."""
-    import time
-
-    state, enter, leave = _serialization_probe()
-
-    def _slow_index(self, directory, owner=None):
-        enter(); time.sleep(0.2); leave()
-        return {"success": True, "indexed_count": 1, "failed_count": 0}
-
-    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
-
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    for name in ("docs_a", "docs_b"):
-        (tmp_path / name).mkdir()
-
-    async with _async_client(app) as ac:
-        results = await asyncio.gather(
-            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
-            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_b")}),
-        )
-
-    assert all(r.status_code == 200 for r in results)
-    assert state["max_active"] == 1, (
-        f"{state['max_active']} index jobs ran in parallel — concurrent "
-        "add_directory requests must serialize"
-    )
-
-
-def test_failed_indexing_still_returns_500(tmp_path, monkeypatch):
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    target = tmp_path / "docs"
-    target.mkdir()
-
-    def _fail(directory, owner=None):
-        return {"success": False, "message": "boom"}
-
-    monkeypatch.setattr(_FakeRag, "index_personal_documents", staticmethod(_fail))
-
-    client = TestClient(app)
-    resp = client.post("/api/personal/add_directory", json={"directory": str(target)})
-    assert resp.status_code == 500
-    assert "boom" in resp.json()["detail"]
-
-
-async def test_add_and_remove_serialize(tmp_path, monkeypatch):
-    """#5634: remove must hold the SAME job lock as add. Otherwise a remove
-    running while an add job is in flight races PersonalDocsManager's
-    unsynchronized list/index mutations — the inconsistent state the PR's
-    'add/remove are serialized' guarantee claims to prevent."""
-    import time
-
-    state, enter, leave = _serialization_probe()
-
-    def _slow_index(self, directory, owner=None):
-        enter(); time.sleep(0.25); leave()
-        return {"success": True, "indexed_count": 1, "failed_count": 0}
-
-    def _slow_remove(self, directory):
-        enter(); time.sleep(0.25); leave()
-
-    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
-    monkeypatch.setattr(_FakeDocsManager, "remove_directory", _slow_remove, raising=False)
-
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    (tmp_path / "docs_a").mkdir()
-    (tmp_path / "docs_b").mkdir()
-
-    async with _async_client(app) as ac:
-        results = await asyncio.gather(
-            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
-            ac.delete("/api/personal/remove_directory", params={"directory": str(tmp_path / "docs_b")}),
-        )
-
-    assert all(r.status_code == 200 for r in results)
-    assert state["max_active"] == 1, (
-        f"{state['max_active']} add/remove critical sections overlapped — "
-        "remove must hold the same index job lock as add"
-    )
-
-
-async def test_reload_serializes_with_add(tmp_path, monkeypatch):
-    """#5634: POST /reload rebuilds the index via refresh_index(); it must hold
-    the same job lock so it cannot race an in-flight add job."""
-    import time
-
-    state, enter, leave = _serialization_probe()
-
-    def _slow_index(self, directory, owner=None):
-        enter(); time.sleep(0.25); leave()
-        return {"success": True, "indexed_count": 1, "failed_count": 0}
-
-    def _slow_refresh(self):
-        enter(); time.sleep(0.25); leave()
-
-    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
-    monkeypatch.setattr(_FakeDocsManager, "refresh_index", _slow_refresh, raising=False)
-
-    record = {}
-    app = _build_app(tmp_path, monkeypatch, record)
-    (tmp_path / "docs_a").mkdir()
-
-    async with _async_client(app) as ac:
-        results = await asyncio.gather(
-            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
-            ac.post("/api/personal/reload"),
-        )
-
-    assert all(r.status_code == 200 for r in results)
-    assert state["max_active"] == 1, (
-        f"{state['max_active']} add/reload critical sections overlapped — "
-        "reload must hold the same index job lock as add"
-    )

tests/test_api_chat_security.py

@@ -76,7 +76,7 @@ def _load_webhook_routes_for_test(monkeypatch):
     module_name = "routes.webhook_routes_under_test"
     spec = importlib.util.spec_from_file_location(
         module_name,
-        Path(__file__).resolve().parent.parent / "routes" / "webhook_routes.py",
+        Path(__file__).resolve().parent.parent / "routes" / "webhook" / "webhook_routes.py",
     )
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)

tests/test_backup_import_cross_user_dedup.py

@@ -27,6 +27,9 @@ def _setup(monkeypatch, store, user="alice"):
 
     mem = MagicMock()
     mem.load_all.return_value = list(store)
+    # import_data reads through the strict loader so a store it cannot read is
+    # never overwritten (#5673); the double has to offer the same entry point.
+    mem.load_all_for_update.return_value = list(store)
     saved = {}
     mem.save.side_effect = lambda entries: saved.__setitem__("entries", entries)
 

tests/test_document_routes_shim.py

@@ -0,0 +1,29 @@
+"""Regression test for the document route shim (slice 2m, #4082/#4071).
+
+The backward-compat shims at ``routes/document_routes.py`` and
+``routes/document_helpers.py`` use ``sys.modules`` replacement so the legacy
+import paths and the canonical ``routes.document.*`` paths resolve to the
+*same* module objects. This is required because multiple tests do
+``import routes.document_routes as droutes`` followed by
+``droutes.SessionLocal = ...`` / ``monkeypatch.setattr(droutes, ...)`` and
+``sys.modules.pop("routes.document_helpers")`` + re-import — for those to
+take effect at runtime, the legacy and canonical module objects must be
+identical.
+"""
+
+import importlib
+
+import routes.document_routes as _shim_routes  # noqa: F401
+import routes.document_helpers as _shim_helpers  # noqa: F401
+
+
+def test_legacy_and_canonical_routes_are_same_object():
+    legacy = importlib.import_module("routes.document_routes")
+    canonical = importlib.import_module("routes.document.document_routes")
+    assert legacy is canonical
+
+
+def test_legacy_and_canonical_helpers_are_same_object():
+    legacy = importlib.import_module("routes.document_helpers")
+    canonical = importlib.import_module("routes.document.document_helpers")
+    assert legacy is canonical

tests/test_email_oauth_connect_smtp_security.py

@@ -0,0 +1,15 @@
+"""Regression coverage for SMTP security saved before Google OAuth."""
+
+from pathlib import Path
+
+
+_REPO = Path(__file__).resolve().parents[1]
+
+
+def test_email_tab_oauth_connect_persists_selected_smtp_security():
+    source = (_REPO / "static" / "js" / "settings.js").read_text(encoding="utf-8")
+    start = source.index("el('eaf-oauth-btn').addEventListener")
+    handler_body = source[start:source.index("if (!body.name)", start)]
+
+    assert "smtp_security: el('eaf-smtp-security').value" in handler_body
+    assert "display_name: el('eaf-display-name').value.trim()" in handler_body

tests/test_email_oauth_settings_redirect.py

@@ -0,0 +1,19 @@
+"""Regression coverage for the settings UI after Google OAuth redirects."""
+
+from pathlib import Path
+
+
+_REPO = Path(__file__).resolve().parents[1]
+
+
+def test_oauth_redirect_uses_the_module_local_settings_api():
+    source = (_REPO / "static" / "js" / "settings.js").read_text(encoding="utf-8")
+    handler = source[
+        source.index("(function _handleOauthRedirect"):
+        source.index("const settingsModule =")
+    ]
+
+    assert "open('integrations');" in handler
+    assert "window.settingsModule" not in handler
+    assert "window.__odysseusAppStarted" not in handler
+    assert "document.addEventListener('DOMContentLoaded', _showResult, { once: true })" in handler

tests/test_imap_mailbox_quoting.py

@@ -87,7 +87,7 @@ def test_known_imap_mailbox_call_sites_are_quoted():
     assert "conn.select(sent_name" not in pollers
     assert "imap.append(sent_folder" not in pollers
 
-    document_routes = Path("routes/document_routes.py").read_text()
+    document_routes = Path("routes/document/document_routes.py").read_text()
     assert "conn.select(doc.source_email_folder" not in document_routes
 
 

tests/test_integration_api_call_ssrf.py

@@ -9,8 +9,13 @@ link-local/metadata is always rejected; RFC-1918/loopback only when
 INTEGRATION_API_BLOCK_PRIVATE_IPS=true (LAN integrations are the primary
 use case, so private stays allowed by default).
 """
+import asyncio
+import ipaddress
+import ssl
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpcore
+import httpx
 import pytest
 
 from src import integrations
@@ -97,3 +102,238 @@ async def test_private_base_url_allowed_by_default_blocked_with_knob(monkeypatch
     assert result["exit_code"] == 1
     assert "rejected" in result["error"].lower()
     client.request.assert_not_called()
+
+
+async def _call_capturing_transport(base_url, path="/items"):
+    """Drive execute_api_call and return (result, transport) where transport is
+    the object passed to httpx.AsyncClient(transport=...)."""
+    resp = MagicMock()
+    resp.status_code = 200
+    resp.headers = {"content-type": "application/json"}
+    resp.json.return_value = {"ok": True}
+    resp.text = '{"ok": true}'
+
+    client = AsyncMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=None)
+    client.request = AsyncMock(return_value=resp)
+
+    captured = {}
+
+    def _fake_async_client(*args, **kwargs):
+        captured.update(kwargs)
+        return client
+
+    with (
+        patch.object(integrations, "_find_integration",
+                     return_value=_integration(base_url)),
+        patch("httpx.AsyncClient", side_effect=_fake_async_client),
+    ):
+        result = await integrations.execute_api_call("test_integ", "GET", path)
+    return result, captured.get("transport"), client
+
+
+@pytest.mark.asyncio
+async def test_connection_is_pinned_to_the_validated_ip(monkeypatch):
+    """DNS-rebinding defense: the guard resolves the host once to a benign
+    public IP, and the request must be pinned to *that* IP so a host that
+    rebinds to the metadata range at connect time can't be reached with the
+    integration's auth headers. Static resolution passing the guard is not
+    enough — a plain client would re-resolve at connect."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["93.184.216.34"])
+    result, transport, client = await _call_capturing_transport(
+        "http://rebinding.attacker.example")
+
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34"]
+
+
+@pytest.mark.asyncio
+async def test_pin_carries_the_whole_validated_ip_set(monkeypatch):
+    """When a host resolves to several records the transport keeps all of them
+    (check_outbound_url validated every one), in resolver order, so it can fall
+    back past a dead first address instead of failing the whole call."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["93.184.216.34", "198.51.100.7"])
+    result, transport, _ = await _call_capturing_transport("http://multi.example")
+
+    assert result.get("exit_code") == 0
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34", "198.51.100.7"]
+
+
+class _FakeStream:
+    """Stand-in for the connected socket the real backend returns."""
+
+
+class _RecordingBackend:
+    """Fake httpcore backend: connect_tcp fails for the addresses in `dead`
+    and succeeds for the rest, recording the order it was asked to connect."""
+
+    def __init__(self, dead):
+        self.dead = set(dead)
+        self.attempts = []
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None,
+                          socket_options=None):
+        self.attempts.append((host, timeout))
+        if host in self.dead:
+            raise httpcore.ConnectError(f"connection refused: {host}")
+        return _FakeStream()
+
+
+def _pinned_backend(ips, dead):
+    """A _PinnedAsyncBackend whose underlying connect is the recording fake."""
+    backend = integrations._PinnedAsyncBackend(ips)
+    backend._real = _RecordingBackend(dead)
+    return backend
+
+
+@pytest.mark.asyncio
+async def test_connect_falls_back_from_dead_first_to_live_second():
+    """first-dead / second-live: the pinned backend must try the next validated
+    address when the first refuses, rather than surfacing the failure. It also
+    ignores the `host` httpcore passes (the original hostname) and connects to
+    the pinned IPs, which is what keeps TLS SNI / Host on the real hostname."""
+    ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")]
+    backend = _pinned_backend(ips, dead={"203.0.113.10"})
+
+    stream = await backend.connect_tcp("original.hostname.example", 443, timeout=5.0)
+
+    assert isinstance(stream, _FakeStream)
+    # Tried the dead address first, then the live one — never the hostname.
+    assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"]
+    # Fallback shared one budget: the second attempt got the time left, not a fresh 5s.
+    assert backend._real.attempts[1][1] <= 5.0
+
+
+@pytest.mark.asyncio
+async def test_connect_raises_when_every_validated_address_is_dead():
+    ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")]
+    backend = _pinned_backend(ips, dead={"203.0.113.10", "198.51.100.7"})
+
+    with pytest.raises(httpcore.ConnectError):
+        await backend.connect_tcp("original.hostname.example", 443, timeout=5.0)
+    assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"]
+
+
+@pytest.mark.asyncio
+async def test_pinned_transport_reuses_httpx_ca_trust(monkeypatch):
+    """TLS trust must come from the same builder the default httpx client uses
+    (certifi + SSL_CERT_FILE / SSL_CERT_DIR via trust_env), not from
+    ssl.create_default_context()'s system roots — otherwise chains that verified
+    under the old default client can silently stop verifying."""
+    sentinel = ssl.create_default_context()
+    calls = []
+
+    def _fake_create(*args, **kwargs):
+        calls.append(kwargs)
+        return sentinel
+
+    monkeypatch.setattr(httpx, "create_ssl_context", _fake_create)
+    transport = integrations._PinnedAsyncTransport([ipaddress.ip_address("93.184.216.34")])
+    try:
+        assert calls, "transport did not build its context via httpx.create_ssl_context"
+        assert transport._pool._ssl_context is sentinel
+    finally:
+        await transport.aclose()
+
+
+@pytest.mark.asyncio
+async def test_real_socket_falls_back_from_dead_first_to_live_second():
+    """End-to-end over real loopback sockets: pin [127.0.0.2 (nothing
+    listening), 127.0.0.1 (live)], and the request must succeed by falling back
+    to the second address while the Host header stays the original hostname —
+    i.e. only the socket destination moved, vhost/SNI routing did not."""
+    captured = {}
+
+    async def handle(reader, writer):
+        request = await reader.read(4096)
+        for line in request.split(b"\r\n"):
+            if line.lower().startswith(b"host:"):
+                captured["host"] = line.split(b":", 1)[1].strip().decode()
+        writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi")
+        await writer.drain()
+        writer.close()
+
+    server = await asyncio.start_server(handle, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    async with server:
+        await server.start_serving()
+        transport = integrations._PinnedAsyncTransport(
+            [ipaddress.ip_address("127.0.0.2"), ipaddress.ip_address("127.0.0.1")]
+        )
+        try:
+            async with httpx.AsyncClient(transport=transport) as client:
+                resp = await client.get(f"http://pinned.example:{port}/health")
+        finally:
+            await transport.aclose()
+
+    assert resp.status_code == 200
+    assert resp.text == "hi"
+    assert captured.get("host") == f"pinned.example:{port}"
+
+
+@pytest.mark.asyncio
+async def test_ip_literal_base_url_still_pins_and_is_not_rejected():
+    """A base_url that is already an IP has nothing to rebind, but it must not
+    trip the "did not resolve" guard either.
+
+    check_outbound_url resolves even a literal (getaddrinfo returns the address
+    itself), so the captured list is populated and the pin is a no-op rather
+    than a rejection. Uses the real resolver on purpose — no monkeypatch — so
+    this would catch the fail-closed branch firing on a literal.
+    """
+    result, transport, client = await _call_capturing_transport(
+        "http://93.184.216.34")
+
+    assert result.get("exit_code") == 0
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34"]
+
+
+@pytest.mark.asyncio
+async def test_ipv6_base_url_pins_every_validated_address(monkeypatch):
+    """IPv6 goes down the same path as v4.
+
+    Resolution is stubbed rather than using a literal so this doesn't depend on
+    the runner having IPv6 configured.
+    """
+    v6 = "2606:2800:220:1:248:1893:25c8:1946"
+    monkeypatch.setattr("src.url_safety._default_resolver", lambda host: [v6])
+    result, transport, client = await _call_capturing_transport("http://v6.example")
+
+    assert result.get("exit_code") == 0
+    assert isinstance(transport, integrations._PinnedAsyncTransport)
+    assert [str(ip) for ip in transport._pinned_ips] == [v6]
+
+
+def test_validated_ips_strips_zone_id_and_drops_junk():
+    """getaddrinfo can hand back a scoped v6 address like 'fe80::1%eth0'."""
+    got = integrations._validated_ips(
+        ["93.184.216.34", "fe80::1%eth0", "not-an-ip", None, "2001:db8::5"]
+    )
+    assert [str(ip) for ip in got] == ["93.184.216.34", "fe80::1", "2001:db8::5"]
+
+
+def test_validated_ips_deduplicates_repeated_addresses():
+    """The resolver is getaddrinfo(host, None) with no socktype filter, so glibc
+    returns one record per socktype and a single-homed host arrives three times
+    over. Duplicates must collapse (first-seen order kept) or the connect
+    fallback wastes its shared deadline retrying one dead address."""
+    got = integrations._validated_ips(
+        ["93.184.216.34", "93.184.216.34", "93.184.216.34"]
+    )
+    assert [str(ip) for ip in got] == ["93.184.216.34"]
+
+    # Order is first-seen, and distinct addresses all survive.
+    got = integrations._validated_ips(
+        ["198.51.100.7", "93.184.216.34", "198.51.100.7", "2001:db8::5"]
+    )
+    assert [str(ip) for ip in got] == ["198.51.100.7", "93.184.216.34", "2001:db8::5"]
+
+    # A zone-id variant is the same address once stripped, so it collapses too.
+    got = integrations._validated_ips(["fe80::1%eth0", "fe80::1%eth1", "fe80::1"])
+    assert [str(ip) for ip in got] == ["fe80::1"]

tests/test_integrations_api_call_truncation.py

@@ -83,9 +83,10 @@ async def _call(json_data, status=200):
     with (
         patch.object(integrations, "_find_integration", return_value=DUMMY_INTEGRATION),
         patch("httpx.AsyncClient", return_value=mock_client),
-        # api.example.com doesn't resolve; the SSRF guard would fail closed.
-        # These tests are about truncation, so stub the guard open.
-        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
+        # api.example.com doesn't resolve. Point the resolver at a public
+        # address instead of stubbing the guard open, so the real check (and
+        # the connect-IP pinning that reads its result) still runs.
+        patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]),
     ):
         return await integrations.execute_api_call("test_integ", "GET", "/items")
 
@@ -101,9 +102,10 @@ async def _call_with_integration(integration, path="/items"):
     with (
         patch.object(integrations, "_find_integration", return_value=integration),
         patch("httpx.AsyncClient", return_value=mock_client),
-        # api.example.com doesn't resolve; the SSRF guard would fail closed.
-        # These tests are about URL joining, so stub the guard open.
-        patch("src.url_safety.check_outbound_url", return_value=(True, "ok")),
+        # api.example.com doesn't resolve. Point the resolver at a public
+        # address instead of stubbing the guard open, so the real check (and
+        # the connect-IP pinning that reads its result) still runs.
+        patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]),
     ):
         result = await integrations.execute_api_call("test_integ", "GET", path)
     return result, mock_client

tests/test_issue_description_check.py

@@ -0,0 +1,86 @@
+"""Regression coverage for issue-description label lifecycle events."""
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+
+_REPO = Path(__file__).resolve().parent.parent
+_CHECKER = _REPO / ".github" / "scripts" / "check-issue-description.js"
+_WORKFLOW = _REPO / ".github" / "workflows" / "issue-description-check.yml"
+pytestmark = pytest.mark.skipif(not shutil.which("node"), reason="node not on PATH")
+
+
+def _run_closed_issue(action):
+    harness = r"""
+const checkIssueDescription = require(process.argv[1]);
+const action = process.argv[2];
+const calls = [];
+const unexpected = (name) => async () => {
+  throw new Error(`${name} should not be called for a closed issue`);
+};
+
+const github = {
+  rest: {
+    issues: {
+      removeLabel: async (params) => calls.push({ method: 'removeLabel', params }),
+      getLabel: unexpected('getLabel'),
+      addLabels: unexpected('addLabels'),
+      listComments: unexpected('listComments'),
+      createComment: unexpected('createComment'),
+      updateComment: unexpected('updateComment'),
+      deleteComment: unexpected('deleteComment'),
+    },
+  },
+};
+const context = {
+  payload: {
+    action,
+    issue: { number: 42, state: 'closed', body: '', labels: [] },
+  },
+  repo: { owner: 'odysseus-dev', repo: 'odysseus' },
+};
+const core = {
+  warning: unexpected('core.warning'),
+  setFailed: unexpected('core.setFailed'),
+};
+
+checkIssueDescription({ github, context, core })
+  .then(() => process.stdout.write(JSON.stringify(calls)))
+  .catch((error) => {
+    console.error(error);
+    process.exitCode = 1;
+  });
+"""
+    proc = subprocess.run(
+        ["node", "-e", harness, str(_CHECKER), action],
+        capture_output=True,
+        text=True,
+        cwd=str(_REPO),
+        timeout=30,
+    )
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def test_workflow_handles_issue_closures():
+    workflow = _WORKFLOW.read_text()
+    assert "types: [opened, edited, reopened, closed]" in workflow
+
+
+@pytest.mark.parametrize("action", ["closed", "edited"])
+def test_closed_issue_only_drops_ready_for_review(action):
+    assert _run_closed_issue(action) == [
+        {
+            "method": "removeLabel",
+            "params": {
+                "owner": "odysseus-dev",
+                "repo": "odysseus",
+                "issue_number": 42,
+                "name": "ready for review",
+            },
+        }
+    ]

tests/test_llm_core_anthropic_temp_omit.py

@@ -29,6 +29,13 @@ from src.llm_core import _anthropic_rejects_temperature, _build_anthropic_payloa
         "anthropic/claude-opus-4-7",  # tolerate a provider-prefixed id
         "claude-opus-4-10",  # future minor still >= 4.7
         "claude-opus-5-0",  # future major
+        # Major-only ids: a missing minor reads as `.0`, so these are >= 4.7 too
+        # (issue #5753). Before the fix the version pattern required a minor, so
+        # these fell through to "accepts temperature" and every call 400'd.
+        "claude-opus-5",
+        "claude-opus-5-20260101",  # major-only + dated snapshot
+        "anthropic/claude-opus-5",  # major-only behind a provider prefix
+        "claude-opus-6",  # future major-only
     ],
 )
 def test_opus_47_plus_rejects_temperature(model):
@@ -48,7 +55,10 @@ def test_opus_47_plus_rejects_temperature(model):
         "claude-opus-4-6-20251201",  # dated 4.6 snapshot — older, still keeps temperature
         "claude-sonnet-4-6",
         "claude-3-5-sonnet",
-        "claude-3-opus-20240229",  # legacy Claude 3 Opus — no opus-N-M pattern, kept
+        "claude-3-opus-20240229",  # legacy Claude 3 Opus — date directly after
+        # "opus-", so the major must not swallow it as version 20240229 (that is
+        # what makes capping the major at 1-2 digits necessary once the minor
+        # became optional in #5753).
         "claude-haiku-4-5",
         "claude-x",
         "octopus-4-8",  # "opus" only as a substring of another word — must not match
@@ -87,6 +97,20 @@ def test_payload_keeps_temperature_for_older_models():
     assert _payload("claude-3-5-sonnet", 1.2)["temperature"] == 1.0
 
 
+def test_payload_omits_temperature_for_major_only_opus_5():
+    # Issue #5753: the scheduled-task path calls stream_agent_loop() without a
+    # temperature and inherits its 0.3 default, so `claude-opus-5` 400'd on every
+    # run and surfaced as "the model returned an empty response". Interactive chat
+    # leaves temperature None and never hit it.
+    assert "temperature" not in _payload("claude-opus-5", 0.3)
+
+
+def test_payload_keeps_temperature_for_legacy_claude_3_opus():
+    # Guards the major-digit cap: `opus-20240229` must not parse as version
+    # 20240229, or Claude 3 Opus would silently lose the caller's temperature.
+    assert _payload("claude-3-opus-20240229", 0.5)["temperature"] == 0.5
+
+
 def test_payload_keeps_temperature_for_dated_opus_4_0():
     # Anthropic's dated id for Opus 4.0 (claude-opus-4-20250514) is in this repo's
     # ANTHROPIC_MODELS list. The date must not be misread as a >= 4.7 minor, or the

tests/test_manage_skills_action_required.py

@@ -0,0 +1,24 @@
+import json
+
+import pytest
+
+from src.tools.system import do_manage_skills
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {},
+        {"action": ""},
+        {"action": "   "},
+        {"name": "demo", "description": "x", "procedure": ["step"]},
+    ],
+)
+async def test_manage_skills_requires_action(payload):
+    result = await do_manage_skills(json.dumps(payload), owner="test")
+
+    assert result == {
+        "error": "action is required (list|view|view_ref|add|edit|patch|publish|delete|search)",
+        "exit_code": 1,
+    }

tests/test_markdown_rendering_js.py

@@ -214,6 +214,50 @@ def test_inline_code_content_is_html_escaped(node_available):
     assert "<b>" not in html
 
 
+def test_fenced_code_keeps_dollar_ampersand(node_available):
+    # Issue #5663: the block-restore pass used a string replacement, so `$&` in a
+    # restored block was read as "the matched text" and re-inserted the
+    # placeholder. `perl -pe 's/world/$& again/'` rendered as
+    # "s/world/___CODE_BLOCK_0___amp; again/" — the trailing "amp;" is the orphan
+    # left behind after `$&` consumed the `$&` of the escaped `$&amp;`.
+    html = _run_markdown_case(
+        "```sh\necho \"hello world\" | perl -pe 's/world/$& again/'\n```"
+    )
+
+    assert "___CODE_BLOCK_" not in html
+    assert "s/world/$&amp; again/" in html
+    assert "amp; again" not in html.replace("$&amp; again", "")
+
+
+def test_fenced_code_keeps_dollar_backtick_and_quote(node_available):
+    # `` $` `` and `$'` splice the text before/after the placeholder into the
+    # block. Unlike `$&` these leave no placeholder behind — the characters just
+    # vanish — so assert the content survives verbatim.
+    html = _run_markdown_case("```sh\nsed \"s/$`/x/\" && sed \"s/$'/y/\"\n```")
+
+    assert "___CODE_BLOCK_" not in html
+    assert "s/$`/x/" in html
+    assert "s/$&#39;/y/" in html
+
+
+def test_fenced_code_keeps_double_dollar(node_available):
+    # `$$` collapsed to a single `$` in the restored block.
+    html = _run_markdown_case('```sh\necho "$$USD and $$"\n```')
+
+    assert "$$USD and $$" in html
+
+
+def test_mermaid_block_keeps_dollar_ampersand(node_available):
+    # The mermaid restore site had the same hazard: a node label containing `$&`
+    # re-inserted the ___MERMAID_BLOCK_n___ placeholder into the diagram source,
+    # which then fails to parse. The math and allowed-HTML sites are fixed the
+    # same way; they need KaTeX/sanitizer conditions this harness doesn't set up.
+    html = _run_markdown_case('```mermaid\ngraph TD; A["$&"] --> B;\n```')
+
+    assert "___MERMAID_BLOCK_" not in html
+    assert "$&amp;" in html
+
+
 def test_currency_dollar_amounts_are_not_rendered_as_math(node_available):
     # "$5 to $10" used to pair the two dollar signs as inline-math delimiters
     # and render "5 to" through KaTeX. Pandoc-style rules now reject it: the

tests/test_mcp_dependency_compatibility.py

@@ -0,0 +1,15 @@
+"""Regression coverage for the built-in MCP servers' SDK compatibility line."""
+
+from pathlib import Path
+
+
+REQUIREMENTS = Path(__file__).resolve().parents[1] / "requirements.txt"
+
+
+def test_mcp_requirement_excludes_breaking_v2_sdk():
+    requirements = [
+        line.split("#", 1)[0].strip().replace(" ", "")
+        for line in REQUIREMENTS.read_text(encoding="utf-8").splitlines()
+    ]
+
+    assert "mcp<2" in requirements

tests/test_memory_extractor_vector_cross_tenant.py

@@ -67,6 +67,12 @@ class FakeMemoryManager:
     def load_all(self):
         return list(self.rows)
 
+    def load_all_for_update(self):
+        # Mirrors the real MemoryManager: extraction is a read-modify-write and
+        # goes through the strict loader (#5673). A healthy store behaves the
+        # same as load_all.
+        return list(self.rows)
+
     def load(self, owner=None):
         return [r for r in self.rows if r.get("owner") == owner]
 

tests/test_memory_store_unreadable_no_wipe.py

@@ -0,0 +1,255 @@
+"""A memory store that cannot be READ must never be overwritten (issue #5673).
+
+`MemoryManager.save` is atomic, and the add/import/extract paths are all
+read-modify-write: load the whole store, append, save it back. `load_all`
+used to answer a *failed read* with `[]` — indistinguishable from "no
+memories" — so a failed read turned into
+
+    load_all() -> []  ->  [].append(new)  ->  save([new])
+
+which atomically replaced the entire store with one entry.
+
+The trigger that actually bites is a store that is **readable but not
+parseable** — a truncated file, or one holding `{}` instead of `[]`. Nothing
+obstructs the write, so the request succeeds with HTTP 200 and every existing
+memory is destroyed silently. Truncation is reachable: `core/database.py`
+rewrites memory.json during migration with a plain `open(..., "w")` +
+`json.dump`, which is not atomic.
+
+A live exclusive lock is NOT the dangerous case: it blocks the read and the
+`os.replace` alike, so the save fails too and the store survives (verified
+end-to-end — clean dev returns 500 there and loses nothing).
+
+`load_all_for_update` is the strict loader those callers now use: it raises
+`MemoryStoreUnreadable` rather than reporting an empty store.
+"""
+
+import asyncio
+import builtins
+import json
+import os
+
+import pytest
+
+from src.memory import MemoryManager, MemoryStoreUnreadable
+
+_SEED = [
+    {"id": "m1", "text": "user prefers dark mode", "owner": "alice"},
+    {"id": "m2", "text": "user lives in Berlin", "owner": "alice"},
+    {"id": "m3", "text": "bob's cat is called Mila", "owner": "bob"},
+]
+
+
+def _seeded(tmp_path):
+    m = MemoryManager(str(tmp_path))
+    m.save([dict(e) for e in _SEED])
+    return m
+
+
+def _break_reads_of(monkeypatch, target, exc):
+    """Make open() raise `exc` for `target` only, leaving every other path alone."""
+    real_open = builtins.open
+
+    def fake_open(file, mode="r", *args, **kwargs):
+        if os.path.abspath(str(file)) == os.path.abspath(target) and "r" in mode:
+            raise exc
+        return real_open(file, mode, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "open", fake_open)
+
+
+# ── the strict loader signals, rather than reporting "empty" ──────────────
+
+def test_strict_load_raises_on_permission_error(tmp_path, monkeypatch):
+    m = _seeded(tmp_path)
+    _break_reads_of(monkeypatch, m.memory_file, PermissionError(13, "locked"))
+    with pytest.raises(MemoryStoreUnreadable):
+        m.load_all_for_update()
+
+
+def test_strict_load_raises_on_corrupt_json(tmp_path):
+    m = _seeded(tmp_path)
+    with open(m.memory_file, "w", encoding="utf-8") as f:
+        f.write('[{"id": "m1", "text": "truncated mid-writ')
+    with pytest.raises(MemoryStoreUnreadable):
+        m.load_all_for_update()
+
+
+def test_strict_load_raises_when_store_is_not_a_list(tmp_path):
+    # A file holding `{}` or `null` is not an empty store, it is a broken one.
+    m = _seeded(tmp_path)
+    with open(m.memory_file, "w", encoding="utf-8") as f:
+        json.dump({}, f)
+    with pytest.raises(MemoryStoreUnreadable):
+        m.load_all_for_update()
+
+
+def test_strict_load_returns_entries_when_healthy(tmp_path):
+    m = _seeded(tmp_path)
+    assert {e["id"] for e in m.load_all_for_update()} == {"m1", "m2", "m3"}
+
+
+def test_strict_load_returns_empty_when_file_genuinely_absent(tmp_path):
+    m = _seeded(tmp_path)
+    os.remove(m.memory_file)
+    # Absent is the one case that legitimately means "no memories yet".
+    assert m.load_all_for_update() == []
+
+
+# ── read paths stay lenient, so an unreadable store can't break chat ──────
+
+def test_read_path_still_degrades_to_empty(tmp_path, monkeypatch):
+    m = _seeded(tmp_path)
+    _break_reads_of(monkeypatch, m.memory_file, PermissionError(13, "locked"))
+    # Context injection / search must not raise; they just see nothing.
+    assert m.load_all() == []
+    assert m.load(owner="alice") == []
+
+
+# ── the actual #5673 regression: the store survives ───────────────────────
+
+def test_add_cycle_under_transient_read_error_does_not_wipe(tmp_path, monkeypatch):
+    """Mirrors routes/memory/memory_routes.py api_add_memory exactly."""
+    m = _seeded(tmp_path)
+    new_entry = m.add_entry("a brand new fact", owner="alice")
+
+    with monkeypatch.context() as mp:
+        _break_reads_of(mp, m.memory_file, PermissionError(13, "locked"))
+        with pytest.raises(MemoryStoreUnreadable):
+            all_mem = m.load_all_for_update()
+            all_mem.append(new_entry)
+            m.save(all_mem)
+
+    # Reads work again; every original memory is still there and the file was
+    # never replaced by the single new entry.
+    assert {e["id"] for e in m.load_all()} == {"m1", "m2", "m3"}
+
+
+def test_audit_merge_cannot_drop_other_tenants(tmp_path, monkeypatch):
+    """The audit path rebuilds the whole file from load_all + one owner's slice.
+
+    Reading [] there would save only the audited owner's entries and destroy
+    every other tenant's memories, so it has to fail closed too.
+    """
+    m = _seeded(tmp_path)
+    alice_slice = [e for e in _SEED if e["owner"] == "alice"]
+
+    with monkeypatch.context() as mp:
+        _break_reads_of(mp, m.memory_file, PermissionError(13, "locked"))
+        with pytest.raises(MemoryStoreUnreadable):
+            all_entries = m.load_all_for_update()
+            others = [e for e in all_entries if e.get("owner") != "alice"]
+            m.save(alice_slice + others)
+
+    assert any(e["id"] == "m3" for e in m.load_all()), "bob's memory was destroyed"
+
+
+def test_uses_bump_skips_write_when_unreadable(tmp_path, monkeypatch):
+    m = _seeded(tmp_path)
+    with monkeypatch.context() as mp:
+        _break_reads_of(mp, m.memory_file, PermissionError(13, "locked"))
+        m.increment_uses(["m1"])  # must not raise, must not write
+    assert {e["id"] for e in m.load_all()} == {"m1", "m2", "m3"}
+
+
+def test_claim_ownerless_skips_write_when_unreadable(tmp_path, monkeypatch):
+    m = _seeded(tmp_path)
+    with monkeypatch.context() as mp:
+        _break_reads_of(mp, m.memory_file, PermissionError(13, "locked"))
+        m.claim_ownerless("alice")
+    assert {e["id"] for e in m.load_all()} == {"m1", "m2", "m3"}
+
+
+# ── the add sinks users actually reach ────────────────────────────────────
+#
+# The tests above replay the read-modify-write shape. These drive the real
+# entry points end to end, because those are what #5673 reports: "remember
+# that I prefer X" in ordinary chat (src/ai_interaction.py do_manage_memory,
+# routed from src/tool_execution.py) and the built-in memory MCP server
+# (mcp_servers/memory_server.py, registered in src/builtin_mcp.py).
+#
+# They use a truncated store rather than a read error on purpose: it reads
+# fine, so nothing stops the save, which is the case that silently destroyed
+# stores. The assertion is that the file is left byte-identical — still broken,
+# but still holding the user's memories, so it can be repaired by hand.
+
+
+def _truncated_store(tmp_path):
+    """Seed a store that reads back fine but no longer parses."""
+    m = _seeded(tmp_path)
+    good = json.dumps([dict(e) for e in _SEED], indent=2)
+    with open(m.memory_file, "w", encoding="utf-8") as f:
+        f.write(good[:good.rindex("]")])  # drop the closing bracket only
+    with open(m.memory_file, "rb") as f:
+        return m, f.read()
+
+
+def _on_disk(manager) -> bytes:
+    with open(manager.memory_file, "rb") as f:
+        return f.read()
+
+
+def test_agent_memory_add_does_not_overwrite_unreadable_store(tmp_path, monkeypatch):
+    """src/ai_interaction.py do_manage_memory, action "add"."""
+    from src import ai_interaction
+
+    manager, before = _truncated_store(tmp_path)
+    monkeypatch.setattr(ai_interaction, "_memory_manager", manager)
+    monkeypatch.setattr(ai_interaction, "_memory_vector", None)
+
+    result = asyncio.run(ai_interaction.do_manage_memory("add\nuser prefers tabs"))
+
+    assert _on_disk(manager) == before, "the unreadable store was overwritten"
+    assert b"m3" in _on_disk(manager)
+    assert "error" in result, "the add reported success over an unreadable store"
+
+
+def test_mcp_memory_add_does_not_overwrite_unreadable_store(tmp_path, monkeypatch):
+    """mcp_servers/memory_server.py, action "add"."""
+    import mcp_servers.memory_server as memory_server
+
+    manager, before = _truncated_store(tmp_path)
+    monkeypatch.setattr(memory_server, "_memory_manager", manager)
+    monkeypatch.setattr(memory_server, "_memory_vector", None)
+    monkeypatch.setattr(memory_server, "_initialized", True)
+    for key in memory_server._OWNER_ENV_KEYS:
+        monkeypatch.delenv(key, raising=False)
+
+    result = asyncio.run(memory_server.call_tool(
+        "manage_memory", {"action": "add", "text": "user prefers tabs"}
+    ))
+
+    assert _on_disk(manager) == before, "the unreadable store was overwritten"
+    assert b"m3" in _on_disk(manager)
+    assert result[0].text.startswith("Error:")
+
+
+def test_native_provider_remember_does_not_overwrite_unreadable_store(tmp_path):
+    """src/memory_provider.py NativeMemoryProvider.remember.
+
+    Registered into app state in src/app_initializer.py but not yet consumed
+    outside tests, so this is the pattern held in place before it goes live.
+    """
+    from src.memory_provider import NativeMemoryProvider
+
+    manager, before = _truncated_store(tmp_path)
+    provider = NativeMemoryProvider(manager)
+
+    with pytest.raises(MemoryStoreUnreadable):
+        asyncio.run(provider.remember("user prefers tabs", owner="alice"))
+
+    assert _on_disk(manager) == before
+
+
+# ── the legacy memory.txt migration is preserved ──────────────────────────
+
+def test_corrupt_store_still_migrates_from_legacy_txt(tmp_path):
+    m = _seeded(tmp_path)
+    with open(m.memory_file, "w", encoding="utf-8") as f:
+        f.write("{ not json")
+    legacy = os.path.join(str(tmp_path), "memory.txt")
+    with open(legacy, "w", encoding="utf-8") as f:
+        f.write("recovered fact one\nrecovered fact two\n")
+
+    entries = m.load_all_for_update()
+    assert [e["text"] for e in entries] == ["recovered fact one", "recovered fact two"]

tests/test_model_helper_owner_scope.py

@@ -14,7 +14,7 @@ def _function_source(path: str, name: str) -> str:
 
 
 def test_document_ai_tidy_resolves_with_owner_scope():
-    body = _function_source("routes/document_routes.py", "ai_tidy_documents")
+    body = _function_source("routes/document/document_routes.py", "ai_tidy_documents")
     assert "resolve_task_endpoint(owner=user or None)" in body
     assert 'resolve_endpoint("default", owner=user or None)' in body
 

tests/test_search_routes_shim.py

@@ -0,0 +1,11 @@
+"""Regression test for the search route shim (slice 2j, #4082/#4071)."""
+
+import importlib
+
+import routes.search_routes as _shim_search  # noqa: F401
+
+
+def test_legacy_and_canonical_search_module_are_same_object():
+    legacy = importlib.import_module("routes.search_routes")
+    canonical = importlib.import_module("routes.search.search_routes")
+    assert legacy is canonical

tests/test_skill_format_timestamp.py

@@ -0,0 +1,58 @@
+"""Regression for issue #5697 — skill timestamps must not use ``datetime.utcnow()``.
+
+``_now_iso()`` builds the ``created`` value in skill frontmatter. ``utcnow()``
+returns a *naive* datetime and has been deprecated since Python 3.12, scheduled
+for removal. The replacement must stay timezone-aware while keeping the
+serialized ``YYYY-MM-DDTHH:MM:SSZ`` shape, so skill files written by older
+versions keep parsing.
+
+The UTC check matters on its own: a bare ``datetime.now()`` also produces the
+right shape, but emits local wall time, which would silently backdate or
+postdate skills for every user outside UTC.
+"""
+
+import os
+import re
+import time
+import warnings
+from datetime import datetime, timezone
+
+import pytest
+
+from services.memory.skill_format import _now_iso
+
+_ISO_Z = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
+
+
+def test_now_iso_keeps_serialized_shape():
+    assert _ISO_Z.match(_now_iso())
+
+
+def test_now_iso_emits_no_deprecation_warning():
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        _now_iso()
+    assert not [w for w in caught if issubclass(w.category, DeprecationWarning)]
+
+
+@pytest.mark.skipif(
+    not hasattr(time, "tzset"),
+    reason="time.tzset is unavailable on this platform",
+)
+def test_now_iso_is_utc_not_local_time():
+    """Pin UTC under a non-UTC local timezone, where the two visibly diverge."""
+    original_tz = os.environ.get("TZ")
+    os.environ["TZ"] = "Asia/Amman"  # UTC+3, never UTC
+    time.tzset()
+    try:
+        emitted = datetime.strptime(_now_iso(), "%Y-%m-%dT%H:%M:%SZ").replace(
+            tzinfo=timezone.utc
+        )
+        drift = abs((emitted - datetime.now(timezone.utc)).total_seconds())
+        assert drift < 60, f"timestamp is {drift}s off UTC — local time leaked in"
+    finally:
+        if original_tz is None:
+            os.environ.pop("TZ", None)
+        else:
+            os.environ["TZ"] = original_tz
+        time.tzset()

tests/test_tts_service_enforce_cache_limit.py

@@ -0,0 +1,97 @@
+import os
+import time
+from pathlib import Path
+import pytest
+
+# Adjust the import path if your file is directly in ./services instead of ./services/tts
+from services.tts.tts_service import TTSService
+
+def test_cache_under_limit(tmp_path, monkeypatch):
+    """Test that writing a file under the size limit does not trigger eviction."""
+    # Set a tiny limit: 100 bytes
+    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "100")
+    
+    # Initialize service with pytest's temporary directory
+    service = TTSService(cache_dir=str(tmp_path))
+    
+    # Write a 40-byte file (under the 100-byte limit)
+    service._put_cache("test_key", b"x" * 40)
+    
+    # Verify the file was written and nothing was deleted
+    files = list(tmp_path.glob("*.*"))
+    assert len(files) == 1
+    assert sum(f.stat().st_size for f in files) == 40
+
+def test_cache_exceeds_limit_triggers_eviction(tmp_path, monkeypatch):
+    """Test that exceeding the limit evicts the oldest files down to 80% capacity."""
+    # Set limit to 100 bytes. 80% target capacity will be 80 bytes.
+    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "100")
+    service = TTSService(cache_dir=str(tmp_path))
+    
+    # 1. Setup: Manually create two older files (40 bytes each)
+    file1 = tmp_path / "oldest.wav"
+    file2 = tmp_path / "middle.wav"
+    
+    file1.write_bytes(b"a" * 40)
+    file2.write_bytes(b"b" * 40)
+    
+    # Spoof timestamps so file1 is explicitly older than file2
+    now = time.time()
+    os.utime(file1, (now - 100, now - 100)) # 100 seconds ago
+    os.utime(file2, (now - 50, now - 50))   # 50 seconds ago
+    
+    # 2. Action: Write a 3rd file using the service method (40 bytes)
+    # Total cache is now 120 bytes, which exceeds 100.
+    # It should delete oldest (file1) to drop to 80 bytes (which matches the 80% target).
+    service._put_cache("newest", b"c" * 40)
+    
+    # 3. Assertions
+    # The newest file should exist (saved as .wav because it lacks MP3 magic bytes)
+    newest_file = tmp_path / "newest.wav"
+    
+    assert not file1.exists(), "The oldest file should have been evicted."
+    assert file2.exists(), "The middle file should still exist."
+    assert newest_file.exists(), "The newest file should have been saved."
+    
+    # Verify the final directory size is <= 80 bytes
+    total_size = sum(f.stat().st_size for f in tmp_path.glob("*.*"))
+    assert total_size <= 80
+
+def test_cache_limit_disabled(tmp_path, monkeypatch):
+    """Test that setting max bytes to 0 disables eviction."""
+    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "0")
+    service = TTSService(cache_dir=str(tmp_path))
+    
+    # Write 3 large files that would normally trigger eviction
+    service._put_cache("file1", b"x" * 1000)
+    service._put_cache("file2", b"x" * 1000)
+    service._put_cache("file3", b"x" * 1000)
+    
+    # Ensure nothing was deleted
+    files = list(tmp_path.glob("*.*"))
+    assert len(files) == 3
+    assert sum(f.stat().st_size for f in files) == 3000
+
+def test_cache_eviction_handles_unlink_error_gracefully(tmp_path, monkeypatch):
+    """Test that if unlinking a file fails, _put_cache still succeeds without raising."""
+    service = TTSService(cache_dir=str(tmp_path))
+    service.max_cache_bytes = 50
+
+    # Create a file to evict
+    old_file = tmp_path / "old.wav"
+    old_file.write_bytes(b"x" * 40)
+
+    # Monkeypatch unlink on Path objects to simulate a PermissionError / file-lock failure
+    def mock_unlink(self_path):
+        raise OSError("Permission denied / file locked")
+
+    monkeypatch.setattr(Path, "unlink", mock_unlink)
+
+    # Writing a new file triggers eviction which encounters the mocked unlink error
+    try:
+        service._put_cache("new_key", b"y" * 40)
+    except Exception as e:
+        pytest.fail(f"_put_cache raised an exception during failed eviction: {e}")
+
+    # The new file should still be written successfully
+    assert (tmp_path / "new_key.wav").exists()

tests/test_vault_routes_shim.py

@@ -0,0 +1,11 @@
+"""Regression test for the vault route shim (slice 2k, #4082/#4071)."""
+
+import importlib
+
+import routes.vault_routes as _shim_vault  # noqa: F401
+
+
+def test_legacy_and_canonical_vault_module_are_same_object():
+    legacy = importlib.import_module("routes.vault_routes")
+    canonical = importlib.import_module("routes.vault.vault_routes")
+    assert legacy is canonical

tests/test_vision_owner_scope.py

@@ -88,7 +88,7 @@ def test_request_vision_call_sites_pass_owner():
     chat_source = (ROOT / "src" / "chat_handler.py").read_text()
     processor_source = (ROOT / "src" / "document_processor.py").read_text()
     upload_source = (ROOT / "routes" / "upload_routes.py").read_text()
-    document_source = (ROOT / "routes" / "document_routes.py").read_text()
+    document_source = (ROOT / "routes" / "document" / "document_routes.py").read_text()
     gallery_source = (ROOT / "routes" / "gallery" / "gallery_routes.py").read_text()
     memory_source = (ROOT / "routes" / "memory" / "memory_routes.py").read_text()
 

tests/test_webhook_routes_shim.py

@@ -0,0 +1,11 @@
+"""Regression test for the webhook route shim (slice 2l, #4082/#4071)."""
+
+import importlib
+
+import routes.webhook_routes as _shim_webhook  # noqa: F401
+
+
+def test_legacy_and_canonical_webhook_module_are_same_object():
+    legacy = importlib.import_module("routes.webhook_routes")
+    canonical = importlib.import_module("routes.webhook.webhook_routes")
+    assert legacy is canonical