From 6e489733ec9fa5e00e0db26aa8e13d7c2a106abd Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Fri, 11 Sep 2026 16:46:49 +0800 Subject: [PATCH] feat(agent): unify async task FinalizeTaskRun projection to chat and runs Extend taskrun executors to return full TaskRunResult, finalize assistant placeholder content with artifact metadata, and map quality_failed terminal state instead of metadata-only updates. Co-authored-by: Cursor --- internal/bizdev/doc_id.go | 35 +++++ internal/bizdev/service.go | 12 +- internal/ctxmgr/message_update.go | 52 ++++++++ internal/server/handlers.go | 20 +-- internal/server/task_finalize.go | 185 ++++++++++++++++++++++++++ internal/server/task_finalize_test.go | 152 +++++++++++++++++++++ internal/server/task_run_handlers.go | 44 +++--- internal/squad/service.go | 10 +- internal/taskrun/manager.go | 81 ++++++++--- 9 files changed, 539 insertions(+), 52 deletions(-) create mode 100644 internal/bizdev/doc_id.go create mode 100644 internal/server/task_finalize.go create mode 100644 internal/server/task_finalize_test.go diff --git a/internal/bizdev/doc_id.go b/internal/bizdev/doc_id.go new file mode 100644 index 00000000..9ea6b0b7 --- /dev/null +++ b/internal/bizdev/doc_id.go @@ -0,0 +1,35 @@ +package bizdev + +import ( + "context" + "fmt" + "strings" +) + +// ResolveAndValidateDocID 解析权威 doc_id；若客户端传入则校验归属。 +func (s *Service) ResolveAndValidateDocID(ctx context.Context, req TaskRunRequest) (string, error) { + docID := strings.TrimSpace(req.DocID) + if docID == "" { + if req.ProjectID <= 0 || req.TaskKey == "" { + return "", nil + } + return fmt.Sprintf("doc_proj%d_%s", req.ProjectID, req.TaskKey), nil + } + + if s.docGov == nil { + return docID, nil + } + + doc, err := s.docGov.GetDocument(ctx, docID) + if err != nil { + return "", fmt.Errorf("doc_id not found: %s", docID) + } + if req.ProjectID > 0 && doc.ProjectID != req.ProjectID { + return "", fmt.Errorf("doc_id %s does not belong to project %d", docID, req.ProjectID) + } + if req.TaskKey != "" && doc.TaskKey != "" && doc.TaskKey != req.TaskKey { + return "", fmt.Errorf("doc_id %s task_key mismatch: expected %s got %s", docID, req.TaskKey, doc.TaskKey) + } + return doc.DocID, nil +} + diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index efc4da38..ffb843ce 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -279,9 +279,15 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun // 文档治理与认领拦截前置判定 artifactTitle := squad.TaskArtifactTitle(req.TaskKey) - docID := req.DocID - if docID == "" { - docID = fmt.Sprintf("doc_proj%d_%s", req.ProjectID, req.TaskKey) + docID, docErr := s.ResolveAndValidateDocID(ctx, req) + if docErr != nil { + return &TaskRunResult{ + TaskKey: req.TaskKey, + ArtifactTitle: artifactTitle, + WorkDir: req.WorkDir, + Success: false, + ErrorReason: docErr.Error(), + }, docErr } // 智能测试经理前置检查：必须依赖同项目已批准的系统需求规格书 (system_reqs) var upstreamSRDoc *docgov.DocumentRecord diff --git a/internal/ctxmgr/message_update.go b/internal/ctxmgr/message_update.go index 2adb4bd9..5c403653 100644 --- a/internal/ctxmgr/message_update.go +++ b/internal/ctxmgr/message_update.go @@ -50,3 +50,55 @@ func (m *ContextManager) UpdateMessageMetadata(ctx context.Context, messageID in } return errors.New("message not found") } + +// FinalizeAssistantTaskMessage 替换助手任务占位正文并写入完整 metadata（含 task_status）。 +func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messageID int64, content string, meta map[string]any, taskStatus string) error { + if messageID <= 0 { + return errors.New("invalid message_id") + } + if meta == nil { + meta = map[string]any{} + } + if taskStatus != "" { + meta["task_status"] = taskStatus + } + + if m.pool != nil { + metaJSON, err := json.Marshal(meta) + if err != nil { + return fmt.Errorf("marshal metadata failed: %w", err) + } + tag, err := m.pool.Exec(ctx, ` + UPDATE chat_messages + SET content = $1, + metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb + WHERE id = $3 + `, content, metaJSON, messageID) + if err != nil { + return fmt.Errorf("finalize assistant task message failed: %w", err) + } + if tag.RowsAffected() == 0 { + return errors.New("message not found") + } + return nil + } + + m.mu.Lock() + defer m.mu.Unlock() + for sid, msgs := range m.messages { + for i := range msgs { + if msgs[i].ID == messageID { + msgs[i].Content = content + if msgs[i].Metadata == nil { + msgs[i].Metadata = map[string]any{} + } + for k, v := range meta { + msgs[i].Metadata[k] = v + } + m.messages[sid][i] = msgs[i] + return nil + } + } + } + return errors.New("message not found") +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 0a2c35f3..57579d1e 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -200,16 +200,17 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { writeTaskError(w, err, stageRes) return } - if runRecord != nil && stageRes != nil { + if runRecord != nil && stageRes != nil && stageRes.PrimaryArtifact != nil { + status := resolveTerminalStatus(stageRes.PrimaryArtifact) outDocID := "" - outSummary := stageRes.StageTitle - if stageRes.PrimaryArtifact != nil { - if stageRes.PrimaryArtifact.DocRecord != nil { - outDocID = stageRes.PrimaryArtifact.DocRecord.DocID - } - outSummary = stageRes.PrimaryArtifact.ArtifactTitle + if stageRes.PrimaryArtifact.DocRecord != nil { + outDocID = stageRes.PrimaryArtifact.DocRecord.DocID } - _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "completed", outDocID, outSummary, "") + outSummary := stageRes.PrimaryArtifact.ArtifactTitle + if outSummary == "" { + outSummary = stageRes.StageTitle + } + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, status, outDocID, outSummary, "") } writeJSONResponse(w, http.StatusOK, stageRes) return @@ -251,11 +252,12 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) { } if runRecord != nil && taskRes != nil { + status := resolveTerminalStatus(taskRes) outDocID := "" if taskRes.DocRecord != nil { outDocID = taskRes.DocRecord.DocID } - _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, "completed", outDocID, taskRes.ArtifactTitle, "") + _ = s.pwSvc.UpdateWorkflowRun(r.Context(), runRecord.RunID, status, outDocID, taskRes.ArtifactTitle, "") } // 隐藏不必要的中间大体积字节 diff --git a/internal/server/task_finalize.go b/internal/server/task_finalize.go new file mode 100644 index 00000000..f0795d45 --- /dev/null +++ b/internal/server/task_finalize.go @@ -0,0 +1,185 @@ +package server + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "path/filepath" + "strings" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/squad" + "chat-agent-system/internal/taskrun" +) + +// FinalizeTaskRunProjection 将任务执行结果投影到 workflow_runs 与聊天助手消息（供回填/管理命令使用）。 +func (s *Server) FinalizeTaskRunProjection(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { + return s.finalizeTaskRun(ctx, record, result) +} + +// finalizeTaskRun 将任务执行结果投影到 workflow_runs 与聊天助手消息。 +func (s *Server) finalizeTaskRun(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { + if record == nil || result == nil { + return "failed", fmt.Errorf("finalize: missing record or result") + } + + workspaceID := result.WorkDir + if workspaceID == "" { + workspaceID = record.WorkspaceID + } + + terminalStatus := resolveTerminalStatus(result) + summaryContent := buildTaskCompletionSummary(result, terminalStatus) + + docID := "" + artifactTitle := result.ArtifactTitle + sha256Hex := "" + if result.DocRecord != nil { + docID = result.DocRecord.DocID + if result.DocRecord.ContentSHA256 != "" { + sha256Hex = result.DocRecord.ContentSHA256 + } + } + if docID == "" && record.ProjectID > 0 && record.TaskKey != "" { + docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey) + } + if artifactTitle == "" && record.TaskKey != "" { + artifactTitle = squad.TaskArtifactTitle(record.TaskKey) + } + + if sha256Hex == "" && workspaceID != "" { + if computed, err := hashOutputFile(workspaceID); err == nil { + sha256Hex = computed + } + } + + meta := map[string]any{ + "run_id": record.RunID, + "workspace_id": workspaceID, + "task_status": terminalStatus, + "task_key": record.TaskKey, + } + if record.ProjectID > 0 { + meta["project_id"] = record.ProjectID + } + if docID != "" { + artifact := map[string]any{ + "doc_id": docID, + "title": artifactTitle, + "task_key": record.TaskKey, + "workspace_id": workspaceID, + } + if sha256Hex != "" { + artifact["sha256"] = sha256Hex + } + if result.QualityReport != nil { + artifact["quality_passed"] = result.QualityReport.Passed + artifact["quality_score"] = result.QualityReport.Score + if len(result.QualityReport.Violations) > 0 { + artifact["quality_violations"] = result.QualityReport.Violations + } + } + meta["artifact"] = artifact + } + + if s.pwSvc != nil && record.ProjectID > 0 && record.RunID != "" { + if err := s.pwSvc.UpdateWorkflowRun(ctx, record.RunID, terminalStatus, docID, artifactTitle, ""); err != nil { + return "finalizing_failed", fmt.Errorf("update workflow run: %w", err) + } + } + + if s.svc != nil && s.svc.ContextManager() != nil && record.AssistantMsgID > 0 { + if err := s.svc.ContextManager().FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summaryContent, meta, terminalStatus); err != nil { + return "finalizing_failed", fmt.Errorf("finalize assistant message: %w", err) + } + } + + return terminalStatus, nil +} + +func resolveTerminalStatus(result *bizdev.TaskRunResult) string { + if result == nil || !result.Success { + return "failed" + } + if result.QualityReport != nil && !result.QualityReport.Passed { + return "quality_failed" + } + return "completed" +} + +func buildTaskCompletionSummary(result *bizdev.TaskRunResult, terminalStatus string) string { + title := result.ArtifactTitle + if title == "" && result.TaskKey != "" { + title = squad.TaskArtifactTitle(result.TaskKey) + } + if title == "" { + title = "交付文档" + } + + var b strings.Builder + fmt.Fprintf(&b, "《%s》已生成。\n\n", title) + + switch terminalStatus { + case "quality_failed": + b.WriteString("当前状态：**质量检查未通过，待整改**") + if result.QualityReport != nil { + fmt.Fprintf(&b, "\n质量评分：%.0f/100", result.QualityReport.Score) + if len(result.QualityReport.Violations) > 0 { + fmt.Fprintf(&b, "\n主要问题：%s", strings.Join(result.QualityReport.Violations, "；")) + } + } + case "completed": + b.WriteString("当前状态：**已完成**") + if result.QualityReport != nil && result.QualityReport.Passed { + fmt.Fprintf(&b, "\n质量评分：%.0f/100", result.QualityReport.Score) + } + default: + if result.ErrorReason != "" { + fmt.Fprintf(&b, "执行未成功：%s", result.ErrorReason) + } else { + b.WriteString("任务执行失败。") + } + } + + if result.DocRecord != nil && result.DocRecord.DocID != "" { + fmt.Fprintf(&b, "\n\n📄 受控文档：`%s`", result.DocRecord.DocID) + } + + return b.String() +} + +func hashOutputFile(workDir string) (string, error) { + path := filepath.Join(workDir, "output.md") + data, err := os.ReadFile(path) + if err != nil { + return "", err + } + if len(data) == 0 { + return "", fmt.Errorf("output.md is empty") + } + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]), nil +} + +func squadResultToTaskRun(res *squad.SquadRunResult, taskKey string) *bizdev.TaskRunResult { + if res == nil { + return nil + } + return &bizdev.TaskRunResult{ + TaskKey: res.TaskKey, + ArtifactTitle: res.ArtifactTitle, + MarkdownBody: res.MarkdownBody, + ExtractedData: res.ExtractedData, + Attachments: res.Attachments, + WorkDir: res.WorkDir, + BackendUsed: res.BackendUsed, + Duration: res.Duration, + Success: res.Success, + ErrorReason: res.ErrorReason, + QualityReport: res.QualityReport, + Manifest: res.Manifest, + AdvisorBrief: res.AdvisorBrief, + } +} diff --git a/internal/server/task_finalize_test.go b/internal/server/task_finalize_test.go new file mode 100644 index 00000000..4463cef5 --- /dev/null +++ b/internal/server/task_finalize_test.go @@ -0,0 +1,152 @@ +package server + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/squad" + "chat-agent-system/internal/stream" + "chat-agent-system/internal/taskrun" +) + +func TestFinalizeTaskRun_ReplacesPlaceholderContent(t *testing.T) { + cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"} + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + + workDir := t.TempDir() + _ = os.WriteFile(filepath.Join(workDir, "output.md"), []byte("# 最终交付\n\n正文内容。"), 0o644) + + ctxMgr := svc.ContextManager() + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-finalize", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{ + "task_status": "running", + }) + if err != nil { + t.Fatalf("RecordUserMessageWithID: %v", err) + } + + pwSvc := projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + _, _ = pwSvc.CreateWorkflowRun(context.Background(), &projectworkflow.WorkflowRun{ + RunID: "run_finalize_test", + ProjectID: 56, + TaskKey: "intake", + Status: "running", + }) + + srv.pwSvc = pwSvc + srv.taskRuns = taskrun.NewManager(pwSvc, ctxMgr) + srv.initTaskRunExecutors() + + record := &taskrun.RunRecord{ + RunID: "run_finalize_test", + SessionID: "sess-finalize", + ProjectID: 56, + TaskKey: "intake", + AssistantMsgID: asstID, + } + + status, finErr := srv.finalizeTaskRun(context.Background(), record, &bizdev.TaskRunResult{ + Success: true, + ArtifactTitle: "用户初始需求收集表", + WorkDir: workDir, + TaskKey: "intake", + QualityReport: &squad.QualityReport{Passed: false, Score: 40, Violations: []string{"不做展开"}}, + DocRecord: &docgov.DocumentRecord{DocID: "doc_proj56_intake", ContentSHA256: "abc"}, + }) + if finErr != nil { + t.Fatalf("finalizeTaskRun: %v", finErr) + } + if status != "quality_failed" { + t.Fatalf("status=%q want quality_failed", status) + } + + msg, err := ctxMgr.GetMessageByID(context.Background(), asstID) + if err != nil { + t.Fatalf("GetMessageByID: %v", err) + } + if msg.Content == "任务执行中…" { + t.Fatalf("content still placeholder: %q", msg.Content) + } + if msg.Metadata["task_status"] != "quality_failed" { + t.Fatalf("metadata task_status=%v", msg.Metadata["task_status"]) + } + artifact, ok := msg.Metadata["artifact"].(map[string]any) + if !ok || artifact["doc_id"] != "doc_proj56_intake" { + t.Fatalf("missing artifact doc_id in metadata: %#v", msg.Metadata["artifact"]) + } + + run, err := pwSvc.GetWorkflowRun(context.Background(), "run_finalize_test") + if err != nil { + t.Fatalf("GetWorkflowRun: %v", err) + } + if run.OutputDocID != "doc_proj56_intake" { + t.Fatalf("OutputDocID=%q", run.OutputDocID) + } + if run.OutputSummary == "" { + t.Fatal("OutputSummary empty") + } +} + +func TestAsyncTaskRun_FinalizesAfterWorkerCompletes(t *testing.T) { + cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"} + svc := bizdev.NewService(cfg) + pwSvc := projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + ctxMgr := svc.ContextManager() + + workDir := t.TempDir() + _ = os.WriteFile(filepath.Join(workDir, "output.md"), []byte("# async finalize\n"), 0o644) + + srv := &Server{cfg: cfg, svc: svc, pwSvc: pwSvc, taskRuns: taskrun.NewManager(pwSvc, ctxMgr)} + srv.taskRuns.SetExecutor( + func(_ context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { + return &bizdev.TaskRunResult{ + Success: true, + ArtifactTitle: "用户初始需求收集表", + WorkDir: workDir, + TaskKey: "intake", + QualityReport: &squad.QualityReport{Passed: true, Score: 95}, + DocRecord: &docgov.DocumentRecord{DocID: "doc_proj1_intake"}, + }, nil + }, + nil, + nil, + ) + srv.taskRuns.SetFinalizer(func(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { + return srv.finalizeTaskRun(ctx, record, result) + }) + + asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-async", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil) + runID := "run_async_finalize" + _, _ = pwSvc.CreateWorkflowRun(context.Background(), &projectworkflow.WorkflowRun{ + RunID: runID, ProjectID: 1, TaskKey: "intake", Status: "running", + }) + + record := &taskrun.RunRecord{ + RunID: runID, SessionID: "sess-async", ProjectID: 1, TaskKey: "intake", AssistantMsgID: asstID, + } + _, err := srv.taskRuns.CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake", ProjectID: 1}, bizdev.StageRunRequest{}, "") + if err != nil { + t.Fatalf("CreateAndStart: %v", err) + } + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + rec, getErr := srv.taskRuns.GetRun(runID) + if getErr == nil && rec.Status != "running" { + msg, _ := ctxMgr.GetMessageByID(context.Background(), asstID) + if msg != nil && msg.Content != "任务执行中…" { + return + } + t.Fatalf("run finished with status=%s but content=%q", rec.Status, msg.Content) + } + time.Sleep(50 * time.Millisecond) + } + t.Fatal("worker did not finish") +} diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index bc740269..02883d13 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -263,29 +263,37 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien func (s *Server) initTaskRunExecutors() { s.taskRuns.SetExecutor( - func(ctx context.Context, runID string, req bizdev.TaskRunRequest, _ *stream.Streamer) (string, error) { - res, err := s.svc.ExecuteTask(ctx, req) - ws := "" - if res != nil { - ws = res.WorkDir - } - return ws, err + func(ctx context.Context, _ string, req bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { + return s.svc.ExecuteTask(ctx, req) }, - func(ctx context.Context, _ string, workDir string, _ *stream.Streamer) (string, error) { + func(ctx context.Context, _ string, workDir string, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { res, err := s.svc.ResumeOrModifyWorkspace(ctx, squad.ModifyWorkflowRequest{WorkDir: workDir}) - ws := workDir - if res != nil && res.WorkDir != "" { - ws = res.WorkDir + if res == nil { + return nil, err } - return ws, err + out := squadResultToTaskRun(res, res.TaskKey) + if out != nil && out.WorkDir == "" { + out.WorkDir = workDir + } + return out, err }, - func(ctx context.Context, _ string, req bizdev.StageRunRequest, _ *stream.Streamer) (string, error) { - res, err := s.svc.ExecuteStage(ctx, req) - ws := "" - if res != nil { - ws = res.WorkDir + func(ctx context.Context, _ string, req bizdev.StageRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) { + stageRes, err := s.svc.ExecuteStage(ctx, req) + if stageRes == nil { + return nil, err } - return ws, err + if stageRes.PrimaryArtifact != nil { + return stageRes.PrimaryArtifact, err + } + return &bizdev.TaskRunResult{ + WorkDir: stageRes.WorkDir, + Success: stageRes.Success, + ErrorReason: stageRes.ErrorReason, + ArtifactTitle: stageRes.StageTitle, + }, err }, ) + s.taskRuns.SetFinalizer(func(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) { + return s.finalizeTaskRun(ctx, record, result) + }) } diff --git a/internal/squad/service.go b/internal/squad/service.go index 0cc66180..9437c2f3 100644 --- a/internal/squad/service.go +++ b/internal/squad/service.go @@ -279,7 +279,15 @@ func (s *Service) ExecuteTask(ctx context.Context, req SquadRunRequest) (*SquadR streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审查告警: %v", err), "leader", 0, 0, 90) } } else if streamer != nil { - streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审核通过（评分: %.1f / 100）", report.Score), "leader", 0, 0, 95) + if report.Passed { + streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁审核通过（评分: %.1f / 100）", report.Score), "leader", 0, 0, 95) + } else { + violations := strings.Join(report.Violations, "；") + if violations == "" { + violations = "未满足质量门禁要求" + } + streamer.EmitPhase("quality_gate", fmt.Sprintf("质量门禁未通过（评分: %.1f / 100）：%s", report.Score, violations), "leader", 0, 0, 95) + } } manifest, _ := s.gate.GenerateManifest(workDir, req.TaskKey, *report) diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 6f963223..68486958 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -13,13 +13,16 @@ import ( ) // Executor 执行具体任务逻辑（由 server 注入）。 -type Executor func(ctx context.Context, runID string, req bizdev.TaskRunRequest, streamer *stream.Streamer) (workspaceID string, err error) +type Executor func(ctx context.Context, runID string, req bizdev.TaskRunRequest, streamer *stream.Streamer) (*bizdev.TaskRunResult, error) // ResumeExecutor 续跑既有工作区。 -type ResumeExecutor func(ctx context.Context, runID string, workDir string, streamer *stream.Streamer) (workspaceID string, err error) +type ResumeExecutor func(ctx context.Context, runID string, workDir string, streamer *stream.Streamer) (*bizdev.TaskRunResult, error) // StageExecutor 执行成熟度阶段任务序列。 -type StageExecutor func(ctx context.Context, runID string, req bizdev.StageRunRequest, streamer *stream.Streamer) (workspaceID string, err error) +type StageExecutor func(ctx context.Context, runID string, req bizdev.StageRunRequest, streamer *stream.Streamer) (*bizdev.TaskRunResult, error) + +// Finalizer 任务成功后投影运行记录与聊天消息（由 server 注入）。 +type Finalizer func(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult) (terminalStatus string, err error) // Manager 后台任务运行观察者：SSE 断线不取消 worker。 type Manager struct { @@ -28,6 +31,7 @@ type Manager struct { exec Executor resume ResumeExecutor stage StageExecutor + finalize Finalizer pwSvc *projectworkflow.Service ctxMgr *ctxmgr.ContextManager } @@ -46,6 +50,10 @@ func (m *Manager) SetExecutor(exec Executor, resume ResumeExecutor, stage StageE m.stage = stage } +func (m *Manager) SetFinalizer(f Finalizer) { + m.finalize = f +} + func (m *Manager) CreateAndStart(parent context.Context, runID string, record *RunRecord, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) (*RunRecord, error) { if runID == "" { runID = fmt.Sprintf("run_%d", time.Now().UnixNano()) @@ -112,7 +120,7 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T } ctxWithS, streamer := stream.WithListener(ctx, false, listener) - var workspaceID string + var taskResult *bizdev.TaskRunResult var execErr error start := time.Now() @@ -122,15 +130,22 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T stageReq.ProjectID = ar.record.ProjectID stageReq.UserID = ar.record.UserID stageReq.ClientTurnID = ar.record.ClientTurnID - workspaceID, execErr = m.stage(ctxWithS, ar.record.RunID, stageReq, streamer) + taskResult, execErr = m.stage(ctxWithS, ar.record.RunID, stageReq, streamer) } else if resumeWorkDir != "" && m.resume != nil { - workspaceID, execErr = m.resume(ctxWithS, ar.record.RunID, resumeWorkDir, streamer) + taskResult, execErr = m.resume(ctxWithS, ar.record.RunID, resumeWorkDir, streamer) } else if m.exec != nil { - workspaceID, execErr = m.exec(ctxWithS, ar.record.RunID, taskReq, streamer) + taskResult, execErr = m.exec(ctxWithS, ar.record.RunID, taskReq, streamer) } else { execErr = fmt.Errorf("task executor not configured") } + workspaceID := "" + if taskResult != nil && taskResult.WorkDir != "" { + workspaceID = taskResult.WorkDir + } else if resumeWorkDir != "" { + workspaceID = resumeWorkDir + } + if workspaceID != "" { ar.mu.Lock() ar.record.WorkspaceID = workspaceID @@ -139,7 +154,7 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T if execErr != nil { if ctx.Err() == context.Canceled { - m.finishRun(ar, "cancelled", workspaceID, execErr.Error()) + m.finishRun(ar, "cancelled", workspaceID, execErr.Error(), nil) info := stream.ClassifyError(fmt.Errorf("task cancelled"), "task_cancel", ar.record.RunID, ar.record.SessionID, workspaceID) m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) return @@ -150,7 +165,7 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T } info := stream.ClassifyError(execErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, classifyWS) m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) - m.finishRun(ar, "failed", workspaceID, execErr.Error()) + m.finishRun(ar, "failed", workspaceID, execErr.Error(), taskResult) return } @@ -162,10 +177,22 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T DurationMs: time.Since(start).Milliseconds(), }, }) - m.finishRun(ar, "completed", workspaceID, "") + + terminalStatus := "completed" + if m.finalize != nil && taskResult != nil { + if ts, finErr := m.finalize(ctx, ar.record, taskResult); finErr != nil { + m.finishRun(ar, "finalizing_failed", workspaceID, finErr.Error(), taskResult) + return + } else if ts != "" { + terminalStatus = ts + } + } else if taskResult != nil && taskResult.QualityReport != nil && !taskResult.QualityReport.Passed { + terminalStatus = "quality_failed" + } + m.finishRun(ar, terminalStatus, workspaceID, "", taskResult) } -func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string) { +func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, result *bizdev.TaskRunResult) { ar.mu.Lock() ar.record.Status = status ar.record.WorkspaceID = workspaceID @@ -173,18 +200,30 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string) { ar.record.UpdatedAt = time.Now() ar.mu.Unlock() - if m.pwSvc != nil && ar.record.ProjectID > 0 { - _ = m.pwSvc.UpdateWorkflowRun(context.Background(), ar.record.RunID, status, "", "", errMsg) - } - - if m.ctxMgr != nil && ar.record.AssistantMsgID > 0 { - meta := map[string]any{ - "run_id": ar.record.RunID, - "workspace_id": workspaceID, - "task_status": status, + // 失败/取消路径：若 Finalize 未执行，仍更新 workflow_runs 与占位消息 metadata + if status == "failed" || status == "cancelled" || status == "finalizing_failed" { + if m.pwSvc != nil && ar.record.ProjectID > 0 { + _ = m.pwSvc.UpdateWorkflowRun(context.Background(), ar.record.RunID, status, "", "", errMsg) + } + if m.ctxMgr != nil && ar.record.AssistantMsgID > 0 { + meta := map[string]any{ + "run_id": ar.record.RunID, + "workspace_id": workspaceID, + "task_status": status, + } + content := "任务执行失败。" + if status == "cancelled" { + content = "任务已取消。" + } else if status == "finalizing_failed" { + content = "任务已生成，但结果落库失败，请联系管理员或稍后重试。" + } + if errMsg != "" { + content += "\n\n" + errMsg + } + _ = m.ctxMgr.FinalizeAssistantTaskMessage(context.Background(), ar.record.AssistantMsgID, content, meta, status) } - _ = m.ctxMgr.UpdateMessageMetadata(context.Background(), ar.record.AssistantMsgID, meta, status) } + _ = result // success path persistence handled by Finalizer } func (m *Manager) appendEvent(ar *activeRun, ev stream.StreamEvent) {