Draft: feature: Add local file intermediary for network-resilient streaming

.gitignore

@@ -8,4 +8,6 @@ backend/dist/
 frontend/demo-recordings/
 frontend/test-results/
 frontend/playwright-report/
-frontend/playwright/.cache/
+frontend/playwright/.cache/
+
+backend/.streaming/

backend/app.ts

@@ -17,6 +17,9 @@ import { handleHistoriesRequest } from "./handlers/histories.ts";
 import { handleConversationRequest } from "./handlers/conversations.ts";
 import { handleChatRequest } from "./handlers/chat.ts";
 import { handleAbortRequest } from "./handlers/abort.ts";
+import { handleResumeRequest } from "./handlers/resume.ts";
+import { handleStatusRequest } from "./handlers/status.ts";
+import { startCleanupInterval } from "./streaming/streamingFileManager.ts";
 
 export interface AppConfig {
   debugMode: boolean;
@@ -33,6 +36,9 @@ export function createApp(
   // Store AbortControllers for each request (shared with chat handler)
   const requestAbortControllers = new Map<string, AbortController>();
 
+  // Start cleanup interval for streaming files
+  startCleanupInterval(runtime);
+
   // CORS middleware
   app.use(
     "*",
@@ -71,10 +77,11 @@ export function createApp(
     (c) => handleAbortRequest(c, requestAbortControllers),
   );
 
-  app.post(
-    "/api/chat",
-    (c) => handleChatRequest(c, requestAbortControllers),
-  );
+  app.post("/api/chat", (c) => handleChatRequest(c, requestAbortControllers));
+
+  app.get("/api/resume/:requestId", (c) => handleResumeRequest(c));
+
+  app.get("/api/status/:requestId", (c) => handleStatusRequest(c));
 
   // Static file serving with SPA fallback
   // Serve static assets (CSS, JS, images, etc.)

backend/deno.json

@@ -11,8 +11,8 @@
   "tasks": {
     "generate-version": "node scripts/generate-version.js",
     "copy-frontend": "node scripts/copy-frontend.js",
-    "dev": "deno task generate-version && dotenvx run --env-file=../.env -- deno run --allow-net --allow-run --allow-read --allow-env --watch cli/deno.ts --debug",
-    "build": "deno task generate-version && deno task copy-frontend && deno compile --allow-net --allow-run --allow-read --allow-env --include ./dist/static --output ../dist/claude-code-webui cli/deno.ts",
+    "dev": "deno task generate-version && dotenvx run --env-file=../.env -- deno run --allow-net --allow-run --allow-read --allow-write=./.streaming --allow-env --watch cli/deno.ts --debug",
+    "build": "deno task generate-version && deno task copy-frontend && deno compile --allow-net --allow-run --allow-read --allow-write=./.streaming --allow-env --include ./dist/static --output ../dist/claude-code-webui cli/deno.ts",
     "format": "deno fmt",
     "lint": "deno lint",
     "check": "deno check",

backend/handlers/chat.ts

@@ -2,6 +2,10 @@ import { Context } from "hono";
 import { AbortError, query } from "@anthropic-ai/claude-code";
 import type { ChatRequest, StreamResponse } from "../../shared/types.ts";
 import type { Runtime } from "../runtime/types.ts";
+import {
+  appendMessage,
+  initializeStreaming,
+} from "../streaming/streamingFileManager.ts";
 
 /**
  * Automatically determines Claude Code execution configuration
@@ -48,6 +52,16 @@ function getClaudeExecutionConfig(claudePath: string, runtime: Runtime) {
   return createNodeConfig(actualPath);
 }
 
+/**
+ * Get encoded project name from working directory
+ */
+function getEncodedProjectName(workingDirectory?: string): string | null {
+  if (!workingDirectory) return null;
+
+  // Encode the directory path to match Claude's project naming convention
+  return encodeURIComponent(workingDirectory.replace(/\//g, "_"));
+}
+
 /**
  * Executes a Claude command and yields streaming responses
  * @param message - User message or command
@@ -73,6 +87,7 @@ async function* executeClaudeCommand(
   debugMode?: boolean,
 ): AsyncGenerator<StreamResponse> {
   let abortController: AbortController;
+  const encodedProjectName = getEncodedProjectName(workingDirectory);
 
   try {
     // Process commands that start with '/'
@@ -86,6 +101,11 @@ async function* executeClaudeCommand(
     abortController = new AbortController();
     requestAbortControllers.set(requestId, abortController);
 
+    // Initialize streaming file if we have a project
+    if (encodedProjectName) {
+      await initializeStreaming(encodedProjectName, requestId, runtime);
+    }
+
     // Use the validated Claude path from startup configuration (passed as parameter)
 
     // Get Claude Code execution configuration for migrate-installer compatibility
@@ -110,25 +130,64 @@ async function* executeClaudeCommand(
         console.debug("---");
       }
 
-      yield {
+      const response: StreamResponse = {
         type: "claude_json",
         data: sdkMessage,
       };
+
+      // Write to streaming file if we have a project
+      if (encodedProjectName) {
+        await appendMessage(encodedProjectName, requestId, response, runtime);
+      }
+
+      yield response;
+    }
+
+    const doneResponse: StreamResponse = { type: "done" };
+
+    // Write done message to streaming file
+    if (encodedProjectName) {
+      await appendMessage(encodedProjectName, requestId, doneResponse, runtime);
     }
 
-    yield { type: "done" };
+    yield doneResponse;
   } catch (error) {
     // Check if error is due to abort
     if (error instanceof AbortError) {
-      yield { type: "aborted" };
+      const abortedResponse: StreamResponse = { type: "aborted" };
+
+      // Write aborted message to streaming file
+      if (encodedProjectName) {
+        await appendMessage(
+          encodedProjectName,
+          requestId,
+          abortedResponse,
+          runtime,
+        );
+      }
+
+      yield abortedResponse;
     } else {
       if (debugMode) {
         console.error("Claude Code execution failed:", error);
       }
-      yield {
+
+      const errorResponse: StreamResponse = {
         type: "error",
         error: error instanceof Error ? error.message : String(error),
       };
+
+      // Write error message to streaming file
+      if (encodedProjectName) {
+        await appendMessage(
+          encodedProjectName,
+          requestId,
+          errorResponse,
+          runtime,
+        );
+      }
+
+      yield errorResponse;
     }
   } finally {
     // Clean up AbortController from map

backend/handlers/resume.ts

@@ -0,0 +1,62 @@
+/**
+ * Resume API handler
+ * Handles requests to resume streaming from a specific point after network interruption
+ */
+
+import { Context } from "hono";
+import type { ResumeResponse } from "../../shared/types.ts";
+import {
+  getRequestStatus,
+  readStreamingFile,
+} from "../streaming/streamingFileManager.ts";
+
+/**
+ * Handles GET /api/resume/:requestId requests
+ * Returns messages from a specific index for resuming interrupted streams
+ */
+export async function handleResumeRequest(c: Context): Promise<Response> {
+  const requestId = c.req.param("requestId");
+  const fromIndex = parseInt(c.req.query("fromIndex") || "0", 10);
+  const { runtime } = c.var.config;
+
+  // Validate requestId
+  if (!requestId) {
+    return c.json({ error: "Request ID is required" }, 400);
+  }
+
+  // Get request status
+  const status = getRequestStatus(requestId);
+  if (!status) {
+    return c.json({ error: "Request not found" }, 404);
+  }
+
+  // Extract encoded project name from the file path
+  // Format: /home/user/.claude/projects/{encodedProjectName}/streaming/{requestId}.jsonl
+  const pathParts = status.filePath.split("/");
+  const projectsIndex = pathParts.indexOf("projects");
+  if (projectsIndex === -1 || projectsIndex + 2 >= pathParts.length) {
+    return c.json({ error: "Invalid file path structure" }, 500);
+  }
+  const encodedProjectName = pathParts[projectsIndex + 1];
+
+  try {
+    // Read messages from the streaming file
+    const messages = await readStreamingFile(
+      encodedProjectName,
+      requestId,
+      fromIndex,
+      runtime,
+    );
+
+    const response: ResumeResponse = {
+      messages,
+      totalMessages: status.totalMessages,
+      isComplete: status.status !== "in_progress",
+    };
+
+    return c.json(response);
+  } catch (error) {
+    console.error("Failed to read streaming file:", error);
+    return c.json({ error: "Failed to read streaming data" }, 500);
+  }
+}

backend/handlers/status.ts

@@ -0,0 +1,44 @@
+/**
+ * Status API handler
+ * Handles requests to check the status of a streaming request
+ */
+
+import { Context } from "hono";
+import type { StatusResponse } from "../../shared/types.ts";
+import { RequestStatus } from "../../shared/types.ts";
+import { getRequestStatus } from "../streaming/streamingFileManager.ts";
+
+/**
+ * Handles GET /api/status/:requestId requests
+ * Returns the current status of a streaming request
+ */
+export function handleStatusRequest(c: Context): Response {
+  const requestId = c.req.param("requestId");
+
+  // Validate requestId
+  if (!requestId) {
+    return c.json({ error: "Request ID is required" }, 400);
+  }
+
+  // Get request status
+  const status = getRequestStatus(requestId);
+
+  if (!status) {
+    const response: StatusResponse = {
+      requestId,
+      status: RequestStatus.NOT_FOUND,
+      totalMessages: 0,
+      lastUpdated: new Date().toISOString(),
+    };
+    return c.json(response);
+  }
+
+  const response: StatusResponse = {
+    requestId,
+    status: status.status,
+    totalMessages: status.totalMessages,
+    lastUpdated: status.lastUpdated.toISOString(),
+  };
+
+  return c.json(response);
+}

backend/history/conversationLoader.ts

@@ -8,6 +8,7 @@ import type { ConversationHistory } from "../../shared/types.ts";
 import type { Runtime } from "../runtime/types.ts";
 import { processConversationMessages } from "./timestampRestore.ts";
 import { validateEncodedProjectName } from "./pathUtils.ts";
+import { getStreamingMessages } from "../streaming/streamingFileManager.ts";
 
 /**
  * Load a specific conversation by session ID
@@ -47,6 +48,66 @@ export async function loadConversation(
       sessionId,
       runtime,
     );
+
+    // Check for additional messages in streaming files
+    const streamingMessages = await getStreamingMessages(
+      encodedProjectName,
+      sessionId,
+      runtime,
+    );
+
+    if (streamingMessages.length > 0) {
+      // Merge streaming messages with conversation history
+      // The streaming messages might contain duplicates or newer messages
+      const existingMessageIds = new Set<string>();
+
+      // Build a set of existing message IDs/timestamps for deduplication
+      for (const msg of conversationHistory.messages) {
+        if (typeof msg === "object" && msg !== null && "timestamp" in msg) {
+          existingMessageIds.add(JSON.stringify(msg));
+        }
+      }
+
+      // Add non-duplicate streaming messages
+      let newMessagesAdded = 0;
+      for (const streamingMsg of streamingMessages) {
+        const msgKey = JSON.stringify(streamingMsg);
+        if (!existingMessageIds.has(msgKey)) {
+          conversationHistory.messages.push(streamingMsg);
+          newMessagesAdded++;
+        }
+      }
+
+      if (newMessagesAdded > 0) {
+        // Re-sort messages by timestamp if new messages were added
+        conversationHistory.messages.sort(
+          (a: unknown, b: unknown) => {
+            const timeA = (a as { timestamp?: number }).timestamp || 0;
+            const timeB = (b as { timestamp?: number }).timestamp || 0;
+            return timeA - timeB;
+          },
+        );
+
+        // Update metadata
+        conversationHistory.metadata.messageCount =
+          conversationHistory.messages.length;
+        if (conversationHistory.messages.length > 0) {
+          const lastMsg = conversationHistory.messages[
+            conversationHistory.messages.length - 1
+          ] as { timestamp?: number };
+          if (lastMsg.timestamp) {
+            conversationHistory.metadata.endTime = new Date(
+              lastMsg.timestamp,
+            ).toISOString();
+          }
+        }
+
+        console.log(
+          `[ConversationLoader] Merged ${newMessagesAdded} additional messages from streaming files for session ${sessionId}`,
+        );
+      }
+    }
+
     return conversationHistory;
   } catch (error) {
     throw error; // Re-throw any parsing errors
@@ -63,7 +124,10 @@ async function parseConversationFile(
   runtime: Runtime,
 ): Promise<ConversationHistory> {
   const content = await runtime.readTextFile(filePath);
-  const lines = content.trim().split("\n").filter((line) => line.trim());
+  const lines = content
+    .trim()
+    .split("\n")
+    .filter((line) => line.trim());
 
   if (lines.length === 0) {
     throw new Error("Empty conversation file");

backend/pathUtils.test.ts

@@ -5,7 +5,7 @@ import type { MiddlewareHandler } from "hono";
 
 // Create a mock runtime for testing
 const mockRuntime: Runtime = {
-  getEnv: (key: string) => key === "HOME" ? "/mock/home" : undefined,
+  getEnv: (key: string) => (key === "HOME" ? "/mock/home" : undefined),
   async *readDir(_path: string) {
     // Mock empty directory - no entries
     // This async generator yields nothing, representing an empty directory
@@ -47,6 +47,10 @@ const mockRuntime: Runtime = {
   serve: () => {},
   createStaticFileMiddleware: (): MiddlewareHandler => () =>
     Promise.resolve(new Response()),
+  appendTextFile: () => Promise.resolve(),
+  ensureDir: () => Promise.resolve(),
+  remove: () => Promise.resolve(),
+  removeDir: () => Promise.resolve(),
 };
 
 describe("pathUtils", () => {