diff --git a/.vscode/launch.json b/.vscode/launch.json
index ea926d23a..8aabbd986 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -41,7 +41,7 @@
       "type": "node-terminal",
       "request": "launch",
       "name": "Debug V3 Deploy CLI",
-      "command": "pnpm exec triggerdev deploy",
+      "command": "pnpm exec triggerdev deploy --self-hosted",
       "cwd": "${workspaceFolder}/references/v3-catalog",
       "sourceMaps": true
     },
diff --git a/packages/cli-v3/package.json b/packages/cli-v3/package.json
index 4d1d89560..6d0606535 100644
--- a/packages/cli-v3/package.json
+++ b/packages/cli-v3/package.json
@@ -116,7 +116,7 @@
     "source-map-support": "0.5.21",
     "terminal-link": "^3.0.0",
     "tiny-invariant": "^1.2.0",
-    "tinyexec": "^0.1.4",
+    "tinyexec": "^0.2.0",
     "ws": "^8.18.0",
     "xdg-app-paths": "^8.3.0",
     "zod": "3.23.8",
diff --git a/packages/cli-v3/src/commands/deploy.ts b/packages/cli-v3/src/commands/deploy.ts
index 61b026e9b..05a0469d1 100644
--- a/packages/cli-v3/src/commands/deploy.ts
+++ b/packages/cli-v3/src/commands/deploy.ts
@@ -355,6 +355,7 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     noCache: options.noCache,
     push: options.push,
     registryHost,
+    registry: options.registry,
     deploymentId: deployment.id,
     deploymentVersion: deployment.version,
     imageTag: deployment.imageTag,
diff --git a/packages/cli-v3/src/deploy/buildImage.ts b/packages/cli-v3/src/deploy/buildImage.ts
index f85f5856c..9fec0f5d8 100644
--- a/packages/cli-v3/src/deploy/buildImage.ts
+++ b/packages/cli-v3/src/deploy/buildImage.ts
@@ -430,7 +430,7 @@ async function generateBunContainerfile(buildManifest: BuildManifest) {
     .join("\n");
 
   return `
-FROM oven/bun:1 AS base
+FROM imbios/bun-node:22-debian AS base
 RUN apt-get update && apt-get --fix-broken install -y && apt-get install -y --no-install-recommends busybox ca-certificates dumb-init git openssl && apt-get clean && rm -rf /var/lib/apt/lists/*
 
 FROM base AS install
@@ -484,6 +484,13 @@ FROM base AS final
 USER bun
 WORKDIR /app
 
+ARG TRIGGER_PROJECT_ID
+ARG TRIGGER_DEPLOYMENT_ID
+ARG TRIGGER_DEPLOYMENT_VERSION
+ARG TRIGGER_CONTENT_HASH
+ARG TRIGGER_PROJECT_REF
+ARG NODE_EXTRA_CA_CERTS
+
 ENV TRIGGER_PROJECT_ID=\${TRIGGER_PROJECT_ID} \
     TRIGGER_DEPLOYMENT_ID=\${TRIGGER_DEPLOYMENT_ID} \
     TRIGGER_DEPLOYMENT_VERSION=\${TRIGGER_DEPLOYMENT_VERSION} \
@@ -498,7 +505,7 @@ COPY --from=install --chown=bun:bun /app ./
 # Copy the index.json file from the indexer stage
 COPY --from=indexer --chown=bun:bun /app/index.json ./
 
-ENTRYPOINT [ "dumb-init", "bun", "run", "${buildManifest.workerEntryPoint}" ]
+ENTRYPOINT [ "dumb-init", "node", "${buildManifest.workerEntryPoint}" ]
 CMD []
   `;
 }
@@ -576,6 +583,13 @@ FROM base AS final
 USER node
 WORKDIR /app
 
+ARG TRIGGER_PROJECT_ID
+ARG TRIGGER_DEPLOYMENT_ID
+ARG TRIGGER_DEPLOYMENT_VERSION
+ARG TRIGGER_CONTENT_HASH
+ARG TRIGGER_PROJECT_REF
+ARG NODE_EXTRA_CA_CERTS
+
 ENV TRIGGER_PROJECT_ID=\${TRIGGER_PROJECT_ID} \
     TRIGGER_DEPLOYMENT_ID=\${TRIGGER_DEPLOYMENT_ID} \
     TRIGGER_DEPLOYMENT_VERSION=\${TRIGGER_DEPLOYMENT_VERSION} \
diff --git a/packages/cli-v3/src/dev/backgroundWorker.ts b/packages/cli-v3/src/dev/backgroundWorker.ts
index b259b06d5..4c2804cf8 100644
--- a/packages/cli-v3/src/dev/backgroundWorker.ts
+++ b/packages/cli-v3/src/dev/backgroundWorker.ts
@@ -9,21 +9,16 @@ import {
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
   WorkerManifest,
-  childToWorkerMessages,
   correctErrorStackTrace,
   indexerToWorkerMessages,
-  workerToChildMessages,
 } from "@trigger.dev/core/v3";
-import {
-  ZodMessageHandler,
-  ZodMessageSender,
-  parseMessageFromCatalog,
-} from "@trigger.dev/core/v3/zodMessageHandler";
+import { parseMessageFromCatalog } from "@trigger.dev/core/v3/zodMessageHandler";
 import { Evt } from "evt";
-import { ChildProcess, fork } from "node:child_process";
-import { chalkError, chalkGrey, chalkRun, prettyPrintDate } from "../utilities/cliOutput.js";
+import { fork } from "node:child_process";
 
+import { execPathForRuntime } from "@trigger.dev/core/v3/build";
 import { join } from "node:path";
+import { TaskRunProcess, TaskRunProcessOptions } from "../executions/taskRunProcess.js";
 import { eventBus } from "../utilities/eventBus.js";
 import { writeJSONFile } from "../utilities/fileSystem.js";
 import { logger } from "../utilities/logger.js";
