diff --git a/agent/harness/toolautocall/autocall.go b/agent/harness/toolautocall/autocall.go
index ed148e83..8baf4abe 100644
--- a/agent/harness/toolautocall/autocall.go
+++ b/agent/harness/toolautocall/autocall.go
@@ -194,19 +194,29 @@ func (f *autocall) Run(next agent.RunFunc, ctx context.Context, messages []*mess
 	}
 	return func(yield func(*agent.ResponseUpdate, error) bool) {
 		var finalConversationID *string
-		response, err := agent.ResponseStream(f.run(next, ctx, messages, opts, &finalConversationID)).Collect()
-		if err != nil {
-			yield(nil, err)
-			return
+		var response agent.Response
+		var runErr error
+		for update, err := range f.run(next, ctx, messages, opts, &finalConversationID) {
+			response.Update(update)
+			if err != nil {
+				runErr = err
+				break
+			}
 		}
+		response.Coalesce()
 		// A completed non-streaming run uses the final provider response's ID,
 		// even when it is nil. Earlier tool rounds still contribute messages.
-		response.ConversationID = finalConversationID
+		if runErr == nil {
+			response.ConversationID = finalConversationID
+		}
 		for _, update := range response.ToUpdates() {
 			if !yield(update, nil) {
 				return
 			}
 		}
+		if runErr != nil {
+			yield(nil, runErr)
+		}
 	}
 }
 
@@ -308,10 +318,6 @@ func (f *autocall) run(next agent.RunFunc, ctx context.Context, messages []*mess
 			// Invoke approved approval responses, which generates some additional FRC wrapped in ChatMessage.
 			var newMsg *message.Message
 			newMsg, errCount, err = f.invokeApprovedToolApprovalResponses(ctx, notInvokedMsgs, tools, 0)
-			if err != nil {
-				yield(nil, err)
-				return
-			}
 			if newMsg != nil {
 				opts = updateOptionsForNextIteration(opts)
 				messages = slices.Insert(messages, approvedResultInsertIdx, newMsg)
@@ -324,6 +330,10 @@ func (f *autocall) run(next agent.RunFunc, ctx context.Context, messages []*mess
 					return
 				}
 			}
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 		}
 		// At this point, we've fully handled all approval responses that were part of the original messages,
 		// and we can now enter the main function calling loop.
