diff --git a/internal/taskrun/finalize.go b/internal/taskrun/finalize.go index 7403299e..b1373353 100644 --- a/internal/taskrun/finalize.go +++ b/internal/taskrun/finalize.go @@ -90,6 +90,11 @@ type VerifiedReceipt struct { } func (c *Commands) Finalize(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult) (string, error) { + return c.FinalizeWithEvents(ctx, record, result, nil, 0) +} + +// FinalizeWithEvents 与 Finalize 相同，额外把 phase 事件折叠进消息 metadata.task_phase_snapshot。 +func (c *Commands) FinalizeWithEvents(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, events []StoredEvent, durationMs int64) (string, error) { if record == nil || result == nil { return StatusFailed, fmt.Errorf("finalize: missing record or result") } @@ -97,7 +102,7 @@ func (c *Commands) Finalize(ctx context.Context, record *RunRecord, result *bizd if _, err := ResolveBackend(result.BackendUsed, ""); err != nil && strings.TrimSpace(result.BackendUsed) != "" { return StatusFailed, err } - out, err := c.apply(ctx, record, result, status) + out, err := c.apply(ctx, record, result, status, events, durationMs) if err != nil { if errors.Is(err, ErrTerminalConflict) && out != nil && out.Status != "" { return out.Status, nil @@ -112,7 +117,7 @@ func (c *Commands) Cancel(ctx context.Context, record *RunRecord) (string, error return StatusCancelled, fmt.Errorf("cancel: missing record") } result := &bizdev.TaskRunResult{Success: false, ErrorReason: "cancelled", TaskKey: record.TaskKey, WorkDir: record.WorkspaceID} - out, err := c.apply(ctx, record, result, StatusCancelled) + out, err := c.apply(ctx, record, result, StatusCancelled, nil, 0) if err != nil { if errors.Is(err, ErrTerminalConflict) && out != nil { return out.Status, nil @@ -122,21 +127,21 @@ func (c *Commands) Cancel(ctx context.Context, record *RunRecord) (string, error return out.Status, nil } -func (c *Commands) apply(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) { +func (c *Commands) apply(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string, events []StoredEvent, durationMs int64) (*FinalizeResult, error) { if c != nil && c.UoW != nil { var out *FinalizeResult err := c.UoW.Within(ctx, func(scope uow.TxScope) error { ctx = uow.WithScope(ctx, scope) - got, err := c.applyLocked(ctx, record, result, want) + got, err := c.applyLocked(ctx, record, result, want, events, durationMs) out = got return err }) return out, err } - return c.applyLocked(ctx, record, result, want) + return c.applyLocked(ctx, record, result, want, events, durationMs) } -func (c *Commands) applyLocked(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) { +func (c *Commands) applyLocked(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string, events []StoredEvent, durationMs int64) (*FinalizeResult, error) { if err := rejectSensitiveText(result.MarkdownBody, result.ErrorReason); err != nil { return nil, err } @@ -163,7 +168,7 @@ func (c *Commands) applyLocked(ctx context.Context, record *RunRecord, result *b if err := c.projectWorkflow(ctx, record, result, want); err != nil { return nil, err } - if err := c.finalizeAssistant(ctx, record, result, want, workspaceID, manifest); err != nil { + if err := c.finalizeAssistant(ctx, record, result, want, workspaceID, manifest, events, durationMs); err != nil { return nil, err } if err := c.appendTerminalEvent(ctx, record, result, want, manifest); err != nil { @@ -344,7 +349,7 @@ func (c *Commands) projectWorkflow(ctx context.Context, record *RunRecord, resul return nil } -func (c *Commands) finalizeAssistant(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status, workspaceID string, man *RunManifest) error { +func (c *Commands) finalizeAssistant(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status, workspaceID string, man *RunManifest, events []StoredEvent, durationMs int64) error { if c == nil || c.Ctx == nil || record.AssistantMsgID <= 0 { return nil } @@ -405,6 +410,13 @@ func (c *Commands) finalizeAssistant(ctx context.Context, record *RunRecord, res } meta["artifact"] = artifact } + title := strings.TrimSpace(result.ArtifactTitle) + if title == "" && result.DocRecord != nil { + title = strings.TrimSpace(result.DocRecord.Title) + } + if snap := BuildPhaseSnapshot(events, durationMs, status, title); snap != nil { + meta["task_phase_snapshot"] = snap + } return c.Ctx.FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summary, meta, status) } diff --git a/internal/taskrun/finalize_test.go b/internal/taskrun/finalize_test.go index 210495d6..272caeb7 100644 --- a/internal/taskrun/finalize_test.go +++ b/internal/taskrun/finalize_test.go @@ -9,6 +9,7 @@ import ( "chat-agent-system/internal/bizdev" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/docgov" + "chat-agent-system/internal/stream" "chat-agent-system/internal/taskrun" ) @@ -42,12 +43,51 @@ func TestCommands_DuplicateFinalizeMemory(t *testing.T) { if st1 != "completed" || st2 != "completed" { t.Fatalf("status %s %s", st1, st2) } - msg2, _ := ctxMgr.GetMessageByID(context.Background(), asstID) - if msg2.ID != msg.ID || msg2.Content != msg.Content { + msgAgain, _ := ctxMgr.GetMessageByID(context.Background(), asstID) + if msgAgain.ID != msg.ID || msgAgain.Content != msg.Content { t.Fatal("duplicate finalize mutated the assistant turn") } } +func TestCommands_FinalizeWithEventsPersistsPhaseSnapshot(t *testing.T) { + ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) + asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-snap", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"task_status": "running"}) + if err != nil { + t.Fatal(err) + } + cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"} + rec := &taskrun.RunRecord{RunID: "run_snap_mem", SessionID: "sess-snap", TaskKey: "intake", AssistantMsgID: asstID} + res := &bizdev.TaskRunResult{ + Success: true, TaskKey: "intake", ArtifactTitle: "用户初始需求收集表", MarkdownBody: "# body\n", + DocRecord: &docgov.DocumentRecord{DocID: "doc_snap", ContentSHA256: strings.Repeat("b", 64), Version: 1}, + } + events := []taskrun.StoredEvent{ + {Seq: 1, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "squad_assembly", Message: "组建小队"}}}, + {Seq: 2, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "quality_gate", Message: "门禁", Progress: 95}}}, + } + if _, err := cmds.FinalizeWithEvents(context.Background(), rec, res, events, 3500); err != nil { + t.Fatal(err) + } + msg, err := ctxMgr.GetMessageByID(context.Background(), asstID) + if err != nil { + t.Fatal(err) + } + raw, ok := msg.Metadata["task_phase_snapshot"] + if !ok || raw == nil { + t.Fatal("expected task_phase_snapshot in metadata") + } + snap, ok := raw.(*taskrun.PhaseSnapshot) + if !ok { + t.Fatalf("unexpected snapshot type %T", raw) + } + if snap.Title != "用户初始需求收集表" || snap.DurationMs != 3500 || len(snap.Steps) == 0 { + t.Fatalf("snapshot %+v", snap) + } + if snap.Steps[len(snap.Steps)-1].Status != "completed" { + t.Fatalf("terminal step should be completed: %+v", snap.Steps) + } +} + func TestCommands_CancelWinsOverFinalizeMemory(t *testing.T) { ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{}) asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-race", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil) diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 6e0288cc..58fa7392 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -380,6 +380,8 @@ 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}) + durationMs := time.Since(start).Milliseconds() + evCopy := m.copyEvents(ar) if m.cmds != nil { failRes := taskResult if failRes == nil { @@ -390,7 +392,7 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T failRes.ErrorReason = execErr.Error() } } - _, _ = m.cmds.Finalize(context.Background(), ar.record, failRes) + _, _ = m.cmds.FinalizeWithEvents(context.Background(), ar.record, failRes, evCopy, durationMs) } m.finishRun(ar, "failed", workspaceID, execErr.Error(), taskResult, &info) return @@ -398,7 +400,18 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T // RUN-002: 先 finalize 落库，成功后再发 done 终态事件 terminalStatus := "completed" - if m.finalize != nil && taskResult != nil { + durationMs := time.Since(start).Milliseconds() + evCopy := m.copyEvents(ar) + if m.cmds != nil && taskResult != nil { + if ts, finErr := m.cmds.FinalizeWithEvents(ctx, ar.record, taskResult, evCopy, durationMs); finErr != nil { + info := stream.ClassifyError(finErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, workspaceID) + m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) + m.finishRun(ar, "finalizing_failed", workspaceID, finErr.Error(), taskResult, &info) + return + } else if ts != "" { + terminalStatus = ts + } + } else if m.finalize != nil && taskResult != nil { if ts, finErr := m.finalize(ctx, ar.record, taskResult); finErr != nil { info := stream.ClassifyError(finErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, workspaceID) m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info}) @@ -416,12 +429,26 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T Done: &stream.DonePayload{ SessionID: "task_run", WorkspaceID: workspaceID, - DurationMs: time.Since(start).Milliseconds(), + DurationMs: durationMs, }, }) m.finishRun(ar, terminalStatus, workspaceID, "", taskResult, nil) } +func (m *Manager) copyEvents(ar *activeRun) []StoredEvent { + if ar == nil { + return nil + } + ar.mu.RLock() + defer ar.mu.RUnlock() + if len(ar.events) == 0 { + return nil + } + out := make([]StoredEvent, len(ar.events)) + copy(out, ar.events) + return out +} + func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, result *bizdev.TaskRunResult, errorInfo *stream.ErrorInfo) { ar.mu.Lock() ar.record.Status = status @@ -459,6 +486,13 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, r if errorInfo != nil { meta["error_info"] = errorInfo } + title := "" + if result != nil { + title = result.ArtifactTitle + } + if snap := BuildPhaseSnapshot(m.copyEvents(ar), 0, status, title); snap != nil { + meta["task_phase_snapshot"] = snap + } content := "任务执行失败。" if status == "cancelled" { content = "任务已取消。" diff --git a/internal/taskrun/phase_snapshot.go b/internal/taskrun/phase_snapshot.go new file mode 100644 index 00000000..ba18d679 --- /dev/null +++ b/internal/taskrun/phase_snapshot.go @@ -0,0 +1,207 @@ +package taskrun + +import ( + "fmt" + "strings" + + "chat-agent-system/internal/stream" +) + +// PhaseStepSnapshot 折叠后的单步进度（与前端 PhaseEvent 字段对齐）。 +type PhaseStepSnapshot struct { + Step string `json:"step"` + Icon string `json:"icon"` + Label string `json:"label"` + Status string `json:"status"` // pending | running | completed | failed + Detail string `json:"detail,omitempty"` +} + +// PhaseSnapshot 写入 chat_messages.metadata.task_phase_snapshot，供刷新后零请求重建卡片。 +type PhaseSnapshot struct { + Steps []PhaseStepSnapshot `json:"steps"` + Label string `json:"label,omitempty"` + DurationMs int64 `json:"duration_ms,omitempty"` + Status string `json:"status,omitempty"` + Title string `json:"title,omitempty"` +} + +type phaseMeta struct { + Icon string + Label string +} + +var phaseMetaTable = map[string]phaseMeta{ + "squad_assembly": {Icon: "👥", Label: "组建小队"}, + "knowledge_retrieval": {Icon: "📚", Label: "知识检索"}, + "knowledge_retrieved": {Icon: "📚", Label: "知识命中"}, + "step_executing": {Icon: "⚙️", Label: "专员执行"}, + "step_completed": {Icon: "✅", Label: "步骤完成"}, + "quality_gate": {Icon: "🛡️", Label: "质量门禁"}, + "advisor_briefing": {Icon: "💡", Label: "参谋简报"}, + "stage_transition": {Icon: "🚀", Label: "阶段跃迁"}, + "stage_started": {Icon: "🚀", Label: "阶段启动"}, + "stage_completed": {Icon: "🎉", Label: "阶段完成"}, + "step_failed": {Icon: "⚠️", Label: "步骤异常"}, + "task_failed": {Icon: "⚠️", Label: "任务异常"}, +} + +var closePhases = map[string]struct{}{ + "step_completed": {}, + "stage_completed": {}, +} + +// BuildPhaseSnapshot 把 run 的 StoredEvents 折叠为可持久化快照（与前端 applyPhaseEvent 语义对齐）。 +func BuildPhaseSnapshot(events []StoredEvent, durationMs int64, status, title string) *PhaseSnapshot { + steps, label := FoldPhaseEvents(events) + if len(steps) == 0 && strings.TrimSpace(label) == "" && durationMs <= 0 && strings.TrimSpace(title) == "" { + if strings.TrimSpace(status) == "" { + return nil + } + } + // 终态时把仍 running 的步骤收口为 completed / failed，避免刷新后卡片卡在旋转态。 + if isTerminalTaskStatus(status) { + fail := status == StatusFailed || status == StatusFinalizingFailed || status == StatusQualityFailed || status == StatusInterrupted + for i := range steps { + if steps[i].Status == "running" { + if fail { + steps[i].Status = "failed" + } else { + steps[i].Status = "completed" + } + } + } + } + return &PhaseSnapshot{ + Steps: steps, + Label: label, + DurationMs: durationMs, + Status: status, + Title: strings.TrimSpace(title), + } +} + +func isTerminalTaskStatus(status string) bool { + _, ok := terminalStatuses[status] + return ok +} + +// FoldPhaseEvents 纯折叠：phase 事件 → 步骤序列 + 最近阶段文案。 +func FoldPhaseEvents(events []StoredEvent) ([]PhaseStepSnapshot, string) { + strip := make([]PhaseStepSnapshot, 0, 8) + label := "" + for _, se := range events { + ev := se.Event + if ev.Type != stream.EventPhase || ev.Phase == nil || ev.Phase.PhaseID == "" { + continue + } + phase := ev.Phase + if msg := strings.TrimSpace(phase.Message); msg != "" { + label = msg + } + strip = applyPhasePayload(strip, phase) + } + return strip, label +} + +func applyPhasePayload(strip []PhaseStepSnapshot, phase *stream.PhasePayload) []PhaseStepSnapshot { + meta, ok := phaseMetaTable[phase.PhaseID] + if !ok { + meta = phaseMeta{Icon: "⏳", Label: "任务推进"} + } + detail := strings.TrimSpace(phase.Message) + base := detail + if base == "" { + base = meta.Label + } + label := decoratePhaseLabel(base, phase) + key := phaseStepKey(phase) + + if idx := indexOfStep(strip, key); idx >= 0 { + cur := strip[idx] + cur.Label = label + if detail != "" { + cur.Detail = detail + } + if cur.Status == "failed" && phase.PhaseID != "step_failed" { + cur.Status = "running" + } + strip[idx] = cur + return strip + } + + if phase.PhaseID == "step_failed" { + if hasRunning(strip) { + out := make([]PhaseStepSnapshot, len(strip)) + copy(out, strip) + for i := range out { + if out[i].Status == "running" { + out[i].Status = "failed" + if detail != "" { + out[i].Detail = detail + } + } + } + return out + } + return append(strip, PhaseStepSnapshot{ + Step: key, Icon: meta.Icon, Label: label, Status: "failed", Detail: detail, + }) + } + + if _, close := closePhases[phase.PhaseID]; close { + out := make([]PhaseStepSnapshot, len(strip)) + copy(out, strip) + for i := range out { + if out[i].Status == "running" { + out[i].Status = "completed" + } + } + return out + } + + settled := make([]PhaseStepSnapshot, len(strip)) + copy(settled, strip) + for i := range settled { + if settled[i].Status == "running" { + settled[i].Status = "completed" + } + } + return append(settled, PhaseStepSnapshot{ + Step: key, Icon: meta.Icon, Label: label, Status: "running", Detail: detail, + }) +} + +func phaseStepKey(phase *stream.PhasePayload) string { + if phase.StepIndex > 0 { + return fmt.Sprintf("%s#%d", phase.PhaseID, phase.StepIndex) + } + return phase.PhaseID +} + +func decoratePhaseLabel(base string, phase *stream.PhasePayload) string { + if phase.StepIndex > 0 && phase.TotalSteps > 0 { + return fmt.Sprintf("%s（第 %d/%d 步）", base, phase.StepIndex, phase.TotalSteps) + } + if phase.Progress > 0 { + return fmt.Sprintf("%s（%d%%）", base, phase.Progress) + } + return base +} + +func indexOfStep(strip []PhaseStepSnapshot, key string) int { + for i := range strip { + if strip[i].Step == key { + return i + } + } + return -1 +} + +func hasRunning(strip []PhaseStepSnapshot) bool { + for i := range strip { + if strip[i].Status == "running" { + return true + } + } + return false +} diff --git a/internal/taskrun/phase_snapshot_test.go b/internal/taskrun/phase_snapshot_test.go new file mode 100644 index 00000000..ce10986e --- /dev/null +++ b/internal/taskrun/phase_snapshot_test.go @@ -0,0 +1,56 @@ +package taskrun_test + +import ( + "testing" + + "chat-agent-system/internal/stream" + "chat-agent-system/internal/taskrun" +) + +func TestFoldPhaseEvents_MirrorsFrontendSemantics(t *testing.T) { + events := []taskrun.StoredEvent{ + {Seq: 1, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "squad_assembly", Message: "组建小队"}}}, + {Seq: 2, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "step_executing", Message: "撰写中", StepIndex: 1, TotalSteps: 2}}}, + {Seq: 3, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "step_completed", Message: "完成"}}}, + {Seq: 4, Event: stream.StreamEvent{Type: stream.EventStatus, Status: &stream.StatusPayload{Message: "ignore"}}}, + } + steps, label := taskrun.FoldPhaseEvents(events) + if len(steps) != 2 { + t.Fatalf("steps=%d want 2: %+v", len(steps), steps) + } + if steps[0].Status != "completed" || steps[1].Status != "completed" { + t.Fatalf("expected both completed, got %+v", steps) + } + if label != "完成" { + t.Fatalf("label=%q", label) + } + if steps[1].Detail != "撰写中" { + t.Fatalf("detail=%q", steps[1].Detail) + } +} + +func TestBuildPhaseSnapshot_SettlesRunningOnTerminal(t *testing.T) { + events := []taskrun.StoredEvent{ + {Seq: 1, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "quality_gate", Message: "门禁中", Progress: 90}}}, + } + snap := taskrun.BuildPhaseSnapshot(events, 12_000, taskrun.StatusCompleted, "用户初始需求收集表") + if snap == nil { + t.Fatal("nil snapshot") + } + if snap.DurationMs != 12_000 || snap.Title != "用户初始需求收集表" { + t.Fatalf("meta %+v", snap) + } + if len(snap.Steps) != 1 || snap.Steps[0].Status != "completed" { + t.Fatalf("steps %+v", snap.Steps) + } +} + +func TestBuildPhaseSnapshot_FailedMarksRunningFailed(t *testing.T) { + events := []taskrun.StoredEvent{ + {Seq: 1, Event: stream.StreamEvent{Type: stream.EventPhase, Phase: &stream.PhasePayload{PhaseID: "step_executing", Message: "执行中"}}}, + } + snap := taskrun.BuildPhaseSnapshot(events, 0, taskrun.StatusFailed, "") + if snap.Steps[0].Status != "failed" { + t.Fatalf("status=%s", snap.Steps[0].Status) + } +}