feat: add direct sessions API for browser/CLI clients

packages/gateway/src/api/index.ts

@@ -0,0 +1,2 @@
+export { ApiPlatform, type ApiPlatformConfig } from "./platform";
+export { ApiResponseRenderer } from "./response-renderer";

packages/gateway/src/api/platform.ts

@@ -0,0 +1,167 @@
+#!/usr/bin/env bun
+
+/**
+ * API Platform Adapter
+ * Handles direct API access for browser extensions, CLI clients, etc.
+ * Does not require external platform integration (no Slack, Discord, etc.)
+ */
+
+import { createLogger, type InstructionProvider, type UserInteraction } from "@peerbot/core";
+import type { CoreServices, PlatformAdapter } from "../platform";
+import { ApiResponseRenderer } from "./response-renderer";
+import type { ResponseRenderer } from "../platform/response-renderer";
+import { broadcastToSession } from "../routes/public/sessions";
+
+const logger = createLogger("api-platform");
+
+/**
+ * API Platform configuration
+ */
+export interface ApiPlatformConfig {
+  /** Whether the API platform is enabled */
+  enabled?: boolean;
+}
+
+/**
+ * API Platform adapter for direct access via HTTP/SSE
+ * This platform doesn't interact with external services like Slack or Discord.
+ * Instead, it provides endpoints for:
+ * - Creating sessions
+ * - Sending messages
+ * - Receiving streaming responses via SSE
+ * - Handling tool approvals
+ */
+export class ApiPlatform implements PlatformAdapter {
+  readonly name = "api";
+
+  private services?: CoreServices;
+  private responseRenderer?: ApiResponseRenderer;
+  private isRunning = false;
+
+  constructor(private readonly config: ApiPlatformConfig = {}) {}
+
+  /**
+   * Initialize with core services
+   */
+  async initialize(services: CoreServices): Promise<void> {
+    logger.info("Initializing API platform...");
+    this.services = services;
+
+    // Create response renderer for routing worker responses to SSE clients
+    this.responseRenderer = new ApiResponseRenderer();
+
+    // Subscribe to interaction events to handle tool approvals
+    const interactionService = services.getInteractionService();
+    interactionService.on("interaction:created", (interaction) => {
+      // Only handle API platform interactions
+      if (interaction.teamId === "api" || interaction.spaceId?.startsWith("api-")) {
+        this.handleToolApproval(interaction).catch((error) => {
+          logger.error("Failed to handle tool approval:", error);
+        });
+      }
+    });
+
+    logger.info("✅ API platform initialized");
+  }
+
+  /**
+   * Start the platform
+   * For API platform, this is mostly a no-op since routes are registered separately
+   */
+  async start(): Promise<void> {
+    this.isRunning = true;
+    logger.info("✅ API platform started");
+  }
+
+  /**
+   * Stop the platform
+   */
+  async stop(): Promise<void> {
+    this.isRunning = false;
+    logger.info("✅ API platform stopped");
+  }
+
+  /**
+   * Check if platform is healthy
+   */
+  isHealthy(): boolean {
+    return this.isRunning;
+  }
+
+  /**
+   * No custom instruction provider for API platform
+   */
+  getInstructionProvider(): InstructionProvider | null {
+    return null;
+  }
+
+  /**
+   * Build deployment metadata
+   * For API sessions, we include session ID and source
+   */
+  buildDeploymentMetadata(
+    threadId: string,
+    channelId: string,
+    platformMetadata: Record<string, any>
+  ): Record<string, string> {
+    return {
+      sessionId: platformMetadata.sessionId || threadId,
+      source: "direct-api",
+      channelId,
+    };
+  }
+
+  /**
+   * Get the response renderer for routing worker responses
+   */
+  getResponseRenderer(): ResponseRenderer | undefined {
+    return this.responseRenderer;
+  }
+
+  /**
+   * Handle tool approval requests by sending them via SSE
+   */
+  private async handleToolApproval(interaction: UserInteraction): Promise<void> {
+    const sessionId = interaction.threadId;
+    if (!sessionId) {
+      logger.warn("No session ID found for tool approval interaction");
+      return;
+    }
+
+    // Send tool approval request to SSE clients
+    broadcastToSession(sessionId, "tool_approval", {
+      type: "tool_approval",
+      interactionId: interaction.id,
+      title: interaction.title,
+      message: interaction.message,
+      fields: interaction.fields,
+      buttons: interaction.buttons,
+      expiresAt: interaction.expiresAt,
+      timestamp: Date.now(),
+    });
+
+    logger.info(`Sent tool approval to session ${sessionId}: ${interaction.id}`);
+  }
+
+  /**
+   * API platform doesn't render interactions via platform UI
+   * Instead, interactions are sent via SSE to the client
+   */
+  async renderInteraction(): Promise<void> {
+    // Interactions are handled via SSE in the response renderer
+  }
+
+  /**
+   * API platform doesn't render suggestions via platform UI
+   */
+  async renderSuggestion(): Promise<void> {
+    // Suggestions are handled via SSE in the response renderer
+  }
+
+  /**
+   * API platform doesn't have thread status indicators
+   */
+  async setThreadStatus(): Promise<void> {
+    // Status is sent via SSE events
+  }
+}

