package openai

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"go.uber.org/zap"
	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
	"net/http"
	"strconv"
	"strings"
)

func (s *Server) handleResponses(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()

	// Bound the body before the first read; no provider or identity work occurs
	// until the canonical request and typed ledger are committed.
	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 preserve Codex/Responses unknown fields (max_output_tokens,
	// tools, store, ...). Strict field validation stays on the normalized path.
	env, err := decodeResponsesEnvelope(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
	}
	ingress, err := buildOpenAIIngressSnapshot(s.maxIngressSnapshotBytes(), rawBody, env)
	if err != nil {
		writeOpenAIIngressError(w, err)
		return
	}
	defer ingress.Close()
	rawBody = nil

	if dispatch.ProviderPool {
		// Provider-pool path: the one-shot SubmitProviderPool selects a candidate
		// and dispatches either tunnel passthrough (OpenAI-compatible) or
		// normalized RunEvent (Ollama/CLI/native) based on the candidate's
		// executionPath. Raw body is preserved for the tunnel branch; the
		// normalized branch performs strict decode + prompt build below.
		requestCtx, ok := s.newResponsesRequestContext(w, r, dispatch, ingress, env)
		if !ok {
			return
		}
		s.handleResponsesProviderPool(w, requestCtx)
		return
	}

	// Non-provider routes: tunnel or normalized. Tunnel routes relay the raw
	// body verbatim; normalized routes require strict decode + prompt build.
	if routeUsesProviderTunnel(dispatch) {
		// Provider routes always relay pure passthrough. Caller metadata is
		// arbitrary context and never selects the route or response shape
		// (SDD S01/S04); there is no caller-facing response mode selector.
		requestCtx, ok := s.newResponsesRequestContext(w, r, dispatch, ingress, env)
		if !ok {
			return
		}
		s.tunnelResponsesPassthrough(w, requestCtx)
		return
	}

	// Normalized RunEvent path: strict decode, stream/background rejection,
	// prompt build, and SubmitRun.
	canonicalBody, err := ingress.canonicalBody()
	if err != nil {
		writeOpenAIIngressError(w, err)
		return
	}
	var req responsesRequest
	if err := decodeResponsesRequest(json.NewDecoder(bytes.NewReader(canonicalBody)), &req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}
	// Keep direct normalized validation ahead of identity resolution, matching
	// the existing strict Responses contract. The dispatch-context constructor
	// repeats these local checks when it is invoked from provider-pool PrepareRun.
	if req.Stream {
		writeError(w, http.StatusBadRequest, "invalid_request_error", "streaming is not supported for /v1/responses")
		return
	}
	if req.Background {
		writeError(w, http.StatusBadRequest, "invalid_request_error", "background is not supported for /v1/responses")
		return
	}
	if _, err := parseResponsesInput(req.Input); err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}
	canonicalBody = nil

	requestCtx, ok := s.newResponsesRequestContext(w, r, dispatch, ingress, env)
	if !ok {
		return
	}

	dc, err := s.newResponsesDispatchContext(requestCtx, req)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}

	s.logger.Info("openai responses input",
		zap.String("model", dc.req.Model),
		zap.String("target", dc.route.Target),
		zap.String("adapter", dc.route.Adapter),
		zap.Bool("strict_output", dc.outputPolicy.Strict),
		zap.String("xml_completion_tool", dc.outputPolicy.XMLCompletionTool),
		zap.Bool("contract_instruction", dc.outputPolicy.ContractInstruction),
		zap.Int("prompt_len", len(dc.prompt)),
	)

	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.logger.Info("openai responses dispatch",
		zap.String("run_id", handle.Dispatch().RunID),
		zap.String("node_id", handle.Dispatch().NodeID),
		zap.String("provider_id", handle.Dispatch().ProviderID),
		zap.String("provider_type", handle.Dispatch().ProviderType),
		zap.String("execution_path", handle.Dispatch().ExecutionPath),
		zap.String("model_group", handle.Dispatch().ModelGroupKey),
		zap.String("adapter", handle.Dispatch().Adapter),
		zap.String("target", handle.Dispatch().Target),
		zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens),
		zap.String("context_class", handle.Dispatch().ContextClass),
		zap.String("queue_reason", handle.Dispatch().QueueReason),
	)

	s.runOpenAIResponsesStreamGate(w, dc, handle)
}

