merge: feature/taskrun-runmanifest

M3 U8 TaskRun 终态与 RunManifest。
2026-09-15 02:17:37 +00:00
parent b8f33b5c55 f4dfada9ae
commit 367473730b
34 changed files with 2078 additions and 138 deletions
@@ -0,0 +1,60 @@
+---
+title: "M3 U8 TaskRun 终态与 RunManifest"
+status: done
+created: 2026-09-15
+updated: 2026-09-15
+owner: Cursor
+branch: feature/taskrun-runmanifest
+worktree: "-"
+mutex: taskrun
+---
+
+# M3 U8 TaskRun 终态与 RunManifest
+
+## 目标
+
+TaskRun 终态在同一 UoW 内 CAS 写入唯一合法状态，并持久化类型化 RunManifest；重复 finalize / cancel race / SSE 断线不得产生第二条助手消息或第二份 artifact；workspace 清理仅在 verified receipt 之后。
+
+## 范围
+
+- Goose 00037 `task_run_manifests` + `finalizing` 状态
+- `internal/taskrun` RunManifest 规范字节 / SHA / members
+- `taskrun.Commands.Finalize` 唯一 finalize 命令；server 变薄
+- `ctxmgr.FinalizeAssistantTaskMessage` 仅此写路径加入 UoW
+- workspace 清理门禁；`ReadArtifact` 替代 `Join(workDir, "output.md")`
+- BackendRef 服务端 policy；隔离库并发测试；占槽 rebuild；U8 报告
+
+## 非目标
+
+- U9 backfill / corpus / 八轨 / 销毁
+- M4–M6；整支 merge FAV；启用 `FAV_DEGRADED_READ`
+- 终止或改写项目 143
+- 动 `chat` / `intent` 刷新任务
+- worktree 内 `env-service`
+
+## 步骤
+
+1. [x] CURRENT 认领 `mutex=taskrun`；从本地 `dev` 开 `feature/taskrun-runmanifest`
+2. [x] Goose 00037 Up+Down；`module-boundaries.yaml` 表属主
+3. [x] Finalize 同 UoW：CAS + 助手 turn + manifest + envelope/staging
+4. [x] 清理门禁与 server 去路径、BackendRef
+5. [x] 隔离库双 pool / race；占槽 rebuild；报告最后一句只能「继续 U9」或「停止扩面」
+
+## 验收标准
+
+- [x] 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍
+- [x] finalize ∥ cancel：恰好一个合法终态
+- [x] SSE `done` 不出现在 TX 失败路径；commit 前断线重试不第二 turn
+- [x] 无 verified receipt 则 cleanup FAIL；有 receipt 则可清 workspace 且 staging 仍能取出原 SHA
+- [x] 报告最后一句为「继续 U9」或「停止扩面」；U9 清单项保持未勾
+
+## 相关链接
+
+- Checklist: [2026-09-14-module-decoupling-refactor-acceptance.md](../checklists/2026-09-14-module-decoupling-refactor-acceptance.md)
+- Remediation: [2026-09-14-module-decoupling-refactor-plan.md](../../remediation/2026-09-14-module-decoupling-refactor-plan.md)
+- 权威停走门: [2026-09-14-document-reliability-m2-test-report.md](../../test-reports/2026-09-14-document-reliability-m2-test-report.md)（继续 M3）
+- Test report: [2026-09-15-taskrun-runmanifest-u8-test-report.md](../../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)
+
+## 证据 / 备注
+
+未签字前禁止 U9–U12。合入 + 容器测占集成槽，主仓 `./scripts/env-service.sh dev rebuild`。
@@ -1,11 +1,11 @@
 # 当前焦点

-**更新**: 2026-09-14
+**更新**: 2026-09-15
 **production_ready**: no（`CHAT-INTENT-STATE` 等无关 blocker 仍开放。文档账本运行时 UI Round20 **PASS**，不把 fallback change report 写成完整 PI，也不把本轮写成全仓可发布）

 ## 焦点

-1. **DOC-RELIABILITY**: [M2 已签，继续 M3](../test-reports/2026-09-14-document-reliability-m2-test-report.md)（`refactor/document-reliability-pilot`；M2 最后一句 **继续 M3**；未开工 U8–U12）
+1. **DOC-RELIABILITY**: [M3 U8 已签，继续 U9](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md)（任务已归档 [2026-09-15-taskrun-runmanifest-u8](../archive/ops/2026-09-15-taskrun-runmanifest-u8.md)；U8 最后一句 **继续 U9**；**未签字前禁止 U10–U12**）
 2. **CHAT-INTENT**: [对话意图运行时状态](../audit/2026-09-13-conversation-intent-runtime-state.md)（`fix/conversation-intent-state`；与文档账本运行时无关）
 3. **CHAT**: [对话历史刷新](../audit/2026-09-13-chat-history-refresh-disappears.md)（认领仍 `in_progress`；gap 行已 PASS，以代码/handoff 为准收敛）

@@ -13,6 +13,7 @@

 | owner | task | branch | worktree | mutex | status |
 |-------|------|--------|----------|-------|--------|
+| Cursor | taskrun-runmanifest-u8 | feature/taskrun-runmanifest | - | taskrun | done |
 | Cursor | document-reliability-m1-live-and-m2 | refactor/document-reliability-pilot | .worktrees/refactor-document-reliability-pilot | workspace | done |
 | Codex | module-decoupling-remediation-plan | feature/module-decoupling-remediation | .worktrees/feature/module-decoupling-remediation | architecture-docs | done |
 | verifier_antigravity | controlled-content-ui-acceptance | fix/controlled-content-ui-acceptance | .worktrees/fix-controlled-content-ui-acceptance | document-content-ui-acceptance | done |