packages/gateway/src/api/response-renderer.ts

@@ -0,0 +1,155 @@
+#!/usr/bin/env bun
+
+/**
+ * API Response Renderer
+ * Broadcasts worker responses to SSE connections for direct API clients
+ */
+
+import { createLogger } from "@peerbot/core";
+import type { ThreadResponsePayload } from "../infrastructure/queue/types";
+import type { ResponseRenderer } from "../platform/response-renderer";
+import { broadcastToSession } from "../routes/public/sessions";
+
+const logger = createLogger("api-response-renderer");
+
+/**
+ * Response renderer for API platform
+ * Broadcasts responses to SSE clients instead of external platforms
+ */
+export class ApiResponseRenderer implements ResponseRenderer {
+  /**
+   * Handle streaming delta content
+   * Broadcasts delta to SSE connections
+   */
+  async handleDelta(
+    payload: ThreadResponsePayload,
+    sessionKey: string
+  ): Promise<string | null> {
+    // Extract session ID from platformMetadata or thread ID
+    const sessionId =
+      (payload.platformMetadata?.sessionId as string) || payload.threadId;
+
+    if (!sessionId) {
+      logger.warn("No session ID found in payload for delta broadcast");
+      return null;
+    }
+
+    // Broadcast delta to SSE clients
+    broadcastToSession(sessionId, "output", {
+      type: "delta",
+      content: payload.delta,
+      timestamp: payload.timestamp || Date.now(),
+      messageId: payload.messageId,
+    });
+
+    logger.debug(
+      `Broadcast delta to session ${sessionId}: ${payload.delta?.length || 0} chars`
+    );
+
+    return payload.messageId;
+  }
+
+  /**
+   * Handle completion of response processing
+   * Sends completion event to SSE clients
+   */
+  async handleCompletion(
+    payload: ThreadResponsePayload,
+    sessionKey: string
+  ): Promise<void> {
+    const sessionId =
+      (payload.platformMetadata?.sessionId as string) || payload.threadId;
+
+    if (!sessionId) {
+      logger.warn("No session ID found in payload for completion broadcast");
+      return;
+    }
+
+    // Broadcast completion to SSE clients
+    broadcastToSession(sessionId, "complete", {
+      type: "complete",
+      messageId: payload.messageId,
+      processedMessageIds: payload.processedMessageIds,
+      timestamp: payload.timestamp || Date.now(),
+    });
+
+    logger.info(`Broadcast completion to session ${sessionId}`);
+  }
+
+  /**
+   * Handle error response
+   * Sends error event to SSE clients
+   */
+  async handleError(
+    payload: ThreadResponsePayload,
+    sessionKey: string
+  ): Promise<void> {
+    const sessionId =
+      (payload.platformMetadata?.sessionId as string) || payload.threadId;
+
+    if (!sessionId) {
+      logger.warn("No session ID found in payload for error broadcast");
+      return;
+    }
+
+    // Broadcast error to SSE clients
+    broadcastToSession(sessionId, "error", {
+      type: "error",
+      error: payload.error,
+      messageId: payload.messageId,
+      timestamp: payload.timestamp || Date.now(),
+    });
+
+    logger.error(`Broadcast error to session ${sessionId}: ${payload.error}`);
+  }
+
+  /**
+   * Handle status updates (heartbeat with elapsed time)
+   * Sends status event to SSE clients
+   */
+  async handleStatusUpdate(payload: ThreadResponsePayload): Promise<void> {
+    const sessionId =
+      (payload.platformMetadata?.sessionId as string) || payload.threadId;
+
+    if (!sessionId) {
+      return;
+    }
+
+    // Broadcast status to SSE clients
+    broadcastToSession(sessionId, "status", {
+      type: "status",
+      status: payload.statusUpdate,
+      messageId: payload.messageId,
+      timestamp: payload.timestamp || Date.now(),
+    });
+  }
+
+  /**
+   * Handle ephemeral messages
+   * For API platform, these are just broadcast as regular events
+   */
+  async handleEphemeral(payload: ThreadResponsePayload): Promise<void> {
+    const sessionId =
+      (payload.platformMetadata?.sessionId as string) || payload.threadId;
+
+    if (!sessionId) {
+      return;
+    }
+
+    // Broadcast ephemeral content to SSE clients
+    broadcastToSession(sessionId, "ephemeral", {
+      type: "ephemeral",
+      content: payload.content,
+      messageId: payload.messageId,
+      timestamp: payload.timestamp || Date.now(),
+    });
+  }
+
+  /**
+   * Stop stream for thread - no-op for API platform
+   * SSE connections handle their own lifecycle
+   */
+  async stopStreamForThread(_userId: string, _threadId: string): Promise<void> {
+    // No-op - SSE connections manage their own lifecycle
+  }
+}

