project_id: str | None = Field(
        default=None,
        description="ID of the project that owns this protocol (None for the system catalog)",
    )
    access_level: str = Field(
        default="project",
        description="'system' for catalog rows, 'project' for project-owned protocols",
    )

    project_id: str | None = Field(
        default=None,
        description="ID of the project that owns this experiment (inherited from its template)",
    )

git add backend/src/pydantic_models/experiments.py
git commit -m "feat(protocols): add project_id/access_level to experiment models"

-- Expand phase: add nullable project_id + access_level to experiment_templates
-- and nullable project_id to experiments, scoping protocols to a project.
-- Mirrors the optimization_templates / jobs access_level pattern. A follow-up
-- contract migration (separate PR, ≥1 month later) validates the FKs and sets
-- project_id NOT NULL once ops task 0022 has run in every environment.

-- ── experiment_templates ────────────────────────────────────────────────────
ALTER TABLE public.experiment_templates
    ADD COLUMN IF NOT EXISTS project_id uuid;

-- Default 'system' (NOT 'project') is load-bearing: the only pre-existing rows
-- are the system-org catalog with project_id NULL, which satisfy only the
-- 'system' arm of the invariant CHECK below. Defaulting to 'project' would make
-- the CHECK fail on add. New project-scoped writes set 'project' explicitly.
ALTER TABLE public.experiment_templates
    ADD COLUMN IF NOT EXISTS access_level text NOT NULL DEFAULT 'system';

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_constraint
        WHERE conname = 'experiment_templates_access_level_check'
    ) THEN
        ALTER TABLE public.experiment_templates
            ADD CONSTRAINT experiment_templates_access_level_check
            CHECK (access_level IN ('system', 'project'));
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_constraint
        WHERE conname = 'experiment_templates_access_level_project_id_check'
    ) THEN
        ALTER TABLE public.experiment_templates
            ADD CONSTRAINT experiment_templates_access_level_project_id_check
            CHECK (
                (access_level = 'project' AND project_id IS NOT NULL)
                OR (access_level = 'system' AND project_id IS NULL)
            );
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_constraint
        WHERE conname = 'experiment_templates_project_id_organization_id_fkey'
    ) THEN
        ALTER TABLE public.experiment_templates
            ADD CONSTRAINT experiment_templates_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 $$;

CREATE INDEX IF NOT EXISTS idx_experiment_templates_project_id
    ON public.experiment_templates (project_id);

-- Project-aware dedup: each project holds its own copy; a single system slot
-- (project_id NULL) stays deduped via NULLS NOT DISTINCT, and the fixed column
-- list stays targetable by PostgREST on_conflict.
DROP INDEX IF EXISTS public.idx_experiment_templates_content_hash_org;
CREATE UNIQUE INDEX IF NOT EXISTS idx_experiment_templates_content_hash_org_proj
    ON public.experiment_templates (content_hash, organization_id, project_id)
    NULLS NOT DISTINCT;

-- ── experiments ──────────────────────────────────────────────────────────────
ALTER TABLE public.experiments
    ADD COLUMN IF NOT EXISTS project_id uuid;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_constraint
        WHERE conname = 'experiments_project_id_organization_id_fkey'
    ) THEN
        ALTER TABLE public.experiments
            ADD CONSTRAINT experiments_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 $$;

CREATE INDEX IF NOT EXISTS idx_experiments_project_id
    ON public.experiments (project_id);

DROP INDEX IF EXISTS public.experiments_template_params_org_unique;
CREATE UNIQUE INDEX IF NOT EXISTS experiments_template_params_org_proj_unique
    ON public.experiments (template_id, parameters, organization_id, project_id)
    NULLS NOT DISTINCT;

-- ── project-aware canonical find ──────────────────────────────────────────────
-- The old signature keyed on (content_hash, organization_id) only, so after the
-- backfill (identical content_hash across project copies) it returned an
-- arbitrary project's row. Add in_project_id; IS NOT DISTINCT FROM matches the
-- NULL catalog slot when passed NULL and only the exact project row otherwise.
CREATE OR REPLACE FUNCTION public.experiment_templates_find_by_content(
    in_protocol_config jsonb,
    in_parameters_schema jsonb,
    in_organization_id uuid,
    in_project_id uuid
) RETURNS SETOF public.experiment_templates
    LANGUAGE sql STABLE
    AS $$
  SELECT *
  FROM public.experiment_templates et
  WHERE et.content_hash = public.generate_experiment_template_content_hash(
          in_protocol_config, in_parameters_schema)
    AND et.organization_id = in_organization_id
    AND et.project_id IS NOT DISTINCT FROM in_project_id
  LIMIT 1;
