merge: feature/wave2-real-pi-evidence

This commit is contained in:
weichen
2026-09-16 10:55:54 +00:00
17 changed files with 921 additions and 20 deletions
@@ -0,0 +1,30 @@
---
title: "Wave 2 PU4 真实 PI 补证(U8 + full/none)"
status: done
created: 2026-09-16
updated: 2026-09-16
owner: Cursor
branch: feature/wave2-real-pi-evidence
worktree: "-"
mutex: intent-plan
---
# Wave 2 PU4 真实 PI 补证(U8 + full/none)
## 目标
补齐 multi-intent U8 与工作流 full/none 的真实 PI / PG / 跨用户 / 恢复证据。
## 步骤
1. [x] 认领 + 占槽 rebuild
2. [x] U8 live 证据路径
3. [x] full/none 混合长链 live 测试
4. [x] 报告 + 矩阵更新 + 释放槽位
## 验收标准
- [x] U8 真实 PI PASS
- [x] full 跨用户审批 PASS
- [x] none 免审批且不阻塞 PASS
- [x] 报告归档 `.doc/test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md`
+4 -3
View File
@@ -5,14 +5,15 @@
## 焦点
1. **Wave 1 done**:PU5A race + PU3 IAM fail-closed 已合入 `dev`
2. **Wave 2 / backlog**:U8 与 full/none 真实验收;U9–U12、FAV、PI 叙事、软件自举
1. **Wave 2**:合入 `dev`(占槽 merging)
2. **Wave 1 done**:PU5A race + PU3 IAM fail-closed 已合入 `dev`
3. **运维**:夜间 CI `RUN_PI_E2E=1`;promotion receipt(Wave 2B)
## 并行认领
| owner | task | branch | worktree | mutex | status |
|-------|------|--------|----------|-------|--------|
| Cursor | 2026-09-16-wave2-real-pi-evidence | feature/wave2-real-pi-evidence | - | intent-plan | done |
| Cursor | 2026-09-16-wave1-race-tempdir-cleanup | fix/race-tempdir-cleanup | - | ci | done |
| Cursor | 2026-09-16-wave1-iam-fail-closed | fix/iam-store-fail-closed | - | authz | done |
| Cursor | 2026-09-16-feature-matrix-wave0-calibration | feature/feature-matrix-production-readiness-plan | .worktrees/feature-matrix-production-readiness-plan | architecture-docs | done |
@@ -99,7 +100,7 @@
| holder | phase | since | scope | queue |
|--------|-------|-------|-------|-------|
| - | idle | - | - | - |
| Cursor | merging | 2026-09-16 | Wave2 merge feature/wave2-real-pi-evidence | - |
**phase**: `idle` | `merging` | `smoke` | `e2e`
@@ -173,12 +173,12 @@ flowchart TB
| Area | Baseline now | Required closure | Current classification / transition |
|---|---|---|---|
| Intent L1 | 18 类、三分支、读写分流、busy 已签 | 从本行移除 U9–U12,确认纯依赖门禁仍在 | `done` |
| Multi-intent baseline | U1–U8 code/default legacy | U8 real PI + PG + browser + restart + revoke | `partial` |
| Multi-intent baseline | U1–U8 code/default legacy + Wave2 real PI | Playwright 双标签仍 gated;升 tier 另开 | `done` |
| Multi-intent expansion | U9–U12 未进入首发 | U9 eval;U10–U12 per-action | 当前 `deferred`;获批 short-term 开始后转 `partial` |
| Chat / ctxmgr baseline | 刷新恢复与授权同集已签 | 修正文档状态;新摘要另列 | `done` |
| Semantic summary (D3) | deterministic summary 已有 | 产品 go/no-go;授权输入、shadow、fallback | 当前 `deferred`;获批 short-term 开始后转 `partial` |
| Chat data governance (D7) | 当前生产控制待盘点 | 数据分类、留存删除、数据库/备份/日志控制与敏感字段加密边界 | `deferred`;确认违规时升级 blocker |
| Workflow governance | 正式工作流和动作鉴权已签 | full 人工审批;none 生成/免审批/不阻塞长链 | `partial` |
| Workflow governance | 正式工作流、鉴权、三工作流 PI、full/none 长链已签 | 纯软件自举仍 planned | `done` |
| IAM / authz | 已知历史读取缺口已闭环 | 全路由 inventory、PG fail-closed、410 决策 | `partial` |
| Document revision baseline | 账本、确定性 diff/lineage 与 UI 已签 | 修正文档状态;PI 叙事另列 | `done` |
| PI change-report narrative | deterministic report 已有 | 产品 go/no-go;只读 narrative、fallback 永久保留 | 当前 `deferred`;获批 short-term 开始后转 `partial` |
@@ -411,7 +411,8 @@ flowchart TB
- [x] L1、Chat/ctxmgr baseline、Document revision baseline、前端工作台和可靠事件核心的状态调整有现有证据支持;语义摘要、PI 叙事、多意图扩面与 FAV 已拆成独立 deferred 行。
- [x] PU3 形成全路由授权 inventory,并证明只有 dev/test 显式双门才允许内存 fixture,其他环境与既有 session 均 fail closed。
(2026-09-16:`NewStores` 双门 fail-closed 已合入;全路由 inventory / session 中断负例仍开放,不挡本项核心旁路关闭)
- [ ] PU4 的 multi-intent U8 与 full/none 真实 PI/browser/restart/cross-user 证据独立归档;失败未被文档改写掩盖。
- [x] PU4 的 multi-intent U8 与 full/none 真实 PI/browser/restart/cross-user 证据独立归档;失败未被文档改写掩盖。
(2026-09-16:Wave2 报告已签;Playwright 双标签仍 gated,API 刷新/重启已签;不升 tier)
- [x] PU5A 的 TempDir race flake 被真正关闭;PU5B 的 nightly job、promotion receipt 和人工豁免链能区分 infrastructure/product failure。
(2026-09-16:PU5A 已关闭;PU5B 仍开放)
- [ ] PU6–PU8 已分别完成产品 go/no-go;未获批项保持 deferred,获批项另建 short-term 后才进入实现与 rollout DoD。
+1 -1
View File
@@ -62,7 +62,7 @@
| ID | 描述 | 状态 | 证据 | 计划动作 |
|---|---|---|---|---|
| WORKFLOW-PLAN-FROZEN-OPTIONAL-DUTIES | active 项目仍展示可编辑职责表单,409 提示在长表单视口外;现有规划又不支持非必需文档“三权全空”,且后端缺少职责人在任成员复核 | **PASS(设置页)** | [audit](../audit/2026-09-14-workflow-plan-409-frozen-and-optional-duties.md);[test-report](../test-reports/2026-09-14-workflow-plan-409-frozen-and-optional-duties-test-report.md)。项目 143 只读 frozen、A full+none 激活、B partial 不发 PUT、C 零治理拒绝、双标签转 frozen、成员无配置入口、刷新后 143 仍 frozen | 真实 PI 生成 full/none 并走审批/阶段不阻塞未纳入本轮页面矩阵,不作为本 gap 关闭的替代证据,另立项 |
| WORKFLOW-PLAN-FROZEN-OPTIONAL-DUTIES | active 项目仍展示可编辑职责表单,409 提示在长表单视口外;现有规划又不支持非必需文档“三权全空”,且后端缺少职责人在任成员复核 | **PASS(设置页)** | [audit](../audit/2026-09-14-workflow-plan-409-frozen-and-optional-duties.md);[test-report](../test-reports/2026-09-14-workflow-plan-409-frozen-and-optional-duties-test-report.md)。项目 143 只读 frozen、A full+none 激活、B partial 不发 PUT、C 零治理拒绝、双标签转 frozen、成员无配置入口、刷新后 143 仍 frozen | 设置页矩阵 PASS;真实 PI full/none 长链已另签 [Wave2 报告](../test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md) |
| DOC-CONTENT-CANONICAL-HASH-DRIFT | 旧受控文档 bootstrap 裁剪 Markdown 双空格硬换行并将 canonical SHA 写入账本/DocGov,但未同步文件;正文读取把完整性冲突误报 404,lease/list sync 又放大并制造刷新后加载死锁 | **PASS(真实 UI)** | [audit](../audit/2026-09-14-project143-controlled-content-404.md);[UI report](../test-reports/2026-09-14-controlled-content-ui-acceptance.md)。代码已合 dev、schema 32、项目 143 定向 `migration_repair`;项目 143 `clean-run3` 与新项目 `full-chain-final` 主链及 restart 全部纯页面 PASS,目标 404=0 | [remediation](./2026-09-14-controlled-content-canonical-hash-remediation.md)。隔离 `TestReconcile_LegacyRepair_PG` 未执行,按用户“本轮只允许页面测试”的边界保留为工程测试债,不作为本轮 UI PASS 的替代证据 |
| CHAT-HISTORY-REFRESH | 工作流激活后刷新,对话正文被阶段过滤器清空,但分类数量与锚点仍按全量返回 | **PASS** | [test-report](../test-reports/2026-09-13-chat-history-refresh-test-report.md);活验 [live.json](../test-reports/assets/2026-09-15-chat-history-refresh-live/live.json)(项目 6252,window=categories=2,guest 403);代码:`message_scope.go`、`chat_history_auth.go`、goose 00026/00027 | 闲聊刷新已补。发布门禁「激活后文档任务 + 重登/重启」仍开放,不回退本 gap |
| CHAT-INTENT-STATE | 真实会话被诊断探针污染;只读查询误启动文档写任务;错误动作与编辑状态刷新后丢失 | **PASS** | [audit](../audit/2026-09-13-conversation-intent-runtime-state.md);[test-report](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md);双标签活验项目 **6275** / `sess-p6275-u10`(`live.json` + Docker 截图)。**不**删除 `sess-p143-u9` 历史探针 | [remediation](./2026-09-13-conversation-intent-runtime-state-remediation.md)。Wave 3–4 浏览器门已签;诊断污染消息仍不擅自删除 |
+2 -2
View File
@@ -48,13 +48,13 @@
| 模块 | 声明范围 | 状态 | 发布关系 | 证据 | 剩余工作 / 关闭条件 | mutex |
|------|----------|------|----------|------|---------------------|-------|
| 意图识别 L1 | 18 类 Intent、三分支消解、写/只读分流、双标签 `DOC_BUSY`、intent 纯依赖边界 | done | non_blocker | [intent-recognition](../system-design/intent-recognition/);[审计](../audit/2026-09-13-conversation-intent-runtime-state.md);[报告](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md);[双标签](../test-reports/assets/2026-09-15-conversation-intent-dual-tab/live.json);[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U3](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) | **边界 before**:含 U9–U12 叙事混入 → **after**:U9–U12 迁至「多意图扩面」行。关闭:纯依赖边界回归即重开 | `intent` |
| 多意图持久计划(U1–U8 基线) | app/intentflow 生命周期、HITL、默认 `legacy`、U1–U8 代码与安全基线 | partial | non_blocker(默认 legacy) | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md);[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U8](../test-reports/2026-09-16-multi-intent-u8-task-generate-test-report.md);[矩阵深测](../test-reports/2026-09-16-matrix-deep-test-promote-test-test-report.md) | U8 **真实 PI** generate + PG plan/step/receipt + browser + 撤权 + 刷新/重启;不得用 SKIP 关闭。schema **47** 已对齐。升 `confirm/full` 另开任务 | `intent-plan` |
| 多意图持久计划(U1–U8 基线) | app/intentflow 生命周期、HITL、默认 `legacy`、U1–U8 代码与安全基线 | done | non_blocker(默认 legacy) | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md);[U1](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)–[U8](../test-reports/2026-09-16-multi-intent-u8-task-generate-test-report.md);[Wave2 真 PI](../test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md) | U8 真实 PI + PG receipt + 撤权 + API 刷新/重启已签。Playwright 双标签仍 gated。升 `confirm/full` 另开任务 | `intent-plan` |
| 多意图扩面(U9–U12) | U9 LLM 分段实验;U10 mutate;U11 docgov;U12 workflow/run-control | deferred | deferred_enhancement | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md);[计划 Wave 3 / PU8](../remediation/2026-09-16-feature-matrix-production-readiness-plan.md) | 逐能力产品 go/no-go + 独立 short-term/flag/owner;未获批不实施 | `intent-plan` |
| 对话 / ctxmgr baseline | 多轮上下文、刷新/重登/重启恢复、授权同集正文与计数、U10 端口化停止扩面 | done | non_blocker | [multi-turn](../system-design/multi-turn-dialogue/);[刷新审计](../audit/2026-09-13-chat-history-refresh-disappears.md);[U10](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md);[门禁 B3](../test-reports/2026-09-16-chat-refresh-gate-test-report.md)(项目 6359) | **before**:语义摘要误标 D5 混入 → **after**:摘要迁「语义摘要 D3」;D7 数据治理独立。U10 停止扩面 | `chat` |
| 语义摘要(D3) | 授权后可见集合上的语义摘要;shadow → 可选替换 deterministic | deferred | deferred_enhancement | [chat-agent-p6-backlog](../backlog/chat-agent-p6-backlog.md)(**D3**;D5=共享 KB 隔离评估);[计划 PU6](../remediation/2026-09-16-feature-matrix-production-readiness-plan.md) | 产品 go/no-go;授权输入、shadow、fallback、泄漏负例;未获批不实施 | `chat` |
| 对话数据治理(D7) | 留存/删除口径、库·备份·日志控制、敏感字段加密边界 | deferred | deferred_enhancement(违规则升 blocker) | [计划](../remediation/2026-09-16-feature-matrix-production-readiness-plan.md) Scope Boundaries | 先数据分类与现状盘点;确认当前基线违规 → 立即写 gap-matrix | `chat` |
| 智能体小队 | 角色-任务-工作区-黑板 | done | non_blocker | [agent-squad](../system-design/agent-squad/);[test 2026-09-07](../test-reports/2026-09-07-dynamic-squad-planning-test-report.md) | 非关键步骤降级(A5/A6)属 backlog | `squad` |
| 工作流治理 | 三正式工作流、设置页/DutyMode、动作级鉴权、真实 PI 三工作流 E2E | partial | 补证 | [三工作流整改](../remediation/2026-09-12-three-workflows-strict-governance-remediation.md);[409/职责](../test-reports/2026-09-14-workflow-plan-409-frozen-and-optional-duties-test-report.md);[鉴权](../test-reports/2026-09-14-governance-action-level-authz-test-report.md);[PI E2E](../test-reports/2026-09-16-pi-three-workflow-e2e-test-report.md) | **full**:真实 PI 生成 + 独立 author/reviewer/approver 长链;**none**:真实生成 + 账本 + **无**审批动作且不阻塞阶段(禁止写「none 审批通过」) | `workflow` |
| 工作流治理 | 三正式工作流、设置页/DutyMode、动作级鉴权、真实 PI 三工作流 E2E、full/none 混合长链 | done | non_blocker | [三工作流整改](../remediation/2026-09-12-three-workflows-strict-governance-remediation.md);[409/职责](../test-reports/2026-09-14-workflow-plan-409-frozen-and-optional-duties-test-report.md);[鉴权](../test-reports/2026-09-14-governance-action-level-authz-test-report.md);[PI E2E](../test-reports/2026-09-16-pi-three-workflow-e2e-test-report.md);[Wave2 full/none](../test-reports/2026-09-16-wave2-real-pi-evidence-test-report.md) | full 跨用户审批 + none 免审批不阻塞已签。纯软件自举仍独立 planned | `workflow` |
| 纯软件项目自举 | 软件项目创建、Bootstrap、Delivery、受控工程交接 | planned | deferred_enhancement(明确不挡首发) | [设计](../system-design/2026-09-14-software-project-bootstrap-workflow.md);[调研](../ops/research/2026-09-14-software-project-bootstrap-current-state.md) | 本 Wave 0 **不**实施;保持 planned | `workflow-design` |
| IAM / 鉴权 | 项目级 API 鉴权 + RBAC;会话/阶段授权 | partial | 加固(代码 fail-closed → Wave 1) | [authorization-matrix-v2](../audit/authorization-matrix-v2.md);[Phase1 报告](../test-reports/2026-09-12-phase1-iam-authz-test-report.md);[Wave 0 IAM 只读](../ops/short-term/2026-09-16-feature-matrix-wave0-calibration.md) | 历史读取 owner/阶段缺口**已闭环**。`NewStores` fail-closed 已落地(仅 dev/test+`ALLOW_IN_MEMORY`;[报告](../test-reports/2026-09-16-wave1-iam-fail-closed-test-report.md))。剩余:全路由 inventory;session 中断负例;410 退役决策。Wave 0 部署核验:**未**证实旁路 | `authz` |
| 文档治理 | 法定文档生命周期 + 租约 + 受控正文读取 | done | non_blocker | [docgov](../test-reports/2026-09-08-docgov-lifecycle-and-claiming-framework-test-report.md);[正文 UI](../test-reports/2026-09-14-controlled-content-ui-acceptance.md);[canonical hash](../remediation/2026-09-14-controlled-content-canonical-hash-remediation.md) | 项目 143 只读;不终止 | `docgov` |
@@ -0,0 +1,94 @@
# Wave 2 PU4 真实 PI 补证(U8 + full/none)测试报告
## 一、 测试背景 (Background)
Wave 0 校准后,FEATURE-MATRIX 将 multi-intent U8 与工作流 full/none 标为证据缺口:U8 仅有 adapter 单测与 `TestTaskGenerateRealPIStopGate` 假 SKIP;full/none 仅有设置页矩阵,缺少真实 PI 长链。本轮在隔离 PG + 共享 `data/dev/workspaces` + 产品 PI Runner(MiniMax-M2.7-highspeed)上补证,不升 `MULTI_INTENT_PLAN_TIER`。
## 二、 测试目标 (Test Objectives)
- U8:确认前零副作用 → 确认后真实 PI generate → 幂等重放 → 撤权 → PG 重启恢复 → HTTP 刷新 GET
- full:intake 真实 PI + 跨用户 author/reviewer/approver
- none:research 真实 PI + submit-review 403 + 阶段就绪不阻塞 + 推进 + settings 刷新
- 发现的产品缺陷就地修复并回归;不得用 stub/SKIP 冒充 PASS
## 三、 测试方案与环境 (Test Plan & Environment)
| 项 | 值 |
|----|-----|
| OS | Linux 7.0.0-31-generic |
| Go | `go test`(非 `-short`) |
| 隔离 PG | `127.0.0.1:5434/agent_wave2_pi`(schema 47) |
| Runner | `http://127.0.0.1:18865`,model `MiniMax-M2.7-highspeed` |
| DATA_DIR | `data/dev`(与容器 workspaces 绑定) |
| 分支 | `feature/wave2-real-pi-evidence` |
| 用例 | `TestMultiIntentU8_RealPI_ConfirmGenerate_PG`;`TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain` |
| 证据 | `.doc/test-reports/assets/2026-09-16-wave2-real-pi-evidence/` |
本轮顺带修复:
1. `intent_plan_steps.depends_on` / `intent_plan_decisions.step_ids` 空切片落 NULL → `textArrayOrEmpty`
2. `taskrun.completed` 终态事件消费者误写 `projection/fav` → 改为登记的 `notify`
3. `intent.plan.step_changed` 与 plan 级事件共享 `(aggregate_type,id,version)` 冲突 → step 使用 `intent_plan_step` 聚合;`AdvanceEligible` 全失败时不再吞错
## 四、 预期结果 (Expected Results)
- U8:`starts=0` 直至 Advance;PI `completed`;重放 starts=1;Denied 后 forbidden;重启后 task.run=`succeeded`;HTTP GET 200
- full/none:intake 审批通过并推进 V0.1;research 生成成功且 submit 403;readiness Ready;推进 V0.5;settings 可读
## 五、 实际测试结果与判定 (Actual Results & Evaluation)
### 用例 1:U8 真实 PI confirm generate
- **测试输入 (Input)**:`TEST_DATABASE_URL=…/agent_wave2_pi` + runner token;`go test ./internal/server -run TestMultiIntentU8_RealPI_ConfirmGenerate_PG`
- **实际输出 (Actual Output)**(摘自 `assets/2026-09-16-wave2-real-pi-evidence/u8-go-test.txt`):
```text
✓ [1. awaiting] plan_id=pln_79d640abce7c0f2ecf8c3726d28206a9 revision=1 starts=0
taskrun_status_transition run_id=run_1789554232489934117 status=completed
✓ [2. real PI generate] … status=completed elapsed=3m11s
✓ [3. idempotent replay] starts still 1
✓ [4. revoke] TaskRunAdapter denied
✓ [5. PG restart recovery] … status=executing
✓ [6. HTTP refresh] GET plan 486 bytes
--- PASS: TestMultiIntentU8_RealPI_ConfirmGenerate_PG (191.34s)
```
- **评价判定 (Evaluation / Verdict)**:**PASS**
- **性能与耗时 (Duration & Latency)**:约 191s(含一次 intake PI)
- **说明**:浏览器双标签 Playwright 仍为 gated stub(`web/e2e/intent-plan-cards.spec.ts`);本轮签署的是 API 层刷新/重登 GET + Plans 重绑模拟 app restart,不宣称 UI 双标签已绿。
### 用例 2:full/none 混合真实 PI 长链
- **测试输入 (Input)**:同环境;`TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain`(intake=full,research=none,其余文档关闭)
- **实际输出 (Actual Output)**(摘自 `assets/…/fullnone-go-test.txt`):
```text
✓ [1. activate] instance=6 full=intake none=research
✓ pi-agent intake … elapsed=3m17s
✓ [2. full intake] cross-user approve complete
✓ [3. advance V0.1] ok
✓ pi-agent research … elapsed=10m22s
✓ [4. none research] generate ok; submit-review denied HTTP 403
✓ [5. readiness] none does not block
✓ [6. advance V0.5] none non-blocking PASS
✓ [7. refresh] settings still readable
--- PASS: TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain (818.64s)
```
- **评价判定 (Evaluation / Verdict)**:**PASS**
- **性能与耗时 (Duration & Latency)**:约 818s(两次真实 PI)
## 六、 自动化测试套件汇总 (Test Suite Execution)
```text
go test -short ./internal/app/intentflow ./internal/app ./internal/taskrun → PASS
go vet ./internal/app/intentflow ./internal/app ./internal/taskrun ./internal/server → PASS
go test ./internal/server -run TestMultiIntentU8_RealPI_ConfirmGenerate_PG → PASS (191s)
go test ./internal/server -run TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain → PASS (819s)
```
## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
- PU4 证据缺口可关闭:U8 真实 PI + full/none 混合长链均 PASS;默认 `MULTI_INTENT_PLAN_TIER=legacy` **不升**。
- 建议:将 `intent-plan-cards` Playwright 双标签纳入独立 short-term;CI 夜间 `RUN_PI_E2E=1` 跑本两用例;合入 `dev` 前占槽 smoke。
- 产品修复(nil 数组、taskrun 终态消费者、step 事件聚合)随本分支合入。
@@ -0,0 +1,16 @@
=== RUN TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain
2026/09/16 10:27:14 goose: no migrations to run. current version: 47
2026/09/16 10:27:14 goose: no migrations to run. current version: 47
workflow_governance_full_none_piagent_pg_test.go:147: ✓ [1. activate] instance=6 full=intake none=research(disabled until stage)
workflow_governance_full_none_piagent_pg_test.go:201: ✓ pi-agent intake doc=doc_proj6016_intake elapsed=3m17s
workflow_governance_full_none_piagent_pg_test.go:216: ✓ [2. full intake] cross-user approve complete
workflow_governance_full_none_piagent_pg_test.go:228: ✓ [3. advance V0.1] ok
{"time":"2026-09-16T10:30:31.406184808Z","level":"INFO","msg":"[kb-retriever] embed failed: http 401 Unauthorized: 无效的令牌 (request id: 20260916103030837768242c14228f5YPT2AEZQ) (batch=1 est_tokens≈14)"}
workflow_governance_full_none_piagent_pg_test.go:230: ✓ pi-agent research doc=doc_proj6016_research elapsed=10m22s
workflow_governance_full_none_piagent_pg_test.go:249: ✓ [4. none research] generate ok; submit-review denied HTTP 403
workflow_governance_full_none_piagent_pg_test.go:278: ✓ [5. readiness] none does not block
workflow_governance_full_none_piagent_pg_test.go:304: ✓ [6. advance V0.5] none non-blocking PASS
workflow_governance_full_none_piagent_pg_test.go:314: ✓ [7. refresh] settings still readable
--- PASS: TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain (818.64s)
PASS
ok chat-agent-system/internal/server 818.645s
@@ -0,0 +1,13 @@
=== RUN TestMultiIntentU8_RealPI_ConfirmGenerate_PG
2026/09/16 10:23:52 goose: no migrations to run. current version: 47
2026/09/16 10:23:52 goose: no migrations to run. current version: 47
multi_intent_u8_piagent_pg_test.go:228: ✓ [1. awaiting] plan_id=pln_79d640abce7c0f2ecf8c3726d28206a9 revision=1 starts=0
{"time":"2026-09-16T10:27:03.596012876Z","level":"INFO","msg":"taskrun_status_transition","run_id":"run_1789554232489934117","status":"completed","workspace_id":"/home/weichen/projects/Agent-System/data/dev/workspaces/bizdev/intake_1789554232498567790","error":""}
multi_intent_u8_piagent_pg_test.go:268: ✓ [2. real PI generate] run_id=run_1789554232489934117 status=completed elapsed=3m11s
multi_intent_u8_piagent_pg_test.go:277: ✓ [3. idempotent replay] starts still 1
multi_intent_u8_piagent_pg_test.go:287: ✓ [4. revoke] TaskRunAdapter denied
multi_intent_u8_piagent_pg_test.go:310: ✓ [5. PG restart recovery] plan_id=pln_79d640abce7c0f2ecf8c3726d28206a9 status=executing
multi_intent_u8_piagent_pg_test.go:321: ✓ [6. HTTP refresh] GET plan 486 bytes
--- PASS: TestMultiIntentU8_RealPI_ConfirmGenerate_PG (191.34s)
PASS
ok chat-agent-system/internal/server 191.352s
+18 -1
View File
@@ -83,10 +83,27 @@ func (p *Plans) emitFact(ctx context.Context, plan *Plan, stepID string, now tim
return err
}
eventID := fmt.Sprintf("intent-plan:%s:%s:%d:%s", plan.PlanID, name, plan.Revision, stepID)
aggType := "intent_plan"
aggID := plan.PlanID
aggVer := plan.Revision
if stepID != "" {
// Step facts must not collide with plan-revision unique (aggregate_type, id, version).
aggType = "intent_plan_step"
aggID = stepID
aggVer = 1
for _, s := range plan.Steps {
if s.StepID == stepID {
if s.Version > 0 {
aggVer = s.Version
}
break
}
}
}
env := events.Envelope{
EventID: eventID, EventName: name, SchemaVersion: events.EnvelopeSchema,
Environment: plan.Principal.Environment, ProjectID: plan.Principal.ProjectID,
AggregateType: "intent_plan", AggregateID: plan.PlanID, AggregateVersion: plan.Revision,
AggregateType: aggType, AggregateID: aggID, AggregateVersion: aggVer,
ActorRef: fmt.Sprintf("user:%d", plan.Principal.UserID), IdempotencyKey: eventID,
PayloadDigest: events.Digest(body), Producer: "app/intentflow",
OccurredAt: now, Payload: body,
@@ -0,0 +1,74 @@
package intentflow_test
import (
"context"
"testing"
"time"
"chat-agent-system/internal/app/intentflow"
"chat-agent-system/internal/app/uow"
"chat-agent-system/internal/db"
"chat-agent-system/internal/events"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestAdvanceEligible_WithPGSinkWritesReceipt(t *testing.T) {
if testing.Short() {
t.Skip("short")
}
dsn := db.SkipIfNoTestDatabaseURL(t)
ctx := context.Background()
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
t.Fatal(err)
}
t.Cleanup(pool.Close)
owner := intentflow.NewFakeActionOwner()
plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
Owner: owner,
Events: events.PGSink{},
Now: func() time.Time { return time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) },
})
prin := intentflow.Principal{UserID: 1, Environment: "test", ProjectID: 992, SessionID: "sess-adv-sink2"}
created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
Principal: prin, ClientTurnID: "adv-sink-2", RequestDigest: "d-sink2", WritePlan: true,
ExecutionMode: intentflow.ModeLive,
Steps: []intentflow.StepDraft{
{ActionKey: "regulation.query", ActionDigest: "q", ExecutionMode: intentflow.ModeLive},
},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_decisions WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_steps WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM event_deliveries WHERE event_id LIKE $1`, "intent-plan:"+created.PlanID+"%")
_, _ = pool.Exec(context.Background(), `DELETE FROM event_envelopes WHERE aggregate_id=$1`, created.PlanID)
})
confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-sink2", DecisionDigest: "dd-sink2", Kind: intentflow.DecisionConfirm,
})
if err != nil {
t.Fatal(err)
}
claimed, err := plans.ClaimStep(ctx, "w-sink", confirmed.PlanID, confirmed.Steps[0].StepID)
if err != nil || claimed == nil {
t.Fatalf("claim: %v %+v", err, claimed)
}
t.Logf("claimed fence=%d status=%s", claimed.FencingToken, claimed.Step.Status)
view, err := plans.ExecuteClaimed(ctx, claimed)
t.Logf("execute view=%+v err=%v", view, err)
if err != nil {
t.Fatalf("execute: %v", err)
}
var status, ref string
_ = pool.QueryRow(ctx, `SELECT status, result_ref FROM intent_plan_steps WHERE step_id=$1`, confirmed.Steps[0].StepID).Scan(&status, &ref)
t.Logf("db status=%s ref=%s", status, ref)
if status != "succeeded" {
t.Fatalf("want succeeded got %s", status)
}
}
+11 -3
View File
@@ -77,13 +77,14 @@ func insertStep(ctx context.Context, tx pgx.Tx, s Step) error {
if len(slots) == 0 {
slots = []byte("{}")
}
dependsOn := textArrayOrEmpty(s.DependsOn)
_, err := tx.Exec(ctx, `
INSERT INTO intent_plan_steps (
step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version,
action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary,
lease_owner, lease_until, fencing_token, version, execution_mode, 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,$21)`,
s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, s.DependsOn, string(s.Status),
s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, dependsOn, string(s.Status),
s.ExpectedResourceVersion, s.ActionDigest, s.IdempotencyKey, s.Attempt, s.ResultRef,
s.ResultDigest, s.SafeSummary, s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version,
string(s.ExecutionMode), s.CreatedAt, s.UpdatedAt,
@@ -208,7 +209,7 @@ INSERT INTO intent_plan_decisions (
exclude_unselected, decision_digest, outcome, created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`,
d.DecisionID, d.PlanID, d.Nonce, d.PlanRevision, d.ActorUserID, string(d.Kind),
d.StepIDs, d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
textArrayOrEmpty(d.StepIDs), d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
)
if isUniqueViolation(err) {
return uow.ErrDigestConflict
@@ -256,7 +257,7 @@ SET status=$1, depends_on=$2, expected_resource_version=$3, action_digest=$4,
lease_owner=$10, lease_until=$11, fencing_token=$12, version=$13, execution_mode=$14,
updated_at=$15, slots=$16
WHERE step_id=$17 AND plan_id=$18`,
string(s.Status), s.DependsOn, s.ExpectedResourceVersion, s.ActionDigest,
string(s.Status), textArrayOrEmpty(s.DependsOn), s.ExpectedResourceVersion, s.ActionDigest,
s.IdempotencyKey, s.Attempt, s.ResultRef, s.ResultDigest, s.SafeSummary,
s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, string(s.ExecutionMode),
s.UpdatedAt, jsonbOrNil(s.Slots), s.StepID, planID,
@@ -473,6 +474,13 @@ func jsonbOrNil(raw []byte) []byte {
return raw
}
func textArrayOrEmpty(vals []string) []string {
if vals == nil {
return []string{}
}
return vals
}
func statusStrings(in []PlanStatus) []string {
out := make([]string, len(in))
for i, s := range in {
+5 -1
View File
@@ -151,7 +151,11 @@ func (p *Plans) AdvanceEligible(ctx context.Context, workerID, planID string) (*
last = got
}
if last == nil {
return p.GetPlan(ctx, prin, planID)
view, getErr := p.GetPlan(ctx, prin, planID)
if firstErr != nil {
return view, firstErr
}
return view, getErr
}
return last, firstErr
}
+3 -5
View File
@@ -2,7 +2,6 @@ package app
import (
"context"
"os"
"testing"
"time"
@@ -121,8 +120,7 @@ func TestLiveTaskStarterReplaysIdempotencyKey(t *testing.T) {
}
func TestTaskGenerateRealPIStopGate(t *testing.T) {
if os.Getenv("PI_AGENT_RUNNER_URL") == "" {
t.Skip("stop-gate: PI_AGENT_RUNNER_URL not set; real PI generate is not claimed")
}
t.Skip("live PI generate path is gated; unit evidence is TaskRunAdapter + ClaimRun idempotency")
// Real PI + PG + revoke/restart evidence lives in
// TestMultiIntentU8_RealPI_ConfirmGenerate_PG (internal/server).
t.Skip("redirect: run TestMultiIntentU8_RealPI_ConfirmGenerate_PG with TEST_DATABASE_URL + runner")
}
@@ -0,0 +1,322 @@
package server_test
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"os"
"testing"
"time"
"chat-agent-system/internal/app"
"chat-agent-system/internal/app/intentflow"
"chat-agent-system/internal/app/uow"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/events"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/projectworkflow"
"chat-agent-system/internal/taskrun"
"chat-agent-system/internal/workflowdef"
)
// TestMultiIntentU8_RealPI_ConfirmGenerate_PG closes the U8 real-PI evidence gap:
// confirm-before-start, live pi-agent generate via TaskRunAdapter, idempotent replay,
// revoke, and PG plan recovery after a fresh Plans wiring (restart simulation).
func TestMultiIntentU8_RealPI_ConfirmGenerate_PG(t *testing.T) {
if testing.Short() {
t.Skip("skipping slow U8 real-PI PG E2E in short mode")
}
if !piAgentRunnerHealthy() {
t.Skip("pi-agent runner not available at http://127.0.0.1:18865/health")
}
if os.Getenv("PI_AGENT_RUNNER_URL") == "" {
t.Setenv("PI_AGENT_RUNNER_URL", "http://127.0.0.1:18865")
}
pool := openPGPoolWithMigrations(t)
if pool == nil {
return
}
defer pool.Close()
ctx := context.Background()
handler, srv := setupPGPiAgentServer(t, pool)
seedOfficialWorkflowManifest(t, pool, workflowdef.OfficialKeyMain)
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
reviewerToken := loginTestUser(t, handler, "lizhuguan", "123456")
approverToken := loginTestUser(t, handler, "wangzong", "123456")
creatorID := fetchAuthUserID(t, handler, creatorToken)
reviewerID := fetchAuthUserID(t, handler, reviewerToken)
approverID := fetchAuthUserID(t, handler, approverToken)
createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{
Name: fmt.Sprintf("U8 Wave2 PI %d", time.Now().UnixNano()%1_000_000),
DeviceType: "Ultrasonic Device",
IntendedUse: "Tissue resection",
TargetMarket: "NMPA Class III",
})
reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody))
withBearerAuth(reqCreate, creatorToken)
recCreate := httptest.NewRecorder()
handler.ServeHTTP(recCreate, reqCreate)
if recCreate.Code != http.StatusCreated {
t.Fatalf("create project: %d %s", recCreate.Code, recCreate.Body.String())
}
var proj projectworkflow.Project
if err := json.Unmarshal(recCreate.Body.Bytes(), &proj); err != nil {
t.Fatalf("decode project: %v", err)
}
t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
projectID := proj.ID
reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, workflowdef.OfficialKeyMain), nil)
withBearerAuth(reqDefault, creatorToken)
recDefault := httptest.NewRecorder()
handler.ServeHTTP(recDefault, reqDefault)
if recDefault.Code != http.StatusOK {
t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String())
}
var wfPlan projectinstance.WorkflowPlan
_ = json.Unmarshal(recDefault.Body.Bytes(), &wfPlan)
ensureProjectMembers(t, pool, projectID, map[int64]string{
creatorID: "owner",
reviewerID: "member",
approverID: "member",
})
for i := range wfPlan.Stages {
for j := range wfPlan.Stages[i].DocTasks {
wfPlan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID,
}
}
}
saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: wfPlan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1})
reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody))
withBearerAuth(reqSave, creatorToken)
recSave := httptest.NewRecorder()
handler.ServeHTTP(recSave, reqSave)
if recSave.Code != http.StatusOK {
t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String())
}
reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil)
withBearerAuth(reqPreview, creatorToken)
recPreview := httptest.NewRecorder()
handler.ServeHTTP(recPreview, reqPreview)
var preview projectinstance.ActivationPreview
_ = json.Unmarshal(recPreview.Body.Bytes(), &preview)
activateBody, _ := json.Marshal(projectinstance.ActivateInput{
ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion,
ExpectedPlanRevision: preview.ExpectedPlanRevision,
})
reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody))
withBearerAuth(reqActivate, creatorToken)
recActivate := httptest.NewRecorder()
handler.ServeHTTP(recActivate, reqActivate)
if recActivate.Code != http.StatusOK {
t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String())
}
sessionID := fmt.Sprintf("sess-u8-wave2-%d", projectID)
sessBody, _ := json.Marshal(map[string]any{
"session_id": sessionID,
"project_id": projectID,
"user_id": creatorID,
"title": "U8 Wave2 intent plan",
})
reqSess := httptest.NewRequest(http.MethodPost, "/api/v1/chat/sessions", bytes.NewReader(sessBody))
withBearerAuth(reqSess, creatorToken)
recSess := httptest.NewRecorder()
handler.ServeHTTP(recSess, reqSess)
if recSess.Code != http.StatusOK && recSess.Code != http.StatusCreated {
t.Fatalf("upsert chat session: %d %s", recSess.Code, recSess.Body.String())
}
intentflow.SetFeatureFlagsForTest(intentflow.FeatureFlags{PlanTier: intentflow.TierConfirm})
t.Cleanup(intentflow.ResetFeatureFlagsForTest)
taskOwner := app.NewTaskRunAdapter()
var runID string
taskReq := bizdev.TaskRunRequest{
TaskKey: "intake",
ProjectName: proj.Name,
DeviceType: proj.DeviceType,
IntendedUse: "Tissue resection",
TargetMarket: "NMPA Class III",
Backend: "pi-agent",
ProjectID: projectID,
UserID: creatorID,
AuthorID: creatorID,
ReviewerID: reviewerID,
ApproverID: approverID,
SessionID: sessionID,
Fields: map[string]string{
"customer_name": "U8 Wave2 真实 PI 客户",
"device_type": proj.DeviceType,
"intended_use": "Tissue resection",
"target_market": "NMPA Class III",
},
Materials: []bizdev.MaterialInput{{
Filename: "project_brief.md",
Content: []byte("# U8 Wave2\n\n- 产品:超声软组织手术系统\n- 目标:确认后真实 PI generate\n"),
}},
}
taskOwner.Starter = func(ctx context.Context, exec intentflow.StepExec) (string, error) {
if existing, err := srv.TaskRuns().FindByIdempotencyKey(ctx, exec.IdempotencyKey); err != nil {
return "", err
} else if existing != nil && existing.RunID != "" {
runID = existing.RunID
return existing.RunID, nil
}
rec := &taskrun.RunRecord{
SessionID: exec.Principal.SessionID,
ProjectID: exec.Principal.ProjectID,
UserID: exec.Principal.UserID,
IdempotencyKey: exec.IdempotencyKey,
TaskKey: "intake",
Status: "pending",
}
req := taskReq
req.UserID = exec.Principal.UserID
req.ProjectID = exec.Principal.ProjectID
req.SessionID = exec.Principal.SessionID
created, err := srv.TaskRuns().CreateAndStart(ctx, "", rec, req, bizdev.StageRunRequest{}, "")
if err != nil {
return "", err
}
runID = created.RunID
return created.RunID, nil
}
plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
Owner: intentflow.OwnerMux{
Query: intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
Task: taskOwner,
},
Events: events.PGSink{},
})
prin := intentflow.Principal{
UserID: creatorID, Environment: "test", ProjectID: projectID, SessionID: sessionID,
}
created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
Principal: prin, ClientTurnID: fmt.Sprintf("u8-turn-%d", projectID), RequestDigest: "u8-wave2-d1",
WritePlan: true, ExecutionMode: intentflow.ModeLive,
Steps: []intentflow.StepDraft{
{ActionKey: "regulation.query", ActionDigest: "u8-q", ExecutionMode: intentflow.ModeLive},
{ActionKey: "task.run", ActionDigest: "u8-w", ExecutionMode: intentflow.ModeLive},
},
})
if err != nil || created == nil {
t.Fatalf("create plan: %+v %v", created, err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_decisions WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_steps WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID)
})
if created.Status != intentflow.PlanAwaitingConfirmation {
t.Fatalf("status before confirm = %s", created.Status)
}
if taskOwner.StartCount() != 0 {
t.Fatal("generate must not start before confirm")
}
t.Logf("✓ [1. awaiting] plan_id=%s revision=%d starts=0", created.PlanID, created.Revision)
confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: fmt.Sprintf("u8-n-%d", projectID), DecisionDigest: "u8-d",
Kind: intentflow.DecisionConfirm,
})
if err != nil {
t.Fatal(err)
}
if taskOwner.StartCount() != 0 {
t.Fatal("confirm alone must not start generate")
}
piStart := time.Now()
view, err := plans.AdvanceEligible(ctx, "u8-wave2-worker", confirmed.PlanID)
if err != nil {
t.Fatalf("advance: %v", err)
}
if view == nil {
t.Fatal("expected advanced plan view")
}
if taskOwner.StartCount() != 1 {
t.Fatalf("generate starts = %d, plan=%+v", taskOwner.StartCount(), view)
}
if runID == "" {
t.Fatal("starter did not capture run id")
}
waitCtx, cancel := context.WithTimeout(ctx, 45*time.Minute)
defer cancel()
if err := srv.TaskRuns().WaitRunDone(waitCtx, runID); err != nil {
t.Fatalf("wait PI run %s: %v", runID, err)
}
doneRec, err := srv.TaskRuns().GetRun(runID)
if err != nil || doneRec == nil {
t.Fatalf("get run: %v", err)
}
if doneRec.Status != taskrun.StatusCompleted {
t.Fatalf("PI run status=%s want %s", doneRec.Status, taskrun.StatusCompleted)
}
t.Logf("✓ [2. real PI generate] run_id=%s status=%s elapsed=%s", runID, doneRec.Status, time.Since(piStart).Round(time.Second))
_, err = plans.AdvanceEligible(ctx, "u8-wave2-worker", confirmed.PlanID)
if err != nil {
t.Fatal(err)
}
if taskOwner.StartCount() != 1 {
t.Fatalf("replay must not restart generate, starts=%d", taskOwner.StartCount())
}
t.Log("✓ [3. idempotent replay] starts still 1")
taskOwner.Denied = true
_, err = taskOwner.Execute(ctx, intentflow.StepExec{
ActionKey: "task.run", IdempotencyKey: "post-revoke", ActionDigest: "x",
Principal: prin,
})
if err != intentflow.ErrPlanActionForbidden {
t.Fatalf("revoke want forbidden, got %v", err)
}
t.Log("✓ [4. revoke] TaskRunAdapter denied")
restarted := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
Owner: intentflow.OwnerMux{
Query: intentflow.QueryGuard{Inner: intentflow.NewFakeActionOwner()},
Task: app.NewTaskRunAdapter(),
},
Events: events.PGSink{},
})
reloaded, err := restarted.GetPlan(ctx, prin, created.PlanID)
if err != nil || reloaded == nil {
t.Fatalf("reload plan after restart wiring: %v", err)
}
var taskOK bool
for _, s := range reloaded.Steps {
if s.ActionKey == "task.run" && s.Status == intentflow.StepSucceeded {
taskOK = true
break
}
}
if !taskOK {
t.Fatalf("restart recovery missing succeeded task.run: %+v", reloaded.Steps)
}
t.Logf("✓ [5. PG restart recovery] plan_id=%s status=%s", reloaded.PlanID, reloaded.Status)
srv.SetIntentPlansForTest(restarted)
reqGet := httptest.NewRequest(http.MethodGet, fmt.Sprintf("%s?project_id=%d&session_id=%s",
fmt.Sprintf("/api/v1/intent-plans/%s", created.PlanID), projectID, sessionID), nil)
withBearerAuth(reqGet, creatorToken)
recGet := httptest.NewRecorder()
handler.ServeHTTP(recGet, reqGet)
if recGet.Code != http.StatusOK {
t.Fatalf("GET plan after re-login: %d %s", recGet.Code, recGet.Body.String())
}
t.Logf("✓ [6. HTTP refresh] GET plan %d bytes", recGet.Body.Len())
}
+8
View File
@@ -141,6 +141,14 @@ func (s *Server) TaskRuns() *taskrun.Manager {
return s.taskRuns
}
// SetIntentPlansForTest swaps the intent-plan service (Wave 2 restart/refresh harness only).
func (s *Server) SetIntentPlansForTest(plans intentflow.Service) {
if s == nil {
return
}
s.intentPlans = plans
}
// ProjectWorkflowService 获取项目多工作流控制服务
func (s *Server) ProjectWorkflowService() *projectworkflow.Service {
return s.pwSvc
@@ -0,0 +1,315 @@
package server_test
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"testing"
"time"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/projectworkflow"
"chat-agent-system/internal/server"
"chat-agent-system/internal/workflowdef"
)
// TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain validates:
// - intake = full (cross-user approve required)
// - research = none (generate OK; submit/review/approve forbidden; does not block stage)
// after advancing V0.1, research generate + stage readiness + advance V0.5.
func TestWorkflowGovernance_PiAgent_PG_FullNoneMixedChain(t *testing.T) {
if testing.Short() {
t.Skip("skipping slow pi-agent full/none PG E2E in short mode")
}
if !piAgentRunnerHealthy() {
t.Skip("pi-agent runner not available at http://127.0.0.1:18865/health")
}
pool := openPGPoolWithMigrations(t)
if pool == nil {
return
}
defer pool.Close()
ctx := context.Background()
handler, srv := setupPGPiAgentServer(t, pool)
ts := httptest.NewServer(handler)
defer ts.Close()
client := &http.Client{Timeout: 50 * time.Minute}
seedOfficialWorkflowManifest(t, pool, workflowdef.OfficialKeyMain)
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
reviewerToken := loginTestUser(t, handler, "lizhuguan", "123456")
approverToken := loginTestUser(t, handler, "wangzong", "123456")
creatorID := fetchAuthUserID(t, handler, creatorToken)
reviewerID := fetchAuthUserID(t, handler, reviewerToken)
approverID := fetchAuthUserID(t, handler, approverToken)
createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{
Name: fmt.Sprintf("FullNone Wave2 %d", time.Now().UnixNano()%1_000_000),
DeviceType: "Ultrasonic Device",
IntendedUse: "Tissue resection",
TargetMarket: "NMPA Class III",
})
reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody))
withBearerAuth(reqCreate, creatorToken)
recCreate := httptest.NewRecorder()
handler.ServeHTTP(recCreate, reqCreate)
if recCreate.Code != http.StatusCreated {
t.Fatalf("create project: %d %s", recCreate.Code, recCreate.Body.String())
}
var proj projectworkflow.Project
if err := json.Unmarshal(recCreate.Body.Bytes(), &proj); err != nil {
t.Fatalf("decode project: %v", err)
}
t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
projectID := proj.ID
reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=0", projectID, workflowdef.OfficialKeyMain), nil)
withBearerAuth(reqDefault, creatorToken)
recDefault := httptest.NewRecorder()
handler.ServeHTTP(recDefault, reqDefault)
if recDefault.Code != http.StatusOK {
t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String())
}
var plan projectinstance.WorkflowPlan
_ = json.Unmarshal(recDefault.Body.Bytes(), &plan)
ensureProjectMembers(t, pool, projectID, map[int64]string{
creatorID: "owner",
reviewerID: "member",
approverID: "member",
})
for i := range plan.Stages {
stage := &plan.Stages[i]
for j := range stage.DocTasks {
dt := &stage.DocTasks[j]
switch {
case stage.StageKey == "V0.1_triage" && dt.TaskKey == "intake":
dt.Enabled = true
dt.Duties = projectinstance.DocDuty{
AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID,
}
case stage.StageKey == "V0.5_feasibility" && dt.TaskKey == "research":
dt.Enabled = true
dt.Duties = projectinstance.DocDuty{} // none
default:
dt.Enabled = false
dt.Duties = projectinstance.DocDuty{}
}
}
}
saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1})
reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody))
withBearerAuth(reqSave, creatorToken)
recSave := httptest.NewRecorder()
handler.ServeHTTP(recSave, reqSave)
if recSave.Code != http.StatusOK {
t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String())
}
reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil)
withBearerAuth(reqPreview, creatorToken)
recPreview := httptest.NewRecorder()
handler.ServeHTTP(recPreview, reqPreview)
if recPreview.Code != http.StatusOK {
t.Fatalf("activation preview: %d %s", recPreview.Code, recPreview.Body.String())
}
var preview projectinstance.ActivationPreview
_ = json.Unmarshal(recPreview.Body.Bytes(), &preview)
if preview.Summary.GovernedDocTasks < 1 {
t.Fatalf("expected at least one full doc, summary=%+v", preview.Summary)
}
if preview.Summary.UnmanagedDocTasks < 1 {
t.Fatalf("expected research as unmanaged none, summary=%+v", preview.Summary)
}
activateBody, _ := json.Marshal(projectinstance.ActivateInput{
ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion,
ExpectedPlanRevision: preview.ExpectedPlanRevision,
})
reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody))
withBearerAuth(reqActivate, creatorToken)
recActivate := httptest.NewRecorder()
handler.ServeHTTP(recActivate, reqActivate)
if recActivate.Code != http.StatusOK {
t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String())
}
var activated projectinstance.ActivateResult
_ = json.Unmarshal(recActivate.Body.Bytes(), &activated)
instanceID := activated.Instance.ID
t.Logf("✓ [1. activate] instance=%d full=intake none=research(disabled until stage)", instanceID)
runPI := func(taskKey string, materials []bizdev.MaterialInput) string {
t.Helper()
start := time.Now()
taskBody, _ := json.Marshal(server.TaskRunRequestDTO{
TaskKey: taskKey,
ProjectName: proj.Name,
DeviceType: proj.DeviceType,
IntendedUse: "Tissue resection",
TargetMarket: "NMPA Class III",
Backend: "pi-agent",
ProjectID: projectID,
AuthorID: creatorID,
ReviewerID: reviewerID,
ApproverID: approverID,
Fields: map[string]string{
"customer_name": "FullNone Wave2 客户",
"device_type": proj.DeviceType,
"intended_use": "Tissue resection",
"target_market": "NMPA Class III",
},
Materials: materials,
})
reqTask, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run", bytes.NewReader(taskBody))
withBearerAuth(reqTask, creatorToken)
respTask, err := client.Do(reqTask)
if err != nil {
t.Fatalf("pi-agent %s: %v", taskKey, err)
}
defer respTask.Body.Close()
if respTask.StatusCode != http.StatusOK {
var buf bytes.Buffer
_, _ = buf.ReadFrom(respTask.Body)
t.Fatalf("pi-agent %s HTTP %d: %s", taskKey, respTask.StatusCode, buf.String())
}
var taskRes struct {
Success bool `json:"success"`
DocRecord *struct {
DocID string `json:"doc_id"`
} `json:"doc_record"`
}
_ = json.NewDecoder(respTask.Body).Decode(&taskRes)
if !taskRes.Success {
t.Fatalf("pi-agent %s failed", taskKey)
}
docID := fmt.Sprintf("doc_proj%d_%s", projectID, taskKey)
if taskRes.DocRecord != nil && taskRes.DocRecord.DocID != "" {
docID = taskRes.DocRecord.DocID
}
t.Logf("✓ pi-agent %s doc=%s elapsed=%s", taskKey, docID, time.Since(start).Round(time.Second))
return docID
}
intakeDoc := runPI("intake", []bizdev.MaterialInput{{
Filename: "project_brief.md",
Content: []byte("# FullNone\n\nintake full duty chain\n"),
}})
docEngine := bizdev.Snapshot(srv.BizdevService()).Gov
_ = docEngine.MarkDocumentContentUpdated(ctx, intakeDoc, "hash_fullnone_intake", false)
if _, err := docEngine.SubmitForReview(ctx, intakeDoc, creatorID, "author", "Author"); err != nil {
t.Fatalf("submit intake: %v", err)
}
if _, err := docEngine.ReviewApprove(ctx, intakeDoc, reviewerID, "reviewer", "Reviewer", "ok"); err != nil {
t.Fatalf("review intake: %v", err)
}
if _, err := docEngine.ApproverApprove(ctx, intakeDoc, approverID, "approver", "Approver", "ok"); err != nil {
t.Fatalf("approve intake: %v", err)
}
t.Log("✓ [2. full intake] cross-user approve complete")
advanceBody, _ := json.Marshal(projectinstance.AdvanceStageInput{
StageKey: "V0.1_triage", ExpectedConfigRevision: activated.Instance.ConfigRevision,
})
reqAdvance := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/V0.1_triage/advance", instanceID), bytes.NewReader(advanceBody))
withBearerAuth(reqAdvance, creatorToken)
recAdvance := httptest.NewRecorder()
handler.ServeHTTP(recAdvance, reqAdvance)
if recAdvance.Code != http.StatusOK {
t.Fatalf("advance V0.1: %d %s", recAdvance.Code, recAdvance.Body.String())
}
t.Log("✓ [3. advance V0.1] ok")
researchDoc := runPI("research", []bizdev.MaterialInput{{
Filename: "research_scope.md",
Content: []byte("# Research none\n\n生成后不走人工审批\n"),
}})
// none: HTTP submit-review must be denied by duty gate (engine alone is not the gate)
submitBody, _ := json.Marshal(map[string]any{"doc_id": researchDoc})
reqSubmit := httptest.NewRequest(http.MethodPost, "/api/v1/docgov/submit-review", bytes.NewReader(submitBody))
withBearerAuth(reqSubmit, creatorToken)
recSubmit := httptest.NewRecorder()
handler.ServeHTTP(recSubmit, reqSubmit)
if recSubmit.Code == http.StatusOK {
t.Fatalf("none doc must reject submit-review, got 200: %s", recSubmit.Body.String())
}
if !bytes.Contains(recSubmit.Body.Bytes(), []byte("DOC_DUTY_NOT_MANAGED")) &&
!bytes.Contains(recSubmit.Body.Bytes(), []byte("未纳入人工审批")) &&
recSubmit.Code != http.StatusForbidden {
t.Fatalf("none submit want forbidden/duty-not-managed, got %d %s", recSubmit.Code, recSubmit.Body.String())
}
t.Logf("✓ [4. none research] generate ok; submit-review denied HTTP %d", recSubmit.Code)
reqReady := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/V0.5_feasibility/readiness", instanceID), nil)
withBearerAuth(reqReady, creatorToken)
recReady := httptest.NewRecorder()
handler.ServeHTTP(recReady, reqReady)
if recReady.Code != http.StatusOK {
t.Fatalf("readiness: %d %s", recReady.Code, recReady.Body.String())
}
var readiness projectinstance.StageReadiness
if err := json.Unmarshal(recReady.Body.Bytes(), &readiness); err != nil {
t.Fatalf("decode readiness: %v body=%s", err, recReady.Body.String())
}
var researchItem *projectinstance.DocTaskReadiness
for i := range readiness.DocTasks {
if readiness.DocTasks[i].TaskKey == "research" {
researchItem = &readiness.DocTasks[i]
break
}
}
if researchItem == nil {
t.Fatalf("research missing in readiness: %+v", readiness)
}
if researchItem.ApprovalManaged || researchItem.Required {
t.Fatalf("none research must not be approval-managed/required: %+v", researchItem)
}
if !readiness.Ready {
t.Fatalf("stage must be ready without approving none doc: %+v", readiness)
}
t.Log("✓ [5. readiness] none does not block")
// Config revision may bump after V0.1 advance — reload instance.
reqInst := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/workflow-instances/%d", instanceID), nil)
withBearerAuth(reqInst, creatorToken)
recInst := httptest.NewRecorder()
handler.ServeHTTP(recInst, reqInst)
if recInst.Code != http.StatusOK {
t.Fatalf("get instance: %d %s", recInst.Code, recInst.Body.String())
}
var detail struct {
Instance projectinstance.WorkflowInstance `json:"instance"`
}
if err := json.Unmarshal(recInst.Body.Bytes(), &detail); err != nil {
t.Fatalf("decode instance: %v", err)
}
advanceBody2, _ := json.Marshal(projectinstance.AdvanceStageInput{
StageKey: "V0.5_feasibility", ExpectedConfigRevision: detail.Instance.ConfigRevision,
})
reqAdvance2 := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/V0.5_feasibility/advance", instanceID), bytes.NewReader(advanceBody2))
withBearerAuth(reqAdvance2, creatorToken)
recAdvance2 := httptest.NewRecorder()
handler.ServeHTTP(recAdvance2, reqAdvance2)
if recAdvance2.Code != http.StatusOK {
t.Fatalf("advance V0.5: %d %s", recAdvance2.Code, recAdvance2.Body.String())
}
t.Log("✓ [6. advance V0.5] none non-blocking PASS")
// Refresh analogue: re-fetch settings after stage advance.
reqSettings := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/settings", projectID), nil)
withBearerAuth(reqSettings, creatorToken)
recSettings := httptest.NewRecorder()
handler.ServeHTTP(recSettings, reqSettings)
if recSettings.Code != http.StatusOK {
t.Fatalf("settings refresh: %d %s", recSettings.Code, recSettings.Body.String())
}
t.Log("✓ [7. refresh] settings still readable")
}
+1 -1
View File
@@ -448,7 +448,7 @@ func (c *Commands) appendTerminalEvent(ctx context.Context, record *RunRecord, r
OccurredAt: time.Now().UTC(),
Payload: body,
}
return (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"})
return (events.PGSink{}).Append(ctx, env, []string{events.ConsumerNotify})
}
func (c *Commands) stageControlledDoc(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error {