feat(assessment): Migrate user prompt to config

backend/app/crud/model_config.py

@@ -167,18 +167,14 @@ def validate_blob_completion_models(
     _validate_completion_model_or_raise(session, completion)
 
     if isinstance(blob, AssessmentConfigBlob) and blob.pre_filters is not None:
-        pre_filters = (
-            blob.pre_filters.topic_relevance,
-            blob.pre_filters.duplicate_detection,
-        )
-        for flt in pre_filters:
-            if flt is not None:
-                _validate_model_or_raise(
-                    session,
-                    raw_provider=flt.provider,
-                    completion_type=CompletionType.TEXT,
-                    params=flt.params,
-                )
+        topic_relevance = blob.pre_filters.topic_relevance
+        if topic_relevance is not None:
+            _validate_model_or_raise(
+                session,
+                raw_provider=topic_relevance.provider,
+                completion_type=CompletionType.TEXT,
+                params=topic_relevance.params,
+            )
 
 
 def _validate_completion_model_or_raise(

backend/app/models/assessment/assessment_api.py

@@ -22,11 +22,10 @@
 
 
 class ResponseInput(SQLModel):
-    """RESPONSE input — a single `query` + optional attachments (no columns)."""
+    """RESPONSE input — optional attachments (no columns). The prompt lives in the config."""
 
     model_config = ConfigDict(extra="forbid")
 
-    query: str = Field(..., min_length=1, description="User query")
     attachments: list[Attachment] = Field(default_factory=list)
 
 
@@ -36,13 +35,10 @@ class ResponseInput(SQLModel):
 
 
 class BatchInput(SQLModel):
-    """BATCH input — a `query` template + a list of submission rows."""
+    """BATCH input — a list of submission rows. The prompt template lives in the config."""
 
     model_config = ConfigDict(extra="forbid")
 
-    query: str = Field(
-        ..., min_length=1, description="Query template with {column} placeholders"
-    )
     data: list[Submission] = Field(
         ..., min_length=1, description="Submission rows; one assessed item each"
     )
@@ -91,8 +87,8 @@ class BatchRunState(TypedDict):
 
 
 # Strict, tagless discrimination via extra=forbid: an input carrying `data` is a
-# BatchInput, one carrying `attachments` (or query-only) is a ResponseInput. The two
-# cannot be confused — neither accepts the other's distinguishing field.
+# BatchInput, one carrying only `attachments` is a ResponseInput. The two cannot be
+# confused — neither accepts the other's distinguishing field.
 AssessmentInput = ResponseInput | BatchInput
 
 
@@ -144,7 +140,6 @@ class PreFilter(BaseModel):
     """Grouped pre-filter verdicts for one item; each is null if not configured."""
 
     topic_relevance: PreFilterVerdict | None = None
-    duplicate_detection: PreFilterVerdict | None = None
 
 
 class AssessmentOutput(BaseModel):

backend/app/models/config/assessment_blob.py

@@ -1,3 +1,5 @@
+import logging
+import re
 from typing import Literal
 
 from pydantic import JsonValue, model_validator
@@ -10,10 +12,15 @@
     TextLLMParams,
 )
 
+logger = logging.getLogger(__name__)
+
 # json_output_schema is validated shallowly at config time: it must be a non-empty
 # object-typed dict. Provider strict-mode normalisation is a run-mode concern.
 JSON_SCHEMA_OBJECT_TYPE = "object"
 
+# {column} placeholders in a submission template; the capture group is the column name.
+PLACEHOLDER_RE = re.compile(r"\{(\w+)\}")
+
 # Default llm for a pre-filter call when the config does not override it.
 DEFAULT_PREFILTER_PROVIDER = Provider.OPENAI
 DEFAULT_PREFILTER_MODEL = "gpt-5.6-luna"
@@ -44,6 +51,13 @@ class PreFilterParams(TextLLMParams):
         min_length=1,
         description="The pre-filter's criteria (system instruction). Mandatory.",
     )
+    submission: str | None = Field(
+        default=None,
+        description=(
+            "Optional per-row prompt template with {column} placeholders for this "
+            "pre-filter. Placeholders must resolve against the blob-level input_schema."
+        ),
+    )
 
 
 class PreFilterBase(SQLModel):
@@ -88,41 +102,21 @@ class TopicRelevanceFilter(PreFilterBase):
     )
 
 
-class DuplicateDetectionFilter(PreFilterBase):
-    """Pre-filter that flags items duplicating prior corpus content.
-
-    Its criteria live in ``params.instructions`` (like the assessment call).
-    """
-
-    knowledge_base_id: str | None = Field(
-        default=None,
-        description="Vector store to compare against; defaults to the platform corpus when unset.",
-    )
-    stop_on_fail: bool = Field(
-        default=False,
-        description=(
-            "If true, a failing verdict stops the chain and skips the assessment for "
-            "that item; if false, the verdict is just recorded and the assessment runs."
-        ),
-    )
-
-
 class AssessmentPreFilters(SQLModel):
     """Optional pre-filters applied before the main assessment call."""
 
     topic_relevance: TopicRelevanceFilter | None = None
