Fix cross-user collision in case-migration session fingerprinting

Found via live multi-user testing: case_migrations.source_fingerprint
was UNIQUE on its own (org-scoped, not per-user), so two org members
who both stage the local-to-shared migration with identical local case
content (most commonly zero local cases, the normal state for anyone
new to a shared org) silently collided onto the same session. The
second caller would inherit the first caller's session -- sometimes
already validated or activated -- and their own "Activate shared
dataset" button would never appear.

Scopes the dedup key to (source_fingerprint, created_by) instead,
matching the same idempotent-retry behavior for one user while no
longer colliding across different users. Fixed in both the Postgres
and in-memory stores, with a migration to update the constraint (and
backfill every already-provisioned tenant schema) plus a regression
test.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: voidstackloop

server/migrations/022_case_migration_fingerprint_scope.sql

@@ -0,0 +1,171 @@
+-- Fixes a real cross-user bug found via live multi-user testing of the
+-- local-to-shared case migration flow: case_migrations.source_fingerprint
+-- (sha256 of the caller's local case list) was UNIQUE on its own, with no
+-- per-user scoping. Two different members of the same organization who
+-- both stage with identical local case content — most commonly, both
+-- having zero local cases, the ordinary state for anyone new to a shared
+-- org — collide onto the exact same migration session via the store's
+-- `ON CONFLICT (source_fingerprint) DO UPDATE`. The second caller silently
+-- inherits the first caller's session (possibly already validated or even
+-- activated), and their own client never recognizes it as a session they
+-- can activate — case-migration.ts's stageLocalCases() only proceeds past
+-- "resume" for a session with a preview, and the "Activate shared dataset"
+-- button (AuditPrivacy.tsx) only renders for status "validated" — so the
+-- second user's own activation silently never becomes available even
+-- though the request nominally "succeeded."
+--
+-- Fix: scope the same idempotency-on-retry behavior (a caller repeating
+-- their own start() call with the same content reuses their own session,
+-- unchanged) to (source_fingerprint, created_by) instead of
+-- source_fingerprint alone. See postgres-case-migration-store.ts's start()
+-- and in-memory-case-migration-store.ts's start() for the matching
+-- application-side fix.
+--
+-- Same two-part structure as every migration since 015: Part 1 replaces
+-- provision_tenant_clinical_schema so every *future* organization's
+-- case_migrations table is created with the corrected constraint; Part 2
+-- backfills every already-provisioned tenant schema in place. Every block
+-- in Part 1 below is copied verbatim from migration 018's version of this
+-- function except the one UNIQUE clause called out inline.
+
+-- --- Part 1: extend provision_tenant_clinical_schema -----------------------
+
+CREATE OR REPLACE FUNCTION provision_tenant_clinical_schema(target_org UUID)
+RETURNS TEXT
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = pg_catalog, public
+AS $provision$
+DECLARE
+    schema_name TEXT := 'tenant_' || replace(target_org::text, '-', '');
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM public.organizations WHERE id = target_org) THEN
+        RAISE EXCEPTION 'Unknown organization';
+    END IF;
+    UPDATE public.organizations SET tenant_schema = schema_name WHERE id = target_org;
+    EXECUTE format('CREATE SCHEMA IF NOT EXISTS %I', schema_name);
+    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'modelforge_runtime') THEN
+        EXECUTE format('GRANT USAGE ON SCHEMA %I TO modelforge_runtime', schema_name);
+        EXECUTE format(
+            'ALTER DEFAULT PRIVILEGES FOR ROLE %I IN SCHEMA %I GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO modelforge_runtime',
+            CURRENT_USER, schema_name
+        );
+    END IF;
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.patient_cases (
+            case_id TEXT PRIMARY KEY,
+            version BIGINT NOT NULL,
+            data JSONB NOT NULL,
+            patient_id TEXT NOT NULL,
+            owner_user_id UUID NOT NULL,
+            workspace_id TEXT,
+            department_id TEXT,
+            assigned_user_ids UUID[] NOT NULL DEFAULT ''{}'',
+            active_consent_scopes TEXT[] NOT NULL DEFAULT ''{}'',
+            staged_migration_id UUID,
+            active BOOLEAN NOT NULL DEFAULT TRUE,
+            updated_at TIMESTAMPTZ NOT NULL
+        )', schema_name
+    );
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.case_change_counter (
+            singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton),
+            next_sequence BIGINT NOT NULL DEFAULT 1
+        )', schema_name
+    );
+    EXECUTE format('INSERT INTO %I.case_change_counter (singleton, next_sequence) VALUES (TRUE, 1) ON CONFLICT DO NOTHING', schema_name);
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.case_changes (
+            sequence BIGINT PRIMARY KEY,
+            kind TEXT NOT NULL CHECK (kind IN (''upsert'', ''delete'')),
+            case_id TEXT NOT NULL,
+            version BIGINT NOT NULL,
+            patient_case JSONB,
+            resource JSONB NOT NULL,
+            changed_at TIMESTAMPTZ NOT NULL
+        )', schema_name
+    );
+    EXECUTE format('ALTER TABLE %I.case_changes ADD COLUMN IF NOT EXISTS resource JSONB', schema_name);
+    EXECUTE format('CREATE INDEX IF NOT EXISTS %I ON %I.case_changes (case_id, sequence DESC)', 'idx_' || schema_name || '_case_changes_case', schema_name);
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.case_migrations (
+            id UUID PRIMARY KEY,
+            status TEXT NOT NULL CHECK (status IN (''staging'', ''validated'', ''active'', ''rolled-back'')),
+            source_fingerprint TEXT NOT NULL,
+            total_items INTEGER NOT NULL,
+            accepted_items INTEGER NOT NULL DEFAULT 0,
+            preview JSONB,
+            created_by UUID NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL,
+            updated_at TIMESTAMPTZ NOT NULL,
+            UNIQUE (source_fingerprint, created_by)
+        )', schema_name
+    );
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.case_migration_items (
+            migration_id UUID NOT NULL REFERENCES %I.case_migrations(id) ON DELETE CASCADE,
+            item_key TEXT NOT NULL,
+            case_id TEXT NOT NULL,
+            data JSONB NOT NULL,
+            data_hash TEXT NOT NULL,
+            status TEXT NOT NULL CHECK (status IN (''pending'', ''accepted'', ''invalid'', ''collision'')),
+            errors JSONB NOT NULL DEFAULT ''[]'',
+            PRIMARY KEY (migration_id, item_key)
+        )', schema_name, schema_name
+    );
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.chat_sessions (
+            id TEXT PRIMARY KEY,
+            version BIGINT NOT NULL,
+            data JSONB NOT NULL,
+            owner_user_id UUID NOT NULL,
+            assigned_user_ids UUID[] NOT NULL DEFAULT ''{}'',
+            updated_at TIMESTAMPTZ NOT NULL
+        )', schema_name
+    );
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.chat_session_change_counter (
+            singleton BOOLEAN PRIMARY KEY DEFAULT TRUE CHECK (singleton),
+            next_sequence BIGINT NOT NULL DEFAULT 1
+        )', schema_name
+    );
+    EXECUTE format('INSERT INTO %I.chat_session_change_counter (singleton, next_sequence) VALUES (TRUE, 1) ON CONFLICT DO NOTHING', schema_name);
+    EXECUTE format(
+        'CREATE TABLE IF NOT EXISTS %I.chat_session_changes (
+            sequence BIGINT PRIMARY KEY,
+            kind TEXT NOT NULL CHECK (kind IN (''upsert'', ''delete'')),
+            session_id TEXT NOT NULL,
+            version BIGINT NOT NULL,
+            session_data JSONB,
+            resource JSONB NOT NULL,
+            changed_at TIMESTAMPTZ NOT NULL
+        )', schema_name
+    );
+    EXECUTE format('CREATE INDEX IF NOT EXISTS %I ON %I.chat_session_changes (session_id, sequence DESC)', 'idx_' || schema_name || '_chat_session_changes_session', schema_name);
+
+    PERFORM provision_tenant_imaging_tables(schema_name);
+    PERFORM provision_tenant_ai_gateway_tables(schema_name);
+
+    RETURN schema_name;
+END
+$provision$;
+
+REVOKE ALL ON FUNCTION provision_tenant_clinical_schema(UUID) FROM PUBLIC;
+
+-- --- Part 2: backfill every already-provisioned tenant schema --------------
+DO $backfill$
+DECLARE
+    org RECORD;
+BEGIN
+    FOR org IN SELECT tenant_schema FROM organizations WHERE tenant_schema IS NOT NULL LOOP
+        -- "case_migrations_source_fingerprint_key" is Postgres's default
+        -- autogenerated name for the original inline UNIQUE (source_fingerprint)
+        -- column constraint declared in migrations 009 through 018.
+        EXECUTE format('ALTER TABLE %I.case_migrations DROP CONSTRAINT IF EXISTS case_migrations_source_fingerprint_key', org.tenant_schema);
+        EXECUTE format(
+            'ALTER TABLE %I.case_migrations ADD CONSTRAINT case_migrations_source_fingerprint_created_by_key UNIQUE (source_fingerprint, created_by)',
+            org.tenant_schema
+        );
+    END LOOP;
+END
+$backfill$;

