fix: cancel Composio requests with agent turns

compose.yml

@@ -21,6 +21,7 @@ services:
       VOICE_API_KEY: ${VOICE_API_KEY:-}
       VOICE_MODEL: ${VOICE_MODEL:-}
       VOICE_NAME: ${VOICE_NAME:-marin}
+      COMPOSIO_API_KEY: ${COMPOSIO_API_KEY:-}
       SLACK_CHANNEL_NAME: ${SLACK_CHANNEL_NAME:-}
       SLACK_TEAM_ID: ${SLACK_TEAM_ID:-}
       SLACK_USER_IDS: ${SLACK_USER_IDS:-}

src/server/composio-tools.ts

@@ -3,9 +3,29 @@ import type { composioSession } from './composio.js';
 import { safeFailure } from './slack-channel.js';
 
 type Session = Awaited<ReturnType<typeof composioSession>>;
+type CancellableSession = Session & {
+  tools(
+    modifiers: Parameters<Session['tools']>[0],
+    requestOptions: { signal?: AbortSignal },
+  ): ReturnType<Session['tools']>;
+  execute(
+    toolSlug: string,
+    input: Record<string, unknown>,
+    options: Parameters<Session['execute']>[2],
+    requestOptions: { signal?: AbortSignal },
+  ): ReturnType<Session['execute']>;
+};
 
