feat: move to read from s3 (#129)

k11kirky · web-flow · commit 2ec3c9e8c3b6 · 2025-11-19T12:55:43.000-08:00
diff --git a/src/main/preload.ts b/src/main/preload.ts
@@ -35,6 +35,8 @@ contextBridge.exposeInMainWorld("electronAPI", {
     ipcRenderer.invoke("store-api-key", apiKey),
   retrieveApiKey: (encryptedKey: string): Promise<string | null> =>
     ipcRenderer.invoke("retrieve-api-key", encryptedKey),
+  fetchS3Logs: (logUrl: string): Promise<string> =>
+    ipcRenderer.invoke("fetch-s3-logs", logUrl),
   // OAuth API
   oauthStartFlow: (
     region: CloudRegion,
diff --git a/src/main/services/agent.ts b/src/main/services/agent.ts
@@ -118,13 +118,6 @@ export function registerAgentIpc(
         posthogApiKey: apiKey,
         posthogApiUrl: apiHost,
         posthogProjectId: projectId,
-        onEvent: (event) => {
-          console.log("agent event", event);
-          if (!event || abortController.signal.aborted) return;
-          const payload =
-            event.type === "done" ? { ...event, success: true } : event;
-          emitToRenderer(payload);
-        },
         debug: true,
       });
 
diff --git a/src/main/services/posthog.ts b/src/main/services/posthog.ts
@@ -30,4 +30,28 @@ export function registerPosthogIpc(): void {
       return encryptedKey;
     },
   );
+
+  // Fetch S3 logs
+  ipcMain.handle(
+    "fetch-s3-logs",
+    async (_event: IpcMainInvokeEvent, logUrl: string): Promise<string> => {
+      try {
+        console.log("Fetching S3 logs from:", logUrl);
+        const response = await fetch(logUrl);
+
+        if (!response.ok) {
+          throw new Error(
+            `Failed to fetch logs: ${response.status} ${response.statusText}`,
+          );
+        }
+
+        const content = await response.text();
+        console.log("S3 logs fetched:", content);
+        return content;
+      } catch (error) {
+        console.error("Failed to fetch S3 logs:", error);
+        throw error;
+      }
+    },
+  );
 }
diff --git a/src/renderer/features/editor/components/PlanEditor.tsx b/src/renderer/features/editor/components/PlanEditor.tsx
@@ -85,11 +85,7 @@ export function PlanEditor({
 
   // Track unsaved changes
   useEffect(() => {
-    if (content !== fetchedContent) {
-      setHasUnsavedChanges(true);
-    } else {
-      setHasUnsavedChanges(false);
-    }
+    setHasUnsavedChanges(content !== fetchedContent);
   }, [content, fetchedContent]);
 
   return (
diff --git a/src/renderer/features/task-detail/stores/taskExecutionStore.ts b/src/renderer/features/task-detail/stores/taskExecutionStore.ts
@@ -6,6 +6,7 @@ import { track } from "@renderer/lib/analytics";
 import type {
   ClarifyingQuestion,
   ExecutionMode,
+  LogEntry,
   PlanModePhase,
   QuestionAnswer,
   Task,
@@ -34,6 +35,23 @@ interface ArtifactEvent {
   }>;
 }
 
+interface TodoItem {
+  content: string;
+  status: "pending" | "in_progress" | "completed";
+  activeForm: string;
+}
+
+interface TodoList {
+  items: TodoItem[];
+  metadata: {
+    total: number;
+    pending: number;
+    in_progress: number;
+    completed: number;
+    last_updated: string;
+  };
+}
+
 const createProgressSignature = (progress: TaskRun): string =>
   [progress.status ?? "", progress.updated_at ?? ""].join("|");
 
@@ -59,22 +77,79 @@ const toClarifyingQuestions = (
     requiresInput: hasCustomOption(q.options),
   }));
 };
-
-export interface TodoItem {
-  content: string;
-  status: "pending" | "in_progress" | "completed";
-  activeForm: string;
+/**
+ * Convert S3 LogEntry to AgentEvent
+ */
+function logEntryToAgentEvent(entry: LogEntry): AgentEvent | null {
+  try {
+    const baseTs = Date.now();
+
+    if (entry.type === "token") {
+      return {
+        type: "token",
+        ts: baseTs,
+        content: entry.message,
+      } as AgentEvent;
+    }
+
+    if (entry.type === "info") {
+      return {
+        type: "token",
+        ts: baseTs,
+        content: entry.message,
+      } as AgentEvent;
+    }
+
+    if (entry.message) {
+      try {
+        const parsed = JSON.parse(entry.message);
+        return {
+          ...parsed,
+          type: entry.type as any,
+          ts: parsed.ts || baseTs,
+        } as AgentEvent;
+      } catch {
+        return {
+          type: entry.type as any,
+          ts: baseTs,
+          message: entry.message,
+        } as AgentEvent;
+      }
+    }
+
+    return null;
+  } catch (err) {
+    console.warn("Failed to convert log entry to agent event", err, entry);
+    return null;
+  }
 }
 
-export interface TodoList {
-  items: TodoItem[];
-  metadata: {
-    total: number;
-    pending: number;
-    in_progress: number;
-    completed: number;
-    last_updated: string;
-  };
+/**
+ * Fetch logs from S3 log URL via main process to avoid CORS issues
+ * Always fetches and returns the entire log file
+ */
+async function fetchLogsFromS3Url(logUrl: string): Promise<AgentEvent[]> {
+  try {
+    const content = await window.electronAPI?.fetchS3Logs(logUrl);
+
+    if (!content || !content.trim()) {
+      return [];
+    }
+
+    const logEntries = content
+      .trim()
+      .split("\n")
+      .map((line: string) => JSON.parse(line) as LogEntry);
+
+    const events = logEntries
+      .map((entry: LogEntry) => logEntryToAgentEvent(entry))
+      .filter((event): event is AgentEvent => event !== null);
+
+    return events;
+  } catch (err) {
+    console.warn("Failed to fetch task logs from S3", err);
+    return [];
+  }
 }
 
 interface TaskExecutionState {
@@ -87,6 +162,8 @@ interface TaskExecutionState {
   unsubscribe: (() => void) | null;
   progress: TaskRun | null;
   progressSignature: string | null;
+  // S3 log polling fields
+  logPoller: ReturnType<typeof setInterval> | null;
   // Plan mode fields
   executionMode: ExecutionMode;
   planModePhase: PlanModePhase;
@@ -162,6 +239,7 @@ const defaultTaskState: TaskExecutionState = {
   unsubscribe: null,
   progress: null,
   progressSignature: null,
+  logPoller: null,
   executionMode: "plan",
   planModePhase: "idle",
   clarifyingQuestions: [],
@@ -258,6 +336,9 @@ export const useTaskExecutionStore = create<TaskExecutionStore>()(
         if (taskState?.unsubscribe) {
           taskState.unsubscribe();
         }
+        if (taskState?.logPoller) {
+          clearInterval(taskState.logPoller);
+        }
         set((state) => {
           const newTaskStates = { ...state.taskStates };
           delete newTaskStates[taskId];
@@ -268,19 +349,22 @@ export const useTaskExecutionStore = create<TaskExecutionStore>()(
       subscribeToAgentEvents: (taskId: string, channel: string) => {
         const store = get();
 
-        // Clean up existing subscription if any
         const existingState = store.taskStates[taskId];
         if (existingState?.unsubscribe) {
           existingState.unsubscribe();
         }
+        if (existingState?.logPoller) {
+          clearInterval(existingState.logPoller);
+        }
+
+        store.updateTaskState(taskId, { logPoller: null });
 
-        // Create new subscription that persists even when component unmounts
+        // Create new subscription that listens only to progress events
         const unsubscribeFn = window.electronAPI?.onAgentEvent(
           channel,
           (ev: AgentEvent | { type: "progress"; progress: TaskRun }) => {
             const currentStore = get();
 
-            // Handle custom progress events from Array backend
             if (ev?.type === "progress" && "progress" in ev) {
               const newProgress = ev.progress;
               const oldProgress = currentStore.getTaskState(taskId).progress;
@@ -289,34 +373,111 @@ export const useTaskExecutionStore = create<TaskExecutionStore>()(
                 : null;
               const newSig = createProgressSignature(newProgress);
 
-              // Always update the stored progress state for UI (like TaskDetail)
               currentStore.setProgress(taskId, newProgress);
 
-              // Only add to logs if signature changed (to avoid duplicate log entries)
               if (oldSig !== newSig) {
                 currentStore.addLog(taskId, {
                   type: "progress",
                   ts: Date.now(),
                   progress: newProgress,
                 } as unknown as AgentEvent);
               }
-              return;
-            }
 
-            // Store AgentEvent directly (ev is now narrowed to AgentEvent)
-            if (ev?.type) {
-              // Handle state changes for special event types
-              if (ev.type === "error" || ev.type === "done") {
-                currentStore.setRunning(taskId, false);
-                if (ev.type === "done") {
-                  currentStore.setUnsubscribe(taskId, null);
+              if (newProgress.log_url) {
+                const currentState = currentStore.getTaskState(taskId);
+
+                // Don't start polling if task is already complete
+                if (
+                  newProgress.status === "completed" ||
+                  newProgress.status === "failed"
+                ) {
+                  // Stop any existing poller
+                  if (currentState.logPoller) {
+                    clearInterval(currentState.logPoller);
+                    currentStore.updateTaskState(taskId, { logPoller: null });
+                  }
+
+                  // Do one final fetch to get all logs
+                  if (newProgress.log_url) {
+                    fetchLogsFromS3Url(newProgress.log_url)
+                      .then((allEvents) => {
+                        if (allEvents.length > 0) {
+                          const store = get();
+                          const hasDone = allEvents.some(
+                            (event) => event.type === "done",
+                          );
+                          if (hasDone) {
+                            store.setRunning(taskId, false);
+                            store.checkPlanCompletion(taskId);
+                          }
+                          store.setLogs(taskId, allEvents);
+                        }
+                      })
+                      .catch((err) =>
+                        console.warn("Failed to fetch final logs", err),
+                      );
+                  }
+                  return;
+                }
+
+                if (!currentState.logPoller) {
+                  const pollLogs = async () => {
+                    const state = get().getTaskState(taskId);
+                    const progress = state.progress;
+
+                    if (
+                      !progress?.log_url ||
+                      progress.status === "completed" ||
+                      progress.status === "failed"
+                    ) {
+                      if (state.logPoller) {
+                        clearInterval(state.logPoller);
+                        get().updateTaskState(taskId, { logPoller: null });
+                      }
+                      return;
+                    }
+
+                    const allEvents = await fetchLogsFromS3Url(
+                      progress.log_url,
+                    );
+
+                    if (allEvents.length > 0) {
+                      const store = get();
+
+                      const hasError = allEvents.some(
+                        (event) => event.type === "error",
+                      );
+                      const hasDone = allEvents.some(
+                        (event) => event.type === "done",
+                      );
+
+                      if (hasError || hasDone) {
+                        store.setRunning(taskId, false);
+                        if (hasDone) {
+                          const currentState = store.getTaskState(taskId);
+                          if (currentState.logPoller) {
+                            clearInterval(currentState.logPoller);
+                            store.updateTaskState(taskId, { logPoller: null });
+                          }
+                        }
+                        store.checkPlanCompletion(taskId);
+                      }
+
+                      store.setLogs(taskId, allEvents);
+                    }
+                  };
+
+                  void pollLogs();
+
+                  const poller = setInterval(() => {
+                    void pollLogs();
+                  }, 2000);
+
+                  currentStore.updateTaskState(taskId, { logPoller: poller });
                 }
-                // Check if plan needs to be loaded after run completes
-                currentStore.checkPlanCompletion(taskId);
               }
 
-              // Add event to logs
-              currentStore.addLog(taskId, ev);
+              return;
             }
           },
         );
@@ -332,6 +493,10 @@ export const useTaskExecutionStore = create<TaskExecutionStore>()(
           taskState.unsubscribe();
           get().setUnsubscribe(taskId, null);
         }
+        if (taskState?.logPoller) {
+          clearInterval(taskState.logPoller);
+          get().updateTaskState(taskId, { logPoller: null });
+        }
       },
 
       // High-level task execution actions
@@ -855,12 +1020,12 @@ export const useTaskExecutionStore = create<TaskExecutionStore>()(
     }),
     {
       name: "task-execution-storage",
-      // Don't persist unsubscribe functions as they can't be serialized
+      // Don't persist unsubscribe functions and pollers as they can't be serialized
       partialize: (state) => ({
         taskStates: Object.fromEntries(
           Object.entries(state.taskStates).map(([taskId, taskState]) => [
             taskId,
-            { ...taskState, unsubscribe: null },
+            { ...taskState, unsubscribe: null, logPoller: null },
           ]),
         ),
       }),
diff --git a/src/renderer/types/electron.d.ts b/src/renderer/types/electron.d.ts
