diff --git a/app/api/v1/routes/call_imports.py b/app/api/v1/routes/call_imports.py index 6e59f992..20cd2b16 100644 --- a/app/api/v1/routes/call_imports.py +++ b/app/api/v1/routes/call_imports.py @@ -14,6 +14,7 @@ import io import json import re +from dataclasses import dataclass, field from datetime import date, datetime, time, timedelta from typing import Any, Dict, Iterable, List, Optional, Tuple from uuid import UUID, uuid4 @@ -87,6 +88,33 @@ ) +@dataclass(frozen=True) +class CallImportParseSkip: + """One source row excluded during CSV/Excel parse (identity / recording URL).""" + + source_row: int + reason: str + message: str + + +@dataclass +class CallImportParseResult: + rows: List[Dict[str, Any]] = field(default_factory=list) + skipped: List[CallImportParseSkip] = field(default_factory=list) + + +def parse_skips_to_json(skips: List[CallImportParseSkip]) -> List[Dict[str, Any]]: + """Persistable JSON shape for ``CallImport.source_row_skips``.""" + return [ + { + "source_row": item.source_row, + "reason": item.reason, + "message": item.message, + } + for item in skips + ] + + def _normalize_dataset(raw: Optional[str]) -> Optional[str]: """Trim and treat empty strings as 'no dataset' (NULL).""" if raw is None: @@ -415,6 +443,14 @@ def _coerce_parameter_value( return cell +def _recording_url_cell_is_valid_http(raw: str) -> bool: + cell = (raw or "").strip() + if not cell: + return False + lower = cell.lower() + return lower.startswith("http://") or lower.startswith("https://") + + def _parameter_is_required(param: CallImportSchemaParameter) -> bool: """Return whether a schema parameter must be mapped on every upload.""" if param.is_required: @@ -438,7 +474,7 @@ def _apply_schema_mapping( *, source_label: str = "CSV", validate_only: bool = False, -) -> List[Dict[str, Any]]: +) -> CallImportParseResult: """Schema-driven row projection: parameter -> CSV header -> typed value. Validates that every required schema parameter is mapped to a CSV @@ -552,9 +588,10 @@ def _apply_schema_mapping( # run; the row loop only matters at IMPORT time. Skip it (and # the "no data rows" guard at the bottom of the function) so # the caller gets a clean pass when the mapping is shaped right. - return [] + return CallImportParseResult() parsed: List[Dict[str, Any]] = [] + skipped: List[CallImportParseSkip] = [] for idx, row in enumerate(rows_iter): # Drop fully-blank lines - matches the legacy parser behavior so # trailing-newline edge cases don't fail an otherwise-good upload. @@ -566,20 +603,59 @@ def _apply_schema_mapping( if not non_blank: continue + source_row = idx + 1 conv_value = (row.get(conv_canonical) or "").strip() if conv_canonical else "" if not conv_value: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=( - f"Row {idx + 1} is missing the '{conv_param.name}' " - "(conversation_id) value." - ), + skipped.append( + CallImportParseSkip( + source_row=source_row, + reason="missing_conversation_id", + message=( + f"Row {source_row} is missing the '{conv_param.name}' " + "(conversation_id) value." + ), + ) + ) + continue + + if rec_canonical and rec_url_param_name: + rec_param = next( + (p for p in parameters if p.name == rec_url_param_name), + None, ) + if rec_param is not None and _parameter_is_required(rec_param): + rec_raw = (row.get(rec_canonical) or "").strip() + if not rec_raw: + skipped.append( + CallImportParseSkip( + source_row=source_row, + reason="missing_recording_url", + message=( + f"Row {source_row} is missing the required " + f"'{rec_url_param_name}' value." + ), + ) + ) + continue + if not _recording_url_cell_is_valid_http(rec_raw): + skipped.append( + CallImportParseSkip( + source_row=source_row, + reason="invalid_recording_url", + message=( + f"Row {source_row}: value for " + f"'{rec_url_param_name}' is not a valid recording " + "URL (must start with http:// or https://)." + ), + ) + ) + continue # Materialize every mapped parameter into the per-row snapshot, # running per-type coercion so a bad cell aborts the upload # rather than silently storing garbage. parameter_values: Dict[str, Any] = {} + row_skipped = False for param in parameters: canonical = canonical_by_param[param.name] if canonical is None: @@ -595,14 +671,29 @@ def _apply_schema_mapping( param_name=param.name, ) if _parameter_is_required(param) and coerced is None: + if param_type == CallImportParameterType.RECORDING_URL: + skipped.append( + CallImportParseSkip( + source_row=source_row, + reason="missing_recording_url", + message=( + f"Row {source_row} is missing the required " + f"'{param.name}' value." + ), + ) + ) + row_skipped = True + break raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( - f"Row {idx + 1} is missing the required " + f"Row {source_row} is missing the required " f"'{param.name}' value." ), ) parameter_values[param.name] = coerced + if row_skipped: + continue rec_value = ( (row.get(rec_canonical) or "").strip() if rec_canonical else "" @@ -628,13 +719,33 @@ def _apply_schema_mapping( } ) - if not parsed: + if not parsed and not skipped: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"{source_label} did not contain any data rows.", ) - return parsed + return CallImportParseResult(rows=parsed, skipped=skipped) + + +def _raise_if_no_importable_rows( + result: CallImportParseResult, *, source_label: str = "CSV" +) -> None: + """Sync upload / API callers fail fast when every data row was skipped.""" + if result.rows: + return + if result.skipped: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + f"No importable rows. {len(result.skipped)} row(s) skipped due " + "to missing or invalid conversation ID or recording URL." + ), + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"{source_label} did not contain any data rows.", + ) def _parse_csv( @@ -642,7 +753,7 @@ def _parse_csv( parameters: List[CallImportSchemaParameter], parameter_mapping: Dict[str, str], skipped_columns: List[str], -) -> List[Dict[str, Any]]: +) -> CallImportParseResult: """Parse a CSV file using the resolved schema parameters.""" if not file_bytes: raise HTTPException( @@ -778,7 +889,7 @@ def _parse_xlsx( parameters: List[CallImportSchemaParameter], parameter_mapping: Dict[str, str], skipped_columns: List[str], -) -> List[Dict[str, Any]]: +) -> CallImportParseResult: """Parse a single worksheet from an xlsx/xlsm workbook. ``sheet_name`` must match one of the workbook's sheets (case @@ -1148,7 +1259,7 @@ def _parse_source_file( parameters: List[CallImportSchemaParameter], cleaned_mapping: Dict[str, str], cleaned_skipped: List[str], -) -> List[Dict[str, Any]]: +) -> CallImportParseResult: """Run the format-appropriate parser against a buffer of file bytes.""" if fmt == "csv": return _parse_csv(file_bytes, parameters, cleaned_mapping, cleaned_skipped) @@ -1913,6 +2024,7 @@ async def upload_call_import_csv( parsed_rows = _parse_source_file( file_bytes, fmt, sheet_name_clean, parameters, cleaned_mapping, cleaned_skipped ) + _raise_if_no_importable_rows(parsed_rows, source_label=fmt) tag_rows = _resolve_tags(db, organization_id, tag_ids) @@ -1932,10 +2044,11 @@ async def upload_call_import_csv( column_mapping={}, extra_columns=[], custom_column_mapping={}, - total_rows=len(parsed_rows), + total_rows=len(parsed_rows.rows), completed_rows=0, failed_rows=0, status=CallImportStatus.PENDING, + source_row_skips=parse_skips_to_json(parsed_rows.skipped), ) if tag_rows: call_import.tags = tag_rows @@ -1947,7 +2060,7 @@ async def upload_call_import_csv( call_import.provider = None row_models = _materialize_rows( - db, call_import, parsed_rows, organization_id + db, call_import, parsed_rows.rows, organization_id ) call_import.status = CallImportStatus.PROCESSING diff --git a/app/api/v1/routes/metrics.py b/app/api/v1/routes/metrics.py index fcba6b8c..760580b5 100644 --- a/app/api/v1/routes/metrics.py +++ b/app/api/v1/routes/metrics.py @@ -22,6 +22,7 @@ MetricResponse, PromoteDiscoveredChildRequest, PromoteDiscoveredMetricRequest, + METRIC_RUBRIC_TEXT_MAX_LENGTH, ) router = APIRouter(prefix="/metrics", tags=["metrics"]) @@ -1718,7 +1719,7 @@ class MetricParseBulkRequest(BaseModel): ) parent_description: Optional[str] = Field( default=None, - max_length=4000, + max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH, description="Optional description used as the parent's LLM rubric.", ) selection_mode: Optional[Literal["single_choice", "multi_label"]] = Field( @@ -1830,7 +1831,7 @@ def _parse_label_blocks(prompt: str) -> List[ParsedLabel]: ParsedLabel( label_name=name[:120], definition=definition[:2000], - examples=examples[:4000], + examples=examples[:METRIC_RUBRIC_TEXT_MAX_LENGTH], ) ) return labels @@ -1908,7 +1909,7 @@ def _llm_parse_labels( ParsedLabel( label_name=name[:120], definition=str(item.get("definition") or "")[:2000].strip(), - examples=str(item.get("examples") or "")[:4000].strip(), + examples=str(item.get("examples") or "")[:METRIC_RUBRIC_TEXT_MAX_LENGTH].strip(), ) ) return labels @@ -1930,7 +1931,7 @@ def _build_description_from_label(label: ParsedLabel) -> str: if label.examples: parts.append(f"Examples:\n{label.examples}") description = "\n\n".join(parts) - return description[:4000] + return description[:METRIC_RUBRIC_TEXT_MAX_LENGTH] def _ensure_unique_metric_name( diff --git a/app/migrations/054_call_import_source_row_skips.py b/app/migrations/054_call_import_source_row_skips.py new file mode 100644 index 00000000..b0e5d64b --- /dev/null +++ b/app/migrations/054_call_import_source_row_skips.py @@ -0,0 +1,42 @@ +""" +Migration: Persist parse-time skip summary on call imports. + +Adds ``source_row_skips`` JSON on ``call_imports`` for rows excluded +during CSV/Excel materialization (missing/invalid conversation ID or URL). +""" + +from sqlalchemy import text +from sqlalchemy.orm import Session + +description = "Add call_imports.source_row_skips for parse-time row skip summary." + + +def _column_exists(db: Session, table_name: str, column_name: str) -> bool: + row = db.execute( + text( + """ + SELECT 1 + FROM information_schema.columns + WHERE table_name = :table_name AND column_name = :column_name + """ + ), + {"table_name": table_name, "column_name": column_name}, + ).first() + return row is not None + + +def upgrade(db: Session): + if not _column_exists(db, "call_imports", "source_row_skips"): + db.execute( + text( + """ + ALTER TABLE call_imports + ADD COLUMN source_row_skips JSON NOT NULL DEFAULT '[]'::json + """ + ) + ) + + +def downgrade(db: Session): + if _column_exists(db, "call_imports", "source_row_skips"): + db.execute(text("ALTER TABLE call_imports DROP COLUMN source_row_skips")) diff --git a/app/models/database.py b/app/models/database.py index 7e3466d5..efe88a16 100644 --- a/app/models/database.py +++ b/app/models/database.py @@ -1810,6 +1810,9 @@ class CallImport(Base): # persisted so the IMPORT stage can re-parse the file with the same # mapping/skip intent. skipped_columns = Column(JSON, nullable=False, default=list) + # Rows skipped at parse time (missing/invalid conversation_id or URL). + # Shape: ``[{"source_row": int, "reason": str, "message": str}, ...]``. + source_row_skips = Column(JSON, nullable=False, default=list) # Free-text high-level segregation label. Powers the "Dataset" filter # at the top of the imports page; multiple imports can share a value. diff --git a/app/models/schemas.py b/app/models/schemas.py index 5f1dc4c2..59c3ced2 100644 --- a/app/models/schemas.py +++ b/app/models/schemas.py @@ -1363,6 +1363,10 @@ class RunEvaluatorsResponse(BaseModel): MetricScope = Literal["workspace", "organization"] +# Max length for metric rubric text (description / example) accepted by +# the API. DB columns are TEXT or unbounded VARCHAR; this cap is validation-only. +METRIC_RUBRIC_TEXT_MAX_LENGTH = 32_000 + class MetricCreate(BaseModel): """Schema for creating a metric. @@ -1387,7 +1391,7 @@ class MetricCreate(BaseModel): # in the LLM judge's rubric. Today this is mainly populated on # child sub-labels (one example per categorization label) but # standalone metrics may carry it too without a schema change. - example: Optional[str] = Field(default=None, max_length=4000) + example: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) metric_type: MetricType = MetricType.RATING metric_category: MetricCategory = MetricCategory.QUALITY trigger: MetricTrigger = MetricTrigger.ALWAYS @@ -1467,11 +1471,11 @@ class MetricChildDraft(BaseModel): """One child sub-metric in a parent + children atomic create body.""" name: str = Field(..., max_length=120) - description: Optional[str] = Field(default=None, max_length=4000) + description: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) # Optional illustrative example for this label. Surfaced alongside # ``description`` in the LLM judge's rubric so each label can carry # both its definition AND a "what does this look like?" example. - example: Optional[str] = Field(default=None, max_length=4000) + example: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) enabled: bool = True capture_rationale: Optional[bool] = True tags: Optional[List[str]] = None @@ -1487,7 +1491,7 @@ class MetricCreateWithChildren(BaseModel): """ name: str = Field(..., max_length=120) - description: Optional[str] = Field(default=None, max_length=4000) + description: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) selection_mode: SelectionMode metric_category: MetricCategory = MetricCategory.QUALITY enabled: bool = True @@ -1517,10 +1521,10 @@ class MetricCreateWithChildren(BaseModel): class MetricUpdate(BaseModel): """Schema for updating a metric.""" name: Optional[str] = None - description: Optional[str] = None + description: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) # ``None`` here means "leave unchanged"; pass an empty string to # clear a previously stored example. - example: Optional[str] = Field(default=None, max_length=4000) + example: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) metric_type: Optional[MetricType] = None trigger: Optional[MetricTrigger] = None enabled: Optional[bool] = None @@ -2639,6 +2643,26 @@ class CallImportPreviewSheet(BaseModel): ) +class CallImportSourceRowSkip(BaseModel): + """One source spreadsheet row skipped during parse (identity / recording URL).""" + + source_row: int = Field( + ..., + description="1-based row index in the source file (same semantics as parse errors).", + ) + reason: str = Field( + ..., + description=( + "Machine-readable skip reason, e.g. missing_conversation_id, " + "missing_recording_url, invalid_recording_url." + ), + ) + message: str = Field( + ..., + description="Human-readable explanation shown in the UI.", + ) + + class CallImportResponse(BaseModel): """Summary of a call-import batch.""" @@ -2664,6 +2688,13 @@ class CallImportResponse(BaseModel): # Persisted "drop these columns" decision captured at MAP time. # Empty for legacy one-shot uploads where the value was ephemeral. skipped_columns: List[str] = Field(default_factory=list) + source_row_skips: List[CallImportSourceRowSkip] = Field( + default_factory=list, + description=( + "Source rows skipped at parse time because of missing/invalid " + "conversation ID or recording URL." + ), + ) # Source-file staging fields populated at UPLOAD time. ``None`` on # legacy batches imported via the one-shot ``POST /upload`` endpoint. source_s3_key: Optional[str] = None @@ -4550,7 +4581,7 @@ class PromoteDiscoveredChildRequest(BaseModel): key: str = Field(..., min_length=1, max_length=120) name: str = Field(..., min_length=1, max_length=120) - description: Optional[str] = Field(default=None, max_length=4000) + description: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) # Default True: when promoting a discovered label we want the new # sub-metric to always capture rationales going forward, since the # candidate was itself proposed *with* a rationale and the user @@ -4639,7 +4670,7 @@ class PromoteDiscoveredMetricRequest(BaseModel): key: str = Field(..., min_length=1, max_length=120) name: str = Field(..., min_length=1, max_length=120) - description: Optional[str] = Field(default=None, max_length=4000) + description: Optional[str] = Field(default=None, max_length=METRIC_RUBRIC_TEXT_MAX_LENGTH) metric_type: DiscoveredMetricSuggestedType = "boolean" capture_rationale: bool = True # Optional per-type config knobs passed through to ``Metric.custom_config``. diff --git a/app/services/call_imports/bulk_ops.py b/app/services/call_imports/bulk_ops.py index 96be5883..cbed6ceb 100644 --- a/app/services/call_imports/bulk_ops.py +++ b/app/services/call_imports/bulk_ops.py @@ -12,6 +12,7 @@ from uuid import UUID, uuid4 from loguru import logger +from fastapi import HTTPException from sqlalchemy.orm import Session, load_only from sqlalchemy import func @@ -598,6 +599,7 @@ def execute_call_import_materialization( _enqueue_row_tasks, _parse_source_file, _resolve_schema, + parse_skips_to_json, ) from app.models.enums import CallImportStatus from app.services.billing.flexprice_service import record_call_import_batch_created @@ -680,20 +682,49 @@ def execute_call_import_materialization( ) parameters = list(schema.parameters) cleaned_skipped = _clean_skipped_columns(list(call_import.skipped_columns or [])) - parsed_rows = _parse_source_file( - file_bytes, - call_import.source_format, - call_import.sheet_name, - parameters, - dict(call_import.parameter_mapping or {}), - cleaned_skipped, - ) + try: + parse_result = _parse_source_file( + file_bytes, + call_import.source_format, + call_import.sheet_name, + parameters, + dict(call_import.parameter_mapping or {}), + cleaned_skipped, + ) + except HTTPException as exc: + call_import.status = CallImportStatus.FAILED + call_import.error_message = ( + exc.detail if isinstance(exc.detail, str) else str(exc.detail) + ) + call_import.source_row_skips = [] + db.commit() + return {"total_rows": 0, "status": "failed"} + + call_import.source_row_skips = parse_skips_to_json(parse_result.skipped) + skipped_count = len(parse_result.skipped) - call_import.total_rows = len(parsed_rows) + if not parse_result.rows: + if parse_result.skipped: + call_import.error_message = ( + f"No importable rows. {skipped_count} row(s) skipped due " + "to missing or invalid conversation ID or recording URL." + ) + else: + call_import.error_message = "Source file did not contain any data rows." + call_import.status = CallImportStatus.FAILED + call_import.total_rows = 0 + db.commit() + return { + "total_rows": 0, + "status": "failed", + "source_rows_skipped": skipped_count, + } + + call_import.total_rows = len(parse_result.rows) call_import.completed_rows = 0 call_import.failed_rows = 0 - row_count = len(parsed_rows) + row_count = len(parse_result.rows) pending_shard_sessions: List[Session] = [] try: if is_sharding_enabled(): @@ -704,7 +735,7 @@ def execute_call_import_materialization( row_count, pending_shard_sessions = bulk_materialize_call_import_rows( db, call_import, - parsed_rows, + parse_result.rows, organization_id, defer_shard_commit=is_sharding_enabled(), ) @@ -750,7 +781,11 @@ def execute_call_import_materialization( call_import.id, organization_id, ) - return {"total_rows": row_count, "status": "processing"} + return { + "total_rows": row_count, + "status": "processing", + "source_rows_skipped": skipped_count, + } def _aggregate_import_row_status_counts( diff --git a/app/workers/tasks/helpers/llm_diarisation.py b/app/workers/tasks/helpers/llm_diarisation.py index 088c8ef1..7156faff 100644 --- a/app/workers/tasks/helpers/llm_diarisation.py +++ b/app/workers/tasks/helpers/llm_diarisation.py @@ -137,13 +137,42 @@ # turn. Without an explicit ``max_tokens`` many providers default to a # low cap (e.g. 4096) and truncate mid-JSON on longer calls. _DIARISATION_MIN_MAX_TOKENS = 4096 -_DIARISATION_MAX_MAX_TOKENS = 16_384 +# Default output ceiling for models without an explicit entry below. +_DIARISATION_DEFAULT_MAX_MAX_TOKENS = 16_384 +# Backward-compatible alias (tests and legacy references). +_DIARISATION_MAX_MAX_TOKENS = _DIARISATION_DEFAULT_MAX_MAX_TOKENS + +# Substrings matched against normalised ``llm_model`` names (LiteLLM / +# provider prefixes like ``gemini/gemini-2.5-flash`` included). Longer +# needles should appear before shorter shared prefixes. +_MODEL_OUTPUT_TOKEN_CEILINGS: tuple[tuple[str, int], ...] = ( + ("gemini-2.5-flash", 65_536), + ("gemini-2.5-pro", 65_536), + ("gemini-2.0-flash", 65_536), + ("gemini-2.0-pro", 65_536), + ("gemini-1.5-pro", 65_536), + ("gemini-1.5-flash", 65_536), +) + + +def _diarisation_output_token_ceiling(llm_model: str) -> int: + """Provider/model-specific max ``max_tokens`` for diariser completions.""" + name = (llm_model or "").strip().lower() + if name: + for needle, ceiling in _MODEL_OUTPUT_TOKEN_CEILINGS: + if needle in name: + return ceiling + return _DIARISATION_DEFAULT_MAX_MAX_TOKENS def _estimate_diarisation_max_tokens( - *, text_length: int = 0, audio_bytes: int = 0 + *, + text_length: int = 0, + audio_bytes: int = 0, + llm_model: str = "", ) -> int: """Scale the diariser output budget to the input payload size.""" + ceiling = _diarisation_output_token_ceiling(llm_model) if text_length > 0: # ~0.45 tokens/char covers JSON keys + per-turn wrapping. estimated = int(text_length * 0.45) + 512 @@ -153,10 +182,7 @@ def _estimate_diarisation_max_tokens( estimated = int(estimated_chars * 0.45) + 512 else: estimated = _DIARISATION_MIN_MAX_TOKENS - return min( - _DIARISATION_MAX_MAX_TOKENS, - max(_DIARISATION_MIN_MAX_TOKENS, estimated), - ) + return min(ceiling, max(_DIARISATION_MIN_MAX_TOKENS, estimated)) def _generate_diarisation_response( @@ -173,11 +199,13 @@ def _generate_diarisation_response( audio_bytes: int = 0, ) -> Dict[str, Any]: """Call the diariser LLM with a scaled ``max_tokens`` and one retry.""" + ceiling = _diarisation_output_token_ceiling(llm_model) base_budget = _estimate_diarisation_max_tokens( text_length=content_length, audio_bytes=audio_bytes, + llm_model=llm_model, ) - retry_budget = min(_DIARISATION_MAX_MAX_TOKENS, base_budget * 2) + retry_budget = min(ceiling, base_budget * 2) budgets = [base_budget] if retry_budget <= base_budget else [base_budget, retry_budget] response: Dict[str, Any] = {} @@ -602,6 +630,14 @@ def _parse_turns_from_response(response: Dict[str, Any]) -> List[Dict[str, Any]] f"{prefix}{snippet}{'…' if len(raw_text) > 400 else ''}" ) + if response.get("truncated"): + raise LLMDiarisationError( + "LLM diariser response was truncated by max_tokens (the " + "transcript was cut off mid-call). Switch to STT + LLM " + "diariser mode, pick a model with a larger output limit, " + "or shorten the diarisation prompt." + ) + turns: List[Dict[str, Any]] = [] cursor: float = 0.0 for entry in parsed: diff --git a/frontend/src/pages/callImports/CallImportDetail.tsx b/frontend/src/pages/callImports/CallImportDetail.tsx index 41ae5a7c..86a27dcb 100644 --- a/frontend/src/pages/callImports/CallImportDetail.tsx +++ b/frontend/src/pages/callImports/CallImportDetail.tsx @@ -1932,6 +1932,36 @@ export default function CallImportDetail() { )} + + {(data.source_row_skips?.length ?? 0) > 0 && data.total_rows > 0 && ( +
+
+ +
+

