fix execution stream attach, cleanup comments

Author: icecrasher321

apps/sim/app/workspace/[workspaceId]/home/components/message-content/message-content.tsx

@@ -182,9 +182,6 @@ function parseBlocks(blocks: ContentBlock[]): MessageSegment[] {
     if (existing) return { group: existing, created: false }
     const group: AgentGroupSegment = {
       type: 'agent_group',
-      // Suffix with segments.length so a later flushLanes / explicit delete
-      // followed by re-ensure for the same key produces a fresh React key
-      // instead of colliding with the stranded prior segment.
       id: `agent-${key}-${segments.length}`,
       agentName: name,
       agentLabel: resolveAgentLabel(name),
@@ -211,11 +208,6 @@ function parseBlocks(blocks: ContentBlock[]): MessageSegment[] {
   }
 
   const flushLanes = () => {
-    // Finalize state on the already-pushed segments before we drop them from
-    // the map. Without this, a thinking/main text/options/stopped block landing
-    // between a subagent's start and its end strands the lane with stale
-    // isOpen/isDelegating flags — the later subagent_end iterates groupsByKey
-    // and can't find it to close.
     for (const g of groupsByKey.values()) {
       g.isOpen = false
       g.isDelegating = false
@@ -300,13 +292,10 @@ function parseBlocks(blocks: ContentBlock[]): MessageSegment[] {
           }
         }
       }
-      // Subagent lanes break the mothership lane: any mothership tool that
-      // arrives after this subagent finishes should render below the lane,
-      // not get back-filled into the mothership group sitting above it.
       groupsByKey.delete(groupKey('mothership', undefined))
       const { group: g } = ensureGroup(key, block.parentToolCallId)
       if (inheritedDelegation) g.isDelegating = true
-      g.isOpen = block.endedAt === undefined
+      g.isOpen = true
       activeGroupKey = resolveGroupKey(key, block.parentToolCallId)
       continue
     }
@@ -331,9 +320,6 @@ function parseBlocks(blocks: ContentBlock[]): MessageSegment[] {
       if (tc.calledBy) {
         const { group: g, created } = ensureGroup(tc.calledBy, block.parentToolCallId)
         g.isDelegating = false
-        // Only mark the lane open when we just created it. Late tool_calls
-        // arriving after a subagent_end (out-of-order persistence, replay,
-        // partial state hand-off) must NOT reopen a closed lane.
         if (created && block.parentToolCallId) g.isOpen = true
         g.items.push({ type: 'tool', data: tool })
         activeGroupKey = resolveGroupKey(tc.calledBy, block.parentToolCallId)
@@ -363,11 +349,6 @@ function parseBlocks(blocks: ContentBlock[]): MessageSegment[] {
           activeGroupKey = null
         }
       } else {
-        // Unparented end blocks come from legacy/partial persisted streams
-        // that predate parentToolCallId. Close lanes that are themselves
-        // legacy-keyed without touching parented lanes — this avoids
-        // force-closing legitimate parallel parented lanes while still
-        // releasing stranded legacy lanes in mixed/migration streams.
         for (const [key, g] of groupsByKey) {
           if (key.endsWith(':legacy') && g.agentName !== 'mothership') {
             g.isOpen = false

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -131,6 +131,7 @@ import type {
   MothershipResource,
   MothershipResourceType,
   QueuedMessage,
+  ToolCallInfo,
 } from '../types'
 import { ToolCallStatus } from '../types'
 
@@ -1217,7 +1218,7 @@ export function useChat(
       reader: ReadableStreamDefaultReader<Uint8Array>,
       assistantId: string,
       expectedGen?: number,
-      options?: { preserveExistingState?: boolean }
+      options?: { preserveExistingState?: boolean; suppressWorkflowToolStarts?: boolean }
     ) => Promise<{ sawStreamError: boolean; sawComplete: boolean }>
   >(async () => ({ sawStreamError: false, sawComplete: false }))
   const attachToExistingStreamRef = useRef<
@@ -1457,8 +1458,50 @@ export function useChat(
         return
       }
       if (handledClientWorkflowToolIdsRef.current.has(toolCallId)) {
+        // #region agent log
+        fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+          method: 'POST',
+          headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+          body: JSON.stringify({
+            sessionId: '21c369',
+            location: 'use-chat.ts:startClientWorkflowTool',
+            message: 'SKIPPED:handled',
+            data: { toolCallId },
+            timestamp: Date.now(),
+          }),
+        }).catch(() => {})
+        // #endregion
+        return
+      }
+      if (recoveringClientWorkflowToolIdsRef.current.has(toolCallId)) {
+        // #region agent log
+        fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+          method: 'POST',
+          headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+          body: JSON.stringify({
+            sessionId: '21c369',
+            location: 'use-chat.ts:startClientWorkflowTool',
+            message: 'SKIPPED:recovering',
+            data: { toolCallId },
+            timestamp: Date.now(),
+          }),
+        }).catch(() => {})
+        // #endregion
         return
       }
+      // #region agent log
+      fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+        body: JSON.stringify({
+          sessionId: '21c369',
+          location: 'use-chat.ts:startClientWorkflowTool',
+          message: 'FIRING',
+          data: { toolCallId, toolName },
+          timestamp: Date.now(),
+        }),
+      }).catch(() => {})
+      // #endregion
       handledClientWorkflowToolIdsRef.current.add(toolCallId)
 
       ensureWorkflowToolResource(toolArgs)
