feat: updating minor features

app/api/v1/routes/call_imports.py

@@ -2975,6 +2975,46 @@ async def delete_call_import(
     )
 
 
+def _locate_call_import_row_or_404(
+    catalog_db: Session,
+    *,
+    call_import_id: UUID,
+    row_id: UUID,
+    organization_id: UUID,
+) -> Tuple[Session, CallImportRow, Optional[Session]]:
+    """Find a call import row on the correct DB session for mutation.
+
+    When sharding is enabled rows live on shard databases; ``get_db`` only
+    opens the catalog. Returns ``(row_db, row, extra_catalog_to_close)``
+    where ``extra_catalog_to_close`` is the catalog session opened by
+    :func:`locate_call_import_row` (distinct from the route's catalog
+    session) and must be closed via :func:`close_row_sessions`.
+    """
+    from app.db_sharding.row_ops import close_row_sessions, locate_call_import_row
+
+    try:
+        row_db, located_catalog, row, _shard_id = locate_call_import_row(row_id)
+    except LookupError:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Call import row not found",
+        ) from None
+    if (
+        row.call_import_id != call_import_id
+        or row.organization_id != organization_id
+    ):
+        close_row_sessions(
+            row_db,
+            located_catalog if located_catalog is not row_db else None,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Call import row not found",
+        )
+    extra_catalog = located_catalog if located_catalog is not row_db else None
+    return row_db, row, extra_catalog
+
+
 @router.delete(
     "/{call_import_id}/rows/{row_id}",
     status_code=status.HTTP_204_NO_CONTENT,
@@ -3009,38 +3049,35 @@ async def delete_call_import_row(
             detail="Call import not found",
         )
 
-    row = (
-        db.query(CallImportRow)
-        .filter(
-            CallImportRow.id == row_id,
-            CallImportRow.call_import_id == call_import.id,
-        )
-        .first()
-    )
-    if not row:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="Call import row not found",
-        )
+    from app.db_sharding.row_ops import close_row_sessions
 
-    _revoke_pending_tasks([row])
+    row_db, row, extra_catalog = _locate_call_import_row_or_404(
+        db,
+        call_import_id=call_import.id,
+        row_id=row_id,
+        organization_id=organization_id,
+    )
+    try:
+        _revoke_pending_tasks([row])
 
-    if row.recording_s3_key and s3_service.is_enabled():
-        try:
-            s3_service.delete_file_by_key(row.recording_s3_key)
-        except Exception as exc:  # noqa: BLE001 — best-effort, DB is source of truth
-            logger.warning(
-                "Failed to delete S3 object {} for row {}: {}",
-                row.recording_s3_key,
-                row.id,
-                exc,
-            )
+        if row.recording_s3_key and s3_service.is_enabled():
+            try:
+                s3_service.delete_file_by_key(row.recording_s3_key)
+            except Exception as exc:  # noqa: BLE001 — best-effort, DB is source of truth
+                logger.warning(
+                    "Failed to delete S3 object {} for row {}: {}",
+                    row.recording_s3_key,
+                    row.id,
+                    exc,
+                )
 
-    db.delete(row)
-    db.flush()
+        row_db.delete(row)
+        row_db.commit()
 
-    _recompute_call_import_counters(db, call_import)
-    db.commit()
+        _recompute_call_import_counters(db, call_import)
+        db.commit()
+    finally:
+        close_row_sessions(row_db, extra_catalog)
 
     logger.info(
         "Deleted call_import_row {} (call_import={}, org={})",
@@ -3529,24 +3566,21 @@ async def cancel_call_import_row_diarisation(
             detail="Call import not found",
         )
 
-    row = (
-        db.query(CallImportRow)
-        .filter(
-            CallImportRow.id == row_id,
-            CallImportRow.call_import_id == call_import_id,
-        )
-        .first()
-    )
-    if not row:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="Call import row not found",
-        )
+    from app.db_sharding.row_ops import close_row_sessions
 