@@ -439,10 +449,6 @@ func (f *autocall) run(next agent.RunFunc, ctx context.Context, messages []*mess
 			var newMsg *message.Message
 			var err error
 			newMsg, errCount, err = f.processFunctionCalls(ctx, tools, functionCallContents, errCount)
-			if err != nil {
-				yield(nil, err)
-				return
-			}
 
 			// Stream any generated function results. This mirrors what's done for ResponseAsync, where the returned messages
 			// includes all activities, including generated function results.
@@ -451,6 +457,10 @@ func (f *autocall) run(next agent.RunFunc, ctx context.Context, messages []*mess
 			if !yield(resultUpdate, nil) {
 				return
 			}
+			if err != nil {
+				yield(nil, err)
+				return
+			}
 
 			// Preserve every response message and content item, not just the
 			// invocable function calls. This includes images, hosted tool results,
@@ -1145,12 +1155,7 @@ func (f *autocall) invokeApprovedToolApprovalResponses(ctx context.Context, appr
 	if len(funcCalls) == 0 {
 		return nil, errCount, nil
 	}
-	newMsg, errCount, err := f.processFunctionCalls(ctx, tools, funcCalls, errCount)
-	if err != nil {
-		return nil, errCount, err
-	}
-
-	return newMsg, errCount, nil
+	return f.processFunctionCalls(ctx, tools, funcCalls, errCount)
 }
 
 type functionInvocationStatus int
@@ -1159,6 +1164,7 @@ const (
 	functionInvocationStatusRanToCompletion functionInvocationStatus = iota
 	functionInvocationStatusNotFound
 	functionInvocationStatusException
+	functionInvocationStatusNotInvoked
 )
 
 type functionInvocationResult struct {
@@ -1177,6 +1183,7 @@ func (f *autocall) processFunctionCalls(ctx context.Context, tools map[string]to
 	captureCurrentIterationErrors := errCount < f.maximumConsecutiveErrorsPerRequest
 	// Process all functions. If there's more than one and concurrent invocation is enabled, do so in parallel.
 	results := make([]functionInvocationResult, 0, len(funcCalls))
+	var runErr error
 	if len(funcCalls) > 1 && f.allowConcurrentInvocations {
 		// Rather than awaiting each function before invoking the next, invoke all of them
 		// and then await all of them. We avoid forcibly introducing parallelism via Task.Run,
@@ -1192,6 +1199,7 @@ func (f *autocall) processFunctionCalls(ctx context.Context, tools map[string]to
 			}()
 		}
 		wg.Wait()
+		results = parallelResults
 		if err := ctx.Err(); err != nil {
 			var toolErrors []error
 			for _, result := range parallelResults {
@@ -1200,29 +1208,40 @@ func (f *autocall) processFunctionCalls(ctx context.Context, tools map[string]to
 				}
 			}
 			if len(toolErrors) == 1 {
-				return nil, errCount, toolErrors[0]
-			}
-			if len(toolErrors) > 1 {
-				return nil, errCount, errors.Join(toolErrors...)
+				runErr = toolErrors[0]
+			} else if len(toolErrors) > 1 {
+				runErr = errors.Join(toolErrors...)
+			} else {
+				runErr = err
 			}
-			return nil, errCount, err
 		}
-		results = parallelResults
 	} else {
 		// Invoke each function serially.
 		for _, fc := range funcCalls {
 			result := f.processFunctionCall(ctx, tools, fc)
+			results = append(results, result)
 			// Request cancellation must bypass the recoverable tool-error path.
 			if err := ctx.Err(); err != nil {
 				if result.err != nil {
-					return nil, errCount, result.err
+					runErr = result.err
+				} else {
+					runErr = err
 				}
-				return nil, errCount, err
+				break
 			}
 			if !captureCurrentIterationErrors && result.status == functionInvocationStatusException {
-				return nil, errCount, result.err
+				runErr = result.err
+				break
 			}
-			results = append(results, result)
+		}
+	}
+	if len(results) < len(funcCalls) {
+		skippedErr := runErr
+		if err := ctx.Err(); err != nil {
+			skippedErr = err
+		}
+		for _, fc := range funcCalls[len(results):] {
+			results = append(results, functionInvocationResult{status: functionInvocationStatusNotInvoked, call: fc, err: skippedErr})
 		}
 	}
 	for _, result := range results {
@@ -1230,13 +1249,11 @@ func (f *autocall) processFunctionCalls(ctx context.Context, tools map[string]to
 	}
 
 	newMsg := f.createResponseMessage(results)
-	var err error
-	errCount, err = f.updateConsecutiveErrorCountOrThrow(ctx, newMsg, errCount)
-	if err != nil {
-		return nil, errCount, err
+	if runErr == nil {
+		errCount, runErr = f.updateConsecutiveErrorCountOrThrow(ctx, newMsg, errCount)
 	}
 
-	return newMsg, errCount, nil
+	return newMsg, errCount, runErr
 }
 
 func (f *autocall) updateConsecutiveErrorCountOrThrow(ctx context.Context, added *message.Message, errCount int) (int, error) {
@@ -1268,7 +1285,7 @@ func (f *autocall) updateConsecutiveErrorCountOrThrow(ctx context.Context, added
 
 func (f *autocall) processFunctionCall(ctx context.Context, tools map[string]tool.SchemaTool, funcCall *message.FunctionCallContent) functionInvocationResult {
 	if err := ctx.Err(); err != nil {
-		return functionInvocationResult{status: functionInvocationStatusException, call: funcCall, err: err}
+		return functionInvocationResult{status: functionInvocationStatusNotInvoked, call: funcCall, err: err}
 	}
 	declaration, ok := tools[funcCall.Name]
 	if !ok {
@@ -1356,6 +1373,11 @@ func (f *autocall) createFunctionResultContent(result functionInvocationResult)
 		if f.includeDetailedErrors && result.err != nil {
 			functionResult = fmt.Sprintf("%s Exception: %v", functionResult, result.err)
 		}
+	case functionInvocationStatusNotInvoked:
+		functionResult = "Error: Function was not invoked."
+		if f.includeDetailedErrors && result.err != nil {
+			functionResult = fmt.Sprintf("%s Exception: %v", functionResult, result.err)
+		}
 	}
 
 	return &message.FunctionResultContent{
diff --git a/agent/harness/toolautocall/autocall_test.go b/agent/harness/toolautocall/autocall_test.go
index 5e650012..618754cf 100644
--- a/agent/harness/toolautocall/autocall_test.go
+++ b/agent/harness/toolautocall/autocall_test.go
@@ -1835,12 +1835,20 @@ func TestFunctionInvoking_SerialDoesNotCaptureErrorsAfterMaximumConsecutiveError
 	}
 
 	var streamErr error
-	for _, err := range toolautocall.New(toolautocall.Config{MaximumConsecutiveErrorsPerRequest: new(0)}).Run(
+	var results []*message.FunctionResultContent
+	for update, err := range toolautocall.New(toolautocall.Config{MaximumConsecutiveErrorsPerRequest: new(0)}).Run(
 		runner.Run,
 		t.Context(),
 		[]*message.Message{message.NewText("hello")},
 		opts...,
 	) {
+		if update != nil {
+			for _, content := range update.Contents {
+				if result, ok := content.(*message.FunctionResultContent); ok {
+					results = append(results, result)
+				}
+			}
+		}
 		if err != nil {
 			streamErr = err
 			break
@@ -1856,6 +1864,10 @@ func TestFunctionInvoking_SerialDoesNotCaptureErrorsAfterMaximumConsecutiveError
 	if got := secondInvoked.Load(); got != 0 {
 		t.Fatalf("second tool invoked %d times, want 0", got)
 	}
+	if len(results) != 2 || results[0].CallID != "callId1" || results[0].Error != firstErr ||
+		results[1].CallID != "callId2" || results[1].Error != firstErr || results[1].Result != "Error: Function was not invoked." {
+		t.Fatalf("results = %+v, want the original failure and an unexecuted-call result", results)
+	}
 }
 
 // TestFunctionInvoking_KeepsFunctionCallingContent tests that function call/result content is preserved
@@ -2381,6 +2393,265 @@ func messageTexts(messages []*message.Message) []string {
 	return texts
 }
 
+func TestFunctionInvoking_CancellationPreservesCompletedResults(t *testing.T) {
+	toolFailure := errors.New("original tool failure")
+	for _, outcome := range []string{"success", "error", "panic error", "panic value"} {
+		for _, stream := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/stream=%t", outcome, stream), func(t *testing.T) {
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				var effects int
+				effect := functool.MustNew(functool.Config{Name: "effect"}, func(callCtx context.Context, _ struct{}) (string, error) {
+					if callCtx.Done() != ctx.Done() {
+						t.Error("tool lost the original cancellation context")
+					}
+					effects++
+					if effects == 3 {
+						cancel()
+						switch outcome {
+						case "error":
+							return "", toolFailure
+						case "panic error":
+							panic(toolFailure)
+						case "panic value":
+							panic("original tool panic")
+						}
+					}
+					return fmt.Sprintf("completed %d", effects), nil
+				})
+				var providerCalls int
+				provider := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
+					return func(yield func(*agent.ResponseUpdate, error) bool) {
+						providerCalls++
+						ids := []string{"earlier"}
+						switch providerCalls {
+						case 2:
+							ids = []string{"first", "interrupted", "skipped"}
+						case 3:
+							yield(&agent.ResponseUpdate{Role: message.RoleAssistant, Contents: message.Contents{&message.TextContent{Text: "done"}}}, nil)
+							return
+						}
+						contents := make(message.Contents, 0, len(ids))
+						for _, id := range ids {
+							contents = append(contents, &message.FunctionCallContent{CallID: id, Name: "effect", Arguments: `{}`})
+						}
+						yield(&agent.ResponseUpdate{
+							Role: message.RoleAssistant, MessageID: fmt.Sprintf("round-%d", providerCalls), Contents: contents,
+						}, nil)
+					}
+				}
+				middleware := toolautocall.New(toolautocall.Config{IncludeDetailedErrors: true})
+				var results []*message.FunctionResultContent
+				var runErr error
+				for update, err := range middleware.Run(
+					provider, ctx, []*message.Message{message.NewText("perform actions")}, agent.WithTool(effect), agent.Stream(stream),
+				) {
+					if update != nil {
+						if !stream && effects != 3 {
+							t.Error("non-streaming run exposed updates before the loop stopped")
+						}
+						for _, content := range update.Contents {
+							if result, ok := content.(*message.FunctionResultContent); ok {
+								results = append(results, result)
+							}
+						}
+					}
+					if err != nil {
+						runErr = err
+						break
+					}
+				}
+				if effects != 3 || providerCalls != 2 || len(results) != 4 {
+					t.Fatalf("effects=%d providers=%d results=%d, want 3/2/4", effects, providerCalls, len(results))
+				}
+				for i, id := range []string{"earlier", "first", "interrupted", "skipped"} {
+					if results[i].CallID != id {
+						t.Errorf("result %d call ID = %q, want %q", i, results[i].CallID, id)
+					}
+				}
+				if results[0].Error != nil || results[0].Result != "completed 1" ||
+					results[1].Error != nil || results[1].Result != "completed 2" {
+					t.Errorf("completed outcomes were lost: %+v", results)
+				}
+				switch outcome {
+				case "success":
+					if !errors.Is(runErr, context.Canceled) || results[2].Error != nil || results[2].Result != "completed 3" {
+						t.Errorf("successful interrupted call was lost: result=%+v err=%v", results[2], runErr)
+					}
+				case "error":
+					if runErr != toolFailure || results[2].Error != toolFailure {
+						t.Errorf("original tool failure changed: result=%+v err=%v", results[2], runErr)
+					}
+				case "panic error":
+					if !errors.Is(runErr, toolFailure) || results[2].Error != runErr {
+						t.Errorf("original panic error was lost: result=%+v err=%v", results[2], runErr)
+					}
+				case "panic value":
+					if runErr == nil || !strings.Contains(runErr.Error(), "original tool panic") || results[2].Error != runErr {
+						t.Errorf("original panic was lost: result=%+v err=%v", results[2], runErr)
+					}
+				}
+				if !errors.Is(results[3].Error, context.Canceled) ||
+					!strings.Contains(fmt.Sprint(results[3].Result), "not invoked") {
+					t.Errorf("skipped call was not reported as unexecuted: %+v", results[3])
+				}
+				response, err := agent.ResponseStream(middleware.Run(
+					provider, t.Context(), []*message.Message{message.NewText("resume")}, agent.WithTool(effect), agent.Stream(stream),
+				)).Collect()
+				if err != nil || response.String() != "done" || effects != 3 || providerCalls != 3 {
+					t.Fatalf("reuse: response=%v err=%v effects=%d providers=%d", response, err, effects, providerCalls)
+				}
+				for content := range response.Contents() {
+					if _, ok := content.(*message.FunctionResultContent); ok {
+						t.Fatal("reusing middleware re-emitted a previous run's results")
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestFunctionInvoking_ApprovedCancellationPreservesResults(t *testing.T) {
+	for _, stream := range []bool{false, true} {
+		t.Run(fmt.Sprintf("stream=%t", stream), func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+			var effects int
+			effect := tool.ApprovalRequiredFunc(functool.MustNew(functool.Config{Name: "effect"}, func(context.Context, struct{}) (string, error) {
+				effects++
+				if effects == 2 {
+					cancel()
+				}
+				return fmt.Sprintf("completed %d", effects), nil
+			}))
+			var requests, approvals message.Contents
+			for _, id := range []string{"a", "b", "c"} {
+				call := &message.FunctionCallContent{CallID: id, Name: "effect", Arguments: `{}`}
+				requests = append(requests, &message.ToolApprovalRequestContent{RequestID: "request-" + id, ToolCall: call})
+				approvals = append(approvals, &message.ToolApprovalResponseContent{RequestID: "request-" + id, ToolCall: call, Approved: true})
+			}
+			messages := []*message.Message{
+				{Role: message.RoleAssistant, Contents: requests},
+				message.New(approvals...),
+			}
+			var providerCalls int
+			provider := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
+				return func(yield func(*agent.ResponseUpdate, error) bool) {
+					providerCalls++
+					yield(&agent.ResponseUpdate{Role: message.RoleAssistant, Contents: message.Contents{&message.TextContent{Text: "unexpected"}}}, nil)
+				}
+			}
+			var results []*message.FunctionResultContent
+			var runErr error
+			for update, err := range toolautocall.New(toolautocall.Config{DisableApprovalResponseBinding: true}).Run(
+				provider, ctx, messages, agent.WithTool(effect), agent.Stream(stream),
+			) {
+				if update != nil {
+					for _, content := range update.Contents {
+						if result, ok := content.(*message.FunctionResultContent); ok {
+							results = append(results, result)
+						}
+					}
+				}
+				if err != nil {
+					runErr = err
+					break
+				}
+			}
+			if !errors.Is(runErr, context.Canceled) || effects != 2 || providerCalls != 0 || len(results) != 3 {
+				t.Fatalf("err=%v effects=%d providers=%d results=%d, want cancelled approval results without a provider call", runErr, effects, providerCalls, len(results))
+			}
+			for i, id := range []string{"a", "b", "c"} {
+				if results[i].CallID != id {
+					t.Errorf("result %d call ID = %q, want %q", i, results[i].CallID, id)
+				}
+			}
+			if results[0].Result != "completed 1" || results[0].Error != nil ||
+				results[1].Result != "completed 2" || results[1].Error != nil ||
+				!errors.Is(results[2].Error, context.Canceled) {
+				t.Errorf("approved outcomes lost or misreported: %+v", results)
+			}
+		})
+	}
+}
+
+func TestFunctionInvoking_ProviderFailurePreservesUpdates(t *testing.T) {
+	providerErr := errors.New("provider failed")
+	for _, stream := range []bool{false, true} {
+		t.Run(fmt.Sprintf("stream=%t", stream), func(t *testing.T) {
+			var providerCalls, toolCalls int
+			effect := functool.MustNew(functool.Config{Name: "effect"}, func(context.Context, struct{}) (string, error) {
+				toolCalls++
+				return "completed", nil
+			})
+			provider := func(context.Context, []*message.Message, ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
+				return func(yield func(*agent.ResponseUpdate, error) bool) {
+					providerCalls++
+					if providerCalls == 1 {
+						yield(&agent.ResponseUpdate{
+							MessageID: "first", ResponseID: "first", Role: message.RoleAssistant,
+							RawRepresentation: json.RawMessage(`{"first":true}`),
+							Contents: message.Contents{
+								&message.TextContent{Text: "started"},
+								&message.FunctionCallContent{CallID: "call", Name: "effect", Arguments: `{}`},
+								&message.UsageContent{Details: message.UsageDetails{InputTokenCount: 2}},
+							},
+						}, nil)
+						return
+					}
+					if !yield(&agent.ResponseUpdate{
+						MessageID: "second", ResponseID: "second", Role: message.RoleAssistant,
+						RawRepresentation: json.RawMessage(`{"second":true}`),
+						Contents:          message.Contents{&message.TextContent{Text: "unfinished"}},
+					}, nil) {
+						return
+					}
+					if !yield(&agent.ResponseUpdate{AdditionalProperties: map[string]any{"checkpoint": "saved"}}, nil) {
+						return
+					}
+					yield(nil, providerErr)
+				}
+			}
+			middleware := toolautocall.New(toolautocall.Config{})
+			var response agent.Response
+			var runErr error
+			for update, err := range middleware.Run(
+				provider, t.Context(), []*message.Message{message.NewText("start")}, agent.WithTool(effect), agent.Stream(stream),
+			) {
+				if update != nil && !stream && providerCalls != 2 {
+					t.Error("non-streaming run exposed updates before the loop stopped")
+				}
+				response.Update(update)
+				if err != nil {
+					runErr = err
+					break
+				}
+			}
+			if runErr != providerErr || providerCalls != 2 || toolCalls != 1 || len(response.Messages) != 3 ||
+				response.String() != "started\nunfinished" || response.Usage().InputTokenCount != 2 ||
+				response.AdditionalProperties["checkpoint"] != "saved" {
+				t.Fatalf("partial response=%+v err=%v providers=%d tools=%d", response, runErr, providerCalls, toolCalls)
+			}
+			result, ok := response.Messages[1].Contents[0].(*message.FunctionResultContent)
+			if !ok || result.CallID != "call" || result.Result != "completed" || result.Error != nil {
+				t.Fatalf("completed tool result was lost: %+v", response.Messages[1])
+			}
+			for i, want := range map[int]string{0: `{"first":true}`, 2: `{"second":true}`} {
+				if raw, ok := response.Messages[i].RawRepresentation.(json.RawMessage); !ok || string(raw) != want {
+					t.Errorf("message %d raw representation = %#v, want %s", i, response.Messages[i].RawRepresentation, want)
+				}
+			}
+			providerCalls, toolCalls = 0, 0
+			collected, err := agent.ResponseStream(middleware.Run(
+				provider, t.Context(), []*message.Message{message.NewText("start")}, agent.WithTool(effect), agent.Stream(stream),
+			)).Collect()
+			if collected != nil || err != providerErr {
+				t.Fatalf("Collect returned %v/%v, want no response and the original error", collected, err)
+			}
+		})
+	}
+}
+
 func TestFunctionInvoking_RequestCancellation(t *testing.T) {
 	toolFailure := errors.New("first failed")
 	secondFailure := errors.New("second failed")
@@ -2490,6 +2761,7 @@ func TestFunctionInvoking_RequestCancellation(t *testing.T) {
 						}
 					}
 					var runErr error
+					var results []*message.FunctionResultContent
 					for update, err := range toolautocall.New(toolautocall.Config{
 						AllowConcurrentInvocations:  tc.concurrent,
 						MaximumIterationsPerRequest: tc.maxIterations,
@@ -2499,6 +2771,13 @@ func TestFunctionInvoking_RequestCancellation(t *testing.T) {
 							runErr = err
 							break
 						}
+						if update != nil {
+							for _, content := range update.Contents {
+								if result, ok := content.(*message.FunctionResultContent); ok {
+									results = append(results, result)
+								}
+							}
+						}
 						if tc.cancelAt == "after results" && update != nil && update.Role == message.RoleTool {
 							stop()
 						}
@@ -2520,6 +2799,50 @@ func TestFunctionInvoking_RequestCancellation(t *testing.T) {
 					if providerCalls != wantProviders {
 						t.Errorf("provider calls = %d, want %d", providerCalls, wantProviders)
 					}
+					wantResults := 2
+					if wantProviders == 0 || tc.maxIterations != nil && *tc.maxIterations == 0 {
+						wantResults = 0
+					}
+					if len(results) != wantResults {
+						t.Fatalf("results = %+v, want %d paired outcomes", results, wantResults)
+					}
+					var skipped int32
+					var successfulResults []string
+					for i, result := range results {
+						if result.CallID != fmt.Sprint(i+1) {
+							t.Errorf("result %d call ID = %q, want original call order", i, result.CallID)
+						}
+						if result.Result == "Error: Function was not invoked." {
+							skipped++
+							if !errors.Is(result.Error, ctx.Err()) {
+								t.Errorf("unexecuted call error = %v, want %v", result.Error, ctx.Err())
+							}
+						} else if result.Error == nil {
+							value, ok := result.Result.(string)
+							if !ok {
+								t.Errorf("completed outcome = %v, want a string", result.Result)
+							}
+							successfulResults = append(successfulResults, value)
+						}
+					}
+					if wantResults > 0 && skipped != int32(wantResults)-tc.wantTools {
+						t.Errorf("unexecuted outcomes = %d, want %d", skipped, int32(wantResults)-tc.wantTools)
+					}
+					var wantSuccessfulResults []string
+					for i := range int(tc.wantTools) {
+						if i < len(tc.toolErrors) {
+							if tc.toolErrors[i] == nil {
+								wantSuccessfulResults = append(wantSuccessfulResults, "")
+							}
+						} else {
+							wantSuccessfulResults = append(wantSuccessfulResults, "result")
+						}
+					}
+					slices.Sort(successfulResults)
+					slices.Sort(wantSuccessfulResults)
+					if !slices.Equal(successfulResults, wantSuccessfulResults) {
+						t.Errorf("completed outcomes = %v, want %v", successfulResults, wantSuccessfulResults)
+					}
 				})
 			})
 		}
diff --git a/agent/messageinjection.go b/agent/messageinjection.go
index 827eca40..f940189d 100644
--- a/agent/messageinjection.go
+++ b/agent/messageinjection.go
@@ -20,6 +20,9 @@ type pendingInjectedMessagesState struct {
 
 // MessageInjector queues messages for injection between provider calls.
 // Its zero value is ready to use.
+// To retain delivered messages across stateless provider calls, enable
+// [Config.RequirePerServiceCallHistoryPersistence] and use a [HistoryProvider]
+// that preserves their content, including images and other non-text parts.
 type MessageInjector struct {
 	mu sync.Mutex
 }
diff --git a/agent/messageinjection_test.go b/agent/messageinjection_test.go
index 2d4263fd..5c9c476b 100644
--- a/agent/messageinjection_test.go
+++ b/agent/messageinjection_test.go
@@ -763,6 +763,118 @@ func TestMessageInjection_ConcurrentEnqueuesDoNotLoseMessages(t *testing.T) {
 	}
 }
 
+func TestMessageInjection_ExternalHistoryRetainsImagesAcrossToolRounds(t *testing.T) {
+	for _, perCall := range []bool{false, true} {
+		for _, stream := range []bool{false, true} {
+			t.Run(fmt.Sprintf("per-call=%t/stream=%t", perCall, stream), func(t *testing.T) {
+				var stored [][]byte
+				history := agent.NewHistoryProvider(agent.HistoryProviderConfig{
+					SourceID: "external-store",
+					Provide: func(context.Context, agent.InvokingContext) ([]*message.Message, error) {
+						messages := make([]*message.Message, 0, len(stored))
+						for _, data := range stored {
+							var msg message.Message
+							if err := json.Unmarshal(data, &msg); err != nil {
+								return nil, fmt.Errorf("decode history: %w", err)
+							}
+							messages = append(messages, &msg)
+						}
+						return messages, nil
+					},
+					Store: func(_ context.Context, invoked agent.InvokedContext) error {
+						for _, batch := range [][]*message.Message{invoked.RequestMessages, invoked.ResponseMessages} {
+							for _, msg := range batch {
+								data, err := json.Marshal(msg)
+								if err != nil {
+									return fmt.Errorf("encode history: %w", err)
+								}
+								stored = append(stored, data)
+							}
+						}
+						return nil
+					},
+				})
+				injection := &agent.MessageInjector{}
+				session := &agent.Session{}
+				var fetchCalls, noopCalls int
+				fetch := functool.MustNew(functool.Config{Name: "fetch"}, func(context.Context, struct{}) (string, error) {
+					fetchCalls++
+					if err := injection.EnqueueMessages(session, message.New(&message.URIContent{
+						URI: "https://example.invalid/image.png", MediaType: "image/png",
+					})); err != nil {
+						return "", err
+					}
+					return "fetched image", nil
+				})
+				noop := functool.MustNew(functool.Config{Name: "noop"}, func(context.Context, struct{}) (string, error) {
+					noopCalls++
+					return "completed", nil
+				})
+				var imageCounts, imagePositions []int
+				run := func(_ context.Context, messages []*message.Message, opts ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] {
+					return func(yield func(*agent.ResponseUpdate, error) bool) {
+						if id, _ := agent.GetOption(opts, agent.WithServiceID); id != "" {
+							t.Errorf("local history ID reached the provider: %q", id)
+						}
+						count, position := 0, -1
+						for i, msg := range messages {
+							for _, content := range msg.Contents {
+								if image, ok := content.(*message.URIContent); ok && image.URI == "https://example.invalid/image.png" {
+									count++
+									position = i
+									if image.MediaType != "image/png" || msg.Role != message.RoleUser {
+										t.Errorf("image content or role was not preserved: %+v", msg)
+									}
+								}
+							}
+						}
+						imageCounts = append(imageCounts, count)
+						imagePositions = append(imagePositions, position)
+						contents := message.Contents{&message.TextContent{Text: "done"}}
+						switch len(imageCounts) {
+						case 1:
+							contents = message.Contents{&message.FunctionCallContent{CallID: "fetch", Name: "fetch", Arguments: `{}`}}
+						case 2:
+							contents = message.Contents{&message.FunctionCallContent{CallID: "noop", Name: "noop", Arguments: `{}`}}
+						}
+						yield(&agent.ResponseUpdate{
+							Role: message.RoleAssistant, MessageID: fmt.Sprintf("round-%d", len(imageCounts)), Contents: contents,
+						}, nil)
+					}
+				}
+				a := agent.New(agent.ProviderConfig{
+					Run: run, Middlewares: []agent.Middleware{toolautocall.New(toolautocall.Config{})},
+				}, agent.Config{
+					HistoryProvider: history, MessageInjector: injection,
+					RequirePerServiceCallHistoryPersistence: perCall,
+					Tools:                                   []tool.Tool{fetch, noop},
+				})
+				response, err := a.RunText(t.Context(), "start", agent.WithSession(session), agent.Stream(stream)).Collect()
+				if err != nil {
+					t.Fatal(err)
+				}
+				wantCounts, wantStored := []int{0, 1, 0}, 6
+				if perCall {
+					wantCounts, wantStored = []int{0, 1, 1}, 7
+				}
+				if !slices.Equal(imageCounts, wantCounts) || len(stored) != wantStored ||
+					fetchCalls != 1 || noopCalls != 1 || response.String() != "done" {
+					t.Fatalf("images=%v stored=%d fetch=%d noop=%d response=%q", imageCounts, len(stored), fetchCalls, noopCalls, response.String())
+				}
+				if perCall && imagePositions[1] != imagePositions[2] {
+					t.Errorf("image moved between provider calls: %v", imagePositions)
+				}
+				if memory, err := session.InMemoryHistory(""); err != nil || len(memory) != 0 {
+					t.Errorf("external history unexpectedly used session in-memory history: %v %v", memory, err)
+				}
+				if pending, err := injection.PendingMessages(session); err != nil || len(pending) != 0 {
+					t.Errorf("delivered image remained queued: %v %v", pending, err)
+				}
+			})
+		}
+	}
+}
+
 func newMessageInjectingAgent(run agent.RunFunc, config toolautocall.Config, tools []tool.Tool, injection *agent.MessageInjector) *agent.Agent {
 	return agent.New(agent.ProviderConfig{
 		Run:         run,
diff --git a/docs/agent-harness.md b/docs/agent-harness.md
index a914450d..650cca22 100644
--- a/docs/agent-harness.md
+++ b/docs/agent-harness.md
@@ -118,6 +118,58 @@ The harness does not override those settings. Approval-required tools can still
 pause a run when `DisableToolApproval` is true; the application must return the
 approval response through the same session. The runnable example shows that flow.
 
+### Partial work and cancellation
+
+Tool invocation does not roll back side effects. When request cancellation stops
+a tool batch, automatic invocation yields the actual outcomes of completed calls
+and failure results for calls that were not invoked, before returning the terminal
+error. The original tool error is preserved when a tool fails during cancellation.
+The same reporting applies when the consecutive-error limit stops a batch.
+
+With [`agent.Stream(false)`](../agent/options.go), updates remain buffered until
+the tool loop stops, including when it fails. To inspect or checkpoint partial
+work, consume the returned sequence directly or from middleware.
+[`ResponseStream.Collect`](../agent/response.go) still returns no response on
+error. Save completed tool exchanges before forwarding the terminal error.
+Cancellation cleanup may need a bounded context that is independent of request
+cancellation; do not use that context to continue tool execution.
+
+For incremental tool-round updates with non-streaming model requests, compose
+`toolautocall` with provider-boundary middleware that applies `agent.Stream(false)`
+only to the individual model call, and run the outer loop with `agent.Stream(true)`.
+Applying the non-streaming option outside the tool loop also enables buffering.
+The normal non-streaming default is unchanged.
+
+### Injected messages and external history
+
+The harness enables per-service-call history persistence. When configuring an
+agent directly, enable `RequirePerServiceCallHistoryPersistence` to retain
+injected messages across stateless model calls. The configured history provider
+can use SQLite or another external store; session-backed in-memory history is
+not required.
+
+Use [`agent.NewHistoryProvider`](../agent/history.go) with `Provide` returning
+only stored history and `Store` saving the new request and response messages:
+
+```go
+cfg := agent.Config{
+	HistoryProvider: agent.NewHistoryProvider(agent.HistoryProviderConfig{
+		SourceID: "external-history",
+		Provide:  loadHistory,
+		Store:    storeMessages,
+	}),
+	MessageInjector:                         &agent.MessageInjector{},
+	RequirePerServiceCallHistoryPersistence: true,
+}
+```
+
+Here, `loadHistory` and `storeMessages` are application-supplied storage callbacks.
+The helper source-stamps loaded messages and excludes them from the default
+stored request messages, avoiding duplicate history. Preserve typed message
+content, for example by JSON-encoding complete `message.Message` values. Saving
+only `Message.String()` loses images and other non-text content; enabling
+per-call persistence cannot recover content discarded by the storage adapter.
+
 ### Provider-managed history
 
 `AgentConfig` does not change provider request-storage settings. The usage example
diff --git a/provider/openaiprovider/chat.go b/provider/openaiprovider/chat.go
index 5acf150f..ad1e4775 100644
--- a/provider/openaiprovider/chat.go
+++ b/provider/openaiprovider/chat.go
@@ -197,23 +197,17 @@ func (a *chatClient) run(ctx context.Context, messages []*message.Message, optio
 		stream := a.client.Chat.Completions.NewStreaming(ctx, body, telemetryOption)
 		defer func() { _ = stream.Close() }()
 		var acc openai.ChatCompletionAccumulator
+		var created int64
 		for stream.Next() {
 			chunk := stream.Current()
 			modelcall.RecordResponse(ctx, chunk.Model, chunk.ID)
 			if !acc.AddChunk(chunk) {
 				continue
 			}
-			var contents []message.Content
-			if tc, ok := acc.JustFinishedToolCall(); ok {
-				contents = append(contents, &message.FunctionCallContent{
-					CallID:    tc.ID,
-					Name:      tc.Name,
-					Arguments: tc.Arguments,
-				})
-			}
-			if refusal, ok := acc.JustFinishedRefusal(); ok {
-				contents = append(contents, &message.ErrorContent{Message: refusal, ErrorCode: "Refusal"})
+			if created <= 0 && chunk.Created > 0 {
+				created = chunk.Created
 			}
+			var contents []message.Content
 			role := message.RoleAssistant
 			if len(chunk.Choices) > 0 {
 				choice := chunk.Choices[0]
@@ -242,8 +236,34 @@ func (a *chatClient) run(ctx context.Context, messages []*message.Message, optio
 				return
 			}
 		}
-		if stream.Err() != nil {
-			yield(nil, stream.Err())
+		if err := stream.Err(); err != nil {
+			yield(nil, err)
+			return
+		}
+		if len(acc.Choices) == 0 {
+			return
+		}
+		choice := acc.Choices[0]
+		var contents []message.Content
+		for _, tc := range choice.Message.ToolCalls {
+			contents = append(contents, &message.FunctionCallContent{
+				CallID:    tc.ID,
+				Name:      tc.Function.Name,
+				Arguments: tc.Function.Arguments,
+			})
+		}
+		if choice.Message.Refusal != "" {
+			contents = append(contents, &message.ErrorContent{Message: choice.Message.Refusal, ErrorCode: "Refusal"})
+		}
+		if len(contents) > 0 {
+			yield(&agent.ResponseUpdate{
+				Contents:     contents,
+				Role:         mapRole(string(choice.Message.Role)),
+				ResponseID:   acc.ID,
+				MessageID:    acc.ID,
+				FinishReason: normalizeChatFinishReason(choice.FinishReason),
+				CreatedAt:    time.Unix(created, 0),
+			}, nil)
 		}
 	}
 }
diff --git a/provider/openaiprovider/chat_test.go b/provider/openaiprovider/chat_test.go
index fd5fa012..e9d29af4 100644
--- a/provider/openaiprovider/chat_test.go
+++ b/provider/openaiprovider/chat_test.go
@@ -1202,6 +1202,27 @@ data: [DONE]
 	}
 }
 
+func TestChatStreamingRefusal_TerminalChunk(t *testing.T) {
+	const input = `{"messages":[{"role":"user","content":"question"}],"model":"gpt-4o-mini","stream":true,"stream_options":{"include_usage":true}}`
+	const output = "data: " + `{"id":"refusal","object":"chat.completion.chunk","created":42,"choices":[{"index":0,"delta":{"role":"assistant","refusal":"Cannot assist."},"finish_reason":"stop"}]}` + "\n\ndata: [DONE]\n\n"
+	server := newTestServerStreaming(t, input, output)
+	defer server.Close()
+
+	response, err := newTestClient(server).RunText(t.Context(), "question", agent.Stream(true)).Collect()
+	if err != nil {
+		t.Fatal(err)
+	}
+	var refusals []*message.ErrorContent
+	for content := range response.Contents() {
+		if refusal, ok := content.(*message.ErrorContent); ok {
+			refusals = append(refusals, refusal)
+		}
+	}
+	if len(refusals) != 1 || refusals[0].Message != "Cannot assist." || refusals[0].ErrorCode != "Refusal" {
+		t.Fatalf("refusals = %+v, want one complete refusal", refusals)
+	}
+}
+
 // TestChatStreamingIncludeUsage_RespectsCallerOverride verifies that when the
 // caller explicitly sets stream_options.include_usage the provider does not
 // override it with the default true value.
@@ -1754,9 +1775,7 @@ data: [DONE]
 		{MessageID: msgID, ResponseID: msgID, Role: message.RoleAssistant, CreatedAt: createdAt},
 		{MessageID: msgID, ResponseID: msgID, Role: message.RoleAssistant, CreatedAt: createdAt},
 		{MessageID: msgID, ResponseID: msgID, Role: message.RoleAssistant, CreatedAt: createdAt},
-		{MessageID: msgID, ResponseID: msgID, FinishReason: "tool_calls", Role: message.RoleAssistant, CreatedAt: createdAt, Contents: []message.Content{
-			&message.FunctionCallContent{CallID: callID, Name: "GetPersonAge", Arguments: `{"personName":"Alice"}`},
-		}},
+		{MessageID: msgID, ResponseID: msgID, FinishReason: "tool_calls", Role: message.RoleAssistant, CreatedAt: createdAt},
 		{MessageID: msgID, ResponseID: msgID, Role: message.RoleAssistant, CreatedAt: createdAt, Contents: []message.Content{
 			&message.UsageContent{Details: message.UsageDetails{
 				InputTokenCount:       61,
@@ -1766,6 +1785,9 @@ data: [DONE]
 				ReasoningTokenCount:   90,
 			}},
 		}},
+		{MessageID: msgID, ResponseID: msgID, FinishReason: "tool_calls", Role: message.RoleAssistant, CreatedAt: createdAt, Contents: []message.Content{
+			&message.FunctionCallContent{CallID: callID, Name: "GetPersonAge", Arguments: `{"personName":"Alice"}`},
+		}},
 	}
 
 	server := newTestServerStreaming(t, input, output)
@@ -1796,6 +1818,187 @@ data: [DONE]
 	}
 }
 
+func TestChatStreamingToolCalls_CompleteOnce(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		choices []string
+	}{
+		{
+			name: "packed terminal calls",
+			choices: []string{
+				`[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"first","arguments":"{\"value\":1}"}},{"index":1,"id":"b","type":"function","function":{"name":"second","arguments":"{\"value\":2}"}}]},"finish_reason":"tool_calls"}]`,
+			},
+		},
+		{
+			name: "packed calls before terminal chunk",
+			choices: []string{
+				`[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"first","arguments":"{\"value\":1}"}},{"index":1,"id":"b","type":"function","function":{"name":"second","arguments":"{\"value\":2}"}}]}}]`,
+				`[{"index":0,"delta":{},"finish_reason":"tool_calls"}]`,
+			},
+		},
+		{
+			name: "interleaved arguments",
+			choices: []string{
+				`[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"first","arguments":"{\"value\":"}},{"index":1,"id":"b","type":"function","function":{"name":"second","arguments":"{\"value\":"}}]}}]`,
+				`[{"index":0,"delta":{"tool_calls":[{"index":1,"function":{"arguments":"2}"}}]}}]`,
+				`[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1}"}}]}}]`,
+				`[{"index":0,"delta":{},"finish_reason":"tool_calls"}]`,
+			},
+		},
+		{
+			name: "sequential calls",
+			choices: []string{
+				`[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"first","arguments":"{\"value\":1}"}}]}}]`,
+				`[{"index":0,"delta":{"tool_calls":[{"index":1,"id":"b","type":"function","function":{"name":"second","arguments":"{\"value\":2}"}}]}}]`,
+				`[{"index":0,"delta":{},"finish_reason":"tool_calls"}]`,
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			const input = `{"messages":[{"role":"user","content":"call both"}],"model":"gpt-4o-mini","stream":true,"stream_options":{"include_usage":true}}`
+			output := "data: " + `{"id":"calls","object":"chat.completion.chunk","created":123,"choices":[{"index":0,"delta":{"role":"assistant","content":"working"}}]}` + "\n\n"
+			for _, choices := range tc.choices {
+				output += fmt.Sprintf("data: {\"id\":\"calls\",\"object\":\"chat.completion.chunk\",\"created\":123,\"choices\":%s}\n\n", choices)
+			}
+			output += "data: " + `{"id":"calls","object":"chat.completion.chunk","choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}` + "\n\ndata: [DONE]\n\n"
+			server := newTestServerStreaming(t, input, output)
+			defer server.Close()
+
+			var response agent.Response
+			var updates []*agent.ResponseUpdate
+			var calls []*message.FunctionCallContent
+			for update, err := range newTestClient(server).RunText(t.Context(), "call both", agent.Stream(true)) {
+				if err != nil {
+					t.Fatal(err)
+				}
+				updates = append(updates, update)
+				response.Update(update)
+				for _, content := range update.Contents {
+					if call, ok := content.(*message.FunctionCallContent); ok {
+						calls = append(calls, call)
+					}
+				}
+			}
+			if len(calls) != 2 {
+				t.Fatalf("calls = %+v, want two complete calls", calls)
+			}
+			for i, want := range []message.FunctionCallContent{
+				{CallID: "a", Name: "first", Arguments: `{"value":1}`},
+				{CallID: "b", Name: "second", Arguments: `{"value":2}`},
+			} {
+				if got := calls[i]; got.CallID != want.CallID || got.Name != want.Name || got.Arguments != want.Arguments {
+					t.Errorf("call %d = %+v, want %+v", i, got, want)
+				}
+			}
+			last := updates[len(updates)-1]
+			if last.MessageID != "calls" || last.ResponseID != "calls" || last.Role != message.RoleAssistant ||
+				last.FinishReason != "tool_calls" || !last.CreatedAt.Equal(time.Unix(123, 0)) || len(last.Contents) != 2 {
+				t.Errorf("completed update lost metadata or calls: %+v", last)
+			}
+			if updates[0].String() != "working" || response.String() != "working" || response.Usage().TotalTokenCount != 3 {
+				t.Errorf("text or usage lost or duplicated: %+v", response)
+			}
+			if raw, ok := response.RawRepresentation.([]any); !ok || len(raw) != len(tc.choices)+2 {
+				t.Errorf("RawRepresentation = %#v, want only the original stream chunks", response.RawRepresentation)
+			}
+		})
+	}
+}
+
+func TestChatStreamingPackedToolCalls_InvokeOnce(t *testing.T) {
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		if requests.Add(1) == 1 {
+			_, _ = fmt.Fprint(w, "data: "+`{"id":"calls","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"first","arguments":"{\"value\":1}"}},{"index":1,"id":"b","type":"function","function":{"name":"second","arguments":"{\"value\":2}"}}]},"finish_reason":"tool_calls"}]}`+"\n\ndata: [DONE]\n\n")
+			return
+		}
+		var body struct {
+			Messages []struct {
+				Role       string `json:"role"`
+				ToolCallID string `json:"tool_call_id"`
+				Content    string `json:"content"`
+			} `json:"messages"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Error(err)
+			return
+		}
+		var ids, values []string
+		for _, msg := range body.Messages {
+			if msg.Role == "tool" {
+				ids = append(ids, msg.ToolCallID)
+				values = append(values, msg.Content)
+			}
+		}
+		if !slices.Equal(ids, []string{"a", "b"}) || !slices.Equal(values, []string{"1", "2"}) {
+			t.Errorf("tool results = %v/%v, want a=1 and b=2", ids, values)
+		}
+		_, _ = fmt.Fprint(w, "data: "+`{"id":"done","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"done"},"finish_reason":"stop"}]}`+"\n\ndata: [DONE]\n\n")
+	}))
+	defer server.Close()
+
+	var values []int
+	invoke := func(_ context.Context, args struct {
+		Value int `json:"value"`
+	},
+	) (int, error) {
+		values = append(values, args.Value)
+		return args.Value, nil
+	}
+	a := openaiprovider.NewChatCompletionsAgent(
+		openai.NewClient(option.WithBaseURL(server.URL), option.WithAPIKey("test"), option.WithMaxRetries(0)),
+		openaiprovider.AgentConfig{Model: "m", Config: agent.Config{Tools: []tool.Tool{
+			functool.MustNew(functool.Config{Name: "first"}, invoke),
+			functool.MustNew(functool.Config{Name: "second"}, invoke),
+		}}},
+	)
+	response, err := a.RunText(t.Context(), "call both", agent.Stream(true)).Collect()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !slices.Equal(values, []int{1, 2}) || requests.Load() != 2 || response.String() != "done" {
+		t.Fatalf("values=%v requests=%d response=%q, want one invocation per call", values, requests.Load(), response.String())
+	}
+}
+
+func TestChatStreamingToolCalls_FailureDoesNotInvokeTools(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = fmt.Fprint(w, "data: "+`{"id":"calls","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"a","type":"function","function":{"name":"effect","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}`+"\n\n")
+		_, _ = fmt.Fprint(w, "data: "+`{"error":{"message":"broken stream","type":"server_error","code":"stream_failed"}}`+"\n\n")
+	}))
+	defer server.Close()
+
+	var effects int
+	effect := functool.MustNew(functool.Config{Name: "effect"}, func(context.Context, struct{}) (string, error) {
+		effects++
+		return "completed", nil
+	})
+	a := openaiprovider.NewChatCompletionsAgent(
+		openai.NewClient(option.WithBaseURL(server.URL), option.WithAPIKey("test"), option.WithMaxRetries(0)),
+		openaiprovider.AgentConfig{Model: "m", Config: agent.Config{Tools: []tool.Tool{effect}}},
+	)
+	var calls int
+	var runErr error
+	for update, err := range a.RunText(t.Context(), "perform action", agent.Stream(true)) {
+		if update != nil {
+			for _, content := range update.Contents {
+				if _, ok := content.(*message.FunctionCallContent); ok {
+					calls++
+				}
+			}
+		}
+		if err != nil {
+			runErr = err
+			break
+		}
+	}
+	if runErr == nil || !strings.Contains(runErr.Error(), "broken stream") || effects != 0 || calls != 0 {
+		t.Fatalf("err=%v effects=%d calls=%d, want original stream failure and no dispatchable calls", runErr, effects, calls)
+	}
+}
+
 func TestChatAssistantMessageWithBothToolsAndContent_NonStreaming(t *testing.T) {
 	const input = `
             {