Fix pnpm lock file

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

.changeset/angry-eagles-trade.md

@@ -0,0 +1,9 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Fix additionalFiles that aren't decendants
+- Stop swallowing uncaught exceptions in prod
+- Improve warnings and errors, fail early on critical warnings
+- New arg to --save-logs even for successful builds

.changeset/beige-pens-dance.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+v3 CLI update command and package manager detection fix

.changeset/eleven-paws-join.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Remove the env var check during deploy (too many false negatives)

.changeset/funny-swans-destroy.md

@@ -0,0 +1,9 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Make msw a normal dependency (for now) to fix Module Not Found error in Next.js.
+
+It turns out that webpack will "hoist" dynamically imported modules and attempt to resolve them at build time, even though it's an optional peer dep:
+
+https://x.com/maverickdotdev/status/1782465214308319404

.changeset/nasty-jars-pump.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fixes an issue that was treating v2 trigger directories as v3

.changeset/pre.json

@@ -44,6 +44,8 @@
     "@trigger.dev/yalt": "2.3.18"
   },
   "changesets": [
+    "angry-eagles-trade",
+    "beige-pens-dance",
     "breezy-gorillas-mate",
     "chilled-hornets-move",
     "clean-pianos-listen",
@@ -51,7 +53,9 @@
     "cuddly-feet-approve",
     "dry-walls-check",
     "eight-pumas-float",
+    "eleven-paws-join",
     "few-students-share",
+    "funny-swans-destroy",
     "green-bags-wink",
     "khaki-apricots-design",
     "khaki-poems-lay",
@@ -63,6 +67,7 @@
     "loud-actors-remember",
     "many-ligers-pump",
     "mighty-camels-joke",
+    "nasty-jars-pump",
     "new-rivers-tell",
     "ninety-pets-travel",
     "odd-poets-own",
@@ -70,9 +75,17 @@
     "poor-flowers-cross",
     "rare-roses-float",
     "real-planets-stare",
+    "rich-kangaroos-unite",
+    "rotten-beers-refuse",
     "rotten-dryers-exercise",
+    "selfish-ducks-sort",
+    "shaggy-spoons-taste",
+    "sharp-emus-compare",
     "sharp-zebras-serve",
+    "shiny-coats-cry",
     "silly-suits-switch",
+    "slow-buses-own",
+    "smart-needles-move",
     "smart-olives-eat",
     "spicy-lamps-smoke",
     "strange-ghosts-matter",
@@ -84,7 +97,11 @@
     "tame-guests-know",
     "tender-oranges-rhyme",
     "tidy-balloons-suffer",
+    "tidy-dryers-sleep",
     "tiny-doors-type",
-    "tricky-bulldogs-heal"
+    "tiny-elephants-scream",
+    "tricky-bulldogs-heal",
+    "tricky-ladybugs-unite",
+    "two-pumas-wait"
   ]
 }

.changeset/rich-kangaroos-unite.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fix a bug where revoking the CLI token would prevent you from ever logging in again with the CLI.

.changeset/rotten-beers-refuse.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add git to prod worker image which fixes private package installs

.changeset/selfish-ducks-sort.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+---
+
+Lock SDK and CLI deps on exact core version

.changeset/shaggy-spoons-taste.md

@@ -0,0 +1,56 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Updates the `trigger`, `batchTrigger` and their `*AndWait` variants to use the first parameter for the payload/items, and the second parameter for options.
+
+Before:
+
+```ts
+await yourTask.trigger({ payload: { foo: "bar" }, options: { idempotencyKey: "key_1234" } });
+await yourTask.triggerAndWait({ payload: { foo: "bar" }, options: { idempotencyKey: "key_1234" } });
+
+await yourTask.batchTrigger({ items: [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }] });
+await yourTask.batchTriggerAndWait({ items: [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }] });
+```
+
+After:
+
+```ts
+await yourTask.trigger({ foo: "bar" }, { idempotencyKey: "key_1234" });
+await yourTask.triggerAndWait({ foo: "bar" }, { idempotencyKey: "key_1234" });
+
+await yourTask.batchTrigger([{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }]);
+await yourTask.batchTriggerAndWait([{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }]);
+```
+
+We've also changed the API of the `triggerAndWait` result. Before, if the subtask that was triggered finished with an error, we would automatically "rethrow" the error in the parent task.
+
+Now instead we're returning a `TaskRunResult` object that allows you to discriminate between successful and failed runs in the subtask:
+
+Before:
+
+```ts
+try {
+  const result = await yourTask.triggerAndWait({ foo: "bar" });
+
+  // result is the output of your task
+  console.log("result", result);
+
+} catch (error) {
+  // handle subtask errors here
+}
+```
+
+After:
+
+```ts
+const result = await yourTask.triggerAndWait({ foo: "bar" });
+
+if (result.ok) {
+  console.log(`Run ${result.id} succeeded with output`, result.output);
+} else {
+  console.log(`Run ${result.id} failed with error`, result.error);
+}
+```

.changeset/sharp-emus-compare.md

@@ -0,0 +1,13 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)
+
+- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
+- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
+- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
+- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task
+

.changeset/shiny-coats-cry.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add additional logging around cleaning up dev workers, and always kill them after 5 seconds if they haven't already exited

.changeset/slow-buses-own.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+"@trigger.dev/cli": patch
+---
+
+Move to our global system from AsyncLocalStorage for the current task context storage

.changeset/smart-needles-move.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Remove unimplemented batchOptions

.changeset/tidy-dryers-sleep.md

