⚡ Bolt: O(1) aggregation for job event sequences

.jules/bolt.md

@@ -0,0 +1,3 @@
+## 2025-05-18 - Optimize Maximum Sequence Number Lookup
+**Learning:** Using array indexing on repository list methods (e.g., `list_for_job(limit=1000)[-1]`) to determine maximum sequence numbers is an anti-pattern. It loads unnecessary records into memory causing N+1 query-like inefficiencies.
+**Action:** Use direct SQLAlchemy aggregation queries (e.g., `select(func.coalesce(func.max(Model.field), 0))`) in the repository methods for $O(1)$ database efficiency.

apps/api/src/cortex_api/services/jobs.py

@@ -110,8 +110,8 @@ async def cancel_job(uow: CortexUnitOfWork, job_id: str) -> JobStatusDetail:
     if updated is None:  # pragma: no cover - defensive
         raise NotFoundError(f"Job `{job_id}` was not found.")
 
-    existing_events = await uow.job_events.list_for_job(job_id, limit=1000)
-    next_sequence = existing_events[-1].sequence_no + 1 if existing_events else 1
+    max_sequence = await uow.job_events.get_max_sequence_no(job_id)
+    next_sequence = max_sequence + 1
     await uow.job_events.add(
         JobEventRecord(
             job_id=job_id,

packages/db/src/cortex_db/repositories.py

@@ -49,7 +49,7 @@
     SynthesisType,
     TenantRecord,
 )
-from sqlalchemy import desc, or_, select, update
+from sqlalchemy import desc, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from .models import (
@@ -1637,6 +1637,14 @@ async def add(self, record: JobEventRecord) -> JobEventRecord:
         await self._session.refresh(model)
         return _job_event_from_model(model)
 
+
+    async def get_max_sequence_no(self, job_id: str) -> int:
+        result = await self._session.execute(
+            select(func.coalesce(func.max(JobEventModel.sequence_no), 0))
+            .where(JobEventModel.job_id == job_id)
+        )
+        return result.scalar_one()
+
     async def list_for_job(
         self,
         job_id: str,

packages/parse/src/cortex_parse/jobs.py

@@ -33,9 +33,7 @@ async def submit(
         idempotency_key: str | None = None,
         request_id: str | None = None,
     ) -> JobAccepted:
-        normalized_key = (
-            normalize_idempotency_key(idempotency_key) if idempotency_key else None
-        )
+        normalized_key = normalize_idempotency_key(idempotency_key) if idempotency_key else None
         if normalized_key is not None:
             existing = await uow.jobs.get_by_idempotency(
                 caller.tenant_id,

workers/evaluation-worker/src/cortex_worker_evaluation/artifacts.py

@@ -2,6 +2,8 @@
 
 from cortex_evaluation import (
     EvaluationStorageCaller as WorkerStorageCaller,
+)
+from cortex_evaluation import (
     persist_evaluation_report,
 )