From 2d65d38b113e5f165bbea8933b2d89fa931dc4eb Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Wed, 27 May 2026 02:49:58 +0000 Subject: [PATCH 1/6] feat: add web WebSocket channel for A2UI app Co-Authored-By: Claude Sonnet 4.6 --- src/channels/channel-identity.ts | 2 +- src/channels/registry.ts | 2 + src/channels/web/plugin.ts | 47 +++++++++ src/channels/web/service.ts | 158 ++++++++++++++++++++++++++++ src/channels/web/session-routing.ts | 43 ++++++++ src/channels/web/transport.ts | 39 +++++++ src/control/runtime-health-store.ts | 12 ++- 7 files changed, 301 insertions(+), 2 deletions(-) create mode 100644 src/channels/web/plugin.ts create mode 100644 src/channels/web/service.ts create mode 100644 src/channels/web/session-routing.ts create mode 100644 src/channels/web/transport.ts diff --git a/src/channels/channel-identity.ts b/src/channels/channel-identity.ts index d1f24985..de1ae0e5 100644 --- a/src/channels/channel-identity.ts +++ b/src/channels/channel-identity.ts @@ -1,5 +1,5 @@ export type ChannelIdentity = { - platform: "slack" | "telegram"; + platform: "slack" | "telegram" | "web"; botId?: string; accountId?: string; conversationKind: "dm" | "channel" | "group" | "topic"; diff --git a/src/channels/registry.ts b/src/channels/registry.ts index 83a08b0a..150b61dd 100644 --- a/src/channels/registry.ts +++ b/src/channels/registry.ts @@ -2,10 +2,12 @@ import type { RuntimeChannel } from "../control/runtime-health-store.ts"; import type { ChannelPlugin } from "./channel-plugin.ts"; import { slackChannelPlugin } from "./slack/plugin.ts"; import { telegramChannelPlugin } from "./telegram/plugin.ts"; +import { webChannelPlugin } from "./web/plugin.ts"; const CHANNEL_PLUGINS: ChannelPlugin[] = [ slackChannelPlugin, telegramChannelPlugin, + webChannelPlugin, ]; export function listChannelPlugins() { diff --git a/src/channels/web/plugin.ts b/src/channels/web/plugin.ts new file mode 100644 index 00000000..55f643dd --- /dev/null +++ b/src/channels/web/plugin.ts @@ -0,0 +1,47 @@ +import type { ChannelPlugin } from "../channel-plugin.ts"; +import type { WebBotConfig } from "./service.ts"; +import { WebRuntimeService } from "./service.ts"; + +function getWebConfig(loadedConfig: Parameters[0]): WebBotConfig | null { + const raw = loadedConfig.raw as { bots?: { web?: { apiKey?: string; port?: number; agentId?: string; ownerId?: string } } }; + const web = raw.bots?.web; + if (!web?.apiKey) return null; + return { + port: web.port ?? 3099, + apiKey: web.apiKey, + agentId: web.agentId ?? loadedConfig.raw.agents.defaults.defaultAgentId ?? "default", + ownerId: web.ownerId, + }; +} + +export const webChannelPlugin: ChannelPlugin = { + id: "web", + isEnabled: (loadedConfig) => !!getWebConfig(loadedConfig), + listBots: (loadedConfig) => { + const config = getWebConfig(loadedConfig); + if (!config) return []; + return [{ botId: "default", config }]; + }, + createRuntimeService: (context, bot) => + new WebRuntimeService( + context.loadedConfig, + context.agentService, + bot.config as WebBotConfig, + context.reportLifecycle, + ), + renderHealthSummary: (state) => { + switch (state) { + case "starting": return "Web channel is starting."; + case "disabled": return "Web channel is disabled (no apiKey configured)."; + case "stopped": return "Web channel is stopped."; + } + }, + renderActiveHealthSummary: () => "Web channel WebSocket server active.", + markStartupFailure: async (store, error) => { + await store.markWebFailure(error); + }, + runMessageCommand: async (_loadedConfig, _command) => { + return { botId: "default", result: { ok: false, error: "Web channel does not support CLI message commands." } }; + }, + resolveMessageReplyTarget: () => null, +}; diff --git a/src/channels/web/service.ts b/src/channels/web/service.ts new file mode 100644 index 00000000..cfeb6090 --- /dev/null +++ b/src/channels/web/service.ts @@ -0,0 +1,158 @@ +import type { AgentService } from "../../agents/agent-service.ts"; +import type { LoadedConfig } from "../../config/load-config.ts"; +import type { ProcessedEventsStore } from "../processed-events-store.ts"; +import type { ActivityStore } from "../../control/activity-store.ts"; +import type { ChannelRuntimeLifecycleEvent, ChannelRuntimeService } from "../channel-plugin.ts"; +import { processChannelInteraction } from "../interaction-processing.ts"; +import { resolveWebConversationTarget } from "./session-routing.ts"; +import { postWebText, reconcileWebText, sendWebDone, sendWebError } from "./transport.ts"; +import { resolveChannelAuth } from "../../auth/resolve.ts"; +import { DEFAULT_PROTECTED_CONTROL_RULE } from "../../auth/defaults.ts"; + +export type WebBotConfig = { + port: number; + apiKey: string; + agentId: string; + ownerId?: string; +}; + +export type WebSocketData = { + authenticated: boolean; + todoId?: string; +}; + +type IncomingMessage = { + type: "message"; + text: string; + todoId?: string; +}; + +const DEFAULT_MAX_CHARS = 32_000; + +export class WebRuntimeService implements ChannelRuntimeService { + private server: ReturnType | null = null; + + constructor( + private readonly loadedConfig: LoadedConfig, + private readonly agentService: AgentService, + private readonly botConfig: WebBotConfig, + private readonly reportLifecycle: (event: ChannelRuntimeLifecycleEvent) => Promise, + ) {} + + async start() { + const { port, apiKey, agentId } = this.botConfig; + const agentService = this.agentService; + const loadedConfig = this.loadedConfig; + + this.server = Bun.serve({ + port, + fetch(req, server) { + const url = new URL(req.url); + const key = url.searchParams.get("key"); + if (key !== apiKey) { + return new Response("Unauthorized", { status: 401 }); + } + const upgraded = server.upgrade(req, { + data: { authenticated: true }, + }); + if (upgraded) return undefined; + return new Response("Upgrade required", { status: 426 }); + }, + websocket: { + open(ws) { + ws.send(JSON.stringify({ type: "ready" })); + }, + async message(ws, raw) { + let msg: IncomingMessage; + try { + msg = JSON.parse(String(raw)); + } catch { + sendWebError(ws, "Invalid JSON"); + return; + } + + if (msg.type !== "message" || !msg.text?.trim()) return; + + const conversationTarget = resolveWebConversationTarget({ + loadedConfig, + agentId, + todoId: msg.todoId, + }); + + const identity = { + platform: "web" as const, + botId: "default", + conversationKind: "dm" as const, + senderId: "web:owner", + senderName: "Bobby", + }; + + const auth = resolveChannelAuth({ + config: loadedConfig.raw, + agentId, + identity, + }); + + // Grant owner-level access — the API key already authenticated this user. + const elevatedAuth = { + ...auth, + appRole: "owner" as const, + agentRole: "owner" as const, + mayBypassPairing: true, + mayManageProtectedResources: true, + canUseShell: true, + }; + + const route = { + agentId, + requireMention: false, + allowBots: false, + commandPrefixes: { slash: "/", at: "@" }, + streaming: "latest" as const, + response: "all" as const, + responseMode: "message-tool" as const, + additionalMessageMode: "queue" as const, + surfaceNotifications: { loopStart: "off" as const, loopEnd: "off" as const }, + verbose: "off" as const, + followUp: { mode: "disabled" as const }, + }; + + try { + await processChannelInteraction({ + agentService, + sessionTarget: conversationTarget, + identity, + auth: elevatedAuth, + senderId: "web:owner", + text: msg.text, + route, + maxChars: DEFAULT_MAX_CHARS, + postText: (text) => postWebText(ws, text), + reconcileText: (chunks, text) => reconcileWebText(ws, chunks, text), + }); + sendWebDone(ws); + } catch (err) { + console.error("[web-channel] interaction error", err); + sendWebError(ws, "Agent error"); + sendWebDone(ws); + } + }, + close(ws) { + // nothing to clean up per-connection + }, + }, + }); + + await this.reportLifecycle({ + connection: "active", + summary: `Web channel listening on port ${port}`, + }); + + console.log(`[web-channel] WebSocket server started on ws://localhost:${port}`); + } + + async stop() { + this.server?.stop(true); + this.server = null; + } +} diff --git a/src/channels/web/session-routing.ts b/src/channels/web/session-routing.ts new file mode 100644 index 00000000..7f7ec1bf --- /dev/null +++ b/src/channels/web/session-routing.ts @@ -0,0 +1,43 @@ +import type { LoadedConfig } from "../../config/load-config.ts"; +import { + buildAgentMainSessionKey, + buildAgentPeerSessionKey, +} from "../../agents/session-key.ts"; + +export type WebConversationTarget = { + agentId: string; + sessionKey: string; + mainSessionKey: string; + todoId?: string; +}; + +export function resolveWebConversationTarget(params: { + loadedConfig: LoadedConfig; + agentId: string; + todoId?: string; +}): WebConversationTarget { + const sessionConfig = params.loadedConfig.raw.session; + const mainSessionKey = buildAgentMainSessionKey({ + agentId: params.agentId, + mainKey: sessionConfig.mainKey, + }); + + // Per-todo sessions get their own conversation context; otherwise falls back to main DM. + const peerId = params.todoId ? `todo:${params.todoId}` : "web:main"; + + return { + agentId: params.agentId, + sessionKey: buildAgentPeerSessionKey({ + agentId: params.agentId, + mainKey: sessionConfig.mainKey, + channel: "web", + botId: "default", + peerKind: "dm", + peerId, + dmScope: sessionConfig.dmScope, + identityLinks: sessionConfig.identityLinks, + }), + mainSessionKey, + todoId: params.todoId, + }; +} diff --git a/src/channels/web/transport.ts b/src/channels/web/transport.ts new file mode 100644 index 00000000..bcf25bb5 --- /dev/null +++ b/src/channels/web/transport.ts @@ -0,0 +1,39 @@ +import type { ServerWebSocket } from "bun"; +import type { WebSocketData } from "./service.ts"; + +export type WebChunk = { id: string; text: string }; + +let chunkCounter = 0; +function nextChunkId() { + return `wc${++chunkCounter}`; +} + +export async function postWebText( + ws: ServerWebSocket, + text: string, +): Promise { + const chunk: WebChunk = { id: nextChunkId(), text }; + ws.send(JSON.stringify({ type: "chunk", id: chunk.id, text })); + return [chunk]; +} + +export async function reconcileWebText( + ws: ServerWebSocket, + chunks: WebChunk[], + text: string, +): Promise { + if (chunks.length === 0) { + return postWebText(ws, text); + } + const firstId = chunks[0].id; + ws.send(JSON.stringify({ type: "update", id: firstId, text })); + return [{ id: firstId, text }]; +} + +export function sendWebDone(ws: ServerWebSocket) { + ws.send(JSON.stringify({ type: "done" })); +} + +export function sendWebError(ws: ServerWebSocket, message: string) { + ws.send(JSON.stringify({ type: "error", message })); +} diff --git a/src/control/runtime-health-store.ts b/src/control/runtime-health-store.ts index a6071400..75daa8ca 100644 --- a/src/control/runtime-health-store.ts +++ b/src/control/runtime-health-store.ts @@ -3,7 +3,7 @@ import { fileExists, readTextFile, writeTextFile } from "../shared/fs.ts"; import { ensureDir, getDefaultRuntimeHealthPath } from "../shared/paths.ts"; import { renderCliCommand } from "../shared/cli-name.ts"; -export type RuntimeChannel = "slack" | "telegram"; +export type RuntimeChannel = "slack" | "telegram" | "web"; export type RuntimeChannelConnection = | "disabled" | "stopped" @@ -207,6 +207,16 @@ export class RuntimeHealthStore { }); } + async markWebFailure(error: unknown) { + await this.setChannel({ + channel: "web", + connection: "failed", + summary: "Web channel failed to start.", + detail: error instanceof Error ? error.message : String(error), + actions: ["Check bots.web.port and bots.web.apiKey in your config."], + }); + } + async setReload(params: { status: "success" | "failed"; reason: "initial" | "watch"; From 68cee467459e0c43b1170d5e5161203936e3682e Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Thu, 28 May 2026 18:35:27 +0000 Subject: [PATCH 2/6] refactor(web): generic MCP proxy, CHANNEL_EVENT annotations, client-side context - Add invoke_tool WebSocket handler: proxies to any configured MCP server via stdio - Add mcp-client.ts: lightweight JSON-RPC 2.0 stdio client with pending-map concurrency - Add get_history handler: reads Claude Code session JSONL for chat history replay - Replace BRIEF_UPDATE with generic [[CHANNEL_EVENT:key:value]] extraction: strips markers from the visible stream, emits typed 'annotation' WS events - Move context injection out of the channel into the client (text field now carries full context) - Add web.senderName config field (defaults to 'Owner') - Add web.mcpServers config schema for declaring allowed MCP servers Co-Authored-By: Claude Sonnet 4.6 --- src/channels/web/history-reader.ts | 85 +++++++ src/channels/web/mcp-client.ts | 93 ++++++++ src/channels/web/plugin.ts | 12 +- src/channels/web/service.ts | 370 +++++++++++++++++++++-------- src/channels/web/transport.ts | 21 +- src/config/schema.ts | 12 + 6 files changed, 485 insertions(+), 108 deletions(-) create mode 100644 src/channels/web/history-reader.ts create mode 100644 src/channels/web/mcp-client.ts diff --git a/src/channels/web/history-reader.ts b/src/channels/web/history-reader.ts new file mode 100644 index 00000000..75bf27bf --- /dev/null +++ b/src/channels/web/history-reader.ts @@ -0,0 +1,85 @@ +import { createReadStream } from "node:fs"; +import { createInterface } from "node:readline"; +import { join } from "node:path"; +import { homedir } from "node:os"; +import type { HistoryMessage } from "./transport.ts"; + +/** + * Reads chat history from the Claude Code session JSONL file. + * Returns last `limit` user/assistant message pairs. + */ +export async function readSessionHistory(params: { + sessionId: string; + workspacePath: string; + limit?: number; +}): Promise { + const { sessionId, workspacePath, limit = 50 } = params; + + // Derive project dir: replace / and . with - + const projectDir = workspacePath.replace(/[/.]/g, "-"); + const jsonlPath = join(homedir(), ".claude", "projects", projectDir, `${sessionId}.jsonl`); + + const messages: HistoryMessage[] = []; + + try { + const stream = createReadStream(jsonlPath, { encoding: "utf-8" }); + const rl = createInterface({ input: stream, crlfDelay: Infinity }); + + for await (const line of rl) { + if (!line.trim()) continue; + let entry: Record; + try { entry = JSON.parse(line); } catch { continue; } + + const type = entry.type as string; + if (type !== "user" && type !== "assistant") continue; + + const msg = entry.message as Record | undefined; + if (!msg) continue; + + const role = msg.role as string; + if (role !== "user" && role !== "assistant") continue; + + const content = msg.content; + let text = ""; + + if (typeof content === "string") { + text = content; + } else if (Array.isArray(content)) { + text = content + .filter((c): c is Record => typeof c === "object" && c !== null) + .filter((c) => c.type === "text") + .map((c) => c.text as string) + .join("\n"); + } + + // Skip empty, strip context injection block (our own prefixes). + // Prefer the [USER_INPUT] marker (new sessions); fall back to lastIndexOf('\n\n') for old ones. + text = text.trim(); + if (!text) continue; + if (text.startsWith("[Context:") || text.startsWith("[iCloud Reminder:") || text.startsWith("[Task:")) { + const markerIdx = text.indexOf('\n[USER_INPUT]\n'); + if (markerIdx !== -1) { + text = text.slice(markerIdx + '\n[USER_INPUT]\n'.length).trim(); + } else { + const sep = text.lastIndexOf('\n\n'); + text = sep !== -1 ? text.slice(sep + 2).trim() : ''; + } + if (!text) continue; + } + + // Strip [[CHANNEL_EVENT:...]] markers (and legacy [[BRIEF_UPDATE:...]]) + text = text.replace(/\[\[CHANNEL_EVENT:[^\]]*?\]\]/gi, "").trim(); + text = text.replace(/\[\[BRIEF_UPDATE:\s*[\s\S]*?\]\]/gi, "").trim(); + text = text.replace(/\[\[CHANNEL_EVENT:[\s\S]*$/i, "").trim(); + if (!text) continue; + + messages.push({ role: role as "user" | "assistant", text }); + } + } catch { + // File not found or unreadable — return empty + return []; + } + + // Return last N messages + return messages.slice(-limit); +} diff --git a/src/channels/web/mcp-client.ts b/src/channels/web/mcp-client.ts new file mode 100644 index 00000000..7cc50e04 --- /dev/null +++ b/src/channels/web/mcp-client.ts @@ -0,0 +1,93 @@ +import { spawn } from "node:child_process"; +import { createInterface } from "node:readline"; + +export interface McpServerConfig { + command: string; + args: string[]; + env?: Record; +} + +interface Pending { + resolve: (v: unknown) => void; + reject: (e: Error) => void; +} + +export interface McpClient { + callTool(tool: string, args: Record): Promise; + close(): void; +} + +export function createMcpClient(config: McpServerConfig): McpClient { + const proc = spawn(config.command, config.args, { + env: { ...process.env, ...(config.env ?? {}) }, + stdio: ["pipe", "pipe", "inherit"], + }); + + const pending = new Map(); + let msgId = 1; + let initResolve: (() => void) | null = null; + const initPromise = new Promise((r) => { initResolve = r; }); + let initialized = false; + + const send = (msg: object) => { + proc.stdin.write(JSON.stringify(msg) + "\n"); + }; + + const rl = createInterface({ input: proc.stdout! }); + rl.on("line", (line) => { + let msg: { id?: number; result?: unknown; error?: { message: string } }; + try { msg = JSON.parse(line); } catch { return; } + if (msg.id !== undefined) { + const p = pending.get(msg.id); + if (p) { + pending.delete(msg.id); + if (msg.error) p.reject(new Error(msg.error.message)); + else p.resolve(msg.result); + } + if (!initialized) { + initialized = true; + initResolve?.(); + } + } + }); + + proc.on("exit", () => { + for (const p of pending.values()) p.reject(new Error("MCP process exited")); + pending.clear(); + }); + + // Perform initialize handshake + const initId = msgId++; + send({ + jsonrpc: "2.0", id: initId, + method: "initialize", + params: { + protocolVersion: "2024-11-05", + capabilities: {}, + clientInfo: { name: "clisbot-web", version: "1.0.0" }, + }, + }); + initPromise.then(() => { + send({ jsonrpc: "2.0", method: "notifications/initialized", params: {} }); + }); + + return { + async callTool(tool, args) { + await initPromise; + const id = msgId++; + const result = await new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }); + send({ jsonrpc: "2.0", id, method: "tools/call", params: { name: tool, arguments: args } }); + }); + // Unwrap MCP content envelope: { content: [{ type: "text", text: "..." }] } + const content = (result as { content?: Array<{ type: string; text: string }> }).content; + if (content?.[0]?.type === "text") { + try { return JSON.parse(content[0].text); } catch { return content[0].text; } + } + return result; + }, + close() { + try { proc.stdin.end(); proc.kill(); } catch { /* already dead */ } + }, + }; +} diff --git a/src/channels/web/plugin.ts b/src/channels/web/plugin.ts index 55f643dd..978191cb 100644 --- a/src/channels/web/plugin.ts +++ b/src/channels/web/plugin.ts @@ -3,7 +3,15 @@ import type { WebBotConfig } from "./service.ts"; import { WebRuntimeService } from "./service.ts"; function getWebConfig(loadedConfig: Parameters[0]): WebBotConfig | null { - const raw = loadedConfig.raw as { bots?: { web?: { apiKey?: string; port?: number; agentId?: string; ownerId?: string } } }; + const raw = loadedConfig.raw as { + bots?: { + web?: { + apiKey?: string; port?: number; agentId?: string; ownerId?: string; + senderName?: string; + mcpServers?: Record }>; + }; + }; + }; const web = raw.bots?.web; if (!web?.apiKey) return null; return { @@ -11,6 +19,8 @@ function getWebConfig(loadedConfig: Parameters[0]): apiKey: web.apiKey, agentId: web.agentId ?? loadedConfig.raw.agents.defaults.defaultAgentId ?? "default", ownerId: web.ownerId, + senderName: web.senderName, + mcpServers: web.mcpServers, }; } diff --git a/src/channels/web/service.ts b/src/channels/web/service.ts index cfeb6090..1e22f483 100644 --- a/src/channels/web/service.ts +++ b/src/channels/web/service.ts @@ -1,36 +1,65 @@ +import { createServer } from "node:http"; +import type { IncomingMessage } from "node:http"; +import { WebSocketServer, WebSocket } from "ws"; import type { AgentService } from "../../agents/agent-service.ts"; import type { LoadedConfig } from "../../config/load-config.ts"; -import type { ProcessedEventsStore } from "../processed-events-store.ts"; -import type { ActivityStore } from "../../control/activity-store.ts"; import type { ChannelRuntimeLifecycleEvent, ChannelRuntimeService } from "../channel-plugin.ts"; import { processChannelInteraction } from "../interaction-processing.ts"; import { resolveWebConversationTarget } from "./session-routing.ts"; -import { postWebText, reconcileWebText, sendWebDone, sendWebError } from "./transport.ts"; +import { + postWebText, + reconcileWebText, + sendWebDone, + sendWebError, + sendWebHistory, + sendAnnotation, +} from "./transport.ts"; import { resolveChannelAuth } from "../../auth/resolve.ts"; -import { DEFAULT_PROTECTED_CONTROL_RULE } from "../../auth/defaults.ts"; +import { readSessionHistory } from "./history-reader.ts"; +import { createMcpClient, type McpClient, type McpServerConfig } from "./mcp-client.ts"; export type WebBotConfig = { port: number; apiKey: string; agentId: string; ownerId?: string; + senderName?: string; + // MCP servers the web UI is allowed to invoke directly. + // Keys are server names; values are spawn configs. + mcpServers?: Record; }; -export type WebSocketData = { - authenticated: boolean; - todoId?: string; -}; - -type IncomingMessage = { - type: "message"; - text: string; - todoId?: string; -}; +type IncomingWsMessage = + | { type: "message"; text: string; todoId?: string } + | { type: "get_history"; todoId: string } + | { type: "invoke_tool"; server: string; tool: string; args?: Record }; const DEFAULT_MAX_CHARS = 32_000; +// Generic structured annotation extraction. +// Agents can embed [[CHANNEL_EVENT:key:value]] markers; the channel strips them +// from the visible stream and emits typed WS events so any client can handle them. +const CHANNEL_EVENT_RE = /\[\[CHANNEL_EVENT:([^:\]]+):([\s\S]*?)\]\]/gi; + +function extractChannelEvents(text: string): { + text: string; + events: Array<{ key: string; value: string }>; +} { + const events: Array<{ key: string; value: string }> = []; + let stripped = text.replace(CHANNEL_EVENT_RE, (_match, key, value) => { + events.push({ key: key.trim(), value: value.trim() }); + return ""; + }); + // Strip incomplete marker at end of partial stream + stripped = stripped.replace(/\[\[CHANNEL_EVENT:[\s\S]*$/, ""); + return { text: stripped.trim(), events }; +} + export class WebRuntimeService implements ChannelRuntimeService { - private server: ReturnType | null = null; + private server: ReturnType | null = null; + private wss: WebSocketServer | null = null; + // Persistent MCP client connections, one per configured server + private mcpClients = new Map(); constructor( private readonly loadedConfig: LoadedConfig, @@ -39,120 +68,259 @@ export class WebRuntimeService implements ChannelRuntimeService { private readonly reportLifecycle: (event: ChannelRuntimeLifecycleEvent) => Promise, ) {} + private getMcpClient(serverName: string): McpClient | null { + if (this.mcpClients.has(serverName)) return this.mcpClients.get(serverName)!; + const config = this.botConfig.mcpServers?.[serverName]; + if (!config) return null; + const client = createMcpClient(config); + this.mcpClients.set(serverName, client); + return client; + } + async start() { const { port, apiKey, agentId } = this.botConfig; const agentService = this.agentService; const loadedConfig = this.loadedConfig; - this.server = Bun.serve({ - port, - fetch(req, server) { - const url = new URL(req.url); - const key = url.searchParams.get("key"); - if (key !== apiKey) { - return new Response("Unauthorized", { status: 401 }); + // In-memory OG image cache: url → {image, ts} + const ogCache = new Map(); + const OG_CACHE_TTL = 3600_000; // 1h + + const httpServer = createServer(async (req, res) => { + // OG image proxy — returns {"image":"..."} or {"image":null} + if (req.method === "GET" && req.url?.startsWith("/og?")) { + const targetUrl = new URL(req.url, "http://localhost").searchParams.get("url"); + res.setHeader("Access-Control-Allow-Origin", "*"); + res.setHeader("Content-Type", "application/json"); + if (!targetUrl) { res.writeHead(400); res.end('{"image":null}'); return; } + + const cached = ogCache.get(targetUrl); + if (cached && Date.now() - cached.ts < OG_CACHE_TTL) { + res.writeHead(200); + res.end(JSON.stringify({ image: cached.image })); + return; } - const upgraded = server.upgrade(req, { - data: { authenticated: true }, - }); - if (upgraded) return undefined; - return new Response("Upgrade required", { status: 426 }); - }, - websocket: { - open(ws) { - ws.send(JSON.stringify({ type: "ready" })); - }, - async message(ws, raw) { - let msg: IncomingMessage; - try { - msg = JSON.parse(String(raw)); - } catch { - sendWebError(ws, "Invalid JSON"); - return; - } - if (msg.type !== "message" || !msg.text?.trim()) return; + try { + const html = await fetch(targetUrl, { + signal: AbortSignal.timeout(5000), + headers: { "User-Agent": "Mozilla/5.0 (compatible; clisbot/1.0)" }, + }).then(r => r.text()); + const match = html.match(/]+(?:property=["']og:image["']|name=["']og:image["'])[^>]+content=["']([^"']+)["']/i) + ?? html.match(/]+content=["']([^"']+)["'][^>]+(?:property=["']og:image["']|name=["']og:image["'])/i); + const image = match?.[1] ?? null; + ogCache.set(targetUrl, { image, ts: Date.now() }); + res.writeHead(200); + res.end(JSON.stringify({ image })); + } catch { + ogCache.set(targetUrl, { image: null, ts: Date.now() }); + res.writeHead(200); + res.end('{"image":null}'); + } + return; + } + + res.writeHead(426, { "Content-Type": "text/plain" }); + res.end("Upgrade required"); + }); + + const wss = new WebSocketServer({ noServer: true }); + + httpServer.on("upgrade", (req: IncomingMessage, socket, head) => { + const url = new URL(req.url ?? "/", `http://localhost`); + const key = url.searchParams.get("key"); + if (key !== apiKey) { + socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n"); + socket.destroy(); + return; + } + wss.handleUpgrade(req, socket, head, (ws) => { + wss.emit("connection", ws, req); + }); + }); + + wss.on("connection", (ws: WebSocket) => { + ws.send(JSON.stringify({ type: "ready" })); + + ws.on("message", async (raw) => { + let msg: IncomingWsMessage; + try { + msg = JSON.parse(String(raw)); + } catch { + sendWebError(ws, "Invalid JSON"); + return; + } - const conversationTarget = resolveWebConversationTarget({ + // ── History request ────────────────────────────────────────────────── + if (msg.type === "get_history") { + const target = resolveWebConversationTarget({ loadedConfig, agentId, todoId: msg.todoId, }); - const identity = { - platform: "web" as const, - botId: "default", - conversationKind: "dm" as const, - senderId: "web:owner", - senderName: "Bobby", - }; + const sessionEntry = await agentService.sessionState.getEntry(target.sessionKey).catch(() => null); + if (sessionEntry?.sessionId && sessionEntry.workspacePath) { + const history = await readSessionHistory({ + sessionId: sessionEntry.sessionId, + workspacePath: sessionEntry.workspacePath, + limit: 60, + }); + sendWebHistory(ws, msg.todoId, history); + } else { + sendWebHistory(ws, msg.todoId, []); + } + return; + } - const auth = resolveChannelAuth({ - config: loadedConfig.raw, - agentId, + // ── MCP tool invocation ────────────────────────────────────────────── + if (msg.type === "invoke_tool") { + const client = this.getMcpClient(msg.server); + if (!client) { + ws.send(JSON.stringify({ + type: "tool_result", server: msg.server, tool: msg.tool, + args: msg.args, error: `MCP server '${msg.server}' is not configured`, + })); + return; + } + try { + const result = await client.callTool(msg.tool, msg.args ?? {}); + ws.send(JSON.stringify({ + type: "tool_result", server: msg.server, tool: msg.tool, + args: msg.args, result, + })); + } catch (err) { + ws.send(JSON.stringify({ + type: "tool_result", server: msg.server, tool: msg.tool, + args: msg.args, error: String(err), + })); + } + return; + } + + // ── Regular message ────────────────────────────────────────────────── + if (msg.type !== "message" || !msg.text?.trim()) return; + + const conversationTarget = resolveWebConversationTarget({ + loadedConfig, + agentId, + todoId: msg.todoId, + }); + + const identity = { + platform: "web" as const, + botId: "default", + conversationKind: "dm" as const, + senderId: "web:owner", + senderName: this.botConfig.senderName ?? "Owner", + }; + + const auth = resolveChannelAuth({ + config: loadedConfig.raw, + agentId, + identity, + }); + + const elevatedAuth = { + ...auth, + appRole: "owner" as const, + agentRole: "owner" as const, + mayBypassPairing: true, + mayManageProtectedResources: true, + canUseShell: true, + }; + + const route = { + agentId, + requireMention: false, + allowBots: false, + commandPrefixes: { slash: ["/"], bash: ["!"] }, + streaming: "latest" as const, + response: "all" as const, + responseMode: "message-tool" as const, + additionalMessageMode: "queue" as const, + surfaceNotifications: { loopStart: "off" as const, loopEnd: "off" as const }, + verbose: "off" as const, + followUp: { mode: "disabled" as const }, + }; + + let accumulatedText = ""; + let activeChunks: { id: string; text: string }[] = []; + + try { + await processChannelInteraction({ + agentService, + sessionTarget: conversationTarget, identity, + auth: elevatedAuth, + senderId: "web:owner", + text: msg.text, + route, + maxChars: DEFAULT_MAX_CHARS, + postText: async (text) => { + accumulatedText = text; + const { text: cleanText } = extractChannelEvents(text); + if (activeChunks.length === 0) { + activeChunks = await postWebText(ws, cleanText); + } else { + activeChunks = await reconcileWebText(ws, activeChunks, cleanText); + } + return activeChunks; + }, + reconcileText: async (_chunks, text) => { + accumulatedText = text; + const { text: cleanText } = extractChannelEvents(text); + if (activeChunks.length === 0) { + activeChunks = await postWebText(ws, cleanText); + } else { + activeChunks = await reconcileWebText(ws, activeChunks, cleanText); + } + return activeChunks; + }, }); - // Grant owner-level access — the API key already authenticated this user. - const elevatedAuth = { - ...auth, - appRole: "owner" as const, - agentRole: "owner" as const, - mayBypassPairing: true, - mayManageProtectedResources: true, - canUseShell: true, - }; - - const route = { - agentId, - requireMention: false, - allowBots: false, - commandPrefixes: { slash: "/", at: "@" }, - streaming: "latest" as const, - response: "all" as const, - responseMode: "message-tool" as const, - additionalMessageMode: "queue" as const, - surfaceNotifications: { loopStart: "off" as const, loopEnd: "off" as const }, - verbose: "off" as const, - followUp: { mode: "disabled" as const }, - }; - - try { - await processChannelInteraction({ - agentService, - sessionTarget: conversationTarget, - identity, - auth: elevatedAuth, - senderId: "web:owner", - text: msg.text, - route, - maxChars: DEFAULT_MAX_CHARS, - postText: (text) => postWebText(ws, text), - reconcileText: (chunks, text) => reconcileWebText(ws, chunks, text), - }); - sendWebDone(ws); - } catch (err) { - console.error("[web-channel] interaction error", err); - sendWebError(ws, "Agent error"); - sendWebDone(ws); + const { events } = extractChannelEvents(accumulatedText); + for (const event of events) { + sendAnnotation(ws, msg.todoId, event.key, event.value); } - }, - close(ws) { - // nothing to clean up per-connection - }, - }, + + sendWebDone(ws); + } catch (err) { + console.error("[web-channel] interaction error", err); + sendWebError(ws, "Agent error"); + sendWebDone(ws); + } + }); }); + await new Promise((resolve, reject) => { + httpServer.listen(port, "0.0.0.0", () => resolve()); + httpServer.once("error", reject); + }); + + // Pre-warm MCP connections so the first user request isn't slow + for (const serverName of Object.keys(this.botConfig.mcpServers ?? {})) { + const client = this.getMcpClient(serverName); + client?.callTool("get_all_reminders", {}).catch(() => { /* ignore warmup errors */ }); + } + + this.server = httpServer; + this.wss = wss; + await this.reportLifecycle({ connection: "active", - summary: `Web channel listening on port ${port}`, + summary: `Web channel WebSocket server active.`, }); console.log(`[web-channel] WebSocket server started on ws://localhost:${port}`); } async stop() { - this.server?.stop(true); + for (const client of this.mcpClients.values()) client.close(); + this.mcpClients.clear(); + this.wss?.close(); + this.server?.close(); + this.wss = null; this.server = null; } } diff --git a/src/channels/web/transport.ts b/src/channels/web/transport.ts index bcf25bb5..4d65c149 100644 --- a/src/channels/web/transport.ts +++ b/src/channels/web/transport.ts @@ -1,15 +1,24 @@ -import type { ServerWebSocket } from "bun"; -import type { WebSocketData } from "./service.ts"; +import type { WebSocket } from "ws"; export type WebChunk = { id: string; text: string }; +export type HistoryMessage = { role: "user" | "assistant"; text: string }; + +export function sendWebHistory(ws: WebSocket, todoId: string, messages: HistoryMessage[]) { + ws.send(JSON.stringify({ type: "history", todoId, messages })); +} + +export function sendAnnotation(ws: WebSocket, todoId: string | undefined, key: string, value: string) { + ws.send(JSON.stringify({ type: "annotation", todoId, key, value })); +} + let chunkCounter = 0; function nextChunkId() { return `wc${++chunkCounter}`; } export async function postWebText( - ws: ServerWebSocket, + ws: WebSocket, text: string, ): Promise { const chunk: WebChunk = { id: nextChunkId(), text }; @@ -18,7 +27,7 @@ export async function postWebText( } export async function reconcileWebText( - ws: ServerWebSocket, + ws: WebSocket, chunks: WebChunk[], text: string, ): Promise { @@ -30,10 +39,10 @@ export async function reconcileWebText( return [{ id: firstId, text }]; } -export function sendWebDone(ws: ServerWebSocket) { +export function sendWebDone(ws: WebSocket) { ws.send(JSON.stringify({ type: "done" })); } -export function sendWebError(ws: ServerWebSocket, message: string) { +export function sendWebError(ws: WebSocket, message: string) { ws.send(JSON.stringify({ type: "error", message })); } diff --git a/src/config/schema.ts b/src/config/schema.ts index 2b54b194..86fc57da 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -971,6 +971,18 @@ export const clisbotConfigSchema = z.object({ }, }, } as any), + web: z.object({ + apiKey: z.string(), + port: z.number().int().default(3099), + agentId: z.string().optional(), + ownerId: z.string().optional(), + senderName: z.string().optional(), + mcpServers: z.record(z.object({ + command: z.string(), + args: z.array(z.string()), + env: z.record(z.string()).optional(), + })).optional(), + }).optional(), }), agents: z.object({ defaults: agentsDefaultsSchema.default({ From 1f12b1479ff9816943c3af6edf79004509c5c88a Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Thu, 28 May 2026 18:43:29 +0000 Subject: [PATCH 3/6] =?UTF-8?q?refactor(web):=20rename=20todoId=20?= =?UTF-8?q?=E2=86=92=20contextId,=20clean=20up=20history-reader?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - All WS protocol fields use contextId (generic opaque thread ID) - history-reader strips context via [USER_INPUT] marker only — no app-specific prefix checks - history-reader strips [[CHANNEL_EVENT:...]] only — no legacy BRIEF_UPDATE Co-Authored-By: Claude Sonnet 4.6 --- src/channels/web/history-reader.ts | 19 ++++++------------- src/channels/web/service.ts | 14 +++++++------- src/channels/web/session-routing.ts | 8 ++++---- src/channels/web/transport.ts | 8 ++++---- 4 files changed, 21 insertions(+), 28 deletions(-) diff --git a/src/channels/web/history-reader.ts b/src/channels/web/history-reader.ts index 75bf27bf..f01b5cc6 100644 --- a/src/channels/web/history-reader.ts +++ b/src/channels/web/history-reader.ts @@ -52,24 +52,17 @@ export async function readSessionHistory(params: { .join("\n"); } - // Skip empty, strip context injection block (our own prefixes). - // Prefer the [USER_INPUT] marker (new sessions); fall back to lastIndexOf('\n\n') for old ones. + // Strip context prefix injected by clients (everything before [USER_INPUT] marker). text = text.trim(); if (!text) continue; - if (text.startsWith("[Context:") || text.startsWith("[iCloud Reminder:") || text.startsWith("[Task:")) { - const markerIdx = text.indexOf('\n[USER_INPUT]\n'); - if (markerIdx !== -1) { - text = text.slice(markerIdx + '\n[USER_INPUT]\n'.length).trim(); - } else { - const sep = text.lastIndexOf('\n\n'); - text = sep !== -1 ? text.slice(sep + 2).trim() : ''; - } + const markerIdx = text.indexOf('\n[USER_INPUT]\n'); + if (markerIdx !== -1) { + text = text.slice(markerIdx + '\n[USER_INPUT]\n'.length).trim(); if (!text) continue; } - // Strip [[CHANNEL_EVENT:...]] markers (and legacy [[BRIEF_UPDATE:...]]) - text = text.replace(/\[\[CHANNEL_EVENT:[^\]]*?\]\]/gi, "").trim(); - text = text.replace(/\[\[BRIEF_UPDATE:\s*[\s\S]*?\]\]/gi, "").trim(); + // Strip [[CHANNEL_EVENT:...]] annotations from historical assistant messages. + text = text.replace(/\[\[CHANNEL_EVENT:[\s\S]*?\]\]/gi, "").trim(); text = text.replace(/\[\[CHANNEL_EVENT:[\s\S]*$/i, "").trim(); if (!text) continue; diff --git a/src/channels/web/service.ts b/src/channels/web/service.ts index 1e22f483..f2a584ec 100644 --- a/src/channels/web/service.ts +++ b/src/channels/web/service.ts @@ -30,8 +30,8 @@ export type WebBotConfig = { }; type IncomingWsMessage = - | { type: "message"; text: string; todoId?: string } - | { type: "get_history"; todoId: string } + | { type: "message"; text: string; contextId?: string } + | { type: "get_history"; contextId: string } | { type: "invoke_tool"; server: string; tool: string; args?: Record }; const DEFAULT_MAX_CHARS = 32_000; @@ -156,7 +156,7 @@ export class WebRuntimeService implements ChannelRuntimeService { const target = resolveWebConversationTarget({ loadedConfig, agentId, - todoId: msg.todoId, + contextId: msg.contextId, }); const sessionEntry = await agentService.sessionState.getEntry(target.sessionKey).catch(() => null); @@ -166,9 +166,9 @@ export class WebRuntimeService implements ChannelRuntimeService { workspacePath: sessionEntry.workspacePath, limit: 60, }); - sendWebHistory(ws, msg.todoId, history); + sendWebHistory(ws, msg.contextId, history); } else { - sendWebHistory(ws, msg.todoId, []); + sendWebHistory(ws, msg.contextId, []); } return; } @@ -204,7 +204,7 @@ export class WebRuntimeService implements ChannelRuntimeService { const conversationTarget = resolveWebConversationTarget({ loadedConfig, agentId, - todoId: msg.todoId, + contextId: msg.contextId, }); const identity = { @@ -281,7 +281,7 @@ export class WebRuntimeService implements ChannelRuntimeService { const { events } = extractChannelEvents(accumulatedText); for (const event of events) { - sendAnnotation(ws, msg.todoId, event.key, event.value); + sendAnnotation(ws, msg.contextId, event.key, event.value); } sendWebDone(ws); diff --git a/src/channels/web/session-routing.ts b/src/channels/web/session-routing.ts index 7f7ec1bf..357cd6f1 100644 --- a/src/channels/web/session-routing.ts +++ b/src/channels/web/session-routing.ts @@ -8,13 +8,13 @@ export type WebConversationTarget = { agentId: string; sessionKey: string; mainSessionKey: string; - todoId?: string; + contextId?: string; }; export function resolveWebConversationTarget(params: { loadedConfig: LoadedConfig; agentId: string; - todoId?: string; + contextId?: string; }): WebConversationTarget { const sessionConfig = params.loadedConfig.raw.session; const mainSessionKey = buildAgentMainSessionKey({ @@ -23,7 +23,7 @@ export function resolveWebConversationTarget(params: { }); // Per-todo sessions get their own conversation context; otherwise falls back to main DM. - const peerId = params.todoId ? `todo:${params.todoId}` : "web:main"; + const peerId = params.contextId ? `todo:${params.contextId}` : "web:main"; return { agentId: params.agentId, @@ -38,6 +38,6 @@ export function resolveWebConversationTarget(params: { identityLinks: sessionConfig.identityLinks, }), mainSessionKey, - todoId: params.todoId, + contextId: params.contextId, }; } diff --git a/src/channels/web/transport.ts b/src/channels/web/transport.ts index 4d65c149..8c27db39 100644 --- a/src/channels/web/transport.ts +++ b/src/channels/web/transport.ts @@ -4,12 +4,12 @@ export type WebChunk = { id: string; text: string }; export type HistoryMessage = { role: "user" | "assistant"; text: string }; -export function sendWebHistory(ws: WebSocket, todoId: string, messages: HistoryMessage[]) { - ws.send(JSON.stringify({ type: "history", todoId, messages })); +export function sendWebHistory(ws: WebSocket, contextId: string, messages: HistoryMessage[]) { + ws.send(JSON.stringify({ type: "history", contextId, messages })); } -export function sendAnnotation(ws: WebSocket, todoId: string | undefined, key: string, value: string) { - ws.send(JSON.stringify({ type: "annotation", todoId, key, value })); +export function sendAnnotation(ws: WebSocket, contextId: string | undefined, key: string, value: string) { + ws.send(JSON.stringify({ type: "annotation", contextId, key, value })); } let chunkCounter = 0; From 0b5ad33f2778391d9a891b308f10cfd30f54922e Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Thu, 28 May 2026 18:44:04 +0000 Subject: [PATCH 4/6] =?UTF-8?q?fix(web):=20rename=20session=20peerId=20pre?= =?UTF-8?q?fix=20todo:=20=E2=86=92=20ctx:?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/channels/web/session-routing.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/channels/web/session-routing.ts b/src/channels/web/session-routing.ts index 357cd6f1..3f8b2bd5 100644 --- a/src/channels/web/session-routing.ts +++ b/src/channels/web/session-routing.ts @@ -22,8 +22,8 @@ export function resolveWebConversationTarget(params: { mainKey: sessionConfig.mainKey, }); - // Per-todo sessions get their own conversation context; otherwise falls back to main DM. - const peerId = params.contextId ? `todo:${params.contextId}` : "web:main"; + // Per-context sessions get their own conversation thread; otherwise falls back to main DM. + const peerId = params.contextId ? `ctx:${params.contextId}` : "web:main"; return { agentId: params.agentId, From e8b0e15ab303e9b0a4bd6cf1396e5904a10f1458 Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Thu, 28 May 2026 23:29:10 +0000 Subject: [PATCH 5/6] fix(web): use capture-pane responseMode to reliably emit CHANNEL_EVENT annotations message-tool responseMode waits 3s for a tool-final reply that never arrives (Claude CLI just outputs text, no send_message tool), then returns early without calling renderResponseText on the completed snapshot. This left accumulatedText as the last streaming chunk value, which was unreliable for CHANNEL_EVENT extraction. capture-pane takes the paneManagedDelivery path which always calls renderResponseText with the final completed snapshot before returning, guaranteeing accumulatedText contains the full response including any [[CHANNEL_EVENT:brief:...]] markers. Co-Authored-By: Claude Sonnet 4.6 --- src/channels/web/service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/channels/web/service.ts b/src/channels/web/service.ts index f2a584ec..e54078d3 100644 --- a/src/channels/web/service.ts +++ b/src/channels/web/service.ts @@ -237,7 +237,7 @@ export class WebRuntimeService implements ChannelRuntimeService { commandPrefixes: { slash: ["/"], bash: ["!"] }, streaming: "latest" as const, response: "all" as const, - responseMode: "message-tool" as const, + responseMode: "capture-pane" as const, additionalMessageMode: "queue" as const, surfaceNotifications: { loopStart: "off" as const, loopEnd: "off" as const }, verbose: "off" as const, From e1c34bd2387966159bf5adf0474d933814c2d307 Mon Sep 17 00:00:00 2001 From: Bobby Marko Date: Fri, 29 May 2026 04:51:45 +0000 Subject: [PATCH 6/6] fix(web): deliver final message from JSONL to preserve markdown code fences MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude Code renders markdown in the terminal, so tmux pane capture strips code fence language identifiers (e.g. \`\`\`a2ui:contact\`). Reading the final assistant message directly from the session JSONL (raw API response) ensures code fences reach the web client intact. Also add 'web' as a passthrough platform in renderPlatformInteraction so the web channel bypasses Slack/Telegram formatting entirely — the web client handles its own markdown rendering. Co-Authored-By: Claude Sonnet 4.6 --- src/channels/rendering.ts | 6 ++++-- src/channels/web/service.ts | 27 +++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/src/channels/rendering.ts b/src/channels/rendering.ts index ef239759..26f7b6ec 100644 --- a/src/channels/rendering.ts +++ b/src/channels/rendering.ts @@ -13,7 +13,7 @@ export type ChannelRenderedMessageState = { }; export function buildRenderedMessageState(params: { - platform: "slack" | "telegram"; + platform: "slack" | "telegram" | "web"; status: "queued" | "running" | "completed" | "timeout" | "detached" | "error"; snapshot: string; queuePosition?: number; @@ -45,7 +45,7 @@ export function buildRenderedMessageState(params: { } export function renderPlatformInteraction(params: { - platform: "slack" | "telegram"; + platform: "slack" | "telegram" | "web"; status: "queued" | "running" | "completed" | "timeout" | "detached" | "error"; content: string; maxChars: number; @@ -54,6 +54,8 @@ export function renderPlatformInteraction(params: { allowTranscriptInspection?: boolean; responsePolicy?: "all" | "final"; }) { + // Web channel receives raw markdown — no platform rendering needed. + if (params.platform === "web") return params.content; return params.platform === "telegram" ? renderTelegramInteraction(params) : renderSlackInteraction(params); diff --git a/src/channels/web/service.ts b/src/channels/web/service.ts index e54078d3..5443d2d3 100644 --- a/src/channels/web/service.ts +++ b/src/channels/web/service.ts @@ -279,6 +279,33 @@ export class WebRuntimeService implements ChannelRuntimeService { }, }); + // Claude Code renders markdown in the terminal, so the tmux-captured text + // has code fences stripped. Read the final assistant message directly from + // the session JSONL (raw API response) and send it as the authoritative text. + const sessionEntry = await agentService.sessionState.getEntry(conversationTarget.sessionKey).catch(() => null); + if (sessionEntry?.sessionId && sessionEntry.workspacePath) { + const history = await readSessionHistory({ + sessionId: sessionEntry.sessionId, + workspacePath: sessionEntry.workspacePath, + limit: 10, + }); + const lastAssistant = history.filter(m => m.role === "assistant").at(-1); + if (lastAssistant?.text) { + const { text: cleanFinal, events: finalEvents } = extractChannelEvents(lastAssistant.text); + if (activeChunks.length === 0) { + activeChunks = await postWebText(ws, cleanFinal); + } else { + await reconcileWebText(ws, activeChunks, cleanFinal); + } + // Use JSONL-derived events (more reliable than tmux-captured accumulatedText) + for (const event of finalEvents) { + sendAnnotation(ws, msg.contextId, event.key, event.value); + } + sendWebDone(ws); + return; + } + } + const { events } = extractChannelEvents(accumulatedText); for (const event of events) { sendAnnotation(ws, msg.contextId, event.key, event.value);