fix: bound Composio startup and forward Compose API key

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

@@ -1,11 +1,26 @@
 import { defineTool, type ToolDefinition } from '@copilotkit/runtime/v2';
+import type { ComposioRequestOptions } from '@composio/core';
 import type { composioSession } from './composio.js';
 import { safeFailure } from './slack-channel.js';
 
 type Session = Awaited<ReturnType<typeof composioSession>>;
+type SessionWithRequestOptions = Session & {
+  tools: (
+    modifiers?: Parameters<Session['tools']>[0],
+    requestOptions?: ComposioRequestOptions,
+  ) => ReturnType<Session['tools']>;
+};
 
-export async function composioTools(session: Session, check: () => void) {
-  const tools = await session.tools();
+export async function composioTools(
+  session: Session,
+  check: () => void,
+  signal: AbortSignal,
+) {
+  // The concrete SDK method accepts request options, but its Session interface
+  // currently exposes only the modifiers argument.
+  const tools = await (session as SessionWithRequestOptions).tools(undefined, {
+    signal,
+  });
   return tools
     .filter((tool) => tool.type === 'function')
     .map(({ function: fn }) =>

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

@@ -69,6 +69,7 @@ export class DotAgent extends AbstractAgent {
       let watcher: ReturnType<typeof setInterval> | undefined;
       let timedOut = false;
       let finished = false;
+      let startupPending = true;
       let configurationFailure = false;
       const turnTimeLimitMs =
         this.config.agentRunTimeoutMs ?? DEFAULT_AGENT_RUN_TIMEOUT_MS;
@@ -85,7 +86,9 @@ export class DotAgent extends AbstractAgent {
         message: `This turn reached the ${turnTimeLimitMs / 1000} second time limit and was stopped. Try a smaller request.`,
       });
       const fail = (error: unknown) => {
+        if (subscriber.closed) return;
         if (!configurationFailure) observe({ type: EventType.RUN_ERROR });
+        finished = true;
         subscriber.next(
           this.channel
             ? channelError()
@@ -99,6 +102,18 @@ export class DotAgent extends AbstractAgent {
         );
         subscriber.complete();
       };
+      controller.signal.addEventListener(
+        'abort',
+        () => {
+          if (!startupPending || subscriber.closed) return;
+          if (timedOut) {
+            finished = true;
+            subscriber.next(this.channel ? channelError() : timeLimitError());
+            subscriber.complete();
+          } else fail(controller.signal.reason);
+        },
+        { once: true },
+      );
       try {
         const dot = this.workspace.dot(this.dotId);
         if (!dot) throw new Error('Specialist Dot not found.');
@@ -336,20 +351,22 @@ export class DotAgent extends AbstractAgent {
           maxRetries: 1,
         });
         const start = async () => {
-          const appTools =
+          let appTools: ToolDefinition[] = [];
+          if (
             this.config.composioApiKey &&
             initialSettings.appsAllowed &&
             dot.appsAllowed
-              ? await composioTools(
-                  await composioSession(
-                    this.config,
-                    this.workspace,
-                    dot,
-                    this.workspace.ownerId,
-                  ),
-                  check,
-                )
-              : [];
+          ) {
+            const session = await composioSession(
+              this.config,
+              this.workspace,
+              dot,
+              this.workspace.ownerId,
+              controller.signal,
+            );
+            check();
+            appTools = await composioTools(session, check, controller.signal);
+          }
           check();
           const serverTools = [
             ...tools,
@@ -461,6 +478,7 @@ export class DotAgent extends AbstractAgent {
                 subscriber.complete();
               },
             });
+          startupPending = false;
         };
         void start().catch(fail);
       } catch (error) {

tests/agent-run-timeout.test.ts

@@ -13,6 +13,17 @@ 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(),
+  tools: vi.fn(),
+}));
+vi.mock('../src/server/composio.js', () => ({
+  composioSession: composio.session,
+}));
+vi.mock('../src/server/composio-tools.js', () => ({
+  composioTools: composio.tools,
+}));
+
 const resources: Array<{ close(): void }> = [];
 function sharedStores() {
   const dir = mkdtempSync(join(tmpdir(), 'opendots-timeout-'));
@@ -27,6 +38,8 @@ function sharedStores() {
 afterEach(() => {
   vi.restoreAllMocks();
   vi.useRealTimers();
+  composio.session.mockReset();
+  composio.tools.mockReset();
   resources.splice(0).forEach((resource) => resource.close());
 });
 
@@ -98,6 +111,179 @@ it('uses the configured timeout for interactive Dot turns', async () => {
   );
 });
 
+it('completes a timed-out turn while Composio session startup is stalled', async () => {
+  vi.useFakeTimers();
+  const store = new Store(':memory:');
+  const workspace = new WorkspaceStore(':memory:', 'owner');
+  resources.push(store, workspace);
+  store.updateSettings({ appsAllowed: true });
+  const dot = workspace.dots()[0];
+  workspace.updateDot(dot.id, {
+    ...dot,
+    appsAllowed: true,
+    composioToolkits: ['gmail'],
+  });
+  workspace.bindThread('thread', dot.id, 'Composio timeout');
+  const agent = new DotAgent(
+    store,
+    workspace,
+    {
+      intelligenceKey: 'fixture',
+      apiKey: 'fixture',
+      model: 'custom-model',
+      baseUrl: 'https://unused.invalid/v1',
+      runtimeUrl: '',
+      voiceName: 'marin',
+      slackUsers: [],
+      composioApiKey: 'composio-fixture',
+    },
+    dot.id,
+  );
+  const input: RunAgentInput = {
+    threadId: 'thread',
+    runId: 'run',
+    state: {},
+    context: [],
+    messages: [{ id: 'user', role: 'user', content: 'Use Gmail.' }],
+    tools: [],
+    forwardedProps: {},
+  };
+  composio.session.mockImplementation(() => new Promise(() => {}));
+
+  const pending = lastValueFrom(agent.run(input).pipe(toArray()));
+  await vi.advanceTimersByTimeAsync(0);
+  expect(composio.session).toHaveBeenCalledWith(
+    expect.anything(),
+    workspace,
+    expect.objectContaining({ id: dot.id }),
+    'owner',
+    expect.any(AbortSignal),
+  );
+
+  await vi.advanceTimersByTimeAsync(91_000);
+  const events = await pending;
+  expect(events).toEqual(
+    expect.arrayContaining([
+      expect.objectContaining({
+        type: EventType.RUN_ERROR,
+        message: expect.stringContaining('90 second time limit'),
+      }),
+    ]),
+  );
+  expect(
+    events.filter((event) => event.type === EventType.RUN_ERROR),
+  ).toHaveLength(1);
+  expect(composio.tools).not.toHaveBeenCalled();
+});
+
+it('completes a timed-out turn while Composio tool discovery is stalled', async () => {
+  vi.useFakeTimers();
+  const store = new Store(':memory:');
+  const workspace = new WorkspaceStore(':memory:', 'owner');
+  resources.push(store, workspace);
+  store.updateSettings({ appsAllowed: true });
+  const dot = workspace.dots()[0];
+  workspace.updateDot(dot.id, { ...dot, appsAllowed: true });
+  workspace.bindThread('thread', dot.id, 'Composio tools timeout');
+  const agent = new DotAgent(
+    store,
+    workspace,
+    {
+      intelligenceKey: 'fixture',
+      apiKey: 'fixture',
+      model: 'custom-model',
+      baseUrl: 'https://unused.invalid/v1',
+      runtimeUrl: '',
+      voiceName: 'marin',
+      slackUsers: [],
+      composioApiKey: 'composio-fixture',
+    },
+    dot.id,
+  );
+  const input: RunAgentInput = {
+    threadId: 'thread',
+    runId: 'run',
+    state: {},
+    context: [],
+    messages: [{ id: 'user', role: 'user', content: 'Use Gmail.' }],
+    tools: [],
+    forwardedProps: {},
+  };
+  composio.session.mockResolvedValue({});
+  composio.tools.mockImplementation(() => new Promise(() => {}));
+
+  const pending = lastValueFrom(agent.run(input).pipe(toArray()));
+  await vi.advanceTimersByTimeAsync(0);
+  expect(composio.tools).toHaveBeenCalledTimes(1);
+  const signal = composio.tools.mock.calls[0][2] as AbortSignal;
+  await vi.advanceTimersByTimeAsync(91_000);
+
+  const events = await pending;
+  expect(signal.aborted).toBe(true);
+  expect(events).toEqual(
+    expect.arrayContaining([
+      expect.objectContaining({
+        type: EventType.RUN_ERROR,
+        message: expect.stringContaining('90 second time limit'),
+      }),
+    ]),
+  );
+  expect(
+    events.filter((event) => event.type === EventType.RUN_ERROR),
+  ).toHaveLength(1);
+});
+
+it('does not continue Composio startup when cancellation precedes session resolution', async () => {
+  vi.useFakeTimers();
+  const store = new Store(':memory:');
+  const workspace = new WorkspaceStore(':memory:', 'owner');
+  resources.push(store, workspace);
+  store.updateSettings({ appsAllowed: true });
+  const dot = workspace.dots()[0];
+  workspace.updateDot(dot.id, { ...dot, appsAllowed: true });
+  workspace.bindThread('thread', dot.id, 'Composio cancellation');
+  const agent = new DotAgent(
+    store,
+    workspace,
+    {
+      intelligenceKey: 'fixture',
+      apiKey: 'fixture',
+      model: 'custom-model',
+      baseUrl: 'https://unused.invalid/v1',
+      runtimeUrl: '',
+      voiceName: 'marin',
+      slackUsers: [],
+      composioApiKey: 'composio-fixture',
+    },
+    dot.id,
+  );
+  const input: RunAgentInput = {
+    threadId: 'thread',
+    runId: 'run',
+    state: {},
+    context: [],
+    messages: [{ id: 'user', role: 'user', content: 'Use Gmail.' }],
+    tools: [],
+    forwardedProps: {},
+  };
+  let resolveSession!: (value: unknown) => void;
+  composio.session.mockImplementation(
+    () => new Promise((resolve) => (resolveSession = resolve)),
+  );
+
+  const pending = lastValueFrom(agent.run(input).pipe(toArray()));
+  expect(composio.session).toHaveBeenCalledTimes(1);
+  const signal = composio.session.mock.calls[0][4] as AbortSignal;
+  agent.abortRun();
+  const events = await pending;
+  expect(signal.aborted).toBe(true);
+  expect(events.at(-1)?.type).toBe(EventType.RUN_ERROR);
+
+  resolveSession({});
+  await vi.advanceTimersByTimeAsync(0);
+  expect(composio.tools).not.toHaveBeenCalled();
+});
+
 it('uses the configured timeout for scheduled/background runs', async () => {
   vi.useFakeTimers();
   const store = new Store(':memory:');

tests/composio-tools.test.ts

@@ -24,8 +24,10 @@ it('passes Composio JSON Schema through and returns only safe execution failures
     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(session.tools).toHaveBeenCalledWith(undefined, { signal });
   expect(tool.parameters).toBe(parameters);
   await expect(tool.execute?.({ query: 'one' })).resolves.toEqual({
     data: { ok: true },