From 63cc8ab1da6dc01a495f034a22187d2f5ab7b06e Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Fri, 11 Sep 2026 15:15:15 +0800 Subject: [PATCH] feat(api): observer task runs decoupled from SSE lifecycle Add POST/GET/POST cancel task run APIs, background worker with event replay, workspace reconcile, runner child cancel, and chat message metadata updates. Co-authored-by: Cursor --- deploy/pi-agent/runner.mjs | 26 +- internal/ctxmgr/message_update.go | 52 ++++ internal/server/handlers.go | 35 +++ internal/server/reconcile_handlers.go | 24 ++ internal/server/server.go | 12 + internal/server/task_run_handlers.go | 291 ++++++++++++++++++++++ internal/server/task_run_observer_test.go | 95 +++++++ internal/squad/reconcile.go | 90 +++++++ internal/taskrun/manager.go | 272 ++++++++++++++++++++ internal/taskrun/types.go | 58 +++++ 10 files changed, 954 insertions(+), 1 deletion(-) create mode 100644 internal/ctxmgr/message_update.go create mode 100644 internal/server/reconcile_handlers.go create mode 100644 internal/server/task_run_handlers.go create mode 100644 internal/server/task_run_observer_test.go create mode 100644 internal/squad/reconcile.go create mode 100644 internal/taskrun/manager.go create mode 100644 internal/taskrun/types.go diff --git a/deploy/pi-agent/runner.mjs b/deploy/pi-agent/runner.mjs index 08265feb..ba22f0ea 100755 --- a/deploy/pi-agent/runner.mjs +++ b/deploy/pi-agent/runner.mjs @@ -16,6 +16,8 @@ const DATA_DIR = process.env.DATA_DIR || "/app/data/workspaces"; const DEFAULT_MODEL = process.env.MINIMAX_MODEL_ID || "MiniMax-M3"; const DEFAULT_PROVIDER = process.env.MINIMAX_PROVIDER || "minimax"; const tavilyPool = TavilyKeyPool.fromEnv(); +/** @type {Map} */ +const activeChildren = new Map(); function safeRel(workspace, target) { const abs = isAbsolute(target) ? target : resolve(workspace, target); @@ -153,6 +155,7 @@ async function handleRun(body) { env, stdio: ["pipe", "pipe", "pipe"], }); + activeChildren.set(workspace, child); const timer = setTimeout(() => { lastErr = `Execution timed out after ${timeoutMs}ms`; @@ -204,10 +207,12 @@ async function handleRun(body) { await new Promise((resolveDone) => { child.on("close", () => { clearTimeout(timer); + activeChildren.delete(workspace); resolveDone(); }); child.on("error", (err) => { clearTimeout(timer); + activeChildren.delete(workspace); lastErr = err && err.message ? err.message : String(err); resolveDone(); }); @@ -314,7 +319,26 @@ const server = http.createServer(async (req, res) => { return; } - // 3. Agent Task Run Endpoint + // 3. Cancel active child process by workspace key + if (method === "POST" && url.pathname === "/v1/cancel") { + try { + const body = await parseJsonBody(req); + const workspace = resolveWorkspace(body.workspace || ""); + const child = activeChildren.get(workspace); + if (child) { + child.kill("SIGKILL"); + activeChildren.delete(workspace); + sendJson(res, 200, { ok: true, cancelled: true, workspace }); + } else { + sendJson(res, 200, { ok: true, cancelled: false, workspace }); + } + } catch (err) { + sendJson(res, 500, { ok: false, error: err.message }); + } + return; + } + + // 4. Agent Task Run Endpoint if (method === "POST" && url.pathname === "/v1/run") { try { const body = await parseJsonBody(req); diff --git a/internal/ctxmgr/message_update.go b/internal/ctxmgr/message_update.go new file mode 100644 index 00000000..2adb4bd9 --- /dev/null +++ b/internal/ctxmgr/message_update.go @@ -0,0 +1,52 @@ +package ctxmgr + +import ( + "context" + "encoding/json" + "errors" + "fmt" +) + +// UpdateMessageMetadata 更新已持久化助手消息的 metadata 与任务状态。 +func (m *ContextManager) UpdateMessageMetadata(ctx context.Context, messageID int64, 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, _ := json.Marshal(meta) + _, err := m.pool.Exec(ctx, ` + UPDATE chat_messages + SET metadata = COALESCE(metadata, '{}'::jsonb) || $1::jsonb + WHERE id = $2 + `, metaJSON, messageID) + if err != nil { + return fmt.Errorf("update message metadata failed: %w", err) + } + return nil + } + + m.mu.Lock() + defer m.mu.Unlock() + for sid, msgs := range m.messages { + for i := range msgs { + if msgs[i].ID == messageID { + if msgs[i].Metadata == nil { + msgs[i].Metadata = map[string]any{} + } + for k, v := range meta { + msgs[i].Metadata[k] = v + } + m.messages[sid][i] = msgs[i] + return nil + } + } + } + return errors.New("message not found") +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 8bcbea49..0a2c35f3 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -65,6 +65,8 @@ type TaskRunRequestDTO struct { ForceDirectEdit bool `json:"force_direct_edit,omitempty"` CustomSteps []squad.StepSpec `json:"custom_steps,omitempty"` WorkflowKey string `json:"workflow_key,omitempty"` + SessionID string `json:"session_id,omitempty"` + ClientTurnID string `json:"client_turn_id,omitempty"` // Materials 用户上传的参考文档材料，由前端 Base64 编码后传入；后端 json 反序列化自动解码 []byte Materials []bizdev.MaterialInput `json:"materials,omitempty"` } @@ -742,6 +744,39 @@ func (s *Server) handleListChatSessions(w http.ResponseWriter, r *http.Request) writeJSONResponse(w, http.StatusOK, sessions) } +type upsertChatSessionRequest struct { + SessionID string `json:"session_id"` + Title string `json:"title"` + UserID int64 `json:"user_id"` + ProjectID int64 `json:"project_id"` +} + +// handleUpsertChatSession 注册或更新会话元数据（立项/成员加入时同步）。 +func (s *Server) handleUpsertChatSession(w http.ResponseWriter, r *http.Request) { + if s.svc == nil || s.svc.ContextManager() == nil { + http.Error(w, "context manager not initialized", http.StatusServiceUnavailable) + return + } + + var req upsertChatSessionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + if req.SessionID == "" { + http.Error(w, "missing session_id", http.StatusBadRequest) + return + } + + summary, err := s.svc.ContextManager().EnsureChatSession(r.Context(), req.SessionID, req.UserID, req.ProjectID, req.Title) + if err != nil { + http.Error(w, "upsert chat session failed: "+err.Error(), http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusOK, summary) +} + // handleGetChatMessages 分页获取指定会话的历史多轮消息记录 func (s *Server) handleGetChatMessages(w http.ResponseWriter, r *http.Request) { if s.svc == nil || s.svc.ContextManager() == nil { diff --git a/internal/server/reconcile_handlers.go b/internal/server/reconcile_handlers.go new file mode 100644 index 00000000..45a93843 --- /dev/null +++ b/internal/server/reconcile_handlers.go @@ -0,0 +1,24 @@ +package server + +import ( + "encoding/json" + "net/http" + + "chat-agent-system/internal/squad" +) + +func (s *Server) handleReconcileWorkspace(w http.ResponseWriter, r *http.Request) { + var body struct { + WorkDir string `json:"work_dir"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + report, err := squad.ReconcileWorkspaceStatus(body.WorkDir) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSONResponse(w, http.StatusOK, report) +} diff --git a/internal/server/server.go b/internal/server/server.go index 3aad90fa..6ed521ac 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -19,6 +19,7 @@ import ( "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/taskrun" ) // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 @@ -31,6 +32,7 @@ type Server struct { authSvc *auth.Service sessions map[string]*chat.Session sessionsMu sync.RWMutex + taskRuns *taskrun.Manager httpServer *http.Server } @@ -61,7 +63,9 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { engine: coordEngine, authSvc: authSvc, sessions: make(map[string]*chat.Session), + taskRuns: taskrun.NewManager(pwSvc, svc.ContextManager()), } + s.initTaskRunExecutors() return s } @@ -111,6 +115,13 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask) 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/{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) + mux.HandleFunc("POST /api/v1/admin/workspaces/reconcile", s.handleReconcileWorkspace) + mux.HandleFunc("POST /api/v1/projects/{id}/assets", s.handleUploadProjectAsset) + mux.HandleFunc("GET /api/v1/projects/{id}/assets", s.handleListProjectAssets) // 6. 多级上下文与偏好记忆管理 mux.HandleFunc("GET /api/v1/context/user", s.handleGetUserContext) @@ -119,6 +130,7 @@ func (s *Server) Routes() http.Handler { // 7. 历史多轮会话管理与标杆案例语义检索 mux.HandleFunc("GET /api/v1/chat/sessions", s.handleListChatSessions) + mux.HandleFunc("POST /api/v1/chat/sessions", s.handleUpsertChatSession) mux.HandleFunc("GET /api/v1/chat/messages", s.handleGetChatMessages) mux.HandleFunc("GET /api/v1/chat/messages/window", s.handleGetChatWindow) mux.HandleFunc("GET /api/v1/chat/messages/day", s.handleGetChatDayMessages) diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go new file mode 100644 index 00000000..bc740269 --- /dev/null +++ b/internal/server/task_run_handlers.go @@ -0,0 +1,291 @@ +package server + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/squad" + "chat-agent-system/internal/stream" + "chat-agent-system/internal/taskrun" +) + +type createTaskRunRequest struct { + TaskRunRequestDTO + SessionID string `json:"session_id"` + ClientTurnID string `json:"client_turn_id"` + UserMessage string `json:"user_message"` + WorkDir string `json:"work_dir,omitempty"` +} + +type createTaskRunResponse struct { + RunID string `json:"run_id"` + WorkspaceID string `json:"workspace_id,omitempty"` + Status string `json:"status"` +} + +func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { + var req createTaskRunRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest) + return + } + if req.UserID <= 0 { + req.UserID = parseUserIDHeader(r) + } + if err := bizdev.ValidateTaskRunIdentity(req.SessionID, req.ProjectID); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if req.TaskKey == "" && req.Stage == "" && req.WorkDir == "" { + req.TaskKey = "intake" + } + + ctxMgr := s.svc.ContextManager() + if ctxMgr != nil { + _, _ = ctxMgr.EnsureChatSession(r.Context(), req.SessionID, req.UserID, req.ProjectID, req.ProjectName) + } + + var assistantMsgID int64 + if ctxMgr != nil && req.UserMessage != "" { + userMeta := map[string]any{ + "project_id": req.ProjectID, + "client_turn_id": req.ClientTurnID, + "task_key": req.TaskKey, + } + _, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, req.UserID, "user", req.UserMessage, "TRIGGER_TASK", "", userMeta) + asstMeta := map[string]any{ + "project_id": req.ProjectID, + "client_turn_id": req.ClientTurnID, + "task_key": req.TaskKey, + "task_status": "running", + } + assistantMsgID, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", asstMeta) + } + + runID := fmt.Sprintf("run_%d", time.Now().UnixNano()) + record := &taskrun.RunRecord{ + RunID: runID, + SessionID: req.SessionID, + ProjectID: req.ProjectID, + UserID: req.UserID, + ClientTurnID: req.ClientTurnID, + TaskKey: req.TaskKey, + Stage: req.Stage, + AssistantMsgID: assistantMsgID, + Status: "running", + } + + taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID) + resumeDir := req.WorkDir + + stageReq := s.buildBizdevStageRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID) + created, err := s.taskRuns.CreateAndStart(r.Context(), runID, record, taskReq, stageReq, resumeDir) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusAccepted, createTaskRunResponse{ + RunID: created.RunID, + Status: created.Status, + }) +} + +func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) { + runID := r.PathValue("run_id") + if runID == "" { + http.Error(w, "run_id required", http.StatusBadRequest) + return + } + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + + afterSeq := int64(0) + if raw := r.URL.Query().Get("after_seq"); raw != "" { + afterSeq, _ = strconv.ParseInt(raw, 10, 64) + } + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("X-Accel-Buffering", "no") + flusher.Flush() + + ch, replay, err := s.taskRuns.Subscribe(runID, afterSeq) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + defer s.taskRuns.Unsubscribe(runID, ch) + + writeStoredEvents := func(events []taskrun.StoredEvent) { + for _, stored := range events { + data, _ := json.Marshal(stored.Event) + fmt.Fprintf(w, "id: %d\n", stored.Seq) + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data)) + } + flusher.Flush() + } + + writeStoredEvents(replay) + + for { + select { + case <-r.Context().Done(): + return + case stored, ok := <-ch: + if !ok { + return + } + writeStoredEvents([]taskrun.StoredEvent{stored}) + if stored.Event.Type == stream.EventDone || stored.Event.Type == stream.EventError { + return + } + } + } +} + +func (s *Server) handleCancelTaskRun(w http.ResponseWriter, r *http.Request) { + runID := r.PathValue("run_id") + if runID == "" { + http.Error(w, "run_id required", http.StatusBadRequest) + return + } + if err := s.taskRuns.Cancel(runID); err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + if rec, getErr := s.taskRuns.GetRun(runID); getErr == nil && rec.WorkspaceID != "" { + s.cancelPiAgentRunner(rec.WorkspaceID) + } + writeJSONResponse(w, http.StatusOK, map[string]string{"run_id": runID, "status": "cancelled"}) +} + +func (s *Server) cancelPiAgentRunner(workspaceID string) { + runnerURL := strings.TrimRight(strings.TrimSpace(s.cfg.PiAgentRunnerURL), "/") + if runnerURL == "" || workspaceID == "" { + return + } + body, _ := json.Marshal(map[string]string{"workspace": workspaceID}) + req, err := http.NewRequest(http.MethodPost, runnerURL+"/v1/cancel", bytes.NewReader(body)) + if err != nil { + return + } + req.Header.Set("Content-Type", "application/json") + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Do(req) + if err != nil { + return + } + _ = resp.Body.Close() +} + +func (s *Server) handleGetTaskRun(w http.ResponseWriter, r *http.Request) { + runID := r.PathValue("run_id") + rec, err := s.taskRuns.GetRun(runID) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + writeJSONResponse(w, http.StatusOK, rec) +} + +func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, clientTurnID, runID string) bizdev.TaskRunRequest { + if dto.TaskKey == "" { + dto.TaskKey = "intake" + } + return bizdev.TaskRunRequest{ + TaskKey: dto.TaskKey, + ProjectName: dto.ProjectName, + CustomerName: dto.CustomerName, + DeviceType: dto.DeviceType, + IntendedUse: dto.IntendedUse, + TargetMarket: dto.TargetMarket, + Fields: dto.Fields, + Materials: dto.Materials, + Backend: dto.Backend, + AutoRepair: dto.AutoRepair, + SessionID: sessionID, + UserID: dto.UserID, + ProjectID: dto.ProjectID, + ClientTurnID: clientTurnID, + RunID: runID, + DocID: dto.DocID, + AuthorID: dto.AuthorID, + AuthorName: dto.AuthorName, + ReviewerID: dto.ReviewerID, + ReviewerName: dto.ReviewerName, + ApproverID: dto.ApproverID, + ApproverName: dto.ApproverName, + ForceDirectEdit: dto.ForceDirectEdit, + CustomSteps: dto.CustomSteps, + } +} + +func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clientTurnID, runID string) bizdev.StageRunRequest { + return bizdev.StageRunRequest{ + Stage: bizdev.MaturityStage(dto.Stage), + ProjectName: dto.ProjectName, + CustomerName: dto.CustomerName, + DeviceType: dto.DeviceType, + IntendedUse: dto.IntendedUse, + TargetMarket: dto.TargetMarket, + Fields: dto.Fields, + Materials: dto.Materials, + Backend: dto.Backend, + AutoRepair: dto.AutoRepair, + SessionID: sessionID, + UserID: dto.UserID, + ProjectID: dto.ProjectID, + ClientTurnID: clientTurnID, + RunID: runID, + DocID: dto.DocID, + AuthorID: dto.AuthorID, + AuthorName: dto.AuthorName, + ReviewerID: dto.ReviewerID, + ReviewerName: dto.ReviewerName, + ApproverID: dto.ApproverID, + ApproverName: dto.ApproverName, + ForceDirectEdit: dto.ForceDirectEdit, + } +} + +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, workDir string, _ *stream.Streamer) (string, error) { + res, err := s.svc.ResumeOrModifyWorkspace(ctx, squad.ModifyWorkflowRequest{WorkDir: workDir}) + ws := workDir + if res != nil && res.WorkDir != "" { + ws = res.WorkDir + } + return ws, 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 + } + return ws, err + }, + ) +} diff --git a/internal/server/task_run_observer_test.go b/internal/server/task_run_observer_test.go new file mode 100644 index 00000000..69d590a3 --- /dev/null +++ b/internal/server/task_run_observer_test.go @@ -0,0 +1,95 @@ +package server + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/stream" + "chat-agent-system/internal/taskrun" +) + +func TestTaskRun_DisconnectDoesNotCancelWorker(t *testing.T) { + cfg := &config.Config{ + Env: "test", + AgentBackend: "stub", + DataDir: t.TempDir(), + } + svc := bizdev.NewService(cfg) + srv := NewServer(cfg, svc) + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + runID := "run_disconnect_test" + record := &taskrun.RunRecord{RunID: runID, TaskKey: "intake", Status: "running"} + taskReq := bizdev.TaskRunRequest{TaskKey: "intake", ProjectName: "disconnect-test"} + _, err := srv.taskRuns.CreateAndStart(context.Background(), runID, record, taskReq, bizdev.StageRunRequest{}, "") + if err != nil { + t.Fatalf("CreateAndStart: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/api/v1/tasks/runs/"+runID+"/events", nil) + req.Header.Set("Accept", "text/event-stream") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("subscribe failed: %v", err) + } + time.Sleep(200 * time.Millisecond) + cancel() + _ = resp.Body.Close() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + rec, getErr := srv.taskRuns.GetRun(runID) + if getErr == nil && rec.Status != "running" { + if rec.Status == "failed" { + t.Fatalf("worker marked failed after disconnect, status=%s err=%s", rec.Status, rec.ErrorMessage) + } + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatal("worker did not finish after disconnect") +} + +func TestCreateTaskRun_RequiresSessionProject(t *testing.T) { + cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"} + srv := NewServer(cfg, bizdev.NewService(cfg)) + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + body, _ := json.Marshal(map[string]any{"task_key": "intake"}) + resp, err := http.Post(ts.URL+"/api/v1/tasks/runs", "application/json", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status=%d want 400", resp.StatusCode) + } +} + +func readFirstEventType(t *testing.T, resp *http.Response) stream.EventType { + t.Helper() + scanner := bufio.NewScanner(resp.Body) + var currentType string + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "event: ") { + currentType = strings.TrimPrefix(line, "event: ") + } else if strings.HasPrefix(line, "data: ") && currentType != "" { + return stream.EventType(currentType) + } + } + return "" +} diff --git a/internal/squad/reconcile.go b/internal/squad/reconcile.go new file mode 100644 index 00000000..8f302787 --- /dev/null +++ b/internal/squad/reconcile.go @@ -0,0 +1,90 @@ +package squad + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// ReconcileReport 工作区状态对账结果。 +type ReconcileReport struct { + WorkDir string `json:"work_dir"` + PreviousStatus string `json:"previous_status"` + NewStatus string `json:"new_status"` + ExecutionOK bool `json:"execution_ok"` + HasOutput bool `json:"has_output"` + Reconciled bool `json:"reconciled"` + Reason string `json:"reason,omitempty"` +} + +// ReconcileWorkspaceStatus 检测外层 failed 但 execution.log 为 SUCCESS 的孤儿任务并修复。 +func ReconcileWorkspaceStatus(workDir string) (*ReconcileReport, error) { + workDir = strings.TrimSpace(workDir) + if workDir == "" { + return nil, fmt.Errorf("work_dir required") + } + metaPath := filepath.Join(workDir, "workspace.json") + metaBytes, err := os.ReadFile(metaPath) + if err != nil { + return nil, err + } + var meta WorkspaceMetadata + if err := json.Unmarshal(metaBytes, &meta); err != nil { + return nil, err + } + + report := &ReconcileReport{ + WorkDir: workDir, + PreviousStatus: meta.Status, + NewStatus: meta.Status, + } + + execOK := scanExecutionSuccess(workDir) + report.ExecutionOK = execOK + report.HasOutput = fileExists(filepath.Join(workDir, "tasks", "01_organize", "output.md")) || + fileExists(filepath.Join(workDir, "output.md")) + + if meta.Status != "failed" || !execOK || !report.HasOutput { + report.Reason = "no reconciliation needed" + return report, nil + } + + meta.Status = "completed" + now := time.Now() + meta.UpdatedAt = now + meta.CompletedAt = &now + if err := writeJSON(metaPath, meta); err != nil { + return nil, err + } + report.NewStatus = "completed" + report.Reconciled = true + report.Reason = "execution log SUCCESS with output present while workspace was failed" + return report, nil +} + +func scanExecutionSuccess(workDir string) bool { + var found bool + _ = filepath.Walk(filepath.Join(workDir, "tasks"), func(path string, info os.FileInfo, err error) error { + if err != nil || info.IsDir() || info.Name() != "execution.log" { + return nil + } + b, readErr := os.ReadFile(path) + if readErr != nil { + return nil + } + if strings.Contains(string(b), "Outcome: SUCCESS") { + found = true + return filepath.SkipAll + } + return nil + }) + return found +} + +func fileExists(path string) bool { + fi, err := os.Stat(path) + return err == nil && !fi.IsDir() +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go new file mode 100644 index 00000000..6f963223 --- /dev/null +++ b/internal/taskrun/manager.go @@ -0,0 +1,272 @@ +package taskrun + +import ( + "context" + "fmt" + "sync" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/stream" +) + +// Executor 执行具体任务逻辑（由 server 注入）。 +type Executor func(ctx context.Context, runID string, req bizdev.TaskRunRequest, streamer *stream.Streamer) (workspaceID string, err error) + +// ResumeExecutor 续跑既有工作区。 +type ResumeExecutor func(ctx context.Context, runID string, workDir string, streamer *stream.Streamer) (workspaceID string, err error) + +// StageExecutor 执行成熟度阶段任务序列。 +type StageExecutor func(ctx context.Context, runID string, req bizdev.StageRunRequest, streamer *stream.Streamer) (workspaceID string, err error) + +// Manager 后台任务运行观察者：SSE 断线不取消 worker。 +type Manager struct { + mu sync.RWMutex + runs map[string]*activeRun + exec Executor + resume ResumeExecutor + stage StageExecutor + pwSvc *projectworkflow.Service + ctxMgr *ctxmgr.ContextManager +} + +func NewManager(pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *Manager { + return &Manager{ + runs: make(map[string]*activeRun), + pwSvc: pwSvc, + ctxMgr: ctxMgr, + } +} + +func (m *Manager) SetExecutor(exec Executor, resume ResumeExecutor, stage StageExecutor) { + m.exec = exec + m.resume = resume + m.stage = stage +} + +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()) + } + if record == nil { + record = &RunRecord{RunID: runID} + } + record.RunID = runID + if record.Status == "" { + record.Status = "running" + } + now := time.Now() + record.CreatedAt = now + record.UpdatedAt = now + + workerCtx, cancel := context.WithCancel(context.WithoutCancel(parent)) + + ar := &activeRun{ + record: record, + subs: make(map[int]chan StoredEvent), + cancel: cancel, + done: make(chan struct{}), + } + + m.mu.Lock() + m.runs[runID] = ar + m.mu.Unlock() + + if m.pwSvc != nil && record.ProjectID > 0 { + startedBy := record.UserID + _, _ = m.pwSvc.CreateWorkflowRun(workerCtx, &projectworkflow.WorkflowRun{ + RunID: runID, + ProjectID: record.ProjectID, + Status: "running", + StartedBy: &startedBy, + StageKey: record.Stage, + TaskKey: record.TaskKey, + InputManifest: map[string]any{ + "session_id": record.SessionID, + "client_turn_id": record.ClientTurnID, + "assistant_msg_id": record.AssistantMsgID, + }, + }) + } + + go m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir) + return record, nil +} + +func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) { + defer close(ar.done) + taskReq.RunID = ar.record.RunID + taskReq.SessionID = ar.record.SessionID + taskReq.ProjectID = ar.record.ProjectID + taskReq.UserID = ar.record.UserID + taskReq.ClientTurnID = ar.record.ClientTurnID + + lastPhaseID := "" + listener := func(ev stream.StreamEvent) { + if ev.Type == stream.EventPhase && ev.Phase != nil && ev.Phase.PhaseID != "" { + lastPhaseID = ev.Phase.PhaseID + } + m.appendEvent(ar, ev) + } + ctxWithS, streamer := stream.WithListener(ctx, false, listener) + + var workspaceID string + var execErr error + start := time.Now() + + if ar.record.Stage != "" && m.stage != nil { + stageReq.RunID = ar.record.RunID + stageReq.SessionID = ar.record.SessionID + 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) + } else if resumeWorkDir != "" && m.resume != nil { + workspaceID, execErr = m.resume(ctxWithS, ar.record.RunID, resumeWorkDir, streamer) + } else if m.exec != nil { + workspaceID, execErr = m.exec(ctxWithS, ar.record.RunID, taskReq, streamer) + } else { + execErr = fmt.Errorf("task executor not configured") + } + + if workspaceID != "" { + ar.mu.Lock() + ar.record.WorkspaceID = workspaceID + ar.mu.Unlock() + } + + if execErr != nil { + if ctx.Err() == context.Canceled { + m.finishRun(ar, "cancelled", workspaceID, execErr.Error()) + 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 + } + classifyWS := workspaceID + if classifyWS == "" && resumeWorkDir != "" { + classifyWS = resumeWorkDir + } + 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()) + return + } + + m.appendEvent(ar, stream.StreamEvent{ + Type: stream.EventDone, + Done: &stream.DonePayload{ + SessionID: "task_run", + WorkspaceID: workspaceID, + DurationMs: time.Since(start).Milliseconds(), + }, + }) + m.finishRun(ar, "completed", workspaceID, "") +} + +func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string) { + ar.mu.Lock() + ar.record.Status = status + ar.record.WorkspaceID = workspaceID + ar.record.ErrorMessage = errMsg + 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, + } + _ = m.ctxMgr.UpdateMessageMetadata(context.Background(), ar.record.AssistantMsgID, meta, status) + } +} + +func (m *Manager) appendEvent(ar *activeRun, ev stream.StreamEvent) { + ar.mu.Lock() + ar.nextSeq++ + stored := StoredEvent{Seq: ar.nextSeq, Event: ev} + ar.events = append(ar.events, stored) + subs := make([]chan StoredEvent, 0, len(ar.subs)) + for _, ch := range ar.subs { + subs = append(subs, ch) + } + ar.mu.Unlock() + + for _, ch := range subs { + select { + case ch <- stored: + default: + } + } +} + +func (m *Manager) GetRun(runID string) (*RunRecord, error) { + m.mu.RLock() + ar, ok := m.runs[runID] + m.mu.RUnlock() + if !ok { + return nil, fmt.Errorf("run not found") + } + ar.mu.RLock() + defer ar.mu.RUnlock() + cpy := *ar.record + return &cpy, nil +} + +func (m *Manager) Subscribe(runID string, afterSeq int64) (<-chan StoredEvent, []StoredEvent, error) { + m.mu.RLock() + ar, ok := m.runs[runID] + m.mu.RUnlock() + if !ok { + return nil, nil, fmt.Errorf("run not found") + } + + ar.mu.Lock() + defer ar.mu.Unlock() + replay := make([]StoredEvent, 0) + for _, ev := range ar.events { + if ev.Seq > afterSeq { + replay = append(replay, ev) + } + } + ch := make(chan StoredEvent, 32) + ar.subSeq++ + id := ar.subSeq + ar.subs[id] = ch + return ch, replay, nil +} + +func (m *Manager) Unsubscribe(runID string, ch <-chan StoredEvent) { + m.mu.RLock() + ar, ok := m.runs[runID] + m.mu.RUnlock() + if !ok { + return + } + ar.mu.Lock() + defer ar.mu.Unlock() + for id, sub := range ar.subs { + if sub == ch { + delete(ar.subs, id) + close(sub) + break + } + } +} + +func (m *Manager) Cancel(runID string) error { + m.mu.RLock() + ar, ok := m.runs[runID] + m.mu.RUnlock() + if !ok { + return fmt.Errorf("run not found") + } + ar.cancel() + return nil +} diff --git a/internal/taskrun/types.go b/internal/taskrun/types.go new file mode 100644 index 00000000..b60b5c21 --- /dev/null +++ b/internal/taskrun/types.go @@ -0,0 +1,58 @@ +package taskrun + +import ( + "context" + "sync" + "time" + + "chat-agent-system/internal/stream" +) + +// CreateRequest 创建后台任务运行实例。 +type CreateRequest struct { + SessionID string + ProjectID int64 + UserID int64 + ClientTurnID string + TaskKey string + Stage string + WorkDir string + UserMessage string + AssistantMsgID int64 + IncludeDetails bool + TaskReq any // *server task payload converted to bizdev request at execution layer +} + +// RunRecord 运行实例状态。 +type RunRecord struct { + RunID string + SessionID string + ProjectID int64 + UserID int64 + ClientTurnID string + TaskKey string + Stage string + Status string // pending | running | completed | failed | cancelled | status_conflict + WorkspaceID string + AssistantMsgID int64 + CreatedAt time.Time + UpdatedAt time.Time + ErrorMessage string +} + +// StoredEvent 持久化 SSE 事件（支持重连重放）。 +type StoredEvent struct { + Seq int64 + Event stream.StreamEvent +} + +type activeRun struct { + record *RunRecord + events []StoredEvent + nextSeq int64 + subs map[int]chan StoredEvent + subSeq int + cancel context.CancelFunc + done chan struct{} + mu sync.RWMutex +}