feat(workflows): support opt-in background runs

packages/cli/src/ui/hooks/useGeminiStream.test.tsx

@@ -206,6 +206,7 @@ describe('useGeminiStream', () => {
   let mockCancelAllToolCalls: Mock;
   let mockMarkToolsAsSubmitted: Mock;
   let mockBackgroundShellRegistry: { setNotificationCallback: Mock };
+  let mockWorkflowRunRegistry: { setCompletionCallback: Mock };
   let mockMonitorRegistry: {
     setNotificationCallback: Mock;
     get: Mock;
@@ -242,6 +243,9 @@ describe('useGeminiStream', () => {
     mockBackgroundShellRegistry = {
       setNotificationCallback: vi.fn(),
     };
+    mockWorkflowRunRegistry = {
+      setCompletionCallback: vi.fn(),
+    };
     mockMonitorRegistry = {
       setNotificationCallback: vi.fn(),
       get: vi.fn().mockReturnValue({ status: 'running' }),
@@ -300,6 +304,7 @@ describe('useGeminiStream', () => {
       })),
       getBackgroundShellRegistry: vi.fn(() => mockBackgroundShellRegistry),
       getMonitorRegistry: vi.fn(() => mockMonitorRegistry),
+      getWorkflowRunRegistry: vi.fn(() => mockWorkflowRunRegistry),
     } as unknown as Config;
     mockOnDebugMessage = vi.fn();
     mockHandleSlashCommand = vi.fn().mockResolvedValue(false);
@@ -472,6 +477,48 @@ describe('useGeminiStream', () => {
     });
   });
 
+  it('queues background workflow completions for the model loop', async () => {
+    const { mockSendMessageStream } = renderTestHook();
+    const displayText = 'Background workflow "research" completed.';
+    const modelText =
+      '<task-notification>\n<kind>workflow</kind>\n<status>completed</status>\n</task-notification>';
+
+    await waitFor(() => {
+      expect(
+        mockWorkflowRunRegistry.setCompletionCallback,
+      ).toHaveBeenCalledWith(expect.any(Function));
+    });
+    const callback = mockWorkflowRunRegistry.setCompletionCallback.mock
+      .calls[0][0] as (
+      displayText: string,
+      modelText: string,
+      meta: { todoWorkChainId?: string },
+    ) => void;
+
+    act(() => {
+      callback(displayText, modelText, { todoWorkChainId: 'workflow-chain' });
+    });
+
+    await waitFor(() => {
+      expect(mockAddItem).toHaveBeenCalledWith(
+        { type: 'notification', text: displayText },
+        expect.any(Number),
+      );
+    });
+    await waitFor(() => {
+      expect(mockSendMessageStream).toHaveBeenCalledWith(
+        modelText,
+        expect.any(AbortSignal),
+        expect.any(String),
+        expect.objectContaining({
+          type: SendMessageType.Notification,
+          notificationDisplayText: displayText,
+          todoWorkChainId: 'workflow-chain',
+        }),
+      );
+    });
+  });
+
   it('forwards submitted prompt provenance only for UserQuery', async () => {
     const { result, mockSendMessageStream } = renderTestHook();
 

packages/cli/src/ui/hooks/useGeminiStream.ts

@@ -3978,6 +3978,24 @@ export const useGeminiStream = (
     };
   }, [config]);
 
+  // Register background workflow completions onto the shared queue. The
+  // registry keeps this separate from its terminal-bell subscriber.
+  useEffect(() => {
+    const registry = config.getWorkflowRunRegistry();
+    registry.setCompletionCallback((displayText, modelText, meta) => {
+      notificationQueueRef.current.push({
+        displayText,
+        modelText,
+        sendMessageType: SendMessageType.Notification,
+        todoWorkChainId: meta.todoWorkChainId,
+      });
+      setNotificationTrigger((n) => n + 1);
+    });
+    return () => {
+      registry.setCompletionCallback(undefined);
+    };
+  }, [config]);
+
   // Register monitor notification callback onto the shared queue.
   useEffect(() => {
     const registry = config.getMonitorRegistry();

packages/core/src/agents/runtime/workflow-runner.test.ts

@@ -5,6 +5,7 @@
  */
 
 import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { getEventListeners } from 'node:events';
 import type { Config } from '../../config/config.js';
 import { WorkflowRunRegistry } from '../workflow-run-registry.js';
 import { AgentEventEmitter } from './agent-events.js';
@@ -91,6 +92,7 @@ describe('WorkflowRunner', () => {
       signal: new AbortController().signal,
       script: 'return await agent("work")',
       args: undefined,
+      runInBackground: true,
     });
     await expect(productionHandle.completion).resolves.toMatchObject({
       ok: true,
@@ -213,6 +215,103 @@ describe('WorkflowRunner', () => {
     expect(logWorkflowRunMock).toHaveBeenCalledTimes(2);
   });
 
+  it('keeps background runs alive after the caller turn ends', async () => {
+    const { config, registry } = configWithRegistry();
+    const observed = observeSettlement(registry);
+    const caller = new AbortController();
+    let resolveDispatch: ((value: string) => void) | undefined;
+    const handle = await WorkflowRunner.start({
+      config,
+      signal: caller.signal,
+      script: 'return await agent("work")',
+      args: undefined,
+      runInBackground: true,
+      dispatch: () =>
+        new Promise<string>((resolve) => {
+          resolveDispatch = resolve;
+        }),
+    });
+    await vi.waitFor(() => expect(resolveDispatch).toBeDefined());
+
+    caller.abort();
+    expect(observed.abortCount()).toBe(0);
+    expect(registry.get(handle.runId)?.status).toBe('running');
+
+    resolveDispatch?.('done');
+    await expect(handle.completion).resolves.toMatchObject({ ok: true });
+    expect(registry.get(handle.runId)?.status).toBe('completed');
+    expect(observed.terminalStatuses).toEqual(['completed']);
+    expect(observed.abortCount()).toBe(1);
+  });
+
+  it('rejects a concurrent resume while the original run is active', async () => {
+    const { config, registry } = configWithRegistry();
+    const runId = 'wf_1234abcd';
+    let resolveDispatch: ((value: string) => void) | undefined;
+    const original = await WorkflowRunner.start({
+      config,
+      signal: new AbortController().signal,
+      script: 'return await agent("original")',
+      args: undefined,
+      resumeFromRunId: runId,
+      runInBackground: true,
+      dispatch: () =>
+        new Promise<string>((resolve) => {
+          resolveDispatch = resolve;
+        }),
+    });
+    await vi.waitFor(() => expect(resolveDispatch).toBeDefined());
+    const replacementCaller = new AbortController();
+    const replacementDispatch = vi.fn(async () => 'replacement');
+
+    try {
+      await expect(
+        WorkflowRunner.start({
+          config,
+          signal: replacementCaller.signal,
+          script: 'return await agent("replacement")',
+          args: undefined,
+          resumeFromRunId: runId,
+          dispatch: replacementDispatch,
+        }),
+      ).rejects.toThrow(/already active/);
+      expect(registry.getHandle(runId)).toBe(original);
+      expect(replacementDispatch).not.toHaveBeenCalled();
+      expect(getEventListeners(replacementCaller.signal, 'abort')).toHaveLength(
+        0,
+      );
+    } finally {
+      resolveDispatch?.('original');
+      await original.completion;
+    }
+
+    expect(registry.get(runId)?.result).toBe('original');
+  });
+
+  it('classifies a background failure after caller abort as failed', async () => {
+    const { config, registry } = configWithRegistry();
+    const caller = new AbortController();
+    let rejectDispatch: ((error: Error) => void) | undefined;
+    const handle = await WorkflowRunner.start({
+      config,
+      signal: caller.signal,
+      script: 'return await agent("work")',
+      args: undefined,
+      runInBackground: true,
+      dispatch: () =>
+        new Promise<string>((_resolve, reject) => {
+          rejectDispatch = reject;
+        }),
+    });
+    await vi.waitFor(() => expect(rejectDispatch).toBeDefined());
+
+    caller.abort();
+    rejectDispatch?.(new Error('background boom'));
+
+    await expect(handle.completion).resolves.toMatchObject({ ok: false });
+    expect(registry.get(handle.runId)?.status).toBe('failed');
+  });
+
   it('routes registry cancellation through each live handle', async () => {
     const cancelCases: Array<{
       cancel: (registry: WorkflowRunRegistry, runId: string) => void;
@@ -234,6 +333,7 @@ describe('WorkflowRunner', () => {
         signal: new AbortController().signal,
         script: 'return await agent("work")',
         args: undefined,
+        runInBackground: true,
         dispatch: () =>
           new Promise<string>((_resolve, reject) => {
             rejectDispatch = reject;
@@ -272,6 +372,7 @@ describe('WorkflowRunner', () => {
         signal: new AbortController().signal,
         script: 'await new Promise(() => {})',
         args: undefined,
+        runInBackground: true,
         dispatch: async () => 'unused',
       });
 

packages/core/src/agents/runtime/workflow-runner.ts

@@ -8,7 +8,10 @@ import { randomBytes } from 'node:crypto';
 import type { Config } from '../../config/config.js';
 import { logWorkflowRun } from '../../telemetry/loggers.js';
 import { WorkflowRunEvent } from '../../telemetry/types.js';
-import { createChildAbortController } from '../../utils/abortController.js';
+import {
+  createAbortController,
+  createChildAbortController,
+} from '../../utils/abortController.js';
 import {
   type WorkflowRunRegistry,
   type WorkflowTask,
@@ -35,6 +38,7 @@ export interface WorkflowRunnerOptions {
   resumeFromRunId?: string;
   dispatch?: WorkflowAgentDispatch;
   onUpdate?: (entry: WorkflowTask) => void;
+  runInBackground?: boolean;
 }
 
 export type WorkflowRunSettlement =
@@ -64,6 +68,7 @@ export class WorkflowRunner {
     options: WorkflowRunnerOptions,
   ): Promise<WorkflowRunHandle> {
     const config = options.config;
+    const runInBackground = options.runInBackground === true;
     const budget = WorkflowBudgetImpl.fromEnv();
     const loaded =
       options.scriptPath && options.script === undefined
@@ -83,7 +88,13 @@ export class WorkflowRunner {
     const resumeReplay: JournalReplay | undefined = options.resumeFromRunId
       ? await journal?.load()
       : undefined;
-    const controller = createChildAbortController(options.signal);
+    if (runInBackground && options.signal.aborted) {
+      throw new Error('Background workflow start was cancelled.');
+    }
+    const callerWasAbortedBeforeStart = options.signal.aborted;
+    const controller = runInBackground
+      ? createAbortController()
+      : createChildAbortController(options.signal);
     const registry = config.getWorkflowRunRegistry?.();
     const dispatch =
       options.dispatch ??
@@ -96,17 +107,24 @@ export class WorkflowRunner {
           : undefined,
       );
     const orchestrator = new WorkflowOrchestrator(dispatch);
-    const entry = registry?.register({
-      runId,
-      meta: null,
-      status: 'running',
-      startTime: Date.now(),
-      outputFile: '',
-      abortController: controller,
-      tokenBudgetTotal: budget.total,
-      script,
-      scriptPath,
-    });
+    let entry: WorkflowTask | undefined;
+    try {
+      entry = registry?.register({
+        runId,
+        meta: null,
+        status: 'running',
+        startTime: Date.now(),
+        outputFile: '',
+        abortController: controller,
+        tokenBudgetTotal: budget.total,
+        script,
+        scriptPath,
+        isBackgrounded: runInBackground,
+      });
+    } catch (error) {
+      controller.abort();
+      throw error;
+    }
     const emitUpdate = (): void => {
       if (!entry || !options.onUpdate) return;
       try {
@@ -172,7 +190,11 @@ export class WorkflowRunner {
           const message = extractErrorMessage(error);
           if (entry && details?.meta && !entry.meta) entry.meta = details.meta;
           if (details?.logs) registry?.setRecentLogs(runId, details.logs);
-          if (options.signal.aborted) {
+          if (
+            callerWasAbortedBeforeStart ||
+            (!runInBackground && options.signal.aborted) ||
+            entry?.status === 'cancelled'
+          ) {
             registry?.cancel(runId, Date.now());
           } else {
             registry?.fail(runId, message, Date.now());

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

@@ -37,13 +37,11 @@
  * they have a separate lifecycle and their inclusion is deferred to a
  * follow-up.
  *
- * `workflow` (P4b) is registered/observed via `WorkflowRunRegistry` and
- * differs from the others in that the registry NEVER emits a
- * `<task-notification>` envelope — `WorkflowTool` already returns its
- * own llmContent + returnDisplay payload to the model on terminal, so
- * a second envelope would duplicate the signal. The kind is widened
- * here so the UI surfaces (pill / dialog / detail body) can switch on
- * `entry.kind === 'workflow'`.
+ * `workflow` (P4b) is registered/observed via `WorkflowRunRegistry`.
+ * Foreground workflows return through their normal tool result; background
+ * workflows emit one terminal notification through a separate completion
+ * channel. The kind is widened here so the UI surfaces (pill / dialog /
+ * detail body) can switch on `entry.kind === 'workflow'`.
  */
 export type TaskKind = 'agent' | 'shell' | 'monitor' | 'workflow';