Staging

.gitignore

@@ -70,6 +70,7 @@ test_*.csv
 
 # Local debugging scripts
 apps/worker/scripts/
+apps/worker/experiments/
 apps/worker/start_celery_worker.py
 apps/worker/start_celery_debug.sh
 apps/worker/clear_celery_queues.sh

apps/api/app/api/dependencies/current_user.py

@@ -10,16 +10,21 @@
 )
 from app.services.rate_limit.job_admission_service import JobAdmissionService
 from fastapi import Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from shared.core.database import get_db
 
 _job_admission_service = JobAdmissionService()
 
 
 async def with_current_user(
     route_context: RouteAdmissionContext = Depends(get_route_admission_context),
     user_id: str = Depends(get_current_user_id),
+    db: AsyncSession = Depends(get_db),
 ) -> AsyncGenerator[CurrentUser, None]:
     current_user = await _job_admission_service.resolve_current_user(
         route_context=route_context,
         user_id=user_id,
+        db=db,
     )
     yield current_user

apps/api/app/services/auth/api_key_authentication_service.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import json
 from datetime import datetime, timezone
 
@@ -11,11 +10,11 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from shared.core.config import redis_pool_manager
-from shared.core.database import get_db_context
 from shared.services.redis.redis_service import RedisService
 from shared.utils.api_keys import hash_api_key
 
 _API_KEY_USER_CACHE_TTL_SECONDS: int = 3600
+_LAST_USED_DEBOUNCE_SECONDS: int = 300
 
 
 class APIKeyAuthenticationService:
