Improve performance of conversation (#692)

* Stream endpoint for execution processes (vibe-kanban c5144da6)

I want an endpoint that's similar to task stream in crates/server/src/routes/tasks.rs but contains execution processes.

The structure of the document should be:

```json
{
    "execution_processes": {
        [EXECUTION_PROCESS_ID]: {
            ... execution process fields
        }
    }
}
```

The endpoint should be at `/api/execution_processes/stream?task_attempt_id=...`

crates/server/src/routes/execution_processes.rs

* add virtualizedlist component

* WIP remove execution processes

* rebase syntax fix

* tmp fix lint

* lint

* VirtuosoMessageList

* cache

* event based hook

* historic

* handle failed historic

* running processes

* user message

* loading

* cleanup

* render user message

* style

* fmt

* better indication for setup/cleanup scripts

* fix ref issue

* virtuoso license

* fmt

* update loader

* loading

* fmt

* loading improvements

* copy as markdown styles

* spacing improvement

* flush all historic at once

* padding fix

* markdown copy sticky

* make user message editable

* edit message

* reset

* cleanup

* hook order

* remove dead code

frontend/package.json

@@ -39,12 +39,14 @@
     "@tanstack/react-query-devtools": "^5.85.5",
     "@types/react-window": "^1.8.8",
     "@uiw/react-codemirror": "^4.25.1",
+    "@virtuoso.dev/message-list": "^1.13.3",
     "class-variance-authority": "^0.7.0",
     "click-to-react-component": "^1.1.2",
     "clsx": "^2.0.0",
     "diff": "^8.0.2",
     "embla-carousel-react": "^8.6.0",
     "fancy-ansi": "^0.1.3",
+    "idb": "^8.0.3",
     "lucide-react": "^0.539.0",
     "react": "^18.2.0",
     "react-diff-viewer-continued": "^3.4.0",
@@ -52,7 +54,7 @@
     "react-markdown": "^10.1.0",
     "react-router-dom": "^6.8.1",
     "react-use-measure": "^2.1.7",
-    "react-virtuoso": "^4.13.0",
+    "react-virtuoso": "^4.14.0",
     "react-window": "^1.8.11",
     "rfc6902": "^5.1.2",
     "tailwind-merge": "^2.2.0",
@@ -82,4 +84,4 @@
     "typescript": "^5.9.2",
     "vite": "^5.0.8"
   }
-}
+}

frontend/src/components/NormalizedConversation/DisplayConversationEntry.tsx

@@ -2,6 +2,7 @@ import MarkdownRenderer from '@/components/ui/markdown-renderer.tsx';
 import {
   ActionType,
   NormalizedEntry,
+  TaskAttempt,
   type NormalizedEntryType,
 } from 'shared/types.ts';
 import type { ProcessStartPayload } from '@/types/logs';
@@ -25,11 +26,14 @@ import {
   User,
 } from 'lucide-react';
 import RawLogText from '../common/RawLogText';
+import UserMessage from './UserMessage';
 
 type Props = {
   entry: NormalizedEntry | ProcessStartPayload;
   expansionKey: string;
   diffDeletable?: boolean;
+  executionProcessId?: string;
+  taskAttempt?: TaskAttempt;
 };
 
 type FileEditAction = Extract<ActionType, { action: 'file_edit' }>;
@@ -89,36 +93,51 @@ const getEntryIcon = (entryType: NormalizedEntryType) => {
   return <Settings className={iconSize} />;
 };
 
+type ExitStatusVisualisation = 'success' | 'error' | 'pending';
+
 const getStatusIndicator = (entryType: NormalizedEntryType) => {
-  const result =
+  let status_visualisation: ExitStatusVisualisation | null = null;
+  if (
     entryType.type === 'tool_use' &&
     entryType.action_type.action === 'command_run'
-      ? entryType.action_type.result?.exit_status
-      : null;
+  ) {
+    status_visualisation = 'pending';
+    if (entryType.action_type.result?.exit_status?.type === 'success') {
+      if (entryType.action_type.result?.exit_status?.success) {
+        status_visualisation = 'success';
+      } else {
+        status_visualisation = 'error';
+      }
+    } else if (
+      entryType.action_type.result?.exit_status?.type === 'exit_code'
+    ) {
+      if (entryType.action_type.result?.exit_status?.code === 0) {
+        status_visualisation = 'success';
+      } else {
+        status_visualisation = 'error';
+      }
+    }
+  }
 
-  const status =
-    result?.type === 'success'
-      ? result.success
-        ? 'success'
-        : 'error'
-      : result?.type === 'exit_code'
-        ? result.code === 0
-          ? 'success'
-          : 'error'
-        : 'unknown';
-
-  if (status === 'unknown') return null;
-
-  const colorMap: Record<typeof status, string> = {
+  // If pending, should be a pulsing primary-foreground
+  const colorMap: Record<ExitStatusVisualisation, string> = {
     success: 'bg-green-300',
     error: 'bg-red-300',
+    pending: 'bg-primary-foreground/50',
   };
 
+  if (!status_visualisation) return null;
+
   return (
     <div className="relative">
       <div
-        className={`${colorMap[status]} h-1.5 w-1.5 rounded-full absolute -left-1 -bottom-4`}
+        className={`${colorMap[status_visualisation]} h-1.5 w-1.5 rounded-full absolute -left-1 -bottom-4`}
       />
+      {status_visualisation === 'pending' && (
+        <div
+          className={`${colorMap[status_visualisation]} h-1.5 w-1.5 rounded-full absolute -left-1 -bottom-4 animate-ping`}
+        />
+      )}
     </div>
   );
 };
@@ -463,11 +482,27 @@ const ToolCallCard: React.FC<{
   );
 };
 
+const LoadingCard = () => {
+  return (
+    <div className="flex animate-pulse space-x-2 items-center">
+      <div className="size-3 bg-foreground/10"></div>
+      <div className="flex-1 h-3 bg-foreground/10"></div>
+      <div className="flex-1 h-3"></div>
+      <div className="flex-1 h-3"></div>
+    </div>
+  );
+};
+
 /*******************
  * Main component  *
  *******************/
 