@@ -0,0 +1,10 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Fixes an issue that caused failed tasks when resuming after calling `triggerAndWait` or `batchTriggerAndWait` in prod/staging (this doesn't effect dev).
+
+The version of Node.js we use for deployed workers (latest 20) would crash with an out-of-memory error when the checkpoint was restored. This crash does not happen on Node 18x or Node21x, so we've decided to upgrade the worker version to Node.js21x, to mitigate this issue.
+
+You'll need to re-deploy to production to fix the issue.

.changeset/tiny-elephants-scream.md

@@ -0,0 +1,7 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Add graceful exit for prod workers
+- Prevent overflow in long waits

.changeset/tricky-ladybugs-unite.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core-apps": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Extracting out all the non-SDK related features from the main @trigger.dev/core/v3 export

.changeset/two-pumas-wait.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/cli": patch
+---
+
+Add support for tasks located in subdirectories inside trigger dirs

.github/workflows/pr_checks.yml

@@ -27,6 +27,6 @@ jobs:
     uses: ./.github/workflows/unit-tests.yml
     secrets: inherit
 
-  e2e:
-    uses: ./.github/workflows/e2e.yml
-    secrets: inherit
+  # e2e:
+  #   uses: ./.github/workflows/e2e.yml
+  #   secrets: inherit

.github/workflows/publish.yml

@@ -49,11 +49,11 @@ jobs:
     uses: ./.github/workflows/unit-tests.yml
     secrets: inherit
 
-  e2e:
-    uses: ./.github/workflows/e2e.yml
-    secrets: inherit
+  # e2e:
+  #   uses: ./.github/workflows/e2e.yml
+  #   secrets: inherit
 
   publish:
-    needs: [typecheck, units, e2e]
+    needs: [typecheck, units]
     uses: ./.github/workflows/publish-docker.yml
     secrets: inherit

apps/coordinator/package.json

@@ -21,8 +21,8 @@
     "execa": "^8.0.1",
     "nanoid": "^5.0.6",
     "prom-client": "^15.1.0",
-    "socket.io": "^4.7.4",
-    "socket.io-client": "^4.7.4"
+    "socket.io": "4.7.4",
+    "socket.io-client": "4.7.4"
   },
   "devDependencies": {
     "@types/node": "^18",

apps/coordinator/src/index.ts

@@ -8,9 +8,9 @@ import {
   PlatformToCoordinatorMessages,
   ProdWorkerSocketData,
   ProdWorkerToCoordinatorMessages,
-  ZodNamespace,
-  ZodSocketConnection,
 } from "@trigger.dev/core/v3";
+import { ZodNamespace } from "@trigger.dev/core/v3/zodNamespace";
+import { ZodSocketConnection } from "@trigger.dev/core/v3/zodSocket";
 import { HttpReply, getTextBody, SimpleLogger } from "@trigger.dev/core-apps";
 
 import { collectDefaultMetrics, register, Gauge } from "prom-client";

apps/docker-provider/src/index.ts

@@ -257,6 +257,7 @@ class DockerTaskOperations implements TaskOperations {
       return await execa("docker", [
         "exec",
         containerName,
+        "busybox",
         "wget",
         "-q",
         "-O-",

apps/kubernetes-provider/src/index.ts

@@ -133,6 +133,7 @@ class KubernetesTaskOperations implements TaskOperations {
         },
         spec: {
           ...this.#defaultPodSpec,
+          terminationGracePeriodSeconds: 60 * 60,
           containers: [
             {
               name: this.#getRunContainerName(opts.runId),
@@ -210,7 +211,7 @@ class KubernetesTaskOperations implements TaskOperations {
             },
             {
               name: "populate-taskinfo",
-              image: "busybox",
+              image: "docker.io/library/busybox",
               command: ["/bin/sh", "-c"],
               args: ["printenv COORDINATOR_HOST | tee /etc/taskinfo/coordinator-host"],
               env: [
@@ -409,7 +410,7 @@ class KubernetesTaskOperations implements TaskOperations {
       `for i in $(seq ${retries}); do sleep 1; busybox wget -q -O- 127.0.0.1:8000/${type}?cause=${cause} && break; done`,
     ];
 
-    logger.log("getLifecycleCommand()", { exec });
+    logger.debug("getLifecycleCommand()", { exec });
 
     return exec;
   }

apps/kubernetes-provider/src/taskMonitor.ts

@@ -408,7 +408,7 @@ export class TaskMonitor {
     const createNonZeroExitPod = async (name: string, labels?: Record<string, string>) => {
       const container = {
         name,
-        image: "busybox",
+        image: "docker.io/library/busybox",
         command: ["sh"],
         args: ["-c", "exit 1"],
       } satisfies k8s.V1Container;
@@ -419,7 +419,7 @@ export class TaskMonitor {
     const createOoDiskPod = async (name: string, labels?: Record<string, string>) => {
       const container = {
         name,
-        image: "busybox",
+        image: "docker.io/library/busybox",
         command: ["sh"],
         args: [
           "-c",

apps/webapp/app/components/primitives/ShortcutKey.tsx

@@ -2,8 +2,14 @@ import { Fragment } from "react";
 import { Modifier, ShortcutDefinition } from "~/hooks/useShortcutKeys";
 import { cn } from "~/utils/cn";
 import { useOperatingSystem } from "./OperatingSystemProvider";
+import {
+  ChevronDownIcon,
+  ChevronLeftIcon,
+  ChevronRightIcon,
+  ChevronUpIcon,
+} from "@heroicons/react/20/solid";
 
-const variants = {
+export const variants = {
   small:
     "text-[0.6rem] font-medium min-w-[17px] rounded-[2px] px-1 ml-1 -mr-0.5 grid place-content-center border border-dimmed/40 text-text-dimmed group-hover:text-text-bright/80 group-hover:border-dimmed/60 transition uppercase",
   medium:
@@ -23,7 +29,7 @@ export function ShortcutKey({ shortcut, variant, className }: ShortcutKeyProps)
   const isMac = platform === "mac";
   let relevantShortcut = "mac" in shortcut ? (isMac ? shortcut.mac : shortcut.windows) : shortcut;
   const modifiers = relevantShortcut.modifiers ?? [];
-  const character = keyString(relevantShortcut.key, isMac);
+  const character = keyString(relevantShortcut.key, isMac, variant);
 
   return (
     <span className={cn(variants[variant], className)}>
@@ -35,10 +41,22 @@ export function ShortcutKey({ shortcut, variant, className }: ShortcutKeyProps)
   );
 }
 
-function keyString(key: String, isMac: boolean) {
+function keyString(key: String, isMac: boolean, size: "small" | "medium") {
+  key = key.toLowerCase();
+
+  const className = size === "small" ? "w-2.5 h-4" : "w-3 h-5";
+
   switch (key) {
     case "enter":
       return isMac ? "↵" : key;
+    case "arrowdown":
+      return <ChevronDownIcon className={className} />;
+    case "arrowup":
+      return <ChevronUpIcon className={className} />;
+    case "arrowleft":
+      return <ChevronLeftIcon className={className} />;
+    case "arrowright":
+      return <ChevronRightIcon className={className} />;
     default:
       return key;
   }

apps/webapp/app/components/primitives/TreeView/TreeView.tsx

@@ -1,10 +1,10 @@
 import { VirtualItem, Virtualizer, useVirtualizer } from "@tanstack/react-virtual";
+import { motion } from "framer-motion";
 import { MutableRefObject, RefObject, useCallback, useEffect, useReducer, useRef } from "react";
 import { UnmountClosed } from "react-collapse";
 import { cn } from "~/utils/cn";
 import { NodeState, NodesState, reducer } from "./reducer";
 import { applyFilterToState, concreteStateFromInput, selectedIdFromState } from "./utils";
-import { motion } from "framer-motion";
 
 export type TreeViewProps<TData> = {
   tree: FlatTree<TData>;
@@ -165,6 +165,11 @@ export type UseTreeStateOutput = {
   expandNode: (id: string, scrollToNode?: boolean) => void;
   collapseNode: (id: string) => void;
   toggleExpandNode: (id: string, scrollToNode?: boolean) => void;
+  expandAllBelowDepth: (depth: number) => void;
+  collapseAllBelowDepth: (depth: number) => void;
+  expandLevel: (level: number) => void;
+  collapseLevel: (level: number) => void;
+  toggleExpandLevel: (level: number) => void;
   selectFirstVisibleNode: (scrollToNode?: boolean) => void;
   selectLastVisibleNode: (scrollToNode?: boolean) => void;
   selectNextVisibleNode: (scrollToNode?: boolean) => void;
@@ -333,6 +338,41 @@ export function useTree<TData>({
     [state]
   );
 
+  const expandAllBelowDepth = useCallback(
+    (depth: number) => {
+      dispatch({ type: "EXPAND_ALL_BELOW_DEPTH", payload: { tree, depth } });
+    },
+    [state]
+  );
+
+  const collapseAllBelowDepth = useCallback(
+    (depth: number) => {
+      dispatch({ type: "COLLAPSE_ALL_BELOW_DEPTH", payload: { tree, depth } });
+    },
+    [state]
+  );
+
+  const expandLevel = useCallback(
+    (level: number) => {
+      dispatch({ type: "EXPAND_LEVEL", payload: { tree, level } });
+    },
+    [state]
+  );
+
+  const collapseLevel = useCallback(
+    (level: number) => {
+      dispatch({ type: "COLLAPSE_LEVEL", payload: { tree, level } });
+    },
+    [state]
+  );
+
+  const toggleExpandLevel = useCallback(
+    (level: number) => {
+      dispatch({ type: "TOGGLE_EXPAND_LEVEL", payload: { tree, level } });
+    },
+    [state]
+  );
+
   const getTreeProps = useCallback(() => {
     return {
       role: "tree",
@@ -368,25 +408,48 @@ export function useTree<TData>({
           }
           case "Left":
           case "ArrowLeft": {
+            e.preventDefault();
+
             const selected = selectedIdFromState(state.nodes);
             if (selected) {
               const treeNode = tree.find((node) => node.id === selected);
-              if (treeNode && treeNode.hasChildren && state.nodes[selected].expanded) {
+
+              if (e.altKey) {
+                if (treeNode && treeNode.hasChildren) {
+                  collapseLevel(treeNode.level);
+                }
+                break;
+              }
+
+              const shouldCollapse =
+                treeNode && treeNode.hasChildren && state.nodes[selected].expanded;
+              if (shouldCollapse) {
                 collapseNode(selected);
               } else {
                 selectParentNode(true);
               }
             }
-            e.preventDefault();
+
             break;
           }
           case "Right":
           case "ArrowRight": {
+            e.preventDefault();
+
             const selected = selectedIdFromState(state.nodes);
+
             if (selected) {
+              const treeNode = tree.find((node) => node.id === selected);
+
+              if (e.altKey) {
+                if (treeNode && treeNode.hasChildren) {
+                  expandLevel(treeNode.level);
+                }
+                break;
+              }
+
               expandNode(selected, true);
             }
-            e.preventDefault();
             break;
           }
           case "Escape": {
@@ -427,6 +490,11 @@ export function useTree<TData>({
     expandNode,
     collapseNode,
     toggleExpandNode,
+    expandAllBelowDepth,
+    collapseAllBelowDepth,
+    expandLevel,
+    collapseLevel,
+    toggleExpandLevel,
     selectFirstVisibleNode,
     selectLastVisibleNode,
     selectNextVisibleNode,

apps/webapp/app/components/primitives/TreeView/reducer.ts

@@ -91,6 +91,46 @@ type ToggleExpandNodeAction = {
   } & WithScrollToNode;
 };
 
+type ExpandAllBelowDepthAction = {
+  type: "EXPAND_ALL_BELOW_DEPTH";
+  payload: {
+    depth: number;
+    tree: FlatTree<any>;
+  };
+};
+
+type CollapseAllBelowDepthAction = {
+  type: "COLLAPSE_ALL_BELOW_DEPTH";
+  payload: {
+    depth: number;
+    tree: FlatTree<any>;
+  };
+};
+
+type ExpandLevelAction = {
+  type: "EXPAND_LEVEL";
+  payload: {
+    level: number;
+    tree: FlatTree<any>;
+  };
+};
+
+type CollapseLevelAction = {
+  type: "COLLAPSE_LEVEL";
+  payload: {
+    level: number;
+    tree: FlatTree<any>;
+  };
+};
+
+type ToggleExpandLevelAction = {
+  type: "TOGGLE_EXPAND_LEVEL";
+  payload: {
+    level: number;
+    tree: FlatTree<any>;
+  };
+};
+
 type SelectFirstVisibleNodeAction = {
   type: "SELECT_FIRST_VISIBLE_NODE";
   payload: {
@@ -135,6 +175,11 @@ export type Action =
   | ExpandNodeAction
   | CollapseNodeAction
   | ToggleExpandNodeAction
+  | ExpandAllBelowDepthAction
+  | CollapseAllBelowDepthAction
+  | ExpandLevelAction
+  | CollapseLevelAction
+  | ToggleExpandLevelAction
   | SelectFirstVisibleNodeAction
   | SelectLastVisibleNodeAction
   | SelectNextVisibleNodeAction
@@ -229,6 +274,109 @@ export function reducer(state: TreeState, action: Action): TreeState {
         });
       }
     }
+    case "EXPAND_ALL_BELOW_DEPTH": {
+      const nodesToExpand = action.payload.tree.filter(
+        (n) => n.level >= action.payload.depth && n.hasChildren
+      );
+
+      const newNodes = Object.fromEntries(
+        Object.entries(state.nodes).map(([key, value]) => [
+          key,
+          {
+            ...value,
+            expanded: nodesToExpand.find((n) => n.id === key) ? true : value.expanded,
+          },
+        ])
+      );
+
+      const visibleNodes = applyVisibility(action.payload.tree, newNodes);
+      return { nodes: visibleNodes, changes: generateChanges(state.nodes, visibleNodes) };
+    }
+    case "COLLAPSE_ALL_BELOW_DEPTH": {
+      const nodesToCollapse = action.payload.tree.filter(
+        (n) => n.level >= action.payload.depth && n.hasChildren
+      );
+
+      const newNodes = Object.fromEntries(
+        Object.entries(state.nodes).map(([key, value]) => [
+          key,
+          {
+            ...value,
+            expanded: nodesToCollapse.find((n) => n.id === key) ? false : value.expanded,
+          },
+        ])
+      );
+
+      const visibleNodes = applyVisibility(action.payload.tree, newNodes);
+      return { nodes: visibleNodes, changes: generateChanges(state.nodes, visibleNodes) };
+    }
+    case "EXPAND_LEVEL": {
+      const nodesToExpand = action.payload.tree.filter(
+        (n) => n.level <= action.payload.level && n.hasChildren
+      );
+
+      const newNodes = Object.fromEntries(
+        Object.entries(state.nodes).map(([key, value]) => [
+          key,
+          {
+            ...value,
+            expanded: nodesToExpand.find((n) => n.id === key) ? true : value.expanded,
+          },
+        ])
+      );
+
+      const visibleNodes = applyVisibility(action.payload.tree, newNodes);
+      return { nodes: visibleNodes, changes: generateChanges(state.nodes, visibleNodes) };
+    }
+    case "COLLAPSE_LEVEL": {
+      const nodesToCollapse = action.payload.tree.filter(
+        (n) => n.level === action.payload.level && n.hasChildren
+      );
+
+      const newNodes = Object.fromEntries(
+        Object.entries(state.nodes).map(([key, value]) => [
+          key,
+          {
+            ...value,
+            expanded: nodesToCollapse.find((n) => n.id === key) ? false : value.expanded,
+          },
+        ])
+      );
+
+      const visibleNodes = applyVisibility(action.payload.tree, newNodes);
+      return { nodes: visibleNodes, changes: generateChanges(state.nodes, visibleNodes) };
+    }
+    case "TOGGLE_EXPAND_LEVEL": {
+      //first get the first item at that level in the tree. If it is expanded, collapse all nodes at that level
+      //if it is collapsed, expand all nodes at that level
+      const nodesAtLevel = action.payload.tree.filter(
+        (n) => n.level === action.payload.level && n.hasChildren
+      );
+      const firstNode = nodesAtLevel[0];
+      if (!firstNode) {
+        return state;
+      }
+
+      const currentlyExpanded = state.nodes[firstNode.id]?.expanded ?? true;
+      const currentVisible = state.nodes[firstNode.id]?.visible ?? true;
+      if (currentlyExpanded && currentVisible) {
+        return reducer(state, {
+          type: "COLLAPSE_LEVEL",
+          payload: {
+            level: action.payload.level,
+            tree: action.payload.tree,
+          },
+        });
+      } else {
+        return reducer(state, {
+          type: "EXPAND_LEVEL",
+          payload: {
+            level: action.payload.level,
+            tree: action.payload.tree,
+          },
+        });
+      }
+    }
     case "SELECT_FIRST_VISIBLE_NODE": {
       const node = firstVisibleNode(action.payload.tree, state.nodes);
       if (node) {

apps/webapp/app/components/runs/RunFilters.tsx

@@ -32,6 +32,7 @@ import {
 import { TimeFrameFilter } from "./TimeFrameFilter";
 import { Button } from "../primitives/Buttons";
 import { useCallback } from "react";
+import assertNever from "assert-never";
 
 export function RunsFilters() {
   const navigate = useNavigate();
@@ -182,8 +183,7 @@ export function FilterStatusIcon({
     case "FAILED":
       return <XCircleIcon className={cn(filterStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -205,8 +205,7 @@ export function filterStatusTitle(status: FilterableStatus): string {
     case "TIMEDOUT":
       return "Timed out";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -228,8 +227,7 @@ export function filterStatusClassNameColor(status: FilterableStatus): string {
     case "TIMEDOUT":
       return "text-amber-300";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/RunStatuses.tsx

@@ -11,6 +11,7 @@ import type { JobRunStatus } from "@trigger.dev/database";
 import { cn } from "~/utils/cn";
 import { Spinner } from "../primitives/Spinner";
 import { z } from "zod";
+import assertNever from "assert-never";
 
 export function RunStatus({ status }: { status: JobRunStatus }) {
   return (
@@ -51,8 +52,7 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla
     case "CANCELED":
       return <NoSymbolIcon className={cn(runStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -89,8 +89,7 @@ export function runStatusTitle(status: JobRunStatus): string {
     case "INVALID_PAYLOAD":
       return "Invalid payload";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -123,8 +122,7 @@ export function runStatusClassNameColor(status: JobRunStatus): string {
     case "CANCELED":
       return "text-charcoal-500";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/v3/DeploymentStatus.tsx

@@ -5,6 +5,7 @@ import {
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { WorkerDeploymentStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
@@ -54,8 +55,7 @@ export function DeploymentStatusIcon({
         />
       );
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -74,8 +74,7 @@ export function deploymentStatusClassNameColor(status: WorkerDeploymentStatus):
     case "FAILED":
       return "text-error";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -97,8 +96,7 @@ export function deploymentStatusTitle(status: WorkerDeploymentStatus): string {
     case "FAILED":
       return "Failed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/v3/RunFilters.tsx

@@ -1,4 +1,4 @@
-import { TrashIcon } from "@heroicons/react/20/solid";
+import { XMarkIcon } from "@heroicons/react/20/solid";
 import { useNavigate } from "@remix-run/react";
 import type { TaskRunStatus as TaskRunStatusType } from "@trigger.dev/database";
 import { RuntimeEnvironment, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
@@ -247,7 +247,7 @@ export function RunsFilters({ possibleEnvironments, possibleTasks }: RunFiltersP
 
       <TimeFrameFilter from={from} to={to} onRangeChanged={handleTimeFrameChange} />
 
-      <Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={TrashIcon} />
+      <Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={XMarkIcon} />
     </div>
   );
 }

apps/webapp/app/components/runs/v3/TaskRunAttemptStatus.tsx

@@ -8,6 +8,7 @@ import {
 } from "@heroicons/react/20/solid";
 import type { TaskRunAttemptStatus as TaskRunAttemptStatusType } from "@trigger.dev/database";
 import { TaskRunAttemptStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { SnowflakeIcon } from "lucide-react";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
@@ -72,8 +73,7 @@ export function TaskRunAttemptStatusIcon({
     case "COMPLETED":
       return <CheckCircleIcon className={cn(runAttemptStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -99,8 +99,7 @@ export function runAttemptStatusClassNameColor(status: ExtendedTaskAttemptStatus
     case "COMPLETED":
       return "text-success";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -126,8 +125,7 @@ export function runAttemptStatusTitle(status: ExtendedTaskAttemptStatus | null):
     case "COMPLETED":
       return "Completed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/v3/TaskRunStatus.tsx

@@ -10,6 +10,7 @@ import {
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { TaskRunStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { SnowflakeIcon } from "lucide-react";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
@@ -88,8 +89,7 @@ export function TaskRunStatusIcon({
       return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
 
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -120,8 +120,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
     case "CRASHED":
       return "text-error";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -153,8 +152,7 @@ export function runStatusTitle(status: TaskRunStatus): string {
     case "CRASHED":
       return "Crashed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/v3/TaskRunsTable.tsx

@@ -1,10 +1,16 @@
+import { ArrowPathIcon, StopCircleIcon } from "@heroicons/react/20/solid";
 import { StopIcon } from "@heroicons/react/24/outline";
 import { BeakerIcon, BookOpenIcon, CheckIcon } from "@heroicons/react/24/solid";
+import { useLocation } from "@remix-run/react";
+import { formatDuration } from "@trigger.dev/core/v3";
 import { User } from "@trigger.dev/database";
+import { Button, LinkButton } from "~/components/primitives/Buttons";
+import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
+import { useEnvironments } from "~/hooks/useEnvironments";
 import { useOrganization } from "~/hooks/useOrganizations";
 import { useProject } from "~/hooks/useProject";
 import { RunListAppliedFilters, RunListItem } from "~/presenters/v3/RunListPresenter.server";
-import { docsPath, v3RunPath, v3TestPath } from "~/utils/pathBuilder";
+import { docsPath, v3RunSpanPath, v3TestPath } from "~/utils/pathBuilder";
 import { EnvironmentLabel } from "../../environments/EnvironmentLabel";
 import { DateTime } from "../../primitives/DateTime";
 import { Paragraph } from "../../primitives/Paragraph";
@@ -14,21 +20,14 @@ import {
   TableBlankRow,
   TableBody,
   TableCell,
-  TableCellChevron,
   TableCellMenu,
   TableHeader,
   TableHeaderCell,
   TableRow,
 } from "../../primitives/Table";
-import { formatDuration } from "@trigger.dev/core/v3";
-import { TaskRunStatusCombo } from "./TaskRunStatus";
-import { useEnvironments } from "~/hooks/useEnvironments";
-import { Button, LinkButton } from "~/components/primitives/Buttons";
-import { ArrowPathIcon, StopCircleIcon } from "@heroicons/react/20/solid";
-import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
 import { CancelRunDialog } from "./CancelRunDialog";
-import { useLocation } from "@remix-run/react";
 import { ReplayRunDialog } from "./ReplayRunDialog";
+import { TaskRunStatusCombo } from "./TaskRunStatus";
 
 type RunsTableProps = {
   total: number;
@@ -78,7 +77,7 @@ export function TaskRunsTable({
           <BlankState isLoading={isLoading} filters={filters} />
         ) : (
           runs.map((run) => {
-            const path = v3RunPath(organization, project, run);
+            const path = v3RunSpanPath(organization, project, run, { spanId: run.spanId });
             const usernameForEnv =
               currentUser.id !== run.environment.userId ? run.environment.userName : undefined;
             return (

apps/webapp/app/db.server.ts

@@ -4,6 +4,7 @@ import { z } from "zod";
 import { logger } from "./services/logger.server";
 import { env } from "./env.server";
 import { singleton } from "./utils/singleton";
+import { isValidDatabaseUrl } from "./utils/db";
 
 export type PrismaTransactionClient = Omit<
   PrismaClient,
@@ -138,3 +139,23 @@ export type { PrismaClient } from "@trigger.dev/database";
 export const PrismaErrorSchema = z.object({
   code: z.string(),
 });
+
+function getDatabaseSchema() {
+  if (!isValidDatabaseUrl(env.DATABASE_URL)) {
+    throw new Error("Invalid Database URL");
+  }
+
+  const databaseUrl = new URL(env.DATABASE_URL);
+  const schemaFromSearchParam = databaseUrl.searchParams.get("schema");
+
+  if (!schemaFromSearchParam) {
+    console.debug("❗ database schema unspecified, will default to `public` schema");
+    return "public";
+  }
+
+  return schemaFromSearchParam;
+}
+
+export const DATABASE_SCHEMA = singleton("DATABASE_SCHEMA", getDatabaseSchema);
+
+export const sqlDatabaseSchema = Prisma.sql([`${DATABASE_SCHEMA}`]);

apps/webapp/app/env.server.ts

@@ -1,13 +1,24 @@
 import { z } from "zod";
 import { SecretStoreOptionsSchema } from "./services/secrets/secretStore.server";
 import { isValidRegex } from "./utils/regex";
+import { isValidDatabaseUrl } from "./utils/db";
 
 const EnvironmentSchema = z.object({
   NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
-  DATABASE_URL: z.string(),
+  DATABASE_URL: z
+    .string()
+    .refine(
+      isValidDatabaseUrl,
+      "DATABASE_URL is invalid, for details please check the additional output above this message."
+    ),
   DATABASE_CONNECTION_LIMIT: z.coerce.number().int().default(10),
   DATABASE_POOL_TIMEOUT: z.coerce.number().int().default(60),
-  DIRECT_URL: z.string(),
+  DIRECT_URL: z
+    .string()
+    .refine(
+      isValidDatabaseUrl,
+      "DIRECT_URL is invalid, for details please check the additional output above this message."
+    ),
   SESSION_SECRET: z.string(),
   MAGIC_LINK_SECRET: z.string(),
   ENCRYPTION_KEY: z.string(),
@@ -65,7 +76,6 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
-  DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
   DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),

apps/webapp/app/hooks/useReplaceLocation.ts

@@ -0,0 +1,32 @@
+import { useCallback, useState } from "react";
+import { useOptimisticLocation } from "./useOptimisticLocation";
+import type { Location } from "@remix-run/react";
+
+export function useReplaceLocation() {
+  const optimisticLocation = useOptimisticLocation();
+  const [location, setLocation] = useState(optimisticLocation);
+
+  const replaceLocation = useCallback((location: Location<any>) => {
+    const fullPath = location.pathname + location.search + location.hash;
+    //replace the URL in the browser
+    history.replaceState(null, "", fullPath);
+    //update the state (new object in case the same location ref was modified)
+    const newLocation = { ...location };
+    setLocation(newLocation);
+  }, []);
+
+  const replaceSearchParam = useCallback(
+    (key: string, value?: string) => {
+      const searchParams = new URLSearchParams(location.search);
+      if (value) {
+        searchParams.set(key, value);
+      } else {
+        searchParams.delete(key);
+      }
+      replaceLocation({ ...optimisticLocation, search: "?" + searchParams.toString() });
+    },
+    [optimisticLocation, replaceLocation]
+  );
+
+  return { location, replaceLocation, replaceSearchParam };
+}

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

@@ -0,0 +1,128 @@
+import {
+  TaskRunError,
+  TaskRunExecutionResult,
+  TaskRunFailedExecutionResult,
+  TaskRunSuccessfulExecutionResult,
+} from "@trigger.dev/core/v3";
+import {
+  BatchTaskRunItemStatus,
+  TaskRun,
+  TaskRunAttempt,
+  TaskRunAttemptStatus,
+  TaskRunStatus,
+} from "@trigger.dev/database";
+import { assertNever } from "assert-never";
+import { logger } from "~/services/logger.server";
+
+const SUCCESSFUL_STATUSES = [TaskRunStatus.COMPLETED_SUCCESSFULLY];
+const FAILURE_STATUSES = [
+  TaskRunStatus.CANCELED,
+  TaskRunStatus.INTERRUPTED,
+  TaskRunStatus.COMPLETED_WITH_ERRORS,
+  TaskRunStatus.SYSTEM_FAILURE,
+  TaskRunStatus.CRASHED,
+];
+
+export type TaskRunWithAttempts = TaskRun & {
+  attempts: TaskRunAttempt[];
+};
+
+export function executionResultForTaskRun(
+  taskRun: TaskRunWithAttempts
+): TaskRunExecutionResult | undefined {
+  if (SUCCESSFUL_STATUSES.includes(taskRun.status)) {
+    // find the last attempt that was successful
+    const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.COMPLETED);
+
+    if (!attempt) {
+      logger.error("Task run is successful but no successful attempt found", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+      });
+
+      return undefined;
+    }
+
+    return {
+      ok: true,
+      id: taskRun.friendlyId,
+      output: attempt.output ?? undefined,
+      outputType: attempt.outputType,
+    } satisfies TaskRunSuccessfulExecutionResult;
+  }
+
+  if (FAILURE_STATUSES.includes(taskRun.status)) {
+    if (taskRun.status === TaskRunStatus.CANCELED) {
+      return {
+        ok: false,
+        id: taskRun.friendlyId,
+        error: {
+          type: "INTERNAL_ERROR",
+          code: "TASK_RUN_CANCELLED",
+        },
+      } satisfies TaskRunFailedExecutionResult;
+    }
+
+    const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.FAILED);
+
+    if (!attempt) {
+      logger.error("Task run is failed but no failed attempt found", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+      });
+
+      return undefined;
+    }
+
+    const error = TaskRunError.safeParse(attempt.error);
+
+    if (!error.success) {
+      logger.error("Failed to parse error from failed task run attempt", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+        error: attempt.error,
+      });
+
+      return {
+        ok: false,
+        id: taskRun.friendlyId,
+        error: {
+          type: "INTERNAL_ERROR",
+          code: "CONFIGURED_INCORRECTLY",
+        },
+      } satisfies TaskRunFailedExecutionResult;
+    }
+
+    return {
+      ok: false,
+      id: taskRun.friendlyId,
+      error: error.data,
+    } satisfies TaskRunFailedExecutionResult;
+  }
+}
+
+export function batchTaskRunItemStatusForRunStatus(status: TaskRunStatus): BatchTaskRunItemStatus {
+  switch (status) {
+    case TaskRunStatus.COMPLETED_SUCCESSFULLY:
+      return BatchTaskRunItemStatus.COMPLETED;
+    case TaskRunStatus.CANCELED:
+    case TaskRunStatus.INTERRUPTED:
+    case TaskRunStatus.COMPLETED_WITH_ERRORS:
+    case TaskRunStatus.SYSTEM_FAILURE:
+    case TaskRunStatus.CRASHED:
+    case TaskRunStatus.COMPLETED_WITH_ERRORS:
+      return BatchTaskRunItemStatus.FAILED;
+    case TaskRunStatus.PENDING:
+    case TaskRunStatus.WAITING_FOR_DEPLOY:
+    case TaskRunStatus.WAITING_TO_RESUME:
+    case TaskRunStatus.RETRYING_AFTER_FAILURE:
+    case TaskRunStatus.EXECUTING:
+    case TaskRunStatus.PAUSED:
+      return BatchTaskRunItemStatus.PENDING;
+    default:
+      assertNever(status);
+  }
+}

apps/webapp/app/presenters/JobListPresenter.server.ts

@@ -3,7 +3,7 @@ import {
   DisplayPropertySchema,
   EventSpecificationSchema,
 } from "@trigger.dev/core";
-import { PrismaClient, Prisma, prisma } from "~/db.server";
+import { PrismaClient, Prisma, prisma, sqlDatabaseSchema } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
@@ -122,7 +122,7 @@ export class JobListPresenter {
               "jobId",
               ROW_NUMBER() OVER(PARTITION BY "jobId" ORDER BY "createdAt" DESC) as rn
           FROM 
-              "JobRun" 
+              ${sqlDatabaseSchema}."JobRun" 
           WHERE 
               "jobId" IN (${Prisma.join(jobs.map((j) => j.id))})
       ) t

apps/webapp/app/presenters/OrgUsagePresenter.server.ts

@@ -1,5 +1,5 @@
 import { estimate } from "@trigger.dev/billing";
-import { PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { featuresForRequest } from "~/features.server";
 import { BillingService } from "~/services/billing.server";
 
@@ -53,7 +53,7 @@ export class OrgUsagePresenter {
         month: string;
         count: number;
       }[]
-    >`SELECT TO_CHAR("createdAt", 'YYYY-MM') as month, COUNT(*) as count FROM "JobRun" WHERE "organizationId" = ${organization.id} AND "createdAt" >= NOW() - INTERVAL '6 months' AND "internal" = FALSE GROUP BY month ORDER BY month ASC`;
+    >`SELECT TO_CHAR("createdAt", 'YYYY-MM') as month, COUNT(*) as count FROM ${sqlDatabaseSchema}."JobRun" WHERE "organizationId" = ${organization.id} AND "createdAt" >= NOW() - INTERVAL '6 months' AND "internal" = FALSE GROUP BY month ORDER BY month ASC`;
 
     const hasMonthlyRunData = monthlyRunsDataRaw.length > 0;
     const monthlyRunsData = monthlyRunsDataRaw.map((obj) => ({
@@ -117,7 +117,7 @@ export class OrgUsagePresenter {
 
     const dailyRunsRawData = await this.#prismaClient.$queryRaw<
       { day: Date; runs: BigInt }[]
-    >`SELECT date_trunc('day', "createdAt") as day, COUNT(*) as runs FROM "JobRun" WHERE "organizationId" = ${organization.id} AND "createdAt" >= NOW() - INTERVAL '30 days' AND "internal" = FALSE GROUP BY day`;
+    >`SELECT date_trunc('day', "createdAt") as day, COUNT(*) as runs FROM ${sqlDatabaseSchema}."JobRun" WHERE "organizationId" = ${organization.id} AND "createdAt" >= NOW() - INTERVAL '30 days' AND "internal" = FALSE GROUP BY day`;
 
     const hasDailyRunsData = dailyRunsRawData.length > 0;
     const dailyRunsDataFilledIn = fillInMissingDailyRuns(ThirtyDaysAgo, 31, dailyRunsRawData);

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

@@ -0,0 +1,46 @@
+import { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
+import { executionResultForTaskRun } from "~/models/taskRun.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { BasePresenter } from "./basePresenter.server";
+
+export class ApiBatchResultsPresenter extends BasePresenter {
+  public async call(
+    friendlyId: string,
+    env: AuthenticatedEnvironment
+  ): Promise<BatchTaskRunExecutionResult | undefined> {
+    return this.traceWithEnv("call", env, async (span) => {
+      const batchRun = await this._prisma.batchTaskRun.findUnique({
+        where: {
+          friendlyId,
+          runtimeEnvironmentId: env.id,
+        },
+        include: {
+          items: {
+            include: {
+              taskRun: {
+                include: {
+                  attempts: {
+                    orderBy: {
+                      createdAt: "desc",
+                    },
+                  },
+                },
+              },
+            },
+          },
+        },
+      });
+
+      if (!batchRun) {
+        return undefined;
+      }
+
+      return {
+        id: batchRun.friendlyId,
+        items: batchRun.items
+          .map((item) => executionResultForTaskRun(item.taskRun))
+          .filter(Boolean),
+      };
+    });
+  }
+}

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

@@ -0,0 +1,33 @@
+import { TaskRunExecutionResult } from "@trigger.dev/core/v3";
+import { executionResultForTaskRun } from "~/models/taskRun.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { BasePresenter } from "./basePresenter.server";
+
+export class ApiRunResultPresenter extends BasePresenter {
+  public async call(
+    friendlyId: string,
+    env: AuthenticatedEnvironment
+  ): Promise<TaskRunExecutionResult | undefined> {
+    return this.traceWithEnv("call", env, async (span) => {
+      const taskRun = await this._prisma.taskRun.findUnique({
+        where: {
+          friendlyId,
+          runtimeEnvironmentId: env.id,
+        },
+        include: {
+          attempts: {
+            orderBy: {
+              createdAt: "desc",
+            },
+          },
+        },
+      });
+
+      if (!taskRun) {
+        return undefined;
+      }
+
+      return executionResultForTaskRun(taskRun);
+    });
+  }
+}

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

@@ -1,5 +1,5 @@
 import { WorkerDeploymentStatus } from "@trigger.dev/database";
-import { PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
@@ -97,7 +97,7 @@ export class DeploymentListPresenter {
   wd."id", 
   wd."shortCode", 
   wd."version", 
-  (SELECT COUNT(*) FROM "BackgroundWorkerTask" WHERE "BackgroundWorkerTask"."workerId" = wd."workerId") AS "tasksCount", 
+  (SELECT COUNT(*) FROM ${sqlDatabaseSchema}."BackgroundWorkerTask" WHERE "BackgroundWorkerTask"."workerId" = wd."workerId") AS "tasksCount", 
   wd."environmentId", 
   wd."status", 
   u."id" AS "userId", 
@@ -106,9 +106,9 @@ export class DeploymentListPresenter {
   u."avatarUrl" AS "userAvatarUrl", 
   wd."deployedAt"
 FROM 
-  "WorkerDeployment" as wd
+  ${sqlDatabaseSchema}."WorkerDeployment" as wd
 INNER JOIN 
-  "User" as u ON wd."triggeredById" = u."id" 
+  ${sqlDatabaseSchema}."User" as u ON wd."triggeredById" = u."id" 
 WHERE 
   wd."projectId" = ${project.id}
 ORDER BY 

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

@@ -58,12 +58,13 @@ export class EditSchedulePresenter {
       },
     });
 
-    const possibleTasks = await this.#prismaClient.$queryRaw<{ slug: string }[]>`
-    SELECT DISTINCT(slug)
-    FROM "BackgroundWorkerTask"
-    WHERE "projectId" = ${project.id} 
-    AND "triggerSource" = 'SCHEDULED';
-    `;
+    const possibleTasks = await this.#prismaClient.backgroundWorkerTask.findMany({
+      distinct: ["slug"],
+      where: {
+        projectId: project.id,
+        triggerSource: "SCHEDULED",
+      },
+    });
 
     const possibleEnvironments = project.environments.map((environment) => {
       let userName: undefined | string;

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

@@ -1,6 +1,6 @@
 import { Prisma, TaskRunStatus } from "@trigger.dev/database";
 import { Direction } from "~/components/runs/RunStatuses";
-import { PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { getUsername } from "~/utils/username";
 import { CANCELLABLE_STATUSES } from "~/v3/services/cancelTaskRun.server";
 
@@ -85,11 +85,12 @@ export class RunListPresenter {
     });
 
     //get all possible tasks
-    const possibleTasks = await this.#prismaClient.$queryRaw<{ slug: string }[]>`
-    SELECT DISTINCT(slug)
-    FROM "BackgroundWorkerTask"
-    WHERE "projectId" = ${project.id};
-    `;
+    const possibleTasks = await this.#prismaClient.backgroundWorkerTask.findMany({
+      distinct: ["slug"],
+      where: {
+        projectId: project.id,
+      },
+    });
 
     //get the runs
     let runs = await this.#prismaClient.$queryRaw<
@@ -105,6 +106,7 @@ export class RunListPresenter {
         lockedAt: Date | null;
         completedAt: Date | null;
         isTest: boolean;
+        spanId: string;
         attempts: BigInt;
       }[]
     >`
@@ -120,17 +122,18 @@ export class RunListPresenter {
     tr."lockedAt" AS "lockedAt",
     tra."completedAt" AS "completedAt",
     tr."isTest" AS "isTest",
+    tr."spanId" AS "spanId",
     COUNT(tra.id) AS attempts
   FROM
-    "TaskRun" tr
+    ${sqlDatabaseSchema}."TaskRun" tr
   LEFT JOIN
     (
       SELECT *,
         ROW_NUMBER() OVER (PARTITION BY "taskRunId" ORDER BY "createdAt" DESC) rn
-      FROM "TaskRunAttempt"
+      FROM ${sqlDatabaseSchema}."TaskRunAttempt"
     ) tra ON tr.id = tra."taskRunId" AND tra.rn = 1
   LEFT JOIN
-    "BackgroundWorker" bw ON tra."backgroundWorkerId" = bw.id
+    ${sqlDatabaseSchema}."BackgroundWorker" bw ON tra."backgroundWorkerId" = bw.id
   WHERE
       -- project
       tr."projectId" = ${project.id}
@@ -224,6 +227,7 @@ export class RunListPresenter {
           status: run.status,
           version: run.version,
           taskIdentifier: run.taskIdentifier,
+          spanId: run.spanId,
           attempts: Number(run.attempts),
           isReplayable: true,
           isCancellable: CANCELLABLE_STATUSES.includes(run.status),

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

@@ -1,6 +1,6 @@
 import { Prisma, RuntimeEnvironmentType } from "@trigger.dev/database";
 import { ScheduleListFilters } from "~/components/runs/v3/ScheduleFilters";
-import { PrismaClient, prisma } from "~/db.server";
+import { PrismaClient, prisma, sqlDatabaseSchema } from "~/db.server";
 import { getUsername } from "~/utils/username";
 import { calculateNextScheduledTimestamp } from "~/v3/utils/calculateNextSchedule.server";
 
@@ -81,12 +81,13 @@ export class ScheduleListPresenter {
     });
 
     //get all possible scheduled tasks
-    const possibleTasks = await this.#prismaClient.$queryRaw<{ slug: string }[]>`
-    SELECT DISTINCT(slug)
-    FROM "BackgroundWorkerTask"
-    WHERE "projectId" = ${project.id}
-    AND "triggerSource" = 'SCHEDULED';
-    `;
+    const possibleTasks = await this.#prismaClient.backgroundWorkerTask.findMany({
+      distinct: ["slug"],
+      where: {
+        projectId: project.id,
+        triggerSource: "SCHEDULED",
+      },
+    });
 
     //do this here to protect against SQL injection
     search = search && search !== "" ? `%${search}%` : undefined;
@@ -201,11 +202,11 @@ export class ScheduleListPresenter {
     SELECT t."scheduleId", t."createdAt"
     FROM (
       SELECT "scheduleId", MAX("createdAt") as "LatestRun"
-      FROM "TaskRun"
+      FROM ${sqlDatabaseSchema}."TaskRun"
       WHERE "scheduleId" IN (${Prisma.join(rawSchedules.map((s) => s.id))})
       GROUP BY "scheduleId"
     ) r
-    JOIN "TaskRun" t
+    JOIN ${sqlDatabaseSchema}."TaskRun" t
     ON t."scheduleId" = r."scheduleId" AND t."createdAt" = r."LatestRun";`
         : [];
 

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

@@ -1,5 +1,5 @@
 import { Prisma, TaskRunStatus, TaskTriggerSource } from "@trigger.dev/database";
-import { PrismaClient, prisma } from "~/db.server";
+import { PrismaClient, prisma, sqlDatabaseSchema } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
@@ -73,7 +73,7 @@ export class TaskListPresenter {
       bwt."createdAt",
       bwt."triggerSource"
     FROM
-      "BackgroundWorkerTask" as bwt
+      ${sqlDatabaseSchema}."BackgroundWorkerTask" as bwt
     WHERE bwt."projectId" = ${project.id}
     ORDER BY
       bwt.slug,
@@ -101,7 +101,7 @@ export class TaskListPresenter {
           "lockedById",
           ROW_NUMBER() OVER (PARTITION BY "lockedById" ORDER BY "updatedAt" DESC) AS rn
         FROM
-          "TaskRun"
+          ${sqlDatabaseSchema}."TaskRun"
         WHERE
           "lockedById" IN(${Prisma.join(tasks.map((t) => t.id))})
             ) t

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

@@ -1,5 +1,5 @@
 import { TaskTriggerSource } from "@trigger.dev/database";
-import { PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { TestSearchParams } from "~/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test/route";
 import { sortEnvironments } from "~/services/environmentSort.server";
 import { createSearchParams } from "~/utils/searchParams";
@@ -97,14 +97,14 @@ export class TestPresenter {
             bw.*,
             ROW_NUMBER() OVER(ORDER BY string_to_array(bw.version, '.')::int[] DESC) AS rn
       FROM 
-            "BackgroundWorker" bw
+            ${sqlDatabaseSchema}."BackgroundWorker" bw
       WHERE "runtimeEnvironmentId" = ${matchingEnvironment.id}
     ),
     latest_workers AS (SELECT * FROM workers WHERE rn = 1)
-    SELECT "BackgroundWorkerTask".id, version, slug as "taskIdentifier", "filePath", "exportName", "BackgroundWorkerTask"."friendlyId", "BackgroundWorkerTask"."triggerSource"
+    SELECT bwt.id, version, slug as "taskIdentifier", "filePath", "exportName", bwt."friendlyId" 
     FROM latest_workers
-    JOIN "BackgroundWorkerTask" ON "BackgroundWorkerTask"."workerId" = latest_workers.id
-    ORDER BY "BackgroundWorkerTask"."exportName" ASC;
+    JOIN ${sqlDatabaseSchema}."BackgroundWorkerTask" bwt ON bwt."workerId" = latest_workers.id
+    ORDER BY bwt."exportName" ASC;
     `;
 
     return {

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

@@ -5,7 +5,7 @@ import {
   TaskRunStatus,
   TaskTriggerSource,
 } from "@trigger.dev/database";
-import { PrismaClient, prisma } from "~/db.server";
+import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
 import { getUsername } from "~/utils/username";
 
 type TestTaskOptions = {
@@ -107,9 +107,9 @@ export class TestTaskPresenter {
       SELECT 
           tr.* 
       FROM 
-          "TaskRun" as tr
+          ${sqlDatabaseSchema}."TaskRun" as tr
       JOIN
-          "BackgroundWorkerTask" as bwt
+          ${sqlDatabaseSchema}."BackgroundWorkerTask" as bwt
       ON
           tr."taskIdentifier" = bwt.slug
       WHERE

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

@@ -0,0 +1,34 @@
+import { Span, SpanKind } from "@opentelemetry/api";
+import { PrismaClientOrTransaction, prisma } from "~/db.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { attributesFromAuthenticatedEnv, tracer } from "../../v3/tracer.server";
+
+export abstract class BasePresenter {
+  constructor(protected readonly _prisma: PrismaClientOrTransaction = prisma) {}
+
+  protected async traceWithEnv<T>(
+    trace: string,
+    env: AuthenticatedEnvironment,
+    fn: (span: Span) => Promise<T>
+  ): Promise<T> {
+    return tracer.startActiveSpan(
+      `${this.constructor.name}.${trace}`,
+      { attributes: attributesFromAuthenticatedEnv(env), kind: SpanKind.SERVER },
+      async (span) => {
+        try {
+          return await fn(span);
+        } catch (e) {
+          if (e instanceof Error) {
+            span.recordException(e);
+          } else {
+            span.recordException(new Error(String(e)));
+          }
+
+          throw e;
+        } finally {
+          span.end();
+        }
+      }
+    );
+  }
+}

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam._index/route.tsx

@@ -60,7 +60,7 @@ export default function Page() {
   const user = useUser();
 
   return (
-    <>
+    <div className="overflow-y-auto scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
       {job.hasIntegrationsRequiringAction && (
         <Callout variant="error" to={organizationIntegrationsPath(organization)} className="mb-2">
           {simplur`This Job has ${
@@ -96,6 +96,6 @@ export default function Page() {
           </div>
         )}
       </Help>
-    </>
+    </div>
   );
 }

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam/route.tsx

@@ -1,10 +1,13 @@
 import {
+  ArrowsPointingInIcon,
+  ArrowsPointingOutIcon,
   ChevronDownIcon,
   ChevronRightIcon,
   MagnifyingGlassMinusIcon,
   MagnifyingGlassPlusIcon,
 } from "@heroicons/react/20/solid";
-import { Outlet, useNavigate, useParams, useRevalidator } from "@remix-run/react";
+import type { Location } from "@remix-run/react";
+import { useParams, useRevalidator } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { Virtualizer } from "@tanstack/react-virtual";
 import {
@@ -23,7 +26,7 @@ import { InlineCode } from "~/components/code/InlineCode";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { MainCenteredContainer, PageBody } from "~/components/layout/AppLayout";
 import { Badge } from "~/components/primitives/Badge";
-import { LinkButton } from "~/components/primitives/Buttons";
+import { Button, LinkButton } from "~/components/primitives/Buttons";
 import { Callout } from "~/components/primitives/Callout";
 import { Input } from "~/components/primitives/Input";
 import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
@@ -33,6 +36,7 @@ import {
   ResizablePanel,
   ResizablePanelGroup,
 } from "~/components/primitives/Resizable";
+import { ShortcutKey, variants } from "~/components/primitives/ShortcutKey";
 import { Slider } from "~/components/primitives/Slider";
 import { Switch } from "~/components/primitives/Switch";
 import * as Timeline from "~/components/primitives/Timeline";
@@ -45,8 +49,9 @@ import { useDebounce } from "~/hooks/useDebounce";
 import { useEventSource } from "~/hooks/useEventSource";
 import { useInitialDimensions } from "~/hooks/useInitialDimensions";
 import { useOrganization } from "~/hooks/useOrganizations";
-import { usePathName } from "~/hooks/usePathName";
 import { useProject } from "~/hooks/useProject";
+import { useReplaceLocation } from "~/hooks/useReplaceLocation";
+import { Shortcut, useShortcutKeys } from "~/hooks/useShortcutKeys";
 import { useUser } from "~/hooks/useUser";
 import { RunEvent, RunPresenter } from "~/presenters/v3/RunPresenter.server";
 import { getResizableRunSettings, setResizableRunSettings } from "~/services/resizablePanel";
@@ -60,6 +65,11 @@ import {
   v3RunStreamingPath,
   v3RunsPath,
 } from "~/utils/pathBuilder";
+import { SpanView } from "../resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route";
+import { number } from "zod";
+import { useHotkeys } from "react-hotkeys-hook";
+import { Popover, PopoverArrowTrigger, PopoverContent } from "~/components/primitives/Popover";
+import { Header3 } from "~/components/primitives/Headers";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
@@ -82,19 +92,18 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   });
 };
 
-function getSpanId(path: string): string | undefined {
-  const regex = /spans\/([^\/]*)/;
-  const match = path.match(regex);
-  return match ? match[1] : undefined;
+function getSpanId(location: Location<any>): string | undefined {
+  const search = new URLSearchParams(location.search);
+  return search.get("span") ?? undefined;
 }
 
 export default function Page() {
   const { run, trace, resizeSettings } = useTypedLoaderData<typeof loader>();
-  const navigate = useNavigate();
   const organization = useOrganization();
-  const pathName = usePathName();
   const project = useProject();
   const user = useUser();
+  const { location, replaceSearchParam } = useReplaceLocation();
+  const selectedSpanId = getSpanId(location);
 
   const usernameForEnv = user.id !== run.environment.userId ? run.environment.userName : undefined;
 
@@ -133,10 +142,8 @@ export default function Page() {
 
   const { events, parentRunFriendlyId, duration, rootSpanStatus, rootStartedAt } = trace;
 
-  const selectedSpanId = getSpanId(pathName);
-
   const changeToSpan = useDebounce((selectedSpan: string) => {
-    navigate(v3RunSpanPath(organization, project, run, { spanId: selectedSpan }));
+    replaceSearchParam("span", selectedSpan);
   }, 250);
 
   const revalidator = useRevalidator();
@@ -166,62 +173,47 @@ export default function Page() {
       </NavBar>
       <PageBody scrollable={false}>
         <div className={cn("grid h-full max-h-full grid-cols-1 overflow-hidden")}>
-          {selectedSpanId === undefined ? (
-            <TasksTreeView
-              selectedId={selectedSpanId}
-              key={events[0]?.id ?? "-"}
-              events={events}
-              parentRunFriendlyId={parentRunFriendlyId}
-              onSelectedIdChanged={(selectedSpan) => {
-                //instantly close the panel if no span is selected
-                if (!selectedSpan) {
-                  navigate(v3RunPath(organization, project, run));
-                  return;
-                }
+          <ResizablePanelGroup
+            direction="horizontal"
+            className="h-full max-h-full"
+            onLayout={(layout) => {
+              if (layout.length !== 2) return;
+              if (!selectedSpanId) return;
+              setResizableRunSettings(document, layout);
+            }}
+          >
+            <ResizablePanel order={1} minSize={30} defaultSize={resizeSettings.layout?.[0]}>
+              <TasksTreeView
+                selectedId={selectedSpanId}
+                key={events[0]?.id ?? "-"}
+                events={events}
+                parentRunFriendlyId={parentRunFriendlyId}
+                onSelectedIdChanged={(selectedSpan) => {
+                  //instantly close the panel if no span is selected
+                  if (!selectedSpan) {
+                    replaceSearchParam("span");
+                    return;
+                  }
 
-                changeToSpan(selectedSpan);
-              }}
-              totalDuration={duration}
-              rootSpanStatus={rootSpanStatus}
-              rootStartedAt={rootStartedAt}
-              environmentType={run.environment.type}
-            />
-          ) : (
-            <ResizablePanelGroup
-              direction="horizontal"
-              className="h-full max-h-full"
-              onLayout={(layout) => {
-                if (layout.length !== 2) return;
-                setResizableRunSettings(document, layout);
-              }}
-            >
-              <ResizablePanel order={1} minSize={30} defaultSize={resizeSettings.layout?.[0]}>
-                <TasksTreeView
-                  selectedId={selectedSpanId}
-                  key={events[0]?.id ?? "-"}
-                  events={events}
-                  parentRunFriendlyId={parentRunFriendlyId}
-                  onSelectedIdChanged={(selectedSpan) => {
-                    //instantly close the panel if no span is selected
-                    if (!selectedSpan) {
-                      navigate(v3RunPath(organization, project, run));
-                      return;
-                    }
-
-                    changeToSpan(selectedSpan);
-                  }}
-                  totalDuration={duration}
-                  rootSpanStatus={rootSpanStatus}
-                  rootStartedAt={rootStartedAt}
-                  environmentType={run.environment.type}
+                  changeToSpan(selectedSpan);
+                }}
+                totalDuration={duration}
+                rootSpanStatus={rootSpanStatus}
+                rootStartedAt={rootStartedAt}
+                environmentType={run.environment.type}
+              />
+            </ResizablePanel>
+            <ResizableHandle withHandle />
+            {selectedSpanId && (
+              <ResizablePanel order={2} minSize={30} defaultSize={resizeSettings.layout?.[1]}>
+                <SpanView
+                  runParam={run.friendlyId}
+                  spanId={selectedSpanId}
+                  closePanel={() => replaceSearchParam("span")}
                 />
               </ResizablePanel>
-              <ResizableHandle withHandle />
-              <ResizablePanel order={2} minSize={30} defaultSize={resizeSettings.layout?.[1]}>
-                <Outlet key={selectedSpanId} />
-              </ResizablePanel>
-            </ResizablePanelGroup>
-          )}
+            )}
+          </ResizablePanelGroup>
         </div>
       </PageBody>
     </>
@@ -263,6 +255,9 @@ function TasksTreeView({
     getNodeProps,
     toggleNodeSelection,
     toggleExpandNode,
+    expandAllBelowDepth,
+    toggleExpandLevel,
+    collapseAllBelowDepth,
     selectNode,
     scrollToNode,
     virtualizer,
@@ -286,7 +281,7 @@ function TasksTreeView({
   });
 
   return (
-    <div className="grid h-full grid-rows-[2.5rem_1fr] overflow-hidden">
+    <div className="grid h-full grid-rows-[2.5rem_1fr_3.25rem] overflow-hidden">
       <div className="mx-3 flex items-center justify-between gap-2 border-b border-grid-dimmed">
         <Input
           placeholder="Search log"
@@ -297,30 +292,12 @@ function TasksTreeView({
           onChange={(e) => setFilterText(e.target.value)}
         />
         <div className="flex items-center gap-2">
-          <LiveReloadingStatus rootSpanCompleted={rootSpanStatus !== "executing"} />
           <Switch
             variant="small"
             label="Errors only"
             checked={errorsOnly}
             onCheckedChange={(e) => setErrorsOnly(e.valueOf())}
           />
-          <Switch
-            variant="small"
-            label="Show durations"
-            checked={showDurations}
-            onCheckedChange={(e) => setShowDurations(e.valueOf())}
-          />
-          <Slider
-            variant={"tertiary"}
-            className="w-20"
-            LeadingIcon={MagnifyingGlassMinusIcon}
-            TrailingIcon={MagnifyingGlassPlusIcon}
-            value={[scale]}
-            onValueChange={(value) => setScale(value[0])}
-            min={0}
-            max={1}
-            step={0.05}
-          />
         </div>
       </div>
       <ResizablePanelGroup
@@ -333,14 +310,15 @@ function TasksTreeView({
         {/* Tree list */}
         <ResizablePanel order={1} minSize={20} defaultSize={50} className="pl-3">
           <div className="grid h-full grid-rows-[2rem_1fr] overflow-hidden">
-            <div className="flex items-center">
+            <div className="flex items-center pr-2">
               {parentRunFriendlyId ? (
                 <ShowParentLink runFriendlyId={parentRunFriendlyId} />
               ) : (
-                <Paragraph variant="small" className="text-charcoal-500">
+                <Paragraph variant="small" className="flex-1 text-charcoal-500">
                   This is the root task
                 </Paragraph>
               )}
+              <LiveReloadingStatus rootSpanCompleted={rootSpanStatus !== "executing"} />
             </div>
             <TreeView
               parentRef={parentRef}
@@ -355,13 +333,13 @@ function TasksTreeView({
                 <>
                   <div
                     className={cn(
-                      "delay-[25ms] flex h-8 cursor-pointer items-center overflow-hidden rounded-l-sm pr-2 transition-colors",
+                      "flex h-8 cursor-pointer items-center overflow-hidden rounded-l-sm pr-2",
                       state.selected
                         ? "bg-grid-dimmed hover:bg-grid-bright"
                         : "bg-transparent hover:bg-grid-dimmed"
                     )}
                     onClick={() => {
-                      toggleNodeSelection(node.id);
+                      selectNode(node.id);
                     }}
                   >
                     <div className="flex h-8 items-center">
@@ -379,7 +357,15 @@ function TasksTreeView({
                         )}
                         onClick={(e) => {
                           e.stopPropagation();
-                          toggleExpandNode(node.id);
+                          if (e.altKey) {
+                            if (state.expanded) {
+                              collapseAllBelowDepth(node.level);
+                            } else {
+                              expandAllBelowDepth(node.level);
+                            }
+                          } else {
+                            toggleExpandNode(node.id);
+                          }
                           scrollToNode(node.id);
                         }}
                       >
@@ -445,6 +431,50 @@ function TasksTreeView({
           />
         </ResizablePanel>
       </ResizablePanelGroup>
+      <div className="flex items-center justify-between gap-2 border-t border-grid-dimmed px-2">
+        <div className="grow @container">
+          <div className="hidden items-center gap-4 @[42rem]:flex">
+            <KeyboardShortcuts
+              expandAllBelowDepth={expandAllBelowDepth}
+              collapseAllBelowDepth={collapseAllBelowDepth}
+              toggleExpandLevel={toggleExpandLevel}
+              setShowDurations={setShowDurations}
+            />
+          </div>
+          <div className="@[42rem]:hidden">
+            <Popover>
+              <PopoverArrowTrigger>Shortcuts</PopoverArrowTrigger>
+              <PopoverContent
+                className="min-w-[20rem] overflow-y-auto p-2 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600"
+                align="start"
+              >
+                <Header3 spacing>Keyboard shortcuts</Header3>
+                <div className="flex flex-col gap-2">
+                  <KeyboardShortcuts
+                    expandAllBelowDepth={expandAllBelowDepth}
+                    collapseAllBelowDepth={collapseAllBelowDepth}
+                    toggleExpandLevel={toggleExpandLevel}
+                    setShowDurations={setShowDurations}
+                  />
+                </div>
+              </PopoverContent>
+            </Popover>
+          </div>
+        </div>
+        <div className="flex items-center gap-4">
+          <Slider
+            variant={"tertiary"}
+            className="w-20"
+            LeadingIcon={MagnifyingGlassMinusIcon}
+            TrailingIcon={MagnifyingGlassPlusIcon}
+            value={[scale]}
+            onValueChange={(value) => setScale(value[0])}
+            min={0}
+            max={1}
+            step={0.05}
+          />
+        </div>
+      </div>
     </div>
   );
 }
@@ -738,6 +768,7 @@ function ShowParentLink({ runFriendlyId }: { runFriendlyId: string }) {
       fullWidth
       textAlignLeft
       shortcut={{ key: "p" }}
+      className="flex-1"
     >
       {mouseOver ? (
         <ShowParentIconSelected className="h-4 w-4 text-indigo-500" />
@@ -884,3 +915,92 @@ function ConnectedDevWarning() {
     </div>
   );
 }
+
+function KeyboardShortcuts({
+  expandAllBelowDepth,
+  collapseAllBelowDepth,
+  toggleExpandLevel,
+  setShowDurations,
+}: {
+  expandAllBelowDepth: (depth: number) => void;
+  collapseAllBelowDepth: (depth: number) => void;
+  toggleExpandLevel: (depth: number) => void;
+  setShowDurations: (show: (show: boolean) => boolean) => void;
+}) {
+  return (
+    <>
+      <ArrowKeyShortcuts />
+      <ShortcutWithAction
+        shortcut={{ key: "e" }}
+        action={() => expandAllBelowDepth(0)}
+        title="Expand all"
+      />
+      <ShortcutWithAction
+        shortcut={{ key: "c" }}
+        action={() => collapseAllBelowDepth(1)}
+        title="Collapse all"
+      />
+      <NumberShortcuts toggleLevel={(number) => toggleExpandLevel(number)} />
+      <ShortcutWithAction
+        shortcut={{ key: "d" }}
+        action={() => setShowDurations((d) => !d)}
+        title="Toggle durations"
+      />
+    </>
+  );
+}
+
+function ArrowKeyShortcuts() {
+  return (
+    <div className="flex items-center gap-0.5">
+      <ShortcutKey shortcut={{ key: "arrowup" }} variant="medium" className="ml-0 mr-0" />
+      <ShortcutKey shortcut={{ key: "arrowdown" }} variant="medium" className="ml-0 mr-0" />
+      <ShortcutKey shortcut={{ key: "arrowleft" }} variant="medium" className="ml-0 mr-0" />
+      <ShortcutKey shortcut={{ key: "arrowright" }} variant="medium" className="ml-0 mr-0" />
+      <Paragraph variant="extra-small" className="ml-1.5 whitespace-nowrap">
+        Navigate
+      </Paragraph>
+    </div>
+  );
+}
+
+function ShortcutWithAction({
+  shortcut,
+  title,
+  action,
+}: {
+  shortcut: Shortcut;
+  title: string;
+  action: () => void;
+}) {
+  useShortcutKeys({
+    shortcut,
+    action,
+  });
+
+  return (
+    <div className="flex items-center gap-0.5">
+      <ShortcutKey shortcut={shortcut} variant="medium" className="ml-0 mr-0" />
+      <Paragraph variant="extra-small" className="ml-1.5 whitespace-nowrap">
+        {title}
+      </Paragraph>
+    </div>
+  );
+}
+
+function NumberShortcuts({ toggleLevel }: { toggleLevel: (depth: number) => void }) {
+  useHotkeys(["1", "2", "3", "4", "5", "6", "7", "8", "9", "0"], (event, hotkeysEvent) => {
+    toggleLevel(Number(event.key));
+  });
+
+  return (
+    <div className="flex items-center gap-0.5">
+      <span className={cn(variants.medium, "ml-0 mr-0")}>0</span>
+      <span className="text-[0.75rem] text-text-dimmed">–</span>
+      <span className={cn(variants.medium, "ml-0 mr-0")}>9</span>
+      <Paragraph variant="extra-small" className="ml-1.5 whitespace-nowrap">
+        Toggle level
+      </Paragraph>
+    </div>
+  );
+}

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test.tasks.$taskParam/route.tsx

@@ -6,7 +6,6 @@ import { ActionFunction, LoaderFunctionArgs, json } from "@remix-run/server-runt
 import { TaskRunStatus } from "@trigger.dev/database";
 import { useCallback, useEffect, useRef, useState } from "react";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
-import { z } from "zod";
 import { JSONEditor } from "~/components/code/JSONEditor";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { Button } from "~/components/primitives/Buttons";
@@ -37,7 +36,7 @@ import {
   TestTaskPresenter,
 } from "~/presenters/v3/TestTaskPresenter.server";
 import { requireUserId } from "~/services/session.server";
-import { docsPath, v3RunPath, v3TaskParamsSchema } from "~/utils/pathBuilder";
+import { docsPath, v3RunSpanPath, v3TaskParamsSchema } from "~/utils/pathBuilder";
 import { TestTaskService } from "~/v3/services/testTask.server";
 import { TestTaskData } from "~/v3/testTask";
 
@@ -77,7 +76,12 @@ export const action: ActionFunction = async ({ request, params }) => {
   }
 
   return redirectWithSuccessMessage(
-    v3RunPath({ slug: organizationSlug }, { slug: projectParam }, { friendlyId: run.friendlyId }),
+    v3RunSpanPath(
+      { slug: organizationSlug },
+      { slug: projectParam },
+      { friendlyId: run.friendlyId },
+      { spanId: run.spanId }
+    ),
     request,
     "Test run created"
   );

apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts

@@ -0,0 +1,45 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { ApiBatchResultsPresenter } from "~/presenters/v3/ApiBatchResultsPresenter.server";
+import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+
+const ParamsSchema = z.object({
+  /* This is the batch friendly ID */
+  batchParam: z.string(),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { batchParam } = parsed.data;
+
+  try {
+    const presenter = new ApiBatchResultsPresenter();
+    const result = await presenter.call(batchParam, authenticationResult.environment);
+
+    if (!result) {
+      return json({ error: "Batch not found" }, { status: 404 });
+    }
+
+    return json(result);
+  } catch (error) {
+    if (error instanceof Error) {
+      return json({ error: error.message }, { status: 500 });
+    } else {
+      return json({ error: JSON.stringify(error) }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts

@@ -1,12 +1,10 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { PrismaErrorSchema, prisma } from "~/db.server";
 import { z } from "zod";
+import { prisma } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
-import { CancelRunService } from "~/services/runs/cancelRun.server";
-import { ApiRunPresenter } from "~/presenters/ApiRunPresenter.server";
-import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
 import { logger } from "~/services/logger.server";
+import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
 
 const ParamsSchema = z.object({
   /* This is the run friendly ID */

apps/webapp/app/routes/api.v1.runs.$runParam.result.ts

@@ -0,0 +1,44 @@
+import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+
+const ParamsSchema = z.object({
+  /* This is the run friendly ID */
+  runParam: z.string(),
+});
+
+export async function loader({ request, params }: LoaderFunctionArgs) {
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { runParam } = parsed.data;
+
+  try {
+    const presenter = new ApiRunResultPresenter();
+    const result = await presenter.call(runParam, authenticationResult.environment);
+
+    if (!result) {
+      return json({ error: "Run either doesn't exist or is not finished" }, { status: 404 });
+    }
+
+    return json(result);
+  } catch (error) {
+    if (error instanceof Error) {
+      return json({ error: error.message }, { status: 500 });
+    } else {
+      return json({ error: JSON.stringify(error) }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/routes/api.v1.tasks.$taskId.batch.ts

@@ -1,6 +1,6 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { parseBatchTriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import { BatchTriggerTaskRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { MAX_BATCH_TRIGGER_ITEMS } from "~/consts";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
@@ -46,7 +46,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = parseBatchTriggerTaskRequestBody(anyBody);
+  const body = BatchTriggerTaskRequestBody.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -1,6 +1,6 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { parseTriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import { TriggerTaskRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
@@ -52,7 +52,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   // Now parse the request body
   const anyBody = await request.json();
 
-  const body = parseTriggerTaskRequestBody(anyBody);
+  const body = TriggerTaskRequestBody.safeParse(anyBody);
 
   if (!body.success) {
     return json({ error: "Invalid request body" }, { status: 400 });

apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts

@@ -2,6 +2,7 @@ import { LoaderFunctionArgs, redirect } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { prisma } from "~/db.server";
 import { requireUserId } from "~/services/session.server";
+import { v3RunSpanPath } from "~/utils/pathBuilder";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -33,8 +34,20 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
+  const run = await prisma.taskRun.findUnique({
+    where: {
+      friendlyId: validatedParams.runParam,
+    },
+  });
+
+  if (!run) {
+    throw new Response("Not found", { status: 404 });
+  }
+
   // Redirect to the project's runs page
   return redirect(
-    `/orgs/${project.organization.slug}/projects/v3/${project.slug}/runs/${validatedParams.runParam}`
+    v3RunSpanPath({ slug: project.organization.slug }, { slug: project.slug }, run, {
+      spanId: run.spanId,
+    })
   );
 }

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -4,10 +4,11 @@ import {
   QueueListIcon,
   StopCircleIcon,
 } from "@heroicons/react/20/solid";
-import { useParams } from "@remix-run/react";
+import { useFetcher, useParams } from "@remix-run/react";
 import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { formatDurationNanoseconds, nanosecondsToMilliseconds } from "@trigger.dev/core/v3";
-import { typedjson, useTypedLoaderData } from "remix-typedjson";
+import { useEffect } from "react";
+import { typedjson, useTypedFetcher, useTypedLoaderData } from "remix-typedjson";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { CodeBlock } from "~/components/code/CodeBlock";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
@@ -17,6 +18,7 @@ import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
 import { Header2 } from "~/components/primitives/Headers";
 import { Paragraph } from "~/components/primitives/Paragraph";
 import { Property, PropertyTable } from "~/components/primitives/PropertyTable";
+import { Spinner } from "~/components/primitives/Spinner";
 import { CancelRunDialog } from "~/components/runs/v3/CancelRunDialog";
 import { LiveTimer } from "~/components/runs/v3/LiveTimer";
 import { ReplayRunDialog } from "~/components/runs/v3/ReplayRunDialog";
@@ -58,19 +60,57 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   return typedjson({ span });
 };
 
-export default function Page() {
-  const {
-    span: { event },
-  } = useTypedLoaderData<typeof loader>();
+export function SpanView({
+  runParam,
+  spanId,
+  closePanel,
+}: {
+  runParam: string;
+  spanId: string | undefined;
+  closePanel: () => void;
+}) {
   const organization = useOrganization();
   const project = useProject();
-  const { runParam } = useParams();
+  const fetcher = useTypedFetcher<typeof loader>();
+
+  useEffect(() => {
+    if (spanId === undefined) return;
+    fetcher.load(
+      `/resources/orgs/${organization.slug}/projects/v3/${project.slug}/runs/${runParam}/spans/${spanId}`
+    );
+  }, [organization.slug, project.slug, runParam, spanId]);
+
+  if (spanId === undefined) {
+    return null;
+  }
+
+  if (fetcher.state !== "idle" || fetcher.data === undefined) {
+    return (
+      <div
+        className={cn(
+          "grid h-full max-h-full grid-rows-[2.5rem_1fr] overflow-hidden bg-background-bright"
+        )}
+      >
+        <div className="mx-3 flex items-center gap-2 overflow-x-hidden border-b border-grid-dimmed">
+          <div className="size-4 bg-grid-dimmed" />
+          <div className="h-6 w-[60%] bg-grid-dimmed" />
+        </div>
+        <div className="flex items-center justify-center">
+          <Spinner />
+        </div>
+      </div>
+    );
+  }
+
+  const {
+    span: { event },
+  } = fetcher.data;
 
   return (
     <div
       className={cn(
         "grid h-full max-h-full overflow-hidden bg-background-bright",
-        event.showActionBar ? "grid-rows-[2.5rem_1fr_2.5rem]" : "grid-rows-[2.5rem_1fr]"
+        event.showActionBar ? "grid-rows-[2.5rem_1fr_3.25rem]" : "grid-rows-[2.5rem_1fr]"
       )}
     >
       <div className="mx-3 flex items-center justify-between gap-2 overflow-x-hidden border-b border-grid-dimmed">
@@ -85,8 +125,8 @@ export default function Page() {
           </Header2>
         </div>
         {runParam && (
-          <LinkButton
-            to={v3RunPath(organization, project, { friendlyId: runParam })}
+          <Button
+            onClick={closePanel}
             variant="minimal/medium"
             LeadingIcon={ExitIcon}
             shortcut={{ key: "esc" }}
@@ -130,6 +170,9 @@ export default function Page() {
             )}
             <Property label="Message">{event.message}</Property>
             <Property label="Task ID">{event.taskSlug}</Property>
+            {event.idempotencyKey && (
+              <Property label="Idempotency key">{event.idempotencyKey}</Property>
+            )}
             {event.taskPath && event.taskExportName && (
               <Property label="Task">
                 <TaskPath
@@ -185,7 +228,7 @@ export default function Page() {
                   { friendlyId: event.runId },
                   { spanId: event.spanId }
                 )}
-                variant="minimal/small"
+                variant="minimal/medium"
                 LeadingIcon={QueueListIcon}
                 shortcut={{ key: "f" }}
               >
@@ -213,7 +256,7 @@ function RunActionButtons({ span }: { span: Span }) {
     return (
       <Dialog>
         <DialogTrigger asChild>
-          <Button variant="danger/small" LeadingIcon={StopCircleIcon}>
+          <Button variant="danger/medium" LeadingIcon={StopCircleIcon}>
             Cancel run
           </Button>
         </DialogTrigger>
@@ -233,7 +276,7 @@ function RunActionButtons({ span }: { span: Span }) {
   return (
     <Dialog>
       <DialogTrigger asChild>
-        <Button variant="tertiary/small" LeadingIcon={ArrowPathIcon}>
+        <Button variant="tertiary/medium" LeadingIcon={ArrowPathIcon}>
           Replay run
         </Button>
       </DialogTrigger>

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

@@ -4,7 +4,7 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
 import { logger } from "~/services/logger.server";
-import { v3RunPath } from "~/utils/pathBuilder";
+import { v3RunSpanPath } from "~/utils/pathBuilder";
 import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
 
 const FormSchema = z.object({
@@ -54,12 +54,13 @@ export const action: ActionFunction = async ({ request, params }) => {
       );
     }
 
-    const runPath = v3RunPath(
+    const runPath = v3RunSpanPath(
       {
         slug: taskRun.project.organization.slug,
       },
       { slug: taskRun.project.slug },
-      { friendlyId: newRun.friendlyId }
+      { friendlyId: newRun.friendlyId },
+      { spanId: newRun.spanId }
     );
 
     return redirectWithSuccessMessage(runPath, request, `Replaying run`);

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

@@ -134,20 +134,27 @@ export async function authenticatePersonalAccessToken(
 
   const hashedToken = hashToken(token);
 
-  const personalAccessToken = await prisma.personalAccessToken.update({
+  const personalAccessToken = await prisma.personalAccessToken.findFirst({
     where: {
       hashedToken,
       revokedAt: null,
     },
+  });
+
+  if (!personalAccessToken) {
+    // The token may have been revoked or is entirely invalid
+    return;
+  }
+
+  await prisma.personalAccessToken.update({
+    where: {
+      id: personalAccessToken.id,
+    },
     data: {
       lastAccessedAt: new Date(),
     },
   });
 
-  if (!personalAccessToken) {
-    return;
-  }
-
   const decryptedToken = decryptPersonalAccessToken(personalAccessToken);
 
   if (decryptedToken !== token) {
@@ -210,6 +217,18 @@ export async function createPersonalAccessTokenFromAuthorizationCode(
       },
     });
 
+    if (existingCliPersonalAccessToken.revokedAt) {
+      // re-activate revoked CLI PAT so we can use it again
+      await prisma.personalAccessToken.update({
+        where: {
+          id: existingCliPersonalAccessToken.id,
+        },
+        data: {
+          revokedAt: null,
+        },
+      });
+    }
+
     //we don't return the decrypted token
     return {
       id: existingCliPersonalAccessToken.id,

apps/webapp/app/utils/db.ts

@@ -0,0 +1,18 @@
+export function isValidDatabaseUrl(url: string) {
+  try {
+    const databaseUrl = new URL(url);
+    const schemaFromSearchParam = databaseUrl.searchParams.get("schema");
+
+    if (schemaFromSearchParam === "") {
+      console.error(
+        "Invalid Database URL: The schema search param can't have an empty value. To use the `public` schema, either omit the schema param entirely or specify it in full: `?schema=public`"
+      );
+      return false;
+    }
+
+    return true;
+  } catch (err) {
+    console.error(err);
+    return false;
+  }
+}

apps/webapp/app/utils/pathBuilder.ts

@@ -14,8 +14,6 @@ import { Job } from "~/models/job.server";
 import type { Organization } from "~/models/organization.server";
 import type { Project } from "~/models/project.server";
 import { objectToSearchParams } from "./searchParams";
-import { ScheduleListFilters } from "~/components/runs/v3/ScheduleFilters";
-import { useLocation } from "@remix-run/react";
 
 export type OrgForPath = Pick<Organization, "slug">;
 export type ProjectForPath = Pick<Project, "slug">;
@@ -368,7 +366,7 @@ export function v3RunSpanPath(
   run: v3RunForPath,
   span: v3SpanForPath
 ) {
-  return `${v3RunPath(organization, project, run)}/spans/${span.spanId}`;
+  return `${v3RunPath(organization, project, run)}?span=${span.spanId}`;
 }
 
 export function v3TraceSpanPath(

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

@@ -1,9 +1,5 @@
-import {
-  ZodMessageHandler,
-  ZodMessageSender,
-  clientWebsocketMessages,
-  serverWebsocketMessages,
-} from "@trigger.dev/core/v3";
+import { clientWebsocketMessages, serverWebsocketMessages } from "@trigger.dev/core/v3";
+import { ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
 import { Evt } from "evt";
 import { randomUUID } from "node:crypto";
 import type { CloseEvent, ErrorEvent, MessageEvent, WebSocket } from "ws";

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

@@ -63,6 +63,7 @@ export type TraceAttributes = Partial<
     | "batchId"
     | "payload"
     | "payloadType"
+    | "idempotencyKey"
   >
 >;
 
@@ -371,6 +372,7 @@ export class EventRepository {
         id: event.spanId,
         parentId: event.parentId ?? undefined,
         runId: event.runId,
+        idempotencyKey: event.idempotencyKey,
         data: {
           message: event.message,
           style: event.style,
@@ -459,7 +461,7 @@ export class EventRepository {
     const links: SpanLink[] = [];
 
     if (messagingEvent.success && messagingEvent.data) {
-      if ("id" in messagingEvent.data.message) {
+      if (messagingEvent.data.message && "id" in messagingEvent.data.message) {
         if (messagingEvent.data.message.id.startsWith("run_")) {
           links.push({
             type: "run",
@@ -719,6 +721,7 @@ export class EventRepository {
       links: links as unknown as Prisma.InputJsonValue,
       payload: options.attributes.payload,
       payloadType: options.attributes.payloadType,
+      idempotencyKey: options.attributes.idempotencyKey,
     };
 
     if (options.immediate) {

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

@@ -5,8 +5,8 @@ import {
   PlatformToProviderMessages,
   ProviderToPlatformMessages,
   SharedQueueToClientMessages,
-  ZodNamespace,
 } from "@trigger.dev/core/v3";
+import { ZodNamespace } from "@trigger.dev/core/v3/zodNamespace";
 import { Server } from "socket.io";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
@@ -71,6 +71,7 @@ function initializeSocketIOServerInstance() {
 
 function createCoordinatorNamespace(io: Server) {
   const coordinator = new ZodNamespace({
+    // @ts-ignore - for some reason the built ZodNamespace Server type is not compatible with the Server type here, but only when doing typechecking
     io,
     name: "coordinator",
     authToken: env.COORDINATOR_SECRET,
@@ -147,6 +148,7 @@ function createCoordinatorNamespace(io: Server) {
 
 function createProviderNamespace(io: Server) {
   const provider = new ZodNamespace({
+    // @ts-ignore - for some reason the built ZodNamespace Server type is not compatible with the Server type here, but only when doing typechecking
     io,
     name: "provider",
     authToken: env.PROVIDER_SECRET,
@@ -181,6 +183,7 @@ function createProviderNamespace(io: Server) {
 
 function createSharedQueueConsumerNamespace(io: Server) {
   const sharedQueue = new ZodNamespace({
+    // @ts-ignore - for some reason the built ZodNamespace Server type is not compatible with the Server type here, but only when doing typechecking
     io,
     name: "shared-queue",
     authToken: env.PROVIDER_SECRET,
@@ -188,7 +191,9 @@ function createSharedQueueConsumerNamespace(io: Server) {
     serverMessages: SharedQueueToClientMessages,
     onConnection: async (socket, handler, sender, logger) => {
       const sharedSocketConnection = new SharedSocketConnection({
+        // @ts-ignore - for some reason the built ZodNamespace Server type is not compatible with the Server type here, but only when doing typechecking
         namespace: sharedQueue.namespace,
+        // @ts-ignore - for some reason the built ZodNamespace Server type is not compatible with the Server type here, but only when doing typechecking
         socket,
         logger,
         poolSize: env.SHARED_QUEUE_CONSUMER_POOL_SIZE,

apps/webapp/app/v3/marqs/devQueueConsumer.server.ts

@@ -3,9 +3,9 @@ import {
   TaskRunExecution,
   TaskRunExecutionPayload,
   TaskRunExecutionResult,
-  ZodMessageSender,
   serverWebsocketMessages,
 } from "@trigger.dev/core/v3";
+import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
 import { BackgroundWorker, BackgroundWorkerTask } from "@trigger.dev/database";
 import { z } from "zod";
 import { prisma } from "~/db.server";
@@ -118,7 +118,7 @@ export class DevQueueConsumer {
     completion: TaskRunExecutionResult,
     execution: TaskRunExecution
   ) {
-    this._inProgressAttempts.delete(completion.id);
+    this._inProgressAttempts.delete(execution.attempt.id);
 
     if (completion.ok) {
       this._taskSuccesses++;
@@ -424,7 +424,7 @@ export class DevQueueConsumer {
           orderBy: { number: "desc" },
         },
         tags: true,
-        batchItem: {
+        batchItems: {
           include: {
             batchTaskRun: true,
           },
@@ -499,6 +499,7 @@ export class DevQueueConsumer {
         createdAt: lockedTaskRun.createdAt,
         tags: lockedTaskRun.tags.map((tag) => tag.name),
         isTest: lockedTaskRun.isTest,
+        idempotencyKey: lockedTaskRun.idempotencyKey ?? undefined,
       },
       queue: {
         id: queue.friendlyId,
@@ -520,9 +521,10 @@ export class DevQueueConsumer {
         slug: this.env.project.slug,
         name: this.env.project.name,
       },
-      batch: lockedTaskRun.batchItem?.batchTaskRun
-        ? { id: lockedTaskRun.batchItem.batchTaskRun.friendlyId }
-        : undefined,
+      batch:
+        lockedTaskRun.batchItems[0] && lockedTaskRun.batchItems[0].batchTaskRun
+          ? { id: lockedTaskRun.batchItems[0].batchTaskRun.friendlyId }
+          : undefined,
     };
 
     const environmentRepository = new EnvironmentVariablesRepository();

apps/webapp/app/v3/marqs/index.server.ts

@@ -39,7 +39,6 @@ const SemanticAttributes = {
 
 export type MarQSOptions = {
   redis: RedisOptions;
-  defaultQueueConcurrency: number;
   defaultEnvConcurrency: number;
   defaultOrgConcurrency: number;
   windowSize?: number;
@@ -58,16 +57,18 @@ export class MarQS {
   public keys: MarQSKeyProducer;
   private queuePriorityStrategy: MarQSQueuePriorityStrategy;
   #requeueingWorkers: Array<AsyncWorker> = [];
+  #rebalanceWorkers: Array<AsyncWorker> = [];
 
   constructor(private readonly options: MarQSOptions) {
     this.redis = new Redis(options.redis);
 
-    // Spawn options.workers workers to requeue visible messages
-    this.#startRequeuingWorkers();
-    this.#registerCommands();
-
     this.keys = options.keysProducer;
     this.queuePriorityStrategy = options.queuePriorityStrategy;
+
+    // Spawn options.workers workers to requeue visible messages
+    this.#startRequeuingWorkers();
+    this.#startRebalanceWorkers();
+    this.#registerCommands();
   }
 
   public async updateQueueConcurrencyLimits(
@@ -90,7 +91,7 @@ export class MarQS {
   public async getQueueConcurrencyLimit(env: AuthenticatedEnvironment, queue: string) {
     const result = await this.redis.get(this.keys.queueConcurrencyLimitKey(env, queue));
 
-    return result ? Number(result) : this.options.defaultQueueConcurrency;
+    return result ? Number(result) : undefined;
   }
 
   public async getEnvConcurrencyLimit(env: AuthenticatedEnvironment) {
@@ -246,6 +247,7 @@ export class MarQS {
           });
 
           await this.#callAcknowledgeMessage({
+            parentQueue,
             messageKey: this.keys.messageKey(messageData.messageId),
             messageQueue: messageQueue,
             visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
@@ -348,6 +350,7 @@ export class MarQS {
         });
 
         await this.#callAcknowledgeMessage({
+          parentQueue: message.parentQueue,
           messageKey: this.keys.messageKey(messageId),
           messageQueue: message.queue,
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
@@ -390,6 +393,7 @@ export class MarQS {
         });
 
         await this.#callAcknowledgeMessage({
+          parentQueue: oldMessage.parentQueue,
           messageKey: this.keys.messageKey(messageId),
           messageQueue: oldMessage.queue,
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
@@ -634,6 +638,17 @@ export class MarQS {
     return result;
   }
 
+  #startRebalanceWorkers() {
+    // Start a new worker to rebalance parent queues periodically
+    for (let i = 0; i < this.options.workers; i++) {
+      const worker = new AsyncWorker(this.#rebalanceParentQueues.bind(this), 60_000);
+
+      this.#rebalanceWorkers.push(worker);
+
+      worker.start();
+    }
+  }
+
   #startRequeuingWorkers() {
     // Start a new worker to requeue visible messages
     for (let i = 0; i < this.options.workers; i++) {
@@ -694,6 +709,106 @@ export class MarQS {
     }
   }
 
+  async #rebalanceParentQueues() {
+    return await new Promise<void>((resolve, reject) => {
+      // Scan for sorted sets with the parent queue pattern
+      const pattern = this.keys.sharedQueueScanPattern();
+      const redis = this.redis.duplicate();
+      const stream = redis.scanStream({
+        match: pattern,
+        type: "zset",
+        count: 100,
+      });
+
+      logger.debug("Streaming parent queues based on pattern", {
+        pattern,
+        component: "marqs",
+        operation: "rebalanceParentQueues",
+      });
+
+      stream.on("data", async (keys) => {
+        stream.pause();
+
+        const uniqueKeys = Array.from(new Set<string>(keys));
+
+        logger.debug("Rebalancing parent queues", {
+          component: "marqs",
+          operation: "rebalanceParentQueues",
+          parentQueues: uniqueKeys,
+        });
+
+        Promise.all(
+          uniqueKeys.map(async (key) => this.#rebalanceParentQueue(this.keys.stripKeyPrefix(key)))
+        ).finally(() => {
+          stream.resume();
+        });
+      });
+
+      stream.on("end", () => {
+        redis.quit().finally(() => {
+          resolve();
+        });
+      });
+
+      stream.on("error", (e) => {
+        redis.quit().finally(() => {
+          reject(e);
+        });
+      });
+    });
+  }
+
+  // Parent queue is a sorted set, the values of which are queue keys and the scores are is the oldest message in the queue
+  // We need to scan the parent queue and rebalance the queues based on the oldest message in the queue
+  async #rebalanceParentQueue(parentQueue: string) {
+    return await new Promise<void>((resolve, reject) => {
+      const redis = this.redis.duplicate();
+
+      const stream = redis.zscanStream(parentQueue, {
+        match: "*",
+        count: 100,
+      });
+
+      stream.on("data", async (childQueues) => {
+        stream.pause();
+
+        // childQueues is a flat array but of the form [queue1, score1, queue2, score2, ...], we want to group them into pairs
+        const childQueuesWithScores: Record<string, string> = {};
+
+        for (let i = 0; i < childQueues.length; i += 2) {
+          childQueuesWithScores[childQueues[i]] = childQueues[i + 1];
+        }
+
+        logger.debug("Rebalancing child queues", {
+          parentQueue,
+          childQueuesWithScores,
+          component: "marqs",
+          operation: "rebalanceParentQueues",
+        });
+
+        await Promise.all(
+          Object.entries(childQueuesWithScores).map(async ([childQueue, currentScore]) =>
+            this.#callRebalanceParentQueueChild({ parentQueue, childQueue, currentScore })
+          )
+        ).finally(() => {
+          stream.resume();
+        });
+      });
+
+      stream.on("end", () => {
+        redis.quit().finally(() => {
+          resolve();
+        });
+      });
+
+      stream.on("error", (e) => {
+        redis.quit().finally(() => {
+          reject(e);
+        });
+      });
+    });
+  }
+
   async #callEnqueueMessage(message: MessagePayload) {
     logger.debug("Calling enqueueMessage", {
       messagePayload: message,
@@ -744,7 +859,6 @@ export class MarQS {
       messageQueue,
       String(this.options.visibilityTimeoutInMs ?? 300000), // 5 minutes
       String(Date.now()),
-      String(this.options.defaultQueueConcurrency),
       String(this.options.defaultEnvConcurrency),
       String(this.options.defaultOrgConcurrency)
     );
@@ -768,6 +882,7 @@ export class MarQS {
   }
 
   async #callAcknowledgeMessage({
+    parentQueue,
     messageKey,
     messageQueue,
     visibilityQueue,
@@ -776,6 +891,7 @@ export class MarQS {
     orgConcurrencyKey,
     messageId,
   }: {
+    parentQueue: string;
     messageKey: string;
     messageQueue: string;
     visibilityQueue: string;
@@ -792,16 +908,19 @@ export class MarQS {
       envConcurrencyKey,
       orgConcurrencyKey,
       messageId,
+      parentQueue,
     });
 
     return this.redis.acknowledgeMessage(
+      parentQueue,
       messageKey,
       messageQueue,
       visibilityQueue,
       concurrencyKey,
       envConcurrencyKey,
       orgConcurrencyKey,
-      messageId
+      messageId,
+      messageQueue
     );
   }
 
@@ -894,16 +1013,22 @@ export class MarQS {
       concurrencyLimitKey,
       envConcurrencyLimitKey,
       orgConcurrencyLimitKey,
-      String(this.options.defaultQueueConcurrency),
       String(this.options.defaultEnvConcurrency),
       String(this.options.defaultOrgConcurrency)
     );
 
+    const queueCurrent = Number(capacities[0]);
+    const envLimit = Number(capacities[3]);
+    const orgLimit = Number(capacities[5]);
+    const queueLimit = capacities[1] ? Number(capacities[1]) : Math.min(envLimit, orgLimit);
+    const envCurrent = Number(capacities[2]);
+    const orgCurrent = Number(capacities[4]);
+
     // [queue current, queue limit, env current, env limit, org current, org limit]
     return {
-      queue: { current: Number(capacities[0]), limit: Number(capacities[1]) },
-      env: { current: Number(capacities[2]), limit: Number(capacities[3]) },
-      org: { current: Number(capacities[4]), limit: Number(capacities[5]) },
+      queue: { current: queueCurrent, limit: queueLimit },
+      env: { current: envCurrent, limit: envLimit },
+      org: { current: orgCurrent, limit: orgLimit },
     };
   }
 
@@ -926,6 +1051,35 @@ export class MarQS {
     );
   }
 
+  async #callRebalanceParentQueueChild({
+    parentQueue,
+    childQueue,
+    currentScore,
+  }: {
+    parentQueue: string;
+    childQueue: string;
+    currentScore: string;
+  }) {
+    const rebalanceResult = await this.redis.rebalanceParentQueueChild(
+      childQueue,
+      parentQueue,
+      childQueue,
+      currentScore
+    );
+
+    if (rebalanceResult) {
+      logger.debug("Rebalanced parent queue child", {
+        parentQueue,
+        childQueue,
+        currentScore,
+        rebalanceResult,
+        operation: "rebalanceParentQueueChild",
+      });
+    }
+
+    return rebalanceResult;
+  }
+
   #registerCommands() {
     this.redis.defineCommand("enqueueMessage", {
       numberOfKeys: 3,
@@ -969,13 +1123,12 @@ local currentConcurrencyKey = KEYS[7]
 local envCurrentConcurrencyKey = KEYS[8]
 local orgCurrentConcurrencyKey = KEYS[9]
 
--- Args: childQueueName, visibilityQueue, currentTime, defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
+-- Args: childQueueName, visibilityQueue, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
 local childQueueName = ARGV[1]
 local visibilityTimeout = tonumber(ARGV[2])
 local currentTime = tonumber(ARGV[3])
-local defaultConcurrencyLimit = ARGV[4]
-local defaultEnvConcurrencyLimit = ARGV[5]
-local defaultOrgConcurrencyLimit = ARGV[6]
+local defaultEnvConcurrencyLimit = ARGV[4]
+local defaultOrgConcurrencyLimit = ARGV[5]
 
 -- Check current org concurrency against the limit
 local orgCurrentConcurrency = tonumber(redis.call('SCARD', orgCurrentConcurrencyKey) or '0')
@@ -995,8 +1148,9 @@ end
 
 -- Check current queue concurrency against the limit
 local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
-local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
+local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or '1000000')
 
+-- Check condition only if concurrencyLimit exists
 if currentConcurrency >= concurrencyLimit then
     return nil
 end
@@ -1032,19 +1186,20 @@ return {messageId, messageScore} -- Return message details
     });
 
     this.redis.defineCommand("acknowledgeMessage", {
-      numberOfKeys: 6,
+      numberOfKeys: 7,
       lua: `
--- Keys: messageKey, messageQueue, visibilityQueue, concurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
-local messageKey = KEYS[1]
-local messageQueue = KEYS[2]
-local visibilityQueue = KEYS[3]
-local concurrencyKey = KEYS[4]
-local envCurrentConcurrencyKey = KEYS[5]
-local orgCurrentConcurrencyKey = KEYS[6]
-local globalCurrentConcurrencyKey = KEYS[7]
+-- Keys: parentQueue, messageKey, messageQueue, visibilityQueue, concurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
+local parentQueue = KEYS[1]
+local messageKey = KEYS[2]
+local messageQueue = KEYS[3]
+local visibilityQueue = KEYS[4]
+local concurrencyKey = KEYS[5]
+local envCurrentConcurrencyKey = KEYS[6]
+local orgCurrentConcurrencyKey = KEYS[7]
 
--- Args: messageId
+-- Args: messageId, messageQueueName
 local messageId = ARGV[1]
+local messageQueueName = ARGV[2]
 
 -- Remove the message from the message key
 redis.call('DEL', messageKey)
@@ -1052,6 +1207,14 @@ redis.call('DEL', messageKey)
 -- Remove the message from the queue
 redis.call('ZREM', messageQueue, messageId)
 
+-- Rebalance the parent queue
+local earliestMessage = redis.call('ZRANGE', messageQueue, 0, 0, 'WITHSCORES')
+if #earliestMessage == 0 then
+    redis.call('ZREM', parentQueue, messageQueueName)
+else
+    redis.call('ZADD', parentQueue, earliestMessage[2], messageQueueName)
+end
+
 -- Remove the message from the timeout queue
 redis.call('ZREM', visibilityQueue, messageId)
 
@@ -1145,10 +1308,9 @@ local concurrencyLimitKey = KEYS[4]
 local envConcurrencyLimitKey = KEYS[5]
 local orgConcurrencyLimitKey = KEYS[6]
 
--- Args defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
-local defaultConcurrencyLimit = tonumber(ARGV[1])
-local defaultEnvConcurrencyLimit = tonumber(ARGV[2])
-local defaultOrgConcurrencyLimit = tonumber(ARGV[3])
+-- Args defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
+local defaultEnvConcurrencyLimit = tonumber(ARGV[1])
+local defaultOrgConcurrencyLimit = tonumber(ARGV[2])
 
 local currentOrgConcurrency = tonumber(redis.call('SCARD', currentOrgConcurrencyKey) or '0')
 local orgConcurrencyLimit = tonumber(redis.call('GET', orgConcurrencyLimitKey) or defaultOrgConcurrencyLimit)
@@ -1157,7 +1319,7 @@ local currentEnvConcurrency = tonumber(redis.call('SCARD', currentEnvConcurrency
 local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
 
 local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
-local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
+local concurrencyLimit = redis.call('GET', concurrencyLimitKey)
 
 -- Return current capacity and concurrency limits for the queue, env, org
 return { currentConcurrency, concurrencyLimit, currentEnvConcurrency, envConcurrencyLimit, currentOrgConcurrency, orgConcurrencyLimit } 
@@ -1179,6 +1341,37 @@ redis.call('SET', envConcurrencyLimitKey, envConcurrencyLimit)
 redis.call('SET', orgConcurrencyLimitKey, orgConcurrencyLimit)
       `,
     });
+
+    this.redis.defineCommand("rebalanceParentQueueChild", {
+      numberOfKeys: 2,
+      lua: `
+-- Keys: childQueueKey, parentQueueKey
+local childQueueKey = KEYS[1]
+local parentQueueKey = KEYS[2]
+
+-- Args: childQueueName, currentScore
+local childQueueName = ARGV[1]
+local currentScore = ARGV[2]
+
+-- Rebalance the parent queue
+local earliestMessage = redis.call('ZRANGE', childQueueKey, 0, 0, 'WITHSCORES')
+if #earliestMessage == 0 then
+    redis.call('ZREM', parentQueueKey, childQueueName)
+
+    -- Return true because the parent queue was rebalanced
+    return true
+else
+    -- If the earliest message is different, update the parent queue and return true, else return false
+    if earliestMessage[2] == currentScore then
+        return false
+    end
+
+    redis.call('ZADD', parentQueueKey, earliestMessage[2], childQueueName)
+
+    return earliestMessage[2]
+end
+`,
+    });
   }
 }
 
@@ -1208,13 +1401,13 @@ declare module "ioredis" {
       childQueueName: string,
       visibilityTimeout: string,
       currentTime: string,
-      defaultConcurrencyLimit: string,
       defaultEnvConcurrencyLimit: string,
       defaultOrgConcurrencyLimit: string,
       callback?: Callback<[string, string]>
     ): Result<[string, string] | null, Context>;
 
     acknowledgeMessage(
+      parentQueue: string,
       messageKey: string,
       messageQueue: string,
       visibilityQueue: string,
@@ -1222,6 +1415,7 @@ declare module "ioredis" {
       envConcurrencyKey: string,
       orgConcurrencyKey: string,
       messageId: string,
+      messageQueueName: string,
       callback?: Callback<void>
     ): Result<void, Context>;
 
@@ -1255,7 +1449,6 @@ declare module "ioredis" {
       concurrencyLimitKey: string,
       envConcurrencyLimitKey: string,
       orgConcurrencyLimitKey: string,
-      defaultConcurrencyLimit: string,
       defaultEnvConcurrencyLimit: string,
       defaultOrgConcurrencyLimit: string,
       callback?: Callback<number[]>
@@ -1268,6 +1461,14 @@ declare module "ioredis" {
       orgConcurrencyLimit: string,
       callback?: Callback<void>
     ): Result<void, Context>;
+
+    rebalanceParentQueueChild(
+      childQueueKey: string,
+      parentQueueKey: string,
+      childQueueName: string,
+      currentScore: string,
+      callback?: Callback<number | string | null>
+    ): Result<number | string | null, Context>;
   }
 }
 
@@ -1292,7 +1493,6 @@ function getMarQSClient() {
         envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
         workers: 1,
         redis: redisOptions,
-        defaultQueueConcurrency: env.DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT,
         defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
         defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
         visibilityTimeoutInMs: 120 * 1000, // 2 minutes,

apps/webapp/app/v3/marqs/marqsKeyProducer.server.ts

@@ -15,6 +15,18 @@ const constants = {
 export class MarQSShortKeyProducer implements MarQSKeyProducer {
   constructor(private _prefix: string) {}
 
+  sharedQueueScanPattern() {
+    return `${this._prefix}*${constants.SHARED_QUEUE}`;
+  }
+
+  stripKeyPrefix(key: string): string {
+    if (key.startsWith(this._prefix)) {
+      return key.slice(this._prefix.length);
+    }
+
+    return key;
+  }
+
   queueConcurrencyLimitKey(env: AuthenticatedEnvironment, queue: string) {
     return [this.queueKey(env, queue), constants.CONCURRENCY_LIMIT_PART].join(":");
   }

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -8,9 +8,9 @@ import {
   TaskRunExecutionResult,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
-  ZodMessageSender,
   serverWebsocketMessages,
 } from "@trigger.dev/core/v3";
+import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
 import {
   BackgroundWorker,
   BackgroundWorkerTask,
@@ -26,7 +26,6 @@ import { EnvironmentVariablesRepository } from "../environmentVariables/environm
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { socketIo } from "../handleSocketIo.server";
 import { findCurrentWorkerDeployment } from "../models/workerDeployment.server";
-import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { RestoreCheckpointService } from "../services/restoreCheckpoint.server";
 import { tracer } from "../tracer.server";
 
@@ -812,7 +811,7 @@ class SharedQueueTasks {
     if (ok) {
       const success: TaskRunSuccessfulExecutionResult = {
         ok,
-        id: attempt.friendlyId,
+        id: attempt.taskRun.friendlyId,
         output: attempt.output ?? undefined,
         outputType: attempt.outputType,
       };
@@ -820,7 +819,7 @@ class SharedQueueTasks {
     } else {
       const failure: TaskRunFailedExecutionResult = {
         ok,
-        id: attempt.friendlyId,
+        id: attempt.taskRun.friendlyId,
         error: attempt.error as TaskRunError,
       };
       return failure;
@@ -848,7 +847,7 @@ class SharedQueueTasks {
         taskRun: {
           include: {
             tags: true,
-            batchItem: {
+            batchItems: {
               include: {
                 batchTaskRun: true,
               },
@@ -956,6 +955,7 @@ class SharedQueueTasks {
         createdAt: taskRun.createdAt,
         tags: taskRun.tags.map((tag) => tag.name),
         isTest: taskRun.isTest,
+        idempotencyKey: taskRun.idempotencyKey ?? undefined,
       },
       queue: {
         id: queue.friendlyId,
@@ -977,9 +977,10 @@ class SharedQueueTasks {
         slug: attempt.runtimeEnvironment.project.slug,
         name: attempt.runtimeEnvironment.project.name,
       },
-      batch: taskRun.batchItem?.batchTaskRun
-        ? { id: taskRun.batchItem.batchTaskRun.friendlyId }
-        : undefined,
+      batch:
+        taskRun.batchItems[0] && taskRun.batchItems[0].batchTaskRun
+          ? { id: taskRun.batchItems[0].batchTaskRun.friendlyId }
+          : undefined,
       worker: {
         id: attempt.backgroundWorkerId,
         contentHash: attempt.backgroundWorker.contentHash,

apps/webapp/app/v3/marqs/types.ts

@@ -1,4 +1,3 @@
-import { RedisOptions } from "ioredis";
 import { z } from "zod";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 
@@ -25,6 +24,7 @@ export interface MarQSKeyProducer {
   orgConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
   queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string): string;
   envSharedQueueKey(env: AuthenticatedEnvironment): string;
+  sharedQueueScanPattern(): string;
   concurrencyLimitKeyFromQueue(queue: string): string;
   currentConcurrencyKeyFromQueue(queue: string): string;
   currentConcurrencyKey(
@@ -39,6 +39,7 @@ export interface MarQSKeyProducer {
   orgCurrentConcurrencyKey(env: AuthenticatedEnvironment): string;
   envCurrentConcurrencyKey(env: AuthenticatedEnvironment): string;
   messageKey(messageId: string): string;
+  stripKeyPrefix(key: string): string;
 }
 
 export type PriorityStrategyChoice = string | { abort: true };

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

@@ -352,6 +352,7 @@ function extractResourceProperties(attributes: KeyValue[]) {
     queueId: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_ID),
     queueName: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_NAME),
     batchId: extractStringAttribute(attributes, SemanticInternalAttributes.BATCH_ID),
+    idempotencyKey: extractStringAttribute(attributes, SemanticInternalAttributes.IDEMPOTENCY_KEY),
   };
 }
 

apps/webapp/app/v3/services/batchTriggerTask.server.ts

@@ -1,9 +1,9 @@
 import { BatchTriggerTaskRequestBody } from "@trigger.dev/core/v3";
-import { nanoid } from "nanoid";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
 import { BaseService } from "./baseService.server";
 import { TriggerTaskService } from "./triggerTask.server";
+import { batchTaskRunItemStatusForRunStatus } from "~/models/taskRun.server";
 
 export type BatchTriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -22,23 +22,23 @@ export class BatchTriggerTaskService extends BaseService {
     return await this.traceWithEnv("call()", environment, async (span) => {
       span.setAttribute("taskId", taskId);
 
-      const idempotencyKey = options.idempotencyKey ?? nanoid();
-
-      const existingBatch = await this._prisma.batchTaskRun.findUnique({
-        where: {
-          runtimeEnvironmentId_idempotencyKey: {
-            runtimeEnvironmentId: environment.id,
-            idempotencyKey,
-          },
-        },
-        include: {
-          items: {
-            include: {
-              taskRun: true,
+      const existingBatch = options.idempotencyKey
+        ? await this._prisma.batchTaskRun.findUnique({
+            where: {
+              runtimeEnvironmentId_idempotencyKey: {
+                runtimeEnvironmentId: environment.id,
+                idempotencyKey: options.idempotencyKey,
+              },
             },
-          },
-        },
-      });
+            include: {
+              items: {
+                include: {
+                  taskRun: true,
+                },
+              },
+            },
+          })
+        : undefined;
 
       if (existingBatch) {
         span.setAttribute("batchId", existingBatch.friendlyId);
@@ -58,7 +58,7 @@ export class BatchTriggerTaskService extends BaseService {
         data: {
           friendlyId: generateFriendlyId("batch"),
           runtimeEnvironmentId: environment.id,
-          idempotencyKey,
+          idempotencyKey: options.idempotencyKey,
           taskIdentifier: taskId,
           dependentTaskAttemptId: dependentAttempt?.id,
         },
@@ -70,8 +70,6 @@ export class BatchTriggerTaskService extends BaseService {
       let index = 0;
 
       for (const item of body.items) {
-        const idempotencyKey = nanoid();
-
         const run = await triggerTaskService.call(
           taskId,
           environment,
@@ -83,7 +81,6 @@ export class BatchTriggerTaskService extends BaseService {
             },
           },
           {
-            idempotencyKey,
             triggerVersion: options.triggerVersion,
             traceContext: options.traceContext,
             spanParentAsLink: options.spanParentAsLink,
@@ -96,6 +93,7 @@ export class BatchTriggerTaskService extends BaseService {
             data: {
               batchTaskRunId: batch.id,
               taskRunId: run.id,
+              status: batchTaskRunItemStatusForRunStatus(run.status),
             },
           });
 

apps/webapp/app/v3/services/cancelTaskRun.server.ts

@@ -4,9 +4,9 @@ import { marqs } from "~/v3/marqs/index.server";
 import { devPubSub } from "../marqs/devPubSub.server";
 import { BaseService } from "./baseService.server";
 import { socketIo } from "../handleSocketIo.server";
-import { assertUnreachable } from "../utils/asserts.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
 import { logger } from "~/services/logger.server";
+import assertNever from "assert-never";
 
 export const CANCELLABLE_STATUSES: Array<TaskRunStatus> = [
   "PENDING",
@@ -148,7 +148,7 @@ export class CancelTaskRunService extends BaseService {
             break;
           }
           default: {
-            assertUnreachable(attempt.status);
+            assertNever(attempt.status);
           }
         }
       }

apps/webapp/app/v3/services/completeAttempt.server.ts

@@ -39,10 +39,12 @@ export class CompleteAttemptService extends BaseService {
     env?: AuthenticatedEnvironment;
     checkpoint?: CheckpointData;
   }): Promise<"COMPLETED" | "RETRIED"> {
-    const taskRunAttempt = await findAttempt(this._prisma, completion.id);
+    const taskRunAttempt = await findAttempt(this._prisma, execution.attempt.id);
 
     if (!taskRunAttempt) {
-      logger.error("[CompleteAttemptService] Task run attempt not found", { id: completion.id });
+      logger.error("[CompleteAttemptService] Task run attempt not found", {
+        id: execution.attempt.id,
+      });
 
       // Update the task run to be failed
       await this._prisma.taskRun.update({
@@ -76,7 +78,7 @@ export class CompleteAttemptService extends BaseService {
     env?: AuthenticatedEnvironment
   ): Promise<"COMPLETED"> {
     await this._prisma.taskRunAttempt.update({
-      where: { friendlyId: completion.id },
+      where: { id: taskRunAttempt.id },
       data: {
         status: "COMPLETED",
         completedAt: new Date(),
@@ -144,7 +146,7 @@ export class CompleteAttemptService extends BaseService {
     }
 
     await this._prisma.taskRunAttempt.update({
-      where: { friendlyId: completion.id },
+      where: { id: taskRunAttempt.id },
       data: {
         status: "FAILED",
         completedAt: new Date(),
@@ -248,14 +250,52 @@ export class CompleteAttemptService extends BaseService {
         },
       });
 
-      await this._prisma.taskRun.update({
-        where: {
-          id: taskRunAttempt.taskRunId,
-        },
-        data: {
-          status: "COMPLETED_WITH_ERRORS",
-        },
-      });
+      if (
+        completion.error.type === "INTERNAL_ERROR" &&
+        completion.error.code === "GRACEFUL_EXIT_TIMEOUT"
+      ) {
+        // We need to fail all incomplete spans
+        const inProgressEvents = await eventRepository.queryIncompleteEvents({
+          attemptId: execution.attempt.id,
+        });
+
+        logger.debug("Failing in-progress events", {
+          inProgressEvents: inProgressEvents.map((event) => event.id),
+        });
+
+        const exception = {
+          type: "Graceful exit timeout",
+          message: completion.error.message,
+        };
+
+        await Promise.all(
+          inProgressEvents.map((event) => {
+            return eventRepository.crashEvent({
+              event: event,
+              crashedAt: new Date(),
+              exception,
+            });
+          })
+        );
+
+        await this._prisma.taskRun.update({
+          where: {
+            id: taskRunAttempt.taskRunId,
+          },
+          data: {
+            status: "SYSTEM_FAILURE",
+          },
+        });
+      } else {
+        await this._prisma.taskRun.update({
+          where: {
+            id: taskRunAttempt.taskRunId,
+          },
+          data: {
+            status: "COMPLETED_WITH_ERRORS",
+          },
+        });
+      }
 
       if (!env || env.type !== "DEVELOPMENT") {
         await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);

apps/webapp/app/v3/services/createCheckpoint.server.ts

@@ -1,4 +1,5 @@
-import { CoordinatorToPlatformMessages, InferSocketMessageSchema } from "@trigger.dev/core/v3";
+import { CoordinatorToPlatformMessages } from "@trigger.dev/core/v3";
+import type { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
 import type {
   CheckpointRestoreEvent,
   TaskRunAttemptStatus,

apps/webapp/app/v3/services/resumeAttempt.server.ts

@@ -1,15 +1,16 @@
 import {
   CoordinatorToPlatformMessages,
-  InferSocketMessageSchema,
   TaskRunExecution,
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
-import { $transaction } from "~/db.server";
+import type { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
+import { $transaction, PrismaClientOrTransaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { socketIo } from "../handleSocketIo.server";
 import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
 import { BaseService } from "./baseService.server";
+import { TaskRunAttempt } from "@trigger.dev/database";
 
 export class ResumeAttemptService extends BaseService {
   public async call(
@@ -24,7 +25,7 @@ export class ResumeAttemptService extends BaseService {
         },
         include: {
           taskRun: true,
-          taskRunDependency: {
+          dependencies: {
             include: {
               taskRun: {
                 include: {
@@ -40,8 +41,12 @@ export class ResumeAttemptService extends BaseService {
                 },
               },
             },
+            orderBy: {
+              createdAt: "desc",
+            },
+            take: 1,
           },
-          batchTaskRunDependency: {
+          batchDependencies: {
             include: {
               items: {
                 include: {
@@ -61,6 +66,10 @@ export class ResumeAttemptService extends BaseService {
                 },
               },
             },
+            orderBy: {
+              createdAt: "desc",
+            },
+            take: 1,
           },
         },
       });
@@ -78,6 +87,8 @@ export class ResumeAttemptService extends BaseService {
         return;
       }
 
+      let completedAttemptIds: string[] = [];
+
       switch (params.type) {
         case "WAIT_FOR_DURATION": {
           logger.error(
@@ -93,12 +104,10 @@ export class ResumeAttemptService extends BaseService {
           });
           break;
         }
-        case "WAIT_FOR_TASK":
-        case "WAIT_FOR_BATCH": {
-          let completedAttemptIds: string[] = [];
-
-          if (attempt.taskRunDependency) {
-            const dependentAttempt = attempt.taskRunDependency.taskRun.attempts[0];
+        case "WAIT_FOR_TASK": {
+          if (attempt.dependencies.length) {
+            // We only care about the latest dependency
+            const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0];
 
             if (!dependentAttempt) {
               logger.error("No dependent attempt", { attemptId: attempt.id });
@@ -106,19 +115,16 @@ export class ResumeAttemptService extends BaseService {
             }
 
             completedAttemptIds = [dependentAttempt.id];
-
-            await tx.taskRunAttempt.update({
-              where: {
-                id: attempt.id,
-              },
-              data: {
-                taskRunDependency: {
-                  disconnect: true,
-                },
-              },
-            });
-          } else if (attempt.batchTaskRunDependency) {
-            const dependentBatchItems = attempt.batchTaskRunDependency.items;
+          } else {
+            logger.error("No task dependency", { attemptId: attempt.id });
+            return;
+          }
+          break;
+        }
+        case "WAIT_FOR_BATCH": {
+          if (attempt.batchDependencies) {
+            // We only care about the latest batch dependency
+            const dependentBatchItems = attempt.batchDependencies[0].items;
 
             if (!dependentBatchItems) {
               logger.error("No dependent batch items", { attemptId: attempt.id });
@@ -126,115 +132,112 @@ export class ResumeAttemptService extends BaseService {
             }
 
             completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id);
-
-            await tx.taskRunAttempt.update({
-              where: {
-                id: attempt.id,
-              },
-              data: {
-                batchTaskRunDependency: {
-                  disconnect: true,
-                },
-              },
-            });
           } else {
-            logger.error("No dependencies", { attemptId: attempt.id });
+            logger.error("No batch dependency", { attemptId: attempt.id });
             return;
           }
-
-          if (completedAttemptIds.length === 0) {
-            logger.error("No completed attempt IDs", { attemptId: attempt.id });
-            return;
-          }
-
-          const completions: TaskRunExecutionResult[] = [];
-          const executions: TaskRunExecution[] = [];
-
-          for (const completedAttemptId of completedAttemptIds) {
-            const completedAttempt = await tx.taskRunAttempt.findUnique({
-              where: {
-                id: completedAttemptId,
-                taskRun: {
-                  lockedAt: {
-                    not: null,
-                  },
-                  lockedById: {
-                    not: null,
-                  },
-                },
-              },
-            });
-
-            if (!completedAttempt) {
-              logger.error("Completed attempt not found", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
-              completedAttempt.id
-            );
-
-            if (!completion) {
-              logger.error("Failed to get completion payload", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            completions.push(completion);
-
-            const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
-              completedAttempt.id
-            );
-
-            if (!executionPayload) {
-              logger.error("Failed to get execution payload", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            executions.push(executionPayload.execution);
-          }
-
-          const updated = await tx.taskRunAttempt.update({
-            where: {
-              id: attempt.id,
-            },
-            data: {
-              status: "EXECUTING",
-              taskRun: {
-                update: {
-                  data: {
-                    status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
-                  },
-                },
-              },
-            },
-          });
-
-          socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
-            version: "v1",
-            runId: attempt.taskRunId,
-            attemptId: attempt.id,
-            attemptFriendlyId: attempt.friendlyId,
-            completions,
-            executions,
-          });
           break;
         }
         default: {
           break;
         }
       }
+
+      await this.#handleDependencyResume(attempt, completedAttemptIds, tx);
+    });
+  }
+
+  async #handleDependencyResume(
+    attempt: TaskRunAttempt,
+    completedAttemptIds: string[],
+    tx: PrismaClientOrTransaction
+  ) {
+    if (completedAttemptIds.length === 0) {
+      logger.error("No completed attempt IDs", { attemptId: attempt.id });
+      return;
+    }
+
+    const completions: TaskRunExecutionResult[] = [];
+    const executions: TaskRunExecution[] = [];
+
+    for (const completedAttemptId of completedAttemptIds) {
+      const completedAttempt = await tx.taskRunAttempt.findUnique({
+        where: {
+          id: completedAttemptId,
+          taskRun: {
+            lockedAt: {
+              not: null,
+            },
+            lockedById: {
+              not: null,
+            },
+          },
+        },
+      });
+
+      if (!completedAttempt) {
+        logger.error("Completed attempt not found", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
+        completedAttempt.id
+      );
+
+      if (!completion) {
+        logger.error("Failed to get completion payload", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      completions.push(completion);
+
+      const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
+        completedAttempt.id
+      );
+
+      if (!executionPayload) {
+        logger.error("Failed to get execution payload", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      executions.push(executionPayload.execution);
+    }
+
+    const updated = await tx.taskRunAttempt.update({
+      where: {
+        id: attempt.id,
+      },
+      data: {
+        status: "EXECUTING",
+        taskRun: {
+          update: {
+            data: {
+              status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
+            },
+          },
+        },
+      },
+    });
+
+    socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
+      version: "v1",
+      runId: attempt.taskRunId,
+      attemptId: attempt.id,
+      attemptFriendlyId: attempt.friendlyId,
+      completions,
+      executions,
     });
   }
 }

apps/webapp/app/v3/services/resumeTaskRunDependencies.server.ts

@@ -12,7 +12,7 @@ export class ResumeTaskRunDependenciesService extends BaseService {
       include: {
         taskRun: {
           include: {
-            batchItem: true,
+            batchItems: true,
             dependency: {
               include: {
                 dependentAttempt: true,
@@ -34,14 +34,16 @@ export class ResumeTaskRunDependenciesService extends BaseService {
       return;
     }
 
-    const { batchItem, dependency } = taskAttempt.taskRun;
+    const { batchItems, dependency } = taskAttempt.taskRun;
 
-    if (!batchItem && !dependency) {
+    if (!batchItems.length && !dependency) {
       return;
     }
 
-    if (batchItem) {
-      await this.#resumeBatchItem(batchItem, taskAttempt);
+    if (batchItems.length) {
+      for (const batchItem of batchItems) {
+        await this.#resumeBatchItem(batchItem, taskAttempt);
+      }
       return;
     }
 

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

@@ -34,18 +34,20 @@ export class TriggerTaskService extends BaseService {
     return await this.traceWithEnv("call()", environment, async (span) => {
       span.setAttribute("taskId", taskId);
 
-      const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey ?? nanoid();
+      const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey;
 
-      const existingRun = await this._prisma.taskRun.findUnique({
-        where: {
-          runtimeEnvironmentId_idempotencyKey: {
-            runtimeEnvironmentId: environment.id,
-            idempotencyKey,
-          },
-        },
-      });
+      const existingRun = idempotencyKey
+        ? await this._prisma.taskRun.findUnique({
+            where: {
+              runtimeEnvironmentId_idempotencyKey: {
+                runtimeEnvironmentId: environment.id,
+                idempotencyKey,
+              },
+            },
+          })
+        : undefined;
 
-      if (existingRun) {
+      if (existingRun && existingRun.taskIdentifier === taskId) {
         span.setAttribute("runId", existingRun.friendlyId);
         return existingRun;
       }
@@ -68,6 +70,7 @@ export class TriggerTaskService extends BaseService {
             },
             runIsTest: body.options?.test ?? false,
             batchId: options.batchId,
+            idempotencyKey,
           },
           incomplete: true,
           immediate: true,

apps/webapp/app/v3/sharedSocketConnection.ts

@@ -1,11 +1,10 @@
+import { clientWebsocketMessages, serverWebsocketMessages } from "@trigger.dev/core/v3";
 import {
-  MessageCatalogToSocketIoEvents,
-  StructuredLogger,
   ZodMessageHandler,
   ZodMessageSender,
-  clientWebsocketMessages,
-  serverWebsocketMessages,
-} from "@trigger.dev/core/v3";
+  MessageCatalogToSocketIoEvents,
+} from "@trigger.dev/core/v3/zodMessageHandler";
+import type { StructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 import { Evt } from "evt";
 import { randomUUID } from "node:crypto";
 import { logger } from "~/services/logger.server";

apps/webapp/app/v3/utils/asserts.server.ts

@@ -1,3 +0,0 @@
-export function assertUnreachable(x: never): never {
-  throw new Error("Didn't expect to get here");
-}

apps/webapp/app/v3/utils/zodPubSub.server.ts

@@ -1,5 +1,5 @@
 import { Logger } from "@trigger.dev/core-backend";
-import { ZodMessageCatalogSchema, ZodMessageHandler, ZodMessageSender } from "@trigger.dev/core/v3";
+import { ZodMessageCatalogSchema, ZodMessageHandler } from "@trigger.dev/core/v3/zodMessageHandler";
 import { Evt } from "evt";
 import Redis, { RedisOptions } from "ioredis";
 import { z } from "zod";

apps/webapp/package.json

@@ -96,6 +96,7 @@
     "@uiw/react-codemirror": "^4.19.5",
     "@upstash/ratelimit": "^1.0.1",
     "@whatwg-node/fetch": "^0.9.14",
+    "assert-never": "^1.2.1",
     "aws4fetch": "^1.0.18",
     "class-variance-authority": "^0.5.2",
     "clsx": "^1.2.1",
@@ -149,7 +150,7 @@
     "simple-oauth2": "^5.0.0",
     "simplur": "^3.0.1",
     "slug": "^6.0.0",
-    "socket.io": "^4.7.4",
+    "socket.io": "4.7.4",
     "socket.io-adapter": "^2.5.4",
     "sonner": "^1.0.3",
     "sqs-consumer": "^7.4.0",

config-packages/tsconfig/package.json

@@ -6,4 +6,4 @@
   "publishConfig": {
     "access": "public"
   }
-}
+}

docs/_snippets/v3/step-cli-dev.mdx

@@ -2,6 +2,8 @@
 
 The CLI `dev` command runs a server for your tasks. It will watches for changes in your `/trigger` directory and communicates with the Trigger.dev platform to register your tasks, perform runs, and send data back and forth.
 
+It can also update your `@trigger.dev/*` packages to prevent version mismatches and failed deploys. You will always be prompted first.
+
 <CodeGroup>
 
 ```bash npm

docs/mint.json

@@ -1,14 +1,8 @@
 {
   "$schema": "https://mintlify.com/schema.json",
   "name": "Trigger.dev",
-  "openapi": [
-    "/openapi.yml",
-    "/v3-openapi.json"
-  ],
-  "versions": [
-    "v3 (Developer Preview)",
-    "v2"
-  ],
+  "openapi": ["/openapi.yml", "/v3-openapi.json"],
+  "versions": ["v3 (Developer Preview)", "v2"],
   "logo": {
     "dark": "/logo/dark.png",
     "light": "/logo/light.png",
@@ -55,7 +49,7 @@
     },
     {
       "name": "v2",
-      "url": "https://trigger.dev/docs",
+      "url": "https://trigger.dev/docs/documentation",
       "version": "v3 (Developer Preview)"
     },
     {
@@ -96,9 +90,7 @@
     {
       "group": "",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/introduction"
-      ]
+      "pages": ["v3/introduction"]
     },
     {
       "group": "Getting Started",
@@ -108,7 +100,6 @@
         "v3/upgrading-from-v2",
         "v3/changelog",
         "v3/feature-matrix",
-        "v3/known-issues",
         "v3/limits-performance"
       ]
     },
@@ -122,12 +113,7 @@
         "v3/apikeys",
         {
           "group": "Task types",
-          "pages": [
-            "v3/tasks-regular",
-            "v3/tasks-scheduled",
-            "v3/tasks-zod",
-            "v3/tasks-webhooks"
-          ]
+          "pages": ["v3/tasks-regular", "v3/tasks-scheduled", "v3/tasks-zod", "v3/tasks-webhooks"]
         },
         "v3/trigger-config"
       ]
@@ -135,10 +121,7 @@
     {
       "group": "Development",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/cli-dev",
-        "v3/run-tests"
-      ]
+      "pages": ["v3/cli-dev", "v3/run-tests"]
     },
     {
       "group": "Deployment",
@@ -149,9 +132,7 @@
         "v3/github-actions",
         {
           "group": "Deployment integrations",
-          "pages": [
-            "v3/vercel-integration"
-          ]
+          "pages": ["v3/vercel-integration"]
         }
       ]
     },
@@ -227,9 +208,7 @@
         },
         {
           "group": "Objects",
-          "pages": [
-            "v3/reference-context"
-          ]
+          "pages": ["v3/reference-context"]
         },
         {
           "group": "CLI",
@@ -280,11 +259,7 @@
     {
       "group": "Open source",
       "version": "v3 (Developer Preview)",
-      "pages": [
-        "v3/github-repo",
-        "v3/open-source-self-hosting",
-        "v3/open-source-contributing"
-      ]
+      "pages": ["v3/github-repo", "v3/open-source-self-hosting", "v3/open-source-contributing"]
     },
     {
       "group": "Help",
@@ -485,10 +460,7 @@
       "pages": [
         {
           "group": "Airtable",
-          "pages": [
-            "integrations/apis/airtable",
-            "integrations/apis/airtable-tasks"
-          ]
+          "pages": ["integrations/apis/airtable", "integrations/apis/airtable-tasks"]
         },
         {
           "group": "GitHub",
@@ -514,25 +486,16 @@
         },
         {
           "group": "Plain",
-          "pages": [
-            "integrations/apis/plain",
-            "integrations/apis/plain-tasks"
-          ]
+          "pages": ["integrations/apis/plain", "integrations/apis/plain-tasks"]
         },
         "integrations/apis/replicate",
         {
           "group": "SendGrid",
-          "pages": [
-            "integrations/apis/sendgrid",
-            "integrations/apis/sendgrid-tasks"
-          ]
+          "pages": ["integrations/apis/sendgrid", "integrations/apis/sendgrid-tasks"]
         },
         {
           "group": "Resend",
-          "pages": [
-            "integrations/apis/resend",
-            "integrations/apis/resend-tasks"
-          ]
+          "pages": ["integrations/apis/resend", "integrations/apis/resend-tasks"]
         },
         {
           "group": "Shopify",
@@ -544,10 +507,7 @@
         },
         {
           "group": "Slack",
-          "pages": [
-            "integrations/apis/slack",
-            "integrations/apis/slack-tasks"
-          ]
+          "pages": ["integrations/apis/slack", "integrations/apis/slack-tasks"]
         },
         "integrations/apis/stripe",
         {
@@ -572,9 +532,7 @@
             "sdk/triggerclient/constructor",
             {
               "group": "Instance properties",
-              "pages": [
-                "sdk/triggerclient/store"
-              ]
+              "pages": ["sdk/triggerclient/store"]
             },
             {
               "group": "Instance methods",
@@ -637,10 +595,7 @@
             "sdk/dynamictrigger/constructor",
             {
               "group": "Instance methods",
-              "pages": [
-                "sdk/dynamictrigger/register",
-                "sdk/dynamictrigger/unregister"
-              ]
+              "pages": ["sdk/dynamictrigger/register", "sdk/dynamictrigger/unregister"]
             }
           ]
         },
@@ -651,10 +606,7 @@
             "sdk/dynamicschedule/constructor",
             {
               "group": "Instance methods",
-              "pages": [
-                "sdk/dynamicschedule/register",
-                "sdk/dynamicschedule/unregister"
-              ]
+              "pages": ["sdk/dynamicschedule/register", "sdk/dynamicschedule/unregister"]
             }
           ]
         },
@@ -666,9 +618,7 @@
     },
     {
       "group": "HTTP Reference",
-      "pages": [
-        "sdk/api-reference/events/create-an-event"
-      ]
+      "pages": ["sdk/api-reference/events/create-an-event"]
     },
     {
       "group": "React SDK",
@@ -683,9 +633,7 @@
     {
       "group": "Overview",
       "version": "v2",
-      "pages": [
-        "examples/introduction"
-      ]
+      "pages": ["examples/introduction"]
     }
   ],
   "footerSocials": {
@@ -693,4 +641,4 @@
     "github": "https://github.com/triggerdotdev",
     "linkedin": "https://www.linkedin.com/company/triggerdotdev"
   }
-}
+}

docs/v3/cli-deploy.mdx

@@ -21,13 +21,16 @@ yarn dlx trigger.dev@beta deploy
 
 </CodeGroup>
 
+<Warning>Will fail in CI if any version mismatches are detected. Ensure everything runs locally first using the [dev](/v3/cli-dev) command and don't bypass the version checks!</Warning>
+
 It performs a few steps to deploy:
 
-1. Typechecks the code.
-2. Compiles and bundles the code.
-3. Checks that [environment variables](/v3/deploy-environment-variables) are set.
-4. Deploys the code to the cloud.
-5. Registers the tasks as a new version in the environment (prod by default).
+1. Optionally updates packages when running locally.
+2. Typechecks the code.
+3. Compiles and bundles the code.
+4. Checks that [environment variables](/v3/deploy-environment-variables) are set.
+5. Deploys the code to the cloud.
+6. Registers the tasks as a new version in the environment (prod by default).
 
 ## Options
 

docs/v3/cli-dev.mdx

@@ -21,6 +21,8 @@ yarn dlx trigger.dev@beta dev
 
 </CodeGroup>
 
+It will first perform an update check to prevent version mismatches, failed deploys, and other errors. You will always be prompted first.
+
 You will see in the terminal that the server is running and listening for requests. When you run a task, you will see it in the terminal along with a link to view it in the dashboard.
 
 It is worth noting that each task runs in a separate Node process. This means that if you have a long-running task, it will not block other tasks from running.

docs/v3/errors-retrying.mdx

@@ -35,7 +35,7 @@ export const myTask = task({
     maxAttempts: 10,
   },
   run: async (payload: string) => {
-    const result = await otherTask.triggerAndWait({ payload: "some data" });
+    const result = await otherTask.triggerAndWait("some data");
     //...do other stuff
   },
 });

docs/v3/feature-matrix.mdx

@@ -14,7 +14,7 @@ description: "What features are currently available in the Developer Preview"
 | [Atomic versioning](/v3/versioning)                                                 | Each deploy creates a new version                  | ✅     |
 | [Deploy via CLI](/v3/cli-deploy)                                                    | Deploy from the command line                       | ✅     |
 | [Deploy via GitHub Actions](/v3/github-actions)                                     | Deploy using GitHub Actions                        | ✅     |
-| [Scheduled tasks](/v3/tasks-scheduled)                                              | A task that can be triggered on a schedule         | ⏳     |
+| [Scheduled tasks](/v3/tasks-scheduled)                                              | A task that can be triggered on a schedule         | ✅     |
 | [Zod tasks](/v3/tasks-zod)                                                          | Define tasks using Zod schemas                     | ⏳     |
 | [Webhook tasks](/v3/tasks-webhooks)                                                 | A task that can be triggered by a webhook          | ⏳     |
 | Full text search of runs                                                            | Find a run by searching the payload and output     | ⏳     |

docs/v3/github-actions.mdx

@@ -5,6 +5,8 @@ description: "You can easily deploy your tasks with GitHub actions."
 
 This simple GitHub action file will deploy you Trigger.dev tasks when new code is pushed to the `main` branch and the `trigger` directory has changes in it.
 
+<Warning>The deploy step will fail if any version mismatches are detected. Please see the [version pinning](/v3/github-actions#version-pinning) section for more details.</Warning>
+
 ```yaml .github/workflows/release-trigger.yml
 name: Deploy to Trigger.dev
 
@@ -42,3 +44,19 @@ If you already have a GitHub action file, you can just add the final step "🚀
 You need to add the `TRIGGER_ACCESS_TOKEN` secret to your repository. You can create a new access token by going to your profile page and then clicking on the "Personal Access Tokens" tab.
 
 To set it in GitHub go to your repository, click on "Settings", "Secrets and variables" and then "Actions". Add a new secret with the name `TRIGGER_ACCESS_TOKEN` and use the value of your access token.
+
+## Version pinning
+
+The CLI and `@trigger.dev/*` package versions need to be in sync, otherwise there will be errors and unpredictable behavior. Hence, the `deploy` command will automatically fail during CI on any version mismatches.
+
+To ensure a smooth CI experience you can pin the CLI version in the deploy step, like so:
+
+```yaml .github/workflows/release-trigger.yml
+- name: 🚀 Deploy Trigger.dev
+  env:
+    TRIGGER_ACCESS_TOKEN: ${{ secrets.TRIGGER_ACCESS_TOKEN }}
+  run: |
+    npx trigger.dev@3.0.0-beta.16 deploy
+```
+
+You should use the version you run locally during dev and manual deploy. The current version is displayed in the banner, but you can also check it by appending `--version` to any command.

docs/v3/known-issues.mdx

@@ -2,11 +2,3 @@
 title: "Known issues"
 description: "Please let us know if you encounter an issue not on this list."
 ---
-
-#### `pnpm` is not currently supported
-
-By default, pnpm has a more complicated package structure than npm. If you see this error when running the CLI then you're experiencing this bug:
-
-```bash
-✘ [ERROR] Could not resolve "@trigger.dev/core/v3"
-```

docs/v3/migration-defer.mdx

@@ -11,7 +11,6 @@ This guide highlights the differences and should help you migrate your project.
 
 Here are some features you might be using in Defer that are coming this month to v3:
 
-- [Scheduled tasks (including CRON)](/v3/tasks-scheduled) will be available in mid-April.
 - Triggering a task with a delay (like `assignOptions` delay in Defer) will be available soon – there is [an alternative](#delay) you can use for now.
 
 You can view the full feature matrix [here](/v3/feature-matrix).
@@ -71,7 +70,7 @@ export async function runLongRunningTask() {
 }
 ```
 
-In Trigger.dev your logic goes in the `run` function of a task. You can then `trigger` and `batchTrigger` that task, with a payload and options.
+In Trigger.dev your logic goes in the `run` function of a task. You can then `trigger` and `batchTrigger` that task, with a payload as the first argument.
 
 ```ts /app/actions/actions.ts
 "use server";
@@ -79,7 +78,7 @@ In Trigger.dev your logic goes in the `run` function of a task. You can then `tr
 import { longRunningTask } from "@/trigger/someTasks";
 
 export async function runLongRunningTask() {
-  return await longRunningTask.trigger({ payload: { foo: "bar" } });
+  return await longRunningTask.trigger({ foo: "bar" });
 }
 ```
 
@@ -243,7 +242,7 @@ export const longRunningTask = task({
 import { longRunningTask } from "@/trigger/longRunningTask";
 
 export async function runLongRunningTask() {
-  return await longRunningTask.trigger({ payload: { foo: "bar" } });
+  return await longRunningTask.trigger({ foo: "bar" });
 }
 ```
 
@@ -256,6 +255,36 @@ export async function runLongRunningTask() {
 
 #### Example 2: A CRON task
 
-<Warning>
-  "Scheduled" tasks will be available by mid-April. This will allow you to replace Defer CRON tasks.
-</Warning>
+We call these [scheduled tasks](/v3/tasks-scheduled) in Trigger.dev.
+
+In Defer you might have a function like this:
+
+```ts
+import { defer } from "@defer/client";
+
+async function sendMondayNewletter() {
+  // business logic here
+}
+
+export default defer.cron(sendMondayNewletter, "0 0 * * 1");
+```
+
+In Trigger.dev the task looks like this:
+
+```ts
+import { schedules } from "@trigger.dev/sdk/v3";
+
+//this task will run when any of the attached schedules trigger
+export const sendMondayNewletter = schedules.task({
+  id: "send-monday-newsletter",
+  run: async (payload) => {
+    // business logic here
+  },
+});
+```
+
+Then you need to attach a schedule to the task, either using the dashboard or in your code. You can attach unlimited schedules to a task.
+
+<Card title="Attaching schedules" icon="clock" href="/v3/tasks-scheduled">
+  How to attach a schedule to a task
+</Card>

docs/v3/queue-concurrency.mdx

@@ -107,24 +107,19 @@ export async function POST(request: Request) {
 
   if (data.branch === "main") {
     //trigger the task, with a different queue
-    const handle = await generatePullRequest.trigger({
-      payload: data,
-      options: {
-        queue: {
-          //the "main-branch" queue will have a concurrency limit of 10
-          //this triggered run will use that queue
-          name: "main-branch",
-          concurrencyLimit: 10,
-        },
+    const handle = await generatePullRequest.trigger(data, {
+      queue: {
+        //the "main-branch" queue will have a concurrency limit of 10
+        //this triggered run will use that queue
+        name: "main-branch",
+        concurrencyLimit: 10,
       },
     });
 
     return Response.json(handle);
   } else {
     //triggered with the default (concurrency of 1)
-    const handle = await generatePullRequest.trigger({
-      payload: data,
-    });
+    const handle = await generatePullRequest.trigger(data);
     return Response.json(handle);
   }
 }
@@ -146,32 +141,26 @@ export async function POST(request: Request) {
 
   if (data.isFreeUser) {
     //free users can only have 1 PR generated at a time
-    const handle = await generatePullRequest.trigger({
-      payload: data,
-      options: {
-        queue: {
-          //every free user gets a queue with a concurrency limit of 1
-          name: "free-users",
-          concurrencyLimit: 1,
-        },
-        concurrencyKey: data.userId,
+    const handle = await generatePullRequest.trigger(data, {
+      queue: {
+        //every free user gets a queue with a concurrency limit of 1
+        name: "free-users",
+        concurrencyLimit: 1,
       },
+      concurrencyKey: data.userId,
     });
 
     //return a success response with the handle
     return Response.json(handle);
   } else {
     //trigger the task, with a different queue
-    const handle = await generatePullRequest.trigger({
-      payload: data,
-      options: {
-        queue: {
-          //every paid user gets a queue with a concurrency limit of 10
-          name: "paid-users",
-          concurrencyLimit: 10,
-        },
-        concurrencyKey: data.userId,
+    const handle = await generatePullRequest.trigger(data, {
+      queue: {
+        //every paid user gets a queue with a concurrency limit of 10
+        name: "paid-users",
+        concurrencyLimit: 10,
       },
+      concurrencyKey: data.userId,
     });
 
     //return a success response with the handle