@@ -35,8 +30,7 @@ import {
   UncaughtExceptionError,
   UnexpectedExitError,
   getFriendlyErrorMessage,
-} from "./errors.js";
-import { execPathForRuntime } from "@trigger.dev/core/v3/build";
+} from "../executions/errors.js";
 
 export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"];
 export class BackgroundWorkerCoordinator {
@@ -212,7 +206,7 @@ export class BackgroundWorker {
   public serverWorker: ServerBackgroundWorker | undefined;
 
   _taskRunProcesses: Map<string, TaskRunProcess> = new Map();
-  private _taskRunProcessesBeingKilled: Set<number> = new Set();
+  private _taskRunProcessesBeingKilled: Map<number, TaskRunProcess> = new Map();
 
   private _closed: boolean = false;
 
@@ -379,14 +373,10 @@ export class BackgroundWorker {
 
     const processOptions: TaskRunProcessOptions = {
       payload,
-      build: this.build,
       env: {
         ...this.params.env,
         ...payload.environment,
         TRIGGER_WORKER_MANIFEST_PATH: this.workerManifestPath,
-        NODE_OPTIONS: this.build.loaderEntryPoint
-          ? `--import=${this.build.loaderEntryPoint} ${process.env.NODE_OPTIONS ?? ""}`
-          : process.env.NODE_OPTIONS ?? "",
       },
       serverWorker: this.serverWorker,
       workerManifest: this.manifest,
@@ -410,9 +400,9 @@ export class BackgroundWorker {
       }
     });
 
-    taskRunProcess.onIsBeingKilled.attach((pid) => {
-      if (pid) {
-        this._taskRunProcessesBeingKilled.add(pid);
+    taskRunProcess.onIsBeingKilled.attach((taskRunProcess) => {
+      if (taskRunProcess.pid) {
+        this._taskRunProcessesBeingKilled.set(taskRunProcess.pid, taskRunProcess);
       }
     });
 
@@ -420,6 +410,10 @@ export class BackgroundWorker {
       this.onTaskRunHeartbeat.post(id);
     });
 
+    taskRunProcess.onReadyToDispose.attach(async () => {
+      await taskRunProcess.kill();
+    });
+
     await taskRunProcess.initialize();
 
     this._taskRunProcesses.set(payload.execution.run.id, taskRunProcess);
@@ -621,353 +615,3 @@ export class BackgroundWorker {
     };
   }
 }
-
-type TaskRunProcessOptions = {
-  payload: TaskRunExecutionPayload;
-  build: BuildManifest;
-  env: Record<string, string>;
-  cwd?: string;
-  // this is the "index" data
-  workerManifest: WorkerManifest;
-  // this is the worker on the server data
-  serverWorker: ServerBackgroundWorker;
-  messageId?: string;
-};
-
-class TaskRunProcess {
-  private _handler = new ZodMessageHandler({
-    schema: childToWorkerMessages,
-  });
-  private _sender: ZodMessageSender<typeof workerToChildMessages>;
-  private _child: ChildProcess | undefined;
-  private _childPid?: number;
-  private _attemptPromises: Map<
-    string,
-    { resolver: (value: TaskRunExecutionResult) => void; rejecter: (err?: any) => void }
-  > = new Map();
-  private _attemptStatuses: Map<string, "PENDING" | "REJECTED" | "RESOLVED"> = new Map();
-  private _currentExecution: TaskRunExecution | undefined;
-  private _isBeingKilled: boolean = false;
-  private _isBeingCancelled: boolean = false;
-  private _stderr: Array<string> = [];
-  /**
-   * @deprecated use onTaskRunHeartbeat instead
-   */
-  public onTaskHeartbeat: Evt<string> = new Evt();
-  public onTaskRunHeartbeat: Evt<string> = new Evt();
-  public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
-    new Evt();
-  public onIsBeingKilled: Evt<number | undefined> = new Evt();
-  private _debuggingPort: number | undefined;
-
-  constructor(public readonly options: TaskRunProcessOptions) {
-    this._sender = new ZodMessageSender({
-      schema: workerToChildMessages,
-      sender: async (message) => {
-        if (this._child?.connected && !this._isBeingKilled && !this._child.killed) {
-          this._child.send(message);
-        }
-      },
-    });
-  }
-
-  async cancel() {
-    this._isBeingCancelled = true;
-
-    await this.cleanup(true);
-  }
-
-  get runId() {
-    return this.options.payload.execution.run.id;
-  }
-
-  get isTest() {
-    return this.options.payload.execution.run.isTest;
-  }
-
-  get payload() {
-    return this.options.payload;
-  }
-
-  async initialize() {
-    const { env, build, cwd } = this.options;
-
-    const fullEnv = {
-      ...(this.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}),
-      ...env,
-    };
-
-    logger.debug(`[${this.runId}] initializing task run process`, {
-      env: fullEnv,
-      path: build.workerEntryPoint,
-      cwd,
-    });
-
-    this._child = fork(build.workerEntryPoint, {
-      stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
-      cwd,
-      env: fullEnv,
-      execArgv: ["--trace-uncaught", "--no-warnings=ExperimentalWarning"],
-      execPath: execPathForRuntime(build.runtime),
-    });
-
-    this._childPid = this._child?.pid;
-
-    this._child.on("message", this.#handleMessage.bind(this));
-    this._child.on("exit", this.#handleExit.bind(this));
-    this._child.stdout?.on("data", this.#handleLog.bind(this));
-    this._child.stderr?.on("data", this.#handleStdErr.bind(this));
-  }
-
-  async cleanup(kill: boolean = false) {
-    if (kill && this._isBeingKilled) {
-      return;
-    }
-
-    if (kill) {
-      this._isBeingKilled = true;
-      this.onIsBeingKilled.post(this._child?.pid);
-    }
-
-    logger.debug(`[${this.runId}] cleaning up task run process`, { kill, pid: this.pid });
-
-    await this._sender.send("CLEANUP", {
-      flush: true,
-      kill,
-    });
-
-    // FIXME: Something broke READY_TO_DISPOSE. We never receive it, so we always have to kill the process after the timeout below.
-
-    if (!kill) {
-      return;
-    }
-
-    // Set a timeout to kill the child process if it hasn't been killed within 5 seconds
-    setTimeout(() => {
-      if (this._child && !this._child.killed) {
-        logger.debug(`[${this.runId}] killing task run process after timeout`, { pid: this.pid });
-
-        this._child.kill();
-      }
-    }, 5000);
-  }
-
-  async execute(): Promise<TaskRunExecutionResult> {
-    let resolver: (value: TaskRunExecutionResult) => void;
-    let rejecter: (err?: any) => void;
-
-    const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => {
-      resolver = resolve;
-      rejecter = reject;
-    });
-
-    this._attemptStatuses.set(this.payload.execution.attempt.id, "PENDING");
-
-    // @ts-expect-error - We know that the resolver and rejecter are defined
-    this._attemptPromises.set(this.payload.execution.attempt.id, { resolver, rejecter });
-
-    const { execution, traceContext } = this.payload;
-
-    this._currentExecution = execution;
-
-    await this._sender.send("EXECUTE_TASK_RUN", {
-      execution,
-      traceContext,
-      metadata: this.options.serverWorker,
-    });
-
-    const result = await promise;
-
-    this._currentExecution = undefined;
-
-    return result;
-  }
-
-  taskRunCompletedNotification(completion: TaskRunExecutionResult) {
-    if (!completion.ok && typeof completion.retry !== "undefined") {
-      return;
-    }
-
-    if (completion.id === this.runId) {
-      // We don't need to notify the task run process if it's the same as the one we're running
-      return;
-    }
-
-    logger.debug(`[${this.runId}] task run completed notification`, {
-      completion,
-    });
-
-    this._sender.send("TASK_RUN_COMPLETED_NOTIFICATION", {
-      version: "v2",
-      completion,
-    });
-  }
-
-  async #handleMessage(msg: any) {
-    const message = this._handler.parseMessage(msg);
-
-    if (!message.success) {
-      logger.error(`Dropping message: ${message.error}`, { message });
-      return;
-    }
-
-    switch (message.data.type) {
-      case "TASK_RUN_COMPLETED": {
-        const { result, execution } = message.data.payload;
-
-        logger.debug(`[${this.runId}] task run completed`, {
-          result,
-        });
-
-        const promiseStatus = this._attemptStatuses.get(execution.attempt.id);
-
-        if (promiseStatus !== "PENDING") {
-          return;
-        }
-
-        this._attemptStatuses.set(execution.attempt.id, "RESOLVED");
-
-        const attemptPromise = this._attemptPromises.get(execution.attempt.id);
-
-        if (!attemptPromise) {
-          return;
-        }
-
-        const { resolver } = attemptPromise;
-
-        resolver(result);
-
-        break;
-      }
-      case "READY_TO_DISPOSE": {
-        logger.debug(`[${this.runId}] task run process is ready to dispose`);
-
-        this.#kill();
-
-        break;
-      }
-      case "TASK_HEARTBEAT": {
-        if (this.options.messageId) {
-          this.onTaskRunHeartbeat.post(this.options.messageId);
-        } else {
-          this.onTaskHeartbeat.post(message.data.payload.id);
-        }
-
-        break;
-      }
-    }
-  }
-
-  async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
-    logger.debug(`[${this.runId}] handle task run process exit`, { code, signal, pid: this.pid });
-
-    // Go through all the attempts currently pending and reject them
-    for (const [id, status] of this._attemptStatuses.entries()) {
-      if (status === "PENDING") {
-        this._attemptStatuses.set(id, "REJECTED");
-
-        const attemptPromise = this._attemptPromises.get(id);
-
-        if (!attemptPromise) {
-          continue;
-        }
-
-        const { rejecter } = attemptPromise;
-
-        if (this._isBeingCancelled) {
-          rejecter(new CancelledProcessError());
-        } else if (this._isBeingKilled) {
-          rejecter(new CleanupProcessError());
-        } else {
-          rejecter(
-            new UnexpectedExitError(
-              code ?? -1,
-              signal,
-              this._stderr.length ? this._stderr.join("\n") : undefined
-            )
-          );
-        }
-      }
-    }
-
-    this.onExit.post({ code, signal, pid: this.pid });
-  }
-
-  #handleLog(data: Buffer) {
-    if (!this._currentExecution) {
-      logger.log(`${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`);
-
-      return;
-    }
-
-    const runId = chalkRun(
-      `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}`
-    );
-
-    logger.log(
-      `${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${data.toString()}`
-    );
-  }
-
-  #handleStdErr(data: Buffer) {
-    if (this._isBeingKilled) {
-      return;
-    }
-
-    if (!this._currentExecution) {
-      logger.log(`${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`);
-
-      return;
-    }
-
-    const runId = chalkRun(
-      `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}`
-    );
-
-    const errorLine = data.toString();
-
-    logger.log(
-      `${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${errorLine}`
-    );
-
-    if (this._stderr.length > 100) {
-      this._stderr.shift();
-    }
-    this._stderr.push(errorLine);
-  }
-
-  #kill() {
-    logger.debug(`[${this.runId}] #kill()`, { pid: this.pid });
-
-    if (this._child && !this._child.killed) {
-      this._child?.kill();
-    }
-  }
-
-  async kill(signal?: number | NodeJS.Signals, timeoutInMs?: number) {
-    logger.debug(`[${this.runId}] killing task run process`, {
-      signal,
-      timeoutInMs,
-      pid: this.pid,
-    });
-
-    this._isBeingKilled = true;
-
-    const killTimeout = this.onExit.waitFor(timeoutInMs);
-
-    this.onIsBeingKilled.post(this._child?.pid);
-    this._child?.kill(signal);
-
-    if (timeoutInMs) {
-      await killTimeout;
-    }
-  }
-
-  get isBeingKilled() {
-    return this._isBeingKilled || this._child?.killed;
-  }
-
-  get pid() {
-    return this._childPid;
-  }
-}
diff --git a/packages/cli-v3/src/entryPoints/deploy-indexer.ts b/packages/cli-v3/src/entryPoints/deploy-indexer.ts
index f9813a9b7..1365fcac1 100644
--- a/packages/cli-v3/src/entryPoints/deploy-indexer.ts
+++ b/packages/cli-v3/src/entryPoints/deploy-indexer.ts
@@ -173,7 +173,13 @@ async function indexDeployment({
       });
     }
 
-    const workerManifest: WorkerManifest = { tasks, configPath: buildManifest.configPath };
+    const workerManifest: WorkerManifest = {
+      tasks,
+      configPath: buildManifest.configPath,
+      runtime: buildManifest.runtime,
+      workerEntryPoint: buildManifest.workerEntryPoint,
+      loaderEntryPoint: buildManifest.loaderEntryPoint,
+    };
 
     console.log("Writing index.json", process.cwd());
 