-function DisplayConversationEntry({ entry, expansionKey }: Props) {
+function DisplayConversationEntry({
+  entry,
+  expansionKey,
+  executionProcessId,
+  taskAttempt,
+}: Props) {
   const isNormalizedEntry = (
     entry: NormalizedEntry | ProcessStartPayload
   ): entry is NormalizedEntry => 'entry_type' in entry;
@@ -492,10 +527,23 @@ function DisplayConversationEntry({ entry, expansionKey }: Props) {
   const isSystem = entryType.type === 'system_message';
   const isError = entryType.type === 'error_message';
   const isToolUse = entryType.type === 'tool_use';
+  const isUserMessage = entryType.type === 'user_message';
+  const isLoading = entryType.type === 'loading';
   const isFileEdit = (a: ActionType): a is FileEditAction =>
     a.action === 'file_edit';
+
+  if (isUserMessage) {
+    return (
+      <UserMessage
+        content={entry.content}
+        executionProcessId={executionProcessId}
+        taskAttempt={taskAttempt}
+      />
+    );
+  }
+
   return (
-    <>
+    <div className="px-4 py-2 text-sm">
       {isSystem || isError ? (
         <CollapsibleEntry
           content={isNormalizedEntry(entry) ? entry.content : ''}
@@ -528,6 +576,8 @@ function DisplayConversationEntry({ entry, expansionKey }: Props) {
           expansionKey={expansionKey}
           entryContent={isNormalizedEntry(entry) ? entry.content : ''}
         />
+      ) : isLoading ? (
+        <LoadingCard />
       ) : (
         <div className={getContentClassName(entryType)}>
           {shouldRenderMarkdown(entryType) ? (
@@ -543,7 +593,7 @@ function DisplayConversationEntry({ entry, expansionKey }: Props) {
           )}
         </div>
       )}
-    </>
+    </div>
   );
 }
 

frontend/src/components/NormalizedConversation/ToolDetails.tsx

@@ -20,7 +20,7 @@ type Props = {
 };
 
 export const renderJson = (v: JsonValue) => (
-  <pre>{JSON.stringify(v, null, 2)}</pre>
+  <pre className="whitespace-pre-wrap">{JSON.stringify(v, null, 2)}</pre>
 );
 
 export default function ToolDetails({

frontend/src/components/NormalizedConversation/UserMessage.tsx

@@ -0,0 +1,70 @@
+import MarkdownRenderer from '@/components/ui/markdown-renderer';
+import { Button } from '@/components/ui/button';
+import { Pencil, Send, X } from 'lucide-react';
+import { useState } from 'react';
+import { Textarea } from '@/components/ui/textarea';
+import { useProcessRetry } from '@/hooks/useProcessRetry';
+import { TaskAttempt } from 'shared/types';
+
+const UserMessage = ({
+  content,
+  executionProcessId,
+  taskAttempt,
+}: {
+  content: string;
+  executionProcessId?: string;
+  taskAttempt?: TaskAttempt;
+}) => {
+  const [isEditing, setIsEditing] = useState(false);
+  const [editContent, setEditContent] = useState(content);
+  const retryHook = useProcessRetry(taskAttempt);
+
+  const handleEdit = () => {
+    if (!executionProcessId) return;
+    retryHook?.retryProcess(executionProcessId, editContent).then(() => {
+      setIsEditing(false);
+    });
+  };
+
+  return (
+    <div className="py-2">
+      <div className="bg-background px-4 py-2 text-sm border-y border-dashed flex gap-2">
+        <div className="flex-1">
+          {isEditing ? (
+            <Textarea
+              value={editContent}
+              onChange={(e) => setEditContent(e.target.value)}
+            />
+          ) : (
+            <MarkdownRenderer
+              content={content}
+              className="whitespace-pre-wrap break-words flex flex-col gap-1 font-light py-3"
+            />
+          )}
+        </div>
+        {executionProcessId && (
+          <div className="flex flex-col">
+            <Button
+              onClick={() => setIsEditing(!isEditing)}
+              variant="ghost"
+              className="p-2"
+            >
+              {isEditing ? (
+                <X className="w-3 h-3" />
+              ) : (
+                <Pencil className="w-3 h-3" />
+              )}
+            </Button>
+            {isEditing && (
+              <Button onClick={handleEdit} variant="ghost" className="p-2">
+                <Send className="w-3 h-3" />
+              </Button>
+            )}
+          </div>
+        )}
+      </div>
+    </div>
+  );
+};
+
+export default UserMessage;

frontend/src/components/dialogs/global/OnboardingDialog.tsx

@@ -50,7 +50,6 @@ const OnboardingDialog = NiceModal.create(() => {
   const [customCommand, setCustomCommand] = useState<string>('');
 
   const handleComplete = () => {
-    console.log('DEBUG1');
     modal.resolve({
       profile,
       editor: {

frontend/src/components/logs/ProcessGroup.tsx

@@ -1,45 +0,0 @@
-import type { UnifiedLogEntry, ProcessStartPayload } from '@/types/logs';
-import ProcessStartCard from '@/components/logs/ProcessStartCard';
-import LogEntryRow from '@/components/logs/LogEntryRow';
-
-type Props = {
-  header: ProcessStartPayload;
-  entries: UnifiedLogEntry[];
-  isCollapsed: boolean;
-  onToggle: (processId: string) => void;
-  retry?: {
-    onRetry: (processId: string, newPrompt: string) => void;
-    retryProcessId?: string;
-    retryDisabled?: boolean;
-    retryDisabledReason?: string;
-  };
-};
-
-export default function ProcessGroup({
-  header,
-  entries,
-  isCollapsed,
-  onToggle,
-  retry,
-}: Props) {
-  return (
-    <div className="px-4 mt-4">
-      <ProcessStartCard
-        payload={header}
-        isCollapsed={isCollapsed}
-        onToggle={onToggle}
-        onRetry={retry?.onRetry}
-        retryProcessId={retry?.retryProcessId}
-        retryDisabled={retry?.retryDisabled}
-        retryDisabledReason={retry?.retryDisabledReason}
-      />
-      <div className="text-sm">
-        {!isCollapsed &&
-          entries.length > 0 &&
-          entries.map((entry, i) => (
-            <LogEntryRow key={entry.id} entry={entry} index={i} />
-          ))}
-      </div>
-    </div>
-  );
-}

frontend/src/components/logs/ProcessStartCard.tsx

@@ -1,252 +0,0 @@
-import { ChevronDown, SquarePen, X, Check } from 'lucide-react';
-import { cn } from '@/lib/utils';
-import {
-  Tooltip,
-  TooltipContent,
-  TooltipProvider,
-  TooltipTrigger,
-} from '@/components/ui/tooltip';
-import type { ProcessStartPayload } from '@/types/logs';
-import type { ExecutorAction } from 'shared/types';
-import { PROCESS_RUN_REASONS } from '@/constants/processes';
-import { useEffect, useMemo, useState } from 'react';
-import { Button } from '@/components/ui/button';
-import { AutoExpandingTextarea } from '@/components/ui/auto-expanding-textarea';
-
-interface ProcessStartCardProps {
-  payload: ProcessStartPayload;
-  isCollapsed: boolean;
-  onToggle: (processId: string) => void;
-  // Retry flow (replaces restore): edit prompt then retry
-  onRetry?: (processId: string, newPrompt: string) => void;
-  retryProcessId?: string; // explicit id if payload lacks it in future
-  retryDisabled?: boolean;
-  retryDisabledReason?: string;
-}
-
-const extractPromptFromAction = (
-  action?: ExecutorAction | null
-): string | null => {
-  if (!action) return null;
-  const t = action.typ;
-  if (!t) return null;
-  if (
-    (t.type === 'CodingAgentInitialRequest' ||
-      t.type === 'CodingAgentFollowUpRequest') &&
-    typeof t.prompt === 'string' &&
-    t.prompt.trim()
-  ) {
-    return t.prompt;
-  }
-  return null;
-};
-
-function ProcessStartCard({
-  payload,
-  isCollapsed,
-  onToggle,
-  onRetry,
-  retryProcessId,
-  retryDisabled,
-  retryDisabledReason,
-}: ProcessStartCardProps) {
-  const getProcessLabel = (p: ProcessStartPayload) => {
-    if (p.runReason === PROCESS_RUN_REASONS.CODING_AGENT) {
-      const prompt = extractPromptFromAction(p.action);
-      return prompt || 'Coding Agent';
-    }
-    switch (p.runReason) {
-      case PROCESS_RUN_REASONS.SETUP_SCRIPT:
-        return 'Setup Script';
-      case PROCESS_RUN_REASONS.CLEANUP_SCRIPT:
-        return 'Cleanup Script';
-      case PROCESS_RUN_REASONS.DEV_SERVER:
-        return 'Dev Server';
-      default:
-        return p.runReason;
-    }
-  };
-
-  const handleClick = () => {
-    onToggle(payload.processId);
-  };
-
-  const handleKeyDown = (e: React.KeyboardEvent) => {
-    if (isEditing) return;
-    if (e.key === 'Enter' || e.key === ' ') {
-      e.preventDefault();
-      onToggle(payload.processId);
-    }
-  };
-
-  const label = getProcessLabel(payload);
-  const shouldTruncate =
-    isCollapsed && payload.runReason === PROCESS_RUN_REASONS.CODING_AGENT;
-
-  // Inline edit state for retry flow
-  const isCodingAgent = payload.runReason === PROCESS_RUN_REASONS.CODING_AGENT;
-  const [isEditing, setIsEditing] = useState(false);
-  const [editValue, setEditValue] = useState(label);
-
-  useEffect(() => {
-    if (!isEditing) setEditValue(label);
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, [label]);
-
-  const canRetry = useMemo(
-    () => !!onRetry && isCodingAgent,
-    [onRetry, isCodingAgent]
-  );
-  const doRetry = () => {
-    if (!onRetry) return;
-    const prompt = (editValue || '').trim();
-    if (!prompt) return; // no-op on empty
-    onRetry(retryProcessId || payload.processId, prompt);
-    setIsEditing(false);
-  };
-
-  return (
-    <div
-      className="p-2 border cursor-pointer select-none transition-colors w-full bg-background"
-      role="button"
-      tabIndex={0}
-      onClick={() => {
-        // Avoid toggling while editing or interacting with controls
-        if (isEditing) return;
-        handleClick();
-      }}
-      onKeyDown={handleKeyDown}
-    >
-      <div className="flex items-center gap-2 text-sm font-light">
-        <div className="flex items-center gap-2 text-foreground min-w-0 flex-1">
-          {isEditing && canRetry ? (
-            <div
-              className="flex items-center w-full"
-              onClick={(e) => e.stopPropagation()}
-            >
-              <AutoExpandingTextarea
-                value={editValue || ''}
-                onChange={(e) => setEditValue(e.target.value)}
-                onKeyDown={(e) => {
-                  if (e.key === 'Escape') {
-                    e.preventDefault();
-                    setIsEditing(false);
-                    setEditValue(label);
-                  }
-                }}
-                className={cn(
-                  'min-h-[36px] text-sm bg-inherit',
-                  shouldTruncate ? 'truncate' : 'whitespace-normal break-words'
-                )}
-                maxRows={12}
-                autoFocus
-              />
-              <Button
-                size="sm"
-                variant="ghost"
-                className="h-7"
-                disabled={!!retryDisabled || !(editValue || '').trim()}
-                onClick={(e) => {
-                  e.stopPropagation();
-                  doRetry();
-                }}
-                aria-label="Confirm edit and retry"
-              >
-                <Check className="h-4 w-4" />
-              </Button>
-              <Button
-                size="sm"
-                variant="ghost"
-                className="h-7"
-                onClick={(e) => {
-                  e.stopPropagation();
-                  setIsEditing(false);
-                  setEditValue(label);
-                }}
-                area-label="Cancel edit"
-              >
-                <X className="h-4 w-4" />
-              </Button>
-            </div>
-          ) : (
-            <>
-              <span
-                className={cn(
-                  shouldTruncate ? 'truncate' : 'whitespace-normal break-words'
-                )}
-                title={shouldTruncate ? label : undefined}
-              >
-                {label}
-              </span>
-            </>
-          )}
-        </div>
-
-        {/* Right controls: edit icon, status, chevron */}
-        {canRetry && !isEditing && (
-          <TooltipProvider>
-            <Tooltip>
-              <TooltipTrigger asChild>
-                {/* Wrap disabled button so tooltip still works */}
-                <span
-                  className="ml-2 inline-flex"
-                  onClick={(e) => e.stopPropagation()}
-                  aria-disabled={retryDisabled ? true : undefined}
-                >
-                  <button
-                    className={cn(
-                      'p-1 rounded transition-colors',
-                      retryDisabled
-                        ? 'cursor-not-allowed text-muted-foreground/60'
-                        : 'text-muted-foreground hover:text-foreground hover:bg-muted/60'
-                    )}
-                    onClick={(e) => {
-                      e.stopPropagation();
-                      if (retryDisabled) return;
-                      setIsEditing(true);
-                      setEditValue(label);
-                    }}
-                    aria-label="Edit prompt and retry from here"
-                    disabled={!!retryDisabled}
-                  >
-                    <SquarePen className="h-4 w-4" />
-                  </button>
-                </span>
-              </TooltipTrigger>
-              <TooltipContent>
-                {retryDisabled
-                  ? retryDisabledReason ||
-                    'Unavailable while another process is running.'
-                  : 'Edit prompt and retry'}
-              </TooltipContent>
-            </Tooltip>
-          </TooltipProvider>
-        )}
-
-        <div
-          className={cn(
-            'ml-1 text-xs px-2 py-1 rounded-full',
-            payload.status === 'running'
-              ? 'bg-blue-100 text-blue-700'
-              : payload.status === 'completed'
-                ? 'bg-green-100 text-green-700'
-                : payload.status === 'failed'
-                  ? 'bg-red-100 text-red-700'
-                  : 'bg-gray-100 text-gray-700'
-          )}
-        >
-          {payload.status}
-        </div>
-
-        <ChevronDown
-          className={cn(
-            'h-4 w-4 text-muted-foreground transition-transform',
-            isCollapsed && '-rotate-90'
-          )}
-        />
-      </div>
-    </div>
-  );
-}
-
-export default ProcessStartCard;

frontend/src/components/logs/VirtualizedList.tsx

@@ -0,0 +1,127 @@
+import {
+  VirtuosoMessageListProps,
+  VirtuosoMessageListMethods,
+  VirtuosoMessageListLicense,
+  VirtuosoMessageList,
+  DataWithScrollModifier,
+  ScrollModifier,
+} from '@virtuoso.dev/message-list';
+import { useEffect, useRef, useState } from 'react';
+import DisplayConversationEntry from '../NormalizedConversation/DisplayConversationEntry';
+import {
+  useConversationHistory,
+  PatchTypeWithKey,
+  AddEntryType,
+} from '@/hooks/useConversationHistory';
+import { TaskAttempt } from 'shared/types';
+import { Loader2 } from 'lucide-react';
+
+interface VirtualizedListProps {
+  attempt: TaskAttempt;
+}
+
+type ChannelData = DataWithScrollModifier<PatchTypeWithKey> | null;
+
+const InitialDataScrollModifier: ScrollModifier = {
+  type: 'item-location',
+  location: {
+    index: 'LAST',
+    align: 'end',
+  },
+  purgeItemSizes: true,
+};
+
+const AutoScrollToBottom: ScrollModifier = {
+  type: 'auto-scroll-to-bottom',
+  autoScroll: ({ atBottom, scrollInProgress }) => {
+    if (atBottom || scrollInProgress) {
+      return 'smooth';
+    }
+    return false;
+  },
+};
+
+const VirtualizedList = ({ attempt }: VirtualizedListProps) => {
+  const [channelData, setChannelData] = useState<ChannelData>(null);
+  const [loading, setLoading] = useState(true);
+
+  // When attempt changes, set loading
+  useEffect(() => {
+    setLoading(true);
+  }, [attempt.id]);
+
+  const onEntriesUpdated = (
+    newEntries: PatchTypeWithKey[],
+    addType: AddEntryType,
+    newLoading: boolean
+  ) => {
+    // initial defaults to scrolling to the latest
+    let scrollModifier: ScrollModifier = InitialDataScrollModifier;
+
+    if (addType === 'running' && !loading) {
+      scrollModifier = AutoScrollToBottom;
+    }
+
+    setChannelData({ data: newEntries, scrollModifier });
+    if (loading) {
+      setLoading(newLoading);
+    }
+  };
+  useConversationHistory({ attempt, onEntriesUpdated });
+
+  const messageListRef = useRef<VirtuosoMessageListMethods | null>(null);
+
+  const ItemContent: VirtuosoMessageListProps<
+    PatchTypeWithKey,
+    null
+  >['ItemContent'] = ({ data }) => {
+    if (data.type === 'STDOUT') {
+      return <p>{data.content}</p>;
+    } else if (data.type === 'STDERR') {
+      return <p>{data.content}</p>;
+    } else if (data.type === 'NORMALIZED_ENTRY') {
+      return (
+        <DisplayConversationEntry
+          key={data.patchKey}
+          expansionKey={data.patchKey}
+          entry={data.content}
+          executionProcessId={data.executionProcessId}
+          taskAttempt={attempt}
+        />
+      );
+    }
+  };
+
+  const computeItemKey: VirtuosoMessageListProps<
+    PatchTypeWithKey,
+    null
+  >['computeItemKey'] = ({ data }) => {
+    return `l-${data.patchKey}`;
+  };
+
+  return (
+    <>
+      <VirtuosoMessageListLicense
+        licenseKey={import.meta.env.PUBLIC_REACT_VIRTUOSO_LICENSE_KEY}
+      >
+        <VirtuosoMessageList<PatchTypeWithKey, null>
+          ref={messageListRef}
+          className="flex-1"
+          data={channelData}
+          computeItemKey={computeItemKey}
+          ItemContent={ItemContent}
+          Header={() => <div className="h-2"></div>} // Padding
+          Footer={() => <div className="h-2"></div>} // Padding
+        />
+      </VirtuosoMessageListLicense>
+      {loading && (
+        <div className="float-left top-0 left-0 w-full h-full bg-primary flex flex-col gap-2 justify-center items-center">
+          <Loader2 className="h-8 w-8 animate-spin" />
+          <p>Loading History</p>
+        </div>
+      )}
+    </>
+  );
+};
+
+export default VirtualizedList;

frontend/src/components/tasks/TaskDetails/ConversationExecutionLogs.tsx

@@ -0,0 +1,26 @@
+import DisplayConversationEntry from '@/components/NormalizedConversation/DisplayConversationEntry';
+import { useNormalizedLogs } from '@/hooks/useNormalizedLogs';
+import { ExecutionProcess } from 'shared/types';
+
+interface ConversationExecutionLogsProps {
+  executionProcess: ExecutionProcess;
+}
+
+const ConversationExecutionLogs = ({
+  executionProcess,
+}: ConversationExecutionLogsProps) => {
+  const { entries } = useNormalizedLogs(executionProcess.id);
+
+  console.log('DEBUG7', entries);
+
+  return entries.map((entry, i) => {
+    return (
+      <DisplayConversationEntry
+        expansionKey={`expansion-${executionProcess.id}-${i}`}
+        entry={entry}
+      />
+    );
+  });
+};
+
+export default ConversationExecutionLogs;

frontend/src/components/tasks/TaskDetails/LogsTab.tsx

@@ -1,494 +1,12 @@
-import {
-  useRef,
-  useCallback,
-  useMemo,
-  useEffect,
-  useReducer,
-  useState,
-} from 'react';
-import { Virtuoso, VirtuosoHandle } from 'react-virtuoso';
-import { Cog } from 'lucide-react';
-import { useAttemptExecution } from '@/hooks/useAttemptExecution';
-import { useBranchStatus } from '@/hooks/useBranchStatus';
-import { useProcessesLogs } from '@/hooks/useProcessesLogs';
-import ProcessGroup from '@/components/logs/ProcessGroup';
-import {
-  shouldShowInLogs,
-  isAutoCollapsibleProcess,
-  isProcessCompleted,
-  isCodingAgent,
-  getLatestCodingAgent,
-  PROCESS_STATUSES,
-  PROCESS_RUN_REASONS,
-} from '@/constants/processes';
-import type { ExecutionProcessStatus, TaskAttempt } from 'shared/types';
-import type { UnifiedLogEntry, ProcessStartPayload } from '@/types/logs';
-import { showModal } from '@/lib/modals';
-
-function addAll<T>(set: Set<T>, items: T[]): Set<T> {
-  items.forEach((i: T) => set.add(i));
-  return set;
-}
-
-// State management types
-type LogsState = {
-  userCollapsed: Set<string>;
-  autoCollapsed: Set<string>;
-  prevStatus: Map<string, ExecutionProcessStatus>;
-  prevLatestAgent?: string;
-};
-
-type LogsAction =
-  | { type: 'RESET_ATTEMPT' }
-  | { type: 'TOGGLE_USER'; id: string }
-  | { type: 'AUTO_COLLAPSE'; ids: string[] }
-  | { type: 'AUTO_EXPAND'; ids: string[] }
-  | { type: 'UPDATE_STATUS'; id: string; status: ExecutionProcessStatus }
-  | { type: 'NEW_RUNNING_AGENT'; id: string };
-
-const initialState: LogsState = {
-  userCollapsed: new Set(),
-  autoCollapsed: new Set(),
-  prevStatus: new Map(),
-  prevLatestAgent: undefined,
-};
-
-function reducer(state: LogsState, action: LogsAction): LogsState {
-  switch (action.type) {
-    case 'RESET_ATTEMPT':
-      return { ...initialState };
-
-    case 'TOGGLE_USER': {
-      const newUserCollapsed = new Set(state.userCollapsed);
-      const newAutoCollapsed = new Set(state.autoCollapsed);
-
-      const isCurrentlyCollapsed =
-        newUserCollapsed.has(action.id) || newAutoCollapsed.has(action.id);
-
-      if (isCurrentlyCollapsed) {
-        // we want to EXPAND
-        newUserCollapsed.delete(action.id);
-        newAutoCollapsed.delete(action.id);
-      } else {
-        // we want to COLLAPSE
-        newUserCollapsed.add(action.id);
-      }
-
-      return {
-        ...state,
-        userCollapsed: newUserCollapsed,
-        autoCollapsed: newAutoCollapsed,
-      };
-    }
-
-    case 'AUTO_COLLAPSE': {
-      const newAutoCollapsed = new Set(state.autoCollapsed);
-      addAll(newAutoCollapsed, action.ids);
-      return {
-        ...state,
-        autoCollapsed: newAutoCollapsed,
-      };
-    }
-
-    case 'AUTO_EXPAND': {
-      const newAutoCollapsed = new Set(state.autoCollapsed);
-      action.ids.forEach((id) => newAutoCollapsed.delete(id));
-      return {
-        ...state,
-        autoCollapsed: newAutoCollapsed,
-      };
-    }
-
-    case 'UPDATE_STATUS': {
-      const newPrevStatus = new Map(state.prevStatus);
-      newPrevStatus.set(action.id, action.status);
-      return {
-        ...state,
-        prevStatus: newPrevStatus,
-      };
-    }
-
-    case 'NEW_RUNNING_AGENT':
-      return {
-        ...state,
-        prevLatestAgent: action.id,
-      };
-
-    default:
-      return state;
-  }
-}
+import type { TaskAttempt } from 'shared/types';
+import VirtualizedList from '@/components/logs/VirtualizedList';
 
 type Props = {
-  selectedAttempt: TaskAttempt | null;
+  selectedAttempt: TaskAttempt;
 };
 
 function LogsTab({ selectedAttempt }: Props) {
-  const { attemptData, refetch } = useAttemptExecution(selectedAttempt?.id);
-  const { data: branchStatus, refetch: refetchBranch } = useBranchStatus(
-    selectedAttempt?.id
-  );
-  const virtuosoRef = useRef<VirtuosoHandle>(null);
-
-  const [state, dispatch] = useReducer(reducer, initialState);
-
-  // Filter out dev server processes before passing to useProcessesLogs
-  const filteredProcesses = useMemo(() => {
-    const processes = attemptData.processes || [];
-    return processes.filter(
-      (process) => shouldShowInLogs(process.run_reason) && !process.dropped
-    );
-  }, [
-    attemptData.processes
-      ?.map((p) => `${p.id}:${p.status}:${p.dropped}`)
-      .join(','),
-  ]);
-
-  // Detect if any process is running
-  const anyRunning = useMemo(
-    () => (attemptData.processes || []).some((p) => p.status === 'running'),
-    [attemptData.processes?.map((p) => p.status).join(',')]
-  );
-
-  const { entries } = useProcessesLogs(filteredProcesses, true);
-  const [restoreBusy, setRestoreBusy] = useState(false);
-
-  // Combined collapsed processes (auto + user)
-  const allCollapsedProcesses = useMemo(() => {
-    const combined = new Set(state.autoCollapsed);
-    state.userCollapsed.forEach((id: string) => combined.add(id));
-    return combined;
-  }, [state.autoCollapsed, state.userCollapsed]);
-
-  // Toggle collapsed state for a process (user action)
-  const toggleProcessCollapse = useCallback((processId: string) => {
-    dispatch({ type: 'TOGGLE_USER', id: processId });
-  }, []);
-
-  // Effect #1: Reset state when attempt changes
-  useEffect(() => {
-    dispatch({ type: 'RESET_ATTEMPT' });
-  }, [selectedAttempt?.id]);
-
-  // Effect #2: Handle setup/cleanup script auto-collapse and auto-expand
-  useEffect(() => {
-    const toCollapse: string[] = [];
-    const toExpand: string[] = [];
-
-    filteredProcesses.forEach((process) => {
-      if (isAutoCollapsibleProcess(process.run_reason)) {
-        const prevStatus = state.prevStatus.get(process.id);
-        const currentStatus = process.status;
-
-        // Auto-collapse completed setup/cleanup scripts
-        const shouldAutoCollapse =
-          (prevStatus === PROCESS_STATUSES.RUNNING ||
-            prevStatus === undefined) &&
-          isProcessCompleted(currentStatus) &&
-          !state.userCollapsed.has(process.id) &&
-          !state.autoCollapsed.has(process.id);
-
-        if (shouldAutoCollapse) {
-          toCollapse.push(process.id);
-        }
-
-        // Auto-expand scripts that restart after completion
-        const becameRunningAgain =
-          prevStatus &&
-          isProcessCompleted(prevStatus) &&
-          currentStatus === PROCESS_STATUSES.RUNNING &&
-          state.autoCollapsed.has(process.id);
-
-        if (becameRunningAgain) {
-          toExpand.push(process.id);
-        }
-
-        // Update status tracking
-        dispatch({
-          type: 'UPDATE_STATUS',
-          id: process.id,
-          status: currentStatus,
-        });
-      }
-    });
-
-    if (toCollapse.length > 0) {
-      dispatch({ type: 'AUTO_COLLAPSE', ids: toCollapse });
-    }
-
-    if (toExpand.length > 0) {
-      dispatch({ type: 'AUTO_EXPAND', ids: toExpand });
-    }
-  }, [filteredProcesses, state.userCollapsed, state.autoCollapsed]);
-
-  // Effect #3: Handle coding agent succession logic
-  useEffect(() => {
-    const latestCodingAgentId = getLatestCodingAgent(filteredProcesses);
-    if (!latestCodingAgentId) return;
-
-    // Collapse previous agents when a new latest agent appears
-    if (latestCodingAgentId !== state.prevLatestAgent) {
-      // Collapse all other coding agents that aren't user-collapsed
-      const toCollapse = filteredProcesses
-        .filter(
-          (p) =>
-            isCodingAgent(p.run_reason) &&
-            p.id !== latestCodingAgentId &&
-            !state.userCollapsed.has(p.id) &&
-            !state.autoCollapsed.has(p.id)
-        )
-        .map((p) => p.id);
-
-      if (toCollapse.length > 0) {
-        dispatch({ type: 'AUTO_COLLAPSE', ids: toCollapse });
-      }
-
-      dispatch({ type: 'NEW_RUNNING_AGENT', id: latestCodingAgentId });
-    }
-  }, [
-    filteredProcesses,
-    state.prevLatestAgent,
-    state.userCollapsed,
-    state.autoCollapsed,
-  ]);
-
-  const groups = useMemo(() => {
-    const map = new Map<
-      string,
-      { header?: ProcessStartPayload; entries: UnifiedLogEntry[] }
-    >();
-
-    filteredProcesses.forEach((p) => {
-      map.set(p.id, { header: undefined, entries: [] });
-    });
-
-    entries.forEach((e: UnifiedLogEntry) => {
-      const bucket = map.get(e.processId);
-      if (!bucket) return;
-
-      if (e.channel === 'process_start') {
-        bucket.header = e.payload as ProcessStartPayload;
-        return;
-      }
-
-      // Always store entries; whether they show is decided by group collapse
-      bucket.entries.push(e);
-    });
-
-    return filteredProcesses
-      .map((p) => ({
-        processId: p.id,
-        ...(map.get(p.id) || { entries: [] }),
-      }))
-      .filter((g) => g.header) as Array<{
-      processId: string;
-      header: ProcessStartPayload;
-      entries: UnifiedLogEntry[];
-    }>;
-  }, [filteredProcesses, entries]);
-
-  const itemContent = useCallback(
-    (
-      _index: number,
-      group: {
-        processId: string;
-        header: ProcessStartPayload;
-        entries: UnifiedLogEntry[];
-      }
-    ) =>
-      (() => {
-        // Compute retry props (replaces restore)
-        let retry:
-          | {
-              onRetry: (pid: string, newPrompt: string) => void;
-              retryProcessId?: string;
-              retryDisabled?: boolean;
-              retryDisabledReason?: string;
-            }
-          | undefined;
-
-        {
-          const proc = (attemptData.processes || []).find(
-            (p) => p.id === group.processId
-          );
-          const isRunningProc = proc?.status === 'running';
-          const isCoding = proc?.run_reason === 'codingagent';
-          // Always show for coding agent processes
-          const shouldShow = !!isCoding;
-
-          if (shouldShow) {
-            const disabled = anyRunning || restoreBusy || isRunningProc;
-            let disabledReason: string | undefined;
-            if (isRunningProc)
-              disabledReason = 'Finish or stop this run to retry.';
-            else if (anyRunning)
-              disabledReason = 'Cannot retry while a process is running.';
-            else if (restoreBusy) disabledReason = 'Retry in progress.';
-
-            retry = {
-              retryProcessId: group.processId,
-              retryDisabled: disabled,
-              retryDisabledReason: disabledReason,
-              onRetry: async (pid: string, newPrompt: string) => {
-                const p2 = (attemptData.processes || []).find(
-                  (p) => p.id === pid
-                );
-                type WithBefore = { before_head_commit?: string | null };
-                const before =
-                  (p2 as WithBefore | undefined)?.before_head_commit || null;
-                let targetSubject = null;
-                let commitsToReset = null;
-                let isLinear = null;
-
-                if (before && selectedAttempt?.id) {
-                  try {
-                    const { commitsApi } = await import('@/lib/api');
-                    const info = await commitsApi.getInfo(
-                      selectedAttempt.id,
-                      before
-                    );
-                    targetSubject = info.subject;
-                    const cmp = await commitsApi.compareToHead(
-                      selectedAttempt.id,
-                      before
-                    );
-                    commitsToReset = cmp.is_linear ? cmp.ahead_from_head : null;
-                    isLinear = cmp.is_linear;
-                  } catch {
-                    /* ignore */
-                  }
-                }
-
-                const head = branchStatus?.head_oid || null;
-                const dirty = !!branchStatus?.has_uncommitted_changes;
-                const needReset = !!(before && (before !== head || dirty));
-                const canGitReset = needReset && !dirty;
-
-                // Calculate later process counts for dialog
-                const procs = (attemptData.processes || []).filter(
-                  (p) => !p.dropped && shouldShowInLogs(p.run_reason)
-                );
-                const idx = procs.findIndex((p) => p.id === pid);
-                const laterCount = idx >= 0 ? procs.length - (idx + 1) : 0;
-                const later = idx >= 0 ? procs.slice(idx + 1) : [];
-                const laterCoding = later.filter((p) =>
-                  isCodingAgent(p.run_reason)
-                ).length;
-                const laterSetup = later.filter(
-                  (p) => p.run_reason === PROCESS_RUN_REASONS.SETUP_SCRIPT
-                ).length;
-                const laterCleanup = later.filter(
-                  (p) => p.run_reason === PROCESS_RUN_REASONS.CLEANUP_SCRIPT
-                ).length;
-
-                try {
-                  const result = await showModal<{
-                    action: 'confirmed' | 'canceled';
-                    performGitReset?: boolean;
-                    forceWhenDirty?: boolean;
-                  }>('restore-logs', {
-                    targetSha: before,
-                    targetSubject,
-                    commitsToReset,
-                    isLinear,
-                    laterCount,
-                    laterCoding,
-                    laterSetup,
-                    laterCleanup,
-                    needGitReset: needReset,
-                    canGitReset,
-                    hasRisk: dirty,
-                    uncommittedCount: branchStatus?.uncommitted_count ?? 0,
-                    untrackedCount: branchStatus?.untracked_count ?? 0,
-                    // Always default to performing a worktree reset
-                    initialWorktreeResetOn: true,
-                    initialForceReset: false,
-                  });
-
-                  if (result.action === 'confirmed' && selectedAttempt?.id) {
-                    const { attemptsApi } = await import('@/lib/api');
-                    try {
-                      setRestoreBusy(true);
-                      // Determine variant from the original process executor profile if available
-                      let variant: string | null = null;
-                      const typ = p2?.executor_action?.typ;
-                      if (
-                        typ &&
-                        (typ.type === 'CodingAgentInitialRequest' ||
-                          typ.type === 'CodingAgentFollowUpRequest')
-                      ) {
-                        variant = typ.executor_profile_id?.variant ?? null;
-                      }
-                      await attemptsApi.replaceProcess(selectedAttempt.id, {
-                        process_id: pid,
-                        prompt: newPrompt,
-                        variant,
-                        perform_git_reset: result.performGitReset ?? true,
-                        force_when_dirty: result.forceWhenDirty ?? false,
-                      });
-                      await refetch();
-                      await refetchBranch();
-                    } finally {
-                      setRestoreBusy(false);
-                    }
-                  }
-                } catch (error) {
-                  // User cancelled - do nothing
-                }
-              },
-            };
-          }
-        }
-
-        return (
-          <ProcessGroup
-            header={group.header}
-            entries={group.entries}
-            isCollapsed={allCollapsedProcesses.has(group.processId)}
-            onToggle={toggleProcessCollapse}
-            retry={retry}
-          />
-        );
-      })(),
-    [
-      allCollapsedProcesses,
-      toggleProcessCollapse,
-      anyRunning,
-      restoreBusy,
-      selectedAttempt?.id,
-      attemptData.processes,
-      branchStatus?.head_oid,
-      branchStatus?.has_uncommitted_changes,
-    ]
-  );
-
-  if (!filteredProcesses || filteredProcesses.length === 0) {
-    return (
-      <div className="flex-1 flex items-center justify-center text-muted-foreground">
-        <div className="text-center">
-          <Cog className="h-12 w-12 mx-auto mb-4 opacity-50" />
-          <p>No execution processes found for this attempt.</p>
-        </div>
-      </div>
-    );
-  }
-
-  return (
-    <div className="w-full h-full flex flex-col">
-      <div className="flex-1">
-        <Virtuoso
-          ref={virtuosoRef}
-          style={{ height: '100%' }}
-          data={groups}
-          itemContent={itemContent}
-          followOutput
-          increaseViewportBy={200}
-          overscan={5}
-          components={{ Footer: () => <div className="pb-4" /> }}
-        />
-      </div>
-    </div>
-  );
+  return <VirtualizedList attempt={selectedAttempt} />;
 }
 
-export default LogsTab; // Filter entries to hide logs from collapsed processes
+export default LogsTab;

frontend/src/components/tasks/TaskDetailsPanel.tsx

@@ -177,29 +177,33 @@ export function TaskDetailsPanel({
 
                       {/* Main content */}
                       <main className="flex-1 min-h-0 min-w-0 flex flex-col">
-                        <TabNavigation
-                          activeTab={activeTab}
-                          setActiveTab={setActiveTab}
-                          selectedAttempt={selectedAttempt}
-                        />
+                        {selectedAttempt && (
+                          <>
+                            <TabNavigation
+                              activeTab={activeTab}
+                              setActiveTab={setActiveTab}
+                              selectedAttempt={selectedAttempt}
+                            />
 
-                        <div className="flex-1 flex flex-col min-h-0">
-                          {activeTab === 'diffs' ? (
-                            <DiffTab selectedAttempt={selectedAttempt} />
-                          ) : activeTab === 'processes' ? (
-                            <ProcessesTab attemptId={selectedAttempt?.id} />
-                          ) : (
-                            <LogsTab selectedAttempt={selectedAttempt} />
-                          )}
-                        </div>
+                            <div className="flex-1 flex flex-col min-h-0">
+                              {activeTab === 'diffs' ? (
+                                <DiffTab selectedAttempt={selectedAttempt} />
+                              ) : activeTab === 'processes' ? (
+                                <ProcessesTab attemptId={selectedAttempt?.id} />
+                              ) : (
+                                <LogsTab selectedAttempt={selectedAttempt} />
+                              )}
+                            </div>
 
-                        <TaskFollowUpSection
-                          task={task}
-                          projectId={projectId}
-                          selectedAttemptId={selectedAttempt?.id}
-                          selectedAttemptProfile={selectedAttempt?.executor}
-                          jumpToLogsTab={jumpToLogsTab}
-                        />
+                            <TaskFollowUpSection
+                              task={task}
+                              projectId={projectId}
+                              selectedAttemptId={selectedAttempt?.id}
+                              selectedAttemptProfile={selectedAttempt?.executor}
+                              jumpToLogsTab={jumpToLogsTab}
+                            />
+                          </>
+                        )}
                       </main>
                     </div>
                   ) : (
@@ -231,7 +235,9 @@ export function TaskDetailsPanel({
                             onJumpToDiffFullScreen={jumpToDiffFullScreen}
                           />
 
-                          <LogsTab selectedAttempt={selectedAttempt} />
+                          {selectedAttempt && (
+                            <LogsTab selectedAttempt={selectedAttempt} />
+                          )}
 
                           <TaskFollowUpSection
                             task={task}

frontend/src/components/ui/markdown-renderer.tsx

@@ -91,7 +91,7 @@ function MarkdownRenderer({
     try {
       await writeClipboardViaBridge(content);
       setCopied(true);
-      window.setTimeout(() => setCopied(false), 1400);
+      window.setTimeout(() => setCopied(false), 400);
     } catch {
       // noop – bridge handles fallback
     }
@@ -100,7 +100,7 @@ function MarkdownRenderer({
   return (
     <div className={`relative group`}>
       {enableCopyButton && (
-        <div className="sticky top-2 z-10 pointer-events-none">
+        <div className="sticky top-2 right-2 z-10 pointer-events-none h-0">
           <div className="flex justify-end pr-1">
             <TooltipProvider>
               <Tooltip>
@@ -113,7 +113,7 @@ function MarkdownRenderer({
                       variant="outline"
                       size="icon"
                       onClick={handleCopy}
-                      className="pointer-events-auto opacity-0 group-hover:opacity-100 group-hover:delay-200 delay-0 transition-opacity duration-150 h-8 w-8 rounded-md bg-background/95 backdrop-blur border border-border shadow-sm"
+                      className="pointer-events-auto opacity-0 group-hover:opacity-100 delay-0 transition-opacity duration-50 h-8 w-8 rounded-md bg-background/95 backdrop-blur border border-border shadow-sm"
                     >
                       {copied ? (
                         <Check className="h-4 w-4 text-green-600" />

frontend/src/hooks/useConversationHistory.ts

@@ -0,0 +1,474 @@
+// useConversationHistory.ts
+import {
+  CommandExitStatus,
+  ExecutionProcess,
+  ExecutorAction,
+  NormalizedEntry,
+  PatchType,
+  TaskAttempt,
+} from 'shared/types';
+import { useExecutionProcesses } from './useExecutionProcesses';
+import { useEffect, useMemo, useRef } from 'react';
+import { streamSseJsonPatchEntries } from '@/utils/streamSseJsonPatchEntries';
+
+export type PatchTypeWithKey = PatchType & {
+  patchKey: string;
+  executionProcessId: string;
+};
+
+export type AddEntryType = 'initial' | 'running' | 'historic';
+
+export type OnEntriesUpdated = (
+  newEntries: PatchTypeWithKey[],
+  addType: AddEntryType,
+  loading: boolean
+) => void;
+
+type ExecutionProcessStaticInfo = {
+  id: string;
+  created_at: string;
+  updated_at: string;
+  executor_action: ExecutorAction;
+};
+
+type ExecutionProcessState = {
+  executionProcess: ExecutionProcessStaticInfo;
+  entries: PatchTypeWithKey[];
+};
+
+type ExecutionProcessStateStore = Record<string, ExecutionProcessState>;
+
+interface UseConversationHistoryParams {
+  attempt: TaskAttempt;
+  onEntriesUpdated: OnEntriesUpdated;
+}
+
+interface UseConversationHistoryResult {}
+
+const MIN_INITIAL_ENTRIES = 10;
+const REMAINING_BATCH_SIZE = 50;
+
+export const useConversationHistory = ({
+  attempt,
+  onEntriesUpdated,
+}: UseConversationHistoryParams): UseConversationHistoryResult => {
+  const { executionProcesses: executionProcessesRaw } = useExecutionProcesses(
+    attempt.id
+  );
+  const executionProcesses = useRef<ExecutionProcess[]>(executionProcessesRaw);
+  const displayedExecutionProcesses = useRef<ExecutionProcessStateStore>({});
+  const loadedInitialEntries = useRef(false);
+  const lastRunningProcessId = useRef<string | null>(null);
+  const onEntriesUpdatedRef = useRef<OnEntriesUpdated | null>(null);
+  useEffect(() => {
+    onEntriesUpdatedRef.current = onEntriesUpdated;
+  }, [onEntriesUpdated]);
+
+  // Keep executionProcesses up to date with executionProcessesRaw
+  useEffect(() => {
+    executionProcesses.current = executionProcessesRaw;
+  }, [executionProcessesRaw]);
+
+  const loadEntriesForHistoricExecutionProcess = (
+    executionProcess: ExecutionProcess
+  ) => {
+    let url = '';
+    if (executionProcess.executor_action.typ.type === 'ScriptRequest') {
+      url = `/api/execution-processes/${executionProcess.id}/raw-logs`;
+    } else {
+      url = `/api/execution-processes/${executionProcess.id}/normalized-logs`;
+    }
+
+    return new Promise<PatchType[]>((resolve) => {
+      const controller = streamSseJsonPatchEntries<PatchType>(url, {
+        onFinished: (allEntries) => {
+          controller.close();
+          resolve(allEntries);
+        },
+        onError: (err) => {
+          console.warn!(
+            `Error loading entries for historic execution process ${executionProcess.id}`,
+            err
+          );
+          controller.close();
+          resolve([]);
+        },
+      });
+    });
+  };
+
+  const getLiveExecutionProcess = (
+    executionProcessId: string
+  ): ExecutionProcess | undefined => {
+    return executionProcesses?.current.find(
+      (executionProcess) => executionProcess.id === executionProcessId
+    );
+  };
+
+  // This emits its own events as they are streamed
+  const loadRunningAndEmit = (
+    executionProcess: ExecutionProcess
+  ): Promise<void> => {
+    return new Promise((resolve, reject) => {
+      let url = '';
+      if (executionProcess.executor_action.typ.type === 'ScriptRequest') {
+        url = `/api/execution-processes/${executionProcess.id}/raw-logs`;
+      } else {
+        url = `/api/execution-processes/${executionProcess.id}/normalized-logs`;
+      }
+      const controller = streamSseJsonPatchEntries<PatchType>(url, {
+        onEntries(entries) {
+          const patchesWithKey = entries.map((entry, index) =>
+            patchWithKey(entry, executionProcess.id, index)
+          );
+          const localEntries = displayedExecutionProcesses.current;
+          localEntries[executionProcess.id] = {
+            executionProcess,
+            entries: patchesWithKey,
+          };
+          displayedExecutionProcesses.current = localEntries;
+          emitEntries(localEntries, 'running', false);
+        },
+        onFinished: () => {
+          controller.close();
+          resolve();
+        },
+        onError: () => {
+          controller.close();
+          reject();
+        },
+      });
+    });
+  };
+
+  // Sometimes it can take a few seconds for the stream to start, wrap the loadRunningAndEmit method
+  const loadRunningAndEmitWithBackoff = async (
+    executionProcess: ExecutionProcess
+  ) => {
+    for (let i = 0; i < 20; i++) {
+      try {
+        await loadRunningAndEmit(executionProcess);
+        break;
+      } catch (_) {
+        await new Promise((resolve) => setTimeout(resolve, 500));
+      }
+    }
+  };
+
+  const getRunningExecutionProcesses = (): ExecutionProcess | null => {
+    // If more than one, throw an error
+    const runningProcesses = executionProcesses?.current.filter(
+      (p) => p.status === 'running'
+    );
+    if (runningProcesses.length > 1) {
+      throw new Error('More than one running execution process found');
+    }
+    return runningProcesses[0] || null;
+  };
+
+  const flattenEntries = (
+    executionProcessState: ExecutionProcessStateStore
+  ): PatchTypeWithKey[] => {
+    return Object.values(executionProcessState)
+      .filter(
+        (p) =>
+          p.executionProcess.executor_action.typ.type ===
+            'CodingAgentFollowUpRequest' ||
+          p.executionProcess.executor_action.typ.type ===
+            'CodingAgentInitialRequest'
+      )
+      .sort(
+        (a, b) =>
+          new Date(
+            a.executionProcess.created_at as unknown as string
+          ).getTime() -
+          new Date(b.executionProcess.created_at as unknown as string).getTime()
+      )
+      .flatMap((p) => p.entries);
+  };
+
+  const loadingPatch: PatchTypeWithKey = {
+    type: 'NORMALIZED_ENTRY',
+    content: {
+      entry_type: {
+        type: 'loading',
+      },
+      content: '',
+      timestamp: null,
+    },
+    patchKey: 'loading',
+    executionProcessId: '',
+  };
+
+  const flattenEntriesForEmit = (
+    executionProcessState: ExecutionProcessStateStore
+  ): PatchTypeWithKey[] => {
+    // Create user messages + tool calls for setup/cleanup scripts
+    const allEntries = Object.values(executionProcessState)
+      .sort(
+        (a, b) =>
+          new Date(
+            a.executionProcess.created_at as unknown as string
+          ).getTime() -
+          new Date(b.executionProcess.created_at as unknown as string).getTime()
+      )
+      .flatMap((p) => {
+        const entries: PatchTypeWithKey[] = [];
+        if (
+          p.executionProcess.executor_action.typ.type ===
+            'CodingAgentInitialRequest' ||
+          p.executionProcess.executor_action.typ.type ===
+            'CodingAgentFollowUpRequest'
+        ) {
+          // New user message
+          const userNormalizedEntry: NormalizedEntry = {
+            entry_type: {
+              type: 'user_message',
+            },
+            content: p.executionProcess.executor_action.typ.prompt,
+            timestamp: null,
+          };
+          const userPatch: PatchType = {
+            type: 'NORMALIZED_ENTRY',
+            content: userNormalizedEntry,
+          };
+          const userPatchTypeWithKey = patchWithKey(
+            userPatch,
+            p.executionProcess.id,
+            'user'
+          );
+          entries.push(userPatchTypeWithKey);
+
+          // Remove all coding agent added user messages, replace with our custom one
+          const entriesExcludingUser = p.entries.filter(
+            (e) =>
+              e.type !== 'NORMALIZED_ENTRY' ||
+              e.content.entry_type.type !== 'user_message'
+          );
+          entries.push(...entriesExcludingUser);
+          if (
+            getLiveExecutionProcess(p.executionProcess.id)?.status === 'running'
+          ) {
+            entries.push(loadingPatch);
+          }
+        } else if (
+          p.executionProcess.executor_action.typ.type === 'ScriptRequest'
+        ) {
+          // Add setup and cleanup script as a tool call
+          let toolName = '';
+          switch (p.executionProcess.executor_action.typ.context) {
+            case 'SetupScript':
+              toolName = 'Setup Script';
+              break;
+            case 'CleanupScript':
+              toolName = 'Cleanup Script';
+              break;
+            default:
+              return [];
+          }
+
+          const executionProcess = getLiveExecutionProcess(
+            p.executionProcess.id
+          );
+
+          const exit_status: CommandExitStatus | null =
+            executionProcess?.status === 'running'
+              ? null
+              : {
+                  type: 'exit_code',
+                  code: Number(executionProcess?.exit_code) || 0,
+                };
+          const output = p.entries.map((line) => line.content).join('\n');
+
+          const toolNormalizedEntry: NormalizedEntry = {
+            entry_type: {
+              type: 'tool_use',
+              tool_name: toolName,
+              action_type: {
+                action: 'command_run',
+                command: p.executionProcess.executor_action.typ.script,
+                result: {
+                  output,
+                  exit_status,
+                },
+              },
+            },
+            content: toolName,
+            timestamp: null,
+          };
+          const toolPatch: PatchType = {
+            type: 'NORMALIZED_ENTRY',
+            content: toolNormalizedEntry,
+          };
+          const toolPatchWithKey: PatchTypeWithKey = patchWithKey(
+            toolPatch,
+            p.executionProcess.id,
+            0
+          );
+
+          entries.push(toolPatchWithKey);
+        }
+
+        return entries;
+      });
+
+    return allEntries;
+  };
+
+  const patchWithKey = (
+    patch: PatchType,
+    executionProcessId: string,
+    index: number | 'user'
+  ) => {
+    return {
+      ...patch,
+      patchKey: `${executionProcessId}:${index}`,
+      executionProcessId,
+    };
+  };
+
+  const loadInitialEntries = async (): Promise<ExecutionProcessStateStore> => {
+    const localDisplayedExecutionProcesses: ExecutionProcessStateStore = {};
+
+    if (!executionProcesses?.current) return localDisplayedExecutionProcesses;
+
+    for (const executionProcess of [...executionProcesses.current].reverse()) {
+      if (executionProcess.status === 'running') continue;
+
+      const entries =
+        await loadEntriesForHistoricExecutionProcess(executionProcess);
+      const entriesWithKey = entries.map((e, idx) =>
+        patchWithKey(e, executionProcess.id, idx)
+      );
+
+      localDisplayedExecutionProcesses[executionProcess.id] = {
+        executionProcess,
+        entries: entriesWithKey,
+      };
+
+      if (
+        flattenEntries(localDisplayedExecutionProcesses).length >
+        MIN_INITIAL_ENTRIES
+      ) {
+        break;
+      }
+    }
+
+    return localDisplayedExecutionProcesses;
+  };
+
+  const loadRemainingEntriesInBatches = async (
+    batchSize: number
+  ): Promise<ExecutionProcessStateStore | null> => {
+    const local = displayedExecutionProcesses.current; // keep ref if intentional
+    if (!executionProcesses?.current) return null;
+
+    let anyUpdated = false;
+    for (const executionProcess of [...executionProcesses.current].reverse()) {
+      if (local[executionProcess.id] || executionProcess.status === 'running')
+        continue;
+
+      const entries =
+        await loadEntriesForHistoricExecutionProcess(executionProcess);
+      const entriesWithKey = entries.map((e, idx) =>
+        patchWithKey(e, executionProcess.id, idx)
+      );
+
+      local[executionProcess.id] = {
+        executionProcess,
+        entries: entriesWithKey,
+      };
+      if (flattenEntries(local).length > batchSize) {
+        anyUpdated = true;
+        break;
+      }
+      anyUpdated = true;
+    }
+    return anyUpdated ? local : null;
+  };
+
+  const emitEntries = (
+    executionProcessState: ExecutionProcessStateStore,
+    addEntryType: AddEntryType,
+    loading: boolean
+  ) => {
+    // Flatten entries in chronological order of process start
+    const entries = flattenEntriesForEmit(executionProcessState);
+    onEntriesUpdatedRef.current?.(entries, addEntryType, loading);
+  };
+
+  // Stable key for dependency arrays when process list changes
+  const idListKey = useMemo(
+    () => executionProcessesRaw?.map((p) => p.id).join(','),
+    [executionProcessesRaw]
+  );
+
+  // Initial load when attempt changes
+  useEffect(() => {
+    let cancelled = false;
+    (async () => {
+      // Waiting for execution processes to load
+      if (
+        executionProcesses?.current.length === 0 ||
+        loadedInitialEntries.current
+      )
+        return;
+
+      // Initial entries
+      const allInitialEntries = await loadInitialEntries();
+      if (cancelled) return;
+      displayedExecutionProcesses.current = allInitialEntries;
+      emitEntries(allInitialEntries, 'initial', false);
+      loadedInitialEntries.current = true;
+
+      // Then load the remaining in batches
+      let updatedEntries;
+      while (
+        !cancelled &&
+        (updatedEntries =
+          await loadRemainingEntriesInBatches(REMAINING_BATCH_SIZE))
+      ) {
+        if (cancelled) return;
+        displayedExecutionProcesses.current = updatedEntries;
+      }
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      emitEntries(displayedExecutionProcesses.current, 'historic', false);
+    })();
+    return () => {
+      cancelled = true;
+    };
+  }, [attempt.id, idListKey]); // include idListKey so new processes trigger reload
+
+  // Running processes
+  useEffect(() => {
+    const runningProcess = getRunningExecutionProcesses();
+    if (runningProcess && lastRunningProcessId.current !== runningProcess.id) {
+      lastRunningProcessId.current = runningProcess.id;
+      loadRunningAndEmitWithBackoff(runningProcess);
+    }
+  }, [attempt.id, idListKey]);
+
+  // If an execution process is removed, remove it from the state
+  useEffect(() => {
+    if (!executionProcessesRaw) return;
+
+    const removedProcessIds = Object.keys(
+      displayedExecutionProcesses.current
+    ).filter((id) => !executionProcessesRaw.some((p) => p.id === id));
+
+    removedProcessIds.forEach((id) => {
+      delete displayedExecutionProcesses.current[id];
+    });
+  }, [attempt.id, idListKey]);
+
+  // Reset state when attempt changes
+  useEffect(() => {
+    displayedExecutionProcesses.current = {};
+    loadedInitialEntries.current = false;
+    lastRunningProcessId.current = null;
+    // Emit blank entries
+    emitEntries(displayedExecutionProcesses.current, 'initial', true);
+  }, [attempt.id]);
+
+  return {};
+};

frontend/src/hooks/useExecutionProcesses.ts

@@ -0,0 +1,54 @@
+import { useCallback } from 'react';
+import { useJsonPatchStream } from './useJsonPatchStream';
+import type { ExecutionProcess } from 'shared/types';
+
+type ExecutionProcessState = {
+  execution_processes: Record<string, ExecutionProcess>;
+};
+
+interface UseExecutionProcessesResult {
+  executionProcesses: ExecutionProcess[];
+  executionProcessesById: Record<string, ExecutionProcess>;
+  isLoading: boolean;
+  isConnected: boolean;
+  error: string | null;
+}
+
+/**
+ * Stream tasks for a project via SSE (JSON Patch) and expose as array + map.
+ * Server sends initial snapshot: replace /tasks with an object keyed by id.
+ * Live updates arrive at /tasks/<id> via add/replace/remove operations.
+ */
+export const useExecutionProcesses = (
+  taskAttemptId: string
+): UseExecutionProcessesResult => {
+  const endpoint = `/api/execution-processes/stream?task_attempt_id=${encodeURIComponent(taskAttemptId)}`;
+
+  const initialData = useCallback(
+    (): ExecutionProcessState => ({ execution_processes: {} }),
+    []
+  );
+
+  const { data, isConnected, error } =
+    useJsonPatchStream<ExecutionProcessState>(
+      endpoint,
+      !!taskAttemptId,
+      initialData
+    );
+
+  const executionProcessesById = data?.execution_processes ?? {};
+  const executionProcesses = Object.values(executionProcessesById).sort(
+    (a, b) =>
+      new Date(a.created_at as unknown as string).getTime() -
+      new Date(b.created_at as unknown as string).getTime()
+  );
+  const isLoading = !data && !error; // until first snapshot
+
+  return {
+    executionProcesses,
+    executionProcessesById,
+    isLoading,
+    isConnected,
+    error,
+  };
+};

frontend/src/hooks/useNormalizedLogs.tsx

@@ -0,0 +1,58 @@
+// useNormalizedLogs.ts
+import { useCallback, useMemo } from 'react';
+import { useJsonPatchStream } from './useJsonPatchStream';
+import { NormalizedEntry } from 'shared/types';
+
+type EntryType = { type: string };
+
+export interface NormalizedEntryContent {
+  timestamp: string | null;
+  entry_type: EntryType;
+  content: string;
+  metadata: Record<string, unknown> | null;
+}
+
+export interface NormalizedLogsState {
+  entries: NormalizedEntry[];
+  session_id: string | null;
+  executor_type: string;
+  prompt: string | null;
+  summary: string | null;
+}
+
+interface UseNormalizedLogsResult {
+  entries: NormalizedEntry[];
+  state: NormalizedLogsState | undefined;
+  isLoading: boolean;
+  isConnected: boolean;
+  error: string | null;
+}
+
+export const useNormalizedLogs = (
+  processId: string,
+  enabled: boolean = true
+): UseNormalizedLogsResult => {
+  const endpoint = `/api/execution-processes/${encodeURIComponent(processId)}/normalized-logs`;
+
+  const initialData = useCallback<() => NormalizedLogsState>(
+    () => ({
+      entries: [],
+      session_id: null,
+      executor_type: '',
+      prompt: null,
+      summary: null,
+    }),
+    []
+  );
+
+  const { data, isConnected, error } = useJsonPatchStream<NormalizedLogsState>(
+    endpoint,
+    Boolean(processId) && enabled,
+    initialData
+  );
+
+  const entries = useMemo(() => data?.entries ?? [], [data?.entries]);
+  const isLoading = !data && !error;
+
+  return { entries, state: data, isLoading, isConnected, error };
+};

frontend/src/hooks/useProcessRetry.ts

@@ -0,0 +1,229 @@
+// hooks/useProcessRetry.ts
+import { useCallback, useMemo, useState } from 'react';
+import { useAttemptExecution } from '@/hooks/useAttemptExecution';
+import { useBranchStatus } from '@/hooks/useBranchStatus';
+import { showModal } from '@/lib/modals';
+import {
+  shouldShowInLogs,
+  isCodingAgent,
+  PROCESS_RUN_REASONS,
+} from '@/constants/processes';
+import type { ExecutionProcess, TaskAttempt } from 'shared/types';
+import type {
+  ExecutorActionType,
+  CodingAgentInitialRequest,
+  CodingAgentFollowUpRequest,
+} from 'shared/types';
+
+function isCodingAgentActionType(
+  t: ExecutorActionType
+): t is
+  | ({ type: 'CodingAgentInitialRequest' } & CodingAgentInitialRequest)
+  | ({ type: 'CodingAgentFollowUpRequest' } & CodingAgentFollowUpRequest) {
+  return (
+    t.type === 'CodingAgentInitialRequest' ||
+    t.type === 'CodingAgentFollowUpRequest'
+  );
+}
+
+/**
+ * Reusable hook to retry a process given its executionProcessId and a new prompt.
+ * Handles:
+ *  - Preventing retry while anything is running (or that process is already running)
+ *  - Optional worktree reset (via modal)
+ *  - Variant extraction for coding-agent processes
+ *  - Refetching attempt + branch data after replace
+ */
+export function useProcessRetry(attempt: TaskAttempt | undefined) {
+  const attemptId = attempt?.id;
+
+  // Fetch attempt + branch state the same way your component did
+  const { attemptData, refetch: refetchAttempt } =
+    useAttemptExecution(attemptId);
+  const { data: branchStatus, refetch: refetchBranch } =
+    useBranchStatus(attemptId);
+
+  const [busy, setBusy] = useState(false);
+
+  // Any process running at all?
+  const anyRunning = useMemo(
+    () => (attemptData.processes || []).some((p) => p.status === 'running'),
+    [attemptData.processes?.map((p) => p.status).join(',')]
+  );
+
+  // Convenience lookups
+  const getProcessById = useCallback(
+    (pid: string): ExecutionProcess | undefined =>
+      (attemptData.processes || []).find((p) => p.id === pid),
+    [attemptData.processes]
+  );
+
+  /**
+   * Returns whether a process is currently allowed to retry, and why not.
+   * Useful if you want to gray out buttons in any component.
+   */
+  const getRetryDisabledState = useCallback(
+    (pid: string) => {
+      const proc = getProcessById(pid);
+      const isRunningProc = proc?.status === 'running';
+      const disabled = busy || anyRunning || isRunningProc;
+      let reason: string | undefined;
+      if (isRunningProc) reason = 'Finish or stop this run to retry.';
+      else if (anyRunning) reason = 'Cannot retry while a process is running.';
+      else if (busy) reason = 'Retry in progress.';
+      return { disabled, reason };
+    },
+    [busy, anyRunning, getProcessById]
+  );
+
+  /**
+   * Primary entrypoint: retry a process with a new prompt.
+   */
+  const retryProcess = useCallback(
+    async (executionProcessId: string, newPrompt: string) => {
+      if (!attemptId) return;
+
+      const proc = getProcessById(executionProcessId);
+      if (!proc) return;
+
+      // Respect current disabled state
+      const { disabled } = getRetryDisabledState(executionProcessId);
+      if (disabled) return;
+
+      type WithBefore = { before_head_commit?: string | null };
+      const before =
+        (proc as WithBefore | undefined)?.before_head_commit || null;
+
+      // Try to gather comparison info (best-effort)
+      let targetSubject: string | null = null;
+      let commitsToReset: number | null = null;
+      let isLinear: boolean | null = null;
+
+      if (before) {
+        try {
+          const { commitsApi } = await import('@/lib/api');
+          const info = await commitsApi.getInfo(attemptId, before);
+          targetSubject = info.subject;
+          const cmp = await commitsApi.compareToHead(attemptId, before);
+          commitsToReset = cmp.is_linear ? cmp.ahead_from_head : null;
+          isLinear = cmp.is_linear;
+        } catch {
+          // ignore best-effort enrichments
+        }
+      }
+
+      const head = branchStatus?.head_oid || null;
+      const dirty = !!branchStatus?.has_uncommitted_changes;
+      const needReset = !!(before && (before !== head || dirty));
+      const canGitReset = needReset && !dirty;
+
+      // Compute “later processes” context for the dialog
+      const procs = (attemptData.processes || []).filter(
+        (p) => !p.dropped && shouldShowInLogs(p.run_reason)
+      );
+      const idx = procs.findIndex((p) => p.id === executionProcessId);
+      const later = idx >= 0 ? procs.slice(idx + 1) : [];
+      const laterCount = later.length;
+      const laterCoding = later.filter((p) =>
+        isCodingAgent(p.run_reason)
+      ).length;
+      const laterSetup = later.filter(
+        (p) => p.run_reason === PROCESS_RUN_REASONS.SETUP_SCRIPT
+      ).length;
+      const laterCleanup = later.filter(
+        (p) => p.run_reason === PROCESS_RUN_REASONS.CLEANUP_SCRIPT
+      ).length;
+
+      // Ask user for confirmation / reset options
+      let modalResult:
+        | {
+            action: 'confirmed' | 'canceled';
+            performGitReset?: boolean;
+            forceWhenDirty?: boolean;
+          }
+        | undefined;
+
+      try {
+        modalResult = await showModal<
+          typeof modalResult extends infer T
+            ? T extends object
+              ? T
+              : never
+            : never
+        >('restore-logs', {
+          targetSha: before,
+          targetSubject,
+          commitsToReset,
+          isLinear,
+          laterCount,
+          laterCoding,
+          laterSetup,
+          laterCleanup,
+          needGitReset: needReset,
+          canGitReset,
+          hasRisk: dirty,
+          uncommittedCount: branchStatus?.uncommitted_count ?? 0,
+          untrackedCount: branchStatus?.untracked_count ?? 0,
+          // Defaults
+          initialWorktreeResetOn: true,
+          initialForceReset: false,
+        });
+      } catch {
+        // user closed dialog
+        return;
+      }
+
+      if (!modalResult || modalResult.action !== 'confirmed') return;
+
+      let variant: string | null = null;
+
+      const typ = proc?.executor_action?.typ; // type: ExecutorActionType
+
+      if (typ && isCodingAgentActionType(typ)) {
+        // executor_profile_id is ExecutorProfileId -> has `variant: string | null`
+        variant = typ.executor_profile_id.variant;
+      }
+
+      // Perform the replacement
+      try {
+        setBusy(true);
+        const { attemptsApi } = await import('@/lib/api');
+        await attemptsApi.replaceProcess(attemptId, {
+          process_id: executionProcessId,
+          prompt: newPrompt,
+          variant,
+          perform_git_reset: modalResult.performGitReset ?? true,
+          force_when_dirty: modalResult.forceWhenDirty ?? false,
+        });
+
+        // Refresh local caches
+        await refetchAttempt();
+        await refetchBranch();
+      } finally {
+        setBusy(false);
+      }
+    },
+    [
+      attemptId,
+      attemptData.processes,
+      branchStatus?.head_oid,
+      branchStatus?.has_uncommitted_changes,
+      branchStatus?.uncommitted_count,
+      branchStatus?.untracked_count,
+      getProcessById,
+      getRetryDisabledState,
+      refetchAttempt,
+      refetchBranch,
+    ]
+  );
+
+  return {
+    retryProcess,
+    busy,
+    anyRunning,
+    /** Helpful for buttons/tooltips */
+    getRetryDisabledState,
+  };
+}
+
+export type UseProcessRetryReturn = ReturnType<typeof useProcessRetry>;

frontend/src/hooks/useProjectTasks.ts

@@ -19,12 +19,8 @@ interface UseProjectTasksResult {
  * Server sends initial snapshot: replace /tasks with an object keyed by id.
  * Live updates arrive at /tasks/<id> via add/replace/remove operations.
  */
-export const useProjectTasks = (
-  projectId: string | undefined
-): UseProjectTasksResult => {
-  const endpoint = projectId
-    ? `/api/tasks/stream?project_id=${encodeURIComponent(projectId)}`
-    : undefined;
+export const useProjectTasks = (projectId: string): UseProjectTasksResult => {
+  const endpoint = `/api/tasks/stream?project_id=${encodeURIComponent(projectId)}`;
 
   const initialData = useCallback((): TasksState => ({ tasks: {} }), []);
 

frontend/src/pages/project-tasks.tsx

@@ -102,7 +102,7 @@ export function ProjectTasks() {
     tasksById,
     isLoading,
     error: streamError,
-  } = useProjectTasks(projectId);
+  } = useProjectTasks(projectId || '');
 
   // Sync selectedTask with URL params and live task updates
   useEffect(() => {

frontend/src/utils/streamSseJsonPatchEntries.ts

@@ -0,0 +1,126 @@
+// sseJsonPatchEntries.ts
+import { applyPatch, type Operation } from 'rfc6902';
+
+type PatchContainer<E = unknown> = { entries: E[] };
+
+export interface StreamOptions<E = unknown> {
+  initial?: PatchContainer<E>;
+  eventSourceInit?: EventSourceInit;
+  /** called after each successful patch application */
+  onEntries?: (entries: E[]) => void;
+  onConnect?: () => void;
+  onError?: (err: unknown) => void;
+  /** called once when a "finished" event is received */
+  onFinished?: (entries: E[]) => void;
+}
+
+/**
+ * Connect to an SSE endpoint that emits:
+ *   event: json_patch
+ *   data: [ { op, path, value? }, ... ]
+ *
+ * Maintains an in-memory { entries: [] } snapshot and returns a controller.
+ */
+export function streamSseJsonPatchEntries<E = unknown>(
+  url: string,
+  opts: StreamOptions<E> = {}
+) {
+  let connected = false;
+  let snapshot: PatchContainer<E> = structuredClone(
+    opts.initial ?? ({ entries: [] } as PatchContainer<E>)
+  );
+
+  const subscribers = new Set<(entries: E[]) => void>();
+  if (opts.onEntries) subscribers.add(opts.onEntries);
+
+  const es = new EventSource(url, opts.eventSourceInit);
+
+  const notify = () => {
+    for (const cb of subscribers) {
+      try {
+        cb(snapshot.entries);
+      } catch {
+        /* swallow subscriber errors */
+      }
+    }
+  };
+
+  const handlePatchEvent = (e: MessageEvent<string>) => {
+    try {
+      const raw = JSON.parse(e.data) as Operation[];
+      const ops = dedupeOps(raw);
+
+      // Apply to a working copy (applyPatch mutates)
+      const next = structuredClone(snapshot);
+      applyPatch(next as unknown as object, ops);
+
+      snapshot = next;
+      notify();
+    } catch (err) {
+      opts.onError?.(err);
+    }
+  };
+
+  es.addEventListener('open', () => {
+    connected = true;
+    opts.onConnect?.();
+  });
+
+  // The server uses a named event: "json_patch"
+  es.addEventListener('json_patch', handlePatchEvent);
+
+  es.addEventListener('finished', () => {
+    opts.onFinished?.(snapshot.entries);
+    es.close();
+  });
+
+  es.addEventListener('error', (err) => {
+    connected = false; // EventSource will auto-retry; this just reflects current state
+    opts.onError?.(err);
+  });
+
+  return {
+    /** Current entries array (immutable snapshot) */
+    getEntries(): E[] {
+      return snapshot.entries;
+    },
+    /** Full { entries } snapshot */
+    getSnapshot(): PatchContainer<E> {
+      return snapshot;
+    },
+    /** Best-effort connection state (EventSource will auto-reconnect) */
+    isConnected(): boolean {
+      return connected;
+    },
+    /** Subscribe to updates; returns an unsubscribe function */
+    onChange(cb: (entries: E[]) => void): () => void {
+      subscribers.add(cb);
+      // push current state immediately
+      cb(snapshot.entries);
+      return () => subscribers.delete(cb);
+    },
+    /** Close the stream */
+    close(): void {
+      es.close();
+      subscribers.clear();
+      connected = false;
+    },
+  };
+}
+
+/**
+ * Dedupe multiple ops that touch the same path within a single event.
+ * Last write for a path wins, while preserving the overall left-to-right
+ * order of the *kept* final operations.
+ *
+ * Example:
+ *   add /entries/4, replace /entries/4  -> keep only the final replace
+ */
+function dedupeOps(ops: Operation[]): Operation[] {
+  const lastIndexByPath = new Map<string, number>();
+  ops.forEach((op, i) => lastIndexByPath.set(op.path, i));
+
+  // Keep only the last op for each path, in ascending order of their final index
+  const keptIndices = [...lastIndexByPath.values()].sort((a, b) => a - b);
+  return keptIndices.map((i) => ops[i]!);
+}