diff --git a/apps/webapp/app/components/primitives/TreeView/TreeView.tsx b/apps/webapp/app/components/primitives/TreeView/TreeView.tsx
index f3d955724..69b437e13 100644
--- a/apps/webapp/app/components/primitives/TreeView/TreeView.tsx
+++ b/apps/webapp/app/components/primitives/TreeView/TreeView.tsx
@@ -1,25 +1,9 @@
 import { VirtualItem, Virtualizer, useVirtualizer } from "@tanstack/react-virtual";
-import {
-  Fragment,
-  MutableRefObject,
-  RefObject,
-  useCallback,
-  useEffect,
-  useImperativeHandle,
-  useReducer,
-  useRef,
-  useState,
-} from "react";
+import { MutableRefObject, RefObject, useCallback, useEffect, useReducer, useRef } from "react";
 import { UnmountClosed } from "react-collapse";
 import { cn } from "~/utils/cn";
-import { Changes, NodeState, NodesState, reducer } from "./reducer";
-import {
-  applyFilterToState,
-  concreteStateFromInput,
-  firstVisibleNode,
-  lastVisibleNode,
-  selectedIdFromState,
-} from "./utils";
+import { NodeState, NodesState, reducer } from "./reducer";
+import { applyFilterToState, concreteStateFromInput, selectedIdFromState } from "./utils";
 
 export type TreeViewProps<TData> = {
   tree: FlatTree<TData>;
@@ -232,6 +216,7 @@ export function useTree<TData>({
         index,
       });
     },
