red-future/ai-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workflow): flow_async_task 统一异步任务表实体与 DAO（物理删除）

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
19904408334
2026-08-25 15:55:03 +08:00
co-authored by Claude Opus 4.7
parent d5b2d90a27
commit 5672c8272e
4 changed files with 236 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workflow/consts/public/table_name.go
+1
View File
@@ -23,4 +23,5 @@ const (
TableNameExecWorkflow = "exec_workflow"
TableNameExecWorkflowResult = "exec_workflow_result"
TableNameFlowSegmentResult = "flow_segment_result"
TableNameFlowAsyncTask = "flow_async_task"
)
workflow/dao/flow/flow_async_task_dao.go
+100
View File
@@ -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
}
workflow/dao/flow/flow_async_task_dao_test.go
+94
View File
@@ -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)
}
workflow/model/entity/flow_async_task.go
+41
View File
@@ -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",
}