Assessment: Implement L1 Filtering and Post-Processing

backend/app/alembic/versions/064_add_prefilter_columns_to_assessment_run.py

@@ -0,0 +1,105 @@
+"""Add prefilter columns and pipeline stage-machine columns to assessment_run
+
+Revision ID: 064
+Revises: 063
+Create Date: 2026-05-27 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "064"
+down_revision = "063"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "prefilter_object_store_url",
+            sa.String(),
+            nullable=True,
+            comment="S3 URL of prefilter results JSON",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "prefilter_total_rows",
+            sa.Integer(),
+            nullable=True,
+            comment="Total rows fed into the prefilter stages",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "prefilter_total_passed",
+            sa.Integer(),
+            nullable=True,
+            comment="Rows that passed the go/no-go gates and went to L2",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "prefilter_total_rejected",
+            sa.Integer(),
+            nullable=True,
+            comment="Rows rejected by a go/no-go gate",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "stage",
+            sa.String(),
+            nullable=True,
+            comment=(
+                "Current pipeline stage: PRE_FILTER_TOPIC_RELEVANCE, "
+                "PRE_FILTER_DUPLICATE_DETECTION, L2_ASSESSMENT, COMPLETED, FAILED"
+            ),
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "stage_status",
+            sa.String(),
+            nullable=True,
+            comment="Status of stage: PENDING, PROCESSING, COMPLETED, FAILED",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "pipeline",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment="Ordered stage config driving execution: {'stages': [...]}",
+        ),
+    )
+    op.add_column(
+        "assessment_run",
+        sa.Column(
+            "stage_batches",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment="Map of stage name -> batch_job id, for per-stage result lookup",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("assessment_run", "stage_batches")
+    op.drop_column("assessment_run", "pipeline")
+    op.drop_column("assessment_run", "stage_status")
+    op.drop_column("assessment_run", "stage")
+    op.drop_column("assessment_run", "prefilter_total_rejected")
+    op.drop_column("assessment_run", "prefilter_total_passed")
+    op.drop_column("assessment_run", "prefilter_total_rows")
+    op.drop_column("assessment_run", "prefilter_object_store_url")

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

@@ -0,0 +1,5 @@
+Resume a failed assessment run from its failed stage.
+
+Re-runs the same child run in place, starting at the stage that failed.
+Stages that already completed are reused (their batch results are not
+recomputed). Only valid when the run is in a failed state.

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

@@ -0,0 +1,15 @@
+Save post-processing config for a single assessment run.
+
+Stores the config inside the run's `input` JSON blob (key
+`post_processing_config`). It is applied at export/preview time and never
+re-runs the LLM, so it can be edited after the run completes.
+
+The config has three optional sections:
+
+- `computed_columns`: derived columns from formulas, e.g.
+  `{"name": "Total_Score", "formula": "@Novelty_score + @Usefulness_score"}`.
+  Formulas reference columns with `@` and support `+ - * /` and parentheses.
+- `filter`: row filters combined with AND logic.
+- `sort`: sort rules applied in priority order.
+
+Pass `null` (or an empty body) to clear post-processing for the run.

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

@@ -3,14 +3,19 @@
 import logging
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Body, Depends, HTTPException, Query
 from fastapi.responses import StreamingResponse
 
 from app.api.deps import AuthContextDep, SessionDep
 from app.api.permissions import Permission, require_permission
 from app.crud.assessment import (
     get_assessment_by_id,
+    update_run_post_processing_config,
+)
+from app.crud.assessment import (
     get_assessment_run_by_id as get_run_by_id,
+)
+from app.crud.assessment import (
     list_assessment_runs as list_runs,
 )
 from app.models.assessment import (
@@ -21,6 +26,9 @@
     AssessmentRunPublic,
 )
 from app.models.evaluation import EvaluationDataset
+from app.services.assessment.service import (
+    resume_assessment_run as resume_run,
+)
 from app.services.assessment.service import (
     retry_assessment_run as retry_run,
 )
@@ -33,6 +41,7 @@
     load_export_rows_for_run,
     sort_export_rows,
 )
+from app.services.assessment.utils.post_processing import apply_post_processing
 from app.utils import APIResponse, load_description
 
 logger = logging.getLogger(__name__)
@@ -65,6 +74,13 @@ def _build_run_public(
         total_items=run.total_items,
         error_message=run.error_message,
         input=run.input,
+        prefilter_total_rows=run.prefilter_total_rows,
+        prefilter_total_passed=run.prefilter_total_passed,
+        prefilter_total_rejected=run.prefilter_total_rejected,
+        stage=run.stage,
+        stage_status=run.stage_status,
+        pipeline=run.pipeline,
+        post_processing_config=(run.input or {}).get("post_processing_config"),
         inserted_at=run.inserted_at,
         updated_at=run.updated_at,
     )
@@ -127,6 +143,34 @@ def retry_assessment_run(
     return APIResponse.success_response(data=result)
 
 
+@router.post(
+    "/runs/{run_id}/resume",
+    description=load_description("assessment/resume_run.md"),
+    response_model=APIResponse[AssessmentResponse],
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+)
+def resume_assessment_run(
+    run_id: int,
+    session: SessionDep,
+    auth_context: AuthContextDep,
+) -> APIResponse[AssessmentResponse]:
+    """Resume a failed child run from its failed stage, reusing completed stages."""
+    run = get_run_by_id(
+        session=session,
+        run_id=run_id,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+
+    result = resume_run(
+        session=session,
+        run=run,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+    return APIResponse.success_response(data=result)
+
+
 @router.get(
     "/runs",
     description=load_description("assessment/list_runs.md"),
@@ -212,12 +256,44 @@ def export_assessment_run_results(
         )
     )
 
+    post_processing_config = (run.input or {}).get("post_processing_config") or None
     base_label = assessment.experiment_name if assessment else f"run_{run.id}"
+
     if export_format != "json":
         return build_export_response(
             export_rows=export_rows,
             export_format=export_format,
             base_name=f"{base_label}_run_{run.id}_results",
+            post_processing_config=post_processing_config,
         )
 
-    return APIResponse.success_response(data=build_json_export_rows(export_rows))
+    rows = build_json_export_rows(export_rows)
+    rows = apply_post_processing(rows, post_processing_config)
+    return APIResponse.success_response(data=rows)
+
+
+@router.patch(
+    "/runs/{run_id}/post-processing",
+    description=load_description("assessment/update_post_processing.md"),
+    response_model=APIResponse[AssessmentRunPublic],
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+)
+def update_post_processing(
+    run_id: int,
+    session: SessionDep,
+    auth_context: AuthContextDep,
+    config: dict[str, Any] | None = Body(default=None),
+) -> APIResponse[AssessmentRunPublic]:
+    """Save post-processing config (computed columns, sort, filter) for a run."""
+    run = get_run_by_id(
+        session=session,
+        run_id=run_id,
+        organization_id=auth_context.organization_.id,
+        project_id=auth_context.project_.id,
+    )
+    if run is None:
+        raise HTTPException(status_code=404, detail="Run not found")
+
+    run = update_run_post_processing_config(session=session, run=run, config=config)
+
+    return APIResponse.success_response(data=_build_run_public(session, run))

backend/app/celery/tasks/job_execution.py

@@ -232,6 +232,29 @@ def run_tts_batch_submission(
     )
 
 
+@celery_app.task(bind=True, queue="low_priority", priority=1)
+@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_assessment_pipeline")
+def run_assessment_pipeline(
+    self,
+    run_id: int,
+    organization_id: int,
+    project_id: int,
+    trace_id: str,
+    **kwargs,
+):
+    from app.services.assessment.tasks import execute_assessment_pipeline
+
+    _set_trace(trace_id)
+    return _run_with_otel_parent(
+        self,
+        lambda: execute_assessment_pipeline(
+            run_id=run_id,
+            organization_id=organization_id,
+            project_id=project_id,
+        ),
+    )
+
+
 @celery_app.task(bind=True, queue="low_priority", priority=1)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_tts_result_processing")
 def run_tts_result_processing(

backend/app/crud/assessment/__init__.py

@@ -13,7 +13,9 @@
     list_assessment_runs,
     list_assessments,
     recompute_assessment_status,
+    update_assessment_run_prefilter_stats,
     update_assessment_run_status,
+    update_run_post_processing_config,
 )
 from app.crud.assessment.dataset import (
     create_assessment_dataset,
@@ -42,5 +44,7 @@
     "list_assessment_datasets",
     "list_assessments",
     "recompute_assessment_status",
+    "update_assessment_run_prefilter_stats",
     "update_assessment_run_status",
+    "update_run_post_processing_config",
 ]