diff --git a/backend/app/api/docs/assessment/get_dataset_rows.md b/backend/app/api/docs/assessment/get_dataset_rows.md new file mode 100644 index 000000000..4f2d6573e --- /dev/null +++ b/backend/app/api/docs/assessment/get_dataset_rows.md @@ -0,0 +1,8 @@ +Return every row of an assessment dataset as column-keyed JSON. + +Fetches and parses the underlying CSV/XLSX file in full, returning the column +`headers`, all `rows` (each a `{column: value}` dict), and `total_rows`. Intended +for the frontend to assemble a directly-usable API-batch input file client-side. + +Unlike the `limit_rows` preview on `GET /datasets/{dataset_id}`, this returns the +entire dataset and always fetches the file, so responses scale with dataset size. diff --git a/backend/app/api/routes/assessment/datasets.py b/backend/app/api/routes/assessment/datasets.py index 22f000c4d..2171a1e98 100644 --- a/backend/app/api/routes/assessment/datasets.py +++ b/backend/app/api/routes/assessment/datasets.py @@ -8,6 +8,7 @@ from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission from app.core.cloud import get_cloud_storage +from app.crud.assessment.batch import list_assessment_dataset_rows from app.crud.assessment.dataset import ( delete_assessment_dataset, get_assessment_dataset_by_id, @@ -16,6 +17,7 @@ from app.models.assessment import ( AssessmentDatasetPreview, AssessmentDatasetResponse, + AssessmentDatasetRows, ) from app.models.evaluation import EvaluationDataset from app.services.assessment.dataset import ( @@ -29,6 +31,11 @@ router = APIRouter() +# Hard cap for the rows-export endpoint. Above this the request is rejected +# up front (from the stored row count, no file download/parse) so a huge sheet +# never triggers heavy computation. +MAX_DATASET_EXPORT_ROWS = 100 + def _dataset_to_response( dataset: EvaluationDataset, @@ -168,6 +175,75 @@ def get_dataset( ) +@router.get( + "/datasets/{dataset_id}/rows", + description=load_description("assessment/get_dataset_rows.md"), + response_model=APIResponse[AssessmentDatasetRows], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def get_dataset_rows( + dataset_id: int, + session: SessionDep, + auth_context: AuthContextDep, +) -> APIResponse[AssessmentDatasetRows]: + """Return all rows of an assessment dataset as column-keyed dicts.""" + dataset = get_assessment_dataset_by_id( + session=session, + dataset_id=dataset_id, + organization_id=auth_context.organization_.id, + project_id=auth_context.project_.id, + ) + + # Reject oversized datasets up front from the stored row count (recorded at + # upload), so no file is downloaded or parsed for a sheet we won't return. + total_items = (dataset.dataset_metadata or {}).get("total_items_count") + if isinstance(total_items, int) and total_items > MAX_DATASET_EXPORT_ROWS: + raise HTTPException( + status_code=422, + detail=( + f"Max size exceeded: dataset has {total_items} rows; the input " + f"rows export is limited to {MAX_DATASET_EXPORT_ROWS}." + ), + ) + + try: + rows = list_assessment_dataset_rows(session, dataset) + except ValueError as e: + logger.warning( + "[get_dataset_rows] Failed to load dataset rows | dataset_id=%s | %s", + dataset_id, + e, + ) + raise HTTPException(status_code=422, detail=str(e)) from e + + # Drop blank-named columns and columns empty across every row (trailing + # empty spreadsheet columns) so the response carries only real fields. + # Single pass over the cells decides which columns to keep (O(rows*cols)), + # so it stays cheap even for large sheets. + ordered_columns: list[str] = [] + seen_columns: set[str] = set() + non_empty_columns: set[str] = set() + for row in rows: + for column, value in row.items(): + if column not in seen_columns: + seen_columns.add(column) + ordered_columns.append(column) + if column.strip() and (value or "").strip(): + non_empty_columns.add(column) + kept_columns = [c for c in ordered_columns if c in non_empty_columns] + cleaned_rows = [ + {column: row.get(column, "") for column in kept_columns} for row in rows + ] + + return APIResponse.success_response( + data=AssessmentDatasetRows( + headers=kept_columns, + rows=cleaned_rows, + total_rows=len(cleaned_rows), + ) + ) + + @router.delete( "/datasets/{dataset_id}", description=load_description("assessment/delete_dataset.md"), diff --git a/backend/app/api/routes/assessment/runs.py b/backend/app/api/routes/assessment/runs.py index 473b5dad9..32b95625f 100644 --- a/backend/app/api/routes/assessment/runs.py +++ b/backend/app/api/routes/assessment/runs.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, Body, Depends, HTTPException, Query from fastapi.responses import StreamingResponse +from pydantic import ValidationError from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission @@ -29,6 +30,7 @@ AssessmentRunCreate, AssessmentRunPublic, AssessmentRunResponse, + RunExecution, ) from app.models.evaluation import EvaluationDataset from app.services.assessment.service import ( @@ -67,6 +69,17 @@ def _build_run_public( run.id, ) dataset = session.get(EvaluationDataset, parent.dataset_id) if parent else None + # Runtime state was folded into the `execution` JSONB bag (migration 078); the frozen + # request input lives on the parent, not the run. Legacy list/get are scoped to + # method=RUN, but guard defensively so a non-RUN-shaped bag can never 500 the list. + try: + bag = RunExecution.model_validate(run.execution or {}) + except ValidationError: + logger.warning( + "[_build_run_public] Non-RUN execution bag for run %s; returning empty bag", + run.id, + ) + bag = RunExecution() return AssessmentRunPublic( id=run.id, assessment_id=run.assessment_id, @@ -78,14 +91,15 @@ def _build_run_public( status=run.status, total_items=run.total_items, error_message=run.error_message, - input=run.input, - prefilter_total_rows=run.prefilter_total_rows, - prefilter_total_passed=run.prefilter_total_passed, - prefilter_total_rejected=run.prefilter_total_rejected, - stage=run.stage, - stage_status=run.stage_status, - pipeline=run.pipeline, - post_processing_config=(run.input or {}).get("post_processing_config"), + input=parent.input if parent else None, + prefilter_total_rows=bag.prefilter_total_rows, + prefilter_total_passed=bag.prefilter_total_passed, + prefilter_total_rejected=bag.prefilter_total_rejected, + stage=bag.stage, + stage_status=bag.stage_status, + pipeline=bag.pipeline, + cost=bag.cost, + post_processing_config=run.post_processing_config, inserted_at=run.inserted_at, updated_at=run.updated_at, ) @@ -261,7 +275,7 @@ def export_assessment_run_results( ) ) - post_processing_config = (run.input or {}).get("post_processing_config") or None + post_processing_config = run.post_processing_config or None base_label = assessment.experiment_name if assessment else f"run_{run.id}" if export_format != "json": diff --git a/backend/app/crud/assessment/__init__.py b/backend/app/crud/assessment/__init__.py index a5cbb23b9..f20c65aca 100644 --- a/backend/app/crud/assessment/__init__.py +++ b/backend/app/crud/assessment/__init__.py @@ -17,6 +17,7 @@ list_assessment_runs, list_assessments, recompute_assessment_status, + resolve_assessment_config_blob, update_assessment_run_prefilter_stats, update_assessment_run_status, update_run_post_processing_config, @@ -49,6 +50,7 @@ "list_assessment_datasets", "list_assessments", "recompute_assessment_status", + "resolve_assessment_config_blob", "update_assessment_run_prefilter_stats", "update_assessment_run_status", "update_run_post_processing_config", diff --git a/backend/app/crud/assessment/batch.py b/backend/app/crud/assessment/batch.py index d77b239e5..5ac2f5d05 100644 --- a/backend/app/crud/assessment/batch.py +++ b/backend/app/crud/assessment/batch.py @@ -28,9 +28,9 @@ AssessmentRun, ) from app.models.batch_job import BatchJob, BatchJobType +from app.models.config.assessment_blob import AssessmentConfigBlob from app.models.evaluation import EvaluationDataset from app.models.llm.constants import DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS -from app.models.llm.request import ConfigBlob from app.services.assessment.mappers import ( map_kaapi_to_anthropic_params, map_kaapi_to_google_params, @@ -80,6 +80,18 @@ def _load_dataset_rows( return _parse_csv_rows(file_content) +def list_assessment_dataset_rows( + session: Session, + dataset: EvaluationDataset, +) -> list[dict[str, str]]: + """Public entry point for loading all dataset rows as column-keyed dicts. + + Thin wrapper over the batch loader so callers outside the batch path (e.g. the + legacy rows endpoint) don't reach into the private helper. + """ + return _load_dataset_rows(session, dataset) + + def _parse_csv_rows(content: bytes) -> list[dict[str, str]]: """Parse CSV content into list of row dicts.""" for encoding in ("utf-8-sig", "utf-8", "latin-1"): @@ -367,7 +379,7 @@ def submit_assessment_batch( run: AssessmentRun, assessment: Assessment, dataset: EvaluationDataset, - config_blob: ConfigBlob, + config_blob: AssessmentConfigBlob, assessment_input: dict[str, Any], organization_id: int, project_id: int, @@ -407,14 +419,22 @@ def submit_assessment_batch( "[submit_assessment_batch] Building JSONL | run_id=%s | rows=%s | provider=%s", run.id, len(rows), - config_blob.completion.provider, + config_blob.assessment.provider, ) # Determine provider and build params - completion = config_blob.completion - provider_name = completion.provider or "openai" - - params = dict(completion.params) + assessment_cfg = config_blob.assessment + provider_name = assessment_cfg.provider or LLMProvider.OPENAI + + # Normalize assessment params for the mappers, mirroring the BATCH path's + # _stage_params: input_schema is request-validation only (not a provider param), + # and json_output_schema is the mappers' output_schema. instructions stays in + # params — the mappers consume it as the system prompt. + params = dict(assessment_cfg.params) + params.pop("input_schema", None) + json_schema = params.pop("json_output_schema", None) + if json_schema is not None: + params["output_schema"] = json_schema # Determine the base provider (openai or google) base_provider = provider_name.replace("-native", "") diff --git a/backend/app/crud/assessment/core.py b/backend/app/crud/assessment/core.py index 450a74ed3..ba0cd781a 100644 --- a/backend/app/crud/assessment/core.py +++ b/backend/app/crud/assessment/core.py @@ -5,10 +5,12 @@ from uuid import UUID from fastapi import HTTPException +from pydantic import ValidationError from sqlalchemy.orm.attributes import flag_modified from sqlmodel import Session, select from app.core.util import now +from app.crud.config.version import ConfigVersionCrud from app.models.assessment import ( Assessment, AssessmentMethod, @@ -17,10 +19,99 @@ AssessmentRunStat, AssessmentStatus, ) +from app.models.config.assessment_blob import AssessmentConfigBlob +from app.models.config.config import ConfigTag logger = logging.getLogger(__name__) +# Keys an input_schema column spec is allowed to carry (mirrors InputColumn). +_PERMITTED_INPUT_COLUMN_KEYS = {"type", "format", "strict"} + + +def _sanitize_input_schema_keys(config_blob: Any) -> Any: + """Drop unknown keys from each ``input_schema`` column spec, logging a warning. + + Config-create rejects extra keys (InputColumn is ``extra='forbid'``), but at + RESOLVE time we stay tolerant: a stored blob that carries a stray column key + (e.g. a pre-strictness config) should still run — strip the extras and warn + rather than raise. Only keys are pruned; invalid values still fail validation. + """ + if not isinstance(config_blob, dict): + return config_blob + params = (config_blob.get("assessment") or {}).get("params") + if not isinstance(params, dict): + return config_blob + input_schema = params.get("input_schema") + if not isinstance(input_schema, dict): + return config_blob + + cleaned: dict[str, Any] = {} + changed = False + for column, spec in input_schema.items(): + if isinstance(spec, dict): + extra = set(spec) - _PERMITTED_INPUT_COLUMN_KEYS + if extra: + changed = True + logger.warning( + "[resolve_assessment_config_blob] Ignoring unknown input_schema " + "key(s) %s on column '%s'", + sorted(extra), + column, + ) + cleaned[column] = { + k: v for k, v in spec.items() if k in _PERMITTED_INPUT_COLUMN_KEYS + } + continue + cleaned[column] = spec + + if not changed: + return config_blob + # Shallow-copy only the mutated path so the stored dict is never touched. + new_blob = dict(config_blob) + new_assessment = dict(config_blob["assessment"]) + new_params = dict(params) + new_params["input_schema"] = cleaned + new_assessment["params"] = new_params + new_blob["assessment"] = new_assessment + return new_blob + + +def resolve_assessment_config_blob( + session: Session, + config_id: UUID, + config_version: int, + project_id: int, +) -> tuple[AssessmentConfigBlob | None, str | None]: + """Resolve a stored ASSESSMENT config version as an ``AssessmentConfigBlob``. + + The RUN pipeline shares its config store with evals but not their blob shape: + ``resolve_evaluation_config`` returns a ``ConfigBlob`` (``{completion}``), while + an ASSESSMENT config is stored as ``{assessment, pre_filters}``. Parse it into + the right model instead, mirroring the BATCH path's ``_resolve_blob``. + + Returns ``(blob, None)`` on success or ``(None, error)`` on a missing version or + an invalid/old-shape blob, so callers surface a clean 400 rather than a 500. + """ + version_crud = ConfigVersionCrud( + session=session, + config_id=config_id, + project_id=project_id, + tag=ConfigTag.ASSESSMENT, + ) + version = version_crud.read_one(version_number=config_version) + if version is None: + return None, f"Config version {config_version} not found" + + try: + blob = AssessmentConfigBlob.model_validate( + _sanitize_input_schema_keys(version.config_blob) + ) + except ValidationError as exc: + return None, f"Stored configuration blob is invalid: {exc}" + return blob, None + + def _read_exec(run: AssessmentRun) -> dict[str, Any]: """Read the RUN runtime bag (RunExecution shape) that replaced the dropped columns.""" return run.execution or {} @@ -85,6 +176,7 @@ def get_assessment_by_id( .where(Assessment.id == assessment_id) .where(Assessment.organization_id == organization_id) .where(Assessment.project_id == project_id) + .where(Assessment.method == AssessmentMethod.RUN) ) assessment = session.exec(statement).first() if not assessment: @@ -107,6 +199,7 @@ def list_assessments( select(Assessment) .where(Assessment.organization_id == organization_id) .where(Assessment.project_id == project_id) + .where(Assessment.method == AssessmentMethod.RUN) .order_by(Assessment.inserted_at.desc()) .limit(limit) .offset(offset) @@ -189,6 +282,7 @@ def get_assessment_run_by_id( .where(AssessmentRun.id == run_id) .where(Assessment.organization_id == organization_id) .where(Assessment.project_id == project_id) + .where(Assessment.method == AssessmentMethod.RUN) ) run = session.exec(statement).first() if not run: @@ -226,6 +320,7 @@ def list_assessment_runs( .join(Assessment, Assessment.id == AssessmentRun.assessment_id) .where(Assessment.organization_id == organization_id) .where(Assessment.project_id == project_id) + .where(Assessment.method == AssessmentMethod.RUN) ) if assessment_id is not None: statement = statement.where(AssessmentRun.assessment_id == assessment_id) diff --git a/backend/app/crud/assessment/processing.py b/backend/app/crud/assessment/processing.py index 4b38762ac..78e4ec137 100644 --- a/backend/app/crud/assessment/processing.py +++ b/backend/app/crud/assessment/processing.py @@ -20,14 +20,21 @@ update_assessment_run_prefilter_stats, update_assessment_run_status, ) -from app.crud.assessment.core import _read_exec, _write_exec +from app.crud.assessment.core import ( + _read_exec, + _write_exec, + resolve_assessment_config_blob, +) from app.crud.job import get_batch_job +from app.crud.model_config import estimate_model_cost from app.models.assessment import ( Assessment, AssessmentRun, AssessmentStatus, + Stage, StageStatus, ) +from app.services.assessment.prefilter.constants import ASSESSMENT_PREFILTER_MODEL from app.services.assessment.stages import ( GATE_STAGES, STAGE_PARSERS, @@ -39,6 +46,16 @@ logger = logging.getLogger(__name__) +_COST_CURRENCY = "USD" +# Gemini pricing rows may be keyed under this provider rather than plain "google". +_GOOGLE_AISTUDIO_PROVIDER = "google-aistudio" +# Pre-filter stage -> cost sub-key under execution.cost.pre_filter. Keyed by str +# (not Stage) because the exec bag stores stage as a plain string. +_PREFILTER_COST_KEYS: dict[str, str] = { + Stage.PRE_FILTER_TOPIC_RELEVANCE: "topic_relevance", + Stage.PRE_FILTER_DUPLICATE_DETECTION: "duplicate_detection", +} + def format_assessment_failure_message(exc: Exception) -> str: """Extract a DB-safe error message from assessment polling exceptions.""" @@ -365,6 +382,132 @@ def _record_gate_stats( ) +def _sum_stage_tokens(outputs: list[dict[str, Any]]) -> tuple[int, int]: + """Sum input/output tokens across a stage's parsed rows; missing usage counts as 0.""" + from app.services.assessment.utils.parsing import usage_totals + + input_total = 0 + output_total = 0 + for row in outputs: + input_tokens, output_tokens, _ = usage_totals(row.get("usage")) + input_total += input_tokens or 0 + output_total += output_tokens or 0 + return input_total, output_total + + +def _estimate_stage_usd( + session: Session, + provider: str, + model_name: str, + input_tokens: int, + output_tokens: int, +) -> float: + """Batch USD for a stage; a missing model/pricing row is treated as 0.0 (warned).""" + result = estimate_model_cost( + session=session, + provider=provider, # type: ignore[arg-type] + model_name=model_name, + input_tokens=input_tokens, + output_tokens=output_tokens, + usage_type="batch", + ) + if result is None and provider == LLMProvider.GOOGLE: + result = estimate_model_cost( + session=session, + provider=_GOOGLE_AISTUDIO_PROVIDER, # type: ignore[arg-type] + model_name=model_name, + input_tokens=input_tokens, + output_tokens=output_tokens, + usage_type="batch", + ) + if result is None: + logger.warning( + "[_estimate_stage_usd] No pricing for provider=%s model=%s — cost=0.0", + provider, + model_name, + ) + return 0.0 + return float(result.get("total_cost") or 0.0) + + +def _resolve_l2_model( + session: Session, run: AssessmentRun, project_id: int +) -> str | None: + """Model name for the L2 assessment call, resolved the way L2 submission does.""" + config_blob, error = resolve_assessment_config_blob( + session=session, + config_id=run.config_id, + config_version=run.config_version, + project_id=project_id, + ) + if error or config_blob is None: + logger.warning( + "[_resolve_l2_model] run_id=%s config resolution failed — %s", run.id, error + ) + return None + return (config_blob.assessment.params or {}).get("model") + + +def _record_stage_cost( + session: Session, run: AssessmentRun, stage: str, batch_job, project_id: int +) -> None: + """Accumulate the just-completed stage's batch USD into execution.cost. + + Best-effort: cost accounting must never fail the pipeline, so every error is + swallowed with a warning. pre_filter.total and the grand total are recomputed + on each call so a resumed/partial run stays internally consistent. + """ + try: + provider = batch_job.provider + if stage in _PREFILTER_COST_KEYS: + model_name = ASSESSMENT_PREFILTER_MODEL + elif stage == Stage.L2_ASSESSMENT: + model_name = _resolve_l2_model(session, run, project_id) + else: + return + + if not model_name: + logger.warning( + "[_record_stage_cost] run_id=%s stage=%s — no model to price, cost=0.0", + run.id, + stage, + ) + + raw = load_raw_batch_results(session, batch_job, project_id) + outputs = parse_assessment_output(raw, provider) + input_tokens, output_tokens = _sum_stage_tokens(outputs) + stage_cost = ( + _estimate_stage_usd( + session, provider, model_name, input_tokens, output_tokens + ) + if model_name + else 0.0 + ) + + cost = dict(_read_exec(run).get("cost") or {}) + pre_filter = dict(cost.get("pre_filter") or {}) + if stage in _PREFILTER_COST_KEYS: + pre_filter[_PREFILTER_COST_KEYS[stage]] = stage_cost + elif stage == Stage.L2_ASSESSMENT: + cost["assessment"] = stage_cost + + if pre_filter: + pre_filter["total"] = sum( + pre_filter[key] + for key in _PREFILTER_COST_KEYS.values() + if pre_filter.get(key) is not None + ) + cost["pre_filter"] = pre_filter + + cost["total"] = pre_filter.get("total", 0.0) + (cost.get("assessment") or 0.0) + cost["currency"] = _COST_CURRENCY + _write_exec(run, cost=cost) + except Exception as exc: + logger.warning( + "[_record_stage_cost] run_id=%s stage=%s — %s", run.id, stage, exc + ) + + def _fail_run_stage( session: Session, run: AssessmentRun, message: str ) -> dict[str, Any]: @@ -425,6 +568,7 @@ async def process_run_batches(run: AssessmentRun, session: Session) -> dict[str, _write_exec(run, stage_status=StageStatus.COMPLETED) if stage in GATE_STAGES: _record_gate_stats(session, run, stage, batch_job, parent.project_id) + _record_stage_cost(session, run, stage, batch_job, parent.project_id) nxt = advance_or_finalize(run) session.add(run) diff --git a/backend/app/models/assessment/__init__.py b/backend/app/models/assessment/__init__.py index dbc581713..293c39c3e 100644 --- a/backend/app/models/assessment/__init__.py +++ b/backend/app/models/assessment/__init__.py @@ -14,6 +14,7 @@ AssessmentConfigRef, AssessmentDatasetPreview, AssessmentDatasetResponse, + AssessmentDatasetRows, AssessmentExecutionPublic, AssessmentExportRow, AssessmentMethod, @@ -29,7 +30,9 @@ AssessmentRunSummary, AssessmentStatus, InputBinding, + RunCost, RunExecution, + RunPreFilterCost, Stage, StageStatus, ) @@ -67,6 +70,8 @@ "StageStatus", "AssessmentAttachment", "InputBinding", + "RunCost", + "RunPreFilterCost", "RunExecution", "AssessmentExecutionPublic", "AssessmentResponse", @@ -81,6 +86,7 @@ "AssessmentRunOverview", "AssessmentDatasetPreview", "AssessmentDatasetResponse", + "AssessmentDatasetRows", # API-client surface "Attachment", "ResponseInput", diff --git a/backend/app/models/assessment/assessment.py b/backend/app/models/assessment/assessment.py index 3b4d9d249..01b50585e 100644 --- a/backend/app/models/assessment/assessment.py +++ b/backend/app/models/assessment/assessment.py @@ -117,6 +117,25 @@ class InputBinding(BaseModel): attachments: list[AssessmentAttachment] = Field(default_factory=list) +# NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. +class RunPreFilterCost(BaseModel): + """Per-submodule pre-filter batch cost (USD); sub-keys present only if the stage ran.""" + + topic_relevance: float | None = None + duplicate_detection: float | None = None + total: float = 0.0 + + +# NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. +class RunCost(BaseModel): + """Batch USD cost of a run, split into pre-filter and assessment stages.""" + + pre_filter: RunPreFilterCost | None = None + assessment: float | None = None + total: float | None = None + currency: str = "USD" + + # NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. class RunExecution(BaseModel): """Legacy RUN pipeline runtime state (assessment_run.execution).""" @@ -130,6 +149,7 @@ class RunExecution(BaseModel): prefilter_total_rejected: int | None = None object_store_url: str | None = None prefilter_object_store_url: str | None = None + cost: RunCost | None = None class Assessment(SQLModel, table=True): @@ -294,18 +314,6 @@ class AssessmentResponse(BaseModel): executions: list[AssessmentExecutionPublic] = [] -# NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. -class AssessmentPublic(BaseModel): - id: UUID - experiment_name: str | None = None - status: AssessmentStatus - executions: list[AssessmentExecutionPublic] = [] - organization_id: int - project_id: int - inserted_at: datetime - updated_at: datetime - - # NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. class AssessmentExportRow(BaseModel): """Flattened result row for CSV/XLSX/JSON export.""" @@ -359,6 +367,23 @@ class AssessmentRunStat(BaseModel): updated_at: datetime | None = None +# NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. +# Declared after AssessmentRunCounts/AssessmentRunStat so the nested annotations resolve. +class AssessmentPublic(BaseModel): + id: UUID + experiment_name: str | None = None + dataset_id: int | None = None + dataset_name: str | None = None + status: AssessmentStatus + counts: AssessmentRunCounts + run_stats: list[AssessmentRunStat] = [] + error_message: str | None = None + organization_id: int + project_id: int + inserted_at: datetime + updated_at: datetime + + # NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. class AssessmentRunSummary(BaseModel): run_id: int @@ -372,14 +397,23 @@ class AssessmentRunSummary(BaseModel): class AssessmentRunPublic(BaseModel): id: int assessment_id: UUID + experiment_name: str | None = None + dataset_id: int | None = None + dataset_name: str | None = None config_id: UUID config_version: int status: AssessmentStatus total_items: int - batch_job_id: int | None = None - execution: RunExecution | None = None - post_processing_config: dict[str, Any] | None = None error_message: str | None = None + input: dict[str, Any] | None = None + prefilter_total_rows: int | None = None + prefilter_total_passed: int | None = None + prefilter_total_rejected: int | None = None + stage: Stage | None = None + stage_status: StageStatus | None = None + pipeline: dict[str, Any] | None = None + post_processing_config: dict[str, Any] | None = None + cost: RunCost | None = None inserted_at: datetime updated_at: datetime @@ -418,6 +452,15 @@ class AssessmentDatasetPreview(BaseModel): truncated: bool = False +# NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. +class AssessmentDatasetRows(BaseModel): + """All rows of a dataset as column-keyed dicts, for client-side batch input assembly.""" + + headers: list[str] + rows: list[dict[str, str]] + total_rows: int = 0 + + # NOTE: Legacy, this is for Assessment Run UI only. The new Assessment pipeline does not use this. class AssessmentDatasetResponse(BaseModel): dataset_id: int diff --git a/backend/app/models/config/assessment_blob.py b/backend/app/models/config/assessment_blob.py index fdfbe0907..4fff240b0 100644 --- a/backend/app/models/config/assessment_blob.py +++ b/backend/app/models/config/assessment_blob.py @@ -17,10 +17,16 @@ class InputColumn(SQLModel): """One BATCH input column: its type (required, no default — every column must - declare one), and how an attachment value is provided (`format`).""" + declare one), and how an attachment value is provided (`format`). + + Only these keys are permitted — an unknown key in a column spec is rejected + (422) rather than silently dropped.""" + + model_config = {"extra": "forbid"} type: Literal["text", "image", "pdf"] format: Literal["url", "base64"] | None = None + strict: bool = False # when true, the column must be present in every row class PreFilterParams(TextLLMParams): diff --git a/backend/app/services/assessment/api/submission.py b/backend/app/services/assessment/api/submission.py index 1034a8681..c5001ff32 100644 --- a/backend/app/services/assessment/api/submission.py +++ b/backend/app/services/assessment/api/submission.py @@ -36,18 +36,20 @@ def _validate_rows_against_schema( - rows: list[dict[str, str]], input_schema: dict[str, dict[str, str | None]] + rows: list[dict[str, str]], input_schema: dict[str, dict[str, str | bool | None]] ) -> None: """Validate every BATCH row against the config's input_schema. - Raises 422 on the first offending row: a missing declared column, an - unexpected column not in the schema, or an attachment column whose value is - not a URL. The row index is named so the client can locate the bad row. + Raises 422 on the first offending row: a missing ``strict`` column, an + unexpected column not in the schema, or a present attachment column whose + value is not a URL. Non-strict columns are optional. The row index is named + so the client can locate the bad row. """ declared = set(input_schema) + required = {c for c, spec in input_schema.items() if (spec or {}).get("strict")} for idx, row in enumerate(rows): row_columns = set(row) - missing = declared - row_columns + missing = required - row_columns if missing: raise HTTPException( status_code=422, @@ -63,7 +65,8 @@ def _validate_rows_against_schema( ), ) for column, spec in input_schema.items(): - if (spec or {}).get("type") in _ATTACHMENT_TYPES: + # Only present values are checked — a non-strict column may be absent. + if column in row_columns and (spec or {}).get("type") in _ATTACHMENT_TYPES: value = row.get(column, "") if not value.startswith(_URL_PREFIXES): raise HTTPException( diff --git a/backend/app/services/assessment/service.py b/backend/app/services/assessment/service.py index 6c2548f34..1588a5488 100644 --- a/backend/app/services/assessment/service.py +++ b/backend/app/services/assessment/service.py @@ -13,10 +13,10 @@ get_assessment_dataset_by_id, get_assessment_runs_for_assessment, recompute_assessment_status, + resolve_assessment_config_blob, ) from app.crud.assessment.core import _read_exec, _write_exec from app.crud.config import ConfigCrud -from app.crud.evaluations.core import resolve_evaluation_config from app.models.assessment import ( Assessment, AssessmentAttachment, @@ -144,12 +144,11 @@ def start_assessment( ), ) - config_blob, error = resolve_evaluation_config( + config_blob, error = resolve_assessment_config_blob( session=session, config_id=cfg.id, config_version=cfg.version, project_id=project_id, - tag=ConfigTag.ASSESSMENT, ) if error or config_blob is None: raise HTTPException( @@ -158,7 +157,7 @@ def start_assessment( f"Failed to resolve config {cfg.id} " f"v{cfg.version}: {error}" ), ) - provider = config_blob.completion.provider or LLMProvider.OPENAI + provider = config_blob.assessment.provider or LLMProvider.OPENAI if provider not in _SUPPORTED_BATCH_PROVIDERS: raise HTTPException( status_code=422, diff --git a/backend/app/services/assessment/tasks.py b/backend/app/services/assessment/tasks.py index c4d6b45a5..7760d1369 100644 --- a/backend/app/services/assessment/tasks.py +++ b/backend/app/services/assessment/tasks.py @@ -11,12 +11,12 @@ from app.crud.assessment import ( get_assessment_dataset_by_id, recompute_assessment_status, + resolve_assessment_config_blob, update_assessment_run_status, ) from app.crud.assessment.batch import _load_dataset_rows, submit_assessment_batch from app.crud.assessment.core import _read_exec, _write_exec from app.crud.assessment.processing import parse_assessment_output -from app.crud.evaluations.core import resolve_evaluation_config from app.crud.job import get_batch_job from app.models.assessment import ( Assessment, @@ -26,7 +26,6 @@ Stage, StageStatus, ) -from app.models.config.config import ConfigTag from app.services.assessment.prefilter import resolve_prefilter_settings from app.services.assessment.stages import ( GATE_STAGES, @@ -116,12 +115,11 @@ def _resolve_run_context( organization_id=organization_id, project_id=project_id, ) - config_blob, error = resolve_evaluation_config( + config_blob, error = resolve_assessment_config_blob( session=session, config_id=run.config_id, config_version=run.config_version, project_id=project_id, - tag=ConfigTag.ASSESSMENT, ) if error or config_blob is None: return assessment, dataset, None, f"Config resolution failed: {error}" diff --git a/backend/app/tests/assessment/test_batch.py b/backend/app/tests/assessment/test_batch.py index 16a9e0f1a..4e430c78b 100644 --- a/backend/app/tests/assessment/test_batch.py +++ b/backend/app/tests/assessment/test_batch.py @@ -18,6 +18,7 @@ submit_assessment_batch, ) from app.models.assessment import AssessmentAttachment +from app.models.config.assessment_blob import AssessmentConfigBlob from app.models.llm.constants import DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS from app.services.assessment.utils.attachments import ( _guess_image_mime_from_url, @@ -56,7 +57,7 @@ def test_openai_native_routes_to_openai_batch(self) -> None: run = _make_run() dataset = _make_dataset() config_blob = SimpleNamespace( - completion=SimpleNamespace( + assessment=SimpleNamespace( provider="openai-native", params={"instructions": "config system"}, ) @@ -115,15 +116,29 @@ def test_openai_native_routes_to_openai_batch(self) -> None: ) assert start_batch.call_args.kwargs["provider_name"] == "openai" - def test_config_instruction_is_used(self) -> None: + def test_config_instruction_and_schema_normalization(self) -> None: + """The resolved AssessmentConfigBlob's params reach the mapper normalized: + instructions kept, input_schema stripped, json_output_schema -> output_schema. + """ session = MagicMock() run = _make_run() dataset = _make_dataset() - config_blob = SimpleNamespace( - completion=SimpleNamespace( - provider="openai", - params={"instructions": "config system", "model": "gpt-4.1-mini"}, - ) + output_schema = { + "type": "object", + "properties": {"score": {"type": "integer"}}, + } + config_blob = AssessmentConfigBlob( + pre_filters=None, + assessment={ + "provider": "openai", + "type": "text", + "params": { + "model": "gpt-4.1-mini", + "instructions": "config system", + "input_schema": {"question": {"type": "text"}}, + "json_output_schema": output_schema, + }, + }, ) batch_job = MagicMock() batch_job.id = 3 @@ -166,18 +181,21 @@ def test_config_instruction_is_used(self) -> None: project_id=1, ) + kaapi_params = map_params.call_args.kwargs["kaapi_params"] assert map_params.call_args.kwargs["session"] is session - assert ( - map_params.call_args.kwargs["kaapi_params"]["instructions"] - == "config system" - ) + assert kaapi_params["instructions"] == "config system" + # input_schema is request-validation only — never a provider param. + assert "input_schema" not in kaapi_params + # json_output_schema is renamed to the mapper's output_schema key. + assert "json_output_schema" not in kaapi_params + assert kaapi_params["output_schema"] == output_schema def test_google_native_routes_to_google_batch(self) -> None: session = MagicMock() run = _make_run() dataset = _make_dataset() config_blob = SimpleNamespace( - completion=SimpleNamespace( + assessment=SimpleNamespace( provider="google-native", params={"instructions": "config system"}, ) @@ -236,7 +254,7 @@ def test_anthropic_native_routes_to_anthropic_batch(self) -> None: run = _make_run() dataset = _make_dataset() config_blob = SimpleNamespace( - completion=SimpleNamespace( + assessment=SimpleNamespace( provider="anthropic-native", params={"instructions": "config system"}, ) diff --git a/backend/app/tests/assessment/test_prefilter_batching.py b/backend/app/tests/assessment/test_prefilter_batching.py index 9c86d8267..dfe8a3d21 100644 --- a/backend/app/tests/assessment/test_prefilter_batching.py +++ b/backend/app/tests/assessment/test_prefilter_batching.py @@ -239,7 +239,7 @@ def test_success(self) -> None: with patch.object( tasks, "get_assessment_dataset_by_id", return_value=MagicMock() ), patch.object( - tasks, "resolve_evaluation_config", return_value=({"x": 1}, None) + tasks, "resolve_assessment_config_blob", return_value=({"x": 1}, None) ): _a, _d, blob, err = tasks._resolve_run_context(session, run, 1, 1) assert blob == {"x": 1} @@ -258,7 +258,7 @@ def test_config_error(self) -> None: with patch.object( tasks, "get_assessment_dataset_by_id", return_value=MagicMock() ), patch.object( - tasks, "resolve_evaluation_config", return_value=(None, "bad config") + tasks, "resolve_assessment_config_blob", return_value=(None, "bad config") ): _a, _d, blob, err = tasks._resolve_run_context(session, _run(), 1, 1) assert blob is None diff --git a/backend/app/tests/assessment/test_processing.py b/backend/app/tests/assessment/test_processing.py index c7afb2750..df1eba881 100644 --- a/backend/app/tests/assessment/test_processing.py +++ b/backend/app/tests/assessment/test_processing.py @@ -10,11 +10,13 @@ from app.crud.assessment.processing import ( _get_batch_provider, _record_gate_stats, + _record_stage_cost, _sanitize_json_output, parse_assessment_output, process_run_batches, ) from app.models.assessment import Stage, StageStatus +from app.services.assessment.prefilter.constants import ASSESSMENT_PREFILTER_MODEL @pytest.fixture(autouse=True) @@ -330,6 +332,126 @@ def test_intersects_with_prior_gate(self) -> None: assert run.execution["pipeline"]["accepted_indices"] == [2] +def _openai_row(input_tokens: int, output_tokens: int) -> dict: + """Raw OpenAI batch row carrying the usage block _sum_stage_tokens reads.""" + return { + "custom_id": "row_0", + "response": { + "status_code": 200, + "body": { + "id": "resp_1", + "output_text": "ok", + "usage": { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + }, + }, + }, + "error": None, + } + + +class TestRecordStageCost: + """execution.cost accumulates per-stage batch USD; totals recompute each call.""" + + def _run(self) -> SimpleNamespace: + return SimpleNamespace( + id=1, assessment_id=2, config_id=None, config_version=1, execution={} + ) + + def _price_by_model(self, mapping: dict[str, float | None]): + """estimate_model_cost stub keyed by model_name -> total_cost (or None).""" + + def _side_effect(**kwargs) -> dict | None: + total = mapping.get(kwargs["model_name"]) + return None if total is None else {"total_cost": total} + + return _side_effect + + def test_breakdown_recomputes_across_two_stages(self) -> None: + run = self._run() + batch_job = SimpleNamespace(provider="openai") + l2_blob = SimpleNamespace( + assessment=SimpleNamespace(params={"model": "l2-model"}) + ) + + with patch.object( + processing_mod, + "load_raw_batch_results", + return_value=[_openai_row(100, 50)], + ), patch.object( + processing_mod, + "estimate_model_cost", + side_effect=self._price_by_model( + {ASSESSMENT_PREFILTER_MODEL: 0.5, "l2-model": 1.25} + ), + ), patch.object( + processing_mod, + "resolve_assessment_config_blob", + return_value=(l2_blob, None), + ): + _record_stage_cost( + MagicMock(), run, Stage.PRE_FILTER_TOPIC_RELEVANCE, batch_job, 1 + ) + + after_stage_1 = run.execution["cost"] + assert after_stage_1["pre_filter"]["topic_relevance"] == 0.5 + assert after_stage_1["pre_filter"]["total"] == 0.5 + assert after_stage_1["total"] == 0.5 + assert "assessment" not in after_stage_1 + assert after_stage_1["currency"] == "USD" + + _record_stage_cost(MagicMock(), run, Stage.L2_ASSESSMENT, batch_job, 1) + + cost = run.execution["cost"] + assert cost["pre_filter"]["topic_relevance"] == 0.5 + assert cost["pre_filter"]["total"] == 0.5 + assert cost["assessment"] == 1.25 + assert cost["total"] == 1.75 + + def test_missing_pricing_yields_zero_cost(self) -> None: + run = self._run() + batch_job = SimpleNamespace(provider="openai") + + with patch.object( + processing_mod, + "load_raw_batch_results", + return_value=[_openai_row(100, 50)], + ), patch.object(processing_mod, "estimate_model_cost", return_value=None): + _record_stage_cost( + MagicMock(), run, Stage.PRE_FILTER_TOPIC_RELEVANCE, batch_job, 1 + ) + + cost = run.execution["cost"] + assert cost["pre_filter"]["topic_relevance"] == 0.0 + assert cost["pre_filter"]["total"] == 0.0 + assert cost["total"] == 0.0 + + def test_google_aistudio_fallback_lookup(self) -> None: + run = self._run() + batch_job = SimpleNamespace(provider="google") + calls: list[str] = [] + + def _side_effect(**kwargs) -> dict | None: + calls.append(kwargs["provider"]) + # Gemini pricing lives under "google-aistudio", not plain "google". + return ( + {"total_cost": 0.9} if kwargs["provider"] == "google-aistudio" else None + ) + + with patch.object( + processing_mod, "load_raw_batch_results", return_value=[_openai_row(10, 5)] + ), patch.object( + processing_mod, "estimate_model_cost", side_effect=_side_effect + ): + _record_stage_cost( + MagicMock(), run, Stage.PRE_FILTER_TOPIC_RELEVANCE, batch_job, 1 + ) + + assert calls == ["google", "google-aistudio"] + assert run.execution["cost"]["pre_filter"]["topic_relevance"] == 0.9 + + class TestGetBatchProvider: def test_unsupported_provider_raises(self) -> None: session = MagicMock() diff --git a/backend/app/tests/assessment/test_routes.py b/backend/app/tests/assessment/test_routes.py index a2fdf9a65..3f2a3b2f1 100644 --- a/backend/app/tests/assessment/test_routes.py +++ b/backend/app/tests/assessment/test_routes.py @@ -80,6 +80,8 @@ def _run() -> SimpleNamespace: total_items=1, error_message=None, input=None, + execution=None, + post_processing_config=None, batch_job_id=None, inserted_at=datetime(2024, 1, 1), updated_at=datetime(2024, 1, 1), @@ -523,3 +525,196 @@ def test_csv_multi_run_returns_zip(self) -> None: ) assert isinstance(resp, StreamingResponse) assert resp.media_type == "application/zip" + + +# ─── Public builders — real logic (no builder mock) ───────────────────────── + +CONFIG_UUID = UUID("00000000-0000-0000-0000-000000000001") + + +def _run_row(status: object, error: str | None = None) -> SimpleNamespace: + """Child-run stub with real enum status for the count/stat helpers.""" + return SimpleNamespace( + id=22, + assessment_id=ASSESSMENT_UUID, + config_id=CONFIG_UUID, + config_version=1, + status=status, + total_items=3, + error_message=error, + updated_at=datetime(2024, 1, 1), + ) + + +def _dispatch_session(parent: object, dataset: object) -> MagicMock: + """Session whose .get() returns the parent Assessment / dataset by model type.""" + from app.models.assessment import Assessment + from app.models.evaluation import EvaluationDataset + + session = MagicMock() + + def _get(model: type, _key: object) -> object: + if model is Assessment: + return parent + if model is EvaluationDataset: + return dataset + return None + + session.get.side_effect = _get + return session + + +class TestBuildAssessmentPublic: + """_build_assessment_public must surface derived counts/stats/dataset info.""" + + def test_populates_derived_fields(self) -> None: + from app.api.routes.assessment.assessments import _build_assessment_public + from app.models.assessment import AssessmentStatus + + assessment = SimpleNamespace( + id=ASSESSMENT_UUID, + experiment_name="exp", + dataset_id=7, + status=AssessmentStatus.COMPLETED_WITH_ERRORS, + organization_id=1, + project_id=1, + inserted_at=datetime(2024, 1, 1), + updated_at=datetime(2024, 1, 1), + ) + runs = [ + _run_row(AssessmentStatus.COMPLETED), + _run_row(AssessmentStatus.FAILED, "boom"), + ] + session = MagicMock() + session.get.return_value = SimpleNamespace(id=7, name="ds") + + with patch( + "app.api.routes.assessment.assessments.get_assessment_runs_for_assessment", + return_value=runs, + ): + public = _build_assessment_public(session, assessment) + + assert public.dataset_id == 7 + assert public.dataset_name == "ds" + assert public.counts.total == 2 + assert public.counts.completed == 1 + assert public.counts.failed == 1 + assert len(public.run_stats) == 2 + assert public.error_message == "1 of 2 run(s) failed" + + +class TestBuildRunPublic: + """_build_run_public must read runtime state from the execution bag, not columns.""" + + def test_sources_from_bag_and_parent(self) -> None: + from app.api.routes.assessment.runs import _build_run_public + from app.models.assessment import AssessmentStatus, Stage, StageStatus + + run = SimpleNamespace( + id=22, + assessment_id=ASSESSMENT_UUID, + config_id=CONFIG_UUID, + config_version=1, + status=AssessmentStatus.PROCESSING, + total_items=10, + error_message=None, + execution={ + "stage": "L2_ASSESSMENT", + "stage_status": "PROCESSING", + "pipeline": {"steps": ["a"]}, + "prefilter_total_rows": 10, + "prefilter_total_passed": 7, + "prefilter_total_rejected": 3, + }, + post_processing_config={"sort": "score"}, + inserted_at=datetime(2024, 1, 1), + updated_at=datetime(2024, 1, 1), + ) + parent = SimpleNamespace( + experiment_name="exp", + dataset_id=7, + input={"prompt": "p"}, + ) + session = _dispatch_session(parent, SimpleNamespace(id=7, name="ds")) + + public = _build_run_public(session, run) + + assert public.stage == Stage.L2_ASSESSMENT + assert public.stage_status == StageStatus.PROCESSING + assert public.pipeline == {"steps": ["a"]} + assert public.prefilter_total_rows == 10 + assert public.prefilter_total_passed == 7 + assert public.prefilter_total_rejected == 3 + assert public.input == {"prompt": "p"} + assert public.post_processing_config == {"sort": "score"} + assert public.experiment_name == "exp" + assert public.dataset_id == 7 + assert public.dataset_name == "ds" + + def test_surfaces_cost_from_bag(self) -> None: + from app.api.routes.assessment.runs import _build_run_public + from app.models.assessment import AssessmentStatus + + run = SimpleNamespace( + id=22, + assessment_id=ASSESSMENT_UUID, + config_id=CONFIG_UUID, + config_version=1, + status=AssessmentStatus.COMPLETED, + total_items=10, + error_message=None, + execution={ + "cost": { + "pre_filter": {"topic_relevance": 0.5, "total": 0.5}, + "assessment": 1.25, + "total": 1.75, + "currency": "USD", + } + }, + post_processing_config=None, + inserted_at=datetime(2024, 1, 1), + updated_at=datetime(2024, 1, 1), + ) + parent = SimpleNamespace(experiment_name="exp", dataset_id=7, input=None) + session = _dispatch_session(parent, SimpleNamespace(id=7, name="ds")) + + public = _build_run_public(session, run) + + assert public.cost is not None + assert public.cost.pre_filter is not None + assert public.cost.pre_filter.topic_relevance == 0.5 + assert public.cost.pre_filter.total == 0.5 + assert public.cost.assessment == 1.25 + assert public.cost.total == 1.75 + assert public.cost.currency == "USD" + + def test_empty_execution_bag_does_not_raise(self) -> None: + """Regression: a run with no execution bag must not 500 (was AttributeError).""" + from app.api.routes.assessment.runs import _build_run_public + from app.models.assessment import AssessmentStatus + + run = SimpleNamespace( + id=22, + assessment_id=ASSESSMENT_UUID, + config_id=CONFIG_UUID, + config_version=1, + status=AssessmentStatus.PENDING, + total_items=0, + error_message=None, + execution=None, + post_processing_config=None, + inserted_at=datetime(2024, 1, 1), + updated_at=datetime(2024, 1, 1), + ) + parent = SimpleNamespace(experiment_name="exp", dataset_id=7, input=None) + session = _dispatch_session(parent, SimpleNamespace(id=7, name="ds")) + + public = _build_run_public(session, run) + + assert public.stage is None + assert public.stage_status is None + assert public.pipeline is None + assert public.prefilter_total_rows is None + assert public.input is None + assert public.post_processing_config is None + assert public.cost is None diff --git a/backend/app/tests/assessment/test_service.py b/backend/app/tests/assessment/test_service.py index 7460676e9..5cb3e1b55 100644 --- a/backend/app/tests/assessment/test_service.py +++ b/backend/app/tests/assessment/test_service.py @@ -103,7 +103,7 @@ def test_config_resolution_failure(self) -> None: return_value=_make_dataset(), ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(None, "missing"), ), _assessment_config_crud_patch(), @@ -119,7 +119,7 @@ def test_config_resolution_failure(self) -> None: def test_rejects_unsupported_provider(self) -> None: session = MagicMock() request = _make_request(UUID("00000000-0000-0000-0000-000000000001")) - config_blob = SimpleNamespace(completion=SimpleNamespace(provider="sarvamai")) + config_blob = SimpleNamespace(assessment=SimpleNamespace(provider="sarvamai")) with ( patch( @@ -127,7 +127,7 @@ def test_rejects_unsupported_provider(self) -> None: return_value=_make_dataset(), ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( @@ -154,7 +154,7 @@ def test_google_provider_is_supported(self) -> None: assessment.id = ASSESSMENT_ID run = _make_run() config_blob = SimpleNamespace( - completion=SimpleNamespace(provider="google", params={"model": "gemini"}) + assessment=SimpleNamespace(provider="google", params={"model": "gemini"}) ) with ( @@ -163,7 +163,7 @@ def test_google_provider_is_supported(self) -> None: return_value=dataset, ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( @@ -210,7 +210,7 @@ def test_supported_batch_providers_are_accepted(self, provider: str) -> None: assessment.id = ASSESSMENT_ID run = _make_run() config_blob = SimpleNamespace( - completion=SimpleNamespace(provider=provider, params={"model": "m"}) + assessment=SimpleNamespace(provider=provider, params={"model": "m"}) ) with ( @@ -219,7 +219,7 @@ def test_supported_batch_providers_are_accepted(self, provider: str) -> None: return_value=dataset, ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( @@ -252,7 +252,7 @@ def test_anthropic_provider_is_supported(self) -> None: assessment.id = ASSESSMENT_ID run = _make_run() config_blob = SimpleNamespace( - completion=SimpleNamespace( + assessment=SimpleNamespace( provider="anthropic", params={"model": "claude-opus-4-8"} ) ) @@ -263,7 +263,7 @@ def test_anthropic_provider_is_supported(self) -> None: return_value=dataset, ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( @@ -298,7 +298,7 @@ def test_defaults_missing_provider_to_openai(self) -> None: assessment.id = ASSESSMENT_ID run = _make_run() config_blob = SimpleNamespace( - completion=SimpleNamespace(provider=None, params={"model": "gpt-4.1-mini"}) + assessment=SimpleNamespace(provider=None, params={"model": "gpt-4.1-mini"}) ) with ( @@ -307,7 +307,7 @@ def test_defaults_missing_provider_to_openai(self) -> None: return_value=dataset, ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( @@ -355,7 +355,7 @@ def test_rejects_default_tagged_config(self) -> None: ), patch("app.services.assessment.service.ConfigCrud", return_value=crud), patch( - "app.services.assessment.service.resolve_evaluation_config" + "app.services.assessment.service.resolve_assessment_config_blob" ) as resolve, ): with pytest.raises( @@ -381,7 +381,7 @@ def test_dispatches_one_celery_task_per_config(self) -> None: assessment.id = ASSESSMENT_ID run = _make_run() config_blob = SimpleNamespace( - completion=SimpleNamespace( + assessment=SimpleNamespace( provider="openai", params={"model": "gpt-4.1-mini"} ) ) @@ -392,7 +392,7 @@ def test_dispatches_one_celery_task_per_config(self) -> None: return_value=dataset, ), patch( - "app.services.assessment.service.resolve_evaluation_config", + "app.services.assessment.service.resolve_assessment_config_blob", return_value=(config_blob, None), ), patch( diff --git a/docs/wiki/modules/assessment.md b/docs/wiki/modules/assessment.md index e8e0b4186..ab06950c5 100644 --- a/docs/wiki/modules/assessment.md +++ b/docs/wiki/modules/assessment.md @@ -24,7 +24,7 @@ Config version (tag=ASSESSMENT, `models/config/assessment_blob.py`) owns system - `services/assessment/` — legacy RUN pipeline (service, stages, processing, batch, cron, tasks) - `services/assessment/api/` — API-client pipeline: `submission.py` (submit), `batch.py` (staged provider batches — gate pre-filters → pass-through → assessment, over `core/batch`; `PREFILTER_VERDICT_SCHEMA`), `results.py` (builds `AssessmentBatchResult`), `callbacks.py` (webhook) - `crud/assessment/api.py` — new API-client crud (method-based Assessment/AssessmentRun writes): `create_assessment`, `set_assessment_job`, `create_execution`, `set_execution_batch_job`, `update_status`, `list_executions` (no `get_assessment` — delivery is webhook-only, so there is no request-time fetch). Namespaced under `api` (`from app.crud.assessment import api`) to avoid colliding with the legacy `create_assessment`. -- `crud/assessment/{core,cron,processing,batch}.py` — legacy RUN pipeline crud. RUN runtime for the dropped columns (`stage`/`stage_status`/`pipeline`/`stage_batches`/`prefilter_total_*`/`object_store_url`) now lives in the `assessment_run.execution` bag via `core._read_exec`/`_write_exec`; `status` is the `AssessmentStatus` enum; the RUN input binding lives on the parent `assessment.input`. `batch.py` builds per-row prompts (`{column}` substitution from the parent `InputBinding.prompt`). +- `crud/assessment/{core,cron,processing,batch}.py` — legacy RUN pipeline crud. RUN runtime for the dropped columns (`stage`/`stage_status`/`pipeline`/`stage_batches`/`prefilter_total_*`/`object_store_url`/`cost`) now lives in the `assessment_run.execution` bag via `core._read_exec`/`_write_exec`; `status` is the `AssessmentStatus` enum; the RUN input binding lives on the parent `assessment.input`. `batch.py` builds per-row prompts (`{column}` substitution from the parent `InputBinding.prompt`). `processing.py::process_run_batches` also records per-stage batch cost (`RunCost`, from `model_config.pricing.batch` via `estimate_model_cost`) into `execution.cost`, surfaced on `AssessmentRunPublic.cost`. ## Async - Rides shared `core/batch/` provider batch infra + cron polling (same as evaluations).