$$;

DROP FUNCTION IF EXISTS public.experiment_templates_find_by_content(jsonb, jsonb, uuid);

GRANT EXECUTE ON FUNCTION public.experiment_templates_find_by_content(jsonb, jsonb, uuid, uuid)
    TO anon, authenticated, service_role;

psql "$SUPABASE_DB_URL" -c "\d public.experiment_templates" | grep -E "access_level|project_id"
psql "$SUPABASE_DB_URL" -c "\df public.experiment_templates_find_by_content"

git add supabase/migrations/20260728174630_add_project_id_to_protocols.sql
git commit -m "feat(protocols): expand migration — project_id, access_level, project-aware dedup + RPC"

@pytest.mark.asyncio
async def test_find_or_create_threads_project_id_into_rpc():
    """project_id is passed to the RPC and into the create payload."""
    repo = _make_repo()
    _wire_rpc(repo, [MagicMock(data=[])])  # no existing → create path
    repo.model_class = MagicMock(side_effect=lambda **kw: MagicMock(**kw))
    created = MagicMock(id="tmpl-new")
    repo.create = AsyncMock(return_value=created)

    await repo.find_or_create_canonical_template(
        name="P",
        protocol_config={"steps": []},
        parameters_schema={},
        organization_id="org-1",
        project_id="proj-1",
    )

    rpc_kwargs = repo.supabase.rpc.call_args.args[1]
    assert rpc_kwargs["in_project_id"] == "proj-1"
    create_payload = repo.create.call_args.args[0]
    assert create_payload["project_id"] == "proj-1"
    assert create_payload["access_level"] == "project"

git add backend/src/repositories/experiment_templates.py backend/tests/test_repositories/test_experiment_templates_conflict.py
git commit -m "feat(protocols): thread project_id through canonical find-or-create"

LIST_DEFAULT_COLUMNS = (
    "id, name, description, description_template, organization_id,"
    " project_id, access_level, created_at, updated_at, created_by"
)

    async def get_templates_by_project_with_email(
        self,
        organization_id: str,
        project_id: str,
        *,
        limit: int | None = None,
        offset: int | None = None,
        include: set[str] | None = None,
    ) -> tuple[list[ExperimentTemplate], int]:
        """List a project's protocols (access_level='project'), with creator email."""
        extra = sorted((include or set()) & INCLUDABLE_COLUMNS)
        select_cols = LIST_DEFAULT_COLUMNS
        if extra:
            select_cols = f"{select_cols}, {', '.join(extra)}"
        select_cols = f"{select_cols}, users(email)"

        query = (
            self.supabase.table(self.table_name)
            .select(select_cols, count="exact")
            .eq("organization_id", organization_id)
            .eq("project_id", project_id)
            .eq("access_level", "project")
            .order("created_at", desc=True)
        )
        if limit is not None and offset is not None:
            query = query.range(offset, offset + limit - 1)
        elif limit is not None:
            query = query.range(0, limit - 1)

        response = await query.execute()
        templates = [
            self.model_class(**self._extract_created_by_email(row))
            for row in (response.data or [])
        ]
        return templates, response.count or 0

git add backend/src/repositories/experiment_templates.py
git commit -m "feat(protocols): project-scoped template listing + expose project_id/access_level"

cd backend && uv run python - <<'PY'
import asyncio
from src.deps import get_supabase_async_service_client  # or however the service client is built in a script
# insert one catalog copy for a real (org, project) via
#   .upsert([{...no content_hash...}], on_conflict="content_hash,organization_id,project_id", ignore_duplicates=True)
# then run it a SECOND time and assert the row count did not grow.
PY

"""Tests for copy_catalog_protocols_into_project."""

from unittest.mock import AsyncMock, MagicMock

import pytest
from src.services.protocol_catalog import copy_catalog_protocols_into_project

SYSTEM_ORG = "00000000-0000-0000-0000-000000000001"


def _catalog_row(name: str) -> dict:
    return {
        "id": f"cat-{name}",
        "name": name,
        "protocol_config": {"steps": []},
        "parameters_schema": {},
        "time_series_spec": {},
        "metrics_spec": {},
        "plot_options": {},
        "description": None,
        "source_protocol": None,
    }


