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..62e7c3dd --- /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. 前端断线自动重试与 DB 约束异常暴露 + +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：同 turn 自动重试路径再次进入处理中（工程修复：`reconnectRunId` 跨 attempt + Claim 幂等） +2. **暴露 `task_runs` 状态约束缺陷**： + 在二次运行与高风险候选流转路径中，后台服务尝试将 `task_runs.status` 更新为 `pending_author_confirmation` 时，暴露了数据库底层约束拦截： + goose **`00019`** 原始 `chk_task_run_status` 仅允许： + ```sql + CHECK (status IN ('pending', 'running', 'completed', 'failed', 'cancelled', 'finalizing_failed', 'quality_failed', 'interrupted')) + ``` + 该枚举**未包含** `pending_author_confirmation`，导致写入状态时触发 PostgreSQL check constraint 违规（由 goose `00030` 补齐）。 + +--- + +## 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 受控文档 | `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/.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..1cc97d9e --- /dev/null +++ b/internal/taskrun/claim_idempotency_test.go @@ -0,0 +1,264 @@ +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)) + } +} + +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 286c7493..2efbd448 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,142 @@ 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) + 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) } } 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 +280,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 +491,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..13abb602 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 } @@ -169,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, @@ -190,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, ) @@ -213,9 +236,35 @@ 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 } +// 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 +477,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..2271cbb9 100644 --- a/web/e2e/document-revision-runtime.spec.ts +++ b/web/e2e/document-revision-runtime.spec.ts @@ -222,40 +222,106 @@ async function adminCreateUser( } /** - * Squad wait: MUST observe start (user bubble + running/thinking) before terminal. - * failed status → FAIL. + * 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) { +function turnAssistantLocator(page: Page, userPromptSnippet: string) { + const userBubble = page + .locator('[data-chat-role="user"]') + .filter({ hasText: userPromptSnippet }) + .last(); + return { + userBubble, + assistant: userBubble.locator( + 'xpath=ancestor::*[@data-index][1]/following-sibling::*[@data-index][1]//*[@data-chat-role="assistant"]', + ), + }; +} + +/** + * 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 userBubble = page.locator('div, article, li').filter({ hasText: userPromptSnippet }).first(); + 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 }); - 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 }); + if (requireInflight) { + await expect(assistant).toHaveAttribute( + 'data-chat-status', + /pending|streaming_status|auto_retrying/, + { timeout: 60_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', + /completed|pending_author_confirmation|failed|cancelled|quality_failed/, + { timeout: timeoutMs }, + ); - await expect(terminal.first()).toBeVisible({ timeout: timeoutMs }); - - if (await page.locator('text=失败').or(page.locator('[data-status="failed"]')).first().isVisible().catch(() => false)) { - fail('squad terminal status is failed'); + const terminal = (await assistant.getAttribute('data-chat-status')) || ''; + if ( + terminal === 'failed' || + terminal === 'cancelled' || + terminal === 'quality_failed' + ) { + fail(`squad terminal status is ${terminal}`); + } + if (terminal !== 'completed' && terminal !== 'pending_author_confirmation') { + fail(`squad terminal status unexpected: ${terminal}`); } - // Ensure running indicator has cleared. - await expect(page.locator('text=正在思考，请稍候...')).toBeHidden({ timeout: 60_000 }).catch(() => undefined); return Date.now() - startedAt; } +async function selectDeliverableCategory( + page: Page, + category: 'workflow' | 'shared' | 'private', +) { + const trigger = page.getByTestId('doc-category-trigger'); + await expect(trigger).toBeVisible({ timeout: 15_000 }); + await trigger.click(); + const item = page.locator(`.dropdown-menu button[data-category="${category}"]`).first(); + if ((await item.count()) === 0) fail(`category menu missing data-category=${category}`); + await item.click(); +} + +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 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) { const box = page.getByPlaceholder(/向项目助手提问/); await expect(box).toBeVisible({ timeout: 15_000 }); @@ -324,8 +390,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); @@ -355,8 +423,11 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { let page: Page; let logs: NetLog[] = []; let controlledTitle = ''; + let controlledDocId = ''; let versionBeforePending = ''; let secondDocTitle = ''; + let secondDocId = ''; + let uploadedDocId = ''; test.beforeAll(async ({ browser }) => { context = await browser.newContext(); @@ -515,15 +586,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', @@ -533,29 +609,34 @@ 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); - // 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(); + // 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 { v, r } = await readVersionBadge(page); - controlledTitle = (await tab.textContent())?.trim() || 'controlled'; await shot(page, '03-first-doc.png'); const st = readPhaseState(); writePhaseState({ ...st, controlledDocTitle: controlledTitle, + controlledDocId, versionText: v, revisionText: r, }); 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}; docId=${controlledDocId}; title=${controlledTitle}`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', @@ -773,33 +854,76 @@ 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() || ''; + 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 page.getByText(label, { exact: false }).first().click(); + 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 + // Page upload: chip draft → send → POST /api/v1/freeform-documents → doc-free pane. const fileInput = page.locator('input[type="file"]'); if ((await fileInput.count()) === 0) fail('no file input for page upload'); const tmp = path.join(ASSETS, `upload-${suffix}.txt`); + const uploadName = `upload-${suffix}.txt`; fs.writeFileSync(tmp, `attachment ${suffix}\n`, 'utf-8'); await fileInput.first().setInputFiles(tmp); - await expect(page.getByText(`upload-${suffix}`).or(page.getByText('attachment'))).toBeVisible({ - timeout: 30_000, - }); + await expect( + page.getByText(uploadName, { exact: false }).or(page.getByText(`upload-${suffix}`, { exact: false })), + ).toBeVisible({ timeout: 15_000 }); + + const archivePrompt = '请将此附件归档为我的私密参考文档。'; + const freeformWait = page.waitForResponse( + (r) => { + const pathOnly = sanitizeUrl(r.url()); + return ( + r.request().method() === 'POST' && + pathOnly.includes('/api/v1/freeform-documents') && + !pathOnly.includes('/from-message') + ); + }, + { timeout: 60_000 }, + ); + const box = page.getByPlaceholder(/向项目助手提问/); + await expect(box).toBeVisible({ timeout: 15_000 }); + // 附件允许空文本；填归档句以便本轮 assistant 可定位并收口终态。 + await box.fill(archivePrompt); + await page.getByRole('button', { name: /发送/ }).click(); + + const freeformRes = await freeformWait; + const freeformStatus = freeformRes.status(); + if (freeformStatus < 200 || freeformStatus >= 300) { + fail(`UI-11 POST /api/v1/freeform-documents must be 2xx, got HTTP ${freeformStatus}`); + } + + const pane = page.getByTestId('active-document-pane'); + await expect(pane).toHaveAttribute('data-doc-id', /^doc-free-/, { timeout: 45_000 }); + uploadedDocId = (await pane.getAttribute('data-doc-id')) || ''; + if (!uploadedDocId.startsWith('doc-free-')) { + fail(`UI-11 upload must activate doc-free-* pane, got ${uploadedDocId}`); + } + if (uploadedDocId.includes('doc-upload')) { + fail(`UI-11 must not leave doc-upload orphan active: ${uploadedDocId}`); + } + + // 发送会拉起普通 assistant：离开 UI-11 前等本轮终态，避免污染 UI-12+。 + await waitForSquadCompletion(page, '请将此附件归档', 180_000); + await shot(page, 'UI-11-multi-doc.png'); recordPass({ id: 'UI-11', name: 'multi doc + upload', - input: 'chat second doc; upload file', - actual: states.join('; '), + input: 'chat second doc; upload file + send for freeform register', + actual: `${states.join('; ')}; freeform HTTP ${freeformStatus}; uploadedDocId=${uploadedDocId}`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', @@ -836,14 +960,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 @@ -879,7 +1011,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}`); @@ -904,6 +1040,7 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { ...st, projectName: PROJECT_NAME, controlledDocTitle: controlledTitle, + controlledDocId, versionText: v, revisionText: r, userA: USER_A, @@ -928,11 +1065,19 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { test('UI-16 registered upload uses server doc-free ledger (no doc-upload orphan)', async () => { const t0 = Date.now(); logs = []; - const uploadTab = page.getByText(`upload-${suffix}`).or(page.getByText(/attachment/)); - if ((await uploadTab.count()) === 0) fail('upload tab missing (UI-11 must have uploaded via page)'); - await uploadTab.first().click(); + if (!uploadedDocId || !uploadedDocId.startsWith('doc-free-')) { + fail(`UI-16 requires uploadedDocId from UI-11 (doc-free-*), got ${uploadedDocId || '(empty)'}`); + } + await selectDeliverableCategory(page, 'private'); + await openWorkflowDocById(page, uploadedDocId); + const pane = page.getByTestId('active-document-pane'); + await expect(pane).toHaveAttribute('data-doc-id', uploadedDocId, { timeout: 15_000 }); + await expect(pane).toHaveAttribute('data-doc-id', /^doc-free-/); + const paneId = (await pane.getAttribute('data-doc-id')) || ''; + if (paneId.startsWith('doc-upload')) { + fail(`UI-16 must not open doc-upload orphan: ${paneId}`); + } await expect(page.locator('text=版本账本未就绪')).toHaveCount(0); - // Registered upload must mount real toolbar (doc-free), not orphan hint. await expect(page.locator('[data-testid="doc-rev-version"]')).toBeVisible({ timeout: 30_000 }); const orphanHits = logs.filter( (l) => l.path.includes('/revision-state') && /\/documents\/doc-upload-/.test(l.path), @@ -944,8 +1089,8 @@ test.describe.serial('document-revision-runtime UI acceptance', () => { recordPass({ id: 'UI-16', name: 'registered upload server ledger', - input: 'open page-uploaded attachment after createFreeform registration', - actual: 'toolbar V visible; no doc-upload revision-state', + input: `category=private; open exact ${uploadedDocId}`, + actual: `pane=${paneId}; toolbar V visible; no doc-upload revision-state`, networkLogs: logs, durationMs: Date.now() - t0, verdict: 'PASS', diff --git a/web/src/App.tsx b/web/src/App.tsx index 52a7b8fb..4e29cd22 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -48,6 +48,16 @@ import { toApiProjectRole, } from './lib/projectMemberAdapter'; import { mergeWorkflowDocumentsFromServer } from './lib/deliverableSync'; +import { + pollUntilPostTaskServerDoc, + shouldSwitchTemplateSlotToServer, +} from './lib/postTaskDeliverableSync'; +import { + isRemoteTaskTerminal, + nextAttemptOpts, + resolveReconnectRunId, + shouldSubscribeExistingRun, +} from './lib/taskRunReconnect'; import { buildControlledLoadErrorSkeleton, buildControlledLoadingSkeleton, @@ -580,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(() => { @@ -2457,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); @@ -2481,7 +2502,7 @@ export default function App() { taskRunStream.setThinking(sessionId, false); return; } - runTaskAttempt(task, asstId, mode, attemptIndex + 1, opts); + runTaskAttempt(task, asstId, mode, attemptIndex + 1, nextAttemptOpts(opts, retainedRunId)); }, delayMs); }; @@ -2500,13 +2521,65 @@ 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 { 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: merged, + 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); @@ -2570,11 +2643,12 @@ 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 () => { let nextStatus: AssistantMessageStatus = 'completed'; + let preferredDocId: string | null = null; if (finishedRunId) { const remote = await getTaskRun(finishedRunId).catch(() => null); if (remote?.status) { @@ -2586,7 +2660,9 @@ 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); })(); break; @@ -2603,6 +2679,35 @@ export default function App() { : buildTaskRequest(task); if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') { + // 断线重试：显式 reconnectRunId 保留；禁止 setMessages updater 取 runId + const existingRunId = resolveReconnectRunId({ + attemptIndex, + optsReconnectRunId: retainedRunId || opts?.reconnectRunId, + }); + if (attemptIndex > 0 && existingRunId) { + const remote = await getTaskRun(existingRunId).catch(() => null); + if (remote && shouldSubscribeExistingRun(remote.status)) { + retainedRunId = existingRunId; + taskRunStream.setRunId(sessionId, existingRunId); + patchAsst({ taskRunId: existingRunId, livePhaseLabel: '正在重连任务…' }); + await subscribeTaskRunEvents(existingRunId, { onEvent }, controller.signal); + return; + } + if (remote && isRemoteTaskTerminal(remote.status)) { + 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, @@ -2611,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); @@ -2627,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); @@ -2637,8 +2743,10 @@ export default function App() { status: 'completed', livePhaseLabel: undefined, taskWorkspaceId: remote.workspace_id, + taskRunId: runId, }); taskRunStream.setThinking(sessionId, false); + await syncControlledDocsAfterTask(task.taskKey || 'intake'); return; } if (remote.status === 'pending_author_confirmation') { @@ -2647,11 +2755,21 @@ 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 (shouldSubscribeExistingRun(remote.status)) { + // 原 run 仍在跑：重订阅，不 POST 新 run + settled = false; + retainedRunId = runId; + patchAsst({ livePhaseLabel: '正在重连任务…', taskRunId: runId }); + await subscribeTaskRunEvents(runId, { onEvent }, controller.signal); + return; + } if (remote.status === 'failed' || remote.status === 'cancelled') { failOrRetry(buildNetworkErrorInfo()); return; @@ -3908,7 +4026,16 @@ export default function App() {

) : ( -
+
d.docId === activeDocId)?.title || + visibleDeliverables.find((d) => d.docId === activeDocId)?.title || + '' + } + > {/* 36px 极简 Tab 条 (含三分类分段切换与私密/共享识别) */} = ({ 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/components/doc/SlimTabBar.tsx b/web/src/components/doc/SlimTabBar.tsx index 0d322fd5..764df614 100644 --- a/web/src/components/doc/SlimTabBar.tsx +++ b/web/src/components/doc/SlimTabBar.tsx @@ -177,6 +177,7 @@ export const SlimTabBar: React.FC = ({