Replace run_report_data table with S3 CSV files for plate reader data

docs/architecture.md

@@ -9,6 +9,7 @@ flowchart LR
     W[Watcher] -->|raw files| S3[S3]
     S3 -->|trigger| L[Lambda]
     W -->|heartbeats, runs| API[API]
+    L -->|processed data| S3
     L -->|results| API
     API -->|presigned URLs| S3
     API --> DB[(PostgreSQL)]

docs/guides/adding-an-instrument.md

@@ -44,7 +44,7 @@ See the [watcher reference](../watcher.md) for details on upload modes, run dete
 
 If the new instrument needs automated preprocessing (metadata extraction, image processing, etc.), you'll need to add a processor to the Lambda function. This requires changes to the codebase.
 
-### 1. Register the instrument in the shared library
+### 4.1 Register the instrument in the shared library
 
 Add the instrument to `packages/shared/src/data_hub_shared/enums.py`:
 
@@ -63,7 +63,7 @@ INSTRUMENT_ID_TO_NAME_MAP: dict[str, str] = {
 }
 ```
 
-### 2. Create a processor module
+### 4.2 Create a processor module
 
 Create a new directory and `process_file.py` under `lambda/src/data_hub_lambda/`:
 
@@ -85,15 +85,16 @@ A typical processor:
 
 1. Gets an API client and the S3 bucket from config.
 2. Calls `client.ensure_run()` to create or find the run.
-3. Calls `client.create_file()` to register the file.
+3. Calls `client.create_file()` to register the raw file.
 4. Downloads the raw file from S3.
 5. Performs instrument-specific preprocessing (parsing, metadata extraction, etc.).
-6. Updates the file status to `completed`.
-7. Returns the web app URL for the run.
+6. Optionally uploads processed artifacts (CSV, images) to the S3 processed bucket and registers them via `client.create_file(..., category="processed")`. This is the pattern used by the SpectraMax plate reader (processed CSV) and Azure 600 Gel Doc (contrast-enhanced PNG).
+7. Updates the raw file status to `completed`.
+8. Returns the web app URL for the run.
 
-See any existing processor (e.g., `lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py`) for a complete example.
+See any existing processor (e.g., `lambda/src/data_hub_lambda/azure_cielo_qpcr/process_file.py` for simple metadata extraction, or `lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py` for the processed-artifact pattern) for complete examples.
 
-### 3. Register the dispatch
+### 4.3 Register the dispatch
 
 Add an `elif` branch in the `lambda_handler` function in `lambda/src/data_hub_lambda/handler.py`:
 
@@ -107,11 +108,11 @@ elif instrument_id == Instrument.BIO_RAD_CFX96.value:
 
 Don't forget to add the import at the top of `handler.py`.
 
-### 4. Add tests
+### 4.4 Add tests
 
 Add unit tests in `lambda/tests/` for the new processor. Integration tests will automatically cover the new instrument if it's registered in the shared library.
 
-### 5. Configure the S3 trigger
+### 4.5 Configure the S3 trigger
 
 Add a `LambdaConfiguration` entry to the `RawDataBucket` resource's `NotificationConfiguration` in `infra/template.yaml`. Each entry specifies a prefix (the instrument ID) and a suffix (the file extension):
 

docs/lambda.md

@@ -20,7 +20,7 @@ The Lambda has two invocation paths that converge on the same processing logic:
 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:
 
 1. The user clicks **Reprocess** on a failed or completed file in the web dashboard.
-2. The web app's `POST /api/v1/files/:fileId/reprocess` endpoint transitions the file to `processing` status, clears any previous error and report data, and sends a POST request to the Lambda Function URL.
+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 request includes an `Authorization: Bearer <LAMBDA_INVOKE_TOKEN>` header and a JSON body containing a synthetic S3 event payload.
 4. The Lambda handler detects the Function URL invocation (via `requestContext` in the event), verifies the Bearer token against the `LAMBDA_INVOKE_TOKEN` environment variable using constant-time comparison, and parses the S3 event from the request body.
 5. From here, processing follows the same dispatch logic as the S3 trigger path (steps 2–6 above).

docs/mcp.md

@@ -77,8 +77,7 @@ All tools return JSON encoded as a single text content block. Error cases set `i
 | --- | --- |
 | `search_runs` | Paginated search across runs with filtering, sorting, and date range. Supports plate-reader metadata filters (`wavelength`, `measurementMode`, `measurementType`). |
 | `get_run` | Get a single run by its natural key (`instrumentId` + `runId`). |
