merge: fix/document-revision-runtime-ui03
This commit is contained in:
weichen
@@ -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)` 卷的迁移实测。
|
||||
+3
-3
@@ -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/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。
|
||||
|
||||
|
||||
@@ -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（历史）
|
||||
|
||||
@@ -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 当验收 |
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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'
|
||||
)
|
||||
);
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
+161
-22
@@ -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]
|
||||
|
||||
+67
-15
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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',
|
||||
|
||||
+134
-7
@@ -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<DeliverableItem[]>(() =>
|
||||
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() {
|
||||
</p>
|
||||
</div>
|
||||
) : (
|
||||
<div style={{ flex: 1, display: 'flex', flexDirection: 'column', height: '100%', overflow: 'hidden', position: 'relative' }}>
|
||||
<div
|
||||
style={{ flex: 1, display: 'flex', flexDirection: 'column', height: '100%', overflow: 'hidden', position: 'relative' }}
|
||||
data-testid="active-document-pane"
|
||||
data-doc-id={activeDocId}
|
||||
data-doc-title={
|
||||
deliverables.find((d) => d.docId === activeDocId)?.title ||
|
||||
visibleDeliverables.find((d) => d.docId === activeDocId)?.title ||
|
||||
''
|
||||
}
|
||||
>
|
||||
{/* 36px 极简 Tab 条 (含三分类分段切换与私密/共享识别) */}
|
||||
<SlimTabBar
|
||||
deliverables={visibleDeliverables}
|
||||
|
||||
@@ -91,6 +91,8 @@ const MessageBubbleInner: React.FC<MessageBubbleProps> = ({
|
||||
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)}
|
||||
|
||||
@@ -177,6 +177,7 @@ export const SlimTabBar: React.FC<SlimTabBarProps> = ({
|
||||
<div style={{ position: 'relative', flexShrink: 0 }}>
|
||||
<button
|
||||
className="dropdown-trigger"
|
||||
data-testid="doc-category-trigger"
|
||||
onClick={() => {
|
||||
setShowCategoryMenu(!showCategoryMenu);
|
||||
setShowDocMenu(false);
|
||||
@@ -239,6 +240,7 @@ export const SlimTabBar: React.FC<SlimTabBarProps> = ({
|
||||
return (
|
||||
<button
|
||||
key={cat}
|
||||
data-category={cat}
|
||||
onClick={() => {
|
||||
handleCategoryClick(cat);
|
||||
setShowCategoryMenu(false);
|
||||
@@ -277,6 +279,7 @@ export const SlimTabBar: React.FC<SlimTabBarProps> = ({
|
||||
<div style={{ position: 'relative', flex: 1, minWidth: 0 }}>
|
||||
<button
|
||||
className="dropdown-trigger"
|
||||
data-testid="doc-selector-trigger"
|
||||
onClick={() => {
|
||||
setShowDocMenu(!showDocMenu);
|
||||
setShowCategoryMenu(false);
|
||||
@@ -385,6 +388,8 @@ export const SlimTabBar: React.FC<SlimTabBarProps> = ({
|
||||
return (
|
||||
<button
|
||||
key={doc.docId}
|
||||
data-doc-id={doc.docId}
|
||||
data-doc-title={doc.title}
|
||||
onClick={() => {
|
||||
onSelectDoc(doc.docId);
|
||||
setShowDocMenu(false);
|
||||
|
||||
@@ -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');
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {
|
||||
resolvePostTaskActiveDocId,
|
||||
shouldSwitchTemplateSlotToServer,
|
||||
pollUntilPostTaskServerDoc,
|
||||
} from './postTaskDeliverableSync';
|
||||
import type { DeliverableItem } from '../types/workbench';
|
||||
import type { WorkflowDocumentSummary } from '../api/projectDocuments';
|
||||
|
||||
const base = (over: Partial<DeliverableItem>): 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,
|
||||
);
|
||||
});
|
||||
|
||||
test('pollUntilPostTaskServerDoc retries until same taskKey server doc appears', async () => {
|
||||
const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })];
|
||||
let calls = 0;
|
||||
const snaps: DeliverableItem[][] = [];
|
||||
const result = await pollUntilPostTaskServerDoc({
|
||||
listDocs: async () => {
|
||||
calls += 1;
|
||||
if (calls < 3) return [];
|
||||
return [
|
||||
{
|
||||
doc_id: 'doc_proj6089_intake',
|
||||
project_id: 6089,
|
||||
task_key: 'intake',
|
||||
title: '用户初始需求收集表',
|
||||
lifecycle_state: 'drafting',
|
||||
is_template_only: false,
|
||||
} satisfies WorkflowDocumentSummary,
|
||||
];
|
||||
},
|
||||
getDeliverables: () => local,
|
||||
merge: (prev, docs) => {
|
||||
if (docs.length === 0) return prev;
|
||||
return [
|
||||
...prev.filter((d) => d.docId !== 'doc_proj6089_intake'),
|
||||
base({
|
||||
docId: docs[0].doc_id,
|
||||
isTemplateOnly: false,
|
||||
taskKey: docs[0].task_key,
|
||||
}),
|
||||
];
|
||||
},
|
||||
projectId: 6089,
|
||||
taskKey: 'intake',
|
||||
maxAttempts: 5,
|
||||
delayMs: 1,
|
||||
sleep: async () => undefined,
|
||||
onMerged: (merged) => {
|
||||
snaps.push(merged);
|
||||
},
|
||||
});
|
||||
assert.equal(result.serverDocId, 'doc_proj6089_intake');
|
||||
assert.equal(result.attempts, 3);
|
||||
assert.ok(snaps.length >= 3);
|
||||
assert.ok(snaps[snaps.length - 1].some((d) => d.docId === 'doc_proj6089_intake'));
|
||||
});
|
||||
|
||||
test('pollUntilPostTaskServerDoc returns null after maxAttempts without relying on setState', async () => {
|
||||
const local = [base({ docId: 'doc-p6089-intake', isTemplateOnly: true })];
|
||||
const result = await pollUntilPostTaskServerDoc({
|
||||
listDocs: async () => [],
|
||||
getDeliverables: () => local,
|
||||
merge: (prev) => prev,
|
||||
projectId: 6089,
|
||||
taskKey: 'intake',
|
||||
maxAttempts: 3,
|
||||
delayMs: 1,
|
||||
sleep: async () => undefined,
|
||||
});
|
||||
assert.equal(result.serverDocId, null);
|
||||
assert.equal(result.attempts, 3);
|
||||
assert.equal(result.merged[0]?.docId, 'doc-p6089-intake');
|
||||
});
|
||||
@@ -0,0 +1,127 @@
|
||||
import type { DeliverableItem } from '../types/workbench';
|
||||
import type { WorkflowDocumentSummary } from '../api/projectDocuments';
|
||||
|
||||
/**
|
||||
* 任务完成后：在已 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');
|
||||
}
|
||||
|
||||
export type PollUntilPostTaskServerDocParams = {
|
||||
listDocs: () => Promise<WorkflowDocumentSummary[]>;
|
||||
/** 每次 attempt 读取当前本地快照（应用侧用 ref.current，禁止依赖 setState updater 回写）。 */
|
||||
getDeliverables: () => DeliverableItem[];
|
||||
merge: (prev: DeliverableItem[], docs: WorkflowDocumentSummary[]) => DeliverableItem[];
|
||||
projectId: number;
|
||||
taskKey: string;
|
||||
preferredDocId?: string | null;
|
||||
maxAttempts?: number;
|
||||
delayMs?: number;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
/** 每轮 merge 后回调（应用侧同步写 ref + setDeliverables(merged)）。 */
|
||||
onMerged?: (merged: DeliverableItem[]) => void;
|
||||
};
|
||||
|
||||
/**
|
||||
* Finalize 后文档可能尚未立即可见：有限重试 list+merge，直到同 taskKey 出现非模板 server doc。
|
||||
* 返回最后一次 merged 快照与解析到的 serverDocId（未出现则为 null）。
|
||||
*/
|
||||
export async function pollUntilPostTaskServerDoc(
|
||||
params: PollUntilPostTaskServerDocParams,
|
||||
): Promise<{ merged: DeliverableItem[]; serverDocId: string | null; attempts: number }> {
|
||||
const {
|
||||
listDocs,
|
||||
getDeliverables,
|
||||
merge,
|
||||
projectId,
|
||||
taskKey,
|
||||
preferredDocId,
|
||||
maxAttempts = 10,
|
||||
delayMs = 800,
|
||||
sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
|
||||
onMerged,
|
||||
} = params;
|
||||
|
||||
let merged = getDeliverables();
|
||||
let serverDocId: string | null = null;
|
||||
let attempts = 0;
|
||||
|
||||
for (let i = 0; i < maxAttempts; i++) {
|
||||
attempts = i + 1;
|
||||
const docs = await listDocs();
|
||||
merged = merge(getDeliverables(), docs);
|
||||
onMerged?.(merged);
|
||||
serverDocId = resolvePostTaskActiveDocId({
|
||||
deliverables: merged,
|
||||
projectId,
|
||||
taskKey,
|
||||
preferredDocId,
|
||||
});
|
||||
if (serverDocId) {
|
||||
return { merged, serverDocId, attempts };
|
||||
}
|
||||
if (i + 1 < maxAttempts) {
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
|
||||
return { merged, serverDocId: null, attempts };
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {
|
||||
resolveReconnectRunId,
|
||||
shouldSubscribeExistingRun,
|
||||
isRemoteTaskTerminal,
|
||||
nextAttemptOpts,
|
||||
} from './taskRunReconnect';
|
||||
|
||||
test('resolveReconnectRunId prefers explicit opts and ignores attempt without opts', () => {
|
||||
assert.equal(
|
||||
resolveReconnectRunId({ attemptIndex: 1, optsReconnectRunId: 'run_abc' }),
|
||||
'run_abc',
|
||||
);
|
||||
assert.equal(resolveReconnectRunId({ attemptIndex: 0, optsReconnectRunId: null }), undefined);
|
||||
assert.equal(resolveReconnectRunId({ attemptIndex: 2, optsReconnectRunId: ' ' }), undefined);
|
||||
assert.equal(resolveReconnectRunId({ attemptIndex: 3 }), undefined);
|
||||
});
|
||||
|
||||
test('shouldSubscribeExistingRun only for in-flight statuses', () => {
|
||||
assert.equal(shouldSubscribeExistingRun('running'), true);
|
||||
assert.equal(shouldSubscribeExistingRun('pending'), true);
|
||||
assert.equal(shouldSubscribeExistingRun('completed'), false);
|
||||
assert.equal(shouldSubscribeExistingRun('failed'), false);
|
||||
});
|
||||
|
||||
test('isRemoteTaskTerminal covers confirmation and quality_failed', () => {
|
||||
assert.equal(isRemoteTaskTerminal('pending_author_confirmation'), true);
|
||||
assert.equal(isRemoteTaskTerminal('completed'), true);
|
||||
assert.equal(isRemoteTaskTerminal('quality_failed'), true);
|
||||
assert.equal(isRemoteTaskTerminal('running'), false);
|
||||
});
|
||||
|
||||
test('nextAttemptOpts carries retained runId for reconnect without new POST', () => {
|
||||
const next = nextAttemptOpts<{ reconnectRunId?: string; workDir?: string }>(
|
||||
{ workDir: '/w' },
|
||||
'run_keep',
|
||||
);
|
||||
assert.equal(next.reconnectRunId, 'run_keep');
|
||||
assert.equal(next.workDir, '/w');
|
||||
assert.equal(nextAttemptOpts(undefined, undefined).reconnectRunId, undefined);
|
||||
});
|
||||
@@ -0,0 +1,38 @@
|
||||
/**
|
||||
* 长任务断线自动重试：用显式 reconnectRunId 跨 attempt 保留，
|
||||
* 禁止依赖 setState updater / 已被清空的 stream.runId。
|
||||
*/
|
||||
|
||||
export function resolveReconnectRunId(input: {
|
||||
attemptIndex: number;
|
||||
optsReconnectRunId?: string | null;
|
||||
}): string | undefined {
|
||||
const fromOpts = input.optsReconnectRunId?.trim();
|
||||
if (fromOpts) return fromOpts;
|
||||
// attempt 0 且无显式 id → 允许新 POST；重试必须靠 opts 传入
|
||||
return undefined;
|
||||
}
|
||||
|
||||
export function shouldSubscribeExistingRun(status: string | null | undefined): boolean {
|
||||
return status === 'running' || status === 'pending';
|
||||
}
|
||||
|
||||
export function isRemoteTaskTerminal(status: string | null | undefined): boolean {
|
||||
return (
|
||||
status === 'completed' ||
|
||||
status === 'pending_author_confirmation' ||
|
||||
status === 'quality_failed'
|
||||
);
|
||||
}
|
||||
|
||||
/** 自动重试调度时把已知 runId 写入下一轮 opts，避免 abort 后丢失。 */
|
||||
export function nextAttemptOpts<T extends { reconnectRunId?: string }>(
|
||||
opts: T | undefined,
|
||||
retainedRunId: string | undefined,
|
||||
): T & { reconnectRunId?: string } {
|
||||
const base = { ...(opts || ({} as T)) };
|
||||
if (retainedRunId) {
|
||||
return { ...base, reconnectRunId: retainedRunId };
|
||||
}
|
||||
return base;
|
||||
}
|
||||
Reference in New Issue
Block a user