-    duplicate_detection: DuplicateDetectionFilter | None = None
 
 
 class AssessmentTextParams(TextLLMParams):
-    """Text params + structured input/output schemas, scoped to assessment."""
+    """Text params + structured output schema, scoped to assessment."""
 
-    input_schema: dict[str, InputColumn] = Field(
+    submission: str = Field(
         ...,
         min_length=1,
         description=(
-            "Per-column spec for BATCH submissions ({type, format}). Mandatory and "
-            "non-empty; every declared column must be present in every submission row."
+            "Per-row prompt template with {column} placeholders. Mandatory; every "
+            "placeholder must resolve against input_schema."
         ),
     )
     json_output_schema: dict[str, JsonValue] | None = Field(
@@ -162,5 +156,49 @@ class AssessmentConfigBlob(SQLModel):
     comes from {{col}} prompt interpolation), and no post_processing.
     """
 
+    input_schema: dict[str, InputColumn] = Field(
+        ...,
+        min_length=1,
+        description=(
+            "Per-column spec for the BATCH `data` rows ({type, format}). Shared by the "
+            "pre-filter and assessment consumers, so it lives once at the blob root. "
+            "Mandatory and non-empty; every declared column must be present in every row."
+        ),
+    )
     pre_filters: AssessmentPreFilters | None = None
     assessment: AssessmentCompletionConfig
+
+    @model_validator(mode="after")
+    def _validate_submission_placeholders(self):
+        """Every {column} in every submission template must be a declared input_schema key.
+
+        Pre-filter and assessment submissions both validate against the blob-level
+        input_schema. Each params dict is already plain (their own validators dumped the
+        typed params), so submissions are read via dict access.
+        """
+        declared = set(self.input_schema)
+
+        blocks: list[tuple[str, str]] = [
+            ("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(
+                (
+                    "pre_filter topic_relevance",
+                    str(topic_relevance.params["submission"]),
+                )
+            )
+
+        for block, template in blocks:
+            unknown = sorted(set(PLACEHOLDER_RE.findall(template)) - declared)
+            if unknown:
+                logger.error(
+                    f"[_validate_submission_placeholders] Unknown placeholder(s) "
+                    f"| block: {block} | unknown: {unknown} | declared: {sorted(declared)}"
+                )
+                raise ValueError(
+                    f"{block} submission references unknown input_schema column(s): "
+                    f"{', '.join(unknown)}"
+                )
+        return self

backend/app/services/assessment/api/batch.py

@@ -59,7 +59,6 @@
 from app.models.config.assessment_blob import (
     AssessmentConfigBlob,
     AssessmentPreFilters,
-    DuplicateDetectionFilter,
     TopicRelevanceFilter,
 )
 from app.models.llm.constants import DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS
@@ -81,7 +80,6 @@ class ApiStage(StrEnum):
     """Pipeline stage identifiers; names match the pre-filter config + result fields."""
 
     TOPIC_RELEVANCE = "topic_relevance"
-    DUPLICATE_DETECTION = "duplicate_detection"
     ASSESSMENT = "assessment"
 
 
@@ -138,14 +136,10 @@ def build_pipeline(
     pre_filters: AssessmentPreFilters | None,
 ) -> list[dict[str, str]]:
     """Ordered stages: GATE pre-filters, then PASS-THROUGH pre-filters, then assessment."""
-    present: list[tuple[ApiStage, TopicRelevanceFilter | DuplicateDetectionFilter]] = []
+    present: list[tuple[ApiStage, TopicRelevanceFilter]] = []
     if pre_filters is not None:
         if pre_filters.topic_relevance is not None:
             present.append((ApiStage.TOPIC_RELEVANCE, pre_filters.topic_relevance))
-        if pre_filters.duplicate_detection is not None:
-            present.append(
-                (ApiStage.DUPLICATE_DETECTION, pre_filters.duplicate_detection)
-            )
 
     gates = [
         {"stage": stage.value, "kind": StageKind.GATE.value}
@@ -213,28 +207,28 @@ def build_rows(
     return rows, text_columns, attachments
 
 
-def _stage_prompt(batch_input: BatchInput, stage: str) -> str | None:
+def _stage_prompt(blob: AssessmentConfigBlob, stage: str) -> str | None:
+    """Per-row prompt template for a stage, sourced from the config's ``submission``.
+
+    The assessment carries a mandatory ``submission``; a pre-filter may carry its own
+    optional one (None when it declares none — its criteria then live entirely in
+    params.instructions with the item's columns + attachments as the user content).
+    """
     if stage == ApiStage.ASSESSMENT:
-        return batch_input.query
-    # Pre-filter criteria live in params.instructions (system prompt); the item's
-    # own columns + attachments are the user content, so there is no per-row
-    # prompt template for a pre-filter stage.
-    return None
+        return blob.assessment.params["submission"]
+    return cast(
+        "str | None", _prefilter_for_stage(blob, stage).params.get("submission")
+    )
 
 
 def _prefilter_for_stage(
     blob: AssessmentConfigBlob, stage: str
-) -> TopicRelevanceFilter | DuplicateDetectionFilter:
-    """The pre-filter config object for a pre-filter stage (topic_relevance / duplicate_detection)."""
+) -> TopicRelevanceFilter:
+    """The pre-filter config object for a pre-filter stage (topic_relevance)."""
     pre = blob.pre_filters
     if pre is not None:
         if stage == ApiStage.TOPIC_RELEVANCE and pre.topic_relevance is not None:
             return pre.topic_relevance
-        if (
-            stage == ApiStage.DUPLICATE_DETECTION
-            and pre.duplicate_detection is not None
-        ):
-            return pre.duplicate_detection
     raise ValueError(
         f"[_prefilter_for_stage] No pre-filter configured for stage {stage}"
     )
@@ -248,28 +242,26 @@ def _stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]:
     if stage == ApiStage.ASSESSMENT:
         params = dict(blob.assessment.params)
         json_schema = params.pop("json_output_schema", None)
-        params.pop(
-            "input_schema", None
-        )  # request-validation only, not a provider param
+        # submission is a prompt concern, not a provider param.
+        params.pop("submission", None)
         if json_schema is not None:
             params["output_schema"] = json_schema  # provider param name
         return params
 
     flt = _prefilter_for_stage(blob, stage)
     params = dict(flt.params)
+    params.pop("submission", None)  # prompt template, not a provider param
     params["output_schema"] = PREFILTER_VERDICT_SCHEMA
     # The config's criteria (mandatory params.instructions) is the system prompt;
     # append the gate directive so the model returns the verdict+reasoning contract.
     criteria = params.get("instructions") or ""
     params["instructions"] = f"{criteria}\n\n{_PREFILTER_INSTRUCTION}"
-    if isinstance(flt, DuplicateDetectionFilter) and flt.knowledge_base_id:
-        params["knowledge_base_ids"] = [flt.knowledge_base_id]
     return params
 
 
 def _stage_provider_model(blob: AssessmentConfigBlob, stage: str) -> tuple[str, str]:
     """Provider + model for a stage: each pre-filter uses its own; assessment uses the config's."""
-    if stage in (ApiStage.TOPIC_RELEVANCE, ApiStage.DUPLICATE_DETECTION):
+    if stage == ApiStage.TOPIC_RELEVANCE:
         flt = _prefilter_for_stage(blob, stage)
         return flt.provider, flt.params["model"]
     return blob.assessment.provider, blob.assessment.params["model"]
@@ -546,7 +538,10 @@ def _submit_stage(
 ) -> bool:
     """Build + submit the current stage's batch on its row subset. Returns success."""
     kind = _stage_kind(bag["pipeline"], stage)
-    input_columns = blob.assessment.params.get("input_schema")
+    input_columns = {
+        name: col.model_dump(exclude_none=True)
+        for name, col in blob.input_schema.items()
+    }
     rows, text_columns, attachments = build_rows(batch_input, input_columns)
     subset = _row_subset(bag, stage, kind, len(rows))
 
@@ -571,7 +566,7 @@ def _submit_stage(
         rows=[rows[i] for i in subset],
         text_columns=text_columns,
         attachments=attachments,
-        prompt=_stage_prompt(batch_input, stage),
+        prompt=_stage_prompt(blob, stage),
         params=_stage_params(blob, stage),
         row_indices=subset,
         organization_id=organization_id,

backend/app/services/assessment/api/results.py

@@ -107,7 +107,6 @@ def build_result(*, session: Session, assessment: Assessment) -> AssessmentBatch
     outputs, load_error = _load_assessment_outputs(session, bag, assessment.project_id)
 
     tr_verdicts = verdicts.get(ApiStage.TOPIC_RELEVANCE.value, {})
-    dd_verdicts = verdicts.get(ApiStage.DUPLICATE_DETECTION.value, {})
 
     items: list[AssessmentResult] = []
     counts = AssessmentCounts()
@@ -125,14 +124,8 @@ def build_result(*, session: Session, assessment: Assessment) -> AssessmentBatch
                 error = load_error
 
         topic_relevance = _verdict_obj(tr_verdicts.get(str(idx)))
-        duplicate_detection = _verdict_obj(dd_verdicts.get(str(idx)))
         pre_filter = (
-            PreFilter(
-                topic_relevance=topic_relevance,
-                duplicate_detection=duplicate_detection,
-            )
-            if (topic_relevance or duplicate_detection)
-            else None
+            PreFilter(topic_relevance=topic_relevance) if topic_relevance else None
         )
         items.append(
             AssessmentResult(

backend/app/services/assessment/api/submission.py

@@ -153,7 +153,10 @@ def submit(
 
     # input_schema is mandatory (enforced on the config), so every row must match it:
     # all declared columns present, no undeclared columns, attachments url-valued.
-    input_columns = blob.assessment.params.get("input_schema") or {}
+    input_columns = {
+        name: col.model_dump(exclude_none=True)
+        for name, col in blob.input_schema.items()
+    }
     _validate_rows_against_schema(batch_input.data, input_columns)
 
     # Validate transposition up front so bad attachment shapes fail as 422, not async.