feat(assessment): Make pre-filter and assessment optional in the config

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": {

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
 {

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

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"

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."

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

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)

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, {})
 

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"],

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"]

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