diff --git a/backend/app/api/docs/config/create.md b/backend/app/api/docs/config/create.md index 4c3740468..a5570d7dc 100644 --- a/backend/app/api/docs/config/create.md +++ b/backend/app/api/docs/config/create.md @@ -89,7 +89,7 @@ shape is documented here: A `strict: true` column must be present and non-blank in every submission row; any other column may be omitted or blank (see the submit docs for per-row validation). Unknown keys in a column spec are rejected. Shared by the pre-filter and the assessment call. -* `assessment` (required) — the grading call. `provider` is `openai` | `google` | +* `assessment` (optional) — the grading call. `provider` is `openai` | `google` | `anthropic`, `type` is `"text"`. `params` carries the `model`, the `instructions` (system prompt), the **mandatory** per-row `submission` template (`{column}` placeholders must name `input_schema` keys), and an optional `json_output_schema` @@ -103,6 +103,10 @@ shape is documented here: failing verdict stops the chain and skips the assessment for that row; `false` = the verdict is just recorded). +At least one of `pre_filters.topic_relevance` or `assessment` is required (`422` otherwise). With only +`pre_filters` a run returns the verdicts and `output.assessment` is `null`; `stop_on_fail` +then has no effect, as no assessment follows. + ```json "config_blob": { "input_schema": { diff --git a/backend/app/api/docs/config/create_version.md b/backend/app/api/docs/config/create_version.md index f1922e54a..b4dd26e9d 100644 --- a/backend/app/api/docs/config/create_version.md +++ b/backend/app/api/docs/config/create_version.md @@ -31,7 +31,7 @@ anything you omit is carried forward. Send the **whole** `config_blob` every time. It replaces the previous version rather than merging onto it, so a column dropped from `input_schema`, a removed `json_output_schema` field, or an omitted `pre_filters` block is genuinely gone in the new version. A partial -body is rejected with `422`, because `input_schema` and `assessment` are mandatory. +body is rejected with `422`, because `input_schema` is mandatory and at least one of `pre_filters` or `assessment` is required. ```json { diff --git a/backend/app/crud/model_config.py b/backend/app/crud/model_config.py index e25e1a531..4a186c522 100644 --- a/backend/app/crud/model_config.py +++ b/backend/app/crud/model_config.py @@ -165,7 +165,8 @@ def validate_blob_completion_models( completion = ( blob.assessment if isinstance(blob, AssessmentConfigBlob) else blob.completion ) - _validate_completion_model_or_raise(session, completion) + if completion is not None: + _validate_completion_model_or_raise(session, completion) if isinstance(blob, AssessmentConfigBlob) and blob.pre_filters is not None: topic_relevance = blob.pre_filters.topic_relevance diff --git a/backend/app/models/config/assessment_blob.py b/backend/app/models/config/assessment_blob.py index 45cb88268..c0fddc74f 100644 --- a/backend/app/models/config/assessment_blob.py +++ b/backend/app/models/config/assessment_blob.py @@ -190,7 +190,16 @@ class AssessmentConfigBlob(SQLModel): ), ) pre_filters: AssessmentPreFilters | None = None - assessment: AssessmentCompletionConfig + assessment: AssessmentCompletionConfig | None = None + + @model_validator(mode="after") + def _require_a_stage(self): + has_pre_filter = bool(self.pre_filters and self.pre_filters.topic_relevance) + if not has_pre_filter and self.assessment is None: + raise ValueError( + "An ASSESSMENT config needs at least one of 'pre_filters' or 'assessment'." + ) + return self def input_columns(self) -> dict[str, dict[str, Any]]: return { @@ -208,9 +217,11 @@ def _validate_submission_placeholders(self): """ declared = set(self.input_schema) - blocks: list[tuple[str, str]] = [ - ("assessment", self.assessment.params.get("submission") or "") - ] + blocks: list[tuple[str, str]] = [] + if self.assessment is not None: + blocks.append( + ("assessment", self.assessment.params.get("submission") or "") + ) topic_relevance = self.pre_filters.topic_relevance if self.pre_filters else None if topic_relevance is not None and topic_relevance.params.get("submission"): blocks.append( @@ -236,7 +247,9 @@ def _validate_submission_placeholders(self): @model_validator(mode="after") def _validate_column_types_supported(self): """Every stage's provider must accept each media column type in input_schema.""" - stage_providers = {"assessment": self.assessment.provider} + stage_providers = {} + if self.assessment is not None: + stage_providers["assessment"] = self.assessment.provider if self.pre_filters and self.pre_filters.topic_relevance: stage_providers[ "pre_filter topic_relevance" diff --git a/backend/app/models/config/config.py b/backend/app/models/config/config.py index a9c30c033..42f8de994 100644 --- a/backend/app/models/config/config.py +++ b/backend/app/models/config/config.py @@ -171,10 +171,6 @@ def _validate_blob_for_tag(cls, blob: object, info: object) -> object: tag = raw_tag.value if isinstance(raw_tag, ConfigTag) else raw_tag is_assessment = tag == ConfigTag.ASSESSMENT.value - if is_assessment and "assessment" not in blob: - raise ValueError( - "An ASSESSMENT config requires an 'assessment' block in config_blob." - ) if not is_assessment and "completion" not in blob: raise ValueError( "A default config requires a 'completion' block in config_blob." diff --git a/backend/app/services/assessment/api/batch.py b/backend/app/services/assessment/api/batch.py index df9d407d3..968663692 100644 --- a/backend/app/services/assessment/api/batch.py +++ b/backend/app/services/assessment/api/batch.py @@ -62,8 +62,8 @@ from app.models.batch_job import BatchJob, BatchJobType from app.models.config.assessment_blob import ( ATTACHMENT_COLUMN_TYPES, + AssessmentCompletionConfig, AssessmentConfigBlob, - AssessmentPreFilters, TopicRelevanceFilter, ) from app.models.llm.constants import DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS @@ -104,6 +104,9 @@ def _google_gcp_credential( # Re-poll cadence for a stage's provider batch, mirroring the assessment cron interval. POLL_COUNTDOWN_SECONDS = settings.CRON_INTERVAL_MINUTES * 60 +# Row error when a provider batch returns neither an output nor an error line for a row. +MISSING_RESULT_ERROR = "No result returned by the provider" + class PipelineStage(StrEnum): """Stage names of the API-client pipeline; they key the config, the run state and the result.""" @@ -163,16 +166,13 @@ def is_supported_provider(provider_name: str) -> bool: return provider_name in _SUPPORTED_PROVIDERS -def build_pipeline( - pre_filters: AssessmentPreFilters | None, -) -> list[dict[str, str]]: - """Ordered stages: GATE pre-filters, then PASS-THROUGH pre-filters, then assessment.""" +def build_pipeline(blob: AssessmentConfigBlob) -> list[dict[str, str]]: + """Ordered stages the config declares: GATE pre-filters, PASS-THROUGH pre-filters, assessment.""" pre_filter_stages: list[tuple[PipelineStage, TopicRelevanceFilter]] = [] - if pre_filters is not None: - if pre_filters.topic_relevance is not None: - pre_filter_stages.append( - (PipelineStage.TOPIC_RELEVANCE, pre_filters.topic_relevance) - ) + if blob.pre_filters is not None and blob.pre_filters.topic_relevance is not None: + pre_filter_stages.append( + (PipelineStage.TOPIC_RELEVANCE, blob.pre_filters.topic_relevance) + ) gates = [ {"stage": stage.value, "kind": StageKind.GATE.value} @@ -184,11 +184,12 @@ def build_pipeline( for stage, pre_filter in pre_filter_stages if not pre_filter.stop_on_fail ] - return [ - *gates, - *passthrough, - {"stage": PipelineStage.ASSESSMENT.value, "kind": StageKind.ASSESSMENT.value}, - ] + assessment = ( + [{"stage": PipelineStage.ASSESSMENT.value, "kind": StageKind.ASSESSMENT.value}] + if blob.assessment is not None + else [] + ) + return [*gates, *passthrough, *assessment] def next_stage(pipeline: list[dict[str, str]], current: str) -> str | None: @@ -257,12 +258,19 @@ def stage_prompt(blob: AssessmentConfigBlob, stage: str) -> str | None: params.instructions with the item's columns + attachments as the user content). """ if stage == PipelineStage.ASSESSMENT: - return blob.assessment.params["submission"] + return _assessment_config(blob).params["submission"] return cast( "str | None", _prefilter_for_stage(blob, stage).params.get("submission") ) +def _assessment_config(blob: AssessmentConfigBlob) -> AssessmentCompletionConfig: + """The assessment block; the pipeline carries an assessment stage only when it exists.""" + if blob.assessment is None: + raise ValueError("[_assessment_config] Config has no assessment block") + return blob.assessment + + def _prefilter_for_stage( blob: AssessmentConfigBlob, stage: str ) -> TopicRelevanceFilter: @@ -282,7 +290,7 @@ def stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]: schema + gate directive layered on. """ if stage == PipelineStage.ASSESSMENT: - params = dict(blob.assessment.params) + params = dict(_assessment_config(blob).params) json_schema = params.pop("json_output_schema", None) # submission is a prompt concern, not a provider param. params.pop("submission", None) @@ -308,7 +316,8 @@ def stage_provider_model(blob: AssessmentConfigBlob, stage: str) -> tuple[str, s if stage == PipelineStage.TOPIC_RELEVANCE: pre_filter = _prefilter_for_stage(blob, stage) return pre_filter.provider, pre_filter.params["model"] - return blob.assessment.provider, blob.assessment.params["model"] + assessment = _assessment_config(blob) + return assessment.provider, assessment.params["model"] def build_stage_requests( @@ -810,7 +819,7 @@ def _finalize( from app.services.assessment.api.result_files import finalize_result_files from app.services.assessment.api.results import build_result - bag["stage"] = PipelineStage.ASSESSMENT.value + bag["stage"] = bag["pipeline"][-1]["stage"] bag["stage_status"] = StageStatus.COMPLETED.value api.save_execution_state(session=session, execution=execution, state=bag) @@ -1122,6 +1131,10 @@ def run_batch_stage( error = out.get("error") if error: stage_errors[str(idx)] = error + # A row absent from both the output and the error file would otherwise look clean. + for idx in _row_subset(bag, stage, kind, execution.total_items): + if idx not in parsed: + stage_errors[str(idx)] = MISSING_RESULT_ERROR bag.setdefault("stage_errors", {})[stage] = stage_errors bag.setdefault("stage_output_urls", {})[stage] = batch_job.raw_output_url diff --git a/backend/app/services/assessment/api/response.py b/backend/app/services/assessment/api/response.py index 1751e7e0f..2bda7bf91 100644 --- a/backend/app/services/assessment/api/response.py +++ b/backend/app/services/assessment/api/response.py @@ -170,7 +170,7 @@ def _assess( state["stage_status"] = StageStatus.COMPLETED.value pre_filter = PreFilter(**verdicts) if verdicts else None - if not state["gate_passed"][0]: + if not state["gate_passed"][0] or blob.assessment is None: return AssessmentResult(output=AssessmentOutput(pre_filter=pre_filter)) _begin_stage(session, execution, state, PipelineStage.ASSESSMENT.value) diff --git a/backend/app/services/assessment/api/results.py b/backend/app/services/assessment/api/results.py index e8d06b11e..e80553683 100644 --- a/backend/app/services/assessment/api/results.py +++ b/backend/app/services/assessment/api/results.py @@ -99,7 +99,9 @@ def build_result( verdicts = bag.get("verdicts") or {} outputs, load_error = _load_assessment_outputs(session, bag, assessment.project_id) # Rows the provider rejected are absent from the output dump; the bag kept their error. - row_errors = (bag.get("stage_errors") or {}).get(PipelineStage.ASSESSMENT.value, {}) + # Errors of the last stage: the assessment, or the pre-filter when there is none. + last_stage = (bag.get("pipeline") or [{}])[-1].get("stage") + row_errors = (bag.get("stage_errors") or {}).get(last_stage, {}) tr_verdicts = verdicts.get(PipelineStage.TOPIC_RELEVANCE.value, {}) diff --git a/backend/app/services/assessment/api/submission.py b/backend/app/services/assessment/api/submission.py index cf21a235a..0f07d5434 100644 --- a/backend/app/services/assessment/api/submission.py +++ b/backend/app/services/assessment/api/submission.py @@ -175,19 +175,17 @@ def _resolve_config( except ValidationError as exc: raise HTTPException(status_code=422, detail=exc.errors()) from exc - provider = blob.assessment.provider + # The last stage (the assessment when present) names the run's provider and model. + last_stage = batch_service.build_pipeline(blob)[-1]["stage"] + provider, model = batch_service.stage_provider_model(blob, last_stage) if not batch_service.is_supported_provider(provider): raise HTTPException( status_code=422, - detail=( - f"Config provider '{blob.assessment.provider}' is not supported for " - f"batch assessment." - ), + detail=f"Config provider '{provider}' is not supported for batch assessment.", ) - model = blob.assessment.params.get("model") if not model: raise HTTPException( - status_code=422, detail="Config assessment params must include a model." + status_code=422, detail="Config params must include a model." ) return blob, provider, model @@ -226,7 +224,7 @@ def _initial_state( model: str, total_items: int, ) -> BatchRunState: - pipeline = batch_service.build_pipeline(blob.pre_filters) + pipeline = batch_service.build_pipeline(blob) return { "pipeline": pipeline, "stage": pipeline[0]["stage"], diff --git a/backend/app/tests/api/routes/configs/test_config.py b/backend/app/tests/api/routes/configs/test_config.py index ec8c44615..093bf46e5 100644 --- a/backend/app/tests/api/routes/configs/test_config.py +++ b/backend/app/tests/api/routes/configs/test_config.py @@ -1,5 +1,6 @@ from uuid import uuid4 +import pytest from fastapi.testclient import TestClient from sqlmodel import Session @@ -567,24 +568,50 @@ def test_list_configs_with_query( assert "other-config" not in returned_names -def test_create_assessment_config_missing_assessment_block_gives_clear_error( +_PRE_FILTERS = { + "topic_relevance": { + "provider": "openai", + "params": {"model": "gpt-4o", "instructions": "On topic?"}, + "stop_on_fail": True, + } +} +_ASSESSMENT = { + "provider": "openai", + "type": "text", + "params": {"model": "gpt-4o", "instructions": "Grade it.", "submission": "{q}"}, +} + + +@pytest.mark.parametrize( + ("stages", "expected_status"), + [ + ({"pre_filters": _PRE_FILTERS}, 201), + ({"assessment": _ASSESSMENT}, 201), + ({"pre_filters": _PRE_FILTERS, "assessment": _ASSESSMENT}, 201), + ({}, 422), + ({"pre_filters": {}}, 422), + ], + ids=["pre_filter_only", "assessment_only", "both", "neither", "empty_pre_filters"], +) +def test_create_assessment_config_needs_a_stage( client: TestClient, user_api_key: TestAuthContext, + stages: dict, + expected_status: int, ) -> None: - """A malformed ASSESSMENT config (no 'assessment' block) returns a single - clear error on config_blob, not raw union noise.""" + """An ASSESSMENT config needs at least one of pre_filters / assessment.""" response = client.post( f"{settings.API_V1_STR}/configs", headers={"X-API-KEY": user_api_key.key}, json={ - "name": "bad-assessment", + "name": "stage-config", "tag": "ASSESSMENT", - "config_blob": {"pre_filters": {}}, + "config_blob": {"input_schema": {"q": {"type": "text"}}, **stages}, }, ) - assert response.status_code == 422 - errors = response.json()["errors"] - assert len(errors) == 1 - assert errors[0]["field"] == "config_blob" - assert "assessment" in errors[0]["message"].lower() + assert response.status_code == expected_status, response.text + if expected_status == 422: + errors = response.json()["errors"] + assert len(errors) == 1 + assert "pre_filters" in errors[0]["message"] diff --git a/backend/app/tests/api/routes/configs/test_version.py b/backend/app/tests/api/routes/configs/test_version.py index 1ce78cebd..298f44872 100644 --- a/backend/app/tests/api/routes/configs/test_version.py +++ b/backend/app/tests/api/routes/configs/test_version.py @@ -1,5 +1,6 @@ from uuid import uuid4 +import pytest from fastapi.testclient import TestClient from sqlmodel import Session @@ -8,6 +9,9 @@ from app.models.config.config import ConfigTag from app.models.llm.request import NativeCompletionConfig from app.tests.utils.auth import TestAuthContext +from app.models.config.assessment_blob import AssessmentConfigBlob +from app.models.config.config import ConfigTag +from app.tests.api.routes.configs.test_config import _ASSESSMENT, _PRE_FILTERS from app.tests.utils.test_data import ( create_test_config, create_test_project, @@ -901,3 +905,40 @@ def test_create_config_with_kaapi_provider_success( assert data["data"]["name"] == config_data["name"] assert data["data"]["version"]["config_blob"]["completion"]["provider"] == "openai" assert data["data"]["version"]["config_blob"]["completion"]["type"] == "text" + + +@pytest.mark.parametrize( + ("stages", "expected_status"), + [ + ({"pre_filters": _PRE_FILTERS}, 201), + ({}, 422), + ], + ids=["pre_filter_only", "neither"], +) +def test_create_assessment_version_needs_a_stage( + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + stages: dict, + expected_status: int, +) -> None: + """A new ASSESSMENT version is held to the same at-least-one-stage rule as create.""" + config = create_test_config( + db=db, + project_id=user_api_key.project_id, + name="assessment-config", + config_blob=AssessmentConfigBlob.model_validate( + {"input_schema": {"q": {"type": "text"}}, "assessment": _ASSESSMENT} + ), + tag=ConfigTag.ASSESSMENT, + ) + + response = client.post( + f"{settings.API_V1_STR}/configs/{config.id}/versions", + headers={"X-API-KEY": user_api_key.key}, + json={"config_blob": {"input_schema": {"q": {"type": "text"}}, **stages}}, + ) + + assert response.status_code == expected_status, response.text + if expected_status == 201: + assert response.json()["data"]["config_blob"]["assessment"] is None diff --git a/backend/app/tests/assessment/test_api_batch.py b/backend/app/tests/assessment/test_api_batch.py index 7750e6d03..df08b1840 100644 --- a/backend/app/tests/assessment/test_api_batch.py +++ b/backend/app/tests/assessment/test_api_batch.py @@ -27,6 +27,7 @@ from app.models.config.config import ConfigTag from app.services.assessment.api import batch as batch_service from app.services.assessment.api.batch import ( + MISSING_RESULT_ERROR, PipelineStage, StageKind, _build_batch_provider, @@ -145,7 +146,7 @@ def test_supported_and_unsupported(self) -> None: class TestBuildPipeline: def test_none_prefilters_is_assessment_only(self) -> None: - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) assert pipeline == [ { "stage": PipelineStage.ASSESSMENT.value, @@ -155,7 +156,7 @@ def test_none_prefilters_is_assessment_only(self) -> None: def test_gate_precedes_assessment(self) -> None: blob = _blob(topic_relevance=True) # stop_on_fail=True -> GATE - pipeline = build_pipeline(blob.pre_filters) + pipeline = build_pipeline(blob) assert [s["stage"] for s in pipeline] == [ PipelineStage.TOPIC_RELEVANCE.value, PipelineStage.ASSESSMENT.value, @@ -167,7 +168,7 @@ def test_gate_precedes_assessment(self) -> None: def test_passthrough_precedes_assessment(self) -> None: blob = _blob(topic_relevance=False) # stop_on_fail=False -> PASS_THROUGH - pipeline = build_pipeline(blob.pre_filters) + pipeline = build_pipeline(blob) assert [s["stage"] for s in pipeline] == [ PipelineStage.TOPIC_RELEVANCE.value, PipelineStage.ASSESSMENT.value, @@ -180,7 +181,7 @@ def test_passthrough_precedes_assessment(self) -> None: class TestNextStageAndKind: def test_next_stage_advances_then_none(self) -> None: - pipeline = build_pipeline(_blob(topic_relevance=True).pre_filters) + pipeline = build_pipeline(_blob(topic_relevance=True)) assert ( next_stage(pipeline, PipelineStage.TOPIC_RELEVANCE.value) == PipelineStage.ASSESSMENT.value @@ -188,7 +189,7 @@ def test_next_stage_advances_then_none(self) -> None: assert next_stage(pipeline, PipelineStage.ASSESSMENT.value) is None def test_stage_kind_lookup_and_missing_raises(self) -> None: - pipeline = build_pipeline(_blob(topic_relevance=True).pre_filters) + pipeline = build_pipeline(_blob(topic_relevance=True)) assert ( _stage_kind(pipeline, PipelineStage.TOPIC_RELEVANCE.value) == StageKind.GATE.value @@ -619,7 +620,7 @@ def test_gate_then_assessment_to_completed_callback(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(blob.pre_filters) + pipeline = build_pipeline(blob) bag = _bag_for(pipeline, total=2) assessment, execution = _seed_assessment( db, @@ -762,7 +763,7 @@ def test_terminal_status_returns_no_requeue(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) _, execution = _seed_assessment( db, org_id=auth.organization_id, @@ -808,7 +809,7 @@ def test_processing_batch_missing_fails(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) bag = _bag_for(pipeline, total=1, callback="") bag["stage_status"] = StageStatus.PROCESSING.value bag["stage_batches"] = {} # no batch job recorded for the stage @@ -1098,7 +1099,7 @@ class TestSubmitStageEmptySubset: def test_all_gated_out_skips_submit(self, db) -> None: auth = get_user_test_auth_context(db) blob = _blob() - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) bag = _bag_for(pipeline, total=2) bag["gate_passed"] = [False, False] batch_input = BatchInput(data=[{"a": "1"}, {"a": "2"}]) @@ -1152,7 +1153,7 @@ def test_every_row_gated_out_reaches_terminal_and_delivers_once(self, db) -> Non config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(blob.pre_filters) + pipeline = build_pipeline(blob) bag = _bag_for(pipeline, total=2) assessment, execution = _seed_assessment( db, @@ -1263,9 +1264,144 @@ def call_tick(): assert len(delivered) == 1 +class TestOptionalStages: + """A config may carry only a pre-filter, only an assessment, or both.""" + + @staticmethod + def _blob_with(*, pre_filter: bool, assessment: bool) -> AssessmentConfigBlob: + raw = _blob_dict(topic_relevance=True if pre_filter else None) + if not assessment: + raw.pop("assessment") + return AssessmentConfigBlob.model_validate(raw) + + @pytest.mark.parametrize( + ("pre_filter", "assessment", "stages"), + [ + (True, False, ["topic_relevance"]), + (False, True, ["assessment"]), + (True, True, ["topic_relevance", "assessment"]), + ], + ) + def test_pipeline_holds_only_present_stages( + self, pre_filter, assessment, stages + ) -> None: + blob = self._blob_with(pre_filter=pre_filter, assessment=assessment) + pipeline = build_pipeline(blob) + assert [step["stage"] for step in pipeline] == stages + + def test_pass_through_only_pipeline_never_gates(self, db) -> None: + raw = _blob_dict(topic_relevance=False) + raw.pop("assessment") + blob = AssessmentConfigBlob.model_validate(raw) + pipeline = build_pipeline(blob) + assert pipeline == [{"stage": "topic_relevance", "kind": "PASS_THROUGH"}] + + auth = get_user_test_auth_context(db) + assessment, execution = _seed_assessment( + db, + org_id=auth.organization_id, + project_id=auth.project_id, + config_id=_make_config_id(db, auth.project_id), + bag=_bag_for(pipeline, total=1, callback=""), + data=[{"a": "1"}], + status=AssessmentStatus.PROCESSING, + ) + bag = dict(execution.execution) + bag["verdicts"] = { + "topic_relevance": {"0": {"verdict": False, "reasoning": ""}} + } + + _finalize(db, execution, assessment, bag) + + db.refresh(assessment) + assert assessment.status == AssessmentStatus.COMPLETED + result = build_result(session=db, assessment=assessment, execution=execution) + assert result.items[0].output.assessment is None + assert result.items[0].output.pre_filter.topic_relevance.verdict is False + assert result.counts.filtered == 0 + + def test_neither_stage_is_rejected(self) -> None: + with pytest.raises(ValueError, match="at least one of"): + self._blob_with(pre_filter=False, assessment=False) + + def test_pre_filter_only_finalizes_on_its_last_stage(self, db) -> None: + auth = get_user_test_auth_context(db) + pipeline = build_pipeline(self._blob_with(pre_filter=True, assessment=False)) + assessment, execution = _seed_assessment( + db, + org_id=auth.organization_id, + project_id=auth.project_id, + config_id=_make_config_id(db, auth.project_id), + bag=_bag_for(pipeline, total=2, callback=""), + data=[{"a": "1"}, {"a": "2"}], + status=AssessmentStatus.PROCESSING, + ) + bag = dict(execution.execution) + bag["verdicts"] = { + "topic_relevance": { + "0": {"verdict": True, "reasoning": "ok"}, + "1": {"verdict": False, "reasoning": "off topic"}, + } + } + bag["gate_passed"] = [True, False] + + _finalize(db, execution, assessment, bag) + + db.refresh(assessment) + db.refresh(execution) + assert assessment.status == AssessmentStatus.COMPLETED + assert execution.execution["stage"] == "topic_relevance" + result = build_result(session=db, assessment=assessment, execution=execution) + assert [item.output.assessment for item in result.items] == [None, None] + assert result.items[0].output.pre_filter.topic_relevance.verdict is True + assert result.items[1].output.pre_filter.topic_relevance.verdict is False + assert result.counts.filtered == 1 + assert result.counts.assessed == 0 + + @pytest.mark.parametrize( + ("errored_rows", "expected"), + [ + ({"0": "provider boom"}, AssessmentStatus.COMPLETED_WITH_ERRORS), + ({"0": "boom", "1": "boom"}, AssessmentStatus.FAILED), + ], + ) + def test_pre_filter_only_surfaces_a_failed_pre_filter_call( + self, db, errored_rows, expected + ) -> None: + auth = get_user_test_auth_context(db) + pipeline = build_pipeline(self._blob_with(pre_filter=True, assessment=False)) + assessment, execution = _seed_assessment( + db, + org_id=auth.organization_id, + project_id=auth.project_id, + config_id=_make_config_id(db, auth.project_id), + bag=_bag_for(pipeline, total=2, callback=""), + data=[{"a": "1"}, {"a": "2"}], + status=AssessmentStatus.PROCESSING, + ) + bag = dict(execution.execution) + # A failed call fails open: the verdict reads "passed" but the error is recorded. + bag["verdicts"] = { + "topic_relevance": { + "0": {"verdict": True, "reasoning": ""}, + "1": {"verdict": True, "reasoning": ""}, + } + } + bag["stage_errors"] = {"topic_relevance": errored_rows} + + _finalize(db, execution, assessment, bag) + + db.refresh(assessment) + assert assessment.status == expected + result = build_result(session=db, assessment=assessment, execution=execution) + assert [bool(item.error) for item in result.items] == [ + str(i) in errored_rows for i in range(2) + ] + + class TestFinalizeAndFail: def _setup(self, db, auth, callback): - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) bag = _bag_for(pipeline, total=1, callback=callback) return _seed_assessment( db, @@ -1362,9 +1498,7 @@ def _processing_execution(self, db, auth, *, stage_pipeline_prefilter=False): config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline( - blob.pre_filters if stage_pipeline_prefilter else None - ) + pipeline = build_pipeline(blob if stage_pipeline_prefilter else _blob()) job = _make_batch_job( db, org_id=auth.organization_id, project_id=auth.project_id ) @@ -1390,6 +1524,71 @@ def _tick(self, db, execution, auth): project_id=auth.project_id, ) + def test_row_without_a_provider_result_gets_an_error(self, db) -> None: + auth = get_user_test_auth_context(db) + raw = _blob_dict(topic_relevance=True) + raw.pop("assessment") + blob = AssessmentConfigBlob.model_validate(raw) + config = create_test_config( + db, + project_id=auth.project_id, + name=f"assess-{random_lower_string()}", + config_blob=blob, + tag=ConfigTag.ASSESSMENT, + ) + pipeline = build_pipeline(blob) + job = _make_batch_job( + db, org_id=auth.organization_id, project_id=auth.project_id + ) + bag = _bag_for(pipeline, total=2, callback="") + bag["stage_status"] = StageStatus.PROCESSING.value + bag["stage_batches"] = {"topic_relevance": job.id} + assessment, execution = _seed_assessment( + db, + org_id=auth.organization_id, + project_id=auth.project_id, + config_id=config.id, + bag=bag, + data=[{"a": "one"}, {"a": "two"}], + status=AssessmentStatus.PROCESSING, + ) + only_row_0 = [ + { + "custom_id": "row_0", + "response": { + "status_code": 200, + "body": { + "output_text": json.dumps({"verdict": True, "reasoning": ""}) + }, + }, + } + ] + with ( + patch( + "app.services.assessment.api.batch.Session", + lambda _e: _SessionCtx(db), + ), + patch("app.services.assessment.api.batch._build_batch_provider"), + patch( + "app.services.assessment.api.batch._poll_outcome", + return_value=("completed", only_row_0), + ), + patch("app.services.assessment.api.result_files.record_stage_dump"), + patch("app.services.assessment.api.result_files.finalize_result_files"), + ): + assert self._tick(db, execution, auth) == {"requeue": False} + + db.refresh(execution) + db.refresh(assessment) + assert execution.execution["stage_errors"]["topic_relevance"] == { + "1": MISSING_RESULT_ERROR + } + assert assessment.status == AssessmentStatus.COMPLETED_WITH_ERRORS + result = build_result(session=db, assessment=assessment, execution=execution) + assert result.items[0].error is None + assert result.items[1].error == MISSING_RESULT_ERROR + assert result.counts.errors == 1 + def test_poll_processing_requeues(self, db) -> None: auth = get_user_test_auth_context(db) _, execution = self._processing_execution(db, auth) @@ -1495,7 +1694,7 @@ def test_pending_submit_value_error_fails_run(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) _, execution = _seed_assessment( db, org_id=auth.organization_id, @@ -1530,7 +1729,7 @@ def test_parent_assessment_missing(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) _, execution = _seed_assessment( db, org_id=auth.organization_id, @@ -1570,7 +1769,7 @@ def _pending_assessment_execution(self, db, auth): config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) return _seed_assessment( db, org_id=auth.organization_id, @@ -1663,7 +1862,7 @@ def test_advance_submit_non_value_error_fails_run(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(blob.pre_filters) + pipeline = build_pipeline(blob) job = _make_batch_job( db, org_id=auth.organization_id, project_id=auth.project_id ) @@ -1732,7 +1931,7 @@ def test_poll_exception_still_requeues_not_fails(self, db) -> None: config_blob=blob, tag=ConfigTag.ASSESSMENT, ) - pipeline = build_pipeline(None) + pipeline = build_pipeline(_blob()) job = _make_batch_job( db, org_id=auth.organization_id, project_id=auth.project_id ) diff --git a/backend/app/tests/assessment/test_api_read.py b/backend/app/tests/assessment/test_api_read.py index 079126277..81fa007be 100644 --- a/backend/app/tests/assessment/test_api_read.py +++ b/backend/app/tests/assessment/test_api_read.py @@ -227,6 +227,33 @@ def test_gated_rows_carry_their_verdict(self, db) -> None: assert gated.output.pre_filter.topic_relevance.verdict is False assert detail.counts.filtered == 1 + def test_pre_filter_only_run_reports_its_own_row_errors(self, db) -> None: + auth = get_user_test_auth_context(db) + bag = _bag( + 2, + pipeline=[{"stage": "topic_relevance", "kind": "GATE"}], + stage="topic_relevance", + stage_status="COMPLETED", + verdicts={ + "topic_relevance": { + "0": {"verdict": True, "reasoning": "on topic"}, + "1": {"verdict": True, "reasoning": ""}, + } + }, + stage_errors={"topic_relevance": {"1": "provider boom"}}, + ) + assessment, _ = _seed(db, auth, total=2, bag=bag) + + with patch(_ROWS, _serving([{"a": "1"}, {"a": "2"}])): + detail = build_detail(session=db, assessment=assessment) + + assert [item.output.assessment for item in detail.items] == [None, None] + assert detail.items[0].error is None + assert detail.items[1].error == "provider boom" + assert detail.stages == ["topic_relevance"] + assert detail.counts.errors == 1 + assert detail.counts.assessed == 0 + def test_input_shorter_than_total_items(self, db) -> None: auth = get_user_test_auth_context(db) assessment, _ = _seed(db, auth, total=3) diff --git a/backend/app/tests/assessment/test_api_response.py b/backend/app/tests/assessment/test_api_response.py index 53cf19c3c..5c949f4ce 100644 --- a/backend/app/tests/assessment/test_api_response.py +++ b/backend/app/tests/assessment/test_api_response.py @@ -54,6 +54,13 @@ def _blob(**kwargs) -> AssessmentConfigBlob: return AssessmentConfigBlob.model_validate(_blob_dict(**kwargs)) +def _blob_with(*, pre_filter: bool, assessment: bool) -> AssessmentConfigBlob: + raw = _blob_dict(topic_relevance=True if pre_filter else None) + if not assessment: + raw.pop("assessment") + return AssessmentConfigBlob.model_validate(raw) + + def _seed_response(db, auth, blob): config = create_test_config( db, @@ -79,7 +86,7 @@ def _seed_response(db, auth, blob): api.save_execution_state( session=db, execution=execution, - state=_bag_for(build_pipeline(blob.pre_filters), total=1), + state=_bag_for(build_pipeline(blob), total=1), ) api.update_status(session=db, obj=assessment, status=AssessmentStatus.PROCESSING) api.update_status(session=db, obj=execution, status=AssessmentStatus.PROCESSING) @@ -285,3 +292,72 @@ def test_missing_run_is_a_no_op(self, db) -> None: delivered = _run(db, 10**9, _stage_output) assert delivered == [] + + +class TestOptionalStages: + @pytest.mark.parametrize( + ("pre_filter", "assessment", "called", "stage"), + [ + (True, False, ["topic_relevance"], "topic_relevance"), + (False, True, ["assessment"], "assessment"), + (True, True, ["topic_relevance", "assessment"], "assessment"), + ], + ids=["pre_filter_only", "assessment_only", "both"], + ) + def test_runs_only_the_configured_stages( + self, db, pre_filter, assessment, called, stage + ) -> None: + auth = get_user_test_auth_context(db) + parent, execution = _seed_response( + db, auth, _blob_with(pre_filter=pre_filter, assessment=assessment) + ) + delivered = _run(db, execution.id, _stage_output) + + db.refresh(parent) + db.refresh(execution) + assert parent.status == AssessmentStatus.COMPLETED + assert execution.execution["stage"] == stage + assert list(execution.execution.get("verdicts", {})) == ( + ["topic_relevance"] if pre_filter else [] + ) + (call,) = delivered + output = call["result"].output + assert (output.assessment is not None) is assessment + assert (output.pre_filter is not None) is pre_filter + assert sorted(execution.execution.get("counters", {})) == sorted( + s for s in called if s != "assessment" + ) + + def test_pre_filter_only_rejected_row_keeps_its_verdict(self, db) -> None: + auth = get_user_test_auth_context(db) + parent, execution = _seed_response( + db, auth, _blob_with(pre_filter=True, assessment=False) + ) + (call,) = _run( + db, execution.id, lambda stage: _stage_output(stage, verdict=False) + ) + + db.refresh(parent) + db.refresh(execution) + assert parent.status == AssessmentStatus.COMPLETED + assert execution.execution["gate_passed"] == [False] + output = call["result"].output + assert output.assessment is None + assert output.pre_filter.topic_relevance.verdict is False + + def test_pre_filter_only_failed_call_is_reported(self, db) -> None: + auth = get_user_test_auth_context(db) + parent, execution = _seed_response( + db, auth, _blob_with(pre_filter=True, assessment=False) + ) + (call,) = _run( + db, execution.id, lambda stage: _stage_output(stage, error="boom") + ) + + db.refresh(parent) + db.refresh(execution) + assert parent.status == AssessmentStatus.COMPLETED_WITH_ERRORS + assert execution.error_message == "topic_relevance: boom" + verdict = call["result"].output.pre_filter.topic_relevance + assert verdict.error == "boom" + assert verdict.verdict is True # fails open, as in BATCH diff --git a/backend/app/tests/assessment/test_api_submission.py b/backend/app/tests/assessment/test_api_submission.py index 0f86a5eed..8c94098cb 100644 --- a/backend/app/tests/assessment/test_api_submission.py +++ b/backend/app/tests/assessment/test_api_submission.py @@ -39,7 +39,13 @@ def _open(**_): def _assessment_config( - db, project_id, *, provider="openai", model="gpt-4o", input_schema=None + db, + project_id, + *, + provider="openai", + model="gpt-4o", + input_schema=None, + pre_filter_only=False, ): """ASSESSMENT config with the mandatory typed ``input_schema`` (default one text column ``a``); callers pass ``input_schema`` to exercise row-validation branches.""" @@ -51,12 +57,19 @@ def _assessment_config( }, "submission": "Assess this submission.", } - blob = AssessmentConfigBlob.model_validate( - { - "input_schema": input_schema or {"a": {"type": "text"}}, - "assessment": {"provider": provider, "type": "text", "params": params}, + raw = { + "input_schema": input_schema or {"a": {"type": "text"}}, + "assessment": {"provider": provider, "type": "text", "params": params}, + } + if pre_filter_only: + raw.pop("assessment") + raw["pre_filters"] = { + "topic_relevance": { + "provider": provider, + "params": {"model": model, "instructions": "On topic?"}, + } } - ) + blob = AssessmentConfigBlob.model_validate(raw) return create_test_config( db, project_id=project_id, @@ -163,6 +176,28 @@ def test_response_without_callback_url_is_rejected(self) -> None: } ) + def test_pre_filter_only_config_runs_just_the_pre_filter(self, db) -> None: + auth = get_user_test_auth_context(db) + config = _assessment_config(db, auth.project_id, pre_filter_only=True) + request = _request(config, [{"a": "one"}]) + + with patch("app.celery.tasks.job_execution.run_assessment_api_batch") as task: + response = submission.submit( + session=db, + request=request, + organization_id=auth.organization_id, + project_id=auth.project_id, + ) + + task.delay.assert_called_once() + (execution,) = api.list_executions( + session=db, assessment_id=response.assessment_id + ) + bag = execution.execution + assert [step["stage"] for step in bag["pipeline"]] == ["topic_relevance"] + assert bag["stage"] == "topic_relevance" + assert bag["model"] == "gpt-4o" + def test_immediate_dispatch_keeps_trace_propagation(self, db) -> None: auth = get_user_test_auth_context(db) config = _assessment_config(db, auth.project_id) diff --git a/docs/wiki/modules/assessment.md b/docs/wiki/modules/assessment.md index 3efb48ada..91e56e333 100644 --- a/docs/wiki/modules/assessment.md +++ b/docs/wiki/modules/assessment.md @@ -17,7 +17,7 @@ All paths relative to `backend/app/`. | `assessment_run` (AssessmentRun; child — one config execution, BATCH/RUN/RESPONSE; FK → assessment, config, batch_job) | `models/assessment/assessment.py` | | `assessment_submission` (AssessmentSubmission; an uploaded CSV/XLSX a run can read its rows from; FK → org, project) | `models/assessment/submission.py` | -Config version (tag=ASSESSMENT, `models/config/assessment_blob.py`) owns system / pre-filters / params / schemas: `input_schema` is a **top-level** field on the blob (sibling of `pre_filters`/`assessment`, **mandatory, non-empty** per-column spec `{type, format}` for the BATCH `data` rows — `type` is required per column; `strict: true` means the column must be present and non-blank in every submission row; it is **off by default** because the console declares every sheet column (the backend rejects undeclared ones) and real sheets have blanks — a blank on a non-strict column passes and `build_rows` fills omitted keys with `""` so the placeholder still resolves; note the config store persists the validated blob *with defaults*, so a saved version pins the default in force at save time; `InputColumn` is `extra=forbid`, so a misspelt flag fails at config save; attachment columns (`image`/`pdf`/`video`/`audio`) are url-format only, and `video`/`audio` are Google-only (AI Studio + GCP share `build_google_jsonl`) — a config whose assessment or pre-filter provider is `openai`/`anthropic` is rejected at save when `input_schema` declares one (`UNSUPPORTED_COLUMN_TYPES_BY_PROVIDER`), so no run reaches the builders with such a column) — it describes the shared input rows once, so both the pre-filter and assessment consumers read the same schema. `submission` (the per-row prompt template with `{column}` placeholders — **mandatory** on `assessment.params`, optional on each pre-filter's `params`; every `{placeholder}` is validated at config save against the top-level `input_schema` keys, and an unknown placeholder rejects the save). `json_output_schema` stays in `assessment.params` (assessment-specific, object-typed structured-output schema, omit for free text). The only API-client pre-filter is `topic_relevance` (duplicate_detection was removed from the API-client pipeline; it survives only in the legacy RUN pipeline); it carries its own `provider` (default `openai`) + `params` (TextLLMParams: model, temperature, ...) and runs its own llm call; its criteria live in `params.instructions` (a **mandatory** field, same shape as the assessment call — pre-filters no longer have a top-level `prompt`/`content`) and it may carry its own `params.submission` template. The prompt template lives on the config, not the request. Strict input types `ResponseInput` (RESPONSE, `{data: }`) / `BatchInput` (BATCH) carry no `query`; they discriminate on the type of `data` (an object ⇒ RESPONSE, a list or a `submission_doc_id` ⇒ BATCH) with `extra=forbid` rejecting a stray `query` — no `mode` tag (`models/assessment/assessment_api.py`). `BatchInput` takes the rows **either** inline as `data` (a list of submission rows, each a flat column→string map, an attachment column's value being a url string) **or** by reference as `submission_doc_id` (an `assessment_submission` id); a `model_validator` requires exactly one, so both-or-neither is a 422. A `submission_doc_id` is resolved at submit: its rows are read, projected onto the schema's columns (a sheet column `input_schema` does not name is dropped, so the schema is the selection and the console need only declare the `@`-referenced columns), validated against `input_schema` like inline ones, and the run streams them from the uploaded submission file instead of copying them, so downstream stays one code path. `assessment.submission_id` records which submission it came from and `delete_submission` refuses while a BATCH run still points at it. That column is set by RUN and by a `submission_doc_id` BATCH alike; it stays NULL only when BATCH sent rows inline. Legacy RUN runtime lives in `assessment_run.execution` (`RunExecution`). +Config version (tag=ASSESSMENT, `models/config/assessment_blob.py`) owns system / pre-filters / params / schemas: `input_schema` is a **top-level** field on the blob (sibling of `pre_filters`/`assessment`, **mandatory, non-empty** per-column spec `{type, format}` for the BATCH `data` rows — `type` is required per column; `strict: true` means the column must be present and non-blank in every submission row; it is **off by default** because the console declares every sheet column (the backend rejects undeclared ones) and real sheets have blanks — a blank on a non-strict column passes and `build_rows` fills omitted keys with `""` so the placeholder still resolves; note the config store persists the validated blob *with defaults*, so a saved version pins the default in force at save time; `InputColumn` is `extra=forbid`, so a misspelt flag fails at config save; attachment columns (`image`/`pdf`/`video`/`audio`) are url-format only, and `video`/`audio` are Google-only (AI Studio + GCP share `build_google_jsonl`) — a config whose assessment or pre-filter provider is `openai`/`anthropic` is rejected at save when `input_schema` declares one (`UNSUPPORTED_COLUMN_TYPES_BY_PROVIDER`), so no run reaches the builders with such a column) — it describes the shared input rows once, so both the pre-filter and assessment consumers read the same schema. `submission` (the per-row prompt template with `{column}` placeholders — **mandatory** on `assessment.params`, optional on each pre-filter's `params`; every `{placeholder}` is validated at config save against the top-level `input_schema` keys, and an unknown placeholder rejects the save). `json_output_schema` stays in `assessment.params` (assessment-specific, object-typed structured-output schema, omit for free text). The only API-client pre-filter is `topic_relevance` (duplicate_detection was removed from the API-client pipeline; it survives only in the legacy RUN pipeline); it carries its own `provider` (default `openai`) + `params` (TextLLMParams: model, temperature, ...) and runs its own llm call; its criteria live in `params.instructions` (a **mandatory** field, same shape as the assessment call — pre-filters no longer have a top-level `prompt`/`content`) and it may carry its own `params.submission` template. The prompt template lives on the config, not the request. Strict input types `ResponseInput` (RESPONSE, `{data: }`) / `BatchInput` (BATCH) carry no `query`; they discriminate on the type of `data` (an object ⇒ RESPONSE, a list or a `submission_doc_id` ⇒ BATCH) with `extra=forbid` rejecting a stray `query` — no `mode` tag (`models/assessment/assessment_api.py`). `BatchInput` takes the rows **either** inline as `data` (a list of submission rows, each a flat column→string map, an attachment column's value being a url string) **or** by reference as `submission_doc_id` (an `assessment_submission` id); a `model_validator` requires exactly one, so both-or-neither is a 422. A `submission_doc_id` is resolved at submit: its rows are read, projected onto the schema's columns (a sheet column `input_schema` does not name is dropped, so the schema is the selection and the console need only declare the `@`-referenced columns), validated against `input_schema` like inline ones, and the run streams them from the uploaded submission file instead of copying them, so downstream stays one code path. `assessment.submission_id` records which submission it came from and `delete_submission` refuses while a BATCH run still points at it. That column is set by RUN and by a `submission_doc_id` BATCH alike; it stays NULL only when BATCH sent rows inline. `assessment` and `pre_filters` are both optional but at least one is required (`_require_a_stage`, 422 at save): `build_pipeline(blob)` holds only the stages the config declares, a pre-filter-only run finalizes on its last stage and returns `output.assessment = null`, and the run's provider/model come from its last stage. Legacy RUN runtime lives in `assessment_run.execution` (`RunExecution`). `assessment.id` is a **UUID** (like config/job/llm_call). Per-item result = `AssessmentResult {output: {assessment, pre_filter}, error}` (no `metadata` — the provider/model/usage block was removed from the API-client output) where `output.assessment` = the LLM output parsed to an object when the config has a `json_output_schema`, else string (null for gated/failed rows), and `output.pre_filter` holds the `{topic_relevance}` verdict (`{verdict, reasoning}` or null) and is itself null when no pre-filter ran. The API-client BATCH rows do **not** live in `assessment.input` (that column is now RESPONSE/RUN only, NULL for BATCH). They are uploaded to `submission.jsonl` at submit and `assessment.submission_input` holds its `s3://` url: 3-6MB of JSONB was dragged along by every full-row `SELECT` of the assessment. `services/assessment/api/submission_store.py` owns the round trip: `upload_submission_rows` at submit, and the `open_submission_rows` context manager on read, which streams the JSONL line by line and closes the body on exit. Rows are read only inside `_submit_stage`, never on a poll, and only the rows the stage's subset needs are held (the count comes from `execution.total_items`, so an empty subset never opens the file). A storage read failure raises `SubmissionUnavailableError`, which requeues the task instead of failing the execution; a corrupt line is a `ValueError` and terminal. An upload failure at submit is a 503 (an assessment without its rows is unrunnable). `build_result` reads `execution.total_items` rather than re-deriving the count from the rows, so the terminal path never fetches them.