-export async function composioTools(session: Session, check: () => void) {
-  const tools = await session.tools();
+export async function composioTools(
+  session: Session,
+  check: () => void,
+  signal?: AbortSignal,
+) {
+  check();
+  // @composio/core 0.22 supports request options at runtime but omits them
+  // from the public Session function aliases.
+  const cancellable = session as CancellableSession;
+  const tools = await cancellable.tools(undefined, { signal });
   return tools
     .filter((tool) => tool.type === 'function')
     .map(({ function: fn }) =>
@@ -16,9 +36,11 @@ export async function composioTools(session: Session, check: () => void) {
         execute: async (input: unknown) => {
           check();
           try {
-            const { data, error } = await session.execute(
+            const { data, error } = await cancellable.execute(
               fn.name,
               input as Record<string, unknown>,
+              undefined,
+              { signal },
             );
             return { data, error };
           } catch (error) {

src/server/composio.ts

@@ -20,6 +20,7 @@ export async function composioSession(
   workspace: WorkspaceStore,
   dot: Dot,
   ownerId: string,
+  signal?: AbortSignal,
 ) {
   if (!config.composioApiKey) throw new Error('Composio is not configured.');
   const toolkits = [...new Set(dot.composioToolkits)].sort();
@@ -29,12 +30,16 @@ export async function composioSession(
     .digest('hex');
   const composio = composioClient(config.composioApiKey);
   if (dot.composioSessionId && dot.composioPolicyHash === policyHash)
-    return composio.sessions.use(dot.composioSessionId);
-  const session = await composio.sessions.create(ownerId, {
-    ...(toolkits.length ? { toolkits } : {}),
-    manageConnections: true,
-    instant: false,
-  });
+    return composio.sessions.use(dot.composioSessionId, undefined, { signal });
+  const session = await composio.sessions.create(
+    ownerId,
+    {
+      ...(toolkits.length ? { toolkits } : {}),
+      manageConnections: true,
+      instant: false,
+    },
+    { signal },
+  );
   workspace.saveComposioSession(dot.id, session.sessionId, policyHash);
   return session;
 }

src/server/dot-agent.ts

@@ -67,24 +67,24 @@ export class DotAgent extends AbstractAgent {
       this.controller = controller;
       let subscription: { unsubscribe(): void } | undefined;
       let watcher: ReturnType<typeof setInterval> | undefined;
-      let timedOut = false;
       let finished = false;
       let configurationFailure = false;
       const turnTimeLimitMs =
         this.config.agentRunTimeoutMs ?? DEFAULT_AGENT_RUN_TIMEOUT_MS;
       const observe = answerObserver((event) =>
         this.setupTelemetry?.capture(event),
       );
-      const timeout = setTimeout(() => {
-        timedOut = true;
-        observe({ type: EventType.RUN_ERROR });
-        this.abortRun();
-      }, turnTimeLimitMs);
       const timeLimitError = () => ({
         type: EventType.RUN_ERROR,
         message: `This turn reached the ${turnTimeLimitMs / 1000} second time limit and was stopped. Try a smaller request.`,
       });
+      const timeout = setTimeout(() => {
+        observe({ type: EventType.RUN_ERROR });
+        subscriber.next(this.channel ? channelError() : timeLimitError());
+        subscriber.complete();
+      }, turnTimeLimitMs);
       const fail = (error: unknown) => {
+        if (subscriber.closed) return;
         if (!configurationFailure) observe({ type: EventType.RUN_ERROR });
         subscriber.next(
           this.channel
@@ -346,8 +346,10 @@ export class DotAgent extends AbstractAgent {
                     this.workspace,
                     dot,
                     this.workspace.ownerId,
+                    controller.signal,
                   ),
                   check,
+                  controller.signal,
                 )
               : [];
           check();
@@ -444,20 +446,11 @@ export class DotAgent extends AbstractAgent {
                 if (this.channel) {
                   subscriber.next(channelError());
                   subscriber.complete();
-                } else if (timedOut && !finished) {
-                  subscriber.next(timeLimitError());
-                  subscriber.complete();
                 } else subscriber.error(error);
               },
               complete: () => {
                 if (controller.signal.aborted && !finished)
                   observe({ type: EventType.RUN_ERROR });
-                if (timedOut && !finished) {
-                  observe({ type: EventType.RUN_ERROR });
-                  subscriber.next(
-                    this.channel ? channelError() : timeLimitError(),
-                  );
-                }
                 subscriber.complete();
               },
             });

tests/agent-run-timeout.test.ts

@@ -13,6 +13,11 @@ import { Runner } from '../src/server/runner.js';
 import { Store } from '../src/server/store.js';
 import { WorkspaceStore } from '../src/server/workspace.js';
 
+const composio = vi.hoisted(() => ({ session: vi.fn() }));
+vi.mock('../src/server/composio.js', () => ({
+  composioSession: composio.session,
+}));
+
 const resources: Array<{ close(): void }> = [];
 function sharedStores() {
   const dir = mkdtempSync(join(tmpdir(), 'opendots-timeout-'));
@@ -26,6 +31,7 @@ function sharedStores() {
 }
 afterEach(() => {
   vi.restoreAllMocks();
+  composio.session.mockReset();
   vi.useRealTimers();
   resources.splice(0).forEach((resource) => resource.close());
 });
@@ -44,12 +50,14 @@ it('rejects invalid timeout overrides at startup', () => {
     ).toThrow(/AGENT_RUN_TIMEOUT_MS/);
 });
 
-it('uses the configured timeout for interactive Dot turns', async () => {
+it('finishes an interactive turn when Composio startup stalls', async () => {
   vi.useFakeTimers();
   const store = new Store(':memory:');
   const workspace = new WorkspaceStore(':memory:', 'owner');
   resources.push(store, workspace);
   const dot = workspace.dots()[0];
+  store.updateSettings({ appsAllowed: true });
+  workspace.updateDot(dot.id, { ...dot, appsAllowed: true });
   workspace.bindThread('thread', dot.id, 'Timeout test');
   const agent = new DotAgent(
     store,
@@ -58,6 +66,7 @@ it('uses the configured timeout for interactive Dot turns', async () => {
       intelligenceKey: 'fixture',
       apiKey: 'fixture',
       model: 'custom-model',
+      composioApiKey: 'fixture',
       baseUrl: 'https://unused.invalid/v1',
       runtimeUrl: '',
       voiceName: 'marin',
@@ -75,27 +84,73 @@ it('uses the configured timeout for interactive Dot turns', async () => {
     tools: [],
     forwardedProps: {},
   };
-  vi.spyOn(globalThis, 'fetch').mockImplementation(
-    (_url, init) =>
-      new Promise((_resolve, reject) => {
-        init?.signal?.addEventListener(
-          'abort',
-          () => reject(init.signal?.reason),
-          { once: true },
-        );
-      }),
-  );
+  composio.session.mockReturnValue(new Promise(() => {}));
+
   const finished = lastValueFrom(agent.run(input).pipe(toArray()));
   await vi.advanceTimersByTimeAsync(1_251);
   const events = await finished;
-  expect(events).toEqual(
-    expect.arrayContaining([
-      expect.objectContaining({
-        type: EventType.RUN_ERROR,
-        message: expect.stringContaining('1.25 second time limit'),
-      }),
-    ]),
+
+  expect(events).toEqual([
+    expect.objectContaining({
+      type: EventType.RUN_ERROR,
+      message: expect.stringContaining('1.25 second time limit'),
+    }),
+  ]);
+  const signal = composio.session.mock.calls[0][4] as AbortSignal;
+  expect(signal.aborted).toBe(true);
+});
+
+it('does not start the inner agent after Composio startup is cancelled', async () => {
+  const store = new Store(':memory:');
+  const workspace = new WorkspaceStore(':memory:', 'owner');
+  resources.push(store, workspace);
+  const dot = workspace.dots()[0];
+  store.updateSettings({ appsAllowed: true });
+  workspace.updateDot(dot.id, { ...dot, appsAllowed: true });
+  workspace.bindThread('thread', dot.id, 'Cancellation test');
+  const agent = new DotAgent(
+    store,
+    workspace,
+    {
+      intelligenceKey: 'fixture',
+      apiKey: 'fixture',
+      model: 'custom-model',
+      composioApiKey: 'fixture',
+      baseUrl: 'https://unused.invalid/v1',
+      runtimeUrl: '',
+      voiceName: 'marin',
+      slackUsers: [],
+    },
+    dot.id,
   );
+  let resolveSession!: (session: unknown) => void;
+  composio.session.mockReturnValue(
+    new Promise((resolve) => {
+      resolveSession = resolve;
+    }),
+  );
+  const network = vi.spyOn(globalThis, 'fetch');
+  const finished = lastValueFrom(
+    agent
+      .run({
+        threadId: 'thread',
+        runId: 'run',
+        state: {},
+        context: [],
+        messages: [{ id: 'user', role: 'user', content: 'Keep working.' }],
+        tools: [],
+        forwardedProps: {},
+      })
+      .pipe(toArray()),
+  );
+
+  const tools = vi.fn().mockResolvedValue([]);
+  agent.abortRun();
+  resolveSession({ tools });
+  await finished;
+
+  expect(tools).not.toHaveBeenCalled();
+  expect(network).not.toHaveBeenCalled();
 });
 
 it('uses the configured timeout for scheduled/background runs', async () => {

tests/composio-tools.test.ts

@@ -14,17 +14,21 @@ it('passes Composio JSON Schema through and returns only safe execution failures
     .fn()
     .mockResolvedValueOnce({ data: { ok: true }, error: null })
     .mockRejectedValueOnce(new Error('provider secret'));
+  const tools = vi.fn().mockResolvedValue([
+    {
+      type: 'function',
+      function: { name: 'COMPOSIO_TEST', description: 'Test', parameters },
+    },
+  ]);
   const session = {
-    tools: vi.fn().mockResolvedValue([
-      {
-        type: 'function',
-        function: { name: 'COMPOSIO_TEST', description: 'Test', parameters },
-      },
-    ]),
+    tools,
     execute,
   } as unknown as Session;
   const check = vi.fn();
-  const [tool] = await composioTools(session, check);
+  const signal = new AbortController().signal;
+  const [tool] = await composioTools(session, check, signal);
+
+  expect(tools).toHaveBeenCalledWith(undefined, { signal });
 
   expect(tool.parameters).toBe(parameters);
   await expect(tool.execute?.({ query: 'one' })).resolves.toEqual({
@@ -35,5 +39,11 @@ it('passes Composio JSON Schema through and returns only safe execution failures
     data: {},
     error: 'Composio tool failed: Error.',
   });
-  expect(check).toHaveBeenCalledTimes(2);
+  expect(execute).toHaveBeenCalledWith(
+    'COMPOSIO_TEST',
+    expect.anything(),
+    undefined,
+    { signal },
+  );
+  expect(check).toHaveBeenCalledTimes(3);
 });

tests/composio.test.ts

@@ -42,24 +42,36 @@ it('creates a non-Instant session and reuses it while the policy is unchanged',
   const created = { sessionId: 'session-one' };
   sdk.create.mockResolvedValue(created);
   sdk.use.mockResolvedValue(created);
+  const signal = new AbortController().signal;
 
   await expect(
-    composioSession(config, workspace, workspace.dot(original.id)!, 'owner'),
+    composioSession(
+      config,
+      workspace,
+      workspace.dot(original.id)!,
+      'owner',
+      signal,
+    ),
   ).resolves.toBe(created);
-  expect(sdk.create).toHaveBeenCalledWith('owner', {
-    toolkits: ['gmail'],
-    manageConnections: true,
-    instant: false,
-  });
+  expect(sdk.create).toHaveBeenCalledWith(
+    'owner',
+    {
+      toolkits: ['gmail'],
+      manageConnections: true,
+      instant: false,
+    },
+    { signal },
+  );
 
   await composioSession(
     config,
     workspace,
     workspace.dot(original.id)!,
     'owner',
+    signal,
   );
   expect(sdk.create).toHaveBeenCalledTimes(1);
-  expect(sdk.use).toHaveBeenCalledWith('session-one');
+  expect(sdk.use).toHaveBeenCalledWith('session-one', undefined, { signal });
 });
 
 it('creates a new session when the toolkit policy changes', async () => {