feat: Implement streaming message integration in conversation loader

- Added functionality to load additional messages from streaming files based on session ID.
- Merged non-duplicate streaming messages with existing conversation history.
- Updated conversation metadata, including message count and end time.
- Introduced a new utility function `getStreamingMessages` to retrieve messages from streaming files.

This enhancement improves the conversation loading process by ensuring that the latest messages are included, providing a more comprehensive chat history.

Author: xiaocang

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/streaming/streamingFileManager.ts

@@ -225,3 +225,85 @@ export async function cleanupAllStreamingFiles(
     console.error(`Failed to clean up streaming directory:`, error);
   }
 }
+
+/**
+ * Get streaming messages for a specific sessionId
+ * This searches through streaming files to find messages with matching sessionId
+ */
+export async function getStreamingMessages(
+  encodedProjectName: string,
+  sessionId: string,
+  runtime: Runtime,
+): Promise<unknown[]> {
+  const streamingDir = getStreamingDir(encodedProjectName, runtime);
+
+  // Check if streaming directory exists
+  if (!(await runtime.exists(streamingDir))) {
+    return [];
+  }
+
+  const messages: unknown[] = [];
+
+  try {
+    // List all files in the streaming directory
+    const entries: string[] = [];
+    for await (const entry of runtime.readDir(streamingDir)) {
+      if (entry.isFile && entry.name.endsWith(".jsonl")) {
+        entries.push(entry.name);
+      }
+    }
+
+    // Process each streaming file
+    for (const file of entries) {
+      const filePath = `${streamingDir}/${file}`;
+      const content = await runtime.readTextFile(filePath);
+      const lines = content
+        .trim()
+        .split("\n")
+        .filter((line) => line.trim());
+
+      // Check each message for matching sessionId
+      for (const line of lines) {
+        try {
+          const parsed = JSON.parse(line);
+
+          // Check if this is a claude_json message with SDK content
+          if (
+            parsed.type === "claude_json" &&
+            parsed.content?.type === "sdk"
+          ) {
+            const sdkMessage = parsed.content.message;
+
+            // Extract sessionId from different message types
+            let messageSessionId: string | undefined;
+
+            if (sdkMessage.type === "system" && sdkMessage.session_id) {
+              messageSessionId = sdkMessage.session_id;
+            } else if (
+              sdkMessage.type === "assistant" &&
+              sdkMessage.session_id
+            ) {
+              messageSessionId = sdkMessage.session_id;
+            } else if (
+              sdkMessage.type === "result" &&
+              sdkMessage.session_id
+            ) {
+              messageSessionId = sdkMessage.session_id;
+            }
+
+            // If sessionId matches, add the SDK message
+            if (messageSessionId === sessionId) {
+              messages.push(sdkMessage);
+            }
+          }
+        } catch (error) {
+          console.error(`Failed to parse streaming message:`, error);
+        }
+      }
+    }
+  } catch (error) {
+    console.error(`Failed to read streaming directory:`, error);
+  }
+
+  return messages;
+}