Merge remote-tracking branch 'origin/main' into fix/testcontainers

Author: nicktrn

.changeset/pre.json

@@ -25,11 +25,14 @@
     "green-lions-relate",
     "hip-cups-wave",
     "honest-files-decide",
+    "itchy-games-sort",
     "late-chairs-ring",
     "moody-squids-count",
     "nice-colts-boil",
     "polite-impalas-care",
     "polite-lies-fix",
+    "real-rats-drop",
+    "red-chairs-begin",
     "red-wasps-cover",
     "shiny-kiwis-beam",
     "smart-coins-hammer",
@@ -42,6 +45,7 @@
     "tricky-houses-invite",
     "two-tigers-dream",
     "weak-jobs-hide",
-    "wet-deers-think"
+    "wet-deers-think",
+    "wet-steaks-reflect"
   ]
 }

.changeset/real-rats-drop.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Add onCancel lifecycle hook

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

@@ -97,6 +97,7 @@ export function RunIcon({ name, className, spanName }: TaskIconProps) {
     case "task-hook-onResume":
     case "task-hook-onComplete":
     case "task-hook-cleanup":
+    case "task-hook-onCancel":
       return <FunctionIcon className={cn(className, "text-text-dimmed")} />;
     case "task-hook-onFailure":
     case "task-hook-catchError":

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

@@ -47,29 +47,6 @@ export class CancelTaskRunService extends BaseService {
       tx: this._prisma,
     });
 
-    const inProgressEvents = await eventRepository.queryIncompleteEvents(
-      getTaskEventStoreTableForRun(taskRun),
-      {
-        runId: taskRun.friendlyId,
-      },
-      taskRun.createdAt,
-      taskRun.completedAt ?? undefined
-    );
-
-    logger.debug("Cancelling in-progress events", {
-      inProgressEvents: inProgressEvents.map((event) => event.id),
-    });
-
-    await Promise.all(
-      inProgressEvents.map((event) => {
-        return eventRepository.cancelEvent(
-          event,
-          options?.cancelledAt ?? new Date(),
-          options?.reason ?? "Run cancelled"
-        );
-      })
-    );
-
     return {
       id: result.run.id,
     };

docs/upgrade-to-v4.mdx

