feat(taskrun): persist runs/events and fix finalize-before-done ordering

Add goose v19 task_runs/task_run_events tables with PG store for restart
replay, orphan interruption on startup, strict /ready checks, and strip
client work_dir from chat/SSE paths per P0 path sandbox requirements.

Co-authored-by: Cursor <cursoragent@cursor.com>

.doc/test-reports/2026-09-12-phase4-task-persistence-readiness-test-report.md

@@ -0,0 +1,91 @@
+# Phase 4 长任务持久化与 Readiness 测试报告
+
+## 一、 测试背景 (Background)
+
+按整改方案建议顺序推进 Phase 4（RUN-001/002）与 P0 剩余项：长任务 run/event 此前仅存进程内存，重启不可重放；`done` 事件先于 Finalize 导致假成功风险；chat/SSE 仍接受客户端 `work_dir`；readiness 未严格校验 schema 与 task store。
+
+## 二、 测试目标 (Test Objectives)
+
+1. `task_runs` / `task_run_events` PG 持久化，重启后可查询与 SSE 重放
+2. Finalize 成功后才发送 `done`；Finalize 失败不发 `done`、状态为 `finalizing_failed`
+3. 进程重启时将孤儿 `running` 任务标记为 `interrupted`
+4. `/api/v1/ready` 在 PG 环境下校验 `task_store_ready`
+5. chat/SSE 忽略客户端 `work_dir`（P0-03 补充）
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+- OS: Linux, Go 1.22+
+- PG: `postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev`
+- 迁移: goose v19 (`00019_2026-09-12-task-runs-events.sql`)
+- 层级: 单元测试 + PG 集成测试
+
+## 四、 预期结果 (Expected Results)
+
+| 用例 | 预期 |
+|------|------|
+| 任务完成持久化 | `task_runs.status=completed`，events 含 `done` |
+| 重启重放 | 新 Manager + Store 可 GetRun/Subscribe replay |
+| Finalize 时序 | 最后事件为 `done`；失败时无 `done` |
+| Readiness | PG 环境 `/api/v1/ready` 200 且 `task_store_ready=true` |
+| P0 匿名 API | 仍返回 401（既有 security_p0 测试） |
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+### 5.1 Finalize 时序（单元）
+
+**输入**: `go test -run TestManager_DoneEmittedAfterFinalize ./internal/taskrun/...`
+
+**输出**:
+```
+ok   chat-agent-system/internal/taskrun   0.116s
+```
+
+**判定**: PASS — 最后事件类型为 `done`
+
+### 5.2 Finalize 失败不发 done（单元）
+
+**输入**: `go test -run TestManager_FinalizeFailureDoesNotEmitDone ./internal/taskrun/...`
+
+**判定**: PASS — 状态 `finalizing_failed`，无 `done` 事件
+
+### 5.3 PG 持久化与重启重放
+
+**输入**: `go test -run TestTaskRun_PersistenceAndReplayAfterRestart_PG ./internal/server/...`
+
+**实际输出（摘要）**:
+```
+goose: successfully migrated database to version: 19
+taskrun_status_transition run_id=run_pg_persist_* status=completed
+ok   chat-agent-system/internal/server   0.989s
+```
+
+**判定**: PASS — Store GetRun/ListEvents/Subscribe replay 均有效
+
+### 5.4 Readiness 严格门禁
+
+**输入**: `GET /api/v1/ready`（PG 已迁移）
+
+**判定**: PASS — 200, `task_store_ready=true`
+
+### 5.5 P0 安全回归
+
+**输入**: `go test -run TestP0_ ./internal/server/...`
+
+**判定**: PASS
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+```
+go test ./internal/taskrun/... ./internal/server/... -count=1 -timeout 180s
+ok   chat-agent-system/internal/taskrun
+ok   chat-agent-system/internal/server
+```
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+**结论**: Phase 4 核心 RUN-001/002 与 P0 readiness/work_dir 收口 **达到本阶段门禁**；尚未覆盖并发双 advance、备份恢复演练、Runner 故障注入（Phase 5 后续项）。
+
+**后续**:
+1. 长任务 worker lease/heartbeat 与真正重启恢复（非仅标记 interrupted）
+2. Phase 5 故障注入矩阵（Finalize 故障、滚动重启、并发 cancel）
+3. Gap P1 余项：Mermaid strict、上传硬限、Runner 服务间认证

