feat(assessment): Implement single response call

backend/app/api/docs/assessment/create.md

@@ -5,17 +5,17 @@ pre-filters (topic relevance / duplicate detection). The run mode is **inferred
 shape** — you do not pass a mode flag.
 
 **Key Features:**
-* Method inferred from the input: a `data` list ⇒ **BATCH**, a single `query` (+ optional
-  `attachments`) ⇒ **RESPONSE**.
+* Method inferred from the input: a `data` list (or `submission_doc_id`) ⇒ **BATCH**, a single
+  row object in `data` ⇒ **RESPONSE**.
 * Pins to a saved config **version** (`config.id` + `config.version`); the config must be tagged
   `ASSESSMENT` (see the config-create docs for the `config_blob` assessment shape).
 * Optional pre-filters run before the grading call and can gate it per item.
 * **Two ways to get the result** — poll `GET /assessments/{assessment_id}`, and/or supply a
   `callback_url` to have the finished result POSTed to you on completion.
 * `request_metadata` is echoed back unchanged in the callback for correlation.
 
-> **RESPONSE mode is not wired yet** — a single-object input currently returns `501 Not Implemented`.
-> Only BATCH (a `data` list) is processed today.
+> **RESPONSE** answers one row and delivers the result only by webhook, so `callback_url` is
+> **required** for it (a RESPONSE without one fails with `422`). BATCH can still be polled.
 
 ---
 
@@ -25,7 +25,6 @@ shape** — you do not pass a mode flag.
 {
   "config": { "id": "3f9b2c10-0000-4a11-8c22-1a2b3c4d5e6f", "version": 1 },
   "input": {
-    "query": "Grade the answer sheet for roll number {roll_number}, student {student_name}.",
     "data": [
       {
         "roll_number": "1",
@@ -50,8 +49,7 @@ shape** — you do not pass a mode flag.
   * `id` (UUID, required) — the config id.
   * `version` (int ≥ 1, required) — the config version. Must reference a config tagged `ASSESSMENT`.
 * `input` (required) — one of two shapes; the shape selects the method:
-  * **BATCH** — `{ "query": "<template>", "data": [ {<column>: <value>}, ... ] }`
-    * `query` (required, non-empty) — a template with `{column}` placeholders substituted per row.
+  * **BATCH** — `{ "data": [ {<column>: <value>}, ... ] }`
     * `data` (required, ≥ 1 row) — submission rows. Each row is a flat `column -> string` map. The
       config's `input_schema` is **mandatory** and defines every column: its `type` (`text` / `image` /
       `pdf`, with an attachment `format`) and `strict` (default `false`). Every row is validated
@@ -60,12 +58,15 @@ shape** — you do not pass a mode flag.
       `submission_doc_id` sheet's undeclared columns are ignored, so the schema selects which
       columns take part), and a non-blank `image`/`pdf` value must be a URL. A row that does not
       match fails with `422` (see Errors).
-  * **RESPONSE** — `{ "query": "<text>", "attachments": [ ... ] }` *(deferred — returns 501)*.
-* `callback_url` (optional) — the webhook the result is POSTed to on completion. Omit it to poll instead.
+  * **RESPONSE** — `{ "data": { <column>: <value>, ... } }` — exactly one row, validated against the
+    config's `input_schema` like a BATCH row. Attachment columns (`image` / `pdf` / `video` / `audio`)
+    carry a URL (`http(s)://` or `gs://`); base64 is not supported.
+* `callback_url` — the webhook the result is POSTed to on completion. Optional for BATCH (omit it to
+  poll instead); **required** for RESPONSE.
 * `request_metadata` (optional) — arbitrary object passed through unchanged in the callback.
 
-The two input shapes are strictly discriminated: a body carrying `data` is BATCH, one carrying
-`attachments` (or query-only) is RESPONSE. Neither accepts the other's distinguishing field.
+The two input shapes are strictly discriminated: `data` as a list (or a `submission_doc_id`) is BATCH,
+`data` as a single object is RESPONSE. Neither accepts the other's distinguishing field.
 
 ---
 
@@ -135,15 +136,20 @@ On completion the platform POSTs this payload to `callback_url`:
       * `output.assessment` — a dict when the config emits structured output (`json_output_schema`),
         a raw string for free-text output, or `null` for a gated/failed row.
       * `output.pre_filter` — per-item `topic_relevance` / `duplicate_detection` verdicts, each
-        `{ "verdict": bool, "reasoning": str }` or `null` if that pre-filter was not configured.
+        `{ "verdict": bool, "reasoning": str, "error": str | null }` or `null` if that pre-filter was not
+        configured. A failed pre-filter call fails open (`verdict: true`) and the assessment still
+        runs; RESPONSE reports that failure in `error`, BATCH leaves it `null`.
       * `error` — a per-row error string, or `null`.
-  * **RESPONSE** — a single item object with the same `output` / `error` shape.
+  * **RESPONSE** — a single item object with the same `output` / `error` shape. `data.status` is
+    `COMPLETED`, `COMPLETED_WITH_ERRORS` (a pre-filter call failed but the assessment ran) or `FAILED`. If the run cannot
+    produce a result at all (for example the config version was deleted or a provider credential is
+    missing), `status` is `FAILED`, `data` is `null`, and the envelope's `error` carries the message.
 
 ---
 
 ## Errors
 
-* `501 Not Implemented` — RESPONSE-mode input (single object) is not wired yet; send a BATCH `data` list.
+* `422 Unprocessable Entity` for a RESPONSE without a `callback_url`.
 * `422 Unprocessable Entity` — the body failed validation (e.g. `config.id`/`config.version` missing,
   an input shape carrying both `data` and `submission_doc_id`, or neither), or a
   submission row does not match the config's `input_schema` (a strict column absent or blank, an

backend/app/api/routes/assessment/api.py

@@ -1,11 +1,12 @@
 """Assessment API-client route.
 
 Run mode is inferred from the input shape:
-    - RESPONSE: single object input (`input` key) — deferred, returns 501.
-    - BATCH: submission-list input (`data` key) — N items over one provider batch series.
+    - RESPONSE: `data` is a row object — the assessment response is delivered by `callback_url`.
+    - BATCH: `data` is a list of rows, or a `submission_doc_id` — N items over one provider
+      batch series.
 
-Results are delivered to the request's `callback_url` when one is given, and are always
-readable from `GET /assessments/{assessment_id}`.
+A BATCH result is delivered to the request's `callback_url` when one is given, and is
+always readable from `GET /assessments/{assessment_id}`.
 """
 
 import logging
@@ -24,7 +25,6 @@
     AssessmentMethod,
     AssessmentSubmitResponse,
     AssessmentSummary,
-    BatchInput,
 )
 from app.services.assessment.api import results, submission
 from app.utils import APIResponse, load_description
@@ -36,8 +36,6 @@
     tags=["Assessment (API)"],
 )
 
