Composio Integration Architecture: Triggers, API Connections & White-Label Auth

┌─────────────────────────────────────────────────────────────────────────────┐
│                           DESKTOP / CLI                                      │
│                                                                              │
│  User: "Connect my Google Drive"                                            │
│  Agent: Opens auth URL → User authenticates → Connection saved              │
│                                                                              │
│  User: "Send a Slack message to #general"                                   │
│  Agent: Uses bytespace_execute tool → Proxied to Composio → Slack API        │
│                                                                              │
│  User: "Notify me when I get a GitHub PR review"                            │
│  Agent: Creates trigger → Webhook configured → Events flow to daemon        │
│                                                                              │
└──────────────────────────────┬──────────────────────────────────────────────┘
                               │ IPC
                               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                            DAEMON                                            │
│                                                                              │
│  ┌──────────────────┐  ┌───────────────────┐  ┌─────────────────────┐      │
│  │ bytespace_search  │  │ bytespace_execute   │  │ bytespace_connect    │      │
│  │                  │  │                    │  │                     │      │
│  │ Discover tools   │  │ Run actions on     │  │ Initiate OAuth for  │      │
│  │ across 500+ apps │  │ connected services │  │ a service           │      │
│  └──────┬───────────┘  └──────┬─────────────┘  └──────┬──────────────┘      │
│         │                     │                        │                     │
│  ┌──────┴─────────────────────┴────────────────────────┴──────────────┐     │
│  │                    ComposioClient (daemon-side)                      │     │
│  │  • Thin wrapper over proxied HTTP                                   │     │
│  │  • Injects session token + device signature                         │     │
│  │  • Routes all calls through Bytespace proxy                         │     │
│  │  • ZERO Composio API key — only session token                       │     │
│  └────────────────────────────┬────────────────────────────────────────┘     │
│                               │                                              │
└───────────────────────────────┼──────────────────────────────────────────────┘
                                │ HTTPS + Device Signature
                                ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                       BYTESPACE PROXY                                        │
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────┐       │
│  │              /api/proxy/composio/[...path]                        │       │
│  │                                                                   │       │
│  │  1. authenticateProxyRequest() — session + device signature       │       │
│  │  2. Decrypt COMPOSIO_API_KEY from encrypted storage               │       │
│  │  3. Map bot0 userId to Composio userId                            │       │
│  │  4. Forward request to Composio API with x-api-key header         │       │
│  │  5. Return response to daemon                                     │       │
│  └──────────────────────────────────────────────────────────────────┘       │
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────┐       │
│  │         /api/webhooks/composio (public endpoint)                  │       │
│  │                                                                   │       │
│  │  1. Verify HMAC-SHA256 webhook signature                          │       │
│  │  2. Parse trigger payload (V3 format)                             │       │
│  │  3. Resolve user_id → device_id → target daemon                   │       │
│  │  4. Route trigger event via Hub to daemon                         │       │
│  │  5. OR queue as proactive task if daemon offline                   │       │
│  └──────────────────────────────────────────────────────────────────┘       │
│                                                                              │
└──────────────────────────────┬──────────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         COMPOSIO API                                         │
│                                                                              │
│  • Tool discovery & execution                                                │
│  • OAuth flow management                                                     │
│  • Trigger subscription & delivery                                           │
│  • Credential storage & refresh                                              │
│  • User isolation                                                            │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

External Service (Gmail, GitHub, Slack, etc.)
         │
         │ Webhook / Poll result
         ▼
┌─────────────────┐
│    Composio      │  Receives event, matches to trigger instance
│    Backend       │  Delivers to configured webhook URL
└────────┬────────┘
         │ POST /api/webhooks/composio
         │ + HMAC-SHA256 signature
         ▼
┌─────────────────┐
│   Bytespace      │  1. Verify signature
│   Webhook        │  2. Parse V3 payload
│   Handler        │  3. Resolve user → daemon
└────────┬────────┘
         │
    ┌────┴────────────────────────┐
    │                             │
    ▼                             ▼
DAEMON ONLINE               DAEMON OFFLINE
    │                             │
    │ Hub: trigger_event          │ Queue in ctx0_trigger_events
    ▼                             │ (pending, TTL-based)
┌─────────────────┐               │
│    Daemon        │               │ On reconnect:
│                  │               │ Daemon polls pending events
│  Proactive       │               │ Processes in order
│  Engine          │◄──────────────┘
│                  │
│  1. Evaluate     │
│  2. Permission   │
│  3. Execute/Ask  │
└─────────────────┘

// apps/bytespace/src/app/api/proxy/composio/[...path]/route.ts

import { NextRequest, NextResponse } from 'next/server';
import { authenticateProxyRequest, createProxyCorsResponse } from '@/lib/auth/proxy';

const COMPOSIO_API_BASE = 'https://backend.composio.dev/api/v3';

export async function POST(
  request: NextRequest,
  { params }: { params: { path: string[] } }
) {
  const bodyText = await request.text();

  // 1. Authenticate: session token + device signature
  const auth = await authenticateProxyRequest(request, bodyText);
  if (!auth.success) return auth.response;

  // 2. Get Composio API key (encrypted on proxy, decrypted here)
  const composioApiKey = await getComposioApiKey(auth.session.userId);
  if (!composioApiKey) {
    return NextResponse.json(
      { error: 'Composio not configured' },
      { status: 400 }
    );
  }

  // 3. Map bot0 userId to Composio userId
  //    bot0 userId IS the Composio userId (direct mapping)
  const composioUserId = auth.session.userId;

  // 4. Forward to Composio API
  const composioPath = params.path.join('/');
  const composioUrl = `${COMPOSIO_API_BASE}/${composioPath}`;

  const response = await fetch(composioUrl, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'x-api-key': composioApiKey,
    },
    body: bodyText,
  });

  // 5. Stream response back to daemon
  return new NextResponse(response.body, {
    status: response.status,
    headers: {
      'Content-Type': response.headers.get('Content-Type') || 'application/json',
    },
  });
}

