Stream Assistant tool argument progress

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"

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

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

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")

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,
+							},
+						})
+					}
 				}
 			}
 		}

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)

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"`