Release 4.0.0-v4-beta.0 (#1907)

* fix package versions in references

* update lockfile

.changeset/breezy-turtles-talk.md

@@ -0,0 +1,10 @@
+---
+"@trigger.dev/react-hooks": patch
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/build": patch
+"@trigger.dev/core": patch
+"@trigger.dev/rsc": patch
+---
+
+Run Engine 2.0 (alpha)

.changeset/config.json

@@ -12,7 +12,13 @@
   "access": "public",
   "baseBranch": "main",
   "updateInternalDependencies": "patch",
-  "ignore": ["webapp", "proxy", "coordinator", "docker-provider", "kubernetes-provider"],
+  "ignore": [
+    "webapp",
+    "coordinator",
+    "docker-provider",
+    "kubernetes-provider",
+    "supervisor"
+  ],
   "___experimentalUnsafeOptions_WILL_CHANGE_IN_PATCH": {
     "onlyUpdatePeerDependentsWhenOutOfRange": true
   }

.changeset/four-needles-add.md

@@ -0,0 +1,12 @@
+---
+"@trigger.dev/redis-worker": major
+"@trigger.dev/react-hooks": major
+"@trigger.dev/sdk": major
+"trigger.dev": major
+"@trigger.dev/python": major
+"@trigger.dev/build": major
+"@trigger.dev/core": major
+"@trigger.dev/rsc": major
+---
+
+Trigger.dev v4 release. Please see our upgrade to v4 docs to view the full changelog: https://trigger.dev/docs/upgrade-to-v4

.changeset/honest-files-decide.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Deprecate toolTask and replace with `ai.tool(mySchemaTask)`

.changeset/nice-colts-boil.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Improve warm start times by eagerly creating the child TaskRunProcess when a previous run as completed

.changeset/pre.json

@@ -0,0 +1,28 @@
+{
+  "mode": "pre",
+  "tag": "v4-beta",
+  "initialVersions": {
+    "coordinator": "0.0.1",
+    "docker-provider": "0.0.1",
+    "kubernetes-provider": "0.0.1",
+    "supervisor": "0.0.1",
+    "webapp": "1.0.0",
+    "@trigger.dev/build": "3.3.17",
+    "trigger.dev": "3.3.17",
+    "@trigger.dev/core": "3.3.17",
+    "@trigger.dev/python": "3.3.17",
+    "@trigger.dev/react-hooks": "3.3.17",
+    "@trigger.dev/redis-worker": "3.3.17",
+    "@trigger.dev/rsc": "3.3.17",
+    "@trigger.dev/sdk": "3.3.17"
+  },
+  "changesets": [
+    "breezy-turtles-talk",
+    "four-needles-add",
+    "honest-files-decide",
+    "nice-colts-boil",
+    "red-wasps-cover",
+    "smart-coins-hammer",
+    "weak-jobs-hide"
+  ]
+}

.changeset/red-wasps-cover.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Suppress external instrumentation for fetch calls from ApiClient

.changeset/smart-coins-hammer.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+fix: Realtime streams: prevent enqueuing into closed ReadableStream

.changeset/weak-jobs-hide.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+v4: New lifecycle hooks

.configs/tsconfig.base.json

@@ -1,7 +1,7 @@
 {
   "compilerOptions": {
     "target": "es2022",
-    "lib": ["ES2022", "DOM", "DOM.Iterable"],
+    "lib": ["ES2022", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
     "module": "NodeNext",
     "moduleResolution": "NodeNext",
     "moduleDetection": "force",

.cursor/mcp.json

@@ -0,0 +1,7 @@
+{
+  "mcpServers": {
+    "trigger.dev": {
+      "url": "http://localhost:3333/sse"
+    }
+  }
+}

.cursor/rules/executing-commands.mdc

@@ -0,0 +1,24 @@
+---
+description: how to run commands in the monorepo
+globs: 
+alwaysApply: true
+---
+Almost all commands in the monorepo should be executed when `pnpm run ...` from the root of the monorepo. For example, running tests for the `@internal/run-engine` internal package:
+
+```
+pnpm run dev --filter webapp
+```
+
+But often, when running tests, it's better to `cd` into the directory and then run tests:
+
+```
+cd apps/webapp
+pnpm run test --run
+```
+
+This way you can run for a single file easily:
+
+```
+cd internal-packages/run-engine
+pnpm run test ./src/engine/tests/ttl.test.ts --run
+```

.cursor/rules/repo.mdc

@@ -0,0 +1,6 @@
+---
+description: understanding the structure of the monorepo
+globs: 
+alwaysApply: true
+---
+We've documented the structure of our monorepo here: [repo.md](mdc:ai/references/repo.md)

.cursor/rules/webapp.mdc

@@ -0,0 +1,37 @@
+---
+description: Making updates to the main trigger.dev remix webapp
+globs: apps/webapp/**/*.tsx,apps/webapp/**/*.ts
+alwaysApply: false
+---
+
+The main trigger.dev webapp, which powers it's API and dashboard and makes up the docker image that is produced as an OSS image, is a Remix 2.1.0 app that uses an express server, written in TypeScript. The following subsystems are either included in the webapp or are used by the webapp in another part of the monorepo:
+
+- `@trigger.dev/database` exports a Prisma 5.4.1 client that is used extensively in the webapp to access a PostgreSQL instance. The schema file is [schema.prisma](mdc:internal-packages/database/prisma/schema.prisma)
+- `@trigger.dev/core` is a published package and is used to share code between the `@trigger.dev/sdk` and the webapp. It includes functionality but also a load of Zod schemas for data validation. When importing from `@trigger.dev/core` in the webapp, we never import the root `@trigger.dev/core` path, instead we favor one of the subpath exports that you can find in [package.json](mdc:packages/core/package.json)
+- `@internal/run-engine` has all the code needed to trigger a run and take it through it's lifecycle to completion.
+- `@trigger.dev/redis-worker` is a custom redis based background job/worker system that's used in the webapp and also used inside the run engine.
+
+## Environment variables and testing
+
+In the webapp, all environment variables are accessed through the `env` export of [env.server.ts](mdc:apps/webapp/app/env.server.ts), instead of directly accessing `process.env`.
+
+Ideally, the `env.server.ts` file would never be imported into a test file, either directly or indirectly. Tests should only imported classes and functions from a file matching `app/**/*.ts` of the webapp, and that file should not use environment variables, everything should be passed through as options instead. This "service/configuration" separation is important, and can be seen in a few places in the code for examples:
+
+- [realtimeClient.server.ts](mdc:apps/webapp/app/services/realtimeClient.server.ts) is the testable service, and [realtimeClientGlobal.server.ts](mdc:apps/webapp/app/services/realtimeClientGlobal.server.ts) is the configuration
+
+Also for writing tests in the webapp, checkout our [tests.md](mdc:ai/references/tests.md) guide
+
+## Legacy run engine vs Run Engine 2.0
+
+We originally the Trigger.dev "Run Engine" not as a single system, but just spread out all over the codebase, with no real separate or encapsulation. And we didn't even call it a "Run Engine". With Run Engine 2.0, we've completely rewritten big parts of the way the system works, and moved it over to an internal package called `@internal/run-engine`. So we've retroactively named the previous run engine "Legacy run engine". We're focused almost exclusively now on moving to Run Engine 2.0 and will be deprecating and removing the legacy run engine code eventually.
+
+## Where to look for code
+
+- The trigger API endpoint is [api.v1.tasks.$taskId.trigger.ts](mdc:apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts)
+- The batch trigger API endpoint is [api.v1.tasks.batch.ts](mdc:apps/webapp/app/routes/api.v1.tasks.batch.ts)
+- Setup code for the prisma client is in [db.server.ts](mdc:apps/webapp/app/db.server.ts)
+- The run engine is configured in [runEngine.server.ts](mdc:apps/webapp/app/v3/runEngine.server.ts)
+- All the "services" that are found in app/v3/services/**/*.server.ts
+- The code for the TaskEvent data, which is the otel data sent from tasks to our servers, is in both the [eventRepository.server.ts](mdc:apps/webapp/app/v3/eventRepository.server.ts) and also the [otlpExporter.server.ts](mdc:apps/webapp/app/v3/otlpExporter.server.ts). The otel endpoints which are hit from production and development otel exporters is [otel.v1.logs.ts](mdc:apps/webapp/app/routes/otel.v1.logs.ts) and [otel.v1.traces.ts](mdc:apps/webapp/app/routes/otel.v1.traces.ts)
+- We use "presenters" to move more complex loader code into a class, and you can find those are app/v3/presenters/**/*.server.ts
+

.cursor/rules/writing-tests.mdc

@@ -0,0 +1,6 @@
+---
+description: How to write tests in the monorepo
+globs: 
+alwaysApply: true
+---
+Follow our [tests.md](mdc:ai/references/tests.md) guide for how to write tests in the monorepo.

.cursorignore

@@ -0,0 +1,7 @@
+apps/docker-provider/
+apps/kubernetes-provider/
+apps/proxy/
+apps/coordinator/
+packages/rsc/
+.changeset
+.zed

.dockerignore

@@ -1,44 +1,23 @@
-\*.log
-.git
-.github
+**/*.log
+**/*.pem
+**/*.tsbuildinfo
 
-# editor
+**/.cache
+**/.env
+**/.next
+**/.output
+**/.trigger
+**/.tshy
+**/.tshy-build
+**/.turbo
+**/.vercel
+**/.wrangler
 
-.idea
-.vscode
+**/dist
+**/node_modules
 
-# dependencies
-
-node_modules
-.pnp
-.pnp.js
-
-# testing
-
-coverage
-
-# next.js
-
-.next/
-build
-
-# packages
-
-build
-dist
-packages/\*\*/dist
-
-# misc
-
-.DS_Store
-\*.pem
-
-.turbo
-.vercel
-.cache
-.output
-.trigger
-apps/\*\*/public/build
+apps/webapp/build
+apps/webapp/public/build
 
 cypress/screenshots
 cypress/videos
@@ -47,8 +26,21 @@ apps/**/styles/tailwind.css
 packages/**/styles/tailwind.css
 
 .changeset
-references
+.DS_Store
+.git
+.github
+.idea
+.pnp
+.pnp.js
+.vscode
+
+coverage
+build
+docs
 examples
+out
+references
+
 CHANGESETS.md
 CONTRIBUTING.md
 README.md

.env.example

@@ -31,10 +31,32 @@ DEV_OTEL_BATCH_PROCESSING_ENABLED="0"
 # AUTH_GITHUB_CLIENT_ID=
 # AUTH_GITHUB_CLIENT_SECRET=
 
-# Resend is an email service used for signing in to Trigger.dev via a Magic Link.
-# Emails will print to the console if you leave these commented out
+# Configure an email transport to allow users to sign in to Trigger.dev via a Magic Link.
+# If none are configured, emails will print to the console instead.
+# Uncomment one of the following blocks to allow delivery of
+
+# Resend
 ### Visit https://resend.com, create an account and get your API key. Then insert it below along with your From and Reply To email addresses. Visit https://resend.com/docs for more information.
-# RESEND_API_KEY=<api_key>
+# EMAIL_TRANSPORT=resend
+# FROM_EMAIL=
+# REPLY_TO_EMAIL=
+# RESEND_API_KEY=
+
+# Generic SMTP
+### Enter the configuration provided by your mail provider. Visit https://nodemailer.com/smtp/ for more information
+### SMTP_SECURE = false will use STARTTLS when connecting to a server that supports it (usually port 587)
+# EMAIL_TRANSPORT=smtp
+# FROM_EMAIL=
+# REPLY_TO_EMAIL=
+# SMTP_HOST=
+# SMTP_PORT=587
+# SMTP_SECURE=false
+# SMTP_USER=
+# SMTP_PASSWORD=
+
+# AWS Simple Email Service
+### Authentication is configured using the default Node.JS credentials provider chain (https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/Package/-aws-sdk-credential-providers/#fromnodeproviderchain)
+# EMAIL_TRANSPORT=aws-ses
 # FROM_EMAIL=
 # REPLY_TO_EMAIL=
 

.eslintrc.js

@@ -1,14 +0,0 @@
-module.exports = {
-  root: true,
-  // This tells ESLint to load the config from the package `eslint-config-custom`
-  extends: ["custom"],
-  settings: {
-    next: {
-      rootDir: ["apps/*/"],
-    },
-  },
-  parserOptions: {
-    sourceType: "module",
-    ecmaVersion: 2020,
-  },
-};

.github/actions/get-image-tag/action.yml

@@ -31,6 +31,11 @@ runs:
             sha=$(echo ${{ github.sha }} | head -c7)
             ts=$(date +%s)
             tag=${env}-${sha}-${ts}
+          elif [[ "${{ github.ref_name }}" == re2-*-* ]]; then
+            env=$(echo ${{ github.ref_name }} | cut -d- -f2)
+            sha=$(echo ${{ github.sha }} | head -c7)
+            ts=$(date +%s)
+            tag=${env}-${sha}-${ts}
           elif [[ "${{ github.ref_name }}" == v.docker.* ]]; then
             version="${GITHUB_REF_NAME#v.docker.}"
             tag="v${version}"

.github/copilot-instructions.md

@@ -0,0 +1 @@
+This is the repo for Trigger.dev, a background jobs platform written in TypeScript. Our webapp at apps/webapp is a Remix 2.1 app that uses Node.js v20. Our SDK is an isomorphic TypeScript SDK at packages/trigger-sdk. Always prefer using isomorphic code like fetch, ReadableStream, etc. instead of Node.js specific code. Our tests are all vitest. We use prisma in internal-packages/database for our database interactions using PostgreSQL. For TypeScript, we usually use types over interfaces. We use zod a lot in packages/core and in the webapp. Avoid enums. Use strict mode. No default exports, use function declarations.

.github/workflows/docs.yml

@@ -39,4 +39,4 @@ jobs:
             ${{ runner.os }}-mintlify
 
       - name: 🔗 Check for broken links
-        run: npx mintlify@4.0.222 broken-links
+        run: npx mintlify@4.0.393 broken-links

.github/workflows/e2e.yml

@@ -17,7 +17,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [buildjet-8vcpu-ubuntu-2204, windows-latest]
+        os: [ubuntu-latest, windows-latest]
         package-manager: ["npm", "pnpm"]
     steps:
       - name: ⬇️ Checkout repo
@@ -38,6 +38,9 @@ jobs:
       - name: 📥 Download deps
         run: pnpm install --frozen-lockfile --filter trigger.dev...
 
+      - name: 📀 Generate Prisma Client
+        run: pnpm run generate
+
       - name: 🔧 Build v3 cli monorepo dependencies
         run: pnpm run build --filter trigger.dev^...
 

.github/workflows/pr_checks.yml

@@ -28,34 +28,3 @@ jobs:
     with:
       package: cli-v3
     secrets: inherit
-
-  preview-release:
-    name: Preview Release
-    needs: [typecheck, units, e2e]
-    if: github.repository == 'triggerdotdev/trigger.dev'
-    runs-on: buildjet-8vcpu-ubuntu-2204
-    steps:
-      - name: ⬇️ Checkout repo
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          version: 8.15.5
-
-      - name: ⎔ Setup node
-        uses: buildjet/setup-node@v4
-        with:
-          node-version: 20.11.1
-          cache: "pnpm"
-
-      - name: 📥 Download deps
-        run: pnpm install --frozen-lockfile
-
-      - name: 🏗️ Build
-        run: pnpm run build --filter "@trigger.dev/*" --filter "trigger.dev"
-
-      - name: ⚡ Publish preview release
-        run: npx pkg-pr-new publish --no-template $(ls -d ./packages/*)

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

@@ -0,0 +1,95 @@
+name: "⚒️ Publish Worker RE2"
+
+on:
+  workflow_call:
+    inputs:
+      image_tag:
+        description: The image tag to publish
+        type: string
+        required: false
+        default: ""
+  push:
+    tags:
+      - "re2-test-*"
+      - "re2-prod-*"
+
+permissions:
+  packages: write
+  contents: read
+
+jobs:
+  # check-branch:
+  #   runs-on: ubuntu-latest
+  #   steps:
+  #     - name: Fail if re2-prod-* is pushed from a non-main branch
+  #       if: startsWith(github.ref_name, 're2-prod-') && github.base_ref != 'main'
+  #       run: |
+  #         echo "🚫 re2-prod-* tags can only be pushed from the main branch."
+  #         exit 1
+  build:
+    # needs: check-branch
+    strategy:
+      matrix:
+        package: [supervisor]
+    runs-on: ubuntu-latest
+    env:
+      DOCKER_BUILDKIT: "1"
+    steps:
+      - name: ⬇️ Checkout git repo
+        uses: actions/checkout@v4
+
+      - name: 📦 Get image repo
+        id: get_repository
+        run: |
+          if [[ "${{ matrix.package }}" == *-provider ]]; then
+            provider_type=$(echo "${{ matrix.package }}" | cut -d- -f1)
+            repo=provider/${provider_type}
+          else
+            repo="${{ matrix.package }}"
+          fi
+          echo "repo=${repo}" >> "$GITHUB_OUTPUT"
+
+      - id: get_tag
+        uses: ./.github/actions/get-image-tag
+        with:
+          tag: ${{ inputs.image_tag }}
+
+      - name: 🐋 Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      # ..to avoid rate limits when pulling images
+      - name: 🐳 Login to DockerHub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: 🚢 Build Container Image
+        run: |
+          docker build -t infra_image -f ./apps/${{ matrix.package }}/Containerfile .
+
+      # ..to push image
+      - name: 🐙 Login to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: 🐙 Push to GitHub Container Registry
+        run: |
+          docker tag infra_image "$REGISTRY/$REPOSITORY:$IMAGE_TAG"
+          docker push "$REGISTRY/$REPOSITORY:$IMAGE_TAG"
+        env:
+          REGISTRY: ghcr.io/triggerdotdev
+          REPOSITORY: ${{ steps.get_repository.outputs.repo }}
+          IMAGE_TAG: ${{ steps.get_tag.outputs.tag }}
+
+      - name: 🐙 Push 'v3' tag to GitHub Container Registry
+        if: steps.get_tag.outputs.is_semver == 'true'
+        run: |
+          docker tag infra_image "$REGISTRY/$REPOSITORY:v3"
+          docker push "$REGISTRY/$REPOSITORY:v3"
+        env:
+          REGISTRY: ghcr.io/triggerdotdev
+          REPOSITORY: ${{ steps.get_repository.outputs.repo }}

.github/workflows/publish-worker.yml

@@ -23,7 +23,7 @@ jobs:
     strategy:
       matrix:
         package: [coordinator, docker-provider, kubernetes-provider]
-    runs-on: buildjet-16vcpu-ubuntu-2204
+    runs-on: ubuntu-latest
     env:
       DOCKER_BUILDKIT: "1"
     steps:

.github/workflows/publish.yml

@@ -24,12 +24,10 @@ on:
       - "packages/**"
       - "!packages/**/*.md"
       - "!packages/**/*.eslintrc"
+      - "internal-packages/**"
       - "apps/**"
       - "!apps/**/*.md"
       - "!apps/**/*.eslintrc"
-      - "integrations/**"
-      - "!integrations/**/*.md"
-      - "!integrations/**/*.eslintrc"
       - "pnpm-lock.yaml"
       - "pnpm-workspace.yaml"
       - "turbo.json"
@@ -58,14 +56,14 @@ jobs:
     secrets: inherit
 
   publish-webapp:
-    needs: [typecheck, units]
+    needs: [typecheck]
     uses: ./.github/workflows/publish-webapp.yml
     secrets: inherit
     with:
       image_tag: ${{ inputs.image_tag }}
 
   publish-worker:
-    needs: [typecheck, units]
+    needs: [typecheck]
     uses: ./.github/workflows/publish-worker.yml
     secrets: inherit
     with:

.github/workflows/release.yml

@@ -17,7 +17,7 @@ concurrency:
 jobs:
   release:
     name: 🦋 Changesets Release
-    runs-on: buildjet-8vcpu-ubuntu-2204
+    runs-on: ubuntu-latest
     if: github.repository == 'triggerdotdev/trigger.dev'
     outputs:
       published: ${{ steps.changesets.outputs.published }}

.github/workflows/typecheck.yml

@@ -5,7 +5,7 @@ on:
 
 jobs:
   typecheck:
-    runs-on: buildjet-8vcpu-ubuntu-2204
+    runs-on: ubuntu-latest
 
     steps:
       - name: ⬇️ Checkout repo

.github/workflows/unit-tests.yml

@@ -6,7 +6,7 @@ on:
 jobs:
   unitTests:
     name: "🧪 Unit Tests"
-    runs-on: buildjet-8vcpu-ubuntu-2204
+    runs-on: ubuntu-latest
     steps:
       - name: ⬇️ Checkout repo
         uses: actions/checkout@v4
@@ -24,11 +24,30 @@ jobs:
           node-version: 20.11.1
           cache: "pnpm"
 
+      # ..to avoid rate limits when pulling images
+      - name: 🐳 Login to DockerHub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
       - name: 📥 Download deps
         run: pnpm install --frozen-lockfile
 
       - name: 📀 Generate Prisma Client
         run: pnpm run generate
 
-      - name: 🧪 Run Unit Tests
-        run: pnpm run test
+      - name: 🧪 Run Webapp Unit Tests
+        run: pnpm run test:webapp
+        env:
+          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
+          DIRECT_URL: postgresql://postgres:postgres@localhost:5432/postgres
+          SESSION_SECRET: "secret"
+          MAGIC_LINK_SECRET: "secret"
+          ENCRYPTION_KEY: "secret"
+      
+      - name: 🧪 Run Package Unit Tests
+        run: pnpm run test:packages
+      
+      - name: 🧪 Run Internal Unit Tests
+        run: pnpm run test:internal

.gitignore

@@ -58,3 +58,8 @@ apps/**/public/build
 .yarn
 *.tsbuildinfo
 /packages/cli-v3/src/package.json
+.husky
+/packages/react-hooks/src/package.json
+/packages/core/src/package.json
+/packages/trigger-sdk/src/package.json
+/packages/python/src/package.json

.npmrc

@@ -1,2 +1,5 @@
 link-workspace-packages=false
-public-hoist-pattern[]=*prisma*
+public-hoist-pattern[]=*prisma*
+prefer-workspace-packages=true
+update-notifier=false
+side-effects-cache=false

.vscode/extensions.json

@@ -1,8 +1,4 @@
 {
-	"recommendations": [
-        "denoland.vscode-deno"
-    ],
-	"unwantedRecommendations": [
-		
-	]
+  "recommendations": ["bierner.comment-tagged-templates"],
+  "unwantedRecommendations": []
 }

.vscode/launch.json

@@ -13,6 +13,24 @@
       "cwd": "${workspaceFolder}",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug realtimeStreams.test.ts",
+      "command": "pnpm run test -t RealtimeStreams",
+      "envFile": "${workspaceFolder}/.env",
+      "cwd": "${workspaceFolder}/apps/webapp",
+      "sourceMaps": true
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug fairDequeuingStrategy.test.ts",
+      "command": "pnpm run test -t FairDequeuingStrategy",
+      "envFile": "${workspaceFolder}/.env",
+      "cwd": "${workspaceFolder}/apps/webapp",
+      "sourceMaps": true
+    },
     {
       "type": "chrome",
       "request": "launch",
@@ -28,6 +46,14 @@
       "cwd": "${workspaceFolder}/references/init-shell",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug V3 init dev CLI",
+      "command": "pnpm exec trigger dev",
+      "cwd": "${workspaceFolder}/references/init-shell",
+      "sourceMaps": true
+    },
     {
       "type": "node-terminal",
       "request": "launch",
@@ -36,6 +62,14 @@
       "cwd": "${workspaceFolder}/references/v3-catalog",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug Dev Next.js Realtime",
+      "command": "pnpm exec trigger dev",
+      "cwd": "${workspaceFolder}/references/nextjs-realtime",
+      "sourceMaps": true
+    },
     {
       "type": "node-terminal",
       "request": "launch",
@@ -99,6 +133,30 @@
       "command": "pnpm exec trigger dev",
       "cwd": "${workspaceFolder}/references/hello-world",
       "sourceMaps": true
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug RunEngine tests",
+      "command": "pnpm run test ./src/engine/tests/releaseConcurrencyTokenBucketQueue.test.ts -t 'Should retrieve metrics for all queues via getQueueMetrics'",
+      "cwd": "${workspaceFolder}/internal-packages/run-engine",
+      "sourceMaps": true
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug RunQueue tests",
+      "command": "pnpm run test ./src/run-queue/index.test.ts",
+      "cwd": "${workspaceFolder}/internal-packages/run-engine",
+      "sourceMaps": true
+    },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug d3-demo",
+      "command": "pnpm exec trigger dev",
+      "cwd": "${workspaceFolder}/references/d3-demo",
+      "sourceMaps": true
     }
   ]
 }

.vscode/settings.json

@@ -5,5 +5,6 @@
   "search.exclude": {
     "**/node_modules/**": true,
     "packages/cli-v3/e2e": true
-  }
+  },
+  "vitest.disableWorkspaceWarning": true
 }

.zed/tasks.json

@@ -0,0 +1,45 @@
+[
+  {
+    "label": "Build packages",
+    "command": "pnpm run build --filter \"@trigger.dev/*\" --filter trigger.dev",
+    //"args": [],
+    // Env overrides for the command, will be appended to the terminal's environment from the settings.
+    "env": { "foo": "bar" },
+    // Current working directory to spawn the command into, defaults to current project root.
+    //"cwd": "/path/to/working/directory",
+    // Whether to use a new terminal tab or reuse the existing one to spawn the process, defaults to `false`.
+    "use_new_terminal": false,
+    // Whether to allow multiple instances of the same task to be run, or rather wait for the existing ones to finish, defaults to `false`.
+    "allow_concurrent_runs": false,
+    // What to do with the terminal pane and tab, after the command was started:
+    // * `always` — always show the task's pane, and focus the corresponding tab in it (default)
+    // * `no_focus` — always show the task's pane, add the task's tab in it, but don't focus it
+    // * `never` — do not alter focus, but still add/reuse the task's tab in its pane
+    "reveal": "always",
+    // What to do with the terminal pane and tab, after the command has finished:
+    // * `never` — Do nothing when the command finishes (default)
+    // * `always` — always hide the terminal tab, hide the pane also if it was the last tab in it
+    // * `on_success` — hide the terminal tab on task success only, otherwise behaves similar to `always`
+    "hide": "never",
+    // Which shell to use when running a task inside the terminal.
+    // May take 3 values:
+    // 1. (default) Use the system's default terminal configuration in /etc/passwd
+    //      "shell": "system"
+    // 2. A program:
+    //      "shell": {
+    //        "program": "sh"
+    //      }
+    // 3. A program with arguments:
+    //     "shell": {
+    //         "with_arguments": {
+    //           "program": "/bin/bash",
+    //           "args": ["--login"]
+    //         }
+    //     }
+    "shell": "system",
+    // Whether to show the task line in the output of the spawned task, defaults to `true`.
+    "show_summary": true,
+    // Whether to show the command line in the output of the spawned task, defaults to `true`.
+    "show_output": true
+  }
+]

