fix(security): prevent ReDoS in verdict-prose and continuation matchers (#4943)

Two py/polynomial-redos sinks ran regexes with two adjacent \s-matching
quantifiers over untrusted model text, backtracking O(n^2) when the tail failed
on a whitespace flood:

  - routes/skills_routes.py: the last-resort verdict-from-prose extractor used
    `["\'\s:]*\s*` — the class already matches \s, so the trailing \s* was a
    redundant second quantifier. Dropped it (extracted to a documented module
    constant _VERDICT_PROSE_RE); the matched text is identical, the scan linear.
  - src/agent_loop.py _EXPLICIT_CONTINUATION_RE: `\s*[.!?]*\s*$` put two \s*
    around `[.!?]*`. Rewrote as `\s*(?:[.!?]+\s*)?$` — same accepted tails (no
    two \s* adjacent), linear. Portable form (no possessive quantifiers).

Both verified output-equivalent to the originals across a fuzz corpus. Adds
tests/test_redos_verdict_continuation.py pinning the unchanged match sets and
bounding the flood inputs (old patterns took seconds at 40k whitespace chars).

app.py

@@ -685,7 +685,7 @@ from routes.signature_routes import setup_signature_routes
 app.include_router(setup_signature_routes())
 
 # Gallery (image library)
-from routes.gallery_routes import setup_gallery_routes
+from routes.gallery.gallery_routes import setup_gallery_routes
 app.include_router(setup_gallery_routes())
 
 # Persisted image-editor drafts (server-backed projects)

routes/gallery/__init__.py

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

routes/gallery/gallery_helpers.py

@@ -0,0 +1,144 @@
+"""gallery_helpers.py — extracted helpers, models, and small utilities.
+
+Imported by gallery_routes.py."""
+
+"""Gallery routes — browsable library for photos and AI-generated images."""
+
+import logging
+from datetime import datetime
+from typing import Dict, Any, Optional
+
+from pydantic import BaseModel
+
+from core.database import GalleryImage
+from src.auth_helpers import _auth_disabled
+
+logger = logging.getLogger(__name__)
+
+
+# ---- Request schemas ----
+
+class GalleryPatch(BaseModel):
+    tags: Optional[str] = None
+    favorite: Optional[bool] = None
+    album_id: Optional[str] = None
+
+
+# ---- EXIF extraction ----
+
+def _extract_exif(content: bytes) -> dict:
+    """Extract EXIF metadata from image bytes. Returns dict of fields."""
+    result = {"width": None, "height": None}
+    try:
+        from PIL import Image
+        from io import BytesIO
+        img = Image.open(BytesIO(content))
+        # Read the raw EXIF before any transpose: exif_transpose strips the
+        # orientation tag and with it the parsed EXIF view.
+        exif = img._getexif() if hasattr(img, '_getexif') else None
+
+        # Record DISPLAY dimensions (EXIF-rotated), matching upload_handler.
+        # A phone photo with Orientation 6/8 is stored landscape but shown
+        # portrait, so the raw width/height swap the aspect ratio.
+        try:
+            from PIL import ImageOps
+            img = ImageOps.exif_transpose(img) or img
+        except Exception:
+            pass
+        result["width"] = img.width
+        result["height"] = img.height
+
+        if not exif:
+            return result
+
+        # EXIF tag IDs
+        # 271=Make, 272=Model, 306=DateTime, 36867=DateTimeOriginal
+        # 34853=GPSInfo
+        result["camera_make"] = str(exif.get(271, "")).strip() or None
+        result["camera_model"] = str(exif.get(272, "")).strip() or None
+
+        # Date taken
+        for tag_id in (36867, 36868, 306):  # DateTimeOriginal, DateTimeDigitized, DateTime
+            raw = exif.get(tag_id)
+            if raw:
+                try:
+                    result["taken_at"] = datetime.strptime(str(raw).strip(), "%Y:%m:%d %H:%M:%S")
+                    break
+                except (ValueError, TypeError):
+                    pass
+
+        # GPS
+        gps_info = exif.get(34853)
+        if gps_info and isinstance(gps_info, dict):
+            try:
+                def _to_deg(vals):
+                    d, m, s = [float(v) for v in vals]
+                    return d + m / 60 + s / 3600
+                if 2 in gps_info and 4 in gps_info:
+                    lat = _to_deg(gps_info[2])
+                    lng = _to_deg(gps_info[4])
+                    if gps_info.get(1) == 'S': lat = -lat
+                    if gps_info.get(3) == 'W': lng = -lng
+                    result["gps_lat"] = f"{lat:.6f}"
+                    result["gps_lng"] = f"{lng:.6f}"
+            except Exception:
+                pass
+    except Exception as e:
+        # User-visible failure (photo loses metadata): surface at WARNING
+        # and record on the result so the upload endpoint can pass it back.
+        logger.warning(f"EXIF extraction failed: {e}")
+        result["exif_error"] = str(e)
+    return result
+
+
+# ---- Helpers ----
+
+def _image_to_dict(img: GalleryImage, session_name: str = None) -> Dict[str, Any]:
+    return {
+        "id": img.id,
+        "filename": img.filename,
+        "url": f"/api/generated-image/{img.filename}",
+        "prompt": img.prompt,
+        "model": img.model,
+        "size": img.size,
+        "quality": img.quality,
+        "tags": img.tags or "",
+        "ai_tags": img.ai_tags or "",
+        "user_tags": img.tags or "",
+        "session_id": img.session_id,
+        "session_name": session_name,
+        "album_id": img.album_id,
+        "is_active": img.is_active,
+        "favorite": img.favorite or False,
+        "taken_at": img.taken_at.isoformat() if img.taken_at else None,
+        "camera": f"{img.camera_make or ''} {img.camera_model or ''}".strip() or None,
+        "gps": {"lat": img.gps_lat, "lng": img.gps_lng} if img.gps_lat else None,
+        "width": img.width,
+        "height": img.height,
+        "file_size": img.file_size,
+        "created_at": img.created_at.isoformat() if img.created_at else None,
+        "updated_at": img.updated_at.isoformat() if img.updated_at else None,
+    }
+
+
+def _owner_filter(q, user, model_cls=GalleryImage):
+    """Apply owner filtering to a gallery query.
+
+    ``get_current_user`` returns None both in auth-disabled single-user mode
+    and when auth is enabled but no current user was resolved. Preserve the
+    single-user behavior, but fail closed for auth-enabled null-user states.
+    """
+    if user is not None:
+        return q.filter(model_cls.owner == user)
+    if _auth_disabled():
+        return q
+    return q.filter(False)
+
+
+
+def _human_size(nbytes):
+    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
+        if abs(nbytes) < 1024:
+            return f"{nbytes:.1f} {unit}"
+        nbytes /= 1024
+    return f"{nbytes:.1f} PB"

routes/gallery_helpers.py

@@ -1,144 +1,14 @@
-"""gallery_helpers.py — extracted helpers, models, and small utilities.
+"""Backward-compat shim — canonical location is routes/gallery/gallery_helpers.py.
 
-Imported by gallery_routes.py."""
+This module is replaced in ``sys.modules`` by the canonical module object so
+that ``import routes.gallery_helpers``, ``from routes.gallery_helpers import X``,
+``importlib.import_module("routes.gallery_helpers")``, and
+``monkeypatch.setattr(routes.gallery_helpers, ...)`` all operate on the *same*
+object. Keeps existing import paths working after slice 2a (#4082/#4071).
+"""
 
-"""Gallery routes — browsable library for photos and AI-generated images."""
+import sys as _sys
 
-import logging
+from routes.gallery import gallery_helpers as _canonical  # noqa: F401
-from datetime import datetime
-from typing import Dict, Any, Optional
 
-from pydantic import BaseModel
+_sys.modules[__name__] = _canonical
-
-from core.database import GalleryImage
-from src.auth_helpers import _auth_disabled
-
-logger = logging.getLogger(__name__)
-
-
-# ---- Request schemas ----
-
-class GalleryPatch(BaseModel):
-    tags: Optional[str] = None
-    favorite: Optional[bool] = None
-    album_id: Optional[str] = None
-
-
-# ---- EXIF extraction ----
-
-def _extract_exif(content: bytes) -> dict:
-    """Extract EXIF metadata from image bytes. Returns dict of fields."""
-    result = {"width": None, "height": None}
-    try:
-        from PIL import Image
-        from io import BytesIO
-        img = Image.open(BytesIO(content))
-        # Read the raw EXIF before any transpose: exif_transpose strips the
-        # orientation tag and with it the parsed EXIF view.
-        exif = img._getexif() if hasattr(img, '_getexif') else None
-
-        # Record DISPLAY dimensions (EXIF-rotated), matching upload_handler.
-        # A phone photo with Orientation 6/8 is stored landscape but shown
-        # portrait, so the raw width/height swap the aspect ratio.
-        try:
-            from PIL import ImageOps
-            img = ImageOps.exif_transpose(img) or img
-        except Exception:
-            pass
-        result["width"] = img.width
-        result["height"] = img.height
-
-        if not exif:
-            return result
-
-        # EXIF tag IDs
-        # 271=Make, 272=Model, 306=DateTime, 36867=DateTimeOriginal
-        # 34853=GPSInfo
-        result["camera_make"] = str(exif.get(271, "")).strip() or None
-        result["camera_model"] = str(exif.get(272, "")).strip() or None
-
-        # Date taken
-        for tag_id in (36867, 36868, 306):  # DateTimeOriginal, DateTimeDigitized, DateTime
-            raw = exif.get(tag_id)
-            if raw:
-                try:
-                    result["taken_at"] = datetime.strptime(str(raw).strip(), "%Y:%m:%d %H:%M:%S")
-                    break
-                except (ValueError, TypeError):
-                    pass
-
-        # GPS
-        gps_info = exif.get(34853)
-        if gps_info and isinstance(gps_info, dict):
-            try:
-                def _to_deg(vals):
-                    d, m, s = [float(v) for v in vals]
-                    return d + m / 60 + s / 3600
-                if 2 in gps_info and 4 in gps_info:
-                    lat = _to_deg(gps_info[2])
-                    lng = _to_deg(gps_info[4])
-                    if gps_info.get(1) == 'S': lat = -lat
-                    if gps_info.get(3) == 'W': lng = -lng
-                    result["gps_lat"] = f"{lat:.6f}"
-                    result["gps_lng"] = f"{lng:.6f}"
-            except Exception:
-                pass
-    except Exception as e:
-        # User-visible failure (photo loses metadata): surface at WARNING
-        # and record on the result so the upload endpoint can pass it back.
-        logger.warning(f"EXIF extraction failed: {e}")
-        result["exif_error"] = str(e)
-    return result
-
-
-# ---- Helpers ----
-
-def _image_to_dict(img: GalleryImage, session_name: str = None) -> Dict[str, Any]:
-    return {
-        "id": img.id,
-        "filename": img.filename,
-        "url": f"/api/generated-image/{img.filename}",
-        "prompt": img.prompt,
-        "model": img.model,
-        "size": img.size,
-        "quality": img.quality,
-        "tags": img.tags or "",
-        "ai_tags": img.ai_tags or "",
-        "user_tags": img.tags or "",
-        "session_id": img.session_id,
-        "session_name": session_name,
-        "album_id": img.album_id,
-        "is_active": img.is_active,
-        "favorite": img.favorite or False,
-        "taken_at": img.taken_at.isoformat() if img.taken_at else None,
-        "camera": f"{img.camera_make or ''} {img.camera_model or ''}".strip() or None,
-        "gps": {"lat": img.gps_lat, "lng": img.gps_lng} if img.gps_lat else None,
-        "width": img.width,
-        "height": img.height,
-        "file_size": img.file_size,
-        "created_at": img.created_at.isoformat() if img.created_at else None,
-        "updated_at": img.updated_at.isoformat() if img.updated_at else None,
-    }
-
-
-def _owner_filter(q, user, model_cls=GalleryImage):
-    """Apply owner filtering to a gallery query.
-
-    ``get_current_user`` returns None both in auth-disabled single-user mode
-    and when auth is enabled but no current user was resolved. Preserve the
-    single-user behavior, but fail closed for auth-enabled null-user states.
-    """
-    if user is not None:
-        return q.filter(model_cls.owner == user)
-    if _auth_disabled():
-        return q
-    return q.filter(False)
-
-
-
-def _human_size(nbytes):
-    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
-        if abs(nbytes) < 1024:
-            return f"{nbytes:.1f} {unit}"
-        nbytes /= 1024
-    return f"{nbytes:.1f} PB"

routes/skills_routes.py

@@ -22,6 +22,16 @@ from core.middleware import require_admin
 
 logger = logging.getLogger(__name__)
 
+# Last-resort verdict extraction from a teacher/verifier model's prose (run when
+# JSON parsing fails). `["\'\s:]*` already consumes whitespace, so the original
+# trailing `\s*` made two adjacent \s-matching quantifiers that backtrack O(n^2)
+# on a `verdict` + whitespace flood in untrusted model output (CodeQL
+# py/polynomial-redos). Without it a single unbounded quantifier remains — the
+# matched text is identical, and the scan is linear.
+_VERDICT_PROSE_RE = re.compile(
+    r'verdict["\'\s:]*["\']?(pass|needs_work|fail|inconclusive)', re.I
+)
+
 
 class SkillAddRequest(BaseModel):
     # New schema (preferred)
@@ -196,7 +206,7 @@ async def _eval_skill_run(skill_md: str, task: str, transcript: str,
         # Last resort: pull the verdict keyword straight out of the prose so a
         # clearly-decided run isn't thrown away as "unparseable".
         if v not in _VERDICTS:
-            km = _re.search(r'verdict["\'\s:]*\s*["\']?(pass|needs_work|fail|inconclusive)', text, _re.I)
+            km = _VERDICT_PROSE_RE.search(text)
             if km:
                 v = km.group(1).lower()
                 if data is None:

scripts/claim_ownerless.py

@@ -27,12 +27,18 @@ def claim_json_entries(entries, owner):
     return count
 
 
+def owner_arg(argv):
+    if len(argv) < 2 or not argv[1].strip():
+        return None
+    return argv[1].strip()
+
+
 def main():
-    if len(sys.argv) < 2:
+    owner = owner_arg(sys.argv)
+    if not owner:
         print("Usage: python scripts/claim_ownerless.py <username>")
         sys.exit(1)
 
-    owner = sys.argv[1]
     print(f"Claiming all ownerless data for: {owner}\n")
 
     # 1. Memories (JSON files)

src/agent_loop.py

@@ -845,7 +845,12 @@ _EXPLICIT_CONTINUATION_RE = re.compile(
     r"run it|launch it|start it|use that|that one|same|the same|"
     r"first|second|third|the first one|the second one|the third one|"
     r"[123]|[abc]"
-    r")\s*[.!?]*\s*$",
+    # `\s*[.!?]*\s*$` put two \s-matching quantifiers around `[.!?]*`, which
+    # backtracks O(n^2) on a terse reply + whitespace flood (py/polynomial-redos).
+    # `\s*(?:[.!?]+\s*)?$` accepts the same "trailing space/punctuation" tails
+    # (the inner \s* only engages after `[.!?]+`, so no two \s* are adjacent) and
+    # is linear.
+    r")\s*(?:[.!?]+\s*)?$",
     re.IGNORECASE,
 )
 _RETRY_CONTINUATION_RE = re.compile(

tests/test_claim_ownerless_json.py

@@ -1,4 +1,4 @@
-from scripts.claim_ownerless import claim_json_entries
+from scripts.claim_ownerless import claim_json_entries, owner_arg
 
 
 def test_claim_json_entries_skips_invalid_rows():
@@ -16,3 +16,9 @@ def test_claim_json_entries_skips_invalid_rows():
         None,
         {"id": "b", "owner": "already"},
     ]
+
+
+def test_owner_arg_rejects_blank_owner():
+    assert owner_arg(["claim_ownerless.py"]) is None
+    assert owner_arg(["claim_ownerless.py", "   "]) is None
+    assert owner_arg(["claim_ownerless.py", " admin "]) == "admin"

tests/test_direct_upload_limits.py

@@ -40,7 +40,7 @@ def test_direct_upload_routes_use_bounded_reads():
         "routes/stt_routes.py": [
             "read_upload_limited(file, STT_MAX_AUDIO_BYTES",
         ],
-        "routes/gallery_routes.py": [
+        "routes/gallery/gallery_routes.py": [
             "read_upload_limited(file, GALLERY_UPLOAD_MAX_BYTES",
             "read_upload_limited(file, GALLERY_TRANSFORM_UPLOAD_MAX_BYTES",
         ],

tests/test_endpoint_owner_scope_followup.py

@@ -377,7 +377,7 @@ def test_compare_endpoint_key_lookup_is_owner_scoped():
 
 
 def test_gallery_image_endpoint_lookups_are_owner_scoped():
-    body = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
+    body = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
     helper_body = body.split("def _visible_image_endpoint_query", 1)[1].split(
         "def _first_visible_image_endpoint", 1
     )[0]

tests/test_gallery_album_owner_scope.py

@@ -12,7 +12,7 @@ from pathlib import Path
 
 
 def _function_sources():
-    source = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
+    source = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
     tree = ast.parse(source)
     return {
         node.name: ast.get_source_segment(source, node) or ""

tests/test_gallery_endpoint_ssrf.py

@@ -15,7 +15,7 @@ metadata range.
 import ast
 from pathlib import Path
 
-SRC = Path(__file__).resolve().parent.parent / "routes" / "gallery_routes.py"
+SRC = Path(__file__).resolve().parent.parent / "routes" / "gallery" / "gallery_routes.py"
 
 
 def _function_source(src_text: str, func_name: str) -> str:

tests/test_gallery_exif_orientation.py

@@ -32,8 +32,8 @@ def extract_exif(monkeypatch):
             return MagicMock()
 
     monkeypatch.setitem(sys.modules, "core.database", _DBStub("core.database"))
-    monkeypatch.delitem(sys.modules, "routes.gallery_helpers", raising=False)
+    monkeypatch.delitem(sys.modules, "routes.gallery.gallery_helpers", raising=False)
-    mod = importlib.import_module("routes.gallery_helpers")
+    mod = importlib.import_module("routes.gallery.gallery_helpers")
     return mod._extract_exif
 
 

tests/test_gallery_filename_confinement.py

@@ -128,7 +128,7 @@ def test_gallery_replace_rejects_symlink_escape(tmp_path, monkeypatch):
 
 
 def test_gallery_file_operations_use_confining_resolver():
-    source = Path("routes/gallery_routes.py").read_text(encoding="utf-8")
+    source = Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
 
     assert 'Path("data/generated_images") / img.filename' not in source
     assert 'os.path.join("data", "generated_images", img.filename)' not in source

tests/test_gallery_image_privileges.py

@@ -15,7 +15,7 @@ GATED_IMAGE_FUNCTIONS = {
 
 
 def _gallery_source():
-    return Path("routes/gallery_routes.py").read_text(encoding="utf-8")
+    return Path("routes/gallery/gallery_routes.py").read_text(encoding="utf-8")
 
 
 def _function_sources(source):

tests/test_gallery_routes_shim.py

@@ -0,0 +1,52 @@
+"""Regression test for the gallery route shim (slice 2a, #4082/#4071).
+
+The backward-compat shims at ``routes/gallery_routes.py`` and
+``routes/gallery_helpers.py`` use ``sys.modules`` replacement so the legacy
+import path and the canonical ``routes.gallery.*`` path resolve to the *same*
+module object. This test pins that contract: if the shim is ever changed to a
+plain ``from ... import *`` (or removed), these assertions catch it before the
+monkeypatch-based gallery tests silently start patching the wrong module.
+"""
+
+import importlib
+
+import routes.gallery_routes as _shim_routes  # noqa: F401
+import routes.gallery_helpers as _shim_helpers  # noqa: F401
+
+
+def test_legacy_and_canonical_route_module_are_same_object():
+    """``import routes.gallery_routes`` must alias the canonical module."""
+    legacy = importlib.import_module("routes.gallery_routes")
+    canonical = importlib.import_module("routes.gallery.gallery_routes")
+    assert legacy is canonical, (
+        "routes.gallery_routes shim must resolve to the canonical "
+        "routes.gallery.gallery_routes module object"
+    )
+
+
+def test_legacy_and_canonical_helpers_module_are_same_object():
+    """``import routes.gallery_helpers`` must alias the canonical module."""
+    legacy = importlib.import_module("routes.gallery_helpers")
+    canonical = importlib.import_module("routes.gallery.gallery_helpers")
+    assert legacy is canonical, (
+        "routes.gallery_helpers shim must resolve to the canonical "
+        "routes.gallery.gallery_helpers module object"
+    )
+
+
+def test_monkeypatch_via_legacy_path_affects_canonical(monkeypatch):
+    """Patching through the legacy path must reach the canonical module.
+
+    Several gallery tests do ``import routes.gallery_routes as gr`` followed by
+    ``monkeypatch.setattr(gr, "get_current_user", ...)``. For that to take
+    effect at runtime, the legacy module object and the canonical one must be
+    identical.
+    """
+    legacy = importlib.import_module("routes.gallery_routes")
+    canonical = importlib.import_module("routes.gallery.gallery_routes")
+
+    sentinel = object()
+    monkeypatch.setattr(legacy, "setup_gallery_routes", sentinel)
+    assert canonical.setup_gallery_routes is sentinel, (
+        "monkeypatch via legacy path did not reach the canonical module"
+    )

tests/test_redos_verdict_continuation.py

@@ -0,0 +1,80 @@
+"""Regression tests for two py/polynomial-redos sinks over untrusted model text.
+
+Both had two adjacent `\\s`-matching quantifiers that backtrack O(n^2) when the
+rest of the pattern fails on a whitespace flood:
+
+  * `routes/skills_routes.py` `_VERDICT_PROSE_RE` — `["\\'\\s:]*\\s*` (the class
+    already matches `\\s`) over a teacher/verifier model's prose verdict.
+  * `src/agent_loop.py` `_EXPLICIT_CONTINUATION_RE` — `\\s*[.!?]*\\s*$` over a
+    user's terse reply.
+
+Each is rewritten to drop the adjacency while keeping the exact match set. The
+tests pin correctness (matches unchanged) and bound the flood inputs; the old
+patterns took seconds, the loose budget is seconds, so the margin is ~100x.
+"""
+
+import time
+
+import pytest
+
+import src.agent_tools  # noqa: F401  (break agent_tools<->agent_loop import cycle)
+from routes.skills_routes import _VERDICT_PROSE_RE
+from src.agent_loop import _EXPLICIT_CONTINUATION_RE, _is_explicit_continuation
+
+_BUDGET_S = 4.0
+
+
+def _timed(fn, *args):
+    start = time.perf_counter()
+    result = fn(*args)
+    return result, time.perf_counter() - start
+
+
+# ── #229 verdict-from-prose: matches unchanged ──────────────────────────────
+
+@pytest.mark.parametrize("text,expected", [
+    ('verdict": "FAIL"', "fail"),
+    ("verdict needs_work", "needs_work"),
+    ("Verdict:   inconclusive", "inconclusive"),
+    ("verdict\t\t'pass'", "pass"),
+    ("verdictpass", "pass"),  # all separators optional — keyword may abut, as before
+    ("the verdict is: pass overall", None),  # intervening "is" breaks the run
+    ("no clear decision here", None),
+])
+def test_verdict_prose_extraction(text, expected):
+    m = _VERDICT_PROSE_RE.search(text)
+    assert (m.group(1).lower() if m else None) == expected
+
+
+def test_verdict_prose_flood_is_fast():
+    evil = "verdict" + "\t" * 40000 + "x"  # `verdict` then whitespace, no keyword
+    (m, dt) = _timed(_VERDICT_PROSE_RE.search, evil)
+    assert dt < _BUDGET_S, f"_VERDICT_PROSE_RE took {dt:.2f}s"
+    assert m is None
+
+
+# ── #472 explicit-continuation: classification unchanged ────────────────────
+
+@pytest.mark.parametrize("text", [
+    "yes", "y", "ok!", "okay ...", "sure!!", "do it", "1", "a", "2.",
+    "the second one", "  yes  ", "continue", "run it!", "third???",
+])
+def test_continuation_accepts_terse_confirmations(text):
+    assert _is_explicit_continuation(text)
+
+
+@pytest.mark.parametrize("text", [
+    "no", "maybe yes", "yesx", "let's not", "y . ! .", "", "run the script please",
+])
+def test_continuation_rejects_non_confirmations(text):
+    assert not _is_explicit_continuation(text)
+
+
+def test_continuation_flood_is_fast():
+    evil = "y" + "\t" * 40000 + "x"  # terse opener then whitespace flood, no `$`
+    (_, dt) = _timed(_is_explicit_continuation, evil)
+    assert dt < _BUDGET_S, f"_is_explicit_continuation took {dt:.2f}s"
+    # Direct on the compiled pattern too (the function strips first).
+    (m, dt2) = _timed(_EXPLICIT_CONTINUATION_RE.match, evil)
+    assert dt2 < _BUDGET_S, f"_EXPLICIT_CONTINUATION_RE took {dt2:.2f}s"
+    assert m is None

tests/test_security_regressions.py

@@ -1099,9 +1099,9 @@ def _import_session_routes_for_filename():
 def _import_gallery_routes_for_filename():
     # Same rationale as the session route helper: import _sanitize_gallery_filename
     # against the real core.database and leave a clean, real module cached.
-    _drop_route_module_cache("routes.gallery_routes")
+    _drop_route_module_cache("routes.gallery.gallery_routes")
-    _drop_route_module_cache("routes.gallery_helpers")
+    _drop_route_module_cache("routes.gallery.gallery_helpers")
-    return importlib.import_module("routes.gallery_routes")
+    return importlib.import_module("routes.gallery.gallery_routes")
 
 
 def test_export_filename_sanitizer_blocks_header_and_path_chars():

tests/test_upload_limits_centralized.py

@@ -80,7 +80,7 @@ def test_non_positive_env_rejected(monkeypatch, env):
 def test_routes_import_from_upload_limits_not_local_defs():
     """Routes must import the constant, not redefine it via raw getenv / literal."""
     forbidden = {
-        "routes/gallery_routes.py": [
+        "routes/gallery/gallery_routes.py": [
             'int(os.getenv("ODYSSEUS_GALLERY_UPLOAD_MAX_BYTES"',
             'int(os.getenv("ODYSSEUS_GALLERY_TRANSFORM_UPLOAD_MAX_BYTES"',
         ],
@@ -97,7 +97,7 @@ def test_routes_import_from_upload_limits_not_local_defs():
 
     # And each imports from upload_limits.
     imports = {
-        "routes/gallery_routes.py": "GALLERY_UPLOAD_MAX_BYTES",
+        "routes/gallery/gallery_routes.py": "GALLERY_UPLOAD_MAX_BYTES",
         "routes/memory_routes.py": "MEMORY_IMPORT_MAX_BYTES",
         "routes/personal_routes.py": "PERSONAL_UPLOAD_MAX_BYTES",
         "routes/email_routes.py": "EMAIL_COMPOSE_UPLOAD_MAX_BYTES",

tests/test_vision_owner_scope.py

@@ -89,7 +89,7 @@ def test_request_vision_call_sites_pass_owner():
     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()
-    gallery_source = (ROOT / "routes" / "gallery_routes.py").read_text()
+    gallery_source = (ROOT / "routes" / "gallery" / "gallery_routes.py").read_text()
     memory_source = (ROOT / "routes" / "memory_routes.py").read_text()
 
     assert 'analyze_image_with_vl_result(file_info["path"], owner=owner)' in chat_source