-| `get_run_report_data` | Get structured experimental results for a run — plate maps, well data, kinetic traces, spectra, etc. This is the primary tool for accessing results. |
-| `list_run_files` | List all files attached to a run, including processing status and metadata. |
+| `list_run_files` | List all files attached to a run, including processing status and metadata. Use `get_file_download_url` on processed CSV files to access experimental results. |
 | `get_run_archive_path` | Get the API path that streams a ZIP archive of all uploaded files for a run. Prepend the Data Hub origin and authenticate with the same Bearer token to download. |
 
 ### Files
@@ -87,7 +86,7 @@ All tools return JSON encoded as a single text content block. Error cases set `i
 | --- | --- |
 | `get_file` | Get detailed metadata for a single file by numeric ID. |
 | `get_file_download_url` | Get a pre-signed S3 URL for downloading a file's raw contents. URLs expire after 15 minutes and can be fetched without additional authentication. |
-| `reprocess_file` | **Write tool.** Re-run the Lambda processing workflow for a `failed` or `completed` file. Clears prior report data and transitions the file back to `processing`. Annotated `destructiveHint: true` so clients can warn before invoking. |
+| `reprocess_file` | **Write tool.** Re-run the Lambda processing workflow for a `failed` or `completed` file. Transitions the file back to `processing`. Annotated `destructiveHint: true` so clients can warn before invoking. |
 
 ### Watchers and system status
 
@@ -123,7 +122,7 @@ Once installed, ask your client questions like:
 
 - *"What instruments are active right now?"* → `list_instruments` with `status="active"`
 - *"Show me all SpectraMax runs from last Friday."* → `search_runs` with an `instrumentId` and date range
-- *"Summarize the well data from run `2026-03-26_experiment` on the plate reader."* → `run_analysis` prompt, which chains `get_run`, `get_run_report_data`, and `list_run_files`
+- *"Summarize the well data from run `2026-03-26_experiment` on the plate reader."* → `run_analysis` prompt, which chains `get_run`, `list_run_files`, and `get_file_download_url` for processed CSVs
 - *"The gel-doc in Lab 3 stopped uploading — what's wrong?"* → `troubleshoot_instrument` prompt, which inspects the watcher list and heartbeat history
 - *"Re-run processing for file 4217, we pushed a parser fix."* → `reprocess_file`. Clients typically confirm the destructive action with the user first.
 

lambda/src/data_hub_lambda/api_client.py

@@ -177,22 +177,17 @@ def update_file(
         *,
         status: str | None = None,
         metadata: dict[str, Any] | None = None,
-        report_data: list[dict[str, Any]] | None = None,
         error_message: str | None = None,
     ) -> FileResponse:
-        """Update a file record (status transition, metadata, report data).
+        """Update a file record (status transition, metadata).
 
         The API enforces a state machine: uploaded → processing → completed|failed.
-        `report_data` is a list of `{data_type, data}` objects inserted into
-        `run_report_data` for instruments that produce tabular data (e.g., plate reader).
         """
         payload: dict[str, Any] = {}
         if status is not None:
             payload["status"] = status
         if metadata is not None:
             payload["metadata"] = metadata
-        if report_data is not None:
-            payload["report_data"] = report_data
         if error_message is not None:
             payload["error_message"] = error_message
 

lambda/src/data_hub_lambda/spectramax_plate_reader/process_file.py

@@ -55,17 +55,27 @@ def process_file(instrument_id: InstrumentType, run_id: str, filename: str) -> s
         well_data = parse_raw_well_data(local_file_path)
         logger.info("Parsed %d well data rows.", len(well_data))
 
-        client.update_run(instrument_id, run_id, metadata=metadata)
-        client.update_file(
-            file_id,
-            status="completed",
-            report_data=[
-                {
-                    "data_type": "raw_well_data",
-                    "data": well_data.to_dict(orient="records"),
-                },
-            ],
+        processed_bucket = config.AWS_S3_PROCESSED_DATA_BUCKET
+        csv_filename = f"{run_id}_raw_well_data.csv"
+        csv_path = config.LOCAL_PROCESSED_DATA_DIRPATH / instrument_id / run_id / csv_filename
+        csv_path.parent.mkdir(parents=True, exist_ok=True)
+        well_data.to_csv(csv_path, index=False)
+
+        csv_s3_key = f"{instrument_id}/{run_id}/{csv_filename}"
+        s3_utils.upload_file(csv_path, f"s3://{processed_bucket}/{csv_s3_key}")
+        logger.info("Uploaded processed CSV to s3://%s/%s", processed_bucket, csv_s3_key)
+
+        client.create_file(
+            instrument_id=instrument_id,
+            run_id=run_id,
+            s3_bucket=processed_bucket or "",
+            s3_key=csv_s3_key,
+            filename=csv_filename,
+            category="processed",
         )
+
+        client.update_run(instrument_id, run_id, metadata=metadata)
+        client.update_file(file_id, status="completed")
         logger.info("File %s marked as completed.", filename)
 
     except Exception as e:

lambda/tests/integration/conftest.py

@@ -108,7 +108,7 @@ def integration_env(
 # Leaf-first (children before parents) to respect FK ordering.
 # Session-scoped rows (instruments, user, personal_access_tokens) are
 # intentionally excluded so tests don't need to re-seed them.
-_DATA_TABLES = ["run_report_data", "files", "instrument_runs"]
+_DATA_TABLES = ["files", "instrument_runs"]
 
 
 @pytest.fixture(autouse=True)

lambda/tests/integration/test_lambda_api.py

@@ -87,8 +87,8 @@ def test_csv_completes_with_dye_channels(
         # and stored as run-level metadata.
         assert run["metadata"]["dye_channels"] == ["ORANGE 560", "TAMRA", "ROX"]
 
-        # qPCR files don't produce tabular report_data (unlike plate readers).
-        assert run["report_data"] == []
+        # qPCR files don't produce processed CSV files (unlike plate readers).
+        assert all(f["category"] == "raw" for f in run["files"])
 
         mock_slack.assert_called_once()
         slack_msg = mock_slack.call_args[0][0]
@@ -97,13 +97,13 @@ def test_csv_completes_with_dye_channels(
 
 
 # ------------------------------------------------------------------
-# Test 4b: SpectraMax plate reader — happy path with report_data
+# Test 4b: SpectraMax plate reader — happy path with processed CSV
 # ------------------------------------------------------------------
 
 
 class TestSpectraMaxHappyPath:
     @pytest.mark.parametrize(
-        ("fixture_file", "run_id", "expected_metadata", "expected_first_plate"),
+        ("fixture_file", "run_id", "expected_metadata"),
         [
             pytest.param(
                 "spectramax_plate_reader_endpoint.xls",
@@ -113,7 +113,6 @@ class TestSpectraMaxHappyPath:
                     "measurement_type": "Endpoint",
                     "wavelength": "750 nm",
                 },
-                "Plate2",
                 id="endpoint",
             ),
             pytest.param(
@@ -124,7 +123,6 @@ class TestSpectraMaxHappyPath:
                     "measurement_type": "Well Scan",
                     "wavelength": "595 nm",
                 },
-                "Plate1",
                 id="well-scan",
             ),
             pytest.param(
@@ -135,33 +133,29 @@ class TestSpectraMaxHappyPath:
                     "measurement_type": "Kinetic",
                     "wavelength": "595 nm",
                 },
-                "Plate4",
                 id="kinetic",
             ),
         ],
     )
-    def test_xls_completes_with_report_data(
+    def test_xls_completes_with_processed_csv(
         self,
         integration_env: IntegrationEnv,
         make_s3_event: Callable[..., dict[str, Any]],
         s3_fixture_files: dict[str, Path],
         mock_context: MagicMock,
         mock_slack: MagicMock,
+        mock_s3_upload: MagicMock,
         fixture_file: str,
         run_id: str,
         expected_metadata: dict[str, str],
-        expected_first_plate: str,
     ) -> None:
-        # Register the real fixture CSV so the patched S3 download can find it.
         filename = f"{run_id}.xls"
         s3_key = f"spectramax-id3-plate-reader/{run_id}/{filename}"
         s3_fixture_files[s3_key] = _FIXTURES_DIR / fixture_file
 
-        # Fire the event to trigger the pipeline.
         event = make_s3_event("spectramax-id3-plate-reader", run_id, filename)
         lambda_handler(event, mock_context)
 
-        # Verify via the real API that the full pipeline wrote correct data.
         run = _api_get(
             integration_env.base_url,
             integration_env.api_token,
@@ -171,36 +165,26 @@ def test_xls_completes_with_report_data(
         assert run["source"] == "lambda"
         assert run["run_id"] == run_id
 
-        assert len(run["files"]) == 1
-        file = run["files"][0]
-        assert file["status"] == "completed"
+        # Two files: the raw .xls and the processed CSV.
+        assert len(run["files"]) == 2
+
+        raw_file = next(f for f in run["files"] if f["category"] == "raw")
+        processed_file = next(f for f in run["files"] if f["category"] == "processed")
+
+        assert raw_file["status"] == "completed"
+        assert processed_file["filename"] == f"{run_id}_raw_well_data.csv"
+        assert processed_file["status"] == "uploaded"
 
-        # Verify the run-level metadata.
         for key, value in expected_metadata.items():
             assert run["metadata"][key] == value
 
-        # Verify the raw well data.
-        assert len(run["report_data"]) == 1
-        run_report_data = run["report_data"][0]
-        assert run_report_data["data_type"] == "raw_well_data"
-        assert isinstance(run_report_data["data"], list)
-        assert len(run_report_data["data"]) > 0
-
-        first_row = run_report_data["data"][0]
-        expected_columns = {
-            "time",
-            "plate_name",
-            "well_position",
-            "temperature_c",
-            "value",
-            "row_label",
-            "column_label",
-            "wavelength",
-        }
-        assert set(first_row.keys()) == expected_columns
-        assert first_row["plate_name"] == expected_first_plate
+        # The pipeline uploads the processed CSV to the processed bucket.
+        mock_s3_upload.assert_called_once()
+        upload_dest = mock_s3_upload.call_args[0][1]
+        assert upload_dest == (
+            f"s3://test-processed-bucket/spectramax-id3-plate-reader/{run_id}/{run_id}_raw_well_data.csv"
+        )
 
-        # Verify the Slack notification.
         mock_slack.assert_called_once()
         slack_msg = mock_slack.call_args[0][0]
         assert run_id in slack_msg
@@ -262,8 +246,8 @@ def test_tif_completes_with_metadata_and_processed_image(
         assert processed_file["filename"] == f"{run_id}.png"
         assert processed_file["status"] == "uploaded"
 
-        # Gel Doc files don't produce tabular report_data.
-        assert run["report_data"] == []
+        # Gel Doc files don't produce additional CSV report data.
+        assert len([f for f in run["files"] if f["filename"].endswith(".csv")]) == 0
 
         # The pipeline uploads the contrast-enhanced PNG to the processed bucket.
         mock_s3_upload.assert_called_once()
@@ -489,42 +473,41 @@ def test_reprocess_clears_failed_state(
         assert "View CloudWatch logs" in mock_slack.call_args_list[0][0][0]
         assert "View in Data Hub" in mock_slack.call_args_list[1][0][0]
 
-    def test_reprocess_does_not_duplicate_report_data(
+    def test_reprocess_does_not_duplicate_processed_csv(
         self,
         integration_env: IntegrationEnv,
         make_s3_event: Callable[..., dict[str, Any]],
         s3_fixture_files: dict[str, Path],
         mock_context: MagicMock,
         mock_slack: MagicMock,
+        mock_s3_upload: MagicMock,
     ) -> None:
-        """Reprocessing a plate reader file must not double the report_data.
+        """Reprocessing a plate reader file must not duplicate the CSV file record.
 
-        The PATCH endpoint deletes existing run_report_data rows for the file
-        when transitioning back to "processing", so the second invocation
-        replaces rather than appends.
+        The create_file API is idempotent on s3_key, so the second invocation
+        returns the existing processed file rather than creating a duplicate.
         """
         run_id = "033126_CM_Od750"
         filename = f"{run_id}.xls"
         s3_key = f"spectramax-id3-plate-reader/{run_id}/{filename}"
         s3_fixture_files[s3_key] = _FIXTURES_DIR / "spectramax_plate_reader_endpoint.xls"
 
-        # Fire the event twice.
         event = make_s3_event("spectramax-id3-plate-reader", run_id, filename)
         lambda_handler(event, mock_context)
         lambda_handler(event, mock_context)
 
-        # Verify via the real API that the full pipeline wrote correct data.
         run = _api_get(
             integration_env.base_url,
             integration_env.api_token,
             f"/api/v1/instruments/spectramax-id3-plate-reader/runs/{run_id}",
         )
 
-        # Verify that the report_data was not duplicated.
-        assert len(run["report_data"]) == 1
-        assert run["report_data"][0]["data_type"] == "raw_well_data"
+        # Two files: one raw .xls and one processed CSV (not duplicated).
+        assert len(run["files"]) == 2
+        processed_files = [f for f in run["files"] if f["category"] == "processed"]
+        assert len(processed_files) == 1
+        assert processed_files[0]["filename"] == f"{run_id}_raw_well_data.csv"
 
-        # Verify the Slack notifications.
         assert mock_slack.call_count == 2
         for call in mock_slack.call_args_list:
             slack_msg = call[0][0]

watcher/tests/integration/conftest.py

@@ -54,7 +54,6 @@ def instrument_id() -> str:
 # ---------------------------------------------------------------------------
 
 _WATCHER_TABLES = [
-    "run_report_data",
     "files",
     "instrument_runs",
     "watcher_events",