CONTRIBUTING.md

@@ -135,23 +135,22 @@ The following steps should be followed any time you start working on a new featu
 
 1. Make sure the webapp is running on localhost:3030
 
-2. Open a terminal window and build the CLI and watch for changes
+2. Open a terminal window and build the CLI and packages and watch for changes
 
 ```sh
-pnpm run dev --filter trigger.dev
+pnpm run dev --filter trigger.dev --filter "@trigger.dev/*"
 ```
 
-2. Open a new terminal window, and anytime changes are made to the `@trigger.dev/core` package, you'll need to manually rebuild the CLI:
-
-```sh
-pnpm run build --filter trigger.dev
-```
-
-Note: You do not need to do the same for `@trigger.dev/sdk`, just core.
-
 3. Open another terminal window, and change into the `<root>/references/v3-catalog` directory.
 
-4. Run the `dev` command, which will register all the local tasks with the platform and allow you to start testing task execution:
+4. You'll need to run the following commands to setup prisma and migrate the database:
+
+```sh
+pnpm exec prisma migrate deploy
+pnpm run generate:prisma
+```
+
+5. Run the `dev` command, which will register all the local tasks with the platform and allow you to start testing task execution:
 
 ```sh
 # in <root>/references/v3-catalog
@@ -165,13 +164,13 @@ If you want additional debug logging, you can use the `--log-level debug` flag:
 pnpm exec trigger dev --log-level debug
 ```
 
-5. If you make any changes in the CLI/Core/SDK, you'll need to `CTRL+C` to exit the `dev` command and restart it to pickup changes. Any changes to the files inside of the `v3-catalog/src/trigger` dir will automatically be rebuilt by the `dev` command.
+6. If you make any changes in the CLI/Core/SDK, you'll need to `CTRL+C` to exit the `dev` command and restart it to pickup changes. Any changes to the files inside of the `v3-catalog/src/trigger` dir will automatically be rebuilt by the `dev` command.
 
-6. Navigate to the `v3-catalog` project in your local dashboard at localhost:3030 and you should see the list of tasks.
+7. Navigate to the `v3-catalog` project in your local dashboard at localhost:3030 and you should see the list of tasks.
 
-7. Go to the "Test" page in the sidebar and select a task. Then enter a payload and click "Run test". You can tell what the payloads should be by looking at the relevant task file inside the `/references/v3-catalog/src/trigger` folder. Many of them accept an empty payload.
+8. Go to the "Test" page in the sidebar and select a task. Then enter a payload and click "Run test". You can tell what the payloads should be by looking at the relevant task file inside the `/references/v3-catalog/src/trigger` folder. Many of them accept an empty payload.
 
-8. Feel free to add additional files in `v3-catalog/src/trigger` to test out specific aspects of the system, or add in edge cases.
+9. Feel free to add additional files in `v3-catalog/src/trigger` to test out specific aspects of the system, or add in edge cases.
 
 ## Running end-to-end webapp tests (deprecated)
 
@@ -230,13 +229,22 @@ pnpm run db:studio
    cd packages/database
    ```
 
-3. Create and apply the migrations
+3. Create a migration
 
    ```
