feat(projects): enforce Project membership and retire the connector

.github/scripts/check-project-rollout.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Read-only, fail-closed ECS retirement check for the Project column contract migration.
+
+The expected digest acknowledges fully deployed #8830 transaction-barrier and durable-authority support
+and retirement of every incompatible server and relevant old worker job before switching. Every live and supported
+rollback application must tolerate project_workspace being absent.
+AWS checks below independently verify ECS retirement, not worker drainage.
+"""
+import argparse
+import datetime
+import json
+import re
+import subprocess
+import sys
+
+
+def aws(region, *args):
+    result = subprocess.run(
+        ['aws', '--region', region, '--no-cli-pager', '--cli-connect-timeout', '10', '--cli-read-timeout', '30', *args, '--output', 'json'],
+        capture_output=True, text=True, timeout=90, check=False,
+    )
+    if result.returncode:
+        raise RuntimeError('AWS preflight read failed; check deployment-read permissions')
+    return json.loads(result.stdout)
+
+
+def latest_execution(region, pipeline):
+    executions = aws(region, 'codepipeline', 'list-pipeline-executions', '--pipeline-name', pipeline).get('pipelineExecutionSummaries', [])
+    def epoch(execution):
+        value = execution['startTime']
+        return value if isinstance(value, (int, float)) else datetime.datetime.fromisoformat(value.replace('Z', '+00:00')).timestamp()
+    if not executions:
+        raise RuntimeError('No application deployment execution was found')
+    return max(executions, key=epoch)
+
+
+def matches_release(execution, digest):
+    return execution.get('status') == 'Succeeded' and any(
+        revision.get('actionName') == 'ECR_Source' and revision.get('revisionId') == digest
+        for revision in execution.get('sourceRevisions', [])
+    )
+
+
+def verify(environment, region, digest):
+    if not re.fullmatch(r'sha256:[0-9a-f]{64}', digest):
+        raise RuntimeError('Set the environment-specific PROJECT_COLUMN_ENFORCEMENT_READY_IMAGE_DIGEST after verifying #8830 authority-aware readers/writers, transaction barriers and all incompatible server/worker drainage')
+    pipeline = f'sim-{environment}-{region}-app-deployment'
+    execution = latest_execution(region, pipeline)
+    if not matches_release(execution, digest):
+        raise RuntimeError('The latest app pipeline has not completed for the acknowledged image; traffic cutover alone is insufficient')
+    execution_id = execution['pipelineExecutionId']
+    group = aws(region, 'deploy', 'get-deployment-group', '--application-name', f'sim-{environment}-{region}-ecs-app',
+                '--deployment-group-name', f'sim-{environment}-{region}-app-dg')['deploymentGroupInfo']
+    services = group.get('ecsServices', [])
+    if len(services) != 1:
+        raise RuntimeError('Expected exactly one application ECS service')
+    cluster, service = services[0]['clusterName'], services[0]['serviceName']
+    description = aws(region, 'ecs', 'describe-services', '--cluster', cluster, '--services', service)
+    if description.get('failures') or len(description.get('services', [])) != 1:
+        raise RuntimeError('Cannot inspect the application ECS service')
+    record = description['services'][0]
+    if record.get('desiredCount', 0) < 1 or record.get('runningCount') != record['desiredCount'] or record.get('pendingCount') != 0:
+        raise RuntimeError('Application ECS service is not stable')
+    arns = set()
+    for status in ('RUNNING', 'STOPPED'):
+        arns.update(aws(region, 'ecs', 'list-tasks', '--cluster', cluster, '--service-name', service,
+                        '--desired-status', status).get('taskArns', []))
+    live = []
+    ordered = sorted(arns)
+    for start in range(0, len(ordered), 100):
+        response = aws(region, 'ecs', 'describe-tasks', '--cluster', cluster, '--tasks', *ordered[start:start + 100])
+        if response.get('failures'):
+            raise RuntimeError('Cannot account for every ECS task')
+        if len(response.get('tasks', [])) != len(ordered[start:start + 100]):
+            raise RuntimeError('Incomplete ECS task response')
+        live.extend(task for task in response['tasks'] if task.get('lastStatus') != 'STOPPED')
+    if len(live) != record['desiredCount']:
+        raise RuntimeError('Old, stopping, or pending ECS tasks remain')
+    for task in live:
+        app = [container for container in task.get('containers', []) if container.get('name') == 'app']
+        if task.get('lastStatus') != 'RUNNING' or task.get('desiredStatus') != 'RUNNING' or len(app) != 1 or app[0].get('imageDigest') != digest:
+            raise RuntimeError('A live ECS task does not match the acknowledged compatible release')
+    latest = latest_execution(region, pipeline)
+    if latest.get('pipelineExecutionId') != execution_id or not matches_release(latest, digest):
+        raise RuntimeError('Application deployment changed during preflight')
+    print(json.dumps({'ecsRetired': True, 'expectedImageDigest': digest, 'pipelineExecutionId': execution_id,
+                      'operatorAcknowledgedColumnWritersAndWorkers': True}))
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--environment', required=True, choices=['production', 'staging'])
+    parser.add_argument('--region', required=True)
+    parser.add_argument('--expected-image-digest', required=True)
+    args = parser.parse_args()
+    try:
+        verify(args.environment, args.region, args.expected_image_digest)
+    except (RuntimeError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired) as error:
+        print(f'Project column rollout preflight refused: {error}', file=sys.stderr)
+        sys.exit(1)

.github/workflows/ci.yml

@@ -113,6 +113,9 @@ jobs:
   # in place before the new app version deploys (replaces the removed ECS
   # migration sidecar)
   migrate:
+    permissions:
+      contents: read
+      id-token: write
     name: migrate
     needs: [checks]
     # Explicit need results instead of the implicit success(): a skipped job
@@ -132,9 +135,16 @@ jobs:
       STAGING_DATABASE_URL: ${{ secrets.STAGING_DATABASE_URL }}
       MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
       STAGING_MIGRATION_DATABASE_URL: ${{ secrets.STAGING_MIGRATION_DATABASE_URL }}
+      AWS_ROLE_TO_ASSUME: ${{ secrets.AWS_ROLE_TO_ASSUME }}
+      STAGING_AWS_ROLE_TO_ASSUME: ${{ secrets.STAGING_AWS_ROLE_TO_ASSUME }}
+      AWS_REGION: ${{ secrets.AWS_REGION }}
+      STAGING_AWS_REGION: ${{ secrets.STAGING_AWS_REGION }}
 
   # Same ordering for dev (schema push before the dev image lands in ECR)
   migrate-dev:
+    permissions:
+      contents: read
+      id-token: write
     name: migrate-dev
     if: github.event_name == 'push' && github.ref == 'refs/heads/dev'
     uses: ./.github/workflows/migrate.yml

.github/workflows/migrate.yml

@@ -18,6 +18,14 @@ on:
         required: false
       STAGING_MIGRATION_DATABASE_URL:
         required: false
+      AWS_ROLE_TO_ASSUME:
+        required: false
+      STAGING_AWS_ROLE_TO_ASSUME:
+        required: false
+      AWS_REGION:
+        required: false
+      STAGING_AWS_REGION:
+        required: false
   workflow_dispatch:
     inputs:
       environment:
@@ -31,6 +39,7 @@ on:
 
 permissions:
   contents: read
+  id-token: write
 
 jobs:
   migrate:
@@ -51,6 +60,33 @@ jobs:
           provider: ${{ vars.CI_PROVIDER }}
           node-version: ''
 
+      - name: Check Project column expansion and whether enforcement needs rollout evidence
+        id: project-contract
+        if: inputs.environment != 'dev'
+        working-directory: ./packages/db
+        env:
+          DATABASE_URL: ${{ inputs.environment == 'production' && secrets.DATABASE_URL || inputs.environment == 'staging' && secrets.STAGING_DATABASE_URL || '' }}
+          MIGRATION_DATABASE_URL: ${{ inputs.environment == 'production' && secrets.MIGRATION_DATABASE_URL || inputs.environment == 'staging' && secrets.STAGING_MIGRATION_DATABASE_URL || '' }}
+        run: bun --no-env-file scripts/project-contract-required.ts >> "$GITHUB_OUTPUT"
+
+      - name: Configure AWS for Project connector retirement verification
+        if: steps.project-contract.outputs.required == 'true'
+        uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b
+        with:
+          role-to-assume: ${{ inputs.environment == 'production' && secrets.AWS_ROLE_TO_ASSUME || secrets.STAGING_AWS_ROLE_TO_ASSUME }}
+          aws-region: ${{ inputs.environment == 'production' && secrets.AWS_REGION || secrets.STAGING_AWS_REGION }}
+
+      # Acknowledge #8830 only after authority-aware readers/writers deploy and all incompatible
+      # servers and relevant old Trigger.dev runs drain. The runner commits the column-authority
+      # switch before backfill, validation, and connector retirement; retries never revert it.
+      # The preflight independently checks ECS retirement; it does not inspect worker runs.
+      - name: Require completed authority-aware rollout before Project cutover
+        if: steps.project-contract.outputs.required == 'true'
+        env:
+          ENVIRONMENT: ${{ inputs.environment }}
+          EXPECTED_IMAGE_DIGEST: ${{ inputs.environment == 'production' && vars.PROJECT_COLUMN_ENFORCEMENT_READY_IMAGE_DIGEST_PRODUCTION || inputs.environment == 'staging' && vars.PROJECT_COLUMN_ENFORCEMENT_READY_IMAGE_DIGEST_STAGING || '' }}
+        run: python3 .github/scripts/check-project-rollout.py --environment "$ENVIRONMENT" --region "$AWS_REGION" --expected-image-digest "$EXPECTED_IMAGE_DIGEST"
+
       # The expression maps the explicit environment input to exactly one repo
       # secret, so the job never holds another environment's database URL. An
       # unknown environment resolves to empty and the guard below fails the job.

apps/desktop/e2e/fixtures/live-sim.ts

@@ -651,9 +651,16 @@ export class SimDatabase {
   }
 
   async close(): Promise<void> {
-    if (this.userIds.length > 0)
-      await this.sql`delete from "user" where id in ${this.sql(this.userIds)}`.catch(() => {})
-    await this.sql.end({ timeout: 5 })
+    try {
+      if (this.userIds.length > 0)
+        await this.sql.begin(async (tx) => {
+          await tx`delete from workspace where owner_id in ${tx(this.userIds)}`
+          await tx`delete from project where owner_id in ${tx(this.userIds)}`
+          await tx`delete from "user" where id in ${tx(this.userIds)}`
+        })
+    } finally {
+      await this.sql.end({ timeout: 5 })
+    }
   }
 
   /** A user with a workspace, a session as the desktop sign-in creates, and one chat per title. */
@@ -669,8 +676,10 @@ export class SimDatabase {
       await tx`insert into "user" (id, name, email, normalized_email, email_verified, created_at, updated_at)
         values (${userId}, ${name}, ${email}, ${email}, true, now(), now())`
       await tx`insert into user_stats (id, user_id) values (${generateId()}, ${userId})`
-      await tx`insert into workspace (id, name, owner_id, billed_account_user_id)
-        values (${workspaceId}, 'Desktop tools E2E', ${userId}, ${userId})`
+      await tx`insert into project (id, name, owner_id)
+        values (${workspaceId}, 'E2E fixture project', ${userId})`
+      await tx`insert into workspace (id, project_id, name, owner_id, billed_account_user_id)
+        values (${workspaceId}, ${workspaceId}, 'Desktop tools E2E', ${userId}, ${userId})`
       await tx`insert into permissions (id, user_id, entity_type, entity_id, permission_type)
         values (${generateId()}, ${userId}, 'workspace', ${workspaceId}, 'admin')`
       await tx`insert into session (id, token, user_id, user_agent, expires_at, created_at, updated_at)

apps/sim/app/api/copilot/chat/stream/desktop-executor.integration.ts

@@ -4,6 +4,7 @@
  * Redis through the production reconnect route, for the live tail and the replay batch a view
  * attaches with after a reload.
  */
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 import { authMock, authMockFns } from '@sim/testing/mocks/auth.mock'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
 
@@ -72,7 +73,7 @@ describe.runIf(Boolean(redisUrl))('desktop executor binding on a chat stream', (
       createdAt: now,
       updatedAt: now,
     })
-    await db.insert(workspace).values({
+    await insertWorkspaceFixture(db, {
       id: workspaceId,
       name: 'Desktop stream fixture',
       ownerId: userId,
@@ -105,7 +106,7 @@ describe.runIf(Boolean(redisUrl))('desktop executor binding on a chat stream', (
     await db.delete(copilotChats).where(eq(copilotChats.id, chatId))
     await db.delete(desktopDevices).where(eq(desktopDevices.id, deviceId))
     await db.delete(permissions).where(eq(permissions.userId, userId))
-    await db.delete(workspace).where(eq(workspace.id, workspaceId))
+    await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId))
     await db.delete(user).where(eq(user.id, userId))
     await closeRedisConnection()
   })

apps/sim/app/api/knowledge/connectors/member-sync/route.integration.ts

@@ -1,3 +1,4 @@
+import { deleteWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 /**
  * The member sync scheduler's reclaim against real PostgreSQL: a members-mode connector whose
  * member lease went stale is put back on the failure ladder, and a connector in any other access
@@ -51,7 +52,7 @@ describe('member sync reclaim in PostgreSQL', () => {
   })
 
   afterAll(async () => {
-    await db.delete(workspace).where(eq(workspace.id, ids.workspaceId))
+    await deleteWorkspaceFixture(db, eq(workspace.id, ids.workspaceId))
     await db.delete(organization).where(eq(organization.id, ids.organizationId))
     await db.delete(user).where(inArray(user.id, [ids.aliceId, ids.bobId]))
     await db.$client.end()

apps/sim/app/api/v1/knowledge/route.integration.ts

@@ -7,6 +7,7 @@
 import type { Principal } from '@sim/auth/principal'
 import { db } from '@sim/db'
 import { document, organization, user, workspace } from '@sim/db/schema'
+import { deleteWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 import { authMock, authMockFns, createMockRequest } from '@sim/testing'
 import { generateId } from '@sim/utils/id'
 import { eq, inArray } from 'drizzle-orm'
@@ -85,7 +86,7 @@ describe('knowledge-base document totals in PostgreSQL', () => {
   })
 
   afterAll(async () => {
-    await db.delete(workspace).where(eq(workspace.id, ids.workspaceId))
+    await deleteWorkspaceFixture(db, eq(workspace.id, ids.workspaceId))
     await db.delete(organization).where(eq(organization.id, ids.organizationId))
     await db.delete(user).where(inArray(user.id, [ids.aliceId, ids.bobId]))
     vi.unstubAllGlobals()

apps/sim/app/api/webhooks/agentmail/thread-scope.integration.ts

@@ -1,3 +1,4 @@
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 /**
  * An inbox reply continues its parent task's chat, and the parent is found by the
  * `In-Reply-To` header — a value every recipient of the agent's reply holds. Against
@@ -97,7 +98,7 @@ async function seedWorkspace(ws: InboxWorkspace): Promise<void> {
     createdAt: now,
     updatedAt: now,
   })
-  await db.insert(workspace).values({
+  await insertWorkspaceFixture(db, {
     id: ws.id,
     name: 'Inbox thread scope fixture',
     ownerId: ws.ownerId,
@@ -224,7 +225,7 @@ describe('inbox reply threading stays inside the receiving workspace', () => {
   afterAll(async () => {
     const ownerIds = [victim.ownerId, attacker.ownerId]
     await db.delete(permissions).where(inArray(permissions.userId, ownerIds))
-    await db.delete(workspace).where(inArray(workspace.id, [victim.id, attacker.id]))
+    await deleteWorkspaceFixture(db, inArray(workspace.id, [victim.id, attacker.id]))
     await db.delete(user).where(inArray(user.id, ownerIds))
   })
 

apps/sim/background/cleanup-soft-deletes.integration.ts

@@ -43,18 +43,28 @@ async function seedRelease(deletedAt: string | null, sizeBytes: number, patch: n
 
 describe('Retention cleanup of deleted owned bodies', () => {
   beforeAll(async () => {
-    await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
-      VALUES (${userId}, 'Test cleanup fixture', ${`${userId}@example.test`}, true, now(), now())`
-    await control`INSERT INTO user_stats (id, user_id, storage_used_bytes)
-      VALUES (${generateId()}, ${userId}, 100)`
-    await control`INSERT INTO workspace (id, name, owner_id, billed_account_user_id, storage_used_bytes)
-      VALUES (${workspaceId}, 'Test cleanup fixtures', ${userId}, ${userId}, 100)`
+    await control.begin(async (tx) => {
+      await tx`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
+        VALUES (${userId}, 'Test cleanup fixture', ${`${userId}@example.test`}, true, now(), now())`
+      await tx`INSERT INTO user_stats (id, user_id, storage_used_bytes)
+        VALUES (${generateId()}, ${userId}, 100)`
+      await tx`INSERT INTO project (id, name, owner_id)
+        VALUES (${workspaceId}, 'Test cleanup project', ${userId})`
+      await tx`INSERT INTO workspace (id, project_id, name, owner_id, billed_account_user_id, storage_used_bytes)
+        VALUES (${workspaceId}, ${workspaceId}, 'Test cleanup fixtures', ${userId}, ${userId}, 100)`
+    })
   })
 
   afterAll(async () => {
-    await control`DELETE FROM workspace WHERE id = ${workspaceId}`
-    await control`DELETE FROM "user" WHERE id = ${userId}`
-    await control.end()
+    try {
+      await control.begin(async (tx) => {
+        await tx`DELETE FROM workspace WHERE id = ${workspaceId}`
+        await tx`DELETE FROM project WHERE id = ${workspaceId} AND owner_id = ${userId}`
+        await tx`DELETE FROM "user" WHERE id = ${userId}`
+      })
+    } finally {
+      await control.end()
+    }
   })
 
   it('purges an expired test with its source and releases the source bytes', async () => {

apps/sim/background/cleanup-table-row-ttl.integration.ts

@@ -109,8 +109,11 @@ describe.skipIf(!migrated)('Expiration with real PostgreSQL transactions', () =>
   beforeAll(async () => {
     await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
       VALUES (${userId}, 'Expiration integration fixture', ${`${userId}@example.test`}, true, now(), now())`
-    await control`INSERT INTO workspace (id, name, owner_id, billed_account_user_id)
-      VALUES (${workspaceId}, 'Expiration integration fixtures', ${userId}, ${userId})`
+    await control.begin(async (tx) => {
+      await tx`INSERT INTO project (id, name, owner_id) VALUES (${workspaceId}, 'Fixture project', ${userId})`
+      await tx`INSERT INTO workspace (id, project_id, name, owner_id, billed_account_user_id)
+        VALUES (${workspaceId}, ${workspaceId}, 'Expiration integration fixtures', ${userId}, ${userId})`
+    })
   })
 
   beforeEach(async () => {
@@ -124,7 +127,10 @@ describe.skipIf(!migrated)('Expiration with real PostgreSQL transactions', () =>
   })
 
   afterAll(async () => {
-    await control`DELETE FROM workspace WHERE id = ${workspaceId}`
+    await control.begin(async (tx) => {
+      await tx`DELETE FROM workspace WHERE id = ${workspaceId}`
+      await tx`DELETE FROM project WHERE id = ${workspaceId}`
+    })
     await control`DELETE FROM "user" WHERE id = ${userId}`
     writeFileSync(
       join(tmpdir(), 'expiration-qa-measurements.json'),

apps/sim/ee/access-requests/lib/application/copilot-requests.integration.ts

@@ -12,6 +12,7 @@ import {
   user,
   workspace,
 } from '@sim/db/schema'
+import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
 import { envFlagsMock } from '@sim/testing/mocks/env-flags.mock'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray } from 'drizzle-orm'
@@ -92,7 +93,8 @@ describe('chat-delegated workspace access requests', () => {
         createdAt: now,
       }))
     )
-    await db.insert(workspace).values(
+    await insertWorkspaceFixture(
+      db,
       [workspaceId, otherWorkspaceId].map((id) => ({
         id,
         name: 'Access request fixture',
@@ -138,7 +140,7 @@ describe('chat-delegated workspace access requests', () => {
       .delete(permissionAccessRequest)
       .where(eq(permissionAccessRequest.organizationId, organizationId))
     await db.delete(auditLog).where(eq(auditLog.workspaceId, workspaceId))
-    await db.delete(workspace).where(inArray(workspace.id, [workspaceId, otherWorkspaceId]))
+    await deleteWorkspaceFixture(db, inArray(workspace.id, [workspaceId, otherWorkspaceId]))
     await db.delete(organization).where(eq(organization.id, organizationId))
     await db.delete(user).where(inArray(user.id, [requesterId, peerId]))
   })