feat(projects): enforce membership after the staged backfill

Author: mzxchandra

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

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Read-only, fail-closed ECS retirement check for the Project contract migration.
+
+The expected digest is the operator's release-scoped acknowledgment that Project
+writers are enabled, relevant old worker jobs are drained, and backfill verification
+passed. AWS checks below independently verify ECS retirement, not those assertions.
+"""
+import argparse
+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 verify(environment, region, digest):
+    if not re.fullmatch(r'sha256:[0-9a-f]{64}', digest):
+        raise RuntimeError('Set the environment-specific PROJECT_ENFORCEMENT_READY_IMAGE_DIGEST after reviewing rollout and backfill evidence')
+    pipeline = f'sim-{environment}-{region}-app-deployment'
+    executions = aws(region, 'codepipeline', 'list-pipeline-executions', '--pipeline-name', pipeline).get('pipelineExecutionSummaries', [])
+    if not executions or executions[0].get('status') != 'Succeeded':
+        raise RuntimeError('The latest app pipeline has not completed; traffic cutover alone is insufficient')
+    execution_id = executions[0]['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 = aws(region, 'codepipeline', 'list-pipeline-executions', '--pipeline-name', pipeline).get('pipelineExecutionSummaries', [])
+    if not latest or latest[0].get('pipelineExecutionId') != execution_id or latest[0].get('status') != 'Succeeded':
+        raise RuntimeError('Application deployment changed during preflight')
+    print(json.dumps({'ecsRetired': True, 'expectedImageDigest': digest, 'pipelineExecutionId': execution_id,
+                      'operatorAcknowledgedWritersWorkersAndBackfill': 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 rollout preflight refused: {error}', file=sys.stderr)
+        sys.exit(1)

.github/workflows/ci.yml

@@ -116,6 +116,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 DB
     needs: [test-build]
     # Explicit need results instead of the implicit success(): a skipped job
@@ -134,6 +137,9 @@ jobs:
 
   # Same ordering for dev (schema push before the dev image lands in ECR)
   migrate-dev:
+    permissions:
+      contents: read
+      id-token: write
     name: Migrate Dev DB
     if: github.event_name == 'push' && github.ref == 'refs/heads/dev'
     uses: ./.github/workflows/migrations.yml

.github/workflows/migrations.yml

@@ -20,6 +20,7 @@ on:
 
 permissions:
   contents: read
+  id-token: write
 
 jobs:
   migrate:
@@ -51,6 +52,32 @@ jobs:
       - name: Install dependencies
         run: bun install --frozen-lockfile --ignore-scripts
 
+      - name: Check whether Project 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 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 }}
+
+      # Set this release-scoped acknowledgment only after writers are enabled everywhere,
+      # relevant old Trigger.dev runs are drained, and the reviewed backfill verifies ready.
+      # The preflight independently checks ECS retirement; it does not inspect worker runs.
+      - name: Require completed compatible rollout before Project enforcement
+        if: steps.project-contract.outputs.required == 'true'
+        env:
+          ENVIRONMENT: ${{ inputs.environment }}
+          EXPECTED_IMAGE_DIGEST: ${{ inputs.environment == 'production' && vars.PROJECT_ENFORCEMENT_READY_IMAGE_DIGEST_PRODUCTION || inputs.environment == 'staging' && vars.PROJECT_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/sim/lib/core/config/env.ts

@@ -669,7 +669,6 @@ export const env = createEnv({
 
     // SSO Configuration (for script-based registration)
     SSO_ENABLED:                           z.boolean().optional(),                 // Enable SSO functionality
-    PROJECT_WRITES_ENABLED:               z.boolean().optional(),                 // Activate new Project assignments after old writers drain
     PROJECT_API_ENABLED:                  z.boolean().optional(),                 // Expose Projects after backfill and contract enforcement
     SCIM_ENABLED:                          z.boolean().optional(),                 // Enable SCIM directory provisioning
     USAGE_MONITORING_ENABLED:              z.boolean().optional(),                 // Enable organization usage monitoring on self-hosted (bypasses hosted requirements)

apps/sim/lib/projects/README.md

@@ -2,7 +2,7 @@
 
 A Project groups environments. An environment is an existing `workspace` record; there is no separate environment table. Every newly created Project starts with an environment.
 
-Project APIs return HTTP 503 until the deployment enables `PROJECT_API_ENABLED`. Automatic assignment on legacy workspace creation requires `PROJECT_WRITES_ENABLED`; both server controls default off. The API requires writers to be enabled. Existing assigned Projects always retain their lifecycle protections, including fork inheritance and disconnect behavior, even if activation is disabled.
+Project APIs return HTTP 503 until the deployment enables `PROJECT_API_ENABLED`. Workspace creation always assigns a Project atomically. The API control defaults off and does not disable assignment. Existing assigned Projects always retain their lifecycle protections, including fork inheritance and disconnect behavior, even if activation is disabled.
 
 ## Choose the creation flow
 

apps/sim/lib/projects/__integration__/foundation.integration.ts

@@ -1,4 +1,4 @@
-import { mkdir, writeFile } from 'node:fs/promises'
+import { mkdir, readFile, writeFile } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
 import { db } from '@sim/db'
 import {
@@ -18,6 +18,7 @@ import {
   workspaceForkPromoteRun,
   workspaceForkResourceMap,
 } from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
 import {
   createSessionPrincipal,
   createWorkspaceApiKeyPrincipal,
@@ -27,6 +28,7 @@ import { getErrorMessage, getPostgresErrorCode } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray, sql } from 'drizzle-orm'
+import postgres from 'postgres'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { removeUserFromOrganization } from '@/lib/billing/organizations/membership'
 import { prepareProjectsForAccountDeletion } from '@/lib/projects/account-deletion'
@@ -194,7 +196,79 @@ afterAll(async () => {
 
 describe('Project foundation at the database and application boundary', () => {
   check(
-    'disabled rollout preserves legacy creation and fork/disconnect without Project rows',
+    'application creation, disconnect, organization deletion and archive commit with SQL enforcement',
+    async () => {
+      const client = postgres(readTestDatabaseUrl(), { max: 1, onnotice: () => undefined })
+      try {
+        await client.unsafe(
+          await readFile(
+            new URL(
+              '../../../../../packages/db/migrations/0394_project_membership_enforcement.sql',
+              import.meta.url
+            ),
+            'utf8'
+          )
+        )
+        const f = await fixture(true, 1)
+        const organizationId = f.organizationId
+        if (!organizationId) throw new Error('Missing organization fixture')
+        const created = await createProject.execute({
+          principal: f.owner,
+          input: {
+            organizationId: f.organizationId,
+            name: 'Enforced',
+            initialEnvironment: { name: 'Production' },
+          },
+          request,
+        })
+        environments.push(created.initialEnvironment.id)
+        const source = await getWorkspaceWithOwner(created.initialEnvironment.id)
+        if (!source) throw new Error('Missing source environment')
+        const fork = await createFork({
+          source,
+          policy: await getWorkspaceCreationPolicy({ userId: f.ownerId }),
+          userId: f.ownerId,
+          name: 'Staging',
+        })
+        environments.push(fork.workspace.id)
+        await unlinkForkEdge({ parentWorkspaceId: source.id, childWorkspaceId: fork.workspace.id })
+        const [detached] = await db
+          .select()
+          .from(projectWorkspace)
+          .where(eq(projectWorkspace.workspaceId, fork.workspace.id))
+        expect(detached.projectId).not.toBe(created.project.id)
+        await db.transaction(async (tx) => {
+          await detachOrganizationWorkspacesTx(tx, organizationId)
+          await tx.delete(organization).where(eq(organization.id, organizationId))
+        })
+        await archiveProject.execute({
+          principal: f.owner,
+          input: { projectId: created.project.id },
+          request,
+        })
+        const [archived] = await db.select().from(project).where(eq(project.id, created.project.id))
+        expect(archived.organizationId).toBeNull()
+        expect(archived.archivedAt).not.toBeNull()
+        expect(
+          await db
+            .select()
+            .from(workflow)
+            .where(and(eq(workflow.workspaceId, source.id), sql`${workflow.archivedAt} IS NULL`))
+        ).toHaveLength(0)
+      } finally {
+        await client.unsafe('ROLLBACK')
+        for (const table of ['project', 'project_workspace', 'workspace', 'workflow']) {
+          await client.unsafe(
+            `DROP TRIGGER IF EXISTS project_contract_lock ON ${table}; DROP TRIGGER IF EXISTS project_contract_check ON ${table}`
+          )
+        }
+        await client.end()
+      }
+    }
+  )
+
+  check(
+    'workspace creation and fork/disconnect assign Projects even with the retired writer flag off',
     async () => {
       vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
       vi.stubEnv('PROJECT_API_ENABLED', 'false')
@@ -214,7 +288,7 @@ describe('Project foundation at the database and application boundary', () => {
       environments.push(source.id)
       expect(
         await db.select().from(projectWorkspace).where(eq(projectWorkspace.workspaceId, source.id))
-      ).toEqual([])
+      ).toHaveLength(1)
       const parent = await getWorkspaceWithOwner(source.id)
       if (!parent) throw new Error('Missing source fixture')
       const fork = await createFork({
@@ -229,11 +303,11 @@ describe('Project foundation at the database and application boundary', () => {
           .select()
           .from(projectWorkspace)
           .where(eq(projectWorkspace.workspaceId, fork.workspace.id))
-      ).toEqual([])
+      ).toHaveLength(1)
       await unlinkForkEdge({ parentWorkspaceId: source.id, childWorkspaceId: fork.workspace.id })
       const [child] = await db.select().from(workspace).where(eq(workspace.id, fork.workspace.id))
       expect(child.forkedFromWorkspaceId).toBeNull()
-      expect(await db.select().from(project).where(eq(project.ownerId, f.ownerId))).toHaveLength(1)
+      expect(await db.select().from(project).where(eq(project.ownerId, f.ownerId))).toHaveLength(3)
     }
   )
 
@@ -316,7 +390,7 @@ describe('Project foundation at the database and application boundary', () => {
     }
   )
 
-  check('writer activation assigns new workspaces while Project APIs remain disabled', async () => {
+  check('new workspaces receive Projects while Project APIs remain disabled', async () => {
     vi.stubEnv('PROJECT_API_ENABLED', 'false')
     const f = await fixture(false, 1)
     const created = await db.transaction((tx) =>
@@ -335,18 +409,6 @@ describe('Project foundation at the database and application boundary', () => {
     expect(
       await db.select().from(projectWorkspace).where(eq(projectWorkspace.workspaceId, created.id))
     ).toHaveLength(1)
-    vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
-    vi.stubEnv('PROJECT_API_ENABLED', 'true')
-    await expect(
-      createProject.execute({
-        principal: f.owner,
-        input: { organizationId: null, name: 'Invalid', initialEnvironment: { name: 'First' } },
-        request,
-      })
-    ).rejects.toThrow('PROJECT_API_ENABLED requires PROJECT_WRITES_ENABLED')
-    expect(await db.select().from(workspace).where(eq(workspace.ownerId, f.ownerId))).toHaveLength(
-      2
-    )
   })
 
   check('legacy fork and disconnect refuse a partially assigned subtree', async () => {

apps/sim/lib/projects/rollout.server.ts

@@ -1,15 +1,6 @@
 import { envBoolean, getEnv } from '@/lib/core/config/env'
 import { HttpError } from '@/lib/core/utils/http-error'
 
-/** Deployment-wide controls; existing Project lifecycle maintenance never depends on these. */
-export function getProjectRollout() {
-  const writesEnabled = envBoolean(getEnv('PROJECT_WRITES_ENABLED')) ?? false
-  const apiEnabled = envBoolean(getEnv('PROJECT_API_ENABLED')) ?? false
-  if (apiEnabled && !writesEnabled)
-    throw new Error('PROJECT_API_ENABLED requires PROJECT_WRITES_ENABLED')
-  return { writesEnabled, apiEnabled }
-}
-
 class ProjectUnavailableError extends HttpError {
   readonly statusCode = 503
   constructor() {
@@ -18,5 +9,5 @@ class ProjectUnavailableError extends HttpError {
 }
 
 export function requireProjectApiEnabled(): void {
-  if (!getProjectRollout().apiEnabled) throw new ProjectUnavailableError()
+  if (!(envBoolean(getEnv('PROJECT_API_ENABLED')) ?? false)) throw new ProjectUnavailableError()
 }