diff --git a/backend/app/api/docs/assessment/create.md b/backend/app/api/docs/assessment/create.md
index d8c140205..c5fe02421 100644
--- a/backend/app/api/docs/assessment/create.md
+++ b/backend/app/api/docs/assessment/create.md
@@ -5,8 +5,8 @@ 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.
@@ -14,8 +14,8 @@ shape** — you do not pass a mode flag.
   `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
diff --git a/backend/app/api/routes/assessment/api.py b/backend/app/api/routes/assessment/api.py
index c8246a4e8..403250dbb 100644
--- a/backend/app/api/routes/assessment/api.py
+++ b/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,
diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py
index b7c7de8f8..ca7f6792b 100644
--- a/backend/app/celery/tasks/job_execution.py
+++ b/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(
diff --git a/backend/app/crud/assessment/api.py b/backend/app/crud/assessment/api.py
index f6f7cd360..31aa16047 100644
--- a/backend/app/crud/assessment/api.py
+++ b/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)
diff --git a/backend/app/models/assessment/__init__.py b/backend/app/models/assessment/__init__.py
index 0e52cdc37..fb4c335d8 100644
--- a/backend/app/models/assessment/__init__.py
+++ b/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",
diff --git a/backend/app/models/assessment/assessment_api.py b/backend/app/models/assessment/assessment_api.py
index 38229d4f9..e744ab948 100644
--- a/backend/app/models/assessment/assessment_api.py
+++ b/backend/app/models/assessment/assessment_api.py
@@ -8,7 +8,16 @@
 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 (
@@ -16,18 +25,6 @@
     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):
diff --git a/backend/app/models/config/assessment_blob.py b/backend/app/models/config/assessment_blob.py
index a5d82468f..45cb88268 100644
--- a/backend/app/models/config/assessment_blob.py
+++ b/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.
diff --git a/backend/app/services/assessment/api/batch.py b/backend/app/services/assessment/api/batch.py
index 4bb4ce5eb..df9d407d3 100644
--- a/backend/app/services/assessment/api/batch.py
+++ b/backend/app/services/assessment/api/batch.py
@@ -105,8 +105,8 @@ def _google_gcp_credential(
 POLL_COUNTDOWN_SECONDS = settings.CRON_INTERVAL_MINUTES * 60
 
 
-class ApiStage(StrEnum):
-    """Pipeline stage identifiers; names match the pre-filter config + result fields."""
+class PipelineStage(StrEnum):
+    """Stage names of the API-client pipeline; they key the config, the run state and the result."""
 
     TOPIC_RELEVANCE = "topic_relevance"
     ASSESSMENT = "assessment"
@@ -167,25 +167,27 @@ 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]] = []
+    pre_filter_stages: list[tuple[PipelineStage, TopicRelevanceFilter]] = []
     if pre_filters is not None:
         if pre_filters.topic_relevance is not None:
-            present.append((ApiStage.TOPIC_RELEVANCE, pre_filters.topic_relevance))
+            pre_filter_stages.append(
+                (PipelineStage.TOPIC_RELEVANCE, pre_filters.topic_relevance)
+            )
 
     gates = [
         {"stage": stage.value, "kind": StageKind.GATE.value}
-        for stage, flt in present
-        if flt.stop_on_fail
+        for stage, pre_filter in pre_filter_stages
+        if pre_filter.stop_on_fail
     ]
     passthrough = [
         {"stage": stage.value, "kind": StageKind.PASS_THROUGH.value}
-        for stage, flt in present
-        if not flt.stop_on_fail
+        for stage, pre_filter in pre_filter_stages
+        if not pre_filter.stop_on_fail
     ]
     return [
         *gates,
         *passthrough,
-        {"stage": ApiStage.ASSESSMENT.value, "kind": StageKind.ASSESSMENT.value},
+        {"stage": PipelineStage.ASSESSMENT.value, "kind": StageKind.ASSESSMENT.value},
     ]
 
 
@@ -247,14 +249,14 @@ def build_rows(
     return rows, text_columns, attachments
 
 
-def _stage_prompt(blob: AssessmentConfigBlob, 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:
+    if stage == PipelineStage.ASSESSMENT:
         return blob.assessment.params["submission"]
     return cast(
         "str | None", _prefilter_for_stage(blob, stage).params.get("submission")
@@ -267,19 +269,19 @@ def _prefilter_for_stage(
     """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:
+        if stage == PipelineStage.TOPIC_RELEVANCE and pre.topic_relevance is not None:
             return pre.topic_relevance
     raise ValueError(
         f"[_prefilter_for_stage] No pre-filter configured for stage {stage}"
     )
 
 
-def _stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]:
+def stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]:
     """Kaapi params for a stage: the assessment uses its own params + output schema; each
     pre-filter uses ITS own params (criteria in params.instructions), with the verdict
     schema + gate directive layered on.
     """
-    if stage == ApiStage.ASSESSMENT:
+    if stage == PipelineStage.ASSESSMENT:
         params = dict(blob.assessment.params)
         json_schema = params.pop("json_output_schema", None)
         # submission is a prompt concern, not a provider param.
@@ -290,8 +292,8 @@ def _stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]:
         params["instructions"] = normalize_llm_text(params.get("instructions") or "")
         return params
 
-    flt = _prefilter_for_stage(blob, stage)
-    params = dict(flt.params)
+    pre_filter = _prefilter_for_stage(blob, stage)
+    params = dict(pre_filter.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;
@@ -301,14 +303,58 @@ def _stage_params(blob: AssessmentConfigBlob, stage: str) -> dict[str, Any]:
     return params
 
 
-def _stage_provider_model(blob: AssessmentConfigBlob, stage: str) -> tuple[str, str]:
+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 == ApiStage.TOPIC_RELEVANCE:
-        flt = _prefilter_for_stage(blob, stage)
-        return flt.provider, flt.params["model"]
+    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"]
 
 
+def build_stage_requests(
+    *,
+    session: Session,
+    provider_name: str,
+    rows: list[dict[str, str]],
+    text_columns: list[str],
+    attachments: list[AssessmentAttachment],
+    prompt: str | None,
+    params: dict[str, Any],
+    organization_id: int,
+    project_id: int,
+    row_indices: list[int] | None = None,
+) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+    """Provider-wire request per row plus the mapped provider params.
+
+    Shared by batch submission and RESPONSE, so both send identical requests.
+    """
+    # Resolve gs:// attachments to provider-reachable URLs before building the requests.
+    rows = rewrite_gcs_attachment_urls(
+        session=session,
+        rows=rows,
+        attachments=attachments,
+        llm_provider=provider_name,
+        project_id=project_id,
+        organization_id=organization_id,
+    )
+    if provider_name == LLMProvider.OPENAI:
+        mapped, _ = map_kaapi_to_openai_params(session=session, kaapi_params=params)
+        builder = build_openai_jsonl
+    elif provider_name in (
+        LLMProvider.GOOGLE,
+        LLMProvider.GOOGLE_AISTUDIO,
+        LLMProvider.GOOGLE_GCP,
+    ):
+        mapped, _ = map_kaapi_to_google_params(params)
+        builder = build_google_jsonl
+    elif provider_name == LLMProvider.ANTHROPIC:
+        mapped, _ = map_kaapi_to_anthropic_params(params)
+        builder = build_anthropic_jsonl
+    else:
+        raise ValueError(f"[build_stage_requests] Unsupported provider {provider_name}")
+    return builder(rows, text_columns, attachments, prompt, mapped, row_indices), mapped
+
+
 def _submit_provider_batch(
     *,
     session: Session,
@@ -325,21 +371,20 @@ def _submit_provider_batch(
     description: str,
 ) -> BatchJob:
     """Build provider JSONL and submit it via the shared batch infra."""
-    # Resolve gs:// attachments to provider-reachable URLs before building JSONL.
-    rows = rewrite_gcs_attachment_urls(
+    jsonl, mapped = build_stage_requests(
         session=session,
+        provider_name=provider_name,
         rows=rows,
+        text_columns=text_columns,
         attachments=attachments,
-        llm_provider=provider_name,
-        project_id=project_id,
+        prompt=prompt,
+        params=params,
+        row_indices=row_indices,
         organization_id=organization_id,
+        project_id=project_id,
     )
 
     if provider_name == LLMProvider.OPENAI:
-        mapped, _ = map_kaapi_to_openai_params(session=session, kaapi_params=params)
-        jsonl = build_openai_jsonl(
-            rows, text_columns, attachments, prompt, mapped, row_indices
-        )
         provider: BatchProvider = OpenAIBatchProvider(
             client=get_openai_client(
                 session=session, org_id=organization_id, project_id=project_id
@@ -355,10 +400,6 @@ def _submit_provider_batch(
         LLMProvider.GOOGLE_AISTUDIO,
         LLMProvider.GOOGLE_GCP,
     ):
-        mapped, _ = map_kaapi_to_google_params(params)
-        jsonl = build_google_jsonl(
-            rows, text_columns, attachments, prompt, mapped, row_indices
-        )
         if provider_name == LLMProvider.GOOGLE_GCP:
             cred = _google_gcp_credential(
                 session=session,
@@ -377,10 +418,6 @@ def _submit_provider_batch(
             config = {"display_name": description, "model": f"models/{model}"}
 
     elif provider_name == LLMProvider.ANTHROPIC:
-        mapped, _ = map_kaapi_to_anthropic_params(params)
-        jsonl = build_anthropic_jsonl(
-            rows, text_columns, attachments, prompt, mapped, row_indices
-        )
         provider = AnthropicBatchProvider(
             client=get_anthropic_client(
                 session=session, org_id=organization_id, project_id=project_id
@@ -481,7 +518,7 @@ def _openai_output_text(output: Any) -> str:
     return "".join(chunks)
 
 
-def _parse_one(result: dict[str, Any], provider_name: str) -> ParsedResult:
+def parse_one(result: dict[str, Any], provider_name: str) -> ParsedResult:
     error = result.get("error")
     if error:
         return {
@@ -555,11 +592,11 @@ def parse_batch_results(
         idx = _row_index(result.get(BATCH_KEY) or result.get("key"))
         if idx is None:
             continue
-        parsed[idx] = _parse_one(result, provider_name)
+        parsed[idx] = parse_one(result, provider_name)
     return parsed
 
 
-def _parse_verdict(output: str | None) -> Verdict:
+def parse_verdict(output: str | None) -> Verdict:
     """Read a pre-filter verdict. Unparseable output fails open (verdict=True)."""
     if not output:
         return {"verdict": True, "reasoning": ""}
@@ -571,7 +608,7 @@ def _parse_verdict(output: str | None) -> Verdict:
         }
     except (json.JSONDecodeError, TypeError, AttributeError):
         logger.warning(
-            "[_parse_verdict] Unparseable verdict, failing open | output_len=%s",
+            "[parse_verdict] Unparseable verdict, failing open | output_len=%s",
             len(output),
         )
         return {"verdict": True, "reasoning": ""}
@@ -587,7 +624,7 @@ def _record_stage(
     verdicts: dict[str, Verdict] = {}
     passed_count = 0
     for idx, out in parsed.items():
-        verdict = _parse_verdict(out.get("output"))
+        verdict = parse_verdict(out.get("output"))
         verdicts[str(idx)] = verdict
         if verdict["verdict"]:
             passed_count += 1
@@ -645,10 +682,7 @@ def _submit_stage(
     from app.services.assessment.api.submission_store import open_submission_rows
 
     kind = _stage_kind(bag["pipeline"], stage)
-    input_columns = {
-        name: col.model_dump(exclude_none=True)
-        for name, col in blob.input_schema.items()
-    }
+    input_columns = blob.input_columns()
     text_columns, attachments = column_kinds(list(input_columns), input_columns)
     subset = _row_subset(bag, stage, kind, execution.total_items)
 
@@ -669,7 +703,7 @@ def _submit_stage(
     with open_submission_rows(session=session, assessment=assessment) as stream:
         rows = [row for idx, row in enumerate(stream) if idx in wanted]
 
-    provider_name, model = _stage_provider_model(blob, stage)
+    provider_name, model = stage_provider_model(blob, stage)
     batch_job = _submit_provider_batch(
         session=session,
         provider_name=provider_name,
@@ -677,8 +711,8 @@ def _submit_stage(
         rows=rows,
         text_columns=text_columns,
         attachments=attachments,
-        prompt=_stage_prompt(blob, stage),
-        params=_stage_params(blob, stage),
+        prompt=stage_prompt(blob, stage),
+        params=stage_params(blob, stage),
         row_indices=subset,
         organization_id=organization_id,
         project_id=project_id,
@@ -776,7 +810,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"] = ApiStage.ASSESSMENT.value
+    bag["stage"] = PipelineStage.ASSESSMENT.value
     bag["stage_status"] = StageStatus.COMPLETED.value
     api.save_execution_state(session=session, execution=execution, state=bag)
 
@@ -932,7 +966,7 @@ def run_batch_stage(
     Idempotent: keyed off ``stage_status`` in the bag — a redelivery either re-polls
     the in-flight batch or re-submits a stage that was never dispatched.
     """
-    # Lazy (like callbacks/results below): result_files imports ApiStage from this module.
+    # Lazy (like callbacks/results below): result_files imports PipelineStage from this module.
     from app.services.assessment.api.result_files import record_stage_dump
     from app.services.assessment.api.submission_store import SubmissionUnavailableError
 
@@ -983,8 +1017,8 @@ def run_batch_stage(
         # invalid/old-shape blob). Route these to _fail so the client gets a terminal
         # callback instead of the run stranding in PROCESSING.
         try:
-            blob = AssessmentConfigBlob.model_validate(
-                _resolve_blob(session, execution, project_id)
+            blob = api.get_config_blob(
+                session=session, execution=execution, project_id=project_id
             )
         except Exception as exc:
             _fail(session, execution, assessment, bag, str(exc))
@@ -1110,20 +1144,3 @@ def run_batch_stage(
             organization_id=organization_id,
             project_id=project_id,
         )
-
-
-def _resolve_blob(
-    session: Session, execution: AssessmentRun, project_id: int
-) -> dict[str, Any]:
-    """Fetch the execution's stored ASSESSMENT config_blob dict for re-parsing."""
-    from app.crud.config.version import ConfigVersionCrud
-    from app.models.config.config import ConfigTag
-
-    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 version.config_blob
diff --git a/backend/app/services/assessment/api/callbacks.py b/backend/app/services/assessment/api/callbacks.py
index 137a34837..e359aaa8f 100644
--- a/backend/app/services/assessment/api/callbacks.py
+++ b/backend/app/services/assessment/api/callbacks.py
@@ -1,8 +1,8 @@
-"""Webhook delivery for the BATCH API-client path.
+"""Webhook delivery for the API-client path.
 
 POSTs the ``AssessmentCallback`` envelope to the client's callback_url on completion via
 the shared SSRF-guarded ``send_callback`` (HMAC-signed with the project webhook secret) —
-the same transport the response path uses. One inline attempt, no retry.
+the same transport the response path uses. The sender retries up to 4 times with backoff.
 """
 
 import logging
@@ -12,8 +12,8 @@
 
 from app.models.assessment import (
     Assessment,
-    AssessmentBatchResult,
     AssessmentCallback,
+    AssessmentResultData,
     AssessmentStatus,
 )
 from app.services.assessment.api.result_files import presign_result_files
@@ -26,7 +26,7 @@ def deliver(
     *,
     session: Session,
     assessment: Assessment,
-    result: AssessmentBatchResult,
+    result: AssessmentResultData | None,
     callback_url: str,
     request_metadata: dict[str, JsonValue] | None,
     failure_message: str | None,
diff --git a/backend/app/services/assessment/api/response.py b/backend/app/services/assessment/api/response.py
new file mode 100644
index 000000000..1751e7e0f
--- /dev/null
+++ b/backend/app/services/assessment/api/response.py
@@ -0,0 +1,279 @@
+"""RESPONSE assessment: the pre-filter and assessment as direct provider calls on one row.
+
+Requests, clients and parsing are the batch / llm-call ones; the run row records the stage
+state and the result goes out by callback.
+"""
+
+import logging
+from functools import partial
+from http import HTTPStatus
+from typing import Any, cast
+
+from gevent import Timeout
+from sqlmodel import Session, col, select
+
+from app.core.db import engine
+from app.crud.assessment import api
+from app.models.assessment import (
+    Assessment,
+    AssessmentOutput,
+    AssessmentResult,
+    AssessmentRun,
+    AssessmentStatus,
+    BatchRunState,
+    ParsedResult,
+    PreFilter,
+    PreFilterVerdict,
+    StageStatus,
+)
+from app.models.config.assessment_blob import AssessmentConfigBlob
+from app.models.llm.constants import DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS
+from app.services.assessment.api.batch import (
+    PipelineStage,
+    StageKind,
+    build_stage_requests,
+    column_kinds,
+    parse_one,
+    parse_verdict,
+    stage_params,
+    stage_prompt,
+    stage_provider_model,
+)
+from app.services.assessment.api.callbacks import deliver
+from app.services.assessment.utils.parsing import parse_assessment_text
+from app.services.llm.providers.registry import LLMProvider, get_llm_provider
+
+logger = logging.getLogger(__name__)
+
+ROW = "0"
+
+
+def _call_openai(client: Any, request: dict[str, Any]) -> dict[str, Any]:
+    response = client.responses.create(**request["body"])
+    body = response.model_dump(mode="json")
+    return {"response": {"status_code": HTTPStatus.OK, "body": body}}
+
+
+def _call_anthropic(client: Any, request: dict[str, Any]) -> dict[str, Any]:
+    # The batch API injects max_tokens at batch level; a single call must carry it.
+    params = {"max_tokens": DEFAULT_ASSESSMENT_BATCH_MAX_TOKENS, **request["params"]}
+    return {"response": client.messages.create(**params).model_dump(mode="json")}
+
+
+def _call_google(client: Any, model: str, request: dict[str, Any]) -> dict[str, Any]:
+    body = request["request"]
+    config = dict(body.get("generationConfig", {}))
+    if "systemInstruction" in body:
+        config["systemInstruction"] = body["systemInstruction"]
+    response = client.models.generate_content(
+        model=model, contents=body["contents"], config=config
+    )
+    return {"response": response.model_dump(mode="json", by_alias=True)}
+
+
+def _call_provider(
+    provider_name: str, client: Any, model: str, request: dict[str, Any]
+) -> dict[str, Any]:
+    """Send one batch-shaped request; the raw response comes back in batch-result shape."""
+    if provider_name == LLMProvider.OPENAI:
+        return _call_openai(client, request)
+    if provider_name == LLMProvider.ANTHROPIC:
+        return _call_anthropic(client, request)
+    if provider_name == LLMProvider.GOOGLE_GCP:
+        return _call_google(client.genai, model, request)
+    return _call_google(client, model, request)
+
+
+def _run_stage(
+    *,
+    session: Session,
+    blob: AssessmentConfigBlob,
+    stage: str,
+    row: dict[str, str],
+    organization_id: int,
+    project_id: int,
+) -> ParsedResult:
+    """One provider call for the row; a provider error comes back as the row's error."""
+    provider_name, model = stage_provider_model(blob, stage)
+    input_columns = blob.input_columns()
+    text_columns, attachments = column_kinds(list(input_columns), input_columns)
+    requests, _ = build_stage_requests(
+        session=session,
+        provider_name=provider_name,
+        rows=[row],
+        text_columns=text_columns,
+        attachments=attachments,
+        prompt=stage_prompt(blob, stage),
+        params=stage_params(blob, stage),
+        organization_id=organization_id,
+        project_id=project_id,
+    )
+    client = get_llm_provider(
+        session=session,
+        provider_type=provider_name,
+        project_id=project_id,
+        organization_id=organization_id,
+    ).client
+    try:
+        raw = _call_provider(provider_name, client, model, requests[0])
+    except Exception as exc:
+        raw = {"error": {"message": str(exc)}}
+    return parse_one(raw, provider_name)
+
+
+def _begin_stage(
+    session: Session, execution: AssessmentRun, state: BatchRunState, stage: str
+) -> None:
+    state["stage"] = stage
+    state["stage_status"] = StageStatus.PROCESSING.value
+    api.save_execution_state(session=session, execution=execution, state=state)
+
+
+def _assess(
+    *,
+    session: Session,
+    assessment: Assessment,
+    execution: AssessmentRun,
+    state: BatchRunState,
+    blob: AssessmentConfigBlob,
+) -> AssessmentResult:
+    run_stage = partial(
+        _run_stage,
+        session=session,
+        blob=blob,
+        row=(assessment.input or {})["data"],
+        organization_id=assessment.organization_id,
+        project_id=assessment.project_id,
+    )
+
+    verdicts: dict[str, PreFilterVerdict] = {}
+    pre_filters = [
+        step for step in state["pipeline"] if step["kind"] != StageKind.ASSESSMENT
+    ]
+    for step in pre_filters:
+        stage = step["stage"]
+        _begin_stage(session, execution, state, stage)
+        out = run_stage(stage=stage)
+        # A failed call fails open, as in BATCH; the error is still reported on the stage.
+        verdict = parse_verdict(out["output"])
+        verdicts[stage] = PreFilterVerdict(**verdict, error=out["error"])
+        state.setdefault("verdicts", {})[stage] = {ROW: verdict}
+        state.setdefault("counters", {})[stage] = {
+            "total": 1,
+            "passed": int(verdict["verdict"]),
+            "rejected": int(not verdict["verdict"]),
+        }
+        if out["error"]:
+            state.setdefault("stage_errors", {})[stage] = {ROW: out["error"]}
+        elif step["kind"] == StageKind.GATE and not verdict["verdict"]:
+            state["gate_passed"][0] = False
+        state["stage_status"] = StageStatus.COMPLETED.value
+
+    pre_filter = PreFilter(**verdicts) if verdicts else None
+    if not state["gate_passed"][0]:
+        return AssessmentResult(output=AssessmentOutput(pre_filter=pre_filter))
+
+    _begin_stage(session, execution, state, PipelineStage.ASSESSMENT.value)
+    out = run_stage(stage=PipelineStage.ASSESSMENT.value)
+    if out["error"]:
+        state.setdefault("stage_errors", {})[PipelineStage.ASSESSMENT.value] = {
+            ROW: out["error"]
+        }
+    state["stage_status"] = StageStatus.COMPLETED.value
+    return AssessmentResult(
+        output=AssessmentOutput(
+            assessment=parse_assessment_text(out), pre_filter=pre_filter
+        ),
+        error=out["error"],
+    )
+
+
+def _outcome(
+    state: BatchRunState, result: AssessmentResult | None, failure: str | None
+) -> tuple[AssessmentStatus, str | None]:
+    """Final status and the run's error message (stage-prefixed)."""
+    if failure is not None:
+        started = state["stage_status"] == StageStatus.PROCESSING.value
+        message = f"{state['stage']}: {failure}" if started else failure
+        return AssessmentStatus.FAILED, message
+
+    errors = state.get("stage_errors") or {}
+    message = (
+        "; ".join(f"{stage}: {rows[ROW]}" for stage, rows in errors.items()) or None
+    )
+    if result and result.error:
+        return AssessmentStatus.FAILED, message
+    status = (
+        AssessmentStatus.COMPLETED_WITH_ERRORS if errors else AssessmentStatus.COMPLETED
+    )
+    return status, message
+
+
+def _finish(
+    session: Session,
+    assessment: Assessment,
+    execution: AssessmentRun,
+    state: BatchRunState,
+    result: AssessmentResult | None,
+    failure: str | None,
+) -> None:
+    status, message = _outcome(state, result, failure)
+    if status == AssessmentStatus.FAILED:
+        if state["stage_status"] != StageStatus.PENDING.value:
+            state["stage_status"] = StageStatus.FAILED.value
+        if message:
+            state["error"] = message
+    if result:
+        state["result"] = result.model_dump(mode="json")
+    api.save_execution_state(session=session, execution=execution, state=state)
+    execution.error_message = message
+    api.update_status(session=session, obj=execution, status=status)
+    api.update_status(session=session, obj=assessment, status=status)
+
+
+def run_response_assessment(*, execution_id: int) -> None:
+    """Assess one row, record the run and deliver the result or the failure by callback."""
+    with Session(engine) as session:
+        # Row lock claims the run: a redelivered task skips one another worker still holds.
+        execution = session.exec(
+            select(AssessmentRun)
+            .where(col(AssessmentRun.id) == execution_id)
+            .with_for_update(skip_locked=True)
+        ).first()
+        assessment = (
+            session.get(Assessment, execution.assessment_id) if execution else None
+        )
+        if execution is None or assessment is None:
+            logger.error(
+                "[run_response_assessment] Run missing | execution_id=%s", execution_id
+            )
+            return
+        if execution.status in AssessmentStatus.terminal():
+            return
+
+        state = cast(BatchRunState, dict(execution.execution or {}))
+        result: AssessmentResult | None = None
+        failure: str | None = None
+        try:
+            blob = api.get_config_blob(
+                session=session, execution=execution, project_id=assessment.project_id
+            )
+            result = _assess(
+                session=session,
+                assessment=assessment,
+                execution=execution,
+                state=state,
+                blob=blob,
+            )
+        except (Exception, Timeout) as exc:  # gevent's Timeout is a BaseException
+            failure = str(exc)
+
+        _finish(session, assessment, execution, state, result, failure)
+        deliver(
+            session=session,
+            assessment=assessment,
+            result=result,
+            callback_url=cast(str, state["callback_url"]),
+            request_metadata=state["request_metadata"],
+            failure_message=failure,
+        )
diff --git a/backend/app/services/assessment/api/results.py b/backend/app/services/assessment/api/results.py
index 32463b238..e8d06b11e 100644
--- a/backend/app/services/assessment/api/results.py
+++ b/backend/app/services/assessment/api/results.py
@@ -6,7 +6,6 @@
 streamed back from object storage.
 """
 
-import json
 import logging
 from typing import Any, cast
 
@@ -33,27 +32,15 @@
     Submission,
     Verdict,
 )
-from app.services.assessment.api.batch import ApiStage, parse_batch_results
-from app.services.assessment.utils.parsing import parse_stored_results
+from app.services.assessment.api.batch import PipelineStage, parse_batch_results
+from app.services.assessment.utils.parsing import (
+    parse_assessment_text,
+    parse_stored_results,
+)
 
 logger = logging.getLogger(__name__)
 
 
-def _parse_assessment(out: ParsedResult) -> dict[str, Any] | str | None:
-    """Stored assessment text as a dict (structured json_schema output) or raw string.
-
-    Null when the row produced no assessment text (gated out or empty/failed call).
-    """
-    text = out.get("output")
-    if not text:
-        return None
-    try:
-        parsed = json.loads(text)
-    except (json.JSONDecodeError, TypeError):
-        return text
-    return parsed if isinstance(parsed, dict) else text
-
-
 def _verdict_obj(verdict: Verdict | None) -> PreFilterVerdict | None:
     if not verdict:
         return None
@@ -72,7 +59,7 @@ def _load_assessment_outputs(
     can flag it per-row instead of mistaking it for a clean run. Missing URL
     (all rows gated) is a legit empty, not an error.
     """
-    url = (bag.get("stage_output_urls") or {}).get(ApiStage.ASSESSMENT.value)
+    url = (bag.get("stage_output_urls") or {}).get(PipelineStage.ASSESSMENT.value)
     if not url:
         return {}, None
     try:
@@ -112,9 +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(ApiStage.ASSESSMENT.value, {})
+    row_errors = (bag.get("stage_errors") or {}).get(PipelineStage.ASSESSMENT.value, {})
 
-    tr_verdicts = verdicts.get(ApiStage.TOPIC_RELEVANCE.value, {})
+    tr_verdicts = verdicts.get(PipelineStage.TOPIC_RELEVANCE.value, {})
 
     items: list[AssessmentResult] = []
     counts = AssessmentCounts()
@@ -126,7 +113,7 @@ def build_result(
         if gate_passed[idx]:
             if idx in outputs:
                 out = outputs[idx]
-                assessment_output = _parse_assessment(out)
+                assessment_output = parse_assessment_text(out)
                 error = out.get("error")
             elif str(idx) in row_errors:
                 error = row_errors[str(idx)]
diff --git a/backend/app/services/assessment/api/submission.py b/backend/app/services/assessment/api/submission.py
index f038d9603..cf21a235a 100644
--- a/backend/app/services/assessment/api/submission.py
+++ b/backend/app/services/assessment/api/submission.py
@@ -1,14 +1,17 @@
-"""BATCH API-client submit entrypoint.
+"""API-client submit entrypoint.
 
-Resolves the ASSESSMENT config, persists the parent assessment + one execution,
-seeds the staged-batch runtime bag, and hands the pipeline off to Celery. Only the
-BATCH method is wired here; RESPONSE stays a route-level 501 stub (deferred).
+Resolves the ASSESSMENT config and validates the rows. BATCH persists the parent
+assessment + one execution, seeds the staged-batch runtime bag and hands the pipeline
+off to Celery; RESPONSE persists the parent assessment and hands the row off.
 """
 
 import logging
+from typing import Any
 from uuid import UUID, uuid4
 
 from asgi_correlation_id import correlation_id
+from celery import Task
+from celery.result import AsyncResult
 from fastapi import HTTPException
 from pydantic import ValidationError
 from sqlmodel import Session
@@ -17,8 +20,10 @@
 from app.crud.assessment.submission import get_submission_by_id
 from app.crud.config import ConfigCrud, ConfigVersionCrud
 from app.models.assessment import (
+    Assessment,
     AssessmentCreate,
     AssessmentMethod,
+    AssessmentRun,
     AssessmentStatus,
     AssessmentSubmitResponse,
     BatchInput,
@@ -187,22 +192,121 @@ def _resolve_config(
     return blob, provider, model
 
 
-def submit(
+def _enqueue(
+    *,
+    session: Session,
+    task: Task,
+    assessment: Assessment,
+    execution: AssessmentRun | None = None,
+    **task_kwargs: Any,
+) -> AsyncResult:
+    """Dispatch ``task``; a broker failure marks the rows FAILED and returns a 503."""
+    try:
+        return task.delay(**task_kwargs)
+    except Exception as exc:
+        logger.exception("[_enqueue] Failed to enqueue | task=%s", task.name)
+        if execution is not None:
+            api.update_status(
+                session=session, obj=execution, status=AssessmentStatus.FAILED
+            )
+        api.update_status(
+            session=session, obj=assessment, status=AssessmentStatus.FAILED
+        )
+        raise HTTPException(
+            status_code=503,
+            detail="Failed to dispatch the assessment for processing. Please retry.",
+        ) from exc
+
+
+def _initial_state(
+    *,
+    request: AssessmentCreate,
+    blob: AssessmentConfigBlob,
+    provider: str,
+    model: str,
+    total_items: int,
+) -> BatchRunState:
+    pipeline = batch_service.build_pipeline(blob.pre_filters)
+    return {
+        "pipeline": pipeline,
+        "stage": pipeline[0]["stage"],
+        "stage_status": AssessmentStatus.PENDING.value,
+        "stage_batches": {},
+        "stage_output_urls": {},
+        "verdicts": {},
+        "counters": {},
+        "gate_passed": [True] * total_items,
+        "provider": provider,
+        "model": model,
+        "input_schema": blob.input_columns() or None,
+        "callback_url": str(request.callback_url) if request.callback_url else None,
+        "request_metadata": request.request_metadata,
+    }
+
+
+def _start_response(
     *,
     session: Session,
     request: AssessmentCreate,
+    blob: AssessmentConfigBlob,
+    provider: str,
+    model: str,
     organization_id: int,
     project_id: int,
 ) -> AssessmentSubmitResponse:
-    """Submit a BATCH assessment: persist state, seed the pipeline, dispatch the task."""
-    from app.celery.tasks.job_execution import run_assessment_api_batch
+    from app.celery.tasks.job_execution import run_assessment_response
 
-    method = derive_method(request.input, submission_id=None)
-    if method != AssessmentMethod.BATCH:
-        # RESPONSE is handled (stubbed) at the route; submit is BATCH-only for now.
-        raise HTTPException(
-            status_code=501, detail="Only BATCH input is wired in the assessment API."
-        )
+    assessment = api.create_assessment(
+        session=session,
+        method=AssessmentMethod.RESPONSE,
+        input=request.input.model_dump(),
+        experiment_name=request.experiment_name,
+        organization_id=organization_id,
+        project_id=project_id,
+    )
+    execution = api.create_execution(
+        session=session,
+        assessment_id=assessment.id,
+        config_id=request.config.id,
+        config_version=request.config.version,
+        total_items=1,
+    )
+    api.save_execution_state(
+        session=session,
+        execution=execution,
+        state=_initial_state(
+            request=request, blob=blob, provider=provider, model=model, total_items=1
+        ),
+    )
+    api.update_status(
+        session=session, obj=assessment, status=AssessmentStatus.PROCESSING
+    )
+    _enqueue(
+        session=session,
+        task=run_assessment_response,
+        assessment=assessment,
+        execution=execution,
+        execution_id=execution.id,
+        trace_id=correlation_id.get() or "",
+    )
+    return AssessmentSubmitResponse(
+        assessment_id=assessment.id,
+        status=assessment.status,
+        message="Your assessment is being processed",
+        inserted_at=assessment.inserted_at,
+        updated_at=assessment.updated_at,
+    )
+
+
+def submit(
+    *,
+    session: Session,
+    request: AssessmentCreate,
+    organization_id: int,
+    project_id: int,
+) -> AssessmentSubmitResponse:
+    """Submit an assessment: persist state, then dispatch the task for its method."""
+    method = derive_method(request.input)
 
     # Reject an unusable callback_url up front (HTTPS + SSRF/private-IP guard) instead
     # of after a full batch run is already paid for. Absent means the client polls.
@@ -212,7 +316,11 @@ def submit(
         except ValueError as exc:
             raise HTTPException(status_code=422, detail=str(exc)) from exc
 
-    batch_input: BatchInput = request.input
+    batch_input = (
+        request.input
+        if isinstance(request.input, BatchInput)
+        else request.input.as_batch()
+    )
     submission_id = batch_input.submission_doc_id
     if submission_id is not None:
         batch_input = _rows_from_submission(
@@ -226,10 +334,7 @@ def submit(
         session=session, request=request, project_id=project_id
     )
 
-    input_columns = {
-        name: col.model_dump(exclude_none=True)
-        for name, col in blob.input_schema.items()
-    }
+    input_columns = blob.input_columns()
     if submission_id is not None:
         batch_input = BatchInput(
             data=[
@@ -246,6 +351,19 @@ def submit(
         raise HTTPException(status_code=422, detail=str(exc)) from exc
     total_items = len(rows)
 
+    if method == AssessmentMethod.RESPONSE:
+        return _start_response(
+            session=session,
+            request=request,
+            blob=blob,
+            provider=provider,
+            model=model,
+            organization_id=organization_id,
+            project_id=project_id,
+        )
+
+    from app.celery.tasks.job_execution import run_assessment_api_batch
+
     assessment_id = uuid4()
     # Only inline rows need storing; an uploaded submission is already immutable in
     # storage, so the run reads it through `submission_id` instead of copying it.
@@ -282,51 +400,28 @@ def submit(
         total_items=total_items,
     )
 
-    pipeline = batch_service.build_pipeline(blob.pre_filters)
-    bag: BatchRunState = {
-        "pipeline": pipeline,
-        "stage": pipeline[0]["stage"],
-        "stage_status": AssessmentStatus.PENDING.value,
-        "stage_batches": {},
-        "stage_output_urls": {},
-        "verdicts": {},
-        "counters": {},
-        "gate_passed": [True] * total_items,
-        "provider": provider,
-        "model": model,
-        "input_schema": input_columns or None,
-        "callback_url": str(request.callback_url) if request.callback_url else None,
-        "request_metadata": request.request_metadata,
-    }
+    bag = _initial_state(
+        request=request,
+        blob=blob,
+        provider=provider,
+        model=model,
+        total_items=total_items,
+    )
     api.save_execution_state(session=session, execution=execution, state=bag)
     api.update_status(
         session=session, obj=assessment, status=AssessmentStatus.PROCESSING
     )
 
-    trace_id = correlation_id.get() or ""
-    try:
-        dispatched = run_assessment_api_batch.delay(
-            execution_id=execution.id,
-            organization_id=organization_id,
-            project_id=project_id,
-            trace_id=trace_id,
-        )
-    except Exception as exc:
-        logger.exception(
-            "[submit] Failed to enqueue BATCH task | assessment_id=%s | execution_id=%s",
-            assessment.id,
-            execution.id,
-        )
-        api.update_status(
-            session=session, obj=execution, status=AssessmentStatus.FAILED
-        )
-        api.update_status(
-            session=session, obj=assessment, status=AssessmentStatus.FAILED
-        )
-        raise HTTPException(
-            status_code=503,
-            detail="Failed to dispatch the assessment for processing. Please retry.",
-        ) from exc
+    dispatched = _enqueue(
+        session=session,
+        task=run_assessment_api_batch,
+        assessment=assessment,
+        execution=execution,
+        execution_id=execution.id,
+        organization_id=organization_id,
+        project_id=project_id,
+        trace_id=correlation_id.get() or "",
+    )
     logger.info(
         "[submit] Dispatched BATCH assessment | assessment_id=%s | execution_id=%s | "
         "task_id=%s | provider=%s | stages=%s | rows=%s",
@@ -334,7 +429,7 @@ def submit(
         execution.id,
         dispatched.id,
         provider,
-        [s["stage"] for s in pipeline],
+        [s["stage"] for s in bag["pipeline"]],
         total_items,
     )
 
diff --git a/backend/app/services/assessment/utils/parsing.py b/backend/app/services/assessment/utils/parsing.py
index fbb064484..3e969dd27 100644
--- a/backend/app/services/assessment/utils/parsing.py
+++ b/backend/app/services/assessment/utils/parsing.py
@@ -3,6 +3,8 @@
 import json
 from typing import Any
 
+from app.models.assessment import ParsedResult
+
 
 def parse_stored_results(raw_content: str) -> list[dict[str, Any]]:
     """Parse stored batch results from JSONL or JSON array."""
@@ -17,6 +19,21 @@ def parse_stored_results(raw_content: str) -> list[dict[str, Any]]:
     return [json.loads(line) for line in content.splitlines() if line.strip()]
 
 
+def parse_assessment_text(out: ParsedResult) -> dict[str, Any] | str | None:
+    """Stored assessment text as a dict (structured json_schema output) or raw string.
+
+    Null when the row produced no assessment text (gated out or empty/failed call).
+    """
+    text = out.get("output")
+    if not text:
+        return None
+    try:
+        parsed = json.loads(text)
+    except (json.JSONDecodeError, TypeError):
+        return text
+    return parsed if isinstance(parsed, dict) else text
+
+
 def usage_totals(usage: Any) -> tuple[int | None, int | None, int | None]:
     """Extract common token totals from provider usage payloads."""
     if not isinstance(usage, dict):
diff --git a/backend/app/tests/assessment/test_api_batch.py b/backend/app/tests/assessment/test_api_batch.py
index 2f193b14c..7750e6d03 100644
--- a/backend/app/tests/assessment/test_api_batch.py
+++ b/backend/app/tests/assessment/test_api_batch.py
@@ -27,24 +27,24 @@
 from app.models.config.config import ConfigTag
 from app.services.assessment.api import batch as batch_service
 from app.services.assessment.api.batch import (
-    ApiStage,
+    PipelineStage,
     StageKind,
     _build_batch_provider,
     _err_str,
     _fail,
     _finalize,
     _openai_output_text,
-    _parse_one,
-    _parse_verdict,
+    parse_one,
+    parse_verdict,
     _poll_outcome,
     _prefilter_for_stage,
     _record_stage,
     _row_index,
     _row_subset,
     _stage_kind,
-    _stage_params,
-    _stage_prompt,
-    _stage_provider_model,
+    stage_params,
+    stage_prompt,
+    stage_provider_model,
     _submit_provider_batch,
     _submit_stage,
     build_pipeline,
@@ -147,15 +147,18 @@ class TestBuildPipeline:
     def test_none_prefilters_is_assessment_only(self) -> None:
         pipeline = build_pipeline(None)
         assert pipeline == [
-            {"stage": ApiStage.ASSESSMENT.value, "kind": StageKind.ASSESSMENT.value}
+            {
+                "stage": PipelineStage.ASSESSMENT.value,
+                "kind": StageKind.ASSESSMENT.value,
+            }
         ]
 
     def test_gate_precedes_assessment(self) -> None:
         blob = _blob(topic_relevance=True)  # stop_on_fail=True -> GATE
         pipeline = build_pipeline(blob.pre_filters)
         assert [s["stage"] for s in pipeline] == [
-            ApiStage.TOPIC_RELEVANCE.value,
-            ApiStage.ASSESSMENT.value,
+            PipelineStage.TOPIC_RELEVANCE.value,
+            PipelineStage.ASSESSMENT.value,
         ]
         assert [s["kind"] for s in pipeline] == [
             StageKind.GATE.value,
@@ -166,8 +169,8 @@ def test_passthrough_precedes_assessment(self) -> None:
         blob = _blob(topic_relevance=False)  # stop_on_fail=False -> PASS_THROUGH
         pipeline = build_pipeline(blob.pre_filters)
         assert [s["stage"] for s in pipeline] == [
-            ApiStage.TOPIC_RELEVANCE.value,
-            ApiStage.ASSESSMENT.value,
+            PipelineStage.TOPIC_RELEVANCE.value,
+            PipelineStage.ASSESSMENT.value,
         ]
         assert [s["kind"] for s in pipeline] == [
             StageKind.PASS_THROUGH.value,
@@ -179,15 +182,15 @@ class TestNextStageAndKind:
     def test_next_stage_advances_then_none(self) -> None:
         pipeline = build_pipeline(_blob(topic_relevance=True).pre_filters)
         assert (
-            next_stage(pipeline, ApiStage.TOPIC_RELEVANCE.value)
-            == ApiStage.ASSESSMENT.value
+            next_stage(pipeline, PipelineStage.TOPIC_RELEVANCE.value)
+            == PipelineStage.ASSESSMENT.value
         )
-        assert next_stage(pipeline, ApiStage.ASSESSMENT.value) is 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)
         assert (
-            _stage_kind(pipeline, ApiStage.TOPIC_RELEVANCE.value)
+            _stage_kind(pipeline, PipelineStage.TOPIC_RELEVANCE.value)
             == StageKind.GATE.value
         )
         with pytest.raises(ValueError, match="not in pipeline"):
@@ -228,40 +231,40 @@ def test_empty_submissions(self) -> None:
 class TestStagePrompt:
     def test_assessment_uses_config_submission(self) -> None:
         blob = _blob(submission="assess {a} now")
-        assert _stage_prompt(blob, ApiStage.ASSESSMENT.value) == "assess {a} now"
+        assert stage_prompt(blob, PipelineStage.ASSESSMENT.value) == "assess {a} now"
 
     def test_prefilter_returns_its_own_submission(self) -> None:
         blob = _blob(topic_relevance=True, topic_submission="rank {a}")
-        assert _stage_prompt(blob, ApiStage.TOPIC_RELEVANCE.value) == "rank {a}"
+        assert stage_prompt(blob, PipelineStage.TOPIC_RELEVANCE.value) == "rank {a}"
 
     def test_prefilter_without_submission_is_none(self) -> None:
         blob = _blob(topic_relevance=True)
-        assert _stage_prompt(blob, ApiStage.TOPIC_RELEVANCE.value) is None
+        assert stage_prompt(blob, PipelineStage.TOPIC_RELEVANCE.value) is None
 
 
 class TestPrefilterForStage:
     def test_returns_configured_filter(self) -> None:
         blob = _blob(topic_relevance=True)
-        tr = _prefilter_for_stage(blob, ApiStage.TOPIC_RELEVANCE.value)
+        tr = _prefilter_for_stage(blob, PipelineStage.TOPIC_RELEVANCE.value)
         assert tr.params["instructions"] == TOPIC_CRITERIA
 
     def test_missing_filter_raises(self) -> None:
         blob = _blob()  # no pre-filters
         with pytest.raises(ValueError, match="No pre-filter configured"):
-            _prefilter_for_stage(blob, ApiStage.TOPIC_RELEVANCE.value)
+            _prefilter_for_stage(blob, PipelineStage.TOPIC_RELEVANCE.value)
 
 
 class TestStageParams:
     def test_assessment_renames_output_schema_and_drops_input_schema(self) -> None:
         blob = _blob(input_schema={"a": {"type": "text"}})
-        params = _stage_params(blob, ApiStage.ASSESSMENT.value)
+        params = stage_params(blob, PipelineStage.ASSESSMENT.value)
         assert params["output_schema"] == OUTPUT_SCHEMA
         assert "json_output_schema" not in params
         assert "input_schema" not in params
 
     def test_prefilter_layers_verdict_schema_and_appends_gate_directive(self) -> None:
         blob = _blob(topic_relevance=True)
-        params = _stage_params(blob, ApiStage.TOPIC_RELEVANCE.value)
+        params = stage_params(blob, PipelineStage.TOPIC_RELEVANCE.value)
         assert params["output_schema"] == batch_service.PREFILTER_VERDICT_SCHEMA
         assert (
             params["instructions"]
@@ -275,21 +278,23 @@ def test_submission_is_not_a_provider_param(self) -> None:
             topic_relevance=True,
             topic_submission="rank {a}",
         )
-        assert "submission" not in _stage_params(blob, ApiStage.ASSESSMENT.value)
-        assert "submission" not in _stage_params(blob, ApiStage.TOPIC_RELEVANCE.value)
+        assert "submission" not in stage_params(blob, PipelineStage.ASSESSMENT.value)
+        assert "submission" not in stage_params(
+            blob, PipelineStage.TOPIC_RELEVANCE.value
+        )
 
 
 class TestStageProviderModel:
     def test_assessment_uses_config_provider_model(self) -> None:
         blob = _blob(provider="anthropic", model="claude-sonnet-4-6")
-        assert _stage_provider_model(blob, ApiStage.ASSESSMENT.value) == (
+        assert stage_provider_model(blob, PipelineStage.ASSESSMENT.value) == (
             "anthropic",
             "claude-sonnet-4-6",
         )
 
     def test_prefilter_uses_own_provider_model(self) -> None:
         blob = _blob(topic_relevance=True)
-        assert _stage_provider_model(blob, ApiStage.TOPIC_RELEVANCE.value) == (
+        assert stage_provider_model(blob, PipelineStage.TOPIC_RELEVANCE.value) == (
             "openai",
             "gpt-4o",
         )
@@ -338,12 +343,12 @@ def test_non_message_items_ignored(self) -> None:
 
 class TestParseOne:
     def test_top_level_error(self) -> None:
-        result = _parse_one({"error": {"message": "bad"}}, "openai")
+        result = parse_one({"error": {"message": "bad"}}, "openai")
         assert result["output"] is None
         assert result["error"] == "bad"
 
     def test_openai_success(self) -> None:
-        result = _parse_one(
+        result = parse_one(
             {
                 "response": {
                     "status_code": 200,
@@ -361,7 +366,7 @@ def test_openai_success(self) -> None:
         assert result["response_id"] == "resp_1"
 
     def test_openai_http_error_status(self) -> None:
-        result = _parse_one(
+        result = parse_one(
             {
                 "response": {
                     "status_code": 500,
@@ -375,12 +380,12 @@ def test_openai_http_error_status(self) -> None:
         assert result["response_id"] == "resp_2"
 
     def test_openai_empty_output(self) -> None:
-        result = _parse_one({"response": {"status_code": 200, "body": {}}}, "openai")
+        result = parse_one({"response": {"status_code": 200, "body": {}}}, "openai")
         assert result["output"] is None
         assert result["error"] == "Empty response output"
 
     def test_anthropic_success(self) -> None:
-        result = _parse_one(
+        result = parse_one(
             {
                 "response": {
                     "content": [
@@ -398,7 +403,7 @@ def test_anthropic_success(self) -> None:
         assert result["response_id"] == "msg_1"
 
     def test_anthropic_empty(self) -> None:
-        result = _parse_one({"response": {"content": []}}, "anthropic")
+        result = parse_one({"response": {"content": []}}, "anthropic")
         assert result["output"] is None
         assert result["error"] == "Empty response"
 
@@ -407,11 +412,11 @@ def test_google_success(self) -> None:
             "app.services.assessment.api.batch.extract_text_from_response_dict",
             return_value="gemini text",
         ):
-            result = _parse_one({"response": {"candidates": []}}, "google")
+            result = parse_one({"response": {"candidates": []}}, "google")
         assert result["output"] == "gemini text"
 
     def test_google_empty_response(self) -> None:
-        result = _parse_one({"response": None}, "google")
+        result = parse_one({"response": None}, "google")
         assert result["output"] is None
         assert result["error"] == "Empty response"
 
@@ -420,7 +425,7 @@ def test_google_gcp_parses_like_google(self) -> None:
             "app.services.assessment.api.batch.extract_text_from_response_dict",
             return_value="vertex text",
         ):
-            result = _parse_one({"response": {"candidates": []}}, "google-gcp")
+            result = parse_one({"response": {"candidates": []}}, "google-gcp")
         assert result["output"] == "vertex text"
 
     def test_google_aistudio_parses_like_google(self) -> None:
@@ -428,12 +433,12 @@ def test_google_aistudio_parses_like_google(self) -> None:
             "app.services.assessment.api.batch.extract_text_from_response_dict",
             return_value="aistudio text",
         ):
-            result = _parse_one({"response": {"candidates": []}}, "google-aistudio")
+            result = parse_one({"response": {"candidates": []}}, "google-aistudio")
         assert result["output"] == "aistudio text"
         assert result["error"] is None
 
     def test_unknown_provider(self) -> None:
-        result = _parse_one({"response": {}}, "cohere")
+        result = parse_one({"response": {}}, "cohere")
         assert result["output"] is None
         assert result["error"] == "Unknown provider cohere"
 
@@ -451,17 +456,17 @@ def test_indexes_by_row_key_and_skips_bad_keys(self) -> None:
 
 class TestParseVerdict:
     def test_empty_output_fails_open(self) -> None:
-        assert _parse_verdict(None) == {"verdict": True, "reasoning": ""}
+        assert parse_verdict(None) == {"verdict": True, "reasoning": ""}
 
     def test_valid_json(self) -> None:
         out = json.dumps({"verdict": False, "reasoning": "off topic"})
-        assert _parse_verdict(out) == {"verdict": False, "reasoning": "off topic"}
+        assert parse_verdict(out) == {"verdict": False, "reasoning": "off topic"}
 
     def test_unparseable_fails_open(self) -> None:
-        assert _parse_verdict("not json {") == {"verdict": True, "reasoning": ""}
+        assert parse_verdict("not json {") == {"verdict": True, "reasoning": ""}
 
     def test_missing_verdict_key_defaults_true(self) -> None:
-        assert _parse_verdict(json.dumps({"reasoning": "x"})) == {
+        assert parse_verdict(json.dumps({"reasoning": "x"})) == {
             "verdict": True,
             "reasoning": "x",
         }
@@ -473,7 +478,9 @@ def _bag(self, total: int) -> dict:
 
     def test_assessment_kind_is_noop(self) -> None:
         bag = self._bag(2)
-        _record_stage(bag, ApiStage.ASSESSMENT.value, StageKind.ASSESSMENT.value, {})
+        _record_stage(
+            bag, PipelineStage.ASSESSMENT.value, StageKind.ASSESSMENT.value, {}
+        )
         assert bag["verdicts"] == {}
 
     def test_gate_marks_failed_rows(self) -> None:
@@ -482,9 +489,11 @@ def test_gate_marks_failed_rows(self) -> None:
             0: {"output": json.dumps({"verdict": True, "reasoning": ""})},
             1: {"output": json.dumps({"verdict": False, "reasoning": "no"})},
         }
-        _record_stage(bag, ApiStage.TOPIC_RELEVANCE.value, StageKind.GATE.value, parsed)
+        _record_stage(
+            bag, PipelineStage.TOPIC_RELEVANCE.value, StageKind.GATE.value, parsed
+        )
         assert bag["gate_passed"] == [True, False]
-        counters = bag["counters"][ApiStage.TOPIC_RELEVANCE.value]
+        counters = bag["counters"][PipelineStage.TOPIC_RELEVANCE.value]
         assert counters == {"total": 2, "passed": 1, "rejected": 1}
 
     def test_pass_through_records_but_does_not_gate(self) -> None:
@@ -495,25 +504,25 @@ def test_pass_through_records_but_does_not_gate(self) -> None:
         }
         _record_stage(
             bag,
-            ApiStage.TOPIC_RELEVANCE.value,
+            PipelineStage.TOPIC_RELEVANCE.value,
             StageKind.PASS_THROUGH.value,
             parsed,
         )
         assert bag["gate_passed"] == [True, True]
-        assert bag["counters"][ApiStage.TOPIC_RELEVANCE.value]["rejected"] == 1
+        assert bag["counters"][PipelineStage.TOPIC_RELEVANCE.value]["rejected"] == 1
 
 
 class TestRowSubset:
     def test_assessment_only_gate_passed(self) -> None:
         bag = {"gate_passed": [True, False, True]}
         assert _row_subset(
-            bag, ApiStage.ASSESSMENT.value, StageKind.ASSESSMENT.value, 3
+            bag, PipelineStage.ASSESSMENT.value, StageKind.ASSESSMENT.value, 3
         ) == [0, 2]
 
     def test_prefilter_all_rows(self) -> None:
         bag = {"gate_passed": [True, False, True]}
         assert _row_subset(
-            bag, ApiStage.TOPIC_RELEVANCE.value, StageKind.GATE.value, 3
+            bag, PipelineStage.TOPIC_RELEVANCE.value, StageKind.GATE.value, 3
         ) == [0, 1, 2]
 
 
@@ -718,7 +727,7 @@ def call_tick():
         assert assessment.status == AssessmentStatus.COMPLETED
         bag_after = execution.execution
         assert bag_after["gate_passed"] == [True, False]
-        tr_verdicts = bag_after["verdicts"][ApiStage.TOPIC_RELEVANCE.value]
+        tr_verdicts = bag_after["verdicts"][PipelineStage.TOPIC_RELEVANCE.value]
         assert tr_verdicts["0"]["verdict"] is True
         assert tr_verdicts["1"]["verdict"] is False
         assert build_result.called
@@ -1114,14 +1123,14 @@ def test_all_gated_out_skips_submit(self, db) -> None:
             assessment=assessment,
             blob=blob,
             bag=bag,
-            stage=ApiStage.ASSESSMENT.value,
+            stage=PipelineStage.ASSESSMENT.value,
             organization_id=auth.organization_id,
             project_id=auth.project_id,
         )
         # Empty subset submits no provider batch and reports False so the driver
         # advances/finalizes instead of re-submitting a PENDING stage forever.
         assert ok is False
-        assert bag["counters"][ApiStage.ASSESSMENT.value] == {
+        assert bag["counters"][PipelineStage.ASSESSMENT.value] == {
             "total": 0,
             "passed": 0,
             "rejected": 0,
@@ -1594,7 +1603,7 @@ def test_blob_resolution_error_fails_run(self, db) -> None:
                 lambda _e: _SessionCtx(db),
             ),
             patch(
-                "app.services.assessment.api.batch._resolve_blob",
+                "app.services.assessment.api.batch.api.get_config_blob",
                 side_effect=HTTPException(
                     status_code=404, detail="config version deleted"
                 ),
diff --git a/backend/app/tests/assessment/test_api_crud.py b/backend/app/tests/assessment/test_api_crud.py
index a4dd0bbda..eef8852d2 100644
--- a/backend/app/tests/assessment/test_api_crud.py
+++ b/backend/app/tests/assessment/test_api_crud.py
@@ -221,19 +221,12 @@ def test_rerecording_a_kind_replaces_only_that_kind(self, db) -> None:
 
 class TestDeriveMethod:
     def test_response_input(self) -> None:
-        assert derive_method(ResponseInput(), None) == AssessmentMethod.RESPONSE
-
-    def test_batch_input(self) -> None:
         assert (
-            derive_method(BatchInput(data=[{"a": "1"}]), None) == AssessmentMethod.BATCH
+            derive_method(ResponseInput(data={"a": "1"})) == AssessmentMethod.RESPONSE
         )
 
-    def test_dataset_only_is_run(self) -> None:
-        assert derive_method(None, 7) == AssessmentMethod.RUN
-
-    def test_nothing_raises(self) -> None:
-        with pytest.raises(ValueError, match="Provide inline"):
-            derive_method(None, None)
+    def test_batch_input(self) -> None:
+        assert derive_method(BatchInput(data=[{"a": "1"}])) == AssessmentMethod.BATCH
 
 
 class TestPreFilterValidators:
@@ -531,10 +524,10 @@ def test_batch_input_without_query_accepted(self) -> None:
 
     def test_response_input_rejects_query_key(self) -> None:
         with pytest.raises(ValidationError):
-            ResponseInput.model_validate({"query": "hi"})
+            ResponseInput.model_validate({"query": "hi", "data": {"a": "1"}})
 
     def test_response_input_without_query_accepted(self) -> None:
-        assert ResponseInput.model_validate({"attachments": []}).attachments == []
+        assert ResponseInput.model_validate({"data": {"a": "1"}}).data == {"a": "1"}
 
 
 class TestInputColumn:
diff --git a/backend/app/tests/assessment/test_api_response.py b/backend/app/tests/assessment/test_api_response.py
new file mode 100644
index 000000000..53cf19c3c
--- /dev/null
+++ b/backend/app/tests/assessment/test_api_response.py
@@ -0,0 +1,287 @@
+"""Tests for the RESPONSE runner (app/services/assessment/api/response.py)."""
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from gevent import Timeout
+
+from app.crud.assessment import api
+from app.models.assessment import AssessmentMethod, AssessmentStatus
+from app.models.config.assessment_blob import AssessmentConfigBlob
+from app.models.config.config import ConfigTag
+from app.services.assessment.api import response
+from app.services.assessment.api.batch import build_pipeline
+from app.tests.assessment.test_api_batch import _SessionCtx, _bag_for, _blob_dict
+from app.tests.utils.auth import get_user_test_auth_context
+from app.tests.utils.test_data import create_test_config
+from app.tests.utils.utils import random_lower_string
+
+_RUN_STAGE = "app.services.assessment.api.response._run_stage"
+_DELIVER = "app.services.assessment.api.response.deliver"
+_PROVIDER = "app.services.assessment.api.response.get_llm_provider"
+
+
+class _Dumpable:
+    """Stand-in for an SDK response object: ``model_dump`` returns the given dict."""
+
+    def __init__(self, payload: dict):
+        self._payload = payload
+
+    def model_dump(self, **_):
+        return self._payload
+
+
+def _fake_client(provider: str, text: str = "ok", raise_exc: Exception | None = None):
+    def _create(**_):
+        if raise_exc:
+            raise raise_exc
+        if provider == "openai":
+            return _Dumpable({"id": "r1", "output_text": text})
+        if provider == "anthropic":
+            return _Dumpable({"id": "m1", "content": [{"type": "text", "text": text}]})
+        return _Dumpable({"candidates": [{"content": {"parts": [{"text": text}]}}]})
+
+    if provider == "openai":
+        return SimpleNamespace(responses=SimpleNamespace(create=_create))
+    if provider == "anthropic":
+        return SimpleNamespace(messages=SimpleNamespace(create=_create))
+    genai = SimpleNamespace(models=SimpleNamespace(generate_content=_create))
+    return SimpleNamespace(genai=genai) if provider == "google-gcp" else genai
+
+
+def _blob(**kwargs) -> AssessmentConfigBlob:
+    return AssessmentConfigBlob.model_validate(_blob_dict(**kwargs))
+
+
+def _seed_response(db, auth, blob):
+    config = create_test_config(
+        db,
+        project_id=auth.project_id,
+        name=f"assess-{random_lower_string()}",
+        config_blob=blob,
+        tag=ConfigTag.ASSESSMENT,
+    )
+    assessment = api.create_assessment(
+        session=db,
+        method=AssessmentMethod.RESPONSE,
+        input={"data": {"a": "one"}},
+        organization_id=auth.organization_id,
+        project_id=auth.project_id,
+    )
+    execution = api.create_execution(
+        session=db,
+        assessment_id=assessment.id,
+        config_id=config.id,
+        config_version=1,
+        total_items=1,
+    )
+    api.save_execution_state(
+        session=db,
+        execution=execution,
+        state=_bag_for(build_pipeline(blob.pre_filters), total=1),
+    )
+    api.update_status(session=db, obj=assessment, status=AssessmentStatus.PROCESSING)
+    api.update_status(session=db, obj=execution, status=AssessmentStatus.PROCESSING)
+    return assessment, execution
+
+
+def _stage_output(
+    stage: str, *, verdict: bool = True, error: str | None = None
+) -> dict:
+    if stage == "assessment":
+        return {"output": '{"score": 1}', "error": None}
+    out = f'{{"verdict": {str(verdict).lower()}, "reasoning": "r"}}'
+    return {"output": None if error else out, "error": error}
+
+
+def _run(db, execution_id: int, stage_output) -> list:
+    delivered: list = []
+    with (
+        patch(_RUN_STAGE, side_effect=lambda **kw: stage_output(kw["stage"])),
+        patch(_DELIVER, side_effect=lambda **kw: delivered.append(kw)),
+        patch.object(response, "Session", lambda _: _SessionCtx(db)),
+    ):
+        response.run_response_assessment(execution_id=execution_id)
+    return delivered
+
+
+class TestRunStage:
+    """One provider call per stage, built from the batch request builders."""
+
+    @pytest.mark.parametrize(
+        "provider", ["openai", "anthropic", "google-aistudio", "google-gcp"]
+    )
+    def test_each_provider_round_trips_through_parse_one(self, db, provider) -> None:
+        auth = get_user_test_auth_context(db)
+        raw = _blob_dict(provider=provider)
+        raw["assessment"]["params"]["instructions"] = "Grade it."
+        blob = AssessmentConfigBlob.model_validate(raw)
+        with patch(
+            _PROVIDER,
+            return_value=SimpleNamespace(client=_fake_client(provider, text="graded")),
+        ):
+            out = response._run_stage(
+                session=db,
+                blob=blob,
+                stage="assessment",
+                row={"a": "one"},
+                organization_id=auth.organization_id,
+                project_id=auth.project_id,
+            )
+        assert out == {
+            "output": "graded",
+            "error": None,
+            "usage": None,
+            "response_id": {"openai": "r1", "anthropic": "m1"}.get(provider),
+        }
+
+    def test_provider_exception_becomes_the_row_error(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        with patch(
+            _PROVIDER,
+            return_value=SimpleNamespace(
+                client=_fake_client("openai", raise_exc=RuntimeError("rate limited"))
+            ),
+        ):
+            out = response._run_stage(
+                session=db,
+                blob=_blob(),
+                stage="assessment",
+                row={"a": "one"},
+                organization_id=auth.organization_id,
+                project_id=auth.project_id,
+            )
+        assert out["output"] is None
+        assert out["error"] == "rate limited"
+
+
+class TestRunResponseAssessment:
+    def test_gate_then_assessment_completes_and_delivers(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob(topic_relevance=True))
+
+        (call,) = _run(db, execution.id, _stage_output)
+
+        db.refresh(parent)
+        db.refresh(execution)
+        assert parent.status == AssessmentStatus.COMPLETED
+        assert execution.execution["stage"] == "assessment"
+        assert execution.execution["stage_status"] == "COMPLETED"
+        assert execution.execution["counters"]["topic_relevance"]["passed"] == 1
+        output = call["result"].output
+        assert output.assessment == {"score": 1}
+        assert output.pre_filter.topic_relevance.verdict is True
+        assert call["failure_message"] is None
+
+    def test_rejected_by_the_gate_skips_the_assessment(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob(topic_relevance=True))
+
+        (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]
+        assert execution.execution["stage"] == "topic_relevance"
+        output = call["result"].output
+        assert output.assessment is None
+        assert output.pre_filter.topic_relevance.verdict is False
+
+    def test_failed_pre_filter_call_fails_open_and_is_reported(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob(topic_relevance=True))
+
+        def stage_output(stage):
+            if stage == "topic_relevance":
+                return _stage_output(stage, error="boom")
+            return _stage_output(stage)
+
+        (call,) = _run(db, execution.id, stage_output)
+
+        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.verdict is True
+        assert verdict.error == "boom"
+        assert call["result"].output.assessment == {"score": 1}
+
+    def test_failed_assessment_call_marks_the_run_failed(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob())
+
+        (call,) = _run(
+            db, execution.id, lambda stage: {"output": None, "error": "provider down"}
+        )
+
+        db.refresh(parent)
+        db.refresh(execution)
+        assert parent.status == AssessmentStatus.FAILED
+        assert execution.error_message == "assessment: provider down"
+        assert execution.execution["stage_status"] == "FAILED"
+        assert call["result"].error == "provider down"
+
+    def test_unexpected_exception_is_a_failure_with_the_stage(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob())
+
+        def explode(stage):
+            raise RuntimeError("kaboom")
+
+        (call,) = _run(db, execution.id, explode)
+
+        db.refresh(parent)
+        db.refresh(execution)
+        assert parent.status == AssessmentStatus.FAILED
+        assert execution.error_message == "assessment: kaboom"
+        assert execution.execution["error"] == "assessment: kaboom"
+        assert call["result"] is None
+        assert call["failure_message"] == "kaboom"
+
+    def test_gevent_timeout_is_a_failure_with_a_callback(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob())
+
+        def time_out(stage):
+            raise Timeout(600)
+
+        (call,) = _run(db, execution.id, time_out)
+
+        db.refresh(parent)
+        assert parent.status == AssessmentStatus.FAILED
+        assert execution.error_message == "assessment: 600 seconds"
+        assert call["result"] is None
+
+    def test_missing_config_version_fails_before_any_stage(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob())
+        execution.config_version = 99
+        db.add(execution)
+        db.commit()
+
+        (call,) = _run(db, execution.id, _stage_output)
+
+        db.refresh(parent)
+        db.refresh(execution)
+        assert parent.status == AssessmentStatus.FAILED
+        assert execution.execution["stage_status"] == "PENDING"
+        assert call["result"] is None
+
+    def test_terminal_run_is_not_rerun(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        parent, execution = _seed_response(db, auth, _blob())
+        api.update_status(session=db, obj=execution, status=AssessmentStatus.COMPLETED)
+
+        delivered = _run(db, execution.id, _stage_output)
+
+        assert delivered == []
+
+    def test_missing_run_is_a_no_op(self, db) -> None:
+        delivered = _run(db, 10**9, _stage_output)
+
+        assert delivered == []
diff --git a/backend/app/tests/assessment/test_api_submission.py b/backend/app/tests/assessment/test_api_submission.py
index 57d748034..0f86a5eed 100644
--- a/backend/app/tests/assessment/test_api_submission.py
+++ b/backend/app/tests/assessment/test_api_submission.py
@@ -15,6 +15,7 @@
 from app.models.assessment import (
     Assessment,
     AssessmentCreate,
+    AssessmentMethod,
     AssessmentStatus,
 )
 from app.models.config.assessment_blob import AssessmentConfigBlob
@@ -127,6 +128,41 @@ def test_creates_assessment_run_and_dispatches(self, db) -> None:
         assert bag["stage_status"] == AssessmentStatus.PENDING.value
         assert bag["callback_url"].startswith("https://hook.example")
 
+    def test_response_creates_run_with_the_row_and_dispatches(self, db) -> None:
+        auth = get_user_test_auth_context(db)
+        config = _assessment_config(db, auth.project_id)
+        request = _request(config, {"a": "one"})
+
+        with patch("app.celery.tasks.job_execution.run_assessment_response") as task:
+            response = submission.submit(
+                session=db,
+                request=request,
+                organization_id=auth.organization_id,
+                project_id=auth.project_id,
+            )
+
+        assert response.status == AssessmentStatus.PROCESSING
+        task.delay.assert_called_once()
+        assessment = db.get(Assessment, response.assessment_id)
+        assert assessment.method == AssessmentMethod.RESPONSE
+        assert assessment.input == {"data": {"a": "one"}}
+        assert assessment.submission_input is None
+        (execution,) = api.list_executions(
+            session=db, assessment_id=response.assessment_id
+        )
+        assert execution.total_items == 1
+        assert execution.execution["gate_passed"] == [True]
+        assert task.delay.call_args.kwargs["execution_id"] == execution.id
+
+    def test_response_without_callback_url_is_rejected(self) -> None:
+        with pytest.raises(ValueError, match="needs a callback_url"):
+            AssessmentCreate.model_validate(
+                {
+                    "config": {"id": str(uuid4()), "version": 1},
+                    "input": {"data": {"a": "one"}},
+                }
+            )
+
     def test_immediate_dispatch_keeps_trace_propagation(self, db) -> None:
         auth = get_user_test_auth_context(db)
         config = _assessment_config(db, auth.project_id)
@@ -290,25 +326,6 @@ def test_config_not_found_is_404(self, db) -> None:
             )
         assert exc.value.status_code == 404
 
-    def test_response_input_method_is_501(self, db) -> None:
-        auth = get_user_test_auth_context(db)
-        config = _assessment_config(db, auth.project_id)
-        request = AssessmentCreate.model_validate(
-            {
-                "config": {"id": str(config.id), "version": 1},
-                "input": {"attachments": []},
-                "callback_url": "https://hook.example/cb",
-            }
-        )
-        with pytest.raises(HTTPException) as exc:
-            submission.submit(
-                session=db,
-                request=request,
-                organization_id=auth.organization_id,
-                project_id=auth.project_id,
-            )
-        assert exc.value.status_code == 501
-
     def test_base64_attachment_column_is_422(self, db) -> None:
         auth = get_user_test_auth_context(db)
         config = _assessment_config(
@@ -539,18 +556,3 @@ def test_batch_input_dispatches_and_returns_202_body(
         assert data["status"] == "PROCESSING"
         assert data["message"] == "Your assessment is being processed"
         task.delay.assert_called_once()
-
-    def test_response_input_is_501(
-        self, client: TestClient, superuser_api_key_header
-    ) -> None:
-        body = {
-            "config": {"id": "00000000-0000-0000-0000-000000000001", "version": 1},
-            "input": {"attachments": []},
-            "callback_url": "https://hook.example/cb",
-        }
-        resp = client.post(
-            f"{settings.API_V1_STR}/assessments",
-            headers=superuser_api_key_header,
-            json=body,
-        )
-        assert resp.status_code == 501
diff --git a/backend/app/tests/assessment/test_result_files.py b/backend/app/tests/assessment/test_result_files.py
index f30423f4e..4eee35969 100644
--- a/backend/app/tests/assessment/test_result_files.py
+++ b/backend/app/tests/assessment/test_result_files.py
@@ -12,7 +12,7 @@
 from app.models.batch_job import BatchJob, BatchJobType
 from app.models.config.assessment_blob import AssessmentConfigBlob
 from app.models.config.config import ConfigTag
-from app.services.assessment.api.batch import ApiStage
+from app.services.assessment.api.batch import PipelineStage
 from app.services.assessment.api.result_files import (
     build_and_upload_errors,
     finalize_result_files,
@@ -39,8 +39,8 @@
 
 def _bag(**overrides) -> BatchRunState:
     bag: dict = {
-        "pipeline": [{"stage": ApiStage.ASSESSMENT.value, "kind": "ASSESSMENT"}],
-        "stage": ApiStage.ASSESSMENT.value,
+        "pipeline": [{"stage": PipelineStage.ASSESSMENT.value, "kind": "ASSESSMENT"}],
+        "stage": PipelineStage.ASSESSMENT.value,
         "stage_status": "COMPLETED",
         "stage_batches": {},
         "stage_output_urls": {},
@@ -141,7 +141,7 @@ def test_dump_is_on_the_parent_row_before_any_terminal_state(self, db) -> None:
         record_stage_dump(
             session=db,
             assessment=assessment,
-            stage=ApiStage.TOPIC_RELEVANCE.value,
+            stage=PipelineStage.TOPIC_RELEVANCE.value,
             url="s3://bucket/batch-1170/output.jsonl",
         )
 
@@ -159,7 +159,7 @@ def test_missing_url_records_nothing(self, db) -> None:
         record_stage_dump(
             session=db,
             assessment=assessment,
-            stage=ApiStage.ASSESSMENT.value,
+            stage=PipelineStage.ASSESSMENT.value,
             url=None,
         )
 
@@ -190,7 +190,7 @@ def test_row_errors_are_flattened_per_stage(self, db) -> None:
         assessment, execution = _seed(db, auth, rows=2)
         uploads = _Uploads()
         bag = _bag(
-            stage_errors={ApiStage.ASSESSMENT.value: {"1": "rate limit exceeded"}}
+            stage_errors={PipelineStage.ASSESSMENT.value: {"1": "rate limit exceeded"}}
         )
 
         with _storage_patch(), _upload_patch(uploads):
@@ -204,7 +204,7 @@ def test_row_errors_are_flattened_per_stage(self, db) -> None:
         assert uploads.rows == [
             {
                 "type": "row_error",
-                "stage": ApiStage.ASSESSMENT.value,
+                "stage": PipelineStage.ASSESSMENT.value,
                 "row_index": 1,
                 "error": "rate limit exceeded",
             }
@@ -236,7 +236,7 @@ def test_openai_error_file_lines_become_rows(self, db) -> None:
                 session=db,
                 execution=execution,
                 assessment=assessment,
-                bag=_bag(stage_batches={ApiStage.ASSESSMENT.value: job.id}),
+                bag=_bag(stage_batches={PipelineStage.ASSESSMENT.value: job.id}),
                 failure_message=None,
             )
         assert [row["type"] for row in uploads.rows] == [
@@ -270,7 +270,7 @@ def test_unreadable_error_file_degrades_to_one_row(self, db) -> None:
                 session=db,
                 execution=execution,
                 assessment=assessment,
-                bag=_bag(stage_batches={ApiStage.ASSESSMENT.value: job.id}),
+                bag=_bag(stage_batches={PipelineStage.ASSESSMENT.value: job.id}),
                 failure_message=None,
             )
         row = uploads.rows[0]
@@ -289,7 +289,7 @@ def test_batch_without_an_error_file_contributes_nothing(self, db) -> None:
                 session=db,
                 execution=execution,
                 assessment=assessment,
-                bag=_bag(stage_batches={ApiStage.ASSESSMENT.value: job.id}),
+                bag=_bag(stage_batches={PipelineStage.ASSESSMENT.value: job.id}),
                 failure_message=None,
             )
 
@@ -337,7 +337,7 @@ def test_pre_provider_failure_records_only_a_synthetic_execution_error(
         assert uploads.rows == [
             {
                 "type": "execution_error",
-                "stage": ApiStage.ASSESSMENT.value,
+                "stage": PipelineStage.ASSESSMENT.value,
                 "error": "Vertex model gemini-2.5-pro not found",
             }
         ]
@@ -350,7 +350,7 @@ def test_completed_run_with_no_errors_carries_only_the_results_record(
         record_stage_dump(
             session=db,
             assessment=assessment,
-            stage=ApiStage.ASSESSMENT.value,
+            stage=PipelineStage.ASSESSMENT.value,
             url="s3://bucket/batch-1173/output.jsonl",
         )
         uploads = _Uploads()
@@ -362,7 +362,7 @@ def test_completed_run_with_no_errors_carries_only_the_results_record(
                 assessment=assessment,
                 bag=_bag(
                     stage_output_urls={
-                        ApiStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl"
+                        PipelineStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl"
                     }
                 ),
             )
@@ -378,16 +378,20 @@ def test_prefilter_and_assessment_dumps_coexist(self, db) -> None:
         uploads = _Uploads()
         bag = _bag(
             stage_output_urls={
-                ApiStage.TOPIC_RELEVANCE.value: "s3://bucket/batch-1170/output.jsonl",
-                ApiStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl",
+                PipelineStage.TOPIC_RELEVANCE.value: "s3://bucket/batch-1170/output.jsonl",
+                PipelineStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl",
             },
             counters={
-                ApiStage.TOPIC_RELEVANCE.value: {
+                PipelineStage.TOPIC_RELEVANCE.value: {
                     "total": 2,
                     "passed": 1,
                     "rejected": 1,
                 },
-                ApiStage.ASSESSMENT.value: {"total": 1, "passed": 1, "rejected": 0},
+                PipelineStage.ASSESSMENT.value: {
+                    "total": 1,
+                    "passed": 1,
+                    "rejected": 0,
+                },
             },
         )
 
@@ -408,7 +412,7 @@ def test_a_second_tick_does_not_duplicate_or_lose_records(self, db) -> None:
         uploads = _Uploads()
         bag = _bag(
             stage_output_urls={
-                ApiStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl"
+                PipelineStage.ASSESSMENT.value: "s3://bucket/batch-1173/output.jsonl"
             }
         )
 
@@ -441,7 +445,7 @@ def test_upload_failure_does_not_raise_into_the_terminal_path(self, db) -> None:
                 assessment=assessment,
                 bag=_bag(
                     stage_output_urls={
-                        ApiStage.ASSESSMENT.value: "s3://bucket/out.jsonl"
+                        PipelineStage.ASSESSMENT.value: "s3://bucket/out.jsonl"
                     }
                 ),
                 failure_message="kaboom",
diff --git a/backend/app/tests/celery/test_job_execution.py b/backend/app/tests/celery/test_job_execution.py
index 74484e3f6..441be7da6 100644
--- a/backend/app/tests/celery/test_job_execution.py
+++ b/backend/app/tests/celery/test_job_execution.py
@@ -59,6 +59,21 @@ def test_no_reenqueue_when_stage_not_requeued(self):
         apply_async.assert_not_called()
 
 
+class TestRunAssessmentResponse:
+    def test_runs_the_service_for_the_execution(self):
+        received: list[int] = []
+        with (
+            patch.object(job_execution, "_set_trace", lambda trace_id: None),
+            patch(
+                "app.services.assessment.api.response.run_response_assessment",
+                lambda *, execution_id: received.append(execution_id),
+            ),
+        ):
+            job_execution.run_assessment_response(execution_id=7, trace_id="trace-1")
+
+        assert received == [7]
+
+
 class TestRunChatbotSessionMessage:
     def test_hands_service_a_uuid_session_id(self):
         session_id = uuid4()
diff --git a/docs/wiki/modules/assessment.md b/docs/wiki/modules/assessment.md
index 655beb110..3efb48ada 100644
--- a/docs/wiki/modules/assessment.md
+++ b/docs/wiki/modules/assessment.md
@@ -1,27 +1,27 @@
 # Module: Assessment
 
-Assessments of datasets/inputs against ASSESSMENT-tagged configs. Three methods, chosen by input shape: **RESPONSE** (object input → `job` → llm_call/llm_chain), **BATCH** (list input → assessment_run → batch_job), **RUN** (legacy UI pipeline). No deep-dive doc yet.
+Assessments of datasets/inputs against ASSESSMENT-tagged configs. Three methods, chosen by input shape: **RESPONSE** (row object input → `assessment` + `assessment_run` rows, direct provider calls in one Celery task, result by callback), **BATCH** (list input → assessment_run → batch_job), **RUN** (legacy UI pipeline). No deep-dive doc yet.
 
 All paths relative to `backend/app/`.
 
 ## Routes
 - `api/routes/assessment/` — legacy UI (assessments, runs, datasets); RUN only, and the results export 422s a BATCH assessment (its rows come from `GET /assessments/{id}`)
-- `api/routes/assessment/api.py` — API-client routes, mounted top-level at `/assessments`: `POST` (submit; method inferred from input shape, BATCH wired, RESPONSE returns 501), `GET` (list, **BATCH only** so the execution join yields one row per assessment; `config_id` + optional `version` filters), `GET /{assessment_id}` (detail/poll; `include_input=true` echoes the submitted rows at the cost of a storage read, off by default). `callback_url` is optional and adds a webhook push on top of polling.
+- `api/routes/assessment/api.py` — API-client routes, mounted top-level at `/assessments`: `POST` (submit; method inferred from input shape: a `data` list or `submission_doc_id` ⇒ BATCH, a `data` object ⇒ RESPONSE, which requires `callback_url`), `GET` (list, **BATCH only** so the execution join yields one row per assessment; `config_id` + optional `version` filters), `GET /{assessment_id}` (detail/poll; `include_input=true` echoes the submitted rows at the cost of a storage read, off by default). `callback_url` is optional and adds a webhook push on top of polling.
 
 ## Tables (SQLModel)
 `models/assessment/` is a package, split by surface: `assessment.py` holds the shared DB tables + `AssessmentStatus`/`AssessmentMethod` enums + `AssessmentConfigRef` + the legacy RUN (UI) models; `assessment_api.py` holds the API-client request/response models; `submission.py` holds the uploaded-submission table and its response models. All are re-exported from the package `__init__`, so `from app.models.assessment import X` resolves either.
 
 | Table | Model |
 |---|---|
-| `assessment` (Assessment; parent — `method`, data source, `submission_input`, `result_files`; FK → job (RESPONSE), assessment_submission (RUN), org, project) | `models/assessment/assessment.py` |
-| `assessment_run` (AssessmentRun; child — one config execution, BATCH/RUN; FK → assessment, config, batch_job) | `models/assessment/assessment.py` |
+| `assessment` (Assessment; parent — `method`, data source, `submission_input`, `result_files`; assessment_submission (RUN), org, project) | `models/assessment/assessment.py` |
+| `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, `{attachments}`) / `BatchInput` (BATCH) no longer carry `query`; they discriminate structurally (a `data` or `submission_doc_id` key ⇒ BATCH, else RESPONSE) with `extra=forbid` keeping them disjoint and 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 copied into that run's own `submission.jsonl`, so downstream stays one code path and the run keeps an immutable snapshot if the submission file is later replaced. `assessment.submission_id` records which submission it came from, so the provenance survives the copy 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: <one row>}`) / `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`).
 
 `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.
 
-`assessment.result_files` is the durable record of every provider batch dump held: JSONB, **NOT NULL default `{}`**, CHECK-constrained to a JSON object (`ck_assessment_result_files_is_object`), keyed by stage name (`assessment` / `topic_relevance`, the `ApiStage` values) plus `errors`, with `{object_store_url}` per key. It stores the raw `s3://` url (presigning is per-delivery, so the column never holds an expiring link); writes go through `crud/assessment/api.py::set_result_files`, which merges **server-side** (`result_files || :files::jsonb`) rather than read-modify-write, because two drivers can touch the row in the same second. `batch_job.provider_error_file_id` persists OpenAI's `error_file_id` (OpenAI only; Anthropic and Gemini report per-item errors inline) so the error dump is still fetchable at terminal time, in a later task than the poll that surfaced it.
+`assessment.result_files` is the durable record of every provider batch dump held: JSONB, **NOT NULL default `{}`**, CHECK-constrained to a JSON object (`ck_assessment_result_files_is_object`), keyed by stage name (`assessment` / `topic_relevance`, the `PipelineStage` values) plus `errors`, with `{object_store_url}` per key. It stores the raw `s3://` url (presigning is per-delivery, so the column never holds an expiring link); writes go through `crud/assessment/api.py::set_result_files`, which merges **server-side** (`result_files || :files::jsonb`) rather than read-modify-write, because two drivers can touch the row in the same second. `batch_job.provider_error_file_id` persists OpenAI's `error_file_id` (OpenAI only; Anthropic and Gemini report per-item errors inline) so the error dump is still fetchable at terminal time, in a later task than the poll that surfaced it.
 
 Delivery is poll-or-push: the `POST /assessments` ack is the flat `AssessmentSubmitResponse {assessment_id, status, message, inserted_at, updated_at}`, the result is always readable from `GET /assessments/{assessment_id}`, and when the request carried a `callback_url` the `AssessmentCallback {assessment_id, status, data, request_metadata}` is also POSTed there on completion — where `data` is a single `AssessmentResult` (RESPONSE) or an `AssessmentBatchResult {total_items, counts, items}` (BATCH); `status` lives on the envelope only. The outer `send_callback` envelope carries `{success, data, error, metadata}`: `error` is `_fail`'s failure message (null on a success path) and `metadata` is `{files: AssessmentResultFiles}` — a flat `{<stage>: url | null, errors: url | null}` map, so a client iterates it without branching on shape; every key is always present (null when the run produced no such file) — the column's `object_store_url` presigned for 1 day at delivery time, never stored, so a client whose receiver rejects the (large, item-inlining) body can still fetch the dumps. Both were hardcoded `None` before. Delivery is one inline attempt with no retry, so a rejected callback is still lost. Pre-filter `stop_on_fail` flag (`config/assessment_blob.py`) drives which filters hard-stop the chain on a failing verdict vs pass-through (record only).
 
