fix user message rendering

joshsny · joshsny · commit f6b12ad60fe5 · 2025-12-04T23:58:01.000Z
diff --git a/apps/array/src/api/fetcher.ts b/apps/array/src/api/fetcher.ts
@@ -10,6 +10,7 @@ export const buildApiFetcher: (config: {
   ): Promise<Response> => {
     const headers = new Headers();
     headers.set("Authorization", `Bearer ${token}`);
+    headers.set("Content-Type", "application/json");
 
     if (input.urlSearchParams) {
       input.url.search = input.urlSearchParams.toString();
@@ -21,10 +22,6 @@ export const buildApiFetcher: (config: {
       ? JSON.stringify(input.parameters?.body)
       : undefined;
 
-    if (body) {
-      headers.set("Content-Type", "application/json");
-    }
-
     if (input.parameters?.header) {
       for (const [key, value] of Object.entries(input.parameters.header)) {
         if (value != null) {
diff --git a/apps/array/src/api/posthogClient.ts b/apps/array/src/api/posthogClient.ts
@@ -246,7 +246,6 @@ export class PostHogAPIClient {
       path: url,
       overrides: {
         body: JSON.stringify({ entries }),
-        headers: { "Content-Type": "application/json" },
       },
     });
     if (!response.ok) {
diff --git a/apps/array/src/renderer/features/sessions/components/SessionView.tsx b/apps/array/src/renderer/features/sessions/components/SessionView.tsx
@@ -106,6 +106,9 @@ function parseSessionNotification(
   notification: SessionNotification,
 ): ParseResult {
   const { update } = notification;
+  if (!update?.sessionUpdate) {
+    return null;
+  }
 
   switch (update.sessionUpdate) {
     case "user_message_chunk":
diff --git a/apps/array/src/renderer/features/sessions/stores/sessionStore.ts b/apps/array/src/renderer/features/sessions/stores/sessionStore.ts
@@ -92,6 +92,7 @@ export interface AgentSession {
   isPromptPending: boolean;
   isCloud: boolean;
   logUrl?: string;
+  processedLineCount?: number;
 }
 
 interface ConnectParams {
@@ -216,6 +217,7 @@ export const useSessionStore = create<SessionStore>((set, get) => ({
               isPromptPending: false,
               isCloud: true,
               logUrl: latestRunLogUrl,
+              processedLineCount: rawEntries.length,
             },
           },
         }));
@@ -393,7 +395,59 @@ export const useSessionStore = create<SessionStore>((set, get) => ({
       throw new Error("No active session for task");
     }
 
-    // Set pending state
+    const blocks: ContentBlock[] =
+      typeof prompt === "string" ? [{ type: "text", text: prompt }] : prompt;
+
+    if (session.isCloud) {
+      // Cloud: send via S3 log - cloud runner polls and picks up
+      // No pending state needed since we just append to log
+      const authState = useAuthStore.getState();
+      const { client } = authState;
+      if (!client) {
+        throw new Error("API client not available");
+      }
+
+      const notification = {
+        type: "notification" as const,
+        timestamp: new Date().toISOString(),
+        direction: "client" as const,
+        notification: {
+          method: "session/update",
+          params: {
+            update: {
+              sessionUpdate: "user_message_chunk",
+              content: blocks[0],
+            },
+          },
+        },
+      };
+
+      await client.appendTaskRunLog(taskId, session.taskRunId, [notification]);
+      log.info("Sent cloud message via S3", { taskId, runId: session.taskRunId });
+
+      // Optimistically add user message to local state immediately
+      const ts = Date.now();
+      const userEvent: SessionEvent = {
+        type: "session_update",
+        ts,
+        notification: notification.notification.params as SessionNotification,
+      };
+      set((state) => ({
+        sessions: {
+          ...state.sessions,
+          [session.taskRunId]: {
+            ...state.sessions[session.taskRunId],
+            events: [...state.sessions[session.taskRunId].events, userEvent],
+            processedLineCount:
+              (state.sessions[session.taskRunId].processedLineCount ?? 0) + 1,
+          },
+        },
+      }));
+
+      return { stopReason: "pending" };
+    }
+
+    // Local: set pending state and send via IPC
     set((state) => ({
       sessions: {
         ...state.sessions,
@@ -405,39 +459,8 @@ export const useSessionStore = create<SessionStore>((set, get) => ({
     }));
 
     try {
-      // Convert text to ContentBlock array if needed
-      const blocks: ContentBlock[] =
-        typeof prompt === "string" ? [{ type: "text", text: prompt }] : prompt;
-
-      if (session.isCloud) {
-        // Cloud: send via S3 log - cloud runner polls and picks up
-        const authState = useAuthStore.getState();
-        const { client } = authState;
-        if (!client) {
-          throw new Error("API client not available");
-        }
-
-        const notification = {
-          type: "notification" as const,
-          timestamp: new Date().toISOString(),
-          notification: {
-            method: "sessionUpdate",
-            params: {
-              sessionUpdate: "user_message",
-              content: blocks,
-            },
-          },
-        };
-
-        await client.appendTaskRunLog(taskId, session.taskRunId, [notification]);
-        log.info("Sent cloud message via S3", { taskId, runId: session.taskRunId });
-        return { stopReason: "pending" };
-      }
-
-      // Local: send via IPC directly to running agent
       return await window.electronAPI.agentPrompt(session.taskRunId, blocks);
     } finally {
-      // Clear pending state
       set((state) => ({
         sessions: {
           ...state.sessions,
@@ -520,29 +543,54 @@ export const useSessionStore = create<SessionStore>((set, get) => ({
         const text = await response.text();
         const lines = text.trim().split("\n").filter(Boolean);
 
-        // Only process new entries
-        const currentEventCount = session.events.length;
-        if (lines.length > currentEventCount) {
-          const newLines = lines.slice(currentEventCount);
+        // Only process new entries (track by line count, not event count)
+        const processedCount = session.processedLineCount ?? 0;
+        if (lines.length > processedCount) {
+          const newLines = lines.slice(processedCount);
           for (const line of newLines) {
             try {
               const entry = JSON.parse(line);
               const ts = entry.timestamp
                 ? new Date(entry.timestamp).getTime()
                 : Date.now();
 
-              const event: SessionEvent = {
+              // Create acp_message for raw log entry
+              const acpEvent: SessionEvent = {
                 type: "acp_message",
                 direction: entry.direction ?? "agent",
                 ts,
                 message: entry.notification,
               };
-
-              get()._handleEvent(taskRunId, event);
+              get()._handleEvent(taskRunId, acpEvent);
+
+              // Also create session_update event for session/update notifications
+              if (
+                entry.type === "notification" &&
+                entry.notification?.method === "session/update" &&
+                entry.notification?.params
+              ) {
+                const sessionUpdateEvent: SessionEvent = {
+                  type: "session_update",
+                  ts,
+                  notification: entry.notification.params as SessionNotification,
+                };
+                get()._handleEvent(taskRunId, sessionUpdateEvent);
+              }
             } catch {
               // Skip invalid JSON
             }
           }
+
+          // Update processed line count
+          set((state) => ({
+            sessions: {
+              ...state.sessions,
+              [taskRunId]: {
+                ...state.sessions[taskRunId],
+                processedLineCount: lines.length,
+              },
+            },
+          }));
         }
       } catch (err) {
         log.warn("Cloud polling error", { error: err });
diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts
@@ -560,7 +560,10 @@ Generated by PostHog Agent`;
 
     // Start session in SessionStore (updates task run status to in_progress)
     const taskRun = await this.sessionStore.start(taskRunId, taskId, taskRunId);
-    this.logger.debug("Session started", { taskRunId, logUrl: taskRun?.log_url });
+    this.logger.debug("Session started", {
+      taskRunId,
+      logUrl: taskRun?.log_url,
+    });
 
     // Create internal ACP connection with S3 persistence
     const acpConnection = createAcpConnection({
@@ -650,7 +653,8 @@ Generated by PostHog Agent`;
               // Look for user_message notifications
               if (
                 entry.notification?.method === "sessionUpdate" &&
-                (entry.notification?.params as any)?.sessionUpdate === "user_message"
+                (entry.notification?.params as any)?.sessionUpdate ===
+                  "user_message"
               ) {
                 const content = (entry.notification?.params as any)?.content;
                 if (content) {
@@ -690,7 +694,8 @@ Generated by PostHog Agent`;
 
       await this.sessionStore.complete(taskRunId);
     } catch (error) {
-      const errorMessage = error instanceof Error ? error.message : String(error);
+      const errorMessage =
+        error instanceof Error ? error.message : String(error);
       this.logger.error("Cloud task execution failed", {
         taskId: task.id,
         error: errorMessage,
