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

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<string, import('node:child_process').ChildProcess>} */
+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);

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

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 {

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

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)

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

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

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

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

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