@pytest.mark.asyncio
async def test_copies_each_catalog_row_scoped_to_project():
    service = MagicMock()
    # catalog read returns two system rows
    read = AsyncMock(return_value=MagicMock(data=[_catalog_row("Charge"), _catalog_row("Discharge")]))
    service.table.return_value.select.return_value.eq.return_value.eq.return_value.execute = read
    insert_exec = AsyncMock(return_value=MagicMock(data=[{"id": "new"}]))
    service.table.return_value.upsert.return_value.execute = insert_exec

    n = await copy_catalog_protocols_into_project(service, "org-1", "proj-1")

    assert n == 2
    # each inserted row is project-scoped
    payloads = [c.args[0] for c in service.table.return_value.upsert.call_args_list]
    flat = [row for p in payloads for row in (p if isinstance(p, list) else [p])]
    assert all(r["project_id"] == "proj-1" for r in flat)
    assert all(r["organization_id"] == "org-1" for r in flat)
    assert all(r["access_level"] == "project" for r in flat)

"""Copy the built-in protocol catalog into a project.

The catalog is the set of system experiment_templates
(``access_level='system'``, ``organization_id`` = the system org). Each project
gets its own editable copies on creation and via the backfill ops task. The copy
is keyed so replays are idempotent against the project-aware dedup index.
"""

import logging

from supabase import AsyncClient

logger = logging.getLogger(__name__)

SYSTEM_ORGANIZATION_ID = "00000000-0000-0000-0000-000000000001"

_COPY_FIELDS = (
    "name",
    "protocol_config",
    "parameters_schema",
    "time_series_spec",
    "metrics_spec",
    "plot_options",
    "description",
    "source_protocol",
)


async def copy_catalog_protocols_into_project(
    service_client: AsyncClient,
    organization_id: str,
    project_id: str,
    created_by: str | None = None,
) -> int:
    """Copy every catalog protocol into ``project_id`` under ``organization_id``.

    Uses the **service** client because catalog rows live in the system org and a
    user client cannot read them. Idempotent: an ``upsert`` with
    ``ignore_duplicates`` against the project-aware unique index makes a replay a
    no-op (a same-content row already present for this (org, project) is skipped).

    Parameters
    ----------
    service_client : AsyncClient
        Supabase service-role client.
    organization_id : str
        Org that will own the copies (the project's org).
    project_id : str
        Destination project.
    created_by : str | None, optional
        User id to stamp on the copies. Defaults to None.

    Returns
    -------
    int
        Number of catalog rows processed (copied or already present).
    """
    catalog = (
        await service_client.table("experiment_templates")
        .select(", ".join(("id",) + _COPY_FIELDS))
        .eq("organization_id", SYSTEM_ORGANIZATION_ID)
        .eq("access_level", "system")
        .execute()
    )
    rows = catalog.data or []
    if not rows:
        return 0

    payload = [
        {
            **{f: row.get(f) for f in _COPY_FIELDS},
            "organization_id": organization_id,
            "project_id": project_id,
            "access_level": "project",
            "created_by": created_by,
        }
        for row in rows
    ]
    # content_hash is recomputed by the BEFORE INSERT trigger; the project-aware
    # unique index (content_hash, organization_id, project_id) NULLS NOT DISTINCT
    # makes ignore_duplicates a safe replay no-op.
    await (
        service_client.table("experiment_templates")
        .upsert(
            payload,
            on_conflict="content_hash,organization_id,project_id",
            ignore_duplicates=True,
        )
        .execute()
    )
    return len(rows)

git add backend/src/services/protocol_catalog.py backend/tests/test_services/test_protocol_catalog.py
git commit -m "feat(protocols): shared catalog-copy helper for project seeding"

"""Seed every existing project with copies of the built-in protocol catalog.

Insert-only (no claim/mutate/delete): the system-org catalog stays intact as the
source of truth through the expand phase. Idempotent — re-running finds every
catalog protocol already present per project and inserts nothing. on_demand so
it is run and reviewed per environment.

Follow-up (separate PR, once run everywhere): the contract migration makes
experiment_templates/experiments project_id NOT NULL and removes the system-org
catalog + union code path.
"""

import asyncio
import logging

from supabase import Client

from src.services.protocol_catalog import (
    SYSTEM_ORGANIZATION_ID,
    copy_catalog_protocols_into_project,
)

RUN_ON = "on_demand"

logger = logging.getLogger(__name__)

_BATCH = 500


