> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ionworks.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 2026 07 13 raw data storage

# Raw Data Storage Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Add an org + project-scoped raw-data subsystem that stores uploaded files as-is in a dedicated `raw-data` Supabase bucket and links them many-to-many to cell measurements, exposed via the FastAPI backend and the `ionworks-api` Python SDK.

**Architecture:** Mirror the analyses trio (repo + storage-repo + service + routes) and the `material_property_datasets` org+project precedent. New `raw_data` table (required `organization_id` + `project_id`, composite FK to `projects`), new `measurement_raw_data_sources` join table (many-to-many, cascade both sides), new `raw-data` bucket with org-path-token storage RLS. Provenance attach/detach is measurement-centric and lives in `cell_measurements` routes/SDK. No web UI this phase.

**Tech Stack:** FastAPI, Supabase (Postgres + Storage), Pydantic v2, pytest (`just test-backend`), `requests`-based Python SDK (`ionworks-api`), mocked-HTTP SDK tests.

**Spec:** `docs/superpowers/specs/2026-07-13-raw-data-storage-design.md`

**Key precedents to copy (read these first):**

* Migration: `supabase/migrations/20260711000000_add_analyses.sql` (table + RLS + trigger idempotency), `supabase/migrations/20260504100000_add_material_property_datasets.sql` (composite `(project_id, organization_id)` FK + storage RLS), `supabase/migrations/20260504100001_create_material_property_datasets_bucket.sql` (bucket INSERT).
* Storage RLS verb set: `measurement-data` policies use `storage:create/read/update/delete` (`baseline.sql:7028-7095`). Table RLS uses domain verbs (`cell_measurement:*`).
* Backend: `backend/src/repositories/analysis.py`, `analysis_bucket.py`, `backend/src/services/analysis_service.py`, `backend/src/routes/analysis.py`, `backend/src/pydantic_models/analysis.py`.
* Bucket repo with raw+org path: `backend/src/repositories/material_property_datasets_bucket.py`.
* SDK: `packages/ionworks-api/ionworks/custom_model.py:229` (`upload_custom` → `upload_multipart`), `client.py:197-218` (sub-client wiring), `client.py:405` (`upload_multipart`), `models.py` (pydantic `extra="allow"`), `cell_measurement.py` (measurement client).
* Tests: `backend/tests/test_repositories/test_analysis_repository.py`, `test_services/test_analysis_service.py`, `test_routes/test_analysis_routes.py`; SDK `packages/ionworks-api/tests/test_measurement_types.py`.

**Conventions (rules that will be enforced by pre-commit / review):**

* Repositories paginate with `.select("*", count="exact")` + `.range(offset, offset+limit-1)`; return `PaginatedList`.
* Services raise domain exceptions (`NotFoundError`, `BadRequestError`, `ConflictError`, `AppError`, `ExternalServiceError`), never `HTTPException` or bare `ValueError`. Updates use `PATCH`. `organization_id` never in URL path — comes from `Depends(get_current_organization_id)`.
* Migrations: DDL-only + the one sanctioned bucket `INSERT ... ON CONFLICT DO NOTHING`; additive, idempotent.
* numpy-style docstrings, ≤88 col.

***

## File Structure

**Create:**

* `supabase/migrations/<ts>_add_raw_data.sql` — `raw_data` table, `measurement_raw_data_sources` join table, indexes, trigger, table + storage RLS.
* `supabase/migrations/<ts>_create_raw_data_bucket.sql` — the `raw-data` bucket INSERT.
* `supabase/seeds/buckets.sql` — **modify**: add `raw-data` to the seed list.
* `backend/src/pydantic_models/raw_data.py` — `RawData`, `UpdateRawData`, `AttachRawData`, `RawDataListResponse` models.
* `backend/src/repositories/raw_data.py` — `RawDataRepository` + join-table methods + dep.
* `backend/src/repositories/raw_data_bucket.py` — `RawDataBucketRepository` + dep.
* `backend/src/services/raw_data_service.py` — `RawDataService` + dep.
* `backend/src/routes/raw_data.py` — `/raw_data` CRUD router.
* `backend/tests/test_repositories/test_raw_data_repository.py`
* `backend/tests/test_services/test_raw_data_service.py`
* `backend/tests/test_routes/test_raw_data_routes.py`
* `packages/ionworks-api/ionworks/raw_data.py` — `RawDataClient`.
* `packages/ionworks-api/tests/test_raw_data.py`

**Modify:**

* `backend/src/main.py` — mount `raw_data_router` at `/raw_data`.
* `backend/src/routes/cell_measurements.py` — add attach/detach/list-raw-data sub-routes on `direct_router`.
* `backend/src/services/cell_measurements.py` (or the service backing those routes) — attach/detach/list orchestration.
* `packages/ionworks-api/ionworks/client.py` — wire `self.raw_data = RawDataClient(self)`.
* `packages/ionworks-api/ionworks/cell_measurement.py` — add `attach_raw_data` / `detach_raw_data` / `list_raw_data`.
* `packages/ionworks-api/ionworks/models.py` — add `RawData` model.
* `packages/ionworks-api/ionworks/__init__.py` — export `RawData`.
* SDK skills (`discover-api`, `upload-data`) via `ionworks-dev-sync` — final task.

***

## Task 1: Database migration — table, join table, bucket, RLS

**Files:**

* Create: `supabase/migrations/<timestamp>_add_raw_data.sql`
* Create: `supabase/migrations/<timestamp>_create_raw_data_bucket.sql`
* Modify: `supabase/seeds/buckets.sql`

Use a timestamp AFTER the latest existing migration. Get it with:
`ls supabase/migrations | sort | tail -1` then pick a larger `YYYYMMDDHHMMSS`.

* [ ] **Step 1: Write `<ts>_add_raw_data.sql`**

Copy the idempotency scaffolding (the `DO $$ ... pg_constraint` guards) verbatim from `20260711000000_add_analyses.sql` and the composite-FK block from `20260504100000_add_material_property_datasets.sql`. Contents:

