From 43e03e4969917b364d8e461c55910bbb05931232 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 14:37:18 +0000 Subject: [PATCH] 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 --- ...09-13-document-revision-sse-idempotency.md | 46 ++++ .doc/ops/CURRENT.md | 12 +- ...09-13-document-revision-sse-idempotency.md | 46 ++++ .doc/remediation/gap-matrix.md | 1 + internal/server/sse.go | 20 ++ internal/server/task_run_handlers.go | 67 +++++- .../server/task_run_idempotent_sse_test.go | 197 ++++++++++++++++++ web/src/api/chatStream.test.ts | 7 + web/src/api/chatStream.ts | 24 ++- web/src/api/taskRuns.test.ts | 73 ++++++- web/src/api/taskRuns.ts | 75 ++++++- 11 files changed, 541 insertions(+), 27 deletions(-) create mode 100644 .doc/audit/2026-09-13-document-revision-sse-idempotency.md create mode 100644 .doc/ops/short-term/2026-09-13-document-revision-sse-idempotency.md create mode 100644 internal/server/task_run_idempotent_sse_test.go diff --git a/.doc/audit/2026-09-13-document-revision-sse-idempotency.md b/.doc/audit/2026-09-13-document-revision-sse-idempotency.md new file mode 100644 index 00000000..df1a1468 --- /dev/null +++ b/.doc/audit/2026-09-13-document-revision-sse-idempotency.md @@ -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`） diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 39a7943b..f5b6bcd4 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -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/` 或 `fix/`，**禁止** `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) | diff --git a/.doc/ops/short-term/2026-09-13-document-revision-sse-idempotency.md b/.doc/ops/short-term/2026-09-13-document-revision-sse-idempotency.md new file mode 100644 index 00000000..cefe97e0 --- /dev/null +++ b/.doc/ops/short-term/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/) diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index 9ed71c8b..70a5f984 100644 --- a/.doc/remediation/gap-matrix.md +++ b/.doc/remediation/gap-matrix.md @@ -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 重跑；不删证据 | --- diff --git a/internal/server/sse.go b/internal/server/sse.go index 3324df79..d19cbc8d 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -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 { diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 8b7fcbd2..3c53b0fa 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -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 } diff --git a/internal/server/task_run_idempotent_sse_test.go b/internal/server/task_run_idempotent_sse_test.go new file mode 100644 index 00000000..838c0e28 --- /dev/null +++ b/internal/server/task_run_idempotent_sse_test.go @@ -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) + } +} diff --git a/web/src/api/chatStream.test.ts b/web/src/api/chatStream.test.ts index d6a4148a..1320dce3 100644 --- a/web/src/api/chatStream.test.ts +++ b/web/src/api/chatStream.test.ts @@ -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); +}); diff --git a/web/src/api/chatStream.ts b/web/src/api/chatStream.ts index fe7716e4..70b921fa 100644 --- a/web/src/api/chatStream.ts +++ b/web/src/api/chatStream.ts @@ -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( diff --git a/web/src/api/taskRuns.test.ts b/web/src/api/taskRuns.test.ts index 602fba6e..a640e244 100644 --- a/web/src/api/taskRuns.test.ts +++ b/web/src/api/taskRuns.test.ts @@ -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; + } +}); diff --git a/web/src/api/taskRuns.ts b/web/src/api/taskRuns.ts index 5d3bad56..eb7a55d0 100644 --- a/web/src/api/taskRuns.ts +++ b/web/src/api/taskRuns.ts @@ -46,13 +46,22 @@ export async function createTaskRun(body: CreateTaskRunBody): Promise { + 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 { - 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; + }, + signal?: AbortSignal, +): Promise { + 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)); } }