def run(supabase: Client) -> None:
    """Stamp the catalog then copy it into every project (see module docstring)."""
    # 1. Defensive re-stamp: any system-org row that isn't labelled 'system'
    #    (should be none after the migration default) is corrected.
    supabase.table("experiment_templates").update({"access_level": "system"}).eq(
        "organization_id", SYSTEM_ORGANIZATION_ID
    ).neq("access_level", "system").execute()

    # 2. Copy the catalog into every project, keyset-paginated by id.
    #    copy_catalog_protocols_into_project is async (built for the AsyncClient
    #    request path); run it on the sync ops client via a fresh event loop.
    projects: list[dict] = []
    cursor: str | None = None
    while True:
        q = (
            supabase.table("projects")
            .select("id, organization_id")
            .order("id")
            .limit(_BATCH)
        )
        if cursor is not None:
            q = q.gt("id", cursor)
        page = q.execute()
        if not page.data:
            break
        projects.extend(page.data)
        cursor = page.data[-1]["id"]
        if len(page.data) < _BATCH:
            break

    for proj in projects:
        _seed_one(supabase, proj["organization_id"], proj["id"])

    logger.info("Seeded protocol catalog into %d projects", len(projects))


def _seed_one(supabase: Client, organization_id: str, project_id: str) -> None:
    """Copy the catalog into one project using a sync PostgREST path."""
    catalog = (
        supabase.table("experiment_templates")
        .select(
            "name, protocol_config, parameters_schema, time_series_spec,"
            " metrics_spec, plot_options, description, source_protocol"
        )
        .eq("organization_id", SYSTEM_ORGANIZATION_ID)
        .eq("access_level", "system")
        .execute()
    )
    rows = catalog.data or []
    if not rows:
        return
    payload = [
        {
            **row,
            "organization_id": organization_id,
            "project_id": project_id,
            "access_level": "project",
        }
        for row in rows
    ]
    supabase.table("experiment_templates").upsert(
        payload,
        on_conflict="content_hash,organization_id,project_id",
        ignore_duplicates=True,
    ).execute()

psql "$SUPABASE_DB_URL" -c "SELECT project_id, count(*) FROM public.experiment_templates WHERE access_level='project' GROUP BY project_id;"

git add backend/src/ops_tasks/0022_seed_project_protocols.py
git commit -m "feat(protocols): ops task 0022 — backfill catalog into existing projects"

@pytest.mark.asyncio
async def test_create_project_seeds_catalog_protocols(monkeypatch, project_service, mock_project_repos):
    # ... wire mock_project_repos so create_project + role assignment succeed
    #     (mirror the existing happy-path test in this file) ...
    seeded = {}
    async def _fake_copy(client, org, proj, created_by=None):
        seeded["args"] = (org, proj)
        return 3
    monkeypatch.setattr(
        "src.services.project_service.copy_catalog_protocols_into_project", _fake_copy
    )

    await project_service.create_project_with_admin("user-1", "org-1", _make_project_data())

    assert seeded["args"][0] == "org-1"  # org threaded through

@pytest.mark.asyncio
async def test_create_project_seed_failure_raises_app_error(monkeypatch, project_service, mock_project_repos):
    # ... happy-path wiring for project + role ...
    async def _boom(*a, **k):
        raise RuntimeError("catalog read failed")
    monkeypatch.setattr(
        "src.services.project_service.copy_catalog_protocols_into_project", _boom
    )
    with pytest.raises(AppError):
        await project_service.create_project_with_admin("user-1", "org-1", _make_project_data())

from src.services.protocol_catalog import copy_catalog_protocols_into_project

        # Seed the project with its own copies of the built-in protocol catalog.
        # Uses the service client because catalog rows live in the system org.
        try:
            await copy_catalog_protocols_into_project(
                self.service_user_project_role_repo.supabase,
                organization_id,
                project_id,
                created_by=user_id,
            )
        except Exception as e:
            logger.exception(
                "Failed to seed protocol catalog for project %s", project_id
            )
            raise AppError(
                f"Project created, but failed to seed protocols: {e}"
            ) from e

git add backend/src/services/project_service.py backend/tests/test_services/test_project_service.py
git commit -m "feat(protocols): seed catalog protocols on project creation"

git add backend/src/repositories/experiments.py
git commit -m "feat(protocols): project-aware experiments upsert arbiter + finder"

    if not body.project_id:
        raise BadRequestError(
            "project_id is required to create a protocol; protocols belong to a project."
        )

route protocols.py:386
  → SimulationService.resolve_template_from_canonical   (simulation_service.py:332, static, no project_id today)
    → SimulationService._create_experiment_from_canonical (simulation_service.py:142)
      → CyclerProtocolService.create_experiment_from_protocol (cycler_protocol_service.py:77)
        → find_or_create_canonical_template (:138) / find_by_template_and_parameters (:153) / create_experiment (:164)

git add backend/src/routes/experiment_templates.py backend/src/routes/protocols.py backend/tests/test_routes/
git commit -m "feat(protocols): require project_id on create, project-filtered listing"

git add frontend/src
git commit -m "fix(protocols): pass project_id on protocol create/parse from in-project UI"