```sql theme={null}
-- Raw data records: original uploaded files stored as-is (provenance),
-- org + project scoped. Files live in the dedicated `raw-data` bucket at
--   {organization_id}/{raw_data_id}/{filename}
-- Many-to-many provenance to cell_measurements via measurement_raw_data_sources.

CREATE TABLE IF NOT EXISTS "public"."raw_data" (
    "id"               "uuid"                   DEFAULT "gen_random_uuid"() NOT NULL,
    "organization_id"  "uuid"                   NOT NULL,
    "project_id"       "uuid"                   NOT NULL,
    "name"             "text"                   NOT NULL,
    "filename"         "text"                   NOT NULL,
    "content_type"     "text",
    "size_bytes"       bigint,
    "storage_path"     "text"                   NOT NULL,
    "metadata"         "jsonb"                  NOT NULL DEFAULT '{}'::"jsonb",
    "created_by"       "uuid",
    "created_at"       timestamp with time zone DEFAULT "now"() NOT NULL,
    "updated_at"       timestamp with time zone DEFAULT "now"() NOT NULL
);
ALTER TABLE "public"."raw_data" OWNER TO "postgres";

COMMENT ON TABLE "public"."raw_data" IS
    'Original uploaded raw files stored as-is (provenance), org + project '
    'scoped. Files live in the raw-data bucket at '
    '{organization_id}/{raw_data_id}/{filename}. Linked many-to-many to '
    'cell_measurements via measurement_raw_data_sources.';

-- PK (guarded)
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" c
        JOIN "pg_class" t ON c."conrelid" = t."oid"
        JOIN "pg_namespace" n ON t."relnamespace" = n."oid"
        WHERE c."conname" = 'raw_data_pkey' AND n."nspname" = 'public'
          AND t."relname" = 'raw_data'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_pkey" PRIMARY KEY ("id");
    END IF;
END $$;

-- FK → organizations
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" WHERE "conname" = 'raw_data_organization_id_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_organization_id_fkey"
            FOREIGN KEY ("organization_id")
            REFERENCES "public"."organizations"("id") ON DELETE CASCADE;
    END IF;
END $$;

-- FK → projects (composite: organization_id cannot drift from owning project).
-- Split ADD ... NOT VALID + VALIDATE, matching the sanctioned pattern in
-- 20260504100000_add_material_property_datasets.sql:100-123. Targets the
-- projects_id_organization_id_key UNIQUE(id, organization_id) added in
-- 20260501000005_add_composite_project_org_fkeys.sql.
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint"
        WHERE "conname" = 'raw_data_project_id_organization_id_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_project_id_organization_id_fkey"
            FOREIGN KEY ("project_id", "organization_id")
            REFERENCES "public"."projects"("id", "organization_id")
            ON DELETE CASCADE NOT VALID;
    END IF;
END $$;
DO $$ BEGIN
    IF EXISTS (
        SELECT 1 FROM "pg_constraint"
        WHERE "conname" = 'raw_data_project_id_organization_id_fkey'
          AND NOT "convalidated"
    ) THEN
        ALTER TABLE "public"."raw_data"
            VALIDATE CONSTRAINT "raw_data_project_id_organization_id_fkey";
    END IF;
END $$;

-- FK → users (nullable)
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" WHERE "conname" = 'raw_data_created_by_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_created_by_fkey"
            FOREIGN KEY ("created_by")
            REFERENCES "public"."users"("id") ON DELETE SET NULL;
    END IF;
END $$;

CREATE INDEX IF NOT EXISTS "idx_raw_data_project_id_created_at"
    ON "public"."raw_data" USING "btree" ("project_id", "created_at" DESC);
CREATE INDEX IF NOT EXISTS "idx_raw_data_organization_id"
    ON "public"."raw_data" USING "btree" ("organization_id");

CREATE OR REPLACE TRIGGER "handle_raw_data_updated_at"
    BEFORE UPDATE ON "public"."raw_data"
    FOR EACH ROW EXECUTE FUNCTION "extensions"."moddatetime"('updated_at');

ALTER TABLE "public"."raw_data" ENABLE ROW LEVEL SECURITY;

-- Table RLS (reuse cell_measurement:* verb set, like analyses)
DROP POLICY IF EXISTS "Users can create raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can create raw_data in their org" ON "public"."raw_data"
    FOR INSERT TO "authenticated"
    WITH CHECK ("public"."has_permission_in_org"("organization_id", 'cell_measurement:create'::"text"));
DROP POLICY IF EXISTS "Users can read raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can read raw_data in their org" ON "public"."raw_data"
    FOR SELECT TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:read'::"text"));
DROP POLICY IF EXISTS "Users can update raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can update raw_data in their org" ON "public"."raw_data"
    FOR UPDATE TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:update'::"text"))
    WITH CHECK ("public"."has_permission_in_org"("organization_id", 'cell_measurement:update'::"text"));
DROP POLICY IF EXISTS "Users can delete raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can delete raw_data in their org" ON "public"."raw_data"
    FOR DELETE TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:delete'::"text"));

GRANT ALL ON TABLE "public"."raw_data" TO "anon";
GRANT ALL ON TABLE "public"."raw_data" TO "authenticated";
GRANT ALL ON TABLE "public"."raw_data" TO "service_role";

-- ---------------------------------------------------------------------------
-- Join table: measurement_raw_data_sources (many-to-many provenance)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS "public"."measurement_raw_data_sources" (
    "id"                   "uuid"                   DEFAULT "gen_random_uuid"() NOT NULL,
    "cell_measurement_id"  "uuid"                   NOT NULL,
    "raw_data_id"          "uuid"                   NOT NULL,
    "created_at"           timestamp with time zone DEFAULT "now"() NOT NULL
);
ALTER TABLE "public"."measurement_raw_data_sources" OWNER TO "postgres";

DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_pkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_pkey" PRIMARY KEY ("id");
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_measurement_fkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_measurement_fkey"
            FOREIGN KEY ("cell_measurement_id")
            REFERENCES "public"."cell_measurements"("id") ON DELETE CASCADE;
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_raw_data_fkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_raw_data_fkey"
            FOREIGN KEY ("raw_data_id")
            REFERENCES "public"."raw_data"("id") ON DELETE CASCADE;
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_unique_pair') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_unique_pair"
            UNIQUE ("cell_measurement_id", "raw_data_id");
    END IF;
END $$;

CREATE INDEX IF NOT EXISTS "idx_mrds_measurement_id"
    ON "public"."measurement_raw_data_sources" USING "btree" ("cell_measurement_id");
CREATE INDEX IF NOT EXISTS "idx_mrds_raw_data_id"
    ON "public"."measurement_raw_data_sources" USING "btree" ("raw_data_id");

ALTER TABLE "public"."measurement_raw_data_sources" ENABLE ROW LEVEL SECURITY;

-- Join-table RLS: gate via the linked raw_data row's org (which is what a user
-- must already be able to read). A row is visible/insertable/deletable iff the
-- caller has the matching cell_measurement:* permission on the raw_data's org.
DROP POLICY IF EXISTS "Users can read mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can read mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR SELECT TO "authenticated"
    USING (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:read'::"text")
    ));
DROP POLICY IF EXISTS "Users can create mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can create mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR INSERT TO "authenticated"
    WITH CHECK (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:update'::"text")
    ));
DROP POLICY IF EXISTS "Users can delete mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can delete mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR DELETE TO "authenticated"
    USING (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:update'::"text")
    ));

GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "anon";
GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "authenticated";
GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "service_role";
```

* [ ] **Step 2: Write `<ts>_create_raw_data_bucket.sql`**

```sql theme={null}
INSERT INTO storage.buckets (id, name, public, file_size_limit, allowed_mime_types)
VALUES ('raw-data', 'raw-data', false, null, null)
ON CONFLICT (id) DO NOTHING;
```

* [ ] **Step 3: Add storage RLS for the `raw-data` bucket**

Append 4 `storage.objects` policies to `<ts>_add_raw_data.sql`, copying the `measurement-data` block from `baseline.sql:7061-7095` but with `bucket_id = 'raw-data'` and the `storage:*` verb set (this is the verb set every bucket uses; do NOT use `cell_measurement:*` here):

```sql theme={null}
-- Storage bucket RLS for raw-data. Path: {organization_id}/{raw_data_id}/{filename}
DROP POLICY IF EXISTS "Allow read storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow read storage for raw_data with permission" ON "storage"."objects"
    FOR SELECT USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:read'));
DROP POLICY IF EXISTS "Allow insert storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow insert storage for raw_data with permission" ON "storage"."objects"
    FOR INSERT WITH CHECK (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:create'));
DROP POLICY IF EXISTS "Allow update storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow update storage for raw_data with permission" ON "storage"."objects"
    FOR UPDATE USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:update'));
DROP POLICY IF EXISTS "Allow delete storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow delete storage for raw_data with permission" ON "storage"."objects"
    FOR DELETE USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:delete'));
```