-    _apply_diarisation_cancel([row])
-    db.commit()
-    db.refresh(row)
-    return CallImportRowResponse.model_validate(row)
+    row_db, row, extra_catalog = _locate_call_import_row_or_404(
+        db,
+        call_import_id=call_import_id,
+        row_id=row_id,
+        organization_id=organization_id,
+    )
+    try:
+        _apply_diarisation_cancel([row])
+        row_db.commit()
+        row_db.refresh(row)
+        return CallImportRowResponse.model_validate(row)
+    finally:
+        close_row_sessions(row_db, extra_catalog)
 
 
 @router.post(
@@ -3702,43 +3736,40 @@ async def toggle_call_import_row_speaker_swap(
             detail="Call import not found",
         )
 
-    row = (
-        db.query(CallImportRow)
-        .filter(
-            CallImportRow.id == row_id,
-            CallImportRow.call_import_id == call_import_id,
-            CallImportRow.organization_id == organization_id,
-        )
-        .first()
+    from app.db_sharding.row_ops import close_row_sessions
+
+    row_db, row, extra_catalog = _locate_call_import_row_or_404(
+        db,
+        call_import_id=call_import_id,
+        row_id=row_id,
+        organization_id=organization_id,
     )
-    if not row:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail="Call import row not found",
+    try:
+        segments = (
+            row.diarised_segments if isinstance(row.diarised_segments, list) else None
         )
+        if not segments:
+            # Without structured turns the swap toggle would have nothing to
+            # re-render — surface a clear error rather than silently
+            # flipping a flag the UI never read.
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail=(
+                    "This row has no structured diarised segments to swap. "
+                    "Re-run diarisation to generate per-speaker turns first."
+                ),
+            )
 
-    segments = row.diarised_segments if isinstance(row.diarised_segments, list) else None
-    if not segments:
-        # Without structured turns the swap toggle would have nothing to
-        # re-render — surface a clear error rather than silently
-        # flipping a flag the UI never read.
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail=(
-                "This row has no structured diarised segments to swap. "
-                "Re-run diarisation to generate per-speaker turns first."
-            ),
+        new_swap = not bool(row.diarised_speaker_swap)
+        row.diarised_speaker_swap = new_swap
+        row.diarised_transcript = (
+            _render_diarised_segments_text(segments, swap=new_swap) or None
         )
-
-    new_swap = not bool(row.diarised_speaker_swap)
-    row.diarised_speaker_swap = new_swap
-    row.diarised_transcript = (
-        _render_diarised_segments_text(segments, swap=new_swap) or None
-    )
-    db.commit()
-    db.refresh(row)
-
-    return CallImportRowResponse.model_validate(row)
+        row_db.commit()
+        row_db.refresh(row)
+        return CallImportRowResponse.model_validate(row)
+    finally:
+        close_row_sessions(row_db, extra_catalog)
 
 
 # ---------------------------------------------------------------------------

app/config/models.json

@@ -305,6 +305,11 @@
     "featured_rank": 7,
     "highlights": ["GA stable", "1M context", "Agentic coding"]
   },
