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 <cursoragent@cursor.com>

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
+}
+

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

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")
+}

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, "")
 	}
 
 	// 隐藏不必要的中间大体积字节

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,
+	}
+}

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")
+}

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)
+	})
 }

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)

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) {