responses 直通超时可配,流式去总超时,代理补阶段超时

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(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error)
-	// GenAiCompatResponsesStream 流式直通 /actions/v1/responses,建立成功返回 SSE body。
-	GenAiCompatResponsesStream(ctx context.Context, cred Credentials, region string, body []byte) (io.ReadCloser, error)
+	// GenAiCompatResponses 直通 OpenAI Responses 请求体到 /actions/v1/responses(xAI 服务端工具通路);
+	// wait 是上游无响应预算(整请求总超时),multi-agent/搜索类模型需远超 SDK 默认 60s。
+	GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte, wait time.Duration) ([]byte, 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
-// `/20231130/actions/v1/responses`(IAM 签名)。xAI 服务端工具(web_search /
-// x_search / code_interpreter)与 mcp 已被 Oracle 文档正式支持,工具参数与限制
-// 遵循 xAI 规格;调用方须自行校验并改写请求体(store/stream)。
-func (c *RealClient) GenAiCompatResponses(ctx context.Context, cred Credentials, region string, body []byte) ([]byte, error) {
-	ic, err := c.genAiInferenceClient(cred, region)
-	if err != nil {
-		return nil, err
+// dispatcherWithTimeout 把 dispatcher 换成指定总超时的拷贝(保留 Transport,
+// 代理链路不受影响);timeout=0 表示无总超时(流式读 body 不能有总时限)。
+// 非 *http.Client 的自定义 dispatcher 保持原样,维持既有超时行为。
+func dispatcherWithTimeout(d common.HTTPRequestDispatcher, timeout time.Duration) common.HTTPRequestDispatcher {
+	hc, ok := d.(*http.Client)
+	if !ok {
+		return d
 	}
-	client := ic.BaseClient
-	common.UpdateEndpointTemplateForOptions(&client)
-	common.SetMissingTemplateParams(&client)
+	return &http.Client{Transport: hc.Transport, Timeout: timeout}
+}
+
+// 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)
+			}
+		})
+	}
+}