@@ -170,6 +170,107 @@ tasks.onComplete(({ ctx, result }) => {
 });
 ```
 
+### onCancel
+
+<Note>Available in v4.0.0-beta.12 and later.</Note>
+
+You can now define an `onCancel` hook that is called when a run is cancelled. This is useful if you want to clean up any resources that were allocated for the run.
+
+```ts
+tasks.onCancel(({ ctx, signal }) => {
+  console.log("Run cancelled", signal);
+});
+```
+
+You can use the `onCancel` hook along with the `signal` passed into the run function to interrupt a call to an external service, for example using the [streamText](https://ai-sdk.dev/docs/reference/ai-sdk-core/stream-text) function from the AI SDK:
+
+```ts
+import { logger, tasks, schemaTask } from "@trigger.dev/sdk";
+import { streamText } from "ai";
+import { z } from "zod";
+
+export const interruptibleChat = schemaTask({
+  id: "interruptible-chat",
+  description: "Chat with the AI",
+  schema: z.object({
+    prompt: z.string().describe("The prompt to chat with the AI"),
+  }),
+  run: async ({ prompt }, { signal }) => {
+    const chunks: TextStreamPart<{}>[] = [];
+
+    // 👇 This is a global onCancel hook, but it's inside of the run function
+    tasks.onCancel(async () => {
+      // We have access to the chunks here, and can save them to the database
+      await saveChunksToDatabase(chunks);
+    });
+
+    try {
+      const result = streamText({
+        model: getModel(),
+        prompt,
+        experimental_telemetry: {
+          isEnabled: true,
+        },
+        tools: {},
+        abortSignal: signal, // 👈 Pass the signal to the streamText function, which aborts with the run is cancelled
+        onChunk: ({ chunk }) => {
+          chunks.push(chunk);
+        },
+      });
+
+      const textParts = [];
+
+      for await (const part of result.textStream) {
+        textParts.push(part);
+      }
+
+      return textParts.join("");
+    } catch (error) {
+      if (error instanceof Error && error.name === "AbortError") {
+        // streamText will throw an AbortError if the signal is aborted, so we can handle it here
+      } else {
+        throw error;
+      }
+    }
+  },
+});
+```
+
+The `onCancel` hook can optionally wait for the `run` function to finish, and access the output of the run:
+
+```ts
+import { logger, task } from "@trigger.dev/sdk";
+import { setTimeout } from "node:timers/promises";
+
+export const cancelExampleTask = task({
+  id: "cancel-example",
+  // Signal will be aborted when the task is cancelled 👇
+  run: async (payload: { message: string }, { signal }) => {
+    try {
+      // We pass the signal to setTimeout to abort the timeout if the task is cancelled
+      await setTimeout(10_000, undefined, { signal });
+    } catch (error) {
+      // Ignore the abort error
+    }
+
+    // Do some more work here
+
+    return {
+      message: "Hello, world!",
+    };
+  },
+  onCancel: async ({ runPromise }) => {
+    // You can await the runPromise to get the output of the task
+    const output = await runPromise;
+  },
+});
+```
+
+<Note>
+  You will have up to 30 seconds to complete the `runPromise` in the `onCancel` hook. After that
+  point the process will be killed.
+</Note>
+
 ### Improved middleware and locals
 
 Our task middleware system is now much more useful. Previously it only ran "around" the `run` function, but now we've hoisted it to the top level and it now runs before/after all the other hooks.
@@ -704,7 +805,7 @@ export const myTask = task({
   id: "my-task",
   onStart: ({ payload, ctx }) => {},
   // The run function still uses separate parameters
-  run: async ( payload, { ctx }) => {},
+  run: async (payload, { ctx }) => {},
 });
 ```
 