export async function GET(
  request: NextRequest,
  { params }: { params: { path: string[] } }
) {
  // Similar pattern for GET requests (tool discovery, connection status, etc.)
  // ...
}

export async function OPTIONS() {
  return createProxyCorsResponse();
}

// packages/daemon/src/composio/client.ts

import { DaemonConfig } from '../config';
import { SigningProvider, createDeviceHeaders } from '../signing';

export interface ComposioClientDeps {
  getConfig: () => DaemonConfig;
  getSigningProvider: () => SigningProvider | null;
}

export class ProxiedComposioClient {
  constructor(private deps: ComposioClientDeps) {}

  /**
   * Search for available tools across Composio toolkits.
   * Used by the bytespace_search tool.
   */
  async searchTools(query: string, options?: {
    toolkits?: string[];
    limit?: number;
  }): Promise<ComposioToolInfo[]> {
    return this.request('POST', 'tools/search', {
      query,
      toolkits: options?.toolkits,
      limit: options?.limit ?? 20,
    });
  }

  /**
   * Execute a Composio tool action.
   * Used by the bytespace_execute tool.
   */
  async executeTool(slug: string, args: Record<string, unknown>): Promise<unknown> {
    return this.request('POST', 'tools/execute', {
      slug,
      arguments: args,
    });
  }

  /**
   * List connected accounts for the current user.
   */
  async listConnections(options?: {
    toolkit?: string;
    status?: 'ACTIVE' | 'EXPIRED' | 'INITIATED';
  }): Promise<ComposioConnection[]> {
    const params = new URLSearchParams();
    if (options?.toolkit) params.set('toolkit', options.toolkit);
    if (options?.status) params.set('status', options.status);
    return this.request('GET', `connected_accounts?${params}`);
  }

  /**
   * Initiate an OAuth connection for a toolkit.
   * Returns a redirect URL the user must visit.
   */
  async initiateConnection(toolkit: string, options?: {
    authConfigId?: string;
    callbackUrl?: string;
  }): Promise<{ redirectUrl: string; connectionId: string }> {
    return this.request('POST', 'connected_accounts', {
      toolkit,
      auth_config_id: options?.authConfigId,
      callback_url: options?.callbackUrl,
    });
  }

  /**
   * Check connection status (polling after OAuth redirect).
   */
  async getConnectionStatus(connectionId: string): Promise<ComposioConnection> {
    return this.request('GET', `connected_accounts/${connectionId}`);
  }

  /**
   * Create a trigger instance.
   */
  async createTrigger(slug: string, config: Record<string, unknown>): Promise<{
    triggerId: string;
  }> {
    return this.request('POST', `trigger_instances/${slug}/upsert`, {
      trigger_config: config,
    });
  }

  /**
   * List active trigger instances.
   */
  async listTriggers(): Promise<ComposioTriggerInstance[]> {
    return this.request('GET', 'trigger_instances/active');
  }

  /**
   * Enable/disable/delete a trigger.
   */
  async manageTrigger(triggerId: string, action: 'enable' | 'disable' | 'delete'): Promise<void> {
    await this.request('POST', `trigger_instances/${triggerId}/${action}`);
  }

  /**
   * List available trigger types for a toolkit.
   */
  async listTriggerTypes(toolkit?: string): Promise<ComposioTriggerType[]> {
    const params = toolkit ? `?toolkit=${toolkit}` : '';
    return this.request('GET', `triggers/types${params}`);
  }

  // ─── Internal ───

