chore: release v4.5.6 (#4317)

## Summary
5 improvements, 9 bug fixes.

## Breaking changes
- Self-hosted deployments no longer ship shared default credentials;
fresh installs generate their own. If yours still uses a previously
published default, set a unique value before upgrading, or set
`ALLOW_INSECURE_DEFAULT_SECRETS=true` to keep booting while you migrate.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))

## Improvements
- Require explicit browser approval for CLI and MCP login, with
resilient polling while approval is pending.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Deployed task telemetry now reports the deployment identifier (e.g.
`deployment_abc123`) in the `worker.id` attribute, instead of an opaque
internal value. Upgrade to get the readable identifier in your own
OpenTelemetry exporters.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Prevent prototype pollution when applying run metadata operations or
reconstructing nested telemetry attributes, while preserving legitimate
`constructor` and `prototype` fields.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Add helpers to mint and verify the deployment-scoped token used to
authenticate run controllers to the platform.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))

## Server changes

These changes affect the self-hosted Docker image and Trigger.dev Cloud:

- Added optional request rate limiting for telemetry ingestion
endpoints.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Background-worker deployment lookups are now scoped to the
authenticated environment.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Updating a GitHub App installation from the callback flow is now
scoped to your own organization, so an installation ID belonging to
another organization can no longer be used to refresh that
organization's installation record. The GitHub App installation session
is also now single-use, so completing an installation callback
invalidates its state and it can no longer be replayed.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Scope schedule and environment-variable writes to the caller's project
and environment
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Reject compute snapshot callbacks that do not match the snapshot
request that created them.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Require secret-key authentication to initialize the session out
(agent→client) stream, matching the append route.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Live run and trace subscriptions now validate their identifiers more
strictly and only return data from your own organization.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Window-function names in the query compiler are now validated against
the allowlist, matching how other function calls are handled.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Authenticate run controllers to the platform with a signed,
deployment-scoped token.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Verify that worker actions (starting, completing, and continuing a
run, and reading its snapshots) target a run belonging to the caller's
environment.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))

<details>
<summary>Raw changeset output</summary>

# Releases
## @trigger.dev/build@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
## trigger.dev@4.5.6

### Patch Changes

- Require explicit browser approval for CLI and MCP login, with
resilient polling while approval is pending.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Deployed task telemetry now reports the deployment identifier (e.g.
`deployment_abc123`) in the `worker.id` attribute, instead of an opaque
internal value. Upgrade to get the readable identifier in your own
OpenTelemetry exporters.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
  - `@trigger.dev/build@4.5.6`
  - `@trigger.dev/schema-to-json@4.5.6`
## @trigger.dev/core@4.5.6

### Patch Changes

- Prevent prototype pollution when applying run metadata operations or
reconstructing nested telemetry attributes, while preserving legitimate
`constructor` and `prototype` fields.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Require explicit browser approval for CLI and MCP login, with
resilient polling while approval is pending.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
- Add helpers to mint and verify the deployment-scoped token used to
authenticate run controllers to the platform.
([#4316](https://github.com/triggerdotdev/trigger.dev/pull/4316))
## @trigger.dev/python@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
  - `@trigger.dev/build@4.5.6`
  - `@trigger.dev/sdk@4.5.6`
## @trigger.dev/react-hooks@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
## @trigger.dev/redis-worker@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
## @trigger.dev/rsc@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
## @trigger.dev/schema-to-json@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`
## @trigger.dev/sdk@4.5.6

### Patch Changes

- Updated dependencies:
  - `@trigger.dev/core@4.5.6`

</details>

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

.changeset/experimental-node-runtimes.md

@@ -1,6 +0,0 @@
----
-"@trigger.dev/core": patch
-"trigger.dev": patch
----
-
-Add experimental Node.js 24 and 26 task runtimes. Set `runtime` to `experimental-node-24` or `experimental-node-26` in `trigger.config.ts`.

.changeset/project-default-region-response.md

@@ -1,5 +0,0 @@
----
-"@trigger.dev/core": patch
----
-
-Add `defaultRegion` to the project GET and list API responses; null when unset.

.env.example

@@ -2,11 +2,12 @@
 SESSION_SECRET=abcdef1234
 MAGIC_LINK_SECRET=abcdef1234
 ENCRYPTION_KEY=ae13021afef0819c3a307ad487071c06 # Must be a random 16 byte hex string. You can generate an encryption key by running `openssl rand -hex 16` in your terminal
+MANAGED_WORKER_SECRET=abcdef1234 # Must match the supervisor's MANAGED_WORKER_SECRET
 LOGIN_ORIGIN=http://localhost:3030
 DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres?schema=public
 # This sets the URL used for direct connections to the database and should only be needed in limited circumstances
 # See: https://www.prisma.io/docs/reference/api-reference/prisma-schema-reference#fields:~:text=the%20shadow%20database.-,directUrl,-No
-DIRECT_URL=${DATABASE_URL}
+DIRECT_URL=postgresql://postgres:postgres@localhost:5432/postgres?schema=public
 # Dedicated run-ops database (@internal/run-ops-database). Only needed to run prisma commands
 # against it or to enable the run-ops split; start it with `docker compose --profile runops up`.
 RUN_OPS_DATABASE_URL=postgresql://postgres:postgres@localhost:5434/postgres?schema=public
@@ -166,4 +167,4 @@ POSTHOG_PROJECT_KEY=
 # Uncomment these to send metrics to the local Prometheus via OTEL Collector:
 # INTERNAL_OTEL_METRIC_EXPORTER_ENABLED=1
 # INTERNAL_OTEL_METRIC_EXPORTER_URL=http://localhost:4318/v1/metrics
-# INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS=15000
+# INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS=15000

.github/workflows/helm-prerelease.yml

@@ -52,12 +52,14 @@ jobs:
 
       - name: Lint Helm Chart
         run: |
-          helm lint ./hosting/k8s/helm/
+          helm lint ./hosting/k8s/helm/ \
+            --values ./hosting/k8s/helm/ci/lint-values.yaml
 
       - name: Render templates
         run: |
           helm template test-release ./hosting/k8s/helm/ \
             --values ./hosting/k8s/helm/values.yaml \
+            --values ./hosting/k8s/helm/ci/lint-values.yaml \
             --output-dir ./helm-output
 
       - name: Validate manifests

.github/workflows/pr_checks.yml

@@ -56,6 +56,7 @@ jobs:
               - '.github/workflows/pr_checks.yml'
               - '.github/workflows/unit-tests-webapp.yml'
               - '.github/workflows/e2e-webapp.yml'
+              - '.github/workflows/runops-guard.yml'
               - '.configs/**'
               - 'package.json'
               - 'pnpm-lock.yaml'
@@ -111,6 +112,11 @@ jobs:
     if: needs.changes.outputs.code == 'true' || needs.changes.outputs.typecheck_self == 'true'
     uses: ./.github/workflows/typecheck.yml
 
+  runops-guard:
+    needs: changes
+    if: needs.changes.outputs.webapp == 'true'
+    uses: ./.github/workflows/runops-guard.yml
+
   webapp:
     needs: changes
     if: needs.changes.outputs.webapp == 'true'
@@ -161,6 +167,7 @@ jobs:
       - changes
       - code-quality
       - typecheck
+      - runops-guard
       - webapp
       - e2e-webapp
       - packages

.github/workflows/publish-worker-v4.yml

@@ -13,6 +13,13 @@ on:
         type: string
         required: false
         default: ""
+    outputs:
+      version:
+        description: The published image tag
+        value: ${{ jobs.build.outputs.version }}
+      image_repo:
+        description: The image repository the build was published to (without tag)
+        value: ${{ jobs.build.outputs.image_repo }}
   push:
     tags:
       - "re2-test-*"
@@ -38,6 +45,11 @@ jobs:
       matrix:
         package: [supervisor]
     runs-on: warp-ubuntu-latest-x64-2x
+    # Single-entry matrix, so these job outputs are unambiguous (consumed by the
+    # scan-supervisor job in publish.yml).
+    outputs:
+      version: ${{ steps.get_tag.outputs.tag }}
+      image_repo: ${{ steps.set_tags.outputs.image_repo }}
     env:
       DOCKER_BUILDKIT: "1"
     steps:
@@ -81,6 +93,7 @@ jobs:
           fi
 
           echo "image_tags=${image_tags}" >> "$GITHUB_OUTPUT"
+          echo "image_repo=${ref_without_tag}" >> "$GITHUB_OUTPUT"
         env:
           IMAGE_REGISTRY: ${{ inputs.image_registry || vars.IMAGE_REGISTRY || format('ghcr.io/{0}', github.repository_owner) }}
           STEPS_GET_REPOSITORY_OUTPUTS_REPO: ${{ steps.get_repository.outputs.repo }}

.github/workflows/publish.yml

@@ -97,10 +97,19 @@ jobs:
     permissions:
       contents: read
       packages: read # pull the just-published image from GHCR
-    uses: ./.github/workflows/trivy-image-webapp.yml
+    uses: ./.github/workflows/trivy-image.yml
     with:
       image-ref: ${{ needs.publish-webapp.outputs.image_repo }}:${{ needs.publish-webapp.outputs.version }}
 
+  scan-supervisor:
+    needs: [publish-worker-v4]
+    permissions:
+      contents: read
+      packages: read # pull the just-published image from GHCR
+    uses: ./.github/workflows/trivy-image.yml
+    with:
+      image-ref: ${{ needs.publish-worker-v4.outputs.image_repo }}:${{ needs.publish-worker-v4.outputs.version }}
+
   # Announce the freshly published mutable `main` webapp image to subscriber
   # repos via repository_dispatch, handing them a digest-pinned ref to build or
   # deploy from. The repo, ref prefix, and dispatch target all default to the

.github/workflows/release-helm.yml

@@ -47,12 +47,14 @@ jobs:
 
       - name: Lint Helm Chart
         run: |
-          helm lint ./hosting/k8s/helm/
+          helm lint ./hosting/k8s/helm/ \
+            --values ./hosting/k8s/helm/ci/lint-values.yaml
 
       - name: Render templates
         run: |
           helm template test-release ./hosting/k8s/helm/ \
             --values ./hosting/k8s/helm/values.yaml \
+            --values ./hosting/k8s/helm/ci/lint-values.yaml \
             --output-dir ./helm-output
 
       - name: Validate manifests

.github/workflows/runops-guard.yml

@@ -0,0 +1,38 @@
+name: "🛡️ Run-ops Legacy Guard"
+
+on:
+  workflow_call:
+
+permissions:
+  contents: read
+
+jobs:
+  runops-guard:
+    runs-on: warp-ubuntu-latest-x64-16x
+
+    steps:
+      - name: ⬇️ Checkout repo
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: ⎔ Setup pnpm
+        uses: pnpm/action-setup@fc06bc1257f339d1d5d8b3a19a8cae5388b55320 # v5.0.0
+        with:
+          version: 10.33.2
+
+      - name: ⎔ Setup node
+        uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
+        with:
+          node-version: 24.18.0
+          cache: "pnpm"
+
+      - name: 📥 Download deps
+        run: pnpm install --frozen-lockfile
+
+      - name: 📀 Generate Prisma Client
+        run: pnpm run generate
+
+      - name: 🛡️ Run-ops legacy guard
+        run: pnpm --filter webapp run guard:runops-legacy -- --check

.github/workflows/trivy-image.yml

@@ -1,7 +1,7 @@
-name: Trivy Image Scan (webapp)
+name: Trivy Image Scan
 
-# OS-level CVE scan of a published webapp image. Called by the publish pipeline
-# (publish.yml) to scan each build right after it's pushed to GHCR — so every
+# OS-level CVE scan of a published image. Called by the publish pipeline
+# (publish.yml) to scan each image right after it's pushed to GHCR — so every
 # main build and every release is scanned, not rebuilt. Also runnable ad-hoc
 # via workflow_dispatch against any image ref.
 #
@@ -27,7 +27,7 @@ on:
 permissions: {}
 
 concurrency:
-  group: trivy-image-webapp-${{ inputs.image-ref }}
+  group: trivy-image-${{ inputs.image-ref }}
   cancel-in-progress: true
 
 jobs:
@@ -59,7 +59,7 @@ jobs:
           ignore-unfixed: true
           severity: HIGH,CRITICAL
           format: table
-          output: trivy-image-webapp.txt
+          output: trivy-image.txt
 
       - name: Job summary
         if: always()
@@ -67,9 +67,9 @@ jobs:
           IMAGE_REF: ${{ inputs.image-ref }}
         run: |
           {
-            echo "## Trivy Image Scan (webapp) — \`${IMAGE_REF}\`"
+            echo "## Trivy Image Scan — \`${IMAGE_REF}\`"
             echo '```'
             # GitHub step summary is capped at 1 MiB; truncate large reports.
-            head -c 900000 trivy-image-webapp.txt 2>/dev/null || echo "(no report produced)"
+            head -c 900000 trivy-image.txt 2>/dev/null || echo "(no report produced)"
             echo '```'
           } >> "$GITHUB_STEP_SUMMARY"

.oxlintrc.json

@@ -1,7 +1,10 @@
 {
   "$schema": "./node_modules/oxlint/configuration_schema.json",
   "plugins": ["typescript", "import", "react"],
-  "jsPlugins": ["./oxlint-plugins/no-thrown-unawaited-redirect.mjs"],
+  "jsPlugins": [
+    "./oxlint-plugins/no-thrown-unawaited-redirect.mjs",
+    "./oxlint-plugins/runops-residency.mjs"
+  ],
   "ignorePatterns": [
     "**/dist/**",
     "**/build/**",
@@ -34,5 +37,21 @@
     "react-hooks/exhaustive-deps": "off",
     "react-hooks/rules-of-hooks": "off",
     "trigger/no-thrown-unawaited-redirect": "error"
-  }
+  },
+  "overrides": [
+    {
+      "files": ["apps/webapp/app/**/*.ts", "apps/webapp/app/**/*.tsx"],
+      "rules": {
+        "trigger-runops/no-control-plane-run-graph-access": "error",
+        "trigger-runops/no-control-plane-in-runops-slot": "error"
+      }
+    },
+    {
+      "files": ["apps/webapp/app/**/*.test.ts", "apps/webapp/app/**/*.test.tsx"],
+      "rules": {
+        "trigger-runops/no-control-plane-run-graph-access": "off",
+        "trigger-runops/no-control-plane-in-runops-slot": "off"
+      }
+    }
+  ]
 }

.server-changes/attio-sync-transient-retry.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-Transient internal sync failures are now retried quietly instead of surfacing as errors.

.server-changes/clickhouse-read-replica.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: improvement
----
-
-Optionally route ClickHouse read traffic to a read replica while writes stay on the primary. Set `CLICKHOUSE_READER_URL` to move all reads, or target the busiest paths with `RUNS_LIST_CLICKHOUSE_URL` (runs list) and `EVENTS_READER_CLICKHOUSE_URL` (traces, spans, logs). All optional; unset keeps current behavior.

.server-changes/fix-batch-idempotency-per-item-keys.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-Fix batchTrigger requests that set a per-item idempotency key failing with an error instead of creating and deduplicating the runs

.server-changes/fix-batch-idempotency-point-lookups.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-Speed up idempotency checks on `batchTrigger` calls that use idempotency keys. Large batches against a task with a big run history no longer degrade to multi-second lookups.

.server-changes/run-ops-legacy-shared-pool.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-Avoid opening a redundant database connection pool when the legacy and primary databases are the same server, preventing connection usage from doubling.

.server-changes/stale-deploy-asset-recovery.md

@@ -1,6 +0,0 @@
----
-area: webapp
-type: fix
----
-
-Fix pages occasionally loading unstyled or failing to load during deploys. The dashboard now detects this and reloads to recover automatically, or prompts you to reload if it can't.

apps/supervisor/.env.example

@@ -1,8 +1,8 @@
 # This needs to match the token of the worker group you want to connect to
 TRIGGER_WORKER_TOKEN=
 
-# This needs to match the MANAGED_WORKER_SECRET env var on the webapp
-MANAGED_WORKER_SECRET=managed-secret
+# Must match the webapp's MANAGED_WORKER_SECRET. Generate with: openssl rand -hex 16
+MANAGED_WORKER_SECRET=
 
 # Point this at the webapp in prod
 TRIGGER_API_URL=http://localhost:3030

apps/supervisor/src/env.ts

@@ -14,8 +14,17 @@ export const Env = z
 
     // Required settings
     TRIGGER_API_URL: z.string().url(),
-    TRIGGER_WORKER_TOKEN: z.string(), // accepts file:// path to read from a file
+    TRIGGER_WORKER_TOKEN: z.string().min(1), // accepts file:// path to read from a file
     MANAGED_WORKER_SECRET: z.string(),
+
+    // Deployment token: sign a token into TRIGGER_DEPLOYMENT_ID at pod creation and verify it on
+    // inbound workload calls. "disabled" = off; "log" = mint + verify + metrics only; "enforce" =
+    // also reject invalid tokens.
+    WORKLOAD_TOKEN_SECRET: z.string().optional(),
+    WORKLOAD_TOKEN_ENFORCEMENT: z.enum(["disabled", "log", "enforce"]).default("disabled"),
+    // Absolute expiry for minted deployment tokens. Deterministic (no wall-clock issued-at) so every
+    // pod of a deployment carries an identical token; bump before this date. Must outlive any run.
+    WORKLOAD_TOKEN_EXP: z.string().datetime().default("2032-01-01T00:00:00.000Z"),
     OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url(), // set on the runners
 
     // Workload API settings (coordinator mode) - the workload API is what the run controller connects to
@@ -365,6 +374,14 @@ export const Env = z
         path: ["TRIGGER_WORKLOAD_API_DOMAIN"],
       });
     }
+    if (data.WORKLOAD_TOKEN_ENFORCEMENT !== "disabled" && !data.WORKLOAD_TOKEN_SECRET) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message:
+          "WORKLOAD_TOKEN_SECRET is required when WORKLOAD_TOKEN_ENFORCEMENT is not disabled",
+        path: ["WORKLOAD_TOKEN_SECRET"],
+      });
+    }
     if (
       data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED &&
       !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST

apps/supervisor/src/index.ts

@@ -27,6 +27,7 @@ import { register } from "./metrics.js";
 import { PodCleaner } from "./services/podCleaner.js";
 import { FailedPodHandler } from "./services/failedPodHandler.js";
 import { getWorkerToken } from "./workerToken.js";
+import { mintDeploymentToken } from "./workloadToken.js";
 import { OtlpTraceService } from "./services/otlpTraceService.js";
 import {
   WarmStartVerificationService,
@@ -96,6 +97,7 @@ class ManagedSupervisor {
       COMPUTE_GATEWAY_AUTH_TOKEN,
       DOCKER_REGISTRY_PASSWORD,
       TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PASSWORD,
+      WORKLOAD_TOKEN_SECRET,
       ...envWithoutSecrets
     } = env;
 
@@ -290,8 +292,10 @@ class ManagedSupervisor {
       });
     }
 