@@ -1469,41 +1512,68 @@ export function useChat(
 
   const recoverPendingClientWorkflowTools = useCallback(
     async (nextMessages: ChatMessage[]) => {
+      const pending: ToolCallInfo[] = []
+
       for (const message of nextMessages) {
         for (const block of message.contentBlocks ?? []) {
           const toolCall = block.toolCall
-          if (!toolCall || !isWorkflowToolName(toolCall.name)) {
-            continue
-          }
-          if (toolCall.status !== 'executing') {
-            continue
-          }
-
+          if (!toolCall || !isWorkflowToolName(toolCall.name)) continue
+          if (toolCall.status !== 'executing') continue
           if (
             handledClientWorkflowToolIdsRef.current.has(toolCall.id) ||
             recoveringClientWorkflowToolIdsRef.current.has(toolCall.id)
           ) {
             continue
           }
-
           recoveringClientWorkflowToolIdsRef.current.add(toolCall.id)
+          pending.push(toolCall)
+        }
+      }
 
-          try {
-            const toolArgs = toolCall.params ?? {}
-            const targetWorkflowId = ensureWorkflowToolResource(toolArgs)
-
-            if (targetWorkflowId) {
-              const rebound = await bindRunToolToExecution(toolCall.id, targetWorkflowId)
-              if (rebound) {
-                handledClientWorkflowToolIdsRef.current.add(toolCall.id)
-                continue
-              }
-            }
+      // #region agent log
+      fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+        body: JSON.stringify({
+          sessionId: '21c369',
+          location: 'use-chat.ts:recoverPending',
+          message: 'scan complete',
+          data: { pendingCount: pending.length, pendingIds: pending.map((t) => t.id) },
+          timestamp: Date.now(),
+        }),
+      }).catch(() => {})
+      // #endregion
 
-            startClientWorkflowTool(toolCall.id, toolCall.name, toolArgs)
-          } finally {
-            recoveringClientWorkflowToolIdsRef.current.delete(toolCall.id)
+      for (const toolCall of pending) {
+        try {
+          const toolArgs = toolCall.params ?? {}
+          const targetWorkflowId = ensureWorkflowToolResource(toolArgs)
+
+          if (targetWorkflowId) {
+            const rebound = await bindRunToolToExecution(toolCall.id, targetWorkflowId)
+            // #region agent log
+            fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+              method: 'POST',
+              headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+              body: JSON.stringify({
+                sessionId: '21c369',
+                location: 'use-chat.ts:recoverPending',
+                message: 'bind result',
+                data: { toolCallId: toolCall.id, targetWorkflowId, rebound },
+                timestamp: Date.now(),
+              }),
+            }).catch(() => {})
+            // #endregion
+            if (rebound) {
+              handledClientWorkflowToolIdsRef.current.add(toolCall.id)
+              continue
+            }
           }
+
+          recoveringClientWorkflowToolIdsRef.current.delete(toolCall.id)
+          startClientWorkflowTool(toolCall.id, toolCall.name, toolArgs)
+        } finally {
+          recoveringClientWorkflowToolIdsRef.current.delete(toolCall.id)
         }
       }
     },