+  "gemini-3.5-flash-lite": {
+    "provider": "google",
+    "model_type": "llm",
+    "description": "Gemini 3.5 Flash Lite \u2014 fastest, lowest-cost 3.5 model for high-throughput extraction, classification, and subagent workflows"
+  },
   "gemini-3.1-flash-lite": {
     "provider": "google",
     "model_type": "llm",

app/db_sharding/scatter_gather.py

@@ -779,6 +779,61 @@ def count_shard(db: Session, _shard_id: str) -> int:
     return total
 
 
+def _non_empty_production_transcript_filter():
+    """Rows whose CSV ``transcript`` column has non-whitespace content."""
+    return (
+        CallImportRow.transcript.isnot(None),
+        func.length(func.trim(CallImportRow.transcript)) > 0,
+    )
+
+
+def count_call_import_rows_with_production_transcript(
+    catalog_db: Session,
+    call_import_id: UUID,
+) -> int:
+    """Import rows with a non-empty production transcript (shard-aware)."""
+    transcript_filters = _non_empty_production_transcript_filter()
+
+    if not is_sharding_enabled():
+        return int(
+            catalog_db.query(func.count(CallImportRow.id))
+            .filter(
+                CallImportRow.call_import_id == call_import_id,
+                *transcript_filters,
+            )
+            .scalar()
+            or 0
+        )
+
+    total = 0
+    shard_ids = shard_ids_for_import(catalog_db, call_import_id)
+
+    def count_shard(db: Session, _shard_id: str) -> int:
+        return int(
+            db.query(func.count(CallImportRow.id))
+            .filter(
+                CallImportRow.call_import_id == call_import_id,
+                *transcript_filters,
+            )
+            .scalar()
+            or 0
+        )
+
+    for part in scatter_gather_on_shards(shard_ids, count_shard):
+        total += int(part)
+    if total == 0:
+        total = int(
+            catalog_db.query(func.count(CallImportRow.id))
+            .filter(
+                CallImportRow.call_import_id == call_import_id,
+                *transcript_filters,
+            )
+            .scalar()
+            or 0
+        )
+    return total
+
+
 def count_completed_call_import_rows(
     catalog_db: Session,
     call_import_id: UUID,

app/models/schemas.py

@@ -3000,11 +3000,10 @@ class CallImportEvaluationCreate(BaseModel):
         min_length=1,
         max_length=1,
         description=(
-            "Which transcript to score against. Only ``diarised`` is "
-            "supported — every evaluation run scores the diarised "
-            "transcript. Pass ``['diarised']`` explicitly or omit the "
-            "field to take the default; any other value (including the "
-            "legacy ``'production'`` source) is rejected with a 400."
+            "Which transcript to score against. ``'diarised'`` (default) "
+            "auto-diarises rows missing a diarised transcript then scores "
+            "``diarised_transcript``. ``'production'`` scores the CSV "
+            "``transcript`` column directly and skips diarisation."
         ),
     )
 
@@ -3013,20 +3012,16 @@ class CallImportEvaluationCreate(BaseModel):
     def _validate_transcript_sources(
         cls, value: List[str]
     ) -> List["CallImportEvaluationTranscriptSource"]:
-        # The Field min_length/max_length constraints catch empty +
-        # over-long payloads; this validator's job is to reject any
-        # non-diarised source value and normalize the result to
-        # ``['diarised']`` for downstream code.
-        invalid = [src for src in value if src != "diarised"]
+        allowed = {"production", "diarised"}
+        invalid = [src for src in value if src not in allowed]
         if invalid:
             raise ValueError(
-                "Only the 'diarised' transcript source is supported "
+                "transcript_sources must be ['production'] or ['diarised'] "
                 "(received: "
                 + ", ".join(repr(src) for src in invalid)
-                + "). Remove 'production' from transcript_sources or "
-                "omit the field to use the default."
+                + ")."
             )
-        return ["diarised"]
+        return value  # type: ignore[return-value]
     # --- Run-level LLM config ---
     llm_provider: Optional[str] = Field(
         default=None,

app/services/call_import_user_insights.py

@@ -143,9 +143,12 @@ def _pick_transcript(
     evaluation: CallImportEvaluation,
     source_row: CallImportRow,
 ) -> str:
-    del evaluation
-    diarised = (source_row.diarised_transcript or "").strip()
-    return diarised[:ROW_TRANSCRIPT_CHAR_CAP]
+    source = (evaluation.transcript_source or "diarised").strip().lower()
+    if source == "production":
+        text = (source_row.transcript or "").strip()
+    else:
+        text = (source_row.diarised_transcript or "").strip()
+    return text[:ROW_TRANSCRIPT_CHAR_CAP]
 
 
 def _metric_name_map(metrics: Sequence[Metric]) -> Dict[str, str]:

app/services/call_imports/bulk_ops.py

@@ -199,6 +199,16 @@ def count_completed_source_rows(db: Session, call_import_id: UUID) -> int:
     return count_completed_call_import_rows(db, call_import_id)
 
 
+def count_source_rows_with_production_transcript(
+    db: Session, call_import_id: UUID
+) -> int:
+    from app.db_sharding.scatter_gather import (
+        count_call_import_rows_with_production_transcript,
+    )
+
+    return count_call_import_rows_with_production_transcript(db, call_import_id)
+
+
 def _completed_source_row_ids(db: Session, call_import_id: UUID) -> List[UUID]:
     from app.db_sharding.scatter_gather import list_completed_source_row_ids_ordered
 
@@ -670,7 +680,19 @@ def execute_call_import_materialization(
         return {"total_rows": 0, "status": "failed"}
 
     try:
+        logger.info(
+            "execute_call_import_materialization downloading staged source "
+            "(call_import={} key={})",
+            call_import_id,
+            call_import.source_s3_key,
+        )
         file_bytes = s3_service.download_file_by_key(call_import.source_s3_key)
+        logger.info(
+            "execute_call_import_materialization downloaded {} bytes "
+            "(call_import={})",
+            len(file_bytes),
+            call_import_id,
+        )
     except StorageError as exc:
         call_import.status = CallImportStatus.FAILED
         call_import.error_message = f"Could not read staged source file from S3: {exc}"