-_RESPONSE_NOT_WIRED = "RESPONSE-mode assessment is not wired yet."
-
 
 @router.post(
     "",
@@ -52,9 +50,6 @@ def create_assessment(
 ) -> APIResponse[AssessmentSubmitResponse]:
     """Submit an assessment; the method is inferred from the input shape. The result is
     pushed to `callback_url` when one is given, and is polled from GET otherwise."""
-    if not isinstance(request.input, BatchInput):
-        raise HTTPException(status_code=501, detail=_RESPONSE_NOT_WIRED)
-
     result = submission.submit(
         session=session,
         request=request,

backend/app/celery/tasks/job_execution.py

@@ -421,6 +421,19 @@ def run_assessment_api_batch(
     return result
 
 
+@celery_app.task(bind=True, queue="default", priority=2)
+@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_assessment_response")
+def run_assessment_response(self, execution_id: int, trace_id: str, **kwargs):
+    """Run a RESPONSE assessment end to end and deliver its result by callback."""
+    from app.services.assessment.api.response import run_response_assessment
+
+    _set_trace(trace_id)
+    return _run_with_otel_parent(
+        self,
+        lambda: run_response_assessment(execution_id=execution_id),
+    )
+
+
 @celery_app.task(bind=True, queue="default", priority=2)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_tts_result_processing")
 def run_tts_result_processing(

backend/app/crud/assessment/api.py

@@ -16,6 +16,7 @@
 from sqlmodel import Session, col, select
 
 from app.core.util import now
+from app.crud.config.version import ConfigVersionCrud
 from app.models.assessment import (
     Assessment,
     AssessmentMethod,
@@ -24,6 +25,8 @@
     AssessmentSubmission,
     BatchRunState,
 )
+from app.models.config.assessment_blob import AssessmentConfigBlob
+from app.models.config.config import ConfigTag
 
 logger = logging.getLogger(__name__)
 
@@ -235,6 +238,20 @@ def list_assessments_with_execution(
     return list(session.exec(statement).all())
 
 
+def get_config_blob(
+    *, session: Session, execution: AssessmentRun, project_id: int
+) -> AssessmentConfigBlob:
+    """The ASSESSMENT config version the run was submitted with, parsed."""
+    version_crud = ConfigVersionCrud(
+        session=session,
+        config_id=execution.config_id,
+        project_id=project_id,
+        tag=ConfigTag.ASSESSMENT,
+    )
+    version = version_crud.exists_or_raise(version_number=execution.config_version)
+    return AssessmentConfigBlob.model_validate(version.config_blob)
+
+
 def list_executions(*, session: Session, assessment_id: UUID) -> list[AssessmentRun]:
     statement = (
         select(AssessmentRun)

backend/app/models/assessment/__init__.py

@@ -42,7 +42,6 @@
     AssessmentResultRow,
     AssessmentSubmitResponse,
     AssessmentSummary,
-    Attachment,
     BatchInput,
     BatchRunState,
     ParsedResult,
@@ -88,7 +87,6 @@
     "AssessmentSubmissionPreview",
     "AssessmentSubmissionResponse",
     # API-client surface
-    "Attachment",
     "ResponseInput",
     "BatchInput",
     "BatchRunState",

backend/app/models/assessment/assessment_api.py

@@ -8,26 +8,23 @@
 from typing import Annotated, Any, NotRequired, TypedDict
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field, HttpUrl, JsonValue, model_validator
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Discriminator,
+    Field,
+    HttpUrl,
+    JsonValue,
+    Tag,
+    model_validator,
+)
 from sqlmodel import SQLModel
 
 from app.models.assessment.assessment import (
     AssessmentConfigRef,
     AssessmentMethod,
     AssessmentStatus,
 )
-from app.models.llm.request import ImageInput, PDFInput
-
-Attachment = Annotated[ImageInput | PDFInput, Field(discriminator="type")]
-
-
-class ResponseInput(SQLModel):
-    """RESPONSE input — optional attachments (no columns). The prompt lives in the config."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    attachments: list[Attachment] = Field(default_factory=list)
-
 
 # One BATCH submission row: column -> string value (text, or a url/base64 value for
 # attachment columns). Column type/strict/format live in the config's input_schema.
@@ -62,6 +59,17 @@ def _exactly_one_row_source(self) -> "BatchInput":
         return self
 
 
+class ResponseInput(SQLModel):
+    """RESPONSE input — one submission row; the prompt lives in the config."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    data: Submission
+
+    def as_batch(self) -> BatchInput:
+        return BatchInput(data=[self.data])
+
+
 class Verdict(TypedDict):
     """A pre-filter's parsed judgement for one row."""
 
@@ -104,25 +112,28 @@ class BatchRunState(TypedDict):
     callback_url: str | None  # None when the client polls instead of receiving a push
     request_metadata: dict[str, Any] | None
     error: NotRequired[str]  # only set on failure (_fail)
+    # RESPONSE only: the AssessmentResult as delivered
+    result: NotRequired[dict[str, Any]]
+
 
+def _input_kind(value: Any) -> str:
+    data = (
+        value.get("data") if isinstance(value, dict) else getattr(value, "data", None)
+    )
+    return "response" if isinstance(data, dict) else "batch"
 
-# Strict, tagless discrimination via extra=forbid: an input carrying `data` is a
-# BatchInput, one carrying only `attachments` is a ResponseInput. The two cannot be
-# confused — neither accepts the other's distinguishing field.
-AssessmentInput = ResponseInput | BatchInput
 
+# A row object is an assessment response; a list of rows or a submission reference is a batch.
+AssessmentInput = Annotated[
+    Annotated[ResponseInput, Tag("response")] | Annotated[BatchInput, Tag("batch")],
+    Discriminator(_input_kind),
+]
 
-def derive_method(
-    input_: AssessmentInput | None, submission_id: UUID | None
-) -> AssessmentMethod:
-    """Infer method: ResponseInput ⇒ RESPONSE, BatchInput ⇒ BATCH, else submission_id ⇒ RUN."""
+
+def derive_method(input_: AssessmentInput) -> AssessmentMethod:
     if isinstance(input_, ResponseInput):
         return AssessmentMethod.RESPONSE
-    if isinstance(input_, BatchInput):
-        return AssessmentMethod.BATCH
-    if submission_id is not None:
-        return AssessmentMethod.RUN
-    raise ValueError("[derive_method] Provide inline `input` or `submission_id`")
+    return AssessmentMethod.BATCH
 
 
 class AssessmentCreate(BaseModel):
@@ -145,6 +156,12 @@ class AssessmentCreate(BaseModel):
         description="Passed through unchanged in the callback for correlation",
     )
 
+    @model_validator(mode="after")
+    def _response_needs_callback(self) -> "AssessmentCreate":
+        if isinstance(self.input, ResponseInput) and self.callback_url is None:
+            raise ValueError("An assessment response needs a callback_url.")
+        return self
+
 
 class AssessmentSubmitResponse(BaseModel):
     """Flat API-client submit ack (one config, so one execution) — mirrors the llm-call ack."""
@@ -157,10 +174,11 @@ class AssessmentSubmitResponse(BaseModel):
 
 
 class PreFilterVerdict(BaseModel):
-    """Structured pre-filter result per item."""
+    """Structured pre-filter result per item; ``error`` is set when the stage's call failed."""
 
     verdict: bool
     reasoning: str = ""
+    error: str | None = None
 
 
 class PreFilter(BaseModel):

backend/app/models/config/assessment_blob.py

@@ -1,6 +1,6 @@
 import logging
 import re
-from typing import Literal, get_args
+from typing import Any, Literal, get_args
 
 from pydantic import JsonValue, model_validator
 from sqlmodel import Field, SQLModel
@@ -192,6 +192,12 @@ class AssessmentConfigBlob(SQLModel):
     pre_filters: AssessmentPreFilters | None = None
     assessment: AssessmentCompletionConfig
 
+    def input_columns(self) -> dict[str, dict[str, Any]]:
+        return {
+            name: column.model_dump(exclude_none=True)
+            for name, column in self.input_schema.items()
+        }
+
     @model_validator(mode="after")
     def _validate_submission_placeholders(self):
         """Every {column} in every submission template must be a declared input_schema key.