Fix: Track Responses API tool calls from added, exclude incomplete items

core/pipeline/extensions.go

@@ -238,16 +238,18 @@ type InferenceExtension struct {
 	FinishReason string `json:"finishReason,omitempty"`
 
 	// ToolCalls are the tool invocations the model requested. Populated on
-	// three of the four response paths — both non-streaming dialects and
-	// Anthropic streaming. An OpenAI *stream* leaves it empty: that dialect
-	// splits each call across `choices[].delta.tool_calls[]` fragments keyed
-	// by their own index, a shape the streaming chunk decoder does not read.
+	// every response path except an OpenAI Chat Completions *stream*: that
+	// dialect splits each call across `choices[].delta.tool_calls[]`
+	// fragments keyed by their own index, a shape the streaming chunk
+	// decoder does not read. Every other path — both non-streaming
+	// dialects, Anthropic streaming, and Responses API streaming — reads or
+	// assembles the call whole.
 	//
-	// So empty means "the model requested no tools" only for a non-streaming
-	// response or an Anthropic stream. A consumer that spans dialects — cost
-	// accounting, per-tool attribution — must not read absence as a negative
-	// on a streamed OpenAI turn, where it is indistinguishable from a turn
-	// whose calls were never captured.
+	// So empty means "the model requested no tools" everywhere except a
+	// streamed OpenAI Chat Completions turn. A consumer that spans dialects
+	// — cost accounting, per-tool attribution — must not read absence as a
+	// negative there, where it is indistinguishable from a turn whose calls
+	// were never captured.
 	ToolCalls []InferenceToolCall `json:"toolCalls,omitempty"`
 
 	// Legacy aggregates, derived from the split fields below via
@@ -384,9 +386,17 @@ type InferenceTool struct {
 }
 
 // InferenceToolCall is a tool invocation the model emitted in its response.
-// Arguments is the raw JSON string as returned by the LLM (often needs
-// json.Unmarshal by the caller) — kept as a string so malformed output
+// Arguments is usually the raw JSON string as returned by the LLM (often
+// needs json.Unmarshal by the caller) — kept as a string so malformed output
 // from the model doesn't prevent capture.
+//
+// NOT ALWAYS JSON, though: the Responses API's "custom_tool_call" items
+// (confirmed on live Codex traffic — its own `exec` tool is this type, see
+// core/plugins/inferenceparser/responses.go) carry freeform text here, by
+// design of that tool type, not malformed output. A consumer that calls
+// json.Unmarshal on Arguments unconditionally — opa and sparc both do today
+// — must handle that failure for this dialect rather than treat it as bad
+// data.
 type InferenceToolCall struct {
 	ID        string `json:"id,omitempty"`
 	Name      string `json:"name"`

core/plugins/inferenceparser/plugin.go

@@ -244,11 +244,20 @@ func (p *InferenceParser) OnResponse(_ context.Context, pctx *pipeline.Context)
 // distinguishable by the block index the provider stamps on each frame.
 // openTool is the fallback for a provider that omits the index.
 //
-// responsesToolCalls holds the Responses API's tool calls instead — a
-// different shape with the same destination. foldResponsesFrame reads a
-// completed call whole off a single response.output_item.done event (see
-// responses.go), so there is nothing to assemble: no index map, no
-// in-progress fallback, just a plain append-as-they-complete list.
+// responsesToolOrder/responsesToolIndex track the Responses API's tool
+// calls instead — a different shape with the same destination. Unlike
+// Anthropic's tool_use blocks, a Responses API call is announced on
+// response.output_item.added (call_id + name, no arguments yet) and
+// restated whole on response.output_item.done (see responses.go), so
+// tracking needs an id-keyed lookup to update the "added" placeholder in
+// place rather than a plain append-on-done list — a plain list would
+// either double-count a call that gets both events, or (if done only
+// updated by index) drop a call that announces "added" but never reaches
+// "done" at all, e.g. a stream cut short by client cancellation.
+// responsesToolOrder preserves emission order for finalize; each entry's
+// excluded flag is set when output_item.done marks the item
+// "incomplete"/"in_progress" (see responsesOutputItem's Status doc) and
+// makes finalize skip it instead of recording cut-off arguments.
 type inferenceStreamState struct {
 	completion strings.Builder
 	usage      parsercommon.TokenUsage
@@ -258,7 +267,64 @@ type inferenceStreamState struct {
 	toolsByIndex map[int]*anthropicToolCallState
 	openTool     *anthropicToolCallState
 
-	responsesToolCalls []pipeline.InferenceToolCall
+	responsesToolOrder []*responsesToolCallState
+	responsesToolIndex map[string]*responsesToolCallState
+}
+
+// responsesToolCallState is one Responses API tool call tracked across its
+// output_item.added/output_item.done lifecycle — see inferenceStreamState's
+// doc for why an id-keyed entry is needed instead of a plain list.
+type responsesToolCallState struct {
+	call     pipeline.InferenceToolCall
+	excluded bool
+}
+
+// addResponsesToolCall records a tool call's call_id and name as soon as
+// response.output_item.added announces it, before its arguments are known.
+// If the stream ends without a matching output_item.done — a cancelled
+// Codex turn — this placeholder (name known, arguments empty) is what
+// finalize emits, instead of losing the call entirely. A duplicate added
+// for the same call_id (not expected on live traffic, but cheap to guard)
+// is a no-op rather than resetting the entry.
+func (s *inferenceStreamState) addResponsesToolCall(callID, name string) {
+	if s.responsesToolIndex == nil {
+		s.responsesToolIndex = make(map[string]*responsesToolCallState)
+	}
+	if _, exists := s.responsesToolIndex[callID]; exists {
+		return
+	}
+	entry := &responsesToolCallState{call: pipeline.InferenceToolCall{ID: callID, Name: name}}
+	s.responsesToolIndex[callID] = entry
+	s.responsesToolOrder = append(s.responsesToolOrder, entry)
+}
+
+// finishResponsesToolCall fills in a tool call's arguments once
+// response.output_item.done delivers them whole, updating the placeholder
+// addResponsesToolCall created rather than appending a second entry — this
+// is what keeps a call seen on both added and done from being double
+// counted. When status marks the item "incomplete" or "in_progress", the
+// entry is excluded instead: its captured arguments were cut off mid-write,
+// so recording them would hand a consumer a call that looks complete but
+// silently isn't. Falls back to creating the entry here if no
+// output_item.added was seen for this call_id — not expected on live
+// traffic (added always precedes done), but tolerated rather than dropping
+// the call.
+func (s *inferenceStreamState) finishResponsesToolCall(callID, name, arguments, status string) {
+	if s.responsesToolIndex == nil {
+		s.responsesToolIndex = make(map[string]*responsesToolCallState)
+	}
+	entry, ok := s.responsesToolIndex[callID]
+	if !ok {
+		entry = &responsesToolCallState{}
+		s.responsesToolIndex[callID] = entry
+		s.responsesToolOrder = append(s.responsesToolOrder, entry)
+	}
+	if status == "incomplete" || status == "in_progress" {
+		entry.excluded = true
+		return
+	}
+	entry.call = pipeline.InferenceToolCall{ID: callID, Name: name, Arguments: arguments}
+	entry.excluded = false
 }
 
 // finalize copies the accumulated stream state onto the public extension
@@ -270,15 +336,20 @@ func (s *inferenceStreamState) finalize(ext *pipeline.InferenceExtension) {
 	if s.hasUsage {
 		s.usage.Fill(ext)
 	}
-	calls := make([]pipeline.InferenceToolCall, 0, len(s.toolCalls)+len(s.responsesToolCalls))
+	calls := make([]pipeline.InferenceToolCall, 0, len(s.toolCalls)+len(s.responsesToolOrder))
 	for _, tc := range s.toolCalls {
 		calls = append(calls, pipeline.InferenceToolCall{
 			ID:        tc.id,
 			Name:      tc.name,
 			Arguments: tc.args.String(),
 		})
 	}
-	calls = append(calls, s.responsesToolCalls...)
+	for _, entry := range s.responsesToolOrder {
+		if entry.excluded {
+			continue
+		}
+		calls = append(calls, entry.call)
+	}
 	if len(calls) == 0 {
 		return
 	}

core/plugins/inferenceparser/responses.go

@@ -344,9 +344,11 @@ type responsesNonStreaming struct {
 // response: output items' text parts -> completion, tool-call items ->
 // ToolCalls (same two shapes as foldResponsesFrame's response.output_item.done
 // case — there's no streaming here to tell "added" from "done", the item
-// just arrives complete), usage -> token counts, status -> a best-effort
-// finish reason (see foldResponsesFrame for why this API has no direct
-// finish_reason/stop_reason equivalent).
+// just arrives complete, so an item marked "incomplete"/"in_progress" is
+// simply skipped rather than tracked and excluded later, see
+// responsesOutputItem's Status doc), usage -> token counts, status -> a
+// best-effort finish reason (see foldResponsesFrame for why this API has no
+// direct finish_reason/stop_reason equivalent).
 func parseResponsesJSON(body []byte, ext *pipeline.InferenceExtension) {
 	var resp responsesNonStreaming
 	if err := json.Unmarshal(body, &resp); err != nil {
@@ -365,10 +367,17 @@ func parseResponsesJSON(body []byte, ext *pipeline.InferenceExtension) {
 			}
 			b.WriteString(part.Text)
 		}
+		incomplete := item.Status == "incomplete" || item.Status == "in_progress"
 		switch item.Type {
 		case "custom_tool_call":
+			if incomplete {
+				continue
+			}
 			calls = append(calls, pipeline.InferenceToolCall{ID: item.CallID, Name: item.Name, Arguments: item.Input})
 		case "function_call":
+			if incomplete {
+				continue
+			}
 			calls = append(calls, pipeline.InferenceToolCall{ID: item.CallID, Name: item.Name, Arguments: item.Arguments})
 		}
 	}
@@ -426,13 +435,25 @@ type responsesStreamEvent struct {
 //     JSON-schema tool, arguments carried in Arguments.
 //
 // A "message" item carries Content instead — its text parts. On the
-// streaming path this struct's Content is never populated: output_text.delta
+// streaming path this struct's Content is never read: output_text.delta
 // events accumulate that text instead, and parsing the same text twice would
 // double it. Content exists on this struct only for the non-streaming path
 // (parseResponsesJSON), which has no deltas to accumulate from and reads the
 // item whole.
+//
+// Status is the item's own completion state — "completed", "incomplete", or
+// "in_progress" per the published schema — distinct from the response-level
+// status foldResponsesFrame reads off the terminal event. A truncated
+// response (hit a token/turn limit, or the client cancelled it) can mark an
+// individual tool-call item "incomplete": its Input/Arguments were cut off
+// mid-write, so capturing them would hand a consumer a call that looks
+// complete but silently isn't. Both call sites below drop such an item
+// rather than record partial arguments. Left empty leniently where absent
+// (older captures, or a dialect variant that omits it) rather than treated
+// as incomplete — the field's absence is not evidence of truncation.
 type responsesOutputItem struct {
 	Type      string `json:"type"`
+	Status    string `json:"status"`
 	CallID    string `json:"call_id"`
 	Name      string `json:"name"`
 	Input     string `json:"input"`     // custom_tool_call's freeform argument text
@@ -444,25 +465,35 @@ type responsesOutputItem struct {
 
 // foldResponsesFrame folds one Responses API SSE event into the running
 // stream state. The completion accumulates from response.output_text.delta
-// events; a completed tool call arrives whole on response.output_item.done
-// (no incremental assembly needed — unlike Anthropic's tool_use blocks,
-// which split call id/name and arguments across separate events, this API's
-// output_item.done restates the full call_id, name, and input/arguments in
-// one event, confirmed on a live Codex turn that ran its `exec` tool); usage
-// and a best-effort finish reason arrive together on whichever terminal
-// event ends the stream — response.completed on the one live sample this
-// file was built from, plus the two other terminal events the published
-// schema documents: response.incomplete (hit a limit, was cancelled) and
-// response.failed (an upstream error). All three carry the same response
-// snapshot shape, so one case handles them identically; OpenAI's own
-// documented example for response.failed shows usage: null, so the
-// hasAny() guard below is what keeps a failure from asserting a zero usage
-// that was never reported. Every other event type in the sequence carries
-// nothing this parser extracts and is silently ignored — not unrecognized,
-// just uninteresting. Unlike foldAnthropicFrame, an unknown type is not
-// logged here: the full event vocabulary was confirmed on live traffic
-// rather than inferred from docs, so anything outside it is more likely a
-// wire change worth its own look than routine.
+// events. A tool call is tracked across TWO events rather than read whole
+// off response.output_item.done alone: output_item.added fires first,
+// carrying call_id and name before arguments exist, and
+// addResponsesToolCall records a placeholder for it. output_item.done then
+// either fills that placeholder in via finishResponsesToolCall — unlike
+// Anthropic's tool_use blocks, which split call id/name and arguments
+// across separate events, this API's output_item.done restates the full
+// call_id, name, and input/arguments in one event, confirmed on a live
+// Codex turn that ran its `exec` tool — or, when the item's own Status
+// marks it "incomplete"/"in_progress", excludes it instead of recording cut-
+// off arguments (see responsesOutputItem's Status doc). Tracking from
+// "added" rather than only "done" is what keeps a call that never reaches
+// "done" at all — a stream cut short mid-call, e.g. by client cancellation
+// — from being silently dropped: finalize still emits the placeholder, call
+// id and name known, arguments empty, rather than losing the call entirely.
+// Usage and a best-effort finish reason arrive together on whichever
+// terminal event ends the stream — response.completed on the one live
+// sample this file was built from, plus the two other terminal events the
+// published schema documents: response.incomplete (hit a limit, was
+// cancelled) and response.failed (an upstream error). All three carry the
+// same response snapshot shape, so one case handles them identically;
+// OpenAI's own documented example for response.failed shows usage: null, so
+// the hasAny() guard below is what keeps a failure from asserting a zero
+// usage that was never reported. Every other event type in the sequence
+// carries nothing this parser extracts and is silently ignored — not
+// unrecognized, just uninteresting. Unlike foldAnthropicFrame, an unknown
+// type is not logged here: the full event vocabulary was confirmed on live
+// traffic rather than inferred from docs, so anything outside it is more
+// likely a wire change worth its own look than routine.
 func foldResponsesFrame(frame []byte, state *inferenceStreamState, ext *pipeline.InferenceExtension) {
 	var ev responsesStreamEvent
 	if err := json.Unmarshal(frame, &ev); err != nil {
@@ -475,23 +506,23 @@ func foldResponsesFrame(frame []byte, state *inferenceStreamState, ext *pipeline
 		if ev.Delta != nil {
 			state.completion.WriteString(*ev.Delta)
 		}
+	case "response.output_item.added":
+		if ev.Item == nil {
+			return
+		}
+		switch ev.Item.Type {
+		case "custom_tool_call", "function_call":
+			state.addResponsesToolCall(ev.Item.CallID, ev.Item.Name)
+		}
 	case "response.output_item.done":
 		if ev.Item == nil {
 			return
 		}
 		switch ev.Item.Type {
 		case "custom_tool_call":
-			state.responsesToolCalls = append(state.responsesToolCalls, pipeline.InferenceToolCall{
-				ID:        ev.Item.CallID,
-				Name:      ev.Item.Name,
-				Arguments: ev.Item.Input,
-			})
+			state.finishResponsesToolCall(ev.Item.CallID, ev.Item.Name, ev.Item.Input, ev.Item.Status)
 		case "function_call":
-			state.responsesToolCalls = append(state.responsesToolCalls, pipeline.InferenceToolCall{
-				ID:        ev.Item.CallID,
-				Name:      ev.Item.Name,
-				Arguments: ev.Item.Arguments,
-			})
+			state.finishResponsesToolCall(ev.Item.CallID, ev.Item.Name, ev.Item.Arguments, ev.Item.Status)
 		}
 	case "response.completed", "response.incomplete", "response.failed":
 		if ev.Response == nil {