@@ -44,8 +43,12 @@ async def validate_api_key(
         if not api_key_record or not api_key_record.is_valid():
             return None
 
-        self._schedule_last_used_update(str(api_key_record.id))
         user_id = str(api_key_record.user_id)
+        await self._update_last_used_best_effort(
+            session,
+            redis_service,
+            str(api_key_record.id),
+        )
         await self._set_cached_user_id(
             redis_service,
             key_hash,
@@ -75,6 +78,10 @@ def _get_user_id_key(api_key_hash: str) -> str:
     def _get_user_api_keys_key(user_id: str) -> str:
         return f"api-key:user-hashes:{user_id}"
 
+    @staticmethod
+    def _get_last_used_debounce_key(api_key_id: str) -> str:
+        return f"api-key:last-used-debounce:{api_key_id}"
+
     async def _get_cached_user_id(
         self,
         redis_service: RedisService,
@@ -159,20 +166,44 @@ def _resolve_api_key_cache_ttl_seconds(self, expires_at: datetime | None) -> int
         remaining_seconds = int((expires_at_utc - now).total_seconds())
         return max(1, min(_API_KEY_USER_CACHE_TTL_SECONDS, remaining_seconds))
 
-    def _schedule_last_used_update(self, api_key_id: str) -> None:
+    async def _update_last_used_best_effort(
+        self,
+        session: AsyncSession,
+        redis_service: RedisService,
+        api_key_id: str,
+    ) -> None:
+        """Update last_used_at on the request session without a nested checkout.
+
+        Redis SET NX debounce skips redundant writes within the debounce window
+        so job polls do not compete for QueuePool via create_task+get_db_context.
+        """
+        debounce_key = self._get_last_used_debounce_key(api_key_id)
         try:
-            asyncio.create_task(
-                self._update_last_used_best_effort(api_key_id),
-                name=f"api_key_last_used:{api_key_id}",
+            acquired = await redis_service.set_nx(
+                debounce_key,
+                "1",
+                ex=_LAST_USED_DEBOUNCE_SECONDS,
             )
-        except Exception as exc:
+            if not acquired:
+                return
+        except Exception:
             logger.warning(
-                f"Failed to schedule API key last-used update (ignored): {exc}"
+                "api_key_authentication: last-used debounce failed for api_key_id={}; "
+                "updating anyway",
+                api_key_id,
             )
 
-    async def _update_last_used_best_effort(self, api_key_id: str) -> None:
         try:
-            async with get_db_context() as db:
-                await self._repository.update_last_used(db, api_key_id)
+            await self._repository.update_last_used(session, api_key_id)
+            # Request-scoped sessions do not auto-commit.
+            await session.commit()
         except Exception as exc:
-            logger.warning(f"Failed to update API key last-used time (ignored): {exc}")
+            logger.warning(
+                f"Failed to update API key last-used time (ignored): {exc}"
+            )
+            try:
+                await session.rollback()
+            except Exception:
+                # Rollback itself can fail if the session is already closed;
+                # ignore so last-used remains best-effort.
+                pass

apps/api/app/services/demo/source_catalog.py

@@ -69,7 +69,7 @@ class DemoSourceDefinition:
                 ),
                 citations=(
                     DemoCitationDefinition(
-                        section_path=("TSLA-Q4-2025-Update.pdf-->OTHER UPDATES"),
+                        section_path=("TSLA-Q4-2025-Update.pdf/OTHER UPDATES"),
                         description="xAI investment",
                         content=(
                             "On January 16, 2026, Tesla entered into an agreement "
@@ -92,7 +92,7 @@ class DemoSourceDefinition:
                 citations=(
                     DemoCitationDefinition(
                         section_path=(
-                            "TSLA-Q4-2025-Update.pdf-->SUMMARY-->"
+                            "TSLA-Q4-2025-Update.pdf/SUMMARY/"
                             "Energy generation and storage"
                         ),
                         description="Storage deployment growth",
@@ -115,7 +115,7 @@ class DemoSourceDefinition:
                 ),
                 citations=(
                     DemoCitationDefinition(
-                        section_path=("TSLA-Q4-2025-Update.pdf-->OUTLOOK-->Product"),
+                        section_path=("TSLA-Q4-2025-Update.pdf/OUTLOOK/Product"),
                         description="2026 production plans",
                         content=(
                             "Cybercab, Tesla Semi and Megapack 3 are on schedule "

apps/api/app/services/demo/source_projection.py

@@ -5,6 +5,11 @@
 from typing import Any, Protocol
 from urllib.parse import quote
 
+from shared.services.chunks.path_segments import (
+    join_document_path,
+    split_escaped_document_path,
+)
+
 _SOURCE_FILE_EXTENSIONS = (
     ".csv",
     ".doc",
@@ -241,22 +246,18 @@ def _publication_path(
     if not raw:
         return prefix
 
-    if "-->" in raw:
-        sections = [part.strip() for part in raw.split("-->")[1:] if part.strip()]
-        return "/".join([prefix, *sections]) if sections else prefix
-
     if raw.startswith("images/") or raw.startswith("tables/"):
         return f"{prefix}/Assets/{raw}"
 
-    parts = [part.strip() for part in raw.split("/") if part.strip()]
+    parts = split_escaped_document_path(raw)
     if parts and parts[0] == source.title:
-        return raw
+        return join_document_path(parts)
     if len(parts) >= 2 and parts[0] == "Default_Root":
         section_parts = parts[2:] if parts[1] == source.title else parts[1:]
-        return "/".join([prefix, *section_parts]) if section_parts else prefix
+        return join_document_path([prefix, *section_parts]) if section_parts else prefix
     if parts and _is_source_file_root(parts[0]):
         section_parts = parts[1:]
-        return "/".join([prefix, *section_parts]) if section_parts else prefix
+        return join_document_path([prefix, *section_parts]) if section_parts else prefix
     return prefix
 
 

apps/api/app/services/jobs/result_projection.py

@@ -15,6 +15,7 @@
 JobStatusValue = Literal[
     "pending", "waiting-file", "running", "converting", "done", "failed"
 ]
+_CHARGED_BILLING_STATUS: str = "charged"
 
 
 def build_error_response(
@@ -112,6 +113,17 @@ def _resolve_duration_seconds(job: Any) -> float | None:
     return None
 
 
+def _resolve_credits_spent(job: Any) -> float:
+    if to_job_status_value(job.status) == "failed":
+        return 0.0
+
+    billing_status = getattr(job, "billing_status", None)
+    if billing_status != _CHARGED_BILLING_STATUS:
+        return 0.0
+
+    return MicroDollar(getattr(job, "credits_charged", 0) or 0).to_credit()
+
+
 async def _resolve_result_delivery(
     job: Any,
 ) -> tuple[dict[str, Any] | None, str | None, datetime]:
@@ -162,9 +174,5 @@ async def build_job_result_response(
         model=parsing_params.get("model"),
         ocr_enabled=parsing_params.get("ocr_enabled"),
         duration_seconds=_resolve_duration_seconds(job),
-        credits_spent=(
-            MicroDollar(job.credits_charged).to_credit()
-            if hasattr(job, "credits_charged")
-            else 0
-        ),
+        credits_spent=_resolve_credits_spent(job),
     )

apps/api/app/services/rate_limit/job_admission_service.py

@@ -33,8 +33,9 @@ async def resolve_current_user(
         *,
         route_context: RouteAdmissionContext,
         user_id: str,
+        db: AsyncSession,
     ) -> CurrentUser:
-        user_tier = await TierService.get_tier(user_id)
+        user_tier = await TierService.get_tier(user_id, session=db)
         self._route_policy_service.enforce_guest_api_key_scope(
             route_context=route_context,
             user_tier=user_tier,

apps/api/app/services/rate_limit/tier_service.py

@@ -31,9 +31,15 @@ class TierService:
     """Manages user tier lookup, caching, and refresh."""
 
     @staticmethod
-    async def get_tier(user_id: str) -> str:
+    async def get_tier(
+        user_id: str,
+        session: AsyncSession | None = None,
+    ) -> str:
         """Return a user's tier from cache or database.
 
+        When ``session`` is provided, reuse it instead of opening a nested
+        ``get_db_context`` checkout (critical on the job-poll auth path).
+
         Missing user tier state is treated as invalid data and raises directly;
         this method never falls back to a default tier for user lookup.
         """
@@ -42,18 +48,43 @@ async def get_tier(user_id: str) -> str:
         if cached_tier is not None:
             return cached_tier
 
-        async with get_db_context() as session:
-            try:
-                user_tier: str = await TierService._get_tier_from_db(session, user_id)
-            except NotFoundException:
-                user_tier = await TierService._initialize_missing_user_tier(
-                    session,
+        if session is not None:
+            user_tier = await TierService._resolve_tier_from_db(
+                session,
+                user_id,
+                commit_on_initialize=True,
+            )
+        else:
+            async with get_db_context() as owned_session:
+                user_tier = await TierService._resolve_tier_from_db(
+                    owned_session,
                     user_id,
+                    commit_on_initialize=False,
                 )
 
         await TierService._set_cached_tier(redis_service, user_id, user_tier)
         return user_tier
 
+    @staticmethod
+    async def _resolve_tier_from_db(
+        session: AsyncSession,
+        user_id: str,
+        *,
+        commit_on_initialize: bool,
+    ) -> str:
+        """Load tier from DB, initializing missing first-use billing state."""
+        try:
+            return await TierService._get_tier_from_db(session, user_id)
+        except NotFoundException:
+            user_tier = await TierService._initialize_missing_user_tier(
+                session,
+                user_id,
+            )
+            # Request-scoped sessions do not auto-commit; get_db_context does.
+            if commit_on_initialize:
+                await session.commit()
+            return user_tier
+
     @staticmethod
     async def refresh_tier(user_id: str, session: AsyncSession) -> str:
         """Called on payment success.

apps/api/tests/contract/test_chunk_document_path_contract.py

@@ -89,38 +89,6 @@ def test_should_not_treat_dotted_section_titles_as_legacy_document_files() -> No
     assert children[0]["path"] == chunk_path
 
 
-def test_should_read_arrow_delimited_document_paths_as_section_paths() -> None:
-    chunk_path = "report.pdf-->Intro-->Subsection"
-
-    doc_nav = ZipResultSchemaBuilder().build_doc_nav(
-        [
-            {
-                "chunk_id": "chunk_arrow_delimited_path",
-                "type": "text",
-                "content": "arrow delimited content",
-                "path": chunk_path,
-                "metadata": {"summary": "arrow delimited summary"},
-            }
-        ],
-        "report.pdf",
-    )
-
-    sections = cast(list[dict[str, object]], doc_nav["sections"])
-    children = cast(list[dict[str, object]], sections[0]["children"])
-
-    assert (
-        section_path_from_chunk_path(
-            chunk_path,
-            source_file_name="report.pdf",
-        )
-        == "Intro / Subsection"
-    )
-    assert sections[0]["title"] == "Intro"
-    assert sections[0]["path"] == "report.pdf/Intro"
-    assert children[0]["title"] == "Subsection"
-    assert children[0]["path"] == "report.pdf/Intro/Subsection"
-
-
 def test_should_preserve_literal_arrow_text_in_slash_paths() -> None:
     chunk_path = "report.pdf/Inputs --> Outputs/Details"
 

apps/api/tests/contract/test_job_creation_contract.py

@@ -417,7 +417,7 @@ async def test_should_create_a_v2_page_memory_job_for_pdf_uploads(
     assert job_metadata["processing_generation"] == "page_memory"
     assert job_metadata["source_file_name"] == payload["file_name"]
     assert page_memory_config["max_pages"] == 1500
-    assert page_memory_config["asset_extraction_enabled"] is False
+    assert page_memory_config["asset_extraction_enabled"] is True
     assert original_request["file_name"] == payload["file_name"]
     assert "parse_track" not in original_request