+    overscan: 20,
   });
 
   const scrollToNodeFn = useCallback(
diff --git a/apps/webapp/app/consts.ts b/apps/webapp/app/consts.ts
index d4c0c4c1d..b576dcfb5 100644
--- a/apps/webapp/app/consts.ts
+++ b/apps/webapp/app/consts.ts
@@ -10,3 +10,4 @@ export const RUN_CHUNK_EXECUTION_BUFFER = 350;
 export const MAX_RUN_CHUNK_EXECUTION_LIMIT = 120000; // 2 minutes
 export const VERCEL_RESPONSE_TIMEOUT_STATUS_CODES = [408, 504];
 export const MAX_BATCH_TRIGGER_ITEMS = 100;
+export const MAX_TASK_RUN_ATTEMPTS = 250;
diff --git a/apps/webapp/app/v3/marqs.server.ts b/apps/webapp/app/v3/marqs.server.ts
index cfc79d082..e9be68233 100644
--- a/apps/webapp/app/v3/marqs.server.ts
+++ b/apps/webapp/app/v3/marqs.server.ts
@@ -611,14 +611,14 @@ export class MarQS {
       String(this.options.defaultConcurrency ?? 10)
     );
 
-    logger.debug("Dequeue message result", {
-      result,
-    });
-
     if (!result) {
       return;
     }
 
+    logger.debug("Dequeue message result", {
+      result,
+    });
+
     if (result.length !== 2) {
       return;
     }
diff --git a/apps/webapp/app/v3/marqs/devQueueConsumer.server.ts b/apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
index fa05a4f05..637ae2c5c 100644
--- a/apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
+++ b/apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
@@ -18,6 +18,7 @@ import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { CompleteAttemptService } from "../services/completeAttempt.server";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";
 import { DevSubscriber, devPubSub } from "./devPubSub.server";
+import { CancelTaskRunService } from "../services/cancelTaskRun.server";
 
 const tracer = trace.getTracer("devQueueConsumer");
 
@@ -50,6 +51,7 @@ export class DevQueueConsumer {
   private _currentSpan: Span | undefined;
   private _endSpanInNextIteration = false;
   private _inProgressAttempts: Map<string, string> = new Map(); // Keys are task attempt friendly IDs, values are TaskRun ids/queue message ids
+  private _inProgressRuns: Map<string, string> = new Map(); // Keys are task run friendly IDs, values are TaskRun internal ids/queue message ids
 
   constructor(
     public env: AuthenticatedEnvironment,
@@ -123,7 +125,11 @@ export class DevQueueConsumer {
     logger.debug("Task run completed", { taskRunCompletion: completion, execution });
 
     const service = new CompleteAttemptService();
-    await service.call(completion, execution, this.env);
+    const result = await service.call(completion, execution, this.env);
+
+    if (result === "COMPLETED") {
+      this._inProgressRuns.delete(execution.run.id);
+    }
   }
 
   public async taskHeartbeat(workerId: string, id: string, seconds: number = 60) {
@@ -148,7 +154,7 @@ export class DevQueueConsumer {
     this._enabled = false;
 
     // We need to cancel all the in progress task run attempts and ack the messages so they will stop processing
-    await this.#cancelInProgressAttempts(reason);
+    await this.#cancelInProgressRunsAndAttempts(reason);
 
     // We need to unsubscribe from the background worker channels
     for (const [id, subscriber] of this._backgroundWorkerSubscriber) {
@@ -161,21 +167,44 @@ export class DevQueueConsumer {
     }
   }
 
-  async #cancelInProgressAttempts(reason: string) {
-    const service = new CancelAttemptService();
+  async #cancelInProgressRunsAndAttempts(reason: string) {
+    const cancelAttemptService = new CancelAttemptService();
+    const cancelTaskRunService = new CancelTaskRunService();
 
     const cancelledAt = new Date();
 
     const inProgressAttempts = new Map(this._inProgressAttempts);
+    const inProgressRuns = new Map(this._inProgressRuns);
 
     this._inProgressAttempts.clear();
+    this._inProgressRuns.clear();
 
-    logger.debug("Cancelling in progress attempts", {
+    const inProgressRunsWithNoInProgressAttempts: string[] = [];
+    const inProgressAttemptRunIds = new Set(inProgressAttempts.values());
+
+    for (const [runId, messageId] of inProgressRuns) {
+      if (!inProgressAttemptRunIds.has(messageId)) {
+        inProgressRunsWithNoInProgressAttempts.push(messageId);
+      }
+    }
+
+    logger.debug("Cancelling in progress runs and attempts", {
       attempts: Array.from(inProgressAttempts.keys()),
+      runs: Array.from(inProgressRuns.keys()),
     });
 
     for (const [attemptId, messageId] of inProgressAttempts) {
-      await this.#cancelInProgressAttempt(attemptId, messageId, service, cancelledAt, reason);
+      await this.#cancelInProgressAttempt(
+        attemptId,
+        messageId,
+        cancelAttemptService,
+        cancelledAt,
+        reason
+      );
+    }
+
+    for (const runId of inProgressRunsWithNoInProgressAttempts) {
+      await this.#cancelInProgressRun(runId, cancelTaskRunService, cancelledAt, reason);
     }
   }
 
@@ -199,6 +228,32 @@ export class DevQueueConsumer {
     }
   }
 
+  async #cancelInProgressRun(
+    runId: string,
+    service: CancelTaskRunService,
+    cancelledAt: Date,
+    reason: string
+  ) {
+    logger.debug("Cancelling in progress run", { runId });
+
+    const taskRun = await prisma.taskRun.findUnique({
+      where: { id: runId },
+    });
+
+    if (!taskRun) {
+      return;
+    }
+
+    try {
+      await service.call(taskRun, { reason, cancelAttempts: false, cancelledAt });
+    } catch (e) {
+      logger.error("Failed to cancel in progress run", {
+        runId,
+        error: e,
+      });
+    }
+  }
+
   #enable() {
     if (this._enabled) {
       return;
@@ -481,6 +536,7 @@ export class DevQueueConsumer {
       });
 
       this._inProgressAttempts.set(taskRunAttempt.friendlyId, message.messageId);
+      this._inProgressRuns.set(lockedTaskRun.friendlyId, message.messageId);
     } catch (e) {
       if (e instanceof Error) {
         this._currentSpan?.recordException(e);
@@ -499,6 +555,7 @@ export class DevQueueConsumer {
           data: {
             lockedAt: null,
             lockedById: null,
+            status: "PENDING",
           },
         }),
         prisma.taskRunAttempt.delete({
@@ -508,6 +565,9 @@ export class DevQueueConsumer {
         }),
       ]);
 
+      this._inProgressAttempts.delete(taskRunAttempt.friendlyId);
+      this._inProgressRuns.delete(lockedTaskRun.friendlyId);
+
       // Finally we need to nack the message so it can be retried
       await marqs?.nackMessage(message.messageId);
     } finally {
diff --git a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
index 78ec796cd..8e9782780 100644
--- a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+++ b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
@@ -300,7 +300,10 @@ export class SharedQueueConsumer {
           return;
         }
 
-        if (existingTaskRun.status !== "PENDING") {
+        if (
+          existingTaskRun.status !== "PENDING" &&
+          existingTaskRun.status !== "RETRYING_AFTER_FAILURE"
+        ) {
           logger.debug("Task run is not pending, aborting", {
             queueMessage: message.data,
             messageId: message.messageId,
diff --git a/apps/webapp/app/v3/services/cancelTaskRun.server.ts b/apps/webapp/app/v3/services/cancelTaskRun.server.ts
index 72e5fcb6d..5fc96991d 100644
--- a/apps/webapp/app/v3/services/cancelTaskRun.server.ts
+++ b/apps/webapp/app/v3/services/cancelTaskRun.server.ts
@@ -23,8 +23,21 @@ const CANCELLABLE_ATTEMPT_STATUSES: Array<TaskRunAttemptStatus> = [
   "PENDING",
 ];
 
+export type CancelTaskRunServiceOptions = {
+  reason?: string;
+  cancelAttempts?: boolean;
+  cancelledAt?: Date;
+};
+
 export class CancelTaskRunService extends BaseService {
-  public async call(taskRun: TaskRun) {
+  public async call(taskRun: TaskRun, options?: CancelTaskRunServiceOptions) {
+    const opts = {
+      reason: "Task run was cancelled by user",
+      cancelAttempts: true,
+      cancelledAt: new Date(),
+      ...options,
+    };
+
     // Make sure the task run is in a cancellable state
     if (!CANCELLABLE_STATUSES.includes(taskRun.status)) {
       return;
@@ -68,59 +81,61 @@ export class CancelTaskRunService extends BaseService {
 
     await Promise.all(
       inProgressEvents.map((event) => {
-        return eventRepository.cancelEvent(event, new Date(), "Task run was cancelled by user");
+        return eventRepository.cancelEvent(event, opts.cancelledAt, opts.reason);
       })
     );
 
     // Cancel any in progress attempts
-    for (const attempt of cancelledTaskRun.attempts) {
-      if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
-        // Signal the task run attempt to stop
-        await devPubSub.publish(
-          `backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
-          "CANCEL_ATTEMPT",
-          {
-            attemptId: attempt.friendlyId,
-            backgroundWorkerId: attempt.backgroundWorker.friendlyId,
-            taskRunId: cancelledTaskRun.friendlyId,
-          }
-        );
-      } else {
-        switch (attempt.status) {
-          case "EXECUTING": {
-            // We need to send a cancel message to the coordinator
-            socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
-              version: "v1",
-              attemptId: attempt.id,
-            });
+    if (opts.cancelAttempts) {
+      for (const attempt of cancelledTaskRun.attempts) {
+        if (attempt.runtimeEnvironment.type === "DEVELOPMENT") {
+          // Signal the task run attempt to stop
+          await devPubSub.publish(
+            `backgroundWorker:${attempt.backgroundWorkerId}:${attempt.id}`,
+            "CANCEL_ATTEMPT",
+            {
+              attemptId: attempt.friendlyId,
+              backgroundWorkerId: attempt.backgroundWorker.friendlyId,
+              taskRunId: cancelledTaskRun.friendlyId,
+            }
+          );
+        } else {
+          switch (attempt.status) {
+            case "EXECUTING": {
+              // We need to send a cancel message to the coordinator
+              socketIo.coordinatorNamespace.emit("REQUEST_ATTEMPT_CANCELLATION", {
+                version: "v1",
+                attemptId: attempt.id,
+              });
 
-            break;
-          }
-          case "PENDING":
-          case "PAUSED": {
-            logger.debug("Cancelling pending or paused attempt", {
-              attempt,
-            });
+              break;
+            }
+            case "PENDING":
+            case "PAUSED": {
+              logger.debug("Cancelling pending or paused attempt", {
+                attempt,
+              });
 
-            const service = new CancelAttemptService();
+              const service = new CancelAttemptService();
 
-            await service.call(
-              attempt.friendlyId,
-              taskRun.id,
-              new Date(),
-              "Task run was cancelled by user"
-            );
+              await service.call(
+                attempt.friendlyId,
+                taskRun.id,
+                new Date(),
+                "Task run was cancelled by user"
+              );
 
-            break;
-          }
-          case "CANCELED":
-          case "COMPLETED":
-          case "FAILED": {
-            // Do nothing
-            break;
-          }
-          default: {
-            assertUnreachable(attempt.status);
+              break;
+            }
+            case "CANCELED":
+            case "COMPLETED":
+            case "FAILED": {
+              // Do nothing
+              break;
+            }
+            default: {
+              assertUnreachable(attempt.status);
+            }
           }
         }
       }
diff --git a/apps/webapp/app/v3/services/completeAttempt.server.ts b/apps/webapp/app/v3/services/completeAttempt.server.ts
index c36e5bd0d..9c76a3436 100644
--- a/apps/webapp/app/v3/services/completeAttempt.server.ts
+++ b/apps/webapp/app/v3/services/completeAttempt.server.ts
@@ -1,12 +1,10 @@
 import { Attributes } from "@opentelemetry/api";
 import {
-  RetryOptions,
   TaskRunContext,
   TaskRunExecution,
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
-  defaultRetryOptions,
   flattenAttributes,
 } from "@trigger.dev/core/v3";
 import { PrismaClientOrTransaction } from "~/db.server";
@@ -16,8 +14,9 @@ import { safeJsonParse } from "~/utils/json";
 import { eventRepository } from "../eventRepository.server";
 import { marqs } from "../marqs.server";
 import { BaseService } from "./baseService.server";
-import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
+import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
+import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
 
 type FoundAttempt = Awaited<ReturnType<typeof findAttempt>>;
 
@@ -26,7 +25,7 @@ export class CompleteAttemptService extends BaseService {
     completion: TaskRunExecutionResult,
     execution: TaskRunExecution,
     env?: AuthenticatedEnvironment
-  ) {
+  ): Promise<"COMPLETED" | "RETRIED"> {
     const taskRunAttempt = await findAttempt(this._prisma, completion.id);
 
     if (!taskRunAttempt) {
@@ -42,7 +41,7 @@ export class CompleteAttemptService extends BaseService {
         },
       });
 
-      return "FAILED";
+      return "COMPLETED";
     }
 
     if (completion.ok) {
@@ -56,7 +55,7 @@ export class CompleteAttemptService extends BaseService {
     completion: TaskRunSuccessfulExecutionResult,
     taskRunAttempt: NonNullable<FoundAttempt>,
     env?: AuthenticatedEnvironment
-  ) {
+  ): Promise<"COMPLETED" | "RETRIED"> {
     await this._prisma.taskRunAttempt.update({
       where: { friendlyId: completion.id },
       data: {
@@ -91,7 +90,7 @@ export class CompleteAttemptService extends BaseService {
       await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);
     }
 
-    return "ACKNOWLEDGED";
+    return "COMPLETED";
   }
 
   async #completeAttemptFailed(
@@ -107,13 +106,15 @@ export class CompleteAttemptService extends BaseService {
       // We need to cancel the task run instead of fail it
       const cancelService = new CancelAttemptService();
 
-      return await cancelService.call(
+      await cancelService.call(
         taskRunAttempt.friendlyId,
         taskRunAttempt.taskRunId,
         new Date(),
         "Cancelled by user",
         env
       );
+
+      return "COMPLETED";
     }
 
     await this._prisma.taskRunAttempt.update({
@@ -125,48 +126,31 @@ export class CompleteAttemptService extends BaseService {
       },
     });
 
-    if (completion.retry !== undefined) {
-      const retryConfig = taskRunAttempt.backgroundWorkerTask.retryConfig
-        ? {
-            ...defaultRetryOptions,
-            ...RetryOptions.parse(taskRunAttempt.backgroundWorkerTask.retryConfig),
-          }
-        : undefined;
-
+    if (completion.retry !== undefined && taskRunAttempt.number < MAX_TASK_RUN_ATTEMPTS) {
       const environment = env ?? (await this.#getEnvironment(execution.environment.id));
 
       const retryAt = new Date(completion.retry.timestamp);
 
       // Retry the task run
-      await eventRepository.recordEvent(
-        retryConfig?.maxAttempts
-          ? `Retry ${execution.attempt.number}/${retryConfig?.maxAttempts - 1} delay`
-          : `Retry #${execution.attempt.number} delay`,
-        {
-          taskSlug: taskRunAttempt.taskRun.taskIdentifier,
-          environment,
-          attributes: {
-            metadata: this.#generateMetadataAttributesForNextAttempt(execution),
-            properties: {
-              retryAt: retryAt.toISOString(),
-              factor: retryConfig?.factor,
-              maxAttempts: retryConfig?.maxAttempts,
-              minTimeoutInMs: retryConfig?.minTimeoutInMs,
-              maxTimeoutInMs: retryConfig?.maxTimeoutInMs,
-              randomize: retryConfig?.randomize,
-            },
-            runId: taskRunAttempt.taskRunId,
-            style: {
-              icon: "schedule-attempt",
-            },
-            queueId: taskRunAttempt.queueId,
-            queueName: taskRunAttempt.taskRun.queue,
+      await eventRepository.recordEvent(`Retry #${execution.attempt.number} delay`, {
+        taskSlug: taskRunAttempt.taskRun.taskIdentifier,
+        environment,
+        attributes: {
+          metadata: this.#generateMetadataAttributesForNextAttempt(execution),
+          properties: {
+            retryAt: retryAt.toISOString(),
           },
-          context: taskRunAttempt.taskRun.traceContext as Record<string, string | undefined>,
-          spanIdSeed: `retry-${taskRunAttempt.number + 1}`,
-          endTime: retryAt,
-        }
-      );
+          runId: taskRunAttempt.taskRunId,
+          style: {
+            icon: "schedule-attempt",
+          },
+          queueId: taskRunAttempt.queueId,
+          queueName: taskRunAttempt.taskRun.queue,
+        },
+        context: taskRunAttempt.taskRun.traceContext as Record<string, string | undefined>,
+        spanIdSeed: `retry-${taskRunAttempt.number + 1}`,
+        endTime: retryAt,
+      });
 
       logger.debug("Retrying", { taskRun: taskRunAttempt.taskRun.friendlyId });
 
@@ -222,7 +206,7 @@ export class CompleteAttemptService extends BaseService {
         await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);
       }
 
-      return "ACKNOWLEDGED";
+      return "COMPLETED";
     }
   }
 
diff --git a/packages/cli-v3/src/commands/deploy.ts b/packages/cli-v3/src/commands/deploy.ts
index 686833efc..f60615fdc 100644
--- a/packages/cli-v3/src/commands/deploy.ts
+++ b/packages/cli-v3/src/commands/deploy.ts
@@ -144,17 +144,17 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
 
   await printStandloneInitialBanner(true);
 
-  const { config } = await readConfig(dir, {
+  const resolvedConfig = await readConfig(dir, {
     configFile: options.data.config,
     projectRef: options.data.projectRef,
   });
 
-  logger.debug("Resolved config", { config });
+  logger.debug("Resolved config", { resolvedConfig });
 
   const apiClient = new CliApiClient(authorization.config.apiUrl, authorization.config.accessToken);
 
   const deploymentEnv = await apiClient.getProjectEnv({
-    projectRef: config.project,
+    projectRef: resolvedConfig.config.project,
     env: options.data.env,
   });
 
@@ -168,11 +168,15 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
   );
 
   intro(
-    `Preparing to deploy "${deploymentEnv.data.name}" (${config.project}) to ${options.data.env}`
+    `Preparing to deploy "${deploymentEnv.data.name}" (${resolvedConfig.config.project}) to ${options.data.env}`
   );
 
   // Step 1: Build the project into a temporary directory
-  const compilation = await compileProject(config, options.data);
+  const compilation = await compileProject(
+    resolvedConfig.config,
+    options.data,
+    resolvedConfig.status === "file" ? resolvedConfig.path : undefined
+  );
 
   logger.debug("Compilation result", { compilation });
 
@@ -180,7 +184,9 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
 
   environmentVariablesSpinner.start("Checking environment variables");
 
-  const environmentVariables = await environmentClient.getEnvironmentVariables(config.project);
+  const environmentVariables = await environmentClient.getEnvironmentVariables(
+    resolvedConfig.config.project
+  );
 
   if (!environmentVariables.success) {
     environmentVariablesSpinner.stop(`Failed to fetch environment variables, skipping check`);
@@ -197,7 +203,7 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
         )}. Aborting deployment. ${chalk.bgBlueBright(
           terminalLink(
             "Manage env vars",
-            `${authorization.config.apiUrl}/projects/v3/${config.project}/environment-variables`
+            `${authorization.config.apiUrl}/projects/v3/${resolvedConfig.config.project}/environment-variables`
           )
         )}`
       );
@@ -239,11 +245,11 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
       return buildAndPushSelfHostedImage({
         imageTag: deploymentResponse.data.imageTag,
         cwd: compilation.path,
-        projectId: config.project,
+        projectId: resolvedConfig.config.project,
         deploymentId: deploymentResponse.data.id,
         deploymentVersion: version,
         contentHash: deploymentResponse.data.contentHash,
-        projectRef: config.project,
+        projectRef: resolvedConfig.config.project,
         buildPlatform: options.data.buildPlatform,
       });
     }
@@ -263,11 +269,11 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
       buildToken: deploymentResponse.data.externalBuildData.buildToken,
       buildProjectId: deploymentResponse.data.externalBuildData.projectId,
       cwd: compilation.path,
-      projectId: config.project,
+      projectId: resolvedConfig.config.project,
       deploymentId: deploymentResponse.data.id,
       deploymentVersion: deploymentResponse.data.version,
       contentHash: deploymentResponse.data.contentHash,
-      projectRef: config.project,
+      projectRef: resolvedConfig.config.project,
       loadImage: options.data.loadImage,
       buildPlatform: options.data.buildPlatform,
     });
@@ -321,7 +327,7 @@ export async function deployCommand(dir: string, anyOptions: unknown) {
 
   const deploymentLink = terminalLink(
     "View deployment",
-    `${authorization.config.apiUrl}/projects/v3/${config.project}/deployments/${finishedDeployment.id}`
+    `${authorization.config.apiUrl}/projects/v3/${resolvedConfig.config.project}/deployments/${finishedDeployment.id}`
   );
 
   switch (finishedDeployment.status) {
@@ -584,7 +590,11 @@ function extractImageDigest(outputs: string[]) {
   }
 }
 
-async function compileProject(config: ResolvedConfig, options: DeployCommandOptions) {
+async function compileProject(
+  config: ResolvedConfig,
+  options: DeployCommandOptions,
+  configPath?: string
+) {
   if (!options.skipTypecheck) {
     await typecheckProject(config, options);
   }
@@ -605,10 +615,24 @@ async function compileProject(config: ResolvedConfig, options: DeployCommandOpti
     importResolve("./workers/prod/worker-setup.js", import.meta.url)
   ).href.replace("file://", "");
 
-  const workerContents = workerFacade
+  let workerContents = workerFacade
     .replace("__TASKS__", createTaskFileImports(taskFiles))
     .replace("__WORKER_SETUP__", `import { tracingSDK } from "${workerSetupPath}";`);
 
+  if (configPath) {
+    logger.debug("Importing project config from", { configPath });
+
+    workerContents = workerContents.replace(
+      "__IMPORTED_PROJECT_CONFIG__",
+      `import importedConfig from "${configPath}";`
+    );
+  } else {
+    workerContents = workerContents.replace(
+      "__IMPORTED_PROJECT_CONFIG__",
+      `const importedConfig = undefined;`
+    );
+  }
+
   const result = await build({
     stdin: {
       contents: workerContents,
diff --git a/packages/cli-v3/src/commands/dev.tsx b/packages/cli-v3/src/commands/dev.tsx
index e628ed70f..b14b5d569 100644
--- a/packages/cli-v3/src/commands/dev.tsx
+++ b/packages/cli-v3/src/commands/dev.tsx
@@ -138,7 +138,7 @@ async function startDev(
         if (config.status === "file") {
           logger.log(`${basename(config.path)} changed...`);
           logger.debug("New config", { config: config.config });
-          rerender(await getDevReactElement(config.config, authorization));
+          rerender(await getDevReactElement(config.config, authorization, config.path));
         } else {
           logger.debug("New config", { config: config.config });
           rerender(await getDevReactElement(config.config, authorization));
@@ -148,7 +148,8 @@ async function startDev(
 
     async function getDevReactElement(
       configParam: ResolvedConfig,
-      authorization: { apiUrl: string; accessToken: string }
+      authorization: { apiUrl: string; accessToken: string },
+      configPath?: string
     ) {
       const accessToken = authorization.accessToken;
       const apiUrl = authorization.apiUrl;
@@ -175,11 +176,18 @@ async function startDev(
           projectName={devEnv.data.name}
           debuggerOn={options.debugger}
           debugOtel={options.debugOtel}
+          configPath={configPath}
         />
       );
     }
 
-    const devReactElement = render(await getDevReactElement(config.config, authorization));
+    const devReactElement = render(
+      await getDevReactElement(
+        config.config,
+        authorization,
+        config.status === "file" ? config.path : undefined
+      )
+    );
 
     rerender = devReactElement.rerender;
 
@@ -205,6 +213,7 @@ type DevProps = {
   projectName: string;
   debuggerOn: boolean;
   debugOtel: boolean;
+  configPath?: string;
 };
 
 function useDev({
@@ -215,6 +224,7 @@ function useDev({
   projectName,
   debuggerOn,
   debugOtel,
+  configPath,
 }: DevProps) {
   useEffect(() => {
     const websocketUrl = new URL(apiUrl);
@@ -334,10 +344,24 @@ function useDev({
         importResolve("./workers/dev/worker-setup.js", import.meta.url)
       ).href.replace("file://", "");
 
-      const entryPointContents = workerFacade
+      let entryPointContents = workerFacade
         .replace("__TASKS__", createTaskFileImports(taskFiles))
         .replace("__WORKER_SETUP__", `import { tracingSDK, sender } from "${workerSetupPath}";`);
 
+      if (configPath) {
+        logger.debug("Importing project config from", { configPath });
+
+        entryPointContents = entryPointContents.replace(
+          "__IMPORTED_PROJECT_CONFIG__",
+          `import importedConfig from "${configPath}";`
+        );
+      } else {
+        entryPointContents = entryPointContents.replace(
+          "__IMPORTED_PROJECT_CONFIG__",
+          `const importedConfig = undefined;`
+        );
+      }
+
       let firstBuild = true;
 
       logger.log(chalk.dim("⎔ Building background worker..."));
diff --git a/packages/cli-v3/src/types.ts b/packages/cli-v3/src/types.ts
index cb8a0dd3e..eaa50ebc7 100644
--- a/packages/cli-v3/src/types.ts
+++ b/packages/cli-v3/src/types.ts
@@ -1,11 +1,24 @@
-import { TaskMetadataWithFilePath } from "@trigger.dev/core/v3";
+import {
+  HandleErrorFnParams,
+  HandleErrorResult,
+  InitFnParams,
+  InitOutput,
+  MiddlewareFnParams,
+  RunFnParams,
+  TaskMetadataWithFilePath,
+} from "@trigger.dev/core/v3";
 
 export type TaskMetadataWithFunctions = TaskMetadataWithFilePath & {
   fns: {
-    run: (payload: any, params: any) => Promise<any>;
-    init?: (payload: any, params: any) => Promise<void>;
-    cleanup?: (payload: any, params: any) => Promise<void>;
-    middleware?: (payload: any, params: any) => Promise<void>;
+    run: (payload: any, params: RunFnParams<any>) => Promise<any>;
+    init?: (payload: any, params: InitFnParams) => Promise<InitOutput>;
+    cleanup?: (payload: any, params: RunFnParams<any>) => Promise<void>;
+    middleware?: (payload: any, params: MiddlewareFnParams) => Promise<void>;
+    handleError?: (
+      payload: any,
+      error: unknown,
+      params: HandleErrorFnParams<any>
+    ) => HandleErrorResult;
   };
 };
 
diff --git a/packages/cli-v3/src/workers/dev/backgroundWorker.ts b/packages/cli-v3/src/workers/dev/backgroundWorker.ts
index 0778aeb38..bb425356f 100644
--- a/packages/cli-v3/src/workers/dev/backgroundWorker.ts
+++ b/packages/cli-v3/src/workers/dev/backgroundWorker.ts
@@ -366,9 +366,9 @@ export class BackgroundWorker {
       const taskRunProcess = new TaskRunProcess(
         this.path,
         {
-          ...this.#readEnvVars(),
           ...this.params.env,
           ...(payload.environment ?? {}),
+          ...this.#readEnvVars(),
         },
         this.metadata,
         this.params
diff --git a/packages/cli-v3/src/workers/dev/worker-facade.ts b/packages/cli-v3/src/workers/dev/worker-facade.ts
index 536264425..da7943bac 100644
--- a/packages/cli-v3/src/workers/dev/worker-facade.ts
+++ b/packages/cli-v3/src/workers/dev/worker-facade.ts
@@ -1,42 +1,36 @@
-import { Config, type TracingSDK } from "@trigger.dev/core/v3";
+import { Config, ProjectConfig, TaskExecutor, type TracingSDK } from "@trigger.dev/core/v3";
 import "source-map-support/register.js";
 
 __WORKER_SETUP__;
 declare const __WORKER_SETUP__: unknown;
 
+__IMPORTED_PROJECT_CONFIG__;
+declare const __IMPORTED_PROJECT_CONFIG__: unknown;
+declare const importedConfig: ProjectConfig | undefined;
+
 declare const __PROJECT_CONFIG__: Config;
 declare const tracingSDK: TracingSDK;
 
 const otelTracer = tracingSDK.getTracer("trigger-dev-worker", packageJson.version);
 const otelLogger = tracingSDK.getLogger("trigger-dev-worker", packageJson.version);
 
-import { SpanKind } from "@opentelemetry/api";
 import {
   ConsoleInterceptor,
   DevRuntimeManager,
   OtelTaskLogger,
-  SemanticInternalAttributes,
   TaskMetadataWithFilePath,
-  TaskRunContext,
   TaskRunErrorCodes,
   TaskRunExecution,
-  TaskRunExecutionRetry,
   TriggerTracer,
   ZodMessageHandler,
   ZodMessageSender,
-  accessoryAttributes,
-  calculateNextRetryDelay,
   childToWorkerMessages,
   logger,
-  parseError,
   runtime,
-  taskContextManager,
   workerToChildMessages,
-  type BackgroundWorkerProperties,
 } from "@trigger.dev/core/v3";
 import * as packageJson from "../../../package.json";
 
-import { flattenAttributes } from "@trigger.dev/core/v3";
 import { TaskMetadataWithFunctions } from "../../types.js";
 
 declare const sender: ZodMessageSender<typeof childToWorkerMessages>;
@@ -64,137 +58,6 @@ const TaskFiles: Record<string, string> = {};
 __TASKS__;
 declare const __TASKS__: Record<string, string>;
 
-class TaskExecutor {
-  constructor(public task: TaskMetadataWithFunctions) {}
-
-  async determineRetrying(
-    execution: TaskRunExecution,
-    error: unknown
-  ): Promise<TaskRunExecutionRetry | "skipped" | undefined> {
-    const retry = this.task.retry ?? __PROJECT_CONFIG__.retries?.default;
-
-    if (!retry) {
-      return;
-    }
-
-    const delay = calculateNextRetryDelay(retry, execution.attempt.number);
-
-    if (
-      typeof __PROJECT_CONFIG__.retries?.enabledInDev === "boolean" &&
-      !__PROJECT_CONFIG__.retries.enabledInDev
-    ) {
-      // TODO: trigger a warning saying that retries are disabled in dev
-      return "skipped";
-    }
-
-    return typeof delay === "undefined" ? undefined : { timestamp: Date.now() + delay, delay };
-  }
-
-  async execute(
-    execution: TaskRunExecution,
-    worker: BackgroundWorkerProperties,
-    traceContext: Record<string, unknown>
-  ) {
-    const parsedPayload = JSON.parse(execution.run.payload);
-    const ctx = TaskRunContext.parse(execution);
-    const attemptMessage = `Attempt ${execution.attempt.number}`;
-
-    const output = await taskContextManager.runWith(
-      {
-        ctx,
-        payload: parsedPayload,
-        worker,
-      },
-      async () => {
-        tracingSDK.asyncResourceDetector.resolveWithAttributes({
-          ...taskContextManager.attributes,
-          [SemanticInternalAttributes.SDK_VERSION]: this.task.packageVersion,
-          [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
-        });
-
-        return await tracer.startActiveSpan(
-          attemptMessage,
-          async (span) => {
-            return await consoleInterceptor.intercept(console, async () => {
-              const init = await this.#callTaskInit(parsedPayload, ctx);
-
-              try {
-                const output = await this.#callRun(parsedPayload, ctx, init);
-
-                span.setAttributes(flattenAttributes(output, SemanticInternalAttributes.OUTPUT));
-
-                return output;
-              } finally {
-                await this.#callTaskCleanup(parsedPayload, ctx, init);
-              }
-            });
-          },
-          {
-            kind: SpanKind.CONSUMER,
-            attributes: {
-              [SemanticInternalAttributes.STYLE_ICON]: "attempt",
-              ...flattenAttributes(parsedPayload, SemanticInternalAttributes.PAYLOAD),
-              ...accessoryAttributes({
-                items: [
-                  {
-                    text: ctx.task.filePath,
-                  },
-                  {
-                    text: `${ctx.task.exportName}.run()`,
-                  },
-                ],
-                style: "codepath",
-              }),
-            },
-          },
-          tracer.extractContext(traceContext)
-        );
-      }
-    );
-
-    return { output: JSON.stringify(output), outputType: "application/json" };
-  }
-
-  async #callRun(payload: unknown, ctx: TaskRunContext, init: unknown) {
-    const runFn = this.task.fns.run;
-    const middlewareFn = this.task.fns.middleware;
-
-    if (!runFn) {
-      throw new Error("Task does not have a run function");
-    }
-
-    if (!middlewareFn) {
-      return runFn(payload, { ctx });
-    }
-
-    return middlewareFn(payload, { ctx, next: async () => runFn(payload, { ctx, init }) });
-  }
-
-  async #callTaskInit(payload: unknown, ctx: TaskRunContext) {
-    const initFn = this.task.fns.init;
-
-    if (!initFn) {
-      return {};
-    }
-
-    return tracer.startActiveSpan("init", async (span) => {
-      return await initFn(payload, { ctx });
-    });
-  }
-
-  async #callTaskCleanup(payload: unknown, ctx: TaskRunContext, init: unknown) {
-    const cleanupFn = this.task.fns.cleanup;
-
-    if (!cleanupFn) {
-      return;
-    }
-
-    return tracer.startActiveSpan("cleanup", async (span) => {
-      return await cleanupFn(payload, { ctx, init });
-    });
-  }
-}
-
 function getTasks(): Array<TaskMetadataWithFunctions> {
   const result: Array<TaskMetadataWithFunctions> = [];
 
@@ -237,7 +100,16 @@ runtime.registerTasks(tasks);
 const taskExecutors: Map<string, TaskExecutor> = new Map();
 
 for (const task of tasks) {
-  taskExecutors.set(task.id, new TaskExecutor(task));
+  taskExecutors.set(
+    task.id,
+    new TaskExecutor(task, {
+      tracer,
+      tracingSDK,
+      consoleInterceptor,
+      projectConfig: __PROJECT_CONFIG__,
+      importedConfig,
+    })
+  );
 }
 
 let _execution: TaskRunExecution | undefined;
@@ -295,24 +167,7 @@ const handler = new ZodMessageHandler({
 
         return sender.send("TASK_RUN_COMPLETED", {
           execution,
-          result: {
-            id: execution.attempt.id,
-            ok: true,
-            ...result,
-          },
-        });
-      } catch (e) {
-        const retryResult = await executor.determineRetrying(execution, e);
-
-        return sender.send("TASK_RUN_COMPLETED", {
-          execution,
-          result: {
-            id: execution.attempt.id,
-            ok: false,
-            error: parseError(e),
-            retry: typeof retryResult === "object" ? retryResult : undefined,
-            skippedRetrying: retryResult === "skipped",
-          },
+          result,
         });
       } finally {
         _execution = undefined;
diff --git a/packages/cli-v3/src/workers/prod/worker-facade.ts b/packages/cli-v3/src/workers/prod/worker-facade.ts
index f9d7f8022..0e4ed89f6 100644
--- a/packages/cli-v3/src/workers/prod/worker-facade.ts
+++ b/packages/cli-v3/src/workers/prod/worker-facade.ts
@@ -1,43 +1,40 @@
 import {
+  Config,
   ProdChildToWorkerMessages,
   ProdWorkerToChildMessages,
+  ProjectConfig,
+  TaskExecutor,
   ZodIpcConnection,
   type TracingSDK,
-  Config,
 } from "@trigger.dev/core/v3";
 import "source-map-support/register.js";
 
 __WORKER_SETUP__;
 declare const __WORKER_SETUP__: unknown;
+
+__IMPORTED_PROJECT_CONFIG__;
+declare const __IMPORTED_PROJECT_CONFIG__: unknown;
+declare const importedConfig: ProjectConfig | undefined;
+
 declare const __PROJECT_CONFIG__: Config;
 declare const tracingSDK: TracingSDK;
 
 const otelTracer = tracingSDK.getTracer("trigger-prod-worker", packageJson.version);
 const otelLogger = tracingSDK.getLogger("trigger-prod-worker", packageJson.version);
 
-import { SpanKind } from "@opentelemetry/api";
 import {
   ConsoleInterceptor,
   OtelTaskLogger,
   ProdRuntimeManager,
-  SemanticInternalAttributes,
   TaskMetadataWithFilePath,
-  TaskRunContext,
   TaskRunErrorCodes,
   TaskRunExecution,
-  TaskRunExecutionRetry,
   TriggerTracer,
-  accessoryAttributes,
-  calculateNextRetryDelay,
   logger,
-  parseError,
   runtime,
-  taskContextManager,
-  type BackgroundWorkerProperties,
 } from "@trigger.dev/core/v3";
 import * as packageJson from "../../../package.json";
 
-import { flattenAttributes } from "@trigger.dev/core/v3";
 import { TaskMetadataWithFunctions } from "../../types";
 
 const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
@@ -59,129 +56,6 @@ const TaskFiles: Record<string, string> = {};
 __TASKS__;
 declare const __TASKS__: Record<string, string>;
 
-class TaskExecutor {
-  constructor(public task: TaskMetadataWithFunctions) {}
-
-  async determineRetrying(
-    execution: TaskRunExecution,
-    error: unknown
-  ): Promise<TaskRunExecutionRetry | undefined> {
-    const retry = this.task.retry ?? __PROJECT_CONFIG__.retries?.default;
-
-    if (!retry) {
-      return;
-    }
-
-    const delay = calculateNextRetryDelay(retry, execution.attempt.number);
-
-    return typeof delay === "undefined" ? undefined : { timestamp: Date.now() + delay, delay };
-  }
-
-  async execute(
-    execution: TaskRunExecution,
-    worker: BackgroundWorkerProperties,
-    traceContext: Record<string, unknown>
-  ) {
-    const parsedPayload = JSON.parse(execution.run.payload);
-    const ctx = TaskRunContext.parse(execution);
-    const attemptMessage = `Attempt ${execution.attempt.number}`;
-
-    const output = await taskContextManager.runWith(
-      {
-        ctx,
-        payload: parsedPayload,
-        worker,
-      },
-      async () => {
-        tracingSDK.asyncResourceDetector.resolveWithAttributes({
-          ...taskContextManager.attributes,
-          [SemanticInternalAttributes.SDK_VERSION]: this.task.packageVersion,
-          [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
-        });
-
-        return await tracer.startActiveSpan(
-          attemptMessage,
-          async (span) => {
-            return await consoleInterceptor.intercept(console, async () => {
-              const init = await this.#callTaskInit(parsedPayload, ctx);
-
-              try {
-                const output = await this.#callRun(parsedPayload, ctx, init);
-
-                span.setAttributes(flattenAttributes(output, SemanticInternalAttributes.OUTPUT));
-
-                return output;
-              } finally {
-                await this.#callTaskCleanup(parsedPayload, ctx, init);
-              }
-            });
-          },
-          {
-            kind: SpanKind.CONSUMER,
-            attributes: {
-              [SemanticInternalAttributes.STYLE_ICON]: "attempt",
-              ...flattenAttributes(parsedPayload, SemanticInternalAttributes.PAYLOAD),
-              ...accessoryAttributes({
-                items: [
-                  {
-                    text: ctx.task.filePath,
-                  },
-                  {
-                    text: `${ctx.task.exportName}.run()`,
-                  },
-                ],
-                style: "codepath",
-              }),
-            },
-          },
-          tracer.extractContext(traceContext)
-        );
-      }
-    );
-
-    return { output: JSON.stringify(output), outputType: "application/json" };
-  }
-
-  async #callRun(payload: unknown, ctx: TaskRunContext, init: unknown) {
-    const runFn = this.task.fns.run;
-    const middlewareFn = this.task.fns.middleware;
-
-    if (!runFn) {
-      throw new Error("Task does not have a run function");
-    }
-
-    if (!middlewareFn) {
-      return runFn(payload, { ctx });
-    }
-
-    return middlewareFn(payload, { ctx, next: async () => runFn(payload, { ctx, init }) });
-  }
-
-  async #callTaskInit(payload: unknown, ctx: TaskRunContext) {
-    const initFn = this.task.fns.init;
-
-    if (!initFn) {
-      return {};
-    }
-
-    return tracer.startActiveSpan("init", async (span) => {
-      return await initFn(payload, { ctx });
-    });
-  }
-
-  async #callTaskCleanup(payload: unknown, ctx: TaskRunContext, init: unknown) {
-    const cleanupFn = this.task.fns.cleanup;
-
-    if (!cleanupFn) {
-      return;
-    }
-
-    return tracer.startActiveSpan("cleanup", async (span) => {
-      return await cleanupFn(payload, { ctx, init });
-    });
-  }
-}
-
 function getTasks(): Array<TaskMetadataWithFunctions> {
   const result: Array<TaskMetadataWithFunctions> = [];
 
@@ -224,7 +98,16 @@ runtime.registerTasks(tasks);
 const taskExecutors: Map<string, TaskExecutor> = new Map();
 
 for (const task of tasks) {
-  taskExecutors.set(task.id, new TaskExecutor(task));
+  taskExecutors.set(
+    task.id,
+    new TaskExecutor(task, {
+      tracer,
+      tracingSDK,
+      consoleInterceptor,
+      projectConfig: __PROJECT_CONFIG__,
+      importedConfig,
+    })
+  );
 }
 
 let _execution: TaskRunExecution | undefined;
@@ -283,21 +166,7 @@ const zodIpc = new ZodIpcConnection({
 
         return sender.send("TASK_RUN_COMPLETED", {
           execution,
-          result: {
-            id: execution.attempt.id,
-            ok: true,
-            ...result,
-          },
-        });
-      } catch (e) {
-        return sender.send("TASK_RUN_COMPLETED", {
-          execution,
-          result: {
-            id: execution.attempt.id,
-            ok: false,
-            error: parseError(e),
-            retry: await executor.determineRetrying(execution, e),
-          },
+          result,
         });
       } finally {
         _execution = undefined;
diff --git a/packages/core/src/v3/consoleInterceptor.ts b/packages/core/src/v3/consoleInterceptor.ts
index c986b8335..0ee04a5a6 100644
--- a/packages/core/src/v3/consoleInterceptor.ts
+++ b/packages/core/src/v3/consoleInterceptor.ts
@@ -10,7 +10,7 @@ export class ConsoleInterceptor {
 
   // Intercept the console and send logs to the OpenTelemetry logger
   // during the execution of the callback
-  async intercept<T, R extends Promise<T>>(console: Console, callback: () => R): Promise<T> {
+  async intercept<T>(console: Console, callback: () => Promise<T>): Promise<T> {
     // Save the original console methods
     const originalConsole = {
       log: console.log,
diff --git a/packages/core/src/v3/index.ts b/packages/core/src/v3/index.ts
index c54f01cf1..14cb25e81 100644
--- a/packages/core/src/v3/index.ts
+++ b/packages/core/src/v3/index.ts
@@ -48,4 +48,5 @@ export { defaultRetryOptions, calculateNextRetryDelay, calculateResetAt } from "
 export { accessoryAttributes } from "./utils/styleAttributes";
 export { eventFilterMatches } from "../eventFilterMatches";
 export { omit } from "./utils/omit";
-export { TracingSDK, type TracingDiagnosticLogLevel } from "./otel";
+export { TracingSDK, type TracingDiagnosticLogLevel, recordSpanException } from "./otel";
+export { TaskExecutor, type TaskExecutorOptions } from "./workers/taskExecutor";
diff --git a/packages/core/src/v3/logger/taskLogger.ts b/packages/core/src/v3/logger/taskLogger.ts
index a37d56583..471b84230 100644
--- a/packages/core/src/v3/logger/taskLogger.ts
+++ b/packages/core/src/v3/logger/taskLogger.ts
@@ -67,7 +67,7 @@ export class OtelTaskLogger implements TaskLogger {
     severityNumber: SeverityNumber,
     properties?: Record<string, unknown>
   ) {
-    let attributes: Attributes = { ...flattenAttributes(properties), "log.type": "logger" };
+    let attributes: Attributes = { ...flattenAttributes(properties) };
 
     const icon = iconStringForSeverity(severityNumber);
     if (icon !== undefined) {
diff --git a/packages/core/src/v3/otel/index.ts b/packages/core/src/v3/otel/index.ts
index 9959735e1..b796a67a5 100644
--- a/packages/core/src/v3/otel/index.ts
+++ b/packages/core/src/v3/otel/index.ts
@@ -1,2 +1,16 @@
+import { Span, SpanStatusCode } from "@opentelemetry/api";
+
 export { TracingSDK, type TracingSDKConfig, type TracingDiagnosticLogLevel } from "./tracingSDK";
 export { HttpInstrumentation, FetchInstrumentation } from "./instrumentations";
+
+export function recordSpanException(span: Span, error: unknown) {
+  if (error instanceof Error) {
+    span.recordException(error);
+  } else if (typeof error === "string") {
+    span.recordException(new Error(error));
+  } else {
+    span.recordException(new Error(JSON.stringify(error)));
+  }
+
+  span.setStatus({ code: SpanStatusCode.ERROR });
+}
diff --git a/packages/core/src/v3/schemas/common.ts b/packages/core/src/v3/schemas/common.ts
index 9a0427296..2892124a2 100644
--- a/packages/core/src/v3/schemas/common.ts
+++ b/packages/core/src/v3/schemas/common.ts
@@ -31,6 +31,8 @@ export const TaskRunErrorCodes = {
   TASK_EXECUTION_ABORTED: "TASK_EXECUTION_ABORTED",
   TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE",
   TASK_RUN_CANCELLED: "TASK_RUN_CANCELLED",
+  TASK_OUTPUT_ERROR: "TASK_OUTPUT_ERROR",
+  HANDLE_ERROR_ERROR: "HANDLE_ERROR_ERROR",
 } as const;
 
 export const TaskRunInternalError = z.object({
@@ -43,7 +45,10 @@ export const TaskRunInternalError = z.object({
     "TASK_EXECUTION_ABORTED",
     "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE",
     "TASK_RUN_CANCELLED",
+    "TASK_OUTPUT_ERROR",
+    "HANDLE_ERROR_ERROR",
   ]),
+  message: z.string().optional(),
 });
 
 export type TaskRunInternalError = z.infer<typeof TaskRunInternalError>;
@@ -156,6 +161,7 @@ export type TaskRunContext = z.infer<typeof TaskRunContext>;
 export const TaskRunExecutionRetry = z.object({
   timestamp: z.number(),
   delay: z.number(),
+  error: z.unknown().optional(),
 });
 
 export type TaskRunExecutionRetry = z.infer<typeof TaskRunExecutionRetry>;
diff --git a/packages/core/src/v3/types/config.ts b/packages/core/src/v3/types/config.ts
new file mode 100644
index 000000000..f73829842
--- /dev/null
+++ b/packages/core/src/v3/types/config.ts
@@ -0,0 +1,17 @@
+import { HandleErrorFnParams, HandleErrorResult } from ".";
+import { RetryOptions } from "../schemas";
+
+export interface ProjectConfig {
+  project: string;
+  triggerDirectories?: string | string[];
+  triggerUrl?: string;
+  retries?: {
+    enabledInDev?: boolean;
+    default?: RetryOptions;
+  };
+  handleError?: (
+    payload: any,
+    error: unknown,
+    params: HandleErrorFnParams<any>
+  ) => HandleErrorResult;
+}
diff --git a/packages/core/src/v3/types/index.ts b/packages/core/src/v3/types/index.ts
index 178cd64f8..35e5a8be6 100644
--- a/packages/core/src/v3/types/index.ts
+++ b/packages/core/src/v3/types/index.ts
@@ -1 +1,63 @@
+import { RetryOptions, TaskMetadataWithFilePath, TaskRunContext } from "../schemas";
+import { Prettify } from "./utils";
+
 export * from "./utils";
+export * from "./config";
+
+export type InitOutput = Record<string, any> | void | undefined;
+
+export type RunFnParams<TInitOutput extends InitOutput> = Prettify<{
+  ctx: Context;
+  init?: TInitOutput;
+}>;
+
+export type MiddlewareFnParams = Prettify<{
+  ctx: Context;
+  next: () => Promise<void>;
+}>;
+
+export type InitFnParams = Prettify<{
+  ctx: Context;
+}>;
+
+export type Context = TaskRunContext;
+
+export type SuccessFnParams<TOutput, TInitOutput extends InitOutput> = RunFnParams<TInitOutput> &
+  Prettify<{
+    output: TOutput;
+  }>;
+
+export type HandleErrorFnParams<TInitOutput extends InitOutput> = RunFnParams<TInitOutput> &
+  Prettify<{
+    retry?: RetryOptions;
+    retryAt?: Date;
+    retryDelayInMs?: number;
+  }>;
+
+export type HandleErrorModificationOptions = {
+  skipRetrying?: boolean | undefined;
+  retryAt?: Date | undefined;
+  retryDelayInMs?: number | undefined;
+  retry?: RetryOptions | undefined;
+  error?: unknown;
+};
+
+export type HandleErrorResult =
+  | undefined
+  | void
+  | HandleErrorModificationOptions
+  | Promise<undefined | void | HandleErrorModificationOptions>;
+
+export type TaskMetadataWithFunctions = TaskMetadataWithFilePath & {
+  fns: {
+    run: (payload: any, params: RunFnParams<any>) => Promise<any>;
+    init?: (payload: any, params: InitFnParams) => Promise<InitOutput>;
+    cleanup?: (payload: any, params: RunFnParams<any>) => Promise<void>;
+    middleware?: (payload: any, params: MiddlewareFnParams) => Promise<void>;
+    handleError?: (
+      payload: any,
+      error: unknown,
+      params: HandleErrorFnParams<any>
+    ) => HandleErrorResult;
+  };
+};
diff --git a/packages/core/src/v3/types/utils.ts b/packages/core/src/v3/types/utils.ts
index 087163e6b..4a6b90fec 100644
--- a/packages/core/src/v3/types/utils.ts
+++ b/packages/core/src/v3/types/utils.ts
@@ -2,3 +2,7 @@ export type RequireKeys<T extends object, K extends keyof T> = Required<Pick<T,
   Omit<T, K> extends infer O
   ? { [P in keyof O]: O[P] }
   : never;
+
+export type Prettify<T> = {
+  [K in keyof T]: T[K];
+} & {};
diff --git a/packages/core/src/v3/workers/taskExecutor.ts b/packages/core/src/v3/workers/taskExecutor.ts
new file mode 100644
index 000000000..aae996b7e
--- /dev/null
+++ b/packages/core/src/v3/workers/taskExecutor.ts
@@ -0,0 +1,315 @@
+import { SpanKind } from "@opentelemetry/api";
+import { TracingSDK, recordSpanException } from "../otel";
+import {
+  BackgroundWorkerProperties,
+  Config,
+  TaskRunContext,
+  TaskRunErrorCodes,
+  TaskRunExecution,
+  TaskRunExecutionResult,
+  TaskRunExecutionRetry,
+} from "../schemas";
+import { SemanticInternalAttributes } from "../semanticInternalAttributes";
+import { ProjectConfig, TaskMetadataWithFunctions } from "../types";
+import { flattenAttributes } from "../utils/flattenAttributes";
+import { accessoryAttributes } from "../utils/styleAttributes";
+import { calculateNextRetryDelay } from "../utils/retries";
+import { taskContextManager } from "../tasks/taskContextManager";
+import { TriggerTracer } from "../tracer";
+import { ConsoleInterceptor } from "../consoleInterceptor";
+import { parseError } from "../errors";
+
+export type TaskExecutorOptions = {
+  tracingSDK: TracingSDK;
+  tracer: TriggerTracer;
+  consoleInterceptor: ConsoleInterceptor;
+  projectConfig: Config;
+  importedConfig: ProjectConfig | undefined;
+};
+
+export class TaskExecutor {
+  private _tracingSDK: TracingSDK;
+  private _tracer: TriggerTracer;
+  private _consoleInterceptor: ConsoleInterceptor;
+  private _config: Config;
+  private _importedConfig: ProjectConfig | undefined;
+
+  constructor(
+    public task: TaskMetadataWithFunctions,
+    options: TaskExecutorOptions
+  ) {
+    this._tracingSDK = options.tracingSDK;
+    this._tracer = options.tracer;
+    this._consoleInterceptor = options.consoleInterceptor;
+    this._config = options.projectConfig;
+    this._importedConfig = options.importedConfig;
+  }
+
+  async execute(
+    execution: TaskRunExecution,
+    worker: BackgroundWorkerProperties,
+    traceContext: Record<string, unknown>
+  ): Promise<TaskRunExecutionResult> {
+    const parsedPayload = JSON.parse(execution.run.payload);
+    const ctx = TaskRunContext.parse(execution);
+    const attemptMessage = `Attempt ${execution.attempt.number}`;
+
+    const result = await taskContextManager.runWith(
+      {
+        ctx,
+        payload: parsedPayload,
+        worker,
+      },
+      async () => {
+        this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
+          ...taskContextManager.attributes,
+          [SemanticInternalAttributes.SDK_VERSION]: this.task.packageVersion,
+          [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript",
+        });
+
+        return await this._tracer.startActiveSpan(
+          attemptMessage,
+          async (span) => {
+            return await this._consoleInterceptor.intercept(console, async () => {
+              const init = await this.#callTaskInit(parsedPayload, ctx);
+
+              try {
+                const output = await this.#callRun(parsedPayload, ctx, init);
+
+                try {
+                  span.setAttributes(flattenAttributes(output, SemanticInternalAttributes.OUTPUT));
+
+                  const serializedOutput = JSON.stringify(output);
+
+                  return {
+                    ok: true,
+                    id: execution.attempt.id,
+                    output: serializedOutput,
+                    outputType: "application/json",
+                  } satisfies TaskRunExecutionResult;
+                } catch (stringifyError) {
+                  recordSpanException(span, stringifyError);
+
+                  return {
+                    ok: false,
+                    id: execution.attempt.id,
+                    error: {
+                      type: "INTERNAL_ERROR",
+                      code: TaskRunErrorCodes.TASK_OUTPUT_ERROR,
+                      message:
+                        stringifyError instanceof Error
+                          ? stringifyError.message
+                          : typeof stringifyError === "string"
+                          ? stringifyError
+                          : undefined,
+                    },
+                  } satisfies TaskRunExecutionResult;
+                }
+              } catch (runError) {
+                try {
+                  const handleErrorResult = await this.#handleError(
+                    execution,
+                    runError,
+                    parsedPayload,
+                    ctx
+                  );
+
+                  recordSpanException(span, handleErrorResult.error ?? runError);
+
+                  return {
+                    id: execution.attempt.id,
+                    ok: false,
+                    error: handleErrorResult.error
+                      ? parseError(handleErrorResult.error)
+                      : parseError(runError),
+                    retry:
+                      handleErrorResult.status === "retry" ? handleErrorResult.retry : undefined,
+                    skippedRetrying: handleErrorResult.status === "skipped",
+                  } satisfies TaskRunExecutionResult;
+                } catch (handleErrorError) {
+                  recordSpanException(span, handleErrorError);
+
+                  return {
+                    ok: false,
+                    id: execution.attempt.id,
+                    error: {
+                      type: "INTERNAL_ERROR",
+                      code: TaskRunErrorCodes.HANDLE_ERROR_ERROR,
+                      message:
+                        handleErrorError instanceof Error
+                          ? handleErrorError.message
+                          : typeof handleErrorError === "string"
+                          ? handleErrorError
+                          : undefined,
+                    },
+                  } satisfies TaskRunExecutionResult;
+                }
+              } finally {
+                await this.#callTaskCleanup(parsedPayload, ctx, init);
+              }
+            });
+          },
+          {
+            kind: SpanKind.CONSUMER,
+            attributes: {
+              [SemanticInternalAttributes.STYLE_ICON]: "attempt",
+              ...flattenAttributes(parsedPayload, SemanticInternalAttributes.PAYLOAD),
+              ...accessoryAttributes({
+                items: [
+                  {
+                    text: ctx.task.filePath,
+                  },
+                  {
+                    text: `${ctx.task.exportName}.run()`,
+                  },
+                ],
+                style: "codepath",
+              }),
+            },
+          },
+          this._tracer.extractContext(traceContext)
+        );
+      }
+    );
+
+    return result;
+  }
+
+  async #callRun(payload: unknown, ctx: TaskRunContext, init: unknown) {
+    const runFn = this.task.fns.run;
+    const middlewareFn = this.task.fns.middleware;
+
+    if (!runFn) {
+      throw new Error("Task does not have a run function");
+    }
+
+    if (!middlewareFn) {
+      return runFn(payload, { ctx });
+    }
+
+    return middlewareFn(payload, { ctx, next: async () => runFn(payload, { ctx, init }) });
+  }
+
+  async #callTaskInit(payload: unknown, ctx: TaskRunContext) {
+    const initFn = this.task.fns.init;
+
+    if (!initFn) {
+      return {};
+    }
+
+    return this._tracer.startActiveSpan("init", async (span) => {
+      return await initFn(payload, { ctx });
+    });
+  }
+
+  async #callTaskCleanup(payload: unknown, ctx: TaskRunContext, init: unknown) {
+    const cleanupFn = this.task.fns.cleanup;
+
+    if (!cleanupFn) {
+      return;
+    }
+
+    return this._tracer.startActiveSpan("cleanup", async (span) => {
+      return await cleanupFn(payload, { ctx, init });
+    });
+  }
+
+  async #handleError(
+    execution: TaskRunExecution,
+    error: unknown,
+    payload: any,
+    ctx: TaskRunContext
+  ): Promise<
+    | { status: "retry"; retry: TaskRunExecutionRetry; error?: unknown }
+    | { status: "skipped"; error?: unknown } // skipped is different than noop, it means that the task was skipped from retrying, instead of just not retrying
+    | { status: "noop"; error?: unknown }
+  > {
+    const retry = this.task.retry ?? this._config.retries?.default;
+
+    if (!retry) {
+      return { status: "noop" };
+    }
+
+    const delay = calculateNextRetryDelay(retry, execution.attempt.number);
+
+    if (
+      typeof this._config.retries?.enabledInDev === "boolean" &&
+      !this._config.retries.enabledInDev
+    ) {
+      return { status: "skipped" };
+    }
+
+    return this._tracer.startActiveSpan(
+      "handleError()",
+      async (span) => {
+        const handleErrorResult = this.task.fns.handleError
+          ? await this.task.fns.handleError(payload, error, {
+              ctx,
+              retry,
+              retryDelayInMs: delay,
+              retryAt: delay ? new Date(Date.now() + delay) : undefined,
+            })
+          : this._importedConfig
+          ? await this._importedConfig.handleError?.(payload, error, {
+              ctx,
+              retry,
+              retryDelayInMs: delay,
+              retryAt: delay ? new Date(Date.now() + delay) : undefined,
+            })
+          : undefined;
+
+        // If handleErrorResult
+        if (!handleErrorResult) {
+          return typeof delay === "undefined"
+            ? { status: "noop" }
+            : { status: "retry", retry: { timestamp: Date.now() + delay, delay } };
+        }
+
+        if (handleErrorResult.skipRetrying) {
+          return { status: "skipped", error: handleErrorResult.error };
+        }
+
+        if (typeof handleErrorResult.retryAt !== "undefined") {
+          return {
+            status: "retry",
+            retry: {
+              timestamp: handleErrorResult.retryAt.getTime(),
+              delay: handleErrorResult.retryAt.getTime() - Date.now(),
+            },
+            error: handleErrorResult.error,
+          };
+        }
+
+        if (typeof handleErrorResult.retryDelayInMs === "number") {
+          return {
+            status: "retry",
+            retry: {
+              timestamp: Date.now() + handleErrorResult.retryDelayInMs,
+              delay: handleErrorResult.retryDelayInMs,
+            },
+            error: handleErrorResult.error,
+          };
+        }
+
+        if (handleErrorResult.retry && typeof handleErrorResult.retry === "object") {
+          const delay = calculateNextRetryDelay(handleErrorResult.retry, execution.attempt.number);
+
+          return typeof delay === "undefined"
+            ? { status: "noop", error: handleErrorResult.error }
+            : {
+                status: "retry",
+                retry: { timestamp: Date.now() + delay, delay },
+                error: handleErrorResult.error,
+              };
+        }
+
+        return { status: "noop", error: handleErrorResult.error };
+      },
+      {
+        attributes: {
+          [SemanticInternalAttributes.STYLE_ICON]: "exclamation-circle",
+        },
+      }
+    );
+  }
+}
diff --git a/packages/trigger-sdk/src/v3/config.ts b/packages/trigger-sdk/src/v3/config.ts
index 10beeaf05..5d1581097 100644
--- a/packages/trigger-sdk/src/v3/config.ts
+++ b/packages/trigger-sdk/src/v3/config.ts
@@ -1,11 +1 @@
-import { RetryOptions } from "./retry";
-
-export interface Config {
-  project: string;
-  triggerDirectories?: string | string[];
-  triggerUrl?: string;
-  retries?: {
-    enabledInDev?: boolean;
-    default?: RetryOptions;
-  };
-}
+export type { ProjectConfig as Config } from "@trigger.dev/core/v3";
diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts
index b55d2566f..e87c5369f 100644
--- a/packages/trigger-sdk/src/v3/shared.ts
+++ b/packages/trigger-sdk/src/v3/shared.ts
@@ -1,9 +1,16 @@
 import { SpanKind } from "@opentelemetry/api";
 import { SemanticAttributes } from "@opentelemetry/semantic-conventions";
 import {
+  HandleErrorFnParams,
+  HandleErrorResult,
+  InitFnParams,
+  InitOutput,
+  MiddlewareFnParams,
   QueueOptions,
   RetryOptions,
+  RunFnParams,
   SemanticInternalAttributes,
+  SuccessFnParams,
   TaskRunContext,
   accessoryAttributes,
   apiClientManager,
@@ -16,34 +23,8 @@ import {
 import * as packageJson from "../../package.json";
 import { tracer } from "./tracer";
 
-export type InitOutput = Record<string, any> | void | undefined;
-
-export type RunFnParams<TInitOutput extends InitOutput> = Prettify<{
-  ctx: Context;
-  init: TInitOutput;
-}>;
-
-export type MiddlewareFnParams = Prettify<{
-  ctx: Context;
-  next: () => Promise<void>;
-}>;
-
-export type InitFnParams = Prettify<{
-  ctx: Context;
-}>;
-
 export type Context = TaskRunContext;
 
-export type SuccessFnParams<TOutput, TInitOutput extends InitOutput> = RunFnParams<TInitOutput> &
-  Prettify<{
-    output: TOutput;
-  }>;
-
-export type ErrorFnParams<TInitOutput extends InitOutput> = RunFnParams<TInitOutput> &
-  Prettify<{
-    error: unknown;
-  }>;
-
 type RequireOne<T, K extends keyof T> = {
   [X in Exclude<keyof T, K>]?: T[X];
 } & {
@@ -66,10 +47,14 @@ export type TaskOptions<TPayload, TOutput = any, TInitOutput extends InitOutput
   };
   run: (payload: TPayload, params: RunFnParams<TInitOutput>) => Promise<TOutput>;
   init?: (payload: TPayload, params: InitFnParams) => Promise<TInitOutput>;
+  handleError?: (
+    payload: TPayload,
+    error: unknown,
+    params: HandleErrorFnParams<TInitOutput>
+  ) => HandleErrorResult;
   cleanup?: (payload: TPayload, params: RunFnParams<TInitOutput>) => Promise<void>;
   middleware?: (payload: TPayload, params: MiddlewareFnParams) => Promise<void>;
   onSuccess?: (payload: TPayload, params: SuccessFnParams<TOutput, TInitOutput>) => Promise<void>;
-  onError?: (payload: TPayload, params: ErrorFnParams<TInitOutput>) => Promise<void>;
 };
 
 type InvokeHandle = {
@@ -437,6 +422,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
         init: params.init,
         cleanup: params.cleanup,
         middleware: params.middleware,
+        handleError: params.handleError,
       },
     },
     enumerable: false,
diff --git a/packages/trigger-sdk/src/v3/tasks.ts b/packages/trigger-sdk/src/v3/tasks.ts
index 2ee68895b..0c9a2ad9f 100644
--- a/packages/trigger-sdk/src/v3/tasks.ts
+++ b/packages/trigger-sdk/src/v3/tasks.ts
@@ -1,4 +1,5 @@
-import { InitOutput, TaskOptions, Task, createTask } from "./shared";
+import { InitOutput } from "@trigger.dev/core/v3";
+import { TaskOptions, Task, createTask } from "./shared";
 
 export function task<TInput, TOutput = any, TInitOutput extends InitOutput = any>(
   options: TaskOptions<TInput, TOutput, TInitOutput>
diff --git a/references/v3-catalog/src/trigger/openai.ts b/references/v3-catalog/src/trigger/openai.ts
index a532af0ae..97db420e1 100644
--- a/references/v3-catalog/src/trigger/openai.ts
+++ b/references/v3-catalog/src/trigger/openai.ts
@@ -1,4 +1,4 @@
-import { task } from "@trigger.dev/sdk/v3";
+import { logger, task } from "@trigger.dev/sdk/v3";
 
 import OpenAI from "openai";
 
@@ -8,6 +8,9 @@ const openai = new OpenAI({
 
 export const openaiTask = task({
   id: "openai-task",
+  retry: {
+    maxAttempts: 1,
+  },
   run: async (payload: { prompt: string }) => {
     const chatCompletion = await openai.chat.completions.create({
       messages: [{ role: "user", content: payload.prompt }],
@@ -16,4 +19,14 @@ export const openaiTask = task({
 
     return chatCompletion.choices[0].message.content;
   },
+  handleError: async (payload, err, { ctx, retryAt }) => {
+    if (err instanceof OpenAI.APIError) {
+      logger.log("OpenAI API error", { err });
+
+      return {
+        error: new Error("Custom OpenAI API error"),
+        retryDelayInMs: 10000,
+      };
+    }
+  },
 });
diff --git a/references/v3-catalog/trigger.config.mjs b/references/v3-catalog/trigger.config.mjs
index 7600746fd..4ac1ca713 100644
--- a/references/v3-catalog/trigger.config.mjs
+++ b/references/v3-catalog/trigger.config.mjs
@@ -1,3 +1,4 @@
+// @ts-check
 /** @type {import('@trigger.dev/sdk/v3').Config} */
 
 export default {
@@ -12,4 +13,7 @@ export default {
       randomize: true,
     },
   },
+  handleError: async (payload, error, { ctx, retryAt, retryDelayInMs, retry }) => {
+    return { skipRetrying: true };
+  },
 };