-   pnpm run db:migrate:dev
+   pnpm run db:migrate:dev:create
    ```
 
-   This creates a migration file and executes the migrations against your database and applies changes to the database schema(s)
+   This creates a migration file. Check the migration file does only what you want. If you're adding any database indexes they must use `CONCURRENTLY`, otherwise they'll lock the table when executed.
+
+4. Run the migration.
+
+```
+pnpm run db:migrate:deploy
+pnpm run generate
+```
+
+This executes the migrations against your database and applies changes to the database schema(s), and then regenerates the Prisma client.
 
 4. Commit generated migrations as well as changes to the schema.prisma file
 5. If you're using VSCode you may need to restart the Typescript server in the webapp to get updated type inference. Open a TypeScript file, then open the Command Palette (View > Command Palette) and run `TypeScript: Restart TS server`.

README.md

@@ -5,7 +5,7 @@
   <img alt="Trigger.dev logo" src="https://imagedelivery.net/3TbraffuDZ4aEf8KWOmI_w/a45d1fa2-0ae8-4a39-4409-f4f934bfae00/public">
 </picture>
   
-### Open source background jobs with no timeouts
+### Open source background jobs and AI infrastructure
 
 [Discord](https://trigger.dev/discord) | [Website](https://trigger.dev) | [Issues](https://github.com/triggerdotdev/trigger.dev/issues) | [Docs](https://trigger.dev/docs)
 
@@ -15,13 +15,22 @@
 
 ## About Trigger.dev
 
-Trigger.dev is an open source platform and SDK which allows you to create long-running background jobs with no timeouts. Write normal async code, deploy, and never hit a timeout.
+Trigger.dev is an open source platform and SDK which allows you to create long-running background jobs. Write normal async code, deploy, and never hit a timeout.
 
-#### Features:
+### Key features:
 
 - JavaScript and TypeScript SDK
-- Write reliable code by default
+- No timeouts
+- Retries (with exponential backoff)
+- Queues and concurrency controls
+- Schedules and crons
+- Full Observability; logs, live trace views, advanced filtering
+- React hooks to interact with the Trigger API from your React app
+- Pipe LLM streams straight to your users through the Realtime API
+- Trigger tasks and display the run status and metadata anywhere in your app
+- Custom alerts, get notified by email, Slack or webhooks
 - No infrastructure to manage
+- Elastic (scaling)
 - Works with your existing tech stack
 
 ## In your codebase
@@ -45,7 +54,7 @@ export const helloWorld = task({
 
 ## Deployment
 
-Use our SDK to write tasks in your codebase. There's no infrastructure to manage, your tasks automatically scale and connect to our cloud. Or you can always [self-host](https://trigger.dev/docs/v3/open-source-self-hosting#overview).
+Use our SDK to write tasks in your codebase. There's no infrastructure to manage, your tasks automatically scale and connect to our cloud. Or you can always self-host.
 
 ## Environments
 
@@ -59,11 +68,19 @@ View every task in every run so you can tell exactly what happened. We provide a
 
 # Getting started
 
-Visit our docs [here](https://trigger.dev/docs/v3/introduction) for a full guide on how to get started with Trigger.dev.
+The quickest way to get started is to create an account and project in our [web app](https://cloud.trigger.dev), and follow the instructions in the onboarding. Build and deploy your first task in minutes.
 
-## Self-host
+### Useful links:
 
-If you prefer to self-host, you can follow our [self-hosting guide](https://trigger.dev/docs/v3/open-source-self-hosting#overview).
+- [Quick start](https://trigger.dev/docs/quick-start) - get up and running in minutes
+- [How it works](https://trigger.dev/docs/v3/how-it-works) - understand how Trigger.dev works under the hood
+- [Guides and examples](https://trigger.dev/docs/guides/introduction) - walk-through guides and code examples for popular frameworks and use cases
+
+## Self-hosting
+
+If you prefer to self-host Trigger.dev, you can follow our [self-hosting guide](https://trigger.dev/docs/v3/open-source-self-hosting#overview).
+
+We also have a dedicated self-hosting channel in our [Discord server](https://trigger.dev/discord) for support.
 
 ## Development
 

ai/references/repo.md

@@ -0,0 +1,37 @@
+## Repo Overview
+
+This is a pnpm 8.15.5 monorepo that uses turborepo @turbo.json. The following workspaces are relevant
+
+## Apps
+
+- <root>/apps/webapp is a remix app that is the main API and dashboard for trigger.dev
+- <root>/apps/supervisor is a node.js app that handles the execution of built tasks, interaction with the webapp through internal "engine" APIs, as well as interfacing with things like docker or kubernetes, to execute the code.
+
+## Public Packages
+
+- <root>/packages/trigger-sdk is the `@trigger.dev/sdk` main SDK package.
+- <root>/packages/cli-v3 is the `trigger.dev` CLI package. See our [CLI dev command](https://trigger.dev/docs/cli-dev.md) and [Deployment](https://trigger.dev/docs/deployment/overview.md) docs for more information.
+- <root>/packages/core is the `@trigger.dev/core` package that is shared across the SDK and other packages
+- <root>/packages/build defines the types and prebuilt build extensions for trigger.dev. See our [build extensions docs](https://trigger.dev/docs/config/extensions/overview.md) for more information.
+- <root>/packages/react-hooks defines some useful react hooks like our realtime hooks. See our [Realtime hooks](https://trigger.dev/docs/frontend/react-hooks/realtime.md) and our [Trigger hooks](https://trigger.dev/docs/frontend/react-hooks/triggering.md) for more information.
+- <root>/packages/redis-worker is the `@trigger.dev/redis-worker` package that implements a custom background job/worker sytem powered by redis for offloading work to the background, used in the webapp and also in the Run Engine 2.0.
+
+## Internal Packages
+
+- <root>/internal-packages/\* are packages that are used internally only, not published, and usually they have a tsc build step and are used in the webapp
+- <root>/internal-packages/database is the `@trigger.dev/database` package that exports a prisma client, has the schema file, and exports a few other helpers.
+- <root>/internal-packages/run-engine is the `@internal/run-engine` package that is "Run Engine 2.0" and handles moving a run all the way through it's lifecycle
+- <root>/internal-packages/redis is the `@internal/redis` package that exports Redis types and the `createRedisClient` function to unify how we create redis clients in the repo. It's not used everywhere yet, but it's the preferred way to create redis clients from now on.
+- <root>/internal-packages/testcontainers is the `@internal/testcontainers` package that exports a few useful functions for spinning up local testcontainers when writing vitest tests. See our [tests.md](./tests.md) file for more information.
+- <root>/internal-packages/zodworker is the `@internal/zodworker` package that implements a wrapper around graphile-worker that allows us to use zod to validate our background jobs. We are moving away from using graphile-worker as our background job system, replacing it with our own redis-worker package.
+
+## References
+
+- <root>/references/\* are test workspaces that we use to write and test the system. Not quite e2e tests or automated, but just a useful place to help develop new features
+
+## Other
+
+- <root>/docs is our trigger.dev/docs mintlify documentation site
+- <root>/docker/Dockerfile is the one that creates the main trigger.dev published image
+- <root>/docker/docker-compose.yml is the file we run locally to start postgresql, redis, and electric when we are doing local development. You can run it with `pnpm run docker`
+- <root>/CONTRIBUTING.md defines the steps it takes for OSS contributors to start contributing.

ai/references/tests.md

@@ -0,0 +1,86 @@
+## Running Tests
+
+We use vitest exclusively for testing. To execute tests for a particular workspace, run the following command:
+
+```bash
+pnpm run test --filter webapp
+```
+
+Prefer running tests on a single file (and first cding into the directory):
+
+```bash
+cd apps/webapp
+pnpm run test ./src/components/Button.test.ts
+```
+
+If you are cd'ing into a directory, you may have to build dependencies first:
+
+```bash
+pnpm run build --filter webapp
+cd apps/webapp
+pnpm run test ./src/components/Button.test.ts
+```
+
+## Writing Tests
+
+We use vitest for testing. We almost NEVER mock anything. Start with a top-level "describe", and have multiple "it" statements inside of it.
+
+New test files should be placed right next to the file being tested. For example:
+
+- Source file: `./src/services/MyService.ts`
+- Test file: `./src/services/MyService.test.ts`
+
+When writing anything that needs redis or postgresql, we have some internal "testcontainers" that are used to spin up a local instance, redis, or both.
+
+redisTest:
+
+```typescript
+import { redisTest } from "@internal/testcontainers";
+import { createRedisClient } from "@internal/redis";
+
+describe("redisTest", () => {
+  redisTest("should use redis", async ({ redisOptions }) => {
+    const redis = createRedisClient(redisOptions);
+
+    await redis.set("test", "test");
+    const result = await redis.get("test");
+    expect(result).toEqual("test");
+  });
+});
+```
+
+postgresTest:
+
+```typescript
+import { postgresTest } from "@internal/testcontainers";
+
+describe("postgresTest", () => {
+  postgresTest("should use postgres", async ({ prisma }) => {
+    // prisma is an instance of PrismaClient
+  });
+});
+```
+
+containerTest:
+
+```typescript
+import { containerTest } from "@internal/testcontainers";
+
+describe("containerTest", () => {
+  containerTest("should use container", async ({ prisma, redisOptions }) => {
+    // container has both prisma and redis
+  });
+});
+```
+
+## Dos and Dont's
+
+- Do not mock anything.
+- Do not use mocks in tests.
+- Do not use spies in tests.
+- Do not use stubs in tests.
+- Do not use fakes in tests.
+- Do not use sinon in tests.
+- Structure each test with a setup, action, and assertion style.
+- Feel free to write long test names.
+- If there is any randomness in the code under test, use `seedrandom` to make it deterministic by allowing the caller to provide a seed.

apps/coordinator/Containerfile

@@ -13,7 +13,7 @@ RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' +
 FROM node-20 AS base
 
 RUN apt-get update \
-  && apt-get install -y buildah ca-certificates dumb-init docker.io \
+  && apt-get install -y buildah ca-certificates dumb-init docker.io busybox \
   && rm -rf /var/lib/apt/lists/*
 
 COPY --chown=node:node .gitignore .gitignore

apps/coordinator/package.json

@@ -23,10 +23,8 @@
     "tinyexec": "^0.3.0"
   },
   "devDependencies": {
-    "@types/node": "^18",
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/coordinator/src/checkpointer.ts

@@ -1,5 +1,5 @@
 import { ExponentialBackoff } from "@trigger.dev/core/v3/apps";
-import { testDockerCheckpoint } from "@trigger.dev/core/v3/apps";
+import { testDockerCheckpoint } from "@trigger.dev/core/v3/serverOnly";
 import { nanoid } from "nanoid";
 import fs from "node:fs/promises";
 import { ChaosMonkey } from "./chaosMonkey";
@@ -27,7 +27,7 @@ type CheckpointAndPushResult =
   | { success: true; checkpoint: CheckpointData }
   | {
       success: false;
-      reason?: "CANCELED" | "DISABLED" | "ERROR" | "IN_PROGRESS" | "NO_SUPPORT" | "SKIP_RETRYING";
+      reason?: "CANCELED" | "ERROR" | "SKIP_RETRYING";
     };
 
 type CheckpointData = {
@@ -87,9 +87,14 @@ export class Checkpointer {
   #dockerMode: boolean;
 
   #logger = new SimpleStructuredLogger("checkpointer");
-  #abortControllers = new Map<string, AbortController>();
+
   #failedCheckpoints = new Map<string, unknown>();
-  #waitingForRetry = new Set<string>();
+
+  // Indexed by run ID
+  #runAbortControllers = new Map<
+    string,
+    { signal: AbortSignal; abort: AbortController["abort"] }
+  >();
 
   private registryHost: string;
   private registryNamespace: string;
@@ -189,29 +194,80 @@ export class Checkpointer {
     }
   }
 
-  async checkpointAndPush(opts: CheckpointAndPushOptions): Promise<CheckpointData | undefined> {
+  async checkpointAndPush(
+    opts: CheckpointAndPushOptions,
+    delayMs?: number
+  ): Promise<CheckpointData | undefined> {
     const start = performance.now();
     this.#logger.log(`checkpointAndPush() start`, { start, opts });
 
-    let interval: NodeJS.Timer | undefined;
+    const { runId } = opts;
 
+    let interval: NodeJS.Timer | undefined;
     if (opts.shouldHeartbeat) {
       interval = setInterval(() => {
-        this.#logger.log("Sending heartbeat", { runId: opts.runId });
-        this.opts.heartbeat(opts.runId);
+        this.#logger.log("Sending heartbeat", { runId });
+        this.opts.heartbeat(runId);
       }, 20_000);
     }
 
+    const controller = new AbortController();
+    const signal = controller.signal;
+    const abort = controller.abort.bind(controller);
+
+    const onAbort = () => {
+      this.#logger.error("Checkpoint aborted", { runId, options: opts });
+    };
+
+    signal.addEventListener("abort", onAbort, { once: true });
+
+    const removeCurrentAbortController = () => {
+      const controller = this.#runAbortControllers.get(runId);
+
+      // Ensure only the current controller is removed
+      if (controller && controller.signal === signal) {
+        this.#runAbortControllers.delete(runId);
+      }
+
+      // Remove the abort listener in case it hasn't fired
+      signal.removeEventListener("abort", onAbort);
+    };
+
+    if (!this.#dockerMode && !this.#canCheckpoint) {
+      this.#logger.error("No checkpoint support. Simulation requires docker.");
+      this.#failCheckpoint(runId, "NO_SUPPORT");
+      return;
+    }
+
+    if (this.#isRunCheckpointing(runId)) {
+      this.#logger.error("Checkpoint procedure already in progress", { options: opts });
+      this.#failCheckpoint(runId, "IN_PROGRESS");
+      return;
+    }
+
+    // This is a new checkpoint, clear any last failure for this run
+    this.#clearFailedCheckpoint(runId);
+
+    if (this.disableCheckpointSupport) {
+      this.#logger.error("Checkpoint support disabled", { options: opts });
+      this.#failCheckpoint(runId, "DISABLED");
+      return;
+    }
+
+    this.#runAbortControllers.set(runId, { signal, abort });
+
     try {
-      const result = await this.#checkpointAndPushWithBackoff(opts);
+      const result = await this.#checkpointAndPushWithBackoff(opts, { delayMs, signal });
 
       const end = performance.now();
       this.#logger.log(`checkpointAndPush() end`, {
         start,
         end,
         diff: end - start,
+        diffWithoutDelay: end - start - (delayMs ?? 0),
         opts,
         success: result.success,
+        delayMs,
       });
 
       if (!result.success) {
@@ -221,53 +277,69 @@ export class Checkpointer {
       return result.checkpoint;
     } finally {
       if (opts.shouldHeartbeat) {
+        // @ts-ignore - Some kind of node incompatible type issue
         clearInterval(interval);
       }
+      removeCurrentAbortController();
     }
   }
 
-  isCheckpointing(runId: string) {
-    return this.#abortControllers.has(runId) || this.#waitingForRetry.has(runId);
+  #isRunCheckpointing(runId: string) {
+    return this.#runAbortControllers.has(runId);
   }
 
-  cancelCheckpoint(runId: string): boolean {
+  cancelAllCheckpointsForRun(runId: string): boolean {
+    this.#logger.log("cancelAllCheckpointsForRun: call", { runId });
+
     // If the last checkpoint failed, pretend we canceled it
     // This ensures tasks don't wait for external resume messages to continue
     if (this.#hasFailedCheckpoint(runId)) {
+      this.#logger.log("cancelAllCheckpointsForRun: hasFailedCheckpoint", { runId });
       this.#clearFailedCheckpoint(runId);
       return true;
     }
 
-    if (this.#waitingForRetry.has(runId)) {
-      this.#waitingForRetry.delete(runId);
-      return true;
-    }
-
-    const controller = this.#abortControllers.get(runId);
+    const controller = this.#runAbortControllers.get(runId);
 
     if (!controller) {
-      this.#logger.debug("Nothing to cancel", { runId });
+      this.#logger.debug("cancelAllCheckpointsForRun: no abort controller", { runId });
       return false;
     }
 
-    if (controller.signal.aborted) {
-      this.#logger.debug("Controller already aborted", { runId });
+    const { abort, signal } = controller;
+
+    if (signal.aborted) {
+      this.#logger.debug("cancelAllCheckpointsForRun: signal already aborted", { runId });
       return false;
     }
 
-    controller.abort("cancelCheckpoint()");
-    this.#abortControllers.delete(runId);
+    abort("cancelCheckpoint()");
+    this.#runAbortControllers.delete(runId);
 
     return true;
   }
 
-  async #checkpointAndPushWithBackoff({
-    runId,
-    leaveRunning = true, // This mirrors kubernetes behaviour more accurately
-    projectRef,
-    deploymentVersion,
-    attemptNumber,
-  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
+  async #checkpointAndPushWithBackoff(
+    {
+      runId,
+      leaveRunning = true, // This mirrors kubernetes behaviour more accurately
+      projectRef,
+      deploymentVersion,
+      attemptNumber,
+    }: CheckpointAndPushOptions,
+    { delayMs, signal }: { delayMs?: number; signal: AbortSignal }
+  ): Promise<CheckpointAndPushResult> {
+    if (delayMs && delayMs > 0) {
+      this.#logger.log("Delaying checkpoint", { runId, delayMs });
+
+      try {
+        await setTimeout(delayMs, undefined, { signal });
+      } catch (error) {
+        this.#logger.log("Checkpoint canceled during initial delay", { runId });
+        return { success: false, reason: "CANCELED" };
+      }
+    }
+
     this.#logger.log("Checkpointing with backoff", {
       runId,
       leaveRunning,
@@ -290,24 +362,24 @@ export class Checkpointer {
             delay,
           });
 
-          this.#waitingForRetry.add(runId);
-          await setTimeout(delay.milliseconds);
-
-          if (!this.#waitingForRetry.has(runId)) {
-            this.#logger.log("Checkpoint canceled while waiting for retry", { runId });
+          try {
+            await setTimeout(delay.milliseconds, undefined, { signal });
+          } catch (error) {
+            this.#logger.log("Checkpoint canceled during retry delay", { runId });
             return { success: false, reason: "CANCELED" };
-          } else {
-            this.#waitingForRetry.delete(runId);
           }
         }
 
-        const result = await this.#checkpointAndPush({
-          runId,
-          leaveRunning,
-          projectRef,
-          deploymentVersion,
-          attemptNumber,
-        });
+        const result = await this.#checkpointAndPush(
+          {
+            runId,
+            leaveRunning,
+            projectRef,
+            deploymentVersion,
+            attemptNumber,
+          },
+          { signal }
+        );
 
         if (result.success) {
           return result;
@@ -319,24 +391,6 @@ export class Checkpointer {
           return result;
         }
 
-        if (result.reason === "IN_PROGRESS") {
-          this.#logger.log("Checkpoint already in progress, won't retry", { runId });
-          this.#failCheckpoint(runId, result.reason);
-          return result;
-        }
-
-        if (result.reason === "NO_SUPPORT") {
-          this.#logger.log("No checkpoint support, won't retry", { runId });
-          this.#failCheckpoint(runId, result.reason);
-          return result;
-        }
-
-        if (result.reason === "DISABLED") {
-          this.#logger.log("Checkpoint support disabled, won't retry", { runId });
-          this.#failCheckpoint(runId, result.reason);
-          return result;
-        }
-
         if (result.reason === "SKIP_RETRYING") {
           this.#logger.log("Skipping retrying", { runId });
           return result;
@@ -364,13 +418,16 @@ export class Checkpointer {
     return { success: false, reason: "ERROR" };
   }
 
-  async #checkpointAndPush({
-    runId,
-    leaveRunning = true, // This mirrors kubernetes behaviour more accurately
-    projectRef,
-    deploymentVersion,
-    attemptNumber,
-  }: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
+  async #checkpointAndPush(
+    {
+      runId,
+      leaveRunning = true, // This mirrors kubernetes behaviour more accurately
+      projectRef,
+      deploymentVersion,
+      attemptNumber,
+    }: CheckpointAndPushOptions,
+    { signal }: { signal: AbortSignal }
+  ): Promise<CheckpointAndPushResult> {
     await this.init();
 
     const options = {
@@ -381,39 +438,12 @@ export class Checkpointer {
       attemptNumber,
     };
 
-    if (!this.#dockerMode && !this.#canCheckpoint) {
-      this.#logger.error("No checkpoint support. Simulation requires docker.");
-      return { success: false, reason: "NO_SUPPORT" };
-    }
-
-    if (this.isCheckpointing(runId)) {
-      this.#logger.error("Checkpoint procedure already in progress", { options });
-      return { success: false, reason: "IN_PROGRESS" };
-    }
-
-    // This is a new checkpoint, clear any last failure for this run
-    this.#clearFailedCheckpoint(runId);
-
-    if (this.disableCheckpointSupport) {
-      this.#logger.error("Checkpoint support disabled", { options });
-      return { success: false, reason: "DISABLED" };
-    }
-
-    const controller = new AbortController();
-    this.#abortControllers.set(runId, controller);
-
-    const onAbort = () => {
-      this.#logger.error("Checkpoint aborted", { options });
-      controller.signal.removeEventListener("abort", onAbort);
-    };
-    controller.signal.addEventListener("abort", onAbort);
-
     const shortCode = nanoid(8);
     const imageRef = this.#getImageRef(projectRef, deploymentVersion, shortCode);
     const exportLocation = this.#getExportLocation(projectRef, deploymentVersion, shortCode);
 
-    const buildah = new Buildah({ id: `${runId}-${shortCode}`, abortSignal: controller.signal });
-    const crictl = new Crictl({ id: `${runId}-${shortCode}`, abortSignal: controller.signal });
+    const buildah = new Buildah({ id: `${runId}-${shortCode}`, abortSignal: signal });
+    const crictl = new Crictl({ id: `${runId}-${shortCode}`, abortSignal: signal });
 
     const cleanup = async () => {
       const metadata = {
@@ -435,29 +465,26 @@ export class Checkpointer {
       } catch (error) {
         this.#logger.error("Error during cleanup", { ...metadata, error });
       }
-
-      this.#abortControllers.delete(runId);
-      controller.signal.removeEventListener("abort", onAbort);
     };
 
     try {
       await this.chaosMonkey.call();
 
-      this.#logger.log("Checkpointing:", { options });
+      this.#logger.log("checkpointAndPush: checkpointing", { options });
 
       const containterName = this.#getRunContainerName(runId);
 
       // Create checkpoint (docker)
       if (this.#dockerMode) {
         await this.#createDockerCheckpoint(
-          controller.signal,
+          signal,
           runId,
           exportLocation,
           leaveRunning,
           attemptNumber
         );
 
-        this.#logger.log("checkpoint created:", {
+        this.#logger.log("checkpointAndPush: checkpoint created", {
           runId,
           location: exportLocation,
         });
@@ -558,13 +585,16 @@ export class Checkpointer {
         }
       }
 
-      this.#logger.error("Unhandled checkpoint error", { options, error });
+      this.#logger.error("Unhandled checkpoint error", {
+        options,
+        error: error instanceof Error ? error.message : error,
+      });
 
       return { success: false, reason: "ERROR" };
     } finally {
       await cleanup();
 
-      if (controller.signal.aborted) {
+      if (signal.aborted) {
         this.#logger.error("Checkpoint canceled: Cleanup", { options });
 
         // Overrides any prior return value
@@ -573,6 +603,16 @@ export class Checkpointer {
     }
   }
 
+  async unpause(runId: string, attemptNumber?: number): Promise<void> {
+    try {
+      const containterNameWithAttempt = this.#getRunContainerName(runId, attemptNumber);
+      const exec = new Exec({ logger: this.#logger });
+      await exec.x("docker", ["unpause", containterNameWithAttempt]);
+    } catch (error) {
+      this.#logger.error("[Docker] Error during unpause", { runId, attemptNumber, error });
+    }
+  }
+
   async #createDockerCheckpoint(
     abortSignal: AbortSignal,
     runId: string,

apps/coordinator/src/exec.ts

@@ -64,7 +64,18 @@ export class Exec {
       command,
       argsRaw: args,
       argsTrimmed,
-      ...output,
+      globalOpts: {
+        trimArgs: this.trimArgs,
+        neverThrow: this.neverThrow,
+        hasAbortSignal: !!this.abortSignal,
+      },
+      localOpts: opts,
+      stdout: output.stdout,
+      stderr: output.stderr,
+      pid: result.pid,
+      exitCode: result.exitCode,
+      aborted: result.aborted,
+      killed: result.killed,
     };
 
     if (this.logOutput) {

apps/coordinator/src/util.ts

@@ -17,3 +17,15 @@ export const numFromEnv = (env: string, defaultValue: number): number => {
 
   return parseInt(value, 10);
 };
+
+export function safeJsonParse(json?: string): unknown {
+  if (!json) {
+    return;
+  }
+
+  try {
+    return JSON.parse(json);
+  } catch (e) {
+    return null;
+  }
+}

apps/coordinator/tsconfig.json

@@ -1,8 +1,6 @@
 {
-  "include": ["./src/**/*.ts"],
-  "exclude": ["node_modules"],
   "compilerOptions": {
-    "target": "es2016",
+    "target": "es2018",
     "module": "commonjs",
     "esModuleInterop": true,
     "resolveJsonModule": true,

apps/docker-provider/package.json

@@ -20,10 +20,8 @@
     "execa": "^8.0.1"
   },
   "devDependencies": {
-    "@types/node": "^18.19.8",
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/docker-provider/src/index.ts

@@ -7,7 +7,8 @@ import {
   TaskOperationsRestoreOptions,
 } from "@trigger.dev/core/v3/apps";
 import { SimpleLogger } from "@trigger.dev/core/v3/apps";
-import { isExecaChildProcess, testDockerCheckpoint } from "@trigger.dev/core/v3/apps";
+import { isExecaChildProcess } from "@trigger.dev/core/v3/apps";
+import { testDockerCheckpoint } from "@trigger.dev/core/v3/serverOnly";
 import { setTimeout } from "node:timers/promises";
 import { PostStartCauses, PreStopCauses } from "@trigger.dev/core/v3";
 
@@ -122,6 +123,7 @@ class DockerTaskOperations implements TaskOperations {
       `--env=POD_NAME=${containerName}`,
       `--env=COORDINATOR_HOST=${COORDINATOR_HOST}`,
       `--env=COORDINATOR_PORT=${COORDINATOR_PORT}`,
+      `--env=TRIGGER_POD_SCHEDULED_AT_MS=${Date.now()}`,
       `--name=${containerName}`,
     ];
 
@@ -129,6 +131,10 @@ class DockerTaskOperations implements TaskOperations {
       runArgs.push(`--cpus=${opts.machine.cpu}`, `--memory=${opts.machine.memory}G`);
     }
 
+    if (opts.dequeuedAt) {
+      runArgs.push(`--env=TRIGGER_RUN_DEQUEUED_AT_MS=${opts.dequeuedAt}`);
+    }
+
     runArgs.push(`${opts.image}`);
 
     try {

apps/kubernetes-provider/package.json

@@ -23,7 +23,6 @@
   "devDependencies": {
     "dotenv": "^16.4.2",
     "esbuild": "^0.19.11",
-    "tsx": "^4.7.0",
-    "typescript": "^5.3.3"
+    "tsx": "^4.7.0"
   }
 }

apps/kubernetes-provider/src/index.ts

@@ -17,6 +17,8 @@ import {
 import { PodCleaner } from "./podCleaner";
 import { TaskMonitor } from "./taskMonitor";
 import { UptimeHeartbeat } from "./uptimeHeartbeat";
+import { assertExhaustive } from "@trigger.dev/core";
+import { CustomLabelHelper } from "./labelHelper";
 
 const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local";
 const NODE_NAME = process.env.NODE_NAME || "local";
@@ -34,6 +36,18 @@ const UPTIME_MAX_PENDING_RUNS = Number(process.env.UPTIME_MAX_PENDING_RUNS || "2
 const UPTIME_MAX_PENDING_INDECES = Number(process.env.UPTIME_MAX_PENDING_INDECES || "10");
 const UPTIME_MAX_PENDING_ERRORS = Number(process.env.UPTIME_MAX_PENDING_ERRORS || "10");
 
+const POD_EPHEMERAL_STORAGE_SIZE_LIMIT = process.env.POD_EPHEMERAL_STORAGE_SIZE_LIMIT || "10Gi";
+const POD_EPHEMERAL_STORAGE_SIZE_REQUEST = process.env.POD_EPHEMERAL_STORAGE_SIZE_REQUEST || "2Gi";
+
+// Image config
+const PRE_PULL_DISABLED = process.env.PRE_PULL_DISABLED === "true";
+const ADDITIONAL_PULL_SECRETS = process.env.ADDITIONAL_PULL_SECRETS;
+const PAUSE_IMAGE = process.env.PAUSE_IMAGE || "registry.k8s.io/pause:3.9";
+const BUSYBOX_IMAGE = process.env.BUSYBOX_IMAGE || "registry.digitalocean.com/trigger/busybox";
+const DEPLOYMENT_IMAGE_PREFIX = process.env.DEPLOYMENT_IMAGE_PREFIX;
+const RESTORE_IMAGE_PREFIX = process.env.RESTORE_IMAGE_PREFIX;
+const UTILITY_IMAGE_PREFIX = process.env.UTILITY_IMAGE_PREFIX;
+
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 logger.log(`running in ${RUNTIME_ENV} mode`);
 
@@ -60,6 +74,8 @@ class KubernetesTaskOperations implements TaskOperations {
     apps: k8s.AppsV1Api;
   };
 
+  #labelHelper = new CustomLabelHelper();
+
   constructor(opts: { namespace?: string } = {}) {
     if (opts.namespace) {
       this.#namespace.metadata.name = opts.namespace;
@@ -98,7 +114,7 @@ class KubernetesTaskOperations implements TaskOperations {
               containers: [
                 {
                   name: this.#getIndexContainerName(opts.shortCode),
-                  image: opts.imageRef,
+                  image: getImageRef("deployment", opts.imageRef),
                   ports: [
                     {
                       containerPort: 8000,
@@ -152,6 +168,7 @@ class KubernetesTaskOperations implements TaskOperations {
           name: containerName,
           namespace: this.#namespace.metadata.name,
           labels: {
+            ...this.#labelHelper.getAdditionalLabels("create"),
             ...this.#getSharedLabels(opts),
             app: "task-run",
             "app.kubernetes.io/part-of": "trigger-worker",
@@ -165,7 +182,7 @@ class KubernetesTaskOperations implements TaskOperations {
           containers: [
             {
               name: containerName,
-              image: opts.image,
+              image: getImageRef("deployment", opts.image),
               ports: [
                 {
                   containerPort: 8000,
@@ -185,6 +202,9 @@ class KubernetesTaskOperations implements TaskOperations {
                   name: "TRIGGER_RUN_ID",
                   value: opts.runId,
                 },
+                ...(opts.dequeuedAt
+                  ? [{ name: "TRIGGER_RUN_DEQUEUED_AT_MS", value: String(opts.dequeuedAt) }]
+                  : []),
               ],
               volumeMounts: [
                 {
@@ -213,6 +233,7 @@ class KubernetesTaskOperations implements TaskOperations {
           name: `${this.#getRunContainerName(opts.runId)}-${opts.checkpointId.slice(-8)}`,
           namespace: this.#namespace.metadata.name,
           labels: {
+            ...this.#labelHelper.getAdditionalLabels("restore"),
             ...this.#getSharedLabels(opts),
             app: "task-run",
             "app.kubernetes.io/part-of": "trigger-worker",
@@ -226,12 +247,12 @@ class KubernetesTaskOperations implements TaskOperations {
           initContainers: [
             {
               name: "pull-base-image",
-              image: opts.imageRef,
+              image: getImageRef("deployment", opts.imageRef),
               command: ["sleep", "0"],
             },
             {
               name: "populate-taskinfo",
-              image: "registry.digitalocean.com/trigger/busybox",
+              image: getImageRef("utility", BUSYBOX_IMAGE),
               imagePullPolicy: "IfNotPresent",
               command: ["/bin/sh", "-c"],
               args: ["printenv COORDINATOR_HOST | tee /etc/taskinfo/coordinator-host"],
@@ -247,7 +268,7 @@ class KubernetesTaskOperations implements TaskOperations {
           containers: [
             {
               name: this.#getRunContainerName(opts.runId),
-              image: opts.checkpointRef,
+              image: getImageRef("restore", opts.checkpointRef),
               ports: [
                 {
                   containerPort: 8000,
@@ -298,6 +319,11 @@ class KubernetesTaskOperations implements TaskOperations {
   }
 
   async prePullDeployment(opts: TaskOperationsPrePullDeploymentOptions) {
+    if (PRE_PULL_DISABLED) {
+      logger.debug("Pre-pull is disabled, skipping.", { opts });
+      return;
+    }
+
     const metaName = this.#getPrePullContainerName(opts.shortCode);
 
     const metaLabels = {
@@ -329,10 +355,26 @@ class KubernetesTaskOperations implements TaskOperations {
             spec: {
               ...this.#defaultPodSpec,
               restartPolicy: "Always",
+              affinity: {
+                nodeAffinity: {
+                  requiredDuringSchedulingIgnoredDuringExecution: {
+                    nodeSelectorTerms: [
+                      {
+                        matchExpressions: [
+                          {
+                            key: "trigger.dev/pre-pull-disabled",
+                            operator: "DoesNotExist",
+                          },
+                        ],
+                      },
+                    ],
+                  },
+                },
+              },
               initContainers: [
                 {
                   name: "prepull",
-                  image: opts.imageRef,
+                  image: getImageRef("deployment", opts.imageRef),
                   command: ["/usr/bin/true"],
                   resources: {
                     limits: {
@@ -346,7 +388,7 @@ class KubernetesTaskOperations implements TaskOperations {
               containers: [
                 {
                   name: "pause",
-                  image: "registry.k8s.io/pause:3.9",
+                  image: getImageRef("utility", PAUSE_IMAGE),
                   resources: {
                     limits: {
                       cpu: "1m",
@@ -377,17 +419,20 @@ class KubernetesTaskOperations implements TaskOperations {
   }
 
   get #defaultPodSpec(): Omit<k8s.V1PodSpec, "containers"> {
+    const pullSecrets = ["registry-trigger", "registry-trigger-failover"];
+
+    if (ADDITIONAL_PULL_SECRETS) {
+      pullSecrets.push(...ADDITIONAL_PULL_SECRETS.split(","));
+    }
+
+    const imagePullSecrets = pullSecrets.map(
+      (name) => ({ name }) satisfies k8s.V1LocalObjectReference
+    );
+
     return {
       restartPolicy: "Never",
       automountServiceAccountToken: false,
-      imagePullSecrets: [
-        {
-          name: "registry-trigger",
-        },
-        {
-          name: "registry-trigger-failover",
-        },
-      ],
+      imagePullSecrets,
       nodeSelector: {
         nodetype: "worker",
       },
@@ -396,13 +441,13 @@ class KubernetesTaskOperations implements TaskOperations {
 
   get #defaultResourceRequests(): ResourceQuantities {
     return {
-      "ephemeral-storage": "2Gi",
+      "ephemeral-storage": POD_EPHEMERAL_STORAGE_SIZE_REQUEST,
     };
   }
 
   get #defaultResourceLimits(): ResourceQuantities {
     return {
-      "ephemeral-storage": "10Gi",
+      "ephemeral-storage": POD_EPHEMERAL_STORAGE_SIZE_LIMIT,
     };
   }
 
@@ -475,6 +520,10 @@ class KubernetesTaskOperations implements TaskOperations {
           },
         },
       },
+      {
+        name: "TRIGGER_POD_SCHEDULED_AT_MS",
+        value: Date.now().toString(),
+      },
       ...this.#coordinatorEnvVars,
     ];
   }
@@ -647,6 +696,26 @@ class KubernetesTaskOperations implements TaskOperations {
   }
 }
 
+type ImageType = "deployment" | "restore" | "utility";
+
+function getImagePrefix(type: ImageType) {
+  switch (type) {
+    case "deployment":
+      return DEPLOYMENT_IMAGE_PREFIX;
+    case "restore":
+      return RESTORE_IMAGE_PREFIX;
+    case "utility":
+      return UTILITY_IMAGE_PREFIX;
+    default:
+      assertExhaustive(type);
+  }
+}
+
+function getImageRef(type: ImageType, ref: string) {
+  const prefix = getImagePrefix(type);
+  return prefix ? `${prefix}/${ref}` : ref;
+}
+
 const provider = new ProviderShell({
   tasks: new KubernetesTaskOperations({
     namespace: KUBERNETES_NAMESPACE,
@@ -658,6 +727,7 @@ provider.listen();
 
 const taskMonitor = new TaskMonitor({
   runtimeEnv: RUNTIME_ENV,
+  namespace: KUBERNETES_NAMESPACE,
   onIndexFailure: async (deploymentId, details) => {
     logger.log("Indexing failed", { deploymentId, details });
 

apps/kubernetes-provider/src/labelHelper.ts

@@ -0,0 +1,153 @@
+import { assertExhaustive } from "@trigger.dev/core";
+
+const CREATE_LABEL_ENV_VAR_PREFIX = "DEPLOYMENT_LABEL_";
+const RESTORE_LABEL_ENV_VAR_PREFIX = "RESTORE_LABEL_";
+const LABEL_SAMPLE_RATE_POSTFIX = "_SAMPLE_RATE";
+
+type OperationType = "create" | "restore";
+
+type CustomLabel = {
+  key: string;
+  value: string;
+  sampleRate: number;
+};
+
+export class CustomLabelHelper {
+  // Labels and sample rates are defined in environment variables so only need to be computed once
+  private createLabels?: CustomLabel[];
+  private restoreLabels?: CustomLabel[];
+
+  private getLabelPrefix(type: OperationType) {
+    const prefix = type === "create" ? CREATE_LABEL_ENV_VAR_PREFIX : RESTORE_LABEL_ENV_VAR_PREFIX;
+    return prefix.toLowerCase();
+  }
+
+  private getLabelSampleRatePostfix() {
+    return LABEL_SAMPLE_RATE_POSTFIX.toLowerCase();
+  }
+
+  // Can only range from 0 to 1
+  private fractionFromPercent(percent: number) {
+    return Math.min(1, Math.max(0, percent / 100));
+  }
+
+  private isLabelSampleRateEnvVar(key: string) {
+    return key.toLowerCase().endsWith(this.getLabelSampleRatePostfix());
+  }
+
+  private isLabelEnvVar(type: OperationType, key: string) {
+    const prefix = this.getLabelPrefix(type);
+    return key.toLowerCase().startsWith(prefix) && !this.isLabelSampleRateEnvVar(key);
+  }
+
+  private getSampleRateEnvVarKey(type: OperationType, envKey: string) {
+    return `${envKey.toLowerCase()}${this.getLabelSampleRatePostfix()}`;
+  }
+
+  private getLabelNameFromEnvVarKey(type: OperationType, key: string) {
+    return key
+      .slice(this.getLabelPrefix(type).length)
+      .toLowerCase()
+      .replace(/___/g, ".")
+      .replace(/__/g, "/")
+      .replace(/_/g, "-");
+  }
+
+  private getCaseInsensitiveEnvValue(key: string) {
+    for (const [envKey, value] of Object.entries(process.env)) {
+      if (envKey.toLowerCase() === key.toLowerCase()) {
+        return value;
+      }
+    }
+  }
+
+  /** Returns the sample rate for a given label as fraction of 100 */
+  private getSampleRateFromEnvVarKey(type: OperationType, envKey: string) {
+    // Apply default: always sample
+    const DEFAULT_SAMPLE_RATE_PERCENT = 100;
+    const defaultSampleRateFraction = this.fractionFromPercent(DEFAULT_SAMPLE_RATE_PERCENT);
+
+    const value = this.getCaseInsensitiveEnvValue(this.getSampleRateEnvVarKey(type, envKey));
+
+    if (!value) {
+      return defaultSampleRateFraction;
+    }
+
+    const sampleRatePercent = parseFloat(value || String(DEFAULT_SAMPLE_RATE_PERCENT));
+
+    if (isNaN(sampleRatePercent)) {
+      return defaultSampleRateFraction;
+    }
+
+    const fractionalSampleRate = this.fractionFromPercent(sampleRatePercent);
+
+    return fractionalSampleRate;
+  }
+
+  private getCustomLabels(type: OperationType): CustomLabel[] {
+    switch (type) {
+      case "create":
+        if (this.createLabels) {
+          return this.createLabels;
+        }
+        break;
+      case "restore":
+        if (this.restoreLabels) {
+          return this.restoreLabels;
+        }
+        break;
+      default:
+        assertExhaustive(type);
+    }
+
+    const customLabels: CustomLabel[] = [];
+
+    for (const [envKey, value] of Object.entries(process.env)) {
+      const key = envKey.toLowerCase();
+
+      // Only process env vars that start with the expected prefix
+      if (!this.isLabelEnvVar(type, key)) {
+        continue;
+      }
+
+      // Skip sample rates - deal with them separately
+      if (this.isLabelSampleRateEnvVar(key)) {
+        continue;
+      }
+
+      const labelName = this.getLabelNameFromEnvVarKey(type, key);
+      const sampleRate = this.getSampleRateFromEnvVarKey(type, key);
+
+      const label = {
+        key: labelName,
+        value: value || "",
+        sampleRate,
+      } satisfies CustomLabel;
+
+      customLabels.push(label);
+    }
+
+    return customLabels;
+  }
+
+  getAdditionalLabels(type: OperationType): Record<string, string> {
+    const labels = this.getCustomLabels(type);
+
+    const additionalLabels: Record<string, string> = {};
+
+    for (const { key, value, sampleRate } of labels) {
+      // Always apply label if sample rate is 1
+      if (sampleRate === 1) {
+        additionalLabels[key] = value;
+        continue;
+      }
+
+      if (Math.random() <= sampleRate) {
+        additionalLabels[key] = value;
+        continue;
+      }
+    }
+
+    return additionalLabels;
+  }
+}

apps/kubernetes-provider/src/taskMonitor.ts

@@ -160,7 +160,10 @@ export class TaskMonitor {
 
     let reason = rawReason || "Unknown error";
     let logs = rawLogs || "";
-    let overrideCompletion = false;
+
+    /** This will only override existing task errors. It will not crash the run.  */
+    let onlyOverrideExistingError = exitCode === EXIT_CODE_CHILD_NONZERO;
+
     let errorCode: TaskRunInternalError["code"] = TaskRunErrorCodes.POD_UNKNOWN_ERROR;
 
     switch (rawReason) {
@@ -185,10 +188,8 @@ export class TaskMonitor {
         }
         break;
       case "OOMKilled":
-        overrideCompletion = true;
-        reason = `${
-          exitCode === EXIT_CODE_CHILD_NONZERO ? "Child process" : "Parent process"
-        } ran out of memory! Try choosing a machine preset with more memory for this task.`;
+        reason =
+          "[TaskMonitor] Your task ran out of memory. Try increasing the machine specs. If this doesn't fix it there might be a memory leak.";
         errorCode = TaskRunErrorCodes.TASK_PROCESS_OOM_KILLED;
         break;
       default:
@@ -199,7 +200,7 @@ export class TaskMonitor {
       exitCode,
       reason,
       logs,
-      overrideCompletion,
+      overrideCompletion: onlyOverrideExistingError,
       errorCode,
     } satisfies FailureDetails;
 

apps/kubernetes-provider/tsconfig.json

@@ -1,6 +1,6 @@
 {
   "compilerOptions": {
-    "target": "es2016",
+    "target": "es2018",
     "module": "commonjs",
     "esModuleInterop": true,
     "forceConsistentCasingInFileNames": true,
@@ -8,6 +8,8 @@
     "strict": true,
     "skipLibCheck": true,
     "paths": {
+      "@trigger.dev/core": ["../../packages/core/src"],
+      "@trigger.dev/core/*": ["../../packages/core/src/*"],
       "@trigger.dev/core/v3": ["../../packages/core/src/v3"],
       "@trigger.dev/core/v3/*": ["../../packages/core/src/v3/*"]
     }

apps/proxy/.dev.vars.example

@@ -1,7 +0,0 @@
-REWRITE_HOSTNAME=
-AWS_SQS_ACCESS_KEY_ID=
-AWS_SQS_SECRET_ACCESS_KEY=
-AWS_SQS_QUEUE_URL=
-AWS_SQS_REGION=
-#optional
-#REWRITE_PORT=

apps/proxy/.editorconfig

@@ -1,13 +0,0 @@
-# http://editorconfig.org
-root = true
-
-[*]
-indent_style = tab
-tab_width = 2
-end_of_line = lf
-charset = utf-8
-trim_trailing_whitespace = true
-insert_final_newline = true
-
-[*.yml]
-indent_style = space

apps/proxy/.gitignore

@@ -1,172 +0,0 @@
-# Logs
-
-logs
-_.log
-npm-debug.log_
-yarn-debug.log*
-yarn-error.log*
-lerna-debug.log*
-.pnpm-debug.log*
-
-# Diagnostic reports (https://nodejs.org/api/report.html)
-
-report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
-
-# Runtime data
-
-pids
-_.pid
-_.seed
-\*.pid.lock
-
-# Directory for instrumented libs generated by jscoverage/JSCover
-
-lib-cov
-
-# Coverage directory used by tools like istanbul
-
-coverage
-\*.lcov
-
-# nyc test coverage
-
-.nyc_output
-
-# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files)
-
-.grunt
-
-# Bower dependency directory (https://bower.io/)
-
-bower_components
-
-# node-waf configuration
-
-.lock-wscript
-
-# Compiled binary addons (https://nodejs.org/api/addons.html)
-
-build/Release
-
-# Dependency directories
-
-node_modules/
-jspm_packages/
-
-# Snowpack dependency directory (https://snowpack.dev/)
-
-web_modules/
-
-# TypeScript cache
-
-\*.tsbuildinfo
-
-# Optional npm cache directory
-
-.npm
-
-# Optional eslint cache
-
-.eslintcache
-
-# Optional stylelint cache
-
-.stylelintcache
-
-# Microbundle cache
-
-.rpt2_cache/
-.rts2_cache_cjs/
-.rts2_cache_es/
-.rts2_cache_umd/
-
-# Optional REPL history
-
-.node_repl_history
-
-# Output of 'npm pack'
-
-\*.tgz
-
-# Yarn Integrity file
-
-.yarn-integrity
-
-# dotenv environment variable files
-
-.env
-.env.development.local
-.env.test.local
-.env.production.local
-.env.local
-
-# parcel-bundler cache (https://parceljs.org/)
-
-.cache
-.parcel-cache
-
-# Next.js build output
-
-.next
-out
-
-# Nuxt.js build / generate output
-
-.nuxt
-dist
-
-# Gatsby files
-
-.cache/
-
-# Comment in the public line in if your project uses Gatsby and not Next.js
-
-# https://nextjs.org/blog/next-9-1#public-directory-support
-
-# public
-
-# vuepress build output
-
-.vuepress/dist
-
-# vuepress v2.x temp and cache directory
-
-.temp
-.cache
-
-# Docusaurus cache and generated files
-
-.docusaurus
-
-# Serverless directories
-
-.serverless/
-
-# FuseBox cache
-
-.fusebox/
-
-# DynamoDB Local files
-
-.dynamodb/
-
-# TernJS port file
-
-.tern-port
-
-# Stores VSCode versions used for testing VSCode extensions
-
-.vscode-test
-
-# yarn v2
-
-.yarn/cache
-.yarn/unplugged
-.yarn/build-state.yml
-.yarn/install-state.gz
-.pnp.\*
-
-# wrangler project
-
-.dev.vars
-.wrangler/

apps/proxy/.prettierrc

@@ -1,11 +0,0 @@
-{
-	"semi": true,
-	"singleQuote": false,
-	"jsxSingleQuote": false,
-	"trailingComma": "es5",
-	"bracketSpacing": true,
-	"bracketSameLine": false,
-	"printWidth": 100,
-	"tabWidth": 2,
-	"useTabs": false
-}

apps/proxy/CHANGELOG.md

@@ -1,72 +0,0 @@
-# proxy
-
-## 0.0.11
-
-### Patch Changes
-
-- @trigger.dev/core@2.3.5
-
-## 0.0.10
-
-### Patch Changes
-
-- @trigger.dev/core@2.3.4
-
-## 0.0.9
-
-### Patch Changes
-
-- @trigger.dev/core@2.3.3
-
-## 0.0.8
-
-### Patch Changes
-
-- @trigger.dev/core@2.3.2
-
-## 0.0.7
-
-### Patch Changes
-
-- Updated dependencies [f3efcc0c]
-  - @trigger.dev/core@2.3.1
-
-## 0.0.6
-
-### Patch Changes
-
-- Updated dependencies [17f6f29d]
-  - @trigger.dev/core@2.3.0
-
-## 0.0.5
-
-### Patch Changes
-
-- @trigger.dev/core@2.2.11
-
-## 0.0.4
-
-### Patch Changes
-
-- @trigger.dev/core@2.2.10
-
-## 0.0.3
-
-### Patch Changes
-
-- Updated dependencies [6ebd435e]
-  - @trigger.dev/core@2.2.9
-
-## 0.0.2
-
-### Patch Changes
-
-- Updated dependencies [067e19fe]
-  - @trigger.dev/core@2.2.8
-
-## 0.0.1
-
-### Patch Changes
-
-- Updated dependencies [756024da]
-  - @trigger.dev/core@2.2.7

apps/proxy/README.md

@@ -1,68 +0,0 @@
-# Trigger.dev proxy
-
-This is an optional module that can be used to proxy and queue requests to the Trigger.dev API.
-
-## Why?
-
-The Trigger.dev API is designed to be fast and reliable. However, if you have a lot of traffic, you may want to use this proxy to queue requests to the API. It intercepts some requests to the API and adds them to an AWS SQS queue, then the webapp can be setup to process the queue.
-
-## Current features
-
-- Intercepts `sendEvent` requests and adds them to an AWS SQS queue. The webapp then reads from the queue and creates the events.
-
-## Setup
-
-### Create an AWS SQS queue
-
-In AWS you should create a new AWS SQS queue with appropriate security settings. You will need the queue URL for the next step.
-
-### Environment variables
-
-#### Cloudflare secrets
-
-Locally you should copy the `.dev.var.example` file to `.dev.var` and fill in the values.
-
-When deploying you should use `wrangler` (the Cloudflare CLI tool) to set secrets. Make sure you set the correct --env ("staging" or "prod")
-
-```bash
-wrangler secret put REWRITE_HOSTNAME --env staging
-wrangler secret put AWS_SQS_ACCESS_KEY_ID --env staging
-wrangler secret put AWS_SQS_SECRET_ACCESS_KEY --env staging
-wrangler secret put AWS_SQS_QUEUE_URL --env staging
-wrangler secret put AWS_SQS_REGION --env staging
-```
-
-You need to set your API CNAME entry to be proxied by Cloudflare. You can do this in the Cloudflare dashboard.
-
-#### Webapp
-
-These env vars also need setting in the webapp.
-
-```bash
-AWS_SQS_REGION
-AWS_SQS_ACCESS_KEY_ID
-AWS_SQS_SECRET_ACCESS_KEY
-AWS_SQS_QUEUE_URL
-AWS_SQS_BATCH_SIZE
-```
-
-## Deployment
-
-Staging:
-
-```bash
-npx wrangler@latest deploy --route "<your-api-subdomain>/*" --env staging
-```
-
-Prod:
-
-```bash
-npx wrangler@latest deploy --route "<your-api-subdomain>/*" --env prod
-```
-
-## Development
-
-Set the environment variables as described above.
-
-1. `pnpm install`
-2. `pnpm run dev --filter proxy`

apps/proxy/package.json

@@ -1,21 +0,0 @@
-{
-  "name": "proxy",
-  "version": "0.0.11",
-  "private": true,
-  "scripts": {
-    "deploy": "wrangler deploy",
-    "dev": "wrangler dev"
-  },
-  "devDependencies": {
-    "@cloudflare/workers-types": "^4.20240512.0",
-    "typescript": "^5.0.4",
-    "wrangler": "^3.57.1"
-  },
-  "dependencies": {
-    "@aws-sdk/client-sqs": "^3.445.0",
-    "@trigger.dev/core": "workspace:*",
-    "ulidx": "^2.2.1",
-    "zod": "3.22.3",
-    "zod-error": "1.5.0"
-  }
-}

apps/proxy/src/apikey.ts

@@ -1,20 +0,0 @@
-import { z } from "zod";
-
-const AuthorizationHeaderSchema = z.string().regex(/^Bearer .+$/);
-
-export function getApiKeyFromRequest(request: Request) {
-  const rawAuthorization = request.headers.get("Authorization");
-
-  const authorization = AuthorizationHeaderSchema.safeParse(rawAuthorization);
-  if (!authorization.success) {
-    return;
-  }
-
-  const apiKey = authorization.data.replace(/^Bearer /, "");
-  const type = isPrivateApiKey(apiKey) ? ("PRIVATE" as const) : ("PUBLIC" as const);
-  return { apiKey, type };
-}
-
-function isPrivateApiKey(key: string) {
-  return key.startsWith("tr_");
-}

apps/proxy/src/events/queueEvent.ts

@@ -1,87 +0,0 @@
-import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
-import { ApiEventLog, SendEventBodySchema } from "@trigger.dev/core";
-import { generateErrorMessage } from "zod-error";
-import { Env } from "..";
-import { getApiKeyFromRequest } from "../apikey";
-import { json } from "../json";
-import { calculateDeliverAt } from "./utils";
-
-/** Adds the event to an AWS SQS queue, so it can be consumed from the main Trigger.dev API */
-export async function queueEvent(request: Request, env: Env): Promise<Response> {
-  //check there's a private API key
-  const apiKeyResult = getApiKeyFromRequest(request);
-  if (!apiKeyResult || apiKeyResult.type !== "PRIVATE") {
-    return json(
-      { error: "Invalid or Missing API key" },
-      {
-        status: 401,
-      }
-    );
-  }
-
-  //parse the request body
-  try {
-    const anyBody = await request.json();
-    const body = SendEventBodySchema.safeParse(anyBody);
-    if (!body.success) {
-      return json(
-        { error: generateErrorMessage(body.error.issues) },
-        {
-          status: 422,
-        }
-      );
-    }
-
-    // The AWS SDK tries to use crypto from off of the window,
-    // so we need to trick it into finding it where it expects it
-    globalThis.global = globalThis;
-
-    const client = new SQSClient({
-      region: env.AWS_SQS_REGION,
-      credentials: {
-        accessKeyId: env.AWS_SQS_ACCESS_KEY_ID,
-        secretAccessKey: env.AWS_SQS_SECRET_ACCESS_KEY,
-      },
-    });
-
-    const timestamp = body.data.event.timestamp ?? new Date();
-
-    //add the event to the queue
-    const send = new SendMessageCommand({
-      // use wrangler secrets to provide this global variable
-      QueueUrl: env.AWS_SQS_QUEUE_URL,
-      MessageBody: JSON.stringify({
-        event: { ...body.data.event, timestamp },
-        options: body.data.options,
-        apiKey: apiKeyResult.apiKey,
-      }),
-    });
-
-    const queuedEvent = await client.send(send);
-    console.log("Queued event", queuedEvent);
-
-    //respond with the event
-    const event: ApiEventLog = {
-      id: body.data.event.id,
-      name: body.data.event.name,
-      payload: body.data.event.payload,
-      context: body.data.event.context,
-      timestamp,
-      deliverAt: calculateDeliverAt(body.data.options),
-    };
-
-    return json(event, {
-      status: 200,
-    });
-  } catch (e) {
-    console.error("queueEvent error", e);
-    return json(
-      {
-        error: `Failed to send event: ${e instanceof Error ? e.message : JSON.stringify(e)}`,
-      },
-      {
-        status: 422,
-      }
-    );
-  }
-}

apps/proxy/src/events/queueEvents.ts

@@ -1,112 +0,0 @@
-import { SQSClient, SendMessageBatchCommand } from "@aws-sdk/client-sqs";
-import { ApiEventLog, SendBulkEventsBodySchema } from "@trigger.dev/core";
-import { generateErrorMessage } from "zod-error";
-import { Env } from "..";
-import { getApiKeyFromRequest } from "../apikey";
-import { json } from "../json";
-import { calculateDeliverAt } from "./utils";
-
-/** Adds the event to an AWS SQS queue, so it can be consumed from the main Trigger.dev API */
-export async function queueEvents(request: Request, env: Env): Promise<Response> {
-  //check there's a private API key
-  const apiKeyResult = getApiKeyFromRequest(request);
-  if (!apiKeyResult || apiKeyResult.type !== "PRIVATE") {
-    return json(
-      { error: "Invalid or Missing API key" },
-      {
-        status: 401,
-      }
-    );
-  }
-
-  //parse the request body
-  try {
-    const anyBody = await request.json();
-    const body = SendBulkEventsBodySchema.safeParse(anyBody);
-    if (!body.success) {
-      return json(
-        { error: generateErrorMessage(body.error.issues) },
-        {
-          status: 422,
-        }
-      );
-    }
-
-    // The AWS SDK tries to use crypto from off of the window,
-    // so we need to trick it into finding it where it expects it
-    globalThis.global = globalThis;
-
-    const client = new SQSClient({
-      region: env.AWS_SQS_REGION,
-      credentials: {
-        accessKeyId: env.AWS_SQS_ACCESS_KEY_ID,
-        secretAccessKey: env.AWS_SQS_SECRET_ACCESS_KEY,
-      },
-    });
-
-    const updatedEvents: ApiEventLog[] = body.data.events.map((event) => {
-      const timestamp = event.timestamp ?? new Date();
-      return {
-        ...event,
-        payload: event.payload,
-        timestamp,
-      };
-    });
-
-    //divide updatedEvents into multiple batches of 10 (max size SQS accepts)
-    const batches: ApiEventLog[][] = [];
-    let currentBatch: ApiEventLog[] = [];
-    for (let i = 0; i < updatedEvents.length; i++) {
-      currentBatch.push(updatedEvents[i]);
-      if (currentBatch.length === 10) {
-        batches.push(currentBatch);
-        currentBatch = [];
-      }
-    }
-    if (currentBatch.length > 0) {
-      batches.push(currentBatch);
-    }
-
-    //loop through the batches and send them
-    for (let i = 0; i < batches.length; i++) {
-      const batch = batches[i];
-      //add the event to the queue
-      const send = new SendMessageBatchCommand({
-        // use wrangler secrets to provide this global variable
-        QueueUrl: env.AWS_SQS_QUEUE_URL,
-        Entries: batch.map((event, index) => ({
-          Id: `event-${index}`,
-          MessageBody: JSON.stringify({
-            event,
-            options: body.data.options,
-            apiKey: apiKeyResult.apiKey,
-          }),
-        })),
-      });
-
-      const queuedEvent = await client.send(send);
-      console.log("Queued events", queuedEvent);
-    }
-
-    //respond with the events
-    const events: ApiEventLog[] = updatedEvents.map((event) => ({
-      ...event,
-      payload: event.payload,
-      deliverAt: calculateDeliverAt(body.data.options),
-    }));
-
-    return json(events, {
-      status: 200,
-    });
-  } catch (e) {
-    console.error("queueEvents error", e);
-    return json(
-      {
-        error: `Failed to send events: ${e instanceof Error ? e.message : JSON.stringify(e)}`,
-      },
-      {
-        status: 422,
-      }
-    );
-  }
-}

apps/proxy/src/events/utils.ts

@@ -1,15 +0,0 @@
-import { SendEventOptions } from "@trigger.dev/core";
-
-export function calculateDeliverAt(options?: SendEventOptions) {
-  // If deliverAt is a string and a valid date, convert it to a Date object
-  if (options?.deliverAt) {
-    return options?.deliverAt;
-  }
-
-  // deliverAfter is the number of seconds to wait before delivering the event
-  if (options?.deliverAfter) {
-    return new Date(Date.now() + options.deliverAfter * 1000);
-  }
-
-  return undefined;
-}

apps/proxy/src/index.ts

@@ -1,71 +0,0 @@
-import { queueEvent } from "./events/queueEvent";
-import { queueEvents } from "./events/queueEvents";
-import { applyRateLimit } from "./rateLimit";
-import { Ratelimit } from "./rateLimiter";
-
-export interface Env {
-  /** The hostname needs to be changed to allow requests to pass to the Trigger.dev platform */
-  REWRITE_HOSTNAME: string;
-  REWRITE_PORT?: string;
-  AWS_SQS_ACCESS_KEY_ID: string;
-  AWS_SQS_SECRET_ACCESS_KEY: string;
-  AWS_SQS_QUEUE_URL: string;
-  AWS_SQS_REGION: string;
-  //rate limiter
-  API_RATE_LIMITER: Ratelimit;
-}
-
-export default {
-  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
-    if (!env.REWRITE_HOSTNAME) throw new Error("Missing REWRITE_HOSTNAME");
-    console.log("url", request.url);
-
-    if (!queueingIsEnabled(env)) {
-      console.log("Missing AWS credentials. Passing through to the origin.");
-      return redirectToOrigin(request, env);
-    }
-
-    const url = new URL(request.url);
-    switch (url.pathname) {
-      case "/api/v1/events": {
-        if (request.method === "POST") {
-          return applyRateLimit(request, env, () => queueEvent(request, env));
-        }
-        break;
-      }
-      case "/api/v1/events/bulk": {
-        if (request.method === "POST") {
-          return applyRateLimit(request, env, () => queueEvents(request, env));
-        }
-        break;
-      }
-    }
-
-    //the same request but with the hostname (and port) changed
-    return redirectToOrigin(request, env);
-  },
-};
-
-function redirectToOrigin(request: Request, env: Env) {
-  const newUrl = new URL(request.url);
-  newUrl.hostname = env.REWRITE_HOSTNAME;
-  newUrl.port = env.REWRITE_PORT || newUrl.port;
-
-  const requestInit: RequestInit = {
-    method: request.method,
-    headers: request.headers,
-    body: request.body,
-  };
-
-  console.log("rewritten url", newUrl.toString());
-  return fetch(newUrl.toString(), requestInit);
-}
-
-function queueingIsEnabled(env: Env) {
-  return (
-    env.AWS_SQS_ACCESS_KEY_ID &&
-    env.AWS_SQS_SECRET_ACCESS_KEY &&
-    env.AWS_SQS_QUEUE_URL &&
-    env.AWS_SQS_REGION
-  );
-}

apps/proxy/src/json.ts

@@ -1,13 +0,0 @@
-export function json(body: any, init?: ResponseInit) {
-  const headers = {
-    "content-type": "application/json",
-    ...(init?.headers ?? {}),
-  };
-
-  const responseInit: ResponseInit = {
-    ...(init ?? {}),
-    headers,
-  };
-
-  return new Response(JSON.stringify(body), responseInit);
-}

apps/proxy/src/rateLimit.ts

@@ -1,46 +0,0 @@
-import { Env } from "src";
-import { getApiKeyFromRequest } from "./apikey";
-import { json } from "./json";
-
-export async function applyRateLimit(
-  request: Request,
-  env: Env,
-  fn: () => Promise<Response>
-): Promise<Response> {
-  const apiKey = getApiKeyFromRequest(request);
-  if (apiKey) {
-    const result = await env.API_RATE_LIMITER.limit({ key: `apikey-${apiKey.apiKey}` });
-    const { success } = result;
-    console.log(`Rate limiter`, {
-      success,
-      key: `${apiKey.apiKey.substring(0, 12)}...`,
-    });
-    if (!success) {
-      //60s in the future
-      const reset = Date.now() + 60 * 1000;
-      const secondsUntilReset = Math.max(0, (reset - new Date().getTime()) / 1000);
-
-      return json(
-        {
-          title: "Rate Limit Exceeded",
-          status: 429,
-          type: "https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/429",
-          detail: `Rate limit exceeded. Retry in ${secondsUntilReset} seconds.`,
-          error: `Rate limit exceeded. Retry in ${secondsUntilReset} seconds.`,
-          reset,
-        },
-        {
-          status: 429,
-          headers: {
-            "x-ratelimit-reset": reset.toString(),
-          },
-        }
-      );
-    }
-  } else {
-    console.log(`Rate limiter: no API key for request`);
-  }
-
-  //call the original function
-  return fn();
-}

apps/proxy/src/rateLimiter.ts

@@ -1,23 +0,0 @@
-export interface Ratelimit {
-  /*
-   * The ratelimit function
-   * @param {RatelimitOptions} options
-   * @returns {Promise<RatelimitResponse>}
-   */
-  limit: (options: RatelimitOptions) => Promise<RatelimitResponse>;
-}
-
-export interface RatelimitOptions {
-  /*
-   * The key to identify the user, can be an IP address, user ID, etc.
-   */
-  key: string;
-}
-
-export interface RatelimitResponse {
-  /*
-   * The ratelimit success status
-   * @returns {boolean}
-   */
-  success: boolean;
-}

apps/proxy/tsconfig.json

@@ -1,35 +0,0 @@
-{
-  "compilerOptions": {
-    "target": "es2021" /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */,
-    "lib": [
-      "es2021"
-    ] /* Specify a set of bundled library declaration files that describe the target runtime environment. */,
-    "jsx": "react" /* Specify what JSX code is generated. */,
-
-    "module": "es2022" /* Specify what module code is generated. */,
-    "moduleResolution": "node" /* Specify how TypeScript looks up a file from a given module specifier. */,
-
-    "types": [
-      "@cloudflare/workers-types"
-    ] /* Specify type package names to be included without being referenced in a source file. */,
-    "resolveJsonModule": true /* Enable importing .json files */,
-
-    "allowJs": true /* Allow JavaScript files to be a part of your program. Use the `checkJS` option to get errors from these files. */,
-    "checkJs": false /* Enable error reporting in type-checked JavaScript files. */,
-
-    "noEmit": true /* Disable emitting files from a compilation. */,
-
-    "isolatedModules": true /* Ensure that each file can be safely transpiled without relying on other imports. */,
-    "allowSyntheticDefaultImports": true /* Allow 'import x from y' when a module doesn't have a default export. */,
-    "forceConsistentCasingInFileNames": true /* Ensure that casing is correct in imports. */,
-
-    "strict": true /* Enable all strict type-checking options. */,
-
-    "skipLibCheck": true /* Skip type checking all .d.ts files. */,
-    "baseUrl": ".",
-    "paths": {
-      "@trigger.dev/core": ["../../packages/core/src/index"],
-      "@trigger.dev/core/*": ["../../packages/core/src/*"]
-    }
-  }
-}

apps/proxy/wrangler.toml

@@ -1,33 +0,0 @@
-name = "proxy"
-main = "src/index.ts"
-compatibility_date = "2024-05-13"
-compatibility_flags = [ "nodejs_compat" ]
-
-[env.staging]
-  # The rate limiting API is in open beta.
-  [[env.staging.unsafe.bindings]]
-  name = "API_RATE_LIMITER"
-  type = "ratelimit"
-  # An identifier you define, that is unique to your Cloudflare account.
-  # Must be an integer.
-  namespace_id = "1"
-
-  # Limit: the number of tokens allowed within a given period in a single
-  # Cloudflare location
-  # Period: the duration of the period, in seconds. Must be either 10 or 60
-  simple = { limit = 100, period = 60 }
-
-
-[env.prod]
-  # The rate limiting API is in open beta.
-  [[env.prod.unsafe.bindings]]
-  name = "API_RATE_LIMITER"
-  type = "ratelimit"
-  # An identifier you define, that is unique to your Cloudflare account.
-  # Must be an integer.
-  namespace_id = "2"
-
-  # Limit: the number of tokens allowed within a given period in a single
-  # Cloudflare location
-  # Period: the duration of the period, in seconds. Must be either 10 or 60
-  simple = { limit = 300, period = 60 }

apps/supervisor/.env.example

@@ -0,0 +1,18 @@
+# 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
+
+# Point this at the webapp in prod
+TRIGGER_API_URL=http://localhost:3030
+
+# Point this at the OTel collector in prod
+OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:3030/otel
+# Use this on macOS
+# OTEL_EXPORTER_OTLP_ENDPOINT=http://host.docker.internal:3030/otel
+
+# Optional settings
+DEBUG=1
+ENFORCE_MACHINE_PRESETS=1
+TRIGGER_DEQUEUE_INTERVAL_MS=1000

apps/supervisor/.nvmrc

@@ -0,0 +1 @@
+v22.12.0

apps/supervisor/Containerfile

@@ -0,0 +1,53 @@
+FROM node:22-alpine@sha256:9bef0ef1e268f60627da9ba7d7605e8831d5b56ad07487d24d1aa386336d1944 AS node-22-alpine
+
+WORKDIR /app
+
+FROM node-22-alpine AS pruner
+
+COPY --chown=node:node . .
+RUN npx -q turbo@1.10.9 prune --scope=supervisor --docker
+
+FROM node-22-alpine AS base
+
+RUN apk add --no-cache dumb-init
+
+COPY --chown=node:node .gitignore .gitignore
+COPY --from=pruner --chown=node:node /app/out/json/ .
+COPY --from=pruner --chown=node:node /app/out/pnpm-lock.yaml ./pnpm-lock.yaml
+COPY --from=pruner --chown=node:node /app/out/pnpm-workspace.yaml ./pnpm-workspace.yaml
+
+RUN corepack enable && corepack prepare --activate
+
+FROM base AS deps-fetcher
+RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm fetch --frozen-lockfile
+
+FROM deps-fetcher AS dev-deps
+ENV NODE_ENV development
+
+RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm install --frozen-lockfile --offline --ignore-scripts
+
+FROM base AS builder
+
+COPY --from=pruner --chown=node:node /app/out/full/ .
+COPY --from=dev-deps --chown=node:node /app/ .
+COPY --chown=node:node turbo.json turbo.json
+COPY --chown=node:node .configs/tsconfig.base.json .configs/tsconfig.base.json
+COPY --chown=node:node scripts/updateVersion.ts scripts/updateVersion.ts
+
+RUN pnpm run generate && \
+  pnpm run --filter supervisor... build&& \
+  pnpm deploy --filter=supervisor --prod /prod/supervisor
+
+FROM base AS runner
+
+ENV NODE_ENV production
+
+COPY --from=builder /prod/supervisor /app/apps/supervisor
+
+EXPOSE 8000
+USER node
+
+# ensure pnpm is installed during build and not silently downloaded at runtime
+RUN pnpm -v
+
+CMD [ "/usr/bin/dumb-init", "--", "pnpm", "run", "--filter", "supervisor", "start"]

apps/supervisor/README.md

@@ -0,0 +1,67 @@
+# Supervisor
+
+## Dev setup
+
+1. Create a worker group
+
+```sh
+api_url=http://localhost:3030
+wg_name=my-worker
+
+# edit this
+admin_pat=tr_pat_...
+
+curl -sS \
+    -X POST \
+    "$api_url/admin/api/v1/workers" \
+    -H "Authorization: Bearer $admin_pat" \
+    -H "Content-Type: application/json" \
+    -d "{\"name\": \"$wg_name\"}"
+```
+
+2. Create `.env` and set the worker token
+
+```sh
+cp .env.example .env
+
+# Then edit your .env and set this to the token.plaintext value
+TRIGGER_WORKER_TOKEN=tr_wgt_...
+```
+
+3. Start the supervisor
+
+```sh
+pnpm dev
+```
+
+4. Build CLI, then deploy a reference project
+
+```sh
+pnpm exec trigger deploy --self-hosted
+
+# The additional network flag is required on linux
+pnpm exec trigger deploy --self-hosted --network host
+```
+
+## Additional worker groups
+
+When adding more worker groups you might also want to make them the default for a specific project. This will allow you to test it without having to change the global default:
+
+```sh
+api_url=http://localhost:3030
+wg_name=my-worker
+
+# edit these
+admin_pat=tr_pat_...
+project_id=clsw6q8wz...
+
+curl -sS \
+    -X POST \
+    "$api_url/admin/api/v1/workers" \
+    -H "Authorization: Bearer $admin_pat" \
+    -H "Content-Type: application/json" \
+    -d "{
+        \"name\": \"$wg_name\",
+        \"makeDefaultForProjectId\": \"$project_id\"
+    }"
+```

apps/supervisor/package.json

@@ -0,0 +1,31 @@
+{
+  "name": "supervisor",
+  "private": true,
+  "version": "0.0.1",
+  "main": "dist/index.js",
+  "type": "module",
+  "scripts": {
+    "build": "tsc",
+    "dev": "tsx --require dotenv/config --watch src/index.ts || (echo '!! Remember to run: nvm use'; exit 1)",
+    "start": "node dist/index.js",
+    "test:run": "vitest --no-file-parallelism --run",
+    "test:watch": "vitest --no-file-parallelism",
+    "typecheck": "tsc --noEmit"
+  },
+  "dependencies": {
+    "@kubernetes/client-node": "^1.0.0",
+    "@trigger.dev/core": "workspace:*",
+    "dockerode": "^4.0.3",
+    "nanoid": "^5.0.9",
+    "prom-client": "^15.1.0",
+    "socket.io": "4.7.4",
+    "std-env": "^3.8.0",
+    "tinyexec": "^0.3.1",
+    "zod": "3.23.8"
+  },
+  "devDependencies": {
+    "@types/dockerode": "^3.3.33",
+    "docker-api-ts": "^0.2.2",
+    "vitest": "^1.4.0"
+  }
+}

apps/supervisor/src/clients/kubernetes.ts

@@ -0,0 +1,52 @@
+import * as k8s from "@kubernetes/client-node";
+import { Informer } from "@kubernetes/client-node";
+import { ListPromise } from "@kubernetes/client-node";
+import { KubernetesObject } from "@kubernetes/client-node";
+import { assertExhaustive } from "@trigger.dev/core/utils";
+
+export const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local";
+
+export function createK8sApi() {
+  const kubeConfig = getKubeConfig();
+
+  function makeInformer<T extends KubernetesObject>(
+    path: string,
+    listPromiseFn: ListPromise<T>,
+    labelSelector?: string,
+    fieldSelector?: string
+  ): Informer<T> {
+    return k8s.makeInformer(kubeConfig, path, listPromiseFn, labelSelector, fieldSelector);
+  }
+
+  const api = {
+    core: kubeConfig.makeApiClient(k8s.CoreV1Api),
+    batch: kubeConfig.makeApiClient(k8s.BatchV1Api),
+    apps: kubeConfig.makeApiClient(k8s.AppsV1Api),
+    makeInformer,
+  };
+
+  return api;
+}
+
+export type K8sApi = ReturnType<typeof createK8sApi>;
+
+function getKubeConfig() {
+  console.log("getKubeConfig()", { RUNTIME_ENV });
+
+  const kubeConfig = new k8s.KubeConfig();
+
+  switch (RUNTIME_ENV) {
+    case "local":
+      kubeConfig.loadFromDefault();
+      break;
+    case "kubernetes":
+      kubeConfig.loadFromCluster();
+      break;
+    default:
+      assertExhaustive(RUNTIME_ENV);
+  }
+
+  return kubeConfig;
+}
+
+export { k8s };

apps/supervisor/src/env.ts

@@ -0,0 +1,77 @@
+import { randomUUID } from "crypto";
+import { env as stdEnv } from "std-env";
+import { z } from "zod";
+import { AdditionalEnvVars, BoolEnv } from "./envUtil.js";
+
+const Env = z.object({
+  // This will come from `spec.nodeName` in k8s
+  TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
+
+  // Required settings
+  TRIGGER_API_URL: z.string().url(),
+  TRIGGER_WORKER_TOKEN: z.string(),
+  MANAGED_WORKER_SECRET: z.string(),
+
+  // Workload API settings (coordinator mode) - the workload API is what the run controller connects to
+  TRIGGER_WORKLOAD_API_ENABLED: BoolEnv.default("true"),
+  TRIGGER_WORKLOAD_API_PROTOCOL: z
+    .string()
+    .transform((s) => z.enum(["http", "https"]).parse(s.toLowerCase()))
+    .default("http"),
+  TRIGGER_WORKLOAD_API_DOMAIN: z.string().optional(), // If unset, will use orchestrator-specific default
+  TRIGGER_WORKLOAD_API_HOST_INTERNAL: z.string().default("0.0.0.0"),
+  TRIGGER_WORKLOAD_API_PORT_INTERNAL: z.coerce.number().default(8020), // This is the port the workload API listens on
+  TRIGGER_WORKLOAD_API_PORT_EXTERNAL: z.coerce.number().default(8020), // This is the exposed port passed to the run controller
+
+  // Runner settings
+  RUNNER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().optional(),
+  RUNNER_SNAPSHOT_POLL_INTERVAL_SECONDS: z.coerce.number().optional(),
+  RUNNER_ADDITIONAL_ENV_VARS: AdditionalEnvVars, // optional (csv)
+
+  // Dequeue settings (provider mode)
+  TRIGGER_DEQUEUE_ENABLED: BoolEnv.default("true"),
+  TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(1000),
+  TRIGGER_DEQUEUE_MAX_RUN_COUNT: z.coerce.number().int().default(10),
+
+  // Optional services
+  TRIGGER_WARM_START_URL: z.string().optional(),
+  TRIGGER_CHECKPOINT_URL: z.string().optional(),
+  TRIGGER_METADATA_URL: z.string().optional(),
+
+  // Used by the workload manager, e.g docker/k8s
+  DOCKER_NETWORK: z.string().default("host"),
+  OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url(),
+  ENFORCE_MACHINE_PRESETS: z.coerce.boolean().default(false),
+  KUBERNETES_IMAGE_PULL_SECRETS: z.string().optional(), // csv
+
+  // Used by the resource monitor
+  OVERRIDE_CPU_TOTAL: z.coerce.number().optional(),
+  OVERRIDE_MEMORY_TOTAL_GB: z.coerce.number().optional(),
+
+  // Kubernetes specific settings
+  KUBERNETES_FORCE_ENABLED: BoolEnv.default(false),
+  KUBERNETES_NAMESPACE: z.string().default("default"),
+  KUBERNETES_WORKER_NODETYPE_LABEL: z.string().default("v4-worker"),
+  EPHEMERAL_STORAGE_SIZE_LIMIT: z.string().default("10Gi"),
+  EPHEMERAL_STORAGE_SIZE_REQUEST: z.string().default("2Gi"),
+
+  // Metrics
+  METRICS_ENABLED: BoolEnv.default(true),
+  METRICS_COLLECT_DEFAULTS: BoolEnv.default(true),
+  METRICS_HOST: z.string().default("127.0.0.1"),
+  METRICS_PORT: z.coerce.number().int().default(9090),
+
+  // Pod cleaner
+  POD_CLEANER_ENABLED: BoolEnv.default(true),
+  POD_CLEANER_INTERVAL_MS: z.coerce.number().int().default(10000),
+  POD_CLEANER_BATCH_SIZE: z.coerce.number().int().default(500),
+
+  // Failed pod handler
+  FAILED_POD_HANDLER_ENABLED: BoolEnv.default(true),
+  FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS: z.coerce.number().int().default(1000),
+
+  // Debug
+  DEBUG: BoolEnv.default(false),
+});
+
+export const env = Env.parse(stdEnv);

apps/supervisor/src/envUtil.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect } from "vitest";
+import { BoolEnv, AdditionalEnvVars } from "./envUtil.js";
+
+describe("BoolEnv", () => {
+  it("should parse string 'true' as true", () => {
+    expect(BoolEnv.parse("true")).toBe(true);
+    expect(BoolEnv.parse("TRUE")).toBe(true);
+    expect(BoolEnv.parse("True")).toBe(true);
+  });
+
+  it("should parse string '1' as true", () => {
+    expect(BoolEnv.parse("1")).toBe(true);
+  });
+
+  it("should parse string 'false' as false", () => {
+    expect(BoolEnv.parse("false")).toBe(false);
+    expect(BoolEnv.parse("FALSE")).toBe(false);
+    expect(BoolEnv.parse("False")).toBe(false);
+  });
+
+  it("should handle whitespace", () => {
+    expect(BoolEnv.parse(" true ")).toBe(true);
+    expect(BoolEnv.parse(" 1 ")).toBe(true);
+  });
+
+  it("should pass through boolean values", () => {
+    expect(BoolEnv.parse(true)).toBe(true);
+    expect(BoolEnv.parse(false)).toBe(false);
+  });
+
+  it("should return false for invalid inputs", () => {
+    expect(BoolEnv.parse("invalid")).toBe(false);
+    expect(BoolEnv.parse("")).toBe(false);
+  });
+});
+
+describe("AdditionalEnvVars", () => {
+  it("should parse single key-value pair", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar")).toEqual({ FOO: "bar" });
+  });
+
+  it("should parse multiple key-value pairs", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar,BAZ=qux")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should handle whitespace", () => {
+    expect(AdditionalEnvVars.parse(" FOO = bar , BAZ = qux ")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should return undefined for empty string", () => {
+    expect(AdditionalEnvVars.parse("")).toBeUndefined();
+  });
+
+  it("should return undefined for invalid format", () => {
+    expect(AdditionalEnvVars.parse("invalid")).toBeUndefined();
+  });
+
+  it("should skip invalid pairs but include valid ones", () => {
+    expect(AdditionalEnvVars.parse("FOO=bar,INVALID,BAZ=qux")).toEqual({
+      FOO: "bar",
+      BAZ: "qux",
+    });
+  });
+
+  it("should pass through undefined", () => {
+    expect(AdditionalEnvVars.parse(undefined)).toBeUndefined();
+  });
+
+  it("should handle empty values", () => {
+    expect(AdditionalEnvVars.parse("FOO=,BAR=value")).toEqual({
+      BAR: "value",
+    });
+  });
+});

apps/supervisor/src/envUtil.ts

@@ -0,0 +1,39 @@
+import { z } from "zod";
+
+export const BoolEnv = z.preprocess((val) => {
+  if (typeof val !== "string") {
+    return val;
+  }
+
+  return ["true", "1"].includes(val.toLowerCase().trim());
+}, z.boolean());
+
+export const AdditionalEnvVars = z.preprocess((val) => {
+  if (typeof val !== "string") {
+    return val;
+  }
+
+  if (!val) {
+    return undefined;
+  }
+
+  try {
+    const result = val.split(",").reduce(
+      (acc, pair) => {
+        const [key, value] = pair.split("=");
+        if (!key || !value) {
+          return acc;
+        }
+        acc[key.trim()] = value.trim();
+        return acc;
+      },
+      {} as Record<string, string>
+    );
+
+    // Return undefined if no valid key-value pairs were found
+    return Object.keys(result).length === 0 ? undefined : result;
+  } catch (error) {
+    console.warn("Failed to parse additional env vars", { error, val });
+    return undefined;
+  }
+}, z.record(z.string(), z.string()).optional());

apps/supervisor/src/index.ts

@@ -0,0 +1,363 @@
+import { SupervisorSession } from "@trigger.dev/core/v3/workers";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import { env } from "./env.js";
+import { WorkloadServer } from "./workloadServer/index.js";
+import type { WorkloadManagerOptions, WorkloadManager } from "./workloadManager/types.js";
+import Docker from "dockerode";
+import { z } from "zod";
+import { type DequeuedMessage } from "@trigger.dev/core/v3";
+import {
+  DockerResourceMonitor,
+  KubernetesResourceMonitor,
+  type ResourceMonitor,
+} from "./resourceMonitor.js";
+import { KubernetesWorkloadManager } from "./workloadManager/kubernetes.js";
+import { DockerWorkloadManager } from "./workloadManager/docker.js";
+import {
+  HttpServer,
+  CheckpointClient,
+  isKubernetesEnvironment,
+} from "@trigger.dev/core/v3/serverOnly";
+import { createK8sApi } from "./clients/kubernetes.js";
+import { collectDefaultMetrics } from "prom-client";
+import { register } from "./metrics.js";
+import { PodCleaner } from "./services/podCleaner.js";
+import { FailedPodHandler } from "./services/failedPodHandler.js";
+
+if (env.METRICS_COLLECT_DEFAULTS) {
+  collectDefaultMetrics({ register });
+}
+
+class ManagedSupervisor {
+  private readonly workerSession: SupervisorSession;
+  private readonly metricsServer?: HttpServer;
+  private readonly workloadServer: WorkloadServer;
+  private readonly workloadManager: WorkloadManager;
+  private readonly logger = new SimpleStructuredLogger("managed-worker");
+  private readonly resourceMonitor: ResourceMonitor;
+  private readonly checkpointClient?: CheckpointClient;
+
+  private readonly podCleaner?: PodCleaner;
+  private readonly failedPodHandler?: FailedPodHandler;
+
+  private readonly isKubernetes = isKubernetesEnvironment(env.KUBERNETES_FORCE_ENABLED);
+  private readonly warmStartUrl = env.TRIGGER_WARM_START_URL;
+
+  constructor() {
+    const { TRIGGER_WORKER_TOKEN, MANAGED_WORKER_SECRET, ...envWithoutSecrets } = env;
+
+    if (env.DEBUG) {
+      console.debug("[ManagedSupervisor] Starting up", { envWithoutSecrets });
+    }
+
+    if (this.warmStartUrl) {
+      this.logger.log("[ManagedWorker] 🔥 Warm starts enabled", {
+        warmStartUrl: this.warmStartUrl,
+      });
+    }
+
+    const workloadManagerOptions = {
+      workloadApiProtocol: env.TRIGGER_WORKLOAD_API_PROTOCOL,
+      workloadApiDomain: env.TRIGGER_WORKLOAD_API_DOMAIN,
+      workloadApiPort: env.TRIGGER_WORKLOAD_API_PORT_EXTERNAL,
+      warmStartUrl: this.warmStartUrl,
+      metadataUrl: env.TRIGGER_METADATA_URL,
+      imagePullSecrets: env.KUBERNETES_IMAGE_PULL_SECRETS?.split(","),
+      heartbeatIntervalSeconds: env.RUNNER_HEARTBEAT_INTERVAL_SECONDS,
+      snapshotPollIntervalSeconds: env.RUNNER_SNAPSHOT_POLL_INTERVAL_SECONDS,
+      additionalEnvVars: env.RUNNER_ADDITIONAL_ENV_VARS,
+    } satisfies WorkloadManagerOptions;
+
+    if (this.isKubernetes) {
+      if (env.POD_CLEANER_ENABLED) {
+        this.logger.log("[ManagedWorker] 🧹 Pod cleaner enabled", {
+          namespace: env.KUBERNETES_NAMESPACE,
+          batchSize: env.POD_CLEANER_BATCH_SIZE,
+          intervalMs: env.POD_CLEANER_INTERVAL_MS,
+        });
+        this.podCleaner = new PodCleaner({
+          register,
+          namespace: env.KUBERNETES_NAMESPACE,
+          batchSize: env.POD_CLEANER_BATCH_SIZE,
+          intervalMs: env.POD_CLEANER_INTERVAL_MS,
+        });
+      } else {
+        this.logger.warn("[ManagedWorker] Pod cleaner disabled");
+      }
+
+      if (env.FAILED_POD_HANDLER_ENABLED) {
+        this.logger.log("[ManagedWorker] 🔁 Failed pod handler enabled", {
+          namespace: env.KUBERNETES_NAMESPACE,
+          reconnectIntervalMs: env.FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS,
+        });
+        this.failedPodHandler = new FailedPodHandler({
+          register,
+          namespace: env.KUBERNETES_NAMESPACE,
+          reconnectIntervalMs: env.FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS,
+        });
+      } else {
+        this.logger.warn("[ManagedWorker] Failed pod handler disabled");
+      }
+
+      this.resourceMonitor = new KubernetesResourceMonitor(
+        createK8sApi(),
+        env.TRIGGER_WORKER_INSTANCE_NAME
+      );
+      this.workloadManager = new KubernetesWorkloadManager(workloadManagerOptions);
+    } else {
+      this.resourceMonitor = new DockerResourceMonitor(new Docker());
+      this.workloadManager = new DockerWorkloadManager(workloadManagerOptions);
+    }
+
+    this.workerSession = new SupervisorSession({
+      workerToken: env.TRIGGER_WORKER_TOKEN,
+      apiUrl: env.TRIGGER_API_URL,
+      instanceName: env.TRIGGER_WORKER_INSTANCE_NAME,
+      managedWorkerSecret: env.MANAGED_WORKER_SECRET,
+      dequeueIntervalMs: env.TRIGGER_DEQUEUE_INTERVAL_MS,
+      queueConsumerEnabled: env.TRIGGER_DEQUEUE_ENABLED,
+      maxRunCount: env.TRIGGER_DEQUEUE_MAX_RUN_COUNT,
+      runNotificationsEnabled: env.TRIGGER_WORKLOAD_API_ENABLED,
+      preDequeue: async () => {
+        if (this.isKubernetes) {
+          // Not used in k8s for now
+          return {};
+        }
+
+        const resources = await this.resourceMonitor.getNodeResources();
+        return {
+          maxResources: {
+            cpu: resources.cpuAvailable,
+            memory: resources.memoryAvailable,
+          },
+          skipDequeue: resources.cpuAvailable < 0.25 || resources.memoryAvailable < 0.25,
+        };
+      },
+      preSkip: async () => {
+        // When the node is full, it should still try to warm start runs
+        // await this.tryWarmStartAllThisNode();
+      },
+    });
+
+    if (env.TRIGGER_CHECKPOINT_URL) {
+      this.logger.log("[ManagedWorker] 🥶 Checkpoints enabled", {
+        checkpointUrl: env.TRIGGER_CHECKPOINT_URL,
+      });
+
+      this.checkpointClient = new CheckpointClient({
+        apiUrl: new URL(env.TRIGGER_CHECKPOINT_URL),
+        workerClient: this.workerSession.httpClient,
+        orchestrator: this.isKubernetes ? "KUBERNETES" : "DOCKER",
+      });
+    }
+
+    // setInterval(async () => {
+    //   const resources = await this.resourceMonitor.getNodeResources(true);
+    //   this.logger.debug("[ManagedWorker] Current resources", { resources });
+    // }, 1000);
+
+    this.workerSession.on("runNotification", async ({ time, run }) => {
+      this.logger.log("[ManagedWorker] runNotification", { time, run });
+
+      this.workloadServer.notifyRun({ run });
+    });
+
+    this.workerSession.on("runQueueMessage", async ({ time, message }) => {
+      this.logger.log(
+        `[ManagedWorker] Received message with timestamp ${time.toLocaleString()}`,
+        message
+      );
+
+      if (message.completedWaitpoints.length > 0) {
+        this.logger.debug("[ManagedWorker] Run has completed waitpoints", {
+          runId: message.run.id,
+          completedWaitpoints: message.completedWaitpoints.length,
+        });
+        // TODO: Do something with them or if we don't need the data here, maybe we shouldn't even send it
+      }
+
+      if (!message.image) {
+        this.logger.error("[ManagedWorker] Run has no image", { runId: message.run.id });
+        return;
+      }
+
+      const { checkpoint, ...rest } = message;
+
+      if (checkpoint) {
+        this.logger.log("[ManagedWorker] Restoring run", { runId: message.run.id });
+
+        if (!this.checkpointClient) {
+          this.logger.error("[ManagedWorker] No checkpoint client", { runId: message.run.id });
+          return;
+        }
+
+        try {
+          const didRestore = await this.checkpointClient.restoreRun({
+            runFriendlyId: message.run.friendlyId,
+            snapshotFriendlyId: message.snapshot.friendlyId,
+            body: {
+              ...rest,
+              checkpoint,
+            },
+          });
+
+          if (didRestore) {
+            this.logger.log("[ManagedWorker] Restore successful", { runId: message.run.id });
+          } else {
+            this.logger.error("[ManagedWorker] Restore failed", { runId: message.run.id });
+          }
+        } catch (error) {
+          this.logger.error("[ManagedWorker] Failed to restore run", { error });
+        }
+
+        return;
+      }
+
+      this.logger.log("[ManagedWorker] Scheduling run", { runId: message.run.id });
+
+      const didWarmStart = await this.tryWarmStart(message);
+
+      if (didWarmStart) {
+        this.logger.log("[ManagedWorker] Warm start successful", { runId: message.run.id });
+        return;
+      }
+
+      try {
+        await this.workloadManager.create({
+          dequeuedAt: message.dequeuedAt,
+          envId: message.environment.id,
+          envType: message.environment.type,
+          image: message.image,
+          machine: message.run.machine,
+          orgId: message.organization.id,
+          projectId: message.project.id,
+          runId: message.run.id,
+          runFriendlyId: message.run.friendlyId,
+          version: message.version,
+          nextAttemptNumber: message.run.attemptNumber,
+          snapshotId: message.snapshot.id,
+          snapshotFriendlyId: message.snapshot.friendlyId,
+        });
+
+        // Disabled for now
+        // this.resourceMonitor.blockResources({
+        //   cpu: message.run.machine.cpu,
+        //   memory: message.run.machine.memory,
+        // });
+      } catch (error) {
+        this.logger.error("[ManagedWorker] Failed to create workload", { error });
+      }
+    });
+
+    if (env.METRICS_ENABLED) {
+      this.metricsServer = new HttpServer({
+        port: env.METRICS_PORT,
+        host: env.METRICS_HOST,
+        metrics: {
+          register,
+          expose: true,
+        },
+      });
+    }
+
+    // Responds to workload requests only
+    this.workloadServer = new WorkloadServer({
+      port: env.TRIGGER_WORKLOAD_API_PORT_INTERNAL,
+      host: env.TRIGGER_WORKLOAD_API_HOST_INTERNAL,
+      workerClient: this.workerSession.httpClient,
+      checkpointClient: this.checkpointClient,
+    });
+
+    this.workloadServer.on("runConnected", this.onRunConnected.bind(this));
+    this.workloadServer.on("runDisconnected", this.onRunDisconnected.bind(this));
+  }
+
+  async onRunConnected({ run }: { run: { friendlyId: string } }) {
+    this.logger.debug("[ManagedWorker] Run connected", { run });
+    this.workerSession.subscribeToRunNotifications([run.friendlyId]);
+  }
+
+  async onRunDisconnected({ run }: { run: { friendlyId: string } }) {
+    this.logger.debug("[ManagedWorker] Run disconnected", { run });
+    this.workerSession.unsubscribeFromRunNotifications([run.friendlyId]);
+  }
+
+  private async tryWarmStart(dequeuedMessage: DequeuedMessage): Promise<boolean> {
+    if (!this.warmStartUrl) {
+      return false;
+    }
+
+    const warmStartUrlWithPath = new URL("/warm-start", this.warmStartUrl);
+
+    try {
+      const res = await fetch(warmStartUrlWithPath.href, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({ dequeuedMessage }),
+      });
+
+      if (!res.ok) {
+        this.logger.error("[ManagedWorker] Warm start failed", {
+          runId: dequeuedMessage.run.id,
+        });
+        return false;
+      }
+
+      const data = await res.json();
+      const parsedData = z.object({ didWarmStart: z.boolean() }).safeParse(data);
+
+      if (!parsedData.success) {
+        this.logger.error("[ManagedWorker] Warm start response invalid", {
+          runId: dequeuedMessage.run.id,
+          data,
+        });
+        return false;
+      }
+
+      return parsedData.data.didWarmStart;
+    } catch (error) {
+      this.logger.error("[ManagedWorker] Warm start error", {
+        runId: dequeuedMessage.run.id,
+        error,
+      });
+      return false;
+    }
+  }
+
+  async start() {
+    this.logger.log("[ManagedWorker] Starting up");
+
+    // Optional services
+    await this.podCleaner?.start();
+    await this.failedPodHandler?.start();
+    await this.metricsServer?.start();
+
+    if (env.TRIGGER_WORKLOAD_API_ENABLED) {
+      this.logger.log("[ManagedWorker] Workload API enabled", {
+        protocol: env.TRIGGER_WORKLOAD_API_PROTOCOL,
+        domain: env.TRIGGER_WORKLOAD_API_DOMAIN,
+        port: env.TRIGGER_WORKLOAD_API_PORT_INTERNAL,
+      });
+      await this.workloadServer.start();
+    } else {
+      this.logger.warn("[ManagedWorker] Workload API disabled");
+    }
+
+    await this.workerSession.start();
+  }
+
+  async stop() {
+    this.logger.log("[ManagedWorker] Shutting down");
+    await this.workerSession.stop();
+
+    // Optional services
+    await this.podCleaner?.stop();
+    await this.failedPodHandler?.stop();
+    await this.metricsServer?.stop();
+  }
+}
+
+const worker = new ManagedSupervisor();
+worker.start();

apps/supervisor/src/metrics.ts

@@ -0,0 +1,3 @@
+import { Registry } from "prom-client";
+
+export const register = new Registry();

apps/supervisor/src/resourceMonitor.ts

@@ -0,0 +1,246 @@
+import type Docker from "dockerode";
+import type * as TDocker from "docker-api-ts";
+import type { MachineResources } from "@trigger.dev/core/v3";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import { env } from "./env.js";
+import type { K8sApi } from "./clients/kubernetes.js";
+
+const logger = new SimpleStructuredLogger("resource-monitor");
+
+interface NodeResources {
+  cpuTotal: number; // in cores
+  cpuAvailable: number;
+  memoryTotal: number; // in bytes
+  memoryAvailable: number;
+}
+
+interface ResourceRequest {
+  cpu: number; // in cores
+  memory: number; // in bytes
+}
+
+export abstract class ResourceMonitor {
+  protected cacheTimeoutMs = 5_000;
+  protected lastUpdateMs = 0;
+
+  protected cachedResources: NodeResources = {
+    cpuTotal: 0,
+    cpuAvailable: 0,
+    memoryTotal: 0,
+    memoryAvailable: 0,
+  };
+
+  protected resourceParser: ResourceParser;
+
+  constructor(Parser: new () => ResourceParser) {
+    this.resourceParser = new Parser();
+  }
+
+  abstract getNodeResources(fromCache?: boolean): Promise<NodeResources>;
+
+  blockResources(resources: MachineResources): void {
+    const { cpu, memory } = this.toResourceRequest(resources);
+
+    logger.debug("[ResourceMonitor] Blocking resources", {
+      raw: resources,
+      converted: { cpu, memory },
+    });
+
+    this.cachedResources.cpuAvailable -= cpu;
+    this.cachedResources.memoryAvailable -= memory;
+  }
+
+  async wouldFit(request: ResourceRequest): Promise<boolean> {
+    const resources = await this.getNodeResources();
+    return resources.cpuAvailable >= request.cpu && resources.memoryAvailable >= request.memory;
+  }
+
+  private toResourceRequest(resources: MachineResources): ResourceRequest {
+    return {
+      cpu: resources.cpu ?? 0,
+      memory: this.gbToBytes(resources.memory ?? 0),
+    };
+  }
+
+  private gbToBytes(gb: number): number {
+    return gb * 1024 * 1024 * 1024;
+  }
+
+  protected isCacheValid(): boolean {
+    return this.cachedResources !== null && Date.now() - this.lastUpdateMs < this.cacheTimeoutMs;
+  }
+
+  protected applyOverrides(resources: NodeResources): NodeResources {
+    if (!env.OVERRIDE_CPU_TOTAL && !env.OVERRIDE_MEMORY_TOTAL_GB) {
+      return resources;
+    }
+
+    logger.debug("[ResourceMonitor] 🛡️ Applying resource overrides", {
+      cpuTotal: env.OVERRIDE_CPU_TOTAL,
+      memoryTotalGb: env.OVERRIDE_MEMORY_TOTAL_GB,
+    });
+
+    const cpuTotal = env.OVERRIDE_CPU_TOTAL ?? resources.cpuTotal;
+    const memoryTotal = env.OVERRIDE_MEMORY_TOTAL_GB
+      ? this.gbToBytes(env.OVERRIDE_MEMORY_TOTAL_GB)
+      : resources.memoryTotal;
+
+    const cpuDiff = cpuTotal - resources.cpuTotal;
+    const memoryDiff = memoryTotal - resources.memoryTotal;
+
+    const cpuAvailable = Math.max(0, resources.cpuAvailable + cpuDiff);
+    const memoryAvailable = Math.max(0, resources.memoryAvailable + memoryDiff);
+
+    return {
+      cpuTotal,
+      cpuAvailable,
+      memoryTotal,
+      memoryAvailable,
+    };
+  }
+}
+
+export class DockerResourceMonitor extends ResourceMonitor {
+  private docker: Docker;
+
+  constructor(docker: Docker) {
+    super(DockerResourceParser);
+    this.docker = docker;
+  }
+
+  async getNodeResources(fromCache?: boolean): Promise<NodeResources> {
+    if (this.isCacheValid() || fromCache) {
+      // logger.debug("[ResourceMonitor] Using cached resources");
+      return this.cachedResources;
+    }
+
+    const info: TDocker.SystemInfo = await this.docker.info();
+    const stats = await this.docker.listContainers({ all: true });
+
+    // Get system-wide resources
+    const cpuTotal = info.NCPU ?? 0;
+    const memoryTotal = info.MemTotal ?? 0;
+
+    // Calculate used resources from running containers
+    let cpuUsed = 0;
+    let memoryUsed = 0;
+
+    for (const container of stats) {
+      if (container.State === "running") {
+        const c = this.docker.getContainer(container.Id);
+        const { HostConfig } = await c.inspect();
+
+        const cpu = this.resourceParser.cpu(HostConfig.NanoCpus ?? 0);
+        const memory = this.resourceParser.memory(HostConfig.Memory ?? 0);
+
+        cpuUsed += cpu;
+        memoryUsed += memory;
+      }
+    }
+
+    this.cachedResources = this.applyOverrides({
+      cpuTotal,
+      cpuAvailable: cpuTotal - cpuUsed,
+      memoryTotal,
+      memoryAvailable: memoryTotal - memoryUsed,
+    });
+
+    this.lastUpdateMs = Date.now();
+
+    return this.cachedResources;
+  }
+}
+
+export class KubernetesResourceMonitor extends ResourceMonitor {
+  private k8s: K8sApi;
+  private nodeName: string;
+
+  constructor(k8s: K8sApi, nodeName: string) {
+    super(KubernetesResourceParser);
+    this.k8s = k8s;
+    this.nodeName = nodeName;
+  }
+
+  async getNodeResources(fromCache?: boolean): Promise<NodeResources> {
+    if (this.isCacheValid() || fromCache) {
+      logger.debug("[ResourceMonitor] Using cached resources");
+      return this.cachedResources;
+    }
+
+    const node = await this.k8s.core.readNode({ name: this.nodeName });
+    const pods = await this.k8s.core.listPodForAllNamespaces({
+      // TODO: ensure this includes all pods that consume resources
+      fieldSelector: `spec.nodeName=${this.nodeName},status.phase=Running`,
+    });
+
+    const allocatable = node.status?.allocatable;
+    const cpuTotal = this.resourceParser.cpu(allocatable?.cpu ?? "0");
+    const memoryTotal = this.resourceParser.memory(allocatable?.memory ?? "0");
+
+    // Sum up resources requested by all pods on this node
+    let cpuRequested = 0;
+    let memoryRequested = 0;
+
+    for (const pod of pods.items) {
+      if (pod.status?.phase === "Running") {
+        if (!pod.spec) {
+          continue;
+        }
+
+        for (const container of pod.spec.containers) {
+          const resources = container.resources?.requests ?? {};
+          cpuRequested += this.resourceParser.cpu(resources.cpu ?? "0");
+          memoryRequested += this.resourceParser.memory(resources.memory ?? "0");
+        }
+      }
+    }
+
+    this.cachedResources = this.applyOverrides({
+      cpuTotal,
+      cpuAvailable: cpuTotal - cpuRequested,
+      memoryTotal,
+      memoryAvailable: memoryTotal - memoryRequested,
+    });
+
+    this.lastUpdateMs = Date.now();
+
+    return this.cachedResources;
+  }
+}
+
+abstract class ResourceParser {
+  abstract cpu(cpu: number | string): number;
+  abstract memory(memory: number | string): number;
+}
+
+class DockerResourceParser extends ResourceParser {
+  cpu(cpu: number): number {
+    return cpu / 1e9;
+  }
+
+  memory(memory: number): number {
+    return memory;
+  }
+}
+
+class KubernetesResourceParser extends ResourceParser {
+  cpu(cpu: string): number {
+    if (cpu.endsWith("m")) {
+      return parseInt(cpu.slice(0, -1)) / 1000;
+    }
+    return parseInt(cpu);
+  }
+
+  memory(memory: string): number {
+    if (memory.endsWith("Ki")) {
+      return parseInt(memory.slice(0, -2)) * 1024;
+    }
+    if (memory.endsWith("Mi")) {
+      return parseInt(memory.slice(0, -2)) * 1024 * 1024;
+    }
+    if (memory.endsWith("Gi")) {
+      return parseInt(memory.slice(0, -2)) * 1024 * 1024 * 1024;
+    }
+    return parseInt(memory);
+  }
+}

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

@@ -0,0 +1,580 @@
+import { describe, it, expect, beforeAll, afterEach } from "vitest";
+import { FailedPodHandler } from "./failedPodHandler.js";
+import { K8sApi, createK8sApi } from "../clients/kubernetes.js";
+import { Registry } from "prom-client";
+import { setTimeout } from "timers/promises";
+
+describe("FailedPodHandler Integration Tests", () => {
+  const k8s = createK8sApi();
+  const namespace = "integration-test";
+  const register = new Registry();
+
+  beforeAll(async () => {
+    // Create the test namespace if it doesn't exist
+    try {
+      await k8s.core.readNamespace({ name: namespace });
+    } catch (error) {
+      await k8s.core.createNamespace({
+        body: {
+          metadata: {
+            name: namespace,
+          },
+        },
+      });
+    }
+
+    // Clear any existing pods in the namespace
+    await deleteAllPodsInNamespace({ k8sApi: k8s, namespace });
+  });
+
+  afterEach(async () => {
+    // Clear metrics to avoid conflicts
+    register.clear();
+
+    // Delete any remaining pods in the namespace
+    await deleteAllPodsInNamespace({ k8sApi: k8s, namespace });
+  });
+
+  it("should process and delete failed pods with app=task-run label", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Create failed pods with the correct label
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 2,
+        shouldFail: true,
+      });
+
+      // Wait for pods to reach Failed state
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Failed",
+      });
+
+      // Start the handler
+      await handler.start();
+
+      // Wait for pods to be deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      // Verify metrics
+      const metrics = handler.getMetrics();
+
+      // Check informer events were recorded
+      const informerEvents = await metrics.informerEventsTotal.get();
+      expect(informerEvents.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "add",
+          }),
+          value: 2,
+        })
+      );
+      expect(informerEvents.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "connect",
+          }),
+          value: 1,
+        })
+      );
+      expect(informerEvents.values).not.toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "error",
+          }),
+        })
+      );
+
+      // Check pods were processed
+      const processedPods = await metrics.processedPodsTotal.get();
+      expect(processedPods.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            status: "Failed",
+          }),
+          value: 2,
+        })
+      );
+
+      // Check pods were deleted
+      const deletedPods = await metrics.deletedPodsTotal.get();
+      expect(deletedPods.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            status: "Failed",
+          }),
+          value: 2,
+        })
+      );
+
+      // Check no deletion errors were recorded
+      const deletionErrors = await metrics.deletionErrorsTotal.get();
+      expect(deletionErrors.values).toHaveLength(0);
+
+      // Check processing durations were recorded
+      const durations = await metrics.processingDurationSeconds.get();
+      const failedDurations = durations.values.filter(
+        (v) => v.labels.namespace === namespace && v.labels.status === "Failed"
+      );
+      expect(failedDurations.length).toBeGreaterThan(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 30000);
+
+  it("should ignore pods without app=task-run label", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Create failed pods without the task-run label
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        shouldFail: true,
+        labels: { app: "not-task-run" },
+      });
+
+      // Wait for pod to reach Failed state
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Failed",
+      });
+
+      await handler.start();
+
+      // Wait a reasonable time to ensure pod isn't deleted
+      await setTimeout(5000);
+
+      // Verify pod still exists
+      const exists = await podExists({ k8sApi: k8s, namespace, podName: podNames[0]! });
+      expect(exists).toBe(true);
+
+      // Verify no metrics were recorded
+      const metrics = handler.getMetrics();
+      const processedPods = await metrics.processedPodsTotal.get();
+      expect(processedPods.values).toHaveLength(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 30000);
+
+  it("should not process pods that are being deleted", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Create a failed pod that we'll mark for deletion
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        shouldFail: true,
+        command: ["/bin/sh", "-c", "sleep 30"],
+      });
+
+      // Wait for pod to reach Failed state
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Running",
+      });
+
+      // Delete the pod but don't wait for deletion
+      await k8s.core.deleteNamespacedPod({
+        namespace,
+        name: podNames[0]!,
+        gracePeriodSeconds: 5,
+      });
+
+      // Start the handler
+      await handler.start();
+
+      // Wait for pod to be fully deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      // Verify metrics show we skipped processing
+      const metrics = handler.getMetrics();
+      const processedPods = await metrics.processedPodsTotal.get();
+      expect(processedPods.values).toHaveLength(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 30000);
+
+  it("should detect and process pods that fail after handler starts", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Start the handler
+      await handler.start();
+
+      // Create failed pods with the correct label
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 3,
+        shouldFail: true,
+      });
+
+      // Wait for pods to be deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      // Verify metrics
+      const metrics = handler.getMetrics();
+
+      // Check informer events were recorded
+      const informerEvents = await metrics.informerEventsTotal.get();
+      expect(informerEvents.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "add",
+          }),
+          value: 3,
+        })
+      );
+      expect(informerEvents.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "connect",
+          }),
+          value: 1,
+        })
+      );
+      expect(informerEvents.values).not.toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "error",
+          }),
+        })
+      );
+
+      // Check pods were processed
+      const processedPods = await metrics.processedPodsTotal.get();
+      expect(processedPods.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            status: "Failed",
+          }),
+          value: 3,
+        })
+      );
+
+      // Check pods were deleted
+      const deletedPods = await metrics.deletedPodsTotal.get();
+      expect(deletedPods.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            status: "Failed",
+          }),
+          value: 3,
+        })
+      );
+
+      // Check no deletion errors were recorded
+      const deletionErrors = await metrics.deletionErrorsTotal.get();
+      expect(deletionErrors.values).toHaveLength(0);
+
+      // Check processing durations were recorded
+      const durations = await metrics.processingDurationSeconds.get();
+      const failedDurations = durations.values.filter(
+        (v) => v.labels.namespace === namespace && v.labels.status === "Failed"
+      );
+      expect(failedDurations.length).toBeGreaterThan(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 60000);
+
+  it("should handle graceful shutdown pods differently", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Create first batch of pods before starting handler
+      const firstBatchPodNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 2,
+        exitCode: FailedPodHandler.GRACEFUL_SHUTDOWN_EXIT_CODE,
+      });
+
+      // Wait for pods to reach Failed state
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames: firstBatchPodNames,
+        phase: "Failed",
+      });
+
+      // Start the handler
+      await handler.start();
+
+      // Wait for first batch to be deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames: firstBatchPodNames,
+      });
+
+      // Create second batch of pods after handler is running
+      const secondBatchPodNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 3,
+        exitCode: FailedPodHandler.GRACEFUL_SHUTDOWN_EXIT_CODE,
+      });
+
+      // Wait for second batch to be deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames: secondBatchPodNames,
+      });
+
+      // Verify metrics
+      const metrics = handler.getMetrics();
+
+      // Check informer events were recorded for both batches
+      const informerEvents = await metrics.informerEventsTotal.get();
+      expect(informerEvents.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            verb: "add",
+          }),
+          value: 5, // 2 from first batch + 3 from second batch
+        })
+      );
+
+      // Check pods were processed as graceful shutdowns
+      const processedPods = await metrics.processedPodsTotal.get();
+
+      // Should not be marked as Failed
+      const failedPods = processedPods.values.find(
+        (v) => v.labels.namespace === namespace && v.labels.status === "Failed"
+      );
+      expect(failedPods).toBeUndefined();
+
+      // Should be marked as GracefulShutdown
+      const gracefulShutdowns = processedPods.values.find(
+        (v) => v.labels.namespace === namespace && v.labels.status === "GracefulShutdown"
+      );
+      expect(gracefulShutdowns).toBeDefined();
+      expect(gracefulShutdowns?.value).toBe(5); // Total from both batches
+
+      // Check pods were still deleted
+      const deletedPods = await metrics.deletedPodsTotal.get();
+      expect(deletedPods.values).toContainEqual(
+        expect.objectContaining({
+          labels: expect.objectContaining({
+            namespace,
+            status: "Failed",
+          }),
+          value: 5, // Total from both batches
+        })
+      );
+
+      // Check no deletion errors were recorded
+      const deletionErrors = await metrics.deletionErrorsTotal.get();
+      expect(deletionErrors.values).toHaveLength(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 30000);
+});
+
+async function createTestPods({
+  k8sApi,
+  namespace,
+  count,
+  labels = { app: "task-run" },
+  shouldFail = false,
+  namePrefix = "test-pod",
+  command = ["/bin/sh", "-c", shouldFail ? "exit 1" : "exit 0"],
+  randomizeName = true,
+  exitCode,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  count: number;
+  labels?: Record<string, string>;
+  shouldFail?: boolean;
+  namePrefix?: string;
+  command?: string[];
+  randomizeName?: boolean;
+  exitCode?: number;
+}) {
+  const createdPods: string[] = [];
+
+  // If exitCode is specified, override the command
+  if (exitCode !== undefined) {
+    command = ["/bin/sh", "-c", `exit ${exitCode}`];
+  }
+
+  for (let i = 0; i < count; i++) {
+    const podName = randomizeName
+      ? `${namePrefix}-${i}-${Math.random().toString(36).substring(2, 15)}`
+      : `${namePrefix}-${i}`;
+    await k8sApi.core.createNamespacedPod({
+      namespace,
+      body: {
+        metadata: {
+          name: podName,
+          labels,
+        },
+        spec: {
+          restartPolicy: "Never",
+          containers: [
+            {
+              name: "run-controller", // Changed to match the name we check in failedPodHandler
+              image: "busybox:1.37.0",
+              command,
+            },
+          ],
+        },
+      },
+    });
+    createdPods.push(podName);
+  }
+
+  return createdPods;
+}
+
+async function waitForPodsDeletion({
+  k8sApi,
+  namespace,
+  podNames,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podNames: string[];
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+  const pendingPods = new Set(podNames);
+
+  while (pendingPods.size > 0 && Date.now() - startTime < timeoutMs) {
+    const pods = await k8sApi.core.listNamespacedPod({ namespace });
+    const existingPods = new Set(pods.items.map((pod) => pod.metadata?.name ?? ""));
+
+    for (const podName of pendingPods) {
+      if (!existingPods.has(podName)) {
+        pendingPods.delete(podName);
+      }
+    }
+
+    if (pendingPods.size > 0) {
+      await setTimeout(waitMs);
+    }
+  }
+
+  if (pendingPods.size > 0) {
+    throw new Error(
+      `Pods [${Array.from(pendingPods).join(", ")}] were not deleted within ${timeoutMs}ms`
+    );
+  }
+}
+
+async function podExists({
+  k8sApi,
+  namespace,
+  podName,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podName: string;
+}) {
+  const pods = await k8sApi.core.listNamespacedPod({ namespace });
+  return pods.items.some((p) => p.metadata?.name === podName);
+}
+
+async function waitForPodsPhase({
+  k8sApi,
+  namespace,
+  podNames,
+  phase,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podNames: string[];
+  phase: "Pending" | "Running" | "Succeeded" | "Failed" | "Unknown";
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+  const pendingPods = new Set(podNames);
+
+  while (pendingPods.size > 0 && Date.now() - startTime < timeoutMs) {
+    const pods = await k8sApi.core.listNamespacedPod({ namespace });
+
+    for (const pod of pods.items) {
+      if (pendingPods.has(pod.metadata?.name ?? "") && pod.status?.phase === phase) {
+        pendingPods.delete(pod.metadata?.name ?? "");
+      }
+    }
+
+    if (pendingPods.size > 0) {
+      await setTimeout(waitMs);
+    }
+  }
+
+  if (pendingPods.size > 0) {
+    throw new Error(
+      `Pods [${Array.from(pendingPods).join(
+        ", "
+      )}] did not reach phase ${phase} within ${timeoutMs}ms`
+    );
+  }
+}
+
+async function deleteAllPodsInNamespace({
+  k8sApi,
+  namespace,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+}) {
+  // Get all pods
+  const pods = await k8sApi.core.listNamespacedPod({ namespace });
+  const podNames = pods.items.map((p) => p.metadata?.name ?? "");
+
+  // Delete all pods
+  await k8sApi.core.deleteCollectionNamespacedPod({ namespace, gracePeriodSeconds: 0 });
+
+  // Wait for all pods to be deleted
+  await waitForPodsDeletion({ k8sApi, namespace, podNames });
+}

apps/supervisor/src/services/failedPodHandler.ts

@@ -0,0 +1,298 @@
+import { LogLevel, SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import { K8sApi } from "../clients/kubernetes.js";
+import { createK8sApi } from "../clients/kubernetes.js";
+import { Informer, V1Pod } from "@kubernetes/client-node";
+import { Counter, Registry, Histogram } from "prom-client";
+import { register } from "../metrics.js";
+import { setTimeout } from "timers/promises";
+
+type PodStatus = "Pending" | "Running" | "Succeeded" | "Failed" | "Unknown" | "GracefulShutdown";
+
+export type FailedPodHandlerOptions = {
+  namespace: string;
+  reconnectIntervalMs?: number;
+  k8s?: K8sApi;
+  register?: Registry;
+};
+
+export class FailedPodHandler {
+  private readonly id: string;
+  private readonly logger: SimpleStructuredLogger;
+  private readonly k8s: K8sApi;
+  private readonly namespace: string;
+
+  private isRunning = false;
+
+  private readonly informer: Informer<V1Pod>;
+  private readonly reconnectIntervalMs: number;
+
+  // Metrics
+  private readonly register: Registry;
+  private readonly processedPodsTotal: Counter;
+  private readonly deletedPodsTotal: Counter;
+  private readonly deletionErrorsTotal: Counter;
+  private readonly processingDurationSeconds: Histogram<string>;
+  private readonly informerEventsTotal: Counter;
+
+  static readonly GRACEFUL_SHUTDOWN_EXIT_CODE = 200;
+
+  constructor(opts: FailedPodHandlerOptions) {
+    this.id = Math.random().toString(36).substring(2, 15);
+    this.logger = new SimpleStructuredLogger("failed-pod-handler", LogLevel.debug, {
+      id: this.id,
+    });
+
+    this.k8s = opts.k8s ?? createK8sApi();
+
+    this.namespace = opts.namespace;
+    this.reconnectIntervalMs = opts.reconnectIntervalMs ?? 1000;
+
+    this.informer = this.k8s.makeInformer(
+      `/api/v1/namespaces/${this.namespace}/pods`,
+      () =>
+        this.k8s.core.listNamespacedPod({
+          namespace: this.namespace,
+          labelSelector: "app=task-run",
+          fieldSelector: "status.phase=Failed",
+        }),
+      "app=task-run",
+      "status.phase=Failed"
+    );
+
+    // Whenever a matching pod is added to the informer cache
+    this.informer.on("add", this.onPodCompleted.bind(this));
+
+    // Informer events
+    this.informer.on("connect", this.makeOnConnect("failed-pod-informer").bind(this));
+    this.informer.on("error", this.makeOnError("failed-pod-informer").bind(this));
+
+    // Initialize metrics
+    this.register = opts.register ?? register;
+
+    this.processedPodsTotal = new Counter({
+      name: "failed_pod_handler_processed_pods_total",
+      help: "Total number of failed pods processed",
+      labelNames: ["namespace", "status"],
+      registers: [this.register],
+    });
+
+    this.deletedPodsTotal = new Counter({
+      name: "failed_pod_handler_deleted_pods_total",
+      help: "Total number of pods deleted",
+      labelNames: ["namespace", "status"],
+      registers: [this.register],
+    });
+
+    this.deletionErrorsTotal = new Counter({
+      name: "failed_pod_handler_deletion_errors_total",
+      help: "Total number of errors encountered while deleting pods",
+      labelNames: ["namespace", "error_type"],
+      registers: [this.register],
+    });
+
+    this.processingDurationSeconds = new Histogram({
+      name: "failed_pod_handler_processing_duration_seconds",
+      help: "The duration of pod processing",
+      labelNames: ["namespace", "status"],
+      registers: [this.register],
+    });
+
+    this.informerEventsTotal = new Counter({
+      name: "failed_pod_handler_informer_events_total",
+      help: "Total number of informer events",
+      labelNames: ["namespace", "verb"],
+      registers: [this.register],
+    });
+  }
+
+  async start() {
+    if (this.isRunning) {
+      this.logger.warn("failed pod handler already running");
+      return;
+    }
+
+    this.isRunning = true;
+
+    this.logger.info("starting failed pod handler");
+    await this.informer.start();
+  }
+
+  async stop() {
+    if (!this.isRunning) {
+      this.logger.warn("failed pod handler not running");
+      return;
+    }
+
+    this.isRunning = false;
+
+    this.logger.info("stopping failed pod handler");
+    await this.informer.stop();
+  }
+
+  private async withHistogram<T>(
+    histogram: Histogram<string>,
+    promise: Promise<T>,
+    labels?: Record<string, string>
+  ): Promise<T> {
+    const end = histogram.startTimer({ namespace: this.namespace, ...labels });
+    try {
+      return await promise;
+    } finally {
+      end();
+    }
+  }
+
+  /**
+   * Returns the non-nullable status of a pod
+   */
+  private podStatus(pod: V1Pod): PodStatus {
+    return (pod.status?.phase ?? "Unknown") as PodStatus;
+  }
+
+  private async onPodCompleted(pod: V1Pod) {
+    this.logger.info("pod-completed", this.podSummary(pod));
+    this.informerEventsTotal.inc({ namespace: this.namespace, verb: "add" });
+
+    if (!pod.metadata?.name) {
+      this.logger.error("pod-completed: no name", this.podSummary(pod));
+      return;
+    }
+
+    if (!pod.status) {
+      this.logger.error("pod-completed: no status", this.podSummary(pod));
+      return;
+    }
+
+    if (pod.metadata?.deletionTimestamp) {
+      this.logger.info("pod-completed: pod is being deleted", this.podSummary(pod));
+      return;
+    }
+
+    const podStatus = this.podStatus(pod);
+
+    switch (podStatus) {
+      case "Succeeded":
+        await this.withHistogram(this.processingDurationSeconds, this.onPodSucceeded(pod), {
+          status: podStatus,
+        });
+        break;
+      case "Failed":
+        await this.withHistogram(this.processingDurationSeconds, this.onPodFailed(pod), {
+          status: podStatus,
+        });
+        break;
+      default:
+        this.logger.error("pod-completed: unknown phase", this.podSummary(pod));
+    }
+  }
+
+  private async onPodSucceeded(pod: V1Pod) {
+    this.logger.info("pod-succeeded", this.podSummary(pod));
+    this.processedPodsTotal.inc({
+      namespace: this.namespace,
+      status: this.podStatus(pod),
+    });
+  }
+
+  private async onPodFailed(pod: V1Pod) {
+    this.logger.info("pod-failed", this.podSummary(pod));
+
+    try {
+      await this.processFailedPod(pod);
+    } catch (error) {
+      this.logger.error("pod-failed: error processing pod", this.podSummary(pod), { error });
+    } finally {
+      await this.deletePod(pod);
+    }
+  }
+
+  private async processFailedPod(pod: V1Pod) {
+    this.logger.info("pod-failed: processing pod", this.podSummary(pod));
+
+    const mainContainer = pod.status?.containerStatuses?.find((c) => c.name === "run-controller");
+
+    // If it's our special "graceful shutdown" exit code, don't process it further, just delete it
+    if (
+      mainContainer?.state?.terminated?.exitCode === FailedPodHandler.GRACEFUL_SHUTDOWN_EXIT_CODE
+    ) {
+      this.logger.debug("pod-failed: graceful shutdown detected", this.podSummary(pod));
+      this.processedPodsTotal.inc({
+        namespace: this.namespace,
+        status: "GracefulShutdown",
+      });
+      return;
+    }
+
+    this.processedPodsTotal.inc({
+      namespace: this.namespace,
+      status: this.podStatus(pod),
+    });
+  }
+
+  private async deletePod(pod: V1Pod) {
+    this.logger.info("pod-failed: deleting pod", this.podSummary(pod));
+    try {
+      await this.k8s.core.deleteNamespacedPod({
+        name: pod.metadata!.name!,
+        namespace: this.namespace,
+      });
+      this.deletedPodsTotal.inc({
+        namespace: this.namespace,
+        status: this.podStatus(pod),
+      });
+    } catch (error) {
+      this.logger.error("pod-failed: error deleting pod", this.podSummary(pod), { error });
+      this.deletionErrorsTotal.inc({
+        namespace: this.namespace,
+        error_type: error instanceof Error ? error.name : "unknown",
+      });
+    }
+  }
+
+  private makeOnError(informerName: string) {
+    return () => this.onError(informerName);
+  }
+
+  private async onError(informerName: string) {
+    if (!this.isRunning) {
+      this.logger.warn("onError: informer not running");
+      return;
+    }
+
+    this.logger.error("error event fired", { informerName });
+    this.informerEventsTotal.inc({ namespace: this.namespace, verb: "error" });
+
+    // Reconnect on errors
+    await setTimeout(this.reconnectIntervalMs);
+    await this.informer.start();
+  }
+
+  private makeOnConnect(informerName: string) {
+    return () => this.onConnect(informerName);
+  }
+
+  private async onConnect(informerName: string) {
+    this.logger.info(`informer connected: ${informerName}`);
+    this.informerEventsTotal.inc({ namespace: this.namespace, verb: "connect" });
+  }
+
+  private podSummary(pod: V1Pod) {
+    return {
+      name: pod.metadata?.name,
+      namespace: pod.metadata?.namespace,
+      status: pod.status?.phase,
+      deletionTimestamp: pod.metadata?.deletionTimestamp,
+    };
+  }
+
+  // Method to expose metrics for testing
+  public getMetrics() {
+    return {
+      processedPodsTotal: this.processedPodsTotal,
+      deletedPodsTotal: this.deletedPodsTotal,
+      deletionErrorsTotal: this.deletionErrorsTotal,
+      informerEventsTotal: this.informerEventsTotal,
+      processingDurationSeconds: this.processingDurationSeconds,
+    };
+  }
+}

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

@@ -0,0 +1,472 @@
+import { PodCleaner } from "./podCleaner.js";
+import { K8sApi, createK8sApi } from "../clients/kubernetes.js";
+import { setTimeout } from "timers/promises";
+import { describe, it, expect, beforeAll, afterEach } from "vitest";
+import { Registry } from "prom-client";
+
+describe("PodCleaner Integration Tests", () => {
+  const k8s = createK8sApi();
+  const namespace = "integration-test";
+  const register = new Registry();
+
+  beforeAll(async () => {
+    // Create the test namespace, only if it doesn't exist
+    try {
+      await k8s.core.readNamespace({ name: namespace });
+    } catch (error) {
+      await k8s.core.createNamespace({
+        body: {
+          metadata: {
+            name: namespace,
+          },
+        },
+      });
+    }
+  });
+
+  afterEach(async () => {
+    // Clear metrics to avoid conflicts
+    register.clear();
+
+    // Delete all pods in the namespace
+    await k8s.core.deleteCollectionNamespacedPod({ namespace, gracePeriodSeconds: 0 });
+  });
+
+  it("should clean up succeeded pods", async () => {
+    const podCleaner = new PodCleaner({ namespace, k8s, register });
+
+    try {
+      // Create a test pod that's in succeeded state
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        namePrefix: "test-succeeded-pod",
+      });
+
+      if (!podNames[0]) {
+        throw new Error("Failed to create test pod");
+      }
+      const podName = podNames[0];
+
+      // Wait for pod to complete
+      await waitForPodPhase({
+        k8sApi: k8s,
+        namespace,
+        podName,
+        phase: "Succeeded",
+      });
+
+      // Start the pod cleaner
+      await podCleaner.start();
+
+      // Wait for pod to be deleted
+      await waitForPodDeletion({
+        k8sApi: k8s,
+        namespace,
+        podName,
+      });
+
+      // Verify pod was deleted
+      expect(await podExists({ k8sApi: k8s, namespace, podName })).toBe(false);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
+  it("should accurately track deletion metrics", async () => {
+    const podCleaner = new PodCleaner({ namespace, k8s, register });
+    try {
+      // Create a test pod that's in succeeded state
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        namePrefix: "test-succeeded-pod",
+      });
+
+      // Wait for pod to be in succeeded state
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Succeeded",
+      });
+
+      await podCleaner.start();
+
+      // Wait for pod to be deleted
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      const metrics = podCleaner.getMetrics();
+      const deletionCycles = await metrics.deletionCyclesTotal.get();
+      const deletionTimestamp = await metrics.lastDeletionTimestamp.get();
+
+      expect(deletionCycles?.values[0]?.value).toBeGreaterThan(0);
+      expect(deletionTimestamp?.values[0]?.value).toBeGreaterThan(0);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
+  it("should handle different batch sizes - small", async () => {
+    const podCleaner = new PodCleaner({
+      namespace,
+      k8s,
+      register,
+      batchSize: 1,
+    });
+
+    try {
+      // Create some pods that will succeed
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 2,
+      });
+
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Succeeded",
+      });
+
+      await podCleaner.start();
+
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      const metrics = podCleaner.getMetrics();
+      const cycles = await metrics.deletionCyclesTotal.get();
+
+      expect(cycles?.values[0]?.value).toBe(2);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
+  it("should handle different batch sizes - large", async () => {
+    const podCleaner = new PodCleaner({
+      namespace,
+      k8s,
+      register,
+      batchSize: 5000,
+    });
+
+    try {
+      // Create some pods that will succeed
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 10,
+      });
+
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Succeeded",
+      });
+
+      await podCleaner.start();
+
+      await waitForPodsDeletion({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+      });
+
+      const metrics = podCleaner.getMetrics();
+      const cycles = await metrics.deletionCyclesTotal.get();
+
+      expect(cycles?.values[0]?.value).toBe(1);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
+  it("should not delete pods without app=task-run label", async () => {
+    const podCleaner = new PodCleaner({ namespace, k8s, register });
+
+    try {
+      // Create a test pod without the task-run label
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        labels: { app: "different-label" },
+        namePrefix: "non-task-run-pod",
+      });
+
+      if (!podNames[0]) {
+        throw new Error("Failed to create test pod");
+      }
+      const podName = podNames[0];
+
+      // Wait for pod to complete
+      await waitForPodPhase({
+        k8sApi: k8s,
+        namespace,
+        podName,
+        phase: "Succeeded",
+      });
+
+      await podCleaner.start();
+
+      // Wait a reasonable time to ensure pod isn't deleted
+      await setTimeout(5000);
+
+      // Verify pod still exists
+      expect(await podExists({ k8sApi: k8s, namespace, podName })).toBe(true);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
+  it("should not delete pods that are still running", async () => {
+    const podCleaner = new PodCleaner({ namespace, k8s, register });
+
+    try {
+      // Create a test pod with a long-running command
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        namePrefix: "running-pod",
+        command: ["sleep", "30"], // Will keep pod running
+      });
+
+      if (!podNames[0]) {
+        throw new Error("Failed to create test pod");
+      }
+      const podName = podNames[0];
+
+      // Wait for pod to be running
+      await waitForPodPhase({
+        k8sApi: k8s,
+        namespace,
+        podName,
+        phase: "Running",
+      });
+
+      await podCleaner.start();
+
+      // Wait a reasonable time to ensure pod isn't deleted
+      await setTimeout(5000);
+
+      // Verify pod still exists
+      expect(await podExists({ k8sApi: k8s, namespace, podName })).toBe(true);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+});
+
+// Helper functions
+async function waitForPodPhase({
+  k8sApi,
+  namespace,
+  podName,
+  phase,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podName: string;
+  phase: string;
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+
+  while (Date.now() - startTime < timeoutMs) {
+    const pod = await k8sApi.core.readNamespacedPod({
+      namespace,
+      name: podName,
+    });
+    if (pod.status?.phase === phase) {
+      return;
+    }
+    await setTimeout(waitMs);
+  }
+
+  throw new Error(`Pod ${podName} did not reach phase ${phase} within ${timeoutMs}ms`);
+}
+
+async function waitForPodDeletion({
+  k8sApi,
+  namespace,
+  podName,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podName: string;
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+
+  while (Date.now() - startTime < timeoutMs) {
+    try {
+      await k8sApi.core.readNamespacedPod({
+        namespace,
+        name: podName,
+      });
+      await setTimeout(waitMs);
+    } catch (error) {
+      // Pod was deleted
+      return;
+    }
+  }
+
+  throw new Error(`Pod ${podName} was not deleted within ${timeoutMs}ms`);
+}
+
+async function createTestPods({
+  k8sApi,
+  namespace,
+  count,
+  labels = { app: "task-run" },
+  shouldFail = false,
+  namePrefix = "test-pod",
+  command = ["/bin/sh", "-c", shouldFail ? "exit 1" : "exit 0"],
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  count: number;
+  labels?: Record<string, string>;
+  shouldFail?: boolean;
+  namePrefix?: string;
+  command?: string[];
+}) {
+  const createdPods: string[] = [];
+
+  for (let i = 0; i < count; i++) {
+    const podName = `${namePrefix}-${i}`;
+    await k8sApi.core.createNamespacedPod({
+      namespace,
+      body: {
+        metadata: {
+          name: podName,
+          labels,
+        },
+        spec: {
+          restartPolicy: "Never",
+          containers: [
+            {
+              name: "test",
+              image: "busybox:1.37.0",
+              command,
+            },
+          ],
+        },
+      },
+    });
+    createdPods.push(podName);
+  }
+
+  return createdPods;
+}
+
+async function waitForPodsPhase({
+  k8sApi,
+  namespace,
+  podNames,
+  phase,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podNames: string[];
+  phase: "Pending" | "Running" | "Succeeded" | "Failed" | "Unknown";
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+  const pendingPods = new Set(podNames);
+
+  while (pendingPods.size > 0 && Date.now() - startTime < timeoutMs) {
+    const pods = await k8sApi.core.listNamespacedPod({ namespace });
+
+    for (const pod of pods.items) {
+      if (pendingPods.has(pod.metadata?.name ?? "") && pod.status?.phase === phase) {
+        pendingPods.delete(pod.metadata?.name ?? "");
+      }
+    }
+
+    if (pendingPods.size > 0) {
+      await setTimeout(waitMs);
+    }
+  }
+
+  if (pendingPods.size > 0) {
+    throw new Error(
+      `Pods [${Array.from(pendingPods).join(
+        ", "
+      )}] did not reach phase ${phase} within ${timeoutMs}ms`
+    );
+  }
+}
+
+async function waitForPodsDeletion({
+  k8sApi,
+  namespace,
+  podNames,
+  timeoutMs = 10000,
+  waitMs = 1000,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podNames: string[];
+  timeoutMs?: number;
+  waitMs?: number;
+}) {
+  const startTime = Date.now();
+  const pendingPods = new Set(podNames);
+
+  while (pendingPods.size > 0 && Date.now() - startTime < timeoutMs) {
+    const pods = await k8sApi.core.listNamespacedPod({ namespace });
+    const existingPods = new Set(pods.items.map((pod) => pod.metadata?.name ?? ""));
+
+    for (const podName of pendingPods) {
+      if (!existingPods.has(podName)) {
+        pendingPods.delete(podName);
+      }
+    }
+
+    if (pendingPods.size > 0) {
+      await setTimeout(waitMs);
+    }
+  }
+
+  if (pendingPods.size > 0) {
+    throw new Error(
+      `Pods [${Array.from(pendingPods).join(", ")}] were not deleted within ${timeoutMs}ms`
+    );
+  }
+}
+
+async function podExists({
+  k8sApi,
+  namespace,
+  podName,
+}: {
+  k8sApi: K8sApi;
+  namespace: string;
+  podName: string;
+}) {
+  const pods = await k8sApi.core.listNamespacedPod({ namespace });
+  return pods.items.some((p) => p.metadata?.name === podName);
+}

apps/supervisor/src/services/podCleaner.ts

@@ -0,0 +1,118 @@
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import { K8sApi } from "../clients/kubernetes.js";
+import { createK8sApi } from "../clients/kubernetes.js";
+import { HeartbeatService } from "@trigger.dev/core/v3";
+import { Counter, Gauge, Registry } from "prom-client";
+import { register } from "../metrics.js";
+
+export type PodCleanerOptions = {
+  namespace: string;
+  k8s?: K8sApi;
+  register?: Registry;
+  batchSize?: number;
+  intervalMs?: number;
+};
+
+export class PodCleaner {
+  private readonly logger = new SimpleStructuredLogger("pod-cleaner");
+  private readonly k8s: K8sApi;
+  private readonly namespace: string;
+
+  private readonly batchSize: number;
+  private readonly deletionHeartbeat: HeartbeatService;
+
+  // Metrics
+  private readonly register: Registry;
+  private readonly deletionCyclesTotal: Counter;
+  private readonly lastDeletionTimestamp: Gauge;
+
+  constructor(opts: PodCleanerOptions) {
+    this.k8s = opts.k8s ?? createK8sApi();
+
+    this.namespace = opts.namespace;
+    this.batchSize = opts.batchSize ?? 500;
+
+    this.deletionHeartbeat = new HeartbeatService({
+      intervalMs: opts.intervalMs ?? 10000,
+      leadingEdge: true,
+      heartbeat: this.deleteCompletedPods.bind(this),
+    });
+
+    // Initialize metrics
+    this.register = opts.register ?? register;
+
+    this.deletionCyclesTotal = new Counter({
+      name: "pod_cleaner_deletion_cycles_total",
+      help: "Total number of pod deletion cycles run",
+      labelNames: ["namespace", "status", "batch_size"],
+      registers: [this.register],
+    });
+
+    this.lastDeletionTimestamp = new Gauge({
+      name: "pod_cleaner_last_deletion_timestamp",
+      help: "Timestamp of the last deletion cycle",
+      labelNames: ["namespace"],
+      registers: [this.register],
+    });
+  }
+
+  async start() {
+    this.deletionHeartbeat.start();
+  }
+
+  async stop() {
+    this.deletionHeartbeat.stop();
+  }
+
+  private async deleteCompletedPods() {
+    let continuationToken: string | undefined;
+
+    do {
+      try {
+        const result = await this.k8s.core.deleteCollectionNamespacedPod({
+          namespace: this.namespace,
+          labelSelector: "app=task-run",
+          fieldSelector: "status.phase=Succeeded",
+          limit: this.batchSize,
+          _continue: continuationToken,
+          gracePeriodSeconds: 0,
+          propagationPolicy: "Background",
+          timeoutSeconds: 30,
+        });
+
+        // Update continuation token for next batch
+        continuationToken = result.metadata?._continue;
+
+        // Increment the deletion cycles counter
+        this.deletionCyclesTotal.inc({
+          namespace: this.namespace,
+          batch_size: this.batchSize,
+          status: "succeeded",
+        });
+
+        this.logger.info("Deleted batch of pods", { continuationToken });
+      } catch (err) {
+        this.logger.error("Failed to delete batch of pods", {
+          err: err instanceof Error ? err.message : String(err),
+        });
+
+        this.deletionCyclesTotal.inc({
+          namespace: this.namespace,
+          batch_size: this.batchSize,
+          status: "failed",
+        });
+        break;
+      }
+    } while (continuationToken);
+
+    this.lastDeletionTimestamp.set({ namespace: this.namespace }, Date.now());
+  }
+
+  // Method to expose metrics for testing
+  public getMetrics() {
+    return {
+      deletionCyclesTotal: this.deletionCyclesTotal,
+      lastDeletionTimestamp: this.lastDeletionTimestamp,
+    };
+  }
+}

apps/supervisor/src/util.ts

@@ -0,0 +1,16 @@
+export function getDockerHostDomain() {
+  const isMacOs = process.platform === "darwin";
+  const isWindows = process.platform === "win32";
+
+  return isMacOs || isWindows ? "host.docker.internal" : "localhost";
+}
+
+export function getRunnerId(runId: string, attemptNumber?: number) {
+  const parts = ["runner", runId.replace("run_", "")];
+
+  if (attemptNumber && attemptNumber > 1) {
+    parts.push(`attempt-${attemptNumber}`);
+  }
+
+  return parts.join("-");
+}

apps/supervisor/src/workloadManager/docker.ts

@@ -0,0 +1,87 @@
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import {
+  type WorkloadManager,
+  type WorkloadManagerCreateOptions,
+  type WorkloadManagerOptions,
+} from "./types.js";
+import { x } from "tinyexec";
+import { env } from "../env.js";
+import { getDockerHostDomain, getRunnerId } from "../util.js";
+
+export class DockerWorkloadManager implements WorkloadManager {
+  private readonly logger = new SimpleStructuredLogger("docker-workload-provider");
+
+  constructor(private opts: WorkloadManagerOptions) {
+    if (opts.workloadApiDomain) {
+      this.logger.warn("[DockerWorkloadProvider] ⚠️ Custom workload API domain", {
+        domain: opts.workloadApiDomain,
+      });
+    }
+  }
+
+  async create(opts: WorkloadManagerCreateOptions) {
+    this.logger.log("[DockerWorkloadProvider] Creating container", { opts });
+
+    const runnerId = getRunnerId(opts.runFriendlyId, opts.nextAttemptNumber);
+
+    const runArgs = [
+      "run",
+      "--detach",
+      `--network=${env.DOCKER_NETWORK}`,
+      `--env=TRIGGER_DEQUEUED_AT_MS=${opts.dequeuedAt.getTime()}`,
+      `--env=TRIGGER_POD_SCHEDULED_AT_MS=${Date.now()}`,
+      `--env=TRIGGER_ENV_ID=${opts.envId}`,
+      `--env=TRIGGER_RUN_ID=${opts.runFriendlyId}`,
+      `--env=TRIGGER_SNAPSHOT_ID=${opts.snapshotFriendlyId}`,
+      `--env=TRIGGER_SUPERVISOR_API_PROTOCOL=${this.opts.workloadApiProtocol}`,
+      `--env=TRIGGER_SUPERVISOR_API_PORT=${this.opts.workloadApiPort}`,
+      `--env=TRIGGER_SUPERVISOR_API_DOMAIN=${this.opts.workloadApiDomain ?? getDockerHostDomain()}`,
+      `--env=TRIGGER_WORKER_INSTANCE_NAME=${env.TRIGGER_WORKER_INSTANCE_NAME}`,
+      `--env=OTEL_EXPORTER_OTLP_ENDPOINT=${env.OTEL_EXPORTER_OTLP_ENDPOINT}`,
+      `--env=TRIGGER_RUNNER_ID=${runnerId}`,
+      `--hostname=${runnerId}`,
+      `--name=${runnerId}`,
+    ];
+
+    if (this.opts.warmStartUrl) {
+      runArgs.push(`--env=TRIGGER_WARM_START_URL=${this.opts.warmStartUrl}`);
+    }
+
+    if (this.opts.metadataUrl) {
+      runArgs.push(`--env=TRIGGER_METADATA_URL=${this.opts.metadataUrl}`);
+    }
+
+    if (this.opts.heartbeatIntervalSeconds) {
+      runArgs.push(
+        `--env=TRIGGER_HEARTBEAT_INTERVAL_SECONDS=${this.opts.heartbeatIntervalSeconds}`
+      );
+    }
+
+    if (this.opts.snapshotPollIntervalSeconds) {
+      runArgs.push(
+        `--env=TRIGGER_SNAPSHOT_POLL_INTERVAL_SECONDS=${this.opts.snapshotPollIntervalSeconds}`
+      );
+    }
+
+    if (this.opts.additionalEnvVars) {
+      Object.entries(this.opts.additionalEnvVars).forEach(([key, value]) => {
+        runArgs.push(`--env=${key}=${value}`);
+      });
+    }
+
+    if (env.ENFORCE_MACHINE_PRESETS) {
+      runArgs.push(`--cpus=${opts.machine.cpu}`, `--memory=${opts.machine.memory}G`);
+      runArgs.push(`--env=TRIGGER_MACHINE_CPU=${opts.machine.cpu}`);
+      runArgs.push(`--env=TRIGGER_MACHINE_MEMORY=${opts.machine.memory}`);
+    }
+
+    runArgs.push(`${opts.image}`);
+
+    try {
+      const { stdout, stderr } = await x("docker", runArgs);
+      this.logger.debug("[DockerWorkloadProvider] Create succeeded", { stdout, stderr });
+    } catch (error) {
+      this.logger.error("[DockerWorkloadProvider] Create failed:", { opts, error });
+    }
+  }
+}

apps/supervisor/src/workloadManager/kubernetes.ts

@@ -0,0 +1,284 @@
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import {
+  type WorkloadManager,
+  type WorkloadManagerCreateOptions,
+  type WorkloadManagerOptions,
+} from "./types.js";
+import type { EnvironmentType, MachinePreset } from "@trigger.dev/core/v3";
+import { env } from "../env.js";
+import { type K8sApi, createK8sApi, type k8s } from "../clients/kubernetes.js";
+import { getRunnerId } from "../util.js";
+
+type ResourceQuantities = {
+  [K in "cpu" | "memory" | "ephemeral-storage"]?: string;
+};
+
+export class KubernetesWorkloadManager implements WorkloadManager {
+  private readonly logger = new SimpleStructuredLogger("kubernetes-workload-provider");
+  private k8s: K8sApi;
+  private namespace = env.KUBERNETES_NAMESPACE;
+
+  constructor(private opts: WorkloadManagerOptions) {
+    this.k8s = createK8sApi();
+
+    if (opts.workloadApiDomain) {
+      this.logger.warn("[KubernetesWorkloadManager] ⚠️ Custom workload API domain", {
+        domain: opts.workloadApiDomain,
+      });
+    }
+  }
+
+  async create(opts: WorkloadManagerCreateOptions) {
+    this.logger.log("[KubernetesWorkloadManager] Creating container", { opts });
+
+    const runnerId = getRunnerId(opts.runFriendlyId, opts.nextAttemptNumber);
+
+    try {
+      await this.k8s.core.createNamespacedPod({
+        namespace: this.namespace,
+        body: {
+          metadata: {
+            name: runnerId,
+            namespace: this.namespace,
+            labels: {
+              ...this.#getSharedLabels(opts),
+              app: "task-run",
+              "app.kubernetes.io/part-of": "trigger-worker",
+              "app.kubernetes.io/component": "create",
+            },
+          },
+          spec: {
+            ...this.#defaultPodSpec,
+            terminationGracePeriodSeconds: 60 * 60,
+            containers: [
+              {
+                name: "run-controller",
+                image: opts.image,
+                ports: [
+                  {
+                    containerPort: 8000,
+                  },
+                ],
+                resources: this.#getResourcesForMachine(opts.machine),
+                env: [
+                  {
+                    name: "TRIGGER_DEQUEUED_AT_MS",
+                    value: opts.dequeuedAt.getTime().toString(),
+                  },
+                  {
+                    name: "TRIGGER_POD_SCHEDULED_AT_MS",
+                    value: Date.now().toString(),
+                  },
+                  {
+                    name: "TRIGGER_RUN_ID",
+                    value: opts.runFriendlyId,
+                  },
+                  {
+                    name: "TRIGGER_ENV_ID",
+                    value: opts.envId,
+                  },
+                  {
+                    name: "TRIGGER_SNAPSHOT_ID",
+                    value: opts.snapshotFriendlyId,
+                  },
+                  {
+                    name: "TRIGGER_SUPERVISOR_API_PROTOCOL",
+                    value: this.opts.workloadApiProtocol,
+                  },
+                  {
+                    name: "TRIGGER_SUPERVISOR_API_PORT",
+                    value: `${this.opts.workloadApiPort}`,
+                  },
+                  {
+                    name: "TRIGGER_SUPERVISOR_API_DOMAIN",
+                    ...(this.opts.workloadApiDomain
+                      ? {
+                          value: this.opts.workloadApiDomain,
+                        }
+                      : {
+                          valueFrom: {
+                            fieldRef: {
+                              fieldPath: "status.hostIP",
+                            },
+                          },
+                        }),
+                  },
+                  {
+                    name: "TRIGGER_WORKER_INSTANCE_NAME",
+                    valueFrom: {
+                      fieldRef: {
+                        fieldPath: "spec.nodeName",
+                      },
+                    },
+                  },
+                  {
+                    name: "OTEL_EXPORTER_OTLP_ENDPOINT",
+                    value: env.OTEL_EXPORTER_OTLP_ENDPOINT,
+                  },
+                  {
+                    name: "TRIGGER_RUNNER_ID",
+                    value: runnerId,
+                  },
+                  {
+                    name: "TRIGGER_MACHINE_CPU",
+                    value: `${opts.machine.cpu}`,
+                  },
+                  {
+                    name: "TRIGGER_MACHINE_MEMORY",
+                    value: `${opts.machine.memory}`,
+                  },
+                  {
+                    name: "LIMITS_CPU",
+                    valueFrom: {
+                      resourceFieldRef: {
+                        resource: "limits.cpu",
+                      },
+                    },
+                  },
+                  {
+                    name: "LIMITS_MEMORY",
+                    valueFrom: {
+                      resourceFieldRef: {
+                        resource: "limits.memory",
+                      },
+                    },
+                  },
+                  ...(this.opts.warmStartUrl
+                    ? [{ name: "TRIGGER_WARM_START_URL", value: this.opts.warmStartUrl }]
+                    : []),
+                  ...(this.opts.metadataUrl
+                    ? [{ name: "TRIGGER_METADATA_URL", value: this.opts.metadataUrl }]
+                    : []),
+                  ...(this.opts.heartbeatIntervalSeconds
+                    ? [
+                        {
+                          name: "TRIGGER_HEARTBEAT_INTERVAL_SECONDS",
+                          value: `${this.opts.heartbeatIntervalSeconds}`,
+                        },
+                      ]
+                    : []),
+                  ...(this.opts.snapshotPollIntervalSeconds
+                    ? [
+                        {
+                          name: "TRIGGER_SNAPSHOT_POLL_INTERVAL_SECONDS",
+                          value: `${this.opts.snapshotPollIntervalSeconds}`,
+                        },
+                      ]
+                    : []),
+                  ...(this.opts.additionalEnvVars
+                    ? Object.entries(this.opts.additionalEnvVars).map(([key, value]) => ({
+                        name: key,
+                        value: value,
+                      }))
+                    : []),
+                ],
+              },
+            ],
+          },
+        },
+      });
+    } catch (err: unknown) {
+      this.#handleK8sError(err);
+    }
+  }
+
+  #throwUnlessRecord(candidate: unknown): asserts candidate is Record<string, unknown> {
+    if (typeof candidate !== "object" || candidate === null) {
+      throw candidate;
+    }
+  }
+
+  #handleK8sError(err: unknown) {
+    this.#throwUnlessRecord(err);
+
+    if ("body" in err && err.body) {
+      this.logger.error("[KubernetesWorkloadManager] Create failed", { rawError: err.body });
+      this.#throwUnlessRecord(err.body);
+
+      if (typeof err.body.message === "string") {
+        throw new Error(err.body?.message);
+      } else {
+        throw err.body;
+      }
+    } else {
+      this.logger.error("[KubernetesWorkloadManager] Create failed", { rawError: err });
+      throw err;
+    }
+  }
+
+  #envTypeToLabelValue(type: EnvironmentType) {
+    switch (type) {
+      case "PRODUCTION":
+        return "prod";
+      case "STAGING":
+        return "stg";
+      case "DEVELOPMENT":
+        return "dev";
+      case "PREVIEW":
+        return "preview";
+    }
+  }
+
+  private getImagePullSecrets(): k8s.V1LocalObjectReference[] | undefined {
+    return this.opts.imagePullSecrets?.map((name) => ({ name }));
+  }
+
+  get #defaultPodSpec(): Omit<k8s.V1PodSpec, "containers"> {
+    return {
+      restartPolicy: "Never",
+      automountServiceAccountToken: false,
+      imagePullSecrets: this.getImagePullSecrets(),
+      nodeSelector: {
+        nodetype: env.KUBERNETES_WORKER_NODETYPE_LABEL,
+      },
+    };
+  }
+
+  get #defaultResourceRequests(): ResourceQuantities {
+    return {
+      "ephemeral-storage": env.EPHEMERAL_STORAGE_SIZE_REQUEST,
+    };
+  }
+
+  get #defaultResourceLimits(): ResourceQuantities {
+    return {
+      "ephemeral-storage": env.EPHEMERAL_STORAGE_SIZE_LIMIT,
+    };
+  }
+
+  #getSharedLabels(opts: WorkloadManagerCreateOptions): Record<string, string> {
+    return {
+      env: opts.envId,
+      envtype: this.#envTypeToLabelValue(opts.envType),
+      org: opts.orgId,
+      project: opts.projectId,
+    };
+  }
+
+  #getResourceRequestsForMachine(preset: MachinePreset): ResourceQuantities {
+    return {
+      cpu: `${preset.cpu * 0.75}`,
+      memory: `${preset.memory}G`,
+    };
+  }
+
+  #getResourceLimitsForMachine(preset: MachinePreset): ResourceQuantities {
+    return {
+      cpu: `${preset.cpu}`,
+      memory: `${preset.memory}G`,
+    };
+  }
+
+  #getResourcesForMachine(preset: MachinePreset): k8s.V1ResourceRequirements {
+    return {
+      requests: {
+        ...this.#defaultResourceRequests,
+        ...this.#getResourceRequestsForMachine(preset),
+      },
+      limits: {
+        ...this.#defaultResourceLimits,
+        ...this.#getResourceLimitsForMachine(preset),
+      },
+    };
+  }
+}

