补提 DASH_VERSION 至 v0.7.2

CHANGELOG.md

@@ -2,6 +2,20 @@
 
 格式参考 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.1.0/)（版本段不记日期），版本号遵循语义化版本。
 
+## [0.7.2]
+
+### Added
+
+- AI 网关设置新增「上游无响应预算」（`upstreamWaitSeconds`，30..900 秒，缺省 300，持久化、即时生效）：非流式为单次尝试总超时，流式为等待响应头上限，供 multi-agent / 搜索类慢模型调宽
+
+### Changed
+
+- 出站代理 Transport 补齐阶段超时（连接 30 秒 / TLS 握手 10 秒，对齐 SDK 直连模板）：连不上的代理快速失败，不再拖满总超时
+
+### Fixed
+
+- 修复 Responses 直通调用 multi-agent / 搜索类模型必然超时：SDK 默认 `http.Client` 60 秒总超时覆盖到 body 读完，流式恰好 60 秒断流、非流式（响应头 >60 秒才返回）重试耗尽后约 121 秒报错。非流式改用预算总超时；流式去掉总超时，以定时取消模拟等待响应头预算，响应头到达后流时长不限、生命周期由客户端连接决定。附带消除此类超时对渠道熔断计数的误伤
+
 ## [0.7.1]
 
 ### Added

DASH_VERSION

@@ -1 +1 @@
-v0.7.1
+v0.7.2

docs/docs.go

@@ -860,7 +860,7 @@ const docTemplate = `{
                 "summary": "更新 AI 网关全局设置",
                 "parameters": [
                     {
-                        "description": "全量提交;保险丝阈值限 1..1024 KB",
+                        "description": "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒",
                         "name": "body",
                         "in": "body",
                         "required": true,
@@ -6102,6 +6102,10 @@ const docTemplate = `{
                 },
                 "streamGuardKB": {
                     "type": "integer"
+                },
+                "upstreamWaitSeconds": {
+                    "description": "UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次\n尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s",
+                    "type": "integer"
                 }
             }
         },

docs/swagger.json

@@ -853,7 +853,7 @@
                 "summary": "更新 AI 网关全局设置",
                 "parameters": [
                     {
-                        "description": "全量提交;保险丝阈值限 1..1024 KB",
+                        "description": "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒",
                         "name": "body",
                         "in": "body",
                         "required": true,
@@ -6095,6 +6095,10 @@
                 },
                 "streamGuardKB": {
                     "type": "integer"
+                },
+                "upstreamWaitSeconds": {
+                    "description": "UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次\n尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s",
+                    "type": "integer"
                 }
             }
         },

docs/swagger.yaml

@@ -65,6 +65,11 @@ definitions:
         type: boolean
       streamGuardKB:
         type: integer
+      upstreamWaitSeconds:
+        description: |-
+          UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次
+          尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s
+        type: integer
     type: object
   internal_api.attachBootVolumeRequest:
     properties:
@@ -3204,7 +3209,7 @@ paths:
       - AI 管理
     put:
       parameters:
-      - description: 全量提交;保险丝阈值限 1..1024 KB
+      - description: 全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒
         in: body
         name: body
         required: true

internal/api/aiadmin.go

