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