diff --git a/.doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md b/.doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md new file mode 100644 index 00000000..0a36da1d --- /dev/null +++ b/.doc/archive/ops/2026-09-15-taskrun-runmanifest-u8.md @@ -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`。 diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 6611776c..fd890979 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -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/` 或 `fix/`，**禁止** `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) | diff --git a/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md b/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md index f2a64590..bbb6df36 100644 --- a/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.md +++ b/.doc/ops/checklists/2026-09-14-module-decoupling-refactor-acceptance.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 和人工批准全部满足。 diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index f63c43eb..36fb99c5 100644 --- a/.doc/remediation/gap-matrix.md +++ b/.doc/remediation/gap-matrix.md @@ -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 业务代码 | --- diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index 518e8bb3..8e40324b 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.md @@ -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` | --- diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 95e59988..92b791ec 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -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 diff --git a/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md b/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md new file mode 100644 index 00000000..ba03a2f8 --- /dev/null +++ b/.doc/test-reports/2026-09-15-taskrun-runmanifest-u8-test-report.md @@ -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 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt new file mode 100644 index 00000000..f7cec0da --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/arch-check-warn.txt @@ -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 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt new file mode 100644 index 00000000..e11b4125 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-short.txt @@ -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 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt new file mode 100644 index 00000000..377f42e5 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/go-test-u8-packages-race.txt @@ -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 diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json new file mode 100644 index 00000000..5720c76a --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/health.json @@ -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} diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json new file mode 100644 index 00000000..5720c76a --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/ready.json @@ -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} diff --git a/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt new file mode 100644 index 00000000..f7448815 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-taskrun-runmanifest-u8/rebuild-health-ready.txt @@ -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} diff --git a/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql b/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql new file mode 100644 index 00000000..c2195c9e --- /dev/null +++ b/deploy/postgres/goose/00037_2026-09-15-task-run-manifests.sql @@ -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' + ) +); diff --git a/internal/ctxmgr/message_update.go b/internal/ctxmgr/message_update.go index 5c403653..e4e655b0 100644 --- a/internal/ctxmgr/message_update.go +++ b/internal/ctxmgr/message_update.go @@ -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...) +} diff --git a/internal/db/migrate_test.go b/internal/db/migrate_test.go index 4c3da960..dfd3f360 100644 --- a/internal/db/migrate_test.go +++ b/internal/db/migrate_test.go @@ -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) diff --git a/internal/db/testdb.go b/internal/db/testdb.go index 4abbb032..e0a8d35a 100644 --- a/internal/db/testdb.go +++ b/internal/db/testdb.go @@ -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") + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 37eccceb..7f9873e8 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -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, diff --git a/internal/server/server.go b/internal/server/server.go index e9731e9f..f0e703d4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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)) diff --git a/internal/server/sse.go b/internal/server/sse.go index d19cbc8d..ffad0f19 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -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 != "" { diff --git a/internal/server/task_finalize.go b/internal/server/task_finalize.go index 1d933c43..1a84ed13 100644 --- a/internal/server/task_finalize.go +++ b/internal/server/task_finalize.go @@ -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 diff --git a/internal/server/task_finalize_test.go b/internal/server/task_finalize_test.go index 37c65bc4..12d21518 100644 --- a/internal/server/task_finalize_test.go +++ b/internal/server/task_finalize_test.go @@ -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) + } +} diff --git a/internal/server/task_run_handlers.go b/internal/server/task_run_handlers.go index 5a6e6b9e..a25babf6 100644 --- a/internal/server/task_run_handlers.go +++ b/internal/server/task_run_handlers.go @@ -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, diff --git a/internal/taskrun/backend.go b/internal/taskrun/backend.go new file mode 100644 index 00000000..3c95a975 --- /dev/null +++ b/internal/taskrun/backend.go @@ -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 +} diff --git a/internal/taskrun/finalize.go b/internal/taskrun/finalize.go new file mode 100644 index 00000000..b88c0766 --- /dev/null +++ b/internal/taskrun/finalize.go @@ -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(¤t) + 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 +} diff --git a/internal/taskrun/finalize_pg_test.go b/internal/taskrun/finalize_pg_test.go new file mode 100644 index 00000000..de352d73 --- /dev/null +++ b/internal/taskrun/finalize_pg_test.go @@ -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") + } +} diff --git a/internal/taskrun/finalize_test.go b/internal/taskrun/finalize_test.go new file mode 100644 index 00000000..210495d6 --- /dev/null +++ b/internal/taskrun/finalize_test.go @@ -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") + } +} diff --git a/internal/taskrun/manager.go b/internal/taskrun/manager.go index 2efbd448..21d86eee 100644 --- a/internal/taskrun/manager.go +++ b/internal/taskrun/manager.go @@ -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) { diff --git a/internal/taskrun/manifest.go b/internal/taskrun/manifest.go new file mode 100644 index 00000000..3e51941c --- /dev/null +++ b/internal/taskrun/manifest.go @@ -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 + } +} diff --git a/internal/taskrun/manifest_test.go b/internal/taskrun/manifest_test.go new file mode 100644 index 00000000..a3657f22 --- /dev/null +++ b/internal/taskrun/manifest_test.go @@ -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") + } +} diff --git a/internal/taskrun/store.go b/internal/taskrun/store.go index 13abb602..c6827259 100644 --- a/internal/taskrun/store.go +++ b/internal/taskrun/store.go @@ -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, diff --git a/internal/workspace/cleanup.go b/internal/workspace/cleanup.go new file mode 100644 index 00000000..57a49c7f --- /dev/null +++ b/internal/workspace/cleanup.go @@ -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) +} diff --git a/internal/workspace/cleanup_test.go b/internal/workspace/cleanup_test.go new file mode 100644 index 00000000..381fa673 --- /dev/null +++ b/internal/workspace/cleanup_test.go @@ -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") + } +} diff --git a/internal/workspace/keys.go b/internal/workspace/keys.go new file mode 100644 index 00000000..abc4e8e2 --- /dev/null +++ b/internal/workspace/keys.go @@ -0,0 +1,4 @@ +package workspace + +// DeliverableKey is the only artifact key for the task run markdown deliverable. +const DeliverableKey = "output.md"