@@ -3,6 +3,7 @@ package api
 import (
 	"net/http"
 	"strconv"
+	"time"
 
 	"github.com/gin-gonic/gin"
 
@@ -450,6 +451,9 @@ type aiSettingsResponse struct {
 	// 请求 tools 已包含同名工具时不覆盖
 	GrokWebSearch bool `json:"grokWebSearch"`
 	GrokXSearch   bool `json:"grokXSearch"`
+	// UpstreamWaitSeconds 是 responses 直通的上游无响应预算(秒):非流式为单次
+	// 尝试总超时,流式为等待响应头预算;multi-agent/搜索类模型需远超 60s
+	UpstreamWaitSeconds int `json:"upstreamWaitSeconds"`
 }
 
 // currentAiSettings 汇总网关运行时设置为响应体。
@@ -457,11 +461,12 @@ func (h *aiAdminHandler) currentAiSettings() aiSettingsResponse {
 	guardOn, guardKB := h.gw.StreamGuard()
 	web, x := h.gw.GrokSearch()
 	return aiSettingsResponse{
-		FilterDeprecated:   h.gw.FilterDeprecated(),
+		FilterDeprecated:    h.gw.FilterDeprecated(),
-		StreamGuardEnabled: guardOn,
+		StreamGuardEnabled:  guardOn,
-		StreamGuardKB:      guardKB,
+		StreamGuardKB:       guardKB,
-		GrokWebSearch:      web,
+		GrokWebSearch:       web,
-		GrokXSearch:        x,
+		GrokXSearch:         x,
+		UpstreamWaitSeconds: int(h.gw.UpstreamWait() / time.Second),
 	}
 }
 
@@ -480,7 +485,7 @@ func (h *aiAdminHandler) aiSettings(c *gin.Context) {
 //
 // @Summary  更新 AI 网关全局设置
 // @Tags     AI 管理
-// @Param    body body aiSettingsResponse true "全量提交;保险丝阈值限 1..1024 KB"
+// @Param    body body aiSettingsResponse true "全量提交;保险丝阈值限 1..1024 KB,上游无响应预算限 30..900 秒"
 // @Success  200 {object} aiSettingsResponse
 // @Failure  400 {object} map[string]string
 // @Security BearerAuth
@@ -495,6 +500,10 @@ func (h *aiAdminHandler) updateAiSettings(c *gin.Context) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "streamGuardKB 须在 1..1024"})
 		return
 	}
+	if req.UpstreamWaitSeconds < 30 || req.UpstreamWaitSeconds > 900 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "upstreamWaitSeconds 须在 30..900"})
+		return
+	}
 	ctx := c.Request.Context()
 	if err := h.gw.SetFilterDeprecated(ctx, req.FilterDeprecated); err != nil {
 		respondError(c, err)
@@ -508,5 +517,9 @@ func (h *aiAdminHandler) updateAiSettings(c *gin.Context) {
 		respondError(c, err)
 		return
 	}
+	if err := h.gw.SetUpstreamWait(ctx, req.UpstreamWaitSeconds); err != nil {
+		respondError(c, err)
+		return
+	}
 	c.JSON(http.StatusOK, h.currentAiSettings())
 }

internal/oci/client.go

@@ -102,10 +102,12 @@ type Client interface {
 	ListGenAiModels(ctx context.Context, cred Credentials, region string) ([]GenAiModel, error)
 	GenAiProbeChat(ctx context.Context, cred Credentials, region, modelOcid, modelName string) (int, error)
 	GenAiEmbed(ctx context.Context, cred Credentials, region, modelOcid string, inputs []string, dimensions *int) ([][]float32, *aiwire.Usage, error)
-	// GenAiCompatResponses 直通 OpenAI Responses 请求体到 /actions/v1/responses(xAI 服务端工具通路)。
+	// GenAiCompatResponses 直通 OpenAI Responses 请求体到 /actions/v1/responses(xAI 服务端工具通路);
-	GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error)
+	// wait 是上游无响应预算(整请求总超时),multi-agent/搜索类模型需远超 SDK 默认 60s。
-	// GenAiCompatResponsesStream 流式直通 /actions/v1/responses,建立成功返回 SSE body。
+	GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) ([]byte, error)
-	GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte) (io.ReadCloser, error)
+	// GenAiCompatResponsesStream 流式直通 /actions/v1/responses,建立成功返回 SSE body;
+	// wait 仅约束等待响应头阶段,建立后的流生命周期由 ctx 决定。
+	GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error)
 	// GenAiCompatSpeech 直通 OpenAI Audio Speech 请求体到 /openai/v1/audio/speech,返回音频与 Content-Type。
 	GenAiCompatSpeech(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, string, error)
 	// GenAiRerank 文档重排,返回按相关度排序的下标与得分。

internal/oci/genai.go