@@ -760,4 +861,4 @@ const batchHandle = await tasks.batchTrigger([
 // Now you need to call runs.list()
 const runs = await batchHandle.runs.list();
 console.log(runs);
-```
+```

packages/build/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @trigger.dev/build
 
+## 4.0.0-v4-beta.12
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@4.0.0-v4-beta.12`
+
 ## 4.0.0-v4-beta.11
 
 ### Patch Changes

packages/build/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/build",
-  "version": "4.0.0-v4-beta.11",
+  "version": "4.0.0-v4-beta.12",
   "description": "trigger.dev build extensions",
   "license": "MIT",
   "publishConfig": {
@@ -69,7 +69,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:4.0.0-v4-beta.11",
+    "@trigger.dev/core": "workspace:4.0.0-v4-beta.12",
     "pkg-types": "^1.1.3",
     "tinyglobby": "^0.2.2",
     "tsconfck": "3.1.3"

packages/cli-v3/CHANGELOG.md

@@ -1,5 +1,16 @@
 # trigger.dev
 
+## 4.0.0-v4-beta.12
+
+### Patch Changes
+
+- Display clickable links in Cursor terminal ([#1998](https://github.com/triggerdotdev/trigger.dev/pull/1998))
+- Added AI assistance link when you have build errors ([#1925](https://github.com/triggerdotdev/trigger.dev/pull/1925))
+- If you pass a directory when calling deploy we validate it exists and give helpful hints ([#2013](https://github.com/triggerdotdev/trigger.dev/pull/2013))
+- Updated dependencies:
+  - `@trigger.dev/build@4.0.0-v4-beta.12`
+  - `@trigger.dev/core@4.0.0-v4-beta.12`
+
 ## 4.0.0-v4-beta.11
 
 ### Patch Changes

packages/cli-v3/package.json

@@ -1,6 +1,6 @@
 {
   "name": "trigger.dev",
-  "version": "4.0.0-v4-beta.11",
+  "version": "4.0.0-v4-beta.12",
   "description": "A Command-Line Interface for Trigger.dev (v3) projects",
   "type": "module",
   "license": "MIT",
@@ -89,8 +89,8 @@
     "@opentelemetry/sdk-trace-base": "1.25.1",
     "@opentelemetry/sdk-trace-node": "1.25.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
-    "@trigger.dev/build": "workspace:4.0.0-v4-beta.11",
-    "@trigger.dev/core": "workspace:4.0.0-v4-beta.11",
+    "@trigger.dev/build": "workspace:4.0.0-v4-beta.12",
+    "@trigger.dev/core": "workspace:4.0.0-v4-beta.12",
     "ansi-escapes": "^7.0.0",
     "c12": "^1.11.1",
     "chalk": "^5.2.0",

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -23,6 +23,7 @@ import {
   TaskRunExecution,
   timeout,
   TriggerConfig,
+  UsageMeasurement,
   waitUntil,
   WorkerManifest,
   WorkerToExecutorMessageCatalog,
@@ -232,7 +233,10 @@ async function bootstrap() {
 
 let _execution: TaskRunExecution | undefined;
 let _isRunning = false;
+let _isCancelled = false;
 let _tracingSDK: TracingSDK | undefined;
+let _executionMeasurement: UsageMeasurement | undefined;
+const cancelController = new AbortController();
 
 const zodIpc = new ZodIpcConnection({
   listenSchema: WorkerToExecutorMessageCatalog,
@@ -403,18 +407,17 @@ const zodIpc = new ZodIpcConnection({
             getNumberEnvVar("TRIGGER_RUN_METADATA_FLUSH_INTERVAL", 1000)
           );
 
-          const measurement = usage.start();
+          _executionMeasurement = usage.start();
 
-          // This lives outside of the executor because this will eventually be moved to the controller level
-          const signal = execution.run.maxDuration
-            ? timeout.abortAfterTimeout(execution.run.maxDuration)
-            : undefined;
+          const timeoutController = timeout.abortAfterTimeout(execution.run.maxDuration);
+
+          const signal = AbortSignal.any([cancelController.signal, timeoutController.signal]);
 
           const { result } = await executor.execute(execution, metadata, traceContext, signal);
 
-          const usageSample = usage.stop(measurement);
+          if (_isRunning && !_isCancelled) {
+            const usageSample = usage.stop(_executionMeasurement);
 
-          if (_isRunning) {
             return sender.send("TASK_RUN_COMPLETED", {
               execution,
               result: {
@@ -458,7 +461,16 @@ const zodIpc = new ZodIpcConnection({
     WAIT_COMPLETED_NOTIFICATION: async () => {
       await managedWorkerRuntime.completeWaitpoints([]);
     },
-    FLUSH: async ({ timeoutInMs }, sender) => {
+    CANCEL: async ({ timeoutInMs }) => {
+      _isCancelled = true;
+      cancelController.abort("run cancelled");
+      await callCancelHooks(timeoutInMs);
+      if (_executionMeasurement) {
+        usage.stop(_executionMeasurement);
+      }
+      await flushAll(timeoutInMs);
+    },
+    FLUSH: async ({ timeoutInMs }) => {
       await flushAll(timeoutInMs);
     },
     WAITPOINT_CREATED: async ({ wait, waitpoint }) => {
@@ -470,6 +482,18 @@ const zodIpc = new ZodIpcConnection({
   },
 });
 
+async function callCancelHooks(timeoutInMs: number = 10_000) {
+  const now = performance.now();
+
+  try {
+    await Promise.race([lifecycleHooks.callOnCancelHookListeners(), setTimeout(timeoutInMs)]);
+  } finally {
+    const duration = performance.now() - now;
+
+    log(`Called cancel hooks in ${duration}ms`);
+  }
+}
+
 async function flushAll(timeoutInMs: number = 10_000) {
   const now = performance.now();