fix(docgov): idempotent lookup before busy; SSE write deadline heartbeat

Return existing task run 202 for the same turn before scope/preflight so SSE disconnect retries do not hit DOC_BUSY or duplicate chat. Extend task-run events write deadlines with heartbeats without dropping global HTTP WriteTimeout. Reconnect subscribeTaskRunEvents on EOF via after_seq. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-13 14:37:18 +00:00
parent f3e4eeb42b
commit 43e03e4969
11 changed files with 541 additions and 27 deletions
@@ -0,0 +1,46 @@
+---
+title: "任务 run SSE 超时断线后同 turn 误击 DOC_BUSY"
+status: in_progress
+created: 2026-09-13
+updated: 2026-09-13
+severity: critical
+---
+
+# 任务 run SSE 超时断线后同 turn 误击 DOC_BUSY
+
+## 现象
+
+真实 Antigravity UI-03（dev schema 30）：
+
+1. 初始 `POST /api/v1/tasks/runs` → **202** `run_1789308825655387640`
+2. `GET .../events` 约 **157s** 后以 `ERR_INCOMPLETE_CHUNKED_ENCODING` 断开（对齐服务端默认 `WriteTimeout=120s`）
+3. 浏览器用**完全相同** `session=sess-p6098-u107`、`client_turn_id=turn_1789308825646`、`doc_id=doc_proj6098_intake` 再 POST
+4. `handleCreateTaskRun` 在幂等 `ClaimRun` **之前**执行 `preflightTaskDocumentBusy` → **409 DOC_BUSY**，并 `recordBusyPreflightMessages` 写入**重复**用户/失败助手消息
+
+## 根因分析
+
+| 层 | 问题 |
+|---|---|
+| HTTP | 全局 `WriteTimeout: 120s` 切断长任务 SSE 响应体，客户端见 incomplete chunked encoding |
+| Handler | 同 turn 重试仍走 scope/preflight；活跃小队租约被当成「别人在编」 |
+| FE | `subscribeTaskRunEvents` 遇意外 EOF 直接结束，上层可能再次 `createTaskRun` |
+
+## 严重度
+
+`critical`
+
+## 关联整改
+
+- Short-term: [../ops/short-term/2026-09-13-document-revision-sse-idempotency.md](../ops/short-term/2026-09-13-document-revision-sse-idempotency.md)
+- 分支: `fix/document-revision-sse-idempotency`
+
+## 证据
+
+- 运行产物目录（勿删）：[../test-reports/assets/2026-09-13-document-revision-runtime-ui/](../test-reports/assets/2026-09-13-document-revision-runtime-ui/)
+- 前序 UI-03 客户端同步审计：[./2026-09-13-controlled-doc-client-sync-ui03.md](./2026-09-13-controlled-doc-client-sync-ui03.md)
+- 现场键：`run_1789308825655387640` / `sess-p6098-u107` / `turn_1789308825646` / `doc_proj6098_intake`
+
+## 开放问题
+
+- 合入前占槽真实 UI 重跑 UI-03（本轮不 rebuild Docker）
+- chat `/stream` 与 `tasks/run/stream` 是否需同样 SSE write-deadline 扩展（本轮优先 `/tasks/runs/{id}/events`）
@@ -1,11 +1,11 @@
 # 当前焦点

 **更新**: 2026-09-13  
-**production_ready**: no（DOCREV 集成槽 e2e；UI-00~02 PASS；UI-03 工程修复在 `fix/document-revision-runtime-ui03`：post-task sync + goose 00030 + Claim 幂等；schema30 卷迁移与真实 UI-03 待合入前验证）
+**production_ready**: no（DOCREV：schema **30** 已合入；真实 UI-03 暴露 SSE WriteTimeout 断线后同 turn 误 409 DOC_BUSY；整改在 `fix/document-revision-sse-idempotency`）

 ## 焦点

