Merge pull request #7 from mariuspruvot/fix/e2e-bugs-batch-2

fix: e2e SSE streaming bugs (batch 2)

Author: mariuspruvot

CLAUDE.md

@@ -47,6 +47,7 @@ infra/
 - **Container orchestration**: `container` module manages ephemeral Docker lifecycle, credential injection, result relay
 - **Skills as agents**: each skill is a self-contained folder with workflow definitions, mounted into containers
 - **SSE passthrough**: backend relays container output to frontend (no AI response generation in backend)
+- **Stream-json protocol**: containers emit NDJSON with 5 event types: `system` (init/retry), `assistant` (one event per content block — thinking/text/tool_use), `user` (tool_result), `result` (session end + metadata), `rate_limit_event`. The `result.result` field duplicates the last assistant text — only display assistant events, use result for status only. No `--include-partial-messages` flag, so no `stream_event` deltas.
 - **API prefix**: all routes under `/api/v1`
 - **Admin panel**: SQLAdmin at `/admin`, configured in `admin/views.py`
 
@@ -87,6 +88,7 @@ Required `.env` at repo root (see docker-compose.yml):
 ## Gotchas
 
 - Always run `make lint` before pushing — ruff + eslint must pass
+- **Debug SSE pipeline**: `claude -p "prompt" --output-format stream-json --verbose 2>/dev/null` captures raw stream-json to validate event parsing
 - DB migrations: `make migrate` inside Docker, or `cd apps/api && uv run alembic upgrade head` locally
 - Test conftest **must** set env vars before importing from `helprs.*`
 - **Agent-readiness**: This repo must be fully understandable by a fresh Claude Code instance with no prior context. Keep docs and CLAUDE.md accurate.

apps/api/src/helprs/modules/container/router.py