// newResponsesRequestContext resolves the identity, estimate, and long-context
// classification a /v1/responses request carries into either dispatch path. It
// writes the client error and reports false when the caller identity is rejected.
func (s *Server) newResponsesRequestContext(w http.ResponseWriter, r *http.Request, dispatch routeDispatch, ingress *openAIIngressSnapshot, env responsesEnvelope) (*responsesRequestContext, bool) {
	canonicalBody, err := ingress.canonicalBody()
	if err != nil {
		writeOpenAIIngressError(w, err)
		return nil, false
	}
	runMeta, err := resolveCallerIdentity(r, dispatch, env.Metadata)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return nil, false
	}
	// The Responses estimate is taken from the raw body so both the tunnel and
	// the normalized branch carry the same observability and long-context
	// information (SDD S02/S03).
	estimate := estimateInputTokensBytes(canonicalBody, runMeta, nil, nil)
	canonicalBody = nil
	return &responsesRequestContext{
		openAIRequestContext: openAIRequestContext{
			r:              r,
			route:          dispatch,
			ingress:        ingress,
			callerMetadata: runMeta,
			estimate:       estimate,
			contextClass:   classifyContext(estimate, s.longContextThreshold()),
			endpoint:       usageEndpointResponses,
			usage:          s.newOpenAIUsageRecorder(r.Context(), env.Model, usageEndpointResponses),
		},
		envelope: env,
	}, true
}

// newResponsesDispatchContext freezes the strict-decoded request and the
// normalized Run request after ingress routing. Raw tunnel data stays on the
// parent responsesRequestContext, and the copied metadata below is the only
// mutable execution metadata for this normalized branch.
func (s *Server) newResponsesDispatchContext(requestCtx *responsesRequestContext, req responsesRequest) (*responsesDispatchContext, error) {
	if req.Stream {
		return nil, fmt.Errorf("streaming is not supported for /v1/responses")
	}
	if req.Background {
		return nil, fmt.Errorf("background is not supported for /v1/responses")
	}

	inputStr, err := parseResponsesInput(req.Input)
	if err != nil {
		return nil, err
	}
	return s.newResponsesDispatchContextFromInput(requestCtx, req, inputStr, "", "", false)
}

// newResponsesResumeDispatchContext constructs a normalized run from the
// already-validated private continuation shape. Public request parsing cannot
// enter this path, so string-only /v1/responses ingress remains unchanged.
func (s *Server) newResponsesResumeDispatchContext(requestCtx *responsesRequestContext, resume responsesResumeRequest) (*responsesDispatchContext, error) {
	resumeInput := resume.Content
	if resume.Reasoning != "" {
		// Keep both channels byte-for-byte while using only the deterministic
		// separator already defined by buildResponsesPrompt.
		resumeInput = buildResponsesPrompt(resume.Reasoning, resume.Content)
	}
	return s.newResponsesDispatchContextFromInput(requestCtx, resume.Request, resumeInput, resume.Content, resume.Reasoning, true)
}