-1. **DOCREV-RT**: [运行时整改收口](short-term/2026-09-13-document-revision-runtime.md)（`fix/document-revision-runtime-ui03`：UI-03 客户端同步/幂等/chk；占槽真实 UI 重跑；不 push）
+1. **DOCREV-RT**: [SSE 幂等与写超时](short-term/2026-09-13-document-revision-sse-idempotency.md)（`fix/document-revision-sse-idempotency`：幂等先于 busy、SSE heartbeat、FE after_seq 重连；不 push）
 2. **CHAT**: [刷新恢复整改已实施](short-term/2026-09-13-chat-history-refresh-remediation.md)（待合 dev + PI E2E 补录）
 3. **CI**: `TEST_DATABASE_URL` + 真实 PI E2E 实录；INTENT 发布 blocker 见 conversation-intent short-term

@@ -16,7 +16,8 @@
 | Cursor | chat-history-refresh-remediation | fix/chat-history-refresh | - | chat | in_progress |
 | Codex | conversation-intent-runtime-state | fix/conversation-intent-state | .worktrees/fix-conversation-intent-state | intent | in_progress |
 | Cursor | document-revision-wave1-6-implementation | feature/document-revision-management | - | docgov | done |
-| Cursor-Auto | document-revision-runtime-ui03 | fix/document-revision-runtime-ui03 | - | docgov | in_progress |
+| Cursor-Auto | document-revision-runtime-ui03 | fix/document-revision-runtime-ui03 | - | docgov | done |
+| Cursor-Auto | document-revision-sse-idempotency | fix/document-revision-sse-idempotency | - | docgov | in_progress |

 > 新任务开始前登记。`branch` 必须为 `feature/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。

@@ -24,7 +25,7 @@

 | holder | phase | since | scope | queue |
 |--------|-------|-------|-------|-------|
-| Codex/Antigravity | e2e | 2026-09-13 11:16 UTC | DOCREV-RT real browser UI + real PI E2E on schema 29 | - |
+| Codex/Antigravity | e2e | 2026-09-13 | DOCREV-RT real UI on **schema 30**；SSE 断线/同 turn 409 现场保留 | Cursor-Auto (sse-idempotency 编码，待占槽冒烟) |

 **phase**: `idle` | `merging` | `smoke` | `e2e`

@@ -34,7 +35,7 @@

 | 环境 | 端口 | schema | 状态 |
 |------|------|--------|------|
-| dev | 8080 | 28 | healthy |
+| dev | 8080 | **30** | healthy（UI-03 现场） |
 | test | 8081 | 25→27* | healthy |
 | main | 8000 | 25→27* | healthy（app + runner + DB） |

@@ -49,3 +50,4 @@
 | Agent 手册 | [AGENTS.md](../../AGENTS.md) |
 | DOCREV ADR | [../adr/2026-09-13-document-version-boundary.md](../adr/2026-09-13-document-version-boundary.md) |
 | DOCREV 测试 | [../test-reports/2026-09-13-document-revision-management-test-report.md](../test-reports/2026-09-13-document-revision-management-test-report.md) |
+| SSE/幂等审计 | [../audit/2026-09-13-document-revision-sse-idempotency.md](../audit/2026-09-13-document-revision-sse-idempotency.md) |
@@ -0,0 +1,46 @@
+---
+title: "任务 SSE 写超时与同 turn 幂等先于 busy 预检"
+status: in_progress
+created: 2026-09-13
+updated: 2026-09-13
+owner: Cursor-Auto
+branch: fix/document-revision-sse-idempotency
+worktree: "-"
+mutex: docgov
+---
+
+# 任务 SSE 写超时与同 turn 幂等先于 busy 预检
+
+## 目标
+
+同 turn 在 SSE 断线重连时返回既有 run 202、不重复聊天、不误 409 DOC_BUSY；长任务 events SSE 不被 120s WriteTimeout 切断；前端 EOF 按 after_seq 重订阅同一 run。
+
+## 范围
+
+- `internal/server/task_run_handlers.go`：授权后幂等查找先于 scope/preflight
+- `internal/server`：SSE 专用 `ResponseController` 写期限延长 + heartbeat
+- `web/src/api/taskRuns.ts` + `chatStream` seq：EOF 重连
+- 单测与文档
+
+## 非目标
+
+- 不改全局 HTTP 超时为「无超时」；不 rebuild Docker；不 push/merge；不删证据 assets
+
+## 步骤
+
+1. [x] 从 `dev` 开 `fix/document-revision-sse-idempotency`；CURRENT 认领
+2. [x] 幂等先查 + 属主校验
+3. [x] SSE write deadline + heartbeat
+4. [x] FE subscribe 重连
+5. [x] 测试与原子提交
+
+## 验收标准
+
+- [x] 同 turn + 活跃租约 → 同 run 202、仅一对消息；不同 turn → 仍 409
+- [x] SSE 心跳/写期限可测；FE EOF/after_seq 单测
+- [x] `go test -short ./...`、`go vet`、`npm test`/`build`、`git diff --check`
+
+## 相关链接
+
+- Audit: [../../audit/2026-09-13-document-revision-sse-idempotency.md](../../audit/2026-09-13-document-revision-sse-idempotency.md)
+- Evidence: [../../test-reports/assets/2026-09-13-document-revision-runtime-ui/](../../test-reports/assets/2026-09-13-document-revision-runtime-ui/)
@@ -62,6 +62,7 @@
 | ENV-RESTART-SOURCE | `docker compose restart` 只重启旧镜像，健康检查无法证明合入代码已部署 | **PASS** | `env-service.sh rebuild` + 真实摘要重测 | 集成冒烟统一使用 `dev rebuild` |
 | DOCREV-RUNTIME-404 | revision-state 404；账本内存丢失；块保存/overlay 无效；agent 假报告 | **FAIL** | [audit](../audit/2026-09-13-document-revision-runtime-404.md)；`internal/docrevision/service.go` 仅 map | [remediation](./2026-09-13-document-revision-runtime-remediation.md) |
 | DOCREV-UI03-CLIENT-SYNC | 小队完成后右栏仍模板；UI-03 选择器误命中 Banner；`pending_author_confirmation` 撞 chk；重试重复建 run | **FIX-LANDED**（工程） | [audit](../audit/2026-09-13-controlled-doc-client-sync-ui03.md)；分支 `fix/document-revision-runtime-ui03`：post-task sync、templateMatch intake 优先、goose 00030、Claim+idempotency | 合入前：schema30 **真实现卷**迁移独立验证；占槽真实 UI-03 重跑；勿把 stub 当验收 |
+| DOCREV-SSE-IDEM-BUSY | SSE WriteTimeout 断线后同 turn 再 POST 误 409 DOC_BUSY + 重复聊天 | **FIX-LANDED**（工程） | [audit](../audit/2026-09-13-document-revision-sse-idempotency.md)；证据 [assets/2026-09-13-document-revision-runtime-ui](../test-reports/assets/2026-09-13-document-revision-runtime-ui/)；`fix/document-revision-sse-idempotency` | 占槽真实 UI-03 重跑；不删证据 |

 ---

@@ -42,6 +42,26 @@ type StreamChatRequest struct {
 	ClientTurnID   string `json:"client_turn_id,omitempty"`
 }

+// 长任务 SSE：不取消全局 WriteTimeout，仅在流式响应上周期性延长写期限并发心跳。
+const (
+	sseWriteDeadlineExtend = 45 * time.Second
+	sseHeartbeatInterval   = 15 * time.Second
+)
+
+func extendSSEWriteDeadline(w http.ResponseWriter) {
+	rc := http.NewResponseController(w)
+	_ = rc.SetWriteDeadline(time.Now().Add(sseWriteDeadlineExtend))
+}
+
+func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error {
+	extendSSEWriteDeadline(w)
+	if _, err := fmt.Fprintf(w, ": heartbeat %d\n\n", time.Now().Unix()); err != nil {
+		return err
+	}
+	flusher.Flush()
+	return nil
+}
+
 func writeSSE(w http.ResponseWriter, flusher http.Flusher, ev stream.StreamEvent) {
 	data, err := json.Marshal(ev)
 	if err != nil {
@@ -76,6 +76,29 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) {
 	}
 	req.WorkDir = ""

+	// 授权后、scope/preflight 前：稳定幂等键命中则直接 202，避免 SSE 断线重试误击 DOC_BUSY / 重复写聊天。
+	idemKey := taskrun.BuildTaskRunIdempotencyKey(req.SessionID, req.ClientTurnID)
+	if idemKey != "" && s.taskRuns != nil {
+		if existing, findErr := s.taskRuns.FindByIdempotencyKey(r.Context(), idemKey); findErr != nil {
+			http.Error(w, findErr.Error(), http.StatusInternalServerError)
+			return
+		} else if existing != nil {
+			if !taskRunMatchesIdempotentOwner(existing, req) {
+				writeJSONResponse(w, http.StatusForbidden, map[string]string{
+					"error_code": "ERR_FORBIDDEN",
+					"message":    "幂等键已存在但不属于当前用户/会话/项目",
+				})
+				return
+			}
+			writeJSONResponse(w, http.StatusAccepted, createTaskRunResponse{
+				RunID:       existing.RunID,
+				WorkspaceID: existing.WorkspaceID,
+				Status:      existing.Status,
+			})
+			return
+		}
+	}
+
 	ctxMgr := s.svc.ContextManager()
 	if ctxMgr != nil {
 		_, _ = ctxMgr.EnsureChatSession(r.Context(), req.SessionID, req.UserID, req.ProjectID, req.ProjectName)
@@ -104,7 +127,6 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) {
 	}

 	// 幂等两阶段：先 Claim 占位，仅 owner 写聊天并 Start；loser 不得写消息。
-	idemKey := taskrun.BuildTaskRunIdempotencyKey(req.SessionID, req.ClientTurnID)
 	runID := fmt.Sprintf("run_%d", time.Now().UnixNano())
 	record := &taskrun.RunRecord{
 		RunID:              runID,
@@ -184,6 +206,16 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) {
 	})
 }

+func taskRunMatchesIdempotentOwner(existing *taskrun.RunRecord, req createTaskRunRequest) bool {
+	if existing == nil {
+		return false
+	}
+	return existing.UserID == req.UserID &&
+		existing.SessionID == req.SessionID &&
+		existing.ProjectID == req.ProjectID &&
+		existing.ClientTurnID == req.ClientTurnID
+}
+
 func (s *Server) recordBusyPreflightMessages(
 	ctx context.Context,
 	ctxMgr *ctxmgr.ContextManager,
@@ -310,6 +342,7 @@ func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) {
 	s.writeCORS(w, r)
 	w.Header().Set("X-Accel-Buffering", "no")
 	flusher.Flush()
+	extendSSEWriteDeadline(w)

 	ch, replay, err := s.taskRuns.Subscribe(runID, afterSeq)
 	if err != nil {
@@ -318,26 +351,48 @@ func (s *Server) handleTaskRunEvents(w http.ResponseWriter, r *http.Request) {
 	}
 	defer s.taskRuns.Unsubscribe(runID, ch)

-	writeStoredEvents := func(events []taskrun.StoredEvent) {
+	writeStoredEvents := func(events []taskrun.StoredEvent) error {
 		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))
+			if _, err := fmt.Fprintf(w, "id: %d\n", stored.Seq); err != nil {
+				return err
+			}
+			if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", stored.Event.Type, string(data)); err != nil {
+				return err
+			}
 		}
 		flusher.Flush()
+		extendSSEWriteDeadline(w)
+		return nil
 	}

-	writeStoredEvents(replay)
+	if err := writeStoredEvents(replay); err != nil {
+		return
+	}
+	for _, stored := range replay {
+		if stored.Event.Type == stream.EventDone || stored.Event.Type == stream.EventError {
+			return
+		}
+	}
+
+	heartbeat := time.NewTicker(sseHeartbeatInterval)
+	defer heartbeat.Stop()

 	for {
 		select {
 		case <-r.Context().Done():
 			return
+		case <-heartbeat.C:
+			if err := writeSSEHeartbeat(w, flusher); err != nil {
+				return
+			}
 		case stored, ok := <-ch:
 			if !ok {
 				return
 			}
-			writeStoredEvents([]taskrun.StoredEvent{stored})
+			if err := writeStoredEvents([]taskrun.StoredEvent{stored}); err != nil {
+				return
+			}
 			if stored.Event.Type == stream.EventDone || stored.Event.Type == stream.EventError {
 				return
 			}
@@ -0,0 +1,197 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/stream"
+)
+
+func TestCreateTaskRun_SameTurnIdempotentReturnsExistingDespiteBusyLease(t *testing.T) {
+	srv := newTestServer(t)
+	ctx := t.Context()
+	token := loginTestAdmin(t, srv)
+	userID := int64(1)
+
+	// 慢执行器：保持 run 在内存中可被 FindByIdempotencyKey 命中
+	block := make(chan struct{})
+	t.Cleanup(func() { close(block) })
+	srv.taskRuns.SetExecutor(
+		func(ctx context.Context, _ string, _ bizdev.TaskRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
+			select {
+			case <-ctx.Done():
+				return &bizdev.TaskRunResult{Success: false}, ctx.Err()
+			case <-block:
+				return &bizdev.TaskRunResult{Success: true}, nil
+			}
+		},
+		nil,
+		func(ctx context.Context, _ string, _ bizdev.StageRunRequest, _ *stream.Streamer) (*bizdev.TaskRunResult, error) {
+			return &bizdev.TaskRunResult{Success: true}, nil
+		},
+	)
+
+	docID := "doc_proj143_intake"
+	if _, err := srv.svc.DocGov().CreateOrGetDocument(
+		ctx, docID, 143, "intake", "用户初始需求收集表", "V0.1_triage", "",
+		userID, "编制人", userID, "审核人", userID, "批准人",
+	); err != nil {
+		t.Fatalf("create document: %v", err)
+	}
+
+	body1, _ := json.Marshal(createTaskRunRequest{
+		TaskRunRequestDTO: TaskRunRequestDTO{TaskKey: "intake", ProjectID: 143, ProjectName: "幂等项目", DocID: docID},
+		SessionID:         "sess-idem-busy",
+		ClientTurnID:      "turn-same-1",
+		UserMessage:       "请编制首轮摸底",
+	})
+	req1 := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/runs", bytes.NewReader(body1))
+	withBearerAuth(req1, token)
+	w1 := httptest.NewRecorder()
+	srv.Routes().ServeHTTP(w1, req1)
+	if w1.Code != http.StatusAccepted {
+		t.Fatalf("first create: %d %s", w1.Code, w1.Body.String())
+	}
+	var first createTaskRunResponse
+	if err := json.Unmarshal(w1.Body.Bytes(), &first); err != nil {
+		t.Fatalf("decode first: %v", err)
+	}
+	if first.RunID == "" {
+		t.Fatal("missing run_id")
+	}
+
+	hist1, err := srv.svc.ContextManager().GetChatHistory(ctx, "sess-idem-busy", 50, 0)
+	if err != nil {
+		t.Fatalf("history1: %v", err)
+	}
+	if len(hist1) != 2 {
+		t.Fatalf("after first create history=%d want 2", len(hist1))
+	}
+
+	// 活跃小队租约：模拟长任务仍占文档（旧路径会 409）
+	if _, err := srv.svc.DocGov().AcquireLease(ctx, docgov.LeaseAcquireRequest{
+		DocID: docID, HolderType: docgov.LeaseHolderSquad, HolderID: first.RunID, TTL: time.Minute,
+	}); err != nil {
+		t.Fatalf("acquire squad lease: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = srv.svc.DocGov().ReleaseLease(context.Background(), docID, string(docgov.LeaseHolderSquad), first.RunID)
+	})
+
+	body2, _ := json.Marshal(createTaskRunRequest{
+		TaskRunRequestDTO: TaskRunRequestDTO{TaskKey: "intake", ProjectID: 143, ProjectName: "幂等项目", DocID: docID},
+		SessionID:         "sess-idem-busy",
+		ClientTurnID:      "turn-same-1",
+		UserMessage:       "请编制首轮摸底",
+	})
+	req2 := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/runs", bytes.NewReader(body2))
+	withBearerAuth(req2, token)
+	w2 := httptest.NewRecorder()
+	srv.Routes().ServeHTTP(w2, req2)
+	if w2.Code != http.StatusAccepted {
+		t.Fatalf("same-turn retry want 202, got %d %s", w2.Code, w2.Body.String())
+	}
+	var second createTaskRunResponse
+	_ = json.Unmarshal(w2.Body.Bytes(), &second)
+	if second.RunID != first.RunID {
+		t.Fatalf("want same run %s, got %s", first.RunID, second.RunID)
+	}
+
+	hist2, err := srv.svc.ContextManager().GetChatHistory(ctx, "sess-idem-busy", 50, 0)
+	if err != nil {
+		t.Fatalf("history2: %v", err)
+	}
+	if len(hist2) != 2 {
+		t.Fatalf("idempotent retry must not append chat; history=%d", len(hist2))
+	}
+
+	// 不同 turn：仍应 DOC_BUSY 409
+	body3, _ := json.Marshal(createTaskRunRequest{
+		TaskRunRequestDTO: TaskRunRequestDTO{TaskKey: "intake", ProjectID: 143, ProjectName: "幂等项目", DocID: docID},
+		SessionID:         "sess-idem-busy",
+		ClientTurnID:      "turn-other-2",
+		UserMessage:       "另一轮摸底",
+	})
+	req3 := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/runs", bytes.NewReader(body3))
+	withBearerAuth(req3, token)
+	w3 := httptest.NewRecorder()
+	srv.Routes().ServeHTTP(w3, req3)
+	if w3.Code != http.StatusConflict {
+		t.Fatalf("other turn want 409, got %d %s", w3.Code, w3.Body.String())
+	}
+	var busy map[string]any
+	_ = json.Unmarshal(w3.Body.Bytes(), &busy)
+	if busy["error_code"] != "DOC_BUSY" {
+		t.Fatalf("want DOC_BUSY, got %+v", busy)
+	}
+}
+
+func TestSSEHeartbeat_ExtendsPastServerWriteTimeout(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/sse", func(w http.ResponseWriter, r *http.Request) {
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			http.Error(w, "no flush", 500)
+			return
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		flusher.Flush()
+		extendSSEWriteDeadline(w)
+		tick := time.NewTicker(60 * time.Millisecond)
+		defer tick.Stop()
+		end := time.After(700 * time.Millisecond)
+		for {
+			select {
+			case <-end:
+				_, _ = fmt.Fprintf(w, "event: done\ndata: {\"type\":\"done\"}\n\n")
+				flusher.Flush()
+				return
+			case <-tick.C:
+				if err := writeSSEHeartbeat(w, flusher); err != nil {
+					return
+				}
+			case <-r.Context().Done():
+				return
+			}
+		}
+	})
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	srv := &http.Server{
+		Handler:      mux,
+		WriteTimeout: 200 * time.Millisecond, // 短于流时长；无 extend 会断
+	}
+	go func() { _ = srv.Serve(ln) }()
+	t.Cleanup(func() { _ = srv.Close() })
+
+	resp, err := http.Get("http://" + ln.Addr().String() + "/sse")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	raw, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("read body: %v", err)
+	}
+	body := string(raw)
+	if !strings.Contains(body, "heartbeat") {
+		t.Fatalf("expected heartbeats, body=%q", body)
+	}
+	if !strings.Contains(body, "event: done") {
+		t.Fatalf("stream should survive WriteTimeout via deadline extend; body=%q", body)
+	}
+}
@@ -26,3 +26,10 @@ test('parseSSEChunk: flush parses final frame without trailing delimiter', () =>
  assert.equal(events[0].type, 'done');
  assert.equal(events[0].done?.session_id, 's1');
 });
+
+test('parseSSEChunk: ignores heartbeat comments and keeps last id', () => {
+  const raw = 'id: 1\nevent: status\ndata: {"type":"status"}\n\n: heartbeat 9\n\nid: 2\nevent: delta\ndata: {"type":"delta","delta":"z"}\n\n';
+  const { events, lastSeq } = parseSSEChunk(raw);
+  assert.equal(events.length, 2);
+  assert.equal(lastSeq, 2);
+});
@@ -102,24 +102,30 @@ function parseActions(raw?: string[]): ChatErrorInfo['actions'] {
  return raw.filter((a): a is ErrorAction => (VALID_ACTIONS as readonly string[]).includes(a));
 }

-function parseFrame(frame: string): StreamEvent | null {
-  if (!frame.trim()) return null;
+function parseFrame(frame: string): { event: StreamEvent | null; seq?: number } {
+  if (!frame.trim()) return { event: null };

  let dataLine: string | undefined;
+  let seq: number | undefined;
  for (const line of frame.split('\n')) {
+    if (line.startsWith('id:')) {
+      const n = Number(line.slice(3).trim());
+      if (Number.isFinite(n)) seq = n;
+    }
    if (line.startsWith('data:')) {
      dataLine = line.slice(5).trim();
    }
  }
-  if (!dataLine) return null;
+  // SSE comment / heartbeat frames (": ...") have no data — ignore.
+  if (!dataLine) return { event: null, seq };

-  return JSON.parse(dataLine) as StreamEvent;
+  return { event: JSON.parse(dataLine) as StreamEvent, seq };
 }

 export function parseSSEChunk(
  buffer: string,
  options?: { flush?: boolean },
-): { events: StreamEvent[]; rest: string } {
+): { events: StreamEvent[]; rest: string; lastSeq?: number } {
  const events: StreamEvent[] = [];
  if (!buffer) return { events, rest: '' };

@@ -131,13 +137,15 @@ export function parseSSEChunk(
    processCount = frames.length;
  }

+  let lastSeq: number | undefined;
  for (let i = 0; i < processCount; i++) {
-    const ev = parseFrame(frames[i]);
-    if (ev) events.push(ev);
+    const parsed = parseFrame(frames[i]);
+    if (parsed.seq !== undefined) lastSeq = parsed.seq;
+    if (parsed.event) events.push(parsed.event);
  }

  const rest = hasTrailingDelimiter || options?.flush ? '' : (frames[frames.length - 1] ?? '');
-  return { events, rest };
+  return { events, rest, lastSeq };
 }

 export async function streamChat(
@@ -1,6 +1,7 @@
 import test from 'node:test';
 import assert from 'node:assert/strict';
-import { createTaskRun } from './taskRuns';
+import { createTaskRun, readTaskRunSSEOnce, subscribeTaskRunEvents } from './taskRuns';
+import { parseSSEChunk } from './chatStream';

 test('createTaskRun exposes DOC_BUSY detail without a retryable run', async () => {
  const originalFetch = globalThis.fetch;
@@ -29,3 +30,73 @@ test('createTaskRun exposes DOC_BUSY detail without a retryable run', async () =
    globalThis.fetch = originalFetch;
  }
 });
+
+test('parseSSEChunk tracks id seq for after_seq reconnect', () => {
+  const raw =
+    'id: 7\nevent: phase\ndata: {"type":"phase","phase":{"phase_id":"x","message":"m"}}\n\n' +
+    ': heartbeat 1\n\n' +
+    'id: 8\nevent: delta\ndata: {"type":"delta","delta":"a"}\n\n';
+  const { events, lastSeq } = parseSSEChunk(raw);
+  assert.equal(events.length, 2);
+  assert.equal(lastSeq, 8);
+});
+
+test('subscribeTaskRunEvents reconnects on EOF with advancing after_seq', async () => {
+  const originalFetch = globalThis.fetch;
+  const calls: string[] = [];
+  let round = 0;
+  globalThis.fetch = async (input: RequestInfo | URL) => {
+    const url = String(input);
+    calls.push(url);
+    round += 1;
+    if (round === 1) {
+      // Unexpected EOF after seq 3 (no done/error)
+      const body = 'id: 3\nevent: phase\ndata: {"type":"phase","phase":{"phase_id":"a","message":"m"}}\n\n';
+      return new Response(body, {
+        status: 200,
+        headers: { 'Content-Type': 'text/event-stream' },
+      });
+    }
+    assert.match(url, /after_seq=3/);
+    const body =
+      'id: 4\nevent: done\ndata: {"type":"done","done":{"session_id":"s","workspace_id":"w"}}\n\n';
+    return new Response(body, {
+      status: 200,
+      headers: { 'Content-Type': 'text/event-stream' },
+    });
+  };
+  try {
+    const seen: string[] = [];
+    await subscribeTaskRunEvents(
+      'run_reconnect',
+      {
+        onEvent: (ev) => seen.push(String(ev.type)),
+        maxReconnects: 3,
+        reconnectDelayMs: 1,
+        sleep: async () => undefined,
+      },
+    );
+    assert.deepEqual(seen, ['phase', 'done']);
+    assert.equal(calls.length, 2);
+    assert.match(calls[0], /after_seq=0/);
+    assert.match(calls[1], /after_seq=3/);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test('readTaskRunSSEOnce marks non-terminal EOF for reconnect', async () => {
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = async () =>
+    new Response('id: 2\nevent: status\ndata: {"type":"status","status":{"message":"x"}}\n\n', {
+      status: 200,
+      headers: { 'Content-Type': 'text/event-stream' },
+    });
+  try {
+    const { terminal, lastSeq } = await readTaskRunSSEOnce('run_x', 0, () => undefined);
+    assert.equal(terminal, false);
+    assert.equal(lastSeq, 2);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
@@ -46,13 +46,22 @@ export async function createTaskRun(body: CreateTaskRunBody): Promise<CreateTask
  return res.json();
 }

-export async function subscribeTaskRunEvents(
+function isTerminalTaskEvent(ev: StreamEvent): boolean {
+  return ev.type === 'done' || ev.type === 'error';
+}
+
+export function sleepMs(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/** 单次 SSE 读取；意外 EOF 时返回 { terminal:false, lastSeq } 供上层重连。 */
+export async function readTaskRunSSEOnce(
  runId: string,
-  handlers: { onEvent: (ev: StreamEvent) => void; afterSeq?: number },
+  afterSeq: number,
+  onEvent: (ev: StreamEvent) => void,
  signal?: AbortSignal,
-): Promise<void> {
-  const after = handlers.afterSeq ?? 0;
-  const url = `/api/v1/tasks/runs/${encodeURIComponent(runId)}/events?after_seq=${after}`;
+): Promise<{ terminal: boolean; lastSeq: number }> {
+  const url = `/api/v1/tasks/runs/${encodeURIComponent(runId)}/events?after_seq=${afterSeq}`;
  const resp = await fetch(url, {
    method: 'GET',
    headers: mergeAuthHeaders({ Accept: 'text/event-stream' }),
@@ -64,17 +73,69 @@ export async function subscribeTaskRunEvents(
  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
+  let lastSeq = afterSeq;
+  let terminal = false;
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const parsed = parseSSEChunk(buf);
    buf = parsed.rest;
-    for (const ev of parsed.events) handlers.onEvent(ev);
+    if (parsed.lastSeq !== undefined) {
+      lastSeq = Math.max(lastSeq, parsed.lastSeq);
+    }
+    for (const ev of parsed.events) {
+      onEvent(ev);
+      if (isTerminalTaskEvent(ev)) terminal = true;
+    }
  }
  if (buf) {
    const flushed = parseSSEChunk(buf, { flush: true });
-    for (const ev of flushed.events) handlers.onEvent(ev);
+    if (flushed.lastSeq !== undefined) {
+      lastSeq = Math.max(lastSeq, flushed.lastSeq);
+    }
+    for (const ev of flushed.events) {
+      onEvent(ev);
+      if (isTerminalTaskEvent(ev)) terminal = true;
+    }
+  }
+  return { terminal, lastSeq };
+}
+
+/**
+ * 订阅任务 SSE：意外 EOF / 不完整 chunked 时按最后 seq 重订阅同一 run，禁止调用方据此 POST 新 run。
+ */
+export async function subscribeTaskRunEvents(
+  runId: string,
+  handlers: {
+    onEvent: (ev: StreamEvent) => void;
+    afterSeq?: number;
+    maxReconnects?: number;
+    reconnectDelayMs?: number;
+    sleep?: (ms: number) => Promise<void>;
+  },
+  signal?: AbortSignal,
+): Promise<void> {
+  let afterSeq = handlers.afterSeq ?? 0;
+  const maxReconnects = handlers.maxReconnects ?? 40;
+  const baseDelay = handlers.reconnectDelayMs ?? 500;
+  const sleep = handlers.sleep ?? sleepMs;
+
+  for (let attempt = 0; attempt <= maxReconnects; attempt++) {
+    if (signal?.aborted) return;
+    try {
+      const { terminal, lastSeq } = await readTaskRunSSEOnce(runId, afterSeq, handlers.onEvent, signal);
+      afterSeq = Math.max(afterSeq, lastSeq);
+      if (terminal) return;
+      // 意外 EOF：未见到 done/error，按 after_seq 重连
+    } catch (err) {
+      if (signal?.aborted) return;
+      if (attempt >= maxReconnects) throw err;
+    }
+    if (attempt >= maxReconnects) {
+      throw new Error(`task run SSE disconnected after ${maxReconnects} reconnects (after_seq=${afterSeq})`);
+    }
+    await sleep(Math.min(baseDelay * (attempt + 1), 5000));
  }
 }