@@ -158,7 +158,8 @@ func (c *RealClient) GenAiProbeChat(ctx context.Context, cred Credentials, regio
 	if err != nil {
 		return 0, err
 	}
-	if _, err = c.GenAiCompatResponses(ctx, cred, region, body); err == nil {
+	// 探测追求快速失败,沿用 SDK 默认量级的 60s 预算即可
+	if _, err = c.GenAiCompatResponses(ctx, cred, region, body, 60*time.Second); err == nil {
 		return http.StatusOK, nil
 	}
 	if status, ok := ServiceStatus(err); ok {

internal/oci/genai_responses.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"time"
 
 	"github.com/oracle/oci-go-sdk/v65/common"
 )
@@ -13,18 +14,19 @@ import (
 // compatResponsesLimit 限制直通响应体大小;web_search 输出含多段引用,给足余量。
 const compatResponsesLimit = int64(8 << 20)
 
-// GenAiCompatResponses 实现 Client:把 OpenAI Responses 请求体直通到 OCI
+// dispatcherWithTimeout 把 dispatcher 换成指定总超时的拷贝(保留 Transport,
-// `/20231130/actions/v1/responses`(IAM 签名)。xAI 服务端工具(web_search /
+// 代理链路不受影响);timeout=0 表示无总超时(流式读 body 不能有总时限)。
-// x_search / code_interpreter)与 mcp 已被 Oracle 文档正式支持,工具参数与限制
+// 非 *http.Client 的自定义 dispatcher 保持原样,维持既有超时行为。
-// 遵循 xAI 规格;调用方须自行校验并改写请求体(store/stream)。
+func dispatcherWithTimeout(d common.HTTPRequestDispatcher, timeout time.Duration) common.HTTPRequestDispatcher {
-func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error) {
+	hc, ok := d.(*http.Client)
-	ic, err := c.genAiInferenceClient(cred, region)
+	if !ok {
-	if err != nil {
+		return d
-		return nil, err
 	}
-	client := ic.BaseClient
+	return &http.Client{Transport: hc.Transport, Timeout: timeout}
-	common.UpdateEndpointTemplateForOptions(&client)
+}
-	common.SetMissingTemplateParams(&client)
+
+// newCompatResponsesRequest 构造 /actions/v1/responses 直通请求。
+func newCompatResponsesRequest(ctx context.Context, cred Credentials, body []byte) (*http.Request, error) {
 	request, err := http.NewRequestWithContext(ctx, http.MethodPost, "/actions/v1/responses", bytes.NewReader(body))
 	if err != nil {
 		return nil, fmt.Errorf("build compat responses request: %w", err)
@@ -32,6 +34,27 @@ func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials,
 	request.Header.Set("Content-Type", "application/json")
 	request.Header.Set("CompartmentId", cred.TenancyOCID)
 	request.Header.Set("opc-compartment-id", cred.TenancyOCID)
+	return request, nil
+}
+
+// GenAiCompatResponses 实现 Client:把 OpenAI Responses 请求体直通到 OCI
+// `/20231130/actions/v1/responses`(IAM 签名)。xAI 服务端工具(web_search /
+// x_search / code_interpreter)与 mcp 已被 Oracle 文档正式支持,工具参数与限制
+// 遵循 xAI 规格;调用方须自行校验并改写请求体(store/stream)。
+// wait 为整请求总超时:非流式上游要等全部生成完才回响应头,SDK 默认 60s 会掐断慢模型。
+func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) ([]byte, error) {
+	ic, err := c.genAiInferenceClient(cred, region)
+	if err != nil {
+		return nil, err
+	}
+	client := ic.BaseClient
+	common.UpdateEndpointTemplateForOptions(&client)
+	common.SetMissingTemplateParams(&client)
+	client.HTTPClient = dispatcherWithTimeout(client.HTTPClient, wait)
+	request, err := newCompatResponsesRequest(ctx, cred, body)
+	if err != nil {
+		return nil, err
+	}
 	response, err := client.Call(ctx, request)
 	if err != nil {
 		return nil, err
@@ -44,10 +67,46 @@ func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials,
 	return payload, nil
 }
 
+// cancelReadCloser 在流关闭时同步取消建立阶段派生的 ctx,避免其随流生命周期泄漏。
+type cancelReadCloser struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
+
+func (c *cancelReadCloser) Close() error {
+	c.cancel()
+	return c.ReadCloser.Close()
+}
+
+// httpCaller 抽象 BaseClient.Call,便于对预算逻辑做无签名单测。
+type httpCaller interface {
+	Call(ctx context.Context, request *http.Request) (*http.Response, error)
+}
+
+// callWithHeaderBudget 以 wait 为等待响应头预算执行调用:预算内未返回则取消
+// 请求(SDK Call 会把 ctx 重绑到请求);响应头到达即解除预算,之后流的生命
+// 周期由 ctx 决定,返回的流 Close 时同步取消派生 ctx。
+func callWithHeaderBudget(ctx context.Context, c httpCaller, req *http.Request, wait time.Duration) (io.ReadCloser, error) {
+	callCtx, cancel := context.WithCancel(ctx)
+	timer := time.AfterFunc(wait, cancel)
+	response, err := c.Call(callCtx, req)
+	timer.Stop()
+	if err != nil {
+		if response != nil && response.Body != nil {
+			response.Body.Close()
+		}
+		cancel()
+		return nil, err
+	}
+	return &cancelReadCloser{ReadCloser: response.Body, cancel: cancel}, nil
+}
+
 // GenAiCompatResponsesStream 实现 Client:以流式直通 OCI `/actions/v1/responses`,
 // 建立成功(2xx)返回 SSE body(调用方负责 Close);建立失败返回 SDK ServiceError,
 // 与既有渠道切换/熔断错误分类兼容。请求体须由调用方置 stream:true。
-func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte) (io.ReadCloser, error) {
+// 总超时置 0(SSE 读 body 不能有总时限);wait 以定时取消模拟等待响应头预算,
+// 响应头到达即解除,此后流的生命周期完全由 ctx(下游客户端断开)决定。
+func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error) {
 	ic, err := c.genAiInferenceClient(cred, region)
 	if err != nil {
 		return nil, err
@@ -55,19 +114,10 @@ func (c *RealClient) GenAiCompatResponsesStream(ctx context.Context, cred Creden
 	client := ic.BaseClient
 	common.UpdateEndpointTemplateForOptions(&client)
 	common.SetMissingTemplateParams(&client)
-	request, err := http.NewRequestWithContext(ctx, http.MethodPost, "/actions/v1/responses", bytes.NewReader(body))
+	client.HTTPClient = dispatcherWithTimeout(client.HTTPClient, 0)
+	request, err := newCompatResponsesRequest(ctx, cred, body)
 	if err != nil {
-		return nil, fmt.Errorf("build compat responses stream request: %w", err)
-	}
-	request.Header.Set("Content-Type", "application/json")
-	request.Header.Set("CompartmentId", cred.TenancyOCID)
-	request.Header.Set("opc-compartment-id", cred.TenancyOCID)
-	response, err := client.Call(ctx, request)
-	if err != nil {
-		if response != nil && response.Body != nil {
-			response.Body.Close()
-		}
 		return nil, err
 	}
-	return response.Body, nil
+	return callWithHeaderBudget(ctx, client, request, wait)
 }

internal/oci/genai_responses_test.go

@@ -0,0 +1,141 @@
+package oci
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/oracle/oci-go-sdk/v65/common"
+)
+
+// staticDispatcher 是非 *http.Client 的自定义 dispatcher,用于降级分支。
+type staticDispatcher struct{}
+
+func (staticDispatcher) Do(*http.Request) (*http.Response, error) { return nil, nil }
+
+func TestDispatcherWithTimeout(t *testing.T) {
+	tr := &http.Transport{}
+	tests := []struct {
+		name    string
+		in      common.HTTPRequestDispatcher
+		timeout time.Duration
+		check   func(t *testing.T, out common.HTTPRequestDispatcher)
+	}{
+		{
+			name: "http.Client 换总超时并保留 Transport", in: &http.Client{Transport: tr, Timeout: 60 * time.Second},
+			timeout: 300 * time.Second,
+			check: func(t *testing.T, out common.HTTPRequestDispatcher) {
+				hc, ok := out.(*http.Client)
+				if !ok || hc.Timeout != 300*time.Second || hc.Transport != tr {
+					t.Fatalf("期望拷贝 client 且 Timeout=300s、Transport 保留, 得到 %#v", out)
+				}
+			},
+		},
+		{
+			name: "timeout=0 表示无总超时", in: &http.Client{Timeout: 60 * time.Second}, timeout: 0,
+			check: func(t *testing.T, out common.HTTPRequestDispatcher) {
+				if hc := out.(*http.Client); hc.Timeout != 0 {
+					t.Fatalf("期望 Timeout=0, 得到 %v", hc.Timeout)
+				}
+			},
+		},
+		{
+			name: "非 http.Client 原样返回", in: staticDispatcher{}, timeout: 300 * time.Second,
+			check: func(t *testing.T, out common.HTTPRequestDispatcher) {
+				if _, ok := out.(staticDispatcher); !ok {
+					t.Fatalf("期望原样返回自定义 dispatcher, 得到 %#v", out)
+				}
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) { tt.check(t, dispatcherWithTimeout(tt.in, tt.timeout)) })
+	}
+}
+
+// ctxReader 模拟真实 HTTP body:请求 ctx 取消后读即失败。
+type ctxReader struct {
+	ctx context.Context
+	r   io.Reader
+}
+
+func (c *ctxReader) Read(p []byte) (int, error) {
+	if err := c.ctx.Err(); err != nil {
+		return 0, err
+	}
+	return c.r.Read(p)
+}
+
+// fakeCaller 在 delay 后返回响应;期间 ctx 取消则按真实行为返回 ctx 错误。
+type fakeCaller struct {
+	delay   time.Duration
+	body    string
+	gotCtx  context.Context
+	failErr error
+}
+
+func (f *fakeCaller) Call(ctx context.Context, _ *http.Request) (*http.Response, error) {
+	f.gotCtx = ctx
+	if f.failErr != nil {
+		return nil, f.failErr
+	}
+	select {
+	case <-time.After(f.delay):
+		body := &ctxReader{ctx: ctx, r: strings.NewReader(f.body)}
+		return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(body)}, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func TestCallWithHeaderBudget(t *testing.T) {
+	req, _ := http.NewRequest(http.MethodPost, "/actions/v1/responses", nil)
+	t.Run("预算内返回响应头后长读不受预算影响", func(t *testing.T) {
+		fc := &fakeCaller{delay: 0, body: "data: hello"}
+		stream, err := callWithHeaderBudget(context.Background(), fc, req, 30*time.Millisecond)
+		if err != nil {
+			t.Fatalf("callWithHeaderBudget = %v", err)
+		}
+		defer stream.Close()
+		time.Sleep(90 * time.Millisecond) // 远超预算,验证响应头到达后预算已解除
+		payload, err := io.ReadAll(stream)
+		if err != nil || string(payload) != "data: hello" {
+			t.Fatalf("预算解除后读流 = %q, %v; 期望完整 body", payload, err)
+		}
+	})
+	t.Run("超预算未返回响应头即取消", func(t *testing.T) {
+		fc := &fakeCaller{delay: time.Minute}
+		start := time.Now()
+		_, err := callWithHeaderBudget(context.Background(), fc, req, 30*time.Millisecond)
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("期望 context.Canceled, 得到 %v", err)
+		}
+		if elapsed := time.Since(start); elapsed > 5*time.Second {
+			t.Fatalf("取消耗时 %v, 未受预算约束", elapsed)
+		}
+	})
+	t.Run("关闭流时取消派生 ctx", func(t *testing.T) {
+		fc := &fakeCaller{delay: 0, body: "x"}
+		stream, err := callWithHeaderBudget(context.Background(), fc, req, time.Minute)
+		if err != nil {
+			t.Fatalf("callWithHeaderBudget = %v", err)
+		}
+		stream.Close()
+		if fc.gotCtx.Err() == nil {
+			t.Fatal("Close 后派生 ctx 应已取消")
+		}
+	})
+	t.Run("建立失败时同样取消派生 ctx", func(t *testing.T) {
+		fc := &fakeCaller{failErr: errors.New("boom")}
+		if _, err := callWithHeaderBudget(context.Background(), fc, req, time.Minute); err == nil {
+			t.Fatal("期望建立失败")
+		}
+		if fc.gotCtx.Err() == nil {
+			t.Fatal("失败路径派生 ctx 应已取消")
+		}
+	})
+}

internal/oci/proxyhttp.go

@@ -2,6 +2,7 @@ package oci
 
 import (
 	"fmt"
+	"net"
 	"net/http"
 	"net/url"
 	"time"
@@ -23,6 +24,13 @@ type ProxySpec struct {
 // proxyClientTimeout 与 SDK 默认 HTTPClient 超时保持一致。
 const proxyClientTimeout = 60 * time.Second
 
+// 阶段超时对齐 SDK 直连 Transport 模板(transport_template_provider):连不上的
+// 代理快速失败,而不是拖满总超时;responses 直通去掉总超时后这是建立阶段的兜底之一。
+const (
+	proxyDialTimeout         = 30 * time.Second
+	proxyTLSHandshakeTimeout = 10 * time.Second
+)
+
 // applyProxy 在 SDK client 构造后统一挂出站代理;未关联代理时不动默认配置。
 // 所有 New*ClientWithConfigurationProvider 调用点构造成功后都必须经过这里。
 func applyProxy(base *common.BaseClient, cred Credentials) {
@@ -52,18 +60,23 @@ func HTTPClientFor(p *ProxySpec) *http.Client {
 // transportFor 构造代理 Transport:http / https 走 CONNECT,socks5 走拨号器。
 func transportFor(p *ProxySpec) *http.Transport {
 	addr := fmt.Sprintf("%s:%d", p.Host, p.Port)
+	dialer := &net.Dialer{Timeout: proxyDialTimeout}
 	if p.Type == "http" || p.Type == "https" {
 		u := &url.URL{Scheme: p.Type, Host: addr}
 		if p.Username != "" {
 			u.User = url.UserPassword(p.Username, p.Password)
 		}
-		return &http.Transport{Proxy: http.ProxyURL(u)}
+		return &http.Transport{
+			Proxy:               http.ProxyURL(u),
+			DialContext:         dialer.DialContext,
+			TLSHandshakeTimeout: proxyTLSHandshakeTimeout,
+		}
 	}
 	var auth *proxy.Auth
 	if p.Username != "" {
 		auth = &proxy.Auth{User: p.Username, Password: p.Password}
 	}
-	d, err := proxy.SOCKS5("tcp", addr, auth, proxy.Direct)
+	d, err := proxy.SOCKS5("tcp", addr, auth, dialer)
 	if err != nil {
 		return nil
 	}
@@ -71,5 +84,8 @@ func transportFor(p *ProxySpec) *http.Transport {
 	if !ok {
 		return nil
 	}
-	return &http.Transport{DialContext: cd.DialContext}
+	return &http.Transport{
+		DialContext:         cd.DialContext,
+		TLSHandshakeTimeout: proxyTLSHandshakeTimeout,
+	}
 }

internal/oci/proxyhttp_test.go

@@ -0,0 +1,34 @@
+package oci
+
+import (
+	"testing"
+)
+
+func TestTransportForStageTimeouts(t *testing.T) {
+	tests := []struct {
+		name      string
+		spec      *ProxySpec
+		wantProxy bool // CONNECT 分支应设 Proxy 函数
+	}{
+		{name: "http CONNECT 代理", spec: &ProxySpec{Type: "http", Host: "127.0.0.1", Port: 8080}, wantProxy: true},
+		{name: "https CONNECT 代理", spec: &ProxySpec{Type: "https", Host: "127.0.0.1", Port: 8443}, wantProxy: true},
+		{name: "socks5 代理", spec: &ProxySpec{Type: "socks5", Host: "127.0.0.1", Port: 1080, Username: "u", Password: "p"}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tr := transportFor(tt.spec)
+			if tr == nil {
+				t.Fatal("transportFor 返回 nil")
+			}
+			if (tr.Proxy != nil) != tt.wantProxy {
+				t.Fatalf("Proxy 函数存在性 = %v, 期望 %v", tr.Proxy != nil, tt.wantProxy)
+			}
+			if tr.DialContext == nil {
+				t.Fatal("应设置带超时的 DialContext")
+			}
+			if tr.TLSHandshakeTimeout != proxyTLSHandshakeTimeout {
+				t.Fatalf("TLSHandshakeTimeout = %v, 期望 %v", tr.TLSHandshakeTimeout, proxyTLSHandshakeTimeout)
+			}
+		})
+	}
+}

internal/service/aigateway.go

@@ -68,6 +68,9 @@ type AiGatewayService struct {
 	// grokWebSearch / grokXSearch 是 xai. 模型服务端搜索工具默认注入开关
 	grokWebSearch atomic.Bool
 	grokXSearch   atomic.Bool
+	// upstreamWaitSec 是 responses 直通的上游无响应预算(秒):非流式为单次尝试
+	// 总超时,流式为等待响应头预算;multi-agent/搜索类模型远超 SDK 默认 60s
+	upstreamWaitSec atomic.Int64
 }
 
 // NewAiGatewayService 组装依赖;调用 StartCleanup 后开始调用日志周期清理。
@@ -78,6 +81,7 @@ func NewAiGatewayService(db *gorm.DB, configs *OciConfigService, client oci.Clie
 	s.streamGuardKB.Store(int64(loadIntSetting(db, settingAiStreamGuardKB, defaultStreamGuardKB)))
 	s.grokWebSearch.Store(loadBoolSetting(db, settingAiGrokWebSearch, true))
 	s.grokXSearch.Store(loadBoolSetting(db, settingAiGrokXSearch, true))
+	s.upstreamWaitSec.Store(int64(loadIntSetting(db, settingAiUpstreamWaitSec, defaultUpstreamWaitSec)))
 	return s
 }
 
@@ -93,11 +97,17 @@ const (
 	// 开关,缺省开。
 	settingAiGrokWebSearch = "ai_grok_web_search"
 	settingAiGrokXSearch   = "ai_grok_x_search"
+	// settingAiUpstreamWaitSec 是 responses 直通的上游无响应预算(秒)。
+	settingAiUpstreamWaitSec = "ai_upstream_wait_seconds"
 )
 
 // defaultStreamGuardKB 是保险丝阈值缺省值,低于实测断流边界留余量。
 const defaultStreamGuardKB = 60
 
+// defaultUpstreamWaitSec 是上游无响应预算缺省值(秒):multi-agent 非流式
+// 实测 100~180s 才回响应头,给足余量;上下限见 SetUpstreamWait。
+const defaultUpstreamWaitSec = 300
+
 // loadBoolSetting 读 settings 表布尔键,无行或值非法时返回缺省。
 func loadBoolSetting(db *gorm.DB, key string, def bool) bool {
 	var row model.Setting
@@ -165,6 +175,25 @@ func (s *AiGatewayService) SetStreamGuard(ctx context.Context, on bool, kb int)
 	return nil
 }
 
+// UpstreamWait 返回 responses 直通的上游无响应预算。
+func (s *AiGatewayService) UpstreamWait() time.Duration {
+	return time.Duration(s.upstreamWaitSec.Load()) * time.Second
+}
+
+// SetUpstreamWait 持久化并即时生效上游无响应预算;sec 限定 30..900。
+func (s *AiGatewayService) SetUpstreamWait(ctx context.Context, sec int) error {
+	if sec < 30 || sec > 900 {
+		return fmt.Errorf("上游无响应预算须在 30..900 秒, 收到 %d", sec)
+	}
+	err := s.db.WithContext(ctx).
+		Save(&model.Setting{Key: settingAiUpstreamWaitSec, Value: strconv.Itoa(sec)}).Error
+	if err != nil {
+		return fmt.Errorf("保存上游无响应预算: %w", err)
+	}
+	s.upstreamWaitSec.Store(int64(sec))
+	return nil
+}
+
 // GrokSearch 返回 grok 服务端搜索工具默认注入开关(web_search, x_search)。
 func (s *AiGatewayService) GrokSearch() (bool, bool) {
 	return s.grokWebSearch.Load(), s.grokXSearch.Load()

internal/service/aigateway_chat.go

@@ -64,7 +64,7 @@ func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidat
 	if err != nil {
 		return nil, err
 	}
-	return s.client.GenAiCompatResponses(ctx, cred, cand.ch.Region, raw)
+	return s.client.GenAiCompatResponses(ctx, cred, cand.ch.Region, raw, s.UpstreamWait())
 }
 
 // RespPassthroughStream 编排流式直通:流建立成功即绑定渠道,建立失败按 switchable
@@ -83,7 +83,7 @@ func (s *AiGatewayService) RespPassthroughStream(ctx context.Context, raw []byte
 		if err != nil {
 			return nil, meta, err
 		}
-		stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw)
+		stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait())
 		if err == nil {
 			s.markSuccess(ctx, cand.ch.ID)
 			return stream, meta, nil

internal/service/aigateway_test.go

@@ -83,7 +83,7 @@ func (f *gatewayStubClient) GenAiApplyGuardrails(ctx context.Context, cred oci.C
 	return f.guardOutcome, f.guardErr
 }
 
-func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.Credentials, region string, body []byte) ([]byte, error) {
+func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.Credentials, region string, body []byte, wait time.Duration) ([]byte, error) {
 	f.passCalls++
 	f.passRegions = append(f.passRegions, region)
 	if len(f.passErrs) > 0 {
@@ -96,7 +96,7 @@ func (f *gatewayStubClient) GenAiCompatResponses(ctx context.Context, cred oci.C
 	return f.passPayload, nil
 }
 
-func (f *gatewayStubClient) GenAiCompatResponsesStream(ctx context.Context, cred oci.Credentials, region string, body []byte) (io.ReadCloser, error) {
+func (f *gatewayStubClient) GenAiCompatResponsesStream(ctx context.Context, cred oci.Credentials, region string, body []byte, wait time.Duration) (io.ReadCloser, error) {
 	f.passCalls++
 	f.passRegions = append(f.passRegions, region)
 	if len(f.passErrs) > 0 {
@@ -1124,3 +1124,38 @@ func TestAggregatedModelsFilterDeprecated(t *testing.T) {
 		t.Fatalf("开关开:弃用模型应被过滤, got %+v", items)
 	}
 }
+
+func TestUpstreamWaitSetting(t *testing.T) {
+	gw, _ := newTestGateway(t, &gatewayStubClient{fakeClient: &fakeClient{}})
+	ctx := context.Background()
+
+	if got := gw.UpstreamWait(); got != 300*time.Second {
+		t.Fatalf("缺省上游无响应预算 = %v, 期望 300s", got)
+	}
+	tests := []struct {
+		name    string
+		sec     int
+		wantErr bool
+	}{
+		{name: "下界 30 有效", sec: 30},
+		{name: "上界 900 有效", sec: 900},
+		{name: "低于下界拒绝", sec: 29, wantErr: true},
+		{name: "高于上界拒绝", sec: 901, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := gw.SetUpstreamWait(ctx, tt.sec)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("SetUpstreamWait(%d) = %v, wantErr=%v", tt.sec, err, tt.wantErr)
+			}
+			if !tt.wantErr && gw.UpstreamWait() != time.Duration(tt.sec)*time.Second {
+				t.Fatalf("UpstreamWait = %v, 期望 %ds", gw.UpstreamWait(), tt.sec)
+			}
+		})
+	}
+	// 持久化后新实例应加载已存值(最后一次成功设置为 900)
+	gw2 := NewAiGatewayService(gw.db, nil, nil)
+	if got := gw2.UpstreamWait(); got != 900*time.Second {
+		t.Fatalf("重建服务加载预算 = %v, 期望 900s", got)
+	}
+}