Protect legacy queue revisions and add fail-closed canary controls

.github/workflows/deploy-cloudflare-worker.yml

@@ -33,11 +33,25 @@ jobs:
           EXPECTED_SHA: ${{ inputs.expected_sha }}
           RELEASE_SHA: ${{ github.sha }}
           ROLLOUT_PREFLIGHT_CONFIRMED: ${{ inputs.rollout_preflight_confirmed }}
+          CANARY_USER_IDS: ${{ secrets.SUPABASE_WORKER_CANARY_USER_IDS }}
         run: |
           set -eu
           test "$ROLLOUT_PREFLIGHT_CONFIRMED" = 'true'
           test "$EXPECTED_SHA" = "$RELEASE_SHA"
           echo "$EXPECTED_SHA" | grep -Eq '^[0-9a-f]{40}$'
+          CANARY_USER_IDS="$CANARY_USER_IDS" node <<'NODE'
+          const ids = String(process.env.CANARY_USER_IDS || '')
+            .split(',')
+            .map(value => value.trim())
+            .filter(Boolean);
+          if (ids.length < 1 || ids.length > 3) {
+            throw new Error('SUPABASE_WORKER_CANARY_USER_IDS must contain 1-3 tenant UUIDs');
+          }
+          const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+          if (new Set(ids.map(value => value.toLowerCase())).size !== ids.length || ids.some(value => !uuid.test(value))) {
+            throw new Error('SUPABASE_WORKER_CANARY_USER_IDS must contain unique tenant UUIDs');
+          }
+          NODE
       - uses: actions/checkout@v4
         with:
           ref: ${{ github.sha }}
@@ -48,7 +62,21 @@ jobs:
           cache: npm
       - run: npm ci
       - run: npm run ci
-      - run: npx wrangler deploy --tag "${{ github.sha }}"
+      - name: Deploy the allowlisted, provider-disabled canary
         env:
           CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }}
           CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
+          CANARY_USER_IDS: ${{ secrets.SUPABASE_WORKER_CANARY_USER_IDS }}
+        run: |
+          set -eu
+          secrets_file="$RUNNER_TEMP/worker-canary-secrets.json"
+          trap 'rm -f "$secrets_file"' EXIT
+          CANARY_USER_IDS="$CANARY_USER_IDS" SECRETS_FILE="$secrets_file" node <<'NODE'
+          const fs = require('node:fs');
+          fs.writeFileSync(
+            process.env.SECRETS_FILE,
+            JSON.stringify({ SUPABASE_WORKER_CANARY_USER_IDS: process.env.CANARY_USER_IDS }),
+            { mode: 0o600 }
+          );
+          NODE
+          npx wrangler deploy --tag "${{ github.sha }}" --secrets-file "$secrets_file"

config.ts

@@ -109,6 +109,10 @@ export interface AppConfig {
   CREDENTIAL_ENCRYPTION_KEY: string;
   SUPABASE_WORKER_POLL_INTERVAL_MS: number;
   SUPABASE_WORKER_BATCH_SIZE: number;
+  SUPABASE_WORKER_CANARY_REQUIRED: boolean;
+  SUPABASE_WORKER_CANARY_USER_IDS: Set<string>;
+  SUPABASE_WORKER_GENERATION_ENABLED: boolean;
+  SUPABASE_PROVIDER_DISPATCH_ENABLED: boolean;
   DAILY_INVENTORY_PLANNER_ENABLED: boolean;
   DAILY_INVENTORY_PLANNER_START_LOCAL_DATE: string;
 }
@@ -241,6 +245,18 @@ function buildBaseConfig(): AppConfig {
     CREDENTIAL_ENCRYPTION_KEY: IS_SOCIAL_CONNECTOR_ONLY ? '' : process.env.CREDENTIAL_ENCRYPTION_KEY || '',
     SUPABASE_WORKER_POLL_INTERVAL_MS: Number.parseInt(process.env.SUPABASE_WORKER_POLL_INTERVAL_MS || '10000', 10),
     SUPABASE_WORKER_BATCH_SIZE: Number.parseInt(process.env.SUPABASE_WORKER_BATCH_SIZE || '10', 10),
+    SUPABASE_WORKER_CANARY_REQUIRED:
+      (process.env.NODE_ENV || 'development') === 'production'
+      || parseBooleanEnv(process.env.SUPABASE_WORKER_CANARY_REQUIRED, false),
+    SUPABASE_WORKER_CANARY_USER_IDS: toSubSet(process.env.SUPABASE_WORKER_CANARY_USER_IDS, ''),
+    SUPABASE_WORKER_GENERATION_ENABLED: parseBooleanEnv(
+      process.env.SUPABASE_WORKER_GENERATION_ENABLED,
+      (process.env.NODE_ENV || 'development') !== 'production'
+    ),
+    SUPABASE_PROVIDER_DISPATCH_ENABLED: parseBooleanEnv(
+      process.env.SUPABASE_PROVIDER_DISPATCH_ENABLED,
+      (process.env.NODE_ENV || 'development') !== 'production'
+    ),
     DAILY_INVENTORY_PLANNER_ENABLED: parseBooleanEnv(process.env.DAILY_INVENTORY_PLANNER_ENABLED, false),
     DAILY_INVENTORY_PLANNER_START_LOCAL_DATE: process.env.DAILY_INVENTORY_PLANNER_START_LOCAL_DATE || '',
   };

