修复跨区间缓存串数据与会话回收泄漏,收敛网关重试

.trellis/spec/backend/database-guidelines.md

@@ -24,7 +24,20 @@ Questions to answer:
 
 <!-- How should queries be written? Batch operations? -->
 
-(To be filled by the team)
+### 租户级数据删除
+
+- 租户主体与本地关联数据必须在同一 GORM transaction 中按“子记录→父记录”删除，每步错误用 `%w` 返回，不得忽略。
+- JSON payload/Setting 键等非外键引用要显式枚举并改写；不能只依赖 `AutoMigrate` 或 ORM association 推断级联范围。
+- 与后台任务、Webhook、解析器等并发写入交叉时，先定义全局一致的行锁顺序，并在写入前重新确认父记录存在。
+- 进程内 cron/缓存只在事务提交后同步；客户端取消不应中断已提交删除的必需运行时对齐。
+- 批量删除/清理遇到**无法解析的 JSON payload** 时记 `log.Printf` 警告并跳过该行(原样保留),不 fail-closed 阻断整个流程——坏数据不应把删除逼到手工修库(tenantdelete.go `logSkippedTask`)。
+
+### 大集合谓词用子查询,禁止展开 IN 列表
+
+- 行数无上界的集合(日志事件、调用日志等)做关联删除/查询时,一律 `WHERE x IN (SELECT …)` 子查询,**不要**先 Pluck ID 再 `IN ?` 展开:绑定变量有硬上限(modernc SQLite 32766,MySQL/PG 65535),数万行即失败且重试无解。
+- GORM 写法:把 `tx.Model(&T{}).Select("id").Where(...)` 作为参数传入 `Where("x IN (?)", sub)`(见 tenantdelete.go `deleteAlertHits` / `alertHitRuleIDs`)。
+- 若原本的 Pluck 兼有 `FOR UPDATE` 锁定语义,保留锁定 SELECT 本身,只是不再把结果拼进后续 SQL(`lockTenantEventRows`)。
+- 行数有小上界的集合(渠道、规则等配置类)可以继续用内存 ID 列表。
 
 ---
 
@@ -48,6 +61,14 @@ Questions to answer:
 
 <!-- Database-related mistakes your team has made -->
 
+### Common Mistake: 用 `Save` 持久化在途任务的陈旧快照
+
+**Symptom**:一条记录已被另一事务删除，在途任务随后执行 `Save` 却把该行重新插入，或覆盖并发更新后的 payload。
+
+**Cause**:GORM `Save` 在 `UPDATE` 零命中时会回退到 `CREATE`/upsert，不适合持久化长时运行开始时读取的快照。
+
+**Fix**:用 `WHERE id = ? AND updated_at = ?` 的条件 `Updates`，并严格要求 `RowsAffected == 1`；零命中表示记录已删除或版本已变，不得补做 `Create`。
+
 ### Common Mistake: gorm 读 NULL 列到已有值的结构体字段不会清零
 
 **Symptom**:UPDATE 把可空列(如 `*time.Time`)写成 NULL 后,用**同一个结构体变量**再次 `First()` 读回,该字段仍是旧值;而新变量读取正常。断言/返回值出现「幽灵旧值」。
