diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index d54eda937..9d5547b2c 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -6826,8 +6826,8 @@ "summary": "Align Assistant live tool progress with OpenCode source workflow", "target_id": "v6-product-lane-expansion", "claimed_at": "2026-06-06T04:02:24Z", - "heartbeat_at": "2026-06-06T04:02:24Z", - "expires_at": "2026-06-06T12:02:24Z", + "heartbeat_at": "2026-06-06T04:06:45Z", + "expires_at": "2026-06-06T12:06:45Z", "work_item": { "kind": "lane-followup", "id": "architecture-post-rc-canonicalization" diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index c0560a885..9a56c428e 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -293,6 +293,17 @@ runtime cost control, and shared AI transport surfaces. stream contract adapts that model with `tool_start`, `tool_progress`, and `tool_end` events that update the same visible pending tool row in place. The referenced OpenCode source at fetched `origin/dev` commit + `09d9cf01f93798939c1284fbe974b6e1f4d2759d` also emits and applies + tool-input lifecycle deltas in + `packages/opencode/src/session/processor.ts` (`tool-input-start`, + `tool-input-delta`, `tool-input-end`) and + `packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx` + (`session.next.tool.input.started`, `.delta`, `.ended`). Pulse's + OpenAI-compatible provider path must surface streamed function-argument + deltas as `tool_progress` so the pending tool row mutates while the model is + still forming the call; waiting until `[DONE]` or execution completion + recreates the delayed batch feeling this contract is meant to prevent. + The referenced OpenCode source at fetched `origin/dev` commit `9ed17da55ab1f7360cc0e01075f763e27fa899e9` creates a tool message part as soon as tool input starts in `packages/opencode/src/session/processor.ts`, mutates that part through running/completed/error states via diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index 3207a9ee6..c672fd7ed 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -245,10 +245,17 @@ func emitToolStartEvent(callback StreamCallback, id, name string, input map[stri } func emitToolProgressEvent(callback StreamCallback, id, name string, input map[string]interface{}, phase, message string) { + emitToolProgressEventWithRawInput(callback, id, name, input, "", phase, message) +} + +func emitToolProgressEventWithRawInput(callback StreamCallback, id, name string, input map[string]interface{}, rawInputOverride, phase, message string) { if callback == nil { return } inputStr, rawInput := formatToolInputForFrontend(name, input, false) + if rawInputOverride != "" { + rawInput = rawInputOverride + } jsonData, _ := json.Marshal(ToolProgressData{ ID: id, Name: name, @@ -895,6 +902,43 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me emitToolStartEvent(callback, data.ID, data.Name, data.Input) } + case "tool_progress": + if data, ok := event.Data.(providers.ToolProgressEvent); ok { + if data.Name == pulseQuestionToolName { + return + } + data.Input = normalizeSummaryOnlyPulseQueryInput(preferSummaryOnlyQueries, data.Name, data.Input) + if toolStartMapHas(suppressedToolStarts, data.ID, data.Name) { + return + } + if len(data.Input) > 0 { + if blockMsg, blocked := a.currentResourcePlaceholderBlock(providers.ToolCall{ID: data.ID, Name: data.Name, Input: data.Input}); blocked { + log.Warn(). + Str("tool", data.Name). + Str("id", data.ID). + Str("reason", blockMsg). + Msg("[AgenticLoop] Suppressed invalid current_resource tool_progress before user-visible execution") + emitCurrentResourceBlock(providers.ToolCall{ID: data.ID, Name: data.Name, Input: data.Input}, blockMsg) + return + } + } + if !toolStartMapHas(visibleToolStarts, data.ID, data.Name) { + attemptEmittedVisibleEvents = true + emitToolStartEvent(callback, data.ID, data.Name, data.Input) + markToolStartMap(visibleToolStarts, data.ID, data.Name) + } + attemptEmittedVisibleEvents = true + emitToolProgressEventWithRawInput( + callback, + data.ID, + data.Name, + data.Input, + data.RawInput, + firstNonEmptyTrimmed(data.Phase, "pending"), + data.Message, + ) + } + case "done": if data, ok := event.Data.(providers.DoneEvent); ok { attemptSawDone = true diff --git a/internal/ai/chat/agentic_test.go b/internal/ai/chat/agentic_test.go index 756c5f15f..beb14bee2 100644 --- a/internal/ai/chat/agentic_test.go +++ b/internal/ai/chat/agentic_test.go @@ -227,6 +227,91 @@ func TestAgenticLoop(t *testing.T) { mockProvider.AssertExpectations(t) }) + t.Run("Provider Tool Argument Progress Mutates Pending Tool", func(t *testing.T) { + mockProvider := &MockProvider{} + loop := NewAgenticLoop(mockProvider, executor, "You are a helper") + ctx := context.Background() + sessionID := "test-session" + messages := []Message{{Role: "user", Content: "List matching resources"}} + + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + return len(req.Messages) == 1 + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + callback := args.Get(2).(providers.StreamCallback) + callback(providers.StreamEvent{ + Type: "tool_start", + Data: providers.ToolStartEvent{ID: "call_query", Name: "pulse_query"}, + }) + callback(providers.StreamEvent{ + Type: "tool_progress", + Data: providers.ToolProgressEvent{ + ID: "call_query", + Name: "pulse_query", + RawInput: `{"action":"sea`, + Phase: "pending", + Message: "Receiving tool input.", + }, + }) + callback(providers.StreamEvent{ + Type: "tool_progress", + Data: providers.ToolProgressEvent{ + ID: "call_query", + Name: "pulse_query", + Input: map[string]interface{}{"action": "search", "query": "prowlarr"}, + RawInput: `{"action":"search","query":"prowlarr"}`, + Phase: "pending", + Message: "Prepared tool input.", + }, + }) + callback(providers.StreamEvent{ + Type: "done", + Data: providers.DoneEvent{ + ToolCalls: []providers.ToolCall{ + { + ID: "call_query", + Name: "pulse_query", + Input: map[string]interface{}{"action": "search", "query": "prowlarr"}, + }, + }, + }, + }) + }).Once() + + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + return len(req.Messages) == 3 + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + callback := args.Get(2).(providers.StreamCallback) + callback(providers.StreamEvent{ + Type: "content", + Data: providers.ContentEvent{Text: "Prowlarr is present."}, + }) + callback(providers.StreamEvent{ + Type: "done", + Data: providers.DoneEvent{}, + }) + }).Once() + + var progressEvents []ToolProgressData + _, err := loop.Execute(ctx, sessionID, messages, func(event StreamEvent) { + if event.Type != "tool_progress" { + return + } + var data ToolProgressData + _ = json.Unmarshal(event.Data, &data) + progressEvents = append(progressEvents, data) + }) + + require.NoError(t, err) + require.GreaterOrEqual(t, len(progressEvents), 3) + assert.Equal(t, "Receiving tool input.", progressEvents[0].Message) + assert.Equal(t, `{"action":"sea`, progressEvents[0].RawInput) + assert.Equal(t, "Prepared tool input.", progressEvents[1].Message) + assert.JSONEq(t, `{"action":"search","query":"prowlarr"}`, progressEvents[1].Input) + assert.Equal(t, "running", progressEvents[2].Phase) + assert.Equal(t, "call_query", progressEvents[2].ID) + mockProvider.AssertExpectations(t) + }) + t.Run("Abort Session", func(t *testing.T) { mockProvider := &MockProvider{} loop := NewAgenticLoop(mockProvider, executor, "You are a helper") diff --git a/internal/ai/providers/openai.go b/internal/ai/providers/openai.go index 23069138d..2f1d2e66d 100644 --- a/internal/ai/providers/openai.go +++ b/internal/ai/providers/openai.go @@ -742,6 +742,17 @@ type openaiStreamToolCallBuilder struct { args strings.Builder } +func parseOpenAIStreamToolInput(rawArgs string) map[string]interface{} { + if strings.TrimSpace(rawArgs) == "" { + return nil + } + var input map[string]interface{} + if err := json.Unmarshal([]byte(rawArgs), &input); err != nil { + return nil + } + return input +} + func finalizeOpenAIStreamToolCalls(builders map[int]*openaiStreamToolCallBuilder) []ToolCall { if len(builders) == 0 { return nil @@ -1082,9 +1093,40 @@ func (c *OpenAIClient) ChatStream(ctx context.Context, req ChatRequest, callback Name: builder.name, }, }) + if rawArgs := builder.args.String(); rawArgs != "" { + callback(StreamEvent{ + Type: "tool_progress", + Data: ToolProgressEvent{ + ID: builder.id, + Name: builder.name, + Input: parseOpenAIStreamToolInput(rawArgs), + RawInput: rawArgs, + Phase: "pending", + Message: "Receiving tool input.", + }, + }) + } } if tc.Function.Arguments != "" { builder.args.WriteString(tc.Function.Arguments) + if builder.name != "" { + rawArgs := builder.args.String() + message := "Receiving tool input." + if parseOpenAIStreamToolInput(rawArgs) != nil { + message = "Prepared tool input." + } + callback(StreamEvent{ + Type: "tool_progress", + Data: ToolProgressEvent{ + ID: builder.id, + Name: builder.name, + Input: parseOpenAIStreamToolInput(rawArgs), + RawInput: rawArgs, + Phase: "pending", + Message: message, + }, + }) + } } } } diff --git a/internal/ai/providers/openai_test.go b/internal/ai/providers/openai_test.go index 006228db5..eb0eab00b 100644 --- a/internal/ai/providers/openai_test.go +++ b/internal/ai/providers/openai_test.go @@ -419,12 +419,18 @@ func TestOpenAIClient_ChatStream_ToolCall(t *testing.T) { var toolCalls []ToolCall var toolStartIndex int + var toolProgressEvents []ToolProgressEvent callback := func(event StreamEvent) { t.Logf("Received event type: %s", event.Type) if event.Type == "tool_start" { toolStartIndex++ } + if event.Type == "tool_progress" { + if data, ok := event.Data.(ToolProgressEvent); ok { + toolProgressEvents = append(toolProgressEvents, data) + } + } if event.Type == "done" { if data, ok := event.Data.(DoneEvent); ok { t.Logf("Received DONE event with %d tool calls", len(data.ToolCalls)) @@ -440,6 +446,15 @@ func TestOpenAIClient_ChatStream_ToolCall(t *testing.T) { // Check that we got a tool_start event assert.Equal(t, 1, toolStartIndex, "Should have received 1 tool_start event") + require.Len(t, toolProgressEvents, 2, "tool argument deltas should stream as progress events") + assert.Equal(t, "call_123", toolProgressEvents[0].ID) + assert.Equal(t, "get_weather", toolProgressEvents[0].Name) + assert.Equal(t, "pending", toolProgressEvents[0].Phase) + assert.Equal(t, "Receiving tool input.", toolProgressEvents[0].Message) + assert.Equal(t, `{"loc`, toolProgressEvents[0].RawInput) + assert.Nil(t, toolProgressEvents[0].Input) + assert.Equal(t, "Prepared tool input.", toolProgressEvents[1].Message) + assert.Equal(t, map[string]interface{}{"location": "NYC"}, toolProgressEvents[1].Input) // Check accumulated tool calls in done event require.Len(t, toolCalls, 1) diff --git a/internal/ai/providers/provider.go b/internal/ai/providers/provider.go index befa11275..45a19b543 100644 --- a/internal/ai/providers/provider.go +++ b/internal/ai/providers/provider.go @@ -165,7 +165,7 @@ type Provider interface { // StreamEvent represents a streaming event from the AI provider type StreamEvent struct { - Type string // "content", "thinking", "tool_start", "tool_end", "done", "error" + Type string // "content", "thinking", "tool_start", "tool_progress", "tool_end", "done", "error" Data interface{} // Type-specific data } @@ -193,6 +193,25 @@ func (e ToolStartEvent) NormalizeCollections() ToolStartEvent { return e } +// ToolProgressEvent is the data for "tool_progress" stream events. +// Providers use this to mutate an already visible pending tool row while the +// model is still streaming tool arguments or while execution advances. +type ToolProgressEvent struct { + ID string `json:"id"` + Name string `json:"name"` + Input map[string]interface{} `json:"input,omitempty"` + RawInput string `json:"raw_input,omitempty"` + Phase string `json:"phase,omitempty"` + Message string `json:"message,omitempty"` +} + +func (e ToolProgressEvent) NormalizeCollections() ToolProgressEvent { + if e.Input == nil { + e.Input = map[string]interface{}{} + } + return e +} + // ToolEndEvent is the data for "tool_end" stream events type ToolEndEvent struct { ID string `json:"id"`