diff --git a/internal/runtime/executor/codex_executor_stream.go b/internal/runtime/executor/codex_executor_stream.go index aedf5ae9..b51e273c 100644 --- a/internal/runtime/executor/codex_executor_stream.go +++ b/internal/runtime/executor/codex_executor_stream.go @@ -107,7 +107,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au }) httpClient := helps.NewUtlsHTTPClient(ctx, e.cfg, auth, 0) - httpClient = reporter.TrackHTTPClient(httpClient) + httpClient = reporter.TrackHTTPClientRoundTripOnly(httpClient) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) @@ -171,6 +171,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au } else if bytes.HasPrefix(line, dataTag) { data := bytes.TrimSpace(line[5:]) data = helps.RestoreCodexMultiAgentV2Response(data, optimizeMultiAgentV2) + observeCodexTokenEvent(reporter, data) translatedLine = append([]byte("data: "), data...) eventType := gjson.GetBytes(data, "type").String() if streamErr, terminalBody, ok := codexTerminalFailureErr(data); ok { @@ -199,14 +200,17 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au switch eventType { case "response.output_item.done": collectCodexOutputItemDone(data, outputItemsByIndex, &outputItemsFallback) - case "response.completed", "response.incomplete": + case "response.completed", "response.incomplete", "response.done": terminalSuccess = true + data = normalizeCodexWebsocketCompletion(data) if detail, ok := helps.ParseCodexUsage(data); ok { reporter.Publish(ctx, detail) + } else { + reporter.EnsurePublished(ctx) } publishCodexImageToolUsage(ctx, reporter, body, data) data = patchCodexCompletedOutput(data, outputItemsByIndex, outputItemsFallback) - if eventType == "response.completed" { + if eventType == "response.completed" || eventType == "response.done" { cacheCodexReasoningReplayFromCompleted(replayScope, data) } translatedLine = append([]byte("data: "), data...) @@ -297,6 +301,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au } else if bytes.HasPrefix(line, dataTag) { data := bytes.TrimSpace(line[5:]) data = helps.RestoreCodexMultiAgentV2Response(data, optimizeMultiAgentV2) + observeCodexTokenEvent(reporter, data) translatedLine = append([]byte("data: "), data...) eventType := gjson.GetBytes(data, "type").String() if streamErr, terminalBody, ok := codexTerminalFailureErr(data); ok { @@ -320,14 +325,17 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au switch eventType { case "response.output_item.done": collectCodexOutputItemDone(data, outputItemsByIndex, &outputItemsFallback) - case "response.completed", "response.incomplete": + case "response.completed", "response.incomplete", "response.done": terminalSuccess = true + data = normalizeCodexWebsocketCompletion(data) if detail, ok := helps.ParseCodexUsage(data); ok { reporter.Publish(ctx, detail) + } else { + reporter.EnsurePublished(ctx) } publishCodexImageToolUsage(ctx, reporter, body, data) data = patchCodexCompletedOutput(data, outputItemsByIndex, outputItemsFallback) - if eventType == "response.completed" { + if eventType == "response.completed" || eventType == "response.done" { cacheCodexReasoningReplayFromCompleted(replayScope, data) } translatedLine = append([]byte("data: "), data...) diff --git a/internal/runtime/executor/codex_executor_terminal.go b/internal/runtime/executor/codex_executor_terminal.go index be69833c..8a934c86 100644 --- a/internal/runtime/executor/codex_executor_terminal.go +++ b/internal/runtime/executor/codex_executor_terminal.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" "github.com/tidwall/gjson" "github.com/tidwall/sjson" ) @@ -427,6 +428,11 @@ func isCodexHandshakeMetadataEvent(eventType string) bool { } } +// observeCodexTokenEvent inspects a stream payload and marks TTFT on the first substantive token event. +func observeCodexTokenEvent(reporter *helps.UsageReporter, payload []byte) { + helps.ObserveResponsesTokenEvent(reporter, payload) +} + // newCodexBootstrapOverloadErr reports a buffered overload rejection with its real status. // // The status is deliberately produced here instead of in codexTerminalFailureStatus: that mapping diff --git a/internal/runtime/executor/codex_websockets_execute.go b/internal/runtime/executor/codex_websockets_execute.go index 548b5b25..a7daf3d0 100644 --- a/internal/runtime/executor/codex_websockets_execute.go +++ b/internal/runtime/executor/codex_websockets_execute.go @@ -283,7 +283,7 @@ func (e *CodexWebsocketsExecutor) Execute(ctx context.Context, auth *cliproxyaut if len(payload) == 0 { continue } - reporter.MarkFirstResponseByte() + observeCodexTokenEvent(reporter, payload) payload = applyCodexIdentityConfuseResponsePayload(payload, identityState) helps.AppendCodexAPIWebsocketResponse(ctx, e.cfg, payload) helps.EmitWebSocketResponseEvent(ctx, opts, auth, e.Identifier(), req.Model, payload) @@ -315,11 +315,15 @@ func (e *CodexWebsocketsExecutor) Execute(ctx context.Context, auth *cliproxyaut switch eventType { case "response.output_item.done": collectCodexOutputItemDone(payload, outputItemsByIndex, &outputItemsFallback) - case "response.completed": + case "response.completed", "response.done", "response.incomplete": payload = patchCodexCompletedOutput(payload, outputItemsByIndex, outputItemsFallback) - cacheCodexReasoningReplayFromCompleted(replayScope, payload) + if eventType != "response.incomplete" { + cacheCodexReasoningReplayFromCompleted(replayScope, payload) + } if detail, ok := helps.ParseCodexUsage(payload); ok { reporter.Publish(ctx, detail) + } else { + reporter.EnsurePublished(ctx) } var param any clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) diff --git a/internal/runtime/executor/codex_websockets_stream.go b/internal/runtime/executor/codex_websockets_stream.go index 7915e971..a158118a 100644 --- a/internal/runtime/executor/codex_websockets_stream.go +++ b/internal/runtime/executor/codex_websockets_stream.go @@ -325,7 +325,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr if len(payload) == 0 { continue } - reporter.MarkFirstResponseByte() + observeCodexTokenEvent(reporter, payload) payload = applyCodexIdentityConfuseResponsePayload(payload, identityState) helps.AppendCodexAPIWebsocketResponse(ctx, e.cfg, payload) helps.EmitWebSocketResponseEvent(ctx, opts, auth, e.Identifier(), req.Model, payload) @@ -387,28 +387,35 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr } eventType := gjson.GetBytes(payload, "type").String() - isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "error" + isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" || eventType == "response.failed" || eventType == "error" if eventType == "response.output_item.done" { collectCodexOutputItemDone(payload, outputItemsByIndex, &outputItemsFallback) } completedPayload := payload - if eventType == "response.completed" || eventType == "response.done" { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { completedPayload = normalizeCodexWebsocketCompletion(completedPayload) completedPayload = patchCodexCompletedOutput(completedPayload, outputItemsByIndex, outputItemsFallback) - cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) + if eventType != "response.incomplete" { + cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) + } if detail, ok := helps.ParseCodexUsage(completedPayload); ok { reporter.Publish(ctx, detail) + } else { + reporter.EnsurePublished(ctx) } } var currentChunks [][]byte if cliproxyexecutor.DownstreamWebsocket(ctx) { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { + payload = completedPayload + } clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) downstreamPayload := helps.EnsureResponsesUsageDetails(clientPayload) currentChunks = [][]byte{downstreamPayload} } else { payload = normalizeCodexWebsocketCompletion(payload) - if eventType == "response.completed" || eventType == "response.done" { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { payload = completedPayload } clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) @@ -519,15 +526,15 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr } if msgType != websocket.TextMessage { if msgType == websocket.BinaryMessage { - err = fmt.Errorf("codex websockets executor: unexpected binary message") + unexpectedErr := fmt.Errorf("codex websockets executor: unexpected binary message") terminateReason = "unexpected_binary" - terminateErr = err - helps.RecordAPIWebsocketError(ctx, e.cfg, "unexpected_binary", err) - reporter.PublishFailure(ctx, err) + terminateErr = unexpectedErr + helps.RecordAPIWebsocketError(ctx, e.cfg, "unexpected_binary", unexpectedErr) + reporter.PublishFailure(ctx, unexpectedErr) if sess != nil { - e.invalidateUpstreamConn(sess, conn, "unexpected_binary", err) + e.invalidateUpstreamConn(sess, conn, "unexpected_binary", unexpectedErr) } - _ = send(cliproxyexecutor.StreamChunk{Err: err}) + _ = send(cliproxyexecutor.StreamChunk{Err: unexpectedErr}) return } continue @@ -537,7 +544,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr if len(payload) == 0 { continue } - reporter.MarkFirstResponseByte() + observeCodexTokenEvent(reporter, payload) payload = applyCodexIdentityConfuseResponsePayload(payload, identityState) helps.AppendCodexAPIWebsocketResponse(ctx, e.cfg, payload) helps.EmitWebSocketResponseEvent(ctx, opts, auth, e.Identifier(), req.Model, payload) @@ -582,22 +589,29 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr } eventType := gjson.GetBytes(payload, "type").String() - isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "error" + isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" || eventType == "response.failed" || eventType == "error" if eventType == "response.output_item.done" { collectCodexOutputItemDone(payload, outputItemsByIndex, &outputItemsFallback) } completedPayload := payload - if eventType == "response.completed" || eventType == "response.done" { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { completedPayload = normalizeCodexWebsocketCompletion(completedPayload) completedPayload = patchCodexCompletedOutput(completedPayload, outputItemsByIndex, outputItemsFallback) - cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) + if eventType != "response.incomplete" { + cacheCodexReasoningReplayFromCompleted(replayScope, completedPayload) + } if detail, ok := helps.ParseCodexUsage(completedPayload); ok { reporter.Publish(ctx, detail) + } else { + reporter.EnsurePublished(ctx) } } clientPayload := applyCodexIdentityExposeResponsePayload(payload, identityState) if cliproxyexecutor.DownstreamWebsocket(ctx) { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { + clientPayload = applyCodexIdentityExposeResponsePayload(completedPayload, identityState) + } downstreamPayload := helps.EnsureResponsesUsageDetails(clientPayload) if !send(cliproxyexecutor.StreamChunk{Payload: downstreamPayload}) { terminateReason = "context_done" @@ -611,7 +625,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr } payload = normalizeCodexWebsocketCompletion(payload) - if eventType == "response.completed" || eventType == "response.done" { + if eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { payload = completedPayload } eventType = gjson.GetBytes(payload, "type").String() @@ -625,7 +639,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr return } } - if eventType == "response.completed" || eventType == "response.done" { + if isTerminalEvent || eventType == "response.completed" || eventType == "response.done" || eventType == "response.incomplete" { return } } diff --git a/internal/runtime/executor/helps/chat_ttft_helpers.go b/internal/runtime/executor/helps/chat_ttft_helpers.go new file mode 100644 index 00000000..ceddd0dc --- /dev/null +++ b/internal/runtime/executor/helps/chat_ttft_helpers.go @@ -0,0 +1,113 @@ +package helps + +import ( + "bytes" + + "github.com/tidwall/gjson" +) + +// IsChatTokenEvent reports whether the given OpenAI-compatible Chat Completions SSE or JSON chunk +// carries substantive output token content (such as text delta, reasoning content, or tool call arguments). +// It filters out container metadata, role announcements, and empty delta frames. +func IsChatTokenEvent(payload []byte) bool { + payload = bytes.TrimSpace(payload) + if len(payload) == 0 { + return false + } + + // Handle SSE stream terminal marker + if bytes.Equal(payload, []byte("data: [DONE]")) || bytes.Equal(payload, []byte("[DONE]")) { + return true + } + + // Handle SSE line format (e.g., "data: {...}") + if bytes.HasPrefix(payload, []byte("data:")) { + prefixLen := len("data:") + if bytes.HasPrefix(payload, []byte("data: ")) { + prefixLen = len("data: ") + } + payload = bytes.TrimSpace(payload[prefixLen:]) + if len(payload) == 0 { + return false + } + if bytes.Equal(payload, []byte("[DONE]")) { + return true + } + } + + // Terminal error envelope fallback + if gjson.GetBytes(payload, "error.message").Exists() || gjson.GetBytes(payload, "error").Exists() { + return true + } + + choices := gjson.GetBytes(payload, "choices").Array() + if len(choices) == 0 { + return false + } + + for _, choice := range choices { + // Streaming delta checks + delta := choice.Get("delta") + if delta.Exists() { + if len(delta.Get("content").String()) > 0 { + return true + } + if len(delta.Get("reasoning_content").String()) > 0 { + return true + } + if len(delta.Get("reasoning").String()) > 0 { + return true + } + if len(delta.Get("refusal").String()) > 0 { + return true + } + for _, tc := range delta.Get("tool_calls").Array() { + if len(tc.Get("function.arguments").String()) > 0 || len(tc.Get("function.name").String()) > 0 { + return true + } + if len(tc.Get("custom.input").String()) > 0 { + return true + } + } + } + + // Non-streaming / completed message fallback + message := choice.Get("message") + if message.Exists() { + if len(message.Get("content").String()) > 0 { + return true + } + if len(message.Get("reasoning_content").String()) > 0 { + return true + } + if len(message.Get("refusal").String()) > 0 { + return true + } + for _, tc := range message.Get("tool_calls").Array() { + if len(tc.Get("function.arguments").String()) > 0 || len(tc.Get("function.name").String()) > 0 { + return true + } + } + } + + // Finish reason terminal fallback + if len(choice.Get("finish_reason").String()) > 0 { + return true + } + } + + return false +} + +// ObserveChatTokenEvent inspects an OpenAI Chat Completions chunk and records TTFT if the frame +// represents the first meaningful token event. It records first-packet arrival time as fallback +// and returns immediately with zero allocations once effective token TTFT is set. +func ObserveChatTokenEvent(reporter *UsageReporter, payload []byte) { + if reporter == nil || len(payload) == 0 { + return + } + if reporter.IsTTFTSet() { + return + } + reporter.ObserveTokenEvent(IsChatTokenEvent(payload)) +} diff --git a/internal/runtime/executor/helps/claude_ttft_helpers.go b/internal/runtime/executor/helps/claude_ttft_helpers.go new file mode 100644 index 00000000..a903d88e --- /dev/null +++ b/internal/runtime/executor/helps/claude_ttft_helpers.go @@ -0,0 +1,101 @@ +package helps + +import ( + "bytes" + + "github.com/tidwall/gjson" +) + +// IsClaudeTokenEvent reports whether an Anthropic Messages SSE chunk carries +// substantive output token content (such as text delta, thinking delta, or tool input delta). +// It filters out message_start, ping, content_block_stop, and empty containers. +func IsClaudeTokenEvent(payload []byte) bool { + payload = bytes.TrimSpace(payload) + if len(payload) == 0 { + return false + } + + // Strip "event: ..." line if present in multi-line SSE buffer + if bytes.HasPrefix(payload, []byte("event:")) { + if idx := bytes.IndexByte(payload, '\n'); idx != -1 { + payload = bytes.TrimSpace(payload[idx+1:]) + } + } + + // Handle SSE data line prefix (e.g., "data: {...}") + if bytes.HasPrefix(payload, []byte("data:")) { + prefixLen := len("data:") + if bytes.HasPrefix(payload, []byte("data: ")) { + prefixLen = len("data: ") + } + payload = bytes.TrimSpace(payload[prefixLen:]) + if len(payload) == 0 { + return false + } + } + + eventType := gjson.GetBytes(payload, "type").String() + switch eventType { + // Substantive streaming block deltas + case "content_block_delta": + delta := gjson.GetBytes(payload, "delta") + if len(delta.Get("text").String()) > 0 { + return true + } + if len(delta.Get("thinking").String()) > 0 { + return true + } + if len(delta.Get("partial_json").String()) > 0 { + return true + } + if len(delta.Get("signature").String()) > 0 { + return true + } + return false + + // Initial block start when populated upfront + case "content_block_start": + cb := gjson.GetBytes(payload, "content_block") + if len(cb.Get("text").String()) > 0 || len(cb.Get("thinking").String()) > 0 { + return true + } + return false + + // Terminal completion events fallback + case "message_delta": + return len(gjson.GetBytes(payload, "delta.stop_reason").String()) > 0 + case "message_stop", "error": + return true + + // Handshake metadata and container boundaries + case "message_start", "ping", "content_block_stop": + return false + + default: + // Non-streaming / complete message payload fallback + if gjson.GetBytes(payload, "content").IsArray() { + for _, block := range gjson.GetBytes(payload, "content").Array() { + if len(block.Get("text").String()) > 0 || len(block.Get("thinking").String()) > 0 { + return true + } + if block.Get("type").String() == "tool_use" && len(block.Get("name").String()) > 0 { + return true + } + } + } + return false + } +} + +// ObserveClaudeTokenEvent inspects an Anthropic Messages SSE chunk and records TTFT if the frame +// represents the first meaningful token event. It records first-packet arrival time as fallback +// and returns immediately with zero allocations once effective token TTFT is set. +func ObserveClaudeTokenEvent(reporter *UsageReporter, payload []byte) { + if reporter == nil || len(payload) == 0 { + return + } + if reporter.IsTTFTSet() { + return + } + reporter.ObserveTokenEvent(IsClaudeTokenEvent(payload)) +} diff --git a/internal/runtime/executor/helps/gemini_ttft_helpers.go b/internal/runtime/executor/helps/gemini_ttft_helpers.go new file mode 100644 index 00000000..005f82bf --- /dev/null +++ b/internal/runtime/executor/helps/gemini_ttft_helpers.go @@ -0,0 +1,81 @@ +package helps + +import ( + "bytes" + + "github.com/tidwall/gjson" +) + +// IsGeminiTokenEvent reports whether a Google Gemini / Antigravity streaming chunk carries +// substantive output token content (such as text parts, thought traces, or function calls). +// It filters out metadata-only chunks (e.g. standalone usageMetadata) and empty parts. +func IsGeminiTokenEvent(payload []byte) bool { + payload = bytes.TrimSpace(payload) + if len(payload) == 0 { + return false + } + + // Strip "data:" SSE prefix if present + if bytes.HasPrefix(payload, []byte("data:")) { + prefixLen := len("data:") + if bytes.HasPrefix(payload, []byte("data: ")) { + prefixLen = len("data: ") + } + payload = bytes.TrimSpace(payload[prefixLen:]) + if len(payload) == 0 { + return false + } + } + + // Terminal error fallback + if gjson.GetBytes(payload, "error.message").Exists() || gjson.GetBytes(payload, "error").Exists() { + return true + } + + candidates := gjson.GetBytes(payload, "candidates").Array() + if len(candidates) == 0 { + return false + } + + for _, candidate := range candidates { + // Inspect content parts + parts := candidate.Get("content.parts").Array() + for _, part := range parts { + if len(part.Get("text").String()) > 0 { + return true + } + if len(part.Get("thoughtText").String()) > 0 { + return true + } + if thought := part.Get("thought"); thought.Type == gjson.String && len(thought.String()) > 0 { + return true + } + if len(part.Get("functionCall.name").String()) > 0 { + return true + } + if len(part.Get("inlineData.data").String()) > 0 { + return true + } + } + + // Terminal finishReason fallback (e.g., STOP, MAX_TOKENS, SAFETY) + if len(candidate.Get("finishReason").String()) > 0 { + return true + } + } + + return false +} + +// ObserveGeminiTokenEvent inspects a Google Gemini / Antigravity streaming chunk and records TTFT +// if the frame represents the first meaningful token event. It records first-packet arrival time +// as fallback and returns immediately with zero allocations once effective token TTFT is set. +func ObserveGeminiTokenEvent(reporter *UsageReporter, payload []byte) { + if reporter == nil || len(payload) == 0 { + return + } + if reporter.IsTTFTSet() { + return + } + reporter.ObserveTokenEvent(IsGeminiTokenEvent(payload)) +} diff --git a/internal/runtime/executor/helps/responses_ttft_helpers.go b/internal/runtime/executor/helps/responses_ttft_helpers.go new file mode 100644 index 00000000..b67ca7bc --- /dev/null +++ b/internal/runtime/executor/helps/responses_ttft_helpers.go @@ -0,0 +1,118 @@ +package helps + +import ( + "bytes" + + "github.com/tidwall/gjson" +) + +// IsResponsesTokenEvent reports whether the given OpenAI/xAI Responses API WebSocket or SSE event +// payload carries an actual output token, reasoning trace, tool call argument, or multimodal delta, +// according to the official Responses API streaming / websocket specification plus Codex-compatible private events. +func IsResponsesTokenEvent(payload []byte) bool { + payload = bytes.TrimSpace(payload) + if len(payload) == 0 { + return false + } + + // Handle SSE line format (e.g., "data: {...}") + if bytes.HasPrefix(payload, []byte("data:")) { + prefixLen := len("data:") + if bytes.HasPrefix(payload, []byte("data: ")) { + prefixLen = len("data: ") + } + payload = bytes.TrimSpace(payload[prefixLen:]) + if len(payload) == 0 { + return false + } + } + + eventType := gjson.GetBytes(payload, "type").String() + switch eventType { + // Substantive text, reasoning, and tool argument streaming deltas + case "response.reasoning_summary_text.delta", + "response.reasoning.delta", + "response.reasoning_text.delta", + "response.output_text.delta", + "response.text.delta", + "response.function_call_arguments.delta", + "response.custom_tool_call_input.delta", + "response.code_interpreter_call_code.delta", + "response.mcp_call_arguments.delta", + "response.shell_call_command.delta", + "response.refusal.delta", + "response.audio.transcript.delta": + return len(gjson.GetBytes(payload, "delta").String()) > 0 + + // Multimodal streaming deltas (audio and image generation preview) + case "response.audio.delta": + return len(gjson.GetBytes(payload, "delta").String()) > 0 || len(gjson.GetBytes(payload, "data").String()) > 0 + case "response.image_generation_call.partial_image": + return len(gjson.GetBytes(payload, "partial_image_b64").String()) > 0 + + // Shell call command added event when command is populated upfront + case "response.shell_call_command.added": + return len(gjson.GetBytes(payload, "command").String()) > 0 + + // Single-chunk / non-stream completed items + case "response.reasoning_summary_text.done", + "response.reasoning_text.done", + "response.output_text.done": + return len(gjson.GetBytes(payload, "text").String()) > 0 + case "response.refusal.done": + return len(gjson.GetBytes(payload, "refusal").String()) > 0 + case "response.function_call_arguments.done", + "response.mcp_call_arguments.done": + return len(gjson.GetBytes(payload, "arguments").String()) > 0 + case "response.custom_tool_call_input.done": + return len(gjson.GetBytes(payload, "input").String()) > 0 + case "response.code_interpreter_call_code.done": + return len(gjson.GetBytes(payload, "code").String()) > 0 + case "response.shell_call_command.done": + return len(gjson.GetBytes(payload, "command").String()) > 0 + case "response.reasoning_summary_part.done": + return len(gjson.GetBytes(payload, "part.text").String()) > 0 + case "response.content_part.done": + return len(gjson.GetBytes(payload, "part.text").String()) > 0 || len(gjson.GetBytes(payload, "part.refusal").String()) > 0 + case "response.output_item.done": + itemType := gjson.GetBytes(payload, "item.type").String() + switch itemType { + case "function_call": + return len(gjson.GetBytes(payload, "item.arguments").String()) > 0 + case "custom_tool_call": + return len(gjson.GetBytes(payload, "item.input").String()) > 0 + case "message": + for _, content := range gjson.GetBytes(payload, "item.content").Array() { + if len(content.Get("text").String()) > 0 || len(content.Get("refusal").String()) > 0 { + return true + } + } + } + return false + + // Terminal events fallback to guarantee TTFT is never missed on truncated, failed, or fast turns + case "response.completed", + "response.done", + "response.incomplete", + "response.failed", + "error": + return true + + // All other container lifecycles (added, created, in_progress, search state machines, metadata) + default: + return false + } +} + +// ObserveResponsesTokenEvent inspects a Responses API frame payload and records TTFT if the frame +// represents the first meaningful token event. It records the first packet arrival time as a fallback +// and exits immediately with zero allocations once effective token TTFT is set. +func ObserveResponsesTokenEvent(reporter *UsageReporter, payload []byte) { + if reporter == nil || len(payload) == 0 { + return + } + if reporter.IsTTFTSet() { + return + } + reporter.ObserveTokenEvent(IsResponsesTokenEvent(payload)) +} diff --git a/internal/runtime/executor/helps/responses_ttft_helpers_test.go b/internal/runtime/executor/helps/responses_ttft_helpers_test.go new file mode 100644 index 00000000..4472e37d --- /dev/null +++ b/internal/runtime/executor/helps/responses_ttft_helpers_test.go @@ -0,0 +1,305 @@ +package helps + +import ( + "context" + "testing" +) + +func TestIsResponsesTokenEvent_Classification(t *testing.T) { + tests := []struct { + name string + payload string + want bool + }{ + { + name: "empty payload", + payload: "", + want: false, + }, + { + name: "whitespace only", + payload: " \n\t ", + want: false, + }, + { + name: "codex rate limits metadata", + payload: `{"type":"codex.rate_limits","rate_limits":{"plan_type":"pro"}}`, + want: false, + }, + { + name: "codex response metadata", + payload: `{"type":"codex.response.metadata","etag":"W/\"123\""}`, + want: false, + }, + { + name: "responsesapi websocket timing", + payload: `{"type":"responsesapi.websocket_timing","timing":{"duration_ms":100}}`, + want: false, + }, + { + name: "response created", + payload: `{"type":"response.created","response":{"id":"resp_123","status":"in_progress"}}`, + want: false, + }, + { + name: "response in progress", + payload: `{"type":"response.in_progress","response":{"id":"resp_123","tools":[{"type":"function"}]}}`, + want: false, + }, + { + name: "response output item added with encrypted content only", + payload: `{"type":"response.output_item.added","item":{"type":"reasoning","encrypted_content":"gAAAAAB..."}}`, + want: false, + }, + { + name: "response content part added", + payload: `{"type":"response.content_part.added","part":{"type":"text","text":""}}`, + want: false, + }, + { + name: "response reasoning summary part added", + payload: `{"type":"response.reasoning_summary_part.added","part":{"type":"summary_text","text":""}}`, + want: false, + }, + { + name: "response reasoning summary text delta empty", + payload: `{"type":"response.reasoning_summary_text.delta","delta":""}`, + want: false, + }, + { + name: "response reasoning summary text delta non-empty", + payload: `{"type":"response.reasoning_summary_text.delta","delta":"**Inspecting**"}`, + want: true, + }, + { + name: "response reasoning delta non-empty", + payload: `{"type":"response.reasoning.delta","delta":"Analyzing requirements..."}`, + want: true, + }, + { + name: "response reasoning text delta non-empty", + payload: `{"type":"response.reasoning_text.delta","delta":"Step 1: Check code"}`, + want: true, + }, + { + name: "response output text delta non-empty", + payload: `{"type":"response.output_text.delta","delta":"Hello world"}`, + want: true, + }, + { + name: "response text delta non-empty", + payload: `{"type":"response.text.delta","delta":"Direct text chunk"}`, + want: true, + }, + { + name: "response function call arguments delta non-empty", + payload: `{"type":"response.function_call_arguments.delta","delta":"{\"query\":\"test\"}"}`, + want: true, + }, + { + name: "response custom_tool_call_input delta non-empty", + payload: `{"type":"response.custom_tool_call_input.delta","delta":"{\"param\":1}"}`, + want: true, + }, + { + name: "response code interpreter call code delta non-empty", + payload: `{"type":"response.code_interpreter_call_code.delta","delta":"import math\n"}`, + want: true, + }, + { + name: "response mcp call arguments delta non-empty", + payload: `{"type":"response.mcp_call_arguments.delta","delta":"{\"tool\":\"lookup\"}"}`, + want: true, + }, + { + name: "response shell call command added with non-empty command", + payload: `{"type":"response.shell_call_command.added","command":"ls -la"}`, + want: true, + }, + { + name: "response shell call command added with empty command", + payload: `{"type":"response.shell_call_command.added","command":""}`, + want: false, + }, + { + name: "response shell call command delta non-empty", + payload: `{"type":"response.shell_call_command.delta","delta":"ls -la\n"}`, + want: true, + }, + { + name: "response shell call output content delta is tool execution output, not model token", + payload: `{"type":"response.shell_call_output_content.delta","delta":{"stdout":"output text\n","stderr":""}}`, + want: false, + }, + { + name: "response shell call output content done is tool execution output, not model token", + payload: `{"type":"response.shell_call_output_content.done","output":[]}`, + want: false, + }, + { + name: "response refusal delta non-empty", + payload: `{"type":"response.refusal.delta","delta":"I cannot fulfill this request"}`, + want: true, + }, + { + name: "response audio transcript delta non-empty", + payload: `{"type":"response.audio.transcript.delta","delta":"Spoken text"}`, + want: true, + }, + { + name: "response audio delta non-empty", + payload: `{"type":"response.audio.delta","delta":"UklGRi..."}`, + want: true, + }, + { + name: "response image generation call partial image non-empty", + payload: `{"type":"response.image_generation_call.partial_image","partial_image_b64":"iVBORw0KGgo..."}`, + want: true, + }, + { + name: "response web search call in progress", + payload: `{"type":"response.web_search_call.in_progress"}`, + want: false, + }, + { + name: "response file search call searching", + payload: `{"type":"response.file_search_call.searching"}`, + want: false, + }, + { + name: "response code interpreter call interpreting", + payload: `{"type":"response.code_interpreter_call.interpreting"}`, + want: false, + }, + { + name: "response mcp call in progress", + payload: `{"type":"response.mcp_call.in_progress"}`, + want: false, + }, + { + name: "response output item done function call empty args", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","name":"lookup","arguments":""}}`, + want: false, + }, + { + name: "response output item done function call non-empty args", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","name":"lookup","arguments":"{\"q\":1}"}}`, + want: true, + }, + { + name: "response output item done message empty", + payload: `{"type":"response.output_item.done","item":{"type":"message","content":[]}}`, + want: false, + }, + { + name: "response output item done message with text", + payload: `{"type":"response.output_item.done","item":{"type":"message","content":[{"type":"text","text":"hello"}]}}`, + want: true, + }, + { + name: "response completed fallback", + payload: `{"type":"response.completed","response":{"id":"resp_123","status":"completed"}}`, + want: true, + }, + { + name: "response done fallback", + payload: `{"type":"response.done","response":{"id":"resp_123"}}`, + want: true, + }, + { + name: "response incomplete fallback", + payload: `{"type":"response.incomplete","response":{"id":"resp_123","status":"incomplete"}}`, + want: true, + }, + { + name: "response failed fallback", + payload: `{"type":"response.failed","response":{"id":"resp_123","status":"failed"}}`, + want: true, + }, + { + name: "generic error fallback", + payload: `{"type":"error","error":{"message":"overloaded","code":"rate_limit_exceeded"}}`, + want: true, + }, + // SSE line format verification + { + name: "SSE data line with output text delta", + payload: `data: {"type":"response.output_text.delta","delta":"Hello SSE"}`, + want: true, + }, + { + name: "SSE data line with response created", + payload: `data: {"type":"response.created","response":{"id":"resp_sse"}}`, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsResponsesTokenEvent([]byte(tt.payload)); got != tt.want { + t.Errorf("IsResponsesTokenEvent(%s) = %v, want %v", tt.name, got, tt.want) + } + }) + } +} + +func TestObserveResponsesTokenEvent_Behavior(t *testing.T) { + ctx := context.Background() + reporter := NewUsageReporter(ctx, "codex", "gpt-5.6-luna", nil) + reporter.StartResponseTTFT() + + // 1. Initial state: TTFT not set + if reporter.IsTTFTSet() { + t.Fatalf("expected IsTTFTSet() to be false initially") + } + + // 2. Metadata event records first packet fallback, but does not mark effective TTFT + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"codex.rate_limits","rate_limits":{"plan_type":"pro"}}`)) + if reporter.IsTTFTSet() { + t.Fatalf("metadata event must not trigger IsTTFTSet()") + } + if !reporter.IsFirstPacketSet() { + t.Fatalf("expected IsFirstPacketSet() == true") + } + + // 3. Response created event does not mark effective TTFT + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"response.created","response":{"id":"resp_1"}}`)) + if reporter.IsTTFTSet() { + t.Fatalf("response.created must not trigger IsTTFTSet()") + } + + // 4. Meaningful token delta triggers effective TTFT + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"response.output_text.delta","delta":"First word"}`)) + if !reporter.IsTTFTSet() { + t.Fatalf("response.output_text.delta must trigger IsTTFTSet()") + } + tokenTTFT := reporter.ttftDuration() + if tokenTTFT <= 0 { + t.Fatalf("expected token TTFT > 0, got %v", tokenTTFT) + } + + // 5. Subsequent calls should be fast-path no-ops and not alter TTFT + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"response.output_text.delta","delta":"Second word"}`)) + if reporter.ttftDuration() != tokenTTFT { + t.Fatalf("subsequent ObserveResponsesTokenEvent must not modify already set TTFT") + } +} + +func TestObserveResponsesTokenEvent_FirstPacketFallback(t *testing.T) { + ctx := context.Background() + reporter := NewUsageReporter(ctx, "codex", "gpt-5.6-luna", nil) + reporter.StartResponseTTFT() + + // Only send metadata events, no token deltas + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"codex.rate_limits","rate_limits":{"plan_type":"pro"}}`)) + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"response.created","response":{"id":"resp_1"}}`)) + + if reporter.IsTTFTSet() { + t.Fatalf("expected IsTTFTSet() to be false when no tokens were received") + } + + // First packet fallback must be returned + if !reporter.IsFirstPacketSet() { + t.Fatalf("expected IsFirstPacketSet() == true") + } +} diff --git a/internal/runtime/executor/helps/usage_helpers.go b/internal/runtime/executor/helps/usage_helpers.go index 7313fb33..247e7006 100644 --- a/internal/runtime/executor/helps/usage_helpers.go +++ b/internal/runtime/executor/helps/usage_helpers.go @@ -22,26 +22,28 @@ import ( ) type UsageReporter struct { - provider string - executorType string - model string - alias string - authID string - authIndex string - authMu sync.RWMutex - accessTokenHash string - authType string - apiKey string - source string - reasoning string - serviceTier string - generate bool - requestedAt time.Time - ttftMu sync.RWMutex - ttft time.Duration - ttftStart time.Time - ttftSet bool - once sync.Once + provider string + executorType string + model string + alias string + authID string + authIndex string + authMu sync.RWMutex + accessTokenHash string + authType string + apiKey string + source string + reasoning string + serviceTier string + generate bool + requestedAt time.Time + ttftMu sync.RWMutex + ttft time.Duration + firstPacketDuration time.Duration + firstPacketSet bool + ttftStart time.Time + ttftSet bool + once sync.Once } type usageExecutor interface { @@ -134,6 +136,19 @@ func (r *UsageReporter) SetTranslatedReasoningEffort(payload []byte, format stri } func (r *UsageReporter) TrackHTTPClient(client *http.Client) *http.Client { + return r.trackHTTPClient(client, false) +} + +// TrackHTTPClientRoundTripOnly records the TTFT start time upon sending the request +// and captures first-packet arrival fallback on initial body reads, while keeping +// effective TTFT unset. This allows protocol-aware streaming executors (like Codex SSE) +// to mark effective TTFT explicitly upon receiving substantive token events, while +// preserving first-packet fallback metrics for non-2xx error bodies or keepalive streams. +func (r *UsageReporter) TrackHTTPClientRoundTripOnly(client *http.Client) *http.Client { + return r.trackHTTPClient(client, true) +} + +func (r *UsageReporter) trackHTTPClient(client *http.Client, packetOnly bool) *http.Client { if r == nil || client == nil { return client } @@ -143,8 +158,9 @@ func (r *UsageReporter) TrackHTTPClient(client *http.Client) *http.Client { transport = http.DefaultTransport } tracked.Transport = usageTTFTRoundTripper{ - base: transport, - reporter: r, + base: transport, + reporter: r, + packetOnly: packetOnly, } return &tracked } @@ -162,6 +178,19 @@ func (r *UsageReporter) ObserveResponse(resp *http.Response) { } } +func (r *UsageReporter) ObserveResponsePacketOnly(resp *http.Response) { + if r == nil || resp == nil || resp.Body == nil { + return + } + r.StartResponseTTFT() + resp.Body = &usageTTFTReadCloser{ + ReadCloser: resp.Body, + mark: func() { + r.RecordFirstPacket() + }, + } +} + func (r *UsageReporter) StartResponseTTFT() { if r == nil { return @@ -173,6 +202,70 @@ func (r *UsageReporter) StartResponseTTFT() { r.ttftMu.Unlock() } +func (r *UsageReporter) IsTTFTSet() bool { + if r == nil { + return false + } + r.ttftMu.RLock() + defer r.ttftMu.RUnlock() + return r.ttftSet +} + +func (r *UsageReporter) IsFirstPacketSet() bool { + if r == nil { + return false + } + r.ttftMu.RLock() + defer r.ttftMu.RUnlock() + return r.firstPacketSet +} + +// RecordFirstPacket records the arrival time of the first packet/chunk from upstream as a fallback. +func (r *UsageReporter) RecordFirstPacket() { + r.ObserveTokenEvent(false) +} + +// ObserveTokenEvent records the first packet fallback time on the first frame, +// and if isToken is true, records the effective TTFT. It uses a fast-path +// read check to return immediately with zero write lock contention once TTFT is set +// or when subsequent non-token metadata frames arrive after the first packet. +func (r *UsageReporter) ObserveTokenEvent(isToken bool) { + if r == nil { + return + } + r.ttftMu.RLock() + if r.ttftSet { + r.ttftMu.RUnlock() + return + } + start := r.ttftStart + alreadyRecordedPacket := r.firstPacketSet + r.ttftMu.RUnlock() + + if start.IsZero() { + return + } + + if !isToken && alreadyRecordedPacket { + return + } + + r.ttftMu.Lock() + defer r.ttftMu.Unlock() + if r.ttftSet { + return + } + if !r.firstPacketSet { + r.firstPacketDuration = time.Since(start) + r.firstPacketSet = true + } + if isToken { + r.ttft = time.Since(start) + r.ttftSet = true + r.ttftStart = time.Time{} + } +} + func (r *UsageReporter) MarkFirstResponseByte() { if r == nil { return @@ -353,12 +446,19 @@ func (r *UsageReporter) ttftDuration() time.Duration { } r.ttftMu.RLock() defer r.ttftMu.RUnlock() - return r.ttft + if r.ttftSet { + return r.ttft + } + if r.firstPacketSet { + return r.firstPacketDuration + } + return 0 } type usageTTFTRoundTripper struct { - base http.RoundTripper - reporter *UsageReporter + base http.RoundTripper + reporter *UsageReporter + packetOnly bool } func (t usageTTFTRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { @@ -367,7 +467,11 @@ func (t usageTTFTRoundTripper) RoundTrip(req *http.Request) (*http.Response, err if errRoundTrip != nil { return resp, errRoundTrip } - t.reporter.ObserveResponse(resp) + if t.packetOnly { + t.reporter.ObserveResponsePacketOnly(resp) + } else { + t.reporter.ObserveResponse(resp) + } return resp, nil } @@ -1101,14 +1205,14 @@ func StripUsageMetadataFromJSON(rawJSON []byte) ([]byte, bool) { var changed bool if usageMetadata = gjson.GetBytes(cleaned, "usageMetadata"); usageMetadata.Exists() { - // Rename usageMetadata to cpaUsageMetadata in the message_start event of Claude + // Rename usageMetadata to cpaUsageMetadata cleaned, _ = sjson.SetRawBytes(cleaned, "cpaUsageMetadata", []byte(usageMetadata.Raw)) cleaned, _ = sjson.DeleteBytes(cleaned, "usageMetadata") changed = true } if usageMetadata = gjson.GetBytes(cleaned, "response.usageMetadata"); usageMetadata.Exists() { - // Rename usageMetadata to cpaUsageMetadata in the message_start event of Claude + // Rename usageMetadata to cpaUsageMetadata cleaned, _ = sjson.SetRawBytes(cleaned, "response.cpaUsageMetadata", []byte(usageMetadata.Raw)) cleaned, _ = sjson.DeleteBytes(cleaned, "response.usageMetadata") changed = true diff --git a/internal/runtime/executor/helps/usage_helpers_test.go b/internal/runtime/executor/helps/usage_helpers_test.go index 3fc77295..afc42b89 100644 --- a/internal/runtime/executor/helps/usage_helpers_test.go +++ b/internal/runtime/executor/helps/usage_helpers_test.go @@ -589,6 +589,132 @@ func TestUsageReporterTrackHTTPClientStartsTTFTBeforeRoundTrip(t *testing.T) { } } +func TestUsageReporterTrackHTTPClientRoundTripOnly_DoesNotTriggerOnBodyRead(t *testing.T) { + reporter := NewUsageReporter(context.Background(), "codex", "gpt-5.6-luna", nil) + client := reporter.TrackHTTPClientRoundTripOnly(&http.Client{ + Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("data: {\"type\":\"response.created\"}\n\n")), + Request: req, + }, nil + }), + }) + + req, errNewRequest := http.NewRequestWithContext(context.Background(), http.MethodPost, "https://example.invalid/v1/responses", strings.NewReader("{}")) + if errNewRequest != nil { + t.Fatalf("NewRequestWithContext() error = %v", errNewRequest) + } + resp, errDo := client.Do(req) + if errDo != nil { + t.Fatalf("Do() error = %v", errDo) + } + bodyBytes, errRead := io.ReadAll(resp.Body) + if errRead != nil { + t.Fatalf("ReadAll() error = %v", errRead) + } + if errClose := resp.Body.Close(); errClose != nil { + t.Fatalf("response body close error = %v", errClose) + } + + // 1. Plain body reading must NOT set TTFT + if reporter.IsTTFTSet() { + t.Fatalf("TrackHTTPClientRoundTripOnly must not set TTFT on plain body read") + } + + // 2. Observing metadata event records fallback, but does NOT set effective TTFT + ObserveResponsesTokenEvent(reporter, bodyBytes) + if reporter.IsTTFTSet() { + t.Fatalf("Observing metadata event must not set effective TTFT") + } + if reporter.ttftDuration() <= 0 { + t.Fatalf("Fallback TTFT should be recorded and > 0, got %v", reporter.ttftDuration()) + } + + // 3. Observing substantive token event sets effective TTFT + ObserveResponsesTokenEvent(reporter, []byte(`{"type":"response.output_text.delta","delta":"hello"}`)) + if !reporter.IsTTFTSet() { + t.Fatalf("Observing token event must set effective TTFT") + } +} + +func TestUsageReporterTrackHTTPClientRoundTripOnly_ErrorResponseRecordsFirstPacketFallback(t *testing.T) { + reporter := NewUsageReporter(context.Background(), "codex", "gpt-5.6-luna", nil) + client := reporter.TrackHTTPClientRoundTripOnly(&http.Client{ + Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Status: "429 Too Many Requests", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"error":{"message":"rate limit"}}`)), + Request: req, + }, nil + }), + }) + + req, errNewRequest := http.NewRequestWithContext(context.Background(), http.MethodPost, "https://example.invalid/v1/responses", strings.NewReader("{}")) + if errNewRequest != nil { + t.Fatalf("NewRequestWithContext() error = %v", errNewRequest) + } + resp, errDo := client.Do(req) + if errDo != nil { + t.Fatalf("Do() error = %v", errDo) + } + _, errRead := io.ReadAll(resp.Body) + if errRead != nil { + t.Fatalf("ReadAll() error = %v", errRead) + } + _ = resp.Body.Close() + + if reporter.IsTTFTSet() { + t.Fatalf("error response read must not set substantive token TTFT") + } + if !reporter.IsFirstPacketSet() { + t.Fatalf("error response read must record first packet set fallback") + } +} + +func TestUsageReporterObserveTokenEvent_FastPathNonTokenAndToken(t *testing.T) { + reporter := NewUsageReporter(context.Background(), "codex", "gpt-5.6-luna", nil) + reporter.StartResponseTTFT() + + // 1. Initial state + if reporter.IsTTFTSet() { + t.Fatalf("expected IsTTFTSet() == false initially") + } + + // 2. First non-token event records firstPacketDuration, but does not mark TTFT set + reporter.ObserveTokenEvent(false) + if reporter.IsTTFTSet() { + t.Fatalf("ObserveTokenEvent(false) must not set TTFT") + } + if !reporter.IsFirstPacketSet() { + t.Fatalf("expected IsFirstPacketSet() == true") + } + firstPacketDuration := reporter.firstPacketDuration + + // 3. Subsequent non-token event is a fast-path return and does not alter firstPacketDuration + reporter.ObserveTokenEvent(false) + if reporter.firstPacketDuration != firstPacketDuration { + t.Fatalf("subsequent ObserveTokenEvent(false) must preserve original firstPacketDuration") + } + + // 4. Substantive token event sets effective TTFT + reporter.ObserveTokenEvent(true) + if !reporter.IsTTFTSet() { + t.Fatalf("ObserveTokenEvent(true) must set IsTTFTSet() == true") + } + tokenTTFT := reporter.ttft + + // 5. Subsequent token event is a fast-path return and does not alter TTFT + reporter.ObserveTokenEvent(true) + if reporter.ttft != tokenTTFT { + t.Fatalf("subsequent ObserveTokenEvent(true) must not alter already recorded TTFT") + } +} + func TestUsageReporterBuildRecordIncludesRequestedModelAlias(t *testing.T) { ctx := usage.WithRequestedModelAlias(context.Background(), "client-gpt") reporter := NewUsageReporter(ctx, "openai", "gpt-5.4", nil)