* [ ] **Step 4: Add `raw-data` to the seed bucket list**

In `supabase/seeds/buckets.sql`, add a `('raw-data', 'raw-data', false)` row to the existing `INSERT ... ON CONFLICT DO NOTHING` list (match the column shape already there).

* [ ] **Step 5: Apply migrations and verify**

Run: `supabase migration up` (or the repo's migration command; do NOT `supabase db reset` — see memory note).
Expected: both migrations apply cleanly; re-running is a no-op (idempotent).
Verify tables exist:
`psql "$SUPABASE_DB_URL" -c "\d public.raw_data" -c "\d public.measurement_raw_data_sources"`
Verify bucket: `psql ... -c "select id from storage.buckets where id='raw-data';"` → 1 row.

* [ ] **Step 6: Commit**

```bash theme={null}
git add supabase/migrations/*_add_raw_data.sql supabase/migrations/*_create_raw_data_bucket.sql supabase/seeds/buckets.sql
git commit -m "feat(raw-data): add raw_data table, join table, bucket + RLS"
```

***

## Task 2: Pydantic models

**Files:**

* Create: `backend/src/pydantic_models/raw_data.py`

Mirror `backend/src/pydantic_models/analysis.py`.

* [ ] **Step 1: Write the models**

```python theme={null}
"""Pydantic models for the raw_data entity.

A raw_data record is an original uploaded file stored as-is (provenance),
owned by an organization and a project. Files live in the ``raw-data`` bucket
at ``{organization_id}/{raw_data_id}/{filename}``. Raw-data records link
many-to-many to cell measurements via ``measurement_raw_data_sources``.
"""

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class RawDataBase(BaseModel):
    """Fields shared by create/read representations of a raw-data record."""

    project_id: str = Field(..., description="Owning project ID (required).")
    name: str = Field(..., description="User-provided label for the raw file.")
    metadata: dict = Field(
        default_factory=dict, description="Loose free-form metadata."
    )


class RawData(RawDataBase):
    """A raw-data record as returned by the API."""

    id: str = Field(..., description="Unique identifier.")
    organization_id: str = Field(..., description="Owning organization.")
    filename: str = Field(..., description="Original uploaded filename.")
    content_type: str | None = Field(None, description="MIME type, if known.")
    size_bytes: int | None = Field(None, description="File size in bytes.")
    storage_path: str = Field(..., description="Path within the raw-data bucket.")
    created_by: str | None = Field(None, description="User ID of the uploader.")
    created_at: datetime = Field(..., description="Creation timestamp.")
    updated_at: datetime = Field(..., description="Last update timestamp.")

    model_config = ConfigDict(from_attributes=True)


class UpdateRawData(BaseModel):
    """Partial update for a raw-data record. Does not replace the file."""

    name: str | None = Field(None, description="New label.")
    metadata: dict | None = Field(None, description="Replacement metadata dict.")


class AttachRawData(BaseModel):
    """Body for bulk-attaching raw-data records to a measurement."""

    raw_data_ids: list[str] = Field(
        ..., min_length=1, description="Raw-data IDs to attach to the measurement."
    )
```

* [ ] **Step 2: Import-check**

Run: `cd backend && uv run python -c "from src.pydantic_models.raw_data import RawData, UpdateRawData, AttachRawData"`
Expected: no error.

* [ ] **Step 3: Commit**

```bash theme={null}
git add backend/src/pydantic_models/raw_data.py
git commit -m "feat(raw-data): add pydantic models"
```

***

## Task 3: DB repository (TDD)

**Files:**

* Create: `backend/src/repositories/raw_data.py`
* Test: `backend/tests/test_repositories/test_raw_data_repository.py`

Mirror `backend/src/repositories/analysis.py`. Read `test_analysis_repository.py` first for the mock-Supabase query-builder pattern.

* [ ] **Step 1: Write failing tests**

Cover: `list_by_project` builds an `eq("project_id")` + `eq("organization_id")` + `range(offset, offset+limit-1)` query and wraps into `PaginatedList` (assert `total` comes from `response.count`); `attach_measurement_sources` issues an upsert with `ignore_duplicates` semantics (idempotent); `list_sources_for_measurement` filters by `cell_measurement_id`. Follow the query-builder mocking in `test_analysis_repository.py`.

* [ ] **Step 2: Run — verify fail**

Run: `just test-backend tests/test_repositories/test_raw_data_repository.py -v`
Expected: FAIL (module/methods not defined).

* [ ] **Step 3: Implement `RawDataRepository`**

```python theme={null}
from typing import Annotated

from fastapi import Depends
from supabase import AsyncClient

from src.deps import get_supabase_user_client
from src.exceptions import AppError
from src.pydantic_models.pagination import PaginatedList
from src.pydantic_models.raw_data import RawData
from src.repositories.base import BaseRepository

_JOIN_TABLE = "measurement_raw_data_sources"


class RawDataRepository(BaseRepository[RawData]):
    """Repository for raw-data records and their measurement links."""

    def __init__(self, supabase: AsyncClient):
        super().__init__(supabase, "raw_data", RawData)

    async def list_by_project(
        self,
        project_id: str,
        organization_id: str,
        limit: int = 100,
        offset: int = 0,
    ) -> PaginatedList[RawData]:
        """List raw-data records for a project, newest first.

        Parameters
        ----------
        project_id : str
            Project whose raw-data records to list.
        organization_id : str
            Organization ID used as an additional RLS guard.
        limit : int, optional
            Maximum records to return. Defaults to 100.
        offset : int, optional
            Records to skip. Defaults to 0.

        Returns
        -------
        PaginatedList[RawData]
            Paginated list with count and total.
        """
        response = await (
            self.supabase.table(self.table_name)
            .select("*", count="exact")
            .eq("project_id", project_id)
            .eq("organization_id", organization_id)
            .order("created_at", desc=True)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [RawData(**row) for row in (response.data or [])]
        return self._wrap_paginated(items, response)

    async def get_raw_data(self, raw_data_id: str) -> RawData | None:
        """Fetch a single raw-data record by ID."""
        return await self.get({"id": raw_data_id})

    async def create_raw_data(
        self, organization_id: str, created_by: str | None, data: dict
    ) -> RawData:
        """Insert a raw-data row.

        Parameters
        ----------
        organization_id : str
            Owning organization.
        created_by : str | None
            Uploader user ID, may be None for service-account creation.
        data : dict
            Fields: project_id, name, filename, content_type, size_bytes,
            storage_path, metadata.

        Returns
        -------
        RawData
            The created record.

        Raises
        ------
        AppError
            If the insert fails.
        """
        payload = {
            "organization_id": organization_id,
            "created_by": created_by,
            **data,
        }
        result = await self.create(payload)
        if not result:
            raise AppError("Failed to create raw_data record")
        return result

    async def delete_raw_data(self, raw_data_id: str) -> bool:
        """Delete a raw-data row by ID."""
        return await self.delete(raw_data_id)

    # --- join table (measurement_raw_data_sources) ---

    async def attach_measurement_sources(
        self, cell_measurement_id: str, raw_data_ids: list[str]
    ) -> None:
        """Idempotently link a measurement to raw-data records (bulk).

        Uses an upsert that ignores duplicate ``(cell_measurement_id,
        raw_data_id)`` pairs, so re-attaching already-linked ids is a no-op.

        Parameters
        ----------
        cell_measurement_id : str
            The measurement to link.
        raw_data_ids : list[str]
            Raw-data IDs to link. A no-op when empty.
        """
        if not raw_data_ids:
            return
        rows = [
            {"cell_measurement_id": cell_measurement_id, "raw_data_id": rid}
            for rid in raw_data_ids
        ]
        await (
            self.supabase.table(_JOIN_TABLE)
            .upsert(rows, ignore_duplicates=True, on_conflict="cell_measurement_id,raw_data_id")
            .execute()
        )

    async def detach_measurement_source(
        self, cell_measurement_id: str, raw_data_id: str
    ) -> None:
        """Remove one measurement↔raw-data link."""
        await (
            self.supabase.table(_JOIN_TABLE)
            .delete()
            .eq("cell_measurement_id", cell_measurement_id)
            .eq("raw_data_id", raw_data_id)
            .execute()
        )

    async def list_sources_for_measurement(
        self, cell_measurement_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[RawData]:
        """List raw-data records linked to a measurement, via the join table."""
        response = await (
            self.supabase.table(_JOIN_TABLE)
            .select("raw_data!inner(*)", count="exact")
            .eq("cell_measurement_id", cell_measurement_id)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [RawData(**row["raw_data"]) for row in (response.data or [])]
        return self._wrap_paginated(items, response)

    async def list_measurement_ids_for_raw_data(
        self, raw_data_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[str]:
        """List measurement IDs linked to a raw-data record."""
        response = await (
            self.supabase.table(_JOIN_TABLE)
            .select("cell_measurement_id", count="exact")
            .eq("raw_data_id", raw_data_id)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [row["cell_measurement_id"] for row in (response.data or [])]
        return self._wrap_paginated(items, response)


def get_raw_data_repository(
    supabase: Annotated[AsyncClient, Depends(get_supabase_user_client)],
) -> RawDataRepository:
    """FastAPI dependency that returns a RawDataRepository."""
    return RawDataRepository(supabase)
```

**Note for implementer:** confirm the Supabase client's `upsert(..., ignore_duplicates=True, on_conflict=...)` kwargs against the installed `supabase`/`postgrest` version. If the kwarg names differ, adjust; the behavior required is "insert these pairs, silently skip ones that already exist." Verify by grepping existing `.upsert(` uses in `backend/src/`.

* [ ] **Step 4: Run — verify pass**

Run: `just test-backend tests/test_repositories/test_raw_data_repository.py -v`
Expected: PASS.

* [ ] **Step 5: Commit**

```bash theme={null}
git add backend/src/repositories/raw_data.py backend/tests/test_repositories/test_raw_data_repository.py
git commit -m "feat(raw-data): add DB repository with join-table methods"
```

***

## Task 4: Storage repository (TDD)

**Files:**

* Create: `backend/src/repositories/raw_data_bucket.py`
* Test: covered via the service test (storage repo is thin); optional dedicated test mirroring `test_cell_data_bucket.py`.

Mirror `analysis_bucket.py` + `material_property_datasets_bucket.py`.

* [ ] **Step 1: Implement `RawDataBucketRepository`**

```python theme={null}
import logging
from typing import Annotated
from urllib.parse import urlparse, urlunparse

from fastapi import Depends
from supabase import AsyncClient

from src.deps import get_supabase_user_client
from src.exceptions import ExternalServiceError
from src.settings import settings
from src.utils.storage_utils import async_upload_file_to_storage

logger = logging.getLogger(__name__)

BUCKET_NAME = "raw-data"


class RawDataBucketRepository:
    """Storage repository for raw-data files.

    Each record stores a single original file at
    ``{organization_id}/{raw_data_id}/{filename}`` in the ``raw-data`` bucket.
    """

    bucket_name = BUCKET_NAME

    def __init__(self, supabase: AsyncClient) -> None:
        self.supabase = supabase
        self.bucket = supabase.storage.from_(self.bucket_name)

    def build_path(self, organization_id: str, raw_data_id: str, filename: str) -> str:
        """Return the storage path for a raw-data file.

        Parameters
        ----------
        organization_id : str
            Organization UUID (first path token, used by RLS).
        raw_data_id : str
            Raw-data record UUID.
        filename : str
            Original uploaded filename.

        Returns
        -------
        str
            Path relative to the bucket root.
        """
        return f"{organization_id}/{raw_data_id}/{filename}"

    async def upload(
        self,
        path: str,
        file_content: bytes,
        content_type: str | None,
    ) -> dict:
        """Upload a raw-data file (no overwrite).

        Raises
        ------
        ExternalServiceError
            If the upload fails.
        """
        options = {"upsert": "false"}
        if content_type:
            options["content_type"] = content_type
        try:
            return await async_upload_file_to_storage(
                supabase=self.supabase,
                filepath=path,
                file_content=file_content,
                bucket_name=self.bucket_name,
                file_options=options,
            )
        except Exception as e:
            raise ExternalServiceError(
                self.bucket_name, f"Failed to upload raw-data file at {path}: {e}"
            ) from e

    async def create_signed_download_url(
        self,
        path: str,
        expires_in: int = 300,
        download_filename: str | None = None,
    ) -> str:
        """Create a short-lived signed download URL. Copied from analysis_bucket."""
        options = {"download": download_filename} if download_filename else {}
        try:
            result = await self.bucket.create_signed_url(
                path=path, expires_in=expires_in, options=options or None
            )
            url = result.get("signedURL") or result.get("signed_url")
            if not url:
                raise ExternalServiceError(
                    self.bucket_name, "Signed URL not returned by storage API"
                )
            if settings.STORAGE_PUBLIC_URL:
                parsed = urlparse(url)
                public = urlparse(settings.STORAGE_PUBLIC_URL)
                url = urlunparse(
                    parsed._replace(scheme=public.scheme, netloc=public.netloc)
                )
            return url
        except ExternalServiceError:
            raise
        except Exception as e:
            raise ExternalServiceError(
                self.bucket_name, f"Failed to create signed URL for {path}: {e}"
            ) from e

    async def delete_at_path(self, path: str) -> None:
        """Delete a single file at the given bucket path (best-effort)."""
        try:
            await self.bucket.remove(paths=[path])
        except Exception as e:
            logger.warning(
                "Failed to delete raw-data file at %s; may be orphaned: %s", path, e
            )


def get_raw_data_bucket_repository(
    supabase: Annotated[AsyncClient, Depends(get_supabase_user_client)],
) -> RawDataBucketRepository:
    """FastAPI dependency that returns a RawDataBucketRepository."""
    return RawDataBucketRepository(supabase)
```

* [ ] **Step 2: Import-check + commit**

Run: `cd backend && uv run python -c "from src.repositories.raw_data_bucket import RawDataBucketRepository"`

```bash theme={null}
git add backend/src/repositories/raw_data_bucket.py
git commit -m "feat(raw-data): add storage bucket repository"
```

***

## Task 5: Service (TDD)

**Files:**

* Create: `backend/src/services/raw_data_service.py`
* Test: `backend/tests/test_services/test_raw_data_service.py`

Mirror `analysis_service.py`. Read `test_analysis_service.py` for the AsyncMock fixture style (note `build_path` must be a `MagicMock`, not `AsyncMock`, since it's sync).

* [ ] **Step 1: Write failing tests**

Cover the ordering invariants and guards:

1. `create` inserts the DB row, uploads, returns the record; upload failure ⇒ row is deleted (rollback) and error re-raised.
2. `delete` removes the DB row first, then best-effort file delete; a storage failure does NOT raise.
3. `get` / `update` / `delete` raise `NotFoundError` when the record is missing or `organization_id` mismatches.
4. `attach` validates the measurement is in-org (else `NotFoundError`) and every raw\_data\_id exists + is in-org (else `BadRequestError` naming the id); then calls `attach_measurement_sources`.
5. `get_download_url` returns the signed URL for the record's `storage_path`.

* [ ] **Step 2: Run — verify fail**

Run: `just test-backend tests/test_services/test_raw_data_service.py -v`
Expected: FAIL.

* [ ] **Step 3: Implement `RawDataService`**

```python theme={null}
import logging
from typing import Annotated

from fastapi import Depends

from src.exceptions import BadRequestError, NotFoundError
from src.pydantic_models.pagination import PaginatedList
from src.pydantic_models.raw_data import RawData
from src.repositories.cell_measurements import (
    CellMeasurementRepository,
    get_cell_measurement_repository,
)
from src.repositories.raw_data import RawDataRepository, get_raw_data_repository
from src.repositories.raw_data_bucket import (
    RawDataBucketRepository,
    get_raw_data_bucket_repository,
)

logger = logging.getLogger(__name__)


class RawDataService:
    """Business logic for raw-data records and measurement provenance links.

    Ordering invariants (same as AnalysisService):

    - **create**: DB row first, then file upload. On upload failure the row is
      deleted and the error re-raised — no orphan row without its file.
    - **delete**: DB row removed first, then best-effort file delete. A storage
      failure orphans a file (recoverable) rather than leaving a visible record
      with inaccessible data.

    Deleting a *measurement* cascades away join rows but never touches raw_data
    rows or files (a raw file may back other measurements).
    """

    resource_type = "raw_data"

    def __init__(
        self,
        raw_data_repo: RawDataRepository,
        bucket_repo: RawDataBucketRepository,
        measurement_repo: CellMeasurementRepository,
    ) -> None:
        self.repo = raw_data_repo
        self.bucket_repo = bucket_repo
        self.measurement_repo = measurement_repo

    async def create(
        self,
        *,
        organization_id: str,
        user_id: str,
        project_id: str,
        name: str,
        filename: str,
        content_type: str | None,
        size_bytes: int | None,
        metadata: dict,
        file_content: bytes,
    ) -> RawData:
        """Create a raw-data record and upload its file (row first, rollback on
        upload failure)."""
        # Insert row without storage_path, then patch it in once we know the id.
        record = await self.repo.create_raw_data(
            organization_id=organization_id,
            created_by=user_id,
            data={
                "project_id": project_id,
                "name": name,
                "filename": filename,
                "content_type": content_type,
                "size_bytes": size_bytes,
                "metadata": metadata,
                "storage_path": "",  # set below
            },
        )
        path = self.bucket_repo.build_path(organization_id, record.id, filename)
        try:
            await self.bucket_repo.upload(path, file_content, content_type)
            updated = await self.repo.update({"id": record.id}, {"storage_path": path})
        except Exception:
            logger.exception("Rolling back raw_data %s after upload failure", record.id)
            try:
                await self.repo.delete_raw_data(record.id)
            except Exception:
                logger.exception("Failed to roll back raw_data row %s", record.id)
            raise
        return updated or record

    async def get(self, raw_data_id: str, organization_id: str) -> RawData:
        """Fetch one record, guarding org ownership."""
        record = await self.repo.get_raw_data(raw_data_id)
        if not record or record.organization_id != organization_id:
            raise NotFoundError(self.resource_type, raw_data_id)
        return record

    async def get_download_url(
        self, raw_data_id: str, organization_id: str, expires_in: int = 300
    ) -> str:
        """Return a signed download URL for a record's file."""
        record = await self.get(raw_data_id, organization_id)
        return await self.bucket_repo.create_signed_download_url(
            record.storage_path,
            expires_in=expires_in,
            download_filename=record.filename,
        )

    async def list_by_project(
        self, project_id: str, organization_id: str, limit: int, offset: int
    ) -> PaginatedList[RawData]:
        """List a project's raw-data records."""
        return await self.repo.list_by_project(
            project_id=project_id,
            organization_id=organization_id,
            limit=limit,
            offset=offset,
        )

    async def update(
        self,
        raw_data_id: str,
        organization_id: str,
        name: str | None,
        metadata: dict | None,
    ) -> RawData:
        """Partial-update name/metadata. Does not replace the file."""
        record = await self.get(raw_data_id, organization_id)
        new_data: dict = {}
        if name is not None:
            new_data["name"] = name
        if metadata is not None:
            new_data["metadata"] = metadata
        if not new_data:
            return record
        updated = await self.repo.update({"id": raw_data_id}, new_data)
        if updated is None:
            raise NotFoundError(self.resource_type, raw_data_id)
        return updated

    async def delete(self, raw_data_id: str, organization_id: str) -> None:
        """Delete a record then best-effort its file."""
        record = await self.get(raw_data_id, organization_id)
        await self.repo.delete_raw_data(raw_data_id)
        try:
            await self.bucket_repo.delete_at_path(record.storage_path)
        except Exception:
            logger.warning(
                "Storage delete failed for raw_data %s; file orphaned", raw_data_id
            )

    # --- provenance (measurement-centric) ---

    async def attach_to_measurement(
        self,
        cell_measurement_id: str,
        raw_data_ids: list[str],
        organization_id: str,
    ) -> PaginatedList[RawData]:
        """Bulk-attach raw-data records to a measurement (idempotent).

        Raises
        ------
        NotFoundError
            If the measurement is missing or not in the caller's org.
        BadRequestError
            If any raw_data_id is missing or in another org.
        """
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        for rid in raw_data_ids:
            rec = await self.repo.get_raw_data(rid)
            if rec is None or rec.organization_id != organization_id:
                raise BadRequestError(
                    f"raw_data '{rid}' does not exist in this organization."
                )
        await self.repo.attach_measurement_sources(cell_measurement_id, raw_data_ids)
        return await self.repo.list_sources_for_measurement(cell_measurement_id)

    async def detach_from_measurement(
        self, cell_measurement_id: str, raw_data_id: str, organization_id: str
    ) -> None:
        """Remove one measurement↔raw-data link (org-guarded via measurement)."""
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        await self.repo.detach_measurement_source(cell_measurement_id, raw_data_id)

    async def list_sources_for_measurement(
        self, cell_measurement_id: str, organization_id: str, limit: int, offset: int
    ) -> PaginatedList[RawData]:
        """List a measurement's raw-data sources (org-guarded via measurement)."""
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        return await self.repo.list_sources_for_measurement(
            cell_measurement_id, limit=limit, offset=offset
        )


def get_raw_data_service(
    raw_data_repo: Annotated[RawDataRepository, Depends(get_raw_data_repository)],
    bucket_repo: Annotated[
        RawDataBucketRepository, Depends(get_raw_data_bucket_repository)
    ],
    measurement_repo: Annotated[
        CellMeasurementRepository, Depends(get_cell_measurement_repository)
    ],
) -> RawDataService:
    """FastAPI dependency constructing a RawDataService."""
    return RawDataService(
        raw_data_repo=raw_data_repo,
        bucket_repo=bucket_repo,
        measurement_repo=measurement_repo,
    )
```

**Implementer note:** confirm `CellMeasurement.organization_id` is populated by `get_by_id`'s default select (`*, users(email)`). The measurements table derives org via join to `cell_specifications`; if `.organization_id` is not directly on the row, use the existing helper `get_by_id_with_org` or the select that includes org (grep `organization_id` in `repositories/cell_measurements.py` — there is an org-scoped select around line 170). Adjust the guard to use whatever attribute/\[method] exposes the measurement's org. If none is convenient, fall back to the repo's existing org-scoped getter used by `AnalysisService` (it calls `measurement_repo.get_by_id(...)` then compares `.organization_id`, so the same attribute is available — verify).

* [ ] **Step 4: Run — verify pass**

Run: `just test-backend tests/test_services/test_raw_data_service.py -v`
Expected: PASS.

* [ ] **Step 5: Commit**

```bash theme={null}
git add backend/src/services/raw_data_service.py backend/tests/test_services/test_raw_data_service.py
git commit -m "feat(raw-data): add service with create/delete invariants and provenance attach"
```

***

## Task 6: Routes + mounting (TDD)

**Files:**

* Create: `backend/src/routes/raw_data.py`
* Modify: `backend/src/main.py`
* Test: `backend/tests/test_routes/test_raw_data_routes.py`

Mirror `backend/src/routes/analysis.py`. Read `test_analysis_routes.py` for the route-test harness (dependency overrides + TestClient).

* [ ] **Step 1: Write failing route tests**

Cover: `POST ""` multipart create returns 201 + record with `content_type`/`size_bytes` derived server-side; `GET ""` requires `project_id` (422 without it) and paginates; `GET /{id}` 200/404; `GET /{id}/download-url` returns `{"url": ...}`; `PATCH /{id}` partial; `DELETE /{id}` 204. Use dependency overrides to inject a fake service.

* [ ] **Step 2: Run — verify fail**

Run: `just test-backend tests/test_routes/test_raw_data_routes.py -v`
Expected: FAIL.

* [ ] **Step 3: Implement the router**

```python theme={null}
import json
import logging
from typing import Annotated

from fastapi import APIRouter, Depends, Form, Query, UploadFile, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

from src.deps import get_current_organization_id, get_current_user_id
from src.exceptions import BadRequestError
from src.pydantic_models.pagination import PaginatedResponse
from src.pydantic_models.raw_data import RawData, UpdateRawData
from src.services.raw_data_service import RawDataService, get_raw_data_service

logger = logging.getLogger(__name__)

router = APIRouter(tags=["Raw Data"])


def _parse_metadata(metadata_json: str) -> dict:
    """Parse a JSON-encoded metadata object (raises BadRequestError on bad input)."""
    try:
        metadata = json.loads(metadata_json)
    except (json.JSONDecodeError, TypeError, ValueError) as exc:
        raise BadRequestError(f"metadata must be a valid JSON object: {exc}") from exc
    if not isinstance(metadata, dict):
        raise BadRequestError("metadata must be a JSON object.")
    return metadata


@router.post("", response_model=RawData, status_code=status.HTTP_201_CREATED,
             summary="Upload a raw-data file")
async def create_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    user_id: Annotated[str, Depends(get_current_user_id)],
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    file: UploadFile,
    project_id: str = Form(...),
    name: str = Form(...),
    metadata: str = Form("{}"),
) -> JSONResponse:
    """Upload a raw file, stored as-is in the raw-data bucket under the project."""
    metadata_dict = _parse_metadata(metadata)
    raw = await file.read()
    record = await service.create(
        organization_id=organization_id,
        user_id=user_id,
        project_id=project_id,
        name=name,
        filename=file.filename or "upload.bin",
        content_type=file.content_type,
        size_bytes=len(raw),
        metadata=metadata_dict,
        file_content=raw,
    )
    return JSONResponse(
        content=record.model_dump(mode="json"),
        status_code=status.HTTP_201_CREATED,
    )


@router.get("", response_model=PaginatedResponse[RawData],
            summary="List raw-data records for a project")
async def list_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    project_id: str = Query(..., description="Filter by owning project ID"),
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """List a project's raw-data records, newest first."""
    result = await service.list_by_project(
        project_id=project_id, organization_id=organization_id,
        limit=limit, offset=offset,
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))


@router.get("/{raw_data_id}", response_model=RawData, summary="Get a raw-data record")
async def get_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Retrieve a single raw-data record by ID."""
    record = await service.get(raw_data_id, organization_id)
    return JSONResponse(content=record.model_dump(mode="json"))


@router.get("/{raw_data_id}/download-url",
            summary="Get a signed download URL for a raw-data file")
async def get_raw_data_download_url(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Return ``{"url": "..."}`` — a signed URL valid for 5 minutes."""
    url = await service.get_download_url(raw_data_id, organization_id, expires_in=300)
    return JSONResponse(content={"url": url})


@router.patch("/{raw_data_id}", response_model=RawData,
              summary="Update a raw-data record's metadata")
async def update_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    body: UpdateRawData,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Update name/metadata. Does not replace the file."""
    updated = await service.update(
        raw_data_id=raw_data_id, organization_id=organization_id,
        name=body.name, metadata=body.metadata,
    )
    return JSONResponse(content=updated.model_dump(mode="json"))


@router.delete("/{raw_data_id}", status_code=status.HTTP_204_NO_CONTENT,
               summary="Delete a raw-data record")
async def delete_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> None:
    """Delete a raw-data record and its file."""
    await service.delete(raw_data_id, organization_id)


@router.get("/{raw_data_id}/cell_measurements",
            response_model=PaginatedResponse[str],
            summary="List measurement IDs produced from a raw-data record")
async def list_raw_data_measurements(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """Reverse provenance read: measurements linked to this raw-data record."""
    await service.get(raw_data_id, organization_id)  # org guard / 404
    result = await service.repo.list_measurement_ids_for_raw_data(
        raw_data_id, limit=limit, offset=offset
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))
```

* [ ] **Step 4: Mount in `main.py`**

After the analyses include (`app.include_router(analysis_router, prefix="/analyses")`), add:

```python theme={null}
from src.routes.raw_data import router as raw_data_router
...
app.include_router(raw_data_router, prefix="/raw_data")
```

* [ ] **Step 5: Run — verify pass**

Run: `just test-backend tests/test_routes/test_raw_data_routes.py -v`
Expected: PASS.

* [ ] **Step 6: Commit**

```bash theme={null}
git add backend/src/routes/raw_data.py backend/src/main.py backend/tests/test_routes/test_raw_data_routes.py
git commit -m "feat(raw-data): add /raw_data CRUD routes and mount router"
```

***

## Task 7: Measurement-centric attach/detach routes (TDD)

**Files:**

* Modify: `backend/src/routes/cell_measurements.py` (add to `direct_router`)
* Test: extend `backend/tests/test_routes/test_raw_data_routes.py` or a new `test_measurement_raw_data_routes.py`.

The `RawDataService` (Task 5) already has `attach_to_measurement` / `detach_from_measurement` / `list_sources_for_measurement`. These routes just call it, but are mounted under `/cell_measurements` for discoverability.

* [ ] **Step 1: Write failing tests**

Cover: `POST /cell_measurements/{id}/raw_data` with `{"raw_data_ids": [...]}` attaches and returns the source list; a bad id ⇒ 400; a foreign measurement ⇒ 404; re-attaching the same id is idempotent (still 200). `DELETE /cell_measurements/{id}/raw_data/{raw_data_id}` ⇒ 204. `GET /cell_measurements/{id}/raw_data` lists.

* [ ] **Step 2: Run — verify fail.** `just test-backend tests/test_routes/test_measurement_raw_data_routes.py -v` → FAIL.

* [ ] **Step 3a: Add the required imports to `cell_measurements.py`**

Verified against the file: `direct_router` (line 53) already imports `Query`,
`status`, `JSONResponse`, `jsonable_encoder`, `get_current_organization_id`,
`Annotated`, `Depends`. It does NOT import these — add them:

```python theme={null}
from src.pydantic_models.pagination import PaginatedResponse
from src.pydantic_models.raw_data import AttachRawData, RawData
from src.services.raw_data_service import RawDataService, get_raw_data_service
```

* [ ] **Step 3b: Add the routes to `direct_router` in `cell_measurements.py`**

```python theme={null}
from src.pydantic_models.raw_data import AttachRawData, RawData
from src.services.raw_data_service import RawDataService, get_raw_data_service

@direct_router.post("/{measurement_id}/raw_data",
                    response_model=PaginatedResponse[RawData],
                    summary="Attach raw-data records to a measurement")
async def attach_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    body: AttachRawData,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Bulk-attach raw-data records to a measurement (idempotent)."""
    result = await service.attach_to_measurement(
        cell_measurement_id=measurement_id,
        raw_data_ids=body.raw_data_ids,
        organization_id=organization_id,
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))

@direct_router.delete("/{measurement_id}/raw_data/{raw_data_id}",
                      status_code=status.HTTP_204_NO_CONTENT,
                      summary="Detach a raw-data record from a measurement")
async def detach_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> None:
    """Remove one measurement↔raw-data link."""
    await service.detach_from_measurement(measurement_id, raw_data_id, organization_id)

@direct_router.get("/{measurement_id}/raw_data",
                   response_model=PaginatedResponse[RawData],
                   summary="List a measurement's raw-data sources")
async def list_measurement_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """List raw-data records linked to a measurement."""
    result = await service.list_sources_for_measurement(
        measurement_id, organization_id, limit=limit, offset=offset
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))
```

**Implementer note (route ordering — verified clean):** `direct_router` has 17
routes; the new static-suffix `/{measurement_id}/raw_data` and
`/{measurement_id}/raw_data/{raw_data_id}` do not collide with any existing
route, and the only `:path` converter is scoped to `filename` in
`/{measurement_id}/files/{filename:path}`. No shadowing — no ordering fix
needed.

* [ ] **Step 4: Run — verify pass.** → PASS.

* [ ] **Step 5: Commit**

```bash theme={null}
git add backend/src/routes/cell_measurements.py backend/tests/test_routes/test_measurement_raw_data_routes.py
git commit -m "feat(raw-data): measurement-centric attach/detach/list provenance routes"
```

***

## Task 8: Backend full-suite gate

* [ ] **Step 1: Run the whole backend suite + prek**

Run: `just test-backend` then invoke the `prek` skill (pre-commit) on staged changes.
Expected: all green; API-convention and PATCH-vs-PUT hooks pass. Fix any failures before proceeding.

* [ ] **Step 2 (optional, if local Supabase up): RLS integration**

If `supabase status` shows a running DB, add/extend a test under `backend/tests/test_search/` (or the RLS test area) asserting org A cannot read org B's raw\_data rows/files. Run `just test-search`. If Supabase isn't running, note it and skip (these auto-skip).

***

## Task 9: SDK — `RawData` model + `RawDataClient` (TDD)

**Files:**

* Modify: `packages/ionworks-api/ionworks/models.py`
* Create: `packages/ionworks-api/ionworks/raw_data.py`
* Modify: `packages/ionworks-api/ionworks/client.py`
* Modify: `packages/ionworks-api/ionworks/__init__.py`
* Test: `packages/ionworks-api/tests/test_raw_data.py`

Read `packages/ionworks-api/ionworks/custom_model.py:180-248` (upload template), `material.py:16-35` (sub-client boilerplate), `models.py` (model + `_build_endpoint`/`_parse_list_response`), and `tests/test_measurement_types.py` (mocked-HTTP test style).

* [ ] **Step 1: Add the `RawData` model to `models.py`**

```python theme={null}
class RawData(BaseModel):
    """A raw-data record. The API defines the full schema; extra fields allowed."""

    model_config = ConfigDict(extra="allow")
    id: str
    project_id: str
    name: str
    filename: str
```

Export it: add `RawData` to the `from .models import (...)` block and `__all__` in `__init__.py`.

* [ ] **Step 2: Write failing SDK tests (`tests/test_raw_data.py`)**

Mirror `test_measurement_types.py`. Construct `RawDataClient(MagicMock())`. Assert:

* `upload(project_id, <path>, name=...)` calls `client.upload_multipart("/raw_data", data={...}, files={...})` with `project_id`/`name`/`metadata` in `data` and a `("file", ...)` tuple in `files`; returns `RawData(**resp)`.

* `list(project_id)` calls `client.get` with the endpoint carrying `project_id` and parses a `PaginatedList[RawData]`.

* `get`, `download_url` (returns `resp["url"]`), `update` (PATCH), `delete`.

* [ ] **Step 3: Run — verify fail.** `cd packages/ionworks-api && uv run python -m pytest tests/test_raw_data.py -v` → FAIL.

* [ ] **Step 4: Implement `RawDataClient` (`ionworks/raw_data.py`)**

```python theme={null}
"""Client for raw-data records (original uploaded files, org + project scoped)."""

import json
import os
from typing import IO, Any

from .models import PaginatedList, RawData, _build_endpoint, _parse_list_response


class RawDataClient:
    """Upload and manage raw-data files and their measurement links."""

    _BASE = "/raw_data"

    def __init__(self, client: Any) -> None:
        self.client = client

    def upload(
        self,
        project_id: str,
        file: str | os.PathLike | IO[bytes],
        name: str | None = None,
        metadata: dict | None = None,
    ) -> RawData:
        """Upload a raw file, stored as-is under the given project.

        Parameters
        ----------
        project_id : str
            Owning project ID.
        file : str | os.PathLike | IO[bytes]
            Path to a file, or an open binary file-like object.
        name : str | None, optional
            Label for the record. Defaults to the file's basename.
        metadata : dict | None, optional
            Free-form metadata.

        Returns
        -------
        RawData
            The created record.
        """
        handle, filename, opened = _resolve_file(file)
        try:
            data = {
                "project_id": project_id,
                "name": name or filename,
                "metadata": json.dumps(metadata or {}),
            }
            files = {"file": (filename, handle, "application/octet-stream")}
            response = self.client.upload_multipart(self._BASE, data=data, files=files)
        finally:
            if opened:
                handle.close()
        if not isinstance(response, dict):
            raise TypeError("Unexpected upload response")
        return RawData(**response)

    def list(
        self, project_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[RawData]:
        """List raw-data records for a project."""
        endpoint = _build_endpoint(
            self._BASE, {"project_id": project_id, "limit": limit, "offset": offset}
        )
        return _parse_list_response(self.client.get(endpoint), RawData)

    def get(self, raw_data_id: str) -> RawData:
        """Fetch one raw-data record."""
        return RawData(**self.client.get(f"{self._BASE}/{raw_data_id}"))

    def download_url(self, raw_data_id: str) -> str:
        """Get a short-lived signed download URL."""
        return self.client.get(f"{self._BASE}/{raw_data_id}/download-url")["url"]

    def update(
        self, raw_data_id: str, name: str | None = None, metadata: dict | None = None
    ) -> RawData:
        """Partial-update a record's name/metadata."""
        payload: dict = {}
        if name is not None:
            payload["name"] = name
        if metadata is not None:
            payload["metadata"] = metadata
        return RawData(**self.client.patch(f"{self._BASE}/{raw_data_id}", payload))

    def delete(self, raw_data_id: str) -> None:
        """Delete a raw-data record and its file."""
        self.client.delete(f"{self._BASE}/{raw_data_id}")

    def list_measurements(
        self, raw_data_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[str]:
        """List measurement IDs produced from this raw-data record.

        The endpoint returns plain string ids, so build the PaginatedList
        directly — do NOT use ``_parse_list_response``, which calls
        ``model_class(**item)`` and would crash on ``str(**"uuid")``.
        """
        endpoint = _build_endpoint(
            f"{self._BASE}/{raw_data_id}/cell_measurements",
            {"limit": limit, "offset": offset},
        )
        resp = self.client.get(endpoint)
        return PaginatedList(
            items=list(resp.get("items", [])),
            total=resp.get("total", 0),
            limit=limit,
            offset=offset,
        )


def _resolve_file(
    file: str | os.PathLike | IO[bytes],
) -> tuple[IO[bytes], str, bool]:
    """Return an open binary handle, a filename, and whether we opened it.

    The third element tells the caller to close the handle in a ``finally`` only
    when this function opened it (a path was passed) — never close a caller-owned
    file-like object.
    """
    if isinstance(file, (str, os.PathLike)):
        return open(file, "rb"), os.path.basename(str(file)), True
    filename = os.path.basename(getattr(file, "name", "upload.bin"))
    return file, filename, False
```

**Implementer note:** `_parse_list_response` (`models.py:104-111`) does `model_class(**item)` per item, so it CANNOT be used for plain-string ids — hence the direct `PaginatedList(...)` construction above. Confirm `PaginatedList`'s constructor kwargs (`items`/`total`/`limit`/`offset`) against `models.py:25` and adjust names to whatever that class defines. The **backend** side is unaffected — `PaginatedResponse[str]` validates fine.

* [ ] **Step 5: Wire the sub-client in `client.py`**

In `Ionworks.__init__` alongside the block at `client.py:197-218`:

```python theme={null}
from .raw_data import RawDataClient
...
self.raw_data = RawDataClient(self)
```

* [ ] **Step 6: Run — verify pass.** → PASS.

* [ ] **Step 7: Commit**

```bash theme={null}
git add packages/ionworks-api/ionworks/raw_data.py packages/ionworks-api/ionworks/models.py packages/ionworks-api/ionworks/client.py packages/ionworks-api/ionworks/__init__.py packages/ionworks-api/tests/test_raw_data.py
git commit -m "feat(sdk): add RawDataClient for raw-data upload/list/get/update/delete"
```

***

## Task 10: SDK — measurement attach/detach (TDD)

**Files:**

* Modify: `packages/ionworks-api/ionworks/cell_measurement.py`

* Test: extend `packages/ionworks-api/tests/test_raw_data.py` or a new test file.

* [ ] **Step 1: Write failing tests**

Assert `measurement.attach_raw_data(mid, [r1, r2])` POSTs `/cell_measurements/{mid}/raw_data` with `{"raw_data_ids": [r1, r2]}`; `detach_raw_data(mid, r1)` DELETEs `/cell_measurements/{mid}/raw_data/{r1}`; `list_raw_data(mid)` GETs and parses `PaginatedList[RawData]`.

* [ ] **Step 2: Run — verify fail.** → FAIL.

* [ ] **Step 3: Implement on `CellMeasurementClient`**

```python theme={null}
from .models import PaginatedList, RawData, _build_endpoint, _parse_list_response

def attach_raw_data(self, cell_measurement_id: str, raw_data_ids: list[str]) -> None:
    """Attach raw-data records to this measurement (bulk, idempotent)."""
    self.client.post(
        f"/cell_measurements/{cell_measurement_id}/raw_data",
        {"raw_data_ids": raw_data_ids},
    )

def detach_raw_data(self, cell_measurement_id: str, raw_data_id: str) -> None:
    """Remove one raw-data link from this measurement."""
    self.client.delete(
        f"/cell_measurements/{cell_measurement_id}/raw_data/{raw_data_id}"
    )

def list_raw_data(
    self, cell_measurement_id: str, limit: int = 100, offset: int = 0
) -> PaginatedList[RawData]:
    """List raw-data records linked to this measurement."""
    endpoint = _build_endpoint(
        f"/cell_measurements/{cell_measurement_id}/raw_data",
        {"limit": limit, "offset": offset},
    )
    return _parse_list_response(self.client.get(endpoint), RawData)
```

**Implementer note:** match the existing import style in `cell_measurement.py` (it already imports from `.models`; extend that import rather than adding a duplicate line).

* [ ] **Step 4: Run — verify pass + full SDK suite**

Run: `cd packages/ionworks-api && uv run python -m pytest -q`
Expected: all green.

* [ ] **Step 5: Commit**

```bash theme={null}
git add packages/ionworks-api/ionworks/cell_measurement.py packages/ionworks-api/tests/
git commit -m "feat(sdk): measurement attach/detach/list raw-data provenance methods"
```

***

## Task 11: Sync SDK skills

**Files:** `packages/skills/skills/discover-api/SKILL.md`, `packages/skills/skills/upload-data/SKILL.md`, and the dev-sync skill's own file list.

* [ ] **Step 1: Invoke the `ionworks-dev-sync` skill**

Follow its checklist to add the `client.raw_data` sub-client row + hierarchy to `discover-api`, and the raw-data upload + measurement attach/detach flow to `upload-data`. Keep edits within the skills' own directories only (per memory: skill edits scope).

* [ ] **Step 2: Commit**

```bash theme={null}
git add packages/skills/skills/discover-api/SKILL.md packages/skills/skills/upload-data/SKILL.md packages/ionworks-api/.claude/skills/ionworks-dev-sync/SKILL.md
git commit -m "docs(skills): document raw-data SDK client and measurement attach"
```

***

## Task 12: Verify end-to-end + finalize

* [ ] **Step 1: Full Python suite**

Run: `just test-python` (packages + backend). Expected: green.

* [ ] **Step 2: Manual smoke via SDK against local backend** (optional but recommended)

With the local backend + Supabase running, write a scratch script: create a project → `client.raw_data.upload(project_id, <tmpfile>)` → `list` → `download_url` → create a measurement → `measurement.attach_raw_data(mid, [rid])` → `measurement.list_raw_data(mid)` shows it → `detach` → gone. Confirm the file exists in the `raw-data` bucket at `{org}/{id}/{filename}`.

* [ ] **Step 3: prek + open PR**

Invoke `prek` on the whole diff; then use the `git-workflow` skill to open a PR titled `feat(raw-data): org+project raw data storage (backend + SDK)`. Reference the spec. Apply the `status: awaiting human review` label via `github-pr-labels`.

***

## Notes / Deferred (from spec §8)

* Table RLS reuses `cell_measurement:*`; storage RLS uses `storage:*` (every bucket does). Join-table RLS gates via the linked `raw_data` row's org. Confirmed in Task 1.
* Org-only RLS — project is a filter dimension, not a DB boundary (confirmed by user).
* No web UI, no auto-processing this phase.
* **Org-guard verified:** `cell_measurements.organization_id` IS a real column
  (added `20260429000001`, NOT NULL in `20260501000003`); `get_by_id`'s `*`
  select populates it and `AnalysisService` uses the identical
  `measurement.organization_id != organization_id` guard. The attach/detach
  guards in Task 5 are correct as written.