fix: 修复请求头转发与任务状态流转问题

移除 util.ForwardHeaders，改为从原始请求精确提取 Authorization 或全部请求头；
任务创建时直接设为 Running 状态，避免二次更新与查询；
模型调用使用独立超时上下文，防止外层取消影响回调；
增加 OSS 上传耗时日志，调整数据库连接池参数。

common/util/mapping.go

@@ -19,17 +19,20 @@ import (
 	tgjson "github.com/tidwall/gjson"
 )
 
-// ParseAndValidate 解析并校验结果
+// ParseAndValidate 解析模型响应，并返回标准格式
 func ParseAndValidate(raw map[string]any, model *entity.ModelGatewayModel) (map[string]any, error) {
-	// 1) 解析 content 字符串为 rounds 数组
+	contentStr := gconv.String(raw[entity.ResponseBody])
-	contentVal, ok := raw[entity.ResponseBody]
+	if strings.TrimSpace(contentStr) == "" {
-	if !ok {
+		return raw, fmt.Errorf("字段 %s 为空", entity.ResponseBody)
-		return raw, fmt.Errorf("字段 %s 不存在", entity.ResponseBody)
 	}
-	contentStr, ok := contentVal.(string)
+
-	if !ok || strings.TrimSpace(contentStr) == "" {
+	contentStr = strings.Map(func(r rune) rune {
-		return raw, fmt.Errorf("字段 %s 为空或不是字符串", entity.ResponseBody)
+		if r < 32 && r != ' ' {
+			return -1
 		}
+		return r
+	}, contentStr)
+
 	var arr []any
 	if err := json.Unmarshal([]byte(contentStr), &arr); err != nil {
 		return raw, fmt.Errorf("JSON解析失败: %w", err)
@@ -38,20 +41,14 @@ func ParseAndValidate(raw map[string]any, model *entity.ModelGatewayModel) (map[
 		return raw, fmt.Errorf("解析后数组为空")
 	}
 
-	// 2) 校验必填字段
-	if len(model.RequiredFields) > 0 {
-		for i, r := range arr {
-			round, ok := r.(map[string]any)
-			if !ok {
-				continue
-			}
 	for _, field := range model.RequiredFields {
-				if gjson.New(round).Get(field).IsNil() {
+		for i, r := range arr {
+			round, _ := r.(map[string]any)
+			if round != nil && gjson.New(round).Get(field).IsNil() {
 				return raw, fmt.Errorf("rounds[%d] 缺少必填字段: %s", i, field)
 			}
 		}
 	}
-	}
 
 	return map[string]any{"total_rounds": len(arr), "rounds": arr}, nil
 }

config.yml

@@ -39,8 +39,8 @@ database:
       dryRun: false
       charset: "utf8"
       timezone: "Asia/Shanghai"
-      maxIdle: 5
+      maxIdle: 15
-      maxOpen: 20
+      maxOpen: 60
       maxLifetime: "30s"
       maxIdleConnTime: "30s"
       createdAt: "created_at"

dao/model_gateway_models_dao.go

@@ -56,6 +56,7 @@ func (d *modelGatewayModelsDao) Get(ctx context.Context, req *entity.ModelGatewa
 		Where(entity.ModelGatewayModelCol.Id, req.Id).
 		Where(entity.ModelGatewayModelCol.Creator, req.Creator).
 		Where(entity.ModelGatewayModelCol.ModelName, req.ModelName).
+		Where(entity.ModelGatewayModelCol.IsChatModel, req.IsChatModel).
 		Fields(fields).One()
 	if err != nil {
 		return nil, err
@@ -122,7 +123,7 @@ func (d *modelGatewayModelsDao) GetByAcrossTenant(ctx context.Context, req *enti
 func (d *modelGatewayModelsDao) GetByCreatorAndPlatform(ctx context.Context, req *dto.ListModelReq) (list []*entity.ModelGatewayModel, total int, err error) {
 	sql := `
 SELECT DISTINCT ON (model_name) *
-FROM asynch_models
+FROM ` + public.TableNameModel + `
 WHERE deleted_at IS NULL 
   AND (? = '' OR model_name LIKE ?)
 `

dao/model_gateway_task_dao.go

@@ -7,7 +7,6 @@ import (
 	"model-gateway/model/entity"
 
 	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
-	"github.com/gogf/gf/v2/database/gdb"
 	"github.com/gogf/gf/v2/util/gconv"
 )
 
@@ -128,32 +127,32 @@ func (d *modelGatewayTaskDao) GetPendingAsyncTasks(ctx context.Context, limit in
 
 // ClaimByID 按主键抢占，返回抢占后的任务
 func (d *modelGatewayTaskDao) ClaimByID(ctx context.Context, id int64) (*entity.ModelGatewayTask, error) {
+	// 1) 先查任务
 	var task entity.ModelGatewayTask
-	err := gfdb.DB(ctx, public.DbNameModelGateway).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
+	r, err := gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameTask).
-		r, err := tx.Model(public.TableNameTask).
 		Where(entity.ModelGatewayTaskCol.Id, id).
 		Where(entity.ModelGatewayTaskCol.State, public.TaskStatusPending).
-			Limit(1).
-			LockUpdate().
 		One()
-		if err != nil {
-			return err
-		}
-		if r.IsEmpty() {
-			return fmt.Errorf("任务已被抢占或不存在: id=%d", id)
-		}
-		if err := r.Struct(&task); err != nil {
-			return err
-		}
-		_, err = tx.Model(public.TableNameTask).
-			Data(&entity.ModelGatewayTask{State: public.TaskStatusRunning}).
-			Where(entity.ModelGatewayTaskCol.Id, id).
-			OmitEmpty().
-			Update()
-		return err
-	})
 	if err != nil {
 		return nil, err
 	}
+	if r.IsEmpty() {
+		return nil, fmt.Errorf("任务已被抢占或不存在: id=%d", id)
+	}
+	if err = r.Struct(&task); err != nil {
+		return nil, err
+	}
+
+	// 2) 改为执行中
+	_, err = gfdb.DB(ctx, public.DbNameModelGateway).Model(ctx, public.TableNameTask).
+		Data(&entity.ModelGatewayTask{State: public.TaskStatusRunning}).
+		Where(entity.ModelGatewayTaskCol.Id, id).
+		Where(entity.ModelGatewayTaskCol.State, public.TaskStatusPending). // 防并发
+		OmitEmpty().
+		Update()
+	if err != nil {
+		return nil, err
+	}
+
 	return &task, nil
 }

model/dto/model_gateway_models_dto.go

@@ -25,6 +25,8 @@ type CreateModelReq struct {
 	Form               []map[string]any `p:"form" json:"form" dc:"动态表单配置"`
 	RequestMapping     map[string]any   `p:"requestMapping" json:"requestMapping" dc:"请求映射"`
 	ResponseMapping    map[string]any   `p:"responseMapping" json:"responseMapping" dc:"返回映射"`
+	ResponseBody       string           `p:"responseBody" json:"responseBody" dc:"返回主体"`
+	ResponseTokenField string           `p:"responseTokenField" json:"responseTokenField" dc:"响应中消耗token的字段映射"`
 	OperatorName       string           `p:"operatorName" json:"operatorName" dc:"运营商名称"`
 	TokenConfig        map[string]any   `p:"tokenConfig" json:"tokenConfig" dc:"token计算配置"`
 	ExtendMapping      map[string]any   `p:"extendMapping" json:"extendMapping" dc:"附加映射"`
@@ -61,6 +63,8 @@ type UpdateModelReq struct {
 	Form               []map[string]any `p:"form" json:"form" dc:"动态表单配置"`
 	RequestMapping     map[string]any   `p:"requestMapping" json:"requestMapping" dc:"请求映射"`
 	ResponseMapping    map[string]any   `p:"responseMapping" json:"responseMapping" dc:"返回映射"`
+	ResponseBody       string           `p:"responseBody" json:"responseBody" dc:"返回主体"`
+	ResponseTokenField string           `p:"responseTokenField" json:"responseTokenField" dc:"响应中消耗token的字段映射"`
 	OperatorName       string           `p:"operatorName" json:"operatorName" dc:"运营商名称"`
 	TokenConfig        map[string]any   `p:"tokenConfig" json:"tokenConfig" dc:"token计算配置"`
 	ExtendMapping      map[string]any   `p:"extendMapping" json:"extendMapping" dc:"附加映射"`

model/entity/model_gateway_model.go

@@ -12,7 +12,6 @@ type modelGatewayModelCol struct {
 	FormJSON         string
 	RequestMapping   string
 	ResponseMapping  string
-	ResponseBody     string
 	RequiredFields   string
 	IsPrivate        string
 	IsChatModel      string
@@ -31,6 +30,7 @@ type modelGatewayModelCol struct {
 	StreamConfig     string
 	FirstFrame       string
 	LastFrame        string
+	MaxTokens        string
 }
 
 var ModelGatewayModelCol = modelGatewayModelCol{
@@ -61,6 +61,7 @@ var ModelGatewayModelCol = modelGatewayModelCol{
 	StreamConfig:     "stream_config",
 	FirstFrame:       "first_frame",
 	LastFrame:        "last_frame",
+	MaxTokens:        "max_tokens",
 }
 
 type ModelGatewayModel struct {
@@ -91,9 +92,10 @@ type ModelGatewayModel struct {
 	StreamConfig     map[string]any   `orm:"stream_config" json:"streamConfig"`
 	FirstFrame       string           `orm:"first_frame" json:"firstFrame"`
 	LastFrame        string           `orm:"last_frame" json:"lastFrame"`
+	MaxTokens        int              `orm:"max_tokens" json:"maxTokens"`
 }
 
-const ( //ResponseMapping 下的字段
+const (
-	ResponseBody = "response_body" //返回主体
+	ResponseBody = "content"      //返回主体（必填）
 	TotalTokens  = "total_tokens" //总token数
 )

service/gateway/gateway_http_service.go

@@ -7,7 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"mime/multipart"
-	"model-gateway/common/util"
 	"model-gateway/model/entity"
 	"time"
 
@@ -43,16 +42,25 @@ func UploadByTask(ctx context.Context, data []byte, fileExt string) (oss *Upload
 	if err != nil {
 		return nil, err
 	}
-	if _, err := part.Write(data); err != nil {
+	if _, err = part.Write(data); err != nil {
 		return nil, err
 	}
-	contentType := writer.FormDataContentType()
+	//contentType := writer.FormDataContentType()
 	if err = writer.Close(); err != nil {
 		return nil, err
 	}
 
-	headers := util.ForwardHeaders(ctx)
+	//headers := util.ForwardHeaders(ctx)
-	headers["Content-Type"] = contentType
+	//headers["Content-Type"] = contentType
+
+	headers := make(map[string]string)
+	headers["Content-Type"] = writer.FormDataContentType()
+	if r := g.RequestFromCtx(ctx); r != nil {
+		if auth := r.Header.Get("Authorization"); auth != "" {
+			headers["Authorization"] = auth
+		}
+	}
+
 	fullURL := "oss/file/uploadFile"
 	g.Log().Infof(ctx, "[OSS] upload start url=%s filename=%s size=%d", fullURL, filename, len(data))
 
@@ -78,15 +86,25 @@ type CallbackPayload struct {
 
 // TriggerCallback 任务的回调
 func TriggerCallback(ctx context.Context, t *entity.ModelGatewayTask) {
-	headers := util.ForwardHeaders(ctx)
+	//headers := util.ForwardHeaders(ctx)
+	headers := make(map[string]string)
+	if r := g.RequestFromCtx(ctx); r != nil {
+		for k, v := range r.Request.Header {
+			if len(v) > 0 {
+				headers[k] = v[0]
+			}
+		}
+	}
 	var resp struct{}
 	payload := CallbackPayload{
 		TaskId:   t.TaskID,
 		State:    t.State,
-		OssFile:  t.ResultFile.OssFile,
-		FileType: t.ResultFile.FileType,
 		ErrorMsg: t.ErrorMsg,
 	}
+	if !g.IsEmpty(t.ResultFile) {
+		payload.OssFile = t.ResultFile.OssFile
+		payload.FileType = t.ResultFile.FileType
+	}
 	jsonData, err := json.Marshal(payload)
 	if err != nil {
 		g.Log().Warningf(ctx, "[回调] JSON序列化失败 taskId=%s 错误=%v", t.TaskID, err)
@@ -112,7 +130,15 @@ type PromptsCallbackPayload struct {
 // TriggerPromptsCallback 任务成功后的提示词回调
 func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epicycleId int64) {
 	callbackURL := "prompts-core/session/callback"
-	headers := util.ForwardHeaders(ctx)
+	//headers := util.ForwardHeaders(ctx)
+	headers := make(map[string]string)
+	if r := g.RequestFromCtx(ctx); r != nil {
+		for k, v := range r.Request.Header {
+			if len(v) > 0 {
+				headers[k] = v[0]
+			}
+		}
+	}
 	var resp struct{}
 	payload := PromptsCallbackPayload{
 		EpicycleId: epicycleId,
@@ -136,7 +162,15 @@ func TriggerPromptsCallback(ctx context.Context, t *entity.ModelGatewayTask, epi
 
 // IsSuperAdmin 调用admin-go服务检查是否是超级管理员
 func IsSuperAdmin(ctx context.Context) (res bool, err error) {
-	headers := util.ForwardHeaders(ctx)
+	//headers := util.ForwardHeaders(ctx)
+	headers := make(map[string]string)
+	if r := g.RequestFromCtx(ctx); r != nil {
+		for k, v := range r.Request.Header {
+			if len(v) > 0 {
+				headers[k] = v[0]
+			}
+		}
+	}
 	var r = make(map[string]bool)
 	if err = commonHttp.Get(ctx, "admin-go/api/v1/system/user/checkIsSuperAdmin", headers, &r); err != nil {
 		return false, err

service/model/model_service.go

@@ -99,7 +99,7 @@ func (s *modelService) Get(ctx context.Context, req *dto.GetModelReq) (*dto.GetM
 		ModelName:   req.ModelName,
 		IsChatModel: req.IsChatModel,
 	})
-	if err != nil {
+	if err != nil || model == nil {
 		return nil, err
 	}
 	return &dto.GetModelRes{

service/task/task_service.go

@@ -65,20 +65,20 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res *
 		Body:    req.RequestPayload,
 		Headers: util.ParseHeadMsgHeaders(model.HeadMsg),
 	}
-	id, err := dao.ModelGatewayTask.Insert(ctx, &entity.ModelGatewayTask{
+	task := new(entity.ModelGatewayTask)
-		ModelName:      req.ModelName,
+	task.ModelName = model.ModelName
-		TaskID:         taskID,
+	task.TaskID = taskID
-		State:          public.TaskStatusPending,
+	task.State = public.TaskStatusRunning
-		BizName:        req.BizName,
+	task.BizName = req.BizName
-		CallbackURL:    req.CallbackUrl,
+	task.CallbackURL = req.CallbackUrl
-		RequestPayload: &requestPayload,
+	task.RequestPayload = &requestPayload
-		EpicycleId:     req.EpicycleId,
+	task.EpicycleId = req.EpicycleId
-	})
+	id, err := dao.ModelGatewayTask.Insert(ctx, task)
 	if err != nil { // 入库失败：回滚闸门占位
 		queue.ReleaseQueueSlot(ctx, req.ModelName, taskID)
 		return nil, err
 	}
-
+	task.Id = id
 	// 4) 写操作日志（不影响主流程，失败忽略）
 	ip := ""
 	ua := ""
@@ -107,13 +107,27 @@ func (s *taskService) Create(ctx context.Context, req *dto.CreateTaskReq) (res *
 		},
 	})
 
-	// 5) 获取任务信息
+	//// 5) 抢占任务：改为执行中
-	task, err := dao.ModelGatewayTask.ClaimByID(ctx, id)
+	//rows, err := dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{
-	if err != nil {
+	//	SQLBaseDO: beans.SQLBaseDO{Id: id},
-		return nil, err
+	//	State:     public.TaskStatusRunning,
-	}
+	//})
+	//if err != nil {
+	//	return nil, err
+	//}
+	//if rows == 0 {
+	//	return nil, fmt.Errorf("任务不存在: id=%d", id)
+	//}
 
-	// 5) 创建成功后立即异步尝试执行当前任务
+	// 6) 查询任务信息
+	//task, err := dao.ModelGatewayTask.Get(ctx, &entity.ModelGatewayTask{
+	//	SQLBaseDO: beans.SQLBaseDO{Id: id},
+	//})
+	//if err != nil {
+	//	return nil, err
+	//}
+
+	// 7) 创建成功后立即异步尝试执行当前任务
 	go AsyncWorker.handleOne(util.AsyncCtx(ctx), task, model, req)
 
 	return &dto.CreateTaskRes{TaskID: taskID}, nil

service/task/worker.go

@@ -67,12 +67,19 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
 	// ============================================
 	// 2) 调用模型
 	// ============================================
+	for attempt := 0; ; attempt++ {
+		if attempt > 0 {
+			g.Log().Infof(ctx, "[执行任务][重试] 调用模型 第%d次 taskId=%s", attempt, task.TaskID)
+			time.Sleep(time.Duration(attempt) * time.Second)
+		}
+
 		switch {
 		case model.CallMode != nil && *model.CallMode == public.CallModeStream:
 			rawBytes, streamErr := w.callModelStream(ctx, task, model, body)
 			if streamErr != nil {
-			w.failTask(ctx, task, startTime, streamErr.Error())
+				err = streamErr
-			return
+				g.Log().Warningf(ctx, "[执行任务][调用失败] taskId=%s attempt=%d err=%v", task.TaskID, attempt, err)
+				continue
 			}
 			result, err = util.ParseStreamResponse(rawBytes, model.StreamConfig)
 		case model.CallMode != nil && *model.CallMode == public.CallModeAsync:
@@ -83,10 +90,17 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
 		default:
 			result, err = w.callModel(ctx, task, model, body)
 		}
-	if err != nil {
+
+		if err == nil {
+			break
+		}
+
+		if !strings.Contains(err.Error(), "Timeout") {
 			w.failTask(ctx, task, startTime, err.Error())
 			return
 		}
+		g.Log().Warningf(ctx, "[执行任务][调用失败] taskId=%s attempt=%d err=%v", task.TaskID, attempt, err)
+	}
 
 	// ============================================
 	// 3) 缓存临时文件
@@ -115,10 +129,14 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
 		if attempt > 0 {
 			g.Log().Infof(ctx, "[执行任务][重试] OSS上传 第%d/%d次 taskId=%s", attempt, maxRetry, task.TaskID)
 		}
+		startUpload := time.Now()
 		oss, err = gateway.UploadByTask(ctx, gjson.New(result).MustToJson(), "json")
 		if err == nil {
 			break
 		}
+		cost := time.Since(startUpload)
+		g.Log().Infof(ctx, "本次上传耗时:%s", cost)
+
 		g.Log().Errorf(ctx, "[执行任务][失败] OSS上传失败 taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, err)
 		if attempt == maxRetry {
 			task.State = public.TaskStatusFailed
@@ -147,9 +165,9 @@ func (w *asyncWorker) handleOne(ctx context.Context, task *entity.ModelGatewayTa
 	}
 
 	queue.ReleaseQueueSlot(ctx, task.ModelName, task.TaskID)
-	go gateway.TriggerCallback(context.WithoutCancel(ctx), task)
+	go gateway.TriggerCallback(util.AsyncCtx(ctx), task)
 	if req.EpicycleId != 0 {
-		go gateway.TriggerPromptsCallback(context.WithoutCancel(ctx), task, req.EpicycleId)
+		go gateway.TriggerPromptsCallback(util.AsyncCtx(ctx), task, req.EpicycleId)
 	}
 
 	g.Log().Infof(ctx, "[执行任务][成功] taskId=%s duration=%ds fileType=%s",
@@ -294,6 +312,8 @@ func (w *asyncWorker) callModel(ctx context.Context, task *entity.ModelGatewayTa
 
 // parseAndRetry 解析模型返回结果，并重试
 func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, task *entity.ModelGatewayTask, model *entity.ModelGatewayModel, req *dto.CreateTaskReq, maxRetry int, startTime time.Time) (map[string]any, error) {
+	var lastErr error
+
 	for attempt := 0; attempt <= maxRetry; attempt++ {
 		if attempt > 0 {
 			g.Log().Infof(ctx, "[执行任务][重试] JSON解析 第%d/%d次 taskId=%s", attempt, maxRetry, task.TaskID)
@@ -302,6 +322,7 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta
 		// 1) 响应映射
 		mapped, err := util.MapResponsePayload(model.ResponseMapping, body)
 		if err != nil {
+			lastErr = err
 			g.Log().Warningf(ctx, "[执行任务][映射失败] taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, err)
 			if attempt == maxRetry {
 				return nil, fmt.Errorf("响应映射重试耗尽: %w", err)
@@ -309,10 +330,10 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta
 			continue
 		}
 
-		// 2) 先存 token 到数据库，防止后续失败丢失
+		// 2) 存 token
 		if _, ok := mapped[entity.TotalTokens]; ok {
 			task.ExpendTokens = gconv.Int64(mapped[entity.TotalTokens])
-			_, err = dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{
+			_, _ = dao.ModelGatewayTask.Update(ctx, &entity.ModelGatewayTask{
 				SQLBaseDO:    beans.SQLBaseDO{Id: task.Id},
 				ExpendTokens: task.ExpendTokens,
 			})
@@ -326,9 +347,9 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta
 			if err == nil {
 				return parsed, nil
 			}
+			lastErr = err
 		case public.BuildTypeStruct:
-			parsed = util.ParseStructResult(mapped, entity.ResponseBody)
+			return util.ParseStructResult(mapped, entity.ResponseBody), nil
-			return parsed, nil
 		default:
 			return mapped, nil
 		}
@@ -336,22 +357,22 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta
 		g.Log().Warningf(ctx, "[执行任务][解析失败] taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, err)
 
 		if attempt == maxRetry {
-			return nil, fmt.Errorf("JSON解析重试耗尽: %w", err)
+			return nil, fmt.Errorf("JSON解析重试耗尽: %w", lastErr)
 		}
 
-		// 4) 重新调模型（直接调，不走缓存）
+		// 4) 拼接错误信息到请求体，重调模型
 		task.RetryCount++
 		_, _ = dao.ModelGatewayTask.Update(ctx, task)
-		rawData, callErr := InvokeModel(ctx, model, task.RequestPayload.Body)
 
+		body = injectErrorMessage(task.RequestPayload.Body, lastErr)
+		rawData, callErr := InvokeModel(ctx, model, body)
 		if callErr != nil {
 			g.Log().Warningf(ctx, "[执行任务][重调模型失败] taskId=%s attempt=%d/%d err=%v", task.TaskID, attempt, maxRetry, callErr)
 			continue
 		}
 
-		// 5) 解析原始响应，覆盖 body 进入下一轮
 		var rawResp map[string]any
-		if err = json.Unmarshal(rawData, &rawResp); err != nil {
+		if err := json.Unmarshal(rawData, &rawResp); err != nil {
 			g.Log().Warningf(ctx, "[执行任务][Unmarshal失败] taskId=%s err=%v", task.TaskID, err)
 			continue
 		}
@@ -361,11 +382,49 @@ func (w *asyncWorker) parseAndRetry(ctx context.Context, body map[string]any, ta
 	return body, nil
 }
 
+// injectErrorMessage 将错误信息拼接到 user 消息中
+func injectErrorMessage(payload map[string]any, err error) map[string]any {
+	if err == nil {
+		return payload
+	}
+
+	messages, _ := payload["messages"].([]any)
+	if len(messages) == 0 {
+		return payload
+	}
+
+	errMsg := fmt.Sprintf("\n\n【上一轮输出错误，请修正】%s", err.Error())
+
+	// 找到最后一个 role=user 的消息，追加错误提示
+	for i := len(messages) - 1; i >= 0; i-- {
+		msg, ok := messages[i].(map[string]any)
+		if !ok {
+			continue
+		}
+		if gconv.String(msg["role"]) != "user" {
+			continue
+		}
+
+		switch c := msg["content"].(type) {
+		case string:
+			msg["content"] = c + errMsg
+		case []any:
+			msg["content"] = append(c, map[string]any{
+				"type": "text",
+				"text": errMsg,
+			})
+		}
+		break
+	}
+
+	return payload
+}
+
 // InvokeModel 调用模型服务，返回二进制结果
 // modelKey 用于覆盖/补充模型配置 head_msg（例如每次请求携带不同的 X-API-Key）
 func InvokeModel(ctx context.Context, model *entity.ModelGatewayModel, body map[string]any) ([]byte, error) {
 	// 1) 记录模型调用次数
-	_ = dao.ModelGatewayLogsStat.IncRequestCount(ctx, time.Now(), model.TenantId, model.Creator, model.ModelName)
+	//_ = dao.ModelGatewayLogsStat.IncRequestCount(ctx, time.Now(), model.TenantId, model.Creator, model.ModelName)
 
 	// 2）请求参数映射：将标准 payload 按模型配置的 requestMapping 转为模型需要的格式
 	//—— 请求映射实际处理为提示词构建请求，因为有附加字段及其他字段的拼接。这里不方便做请求映射
@@ -392,13 +451,20 @@ func InvokeModel(ctx context.Context, model *entity.ModelGatewayModel, body map[
 				baseURL = baseURL + "?" + q.Encode()
 			}
 		}
-		req, err = http.NewRequestWithContext(ctx, http.MethodGet, baseURL, nil)
+		// 改用独立超时ctx，隔绝外层截止
+		reqCtx, reqCancel := context.WithTimeout(context.Background(), timeout)
+		defer reqCancel()
+		req, err = http.NewRequestWithContext(reqCtx, http.MethodGet, baseURL, nil)
+		//req, err = http.NewRequestWithContext(ctx, http.MethodGet, baseURL, nil)
 	default:
 		bodyBytes, err := json.Marshal(body)
 		if err != nil {
 			return nil, err
 		}
-		req, err = http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes))
+		reqCtx, reqCancel := context.WithTimeout(context.Background(), timeout)
+		defer reqCancel()
+		req, err = http.NewRequestWithContext(reqCtx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes))
+		//req, err = http.NewRequestWithContext(ctx, http.MethodPost, baseURL, bytes.NewReader(bodyBytes))
 	}
 
 	// 5）注入请求头：先模型静态配置，再动态 modelKey（后者可覆盖前者）
@@ -497,5 +563,5 @@ func (w *asyncWorker) failTask(ctx context.Context, t *entity.ModelGatewayTask,
 		g.Log().Warningf(ctx, "[执行任务][更新数据库失败] taskId=%s err=%v", t.TaskID, err)
 	}
 	queue.ReleaseQueueSlot(ctx, t.ModelName, t.TaskID)
-	go gateway.TriggerCallback(context.WithoutCancel(ctx), t)
+	go gateway.TriggerCallback(util.AsyncCtx(ctx), t)
 }

update.sql

@@ -33,6 +33,7 @@ CREATE TABLE IF NOT EXISTS model_gateway_models (
     max_concurrency int4 NOT NULL DEFAULT 10,
     timeout_seconds int4 NOT NULL DEFAULT 600,
     retry_times     int2 NOT NULL DEFAULT 3,
+    auto_clean_seconds int4 NOT NULL DEFAULT 86400,
     response_token_field varchar(128) NOT NULL DEFAULT '',
     call_mode       int2 NOT NULL DEFAULT 0,
     required_fields jsonb NOT NULL DEFAULT '[]',
@@ -54,7 +55,6 @@ COMMENT ON COLUMN model_gateway_models.created_at IS '创建时间';
 COMMENT ON COLUMN model_gateway_models.updater IS '更新人';
 COMMENT ON COLUMN model_gateway_models.updated_at IS '更新时间';
 COMMENT ON COLUMN model_gateway_models.deleted_at IS '删除时间(软删)';
-
 COMMENT ON COLUMN model_gateway_models.model_name IS '模型名称';
 COMMENT ON COLUMN model_gateway_models.model_type IS '模型类型';
 COMMENT ON COLUMN model_gateway_models.operator_name IS '运营商名称';
@@ -62,11 +62,12 @@ COMMENT ON COLUMN model_gateway_models.base_url IS '模型地址';
 COMMENT ON COLUMN model_gateway_models.http_method IS '请求方式 GET/POST';
 COMMENT ON COLUMN model_gateway_models.head_msg IS '请求头信息';
 COMMENT ON COLUMN model_gateway_models.api_key IS '调用凭证/密钥';
+
 COMMENT ON COLUMN model_gateway_models.is_private IS '是否私有化：0-私有 1-公共';
 COMMENT ON COLUMN model_gateway_models.enabled IS '是否启用：0-停用 1-启用';
 COMMENT ON COLUMN model_gateway_models.is_chat_model IS '是否为对话模型：0-否 1-是';
 COMMENT ON COLUMN model_gateway_models.is_owner IS '1=当前用户创建 0=超级管理员';
-COMMENT ON COLUMN model_gateway_models.call_mode IS '调用模式：0-同步 1-异步 2-流式';
+
 COMMENT ON COLUMN model_gateway_models.form_json IS '动态表单结构';
 COMMENT ON COLUMN model_gateway_models.request_mapping IS '请求映射';
 COMMENT ON COLUMN model_gateway_models.response_mapping IS '返回映射';
@@ -80,7 +81,9 @@ COMMENT ON COLUMN model_gateway_models.last_frame IS '尾帧图片参数';
 COMMENT ON COLUMN model_gateway_models.max_concurrency IS '最大并发数';
 COMMENT ON COLUMN model_gateway_models.timeout_seconds IS '调用模型超时(秒)';
 COMMENT ON COLUMN model_gateway_models.retry_times IS '失败重试次数';
+COMMENT ON COLUMN model_gateway_models.auto_clean_seconds IS '任务完成后自动清理时间(秒)';
 COMMENT ON COLUMN model_gateway_models.response_token_field IS '响应中消耗token的字段映射';
+COMMENT ON COLUMN model_gateway_models.call_mode IS '调用模式：0-同步 1-异步 2-流式';
 COMMENT ON COLUMN model_gateway_models.required_fields IS '必选字段列表';
 COMMENT ON COLUMN model_gateway_models.max_tokens IS '最大 token 数，0 表示不传';