Delete the old SSE util and updated the test endpoint to use the new one

* Use completedAt instead of updatedAt

* Disable verbose SSE logs

* Engine rate limiter: whitelist engine worker-actions

* Show the compute column in the runs table

* The compute column should be invocation + compute cost

.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/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/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

.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/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

@@ -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/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.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/unit-tests.yml

@@ -24,6 +24,13 @@ 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
 
@@ -31,7 +38,7 @@ jobs:
         run: pnpm run generate
 
       - name: 🧪 Run Webapp Unit Tests
-        run: pnpm run test --filter webapp
+        run: pnpm run test:webapp
         env:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
           DIRECT_URL: postgresql://postgres:postgres@localhost:5432/postgres
@@ -40,7 +47,7 @@ jobs:
           ENCRYPTION_KEY: "secret"
       
       - name: 🧪 Run Package Unit Tests
-        run: pnpm run test --filter "@trigger.dev/*"
+        run: pnpm run test:packages
       
       - name: 🧪 Run Internal Unit Tests
-        run: pnpm run test --filter "@internal/*"
+        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,3 +1,5 @@
 link-workspace-packages=false
 public-hoist-pattern[]=*prisma*
-prefer-workspace-packages=true
+prefer-workspace-packages=true
+update-notifier=false
+side-effects-cache=false

.vscode/launch.json

@@ -133,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)
 
@@ -20,12 +20,14 @@ Trigger.dev is an open source platform and SDK which allows you to create long-r
 ### 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)

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,32 +465,26 @@ export class Checkpointer {
       } catch (error) {
         this.#logger.error("Error during cleanup", { ...metadata, error });
       }
-
-      // Ensure only the current controller is removed
-      if (this.#abortControllers.get(runId) === controller) {
-        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,
         });
@@ -561,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
@@ -576,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/index.ts

@@ -11,10 +11,10 @@ import {
 } from "@trigger.dev/core/v3";
 import { ZodNamespace } from "@trigger.dev/core/v3/zodNamespace";
 import { ZodSocketConnection } from "@trigger.dev/core/v3/zodSocket";
-import { HttpReply, getTextBody } from "@trigger.dev/core/v3/apps";
+import { ExponentialBackoff, HttpReply, getTextBody } from "@trigger.dev/core/v3/apps";
 import { ChaosMonkey } from "./chaosMonkey";
 import { Checkpointer } from "./checkpointer";
-import { boolFromEnv, numFromEnv } from "./util";
+import { boolFromEnv, numFromEnv, safeJsonParse } from "./util";
 
 import { collectDefaultMetrics, register, Gauge } from "prom-client";
 import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
@@ -30,6 +30,16 @@ const PLATFORM_WS_PORT = process.env.PLATFORM_WS_PORT || 3030;
 const PLATFORM_SECRET = process.env.PLATFORM_SECRET || "coordinator-secret";
 const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?? "false");
 