+ Imported {data.total_rows} row + {data.total_rows === 1 ? '' : 's'}.{' '} + {data.source_row_skips!.length} source row + {data.source_row_skips!.length === 1 ? '' : 's'} skipped + — missing or invalid conversation ID / recording URL. +

+
    + {data.source_row_skips!.slice(0, 50).map((skip) => ( +
  • + {skip.message} +
    • + ))} +
+ {data.source_row_skips!.length > 50 && ( +

+ and {data.source_row_skips!.length - 50} more skipped row + {data.source_row_skips!.length - 50 === 1 ? '' : 's'}. +

+ )} +
+
+
+ )} {/* Stage tracker: only meaningful for batches that came through @@ -3736,8 +3766,8 @@ export default function CallImportDetail() {

{transcribeMode === 'llm_only' - ? 'Recommended. Single-stage pipeline: the audio is fed directly to a multimodal LLM along with your prompt; the model both transcribes and diarises in one call. Pick a model that accepts audio input (e.g. Gemini 1.5/2.0, GPT-4o audio-preview).' - : 'Advanced fallback. Two-stage pipeline: STT transcribes the audio, then an LLM splits it into agent / user turns using your prompt. Use this when you need a specific STT contract or to reuse an existing transcript artefact.'} + ? 'Recommended. Single-stage pipeline: the audio is fed directly to a multimodal LLM along with your prompt; the model both transcribes and diarises in one call. Pick a model that accepts audio input (e.g. Gemini 1.5/2.0, GPT-4o audio-preview). Recordings longer than about 8 minutes depend on the model’s output token limit—if diarisation fails or looks cut off, switch to STT + LLM diariser.' + : 'Advanced fallback. Two-stage pipeline: STT transcribes the audio, then an LLM splits it into agent / user turns using your prompt. Use this when you need a specific STT contract, longer recordings, or to reuse an existing transcript artefact.'}

{transcribeMode === 'stt_llm' && ( @@ -4473,7 +4503,7 @@ export default function CallImportDetail() { {evalTranscribeMode === 'stt_llm' ? "Every evaluation scores the diarised transcript. Rows that don't already have one are diarised first via the STT provider you pick below." - : "Every evaluation scores the diarised transcript. Rows that don't already have one are diarised by feeding the audio directly to the multimodal LLM you pick below."} + : "Every evaluation scores the diarised transcript. Rows that don't already have one are diarised by feeding the audio directly to the multimodal LLM you pick below. Calls longer than about 8 minutes need enough output headroom—if results look cut off, use STT + LLM diariser instead."} diff --git a/frontend/src/pages/callImports/components/ImportPanel.tsx b/frontend/src/pages/callImports/components/ImportPanel.tsx index 1147d265..7efd1ce2 100644 --- a/frontend/src/pages/callImports/components/ImportPanel.tsx +++ b/frontend/src/pages/callImports/components/ImportPanel.tsx @@ -63,7 +63,7 @@ export default function ImportPanel({ callImport }: ImportPanelProps) { const helperText = useMemo(() => { if (selectedIntegrationId === DIRECT_URL_CREDENTIAL) { - return 'Recordings will be downloaded directly from the recording URL column mapped in your schema. Each row must include a valid URL.' + return 'Recordings are downloaded from the recording URL column. Rows with a missing or invalid URL are skipped; the rest import normally.' } return 'Pick the telephony provider + credential to use when fetching recordings for this batch. Once started, the per-row workers pick up automatically and progress is reflected on this page.' }, [selectedIntegrationId]) diff --git a/frontend/src/types/api.ts b/frontend/src/types/api.ts index 14106744..3820aaac 100644 --- a/frontend/src/types/api.ts +++ b/frontend/src/types/api.ts @@ -985,6 +985,12 @@ export interface WorkspaceRoleUpdate { capabilities?: string[] } +export interface CallImportSourceRowSkip { + source_row: number + reason: string + message: string +} + export interface CallImport { id: string organization_id: string @@ -1032,6 +1038,10 @@ export interface CallImport { * ephemeral. */ skipped_columns: string[] + /** + * Source rows skipped at parse time (missing/invalid conversation ID or URL). + */ + source_row_skips?: CallImportSourceRowSkip[] /** S3 key for the staged source file. ``null`` on legacy batches. */ source_s3_key: string | null /** ``'csv'`` / ``'xlsx'`` for staged files, or ``'audio'`` for manual uploads. */ diff --git a/tests/test_api/test_call_import_diarization_and_eval_llm.py b/tests/test_api/test_call_import_diarization_and_eval_llm.py index 5bad31ef..0f92599e 100644 --- a/tests/test_api/test_call_import_diarization_and_eval_llm.py +++ b/tests/test_api/test_call_import_diarization_and_eval_llm.py @@ -1341,6 +1341,44 @@ def test_estimate_diarisation_max_tokens_scales_with_transcript(): assert long <= llm_diarisation._DIARISATION_MAX_MAX_TOKENS +def test_estimate_diarisation_max_tokens_gemini_long_audio(): + from app.workers.tasks.helpers import llm_diarisation + + # ~15 min mono at the worker's byte heuristic (~320 KiB/min). + fifteen_min_bytes = 15 * 320_000 + budget = llm_diarisation._estimate_diarisation_max_tokens( + audio_bytes=fifteen_min_bytes, + llm_model="gemini-2.5-flash", + ) + assert budget > llm_diarisation._DIARISATION_MAX_MAX_TOKENS + assert budget <= 65_536 + + +def test_diarisation_output_token_ceiling_gemini(): + from app.workers.tasks.helpers import llm_diarisation + + assert ( + llm_diarisation._diarisation_output_token_ceiling("gemini/gemini-2.5-flash") + == 65_536 + ) + assert ( + llm_diarisation._diarisation_output_token_ceiling("gpt-4o-mini") + == llm_diarisation._DIARISATION_MAX_MAX_TOKENS + ) + + +def test_parse_turns_rejects_truncated_response_with_repairable_json(): + from app.workers.tasks.helpers import llm_diarisation + + response = { + "text": '{"turns": [{"speaker": "agent", "text": "Hello"}, {"speaker": "user", "text": "partial', + "truncated": True, + } + with pytest.raises(llm_diarisation.LLMDiarisationError) as exc_info: + llm_diarisation._parse_turns_from_response(response) + assert "truncated" in str(exc_info.value).lower() + + def test_diariser_passes_scaled_max_tokens(monkeypatch): from app.workers.tasks.helpers import llm_diarisation diff --git a/tests/test_api/test_call_imports_routes.py b/tests/test_api/test_call_imports_routes.py index 9ae12f78..8ad478ad 100644 --- a/tests/test_api/test_call_imports_routes.py +++ b/tests/test_api/test_call_imports_routes.py @@ -111,13 +111,21 @@ def _standard_skipped() -> list[str]: return [] +def _parse_csv_rows(*args, **kwargs): + return _parse_csv(*args, **kwargs).rows + + +def _parse_xlsx_rows(*args, **kwargs): + return _parse_xlsx(*args, **kwargs).rows + + def test_parse_csv_accepts_canonical_headers(): csv_text = ( "CallID,Recording Date,Recording URL,Transcript\n" "abc-1,18/05/2026,https://api.exotel.com/recordings/abc-1.mp3,Hello world\n" "abc-2,19/05/2026,https://api.exotel.com/recordings/abc-2.mp3,Another call\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -146,7 +154,7 @@ def test_parse_csv_accepts_flexible_day_first_recording_dates(raw_date, normaliz "CallID,Recording Date,Recording URL,Transcript\n" f"abc-1,{raw_date},https://api.exotel.com/recordings/abc-1.mp3,Hello world\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -160,7 +168,7 @@ def test_parse_csv_is_case_insensitive_on_headers(): "callid,recording date,recording url,TRANSCRIPT\n" "id-1,18/05/2026,https://x/recording.mp3,Some transcript\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -216,20 +224,99 @@ def test_parse_csv_rejects_unmapped_required_recording_url(): assert "recording_url" in exc.value.detail.lower() -def test_parse_csv_rejects_row_missing_conversation_id(): +def test_parse_csv_skips_row_missing_conversation_id(): csv_text = ( "CallID,Recording Date,Recording URL,Transcript\n" - ",18/05/2026,https://x/recording.mp3,Some transcript\n" + "good-1,18/05/2026,https://x/recording.mp3,Some transcript\n" + ",19/05/2026,https://x/2.mp3,Missing conv id\n" ) - with pytest.raises(HTTPException) as exc: - _parse_csv( - _csv_bytes(csv_text), - _standard_params(), - _standard_mapping(), - _standard_skipped(), + result = _parse_csv( + _csv_bytes(csv_text), + _standard_params(), + _standard_mapping(), + _standard_skipped(), + ) + assert len(result.rows) == 1 + assert result.rows[0]["conversation_id"] == "good-1" + assert len(result.skipped) == 1 + assert result.skipped[0].reason == "missing_conversation_id" + assert result.skipped[0].source_row == 2 + + +def test_parse_csv_skips_row_missing_recording_url(): + csv_text = ( + "CallID,Recording Date,Recording URL,Transcript\n" + "abc-1,18/05/2026,https://x/recording.mp3,Some transcript\n" + "abc-2,19/05/2026,,No URL\n" + ) + result = _parse_csv( + _csv_bytes(csv_text), + _standard_params(), + _standard_mapping(), + _standard_skipped(), + ) + assert len(result.rows) == 1 + assert len(result.skipped) == 1 + assert result.skipped[0].reason == "missing_recording_url" + assert result.skipped[0].source_row == 2 + + +def test_parse_csv_skips_row_invalid_recording_url(): + csv_text = ( + "CallID,Recording Date,Recording URL,Transcript\n" + "abc-1,18/05/2026,ftp://bad/recording.mp3,Some transcript\n" + ) + result = _parse_csv( + _csv_bytes(csv_text), + _standard_params(), + _standard_mapping(), + _standard_skipped(), + ) + assert len(result.rows) == 0 + assert len(result.skipped) == 1 + assert result.skipped[0].reason == "invalid_recording_url" + + +def test_parse_csv_skips_multiple_bad_rows(): + lines = ["CallID,Recording Date,Recording URL,Transcript"] + for i in range(1, 9): + lines.append( + f"conv-{i},18/05/2026,https://x/{i}.mp3,row {i}" ) + lines.append(",18/05/2026,https://x/bad1.mp3,missing conv") + lines.append("conv-10,18/05/2026,not-a-url,invalid url") + result = _parse_csv( + _csv_bytes("\n".join(lines) + "\n"), + _standard_params(), + _standard_mapping(), + _standard_skipped(), + ) + assert len(result.rows) == 8 + assert len(result.skipped) == 2 + assert {s.source_row for s in result.skipped} == {9, 10} + + +def test_raise_if_no_importable_rows_when_all_skipped(): + from app.api.v1.routes.call_imports import ( + CallImportParseResult, + CallImportParseSkip, + _raise_if_no_importable_rows, + ) + + result = CallImportParseResult( + rows=[], + skipped=[ + CallImportParseSkip( + source_row=2, + reason="missing_recording_url", + message="Row 2 is missing the required 'recording_url' value.", + ) + ], + ) + with pytest.raises(HTTPException) as exc: + _raise_if_no_importable_rows(result) assert exc.value.status_code == 400 - assert "conversation_id" in exc.value.detail.lower() + assert "No importable rows" in exc.value.detail def test_parse_csv_rejects_missing_recording_date_value(): @@ -277,7 +364,7 @@ def test_parse_csv_accepts_blank_recording_date_when_optional(): "CallID,Recording Date,Recording URL,Transcript\n" "abc-1,,https://x/recording.mp3,Some transcript\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), params, _standard_mapping(), @@ -315,7 +402,7 @@ def test_parse_csv_works_without_recording_date_parameter(): "CallID,Recording URL,Transcript\n" "abc-1,https://x/recording.mp3,Some transcript\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), params, mapping, @@ -350,7 +437,7 @@ def test_parse_csv_skips_completely_blank_rows(): ",,,\n" "abc-2,19/05/2026,https://x/2.mp3,Other transcript\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -365,7 +452,7 @@ def test_parse_csv_strips_utf8_bom(): "\ufeffCallID,Recording Date,Recording URL,Transcript\n" "abc-1,18/05/2026,https://x/recording.mp3,T1\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -398,7 +485,7 @@ def test_parse_csv_accepts_explicitly_skipped_columns(): "CallID,Recording Date,Recording URL,Transcript,AgentName\n" "abc-1,18/05/2026,https://x/r.mp3,hi,alice\n" ) - rows = _parse_csv( + rows = _parse_csv_rows( _csv_bytes(csv_text), _standard_params(), _standard_mapping(), @@ -421,7 +508,7 @@ def test_parse_csv_with_custom_text_parameter_is_preserved_per_row(): "CallID,Recording Date,Recording URL,Transcript,AgentName\n" "conv-1,18/05/2026,https://x/r1.mp3,hello there,alice\n" ) - rows = _parse_csv(_csv_bytes(csv_text), params, mapping, skipped_columns=[]) + rows = _parse_csv_rows(_csv_bytes(csv_text), params, mapping, skipped_columns=[]) assert rows[0]["parameter_values"]["agent_name"] == "alice" assert rows[0]["parameter_values"]["conversation_id"] == "conv-1" @@ -440,7 +527,7 @@ def test_parse_csv_coerces_typed_parameter_values(): "CallID,Recording Date,Recording URL,Transcript,Latency,Answered\n" "conv-1,18/05/2026,https://x/r.mp3,hi,123.5,true\n" ) - rows = _parse_csv(_csv_bytes(csv_text), params, mapping, skipped_columns=[]) + rows = _parse_csv_rows(_csv_bytes(csv_text), params, mapping, skipped_columns=[]) assert rows[0]["parameter_values"]["latency_ms"] == 123.5 assert rows[0]["parameter_values"]["answered"] is True @@ -1280,7 +1367,7 @@ def test_parse_xlsx_accepts_canonical_headers(): ] } ) - rows = _parse_xlsx( + rows = _parse_xlsx_rows( blob, "Calls", _standard_params(), _standard_mapping(), _standard_skipped() ) assert len(rows) == 2 @@ -1299,7 +1386,7 @@ def test_parse_xlsx_accepts_native_excel_recording_date_cells(): ] } ) - rows = _parse_xlsx( + rows = _parse_xlsx_rows( blob, "Calls", _standard_params(), _standard_mapping(), _standard_skipped() ) assert rows[0]["recording_date"] == "08/04/2026" @@ -1314,7 +1401,7 @@ def test_parse_xlsx_coerces_numeric_call_ids_without_decimal_suffix(): ] } ) - rows = _parse_xlsx( + rows = _parse_xlsx_rows( blob, "Calls", _standard_params(), _standard_mapping(), _standard_skipped() ) assert rows[0]["conversation_id"] == "12345" @@ -1331,7 +1418,7 @@ def test_parse_xlsx_skips_fully_blank_rows(): ] } ) - rows = _parse_xlsx( + rows = _parse_xlsx_rows( blob, "Calls", _standard_params(), _standard_mapping(), _standard_skipped() ) assert [r["conversation_id"] for r in rows] == ["abc-1", "abc-2"] diff --git a/tests/test_models/test_metric_rubric_max_length.py b/tests/test_models/test_metric_rubric_max_length.py new file mode 100644 index 00000000..bfe82762 --- /dev/null +++ b/tests/test_models/test_metric_rubric_max_length.py @@ -0,0 +1,35 @@ +"""Smoke test for metric rubric text length validation.""" + +from pydantic import ValidationError + +from app.models.schemas import ( + METRIC_RUBRIC_TEXT_MAX_LENGTH, + MetricChildDraft, + MetricCreateWithChildren, +) + + +def test_metric_rubric_accepts_text_over_legacy_4000_limit(): + long_text = "x" * 5000 + payload = MetricCreateWithChildren( + name="Test", + description=long_text, + selection_mode="single_choice", + children=[ + MetricChildDraft(name="Label1", description="def", example=long_text) + ], + ) + assert len(payload.description) == 5000 + assert len(payload.children[0].example) == 5000 + + +def test_metric_rubric_rejects_text_over_max_length(): + try: + MetricCreateWithChildren( + name="Test", + description="x" * (METRIC_RUBRIC_TEXT_MAX_LENGTH + 1), + selection_mode="single_choice", + ) + except ValidationError: + return + raise AssertionError("expected ValidationError for over-limit description")