feat(web): sse event streaming and activity tracking

2025-12-24 03:25:52 +02:00
parent de0d7b698a
commit 6bcea3eb4d
2 changed files with 191 additions and 9 deletions
@@ -407,7 +407,6 @@ export const useEventStream = () => {
        break;
      }
      case 'openchamber:session-activity': {
-        if (!isDesktopRuntimeRef.current) break;
        const sessionId = typeof props.sessionId === 'string' ? props.sessionId : null;
        const phase = typeof props.phase === 'string' ? props.phase : null;
        if (sessionId && (phase === 'idle' || phase === 'busy' || phase === 'cooldown')) {
@@ -804,16 +803,21 @@ export const useEventStream = () => {

          completeStreamingMessage(sessionId, messageId);

+          // For web/vscode: trigger cooldown only when assistant message has finish === "stop"
+          // This matches the desktop backend logic in session_activity.rs
          if (!isDesktopRuntimeRef.current) {
-            const rawCompletedSessionId = (message as { sessionID?: string }).sessionID;
-            const completedSessionId: string =
-              typeof rawCompletedSessionId === 'string' && rawCompletedSessionId.length > 0
-                ? rawCompletedSessionId
-                : sessionId;
+            const finish = (message as { finish?: string }).finish;
+            if (finish === 'stop') {
+              const rawCompletedSessionId = (message as { sessionID?: string }).sessionID;
+              const completedSessionId: string =
+                typeof rawCompletedSessionId === 'string' && rawCompletedSessionId.length > 0
+                  ? rawCompletedSessionId
+                  : sessionId;

-            const currentPhase = sessionActivityPhaseRef.current.get(completedSessionId);
-            if (currentPhase === 'busy') {
-              updateSessionActivityPhase(completedSessionId, 'cooldown');
+              const currentPhase = sessionActivityPhaseRef.current.get(completedSessionId);
+              if (currentPhase === 'busy') {
+                updateSessionActivityPhase(completedSessionId, 'cooldown');
+              }
            }
          }

@@ -677,6 +677,61 @@ function buildOpenCodeUrl(path, prefixOverride) {
  return `http://localhost:${openCodePort}${fullPath}`;
 }

+function parseSseDataPayload(block) {
+  if (!block || typeof block !== 'string') {
+    return null;
+  }
+  const dataLines = block
+    .split('\n')
+    .filter((line) => line.startsWith('data:'))
+    .map((line) => line.slice(5).replace(/^\s/, ''));
+
+  if (dataLines.length === 0) {
+    return null;
+  }
+
+  const payloadText = dataLines.join('\n').trim();
+  if (!payloadText) {
+    return null;
+  }
+
+  try {
+    return JSON.parse(payloadText);
+  } catch {
+    return null;
+  }
+}
+
+function deriveSessionActivity(payload) {
+  if (!payload || typeof payload !== 'object') {
+    return null;
+  }
+
+  if (payload.type === 'session.status') {
+    const status = payload.properties?.status;
+    const sessionId = payload.properties?.sessionID;
+    const statusType = status?.type;
+
+    if (typeof sessionId === 'string' && sessionId.length > 0 && typeof statusType === 'string') {
+      const phase = statusType === 'busy' || statusType === 'retry' ? 'busy' : 'idle';
+      return { sessionId, phase };
+    }
+  }
+
+  if (payload.type === 'session.idle') {
+    const sessionId = payload.properties?.sessionID;
+    if (typeof sessionId === 'string' && sessionId.length > 0) {
+      return { sessionId, phase: 'idle' };
+    }
+  }
+
+  return null;
+}
+
+function writeSseEvent(res, payload) {
+  res.write(`data: ${JSON.stringify(payload)}\n\n`);
+}
+
 function extractApiPrefixFromUrl(urlString, expectedSuffix) {
  if (!urlString) {
    return null;
@@ -1764,6 +1819,129 @@ async function main(options = {}) {
    }
  });

+  app.get('/api/event', async (req, res) => {
+    if (!openCodePort) {
+      return res.status(503).json({ error: 'OpenCode service unavailable' });
+    }
+
+    if (!openCodeApiPrefixDetected) {
+      try {
+        await detectOpenCodeApiPrefix();
+      } catch {
+        // ignore detection failures
+      }
+    }
+
+    let targetUrl;
+    try {
+      const prefix = openCodeApiPrefixDetected ? openCodeApiPrefix : '';
+      targetUrl = new URL(buildOpenCodeUrl('/event', prefix));
+    } catch (error) {
+      return res.status(503).json({ error: 'OpenCode service unavailable' });
+    }
+
+    const directoryParam = Array.isArray(req.query.directory)
+      ? req.query.directory[0]
+      : req.query.directory;
+    if (typeof directoryParam === 'string' && directoryParam.trim().length > 0) {
+      targetUrl.searchParams.set('directory', directoryParam.trim());
+    }
+
+    const headers = {
+      Accept: 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive'
+    };
+
+    const lastEventId = req.header('Last-Event-ID');
+    if (typeof lastEventId === 'string' && lastEventId.length > 0) {
+      headers['Last-Event-ID'] = lastEventId;
+    }
+
+    const controller = new AbortController();
+    const cleanup = () => {
+      if (!controller.signal.aborted) {
+        controller.abort();
+      }
+    };
+
+    req.on('close', cleanup);
+    req.on('error', cleanup);
+
+    let upstream;
+    try {
+      upstream = await fetch(targetUrl.toString(), {
+        headers,
+        signal: controller.signal,
+      });
+    } catch (error) {
+      return res.status(502).json({ error: 'Failed to connect to OpenCode event stream' });
+    }
+
+    if (!upstream.ok || !upstream.body) {
+      return res.status(502).json({ error: `OpenCode event stream unavailable (${upstream.status})` });
+    }
+
+    res.setHeader('Content-Type', 'text/event-stream');
+    res.setHeader('Cache-Control', 'no-cache');
+    res.setHeader('Connection', 'keep-alive');
+    res.setHeader('X-Accel-Buffering', 'no');
+
+    if (typeof res.flushHeaders === 'function') {
+      res.flushHeaders();
+    }
+
+    const decoder = new TextDecoder();
+    const reader = upstream.body.getReader();
+    let buffer = '';
+
+    const forwardBlock = (block) => {
+      if (!block) return;
+      res.write(`${block}\n\n`);
+      const payload = parseSseDataPayload(block);
+      const activity = deriveSessionActivity(payload);
+      if (activity) {
+        writeSseEvent(res, {
+          type: 'openchamber:session-activity',
+          properties: {
+            sessionId: activity.sessionId,
+            phase: activity.phase,
+          }
+        });
+      }
+    };
+
+    try {
+      while (true) {
+        const { value, done } = await reader.read();
+        if (done) break;
+        buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n');
+
+        let separatorIndex;
+        while ((separatorIndex = buffer.indexOf('\n\n')) !== -1) {
+          const block = buffer.slice(0, separatorIndex);
+          buffer = buffer.slice(separatorIndex + 2);
+          forwardBlock(block);
+        }
+      }
+
+      if (buffer.trim().length > 0) {
+        forwardBlock(buffer.trim());
+      }
+    } catch (error) {
+      if (!controller.signal.aborted) {
+        console.warn('SSE proxy stream error:', error);
+      }
+    } finally {
+      cleanup();
+      try {
+        res.end();
+      } catch {
+        // ignore
+      }
+    }
+  });
+
  app.get('/api/config/settings', async (_req, res) => {
    try {
      const settings = await readSettingsFromDisk();