deploy/postgres/goose/00019_2026-09-12-task-runs-events.sql

@@ -0,0 +1,41 @@
+-- +goose Up
+-- Phase 4: 长任务持久化与 SSE 事件重放
+
+CREATE TABLE IF NOT EXISTS task_runs (
+    run_id               TEXT PRIMARY KEY,
+    session_id           TEXT NOT NULL DEFAULT '',
+    project_id           BIGINT REFERENCES projects(id) ON DELETE SET NULL,
+    user_id              BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    client_turn_id       TEXT NOT NULL DEFAULT '',
+    task_key             TEXT NOT NULL DEFAULT '',
+    stage                TEXT NOT NULL DEFAULT '',
+    status               TEXT NOT NULL DEFAULT 'running',
+    workspace_id         TEXT NOT NULL DEFAULT '',
+    assistant_message_id BIGINT,
+    error_message        TEXT NOT NULL DEFAULT '',
+    idempotency_key      TEXT,
+    created_at           TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at           TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT chk_task_run_status CHECK (
+        status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'finalizing_failed', 'quality_failed', 'interrupted')
+    )
+);
+
+CREATE INDEX IF NOT EXISTS idx_task_runs_user ON task_runs(user_id, updated_at DESC);
+CREATE INDEX IF NOT EXISTS idx_task_runs_project ON task_runs(project_id, updated_at DESC);
+CREATE INDEX IF NOT EXISTS idx_task_runs_running ON task_runs(status) WHERE status = 'running';
+
+CREATE TABLE IF NOT EXISTS task_run_events (
+    run_id     TEXT NOT NULL REFERENCES task_runs(run_id) ON DELETE CASCADE,
+    seq        BIGINT NOT NULL,
+    event_type TEXT NOT NULL,
+    payload    JSONB NOT NULL DEFAULT '{}'::jsonb,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (run_id, seq)
+);
+
+CREATE INDEX IF NOT EXISTS idx_task_run_events_run ON task_run_events(run_id, seq);
+
+-- +goose Down
+DROP TABLE IF EXISTS task_run_events;
+DROP TABLE IF EXISTS task_runs;

internal/server/auth_middleware.go

