Skip processed artifacts when reprocessing a run

developer-docs/lambda.md

@@ -18,13 +18,14 @@ Slack notifications are sent by the **web app**, not the Lambda — see [Slack n
 
 ### Function URL (manual reprocessing)
 
-When a file fails processing (or needs to be re-run), users can trigger reprocessing from the run detail page in the web app. This invokes the Lambda's Function URL instead of going through S3:
+When a raw file fails processing (or needs to be re-run), users can trigger reprocessing from the run detail page in the web app. Processed artifacts (preview CSVs, posters, MP4s) are not eligible — they often share extensions with the watcher's raw patterns, and Lambda only ingests the original instrument files. This invokes the Lambda's Function URL instead of going through S3:
 
-1. The user clicks **Reprocess** on an uploaded, failed, or completed file in the web dashboard.
+1. The user clicks **Reprocess** on an uploaded, failed, or completed raw file in the web dashboard, or **Reprocess** on the run (which queues every eligible raw file).
 2. The web app's `POST /api/v1/files/:fileId/reprocess` endpoint transitions the file to `processing` status, clears any previous error, and sends a POST request to the Lambda Function URL.
 3. The Function URL is configured with `AuthType: AWS_IAM`, so the web app SigV4-signs the request using credentials it gets via Vercel OIDC federation (the `WebAppS3Role` IAM role, which has `lambda:InvokeFunctionUrl` on this function's ARN). The body is a JSON payload containing a synthetic S3 event.
 4. The Lambda handler detects the Function URL invocation (via `requestContext` in the event) and parses the S3 event from the request body. Inbound auth is enforced by AWS itself in front of the function — the handler never sees an unauthenticated request.
-5. From here, processing follows the same type-based dispatch as the S3 trigger path, except **filename gates are skipped** — a user clicking Reprocess has stated intent, so the handler must not leave the file stranded in `processing`.
+5. From here, processing follows the same type-based dispatch as the S3 trigger path, except the cheap **union gate is skipped** so the handler always reaches the instrument lookup.
+6. If the matched processor's own filename gate rejects the file, the handler PATCHes it to `failed` with `The {instrument_type} processor does not handle '{filename}'` instead of returning quietly. The web app moved the row to `processing` before invoking, and processors return without touching the row when the name isn't theirs — so a silent no-op would strand it there forever. The same recovery covers a missing instrument and an unmapped `instrument_type`.
 
 ### Function URL (archive build)
 
@@ -41,7 +42,7 @@ See [Run archives](run-archives.md) for the full flow, S3 bucket layout, cache s
 
 Dispatch is by `instrument_type` (Postgres/TS enum), not instrument ID. The registry lives in `lambda/src/data_hub_lambda/processors.py`; the web reprocess gate mirrors the same keys in `web/lib/instruments/processable-types.ts`.
 
-| `instrument_type` | Module | Filename gate (S3 events only) |
+| `instrument_type` | Module | Filename gate |
 | --- | --- | --- |
 | `tape_station` | `agilent_4150_tapestation` | `.pdf` |
 | `fplc` | `akta_fplc` | `.pdf` |

lambda/src/data_hub_lambda/handler.py

@@ -309,15 +309,16 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] |
     """Top-level Lambda handler dispatching to instrument workflows.
 
     Dispatch is by ``instrument_type`` (fetched from the API), not instrument
-    ID. Filename gates filter the S3 firehose; explicit reprocess via the
-    Function URL bypasses those gates so user-initiated work can't strand
-    a file in ``processing``.
+    ID. Filename gates filter the S3 firehose, which delivers every upload.
+    A reprocess via the Function URL skips the cheap union gate so it always
+    reaches the instrument lookup, then fails the file if the processor's own
+    gate rejects it — the web app already moved the row to ``processing``, so
+    a silent return would strand it.
     """
     logger.info("Received event: %s", pformat(event))
 
-    # Capture before unwrapping so reprocess (Function URL) can skip gates.
+    # Capture before unwrapping: the payload loses the Function URL envelope.
     is_function_url = _is_function_url_event(event)
-    apply_filename_gates = not is_function_url
 
     # Function URL invocations carry a requestContext with an http key.
     # AWS_IAM auth is enforced by Lambda before the handler runs, so we
@@ -354,7 +355,8 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] |
 
     try:
         # Skip the API call when no processor could possibly want this filename.