+    const workerToken = getWorkerToken();
+
     this.workerSession = new SupervisorSession({
-      workerToken: getWorkerToken(),
+      workerToken,
       apiUrl: env.TRIGGER_API_URL,
       instanceName: env.TRIGGER_WORKER_INSTANCE_NAME,
       managedWorkerSecret: env.MANAGED_WORKER_SECRET,
@@ -569,6 +573,7 @@ class ManagedSupervisor {
       checkpointClient: this.checkpointClient,
       computeManager: this.computeManager,
       tracing: this.tracing,
+      snapshotCallbackSecret: workerToken,
       wideEventOpts: this.wideEventOpts,
       wideEventsNoisyRoutes: this.wideEventsNoisyRoutes,
     });
@@ -603,6 +608,15 @@ class ManagedSupervisor {
         throw new Error("Image is missing");
       }
 
+      const deploymentToken = await mintDeploymentToken({
+        deployment: message.deployment.friendlyId,
+        deployment_version: message.backgroundWorker.version,
+        environment_id: message.environment.id,
+        environment_type: message.environment.type,
+        org_id: message.organization.id,
+        project_id: message.project.id,
+      });
+
       await this.workloadManager.create({
         dequeuedAt: message.dequeuedAt,
         dequeueResponseMs: timings.dequeueResponseMs,
@@ -617,6 +631,7 @@ class ManagedSupervisor {
         deploymentFriendlyId: message.deployment.friendlyId,
         deploymentVersion: message.backgroundWorker.version,
         runtime: message.backgroundWorker.runtime,
+        deploymentToken,
         runId: message.run.id,
         runFriendlyId: message.run.friendlyId,
         version: message.version,

apps/supervisor/src/services/computeSnapshotService.test.ts

@@ -20,13 +20,26 @@ function createService() {
     snapshot,
   } as unknown as ComputeWorkloadManager;
 
+  const submitSuspendCompletion = vi.fn(async () => ({ success: true }));
+
   const service = new ComputeSnapshotService({
     computeManager,
-    workerClient: {} as SupervisorHttpClient,
+    workerClient: { submitSuspendCompletion } as unknown as SupervisorHttpClient,
     wideEventOpts: { service: "supervisor-test", env: {}, enabled: false },
+    snapshotCallbackSecret: "test-secret",
   });
 
-  return { service, snapshot };
+  return { service, snapshot, submitSuspendCompletion };
+}
+
+function dispatchedMetadata(snapshot: {
+  mock: { calls: Array<Array<{ metadata?: Record<string, string> }>> };
+}) {
+  const metadata = snapshot.mock.calls[0]?.[0]?.metadata;
+  if (!metadata) {
+    throw new Error("Snapshot was not dispatched");
+  }
+  return metadata;
 }
 
 function delayedSnapshot(runnerId = "runner-1") {
@@ -38,6 +51,24 @@ function delayedSnapshot(runnerId = "runner-1") {
 }
 
 describe("ComputeSnapshotService", () => {
+  it("refuses to construct with an empty callback secret", () => {
+    const computeManager = {
+      snapshotDelayMs: DELAY_MS,
+      snapshotDispatchLimit: 1,
+      snapshot: vi.fn(async () => true),
+    } as unknown as ComputeWorkloadManager;
+
+    expect(
+      () =>
+        new ComputeSnapshotService({
+          computeManager,
+          workerClient: {} as SupervisorHttpClient,
+          wideEventOpts: { service: "supervisor-test", env: {}, enabled: false },
+          snapshotCallbackSecret: "",
+        })
+    ).toThrow();
+  });
+
   it("dispatches a scheduled snapshot after the delay", async () => {
     const { service, snapshot } = createService();
     try {
@@ -46,7 +77,12 @@ describe("ComputeSnapshotService", () => {
       await vi.waitFor(() => expect(snapshot).toHaveBeenCalledTimes(1), { timeout: 2_000 });
       expect(snapshot).toHaveBeenCalledWith({
         runnerId: "runner-1",
-        metadata: { runId: "run_1", snapshotFriendlyId: "snapshot_1" },
+        metadata: expect.objectContaining({
+          runId: "run_1",
+          snapshotFriendlyId: "snapshot_1",
+          snapshotCallbackNonce: expect.any(String),
+          snapshotCallbackToken: expect.any(String),
+        }),
       });
     } finally {
       service.stop();
@@ -121,10 +157,86 @@ describe("ComputeSnapshotService", () => {
       expect(snapshot).toHaveBeenCalledTimes(1);
       expect(snapshot).toHaveBeenCalledWith({
         runnerId: "runner-1",
-        metadata: { runId: "run_1", snapshotFriendlyId: "snapshot_2" },
+        metadata: expect.objectContaining({
+          runId: "run_1",
+          snapshotFriendlyId: "snapshot_2",
+          snapshotCallbackNonce: expect.any(String),
+          snapshotCallbackToken: expect.any(String),
+        }),
       });
     } finally {
       service.stop();
     }
   });
+
+  it("accepts a snapshot callback with the dispatched token", async () => {
+    const { service, snapshot, submitSuspendCompletion } = createService();
+    try {
+      service.schedule("run_1", delayedSnapshot());
+
+      await vi.waitFor(() => expect(snapshot).toHaveBeenCalledTimes(1), { timeout: 2_000 });
+      const metadata = dispatchedMetadata(snapshot);
+
+      const result = await service.handleCallback({
+        status: "completed",
+        instance_id: "instance_1",
+        snapshot_id: "compute_snapshot_1",
+        metadata,
+      });
+
+      expect(result).toEqual({ ok: true, status: 200 });
+      expect(submitSuspendCompletion).toHaveBeenCalledWith({
+        runId: "run_1",
+        snapshotId: "snapshot_1",
+        body: {
+          success: true,
+          checkpoint: {
+            type: "COMPUTE",
+            location: "compute_snapshot_1",
+          },
+        },
+      });
+    } finally {
+      service.stop();
+    }
+  });
+
+  it("rejects a snapshot callback without a valid token", async () => {
+    const { service, submitSuspendCompletion } = createService();
+    try {
+      const result = await service.handleCallback({
+        status: "completed",
+        instance_id: "instance_1",
+        snapshot_id: "compute_snapshot_1",
+        metadata: { runId: "run_1", snapshotFriendlyId: "snapshot_1" },
+      });
+
+      expect(result).toEqual({ ok: false, status: 401 });
+      expect(submitSuspendCompletion).not.toHaveBeenCalled();
+    } finally {
+      service.stop();
+    }
+  });
+
+  it("rejects a snapshot callback whose token is for a different snapshot", async () => {
+    const { service, snapshot, submitSuspendCompletion } = createService();
+    try {
+      service.schedule("run_1", delayedSnapshot());
+
+      await vi.waitFor(() => expect(snapshot).toHaveBeenCalledTimes(1), { timeout: 2_000 });
+      const metadata = dispatchedMetadata(snapshot);
+
+      const result = await service.handleCallback({
+        status: "completed",
+        instance_id: "instance_1",
+        snapshot_id: "compute_snapshot_1",
+        metadata: { ...metadata, snapshotFriendlyId: "snapshot_2" },
+      });
+
+      expect(result).toEqual({ ok: false, status: 401 });
+      expect(submitSuspendCompletion).not.toHaveBeenCalled();
+    } finally {
+      service.stop();
+    }
+  });
 });

apps/supervisor/src/services/computeSnapshotService.ts

@@ -1,3 +1,4 @@
+import { createHmac, randomBytes, timingSafeEqual } from "node:crypto";
 import pLimit from "p-limit";
 import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 import { parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
@@ -16,6 +17,13 @@ import {
   type WideEventOptions,
 } from "../wideEvents/index.js";
 
+const SNAPSHOT_CALLBACK_NONCE_METADATA_KEY = "snapshotCallbackNonce";
+const SNAPSHOT_CALLBACK_TOKEN_METADATA_KEY = "snapshotCallbackToken";
+
+// Domain-separation label so the callback-signing key is derived from, rather
+// than equal to, the secret used for other protocols. Bump the suffix to rotate.
+const SNAPSHOT_CALLBACK_KEY_INFO = "compute-snapshot-callback-v1";
+
 type DelayedSnapshot = {
   runnerId: string;
   runFriendlyId: string;
@@ -34,6 +42,7 @@ export type ComputeSnapshotServiceOptions = {
   workerClient: SupervisorHttpClient;
   tracing?: OtlpTraceService;
   wideEventOpts: WideEventOptions;
+  snapshotCallbackSecret: string;
 };
 
 export class ComputeSnapshotService {
@@ -48,6 +57,7 @@ export class ComputeSnapshotService {
   private readonly workerClient: SupervisorHttpClient;
   private readonly tracing?: OtlpTraceService;
   private readonly wideEventOpts: WideEventOptions;
+  private readonly snapshotCallbackKey: Buffer;
 
   constructor(opts: ComputeSnapshotServiceOptions) {
     this.computeManager = opts.computeManager;
@@ -55,6 +65,18 @@ export class ComputeSnapshotService {
     this.tracing = opts.tracing;
     this.wideEventOpts = opts.wideEventOpts;
 
+    // Reject an empty secret up front: an empty HMAC key would make callback
+    // tokens forgeable by anyone. Guarding here (rather than only at env parse)
+    // also covers the case where the secret is read from an empty file.
+    if (!opts.snapshotCallbackSecret) {
+      throw new Error("snapshotCallbackSecret must not be empty");
+    }
+    // Derive a dedicated key by domain separation so the raw secret is never
+    // used directly as a MAC key for this protocol.
+    this.snapshotCallbackKey = createHmac("sha256", opts.snapshotCallbackSecret)
+      .update(SNAPSHOT_CALLBACK_KEY_INFO)
+      .digest();
+
     this.dispatchLimit = pLimit(this.computeManager.snapshotDispatchLimit);
     this.timerWheel = new TimerWheel<DelayedSnapshot>({
       delayMs: this.computeManager.snapshotDelayMs,
@@ -146,15 +168,29 @@ export class ComputeSnapshotService {
       instanceId: body.instance_id,
       status: body.status,
       error: body.status === "failed" ? body.error : undefined,
-      metadata: body.metadata,
+      runId,
+      snapshotFriendlyId,
       durationMs: body.duration_ms,
     });
 
     if (!runId || !snapshotFriendlyId) {
-      this.logger.error("Snapshot callback missing metadata", { body });
+      this.logger.error("Snapshot callback missing metadata", {
+        status: body.status,
+        instanceId: body.instance_id,
+        metadataKeys: Object.keys(body.metadata ?? {}),
+      });
       return { ok: false as const, status: 400 };
     }
 
+    if (!this.#verifyCallbackToken(body.metadata, runId, snapshotFriendlyId)) {
+      this.logger.error("Snapshot callback failed token verification", {
+        runId,
+        snapshotFriendlyId,
+        instanceId: body.instance_id,
+      });
+      return { ok: false as const, status: 401 };
+    }
+
     this.#emitSnapshotSpan(runId, body.duration_ms, snapshotId);
 
     if (body.status === "completed") {
@@ -266,11 +302,18 @@ export class ComputeSnapshotService {
         },
       },
       async () => {
+        const callbackNonce = randomBytes(16).toString("hex");
         const result = await this.computeManager.snapshot({
           runnerId: snapshot.runnerId,
           metadata: {
             runId: snapshot.runFriendlyId,
             snapshotFriendlyId: snapshot.snapshotFriendlyId,
+            [SNAPSHOT_CALLBACK_NONCE_METADATA_KEY]: callbackNonce,
+            [SNAPSHOT_CALLBACK_TOKEN_METADATA_KEY]: this.#createCallbackToken(
+              callbackNonce,
+              snapshot.runFriendlyId,
+              snapshot.snapshotFriendlyId
+            ),
           },
         });
 
@@ -281,6 +324,51 @@ export class ComputeSnapshotService {
     );
   }
 
+  #createCallbackToken(nonce: string, runFriendlyId: string, snapshotFriendlyId: string): string {
+    return createHmac("sha256", this.snapshotCallbackKey)
+      .update(nonce)
+      .update("\0")
+      .update(runFriendlyId)
+      .update("\0")
+      .update(snapshotFriendlyId)
+      .digest("hex");
+  }
+
+  /**
+   * Verify that a callback carries a token this supervisor issued for the given
+   * run and snapshot. The token binds only the identifiers known at dispatch
+   * time (nonce, run, snapshot); it intentionally does not cover result fields
+   * such as the snapshot location or status/error, which are produced by the
+   * gateway after the snapshot and so cannot be signed in advance. Verification
+   * is also stateless, so a token is not single-use.
+   *
+   * This closes the primary risk (a caller that can merely reach the endpoint
+   * cannot mint a valid token, so cannot forge a result for an arbitrary run).
+   * It does not defend against an attacker who can observe a genuine callback
+   * and then replay it or alter its unsigned result fields - that relies on the
+   * gateway->supervisor callback channel being authenticated and encrypted.
+   */
+  #verifyCallbackToken(
+    metadata: Record<string, string> | undefined,
+    runFriendlyId: string,
+    snapshotFriendlyId: string
+  ): boolean {
+    const nonce = metadata?.[SNAPSHOT_CALLBACK_NONCE_METADATA_KEY];
+    const token = metadata?.[SNAPSHOT_CALLBACK_TOKEN_METADATA_KEY];
+
+    if (!nonce || !token) {
+      return false;
+    }
+
+    const expected = this.#createCallbackToken(nonce, runFriendlyId, snapshotFriendlyId);
+    const expectedBuffer = Buffer.from(expected, "hex");
+    const tokenBuffer = Buffer.from(token, "hex");
+
+    return (
+      expectedBuffer.length === tokenBuffer.length && timingSafeEqual(expectedBuffer, tokenBuffer)
+    );
+  }
+
   #emitSnapshotSpan(runFriendlyId: string, durationMs?: number, snapshotId?: string) {
     if (!this.tracing) return;
 

apps/supervisor/src/workloadManager/compute.ts

@@ -151,7 +151,9 @@ export class ComputeWorkloadManager implements WorkloadManager {
       TRIGGER_DEQUEUED_AT_MS: String(opts.dequeuedAt.getTime()),
       TRIGGER_POD_SCHEDULED_AT_MS: String(Date.now()),
       TRIGGER_ENV_ID: opts.envId,
-      TRIGGER_DEPLOYMENT_ID: opts.deploymentFriendlyId,
+      TRIGGER_DEPLOYMENT_ID: opts.deploymentToken ?? opts.deploymentFriendlyId,
+      // Plain friendlyId for telemetry (worker.id), so it isn't the opaque token in DEPLOYMENT_ID.
+      TRIGGER_DEPLOYMENT_FRIENDLY_ID: opts.deploymentFriendlyId,
       TRIGGER_DEPLOYMENT_VERSION: opts.deploymentVersion,
       TRIGGER_RUN_ID: opts.runFriendlyId,
       TRIGGER_SNAPSHOT_ID: opts.snapshotFriendlyId,

apps/supervisor/src/workloadManager/docker.ts

@@ -72,7 +72,9 @@ export class DockerWorkloadManager implements WorkloadManager {
       `TRIGGER_DEQUEUED_AT_MS=${opts.dequeuedAt.getTime()}`,
       `TRIGGER_POD_SCHEDULED_AT_MS=${Date.now()}`,
       `TRIGGER_ENV_ID=${opts.envId}`,
-      `TRIGGER_DEPLOYMENT_ID=${opts.deploymentFriendlyId}`,
+      `TRIGGER_DEPLOYMENT_ID=${opts.deploymentToken ?? opts.deploymentFriendlyId}`,
+      // Plain friendlyId for telemetry (worker.id), so it isn't the opaque token in DEPLOYMENT_ID.
+      `TRIGGER_DEPLOYMENT_FRIENDLY_ID=${opts.deploymentFriendlyId}`,
       `TRIGGER_DEPLOYMENT_VERSION=${opts.deploymentVersion}`,
       `TRIGGER_RUN_ID=${opts.runFriendlyId}`,
       `TRIGGER_SNAPSHOT_ID=${opts.snapshotFriendlyId}`,

apps/supervisor/src/workloadManager/kubernetes.ts

@@ -158,6 +158,11 @@ export class KubernetesWorkloadManager implements WorkloadManager {
                   },
                   {
                     name: "TRIGGER_DEPLOYMENT_ID",
+                    value: opts.deploymentToken ?? opts.deploymentFriendlyId,
+                  },
+                  {
+                    // Plain friendlyId for telemetry (worker.id), not the opaque token in DEPLOYMENT_ID.
+                    name: "TRIGGER_DEPLOYMENT_FRIENDLY_ID",
                     value: opts.deploymentFriendlyId,
                   },
                   {

apps/supervisor/src/workloadManager/types.ts

@@ -44,6 +44,8 @@ export interface WorkloadManagerCreateOptions {
   deploymentVersion: string;
   // Canonical runtime identifier (e.g. "node", "node-22", "node-24")
   runtime?: string;
+  // When set, overrides the TRIGGER_DEPLOYMENT_ID value the runner forwards as its identity header.
+  deploymentToken?: string;
   runId: string;
   runFriendlyId: string;
   snapshotId: string;

apps/supervisor/src/workloadServer/index.ts

@@ -25,6 +25,11 @@ import { type Namespace, Server, type Socket } from "socket.io";
 import { z } from "zod";
 import { env } from "../env.js";
 import { register } from "../metrics.js";
+import {
+  verifyDeploymentIdHeader,
+  workloadTokenEnforced,
+  workloadTokensEnabled,
+} from "../workloadToken.js";
 import {
   ComputeSnapshotService,
   type RunTraceContext,
@@ -86,6 +91,7 @@ type WorkloadServerOptions = {
   checkpointClient?: CheckpointClient;
   computeManager?: ComputeWorkloadManager;
   tracing?: OtlpTraceService;
+  snapshotCallbackSecret: string;
   wideEventOpts: WideEventOptions;
   /** When true, high-frequency HTTP routes also emit wide events. */
   wideEventsNoisyRoutes: boolean;
@@ -136,6 +142,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         workerClient: opts.workerClient,
         tracing: opts.tracing,
         wideEventOpts: this.wideEventOpts,
+        snapshotCallbackSecret: opts.snapshotCallbackSecret,
       });
     }
 
@@ -169,6 +176,34 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
     return this.headerValueFromRequest(req, WORKLOAD_HEADERS.PROJECT_REF);
   }
 
+  /**
+   * Verify the deployment token from the workload deployment-id header and return the verified
+   * environment_id to forward upstream. The env id is only forwarded in enforce mode: in log mode
+   * we still verify + record metrics but attach no header (so the platform never scopes). Only
+   * enforce fails a request, and only for a present-but-invalid token; absent and legacy ids pass.
+   */
+  private async authorizeWorkloadRequest(
+    req: IncomingMessage
+  ): Promise<{ ok: true; environmentId?: string } | { ok: false }> {
+    if (!workloadTokensEnabled) {
+      return { ok: true };
+    }
+
+    const result = await verifyDeploymentIdHeader(this.deploymentIdFromRequest(req), "http");
+
+    if (result.outcome === "jwt_invalid" && workloadTokenEnforced) {
+      return { ok: false };
+    }
+
+    return {
+      ok: true,
+      environmentId:
+        workloadTokenEnforced && result.outcome === "jwt_valid"
+          ? result.claims.environment_id
+          : undefined,
+    };
+  }
+
   /**
    * Sets common route meta on the wide-event state from URL params.
    */
@@ -250,11 +285,17 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "POST",
               async () => {
                 const { req, reply, params, body } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 const startResponse = await this.workerClient.startRunAttempt(
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
                   body,
-                  this.runnerIdFromRequest(req)
+                  this.runnerIdFromRequest(req),
+                  auth.environmentId
                 );
 
                 if (!startResponse.success) {
@@ -286,6 +327,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "POST",
               async () => {
                 const { req, reply, params, body } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 const runnerId = this.runnerIdFromRequest(req);
 
                 // A completion attempt invalidates any pending delayed snapshot
@@ -304,7 +350,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
                   body,
-                  runnerId
+                  runnerId,
+                  auth.environmentId
                 );
 
                 if (!completeResponse.success) {
@@ -336,6 +383,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "POST",
               async () => {
                 const { req, reply, params, body } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 const heartbeatResponse = await this.workerClient.heartbeatRun(
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
@@ -373,6 +425,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "GET",
               async () => {
                 const { reply, params, req } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 const runnerId = this.runnerIdFromRequest(req);
                 const deploymentVersion = this.deploymentVersionFromRequest(req);
                 const projectRef = this.projectRefFromRequest(req);
@@ -469,6 +526,11 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "GET",
               async () => {
                 const { req, reply, params } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 this.logger.debug("Run continuation request", { params });
 
                 // Cancel any pending delayed snapshot for this run
@@ -477,7 +539,8 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
                 const continuationResult = await this.workerClient.continueRunExecution(
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
-                  this.runnerIdFromRequest(req)
+                  this.runnerIdFromRequest(req),
+                  auth.environmentId
                 );
 
                 if (!continuationResult.success) {
@@ -511,10 +574,16 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               "GET",
               async () => {
                 const { req, reply, params } = ctx;
+                const auth = await this.authorizeWorkloadRequest(req);
+                if (!auth.ok) {
+                  reply.empty(401);
+                  return;
+                }
                 const sinceSnapshotResponse = await this.workerClient.getSnapshotsSince(
                   params.runFriendlyId,
                   params.snapshotFriendlyId,
-                  this.runnerIdFromRequest(req)
+                  this.runnerIdFromRequest(req),
+                  auth.environmentId
                 );
 
                 if (!sinceSnapshotResponse.success) {
@@ -585,9 +654,18 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
               const { req, reply, params, body } = ctx;
               reply.empty(204);
 
+              // Redact TRIGGER_DEPLOYMENT_ID before relaying to the platform.
+              const sanitizedBody =
+                body.properties && "TRIGGER_DEPLOYMENT_ID" in body.properties
+                  ? {
+                      ...body,
+                      properties: { ...body.properties, TRIGGER_DEPLOYMENT_ID: "[redacted]" },
+                    }
+                  : body;
+
               await this.workerClient.sendDebugLog(
                 params.runFriendlyId,
-                body,
+                sanitizedBody,
                 this.runnerIdFromRequest(req)
               );
             },
@@ -681,7 +759,31 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
         return;
       }
 
-      this.logger.debug("[WS] auth success", socket.data);
+      if (workloadTokensEnabled) {
+        const result = await verifyDeploymentIdHeader(socket.data.deploymentId, "ws");
+
+        if (result.outcome === "jwt_invalid" && workloadTokenEnforced) {
+          this.logger.error("[WS] deployment token verification failed", {
+            runnerId: socket.data.runnerId,
+          });
+          socket.disconnect(true);
+          return;
+        }
+
+        // Re-source the deployment id from the verified claim; the raw header may be an opaque token.
+        // A legacy bare id is itself the friendlyId, so it's safe to keep.
+        socket.data.deploymentFriendlyId =
+          result.outcome === "jwt_valid"
+            ? result.claims.deployment
+            : result.outcome === "legacy_bare"
+              ? socket.data.deploymentId
+              : undefined;
+      }
+
+      this.logger.debug("[WS] handshake complete", {
+        runnerId: socket.data.runnerId,
+        deploymentFriendlyId: socket.data.deploymentFriendlyId,
+      });
 
       next();
     });
@@ -693,7 +795,7 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
 
       const getSocketMetadata = () => {
         return {
-          deploymentId: socket.data.deploymentId,
+          deploymentId: socket.data.deploymentFriendlyId ?? socket.data.deploymentId,
           runId: socket.data.runFriendlyId,
           snapshotId: socket.data.snapshotId,
           runnerId: socket.data.runnerId,
@@ -712,8 +814,9 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
           populate: (state) => {
             state.extras.event = event;
             setMeta(state, "run_id", friendlyId);
-            if (socket.data.deploymentId) {
-              setMeta(state, "deployment_id", socket.data.deploymentId);
+            const deploymentId = socket.data.deploymentFriendlyId ?? socket.data.deploymentId;
+            if (deploymentId) {
+              setMeta(state, "deployment_id", deploymentId);
             }
             if (socket.data.runnerId) setMeta(state, "runner_id", socket.data.runnerId);
             state.extras.socket_id = socket.id;
@@ -725,6 +828,33 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
       const runConnected = (friendlyId: string) => {
         socketLogger.debug("runConnected", { ...getSocketMetadata() });
 
+        // Only the owning runner may (re)bind a run. A live socket from a *different*
+        // runner keeps its binding so an unrelated connection can't hijack the run. But
+        // the newest socket for the *same* runner is a legitimate reconnection/handoff and
+        // is allowed to take over even while the stale socket still reports connected -
+        // otherwise, during a reconnect race the fresh socket would silently stay unbound
+        // (missing continue/cancel/suspend notifications) until the dead socket times out.
+        const existing = this.runSockets.get(friendlyId);
+        if (existing && existing.id !== socket.id && existing.connected) {
+          const sameRunner =
+            !!socket.data.runnerId && existing.data.runnerId === socket.data.runnerId;
+
+          if (!sameRunner) {
+            socketLogger.warn("runConnected: run already bound to another socket", {
+              ...getSocketMetadata(),
+              friendlyId,
+              existingSocketId: existing.id,
+            });
+            return;
+          }
+
+          socketLogger.debug("runConnected: replacing stale socket for same runner", {
+            ...getSocketMetadata(),
+            friendlyId,
+            existingSocketId: existing.id,
+          });
+        }
+
         // If there's already a run ID set, we should "disconnect" it from this socket
         if (socket.data.runFriendlyId && socket.data.runFriendlyId !== friendlyId) {
           socketLogger.debug("runConnected: disconnecting existing run", {
@@ -744,6 +874,22 @@ export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
       const runDisconnected = (friendlyId: string, reason: string) => {
         socketLogger.debug("runDisconnected", { ...getSocketMetadata() });
 
+        // A newer socket may have taken over this run (same-runner reconnect race). If the
+        // run is now bound to a different socket, this stale socket must not clear the fresh
+        // binding or emit a spurious disconnect - just drop its own reference and bail.
+        const bound = this.runSockets.get(friendlyId);
+        if (bound && bound.id !== socket.id) {
+          socketLogger.debug("runDisconnected: run rebound to another socket, skipping", {
+            ...getSocketMetadata(),
+            friendlyId,
+            boundSocketId: bound.id,
+          });
+          if (socket.data.runFriendlyId === friendlyId) {
+            socket.data.runFriendlyId = undefined;
+          }
+          return;
+        }
+
         // The run is gone from this runner (crash, exit, or replaced by a new
         // run), so a pending delayed snapshot for it is stale. Genuine
         // waitpoint suspensions keep the socket connected, so this doesn't

apps/supervisor/src/workloadServer/workloadServer.auth.test.ts

@@ -0,0 +1,107 @@
+import { mintWorkloadDeploymentToken } from "@trigger.dev/core/v3";
+import { WORKLOAD_HEADERS } from "@trigger.dev/core/v3/workers";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+// Set enforce mode + secret before env.ts parses (vi.mock is hoisted above imports, so the secret
+// must be a literal here). SECRET below mirrors it for use in the test body.
+vi.mock("std-env", () => ({
+  env: {
+    TRIGGER_API_URL: "http://localhost:3030",
+    TRIGGER_WORKER_TOKEN: "test-token",
+    MANAGED_WORKER_SECRET: "test-secret",
+    OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+    WORKLOAD_TOKEN_SECRET: "integration-test-secret",
+    WORKLOAD_TOKEN_ENFORCEMENT: "enforce",
+  },
+}));
+
+const SECRET = "integration-test-secret";
+const EXP = Math.floor(Date.UTC(2032, 0, 1) / 1000);
+
+const { WorkloadServer } = await import("./index.js");
+
+const PORT = 18732;
+const BASE = `http://127.0.0.1:${PORT}`;
+
+function claims(environmentId = "env_test_123") {
+  return {
+    deployment: "deployment_test",
+    deployment_version: "20260710.1",
+    environment_id: environmentId,
+    environment_type: "PRODUCTION",
+    org_id: "org_1",
+    project_id: "proj_1",
+  };
+}
+
+// Records the args each relay method is called with so we can assert the forwarded claim.
+const calls: { getSnapshotsSince: any[][] } = { getSnapshotsSince: [] };
+
+const workerClient = {
+  getSnapshotsSince: vi.fn(async (...args: any[]) => {
+    calls.getSnapshotsSince.push(args);
+    return { success: true as const, data: { snapshots: [] } };
+  }),
+} as any;
+
+let server: InstanceType<typeof WorkloadServer>;
+
+beforeAll(async () => {
+  server = new WorkloadServer({
+    port: PORT,
+    workerClient,
+    snapshotCallbackSecret: "snapshot-callback-secret",
+    wideEventOpts: { service: "supervisor", env: { nodeId: "test" }, enabled: false },
+    wideEventsNoisyRoutes: false,
+  });
+  await server.start();
+});
+
+afterAll(async () => {
+  await server.stop();
+});
+
+function snapshotsSince(deploymentIdHeader?: string) {
+  const headers: Record<string, string> = {
+    [WORKLOAD_HEADERS.RUNNER_ID]: "runner_1",
+  };
+  if (deploymentIdHeader !== undefined) {
+    headers[WORKLOAD_HEADERS.DEPLOYMENT_ID] = deploymentIdHeader;
+  }
+  return fetch(`${BASE}/api/v1/workload-actions/runs/run_1/snapshots/since/snap_1`, { headers });
+}
+
+describe("WorkloadServer auth (enforce mode)", () => {
+  it("allows a valid token and forwards the verified environment_id", async () => {
+    const token = await mintWorkloadDeploymentToken(claims("env_forwarded_42"), SECRET, EXP);
+    const res = await snapshotsSince(token);
+
+    expect(res.status).toBe(200);
+    const lastCall = calls.getSnapshotsSince.at(-1)!;
+    // getSnapshotsSince(runId, snapshotId, runnerId, environmentId)
+    expect(lastCall[3]).toBe("env_forwarded_42");
+  });
+
+  it("rejects a token signed with the wrong secret (401) and does not relay", async () => {
+    const before = calls.getSnapshotsSince.length;
+    const badToken = await mintWorkloadDeploymentToken(claims(), "wrong-secret", EXP);
+    const res = await snapshotsSince(badToken);
+
+    expect(res.status).toBe(401);
+    expect(calls.getSnapshotsSince.length).toBe(before);
+  });
+
+  it("allows a legacy bare friendlyId and forwards no environment_id", async () => {
+    const res = await snapshotsSince("deployment_legacy_bare");
+
+    expect(res.status).toBe(200);
+    expect(calls.getSnapshotsSince.at(-1)![3]).toBeUndefined();
+  });
+
+  it("allows an absent token and forwards no environment_id", async () => {
+    const res = await snapshotsSince(undefined);
+
+    expect(res.status).toBe(200);
+    expect(calls.getSnapshotsSince.at(-1)![3]).toBeUndefined();
+  });
+});

apps/supervisor/src/workloadServer/workloadServer.authLog.test.ts

@@ -0,0 +1,87 @@
+import { mintWorkloadDeploymentToken } from "@trigger.dev/core/v3";
+import { WORKLOAD_HEADERS } from "@trigger.dev/core/v3/workers";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+// Log mode: mint + verify + metrics, but the platform must NOT be scoped, so no environment_id is
+// forwarded even for a valid token. (vi.mock is hoisted; secret literal here, mirrored below.)
+vi.mock("std-env", () => ({
+  env: {
+    TRIGGER_API_URL: "http://localhost:3030",
+    TRIGGER_WORKER_TOKEN: "test-token",
+    MANAGED_WORKER_SECRET: "test-secret",
+    OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+    WORKLOAD_TOKEN_SECRET: "integration-test-secret",
+    WORKLOAD_TOKEN_ENFORCEMENT: "log",
+  },
+}));
+
+const SECRET = "integration-test-secret";
+const EXP = Math.floor(Date.UTC(2032, 0, 1) / 1000);
+
+const { WorkloadServer } = await import("./index.js");
+
+const PORT = 18733;
+const BASE = `http://127.0.0.1:${PORT}`;
+
+const claims = {
+  deployment: "deployment_test",
+  deployment_version: "20260710.1",
+  environment_id: "env_should_not_forward",
+  environment_type: "PRODUCTION",
+  org_id: "org_1",
+  project_id: "proj_1",
+};
+
+const calls: { getSnapshotsSince: any[][] } = { getSnapshotsSince: [] };
+
+const workerClient = {
+  getSnapshotsSince: vi.fn(async (...args: any[]) => {
+    calls.getSnapshotsSince.push(args);
+    return { success: true as const, data: { snapshots: [] } };
+  }),
+} as any;
+
+let server: InstanceType<typeof WorkloadServer>;
+
+beforeAll(async () => {
+  server = new WorkloadServer({
+    port: PORT,
+    workerClient,
+    snapshotCallbackSecret: "snapshot-callback-secret",
+    wideEventOpts: { service: "supervisor", env: { nodeId: "test" }, enabled: false },
+    wideEventsNoisyRoutes: false,
+  });
+  await server.start();
+});
+
+afterAll(async () => {
+  await server.stop();
+});
+
+describe("WorkloadServer auth (log mode)", () => {
+  it("allows a valid token but forwards no environment_id", async () => {
+    const token = await mintWorkloadDeploymentToken(claims, SECRET, EXP);
+    const res = await fetch(`${BASE}/api/v1/workload-actions/runs/run_1/snapshots/since/snap_1`, {
+      headers: {
+        [WORKLOAD_HEADERS.RUNNER_ID]: "runner_1",
+        [WORKLOAD_HEADERS.DEPLOYMENT_ID]: token,
+      },
+    });
+
+    expect(res.status).toBe(200);
+    expect(calls.getSnapshotsSince.at(-1)![3]).toBeUndefined();
+  });
+
+  it("does not reject an invalid token in log mode", async () => {
+    const badToken = await mintWorkloadDeploymentToken(claims, "wrong-secret", EXP);
+    const res = await fetch(`${BASE}/api/v1/workload-actions/runs/run_1/snapshots/since/snap_1`, {
+      headers: {
+        [WORKLOAD_HEADERS.RUNNER_ID]: "runner_1",
+        [WORKLOAD_HEADERS.DEPLOYMENT_ID]: badToken,
+      },
+    });
+
+    expect(res.status).toBe(200);
+    expect(calls.getSnapshotsSince.at(-1)![3]).toBeUndefined();
+  });
+});

apps/supervisor/src/workloadToken.ts

@@ -0,0 +1,85 @@
+import {
+  classifyDeploymentIdHeader,
+  mintWorkloadDeploymentToken,
+  type WorkloadDeploymentTokenClaims,
+  type WorkloadDeploymentTokenInput,
+} from "@trigger.dev/core/v3";
+import { Counter } from "prom-client";
+import { env } from "./env.js";
+import { register } from "./metrics.js";
+
+const secret = env.WORKLOAD_TOKEN_SECRET;
+
+// Absolute expiry (epoch seconds) shared by every mint, so tokens stay byte-deterministic per
+// deployment regardless of when/where a pod is created.
+const tokenExpSeconds = Math.floor(new Date(env.WORKLOAD_TOKEN_EXP).getTime() / 1000);
+
+/** Mint + verify run in "log" (dry-run) and "enforce"; the env superRefine guarantees a secret then. */
+export const workloadTokensEnabled = env.WORKLOAD_TOKEN_ENFORCEMENT !== "disabled";
+
+/** Only "enforce" rejects a present-but-invalid token; "log" observes and always allows. */
+export const workloadTokenEnforced = env.WORKLOAD_TOKEN_ENFORCEMENT === "enforce";
+
+const mintCounter = new Counter({
+  name: "workload_token_minted_total",
+  help: "Deployment tokens minted and injected into TRIGGER_DEPLOYMENT_ID at pod creation",
+  labelNames: ["env_type"] as const,
+  registers: [register],
+});
+
+export type WorkloadAuthTransport = "http" | "ws";
+export type WorkloadAuthOutcome = "jwt_valid" | "jwt_invalid" | "legacy_bare" | "token_absent";
+
+const verifyCounter = new Counter({
+  name: "workload_auth_verify_total",
+  help: "Runner-boundary token verification outcomes at the supervisor workload server",
+  labelNames: ["outcome", "transport", "env_type"] as const,
+  registers: [register],
+});
+
+export async function mintDeploymentToken(
+  claims: WorkloadDeploymentTokenInput
+): Promise<string | undefined> {
+  if (!workloadTokensEnabled || !secret) {
+    return undefined;
+  }
+
+  const token = await mintWorkloadDeploymentToken(claims, secret, tokenExpSeconds);
+  mintCounter.inc({ env_type: claims.environment_type });
+  return token;
+}
+
+export type VerifiedDeploymentHeader =
+  | { outcome: "jwt_valid"; claims: WorkloadDeploymentTokenClaims }
+  | { outcome: "jwt_invalid" | "legacy_bare" | "token_absent"; claims?: undefined };
+
+/**
+ * Verify the deployment-id header value and record the outcome. "jwt_valid" returns the claims so the
+ * caller can forward the verified environment_id upstream; other outcomes carry no trusted data.
+ */
+export async function verifyDeploymentIdHeader(
+  value: string | undefined,
+  transport: WorkloadAuthTransport
+): Promise<VerifiedDeploymentHeader> {
+  const result = await classify(value);
+  verifyCounter.inc({
+    outcome: result.outcome,
+    transport,
+    env_type: result.outcome === "jwt_valid" ? result.claims.environment_type : "unknown",
+  });
+  return result;
+}
+
+async function classify(value: string | undefined): Promise<VerifiedDeploymentHeader> {
+  if (!value || !secret) {
+    return { outcome: "token_absent" };
+  }
+
+  const result = await classifyDeploymentIdHeader(value, secret);
+
+  if (result.outcome === "jwt_valid" && result.claims) {
+    return { outcome: "jwt_valid", claims: result.claims };
+  }
+
+  return { outcome: result.outcome === "jwt_valid" ? "jwt_invalid" : result.outcome };
+}

apps/webapp/app/components/StaleAssetRecovery.test.ts

@@ -0,0 +1,56 @@
+// @vitest-environment jsdom
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { staleAssetRecoveryScript } from "./StaleAssetRecovery";
+
+// Each staleAssetRecoveryScript() call models a fresh page load: it reads the shared
+// sessionStorage budget and returns its own `recover`. We drive recover() directly rather
+// than dispatching resource-error events, so accumulated window listeners never fire.
+describe("staleAssetRecoveryScript", () => {
+  let reload: ReturnType<typeof vi.fn>;
+
+  beforeEach(() => {
+    sessionStorage.clear();
+    reload = vi.fn();
+    vi.stubGlobal("location", { reload });
+    vi.stubGlobal("navigator", { onLine: true });
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+  });
+
+  it("reloads on a recovery", () => {
+    staleAssetRecoveryScript().recover();
+    expect(reload).toHaveBeenCalledTimes(1);
+  });
+
+  it("reloads only once per page even if several assets fail (re-entrancy guard)", () => {
+    const { recover } = staleAssetRecoveryScript();
+    recover();
+    recover();
+    recover();
+    expect(reload).toHaveBeenCalledTimes(1);
+  });
+
+  it("stops reloading once the budget is spent across reloads", () => {
+    staleAssetRecoveryScript().recover(); // reload 1
+    staleAssetRecoveryScript().recover(); // reload 2
+    staleAssetRecoveryScript().recover(); // budget spent -> no reload
+    expect(reload).toHaveBeenCalledTimes(2);
+  });
+
+  it("does not reload when offline", () => {
+    vi.stubGlobal("navigator", { onLine: false });
+    staleAssetRecoveryScript().recover();
+    expect(reload).not.toHaveBeenCalled();
+  });
+
+  it("does not reload when sessionStorage is unavailable", () => {
+    vi.spyOn(Storage.prototype, "setItem").mockImplementation(() => {
+      throw new Error("blocked");
+    });
+    staleAssetRecoveryScript().recover();
+    expect(reload).not.toHaveBeenCalled();
+  });
+});

apps/webapp/app/components/StaleAssetRecovery.tsx

@@ -1,272 +1,94 @@
-// Recovers from deploys rotating the content-hashed /build assets out from
-// under a page. Policy:
+// Recovers from a rolling deploy rotating the content-hashed /build assets out from
+// under a page. Each image serves only its own build and hard-404s unknown hashes, so
+// a client can request a hash the serving replica doesn't have and get missing styles
+// or a failed asset load. On such a /build load failure we do a bounded full document
+// reload: the fresh document (and, under sticky routing, all of its assets) lands on a
+// single live build, so the asset resolves. Bounded via sessionStorage so it can never
+// loop; when the budget is spent it stops rather than reloading forever.
 //
-// 1. A working page is never touched just because a new build exists. Remix
-//    loader fetches (?_data=) are stamped with X-Build-Id by the server;
-//    when a navigation reveals a different build, the navigation is turned
-//    into a full document load — the user lands on the new build without
-//    ever seeing an incompatible state.
-// 2. Only real incompatibility (a /build stylesheet/script 404 or a failed
-//    chunk import) triggers recovery: an overlay goes up immediately, the
-//    script polls /build-version with backoff, and reloads once the server
-//    reports a different build than the page was rendered with
-//    (window.__remixManifest.version). If versions never diverge or the
-//    reload budget (one per observed build, 2 total) is spent, the overlay
-//    offers a manual reload instead of leaving a dead page.
-// 3. Before a recovery reload, form fields and scroll position are
-//    snapshotted to sessionStorage and restored (best-effort) after the
-//    reload. history.state is deliberately not restored — the Remix router
-//    owns it, and reviving a stale one can desync the router.
-//
-// Must render before <Links /> so the listener precedes the stylesheet.
-const script = `(function () {
-  var VKEY = "trigger:assetRecovery";
-  var SKEY = "trigger:recoverySnapshot";
+// Deliberately minimal — no fetch interception, no build-version polling, no server
+// build-id contract, no form snapshot, no blocking overlay.
+
+// The recovery logic runs as an inline <script> injected before <Links /> (see the
+// component below), so it must execute before the app bundle and before the stylesheet
+// can fail to load. It is authored as a normal, type-checked and lint-checked function
+// and serialized with .toString() at render time — NOT hand-written into a string — so
+// the logic is real code the compiler and linter can see. Because it is serialized, it
+// must stay fully self-contained: no imports, no references to module scope, and plain
+// ES that the bundler won't rewrite to reach a hoisted helper. It returns its `recover`
+// closure purely so the unit test can drive the logic directly (the inline IIFE that
+// runs in the browser ignores the return value).
+export function staleAssetRecoveryScript() {
+  var KEY = "trigger:assetReload";
   var MAX_RELOADS = 2;
-  var RESET_AFTER = 300000;
-  var SNAPSHOT_TTL = 30000;
-  var CHECK_DELAYS = [0, 2000, 4000, 8000, 15000, 30000];
+  var WINDOW_MS = 300000;
   var recovering = false;
-  var navigated = false;
 
-  function ownVersion() {
-    return window.__remixManifest && window.__remixManifest.version;
-  }
-
-  function readJson(key) {
+  function budgetAllows() {
     try {
-      return JSON.parse(sessionStorage.getItem(key) || "null");
-    } catch (e) {
-      return undefined;
-    }
-  }
-
-  function writeJson(key, value) {
-    try {
-      sessionStorage.setItem(key, JSON.stringify(value));
+      var raw = sessionStorage.getItem(KEY);
+      var state = raw ? (JSON.parse(raw) as { n: number; t: number }) : { n: 0, t: 0 };
+      if (Date.now() - state.t > WINDOW_MS) state = { n: 0, t: 0 };
+      if (state.n >= MAX_RELOADS) return false;
+      sessionStorage.setItem(KEY, JSON.stringify({ n: state.n + 1, t: Date.now() }));
       return true;
-    } catch (e) {
+    } catch {
+      // Storage blocked (private mode / quota): can't bound reloads, so don't auto-reload.
       return false;
     }
   }
 
-  // ---- form + scroll snapshot ------------------------------------------
-
-  function takeSnapshot() {
-    var fields = [];
-    var els = document.querySelectorAll("input, textarea, select");
-    for (var i = 0; i < els.length; i++) {
-      var el = els[i];
-      var type = (el.getAttribute("type") || "").toLowerCase();
-      if (type === "password" || type === "file" || type === "hidden") continue;
-      fields.push({
-        i: i,
-        id: el.id || null,
-        name: el.name || null,
-        tag: el.tagName,
-        value: el.value,
-        checked: el.checked === true,
-      });
-    }
-    writeJson(SKEY, {
-      t: Date.now(),
-      path: location.pathname,
-      scrollY: window.scrollY,
-      fields: fields,
-    });
-  }
-
-  function setNativeValue(el, value) {
-    // Go through the prototype setter so React's value tracking notices the
-    // change when the input event fires.
-    var proto =
-      el.tagName === "TEXTAREA"
-        ? window.HTMLTextAreaElement
-        : el.tagName === "SELECT"
-          ? window.HTMLSelectElement
-          : window.HTMLInputElement;
-    var descriptor = Object.getOwnPropertyDescriptor(proto.prototype, "value");
-    if (descriptor && descriptor.set) {
-      descriptor.set.call(el, value);
-    } else {
-      el.value = value;
-    }
-  }
-
-  function restoreSnapshot() {
-    var snapshot = readJson(SKEY);
-    try {
-      sessionStorage.removeItem(SKEY);
-    } catch (e) {}
-    if (!snapshot || snapshot.path !== location.pathname) return;
-    if (Date.now() - (snapshot.t || 0) > SNAPSHOT_TTL) return;
-    var els = document.querySelectorAll("input, textarea, select");
-    for (var i = 0; i < snapshot.fields.length; i++) {
-      var field = snapshot.fields[i];
-      var el = (field.id && document.getElementById(field.id)) || els[field.i];
-      if (!el || el.tagName !== field.tag || (el.name || null) !== field.name) continue;
-      if (el.type === "checkbox" || el.type === "radio") {
-        // click() keeps React state in sync with the DOM
-        if (el.checked !== field.checked) el.click();
-      } else if (field.value != null && el.value !== field.value) {
-        setNativeValue(el, field.value);
-        el.dispatchEvent(new Event("input", { bubbles: true }));
-        el.dispatchEvent(new Event("change", { bubbles: true }));
-      }
-    }
-    if (snapshot.scrollY) window.scrollTo(0, snapshot.scrollY);
-  }
-
-  window.addEventListener("load", function () {
-    setTimeout(restoreSnapshot, 100);
-  });
-
-  function doReload() {
-    takeSnapshot();
-    location.reload();
-  }
-
-  // ---- scenario 1: working page, navigation as the update point --------
-
-  var origFetch = window.fetch;
-  window.fetch = function (input) {
-    var result = origFetch.apply(this, arguments);
-    try {
-      var url = typeof input === "string" ? input : input && input.url;
-      if (url && url.indexOf("_data=") !== -1) {
-        result.then(function (response) {
-          var server = response.headers.get("X-Build-Id");
-          var mine = ownVersion();
-          if (server && mine && server !== mine && !navigated && !recovering) {
-            navigated = true;
-            var target = new URL(url, location.origin);
-            target.searchParams.delete("_data");
-            location.assign(target.toString());
-          }
-        }, function () {});
-      }
-    } catch (e) {}
-    return result;
-  };
-
-  // ---- scenario 2: page is actually broken ------------------------------
-
-  function showOverlay(final) {
-    if (!document.body) {
-      document.addEventListener("DOMContentLoaded", function () {
-        showOverlay(final);
-      });
-      return;
-    }
-    // ASCII only: documents are served without an explicit charset, so
-    // non-ASCII here can render as mojibake on a broken page.
-    var text = final ? "This page failed to load properly. Please reload." : "Loading...";
-    var existing = document.getElementById("stale-asset-overlay-text");
-    if (existing) {
-      existing.textContent = text;
-      if (final) document.getElementById("stale-asset-overlay-button").style.display = "";
-      return;
-    }
-    var overlay = document.createElement("div");
-    overlay.id = "stale-asset-overlay";
-    overlay.style.cssText =
-      "position:fixed;inset:0;z-index:2147483647;display:flex;flex-direction:column;gap:16px;align-items:center;justify-content:center;background:#121317;color:#d7d9dd;font:15px/1.5 system-ui,sans-serif;text-align:center;padding:24px";
-    var message = document.createElement("p");
-    message.id = "stale-asset-overlay-text";
-    message.textContent = text;
-    var button = document.createElement("button");
-    button.id = "stale-asset-overlay-button";
-    button.textContent = "Reload";
-    button.style.cssText =
-      "border:0;border-radius:4px;padding:7px 18px;background:#6366f1;color:#fff;font:inherit;cursor:pointer" +
-      (final ? "" : ";display:none");
-    button.onclick = doReload;
-    overlay.appendChild(message);
-    overlay.appendChild(button);
-    document.body.appendChild(overlay);
-  }
-
-  function reloadFor(serverVersion) {
-    var state = readJson(VKEY);
-    if (state === undefined) return showOverlay(true);
-    state = state || {};
-    if (Date.now() - (state.t || 0) > RESET_AFTER) state = {};
-    // One reload per observed server version, MAX_RELOADS total: a page that
-    // is still broken after reloading for this build gets the manual overlay
-    // instead of reloading again.
-    if (state.v === serverVersion || (state.reloads || 0) >= MAX_RELOADS) return showOverlay(true);
-    if (!writeJson(VKEY, { v: serverVersion, reloads: (state.reloads || 0) + 1, t: Date.now() })) {
-      return showOverlay(true);
-    }
-    doReload();
-  }
-
-  function check(attempt) {
-    origFetch("/build-version", { cache: "no-store" })
-      .then(function (response) {
-        return response.json();
-      })
-      .then(function (data) {
-        var mine = ownVersion();
-        if (data && data.version && mine && data.version !== mine) {
-          reloadFor(data.version);
-        } else {
-          scheduleNext(attempt);
-        }
-      })
-      .catch(function () {
-        scheduleNext(attempt);
-      });
-  }
-
-  function scheduleNext(attempt) {
-    var next = attempt + 1;
-    if (next >= CHECK_DELAYS.length) return showOverlay(true);
-    setTimeout(function () {
-      check(next);
-    }, CHECK_DELAYS[next]);
-  }
-
   function recover() {
+    // One recovery per page: a broken load fails several /build assets at once and each
+    // fires its own error event before location.reload() commits — without this guard a
+    // single incident would burn the entire reload budget.
     if (recovering) return;
     recovering = true;
-    showOverlay(false);
-    // __remixManifest is set by an inline script near the end of body; wait
-    // for the document to finish parsing before comparing versions.
-    if (document.readyState === "loading") {
-      document.addEventListener("DOMContentLoaded", function () {
-        check(0);
-      });
-    } else {
-      check(0);
-    }
+    // Don't reload into the browser's offline error page.
+    if (navigator.onLine === false) return;
+    if (budgetAllows()) location.reload();
   }
 
+  // Non-bubbling resource load failures (stylesheet, modulepreload, entry <script>) at
+  // document load — the failure class nothing else covers. Capture phase is required.
   window.addEventListener(
     "error",
     function (event) {
-      var el = event.target;
-      if (!el || el === window) return;
-      var url = el.tagName === "LINK" ? el.href : el.tagName === "SCRIPT" ? el.src : null;
+      var el = event.target as Element | null;
+      if (!el || typeof el.tagName !== "string") return; // window/global errors have no tagName
+      var url =
+        el.tagName === "LINK"
+          ? (el as HTMLLinkElement).href
+          : el.tagName === "SCRIPT"
+            ? (el as HTMLScriptElement).src
+            : null;
       if (url && url.indexOf("/build/") !== -1) recover();
     },
     true
   );
+
+  // Raw dynamic import() failures in app code. (Remix reloads its own route chunks, so
+  // that path rarely reaches here.) The message URL isn't reliable cross-browser, so
+  // match the chunk-load error shape; the once-guard + bounded budget make a rare stray
+  // reload harmless.
   window.addEventListener("unhandledrejection", function (event) {
-    var message = event.reason && event.reason.message;
+    var message = (event.reason && event.reason.message) || "";
     if (
-      typeof message === "string" &&
       /dynamically imported module|Importing a module script failed|ChunkLoadError/i.test(message)
     ) {
       recover();
     }
   });
-})();`;
+
+  return { recover };
+}
 
 export function StaleAssetRecovery({ isProduction }: { isProduction: boolean }) {
   if (!isProduction) {
     return null;
   }
 
-  return <script dangerouslySetInnerHTML={{ __html: script }} />;
+  return (
+    <script dangerouslySetInnerHTML={{ __html: `(${staleAssetRecoveryScript.toString()})()` }} />
+  );
 }

apps/webapp/app/entry.server.tsx

@@ -300,6 +300,7 @@ singleton("SentryTenantContextProcessor", () => {
 
 export { apiRateLimiter } from "./services/apiRateLimit.server";
 export { engineRateLimiter } from "./services/engineRateLimit.server";
+export { otlpRateLimiter } from "./services/otlpRateLimit.server";
 export { runWithHttpContext } from "./services/httpAsyncStorage.server";
 export { tenantContextMiddleware } from "./services/tenantContextResolver.server";
 export { socketIo } from "./v3/handleSocketIo.server";

apps/webapp/app/env.server.ts

@@ -85,6 +85,29 @@ const S2EnvSchema = z.preprocess(
   ])
 );
 
+// Previously published secret values must never be accepted, including when
+// an existing deployment or external secret manager still supplies one.
+const INSECURE_SECRET_VALUES = [
+  "managed-secret",
+  "2818143646516f6fffd707b36f334bbb",
+  "44da78b7bbb0dfe709cf38931d25dcdd",
+  "f686147ab967943ebbe9ed3b496e465a",
+  "447c29678f9eaf289e9c4b70d3dd8a7f",
+];
+
+// Escape hatch for deployments that can't rotate a published default yet (e.g.
+// ENCRYPTION_KEY protects existing data). Read raw: a refine can't see the
+// sibling parsed flag.
+const allowInsecureDefaultSecrets = ["true", "1"].includes(
+  (process.env.ALLOW_INSECURE_DEFAULT_SECRETS ?? "").toLowerCase().trim()
+);
+
+const isNotInsecureSecret = (value: string) =>
+  allowInsecureDefaultSecrets || !INSECURE_SECRET_VALUES.includes(value);
+
+const INSECURE_SECRET_MESSAGE =
+  "must not be a known-insecure published default; set a strong, unique value. If you cannot rotate it yet (e.g. it protects existing encrypted data or active sessions), set ALLOW_INSECURE_DEFAULT_SECRETS=1 to boot while you migrate.";
+
 const EnvironmentSchema = z
   .object({
     NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
@@ -188,14 +211,15 @@ const EnvironmentSchema = z
     // Control-plane cache relax knobs. Unset -> defaults (DEFAULT_CP_CACHE_TTL_MS / _MAX_ENTRIES).
     CONTROL_PLANE_CACHE_TTL_MS: z.coerce.number().int().optional(),
     CONTROL_PLANE_CACHE_MAX_ENTRIES: z.coerce.number().int().optional(),
-    SESSION_SECRET: z.string(),
-    MAGIC_LINK_SECRET: z.string(),
+    SESSION_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
+    MAGIC_LINK_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
     ENCRYPTION_KEY: z
       .string()
       .refine(
         (val) => Buffer.from(val, "utf8").length === 32,
         "ENCRYPTION_KEY must be exactly 32 bytes"
-      ),
+      )
+      .refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
     WHITELISTED_EMAILS: z
       .string()
       .refine(isValidRegex, "WHITELISTED_EMAILS must be a valid regex.")
@@ -547,6 +571,22 @@ const EnvironmentSchema = z
     API_RATE_LIMIT_JWT_WINDOW: z.string().default("1m"),
     API_RATE_LIMIT_JWT_TOKENS: z.coerce.number().int().default(60),
 
+    // Per-IP rate limit for the unauthenticated OTLP ingestion endpoints
+    // (/otel/*). Bounds unauthenticated request rates. Opt-in
+    // (disabled by default): because it keys on the source IP, it is only
+    // safe to enable when each client presents a distinct IP through a proxy
+    // that appends the real client IP to X-Forwarded-For. Enabling it where
+    // many clients share one egress IP (e.g. behind NAT or a shared proxy)
+    // would collapse that traffic into a single bucket and could throttle
+    // legitimate telemetry. Set OTLP_RATE_LIMIT_ENABLED=1 to enable, then tune
+    // OTLP_RATE_LIMIT_MAX / OTLP_RATE_LIMIT_WINDOW for expected volume.
+    OTLP_RATE_LIMIT_ENABLED: z.string().default("0"),
+    OTLP_RATE_LIMIT_WINDOW: z
+      .string()
+      .regex(/^\d+ ?(?:ms|s|m|h|d)$/)
+      .default("1m"),
+    OTLP_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(3000),
+
     DEPOT_TOKEN: z.string().optional(),
     DEPOT_ORG_ID: z.string().optional(),
     DEPOT_REGION: z.string().default("us-east-1"),
@@ -668,7 +708,18 @@ const EnvironmentSchema = z
     EVENTS_LOAD_SHEDDING_THRESHOLD: z.coerce.number().int().default(100000),
     EVENTS_LOAD_SHEDDING_ENABLED: z.string().default("1"),
 
-    MANAGED_WORKER_SECRET: z.string().default("managed-secret"),
+    MANAGED_WORKER_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
+
+    // Allow booting with a known-insecure published default secret. Temporary
+    // bridge for deployments that can't rotate yet; rotate as soon as possible.
+    ALLOW_INSECURE_DEFAULT_SECRETS: BoolEnv.default(false),
+
+    // Tenant scoping on worker actions is header-driven (folded into the engine snapshot read) and
+    // needs no flag. This is only the no-header fallback: when "1", a worker action on a run created
+    // after WORKLOAD_TOKEN_CUTOFF without a verified env header is rejected; runs on or before the
+    // cutoff pass (grandfathered). Default off = no run-row read, byte-for-byte today's behavior.
+    WORKLOAD_CREATED_AT_GATE_ENABLED: z.string().default("0"),
+    WORKLOAD_TOKEN_CUTOFF: z.string().datetime().optional(),
 
     // Development OTEL environment variables
     DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
@@ -1311,6 +1362,9 @@ const EnvironmentSchema = z
     // claim TTL), how long a waiter blocks before timing out, and the
     // waiter poll interval.
     TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS: z.coerce.number().int().positive().default(30),
+    // Pipeline floor: the claim never shrinks below this even for a short customer key TTL, so it
+    // can't expire mid-pipeline and let a loser re-claim (cross-DB duplicate under the split).
+    TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS: z.coerce.number().int().positive().default(5),
     TRIGGER_MOLLIFIER_CLAIM_WAIT_MS: z.coerce.number().int().positive().default(5_000),
     TRIGGER_MOLLIFIER_CLAIM_POLL_MS: z.coerce.number().int().positive().default(25),
 
@@ -2087,3 +2141,24 @@ const EnvironmentSchema = z
 
 export type Environment = z.infer<typeof EnvironmentSchema>;
 export const env = EnvironmentSchema.parse(process.env);
+
+if (env.ALLOW_INSECURE_DEFAULT_SECRETS) {
+  const insecure = (
+    [
+      ["SESSION_SECRET", env.SESSION_SECRET],
+      ["MAGIC_LINK_SECRET", env.MAGIC_LINK_SECRET],
+      ["ENCRYPTION_KEY", env.ENCRYPTION_KEY],
+      ["MANAGED_WORKER_SECRET", env.MANAGED_WORKER_SECRET],
+    ] as const
+  )
+    .filter(([, value]) => INSECURE_SECRET_VALUES.includes(value))
+    .map(([name]) => name);
+
+  if (insecure.length > 0) {
+    console.warn(
+      `⚠️  ALLOW_INSECURE_DEFAULT_SECRETS is enabled and these secrets still use a known-insecure published default: ${insecure.join(
+        ", "
+      )}. This is insecure - rotate them as soon as you can.`
+    );
+  }
+}

apps/webapp/app/models/runtimeEnvironment.server.ts

@@ -274,19 +274,17 @@ export async function findEnvironmentFromRun(
 ): Promise<EnvironmentFromRun | null> {
   // Run-ops scalars (runTags/batchId/runtimeEnvironmentId) from the run store; the env half is
   // resolved via the control-plane resolver so the run-ops DB can split without a cross-DB join.
-  const taskRun = await runStore.findRun(
-    {
-      id: runId,
-    },
-    {
-      select: {
-        runTags: true,
-        batchId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    tx ?? $replica
-  );
+  const select = {
+    runTags: true,
+    batchId: true,
+    runtimeEnvironmentId: true,
+  } as const;
+  let taskRun = await runStore.findRun({ id: runId }, { select }, tx ?? $replica);
+  if (!taskRun) {
+    // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary before
+    // treating it as absent, so runMetadataUpdated doesn't drop a live run's final metadata + publish.
+    taskRun = await runStore.findRun({ id: runId }, { select }, prisma);
+  }
   if (!taskRun) {
     return null;
   }

apps/webapp/app/models/schedules.server.ts

@@ -1,4 +1,4 @@
-import type { Prisma } from "~/db.server";
+import type { Prisma, PrismaClientOrTransaction, TaskSchedule } from "@trigger.dev/database";
 
 export function scheduleUniqWhereClause(
   projectId: string,
@@ -35,3 +35,48 @@ export function scheduleWhereClause(
     deduplicationKey: scheduleId,
   };
 }
+
+/**
+ * Resolve a schedule's visibility for an environment-scoped caller.
+ *
+ *   - "visible": the schedule exists in the project and has at least one
+ *     instance bound to `environmentId` (or has no instances yet).
+ *   - "hidden": the schedule exists but none of its instances live in the
+ *     caller's environment.
+ *   - "missing": no schedule exists for the (project, scheduleId) pair.
+ *
+ * A schedule can be bound to several environments at once, so visibility
+ * mirrors the "some instance is in this environment" rule the schedule
+ * list uses: a schedule that is listed for a key must also be readable
+ * and mutable by that key. This still rejects cross-environment access to
+ * schedules the caller has no instance in, and `scheduleWhereClause`
+ * already confines the lookup to the caller's project.
+ *
+ * The tri-state lets PUT (upsert) disambiguate "hidden" (refuse) from
+ * "missing" (fall through to create). DELETE/GET treat hidden and
+ * missing the same way.
+ */
+export type ScheduleEnvVisibility =
+  | { status: "visible"; schedule: TaskSchedule }
+  | { status: "hidden" }
+  | { status: "missing" };
+
+export async function getScheduleEnvVisibility(
+  prisma: PrismaClientOrTransaction,
+  projectId: string,
+  scheduleId: string,
+  environmentId: string
+): Promise<ScheduleEnvVisibility> {
+  const schedule = await prisma.taskSchedule.findFirst({
+    where: scheduleWhereClause(projectId, scheduleId),
+    include: { instances: { select: { environmentId: true } } },
+  });
+
+  if (!schedule) return { status: "missing" };
+
+  const { instances, ...rest } = schedule;
+  if (instances.length === 0) return { status: "visible", schedule: rest };
+  const scoped = instances.some((i) => i.environmentId === environmentId);
+  if (!scoped) return { status: "hidden" };
+  return { status: "visible", schedule: rest };
+}

apps/webapp/app/models/waitpointTag.server.ts

@@ -8,10 +8,14 @@ export async function createWaitpointTag({
   tag,
   environmentId,
   projectId,
+  residency,
 }: {
   tag: string;
   environmentId: string;
   projectId: string;
+  // Residency from the env mint kind: a tag has no owning run, so a minted-new env pins it to NEW
+  // instead of defaulting to the draining legacy DB.
+  residency?: "NEW" | "LEGACY";
 }) {
   if (tag.trim().length === 0) return;
 
@@ -19,11 +23,15 @@ export async function createWaitpointTag({
 
   while (attempts < MAX_RETRIES) {
     try {
-      return await runStore.upsertWaitpointTag({
-        environmentId,
-        name: tag,
-        projectId,
-      });
+      return await runStore.upsertWaitpointTag(
+        {
+          environmentId,
+          name: tag,
+          projectId,
+        },
+        undefined,
+        residency
+      );
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {
         // Handle unique constraint violation (conflict)

apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts

@@ -42,29 +42,36 @@ export class ApiWaitpointPresenter extends BasePresenter {
     return this.trace("call", async (span) => {
       // The store routes by the waitpointId's residency (id shape) and reads the owning
       // store's replica. waitpointId is pre-decoded from the friendlyId via WaitpointId.toId.
-      const waitpoint = await this.runStore.findWaitpoint({
-        where: {
-          id: waitpointId,
-          environmentId: environment.id,
-        },
-        select: {
-          id: true,
-          friendlyId: true,
-          type: true,
-          status: true,
-          idempotencyKey: true,
-          userProvidedIdempotencyKey: true,
-          idempotencyKeyExpiresAt: true,
-          inactiveIdempotencyKey: true,
-          output: true,
-          outputType: true,
-          outputIsError: true,
-          completedAfter: true,
-          completedAt: true,
-          createdAt: true,
-          tags: true,
-        },
-      });
+      const where = {
+        id: waitpointId,
+        environmentId: environment.id,
+      };
+      const select = {
+        id: true,
+        friendlyId: true,
+        type: true,
+        status: true,
+        idempotencyKey: true,
+        userProvidedIdempotencyKey: true,
+        idempotencyKeyExpiresAt: true,
+        inactiveIdempotencyKey: true,
+        output: true,
+        outputType: true,
+        outputIsError: true,
+        completedAfter: true,
+        completedAt: true,
+        createdAt: true,
+        tags: true,
+      } as const;
+
+      let waitpoint = await this.runStore.findWaitpoint({ where, select });
+
+      // Read-your-writes on a public GET: a just-minted token may not be on the owning store's
+      // replica yet, so a replica miss would 404 a live token. Re-read the owning primary before
+      // concluding it doesn't exist (mirrors the metadata GET loader + the complete/callback paths).
+      if (!waitpoint) {
+        waitpoint = await this.runStore.findWaitpointOnPrimary({ where, select });
+      }
 
       if (!waitpoint) {
         logger.error(`WaitpointPresenter: Waitpoint not found`, {

apps/webapp/app/presenters/v3/BatchListPresenter.server.ts

@@ -43,6 +43,28 @@ type BatchRow = {
   batchVersion: string;
 };
 
+// Composite keyset cursor "<createdAt-epoch-ms>_<id>". Ordering is by createdAt then id: a batch id is
+// a cuid (legacy) OR a run-ops id (new), and the two schemes occupy different lexical ranges, so `id`
+// alone is not a valid chronological order across the residency split. `id` is the stable tiebreak.
+// Old plain-id cursors (no "_") decode to undefined and restart from page 1 (self-healing).
+type BatchCursor = { createdAt: Date; id: string };
+function encodeBatchCursor(row: BatchCursor): string {
+  return `${row.createdAt.getTime()}_${row.id}`;
+}
+function decodeBatchCursor(cursor: string | undefined): BatchCursor | undefined {
+  if (!cursor) return undefined;
+  const sep = cursor.indexOf("_");
+  if (sep === -1) return undefined;
+  const ms = Number(cursor.slice(0, sep));
+  const id = cursor.slice(sep + 1);
+  // Number.isFinite accepts e.g. 1e20, but new Date(1e20) is Invalid Date — reject it so a malformed
+  // URL cursor self-heals to page 1 instead of reaching Prisma with an invalid date.
+  const createdAt = new Date(ms);
+  if (!Number.isFinite(ms) || Number.isNaN(createdAt.getTime()) || id.length === 0)
+    return undefined;
+  return { createdAt, id };
+}
+
 export class BatchListPresenter extends BasePresenter {
   // Optional run-ops read-routing. Omitted (single-DB / self-host) => everything
   // reads from `_replica` exactly as today (passthrough). Field names are local to
@@ -86,17 +108,16 @@ export class BatchListPresenter extends BasePresenter {
       return scan(passthrough);
     }
 
-    const newRows = await scan(this.readRoute.runOpsNew ?? passthrough);
+    // Always read BOTH stores and merge. The old "skip legacy when new fills the page" shortcut is
+    // unsound across the residency split: legacy cuid ids ("c…") sort ABOVE new run-ops ids ("0…")
+    // under id order, so a new-only page can hide pre-flip legacy batches that belong ahead of it.
+    // Ordering is by createdAt (id tiebreak), which is chronologically correct across both schemes.
+    const [newRows, legacyRows] = await Promise.all([
+      scan(this.readRoute.runOpsNew ?? passthrough),
+      scan(this.readRoute.runOpsLegacyReplica ?? passthrough),
+    ]);
 
-    // New DB filled the page — skip the legacy read entirely; older rows fall on a later page.
-    if (newRows.length >= pageSize + 1) {
-      return newRows;
-    }
-
-    const legacyRows = await scan(this.readRoute.runOpsLegacyReplica ?? passthrough);
-
-    // De-dupe by id (new wins), re-sort under the page's keyset order, re-apply the over-fetch
-    // LIMIT — reproduces the pageSize+1 window a single union scan would return.
+    // De-dupe by id (new wins), re-sort under the page's keyset order, re-apply the over-fetch LIMIT.
     const byId = new Map<string, BatchRow>();
     for (const row of newRows) {
       byId.set(row.id, row);
@@ -107,10 +128,16 @@ export class BatchListPresenter extends BasePresenter {
       }
     }
 
-    // codepoint comparator (NEVER localeCompare): BatchTaskRun.id is ASCII (cuid or run-ops id).
-    const sign = direction === "forward" ? 1 : -1; // forward => DESC; backward => ASC
+    // forward => newest-first (createdAt DESC), backward => oldest-first (ASC); id is the stable
+    // tiebreak (ASCII codepoint, NEVER localeCompare).
+    const sign = direction === "forward" ? 1 : -1;
     return Array.from(byId.values())
-      .sort((a, b) => (a.id < b.id ? sign : a.id > b.id ? -sign : 0))
+      .sort((a, b) => {
+        const at = a.createdAt.getTime();
+        const bt = b.createdAt.getTime();
+        if (at !== bt) return at < bt ? sign : -sign;
+        return a.id < b.id ? sign : a.id > b.id ? -sign : 0;
+      })
       .slice(0, pageSize + 1);
   }
 
@@ -212,11 +239,28 @@ export class BatchListPresenter extends BasePresenter {
     }
     const createdAtLte: Date | undefined = time.to;
 
+    // Composite (createdAt, id) keyset — see encodeBatchCursor. An old plain-id cursor decodes to
+    // undefined and restarts from page 1.
+    const keyCursor = decodeBatchCursor(cursor);
+
     const batches = await this.#scanBatchTaskRun(pageSize, direction, (client) =>
       client.batchTaskRun.findMany({
         where: {
           runtimeEnvironmentId: environmentId,
-          ...(cursor ? { id: direction === "forward" ? { lt: cursor } : { gt: cursor } } : {}),
+          ...(keyCursor
+            ? {
+                OR:
+                  direction === "forward"
+                    ? [
+                        { createdAt: { lt: keyCursor.createdAt } },
+                        { createdAt: keyCursor.createdAt, id: { lt: keyCursor.id } },
+                      ]
+                    : [
+                        { createdAt: { gt: keyCursor.createdAt } },
+                        { createdAt: keyCursor.createdAt, id: { gt: keyCursor.id } },
+                      ],
+              }
+            : {}),
           ...(friendlyId ? { friendlyId } : {}),
           ...(statuses && statuses.length > 0
             ? { status: { in: statuses }, batchVersion: { not: "v1" } }
@@ -230,7 +274,10 @@ export class BatchListPresenter extends BasePresenter {
               }
             : {}),
         },
-        orderBy: { id: direction === "forward" ? "desc" : "asc" },
+        orderBy: [
+          { createdAt: direction === "forward" ? "desc" : "asc" },
+          { id: direction === "forward" ? "desc" : "asc" },
+        ],
         take: pageSize + 1,
         select: {
           id: true,
@@ -248,23 +295,24 @@ export class BatchListPresenter extends BasePresenter {
 
     const hasMore = batches.length > pageSize;
 
-    //get cursors for next and previous pages
+    //get cursors for next and previous pages (composite (createdAt, id) keyset)
+    const cur = (row?: BatchRow) => (row ? encodeBatchCursor(row) : undefined);
     let next: string | undefined;
     let previous: string | undefined;
     switch (direction) {
       case "forward":
-        previous = cursor ? batches.at(0)?.id : undefined;
+        previous = cursor ? cur(batches.at(0)) : undefined;
         if (hasMore) {
-          next = batches[pageSize - 1]?.id;
+          next = cur(batches[pageSize - 1]);
         }
         break;
       case "backward":
         batches.reverse();
         if (hasMore) {
-          previous = batches[1]?.id;
-          next = batches[pageSize]?.id;
+          previous = cur(batches[1]);
+          next = cur(batches[pageSize]);
         } else {
-          next = batches[pageSize - 1]?.id;
+          next = cur(batches[pageSize - 1]);
         }
         break;
     }

apps/webapp/app/presenters/v3/BatchPresenter.server.ts

@@ -47,13 +47,25 @@ export class BatchPresenter extends BasePresenter {
     // The BatchTaskRun (run-ops) is read through the run store, which routes by residency. The
     // runtimeEnvironment (control-plane) is resolved separately because the cross-seam FK is
     // dropped, so the batch row cannot single-SQL join to control-plane RuntimeEnvironment.
-    const batch = await this.runStore.findBatchTaskRunByFriendlyId(
+    let batch = await this.runStore.findBatchTaskRunByFriendlyId(
       batchId,
       environmentId,
       { include: BATCH_INCLUDE },
       this._replica
     );
 
+    // Read-your-writes: findBatchTaskRunByFriendlyId defaults to (and here reads) the replica, so a
+    // batch created within the replica's apply window returns null under lag. Re-read from the owning
+    // primary on a miss so a live batch's detail page never spuriously 404s ("Batch not found").
+    if (!batch) {
+      batch = await this.runStore.findBatchTaskRunByFriendlyId(
+        batchId,
+        environmentId,
+        { include: BATCH_INCLUDE },
+        this._prisma
+      );
+    }
+
     if (!batch) {
       throw new Error("Batch not found");
     }

apps/webapp/app/presenters/v3/LimitsPresenter.server.ts

@@ -155,7 +155,8 @@ export class LimitsPresenter extends BasePresenter {
     const activeBranchCount = await this._replica.runtimeEnvironment.count({
       where: {
         projectId,
-        branchName: {
+        type: "PREVIEW",
+        parentEnvironmentId: {
           not: null,
         },
         archivedAt: null,

apps/webapp/app/routes/_app.github.callback/route.tsx

@@ -1,6 +1,9 @@
 import { type LoaderFunctionArgs } from "@remix-run/node";
 import { z } from "zod";
-import { validateGitHubAppInstallSession } from "~/services/gitHubSession.server";
+import {
+  destroyGitHubAppInstallSession,
+  validateGitHubAppInstallSession,
+} from "~/services/gitHubSession.server";
 import { linkGitHubAppInstallation, updateGitHubAppInstallation } from "~/services/gitHub.server";
 import { logger } from "~/services/logger.server";
 import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
@@ -75,6 +78,15 @@ export async function loader({ request }: LoaderFunctionArgs) {
     return redirectWithErrorMessage(redirectTo, request, "Failed to install GitHub app");
   }
 
+  // The install session is single-use: once a callback consumes an
+  // installation_id, invalidate the state cookie so the same initiation
+  // cannot be replayed against other installation_ids.
+  const clearInstallSession = await destroyGitHubAppInstallSession(cookieHeader);
+  const consumingSession = (response: Response) => {
+    response.headers.append("Set-Cookie", clearInstallSession);
+    return response;
+  };
+
   switch (callbackData.setup_action) {
     case "install": {
       const [error] = await tryCatch(
@@ -85,23 +97,33 @@ export async function loader({ request }: LoaderFunctionArgs) {
         logger.error("Failed to link GitHub App installation", {
           error,
         });
-        return redirectWithErrorMessage(redirectTo, request, "Failed to install GitHub app");
+        return consumingSession(
+          await redirectWithErrorMessage(redirectTo, request, "Failed to install GitHub app")
+        );
       }
 
-      return redirectWithSuccessMessage(redirectTo, request, "GitHub App installed successfully");
+      return consumingSession(
+        await redirectWithSuccessMessage(redirectTo, request, "GitHub App installed successfully")
+      );
     }
 
     case "update": {
-      const [error] = await tryCatch(updateGitHubAppInstallation(callbackData.installation_id));
+      const [error] = await tryCatch(
+        updateGitHubAppInstallation(callbackData.installation_id, organizationId)
+      );
 
       if (error) {
         logger.error("Failed to update GitHub App installation", {
           error,
         });
-        return redirectWithErrorMessage(redirectTo, request, "Failed to update GitHub App");
+        return consumingSession(
+          await redirectWithErrorMessage(redirectTo, request, "Failed to update GitHub App")
+        );
       }
 
-      return redirectWithSuccessMessage(redirectTo, request, "GitHub App updated successfully");
+      return consumingSession(
+        await redirectWithSuccessMessage(redirectTo, request, "GitHub App updated successfully")
+      );
     }
 
     case "request": {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments/route.tsx

@@ -303,6 +303,7 @@ export default function Page() {
                               <RuntimeIcon
                                 runtime={deployment.runtime}
                                 runtimeVersion={deployment.runtimeVersion}
+                                withLabel
                               />
                             </TableCell>
                             <TableCell to={path} isSelected={isSelected}>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables.new/route.tsx

@@ -55,6 +55,7 @@ import {
 } from "~/utils/pathBuilder";
 import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
 import { EnvironmentVariableKey } from "~/v3/environmentVariables/repository";
+import { findUnauthorizedEnvironmentId } from "~/v3/writableEnvironments";
 
 const Variable = z.object({
   key: EnvironmentVariableKey,
@@ -164,6 +165,31 @@ export const action = dashboardAction(
       return json(submission.reply({ formErrors: ["Project not found"] }));
     }
 
+    // The submitted `environmentIds` are user-supplied. Shared env types are
+    // writable by any member; a DEV env only by its owner. See
+    // findUnauthorizedEnvironmentId.
+    const submittedEnvs = await prisma.runtimeEnvironment.findMany({
+      where: {
+        projectId: project.id,
+        id: { in: submission.value.environmentIds },
+      },
+      select: { id: true, type: true, orgMember: { select: { userId: true } } },
+    });
+    const unauthorizedEnvironmentId = findUnauthorizedEnvironmentId(
+      submittedEnvs,
+      submission.value.environmentIds,
+      userId
+    );
+    if (unauthorizedEnvironmentId) {
+      return json(
+        submission.reply({
+          fieldErrors: {
+            environmentIds: ["One or more of the selected environments is not writable by you."],
+          },
+        })
+      );
+    }
+
     const repository = new EnvironmentVariablesRepository(prisma);
     const result = await repository.create(project.id, {
       ...submission.value,

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.environment-variables/route.tsx

@@ -78,6 +78,7 @@ import {
   v3NewEnvironmentVariablesPath,
 } from "~/utils/pathBuilder";
 import { EnvironmentVariablesRepository } from "~/v3/environmentVariables/environmentVariablesRepository.server";
+import { findUnauthorizedEnvironmentId } from "~/v3/writableEnvironments";
 import {
   DeleteEnvironmentVariableValue,
   EditEnvironmentVariableValue,
@@ -267,6 +268,24 @@ export const action = dashboardAction(
       return json(submission.reply({ formErrors: ["Project not found"] }));
     }
 
+    // Per-env write gate for the mutating value actions: `environmentId` is a
+    // user-supplied hidden field and the repository only checks project
+    // membership. Mirrors the create route's check.
+    if (submission.value.action === "edit" || submission.value.action === "delete") {
+      const submittedEnvs = await prisma.runtimeEnvironment.findMany({
+        where: { projectId: project.id, id: submission.value.environmentId },
+        select: { id: true, type: true, orgMember: { select: { userId: true } } },
+      });
+      const unauthorizedEnvironmentId = findUnauthorizedEnvironmentId(
+        submittedEnvs,
+        [submission.value.environmentId],
+        userId
+      );
+      if (unauthorizedEnvironmentId) {
+        return json(submission.reply({ formErrors: ["This environment is not writable by you."] }));
+      }
+    }
+
     switch (submission.value.action) {
       case "edit": {
         const repository = new EnvironmentVariablesRepository(prisma);

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.schedules.$scheduleParam/route.tsx

@@ -6,7 +6,6 @@ import { z } from "zod";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { LinkButton } from "~/components/primitives/Buttons";
 import { ScheduleInspector } from "~/components/schedules/ScheduleInspector";
-import { prisma } from "~/db.server";
 import { useEnvironment } from "~/hooks/useEnvironment";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
@@ -78,11 +77,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
   // `_format=json` → return JSON instead of redirecting; caller stays put.
   const wantsJson = formData.get("_format") === "json";
 
-  const project = await prisma.project.findFirst({
-    where: {
-      slug: projectParam,
-    },
-  });
+  const project = await findProjectBySlug(organizationSlug, projectParam, userId);
 
   if (!project) {
     const message = `No project found with slug ${projectParam}`;

apps/webapp/app/routes/account.authorization-code.$authorizationCode/route.tsx

@@ -1,14 +1,19 @@
 import { CheckCircleIcon } from "@heroicons/react/24/solid";
-import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
-import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { Form } from "@remix-run/react";
+import type { ActionFunctionArgs, LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { typedjson, useTypedActionData, useTypedLoaderData } from "remix-typedjson";
 import { z } from "zod";
 import { AppContainer, MainCenteredContainer } from "~/components/layout/AppLayout";
+import { Button } from "~/components/primitives/Buttons";
 import { Callout } from "~/components/primitives/Callout";
 import { Header1 } from "~/components/primitives/Headers";
 import { Icon } from "~/components/primitives/Icon";
 import { Paragraph } from "~/components/primitives/Paragraph";
 import { logger } from "~/services/logger.server";
-import { createPersonalAccessTokenFromAuthorizationCode } from "~/services/personalAccessToken.server";
+import {
+  createPersonalAccessTokenFromAuthorizationCode,
+  isAuthorizationCodeMintable,
+} from "~/services/personalAccessToken.server";
 import { requireUserId } from "~/services/session.server";
 
 const ParamsSchema = z.object({
@@ -20,49 +25,57 @@ const SearchParamsSchema = z.object({
   clientName: z.string().optional(),
 });
 
-export const loader = async ({ request, params }: LoaderFunctionArgs) => {
-  const userId = await requireUserId(request);
-
+function parseParams(params: unknown) {
   const parsedParams = ParamsSchema.safeParse(params);
-
   if (!parsedParams.success) {
     logger.info("Invalid params", { params });
-    throw new Response(undefined, {
-      status: 400,
-      statusText: "Invalid params",
-    });
+    throw new Response(undefined, { status: 400, statusText: "Invalid params" });
   }
+  return parsedParams.data;
+}
 
+function parseSearch(request: Request) {
   const url = new URL(request.url);
   const searchObject = Object.fromEntries(url.searchParams.entries());
-
   const searchParams = SearchParamsSchema.safeParse(searchObject);
-
   const source = (searchParams.success ? searchParams.data.source : undefined) ?? "cli";
   const clientName = (searchParams.success ? searchParams.data.clientName : undefined) ?? "unknown";
+  return { source, clientName };
+}
+
+// The loader only renders a consent screen; minting/binding a PAT happens in
+// the `action`, behind an explicit "Authorize" POST.
+export const loader = async ({ request, params }: LoaderFunctionArgs) => {
+  await requireUserId(request);
+
+  const { authorizationCode } = parseParams(params);
+  const { source, clientName } = parseSearch(request);
+
+  const mintable = await isAuthorizationCodeMintable(authorizationCode);
+
+  return typedjson({
+    status: mintable ? ("consent" as const) : ("invalid" as const),
+    source,
+    clientName,
+  });
+};
+
+export const action = async ({ request, params }: ActionFunctionArgs) => {
+  const userId = await requireUserId(request);
+
+  const { authorizationCode } = parseParams(params);
+  const { source, clientName } = parseSearch(request);
 
   try {
-    const _personalAccessToken = await createPersonalAccessTokenFromAuthorizationCode(
-      parsedParams.data.authorizationCode,
-      userId
-    );
-    return typedjson({
-      success: true as const,
-      source,
-      clientName,
-    });
+    await createPersonalAccessTokenFromAuthorizationCode(authorizationCode, userId);
+    return typedjson({ success: true as const, source, clientName });
   } catch (error) {
     if (error instanceof Response) {
       throw error;
     }
 
     if (error instanceof Error) {
-      return typedjson({
-        success: false as const,
-        error: error.message,
-        source,
-        clientName,
-      });
+      return typedjson({ success: false as const, error: error.message, source, clientName });
     }
 
     logger.error(JSON.stringify(error));
@@ -74,32 +87,78 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
 };
 
 export default function Page() {
-  const result = useTypedLoaderData<typeof loader>();
+  const loaderData = useTypedLoaderData<typeof loader>();
+  const actionData = useTypedActionData<typeof action>();
 
+  // After the consent POST: success or failure.
+  if (actionData) {
+    return (
+      <AuthShell>
+        {actionData.success ? (
+          <div>
+            <Header1 className="mb-2 flex items-center gap-1">
+              <Icon icon={CheckCircleIcon} className="h-6 w-6 text-emerald-500" /> Successfully
+              authenticated
+            </Header1>
+            <Paragraph>
+              {getInstructionsForSource(actionData.source, actionData.clientName)}
+            </Paragraph>
+          </div>
+        ) : (
+          <div>
+            <Header1 className="mb-2">Authentication failed</Header1>
+            <Callout variant="error" className="my-2">
+              {actionData.error}
+            </Callout>
+            <Paragraph spacing>
+              There was a problem authenticating you, please try logging in with your CLI again.
+            </Paragraph>
+          </div>
+        )}
+      </AuthShell>
+    );
+  }
+
+  // Initial GET: invalid/expired code, or the consent prompt.
+  if (loaderData.status === "invalid") {
+    return (
+      <AuthShell>
+        <div>
+          <Header1 className="mb-2">Authentication failed</Header1>
+          <Callout variant="error" className="my-2">
+            This login link is invalid or has expired.
+          </Callout>
+          <Paragraph spacing>
+            Please try logging in with your CLI again to get a fresh link.
+          </Paragraph>
+        </div>
+      </AuthShell>
+    );
+  }
+
+  return (
+    <AuthShell>
+      <div className="flex flex-col gap-4">
+        <Header1>Authorize login</Header1>
+        <Paragraph>{getConsentPrompt(loaderData.source, loaderData.clientName)}</Paragraph>
+        <Form method="post">
+          <Button type="submit" variant="primary/medium" fullWidth>
+            Authorize
+          </Button>
+        </Form>
+        <Paragraph variant="extra-small">
+          Only authorize if you started this login yourself. If you didn't, close this page.
+        </Paragraph>
+      </div>
+    </AuthShell>
+  );
+}
+
+function AuthShell({ children }: { children: React.ReactNode }) {
   return (
     <AppContainer>
       <MainCenteredContainer className="max-w-88">
-        <div className="flex flex-col items-center space-y-4">
-          {result.success ? (
-            <div>
-              <Header1 className="mb-2 flex items-center gap-1">
-                <Icon icon={CheckCircleIcon} className="h-6 w-6 text-emerald-500" /> Successfully
-                authenticated
-              </Header1>
-              <Paragraph>{getInstructionsForSource(result.source, result.clientName)}</Paragraph>
-            </div>
-          ) : (
-            <div>
-              <Header1 className="mb-2">Authentication failed</Header1>
-              <Callout variant="error" className="my-2">
-                {result.error}
-              </Callout>
-              <Paragraph spacing>
-                There was a problem authenticating you, please try logging in with your CLI again.
-              </Paragraph>
-            </div>
-          )}
-        </div>
+        <div className="flex flex-col items-center space-y-4">{children}</div>
       </MainCenteredContainer>
     </AppContainer>
   );
@@ -113,6 +172,18 @@ const prettyClientNames: Record<string, string> = {
   "claude-ai": "Claude Desktop",
 };
 
+function getConsentPrompt(source: string, clientName: string) {
+  if (source === "mcp") {
+    const pretty = prettyClientNames[clientName] ?? clientName;
+    if (pretty && pretty !== "unknown") {
+      return `Authorize ${pretty} to access your Trigger.dev account?`;
+    }
+    return `Authorize this MCP client to access your Trigger.dev account?`;
+  }
+
+  return `Authorize the Trigger.dev CLI to access your account?`;
+}
+
 function getInstructionsForSource(source: string, clientName: string) {
   if (source === "mcp") {
     if (clientName) {

apps/webapp/app/routes/api.v1.authorization-code.ts

@@ -3,7 +3,12 @@ import { json } from "@remix-run/server-runtime";
 import type { CreateAuthorizationCodeResponse } from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
+import {
+  AuthorizationCodeRateLimitError,
+  checkAuthorizationCodeMintRateLimit,
+} from "~/services/authCodeRateLimiter.server";
 import { createAuthorizationCode } from "~/services/personalAccessToken.server";
+import { extractClientIp } from "~/utils/extractClientIp.server";
 
 /** Used to create an AuthorizationCode, that can then be used to obtain a Personal Access Token by logging in with the provided URL */
 export async function action({ request }: ActionFunctionArgs) {
@@ -14,8 +19,24 @@ export async function action({ request }: ActionFunctionArgs) {
     return { status: 405, body: "Method Not Allowed" };
   }
 
-  //there is no authentication on this endpoint, anyone can create an AuthorizationCode.
-  //they're only used to allow a user to login, when they'll then receive a Personal Access Token
+  //this endpoint is unauthenticated (codes only allow a user to log in), so it's
+  //rate-limited per client IP. Keyed by X-Forwarded-For; if there's no trustworthy
+  //client IP we skip the limit rather than bucket everyone together. Self-hosters
+  //wanting per-IP limiting should front the app with a proxy that sets X-Forwarded-For.
+  const clientIp = extractClientIp(request.headers.get("x-forwarded-for"));
+  if (clientIp) {
+    try {
+      await checkAuthorizationCodeMintRateLimit(clientIp);
+    } catch (error) {
+      if (error instanceof AuthorizationCodeRateLimitError) {
+        return json(
+          { error: "Too many requests, please try again later." },
+          { status: 429, headers: { "Retry-After": Math.ceil(error.retryAfter / 1000).toString() } }
+        );
+      }
+      throw error;
+    }
+  }
 
   try {
     const authorizationCode = await createAuthorizationCode();

apps/webapp/app/routes/api.v1.batches.$batchId.ts

@@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
+    // A just-created batch may not yet have replicated to the read replica this client-less
+    // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through
+    // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes,
+    // e.g. api.v3.runs.$runId).
+    shouldRetryNotFound: true,
     findResource: (params, auth) => {
       return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, {
         include: { errors: true },

apps/webapp/app/routes/api.v1.deployments.$deploymentId.cancel.ts

@@ -45,7 +45,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
     const deploymentService = new DeploymentService();
 
     return await deploymentService
-      .cancelDeployment(authenticatedEnv, deploymentId, {
+      .cancelDeployment({ id: authenticatedEnv.id }, deploymentId, {
         canceledReason: body.data.reason,
       })
       .match(

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts

@@ -33,21 +33,23 @@ const { action, loader } = createActionApiRoute(
   },
   async ({ authentication, body, params }) => {
     try {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runFriendlyId,
-          runtimeEnvironmentId: authentication.environment.id,
+      const where = {
+        friendlyId: params.runFriendlyId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          realtimeStreamsVersion: true,
+          streamBasinName: true,
         },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            realtimeStreamsVersion: true,
-            streamBasinName: true,
-          },
-        },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run
+      // that exists. Re-read the owning primary on a replica miss.
+      const run =
+        (await runStore.findRun(where, args, $replica)) ??
+        (await runStore.findRunOnPrimary(where, args));
 
       if (!run) {
         return json({ error: "Run not found" }, { status: 404 });

apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts

@@ -39,20 +39,22 @@ const { action, loader } = createActionApiRoute(
   },
   async ({ authentication, body, params }) => {
     try {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runFriendlyId,
-          runtimeEnvironmentId: authentication.environment.id,
+      const where = {
+        friendlyId: params.runFriendlyId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          realtimeStreamsVersion: true,
         },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            realtimeStreamsVersion: true,
-          },
-        },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run
+      // that exists. Re-read the owning primary on a replica miss.
+      const run =
+        (await runStore.findRun(where, args, $replica)) ??
+        (await runStore.findRunOnPrimary(where, args));
 
       if (!run) {
         return json({ error: "Run not found" }, { status: 404 });

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -64,6 +64,19 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     );
   }
 
+  // Read-your-writes: a run drained from the buffer to the primary but not yet replicated misses
+  // both the replica read and the buffer. Re-read the owning primary before 404ing.
+  const primaryRun = await runStore.findRunOnPrimary(
+    { friendlyId: parsed.data.runId, runtimeEnvironmentId: env.id },
+    { select: { metadata: true, metadataType: true } }
+  );
+  if (primaryRun) {
+    return json(
+      { metadata: primaryRun.metadata, metadataType: primaryRun.metadataType },
+      { status: 200 }
+    );
+  }
+
   return json({ error: "Run not found" }, { status: 404 });
 }
 

apps/webapp/app/routes/api.v1.schedules.$scheduleId.activate.ts

@@ -2,7 +2,7 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
-import { scheduleUniqWhereClause, scheduleWhereClause } from "~/models/schedules.server";
+import { getScheduleEnvVisibility, scheduleUniqWhereClause } from "~/models/schedules.server";
 import { ViewSchedulePresenter } from "~/presenters/v3/ViewSchedulePresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -34,14 +34,17 @@ export async function action({ request, params }: ActionFunctionArgs) {
   }
 
   try {
-    const existingSchedule = await prisma.taskSchedule.findFirst({
-      where: scheduleWhereClause(
-        authenticationResult.environment.projectId,
-        parsedParams.data.scheduleId
-      ),
-    });
-
-    if (!existingSchedule) {
+    // Env-scoped API keys can only toggle schedules that have an instance in
+    // their own environment. Without this a key scoped to one environment
+    // could enable/disable a schedule that only runs in another environment
+    // of the same project.
+    const visibility = await getScheduleEnvVisibility(
+      prisma,
+      authenticationResult.environment.projectId,
+      parsedParams.data.scheduleId,
+      authenticationResult.environment.id
+    );
+    if (visibility.status !== "visible") {
       return json({ error: "Schedule not found" }, { status: 404 });
     }
 

apps/webapp/app/routes/api.v1.schedules.$scheduleId.deactivate.ts

@@ -2,7 +2,7 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
-import { scheduleUniqWhereClause, scheduleWhereClause } from "~/models/schedules.server";
+import { getScheduleEnvVisibility, scheduleUniqWhereClause } from "~/models/schedules.server";
 import { ViewSchedulePresenter } from "~/presenters/v3/ViewSchedulePresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -34,14 +34,17 @@ export async function action({ request, params }: ActionFunctionArgs) {
   }
 
   try {
-    const existingSchedule = await prisma.taskSchedule.findFirst({
-      where: scheduleWhereClause(
-        authenticationResult.environment.projectId,
-        parsedParams.data.scheduleId
-      ),
-    });
-
-    if (!existingSchedule) {
+    // Env-scoped API keys can only toggle schedules that have an instance in
+    // their own environment. Without this a key scoped to one environment
+    // could enable/disable a schedule that only runs in another environment
+    // of the same project.
+    const visibility = await getScheduleEnvVisibility(
+      prisma,
+      authenticationResult.environment.projectId,
+      parsedParams.data.scheduleId,
+      authenticationResult.environment.id
+    );
+    if (visibility.status !== "visible") {
       return json({ error: "Schedule not found" }, { status: 404 });
     }
 

apps/webapp/app/routes/api.v1.schedules.$scheduleId.ts

@@ -5,7 +5,7 @@ import { UpdateScheduleOptions } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { Prisma, prisma } from "~/db.server";
 import { clientSafeErrorMessage } from "~/utils/prismaErrors";
-import { scheduleUniqWhereClause } from "~/models/schedules.server";
+import { getScheduleEnvVisibility, scheduleUniqWhereClause } from "~/models/schedules.server";
 import { ViewSchedulePresenter } from "~/presenters/v3/ViewSchedulePresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -38,6 +38,16 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   switch (method) {
     case "DELETE": {
+      const visibility = await getScheduleEnvVisibility(
+        prisma,
+        authenticationResult.environment.projectId,
+        parsedParams.data.scheduleId,
+        authenticationResult.environment.id
+      );
+      if (visibility.status !== "visible") {
+        return json({ error: "Schedule not found" }, { status: 404 });
+      }
+
       try {
         const deletedSchedule = await prisma.taskSchedule.delete({
           where: scheduleUniqWhereClause(
@@ -76,6 +86,19 @@ export async function action({ request, params }: ActionFunctionArgs) {
         return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 });
       }
 
+      // Env-scoped API keys can't see or mutate a schedule whose
+      // instances live in a different environment. "hidden" → refuse;
+      // "missing" → fall through to the upsert's create path.
+      const visibility = await getScheduleEnvVisibility(
+        prisma,
+        authenticationResult.environment.projectId,
+        parsedParams.data.scheduleId,
+        authenticationResult.environment.id
+      );
+      if (visibility.status === "hidden") {
+        return json({ error: "Schedule not found" }, { status: 404 });
+      }
+
       const service = new UpsertTaskScheduleService();
 
       try {
@@ -137,6 +160,16 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     );
   }
 
+  const visibility = await getScheduleEnvVisibility(
+    prisma,
+    authenticationResult.environment.projectId,
+    parsedParams.data.scheduleId,
+    authenticationResult.environment.id
+  );
+  if (visibility.status !== "visible") {
+    return json({ error: "Schedule not found" }, { status: 404 });
+  }
+
   const presenter = new ViewSchedulePresenter();
 
   const result = await presenter.call({

apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts

@@ -75,7 +75,7 @@ const { action, loader } = createActionApiRoute(
     // SDK exposes via `ctx.run.id`). Internally `Session.currentRunId`
     // stores the TaskRun.id cuid, so resolve before handing to the
     // optimistic-claim service.
-    const callingRun = await runStore.findRun(
+    let callingRun = await runStore.findRun(
       {
         friendlyId: body.callingRunId,
         runtimeEnvironmentId: authentication.environment.id,
@@ -83,6 +83,19 @@ const { action, loader } = createActionApiRoute(
       { select: { id: true } },
       $replica
     );
+    if (!callingRun) {
+      // Replica lag: `callingRunId` is the agent's own live run (it is executing this request), so it
+      // exists on the owning primary even when the read replica has not caught up. Re-read the primary
+      // before 404ing — otherwise a lagging replica turns a legitimate handoff into a spurious
+      // "callingRunId not found in this environment".
+      callingRun = await runStore.findRunOnPrimary(
+        {
+          friendlyId: body.callingRunId,
+          runtimeEnvironmentId: authentication.environment.id,
+        },
+        { select: { id: true } }
+      );
+    }
     if (!callingRun) {
       return json({ error: "callingRunId not found in this environment" }, { status: 404 });
     }

apps/webapp/app/routes/api.v1.token.ts

@@ -4,6 +4,10 @@ import type { GetPersonalAccessTokenResponse } from "@trigger.dev/core/v3";
 import { GetPersonalAccessTokenRequestSchema } from "@trigger.dev/core/v3";
 import { generateErrorMessage } from "zod-error";
 import { logger } from "~/services/logger.server";
+import {
+  AuthorizationCodeRateLimitError,
+  checkAuthorizationCodeTokenPollRateLimit,
+} from "~/services/authCodeRateLimiter.server";
 import { getPersonalAccessTokenFromAuthorizationCode } from "~/services/personalAccessToken.server";
 import { clientSafeErrorMessage } from "~/utils/prismaErrors";
 
@@ -25,6 +29,20 @@ export async function action({ request }: ActionFunctionArgs) {
     return json({ error: generateErrorMessage(body.error.issues) }, { status: 422 });
   }
 
+  // Per-code rate limit (keyed by the code, not the IP, so the CLI's poll loop
+  // isn't broken behind a shared NAT).
+  try {
+    await checkAuthorizationCodeTokenPollRateLimit(body.data.authorizationCode);
+  } catch (error) {
+    if (error instanceof AuthorizationCodeRateLimitError) {
+      return json(
+        { error: "Too many requests, please try again later." },
+        { status: 429, headers: { "Retry-After": Math.ceil(error.retryAfter / 1000).toString() } }
+      );
+    }
+    throw error;
+  }
+
   try {
     const personalAccessToken = await getPersonalAccessTokenFromAuthorizationCode(
       body.data.authorizationCode

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts

@@ -36,13 +36,22 @@ export async function action({ request, params }: ActionFunctionArgs) {
     // Resolve wherever the waitpoint resides. The store routes by the waitpoint id's residency
     // (id-shape) and probes both run-ops DBs, so a token on either store resolves; the env is
     // resolved below from the row via the control-plane resolver.
-    const waitpoint = await runStore.findWaitpoint({
+    let waitpoint = await runStore.findWaitpoint({
       where: {
         id: waitpointId,
       },
       select: { id: true, status: true, environmentId: true },
     });
 
+    if (!waitpoint) {
+      // Read-your-writes: a token whose callback fires right after mint may not have replicated
+      // yet. Re-read the owning primary before 404ing (mirrors complete.ts's primary fallback).
+      waitpoint = await runStore.findWaitpointOnPrimary({
+        where: { id: waitpointId },
+        select: { id: true, status: true, environmentId: true },
+      });
+    }
+
     if (!waitpoint) {
       return json({ error: "Waitpoint not found" }, { status: 404 });
     }

apps/webapp/app/routes/api.v1.waitpoints.tokens.ts

@@ -16,6 +16,7 @@ import {
   type PrismaClientOrTransaction,
 } from "~/db.server";
 import { type AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
 import { logger } from "~/services/logger.server";
 import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import {
@@ -58,6 +59,16 @@ const { action } = createActionApiRoute(
 
       const timeout = await parseDelay(body.timeout);
 
+      // A token (and its tags) has no owning run, so it can't co-locate. Resolve the env mint kind so a
+      // minted-new env creates them on the run-ops DB (NEW) instead of defaulting to the draining LEGACY
+      // DB by their cuid id-shape.
+      const mintKind = await resolveRunIdMintKind({
+        organizationId: authentication.environment.organizationId,
+        id: authentication.environment.id,
+        orgFeatureFlags: authentication.environment.organization.featureFlags,
+      });
+      const residency = mintKind === "runOpsId" ? "NEW" : "LEGACY";
+
       //upsert tags
       let tags: { id: string; name: string }[] = [];
       const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags;
@@ -74,6 +85,7 @@ const { action } = createActionApiRoute(
             tag,
             environmentId: authentication.environment.id,
             projectId: authentication.environment.projectId,
+            residency,
           });
           if (tagRecord) {
             tags.push(tagRecord);
@@ -88,6 +100,7 @@ const { action } = createActionApiRoute(
         idempotencyKeyExpiresAt,
         timeout,
         tags: bodyTags,
+        standaloneResidency: residency,
       });
 
       const $responseHeaders = await responseHeaders(authentication.environment);

apps/webapp/app/routes/api.v2.batches.$batchId.ts

@@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
+    // A just-created batch may not yet have replicated to the read replica this client-less
+    // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through
+    // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes,
+    // e.g. api.v3.runs.$runId).
+    shouldRetryNotFound: true,
     findResource: (params, auth) => {
       return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, {
         include: { errors: true },

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.complete.ts

@@ -18,6 +18,7 @@ export const action = createActionWorkerApiRoute(
     body,
     params,
     runnerId,
+    environmentId,
   }): Promise<TypedResponse<WorkerApiRunAttemptCompleteResponseBody>> => {
     const { completion } = body;
     const { runFriendlyId, snapshotFriendlyId } = params;
@@ -27,6 +28,7 @@ export const action = createActionWorkerApiRoute(
       snapshotFriendlyId,
       completion,
       runnerId,
+      environmentId,
     });
 
     return json({ result: completeResult });

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.attempts.start.ts

@@ -18,6 +18,7 @@ export const action = createActionWorkerApiRoute(
     body,
     params,
     runnerId,
+    environmentId,
   }): Promise<TypedResponse<WorkerApiRunAttemptStartResponseBody>> => {
     const { runFriendlyId, snapshotFriendlyId } = params;
 
@@ -26,6 +27,7 @@ export const action = createActionWorkerApiRoute(
       snapshotFriendlyId,
       isWarmStart: body.isWarmStart,
       runnerId,
+      environmentId,
     });
 
     return json(runExecutionData);

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.$snapshotFriendlyId.continue.ts

@@ -17,6 +17,7 @@ export const loader = createLoaderWorkerApiRoute(
     authenticatedWorker,
     params,
     runnerId,
+    environmentId,
   }): Promise<TypedResponse<WorkerApiContinueRunExecutionRequestBody>> => {
     const { runFriendlyId, snapshotFriendlyId } = params;
 
@@ -27,10 +28,17 @@ export const loader = createLoaderWorkerApiRoute(
         runFriendlyId,
         snapshotFriendlyId,
         runnerId,
+        environmentId,
       });
 
       return json(continuationResult);
     } catch (error) {
+      // An authorization rejection is thrown as a Response; propagate it as-is rather than
+      // masking it as a generic 422.
+      if (error instanceof Response) {
+        throw error;
+      }
+
       logger.warn("Failed to continue run execution", {
         runFriendlyId,
         snapshotFriendlyId,

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.latest.ts

@@ -13,11 +13,13 @@ export const loader = createLoaderWorkerApiRoute(
   async ({
     authenticatedWorker,
     params,
+    environmentId,
   }): Promise<TypedResponse<WorkerApiRunLatestSnapshotResponseBody>> => {
     const { runFriendlyId } = params;
 
     const executionData = await authenticatedWorker.getLatestSnapshot({
       runFriendlyId,
+      environmentId,
     });
 
     if (!executionData) {

apps/webapp/app/routes/engine.v1.worker-actions.runs.$runFriendlyId.snapshots.since.$snapshotId.ts

@@ -14,12 +14,14 @@ export const loader = createLoaderWorkerApiRoute(
   async ({
     authenticatedWorker,
     params,
+    environmentId,
   }): Promise<TypedResponse<WorkerApiRunSnapshotsSinceResponseBody>> => {
     const { runFriendlyId, snapshotId } = params;
 
     const snapshots = await authenticatedWorker.getSnapshotsSince({
       runFriendlyId,
       snapshotId,
+      environmentId,
     });
 
     if (!snapshots) {

apps/webapp/app/routes/logout.tsx

@@ -1,6 +1,8 @@
-import { type ActionFunction, type LoaderFunction } from "@remix-run/node";
+import { redirect, type ActionFunction, type LoaderFunction } from "@remix-run/node";
+import { env } from "~/env.server";
 import { authenticator } from "~/services/auth.server";
 import { sanitizeRedirectPath } from "~/utils";
+import { isSameOriginNavigation } from "~/utils/sameOriginNavigation";
 import { SSO_SESSION_EXPIRED_REASON } from "~/utils/ssoSession";
 
 function logoutRedirectTo(request: Request): string {
@@ -18,5 +20,10 @@ export const action: ActionFunction = async ({ request }) => {
 };
 
 export const loader: LoaderFunction = async ({ request }) => {
+  // GET /logout is state-changing, so reject cross-site navigations.
+  if (!isSameOriginNavigation(request, env.LOGIN_ORIGIN)) {
+    throw redirect("/");
+  }
+
   return await authenticator.logout(request, { redirectTo: logoutRedirectTo(request) });
 };

apps/webapp/app/routes/realtime.v1.batches.$batchId.ts

@@ -2,7 +2,7 @@ import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { resolveRealtimeStreamClient } from "~/services/realtime/resolveRealtimeStreamClient.server";
 import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { runStore } from "~/v3/runStore.server";
+import { resolveBatchTaskRunForRealtime } from "~/v3/realtime/resolveBatchForRealtime.server";
 
 const ParamsSchema = z.object({
   batchId: z.string(),
@@ -13,9 +13,13 @@ export const loader = createLoaderApiRoute(
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
-    findResource: (params, auth) => {
-      return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id);
-    },
+    // A just-created batch may not yet have replicated to the read replica the client-less lookup uses.
+    // shouldRetryNotFound stamps a retryable 404 for the zodfetch GET; the realtime resolver ALSO
+    // re-reads the owning primary on a replica miss, so the Electric ShapeStream consumer (which ignores
+    // x-should-retry) doesn't strand a live batch on a permanent 404. Mirrors the run-get routes.
+    shouldRetryNotFound: true,
+    findResource: (params, auth) =>
+      resolveBatchTaskRunForRealtime(params.batchId, auth.environment.id),
     authorization: {
       action: "read",
       // See sibling note in api.v1.batches.$batchId.ts — `{type: "runs"}`

apps/webapp/app/routes/realtime.v1.runs.$runId.ts

@@ -15,22 +15,24 @@ export const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, authentication) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: authentication.environment.id,
-        },
-        {
-          include: {
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a run that already exists on the owning primary. A spurious 404
+      // here permanently fails the client's realtime subscription (the SSE client treats 404 as
+      // "stream gone" — nonRetryableStatuses). Re-read the primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/realtime.v1.runs.ts

@@ -2,6 +2,7 @@ import { z } from "zod";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { resolveRealtimeStreamClient } from "~/services/realtime/resolveRealtimeStreamClient.server";
 import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { UNSAFE_REALTIME_TAG_CHARS } from "~/v3/electricShape.server";
 
 const SearchParamsSchema = z.object({
   tags: z
@@ -9,6 +10,20 @@ const SearchParamsSchema = z.object({
     .optional()
     .transform((value) => {
       return value ? value.split(",") : undefined;
+    })
+    .superRefine((tags, ctx) => {
+      if (!tags) return;
+      for (const tag of tags) {
+        // Mirror the runtime sanitiser's reject list so the API returns 400
+        // instead of a 500. Single quotes are allowed — escaped downstream.
+        if (UNSAFE_REALTIME_TAG_CHARS.test(tag) || tag.length === 0) {
+          ctx.addIssue({
+            code: z.ZodIssueCode.custom,
+            message: `Invalid tag: ${JSON.stringify(tag)}`,
+          });
+          return;
+        }
+      }
     }),
   createdAt: z.string().optional(),
 });

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.ts

@@ -34,6 +34,16 @@ const { action } = createActionApiRoute(
     },
   },
   async ({ params, authentication }) => {
+    // `.out` is the agent→client channel. Only PRIVATE (secret key) auth —
+    // i.e. the agent run itself — may initialize it. Session-scoped JWTs carry
+    // `write:sessions:<key>` for `.in`; without this gate they could obtain
+    // credentials to forge assistant chunks on their own session's `.out`.
+    if (params.io === "out" && authentication.type !== "PRIVATE") {
+      return new Response("Initializing the out channel requires secret key authentication", {
+        status: 403,
+      });
+    }
+
     // Row-optional addressing. The agent calls PUT initialize as part
     // of `session.out.writer()`, by which time it has already created
     // the row at bind, so a missing row here is an unusual case

apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts

@@ -1,115 +1,44 @@
-import { type ActionFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { runStore } from "~/v3/runStore.server";
-import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 
 const ParamsSchema = z.object({
   runId: z.string(),
   streamId: z.string(),
 });
 
-// Plain action for backwards compatibility with older clients that don't send auth headers
-export async function action({ request, params }: ActionFunctionArgs) {
-  const parsedParams = ParamsSchema.safeParse(params);
-
-  if (!parsedParams.success) {
-    return new Response("Invalid parameters", { status: 400 });
-  }
-
-  const { runId, streamId } = parsedParams.data;
-
-  // Look up the run without environment scoping for backwards compatibility
-  const run = await runStore.findRun(
-    {
-      friendlyId: runId,
-    },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        streamBasinName: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    $replica
-  );
-
-  if (!run) {
-    return new Response("Run not found", { status: 404 });
-  }
-
-  const environment = await controlPlaneResolver.resolveAuthenticatedEnv(run.runtimeEnvironmentId);
-
-  if (!environment) {
-    return new Response("Run not found", { status: 404 });
-  }
-
-  // Extract client ID from header, default to "default" if not provided
-  const clientId = request.headers.get("X-Client-Id") || "default";
-  const streamVersion = request.headers.get("X-Stream-Version") || "v1";
-
-  if (!request.body) {
-    return new Response("No body provided", { status: 400 });
-  }
-
-  const resumeFromChunk = request.headers.get("X-Resume-From-Chunk");
-  let resumeFromChunkNumber: number | undefined = undefined;
-  if (resumeFromChunk) {
-    const parsed = parseInt(resumeFromChunk, 10);
-    if (isNaN(parsed) || parsed < 0) {
-      return new Response(`Invalid X-Resume-From-Chunk header value: ${resumeFromChunk}`, {
-        status: 400,
-      });
-    }
-    resumeFromChunkNumber = parsed;
-  }
-
-  const realtimeStream = getRealtimeStreamInstance(environment, streamVersion, {
-    run,
-  });
-
-  return realtimeStream.ingestData(
-    request.body,
-    run.friendlyId,
-    streamId,
-    clientId,
-    resumeFromChunkNumber
-  );
-}
-
 export const loader = createLoaderApiRoute(
   {
     params: ParamsSchema,
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, auth) => {
-      const run = await runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: auth.environment.id,
-        },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            taskIdentifier: true,
-            runTags: true,
-            realtimeStreamsVersion: true,
-            streamBasinName: true,
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: auth.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          taskIdentifier: true,
+          runTags: true,
+          realtimeStreamsVersion: true,
+          streamBasinName: true,
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
-      return run;
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the SSE subscription
+      // (the client treats 404 as "stream gone"). Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts

@@ -27,29 +27,31 @@ const { action } = createActionApiRoute(
     maxContentLength: MAX_APPEND_BODY_BYTES,
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
-      },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          parentTaskRun: {
-            select: {
-              friendlyId: true,
-            },
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        parentTaskRun: {
+          select: {
+            friendlyId: true,
           },
-          rootTaskRun: {
-            select: {
-              friendlyId: true,
-            },
+        },
+        rootTaskRun: {
+          select: {
+            friendlyId: true,
           },
         },
       },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the append. Re-read the
+    // owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts

@@ -19,32 +19,34 @@ const { action } = createActionApiRoute(
     params: ParamsSchema,
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
-      },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          streamBasinName: true,
-          parentTaskRun: {
-            select: {
-              friendlyId: true,
-              streamBasinName: true,
-            },
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        streamBasinName: true,
+        parentTaskRun: {
+          select: {
+            friendlyId: true,
+            streamBasinName: true,
           },
-          rootTaskRun: {
-            select: {
-              friendlyId: true,
-              streamBasinName: true,
-            },
+        },
+        rootTaskRun: {
+          select: {
+            friendlyId: true,
+            streamBasinName: true,
           },
         },
       },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the ingest client.
+    // Re-read the owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return new Response("Run not found", { status: 404 });
@@ -154,32 +156,33 @@ const loader = createLoaderApiRoute(
     allowJWT: false,
     corsStrategy: "none",
     findResource: async (params, authentication) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: authentication.environment.id,
-        },
-        {
-          select: {
-            id: true,
-            friendlyId: true,
-            streamBasinName: true,
-            parentTaskRun: {
-              select: {
-                friendlyId: true,
-                streamBasinName: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: authentication.environment.id,
+      };
+      const args = {
+        select: {
+          id: true,
+          friendlyId: true,
+          streamBasinName: true,
+          parentTaskRun: {
+            select: {
+              friendlyId: true,
+              streamBasinName: true,
             },
-            rootTaskRun: {
-              select: {
-                friendlyId: true,
-                streamBasinName: true,
-              },
+          },
+          rootTaskRun: {
+            select: {
+              friendlyId: true,
+              streamBasinName: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the HEAD probe.
+      // Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
   },
   async ({ request, params, resource: run, authentication }) => {

apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts

@@ -39,22 +39,24 @@ const { action } = createActionApiRoute(
     },
   },
   async ({ request, params, authentication }) => {
-    const run = await runStore.findRun(
-      {
-        friendlyId: params.runId,
-        runtimeEnvironmentId: authentication.environment.id,
+    const where = {
+      friendlyId: params.runId,
+      runtimeEnvironmentId: authentication.environment.id,
+    };
+    const args = {
+      select: {
+        id: true,
+        friendlyId: true,
+        completedAt: true,
+        realtimeStreamsVersion: true,
+        streamBasinName: true,
       },
-      {
-        select: {
-          id: true,
-          friendlyId: true,
-          completedAt: true,
-          realtimeStreamsVersion: true,
-          streamBasinName: true,
-        },
-      },
-      $replica
-    );
+    };
+    // Replica lag can null out a live run; a spurious 404 permanently fails the input send. Re-read
+    // the owning primary on a replica miss.
+    const run =
+      (await runStore.findRun(where, args, $replica)) ??
+      (await runStore.findRunOnPrimary(where, args));
 
     if (!run) {
       return json({ ok: false, error: "Run not found" }, { status: 404 });
@@ -133,22 +135,23 @@ const loader = createLoaderApiRoute(
     allowJWT: true,
     corsStrategy: "all",
     findResource: async (params, auth) => {
-      return runStore.findRun(
-        {
-          friendlyId: params.runId,
-          runtimeEnvironmentId: auth.environment.id,
-        },
-        {
-          include: {
-            batch: {
-              select: {
-                friendlyId: true,
-              },
+      const where = {
+        friendlyId: params.runId,
+        runtimeEnvironmentId: auth.environment.id,
+      };
+      const args = {
+        include: {
+          batch: {
+            select: {
+              friendlyId: true,
             },
           },
         },
-        $replica
-      );
+      };
+      // Replica lag can null out a live run; a spurious 404 permanently fails the SSE tail (the
+      // client treats 404 as "stream gone"). Re-read the owning primary on a replica miss.
+      const run = await runStore.findRun(where, args, $replica);
+      return run ?? runStore.findRunOnPrimary(where, args);
     },
     authorization: {
       action: "read",

apps/webapp/app/routes/resources.$projectId.deployments.$deploymentShortCode.cancel.ts

@@ -75,7 +75,7 @@ export const action = dashboardAction(
         prisma.workerDeployment.findUnique({
           select: {
             friendlyId: true,
-            projectId: true,
+            environmentId: true,
           },
           where: {
             projectId_shortCode: {
@@ -96,10 +96,7 @@ export const action = dashboardAction(
     const result = await verifyProjectMembership()
       .andThen(findDeploymentFriendlyId)
       .andThen((deployment) =>
-        deploymentService.cancelDeployment(
-          { projectId: deployment.projectId },
-          deployment.friendlyId
-        )
+        deploymentService.cancelDeployment({ id: deployment.environmentId }, deployment.friendlyId)
       );
 
     if (result.isErr()) {

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx

@@ -12,20 +12,19 @@ export const action: ActionFunction = async ({ request, params }) => {
   const { projectParam, organizationSlug, envParam, runParam } = v3RunParamsSchema.parse(params);
 
   try {
-    const taskRun = await runStore.findRun(
-      {
-        friendlyId: runParam,
-      },
-      {
-        select: {
-          id: true,
-          idempotencyKey: true,
-          taskIdentifier: true,
-          projectId: true,
-          runtimeEnvironmentId: true,
-        },
-      }
-    );
+    const resetSelect = {
+      id: true,
+      idempotencyKey: true,
+      taskIdentifier: true,
+      projectId: true,
+      runtimeEnvironmentId: true,
+    };
+    let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: resetSelect });
+    if (!taskRun) {
+      // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+      // before 404ing — this null gates the reset mutation below (mirrors cancel/replay).
+      taskRun = await runStore.findRunOnPrimary({ friendlyId: runParam }, { select: resetSelect });
+    }
 
     if (!taskRun) {
       return jsonWithErrorMessage({}, request, "Run not found");

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts

@@ -1,6 +1,6 @@
 import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import { runStore } from "~/v3/runStore.server";
 import { findProjectBySlug } from "~/models/project.server";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
@@ -8,7 +8,7 @@ import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
 import {
   canonicalSessionAddressingKey,
-  resolveSessionByIdOrExternalId,
+  resolveSessionWithWriterFallback,
 } from "~/services/realtime/sessions.server";
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
@@ -51,22 +51,27 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
   // Verify the run lives in this environment — keeps callers from
   // subscribing to arbitrary sessions via `/runs/$runParam/...`.
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-      runtimeEnvironmentId: environment.id,
-    },
-    {
-      select: { id: true, friendlyId: true },
-    },
-    $replica
-  );
+  const runWhere = {
+    friendlyId: runParam,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: { id: true, friendlyId: true },
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
+  // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });
   }
 
-  const session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionId);
+  // Replica lag can null out a just-created session; a spurious 404 breaks the dashboard Agent tab
+  // subscription (useRealtimeStream surfaces the error and does not auto-retry). Resolve replica-first
+  // with a writer fallback — the same helper the sibling `.in/append` route uses.
+  const session = await resolveSessionWithWriterFallback(environment.id, sessionId);
 
   if (!session) {
     return new Response("Session not found", { status: 404 });
@@ -76,10 +81,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // this environment is enough to subscribe to any session in the same
   // environment — defeats the point of scoping subscriptions through the
   // run route. SessionRun.runId is indexed (@unique), so this is cheap.
-  const linkedSessionRun = await $replica.sessionRun.findFirst({
-    where: { runId: run.id, sessionId: session.id },
-    select: { id: true },
-  });
+  // Replica lag can null out the just-created run↔session linkage row; a spurious 404 breaks the
+  // dashboard Agent tab subscription (client does not auto-retry). Re-read the primary on a miss.
+  const linkWhere = { runId: run.id, sessionId: session.id };
+  const linkedSessionRun =
+    (await $replica.sessionRun.findFirst({ where: linkWhere, select: { id: true } })) ??
+    (await prisma.sessionRun.findFirst({ where: linkWhere, select: { id: true } }));
 
   if (!linkedSessionRun) {
     return new Response("Session not found for run", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts

@@ -45,21 +45,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Environment not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runId,
-      runtimeEnvironmentId: environment.id,
+  const runWhere = {
+    friendlyId: runId,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
     },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-      },
-    },
-    $replica
-  );
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
+  // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts

@@ -47,21 +47,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     return new Response("Environment not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runId,
-      runtimeEnvironmentId: environment.id,
+  const runWhere = {
+    friendlyId: runId,
+    runtimeEnvironmentId: environment.id,
+  };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
     },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-      },
-    },
-    $replica
-  );
+  };
+  // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream
+  // subscription (useRealtimeStream surfaces the error, no auto-retry). Re-read the primary on a miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs, $replica)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx

@@ -60,18 +60,22 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     throw new Response("Not Found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    { friendlyId: runParam, projectId: project.id },
-    {
-      select: {
-        id: true,
-        friendlyId: true,
-        realtimeStreamsVersion: true,
-        streamBasinName: true,
-        runtimeEnvironmentId: true,
-      },
-    }
-  );
+  const runWhere = { friendlyId: runParam, projectId: project.id };
+  const runArgs = {
+    select: {
+      id: true,
+      friendlyId: true,
+      realtimeStreamsVersion: true,
+      streamBasinName: true,
+      runtimeEnvironmentId: true,
+    },
+  };
+  // Client-less findRun defaults to the read replica; replica lag can null out a live run and 404 a
+  // valid stream-viewer request (useRealtimeStream surfaces the error, no auto-retry). Re-read the
+  // owning primary on a replica miss.
+  const run =
+    (await runStore.findRun(runWhere, runArgs)) ??
+    (await runStore.findRunOnPrimary(runWhere, runArgs));
 
   if (!run) {
     throw new Response("Not Found", { status: 404 });

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx

@@ -81,7 +81,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
 
     const waitpointId = WaitpointId.toId(waitpointFriendlyId);
 
-    const waitpoint = await runStore.findWaitpoint({
+    let waitpoint = await runStore.findWaitpoint({
       select: {
         projectId: true,
         environmentId: true,
@@ -90,6 +90,14 @@ export const action = async ({ request, params }: ActionFunctionArgs) => {
         id: waitpointId,
       },
     });
+    if (!waitpoint) {
+      // Read-your-writes: a just-minted token may not have replicated. Re-read the owning primary
+      // before the auth guard / "No waitpoint found" (mirrors the token complete/callback routes).
+      waitpoint = await runStore.findWaitpointOnPrimary({
+        select: { projectId: true, environmentId: true },
+        where: { id: waitpointId },
+      });
+    }
 
     if (waitpoint?.projectId !== project.id) {
       return redirectWithErrorMessage(

apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts

@@ -80,21 +80,26 @@ export const action = dashboardAction(
       // The project-scope + membership auth is a control-plane concern resolved
       // separately below; joining project/organization here is a cross-DB join
       // that returns nothing once the run lives in the run-ops DB.
-      const taskRun = await runStore.findRun(
-        { friendlyId: runParam },
-        {
-          select: {
-            id: true,
-            engine: true,
-            status: true,
-            friendlyId: true,
-            taskEventStore: true,
-            createdAt: true,
-            completedAt: true,
-            projectId: true,
-          },
-        }
-      );
+      const cancelRunSelect = {
+        id: true,
+        engine: true,
+        status: true,
+        friendlyId: true,
+        taskEventStore: true,
+        createdAt: true,
+        completedAt: true,
+        projectId: true,
+      };
+      let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: cancelRunSelect });
+      if (!taskRun) {
+        // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+        // before treating it as absent (mirrors resolveRunOrganizationId's primary fallback above).
+        taskRun = await runStore.findRun(
+          { friendlyId: runParam },
+          { select: cancelRunSelect },
+          prisma
+        );
+      }
 
       // Project-scope + membership auth is control-plane only — keyed by the
       // run's projectId. A miss is treated as not-found (mirrors the old where).

apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts

@@ -323,7 +323,12 @@ export const action = dashboardAction(
     try {
       // Run-ops read keyed by friendlyId only; membership auth is re-checked on the
       // control plane below, keyed off the resolved run's projectId.
-      const pgRun = await runStore.findRun({ friendlyId: runParam });
+      let pgRun = await runStore.findRun({ friendlyId: runParam });
+      if (!pgRun) {
+        // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary
+        // before falling back to the mollifier buffer (mirrors resolveRunOrganizationId above).
+        pgRun = await runStore.findRun({ friendlyId: runParam }, prisma);
+      }
 
       // Mollifier read-fallback: if the original isn't in PG yet, synthesise a
       // TaskRun from the buffered snapshot. Needs project/org/env slugs for the

apps/webapp/app/routes/sync.traces.$traceId.ts

@@ -1,15 +1,33 @@
 import type { LoaderFunctionArgs } from "@remix-run/node";
+import { z } from "zod";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { getUserId } from "~/services/session.server";
 import { longPollingFetch } from "~/utils/longPollingFetch";
+import {
+  OtelTraceIdSchema,
+  RESERVED_ELECTRIC_SHAPE_PARAMS,
+  buildElectricTraceWhereClause,
+} from "~/v3/electricShape.server";
+
+const Params = z.object({
+  traceId: OtelTraceIdSchema,
+});
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   try {
     const userId = await getUserId(request);
 
-    logger.log(`/sync/traces/${params.traceId}`, { userId });
+    const parsedParams = Params.safeParse(params);
+    if (!parsedParams.success) {
+      // Treat a malformed traceId as not-found rather than 400 to avoid
+      // signalling the validator.
+      return new Response("Not found", { status: 404 });
+    }
+    const { traceId } = parsedParams.data;
+
+    logger.log(`/sync/traces/${traceId}`, { userId });
 
     if (!userId) {
       return new Response("No user found in cookie", { status: 401 });
@@ -20,7 +38,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
         organizationId: true,
       },
       where: {
-        traceId: params.traceId,
+        traceId,
       },
     });
 
@@ -41,11 +59,19 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
     const url = new URL(request.url);
     const originUrl = new URL(`${env.ELECTRIC_ORIGIN}/v1/shape/public."TaskEvent"`);
+    // Strip params we set ourselves so the caller can't override them.
     url.searchParams.forEach((value, key) => {
+      if (RESERVED_ELECTRIC_SHAPE_PARAMS.has(key)) return;
       originUrl.searchParams.set(key, value);
     });
 
-    originUrl.searchParams.set("where", `"traceId"='${params.traceId}'`);
+    originUrl.searchParams.set(
+      "where",
+      buildElectricTraceWhereClause({
+        traceId,
+        scope: { column: "organizationId", id: trace.organizationId },
+      })
+    );
 
     const finalUrl = originUrl.toString();
 

apps/webapp/app/routes/sync.traces.runs.$traceId.ts

@@ -5,17 +5,27 @@ import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { getUserId } from "~/services/session.server";
 import { longPollingFetch } from "~/utils/longPollingFetch";
-import { runStore } from "~/v3/runStore.server";
+import {
+  OtelTraceIdSchema,
+  RESERVED_ELECTRIC_SHAPE_PARAMS,
+  buildElectricTraceWhereClause,
+} from "~/v3/electricShape.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { runStore } from "~/v3/runStore.server";
 
 const Params = z.object({
-  traceId: z.string(),
+  traceId: OtelTraceIdSchema,
 });
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   try {
     const userId = await getUserId(request);
-    const { traceId } = Params.parse(params);
+
+    const parsedParams = Params.safeParse(params);
+    if (!parsedParams.success) {
+      return new Response("Not found", { status: 404 });
+    }
+    const { traceId } = parsedParams.data;
 
     logger.log(`/sync/runs/${traceId}`, { userId });
 
@@ -23,18 +33,28 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
       return new Response("No user found in cookie", { status: 401 });
     }
 
-    const run = await runStore.findRun(
+    let run = await runStore.findRun(
       {
         traceId,
       },
       {
         select: {
+          projectId: true,
           runtimeEnvironmentId: true,
         },
       },
       $replica
     );
 
+    if (!run) {
+      // Read-your-writes: a just-created run may not have replicated yet. Re-read the owning
+      // primary before 404ing so a live run's realtime trace feed isn't spuriously not-found.
+      run = await runStore.findRunOnPrimary(
+        { traceId },
+        { select: { projectId: true, runtimeEnvironmentId: true } }
+      );
+    }
+
     if (!run) {
       return new Response("No run found", { status: 404 });
     }
@@ -58,11 +78,22 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
 
     const url = new URL(request.url);
     const originUrl = new URL(`${env.ELECTRIC_ORIGIN}/v1/shape/public."TaskRun"`);
+    // Strip params we set ourselves so the caller can't override them.
     url.searchParams.forEach((value, key) => {
+      if (RESERVED_ELECTRIC_SHAPE_PARAMS.has(key)) return;
       originUrl.searchParams.set(key, value);
     });
 
-    originUrl.searchParams.set("where", `"traceId"='${traceId}'`);
+    originUrl.searchParams.set(
+      "where",
+      // Scope by non-null projectId, not the nullable organizationId (legacy
+      // rows would vanish). Tenant-safe: membership was verified against this
+      // project's org and a trace's runs all live in one project.
+      buildElectricTraceWhereClause({
+        traceId,
+        scope: { column: "projectId", id: run.projectId },
+      })
+    );
 
     const finalUrl = originUrl.toString();
 

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -1,5 +1,5 @@
-import { RunId } from "@trigger.dev/core/v3/isomorphic";
-import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database";
+import { ownerEngine, RunId } from "@trigger.dev/core/v3/isomorphic";
+import type { PrismaClientOrTransaction, TaskRun, Waitpoint } from "@trigger.dev/database";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
@@ -8,7 +8,8 @@ import type { RunEngine } from "~/v3/runEngine.server";
 import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus";
 import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
-import { claimOrAwait } from "~/v3/mollifier/idempotencyClaim.server";
+import { claimOrAwait, resetResolvedClaim } from "~/v3/mollifier/idempotencyClaim.server";
+import { computeClaimTtlSeconds } from "~/v3/mollifier/claimTtl";
 import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server";
 import { runStore } from "~/v3/runStore.server";
 import { runOpsLegacyPrisma, runOpsNewPrisma } from "~/db.server";
@@ -24,6 +25,13 @@ import type { TraceEventConcern, TriggerTaskRequest } from "../types";
 // handleTriggerRequest.
 const resolveOrgMollifierFlag = makeResolveMollifierFlag();
 
+// Cap on the claim-loser recreate re-acquisition loop (see
+// reacquireClearedGlobalWinner). Each pass reopens a stale resolved slot and
+// re-enters the claim; bounded so a pathological stream of expired/failed
+// winners can't spin forever. On exhaustion we fall open to the create with
+// PG's unique index as the backstop.
+const MAX_CLEARED_WINNER_REACQUIRES = 5;
+
 // Claim ownership context returned to the caller when the
 // IdempotencyKeyConcern won a pre-gate claim. Caller MUST publish the
 // winning runId on pipeline success (`publishClaim`) or release the
@@ -173,6 +181,14 @@ export class IdempotencyKeyConcern {
       }
     );
 
+    // `global`-scope (or scope-absent) keys under the split have no per-run salt, so the Redis claim is
+    // their only cross-DB dedup mutex. Computed here (not just in the claim block below) because the
+    // expired/failed clear-and-recreate path must serialise through it too.
+    const idempotencyKeyScope = request.body.options?.idempotencyKeyOptions?.scope;
+    const globalUnderSplit =
+      (idempotencyKeyScope === "global" || idempotencyKeyScope === undefined) &&
+      (await isSplitEnabled());
+
     const existingRun = idempotencyKey
       ? await runStore.findRun(
           {
@@ -209,107 +225,37 @@ export class IdempotencyKeyConcern {
     }
 
     if (existingRun) {
-      // The idempotency key has expired
-      if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
-        logger.debug("[TriggerTaskService][call] Idempotency key has expired", {
-          idempotencyKey: request.options?.idempotencyKey,
-          run: existingRun,
+      const handled = await this.handleExistingRun(request, parentStore, existingRun, {
+        idempotencyKey,
+        idempotencyKeyExpiresAt,
+        dedupClient,
+      });
+      // A LIVE cached hit (or andWait waitpoint wiring) is terminal.
+      if (handled.isCached) {
+        return handled;
+      }
+      // isCached === false → the existing run was EXPIRED/FAILED, so handleExistingRun cleared its key
+      // and we must recreate. For a global-scope key under split that recreate has to be claim-
+      // serialised too — otherwise two concurrent cross-residency recreates each create a run the
+      // per-DB unique index can't dedup (the same hole reacquireClearedGlobalWinner closes on the
+      // claim-loser path). Non-split / non-global: the plain unserialised recreate is safe.
+      if (globalUnderSplit) {
+        return await this.reacquireClearedGlobalWinner(request, parentStore, {
+          idempotencyKey,
+          idempotencyKeyExpiresAt,
+          dedupClient,
+          ttlSeconds: computeClaimTtlSeconds({
+            keyExpiresAt: idempotencyKeyExpiresAt,
+            now: Date.now(),
+            minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS,
+            maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
+          }),
+          clearedRunId: existingRun.friendlyId,
+          safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS,
+          pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
         });
-
-        // Update the existing run to remove the idempotency key
-        await runStore.clearIdempotencyKey(
-          { byId: { runId: existingRun.id, idempotencyKey } },
-          dedupClient
-        );
-
-        return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
       }
-
-      // If the existing run failed or was expired, we clear the key and do a new run
-      if (shouldIdempotencyKeyBeCleared(existingRun.status)) {
-        logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", {
-          idempotencyKey: request.options?.idempotencyKey,
-          runStatus: existingRun.status,
-          runId: existingRun.id,
-        });
-
-        // Update the existing run to remove the idempotency key
-        await runStore.clearIdempotencyKey(
-          { byId: { runId: existingRun.id, idempotencyKey } },
-          dedupClient
-        );
-
-        return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
-      }
-
-      // We have an idempotent run, so we return it
-      const parentRunId = request.body.options?.parentRunId;
-      const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
-
-      //We're using `andWait` so we need to block the parent run with a waitpoint
-      if (resumeParentOnCompletion && parentRunId) {
-        // `parentRunId` comes from the request body and isn't re-validated
-        // here, so confirm the parent run is in the caller's environment
-        // before wiring a waitpoint against it.
-        const parentRunInternalId = RunId.fromFriendlyId(parentRunId);
-        const parentRunInCallerEnv = await runStore.findRun(
-          {
-            id: parentRunInternalId,
-            runtimeEnvironmentId: request.environment.id,
-          },
-          { select: { id: true } },
-          this.prisma
-        );
-        if (!parentRunInCallerEnv) {
-          throw new ServiceValidationError("Parent run not found in the calling environment", 404);
-        }
-
-        // Get or create waitpoint lazily (existing run may not have one if it was standalone)
-        let associatedWaitpoint = existingRun.associatedWaitpoint;
-        if (!associatedWaitpoint) {
-          associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({
-            runId: existingRun.id,
-            projectId: request.environment.projectId,
-            environmentId: request.environment.id,
-          });
-        }
-
-        await this.traceEventConcern.traceIdempotentRun(
-          request,
-          parentStore,
-          {
-            existingRun,
-            idempotencyKey,
-            incomplete: associatedWaitpoint.status === "PENDING",
-            isError: associatedWaitpoint.outputIsError,
-          },
-          async (event) => {
-            const spanId =
-              request.options?.parentAsLinkType === "replay"
-                ? event.spanId
-                : event.traceparent?.spanId
-                  ? `${event.traceparent.spanId}:${event.spanId}`
-                  : event.spanId;
-
-            await this.engine.blockRunWithWaitpoint({
-              runId: parentRunInternalId,
-              waitpoints: associatedWaitpoint!.id,
-              spanIdToComplete: spanId,
-              batch: request.options?.batchId
-                ? {
-                    id: request.options.batchId,
-                    index: request.options.batchIndex ?? 0,
-                  }
-                : undefined,
-              projectId: request.environment.projectId,
-              organizationId: request.environment.organizationId,
-              tx: dedupClient,
-            });
-          }
-        );
-      }
-
-      return { isCached: true, run: existingRun };
+      return handled;
     }
 
     // Pre-gate claim — closes the PG+buffer race during gate transition.
@@ -325,28 +271,41 @@ export class IdempotencyKeyConcern {
     // claim's Redis SETNX keeps its RTT off the hot path for those requests
     // during staged rollout. The org-flag check is a pure in-memory read of
     // `Organization.featureFlags`, no DB query.
+    //
+    // Under the run-ops split the claim ALSO acts as the only cross-DB mutex a
+    // `global`-scope key has: that key is per (environment, task), so it can be
+    // triggered concurrently from two parents on DIFFERENT physical DBs where
+    // each probe misses and the per-DB unique index can't enforce uniqueness.
+    // For that case the claim is eligible regardless of the per-org flag AND of
+    // resumeParentOnCompletion (the loser wires its parent waitpoint against the
+    // winner in the resolved branch below). An absent scope (pre-hashed key /
+    // older SDK) is treated conservatively as possibly-global — harmless for a
+    // real run/attempt key, whose hash already embeds the parent id so two
+    // parents mint DISTINCT keys that never share a claim slot.
+    // (idempotencyKeyScope / globalUnderSplit are computed above — they also gate the expired/failed
+    // recreate serialisation.)
     const claimEligible =
-      !request.body.options?.resumeParentOnCompletion &&
       !request.body.options?.debounce &&
       !request.options?.oneTimeUseToken &&
-      (await resolveOrgMollifierFlag({
-        envId: request.environment.id,
-        orgId: request.environment.organizationId,
-        taskId: request.taskId,
-        orgFeatureFlags:
-          (request.environment.organization?.featureFlags as
-            | Record<string, unknown>
-            | null
-            | undefined) ?? null,
-      }));
+      (globalUnderSplit ||
+        (!request.body.options?.resumeParentOnCompletion &&
+          (await resolveOrgMollifierFlag({
+            envId: request.environment.id,
+            orgId: request.environment.organizationId,
+            taskId: request.taskId,
+            orgFeatureFlags:
+              (request.environment.organization?.featureFlags as
+                | Record<string, unknown>
+                | null
+                | undefined) ?? null,
+          }))));
     if (claimEligible) {
-      const ttlSeconds = Math.max(
-        1,
-        Math.min(
-          env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
-          Math.ceil((idempotencyKeyExpiresAt.getTime() - Date.now()) / 1000)
-        )
-      );
+      const ttlSeconds = computeClaimTtlSeconds({
+        keyExpiresAt: idempotencyKeyExpiresAt,
+        now: Date.now(),
+        minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS,
+        maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
+      });
       const outcome = await claimOrAwait({
         envId: request.environment.id,
         taskIdentifier: request.taskId,
@@ -356,6 +315,44 @@ export class IdempotencyKeyConcern {
         pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
       });
       if (outcome.kind === "resolved") {
+        // Global-under-split loser: the winner lives on ITS parent's DB, which
+        // may differ from this loser's parent DB. Resolve the winner by id
+        // across both DBs (classify the winner friendlyId → NEW/LEGACY client,
+        // then let the router route+fall-back by id-shape) and feed it through
+        // the same existing-run handling the PG-hit path uses — so expiry-clear,
+        // status-clear, and the resumeParentOnCompletion waitpoint wiring all
+        // apply to the loser exactly as they would to a plain cached hit.
+        if (globalUnderSplit) {
+          const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id);
+          if (winner) {
+            const resolved = await this.handleExistingRun(request, parentStore, winner, {
+              idempotencyKey,
+              idempotencyKeyExpiresAt,
+              dedupClient,
+            });
+            // A LIVE winner (cached hit, or andWait waitpoint wired) is terminal.
+            if (resolved.isCached) {
+              return resolved;
+            }
+            // CRITICAL (CodeRabbit): the resolved winner was EXPIRED or FAILED, so
+            // handleExistingRun cleared its key and would have us CREATE a new run.
+            // The initial create is serialised by the claim, but this clear-and-
+            // recreate is not — and under the split the per-DB unique index can't
+            // dedup a cross-residency recreate, so two concurrent losers clearing
+            // the SAME winner would each create → duplicate run. Re-serialise the
+            // recreate through the claim so exactly one caller recreates (and
+            // publishes) while the rest resolve to the fresh run.
+            return await this.reacquireClearedGlobalWinner(request, parentStore, {
+              idempotencyKey,
+              idempotencyKeyExpiresAt,
+              dedupClient,
+              ttlSeconds,
+              clearedRunId: outcome.runId,
+              safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS,
+              pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS,
+            });
+          }
+        }
         // Another concurrent trigger committed first. Re-resolve via the
         // existing checks: writer-side PG findFirst first (defeats
         // replica lag), then buffer fallback for the buffered case.
@@ -421,4 +418,238 @@ export class IdempotencyKeyConcern {
 
     return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
   }
+
+  // Resolve an already-existing idempotent run: honour key expiry / status
+  // clearing, and for `andWait` (resumeParentOnCompletion) block the calling
+  // parent on the run's waitpoint. Extracted so both the PG-hit path and the
+  // cross-DB claim-loser path resolve an existing run identically.
+  private async handleExistingRun(
+    request: TriggerTaskRequest,
+    parentStore: string | undefined,
+    existingRun: TaskRun & { associatedWaitpoint?: Waitpoint | null },
+    ctx: {
+      idempotencyKey: string;
+      idempotencyKeyExpiresAt: Date;
+      dedupClient: PrismaClientOrTransaction;
+    }
+  ): Promise<IdempotencyKeyConcernResult> {
+    const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient } = ctx;
+
+    // The idempotency key has expired
+    if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
+      logger.debug("[TriggerTaskService][call] Idempotency key has expired", {
+        idempotencyKey: request.options?.idempotencyKey,
+        run: existingRun,
+      });
+
+      // Update the existing run to remove the idempotency key
+      await runStore.clearIdempotencyKey(
+        { byId: { runId: existingRun.id, idempotencyKey } },
+        dedupClient
+      );
+
+      return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+    }
+
+    // If the existing run failed or was expired, we clear the key and do a new run
+    if (shouldIdempotencyKeyBeCleared(existingRun.status)) {
+      logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", {
+        idempotencyKey: request.options?.idempotencyKey,
+        runStatus: existingRun.status,
+        runId: existingRun.id,
+      });
+
+      // Update the existing run to remove the idempotency key
+      await runStore.clearIdempotencyKey(
+        { byId: { runId: existingRun.id, idempotencyKey } },
+        dedupClient
+      );
+
+      return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+    }
+
+    // We have an idempotent run, so we return it
+    const parentRunId = request.body.options?.parentRunId;
+    const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
+
+    //We're using `andWait` so we need to block the parent run with a waitpoint
+    if (resumeParentOnCompletion && parentRunId) {
+      // `parentRunId` comes from the request body and isn't re-validated
+      // here, so confirm the parent run is in the caller's environment
+      // before wiring a waitpoint against it.
+      const parentRunInternalId = RunId.fromFriendlyId(parentRunId);
+      const parentRunInCallerEnv = await runStore.findRun(
+        {
+          id: parentRunInternalId,
+          runtimeEnvironmentId: request.environment.id,
+        },
+        { select: { id: true } },
+        this.prisma
+      );
+      if (!parentRunInCallerEnv) {
+        throw new ServiceValidationError("Parent run not found in the calling environment", 404);
+      }
+
+      // Get or create waitpoint lazily (existing run may not have one if it was standalone)
+      let associatedWaitpoint = existingRun.associatedWaitpoint;
+      if (!associatedWaitpoint) {
+        associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({
+          runId: existingRun.id,
+          projectId: request.environment.projectId,
+          environmentId: request.environment.id,
+        });
+      }
+
+      await this.traceEventConcern.traceIdempotentRun(
+        request,
+        parentStore,
+        {
+          existingRun,
+          idempotencyKey,
+          incomplete: associatedWaitpoint.status === "PENDING",
+          isError: associatedWaitpoint.outputIsError,
+        },
+        async (event) => {
+          const spanId =
+            request.options?.parentAsLinkType === "replay"
+              ? event.spanId
+              : event.traceparent?.spanId
+                ? `${event.traceparent.spanId}:${event.spanId}`
+                : event.spanId;
+
+          await this.engine.blockRunWithWaitpoint({
+            runId: parentRunInternalId,
+            waitpoints: associatedWaitpoint!.id,
+            spanIdToComplete: spanId,
+            batch: request.options?.batchId
+              ? {
+                  id: request.options.batchId,
+                  index: request.options.batchIndex ?? 0,
+                }
+              : undefined,
+            projectId: request.environment.projectId,
+            organizationId: request.environment.organizationId,
+            tx: dedupClient,
+          });
+        }
+      );
+    }
+
+    return { isCached: true, run: existingRun };
+  }
+
+  // Re-serialise a cross-DB recreate through the claim after a claim-loser's
+  // resolved winner turned out to be EXPIRED / FAILED (its key was cleared by
+  // handleExistingRun). Without this, concurrent losers clearing the same
+  // winner each create a new run on their own DB — the per-DB unique index
+  // can't dedup a cross-residency pair, so the initial-create's serialisation
+  // is lost on the recreate. Each pass: compare-and-delete the stale resolved
+  // slot (keyed on the cleared runId — never an unconditional DEL, so a
+  // reacquirer that already re-published a NEW winner is not wiped), then
+  // re-enter claimOrAwait. Exactly one caller wins the re-claim and recreates
+  // (returning its claim to publish); the rest resolve to the fresh run. If a
+  // re-claim resolves to ANOTHER cleared winner we advance and loop, bounded
+  // by MAX_CLEARED_WINNER_REACQUIRES; on exhaustion (or an unfindable
+  // resolution) we fail CLOSED with a retryable 503 so the SDK retry re-serialises,
+  // rather than fall open to an unserialised cross-DB create.
+  private async reacquireClearedGlobalWinner(
+    request: TriggerTaskRequest,
+    parentStore: string | undefined,
+    ctx: {
+      idempotencyKey: string;
+      idempotencyKeyExpiresAt: Date;
+      dedupClient: PrismaClientOrTransaction;
+      ttlSeconds: number;
+      clearedRunId: string;
+      safetyNetMs: number;
+      pollStepMs: number;
+    }
+  ): Promise<IdempotencyKeyConcernResult> {
+    const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient, ttlSeconds } = ctx;
+    const claimInput = {
+      envId: request.environment.id,
+      taskIdentifier: request.taskId,
+      idempotencyKey,
+    };
+    let staleRunId = ctx.clearedRunId;
+    for (let attempt = 0; attempt < MAX_CLEARED_WINNER_REACQUIRES; attempt++) {
+      await resetResolvedClaim({ ...claimInput, runId: staleRunId });
+
+      const outcome = await claimOrAwait({
+        ...claimInput,
+        ttlSeconds,
+        safetyNetMs: ctx.safetyNetMs,
+        pollStepMs: ctx.pollStepMs,
+      });
+
+      if (outcome.kind === "timed_out") {
+        throw new ServiceValidationError("Idempotency claim resolution timed out", 503);
+      }
+      if (outcome.kind === "claimed") {
+        // We own the recreate. Caller MUST publish the new runId / release on error.
+        return {
+          isCached: false,
+          idempotencyKey,
+          idempotencyKeyExpiresAt,
+          claim: { ...claimInput, token: outcome.token },
+        };
+      }
+      // resolved: another caller won the recreate. Honour it like any cached
+      // hit (incl. andWait wiring). If it too was cleared, advance and loop.
+      const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id);
+      if (!winner) {
+        logger.warn("idempotency reacquire resolved but runId not findable", {
+          envId: request.environment.id,
+          taskIdentifier: request.taskId,
+          claimedRunId: outcome.runId,
+        });
+        break;
+      }
+      const resolved = await this.handleExistingRun(request, parentStore, winner, {
+        idempotencyKey,
+        idempotencyKeyExpiresAt,
+        dedupClient,
+      });
+      if (resolved.isCached) {
+        return resolved;
+      }
+      staleRunId = outcome.runId;
+    }
+    // Exhausted the bounded reacquires (or the winner was unfindable). Rather than fall through to an
+    // UNSERIALISED create — which under global-scope-split can dual-create across DBs (the per-DB unique
+    // index can't dedup cross-residency) — fail closed with a retryable 503 so the SDK retry re-serialises
+    // through a fresh claim (mirrors the timed_out branch above).
+    throw new ServiceValidationError(
+      "Idempotency claim could not be re-serialised after repeated cleared winners",
+      503
+    );
+  }
+
+  // Resolve a claim winner (a run friendlyId) across both split DBs. Classify
+  // the id-shape to pick the writer client for read-your-writes, then read by
+  // id — the routing store routes to the owning store and falls back to the
+  // other, so a winner on either DB is found. Returns null when the id can't be
+  // classified or the row genuinely isn't there (caller falls through).
+  private async resolveWinnerAcrossDbs(
+    winnerFriendlyId: string,
+    environmentId: string
+  ): Promise<(TaskRun & { associatedWaitpoint?: Waitpoint | null }) | null> {
+    let internalId: string;
+    try {
+      internalId = RunId.fromFriendlyId(winnerFriendlyId);
+    } catch {
+      return null;
+    }
+    let client: PrismaClientOrTransaction;
+    try {
+      client = ownerEngine(internalId) === "NEW" ? runOpsNewPrisma : runOpsLegacyPrisma;
+    } catch {
+      client = this.prisma;
+    }
+    return runStore.findRun(
+      { id: internalId, runtimeEnvironmentId: environmentId },
+      { include: { associatedWaitpoint: true } },
+      client
+    );
+  }
 }

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -752,7 +752,7 @@ export class RunEngineTriggerTaskService {
       // Pipeline returned successfully — publish the claim if we held
       // one. Waiters polling for our key resolve to this runId.
       if (idempotencyClaim && result?.run?.friendlyId) {
-        await publishMollifierClaim({
+        const published = await publishMollifierClaim({
           envId: idempotencyClaim.envId,
           taskIdentifier: idempotencyClaim.taskIdentifier,
           idempotencyKey: idempotencyClaim.idempotencyKey,
@@ -760,6 +760,17 @@ export class RunEngineTriggerTaskService {
           runId: result.run.friendlyId,
           ttlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
         });
+        if (!published) {
+          // Our claim expired mid-pipeline and another claimant took it, so this publish no-op'd: a
+          // different run is now canonical for this key while we return ours (a cross-DB dup under the
+          // split). Rare now the claim TTL is floored (C1); surfaced for monitoring pending auto-
+          // convergence (re-resolve the current winner + cancel this orphan).
+          logger.warn("mollifier claim publish no-op'd; winner lost the claim mid-pipeline", {
+            envId: idempotencyClaim.envId,
+            taskIdentifier: idempotencyClaim.taskIdentifier,
+            runId: result.run.friendlyId,
+          });
+        }
       }
       return result;
     } catch (err) {

apps/webapp/app/services/apiRateLimit.server.ts

@@ -48,6 +48,9 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
   // Allow /api/v1/tasks/:id/callback/:secret
   pathWhiteList: [
     "/api/internal/stripe_webhooks",
+    // Keep allowlisted: these CLI endpoints are intentionally unauthenticated,
+    // so this Authorization-header-keyed limiter would 401 them. They are
+    // throttled separately by authCodeRateLimiter.server.ts.
     "/api/v1/authorization-code",
     "/api/v1/token",
     "/api/v1/usage/ingest",

apps/webapp/app/services/authCodeRateLimiter.server.ts

@@ -0,0 +1,85 @@
+import { Ratelimit } from "@upstash/ratelimit";
+import { createHash } from "node:crypto";
+import { env } from "~/env.server";
+import { createRedisRateLimitClient, RateLimiter } from "~/services/rateLimiter.server";
+import { singleton } from "~/utils/singleton";
+
+/**
+ * Rate limiting for the unauthenticated CLI auth-code endpoints
+ * (`/api/v1/authorization-code` mint + `/api/v1/token` poll). The global
+ * limiter keys on the Authorization header, which these endpoints don't carry,
+ * so it can't throttle them — this module does.
+ */
+export class AuthorizationCodeRateLimitError extends Error {
+  public readonly retryAfter: number;
+
+  constructor(retryAfter: number) {
+    super("Authorization code rate limit exceeded.");
+    this.name = "AuthorizationCodeRateLimitError";
+    this.retryAfter = retryAfter;
+  }
+}
+
+function getRedisClient() {
+  return createRedisRateLimitClient({
+    port: env.RATE_LIMIT_REDIS_PORT,
+    host: env.RATE_LIMIT_REDIS_HOST,
+    username: env.RATE_LIMIT_REDIS_USERNAME,
+    password: env.RATE_LIMIT_REDIS_PASSWORD,
+    tlsDisabled: env.RATE_LIMIT_REDIS_TLS_DISABLED === "true",
+    clusterMode: env.RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED === "1",
+  });
+}
+
+// Minting is unauthenticated and a real login mints one code. Cap per IP, with
+// headroom for many users behind a shared NAT.
+const authorizationCodeMintIpRateLimiter = singleton(
+  "authorizationCodeMintIpRateLimiter",
+  () =>
+    new RateLimiter({
+      redisClient: getRedisClient(),
+      keyPrefix: "auth:authcode:mint:ip",
+      limiter: Ratelimit.slidingWindow(30, "1 m"), // 30 code mints / min / IP
+      logSuccess: false,
+      logFailure: true,
+    })
+);
+
+// Keyed by the code, not the IP: the CLI polls this endpoint ~1/s per login, so
+// IP-keying would break logins behind a shared NAT. The ~60/min cadence stays
+// under the cap. The code is hashed first so it never lands in a Redis key or log.
+const authorizationCodeTokenPollRateLimiter = singleton(
+  "authorizationCodeTokenPollRateLimiter",
+  () =>
+    new RateLimiter({
+      redisClient: getRedisClient(),
+      keyPrefix: "auth:authcode:token:code",
+      limiter: Ratelimit.slidingWindow(100, "1 m"), // 100 polls / min / code (CLI polls ~60/min)
+      logSuccess: false,
+      logFailure: false,
+    })
+);
+
+function hashCode(code: string): string {
+  return createHash("sha256").update(code).digest("hex").slice(0, 32);
+}
+
+export async function checkAuthorizationCodeMintRateLimit(ip: string): Promise<void> {
+  const result = await authorizationCodeMintIpRateLimiter.limit(ip);
+
+  if (!result.success) {
+    const retryAfter = new Date(result.reset).getTime() - Date.now();
+    throw new AuthorizationCodeRateLimitError(retryAfter);
+  }
+}
+
+export async function checkAuthorizationCodeTokenPollRateLimit(
+  authorizationCode: string
+): Promise<void> {
+  const result = await authorizationCodeTokenPollRateLimiter.limit(hashCode(authorizationCode));
+
+  if (!result.success) {
+    const retryAfter = new Date(result.reset).getTime() - Date.now();
+    throw new AuthorizationCodeRateLimitError(retryAfter);
+  }
+}

apps/webapp/app/services/dashboardAgent.server.ts

@@ -212,7 +212,9 @@ export async function resolveRunCommit(
   environmentId: string,
   runFriendlyId: string
 ): Promise<{ sha: string; version: string; dirty: boolean } | null> {
-  const run = await runStore.findRun(
+  // Read-your-writes: a just-locked run's lockedToVersionId may not have replicated. Read the owning
+  // primary so a live, pinned run resolves its commit instead of silently falling back to branch head.
+  const run = await runStore.findRunOnPrimary(
     { friendlyId: runFriendlyId, runtimeEnvironmentId: environmentId },
     { select: { lockedToVersionId: true } }
   );

apps/webapp/app/services/gitHub.server.ts

@@ -58,26 +58,32 @@ export async function linkGitHubAppInstallation(
 }
 
 /**
- * Links a GitHub App installation to a Trigger organization
+ * Updates a GitHub App installation owned by the given Trigger organization
  */
-export async function updateGitHubAppInstallation(installationId: number): Promise<void> {
+export async function updateGitHubAppInstallation(
+  installationId: number,
+  organizationId: string
+): Promise<void> {
   if (!githubApp) {
     throw new Error("GitHub App is not enabled");
   }
 
+  // Scope the lookup to the caller's organization so a cross-tenant
+  // installation_id cannot update another org's record. Resolve ownership
+  // before calling GitHub to avoid burning the victim's API rate limit.
+  const existingInstallation = await prisma.githubAppInstallation.findFirst({
+    where: { appInstallationId: installationId, organizationId },
+  });
+
+  if (!existingInstallation) {
+    throw new Error("GitHub App installation not found");
+  }
+
   const octokit = await githubApp.getInstallationOctokit(installationId);
   const { data: installation } = await octokit.rest.apps.getInstallation({
     installation_id: installationId,
   });
 
-  const existingInstallation = await prisma.githubAppInstallation.findFirst({
-    where: { appInstallationId: installationId },
-  });
-
-  if (!existingInstallation) {
-    throw new Error("GitHub App installation not found");
-  }
-
   const repositorySelection = installation.repository_selection === "all" ? "ALL" : "SELECTED";
 
   // repos are updated asynchronously via webhook events

apps/webapp/app/services/otlpRateLimit.server.ts

@@ -0,0 +1,91 @@
+import { Ratelimit } from "@upstash/ratelimit";
+import type { NextFunction, Request, Response } from "express";
+import { env } from "~/env.server";
+import { createRedisRateLimitClient, RateLimiter } from "~/services/rateLimiter.server";
+import { extractClientIp } from "~/utils/extractClientIp.server";
+import { singleton } from "~/utils/singleton";
+import { logger } from "./logger.server";
+
+const OTLP_PATH = /^\/otel\//i;
+
+function getOtlpIpRateLimiter() {
+  return singleton(
+    "otlpIpRateLimiter",
+    () =>
+      new RateLimiter({
+        redisClient: createRedisRateLimitClient({
+          port: env.RATE_LIMIT_REDIS_PORT,
+          host: env.RATE_LIMIT_REDIS_HOST,
+          username: env.RATE_LIMIT_REDIS_USERNAME,
+          password: env.RATE_LIMIT_REDIS_PASSWORD,
+          tlsDisabled: env.RATE_LIMIT_REDIS_TLS_DISABLED === "true",
+          clusterMode: env.RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED === "1",
+        }),
+        keyPrefix: "otlp:ip",
+        limiter: Ratelimit.slidingWindow(
+          env.OTLP_RATE_LIMIT_MAX,
+          env.OTLP_RATE_LIMIT_WINDOW as Parameters<typeof Ratelimit.slidingWindow>[1]
+        ),
+        logSuccess: false,
+        logFailure: true,
+      })
+  );
+}
+
+/**
+ * Per-IP rate limiter for the OTLP ingestion endpoints (`/otel/*`).
+ *
+ * These endpoints are currently unauthenticated (see SEC-98), so the source IP
+ * is the only identity available to key on. This bounds unauthenticated
+ * request rates and is NOT a substitute for authenticating the
+ * endpoints. It fails open (allows the request) whenever the source cannot be
+ * identified or the limiter backend errors, so a limiter outage never drops
+ * legitimate telemetry.
+ *
+ * Opt-in (disabled unless `OTLP_RATE_LIMIT_ENABLED=1`). Because it keys on the
+ * source IP, two preconditions must hold before enabling it, or it can drop
+ * legitimate telemetry:
+ *
+ * 1. Each client must present a distinct IP. Where many clients share one
+ *    egress IP (e.g. behind NAT or a shared proxy) their traffic collapses
+ *    into a single bucket and can be throttled together. Size
+ *    `OTLP_RATE_LIMIT_MAX` for the aggregate volume of a shared source, not a
+ *    single client.
+ * 2. The IP must be trustworthy. `extractClientIp` takes the last
+ *    `X-Forwarded-For` hop, which is only spoof-resistant behind a proxy that
+ *    appends the real client IP. Without such a proxy the value is
+ *    client-controlled and the per-IP bound is bypassable.
+ */
+export async function otlpRateLimiter(req: Request, res: Response, next: NextFunction) {
+  if (env.OTLP_RATE_LIMIT_ENABLED !== "1") {
+    return next();
+  }
+
+  if (req.method.toUpperCase() === "OPTIONS" || !OTLP_PATH.test(req.path)) {
+    return next();
+  }
+
+  const xff = req.headers["x-forwarded-for"];
+  const ip = extractClientIp(Array.isArray(xff) ? xff.join(",") : (xff ?? null)) ?? req.ip;
+
+  if (!ip) {
+    // Fail open: without a source we cannot fairly rate limit.
+    return next();
+  }
+
+  try {
+    const { success, reset } = await getOtlpIpRateLimiter().limit(ip);
+
+    if (!success) {
+      const retryAfterSeconds = Math.max(1, Math.ceil((reset - Date.now()) / 1000));
+      res.setHeader("Retry-After", retryAfterSeconds.toString());
+      res.status(429).send("Too Many Requests");
+      return;
+    }
+  } catch (error) {
+    // Fail open: a rate-limiter backend outage must not drop telemetry.
+    logger.warn("otlpRateLimiter: limiter error, allowing request", { error });
+  }
+
+  return next();
+}

apps/webapp/app/services/personalAccessToken.server.ts

@@ -18,6 +18,10 @@ const tokenGenerator = customAlphabet("123456789abcdefghijkmnopqrstuvwxyz", toke
 // staleness is fine.
 export const PAT_LAST_ACCESSED_THROTTLE_MS = 5 * 60 * 1000;
 
+// How long an unconsumed CLI authorization code stays valid. Shared constant so
+// the mint and read paths can't drift.
+export const AUTHORIZATION_CODE_TTL_MS = 10 * 60 * 1000;
+
 type CreatePersonalAccessTokenOptions = {
   name: string;
   userId: string;
@@ -60,16 +64,16 @@ export type ObfuscatedPersonalAccessToken = Awaited<
 
 /** Gets a PersonalAccessToken from an Auth Code, this only works within 10 mins of the auth code being created */
 export async function getPersonalAccessTokenFromAuthorizationCode(authorizationCode: string) {
-  //only allow authorization codes that were created less than 10 mins ago
-  const tenMinutesAgo = new Date(Date.now() - 10 * 60 * 1000);
-  const code = await prisma.authorizationCode.findUnique({
+  // Only allow authorization codes created within the short consent window.
+  const validAfter = new Date(Date.now() - AUTHORIZATION_CODE_TTL_MS);
+  const code = await prisma.authorizationCode.findFirst({
     select: {
       personalAccessToken: true,
     },
     where: {
       code: authorizationCode,
       createdAt: {
-        gte: tenMinutesAgo,
+        gte: validAfter,
       },
     },
   });
@@ -280,6 +284,27 @@ export function isPersonalAccessToken(token: string) {
   return token.startsWith(tokenPrefix);
 }
 
+/**
+ * Read-only check that an authorization code is still mintable: it exists, is
+ * unconsumed (`personalAccessTokenId: null`), and within the TTL. Lets the
+ * consent-screen loader show Authorize vs expired/invalid without minting a PAT.
+ */
+export async function isAuthorizationCodeMintable(
+  authorizationCode: string,
+  prismaClient = prisma
+): Promise<boolean> {
+  const validAfter = new Date(Date.now() - AUTHORIZATION_CODE_TTL_MS);
+  const code = await prismaClient.authorizationCode.findFirst({
+    where: {
+      code: authorizationCode,
+      personalAccessTokenId: null,
+      createdAt: { gte: validAfter },
+    },
+    select: { id: true },
+  });
+  return code !== null;
+}
+
 export function createAuthorizationCode() {
   return prisma.authorizationCode.create({
     data: {
@@ -293,14 +318,14 @@ export async function createPersonalAccessTokenFromAuthorizationCode(
   authorizationCode: string,
   userId: string
 ) {
-  //only allow authorization codes that were created less than 10 mins ago
-  const tenMinutesAgo = new Date(Date.now() - 10 * 60 * 1000);
-  const code = await prisma.authorizationCode.findUnique({
+  // Only allow authorization codes created within the short consent window.
+  const validAfter = new Date(Date.now() - AUTHORIZATION_CODE_TTL_MS);
+  const code = await prisma.authorizationCode.findFirst({
     where: {
       code: authorizationCode,
       personalAccessTokenId: null,
       createdAt: {
-        gte: tenMinutesAgo,
+        gte: validAfter,
       },
     },
   });

apps/webapp/app/services/realtime/sessionRunManager.server.ts

@@ -478,8 +478,10 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise<Swap
 async function getRunStatusAndFriendlyId(
   runId: string
 ): Promise<{ status: TaskRunStatus; friendlyId: string } | null> {
-  // Use the read replica — this is a hot-path probe and stale-by-ms is
-  // fine. The append handler re-checks if it ends up reusing the runId.
+  // Use the read replica — hot-path probe, stale-by-ms is fine. The dangerous
+  // stale shape (looks-vanished → double-trigger) is closed by the writer re-probe
+  // in ensureRunForSession; a stale-non-final reuse self-heals via the durable S2
+  // stream + next-append re-probe (there is no append-handler re-check).
   // `friendlyId` is fetched alongside `status` so the dead-run-detection
   // branch in `ensureRunForSession` can forward the public-form id as
   // `payload.previousRunId` without a second read. `Session.currentRunId`

apps/webapp/app/services/realtimeClient.server.ts

@@ -15,6 +15,7 @@ import { RedisCacheStore } from "./unkey/redisCacheStore.server";
 import { env } from "~/env.server";
 import type { API_VERSIONS } from "~/api/versions";
 import { CURRENT_API_VERSION } from "~/api/versions";
+import { sanitizeRealtimeTagsForSql } from "~/v3/electricShape.server";
 
 export interface CachedLimitProvider {
   getCachedLimit: (organizationId: string, defaultValue: number) => Promise<number | undefined>;
@@ -171,7 +172,10 @@ export class RealtimeClient {
     const whereClauses: string[] = [`"runtimeEnvironmentId"='${environment.id}'`];
 
     if (params.tags) {
-      whereClauses.push(`"runTags" @> ARRAY[${params.tags.map((t) => `'${t}'`).join(",")}]`);
+      // Reject unsafe chars and escape single quotes so tag values can't
+      // break out of the Electric SQL string literal.
+      const safeTags = sanitizeRealtimeTagsForSql(params.tags);
+      whereClauses.push(`"runTags" @> ARRAY[${safeTags.map((t) => `'${t}'`).join(",")}]`);
     }
 
     const createdAtFilter = await this.#calculateCreatedAtFilter(url, params.createdAt);

apps/webapp/app/services/routeBuilders/apiBuilder.server.ts

@@ -1537,6 +1537,7 @@ type WorkerLoaderHandlerFunction<
     ? z.infer<THeadersSchema>
     : undefined;
   runnerId?: string;
+  environmentId?: string;
 }) => Promise<Response>;
 
 export function createLoaderWorkerApiRoute<
@@ -1601,6 +1602,8 @@ export function createLoaderWorkerApiRoute<
       }
 
       const runnerId = request.headers.get(WORKER_HEADERS.RUNNER_ID) ?? undefined;
+      // `|| undefined` so a blank header can't become a zero-match snapshot filter (→ false reject).
+      const environmentId = request.headers.get(WORKER_HEADERS.ENVIRONMENT_ID) || undefined;
 
       const result = await handler({
         params: parsedParams,
@@ -1609,6 +1612,7 @@ export function createLoaderWorkerApiRoute<
         request,
         headers: parsedHeaders,
         runnerId,
+        environmentId,
       });
       return result;
     } catch (error) {
@@ -1660,6 +1664,7 @@ type WorkerActionHandlerFunction<
     ? z.infer<TBodySchema>
     : undefined;
   runnerId?: string;
+  environmentId?: string;
 }) => Promise<Response>;
 
 export function createActionWorkerApiRoute<
@@ -1758,6 +1763,8 @@ export function createActionWorkerApiRoute<
       }
 
       const runnerId = request.headers.get(WORKER_HEADERS.RUNNER_ID) ?? undefined;
+      // `|| undefined` so a blank header can't become a zero-match snapshot filter (→ false reject).
+      const environmentId = request.headers.get(WORKER_HEADERS.ENVIRONMENT_ID) || undefined;
 
       const result = await handler({
         params: parsedParams,
@@ -1767,6 +1774,7 @@ export function createActionWorkerApiRoute<
         body: parsedBody,
         headers: parsedHeaders,
         runnerId,
+        environmentId,
       });
       return result;
     } catch (error) {

apps/webapp/app/v3/electricShape.server.ts

@@ -0,0 +1,69 @@
+import { z } from "zod";
+
+/**
+ * OTel trace IDs are 32 lowercase hex chars. The traceparent parser only
+ * checks the dash-delimited format, so crafted ids can be persisted and later
+ * interpolated into shape `where` clauses. Validate here to close the SQLi vector.
+ */
+export const OtelTraceIdSchema = z
+  .string()
+  .regex(/^[0-9a-f]{32}$/, "traceId must be 32 lowercase hex characters");
+
+/** Params the sync routes set themselves; stripped from incoming requests. */
+export const RESERVED_ELECTRIC_SHAPE_PARAMS = new Set(["where", "table", "columns"]);
+
+const CUID_LIKE = /^[a-z][a-z0-9_]*$/i;
+
+/**
+ * Tenant column a trace shape is scoped by. TaskEvent scopes by non-null
+ * organizationId; TaskRun scopes by non-null projectId (its organizationId is
+ * nullable). The column is from this fixed union, never user input, so it's
+ * safe to interpolate.
+ */
+export type TraceScope =
+  | { column: "organizationId"; id: string }
+  | { column: "projectId"; id: string };
+
+/**
+ * Build the Electric Shape `where` clause for the trace sync routes. Both ids
+ * are re-validated as defense-in-depth so a missed call site can't bypass scope.
+ */
+export function buildElectricTraceWhereClause(args: {
+  traceId: string;
+  scope: TraceScope;
+}): string {
+  const { traceId, scope } = args;
+  if (!OtelTraceIdSchema.safeParse(traceId).success) {
+    throw new Error("buildElectricTraceWhereClause: unsafe traceId");
+  }
+  if (!CUID_LIKE.test(scope.id)) {
+    throw new Error("buildElectricTraceWhereClause: unsafe scope id");
+  }
+  return `"traceId"='${traceId}' AND "${scope.column}"='${scope.id}'`;
+}
+
+/**
+ * Characters rejected in realtime tag values — the single source of truth
+ * shared by the apiBuilder Zod refine (`realtime.v1.runs.ts`) and the runtime
+ * sanitiser. Rejects control chars/DEL, backslash, and double-quote. Single
+ * quotes are allowed and escaped (`'` → `''`) in `sanitizeRealtimeTagForSql`.
+ */
+export const UNSAFE_REALTIME_TAG_CHARS = /[\x00-\x1f\x7f\\"]/;
+
+/**
+ * Sanitise a tag value for interpolation into an Electric Shape `where` clause:
+ * reject unsafe chars, escape single quotes per SQL standard.
+ */
+export function sanitizeRealtimeTagForSql(tag: string): string {
+  if (typeof tag !== "string" || tag.length === 0) {
+    throw new Error("Invalid realtime tag: empty");
+  }
+  if (UNSAFE_REALTIME_TAG_CHARS.test(tag)) {
+    throw new Error(`Invalid realtime tag: ${JSON.stringify(tag)} — contains unsafe character`);
+  }
+  return tag.replace(/'/g, "''");
+}
+
+export function sanitizeRealtimeTagsForSql(tags: string[]): string[] {
+  return tags.map(sanitizeRealtimeTagForSql);
+}

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -82,7 +82,10 @@ export class EnvironmentVariablesRepository implements Repository {
       return { success: false as const, error: "Project not found" };
     }
 
-    if (options.environmentIds.every((v) => !project.environments.some((e) => e.id === v))) {
+    // Reject if ANY supplied environmentId is outside the caller's project.
+    // `.some` (not `.every`) so one in-project id can't let a mixed array
+    // through.
+    if (options.environmentIds.some((v) => !project.environments.some((e) => e.id === v))) {
       return { success: false as const, error: `Environment not found` };
     }
 
@@ -291,7 +294,9 @@ export class EnvironmentVariablesRepository implements Repository {
       return { success: false as const, error: "Project not found" };
     }
 
-    if (options.values.every((v) => !project.environments.some((e) => e.id === v.environmentId))) {
+    // Same guard as `create()`: reject if ANY supplied environmentId is
+    // outside the caller's project (`.some`, not `.every`).
+    if (options.values.some((v) => !project.environments.some((e) => e.id === v.environmentId))) {
       return { success: false as const, error: `Environment not found` };
     }
 

apps/webapp/app/v3/mollifier/claimTtl.ts

@@ -0,0 +1,12 @@
+// The claim is a serialization lock that must outlive the winner's create-and-publish pipeline. A short
+// customer key TTL must NOT shrink it below a pipeline floor (else the claim expires mid-pipeline and a
+// polling loser re-claims → cross-DB duplicate). Floor at `minTtlSeconds` independent of key TTL; cap at max.
+export function computeClaimTtlSeconds(input: {
+  keyExpiresAt: Date;
+  now: number;
+  minTtlSeconds: number;
+  maxTtlSeconds: number;
+}): number {
+  const keyTtlSeconds = Math.ceil((input.keyExpiresAt.getTime() - input.now) / 1000);
+  return Math.min(input.maxTtlSeconds, Math.max(input.minTtlSeconds, keyTtlSeconds));
+}