func (s *Server) newResponsesDispatchContextFromInput(requestCtx *responsesRequestContext, req responsesRequest, inputStr, resumeContent, resumeReasoning string, isResume bool) (*responsesDispatchContext, error) {
	if req.Stream {
		return nil, fmt.Errorf("streaming is not supported for /v1/responses")
	}
	if req.Background {
		return nil, fmt.Errorf("background is not supported for /v1/responses")
	}
	prompt := buildResponsesPrompt(req.Instructions, inputStr)
	outputPolicy := s.resolveOutputPolicy(prompt)
	if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" {
		prompt = instruction + "\n" + prompt
	}

	defaultThinkingTokenBudget := 0
	if catalogEntry := s.findProviderPoolEntry(requestCtx.route.effectiveModelGroupKey(req.Model)); catalogEntry != nil {
		applyModelCatalogGenerationPolicyToResponses(&req, *catalogEntry)
		defaultThinkingTokenBudget = catalogEntry.DefaultThinkingTokenBudget
	}

	runMetadata := cloneMetadata(requestCtx.callerMetadata)
	runMetadata["openai_model"] = req.Model
	runMetadata["openai_stream"] = fmt.Sprintf("%t", req.Stream)
	runMetadata["strict_output"] = fmt.Sprintf("%t", outputPolicy.Strict)
	input := map[string]any{"prompt": prompt}
	if isResume {
		// Preserve recovery channel provenance for the normalized execution
		// path without exposing it in the public request contract.
		input["responses_resume_content"] = resumeContent
		input["responses_resume_reasoning"] = resumeReasoning
	}
	if defaultThinkingTokenBudget > 0 {
		input["think"] = true
		input["thinking_token_budget"] = defaultThinkingTokenBudget
	} else if outputPolicy.Strict {
		input["think"] = false
	}
	if options := req.providerOptions(); len(options) > 0 {
		input["options"] = options
	}

	estimate := estimateInputTokens(prompt, runMetadata, nil, nil)
	contextClass := classifyContext(estimate, s.longContextThreshold())
	runMetadata["estimated_input_tokens"] = strconv.Itoa(estimate)
	runMetadata["context_class"] = contextClass

	dc := &responsesDispatchContext{
		responsesRequestContext: requestCtx,
		req:                     req,
		prompt:                  prompt,
		input:                   input,
		outputPolicy:            outputPolicy,
		runMetadata:             runMetadata,
	}
	dc.submitReq = edgeservice.SubmitRunRequest{
		NodeRef:              dc.route.NodeRef,
		ModelGroupKey:        dc.route.effectiveModelGroupKey(dc.req.Model),
		ProviderID:           dc.route.ProviderID,
		UsageAttribution:     dc.route.UsageAttribution,
		Adapter:              dc.route.Adapter,
		Target:               dc.route.Target,
		SessionID:            dc.route.SessionID,
		Prompt:               dc.prompt,
		Input:                dc.input,
		TimeoutSec:           dc.route.TimeoutSec,
		MaxQueue:             dc.route.MaxQueue,
		QueueTimeoutMS:       dc.route.QueueTimeoutMS,
		Metadata:             dc.runMetadata,
		EstimatedInputTokens: estimate,
		ContextClass:         contextClass,
		ProviderPool:         dc.route.ProviderPool,
	}
	return dc, nil
}