@@ -17,7 +17,8 @@ func (s *Server) allowLegacyTestPrincipal() bool {
 }
 
 func isPublicRoute(method, path string) bool {
-	if path == "/" || path == "/health" || path == "/api/v1/health" {
+	if path == "/" || path == "/health" || path == "/api/v1/health" ||
+		path == "/ready" || path == "/api/v1/ready" {
 		return true
 	}
 	if method == http.MethodPost && path == "/api/v1/auth/login" {

internal/server/handlers.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"net/http"
@@ -14,6 +15,7 @@ import (
 	"chat-agent-system/internal/docgov"
 	"chat-agent-system/internal/projectworkflow"
 	"chat-agent-system/internal/squad"
+	"chat-agent-system/internal/taskrun"
 )
 
 // HealthResponse 健康检查响应
@@ -25,6 +27,8 @@ type HealthResponse struct {
 	SharedKBPort  int    `json:"shared_kb_port"`
 	DatabaseReady bool   `json:"database_ready"`
 	SchemaVersion int64  `json:"schema_version,omitempty"`
+	SchemaReady   bool   `json:"schema_ready,omitempty"`
+	TaskStoreReady bool  `json:"task_store_ready,omitempty"`
 }
 
 // ChatRequest 对话请求
@@ -74,25 +78,47 @@ type TaskRunRequestDTO struct {
 }
 
 func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
+	resp, code := s.buildHealthResponse(false)
+	writeJSONResponse(w, code, resp)
+}
+
+func (s *Server) handleReadiness(w http.ResponseWriter, r *http.Request) {
+	resp, code := s.buildHealthResponse(true)
+	writeJSONResponse(w, code, resp)
+}
+
+func (s *Server) buildHealthResponse(strict bool) (HealthResponse, int) {
 	dbReady := s.svc != nil && s.svc.DatabaseReady()
+	schemaVersion := int64(0)
+	if s.svc != nil {
+		schemaVersion = s.svc.SchemaVersion()
+	}
+	schemaReady := schemaVersion > 0 || (s.cfg != nil && s.cfg.AllowInMemory)
+	taskStoreReady := s.svc != nil && taskrun.TablesReady(context.Background(), s.svc.Pool())
+
 	status := "ok"
 	code := http.StatusOK
-	if s.cfg != nil && s.cfg.ContextDbHost != "" && !s.cfg.AllowInMemory && !dbReady {
-		status = "degraded"
-		code = http.StatusServiceUnavailable
+	if s.cfg != nil && s.cfg.ContextDbHost != "" && !s.cfg.AllowInMemory {
+		if !dbReady {
+			status = "degraded"
+			code = http.StatusServiceUnavailable
+		} else if strict && (!schemaReady || !taskStoreReady) {
+			status = "not_ready"
+			code = http.StatusServiceUnavailable
+		}
 	}
-	resp := HealthResponse{
-		Status:        status,
-		Env:           s.cfg.Env,
-		AgentBackend:  s.cfg.AgentBackend,
-		Version:       "v1.0",
-		SharedKBPort:  s.cfg.KbDbPort,
-		DatabaseReady: dbReady,
-	}
-	if s.svc != nil {
-		resp.SchemaVersion = s.svc.SchemaVersion()
-	}
-	writeJSONResponse(w, code, resp)
+
+	return HealthResponse{
+		Status:         status,
+		Env:            s.cfg.Env,
+		AgentBackend:   s.cfg.AgentBackend,
+		Version:        "v1.0",
+		SharedKBPort:   s.cfg.KbDbPort,
+		DatabaseReady:  dbReady,
+		SchemaVersion:  schemaVersion,
+		SchemaReady:    schemaReady,
+		TaskStoreReady: taskStoreReady,
+	}, code
 }
 
 func (s *Server) handleGetProject(w http.ResponseWriter, r *http.Request) {
@@ -126,6 +152,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "message cannot be empty", http.StatusBadRequest)
 		return
 	}
+	req.WorkDir = ""
 
 	sess := s.getOrCreateSession(
 		req.SessionID,

internal/server/server.go

@@ -86,6 +86,11 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
 		docChecker := projectinstance.NewDocGovChecker(svc.DocGov())
 		instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
 	}
+	taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager())
+	if pool != nil {
+		taskMgr.SetStore(taskrun.NewPGStore(pool))
+	}
+
 	s := &Server{
 		cfg:        cfg,
 		svc:        svc,
@@ -94,7 +99,7 @@ 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()),
+		taskRuns:   taskMgr,
 		authorizer: authz.NewAuthorizer(pool),
 		authzSvc:   authzSvc,
 		wfDefSvc:   wfDefSvc,
@@ -121,6 +126,11 @@ func (s *Server) BizdevService() *bizdev.Service {
 	return s.svc
 }
 
+// TaskRuns 获取长任务管理器（测试与运维钩子）。
+func (s *Server) TaskRuns() *taskrun.Manager {
+	return s.taskRuns
+}
+
 // ProjectWorkflowService 获取项目多工作流控制服务
 func (s *Server) ProjectWorkflowService() *projectworkflow.Service {
 	return s.pwSvc
@@ -133,6 +143,8 @@ func (s *Server) Routes() http.Handler {
 	// 1. 健康检查与系统自检
 	mux.HandleFunc("GET /health", s.handleHealth)
 	mux.HandleFunc("GET /api/v1/health", s.handleHealth)
+	mux.HandleFunc("GET /ready", s.handleReadiness)
+	mux.HandleFunc("GET /api/v1/ready", s.handleReadiness)
 
 	// 2. 项目与多工作流控制面
 	mux.HandleFunc("POST /api/v1/projects", s.handleCreateProject)

internal/server/sse.go

@@ -150,6 +150,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	req.UserID = user.ID
+	req.WorkDir = ""
 
 	// 1. 设置 SSE 响应头 + 请求追踪 ID
 	requestID := fmt.Sprintf("req_%d", time.Now().UnixNano())

internal/server/task_run_persistence_pg_test.go

@@ -0,0 +1,150 @@
+package server_test
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/stream"
+	"chat-agent-system/internal/taskrun"
+)
+
+func TestTaskRun_PersistenceAndReplayAfterRestart_PG(t *testing.T) {
+	pool := openPGPoolWithMigrations(t)
+	if pool == nil {
+		return
+	}
+	defer pool.Close()
+	if !taskrun.TablesReady(context.Background(), pool) {
+		t.Skip("task_runs table not ready")
+	}
+
+	handler, srv := setupPGGovernanceServer(t, pool)
+	token := loginTestUser(t, handler, "zhanggong", "123456")
+
+	runID := "run_pg_persist_" + time.Now().Format("150405")
+	store := taskrun.NewPGStore(pool)
+	srv.TaskRuns().SetStore(store)
+
+	srv.TaskRuns().SetExecutor(
+		func(_ context.Context, _ string, _ bizdev.TaskRunRequest, st *stream.Streamer) (*bizdev.TaskRunResult, error) {
+			if st != nil {
+				st.EmitPhase("test", "running", "step1", 1, 1, 50)
+			}
+			return &bizdev.TaskRunResult{Success: true, WorkDir: t.TempDir(), TaskKey: "intake"}, nil
+		},
+		nil,
+		nil,
+	)
+	srv.TaskRuns().SetFinalizer(func(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) {
+		return srv.FinalizeTaskRunProjection(ctx, record, result)
+	})
+
+	record := &taskrun.RunRecord{
+		RunID: runID, SessionID: "sess_pg_persist", ProjectID: 0,
+		UserID: fetchAuthUserID(t, handler, token), TaskKey: "intake", Status: "running",
+	}
+	_, err := srv.TaskRuns().CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "")
+	if err != nil {
+		t.Fatalf("CreateAndStart: %v", err)
+	}
+
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		rec, getErr := store.GetRun(context.Background(), runID)
+		if getErr == nil && rec.Status == "completed" {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	persisted, err := store.GetRun(context.Background(), runID)
+	if err != nil {
+		t.Fatalf("GetRun from store: %v", err)
+	}
+	if persisted.Status != "completed" {
+		t.Fatalf("expected completed, got %s", persisted.Status)
+	}
+
+	events, err := store.ListEvents(context.Background(), runID, 0)
+	if err != nil {
+		t.Fatalf("ListEvents: %v", err)
+	}
+	if len(events) == 0 {
+		t.Fatal("expected persisted events")
+	}
+	hasDone := false
+	for _, ev := range events {
+		if ev.Event.Type == stream.EventDone {
+			hasDone = true
+		}
+	}
+	if !hasDone {
+		t.Fatal("expected done event in persisted events")
+	}
+
+	// 模拟进程重启：新 Manager 仅挂载 Store
+	restarted := taskrun.NewManager(srv.ProjectWorkflowService(), srv.BizdevService().ContextManager())
+	restarted.SetStore(store)
+
+	reloaded, err := restarted.GetRun(runID)
+	if err != nil {
+		t.Fatalf("GetRun after restart: %v", err)
+	}
+	if reloaded.Status != "completed" {
+		t.Fatalf("reloaded status=%s", reloaded.Status)
+	}
+
+	ch, replay, err := restarted.Subscribe(runID, 0)
+	if err != nil {
+		t.Fatalf("Subscribe after restart: %v", err)
+	}
+	if len(replay) == 0 {
+		t.Fatal("expected replay events from DB")
+	}
+	_, ok := <-ch
+	if ok {
+		t.Fatal("replay-only channel should be closed")
+	}
+
+	// HTTP GET 鉴权 + 查询
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/tasks/runs/"+runID, nil)
+	withBearerAuth(req, token)
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("GET task run: %d %s", rec.Code, rec.Body.String())
+	}
+	var body taskrun.RunRecord
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if body.RunID != runID {
+		t.Fatalf("run_id=%s", body.RunID)
+	}
+}
+
+func TestReadiness_StrictWhenPGConfigured(t *testing.T) {
+	pool := openPGPoolWithMigrations(t)
+	if pool == nil {
+		return
+	}
+	defer pool.Close()
+
+	handler, _ := setupPGGovernanceServer(t, pool)
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/ready", nil)
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("ready expected 200, got %d %s", rec.Code, rec.Body.String())
+	}
+	var resp map[string]any
+	_ = json.Unmarshal(rec.Body.Bytes(), &resp)
+	if resp["task_store_ready"] != true {
+		t.Fatalf("task_store_ready=%v", resp["task_store_ready"])
+	}
+}