apps/supervisor/src/workloadManager/types.ts

@@ -0,0 +1,34 @@
+import { type EnvironmentType, type MachinePreset } from "@trigger.dev/core/v3";
+
+export interface WorkloadManagerOptions {
+  workloadApiProtocol: "http" | "https";
+  workloadApiDomain?: string; // If unset, will use orchestrator-specific default
+  workloadApiPort: number;
+  warmStartUrl?: string;
+  metadataUrl?: string;
+  imagePullSecrets?: string[];
+  heartbeatIntervalSeconds?: number;
+  snapshotPollIntervalSeconds?: number;
+  additionalEnvVars?: Record<string, string>;
+}
+
+export interface WorkloadManager {
+  create: (opts: WorkloadManagerCreateOptions) => Promise<unknown>;
+}
+
+export interface WorkloadManagerCreateOptions {
+  image: string;
+  machine: MachinePreset;
+  version: string;
+  nextAttemptNumber?: number;
+  dequeuedAt: Date;
+  // identifiers
+  envId: string;
+  envType: EnvironmentType;
+  orgId: string;
+  projectId: string;
+  runId: string;
+  runFriendlyId: string;
+  snapshotId: string;
+  snapshotFriendlyId: string;
+}

apps/supervisor/src/workloadServer/index.ts