@@ -66,3 +87,19 @@ db.Model(&model.AiChannel{}).
     Where("id = ? AND (fail_count > 0 OR disabled_until IS NOT NULL)", id).
     Updates(map[string]any{"fail_count": 0, "disabled_until": gorm.Expr("NULL")})
 ```
+
+### `serializer:json` 字段走 map Updates 时手动 marshal
+
+- 切片/结构体字段用 `gorm:"serializer:json;type:text"` 声明(如 `AiKey.Models []string`),Create/First/struct 路径自动序列化;
+- 但 `Updates(map[string]any{...})` 路径不要依赖 GORM 对 map 值应用 serializer——把值 `json.Marshal` 成 string 放进 map(见 aigateway.go `UpdateKey`),行为版本无关且可测;
+- 存储格式与 serializer 一致(JSON 文本),读回仍走自动反序列化;`nil` 切片 marshal 为 `null`,读回 nil,天然表达「空 = 不限」语义。
+
+### Common Mistake: 进程内缓存键漏掉查询维度
+
+**Symptom**:多区间(compartment)租户在前端切换区间后,实例/卷/VCN 列表短暂显示上一个区间的数据(TTL 窗口内)。
+
+**Cause**:`internal/oci/cached.go` 的 `ckey` 只拼了租户 OCID+资源名+region,而底层查询按 `cred.EffectiveCompartment()` 过滤——影响结果的维度没有全部进键,不同参数命中同一条缓存。
+
+**Fix**:键值加入 `cred.CompartmentID`(空 = 租户根,天然区分)。
+
+**Prevention**:缓存键必须覆盖影响回源结果的**全部**输入维度(租户、区间、区域、过滤参数);给 Credentials/查询结构体新增会改变结果的字段时,同步检查 `ckey` 调用点;隔离行为写进 `cached_test.go` 的 isolation 用例。

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/AI网关.md

@@ -218,6 +218,17 @@ Responses 合成的最小事件序列为：`response.created` →
 
 </details>
 
+### `multi-agent` 加密推理内容流式断流
+
+> [!WARNING]
+> 当 `multi-agent` 请求同时启用 `stream: true` 与
+> `include: ["reasoning.encrypted_content"]` 时，上游可能在序列化大体量
+> `encrypted_content` 事件期间静默断开连接，且不返回 `error` 或终态事件。
+>
+> 已复现场景中的断点位于 `output_index: 8` 附近：第三次搜索结束后，上游准备
+> 发送较大的加密推理块时连接被中止。`multi-agent` 产生的加密推理内容体积较大，
+> 因而更容易暴露该上游流式序列化缺陷。
+
 ### ZDR 与文件输入
 
 Responses 的 `input_file` 内容块（`file_url` / `file_data`）实测会被上游拒绝：

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(),
-		StreamGuardEnabled: guardOn,
-		StreamGuardKB:      guardKB,
-		GrokWebSearch:      web,
-		GrokXSearch:        x,
+		FilterDeprecated:    h.gw.FilterDeprecated(),
+		StreamGuardEnabled:  guardOn,
+		StreamGuardKB:       guardKB,
+		GrokWebSearch:       web,
+		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/cached.go

@@ -32,8 +32,10 @@ func NewCachedClient(inner Client) *CachedClient {
 }
 
 // ckey 组缓存键;租户 OCID 在最前,写失效按前缀一锅端。
+// compartment 必须参与键值:列表查询按 EffectiveCompartment 过滤,
+// 同租户切换区间时若共用键会串到上一个区间的缓存结果。
 func ckey(cred Credentials, parts ...string) string {
-	return cred.TenancyOCID + "|" + strings.Join(parts, "|")
+	return cred.TenancyOCID + "|" + cred.CompartmentID + "|" + strings.Join(parts, "|")
 }
 
 // bust 写操作成功后失效该租户全部读缓存。

internal/oci/cached_test.go

@@ -49,6 +49,13 @@ func TestCachedClientHitAndIsolation(t *testing.T) {
 	if inner.instCalls != 3 {
 		t.Errorf("跨租户/区域回源 %d 次, want 3", inner.instCalls)
 	}
+	// 同租户不同 compartment 各自回源,不得共用缓存
+	inCompartment := testCred("t1")
+	inCompartment.CompartmentID = "ocid1.compartment.a"
+	_, _ = c.ListInstances(ctx, inCompartment, "r1")
+	if inner.instCalls != 4 {
+		t.Errorf("跨 compartment 回源 %d 次, want 4", inner.instCalls)
+	}
 }
 
 func TestCachedClientWriteBusts(t *testing.T) {

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)
+			}
+		})
+	}
+}

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

@@ -27,27 +27,27 @@ type aiCandidate struct {
 	modelOcid string
 }
 
-// RespPassthrough 编排一次非流式直通调用:选渠道(priority→加权随机)→ 调用 →
-// 可重试错误换渠道(整请求上限 3 次)并维护熔断;group 非空时只在同分组渠道内路由。
-// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。
-func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) {
+// routeRetry 统一编排「选渠道(priority→加权随机)→ 调用 → 可重试错误换渠道」,
+// 整请求上限 3 次并维护熔断;group 非空时只在同分组渠道内路由。
+func routeRetry[T any](ctx context.Context, s *AiGatewayService, modelName, group, capability string, once func(*aiCandidate) (T, error)) (T, ChatMeta, error) {
+	var zero T
 	meta := ChatMeta{}
 	excluded := map[uint]bool{}
 	var lastErr error
 	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, modelName, group, "CHAT", excluded)
+		cand, err := s.pick(ctx, modelName, group, capability, excluded)
 		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
+			return zero, meta, firstErr(lastErr, err)
 		}
 		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
-		payload, err := s.passthroughOnce(ctx, cand, raw)
+		out, err := once(cand)
 		if err == nil {
 			s.markSuccess(ctx, cand.ch.ID)
-			return payload, meta, nil
+			return out, meta, nil
 		}
 		retry, penalize := switchable(err)
 		if !retry {
-			return nil, meta, err
+			return zero, meta, err
 		}
 		if penalize {
 			s.markFailure(ctx, cand.ch.ID)
@@ -56,7 +56,15 @@ func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, mode
 		meta.Retries++
 		lastErr = err
 	}
-	return nil, meta, lastErr
+	return zero, meta, lastErr
+}
+
+// RespPassthrough 编排一次非流式直通调用。
+// 上游为 OpenAI-compatible /actions/v1/responses(实测可用,无 Oracle 文档合同)。
+func (s *AiGatewayService) RespPassthrough(ctx context.Context, raw []byte, modelName, group string) ([]byte, ChatMeta, error) {
+	return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) ([]byte, error) {
+		return s.passthroughOnce(ctx, cand, raw)
+	})
 }
 
 func (s *AiGatewayService) passthroughOnce(ctx context.Context, cand *aiCandidate, raw []byte) ([]byte, error) {
@@ -64,42 +72,19 @@ 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
 // 换渠道重试;建立后的中断不重试、不计熔断(与 OpenStream 语义一致)。
 func (s *AiGatewayService) RespPassthroughStream(ctx context.Context, raw []byte, modelName, group string) (io.ReadCloser, ChatMeta, error) {
-	meta := ChatMeta{}
-	excluded := map[uint]bool{}
-	var lastErr error
-	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, modelName, group, "CHAT", excluded)
-		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
-		}
-		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
+	return routeRetry(ctx, s, modelName, group, "CHAT", func(cand *aiCandidate) (io.ReadCloser, error) {
 		cred, err := s.configs.credentialsByID(ctx, cand.ch.OciConfigID)
 		if err != nil {
-			return nil, meta, err
+			return nil, err
 		}
-		stream, err := s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw)
-		if err == nil {
-			s.markSuccess(ctx, cand.ch.ID)
-			return stream, meta, nil
-		}
-		retry, penalize := switchable(err)
-		if !retry {
-			return nil, meta, err
-		}
-		if penalize {
-			s.markFailure(ctx, cand.ch.ID)
-		}
-		excluded[cand.ch.ID] = true
-		meta.Retries++
-		lastErr = err
-	}
-	return nil, meta, lastErr
+		return s.client.GenAiCompatResponsesStream(ctx, cred, cand.ch.Region, raw, s.UpstreamWait())
+	})
 }
 
 // firstErr 在换渠道后仍失败时优先返回上游错误(而非「无渠道」)。
@@ -221,34 +206,11 @@ func weightedPick(chs []model.AiChannel) model.AiChannel {
 	return chs[len(chs)-1]
 }
 
-// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道(整请求上限 3 次)。
+// Embeddings 编排向量化调用:按 EMBEDDING 能力选渠道,可重试错误换渠道。
 func (s *AiGatewayService) Embeddings(ctx context.Context, req aiwire.EmbeddingsRequest, group string) (*aiwire.EmbeddingsResponse, ChatMeta, error) {
-	meta := ChatMeta{}
-	excluded := map[uint]bool{}
-	var lastErr error
-	for attempt := 0; attempt < 3; attempt++ {
-		cand, err := s.pick(ctx, req.Model, group, "EMBEDDING", excluded)
-		if err != nil {
-			return nil, meta, firstErr(lastErr, err)
-		}
-		meta.ChannelID, meta.ChannelName = cand.ch.ID, cand.ch.Name
-		resp, err := s.embedOnce(ctx, cand, req)
-		if err == nil {
-			s.markSuccess(ctx, cand.ch.ID)
-			return resp, meta, nil
-		}
-		retry, penalize := switchable(err)
-		if !retry {
-			return nil, meta, err
-		}
-		if penalize {
-			s.markFailure(ctx, cand.ch.ID)
-		}
-		excluded[cand.ch.ID] = true
-		meta.Retries++
-		lastErr = err
-	}
-	return nil, meta, lastErr
+	return routeRetry(ctx, s, req.Model, group, "EMBEDDING", func(cand *aiCandidate) (*aiwire.EmbeddingsResponse, error) {
+		return s.embedOnce(ctx, cand, req)
+	})
 }
 
 // embedOnce 调用渠道向量化并装配 OpenAI 形态响应。

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)
+	}
+}

internal/service/webconsole.go

@@ -46,6 +46,7 @@ type ConsoleSession struct {
 type ConsoleService struct {
 	configs      *OciConfigService
 	pollInterval time.Duration // 清理残留连接的轮询间隔，测试注入缩短
+	sessionTTL   time.Duration // 会话回收检查周期，测试注入缩短
 
 	mu       sync.Mutex
 	sessions map[string]*ConsoleSession
@@ -55,6 +56,7 @@ func NewConsoleService(configs *OciConfigService) *ConsoleService {
 	return &ConsoleService{
 		configs:      configs,
 		pollInterval: 2 * time.Second,
+		sessionTTL:   consoleSessionTTL,
 		sessions:     map[string]*ConsoleSession{},
 	}
 }
@@ -161,11 +163,12 @@ func (s *ConsoleService) storeSession(cfgID uint, instanceID, region, typ, connI
 	s.mu.Lock()
 	s.sessions[sess.ID] = sess
 	s.mu.Unlock()
-	time.AfterFunc(consoleSessionTTL, func() { s.expire(sess.ID) })
+	time.AfterFunc(s.sessionTTL, func() { s.expire(sess.ID) })
 	return sess
 }
 
-// expire TTL 到期回收：正在使用的会话跳过（连接断开后自然停止，无续期）。
+// expire TTL 到期回收：正在使用的会话跳过并重挂下一轮检查，
+// 连接断开后由后续轮次回收，避免会话与云端连接常驻到进程退出。
 func (s *ConsoleService) expire(id string) {
 	s.mu.Lock()
 	sess, ok := s.sessions[id]
@@ -173,6 +176,7 @@ func (s *ConsoleService) expire(id string) {
 		sess.mu.Lock()
 		if sess.inUse {
 			ok = false
+			time.AfterFunc(s.sessionTTL, func() { s.expire(id) })
 		} else {
 			delete(s.sessions, id)
 		}

internal/service/webconsole_test.go

@@ -116,3 +116,44 @@ func TestCreateConsoleSession(t *testing.T) {
 		})
 	}
 }
+
+// TestConsoleSessionExpire 验证过期回收:在用会话跳过本轮,断开后下一轮回收。
+func TestConsoleSessionExpire(t *testing.T) {
+	tests := []struct {
+		name      string
+		inUse     bool
+		wantAlive bool
+	}{
+		{name: "在用会话跳过回收", inUse: true, wantAlive: true},
+		{name: "空闲会话回收并删云端连接", inUse: false, wantAlive: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			client := &consoleFakeClient{}
+			console, cfgID := newConsoleTestService(t, client)
+			sess, err := console.CreateSession(context.Background(), cfgID, "ocid1.instance.i", "", ConsoleTypeSerial)
+			if err != nil {
+				t.Fatalf("CreateSession: %v", err)
+			}
+			sess.MarkUse(tt.inUse)
+			console.expire(sess.ID)
+			if alive := console.Get(sess.ID) != nil; alive != tt.wantAlive {
+				t.Fatalf("session alive = %v, want %v", alive, tt.wantAlive)
+			}
+			if wantDel := !tt.wantAlive; (len(client.deleted) == 1) != wantDel {
+				t.Errorf("cloud connection deleted %v, want %v", client.deleted, wantDel)
+			}
+			if tt.inUse {
+				// 断开后下一轮回收(此前的缺陷:跳过后不再检查,会话常驻)
+				sess.MarkUse(false)
+				console.expire(sess.ID)
+				if console.Get(sess.ID) != nil {
+					t.Fatal("session still alive after idle expire round")
+				}
+				if len(client.deleted) != 1 {
+					t.Errorf("cloud connection not deleted after idle expire: %v", client.deleted)
+				}
+			}
+		})
+	}
+}