Support listening to top-level composio events (#3125)

## Summary
- Allow `composio listen` to subscribe directly to top-level
`composio.*` project event types without creating a temporary trigger.
- Keep the existing temporary-trigger flow for standard trigger slugs.
- Update help text and root command descriptions to document the new
behavior.
- Add analytics/test layer coverage for the new listen path and the
existing trigger path.

## Testing
- Added CLI test coverage for listening to
`composio.connected_account.expired` and verifying no temporary trigger
is created.
- Added CLI test coverage for the existing trigger-slug flow to confirm
it still creates and disables a temporary trigger.
- Added analytics test coverage to ensure `cli_version` is present on
tracked events.
- Not run: full repository test suite.

Author: CryogenicPlanet

ts/packages/cli/src/analytics/dispatch.ts

@@ -212,12 +212,12 @@ export const getCurrentCwdSessionId = (): string | undefined => {
 const withCliSessionId = (event: TrackEvent): TrackEvent => {
   if (!event) return event;
   const cliSessionId = getCurrentCwdSessionId();
-  if (!cliSessionId) return event;
   return {
     ...event,
     properties: {
       ...(event.properties ?? {}),
-      cli_session_id: cliSessionId,
+      cli_version: event.properties?.cli_version ?? constants.APP_VERSION,
+      ...(cliSessionId ? { cli_session_id: cliSessionId } : {}),
     },
   };
 };

ts/packages/cli/src/commands/listen.cmd.ts

@@ -1,5 +1,6 @@
 import { Args, Command, Options } from '@effect/cli';
 import { FileSystem } from '@effect/platform';
+import type { Composio as RawComposioClient } from '@composio/client';
 import { Deferred, Effect, Option, Runtime } from 'effect';
 import path from 'node:path';
 import { requireAuth } from 'src/effects/require-auth';
@@ -21,17 +22,21 @@ import { matchesTriggerListenFilters } from './triggers/filter';
 import { parseTriggerListenEvent } from './triggers/parse';
 
 const slug = Args.text({ name: 'slug' }).pipe(
-  Args.withDescription('Trigger slug (e.g. "GMAIL_NEW_GMAIL_MESSAGE")')
+  Args.withDescription(
+    'Trigger slug (e.g. "GMAIL_NEW_GMAIL_MESSAGE") or project event type (e.g. "composio.connected_account.expired")'
+  )
 );
 
 const params = Options.text('params').pipe(
   Options.withAlias('p'),
-  Options.withDescription('Trigger create params as JSON/JS object, @file, or - for stdin'),
+  Options.withDescription(
+    'Trigger create params as JSON/JS object, @file, or - for stdin. Only valid for trigger slugs.'
+  ),
   Options.optional
 );
 
 const maxEvents = Options.integer('max-events').pipe(
-  Options.withDescription('Stop after receiving N events for this temporary trigger'),
+  Options.withDescription('Stop after receiving N matching events'),
   Options.optional
 );
 
@@ -57,6 +62,8 @@ const debug = Options.boolean('debug').pipe(
 const sanitizePathPart = (value: string): string =>
   value.replace(/[^A-Z0-9._-]+/gi, '-').replace(/^-+|-+$/g, '') || 'unknown';
 
+const isProjectEventType = (value: string): boolean => value.startsWith('composio.');
+
 const resolveParamsInput = (input: Option.Option<string>) =>
   resolveOptionalTextInput(input, { missingValue: '{}' });
 
@@ -79,6 +86,58 @@ const parseCreateParams = (raw: string) =>
     return parsed as Record<string, unknown>;
   });
 
+const assertSupportedListenParams = (params: {
+  listeningToProjectEvent: boolean;
+  slug: string;
+  createParamsInput: Record<string, unknown>;
+}) =>
+  params.listeningToProjectEvent && Object.keys(params.createParamsInput).length > 0
+    ? Effect.fail(
+        new Error(
+          `--params is only supported for trigger slugs. "${params.slug}" is a project-level composio.* event type and does not create a temporary trigger.`
+        )
+      )
+    : Effect.void;
+
+const resolveConnectedAccountIdForTrigger = (params: {
+  client: RawComposioClient;
+  slug: string;
+  consumerUserId: string;
+}) =>
+  Effect.gen(function* () {
+    const toolkitSlug = toolkitFromToolSlug(params.slug);
+    if (!toolkitSlug) {
+      return yield* Effect.fail(
+        new Error(
+          `Could not infer a toolkit from trigger slug "${params.slug}". Use a standard trigger slug such as "GMAIL_NEW_GMAIL_MESSAGE", or a project event type such as "composio.connected_account.expired".`
+        )
+      );
+    }
+
+    const connectedAccounts = yield* Effect.tryPromise({
+      try: () =>
+        params.client.connectedAccounts.list({
+          toolkit_slugs: [toolkitSlug],
+          user_ids: [params.consumerUserId],
+          statuses: ['ACTIVE'],
+          limit: 100,
+        }),
+      catch: error =>
+        new Error(`Failed to list connected accounts for "${toolkitSlug}": ${String(error)}`),
+    });
+    const connectedAccountId = selectConnectedAccountId(connectedAccounts.items);
+
+    if (!connectedAccountId) {
+      return yield* Effect.fail(
+        new Error(
+          `No active connected account found for toolkit "${toolkitSlug}" and consumer user "${params.consumerUserId}". Run \`composio link ${toolkitSlug}\` first.`
+        )
+      );
+    }
+
+    return connectedAccountId;
+  });
+
 const selectConnectedAccountId = (
   items: ReadonlyArray<{
     id: string;
@@ -102,6 +161,9 @@ const emitStreamLine = (line: string, ui: TerminalUI) =>
     yield* ui.output(line, { force: true });
   });
 
+const eventTypeOf = (eventData: Record<string, unknown>): string | undefined =>
+  typeof eventData.type === 'string' && eventData.type.length > 0 ? eventData.type : undefined;
+
 const extractEventFileId = (eventData: Record<string, unknown>): string => {
   const candidates = [
     eventData.id,
@@ -180,6 +242,23 @@ const formatStreamValue = (value: unknown): string => {
   return JSON.stringify(value);
 };
 
+const formatStopMessage = (params: {
+  matchingEvents: number;
+  timedOut: boolean;
+  temporaryTriggerDisabled: boolean;
+}): string => {
+  const eventLabel = `event${params.matchingEvents === 1 ? '' : 's'}`;
+  if (params.timedOut) {
+    return params.temporaryTriggerDisabled
+      ? `Stopped after timeout with ${params.matchingEvents} matching ${eventLabel}. Temporary trigger disabled.`
+      : `Stopped after timeout with ${params.matchingEvents} matching ${eventLabel}.`;
+  }
+
+  return params.temporaryTriggerDisabled
+    ? `Stopped after receiving ${params.matchingEvents} ${eventLabel}. Temporary trigger disabled.`
+    : `Stopped after receiving ${params.matchingEvents} ${eventLabel}.`;
+};
+
 const TIMEOUT_UNITS_MS: Record<string, number> = {
   ms: 1,
   millisecond: 1,
@@ -252,44 +331,27 @@ export const listenCmd = Command.make(
         );
       }
 
+      const listeningToProjectEvent = isProjectEventType(slug);
       const rawParams = Option.isSome(params)
         ? (yield* resolveParamsInput(params))?.trim() || '{}'
         : '{}';
       const createParamsInput = yield* parseCreateParams(rawParams);
-      const toolkitSlug = toolkitFromToolSlug(slug);
-      if (!toolkitSlug) {
-        return yield* Effect.fail(
-          new Error(
-            `Could not infer a toolkit from trigger slug "${slug}". Use a standard trigger slug such as "GMAIL_NEW_GMAIL_MESSAGE".`
-          )
-        );
-      }
-
-      const connectedAccounts = yield* Effect.tryPromise({
-        try: () =>
-          client.connectedAccounts.list({
-            toolkit_slugs: [toolkitSlug],
-            user_ids: resolvedProject.consumerUserId ? [resolvedProject.consumerUserId] : undefined,
-            statuses: ['ACTIVE'],
-            limit: 100,
-          }),
-        catch: error =>
-          new Error(`Failed to list connected accounts for "${toolkitSlug}": ${String(error)}`),
-      });
-      const resolvedConnectedAccountId = selectConnectedAccountId(connectedAccounts.items);
-
-      if (!resolvedConnectedAccountId) {
-        return yield* Effect.fail(
-          new Error(
-            `No active connected account found for toolkit "${toolkitSlug}" and consumer user "${resolvedProject.consumerUserId}". Run \`composio link ${toolkitSlug}\` first.`
-          )
-        );
-      }
-
-      const createParams = {
-        ...createParamsInput,
-        connected_account_id: resolvedConnectedAccountId,
-      } as Parameters<typeof client.triggerInstances.upsert>[1];
+      yield* assertSupportedListenParams({ listeningToProjectEvent, slug, createParamsInput });
+
+      const resolvedConnectedAccountId = listeningToProjectEvent
+        ? undefined
+        : yield* resolveConnectedAccountIdForTrigger({
+            client,
+            slug,
+            consumerUserId: resolvedProject.consumerUserId,
+          });
+
+      const createParams = listeningToProjectEvent
+        ? undefined
+        : ({
+            ...createParamsInput,
+            connected_account_id: resolvedConnectedAccountId,
+          } as Parameters<typeof client.triggerInstances.upsert>[1]);
       const timeoutMs = Option.match(timeout, {
         onNone: () => undefined,
         onSome: value => parseTimeoutMs(value),
@@ -311,7 +373,11 @@ export const listenCmd = Command.make(
         onNone: () => resolveFallbackArtifactsDir(),
         onSome: value => value.directoryPath,
       });
-      const triggerDir = path.join(artifactsRoot, 'triggers', sanitizePathPart(slug));
+      const triggerDir = path.join(
+        artifactsRoot,
+        listeningToProjectEvent ? 'events' : 'triggers',
+        sanitizePathPart(slug)
+      );
       const streamFilePath = path.join(triggerDir, 'events.jsonl');
 
       yield* fs.makeDirectory(triggerDir, { recursive: true });
@@ -323,21 +389,28 @@ export const listenCmd = Command.make(
       const seenEventIds = new Set<string>();
 
       yield* Effect.acquireUseRelease(
-        Effect.tryPromise({
-          try: () => client.triggerInstances.upsert(slug, createParams),
-          catch: error =>
-            new Error(`Failed to create temporary trigger "${slug}": ${String(error)}`),
-        }),
+        listeningToProjectEvent
+          ? Effect.succeed<null | { trigger_id: string }>(null)
+          : Effect.tryPromise({
+              try: () => client.triggerInstances.upsert(slug, createParams),
+              catch: error =>
+                new Error(`Failed to create temporary trigger "${slug}": ${String(error)}`),
+            }),
         createdTrigger =>
           Effect.gen(function* () {
             yield* emitStreamLine(`listening for events ${slug} (tail at ${streamFilePath})`, ui);
             if (debug) {
-              const debugMsg = `[debug] trigger_id=${createdTrigger.trigger_id} project=${resolvedProject.projectId} org=${resolvedProject.orgId} createParams=${JSON.stringify(createParams)}`;
+              const debugMsg =
+                createdTrigger === null
+                  ? `[debug] event_type=${slug} project=${resolvedProject.projectId} org=${resolvedProject.orgId}`
+                  : `[debug] trigger_id=${createdTrigger.trigger_id} project=${resolvedProject.projectId} org=${resolvedProject.orgId} createParams=${JSON.stringify(createParams)}`;
               yield* emitStreamLine(debugMsg, ui);
-              yield* emitStreamLine(
-                `[debug] upsert response: ${JSON.stringify(createdTrigger)}`,
-                ui
-              );
+              if (createdTrigger !== null) {
+                yield* emitStreamLine(
+                  `[debug] upsert response: ${JSON.stringify(createdTrigger)}`,
+                  ui
+                );
+              }
             }
 
             const onEvent = (eventData: Record<string, unknown>) => {
@@ -353,14 +426,20 @@ export const listenCmd = Command.make(
                       ui
                     );
                   }
-                  const parsed = parseTriggerListenEvent(eventData);
-                  const filterResult = matchesTriggerListenFilters(
-                    { triggerId: createdTrigger.trigger_id },
-                    parsed
-                  );
+                  const parsedTriggerEvent =
+                    createdTrigger === null ? undefined : parseTriggerListenEvent(eventData);
+                  const filterResult =
+                    createdTrigger === null
+                      ? eventTypeOf(eventData) === slug
+                      : matchesTriggerListenFilters(
+                          { triggerId: createdTrigger.trigger_id },
+                          parsedTriggerEvent!
+                        );
                   if (debug) {
                     yield* emitStreamLine(
-                      `[debug] parsed.id=${parsed.id} triggerSlug=${parsed.triggerSlug} trigger_id=${createdTrigger.trigger_id} match=${filterResult}`,
+                      createdTrigger === null
+                        ? `[debug] event.type=${eventTypeOf(eventData) ?? '<missing>'} match=${filterResult}`
+                        : `[debug] parsed.id=${parsedTriggerEvent!.id} triggerSlug=${parsedTriggerEvent!.triggerSlug} trigger_id=${createdTrigger.trigger_id} match=${filterResult}`,
                       ui
                     );
                   }
@@ -382,8 +461,9 @@ export const listenCmd = Command.make(
                   const eventJson = JSON.stringify(eventData, null, 2);
                   const streamEntry = JSON.stringify({
                     event_id: eventFileId,
-                    trigger_id: createdTrigger.trigger_id,
-                    trigger_slug: slug,
+                    event_type: eventTypeOf(eventData),
+                    trigger_id: createdTrigger?.trigger_id,
+                    trigger_slug: createdTrigger ? slug : undefined,
                     file_path: eventFilePath,
                     received_at: new Date().toISOString(),
                   });
@@ -440,34 +520,44 @@ export const listenCmd = Command.make(
             const stopReason = yield* Effect.raceFirst(listenEffect, Deferred.await(stopWhenDone));
             if (stopReason === 'max-events') {
               yield* ui.outro(
-                `Stopped after receiving ${matchingEvents} events. Temporary trigger disabled.`
+                formatStopMessage({
+                  matchingEvents,
+                  timedOut: false,
+                  temporaryTriggerDisabled: createdTrigger !== null,
+                })
               );
               return;
             }
 
             if (stopReason === 'timeout') {
               yield* ui.outro(
-                `Stopped after timeout with ${matchingEvents} matching event${matchingEvents === 1 ? '' : 's'}. Temporary trigger disabled.`
+                formatStopMessage({
+                  matchingEvents,
+                  timedOut: true,
+                  temporaryTriggerDisabled: createdTrigger !== null,
+                })
               );
             }
           }),
         created =>
-          Effect.tryPromise({
-            try: () =>
-              client.triggerInstances.manage.update(created.trigger_id, { status: 'disable' }),
-            catch: error =>
-              new Error(
-                `Failed to disable temporary trigger "${created.trigger_id}": ${String(error)}`
-              ),
-          }).pipe(
-            Effect.catchAll(error =>
-              ui.log.warn(error instanceof Error ? error.message : String(error))
-            )
-          )
+          created === null
+            ? Effect.void
+            : Effect.tryPromise({
+                try: () =>
+                  client.triggerInstances.manage.update(created.trigger_id, { status: 'disable' }),
+                catch: error =>
+                  new Error(
+                    `Failed to disable temporary trigger "${created.trigger_id}": ${String(error)}`
+                  ),
+              }).pipe(
+                Effect.catchAll(error =>
+                  ui.log.warn(error instanceof Error ? error.message : String(error))
+                )
+              )
       );
     })
 ).pipe(
   Command.withDescription(
-    'Create a temporary subscription for consumer-project events and write each event to artifacts for easy background-agent consumption.'
+    'Listen to consumer-project realtime events. Trigger slugs create a temporary trigger; top-level composio.* event types subscribe directly.'
   )
 );