From 09ca12e885d5b3727e656a6cf1b8a5a844add380 Mon Sep 17 00:00:00 2001 From: Twil3akine <156310846+Twil3akine@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:45:40 +0900 Subject: [PATCH 1/2] fix: abort Composio startup with Dot runs --- src/server/composio-tools.ts | 19 +++- src/server/composio.ts | 17 +-- src/server/dot-agent.ts | 40 +++++-- tests/agent-run-timeout.test.ts | 186 ++++++++++++++++++++++++++++++++ tests/composio-tools.test.ts | 4 +- tests/composio.test.ts | 38 +++++-- 6 files changed, 275 insertions(+), 29 deletions(-) diff --git a/src/server/composio-tools.ts b/src/server/composio-tools.ts index b2c04275..4db7ca64 100644 --- a/src/server/composio-tools.ts +++ b/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>; +type SessionWithRequestOptions = Session & { + tools: ( + modifiers?: Parameters[0], + requestOptions?: ComposioRequestOptions, + ) => ReturnType; +}; -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 }) => diff --git a/src/server/composio.ts b/src/server/composio.ts index 890649f9..b5897852 100644 --- a/src/server/composio.ts +++ b/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; } diff --git a/src/server/dot-agent.ts b/src/server/dot-agent.ts index f5e34ab7..9d24df35 100644 --- a/src/server/dot-agent.ts +++ b/src/server/dot-agent.ts @@ -69,6 +69,7 @@ export class DotAgent extends AbstractAgent { let watcher: ReturnType | 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) { diff --git a/tests/agent-run-timeout.test.ts b/tests/agent-run-timeout.test.ts index 6074393d..9c5050fd 100644 --- a/tests/agent-run-timeout.test.ts +++ b/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:'); diff --git a/tests/composio-tools.test.ts b/tests/composio-tools.test.ts index 88f826ba..1213ca75 100644 --- a/tests/composio-tools.test.ts +++ b/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 }, diff --git a/tests/composio.test.ts b/tests/composio.test.ts index b50414dd..70948855 100644 --- a/tests/composio.test.ts +++ b/tests/composio.test.ts @@ -40,26 +40,38 @@ it('creates a non-Instant session and reuses it while the policy is unchanged', composioToolkits: ['gmail'], }); const created = { sessionId: 'session-one' }; + const signal = new AbortController().signal; sdk.create.mockResolvedValue(created); sdk.use.mockResolvedValue(created); 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 () => { @@ -67,16 +79,23 @@ it('creates a new session when the toolkit policy changes', async () => { const workspace = new WorkspaceStore(':memory:', 'owner'); stores.push(workspace); const dot = workspace.dots()[0]; + const signal = new AbortController().signal; sdk.create .mockResolvedValueOnce({ sessionId: 'session-one' }) .mockResolvedValueOnce({ sessionId: 'session-two' }); - await composioSession(config, workspace, dot, 'owner'); + await composioSession(config, workspace, dot, 'owner', signal); workspace.updateDot(dot.id, { ...workspace.dot(dot.id)!, composioToolkits: ['gmail'], }); - await composioSession(config, workspace, workspace.dot(dot.id)!, 'owner'); + await composioSession( + config, + workspace, + workspace.dot(dot.id)!, + 'owner', + signal, + ); expect(sdk.create).toHaveBeenCalledTimes(2); expect(sdk.use).not.toHaveBeenCalled(); @@ -89,4 +108,5 @@ it('keeps Composio optional in setup status', async () => { expect(without.composio).toBe(false); expect(without.missing).not.toContain('COMPOSIO_API_KEY'); expect(setupStatus(config).composio).toBe(true); + expect(setupStatus({ ...config, composioApiKey: '' }).composio).toBe(false); }); From 7cca89029e7c426ce7bad112d3324e63c9fb69f6 Mon Sep 17 00:00:00 2001 From: Twil3akine <156310846+Twil3akine@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:45:58 +0900 Subject: [PATCH 2/2] fix: forward Composio key to Compose app --- compose.yml | 1 + tests/telemetry.test.ts | 1 + 2 files changed, 2 insertions(+) diff --git a/compose.yml b/compose.yml index ecac4711..93d96246 100644 --- a/compose.yml +++ b/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:-} diff --git a/tests/telemetry.test.ts b/tests/telemetry.test.ts index 2230ed05..9b0f03e0 100644 --- a/tests/telemetry.test.ts +++ b/tests/telemetry.test.ts @@ -130,6 +130,7 @@ it('forwards the CLI identity and both opt-outs into the Docker app', async () = expect(compose).toContain( 'COPILOTKIT_TELEMETRY_DISABLED: ${COPILOTKIT_TELEMETRY_DISABLED:-false}', ); + expect(compose).toContain('COMPOSIO_API_KEY: ${COMPOSIO_API_KEY:-}'); }); it('hands the persistent installation fallback to the installed runtime when CLI identity is absent', async () => {