refactor(gateway): ModelCallResult 拆分为 SubmitModelCall + WaitModelCallResult

- SubmitModelCall: 提交段（异步模型生成唯一 msgTopic），返回 (res, msgTopic, err) - WaitModelCallResult: 订阅段（重订阅可恢复结果，供 Task5 AsyncModelCallWithRecovery 复用） - ModelCallResult 签名不变，内部复用二者；订阅参数与拆分前一致 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-08-25 16:04:13 +08:00
parent 5672c8272e
commit 3620e1ea61
2 changed files with 73 additions and 44 deletions
@@ -132,12 +132,11 @@ func GetModelInfoById(ctx context.Context, req *GetModelInfoByIdReq) (res *GetMo
 	return
 }

-// ModelCallResult 调模型网关生成一段内容并等待结果。
-// 视频等异步模型为"提交+等待"一体：内部订阅 GMQ 直到结果返回。
-func ModelCallResult(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (responseParams *ModelCallRes, err error) {
+// SubmitModelCall 提交模型调用（异步模型自动生成唯一 msgTopic），返回提交响应与消息主题。
+// taskId 在 res.TaskId；同步模型 msgTopic 为空串。结果等待由 WaitModelCallResult 完成。
+func SubmitModelCall(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (res *ModelCallRes, msgTopic string, err error) {
 	// 异步模型必须绑定消息主题接收结果：自动生成唯一主题，
 	// 带业务标识（bizName/modelId/sessionId）便于排查，每次调用唯一避免并发串结果
-	msgTopic := ""
 	if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
 		msgTopic = modelCallTopic(g.Cfg().MustGet(ctx, "server.name").String(), modelId, sessionId)
 	}
@@ -150,62 +149,77 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp
 		BusinessParams: businessParams,
 		MsgTopic:       msgTopic,
 	}
-	// 2. 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout
+	// 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout
 	client := commonHttp.Httpclient.Clone()
 	client.SetTimeout(30 * time.Minute)
 	if tr, ok := client.Transport.(*http.Transport); ok {
 		tr.ResponseHeaderTimeout = 30 * time.Minute
 	}

-	res := new(ModelCallRes)
+	res = new(ModelCallRes)
 	err = commonHttp.Post(ctx, "model-gateway/model/call/modelCall", requestHeaders(ctx), res, &req)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	if g.IsEmpty(res.TaskId) || !g.IsEmpty(res.ErrorMsg) {
-		return nil, fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg)
+		return nil, "", fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg)
 	}
-	// 3. 订阅模型结果（异步模型）
-	if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
-		resultCh := make(chan *ModelCallRes, 1)
-		errCh := make(chan error, 1)
+	return res, msgTopic, nil
+}

-		_, err = gmq.GetGmq(public.GmqMsgPluginsName).GmqSubscribe(ctx, &mq.NatsSubMessage{
-			SubMessage: types.SubMessage{
-				Topic:           msgTopic,
-				ConsumerName:    fmt.Sprintf("model-call-result-%s", uuid.NewString()),
-				AutoAck:         false,
-				AutoUnsubscribe: true,
-				FetchCount:      1,
-				HandleFunc: func(ctx context.Context, msg any) error {
-					r := new(ModelCallRes)
-					if err := gconv.Struct(msg, r); err != nil {
-						errCh <- err
-						return nil
-					}
-					if g.IsEmpty(r.TaskId) || !g.IsEmpty(r.ErrorMsg) {
-						errCh <- fmt.Errorf("创建模型任务失败:%v", r.ErrorMsg)
-						return nil
-					}
-					resultCh <- r
+// WaitModelCallResult 订阅 msgTopic 等待异步模型结果。
+// 重订阅可拿回已发布结果：gmq NATS 侧 Retention=LimitsPolicy+MaxAge=7天（消息不随 ack 删除、保留 7 天），
+// 订阅用 DeliverPolicy=DeliverAllPolicy（新订阅从 stream 序头重放），崩溃后重订阅同一 msgTopic 即可恢复。
+func WaitModelCallResult(ctx context.Context, msgTopic string) (responseParams *ModelCallRes, err error) {
+	resultCh := make(chan *ModelCallRes, 1)
+	errCh := make(chan error, 1)
+
+	_, err = gmq.GetGmq(public.GmqMsgPluginsName).GmqSubscribe(ctx, &mq.NatsSubMessage{
+		SubMessage: types.SubMessage{
+			Topic:           msgTopic,
+			ConsumerName:    fmt.Sprintf("model-call-result-%s", uuid.NewString()),
+			AutoAck:         false,
+			AutoUnsubscribe: true,
+			FetchCount:      1,
+			HandleFunc: func(ctx context.Context, msg any) error {
+				r := new(ModelCallRes)
+				if err := gconv.Struct(msg, r); err != nil {
+					errCh <- err
 					return nil
-				},
+				}
+				if g.IsEmpty(r.TaskId) || !g.IsEmpty(r.ErrorMsg) {
+					errCh <- fmt.Errorf("创建模型任务失败:%v", r.ErrorMsg)
+					return nil
+				}
+				resultCh <- r
+				return nil
 			},
-			Durable: true,
-		})
-		if err != nil {
-			return
-		}
-
-		select {
-		case responseParams = <-resultCh:
-		case err = <-errCh:
-			return nil, err
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		}
+		},
+		Durable: true,
+	})
+	if err != nil {
 		return
 	}

+	select {
+	case responseParams = <-resultCh:
+	case err = <-errCh:
+		return nil, err
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	return
+}
+
+// ModelCallResult 调模型网关生成一段内容并等待结果（内部复用 SubmitModelCall + WaitModelCallResult）。
+// 同步模型提交即返回；异步模型为"提交+等待"一体。
+func ModelCallResult(ctx context.Context, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (responseParams *ModelCallRes, err error) {
+	res, msgTopic, err := SubmitModelCall(ctx, modelId, responseType, sessionId, requestParams, businessParams)
+	if err != nil {
+		return nil, err
+	}
+	if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
+		return WaitModelCallResult(ctx, msgTopic)
+	}
 	return res, nil
 }
@@ -0,0 +1,15 @@
+package gateway
+
+import (
+	"context"
+	"testing"
+
+	"ai-agent/workflow/consts/model"
+)
+
+// TestSplitSignatures 只做编译期签名验证：拆分后的函数存在且签名符合 memo 包装契约。
+// 不做实际调用（依赖 live model-gateway + Consul），实际行为由端到端走查覆盖。
+func TestSplitSignatures(t *testing.T) {
+	var _ func(context.Context, int64, model.ResponseType, string, map[string]any, map[string]any) (*ModelCallRes, string, error) = SubmitModelCall
+	var _ func(context.Context, string) (*ModelCallRes, error) = WaitModelCallResult
+}