From 5672c8272eeae0f2ab63679a3663960ca9daa1f3 Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Tue, 25 Aug 2026 15:55:03 +0800 Subject: [PATCH] =?UTF-8?q?feat(workflow):=20flow=5Fasync=5Ftask=20?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E5=BC=82=E6=AD=A5=E4=BB=BB=E5=8A=A1=E8=A1=A8?= =?UTF-8?q?=E5=AE=9E=E4=BD=93=E4=B8=8E=20DAO=EF=BC=88=E7=89=A9=E7=90=86?= =?UTF-8?q?=E5=88=A0=E9=99=A4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- workflow/consts/public/table_name.go | 1 + workflow/dao/flow/flow_async_task_dao.go | 100 ++++++++++++++++++ workflow/dao/flow/flow_async_task_dao_test.go | 94 ++++++++++++++++ workflow/model/entity/flow_async_task.go | 41 +++++++ 4 files changed, 236 insertions(+) create mode 100644 workflow/dao/flow/flow_async_task_dao.go create mode 100644 workflow/dao/flow/flow_async_task_dao_test.go create mode 100644 workflow/model/entity/flow_async_task.go diff --git a/workflow/consts/public/table_name.go b/workflow/consts/public/table_name.go index 2d78460..9ec131b 100644 --- a/workflow/consts/public/table_name.go +++ b/workflow/consts/public/table_name.go @@ -23,4 +23,5 @@ const ( TableNameExecWorkflow = "exec_workflow" TableNameExecWorkflowResult = "exec_workflow_result" TableNameFlowSegmentResult = "flow_segment_result" + TableNameFlowAsyncTask = "flow_async_task" ) diff --git a/workflow/dao/flow/flow_async_task_dao.go b/workflow/dao/flow/flow_async_task_dao.go new file mode 100644 index 0000000..f72cb4f --- /dev/null +++ b/workflow/dao/flow/flow_async_task_dao.go @@ -0,0 +1,100 @@ +package flow + +import ( + "ai-agent/workflow/consts/public" + "ai-agent/workflow/model/entity" + "context" + "fmt" + + "gitea.redpowerfuture.com/red-future/common/db/gfdb" +) + +const ( + FlowAsyncStateInflight = 0 // 任务已提交，结果未取（执行中/结果已发布未取/失败） + FlowAsyncStateDone = 1 // 成功，结果已缓存 + FlowAsyncStateFailed = 2 // 确定失败 + FlowAsyncSegSentinel = -1 // 非段异步调用的段序号哨兵值 +) + +var FlowAsyncTaskDao = &flowAsyncTaskDao{} + +type flowAsyncTaskDao struct{} + +// Get 查询唯一键 (execution_id, node_id, segment_index) 的记录 +func (d *flowAsyncTaskDao) Get(ctx context.Context, execId int64, nodeId string, segIdx int) (res *entity.FlowAsyncTask, err error) { + r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask). + Where(entity.FlowAsyncTaskCol.ExecutionId, execId). + Where(entity.FlowAsyncTaskCol.NodeId, nodeId). + Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx). + One() + if err != nil { + return nil, err + } + if r.IsEmpty() { + return nil, nil + } + err = r.Struct(&res) + return +} + +// Upsert 提交时写入/更新 in-flight 行：唯一键冲突则更新 model_id/task_id/msg_topic/state（保留已完成结果不动）。 +// 与 flow_segment_result 相同走 OnConflict().Save()；两点相对初稿的调整： +// 1. Result 列是 JSONB，空串会被 PG 拒绝（invalid input syntax for type json）， +// 提交时本无结果，统一写 '{}' 占位（与表列默认值一致）。 +// 2. 必须用 OnDuplicate 限定冲突更新列——GoFrame Save 默认把 Data 里所有列写进 +// ON CONFLICT DO UPDATE SET，若不限定会把已缓存的结果覆盖成 '{}'，与"保留已完成结果"矛盾。 +func (d *flowAsyncTaskDao) Upsert(ctx context.Context, execId int64, nodeId string, segIdx int, modelId, taskId int64, msgTopic string) error { + rec := &entity.FlowAsyncTask{ + ExecutionId: execId, + NodeId: nodeId, + SegmentIndex: segIdx, + ModelId: modelId, + TaskId: taskId, + MsgTopic: msgTopic, + State: FlowAsyncStateInflight, + Result: "{}", + } + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). + Model(ctx, public.TableNameFlowAsyncTask). + Data(rec). + OnConflict(entity.FlowAsyncTaskCol.ExecutionId, entity.FlowAsyncTaskCol.NodeId, entity.FlowAsyncTaskCol.SegmentIndex). + OnDuplicate(entity.FlowAsyncTaskCol.ModelId, entity.FlowAsyncTaskCol.TaskId, entity.FlowAsyncTaskCol.MsgTopic, entity.FlowAsyncTaskCol.State). + Save() + return err +} + +// UpdateByKey 按唯一键更新 state/result（OmitNil 丢弃 nil 字段，map 值非 nil 全写入； +// state=0 也能落库）。Result 列是 JSONB，空串无法写入，统一落 '{}' 表示无结果。 +func (d *flowAsyncTaskDao) UpdateByKey(ctx context.Context, execId int64, nodeId string, segIdx int, state int, result string) error { + resultVal := result + if resultVal == "" { + resultVal = "{}" + } + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask). + Where(entity.FlowAsyncTaskCol.ExecutionId, execId). + Where(entity.FlowAsyncTaskCol.NodeId, nodeId). + Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx). + Data(map[string]any{ + entity.FlowAsyncTaskCol.State: state, + entity.FlowAsyncTaskCol.Result: resultVal, + }). + Update() + return err +} + +// DeleteByKey 物理删除：实体嵌 SQLBaseDO 软删后同键重存无法复活（ON CONFLICT 不含 deleted_at）， +// 与 flow_segment_result 相同约束，须 raw Exec 用物理全名 +func (d *flowAsyncTaskDao) DeleteByKey(ctx context.Context, execId int64, nodeId string, segIdx int) error { + const physicalTable = "black_deacon_flow_async_task" + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). + Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ? AND node_id = ? AND segment_index = ?", physicalTable), execId, nodeId, segIdx) + return err +} + +// DeleteByExecution 清理指定执行的异步任务缓存（exec 成功后调用，与段清理同处） +func (d *flowAsyncTaskDao) DeleteByExecution(ctx context.Context, execId int64) error { + const physicalTable = "black_deacon_flow_async_task" + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). + Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ?", physicalTable), execId) + return err +} diff --git a/workflow/dao/flow/flow_async_task_dao_test.go b/workflow/dao/flow/flow_async_task_dao_test.go new file mode 100644 index 0000000..0d235a9 --- /dev/null +++ b/workflow/dao/flow/flow_async_task_dao_test.go @@ -0,0 +1,94 @@ +package flow + +import ( + "context" + "os" + "testing" + + "ai-agent/workflow/consts/public" + + "gitea.redpowerfuture.com/red-future/common/beans" + "gitea.redpowerfuture.com/red-future/common/db/gfdb" + _ "github.com/gogf/gf/contrib/drivers/pgsql/v2" +) + +// resultJSONB 是 JSONB 读回的规范化形式：PG 对 {"taskId":2026} 做 JSONB 归一化后输出 +// {"taskId": 2026}（冒号后带空格），直接比较字面量会失配，以实际读回为准。 +const resultJSONB = `{"taskId": 2026}` + +func TestFlowAsyncTaskDaoRoundTrip(t *testing.T) { + if os.Getenv("AI_AGENT_TEST_DB") != "1" { + t.Skip("skip: AI_AGENT_TEST_DB not set") + } + // gfdb 的 Hook 依赖 ctx 中注入用户信息（租户/创建人），否则 Insert/Select 报 "token 数据为空" + ctx := context.WithValue(context.Background(), "user", &beans.User{ + UserName: "test-async", TenantId: 1, + }) + const execId, nodeId, seg = 900001, "node-x", 3 + // 清理可能残留 + if err := FlowAsyncTaskDao.DeleteByKey(ctx, execId, nodeId, seg); err != nil { + t.Fatalf("cleanup: %v", err) + } + + if err := FlowAsyncTaskDao.Upsert(ctx, execId, nodeId, seg, 1001, 2026, "model-call-test"); err != nil { + t.Fatalf("upsert: %v", err) + } + rec, err := FlowAsyncTaskDao.Get(ctx, execId, nodeId, seg) + if err != nil || rec == nil { + t.Fatalf("get: %v", err) + } + if rec.TaskId != 2026 || rec.MsgTopic != "model-call-test" || rec.State != FlowAsyncStateInflight { + t.Fatalf("upsert state wrong: %+v", rec) + } + + if err := FlowAsyncTaskDao.UpdateByKey(ctx, execId, nodeId, seg, FlowAsyncStateDone, `{"taskId":2026}`); err != nil { + t.Fatalf("update result: %v", err) + } + rec, err = FlowAsyncTaskDao.Get(ctx, execId, nodeId, seg) + if err != nil || rec == nil { + t.Fatalf("get: %v", err) + } + if rec.State != FlowAsyncStateDone || rec.Result != resultJSONB { + t.Fatalf("result not persisted: %+v", rec) + } + + // 唯一键 upsert：同 (exec,node,seg) 再次写入应更新而非新建 + if err := FlowAsyncTaskDao.Upsert(ctx, execId, nodeId, seg, 1001, 9999, "model-call-test2"); err != nil { + t.Fatalf("re-upsert: %v", err) + } + rec, err = FlowAsyncTaskDao.Get(ctx, execId, nodeId, seg) + if err != nil || rec == nil { + t.Fatalf("get: %v", err) + } + if rec.TaskId != 9999 { + t.Fatalf("re-upsert should update task_id, got %d", rec.TaskId) + } + // OnDuplicate 限定了冲突更新列，已缓存结果不应被覆盖 + if rec.Result != resultJSONB { + t.Fatalf("re-upsert should preserve cached result, got %q", rec.Result) + } + + // 物理删除验证：DeleteByKey 后 Get 为 nil（软删会留下 deleted_at 行且 Get 仍返回） + if err := FlowAsyncTaskDao.DeleteByKey(ctx, execId, nodeId, seg); err != nil { + t.Fatalf("delete: %v", err) + } + rec, err = FlowAsyncTaskDao.Get(ctx, execId, nodeId, seg) + if err != nil { + t.Fatalf("get after delete: %v", err) + } + if rec != nil { + t.Fatalf("物理删除后不应查到残留: %+v", rec) + } + // DeleteByExecution + if err := FlowAsyncTaskDao.Upsert(ctx, execId, nodeId, seg, 1001, 1, "t"); err != nil { + t.Fatalf("upsert: %v", err) + } + if err := FlowAsyncTaskDao.DeleteByExecution(ctx, execId); err != nil { + t.Fatalf("delete by exec: %v", err) + } + if rec, _ := FlowAsyncTaskDao.Get(ctx, execId, nodeId, seg); rec != nil { + t.Fatalf("DeleteByExecution 后不应查到残留") + } + // 清理表残留（跨测试） + gfdb.DB(ctx, public.DbNameBlackDeacon).Exec(ctx, "DELETE FROM black_deacon_flow_async_task WHERE execution_id = ?", execId) +} diff --git a/workflow/model/entity/flow_async_task.go b/workflow/model/entity/flow_async_task.go new file mode 100644 index 0000000..0951781 --- /dev/null +++ b/workflow/model/entity/flow_async_task.go @@ -0,0 +1,41 @@ +package entity + +import "gitea.redpowerfuture.com/red-future/common/beans" + +// FlowAsyncTask 统一异步模型任务表：提交时写入（in-flight），结果回来更新； +// 崩溃恢复靠持久化的 msg_topic 重订阅 NATS 拿回结果，避免重复调用模型 +type FlowAsyncTask struct { + beans.SQLBaseDO `orm:",inherit"` + ExecutionId int64 `orm:"execution_id" json:"executionId" description:"所属执行ID"` + NodeId string `orm:"node_id" json:"nodeId" description:"所属节点ID"` + SegmentIndex int `orm:"segment_index" json:"segmentIndex" description:"段序号；非段调用为-1"` + ModelId int64 `orm:"model_id" json:"modelId" description:"模型ID"` + TaskId int64 `orm:"task_id" json:"taskId" description:"model-gateway任务ID"` + MsgTopic string `orm:"msg_topic" json:"msgTopic" description:"结果消息主题"` + State int `orm:"state" json:"state" description:"0=in-flight,1=done,2=failed"` + Result string `orm:"result" json:"result" description:"成功结果JSON(ModelCallRes)"` +} + +type flowAsyncTaskCol struct { + beans.SQLBaseCol + ExecutionId string + NodeId string + SegmentIndex string + ModelId string + TaskId string + MsgTopic string + State string + Result string +} + +var FlowAsyncTaskCol = flowAsyncTaskCol{ + SQLBaseCol: beans.DefSQLBaseCol, + ExecutionId: "execution_id", + NodeId: "node_id", + SegmentIndex: "segment_index", + ModelId: "model_id", + TaskId: "task_id", + MsgTopic: "msg_topic", + State: "state", + Result: "result", +}