packages/gateway/src/cli/gateway.ts

@@ -173,6 +173,29 @@ function setupHealthEndpoints(
     logger.info("✅ Messaging routes enabled at :8080/api/messaging/send");
   }
 
+  // Setup sessions API routes (direct API access without platform adapters)
+  if (coreServices) {
+    const queueProducer = coreServices.getQueueProducer();
+    const sessionMgr = coreServices.getSessionManager();
+    const interactionSvc = coreServices.getInteractionService();
+    const publicUrl = coreServices.getPublicGatewayUrl();
+
+    if (queueProducer && sessionMgr && interactionSvc) {
+      const { Router } = require("express");
+      const sessionsRouter = Router();
+      const { registerSessionsRoutes } = require("../routes/public/sessions");
+      registerSessionsRoutes(
+        sessionsRouter,
+        queueProducer,
+        sessionMgr,
+        interactionSvc,
+        publicUrl
+      );
+      proxyApp.use(sessionsRouter);
+      logger.info("✅ Sessions API routes enabled at :8080/api/sessions/*");
+    }
+  }
+
   // Setup auth callback routes for WhatsApp and other non-modal platforms
   if (coreServices) {
     const stateStore = coreServices.getClaudeOAuthStateStore();
@@ -273,6 +296,12 @@ export async function startGateway(
     logger.info("✅ WhatsApp platform registered");
   }
 
+  // Register API platform (always enabled for direct API access)
+  const { ApiPlatform } = await import("../api");
+  const apiPlatform = new ApiPlatform({ enabled: true });
+  gateway.registerPlatform(apiPlatform);
+  logger.info("✅ API platform registered");
+
   // Start gateway (initializes core services + platforms)
   await gateway.start();
   logger.info("✅ Gateway started");

packages/gateway/src/infrastructure/queue/types.ts

@@ -101,7 +101,7 @@ export interface ThreadResponsePayload {
   threadId: string;
   userId: string;
   teamId: string;
-  platform?: string; // Platform identifier (slack, whatsapp, etc.) for multi-platform routing
+  platform?: string; // Platform identifier (slack, whatsapp, api, etc.) for multi-platform routing
   content?: string; // Used only for ephemeral messages (OAuth/auth flows)
   delta?: string;
   isFullReplacement?: boolean;
@@ -116,4 +116,5 @@ export interface ThreadResponsePayload {
     elapsedSeconds: number;
     state: string;
   };
+  platformMetadata?: Record<string, unknown>; // Platform-specific metadata (e.g., sessionId for API)
 }