wlw/ai-agent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add AI workflow run management with listing and detail retrieval endpoints, including corresponding response structures and UI integration

Browse Source
This commit is contained in:
mlogclub
2026-06-23 23:05:17 +08:00
parent a670084374
commit 1440c3c4c3
10 changed files with 594 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/bootstrap/routes.go
+2
View File
@@ -231,6 +231,8 @@ func registerDashboardAIAgentRoutes(group *gin.RouterGroup) {
func registerDashboardAIWorkflowRoutes(group *gin.RouterGroup) {
group.GET("/node-spec/list", dashboard.AIWorkflowGetNodeSpecList)
group.POST("/validate", dashboard.AIWorkflowPostValidate)
group.Any("/run/list", dashboard.AIWorkflowAnyRunList)
group.GET("/run/:id", dashboard.AIWorkflowGetRunBy)
group.Any("/version/list", dashboard.AIWorkflowAnyVersionList)
group.GET("/version/:id", dashboard.AIWorkflowGetVersionBy)
}
internal/bootstrap/server_route_test.go
+2
View File
@@ -41,6 +41,8 @@ func TestNewServerRegistersGinRoutes(t *testing.T) {
http.MethodGet + " /api/dashboard/user/:id",
http.MethodPost + " /api/dashboard/user/create",
http.MethodPost + " /api/dashboard/conversation/send_message",
http.MethodGet + " /api/dashboard/ai-workflow/run/list",
http.MethodGet + " /api/dashboard/ai-workflow/run/:id",
http.MethodGet + " /api/ws/dashboard",
http.MethodGet + " /api/ws/open",
}
internal/builders/ai_workflow_builder.go
+93
View File
@@ -2,6 +2,7 @@ package builders
import (
"encoding/json"
"time"
"agent-desk/internal/ai/workflow/dsl"
workflowregistry "agent-desk/internal/ai/workflow/registry"
@@ -87,6 +88,71 @@ func BuildAIWorkflowNodeSpecs(list []workflowregistry.NodeSpec) []response.AIWor
return ret
}
func BuildAIWorkflowRun(item *models.AIWorkflowRun) response.AIWorkflowRunResponse {
if item == nil {
return response.AIWorkflowRunResponse{}
}
return response.AIWorkflowRunResponse{
ID: item.ID,
WorkflowID: item.WorkflowID,
WorkflowVersionID: item.WorkflowVersionID,
ConversationID: item.ConversationID,
AIAgentID: item.AIAgentID,
MessageID: item.MessageID,
Status: item.Status,
StatusName: workflowRunStatusName(item.Status),
StartedAt: formatWorkflowTime(item.StartedAt),
EndedAt: formatWorkflowTimePtr(item.EndedAt),
InterruptType: item.InterruptType,
InterruptNodeID: item.InterruptNodeID,
ErrorMessage: item.ErrorMessage,
CreatedAt: formatWorkflowTime(item.CreatedAt),
UpdatedAt: formatWorkflowTime(item.UpdatedAt),
}
}
func BuildAIWorkflowRunDetail(item *models.AIWorkflowRun, nodes []models.AIWorkflowNodeRun) response.AIWorkflowRunResponse {
ret := BuildAIWorkflowRun(item)
ret.Nodes = BuildAIWorkflowNodeRunList(nodes)
return ret
}
func BuildAIWorkflowRunList(list []models.AIWorkflowRun) []response.AIWorkflowRunResponse {
ret := make([]response.AIWorkflowRunResponse, 0, len(list))
for i := range list {
ret = append(ret, BuildAIWorkflowRun(&list[i]))
}
return ret
}
func BuildAIWorkflowNodeRun(item *models.AIWorkflowNodeRun) response.AIWorkflowNodeRunResponse {
if item == nil {
return response.AIWorkflowNodeRunResponse{}
}
return response.AIWorkflowNodeRunResponse{
ID: item.ID,
WorkflowRunID: item.WorkflowRunID,
NodeID: item.NodeID,
NodeType: item.NodeType,
Status: item.Status,
StatusName: workflowRunStatusName(item.Status),
InputPreview: item.InputPreview,
OutputPreview: item.OutputPreview,
ErrorMessage: item.ErrorMessage,
StartedAt: formatWorkflowTime(item.StartedAt),
EndedAt: formatWorkflowTimePtr(item.EndedAt),
DurationMS: item.DurationMS,
}
}
func BuildAIWorkflowNodeRunList(list []models.AIWorkflowNodeRun) []response.AIWorkflowNodeRunResponse {
ret := make([]response.AIWorkflowNodeRunResponse, 0, len(list))
for i := range list {
ret = append(ret, BuildAIWorkflowNodeRun(&list[i]))
}
return ret
}
func parseWorkflowDefinition(raw string) dsl.Definition {
var ret dsl.Definition
if raw == "" {
@@ -95,3 +161,30 @@ func parseWorkflowDefinition(raw string) dsl.Definition {
_ = json.Unmarshal([]byte(raw), &ret)
return ret
}
func workflowRunStatusName(status int) string {
switch status {
case 1:
return "completed"
case 2:
return "interrupted"
case 3:
return "failed"
default:
return "unknown"
}
}
func formatWorkflowTime(value time.Time) string {
if value.IsZero() {
return ""
}
return value.Format("2006-01-02 15:04:05")
}
func formatWorkflowTimePtr(value *time.Time) string {
if value == nil {
return ""
}
return formatWorkflowTime(*value)
}
internal/handlers/dashboard/ai_workflow_handler.go
+34
View File
@@ -227,3 +227,37 @@ func AIWorkflowGetVersionBy(ctx *gin.Context) {
}
httpx.WriteJSON(ctx, builders.BuildAIWorkflowVersion(item))
}
func AIWorkflowAnyRunList(ctx *gin.Context) {
if _, err := services.AuthService.RequirePermission(ctx, constants.PermissionAIAgentView); err != nil {
httpx.WriteJSON(ctx, err)
return
}
cnd := params.NewPagedSqlCnd(ctx,
params.QueryFilter{ParamName: "workflowId"},
params.QueryFilter{ParamName: "workflowVersionId"},
params.QueryFilter{ParamName: "conversationId"},
params.QueryFilter{ParamName: "aiAgentId"},
params.QueryFilter{ParamName: "messageId"},
params.QueryFilter{ParamName: "status"},
).Desc("id")
list, paging := services.AIWorkflowService.FindRunPageByCnd(cnd)
httpx.WriteJSON(ctx, &web.PageResult{Results: builders.BuildAIWorkflowRunList(list), Page: paging})
}
func AIWorkflowGetRunBy(ctx *gin.Context) {
id, ok := httpx.GetPathInt64(ctx, "id")
if !ok {
return
}
if _, err := services.AuthService.RequirePermission(ctx, constants.PermissionAIAgentView); err != nil {
httpx.WriteJSON(ctx, err)
return
}
item, nodes := services.AIWorkflowService.GetRunDetail(id)
if item == nil {
httpx.WriteJSON(ctx, httpx.JsonErrorMsg(ctx, "error.e0002"))
return
}
httpx.WriteJSON(ctx, builders.BuildAIWorkflowRunDetail(item, nodes))
}
internal/pkg/dto/response/ai_workflow_response.go
+34
View File
@@ -53,3 +53,37 @@ type AIWorkflowNodeSpecResponse struct {
OutputSchema []workflowregistry.VariableSpec `json:"outputSchema,omitempty"`
DefaultInputs map[string]dsl.VariableSelector `json:"defaultInputs,omitempty"`
}
type AIWorkflowRunResponse struct {
ID int64 `json:"id"`
WorkflowID int64 `json:"workflowId"`
WorkflowVersionID int64 `json:"workflowVersionId"`
ConversationID int64 `json:"conversationId"`
AIAgentID int64 `json:"aiAgentId"`
MessageID int64 `json:"messageId"`
Status int `json:"status"`
StatusName string `json:"statusName"`
StartedAt string `json:"startedAt"`
EndedAt string `json:"endedAt"`
InterruptType string `json:"interruptType"`
InterruptNodeID string `json:"interruptNodeId"`
ErrorMessage string `json:"errorMessage"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
Nodes []AIWorkflowNodeRunResponse `json:"nodes,omitempty"`
}
type AIWorkflowNodeRunResponse struct {
ID int64 `json:"id"`
WorkflowRunID int64 `json:"workflowRunId"`
NodeID string `json:"nodeId"`
NodeType string `json:"nodeType"`
Status int `json:"status"`
StatusName string `json:"statusName"`
InputPreview string `json:"inputPreview"`
OutputPreview string `json:"outputPreview"`
ErrorMessage string `json:"errorMessage"`
StartedAt string `json:"startedAt"`
EndedAt string `json:"endedAt"`
DurationMS int `json:"durationMs"`
}
internal/repositories/ai_workflow_run_repository.go
+16
View File
@@ -2,6 +2,7 @@ package repositories
import (
"agent-desk/internal/models"
"agent-desk/internal/pkg/httpx/params"
"github.com/mlogclub/simple/sqls"
"gorm.io/gorm"
@@ -28,6 +29,21 @@ func (r *aiWorkflowRunRepository) Find(db *gorm.DB, cnd *sqls.Cnd) (list []model
return
}
func (r *aiWorkflowRunRepository) FindPageByParams(db *gorm.DB, params *params.QueryParams) (list []models.AIWorkflowRun, paging *sqls.Paging) {
return r.FindPageByCnd(db, &params.Cnd)
}
func (r *aiWorkflowRunRepository) FindPageByCnd(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflowRun, paging *sqls.Paging) {
cnd.Find(db, &list)
count := cnd.Count(db, &models.AIWorkflowRun{})
paging = &sqls.Paging{
Page: cnd.Paging.Page,
Limit: cnd.Paging.Limit,
Total: count,
}
return
}
func (r *aiWorkflowRunRepository) Create(db *gorm.DB, t *models.AIWorkflowRun) error {
return db.Create(t).Error
}
internal/services/ai_workflow_service.go
+16
View File
@@ -57,6 +57,22 @@ func (s *aiWorkflowService) FindVersionPageByParams(params *params.QueryParams)
return repositories.AIWorkflowVersionRepository.FindPageByParams(sqls.DB(), params)
}
func (s *aiWorkflowService) FindRunPageByCnd(cnd *sqls.Cnd) (list []models.AIWorkflowRun, paging *sqls.Paging) {
return repositories.AIWorkflowRunRepository.FindPageByCnd(sqls.DB(), cnd)
}
func (s *aiWorkflowService) GetRunDetail(id int64) (*models.AIWorkflowRun, []models.AIWorkflowNodeRun) {
if id <= 0 {
return nil, nil
}
run := repositories.AIWorkflowRunRepository.Get(sqls.DB(), id)
if run == nil {
return nil, nil
}
nodes := repositories.AIWorkflowNodeRunRepository.Find(sqls.DB(), sqls.NewCnd().Eq("workflow_run_id", id).Asc("id"))
return run, nodes
}
func (s *aiWorkflowService) GetByAgentID(agentID int64) *models.AIWorkflow {
if agentID <= 0 {
return nil
internal/services/ai_workflow_service_test.go
+72
View File
@@ -3,6 +3,7 @@ package services
import (
"encoding/json"
"testing"
"time"
"agent-desk/internal/ai/workflow/dsl"
"agent-desk/internal/models"
@@ -151,6 +152,77 @@ func TestAIWorkflowServicePublishRejectsInvalidDSL(t *testing.T) {
}
}
func TestAIWorkflowServiceRunListAndDetail(t *testing.T) {
setupAIWorkflowTestDB(t)
now := time.Now()
run := models.AIWorkflowRun{
WorkflowID: 101,
WorkflowVersionID: 202,
ConversationID: 303,
AIAgentID: 12,
MessageID: 404,
Status: 1,
StartedAt: now,
EndedAt: &now,
}
if err := sqls.DB().Create(&run).Error; err != nil {
t.Fatalf("create workflow run: %v", err)
}
otherRun := models.AIWorkflowRun{
WorkflowID: 101,
WorkflowVersionID: 202,
ConversationID: 999,
AIAgentID: 12,
MessageID: 505,
Status: 1,
StartedAt: now,
}
if err := sqls.DB().Create(&otherRun).Error; err != nil {
t.Fatalf("create other workflow run: %v", err)
}
nodes := []models.AIWorkflowNodeRun{
{
WorkflowRunID: run.ID,
NodeID: "start_1",
NodeType: "start",
Status: 1,
InputPreview: `{"inputs":{}}`,
OutputPreview: `{"messageId":404}`,
StartedAt: now,
EndedAt: &now,
},
{
WorkflowRunID: run.ID,
NodeID: "reply_1",
NodeType: "llm_reply",
Status: 1,
OutputPreview: `{"replyText":"hello"}`,
StartedAt: now,
EndedAt: &now,
DurationMS: 8,
},
}
if err := sqls.DB().Create(&nodes).Error; err != nil {
t.Fatalf("create workflow node runs: %v", err)
}
list, paging := AIWorkflowService.FindRunPageByCnd(sqls.NewCnd().Eq("conversation_id", 303).Desc("id").Page(1, 20))
if paging.Total != 1 || len(list) != 1 || list[0].ID != run.ID {
t.Fatalf("unexpected run list: total=%d list=%#v", paging.Total, list)
}
detail, nodeRuns := AIWorkflowService.GetRunDetail(run.ID)
if detail == nil || detail.ID != run.ID {
t.Fatalf("unexpected detail run: %#v", detail)
}
if len(nodeRuns) != 2 || nodeRuns[0].NodeID != "start_1" || nodeRuns[1].NodeID != "reply_1" {
t.Fatalf("unexpected detail nodes: %#v", nodeRuns)
}
if missing, missingNodes := AIWorkflowService.GetRunDetail(999999); missing != nil || len(missingNodes) != 0 {
t.Fatalf("expected missing detail to be empty, got run=%#v nodes=%#v", missing, missingNodes)
}
}
func setupAIWorkflowTestDB(t *testing.T) {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{})