-        if apply_filename_gates and not matches_any_processor_gate(filename):
+        # Reprocess skips this so it can name the instrument type in the error.
+        if not is_function_url and not matches_any_processor_gate(filename):
             logger.info(
                 "No processor filename gate matches %s; skipping.",
                 filename,
@@ -410,12 +412,17 @@ def lambda_handler(event: dict[str, Any], context: Context) -> dict[str, Any] |
                 _fail_reprocess_file(instrument_id, run_id, filename, message)
             return None
 
-        if apply_filename_gates and not processor.matches_filename(filename):
+        if not processor.matches_filename(filename):
+            message = f"The {instrument.instrument_type} processor does not handle '{filename}'"
             logger.info(
                 "Filename %s does not match gate for instrument_type=%s; skipping.",
                 filename,
                 instrument.instrument_type,
             )
+            # Processors return without touching the file row when the name
+            # isn't theirs, so a reprocess has to report the mismatch itself.
+            if is_function_url:
+                _fail_reprocess_file(instrument_id, run_id, filename, message)
             return None
 
         logger.info(

lambda/tests/test_processors.py

@@ -340,14 +340,16 @@ def test_403_raises(self) -> None:
 
         assert exc_info.value.status_code == 403
 
-    def test_reprocess_bypasses_filename_gate(self) -> None:
+    def test_reprocess_marks_failed_when_filename_unsupported(self) -> None:
+        """A reprocess the processor would ignore must not stay in `processing`."""
         client = MagicMock()
         client.get_instrument.return_value = InstrumentResponse(
             id="azure-cielo-qpcr",
             display_name="Azure Cielo qPCR",
             status="active",
             instrument_type="qpcr",
         )
+        client.create_file.return_value = self._file_response(11)
         process_file = MagicMock()
         entry = MagicMock()
         entry.matches_filename.return_value = False
@@ -360,11 +362,16 @@ def test_reprocess_bypasses_filename_gate(self) -> None:
                 "data_hub_lambda.handler.get_processor",
                 return_value=entry,
             ),
-            # Union gate would also block S3 events; reprocess must skip it.
+            # Union gate would short-circuit an S3 event before the API call;
+            # reprocess must skip it so the error can name the type.
             patch(
                 "data_hub_lambda.handler.matches_any_processor_gate",
                 return_value=False,
             ),
+            patch(
+                "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET",
+                "test-bucket",
+            ),
         ):
             result = lambda_handler(
                 self._function_url_event(
@@ -375,12 +382,51 @@ def test_reprocess_bypasses_filename_gate(self) -> None:
                 MagicMock(),
             )
 
+        assert result is None
+        client.get_instrument.assert_called_once_with("azure-cielo-qpcr")
+        process_file.assert_not_called()
+        client.update_file.assert_called_once_with(
+            11,
+            status="failed",
+            error_message=(
+                "The qpcr processor does not handle 'Experiment_Amplification Results.csv'"
+            ),
+        )
+
+    def test_reprocess_runs_when_filename_matches(self) -> None:
+        client = MagicMock()
+        client.get_instrument.return_value = InstrumentResponse(
+            id="azure-cielo-qpcr",
+            display_name="Azure Cielo qPCR",
+            status="active",
+            instrument_type="qpcr",
+        )
+        process_file = MagicMock()
+        entry = MagicMock()
+        entry.matches_filename.return_value = True
+        entry.process_file = process_file
+
+        with (
+            patch("data_hub_lambda.handler.get_client", return_value=client),
+            patch("data_hub_lambda.handler._cleanup_tmp"),
+            patch("data_hub_lambda.handler.get_processor", return_value=entry),
+        ):
+            result = lambda_handler(
+                self._function_url_event(
+                    "azure-cielo-qpcr",
+                    "run-1",
+                    "Experiment_Cq Values.csv",
+                ),
+                MagicMock(),
+            )
+
         assert result is None
         process_file.assert_called_once_with(
             "azure-cielo-qpcr",
             "run-1",
-            "Experiment_Amplification Results.csv",
+            "Experiment_Cq Values.csv",
         )
+        client.update_file.assert_not_called()
 
     def test_s3_event_applies_per_type_gate(self) -> None:
         client = MagicMock()

web/app/api/v1/files/[fileId]/reprocess/route.ts

@@ -14,9 +14,9 @@ interface RouteContext {
 // ---------------------------------------------------------------------------
 // POST /api/v1/files/:fileId/reprocess
 //
-// Transitions an uploaded, failed, or completed file back to "processing"
-// and invokes the Lambda Function URL to re-run the instrument's
-// process_file workflow.
+// Transitions an uploaded, failed, or completed raw file back to
+// "processing" and invokes the Lambda Function URL to re-run the
+// instrument's process_file workflow. Processed artifacts are rejected.
 // The core logic lives in lib/api/file-reprocessing.ts so the MCP server
 // can reuse it.
 // ---------------------------------------------------------------------------

web/app/api/v1/instruments/[instrumentId]/runs/[runId]/reprocess/route.ts

@@ -11,8 +11,8 @@ interface RouteContext {
 // POST /api/v1/instruments/:instrumentId/runs/:runId/reprocess
 //
 // Run-level convenience endpoint that reprocesses every `uploaded`,
-// `completed`, or `failed` file on the run. Used by the runs list
-// row/bulk actions.
+// `completed`, or `failed` raw file on the run. Processed artifacts
+// are skipped. Used by the runs list row/bulk actions.
 // ---------------------------------------------------------------------------
 
 export async function POST(request: NextRequest, { params }: RouteContext) {

web/components/runs/file-selection-provider.tsx

@@ -3,8 +3,7 @@
 import { createContext, use, useCallback, useMemo, useState } from "react";
 import type { RunFile } from "@/lib/api/instrument-runs";
 import type { InstrumentType } from "@/lib/db/schema";
-import { isProcessableInstrumentType } from "@/lib/instruments/processable-types";
-import { REPROCESSABLE_STATUSES } from "@/lib/runs/reprocessable-statuses";
+import { canReprocessFile } from "@/lib/runs/reprocessable-statuses";
 
 // ---------------------------------------------------------------------------
 // File selection provider for the run files table. Mirrors RunSelectionProvider
@@ -21,8 +20,6 @@ const DOWNLOADABLE_STATUSES = new Set([
   "failed",
 ]);
 
-const REPROCESSABLE_STATUS_SET = new Set<string>(REPROCESSABLE_STATUSES);
-
 export interface FileCaps {
   dismiss: boolean;
   download: boolean;
@@ -48,10 +45,7 @@ export function buildFileRef(
   }
   const isDetected = file.status === "detected";
   const canDownload = DOWNLOADABLE_STATUSES.has(file.status);
-  const canReprocess =
-    isProcessableInstrumentType(instrumentType) &&
-    REPROCESSABLE_STATUS_SET.has(file.status) &&
-    file.s3Key !== null;
+  const canReprocess = canReprocessFile(file, instrumentType);
   if (!(isDetected || canDownload)) {
     return null;
   }

web/components/runs/run-files-table.tsx

@@ -25,8 +25,7 @@ import {
 import type { RunFile } from "@/lib/api/instrument-runs";
 import { formatDateTime } from "@/lib/date";
 import type { InstrumentType } from "@/lib/db/schema";
-import { isProcessableInstrumentType } from "@/lib/instruments/processable-types";
-import { REPROCESSABLE_STATUSES } from "@/lib/runs/reprocessable-statuses";
+import { canReprocessFile } from "@/lib/runs/reprocessable-statuses";
 import { cn, formatBytes } from "@/lib/utils";
 import {
   FileSelectAllCheckbox,
@@ -44,8 +43,6 @@ const DOWNLOADABLE_STATUSES = new Set([
   "failed",
 ]);
 
-const REPROCESSABLE_STATUS_SET = new Set<string>(REPROCESSABLE_STATUSES);
-
 const CATEGORY_BADGE_CLASSES: Record<string, string> = {
   raw: "bg-blue-100 text-blue-700 dark:bg-blue-950 dark:text-blue-300",
   processed:
@@ -215,19 +212,10 @@ function UploadDismissActions({
   );
 }
 
-function canReprocess(file: RunFile, instrumentType: InstrumentType): boolean {
-  return (
-    file.deletedAt === null &&
-    isProcessableInstrumentType(instrumentType) &&
-    REPROCESSABLE_STATUS_SET.has(file.status) &&
-    file.s3Key !== null
-  );
-}
-
 // ---------------------------------------------------------------------------
 // Read-only variant: no selection column, no upload/dismiss. Reprocessing is
-// still allowed for uploaded/completed/failed files so operators can recover
-// report data or kick stuck uploads without restoring the run.
+// still allowed for uploaded/completed/failed raw files so operators can
+// recover report data or kick stuck uploads without restoring the run.
 // ---------------------------------------------------------------------------
 
 export interface ReadOnlyRunFilesTableProps {
@@ -261,7 +249,7 @@ export function ReadOnlyRunFilesTable({
             >
               <FileInfoCells file={file} />
               <TableCell className="py-2 pr-3">
-                {canReprocess(file, instrumentType) ? (
+                {canReprocessFile(file, instrumentType) ? (
                   <div className="flex items-center gap-1 opacity-0 transition-opacity focus-within:opacity-100 group-hover:opacity-100">
                     <ReprocessAction
                       file={file}
@@ -342,7 +330,7 @@ export function EditableRunFilesTable({
           const ref = refsByFileId.get(file.id) ?? null;
           const isSelected = ref ? meta.isSelected(ref.id) : false;
           const canDoUploadDismiss = !isDismissed && file.status === "detected";
-          const canDoReprocess = canReprocess(file, instrumentType);
+          const canDoReprocess = canReprocessFile(file, instrumentType);
 
           // Reveal classes: hide per-row actions while a bulk selection is
           // active (the bar is the single entry point) but keep the JSX

web/lib/api/file-reprocessing.ts

@@ -56,6 +56,19 @@ export async function reprocessFile(fileId: number): Promise<ReprocessResult> {
     };
   }
 
+  // Processed artifacts often share watcher extensions. Lambda only
+  // ingests raw inputs; queuing them marks them processing then strands
+  // them when the processor ignores the filename.
+  if (file.category !== "raw") {
+    return {
+      ok: false,
+      status: 409,
+      code: "CONFLICT",
+      message:
+        "Cannot reprocess a processed artifact — only raw files can be reprocessed",
+    };
+  }
+
   if (!(REPROCESSABLE_STATUSES as readonly string[]).includes(file.status)) {
     return {
       ok: false,
@@ -226,6 +239,7 @@ export async function reprocessRun(
     .where(
       and(
         eq(files.instrumentRunId, run.id),
+        eq(files.category, "raw"),
         inArray(files.status, [...REPROCESSABLE_STATUSES]),
         isNull(files.deletedAt)
       )

web/lib/api/openapi/paths/files.ts

@@ -86,7 +86,7 @@ registry.registerPath({
   operationId: "reprocessFile",
   summary: "Reprocess a file",
   description:
-    "Requires scope `files:reprocess`. Eligible statuses: `uploaded`, `failed`, or `completed` (file must have an S3 location). The file's instrument must have a Lambda processor.",
+    "Requires scope `files:reprocess`. Raw files only. Eligible statuses: `uploaded`, `failed`, or `completed` (file must have an S3 location). The file's instrument must have a Lambda processor.",
   tags: ["Files"],
   security: bearerSecurity,
   request: { params: fileParams },

web/lib/api/openapi/paths/runs.ts

@@ -141,7 +141,7 @@ registry.registerPath({
   path: "/instruments/{instrumentId}/runs/{runId}/reprocess",
   operationId: "reprocessInstrumentRun",
   summary: "Reprocess a run",
-  description: `${scoped("runs:reprocess")} Reprocesses every \`uploaded\`, \`failed\`, or \`completed\` file on the run. The instrument must have a Lambda processor.`,
+  description: `${scoped("runs:reprocess")} Reprocesses every \`uploaded\`, \`failed\`, or \`completed\` raw file on the run. Processed artifacts are skipped. The instrument must have a Lambda processor.`,
   tags: tag,
   security: bearerSecurity,
   request: { params: runParams },

web/lib/mcp/tools/files.defs.ts

@@ -51,7 +51,7 @@ export const reprocessFileTool = {
   name: "reprocess_file",
   title: "Reprocess File",
   description:
-    "Re-run the Lambda processing workflow for an uploaded, failed, or completed file on an instrument that has a Lambda processor. Transitions the file back to 'processing'. Use this to retry after a parser fix, transient Lambda failure, or a stuck upload that never entered processing.",
+    "Re-run the Lambda processing workflow for an uploaded, failed, or completed raw file on an instrument that has a Lambda processor. Processed artifacts are rejected. Transitions the file back to 'processing'. Use this to retry after a parser fix, transient Lambda failure, or a stuck upload that never entered processing.",
   group: "files",
   scope: "files:reprocess",
   inputSchema: { fileId: z.number().int().describe("Numeric file ID") },

web/lib/mcp/tools/runs.defs.ts

@@ -317,7 +317,7 @@ export const reprocessRunTool = {
   scope: "runs:reprocess",
   title: "Reprocess Run",
   description:
-    "Re-run Lambda processing for every uploaded, completed, or failed file on a run. The instrument must have a Lambda processor. Prefer this over looping reprocess_file for bulk retries after a parser fix or to kick stuck uploads.",
+    "Re-run Lambda processing for every uploaded, completed, or failed raw file on a run. Processed artifacts are skipped. The instrument must have a Lambda processor. Prefer this over looping reprocess_file for bulk retries after a parser fix or to kick stuck uploads.",
   inputSchema: {
     instrumentId: z.string().describe("Instrument identifier"),
     runId: z.string().describe("Run identifier within the instrument"),

web/lib/runs/reprocessable-statuses.ts

@@ -1,3 +1,6 @@
+import type { files, InstrumentType } from "@/lib/db/schema";
+import { isProcessableInstrumentType } from "@/lib/instruments/processable-types";
+
 // Statuses eligible for POST /files/:id/reprocess (and run-level reprocess).
 // Includes `uploaded` so stuck S3 uploads can be kicked when the Lambda
 // trigger never fired. Client-safe — imported by UI tables and the API.
@@ -6,3 +9,25 @@ export const REPROCESSABLE_STATUSES = [
   "failed",
   "completed",
 ] as const;
+
+const REPROCESSABLE_STATUS_SET = new Set<string>(REPROCESSABLE_STATUSES);
+
+// Only the columns the predicate reads, taken from the `files` row so a typo
+// in a category or status literal fails to compile.
+export type ReprocessableFile = Pick<
+  typeof files.$inferSelect,
+  "category" | "deletedAt" | "s3Key" | "status"
+>;
+
+export function canReprocessFile(
+  file: ReprocessableFile,
+  instrumentType: InstrumentType
+): boolean {
+  return (
+    file.category === "raw" &&
+    file.deletedAt === null &&
+    file.s3Key !== null &&
+    REPROCESSABLE_STATUS_SET.has(file.status) &&
+    isProcessableInstrumentType(instrumentType)
+  );
+}

web/lib/runs/row-actions.ts

@@ -14,8 +14,8 @@ import { isProcessableInstrumentType } from "@/lib/instruments/processable-types
 // - Upload:    at least one file still waiting to be uploaded.
 // - Download:  at least one file has made it to S3 (i.e. is not `detected`
 //              or `upload_requested`).
-// - Reprocess: instrument_type has a Lambda processor and at least one file
-//              is in `uploaded`, `completed`, or `failed`.
+// - Reprocess: instrument_type has a Lambda processor and at least one raw
+//              file is in `uploaded`, `completed`, or `failed`.
 // - Delete:    the run isn't already soft-deleted.
 // ---------------------------------------------------------------------------