From abb0f6f4ae98046da10a985ab9a515118991ae92 Mon Sep 17 00:00:00 2001 From: qmuntal Date: Tue, 3 Nov 2026 11:12:30 +0100 Subject: [PATCH] Preserve tool outcomes and fix streamed call assembly Assemble complete streamed tool calls and refusals after successful stream completion. Preserve completed and unexecuted tool outcomes before terminal errors, including buffered runs, concurrent batches, and approval resumption. Document per-call image history and add regression coverage without changing non-streaming defaults or introducing new public APIs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- agent/harness/toolautocall/autocall.go | 92 +++--- agent/harness/toolautocall/autocall_test.go | 325 +++++++++++++++++++- agent/messageinjection.go | 3 + agent/messageinjection_test.go | 112 +++++++ docs/agent-harness.md | 52 ++++ provider/openaiprovider/chat.go | 44 ++- provider/openaiprovider/chat_test.go | 209 ++++++++++++- 7 files changed, 786 insertions(+), 51 deletions(-) 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 = ` {