diff --git a/packages/cli-v3/src/entryPoints/deploy.ts b/packages/cli-v3/src/entryPoints/deploy.ts
index cb0ff5c3b..02528c066 100644
--- a/packages/cli-v3/src/entryPoints/deploy.ts
+++ b/packages/cli-v3/src/entryPoints/deploy.ts
@@ -1 +1,1110 @@
-export {};
+import {
+  CoordinatorToProdWorkerMessages,
+  PostStartCauses,
+  PreStopCauses,
+  ProdTaskRunExecution,
+  ProdWorkerToCoordinatorMessages,
+  TaskRunErrorCodes,
+  TaskRunExecutionResult,
+  TaskRunFailedExecutionResult,
+  WaitReason,
+  WorkerManifest,
+} from "@trigger.dev/core/v3";
+import {
+  EXIT_CODE_CHILD_NONZERO,
+  ExponentialBackoff,
+  HttpReply,
+  SimpleLogger,
+  getRandomPortNumber,
+} from "@trigger.dev/core/v3/apps";
+import { ZodSocketConnection } from "@trigger.dev/core/v3/zodSocket";
+import { Evt } from "evt";
+import { randomUUID } from "node:crypto";
+import { readFile } from "node:fs/promises";
+import { createServer } from "node:http";
+import { setTimeout as timeout } from "node:timers/promises";
+import { OnWaitForBatchMessage, OnWaitForTaskMessage } from "../executions/taskRunProcess.js";
+
+const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || getRandomPortNumber());
+const COORDINATOR_HOST = process.env.COORDINATOR_HOST || "127.0.0.1";
+const COORDINATOR_PORT = Number(process.env.COORDINATOR_PORT || 50080);
+const MACHINE_NAME = process.env.MACHINE_NAME || "local";
+const POD_NAME = process.env.POD_NAME || "some-pod";
+const SHORT_HASH = process.env.TRIGGER_CONTENT_HASH!.slice(0, 9);
+
+const logger = new SimpleLogger(`[${MACHINE_NAME}][${SHORT_HASH}]`);
+
+const defaultBackoff = new ExponentialBackoff("FullJitter", {
+  maxRetries: 5,
+});
+
+class ProdWorker {
+  private contentHash = process.env.TRIGGER_CONTENT_HASH!;
+  private projectRef = process.env.TRIGGER_PROJECT_REF!;
+  private envId = process.env.TRIGGER_ENV_ID!;
+  private runId = process.env.TRIGGER_RUN_ID!;
+  private deploymentId = process.env.TRIGGER_DEPLOYMENT_ID!;
+  private deploymentVersion = process.env.TRIGGER_DEPLOYMENT_VERSION!;
+  private runningInKubernetes = !!process.env.KUBERNETES_PORT;
+
+  private executing = false;
+  private completed = new Set<string>();
+  private paused = false;
+  private attemptFriendlyId?: string;
+  private attemptNumber?: number;
+
+  private nextResumeAfter?: WaitReason;
+  private waitForPostStart = false;
+  private connectionCount = 0;
+
+  private restoreNotification = Evt.create();
+
+  private waitForTaskReplay:
+    | {
+        idempotencyKey: string;
+        message: OnWaitForTaskMessage;
+        attempt: number;
+      }
+    | undefined;
+  private waitForBatchReplay:
+    | {
+        idempotencyKey: string;
+        message: OnWaitForBatchMessage;
+        attempt: number;
+      }
+    | undefined;
+  private readyForLazyAttemptReplay:
+    | {
+        idempotencyKey: string;
+      }
+    | undefined;
+  private durationResumeFallback:
+    | {
+        idempotencyKey: string;
+      }
+    | undefined;
+
+  #httpPort: number;
+  #httpServer: ReturnType<typeof createServer>;
+  #coordinatorSocket: ZodSocketConnection<
+    typeof ProdWorkerToCoordinatorMessages,
+    typeof CoordinatorToProdWorkerMessages
+  >;
+
+  constructor(
+    port: number,
+    private workerManifest: WorkerManifest,
+    private host = "0.0.0.0"
+  ) {
+    process.on("SIGTERM", this.#handleSignal.bind(this, "SIGTERM"));
+
+    this.#coordinatorSocket = this.#createCoordinatorSocket(COORDINATOR_HOST);
+
+    this.#httpPort = port;
+    this.#httpServer = this.#createHttpServer();
+  }
+
+  async #handleSignal(signal: NodeJS.Signals) {
+    logger.log("Received signal", { signal });
+
+    if (signal === "SIGTERM") {
+      let gracefulExitTimeoutElapsed = false;
+
+      if (this.executing) {
+        const terminationGracePeriodSeconds = 60 * 60;
+
+        logger.log("Waiting for attempt to complete before exiting", {
+          terminationGracePeriodSeconds,
+        });
+
+        // Wait for termination grace period minus 5s to give cleanup a chance to complete
+        await timeout(terminationGracePeriodSeconds * 1000 - 5000);
+        gracefulExitTimeoutElapsed = true;
+
+        logger.log("Termination timeout reached, exiting gracefully.");
+      } else {
+        logger.log("Not executing, exiting immediately.");
+      }
+
+      await this.#exitGracefully(gracefulExitTimeoutElapsed);
+      return;
+    }
+
+    logger.log("Unhandled signal", { signal });
+  }
+
+  async #exitGracefully(gracefulExitTimeoutElapsed = false, exitCode = 0) {
+    // TODO: close the worker process
+    // await this.#backgroundWorker.close(gracefulExitTimeoutElapsed);
+
+    if (!gracefulExitTimeoutElapsed) {
+      // TODO: Maybe add a sensible timeout instead of a conditional to avoid zombies
+      process.exit(exitCode);
+    }
+  }
+
+  async #reconnectAfterPostStart() {
+    this.waitForPostStart = false;
+
+    this.#coordinatorSocket.close();
+    this.connectionCount = 0;
+
+    let coordinatorHost = COORDINATOR_HOST;
+
+    try {
+      if (this.runningInKubernetes) {
+        coordinatorHost = (await readFile("/etc/taskinfo/coordinator-host", "utf-8")).replace(
+          "\n",
+          ""
+        );
+
+        logger.log("reconnecting", {
+          coordinatorHost: {
+            fromEnv: COORDINATOR_HOST,
+            fromVolume: coordinatorHost,
+            current: this.#coordinatorSocket.socket.io.opts.hostname,
+          },
+        });
+      }
+    } catch (error) {
+      logger.error("taskinfo read error during reconnect", {
+        error: error instanceof Error ? error.message : error,
+      });
+    } finally {
+      this.#coordinatorSocket = this.#createCoordinatorSocket(coordinatorHost);
+    }
+  }
+
+  // MARK: TASK WAIT
+  #waitForTaskHandlerFactory(workerId?: string) {
+    return async (message: OnWaitForTaskMessage, replayIdempotencyKey?: string) => {
+      logger.log("onWaitForTask", { workerId, message });
+
+      if (this.nextResumeAfter) {
+        logger.error("Already waiting for resume, skipping wait for task", {
+          nextResumeAfter: this.nextResumeAfter,
+        });
+
+        return;
+      }
+
+      const waitForTask = await defaultBackoff.execute(async ({ retry }) => {
+        logger.log("Wait for task with backoff", { retry });
+
+        if (!this.attemptFriendlyId) {
+          logger.error("Failed to send wait message, attempt friendly ID not set", { message });
+
+          throw new ExponentialBackoff.StopRetrying("No attempt ID");
+        }
+
+        return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_TASK", {
+          version: "v2",
+          friendlyId: message.friendlyId,
+          attemptFriendlyId: this.attemptFriendlyId,
+        });
+      });
+
+      if (!waitForTask.success) {
+        logger.error("Failed to wait for task with backoff", {
+          cause: waitForTask.cause,
+          error: waitForTask.error,
+        });
+
+        this.#emitUnrecoverableError(
+          "WaitForTaskFailed",
+          `${waitForTask.cause}: ${waitForTask.error}`
+        );
+
+        return;
+      }
+
+      const { willCheckpointAndRestore } = waitForTask.result;
+
+      await this.#prepareForWait("WAIT_FOR_TASK", willCheckpointAndRestore);
+
+      if (willCheckpointAndRestore) {
+        // We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
+        if (!this.waitForTaskReplay) {
+          this.waitForTaskReplay = {
+            message,
+            attempt: 1,
+            idempotencyKey: randomUUID(),
+          };
+        } else {
+          if (
+            replayIdempotencyKey &&
+            replayIdempotencyKey !== this.waitForTaskReplay.idempotencyKey
+          ) {
+            logger.error(
+              "wait for task handler called with mismatched idempotency key, won't overwrite replay request"
+            );
+            return;
+          }
+
+          this.waitForTaskReplay.attempt++;
+        }
+      }
+    };
+  }
+
+  // MARK: BATCH WAIT
+  #waitForBatchHandlerFactory(workerId?: string) {
+    return async (message: OnWaitForBatchMessage, replayIdempotencyKey?: string) => {
+      logger.log("onWaitForBatch", { workerId, message });
+
+      if (this.nextResumeAfter) {
+        logger.error("Already waiting for resume, skipping wait for batch", {
+          nextResumeAfter: this.nextResumeAfter,
+        });
+
+        return;
+      }
+
+      const waitForBatch = await defaultBackoff.execute(async ({ retry }) => {
+        logger.log("Wait for batch with backoff", { retry });
+
+        if (!this.attemptFriendlyId) {
+          logger.error("Failed to send wait message, attempt friendly ID not set", { message });
+
+          throw new ExponentialBackoff.StopRetrying("No attempt ID");
+        }
+
+        return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_BATCH", {
+          version: "v2",
+          batchFriendlyId: message.batchFriendlyId,
+          runFriendlyIds: message.runFriendlyIds,
+          attemptFriendlyId: this.attemptFriendlyId,
+        });
+      });
+
+      if (!waitForBatch.success) {
+        logger.error("Failed to wait for batch with backoff", {
+          cause: waitForBatch.cause,
+          error: waitForBatch.error,
+        });
+
+        this.#emitUnrecoverableError(
+          "WaitForBatchFailed",
+          `${waitForBatch.cause}: ${waitForBatch.error}`
+        );
+
+        return;
+      }
+
+      const { willCheckpointAndRestore } = waitForBatch.result;
+
+      await this.#prepareForWait("WAIT_FOR_BATCH", willCheckpointAndRestore);
+
+      if (willCheckpointAndRestore) {
+        // We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
+        if (!this.waitForBatchReplay) {
+          this.waitForBatchReplay = {
+            message,
+            attempt: 1,
+            idempotencyKey: randomUUID(),
+          };
+        } else {
+          if (
+            replayIdempotencyKey &&
+            replayIdempotencyKey !== this.waitForBatchReplay.idempotencyKey
+          ) {
+            logger.error(
+              "wait for task handler called with mismatched idempotency key, won't overwrite replay request"
+            );
+            return;
+          }
+
+          this.waitForBatchReplay.attempt++;
+        }
+      }
+    };
+  }
+
+  async #prepareForWait(reason: WaitReason, willCheckpointAndRestore: boolean) {
+    logger.log(`prepare for ${reason}`, { willCheckpointAndRestore });
+
+    if (this.nextResumeAfter) {
+      logger.error("Already waiting for resume, skipping prepare for wait", {
+        nextResumeAfter: this.nextResumeAfter,
+        params: {
+          reason,
+          willCheckpointAndRestore,
+        },
+      });
+
+      return;
+    }
+
+    if (!willCheckpointAndRestore) {
+      return;
+    }
+
+    this.paused = true;
+    this.nextResumeAfter = reason;
+    this.waitForPostStart = true;
+
+    await this.#prepareForCheckpoint();
+  }
+
+  // MARK: RETRY PREP
+  async #prepareForRetry(shouldExit: boolean, exitCode?: number) {
+    logger.log("prepare for retry", { shouldExit, exitCode });
+
+    // Graceful shutdown on final attempt
+    if (shouldExit) {
+      await this.#exitGracefully(false, exitCode);
+      return;
+    }
+
+    // Clear state for next execution
+    this.paused = false;
+    this.waitForPostStart = false;
+    this.executing = false;
+    this.attemptFriendlyId = undefined;
+    this.attemptNumber = undefined;
+  }
+
+  // MARK: CHECKPOINT PREP
+  async #prepareForCheckpoint(flush = true) {
+    if (flush) {
+      // Flush before checkpointing so we don't flush the same spans again after restore
+      try {
+        // TODO: flush the telemetry
+        // await this.#backgroundWorker.flushTelemetry();
+      } catch (error) {
+        logger.error(
+          "Failed to flush telemetry while preparing for checkpoint, will proceed anyway",
+          { error }
+        );
+      }
+    }
+
+    try {
+      // Kill the previous worker process to prevent large checkpoints
+      // TODO: do we need this?
+      // await this.#backgroundWorker.forceKillOldTaskRunProcesses();
+    } catch (error) {
+      logger.error(
+        "Failed to kill previous worker while preparing for checkpoint, will proceed anyway",
+        { error }
+      );
+    }
+
+    this.#readyForCheckpoint();
+  }
+
+  #resumeAfterDuration() {
+    this.paused = false;
+    this.nextResumeAfter = undefined;
+    this.waitForPostStart = false;
+
+    this.durationResumeFallback = undefined;
+
+    // TODO: signal to the worker that is can resume after duration
+    // this.#backgroundWorker.waitCompletedNotification();
+  }
+
+  async #readyForLazyAttempt() {
+    const idempotencyKey = randomUUID();
+
+    this.readyForLazyAttemptReplay = {
+      idempotencyKey,
+    };
+
+    // Retry if we don't receive EXECUTE_TASK_RUN_LAZY_ATTEMPT in a reasonable time
+    // ..but we also have to be fast to avoid failing the task due to missing heartbeat
+    for await (const { delay, retry } of defaultBackoff.min(10).maxRetries(3)) {
+      if (retry > 0) {
+        logger.log("retrying ready for lazy attempt", { retry });
+      }
+
+      this.#coordinatorSocket.socket.emit("READY_FOR_LAZY_ATTEMPT", {
+        version: "v1",
+        runId: this.runId,
+        totalCompletions: this.completed.size,
+      });
+
+      await timeout(delay.milliseconds);
+
+      if (!this.readyForLazyAttemptReplay) {
+        logger.error("replay ready for lazy attempt cancelled, discarding", {
+          idempotencyKey,
+        });
+
+        return;
+      }
+
+      if (idempotencyKey !== this.readyForLazyAttemptReplay.idempotencyKey) {
+        logger.error("replay ready for lazy attempt idempotency key mismatch, discarding", {
+          idempotencyKey,
+          newIdempotencyKey: this.readyForLazyAttemptReplay.idempotencyKey,
+        });
+
+        return;
+      }
+    }
+
+    // Fail the task with a more descriptive message as it likely failed with a generic missing heartbeat error
+    this.#failRun(this.runId, "Failed to receive execute request in a reasonable time");
+  }
+
+  #readyForCheckpoint() {
+    this.#coordinatorSocket.socket.emit("READY_FOR_CHECKPOINT", { version: "v1" });
+  }
+
+  #failRun(anyRunId: string, error: unknown) {
+    logger.error("Failing run", { anyRunId, error });
+
+    const completion: TaskRunFailedExecutionResult = {
+      ok: false,
+      id: anyRunId,
+      retry: undefined,
+      error:
+        error instanceof Error
+          ? {
+              type: "BUILT_IN_ERROR",
+              name: error.name,
+              message: error.message,
+              stackTrace: error.stack ?? "",
+            }
+          : {
+              type: "BUILT_IN_ERROR",
+              name: "UnknownError",
+              message: String(error),
+              stackTrace: "",
+            },
+    };
+
+    this.#coordinatorSocket.socket.emit("TASK_RUN_FAILED_TO_RUN", {
+      version: "v1",
+      completion,
+    });
+  }
+
+  // MARK: ATTEMPT COMPLETION
+  async #submitAttemptCompletion(
+    execution: ProdTaskRunExecution,
+    completion: TaskRunExecutionResult,
+    replayIdempotencyKey?: string
+  ) {
+    const taskRunCompleted = await defaultBackoff.execute(async ({ retry }) => {
+      logger.log("Submit attempt completion with backoff", { retry });
+
+      return await this.#coordinatorSocket.socket
+        .timeout(20_000)
+        .emitWithAck("TASK_RUN_COMPLETED", {
+          version: "v1",
+          execution,
+          completion,
+        });
+    });
+
+    if (!taskRunCompleted.success) {
+      logger.error("Failed to complete lazy attempt with backoff", {
+        cause: taskRunCompleted.cause,
+        error: taskRunCompleted.error,
+      });
+
+      this.#failRun(execution.run.id, taskRunCompleted.error);
+
+      return;
+    }
+
+    const { willCheckpointAndRestore, shouldExit } = taskRunCompleted.result;
+
+    logger.log("completion acknowledged", { willCheckpointAndRestore, shouldExit });
+
+    const exitCode =
+      !completion.ok &&
+      completion.error.type === "INTERNAL_ERROR" &&
+      completion.error.code === TaskRunErrorCodes.TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE
+        ? EXIT_CODE_CHILD_NONZERO
+        : 0;
+
+    await this.#prepareForRetry(shouldExit, exitCode);
+
+    if (willCheckpointAndRestore) {
+      logger.error("This worker should never be checkpointed between attempts. This is a bug.");
+    }
+  }
+
+  #returnValidatedExtraHeaders(headers: Record<string, string>) {
+    for (const [key, value] of Object.entries(headers)) {
+      if (value === undefined) {
+        throw new Error(`Extra header is undefined: ${key}`);
+      }
+    }
+
+    return headers;
+  }
+
+  // MARK: COORDINATOR SOCKET
+  #createCoordinatorSocket(host: string) {
+    const extraHeaders = this.#returnValidatedExtraHeaders({
+      "x-machine-name": MACHINE_NAME,
+      "x-pod-name": POD_NAME,
+      "x-trigger-content-hash": this.contentHash,
+      "x-trigger-project-ref": this.projectRef,
+      "x-trigger-env-id": this.envId,
+      "x-trigger-deployment-id": this.deploymentId,
+      "x-trigger-run-id": this.runId,
+      "x-trigger-deployment-version": this.deploymentVersion,
+    });
+
+    if (this.attemptFriendlyId) {
+      extraHeaders["x-trigger-attempt-friendly-id"] = this.attemptFriendlyId;
+    }
+
+    if (this.attemptNumber !== undefined) {
+      extraHeaders["x-trigger-attempt-number"] = String(this.attemptNumber);
+    }
+
+    logger.log(`connecting to coordinator: ${host}:${COORDINATOR_PORT}`);
+    logger.debug(`connecting with extra headers`, { extraHeaders });
+
+    const coordinatorConnection = new ZodSocketConnection({
+      namespace: "prod-worker",
+      host,
+      port: COORDINATOR_PORT,
+      clientMessages: ProdWorkerToCoordinatorMessages,
+      serverMessages: CoordinatorToProdWorkerMessages,
+      extraHeaders,
+      ioOptions: {
+        reconnectionDelay: 1000,
+        reconnectionDelayMax: 3000,
+      },
+      handlers: {
+        RESUME_AFTER_DEPENDENCY: async ({ completions }) => {
+          if (!this.paused) {
+            logger.error("Failed to resume after dependency: Worker not paused");
+            return;
+          }
+
+          if (completions.length === 0) {
+            logger.error("Failed to resume after dependency: No completions");
+            return;
+          }
+
+          if (
+            this.nextResumeAfter !== "WAIT_FOR_TASK" &&
+            this.nextResumeAfter !== "WAIT_FOR_BATCH"
+          ) {
+            logger.error("Failed to resume after dependency: Invalid next resume", {
+              nextResumeAfter: this.nextResumeAfter,
+            });
+            return;
+          }
+
+          if (this.nextResumeAfter === "WAIT_FOR_TASK" && completions.length > 1) {
+            logger.error(
+              "Failed to resume after dependency: Waiting for single task but got multiple completions",
+              {
+                completions: completions,
+              }
+            );
+            return;
+          }
+
+          switch (this.nextResumeAfter) {
+            case "WAIT_FOR_TASK": {
+              this.waitForTaskReplay = undefined;
+              break;
+            }
+            case "WAIT_FOR_BATCH": {
+              this.waitForBatchReplay = undefined;
+              break;
+            }
+          }
+
+          this.paused = false;
+          this.nextResumeAfter = undefined;
+          this.waitForPostStart = false;
+
+          for (let i = 0; i < completions.length; i++) {
+            const completion = completions[i];
+
+            if (!completion) continue;
+
+            // TODO: signal to the worker that a task run has completed that it was waiting for
+            // this.#backgroundWorker.taskRunCompletedNotification(completion);
+          }
+        },
+        RESUME_AFTER_DURATION: async (message) => {
+          if (!this.paused) {
+            logger.error("worker not paused", {
+              attemptId: message.attemptId,
+            });
+            return;
+          }
+
+          if (this.nextResumeAfter !== "WAIT_FOR_DURATION") {
+            logger.error("not waiting to resume after duration", {
+              nextResumeAfter: this.nextResumeAfter,
+            });
+            return;
+          }
+
+          this.#resumeAfterDuration();
+        },
+        EXECUTE_TASK_RUN: async () => {
+          // These messages should only be received by old workers that don't support lazy attempts
+          this.#failRun(
+            this.runId,
+            "Received deprecated EXECUTE_TASK_RUN message. Please contact us if you see this error."
+          );
+        },
+        EXECUTE_TASK_RUN_LAZY_ATTEMPT: async (message) => {
+          this.readyForLazyAttemptReplay = undefined;
+
+          if (this.executing) {
+            logger.error("dropping execute request, already executing");
+            return;
+          }
+
+          const attemptCount = message.lazyPayload.attemptCount ?? 0;
+
+          logger.log("execute attempt counts", { attemptCount, completed: this.completed.size });
+
+          if (this.completed.size > 0 && this.completed.size >= attemptCount + 1) {
+            logger.error("dropping execute request, already completed");
+            return;
+          }
+
+          this.executing = true;
+
+          const createAttempt = await defaultBackoff.execute(async ({ retry }) => {
+            logger.log("Create task run attempt with backoff", { retry });
+
+            return await this.#coordinatorSocket.socket
+              .timeout(15_000)
+              .emitWithAck("CREATE_TASK_RUN_ATTEMPT", {
+                version: "v1",
+                runId: message.lazyPayload.runId,
+              });
+          });
+
+          if (!createAttempt.success) {
+            this.#failRun(
+              message.lazyPayload.runId,
+              `Failed to create attempt: ${createAttempt.cause}. ${createAttempt.error}`
+            );
+            return;
+          }
+
+          if (!createAttempt.result.success) {
+            this.#failRun(
+              message.lazyPayload.runId,
+              createAttempt.result.reason ?? "Failed to create attempt"
+            );
+            return;
+          }
+
+          const { execution } = createAttempt.result.executionPayload;
+          const { traceContext, environment } = message.lazyPayload;
+
+          // TODO: execute the task run lazy attempt
+          // try {
+          //   const { completion, execution } =
+          //     await this.#backgroundWorker.executeTaskRunLazyAttempt(message.lazyPayload);
+
+          //   logger.log("completed", completion);
+
+          //   this.completed.add(execution.attempt.id);
+
+          //   await this.#submitAttemptCompletion(execution, completion);
+          // } catch (error) {
+          //   logger.error("Failed to complete lazy attempt", {
+          //     error,
+          //   });
+
+          //   this.#failRun(message.lazyPayload.runId, error);
+          // }
+        },
+        REQUEST_ATTEMPT_CANCELLATION: async (message) => {
+          if (!this.executing) {
+            logger.log("dropping cancel request, not executing", { status: this.#status });
+            return;
+          }
+
+          logger.log("cancelling attempt", { attemptId: message.attemptId, status: this.#status });
+
+          // TODO: cancel the attempt
+          // await this.#backgroundWorker.cancelAttempt(message.attemptId);
+        },
+        REQUEST_EXIT: async (message) => {
+          if (message.version === "v2" && message.delayInMs) {
+            logger.log("exit requested with delay", { delayInMs: message.delayInMs });
+            await timeout(message.delayInMs);
+          }
+
+          this.#coordinatorSocket.close();
+          process.exit(0);
+        },
+        READY_FOR_RETRY: async (message) => {
+          if (this.completed.size < 1) {
+            logger.error("Received READY_FOR_RETRY but no completions yet. This is a bug.");
+            return;
+          }
+
+          await this.#readyForLazyAttempt();
+        },
+      },
+      // MARK: ON CONNECTION
+      onConnection: async (socket, handler, sender, logger) => {
+        logger.log("connected to coordinator", {
+          status: this.#status,
+          connectionCount: ++this.connectionCount,
+        });
+
+        // We need to send our current state to the coordinator
+        socket.emit("SET_STATE", {
+          version: "v1",
+          attemptFriendlyId: this.attemptFriendlyId,
+        });
+
+        try {
+          if (this.waitForPostStart) {
+            logger.log("skip connection handler, waiting for post start hook");
+            return;
+          }
+
+          if (this.paused) {
+            if (!this.nextResumeAfter) {
+              logger.error("Missing next resume reason", { status: this.#status });
+
+              this.#emitUnrecoverableError(
+                "NoNextResume",
+                "Next resume reason not set while resuming from paused state"
+              );
+
+              return;
+            }
+
+            if (!this.attemptFriendlyId) {
+              logger.error("Missing attempt friendly ID", { status: this.#status });
+
+              this.#emitUnrecoverableError(
+                "NoAttemptId",
+                "Attempt ID not set while resuming from paused state"
+              );
+
+              return;
+            }
+
+            socket.emit("READY_FOR_RESUME", {
+              version: "v1",
+              attemptFriendlyId: this.attemptFriendlyId,
+              type: this.nextResumeAfter,
+            });
+
+            return;
+          }
+
+          if (this.executing) {
+            return;
+          }
+
+          process.removeAllListeners("uncaughtException");
+          process.on("uncaughtException", (error) => {
+            console.error("Uncaught exception during run", error);
+            this.#failRun(this.runId, error);
+          });
+
+          await this.#readyForLazyAttempt();
+        } catch (error) {
+          logger.error("connection handler error", { error });
+        } finally {
+          if (this.connectionCount === 1) {
+            // Skip replays if this is the first connection, including post start
+            return;
+          }
+
+          // This is a reconnect, so handle replays
+          this.#handleReplays();
+        }
+      },
+      onError: async (socket, err, logger) => {
+        logger.error("onError", {
+          error: {
+            name: err.name,
+            message: err.message,
+          },
+        });
+      },
+    });
+
+    return coordinatorConnection;
+  }
+
+  // MARK: REPLAYS
+  async #handleReplays() {
+    const backoff = new ExponentialBackoff().type("FullJitter").maxRetries(3);
+    const replayCancellationDelay = 20_000;
+
+    if (this.waitForTaskReplay) {
+      logger.log("replaying wait for task", { ...this.waitForTaskReplay });
+
+      const { idempotencyKey, message, attempt } = this.waitForTaskReplay;
+
+      // Give the platform some time to send RESUME_AFTER_DEPENDENCY
+      await timeout(replayCancellationDelay);
+
+      if (!this.waitForTaskReplay) {
+        logger.error("wait for task replay cancelled, discarding", {
+          originalMessage: { idempotencyKey, message, attempt },
+        });
+
+        return;
+      }
+
+      if (idempotencyKey !== this.waitForTaskReplay.idempotencyKey) {
+        logger.error("wait for task replay idempotency key mismatch, discarding", {
+          originalMessage: { idempotencyKey, message, attempt },
+          newMessage: this.waitForTaskReplay,
+        });
+
+        return;
+      }
+
+      try {
+        await backoff.wait(attempt + 1);
+
+        await this.#waitForTaskHandlerFactory("replay")(message, idempotencyKey);
+      } catch (error) {
+        if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
+          logger.error("wait for task replay retry limit exceeded", { error });
+        } else {
+          logger.error("wait for task replay error", { error });
+        }
+      }
+
+      return;
+    }
+
+    if (this.waitForBatchReplay) {
+      logger.log("replaying wait for batch", {
+        ...this.waitForBatchReplay,
+        cancellationDelay: replayCancellationDelay,
+      });
+
+      const { idempotencyKey, message, attempt } = this.waitForBatchReplay;
+
+      // Give the platform some time to send RESUME_AFTER_DEPENDENCY
+      await timeout(replayCancellationDelay);
+
+      if (!this.waitForBatchReplay) {
+        logger.error("wait for batch replay cancelled, discarding", {
+          originalMessage: { idempotencyKey, message, attempt },
+        });
+
+        return;
+      }
+
+      if (idempotencyKey !== this.waitForBatchReplay.idempotencyKey) {
+        logger.error("wait for batch replay idempotency key mismatch, discarding", {
+          originalMessage: { idempotencyKey, message, attempt },
+          newMessage: this.waitForBatchReplay,
+        });
+
+        return;
+      }
+
+      try {
+        await backoff.wait(attempt + 1);
+
+        await this.#waitForBatchHandlerFactory("replay")(message, idempotencyKey);
+      } catch (error) {
+        if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
+          logger.error("wait for batch replay retry limit exceeded", { error });
+        } else {
+          logger.error("wait for batch replay error", { error });
+        }
+      }
+
+      return;
+    }
+  }
+
+  // MARK: HTTP SERVER
+  #createHttpServer() {
+    const httpServer = createServer(async (req, res) => {
+      logger.log(`[${req.method}]`, req.url);
+      const reply = new HttpReply(res);
+
+      try {
+        const url = new URL(req.url ?? "", `http://${req.headers.host}`);
+
+        switch (url.pathname) {
+          case "/health": {
+            return reply.text("ok");
+          }
+
+          case "/status": {
+            return reply.json(this.#status);
+          }
+
+          case "/connect": {
+            this.#coordinatorSocket.connect();
+
+            return reply.text("Connected to coordinator");
+          }
+
+          case "/close": {
+            this.#coordinatorSocket.close();
+            this.connectionCount = 0;
+
+            return reply.text("Disconnected from coordinator");
+          }
+
+          case "/test": {
+            await this.#coordinatorSocket.socket.timeout(10_000).emitWithAck("TEST", {
+              version: "v1",
+            });
+
+            return reply.text("Received ACK from coordinator");
+          }
+
+          case "/preStop": {
+            const cause = PreStopCauses.safeParse(url.searchParams.get("cause"));
+
+            if (!cause.success) {
+              logger.error("Failed to parse cause", { cause });
+              return reply.text("Failed to parse cause", 400);
+            }
+
+            switch (cause.data) {
+              case "terminate": {
+                break;
+              }
+              default: {
+                logger.error("Unhandled cause", { cause: cause });
+                break;
+              }
+            }
+
+            return reply.text("preStop ok");
+          }
+
+          case "/postStart": {
+            const cause = PostStartCauses.safeParse(url.searchParams.get("cause"));
+
+            if (!cause.success) {
+              logger.error("Failed to parse cause", { cause });
+              return reply.text("Failed to parse cause", 400);
+            }
+
+            switch (cause.data) {
+              case "index": {
+                break;
+              }
+              case "create": {
+                break;
+              }
+              case "restore": {
+                await this.#reconnectAfterPostStart();
+                this.restoreNotification.post();
+                break;
+              }
+              default: {
+                logger.error("Unhandled cause", { cause: cause });
+                break;
+              }
+            }
+
+            return reply.text("postStart ok");
+          }
+
+          default: {
+            return reply.empty(404);
+          }
+        }
+      } catch (error) {
+        logger.error("HTTP server error", { error });
+        reply.empty(500);
+      }
+
+      return;
+    });
+
+    httpServer.on("clientError", (err, socket) => {
+      socket.end("HTTP/1.1 400 Bad Request\r\n\r\n");
+    });
+
+    httpServer.on("listening", () => {
+      logger.log("http server listening on port", this.#httpPort);
+    });
+
+    httpServer.on("error", async (error) => {
+      // @ts-expect-error
+      if (error.code != "EADDRINUSE") {
+        return;
+      }
+
+      logger.error(`port ${this.#httpPort} already in use, retrying with random port..`);
+
+      this.#httpPort = getRandomPortNumber();
+
+      await timeout(100);
+      this.start();
+    });
+
+    return httpServer;
+  }
+
+  get #status() {
+    return {
+      executing: this.executing,
+      paused: this.paused,
+      completed: this.completed.size,
+      nextResumeAfter: this.nextResumeAfter,
+      waitForPostStart: this.waitForPostStart,
+      attemptFriendlyId: this.attemptFriendlyId,
+      attemptNumber: this.attemptNumber,
+      waitForTaskReplay: this.waitForTaskReplay,
+      waitForBatchReplay: this.waitForBatchReplay,
+    };
+  }
+
+  #emitUnrecoverableError(name: string, message: string) {
+    this.#coordinatorSocket.socket.emit("UNRECOVERABLE_ERROR", {
+      version: "v1",
+      error: {
+        name,
+        message,
+      },
+    });
+  }
+
+  async start() {
+    this.#httpServer.listen(this.#httpPort, this.host);
+  }
+}
+
+const workerManifest = await loadWorkerManifest();
+
+const prodWorker = new ProdWorker(HTTP_SERVER_PORT, workerManifest);
+await prodWorker.start();
+
+function gatherProcessEnv() {
+  const env = {
+    NODE_ENV: process.env.NODE_ENV ?? "production",
+    PATH: process.env.PATH,
+    USER: process.env.USER,
+    SHELL: process.env.SHELL,
+    LANG: process.env.LANG,
+    TERM: process.env.TERM,
+    NODE_PATH: process.env.NODE_PATH,
+    HOME: process.env.HOME,
+    NODE_EXTRA_CA_CERTS: process.env.NODE_EXTRA_CA_CERTS,
+  };
+
+  // Filter out undefined values
+  return Object.fromEntries(Object.entries(env).filter(([key, value]) => value !== undefined));
+}
+
+async function loadWorkerManifest() {
+  const manifestContents = await readFile("./index.json", "utf-8");
+  const raw = JSON.parse(manifestContents);
+
+  return WorkerManifest.parse(raw);
+}
diff --git a/packages/cli-v3/src/entryPoints/dev.ts b/packages/cli-v3/src/entryPoints/dev.ts
index 2980f83f5..2c2580c8c 100644
--- a/packages/cli-v3/src/entryPoints/dev.ts
+++ b/packages/cli-v3/src/entryPoints/dev.ts
@@ -1,7 +1,6 @@
 import type { Tracer } from "@opentelemetry/api";
 import type { Logger } from "@opentelemetry/api-logs";
 import {
-  childToWorkerMessages,
   clock,
   type HandleErrorFunction,
   logger,
@@ -10,10 +9,11 @@ import {
   taskCatalog,
   TaskRunErrorCodes,
   TaskRunExecution,
+  WorkerToExecutorMessageCatalog,
   TriggerConfig,
   TriggerTracer,
   WorkerManifest,
-  workerToChildMessages,
+  ExecutorToWorkerMessageCatalog,
 } from "@trigger.dev/core/v3";
 import { DevRuntimeManager } from "@trigger.dev/core/v3/dev";
 import {
@@ -29,7 +29,7 @@ import {
   TracingSDK,
   usage,
 } from "@trigger.dev/core/v3/workers";
-import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
+import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
 import { readFile } from "node:fs/promises";
 import sourceMapSupport from "source-map-support";
 import { VERSION } from "../version.js";
@@ -67,13 +67,6 @@ process.on("uncaughtException", function (error, origin) {
   }
 });
 
-const sender = new ZodMessageSender({
-  schema: childToWorkerMessages,
-  sender: async (message) => {
-    process.send?.(message);
-  },
-});
-
 taskCatalog.setGlobalTaskCatalog(new StandardTaskCatalog());
 const durableClock = new DurableClock();
 clock.setGlobalClock(durableClock);
@@ -156,10 +149,12 @@ let _execution: TaskRunExecution | undefined;
 let _isRunning = false;
 let _tracingSDK: TracingSDK | undefined;
 
-const handler = new ZodMessageHandler({
-  schema: workerToChildMessages,
-  messages: {
-    EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }) => {
+const zodIpc = new ZodIpcConnection({
+  listenSchema: ExecutorToWorkerMessageCatalog,
+  emitSchema: WorkerToExecutorMessageCatalog,
+  process,
+  handlers: {
+    EXECUTE_TASK_RUN: async ({ execution, traceContext, metadata }, sender) => {
       if (_isRunning) {
         console.error("Worker is already running a task");
 
@@ -184,6 +179,8 @@ const handler = new ZodMessageHandler({
       const { tracer, tracingSDK, consoleInterceptor, config, handleErrorFn, workerManifest } =
         await bootstrap();
 
+      _tracingSDK = tracingSDK;
+
       const taskManifest = workerManifest.tasks.find((t) => t.id === execution.task.id);
 
       if (!taskManifest) {
@@ -300,7 +297,7 @@ const handler = new ZodMessageHandler({
         }
       }
     },
-    CLEANUP: async ({ flush, kill }) => {
+    CLEANUP: async ({ flush, kill }, sender) => {
       if (kill) {
         await _tracingSDK?.flush();
         // Now we need to exit the process
@@ -314,10 +311,6 @@ const handler = new ZodMessageHandler({
   },
 });
 
-process.on("message", async (msg: any) => {
-  await handler.handleMessage(msg);
-});
-
 process.title = "trigger-dev-worker";
 
 async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeconds: number = 30) {
@@ -325,7 +318,7 @@ async function asyncHeartbeat(initialDelayInSeconds: number = 30, intervalInSeco
     while (true) {
       if (_isRunning && _execution) {
         try {
-          await sender.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
+          await zodIpc.send("TASK_HEARTBEAT", { id: _execution.attempt.id });
         } catch (err) {
           console.error("Failed to send HEARTBEAT message", err);
         }
diff --git a/packages/cli-v3/src/entryPoints/indexer.ts b/packages/cli-v3/src/entryPoints/indexer.ts
index aa8577868..ed56bf38b 100644
--- a/packages/cli-v3/src/entryPoints/indexer.ts
+++ b/packages/cli-v3/src/entryPoints/indexer.ts
@@ -122,6 +122,9 @@ await sendMessageInCatalog(
     manifest: {
       tasks,
       configPath: buildManifest.configPath,
+      runtime: buildManifest.runtime,
+      workerEntryPoint: buildManifest.workerEntryPoint,
+      loaderEntryPoint: buildManifest.loaderEntryPoint,
     },
   },
   async (msg) => {
diff --git a/packages/cli-v3/src/dev/errors.ts b/packages/cli-v3/src/executions/errors.ts
similarity index 100%
rename from packages/cli-v3/src/dev/errors.ts
rename to packages/cli-v3/src/executions/errors.ts
diff --git a/packages/cli-v3/src/executions/taskRunProcess.ts b/packages/cli-v3/src/executions/taskRunProcess.ts
new file mode 100644
index 000000000..eec0b0840
--- /dev/null
+++ b/packages/cli-v3/src/executions/taskRunProcess.ts
@@ -0,0 +1,407 @@
+import {
+  ServerBackgroundWorker,
+  TaskRunExecution,
+  TaskRunExecutionPayload,
+  TaskRunExecutionResult,
+  WorkerToExecutorMessageCatalog,
+  ExecutorToWorkerMessageCatalog,
+  WorkerManifest,
+} from "@trigger.dev/core/v3";
+import {
+  type WorkerToExecutorProcessConnection,
+  ZodIpcConnection,
+} from "@trigger.dev/core/v3/zodIpc";
+import { Evt } from "evt";
+import { ChildProcess, fork } from "node:child_process";
+import { chalkError, chalkGrey, chalkRun, prettyPrintDate } from "../utilities/cliOutput.js";
+
+import { execPathForRuntime } from "@trigger.dev/core/v3/build";
+import { logger } from "../utilities/logger.js";
+import {
+  CancelledProcessError,
+  CleanupProcessError,
+  GracefulExitTimeoutError,
+  UnexpectedExitError,
+} from "./errors.js";
+import { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
+
+export type OnWaitForDurationMessage = InferSocketMessageSchema<
+  typeof WorkerToExecutorMessageCatalog,
+  "WAIT_FOR_DURATION"
+>;
+export type OnWaitForTaskMessage = InferSocketMessageSchema<
+  typeof WorkerToExecutorMessageCatalog,
+  "WAIT_FOR_TASK"
+>;
+export type OnWaitForBatchMessage = InferSocketMessageSchema<
+  typeof WorkerToExecutorMessageCatalog,
+  "WAIT_FOR_BATCH"
+>;
+
+export type TaskRunProcessOptions = {
+  workerManifest: WorkerManifest;
+  serverWorker: ServerBackgroundWorker;
+  env: Record<string, string>;
+  payload: TaskRunExecutionPayload;
+
+  cwd?: string;
+  messageId?: string;
+};
+
+export class TaskRunProcess {
+  private _ipc?: WorkerToExecutorProcessConnection;
+  private _child: ChildProcess | undefined;
+  private _childPid?: number;
+  private _attemptPromises: Map<
+    string,
+    { resolver: (value: TaskRunExecutionResult) => void; rejecter: (err?: any) => void }
+  > = new Map();
+  private _attemptStatuses: Map<string, "PENDING" | "REJECTED" | "RESOLVED"> = new Map();
+  private _currentExecution: TaskRunExecution | undefined;
+  private _gracefulExitTimeoutElapsed: boolean = false;
+  private _isBeingKilled: boolean = false;
+  private _isBeingCancelled: boolean = false;
+  private _stderr: Array<string> = [];
+  /**
+   * @deprecated use onTaskRunHeartbeat instead
+   */
+  public onTaskHeartbeat: Evt<string> = new Evt();
+  public onTaskRunHeartbeat: Evt<string> = new Evt();
+  public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
+    new Evt();
+  public onIsBeingKilled: Evt<TaskRunProcess> = new Evt();
+  public onReadyToDispose: Evt<TaskRunProcess> = new Evt();
+
+  public onWaitForDuration: Evt<OnWaitForDurationMessage> = new Evt();
+  public onWaitForTask: Evt<OnWaitForTaskMessage> = new Evt();
+  public onWaitForBatch: Evt<OnWaitForBatchMessage> = new Evt();
+
+  constructor(public readonly options: TaskRunProcessOptions) {}
+
+  async cancel() {
+    this._isBeingCancelled = true;
+
+    await this.cleanup(true);
+  }
+
+  get runId() {
+    return this.options.payload.execution.run.id;
+  }
+
+  get isTest() {
+    return this.options.payload.execution.run.isTest;
+  }
+
+  get payload() {
+    return this.options.payload;
+  }
+
+  async initialize() {
+    const { env, workerManifest, cwd, messageId } = this.options;
+
+    const fullEnv = {
+      ...(this.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}),
+      ...env,
+      NODE_OPTIONS: workerManifest.loaderEntryPoint
+        ? `--import=${workerManifest.loaderEntryPoint} ${
+            env.NODE_OPTIONS ?? process.env.NODE_OPTIONS ?? ""
+          }`
+        : env.NODE_OPTIONS ?? process.env.NODE_OPTIONS ?? "",
+    };
+
+    logger.debug(`[${this.runId}] initializing task run process`, {
+      env: fullEnv,
+      path: workerManifest.workerEntryPoint,
+      cwd,
+    });
+
+    this._child = fork(workerManifest.workerEntryPoint, {
+      stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
+      cwd,
+      env: fullEnv,
+      execArgv: ["--trace-uncaught", "--no-warnings=ExperimentalWarning"],
+      execPath: execPathForRuntime(workerManifest.runtime),
+    });
+
+    this._childPid = this._child?.pid;
+
+    this._ipc = new ZodIpcConnection({
+      listenSchema: WorkerToExecutorMessageCatalog,
+      emitSchema: ExecutorToWorkerMessageCatalog,
+      process: this._child,
+      handlers: {
+        TASK_RUN_COMPLETED: async (message) => {
+          const { result, execution } = message;
+
+          const promiseStatus = this._attemptStatuses.get(execution.attempt.id);
+
+          if (promiseStatus !== "PENDING") {
+            return;
+          }
+
+          this._attemptStatuses.set(execution.attempt.id, "RESOLVED");
+
+          const attemptPromise = this._attemptPromises.get(execution.attempt.id);
+
+          if (!attemptPromise) {
+            return;
+          }
+
+          const { resolver } = attemptPromise;
+
+          resolver(result);
+        },
+        READY_TO_DISPOSE: async (message) => {
+          logger.debug(`[${this.runId}] task run process is ready to dispose`);
+
+          this.onReadyToDispose.post(this);
+        },
+        TASK_HEARTBEAT: async (message) => {
+          if (messageId) {
+            this.onTaskRunHeartbeat.post(messageId);
+          } else {
+            logger.debug(
+              "No message id for task heartbeat, falling back to (deprecated) attempt heartbeat",
+              { id: message.id }
+            );
+            this.onTaskHeartbeat.post(message.id);
+          }
+        },
+        WAIT_FOR_TASK: async (message) => {
+          this.onWaitForTask.post(message);
+        },
+        WAIT_FOR_BATCH: async (message) => {
+          this.onWaitForBatch.post(message);
+        },
+        WAIT_FOR_DURATION: async (message) => {
+          this.onWaitForDuration.post(message);
+        },
+      },
+    });
+
+    this._child.on("exit", this.#handleExit.bind(this));
+    this._child.stdout?.on("data", this.#handleLog.bind(this));
+    this._child.stderr?.on("data", this.#handleStdErr.bind(this));
+  }
+
+  async cleanup(kill = false, gracefulExitTimeoutElapsed = false) {
+    logger.debug("cleanup()", { kill, gracefulExitTimeoutElapsed });
+
+    if (kill && this._isBeingKilled) {
+      return;
+    }
+
+    if (kill) {
+      this._isBeingKilled = true;
+      this.onIsBeingKilled.post(this);
+    }
+
+    logger.debug("Cleaning up task run process", {
+      kill,
+      childPid: this._childPid,
+      realChildPid: this._child?.pid,
+    });
+
+    try {
+      await this._ipc?.sendWithAck(
+        "CLEANUP",
+        {
+          flush: true,
+          kill,
+        },
+        30_000
+      );
+    } catch (error) {
+      logger.debug("Error while cleaning up task run process", error);
+
+      if (kill) {
+        this.onReadyToDispose.post(this);
+      }
+    }
+
+    if (!kill) {
+      return;
+    }
+
+    // Set a timeout to kill the child process if it hasn't been killed within 5 seconds
+    setTimeout(() => {
+      if (this._child && !this._child.killed) {
+        logger.debug(`[${this.runId}] killing task run process after timeout`, { pid: this.pid });
+
+        this._child.kill();
+      }
+    }, 5000);
+  }
+
+  async execute(): Promise<TaskRunExecutionResult> {
+    let resolver: (value: TaskRunExecutionResult) => void;
+    let rejecter: (err?: any) => void;
+
+    const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => {
+      resolver = resolve;
+      rejecter = reject;
+    });
+
+    this._attemptStatuses.set(this.payload.execution.attempt.id, "PENDING");
+
+    // @ts-expect-error - We know that the resolver and rejecter are defined
+    this._attemptPromises.set(this.payload.execution.attempt.id, { resolver, rejecter });
+
+    const { execution, traceContext } = this.payload;
+
+    this._currentExecution = execution;
+
+    if (this._child?.connected && !this._isBeingKilled && !this._child.killed) {
+      await this._ipc?.send("EXECUTE_TASK_RUN", {
+        execution,
+        traceContext,
+        metadata: this.options.serverWorker,
+      });
+    }
+
+    const result = await promise;
+
+    this._currentExecution = undefined;
+
+    return result;
+  }
+
+  taskRunCompletedNotification(completion: TaskRunExecutionResult) {
+    if (!completion.ok && typeof completion.retry !== "undefined") {
+      logger.debug(
+        "Task run completed with error and wants to retry, won't send task run completed notification"
+      );
+      return;
+    }
+
+    if (!this._child?.connected || this._isBeingKilled || this._child.killed) {
+      logger.debug(
+        "Child process not connected or being killed, can't send task run completed notification"
+      );
+      return;
+    }
+
+    this._ipc?.send("TASK_RUN_COMPLETED_NOTIFICATION", {
+      version: "v2",
+      completion,
+    });
+  }
+
+  async #handleExit(code: number | null, signal: NodeJS.Signals | null) {
+    logger.debug("handling child exit", { code, signal });
+
+    // Go through all the attempts currently pending and reject them
+    for (const [id, status] of this._attemptStatuses.entries()) {
+      if (status === "PENDING") {
+        logger.debug("found pending attempt", { id });
+
+        this._attemptStatuses.set(id, "REJECTED");
+
+        const attemptPromise = this._attemptPromises.get(id);
+
+        if (!attemptPromise) {
+          continue;
+        }
+
+        const { rejecter } = attemptPromise;
+
+        if (this._isBeingCancelled) {
+          rejecter(new CancelledProcessError());
+        } else if (this._gracefulExitTimeoutElapsed) {
+          // Order matters, this has to be before the graceful exit timeout
+          rejecter(new GracefulExitTimeoutError());
+        } else if (this._isBeingKilled) {
+          rejecter(new CleanupProcessError());
+        } else {
+          rejecter(
+            new UnexpectedExitError(
+              code ?? -1,
+              signal,
+              this._stderr.length ? this._stderr.join("\n") : undefined
+            )
+          );
+        }
+      }
+    }
+
+    this.onExit.post({ code, signal, pid: this.pid });
+  }
+
+  #handleLog(data: Buffer) {
+    if (!this._currentExecution) {
+      logger.log(`${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`);
+
+      return;
+    }
+
+    const runId = chalkRun(
+      `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}`
+    );
+
+    logger.log(
+      `${chalkGrey("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${data.toString()}`
+    );
+  }
+
+  #handleStdErr(data: Buffer) {
+    if (this._isBeingKilled) {
+      return;
+    }
+
+    if (!this._currentExecution) {
+      logger.log(`${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${data.toString()}`);
+
+      return;
+    }
+
+    const runId = chalkRun(
+      `${this._currentExecution.run.id}.${this._currentExecution.attempt.number}`
+    );
+
+    const errorLine = data.toString();
+
+    logger.log(
+      `${chalkError("○")} ${chalkGrey(prettyPrintDate(new Date()))} ${runId} ${errorLine}`
+    );
+
+    if (this._stderr.length > 100) {
+      this._stderr.shift();
+    }
+    this._stderr.push(errorLine);
+  }
+
+  #kill() {
+    logger.debug(`[${this.runId}] #kill()`, { pid: this.pid });
+
+    if (this._child && !this._child.killed) {
+      this._child?.kill();
+    }
+  }
+
+  async kill(signal?: number | NodeJS.Signals, timeoutInMs?: number) {
+    logger.debug(`[${this.runId}] killing task run process`, {
+      signal,
+      timeoutInMs,
+      pid: this.pid,
+    });
+
+    this._isBeingKilled = true;
+
+    const killTimeout = this.onExit.waitFor(timeoutInMs);
+
+    this.onIsBeingKilled.post(this);
+    this._child?.kill(signal);
+
+    if (timeoutInMs) {
+      await killTimeout;
+    }
+  }
+
+  get isBeingKilled() {
+    return this._isBeingKilled || this._child?.killed;
+  }
+
+  get pid() {
+    return this._childPid;
+  }
+}
diff --git a/packages/core/src/v3/runtime/prodRuntimeManager.ts b/packages/core/src/v3/runtime/prodRuntimeManager.ts
index 6b9abfe30..4fd63e04b 100644
--- a/packages/core/src/v3/runtime/prodRuntimeManager.ts
+++ b/packages/core/src/v3/runtime/prodRuntimeManager.ts
@@ -1,12 +1,10 @@
 import { clock } from "../clock-api.js";
 import {
   BatchTaskRunExecutionResult,
-  ProdChildToWorkerMessages,
-  ProdWorkerToChildMessages,
   TaskRunContext,
   TaskRunExecutionResult,
 } from "../schemas/index.js";
-import { ZodIpcConnection } from "../zodIpc.js";
+import { ExecutorToWorkerProcessConnection } from "../zodIpc.js";
 import { RuntimeManager } from "./manager.js";
 
 export type ProdRuntimeManagerOptions = {
@@ -24,10 +22,7 @@ export class ProdRuntimeManager implements RuntimeManager {
   _waitForDuration: { resolve: (value: void) => void; reject: (err?: any) => void } | undefined;
 
   constructor(
-    private ipc: ZodIpcConnection<
-      typeof ProdWorkerToChildMessages,
-      typeof ProdChildToWorkerMessages
-    >,
+    private ipc: ExecutorToWorkerProcessConnection,
     private options: ProdRuntimeManagerOptions = {}
   ) {}
 
diff --git a/packages/core/src/v3/schemas/build.ts b/packages/core/src/v3/schemas/build.ts
index c8f51922c..3b93a72bc 100644
--- a/packages/core/src/v3/schemas/build.ts
+++ b/packages/core/src/v3/schemas/build.ts
@@ -66,6 +66,9 @@ export type IndexMessage = z.infer<typeof IndexMessage>;
 export const WorkerManifest = z.object({
   configPath: z.string(),
   tasks: TaskManifest.array(),
+  workerEntryPoint: z.string(),
+  loaderEntryPoint: z.string().optional(),
+  runtime: BuildRuntime,
 });
 
 export type WorkerManifest = z.infer<typeof WorkerManifest>;
diff --git a/packages/core/src/v3/schemas/messages.ts b/packages/core/src/v3/schemas/messages.ts
index cfb103281..bba2e99a0 100644
--- a/packages/core/src/v3/schemas/messages.ts
+++ b/packages/core/src/v3/schemas/messages.ts
@@ -1,21 +1,20 @@
 import { z } from "zod";
+import { WorkerManifest } from "./build.js";
 import {
   MachinePreset,
   TaskRunExecution,
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
 } from "./common.js";
+import { TaskResource } from "./resources.js";
 import {
   EnvironmentType,
   ProdTaskRunExecution,
   ProdTaskRunExecutionPayload,
   TaskManifest,
   TaskRunExecutionLazyAttemptPayload,
-  TaskRunExecutionPayload,
   WaitReason,
 } from "./schemas.js";
-import { TaskResource } from "./resources.js";
-import { BuildManifest, WorkerManifest } from "./build.js";
 
 export const BackgroundWorkerServerMessages = z.discriminatedUnion("type", [
   z.object({
@@ -108,32 +107,6 @@ export const clientWebsocketMessages = {
   }),
 };
 
-export const workerToChildMessages = {
-  INDEX: z.object({}),
-  EXECUTE_TASK_RUN: z.object({
-    version: z.literal("v1").default("v1"),
-    execution: TaskRunExecution,
-    traceContext: z.record(z.unknown()),
-    metadata: ServerBackgroundWorker,
-  }),
-  TASK_RUN_COMPLETED_NOTIFICATION: z.discriminatedUnion("version", [
-    z.object({
-      version: z.literal("v1"),
-      completion: TaskRunExecutionResult,
-      execution: TaskRunExecution,
-    }),
-    z.object({
-      version: z.literal("v2"),
-      completion: TaskRunExecutionResult,
-    }),
-  ]),
-  CLEANUP: z.object({
-    version: z.literal("v1").default("v1"),
-    flush: z.boolean().default(false),
-    kill: z.boolean().default(true),
-  }),
-};
-
 export const UncaughtExceptionMessage = z.object({
   version: z.literal("v1").default("v1"),
   error: z.object({
@@ -152,37 +125,6 @@ export const TaskMetadataFailedToParseData = z.object({
   }),
 });
 
-export const childToWorkerMessages = {
-  TASK_RUN_COMPLETED: z.object({
-    version: z.literal("v1").default("v1"),
-    execution: TaskRunExecution,
-    result: TaskRunExecutionResult,
-  }),
-  TASK_HEARTBEAT: z.object({
-    version: z.literal("v1").default("v1"),
-    id: z.string(),
-  }),
-  TASK_RUN_HEARTBEAT: z.object({
-    version: z.literal("v1").default("v1"),
-    id: z.string(),
-  }),
-  READY_TO_DISPOSE: z.undefined(),
-  WAIT_FOR_DURATION: z.object({
-    version: z.literal("v1").default("v1"),
-    ms: z.number(),
-  }),
-  WAIT_FOR_TASK: z.object({
-    version: z.literal("v1").default("v1"),
-    id: z.string(),
-  }),
-  WAIT_FOR_BATCH: z.object({
-    version: z.literal("v1").default("v1"),
-    id: z.string(),
-    runs: z.string().array(),
-  }),
-  UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
-};
-
 export const indexerToWorkerMessages = {
   INDEX_COMPLETE: z.object({
     version: z.literal("v1").default("v1"),
@@ -192,7 +134,7 @@ export const indexerToWorkerMessages = {
   UNCAUGHT_EXCEPTION: UncaughtExceptionMessage,
 };
 
-export const ProdChildToWorkerMessages = {
+export const WorkerToExecutorMessageCatalog = {
   TASK_RUN_COMPLETED: {
     message: z.object({
       version: z.literal("v1").default("v1"),
@@ -200,15 +142,6 @@ export const ProdChildToWorkerMessages = {
       result: TaskRunExecutionResult,
     }),
   },
-  TASKS_READY: {
-    message: z.object({
-      version: z.literal("v1").default("v1"),
-      tasks: TaskManifest.array(),
-    }),
-  },
-  TASKS_FAILED_TO_PARSE: {
-    message: TaskMetadataFailedToParseData,
-  },
   TASK_HEARTBEAT: {
     message: z.object({
       version: z.literal("v1").default("v1"),
@@ -244,7 +177,7 @@ export const ProdChildToWorkerMessages = {
   },
 };
 
-export const ProdWorkerToChildMessages = {
+export const ExecutorToWorkerMessageCatalog = {
   EXECUTE_TASK_RUN: {
     message: z.object({
       version: z.literal("v1").default("v1"),
diff --git a/packages/core/src/v3/zodIpc.ts b/packages/core/src/v3/zodIpc.ts
index e0d2113a2..40dd32b2f 100644
--- a/packages/core/src/v3/zodIpc.ts
+++ b/packages/core/src/v3/zodIpc.ts
@@ -11,6 +11,10 @@ import {
 import { z } from "zod";
 import { ZodSchemaParsedError } from "./zodMessageHandler.js";
 import { inspect } from "node:util";
+import {
+  ExecutorToWorkerMessageCatalog,
+  WorkerToExecutorMessageCatalog,
+} from "./schemas/messages.js";
 
 interface ZodIpcMessageSender<TEmitCatalog extends ZodSocketMessageCatalogSchema> {
   send<K extends GetSocketMessagesWithoutCallback<TEmitCatalog>>(
@@ -335,3 +339,13 @@ export class ZodIpcConnection<
     });
   }
 }
+
+export type WorkerToExecutorProcessConnection = ZodIpcConnection<
+  typeof WorkerToExecutorMessageCatalog,
+  typeof ExecutorToWorkerMessageCatalog
+>;
+
+export type ExecutorToWorkerProcessConnection = ZodIpcConnection<
+  typeof ExecutorToWorkerMessageCatalog,
+  typeof WorkerToExecutorMessageCatalog
+>;
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index cefa9c012..99637c863 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -985,8 +985,8 @@ importers:
         specifier: ^1.2.0
         version: 1.3.1
       tinyexec:
-        specifier: ^0.1.4
-        version: 0.1.4
+        specifier: ^0.2.0
+        version: 0.2.0
       ws:
         specifier: ^8.18.0
         version: 8.18.0
@@ -26153,8 +26153,8 @@ packages:
     resolution: {integrity: sha512-XPaBkWQJdsf3pLKJV9p4qN/S+fm2Oj8AIPo1BTUhg5oxkvm9+SVEGFdhyOz7tTdUTfvxMiAs4sp6/eZO2Ew+pw==}
     dev: false
 
-  /tinyexec@0.1.4:
-    resolution: {integrity: sha512-Ba2ELcNnnWkgqnAJBouhcsDsYitbD9LIAVNSz3746u50f+tlF3wO0uB3uqyz8NHFSTpv23qtT47XGDw8pXW5DA==}
+  /tinyexec@0.2.0:
+    resolution: {integrity: sha512-au8dwv4xKSDR+Fw52csDo3wcDztPdne2oM1o/7LFro4h6bdFmvyUAeAfX40pwDtzHgRFqz1XWaUqgKS2G83/ig==}
     dev: false
 
   /tinyglobby@0.2.2:
diff --git a/references/bun-catalog/package.json b/references/bun-catalog/package.json
index 412676d34..fa1161c1c 100644
--- a/references/bun-catalog/package.json
+++ b/references/bun-catalog/package.json
@@ -4,7 +4,8 @@
   "private": true,
   "type": "module",
   "scripts": {
-    "dev:trigger": "triggerdev dev"
+    "dev:trigger": "triggerdev dev",
+    "deploy": "triggerdev deploy"
   },
   "dependencies": {
     "@trigger.dev/sdk": "workspace:*"