  private async request<T>(method: string, path: string, body?: unknown): Promise<T> {
    const config = this.deps.getConfig();
    const signingProvider = this.deps.getSigningProvider();
    const url = `${config.proxyUrl}/api/proxy/composio/${path}`;

    const bodyStr = body ? JSON.stringify(body) : undefined;
    const headers: Record<string, string> = {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${config.sessionToken}`,
    };

    // Inject device signature
    if (signingProvider && bodyStr) {
      const deviceHeaders = await createDeviceHeaders(signingProvider, bodyStr);
      Object.assign(headers, deviceHeaders);
    }

    const response = await fetch(url, { method, headers, body: bodyStr });

    if (!response.ok) {
      const error = await response.text();
      throw new Error(`Composio proxy error (${response.status}): ${error}`);
    }

    return response.json();
  }
}

// packages/daemon/src/tools/composio-search.ts

import { Tool, ToolResult } from './types';

export function createComposioSearchTool(deps: {
  getClient: () => ProxiedComposioClient | null;
}): Tool {
  return {
    name: 'bytespace_search',
    description: `Search for available API tools across 500+ connected services (Gmail, Slack, GitHub, Salesforce, etc.).
Use this to discover what actions are available before executing them.
Returns tool slugs, descriptions, required parameters, and connection status.`,
    inputSchema: {
      type: 'object',
      properties: {
        query: {
          type: 'string',
          description: 'Natural language search query (e.g., "send email", "create github issue", "list slack channels")',
        },
        toolkit: {
          type: 'string',
          description: 'Filter to a specific toolkit (e.g., "gmail", "github", "slack")',
        },
      },
      required: ['query'],
    },
    execute: async (input) => {
      const client = deps.getClient();
      if (!client) return 'Composio is not configured. The user needs to set up Composio in settings.';

      const results = await client.searchTools(input.query, {
        toolkits: input.toolkit ? [input.toolkit] : undefined,
        limit: 15,
      });

      if (results.length === 0) {
        return `No tools found for "${input.query}". Try a broader search or check if the toolkit is connected.`;
      }

      const formatted = results.map(t =>
        `- **${t.slug}** (${t.toolkit}): ${t.description}\n  Connected: ${t.connected ? 'Yes' : 'No — user must connect first'}\n  Params: ${t.parameters?.map(p => p.name).join(', ') || 'none'}`
      ).join('\n\n');

      return {
        output: `Found ${results.length} tools:\n\n${formatted}`,
        metadata: {
          type: 'bytespace_search',
          resultCount: results.length,
          toolkits: [...new Set(results.map(t => t.toolkit))],
        },
      } as ToolResult;
    },
  };
}

// packages/daemon/src/tools/composio-execute.ts

import { Tool, ToolResult } from './types';

export function createComposioExecuteTool(deps: {
  getClient: () => ProxiedComposioClient | null;
}): Tool {
  return {
    name: 'bytespace_execute',
    description: `Execute an action on a connected service via Composio.
Use bytespace_search first to discover available tools and their parameters.
The user must have an active connection to the toolkit for the action to succeed.
If the user is not connected, use bytespace_connect to initiate authentication.`,
    inputSchema: {
      type: 'object',
      properties: {
        slug: {
          type: 'string',
          description: 'The tool slug from bytespace_search (e.g., "GMAIL_SEND_EMAIL", "GITHUB_CREATE_ISSUE")',
        },
        arguments: {
          type: 'object',
          description: 'Arguments for the tool as key-value pairs',
          additionalProperties: true,
        },
        connected_account_id: {
          type: 'string',
          description: 'Optional: specific connected account ID if user has multiple accounts for this toolkit',
        },
      },
      required: ['slug', 'arguments'],
    },
    execute: async (input) => {
      const client = deps.getClient();
      if (!client) return 'Composio is not configured.';

      try {
        const result = await client.executeTool(input.slug, {
          ...input.arguments,
          ...(input.connected_account_id && {
            connected_account_id: input.connected_account_id,
          }),
        });

        const output = typeof result === 'string'
          ? result
          : JSON.stringify(result, null, 2);

        return {
          output: `Action ${input.slug} executed successfully:\n\n${output}`,
          metadata: {
            type: 'bytespace_execute',
            slug: input.slug,
            success: true,
          },
        } as ToolResult;
      } catch (error) {
        const message = error instanceof Error ? error.message : String(error);

        // Detect auth errors and guide agent to connect
        if (message.includes('401') || message.includes('not connected') || message.includes('no active connection')) {
          const toolkit = input.slug.split('_')[0].toLowerCase();
          return `The user is not connected to ${toolkit}. Use bytespace_connect to initiate authentication before retrying this action.`;
        }

        return `Action ${input.slug} failed: ${message}`;
      }
    },
  };
}

// packages/daemon/src/tools/composio-connect.ts

import { Tool, ToolResult } from './types';

export function createComposioConnectTool(deps: {
  getClient: () => ProxiedComposioClient | null;
}): Tool {
  return {
    name: 'bytespace_connect',
    description: `Connect the user to an external service (Gmail, GitHub, Slack, etc.) via OAuth or API key.
Returns an authentication URL that the user must visit to authorize access.
Use this when:
- The user explicitly asks to connect a service
- A bytespace_execute call fails because the user is not connected
- You need to check which services are already connected

Actions: "connect" (initiate OAuth), "list" (show connections), "disconnect" (remove connection)`,
    inputSchema: {
      type: 'object',
      properties: {
        action: {
          type: 'string',
          enum: ['connect', 'list', 'disconnect'],
          description: 'The action to perform',
        },
        toolkit: {
          type: 'string',
          description: 'The service to connect (e.g., "gmail", "github", "slack", "notion")',
        },
        connected_account_id: {
          type: 'string',
          description: 'For disconnect: the connected account ID to remove',
        },
      },
      required: ['action'],
    },
    execute: async (input) => {
      const client = deps.getClient();
      if (!client) return 'Composio is not configured.';

      switch (input.action) {
        case 'connect': {
          if (!input.toolkit) return 'Please specify a toolkit to connect (e.g., "gmail", "github").';

          const result = await client.initiateConnection(input.toolkit);

          return {
            output: `To connect ${input.toolkit}, the user needs to authenticate:\n\n**Auth URL:** ${result.redirectUrl}\n\nPlease open this URL to authorize bot0 to access your ${input.toolkit} account. The connection will be saved automatically.`,
            metadata: {
              type: 'bytespace_connect',
              action: 'connect',
              toolkit: input.toolkit,
              authUrl: result.redirectUrl,
              connectionId: result.connectionId,
            },
          } as ToolResult;
        }

        case 'list': {
          const connections = await client.listConnections({
            toolkit: input.toolkit,
            status: 'ACTIVE',
          });

          if (connections.length === 0) {
            return input.toolkit
              ? `No active connections to ${input.toolkit}. Use bytespace_connect with action "connect" to set one up.`
              : 'No active connections. The user has not connected any external services yet.';
          }

          const formatted = connections.map(c =>
            `- **${c.toolkit}** (${c.id}): ${c.status} — connected ${c.createdAt}`
          ).join('\n');

          return {
            output: `Active connections:\n\n${formatted}`,
            metadata: {
              type: 'bytespace_connect',
              action: 'list',
              connectionCount: connections.length,
            },
          } as ToolResult;
        }

        case 'disconnect': {
          if (!input.connected_account_id) return 'Please specify connected_account_id to disconnect.';
          // Composio API: DELETE connected account
          return `Connection ${input.connected_account_id} has been disconnected.`;
        }
      }
    },
  };
}

// packages/daemon/src/tools/composio-trigger.ts

import { Tool, ToolResult } from './types';

export function createComposioTriggerTool(deps: {
  getClient: () => ProxiedComposioClient | null;
}): Tool {
  return {
    name: 'bytespace_trigger',
    description: `Manage event triggers for connected services.
Triggers notify the daemon when something happens (new email, GitHub PR, Slack message, etc.).
This is how the agent becomes proactive — reacting to real-world events.

Actions:
- "create": Set up a new trigger (e.g., "notify me when I get a GitHub PR review")
- "list": Show active triggers
- "list_types": Show available trigger types for a toolkit
- "disable" / "enable" / "delete": Manage existing triggers`,
    inputSchema: {
      type: 'object',
      properties: {
        action: {
          type: 'string',
          enum: ['create', 'list', 'list_types', 'enable', 'disable', 'delete'],
        },
        trigger_slug: {
          type: 'string',
          description: 'Trigger type slug (e.g., "GMAIL_NEW_GMAIL_MESSAGE", "GITHUB_COMMIT_EVENT")',
        },
        trigger_config: {
          type: 'object',
          description: 'Configuration for the trigger (e.g., { "owner": "myorg", "repo": "myrepo" })',
          additionalProperties: true,
        },
        trigger_id: {
          type: 'string',
          description: 'For enable/disable/delete: the trigger instance ID',
        },
        toolkit: {
          type: 'string',
          description: 'For list_types: filter by toolkit',
        },
      },
      required: ['action'],
    },
    execute: async (input) => {
      const client = deps.getClient();
      if (!client) return 'Composio is not configured.';

      switch (input.action) {
        case 'create': {
          if (!input.trigger_slug) return 'Specify trigger_slug (e.g., "GMAIL_NEW_GMAIL_MESSAGE").';
          const result = await client.createTrigger(
            input.trigger_slug,
            input.trigger_config ?? {}
          );
          return {
            output: `Trigger created: ${result.triggerId}\nType: ${input.trigger_slug}\nThe daemon will now receive events when this trigger fires.`,
            metadata: { type: 'bytespace_trigger', action: 'create', triggerId: result.triggerId },
          } as ToolResult;
        }

        case 'list': {
          const triggers = await client.listTriggers();
          if (triggers.length === 0) return 'No active triggers.';
          const formatted = triggers.map(t =>
            `- **${t.slug}** (${t.id}): ${t.status} — ${t.toolkit}`
          ).join('\n');
          return { output: `Active triggers:\n\n${formatted}`, metadata: { type: 'bytespace_trigger', action: 'list', count: triggers.length } } as ToolResult;
        }

        case 'list_types': {
          const types = await client.listTriggerTypes(input.toolkit);
          if (types.length === 0) return `No trigger types available${input.toolkit ? ` for ${input.toolkit}` : ''}.`;
          const formatted = types.map(t =>
            `- **${t.slug}**: ${t.description}\n  Config: ${JSON.stringify(t.config ?? {})}`
          ).join('\n\n');
          return { output: `Available trigger types:\n\n${formatted}`, metadata: { type: 'bytespace_trigger', action: 'list_types' } } as ToolResult;
        }

        case 'enable':
        case 'disable':
        case 'delete': {
          if (!input.trigger_id) return `Specify trigger_id to ${input.action}.`;
          await client.manageTrigger(input.trigger_id, input.action);
          return `Trigger ${input.trigger_id} has been ${input.action}d.`;
        }
      }
    },
  };
}

// In packages/daemon/src/index.ts — createDefaultToolRegistry()

import { createComposioSearchTool, createComposioExecuteTool, createComposioConnectTool, createComposioTriggerTool } from './tools';
import { ProxiedComposioClient } from './composio/client';

export function createDefaultToolRegistry(options?: {
  taskToolDeps?: TaskToolDeps;
  composioDeps?: { getConfig: () => DaemonConfig; getSigningProvider: () => SigningProvider | null };
}) {
  const registry = new ToolRegistry();

  // ... existing tools ...

  // Composio tools (lazy client)
  let composioClient: ProxiedComposioClient | null = null;
  const getComposioClient = () => {
    if (!composioClient && options?.composioDeps) {
      composioClient = new ProxiedComposioClient(options.composioDeps);
    }
    return composioClient;
  };

  registry.register(createComposioSearchTool({ getClient: getComposioClient }));
  registry.register(createComposioExecuteTool({ getClient: getComposioClient }));
  registry.register(createComposioConnectTool({ getClient: getComposioClient }));
  registry.register(createComposioTriggerTool({ getClient: getComposioClient }));

  return registry;
}

// In buildSystemPrompt(), add after tool usage section:

const composioPrompt = `
## External Service Integration (Composio)

You have access to 500+ external services through the bytespace_* tools:
- bytespace_search: Find available tools across Gmail, Slack, GitHub, Salesforce, etc.
- bytespace_execute: Run actions on connected services
- bytespace_connect: Help the user connect new services (OAuth)
- bytespace_trigger: Set up event triggers for proactive notifications

**Workflow for service actions:**
1. If unsure what tools exist, use bytespace_search first
2. If the user isn't connected, use bytespace_connect to initiate OAuth
3. Execute actions with bytespace_execute
4. Set up triggers with bytespace_trigger for ongoing monitoring

**Connection handling:**
- If an action fails with "not connected", offer to connect the service
- The user authenticates via a URL you provide — never ask for passwords or API keys
- Connected services persist across sessions — no need to reconnect

**Triggers:**
- Triggers make you proactive — you'll be notified of events automatically
- Always confirm with the user before creating triggers (they generate ongoing notifications)
- Common triggers: new emails, GitHub PRs, Slack messages, calendar events
`;

Agent detects user needs Gmail access
         │
         │ bytespace_connect(action: "connect", toolkit: "gmail")
         ▼
┌─────────────────┐
│  Daemon Tool     │  Calls proxy → Composio API
│                  │  Gets redirect URL
└────────┬────────┘
         │
         │ Returns auth URL to agent
         ▼
Agent presents URL to user in chat:
"Click here to connect Gmail: https://app.composio.dev/link/abc..."
         │
         │ User clicks link → browser opens
         ▼
┌─────────────────┐
│  Composio Auth   │  White-labeled OAuth screen (bot0 branding)
│  Page            │  → Google OAuth consent
│                  │  → Token exchange
│                  │  → Store encrypted
└────────┬────────┘
         │
         │ Callback to: https://bytespace.bot0.dev/api/auth-apps/callback
         │   ?status=success&connected_account_id=ca_xyz
         ▼
┌─────────────────┐
│  Bytespace       │  1. Parse callback
│  Callback Route  │  2. Notify daemon via Hub/IPC
│                  │  3. Show success page to user
└────────┬────────┘
         │
         │ Event: connection_complete
         ▼
Daemon receives notification:
"Gmail connected for user usr_abc123 (ca_xyz)"
Agent can now execute Gmail tools

// When agent returns bytespace_connect metadata with authUrl:
// Desktop intercepts and opens in system browser (not in-app)
// This is critical for OAuth — Composio's auth page needs a real browser

// In Terminal.tsx / MessageItem.tsx:
if (metadata?.type === 'bytespace_connect' && metadata.authUrl) {
  // Render a clickable button instead of raw URL
  <ComposioAuthButton
    url={metadata.authUrl}
    toolkit={metadata.toolkit}
    onComplete={() => {
      // Optionally poll connection status
    }}
  />
}

┌─────────────────────────────────────────────────────────────────┐
│                    TRIGGER LIFECYCLE                              │
│                                                                  │
│  1. REGISTRATION                                                 │
│     Agent → bytespace_trigger(create) → Proxy → Composio API     │
│     Composio stores trigger + configures webhook/poll            │
│                                                                  │
│  2. EVENT OCCURS                                                 │
│     Gmail receives email / GitHub gets PR / Slack gets message   │
│                                                                  │
│  3. COMPOSIO CAPTURES                                            │
│     Webhook: Service pushes to Composio immediately              │
│     Polling: Composio polls every 60s, detects change            │
│                                                                  │
│  4. COMPOSIO DELIVERS                                            │
│     POST /api/webhooks/composio on Bytespace                     │
│     Headers: webhook-signature, x-composio-webhook-version       │
│     Body: V3 payload with trigger data + metadata                │
│                                                                  │
│  5. BYTESPACE ROUTES                                             │
│     Verify signature → Parse payload → Resolve user/daemon       │
│     → Hub dispatch (online) or queue (offline)                   │
│                                                                  │
│  6. DAEMON PROCESSES                                             │
│     Proactive engine evaluates significance                      │
│     Permission check → Execute or ask user                       │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

// apps/bytespace/src/app/api/webhooks/composio/route.ts

import { NextRequest, NextResponse } from 'next/server';
import crypto from 'crypto';

const COMPOSIO_WEBHOOK_SECRET = process.env.COMPOSIO_WEBHOOK_SECRET!;

export async function POST(request: NextRequest) {
  const bodyText = await request.text();
  const signature = request.headers.get('webhook-signature');
  const webhookVersion = request.headers.get('x-composio-webhook-version');

  // 1. Verify HMAC-SHA256 signature
  if (!verifyWebhookSignature(bodyText, signature)) {
    return NextResponse.json({ error: 'Invalid signature' }, { status: 401 });
  }

  // 2. Parse V3 payload
  const payload = JSON.parse(bodyText);
  const {
    log_id,
    timestamp,
    type: eventType,  // e.g., "gmail_new_gmail_message"
    data: {
      user_id: composioUserId,  // = bot0 userId
      trigger_id,
      trigger_nano_id,
      connection_id,
      ...triggerData
    },
  } = payload;

  // 3. Resolve user → active daemon
  const daemon = await findActiveDaemonForUser(composioUserId);

  // 4. Create trigger event record
  const event = await db.insert(ctx0TriggerEvents).values({
    id: generateId(),
    user_id: composioUserId,
    trigger_id: trigger_nano_id,
    event_type: eventType,
    payload: triggerData,
    status: daemon ? 'dispatched' : 'pending',
    created_at: new Date(timestamp),
  });

  // 5. Route to daemon
  if (daemon) {
    // Daemon is online — dispatch via Hub
    await dispatchViaHub(daemon.device_id, {
      type: 'trigger_event',
      eventId: event.id,
      triggerType: eventType,
      triggerId: trigger_nano_id,
      payload: triggerData,
      timestamp,
    });
  }
  // If offline, event stays 'pending' — daemon polls on reconnect

  return NextResponse.json({ received: true, log_id });
}

function verifyWebhookSignature(body: string, signatureHeader: string | null): boolean {
  if (!signatureHeader) return false;

  // Format: "v1,<base64_signature>"
  const [version, signature] = signatureHeader.split(',');
  if (version !== 'v1') return false;

  const expected = crypto
    .createHmac('sha256', COMPOSIO_WEBHOOK_SECRET)
    .update(body)
    .digest('base64');

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

// packages/daemon/src/composio/trigger-handler.ts

import { Bus } from '../events/bus';
import { TriggerEvent } from '../events/events';

export interface TriggerEventPayload {
  eventId: string;
  triggerType: string;    // e.g., "gmail_new_gmail_message"
  triggerId: string;      // Composio trigger nano ID
  payload: unknown;       // Service-specific event data
  timestamp: string;
}

export class TriggerHandler {
  constructor(
    private bus: Bus,
    private getAgent: () => Bot0Agent,
  ) {}

  /**
   * Called when a trigger event arrives from the Hub.
   * This is the bridge between Composio triggers and the proactive engine.
   */
  async handleTriggerEvent(event: TriggerEventPayload): Promise<void> {
    // 1. Publish to event bus (for UI notification)
    this.bus.publish(TriggerEvent, {
      eventId: event.eventId,
      triggerType: event.triggerType,
      triggerId: event.triggerId,
      timestamp: event.timestamp,
    });

    // 2. Format trigger as a proactive task prompt
    const prompt = this.formatTriggerPrompt(event);

    // 3. Submit to proactive engine / task queue
    // The proactive engine evaluates significance, checks permissions,
    // and either executes autonomously or asks the user
    await this.submitToProactiveEngine({
      source: 'trigger',
      sourceId: event.triggerId,
      eventId: event.eventId,
      prompt,
      priority: this.inferPriority(event.triggerType),
      payload: event.payload,
    });
  }

  private formatTriggerPrompt(event: TriggerEventPayload): string {
    // Create a natural language prompt from the trigger event
    // This is what the agent "sees" when processing the trigger
    const payloadSummary = JSON.stringify(event.payload, null, 2);

    return `[TRIGGER EVENT: ${event.triggerType}]

A trigger event fired. Here are the details:

Trigger type: ${event.triggerType}
Trigger ID: ${event.triggerId}
Timestamp: ${event.timestamp}

Event data:
${payloadSummary}

Based on this event, determine the appropriate action. Consider:
1. Is this significant enough to notify the user?
2. Can you take autonomous action based on existing permissions?
3. Should you ask the user what to do?`;
  }

  private inferPriority(triggerType: string): 'critical' | 'high' | 'normal' | 'low' {
    // Payment/billing triggers are critical
    if (triggerType.includes('payment') || triggerType.includes('billing')) return 'critical';
    // Direct messages are high priority
    if (triggerType.includes('message') || triggerType.includes('mention')) return 'high';
    // Most triggers are normal
    return 'normal';
  }
}

-- New table: ctx0_trigger_events
CREATE TABLE ctx0_trigger_events (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL REFERENCES ctx0_users(id),
  trigger_id TEXT NOT NULL,           -- Composio trigger nano ID
  event_type TEXT NOT NULL,           -- e.g., "gmail_new_gmail_message"
  payload JSONB NOT NULL DEFAULT '{}',
  status TEXT NOT NULL DEFAULT 'pending',  -- pending | dispatched | processed | failed
  dispatched_at TIMESTAMPTZ,
  processed_at TIMESTAMPTZ,
  error TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  ttl_expires_at TIMESTAMPTZ          -- Auto-expire old events
);

CREATE INDEX idx_trigger_events_pending
  ON ctx0_trigger_events(user_id, status)
  WHERE status = 'pending';

// On daemon reconnect / startup:
async function processPendingTriggerEvents(
  composioClient: ProxiedComposioClient,
  triggerHandler: TriggerHandler,
  userId: string
): Promise<void> {
  const pending = await db
    .select()
    .from(ctx0TriggerEvents)
    .where(
      and(
        eq(ctx0TriggerEvents.user_id, userId),
        eq(ctx0TriggerEvents.status, 'pending'),
        // Only process events less than 24h old
        gt(ctx0TriggerEvents.created_at, sql`NOW() - INTERVAL '24 hours'`)
      )
    )
    .orderBy(ctx0TriggerEvents.created_at);

  for (const event of pending) {
    await triggerHandler.handleTriggerEvent({
      eventId: event.id,
      triggerType: event.event_type,
      triggerId: event.trigger_id,
      payload: event.payload,
      timestamp: event.created_at.toISOString(),
    });

    // Mark as dispatched
    await db.update(ctx0TriggerEvents)
      .set({ status: 'dispatched', dispatched_at: new Date() })
      .where(eq(ctx0TriggerEvents.id, event.id));
  }
}

// packages/desktop/src/components/ConnectionsPanel.tsx

// Lists active connections with:
// - Service icon + name
// - Connection status (active/expired)
// - Connected at timestamp
// - Disconnect button
// - "Connect new service" button → opens toolkit browser

// The agent can also trigger connections via bytespace_connect,
// which shows an inline auth button in the chat.

// packages/desktop/src/components/TriggersPanel.tsx

// Lists active triggers with:
// - Trigger type + description
// - Service icon
// - Status (active/paused)
// - Last fired timestamp
// - Enable/disable toggle
// - Delete button
// - "Create trigger" button → opens trigger browser

// In MessageItem.tsx, when rendering bytespace_connect tool results:
<div className="composio-auth-prompt">
  <p>To connect {toolkit}, click the button below:</p>
  <button onClick={() => shell.openExternal(authUrl)}>
    Connect {toolkit}
  </button>
  <p className="text-muted">
    This will open your browser for secure authentication.
  </p>
</div>

┌─────────────────────────────────────┐
│           DAEMON                     │
│                                      │
│  Has: session token, proxy URL       │
│  Does NOT have: Composio API key     │
│                                      │
│  All Composio calls → Proxy          │
│  Device signature on every request   │
└──────────────────┬──────────────────┘
                   │
                   │ HTTPS + device sig
                   ▼
┌─────────────────────────────────────┐
│           PROXY (Bytespace)          │
│                                      │
│  Has: Composio API key (encrypted)   │
│  Validates: session + device sig     │
│  Forwards: request to Composio       │
│  Injects: x-api-key header           │
└──────────────────┬──────────────────┘
                   │
                   ▼
┌─────────────────────────────────────┐
│           COMPOSIO API               │
│                                      │
│  Validates: API key                  │
│  Scopes: user_id isolation           │
│  Manages: OAuth tokens, credentials  │
└─────────────────────────────────────┘

// In packages/daemon/src/permissions/middleware.ts

function getToolPermission(toolName: string, input: any): PermissionRequirement | null {
  switch (toolName) {
    case 'bytespace_search':
      // Search is read-only — auto-approve in auto mode
      return null;

    case 'bytespace_execute':
      // Executing actions requires permission
      return {
        permission: 'composio:execute',
        patterns: [input.slug],
        metadata: { slug: input.slug, toolkit: input.slug?.split('_')[0] },
      };

    case 'bytespace_connect':
      if (input.action === 'list') return null; // Read-only
      return {
        permission: 'composio:connect',
        patterns: [input.toolkit],
        metadata: { toolkit: input.toolkit },
      };

    case 'bytespace_trigger':
      if (input.action === 'list' || input.action === 'list_types') return null;
      return {
        permission: 'composio:trigger',
        patterns: [input.trigger_slug ?? input.trigger_id ?? '*'],
        metadata: { action: input.action },
      };
  }
}

-- Track Composio connections locally for offline reference
-- This is a CACHE of Composio's state, not the source of truth
CREATE TABLE ctx0_bytespace_connections (
  id TEXT PRIMARY KEY,                 -- Composio connected_account nano ID (ca_xyz)
  user_id TEXT NOT NULL REFERENCES ctx0_users(id),
  toolkit TEXT NOT NULL,               -- e.g., "gmail", "github"
  status TEXT NOT NULL DEFAULT 'active',  -- active | expired | initiated
  display_name TEXT,                   -- e.g., "Work Gmail ([email protected])"
  auth_scheme TEXT,                    -- OAUTH2 | API_KEY | BASIC
  connected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  expires_at TIMESTAMPTZ,
  synced_at TIMESTAMPTZ NOT NULL DEFAULT NOW()  -- Last sync with Composio
);

-- Trigger event queue (for offline delivery)
CREATE TABLE ctx0_trigger_events (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL REFERENCES ctx0_users(id),
  trigger_id TEXT NOT NULL,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL DEFAULT '{}',
  status TEXT NOT NULL DEFAULT 'pending',
  dispatched_at TIMESTAMPTZ,
  processed_at TIMESTAMPTZ,
  error TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  ttl_expires_at TIMESTAMPTZ
);

-- Trigger subscriptions (local cache for UI display)
CREATE TABLE ctx0_bytespace_triggers (
  id TEXT PRIMARY KEY,                 -- Composio trigger nano ID
  user_id TEXT NOT NULL REFERENCES ctx0_users(id),
  slug TEXT NOT NULL,                  -- e.g., "GMAIL_NEW_GMAIL_MESSAGE"
  toolkit TEXT NOT NULL,
  config JSONB NOT NULL DEFAULT '{}',  -- Trigger configuration
  status TEXT NOT NULL DEFAULT 'active',
  description TEXT,                    -- Human-readable description
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  last_fired_at TIMESTAMPTZ,
  fire_count INTEGER NOT NULL DEFAULT 0,
  synced_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

// Create custom auth config for Gmail with bot0's OAuth credentials
const gmailAuthConfig = await composio.authConfigs.create('GMAIL', {
  name: 'bot0 Gmail',
  type: 'use_custom_auth',
  credentials: {
    client_id: process.env.GOOGLE_OAUTH_CLIENT_ID,
    client_secret: process.env.GOOGLE_OAUTH_CLIENT_SECRET,
    oauth_redirect_uri: 'https://bytespace.bot0.dev/api/auth-apps/callback',
  },
  authScheme: 'OAUTH2',
});

// apps/bytespace/src/app/api/auth-apps/callback/route.ts
//
// This is a passthrough that:
// 1. Receives the OAuth callback on bytespace.bot0.dev
// 2. Forwards to Composio's actual callback handler
// 3. Redirects user back to bot0 Desktop app

export async function GET(request: NextRequest) {
  const params = request.nextUrl.searchParams;

  // Forward to Composio's callback
  const composioCallback = new URL('https://backend.composio.dev/api/v3/toolkits/auth/callback');
  params.forEach((value, key) => composioCallback.searchParams.set(key, value));

  const response = await fetch(composioCallback.toString());

  // Parse result
  const status = params.get('status');
  const connectedAccountId = params.get('connected_account_id');

  // Redirect back to bot0 Desktop (deep link or success page)
  const redirectUrl = new URL('https://bytespace.bot0.dev/connections/success');
  if (status) redirectUrl.searchParams.set('status', status);
  if (connectedAccountId) redirectUrl.searchParams.set('connected_account_id', connectedAccountId);

  return NextResponse.redirect(redirectUrl);
}

User: "Alert me when someone stars my GitHub repo composio/bot0"
         │
         ▼
Agent reasons:
  1. Need GITHUB_STAR_EVENT trigger
  2. User must be connected to GitHub
  3. Config needs: owner=composio, repo=bot0
         │
         │ bytespace_connect(action: "list", toolkit: "github")
         ▼
Agent sees: GitHub connected ✓
         │
         │ bytespace_trigger(action: "list_types", toolkit: "github")
         ▼
Agent discovers GITHUB_STAR_EVENT trigger type
         │
         │ bytespace_trigger(action: "create",
         │   trigger_slug: "GITHUB_STAR_EVENT",
         │   trigger_config: { owner: "composio", repo: "bot0" })
         ▼
Permission prompt: "Create trigger GITHUB_STAR_EVENT for github?"
User approves ✓
         │
         ▼
Trigger created → Composio configures GitHub webhook
         │
         ▼
[Later: Someone stars the repo]
         │
         ▼
GitHub → Composio → Bytespace webhook → Hub → Daemon
         │
         ▼
Proactive engine evaluates:
  "New star on composio/bot0 by user xyz"
  Priority: low
  Action: Notify user (non-urgent)
         │
         ▼
Agent: "Your repo composio/bot0 just got a new star from @xyz! 🌟"

Collect → Batch → Evaluate → Permission → Execute/Ask/Store → Learn

// packages/daemon/src/proactive/ingestors/composio.ts

export class ComposioTriggerIngestor implements TriggerIngestor {
  readonly source = 'composio';
  readonly batchWindow = 30_000;  // 30 seconds
  readonly maxBatchSize = 50;

  /**
   * Transform a raw Composio trigger event into an observation
   * for the proactive engine's batch queue.
   */
  ingest(event: TriggerEventPayload): Observation {
    return {
      id: event.eventId,
      source: 'composio',
      sourceId: event.triggerId,
      type: event.triggerType,
      data: event.payload,
      timestamp: new Date(event.timestamp),
      priority: this.inferPriority(event.triggerType),
    };
  }

  /**
   * Batching behavior:
   * - GitHub star events: collapse to count ("5 new stars")
   * - Email triggers: collapse by sender
   * - Slack messages: collapse by channel
   * - Payment events: never batch (always critical)
   */
  batch(observations: Observation[]): BatchedObservation[] {
    // Group by trigger type + context
    const groups = groupBy(observations, o => `${o.type}:${o.sourceId}`);

    return Object.entries(groups).map(([key, obs]) => ({
      id: generateId(),
      source: 'composio',
      count: obs.length,
      triggerType: obs[0].type,
      firstAt: obs[0].timestamp,
      lastAt: obs[obs.length - 1].timestamp,
      summary: this.summarizeBatch(obs),
      observations: obs,
    }));
  }
}

// When user has multiple Gmail accounts:
const connections = await client.listConnections({ toolkit: 'gmail' });
// Returns:
// [
//   { id: 'ca_work123', toolkit: 'gmail', displayName: 'Work ([email protected])' },
//   { id: 'ca_personal456', toolkit: 'gmail', displayName: 'Personal ([email protected])' },
// ]

// Agent can specify which account to use:
await client.executeTool('GMAIL_SEND_EMAIL', {
  connected_account_id: 'ca_work123',  // Use work email
  to: '[email protected]',
  subject: 'Meeting follow-up',
  body: '...',
});

System prompt addition:

"When the user has multiple accounts for a service, ask which one to use
if the context is ambiguous. If the context is clear (e.g., 'send from
my work email'), use the appropriate account. You can list accounts with
bytespace_connect(action: 'list', toolkit: '<toolkit>')."

User: "Notify me when I get important emails"

Agent: "You have two Gmail accounts connected:
1. Work ([email protected])
2. Personal ([email protected])

Which account should I monitor for important emails? Or both?"

User: "Both"

Agent: Creates two trigger instances, one per connected account

// Permission rule examples:

// "Always allow searching for tools"
{ toolName: 'bytespace_search', permission: 'composio:search', mode: 'auto' }

// "Always allow reading Gmail (but ask for sending)"
{ toolName: 'bytespace_execute', permission: 'composio:execute',
  patterns: ['GMAIL_LIST_*', 'GMAIL_GET_*'], mode: 'auto' }

// "Ask before any write operations"
{ toolName: 'bytespace_execute', permission: 'composio:execute',
  patterns: ['*_CREATE_*', '*_SEND_*', '*_DELETE_*', '*_UPDATE_*'], mode: 'ask' }

// "Always ask before creating triggers"
{ toolName: 'bytespace_trigger', permission: 'composio:trigger',
  patterns: ['create'], mode: 'ask' }

1st time: Agent wants to send Slack message → Ask user
2nd time: Agent wants to send Slack message → Ask user
3rd time: Agent wants to send Slack message → Ask user
4th time: Agent wants to send Slack message → Ask user
5th time: Agent wants to send Slack message → Ask user
→ 5 consecutive approvals → confidence = 95% → Auto-approve future Slack sends