server/src/store/in-memory-case-migration-store.ts

@@ -7,7 +7,7 @@ import { MigrationStateError, type CaseMigrationStore, type TenantCaseMigrationR
 import type { TenantCaseRepository } from "./case-store.js";
 
 interface StagedItem { itemKey: string; patientCase: unknown; hash: string; status: "pending" | "accepted" | "invalid" | "collision"; errors: string[] }
-interface State { session: MigrationSession; items: Map<string, StagedItem>; activated: Map<string, string> }
+interface State { session: MigrationSession; items: Map<string, StagedItem>; activated: Map<string, string>; createdBy: string | undefined }
 const digest = (value: unknown): string => createHash("sha256").update(JSON.stringify(value)).digest("hex");
 
 export class InMemoryCaseMigrationStore implements CaseMigrationStore {
@@ -24,11 +24,17 @@ export class InMemoryCaseMigrationStore implements CaseMigrationStore {
         const requireState = (id: string): State => { const state = sessions.get(id); if (!state) throw Object.assign(new Error("Migration not found."), { statusCode: 404 }); return state; };
         return {
             start: async (input, actor) => {
-                const existing = [...sessions.values()].find((state) => state.session.sourceFingerprint === input.sourceFingerprint && state.session.status !== "rolled-back");
+                // Scoped by creator as well as content — see the Postgres
+                // store's identical fix and migrations/022_case_migration_fingerprint_scope.sql
+                // for why sourceFingerprint alone is not a safe dedup key
+                // across different users.
+                const existing = [...sessions.values()].find(
+                    (state) => state.session.sourceFingerprint === input.sourceFingerprint && state.createdBy === actor.userId && state.session.status !== "rolled-back"
+                );
                 if (existing) return existing.session;
                 const now = new Date().toISOString();
                 const session: MigrationSession = { id: randomUUID(), organizationId: context.organizationId, status: "staging", sourceFingerprint: input.sourceFingerprint, totalItems: input.totalItems, acceptedItems: 0, createdAt: now, updatedAt: now };
-                sessions.set(session.id, { session, items: new Map(), activated: new Map() });
+                sessions.set(session.id, { session, items: new Map(), activated: new Map(), createdBy: actor.userId });
                 await this.audit.record({ organizationId: context.organizationId, actorUserId: actor.userId, actorExternalSubject: actor.externalSubject, action: "caseMigration.start", targetType: "caseMigration", targetId: session.id });
                 return session;
             },

server/src/store/postgres-case-migration-store.test.ts

@@ -78,6 +78,24 @@ describe.skipIf(!DATABASE_URL)("PostgresCaseMigrationStore (integration — requ
         expect(rows.rows[0].n).toBe(1);
     });
 
+    it("start() scopes the fingerprint dedup key per user — two org members with identical local case content (e.g. both empty) get distinct sessions, not the same one", async () => {
+        const otherActor: AuditActor = { externalSubject: "idp|test-actor-2", userId: randomUUID() };
+        const mine = await repo.start({ sourceFingerprint: "identical-content", totalItems: 0 }, ACTOR);
+        const theirs = await repo.start({ sourceFingerprint: "identical-content", totalItems: 0 }, otherActor);
+        expect(theirs.id).not.toBe(mine.id);
+
+        // Regression coverage for the real bug: activating MY session must
+        // never affect the OTHER actor's still-independent session — before
+        // this fix, they were the exact same row, so this validate() would
+        // have failed with "cannot be validated in its current state" once
+        // the shared session moved to "active".
+        await repo.validate(mine.id, ACTOR);
+        const activated = await repo.activate(mine.id, ACTOR);
+        expect(activated.status).toBe("active");
+        const theirsStillStaging = await repo.get(theirs.id);
+        expect(theirsStillStaging?.status).toBe("staging");
+    });
+
     it("upload() is idempotent per item key — replaying the same batch does not duplicate items, but reusing a key with different data is rejected", async () => {
         const session = await repo.start({ sourceFingerprint: "fp-upload", totalItems: 1 }, ACTOR);
         const original = patientCaseFixture("case-1");

server/src/store/postgres-case-migration-store.ts

@@ -29,7 +29,16 @@ export class PostgresCaseMigrationStore implements CaseMigrationStore {
     private async begin(client: PoolClient, context: TenantContext): Promise<void> { await client.query("BEGIN"); await client.query("SELECT set_config('app.tenant_id',$1,true)",[context.organizationId]); }
     private async get(context: TenantContext, schema: string, id: string): Promise<MigrationSession | null> { const result = await this.pool.query<SessionRow>(`SELECT * FROM ${schema}.case_migrations WHERE id=$1`,[id]); return result.rows[0] ? mapSession(result.rows[0],context.organizationId) : null; }
     private async start(context: TenantContext, schema: string, input: { sourceFingerprint: string; totalItems: number }, actor: AuditActor): Promise<MigrationSession> {
-        const client=await this.pool.connect(); try { await this.begin(client,context); const now=new Date(); const result=await client.query<SessionRow>(`INSERT INTO ${schema}.case_migrations (id,status,source_fingerprint,total_items,accepted_items,created_by,created_at,updated_at) VALUES ($1,'staging',$2,$3,0,$4,$5,$5) ON CONFLICT (source_fingerprint) DO UPDATE SET source_fingerprint=EXCLUDED.source_fingerprint RETURNING *`,[randomUUID(),input.sourceFingerprint,input.totalItems,actor.userId,now]); await insertAuditEntry(client,{organizationId:context.organizationId,actorUserId:actor.userId,actorExternalSubject:actor.externalSubject,action:"caseMigration.start",targetType:"caseMigration",targetId:result.rows[0].id}); await client.query("COMMIT"); return mapSession(result.rows[0],context.organizationId); } catch(e){await client.query("ROLLBACK");throw e;} finally{client.release();}
+        // The conflict target is (source_fingerprint, created_by), not
+        // source_fingerprint alone — see migrations/022_case_migration_fingerprint_scope.sql.
+        // A bare source_fingerprint key means any two org members (or the
+        // same user across two separate attempts, since sha256("[]") is
+        // identical every time for someone with zero local cases) who stage
+        // with identical local case content collide onto the exact same
+        // session, silently handing the second caller a session — possibly
+        // already validated or activated — that they never started and that
+        // their own client's Activate button won't recognize as theirs.
+        const client=await this.pool.connect(); try { await this.begin(client,context); const now=new Date(); const result=await client.query<SessionRow>(`INSERT INTO ${schema}.case_migrations (id,status,source_fingerprint,total_items,accepted_items,created_by,created_at,updated_at) VALUES ($1,'staging',$2,$3,0,$4,$5,$5) ON CONFLICT (source_fingerprint, created_by) DO UPDATE SET source_fingerprint=EXCLUDED.source_fingerprint RETURNING *`,[randomUUID(),input.sourceFingerprint,input.totalItems,actor.userId,now]); await insertAuditEntry(client,{organizationId:context.organizationId,actorUserId:actor.userId,actorExternalSubject:actor.externalSubject,action:"caseMigration.start",targetType:"caseMigration",targetId:result.rows[0].id}); await client.query("COMMIT"); return mapSession(result.rows[0],context.organizationId); } catch(e){await client.query("ROLLBACK");throw e;} finally{client.release();}
     }
     private async upload(context: TenantContext, schema: string, id: string, items: { itemKey:string;patientCase:unknown }[], actor: AuditActor): Promise<MigrationSession> {
         const client=await this.pool.connect(); try { await this.begin(client,context); const session=await client.query<SessionRow>(`SELECT * FROM ${schema}.case_migrations WHERE id=$1 FOR UPDATE`,[id]); if(!session.rows[0]) throw Object.assign(new Error("Migration not found."),{statusCode:404}); if(session.rows[0].status!=="staging") throw new MigrationStateError("Migration is not accepting batches.");