@@ -54,7 +55,7 @@
 > 新任务开始前登记。`branch` 必须为 `feature/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。

 > workflow-plan 409 与治理鉴权 G1–G5 已收口（见 [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)）。禁止终止或改写项目 143。
-> `document-reliability-pilot` M2 已签（继续 M3）。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U8–U12。
+> `document-reliability-pilot` U8 已签（继续 U9）。禁止 worktree 内 `env-service`。项目 143 只读，禁止终止。未签字前禁止 U10–U12。

 ## 集成槽位

@@ -70,7 +71,7 @@

 | 环境 | 端口 | schema | 状态 |
 |------|------|--------|------|
-| dev | 8080 | **36** | healthy（`pi-agent`；M2 rebuild 后 health+ready PASS） |
+| dev | 8080 | **37** | healthy（`pi-agent`；U8 rebuild 后 health+ready PASS） |
 | test | 8081 | 25→27* | healthy |
 | main | 8000 | 25→27* | healthy（app + runner + DB） |

@@ -93,6 +94,7 @@
 | 文档可靠性 M1 报告 | [../test-reports/2026-09-14-document-reliability-pilot-test-report.md](../test-reports/2026-09-14-document-reliability-pilot-test-report.md) |
 | 文档可靠性 M1 活验 | [../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md) |
 | 文档可靠性 M2 | [../test-reports/2026-09-14-document-reliability-m2-test-report.md](../test-reports/2026-09-14-document-reliability-m2-test-report.md) |
+| 文档可靠性 M3 U8 | [../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) |
 | SSE/幂等审计 | [../audit/2026-09-13-document-revision-sse-idempotency.md](../audit/2026-09-13-document-revision-sse-idempotency.md) |
 | 质量门禁自污染审计 | [../audit/2026-09-13-document-revision-quality-gate-self-pollution.md](../audit/2026-09-13-document-revision-quality-gate-self-pollution.md) |
 | 当前文档路由审计 | [../audit/2026-09-13-document-revision-active-doc-routing.md](../audit/2026-09-13-document-revision-active-doc-routing.md) |
@@ -2,7 +2,7 @@
 title: "模块解耦重构验收清单"
 status: in_progress
 created: 2026-09-14
-updated: 2026-09-14
+updated: 2026-09-15
 owner: Cursor
 ---

@@ -37,11 +37,11 @@ owner: Cursor

 ## M3 TaskRun、workspace 与数据生命周期

- [ ] RunManifest 规范字节、SHA、成员和 storage identity 持久化。
- [ ] 重复 finalize、SSE 断线和 cancel race 不重复消息或 artifact。
- [ ] WorkspaceRef 无物理路径，路径遍历和链接逃逸负例 PASS。
- [ ] 旧 lease writer 被 fencing token 拒绝。
- [ ] workspace 清空后最终成果仍可定位和校验。
+- [x] RunManifest 规范字节、SHA、成员和 storage identity 持久化。
+- [x] 重复 finalize、SSE 断线和 cancel race 不重复消息或 artifact。
+- [x] WorkspaceRef 无物理路径，路径遍历和链接逃逸负例 PASS。
+- [x] 旧 lease writer 被 fencing token 拒绝。
+- [x] workspace 清空后最终成果仍可定位和校验。
 - [ ] backfill 可暂停、续跑和重放；异常进入 quarantine。
 - [ ] 八条 migration/reload/recovery 轨道 PASS。
 - [ ] 数据销毁前完整备份、corpus、FAV coverage 和人工批准全部满足。
@@ -70,7 +70,7 @@
 | DOCREV-ACTIVE-DOC-ROUTING | UI-05「当前受控文档」被法规关键词改路由到 `feasibility_reg`，intake 无 pending | **PASS** | [audit](../audit/2026-09-13-document-revision-active-doc-routing.md)；Round20 UI-05 pending 停在 intake V2 | **不改弱 E2E**；不删证据 |
 | DOCREV-QUALITY-STATUS-FP | 行动计划表状态列「待续」被门禁当敷衍，两轮 repair 后 quality_failed 80 | **PASS** | [audit](../audit/2026-09-13-document-revision-quality-status-false-positive.md)；Round20 UI-03 | **不删禁词、不放宽正文**；不删证据 |
 | GOVERNANCE-ACTION-AUTHZ | 文档动作按职责鉴权；未激活 202-blocked；关闭创建直通 | **PASS** | [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)；[audit 异态](../audit/2026-09-14-planning-with-workflow-key-anomaly.md) | 不改 83 条 planning+key 夹具；none 长 PI 审批链另开 |
-| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md) | M1 活验 **继续 M2**。M2 **继续 M3**。禁止在 M3 签字前写 U8–U12 业务代码 |
+| DOC-RELIABILITY-PILOT | 受控文档写路径跨表非同事务、读路径被 projection 绑架、授权 TOCTOU；刷新/重启后可能 404/409/projection-missing | OPEN | [复核](../audit/2026-09-14-module-decoupling-plan-review.md)；[重构计划](./2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | M2 **继续 M3**。U8 **继续 U9**。禁止在 U9 签字前写 U10–U12 业务代码 |

 ---

@@ -1,6 +1,6 @@
 # 模块与功能完成度（SSOT）

-**更新**: 2026-09-14
+**更新**: 2026-09-15
 **production_ready**: no
 **发布缺口**: [../remediation/gap-matrix.md](../remediation/gap-matrix.md)
 **协作焦点**: [../ops/CURRENT.md](../ops/CURRENT.md)
@@ -27,7 +27,7 @@
 | PI Runner / 配置链 | pi-agent + MiniMax 真实链路 | done | [test piagent](../test-reports/2026-09-13-codex-audit-p0-remediation-test-report.md)；[container-env](../test-reports/2026-09-13-container-env-configuration-test-report.md) | — | `pi-agent` |
 | 三环境运维 | dev/test/main + 共享 KB | done | [PROJECT.md](../../PROJECT.md) §4；四环境 healthy | — | `infra` |
 | CI / E2E | PG 集成 + 真实 PI E2E | partial | [gap-matrix BUG-08](../remediation/gap-matrix.md)；[test pg-ci](../test-reports/2026-09-12-pg-ci-soak-e2e-test-report.md)；[运行时 UI Round20](../test-reports/2026-09-14-document-revision-runtime-ui-test-report.md) | CI `TEST_DATABASE_URL` 隔离库仍缺；文档运行时真实 UI 已按 UI_PHASE 分文件且 post-restart 断言最终 A1 正文。`-race` 全包并行 TempDir flake 为基建残留 | `ci` |
-| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md) | U5–U7 PASS（WorkspaceRef/fencing、耐久投递、按文件 FAV）。活验 **继续 M2**；M2 最后一句 **继续 M3**。未开工 U8–U12 | `workspace` |
+| 模块解耦与可靠事件 | 文档可靠性纵向试点、跨属主事务、多副本互斥、FAV 与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[重构计划](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[试点任务](../ops/short-term/2026-09-14-document-reliability-pilot.md)；[M1 停止扩面](../test-reports/2026-09-14-document-reliability-pilot-test-report.md)；[M1 活验](../test-reports/2026-09-14-document-reliability-pilot-live-test-report.md)；[M2](../test-reports/2026-09-14-document-reliability-m2-test-report.md)；[M3 U8](../test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md) | U5–U8 PASS。M2 **继续 M3**；U8 最后一句 **继续 U9**。未开工 U9–U12 | `taskrun` |

 ---

@@ -49,6 +49,7 @@ imports:
    - squad
    - stream
  ctxmgr:
+    - app/uow
    - docgov
    - replyexport
    - storage
@@ -124,6 +125,7 @@ imports:
    - upload
    - workflowdef
    - workspace
+    - artifactvault
  squad:
    - agent
    - config
@@ -135,10 +137,13 @@ imports:
  stream: []
  taskrun:
    - app/uow
+    - artifactvault
    - bizdev
    - ctxmgr
+    - events
    - projectworkflow
    - stream
+    - workspace
  template: []
  upload: []
  workflow: []
@@ -169,6 +174,8 @@ table_owners:
    - freedoc
  task_runs:
    - taskrun
+  task_run_manifests:
+    - taskrun
  document_versions:
    - docrevision
  document_snapshots:
@@ -222,6 +229,10 @@ event_producers:
    - docrevision
  document.change_report_degraded:
    - docrevision
+  taskrun.completed:
+    - taskrun
+  taskrun.cancelled:
+    - taskrun

 event_consumers:
  outbox.notify:
@@ -235,6 +246,7 @@ unwrap_owners:
  - freedoc
  - events
  - artifactvault
+  - ctxmgr

 exceptions:
  - kind: sql_write
@@ -397,14 +409,6 @@ exceptions:
    reason: server still concatenates workspaces/ roots; U5
    milestone: U5
    expires: "2026-10-12"
-  - kind: path
-    package: server
-    file: internal/server/task_finalize.go
-    symbol: output.md
-    owner: server
-    reason: finalize still joins output.md; U8 verified receipt
-    milestone: U8
-    expires: "2026-10-26"
  - kind: path
    package: storage
    file: internal/storage/cas.go
@@ -417,5 +421,5 @@ exceptions:
 baseline_counts:
  import: 0
  sql_write: 5
-  path: 17
+  path: 16
  event: 0
@@ -0,0 +1,124 @@
+# M3 U8 TaskRun 终态与 RunManifest 测试报告
+
+## 一、 测试背景 (Background)
+
+权威停走门是 M2 报告最后一句 **继续 M3**（[2026-09-14-document-reliability-m2-test-report.md](./2026-09-14-document-reliability-m2-test-report.md)）。整改计划 U8 要求 TaskRun 终态与类型化 RunManifest 同一 UoW，并在重复 finalize、cancel 竞态、SSE 断线和 workspace 清理后保持一致。本轮不做 U9（dataset/八轨/销毁），不碰 M4–M6 / `chat` / `intent`。
+
+## 二、 测试目标 (Test Objectives)
+
+- RunManifest 规范字节、SHA-256、workspace_id、storage_identity、members 持久化；Evidence/Gaps 不进核心协议。
+- 重复 finalize：同一 status、同一 manifest SHA、助手消息 1 条、envelope/staging 不加倍。
+- finalize ∥ cancel：恰好一个合法终态（双 pool，不用进程全局 mutex）。
+- SSE `done` 不出现在 TX 失败路径；commit 前失败后占位正文不落终态。
+- workspace 无 verified receipt 则 cleanup FAIL；有 receipt 则可清，且 staging 仍能取出原 SHA。
+- server 不再 `Join(workDir, "output.md")`；Backend 由服务端 policy 解析。
+- Goose 00037 Up+Down；占槽 rebuild 后 schema **37**。
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+| 项 | 值 |
+|----|-----|
+| OS | Linux 7.0.0-31-generic |
+| Go | 本机 `go test`；容器 `agent-dev-app` |
+| 分支 | `feature/taskrun-runmanifest`（从本地 `dev` 检出） |
+| 部署 | 主仓 `./scripts/env-service.sh dev rebuild`（禁止 worktree 内 env-service） |
+| Compose | 项目 `agent-dev`，HTTP `:8080` |
+| Agent | `agent_backend=pi-agent` |
+| Schema | **37**（00037 `task_run_manifests`） |
+| PostgreSQL | 隔离库 `agent_rel_m3_u8`（admin 连 `shuimu-system-pg-test:5434/postgres` 自建）；测完 `DROP`；禁止 `agent_system_dev/test/main` |
+| 项目 143 | **只读**；本轮未 terminate，未改 workflow-plan |
+| 证据 | `.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/` |
+
+镜像身份：`app_image=sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9`。见 `rebuild-health-ready.txt`。
+
+提交：`2234af4` docs(ops) 认领 → `eb75f85` feat(db) → `7be8730` feat(taskrun) → `2ddaf76` feat(workspace) → `3997a04` test(taskrun)。
+
+## 四、 预期结果 (Expected Results)
+
+| 栏 | 期望 |
+|----|------|
+| Unit | 规范 SHA 稳定；未授权 backend / 敏感正文拒绝；重复 finalize 不改 turn；cancel 后 finalize 采用已有终态 |
+| PG | 双 pool 重复 finalize 计数均为 1；cancel race 唯一终态；commit 失败无 `done`；无 receipt 拒清理 |
+| 故障注入 | `BeforeCommit` 失败回滚助手占位；`done` 不出现 |
+| 真实 PI | 本轮未新开编制任务；dev 已 `pi-agent` ready；143 只读 |
+| 浏览器 | web 单测 + production build 必做；页面回归继承 M1 |
+| restart-rebuild | health/ready schema 37，`pi-agent` |
+| 跨用户 | Backend 调用方不能塞任意字符串；敏感字段从 metadata 剥离 |
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+### 5.1 Unit / stub
+
+- **测试输入**: `unset TEST_DATABASE_URL; go test -short ./internal/taskrun ./internal/workspace ./internal/server -run 'TestFinalize|TestCommands|TestRunManifest|TestCleanup|TestOpaque|TestManager_Done'`
+- **实际输出**: `TestRunManifest_FreezeStableSHA`、`TestCommands_DuplicateFinalizeMemory`、`TestCommands_CancelWinsOverFinalizeMemory`、`TestCleanup_RequiresVerifiedReceipts`、`TestFinalizeTaskRun_HashesViaWorkspaceCapability`、`TestManager_FinalizeFailureDoesNotEmitDone` 均为 PASS。
+- **评价判定**: **PASS**
+- **性能与耗时**: 亚秒。日志：`go-test-short.txt`
+
+### 5.2 PostgreSQL 隔离库（双 pool）
+
+- **测试输入**: `TEST_DATABASE_URL` 指向 5434 上非共享库名；`go test ./internal/taskrun -run TestPG_`。Admin 连 `postgres`，`CREATE DATABASE agent_rel_m3_u8`，goose 至 37。
+- **实际输出**:
+  - `TestPG_DuplicateFinalizeIdempotent`：messages=1 manifests=1 envelopes=1 staging=1，两侧 SHA 相同
+  - `TestPG_FinalizeCancelRaceOneTerminal`：status ∈ {completed, cancelled}，助手 turn=1
+  - `TestPG_CleanupWithoutReceiptFailsAndStagingSurvives`：无 receipt 目录仍在；有 receipt 后可删；`LoadStaging` SHA 与原文一致
+- **评价判定**: **PASS**
+- **性能与耗时**: 含 goose 约 0.35s。测完 `DROP DATABASE`；`psql` 确认 `agent_rel_m3%` 为 0 行。
+
+### 5.3 故障注入 / SSE
+
+- **测试输入**: `uow.PG.BeforeCommit` 返回 `inject commit fail`；Manager worker finalize。
+- **实际输出**: status=`finalizing_failed`；`task_run_events` 无 `done`；助手正文仍为「任务执行中…」。
+- **评价判定**: **PASS**
+- **性能与耗时**: 40ms
+
+### 5.4 真实 PI
+
+- **测试输入**: 本轮未对新项目发起编制；`GET /health` 显示 `agent_backend=pi-agent`。项目 143 只读未打开。
+- **实际输出**: 模型容器在线，但 U8 未跑一条新的真实编制链路。
+- **评价判定**: **FAIL（未测）** — 不构成 U8 停走项（停走项是重复 finalize / 双终态 / 提前 `done` / 无 receipt 清理 / 仅进程 mutex）。浏览器页面回归继承 M1。
+- **性能与耗时**: 未测
+
+### 5.5 浏览器
+
+- **测试输入**: `cd web && npm test`；`npm run build`
+- **实际输出**: tsx 365 passed + vitest 30 passed；production build ✓ 8.29s
+- **评价判定**: **PASS**（单测+build）。页面 finalize 后刷新继承 M1，本轮未重跑 Playwright。
+- **性能与耗时**: 单测 ~1.3s；build 8.29s
+
+### 5.6 restart / rebuild
+
+- **测试输入**: 主仓占槽 `./scripts/env-service.sh dev rebuild`；`curl -sf http://127.0.0.1:8080/health` 与 `/ready`
+- **实际输出**:
+
+```json
+{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}
+```
+
+- **评价判定**: **PASS**
+- **性能与耗时**: rebuild 约 23s。日志：`rebuild.log`、`health.json`、`ready.json`
+
+### 5.7 跨用户 / Backend policy
+
+- **测试输入**: `TestResolveBackend_StripsUnauthorized`；`TestStripSecretsAndSensitive`；`TestCommands_RejectsSensitiveMarkdown`
+- **实际输出**: `evil-shell` 拒绝；`api_key` 从 metadata 剥离；含 `api_key=` 的正文 finalize 失败
+- **评价判定**: **PASS**
+- **性能与耗时**: 亚秒
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+| 命令 | 结果 |
+|------|------|
+| `go vet ./...` | PASS |
+| `go test -short ./...` | PASS |
+| 隔离库 goose 00037 Up + 双 pool 竞态 + `DROP DATABASE` | PASS |
+| 目标包 `-race`（taskrun/workspace/ctxmgr） | PASS |
+| `arch-check --mode warn` | PASS |
+| web `npm test` | 365 + 30 PASS |
+| web production build | PASS |
+| 占槽 rebuild + health/ready | PASS，`schema_version=37` |
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+U8 停走项全部观测为 PASS：重复 finalize 不产生第二 turn / 第二份 staging；cancel 与 finalize 只有一个合法终态；TX 失败路径无 SSE `done`；无 verified receipt 不得删 workspace；正确性由 `SELECT … FOR UPDATE` 与 UNIQUE(run_id) 承担，而不是进程 mutex。Goose 00037 可随隔离库 Down（随 Down to 33 验证表消失）。server 已去掉 `filepath.Join(workDir, "output.md")`。未做 U9 backfill/corpus/八轨/销毁，未整支 merge FAV，未启用 `FAV_DEGRADED_READ`，未终止项目 143。真实 PI 编制本轮未新开，不作为 U8 扩面失败理由。
+
+继续 U9
@@ -0,0 +1,4 @@
+arch-check mode=warn
+boundaries=/home/weichen/projects/Agent-System/.doc/system-design/module-boundaries.yaml
+ok  	chat-agent-system/internal/archtest	0.102s
+arch-check warn: production gate passed
@@ -0,0 +1,47 @@
+ok  	chat-agent-system/cmd/bizdev-cli	2.342s
+ok  	chat-agent-system/cmd/document-content-reconcile	(cached)
+?   	chat-agent-system/cmd/document-revision-backfill	[no test files]
+?   	chat-agent-system/cmd/recategorize	[no test files]
+?   	chat-agent-system/cmd/server	[no test files]
+ok  	chat-agent-system/internal/agent	1.178s
+ok  	chat-agent-system/internal/app/docflow	0.003s
+ok  	chat-agent-system/internal/app/uow	0.003s
+ok  	chat-agent-system/internal/archtest	0.121s
+ok  	chat-agent-system/internal/artifactvault	0.004s
+ok  	chat-agent-system/internal/auth	(cached)
+ok  	chat-agent-system/internal/authz	(cached)
+ok  	chat-agent-system/internal/bizdev	0.161s
+ok  	chat-agent-system/internal/chat	0.015s
+ok  	chat-agent-system/internal/config	(cached)
+ok  	chat-agent-system/internal/coordinator	0.011s
+ok  	chat-agent-system/internal/ctxmgr	(cached)
+ok  	chat-agent-system/internal/db	0.003s
+ok  	chat-agent-system/internal/docduty	(cached)
+ok  	chat-agent-system/internal/docgov	0.018s
+ok  	chat-agent-system/internal/docquery	0.003s
+ok  	chat-agent-system/internal/docrevision	0.006s
+ok  	chat-agent-system/internal/events	0.003s
+ok  	chat-agent-system/internal/freedoc	0.027s
+ok  	chat-agent-system/internal/fsutil	(cached)
+ok  	chat-agent-system/internal/intent	0.018s
+ok  	chat-agent-system/internal/kb	(cached)
+ok  	chat-agent-system/internal/notification	0.070s
+?   	chat-agent-system/internal/observability	[no test files]
+ok  	chat-agent-system/internal/projectcontext	0.004s
+ok  	chat-agent-system/internal/projectinstance	0.004s
+ok  	chat-agent-system/internal/projectworkflow	(cached)
+ok  	chat-agent-system/internal/quote	(cached)
+ok  	chat-agent-system/internal/replyexport	(cached)
+ok  	chat-agent-system/internal/server	10.015s
+ok  	chat-agent-system/internal/squad	0.034s
+ok  	chat-agent-system/internal/storage	(cached)
+ok  	chat-agent-system/internal/stream	(cached)
+ok  	chat-agent-system/internal/taskrun	0.146s
+ok  	chat-agent-system/internal/template	0.709s
+ok  	chat-agent-system/internal/upload	(cached)
+?   	chat-agent-system/internal/workflow	[no test files]
+ok  	chat-agent-system/internal/workflow/runtime	(cached)
+ok  	chat-agent-system/internal/workflowdef	(cached)
+ok  	chat-agent-system/internal/workspace	(cached)
+?   	chat-agent-system/templates	[no test files]
+SHORT_EXIT:0
@@ -0,0 +1,4 @@
+ok  	chat-agent-system/internal/taskrun	1.961s
+ok  	chat-agent-system/internal/workspace	1.012s
+ok  	chat-agent-system/internal/ctxmgr	1.146s
+RACE_EXIT:0
@@ -0,0 +1 @@
+{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}
@@ -0,0 +1 @@
+{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}
@@ -0,0 +1,4 @@
+sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9 [agent-dev-app@sha256:fb3eec56ae5684fd323b1bd9e3c5c521c2e415730c2edd7f8c65a86b85aa6fb9]
+{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}
+
+{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}
@@ -0,0 +1,54 @@
+-- +goose Up
+-- U8: typed RunManifest persistence and optional finalizing status.
+
+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',
+        'finalizing',
+        'completed',
+        'failed',
+        'cancelled',
+        'finalizing_failed',
+        'quality_failed',
+        'interrupted',
+        'pending_author_confirmation',
+        'blocked_pending_governance'
+    )
+);
+
+CREATE TABLE IF NOT EXISTS task_run_manifests (
+    run_id            TEXT PRIMARY KEY REFERENCES task_runs(run_id) ON DELETE CASCADE,
+    canonical_bytes   BYTEA NOT NULL,
+    sha256            CHAR(64) NOT NULL,
+    workspace_id      TEXT NOT NULL DEFAULT '',
+    storage_identity  TEXT NOT NULL DEFAULT '',
+    members           JSONB NOT NULL DEFAULT '[]'::jsonb,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT uq_task_run_manifests_run UNIQUE (run_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_task_run_manifests_sha ON task_run_manifests (sha256);
+
+-- +goose Down
+DROP INDEX IF EXISTS idx_task_run_manifests_sha;
+DROP TABLE IF EXISTS task_run_manifests;
+
+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',
+        'blocked_pending_governance'
+    )
+);
@@ -5,6 +5,11 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+
+	"chat-agent-system/internal/app/uow"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
 )

 // UpdateMessageMetadata 更新已持久化助手消息的 metadata 与任务状态。
