fix(attribution): keep owner backfill tenant-scoped

Author: kyle-compute

backend/alembic/versions/expownercreate001_backfill_creation_owner.py

@@ -19,12 +19,15 @@
 def upgrade() -> None:
     # Include deleted and superseded trial history deliberately. Only the exact
     # first trial can identify an experiment's creator; selecting the earliest
-    # later trial with billing data could assign it to an appender.
+    # later trial with billing data could assign it to an appender. Validate the
+    # first trial's payer only after ranking because billed_user_id is not an FK.
+    # Deleted users remain eligible: creator provenance survives offboarding.
     op.execute(
         """
         WITH earliest_trial AS MATERIALIZED (
             SELECT DISTINCT ON (t.experiment_id)
                 t.experiment_id,
+                t.org_id,
                 t.billed_user_id
             FROM trials AS t
             WHERE t.experiment_id IS NOT NULL
@@ -33,8 +36,11 @@ def upgrade() -> None:
         UPDATE experiments AS e
         SET owner_user_id = earliest_trial.billed_user_id
         FROM earliest_trial
+        JOIN users AS billed_user
+          ON billed_user.id = earliest_trial.billed_user_id
         WHERE e.id = earliest_trial.experiment_id
-          AND earliest_trial.billed_user_id IS NOT NULL
+          AND earliest_trial.org_id = e.org_id
+          AND billed_user.org_id = e.org_id
           AND (
               e.owner_user_id IS NULL
               OR e.owner_user_id = '__unattributed__'

backend/tests/test_experiment_owner_creation_backfill.py

@@ -1,10 +1,19 @@
 from __future__ import annotations
 
 import importlib.util
+import os
 from pathlib import Path
 
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
 
-def test_backfill_uses_the_exact_first_trial_including_history() -> None:
+
+DB_URL = os.environ.get("ODDISH_DATABASE_URL")
+requires_db = pytest.mark.skipif(not DB_URL, reason="ODDISH_DATABASE_URL not set")
+
+
+def _upgrade_sql() -> str:
     migration_path = (
         Path(__file__).resolve().parents[1]
         / "alembic/versions/expownercreate001_backfill_creation_owner.py"
@@ -24,10 +33,100 @@ def test_backfill_uses_the_exact_first_trial_including_history() -> None:
     finally:
         module.op.execute = original_execute
 
-    sql = statements[0]
+    assert len(statements) == 1
+    return statements[0]
+
+
+def test_backfill_uses_the_exact_first_trial_including_history() -> None:
+    sql = _upgrade_sql()
     assert "DISTINCT ON (t.experiment_id)" in sql
     assert "ORDER BY t.experiment_id, t.created_at ASC, t.id ASC" in sql
-    assert "earliest_trial.billed_user_id IS NOT NULL" in sql
     assert "owner_user_id IS NULL" in sql
     assert "t.deleted_at" not in sql
     assert "superseded_by_trial_id" not in sql
+
+
+def test_backfill_only_accepts_a_first_trial_payer_from_the_experiment_org() -> None:
+    sql = _upgrade_sql()
+
+    # Validate only after selecting the exact first trial. Joining users inside
+    # the CTE would silently fall forward to a later trial when the first trial
+    # has stale or cross-tenant billing attribution.
+    first_trial_query, update_query = sql.split("UPDATE experiments AS e", maxsplit=1)
+    assert "users" not in first_trial_query
+    assert "JOIN users AS billed_user" in update_query
+    assert "billed_user.id = earliest_trial.billed_user_id" in update_query
+    assert "earliest_trial.org_id = e.org_id" in update_query
+    assert "billed_user.org_id = e.org_id" in update_query
+
+
+@requires_db
+@pytest.mark.asyncio
+async def test_backfill_leaves_cross_org_attribution_unowned() -> None:
+    assert DB_URL is not None
+    engine = create_async_engine(DB_URL)
+    try:
+        async with engine.begin() as conn:
+            # Temporary tables shadow the real schema for this connection, so
+            # this runs the migration SQL without mutating shared test data.
+            statements = (
+                """
+                CREATE TEMP TABLE experiments (
+                        id TEXT PRIMARY KEY,
+                        org_id TEXT NOT NULL,
+                        owner_user_id TEXT
+                )
+                """,
+                """
+                CREATE TEMP TABLE trials (
+                        id TEXT PRIMARY KEY,
+                        experiment_id TEXT,
+                        org_id TEXT NOT NULL,
+                        billed_user_id TEXT,
+                        created_at TIMESTAMPTZ NOT NULL
+                )
+                """,
+                """
+                CREATE TEMP TABLE users (
+                        id TEXT PRIMARY KEY,
+                        org_id TEXT NOT NULL
+                )
+                """,
+                """
+                INSERT INTO users (id, org_id) VALUES
+                    ('user-a', 'org-a'),
+                    ('user-b', 'org-b')
+                """,
+                """
+                INSERT INTO experiments (id, org_id) VALUES
+                    ('valid', 'org-a'),
+                    ('foreign-payer', 'org-a'),
+                    ('foreign-trial', 'org-a'),
+                    ('no-fall-forward', 'org-a')
+                """,
+                """
+                INSERT INTO trials (
+                    id, experiment_id, org_id, billed_user_id, created_at
+                ) VALUES
+                    ('t-valid', 'valid', 'org-a', 'user-a', '2026-01-01'),
+                    ('t-payer', 'foreign-payer', 'org-a', 'user-b', '2026-01-01'),
+                    ('t-trial', 'foreign-trial', 'org-b', 'user-a', '2026-01-01'),
+                    ('t-first', 'no-fall-forward', 'org-a', 'user-b', '2026-01-01'),
+                    ('t-later', 'no-fall-forward', 'org-a', 'user-a', '2026-01-02')
+                """,
+            )
+            for statement in statements:
+                await conn.execute(text(statement))
+            await conn.execute(text(_upgrade_sql()))
+            result = await conn.execute(
+                text("SELECT id, owner_user_id FROM experiments ORDER BY id")
+            )
+
+        assert dict(result.all()) == {
+            "foreign-payer": None,
+            "foreign-trial": None,
+            "no-fall-forward": None,
+            "valid": "user-a",
+        }
+    finally:
+        await engine.dispose()