fix(executor): skip content persistence when stream consumer exits early

Previously, if the onStream consumer caught an internal error without
re-throwing, the block-executor would treat the shortened accumulator
as the complete response, persist a truncated string to memory via
appendToMemory, and set it as executionOutput.content.

Track whether the source ReadableStream actually closed (done=true) in
the pull handler. If onStream returns before the source drains, skip
content persistence and log a warning — the old tee()-based flow was
immune to this because the executor branch drained independently of
the client branch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: TheodoreSpeaks

apps/sim/executor/execution/block-executor.ts

@@ -618,6 +618,7 @@ export class BlockExecutor {
     const decoder = new TextDecoder()
     const accumulated: string[] = []
     let drainError: unknown
+    let sourceFullyDrained = false
 
     const clientSource = new ReadableStream<Uint8Array>({
       async pull(controller) {
@@ -626,6 +627,7 @@ export class BlockExecutor {
           if (done) {
             const tail = decoder.decode()
             if (tail) accumulated.push(tail)
+            sourceFullyDrained = true
             controller.close()
             return
           }
@@ -669,6 +671,17 @@ export class BlockExecutor {
       return
     }
 
+    // If the onStream consumer exited before the source drained (e.g. it caught
+    // an internal error and returned normally), `accumulated` holds a truncated
+    // response. Persisting that to memory or setting it as the block output
+    // would corrupt downstream state — skip and log instead.
+    if (!sourceFullyDrained) {
+      this.execLogger.warn('Stream consumer exited before source drained; skipping content persistence', {
+        blockId,
+      })
+      return
+    }
+
     const fullContent = accumulated.join('')
     if (fullContent) {
       const executionOutput = streamingExec.execution?.output