From c0d6ef7d8024bfb34f0bfe315362525dad88d87c Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Thu, 10 Sep 2026 23:41:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(workflow):=20Phase=203=20Coordinator=20?= =?UTF-8?q?=E4=B8=8A=E4=B8=8B=E6=96=87=E3=80=81=E5=8A=A8=E4=BD=9C=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E4=B8=8E=E5=B7=A5=E4=BD=9C=E6=B5=81=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=20API=20=E9=9B=86=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/coordinator/context.go | 48 +++++++++++---- internal/ctxmgr/manager.go | 13 +++- internal/ctxmgr/types.go | 7 ++- internal/server/handlers.go | 54 +++++++++++++++++ internal/server/project_handlers.go | 63 ++++++++++++++++++++ internal/server/project_workflow_api_test.go | 23 +++++++ internal/server/server.go | 2 + 7 files changed, 196 insertions(+), 14 deletions(-) diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index ebbb784b..ab0d8b4b 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "strconv" "strings" "time" @@ -25,15 +26,18 @@ type UserInfo struct { // ProjectInfo 项目与工作流当前基准 type ProjectInfo struct { - ID string `json:"id"` - Name string `json:"name"` - CustomerName string `json:"customer_name"` - DeviceType string `json:"device_type"` - IntendedUse string `json:"intended_use"` - TargetMarket string `json:"target_market"` - CurrentWorkflow string `json:"current_workflow"` - CurrentStage bizdev.MaturityStage `json:"current_stage"` - StageRules []string `json:"stage_rules,omitempty"` + ID string `json:"id"` + Name string `json:"name"` + CustomerName string `json:"customer_name"` + DeviceType string `json:"device_type"` + IntendedUse string `json:"intended_use"` + TargetMarket string `json:"target_market"` + CreatorUserID int64 `json:"creator_user_id,omitempty"` + ActiveWorkflowKey string `json:"active_workflow_key,omitempty"` + WorkflowSelectionVersion int `json:"workflow_selection_version,omitempty"` + CurrentWorkflow string `json:"current_workflow"` + CurrentStage bizdev.MaturityStage `json:"current_stage"` + StageRules []string `json:"stage_rules,omitempty"` } // UserDocsInfo 用户相关文档清单与认领状态 @@ -120,6 +124,21 @@ func ComputeAvailableActions(stage bizdev.MaturityStage, pRole string) []string return actions } +// ComputeAvailableActionsWithWorkflow 结合激活工作流、成熟度阶段与角色权限计算可用操作集合 +func ComputeAvailableActionsWithWorkflow(stage bizdev.MaturityStage, pRole string, activeWorkflow string, isCreator bool) []string { + actions := ComputeAvailableActions(stage, pRole) + if isCreator { + actions = append(actions, "SWITCH_WORKFLOW") + } + switch activeWorkflow { + case "product_manager": + actions = append(actions, "TRIGGER_TASK:system_reqs", "TRIGGER_WORKFLOW:product_manager") + case "test_manager": + actions = append(actions, "TRIGGER_TASK:test_manager", "TRIGGER_WORKFLOW:test_manager") + } + return actions +} + // ToSnapshot 转换为 L1 意图识别器所需的轻量上下文快照 func (ctx *UserProjectContext) ToSnapshot() *intent.ContextSnapshot { if ctx == nil { @@ -316,6 +335,9 @@ func (a *ContextAssembler) AssembleFromWorkspace(workDir string, user UserInfo) } // 3. 尝试提取阶段与文档 + if ctx.Project.ActiveWorkflowKey != "" { + ctx.Project.CurrentWorkflow = ctx.Project.ActiveWorkflowKey + } if strings.HasPrefix(ctx.Project.CurrentWorkflow, "V0.1") || ctx.Project.CurrentWorkflow == "intake" { ctx.Project.CurrentStage = bizdev.StageV01_Triage } else if strings.HasPrefix(ctx.Project.CurrentWorkflow, "V0.5") { @@ -327,7 +349,13 @@ func (a *ContextAssembler) AssembleFromWorkspace(workDir string, user UserInfo) if ctx.Project.CurrentStage == "" { ctx.Project.CurrentStage = bizdev.StageV01_Triage } - ctx.AvailableActions = ComputeAvailableActions(ctx.Project.CurrentStage, user.ProjectRole) + + isCreator := false + if user.ID != "" && ctx.Project.CreatorUserID > 0 { + uid, _ := strconv.ParseInt(user.ID, 10, 64) + isCreator = (uid == ctx.Project.CreatorUserID) + } + ctx.AvailableActions = ComputeAvailableActionsWithWorkflow(ctx.Project.CurrentStage, user.ProjectRole, ctx.Project.ActiveWorkflowKey, isCreator) // 4. 优先尝试直接读取持久化的 advisor_brief.json briefPath := filepath.Join(workDir, "advisor_brief.json") diff --git a/internal/ctxmgr/manager.go b/internal/ctxmgr/manager.go index d48cb62e..e5b258c1 100644 --- a/internal/ctxmgr/manager.go +++ b/internal/ctxmgr/manager.go @@ -416,11 +416,20 @@ func (m *ContextManager) GetProjectContext(ctx context.Context, projectID int64) if m.pool != nil { // 1. 获取项目基准与黑板事实 - var pCode, pName, cStage string - _ = m.pool.QueryRow(ctx, "SELECT code, name, current_stage FROM projects WHERE id = $1", projectID).Scan(&pCode, &pName, &cStage) + var pCode, pName, cStage, actWf string + var creatorID int64 + var wfVer int + _ = m.pool.QueryRow(ctx, ` + SELECT code, name, current_stage, COALESCE(creator_user_id, 1), + COALESCE(active_workflow_key, 'main'), COALESCE(workflow_selection_version, 1) + FROM projects WHERE id = $1 + `, projectID).Scan(&pCode, &pName, &cStage, &creatorID, &actWf, &wfVer) pCtx.ProjectCode = pCode pCtx.ProjectName = pName pCtx.CurrentStage = cStage + pCtx.CreatorUserID = creatorID + pCtx.ActiveWorkflowKey = actWf + pCtx.WorkflowSelectionVersion = wfVer var factsJSON []byte err := m.pool.QueryRow(ctx, "SELECT facts FROM project_blackboards WHERE project_id = $1", projectID).Scan(&factsJSON) diff --git a/internal/ctxmgr/types.go b/internal/ctxmgr/types.go index d8b6c5da..aba22528 100644 --- a/internal/ctxmgr/types.go +++ b/internal/ctxmgr/types.go @@ -81,8 +81,11 @@ type ProjectCollectiveContext struct { ProjectID int64 `json:"project_id"` ProjectCode string `json:"project_code"` ProjectName string `json:"project_name"` - CurrentStage string `json:"current_stage"` - BlackboardFacts map[string]string `json:"blackboard_facts"` + CurrentStage string `json:"current_stage"` + CreatorUserID int64 `json:"creator_user_id,omitempty"` + ActiveWorkflowKey string `json:"active_workflow_key,omitempty"` + WorkflowSelectionVersion int `json:"workflow_selection_version,omitempty"` + BlackboardFacts map[string]string `json:"blackboard_facts"` AllActivities []ActivityItem `json:"all_activities"` MilestoneDecisions []ActivityItem `json:"milestone_decisions"` SharedFiles []*storage.FileReference `json:"shared_files"` diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 79d0772d..4f5833d2 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -12,6 +12,7 @@ import ( "chat-agent-system/internal/bizdev" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/docgov" + "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/squad" ) @@ -63,6 +64,7 @@ type TaskRunRequestDTO struct { ApproverName string `json:"approver_name,omitempty"` ForceDirectEdit bool `json:"force_direct_edit,omitempty"` CustomSteps []squad.StepSpec `json:"custom_steps,omitempty"` + WorkflowKey string `json:"workflow_key,omitempty"` } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { @@ -137,6 +139,33 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { return } + // 0. 如果关联了项目，在 projectworkflow 服务中记录运行实例 + var runRecord *projectworkflow.WorkflowRun + if s.pwSvc != nil && req.ProjectID > 0 { + wfKey := req.WorkflowKey + if wfKey == "" { + if p, _ := s.pwSvc.GetProject(r.Context(), req.ProjectID, req.UserID); p != nil { + wfKey = p.ActiveWorkflowKey + } + } + if wfKey == "" { + wfKey = "main" + } + manifest := map[string]interface{}{ + "fields": req.Fields, + "doc_id": req.DocID, + } + runRecord, _ = s.pwSvc.CreateWorkflowRun(r.Context(), &projectworkflow.WorkflowRun{ + ProjectID: req.ProjectID, + WorkflowKey: wfKey, + Status: "running", + StartedBy: &req.UserID, + StageKey: req.Stage, + TaskKey: req.TaskKey, + InputManifest: manifest, + }) + } + // 1. 如果指定了 stage，则执行成熟度阶段编排 if req.Stage != "" { stageRes, err := s.svc.ExecuteStage(r.Context(), bizdev.StageRunRequest{ @@ -161,9 +190,23 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { ForceDirectEdit: req.ForceDirectEdit, }) if err != nil { + if runRecord != nil { + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "failed", "", "", err.Error()) + } writeTaskError(w, err, stageRes) return } + if runRecord != nil && stageRes != nil { + outDocID := "" + outSummary := stageRes.StageTitle + if stageRes.PrimaryArtifact != nil { + if stageRes.PrimaryArtifact.DocRecord != nil { + outDocID = stageRes.PrimaryArtifact.DocRecord.DocID + } + outSummary = stageRes.PrimaryArtifact.ArtifactTitle + } + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "completed", outDocID, outSummary, "") + } writeJSONResponse(w, http.StatusOK, stageRes) return } @@ -195,10 +238,21 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { CustomSteps: req.CustomSteps, }) if err != nil { + if runRecord != nil { + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "failed", "", "", err.Error()) + } writeTaskError(w, err, taskRes) return } + if runRecord != nil && taskRes != nil { + outDocID := "" + if taskRes.DocRecord != nil { + outDocID = taskRes.DocRecord.DocID + } + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "completed", outDocID, taskRes.ArtifactTitle, "") + } + // 隐藏不必要的中间大体积字节 if taskRes.WorkDir != "" { taskRes.WorkDir = filepath.Clean(taskRes.WorkDir) diff --git a/internal/server/project_handlers.go b/internal/server/project_handlers.go index 5517034d..2ea00234 100644 --- a/internal/server/project_handlers.go +++ b/internal/server/project_handlers.go @@ -1,8 +1,10 @@ package server import ( + "bytes" "encoding/json" "errors" + "io" "net/http" "strconv" @@ -165,3 +167,64 @@ func (s *Server) handleListWorkflowDefinitions(w http.ResponseWriter, r *http.Re defs := s.pwSvc.Catalog().List() writeJSONResponse(w, http.StatusOK, defs) } + +// handleRunProjectWorkflow 针对特定项目与工作流触发运行并落盘运行记录 +func (s *Server) handleRunProjectWorkflow(w http.ResponseWriter, r *http.Request) { + idStr := r.PathValue("id") + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`, http.StatusBadRequest) + return + } + wfKey := r.PathValue("workflow_key") + if wfKey == "" { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"workflow_key is required"}`, http.StatusBadRequest) + return + } + def, ok := s.pwSvc.Catalog().Get(wfKey) + if !ok { + http.Error(w, `{"error":"ERR_WORKFLOW_NOT_FOUND","message":"unknown workflow"}`, http.StatusNotFound) + return + } + + user, _ := s.authenticateUser(r) + var uid int64 = 1 + if user != nil { + uid = user.ID + } + + var req TaskRunRequestDTO + if r.Body != nil && r.ContentLength > 0 { + _ = json.NewDecoder(r.Body).Decode(&req) + } + req.ProjectID = id + req.WorkflowKey = wfKey + req.UserID = uid + if req.TaskKey == "" { + req.TaskKey = def.TaskKey + } + + bodyBytes, _ := json.Marshal(req) + r.Body = io.NopCloser(bytes.NewReader(bodyBytes)) + s.handleRunTask(w, r) +} + +// handleListProjectWorkflowRuns 查询项目某工作流的历史运行记录 +func (s *Server) handleListProjectWorkflowRuns(w http.ResponseWriter, r *http.Request) { + idStr := r.PathValue("id") + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`, http.StatusBadRequest) + return + } + wfKey := r.PathValue("workflow_key") + runs, err := s.pwSvc.ListWorkflowRuns(r.Context(), id, wfKey, 20) + if err != nil { + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + if runs == nil { + runs = make([]*projectworkflow.WorkflowRun, 0) + } + writeJSONResponse(w, http.StatusOK, runs) +} diff --git a/internal/server/project_workflow_api_test.go b/internal/server/project_workflow_api_test.go index 2a20e886..601daca2 100644 --- a/internal/server/project_workflow_api_test.go +++ b/internal/server/project_workflow_api_test.go @@ -132,4 +132,27 @@ func TestProjectWorkflowAPI_Lifecycle(t *testing.T) { err = json.Unmarshal(recDefs.Body.Bytes(), &defs) require.NoError(t, err) assert.Equal(t, 3, len(defs)) + + // 9. 触发工作流运行 -> 200 OK + bodyRun, _ := json.Marshal(map[string]any{ + "task_key": "intake", + "project_name": "智能高频电刀", + "device_type": "高频手术设备", + }) + reqRun := httptest.NewRequest(http.MethodPost, "/api/v1/projects/1001/workflows/main/run", bytes.NewReader(bodyRun)) + reqRun.Header.Set("X-User-ID", "2") + recRun := httptest.NewRecorder() + handler.ServeHTTP(recRun, reqRun) + assert.Equal(t, http.StatusOK, recRun.Code) + + // 10. 查询运行历史 -> 200 OK 且包含刚运行的记录 + reqRuns := httptest.NewRequest(http.MethodGet, "/api/v1/projects/1001/workflows/main/runs", nil) + recRuns := httptest.NewRecorder() + handler.ServeHTTP(recRuns, reqRuns) + assert.Equal(t, http.StatusOK, recRuns.Code) + var runsList []*projectworkflow.WorkflowRun + err = json.Unmarshal(recRuns.Body.Bytes(), &runsList) + require.NoError(t, err) + assert.NotEmpty(t, runsList) + assert.Equal(t, "completed", runsList[0].Status) } diff --git a/internal/server/server.go b/internal/server/server.go index c7af7685..e8d439a8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -85,6 +85,8 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("GET /api/v1/projects/{id}/workflows", s.handleListProjectWorkflows) mux.HandleFunc("PUT /api/v1/projects/{id}/active-workflow", s.handleActivateWorkflow) mux.HandleFunc("GET /api/v1/workflows", s.handleListWorkflowDefinitions) + mux.HandleFunc("POST /api/v1/projects/{id}/workflows/{workflow_key}/run", s.handleRunProjectWorkflow) + mux.HandleFunc("GET /api/v1/projects/{id}/workflows/{workflow_key}/runs", s.handleListProjectWorkflowRuns) // 3. 对话协同服务 (标准 JSON 阻塞) mux.HandleFunc("POST /api/v1/chat", s.handleChat)