// handleResponsesProviderPool dispatches a provider-pool /v1/responses request
// 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 (OpenAI-compatible providers)
// or the normalized RunEvent path (Ollama/CLI/native). Provider auth is
// validated via providerTunnelAuthHeaders for tunnel-path only (SDD D01).
// Strict-output policy only applies to normalized dispatch, not to raw tunnel
// passthrough (SDD D02).
//
// Selection-first flow: the provider is selected by SubmitProviderPool before
// any path-specific decode, validation, or stream gating. The tunnel branch
// only rewrites the model field and relays the raw body (SDD S04). The
// normalized branch performs strict decode + stream/background rejection +
// prompt build, but only after the candidate is known. This prevents
// tunnel-preferring unknown or streaming requests from bypassing the selected
// provider's executionPath rule (SDD S02/S03).
func (s *Server) handleResponsesProviderPool(w http.ResponseWriter, requestCtx *responsesRequestContext) {
	// Build a populated base Run for the normalized branch and a Tunnel for
	// the tunnel branch. Both are sent to SubmitProviderPool so the one-shot
	// admission selects the provider before any path-specific behavior.
	r := requestCtx.r
	dispatch := requestCtx.route
	rawBody, err := requestCtx.canonicalBody()
	if err != nil {
		writeOpenAIIngressError(w, err)
		return
	}
	estimate := requestCtx.estimate
	contextClass := requestCtx.contextClass
	requirements, err := decodeResponsesProviderRequirements(rawBody)
	if err != nil {
		requestCtx.finishUsageRequest(usageStatusError, responseModePassthrough)
		writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		return
	}

	env := requestCtx.envelope
	bridgeBody := rawBody
	if catalogEntry := s.findProviderPoolEntry(dispatch.effectiveModelGroupKey(env.Model)); catalogEntry != nil {
		var effective responsesRequest
		if err := json.Unmarshal(rawBody, &effective); err == nil {
			applyModelCatalogGenerationPolicyToResponses(&effective, *catalogEntry)
			if effective.MaxOutputTokens != nil {
				maxTokensJSON, marshalErr := json.Marshal(*effective.MaxOutputTokens)
				if marshalErr != nil {
					requestCtx.finishUsageRequest(usageStatusError, responseModePassthrough)
					writeError(w, http.StatusBadRequest, "invalid_request_error", "invalid max_output_tokens")
					return
				}
				patchPlan, patchErr := planTopLevelJSONPatches(rawBody, []topLevelJSONPatch{{name: "max_output_tokens", value: maxTokensJSON}})
				if patchErr != nil {
					requestCtx.finishUsageRequest(usageStatusError, responseModePassthrough)
					writeError(w, http.StatusBadRequest, "invalid_request_error", "invalid max_output_tokens")
					return
				}
				bridgeBody = patchPlan.apply()
			}
		}
	}
	runMeta := cloneMetadata(requestCtx.callerMetadata)
	runMeta["openai_model"] = env.Model
	runMeta["openai_stream"] = strconv.FormatBool(env.Stream)
	runMeta["estimated_input_tokens"] = strconv.Itoa(estimate)
	runMeta["context_class"] = contextClass

	// Base Run carries the model-group key that SubmitProviderPool requires
	// for candidate resolution. Metadata, estimate, and context class are
	// preserved by PrepareRun on the normalized branch; the tunnel branch
	// inherits them from the base so tunnel dispatch keeps observability.
	baseRun := edgeservice.SubmitRunRequest{
		NodeRef:              dispatch.NodeRef,
		ModelGroupKey:        dispatch.effectiveModelGroupKey(env.Model),
		ProviderID:           dispatch.ProviderID,
		UsageAttribution:     dispatch.UsageAttribution,
		SessionID:            dispatch.SessionID,
		Metadata:             runMeta,
		EstimatedInputTokens: estimate,
		ContextClass:         contextClass,
		ProviderPool:         true,
	}

	baseTunnel := edgeservice.SubmitProviderTunnelRequest{
		CredentialBinding:    dispatch.credentialBinding(),
		Metadata:             runMeta,
		EstimatedInputTokens: estimate,
		ContextClass:         contextClass,
		ModelGroupKey:        dispatch.effectiveModelGroupKey(env.Model),
		ProviderID:           dispatch.ProviderID,
		UsageAttribution:     dispatch.UsageAttribution,
		SessionID:            dispatch.SessionID,
		Method:               http.MethodPost,
		Path:                 "/v1/responses",
		Operation:            "",
		Stream:               env.Stream,
		TimeoutSec:           dispatch.TimeoutSec,
		MaxQueue:             dispatch.MaxQueue,
		QueueTimeoutMS:       dispatch.QueueTimeoutMS,
		ProviderPool:         true,
	}

	poolReq := edgeservice.ProviderPoolDispatchRequest{
		Run:    baseRun,
		Tunnel: baseTunnel,
	}
	poolReq.AcceptCandidate = responsesCandidatePredicate(requirements)

	if s.streamGateSemanticEnabled() {
		fctx, err := s.openAIResponsesOutputFilterContext(requestCtx)
		if err != nil {
			requestCtx.finishUsageRequest(usageStatusError, responseModePassthrough)
			writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
			return
		}
		predicate, err := openAIStreamGateCandidatePredicate(s.streamGateConfig(), fctx)
		if err != nil {
			requestCtx.finishUsageRequest(usageStatusError, responseModePassthrough)
			writeError(w, http.StatusInternalServerError, "run_error", "stream gate runtime unavailable")
			return
		}
		poolReq.AcceptCandidate = composeCandidatePredicates(poolReq.AcceptCandidate, predicate)
	}
	if requestCtx.route.Managed {
		poolReq.AcceptCandidate = composeCandidatePredicates(poolReq.AcceptCandidate, requestCtx.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
	}
	poolReq.PrepareProtocolTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest, selected edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
		if selected.ProtocolProfile == nil {
			prepared, err := s.protocolTunnelPreparer(r, config.OperationResponses)(tunnelReq, selected)
			prepared.Operation = string(config.OperationResponses)
			return prepared, err
		}
		plan, err := selectProviderOperation(*selected.ProtocolProfile, config.OperationResponses, requirements)
		if err != nil {
			return tunnelReq, err
		}
		if plan.Operation != config.OperationResponses {
			if err := validateResponsesBridgeControls(rawBody); err != nil {
				return tunnelReq, fmt.Errorf("%w: %v", errProviderRequestValidation, err)
			}
		}
		headers, err := s.providerTunnelAuthHeaders(r)
		if err != nil {
			return tunnelReq, err
		}
		if len(headers) > 0 {
			credential := ""
			for _, value := range headers {
				credential = strings.TrimSpace(value)
				break
			}
			if fields := strings.Fields(credential); len(fields) > 1 {
				credential = strings.Join(fields[1:], " ")
			}
			if scheme := strings.TrimSpace(selected.ProtocolProfile.Auth.Scheme); scheme != "" {
				credential = scheme + " " + credential
			}
			headers = map[string]string{selected.ProtocolProfile.Auth.Header: credential}
		}
		tunnelReq.Headers = headers
		tunnelReq.Operation = string(plan.Operation)
		switch plan.Operation {
		case config.OperationResponses:
			tunnelReq.Path = "/v1/responses"
			tunnelReq.BuildBody = func(target string) ([]byte, error) { return rewriteResponsesProviderControls(rawBody, target, plan) }
		case config.OperationMessages:
			tunnelReq.Path = "/v1/messages"
			if _, err := prepareResponsesMessagesBridge(bridgeBody, "validation-target", *selected.ProtocolProfile, plan); err != nil {
				return tunnelReq, fmt.Errorf("%w: %v", errProviderRequestValidation, err)
			}
			tunnelReq.BuildBody = func(target string) ([]byte, error) {
				return prepareResponsesMessagesBridge(bridgeBody, target, *selected.ProtocolProfile, plan)
			}
		case config.OperationChatCompletions:
			tunnelReq.Path = "/v1/chat/completions"
			tunnelReq.BuildBody = func(target string) ([]byte, error) {
				return prepareResponsesChatBridge(rawBody, target, *selected.ProtocolProfile, plan)
			}
		default:
			return tunnelReq, fmt.Errorf("unsupported Responses provider operation %q", plan.Operation)
		}
		return tunnelReq, nil
	}

	// Tunnel branch rewrites the model and, for concrete profiles, the mapped
	// reasoning effort. All other fields (tools, max_output_tokens, custom
	// fields) are preserved. Stream/background gating is skipped: the provider
	// itself enforces those constraints (SDD S04).
	bodyBuilder := newOpenAIProviderBodyBuilder(func(target string) (*openAIRebuiltLease, error) {
		return rewriteResponsesModelFromIngress(requestCtx.ingress, target)
	})
	poolReq.Tunnel.BuildBody = bodyBuilder.BuildBody

	// The normalized context is created only after candidate selection. That
	// preserves raw passthrough for tunnel candidates while giving the
	// normalized completion one immutable request/policy/metadata snapshot.
	var preparedDispatch *responsesDispatchContext

	// Normalized branch: strict decode, stream/background rejection, prompt
	// build, and generation policy. Only runs AFTER the candidate is selected
	// by SubmitProviderPool (i.e. only when executionPath is normalized).
	// Validation errors are wrapped with errProviderRequestValidation so
	// SubmitProviderPool can relay them as HTTP 400.
	poolReq.PrepareRun = func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
		req := responsesRequest{Model: env.Model}
		if err := decodeResponsesRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
			return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: invalid /v1/responses request format: %s", errProviderRequestValidation, err.Error())
		}
		dc, err := s.newResponsesDispatchContext(requestCtx, req)
		if err != nil {
			return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: %s", errProviderRequestValidation, err.Error())
		}
		// SubmitProviderPool has already filled the common queue-admission
		// fields in runReq. Keep those and overlay the normalized context's
		// body, metadata, and execution values.
		runReq.Prompt = dc.submitReq.Prompt
		runReq.Input = dc.submitReq.Input
		runReq.Metadata = dc.submitReq.Metadata
		runReq.EstimatedInputTokens = dc.submitReq.EstimatedInputTokens
		runReq.ContextClass = dc.submitReq.ContextClass
		runReq.TimeoutSec = dc.submitReq.TimeoutSec
		runReq.MaxQueue = dc.submitReq.MaxQueue
		runReq.QueueTimeoutMS = dc.submitReq.QueueTimeoutMS
		preparedDispatch = dc
		return runReq, nil
	}

	result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
	bodyBuilder.Close()
	if err != nil {
		requestCtx.finishUsageRequest(usageStatusForError(err), responseModePassthrough)
		// Provider auth failure is a client request error (400), not a
		// backend dispatch error. The auth check runs inside
		// SubmitProviderPool (PrepareTunnel) before any tunnel request is sent.
		// Normalized validation failures (e.g. strict decode, stream) are
		// also client errors: map to 400.
		if isProviderCredentialClientError(err) {
			writeError(w, http.StatusBadRequest, "invalid_request_error", providerCredentialClientMessage(err))
			return
		}
		if errors.Is(err, edgeservice.ErrProviderPoolCandidateRejected) {
			var operationErr *edgeservice.ProviderPoolOperationUnsupportedError
			if errors.As(err, &operationErr) {
				writeError(w, http.StatusBadRequest, "invalid_request_error", "no provider profile supports the requested Responses operation")
				return
			}
			writeError(w, http.StatusBadRequest, "invalid_request_error", openAIStreamGateCandidateRejectedMessage)
			return
		}
		if isValidationError(err) {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
			return
		}
		writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
		return
	}

	s.logger.Info("openai responses provider-pool dispatch",
		zap.String("run_id", result.DispatchInfo.RunID),
		zap.String("node_id", result.DispatchInfo.NodeID),
		zap.String("provider_id", result.DispatchInfo.ProviderID),
		zap.String("credential_slot_ref", result.DispatchInfo.CredentialSlotRef),
		zap.Uint64("credential_revision", result.DispatchInfo.CredentialRevision),
		zap.String("provider_type", result.DispatchInfo.ProviderType),
		zap.String("execution_path", result.DispatchInfo.ExecutionPath),
		zap.String("model_group", result.DispatchInfo.ModelGroupKey),
		zap.String("adapter", result.DispatchInfo.Adapter),
		zap.String("target", result.DispatchInfo.Target),
		zap.String("path", string(result.Path)),
		zap.Int("estimated_input_tokens", result.DispatchInfo.EstimatedInputTokens),
		zap.String("context_class", result.DispatchInfo.ContextClass),
		zap.String("queue_reason", result.DispatchInfo.QueueReason),
	)

	switch result.Path {
	case edgeservice.ProviderPoolPathTunnel:
		// Tunnel path: provider auth was already validated and injected via
		// PrepareTunnel before dispatch; on failure SubmitProviderPool returns
		// an error and no tunnel handle exists. Provider bytes are relayed as
		// pure passthrough; caller metadata never selects a sideband surface.
		// The Core request runtime owns response-start staging,
		// and every recovery re-enters SubmitProviderPool through the
		// Responses-specific runtime instead of pinning the initially selected
		// candidate or reusing the caller-derived normalized context.
		s.runOpenAIResponsesPoolStreamGate(w, requestCtx, poolReq, result.Tunnel)

	case edgeservice.ProviderPoolPathNormalized:
		// Normalized path: no auth required, collect from RunEvent stream.
		handle := result.Run
		if handle == nil {
			requestCtx.finishUsageRequest(usageStatusError, responseModeNormalized)
			writeError(w, http.StatusInternalServerError, "run_error", "provider-pool selection returned normalized path but no run result")
			return
		}
		// PrepareRun is only invoked on the normalized path. A normalized
		// result without a successful PrepareRun is an internal mismatch and
		// must surface as run_error (SDD D02).
		if preparedDispatch == nil {
			requestCtx.finishUsageRequest(usageStatusError, responseModeNormalized)
			writeError(w, http.StatusInternalServerError, "run_error", "provider-pool normalized result received without a successful PrepareRun")
			return
		}
		// Relay the prepared normalized context so strict-output XML wrapping
		// and the exact derived metadata survive the provider-pool path.
		s.runOpenAIResponsesStreamGate(w, preparedDispatch.withPoolDispatch(poolReq), handle)
	}
}