internal/taskrun/manager.go

@@ -25,10 +25,11 @@ type StageExecutor func(ctx context.Context, runID string, req bizdev.StageRunRe
 // Finalizer 任务成功后投影运行记录与聊天消息（由 server 注入）。
 type Finalizer func(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult) (terminalStatus string, err error)
 
-// Manager 后台任务运行观察者：SSE 断线不取消 worker。
+// Manager 后台任务运行观察者：SSE 断线不取消 worker；可选 PG 持久化支持重启重放。
 type Manager struct {
 	mu       sync.RWMutex
 	runs     map[string]*activeRun
+	store    Store
 	exec     Executor
 	resume   ResumeExecutor
 	stage    StageExecutor
@@ -42,6 +43,7 @@ const runRetention = time.Hour
 func NewManager(pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *Manager {
 	m := &Manager{
 		runs:   make(map[string]*activeRun),
+		store:  noopStore{},
 		pwSvc:  pwSvc,
 		ctxMgr: ctxMgr,
 	}
@@ -49,6 +51,24 @@ func NewManager(pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *
 	return m
 }
 
+// SetStore 注入持久化存储（PG）；启动时标记孤儿 running 任务为 interrupted。
+func (m *Manager) SetStore(store Store) {
+	if m == nil {
+		return
+	}
+	if store == nil {
+		store = noopStore{}
+	}
+	m.store = store
+	if store.Available() {
+		if n, err := store.MarkOrphanedRunning(context.Background()); err != nil {
+			slog.Warn("taskrun_orphan_recovery_failed", "error", err)
+		} else if n > 0 {
+			slog.Info("taskrun_orphans_marked_interrupted", "count", n)
+		}
+	}
+}
+
 func (m *Manager) cleanupLoop() {
 	ticker := time.NewTicker(10 * time.Minute)
 	defer ticker.Stop()
@@ -97,6 +117,12 @@ func (m *Manager) CreateAndStart(parent context.Context, runID string, record *R
 	record.CreatedAt = now
 	record.UpdatedAt = now
 
+	if m.store != nil && m.store.Available() {
+		if err := m.store.InsertRun(parent, record); err != nil {
+			slog.Warn("taskrun_insert_failed", "run_id", runID, "error", err)
+		}
+	}
+
 	workerCtx, cancel := context.WithCancel(context.WithoutCancel(parent))
 
 	ar := &activeRun{
@@ -177,7 +203,9 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T
 	if workspaceID != "" {
 		ar.mu.Lock()
 		ar.record.WorkspaceID = workspaceID
+		ar.record.UpdatedAt = time.Now()
 		ar.mu.Unlock()
+		m.persistRun(ar)
 	}
 
 	if execErr != nil {
@@ -197,18 +225,12 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T
 		return
 	}
 
-	m.appendEvent(ar, stream.StreamEvent{
-		Type: stream.EventDone,
-		Done: &stream.DonePayload{
-			SessionID:   "task_run",
-			WorkspaceID: workspaceID,
-			DurationMs:  time.Since(start).Milliseconds(),
-		},
-	})
-
+	// RUN-002: 先 finalize 落库，成功后再发 done 终态事件
 	terminalStatus := "completed"
 	if m.finalize != nil && taskResult != nil {
 		if ts, finErr := m.finalize(ctx, ar.record, taskResult); finErr != nil {
+			info := stream.ClassifyError(finErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, workspaceID)
+			m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info})
 			m.finishRun(ar, "finalizing_failed", workspaceID, finErr.Error(), taskResult)
 			return
 		} else if ts != "" {
@@ -217,6 +239,15 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T
 	} else if taskResult != nil && taskResult.QualityReport != nil && !taskResult.QualityReport.Passed {
 		terminalStatus = "quality_failed"
 	}
+
+	m.appendEvent(ar, stream.StreamEvent{
+		Type: stream.EventDone,
+		Done: &stream.DonePayload{
+			SessionID:   "task_run",
+			WorkspaceID: workspaceID,
+			DurationMs:  time.Since(start).Milliseconds(),
+		},
+	})
 	m.finishRun(ar, terminalStatus, workspaceID, "", taskResult)
 }
 
@@ -234,6 +265,7 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, r
 		"workspace_id", workspaceID,
 		"error", errMsg,
 	)
+	m.persistRun(ar)
 
 	// 失败/取消路径：若 Finalize 未执行，仍更新 workflow_runs 与占位消息 metadata
 	if status == "failed" || status == "cancelled" || status == "finalizing_failed" {
@@ -261,17 +293,37 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, r
 	_ = result // success path persistence handled by Finalizer
 }
 
+func (m *Manager) persistRun(ar *activeRun) {
+	if m.store == nil || !m.store.Available() || ar == nil {
+		return
+	}
+	ar.mu.RLock()
+	rec := *ar.record
+	ar.mu.RUnlock()
+	if err := m.store.UpdateRun(context.Background(), &rec); err != nil {
+		slog.Warn("taskrun_update_failed", "run_id", rec.RunID, "error", err)
+	}
+}
+
 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)
+	runID := ar.record.RunID
+	seq := ar.nextSeq
 	subs := make([]chan StoredEvent, 0, len(ar.subs))
 	for _, ch := range ar.subs {
 		subs = append(subs, ch)
 	}
 	ar.mu.Unlock()
 
+	if m.store != nil && m.store.Available() {
+		if err := m.store.AppendEvent(context.Background(), runID, seq, ev); err != nil {
+			slog.Warn("taskrun_event_persist_failed", "run_id", runID, "seq", seq, "error", err)
+		}
+	}
+
 	for _, ch := range subs {
 		select {
 		case ch <- stored:
@@ -284,36 +336,51 @@ 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")
+	if ok {
+		ar.mu.RLock()
+		defer ar.mu.RUnlock()
+		cpy := *ar.record
+		return &cpy, nil
 	}
-	ar.mu.RLock()
-	defer ar.mu.RUnlock()
-	cpy := *ar.record
-	return &cpy, nil
+	if m.store != nil && m.store.Available() {
+		return m.store.GetRun(context.Background(), runID)
+	}
+	return nil, fmt.Errorf("run not found")
 }
 
 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")
+	if ok {
+		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
 	}
 
-	ar.mu.Lock()
-	defer ar.mu.Unlock()
-	replay := make([]StoredEvent, 0)
-	for _, ev := range ar.events {
-		if ev.Seq > afterSeq {
-			replay = append(replay, ev)
+	if m.store != nil && m.store.Available() {
+		if _, err := m.store.GetRun(context.Background(), runID); err != nil {
+			return nil, nil, fmt.Errorf("run not found")
 		}
+		events, err := m.store.ListEvents(context.Background(), runID, afterSeq)
+		if err != nil {
+			return nil, nil, err
+		}
+		ch := make(chan StoredEvent)
+		close(ch)
+		return ch, events, nil
 	}
-	ch := make(chan StoredEvent, 32)
-	ar.subSeq++
-	id := ar.subSeq
-	ar.subs[id] = ch
-	return ch, replay, nil
+	return nil, nil, fmt.Errorf("run not found")
 }
 
 func (m *Manager) Unsubscribe(runID string, ch <-chan StoredEvent) {

internal/taskrun/manager_finalize_order_test.go

@@ -0,0 +1,105 @@
+package taskrun
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/stream"
+)
+
+func TestManager_DoneEmittedAfterFinalize(t *testing.T) {
+	m := NewManager(nil, nil)
+	finalizeCalled := false
+
+	m.SetExecutor(
+		func(_ context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
+			return &bizdev.TaskRunResult{Success: true, WorkDir: "/tmp/ws"}, nil
+		},
+		nil,
+		nil,
+	)
+	m.SetFinalizer(func(_ context.Context, _ *RunRecord, _ *bizdev.TaskRunResult) (string, error) {
+		finalizeCalled = true
+		time.Sleep(20 * time.Millisecond)
+		return "completed", nil
+	})
+
+	runID := "run_finalize_order"
+	record := &RunRecord{RunID: runID, Status: "running"}
+	_, err := m.CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "")
+	if err != nil {
+		t.Fatalf("CreateAndStart: %v", err)
+	}
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		rec, getErr := m.GetRun(runID)
+		if getErr == nil && rec.Status != "running" {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	m.mu.RLock()
+	ar := m.runs[runID]
+	m.mu.RUnlock()
+	if ar == nil {
+		t.Fatal("active run evicted too early")
+	}
+	ar.mu.RLock()
+	defer ar.mu.RUnlock()
+	if !finalizeCalled {
+		t.Fatal("finalize was not called")
+	}
+	if len(ar.events) == 0 {
+		t.Fatal("expected events")
+	}
+	last := ar.events[len(ar.events)-1]
+	if last.Event.Type != stream.EventDone {
+		t.Fatalf("last event type=%s want done", last.Event.Type)
+	}
+}
+
+func TestManager_FinalizeFailureDoesNotEmitDone(t *testing.T) {
+	m := NewManager(nil, nil)
+	m.SetExecutor(
+		func(_ context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
+			return &bizdev.TaskRunResult{Success: true}, nil
+		},
+		nil,
+		nil,
+	)
+	m.SetFinalizer(func(_ context.Context, _ *RunRecord, _ *bizdev.TaskRunResult) (string, error) {
+		return "", errors.New("db write failed")
+	})
+
+	runID := "run_finalize_fail"
+	record := &RunRecord{RunID: runID, Status: "running"}
+	_, _ = m.CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "")
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		rec, getErr := m.GetRun(runID)
+		if getErr == nil && rec.Status == "finalizing_failed" {
+			m.mu.RLock()
+			ar := m.runs[runID]
+			m.mu.RUnlock()
+			if ar == nil {
+				t.Fatal("missing active run")
+			}
+			ar.mu.RLock()
+			defer ar.mu.RUnlock()
+			for _, ev := range ar.events {
+				if ev.Event.Type == stream.EventDone {
+					t.Fatal("done must not be emitted when finalize fails")
+				}
+			}
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("expected finalizing_failed status")
+}

internal/taskrun/store.go

@@ -0,0 +1,234 @@
+package taskrun
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"chat-agent-system/internal/stream"
+)
+
+var ErrStoreNotAvailable = errors.New("taskrun: store not available")
+
+// Store 持久化 task run 与 SSE 事件，支持重启后查询与重放。
+type Store interface {
+	Available() bool
+	InsertRun(ctx context.Context, rec *RunRecord) error
+	UpdateRun(ctx context.Context, rec *RunRecord) error
+	AppendEvent(ctx context.Context, runID string, seq int64, ev stream.StreamEvent) error
+	GetRun(ctx context.Context, runID string) (*RunRecord, error)
+	ListEvents(ctx context.Context, runID string, afterSeq int64) ([]StoredEvent, error)
+	MarkOrphanedRunning(ctx context.Context) (int, error)
+}
+
+// PGStore PostgreSQL 实现。
+type PGStore struct {
+	pool *pgxpool.Pool
+}
+
+func NewPGStore(pool *pgxpool.Pool) *PGStore {
+	return &PGStore{pool: pool}
+}
+
+func (s *PGStore) Available() bool {
+	if s == nil || s.pool == nil {
+		return false
+	}
+	var ok bool
+	_ = s.pool.QueryRow(context.Background(), `
+		SELECT EXISTS (
+			SELECT 1 FROM information_schema.tables
+			WHERE table_schema = 'public' AND table_name = 'task_runs'
+		)
+	`).Scan(&ok)
+	return ok
+}
+
+func (s *PGStore) InsertRun(ctx context.Context, rec *RunRecord) error {
+	if !s.Available() {
+		return ErrStoreNotAvailable
+	}
+	if rec == nil || rec.RunID == "" {
+		return fmt.Errorf("taskrun: run_id required")
+	}
+	_, err := s.pool.Exec(ctx, `
+		INSERT INTO task_runs (
+			run_id, session_id, project_id, user_id, client_turn_id,
+			task_key, stage, status, workspace_id, assistant_message_id,
+			error_message, created_at, updated_at
+		) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
+		ON CONFLICT (run_id) DO NOTHING
+	`,
+		rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID,
+		rec.TaskKey, rec.Stage, rec.Status, rec.WorkspaceID, nullInt64(rec.AssistantMsgID),
+		rec.ErrorMessage, rec.CreatedAt, rec.UpdatedAt,
+	)
+	return err
+}
+
+func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error {
+	if !s.Available() {
+		return ErrStoreNotAvailable
+	}
+	if rec == nil || rec.RunID == "" {
+		return fmt.Errorf("taskrun: run_id required")
+	}
+	_, err := s.pool.Exec(ctx, `
+		UPDATE task_runs SET
+			status = $2,
+			workspace_id = $3,
+			error_message = $4,
+			updated_at = $5
+		WHERE run_id = $1
+	`, rec.RunID, rec.Status, rec.WorkspaceID, rec.ErrorMessage, rec.UpdatedAt)
+	return err
+}
+
+func (s *PGStore) AppendEvent(ctx context.Context, runID string, seq int64, ev stream.StreamEvent) error {
+	if !s.Available() {
+		return ErrStoreNotAvailable
+	}
+	payload, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	_, err = s.pool.Exec(ctx, `
+		INSERT INTO task_run_events (run_id, seq, event_type, payload)
+		VALUES ($1, $2, $3, $4::jsonb)
+		ON CONFLICT (run_id, seq) DO NOTHING
+	`, runID, seq, string(ev.Type), payload)
+	return err
+}
+
+func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) {
+	if !s.Available() {
+		return nil, ErrStoreNotAvailable
+	}
+	var rec RunRecord
+	var projectID, assistantMsgID *int64
+	err := s.pool.QueryRow(ctx, `
+		SELECT run_id, session_id, project_id, user_id, client_turn_id,
+			task_key, stage, status, workspace_id, assistant_message_id,
+			error_message, created_at, updated_at
+		FROM task_runs
+		WHERE run_id = $1
+	`, runID).Scan(
+		&rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID,
+		&rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID,
+		&rec.ErrorMessage, &rec.CreatedAt, &rec.UpdatedAt,
+	)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, fmt.Errorf("run not found")
+		}
+		return nil, err
+	}
+	if projectID != nil {
+		rec.ProjectID = *projectID
+	}
+	if assistantMsgID != nil {
+		rec.AssistantMsgID = *assistantMsgID
+	}
+	return &rec, nil
+}
+
+func (s *PGStore) ListEvents(ctx context.Context, runID string, afterSeq int64) ([]StoredEvent, error) {
+	if !s.Available() {
+		return nil, ErrStoreNotAvailable
+	}
+	rows, err := s.pool.Query(ctx, `
+		SELECT seq, payload
+		FROM task_run_events
+		WHERE run_id = $1 AND seq > $2
+		ORDER BY seq ASC
+	`, runID, afterSeq)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := make([]StoredEvent, 0)
+	for rows.Next() {
+		var seq int64
+		var payload []byte
+		if err := rows.Scan(&seq, &payload); err != nil {
+			return nil, err
+		}
+		var ev stream.StreamEvent
+		if err := json.Unmarshal(payload, &ev); err != nil {
+			return nil, err
+		}
+		out = append(out, StoredEvent{Seq: seq, Event: ev})
+	}
+	return out, rows.Err()
+}
+
+func (s *PGStore) MarkOrphanedRunning(ctx context.Context) (int, error) {
+	if !s.Available() {
+		return 0, ErrStoreNotAvailable
+	}
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE task_runs
+		SET status = 'interrupted',
+		    error_message = '进程重启导致任务中断，请重新提交',
+		    updated_at = now()
+		WHERE status = 'running'
+	`)
+	if err != nil {
+		return 0, err
+	}
+	return int(tag.RowsAffected()), nil
+}
+
+func nullInt64(v int64) *int64 {
+	if v <= 0 {
+		return nil
+	}
+	return &v
+}
+
+// TablesReady 检查 task_runs 表是否已迁移。
+func TablesReady(ctx context.Context, pool *pgxpool.Pool) bool {
+	if pool == nil {
+		return false
+	}
+	var ok bool
+	_ = pool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1 FROM information_schema.tables
+			WHERE table_schema = 'public' AND table_name = 'task_runs'
+		)
+	`).Scan(&ok)
+	return ok
+}
+
+// noopStore 无持久化时的空实现。
+type noopStore struct{}
+
+func (noopStore) Available() bool { return false }
+func (noopStore) InsertRun(context.Context, *RunRecord) error {
+	return ErrStoreNotAvailable
+}
+func (noopStore) UpdateRun(context.Context, *RunRecord) error { return ErrStoreNotAvailable }
+func (noopStore) AppendEvent(context.Context, string, int64, stream.StreamEvent) error {
+	return ErrStoreNotAvailable
+}
+func (noopStore) GetRun(context.Context, string) (*RunRecord, error) {
+	return nil, ErrStoreNotAvailable
+}
+func (noopStore) ListEvents(context.Context, string, int64) ([]StoredEvent, error) {
+	return nil, ErrStoreNotAvailable
+}
+func (noopStore) MarkOrphanedRunning(context.Context) (int, error) { return 0, ErrStoreNotAvailable }
+
+// EnsureUpdatedAt 若 UpdatedAt 为零则设为当前时间。
+func EnsureUpdatedAt(rec *RunRecord) {
+	if rec != nil && rec.UpdatedAt.IsZero() {
+		rec.UpdatedAt = time.Now()
+	}
+}