From 76e2028761f41c616ff21005b3722094f06c9c18 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 13:39:07 +0000 Subject: [PATCH 1/5] fix(docgov): UI-03 post-task doc sync, claim idempotency, schema30 Restore controlled-doc merge after task terminal states, prioritize intake template match, ClaimRun before chat writes, and extend chk for pending_author_confirmation via goose 00030. Real volume migrate + UI re-run remain pre-merge gates. Co-authored-by: Cursor --- ...6-09-13-controlled-doc-client-sync-ui03.md | 106 +++++++++ .doc/ops/CURRENT.md | 6 +- .../2026-09-13-document-revision-runtime.md | 8 +- .doc/remediation/gap-matrix.md | 1 + ...3-task-run-pending-author-confirmation.sql | 42 ++++ internal/server/task_run_handlers.go | 59 +++-- internal/taskrun/claim_idempotency_test.go | 222 ++++++++++++++++++ internal/taskrun/manager.go | 181 ++++++++++++-- internal/taskrun/store.go | 70 +++++- internal/taskrun/types.go | 27 ++- web/e2e/document-revision-runtime.spec.ts | 35 ++- web/src/App.tsx | 108 ++++++++- web/src/engine/templateMatch.test.ts | 18 ++ web/src/engine/templateMatch.ts | 26 +- web/src/lib/postTaskDeliverableSync.test.ts | 90 +++++++ web/src/lib/postTaskDeliverableSync.ts | 65 +++++ 16 files changed, 976 insertions(+), 88 deletions(-) create mode 100644 .doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md create mode 100644 deploy/postgres/goose/00030_2026-09-13-task-run-pending-author-confirmation.sql create mode 100644 internal/taskrun/claim_idempotency_test.go create mode 100644 web/src/lib/postTaskDeliverableSync.test.ts create mode 100644 web/src/lib/postTaskDeliverableSync.ts diff --git a/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md new file mode 100644 index 00000000..024e1d5e --- /dev/null +++ b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md @@ -0,0 +1,106 @@ +--- +title: "小队完成未同步受控文档状态 + UI-03 选择器非 Tab + Retry 暴露 DB 约束（UI-03）" +status: done +created: 2026-09-13 +updated: 2026-09-13 +severity: critical +--- + +# 小队完成未同步受控文档状态 + UI-03 选择器非 Tab + Retry 暴露 DB 约束（UI-03） + +## 1. 现场实战概况与判定 + +在真实 UI 验收（`UI_PHASE=main`，dev 环境 `schema 29`，真实 `pi-agent` + `MiniMax-M2.7-highspeed`）： + +- **UI-00 PASS**（4.9s）：管理员真实创建 A/B/C，准确按 DOM 作用域关闭模态，三用户完成首次登录改密。 +- **UI-01 PASS**（774ms）：A 页面立项并分配 B 为 reviewer，项目工作台就绪，零 404，模板预览正确。 +- **UI-02 PASS**（191ms）：真实发任务，`POST /api/v1/tasks/runs` 返回 202 Accepted，小队编排启动；真实模型耗时约 157 秒完成编排，后端成功生成受控文档 `doc_proj6089_user_requirements`（质量评分 100/100）。 +- **UI-03 FAIL**：`await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 60_000 });` 超时失败。 + - **页面真实状态**：页面已明确显示小队任务完成，正文标出 `📄 受控文档：doc_proj6089_user_requirements`。 + - **右栏面板状态**：右栏仍保持为 `doc-p intake` 初始模板（`用户初始需求收集表 模板`，`模板预览：首轮编制后启用版本账本`），版本工具条 `[data-testid="doc-rev-version"]` 根本未挂载。 +- **UI-04 ~ UI-16**：**NOT RUN**（因 UI-03 失败立即停机保全现场，未执行后续用例）。 + +--- + +## 2. 根因与现场异常分析 + +### A. 产品缺陷：前端在小队完成后未同步工作流文档列表 + +1. **服务端已落库受控文档**： + 后端 `document_records` 表已落库 `doc_id="doc_proj6089_user_requirements"`，`is_template_only=false`，`current_state="drafting"`。且前端每 5 秒发起的轮询 `GET /api/v1/projects/6089/documents?category=workflow` 已正确返回该受控文档记录。 +2. **`syncLeases` 轮询丢弃新文档**： + 在 `web/src/App.tsx` 第 1664–1697 行，每 5 秒执行的 `syncLeases` 仅提取 `doc.lease_holder_type` 并更新既有 `doc.docId` 的租约。由于本地 React 状态 `deliverables` 初始仅有 `doc-p6089-*` 模板，`leases.has(doc.docId)` 匹配不上 `doc_proj*`，新文档从未被合并进 `deliverables`。 +3. **`syncWorkflowDocs` 依赖项未触发**： + 在 `App.tsx` 第 1618–1661 行，`syncWorkflowDocs`（调用 `mergeWorkflowDocumentsFromServer`）仅在 `currentProjectId` 或 `session` 变动时执行；同一会话内小队完成不会触发该 effect。 +4. **流式结束 `case 'done'` 遗漏文档同步**： + 在 `App.tsx` 第 2571–2593 行，任务流完成时仅调用了 `await chatWindow.refresh().catch(() => undefined)`（仅刷新聊天消息），未同步工作流文档列表。 +5. **连锁阻断工具条挂载**： + 前端 `deliverables` 中的 `user_requirements` 一直是 `isTemplateOnly: true`。根据 `revisionToolbarMountMode` 契约，`isTemplateOnly === true` 时返回 `'template_hint'`，阻断 `DocumentRevisionToolbar`（`[data-testid="doc-rev-version"]`）挂载。 + +### B. 测试脚本缺陷：定位器假设水平 Tab 且命中页头 Banner 徽章 + +1. 在 `web/e2e/document-revision-runtime.spec.ts` 第 538 行： + `const tab = page.locator('[role="tab"], button, div').filter({ hasText: /需求|摸底|URS|用户/ }).first();` +2. **穿透到页头徽章**： + `[role="tab"], button, div` 包含全局所有 `div`。页面顶部 Header Banner 含有 `当前项目: ... V0.1 需求摸底`（`div` 元素）。`.first()` 命中了该非交互的 Banner 徽章 `div`，根本没有点击文档切换器。 +3. **真实 UI 为下拉/抽屉交互**： + `SlimTabBar.tsx` 为面包屑下拉（`button.dropdown-trigger` -> `.dropdown-menu`）或大盘抽屉（`DeckDrawer`），页面不存在水平文档 Tab。 + +### C. Playwright 串行 Retry 现象与 DB 约束异常暴露 + +1. **Playwright 串行重试导致任务重发**： + Trace 与聊天现场显示：同一项目下，13:02 成功执行并完成首轮任务后，因重试机制或步骤重入，13:05 再次向同一个会话发送了相同的首轮指令： + - 13:02：`TRIGGER_TASK 请基于本项目编制首轮...` → `《用户需求表》已生成。当前状态：已完成` + - 13:05：`TRIGGER_TASK 请基于本项目编制首轮...` → 触发二次运行，助手再次进入处理中 +2. **暴露 `task_runs` 状态约束缺陷**： + 在二次运行与高风险候选流转路径中，后台服务尝试将 `task_runs.status` 更新为 `pending_author_confirmation` 时，暴露了数据库底层约束拦截： + `task_runs` 表定义中的 `chk_task_run_status` 仅允许： + ```sql + CHECK (status IN ('queued', 'running', 'completed', 'failed', 'cancelled', 'quality_failed')) + ``` + 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规。 + +--- + +## 3. 严重度与阻断 + +- **严重度**：`critical` +- **阻断项**： + 1. 产品前端：`App.tsx` 在小队完成（`case 'done'`）以及定期轮询中必须调用 `mergeWorkflowDocumentsFromServer` 动态更新受控文档。 + 2. 数据库约束：`task_runs` 表的 `chk_task_run_status` 约束必须补齐 `pending_author_confirmation`。 + 3. 测试脚本：修正 UI-03 文档切换定位器，禁止使用全局 `div` 通配，走真实 `SlimTabBar` 下拉或 `DeckDrawer` 动作；关闭非幂等的单用例重试。 + +--- + +## 4. 证据保全清单 + +- **归档路径**：`.doc/test-reports/assets/2026-09-13-document-revision-runtime-ui/ui03-controlled-doc-sync-failure/` + - `playwright-output/`（含完整 `trace.zip`、`test-failed-1.png`、`error-context.md`） + - `playwright-report/`（HTML 完整执行报告） + - `screenshots/`（UI-00~UI-02 真实过程截图） + - `traces/`、`videos/` + - `ui-acceptance-results.json`（UI-00~UI-02 逐项 PASS 记录） + - `phase-secrets.local.json`、`phase-state.json` +- 历史归档目录全部原位保全： + - `post-auth-ui00-backdrop-click-failure/` + - `post-auth-ui00-close-selector-failure/` + - `post-auth-ui00-wrong-close-locator-failure/` + - `pre-auth-fix-ui00-failure/` + - `ui01-parent-directory-sync-failure/` + - `ui01-strict-mode-title-selector-failure/` + - `ui02-task-scope-409-failure/` + - `ui02-workflow-access-failure/` + +--- + +## 5. 工程修复落盘（`fix/document-revision-runtime-ui03`，合入前） + +| 根因 | 修复 | 门禁 | +|---|---|---| +| A 任务完成后未 merge 受控文档 | `postTaskDeliverableSync` + `App.tsx` done/pending/reconnect 后 `listProjectWorkflowDocuments` → merge → 切到 `doc_proj*` | `npm test` 232 PASS；`npm run build` PASS | +| A′ intake vs URS 误匹配 | `templateMatch`：**intake 标记优先**于裸「用户需求」 | `templateMatch` + sync 单测（`npx tsx --test` 18/18） | +| B UI-03 选择器命中 Banner | E2E 改为 SlimTabBar/DeckDrawer 精确选择器；要求服务端 `doc_proj*` | 待真实 UI 重跑 | +| C `pending_author_confirmation` 约束 | goose `00030` 扩展 `chk_task_run_status` | **真实现卷迁移由合入前独立验证**（本提交不跑复杂 PG 集成测） | +| C′ 重试重复建 run / 重复写 chat | `ClaimRun` 先占 `idempotency_key`；非 owner 直接 202；FE reconnect 同 `run_id` 订阅 | `go test -short ./internal/taskrun ./internal/server` PASS | + +**未做**：占槽 rebuild、真实 UI UI-03 重验、schema30 对现有重复 `(session_id,client_turn_id)` 卷的迁移实测。 diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index d2c80515..39a7943b 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 PASS；UI-01/02 工程修复已入 fix 分支：governing 实例 + chat 项目门禁 + 立项成员落库；待重跑真实 UI） +**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 待合入前验证） ## 焦点 -1. **DOCREV-RT**: [运行时整改收口](short-term/2026-09-13-document-revision-runtime.md)（`fix/document-revision-runtime`：上传注册 + auth 映射 + 用户目录同步 + 无 live/chat 授权 + 立项成员持久化；占槽真实 UI E2E；不 push） +1. **DOCREV-RT**: [运行时整改收口](short-term/2026-09-13-document-revision-runtime.md)（`fix/document-revision-runtime-ui03`：UI-03 客户端同步/幂等/chk；占槽真实 UI 重跑；不 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,7 @@ | 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 | -| Codex + Herdr(Cursor/Antigravity) | document-revision-runtime | fix/document-revision-runtime | - | docgov | in_progress | +| Cursor-Auto | document-revision-runtime-ui03 | fix/document-revision-runtime-ui03 | - | docgov | in_progress | > 新任务开始前登记。`branch` 必须为 `feature/` 或 `fix/`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。 diff --git a/.doc/ops/short-term/2026-09-13-document-revision-runtime.md b/.doc/ops/short-term/2026-09-13-document-revision-runtime.md index 4a499709..4d6ca355 100644 --- a/.doc/ops/short-term/2026-09-13-document-revision-runtime.md +++ b/.doc/ops/short-term/2026-09-13-document-revision-runtime.md @@ -4,7 +4,7 @@ status: in_progress created: 2026-09-13 updated: 2026-09-13 owner: Cursor-Auto -branch: fix/document-revision-runtime +branch: fix/document-revision-runtime-ui03 worktree: "-" mutex: docgov --- @@ -87,16 +87,18 @@ UI-01/02 访问与成员落库门禁（2026-09-13）：`go test ./internal/proje - P：**无 live 访问 + chat 项目门禁 + 立项成员落库（UI-01/02）**：`GetGoverningInstanceForAccess` 区分从未激活 vs completed/terminated；chat/chatStream 先 `RequireProjectAccess` 再 stage（SSE 拒绝不得先 200）；`createProject` 后持久化非创建者成员。详见 [../../audit/2026-09-13-workflow-access-chat-members-ui01-02.md](../../audit/2026-09-13-workflow-access-chat-members-ui01-02.md) - Q：**立项确认双击竞态**：`handleCreateProjectConfirm` 加同步 in-flight `useRef` 锁 + `createProjectBootstrapIdRef`（取代 state）；`createProjectSubmitting`/`submitPhase` 驱动 Modal「立项中/同步成员中」、禁用确认与关闭；失败可串行幂等重试，成功清 bootstrap。证据：`createProjectSubmitGuard.test.ts` - R：**UI-02 任务作用域契约**：`ResolveMessageScopeForTask` 三态（preactivation / live stage / closed→Err）；handler 无 live 创建 202 非 409；E2E UI-02 以 `POST /api/v1/tasks/runs` 2xx 为硬门禁（409/连接异常 FAIL）。证据：`message_scope_test.go`、`TestCreateTaskRun_PreActivationAcceptedNot409`、`document-revision-runtime.spec.ts` UI-02 +- S：**UI-03 客户端同步 + 任务幂等（本分支）**：`templateMatch` intake 标记优先；任务终态后 `postTaskDeliverableSync` merge `doc_proj*`；goose `00030` 放行 `pending_author_confirmation` + `idempotency_key` 部分唯一；`ClaimRun` 先占键再写 chat；FE reconnect 同 run 订阅。审计：[../../audit/2026-09-13-controlled-doc-client-sync-ui03.md](../../audit/2026-09-13-controlled-doc-client-sync-ui03.md)。门禁：`go test -short ./internal/taskrun ./internal/server`；`web npm test` 232+27；`npm run build`。**schema30 真实现卷迁移合入前独立验证**；真实 UI-03 待占槽重跑。 ### 用户 UI 实战验收 -见扩展 checklist（UI-01…UI-16）。Antigravity：UI-00 PASS；UI-01 目录死 prop 已修；UI-02 暴露无 live 误拒 + 成员未落库（本轮工程修复，待重跑 UI）。 +见扩展 checklist（UI-01…UI-16）。Antigravity：UI-00/01/02 PASS；UI-03 FAIL（客户端未 sync + 选择器 + DB chk）→ 工程修复已落本分支，待重跑。 ### Blockers - **DOCREV-RT-PI-NARRATIVE**：真实 PI 业务叙事摘要/风险解释（确定性 fallback 已可见，不得宣称为完整 PI） - **DOCREV-RT-UPLOAD-ORPHAN**：仅 IndexedDB 上传、无服务端正文 → 禁止假 V1 -- **DOCREV-RT-UI-E2E**：需占集成槽 + 真实小队 + `npm run test:ui-acceptance` 无头 Playwright 实录 +- **DOCREV-RT-UI-E2E**：需占集成槽 + 真实小队 + `npm run test:ui-acceptance` 无头 Playwright 实录（含 UI-03 重验） +- **DOCREV-RT-SCHEMA30-VOLUME**：goose 00030 对既有重复 `(session,client_turn)` / NULL idempotency 卷的迁移由合入前独立验证 - **DOCREV-RT-USER-DIR-CROSS-DEVICE**：立项 `availableUsers` 仍依赖本机 `allUsers`/localStorage；非 admin 无成员目录 API；跨设备/新浏览器可能仍缺 B（见 [../../audit/2026-09-13-admin-user-directory-sync-ui01.md](../../audit/2026-09-13-admin-user-directory-sync-ui01.md)） ### Commits（历史） diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index 11157429..9ed71c8b 100644 --- a/.doc/remediation/gap-matrix.md +++ b/.doc/remediation/gap-matrix.md @@ -61,6 +61,7 @@ | CHAT-INTENT-STATE | 真实会话被诊断探针污染；只读查询误启动文档写任务；错误动作与编辑状态刷新后丢失 | **FAIL** | [audit](../audit/2026-09-13-conversation-intent-runtime-state.md)；dev `sess-p143-u9` / task run / lease 现场 | [remediation](./2026-09-13-conversation-intent-runtime-state-remediation.md) | | 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 当验收 | --- diff --git a/deploy/postgres/goose/00030_2026-09-13-task-run-pending-author-confirmation.sql b/deploy/postgres/goose/00030_2026-09-13-task-run-pending-author-confirmation.sql new file mode 100644 index 00000000..bacac9c3 --- /dev/null +++ b/deploy/postgres/goose/00030_2026-09-13-task-run-pending-author-confirmation.sql @@ -0,0 +1,42 @@ +-- +goose Up +-- Allow pending_author_confirmation; idempotent creates via nullable idempotency_key. +-- Do NOT unique-index (session_id, client_turn_id): existing volumes already have duplicates. + +ALTER TABLE task_runs DROP CONSTRAINT IF EXISTS chk_task_run_status; + +ALTER TABLE task_runs ADD CONSTRAINT chk_task_run_status CHECK ( + status IN ( + 'pending', + 'running', + 'completed', + 'failed', + 'cancelled', + 'finalizing_failed', + 'quality_failed', + 'interrupted', + 'pending_author_confirmation' + ) +); + +-- Only new keyed rows participate; legacy NULL rows (incl. historical duplicates) untouched. +CREATE UNIQUE INDEX IF NOT EXISTS uq_task_runs_idempotency_key + ON task_runs (idempotency_key) + WHERE idempotency_key IS NOT NULL; + +-- +goose Down +DROP INDEX IF EXISTS uq_task_runs_idempotency_key; + +ALTER TABLE task_runs DROP CONSTRAINT IF EXISTS chk_task_run_status; + +ALTER TABLE task_runs ADD CONSTRAINT chk_task_run_status CHECK ( + status IN ( + 'pending', + 'running', + 'completed', + 'failed', + 'cancelled', + 'finalizing_failed', + 'quality_failed', + 'interrupted' + ) +); diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 3bbec1a0..8b7fcbd2 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -103,7 +103,36 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { return } + // 幂等两阶段：先 Claim 占位，仅 owner 写聊天并 Start；loser 不得写消息。 + idemKey := taskrun.BuildTaskRunIdempotencyKey(req.SessionID, req.ClientTurnID) 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, + IdempotencyKey: idemKey, + TaskKey: req.TaskKey, + Stage: req.Stage, + WorkflowInstanceID: taskScope.WorkflowInstanceID, + StageInstanceID: taskScope.StageInstanceID, + WorkflowStageKey: taskScope.StageKey, + Status: "pending", + } + claimed, created, claimErr := s.taskRuns.ClaimRun(r.Context(), record) + if claimErr != nil { + http.Error(w, claimErr.Error(), http.StatusInternalServerError) + return + } + if !created { + writeJSONResponse(w, http.StatusAccepted, createTaskRunResponse{ + RunID: claimed.RunID, + Status: claimed.Status, + }) + return + } + var assistantMsgID int64 if ctxMgr != nil && req.UserMessage != "" { userMeta := map[string]any{ @@ -114,7 +143,7 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { projectinstance.ApplyScopeToMetadata(userMeta, taskScope) _, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, req.UserID, "user", req.UserMessage, "TRIGGER_TASK", "", userMeta) asstMeta := map[string]any{ - "run_id": runID, + "run_id": claimed.RunID, "project_id": req.ProjectID, "client_turn_id": req.ClientTurnID, "task_key": req.TaskKey, @@ -136,36 +165,22 @@ func (s *Server) handleCreateTaskRun(w http.ResponseWriter, r *http.Request) { } projectinstance.ApplyScopeToMetadata(asstMeta, taskScope) assistantMsgID, _ = ctxMgr.RecordUserMessageWithID(r.Context(), req.SessionID, 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", asstMeta) + claimed.AssistantMsgID = assistantMsgID + _ = s.taskRuns.BindAssistantMsgID(r.Context(), claimed.RunID, assistantMsgID) } - record := &taskrun.RunRecord{ - RunID: runID, - SessionID: req.SessionID, - ProjectID: req.ProjectID, - UserID: req.UserID, - ClientTurnID: req.ClientTurnID, - TaskKey: req.TaskKey, - Stage: req.Stage, - WorkflowInstanceID: taskScope.WorkflowInstanceID, - StageInstanceID: taskScope.StageInstanceID, - WorkflowStageKey: taskScope.StageKey, - AssistantMsgID: assistantMsgID, - Status: "running", - } - - taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID) + taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, claimed.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) + stageReq := s.buildBizdevStageRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, claimed.RunID) + createdRun, err := s.taskRuns.StartClaimedRun(r.Context(), claimed, 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, + RunID: createdRun.RunID, + Status: createdRun.Status, }) } diff --git a/internal/taskrun/claim_idempotency_test.go b/internal/taskrun/claim_idempotency_test.go new file mode 100644 index 00000000..d143966d --- /dev/null +++ b/internal/taskrun/claim_idempotency_test.go @@ -0,0 +1,222 @@ +package taskrun + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/stream" +) + +// memIdemStore 最小内存 store：支持 idempotency_key 唯一占位。 +type memIdemStore struct { + mu sync.Mutex + byID map[string]*RunRecord + byKey map[string]string + events map[string][]StoredEvent +} + +func newMemIdemStore() *memIdemStore { + return &memIdemStore{ + byID: make(map[string]*RunRecord), + byKey: make(map[string]string), + events: make(map[string][]StoredEvent), + } +} + +func (m *memIdemStore) Available() bool { return true } +func (m *memIdemStore) InsertRun(_ context.Context, rec *RunRecord) error { + m.mu.Lock() + defer m.mu.Unlock() + if rec.IdempotencyKey != "" { + if _, ok := m.byKey[rec.IdempotencyKey]; ok { + return ErrIdempotencyConflict + } + m.byKey[rec.IdempotencyKey] = rec.RunID + } + cpy := *rec + m.byID[rec.RunID] = &cpy + return nil +} +func (m *memIdemStore) UpdateRun(_ context.Context, rec *RunRecord) error { + m.mu.Lock() + defer m.mu.Unlock() + cpy := *rec + m.byID[rec.RunID] = &cpy + return nil +} +func (m *memIdemStore) GetByIdempotencyKey(_ context.Context, key string) (*RunRecord, error) { + m.mu.Lock() + defer m.mu.Unlock() + id, ok := m.byKey[key] + if !ok { + return nil, nil + } + rec := m.byID[id] + if rec == nil { + return nil, nil + } + cpy := *rec + return &cpy, nil +} +func (m *memIdemStore) GetRun(_ context.Context, runID string) (*RunRecord, error) { + m.mu.Lock() + defer m.mu.Unlock() + rec := m.byID[runID] + if rec == nil { + return nil, errors.New("run not found") + } + cpy := *rec + return &cpy, nil +} +func (m *memIdemStore) AppendEvent(_ context.Context, runID string, seq int64, ev stream.StreamEvent) error { + m.mu.Lock() + defer m.mu.Unlock() + m.events[runID] = append(m.events[runID], StoredEvent{Seq: seq, Event: ev}) + return nil +} +func (m *memIdemStore) ListEvents(_ context.Context, runID string, afterSeq int64) ([]StoredEvent, error) { + return nil, nil +} +func (m *memIdemStore) AcquireLease(context.Context, string, string, time.Duration) error { + return nil +} +func (m *memIdemStore) HeartbeatLease(context.Context, string, string, time.Duration) error { + return nil +} +func (m *memIdemStore) ReleaseLease(context.Context, string, string) error { return nil } +func (m *memIdemStore) ResetForResume(context.Context, string, int) error { return nil } +func (m *memIdemStore) MarkOrphanedRunning(context.Context) (int, error) { return 0, nil } + +func TestClaimRun_ConcurrentSameKey_OneOwner(t *testing.T) { + store := newMemIdemStore() + mgr := NewManager(nil, nil) + mgr.SetStore(store) + + const n = 16 + var owners atomic.Int32 + var losers atomic.Int32 + var wg sync.WaitGroup + key := BuildTaskRunIdempotencyKey("sess-conc", "turn-conc-1") + runIDs := make(chan string, n) + + wg.Add(n) + for i := 0; i < n; i++ { + go func(i int) { + defer wg.Done() + rec := &RunRecord{ + RunID: "", + SessionID: "sess-conc", + ClientTurnID: "turn-conc-1", + IdempotencyKey: key, + UserID: 1, + TaskKey: "intake", + Status: "pending", + } + claimed, created, err := mgr.ClaimRun(context.Background(), rec) + if err != nil { + t.Errorf("claim: %v", err) + return + } + if created { + owners.Add(1) + runIDs <- claimed.RunID + } else { + losers.Add(1) + runIDs <- claimed.RunID + } + }(i) + } + wg.Wait() + close(runIDs) + + if owners.Load() != 1 { + t.Fatalf("want exactly 1 claim owner, got %d", owners.Load()) + } + if losers.Load() != n-1 { + t.Fatalf("want %d losers, got %d", n-1, losers.Load()) + } + first := "" + for id := range runIDs { + if first == "" { + first = id + } else if id != first { + t.Fatalf("all claimants must share same run_id, got %s vs %s", first, id) + } + } + if len(store.byID) != 1 { + t.Fatalf("store must have 1 run, got %d", len(store.byID)) + } +} + +func TestClaimThenStart_Concurrent_OneExecutor(t *testing.T) { + store := newMemIdemStore() + mgr := NewManager(nil, nil) + mgr.SetStore(store) + + var execCalls atomic.Int32 + mgr.SetExecutor(func(ctx context.Context, runID string, req bizdev.TaskRunRequest, st *stream.Streamer) (*bizdev.TaskRunResult, error) { + execCalls.Add(1) + time.Sleep(20 * time.Millisecond) + return &bizdev.TaskRunResult{Success: true, WorkDir: t.TempDir(), TaskKey: "intake"}, nil + }, nil, nil) + + const n = 12 + var chatWrites atomic.Int32 + var starts atomic.Int32 + key := BuildTaskRunIdempotencyKey("sess-exec", "turn-exec-1") + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + rec := &RunRecord{ + SessionID: "sess-exec", + ClientTurnID: "turn-exec-1", + IdempotencyKey: key, + UserID: 7, + TaskKey: "intake", + Status: "pending", + } + claimed, created, err := mgr.ClaimRun(context.Background(), rec) + if err != nil { + t.Errorf("claim: %v", err) + return + } + if !created { + return + } + // 模拟 handler：仅 owner 写 1 user + 1 assistant + chatWrites.Add(2) + claimed.AssistantMsgID = 42 + _ = mgr.BindAssistantMsgID(context.Background(), claimed.RunID, 42) + starts.Add(1) + _, err = mgr.StartClaimedRun(context.Background(), claimed, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") + if err != nil { + t.Errorf("start: %v", err) + } + }() + } + wg.Wait() + + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if execCalls.Load() >= 1 { + break + } + time.Sleep(20 * time.Millisecond) + } + if starts.Load() != 1 || chatWrites.Load() != 2 { + t.Fatalf("want 1 start + 2 chat writes, got starts=%d chats=%d", starts.Load(), chatWrites.Load()) + } + if execCalls.Load() != 1 { + t.Fatalf("want 1 executor call, got %d", execCalls.Load()) + } + if len(store.byID) != 1 { + t.Fatalf("want 1 persisted run, got %d", len(store.byID)) + } +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 286c7493..d0845dc2 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -38,6 +38,9 @@ type Manager struct { finalize Finalizer pwSvc *projectworkflow.Service ctxMgr *ctxmgr.ContextManager + + idemMu sync.Mutex + idemLocks map[string]*sync.Mutex // keyed by idempotency_key } const runRetention = time.Hour @@ -49,16 +52,32 @@ var ( func NewManager(pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *Manager { m := &Manager{ - runs: make(map[string]*activeRun), - store: noopStore{}, - workerID: newWorkerID(), - pwSvc: pwSvc, - ctxMgr: ctxMgr, + runs: make(map[string]*activeRun), + store: noopStore{}, + workerID: newWorkerID(), + pwSvc: pwSvc, + ctxMgr: ctxMgr, + idemLocks: make(map[string]*sync.Mutex), } go m.cleanupLoop() return m } +func (m *Manager) lockIdempotencyKey(key string) func() { + if key == "" { + return func() {} + } + m.idemMu.Lock() + lk, ok := m.idemLocks[key] + if !ok { + lk = &sync.Mutex{} + m.idemLocks[key] = lk + } + m.idemMu.Unlock() + lk.Lock() + return lk.Unlock +} + // SetStore 注入持久化存储（PG）；启动时标记 lease 过期的 running 任务为 interrupted。 func (m *Manager) SetStore(store Store) { if m == nil { @@ -111,49 +130,140 @@ func (m *Manager) SetFinalizer(f Finalizer) { } 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 = &RunRecord{} + } + if runID != "" { + record.RunID = runID + } + claimed, created, err := m.ClaimRun(parent, record) + if err != nil { + return nil, err + } + if !created { + return claimed, nil + } + return m.StartClaimedRun(parent, claimed, taskReq, stageReq, resumeWorkDir) +} + +// ClaimRun 原子占位：在写聊天 / 启 worker 之前确认本请求是否为 owner。 +// created=false 表示已有同 idempotency_key 的 run，调用方不得再写消息。 +func (m *Manager) ClaimRun(parent context.Context, record *RunRecord) (*RunRecord, bool, error) { + if record == nil { + return nil, false, fmt.Errorf("taskrun: record required") + } + if record.IdempotencyKey == "" && record.SessionID != "" && record.ClientTurnID != "" { + record.IdempotencyKey = BuildTaskRunIdempotencyKey(record.SessionID, record.ClientTurnID) + } + unlock := m.lockIdempotencyKey(record.IdempotencyKey) + defer unlock() + + if record.IdempotencyKey != "" && m.store != nil && m.store.Available() { + if existing, err := m.store.GetByIdempotencyKey(parent, record.IdempotencyKey); err != nil { + return nil, false, err + } else if existing != nil { + return existing, false, nil + } + } + + if record.RunID == "" { + record.RunID = fmt.Sprintf("run_%d", time.Now().UnixNano()) } - record.RunID = runID if record.Status == "" { - record.Status = "running" + record.Status = "pending" } now := time.Now() - record.CreatedAt = now + if record.CreatedAt.IsZero() { + record.CreatedAt = now + } record.UpdatedAt = now + if record.Attempt <= 0 { + record.Attempt = 1 + } + + if m.store != nil && m.store.Available() { + if err := m.store.InsertRun(parent, record); err != nil { + if errors.Is(err, ErrIdempotencyConflict) && record.IdempotencyKey != "" { + existing, lookupErr := m.store.GetByIdempotencyKey(parent, record.IdempotencyKey) + if lookupErr != nil { + return nil, false, lookupErr + } + if existing != nil { + return existing, false, nil + } + } + return nil, false, err + } + } + cpy := *record + return &cpy, true, nil +} + +// BindAssistantMsgID 在启 worker 前回填 assistant 消息 id，覆盖快速完成竞态。 +func (m *Manager) BindAssistantMsgID(ctx context.Context, runID string, assistantMsgID int64) error { + if runID == "" || assistantMsgID <= 0 { + return nil + } + m.mu.RLock() + ar, ok := m.runs[runID] + m.mu.RUnlock() + if ok { + ar.mu.Lock() + ar.record.AssistantMsgID = assistantMsgID + ar.record.UpdatedAt = time.Now() + rec := *ar.record + ar.mu.Unlock() + if m.store != nil && m.store.Available() { + return m.store.UpdateRun(ctx, &rec) + } + return nil + } + if m.store != nil && m.store.Available() { + rec, err := m.store.GetRun(ctx, runID) + if err != nil { + return err + } + rec.AssistantMsgID = assistantMsgID + rec.UpdatedAt = time.Now() + return m.store.UpdateRun(ctx, rec) + } + return nil +} + +// StartClaimedRun 仅对 ClaimRun 成功的 owner 启 worker；禁止在 Insert 冲突后调用。 +func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) (*RunRecord, error) { + if record == nil || record.RunID == "" { + return nil, fmt.Errorf("taskrun: claimed record required") + } + record.Status = "running" record.WorkerID = m.workerID - record.Attempt = 1 + now := time.Now() + record.UpdatedAt = now leaseUntil := now.Add(defaultLeaseTTL) record.LeaseExpiresAt = &leaseUntil 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) - } else if err := m.store.AcquireLease(parent, runID, m.workerID, defaultLeaseTTL); err != nil { - slog.Warn("taskrun_lease_acquire_failed", "run_id", runID, "error", err) + _ = m.store.UpdateRun(parent, record) + if err := m.store.AcquireLease(parent, record.RunID, m.workerID, defaultLeaseTTL); err != nil { + slog.Warn("taskrun_lease_acquire_failed", "run_id", record.RunID, "error", err) } } 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.runs[record.RunID] = ar m.mu.Unlock() if m.pwSvc != nil && record.ProjectID > 0 { startedBy := record.UserID _, _ = m.pwSvc.CreateWorkflowRun(workerCtx, &projectworkflow.WorkflowRun{ - RunID: runID, + RunID: record.RunID, ProjectID: record.ProjectID, Status: "running", StartedBy: &startedBy, @@ -168,7 +278,8 @@ func (m *Manager) CreateAndStart(parent context.Context, runID string, record *R } go m.runWorker(workerCtx, ar, taskReq, stageReq, resumeWorkDir) - return record, nil + cpy := *record + return &cpy, nil } func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) { @@ -378,6 +489,32 @@ func (m *Manager) GetRun(runID string) (*RunRecord, error) { return nil, fmt.Errorf("run not found") } +// FindByIdempotencyKey 查找既有幂等 run（内存优先，再查 store）。 +func (m *Manager) FindByIdempotencyKey(ctx context.Context, key string) (*RunRecord, error) { + if key == "" { + return nil, nil + } + m.mu.RLock() + for _, ar := range m.runs { + ar.mu.RLock() + match := ar.record != nil && ar.record.IdempotencyKey == key + var cpy RunRecord + if match { + cpy = *ar.record + } + ar.mu.RUnlock() + if match { + m.mu.RUnlock() + return &cpy, nil + } + } + m.mu.RUnlock() + if m.store != nil && m.store.Available() { + return m.store.GetByIdempotencyKey(ctx, key) + } + return nil, nil +} + func (m *Manager) Subscribe(runID string, afterSeq int64) (<-chan StoredEvent, []StoredEvent, error) { m.mu.RLock() ar, ok := m.runs[runID] diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 21be8f7c..5ef80854 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -5,21 +5,25 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "chat-agent-system/internal/stream" ) var ErrStoreNotAvailable = errors.New("taskrun: store not available") +var ErrIdempotencyConflict = errors.New("taskrun: idempotency key already exists") // Store 持久化 task run 与 SSE 事件，支持重启后查询与重放。 type Store interface { Available() bool InsertRun(ctx context.Context, rec *RunRecord) error UpdateRun(ctx context.Context, rec *RunRecord) error + GetByIdempotencyKey(ctx context.Context, key string) (*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) @@ -75,6 +79,7 @@ func (s *PGStore) InsertRun(ctx context.Context, rec *RunRecord) error { if attempt <= 0 { attempt = 1 } + idemKey := nullScopeText(rec.IdempotencyKey) if s.hasLeaseColumns(ctx) { var leaseHeartbeat *time.Time if rec.LeaseExpiresAt != nil { @@ -83,36 +88,51 @@ func (s *PGStore) InsertRun(ctx context.Context, rec *RunRecord) error { } _, err := s.pool.Exec(ctx, ` INSERT INTO task_runs ( - run_id, session_id, project_id, user_id, client_turn_id, + run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, task_key, stage, status, workspace_id, assistant_message_id, error_message, worker_id, lease_expires_at, lease_heartbeat_at, attempt, workflow_instance_id, stage_instance_id, workflow_stage_key, created_at, updated_at - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20) + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21) ON CONFLICT (run_id) DO NOTHING `, - rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, + rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, idemKey, rec.TaskKey, rec.Stage, rec.Status, rec.WorkspaceID, nullInt64(rec.AssistantMsgID), rec.ErrorMessage, rec.WorkerID, rec.LeaseExpiresAt, leaseHeartbeat, attempt, nullInt64(rec.WorkflowInstanceID), nullInt64(rec.StageInstanceID), nullScopeText(rec.WorkflowStageKey), rec.CreatedAt, rec.UpdatedAt, ) - return err + return mapIdempotencyConflict(err) } _, err := s.pool.Exec(ctx, ` INSERT INTO task_runs ( - run_id, session_id, project_id, user_id, client_turn_id, + run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, task_key, stage, status, workspace_id, assistant_message_id, error_message, workflow_instance_id, stage_instance_id, workflow_stage_key, created_at, updated_at - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) ON CONFLICT (run_id) DO NOTHING `, - rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, + rec.RunID, rec.SessionID, nullInt64(rec.ProjectID), rec.UserID, rec.ClientTurnID, idemKey, rec.TaskKey, rec.Stage, rec.Status, rec.WorkspaceID, nullInt64(rec.AssistantMsgID), rec.ErrorMessage, nullInt64(rec.WorkflowInstanceID), nullInt64(rec.StageInstanceID), nullScopeText(rec.WorkflowStageKey), rec.CreatedAt, rec.UpdatedAt, ) + return mapIdempotencyConflict(err) +} + +func mapIdempotencyConflict(err error) error { + if err == nil { + return nil + } + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" { + return fmt.Errorf("%w: %v", ErrIdempotencyConflict, err) + } + msg := strings.ToLower(err.Error()) + if strings.Contains(msg, "idempotency_key") && (strings.Contains(msg, "duplicate") || strings.Contains(msg, "unique") || strings.Contains(msg, "23505")) { + return fmt.Errorf("%w: %v", ErrIdempotencyConflict, err) + } return err } @@ -131,9 +151,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error { error_message = $4, worker_id = $5, lease_expires_at = $6, - updated_at = $7 + assistant_message_id = $7, + updated_at = $8 WHERE run_id = $1 - `, rec.RunID, rec.Status, rec.WorkspaceID, rec.ErrorMessage, rec.WorkerID, rec.LeaseExpiresAt, rec.UpdatedAt) + `, rec.RunID, rec.Status, rec.WorkspaceID, rec.ErrorMessage, rec.WorkerID, rec.LeaseExpiresAt, nullInt64(rec.AssistantMsgID), rec.UpdatedAt) return err } _, err := s.pool.Exec(ctx, ` @@ -141,9 +162,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error { status = $2, workspace_id = $3, error_message = $4, - updated_at = $5 + assistant_message_id = $5, + updated_at = $6 WHERE run_id = $1 - `, rec.RunID, rec.Status, rec.WorkspaceID, rec.ErrorMessage, rec.UpdatedAt) + `, rec.RunID, rec.Status, rec.WorkspaceID, rec.ErrorMessage, nullInt64(rec.AssistantMsgID), rec.UpdatedAt) return err } @@ -216,6 +238,29 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) return &rec, nil } +// GetByIdempotencyKey 按稳定幂等键查找既有 run；无键或未命中返回 (nil, nil)。 +func (s *PGStore) GetByIdempotencyKey(ctx context.Context, key string) (*RunRecord, error) { + if !s.Available() { + return nil, ErrStoreNotAvailable + } + if key == "" { + return nil, nil + } + var runID string + err := s.pool.QueryRow(ctx, ` + SELECT run_id FROM task_runs + WHERE idempotency_key = $1 + LIMIT 1 + `, key).Scan(&runID) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + return s.GetRun(ctx, runID) +} + func (s *PGStore) ListEvents(ctx context.Context, runID string, afterSeq int64) ([]StoredEvent, error) { if !s.Available() { return nil, ErrStoreNotAvailable @@ -428,6 +473,9 @@ func (noopStore) AppendEvent(context.Context, string, int64, stream.StreamEvent) func (noopStore) GetRun(context.Context, string) (*RunRecord, error) { return nil, ErrStoreNotAvailable } +func (noopStore) GetByIdempotencyKey(context.Context, string) (*RunRecord, error) { + return nil, ErrStoreNotAvailable +} func (noopStore) ListEvents(context.Context, string, int64) ([]StoredEvent, error) { return nil, ErrStoreNotAvailable } diff --git a/internal/taskrun/types.go b/internal/taskrun/types.go index 6546cb2a..9f7c707b 100644 --- a/internal/taskrun/types.go +++ b/internal/taskrun/types.go @@ -30,12 +30,13 @@ type RunRecord struct { ProjectID int64 UserID int64 ClientTurnID string + IdempotencyKey string // 新 run 稳定键；旧行保持 NULL 不参与唯一索引 TaskKey string Stage string WorkflowInstanceID int64 StageInstanceID int64 WorkflowStageKey string - Status string // pending | running | completed | failed | cancelled | status_conflict + Status string // pending | running | completed | failed | cancelled | pending_author_confirmation | … WorkspaceID string AssistantMsgID int64 WorkerID string @@ -46,6 +47,14 @@ type RunRecord struct { Attempt int } +// BuildTaskRunIdempotencyKey 新创建用稳定幂等键；空 client_turn 不设键。 +func BuildTaskRunIdempotencyKey(sessionID, clientTurnID string) string { + if sessionID == "" || clientTurnID == "" { + return "" + } + return "turn:" + sessionID + ":" + clientTurnID +} + // StoredEvent 持久化 SSE 事件（支持重连重放）。 type StoredEvent struct { Seq int64 @@ -53,12 +62,12 @@ type StoredEvent struct { } 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 + record *RunRecord + events []StoredEvent + nextSeq int64 + subs map[int]chan StoredEvent + subSeq int + cancel context.CancelFunc + done chan struct{} + mu sync.RWMutex } diff --git a/web/e2e/document-revision-runtime.spec.ts b/web/e2e/document-revision-runtime.spec.ts index 9a15a380..71a21d1f 100644 --- a/web/e2e/document-revision-runtime.spec.ts +++ b/web/e2e/document-revision-runtime.spec.ts @@ -534,15 +534,36 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { const t0 = Date.now(); const promptSlice = '请基于本项目编制首轮'; await waitForSquadCompletion(page, promptSlice); - // Open a workflow doc tab that is no longer template-only (server doc_proj*). - const tab = page.locator('[role="tab"], button, div').filter({ hasText: /需求|摸底|URS|用户/ }).first(); - if ((await tab.count()) === 0) fail('no controlled deliverable tab after first squad round'); - await tab.click(); + // 精确打开工作流文档槽：SlimTabBar 下拉 / DeckDrawer，禁止全局 div 泛匹配。 + const trigger = page.locator('button.dropdown-trigger').first(); + if ((await trigger.count()) > 0) { + await trigger.click(); + const item = page.locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]').filter({ + hasText: /用户初始需求|需求摸底|intake/i, + }).first(); + if ((await item.count()) === 0) fail('UI-03: no intake slot in SlimTabBar dropdown'); + await item.click(); + } else { + const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); + if ((await deck.count()) > 0) await deck.click(); + const card = page.locator('[data-doc-id^="doc_proj"], [data-doc-id*="intake"]').filter({ + hasText: /初始需求|intake/i, + }).first(); + if ((await card.count()) === 0) { + fail('UI-03: no server-backed intake deliverable card (doc_proj* / intake)'); + } + await card.click(); + } await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 60_000 }); await expect(page.getByTestId('doc-rev-template-hint')).toHaveCount(0); await expect(page.locator('text=版本账本未就绪')).toHaveCount(0); + const activeDoc = page.locator('[data-doc-id^="doc_proj"]').first(); + if ((await activeDoc.count()) === 0) fail('UI-03: active pane must be server doc_proj* not doc-p template'); const { v, r } = await readVersionBadge(page); - controlledTitle = (await tab.textContent())?.trim() || 'controlled'; + controlledTitle = (await page.locator('[data-testid="doc-rev-version"]').evaluate((el) => { + const root = el.closest('[data-doc-id]') || el.parentElement; + return root?.getAttribute('data-doc-id') || 'controlled'; + }).catch(() => 'controlled')) as string; await shot(page, '03-first-doc.png'); const st = readPhaseState(); writePhaseState({ @@ -554,8 +575,8 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { recordPass({ id: 'UI-03', name: 'first controlled doc ledger', - input: 'wait squad; open deliverable tab', - actual: `toolbar ${v}/${r}`, + input: 'wait squad; open intake server doc slot', + actual: `toolbar ${v}/${r}; server doc visible`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', diff --git a/web/src/App.tsx b/web/src/App.tsx index 52a7b8fb..2ed6abe0 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -48,6 +48,10 @@ import { toApiProjectRole, } from './lib/projectMemberAdapter'; import { mergeWorkflowDocumentsFromServer } from './lib/deliverableSync'; +import { + resolvePostTaskActiveDocId, + shouldSwitchTemplateSlotToServer, +} from './lib/postTaskDeliverableSync'; import { buildControlledLoadErrorSkeleton, buildControlledLoadingSkeleton, @@ -2481,6 +2485,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } + // 重试必须携带既有 run_id，禁止无条件新 POST runTaskAttempt(task, asstId, mode, attemptIndex + 1, opts); }, delayMs); }; @@ -2500,13 +2505,70 @@ export default function App() { taskRunStream.setThinking(sessionId, false); }; + const syncControlledDocsAfterTask = async (taskKey: string, preferredDocId?: string | null) => { + if (!task.projectId || task.projectId <= 0 || !taskKey) return; + const workflowDocs = await listProjectWorkflowDocuments(task.projectId).catch(() => []); + if (workflowDocs.length === 0) return; + let mergedSnap: DeliverableItem[] = []; + setDeliverables((prev) => { + const merged = mergeWorkflowDocumentsFromServer( + prev, + workflowDocs, + task.projectId!, + currentUser, + ); + mergedSnap = merged; + const serverBackedIds = merged + .filter((d) => d.projectId === task.projectId && shouldLoadControlledContent(d)) + .map((d) => d.docId); + if (serverBackedIds.length > 0) { + setDocCache((cache) => { + const next = { ...cache }; + serverBackedIds.forEach((docId) => { + delete next[docId]; + }); + return next; + }); + setControlledContentReady((ready) => { + const next = { ...ready }; + serverBackedIds.forEach((docId) => { + delete next[docId]; + }); + return next; + }); + } + return merged; + }); + const serverDocId = resolvePostTaskActiveDocId({ + deliverables: mergedSnap, + projectId: task.projectId, + taskKey, + preferredDocId, + }); + if ( + serverDocId && + shouldSwitchTemplateSlotToServer({ + activeDocId, + deliverables: mergedSnap, + projectId: task.projectId, + taskKey, + serverDocId, + }) + ) { + setActiveDocId(serverDocId); + setActiveCategory('workflow'); + } + }; + // 等待态：resume 保留已有 phaseEvents 续接；run/restart 重置正文与步骤条 patchAsst({ status: 'pending', livePhaseLabel: mode === 'resume' ? '正在继续任务…' : mode === 'restart' ? '正在重新开始任务…' : '任务执行中…', errorInfo: undefined, - ...(mode === 'resume' ? {} : { content: '', phaseEvents: [], taskWorkspaceId: undefined }), + ...(mode === 'resume' || attemptIndex > 0 + ? {} + : { content: '', phaseEvents: [], taskWorkspaceId: undefined }), }); taskRunStream.setThinking(sessionId, true); @@ -2575,6 +2637,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); void (async () => { let nextStatus: AssistantMessageStatus = 'completed'; + let preferredDocId: string | null = null; if (finishedRunId) { const remote = await getTaskRun(finishedRunId).catch(() => null); if (remote?.status) { @@ -2587,6 +2650,7 @@ export default function App() { livePhaseLabel: undefined, taskWorkspaceId: ev.done?.workspace_id, }); + await syncControlledDocsAfterTask(task.taskKey || 'intake', preferredDocId); await chatWindow.refresh().catch(() => undefined); })(); break; @@ -2603,6 +2667,39 @@ export default function App() { : buildTaskRequest(task); if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { + // 断线重试：若已有 run 仍在执行，只订阅，禁止新 POST + let existingRunId: string | undefined; + setMessages((prev) => { + const asst = prev.find((m) => m.id === asstId); + existingRunId = asst?.taskRunId || undefined; + return prev; + }); + if (!existingRunId) { + existingRunId = taskRunStream.activeTaskRunId || undefined; + } + if (attemptIndex > 0 && existingRunId) { + const remote = await getTaskRun(existingRunId).catch(() => null); + if (remote && (remote.status === 'running' || remote.status === 'pending')) { + taskRunStream.setRunId(sessionId, existingRunId); + patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); + await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); + return; + } + if (remote?.status === 'pending_author_confirmation' || remote?.status === 'completed') { + settled = true; + const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant') || 'completed'; + patchAsst({ + status: mapped, + livePhaseLabel: undefined, + taskWorkspaceId: remote.workspace_id, + taskRunId: existingRunId, + }); + taskRunStream.setThinking(sessionId, false); + await syncControlledDocsAfterTask(task.taskKey || 'intake'); + return; + } + } + const created = await createTaskRun({ ...body, session_id: task.sessionId, @@ -2639,6 +2736,7 @@ export default function App() { taskWorkspaceId: remote.workspace_id, }); taskRunStream.setThinking(sessionId, false); + await syncControlledDocsAfterTask(task.taskKey || 'intake'); return; } if (remote.status === 'pending_author_confirmation') { @@ -2649,9 +2747,17 @@ export default function App() { taskWorkspaceId: remote.workspace_id, }); taskRunStream.setThinking(sessionId, false); + await syncControlledDocsAfterTask(task.taskKey || 'intake'); await chatWindow.refresh().catch(() => undefined); return; } + if (remote.status === 'running' || remote.status === 'pending') { + // 原 run 仍在跑：重订阅，不 POST 新 run + settled = false; + patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); + await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); + return; + } if (remote.status === 'failed' || remote.status === 'cancelled') { failOrRetry(buildNetworkErrorInfo()); return; diff --git a/web/src/engine/templateMatch.test.ts b/web/src/engine/templateMatch.test.ts index 01eff442..221e343b 100644 --- a/web/src/engine/templateMatch.test.ts +++ b/web/src/engine/templateMatch.test.ts @@ -44,6 +44,24 @@ test('resolveRequirementsTaskKey selects user_requirements for explicit URS', () assert.equal(resolveRequirementsTaskKey('请编制用户需求表 URS'), 'user_requirements'); }); +const UI02_FIRST_ROUND_PROMPT = + '请基于本项目编制首轮《用户需求》摸底文档，覆盖预期用途、关键性能与适用法规框架，直接生成受控交付物。'; + +test('UI-02 first-round prompt with 首轮+用户需求+摸底 must be intake not user_requirements', () => { + assert.equal(resolveRequirementsTaskKey(UI02_FIRST_ROUND_PROMPT), 'intake'); + assert.equal(resolveTaskTemplateForMessage(UI02_FIRST_ROUND_PROMPT, bothPool)?.taskKey, 'intake'); +}); + +test('首轮 / 需求摸底 / 摸底 beat bare 用户需求', () => { + assert.equal(resolveRequirementsTaskKey('请编制首轮用户需求文档'), 'intake'); + assert.equal(resolveRequirementsTaskKey('进入需求摸底，覆盖用户需求要点'), 'intake'); + assert.equal(resolveRequirementsTaskKey('生成本轮摸底文档（含用户需求）'), 'intake'); +}); + +test('bare 用户需求 without V0.1 markers stays user_requirements', () => { + assert.equal(resolveRequirementsTaskKey('请更新用户需求规格'), 'user_requirements'); +}); + test('matchTemplateDeliverable uses intake key when pool only has URS template', () => { const hit = matchTemplateDeliverable('开始编写用户初始需求表', ursOnlyPool); assert.equal(hit?.taskKey, 'intake'); diff --git a/web/src/engine/templateMatch.ts b/web/src/engine/templateMatch.ts index b3c43ec3..209b4857 100644 --- a/web/src/engine/templateMatch.ts +++ b/web/src/engine/templateMatch.ts @@ -4,21 +4,27 @@ function templatePool(deliverables: DeliverableItem[]): DeliverableItem[] { return deliverables.filter((d) => d.isTemplateOnly); } -/** 从文案判定 intake（V0.1）还是 user_requirements（V0.5），与交付物池无关。 */ +/** + * 从文案判定 intake（V0.1）还是 user_requirements（V0.5），与交付物池无关。 + * 含「首轮 / 初始 / 需求摸底 / 摸底」的明确 V0.1 文案必须 intake，即使同时含「用户需求」。 + * intake 与 user_requirements 仍是两个真实任务键，不做全局别名。 + */ export function resolveRequirementsTaskKey(text: string): 'intake' | 'user_requirements' { + const wantsIntake = + text.includes('首轮') || + text.includes('初始') || + text.includes('需求摸底') || + text.includes('摸底') || + text.includes('初始需求') || + text.includes('初始需求表'); + if (wantsIntake) return 'intake'; + const wantsUrs = text.includes('用户需求表') || text.includes('URS') || - (text.includes('用户需求') && !text.includes('初始')); + text.includes('用户需求'); if (wantsUrs) return 'user_requirements'; - const wantsIntake = - text.includes('初始需求') || - text.includes('需求摸底') || - text.includes('摸底') || - text.includes('初始需求表'); - if (wantsIntake) return 'intake'; - // 含糊表述默认 V0.1 摸底表，避免误触 V0.5 用户需求表导致阶段门禁 403 return 'intake'; } @@ -52,7 +58,7 @@ export function resolveTaskKeyFromMessage(text: string): string | undefined { // 模板名只是操作对象，不代表用户授权写入。只读动词优先；只有明确写动作才走任务 API。 if (readOnlyIntent.test(normalized) || !writeIntent.test(normalized)) return undefined; - if (normalized.includes('需求') || normalized.includes('摸底') || normalized.includes('用户')) { + if (normalized.includes('需求') || normalized.includes('摸底') || normalized.includes('用户') || normalized.includes('首轮')) { return resolveRequirementsTaskKey(normalized); } if (normalized.includes('调研') || normalized.includes('竞品')) return 'research'; diff --git a/web/src/lib/postTaskDeliverableSync.test.ts b/web/src/lib/postTaskDeliverableSync.test.ts new file mode 100644 index 00000000..a6a2d363 --- /dev/null +++ b/web/src/lib/postTaskDeliverableSync.test.ts @@ -0,0 +1,90 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { + resolvePostTaskActiveDocId, + shouldSwitchTemplateSlotToServer, +} from './postTaskDeliverableSync'; +import type { DeliverableItem } from '../types/workbench'; + +const base = (over: Partial): DeliverableItem => ({ + docId: 'doc-x', + projectId: 6089, + taskKey: 'intake', + title: '用户初始需求收集表', + stage: 'V0.1', + category: 'workflow', + creatorId: 1, + creatorName: 'A', + isFixedWorkflow: true, + isTemplateOnly: true, + lifecycleState: 'drafting', + isClaimed: false, + author: '', + reviewer: '', + approver: '', + updatedAt: '', + governanceMode: 'controlled', + ...over, +}); + +test('resolvePostTaskActiveDocId prefers server doc_proj for taskKey', () => { + const docs = [ + base({ docId: 'doc-p6089-intake', isTemplateOnly: true }), + base({ + docId: 'doc_proj6089_intake', + isTemplateOnly: false, + title: '用户初始需求收集表', + }), + ]; + assert.equal( + resolvePostTaskActiveDocId({ deliverables: docs, projectId: 6089, taskKey: 'intake' }), + 'doc_proj6089_intake', + ); +}); + +test('resolvePostTaskActiveDocId honors preferred server doc id', () => { + const docs = [ + base({ docId: 'doc_proj6089_intake', isTemplateOnly: false }), + base({ + docId: 'doc_proj6089_user_requirements', + taskKey: 'user_requirements', + isTemplateOnly: false, + }), + ]; + assert.equal( + resolvePostTaskActiveDocId({ + deliverables: docs, + projectId: 6089, + taskKey: 'intake', + preferredDocId: 'doc_proj6089_intake', + }), + 'doc_proj6089_intake', + ); +}); + +test('shouldSwitchTemplateSlotToServer when active is local template slot', () => { + const docs = [ + base({ docId: 'doc-p6089-intake', isTemplateOnly: true }), + base({ docId: 'doc_proj6089_intake', isTemplateOnly: false }), + ]; + assert.equal( + shouldSwitchTemplateSlotToServer({ + activeDocId: 'doc-p6089-intake', + deliverables: docs, + projectId: 6089, + taskKey: 'intake', + serverDocId: 'doc_proj6089_intake', + }), + true, + ); + assert.equal( + shouldSwitchTemplateSlotToServer({ + activeDocId: 'doc_proj6089_intake', + deliverables: docs, + projectId: 6089, + taskKey: 'intake', + serverDocId: 'doc_proj6089_intake', + }), + false, + ); +}); diff --git a/web/src/lib/postTaskDeliverableSync.ts b/web/src/lib/postTaskDeliverableSync.ts new file mode 100644 index 00000000..545bb754 --- /dev/null +++ b/web/src/lib/postTaskDeliverableSync.ts @@ -0,0 +1,65 @@ +import type { DeliverableItem } from '../types/workbench'; + +/** + * 任务完成后：在已 merge 的 deliverables 中定位同 taskKey 的服务端受控文档槽， + * 用于把当前模板页切换到 doc_proj*（toolbar V/r 可挂载）。 + */ +export function resolvePostTaskActiveDocId(params: { + deliverables: DeliverableItem[]; + projectId: number; + taskKey: string; + preferredDocId?: string | null; +}): string | null { + const { deliverables, projectId, taskKey, preferredDocId } = params; + if (!taskKey || projectId <= 0) return null; + + if (preferredDocId) { + const preferred = deliverables.find( + (d) => + d.docId === preferredDocId && + d.projectId === projectId && + d.taskKey === taskKey && + d.governanceMode !== 'freeform' && + !d.isTemplateOnly, + ); + if (preferred) return preferred.docId; + } + + const serverSlot = deliverables.find( + (d) => + d.projectId === projectId && + d.taskKey === taskKey && + d.governanceMode !== 'freeform' && + d.isFixedWorkflow !== false && + !d.isTemplateOnly && + /^doc_proj\d+_/.test(d.docId), + ); + if (serverSlot) return serverSlot.docId; + + const anyNonTemplate = deliverables.find( + (d) => + d.projectId === projectId && + d.taskKey === taskKey && + d.governanceMode !== 'freeform' && + !d.isTemplateOnly, + ); + return anyNonTemplate?.docId ?? null; +} + +/** 若当前 active 仍是同 taskKey 的模板槽，则应切换到 serverDocId。 */ +export function shouldSwitchTemplateSlotToServer(params: { + activeDocId: string | null | undefined; + deliverables: DeliverableItem[]; + projectId: number; + taskKey: string; + serverDocId: string; +}): boolean { + const { activeDocId, deliverables, projectId, taskKey, serverDocId } = params; + if (!serverDocId || !taskKey) return false; + if (!activeDocId) return true; + if (activeDocId === serverDocId) return false; + const active = deliverables.find((d) => d.docId === activeDocId); + if (!active) return true; + if (active.projectId !== projectId || active.taskKey !== taskKey) return false; + return Boolean(active.isTemplateOnly) || active.docId.startsWith('doc-p'); +} From 7466a15319f35b457a325e5f3061028c5a44e7b3 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 13:54:49 +0000 Subject: [PATCH 2/5] fix(docgov): harden UI-03 sync poll, reconnectRunId, and E2E status attrs Use deliverablesRef + limited poll for post-task server docs; pass reconnectRunId across auto-retry; GetRun reads idempotency_key; StartClaimedRun fails closed on UpdateRun; MessageBubble/E2E scope this-turn assistant; correct audit retry/schema19 wording. Co-authored-by: Cursor --- ...6-09-13-controlled-doc-client-sync-ui03.md | 24 ++-- internal/taskrun/claim_idempotency_test.go | 42 ++++++ internal/taskrun/manager.go | 4 +- internal/taskrun/store.go | 12 +- web/e2e/document-revision-runtime.spec.ts | 90 ++++++------- web/src/App.tsx | 120 ++++++++++-------- web/src/components/chat/MessageBubble.tsx | 2 + web/src/lib/postTaskDeliverableSync.test.ts | 65 ++++++++++ web/src/lib/postTaskDeliverableSync.ts | 62 +++++++++ web/src/lib/taskRunReconnect.test.ts | 42 ++++++ web/src/lib/taskRunReconnect.ts | 38 ++++++ 11 files changed, 385 insertions(+), 116 deletions(-) create mode 100644 web/src/lib/taskRunReconnect.test.ts create mode 100644 web/src/lib/taskRunReconnect.ts diff --git a/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md index 024e1d5e..62e7c3dd 100644 --- a/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md +++ b/.doc/audit/2026-09-13-controlled-doc-client-sync-ui03.md @@ -46,19 +46,19 @@ severity: critical 3. **真实 UI 为下拉/抽屉交互**： `SlimTabBar.tsx` 为面包屑下拉（`button.dropdown-trigger` -> `.dropdown-menu`）或大盘抽屉（`DeckDrawer`），页面不存在水平文档 Tab。 -### C. Playwright 串行 Retry 现象与 DB 约束异常暴露 +### C. 前端断线自动重试与 DB 约束异常暴露 -1. **Playwright 串行重试导致任务重发**： - Trace 与聊天现场显示：同一项目下，13:02 成功执行并完成首轮任务后，因重试机制或步骤重入，13:05 再次向同一个会话发送了相同的首轮指令： +1. **前端断线自动重试导致同 turn 二次触发（非 Playwright retries）**： + `web/playwright.config.ts` 中 **`retries: 0`**。现场 13:02 完成首轮后，13:05 再次出现同 prompt 处理中，根因是前端 `runTaskAttempt` 在连接异常路径上自动重试，且重试开头 `setRunId(null)` / 未显式传递 `reconnectRunId`，导致重复 `POST /tasks/runs` 而非订阅既有 run： - 13:02：`TRIGGER_TASK 请基于本项目编制首轮...` → `《用户需求表》已生成。当前状态：已完成` - - 13:05：`TRIGGER_TASK 请基于本项目编制首轮...` → 触发二次运行，助手再次进入处理中 + - 13:05：同 turn 自动重试路径再次进入处理中（工程修复：`reconnectRunId` 跨 attempt + Claim 幂等） 2. **暴露 `task_runs` 状态约束缺陷**： 在二次运行与高风险候选流转路径中，后台服务尝试将 `task_runs.status` 更新为 `pending_author_confirmation` 时，暴露了数据库底层约束拦截： - `task_runs` 表定义中的 `chk_task_run_status` 仅允许： + goose **`00019`** 原始 `chk_task_run_status` 仅允许： ```sql - CHECK (status IN ('queued', 'running', 'completed', 'failed', 'cancelled', 'quality_failed')) + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'finalizing_failed', 'quality_failed', 'interrupted')) ``` - 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规。 + 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规（由 goose `00030` 补齐）。 --- @@ -97,10 +97,10 @@ severity: critical | 根因 | 修复 | 门禁 | |---|---|---| -| A 任务完成后未 merge 受控文档 | `postTaskDeliverableSync` + `App.tsx` done/pending/reconnect 后 `listProjectWorkflowDocuments` → merge → 切到 `doc_proj*` | `npm test` 232 PASS；`npm run build` PASS | -| A′ intake vs URS 误匹配 | `templateMatch`：**intake 标记优先**于裸「用户需求」 | `templateMatch` + sync 单测（`npx tsx --test` 18/18） | -| B UI-03 选择器命中 Banner | E2E 改为 SlimTabBar/DeckDrawer 精确选择器；要求服务端 `doc_proj*` | 待真实 UI 重跑 | -| C `pending_author_confirmation` 约束 | goose `00030` 扩展 `chk_task_run_status` | **真实现卷迁移由合入前独立验证**（本提交不跑复杂 PG 集成测） | -| C′ 重试重复建 run / 重复写 chat | `ClaimRun` 先占 `idempotency_key`；非 owner 直接 202；FE reconnect 同 `run_id` 订阅 | `go test -short ./internal/taskrun ./internal/server` PASS | +| A 任务完成后未 merge 受控文档 | `pollUntilPostTaskServerDoc` + `deliverablesRef` 显式快照（禁止 setState updater 取 mergedSnap）+ 有限重试至 `doc_proj*` | focused + `npm test` / `build` | +| A′ intake vs URS 误匹配 | `templateMatch`：**intake 标记优先**于裸「用户需求」 | 单测 PASS | +| B UI-03 选择器命中 Banner | SlimTabBar/DeckDrawer；`data-chat-role/status` + 本轮 assistant sibling 等待；UI-11 禁止全局 div | 待真实 UI 重跑 | +| C `pending_author_confirmation` 约束 | goose `00030` 扩展 `chk_task_run_status`（相对 **00019** 枚举） | 真实现卷迁移合入前独立验证 | +| C′ 断线自动重试重复 POST | `reconnectRunId` 显式跨 retry；`ClaimRun`；`GetRun` 读回 `idempotency_key`；`StartClaimedRun` UpdateRun 失败不启 worker | `go test -short ./internal/taskrun` | **未做**：占槽 rebuild、真实 UI UI-03 重验、schema30 对现有重复 `(session_id,client_turn_id)` 卷的迁移实测。 diff --git a/internal/taskrun/claim_idempotency_test.go b/internal/taskrun/claim_idempotency_test.go index d143966d..1cc97d9e 100644 --- a/internal/taskrun/claim_idempotency_test.go +++ b/internal/taskrun/claim_idempotency_test.go @@ -220,3 +220,45 @@ func TestClaimThenStart_Concurrent_OneExecutor(t *testing.T) { t.Fatalf("want 1 persisted run, got %d", len(store.byID)) } } + +type failUpdateStore struct { + *memIdemStore + updateErr error +} + +func (f *failUpdateStore) UpdateRun(ctx context.Context, rec *RunRecord) error { + if f.updateErr != nil { + return f.updateErr + } + return f.memIdemStore.UpdateRun(ctx, rec) +} + +func TestStartClaimedRun_UpdateRunFailureDoesNotStartWorker(t *testing.T) { + base := newMemIdemStore() + store := &failUpdateStore{memIdemStore: base, updateErr: errors.New("persist boom")} + mgr := NewManager(nil, nil) + mgr.SetStore(store) + + rec := &RunRecord{ + SessionID: "sess-upd-fail", + ClientTurnID: "turn-upd-fail", + IdempotencyKey: BuildTaskRunIdempotencyKey("sess-upd-fail", "turn-upd-fail"), + UserID: 1, + TaskKey: "intake", + Status: "pending", + } + claimed, created, err := mgr.ClaimRun(context.Background(), rec) + if err != nil || !created { + t.Fatalf("claim: created=%v err=%v", created, err) + } + _, err = mgr.StartClaimedRun(context.Background(), claimed, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") + if err == nil { + t.Fatal("expected StartClaimedRun to fail when UpdateRun fails") + } + mgr.mu.Lock() + _, live := mgr.runs[claimed.RunID] + mgr.mu.Unlock() + if live { + t.Fatal("worker must not start when UpdateRun persistence fails") + } +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index d0845dc2..2efbd448 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -243,7 +243,9 @@ func (m *Manager) StartClaimedRun(parent context.Context, record *RunRecord, tas record.LeaseExpiresAt = &leaseUntil if m.store != nil && m.store.Available() { - _ = m.store.UpdateRun(parent, record) + if err := m.store.UpdateRun(parent, record); err != nil { + return nil, fmt.Errorf("taskrun: persist running status before start: %w", err) + } if err := m.store.AcquireLease(parent, record.RunID, m.workerID, defaultLeaseTTL); err != nil { slog.Warn("taskrun_lease_acquire_failed", "run_id", record.RunID, "error", err) } diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 5ef80854..13abb602 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -191,15 +191,16 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) } var rec RunRecord var projectID, assistantMsgID *int64 + var idempotencyKey *string if s.hasLeaseColumns(ctx) { err := s.pool.QueryRow(ctx, ` - SELECT run_id, session_id, project_id, user_id, client_turn_id, + SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, task_key, stage, status, workspace_id, assistant_message_id, error_message, worker_id, lease_expires_at, attempt, created_at, updated_at FROM task_runs WHERE run_id = $1 `, runID).Scan( - &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, + &rec.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, &rec.ErrorMessage, &rec.WorkerID, &rec.LeaseExpiresAt, &rec.Attempt, &rec.CreatedAt, &rec.UpdatedAt, @@ -212,13 +213,13 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) } } else { err := s.pool.QueryRow(ctx, ` - SELECT run_id, session_id, project_id, user_id, client_turn_id, + SELECT run_id, session_id, project_id, user_id, client_turn_id, idempotency_key, 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.RunID, &rec.SessionID, &projectID, &rec.UserID, &rec.ClientTurnID, &idempotencyKey, &rec.TaskKey, &rec.Stage, &rec.Status, &rec.WorkspaceID, &assistantMsgID, &rec.ErrorMessage, &rec.CreatedAt, &rec.UpdatedAt, ) @@ -235,6 +236,9 @@ func (s *PGStore) GetRun(ctx context.Context, runID string) (*RunRecord, error) if assistantMsgID != nil { rec.AssistantMsgID = *assistantMsgID } + if idempotencyKey != nil { + rec.IdempotencyKey = *idempotencyKey + } return &rec, nil } diff --git a/web/e2e/document-revision-runtime.spec.ts b/web/e2e/document-revision-runtime.spec.ts index 71a21d1f..630cd2f3 100644 --- a/web/e2e/document-revision-runtime.spec.ts +++ b/web/e2e/document-revision-runtime.spec.ts @@ -222,40 +222,60 @@ async function adminCreateUser( } /** - * Squad wait: MUST observe start (user bubble + running/thinking) before terminal. - * failed status → FAIL. + * Squad wait: observe THIS turn's user bubble, then its following assistant sibling + * running → terminal. Never match a prior completed bubble globally. */ async function waitForSquadCompletion(page: Page, userPromptSnippet: string, timeoutMs = 420_000) { const startedAt = Date.now(); - const userBubble = page.locator('div, article, li').filter({ hasText: userPromptSnippet }).first(); + const userBubble = page + .locator('[data-chat-role="user"]') + .filter({ hasText: userPromptSnippet }) + .last(); await expect(userBubble).toBeVisible({ timeout: 30_000 }); - const running = page - .locator('text=正在思考') - .or(page.locator('text=生成中')) - .or(page.locator('.agent-pulse-active')) - .or(page.locator('text=执行中')) - .or(page.locator('text=编排中')); - await expect(running.first()).toBeVisible({ timeout: 60_000 }); + const assistant = userBubble.locator( + 'xpath=following-sibling::*[@data-chat-role="assistant"][1]', + ); + await expect(assistant).toBeVisible({ timeout: 30_000 }); - const terminal = page - .locator('text=已完成') - .or(page.locator('text=待编制人确认')) - .or(page.locator('text=失败')) - .or(page.locator('[data-status="completed"]')) - .or(page.locator('[data-status="pending_author_confirmation"]')) - .or(page.locator('[data-status="failed"]')); + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /pending|streaming_status|auto_retrying/, + { timeout: 60_000 }, + ); - await expect(terminal.first()).toBeVisible({ timeout: timeoutMs }); + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /completed|pending_author_confirmation|failed|cancelled|quality_failed/, + { timeout: timeoutMs }, + ); - if (await page.locator('text=失败').or(page.locator('[data-status="failed"]')).first().isVisible().catch(() => false)) { + const terminal = (await assistant.getAttribute('data-chat-status')) || ''; + if (terminal === 'failed') { fail('squad terminal status is failed'); } - // Ensure running indicator has cleared. - await expect(page.locator('text=正在思考，请稍候...')).toBeHidden({ timeout: 60_000 }).catch(() => undefined); return Date.now() - startedAt; } +async function openWorkflowDocByLabel(page: Page, label: RegExp) { + const trigger = page.locator('button.dropdown-trigger').first(); + if ((await trigger.count()) > 0) { + await trigger.click(); + const item = page + .locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]') + .filter({ hasText: label }) + .first(); + if ((await item.count()) === 0) fail(`workflow slot not in SlimTabBar dropdown: ${label}`); + await item.click(); + return; + } + const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); + if ((await deck.count()) > 0) await deck.click(); + const card = page.locator('[data-doc-id^="doc_proj"]').filter({ hasText: label }).first(); + if ((await card.count()) === 0) fail(`no server-backed workflow card for ${label}`); + await card.click(); +} + async function sendChat(page: Page, text: string) { const box = page.getByPlaceholder(/向项目助手提问/); await expect(box).toBeVisible({ timeout: 15_000 }); @@ -535,25 +555,7 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { const promptSlice = '请基于本项目编制首轮'; await waitForSquadCompletion(page, promptSlice); // 精确打开工作流文档槽：SlimTabBar 下拉 / DeckDrawer，禁止全局 div 泛匹配。 - const trigger = page.locator('button.dropdown-trigger').first(); - if ((await trigger.count()) > 0) { - await trigger.click(); - const item = page.locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]').filter({ - hasText: /用户初始需求|需求摸底|intake/i, - }).first(); - if ((await item.count()) === 0) fail('UI-03: no intake slot in SlimTabBar dropdown'); - await item.click(); - } else { - const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); - if ((await deck.count()) > 0) await deck.click(); - const card = page.locator('[data-doc-id^="doc_proj"], [data-doc-id*="intake"]').filter({ - hasText: /初始需求|intake/i, - }).first(); - if ((await card.count()) === 0) { - fail('UI-03: no server-backed intake deliverable card (doc_proj* / intake)'); - } - await card.click(); - } + await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 60_000 }); await expect(page.getByTestId('doc-rev-template-hint')).toHaveCount(0); await expect(page.locator('text=版本账本未就绪')).toHaveCount(0); @@ -794,15 +796,13 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { '请另行生成一份《系统设计输入》受控文档（第二份），不要覆盖上一份需求文档，并给出可打开的交付物。'; await sendChat(page, prompt); await waitForSquadCompletion(page, '系统设计输入'); - const secondTab = page.locator('[role="tab"], button, div').filter({ hasText: /设计输入|系统设计/ }).first(); - if ((await secondTab.count()) === 0) fail('second controlled document tab not found after chat'); - await secondTab.click(); - secondDocTitle = (await secondTab.textContent())?.trim() || ''; + await openWorkflowDocByLabel(page, /设计输入|系统设计/); + secondDocTitle = '系统设计输入'; await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 45_000 }); const states: string[] = []; for (const label of [controlledTitle, secondDocTitle]) { if (!label) continue; - await page.getByText(label, { exact: false }).first().click(); + await openWorkflowDocByLabel(page, new RegExp(label.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'))); const { v, r } = await readVersionBadge(page); states.push(`${label}:${v}/${r}`); } diff --git a/web/src/App.tsx b/web/src/App.tsx index 2ed6abe0..87ea8eb6 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -49,9 +49,15 @@ import { } from './lib/projectMemberAdapter'; import { mergeWorkflowDocumentsFromServer } from './lib/deliverableSync'; import { - resolvePostTaskActiveDocId, + pollUntilPostTaskServerDoc, shouldSwitchTemplateSlotToServer, } from './lib/postTaskDeliverableSync'; +import { + isRemoteTaskTerminal, + nextAttemptOpts, + resolveReconnectRunId, + shouldSubscribeExistingRun, +} from './lib/taskRunReconnect'; import { buildControlledLoadErrorSkeleton, buildControlledLoadingSkeleton, @@ -584,6 +590,8 @@ export default function App() { const [deliverables, setDeliverables] = useState(() => loadDeliverables(ALL_INITIAL_DELIVERABLES) ); + const deliverablesRef = useRef(deliverables); + deliverablesRef.current = deliverables; // 自动持久化 deliverables（序列化时剔除不可持久化的 File 对象，见 storage.ts） useEffect(() => { @@ -2461,11 +2469,20 @@ export default function App() { asstId: string, mode: 'run' | 'resume' | 'restart', attemptIndex: number, - opts?: { phaseEvents?: PhaseEvent[]; workDir?: string } + opts?: { phaseEvents?: PhaseEvent[]; workDir?: string; reconnectRunId?: string } ) => { const sessionId = task.sessionId || activeSessionId; + let retainedRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: opts?.reconnectRunId, + }); taskRunStream.abortSession(sessionId); - taskRunStream.setRunId(sessionId, null); + // 重连保留显式 runId；禁止无条件 setRunId(null) 导致自动重试丢失 + if (retainedRunId) { + taskRunStream.setRunId(sessionId, retainedRunId); + } else { + taskRunStream.setRunId(sessionId, null); + } const controller = new AbortController(); taskRunStream.setController(sessionId, controller); @@ -2485,8 +2502,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } - // 重试必须携带既有 run_id，禁止无条件新 POST - runTaskAttempt(task, asstId, mode, attemptIndex + 1, opts); + runTaskAttempt(task, asstId, mode, attemptIndex + 1, nextAttemptOpts(opts, retainedRunId)); }, delayMs); }; @@ -2507,49 +2523,44 @@ export default function App() { const syncControlledDocsAfterTask = async (taskKey: string, preferredDocId?: string | null) => { if (!task.projectId || task.projectId <= 0 || !taskKey) return; - const workflowDocs = await listProjectWorkflowDocuments(task.projectId).catch(() => []); - if (workflowDocs.length === 0) return; - let mergedSnap: DeliverableItem[] = []; - setDeliverables((prev) => { - const merged = mergeWorkflowDocumentsFromServer( - prev, - workflowDocs, - task.projectId!, - currentUser, - ); - mergedSnap = merged; - const serverBackedIds = merged - .filter((d) => d.projectId === task.projectId && shouldLoadControlledContent(d)) - .map((d) => d.docId); - if (serverBackedIds.length > 0) { - setDocCache((cache) => { - const next = { ...cache }; - serverBackedIds.forEach((docId) => { - delete next[docId]; - }); - return next; - }); - setControlledContentReady((ready) => { - const next = { ...ready }; - serverBackedIds.forEach((docId) => { - delete next[docId]; - }); - return next; - }); - } - return merged; - }); - const serverDocId = resolvePostTaskActiveDocId({ - deliverables: mergedSnap, + const { merged, serverDocId } = await pollUntilPostTaskServerDoc({ + listDocs: () => listProjectWorkflowDocuments(task.projectId!).catch(() => []), + getDeliverables: () => deliverablesRef.current, + merge: (prev, docs) => + mergeWorkflowDocumentsFromServer(prev, docs, task.projectId!, currentUser), projectId: task.projectId, taskKey, preferredDocId, + maxAttempts: 10, + delayMs: 800, + onMerged: (next) => { + deliverablesRef.current = next; + setDeliverables(next); + const serverBackedIds = next + .filter((d) => d.projectId === task.projectId && shouldLoadControlledContent(d)) + .map((d) => d.docId); + if (serverBackedIds.length === 0) return; + setDocCache((cache) => { + const cleared = { ...cache }; + serverBackedIds.forEach((docId) => { + delete cleared[docId]; + }); + return cleared; + }); + setControlledContentReady((ready) => { + const cleared = { ...ready }; + serverBackedIds.forEach((docId) => { + delete cleared[docId]; + }); + return cleared; + }); + }, }); if ( serverDocId && shouldSwitchTemplateSlotToServer({ activeDocId, - deliverables: mergedSnap, + deliverables: merged, projectId: task.projectId, taskKey, serverDocId, @@ -2632,7 +2643,7 @@ export default function App() { } case 'done': { settled = true; - const finishedRunId = taskRunStream.activeTaskRunId; + const finishedRunId = retainedRunId || taskRunStream.activeTaskRunId; taskRunStream.setRunId(sessionId, null); taskRunStream.setThinking(sessionId, false); void (async () => { @@ -2649,6 +2660,7 @@ export default function App() { status: nextStatus, livePhaseLabel: undefined, taskWorkspaceId: ev.done?.workspace_id, + ...(finishedRunId ? { taskRunId: finishedRunId } : {}), }); await syncControlledDocsAfterTask(task.taskKey || 'intake', preferredDocId); await chatWindow.refresh().catch(() => undefined); @@ -2667,25 +2679,21 @@ export default function App() { : buildTaskRequest(task); if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { - // 断线重试：若已有 run 仍在执行，只订阅，禁止新 POST - let existingRunId: string | undefined; - setMessages((prev) => { - const asst = prev.find((m) => m.id === asstId); - existingRunId = asst?.taskRunId || undefined; - return prev; + // 断线重试：显式 reconnectRunId 保留；禁止 setMessages updater 取 runId + const existingRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: retainedRunId || opts?.reconnectRunId, }); - if (!existingRunId) { - existingRunId = taskRunStream.activeTaskRunId || undefined; - } if (attemptIndex > 0 && existingRunId) { const remote = await getTaskRun(existingRunId).catch(() => null); - if (remote && (remote.status === 'running' || remote.status === 'pending')) { + if (remote && shouldSubscribeExistingRun(remote.status)) { + retainedRunId = existingRunId; taskRunStream.setRunId(sessionId, existingRunId); patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); return; } - if (remote?.status === 'pending_author_confirmation' || remote?.status === 'completed') { + if (remote && isRemoteTaskTerminal(remote.status)) { settled = true; const mapped = mapTaskStatusToMessageStatus(remote.status, 'assistant') || 'completed'; patchAsst({ @@ -2708,6 +2716,7 @@ export default function App() { client_turn_id: task.clientTurnId, user_message: task.userMessage, }); + retainedRunId = created.run_id; taskRunStream.setRunId(sessionId, created.run_id); patchAsst({ taskRunId: created.run_id }); await subscribeTaskRunEvents(created.run_id, { onEvent }, controller.signal); @@ -2724,7 +2733,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } - const runId = taskRunStream.activeTaskRunId; + const runId = retainedRunId || taskRunStream.activeTaskRunId || undefined; if (runId) { try { const remote = await getTaskRun(runId); @@ -2734,6 +2743,7 @@ export default function App() { status: 'completed', livePhaseLabel: undefined, taskWorkspaceId: remote.workspace_id, + taskRunId: runId, }); taskRunStream.setThinking(sessionId, false); await syncControlledDocsAfterTask(task.taskKey || 'intake'); @@ -2745,15 +2755,17 @@ export default function App() { status: 'pending_author_confirmation', livePhaseLabel: undefined, taskWorkspaceId: remote.workspace_id, + taskRunId: runId, }); taskRunStream.setThinking(sessionId, false); await syncControlledDocsAfterTask(task.taskKey || 'intake'); await chatWindow.refresh().catch(() => undefined); return; } - if (remote.status === 'running' || remote.status === 'pending') { + if (shouldSubscribeExistingRun(remote.status)) { // 原 run 仍在跑：重订阅，不 POST 新 run settled = false; + retainedRunId = runId; patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); return; diff --git a/web/src/components/chat/MessageBubble.tsx b/web/src/components/chat/MessageBubble.tsx index afcc5df9..4ff483a4 100644 --- a/web/src/components/chat/MessageBubble.tsx +++ b/web/src/components/chat/MessageBubble.tsx @@ -91,6 +91,8 @@ const MessageBubbleInner: React.FC = ({ id={message.serverId ? `chat-anchor-${message.serverId}` : (message.id ? `chat-anchor-${message.id}` : undefined)} data-server-id={message.serverId} data-category={message.category} + data-chat-role={message.role} + data-chat-status={status || (isUser ? 'user' : 'unknown')} className={isCategoryDimmed ? 'chat-category-dimmed' : undefined} onMouseEnter={() => setIsHovered(true)} onMouseLeave={() => setIsHovered(false)} diff --git a/web/src/lib/postTaskDeliverableSync.test.ts b/web/src/lib/postTaskDeliverableSync.test.ts index a6a2d363..e944fc4c 100644 --- a/web/src/lib/postTaskDeliverableSync.test.ts +++ b/web/src/lib/postTaskDeliverableSync.test.ts @@ -3,8 +3,10 @@ import assert from 'node:assert/strict'; import { resolvePostTaskActiveDocId, shouldSwitchTemplateSlotToServer, + pollUntilPostTaskServerDoc, } from './postTaskDeliverableSync'; import type { DeliverableItem } from '../types/workbench'; +import type { WorkflowDocumentSummary } from '../api/projectDocuments'; const base = (over: Partial): DeliverableItem => ({ docId: 'doc-x', @@ -88,3 +90,66 @@ test('shouldSwitchTemplateSlotToServer when active is local template slot', () = false, ); }); + +test('pollUntilPostTaskServerDoc retries until same taskKey server doc appears', async () => { + const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })]; + let calls = 0; + const snaps: DeliverableItem[][] = []; + const result = await pollUntilPostTaskServerDoc({ + listDocs: async () => { + calls += 1; + if (calls < 3) return []; + return [ + { + doc_id: 'doc_proj6089_intake', + project_id: 6089, + task_key: 'intake', + title: '用户初始需求收集表', + lifecycle_state: 'drafting', + is_template_only: false, + } satisfies WorkflowDocumentSummary, + ]; + }, + getDeliverables: () => local, + merge: (prev, docs) => { + if (docs.length === 0) return prev; + return [ + ...prev.filter((d) => d.docId !== 'doc_proj6089_intake'), + base({ + docId: docs[0].doc_id, + isTemplateOnly: false, + taskKey: docs[0].task_key, + }), + ]; + }, + projectId: 6089, + taskKey: 'intake', + maxAttempts: 5, + delayMs: 1, + sleep: async () => undefined, + onMerged: (merged) => { + snaps.push(merged); + }, + }); + assert.equal(result.serverDocId, 'doc_proj6089_intake'); + assert.equal(result.attempts, 3); + assert.ok(snaps.length >= 3); + assert.ok(snaps[snaps.length - 1].some((d) => d.docId === 'doc_proj6089_intake')); +}); + +test('pollUntilPostTaskServerDoc returns null after maxAttempts without relying on setState', async () => { + const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })]; + const result = await pollUntilPostTaskServerDoc({ + listDocs: async () => [], + getDeliverables: () => local, + merge: (prev) => prev, + projectId: 6089, + taskKey: 'intake', + maxAttempts: 3, + delayMs: 1, + sleep: async () => undefined, + }); + assert.equal(result.serverDocId, null); + assert.equal(result.attempts, 3); + assert.equal(result.merged[0]?.docId, 'doc-p6089-intake'); +}); diff --git a/web/src/lib/postTaskDeliverableSync.ts b/web/src/lib/postTaskDeliverableSync.ts index 545bb754..f49a040d 100644 --- a/web/src/lib/postTaskDeliverableSync.ts +++ b/web/src/lib/postTaskDeliverableSync.ts @@ -1,4 +1,5 @@ import type { DeliverableItem } from '../types/workbench'; +import type { WorkflowDocumentSummary } from '../api/projectDocuments'; /** * 任务完成后：在已 merge 的 deliverables 中定位同 taskKey 的服务端受控文档槽， @@ -63,3 +64,64 @@ export function shouldSwitchTemplateSlotToServer(params: { if (active.projectId !== projectId || active.taskKey !== taskKey) return false; return Boolean(active.isTemplateOnly) || active.docId.startsWith('doc-p'); } + +export type PollUntilPostTaskServerDocParams = { + listDocs: () => Promise; + /** 每次 attempt 读取当前本地快照（应用侧用 ref.current，禁止依赖 setState updater 回写）。 */ + getDeliverables: () => DeliverableItem[]; + merge: (prev: DeliverableItem[], docs: WorkflowDocumentSummary[]) => DeliverableItem[]; + projectId: number; + taskKey: string; + preferredDocId?: string | null; + maxAttempts?: number; + delayMs?: number; + sleep?: (ms: number) => Promise; + /** 每轮 merge 后回调（应用侧同步写 ref + setDeliverables(merged)）。 */ + onMerged?: (merged: DeliverableItem[]) => void; +}; + +/** + * Finalize 后文档可能尚未立即可见：有限重试 list+merge，直到同 taskKey 出现非模板 server doc。 + * 返回最后一次 merged 快照与解析到的 serverDocId（未出现则为 null）。 + */ +export async function pollUntilPostTaskServerDoc( + params: PollUntilPostTaskServerDocParams, +): Promise<{ merged: DeliverableItem[]; serverDocId: string | null; attempts: number }> { + const { + listDocs, + getDeliverables, + merge, + projectId, + taskKey, + preferredDocId, + maxAttempts = 10, + delayMs = 800, + sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + onMerged, + } = params; + + let merged = getDeliverables(); + let serverDocId: string | null = null; + let attempts = 0; + + for (let i = 0; i < maxAttempts; i++) { + attempts = i + 1; + const docs = await listDocs(); + merged = merge(getDeliverables(), docs); + onMerged?.(merged); + serverDocId = resolvePostTaskActiveDocId({ + deliverables: merged, + projectId, + taskKey, + preferredDocId, + }); + if (serverDocId) { + return { merged, serverDocId, attempts }; + } + if (i + 1 < maxAttempts) { + await sleep(delayMs); + } + } + + return { merged, serverDocId: null, attempts }; +} diff --git a/web/src/lib/taskRunReconnect.test.ts b/web/src/lib/taskRunReconnect.test.ts new file mode 100644 index 00000000..8bfb8212 --- /dev/null +++ b/web/src/lib/taskRunReconnect.test.ts @@ -0,0 +1,42 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { + resolveReconnectRunId, + shouldSubscribeExistingRun, + isRemoteTaskTerminal, + nextAttemptOpts, +} from './taskRunReconnect'; + +test('resolveReconnectRunId prefers explicit opts and ignores attempt without opts', () => { + assert.equal( + resolveReconnectRunId({ attemptIndex: 1, optsReconnectRunId: 'run_abc' }), + 'run_abc', + ); + assert.equal(resolveReconnectRunId({ attemptIndex: 0, optsReconnectRunId: null }), undefined); + assert.equal(resolveReconnectRunId({ attemptIndex: 2, optsReconnectRunId: ' ' }), undefined); + assert.equal(resolveReconnectRunId({ attemptIndex: 3 }), undefined); +}); + +test('shouldSubscribeExistingRun only for in-flight statuses', () => { + assert.equal(shouldSubscribeExistingRun('running'), true); + assert.equal(shouldSubscribeExistingRun('pending'), true); + assert.equal(shouldSubscribeExistingRun('completed'), false); + assert.equal(shouldSubscribeExistingRun('failed'), false); +}); + +test('isRemoteTaskTerminal covers confirmation and quality_failed', () => { + assert.equal(isRemoteTaskTerminal('pending_author_confirmation'), true); + assert.equal(isRemoteTaskTerminal('completed'), true); + assert.equal(isRemoteTaskTerminal('quality_failed'), true); + assert.equal(isRemoteTaskTerminal('running'), false); +}); + +test('nextAttemptOpts carries retained runId for reconnect without new POST', () => { + const next = nextAttemptOpts<{ reconnectRunId?: string; workDir?: string }>( + { workDir: '/w' }, + 'run_keep', + ); + assert.equal(next.reconnectRunId, 'run_keep'); + assert.equal(next.workDir, '/w'); + assert.equal(nextAttemptOpts(undefined, undefined).reconnectRunId, undefined); +}); diff --git a/web/src/lib/taskRunReconnect.ts b/web/src/lib/taskRunReconnect.ts new file mode 100644 index 00000000..237d28e5 --- /dev/null +++ b/web/src/lib/taskRunReconnect.ts @@ -0,0 +1,38 @@ +/** + * 长任务断线自动重试：用显式 reconnectRunId 跨 attempt 保留， + * 禁止依赖 setState updater / 已被清空的 stream.runId。 + */ + +export function resolveReconnectRunId(input: { + attemptIndex: number; + optsReconnectRunId?: string | null; +}): string | undefined { + const fromOpts = input.optsReconnectRunId?.trim(); + if (fromOpts) return fromOpts; + // attempt 0 且无显式 id → 允许新 POST；重试必须靠 opts 传入 + return undefined; +} + +export function shouldSubscribeExistingRun(status: string | null | undefined): boolean { + return status === 'running' || status === 'pending'; +} + +export function isRemoteTaskTerminal(status: string | null | undefined): boolean { + return ( + status === 'completed' || + status === 'pending_author_confirmation' || + status === 'quality_failed' + ); +} + +/** 自动重试调度时把已知 runId 写入下一轮 opts，避免 abort 后丢失。 */ +export function nextAttemptOpts( + opts: T | undefined, + retainedRunId: string | undefined, +): T & { reconnectRunId?: string } { + const base = { ...(opts || ({} as T)) }; + if (retainedRunId) { + return { ...base, reconnectRunId: retainedRunId }; + } + return base; +} From 301e1f401ee2d32994ec9568d7c394ecb081b34b Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 13 Sep 2026 14:01:41 +0000 Subject: [PATCH 3/5] fix(e2e): stable doc selector and this-turn assistant waits Add doc-selector-trigger / active-document-pane data attrs; resolve assistant via virtualizer data-index siblings; UI-02/03 and open-by-id helpers stop relying on global text or first dropdown-trigger. Co-authored-by: Cursor --- web/e2e/document-revision-runtime.spec.ts | 165 +++++++++++++++------- web/src/App.tsx | 11 +- web/src/components/doc/SlimTabBar.tsx | 3 + 3 files changed, 124 insertions(+), 55 deletions(-) diff --git a/web/e2e/document-revision-runtime.spec.ts b/web/e2e/document-revision-runtime.spec.ts index 630cd2f3..47973b4f 100644 --- a/web/e2e/document-revision-runtime.spec.ts +++ b/web/e2e/document-revision-runtime.spec.ts @@ -222,27 +222,45 @@ async function adminCreateUser( } /** - * Squad wait: observe THIS turn's user bubble, then its following assistant sibling - * running → terminal. Never match a prior completed bubble globally. + * Locate the assistant bubble for THIS user turn. + * MessageBubble is nested under a virtualizer outer with data-index; siblings are those outers. */ -async function waitForSquadCompletion(page: Page, userPromptSnippet: string, timeoutMs = 420_000) { - const startedAt = Date.now(); +function turnAssistantLocator(page: Page, userPromptSnippet: string) { const userBubble = page .locator('[data-chat-role="user"]') .filter({ hasText: userPromptSnippet }) .last(); - await expect(userBubble).toBeVisible({ timeout: 30_000 }); + return { + userBubble, + assistant: userBubble.locator( + 'xpath=ancestor::*[@data-index][1]/following-sibling::*[@data-index][1]//*[@data-chat-role="assistant"]', + ), + }; +} - const assistant = userBubble.locator( - 'xpath=following-sibling::*[@data-chat-role="assistant"][1]', - ); +/** + * Wait for this-turn assistant terminal. Optionally require observing inflight first. + * UI-03 may skip inflight (already seen in UI-02) to avoid fast-complete race. + */ +async function waitForSquadCompletion( + page: Page, + userPromptSnippet: string, + timeoutMs = 420_000, + opts?: { requireInflight?: boolean }, +) { + const startedAt = Date.now(); + const requireInflight = opts?.requireInflight !== false; + const { userBubble, assistant } = turnAssistantLocator(page, userPromptSnippet); + await expect(userBubble).toBeVisible({ timeout: 30_000 }); await expect(assistant).toBeVisible({ timeout: 30_000 }); - await expect(assistant).toHaveAttribute( - 'data-chat-status', - /pending|streaming_status|auto_retrying/, - { timeout: 60_000 }, - ); + if (requireInflight) { + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /pending|streaming_status|auto_retrying/, + { timeout: 60_000 }, + ); + } await expect(assistant).toHaveAttribute( 'data-chat-status', @@ -257,23 +275,32 @@ async function waitForSquadCompletion(page: Page, userPromptSnippet: string, tim return Date.now() - startedAt; } -async function openWorkflowDocByLabel(page: Page, label: RegExp) { - const trigger = page.locator('button.dropdown-trigger').first(); - if ((await trigger.count()) > 0) { - await trigger.click(); - const item = page - .locator('.dropdown-menu button, .dropdown-menu [role="menuitem"]') - .filter({ hasText: label }) - .first(); - if ((await item.count()) === 0) fail(`workflow slot not in SlimTabBar dropdown: ${label}`); - await item.click(); - return; +async function openWorkflowDocByLabel( + page: Page, + label: RegExp, +): Promise<{ title: string; docId: string }> { + const trigger = page.getByTestId('doc-selector-trigger'); + await expect(trigger).toBeVisible({ timeout: 15_000 }); + await trigger.click(); + const item = page.locator('.dropdown-menu button[data-doc-id]').filter({ hasText: label }).first(); + if ((await item.count()) === 0) { + fail(`workflow slot not in doc-selector menu: ${label}`); } - const deck = page.locator('[data-testid="deck-drawer"], .deck-drawer').first(); - if ((await deck.count()) > 0) await deck.click(); - const card = page.locator('[data-doc-id^="doc_proj"]').filter({ hasText: label }).first(); - if ((await card.count()) === 0) fail(`no server-backed workflow card for ${label}`); - await card.click(); + const docId = (await item.getAttribute('data-doc-id')) || ''; + const title = (await item.getAttribute('data-doc-title')) || ''; + await item.click(); + if (!docId) fail(`selected menu item missing data-doc-id for ${label}`); + return { title: title || docId, docId }; +} + +async function openWorkflowDocById(page: Page, docId: string): Promise { + if (!docId) fail('openWorkflowDocById requires docId'); + const trigger = page.getByTestId('doc-selector-trigger'); + await expect(trigger).toBeVisible({ timeout: 15_000 }); + await trigger.click(); + const item = page.locator(`.dropdown-menu button[data-doc-id="${docId}"]`).first(); + if ((await item.count()) === 0) fail(`doc-selector menu missing data-doc-id=${docId}`); + await item.click(); } async function sendChat(page: Page, text: string) { @@ -344,8 +371,10 @@ test.describe('document-revision-runtime UI @post-restart', () => { // Locate project by visible name. await page.getByText(st.projectName, { exact: false }).first().click(); await expect(page.getByText(st.projectName)).toBeVisible({ timeout: 20_000 }); - if (st.controlledDocTitle) { - await page.getByText(st.controlledDocTitle, { exact: false }).first().click(); + if (st.controlledDocId) { + await openWorkflowDocById(page, st.controlledDocId); + } else if (st.controlledDocTitle) { + await openWorkflowDocByLabel(page, new RegExp(st.controlledDocTitle.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'))); } await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 30_000 }); const { v, r } = await readVersionBadge(page); @@ -375,8 +404,10 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { let page: Page; let logs: NetLog[] = []; let controlledTitle = ''; + let controlledDocId = ''; let versionBeforePending = ''; let secondDocTitle = ''; + let secondDocId = ''; test.beforeAll(async ({ browser }) => { context = await browser.newContext(); @@ -535,15 +566,20 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { if (await connErr.first().isVisible().catch(() => false)) { fail('UI-02 connection-error banner visible after tasks/runs — product FAIL'); } - // 运行态仅为额外 UI 断言，不能单独构成 PASS。 - const running = page.locator('text=正在思考').or(page.locator('text=生成中')); - await expect(running.first()).toBeVisible({ timeout: 60_000 }); + // 对本轮 user 对应 assistant 精确断言 inflight（禁止全局「生成中」）。 + const { assistant } = turnAssistantLocator(page, prompt.slice(0, 12)); + await expect(assistant).toBeVisible({ timeout: 30_000 }); + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /pending|streaming_status|auto_retrying/, + { timeout: 60_000 }, + ); await shot(page, 'UI-02-squad-started.png'); recordPass({ id: 'UI-02', name: 'start real squad', input: prompt, - actual: `POST /api/v1/tasks/runs HTTP ${status}; running/thinking visible`, + actual: `POST /api/v1/tasks/runs HTTP ${status}; this-turn assistant inflight`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', @@ -553,24 +589,26 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { test('UI-03 wait first round; open controlled doc with V/r', async () => { const t0 = Date.now(); const promptSlice = '请基于本项目编制首轮'; - await waitForSquadCompletion(page, promptSlice); - // 精确打开工作流文档槽：SlimTabBar 下拉 / DeckDrawer，禁止全局 div 泛匹配。 - await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); + // UI-02 已观测 inflight；此处只等同一 assistant 到 terminal，避免快速完成竞态。 + await waitForSquadCompletion(page, promptSlice, 420_000, { requireInflight: false }); + const selected = await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); + controlledDocId = selected.docId; + controlledTitle = selected.title; + if (!/^doc_proj\d+_/.test(controlledDocId)) { + fail(`UI-03: selected slot must be server doc_proj*, got ${controlledDocId}`); + } + const pane = page.getByTestId('active-document-pane'); + await expect(pane).toHaveAttribute('data-doc-id', controlledDocId, { timeout: 15_000 }); await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 60_000 }); await expect(page.getByTestId('doc-rev-template-hint')).toHaveCount(0); await expect(page.locator('text=版本账本未就绪')).toHaveCount(0); - const activeDoc = page.locator('[data-doc-id^="doc_proj"]').first(); - if ((await activeDoc.count()) === 0) fail('UI-03: active pane must be server doc_proj* not doc-p template'); const { v, r } = await readVersionBadge(page); - controlledTitle = (await page.locator('[data-testid="doc-rev-version"]').evaluate((el) => { - const root = el.closest('[data-doc-id]') || el.parentElement; - return root?.getAttribute('data-doc-id') || 'controlled'; - }).catch(() => 'controlled')) as string; await shot(page, '03-first-doc.png'); const st = readPhaseState(); writePhaseState({ ...st, controlledDocTitle: controlledTitle, + controlledDocId, versionText: v, revisionText: r, }); @@ -578,7 +616,7 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { id: 'UI-03', name: 'first controlled doc ledger', input: 'wait squad; open intake server doc slot', - actual: `toolbar ${v}/${r}; server doc visible`, + actual: `toolbar ${v}/${r}; docId=${controlledDocId}; title=${controlledTitle}`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', @@ -796,15 +834,21 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { '请另行生成一份《系统设计输入》受控文档（第二份），不要覆盖上一份需求文档，并给出可打开的交付物。'; await sendChat(page, prompt); await waitForSquadCompletion(page, '系统设计输入'); - await openWorkflowDocByLabel(page, /设计输入|系统设计/); - secondDocTitle = '系统设计输入'; + const second = await openWorkflowDocByLabel(page, /设计输入|系统设计/); + secondDocTitle = second.title; + secondDocId = second.docId; + await expect(page.getByTestId('active-document-pane')).toHaveAttribute('data-doc-id', secondDocId); await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 45_000 }); const states: string[] = []; - for (const label of [controlledTitle, secondDocTitle]) { - if (!label) continue; - await openWorkflowDocByLabel(page, new RegExp(label.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'))); + for (const target of [ + { id: controlledDocId, title: controlledTitle }, + { id: secondDocId, title: secondDocTitle }, + ]) { + if (!target.id) continue; + await openWorkflowDocById(page, target.id); + await expect(page.getByTestId('active-document-pane')).toHaveAttribute('data-doc-id', target.id); const { v, r } = await readVersionBadge(page); - states.push(`${label}:${v}/${r}`); + states.push(`${target.title}:${v}/${r}`); } // Page upload attachment const fileInput = page.locator('input[type="file"]'); @@ -857,14 +901,22 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { const t0 = Date.now(); logs = []; // Ensure A is on controlled doc - await page.getByText(controlledTitle || PROJECT_NAME, { exact: false }).first().click(); + if (controlledDocId) { + await openWorkflowDocById(page, controlledDocId); + } else { + await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); + } await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 20_000 }); const ctxB = await page.context().browser()!.newContext(); const pageB = await ctxB.newPage(); await login(pageB, USER_B, FINAL_PASS_B); await pageB.getByText(PROJECT_NAME, { exact: false }).first().click(); - await pageB.getByText(controlledTitle || /需求/, { exact: false }).first().click(); + if (controlledDocId) { + await openWorkflowDocById(pageB, controlledDocId); + } else { + await openWorkflowDocByLabel(pageB, /用户初始需求|需求摸底|intake/i); + } await expect(pageB.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 30_000 }); // Both open same old revision — A saves first @@ -900,7 +952,11 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { await logoutViaUI(page); await login(page, USER_A, FINAL_PASS_A); await page.getByText(PROJECT_NAME, { exact: false }).first().click(); - await page.getByText(controlledTitle || /需求/, { exact: false }).first().click(); + if (controlledDocId) { + await openWorkflowDocById(page, controlledDocId); + } else { + await openWorkflowDocByLabel(page, /用户初始需求|需求摸底|intake/i); + } await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 30_000 }); const again = await readVersionBadge(page); if (again.v !== v) fail(`version changed across re-login unexpectedly ${v}→${again.v}`); @@ -925,6 +981,7 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { ...st, projectName: PROJECT_NAME, controlledDocTitle: controlledTitle, + controlledDocId, versionText: v, revisionText: r, userA: USER_A, diff --git a/web/src/App.tsx b/web/src/App.tsx index 87ea8eb6..4e29cd22 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -4026,7 +4026,16 @@ export default function App() {

) : ( -
+
d.docId === activeDocId)?.title || + visibleDeliverables.find((d) => d.docId === activeDocId)?.title || + '' + } + > {/* 36px 极简 Tab 条 (含三分类分段切换与私密/共享识别) */} = ({