toki
0110ca1af8
OpenCode의 일반 Chat 요청을 GPT provider가 거부한 뒤 재시도 가능한 오류로 왜곡해 벤치가 장시간 정체됐다. 선택된 protocol profile에 맞춰 출력 토큰 필드를 정규화하고 upstream 400을 비재시도 validation 오류로 유지한다.
491 lines
20 KiB
Go
491 lines
20 KiB
Go
package openai
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"go.uber.org/zap"
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
)
|
|
|
|
// chatHotPathDispositionPolicy is the OpenAI Chat projection of the common
|
|
// Hot Path disposition. Commit state is intentionally absent: the Chat codec
|
|
// selects either the normal JSON error/status contract or the committed SSE
|
|
// error + [DONE] sequence without changing these semantics.
|
|
type chatHotPathDispositionPolicy struct {
|
|
status int
|
|
errorType string
|
|
finishReason string
|
|
silent bool
|
|
errorTerminal bool
|
|
}
|
|
|
|
func chatHotPathPolicy(disposition hotPathTerminalDisposition) chatHotPathDispositionPolicy {
|
|
switch disposition.Kind {
|
|
case hotPathDispositionSuccess:
|
|
return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "stop"}
|
|
case hotPathDispositionToolTurn:
|
|
return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "tool_calls"}
|
|
case hotPathDispositionLength:
|
|
return chatHotPathDispositionPolicy{status: http.StatusOK, finishReason: "length"}
|
|
case hotPathDispositionValidationError:
|
|
return chatHotPathDispositionPolicy{
|
|
status: http.StatusBadRequest, errorType: "invalid_request_error", errorTerminal: true,
|
|
}
|
|
case hotPathDispositionProviderError, hotPathDispositionTimeout:
|
|
return chatHotPathDispositionPolicy{
|
|
status: http.StatusBadGateway, errorType: "run_error", errorTerminal: true,
|
|
}
|
|
case hotPathDispositionCallerCancel:
|
|
return chatHotPathDispositionPolicy{silent: true}
|
|
default:
|
|
return chatHotPathDispositionPolicy{
|
|
status: http.StatusBadGateway, errorType: "run_error", errorTerminal: true,
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleChatCompletions(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
|
|
return
|
|
}
|
|
defer r.Body.Close()
|
|
|
|
// Apply the request-start policy before reading. MaxBytesReader performs a
|
|
// limit+1 probe, so no unbounded body exists before the Core snapshot.
|
|
rawBody, err := readOpenAIIngressBody(w, r, s.maxIngressSnapshotBytes())
|
|
if err != nil {
|
|
writeOpenAIIngressError(w, err)
|
|
return
|
|
}
|
|
|
|
// The route decision runs before strict normalization: only the
|
|
// routing-relevant envelope fields are decoded leniently so provider
|
|
// passthrough can forward Codex/provider-specific payloads verbatim.
|
|
// Strict field validation stays on the non-provider path.
|
|
env, err := decodeChatCompletionEnvelope(rawBody)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
dispatch, err := s.resolveRouteDispatchForPrincipal(r.Context(), env.Model)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "model is required")
|
|
return
|
|
}
|
|
|
|
providerNativeThinking := chatRequestHasProviderNativeThinking(rawBody)
|
|
var req chatCompletionRequest
|
|
if dispatch.ProviderPool {
|
|
// Provider-pool path: lenient decode so caller unknown/Codex/
|
|
// provider-specific fields survive into the tunnel body.
|
|
if err := decodeChatCompletionRequestLenient(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
} else {
|
|
// Non-provider path keeps the strict allowlist-based decode.
|
|
if err := decodeChatCompletionRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
}
|
|
providerTunnelRoute := routeUsesProviderTunnel(dispatch)
|
|
basePrompt := promptFromMessages(req.Messages)
|
|
if strings.TrimSpace(basePrompt) == "" {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "messages are required")
|
|
return
|
|
}
|
|
if dispatch.SingleRequest != nil {
|
|
capability, ok := s.service.(singleRequestService)
|
|
if !ok {
|
|
writeError(w, http.StatusServiceUnavailable, "run_error", "single-request execution is unavailable")
|
|
return
|
|
}
|
|
recordSingleRequestIngress()
|
|
if req.Stream {
|
|
s.handleChatSingleRequestStream(w, r, capability, dispatch, rawBody)
|
|
} else {
|
|
s.handleChatSingleRequest(w, r, capability, dispatch, rawBody)
|
|
}
|
|
return
|
|
}
|
|
outputPolicy := s.resolveOutputPolicy(basePrompt)
|
|
// The mutable model-catalog generation policy is applied here, at the single
|
|
// ingress point, before req is frozen into the dispatch context. Every stage
|
|
// downstream reads the post-policy request.
|
|
if catalogEntry := s.findProviderPoolEntry(dispatch.effectiveModelGroupKey(req.Model)); catalogEntry != nil {
|
|
applyModelCatalogGenerationPolicyToChat(&req, *catalogEntry, outputPolicy.Strict, providerNativeThinking)
|
|
}
|
|
|
|
// Freeze canonical bytes plus the effective typed request before identity
|
|
// derivation or any provider admission. A typed-view overflow is therefore
|
|
// a zero-dispatch request-too-large failure.
|
|
ingress, err := buildOpenAIIngressSnapshot(s.maxIngressSnapshotBytes(), rawBody, req)
|
|
if err != nil {
|
|
writeOpenAIIngressError(w, err)
|
|
return
|
|
}
|
|
defer ingress.Close()
|
|
rawBody = nil
|
|
|
|
runMeta, err := resolveCallerIdentity(r, dispatch, req.Metadata)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
if dispatch.IsPreset {
|
|
applyHotPathOutputTokenCap(runMeta, req.MaxTokens, req.MaxCompletionTokens)
|
|
presetChatCodec := newHotPathChatOuterCodec(req.Stream, req.Model, hotPathOutputTokenCap(runMeta))
|
|
r = withHotPathChatOuterCodec(r, presetChatCodec)
|
|
}
|
|
|
|
var presetIngress presetIngressResult
|
|
if dispatch.IsPreset {
|
|
rawBytes, err := ingress.canonicalBody()
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
presetIngress, err = s.joinPresetChatIngress(r, dispatch, rawBytes, runMeta)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
if presetIngress.localStageEligible() {
|
|
_ = s.runHotPathLocalEligible(w, r, dispatch, "openai", req.Stream, runMeta)
|
|
return
|
|
}
|
|
if presetIngress.lightStageContinuation() {
|
|
_ = s.runHotPathLightContinuation(w, r, dispatch, "openai", req.Stream, runMeta)
|
|
return
|
|
}
|
|
if presetIngress.cleanupIssued() {
|
|
_ = s.writeHotPathStageResponse(w, r, dispatch, "openai", req.Stream, presetIngress.Cleanup.RequestID, presetIngress.Cleanup.Output)
|
|
return
|
|
}
|
|
if presetIngress.terminalReady() {
|
|
_ = s.writeHotPathTerminal(w, r, dispatch, "openai", req.Stream, runMeta["iop_logical_request_id"], *presetIngress.Terminal)
|
|
return
|
|
}
|
|
}
|
|
|
|
// The response path is decided by the resolved route, never by caller
|
|
// metadata: provider routes relay pure passthrough over the raw tunnel;
|
|
// every other route uses the normalized RunEvent path. Caller metadata is
|
|
// arbitrary context and does not select route or response shape (SDD S01).
|
|
// Pre-compute estimate/contextClass early so provider-pool path can use them.
|
|
estimate := s.estimateChatInputTokens(basePrompt, runMeta, req.Tools, req.ToolChoice)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
|
|
requestCtx := openAIRequestContext{
|
|
r: r,
|
|
route: dispatch,
|
|
ingress: ingress,
|
|
callerMetadata: runMeta,
|
|
estimate: estimate,
|
|
contextClass: contextClass,
|
|
endpoint: usageEndpointChatCompletions,
|
|
usage: s.newOpenAIUsageRecorder(r.Context(), req.Model, usageEndpointChatCompletions),
|
|
}
|
|
|
|
// Provider-pool: use one-shot SubmitProviderPool which dispatches either
|
|
// tunnel or normalized based on the selected provider's executionPath.
|
|
dc := s.newChatDispatchContext(requestCtx, req, basePrompt, outputPolicy)
|
|
if dispatch.ProviderPool {
|
|
s.handleChatCompletionsProviderPool(w, dc)
|
|
return
|
|
}
|
|
|
|
if providerTunnelRoute {
|
|
s.tunnelChatCompletionPassthrough(w, dc)
|
|
return
|
|
}
|
|
|
|
handle, err := s.service.SubmitRun(r.Context(), dc.submitReq)
|
|
if err != nil {
|
|
dc.finishUsageRequest(usageStatusForError(err), responseModeNormalized)
|
|
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
|
return
|
|
}
|
|
s.logChatDispatch("openai chat completion dispatch", handle.Dispatch())
|
|
|
|
dc = dc.withRetrySubmit(func(ctx context.Context, req edgeservice.SubmitRunRequest) (any, error) {
|
|
return s.service.SubmitRun(ctx, req)
|
|
})
|
|
if req.Stream {
|
|
s.streamChatCompletion(w, dc, handle)
|
|
return
|
|
}
|
|
s.completeChatCompletion(w, dc, handle)
|
|
}
|
|
|
|
// newChatDispatchContext freezes the ingress-resolved values into the immutable
|
|
// dispatch context for the resolved route. The provider-pool route leaves
|
|
// adapter/target unset: the service layer resolves them per candidate at
|
|
// admission and rewrites the target to the winning provider's served model.
|
|
func (s *Server) newChatDispatchContext(requestCtx openAIRequestContext, req chatCompletionRequest, basePrompt string, outputPolicy strictOutputPolicy) *chatDispatchContext {
|
|
dc := &chatDispatchContext{
|
|
openAIRequestContext: requestCtx,
|
|
req: req,
|
|
basePrompt: basePrompt,
|
|
outputPolicy: outputPolicy,
|
|
validation: toolValidationContractFromRequest(req),
|
|
}
|
|
|
|
messages := req.Messages
|
|
if !requestCtx.route.ProviderPool {
|
|
// Strict output is a normalized-dispatch contract only; the pool route
|
|
// decides the path after admission, so the instruction is not baked in
|
|
// here (SDD D02).
|
|
if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" {
|
|
messages = prependSystemMessage(messages, instruction)
|
|
}
|
|
}
|
|
dc.prompt = promptFromMessages(messages)
|
|
dc.input = req.runInput(dc.prompt, messages, outputPolicy.Strict, routeSupportsNativeToolCalls(requestCtx.route))
|
|
|
|
if !requestCtx.route.ProviderPool {
|
|
// Re-estimate against the instruction-augmented prompt for the
|
|
// normalized path.
|
|
dc.estimate = s.estimateChatInputTokens(dc.prompt, requestCtx.callerMetadata, req.Tools, req.ToolChoice)
|
|
dc.contextClass = classifyContext(dc.estimate, s.longContextThreshold())
|
|
}
|
|
|
|
dc.runMetadata = chatRunMetadata(cloneMetadata(requestCtx.callerMetadata), req, outputPolicy)
|
|
dc.runMetadata["estimated_input_tokens"] = strconv.Itoa(dc.estimate)
|
|
dc.runMetadata["context_class"] = dc.contextClass
|
|
|
|
if requestCtx.route.ProviderPool {
|
|
modelGroupKey := requestCtx.route.effectiveModelGroupKey(req.Model)
|
|
if requestCtx.route.IsPreset && strings.TrimSpace(requestCtx.route.Preset.Selector.Model) != "" {
|
|
modelGroupKey = presetSelectorModelGroupKey(requestCtx.route, req.Model)
|
|
}
|
|
dc.submitReq = edgeservice.SubmitRunRequest{
|
|
NodeRef: requestCtx.route.NodeRef,
|
|
ModelGroupKey: modelGroupKey,
|
|
ProviderID: requestCtx.route.ProviderID,
|
|
UsageAttribution: requestCtx.route.UsageAttribution,
|
|
SessionID: requestCtx.route.SessionID,
|
|
Prompt: dc.prompt,
|
|
Input: dc.input,
|
|
TimeoutSec: requestCtx.route.TimeoutSec,
|
|
MaxQueue: requestCtx.route.MaxQueue,
|
|
QueueTimeoutMS: requestCtx.route.QueueTimeoutMS,
|
|
Metadata: dc.runMetadata,
|
|
EstimatedInputTokens: dc.estimate,
|
|
ContextClass: dc.contextClass,
|
|
ProviderPool: true,
|
|
}
|
|
return dc
|
|
}
|
|
|
|
if dc.validation.enabled {
|
|
dc.runMetadata = toolValidationAttemptMetadata(dc.runMetadata, 1, "", "")
|
|
}
|
|
dc.submitReq = chatSubmitRunRequest(requestCtx.route, req, dc.prompt, dc.input, dc.runMetadata)
|
|
dc.submitReq.EstimatedInputTokens = dc.estimate
|
|
dc.submitReq.ContextClass = dc.contextClass
|
|
return dc
|
|
}
|
|
|
|
func (s *Server) logChatDispatch(msg string, disp edgeservice.RunDispatch, extra ...zap.Field) {
|
|
fields := []zap.Field{
|
|
zap.String("run_id", disp.RunID),
|
|
zap.String("node_id", disp.NodeID),
|
|
zap.String("provider_id", disp.ProviderID),
|
|
zap.String("credential_slot_ref", disp.CredentialSlotRef),
|
|
zap.Uint64("credential_revision", disp.CredentialRevision),
|
|
zap.String("provider_type", disp.ProviderType),
|
|
zap.String("execution_path", disp.ExecutionPath),
|
|
zap.String("model_group", disp.ModelGroupKey),
|
|
zap.String("adapter", disp.Adapter),
|
|
zap.String("target", disp.Target),
|
|
zap.Int("estimated_input_tokens", disp.EstimatedInputTokens),
|
|
zap.String("context_class", disp.ContextClass),
|
|
zap.String("queue_reason", disp.QueueReason),
|
|
}
|
|
s.logger.Info(msg, append(fields, extra...)...)
|
|
}
|
|
|
|
// handleChatCompletionsProviderPool dispatches a provider-pool chat completion
|
|
// through the one-shot SubmitProviderPool surface. After a single queue admission,
|
|
// the selected candidate's executionPath determines whether the response is
|
|
// delivered via tunnel passthrough or the normalized RunEvent path.
|
|
//
|
|
// Provider auth is validated via providerTunnelAuthHeaders which is called for
|
|
// tunnel-path only. For normalized paths, the auth check is skipped (SDD D01).
|
|
func (s *Server) handleChatCompletionsProviderPool(w http.ResponseWriter, dc *chatDispatchContext) {
|
|
r := dc.r
|
|
req := dc.req
|
|
modelGroupKey := dc.route.effectiveModelGroupKey(req.Model)
|
|
if dc.route.IsPreset && strings.TrimSpace(dc.route.Preset.Selector.Model) != "" {
|
|
modelGroupKey = presetSelectorModelGroupKey(dc.route, req.Model)
|
|
}
|
|
poolReq := edgeservice.ProviderPoolDispatchRequest{
|
|
Run: dc.submitReq,
|
|
Tunnel: edgeservice.SubmitProviderTunnelRequest{
|
|
CredentialBinding: dc.route.credentialBinding(),
|
|
ModelGroupKey: modelGroupKey,
|
|
ProviderID: dc.route.ProviderID,
|
|
UsageAttribution: dc.route.UsageAttribution,
|
|
SessionID: dc.route.SessionID,
|
|
Method: http.MethodPost,
|
|
Path: "/v1/chat/completions",
|
|
Operation: string(config.OperationChatCompletions),
|
|
Stream: req.Stream,
|
|
TimeoutSec: dc.route.TimeoutSec,
|
|
MaxQueue: dc.route.MaxQueue,
|
|
QueueTimeoutMS: dc.route.QueueTimeoutMS,
|
|
Metadata: dc.runMetadata,
|
|
EstimatedInputTokens: dc.estimate,
|
|
ContextClass: dc.contextClass,
|
|
ProviderPool: true,
|
|
},
|
|
}
|
|
|
|
if s.streamGateSemanticEnabled() {
|
|
fctx, err := s.openAIChatOutputFilterContext(dc)
|
|
if err != nil {
|
|
dc.finishUsageRequest(usageStatusError, responseModePassthrough)
|
|
writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
|
|
return
|
|
}
|
|
predicate, err := openAIStreamGateCandidatePredicate(s.streamGateConfig(), fctx)
|
|
if err != nil {
|
|
dc.finishUsageRequest(usageStatusError, responseModePassthrough)
|
|
writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
|
|
return
|
|
}
|
|
poolReq.AcceptCandidate = predicate
|
|
}
|
|
if dc.route.Managed {
|
|
poolReq.AcceptCandidate = composeCandidatePredicates(poolReq.AcceptCandidate, dc.route.CandidatePredicate())
|
|
}
|
|
|
|
// Pre-dispatch provider auth header injection. Runs inside SubmitProviderPool
|
|
// BEFORE buildProviderTunnelRequest and the Node Send step, so the auth
|
|
// header lands on the actual wire request. On failure the slot is released
|
|
// and no tunnel request is sent (SDD S03).
|
|
poolReq.PrepareTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
headers, err := s.providerTunnelAuthHeaders(r)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
tunnelReq.Headers = headers
|
|
return tunnelReq, nil
|
|
}
|
|
baseProtocolPreparer := s.protocolTunnelPreparer(r, config.OperationChatCompletions)
|
|
selectorInstruction, err := s.hotPathSelectorProviderInstruction(dc.runMetadata)
|
|
if err != nil {
|
|
dc.finishUsageRequest(usageStatusError, responseModePassthrough)
|
|
writeError(w, http.StatusInternalServerError, "run_error", "selector provider instruction is unavailable")
|
|
return
|
|
}
|
|
poolReq.PrepareProtocolTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest, selected edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
prepared, err := baseProtocolPreparer(tunnelReq, selected)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
prepared, err = prepareProviderChatRequestNormalization(prepared, selected)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
prepared, err = prepareProviderChatToolCallNormalization(prepared, selected)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
return prepareHotPathSelectorProviderInstruction(prepared, selectorInstruction)
|
|
}
|
|
|
|
// strict-output output policy only applies to normalized dispatch,
|
|
// not to raw tunnel passthrough (SDD D02). The caller's raw body is the
|
|
// passthrough source; only the model field is rewritten to the served target.
|
|
bodyBuilder := newOpenAIProviderBodyBuilder(func(target string) (*openAIRebuiltLease, error) {
|
|
return rewriteChatCompletionModelFromIngress(dc.ingress, target, req)
|
|
})
|
|
poolReq.Tunnel.BuildBody = bodyBuilder.BuildBody
|
|
|
|
// Normalized branch: chat completions use lenient decode so unknown
|
|
// fields survive; PrepareRun passes through the already-built Run
|
|
// without strict validation (SDD S02/S03). The tunnel branch is
|
|
// unaffected.
|
|
poolReq.PrepareRun = func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
|
|
return runReq, nil
|
|
}
|
|
|
|
result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
|
|
bodyBuilder.Close()
|
|
if err != nil {
|
|
dc.finishUsageRequest(usageStatusForError(err), responseModePassthrough)
|
|
// Provider auth failure is a client request error (400), not a
|
|
// backend dispatch error. The auth check now runs inside
|
|
// SubmitProviderPool (PrepareTunnel) before any tunnel request is sent.
|
|
if isProviderCredentialClientError(err) {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", providerCredentialClientMessage(err))
|
|
return
|
|
}
|
|
if errors.Is(err, edgeservice.ErrProviderPoolCandidateRejected) {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", openAIStreamGateCandidateRejectedMessage)
|
|
return
|
|
}
|
|
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
|
return
|
|
}
|
|
|
|
s.logChatDispatch("openai chat completion provider-pool dispatch", result.DispatchInfo,
|
|
zap.String("path", string(result.Path)),
|
|
)
|
|
if presetHotPathEnabled(dc.route) {
|
|
mode := responseModeNormalized
|
|
if result.Path == edgeservice.ProviderPoolPathTunnel {
|
|
mode = responseModePassthrough
|
|
}
|
|
presetChatCodec := hotPathChatOuterCodecFromRequest(r)
|
|
if presetChatCodec == nil {
|
|
s.terminalPresetRequest(dc.runMetadata["iop_logical_request_id"], s.edgeIDValue())
|
|
dc.finishUsageRequest(usageStatusError, mode)
|
|
writeError(w, http.StatusInternalServerError, "run_error", "Chat outer codec is unavailable")
|
|
return
|
|
}
|
|
stage, collected, turnErr := presetChatCodec.runInitialPresetTurn(s, w, r, dc.route, dc.runMetadata, result)
|
|
if !collected {
|
|
s.terminalPresetRequest(dc.runMetadata["iop_logical_request_id"], s.edgeIDValue())
|
|
dc.finishUsageRequest(usageStatusForError(turnErr), mode)
|
|
disposition, ok := hotPathDispositionFromError(turnErr)
|
|
if !ok {
|
|
disposition = hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionForError(turnErr), Cause: turnErr.Error(), Source: "selector_collection",
|
|
}
|
|
}
|
|
errorType := "run_error"
|
|
if disposition.Kind == hotPathDispositionValidationError {
|
|
errorType = "invalid_request_error"
|
|
}
|
|
_ = presetChatCodec.writeDisposition(
|
|
w, disposition, httpStatusForRunError(turnErr), errorType, turnErr.Error(),
|
|
)
|
|
return
|
|
}
|
|
dc.recordUsageAttempt(result.DispatchInfo, mode, usageObservationFromOpenAIUsage(stage.OpenAIUsage, len(stage.Reasoning)))
|
|
if turnErr != nil {
|
|
dc.finishUsageRequest(usageStatusError, mode)
|
|
return
|
|
}
|
|
dc.finishUsageRequest(usageStatusSuccess, mode)
|
|
return
|
|
}
|
|
|
|
// The Core request runtime owns the whole response for both
|
|
// selected paths. The initial admission result becomes the initial attempt
|
|
// binding, and every recovery re-enters SubmitProviderPool through the same
|
|
// runtime, so the actual provider/model/execution path may still change
|
|
// while the transport is uncommitted.
|
|
s.runOpenAIChatPoolStreamGate(w, dc.withPoolDispatch(poolReq), result)
|
|
}
|