webservice

Author: linyuan.yang

packages/sbot/src/Channel/ChannelManager.ts

@@ -211,6 +211,11 @@ export class ChannelManager {
     getChannel(channelId: string) { return config.getChannel(channelId); }
     getService(channelId: string) { return this.services.get(channelId); }
 
+    /** 由内置 channel（WEB_CHANNEL_ID）等不走 plugin 路径的 service 主动注册。dispose 时统一回收。 */
+    registerService(channelId: string, service: IChannelService): void {
+        this.services.set(channelId, service);
+    }
+
     async sendText(channelId: string, sessionId: string, text: string): Promise<boolean> {
         const service = this.services.get(channelId);
         if (!service) return false;

packages/sbot/src/Channel/web/WebService.ts

@@ -0,0 +1,145 @@
+import fs from 'fs';
+import path from 'path';
+import http from 'http';
+import { randomUUID } from 'crypto';
+import { WebSocket, WebSocketServer } from 'ws';
+import { isEmptyContent, resizeImageIfNeeded, type MessageContent } from "scorpio.ai";
+import { IChannelService, ChannelSessionHandler, SessionService } from "channel.base";
+import { WsCommandType, WEB_CHANNEL_ID, WEB_CHANNEL_TYPE } from 'sbot.commons';
+import { ensureChannelSession } from "../../Core/Database";
+import { sessionManager } from "../../Session/SessionManager";
+import { LoggerService } from "../../Core/LoggerService";
+import { WebSocketSessionHandler } from "./WebSocketSessionHandler";
+
+const logger = LoggerService.getLogger("WebService.ts");
+
+type AttachmentInput = { name: string; dataUrl?: string; content?: string };
+type ContentPartInput = { type: 'text'; text: string } | { type: 'image'; dataUrl: string };
+
+function isImageDataUrl(dataUrl: string): boolean {
+    return /^data:image\//.test(dataUrl);
+}
+
+/**
+ * 将编辑器交错的 text/image parts + 附件文件构造为 MessageContent。
+ * - parts 保持编辑器顺序
+ * - 非内联的文件附件追加在末尾，落盘到 uploadDir 后以 markdown 文件链接的形式插入文本
+ */
+async function processMessage(parts: ContentPartInput[], attachments: AttachmentInput[] | undefined, uploadDir: string): Promise<MessageContent> {
+    const msgParts: Array<{ type: string; text?: string;[key: string]: any }> = [];
+    let hasImage = false;
+
+    for (const p of parts) {
+        if (p.type === 'text') {
+            msgParts.push({ type: 'text', text: p.text });
+        } else if (p.type === 'image' && p.dataUrl) {
+            const url = await resizeImageIfNeeded(p.dataUrl);
+            msgParts.push({ type: 'image_url', image_url: { url } });
+            hasImage = true;
+        }
+    }
+
+    if (attachments?.length) {
+        for (const att of attachments) {
+            if (att.dataUrl && isImageDataUrl(att.dataUrl)) {
+                const url = await resizeImageIfNeeded(att.dataUrl);
+                msgParts.push({ type: 'image_url', image_url: { url } });
+                hasImage = true;
+            } else if (att.dataUrl) {
+                const filePath = path.join(uploadDir, `${randomUUID()}-${att.name}`);
+                fs.writeFileSync(filePath, Buffer.from(att.dataUrl.replace(/^data:[^;]+;base64,/, ''), 'base64'));
+                msgParts.push({ type: 'text', text: `[file: ${att.name}](${filePath})` });
+            } else if (att.content != null) {
+                const filePath = path.join(uploadDir, `${randomUUID()}-${att.name}`);
+                fs.writeFileSync(filePath, att.content);
+                msgParts.push({ type: 'text', text: `[file: ${att.name}](${filePath})` });
+            }
+        }
+    }
+
+    if (msgParts.length === 0) return '';
+    if (!hasImage) return msgParts.map(p => p.text!).join('\n');
+    return msgParts;
+}
+
+/**
+ * Web channel 的 IChannelService 实现：管理 ws 客户端连接、广播事件、为每个 SbotSession 创建 WebSocketSessionHandler。
+ * 由 HttpServer 在创建 http.Server 后调用 attach() 完成 WS 升级路径绑定，并通过 channelManager.registerService 注册。
+ */
+export class WebService implements IChannelService {
+    private readonly wsClients = new Set<WebSocket>();
+    private wss?: WebSocketServer;
+
+    attach(server: http.Server, uploadDir: string): void {
+        const wss = this.wss = new WebSocketServer({ server, path: '/ws/chat' });
+        wss.on('connection', (ws) => {
+            this.wsClients.add(ws);
+            ws.on('close', () => { this.wsClients.delete(ws); });
+            ws.on('message', async (data) => {
+                try {
+                    const msg = JSON.parse(data.toString()) as { type?: string;[key: string]: any };
+                    const sid = msg.sessionId as string | undefined;
+                    if (!sid) throw new Error('sessionId is required');
+                    const { session, profile } = await ensureChannelSession(WEB_CHANNEL_ID, sid);
+                    const threadId = String(profile.id);
+                    switch (msg.type) {
+                        case WsCommandType.Query: {
+                            const enriched = await processMessage(msg.parts ?? [], msg.attachments, uploadDir);
+                            if (isEmptyContent(enriched)) break;
+                            sessionManager.onReceiveChannelMessage(threadId, enriched, {
+                                channelType: WEB_CHANNEL_TYPE,
+                                channelId: WEB_CHANNEL_ID,
+                                dbSessionId: session.id,
+                                sessionId: sid,
+                            });
+                            break;
+                        }
+                        case WsCommandType.Approval:
+                        case WsCommandType.Ask:
+                        case WsCommandType.Abort: {
+                            sessionManager.onTriggerChannelAction(threadId, msg.type!, msg).catch(e => logger.error(`ws trigger error: ${e?.message ?? e}`));
+                            break;
+                        }
+                    }
+                } catch (e: any) {
+                    logger.error(`ws message error: ${e?.message ?? e}`);
+                }
+            });
+        });
+    }
+
+    broadcast(data: string): void {
+        for (const ws of this.wsClients) {
+            if (ws.readyState === WebSocket.OPEN) ws.send(data);
+        }
+    }
+
+    // ── IChannelService ──
+
+    createSessionHandler(session: SessionService): ChannelSessionHandler {
+        return new WebSocketSessionHandler(session);
+    }
+
+    async sendText(_sessionId: string, _text: string): Promise<void> {
+        // Web channel 没有"原文直发"路径，输出统一通过 WebSocketSessionHandler 经 ws 广播给前端。
+    }
+
+    async sendFile(_sessionId: string, _file: string | Buffer, _fileName?: string): Promise<void> {
+        // 同上：web 没有独立的文件直发通道。
+    }
+
+    async sendNative(_sessionId: string, _payload: any): Promise<void> {
+        // 同上。
+    }
+
+    dispose(): void {
+        for (const ws of this.wsClients) {
+            try { ws.close(); } catch (_) { /* ignore */ }
+        }
+        this.wsClients.clear();
+        try { this.wss?.close(); } catch (_) { /* ignore */ }
+        this.wss = undefined;
+    }
+}
+
+export const webService = new WebService();

packages/sbot/src/Channel/web/WebSocketSessionHandler.ts

@@ -7,7 +7,7 @@ import {
     type ChannelMessageArgs, type ChatToolCall,
 } from "channel.base";
 import { WebChatEventType, WsCommandType, ApprovalTimeoutValue } from 'sbot.commons';
-import { httpServer } from "../../Server/HttpServer";
+import { webService } from "./WebService";
 
 const WEB_ASK_PROMPT = `Ask the user one or more structured questions and wait for their response. Use this tool whenever you need clarification, a decision, or input before proceeding.
 
@@ -122,6 +122,6 @@ export class WebSocketSessionHandler extends ChannelSessionHandler {
     // ── Emit helpers ──
 
     private emit(type: WebChatEventType, data: Record<string, any>): void {
-        httpServer.broadcastToWs(JSON.stringify({ sessionId: this.sessionId, type, data }));
+        webService.broadcast(JSON.stringify({ sessionId: this.sessionId, type, data }));
     }
 }

packages/sbot/src/Processing/createProcessAIHandler.ts

@@ -5,7 +5,7 @@ import { config, AgentMode } from "../Core/Config";
 import { buildExecuteTool } from "./buildExecuteTool";
 import { updateUsageStats, type UsageContext } from "./updateUsageStats";
 import { WebChatEventType, WEB_CHANNEL_ID, ApprovalTimeoutValue } from "sbot.commons";
-import { httpServer } from "../Server/HttpServer";
+import { webService } from "../Channel/web/WebService";
 import { AgentRunner } from "../Agent/AgentRunner";
 
 export function createProcessAIHandler(): ProcessAIHandler {
@@ -22,7 +22,7 @@ export function createProcessAIHandler(): ProcessAIHandler {
         if (!channel) throw new Error(`Channel config not found: ${channelId}`);
 
         if (channelId === WEB_CHANNEL_ID) {
-            httpServer.broadcastToWs(JSON.stringify({ sessionId, type: WebChatEventType.Human, data: { content: query } }));
+            webService.broadcast(JSON.stringify({ sessionId, type: WebChatEventType.Human, data: { content: query } }));
         }
 
         const agentId = resolved.agentId;

packages/sbot/src/Processing/updateUsageStats.ts

@@ -1,7 +1,7 @@
 import { type TokenUsage } from "scorpio.ai";
 import { WebChatEventType, WEB_CHANNEL_ID } from "sbot.commons";
 import { database, getChannelSession } from "../Core/Database";
-import { httpServer } from "../Server/HttpServer";
+import { webService } from "../Channel/web/WebService";
 
 export interface UsageContext {
     agentId: string;
@@ -55,7 +55,7 @@ export async function updateUsageStats(
 
     const row = await getChannelSession(dbSessionId);
     if (row && row.channelId === WEB_CHANNEL_ID) {
-        httpServer.broadcastToWs(JSON.stringify({
+        webService.broadcast(JSON.stringify({
             sessionId: row.sessionId,
             type: WebChatEventType.Usage,
             data: { inputTokens: usage.input_tokens, outputTokens: usage.output_tokens, totalTokens: usage.total_tokens, cacheCreationTokens: cacheCreation, cacheReadTokens: cacheRead },

packages/sbot/src/Server/HttpServer.ts

@@ -7,8 +7,7 @@ import { randomUUID } from 'crypto';
 import { execFile } from 'child_process';
 import { promisify } from 'util';
 import { z } from 'zod';
-import { WebSocket, WebSocketServer } from 'ws';
-import { AgentToolService, SkillService, ModelProvider, listThreadIds, isEmptyContent, resizeImageIfNeeded, setMaxImageSize, type StoredMessage, type MessageContent } from "scorpio.ai";
+import { AgentToolService, SkillService, ModelProvider, listThreadIds, setMaxImageSize, type StoredMessage } from "scorpio.ai";
 import { config, isDev, isValidAgentId } from '../Core/Config';
 import { AgentRunner } from '../Agent/AgentRunner';
 import { ACPAgentPool } from '../Agent/ACPAgentPool';
@@ -25,9 +24,10 @@ import { sessionManager } from '../Session/SessionManager';
 import { schedulerService } from '../Scheduler/SchedulerService';
 import { heartbeatService } from '../Heartbeat/HeartbeatService';
 import { channelManager } from '../Channel/ChannelManager';
-import { WsCommandType, WEB_CHANNEL_ID, WEB_CHANNEL_TYPE } from 'sbot.commons';
+import { WEB_CHANNEL_ID } from 'sbot.commons';
 import { getModelMeta, getKnownModels } from './modelCatalog';
 import { FsApi } from './FsApi';
+import { webService } from '../Channel/web/WebService';
 
 const logger = LoggerService.getLogger('HttpServer.ts');
 const execFileAsync = promisify(execFile);
@@ -317,57 +317,6 @@ function buildPromptTree(dir: string, basePath = '', userBaseDir = ''): PromptNo
     return result;
 }
 
-// ===== 附件处理 =====
-type AttachmentInput = { name: string; dataUrl?: string; content?: string };
-type ContentPartInput = { type: 'text'; text: string } | { type: 'image'; dataUrl: string };
-
-function isImageDataUrl(dataUrl: string): boolean {
-    return /^data:image\//.test(dataUrl);
-}
-
-/**
- * Build MessageContent from ordered parts (interleaved text/image) + file attachments.
- * Parts preserve the interleaved order from the editor.
- * File attachments (non-inline) are appended at the end.
- */
-async function processMessage(parts: ContentPartInput[], attachments: AttachmentInput[] | undefined, uploadDir: string): Promise<MessageContent> {
-    const msgParts: Array<{ type: string; text?: string; [key: string]: any }> = [];
-    let hasImage = false;
-
-    for (const p of parts) {
-        if (p.type === 'text') {
-            msgParts.push({ type: 'text', text: p.text });
-        } else if (p.type === 'image' && p.dataUrl) {
-            const url = await resizeImageIfNeeded(p.dataUrl);
-            msgParts.push({ type: 'image_url', image_url: { url } });
-            hasImage = true;
-        }
-    }
-
-    // Append file attachments (non-inline files from the attachment picker)
-    if (attachments?.length) {
-        for (const att of attachments) {
-            if (att.dataUrl && isImageDataUrl(att.dataUrl)) {
-                const url = await resizeImageIfNeeded(att.dataUrl);
-                msgParts.push({ type: 'image_url', image_url: { url } });
-                hasImage = true;
-            } else if (att.dataUrl) {
-                const filePath = path.join(uploadDir, `${randomUUID()}-${att.name}`);
-                fs.writeFileSync(filePath, Buffer.from(att.dataUrl.replace(/^data:[^;]+;base64,/, ''), 'base64'));
-                msgParts.push({ type: 'text', text: `[file: ${att.name}](${filePath})` });
-            } else if (att.content != null) {
-                const filePath = path.join(uploadDir, `${randomUUID()}-${att.name}`);
-                fs.writeFileSync(filePath, att.content);
-                msgParts.push({ type: 'text', text: `[file: ${att.name}](${filePath})` });
-            }
-        }
-    }
-
-    if (msgParts.length === 0) return '';
-    if (!hasImage) return msgParts.map(p => p.text!).join('\n');
-    return msgParts;
-}
-
 // ===== Skills 辅助函数 =====
 function listSkills(skillsDir: string) {
     if (!fs.existsSync(skillsDir)) return [];
@@ -422,23 +371,14 @@ function api(fn: (req: Request, res: Response) => any) {
 class HttpServer {
     private readonly skillHubService = new SkillHubService();
     private readonly agentStoreService = new AgentStoreService();
-    private readonly wsClients = new Set<WebSocket>();
     private server?: http.Server;
 
-    broadcastToWs(data: string): void {
-        for (const ws of this.wsClients) {
-            if (ws.readyState === WebSocket.OPEN) ws.send(data);
-        }
-    }
-
     async shutdown(): Promise<void> {
         logger.info('Shutting down services...');
         try {
             schedulerService.stopAll();
             await ACPAgentPool.getInstance().disposeAll();
             await channelManager.dispose();
-            for (const ws of this.wsClients) ws.close();
-            this.wsClients.clear();
             if (this.server) {
                 await new Promise<void>((resolve, reject) =>
                     this.server!.close(err => err ? reject(err) : resolve()),
@@ -499,45 +439,11 @@ class HttpServer {
         this.registerUserRoutes(app);
         this.registerChatRoutes(app);
 
-        // HTTP + WebSocket 服务
+        // HTTP + WebSocket 服务：把 ws 升级路径与 web channel 运行时交给 WebService，
+        // 然后注册到 channelManager，让消息出路与 dispose 生命周期与其他 channel 对齐
         const server = this.server = http.createServer(app);
-
-        const wss = new WebSocketServer({ server, path: '/ws/chat' });
-        wss.on('connection', (ws) => {
-            this.wsClients.add(ws);
-            ws.on('close', () => { this.wsClients.delete(ws); });
-            ws.on('message', async (data) => {
-                try {
-                    const msg = JSON.parse(data.toString()) as { type?: string; [key: string]: any };
-                    const sid = msg.sessionId as string | undefined;
-                    if (!sid) throw new Error('sessionId is required');
-                    // 与 channel plugin 路径对齐：先 ensure session+profile，再交给 sessionManager
-                    const { session, profile } = await ensureChannelSession(WEB_CHANNEL_ID, sid);
-                    const threadId = String(profile.id);
-                    switch (msg.type) {
-                        case WsCommandType.Query: {
-                            const enriched = await processMessage(msg.parts ?? [], msg.attachments, uploadDir);
-                            if (isEmptyContent(enriched)) break;
-                            sessionManager.onReceiveChannelMessage(threadId, enriched, {
-                                channelType: WEB_CHANNEL_TYPE,
-                                channelId: WEB_CHANNEL_ID,
-                                dbSessionId: session.id,
-                                sessionId: sid,
-                            });
-                            break;
-                        }
-                        case WsCommandType.Approval:
-                        case WsCommandType.Ask:
-                        case WsCommandType.Abort: {
-                            sessionManager.onTriggerChannelAction(threadId, msg.type!, msg).catch(e => logger.error(`ws trigger error: ${e?.message ?? e}`));
-                            break;
-                        }
-                    }
-                } catch (e: any) {
-                    logger.error(`ws message error: ${e?.message ?? e}`);
-                }
-            });
-        });
+        webService.attach(server, uploadDir);
+        channelManager.registerService(WEB_CHANNEL_ID, webService);
 
         server.listen(port, () => {
             logger.info(`HTTP server started, admin UI available at: http://127.0.0.1:${port}`);