+const TASK_RUN_COMPLETED_WITH_ACK_TIMEOUT_MS =
+  parseInt(process.env.TASK_RUN_COMPLETED_WITH_ACK_TIMEOUT_MS || "") || 30_000;
+const TASK_RUN_COMPLETED_WITH_ACK_MAX_RETRIES =
+  parseInt(process.env.TASK_RUN_COMPLETED_WITH_ACK_MAX_RETRIES || "") || 7;
+
+const WAIT_FOR_TASK_CHECKPOINT_DELAY_MS =
+  parseInt(process.env.WAIT_FOR_TASK_CHECKPOINT_DELAY_MS || "") || 0;
+const WAIT_FOR_BATCH_CHECKPOINT_DELAY_MS =
+  parseInt(process.env.WAIT_FOR_BATCH_CHECKPOINT_DELAY_MS || "") || 0;
+
 const logger = new SimpleStructuredLogger("coordinator", undefined, { nodeName: NODE_NAME });
 const chaosMonkey = new ChaosMonkey(
   !!process.env.CHAOS_MONKEY_ENABLED,
@@ -42,6 +52,8 @@ class CheckpointCancelError extends Error {}
 
 class TaskCoordinator {
   #httpServer: ReturnType<typeof createServer>;
+  #internalHttpServer: ReturnType<typeof createServer>;
+
   #checkpointer = new Checkpointer({
     dockerMode: !process.env.KUBERNETES_PORT,
     forceSimulate: boolFromEnv("FORCE_CHECKPOINT_SIMULATION", false),
@@ -79,6 +91,8 @@ class TaskCoordinator {
     private host = "0.0.0.0"
   ) {
     this.#httpServer = this.#createHttpServer();
+    this.#internalHttpServer = this.#createInternalHttpServer();
+
     this.#checkpointer.init();
     this.#platformSocket = this.#createPlatformSocket();
 
@@ -134,6 +148,7 @@ class TaskCoordinator {
       authToken: PLATFORM_SECRET,
       logHandlerPayloads: false,
       handlers: {
+        // This is used by resumeAttempt
         RESUME_AFTER_DEPENDENCY: async (message) => {
           const log = platformLogger.child({
             eventName: "RESUME_AFTER_DEPENDENCY",
@@ -159,11 +174,15 @@ class TaskCoordinator {
 
           await chaosMonkey.call();
 
-          // In case the task resumed faster than we could checkpoint
-          this.#cancelCheckpoint(message.runId);
+          // In case the task resumes before the checkpoint is created
+          this.#cancelCheckpoint(message.runId, {
+            event: "RESUME_AFTER_DEPENDENCY",
+            completions: message.completions.length,
+          });
 
           taskSocket.emit("RESUME_AFTER_DEPENDENCY", message);
         },
+        // This is used by sharedQueueConsumer
         RESUME_AFTER_DEPENDENCY_WITH_ACK: async (message) => {
           const log = platformLogger.child({
             eventName: "RESUME_AFTER_DEPENDENCY_WITH_ACK",
@@ -209,8 +228,11 @@ class TaskCoordinator {
 
           await chaosMonkey.call();
 
-          // In case the task resumed faster than we could checkpoint
-          this.#cancelCheckpoint(message.runId);
+          // In case the task resumes before the checkpoint is created
+          this.#cancelCheckpoint(message.runId, {
+            event: "RESUME_AFTER_DEPENDENCY_WITH_ACK",
+            completions: message.completions.length,
+          });
 
           taskSocket.emit("RESUME_AFTER_DEPENDENCY", message);
 
@@ -278,7 +300,7 @@ class TaskCoordinator {
           log.addFields({ socketId: taskSocket.id, socketData: taskSocket.data });
           log.log("Found task socket for REQUEST_RUN_CANCELLATION");
 
-          this.#cancelCheckpoint(message.runId);
+          this.#cancelCheckpoint(message.runId, { event: "REQUEST_RUN_CANCELLATION", ...message });
 
           if (message.delayInMs) {
             taskSocket.emit("REQUEST_EXIT", {
@@ -641,9 +663,25 @@ class TaskCoordinator {
 
             await chaosMonkey.call();
 
+            const lazyPayload = {
+              ...lazyAttempt.lazyPayload,
+              metrics: [
+                ...(message.startTime
+                  ? [
+                      {
+                        name: "start",
+                        event: "lazy_payload",
+                        timestamp: message.startTime,
+                        duration: Date.now() - message.startTime,
+                      },
+                    ]
+                  : []),
+              ],
+            };
+
             socket.emit("EXECUTE_TASK_RUN_LAZY_ATTEMPT", {
               version: "v1",
-              lazyPayload: lazyAttempt.lazyPayload,
+              lazyPayload,
             });
           } catch (error) {
             if (error instanceof ChaosMonkey.Error) {
@@ -653,11 +691,11 @@ class TaskCoordinator {
 
             log.error("READY_FOR_LAZY_ATTEMPT error", { error });
 
-            await crashRun({
-              name: "ReadyForLazyAttemptError",
-              message:
-                error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
-            });
+            // await crashRun({
+            //   name: "ReadyForLazyAttemptError",
+            //   message:
+            //     error instanceof Error ? `Unexpected error: ${error.message}` : "Unexpected error",
+            // });
 
             return;
           }
@@ -712,23 +750,91 @@ class TaskCoordinator {
             const { completion, execution } = message;
 
             // Cancel all in-progress checkpoints (if any)
-            this.#cancelCheckpoint(socket.data.runId);
+            this.#cancelCheckpoint(socket.data.runId, {
+              event: "TASK_RUN_COMPLETED",
+              attemptNumber: execution.attempt.number,
+            });
 
             await chaosMonkey.call({ throwErrors: false });
 
-            const completeWithoutCheckpoint = (shouldExit: boolean) => {
+            const sendCompletionWithAck = async (): Promise<boolean> => {
+              try {
+                const response = await this.#platformSocket?.sendWithAck(
+                  "TASK_RUN_COMPLETED_WITH_ACK",
+                  {
+                    version: "v2",
+                    execution,
+                    completion,
+                  },
+                  TASK_RUN_COMPLETED_WITH_ACK_TIMEOUT_MS
+                );
+
+                if (!response) {
+                  log.error("TASK_RUN_COMPLETED_WITH_ACK: no response");
+                  return false;
+                }
+
+                if (!response.success) {
+                  log.error("TASK_RUN_COMPLETED_WITH_ACK: error response", {
+                    error: response.error,
+                  });
+                  return false;
+                }
+
+                log.log("TASK_RUN_COMPLETED_WITH_ACK: successful response");
+                return true;
+              } catch (error) {
+                log.error("TASK_RUN_COMPLETED_WITH_ACK: threw error", { error });
+                return false;
+              }
+            };
+
+            const completeWithoutCheckpoint = async (shouldExit: boolean) => {
               const supportsRetryCheckpoints = message.version === "v1";
 
-              this.#platformSocket?.send("TASK_RUN_COMPLETED", {
-                version: supportsRetryCheckpoints ? "v1" : "v2",
-                execution,
-                completion,
-              });
               callback({ willCheckpointAndRestore: false, shouldExit });
+
+              if (supportsRetryCheckpoints) {
+                // This is only here for backwards compat
+                this.#platformSocket?.send("TASK_RUN_COMPLETED", {
+                  version: "v1",
+                  execution,
+                  completion,
+                });
+              } else {
+                // 99.99% of runs should end up here
+
+                const completedWithAckBackoff = new ExponentialBackoff("FullJitter").maxRetries(
+                  TASK_RUN_COMPLETED_WITH_ACK_MAX_RETRIES
+                );
+
+                const result = await completedWithAckBackoff.execute(
+                  async ({ retry, delay, elapsedMs }) => {
+                    logger.log("TASK_RUN_COMPLETED_WITH_ACK: sending with backoff", {
+                      retry,
+                      delay,
+                      elapsedMs,
+                    });
+
+                    const success = await sendCompletionWithAck();
+
+                    if (!success) {
+                      throw new Error("Failed to send completion with ack");
+                    }
+                  }
+                );
+
+                if (!result.success) {
+                  logger.error("TASK_RUN_COMPLETED_WITH_ACK: failed to send with backoff", result);
+                  return;
+                }
+
+                logger.log("TASK_RUN_COMPLETED_WITH_ACK: sent with backoff", result);
+              }
             };
 
             if (completion.ok) {
-              completeWithoutCheckpoint(true);
+              await completeWithoutCheckpoint(true);
               return;
             }
 
@@ -736,17 +842,17 @@ class TaskCoordinator {
               completion.error.type === "INTERNAL_ERROR" &&
               completion.error.code === "TASK_RUN_CANCELLED"
             ) {
-              completeWithoutCheckpoint(true);
+              await completeWithoutCheckpoint(true);
               return;
             }
 
             if (completion.retry === undefined) {
-              completeWithoutCheckpoint(true);
+              await completeWithoutCheckpoint(true);
               return;
             }
 
             if (completion.retry.delay < this.#delayThresholdInMs) {
-              completeWithoutCheckpoint(false);
+              await completeWithoutCheckpoint(false);
 
               // Prevents runs that fail fast from never sending a heartbeat
               this.#sendRunHeartbeat(socket.data.runId);
@@ -755,7 +861,7 @@ class TaskCoordinator {
             }
 
             if (message.version === "v2") {
-              completeWithoutCheckpoint(true);
+              await completeWithoutCheckpoint(true);
               return;
             }
 
@@ -764,7 +870,7 @@ class TaskCoordinator {
             const willCheckpointAndRestore = canCheckpoint || willSimulate;
 
             if (!willCheckpointAndRestore) {
-              completeWithoutCheckpoint(false);
+              await completeWithoutCheckpoint(false);
               return;
             }
 
@@ -788,7 +894,7 @@ class TaskCoordinator {
 
             if (!checkpoint) {
               log.error("Failed to checkpoint");
-              completeWithoutCheckpoint(false);
+              await completeWithoutCheckpoint(false);
               return;
             }
 
@@ -832,7 +938,10 @@ class TaskCoordinator {
 
           try {
             // Cancel all in-progress checkpoints (if any)
-            this.#cancelCheckpoint(socket.data.runId);
+            this.#cancelCheckpoint(socket.data.runId, {
+              event: "TASK_RUN_FAILED_TO_RUN",
+              errorType: completion.error.type,
+            });
 
             this.#platformSocket?.send("TASK_RUN_FAILED_TO_RUN", {
               version: "v1",
@@ -885,12 +994,15 @@ class TaskCoordinator {
 
           try {
             if (message.version === "v1") {
-              this.#cancelCheckpoint(socket.data.runId);
+              this.#cancelCheckpoint(socket.data.runId, { event: "CANCEL_CHECKPOINT", ...message });
               // v1 has no callback
               return;
             }
 
-            const checkpointCanceled = this.#cancelCheckpoint(socket.data.runId);
+            const checkpointCanceled = this.#cancelCheckpoint(socket.data.runId, {
+              event: "CANCEL_CHECKPOINT",
+              ...message,
+            });
 
             callback({ version: "v2", checkpointCanceled });
           } catch (error) {
@@ -934,11 +1046,14 @@ class TaskCoordinator {
               return;
             }
 
+            const runId = socket.data.runId;
+            const attemptNumber = getAttemptNumber();
+
             const checkpoint = await this.#checkpointer.checkpointAndPush({
-              runId: socket.data.runId,
+              runId,
               projectRef: socket.data.projectRef,
               deploymentVersion: socket.data.deploymentVersion,
-              attemptNumber: getAttemptNumber(),
+              attemptNumber,
             });
 
             if (!checkpoint) {
@@ -964,6 +1079,13 @@ class TaskCoordinator {
 
             if (ack?.keepRunAlive) {
               log.log("keeping run alive after duration checkpoint");
+
+              if (checkpoint.docker && willSimulate) {
+                // The container is still paused so we need to unpause it
+                log.log("unpausing container after duration checkpoint");
+                this.#checkpointer.unpause(runId, attemptNumber);
+              }
+
               return;
             }
 
@@ -1022,12 +1144,18 @@ class TaskCoordinator {
               }
             }
 
-            const checkpoint = await this.#checkpointer.checkpointAndPush({
-              runId: socket.data.runId,
-              projectRef: socket.data.projectRef,
-              deploymentVersion: socket.data.deploymentVersion,
-              attemptNumber: getAttemptNumber(),
-            });
+            const runId = socket.data.runId;
+            const attemptNumber = getAttemptNumber();
+
+            const checkpoint = await this.#checkpointer.checkpointAndPush(
+              {
+                runId,
+                projectRef: socket.data.projectRef,
+                deploymentVersion: socket.data.deploymentVersion,
+                attemptNumber,
+              },
+              WAIT_FOR_TASK_CHECKPOINT_DELAY_MS
+            );
 
             if (!checkpoint) {
               log.error("Failed to checkpoint");
@@ -1057,6 +1185,13 @@ class TaskCoordinator {
             if (ack?.keepRunAlive) {
               socket.data.requiresCheckpointResumeWithMessage = undefined;
               log.log("keeping run alive after task checkpoint");
+
+              if (checkpoint.docker && willSimulate) {
+                // The container is still paused so we need to unpause it
+                log.log("unpausing container after duration checkpoint");
+                this.#checkpointer.unpause(runId, attemptNumber);
+              }
+
               return;
             }
 
@@ -1115,12 +1250,18 @@ class TaskCoordinator {
               }
             }
 
-            const checkpoint = await this.#checkpointer.checkpointAndPush({
-              runId: socket.data.runId,
-              projectRef: socket.data.projectRef,
-              deploymentVersion: socket.data.deploymentVersion,
-              attemptNumber: getAttemptNumber(),
-            });
+            const runId = socket.data.runId;
+            const attemptNumber = getAttemptNumber();
+
+            const checkpoint = await this.#checkpointer.checkpointAndPush(
+              {
+                runId,
+                projectRef: socket.data.projectRef,
+                deploymentVersion: socket.data.deploymentVersion,
+                attemptNumber,
+              },
+              WAIT_FOR_BATCH_CHECKPOINT_DELAY_MS
+            );
 
             if (!checkpoint) {
               log.error("Failed to checkpoint");
@@ -1151,6 +1292,13 @@ class TaskCoordinator {
             if (ack?.keepRunAlive) {
               socket.data.requiresCheckpointResumeWithMessage = undefined;
               log.log("keeping run alive after batch checkpoint");
+
+              if (checkpoint.docker && willSimulate) {
+                // The container is still paused so we need to unpause it
+                log.log("unpausing container after batch checkpoint");
+                this.#checkpointer.unpause(runId, attemptNumber);
+              }
+
               return;
             }
 
@@ -1338,7 +1486,9 @@ class TaskCoordinator {
     });
   }
 
-  #cancelCheckpoint(runId: string): boolean {
+  #cancelCheckpoint(runId: string, reason?: any): boolean {
+    logger.log("cancelCheckpoint: call", { runId, reason });
+
     const checkpointWait = this.#checkpointableTasks.get(runId);
 
     if (checkpointWait) {
@@ -1347,9 +1497,14 @@ class TaskCoordinator {
     }
 
     // Cancel checkpointing procedure
-    const checkpointCanceled = this.#checkpointer.cancelCheckpoint(runId);
+    const checkpointCanceled = this.#checkpointer.cancelAllCheckpointsForRun(runId);
 
-    logger.log("cancelCheckpoint()", { runId, checkpointCanceled });
+    logger.log("cancelCheckpoint: result", {
+      runId,
+      reason,
+      checkpointCanceled,
+      hadCheckpointWait: !!checkpointWait,
+    });
 
     return checkpointCanceled;
   }
@@ -1368,14 +1523,6 @@ class TaskCoordinator {
         case "/metrics": {
           return reply.text(await register.metrics(), 200, register.contentType);
         }
-        case "/whoami": {
-          return reply.text(NODE_NAME);
-        }
-        case "/checkpoint": {
-          const body = await getTextBody(req);
-          // await this.#checkpointer.checkpointAndPush(body);
-          return reply.text(`sent restore request: ${body}`);
-        }
         default: {
           return reply.empty(404);
         }
@@ -1393,8 +1540,240 @@ class TaskCoordinator {
     return httpServer;
   }
 
+  #createInternalHttpServer() {
+    const httpServer = createServer(async (req, res) => {
+      logger.log(`[${req.method}]`, { url: req.url });
+
+      const reply = new HttpReply(res);
+
+      switch (req.url) {
+        case "/whoami": {
+          return reply.text(NODE_NAME);
+        }
+        case "/checkpoint/duration": {
+          try {
+            const body = await getTextBody(req);
+            const json = safeJsonParse(body);
+
+            if (typeof json !== "object" || !json) {
+              return reply.text("Invalid body", 400);
+            }
+
+            if (!("runId" in json) || typeof json.runId !== "string") {
+              return reply.text("Missing or invalid: runId", 400);
+            }
+
+            if (!("now" in json) || typeof json.now !== "number") {
+              return reply.text("Missing or invalid: now", 400);
+            }
+
+            if (!("ms" in json) || typeof json.ms !== "number") {
+              return reply.text("Missing or invalid: ms", 400);
+            }
+
+            let keepRunAlive = false;
+            if ("keepRunAlive" in json && typeof json.keepRunAlive === "boolean") {
+              keepRunAlive = json.keepRunAlive;
+            }
+
+            let async = false;
+            if ("async" in json && typeof json.async === "boolean") {
+              async = json.async;
+            }
+
+            const { runId, now, ms } = json;
+
+            if (!runId) {
+              return reply.text("Missing runId", 400);
+            }
+
+            const runSocket = await this.#getRunSocket(runId);
+            if (!runSocket) {
+              return reply.text("Run socket not found", 404);
+            }
+
+            const { data } = runSocket;
+
+            console.log("Manual duration checkpoint", data);
+
+            if (async) {
+              reply.text("Creating checkpoint in the background", 202);
+            }
+
+            const checkpoint = await this.#checkpointer.checkpointAndPush({
+              runId: data.runId,
+              projectRef: data.projectRef,
+              deploymentVersion: data.deploymentVersion,
+              attemptNumber: data.attemptNumber ? parseInt(data.attemptNumber) : undefined,
+            });
+
+            if (!checkpoint) {
+              return reply.text("Failed to checkpoint", 500);
+            }
+
+            if (!data.attemptFriendlyId) {
+              return reply.text("Socket data missing attemptFriendlyId", 500);
+            }
+
+            const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
+              version: "v1",
+              runId,
+              attemptFriendlyId: data.attemptFriendlyId,
+              docker: checkpoint.docker,
+              location: checkpoint.location,
+              reason: {
+                type: "WAIT_FOR_DURATION",
+                ms,
+                now,
+              },
+            });
+
+            if (ack?.keepRunAlive || keepRunAlive) {
+              return reply.json({
+                message: `keeping run ${runId} alive after checkpoint`,
+                checkpoint,
+                requestJson: json,
+                platformAck: ack,
+              });
+            }
+
+            runSocket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+
+            return reply.json({
+              message: `checkpoint created for run ${runId}`,
+              checkpoint,
+              requestJson: json,
+              platformAck: ack,
+            });
+          } catch (error) {
+            return reply.json({
+              message: `error`,
+              error,
+            });
+          }
+        }
+        case "/checkpoint/manual": {
+          try {
+            const body = await getTextBody(req);
+            const json = safeJsonParse(body);
+
+            if (typeof json !== "object" || !json) {
+              return reply.text("Invalid body", 400);
+            }
+
+            if (!("runId" in json) || typeof json.runId !== "string") {
+              return reply.text("Missing or invalid: runId", 400);
+            }
+
+            let restoreAtUnixTimeMs: number | undefined;
+            if ("restoreAtUnixTimeMs" in json && typeof json.restoreAtUnixTimeMs === "number") {
+              restoreAtUnixTimeMs = json.restoreAtUnixTimeMs;
+            }
+
+            let keepRunAlive = false;
+            if ("keepRunAlive" in json && typeof json.keepRunAlive === "boolean") {
+              keepRunAlive = json.keepRunAlive;
+            }
+
+            let async = false;
+            if ("async" in json && typeof json.async === "boolean") {
+              async = json.async;
+            }
+
+            const { runId } = json;
+
+            if (!runId) {
+              return reply.text("Missing runId", 400);
+            }
+
+            const runSocket = await this.#getRunSocket(runId);
+            if (!runSocket) {
+              return reply.text("Run socket not found", 404);
+            }
+
+            const { data } = runSocket;
+
+            console.log("Manual checkpoint", data);
+
+            if (async) {
+              reply.text("Creating checkpoint in the background", 202);
+            }
+
+            const checkpoint = await this.#checkpointer.checkpointAndPush({
+              runId: data.runId,
+              projectRef: data.projectRef,
+              deploymentVersion: data.deploymentVersion,
+              attemptNumber: data.attemptNumber ? parseInt(data.attemptNumber) : undefined,
+            });
+
+            if (!checkpoint) {
+              return reply.text("Failed to checkpoint", 500);
+            }
+
+            if (!data.attemptFriendlyId) {
+              return reply.text("Socket data missing attemptFriendlyId", 500);
+            }
+
+            const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
+              version: "v1",
+              runId,
+              attemptFriendlyId: data.attemptFriendlyId,
+              docker: checkpoint.docker,
+              location: checkpoint.location,
+              reason: {
+                type: "MANUAL",
+                restoreAtUnixTimeMs,
+              },
+            });
+
+            if (ack?.keepRunAlive || keepRunAlive) {
+              return reply.json({
+                message: `keeping run ${runId} alive after checkpoint`,
+                checkpoint,
+                requestJson: json,
+                platformAck: ack,
+              });
+            }
+
+            runSocket.emit("REQUEST_EXIT", {
+              version: "v1",
+            });
+
+            return reply.json({
+              message: `checkpoint created for run ${runId}`,
+              checkpoint,
+              requestJson: json,
+              platformAck: ack,
+            });
+          } catch (error) {
+            return reply.json({
+              message: `error`,
+              error,
+            });
+          }
+        }
+        default: {
+          return reply.empty(404);
+        }
+      }
+    });
+
+    httpServer.on("clientError", (err, socket) => {
+      socket.end("HTTP/1.1 400 Bad Request\r\n\r\n");
+    });
+
+    httpServer.on("listening", () => {
+      logger.log("internal server listening on port", { port: HTTP_SERVER_PORT + 100 });
+    });
+
+    return httpServer;
+  }
+
   listen() {
     this.#httpServer.listen(this.port, this.host);
+    this.#internalHttpServer.listen(this.port + 100, "127.0.0.1");
   }
 }
 

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";
@@ -37,7 +39,14 @@ const UPTIME_MAX_PENDING_ERRORS = Number(process.env.UPTIME_MAX_PENDING_ERRORS |
 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`);
@@ -65,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;
@@ -103,7 +114,7 @@ class KubernetesTaskOperations implements TaskOperations {
               containers: [
                 {
                   name: this.#getIndexContainerName(opts.shortCode),
-                  image: opts.imageRef,
+                  image: getImageRef("deployment", opts.imageRef),
                   ports: [
                     {
                       containerPort: 8000,
@@ -157,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",
@@ -170,7 +182,7 @@ class KubernetesTaskOperations implements TaskOperations {
           containers: [
             {
               name: containerName,
-              image: opts.image,
+              image: getImageRef("deployment", opts.image),
               ports: [
                 {
                   containerPort: 8000,
@@ -190,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: [
                 {
@@ -218,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",
@@ -231,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"],
@@ -252,7 +268,7 @@ class KubernetesTaskOperations implements TaskOperations {
           containers: [
             {
               name: this.#getRunContainerName(opts.runId),
-              image: opts.checkpointRef,
+              image: getImageRef("restore", opts.checkpointRef),
               ports: [
                 {
                   containerPort: 8000,
@@ -358,7 +374,7 @@ class KubernetesTaskOperations implements TaskOperations {
               initContainers: [
                 {
                   name: "prepull",
-                  image: opts.imageRef,
+                  image: getImageRef("deployment", opts.imageRef),
                   command: ["/usr/bin/true"],
                   resources: {
                     limits: {
@@ -372,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",
@@ -403,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",
       },
@@ -501,6 +520,10 @@ class KubernetesTaskOperations implements TaskOperations {
           },
         },
       },
+      {
+        name: "TRIGGER_POD_SCHEDULED_AT_MS",
+        value: Date.now().toString(),
+      },
       ...this.#coordinatorEnvVars,
     ];
   }
@@ -673,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,

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/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,22 +0,0 @@
-{
-  "name": "proxy",
-  "version": "0.0.11",
-  "private": true,
-  "scripts": {
-    "deploy": "wrangler deploy",
-    "dev": "wrangler dev",
-    "dry-run:staging": "wrangler deploy --dry-run --outdir=dist --env staging"
-  },
-  "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.23.8",
-    "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,53 +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 (!queueingIsEnabled(env)) {
-      console.log("Missing AWS credentials. Passing through to the origin.");
-      return fetch(request);
-    }
-
-    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 fetch(request);
-  },
-};
-
-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>
+  );
+}

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

@@ -0,0 +1,153 @@
+export function DevEnvironmentIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <circle cx="12" cy="12" r="3" fill="currentColor" />
+      <path
+        d="M8.35714 3H5C3.89543 3 3 3.89543 3 5V8.35714"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M21 8.35714L21 5C21 3.89543 20.1046 3 19 3L15.6429 3"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M15.6429 21L19 21C20.1046 21 21 20.1046 21 19L21 15.6429"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M3 15.6429L3 19C3 20.1046 3.89543 21 5 21L8.35714 21"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}
+
+export function DevEnvironmentIconSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <circle cx="10" cy="10" r="2" fill="currentColor" />
+      <path
+        d="M7 3H5C3.89543 3 3 3.89543 3 5V7"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M17 7L17 5C17 3.89543 16.1046 3 15 3L13 3"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M13 17L15 17C16.1046 17 17 16.1046 17 15L17 13"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M3 13L3 15C3 16.1046 3.89543 17 5 17L7 17"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}
+
+export function ProdEnvironmentIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M12.5037 7.33603C12.3174 6.88799 11.6827 6.88799 11.4963 7.33603L10.4338 9.89064L7.6759 10.1117C7.1922 10.1505 6.99606 10.7542 7.36459 11.0698L9.46583 12.8698L8.82387 15.561C8.71128 16.033 9.22477 16.4061 9.63888 16.1532L12 14.711L14.3612 16.1532C14.7753 16.4061 15.2888 16.0331 15.1762 15.561L14.5343 12.8698L16.6355 11.0698C17.004 10.7542 16.8079 10.1505 16.3242 10.1117L13.5663 9.89064L12.5037 7.33603Z"
+        fill="white"
+      />
+      <rect
+        x="3"
+        y="3"
+        width="18"
+        height="18"
+        rx="2"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}
+
+export function ProdEnvironmentIconSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="3"
+        y="3"
+        width="14"
+        height="14"
+        rx="3"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+      <path
+        d="M10.4174 6.23514C10.263 5.86384 9.73701 5.86384 9.58258 6.23514L8.70207 8.35213L6.4166 8.53536C6.01575 8.56749 5.85322 9.06773 6.15862 9.32934L7.89991 10.8209L7.36792 13.0512C7.27461 13.4423 7.70014 13.7515 8.04332 13.5419L10 12.3467L11.9567 13.5419C12.2999 13.7515 12.7254 13.4423 12.6321 13.0512L12.1001 10.8209L13.8414 9.32934C14.1468 9.06773 13.9842 8.56749 13.5834 8.53536L11.2979 8.35213L10.4174 6.23514Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}
+
+export function DeployedEnvironmentIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <circle cx="12" cy="12" r="3" fill="currentColor" />
+      <rect
+        x="3"
+        y="3"
+        width="18"
+        height="18"
+        rx="2"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}
+
+export function DeployedEnvironmentIconSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <circle cx="10" cy="10" r="2" fill="currentColor" />
+      <rect
+        x="3"
+        y="3"
+        width="14"
+        height="14"
+        rx="3"
+        stroke="currentColor"
+        strokeWidth="2"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}

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

@@ -0,0 +1,21 @@
+export function FunctionIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="2.5"
+        y="2.5"
+        width="19"
+        height="19"
+        rx="2.5"
+        stroke="currentColor"
+        strokeOpacity={0.5}
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M14.9987 7.99477C13.8955 7.99477 13.0012 8.88907 13.0012 9.99225V10.6649H14.3363H15.3375V12.6675H14.3363H13.0012V14.0084C13.0012 16.2171 11.2107 18.0077 9.0019 18.0077H8.00061V16.0051H9.0019C10.1047 16.0051 10.9986 15.1111 10.9986 14.0084V12.6675H9.66357H8.66228V10.6649H9.66357H10.9986V9.99225C10.9986 7.78308 12.7895 5.99219 14.9987 5.99219H16V7.99477H14.9987Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}

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

@@ -0,0 +1,21 @@
+export function MiddlewareIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="2.5"
+        y="2.5"
+        width="19"
+        height="19"
+        rx="2.5"
+        stroke="currentColor"
+        strokeOpacity={0.5}
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M8.72775 14.8717V11.1087C8.72775 10.4761 9.24051 9.96338 9.87303 9.96338C10.5056 9.96338 11.0183 10.4761 11.0183 11.1087V15.5262V16.5079H12.9817V15.5262V11.1087C12.9817 10.4761 13.4944 9.96334 14.1269 9.96334C14.7594 9.96334 15.2722 10.4761 15.2722 11.1086V15.5262V16.5079H17.2356V15.5262V11.1086C17.2356 9.39177 15.8438 8 14.1269 8C13.3041 8 12.556 8.31966 12 8.84155C11.4439 8.31968 10.6958 8.00004 9.87303 8.00004C9.46858 8.00004 9.08217 8.07728 8.72775 8.21781V8.00004H6.7644V8.98171V11.1087V14.8717V15.5262V15.8534V16.5079H8.72775V15.8534V15.5262V14.8717Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}

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

@@ -0,0 +1,19 @@
+export function PauseIcon({ className }: { className?: string }) {
+  return (
+    <svg
+      className={className}
+      width="24"
+      height="24"
+      viewBox="0 0 24 24"
+      fill="none"
+      xmlns="http://www.w3.org/2000/svg"
+    >
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M2 12C2 6.47715 6.47715 2 12 2C17.5228 2 22 6.47715 22 12C22 17.5228 17.5228 22 12 22C6.47715 22 2 17.5228 2 12ZM8 8.25C8 7.55964 8.55964 7 9.25 7C9.94036 7 10.5 7.55964 10.5 8.25V15.75C10.5 16.4404 9.94036 17 9.25 17C8.55964 17 8 16.4404 8 15.75V8.25ZM14.75 7C14.0596 7 13.5 7.55964 13.5 8.25V15.75C13.5 16.4404 14.0596 17 14.75 17C15.4404 17 16 16.4404 16 15.75V8.25C16 7.55964 15.4404 7 14.75 7Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}

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

@@ -0,0 +1,24 @@
+export function PromoteIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        d="M9.5 15.75H5.25493C4.4265 15.75 3.75493 16.4216 3.75493 17.25V19.741C3.75493 20.5695 4.4265 21.241 5.25493 21.241H18.7582C19.5866 21.241 20.2582 20.5695 20.2582 19.741V17.25C20.2582 16.4216 19.5866 15.75 18.7582 15.75H14.5"
+        stroke="currentColor"
+        strokeWidth="1.5"
+      />
+      <path
+        d="M9.5 7.75H5.25493C4.4265 7.75 3.75493 8.42157 3.75493 9.25V11.741C3.75493 12.5695 4.4265 13.241 5.25493 13.241H9.5M14.5 7.75002H18.7582C19.5866 7.75002 20.2582 8.42159 20.2582 9.25002V11.741C20.2582 12.5695 19.5866 13.241 18.7582 13.241H14.5"
+        stroke="currentColor"
+        strokeWidth="1.5"
+      />
+      <path d="M12 18L12 3" stroke="currentColor" strokeWidth="1.5" strokeLinecap="round" />
+      <path
+        d="M9 5.22892L12 2.75L15 5.22892"
+        stroke="currentColor"
+        strokeWidth="1.5"
+        strokeLinecap="round"
+        strokeLinejoin="round"
+      />
+    </svg>
+  );
+}

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

@@ -0,0 +1,21 @@
+export function PythonLogoIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <g clipPath="url(#clip0_16722_101857)">
+        <path
+          d="M11.9281 2C6.85063 2 7.1675 4.20188 7.1675 4.20188L7.17375 6.48313H12.0187V7.1675H5.24813C5.24813 7.1675 2 6.79875 2 11.9225C2 17.045 4.83563 16.8638 4.83563 16.8638H6.52813V14.4863C6.52813 14.4863 6.43687 11.6506 9.31875 11.6506H14.1237C14.1237 11.6506 16.8237 11.6944 16.8237 9.04125V4.65438C16.8237 4.65438 17.2338 2 11.9281 2ZM9.25625 3.53375C9.73813 3.53375 10.1281 3.92375 10.1281 4.40563C10.1281 4.8875 9.73813 5.2775 9.25625 5.2775C9.14171 5.27766 9.02826 5.25523 8.9224 5.21147C8.81655 5.16771 8.72037 5.10349 8.63937 5.0225C8.55838 4.94151 8.49417 4.84533 8.45041 4.73947C8.40665 4.63362 8.38421 4.52017 8.38438 4.40563C8.38438 3.92375 8.77438 3.53375 9.25625 3.53375Z"
+          fill="#387EB8"
+        />
+        <path
+          d="M12.0719 21.9839C17.1494 21.9839 16.8325 19.782 16.8325 19.782L16.8263 17.5008H11.9813V16.8164H18.7513C18.7513 16.8164 22 17.1852 22 12.062C22 6.93892 19.1644 7.12079 19.1644 7.12079H17.4719V9.49767C17.4719 9.49767 17.5631 12.3333 14.6813 12.3333H9.87626C9.87626 12.3333 7.17626 12.2895 7.17626 14.9427V19.3295C7.17626 19.3295 6.76626 21.9839 12.0719 21.9839ZM14.7438 20.4502C14.6292 20.4503 14.5158 20.4279 14.4099 20.3841C14.3041 20.3404 14.2079 20.2762 14.1269 20.1952C14.0459 20.1142 13.9817 20.018 13.9379 19.9121C13.8942 19.8063 13.8717 19.6928 13.8719 19.5783C13.8719 19.097 14.2619 18.707 14.7438 18.707C15.2256 18.707 15.6156 19.0964 15.6156 19.5783C15.6156 20.0608 15.2256 20.4502 14.7438 20.4502Z"
+          fill="#F4D441"
+        />
+      </g>
+      <defs>
+        <clipPath id="clip0_16722_101857">
+          <rect width="20" height="20" fill="white" transform="translate(2 2)" />
+        </clipPath>
+      </defs>
+    </svg>
+  );
+}

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

@@ -0,0 +1,21 @@
+export function RunFunctionIcon({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <rect
+        x="2.5"
+        y="2.5"
+        width="19"
+        height="19"
+        rx="2.5"
+        stroke="currentColor"
+        strokeOpacity={0.5}
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M12 8H10V12V16.5159H12V12C12 10.8954 12.8954 10 14 10H15.5V8H14C13.2714 8 12.5883 8.19479 12 8.53513V8Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}

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

@@ -2,15 +2,53 @@ export function RunsIcon({ className }: { className?: string }) {
   return (
     <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
       <path
-        d="M12.2964 3.66962C11.97 3.83662 11.6961 4.09048 11.5047 4.40323C11.3134 4.71598 11.2121 5.07547 11.212 5.44211L11.211 8.0162L9.22047 8.0172C8.95637 8.0172 8.70309 8.12211 8.51635 8.30886C8.3296 8.4956 8.22469 8.74888 8.22469 9.01298V14.9877L8.23166 15.1042C8.26019 15.3464 8.37664 15.5697 8.5589 15.7318C8.74116 15.8939 8.97657 15.9834 9.22047 15.9834L11.211 15.9825L11.212 18.5585C11.2121 18.9524 11.329 19.3373 11.5478 19.6648C11.7666 19.9922 12.0777 20.2474 12.4415 20.3981C12.8054 20.5488 13.2057 20.5882 13.592 20.5114C13.9783 20.4346 14.3331 20.245 14.6116 19.9666L21.1698 13.4084C21.5432 13.0349 21.7529 12.5284 21.7529 12.0003C21.7529 11.4722 21.5432 10.9658 21.1698 10.5923L14.6116 4.03408C14.3331 3.75539 13.9782 3.56558 13.5918 3.48865C13.2054 3.41172 12.8048 3.45114 12.4408 3.60191L12.2964 3.66962Z"
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M21.4075 8.80505C21.9499 8.40538 21.9499 7.59459 21.4075 7.19493L14.5932 2.17391C13.9329 1.68741 13 2.15883 13 2.97897V5.99999L4.5 5.99999C4.22386 5.99999 4 6.22385 4 6.49999L4 9.49999C4 9.77613 4.22386 9.99999 4.5 9.99999L13 9.99999V13.021C13 13.8411 13.9329 14.3126 14.5932 13.8261L21.4075 8.80505Z"
         fill="currentColor"
       />
       <path
-        d="M3.24578 8.0172C3.48968 8.01723 3.72509 8.10677 3.90735 8.26885C4.08961 8.43092 4.20606 8.65425 4.23459 8.89647L4.24156 9.01298V14.9877C4.24128 15.2415 4.1441 15.4856 3.96986 15.6701C3.79563 15.8547 3.55751 15.9658 3.30414 15.9806C3.05077 15.9955 2.80128 15.9131 2.60666 15.7502C2.41203 15.5873 2.28695 15.3562 2.25697 15.1042L2.25 14.9877V9.01298C2.25 8.74888 2.35491 8.4956 2.54166 8.30885C2.7284 8.12211 2.98168 8.0172 3.24578 8.0172Z"
-        fill="currentColor"
-      />
-      <path
-        d="M6.23312 8.0172C6.47702 8.01723 6.71243 8.10677 6.89469 8.26885C7.07696 8.43092 7.1934 8.65425 7.22193 8.89647L7.2289 9.01298V14.9877C7.22862 15.2415 7.13144 15.4856 6.95721 15.6701C6.78298 15.8547 6.54485 15.9658 6.29148 15.9806C6.03811 15.9955 5.78863 15.9131 5.594 15.7502C5.39937 15.5873 5.27429 15.3562 5.24431 15.1042L5.23734 14.9877V9.01298C5.23734 8.74888 5.34226 8.4956 5.529 8.30885C5.71575 8.12211 5.96903 8.0172 6.23312 8.0172Z"
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M12.2494 17.5898C12.6316 17.2895 12.6316 16.7106 12.2494 16.4103L7.21336 12.4534C6.72127 12.0668 6 12.4173 6 13.0431L6 15.5L2.5 15.5C2.22386 15.5 2 15.7239 2 16V18C2 18.2762 2.22386 18.5 2.5 18.5L6 18.5L6 20.9569C6 21.5828 6.72127 21.9333 7.21336 21.5467L12.2494 17.5898Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}
+
+export function RunsIconSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M17.6527 7.18006C18.145 6.86034 18.145 6.1396 17.6527 5.81987L12.2525 2.31329C11.713 1.96299 11 2.35016 11 2.99339L11 4.99998L4.40545 4.99998C4.18153 4.99998 4 5.1815 4 5.40543V7.59453C4 7.81845 4.18153 7.99998 4.40545 7.99998L11 7.99998L11 10.0065C11 10.6498 11.713 11.037 12.2525 10.6866L17.6527 7.18006Z"
+        fill="currentColor"
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M10.4064 14.4748C10.7107 14.2314 10.7107 13.7685 10.4064 13.525L6.98808 10.7904C6.58987 10.4718 5.99998 10.7554 5.99998 11.2653L5.99998 13L2.40539 13C2.18147 13 1.99994 13.1815 1.99994 13.4054V14.5945C1.99994 14.8185 2.18147 15 2.40539 15L5.99998 15L5.99999 16.7346C5.99999 17.2445 6.58987 17.528 6.98808 17.2095L10.4064 14.4748Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}
+
+export function RunsIconExtraSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 18 18" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M15.6332 6.60006C16.0524 6.30985 16.0525 5.69018 15.6333 5.39997L11.1452 2.29277C10.6612 1.95767 9.99998 2.3041 9.99998 2.89281L9.99998 4.5L4.3649 4.5C4.16337 4.5 4 4.66337 4 4.8649V7.13509C4 7.33663 4.16337 7.5 4.3649 7.5L9.99998 7.5V9.10703C9.99998 9.69574 10.6612 10.0422 11.1452 9.70708L15.6332 6.60006Z"
+        fill="currentColor"
+      />
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M9.46572 12.9274C9.73962 12.7083 9.73961 12.2917 9.46572 12.0726L5.88926 9.21142C5.53087 8.92471 4.99998 9.17988 4.99998 9.63884V11.25L2.3649 11.25C2.16337 11.25 2 11.4134 2 11.6149V13.3851C2 13.5866 2.16337 13.75 2.3649 13.75L4.99998 13.75V15.3612C4.99998 15.8202 5.53088 16.0754 5.88927 15.7887L9.46572 12.9274Z"
         fill="currentColor"
       />
     </svg>

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

@@ -0,0 +1,9 @@
+import { cn } from "~/utils/cn";
+
+export function StatusIcon({ className }: { className?: string }) {
+  return (
+    <div className={cn("grid place-items-center", className)}>
+      <div className="size-[75%] rounded-full border-2 border-text-dimmed" />
+    </div>
+  );
+}

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

@@ -0,0 +1,49 @@
+export function TaskCachedIcon({ className }: { className?: string }) {
+  return (
+    <svg
+      className={className}
+      width="16"
+      height="16"
+      viewBox="0 0 16 16"
+      fill="none"
+      xmlns="http://www.w3.org/2000/svg"
+    >
+      <g clipPath="url(#clip0_15584_76102)">
+        <path
+          d="M0.5 3.5L0.5 2.5C0.5 1.39543 1.39543 0.5 2.5 0.5H3.5"
+          stroke="#3B82F6"
+          strokeLinecap="square"
+          strokeLinejoin="round"
+        />
+        <path
+          d="M15.5 12.5L15.5 13.5C15.5 14.6046 14.6046 15.5 13.5 15.5L12.5 15.5"
+          stroke="#3B82F6"
+          strokeLinecap="square"
+          strokeLinejoin="round"
+        />
+        <path
+          d="M12.5 0.5L13.5 0.5C14.6046 0.5 15.5 1.39543 15.5 2.5L15.5 3.5"
+          stroke="#3B82F6"
+          strokeLinecap="square"
+          strokeLinejoin="round"
+        />
+        <path
+          d="M3.5 15.5L2.5 15.5C1.39543 15.5 0.5 14.6046 0.5 13.5L0.5 12.5"
+          stroke="#3B82F6"
+          strokeLinecap="square"
+          strokeLinejoin="round"
+        />
+        <path d="M11.1799 4.19V5.598H8.8479V12H7.1649V5.598H4.8219V4.19H11.1799Z" fill="#3B82F6" />
+        <line x1="6" y1="15.5" x2="10" y2="15.5" stroke="#3B82F6" />
+        <line x1="6" y1="0.5" x2="10" y2="0.5" stroke="#3B82F6" />
+        <line x1="15.5" y1="6" x2="15.5" y2="10" stroke="#3B82F6" />
+        <line x1="0.5" y1="6" x2="0.5" y2="10" stroke="#3B82F6" />
+      </g>
+      <defs>
+        <clipPath id="clip0_15584_76102">
+          <rect width="16" height="16" fill="white" />
+        </clipPath>
+      </defs>
+    </svg>
+  );
+}

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

@@ -1,29 +1,25 @@
 export function TaskIcon({ className }: { className?: string }) {
   return (
-    <svg
-      className={className}
-      width="16"
-      height="16"
-      viewBox="0 0 16 16"
-      fill="none"
-      xmlns="http://www.w3.org/2000/svg"
-    >
-      <g clipPath="url(#clip0_9221_99732)">
-        <path
-          fillRule="evenodd"
-          clipRule="evenodd"
-          d="M16 0H0V16H16V0ZM10.8901 5.73995V4.44995H5.11011V5.73995H7.23011V11.55H8.77011V5.73995H10.8901Z"
-          fill="currentColor"
-        />
-      </g>
-      <defs>
-        <clipPath id="clip0_9221_99732">
-          <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>
+    <svg className={className} viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M2 5C2 3.34315 3.34315 2 5 2H19C20.6569 2 22 3.34315 22 5V19C22 20.6569 20.6569 22 19 22H5C3.34315 22 2 20.6569 2 19V5ZM7.5035 8C7.5035 8.55228 7.95122 9 8.5035 9H11V16.4954C11 17.0476 11.4477 17.4954 12 17.4954C12.5523 17.4954 13 17.0476 13 16.4954V9H15.4965C16.0488 9 16.4965 8.55228 16.4965 8C16.4965 7.44772 16.0488 7 15.4965 7H8.5035C7.95122 7 7.5035 7.44772 7.5035 8Z"
+        fill="currentColor"
+      />
+    </svg>
+  );
+}
+
+export function TaskIconSmall({ className }: { className?: string }) {
+  return (
+    <svg className={className} viewBox="0 0 20 20" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path
+        fillRule="evenodd"
+        clipRule="evenodd"
+        d="M2 4C2 2.89543 2.89543 2 4 2H16C17.1046 2 18 2.89543 18 4V16C18 17.1046 17.1046 18 16 18H4C2.89543 18 2 17.1046 2 16V4ZM6 7C6 7.55228 6.44772 8 7 8H9L9 14C9 14.5523 9.44772 15 10 15C10.5523 15 11 14.5523 11 14V8H13C13.5523 8 14 7.55228 14 7C14 6.44772 13.5523 6 13 6H10H7C6.44772 6 6 6.44772 6 7Z"
+        fill="currentColor"
+      />
     </svg>
   );
 }