@@ -38,6 +38,13 @@ Delivery is poll-or-push: the `POST /assessments` ack is the flat `AssessmentSub
 - `crud/assessment/api.py` — new API-client crud (method-based Assessment/AssessmentRun writes): `create_assessment` (takes a pre-generated `assessment_id` + `submission_input`, so submit uploads `submission.jsonl` first and then does one insert — no FAILED orphan when the upload fails), `set_assessment_job`, `create_execution`, `set_execution_batch_job`, `save_execution_state`, `set_result_files`, `update_status`, `list_executions`, `list_assessments_with_execution` (BATCH only; the detail route reads through the legacy `core.get_assessment_by_id`). Namespaced under `api` (`from app.crud.assessment import api`) to avoid colliding with the legacy `create_assessment`.
 - `crud/assessment/{core,cron,processing,batch}.py` — legacy RUN pipeline crud. RUN runtime for the dropped columns (`stage`/`stage_status`/`pipeline`/`stage_batches`/`prefilter_total_*`/`object_store_url`) now lives in the `assessment_run.execution` bag via `core._read_exec`/`_write_exec`; `status` is the `AssessmentStatus` enum; the RUN input binding lives on the parent `assessment.input`. `batch.py` builds per-row prompts (`{column}` substitution from the parent `InputBinding.prompt`).
 
+## RESPONSE
+`submission.submit` validates the row exactly like a BATCH row, then inserts one `assessment` row (`method=RESPONSE`, `input` = the submitted `{data: row}`) and one `assessment_run` row (`config_id` + `config_version` used, `total_items=1`, `execution` state bag seeded by the same `_initial_state` BATCH uses) and enqueues `celery/tasks/job_execution.py::run_assessment_response(execution_id)`. No `assessment_submission`, no `submission.jsonl`, no `job`, no stored results file.
+
+`services/assessment/api/response.py::run_response_assessment` runs `build_pipeline` (gate pre-filters, then pass-through, then the assessment) as one synchronous provider call per stage: `batch.build_stage_requests` builds the same provider-wire request the batch path sends, the call goes through the llm-call client from `get_llm_provider(...).client` (OpenAI / Anthropic / Gemini; Vertex via `.genai`), and `batch.parse_one` / `parse_verdict` / `parsing.parse_assessment_text` read the result. A failed gate verdict skips the assessment. The state bag is saved at each stage boundary: `stage` / `stage_status`, `verdicts`, `stage_errors` (`{stage: {"0": error}}`), and at the end the `result` exactly as delivered, so a lost callback loses nothing.
+
+Errors: a failed pre-filter call **fails open as in BATCH** (`verdict: true`, the assessment still runs) but is reported: `PreFilterVerdict.error` is set in the callback body, recorded in `stage_errors`, and the run ends `COMPLETED_WITH_ERRORS` (`success` stays true). An assessment provider error, or an unrecoverable failure (config version gone, credential missing), ends `FAILED`; the failed stage is `execution.stage` with `stage_status: FAILED` (a failure before any stage started leaves it `PENDING`), and `assessment_run.error_message` carries the stage-prefixed text. The callback is sent with `callbacks.deliver` (single `AssessmentResult`, or `data=null` plus the message on an unrecoverable failure); the shared sender retries up to 4 times with backoff. The task has no autoretry (a retry would re-bill the provider and re-send the callback), so a lost worker leaves the rows `PROCESSING`. `GET /assessments` and `GET /assessments/{id}` stay BATCH-only and `callback_url` is required for RESPONSE.
+
 ## Async
 - Rides shared `core/batch/` provider batch infra + cron polling (same as evaluations).
 - **Two drivers, strictly separated:** the cron poller (`crud/assessment/cron.py::poll_all_pending_assessment_evaluations`) selects `method == RUN` only and owns the legacy pipeline; the API-client BATCH path is driven solely by the Celery self-re-enqueue. The poller used to pick up BATCH runs too and crash on their list-shaped `pipeline`, leaving the run un-finalised and the callback unfired. Its deterministic-error branch (`DETERMINISTIC_ERRORS`) marks a run FAILED on a programming error instead of logging "will retry" forever; bare `Exception` stays the transient path.