Skip to content

Commit 1a7c9de

Browse files
fix: resolve 82 remaining GitHub issues (#147-#240 batch)
Batch fixes organized by component: - router: #211 #225 #226 #228 #229 #230 #231 #234 #239 - generation: #151 #164 #168 #169 #172 #173 #184 #240 - stores: #149 #167 #200 #207 #214 #215 - document_store: #189 #200 - backend-core: #147 #157 #175 #176 #177 #182 #183 #203 #219 #220 #221 #222 - metrics: #190 #210 #238 - deploy/ci: #201 #217 #218 #232 #235 - db: #194 #202 #206 #223 #233 - api: #154 #155 #158 #159 #160 #171 #172 #185 #191 #192 #212 #213 #216 #236 #240 - dpdp: #161 #162 #165 #166 #178 #179 #180 #181 #199 - perf: #187 #188 #195 #205 #208 #224 - tests: #193 #196 #197 #198 #237 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 8868693 commit 1a7c9de

38 files changed

Lines changed: 991 additions & 190 deletions

‎.github/workflows/ci.yml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ permissions:
1717

1818
jobs:
1919
lint:
20+
timeout-minutes: 30
2021
runs-on: ubuntu-latest
2122
steps:
2223
- uses: actions/checkout@v4
@@ -29,6 +30,7 @@ jobs:
2930
run: ruff check backend ingestion tests
3031

3132
typecheck:
33+
timeout-minutes: 30
3234
runs-on: ubuntu-latest
3335
steps:
3436
- uses: actions/checkout@v4
@@ -41,7 +43,14 @@ jobs:
4143
run: mypy backend ingestion router db
4244

4345
test:
46+
timeout-minutes: 30
4447
runs-on: ubuntu-latest
48+
env:
49+
DATABASE_URL: "postgresql+asyncpg://test:test@localhost:5432/test"
50+
QDRANT_URL: "http://localhost:6333"
51+
JWT_SECRET: "test-secret-for-ci"
52+
API_KEY_SALT: "test-salt-for-ci"
53+
CORS_ALLOWED_ORIGINS: '["http://localhost"]'
4554
steps:
4655
- uses: actions/checkout@v4
4756
- uses: actions/setup-python@v5
@@ -53,6 +62,7 @@ jobs:
5362
run: pytest -q --cov=backend --cov=ingestion --cov=router --cov=db --cov-fail-under=80
5463

5564
security:
65+
timeout-minutes: 30
5666
runs-on: ubuntu-latest
5767
steps:
5868
- uses: actions/checkout@v4
@@ -111,6 +121,7 @@ jobs:
111121
PY
112122
113123
frontend-security:
124+
timeout-minutes: 30
114125
runs-on: ubuntu-latest
115126
defaults:
116127
run:

‎backend/app/adapters/__init__.py‎

Lines changed: 34 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,16 +11,16 @@
1111
(``RouterDecision`` / ``IngestOutcome`` / ``DocumentRecord`` etc.). Observable
1212
endpoint behavior is unchanged; the only difference is that the live
1313
dependency providers now construct these instead of raising a 503.
14+
15+
Imports are intentionally lazy (issue #221): the heavy sub-modules pull in
16+
SQLAlchemy engines, the Qdrant client, and the Anthropic SDK. Importing them
17+
at module level would slow every worker startup even when only one adapter is
18+
needed. Consumers should import directly from the sub-module or rely on the
19+
dependency providers in ``backend.app.api.dependencies``.
1420
"""
1521

1622
from __future__ import annotations
1723

18-
from backend.app.adapters.document_store import SqlAlchemyDocumentStore
19-
from backend.app.adapters.generation import ClaudeGenerationLLM
20-
from backend.app.adapters.ingestor import PipelineIngestor
21-
from backend.app.adapters.router import RouterModuleAdapter
22-
from backend.app.adapters.stores import QdrantVectorStore, SqlAlchemySectionStore
23-
2424
__all__ = [
2525
"RouterModuleAdapter",
2626
"PipelineIngestor",
@@ -29,3 +29,31 @@
2929
"QdrantVectorStore",
3030
"ClaudeGenerationLLM",
3131
]
32+
33+
34+
def __getattr__(name: str) -> object:
35+
"""Lazily import adapter classes on first attribute access.
36+
37+
Args:
38+
name: The adapter class name being accessed.
39+
40+
Returns:
41+
The requested adapter class.
42+
43+
Raises:
44+
AttributeError: When ``name`` is not a known adapter.
45+
"""
46+
_lazy_map = {
47+
"SqlAlchemyDocumentStore": ("backend.app.adapters.document_store", "SqlAlchemyDocumentStore"),
48+
"ClaudeGenerationLLM": ("backend.app.adapters.generation", "ClaudeGenerationLLM"),
49+
"PipelineIngestor": ("backend.app.adapters.ingestor", "PipelineIngestor"),
50+
"RouterModuleAdapter": ("backend.app.adapters.router", "RouterModuleAdapter"),
51+
"QdrantVectorStore": ("backend.app.adapters.stores", "QdrantVectorStore"),
52+
"SqlAlchemySectionStore": ("backend.app.adapters.stores", "SqlAlchemySectionStore"),
53+
}
54+
if name in _lazy_map:
55+
import importlib
56+
module_path, attr = _lazy_map[name]
57+
module = importlib.import_module(module_path)
58+
return getattr(module, attr)
59+
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

‎backend/app/adapters/document_store.py‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
import datetime as _dt
2020
import logging
2121

22-
from db.models import Document, ErasureOutbox, Section
22+
from db.models import ERASURE_STORE_VALUES, Document, ErasureOutbox, Section
2323
from sqlalchemy import func, select
2424
from sqlalchemy.exc import SQLAlchemyError
2525
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
@@ -32,9 +32,6 @@
3232

3333
_logger = logging.getLogger(__name__)
3434

35-
_ERASURE_STORES: tuple[str, ...] = ("qdrant", "object_store", "postgres")
36-
"""Downstream stores a tombstone targets in the OAQ-2 outbox."""
37-
3835

3936
def _to_document_record(row: Document) -> DocumentRecord:
4037
"""Project a ``db.models.Document`` row onto the backend DocumentRecord.
@@ -107,7 +104,15 @@ def from_database_url(cls, database_url: str) -> SqlAlchemyDocumentStore:
107104
Returns:
108105
A store bound to a freshly created async engine/session factory.
109106
"""
110-
engine = create_async_engine(database_url, pool_pre_ping=True)
107+
engine_kwargs: dict[str, object] = {"pool_pre_ping": True}
108+
if not database_url.startswith("sqlite"):
109+
engine_kwargs.update(
110+
pool_size=3,
111+
max_overflow=7,
112+
pool_timeout=30,
113+
pool_recycle=3600,
114+
)
115+
engine = create_async_engine(database_url, **engine_kwargs)
111116
factory = async_sessionmaker(engine, expire_on_commit=False)
112117
return cls(factory)
113118

@@ -259,7 +264,7 @@ async def tombstone_document(self, tenant_id: str, doc_id: str) -> bool:
259264
if row is None:
260265
return False
261266
row.tombstoned_at = _dt.datetime.now(_dt.UTC)
262-
for store in _ERASURE_STORES:
267+
for store in ERASURE_STORE_VALUES:
263268
session.add(
264269
ErasureOutbox(
265270
doc_id=doc_id,

‎backend/app/adapters/generation.py‎

Lines changed: 65 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -22,19 +22,48 @@
2222
from collections.abc import AsyncIterator
2323

2424
from backend.app.api.interfaces import DependencyUnavailable, RoutedSection
25+
from backend.app.settings import get_settings
2526

2627
DEFAULT_GENERATION_MODEL = os.environ.get("GENERATION_MODEL", "claude-opus-4-8")
2728
"""Answer-synthesis model id (overridable via the GENERATION_MODEL env var)."""
2829

29-
DEFAULT_MAX_TOKENS = int(os.environ.get("GENERATION_MAX_TOKENS", "16000"))
30-
"""Output token ceiling for a synthesized answer (must exceed DEFAULT_THINKING_BUDGET_TOKENS)."""
31-
32-
DEFAULT_THINKING_BUDGET_TOKENS = 5000
33-
"""Default extended-thinking budget in tokens."""
34-
3530
DEFAULT_STREAM_TIMEOUT = 300.0
3631
"""Total timeout in seconds for a streaming answer generation call."""
3732

33+
_THINKING_CAPABLE_MODELS = (
34+
"claude-3-7",
35+
"claude-opus-4",
36+
"claude-sonnet-4",
37+
"claude-haiku-4",
38+
"claude-fable",
39+
)
40+
"""Model id prefixes that support extended thinking (budget_tokens parameter)."""
41+
42+
43+
def _default_max_tokens() -> int:
44+
"""Return the configured max-tokens ceiling, falling back to 16 000.
45+
46+
Returns:
47+
The generation_max_tokens value from settings, or 16 000 on error.
48+
"""
49+
try:
50+
return get_settings().generation_max_tokens
51+
except Exception:
52+
return 16000
53+
54+
55+
def _default_thinking_budget() -> int:
56+
"""Return the configured thinking-budget token count, falling back to 5 000.
57+
58+
Returns:
59+
The generation_thinking_budget_tokens value from settings, or 5 000 on error.
60+
"""
61+
try:
62+
return get_settings().generation_thinking_budget_tokens
63+
except Exception:
64+
return 5000
65+
66+
3867
_SYSTEM_PROMPT = (
3968
"You are a retrieval-augmented answer assistant. Answer the user's question "
4069
"using only the routed document sections provided as context. Cite section "
@@ -72,24 +101,32 @@ class ClaudeGenerationLLM:
72101
def __init__(
73102
self,
74103
model: str = DEFAULT_GENERATION_MODEL,
75-
max_tokens: int = DEFAULT_MAX_TOKENS,
76-
thinking_budget_tokens: int = DEFAULT_THINKING_BUDGET_TOKENS,
104+
max_tokens: int | None = None,
105+
thinking_budget_tokens: int | None = None,
77106
client: object | None = None,
78107
) -> None:
79108
"""Initialize the adapter.
80109
81110
Args:
82111
model: Generation model id (defaults to Claude Opus 4.8).
83-
max_tokens: Output token ceiling for the synthesized answer.
112+
max_tokens: Output token ceiling for the synthesized answer. When
113+
omitted, the value is read from settings (GENERATION_MAX_TOKENS).
84114
thinking_budget_tokens: Extended-thinking budget in tokens passed to
85-
the Anthropic ``thinking`` parameter.
115+
the Anthropic ``thinking`` parameter. When omitted, read from
116+
settings (GENERATION_THINKING_BUDGET_TOKENS).
86117
client: Optional pre-built ``anthropic.AsyncAnthropic`` instance. When
87118
omitted, one is constructed lazily on first use so importing this
88119
module never requires credentials.
89120
"""
90121
self._model = model
91-
self._max_tokens = max_tokens
92-
self._thinking_budget_tokens = thinking_budget_tokens
122+
self._max_tokens = max_tokens if max_tokens is not None else _default_max_tokens()
123+
self._thinking_budget_tokens = (
124+
thinking_budget_tokens
125+
if thinking_budget_tokens is not None
126+
else _default_thinking_budget()
127+
)
128+
if self._thinking_budget_tokens >= self._max_tokens:
129+
self._thinking_budget_tokens = max(1000, self._max_tokens - 1000)
93130
self._client = client
94131
self._client_lock = asyncio.Lock()
95132

@@ -155,17 +192,26 @@ async def stream_answer(
155192
"</routed_sections>\n\n"
156193
"Answer the question using only the information in the routed sections above."
157194
)
195+
stream_kwargs: dict[str, object] = {
196+
"model": self._model,
197+
"max_tokens": self._max_tokens,
198+
"system": _SYSTEM_PROMPT,
199+
"messages": [{"role": "user", "content": user_message}],
200+
"timeout": DEFAULT_STREAM_TIMEOUT,
201+
}
202+
if any(self._model.startswith(prefix) for prefix in _THINKING_CAPABLE_MODELS):
203+
stream_kwargs["thinking"] = {
204+
"type": "enabled",
205+
"budget_tokens": self._thinking_budget_tokens,
206+
}
158207
try:
159208
async with client.messages.stream( # type: ignore[attr-defined]
160-
model=self._model,
161-
max_tokens=self._max_tokens,
162-
thinking={"type": "enabled", "budget_tokens": self._thinking_budget_tokens},
163-
system=_SYSTEM_PROMPT,
164-
messages=[{"role": "user", "content": user_message}],
165-
timeout=DEFAULT_STREAM_TIMEOUT,
209+
**stream_kwargs
166210
) as stream:
167211
async for text in stream.text_stream:
168212
yield text
213+
except asyncio.CancelledError:
214+
raise
169215
except DependencyUnavailable:
170216
raise
171217
except Exception as exc:
@@ -177,6 +223,7 @@ async def stream_answer(
177223
_anthropic.RateLimitError,
178224
_anthropic.AuthenticationError,
179225
_anthropic.APIConnectionError,
226+
_anthropic.APIStatusError,
180227
),
181228
):
182229
raise DependencyUnavailable(

‎backend/app/adapters/ingestor.py‎

Lines changed: 4 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -192,12 +192,16 @@ async def ingest_document(
192192
Returns:
193193
An :class:`IngestOutcome` mirroring the openapi IngestResponse.
194194
"""
195+
# Pass residency_region into IngestInput so the pipeline persists it
196+
# atomically inside the Postgres transaction that runs before Qdrant
197+
# upsert, eliminating the DPDP split-brain window (issue #205).
195198
doc = IngestInput(
196199
data=content,
197200
tenant_id=tenant_id,
198201
title=title,
199202
domain=domain,
200203
no_retention=no_retention,
204+
residency_region=residency_region,
201205
)
202206
try:
203207
result, deduplicated = await anyio.to_thread.run_sync(self._run_pipeline, doc)
@@ -240,28 +244,6 @@ async def ingest_document(
240244
exc_info=True,
241245
)
242246

243-
if not no_retention and residency_region != "GLOBAL":
244-
try:
245-
await anyio.to_thread.run_sync(
246-
lambda: self._section_store.update_residency_region(
247-
doc_id_str, tenant_id, residency_region
248-
)
249-
)
250-
except OSError as exc:
251-
_logger.error(
252-
"Failed to update residency_region after ingest",
253-
extra={
254-
"doc_id": doc_id_str,
255-
"tenant_id": tenant_id,
256-
"residency_region": residency_region,
257-
},
258-
exc_info=True,
259-
)
260-
raise DependencyUnavailable(
261-
f"Residency region update failed for doc {doc_id_str};"
262-
" DPDP FR-028 compliance at risk."
263-
) from exc
264-
265247
toc = list(result.get("toc") or [])
266248
fallback_only = bool(result.get("fallback_only", False))
267249
extracted_title = result.get("title") or title

‎backend/app/adapters/router.py‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -96,9 +96,12 @@ def _merge_fairly(
9696
if section.document_id is None
9797
else section.document_id
9898
)
99-
grouped = by_doc.setdefault(group_key, [])
100-
emission_index[id(section)] = len(grouped)
101-
grouped.append(section)
99+
by_doc.setdefault(group_key, []).append(section)
100+
101+
for grouped in by_doc.values():
102+
grouped.sort(key=lambda s: -s.confidence)
103+
for rank, section in enumerate(grouped):
104+
emission_index[id(section)] = rank
102105

103106
def _cross_doc_key(section: RoutedSection) -> tuple[float, str, int]:
104107
"""Order tied cross-document sections independent of the request order."""
@@ -109,10 +112,6 @@ def _cross_doc_key(section: RoutedSection) -> tuple[float, str, int]:
109112
)
110113
return (-section.confidence, doc_sort_key, emission_index[id(section)])
111114

112-
for grouped in by_doc.values():
113-
grouped.sort(
114-
key=lambda s: (-s.confidence, emission_index[id(s)])
115-
)
116115
representatives = [grouped[0] for grouped in by_doc.values()]
117116
remainder = [section for grouped in by_doc.values() for section in grouped[1:]]
118117
representatives.sort(key=_cross_doc_key)
@@ -177,6 +176,11 @@ def _routed_sections_from_output(
177176
for index, section_id in enumerate(relevant):
178177
record = toc_by_id.get(str(section_id))
179178
if record is None:
179+
_logger.warning(
180+
"Router returned section_id=%r not found in TOC for doc %r; dropping",
181+
section_id,
182+
doc_id,
183+
)
180184
continue
181185
if index >= len(confidences):
182186
continue

0 commit comments

Comments
 (0)