feat(ai): add triggerAndSubscribe method and use it in ai.tool

Replace triggerAndWait with triggerAndSubscribe in ai.tool to fix:
- Parallel tool calls (no more preventMultipleWaits errors)
- Stop signal while suspended (parent stays alive, child gets cancelled)

New task.triggerAndSubscribe() method: trigger + subscribeToRun in a
single span, with abort signal support and cancelOnAbort option.

Convert deepResearch to a schemaTask + ai.tool in the reference app.

refs TRI-7986

Author: ericallam

packages/core/src/v3/types/tasks.ts

@@ -618,6 +618,30 @@ export interface Task<TIdentifier extends string, TInput = void, TOutput = any>
     requestOptions?: TriggerApiRequestOptions
   ) => TaskRunPromise<TIdentifier, TOutput>;
 
+  /**
+   * Trigger a task and subscribe to its updates via realtime. Unlike `triggerAndWait`,
+   * this does NOT suspend the parent run — the parent stays alive and polls for updates.
+   * This enables parallel tool calls and proper abort signal handling.
+   *
+   * @param payload
+   * @param options - Options for the task run, including an optional `signal` to cancel the subscription and child run
+   * @returns TaskRunPromise
+   * @example
+   * ```
+   * const result = await task.triggerAndSubscribe({ foo: "bar" }, { signal: abortSignal });
+   *
+   * if (result.ok) {
+   *   console.log(result.output);
+   * } else {
+   *   console.error(result.error);
+   * }
+   * ```
+   */
+  triggerAndSubscribe: (
+    payload: TInput,
+    options?: TriggerAndSubscribeOptions,
+  ) => TaskRunPromise<TIdentifier, TOutput>;
+
   /**
    * Batch trigger multiple task runs with the given payloads, and wait for the results. Returns the results of the task runs.
    * @param items - Array, AsyncIterable, or ReadableStream of batch items
@@ -966,6 +990,16 @@ export type TriggerOptions = {
 };
 
 export type TriggerAndWaitOptions = Omit<TriggerOptions, "version">;
+
+export type TriggerAndSubscribeOptions = Omit<TriggerOptions, "version"> & {
+  /** An AbortSignal to cancel the subscription. When fired, the subscription closes and the promise rejects. */
+  signal?: AbortSignal;
+  /**
+   * Whether to cancel the child run when the abort signal fires.
+   * @default true
+   */
+  cancelOnAbort?: boolean;
+};
 export type BatchTriggerOptions = {
   /**
    * If no idempotencyKey is set on an individual item in the batch, it will use this key on each item + the array index.

packages/trigger-sdk/src/v3/ai.ts

@@ -147,11 +147,12 @@ function toolFromTask<
       }
 
       return await task
-        .triggerAndWait(input as inferSchemaIn<TTaskSchema>, {
+        .triggerAndSubscribe(input as inferSchemaIn<TTaskSchema>, {
           metadata: {
             [METADATA_KEY]: toolMeta as any,
           },
           tags: options?.toolCallId ? [`toolCallId:${options.toolCallId}`] : undefined,
+          signal: options?.abortSignal,
         })
         .unwrap();
     },

packages/trigger-sdk/src/v3/runs.ts

@@ -358,6 +358,14 @@ export type SubscribeToRunOptions = {
    * ```
    */
   skipColumns?: RealtimeRunSkipColumns;
+
+  /**
+   * An AbortSignal to cancel the subscription.
+   *
+   * When the signal is aborted, the underlying SSE connection is closed
+   * and the async iterator completes.
+   */
+  signal?: AbortSignal;
 };
 
 /**
@@ -403,6 +411,7 @@ function subscribeToRun<TRunId extends AnyRunHandle | AnyTask | string>(
     closeOnComplete:
       typeof options?.stopOnCompletion === "boolean" ? options.stopOnCompletion : true,
     skipColumns: options?.skipColumns,
+    signal: options?.signal,
   });
 }
 

packages/trigger-sdk/src/v3/shared.ts

@@ -90,6 +90,7 @@ import type {
   TaskWithToolOptions,
   ToolTask,
   ToolTaskParameters,
+  TriggerAndSubscribeOptions,
   TriggerAndWaitOptions,
   TriggerApiRequestOptions,
   TriggerOptions,
@@ -214,6 +215,26 @@ export function createTask<
           });
       }, params.id);
     },
+    triggerAndSubscribe: (payload, options) => {
+      return new TaskRunPromise<TIdentifier, TOutput>((resolve, reject) => {
+        triggerAndSubscribe_internal<TIdentifier, TInput, TOutput>(
+          "triggerAndSubscribe()",
+          params.id,
+          payload,
+          undefined,
+          {
+            queue: params.queue?.name,
+            ...options,
+          }
+        )
+          .then((result) => {
+            resolve(result);
+          })
+          .catch((error) => {
+            reject(error);
+          });
+      }, params.id);
+    },
     batchTriggerAndWait: async (items, options) => {
       return await batchTriggerAndWait_internal<TIdentifier, TInput, TOutput>(
         "batchTriggerAndWait()",
@@ -345,6 +366,26 @@ export function createSchemaTask<
           });
       }, params.id);
     },
+    triggerAndSubscribe: (payload, options) => {
+      return new TaskRunPromise<TIdentifier, TOutput>((resolve, reject) => {
+        triggerAndSubscribe_internal<TIdentifier, inferSchemaIn<TSchema>, TOutput>(
+          "triggerAndSubscribe()",
+          params.id,
+          payload,
+          parsePayload,
+          {
+            queue: params.queue?.name,
+            ...options,
+          }
+        )
+          .then((result) => {
+            resolve(result);
+          })
+          .catch((error) => {
+            reject(error);
+          });
+      }, params.id);
+    },
     batchTriggerAndWait: async (items, options) => {
       return await batchTriggerAndWait_internal<TIdentifier, inferSchemaIn<TSchema>, TOutput>(
         "batchTriggerAndWait()",
@@ -463,6 +504,49 @@ export function triggerAndWait<TTask extends AnyTask>(
   }, id);
 }
 
+/**
+ * Trigger a task and subscribe to its updates via realtime. Unlike `triggerAndWait`,
+ * this does NOT suspend the parent run — the parent stays alive and subscribes to updates.
+ * This enables parallel execution and proper abort signal handling.
+ *
+ * @param id - The id of the task to trigger
+ * @param payload
+ * @param options - Options for the task run, including an optional `signal` to cancel the subscription and child run
+ * @returns TaskRunPromise
+ * @example
+ * ```ts
+ * import { tasks } from "@trigger.dev/sdk/v3";
+ * const result = await tasks.triggerAndSubscribe("my-task", { foo: "bar" });
+ *
+ * if (result.ok) {
+ *   console.log(result.output);
+ * } else {
+ *   console.error(result.error);
+ * }
+ * ```
+ */
+export function triggerAndSubscribe<TTask extends AnyTask>(
+  id: TaskIdentifier<TTask>,
+  payload: TaskPayload<TTask>,
+  options?: TriggerAndSubscribeOptions
+): TaskRunPromise<TaskIdentifier<TTask>, TaskOutput<TTask>> {
+  return new TaskRunPromise<TaskIdentifier<TTask>, TaskOutput<TTask>>((resolve, reject) => {
+    triggerAndSubscribe_internal<TaskIdentifier<TTask>, TaskPayload<TTask>, TaskOutput<TTask>>(
+      "tasks.triggerAndSubscribe()",
+      id,
+      payload,
+      undefined,
+      options
+    )
+      .then((result) => {
+        resolve(result);
+      })
+      .catch((error) => {
+        reject(error);
+      });
+  }, id);
+}
+
 /**
  * Batch trigger multiple task runs with the given payloads, and wait for the results. Returns the results of the task runs.
  * @param id - The id of the task to trigger
@@ -2439,6 +2523,128 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
   );
 }
 
+async function triggerAndSubscribe_internal<TIdentifier extends string, TPayload, TOutput>(
+  name: string,
+  id: TIdentifier,
+  payload: TPayload,
+  parsePayload?: SchemaParseFn<TPayload>,
+  options?: TriggerAndSubscribeOptions
+): Promise<TaskRunResult<TIdentifier, TOutput>> {
+  const ctx = taskContext.ctx;
+
+  if (!ctx) {
+    throw new Error("triggerAndSubscribe can only be used from inside a task.run()");
+  }
+
+  const apiClient = apiClientManager.clientOrThrow();
+
+  const parsedPayload = parsePayload ? await parsePayload(payload) : payload;
+  const payloadPacket = await stringifyIO(parsedPayload);
+
+  const processedIdempotencyKey = await makeIdempotencyKey(options?.idempotencyKey);
+  const idempotencyKeyOptions = processedIdempotencyKey
+    ? getIdempotencyKeyOptions(processedIdempotencyKey)
+    : undefined;
+
+  return await tracer.startActiveSpan(
+    name,
+    async (span) => {
+      const response = await apiClient.triggerTask(
+        id,
+        {
+          payload: payloadPacket.data,
+          options: {
+            lockToVersion: taskContext.worker?.version,
+            queue: options?.queue ? { name: options.queue } : undefined,
+            concurrencyKey: options?.concurrencyKey,
+            test: taskContext.ctx?.run.isTest,
+            payloadType: payloadPacket.dataType,
+            delay: options?.delay,
+            ttl: options?.ttl,
+            tags: options?.tags,
+            maxAttempts: options?.maxAttempts,
+            metadata: options?.metadata,
+            maxDuration: options?.maxDuration,
+            parentRunId: ctx.run.id,
+            // NOTE: no resumeParentOnCompletion — parent stays alive and subscribes
+            idempotencyKey: processedIdempotencyKey?.toString(),
+            idempotencyKeyTTL: options?.idempotencyKeyTTL,
+            idempotencyKeyOptions,
+            machine: options?.machine,
+            priority: options?.priority,
+            region: options?.region,
+            debounce: options?.debounce,
+          },
+        },
+        {}
+      );
+
+      // Set attributes after trigger so the dashboard can link to the child run
+      span.setAttribute("messaging.message.id", response.id);
+      span.setAttribute("runId", response.id);
+      span.setAttribute(SemanticInternalAttributes.ENTITY_TYPE, "run");
+      span.setAttribute(SemanticInternalAttributes.ENTITY_ID, response.id);
+
+      // Optionally cancel the child run when the abort signal fires (default: true)
+      const cancelOnAbort = options?.cancelOnAbort !== false;
+      if (options?.signal && cancelOnAbort) {
+        const onAbort = () => {
+          apiClient.cancelRun(response.id).catch(() => {});
+        };
+        if (options.signal.aborted) {
+          await apiClient.cancelRun(response.id).catch(() => {});
+          throw new Error("Aborted");
+        }
+        options.signal.addEventListener("abort", onAbort, { once: true });
+      }
+
+      for await (const run of apiClient.subscribeToRun(response.id, {
+        closeOnComplete: true,
+        signal: options?.signal,
+        skipColumns: ["payload"],
+      })) {
+        if (run.isSuccess) {
+          // run.output from subscribeToRun is already deserialized
+          return {
+            ok: true as const,
+            id: response.id,
+            taskIdentifier: id as TIdentifier,
+            output: run.output as TOutput,
+          };
+        }
+        if (run.isFailed || run.isCancelled) {
+          const error = new Error(run.error?.message ?? `Task ${id} failed (${run.status})`);
+          if (run.error?.name) error.name = run.error.name;
+
+          return {
+            ok: false as const,
+            id: response.id,
+            taskIdentifier: id as TIdentifier,
+            error,
+          };
+        }
+      }
+
+      throw new Error(`Task ${id}: subscription ended without completion`);
+    },
+    {
+      kind: SpanKind.PRODUCER,
+      attributes: {
+        [SemanticInternalAttributes.STYLE_ICON]: "trigger",
+        ...accessoryAttributes({
+          items: [
+            {
+              text: id,
+              variant: "normal",
+            },
+          ],
+          style: "codepath",
+        }),
+      },
+    }
+  );
+}
+
 async function batchTriggerAndWait_internal<TIdentifier extends string, TPayload, TOutput>(
   name: string,
   id: TIdentifier,

packages/trigger-sdk/src/v3/tasks.ts

@@ -20,6 +20,7 @@ import {
   SubtaskUnwrapError,
   trigger,
   triggerAndWait,
+  triggerAndSubscribe,
 } from "./shared.js";
 
 export { SubtaskUnwrapError };
@@ -96,6 +97,7 @@ export const tasks = {
   trigger,
   batchTrigger,
   triggerAndWait,
+  triggerAndSubscribe,
   batchTriggerAndWait,
   /** @deprecated Use onStartAttempt instead */
   onStart,