package.json

@@ -6,7 +6,7 @@
   "scripts": {
     "build": "tsx scripts/build.ts",
     "typecheck": "tsc --noEmit --project tsconfig.json",
-    "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js",
+    "test": "npm run build && node dist/test/security-hardening.test.js && node dist/test/source-ssrf.test.js && node dist/test/browser-collector-ingest.test.js && node dist/test/slot-scheduler.test.js && node dist/test/daily-inventory-planner.test.js && node dist/test/refresh-queue-finalization.test.js && node dist/test/publish-summary.test.js && node dist/test/threads-refresh.test.js && node dist/test/linkedin-refresh.test.js && node dist/test/instagram-image-timeout.test.js && node dist/test/cost-control.test.js && node dist/test/recovery-scheduler.test.js && node dist/test/social-connector.test.js && node dist/test/meta-publication-boundary.test.js && node dist/test/cloudflare-health.test.js && node dist/test/canary-policy.test.js && node dist/test/exclusive-run-gate.test.js && node dist/test/runtime-scope.test.js && node dist/test/tenant-platform-policy.test.js && node dist/test/supabase-client-retry.test.js && node dist/test/worker-claims.test.js && node dist/test/publication-ledger.test.js && node dist/test/publication-outcome.test.js && node dist/test/publication-executor.test.js && node dist/test/provider-single-dispatch.test.js && node dist/test/publication-receipts.test.js && node dist/test/legacy-revision-hold.test.js",
     "smoke:dist": "node dist/src/cli.js status",
     "ci": "npm run typecheck && npm test && npm run smoke:dist",
     "dev": "tsx src/agent.ts",

src/canary-policy.ts

@@ -0,0 +1,68 @@
+export type RolloutControlledJobKind =
+  | 'fetch_sources'
+  | 'refresh_queue'
+  | 'publish_now'
+  | 'publish_all'
+  | 'skip_slot'
+  | 'release_slot';
+
+export interface RolloutPolicy {
+  canaryRequired: boolean;
+  canaryUserIds: ReadonlySet<string>;
+  generationEnabled: boolean;
+  providerDispatchEnabled: boolean;
+}
+
+const GENERATION_JOB_KINDS = new Set<RolloutControlledJobKind>([
+  'fetch_sources',
+  'refresh_queue',
+]);
+
+const PUBLICATION_JOB_KINDS = new Set<RolloutControlledJobKind>([
+  'publish_now',
+  'publish_all',
+]);
+
+const OPERATOR_JOB_KINDS: RolloutControlledJobKind[] = [
+  'skip_slot',
+  'release_slot',
+];
+
+export function allowedCanaryUserIds(policy: RolloutPolicy): string[] | undefined {
+  if (!policy.canaryRequired) return undefined;
+  return [...policy.canaryUserIds]
+    .map(userId => userId.trim().toLowerCase())
+    .filter(Boolean)
+    .sort();
+}
+
+export function canaryAllowsUser(policy: RolloutPolicy, userId: string): boolean {
+  if (!policy.canaryRequired) return true;
+  const normalized = userId.trim().toLowerCase();
+  return [...policy.canaryUserIds].some(
+    allowed => allowed.trim().toLowerCase() === normalized
+  );
+}
+
+export function runnableJobKinds(policy: RolloutPolicy): RolloutControlledJobKind[] {
+  return [
+    ...(policy.generationEnabled ? [...GENERATION_JOB_KINDS] : []),
+    ...(policy.providerDispatchEnabled ? [...PUBLICATION_JOB_KINDS] : []),
+    ...OPERATOR_JOB_KINDS,
+  ];
+}
+
+export function rolloutBlockCode(
+  policy: RolloutPolicy,
+  userId: string,
+  kind: RolloutControlledJobKind
+): string | undefined {
+  if (!canaryAllowsUser(policy, userId)) return 'rollout_canary_tenant_blocked';
+  if (GENERATION_JOB_KINDS.has(kind) && !policy.generationEnabled) {
+    return 'rollout_generation_disabled';
+  }
+  if (PUBLICATION_JOB_KINDS.has(kind) && !policy.providerDispatchEnabled) {
+    return 'rollout_provider_dispatch_disabled';
+  }
+  return undefined;
+}

src/cloudflare-worker.ts

@@ -15,6 +15,10 @@ interface Env {
   SERVICE_ROLE_KEY?: string;
   CREDENTIAL_ENCRYPTION_KEY?: string;
   SUPABASE_WORKER_BATCH_SIZE?: string;
+  SUPABASE_WORKER_CANARY_REQUIRED?: string;
+  SUPABASE_WORKER_CANARY_USER_IDS?: string;
+  SUPABASE_WORKER_GENERATION_ENABLED?: string;
+  SUPABASE_PROVIDER_DISPATCH_ENABLED?: string;
   DAILY_INVENTORY_PLANNER_ENABLED?: string;
   DAILY_INVENTORY_PLANNER_START_LOCAL_DATE?: string;
   HTTP_TIMEOUT_MS?: string;
@@ -87,8 +91,25 @@ function publicationCapabilities(): Record<string, { state: string; code: string
   };
 }
 
+function booleanBinding(value: string | undefined, fallback: boolean): boolean {
+  if (value === undefined || value === '') return fallback;
+  return /^(1|true|yes|on)$/i.test(value.trim());
+}
+
+function canaryUserCount(value: string | undefined): number {
+  return new Set(
+    String(value || '')
+      .split(',')
+      .map(item => item.trim().toLowerCase())
+      .filter(Boolean)
+  ).size;
+}
+
 function healthPayload(env: Env): Record<string, unknown> {
   const metadata = env.CF_VERSION_METADATA;
+  const production = env.NODE_ENV === 'production';
+  const canaryRequired = production || booleanBinding(env.SUPABASE_WORKER_CANARY_REQUIRED, false);
+  const allowedTenantCount = canaryUserCount(env.SUPABASE_WORKER_CANARY_USER_IDS);
   return {
     ok: true,
     liveness: 'ok',
@@ -104,6 +125,17 @@ function healthPayload(env: Env): Record<string, unknown> {
       appliedSchema: 'unverified',
     },
     publicationCapabilities: publicationCapabilities(),
+    rollout: {
+      tenantScope: canaryRequired
+        ? allowedTenantCount > 0
+          ? 'allowlisted'
+          : 'blocked_empty_allowlist'
+        : 'unrestricted',
+      canaryRequired,
+      allowedTenantCount,
+      generationEnabled: booleanBinding(env.SUPABASE_WORKER_GENERATION_ENABLED, !production),
+      providerDispatchEnabled: booleanBinding(env.SUPABASE_PROVIDER_DISPATCH_ENABLED, !production),
+    },
     executionGate: scheduledTickGate.snapshot(),
   };
 }

src/publication-executor.ts

@@ -352,13 +352,20 @@ export async function reconcilePublication(
 }
 
 /** Also recovers orphaned publish_all attempts after their parent job has already ended. */
-export async function recoverStalePublications(now = Date.now()): Promise<number> {
+export async function recoverStalePublications(
+  now = Date.now(),
+  allowedUserIds?: readonly string[]
+): Promise<number> {
+  if (allowedUserIds?.length === 0) return 0;
   try {
     await ledger.assertPublicationLedgerContract();
     const intents = await supabaseSelect<ledger.PublicationIntent>('publication_intents', {
       filters: [
         { column: 'state', operator: 'in', value: ['claimed', 'dispatching'] },
         { column: 'updated_at', operator: 'lte', value: new Date(now - 180_000).toISOString() },
+        ...(allowedUserIds
+          ? [{ column: 'user_id', operator: 'in' as const, value: [...allowedUserIds] }]
+          : []),
       ],
       order: 'updated_at.asc',
       limit: 50,

src/publication-ledger.ts

@@ -17,6 +17,7 @@ export const REQUIRED_PUBLICATION_CAPABILITIES = [
   'publication-exact-history-receipt-v1',
   'publication-queue-compatibility-fence-v1',
   'publication-provenance-snapshot-v1',
+  'publication-legacy-queue-hold-v1',
 ] as const;
 
 export interface PublicationSchemaContract {