@@ -0,0 +1,582 @@
+import { type Namespace, Server, type Socket } from "socket.io";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
+import EventEmitter from "node:events";
+import { z } from "zod";
+import {
+  type SupervisorHttpClient,
+  WORKLOAD_HEADERS,
+  type WorkloadClientSocketData,
+  type WorkloadClientToServerEvents,
+  type WorkloadContinueRunExecutionResponseBody,
+  WorkloadDebugLogRequestBody,
+  type WorkloadDequeueFromVersionResponseBody,
+  WorkloadHeartbeatRequestBody,
+  type WorkloadHeartbeatResponseBody,
+  WorkloadRunAttemptCompleteRequestBody,
+  type WorkloadRunAttemptCompleteResponseBody,
+  WorkloadRunAttemptStartRequestBody,
+  type WorkloadRunAttemptStartResponseBody,
+  type WorkloadRunLatestSnapshotResponseBody,
+  type WorkloadServerToClientEvents,
+  type WorkloadSuspendRunResponseBody,
+} from "@trigger.dev/core/v3/workers";
+import { HttpServer, type CheckpointClient } from "@trigger.dev/core/v3/serverOnly";
+import { type IncomingMessage } from "node:http";
+import { register } from "../metrics.js";
+
+// Use the official export when upgrading to socket.io@4.8.0
+interface DefaultEventsMap {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  [event: string]: (...args: any[]) => void;
+}
+
+const WorkloadActionParams = z.object({
+  runFriendlyId: z.string(),
+  snapshotFriendlyId: z.string(),
+});
+
+type WorkloadServerEvents = {
+  runConnected: [
+    {
+      run: {
+        friendlyId: string;
+      };
+    },
+  ];
+  runDisconnected: [
+    {
+      run: {
+        friendlyId: string;
+      };
+    },
+  ];
+};
+
+type WorkloadServerOptions = {
+  port: number;
+  host?: string;
+  workerClient: SupervisorHttpClient;
+  checkpointClient?: CheckpointClient;
+};
+
+export class WorkloadServer extends EventEmitter<WorkloadServerEvents> {
+  private checkpointClient?: CheckpointClient;
+
+  private readonly httpServer: HttpServer;
+  private readonly websocketServer: Namespace<
+    WorkloadClientToServerEvents,
+    WorkloadServerToClientEvents,
+    DefaultEventsMap,
+    WorkloadClientSocketData
+  >;
+
+  private readonly runSockets = new Map<
+    string,
+    Socket<
+      WorkloadClientToServerEvents,
+      WorkloadServerToClientEvents,
+      DefaultEventsMap,
+      WorkloadClientSocketData
+    >
+  >();
+
+  private readonly workerClient: SupervisorHttpClient;
+
+  constructor(opts: WorkloadServerOptions) {
+    super();
+
+    const host = opts.host ?? "0.0.0.0";
+    const port = opts.port;
+
+    this.workerClient = opts.workerClient;
+    this.checkpointClient = opts.checkpointClient;
+
+    this.httpServer = this.createHttpServer({ host, port });
+    this.websocketServer = this.createWebsocketServer();
+  }
+
+  private headerValueFromRequest(req: IncomingMessage, headerName: string): string | undefined {
+    const value = req.headers[headerName];
+
+    if (Array.isArray(value)) {
+      return value[0];
+    }
+
+    return value;
+  }
+
+  private runnerIdFromRequest(req: IncomingMessage): string | undefined {
+    return this.headerValueFromRequest(req, WORKLOAD_HEADERS.RUNNER_ID);
+  }
+
+  private deploymentIdFromRequest(req: IncomingMessage): string | undefined {
+    return this.headerValueFromRequest(req, WORKLOAD_HEADERS.DEPLOYMENT_ID);
+  }
+
+  private deploymentVersionFromRequest(req: IncomingMessage): string | undefined {
+    return this.headerValueFromRequest(req, WORKLOAD_HEADERS.DEPLOYMENT_VERSION);
+  }
+
+  private projectRefFromRequest(req: IncomingMessage): string | undefined {
+    return this.headerValueFromRequest(req, WORKLOAD_HEADERS.PROJECT_REF);
+  }
+
+  private createHttpServer({ host, port }: { host: string; port: number }) {
+    return new HttpServer({
+      port,
+      host,
+      metrics: {
+        register,
+        expose: false,
+      },
+    })
+      .route("/health", "GET", {
+        handler: async ({ reply }) => {
+          reply.text("OK");
+        },
+      })
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/attempts/start",
+        "POST",
+        {
+          paramsSchema: WorkloadActionParams,
+          bodySchema: WorkloadRunAttemptStartRequestBody,
+          handler: async ({ req, reply, params, body }) => {
+            const startResponse = await this.workerClient.startRunAttempt(
+              params.runFriendlyId,
+              params.snapshotFriendlyId,
+              body,
+              this.runnerIdFromRequest(req)
+            );
+
+            if (!startResponse.success) {
+              console.error("Failed to start run", {
+                params,
+                error: startResponse.error,
+              });
+              reply.empty(500);
+              return;
+            }
+
+            reply.json(startResponse.data satisfies WorkloadRunAttemptStartResponseBody);
+            return;
+          },
+        }
+      )
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/attempts/complete",
+        "POST",
+        {
+          paramsSchema: WorkloadActionParams,
+          bodySchema: WorkloadRunAttemptCompleteRequestBody,
+          handler: async ({ req, reply, params, body }) => {
+            console.log("headers", req.headers);
+            const completeResponse = await this.workerClient.completeRunAttempt(
+              params.runFriendlyId,
+              params.snapshotFriendlyId,
+              body,
+              this.runnerIdFromRequest(req)
+            );
+
+            if (!completeResponse.success) {
+              console.error("Failed to complete run", {
+                params,
+                error: completeResponse.error,
+              });
+              reply.empty(500);
+              return;
+            }
+
+            reply.json(completeResponse.data satisfies WorkloadRunAttemptCompleteResponseBody);
+            return;
+          },
+        }
+      )
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/heartbeat",
+        "POST",
+        {
+          paramsSchema: WorkloadActionParams,
+          bodySchema: WorkloadHeartbeatRequestBody,
+          handler: async ({ req, reply, params, body }) => {
+            const heartbeatResponse = await this.workerClient.heartbeatRun(
+              params.runFriendlyId,
+              params.snapshotFriendlyId,
+              body,
+              this.runnerIdFromRequest(req)
+            );
+
+            if (!heartbeatResponse.success) {
+              console.error("Failed to heartbeat run", {
+                params,
+                error: heartbeatResponse.error,
+              });
+              reply.empty(500);
+              return;
+            }
+
+            reply.json({
+              ok: true,
+            } satisfies WorkloadHeartbeatResponseBody);
+          },
+        }
+      )
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/suspend",
+        "GET",
+        {
+          paramsSchema: WorkloadActionParams,
+          handler: async ({ reply, params, req }) => {
+            console.debug("Suspend request", { params, headers: req.headers });
+
+            if (!this.checkpointClient) {
+              reply.json(
+                {
+                  ok: false,
+                  error: "Checkpoints disabled",
+                } satisfies WorkloadSuspendRunResponseBody,
+                false,
+                400
+              );
+              return;
+            }
+
+            const runnerId = this.runnerIdFromRequest(req);
+            const deploymentVersion = this.deploymentVersionFromRequest(req);
+            const projectRef = this.projectRefFromRequest(req);
+
+            if (!runnerId || !deploymentVersion || !projectRef) {
+              console.error("Invalid headers for suspend request", {
+                ...params,
+                headers: req.headers,
+              });
+              reply.json(
+                {
+                  ok: false,
+                  error: "Invalid headers",
+                } satisfies WorkloadSuspendRunResponseBody,
+                false,
+                400
+              );
+              return;
+            }
+
+            reply.json(
+              {
+                ok: true,
+              } satisfies WorkloadSuspendRunResponseBody,
+              false,
+              202
+            );
+
+            const suspendResult = await this.checkpointClient.suspendRun({
+              runFriendlyId: params.runFriendlyId,
+              snapshotFriendlyId: params.snapshotFriendlyId,
+              body: {
+                runnerId,
+                runId: params.runFriendlyId,
+                snapshotId: params.snapshotFriendlyId,
+                projectRef,
+                deploymentVersion,
+              },
+            });
+
+            if (!suspendResult) {
+              console.error("Failed to suspend run", { params });
+              return;
+            }
+          },
+        }
+      )
+      .route(
+        "/api/v1/workload-actions/runs/:runFriendlyId/snapshots/:snapshotFriendlyId/continue",
+        "GET",
+        {
+          paramsSchema: WorkloadActionParams,
+          handler: async ({ req, reply, params }) => {
+            console.debug("Run continuation request", { params });
+
+            const continuationResult = await this.workerClient.continueRunExecution(
+              params.runFriendlyId,
+              params.snapshotFriendlyId,
+              this.runnerIdFromRequest(req)
+            );
+
+            if (!continuationResult.success) {
+              console.error("Failed to continue run execution", { params });
+              reply.json(
+                {
+                  ok: false,
+                  error: "Failed to continue run execution",
+                },
+                false,
+                400
+              );
+              return;
+            }
+
+            reply.json(continuationResult.data as WorkloadContinueRunExecutionResponseBody);
+          },
+        }
+      )
+      .route("/api/v1/workload-actions/runs/:runFriendlyId/snapshots/latest", "GET", {
+        paramsSchema: WorkloadActionParams.pick({ runFriendlyId: true }),
+        handler: async ({ req, reply, params }) => {
+          const latestSnapshotResponse = await this.workerClient.getLatestSnapshot(
+            params.runFriendlyId,
+            this.runnerIdFromRequest(req)
+          );
+
+          if (!latestSnapshotResponse.success) {
+            console.error("Failed to get latest snapshot", {
+              runId: params.runFriendlyId,
+              error: latestSnapshotResponse.error,
+            });
+            reply.empty(500);
+            return;
+          }
+
+          reply.json({
+            execution: latestSnapshotResponse.data.execution,
+          } satisfies WorkloadRunLatestSnapshotResponseBody);
+        },
+      })
+      .route("/api/v1/workload-actions/runs/:runFriendlyId/logs/debug", "POST", {
+        paramsSchema: WorkloadActionParams.pick({ runFriendlyId: true }),
+        bodySchema: WorkloadDebugLogRequestBody,
+        handler: async ({ req, reply, params, body }) => {
+          reply.empty(204);
+
+          await this.workerClient.sendDebugLog(
+            params.runFriendlyId,
+            body,
+            this.runnerIdFromRequest(req)
+          );
+        },
+      })
+      .route("/api/v1/workload-actions/deployments/:deploymentId/dequeue", "GET", {
+        paramsSchema: z.object({
+          deploymentId: z.string(),
+        }),
+
+        handler: async ({ req, reply, params }) => {
+          const dequeueResponse = await this.workerClient.dequeueFromVersion(
+            params.deploymentId,
+            1,
+            this.runnerIdFromRequest(req)
+          );
+
+          if (!dequeueResponse.success) {
+            console.error("Failed to get latest snapshot", {
+              deploymentId: params.deploymentId,
+              error: dequeueResponse.error,
+            });
+            reply.empty(500);
+            return;
+          }
+
+          reply.json(dequeueResponse.data satisfies WorkloadDequeueFromVersionResponseBody);
+        },
+      });
+  }
+
+  private createWebsocketServer() {
+    const io = new Server(this.httpServer.server);
+
+    const websocketServer: Namespace<
+      WorkloadClientToServerEvents,
+      WorkloadServerToClientEvents,
+      DefaultEventsMap,
+      WorkloadClientSocketData
+    > = io.of("/workload");
+
+    websocketServer.on("disconnect", (socket) => {
+      console.log("[WorkloadSocket] disconnect", socket.id);
+    });
+    websocketServer.use(async (socket, next) => {
+      function setSocketDataFromHeader(
+        dataKey: keyof typeof socket.data,
+        headerName: string,
+        required: boolean = true
+      ) {
+        const value = socket.handshake.headers[headerName];
+
+        if (value) {
+          if (Array.isArray(value)) {
+            if (value[0]) {
+              socket.data[dataKey] = value[0];
+              return;
+            }
+          } else {
+            socket.data[dataKey] = value;
+            return;
+          }
+        }
+
+        if (required) {
+          console.error("[WorkloadSocket] missing required header", { headerName });
+          throw new Error("missing header");
+        }
+      }
+
+      try {
+        setSocketDataFromHeader("deploymentId", WORKLOAD_HEADERS.DEPLOYMENT_ID);
+        setSocketDataFromHeader("runnerId", WORKLOAD_HEADERS.RUNNER_ID);
+      } catch (error) {
+        console.error("[WorkloadSocket] setSocketDataFromHeader error", { error });
+        socket.disconnect(true);
+        return;
+      }
+
+      console.debug("[WorkloadSocket] auth success", socket.data);
+
+      next();
+    });
+    websocketServer.on("connection", (socket) => {
+      const logger = new SimpleStructuredLogger("workload-namespace", undefined, {
+        namespace: "workload",
+        socketId: socket.id,
+        socketData: socket.data,
+      });
+
+      const getSocketMetadata = () => {
+        return {
+          deploymentId: socket.data.deploymentId,
+          runId: socket.data.runFriendlyId,
+          snapshotId: socket.data.snapshotId,
+          runnerId: socket.data.runnerId,
+        };
+      };
+
+      const runConnected = (friendlyId: string) => {
+        logger.debug("runConnected", { ...getSocketMetadata() });
+
+        // If there's already a run ID set, we should "disconnect" it from this socket
+        if (socket.data.runFriendlyId) {
+          logger.debug("runConnected: disconnecting existing run", {
+            ...getSocketMetadata(),
+            newRunId: friendlyId,
+            oldRunId: socket.data.runFriendlyId,
+          });
+          runDisconnected(socket.data.runFriendlyId);
+        }
+
+        this.runSockets.set(friendlyId, socket);
+        this.emit("runConnected", { run: { friendlyId } });
+        socket.data.runFriendlyId = friendlyId;
+      };
+
+      const runDisconnected = (friendlyId: string) => {
+        logger.debug("runDisconnected", { ...getSocketMetadata() });
+
+        this.runSockets.delete(friendlyId);
+        this.emit("runDisconnected", { run: { friendlyId } });
+        socket.data.runFriendlyId = undefined;
+      };
+
+      logger.log("wsServer socket connected", { ...getSocketMetadata() });
+
+      // FIXME: where does this get set?
+      if (socket.data.runFriendlyId) {
+        runConnected(socket.data.runFriendlyId);
+      }
+
+      socket.on("disconnecting", (reason, description) => {
+        logger.log("Socket disconnecting", { ...getSocketMetadata(), reason, description });
+
+        if (socket.data.runFriendlyId) {
+          runDisconnected(socket.data.runFriendlyId);
+        }
+      });
+
+      socket.on("disconnect", (reason, description) => {
+        logger.log("Socket disconnected", { ...getSocketMetadata(), reason, description });
+      });
+
+      socket.on("error", (error) => {
+        logger.error("Socket error", {
+          ...getSocketMetadata(),
+          error: {
+            name: error.name,
+            message: error.message,
+            stack: error.stack,
+          },
+        });
+      });
+
+      socket.on("run:start", async (message) => {
+        const log = logger.child({
+          eventName: "run:start",
+          ...getSocketMetadata(),
+          ...message,
+        });
+
+        log.log("Handling run:start");
+
+        try {
+          runConnected(message.run.friendlyId);
+        } catch (error) {
+          log.error("run:start error", { error });
+        }
+      });
+
+      socket.on("run:stop", async (message) => {
+        const log = logger.child({
+          eventName: "run:stop",
+          ...getSocketMetadata(),
+          ...message,
+        });
+
+        log.log("Handling run:stop");
+
+        try {
+          runDisconnected(message.run.friendlyId);
+        } catch (error) {
+          log.error("run:stop error", { error });
+        }
+      });
+    });
+
+    return websocketServer;
+  }
+
+  notifyRun({ run }: { run: { friendlyId: string } }) {
+    try {
+      const runSocket = this.runSockets.get(run.friendlyId);
+
+      if (!runSocket) {
+        console.debug("[WorkloadServer] notifyRun: Run socket not found", { run });
+
+        this.workerClient.sendDebugLog(run.friendlyId, {
+          time: new Date(),
+          message: "run:notify socket not found on supervisor",
+        });
+
+        return;
+      }
+
+      runSocket.emit("run:notify", { version: "1", run });
+      console.debug("[WorkloadServer] run:notify sent", { run });
+
+      this.workerClient.sendDebugLog(run.friendlyId, {
+        time: new Date(),
+        message: "run:notify supervisor -> runner",
+      });
+    } catch (error) {
+      console.error("[WorkloadServer] Error in notifyRun", { run, error });
+
+      this.workerClient.sendDebugLog(run.friendlyId, {
+        time: new Date(),
+        message: "run:notify error on supervisor",
+      });
+    }
+  }
+
+  async start() {
+    await this.httpServer.start();
+  }
+
+  async stop() {
+    await this.httpServer.stop();
+  }
+}