@@ -19,9 +24,9 @@ func (m *ContextManager) UpdateMessageMetadata(ctx context.Context, messageID in
 		meta["task_status"] = taskStatus
 	}

-	if m.pool != nil {
+	if m.pool != nil || hasUoW(ctx) {
 		metaJSON, _ := json.Marshal(meta)
-		_, err := m.pool.Exec(ctx, `
+		_, err := execChat(ctx, m.pool, `
 			UPDATE chat_messages
 			SET metadata = COALESCE(metadata, '{}'::jsonb) || $1::jsonb
 			WHERE id = $2
@@ -63,12 +68,12 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 		meta["task_status"] = taskStatus
 	}

-	if m.pool != nil {
+	if m.pool != nil || hasUoW(ctx) {
 		metaJSON, err := json.Marshal(meta)
 		if err != nil {
 			return fmt.Errorf("marshal metadata failed: %w", err)
 		}
-		tag, err := m.pool.Exec(ctx, `
+		tag, err := execChat(ctx, m.pool, `
 			UPDATE chat_messages
 			SET content = $1,
 			    metadata = COALESCE(metadata, '{}'::jsonb) || $2::jsonb
@@ -102,3 +107,22 @@ func (m *ContextManager) FinalizeAssistantTaskMessage(ctx context.Context, messa
 	}
 	return errors.New("message not found")
 }
+
+func hasUoW(ctx context.Context) bool {
+	_, ok := uow.ScopeFrom(ctx)
+	return ok
+}
+
+func execChat(ctx context.Context, pool *pgxpool.Pool, q string, args ...any) (pgconn.CommandTag, error) {
+	if scope, ok := uow.ScopeFrom(ctx); ok {
+		tx, err := uow.AsPG(scope)
+		if err != nil {
+			return pgconn.CommandTag{}, err
+		}
+		return tx.Exec(ctx, q, args...)
+	}
+	if pool == nil {
+		return pgconn.CommandTag{}, errors.New("ctxmgr: database required")
+	}
+	return pool.Exec(ctx, q, args...)
+}
@@ -100,6 +100,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
 		"00034_2026-09-14-workspace-lease-fencing.sql",
 		"00035_2026-09-14-durable-event-delivery.sql",
 		"00036_2026-09-14-final-artifact-vault.sql",
+		"00037_2026-09-15-task-run-manifests.sql",
 	}
 	joined := strings.Join(files, "\n")
 	for _, name := range want {
@@ -135,7 +136,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
 		t.Fatalf("up: %v", err)
 	}
 	ver, err := CurrentVersion(ctx, pool)
-	if err != nil || ver < 36 {
+	if err != nil || ver < 37 {
 		t.Fatalf("version after up: %d %v", ver, err)
 	}

@@ -148,7 +149,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) {
 	if err := goose.DownToContext(ctx, sqlDB, migDir, 33); err != nil {
 		t.Fatalf("down to 33: %v", err)
 	}
-	for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts"} {
+	for _, table := range []string{"event_envelopes", "event_deliveries", "event_dead_letters", "final_artifacts", "task_run_manifests"} {
 		var n int
 		if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM information_schema.tables WHERE table_name=$1`, table).Scan(&n); err != nil {
 			t.Fatal(err)
@@ -49,3 +49,13 @@ func SkipIfNoTestDatabaseURL(t *testing.T) string {
 	}
 	return dsn
 }
+
+// RefuseSharedEnvDSN fails if the DSN targets a shared env database.
+func RefuseSharedEnvDSN(t *testing.T, dsn string) {
+	t.Helper()
+	if strings.Contains(dsn, "5433") || strings.Contains(dsn, "agent_system_dev") ||
+		(strings.Contains(dsn, "5434") && strings.Contains(dsn, "agent_system_test")) ||
+		strings.Contains(dsn, "5435") {
+		t.Fatal("TEST_DATABASE_URL points at a shared env port; refuse agent_system_dev/test/main")
+	}
+}
@@ -259,7 +259,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 			IntendedUse:     req.IntendedUse,
 			TargetMarket:    req.TargetMarket,
 			Fields:          req.Fields,
-			Backend:         req.Backend,
+			Backend:         resolveTaskBackend(s, req.Backend),
 			AutoRepair:      req.AutoRepair,
 			UserID:          req.UserID,
 			ProjectID:       req.ProjectID,
@@ -308,7 +308,7 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 		TargetMarket:    req.TargetMarket,
 		Fields:          req.Fields,
 		Materials:       req.Materials,
-		Backend:         req.Backend,
+		Backend:         resolveTaskBackend(s, req.Backend),
 		AutoRepair:      req.AutoRepair,
 		UserID:          req.UserID,
 		ProjectID:       req.ProjectID,
@@ -11,6 +11,8 @@ import (
 	"time"

 	"chat-agent-system/internal/agent"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/artifactvault"
 	"chat-agent-system/internal/auth"
 	"chat-agent-system/internal/authz"
 	"chat-agent-system/internal/bizdev"
@@ -45,6 +47,8 @@ type Server struct {
 	instSvc    *projectinstance.Service
 	notifySvc  *notification.Service
 	resolver   *workspace.Resolver
+	wsCap      *workspace.Capability
+	cmds       *taskrun.Commands
 	httpServer *http.Server
 }

@@ -93,6 +97,18 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
 	if pool != nil {
 		taskMgr.SetStore(taskrun.NewPGStore(pool))
 	}
+	wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
+	cmds := &taskrun.Commands{
+		Ctx:             svc.ContextManager(),
+		PW:              pwSvc,
+		Env:             cfg.Env,
+		StorageIdentity: "env:" + cfg.Env,
+	}
+	if pool != nil {
+		cmds.UoW = uow.NewPG(pool)
+		cmds.Fav = &artifactvault.Ingest{Env: cfg.Env}
+	}
+	taskMgr.SetCommands(cmds)

 	s := &Server{
 		cfg:        cfg,
@@ -109,6 +125,8 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
 		instSvc:    instSvc,
 		notifySvc:  notification.NewService(pool),
 		resolver:   workspace.NewResolver(cfg.DataDir, pool),
+		wsCap:      wsCap,
+		cmds:       cmds,
 	}
 	if instSvc != nil && instSvc.Available() {
 		svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc))
@@ -334,7 +334,7 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
 		req.DeviceType = q.Get("device_type")
 		req.IntendedUse = q.Get("intended_use")
 		req.TargetMarket = q.Get("target_market")
-		req.Backend = q.Get("backend")
+		req.Backend = resolveTaskBackend(s, q.Get("backend"))
 		req.WorkDir = q.Get("work_dir")
 		req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair"))
 		if pid := q.Get("project_id"); pid != "" {
@@ -5,12 +5,11 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
-	"os"
-	"path/filepath"

 	"chat-agent-system/internal/bizdev"
 	"chat-agent-system/internal/squad"
 	"chat-agent-system/internal/taskrun"
+	"chat-agent-system/internal/workspace"
 )

 // FinalizeTaskRunProjection 将任务执行结果投影到 workflow_runs 与聊天助手消息（供回填/管理命令使用）。
@@ -18,101 +17,69 @@ func (s *Server) FinalizeTaskRunProjection(ctx context.Context, record *taskrun.
 	return s.finalizeTaskRun(ctx, record, result)
 }

-// finalizeTaskRun 将任务执行结果投影到 workflow_runs 与聊天助手消息。
+// finalizeTaskRun 是薄适配：产物哈希在事务外经 workspace.Capability 读取，终态命令交给 taskrun.Commands。
 func (s *Server) finalizeTaskRun(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) (string, error) {
 	if record == nil || result == nil {
 		return "failed", fmt.Errorf("finalize: missing record or result")
 	}
+	if result.DocRecord != nil && result.DocRecord.ContentSHA256 == "" {
+		if hexSum := s.hashDeliverable(ctx, record, result); hexSum != "" {
+			result.DocRecord.ContentSHA256 = hexSum
+		}
+	}
+	cmds := s.taskCommands()
+	if s != nil {
+		cmds.PW = s.pwSvc
+		if s.svc != nil {
+			cmds.Ctx = s.svc.ContextManager()
+		}
+	}
+	return cmds.Finalize(ctx, record, result)
+}

-	workspaceID := result.WorkDir
-	if workspaceID == "" {
-		workspaceID = record.WorkspaceID
+func (s *Server) taskCommands() *taskrun.Commands {
+	if s != nil && s.cmds != nil {
+		return s.cmds
 	}
+	cmds := &taskrun.Commands{
+		PW:  s.pwSvc,
+		Env: "test",
+	}
+	if s != nil && s.cfg != nil {
+		cmds.Env = s.cfg.Env
+		cmds.StorageIdentity = "env:" + s.cfg.Env
+	}
+	if s != nil && s.svc != nil {
+		cmds.Ctx = s.svc.ContextManager()
+	}
+	return cmds
+}

-	terminalStatus := result.TerminalStatus()
-	summaryContent := result.CompletionSummary()
-
-	docID := ""
-	artifactTitle := result.ArtifactTitle
-	sha256Hex := ""
-	if result.DocRecord != nil {
-		docID = result.DocRecord.DocID
-		if result.DocRecord.ContentSHA256 != "" {
-			sha256Hex = result.DocRecord.ContentSHA256
-		}
+func (s *Server) hashDeliverable(ctx context.Context, record *taskrun.RunRecord, result *bizdev.TaskRunResult) string {
+	if s == nil || s.wsCap == nil || record == nil {
+		return ""
 	}
-	if docID == "" && record.ProjectID > 0 && record.TaskKey != "" {
-		docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey)
+	id := workspace.OpaqueID(result.WorkDir, record.WorkspaceID)
+	if id == "" {
+		return ""
 	}
-	if artifactTitle == "" && record.TaskKey != "" {
-		artifactTitle = squad.TaskArtifactTitle(record.TaskKey)
+	env := ""
+	if s.cfg != nil {
+		env = s.cfg.Env
 	}
-
-	if sha256Hex == "" && workspaceID != "" {
-		if computed, err := hashOutputFile(workspaceID); err == nil {
-			sha256Hex = computed
-		}
+	ref := workspace.WorkspaceRef{
+		ID:            id,
+		SchemaVersion: workspace.SchemaVersion,
+		ProjectScope:  record.ProjectID,
+		RunID:         record.RunID,
+		Environment:   env,
 	}
-
-	meta := map[string]any{
-		"run_id":       record.RunID,
-		"workspace_id": workspaceID,
-		"task_status":  terminalStatus,
-		"task_key":     record.TaskKey,
+	body, err := s.wsCap.ReadArtifact(ctx, ref, workspace.DeliverableKey)
+	if err != nil || len(body) == 0 {
+		return ""
 	}
-	if record.ProjectID > 0 {
-		meta["project_id"] = record.ProjectID
-	}
-	if art := result.ArtifactMetadata(); art != nil {
-		if workspaceID != "" {
-			art["workspace_id"] = workspaceID
-		}
-		if sha256Hex != "" {
-			art["sha256"] = sha256Hex
-		}
-		if artifactTitle != "" {
-			art["title"] = artifactTitle
-		}
-		if record.TaskKey != "" {
-			art["task_key"] = record.TaskKey
-		}
-		if docID != "" {
-			art["doc_id"] = docID
-		}
-		meta["artifact"] = art
-	} else if docID != "" {
-		artifact := map[string]any{
-			"doc_id":       docID,
-			"title":        artifactTitle,
-			"task_key":     record.TaskKey,
-			"workspace_id": workspaceID,
-		}
-		if sha256Hex != "" {
-			artifact["sha256"] = sha256Hex
-		}
-		if result.QualityReport != nil {
-			artifact["quality_passed"] = result.QualityReport.Passed
-			artifact["quality_score"] = result.QualityReport.Score
-			if len(result.QualityReport.Violations) > 0 {
-				artifact["quality_violations"] = result.QualityReport.Violations
-			}
-		}
-		meta["artifact"] = artifact
-	}
-
-	if s.pwSvc != nil && record.ProjectID > 0 && record.RunID != "" {
-		if err := s.pwSvc.UpdateWorkflowRun(ctx, record.RunID, terminalStatus, docID, artifactTitle, ""); err != nil {
-			return "finalizing_failed", fmt.Errorf("update workflow run: %w", err)
-		}
-	}
-
-	if s.svc != nil && s.svc.ContextManager() != nil && record.AssistantMsgID > 0 {
-		if err := s.svc.ContextManager().FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summaryContent, meta, terminalStatus); err != nil {
-			return "finalizing_failed", fmt.Errorf("finalize assistant message: %w", err)
-		}
-	}
-
-	return terminalStatus, nil
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:])
 }

 func resolveTerminalStatus(result *bizdev.TaskRunResult) string {
@@ -124,19 +91,6 @@ func buildTaskCompletionSummary(result *bizdev.TaskRunResult, terminalStatus str
 	return result.CompletionSummary()
 }

-func hashOutputFile(workDir string) (string, error) {
-	path := filepath.Join(workDir, "output.md")
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return "", err
-	}
-	if len(data) == 0 {
-		return "", fmt.Errorf("output.md is empty")
-	}
-	sum := sha256.Sum256(data)
-	return hex.EncodeToString(sum[:]), nil
-}
-
 func squadResultToTaskRun(res *squad.SquadRunResult, taskKey string) *bizdev.TaskRunResult {
 	if res == nil {
 		return nil
@@ -2,6 +2,8 @@ package server

 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"os"
 	"path/filepath"
 	"strings"
@@ -15,6 +17,7 @@ import (
 	"chat-agent-system/internal/squad"
 	"chat-agent-system/internal/stream"
 	"chat-agent-system/internal/taskrun"
+	"chat-agent-system/internal/workspace"
 )

 func TestFinalizeTaskRun_ReplacesPlaceholderContent(t *testing.T) {
@@ -298,3 +301,52 @@ func TestAsyncTaskRun_PendingConfirmation_FinalizesAsPending(t *testing.T) {
 	}
 	t.Fatal("worker did not finish")
 }
+
+func TestFinalizeTaskRun_HashesViaWorkspaceCapability(t *testing.T) {
+	cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"}
+	svc := bizdev.MustNewService(t, cfg)
+	srv := NewServer(cfg, svc)
+	ctx := context.Background()
+	ref, err := srv.wsCap.Issue(workspace.WorkspaceRef{ID: "ws_hash", ProjectScope: 1, RunID: "run_hash"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	body := []byte("# hashed via capability\n")
+	if err := srv.wsCap.WriteArtifact(ctx, ref, workspace.DeliverableKey, body); err != nil {
+		t.Fatal(err)
+	}
+	ctxMgr := svc.ContextManager()
+	asstID, err := ctxMgr.RecordUserMessageWithID(ctx, "sess-hash", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pwSvc := projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
+	_, _ = pwSvc.CreateWorkflowRun(ctx, &projectworkflow.WorkflowRun{
+		RunID: "run_hash", ProjectID: 1, TaskKey: "intake", Status: "running",
+	})
+	srv.pwSvc = pwSvc
+	record := &taskrun.RunRecord{RunID: "run_hash", SessionID: "sess-hash", ProjectID: 1, TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: ref.ID}
+	status, err := srv.finalizeTaskRun(ctx, record, &bizdev.TaskRunResult{
+		Success: true, TaskKey: "intake", WorkDir: ref.ID,
+		DocRecord: &docgov.DocumentRecord{DocID: "doc_hash"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status != "completed" {
+		t.Fatalf("status=%s", status)
+	}
+	msg, err := ctxMgr.GetMessageByID(ctx, asstID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	art, _ := msg.Metadata["artifact"].(map[string]any)
+	if art == nil {
+		t.Fatalf("artifact missing: %#v", msg.Metadata)
+	}
+	sum := sha256.Sum256(body)
+	want := hex.EncodeToString(sum[:])
+	if art["sha256"] != want {
+		t.Fatalf("sha256=%v want %s", art["sha256"], want)
+	}
+}
@@ -645,7 +645,7 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client
 		TargetMarket:    dto.TargetMarket,
 		Fields:          dto.Fields,
 		Materials:       dto.Materials,
-		Backend:         dto.Backend,
+		Backend:         resolveTaskBackend(s, dto.Backend),
 		AutoRepair:      true, // 产品契约：受控小队任务服务端启用修复，不依赖浏览器字段
 		SessionID:       sessionID,
 		UserID:          dto.UserID,
@@ -664,6 +664,19 @@ func (s *Server) buildBizdevTaskRequest(dto TaskRunRequestDTO, sessionID, client
 	}
 }

+func resolveTaskBackend(s *Server, requested string) string {
+	configured := ""
+	if s != nil && s.cfg != nil {
+		configured = s.cfg.AgentBackend
+	}
+	resolved, err := taskrun.ResolveBackend(requested, configured)
+	if err != nil {
+		fallback, _ := taskrun.ResolveBackend("", configured)
+		return fallback
+	}
+	return resolved
+}
+
 func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clientTurnID, runID string) bizdev.StageRunRequest {
 	return bizdev.StageRunRequest{
 		Stage:           bizdev.MaturityStage(dto.Stage),
@@ -674,7 +687,7 @@ func (s *Server) buildBizdevStageRequest(dto TaskRunRequestDTO, sessionID, clien
 		TargetMarket:    dto.TargetMarket,
 		Fields:          dto.Fields,
 		Materials:       dto.Materials,
-		Backend:         dto.Backend,
+		Backend:         resolveTaskBackend(s, dto.Backend),
 		AutoRepair:      true, // 产品契约：阶段任务同样启用修复预算
 		SessionID:       sessionID,
 		UserID:          dto.UserID,
@@ -0,0 +1,64 @@
+package taskrun
+
+import (
+	"errors"
+	"strings"
+)
+
+var (
+	ErrUnauthorizedBackend = errors.New("taskrun: unauthorized backend")
+	ErrSensitivePayload    = errors.New("taskrun: sensitive input rejected")
+)
+
+// Allowed backends are resolved by server policy. Callers cannot inject arbitrary strings.
+var allowedBackends = map[string]struct{}{
+	"stub":     {},
+	"pi-agent": {},
+	"cursor":   {},
+}
+
+// ResolveBackend maps a caller hint onto the server allowlist.
+// An empty requested value falls back to configured. Unauthorized values are rejected.
+func ResolveBackend(requested, configured string) (string, error) {
+	requested = strings.TrimSpace(requested)
+	configured = strings.TrimSpace(configured)
+	if requested == "" {
+		if configured == "" {
+			return "stub", nil
+		}
+		if _, ok := allowedBackends[configured]; !ok {
+			return "", ErrUnauthorizedBackend
+		}
+		return configured, nil
+	}
+	if _, ok := allowedBackends[requested]; !ok {
+		return "", ErrUnauthorizedBackend
+	}
+	return requested, nil
+}
+
+func StripSecrets(meta map[string]any) map[string]any {
+	if meta == nil {
+		return map[string]any{}
+	}
+	out := make(map[string]any, len(meta))
+	for k, v := range meta {
+		lk := strings.ToLower(strings.TrimSpace(k))
+		if lk == "authorization" || lk == "api_key" || lk == "token" || lk == "password" ||
+			strings.Contains(lk, "secret") || strings.Contains(lk, "credential") {
+			continue
+		}
+		out[k] = v
+	}
+	return out
+}
+
+func rejectSensitiveText(parts ...string) error {
+	for _, p := range parts {
+		low := strings.ToLower(p)
+		if strings.Contains(low, "api_key=") || strings.Contains(low, "authorization: bearer") {
+			return ErrSensitivePayload
+		}
+	}
+	return nil
+}
@@ -0,0 +1,561 @@
+package taskrun
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/artifactvault"
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/events"
+	"chat-agent-system/internal/projectworkflow"
+	"chat-agent-system/internal/workspace"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const (
+	StatusPending                   = "pending"
+	StatusRunning                   = "running"
+	StatusFinalizing                = "finalizing"
+	StatusCompleted                 = "completed"
+	StatusFailed                    = "failed"
+	StatusCancelled                 = "cancelled"
+	StatusFinalizingFailed          = "finalizing_failed"
+	StatusQualityFailed             = "quality_failed"
+	StatusInterrupted               = "interrupted"
+	StatusPendingAuthorConfirmation = "pending_author_confirmation"
+	StatusBlockedPendingGovernance  = "blocked_pending_governance"
+
+	EventCompleted = "taskrun.completed"
+	EventCancelled = "taskrun.cancelled"
+)
+
+var (
+	ErrTerminalConflict = errors.New("taskrun: terminal status already set")
+	ErrCleanupBlocked   = errors.New("taskrun: cleanup blocked until verified receipts match the frozen manifest")
+)
+
+var openStatuses = map[string]struct{}{
+	StatusPending:                  {},
+	StatusRunning:                  {},
+	StatusFinalizing:               {},
+	StatusInterrupted:              {},
+	StatusBlockedPendingGovernance: {},
+}
+
+var terminalStatuses = map[string]struct{}{
+	StatusCompleted:                 {},
+	StatusFailed:                    {},
+	StatusCancelled:                 {},
+	StatusQualityFailed:             {},
+	StatusPendingAuthorConfirmation: {},
+	StatusFinalizingFailed:          {},
+}
+
+// Commands is the unique finalize/cancel command. Server is a thin adapter.
+type Commands struct {
+	UoW             uow.UnitOfWork
+	Ctx             *ctxmgr.ContextManager
+	PW              *projectworkflow.Service
+	Fav             *artifactvault.Ingest
+	Env             string
+	StorageIdentity string
+
+	mu  sync.Mutex
+	mem map[string]*FinalizeResult
+}
+
+// FinalizeResult is the durable outcome of a terminal command.
+type FinalizeResult struct {
+	Status   string
+	Manifest *RunManifest
+	Applied  bool
+}
+
+type VerifiedReceipt struct {
+	LogicalID       string
+	Digest          string
+	StorageIdentity string
+	ArtifactUUID    string
+	Verified        bool
+}
+
+func (c *Commands) Finalize(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult) (string, error) {
+	if record == nil || result == nil {
+		return StatusFailed, fmt.Errorf("finalize: missing record or result")
+	}
+	status := result.TerminalStatus()
+	if _, err := ResolveBackend(result.BackendUsed, ""); err != nil && strings.TrimSpace(result.BackendUsed) != "" {
+		return StatusFailed, err
+	}
+	out, err := c.apply(ctx, record, result, status)
+	if err != nil {
+		if errors.Is(err, ErrTerminalConflict) && out != nil && out.Status != "" {
+			return out.Status, nil
+		}
+		return StatusFinalizingFailed, err
+	}
+	return out.Status, nil
+}
+
+func (c *Commands) Cancel(ctx context.Context, record *RunRecord) (string, error) {
+	if record == nil {
+		return StatusCancelled, fmt.Errorf("cancel: missing record")
+	}
+	result := &bizdev.TaskRunResult{Success: false, ErrorReason: "cancelled", TaskKey: record.TaskKey, WorkDir: record.WorkspaceID}
+	out, err := c.apply(ctx, record, result, StatusCancelled)
+	if err != nil {
+		if errors.Is(err, ErrTerminalConflict) && out != nil {
+			return out.Status, nil
+		}
+		return StatusCancelled, err
+	}
+	return out.Status, nil
+}
+
+func (c *Commands) apply(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) {
+	if c != nil && c.UoW != nil {
+		var out *FinalizeResult
+		err := c.UoW.Within(ctx, func(scope uow.TxScope) error {
+			ctx = uow.WithScope(ctx, scope)
+			got, err := c.applyLocked(ctx, record, result, want)
+			out = got
+			return err
+		})
+		return out, err
+	}
+	return c.applyLocked(ctx, record, result, want)
+}
+
+func (c *Commands) applyLocked(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, want string) (*FinalizeResult, error) {
+	if err := rejectSensitiveText(result.MarkdownBody, result.ErrorReason); err != nil {
+		return nil, err
+	}
+	workspaceID := workspace.OpaqueID(result.WorkDir, record.WorkspaceID)
+	manifest, err := buildRunManifest(record, result, workspaceID, c.storageID(), want)
+	if err != nil {
+		return nil, err
+	}
+
+	existing, applied, err := c.casTerminal(ctx, record.RunID, want)
+	if err != nil {
+		return existing, err
+	}
+	if !applied {
+		if existing != nil {
+			return existing, nil
+		}
+		return &FinalizeResult{Status: want, Manifest: manifest, Applied: false}, nil
+	}
+
+	if err := c.persistManifest(ctx, record.RunID, manifest); err != nil {
+		return nil, err
+	}
+	if err := c.projectWorkflow(ctx, record, result, want); err != nil {
+		return nil, err
+	}
+	if err := c.finalizeAssistant(ctx, record, result, want, workspaceID, manifest); err != nil {
+		return nil, err
+	}
+	if err := c.appendTerminalEvent(ctx, record, result, want, manifest); err != nil {
+		return nil, err
+	}
+	if err := c.stageControlledDoc(ctx, record, result, want, manifest); err != nil {
+		return nil, err
+	}
+
+	out := &FinalizeResult{Status: want, Manifest: manifest, Applied: true}
+	c.remember(record.RunID, out)
+	return out, nil
+}
+
+func (c *Commands) storageID() string {
+	if c != nil && strings.TrimSpace(c.StorageIdentity) != "" {
+		return c.StorageIdentity
+	}
+	env := "dev"
+	if c != nil && c.Env != "" {
+		env = c.Env
+	}
+	return "env:" + env
+}
+
+func (c *Commands) casTerminal(ctx context.Context, runID, want string) (*FinalizeResult, bool, error) {
+	if scope, ok := uow.ScopeFrom(ctx); ok {
+		tx, err := uow.AsPG(scope)
+		if err != nil {
+			return nil, false, err
+		}
+		return casPG(ctx, tx, runID, want)
+	}
+	return c.casMemory(runID, want)
+}
+
+func casPG(ctx context.Context, tx pgx.Tx, runID, want string) (*FinalizeResult, bool, error) {
+	var current string
+	err := tx.QueryRow(ctx, `SELECT status FROM task_runs WHERE run_id=$1 FOR UPDATE`, runID).Scan(&current)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, true, nil
+	}
+	if err != nil {
+		return nil, false, err
+	}
+	if current == want {
+		man, _ := loadManifestTX(ctx, tx, runID)
+		return &FinalizeResult{Status: current, Manifest: man, Applied: false}, false, nil
+	}
+	if _, open := openStatuses[current]; !open {
+		man, _ := loadManifestTX(ctx, tx, runID)
+		return &FinalizeResult{Status: current, Manifest: man, Applied: false}, false, ErrTerminalConflict
+	}
+	tag, err := tx.Exec(ctx, `
+		UPDATE task_runs SET status=$2, updated_at=now()
+		WHERE run_id=$1 AND status=$3
+	`, runID, want, current)
+	if err != nil {
+		return nil, false, err
+	}
+	if tag.RowsAffected() != 1 {
+		return nil, false, ErrTerminalConflict
+	}
+	return nil, true, nil
+}
+
+func (c *Commands) casMemory(runID, want string) (*FinalizeResult, bool, error) {
+	if c == nil {
+		return nil, true, nil
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.mem == nil {
+		c.mem = map[string]*FinalizeResult{}
+	}
+	if snap, ok := c.mem[runID]; ok {
+		if snap.Status == want {
+			cpy := *snap
+			cpy.Applied = false
+			return &cpy, false, nil
+		}
+		cpy := *snap
+		cpy.Applied = false
+		return &cpy, false, ErrTerminalConflict
+	}
+	c.mem[runID] = &FinalizeResult{Status: want, Applied: true}
+	return nil, true, nil
+}
+
+func (c *Commands) remember(runID string, out *FinalizeResult) {
+	if c == nil || out == nil || runID == "" {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.mem == nil {
+		c.mem = map[string]*FinalizeResult{}
+	}
+	cpy := *out
+	c.mem[runID] = &cpy
+}
+
+func (c *Commands) persistManifest(ctx context.Context, runID string, man *RunManifest) error {
+	scope, ok := uow.ScopeFrom(ctx)
+	if !ok {
+		return nil
+	}
+	tx, err := uow.AsPG(scope)
+	if err != nil {
+		return err
+	}
+	members, err := json.Marshal(man.Members)
+	if err != nil {
+		return err
+	}
+	var existingSHA string
+	err = tx.QueryRow(ctx, `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, runID).Scan(&existingSHA)
+	if err == nil {
+		if strings.TrimSpace(existingSHA) != man.SHA256 {
+			return uow.ErrDigestConflict
+		}
+		return nil
+	}
+	if !errors.Is(err, pgx.ErrNoRows) {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+		INSERT INTO task_run_manifests (run_id, canonical_bytes, sha256, workspace_id, storage_identity, members)
+		VALUES ($1,$2,$3,$4,$5,$6::jsonb)
+		ON CONFLICT (run_id) DO NOTHING
+	`, runID, man.CanonicalBytes, man.SHA256, man.WorkspaceID, man.StorageIdentity, members)
+	if err != nil {
+		// Unit tests without a durable task_runs row skip FK persistence.
+		if strings.Contains(strings.ToLower(err.Error()), "foreign key") {
+			return nil
+		}
+		return err
+	}
+	return nil
+}
+
+func loadManifestTX(ctx context.Context, tx pgx.Tx, runID string) (*RunManifest, error) {
+	var man RunManifest
+	var members []byte
+	err := tx.QueryRow(ctx, `
+		SELECT run_id, canonical_bytes, sha256, workspace_id, storage_identity, members
+		FROM task_run_manifests WHERE run_id=$1
+	`, runID).Scan(&man.RunID, &man.CanonicalBytes, &man.SHA256, &man.WorkspaceID, &man.StorageIdentity, &members)
+	if err != nil {
+		return nil, err
+	}
+	_ = json.Unmarshal(members, &man.Members)
+	man.SchemaVersion = ManifestSchemaVersion
+	return &man, nil
+}
+
+func (c *Commands) projectWorkflow(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string) error {
+	if c == nil || c.PW == nil || record.ProjectID <= 0 || record.RunID == "" {
+		return nil
+	}
+	docID := ""
+	title := result.ArtifactTitle
+	if result.DocRecord != nil {
+		docID = result.DocRecord.DocID
+	}
+	if docID == "" && record.TaskKey != "" {
+		docID = fmt.Sprintf("doc_proj%d_%s", record.ProjectID, record.TaskKey)
+	}
+	if title == "" && record.TaskKey != "" {
+		title = record.TaskKey
+	}
+	if err := c.PW.UpdateWorkflowRun(ctx, record.RunID, status, docID, title, ""); err != nil {
+		if strings.Contains(strings.ToLower(err.Error()), "not found") {
+			return nil
+		}
+		return err
+	}
+	return nil
+}
+
+func (c *Commands) finalizeAssistant(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status, workspaceID string, man *RunManifest) error {
+	if c == nil || c.Ctx == nil || record.AssistantMsgID <= 0 {
+		return nil
+	}
+	summary := result.CompletionSummary()
+	if status == StatusCancelled {
+		summary = "任务已取消。"
+		if result.ErrorReason != "" && result.ErrorReason != "cancelled" {
+			summary += "\n\n" + result.ErrorReason
+		}
+	}
+	meta := StripSecrets(map[string]any{
+		"run_id":       record.RunID,
+		"workspace_id": workspaceID,
+		"task_status":  status,
+		"task_key":     record.TaskKey,
+	})
+	if record.ProjectID > 0 {
+		meta["project_id"] = record.ProjectID
+	}
+	if man != nil {
+		meta["run_manifest_sha256"] = man.SHA256
+		meta["storage_identity"] = man.StorageIdentity
+	}
+	if art := result.ArtifactMetadata(); art != nil {
+		if workspaceID != "" {
+			art["workspace_id"] = workspaceID
+		}
+		if result.DocRecord != nil && result.DocRecord.ContentSHA256 != "" {
+			art["sha256"] = result.DocRecord.ContentSHA256
+		}
+		meta["artifact"] = StripSecrets(art)
+	} else if result.DocRecord != nil && result.DocRecord.DocID != "" {
+		artifact := map[string]any{
+			"doc_id":       result.DocRecord.DocID,
+			"title":        result.ArtifactTitle,
+			"task_key":     record.TaskKey,
+			"workspace_id": workspaceID,
+		}
+		if result.DocRecord.ContentSHA256 != "" {
+			artifact["sha256"] = result.DocRecord.ContentSHA256
+		}
+		if result.QualityReport != nil {
+			artifact["quality_passed"] = result.QualityReport.Passed
+			artifact["quality_score"] = result.QualityReport.Score
+			if len(result.QualityReport.Violations) > 0 {
+				artifact["quality_violations"] = result.QualityReport.Violations
+			}
+		}
+		if result.AwaitsAgentConfirmation() {
+			artifact["pending_agent_candidate_id"] = result.PendingAgentCandidateID
+			if result.PendingChangeReportID > 0 {
+				artifact["pending_change_report_id"] = result.PendingChangeReportID
+			}
+			if result.PendingRiskLevel != "" {
+				artifact["pending_risk_level"] = result.PendingRiskLevel
+			}
+			artifact["awaiting_author_confirmation"] = true
+		}
+		meta["artifact"] = artifact
+	}
+	return c.Ctx.FinalizeAssistantTaskMessage(ctx, record.AssistantMsgID, summary, meta, status)
+}
+
+func (c *Commands) appendTerminalEvent(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error {
+	if _, ok := uow.ScopeFrom(ctx); !ok {
+		return nil
+	}
+	name := EventCompleted
+	if status == StatusCancelled {
+		name = EventCancelled
+	}
+	body, err := json.Marshal(map[string]any{
+		"run_id":           record.RunID,
+		"status":           status,
+		"workspace_id":     man.WorkspaceID,
+		"storage_identity": man.StorageIdentity,
+		"manifest_sha256":  man.SHA256,
+		"task_key":         record.TaskKey,
+		"doc_id":           docIDOf(result),
+	})
+	if err != nil {
+		return err
+	}
+	envName := "dev"
+	if c != nil && c.Env != "" {
+		envName = c.Env
+	}
+	env := events.Envelope{
+		EventID:          fmt.Sprintf("taskrun:%s:%s", record.RunID, status),
+		EventName:        name,
+		SchemaVersion:    events.EnvelopeSchema,
+		Environment:      envName,
+		ProjectID:        record.ProjectID,
+		AggregateType:    "task_run",
+		AggregateID:      record.RunID,
+		AggregateVersion: 1,
+		ActorRef:         fmt.Sprintf("user:%d", record.UserID),
+		IdempotencyKey:   fmt.Sprintf("finalize:%s:%s", record.RunID, status),
+		PayloadDigest:    events.Digest(body),
+		Producer:         "taskrun",
+		OccurredAt:       time.Now().UTC(),
+		Payload:          body,
+	}
+	return (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"})
+}
+
+func (c *Commands) stageControlledDoc(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error {
+	if c == nil || c.Fav == nil {
+		return nil
+	}
+	if status != StatusCompleted {
+		return nil
+	}
+	if result.DocRecord == nil || result.DocRecord.DocID == "" {
+		return nil
+	}
+	payload := []byte(result.MarkdownBody)
+	if len(payload) == 0 {
+		payload = man.CanonicalBytes
+	}
+	if len(payload) == 0 {
+		return nil
+	}
+	snapID := int64(result.DocRecord.Version)
+	if snapID <= 0 {
+		snapID = 1
+	}
+	eventID := fmt.Sprintf("taskrun:%s:%s", record.RunID, status)
+	return c.Fav.StageAgentCommit(ctx, record.ProjectID, result.DocRecord.DocID, eventID, snapID, payload)
+}
+
+func buildRunManifest(record *RunRecord, result *bizdev.TaskRunResult, workspaceID, storageID, status string) (*RunManifest, error) {
+	man := &RunManifest{
+		SchemaVersion:   ManifestSchemaVersion,
+		RunID:           record.RunID,
+		WorkspaceID:     workspaceID,
+		StorageIdentity: storageID,
+	}
+	if result.DocRecord != nil && result.DocRecord.DocID != "" {
+		digest := strings.ToLower(strings.TrimSpace(result.DocRecord.ContentSHA256))
+		if digest == "" && result.MarkdownBody != "" {
+			sum := sha256Hex([]byte(result.MarkdownBody))
+			digest = sum
+		}
+		if digest != "" {
+			man.Members = append(man.Members, ManifestMember{
+				LogicalID: result.DocRecord.DocID,
+				Digest:    digest,
+				Kind:      MemberKindDocument,
+			})
+		}
+	}
+	if record.AssistantMsgID > 0 {
+		digest := sha256Hex([]byte(result.CompletionSummary()))
+		if status == StatusCancelled {
+			digest = sha256Hex([]byte("任务已取消。"))
+		}
+		man.Members = append(man.Members, ManifestMember{
+			LogicalID: fmt.Sprintf("assistant:%d", record.AssistantMsgID),
+			Digest:    digest,
+			Kind:      MemberKindAssistantTurn,
+		})
+	}
+	if err := man.Freeze(); err != nil {
+		return nil, err
+	}
+	return man, nil
+}
+
+func sha256Hex(b []byte) string {
+	sum := sha256.Sum256(b)
+	return hex.EncodeToString(sum[:])
+}
+
+func docIDOf(result *bizdev.TaskRunResult) string {
+	if result != nil && result.DocRecord != nil {
+		return result.DocRecord.DocID
+	}
+	return ""
+}
+
+func IsTerminalStatus(status string) bool {
+	_, ok := terminalStatuses[status]
+	return ok
+}
+
+func IsOpenStatus(status string) bool {
+	_, ok := openStatuses[status]
+	return ok
+}
+
+// AssertCleanupAllowed requires a frozen manifest and a verified receipt for every retained member.
+func (m *RunManifest) AssertCleanupAllowed(receipts []VerifiedReceipt) error {
+	if m == nil || m.SHA256 == "" || m.StorageIdentity == "" {
+		return ErrCleanupBlocked
+	}
+	byLogical := map[string]VerifiedReceipt{}
+	for _, r := range receipts {
+		byLogical[r.LogicalID] = r
+	}
+	for _, mem := range m.Members {
+		if !memberNeedsRetention(mem.Kind) {
+			continue
+		}
+		r, ok := byLogical[mem.LogicalID]
+		if !ok || !r.Verified {
+			return ErrCleanupBlocked
+		}
+		if r.Digest != mem.Digest || r.StorageIdentity != m.StorageIdentity {
+			return ErrCleanupBlocked
+		}
+	}
+	return nil
+}
@@ -0,0 +1,363 @@
+package taskrun_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/artifactvault"
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/stream"
+	"chat-agent-system/internal/taskrun"
+	"chat-agent-system/internal/workspace"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func init() {
+	if _, err := os.Stat("deploy/postgres/goose"); err == nil {
+		return
+	}
+	for _, cand := range []string{"../..", "../../.."} {
+		if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
+			_ = os.Chdir(cand)
+			return
+		}
+	}
+}
+
+var (
+	isoCleanup func()
+)
+
+func TestMain(m *testing.M) {
+	code := m.Run()
+	if isoCleanup != nil {
+		isoCleanup()
+	}
+	os.Exit(code)
+}
+
+func rewriteDatabaseName(dsn, name string) string {
+	if u, err := url.Parse(dsn); err == nil && u.Scheme != "" {
+		u.Path = "/" + name
+		return u.String()
+	}
+	return dsn
+}
+
+var isoOnce sync.Once
+var isoPoolA, isoPoolB *pgxpool.Pool
+var isoErr error
+
+func openIsolatedM3(t *testing.T) (*pgxpool.Pool, *pgxpool.Pool) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("U8 PG races are not part of -short")
+	}
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	db.RefuseSharedEnvDSN(t, dsn)
+
+	isoOnce.Do(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+		defer cancel()
+		adminDSN := rewriteDatabaseName(dsn, "postgres")
+		admin, err := pgxpool.New(ctx, adminDSN)
+		if err != nil {
+			isoErr = err
+			return
+		}
+		const isolated = "agent_rel_m3_u8"
+		_, _ = admin.Exec(ctx, `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated)
+		_, _ = admin.Exec(ctx, `DROP DATABASE IF EXISTS `+isolated)
+		if _, err := admin.Exec(ctx, `CREATE DATABASE `+isolated); err != nil {
+			admin.Close()
+			isoErr = err
+			return
+		}
+		isoDSN := rewriteDatabaseName(dsn, isolated)
+		poolA, err := pgxpool.New(ctx, isoDSN)
+		if err != nil {
+			admin.Close()
+			isoErr = err
+			return
+		}
+		if err := db.RunMigrations(ctx, poolA, db.ResolveMigrationsDir()); err != nil {
+			poolA.Close()
+			admin.Close()
+			isoErr = err
+			return
+		}
+		poolB, err := pgxpool.New(ctx, isoDSN)
+		if err != nil {
+			poolA.Close()
+			admin.Close()
+			isoErr = err
+			return
+		}
+		isoPoolA, isoPoolB = poolA, poolB
+		isoCleanup = func() {
+			poolA.Close()
+			poolB.Close()
+			_, _ = admin.Exec(context.Background(), `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname=$1 AND pid <> pg_backend_pid()`, isolated)
+			_, _ = admin.Exec(context.Background(), `DROP DATABASE IF EXISTS `+isolated)
+			admin.Close()
+		}
+	})
+	if isoErr != nil {
+		t.Fatalf("isolated db: %v", isoErr)
+	}
+	return isoPoolA, isoPoolB
+}
+
+func seedUserAndProject(t *testing.T, pool *pgxpool.Pool) (userID, projectID int64) {
+	t.Helper()
+	ctx := context.Background()
+	err := pool.QueryRow(ctx, `SELECT id FROM users ORDER BY id LIMIT 1`).Scan(&userID)
+	if err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+	err = pool.QueryRow(ctx, `SELECT id FROM projects ORDER BY id LIMIT 1`).Scan(&projectID)
+	if err != nil {
+		t.Fatalf("seed project: %v", err)
+	}
+	return userID, projectID
+}
+
+func insertRunning(t *testing.T, pool *pgxpool.Pool, rec *taskrun.RunRecord) {
+	t.Helper()
+	_, err := pool.Exec(context.Background(), `
+		INSERT INTO task_runs (
+			run_id, session_id, project_id, user_id, client_turn_id, task_key, status,
+			workspace_id, assistant_message_id, created_at, updated_at
+		) VALUES ($1,$2,$3,$4,$5,$6,'running',$7,$8,now(),now())
+	`, rec.RunID, rec.SessionID, rec.ProjectID, rec.UserID, rec.ClientTurnID, rec.TaskKey, rec.WorkspaceID, rec.AssistantMsgID)
+	if err != nil {
+		t.Fatalf("insert run: %v", err)
+	}
+}
+
+func TestPG_DuplicateFinalizeIdempotent(t *testing.T) {
+	poolA, poolB := openIsolatedM3(t)
+	userID, projectID := seedUserAndProject(t, poolA)
+	ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{})
+	sessionID := "sess_u8_dup"
+	asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := &taskrun.RunRecord{
+		RunID: "run_u8_dup", SessionID: sessionID, ProjectID: projectID, UserID: userID,
+		TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: "ws_u8_dup",
+	}
+	insertRunning(t, poolA, rec)
+	body := []byte("# committed intake body\n")
+	res := &bizdev.TaskRunResult{
+		Success: true, TaskKey: "intake", ArtifactTitle: "用户初始需求收集表", MarkdownBody: string(body),
+		DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_dup", ContentSHA256: artifactvault.SHA256Hex(body), Version: 1, ProjectID: projectID},
+	}
+	cmdsA := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"}
+	cmdsB := &taskrun.Commands{UoW: uow.NewPG(poolB), Ctx: ctxmgr.NewContextManager(poolB, nil, ctxmgr.TimeDecayConfig{}), Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"}
+	st1, err := cmdsA.Finalize(context.Background(), rec, res)
+	if err != nil {
+		t.Fatal(err)
+	}
+	st2, err := cmdsB.Finalize(context.Background(), rec, res)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st1 != "completed" || st2 != "completed" {
+		t.Fatalf("status %s %s", st1, st2)
+	}
+	var msgs, mans, envs, stgs int
+	_ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM chat_messages WHERE session_id=$1 AND role='assistant'`, sessionID).Scan(&msgs)
+	_ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&mans)
+	_ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1`, rec.RunID).Scan(&envs)
+	_ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM artifact_payload_staging WHERE source_event_id=$1`, "taskrun:"+rec.RunID+":completed").Scan(&stgs)
+	if msgs != 1 || mans != 1 || envs != 1 || stgs != 1 {
+		t.Fatalf("counts messages=%d manifests=%d envelopes=%d staging=%d", msgs, mans, envs, stgs)
+	}
+	var sha1, sha2 string
+	_ = poolA.QueryRow(context.Background(), `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&sha1)
+	_ = poolB.QueryRow(context.Background(), `SELECT sha256 FROM task_run_manifests WHERE run_id=$1`, rec.RunID).Scan(&sha2)
+	if sha1 == "" || sha1 != sha2 {
+		t.Fatalf("manifest sha %q %q", sha1, sha2)
+	}
+}
+
+func TestPG_FinalizeCancelRaceOneTerminal(t *testing.T) {
+	poolA, poolB := openIsolatedM3(t)
+	userID, projectID := seedUserAndProject(t, poolA)
+	ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{})
+	sessionID := "sess_u8_race"
+	asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := &taskrun.RunRecord{
+		RunID: "run_u8_race", SessionID: sessionID, ProjectID: projectID, UserID: userID,
+		TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: "ws_u8_race",
+	}
+	insertRunning(t, poolA, rec)
+	body := []byte("# race body\n")
+	res := &bizdev.TaskRunResult{
+		Success: true, TaskKey: "intake", MarkdownBody: string(body),
+		DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_race", ContentSHA256: artifactvault.SHA256Hex(body), Version: 1},
+	}
+	cmdsA := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"}
+	cmdsB := &taskrun.Commands{UoW: uow.NewPG(poolB), Ctx: ctxmgr.NewContextManager(poolB, nil, ctxmgr.TimeDecayConfig{}), Env: "dev", StorageIdentity: "env:dev"}
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		_, _ = cmdsA.Finalize(context.Background(), rec, res)
+	}()
+	go func() {
+		defer wg.Done()
+		_, _ = cmdsB.Cancel(context.Background(), rec)
+	}()
+	wg.Wait()
+	var status string
+	if err := poolA.QueryRow(context.Background(), `SELECT status FROM task_runs WHERE run_id=$1`, rec.RunID).Scan(&status); err != nil {
+		t.Fatal(err)
+	}
+	if status != "completed" && status != "cancelled" {
+		t.Fatalf("illegal terminal %q", status)
+	}
+	var n int
+	_ = poolA.QueryRow(context.Background(), `SELECT COUNT(*) FROM chat_messages WHERE session_id=$1 AND role='assistant'`, sessionID).Scan(&n)
+	if n != 1 {
+		t.Fatalf("assistant turns=%d", n)
+	}
+}
+
+func TestPG_SSEDoneNotEmittedWhenCommitFails(t *testing.T) {
+	poolA, _ := openIsolatedM3(t)
+	userID, projectID := seedUserAndProject(t, poolA)
+	ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{})
+	sessionID := "sess_u8_sse"
+	asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := &taskrun.RunRecord{
+		RunID: "run_u8_sse", SessionID: sessionID, ProjectID: projectID, UserID: userID,
+		TaskKey: "intake", AssistantMsgID: asstID, Status: "pending",
+	}
+	unit := uow.NewPG(poolA)
+	unit.BeforeCommit = func(uow.TxScope) error { return errors.New("inject commit fail") }
+	cmds := &taskrun.Commands{UoW: unit, Ctx: ctxMgr, Env: "dev", StorageIdentity: "env:dev"}
+	mgr := taskrun.NewManager(nil, ctxMgr)
+	mgr.SetStore(taskrun.NewPGStore(poolA))
+	mgr.SetCommands(cmds)
+	mgr.SetExecutor(func(context.Context, string, bizdev.TaskRunRequest, *stream.Streamer) (*bizdev.TaskRunResult, error) {
+		return &bizdev.TaskRunResult{Success: true, MarkdownBody: "# sse\n"}, nil
+	}, nil, nil)
+	if _, err := mgr.CreateAndStart(context.Background(), rec.RunID, rec, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, ""); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(8 * time.Second)
+	for time.Now().Before(deadline) {
+		got, err := mgr.GetRun(rec.RunID)
+		if err == nil && got.Status == "finalizing_failed" {
+			evs, listErr := taskrun.NewPGStore(poolA).ListEvents(context.Background(), rec.RunID, 0)
+			if listErr != nil {
+				t.Fatal(listErr)
+			}
+			for _, ev := range evs {
+				if ev.Event.Type == stream.EventDone {
+					t.Fatal("done emitted on failed commit")
+				}
+			}
+			msg, _ := ctxMgr.GetMessageByID(context.Background(), asstID)
+			if msg != nil && msg.Content != "任务执行中…" {
+				t.Fatalf("placeholder mutated after rollback: %q", msg.Content)
+			}
+			return
+		}
+		time.Sleep(30 * time.Millisecond)
+	}
+	t.Fatal("expected finalizing_failed")
+}
+
+func TestPG_CleanupWithoutReceiptFailsAndStagingSurvives(t *testing.T) {
+	poolA, _ := openIsolatedM3(t)
+	userID, projectID := seedUserAndProject(t, poolA)
+	ctxMgr := ctxmgr.NewContextManager(poolA, nil, ctxmgr.TimeDecayConfig{})
+	sessionID := "sess_u8_clean"
+	asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), sessionID, userID, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"project_id": projectID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	dataRoot := t.TempDir()
+	cap := workspace.NewCapability(dataRoot, nil)
+	ref, err := cap.Issue(workspace.WorkspaceRef{ID: "ws_u8_clean", ProjectScope: projectID, RunID: "run_u8_clean"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	body := []byte("# vaulted body for cleanup\n")
+	if err := cap.WriteArtifact(context.Background(), ref, workspace.DeliverableKey, body); err != nil {
+		t.Fatal(err)
+	}
+	rec := &taskrun.RunRecord{
+		RunID: "run_u8_clean", SessionID: sessionID, ProjectID: projectID, UserID: userID,
+		TaskKey: "intake", AssistantMsgID: asstID, WorkspaceID: ref.ID,
+	}
+	insertRunning(t, poolA, rec)
+	res := &bizdev.TaskRunResult{
+		Success: true, TaskKey: "intake", MarkdownBody: string(body), WorkDir: ref.ID,
+		DocRecord: &docgov.DocumentRecord{DocID: "doc_u8_clean", ContentSHA256: artifactvault.SHA256Hex(body), Version: 2},
+	}
+	cmds := &taskrun.Commands{UoW: uow.NewPG(poolA), Ctx: ctxMgr, Fav: &artifactvault.Ingest{Env: "dev"}, Env: "dev", StorageIdentity: "env:dev"}
+	if _, err := cmds.Finalize(context.Background(), rec, res); err != nil {
+		t.Fatal(err)
+	}
+	var manSHA, storageID string
+	var members []byte
+	if err := poolA.QueryRow(context.Background(), `SELECT sha256, storage_identity, members FROM task_run_manifests WHERE run_id=$1`, rec.RunID).
+		Scan(&manSHA, &storageID, &members); err != nil {
+		t.Fatal(err)
+	}
+	var parsed []taskrun.ManifestMember
+	if err := json.Unmarshal(members, &parsed); err != nil {
+		t.Fatal(err)
+	}
+	frozen := workspace.FrozenManifest{SHA256: manSHA, StorageIdentity: storageID, WorkspaceID: ref.ID}
+	for _, m := range parsed {
+		frozen.Members = append(frozen.Members, workspace.FrozenMember{LogicalID: m.LogicalID, Digest: m.Digest, Kind: m.Kind})
+	}
+	if err := cap.Cleanup(context.Background(), ref, frozen, nil); err == nil {
+		t.Fatal("cleanup without receipts must fail")
+	}
+	if _, err := os.Stat(filepath.Join(dataRoot, "workspaces", ref.ID, workspace.DeliverableKey)); err != nil {
+		t.Fatalf("workspace removed without receipts: %v", err)
+	}
+	var receipts []workspace.RetentionReceipt
+	for _, m := range parsed {
+		receipts = append(receipts, workspace.RetentionReceipt{
+			LogicalID: m.LogicalID, Digest: m.Digest, StorageIdentity: storageID, Verified: true,
+		})
+	}
+	if err := cap.Cleanup(context.Background(), ref, frozen, receipts); err != nil {
+		t.Fatal(err)
+	}
+	eventID := "taskrun:" + rec.RunID + ":completed"
+	reg := artifactvault.NewPGRegistry(poolA)
+	st, err := reg.LoadStaging(context.Background(), eventID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if artifactvault.SHA256Hex(st.Payload) != artifactvault.SHA256Hex(body) {
+		t.Fatalf("staging sha drifted")
+	}
+}
@@ -0,0 +1,131 @@
+package taskrun_test
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+
+	"chat-agent-system/internal/bizdev"
+	"chat-agent-system/internal/ctxmgr"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/taskrun"
+)
+
+func TestCommands_DuplicateFinalizeMemory(t *testing.T) {
+	ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{})
+	asstID, err := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-dup", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", map[string]any{"task_status": "running"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"}
+	rec := &taskrun.RunRecord{RunID: "run_dup_mem", SessionID: "sess-dup", TaskKey: "intake", AssistantMsgID: asstID}
+	res := &bizdev.TaskRunResult{
+		Success: true, TaskKey: "intake", ArtifactTitle: "用户初始需求收集表", MarkdownBody: "# body\n",
+		DocRecord: &docgov.DocumentRecord{DocID: "doc_dup", ContentSHA256: strings.Repeat("a", 64), Version: 1},
+	}
+	st1, err := cmds.Finalize(context.Background(), rec, res)
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg, err := ctxMgr.GetMessageByID(context.Background(), asstID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg.Content == "任务执行中…" {
+		t.Fatal("placeholder not replaced")
+	}
+	st2, err := cmds.Finalize(context.Background(), rec, res)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st1 != "completed" || st2 != "completed" {
+		t.Fatalf("status %s %s", st1, st2)
+	}
+	msg2, _ := ctxMgr.GetMessageByID(context.Background(), asstID)
+	if msg2.ID != msg.ID || msg2.Content != msg.Content {
+		t.Fatal("duplicate finalize mutated the assistant turn")
+	}
+}
+
+func TestCommands_CancelWinsOverFinalizeMemory(t *testing.T) {
+	ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{})
+	asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-race", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil)
+	cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"}
+	rec := &taskrun.RunRecord{RunID: "run_race_mem", SessionID: "sess-race", AssistantMsgID: asstID}
+	st, err := cmds.Cancel(context.Background(), rec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st != taskrun.StatusCancelled {
+		t.Fatalf("status=%s", st)
+	}
+	res := &bizdev.TaskRunResult{Success: true, MarkdownBody: "# later\n"}
+	st2, err := cmds.Finalize(context.Background(), rec, res)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st2 != taskrun.StatusCancelled {
+		t.Fatalf("finalize must adopt cancelled, got %s", st2)
+	}
+}
+
+func TestCommands_ConcurrentMemorySingleTerminal(t *testing.T) {
+	ctxMgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.TimeDecayConfig{})
+	asstID, _ := ctxMgr.RecordUserMessageWithID(context.Background(), "sess-conc", 0, "assistant", "任务执行中…", "TRIGGER_TASK", "", nil)
+	cmds := &taskrun.Commands{Ctx: ctxMgr, Env: "test", StorageIdentity: "env:test"}
+	rec := &taskrun.RunRecord{RunID: "run_conc_mem", SessionID: "sess-conc", AssistantMsgID: asstID}
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		_, _ = cmds.Finalize(context.Background(), rec, &bizdev.TaskRunResult{Success: true, MarkdownBody: "# ok\n"})
+	}()
+	go func() {
+		defer wg.Done()
+		_, _ = cmds.Cancel(context.Background(), rec)
+	}()
+	wg.Wait()
+	msg, err := ctxMgr.GetMessageByID(context.Background(), asstID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg.ID != asstID {
+		t.Fatalf("assistant id changed %d", msg.ID)
+	}
+}
+
+func TestRunManifest_CleanupGate(t *testing.T) {
+	man := &taskrun.RunManifest{
+		RunID: "run_clean", WorkspaceID: "ws", StorageIdentity: "env:dev",
+		Members: []taskrun.ManifestMember{{LogicalID: "doc_a", Digest: "abc", Kind: taskrun.MemberKindDocument}},
+	}
+	if err := man.Freeze(); err != nil {
+		t.Fatal(err)
+	}
+	if err := man.AssertCleanupAllowed(nil); err == nil {
+		t.Fatal("cleanup without receipts must fail")
+	}
+	receipts := []taskrun.VerifiedReceipt{{
+		LogicalID: "doc_a", Digest: "abc", StorageIdentity: "env:dev", Verified: true,
+	}}
+	if err := man.AssertCleanupAllowed(receipts); err != nil {
+		t.Fatal(err)
+	}
+	bad := []taskrun.VerifiedReceipt{{
+		LogicalID: "doc_a", Digest: "zzz", StorageIdentity: "env:dev", Verified: true,
+	}}
+	if err := man.AssertCleanupAllowed(bad); err == nil {
+		t.Fatal("digest mismatch must fail")
+	}
+}
+
+func TestCommands_RejectsSensitiveMarkdown(t *testing.T) {
+	cmds := &taskrun.Commands{Env: "test", StorageIdentity: "env:test"}
+	_, err := cmds.Finalize(context.Background(), &taskrun.RunRecord{RunID: "run_secret"}, &bizdev.TaskRunResult{
+		Success: true, MarkdownBody: "api_key=super-secret",
+	})
+	if err == nil {
+		t.Fatal("expected sensitive reject")
+	}
+}
@@ -36,6 +36,7 @@ type Manager struct {
 	resume   ResumeExecutor
 	stage    StageExecutor
 	finalize Finalizer
+	cmds     *Commands
 	pwSvc    *projectworkflow.Service
 	ctxMgr   *ctxmgr.ContextManager

@@ -129,6 +130,13 @@ func (m *Manager) SetFinalizer(f Finalizer) {
 	m.finalize = f
 }

+func (m *Manager) SetCommands(cmds *Commands) {
+	m.cmds = cmds
+	if cmds != nil {
+		m.finalize = cmds.Finalize
+	}
+}
+
 func (m *Manager) CreateAndStart(parent context.Context, runID string, record *RunRecord, taskReq bizdev.TaskRunRequest, stageReq bizdev.StageRunRequest, resumeWorkDir string) (*RunRecord, error) {
 	if record == nil {
 		record = &RunRecord{}
@@ -343,6 +351,9 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T

 	if execErr != nil {
 		if ctx.Err() == context.Canceled {
+			if m.cmds != nil {
+				_, _ = m.cmds.Cancel(context.Background(), ar.record)
+			}
 			info := stream.ClassifyError(fmt.Errorf("task cancelled"), "task_cancel", ar.record.RunID, ar.record.SessionID, workspaceID)
 			m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info})
 			m.finishRun(ar, "cancelled", workspaceID, execErr.Error(), nil, &info)
@@ -354,6 +365,18 @@ func (m *Manager) runWorker(ctx context.Context, ar *activeRun, taskReq bizdev.T
 		}
 		info := stream.ClassifyError(execErr, lastPhaseID, ar.record.RunID, ar.record.SessionID, classifyWS)
 		m.appendEvent(ar, stream.StreamEvent{Type: stream.EventError, ErrorInfo: &info})
+		if m.cmds != nil {
+			failRes := taskResult
+			if failRes == nil {
+				failRes = &bizdev.TaskRunResult{Success: false, ErrorReason: execErr.Error(), WorkDir: workspaceID}
+			} else {
+				failRes.Success = false
+				if failRes.ErrorReason == "" {
+					failRes.ErrorReason = execErr.Error()
+				}
+			}
+			_, _ = m.cmds.Finalize(context.Background(), ar.record, failRes)
+		}
 		m.finishRun(ar, "failed", workspaceID, execErr.Error(), taskResult, &info)
 		return
 	}
@@ -408,7 +431,7 @@ func (m *Manager) finishRun(ar *activeRun, status, workspaceID, errMsg string, r
 	m.persistRun(ar)

 	// 失败/取消路径：若 Finalize 未执行，仍更新 workflow_runs 与占位消息 metadata
-	if status == "failed" || status == "cancelled" || status == "finalizing_failed" {
+	if m.cmds == nil && (status == "failed" || status == "cancelled" || status == "finalizing_failed") {
 		if m.pwSvc != nil && ar.record.ProjectID > 0 {
 			_ = m.pwSvc.UpdateWorkflowRun(context.Background(), ar.record.RunID, status, "", "", errMsg)
 		}
@@ -633,14 +656,23 @@ func (m *Manager) ResumeInterrupted(parent context.Context, record *RunRecord, r
 }

 func (m *Manager) Cancel(runID string) error {
+	rec, _ := m.GetRun(runID)
+	if m.cmds != nil && rec != nil {
+		if _, err := m.cmds.Cancel(context.Background(), rec); err != nil && !errors.Is(err, ErrTerminalConflict) {
+			return err
+		}
+	}
 	m.mu.RLock()
 	ar, ok := m.runs[runID]
 	m.mu.RUnlock()
-	if !ok {
-		return fmt.Errorf("run not found")
+	if ok {
+		ar.cancel()
+		return nil
 	}
-	ar.cancel()
-	return nil
+	if rec != nil {
+		return nil
+	}
+	return fmt.Errorf("run not found")
 }

 func (m *Manager) heartbeatLoop(ctx context.Context, runID string) {
@@ -0,0 +1,151 @@
+package taskrun
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+const ManifestSchemaVersion = 1
+
+const (
+	MemberKindDocument      = "document"
+	MemberKindAssistantTurn = "assistant_turn"
+	MemberKindRunManifest   = "run_manifest"
+	MemberKindAttachment    = "attachment"
+	MemberKindEphemeral     = "ephemeral"
+)
+
+// ManifestMember is a retained output of a task run. Evidence/Gaps are not members.
+type ManifestMember struct {
+	LogicalID string `json:"logical_id"`
+	Digest    string `json:"digest"`
+	Kind      string `json:"kind"`
+}
+
+// RunManifest is the typed terminal protocol. Workspace copies are rebuildable, not the source of truth.
+type RunManifest struct {
+	SchemaVersion   int                        `json:"schema_version"`
+	RunID           string                     `json:"run_id"`
+	WorkspaceID     string                     `json:"workspace_id"`
+	StorageIdentity string                     `json:"storage_identity"`
+	Members         []ManifestMember           `json:"members"`
+	Extensions      map[string]json.RawMessage `json:"extensions,omitempty"`
+	CanonicalBytes  []byte                     `json:"-"`
+	SHA256          string                     `json:"-"`
+}
+
+type canonicalManifest struct {
+	SchemaVersion   int              `json:"schema_version"`
+	RunID           string           `json:"run_id"`
+	WorkspaceID     string           `json:"workspace_id"`
+	StorageIdentity string           `json:"storage_identity"`
+	Members         []ManifestMember `json:"members"`
+	Extensions      json.RawMessage  `json:"extensions,omitempty"`
+}
+
+// Freeze canonicalizes members, encodes stable JSON, and fills SHA-256.
+func (m *RunManifest) Freeze() error {
+	if m == nil {
+		return fmt.Errorf("taskrun: manifest required")
+	}
+	if strings.TrimSpace(m.RunID) == "" {
+		return fmt.Errorf("taskrun: manifest run_id required")
+	}
+	if m.SchemaVersion == 0 {
+		m.SchemaVersion = ManifestSchemaVersion
+	}
+	if m.SchemaVersion != ManifestSchemaVersion {
+		return fmt.Errorf("taskrun: unsupported manifest schema %d", m.SchemaVersion)
+	}
+	members := append([]ManifestMember(nil), m.Members...)
+	sort.Slice(members, func(i, j int) bool {
+		if members[i].LogicalID == members[j].LogicalID {
+			return members[i].Kind < members[j].Kind
+		}
+		return members[i].LogicalID < members[j].LogicalID
+	})
+	for i := range members {
+		members[i].LogicalID = strings.TrimSpace(members[i].LogicalID)
+		members[i].Digest = strings.ToLower(strings.TrimSpace(members[i].Digest))
+		members[i].Kind = strings.TrimSpace(members[i].Kind)
+		if members[i].LogicalID == "" || members[i].Digest == "" || members[i].Kind == "" {
+			return fmt.Errorf("taskrun: manifest member incomplete")
+		}
+	}
+	m.Members = members
+
+	canon := canonicalManifest{
+		SchemaVersion:   m.SchemaVersion,
+		RunID:           m.RunID,
+		WorkspaceID:     m.WorkspaceID,
+		StorageIdentity: m.StorageIdentity,
+		Members:         members,
+		Extensions:      marshalSortedExtensions(m.Extensions),
+	}
+	raw, err := marshalCanonical(canon)
+	if err != nil {
+		return err
+	}
+	sum := sha256.Sum256(raw)
+	m.CanonicalBytes = raw
+	m.SHA256 = hex.EncodeToString(sum[:])
+	return nil
+}
+
+func marshalCanonical(v any) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		return nil, err
+	}
+	return bytes.TrimSpace(buf.Bytes()), nil
+}
+
+func marshalSortedExtensions(ext map[string]json.RawMessage) json.RawMessage {
+	if len(ext) == 0 {
+		return nil
+	}
+	keys := make([]string, 0, len(ext))
+	for k := range ext {
+		if strings.TrimSpace(k) == "" || !strings.Contains(k, "/") {
+			continue
+		}
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	if len(keys) == 0 {
+		return nil
+	}
+	var buf bytes.Buffer
+	buf.WriteByte('{')
+	for i, k := range keys {
+		if i > 0 {
+			buf.WriteByte(',')
+		}
+		kb, _ := json.Marshal(k)
+		buf.Write(kb)
+		buf.WriteByte(':')
+		if len(ext[k]) == 0 {
+			buf.WriteString("null")
+		} else {
+			buf.Write(ext[k])
+		}
+	}
+	buf.WriteByte('}')
+	return buf.Bytes()
+}
+
+func memberNeedsRetention(kind string) bool {
+	switch kind {
+	case MemberKindDocument, MemberKindAssistantTurn, MemberKindRunManifest, MemberKindAttachment:
+		return true
+	default:
+		return false
+	}
+}
@@ -0,0 +1,77 @@
+package taskrun
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestRunManifest_FreezeStableSHA(t *testing.T) {
+	a := &RunManifest{
+		RunID:           "run_a",
+		WorkspaceID:     "ws1",
+		StorageIdentity: "env:dev",
+		Members: []ManifestMember{
+			{LogicalID: "doc_b", Digest: "bbb", Kind: MemberKindDocument},
+			{LogicalID: "doc_a", Digest: "aaa", Kind: MemberKindDocument},
+		},
+		Extensions: map[string]json.RawMessage{
+			"fav/policy": json.RawMessage(`{"v":1}`),
+			"ext/other":  json.RawMessage(`{"k":"v"}`),
+		},
+	}
+	b := &RunManifest{
+		RunID:           "run_a",
+		WorkspaceID:     "ws1",
+		StorageIdentity: "env:dev",
+		Members: []ManifestMember{
+			{LogicalID: "doc_a", Digest: "AAA", Kind: MemberKindDocument},
+			{LogicalID: "doc_b", Digest: "bbb", Kind: MemberKindDocument},
+		},
+		Extensions: map[string]json.RawMessage{
+			"ext/other":  json.RawMessage(`{"k":"v"}`),
+			"fav/policy": json.RawMessage(`{"v":1}`),
+		},
+	}
+	if err := a.Freeze(); err != nil {
+		t.Fatal(err)
+	}
+	if err := b.Freeze(); err != nil {
+		t.Fatal(err)
+	}
+	if a.SHA256 != b.SHA256 {
+		t.Fatalf("canonical sha diverged: %s vs %s\n%s\n%s", a.SHA256, b.SHA256, a.CanonicalBytes, b.CanonicalBytes)
+	}
+	if len(a.CanonicalBytes) == 0 || a.Members[0].LogicalID != "doc_a" {
+		t.Fatalf("members not sorted: %+v", a.Members)
+	}
+}
+
+func TestRunManifest_RejectsIncompleteMember(t *testing.T) {
+	m := &RunManifest{RunID: "run_x", Members: []ManifestMember{{LogicalID: "x", Kind: MemberKindDocument}}}
+	if err := m.Freeze(); err == nil {
+		t.Fatal("expected incomplete member error")
+	}
+}
+
+func TestResolveBackend_StripsUnauthorized(t *testing.T) {
+	if _, err := ResolveBackend("evil-shell", "stub"); err == nil {
+		t.Fatal("expected unauthorized")
+	}
+	got, err := ResolveBackend("", "pi-agent")
+	if err != nil || got != "pi-agent" {
+		t.Fatalf("got %q %v", got, err)
+	}
+}
+
+func TestStripSecretsAndSensitive(t *testing.T) {
+	meta := StripSecrets(map[string]any{"api_key": "secret", "run_id": "r1", "authorization": "Bearer x"})
+	if _, ok := meta["api_key"]; ok {
+		t.Fatal("api_key leaked")
+	}
+	if meta["run_id"] != "r1" {
+		t.Fatalf("run_id stripped: %#v", meta)
+	}
+	if err := rejectSensitiveText("hello api_key=abcd"); err == nil {
+		t.Fatal("expected sensitive reject")
+	}
+}
@@ -146,7 +146,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error {
 	if s.hasLeaseColumns(ctx) {
 		_, err := s.pool.Exec(ctx, `
 			UPDATE task_runs SET
-				status = $2,
+				status = CASE
+					WHEN status IN ('completed','cancelled','failed','quality_failed','pending_author_confirmation','finalizing_failed') THEN status
+					ELSE $2
+				END,
 				workspace_id = $3,
 				error_message = $4,
 				worker_id = $5,
@@ -159,7 +162,10 @@ func (s *PGStore) UpdateRun(ctx context.Context, rec *RunRecord) error {
 	}
 	_, err := s.pool.Exec(ctx, `
 		UPDATE task_runs SET
-			status = $2,
+			status = CASE
+				WHEN status IN ('completed','cancelled','failed','quality_failed','pending_author_confirmation','finalizing_failed') THEN status
+				ELSE $2
+			END,
 			workspace_id = $3,
 			error_message = $4,
 			assistant_message_id = $5,
@@ -0,0 +1,111 @@
+package workspace
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// OpaqueID turns a stored work dir or already-opaque id into a WorkspaceRef.ID.
+// It never returns a physical path.
+func OpaqueID(parts ...string) string {
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		slash := filepath.ToSlash(p)
+		if i := strings.LastIndex(slash, "/workspaces/"); i >= 0 {
+			rest := slash[i+len("/workspaces/"):]
+			if rest == "" {
+				continue
+			}
+			return strings.Split(rest, "/")[0]
+		}
+		if !strings.Contains(slash, "/") {
+			return p
+		}
+		return filepath.Base(filepath.Clean(p))
+	}
+	return ""
+}
+
+type RetentionReceipt struct {
+	LogicalID       string
+	Digest          string
+	StorageIdentity string
+	Verified        bool
+}
+
+type FrozenManifest struct {
+	SHA256          string
+	StorageIdentity string
+	WorkspaceID     string
+	Members         []FrozenMember
+}
+
+type FrozenMember struct {
+	LogicalID string
+	Digest    string
+	Kind      string
+}
+
+func memberNeedsRetention(kind string) bool {
+	switch kind {
+	case "document", "assistant_turn", "run_manifest", "attachment":
+		return true
+	default:
+		return false
+	}
+}
+
+var ErrCleanupBlocked = fmt.Errorf("workspace: cleanup blocked until verified receipts match the frozen manifest")
+
+func AssertCleanupAllowed(frozen FrozenManifest, receipts []RetentionReceipt) error {
+	if frozen.SHA256 == "" || frozen.StorageIdentity == "" {
+		return ErrCleanupBlocked
+	}
+	byLogical := map[string]RetentionReceipt{}
+	for _, r := range receipts {
+		byLogical[r.LogicalID] = r
+	}
+	for _, mem := range frozen.Members {
+		if !memberNeedsRetention(mem.Kind) {
+			continue
+		}
+		r, ok := byLogical[mem.LogicalID]
+		if !ok || !r.Verified {
+			return ErrCleanupBlocked
+		}
+		if r.Digest != mem.Digest || r.StorageIdentity != frozen.StorageIdentity {
+			return ErrCleanupBlocked
+		}
+	}
+	return nil
+}
+
+// Cleanup removes the workspace directory only after verified receipts match the frozen manifest.
+func (c *Capability) Cleanup(ctx context.Context, ref WorkspaceRef, frozen FrozenManifest, receipts []RetentionReceipt) error {
+	if err := AssertCleanupAllowed(frozen, receipts); err != nil {
+		return err
+	}
+	if err := c.guard(ctx, ref, false); err != nil {
+		return err
+	}
+	root := filepath.Join(c.dataRoot, "workspaces", ref.ID)
+	abs, err := filepath.Abs(root)
+	if err != nil {
+		return err
+	}
+	dataRoot, err := filepath.Abs(c.dataRoot)
+	if err != nil {
+		return err
+	}
+	rel, err := filepath.Rel(dataRoot, abs)
+	if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) {
+		return ErrPathEscape
+	}
+	return os.RemoveAll(abs)
+}
@@ -0,0 +1,63 @@
+package workspace
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestOpaqueID_StripsPhysicalPath(t *testing.T) {
+	got := OpaqueID("/data/dev/workspaces/run_abc/output.md", "")
+	if got != "run_abc" {
+		t.Fatalf("got %q", got)
+	}
+	if OpaqueID("already-opaque") != "already-opaque" {
+		t.Fatal("opaque id changed")
+	}
+}
+
+func TestCleanup_RequiresVerifiedReceipts(t *testing.T) {
+	root := t.TempDir()
+	cap := NewCapability(root, nil)
+	ctx := context.Background()
+	ref, err := cap.Issue(WorkspaceRef{ID: "ws_clean", ProjectScope: 1, RunID: "run_clean"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := cap.WriteArtifact(ctx, ref, DeliverableKey, []byte("# body\n")); err != nil {
+		t.Fatal(err)
+	}
+	frozen := FrozenManifest{
+		SHA256: "deadbeef", StorageIdentity: "env:dev", WorkspaceID: ref.ID,
+		Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}},
+	}
+	if err := cap.Cleanup(ctx, ref, frozen, nil); err == nil {
+		t.Fatal("cleanup without receipts must fail")
+	}
+	if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID, DeliverableKey)); err != nil {
+		t.Fatalf("workspace removed without receipts: %v", err)
+	}
+	receipts := []RetentionReceipt{{
+		LogicalID: "doc_a", Digest: "abc", StorageIdentity: "env:dev", Verified: true,
+	}}
+	if err := cap.Cleanup(ctx, ref, frozen, receipts); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(filepath.Join(root, "workspaces", ref.ID)); !os.IsNotExist(err) {
+		t.Fatalf("workspace still present: %v", err)
+	}
+}
+
+func TestCleanup_MismatchDigestRejected(t *testing.T) {
+	frozen := FrozenManifest{
+		SHA256: "x", StorageIdentity: "env:dev",
+		Members: []FrozenMember{{LogicalID: "doc_a", Digest: "abc", Kind: "document"}},
+	}
+	err := AssertCleanupAllowed(frozen, []RetentionReceipt{{
+		LogicalID: "doc_a", Digest: "other", StorageIdentity: "env:dev", Verified: true,
+	}})
+	if err == nil {
+		t.Fatal("mismatch must fail")
+	}
+}
@@ -0,0 +1,4 @@
+package workspace
+
+// DeliverableKey is the only artifact key for the task run markdown deliverable.
+const DeliverableKey = "output.md"
				`@@ -0,0 +1 @@`
				`{"status":"ok","env":"dev","agent_backend":"pi-agent","version":"v1.0","shared_kb_port":5432,"database_ready":true,"schema_version":37,"schema_ready":true,"task_store_ready":true}`