From b820e0070ddf67e2e9ee490ac1210e38c955e31e Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 28 Sep 2026 06:09:42 +0000 Subject: [PATCH] =?UTF-8?q?fix(taskrun):=20=E5=88=B7=E6=96=B0=E5=90=8E?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E8=BF=9B=E8=A1=8C=E4=B8=AD=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E5=B9=B6=E7=BA=A0=E6=AD=A3=E8=BF=9B=E5=BA=A6=E9=97=AE=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 进行中的 run 提前写下消息锚点，并按会话列出活跃任务；进度疑问句改为查询真实状态，交付物已落盘且不是传输失败时不再被空结果否决。 --- internal/agent/deliverable_test.go | 25 +++ internal/app/app.go | 1 + internal/app/chat_adapters.go | 27 +++ internal/chat/ports.go | 1 + internal/chat/session.go | 16 ++ internal/coordinator/context.go | 2 + internal/coordinator/engine.go | 22 ++ internal/coordinator/engine_query.go | 50 +++++ .../coordinator/session_run_reply_test.go | 45 ++++ internal/intent/recognizer_matchers_task.go | 29 +++ internal/intent/recognizer_test.go | 24 +++ internal/server/server.go | 1 + internal/server/sse_wave0_test.go | 6 + internal/server/task_run_handlers.go | 42 ++++ internal/squad/gov.go | 31 +-- internal/squad/workflow.go | 192 +++++++++++++----- internal/squad/workflow_outcome_test.go | 31 +++ internal/taskrun/active_session_test.go | 36 ++++ internal/taskrun/claim_idempotency_test.go | 31 +++ internal/taskrun/manager.go | 119 ++++++++++- internal/taskrun/store.go | 111 +++++++++- web/src/App.tsx | 9 + web/src/api/chatWindow.ts | 15 +- web/src/api/taskRuns.ts | 20 ++ web/src/engine/templateMatch.test.ts | 6 + web/src/hooks/useActiveTaskRecovery.ts | 114 +++++++++++ web/src/hooks/useChatWindow.test.ts | 15 ++ web/src/hooks/useRecoveredTaskPhases.test.ts | 18 ++ web/src/hooks/useRecoveredTaskPhases.ts | 41 +++- web/src/hooks/useTaskRunAttempts.ts | 36 +++- web/src/lib/activeTaskCache.test.ts | 44 ++++ web/src/lib/activeTaskCache.ts | 95 +++++++++ 32 files changed, 1173 insertions(+), 82 deletions(-) create mode 100644 internal/coordinator/session_run_reply_test.go create mode 100644 internal/squad/workflow_outcome_test.go create mode 100644 internal/taskrun/active_session_test.go create mode 100644 web/src/hooks/useActiveTaskRecovery.ts create mode 100644 web/src/hooks/useRecoveredTaskPhases.test.ts create mode 100644 web/src/lib/activeTaskCache.test.ts create mode 100644 web/src/lib/activeTaskCache.ts diff --git a/internal/agent/deliverable_test.go b/internal/agent/deliverable_test.go index 1539a259..023e9e4f 100644 --- a/internal/agent/deliverable_test.go +++ b/internal/agent/deliverable_test.go @@ -83,6 +83,31 @@ func TestMaterializeTargets_MissingAndEmptyFail(t *testing.T) { } } +func TestMaterializeTargets_EmptyPatchKeepsExistingFile(t *testing.T) { + workDir := t.TempDir() + body := []byte("# 调研报告\n已写在盘上的正文\n") + path := filepath.Join(workDir, "output.md") + if err := os.WriteFile(path, body, 0o644); err != nil { + t.Fatal(err) + } + got, err := MaterializeTargets(workDir, []string{"output.md"}, map[string][]byte{ + "output.md": nil, + }) + if err != nil { + t.Fatal(err) + } + if string(got["output.md"]) != string(body) { + t.Fatalf("collected %q", got["output.md"]) + } + disk, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(disk) != string(body) { + t.Fatalf("disk overwritten with %q", disk) + } +} + func TestMaterializeTargets_RejectsSymlinkEscape(t *testing.T) { root := t.TempDir() outside := t.TempDir() diff --git a/internal/app/app.go b/internal/app/app.go index 2d160c4b..10421fd9 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -244,6 +244,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { stages := stageflow.New(svc) coordEngine.SetStageTaskRunner(stages) coordEngine.SetContextServices(parts.Ctx, parts.KB) + coordEngine.SetSessionRunReader(sessionRunAdapter{mgr: taskMgr}) if instSvc != nil && instSvc.Available() { coordEngine.SetInstanceService(instSvc) } diff --git a/internal/app/chat_adapters.go b/internal/app/chat_adapters.go index c1a5c136..6c7fd273 100644 --- a/internal/app/chat_adapters.go +++ b/internal/app/chat_adapters.go @@ -7,6 +7,7 @@ import ( "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/chat" + "chat-agent-system/internal/coordinator" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/taskrun" ) @@ -117,6 +118,10 @@ func (a chatTaskStarter) CreateAndStart(ctx context.Context, runID string, recor }, bizdev.StageRunRequest{}, resumeDir) } +func (a chatTaskStarter) BindAssistantMsgID(ctx context.Context, runID string, assistantMsgID int64) error { + return a.inner.BindAssistantMsgID(ctx, runID, assistantMsgID) +} + // BindChatSession attaches composition-root adapters onto a chat session. // Callers that keep a mutable intentflow.Service (e.g. server test hooks) should // prefer AdaptChat* helpers with their own fields so swaps stay visible. @@ -138,3 +143,25 @@ func (rt *Runtime) BindChatSession(sess *chat.Session, env string) { Environment: env, } } + +type sessionRunAdapter struct { + mgr *taskrun.Manager +} + +func (a sessionRunAdapter) LatestSessionRun(ctx context.Context, sessionID string) (*coordinator.SessionRunSnapshot, error) { + if a.mgr == nil { + return nil, nil + } + rec, err := a.mgr.LatestSessionRun(ctx, sessionID) + if err != nil || rec == nil { + return nil, err + } + return &coordinator.SessionRunSnapshot{ + RunID: rec.RunID, + TaskKey: rec.TaskKey, + Status: rec.Status, + WorkspaceID: rec.WorkspaceID, + ErrorMessage: rec.ErrorMessage, + UpdatedAt: rec.UpdatedAt, + }, nil +} diff --git a/internal/chat/ports.go b/internal/chat/ports.go index 5acd651f..4fdb5bc2 100644 --- a/internal/chat/ports.go +++ b/internal/chat/ports.go @@ -70,6 +70,7 @@ type IntentStepView struct { // TaskStarter starts a task run without exposing bizdev/stageflow request types to chat. type TaskStarter interface { CreateAndStart(ctx context.Context, runID string, record *taskrun.RunRecord, start TaskStartRequest, resumeDir string) (*taskrun.RunRecord, error) + BindAssistantMsgID(ctx context.Context, runID string, assistantMsgID int64) error } // TaskStartRequest carries the chat-owned fields needed to start or resume a run. diff --git a/internal/chat/session.go b/internal/chat/session.go index 6bb6ea48..7ae7f034 100644 --- a/internal/chat/session.go +++ b/internal/chat/session.go @@ -465,6 +465,9 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str streamer.EmitPhase("dialogue_generating", "正在整理回复...", "assistant", 0, 0, 40) } s.EnsureDocResolver() + if s.UserCtx != nil { + s.UserCtx.ChatSessionID = s.ID + } coordRes, err := s.Coordinator.ProcessIntent(ctx, recRes, s.UserCtx) if err != nil { if streamer != nil { @@ -473,8 +476,12 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str return nil, fmt.Errorf("coordination execution failed: %w", err) } + var startedRunID string if coordRes.PendingAction != nil { execResult, execErr := s.executeAction(ctx, coordRes.PendingAction, s.UserCtx) + if execResult != nil { + startedRunID = execResult.RunID + } coordRes.NaturalReply = renderActionReply(coordRes.PendingAction, execResult, execErr) coordRes.PendingAction = nil } @@ -592,8 +599,17 @@ func (s *Session) sendMessageInternal(ctx context.Context, userInput string, str asstMeta["client_turn_id"] = clientTurnID asstMeta["client_id"] = clientTurnID + ":assistant" } + if startedRunID != "" { + asstMeta["run_id"] = startedRunID + if asstMeta["task_status"] == nil || asstMeta["task_status"] == "" { + asstMeta["task_status"] = "running" + } + } asstMsgID, _ := s.CtxMgr.RecordUserMessageWithID(ctx, s.ID, 0, "assistant", coordRes.NaturalReply, string(coordRes.Intent), string(recRes.Branch), asstMeta) + if startedRunID != "" && asstMsgID > 0 && s.Actions != nil && s.Actions.TaskRuns != nil { + _ = s.Actions.TaskRuns.BindAssistantMsgID(ctx, startedRunID, asstMsgID) + } // 异步记录问答切片到向量记忆库 (P2 检索增强，不阻塞流式或同步答复) sessID := s.ID diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index 831d0458..7dfb9ab7 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -222,6 +222,8 @@ type UserProjectContext struct { RegulationHits []kb.HitItem `json:"-"` // UpgradeOffered 路由升级建议冷却标记（每会话至多主动建议一次，防打扰——对齐"参谋不强推"人设） UpgradeOffered bool `json:"-"` + // ChatSessionID 当前对话会话，供进度问询读取真实 task run。不序列化。 + ChatSessionID string `json:"-"` // DocumentSummaryReader 仅用于服务端鉴权后的只读正文访问，不序列化。 DocumentSummaryReader DocumentSummaryReader `json:"-"` // RealtimeSnippet 本轮实时检索（天气/新闻等）成功结果，供 Prompt 注入；不序列化。 diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 260bb712..c8ecc344 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -83,6 +83,7 @@ type CoordinatorEngine struct { docResolver DocResolver pendingSupps PendingSupplementReader reopenLister ReopenLister + sessionRuns SessionRunReader // OnSupplement is optional; composition typically wires supplementimpact.TriggerAsync. // Nil-safe: unset hooks are skipped. OnSupplement func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string) @@ -108,6 +109,27 @@ func (e *CoordinatorEngine) tasks() StageTaskRunner { return e.stageTasks } +// SessionRunSnapshot 是进度问询可读的任务运行投影，不含工作目录绝对路径。 +type SessionRunSnapshot struct { + RunID string + TaskKey string + Status string + WorkspaceID string + ErrorMessage string + UpdatedAt time.Time +} + +// SessionRunReader 按会话读取最近一次任务运行。未注入时进度问询保持模板答复。 +type SessionRunReader interface { + LatestSessionRun(ctx context.Context, sessionID string) (*SessionRunSnapshot, error) +} + +func (e *CoordinatorEngine) SetSessionRunReader(reader SessionRunReader) { + if e != nil { + e.sessionRuns = reader + } +} + func (e *CoordinatorEngine) SetContextServices(ctxMgr *ctxmgr.ContextManager, kbClient *kb.Client) { if e != nil { e.ctxMgr = ctxMgr diff --git a/internal/coordinator/engine_query.go b/internal/coordinator/engine_query.go index dfdc47ce..7d2d0e45 100644 --- a/internal/coordinator/engine_query.go +++ b/internal/coordinator/engine_query.go @@ -151,6 +151,13 @@ func (e *CoordinatorEngine) handleQueryProgress(ctx context.Context, cand *inten sb.WriteString(brief.CommunicationKit) case "progress": + if snap := e.latestSessionRun(ctx, userCtx); snap != nil { + sb.WriteString(formatSessionRunReply(snap)) + if brief != nil { + sb.WriteString(fmt.Sprintf("\n项目成熟度仍为 %s（%d/100）。", brief.MaturityLevel, brief.MaturityScore)) + } + break + } sb.WriteString(fmt.Sprintf("【进展汇报】 项目当前成熟度等级为 %s（评分 %d/100）。\n", brief.MaturityLevel, brief.MaturityScore)) sb.WriteString(fmt.Sprintf("诊断结论：%s\n\n", brief.MaturityHeadline)) if len(brief.NextBestActions) > 0 { @@ -196,6 +203,49 @@ func (e *CoordinatorEngine) handleQueryProgress(ctx context.Context, cand *inten }), nil } +func (e *CoordinatorEngine) latestSessionRun(ctx context.Context, userCtx *UserProjectContext) *SessionRunSnapshot { + if e == nil || e.sessionRuns == nil || userCtx == nil || strings.TrimSpace(userCtx.ChatSessionID) == "" { + return nil + } + snap, err := e.sessionRuns.LatestSessionRun(ctx, userCtx.ChatSessionID) + if err != nil || snap == nil || snap.RunID == "" { + return nil + } + return snap +} + +func formatSessionRunReply(snap *SessionRunSnapshot) string { + label := map[string]string{ + "pending": "排队中", + "running": "正在编写", + "completed": "已完成", + "failed": "执行失败", + "finalizing_failed": "收尾失败", + "interrupted": "已中断，可以继续", + "cancelled": "已取消", + "pending_author_confirmation": "待编制人确认", + "quality_failed": "质量未通过", + "blocked_pending_governance": "待激活工作流", + } + status := label[snap.Status] + if status == "" { + status = snap.Status + } + taskName := snap.TaskKey + if taskName == "" { + taskName = "当前任务" + } + var b strings.Builder + b.WriteString(fmt.Sprintf("「%s」当前状态：%s。", taskName, status)) + if snap.WorkspaceID != "" { + b.WriteString(fmt.Sprintf("工作区标识：%s。", snap.WorkspaceID)) + } + if snap.ErrorMessage != "" && (snap.Status == "failed" || snap.Status == "interrupted" || snap.Status == "finalizing_failed") { + b.WriteString("原因：" + snap.ErrorMessage + "。") + } + return b.String() +} + // synthesizeBaselineBrief 在用户尚未执行过任何阶段/任务（无持久化参谋简报）时， // 基于黑板事实与项目信息合成 intake 级基准参谋简报，确保查询类意图始终有实质研判可用。 func synthesizeBaselineBrief(userCtx *UserProjectContext) *squad.AdvisorBriefing { diff --git a/internal/coordinator/session_run_reply_test.go b/internal/coordinator/session_run_reply_test.go new file mode 100644 index 00000000..374e2d76 --- /dev/null +++ b/internal/coordinator/session_run_reply_test.go @@ -0,0 +1,45 @@ +package coordinator + +import ( + "context" + "strings" + "testing" +) + +type fakeSessionRuns struct { + snap *SessionRunSnapshot + err error +} + +func (f fakeSessionRuns) LatestSessionRun(context.Context, string) (*SessionRunSnapshot, error) { + return f.snap, f.err +} + +func TestFormatSessionRunReply_RunningIncludesWorkspace(t *testing.T) { + got := formatSessionRunReply(&SessionRunSnapshot{ + TaskKey: "research", + Status: "running", + WorkspaceID: "ws_research_1", + }) + for _, want := range []string{"research", "正在编写", "ws_research_1"} { + if !strings.Contains(got, want) { + t.Fatalf("reply %q missing %q", got, want) + } + } +} + +func TestLatestSessionRun_ReadsInjectedReader(t *testing.T) { + engine := NewCoordinatorEngine(nil) + engine.SetSessionRunReader(fakeSessionRuns{snap: &SessionRunSnapshot{ + RunID: "run-1", + TaskKey: "research", + Status: "running", + }}) + got := engine.latestSessionRun(context.Background(), &UserProjectContext{ChatSessionID: "sess-1"}) + if got == nil || got.RunID != "run-1" { + t.Fatalf("latest = %#v", got) + } + if engine.latestSessionRun(context.Background(), &UserProjectContext{}) != nil { + t.Fatal("empty chat session must not invent a run") + } +} diff --git a/internal/intent/recognizer_matchers_task.go b/internal/intent/recognizer_matchers_task.go index d9a4e9a8..2511e64b 100644 --- a/internal/intent/recognizer_matchers_task.go +++ b/internal/intent/recognizer_matchers_task.go @@ -11,6 +11,30 @@ import ( "chat-agent-system/internal/squad" ) +// isProgressInquiry 为真时，带任务名词的疑问句应让行 QUERY_PROGRESS，而不是 TRIGGER_TASK。 +func isProgressInquiry(input string) bool { + if !isInterrogative(input) { + return false + } + markers := []string{"进度", "进展", "怎么样了", "状态如何", "写完", "完成了吗", "完成没有", "好了吗", "好了没有"} + hit := false + for _, marker := range markers { + if strings.Contains(input, marker) { + hit = true + break + } + } + if !hit { + return false + } + for _, verb := range []string{"开始", "重新", "再来", "重跑", "再生成", "重新生成"} { + if strings.Contains(input, verb) { + return false + } + } + return true +} + // isInterrogative 判断输入是否疑问句：句尾问号/「吗」「呢」，或句中出现 // 「怎么/如何/什么/多少/哪/是否/大概/多久/为什么」等疑问词。 // 用于把「查询/咨询」类输入从祈使句「指令」类意图（阶段推进等）中排除。 @@ -248,6 +272,11 @@ func requirementsTriggerPhrase(taskKey string) string { } func matchTriggerTask(input string, userCtx *ContextSnapshot) *CandidateIntent { + // 疑问句里的「编写进度怎么样了」是在问状态，不是再启动一份交付物。 + // 显式启动词（开始/重新/重跑）仍走任务触发。 + if isProgressInquiry(input) { + return nil + } type phrase struct { text string key string diff --git a/internal/intent/recognizer_test.go b/internal/intent/recognizer_test.go index 78f58015..9670f065 100644 --- a/internal/intent/recognizer_test.go +++ b/internal/intent/recognizer_test.go @@ -72,6 +72,30 @@ func TestIntentRecognition_AdvanceStage(t *testing.T) { } } +func TestIntentRecognition_ProgressQuestionDoesNotStartTask(t *testing.T) { + rec := intent.NewRecognizer() + ctx := &intent.ContextSnapshot{SnapshotID: "ctx_progress"} + + progress, err := rec.Recognize(context.Background(), "调研报告编写进度怎么样了", ctx) + if err != nil { + t.Fatalf("recognize progress: %v", err) + } + if progress.TopIntent.Intent != intent.IntentQueryProgress { + t.Fatalf("progress question intent = %s, want QUERY_PROGRESS", progress.TopIntent.Intent) + } + + start, err := rec.Recognize(context.Background(), "开始做调研报告", ctx) + if err != nil { + t.Fatalf("recognize start: %v", err) + } + if start.TopIntent.Intent != intent.IntentTriggerTask { + t.Fatalf("start intent = %s, want TRIGGER_TASK", start.TopIntent.Intent) + } + if start.TopIntent.FilledSlots[intent.SlotTargetTask] != "research" { + t.Fatalf("start task = %s, want research", start.TopIntent.FilledSlots[intent.SlotTargetTask]) + } +} + func TestIntentRecognition_TriggerTask(t *testing.T) { rec := intent.NewRecognizer() ctx := &intent.ContextSnapshot{ diff --git a/internal/server/server.go b/internal/server/server.go index a6059857..b95426e6 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -230,6 +230,7 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("POST /api/v1/tasks/run/stream", s.handleRunTaskStream) mux.HandleFunc("GET /api/v1/tasks/run/stream", s.handleRunTaskStream) mux.HandleFunc("POST /api/v1/tasks/runs", s.handleCreateTaskRun) + mux.HandleFunc("GET /api/v1/tasks/runs", s.handleListTaskRuns) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}", s.handleGetTaskRun) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}/events", s.handleTaskRunEvents) mux.HandleFunc("POST /api/v1/tasks/runs/{run_id}/cancel", s.handleCancelTaskRun) diff --git a/internal/server/sse_wave0_test.go b/internal/server/sse_wave0_test.go index 6c967ce1..cba91b0f 100644 --- a/internal/server/sse_wave0_test.go +++ b/internal/server/sse_wave0_test.go @@ -49,6 +49,12 @@ func (failInsertStore) HeartbeatLease(context.Context, string, string, time.Dura func (failInsertStore) ReleaseLease(context.Context, string, string) error { return nil } func (failInsertStore) ResetForResume(context.Context, string, int) error { return nil } func (failInsertStore) MarkOrphanedRunning(context.Context) (int, error) { return 0, nil } +func (failInsertStore) ListActiveBySession(context.Context, string) ([]*taskrun.RunRecord, error) { + return nil, nil +} +func (failInsertStore) LatestBySession(context.Context, string) (*taskrun.RunRecord, error) { + return nil, nil +} func TestRunTaskStream_InvalidJSON_Returns400(t *testing.T) { srv := newTestServer(t) diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index e2dc98bc..021986e9 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -209,6 +209,9 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { "doc_id": req.DocID, }, } + if req.ClientTurnID != "" { + asstMeta["client_id"] = req.ClientTurnID + ":assistant" + } projectinstance.ApplyScopeToMetadata(asstMeta, taskScope) assistantMsgID, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", asstMeta) claimed.AssistantMsgID = assistantMsgID @@ -613,6 +616,45 @@ func (s *Server) cancelPiAgentRunner(workspaceID string) { _ = resp.Body.Close() } +func (s *Server) handleListTaskRuns(w http.ResponseWriter, r *http.Request) { + user, ok := s.requirePrincipal(w, r) + if !ok { + return + } + sessionID := strings.TrimSpace(r.URL.Query().Get("session_id")) + if sessionID == "" { + writeSafeError(w, r, http.StatusBadRequest, "session_id required") + return + } + if r.URL.Query().Get("active") != "1" { + writeSafeError(w, r, http.StatusBadRequest, "active=1 required") + return + } + if !s.assertSessionOwnedByUser(w, r, sessionID, user.ID) { + return + } + if s.taskRuns() == nil { + writeJSONResponse(w, http.StatusOK, map[string]any{"runs": []taskRunResponse{}}) + return + } + listed, err := s.taskRuns().ListActiveBySession(r.Context(), sessionID) + if err != nil { + writeDomainError(w, r, http.StatusInternalServerError, err) + return + } + out := make([]taskRunResponse, 0, len(listed)) + for _, rec := range listed { + if rec == nil { + continue + } + if !s.isSystemAdminUser(r.Context(), user) && rec.UserID > 0 && rec.UserID != user.ID { + continue + } + out = append(out, s.buildTaskRunResponse(rec)) + } + writeJSONResponse(w, http.StatusOK, map[string]any{"runs": out}) +} + func (s *Server) handleGetTaskRun(w http.ResponseWriter, r *http.Request) { user, ok := s.requirePrincipal(w, r) if !ok { diff --git a/internal/squad/gov.go b/internal/squad/gov.go index 28efe723..354b3614 100644 --- a/internal/squad/gov.go +++ b/internal/squad/gov.go @@ -26,7 +26,7 @@ import ( // 步骤状态常量（与既有 workflow.json 的 legacy 状态字符串完全兼容） const ( StepStatePending = "pending" - StepStateExecuting = "running" // legacy 值保持不变 + StepStateExecuting = "running" // legacy 值保持不变 StepStateVerifying = "verifying" // 新增：校验态 StepStateRepairing = "repairing" // 新增：修复态 StepStateCompleted = "completed" @@ -213,15 +213,23 @@ func (l *BudgetLedger) UsageSummary() string { // TraceEntry 单次底层 agent 调用的轨迹记录 type TraceEntry struct { - Seq int `json:"seq"` - StepID string `json:"step_id"` - Role string `json:"role"` - Timestamp time.Time `json:"ts"` - PromptDigest string `json:"prompt_digest"` - OutputDigest string `json:"output_digest"` - SimilarityToPrev float64 `json:"similarity_to_prev"` - DurationMs int64 `json:"duration_ms"` - Success bool `json:"success"` + Seq int `json:"seq"` + StepID string `json:"step_id"` + Role string `json:"role"` + Timestamp time.Time `json:"ts"` + PromptDigest string `json:"prompt_digest"` + OutputDigest string `json:"output_digest"` + SimilarityToPrev float64 `json:"similarity_to_prev"` + DurationMs int64 `json:"duration_ms"` + Success bool `json:"success"` + ExecErr string `json:"exec_err,omitempty"` + ResultSuccess bool `json:"res_success,omitempty"` + ResultError string `json:"res_error_reason,omitempty"` + HasTrustedPatch bool `json:"has_trusted_patch,omitempty"` + WroteThisInvocation bool `json:"wrote_this_invocation,omitempty"` + MatErr string `json:"mat_err,omitempty"` + RecoveredFromDisk bool `json:"recovered_from_disk,omitempty"` + CancelReason string `json:"cancel_reason,omitempty"` } // AppendTrace 追加一条执行轨迹（jsonl，天然即历史轨迹审计源） @@ -462,6 +470,3 @@ func SortedStepIDs(wf *WorkflowPlan) []string { sort.Strings(ids) return ids } - - - diff --git a/internal/squad/workflow.go b/internal/squad/workflow.go index d8ca8a33..9889d43d 100644 --- a/internal/squad/workflow.go +++ b/internal/squad/workflow.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "log/slog" "os" "path/filepath" "strings" @@ -264,28 +265,31 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen for _, s := range parallelGroup { sCopy := s wg.Add(1) - go func() { - defer wg.Done() - defer func() { - if rec := recover(); rec != nil { - mu.Lock() - if groupErr == nil { - groupErr = fmt.Errorf("parallel step panic: %v", rec) - } - mu.Unlock() - } - }() - if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { - mu.Lock() - if groupErr == nil { - groupErr = err - } - mu.Unlock() - } - }() + go func() { + defer wg.Done() + defer func() { + if rec := recover(); rec != nil { + mu.Lock() + if groupErr == nil { + groupErr = fmt.Errorf("parallel step panic: %v", rec) + } + mu.Unlock() + } + }() + if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { + mu.Lock() + if groupErr == nil { + groupErr = err + } + mu.Unlock() + } + }() } wg.Wait() if groupErr != nil { + if cerr := contextErrString(runCtx); cerr != "" { + slog.Warn("squad_parallel_cancelled", "error", groupErr.Error(), "cancel_reason", cerr, "parent_cancel", contextErrString(ctx)) + } wf.Status = "failed" _ = d.SaveWorkflow(wsDir, wf) _ = d.updateWorkspaceStatus(wsDir, "failed") @@ -335,25 +339,25 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen Duration: time.Since(start), }, rErr } - // 治理：受控迁移至执行态（legacy 重跑场景直接放行） - markStepExecuting(wf, step.ID) - if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { - // A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded - if !step.IsCritical() { - markStepDegraded(wsDir, wf, step, err) - stepsRun++ + // 治理：受控迁移至执行态（legacy 重跑场景直接放行） + markStepExecuting(wf, step.ID) + if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil { + // A5 非关键步骤降级：记录 degraded 继续推进，收口统一标 degraded + if !step.IsCritical() { + markStepDegraded(wsDir, wf, step, err) + stepsRun++ + _ = d.SaveWorkflow(wsDir, wf) + continue + } + wf.Status = "failed" _ = d.SaveWorkflow(wsDir, wf) - continue + _ = d.updateWorkspaceStatus(wsDir, "failed") + return &WorkflowExecutionResult{ + Success: false, + ErrorReason: err.Error(), + Duration: time.Since(start), + }, err } - wf.Status = "failed" - _ = d.SaveWorkflow(wsDir, wf) - _ = d.updateWorkspaceStatus(wsDir, "failed") - return &WorkflowExecutionResult{ - Success: false, - ErrorReason: err.Error(), - Duration: time.Since(start), - }, err - } stepsRun++ // 治理：FSM 受控迁移 + 步骤归档锁定 + 软线提醒 if tErr := TransitionStep(wf, step.ID, StepStateCompleted); tErr == nil { @@ -549,13 +553,13 @@ func (d *WorkflowDriver) runOneStep( } var ( - res *agent.Result - execErr error - matErr error - duration int64 - ok bool - lastAssess agent.FailureAssessment - basePrompt = string(taskData) + res *agent.Result + execErr error + matErr error + duration int64 + ok bool + lastAssess agent.FailureAssessment + basePrompt = string(taskData) ) for attempt := 1; attempt <= agent.MaxTransientStepInvocations; attempt++ { onDisk, _ := os.ReadFile(prevOutPath) @@ -611,22 +615,51 @@ func (d *WorkflowDriver) runOneStep( if newContent == "" && wroteThisInvocation { newContent = string(onDisk) } - ok = execErr == nil && res != nil && res.Success && matErr == nil + recoveredFromDisk := false + if strings.TrimSpace(newContent) == "" { + if again, rerr := os.ReadFile(prevOutPath); rerr == nil && deliverableWroteThisInvocation(prevContent, again) { + newContent = string(again) + wroteThisInvocation = true + recoveredFromDisk = true + } + } + ok = stepOutputAccepted(execErr, matErr, wroteThisInvocation, newContent, res) + if !ok || recoveredFromDisk { + slog.Info("squad_step_output_judge", + "step_id", step.ID, + "exec_err", errString(execErr), + "res_success", res != nil && res.Success, + "res_error_reason", resultErrorReason(res), + "has_trusted_patch", hasTrustedPatch, + "wrote_this_invocation", wroteThisInvocation, + "mat_err", errString(matErr), + "recovered_from_disk", recoveredFromDisk, + "accepted", ok, + ) + } sim := OutputSimilarity(prevContent, newContent) if mu != nil { mu.Lock() } traceSeq := len(LoadTrace(wsDir)) + 1 _ = AppendTrace(wsDir, TraceEntry{ - Seq: traceSeq, - StepID: step.ID, - Role: step.Role, - Timestamp: time.Now(), - PromptDigest: digestString(string(taskData)), - OutputDigest: digestString(newContent), - SimilarityToPrev: sim, - DurationMs: duration, - Success: ok, + Seq: traceSeq, + StepID: step.ID, + Role: step.Role, + Timestamp: time.Now(), + PromptDigest: digestString(string(taskData)), + OutputDigest: digestString(newContent), + SimilarityToPrev: sim, + DurationMs: duration, + Success: ok, + ExecErr: errString(execErr), + ResultSuccess: res != nil && res.Success, + ResultError: resultErrorReason(res), + HasTrustedPatch: hasTrustedPatch, + WroteThisInvocation: wroteThisInvocation, + MatErr: errString(matErr), + RecoveredFromDisk: strings.TrimSpace(newContent) != "" && !(res != nil && res.Success), + CancelReason: contextErrString(ctx), }) if mu != nil { mu.Unlock() @@ -662,7 +695,7 @@ func (d *WorkflowDriver) runOneStep( } } - if execErr != nil { + if !ok && execErr != nil { if streamer != nil { streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行异常：%s", roleTitle, execErr.Error()), step.Role, 0, 0, 0) } @@ -680,7 +713,7 @@ func (d *WorkflowDriver) runOneStep( return fmt.Errorf("step %s (%s) backend execute: %w", step.ID, step.Role, execErr) } - if res != nil && !res.Success { + if !ok && res != nil && !res.Success { errMsg := res.ErrorReason if streamer != nil { streamer.EmitPhase("step_failed", fmt.Sprintf("%s执行未成功：%s", roleTitle, errMsg), step.Role, 0, 0, 0) @@ -761,6 +794,55 @@ func (d *WorkflowDriver) runOneStep( return nil } +// stepOutputAccepted 以本轮写盘产物为事实源。runner 把 Success 标成 false +// （例如 typed result 没带回 patch）但文件已经写好时，不让这一处否决步骤。 +func stepOutputAccepted(execErr, matErr error, wrote bool, newContent string, res *agent.Result) bool { + if execErr != nil || matErr != nil { + return false + } + if res != nil && res.Success { + return true + } + if !wrote || strings.TrimSpace(newContent) == "" { + return false + } + // 连接中断或可重试失败留下的文件可能是截断的，不能当成交付成功。 + // 交付物缺失/空结果但盘上已有正文时，以文件为准。 + if res != nil && (res.Retryable || hardRunnerFailure(res.ErrorReason)) { + return false + } + return true +} + +func hardRunnerFailure(reason string) bool { + reason = strings.ToLower(strings.TrimSpace(reason)) + if reason == "" || strings.Contains(reason, "missing target deliverable") { + return false + } + return true +} + +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +func resultErrorReason(res *agent.Result) string { + if res == nil { + return "" + } + return res.ErrorReason +} + +func contextErrString(ctx context.Context) string { + if ctx == nil || ctx.Err() == nil { + return "" + } + return ctx.Err().Error() +} + func deliverableWroteThisInvocation(baseline string, onDisk []byte) bool { if len(bytes.TrimSpace(onDisk)) == 0 { return false diff --git a/internal/squad/workflow_outcome_test.go b/internal/squad/workflow_outcome_test.go new file mode 100644 index 00000000..bfcfd083 --- /dev/null +++ b/internal/squad/workflow_outcome_test.go @@ -0,0 +1,31 @@ +package squad + +import ( + "errors" + "testing" + + "chat-agent-system/internal/agent" +) + +func TestStepOutputAccepted_DiskContentBeatsMissingDeliverable(t *testing.T) { + missing := &agent.Result{Success: false, ErrorReason: "missing target deliverable output.md"} + if !stepOutputAccepted(nil, nil, true, "# 调研报告\n正文", missing) { + t.Fatal("non-empty file must be accepted when the runner only reported a missing deliverable") + } + truncated := &agent.Result{Success: false, ErrorReason: "Connection error.", Retryable: true} + if stepOutputAccepted(nil, nil, true, "# partial", truncated) { + t.Fatal("retryable connection failure must stay fail-closed even if a file exists") + } + if stepOutputAccepted(nil, nil, true, " \n", missing) { + t.Fatal("blank content must not override a failed runner result") + } + if stepOutputAccepted(errors.New("backend"), nil, true, "# 调研报告", missing) { + t.Fatal("execute error must reject the step") + } + if stepOutputAccepted(nil, errors.New("materialize"), true, "# 调研报告", &agent.Result{Success: true}) { + t.Fatal("materialize error must reject the step") + } + if !stepOutputAccepted(nil, nil, false, "", &agent.Result{Success: true}) { + t.Fatal("runner success without a new file still counts") + } +} diff --git a/internal/taskrun/active_session_test.go b/internal/taskrun/active_session_test.go new file mode 100644 index 00000000..3dd7c631 --- /dev/null +++ b/internal/taskrun/active_session_test.go @@ -0,0 +1,36 @@ +package taskrun + +import ( + "context" + "testing" +) + +func TestListActiveBySession_MemoryOnly(t *testing.T) { + m := NewManager(nil, nil) + m.mu.Lock() + m.runs["run-live"] = &activeRun{record: &RunRecord{ + RunID: "run-live", SessionID: "sess-1", Status: "running", TaskKey: "research", + }} + m.runs["run-done"] = &activeRun{record: &RunRecord{ + RunID: "run-done", SessionID: "sess-1", Status: "completed", + }} + m.runs["other"] = &activeRun{record: &RunRecord{ + RunID: "other", SessionID: "sess-2", Status: "running", + }} + m.mu.Unlock() + + listed, err := m.ListActiveBySession(context.Background(), "sess-1") + if err != nil { + t.Fatal(err) + } + if len(listed) != 1 || listed[0].RunID != "run-live" { + t.Fatalf("listed = %#v", listed) + } +} + +func TestWriteRunningAnchor_NoContextManagerIsNoop(t *testing.T) { + m := NewManager(nil, nil) + m.writeRunningAnchor(context.Background(), &RunRecord{ + RunID: "run-1", AssistantMsgID: 9, TaskKey: "research", Status: "running", + }) +} diff --git a/internal/taskrun/claim_idempotency_test.go b/internal/taskrun/claim_idempotency_test.go index 1cc97d9e..be213e4d 100644 --- a/internal/taskrun/claim_idempotency_test.go +++ b/internal/taskrun/claim_idempotency_test.go @@ -91,6 +91,37 @@ func (m *memIdemStore) HeartbeatLease(context.Context, string, string, time.Dura func (m *memIdemStore) ReleaseLease(context.Context, string, string) error { return nil } func (m *memIdemStore) ResetForResume(context.Context, string, int) error { return nil } func (m *memIdemStore) MarkOrphanedRunning(context.Context) (int, error) { return 0, nil } +func (m *memIdemStore) ListActiveBySession(_ context.Context, sessionID string) ([]*RunRecord, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]*RunRecord, 0) + for _, rec := range m.byID { + if rec == nil || rec.SessionID != sessionID { + continue + } + if rec.Status != "pending" && rec.Status != "running" { + continue + } + cpy := *rec + out = append(out, &cpy) + } + return out, nil +} +func (m *memIdemStore) LatestBySession(_ context.Context, sessionID string) (*RunRecord, error) { + m.mu.Lock() + defer m.mu.Unlock() + var best *RunRecord + for _, rec := range m.byID { + if rec == nil || rec.SessionID != sessionID { + continue + } + if best == nil || rec.UpdatedAt.After(best.UpdatedAt) { + cpy := *rec + best = &cpy + } + } + return best, nil +} func TestClaimRun_ConcurrentSameKey_OneOwner(t *testing.T) { store := newMemIdemStore() diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 0500d316..452ba5a5 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -255,8 +255,11 @@ func (m *Manager) BindAssistantMsgID(ctx context.Context, runID string, assistan rec := *ar.record ar.mu.Unlock() if m.store != nil && m.store.Available() { - return m.store.UpdateRun(ctx, &rec) + if err := m.store.UpdateRun(ctx, &rec); err != nil { + return err + } } + m.writeRunningAnchor(ctx, &rec) return nil } if m.store != nil && m.store.Available() { @@ -266,11 +269,121 @@ func (m *Manager) BindAssistantMsgID(ctx context.Context, runID string, assistan } rec.AssistantMsgID = assistantMsgID rec.UpdatedAt = time.Now() - return m.store.UpdateRun(ctx, rec) + if err := m.store.UpdateRun(ctx, rec); err != nil { + return err + } + m.writeRunningAnchor(ctx, rec) + return nil } return nil } +// writeRunningAnchor 在 run 仍进行中时就把 run_id / task_status 写进助手消息。 +// 刷新后窗口能按 taskRunId 重放进度，不必等终态 finalize。 +func (m *Manager) writeRunningAnchor(ctx context.Context, rec *RunRecord) { + if m == nil || m.ctxMgr == nil || rec == nil || rec.AssistantMsgID <= 0 || rec.RunID == "" { + return + } + meta := map[string]any{ + "run_id": rec.RunID, + "task_key": rec.TaskKey, + "task_status": "running", + } + if rec.WorkspaceID != "" { + meta["workspace_id"] = rec.WorkspaceID + } + if err := m.ctxMgr.UpdateMessageMetadata(ctx, rec.AssistantMsgID, meta, "running"); err != nil { + slog.Warn("taskrun_running_anchor_failed", "run_id", rec.RunID, "message_id", rec.AssistantMsgID, "error", err) + } +} + +// ListActiveBySession 返回该会话仍在 pending/running 且租约未过期的 run。 +// 内存中的活跃 run 与存储合并，覆盖尚未落库的窗口。 +func (m *Manager) ListActiveBySession(ctx context.Context, sessionID string) ([]*RunRecord, error) { + if m == nil || sessionID == "" { + return nil, nil + } + var stored []*RunRecord + if m.store != nil && m.store.Available() { + listed, err := m.store.ListActiveBySession(ctx, sessionID) + if err != nil { + return nil, err + } + stored = listed + } + return mergeRuns(stored, m.memorySessionRuns(sessionID, true)), nil +} + +// LatestSessionRun 返回该会话最近一次 run（含已结束），供进度问询读取真实状态。 +func (m *Manager) LatestSessionRun(ctx context.Context, sessionID string) (*RunRecord, error) { + if m == nil || sessionID == "" { + return nil, nil + } + var best *RunRecord + if m.store != nil && m.store.Available() { + rec, err := m.store.LatestBySession(ctx, sessionID) + if err != nil { + return nil, err + } + best = rec + } + for _, mem := range m.memorySessionRuns(sessionID, false) { + if best == nil || mem.UpdatedAt.After(best.UpdatedAt) { + cpy := *mem + best = &cpy + } + } + return best, nil +} + +func (m *Manager) memorySessionRuns(sessionID string, activeOnly bool) []*RunRecord { + m.mu.RLock() + defer m.mu.RUnlock() + out := make([]*RunRecord, 0) + for _, ar := range m.runs { + if ar == nil || ar.record == nil || ar.record.SessionID != sessionID { + continue + } + ar.mu.Lock() + rec := *ar.record + ar.mu.Unlock() + if activeOnly && rec.Status != "pending" && rec.Status != "running" { + continue + } + out = append(out, &rec) + } + return out +} + +func mergeRuns(stored, memory []*RunRecord) []*RunRecord { + byID := make(map[string]*RunRecord, len(stored)+len(memory)) + order := make([]string, 0, len(stored)+len(memory)) + add := func(rec *RunRecord) { + if rec == nil || rec.RunID == "" { + return + } + if prev, ok := byID[rec.RunID]; ok { + if rec.UpdatedAt.After(prev.UpdatedAt) { + byID[rec.RunID] = rec + } + return + } + byID[rec.RunID] = rec + order = append(order, rec.RunID) + } + for _, rec := range stored { + add(rec) + } + for _, rec := range memory { + add(rec) + } + out := make([]*RunRecord, 0, len(order)) + for _, id := range order { + out = append(out, byID[id]) + } + return out +} + // StartClaimedRun 仅对 ClaimRun 成功的 owner 启 worker；禁止在 Insert 冲突后调用。 func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) (*RunRecord, error) { if record == nil || record.RunID == "" { @@ -303,6 +416,8 @@ func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, tas m.runs[record.RunID] = ar m.mu.Unlock() + m.writeRunningAnchor(parent, record) + if m.pwSvc != nil && record.ProjectID > 0 { startedBy := record.UserID _, _ = m.pwSvc.CreateWorkflowRun(workerCtx, &projectworkflow.WorkflowRun{ diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 93613d8e..65aadc67 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -26,6 +26,8 @@ type Store interface { GetByIdempotencyKey(ctx context.Context, key string) (*RunRecord, error) AppendEvent(ctx context.Context, runID string, seq int64, ev stream.StreamEvent) error GetRun(ctx context.Context, runID string) (*RunRecord, error) + ListActiveBySession(ctx context.Context, sessionID string) ([]*RunRecord, error) + LatestBySession(ctx context.Context, sessionID string) (*RunRecord, error) ListEvents(ctx context.Context, runID string, afterSeq int64) ([]StoredEvent, error) AcquireLease(ctx context.Context, runID, workerID string, ttl time.Duration) error HeartbeatLease(ctx context.Context, runID, workerID string, ttl time.Duration) error @@ -248,6 +250,108 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) return &rec, nil } +func (s *PGStore) ListActiveBySession(ctx context.Context, sessionID string) ([]*RunRecord, error) { + if !s.Available() { + return nil, ErrStoreNotAvailable + } + if sessionID == "" { + return nil, nil + } + where := `WHERE session_id = $1 AND status IN ('pending', 'running')` + if s.hasLeaseColumns(ctx) { + where += ` AND (lease_expires_at IS NULL OR lease_expires_at > now())` + } + where += ` ORDER BY updated_at DESC` + return s.queryRuns(ctx, where, sessionID) +} + +func (s *PGStore) LatestBySession(ctx context.Context, sessionID string) (*RunRecord, error) { + if !s.Available() { + return nil, ErrStoreNotAvailable + } + if sessionID == "" { + return nil, nil + } + rows, err := s.queryRuns(ctx, `WHERE session_id = $1 ORDER BY updated_at DESC LIMIT 1`, sessionID) + if err != nil { + return nil, err + } + if len(rows) == 0 { + return nil, nil + } + return rows[0], nil +} + +func (s *PGStore) queryRuns(ctx context.Context, where string, args ...any) ([]*RunRecord, error) { + withLease := s.hasLeaseColumns(ctx) + q := runSelectSQL(withLease) + " " + where + rows, err := s.pool.Query(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]*RunRecord, 0) + for rows.Next() { + rec, err := scanRunRecord(rows, withLease) + if err != nil { + return nil, err + } + out = append(out, rec) + } + return out, rows.Err() +} + +func runSelectSQL(withLease bool) string { + if withLease { + return `SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, + task_key, stage, status, workspace_id, assistant_message_id, + error_message, worker_id, lease_expires_at, attempt, created_at, updated_at + FROM task_runs` + } + return `SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, + task_key, stage, status, workspace_id, assistant_message_id, + error_message, created_at, updated_at + FROM task_runs` +} + +type runScanner interface { + Scan(dest ...any) error +} + +func scanRunRecord(row runScanner, withLease bool) (*RunRecord, error) { + var rec RunRecord + var projectID, assistantMsgID *int64 + var idempotencyKey *string + var err error + if withLease { + err = row.Scan( + &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, + &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, + &rec.ErrorMessage, &rec.WorkerID, &rec.LeaseExpiresAt, &rec.Attempt, + &rec.CreatedAt, &rec.UpdatedAt, + ) + } else { + err = row.Scan( + &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, + &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, + &rec.ErrorMessage, &rec.CreatedAt, &rec.UpdatedAt, + ) + } + if err != nil { + return nil, err + } + if projectID != nil { + rec.ProjectID = *projectID + } + if assistantMsgID != nil { + rec.AssistantMsgID = *assistantMsgID + } + if idempotencyKey != nil { + rec.IdempotencyKey = *idempotencyKey + } + return &rec, nil +} + // GetByIdempotencyKey 按稳定幂等键查找既有 run；无键或未命中返回 (nil, nil)。 func (s *PGStore) GetByIdempotencyKey(ctx context.Context, key string) (*RunRecord, error) { if !s.Available() { @@ -302,7 +406,6 @@ func (s *PGStore) ListEvents(ctx context.Context, runID string, afterSeq int64) return out, rows.Err() } - // CancelOpenByProject cancels pending/running runs for a project inside an existing transaction. // Table ownership stays with taskrun; workflow terminate must call this instead of writing task_runs directly. func CancelOpenByProject(ctx context.Context, tx pgx.Tx, projectID int64, at time.Time, errMsg string) error { @@ -500,6 +603,12 @@ func (noopStore) AppendEvent(context.Context, string, int64, stream.StreamEvent) func (noopStore) GetRun(context.Context, string) (*RunRecord, error) { return nil, ErrStoreNotAvailable } +func (noopStore) ListActiveBySession(context.Context, string) ([]*RunRecord, error) { + return nil, ErrStoreNotAvailable +} +func (noopStore) LatestBySession(context.Context, string) (*RunRecord, error) { + return nil, ErrStoreNotAvailable +} func (noopStore) GetByIdempotencyKey(context.Context, string) (*RunRecord, error) { return nil, ErrStoreNotAvailable } diff --git a/web/src/App.tsx b/web/src/App.tsx index 94dbae78..897d51a6 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -120,6 +120,7 @@ import { saveFile } from './data/fileStore'; import { cancelTaskRun, getTaskRun } from './api/taskRuns'; import { cancelInflightAssistantMessages } from './lib/squadCompletionObservation'; import { useRecoveredTaskPhases } from './hooks/useRecoveredTaskPhases'; +import { useActiveTaskRecovery } from './hooks/useActiveTaskRecovery'; export { ALL_INITIAL_DELIVERABLES, @@ -1012,6 +1013,14 @@ export default function App() { chatWindow.updateLiveMessage, ); + useActiveTaskRecovery( + authenticatedSessionId, + chatWindow.phase === 'ready', + chatWindow.allMessages, + chatWindow.updateLiveMessage, + chatWindow.appendLiveMessage, + ); + // P1: 多维度历史导航与分类统计、日历、引用状态 const [categoryStats, setCategoryStats] = useState([]); const [dateStats, setDateStats] = useState([]); diff --git a/web/src/api/chatWindow.ts b/web/src/api/chatWindow.ts index 2c52a8be..a6616c61 100644 --- a/web/src/api/chatWindow.ts +++ b/web/src/api/chatWindow.ts @@ -230,6 +230,19 @@ export function mapRecordToMessage(r: RawChatMessageRecord): ChatMessage { : undefined; const msgId = meta.client_id || (r.id ? `msg-${r.id}` : `msg-tmp-${Date.now()}-${Math.random().toString(36).substring(2, 6)}`); const phaseSnapshot = parseTaskPhaseSnapshot(meta.task_phase_snapshot); + const hasRunID = typeof meta.run_id === 'string' && meta.run_id !== ''; + const rawTaskStatus = typeof meta.task_status === 'string' ? meta.task_status : ''; + let messageStatus = mapTaskStatusToMessageStatus(meta.task_status, r.role); + // 有 run 锚点但还没有终态快照、也没有 task_status 时，不能当成已完成。 + if ( + r.role === 'assistant' && + hasRunID && + !(phaseSnapshot?.steps && phaseSnapshot.steps.length > 0) && + rawTaskStatus === '' && + messageStatus === 'completed' + ) { + messageStatus = 'streaming_status'; + } return { id: msgId, clientTurnId: meta.client_turn_id || undefined, @@ -249,7 +262,7 @@ export function mapRecordToMessage(r: RawChatMessageRecord): ChatMessage { interactiveCard: meta.interactive_card, recommendedCases: meta.recommended_cases, quotes: meta.quotes || (meta.quote ? [meta.quote] : undefined), - status: mapTaskStatusToMessageStatus(meta.task_status, r.role), + status: messageStatus, errorInfo: persistedError || legacyBusyError || legacyTaskError, taskRequest, taskRunId: typeof meta.run_id === 'string' ? meta.run_id : undefined, diff --git a/web/src/api/taskRuns.ts b/web/src/api/taskRuns.ts index 9dd2583b..e09f9f17 100644 --- a/web/src/api/taskRuns.ts +++ b/web/src/api/taskRuns.ts @@ -180,6 +180,26 @@ export async function cancelTaskRun(runId: string): Promise { } } +export type ActiveTaskRun = { + run_id: string; + status: string; + task_key?: string; + workspace_id?: string; + assistant_message_id?: number; + error_message?: string; + session_id?: string; +}; + +export async function listActiveTaskRuns(sessionId: string): Promise { + const url = `/api/v1/tasks/runs?session_id=${encodeURIComponent(sessionId)}&active=1`; + const res = await apiFetch(url, { headers: getBearerAuthHeaders() }); + if (!res.ok) { + throw new Error(`list active task runs failed: HTTP ${res.status}`); + } + const body = (await res.json()) as { runs?: ActiveTaskRun[] }; + return Array.isArray(body.runs) ? body.runs : []; +} + export async function getTaskRun(runId: string) { const res = await apiFetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}`, { headers: getBearerAuthHeaders(), diff --git a/web/src/engine/templateMatch.test.ts b/web/src/engine/templateMatch.test.ts index e43e0a8e..060a9472 100644 --- a/web/src/engine/templateMatch.test.ts +++ b/web/src/engine/templateMatch.test.ts @@ -359,3 +359,9 @@ test('weighted mode does not let a cross-workflow title hijack an in-scope write const fuzzy = resolveTaskTemplateForMessage('编写建议', pool, null, 'main', 'weighted'); assert.equal(fuzzy?.wrongWorkflow, undefined); }); + +test('progress question is read-only; explicit start still resolves research', () => { + assert.equal(hasDocumentWriteIntent('调研报告编写进度怎么样了'), false); + assert.equal(resolveTaskKeyFromMessage('调研报告编写进度怎么样了'), undefined); + assert.equal(resolveTaskKeyFromMessage('开始做调研报告'), 'research'); +}); diff --git a/web/src/hooks/useActiveTaskRecovery.ts b/web/src/hooks/useActiveTaskRecovery.ts new file mode 100644 index 00000000..afea9b2c --- /dev/null +++ b/web/src/hooks/useActiveTaskRecovery.ts @@ -0,0 +1,114 @@ +import { useEffect } from 'react'; +import { listActiveTaskRuns, type ActiveTaskRun } from '../api/taskRuns'; +import { + placeholderFromActiveTask, + readActiveTasks, + removeActiveTask, + upsertActiveTask, + type ActiveTaskCacheEntry, +} from '../lib/activeTaskCache'; +import type { ChatMessage } from '../types/workbench'; + +const FRESH_MS = 20_000; + +/** + * 刷新后先用本地进行中缓存画出进度卡片，再用会话活跃 run 列表收敛。 + * 服务端没有的旧缓存会被清掉；刚写入、列表还没跟上的条目保留一小段时间。 + */ +export function useActiveTaskRecovery( + sessionId: string, + windowReady: boolean, + messages: ChatMessage[], + updateMessage: (id: string, patch: Partial) => void, + appendMessage: (message: ChatMessage) => void, +) { + useEffect(() => { + if (!sessionId || !windowReady) return; + let cancelled = false; + const cached = readActiveTasks(sessionId); + for (const entry of cached) { + if (messages.some((message) => message.taskRunId === entry.runId)) continue; + const byClient = messages.find( + (message) => message.id === entry.assistantClientId || message.clientMessageId === entry.assistantClientId, + ); + if (byClient) { + updateMessage(byClient.id, { + taskRunId: entry.runId, + status: 'streaming_status', + livePhaseLabel: entry.lastPhaseLabel || byClient.livePhaseLabel || '任务执行中…', + serverId: byClient.serverId || entry.serverMessageId, + }); + continue; + } + appendMessage(placeholderFromActiveTask(entry)); + } + + void listActiveTaskRuns(sessionId) + .then((runs) => { + if (cancelled) return; + const activeIDs = new Set(runs.map((run) => run.run_id).filter(Boolean)); + for (const run of runs) { + applyServerRun(sessionId, run, messages, updateMessage, appendMessage); + } + for (const entry of readActiveTasks(sessionId)) { + if (activeIDs.has(entry.runId)) continue; + const age = Date.now() - Date.parse(entry.startedAt); + if (Number.isFinite(age) && age >= 0 && age < FRESH_MS) continue; + removeActiveTask(entry.runId); + } + }) + .catch((err) => { + console.warn('active task sync failed', err); + }); + + return () => { + cancelled = true; + }; + // 窗口 ready 后再画本地缓存并拉活跃 run，避免被随后的窗口替换清掉。 + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [sessionId, windowReady]); +} + +function applyServerRun( + sessionId: string, + run: ActiveTaskRun, + messages: ChatMessage[], + updateMessage: (id: string, patch: Partial) => void, + appendMessage: (message: ChatMessage) => void, +) { + if (!run.run_id) return; + const existing = + messages.find((message) => message.taskRunId === run.run_id) || + messages.find((message) => run.assistant_message_id && message.serverId === run.assistant_message_id); + const label = existing?.livePhaseLabel || '任务执行中…'; + if (existing) { + updateMessage(existing.id, { + taskRunId: run.run_id, + status: existing.status === 'completed' ? existing.status : 'streaming_status', + taskWorkspaceId: run.workspace_id || existing.taskWorkspaceId, + serverId: existing.serverId || run.assistant_message_id, + livePhaseLabel: label, + }); + } else { + const entry: ActiveTaskCacheEntry = { + sessionId, + runId: run.run_id, + assistantClientId: run.assistant_message_id ? `msg-${run.assistant_message_id}` : `ast-active-${run.run_id}`, + taskKey: run.task_key || '', + startedAt: new Date().toISOString(), + lastPhaseLabel: label, + serverMessageId: run.assistant_message_id, + }; + appendMessage(placeholderFromActiveTask(entry)); + } + const previous = readActiveTasks(sessionId).find((row) => row.runId === run.run_id); + upsertActiveTask({ + sessionId, + runId: run.run_id, + assistantClientId: previous?.assistantClientId || existing?.id || `ast-active-${run.run_id}`, + taskKey: run.task_key || previous?.taskKey || '', + startedAt: previous?.startedAt || new Date().toISOString(), + lastPhaseLabel: previous?.lastPhaseLabel || label, + serverMessageId: run.assistant_message_id || previous?.serverMessageId, + }); +} diff --git a/web/src/hooks/useChatWindow.test.ts b/web/src/hooks/useChatWindow.test.ts index b0df1a02..e6042c22 100644 --- a/web/src/hooks/useChatWindow.test.ts +++ b/web/src/hooks/useChatWindow.test.ts @@ -165,3 +165,18 @@ test('mapRecordToMessage: restores task phase snapshot into card fields', () => assert.equal(msg.phaseSnapshot?.title, '用户初始需求收集表'); assert.equal(msg.phaseEvents?.[1].detail, '门禁'); }); + +test('mapRecordToMessage: run anchor without task_status stays streaming', () => { + const msg = mapRecordToMessage({ + id: 42, + session_id: 'sess-1', + user_id: 1, + role: 'assistant', + content: '任务执行中…', + metadata: { run_id: 'run-live', task_key: 'research' }, + day_key: '2026-09-28', + created_at: '2026-09-28T03:27:00Z', + }); + assert.equal(msg.taskRunId, 'run-live'); + assert.equal(msg.status, 'streaming_status'); +}); diff --git a/web/src/hooks/useRecoveredTaskPhases.test.ts b/web/src/hooks/useRecoveredTaskPhases.test.ts new file mode 100644 index 00000000..f978e312 --- /dev/null +++ b/web/src/hooks/useRecoveredTaskPhases.test.ts @@ -0,0 +1,18 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + replayAttemptLimit, + replayBackoffMs, + shouldAbandonTaskReplay, +} from './useRecoveredTaskPhases'; + +test('replay retries 502 and stops on auth or not-found', () => { + assert.equal(replayAttemptLimit(false), 4); + assert.equal(replayAttemptLimit(true), 2); + assert.equal(replayBackoffMs(0), 500); + assert.equal(replayBackoffMs(20), 5000); + assert.equal(shouldAbandonTaskReplay(502, 0, 4), false); + assert.equal(shouldAbandonTaskReplay(502, 3, 4), true); + assert.equal(shouldAbandonTaskReplay(401, 0, 4), true); + assert.equal(shouldAbandonTaskReplay(404, 0, 4), true); +}); diff --git a/web/src/hooks/useRecoveredTaskPhases.ts b/web/src/hooks/useRecoveredTaskPhases.ts index 3ce61b4c..f1e90bf7 100644 --- a/web/src/hooks/useRecoveredTaskPhases.ts +++ b/web/src/hooks/useRecoveredTaskPhases.ts @@ -1,6 +1,6 @@ import { useEffect, useRef } from 'react'; import type { StreamEvent } from '../api/chatStream'; -import { readTaskRunSSEOnce, subscribeTaskRunEvents } from '../api/taskRuns'; +import { readTaskRunSSEOnce, subscribeTaskRunEvents, TaskRunHttpError } from '../api/taskRuns'; import { applyPhaseEvent } from '../engine/phaseEventMapper'; import { messageNeedsPhaseReplay, @@ -80,15 +80,38 @@ async function replayRun( cache.set(runId, { events: strip, label }); updateMessage(messageId, { phaseEvents: strip, livePhaseLabel: label }); }; - try { - if (live) { - await subscribeTaskRunEvents(runId, { onEvent }, signal); + const attempts = replayAttemptLimit(live); + for (let attempt = 0; attempt < attempts; attempt++) { + if (signal.aborted) return; + try { + if (live) { + await subscribeTaskRunEvents(runId, { onEvent }, signal); + return; + } + await readTaskRunSSEOnce(runId, 0, onEvent, signal, { replayOnly: true }); return; - } - await readTaskRunSSEOnce(runId, 0, onEvent, signal, { replayOnly: true }); - } catch (err) { - if (!signal.aborted) { - console.warn('task phase replay failed', runId, err); + } catch (err) { + if (signal.aborted) return; + const status = err instanceof TaskRunHttpError ? err.status : 0; + if (shouldAbandonTaskReplay(status, attempt, attempts)) { + console.warn('task phase replay failed', runId, err); + return; + } + await new Promise((resolve) => setTimeout(resolve, replayBackoffMs(attempt))); } } } + +/** 终态重放多试几次；进行中订阅内部已有重试，外层只再补一次。 */ +export function replayAttemptLimit(live: boolean): number { + return live ? 2 : 4; +} + +export function replayBackoffMs(attempt: number): number { + return Math.min(500 * (attempt + 1), 5000); +} + +export function shouldAbandonTaskReplay(status: number, attempt: number, attempts: number): boolean { + if (status === 401 || status === 404) return true; + return attempt >= attempts - 1; +} diff --git a/web/src/hooks/useTaskRunAttempts.ts b/web/src/hooks/useTaskRunAttempts.ts index 3367ec47..3b9ba574 100644 --- a/web/src/hooks/useTaskRunAttempts.ts +++ b/web/src/hooks/useTaskRunAttempts.ts @@ -9,6 +9,7 @@ import { fetchChatTurnStatus } from '../api/chatTurn'; import { mapTaskStatusToMessageStatus } from '../api/chatWindow'; import { listProjectWorkflowDocuments } from '../api/projectDocuments'; import { createTaskRun, getTaskRun, resumeTaskRun, subscribeTaskRunEvents } from '../api/taskRuns'; +import { readActiveTasks, removeActiveTask, touchActiveTaskPhase, upsertActiveTask } from '../lib/activeTaskCache'; import { buildTaskRequest, streamTask } from '../api/taskStream'; import { applyPhaseEvent } from '../engine/phaseEventMapper'; import { resolveTaskTemplateForMessage, wrongWorkflowGuidance } from '../engine/templateMatch'; @@ -371,6 +372,24 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { let accumulated = ''; let settled = false; + const rememberRun = (runId: string, label?: string, serverMessageId?: number) => { + if (!sessionId || !runId) return; + const previous = readActiveTasks(sessionId).find((row) => row.runId === runId); + upsertActiveTask({ + sessionId, + runId, + assistantClientId: asstId, + taskKey: task.taskKey || previous?.taskKey || '', + startedAt: previous?.startedAt || new Date().toISOString(), + lastPhaseLabel: label || previous?.lastPhaseLabel, + serverMessageId: serverMessageId || previous?.serverMessageId, + }); + }; + const forgetRun = (runId?: string) => { + const id = runId || retainedRunId; + if (id) removeActiveTask(id); + }; + const historyOnly = !messages.some((m) => m.id === asstId); let historySnapshot = historyOnly ? chatWindow.allMessages.find((m) => m.id === asstId) @@ -412,6 +431,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { scheduleRetry(delayMs); return; } + forgetRun(); patchAsst({ status: 'failed', errorInfo: info, livePhaseLabel: undefined }); taskRunStream.setThinking(sessionId, false); }; @@ -472,9 +492,11 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { if (controller.signal.aborted || settled) return; switch (ev.type) { case 'phase': { + const phaseLabel = ev.phase?.message; + if (retainedRunId) touchActiveTaskPhase(retainedRunId, phaseLabel); patchAsst((m) => ({ status: 'streaming_status', - livePhaseLabel: ev.phase?.message || m.livePhaseLabel, + livePhaseLabel: phaseLabel || m.livePhaseLabel, phaseEvents: applyPhaseEvent(opts?.phaseEvents || m.phaseEvents || [], ev.phase), })); break; @@ -542,6 +564,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { case 'done': { settled = true; const finishedRunId = retainedRunId || taskRunStream.activeTaskRunId; + forgetRun(finishedRunId || undefined); taskRunStream.setRunId(sessionId, null); taskRunStream.setThinking(sessionId, false); void (async () => { @@ -583,6 +606,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { const view = readRemoteTaskRun(remote); if (view.status === 'quality_failed') { settled = true; + forgetRun(runId); patchAsst({ status: 'quality_failed', livePhaseLabel: undefined, @@ -594,6 +618,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { } if (isBlockedGovernanceStatus(view.status)) { settled = true; + forgetRun(runId); patchAsst({ status: 'blocked_pending_governance', content: GOVERNANCE_BLOCKED_HINT, @@ -606,6 +631,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { } if (view.status === 'completed' || view.status === 'pending_author_confirmation') { settled = true; + forgetRun(runId); const mapped = mapTaskStatusToMessageStatus(view.status, 'assistant') || 'completed'; patchAsst({ status: mapped, @@ -624,6 +650,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { } if (view.status === 'interrupted') { settled = true; + forgetRun(runId); patchAsst({ status: 'failed', livePhaseLabel: undefined, @@ -693,6 +720,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { } const resumed = await resumeTaskRun(runId); retainedRunId = resumed.run_id || runId; + rememberRun(retainedRunId, '正在继续任务…'); taskRunStream.setRunId(sessionId, retainedRunId); taskRunStream.setThinking(sessionId, true); patchAsst({ @@ -720,6 +748,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { const view = readRemoteTaskRun(remote); if (remote && shouldSubscribeExistingRun(view.status)) { retainedRunId = existingRunId; + rememberRun(existingRunId, '正在重连任务…'); taskRunStream.setRunId(sessionId, existingRunId); patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); await subscribeWithGapHeal(existingRunId); @@ -738,6 +767,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { ? buildInterruptedTaskErrorInfo(view.errorMessage, view.workspaceId) : undefined, }); + forgetRun(existingRunId); taskRunStream.setThinking(sessionId, false); if (mapped === 'completed' || mapped === 'pending_author_confirmation') { await syncControlledDocsAfterTask(task.taskKey || 'intake'); @@ -760,6 +790,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { typeof created.assistant_msg_id === 'number' && created.assistant_msg_id > 0 ? created.assistant_msg_id : 0; + rememberRun(created.run_id, '任务执行中…', assistantMsgId > 0 ? assistantMsgId : undefined); if (assistantMsgId > 0) { chatWindow.replaceMessageServerId(asstId, assistantMsgId); patchAsst({ taskRunId: created.run_id, serverId: assistantMsgId }); @@ -774,6 +805,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { livePhaseLabel: undefined, taskRunId: created.run_id, }); + forgetRun(created.run_id); taskRunStream.setRunId(sessionId, null); taskRunStream.setThinking(sessionId, false); await chatWindow.refresh().catch(() => undefined); @@ -789,6 +821,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { void startObserver().catch(async (_err) => { if (settled) return; if (controller.signal.aborted) { + forgetRun(); patchAsst({ status: 'cancelled', livePhaseLabel: undefined }); taskRunStream.setThinking(sessionId, false); return; @@ -805,6 +838,7 @@ export function useTaskRunAttempts(input: TaskRunAttemptsInput) { // 原 run 仍在跑：重订阅，不 POST 新 run settled = false; retainedRunId = runId; + rememberRun(runId, '正在重连任务…'); patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); await subscribeWithGapHeal(runId); return; diff --git a/web/src/lib/activeTaskCache.test.ts b/web/src/lib/activeTaskCache.test.ts new file mode 100644 index 00000000..7a2a4f57 --- /dev/null +++ b/web/src/lib/activeTaskCache.test.ts @@ -0,0 +1,44 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + readActiveTasks, + removeActiveTask, + touchActiveTaskPhase, + upsertActiveTask, +} from './activeTaskCache'; + +const mem = new Map(); + +test('active task cache keeps an in-progress run and drops it when finished', () => { + (globalThis as { window?: unknown }).window = { + localStorage: { + getItem: (key: string) => mem.get(key) ?? null, + setItem: (key: string, value: string) => { + mem.set(key, value); + }, + }, + }; + mem.clear(); + upsertActiveTask({ + sessionId: 'sess-1', + runId: 'run-1', + assistantClientId: 'turn:assistant', + taskKey: 'research', + startedAt: '2026-09-28T03:27:00Z', + }); + upsertActiveTask({ + sessionId: 'sess-2', + runId: 'run-2', + assistantClientId: 'other', + taskKey: 'intake', + startedAt: '2026-09-28T03:28:00Z', + }); + touchActiveTaskPhase('run-1', '正在撰写调研报告'); + const mine = readActiveTasks('sess-1'); + assert.equal(mine.length, 1); + assert.equal(mine[0].runId, 'run-1'); + assert.equal(mine[0].lastPhaseLabel, '正在撰写调研报告'); + removeActiveTask('run-1'); + assert.equal(readActiveTasks('sess-1').length, 0); + assert.equal(readActiveTasks('sess-2').length, 1); +}); diff --git a/web/src/lib/activeTaskCache.ts b/web/src/lib/activeTaskCache.ts new file mode 100644 index 00000000..57cfd765 --- /dev/null +++ b/web/src/lib/activeTaskCache.ts @@ -0,0 +1,95 @@ +import type { ChatMessage } from '../types/workbench'; + +/** 进行中任务锚点。窗口缓存只存服务端已落库消息，刷新会丢掉还没写回的进度。 */ +const STORAGE_KEY = 'smdf_agent_workbench_v6_active_tasks'; + +export type ActiveTaskCacheEntry = { + sessionId: string; + runId: string; + assistantClientId: string; + taskKey: string; + startedAt: string; + lastPhaseLabel?: string; + serverMessageId?: number; +}; + +function readAll(): ActiveTaskCacheEntry[] { + if (typeof window === 'undefined' || !window.localStorage) return []; + try { + const raw = window.localStorage.getItem(STORAGE_KEY); + if (!raw) return []; + const parsed = JSON.parse(raw) as unknown; + if (!Array.isArray(parsed)) return []; + return parsed.filter(isEntry); + } catch { + return []; + } +} + +function writeAll(entries: ActiveTaskCacheEntry[]): void { + if (typeof window === 'undefined' || !window.localStorage) return; + try { + window.localStorage.setItem(STORAGE_KEY, JSON.stringify(entries)); + } catch { + // 配额不足时放弃缓存，服务端锚点仍可恢复。 + } +} + +function isEntry(value: unknown): value is ActiveTaskCacheEntry { + if (!value || typeof value !== 'object') return false; + const row = value as Partial; + return typeof row.sessionId === 'string' && typeof row.runId === 'string' && row.runId !== ''; +} + +export function readActiveTasks(sessionId: string): ActiveTaskCacheEntry[] { + if (!sessionId) return []; + return readAll().filter((entry) => entry.sessionId === sessionId); +} + +export function upsertActiveTask(entry: ActiveTaskCacheEntry): void { + if (!entry.sessionId || !entry.runId) return; + const rest = readAll().filter((row) => row.runId !== entry.runId); + writeAll([...rest, entry]); +} + +export function touchActiveTaskPhase(runId: string, label: string | undefined): void { + if (!runId || !label) return; + const all = readAll(); + const hit = all.find((row) => row.runId === runId); + if (!hit || hit.lastPhaseLabel === label) return; + hit.lastPhaseLabel = label; + writeAll(all); +} + +export function removeActiveTask(runId: string): void { + if (!runId) return; + writeAll(readAll().filter((row) => row.runId !== runId)); +} + +export function placeholderFromActiveTask(entry: ActiveTaskCacheEntry): ChatMessage { + return { + id: entry.assistantClientId || `ast-active-${entry.runId}`, + clientMessageId: entry.assistantClientId || undefined, + serverId: entry.serverMessageId, + sessionId: entry.sessionId, + role: 'assistant', + authorName: '项目助手', + content: '任务执行中…', + timestamp: new Date().toTimeString().substring(0, 5), + status: 'streaming_status', + livePhaseLabel: entry.lastPhaseLabel || '任务执行中…', + taskRunId: entry.runId, + taskRequest: entry.taskKey + ? { + taskKey: entry.taskKey, + projectName: '', + customerName: '', + deviceType: '', + sessionId: entry.sessionId, + userId: 0, + projectId: 0, + clientTurnId: '', + } + : undefined, + }; +}