apps/supervisor/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "../../.configs/tsconfig.base.json",
+  "include": ["src/**/*.ts"],
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "dist"
+  }
+}

apps/webapp/.eslintrc

@@ -1,17 +1,12 @@
 {
-  "plugins": [
-    "@trigger.dev/eslint-plugin",
-    "react-hooks",
-    "@typescript-eslint/eslint-plugin",
-    "import"
-  ],
+  "plugins": ["react-hooks", "@typescript-eslint/eslint-plugin", "import"],
   "parser": "@typescript-eslint/parser",
   "overrides": [
     {
       "files": ["*.ts", "*.tsx"],
       "rules": {
         // Autofixes imports from "@trigger.dev/core" to fine grained modules
-        "@trigger.dev/no-trigger-core-import": "error",
+        // "@trigger.dev/no-trigger-core-import": "error",
         // Normalize `import type {}` and `import { type }`
         "@typescript-eslint/consistent-type-imports": [
           "warn",

apps/webapp/app/api.server.ts

@@ -1,15 +0,0 @@
-import { ApiEventLog } from "@trigger.dev/core";
-import { EventRecord } from "@trigger.dev/database";
-
-export function eventRecordToApiJson(eventRecord: EventRecord): ApiEventLog {
-  return {
-    id: eventRecord.eventId,
-    name: eventRecord.name,
-    payload: eventRecord.payload as any,
-    context: eventRecord.context as any,
-    timestamp: eventRecord.timestamp,
-    deliverAt: eventRecord.deliverAt,
-    deliveredAt: eventRecord.deliveredAt,
-    cancelledAt: eventRecord.cancelledAt,
-  };
-}

apps/webapp/app/assets/icons/AnimatedHourglassIcon.tsx

@@ -0,0 +1,27 @@
+import { useAnimate } from "framer-motion";
+import { HourglassIcon } from "lucide-react";
+import { useEffect } from "react";
+
+export function AnimatedHourglassIcon({
+  className,
+  delay,
+}: {
+  className?: string;
+  delay?: number;
+}) {
+  const [scope, animate] = useAnimate();
+
+  useEffect(() => {
+    animate(
+      [
+        [scope.current, { rotate: 0 }, { duration: 0.7 }],
+        [scope.current, { rotate: 180 }, { duration: 0.3 }],
+        [scope.current, { rotate: 180 }, { duration: 0.7 }],
+        [scope.current, { rotate: 360 }, { duration: 0.3 }],
+      ],
+      { repeat: Infinity, delay }
+    );
+  }, []);
+
+  return <HourglassIcon ref={scope} className={className} />;
+}

apps/webapp/app/assets/icons/AttemptIcon.tsx

@@ -2,28 +2,18 @@ export function AttemptIcon({ className }: { className?: string }) {
   return (
     <svg
       className={className}
-      width="16"
-      height="16"
-      viewBox="0 0 16 16"
+      width="24"
+      height="24"
+      viewBox="0 0 24 24"
       fill="none"
       xmlns="http://www.w3.org/2000/svg"
     >
-      <g clipPath="url(#clip0_9964_113464)">
-        <path
-          fillRule="evenodd"
-          clipRule="evenodd"
-          d="M16 0H0V16H16V0ZM7.09906 4.4L4.53906 11.5H6.11906L6.63906 10H9.35906L9.87906 11.5H11.4591L8.89906 4.4H7.09906ZM7.99906 6L8.92906 8.73H7.06906L7.99906 6Z"
-          fill="currentColor"
-        />
-      </g>
-      <defs>
-        <clipPath id="clip0_9964_113464">
-          <path
-            d="M0 2C0 0.895431 0.895431 0 2 0H14C15.1046 0 16 0.895431 16 2V14C16 15.1046 15.1046 16 14 16H2C0.895431 16 0 15.1046 0 14V2Z"
-            fill="white"
-          />
-        </clipPath>
-      </defs>
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M5 2C3.34315 2 2 3.34315 2 5V19C2 20.6569 3.34315 22 5 22H19C20.6569 22 22 20.6569 22 19V5C22 3.34315 20.6569 2 19 2H5ZM7.65398 16.4367C7.64829 16.4871 7.64671 16.5374 7.64905 16.5873C7.66648 16.959 7.90169 17.3021 8.27281 17.4389C8.7645 17.6202 9.31042 17.3713 9.49607 16.8813L10.2088 15H13.8788L14.5915 16.8813C14.7772 17.3713 15.3231 17.6202 15.8148 17.4389C16.3096 17.2565 16.5628 16.7075 16.3804 16.2127L13.2478 7.71625C13.1255 7.30217 12.7423 7 12.2886 7H11.8041C11.5337 7 11.2884 7.10733 11.1084 7.28171C11.0015 7.38343 10.9157 7.51053 10.8613 7.65817L7.70722 16.2126C7.67996 16.2865 7.66243 16.3617 7.65398 16.4367ZM12.0438 10.1564L10.9665 13H13.1211L12.0438 10.1564Z"
+        fill="currentColor"
+      />
     </svg>
   );
 }

apps/webapp/app/assets/icons/ConnectionIcons.tsx

@@ -0,0 +1,73 @@
+export function ConnectedIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="0.5"
+        y="-0.5"
+        width="19"
+        height="1"
+        rx="0.5"
+        transform="matrix(1 0 0 -1 1.99998 20)"
+        stroke="#878C99"
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M14.4187 5L6.00002 5C4.89545 5 4.00002 5.89543 4.00002 7L4.00001 15C4.00002 16.1046 4.89545 17 6.00002 17L18 17C19.1046 17 20 16.1046 20 15L20 7.9816L18.5 9.4816L18.5 15C18.5 15.2761 18.2762 15.5 18 15.5L6.00001 15.5C5.72387 15.5 5.50001 15.2761 5.50001 15L5.50002 7C5.50002 6.72386 5.72387 6.5 6.00002 6.5L12.9187 6.5L14.4187 5Z"
+        fill="#878C99"
+      />
+      <path
+        d="M8.50002 9.75L11.5 12.75L18 6"
+        stroke="#28BF5C"
+        strokeWidth="1.5"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}
+
+export function DisconnectedIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M14.673 19L2.99998 19C2.44769 19 1.99998 19.4477 1.99998 20C1.99998 20.5523 2.44769 21 2.99998 21L16.673 21L14.673 19Z"
+        fill="#878C99"
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M3.99999 8.32703L3.99999 15C3.99999 16.1046 4.89542 17 5.99999 17L12.673 17L11.173 15.5L5.99999 15.5C5.72385 15.5 5.49999 15.2761 5.49999 15L5.49999 9.82703L3.99999 8.32703ZM18.5 14.2641L18.5 7C18.5 6.72386 18.2761 6.5 18 6.5L10.7358 6.5L9.23585 5L18 5C19.1046 5 20 5.89543 20 7L20 15C20 15.2292 19.9614 15.4495 19.8904 15.6546L18.5 14.2641Z"
+        fill="#878C99"
+      />
+      <path d="M3.00001 3L21 21" stroke="#E11D48" strokeWidth="1.5" strokeLinecap="round" />
+    </svg>
+  );
+}
+
+export function CheckingConnectionIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="0.5"
+        y="-0.5"
+        width="19"
+        height="1"
+        rx="0.5"
+        transform="matrix(1 0 0 -1 1.99998 20)"
+        stroke="#878C99"
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M18 6.5L6.00001 6.5C5.72387 6.5 5.50001 6.72386 5.50001 7L5.50001 15C5.50001 15.2761 5.72387 15.5 6.00001 15.5L18 15.5C18.2762 15.5 18.5 15.2761 18.5 15V7C18.5 6.72386 18.2762 6.5 18 6.5ZM6.00001 5C4.89545 5 4.00001 5.89543 4.00001 7L4.00001 15C4.00001 16.1046 4.89545 17 6.00001 17L18 17C19.1046 17 20 16.1046 20 15V7C20 5.89543 19.1046 5 18 5L6.00001 5Z"
+        fill="#878C99"
+      />
+      <circle cx="9" cy="11" r="1" fill="#878C99" />
+      <circle cx="12" cy="11" r="1" fill="#878C99" />
+      <circle cx="15" cy="11" r="1" fill="#878C99" />
+    </svg>
+  );
+}

apps/webapp/app/assets/icons/DropdownIcon.tsx

@@ -0,0 +1,20 @@
+export function DropdownIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M7.5 9L12 5L16.5 9"
+        stroke="currentColor"
+        strokeWidth="1.5"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M7.5 15L12 19L16.5 15"
+        stroke="currentColor"
+        strokeWidth="1.5"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}