@@ -114,13 +114,29 @@ async def get_container_session(
 
 
 @router.get("/sessions/{session_id}/stream")
-@limiter.limit("10/minute")
+@limiter.limit("60/minute")
 async def stream_container_output(
     session_id: UUID,
     request: Request,
     db: DbSession,
+    offset: int = 0,
 ):
-    """SSE endpoint streaming container stdout/stderr."""
+    """SSE endpoint streaming container stdout/stderr.
+
+    Accepts an ``offset`` query parameter: the number of events to skip.
+    Clients should pass the last received event ``id`` so that reconnects
+    resume from where they left off instead of replaying the full log.
+
+    Also reads the ``Last-Event-ID`` header (sent automatically by
+    EventSource on native auto-reconnect) as a fallback when the query
+    parameter is absent or zero.
+    """
+    # EventSource auto-reconnect sends Last-Event-ID header, not query params.
+    if offset == 0:
+        last_event_id = request.headers.get("last-event-id", "")
+        if last_event_id.isdigit():
+            offset = int(last_event_id)
+
     cs = await get_session_or_404(db, session_id)
 
     if cs.status != ContainerStatus.RUNNING or not cs.container_id:
@@ -130,7 +146,7 @@ async def stream_container_output(
 
     async def _event_stream():
         try:
-            async for event in stream_output(docker, cs.container_id):
+            async for event in stream_output(docker, cs.container_id, offset=offset):
                 yield event
         finally:
             await docker.close()

apps/api/src/helprs/modules/container/service.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import contextlib
 import json
 import os
@@ -137,21 +138,24 @@ async def remove_container(self, container_id: str, force: bool = False) -> None
 
     async def container_logs(self, container_id: str, follow: bool = False) -> AsyncIterator[str]:
         container = await self._docker.containers.get(container_id)
-        async for line in container.log(stdout=True, stderr=True, follow=follow):
+        # Only read stdout — Claude Code writes stream-json to stdout.
+        # stderr contains verbose/diagnostic output that duplicates events.
+        async for line in container.log(stdout=True, stderr=False, follow=follow):
             yield line
 
     async def write_to_container(self, container_id: str, data: str) -> None:
         """Write a message to the container's FIFO via docker exec.
 
         The entrypoint reads from /tmp/claude-input FIFO, so we write there.
+        Uses base64 encoding to safely pass arbitrary JSON through the shell
+        without risking interpretation of special characters ($, `, !, etc.).
         """
         container = await self._docker.containers.get(container_id)
+        encoded = base64.b64encode(data.encode()).decode()
         exec_obj = await container.exec(
-            cmd=["bash", "-c", f"echo {json.dumps(data)} > /tmp/claude-input"],
-            stdout=False,
-            stderr=False,
+            cmd=["sh", "-c", f"echo {encoded} | base64 -d > /tmp/claude-input"],
         )
-        await exec_obj.start()
+        await exec_obj.start(detach=True)
 
     async def wait_container(self, container_id: str) -> int:
         container = await self._docker.containers.get(container_id)
@@ -293,10 +297,49 @@ async def start_container(
 async def stream_output(
     docker: DockerClient,
     container_id: str,
+    offset: int = 0,
 ) -> AsyncIterator[str]:
-    """Async generator yielding container log lines as SSE events."""
-    async for line in docker.container_logs(container_id, follow=True):
-        yield f"data: {line}\n\n"
+    """Async generator yielding container log lines as SSE events.
+
+    Docker may split long stdout lines (e.g. stream-json tool_result
+    events with full file contents) across multiple log frames.  We
+    buffer chunks and only emit complete newline-delimited lines so
+    the frontend always receives valid, parseable JSON per SSE event.
+
+    Each event includes an incrementing ``id:`` field so that clients
+    can resume from the last received event via the ``offset`` query
+    parameter (number of events to skip).
+
+    When the container is quiet (Claude is thinking), the Docker log
+    stream produces no data.  We send SSE comments (``:``) every
+    ``KEEPALIVE_INTERVAL`` seconds to prevent idle-timeout disconnects.
+    """
+    keepalive_interval = 15.0
+    event_id = 0
+    buffer = ""
+
+    log_iter = docker.container_logs(container_id, follow=True).__aiter__()
+    while True:
+        try:
+            chunk = await asyncio.wait_for(log_iter.__anext__(), timeout=keepalive_interval)
+            buffer += chunk
+            while "\n" in buffer:
+                line, buffer = buffer.split("\n", 1)
+                line = line.strip()
+                if not line:
+                    continue
+                event_id += 1
+                if event_id <= offset:
+                    continue
+                yield f"id: {event_id}\ndata: {line}\n\n"
+        except TimeoutError:
+            # No data from container — send SSE keepalive comment to
+            # prevent the HTTP connection from being closed by proxies
+            # or the browser.  SSE comments are silently ignored by
+            # EventSource clients.
+            yield ": keepalive\n\n"
+        except StopAsyncIteration:
+            break
 
 
 async def send_message(

apps/web/src/features/session/ContainerSession.test.tsx

@@ -171,7 +171,7 @@ describe('ContainerSession', () => {
       expect(mockEventSources.length).toBeGreaterThan(0)
     })
 
-    expect(mockedBuildStreamUrl).toHaveBeenCalledWith('test-session-id', 'test-token')
+    expect(mockedBuildStreamUrl).toHaveBeenCalledWith('test-session-id', 'test-token', 0)
   })
 
   test('calls onBack when back button is clicked', async () => {

apps/web/src/features/session/ContainerSession.tsx

@@ -50,10 +50,15 @@ export default function ContainerSession({
   const [sending, setSending] = useState(false)
 
   const lineIdRef = useRef(0)
+  const sseOffsetRef = useRef(0)
+  const reconnectCountRef = useRef(0)
   const eventSourceRef = useRef<EventSource | null>(null)
   const mountedRef = useRef(true)
   const inputRef = useRef<HTMLInputElement>(null)
 
+  const MAX_RECONNECTS = 5
+  const RECONNECT_BASE_DELAY_MS = 2000
+
   const appendLine = useCallback((text: string, kind?: TerminalLine['kind']) => {
     if (!mountedRef.current) return
     lineIdRef.current += 1
@@ -71,6 +76,8 @@ export default function ContainerSession({
     mountedRef.current = true
     const abortController = new AbortController()
     lineIdRef.current = 0
+    sseOffsetRef.current = 0
+    reconnectCountRef.current = 0
     setLines([])
 
     async function init() {
@@ -109,6 +116,19 @@ export default function ContainerSession({
     }
 
     async function pollSessionStatus(sessionId: string) {
+      reconnectCountRef.current += 1
+      if (reconnectCountRef.current > MAX_RECONNECTS) {
+        setStatus('failed')
+        setError('Lost connection to session')
+        appendLine('[error] Lost connection after multiple retries.')
+        return
+      }
+
+      // Exponential backoff: 2s, 4s, 8s, 16s, 32s
+      const delay = RECONNECT_BASE_DELAY_MS * Math.pow(2, reconnectCountRef.current - 1)
+      await new Promise((r) => setTimeout(r, delay))
+      if (!mountedRef.current) return
+
       try {
         const fresh = await getContainerSession(sessionId)
         if (!mountedRef.current) return
@@ -119,7 +139,6 @@ export default function ContainerSession({
           setError('Container failed')
           appendLine('[error] Container failed.')
         } else if (fresh.status === 'running') {
-          appendLine('[reconnecting] Stream dropped, container still running...')
           connectStream(sessionId)
         }
       } catch {
@@ -137,18 +156,24 @@ export default function ContainerSession({
         return
       }
 
-      const url = buildStreamUrl(sessionId, accessToken)
+      const url = buildStreamUrl(sessionId, accessToken, sseOffsetRef.current)
       const source = new EventSource(url, { withCredentials: true })
       eventSourceRef.current = source
 
       source.addEventListener('open', () => {
         if (!mountedRef.current) return
         setStatus('running')
+        // Connection succeeded — reset reconnect counter
+        reconnectCountRef.current = 0
       })
 
       // Default message event — parse Claude Code stream-json
       source.onmessage = (event: MessageEvent) => {
         if (!mountedRef.current) return
+        // Track the last event id so reconnects can skip already-received events
+        if (event.lastEventId) {
+          sseOffsetRef.current = parseInt(event.lastEventId, 10) || sseOffsetRef.current
+        }
         const parsed = parseStreamEvent(event.data as string)
         if (parsed) {
           appendLine(parsed.text, parsed.kind)

apps/web/src/features/session/TerminalOutput.tsx

@@ -89,17 +89,11 @@ export default function TerminalOutput({ lines, isRunning }: TerminalOutputProps
             key={line.id}
             className="whitespace-pre-wrap break-words"
             style={
-              line.kind === 'tool'
-                ? { color: '#6e6e73', fontSize: '12px' }
-                : line.kind === 'system'
-                  ? { color: '#6e6e73', fontSize: '12px', fontStyle: 'italic' }
-                  : line.kind === 'error'
-                    ? { color: '#ff6961' }
-                    : line.kind === 'user'
-                      ? { color: '#E2A039' }
-                      : line.kind === 'result'
-                        ? { color: '#E0E0E0', marginTop: '16px', paddingTop: '16px', borderTop: '1px solid rgba(255,255,255,0.08)' }
-                        : undefined
+              line.kind === 'error'
+                ? { color: '#ff6961' }
+                : line.kind === 'status'
+                  ? { color: '#9a9898', fontStyle: 'italic' }
+                  : undefined
             }
           >
             {line.text}

apps/web/src/features/session/containerApi.ts

@@ -83,7 +83,12 @@ export async function sendMessage(sessionId: string, content: string): Promise<v
 /**
  * Build the SSE stream URL for a container session.
  * Auth token is appended as a query parameter since EventSource cannot send headers.
+ * Optional `offset` skips already-received events on reconnect.
  */
-export function buildStreamUrl(sessionId: string, accessToken: string): string {
-  return `${API_BASE}/api/v1/containers/sessions/${sessionId}/stream?access_token=${encodeURIComponent(accessToken)}`
+export function buildStreamUrl(sessionId: string, accessToken: string, offset?: number): string {
+  let url = `${API_BASE}/api/v1/containers/sessions/${sessionId}/stream?access_token=${encodeURIComponent(accessToken)}`
+  if (offset && offset > 0) {
+    url += `&offset=${offset}`
+  }
+  return url
 }