@@ -1703,7 +1773,7 @@ export function useChat(
       reader: ReadableStreamDefaultReader<Uint8Array>,
       assistantId: string,
       expectedGen?: number,
-      options?: { preserveExistingState?: boolean }
+      options?: { preserveExistingState?: boolean; suppressWorkflowToolStarts?: boolean }
     ) => {
       const decoder = new TextDecoder()
       streamReaderRef.current = reader
@@ -1818,6 +1888,19 @@ export function useChat(
         return activeSubagent
       }
 
+      const resolveParentForSubagentBlock = (
+        subagent: string | undefined,
+        scopedParent: string | undefined
+      ): string | undefined => {
+        if (!subagent) return undefined
+        if (scopedParent) return scopedParent
+        if (activeSubagent === subagent) return activeSubagentParentToolCallId
+        for (const [parent, name] of subagentByParentToolCallId) {
+          if (name === subagent) return parent
+        }
+        return undefined
+      }
+
       const appendInlineErrorTag = (
         tag: string,
         subagentName?: string,
@@ -2038,9 +2121,10 @@ export function useChat(
               if (chunk) {
                 const eventTs = typeof parsed.ts === 'string' ? parsed.ts : undefined
                 if (parsed.payload.channel === MothershipStreamV1TextChannel.thinking) {
-                  const scopedParentForBlock = scopedSubagent
-                    ? (scopedParentToolCallId ?? activeSubagentParentToolCallId)
-                    : undefined
+                  const scopedParentForBlock = resolveParentForSubagentBlock(
+                    scopedSubagent,
+                    scopedParentToolCallId
+                  )
                   const tb = ensureThinkingBlock(scopedSubagent, scopedParentForBlock, eventTs)
                   tb.content = (tb.content ?? '') + chunk
                   flushText()
@@ -2052,9 +2136,10 @@ export function useChat(
                   lastContentSource !== contentSource &&
                   runningText.length > 0 &&
                   !runningText.endsWith('\n')
-                const scopedParentForBlock = scopedSubagent
-                  ? (scopedParentToolCallId ?? activeSubagentParentToolCallId)
-                  : undefined
+                const scopedParentForBlock = resolveParentForSubagentBlock(
+                  scopedSubagent,
+                  scopedParentToolCallId
+                )
                 const tb = ensureTextBlock(scopedSubagent, scopedParentForBlock, eventTs)
                 const normalizedChunk = needsBoundaryNewline ? `\n${chunk}` : chunk
                 tb.content = (tb.content ?? '') + normalizedChunk
@@ -2391,12 +2476,17 @@ export function useChat(
                 }
               }
 
-              if (!toolMap.has(id)) {
+              const existingToolCall = toolMap.has(id)
+                ? blocks[toolMap.get(id)!]?.toolCall
+                : undefined
+              const isNewToolCall = !existingToolCall
+              if (isNewToolCall) {
                 stampBlockEnd(blocks[blocks.length - 1])
                 toolMap.set(id, blocks.length)
-                const parentToolCallIdForBlock = scopedSubagent
-                  ? (scopedParentToolCallId ?? activeSubagentParentToolCallId)
-                  : undefined
+                const parentToolCallIdForBlock = resolveParentForSubagentBlock(
+                  scopedSubagent,
+                  scopedParentToolCallId
+                )
                 blocks.push({
                   type: 'tool_call',
                   toolCall: {
@@ -2427,7 +2517,28 @@ export function useChat(
               flush()
 
               if (isWorkflowToolName(name) && !isPartial) {
-                startClientWorkflowTool(id, name, args ?? {})
+                const shouldStartWorkflowTool =
+                  !options?.suppressWorkflowToolStarts &&
+                  (isNewToolCall ||
+                    (existingToolCall?.status === ToolCallStatus.executing &&
+                      !existingToolCall.result))
+                if (shouldStartWorkflowTool) {
+                  startClientWorkflowTool(id, name, args ?? {})
+                } else {
+                  // #region agent log
+                  fetch('http://127.0.0.1:1025/ingest/85045d0a-92f7-4ee2-9de1-e2f99930c6bc', {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json', 'X-Debug-Session-Id': '21c369' },
+                    body: JSON.stringify({
+                      sessionId: '21c369',
+                      location: 'use-chat.ts:processSSE',
+                      message: 'SKIPPED:replay',
+                      data: { toolCallId: id, toolName: name },
+                      timestamp: Date.now(),
+                    }),
+                  }).catch(() => {})
+                  // #endregion
+                }
               }
               break
             }
@@ -2627,9 +2738,7 @@ export function useChat(
               appendInlineErrorTag(
                 buildInlineErrorTag(parsed.payload),
                 scopedSubagent,
-                scopedSubagent
-                  ? (scopedParentToolCallId ?? activeSubagentParentToolCallId)
-                  : undefined,
+                resolveParentForSubagentBlock(scopedSubagent, scopedParentToolCallId),
                 typeof parsed.ts === 'string' ? parsed.ts : undefined
               )
               break
@@ -2734,6 +2843,7 @@ export function useChat(
       let latestCursor = afterCursor
       let seedEvents = opts.initialBatch?.events ?? []
       let streamStatus = opts.initialBatch?.status ?? 'unknown'
+      let suppressSeedWorkflowStarts = seedEvents.length > 0
 
       const isStaleReconnect = () =>
         streamGenRef.current !== expectedGen || abortControllerRef.current?.signal.aborted === true
@@ -2752,11 +2862,15 @@ export function useChat(
               buildReplayStream(seedEvents).getReader(),
               assistantId,
               expectedGen,
-              { preserveExistingState: true }
+              {
+                preserveExistingState: true,
+                suppressWorkflowToolStarts: suppressSeedWorkflowStarts,
+              }
             )
             latestCursor = String(seedEvents[seedEvents.length - 1]?.eventId ?? latestCursor)
             lastCursorRef.current = latestCursor
             seedEvents = []
+            suppressSeedWorkflowStarts = false
 
             if (replayResult.sawStreamError) {
               return { error: true, aborted: false }
@@ -2800,7 +2914,7 @@ export function useChat(
             sseRes.body.getReader(),
             assistantId,
             expectedGen,
-            { preserveExistingState: true }
+            { preserveExistingState: true, suppressWorkflowToolStarts: true }
           )
 
           if (liveResult.sawStreamError) {