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/src/server/composio-tools.ts b/src/server/composio-tools.ts index b2c04275..0c84d353 100644 --- a/src/server/composio-tools.ts +++ b/src/server/composio-tools.ts @@ -3,9 +3,29 @@ import type { composioSession } from './composio.js'; import { safeFailure } from './slack-channel.js'; type Session = Awaited>; +type CancellableSession = Session & { + tools( + modifiers: Parameters[0], + requestOptions: { signal?: AbortSignal }, + ): ReturnType; + execute( + toolSlug: string, + input: Record, + options: Parameters[2], + requestOptions: { signal?: AbortSignal }, + ): ReturnType; +}; -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, + undefined, + { signal }, ); return { data, error }; } catch (error) { diff --git a/src/server/composio.ts b/src/server/composio.ts index 890649f9..d7b48915 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..57fd2948 100644 --- a/src/server/dot-agent.ts +++ b/src/server/dot-agent.ts @@ -67,7 +67,6 @@ export class DotAgent extends AbstractAgent { this.controller = controller; let subscription: { unsubscribe(): void } | undefined; let watcher: ReturnType | undefined; - let timedOut = false; let finished = false; let configurationFailure = false; const turnTimeLimitMs = @@ -75,16 +74,17 @@ export class DotAgent extends AbstractAgent { 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(); }, }); diff --git a/tests/agent-run-timeout.test.ts b/tests/agent-run-timeout.test.ts index 6074393d..e1891361 100644 --- a/tests/agent-run-timeout.test.ts +++ b/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 () => { diff --git a/tests/composio-tools.test.ts b/tests/composio-tools.test.ts index 88f826ba..6f84711b 100644 --- a/tests/composio-tools.test.ts +++ b/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); }); diff --git a/tests/composio.test.ts b/tests/composio.test.ts index b50414dd..2143fcbe 100644 --- a/tests/composio.test.ts +++ b/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 () => {