diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index 81d42d3e..2519e162 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -5,7 +5,7 @@ ## 焦点 -1. **MULTI-INTENT HITL**：U2 规则分段已签（[U2 报告](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)）；下一单元 U3 PG/CAS 待认领 +1. **MULTI-INTENT HITL**：U3 PG/CAS 已签（[U3 报告](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md)）；下一单元 U4 outbox/投影待认领 2. **发布门禁**：CI `TEST_DATABASE_URL`、真实 PI 三工作流 E2E、前端人工抽检仍开放（[gap-matrix](../remediation/gap-matrix.md)） 3. **CHAT-INTENT**：双标签 busy 已签（项目 **6275**）；不删 `sess-p143-u9`；项目 143 只读 @@ -13,6 +13,7 @@ | owner | task | branch | worktree | mutex | status | |-------|------|--------|----------|-------|--------| +| Cursor | 2026-09-15-multi-intent-u3-intentflow-pg | feature/multi-intent-u3-intentflow-pg | - | intent-plan | done | | Cursor | 2026-09-15-multi-intent-u2-segmenter | feature/multi-intent-u2-segmenter | - | intent | done | | Cursor | 2026-09-15-multi-intent-u1-contracts | feature/multi-intent-u1-contracts | - | architecture-docs | done | | Cursor | app-composition-u11-u12 | feature/app-composition-u11-u12 | - | app | done | @@ -86,7 +87,7 @@ | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| - | idle | - | - | - | +| Cursor | merging | 2026-09-15 | multi-intent-u3-intentflow-pg（schema 文件 44，不 rebuild 共享环境） | - | **phase**: `idle` | `merging` | `smoke` | `e2e` @@ -122,6 +123,7 @@ | 多意图 HITL 复核方案 | [../system-design/2026-09-15-multi-intent-hitl-plan.md](../system-design/2026-09-15-multi-intent-hitl-plan.md) | | 多意图 HITL U1 契约 | [../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md) | | 多意图 HITL U2 分段 | [../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md) | +| 多意图 HITL U3 PG | [../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) | | 文档可靠性 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) | diff --git a/.doc/ops/checklists/2026-09-15-multi-intent-u3-intentflow-pg.md b/.doc/ops/checklists/2026-09-15-multi-intent-u3-intentflow-pg.md new file mode 100644 index 00000000..6e56f372 --- /dev/null +++ b/.doc/ops/checklists/2026-09-15-multi-intent-u3-intentflow-pg.md @@ -0,0 +1,22 @@ +--- +title: "多意图 HITL U3 intentflow PG checklist" +status: done +created: 2026-09-15 +updated: 2026-09-15 +related_task: "2026-09-15-multi-intent-u3-intentflow-pg" +--- + +# 多意图 HITL U3 intentflow PG checklist + +## Wave 2 + +- [x] Goose 00044 三表 + unique/CAS/lease 列 — 证据：`deploy/postgres/goose/00044_2026-09-15-intent-plans.sql`；`TestGooseU3_00044_HasUpAndDown` +- [x] 创建幂等与同 turn 异 digest — 证据：[plans-verbose.txt](../../test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/plans-verbose.txt) +- [x] 决策 nonce / revision CAS / 过期 — 证据：同上 +- [x] 两 worker 无双 owner；租约接管；旧 fence 拒绝 — 证据：`TestTwoWorkersOneOwnerAndLeaseTakeover` +- [x] receipt 对账不盲跑；环境隔离；legacy HandleTurn 不落盘 — 证据：`TestReceiptWritebackFailureDoesNotRerun` / `TestHandleTurnLegacyDoesNotPersist` +- [x] Assemble 接线且 chat 不调用；arch-check 无新例外 — 证据：[short.txt](../../test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/short.txt) + +## 完成定义 + +- [x] 全部勾选且有证据链接 diff --git a/.doc/ops/short-term/2026-09-15-multi-intent-hitl.md b/.doc/ops/short-term/2026-09-15-multi-intent-hitl.md index c9e2ea08..1ace26d5 100644 --- a/.doc/ops/short-term/2026-09-15-multi-intent-hitl.md +++ b/.doc/ops/short-term/2026-09-15-multi-intent-hitl.md @@ -3,7 +3,7 @@ title: "多意图持久计划与 HITL（核心 U1–U8，扩面 U9–U12）" status: in_progress created: 2026-09-15 updated: 2026-09-15 -owner: 待认领（下一单元 U3） +owner: 待认领（下一单元 U4） branch: "-" worktree: "-" mutex: "-" @@ -53,7 +53,7 @@ mutex: "-" 1. [x] U1 冻结边界、状态机、批量策略、错误码、事件和默认 legacy tier（[独立任务](./2026-09-15-multi-intent-u1-contracts.md)；[报告](../../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)）； 2. [x] U2 characterization-first 建 golden/hard-negative 与 shadow 基线（[独立任务](./2026-09-15-multi-intent-u2-segmenter.md)；[报告](../../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)）； -3. [ ] U3 建 intentflow owner、PG schema、CAS/lease/fencing/retention； +3. [x] U3 建 intentflow owner、PG schema、CAS/lease/fencing/retention（[独立任务](./2026-09-15-multi-intent-u3-intentflow-pg.md)；[报告](../../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md)）； 4. [ ] U4 接 Track E outbox/dispatcher，只投影状态事实； 5. [ ] U5 建统一 decision command 与动作 owner 授权端口； 6. [ ] U6 取消计划动作的前端本地成功路径，以 PlanView 恢复； @@ -92,13 +92,14 @@ mutex: "-" ## 相关链接 -- Checklist: [U1](../checklists/2026-09-15-multi-intent-u1-contracts.md)；[U2](../checklists/2026-09-15-multi-intent-u2-segmenter.md) +- Checklist: [U1](../checklists/2026-09-15-multi-intent-u1-contracts.md)；[U2](../checklists/2026-09-15-multi-intent-u2-segmenter.md)；[U3](../checklists/2026-09-15-multi-intent-u3-intentflow-pg.md) - Handoff: 每个 U-ID 独立建立 -- Test report: [U1](../../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2](../../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md) +- Test report: [U1](../../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2](../../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)；[U3](../../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) ## 证据 / 备注 - 当前 `plan_builder.go` 最多两段且写意图返回 nil；`action_planner.go` 的确认仍是文字提示。 - 当前 `module-boundaries.yaml` 中 intent 仅依赖 maturity；Track E live，FAV off。 - U1 已冻结 intentflow 契约、事件与 BatchActionPolicy，默认 `MULTI_INTENT_PLAN_TIER=legacy`。 -- U2 已落地规则分段与 shadow `PlanProposal`；生产 TopIntent / IntentPlan 不变；反事实 `false_auto_execute=0`。下一单元 U3：intentflow PG/CAS/lease。 +- U2 已落地规则分段与 shadow `PlanProposal`；生产 TopIntent / IntentPlan 不变；反事实 `false_auto_execute=0`。 +- U3 已落地 PG 计划账本与 CAS/lease/fencing；HandleTurn 默认不落盘。下一单元 U4：outbox/投影。共享环境 schema 44 待晋升窗口。 diff --git a/.doc/ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md b/.doc/ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md new file mode 100644 index 00000000..ac96c76b --- /dev/null +++ b/.doc/ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md @@ -0,0 +1,68 @@ +--- +title: "多意图 HITL U3：intentflow PG/CAS/lease" +status: done +created: 2026-09-15 +updated: 2026-09-15 +owner: Cursor +branch: feature/multi-intent-u3-intentflow-pg +worktree: "-" +mutex: intent-plan +--- + +# 多意图 HITL U3：intentflow PG/CAS/lease + +## 目标 + +建立持久计划/步骤/决策与并发协议（CAS revision、decision nonce、lease/fencing、创建幂等、retention）；不接真实写动作，不改 chat/coordinator 生产路由。 + +## 范围 + +- Goose `00044`：`intent_plans` / `intent_plan_steps` / `intent_plan_decisions` +- `app/intentflow` owner repository + UoW；owner 可 unwrap pgx +- 创建幂等 `(environment, session_id, client_turn_id)`；同 turn 异 digest 冲突 +- 决策 CAS + nonce 同/异 digest；确认/取消/过期竞态 +- 步骤 lease + 单调 fencing；两 worker 只一个 owner +- 业务成功但 receipt 写回失败时按幂等键对账，不盲跑 +- 环境隔离；关闭计划 retention 由 owner 删除 +- `app.Assemble` 装配 `*intentflow.Plans`，chat 仍不调用 + +## 非目标 + +- 不接 HTTP（U5）、不 Append 事件（U4）、不接真实 action adapter（U7/U8） +- 不把 HITL/tier 回灌 ActionRegistry +- 不改 `internal/intent` 生产 TopIntent / IntentPlan +- 不 rebuild 共享 `:8080`（schema 44 仅隔离库/后续晋升窗口应用） + +## 步骤 + +1. [x] 认领 CURRENT（mutex `intent-plan`） +2. [x] Goose 00044 + migrate gate +3. [x] Store/Engine：Create/Decide/Claim/Receipt/Expire/Purge +4. [x] 内存并发 + 可选 PG 测试；Assemble 接线 +5. [x] 报告并合回本地 dev + +## 验收标准 + +- [x] 双创建同 digest 返回同一 plan；异 digest 冲突 +- [x] nonce 同 digest 返回 receipt；异 digest 冲突 +- [x] 确认/取消/过期只一个 CAS 胜出 +- [x] 两 worker 抢 eligible 步骤无双 owner +- [x] 租约过期可接管；旧 fencing token 被动作属主拒绝 +- [x] 重启后 succeeded 不重跑；receipt 写回失败按幂等键恢复 +- [x] 跨 environment 隐藏 +- [x] 默认 legacy 时 HandleTurn 不落盘 +- [x] `go test -short ./internal/app/intentflow ./internal/app ./internal/archtest ./internal/db` 绿 + +## 相关链接 + +- 总览: [2026-09-15-multi-intent-hitl.md](./2026-09-15-multi-intent-hitl.md) +- 设计: [../../system-design/2026-09-15-multi-intent-hitl-plan.md](../../system-design/2026-09-15-multi-intent-hitl-plan.md) +- Checklist: [../checklists/2026-09-15-multi-intent-u3-intentflow-pg.md](../checklists/2026-09-15-multi-intent-u3-intentflow-pg.md) +- Test report: [../../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md](../../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) + +## 证据 / 备注 + +- HandleTurn 在默认 legacy 下不落盘；CreatePlan 是 U3 测试与后续 U7 入口。 +- 动作属主端口仅 FakeActionOwner；真实 adapter 从 U7/U8 按 key 接入。 +- PG 活验需隔离 `TEST_DATABASE_URL`；本机 short 门禁跳过。共享 `:8080` 仍 schema 43。 +- 提交：`a75aa45` docs(ops) 认领；`7dfd6d6` feat(db)；`da0b3d8` feat(intentflow)；`1385002` feat(app)。 diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index 42bc2890..e7f8f9d0 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.md @@ -15,8 +15,8 @@ | 模块 | 能力 | 状态 | 证据 | 缺口/后续 | mutex | |------|------|------|------|-----------|-------| -| 意图识别 L1 | 18 类 Intent + 三分支消解 | partial | [system-design/intent-recognition](../system-design/intent-recognition/)；[现场专项审计](../audit/2026-09-13-conversation-intent-runtime-state.md)；[运行态报告](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md)；[双标签活验](../test-reports/assets/2026-09-15-conversation-intent-dual-tab/live.json)；[多意图复核方案](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1 契约](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2 分段](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md) | 明确写动作与只读总结已分流；双标签 `DOC_BUSY` 已签（项目 6275）。多意图 U1 契约已冻；U2 shadow `PlanProposal` 已落地且生产路由不变；U3–U8 未实施；intent 纯依赖边界不得回退 | `intent` | -| 多意图持久计划 | app/intentflow 生命周期与 HITL 契约 | planned | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1 报告](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2 报告](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)；[U2 任务](../ops/short-term/2026-09-15-multi-intent-u2-segmenter.md) | U1 契约 + U2 规则分段 shadow 已完成。下一单元 U3 PG/CAS/lease；U4 投影；U5–U8 决策/Web/只读执行/task.generate | `intent-plan` | +| 意图识别 L1 | 18 类 Intent + 三分支消解 | partial | [system-design/intent-recognition](../system-design/intent-recognition/)；[现场专项审计](../audit/2026-09-13-conversation-intent-runtime-state.md)；[运行态报告](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md)；[双标签活验](../test-reports/assets/2026-09-15-conversation-intent-dual-tab/live.json)；[多意图复核方案](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1 契约](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2 分段](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)；[U3 PG](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) | 明确写动作与只读总结已分流；双标签 `DOC_BUSY` 已签（项目 6275）。多意图 U1–U3 已完成（契约/shadow 提案/PG CAS）；U4–U8 未实施；intent 纯依赖边界不得回退 | `intent` | +| 多意图持久计划 | app/intentflow 生命周期与 HITL 契约 | partial | [设计](../system-design/2026-09-15-multi-intent-hitl-plan.md)；[U1 报告](../test-reports/2026-09-15-multi-intent-u1-contracts-test-report.md)；[U2 报告](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md)；[U3 报告](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md)；[U3 任务](../ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md) | U3 计划表/CAS/lease 已落地，默认 HandleTurn 不落盘。下一单元 U4 投影；U5–U8 决策/Web/只读执行/task.generate。共享环境 schema 44 待晋升 | `intent-plan` | | 对话 / ctxmgr | 多轮上下文、记忆、摘要 | partial | [system-design/multi-turn-dialogue](../system-design/multi-turn-dialogue/)；[刷新恢复专项审计](../audit/2026-09-13-chat-history-refresh-disappears.md)；[U10 端口化](../archive/ops/2026-09-15-chatlog-contextasm-u10.md)；[U10 报告](../test-reports/2026-09-15-chatlog-contextasm-u10-test-report.md) | `CHAT-HISTORY-REFRESH` 闲聊刷新活验已补。U10 原地端口 **停止扩面**（未物理拆包）。语义摘要(D5 backlog) | `chat` | | 智能体小队 | 角色-任务-工作区-黑板 | done | [system-design/agent-squad](../system-design/agent-squad/)；[test 2026-09-07](../test-reports/2026-09-07-dynamic-squad-planning-test-report.md) | 非关键步骤降级(A5/A6 backlog) | `squad` | | 工作流治理 | 三正式工作流 + 阶段门禁 | partial | [remediation/three-workflows](../remediation/2026-09-12-three-workflows-strict-governance-remediation.md)；[test 2026-09-12](../test-reports/2026-09-12-three-workflows-governance-remediation-test-report.md)；[冻结态/职责可选整改](../remediation/2026-09-14-workflow-plan-409-frozen-and-optional-duties-remediation.md)；[Wave 5 页面矩阵](../test-reports/2026-09-14-workflow-plan-409-frozen-and-optional-duties-test-report.md)；[职责-动作鉴权](../test-reports/2026-09-14-governance-action-level-authz-test-report.md) | 设置页/DutyMode + 动作级鉴权/未激活 202-blocked/关闭创建直通已落地。后续：none 文档真实 PI 审批长链路、CI 三工作流 E2E | `workflow` | diff --git a/.doc/system-design/README.md b/.doc/system-design/README.md index 103992d7..f2fe11b4 100644 --- a/.doc/system-design/README.md +++ b/.doc/system-design/README.md @@ -128,6 +128,7 @@ | 任务总览 | [../ops/short-term/2026-09-15-multi-intent-hitl.md](../ops/short-term/2026-09-15-multi-intent-hitl.md) | | U1 契约 | [../ops/short-term/2026-09-15-multi-intent-u1-contracts.md](../ops/short-term/2026-09-15-multi-intent-u1-contracts.md) | | U2 分段 | [../ops/short-term/2026-09-15-multi-intent-u2-segmenter.md](../ops/short-term/2026-09-15-multi-intent-u2-segmenter.md)；[报告](../test-reports/2026-09-15-multi-intent-u2-segmenter-test-report.md) | +| U3 PG | [../ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md](../ops/short-term/2026-09-15-multi-intent-u3-intentflow-pg.md)；[报告](../test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md) | | 一句话总结 | intent 保持纯规则，intentflow 持久化计划并统一确认、授权、幂等、恢复和逐动作执行 | **硬边界**：确认不等于授权；PG 是计划真相，卡片/SSE 是投影；写动作按 owner 端口逐项接入；LLM 只能提出不可信分段建议。 diff --git a/.doc/test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md b/.doc/test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md new file mode 100644 index 00000000..2375e19e --- /dev/null +++ b/.doc/test-reports/2026-09-15-multi-intent-u3-intentflow-pg-test-report.md @@ -0,0 +1,112 @@ +# 多意图 HITL U3 intentflow PG/CAS/lease 测试报告 + +## 一、 测试背景 (Background) + +U1 冻结了 intentflow 契约，U2 在 L1 产出 shadow `PlanProposal`，生产路由仍不落盘。U3 要把计划/步骤/决策变成 PostgreSQL 权威账本，并用 CAS、nonce、lease 与单调 fencing 保证两 worker 不能双 owner / 双执行。本单元不接真实写动作，也不改 chat 路径。 + +## 二、 测试目标 (Test Objectives) + +- Goose `00044` 建立 `intent_plans` / `intent_plan_steps` / `intent_plan_decisions` +- 同 turn 同 digest 幂等；异 digest 冲突；同一作用域不能有第二个活动写计划 +- 决策 nonce 同 digest 返回 receipt，异 digest 冲突；确认/取消竞态只一个 CAS 胜出 +- 两 worker 抢 eligible 步骤只有一个 lease；过期可接管；旧 fencing token 被动作属主拒绝 +- 业务成功但 receipt 写回失败时按幂等键恢复，不盲跑 +- 跨 environment 隐藏；关闭计划可由 owner purge +- 默认 legacy 下 `HandleTurn` 不落盘；chat/coordinator 无回归 +- 无 HTTP、无真实 adapter、无共享环境 rebuild + +## 三、 测试方案与环境 (Test Plan & Environment) + +| 项 | 值 | +|----|-----| +| OS | Linux 7.0.0-31-generic | +| Go | go1.27.1 linux/amd64 | +| 分支 | `feature/multi-intent-u3-intentflow-pg`（从本地 `dev@53c0722` 检出） | +| Schema | 文件 **44**（共享 `:8080` 仍为 **43**，本单元不 rebuild） | +| 容器 | 不 rebuild | +| 项目 143 | **只读** | +| 证据 | `.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/` | + +测试层级：内存 CAS/lease 单测（`-short`）/ goose 文件门禁 / arch-check / `go test -short ./...`。PG 集成 `TestPlansPG_CreateClaimAndFence` 在 `-short` 下跳过，需隔离 `TEST_DATABASE_URL`。 + +## 四、 预期结果 (Expected Results) + +| 用例 | 判定 | +|------|------| +| HandleTurn legacy | 不创建计划，GetActive 为空 | +| 双创建同 digest | 同一 plan_id | +| 同 turn 异 digest | `PLAN_DECISION_CONFLICT` | +| 第二份活动写计划 | 冲突并返回已有活动计划 | +| nonce 同/异 digest | receipt / conflict | +| 确认 vs 取消 | 恰好一个 CAS 胜出 | +| 两 worker | 仅一个 owner；租约过期接管；旧 fence `PLAN_ACTION_VERSION_CONFLICT` | +| receipt 对账 | execute count=1，步骤 succeeded | +| 跨 env / purge | `PLAN_NOT_FOUND_OR_HIDDEN` | +| arch-check / 全仓 short | PASS | + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 创建幂等与活动写计划锁 + +- **测试输入 (Input)**：`go test -v -count=1 ./internal/app/intentflow -run 'TestCreate'` +- **实际输出 (Actual Output)**： + +``` +--- PASS: TestCreateIdempotentSameDigest (0.00s) +--- PASS: TestCreateRejectsSecondActiveWritePlan (0.00s) +``` + +- **评价判定 (Evaluation / Verdict)**：PASS。 +- **性能与耗时 (Duration & Latency)**：<10ms + +### 5.2 决策 CAS / nonce / 过期隔离 + +- **测试输入**：`TestDecisionNonceSameAndDifferentDigest`、`TestConfirmCancelRaceOneWinner`、`TestExpireAndEnvironmentIsolationAndRetention` +- **实际输出**：见 [plans-verbose.txt](assets/2026-09-15-multi-intent-u3-intentflow-pg/plans-verbose.txt)。三用例均 PASS。 +- **评价判定**：PASS。确认/取消竞态恰好一个胜出；过期后确认失败；跨 env 隐藏；purge 后不可见。 +- **性能与耗时**：<10ms + +### 5.3 lease / fencing / receipt 对账 + +- **测试输入**：`TestTwoWorkersOneOwnerAndLeaseTakeover`、`TestReceiptWritebackFailureDoesNotRerun` +- **实际输出**： + +``` +--- PASS: TestTwoWorkersOneOwnerAndLeaseTakeover (0.00s) +--- PASS: TestReceiptWritebackFailureDoesNotRerun (0.00s) +``` + +- **评价判定**：PASS。旧 fencing token 由 FakeActionOwner 拒绝；写回失败路径只按幂等键恢复，不二次执行。 +- **性能与耗时**：<10ms + +### 5.4 HandleTurn 与边界 + +- **测试输入**：`TestHandleTurnLegacyDoesNotPersist`、`TestUnauthenticatedRejected`、`TestGooseU3_00044_HasUpAndDown` +- **实际输出**：intentflow 与 db 包均 PASS。完整包输出见 [unit.txt](assets/2026-09-15-multi-intent-u3-intentflow-pg/unit.txt)。 +- **评价判定**：PASS。生产 HandleTurn 仍不落盘。00044 含 Up/Down 与三表、turn/write/nonce 唯一索引。 +- **性能与耗时**：db 约 3ms + +### 5.5 全仓 short 与 arch-check + +- **测试输入**：`go vet ./internal/app/intentflow ./internal/app ./internal/db`；`./scripts/arch-check.sh --mode block`；`unset TEST_DATABASE_URL; go test -short -count=1 ./...` +- **实际输出**：`go vet` 无输出。`arch-check block: production gate passed`。全仓见 [short.txt](assets/2026-09-15-multi-intent-u3-intentflow-pg/short.txt)。`internal/server` 约 9.9s；全包 ok。 +- **评价判定**：PASS。`intent -> maturity` 保持；chat/coordinator 无回归。SQL 写仍仅 `app/intentflow`。 +- **性能与耗时**：全仓 short 约 12s + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 命令 | 结果 | +|------|------| +| `go vet ./internal/app/intentflow ./internal/app ./internal/db` | PASS | +| `go test -count=1 ./internal/app/intentflow` | PASS（PG 用例 `-short` 跳过） | +| `go test -short -count=1 ./internal/db ./internal/archtest ./internal/intent ./internal/coordinator ./internal/app` | PASS | +| `go test -short -count=1 ./...` | PASS | +| `./scripts/arch-check.sh --mode block` | PASS | + +通过率 100%。隔离库 `TEST_DATABASE_URL` 未在本机设置，PG 活验按 skip 记录，不冒充已跑。共享 `:8080` schema 仍为 43。 + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +U3 达到合入标准：intentflow 成为计划表唯一写者，CAS/lease/fencing/retention 在内存路径闭环，默认 legacy 不改变对话路由。 + +后续：U4 在独立分支接 Track E outbox/投影；不得修改通用 dispatcher lease 语义。U5 再挂 HTTP。共享环境应用 schema 44 放到后续晋升/rebuild 窗口，不在本单元对项目 143 做迁移。 diff --git a/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/plans-verbose.txt b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/plans-verbose.txt new file mode 100644 index 00000000..a9bbc4df --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/plans-verbose.txt @@ -0,0 +1,20 @@ +=== RUN TestHandleTurnLegacyDoesNotPersist +--- PASS: TestHandleTurnLegacyDoesNotPersist (0.00s) +=== RUN TestCreateIdempotentSameDigest +--- PASS: TestCreateIdempotentSameDigest (0.00s) +=== RUN TestCreateRejectsSecondActiveWritePlan +--- PASS: TestCreateRejectsSecondActiveWritePlan (0.00s) +=== RUN TestDecisionNonceSameAndDifferentDigest +--- PASS: TestDecisionNonceSameAndDifferentDigest (0.00s) +=== RUN TestConfirmCancelRaceOneWinner +--- PASS: TestConfirmCancelRaceOneWinner (0.00s) +=== RUN TestTwoWorkersOneOwnerAndLeaseTakeover +--- PASS: TestTwoWorkersOneOwnerAndLeaseTakeover (0.00s) +=== RUN TestReceiptWritebackFailureDoesNotRerun +--- PASS: TestReceiptWritebackFailureDoesNotRerun (0.00s) +=== RUN TestExpireAndEnvironmentIsolationAndRetention +--- PASS: TestExpireAndEnvironmentIsolationAndRetention (0.00s) +=== RUN TestUnauthenticatedRejected +--- PASS: TestUnauthenticatedRejected (0.00s) +PASS +ok chat-agent-system/internal/app/intentflow 0.005s diff --git a/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/short.txt b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/short.txt new file mode 100644 index 00000000..aaeb4822 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/short.txt @@ -0,0 +1,61 @@ +ok chat-agent-system/cmd/bizdev-cli 1.946s +? chat-agent-system/cmd/dataset [no test files] +ok chat-agent-system/cmd/document-content-reconcile 0.002s +? chat-agent-system/cmd/document-revision-backfill [no test files] +? chat-agent-system/cmd/eventmode [no test files] +? chat-agent-system/cmd/events [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 3.163s +ok chat-agent-system/internal/app 0.015s +ok chat-agent-system/internal/app/chatflow 0.007s +ok chat-agent-system/internal/app/docadvance 0.005s +ok chat-agent-system/internal/app/docflow 0.005s +ok chat-agent-system/internal/app/docnotify 0.005s +ok chat-agent-system/internal/app/docproject 0.005s +ok chat-agent-system/internal/app/intentflow 0.005s +ok chat-agent-system/internal/app/stageflow 0.005s +ok chat-agent-system/internal/app/uow 0.006s +ok chat-agent-system/internal/archtest 0.200s +ok chat-agent-system/internal/artifactvault 0.007s +ok chat-agent-system/internal/auth 0.840s +ok chat-agent-system/internal/authz 0.004s +ok chat-agent-system/internal/bizdev 0.169s +ok chat-agent-system/internal/chat 0.018s +ok chat-agent-system/internal/config 0.004s +ok chat-agent-system/internal/coordinator 0.013s +ok chat-agent-system/internal/ctxmgr 0.096s +ok chat-agent-system/internal/dataset 0.005s +ok chat-agent-system/internal/db 0.005s +ok chat-agent-system/internal/doccontent 0.002s +ok chat-agent-system/internal/docduty 0.005s +ok chat-agent-system/internal/docgov 0.020s +ok chat-agent-system/internal/docquery 0.005s +ok chat-agent-system/internal/docrevision 0.007s +ok chat-agent-system/internal/events 0.007s +ok chat-agent-system/internal/freedoc 0.029s +ok chat-agent-system/internal/fsutil 0.005s +ok chat-agent-system/internal/httpx 0.005s +ok chat-agent-system/internal/intent 0.029s +ok chat-agent-system/internal/kb 0.008s +ok chat-agent-system/internal/maturity 0.004s +ok chat-agent-system/internal/notification 0.083s +? chat-agent-system/internal/observability [no test files] +ok chat-agent-system/internal/projectcontext 0.007s +ok chat-agent-system/internal/projectinstance 0.007s +ok chat-agent-system/internal/projectworkflow 0.006s +ok chat-agent-system/internal/quote 0.009s +ok chat-agent-system/internal/replyexport 0.002s +ok chat-agent-system/internal/server 9.896s +ok chat-agent-system/internal/squad 0.030s +ok chat-agent-system/internal/storage 0.015s +ok chat-agent-system/internal/stream 0.003s +ok chat-agent-system/internal/stream/transport 0.002s +ok chat-agent-system/internal/taskrun 0.146s +ok chat-agent-system/internal/template 0.721s +ok chat-agent-system/internal/upload 0.002s +? chat-agent-system/internal/workflow [no test files] +ok chat-agent-system/internal/workflow/runtime 0.003s +ok chat-agent-system/internal/workflowdef 0.003s +ok chat-agent-system/internal/workspace 0.003s +? chat-agent-system/templates [no test files] diff --git a/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/unit.txt b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/unit.txt new file mode 100644 index 00000000..bb6c3fc6 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-multi-intent-u3-intentflow-pg/unit.txt @@ -0,0 +1,6 @@ +ok chat-agent-system/internal/app/intentflow 0.003s +ok chat-agent-system/internal/app 0.013s +ok chat-agent-system/internal/db 0.002s +ok chat-agent-system/internal/archtest 0.157s +ok chat-agent-system/internal/intent 0.017s +ok chat-agent-system/internal/coordinator 0.009s diff --git a/deploy/postgres/goose/00044_2026-09-15-intent-plans.sql b/deploy/postgres/goose/00044_2026-09-15-intent-plans.sql new file mode 100644 index 00000000..c2d2c5f7 --- /dev/null +++ b/deploy/postgres/goose/00044_2026-09-15-intent-plans.sql @@ -0,0 +1,160 @@ +-- +goose Up +-- U3: persistent multi-intent plans. Owner is app/intentflow; other packages use the Service port. +-- JSONB is only for versioned slots / display snapshots. CAS, lease, and audit columns are typed. + +CREATE TABLE IF NOT EXISTS intent_plans ( + plan_id TEXT PRIMARY KEY, + environment TEXT NOT NULL, + project_id BIGINT NOT NULL, + session_id TEXT NOT NULL, + user_id BIGINT NOT NULL, + client_turn_id TEXT NOT NULL, + request_digest TEXT NOT NULL, + segmenter_version TEXT NOT NULL DEFAULT '', + policy_version TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + revision BIGINT NOT NULL DEFAULT 1, + expires_at TIMESTAMPTZ NOT NULL, + last_fencing_token BIGINT NOT NULL DEFAULT 0, + write_plan BOOLEAN NOT NULL DEFAULT FALSE, + execution_mode TEXT NOT NULL DEFAULT 'validated_only', + safe_summary TEXT NOT NULL DEFAULT '', + display_snapshot JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_intent_plan_status CHECK (status IN ( + 'proposed', + 'awaiting_clarification', + 'awaiting_confirmation', + 'executing', + 'partially_completed', + 'completed', + 'failed', + 'cancelled', + 'expired', + 'superseded' + )), + CONSTRAINT chk_intent_plan_execution_mode CHECK (execution_mode IN ( + 'validated_only', + 'live' + )) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_intent_plans_turn + ON intent_plans (environment, session_id, client_turn_id); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_intent_plans_one_active_write + ON intent_plans (environment, project_id, session_id, user_id) + WHERE write_plan AND status IN ( + 'proposed', + 'awaiting_clarification', + 'awaiting_confirmation', + 'executing', + 'partially_completed' + ); + +CREATE INDEX IF NOT EXISTS idx_intent_plans_active_scope + ON intent_plans (environment, project_id, session_id, user_id, status); + +CREATE INDEX IF NOT EXISTS idx_intent_plans_expire + ON intent_plans (expires_at) + WHERE status IN ('proposed', 'awaiting_clarification', 'awaiting_confirmation'); + +CREATE INDEX IF NOT EXISTS idx_intent_plans_retention + ON intent_plans (updated_at) + WHERE status IN ('completed', 'failed', 'cancelled', 'expired', 'superseded'); + +CREATE TABLE IF NOT EXISTS intent_plan_steps ( + step_id TEXT PRIMARY KEY, + plan_id TEXT NOT NULL REFERENCES intent_plans(plan_id) ON DELETE CASCADE, + ordinal INT NOT NULL, + action_key TEXT NOT NULL, + slots JSONB NOT NULL DEFAULT '{}'::jsonb, + depends_on TEXT[] NOT NULL DEFAULT '{}', + status TEXT NOT NULL, + expected_resource_version TEXT NOT NULL DEFAULT '', + action_digest TEXT NOT NULL DEFAULT '', + idempotency_key TEXT NOT NULL DEFAULT '', + attempt INT NOT NULL DEFAULT 0, + result_ref TEXT NOT NULL DEFAULT '', + result_digest TEXT NOT NULL DEFAULT '', + safe_summary TEXT NOT NULL DEFAULT '', + lease_owner TEXT NOT NULL DEFAULT '', + lease_until TIMESTAMPTZ, + fencing_token BIGINT NOT NULL DEFAULT 0, + version BIGINT NOT NULL DEFAULT 1, + execution_mode TEXT NOT NULL DEFAULT 'validated_only', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_intent_plan_step_status CHECK (status IN ( + 'pending', + 'eligible', + 'running', + 'succeeded', + 'failed', + 'denied', + 'conflicted', + 'blocked_dependency', + 'skipped', + 'cancelled' + )), + CONSTRAINT chk_intent_plan_step_execution_mode CHECK (execution_mode IN ( + 'validated_only', + 'live' + )) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_intent_plan_steps_ordinal + ON intent_plan_steps (plan_id, ordinal); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_intent_plan_steps_idempotency + ON intent_plan_steps (idempotency_key) + WHERE idempotency_key <> ''; + +CREATE INDEX IF NOT EXISTS idx_intent_plan_steps_claim + ON intent_plan_steps (status, lease_until); + +CREATE TABLE IF NOT EXISTS intent_plan_decisions ( + decision_id TEXT PRIMARY KEY, + plan_id TEXT NOT NULL REFERENCES intent_plans(plan_id) ON DELETE CASCADE, + nonce TEXT NOT NULL, + plan_revision BIGINT NOT NULL, + actor_user_id BIGINT NOT NULL, + kind TEXT NOT NULL, + step_ids TEXT[] NOT NULL DEFAULT '{}', + exclude_unselected BOOLEAN NOT NULL DEFAULT FALSE, + decision_digest TEXT NOT NULL, + outcome TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT chk_intent_plan_decision_kind CHECK (kind IN ( + 'confirm', + 'cancel', + 'exclude', + 'clarify', + 'edit', + 'confirm_retry', + 'continue', + 'replace' + )) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_intent_plan_decisions_nonce + ON intent_plan_decisions (plan_id, nonce); + +CREATE INDEX IF NOT EXISTS idx_intent_plan_decisions_plan + ON intent_plan_decisions (plan_id, created_at); + +-- +goose Down +DROP INDEX IF EXISTS idx_intent_plan_decisions_plan; +DROP INDEX IF EXISTS uq_intent_plan_decisions_nonce; +DROP TABLE IF EXISTS intent_plan_decisions; +DROP INDEX IF EXISTS idx_intent_plan_steps_claim; +DROP INDEX IF EXISTS uq_intent_plan_steps_idempotency; +DROP INDEX IF EXISTS uq_intent_plan_steps_ordinal; +DROP TABLE IF EXISTS intent_plan_steps; +DROP INDEX IF EXISTS idx_intent_plans_retention; +DROP INDEX IF EXISTS idx_intent_plans_expire; +DROP INDEX IF EXISTS idx_intent_plans_active_scope; +DROP INDEX IF EXISTS uq_intent_plans_one_active_write; +DROP INDEX IF EXISTS uq_intent_plans_turn; +DROP TABLE IF EXISTS intent_plans; diff --git a/internal/app/app.go b/internal/app/app.go index 27fe690c..9002082d 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -11,6 +11,7 @@ import ( "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/app/docflow" "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/app/uow" "chat-agent-system/internal/artifactvault" @@ -43,33 +44,34 @@ const defaultReminderPollInterval = 15 * time.Minute // Runtime is the unique composition root. cmd/server, tests and workers // construct it once; HTTP adapters must not re-wire domain services. type Runtime struct { - Cfg *config.Config - Biz *bizdev.Service - Engine *coordinator.CoordinatorEngine - Auth *auth.Service - PW *projectworkflow.Service - TaskRuns *taskrun.Manager - Authorizer *authz.Authorizer - Authz *authz.Service - WFDef *workflowdef.Service - Inst *projectinstance.Service - Notify *notification.Service - Resolver *workspace.Resolver - WSCap *workspace.Capability - Commands *taskrun.Commands - Rec intent.Recognizer - Stageflow *stageflow.UseCase - DocNotify *docnotify.Subscriber - ChatLog chatflow.ChatLog - Assembler chatflow.ContextAssembler - Docflow *docflow.Commands - Pool *pgxpool.Pool - Gov *docgov.GovernanceEngine - Rev *docrevision.Service - Ctx *ctxmgr.ContextManager - CAS *storage.CAS - FreeDoc *freedoc.Service - KB *kb.Client + Cfg *config.Config + Biz *bizdev.Service + Engine *coordinator.CoordinatorEngine + Auth *auth.Service + PW *projectworkflow.Service + TaskRuns *taskrun.Manager + Authorizer *authz.Authorizer + Authz *authz.Service + WFDef *workflowdef.Service + Inst *projectinstance.Service + Notify *notification.Service + Resolver *workspace.Resolver + WSCap *workspace.Capability + Commands *taskrun.Commands + Rec intent.Recognizer + Stageflow *stageflow.UseCase + DocNotify *docnotify.Subscriber + ChatLog chatflow.ChatLog + Assembler chatflow.ContextAssembler + Docflow *docflow.Commands + IntentPlans *intentflow.Plans + Pool *pgxpool.Pool + Gov *docgov.GovernanceEngine + Rev *docrevision.Service + Ctx *ctxmgr.ContextManager + CAS *storage.CAS + FreeDoc *freedoc.Service + KB *kb.Client mu sync.Mutex running bool @@ -190,6 +192,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { if pool != nil { rt.Docflow.UoW = uow.NewPG(pool) rt.Docflow.Events = events.PGSink{} + rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{}) } svc.SetAgentMaterializer(rt.Docflow) rt.bindDocumentNotify() diff --git a/internal/app/intentflow/doc.go b/internal/app/intentflow/doc.go index e02249e1..a2367607 100644 --- a/internal/app/intentflow/doc.go +++ b/internal/app/intentflow/doc.go @@ -9,5 +9,6 @@ // - Confirmation is not authorization; each step re-checks the current principal. // - PostgreSQL plan/step/decision rows are authoritative; cards and SSE are projections. // - internal/intent stays a pure rule module (intent -> maturity). -// - This package must not import docgov, taskrun, coordinator, server, or pgx. +// - This package must not import docgov, taskrun, coordinator, or server. +// - Only the owner repository unwraps pgx via app/uow (see unwrap_owners). package intentflow diff --git a/internal/app/intentflow/errors.go b/internal/app/intentflow/errors.go index 6a7ccfc2..e33ebf7b 100644 --- a/internal/app/intentflow/errors.go +++ b/internal/app/intentflow/errors.go @@ -14,6 +14,8 @@ var ( ErrUnauthenticated = errors.New("ERR_UNAUTHENTICATED") ) +var errTurnExists = errors.New("intentflow: turn exists") + // ErrorCode returns the stable machine code for a sentinel, or empty. func ErrorCode(err error) string { switch { diff --git a/internal/app/intentflow/memory.go b/internal/app/intentflow/memory.go new file mode 100644 index 00000000..547ea47b --- /dev/null +++ b/internal/app/intentflow/memory.go @@ -0,0 +1,322 @@ +package intentflow + +import ( + "context" + "fmt" + "slices" + "sync" + "time" + + "chat-agent-system/internal/app/uow" +) + +type serialScope struct{ ctx context.Context } + +func (s serialScope) Context() context.Context { return s.ctx } + +// SerialUoW serializes commands. Memory tests use it so CAS races have one winner. +type SerialUoW struct{ mu sync.Mutex } + +func (s *SerialUoW) Within(ctx context.Context, fn func(uow.TxScope) error) error { + if s == nil { + return fmt.Errorf("intentflow: serial uow required") + } + s.mu.Lock() + defer s.mu.Unlock() + return fn(serialScope{ctx: ctx}) +} + +type MemoryStore struct { + mu sync.Mutex + plans map[string]*Plan + turn map[string]string + decisions map[string]*Decision +} + +func NewMemoryStore() *MemoryStore { + return &MemoryStore{ + plans: map[string]*Plan{}, + turn: map[string]string{}, + decisions: map[string]*Decision{}, + } +} + +func turnKey(env, session, turn string) string { + return env + "\x1f" + session + "\x1f" + turn +} + +func nonceKey(planID, nonce string) string { + return planID + "\x1f" + nonce +} + +func (m *MemoryStore) InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error { + m.mu.Lock() + defer m.mu.Unlock() + if plan == nil || plan.PlanID == "" { + return fmt.Errorf("intentflow: plan required") + } + tk := turnKey(plan.Principal.Environment, plan.Principal.SessionID, plan.ClientTurnID) + if existingID, ok := m.turn[tk]; ok { + _ = existingID + return errTurnExists + } + if plan.WritePlan && isActiveStatus(plan.Status) { + for _, p := range m.plans { + if sameScope(p.Principal, plan.Principal) && p.WritePlan && isActiveStatus(p.Status) { + return uow.ErrBusy + } + } + } + cloned := clonePlan(plan) + m.plans[plan.PlanID] = cloned + m.turn[tk] = plan.PlanID + return nil +} + +func (m *MemoryStore) GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) { + m.mu.Lock() + defer m.mu.Unlock() + p := m.plans[planID] + if p == nil { + return nil, nil + } + return clonePlan(p), nil +} + +func (m *MemoryStore) GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) { + return m.GetPlan(ctx, scope, planID) +} + +func (m *MemoryStore) GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error) { + m.mu.Lock() + defer m.mu.Unlock() + id := m.turn[turnKey(env, sessionID, clientTurnID)] + if id == "" { + return nil, nil + } + return clonePlan(m.plans[id]), nil +} + +func (m *MemoryStore) GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) { + m.mu.Lock() + defer m.mu.Unlock() + for _, plan := range m.plans { + if sameScope(plan.Principal, p) && plan.WritePlan && isActiveStatus(plan.Status) { + return clonePlan(plan), nil + } + } + return nil, nil +} + +func (m *MemoryStore) GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) { + m.mu.Lock() + defer m.mu.Unlock() + var best *Plan + for _, plan := range m.plans { + if !sameScope(plan.Principal, p) || !isActiveStatus(plan.Status) { + continue + } + if best == nil || plan.UpdatedAt.After(best.UpdatedAt) { + cp := clonePlan(plan) + best = cp + } + } + return best, nil +} + +func (m *MemoryStore) CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (int64, error) { + m.mu.Lock() + defer m.mu.Unlock() + p := m.plans[planID] + if p == nil || p.Revision != expectedRev || !slices.Contains(from, p.Status) { + return 0, nil + } + p.Status = to + p.Revision++ + p.UpdatedAt = now + return p.Revision, nil +} + +func (m *MemoryStore) InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error { + m.mu.Lock() + defer m.mu.Unlock() + k := nonceKey(d.PlanID, d.Nonce) + if _, ok := m.decisions[k]; ok { + return uow.ErrDigestConflict + } + cp := *d + cp.StepIDs = slices.Clone(d.StepIDs) + m.decisions[k] = &cp + return nil +} + +func (m *MemoryStore) GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error) { + m.mu.Lock() + defer m.mu.Unlock() + d := m.decisions[nonceKey(planID, nonce)] + if d == nil { + return nil, nil + } + cp := *d + cp.StepIDs = slices.Clone(d.StepIDs) + return &cp, nil +} + +func (m *MemoryStore) ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error { + m.mu.Lock() + defer m.mu.Unlock() + p := m.plans[planID] + if p == nil { + return fmt.Errorf("intentflow: plan %s not found", planID) + } + p.Steps = cloneSteps(steps) + p.UpdatedAt = time.Now().UTC() + return nil +} + +func (m *MemoryStore) ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error) { + m.mu.Lock() + defer m.mu.Unlock() + p := m.plans[planID] + if p == nil { + return nil, 0, nil + } + for i := range p.Steps { + s := &p.Steps[i] + if s.StepID != stepID { + continue + } + claimable := s.Status == StepEligible || + (s.Status == StepRunning && (s.LeaseUntil == nil || s.LeaseUntil.Before(now))) + if !claimable { + return nil, 0, nil + } + p.LastFencingToken++ + s.Status = StepRunning + s.LeaseOwner = worker + untilCopy := until + s.LeaseUntil = &untilCopy + s.Attempt++ + s.Version++ + s.FencingToken = p.LastFencingToken + if s.IdempotencyKey == "" { + s.IdempotencyKey = idempotencyKey + } + s.UpdatedAt = now + p.UpdatedAt = now + return cloneStep(s), s.FencingToken, nil + } + return nil, 0, nil +} + +func (m *MemoryStore) WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + p := m.plans[step.PlanID] + if p == nil { + return false, nil + } + for i := range p.Steps { + s := &p.Steps[i] + if s.StepID != step.StepID { + continue + } + if s.Status == StepSucceeded && s.ResultDigest == step.ResultDigest { + return true, nil + } + if s.Status != StepRunning || s.LeaseOwner != worker || s.FencingToken != token { + return false, nil + } + s.Status = StepSucceeded + s.ResultRef = step.ResultRef + s.ResultDigest = step.ResultDigest + s.SafeSummary = step.SafeSummary + s.LeaseOwner = "" + s.LeaseUntil = nil + s.Version++ + s.UpdatedAt = now + p.UpdatedAt = now + return true, nil + } + return false, nil +} + +func (m *MemoryStore) ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error) { + m.mu.Lock() + defer m.mu.Unlock() + n := 0 + for _, p := range m.plans { + if !isWaitingStatus(p.Status) || !now.After(p.ExpiresAt) { + continue + } + p.Status = PlanExpired + p.Revision++ + p.UpdatedAt = now + for i := range p.Steps { + if p.Steps[i].Status == StepPending || p.Steps[i].Status == StepEligible { + p.Steps[i].Status = StepCancelled + p.Steps[i].UpdatedAt = now + } + } + n++ + } + return n, nil +} + +func (m *MemoryStore) PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error) { + m.mu.Lock() + defer m.mu.Unlock() + n := 0 + for id, p := range m.plans { + switch p.Status { + case PlanCompleted, PlanFailed, PlanCancelled, PlanExpired, PlanSuperseded: + default: + continue + } + if !p.UpdatedAt.Before(before) { + continue + } + delete(m.turn, turnKey(p.Principal.Environment, p.Principal.SessionID, p.ClientTurnID)) + for k, d := range m.decisions { + if d.PlanID == id { + delete(m.decisions, k) + } + } + delete(m.plans, id) + n++ + } + return n, nil +} + +func sameScope(a, b Principal) bool { + return a.Environment == b.Environment && a.ProjectID == b.ProjectID && a.SessionID == b.SessionID && a.UserID == b.UserID +} + +func clonePlan(p *Plan) *Plan { + if p == nil { + return nil + } + cp := *p + cp.DisplaySnapshot = slices.Clone(p.DisplaySnapshot) + cp.Steps = cloneSteps(p.Steps) + return &cp +} + +func cloneSteps(in []Step) []Step { + out := make([]Step, len(in)) + for i := range in { + out[i] = *cloneStep(&in[i]) + } + return out +} + +func cloneStep(s *Step) *Step { + cp := *s + cp.Slots = slices.Clone(s.Slots) + cp.DependsOn = slices.Clone(s.DependsOn) + if s.LeaseUntil != nil { + t := *s.LeaseUntil + cp.LeaseUntil = &t + } + return &cp +} diff --git a/internal/app/intentflow/owner.go b/internal/app/intentflow/owner.go new file mode 100644 index 00000000..c5e1fd5c --- /dev/null +++ b/internal/app/intentflow/owner.go @@ -0,0 +1,88 @@ +package intentflow + +import ( + "context" + "sync" +) + +// ActionOwner executes one step. Old fencing tokens must be rejected here, +// not only in the plan store. The owner also stores receipts by idempotency key. +type ActionOwner interface { + Execute(ctx context.Context, exec StepExec) (Receipt, error) + LookupReceipt(ctx context.Context, idempotencyKey string) (Receipt, bool, error) +} + +// FenceObserver is optional; ClaimStep notifies it so fakes can reject stale tokens. +type FenceObserver interface { + ObserveFence(stepID string, token int64) +} + +// FakeActionOwner is the U3 test double. It records executions and receipts +// without touching business tables. +type FakeActionOwner struct { + mu sync.Mutex + execCount map[string]int + receipts map[string]Receipt + fences map[string]int64 + FailWrite bool +} + +func NewFakeActionOwner() *FakeActionOwner { + return &FakeActionOwner{ + execCount: map[string]int{}, + receipts: map[string]Receipt{}, + fences: map[string]int64{}, + } +} + +func (f *FakeActionOwner) ObserveFence(stepID string, token int64) { + if f == nil { + return + } + f.mu.Lock() + defer f.mu.Unlock() + f.fences[stepID] = token +} + +func (f *FakeActionOwner) Execute(ctx context.Context, exec StepExec) (Receipt, error) { + if f == nil { + return Receipt{}, ErrPlanActionForbidden + } + f.mu.Lock() + defer f.mu.Unlock() + if cur, ok := f.fences[exec.StepID]; ok && exec.FencingToken != cur { + return Receipt{}, ErrPlanActionVersionConflict + } + if existing, ok := f.receipts[exec.IdempotencyKey]; ok { + if existing.ActionDigest != exec.ActionDigest { + return Receipt{}, ErrPlanDecisionConflict + } + return existing, nil + } + f.execCount[exec.IdempotencyKey]++ + rec := Receipt{ + IdempotencyKey: exec.IdempotencyKey, + ActionDigest: exec.ActionDigest, + ResultRef: "fake:" + exec.StepID, + ResultDigest: "digest:" + exec.ActionDigest, + SafeSummary: "validated_only", + } + f.receipts[exec.IdempotencyKey] = rec + return rec, nil +} + +func (f *FakeActionOwner) LookupReceipt(ctx context.Context, idempotencyKey string) (Receipt, bool, error) { + if f == nil { + return Receipt{}, false, nil + } + f.mu.Lock() + defer f.mu.Unlock() + rec, ok := f.receipts[idempotencyKey] + return rec, ok, nil +} + +func (f *FakeActionOwner) ExecCount(idempotencyKey string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.execCount[idempotencyKey] +} diff --git a/internal/app/intentflow/pg_test.go b/internal/app/intentflow/pg_test.go new file mode 100644 index 00000000..addc9557 --- /dev/null +++ b/internal/app/intentflow/pg_test.go @@ -0,0 +1,113 @@ +package intentflow_test + +import ( + "context" + "errors" + "os" + "testing" + "time" + + "chat-agent-system/internal/app/intentflow" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/db" + + "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 + } + } +} + +func TestPlansPG_CreateClaimAndFence(t *testing.T) { + if testing.Short() { + t.Skip("intentflow PG is not part of -short") + } + dsn := db.SkipIfNoTestDatabaseURL(t) + db.RefuseSharedEnvDSN(t, dsn) + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { + t.Fatalf("migrate: %v", err) + } + + now := time.Date(2026, 9, 15, 15, 0, 0, 0, time.UTC) + clock := now + owner := intentflow.NewFakeActionOwner() + plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{ + Owner: owner, + Now: func() time.Time { return clock }, + Lease: 5 * time.Second, + }) + prin := intentflow.Principal{UserID: 9, Environment: "ci", ProjectID: 1001, SessionID: "sess-pg-u3"} + created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{ + Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1", + WritePlan: true, + Steps: []intentflow.StepDraft{ + {ActionKey: "regulation.query", ActionDigest: "q"}, + {ActionKey: "task.run", ActionDigest: "w"}, + }, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID) + }) + again, err := plans.CreatePlan(ctx, intentflow.CreateCommand{ + Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1", + WritePlan: true, + Steps: []intentflow.StepDraft{ + {ActionKey: "regulation.query", ActionDigest: "q"}, + {ActionKey: "task.run", ActionDigest: "w"}, + }, + }) + if err != nil || again.PlanID != created.PlanID { + t.Fatalf("idempotent create: %v %+v", err, again) + } + confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision, + DecisionNonce: "pg-n", DecisionDigest: "pg-dec", Kind: intentflow.DecisionConfirm, + }) + if err != nil { + t.Fatal(err) + } + a, err := plans.ClaimStep(ctx, "pg-a", created.PlanID, confirmed.Steps[0].StepID) + if err != nil || a == nil { + t.Fatalf("claim a: %v", err) + } + b, err := plans.ClaimStep(ctx, "pg-b", created.PlanID, confirmed.Steps[0].StepID) + if err != nil { + t.Fatal(err) + } + if b != nil { + t.Fatal("second worker must not own a live lease") + } + clock = clock.Add(time.Minute) + c, err := plans.ClaimStep(ctx, "pg-c", created.PlanID, confirmed.Steps[0].StepID) + if err != nil || c == nil { + t.Fatalf("takeover: %v", err) + } + if _, err := plans.ExecuteClaimed(ctx, a); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) { + t.Fatalf("old fence: %v", err) + } + if _, err := plans.ExecuteClaimed(ctx, c); err != nil { + t.Fatal(err) + } + other := prin + other.Environment = "dev" + if _, err := plans.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) { + t.Fatalf("env isolation: %v", err) + } +} diff --git a/internal/app/intentflow/plans.go b/internal/app/intentflow/plans.go new file mode 100644 index 00000000..552ec322 --- /dev/null +++ b/internal/app/intentflow/plans.go @@ -0,0 +1,563 @@ +package intentflow + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "slices" + "time" + + "chat-agent-system/internal/app/uow" +) + +const ( + defaultLeaseTTL = 15 * time.Second + defaultPlanTTL = 30 * time.Minute +) + +// Plans is the owner implementation of Service plus worker/recovery methods. +type Plans struct { + uow uow.UnitOfWork + store Store + owner ActionOwner + now func() time.Time + newID func(prefix string) string + lease time.Duration + ttl time.Duration +} + +type Options struct { + Owner ActionOwner + Now func() time.Time + NewID func(prefix string) string + Lease time.Duration + PlanTTL time.Duration +} + +func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans { + now := opt.Now + if now == nil { + now = func() time.Time { return time.Now().UTC() } + } + newID := opt.NewID + if newID == nil { + newID = randomID + } + lease := opt.Lease + if lease <= 0 { + lease = defaultLeaseTTL + } + ttl := opt.PlanTTL + if ttl <= 0 { + ttl = defaultPlanTTL + } + return &Plans{uow: unit, store: store, owner: opt.Owner, now: now, newID: newID, lease: lease, ttl: ttl} +} + +func randomID(prefix string) string { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return prefix + fmt.Sprintf("%d", time.Now().UnixNano()) + } + return prefix + hex.EncodeToString(b[:]) +} + +func requirePrincipal(p Principal) error { + if p.UserID <= 0 || p.ProjectID <= 0 || p.SessionID == "" || p.Environment == "" { + return ErrUnauthenticated + } + return nil +} + +func visibleTo(plan *Plan, p Principal) bool { + return plan != nil && sameScope(plan.Principal, p) +} + +// HandleTurn stays a no-op while MULTI_INTENT_PLAN_TIER is legacy/shadow (default). +// U7 will map utterances onto CreatePlan; chat must not grow a second owner. +func (p *Plans) HandleTurn(ctx context.Context, in TurnInput) (*TurnResult, error) { + if err := requirePrincipal(in.Principal); err != nil { + return nil, err + } + return &TurnResult{}, nil +} + +func (p *Plans) GetPlan(ctx context.Context, principal Principal, planID string) (*PlanView, error) { + if err := requirePrincipal(principal); err != nil { + return nil, err + } + var view *PlanView + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + plan, err := p.store.GetPlan(ctx, scope, planID) + if err != nil { + return err + } + if !visibleTo(plan, principal) { + return ErrPlanNotFoundOrHidden + } + view = toView(plan) + return nil + }) + return view, err +} + +func (p *Plans) GetActivePlan(ctx context.Context, principal Principal) (*PlanView, error) { + if err := requirePrincipal(principal); err != nil { + return nil, err + } + var view *PlanView + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + plan, err := p.store.GetActive(ctx, scope, principal) + if err != nil { + return err + } + if plan == nil { + return nil + } + view = toView(plan) + return nil + }) + return view, err +} + +func (p *Plans) CreatePlan(ctx context.Context, in CreateCommand) (*PlanView, error) { + if err := requirePrincipal(in.Principal); err != nil { + return nil, err + } + if in.ClientTurnID == "" || in.RequestDigest == "" { + return nil, ErrPlanDecisionConflict + } + if len(in.Steps) == 0 || len(in.Steps) > 4 { + return nil, ErrPlanDependencyInvalid + } + now := p.now() + status := in.Status + if status == "" { + status = PlanAwaitingConfirmation + } + mode := in.ExecutionMode + if mode == "" { + mode = ModeValidatedOnly + } + expires := in.ExpiresAt + if expires.IsZero() { + expires = now.Add(p.ttl) + } + seg := in.SegmenterVersion + if seg == "" { + seg = SegmenterVersionUnknown + } + pol := in.PolicyVersion + if pol == "" { + pol = PolicyVersion + } + + var view *PlanView + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + existing, err := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID) + if err != nil { + return err + } + if existing != nil { + if existing.RequestDigest != in.RequestDigest { + return ErrPlanDecisionConflict + } + view = toView(existing) + return nil + } + if in.WritePlan { + active, err := p.store.GetActiveWrite(ctx, scope, in.Principal) + if err != nil { + return err + } + if active != nil { + view = toView(active) + return ErrPlanDecisionConflict + } + } + plan := &Plan{ + PlanID: p.newID("pln_"), + Principal: in.Principal, + ClientTurnID: in.ClientTurnID, + RequestDigest: in.RequestDigest, + SegmenterVersion: seg, + PolicyVersion: pol, + Status: status, + Revision: 1, + ExpiresAt: expires, + WritePlan: in.WritePlan, + ExecutionMode: mode, + SafeSummary: in.SafeSummary, + CreatedAt: now, + UpdatedAt: now, + } + for i, d := range in.Steps { + mode := d.ExecutionMode + if mode == "" { + mode = plan.ExecutionMode + } + slots := d.Slots + if len(slots) == 0 { + slots = []byte("{}") + } + plan.Steps = append(plan.Steps, Step{ + PlanID: plan.PlanID, + StepID: p.newID("stp_"), + Ordinal: i + 1, + ActionKey: d.ActionKey, + Slots: slots, + DependsOn: slices.Clone(d.DependsOn), + Status: StepPending, + ExpectedResourceVersion: d.ExpectedResourceVersion, + ActionDigest: d.ActionDigest, + SafeSummary: d.SafeSummary, + Version: 1, + ExecutionMode: mode, + CreatedAt: now, + UpdatedAt: now, + }) + } + if err := p.store.InsertPlan(ctx, scope, plan); err != nil { + if errors.Is(err, errTurnExists) { + again, loadErr := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID) + if loadErr != nil { + return loadErr + } + if again != nil && again.RequestDigest == in.RequestDigest { + view = toView(again) + return nil + } + return ErrPlanDecisionConflict + } + if errors.Is(err, uow.ErrBusy) { + active, loadErr := p.store.GetActiveWrite(ctx, scope, in.Principal) + if loadErr != nil { + return loadErr + } + if active != nil { + view = toView(active) + } + return ErrPlanDecisionConflict + } + return err + } + view = toView(plan) + return nil + }) + return view, err +} + +func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView, error) { + if err := requirePrincipal(in.Principal); err != nil { + return nil, err + } + if in.PlanID == "" || in.DecisionNonce == "" || in.DecisionDigest == "" { + return nil, ErrPlanDecisionConflict + } + now := p.now() + var view *PlanView + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + plan, err := p.store.GetPlanForUpdate(ctx, scope, in.PlanID) + if err != nil { + return err + } + if !visibleTo(plan, in.Principal) { + return ErrPlanNotFoundOrHidden + } + existing, err := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce) + if err != nil { + return err + } + if existing != nil { + if existing.DecisionDigest != in.DecisionDigest { + return ErrPlanDecisionConflict + } + fresh, err := p.store.GetPlan(ctx, scope, in.PlanID) + if err != nil { + return err + } + view = toView(fresh) + return nil + } + if isWaitingStatus(plan.Status) && !now.Before(plan.ExpiresAt) { + if _, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, waitingStatuses(), PlanExpired, now); err != nil { + return err + } + cancelUnstarted(plan, now) + if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil { + return err + } + return ErrPlanExpired + } + if in.ExpectedRevision != plan.Revision { + return ErrPlanStaleRevision + } + dec := &Decision{ + DecisionID: p.newID("dec_"), + PlanID: plan.PlanID, + Nonce: in.DecisionNonce, + PlanRevision: plan.Revision, + ActorUserID: in.Principal.UserID, + Kind: in.Kind, + StepIDs: slices.Clone(in.StepIDs), + ExcludeUnselected: in.ExcludeUnselected, + DecisionDigest: in.DecisionDigest, + CreatedAt: now, + } + if err := p.store.InsertDecision(ctx, scope, dec); err != nil { + if errors.Is(err, uow.ErrDigestConflict) || errors.Is(err, errTurnExists) { + again, loadErr := p.store.GetDecisionByNonce(ctx, scope, in.PlanID, in.DecisionNonce) + if loadErr != nil { + return loadErr + } + if again != nil && again.DecisionDigest == in.DecisionDigest { + fresh, err := p.store.GetPlan(ctx, scope, in.PlanID) + if err != nil { + return err + } + view = toView(fresh) + return nil + } + return ErrPlanDecisionConflict + } + return err + } + from := []PlanStatus{plan.Status} + var to PlanStatus + switch in.Kind { + case DecisionConfirm: + if plan.Status != PlanAwaitingConfirmation && plan.Status != PlanAwaitingClarification && plan.Status != PlanProposed { + return ErrPlanStaleRevision + } + if err := applyConfirm(plan, in.StepIDs, in.ExcludeUnselected, now); err != nil { + return err + } + to = PlanExecuting + case DecisionCancel: + if !isWaitingStatus(plan.Status) && plan.Status != PlanExecuting && plan.Status != PlanPartiallyCompleted { + return ErrPlanStaleRevision + } + cancelUnstarted(plan, now) + to = PlanCancelled + default: + return ErrPlanDependencyInvalid + } + newRev, err := p.store.CASPlan(ctx, scope, plan.PlanID, plan.Revision, from, to, now) + if err != nil { + return err + } + if newRev == 0 { + return ErrPlanStaleRevision + } + if err := p.store.ReplaceSteps(ctx, scope, plan.PlanID, plan.Steps); err != nil { + return err + } + plan.Status = to + plan.Revision = newRev + view = toView(plan) + return nil + }) + return view, err +} + +func applyConfirm(plan *Plan, selected []string, excludeUnselected bool, now time.Time) error { + byID := map[string]*Step{} + for i := range plan.Steps { + byID[plan.Steps[i].StepID] = &plan.Steps[i] + } + chosen := map[string]bool{} + if len(selected) == 0 { + for i := range plan.Steps { + chosen[plan.Steps[i].StepID] = true + } + } else { + for _, id := range selected { + if _, ok := byID[id]; !ok { + return ErrPlanDependencyInvalid + } + chosen[id] = true + } + for id := range chosen { + for _, dep := range byID[id].DependsOn { + if !chosen[dep] && byID[dep].Status != StepSucceeded && byID[dep].Status != StepSkipped { + return ErrPlanDependencyInvalid + } + } + } + } + for i := range plan.Steps { + s := &plan.Steps[i] + if chosen[s.StepID] { + if s.Status == StepPending || s.Status == StepEligible { + s.Status = StepEligible + s.UpdatedAt = now + } + continue + } + if excludeUnselected && (s.Status == StepPending || s.Status == StepEligible) { + s.Status = StepCancelled + s.UpdatedAt = now + } + } + return nil +} + +func cancelUnstarted(plan *Plan, now time.Time) { + for i := range plan.Steps { + s := &plan.Steps[i] + if s.Status == StepPending || s.Status == StepEligible { + s.Status = StepCancelled + s.UpdatedAt = now + } + } +} + +func (p *Plans) ClaimStep(ctx context.Context, workerID, planID, stepID string) (*ClaimedStep, error) { + if workerID == "" || planID == "" || stepID == "" { + return nil, ErrPlanDependencyInvalid + } + now := p.now() + until := now.Add(p.lease) + var out *ClaimedStep + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + plan, err := p.store.GetPlanForUpdate(ctx, scope, planID) + if err != nil { + return err + } + if plan == nil { + return ErrPlanNotFoundOrHidden + } + var digest string + for _, s := range plan.Steps { + if s.StepID == stepID { + digest = s.ActionDigest + break + } + } + key := fmt.Sprintf("%s:%d:%s:%s", planID, plan.Revision, stepID, digest) + step, token, err := p.store.ClaimStep(ctx, scope, planID, stepID, workerID, until, now, key) + if err != nil { + return err + } + if step == nil { + return nil + } + if obs, ok := p.owner.(FenceObserver); ok { + obs.ObserveFence(step.StepID, token) + } + out = &ClaimedStep{Plan: *plan, Step: *step, WorkerID: workerID, FencingToken: token, IdempotencyKey: step.IdempotencyKey} + return nil + }) + return out, err +} + +func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) { + if claimed == nil { + return nil, ErrPlanNotFoundOrHidden + } + if p.owner == nil { + return nil, ErrPlanActionForbidden + } + rec, err := p.owner.Execute(ctx, StepExec{ + Principal: claimed.Plan.Principal, + PlanID: claimed.Plan.PlanID, + StepID: claimed.Step.StepID, + ActionKey: claimed.Step.ActionKey, + IdempotencyKey: claimed.IdempotencyKey, + ActionDigest: claimed.Step.ActionDigest, + ExpectedResourceVersion: claimed.Step.ExpectedResourceVersion, + FencingToken: claimed.FencingToken, + }) + if err != nil { + return nil, err + } + return p.writeReceipt(ctx, claimed, rec) +} + +func (p *Plans) RecoverReceipt(ctx context.Context, claimed *ClaimedStep) (*PlanView, error) { + if claimed == nil || p.owner == nil { + return nil, ErrPlanNotFoundOrHidden + } + rec, ok, err := p.owner.LookupReceipt(ctx, claimed.IdempotencyKey) + if err != nil { + return nil, err + } + if !ok { + return nil, ErrPlanActionVersionConflict + } + return p.writeReceipt(ctx, claimed, rec) +} + +func (p *Plans) writeReceipt(ctx context.Context, claimed *ClaimedStep, rec Receipt) (*PlanView, error) { + now := p.now() + var view *PlanView + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + ok, err := p.store.WriteStepReceipt(ctx, scope, claimed.WorkerID, claimed.FencingToken, Step{ + PlanID: claimed.Plan.PlanID, + StepID: claimed.Step.StepID, + ResultRef: rec.ResultRef, + ResultDigest: rec.ResultDigest, + SafeSummary: rec.SafeSummary, + }, now) + if err != nil { + return err + } + if !ok { + return ErrPlanActionVersionConflict + } + plan, err := p.store.GetPlan(ctx, scope, claimed.Plan.PlanID) + if err != nil { + return err + } + view = toView(plan) + return nil + }) + return view, err +} + +func (p *Plans) ExpireDue(ctx context.Context) (int, error) { + var n int + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + var err error + n, err = p.store.ExpireWaiting(ctx, scope, p.now()) + return err + }) + return n, err +} + +func (p *Plans) PurgeClosed(ctx context.Context, olderThan time.Time) (int, error) { + var n int + err := p.uow.Within(ctx, func(scope uow.TxScope) error { + var err error + n, err = p.store.PurgeClosed(ctx, scope, olderThan) + return err + }) + return n, err +} + +func toView(plan *Plan) *PlanView { + if plan == nil { + return nil + } + v := &PlanView{ + PlanID: plan.PlanID, + Revision: plan.Revision, + Status: plan.Status, + ExecutionMode: plan.ExecutionMode, + ExpiresAt: plan.ExpiresAt, + SafeSummary: plan.SafeSummary, + } + for _, s := range plan.Steps { + v.Steps = append(v.Steps, StepView{ + StepID: s.StepID, + ActionKey: s.ActionKey, + Status: s.Status, + ExecutionMode: s.ExecutionMode, + DependsOn: slices.Clone(s.DependsOn), + SafeSummary: s.SafeSummary, + }) + } + return v +} diff --git a/internal/app/intentflow/plans_test.go b/internal/app/intentflow/plans_test.go new file mode 100644 index 00000000..708936a0 --- /dev/null +++ b/internal/app/intentflow/plans_test.go @@ -0,0 +1,321 @@ +package intentflow_test + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "chat-agent-system/internal/app/intentflow" +) + +func testClock() (now *time.Time, opts intentflow.Options, owner *intentflow.FakeActionOwner) { + t0 := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC) + now = &t0 + owner = intentflow.NewFakeActionOwner() + opts = intentflow.Options{ + Owner: owner, + Now: func() time.Time { return now.UTC() }, + Lease: 10 * time.Second, + PlanTTL: 30 * time.Minute, + } + return now, opts, owner +} + +func testPlans() (*intentflow.Plans, *time.Time, *intentflow.FakeActionOwner) { + now, opts, owner := testClock() + return intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), opts), now, owner +} + +func principal() intentflow.Principal { + return intentflow.Principal{UserID: 7, Environment: "dev", ProjectID: 42, SessionID: "sess-u3"} +} + +func draft(digest string) intentflow.CreateCommand { + return intentflow.CreateCommand{ + Principal: principal(), + ClientTurnID: "turn-1", + RequestDigest: digest, + WritePlan: true, + SafeSummary: "query then generate", + Steps: []intentflow.StepDraft{ + {ActionKey: "regulation.query", ActionDigest: "q1"}, + {ActionKey: "task.run", ActionDigest: "w1"}, + }, + } +} + +func TestHandleTurnLegacyDoesNotPersist(t *testing.T) { + p, _, _ := testPlans() + ctx := context.Background() + res, err := p.HandleTurn(ctx, intentflow.TurnInput{Principal: principal(), Text: "生成报价单", ClientTurnID: "turn-1", RequestDigest: "d1"}) + if err != nil { + t.Fatal(err) + } + if res == nil || res.Plan != nil { + t.Fatalf("legacy HandleTurn must not create a plan: %+v", res) + } + active, err := p.GetActivePlan(ctx, principal()) + if err != nil { + t.Fatal(err) + } + if active != nil { + t.Fatal("no plan should be visible") + } +} + +func TestCreateIdempotentSameDigest(t *testing.T) { + p, _, _ := testPlans() + ctx := context.Background() + a, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + b, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + if a.PlanID != b.PlanID { + t.Fatalf("same turn/digest must reuse plan %s vs %s", a.PlanID, b.PlanID) + } + other := draft("digest-b") + if _, err := p.CreatePlan(ctx, other); !errors.Is(err, intentflow.ErrPlanDecisionConflict) { + t.Fatalf("different digest must conflict, got %v", err) + } +} + +func TestCreateRejectsSecondActiveWritePlan(t *testing.T) { + p, _, _ := testPlans() + ctx := context.Background() + first, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + second := draft("digest-x") + second.ClientTurnID = "turn-2" + got, err := p.CreatePlan(ctx, second) + if !errors.Is(err, intentflow.ErrPlanDecisionConflict) { + t.Fatalf("want conflict, got %v", err) + } + if got == nil || got.PlanID != first.PlanID { + t.Fatalf("must return the active write plan, got %+v", got) + } +} + +func TestDecisionNonceSameAndDifferentDigest(t *testing.T) { + p, _, _ := testPlans() + ctx := context.Background() + created, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + cmd := intentflow.DecisionCommand{ + Principal: principal(), + PlanID: created.PlanID, + ExpectedRevision: created.Revision, + DecisionNonce: "nonce-1", + DecisionDigest: "dec-1", + Kind: intentflow.DecisionConfirm, + } + first, err := p.DecidePlan(ctx, cmd) + if err != nil { + t.Fatal(err) + } + again, err := p.DecidePlan(ctx, cmd) + if err != nil { + t.Fatal(err) + } + if first.Revision != again.Revision { + t.Fatal("same nonce/digest must return the same receipt") + } + cmd.DecisionDigest = "dec-other" + if _, err := p.DecidePlan(ctx, cmd); !errors.Is(err, intentflow.ErrPlanDecisionConflict) { + t.Fatalf("different digest must conflict, got %v", err) + } +} + +func TestConfirmCancelRaceOneWinner(t *testing.T) { + p, _, _ := testPlans() + ctx := context.Background() + created, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + var okConfirm, okCancel atomic.Int32 + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, err := p.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision, + DecisionNonce: "n-confirm", DecisionDigest: "d-confirm", Kind: intentflow.DecisionConfirm, + }) + if err == nil { + okConfirm.Add(1) + } + }() + go func() { + defer wg.Done() + _, err := p.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision, + DecisionNonce: "n-cancel", DecisionDigest: "d-cancel", Kind: intentflow.DecisionCancel, + }) + if err == nil { + okCancel.Add(1) + } + }() + wg.Wait() + if okConfirm.Load()+okCancel.Load() != 1 { + t.Fatalf("exactly one CAS winner, confirm=%d cancel=%d", okConfirm.Load(), okCancel.Load()) + } +} + +func TestTwoWorkersOneOwnerAndLeaseTakeover(t *testing.T) { + p, clock, owner := testPlans() + ctx := context.Background() + created, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision, + DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm, + }) + if err != nil { + t.Fatal(err) + } + stepID := confirmed.Steps[0].StepID + var gotA, gotB atomic.Int32 + var first *intentflow.ClaimedStep + var mu sync.Mutex + var wg sync.WaitGroup + wg.Add(2) + claim := func(worker string, slot *atomic.Int32) { + defer wg.Done() + c, err := p.ClaimStep(ctx, worker, created.PlanID, stepID) + if err != nil { + t.Errorf("claim %s: %v", worker, err) + return + } + if c != nil { + slot.Add(1) + mu.Lock() + if first == nil { + first = c + } + mu.Unlock() + } + } + go claim("worker-a", &gotA) + go claim("worker-b", &gotB) + wg.Wait() + if gotA.Load()+gotB.Load() != 1 { + t.Fatalf("two workers must not both own, a=%d b=%d", gotA.Load(), gotB.Load()) + } + stale := first + *clock = clock.Add(time.Minute) + takeover, err := p.ClaimStep(ctx, "worker-c", created.PlanID, stepID) + if err != nil || takeover == nil { + t.Fatalf("expired lease must be taken over: %v %+v", err, takeover) + } + if takeover.FencingToken <= stale.FencingToken { + t.Fatalf("fencing must be monotonic: old=%d new=%d", stale.FencingToken, takeover.FencingToken) + } + if _, err := p.ExecuteClaimed(ctx, stale); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) { + t.Fatalf("old fence must be rejected by owner, got %v", err) + } + if _, err := p.ExecuteClaimed(ctx, takeover); err != nil { + t.Fatal(err) + } + if owner.ExecCount(takeover.IdempotencyKey) != 1 { + t.Fatalf("execute count = %d", owner.ExecCount(takeover.IdempotencyKey)) + } +} + +func TestReceiptWritebackFailureDoesNotRerun(t *testing.T) { + p, _, owner := testPlans() + ctx := context.Background() + created, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision, + DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm, + }) + if err != nil { + t.Fatal(err) + } + claimed, err := p.ClaimStep(ctx, "w1", created.PlanID, confirmed.Steps[0].StepID) + if err != nil || claimed == nil { + t.Fatalf("claim: %v %+v", err, claimed) + } + if _, err := owner.Execute(ctx, intentflow.StepExec{ + Principal: principal(), PlanID: claimed.Plan.PlanID, StepID: claimed.Step.StepID, + ActionKey: claimed.Step.ActionKey, IdempotencyKey: claimed.IdempotencyKey, + ActionDigest: claimed.Step.ActionDigest, FencingToken: claimed.FencingToken, + }); err != nil { + t.Fatal(err) + } + if _, err := p.RecoverReceipt(ctx, claimed); err != nil { + t.Fatal(err) + } + if owner.ExecCount(claimed.IdempotencyKey) != 1 { + t.Fatalf("recover must not re-execute, count=%d", owner.ExecCount(claimed.IdempotencyKey)) + } + got, err := p.GetPlan(ctx, principal(), created.PlanID) + if err != nil { + t.Fatal(err) + } + if got.Steps[0].Status != intentflow.StepSucceeded { + t.Fatalf("step status %s", got.Steps[0].Status) + } +} + +func TestExpireAndEnvironmentIsolationAndRetention(t *testing.T) { + p, clock, _ := testPlans() + ctx := context.Background() + created, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + other := principal() + other.Environment = "test" + if _, err := p.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) { + t.Fatalf("cross-env must hide, got %v", err) + } + *clock = clock.Add(time.Hour) + n, err := p.ExpireDue(ctx) + if err != nil || n != 1 { + t.Fatalf("expire n=%d err=%v", n, err) + } + if _, err := p.DecidePlan(ctx, intentflow.DecisionCommand{ + Principal: principal(), PlanID: created.PlanID, ExpectedRevision: 1, + DecisionNonce: "late", DecisionDigest: "late", Kind: intentflow.DecisionConfirm, + }); !errors.Is(err, intentflow.ErrPlanExpired) && !errors.Is(err, intentflow.ErrPlanStaleRevision) { + got, _ := p.GetPlan(ctx, principal(), created.PlanID) + if got == nil || got.Status != intentflow.PlanExpired { + t.Fatalf("expired plan confirm: %v view=%+v", err, got) + } + } + purged, err := p.PurgeClosed(ctx, clock.Add(time.Second)) + if err != nil || purged != 1 { + t.Fatalf("purge n=%d err=%v", purged, err) + } + if _, err := p.GetPlan(ctx, principal(), created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) { + t.Fatalf("purged plan must be hidden, got %v", err) + } +} + +func TestUnauthenticatedRejected(t *testing.T) { + p, _, _ := testPlans() + _, err := p.CreatePlan(context.Background(), intentflow.CreateCommand{ + Principal: intentflow.Principal{}, ClientTurnID: "t", RequestDigest: "d", + Steps: []intentflow.StepDraft{{ActionKey: "regulation.query"}}, + }) + if !errors.Is(err, intentflow.ErrUnauthenticated) { + t.Fatalf("got %v", err) + } +} diff --git a/internal/app/intentflow/repository.go b/internal/app/intentflow/repository.go new file mode 100644 index 00000000..a4ca9e85 --- /dev/null +++ b/internal/app/intentflow/repository.go @@ -0,0 +1,486 @@ +package intentflow + +import ( + "context" + "encoding/json" + "errors" + "time" + + "chat-agent-system/internal/app/uow" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +// Repository is the PostgreSQL owner of intent_plans*. +type Repository struct{} + +func NewRepository() Repository { return Repository{} } + +func txFrom(scope uow.TxScope) (pgx.Tx, error) { + return uow.AsPG(scope) +} + +func isUniqueViolation(err error) bool { + var pgErr *pgconn.PgError + return errors.As(err, &pgErr) && pgErr.Code == "23505" +} + +func uniqueConstraint(err error) string { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return pgErr.ConstraintName + } + return "" +} + +func (Repository) InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error { + tx, err := txFrom(scope) + if err != nil { + return err + } + snap := plan.DisplaySnapshot + if len(snap) == 0 { + snap = nil + } + _, err = tx.Exec(ctx, ` +INSERT INTO intent_plans ( + plan_id, environment, project_id, session_id, user_id, client_turn_id, request_digest, + segmenter_version, policy_version, status, revision, expires_at, last_fencing_token, + write_plan, execution_mode, safe_summary, display_snapshot, created_at, updated_at +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19)`, + plan.PlanID, plan.Principal.Environment, plan.Principal.ProjectID, plan.Principal.SessionID, + plan.Principal.UserID, plan.ClientTurnID, plan.RequestDigest, plan.SegmenterVersion, + plan.PolicyVersion, string(plan.Status), plan.Revision, plan.ExpiresAt, plan.LastFencingToken, + plan.WritePlan, string(plan.ExecutionMode), plan.SafeSummary, jsonbOrNil(snap), + plan.CreatedAt, plan.UpdatedAt, + ) + if err != nil { + if isUniqueViolation(err) { + if uniqueConstraint(err) == "uq_intent_plans_one_active_write" { + return uow.ErrBusy + } + return errTurnExists + } + return err + } + for _, s := range plan.Steps { + if err := insertStep(ctx, tx, s); err != nil { + return err + } + } + return nil +} + +func insertStep(ctx context.Context, tx pgx.Tx, s Step) error { + slots := s.Slots + if len(slots) == 0 { + slots = []byte("{}") + } + _, err := tx.Exec(ctx, ` +INSERT INTO intent_plan_steps ( + step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version, + action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary, + lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)`, + s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, s.DependsOn, string(s.Status), + s.ExpectedResourceVersion, s.ActionDigest, s.IdempotencyKey, s.Attempt, s.ResultRef, + s.ResultDigest, s.SafeSummary, s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, + string(s.ExecutionMode), s.CreatedAt, s.UpdatedAt, + ) + return err +} + +func (r Repository) GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) { + return r.loadPlan(ctx, scope, planID, false) +} + +func (r Repository) GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) { + return r.loadPlan(ctx, scope, planID, true) +} + +func (Repository) loadPlan(ctx context.Context, scope uow.TxScope, planID string, forUpdate bool) (*Plan, error) { + tx, err := txFrom(scope) + if err != nil { + return nil, err + } + q := planSelect + " WHERE plan_id=$1" + if forUpdate { + q += " FOR UPDATE" + } + plan, err := scanPlan(tx.QueryRow(ctx, q, planID)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + steps, err := loadSteps(ctx, tx, plan.PlanID) + if err != nil { + return nil, err + } + plan.Steps = steps + return plan, nil +} + +func (Repository) GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error) { + tx, err := txFrom(scope) + if err != nil { + return nil, err + } + plan, err := scanPlan(tx.QueryRow(ctx, planSelect+` +WHERE environment=$1 AND session_id=$2 AND client_turn_id=$3`, env, sessionID, clientTurnID)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + steps, err := loadSteps(ctx, tx, plan.PlanID) + if err != nil { + return nil, err + } + plan.Steps = steps + return plan, nil +} + +func (Repository) GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) { + return loadActive(ctx, scope, p, true) +} + +func (Repository) GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) { + return loadActive(ctx, scope, p, false) +} + +func loadActive(ctx context.Context, scope uow.TxScope, p Principal, writeOnly bool) (*Plan, error) { + tx, err := txFrom(scope) + if err != nil { + return nil, err + } + q := planSelect + ` +WHERE environment=$1 AND project_id=$2 AND session_id=$3 AND user_id=$4 + AND status = ANY($5)` + args := []any{p.Environment, p.ProjectID, p.SessionID, p.UserID, activeStatusStrings()} + if writeOnly { + q += ` AND write_plan` + } + q += ` ORDER BY updated_at DESC LIMIT 1` + plan, err := scanPlan(tx.QueryRow(ctx, q, args...)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + steps, err := loadSteps(ctx, tx, plan.PlanID) + if err != nil { + return nil, err + } + plan.Steps = steps + return plan, nil +} + +func (Repository) CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (int64, error) { + tx, err := txFrom(scope) + if err != nil { + return 0, err + } + var newRev int64 + err = tx.QueryRow(ctx, ` +UPDATE intent_plans +SET status=$1, revision=revision+1, updated_at=$2 +WHERE plan_id=$3 AND revision=$4 AND status = ANY($5) +RETURNING revision`, string(to), now, planID, expectedRev, statusStrings(from)).Scan(&newRev) + if errors.Is(err, pgx.ErrNoRows) { + return 0, nil + } + return newRev, err +} + +func (Repository) InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error { + tx, err := txFrom(scope) + if err != nil { + return err + } + _, err = tx.Exec(ctx, ` +INSERT INTO intent_plan_decisions ( + decision_id, plan_id, nonce, plan_revision, actor_user_id, kind, step_ids, + exclude_unselected, decision_digest, outcome, created_at +) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, + d.DecisionID, d.PlanID, d.Nonce, d.PlanRevision, d.ActorUserID, string(d.Kind), + d.StepIDs, d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt, + ) + if isUniqueViolation(err) { + return uow.ErrDigestConflict + } + return err +} + +func (Repository) GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error) { + tx, err := txFrom(scope) + if err != nil { + return nil, err + } + var d Decision + var kind string + err = tx.QueryRow(ctx, ` +SELECT decision_id, plan_id, nonce, plan_revision, actor_user_id, kind, step_ids, + exclude_unselected, decision_digest, outcome, created_at +FROM intent_plan_decisions WHERE plan_id=$1 AND nonce=$2`, planID, nonce).Scan( + &d.DecisionID, &d.PlanID, &d.Nonce, &d.PlanRevision, &d.ActorUserID, &kind, + &d.StepIDs, &d.ExcludeUnselected, &d.DecisionDigest, &d.Outcome, &d.CreatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + d.Kind = DecisionKind(kind) + return &d, nil +} + +func (Repository) ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error { + tx, err := txFrom(scope) + if err != nil { + return err + } + for _, s := range steps { + if s.PlanID == "" { + s.PlanID = planID + } + _, err := tx.Exec(ctx, ` +UPDATE intent_plan_steps +SET status=$1, depends_on=$2, expected_resource_version=$3, action_digest=$4, + idempotency_key=$5, attempt=$6, result_ref=$7, result_digest=$8, safe_summary=$9, + lease_owner=$10, lease_until=$11, fencing_token=$12, version=$13, execution_mode=$14, + updated_at=$15, slots=$16 +WHERE step_id=$17 AND plan_id=$18`, + string(s.Status), s.DependsOn, s.ExpectedResourceVersion, s.ActionDigest, + s.IdempotencyKey, s.Attempt, s.ResultRef, s.ResultDigest, s.SafeSummary, + s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, string(s.ExecutionMode), + s.UpdatedAt, jsonbOrNil(s.Slots), s.StepID, planID, + ) + if err != nil { + return err + } + } + return nil +} + +func (Repository) ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error) { + tx, err := txFrom(scope) + if err != nil { + return nil, 0, err + } + var token int64 + err = tx.QueryRow(ctx, ` +UPDATE intent_plans +SET last_fencing_token = last_fencing_token + 1, updated_at = $2 +WHERE plan_id = $1 + AND EXISTS ( + SELECT 1 FROM intent_plan_steps s + WHERE s.plan_id = $1 AND s.step_id = $3 + AND ( + s.status = 'eligible' + OR (s.status = 'running' AND (s.lease_until IS NULL OR s.lease_until < $2)) + ) + ) +RETURNING last_fencing_token`, planID, now, stepID).Scan(&token) + if errors.Is(err, pgx.ErrNoRows) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + row := tx.QueryRow(ctx, ` +UPDATE intent_plan_steps +SET status='running', + lease_owner=$1, + lease_until=$2, + attempt=attempt+1, + version=version+1, + fencing_token=$3, + idempotency_key=CASE WHEN idempotency_key='' THEN $4 ELSE idempotency_key END, + updated_at=$5 +WHERE step_id=$6 AND plan_id=$7 + AND ( + status='eligible' + OR (status='running' AND (lease_until IS NULL OR lease_until < $5)) + ) +RETURNING `+stepColumns, worker, until, token, idempotencyKey, now, stepID, planID) + step, err := scanStep(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, 0, nil + } + return step, token, err +} + +func (Repository) WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error) { + tx, err := txFrom(scope) + if err != nil { + return false, err + } + tag, err := tx.Exec(ctx, ` +UPDATE intent_plan_steps +SET status='succeeded', result_ref=$1, result_digest=$2, safe_summary=$3, + lease_owner='', lease_until=NULL, version=version+1, updated_at=$4 +WHERE step_id=$5 AND plan_id=$6 AND fencing_token=$7 AND lease_owner=$8 AND status='running'`, + step.ResultRef, step.ResultDigest, step.SafeSummary, now, step.StepID, step.PlanID, token, worker) + if err != nil { + return false, err + } + if tag.RowsAffected() == 1 { + return true, nil + } + var status, digest string + err = tx.QueryRow(ctx, `SELECT status, result_digest FROM intent_plan_steps WHERE step_id=$1 AND plan_id=$2`, + step.StepID, step.PlanID).Scan(&status, &digest) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + return status == string(StepSucceeded) && digest == step.ResultDigest, nil +} + +func (Repository) ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error) { + tx, err := txFrom(scope) + if err != nil { + return 0, err + } + rows, err := tx.Query(ctx, ` +UPDATE intent_plans +SET status='expired', revision=revision+1, updated_at=$1 +WHERE status = ANY($2) AND expires_at <= $1 +RETURNING plan_id`, now, waitingStatusStrings()) + if err != nil { + return 0, err + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return 0, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + return 0, err + } + if len(ids) == 0 { + return 0, nil + } + _, err = tx.Exec(ctx, ` +UPDATE intent_plan_steps +SET status='cancelled', updated_at=$1 +WHERE plan_id = ANY($2) AND status IN ('pending','eligible')`, now, ids) + return len(ids), err +} + +func (Repository) PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error) { + tx, err := txFrom(scope) + if err != nil { + return 0, err + } + tag, err := tx.Exec(ctx, ` +DELETE FROM intent_plans +WHERE updated_at < $1 + AND status IN ('completed','failed','cancelled','expired','superseded')`, before) + if err != nil { + return 0, err + } + return int(tag.RowsAffected()), nil +} + +const planSelect = ` +SELECT plan_id, environment, project_id, session_id, user_id, client_turn_id, request_digest, + segmenter_version, policy_version, status, revision, expires_at, last_fencing_token, + write_plan, execution_mode, safe_summary, display_snapshot, created_at, updated_at +FROM intent_plans` + +const stepColumns = ` +step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version, +action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary, +lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at` + +func loadSteps(ctx context.Context, tx pgx.Tx, planID string) ([]Step, error) { + rows, err := tx.Query(ctx, `SELECT `+stepColumns+` FROM intent_plan_steps WHERE plan_id=$1 ORDER BY ordinal`, planID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Step + for rows.Next() { + s, err := scanStep(rows) + if err != nil { + return nil, err + } + out = append(out, *s) + } + return out, rows.Err() +} + +type scannable interface { + Scan(dest ...any) error +} + +func scanPlan(row scannable) (*Plan, error) { + var p Plan + var status, mode string + var snap []byte + err := row.Scan( + &p.PlanID, &p.Principal.Environment, &p.Principal.ProjectID, &p.Principal.SessionID, + &p.Principal.UserID, &p.ClientTurnID, &p.RequestDigest, &p.SegmenterVersion, + &p.PolicyVersion, &status, &p.Revision, &p.ExpiresAt, &p.LastFencingToken, + &p.WritePlan, &mode, &p.SafeSummary, &snap, &p.CreatedAt, &p.UpdatedAt, + ) + if err != nil { + return nil, err + } + p.Status = PlanStatus(status) + p.ExecutionMode = ExecutionMode(mode) + p.DisplaySnapshot = snap + return &p, nil +} + +func scanStep(row scannable) (*Step, error) { + var s Step + var status, mode string + err := row.Scan( + &s.StepID, &s.PlanID, &s.Ordinal, &s.ActionKey, &s.Slots, &s.DependsOn, &status, + &s.ExpectedResourceVersion, &s.ActionDigest, &s.IdempotencyKey, &s.Attempt, + &s.ResultRef, &s.ResultDigest, &s.SafeSummary, &s.LeaseOwner, &s.LeaseUntil, + &s.FencingToken, &s.Version, &mode, &s.CreatedAt, &s.UpdatedAt, + ) + if err != nil { + return nil, err + } + s.Status = StepStatus(status) + s.ExecutionMode = ExecutionMode(mode) + return &s, nil +} + +func jsonbOrNil(raw []byte) []byte { + if len(raw) == 0 { + return []byte("{}") + } + if !json.Valid(raw) { + return []byte("{}") + } + return raw +} + +func statusStrings(in []PlanStatus) []string { + out := make([]string, len(in)) + for i, s := range in { + out[i] = string(s) + } + return out +} + +func activeStatusStrings() []string { return statusStrings(activeStatuses()) } + +func waitingStatusStrings() []string { return statusStrings(waitingStatuses()) } diff --git a/internal/app/intentflow/store.go b/internal/app/intentflow/store.go new file mode 100644 index 00000000..46a681e7 --- /dev/null +++ b/internal/app/intentflow/store.go @@ -0,0 +1,55 @@ +package intentflow + +import ( + "context" + "time" + + "chat-agent-system/internal/app/uow" +) + +// Store is the intent_plans* owner port. Callers pass the command TxScope. +type Store interface { + InsertPlan(ctx context.Context, scope uow.TxScope, plan *Plan) error + GetPlan(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) + GetPlanForUpdate(ctx context.Context, scope uow.TxScope, planID string) (*Plan, error) + GetByTurn(ctx context.Context, scope uow.TxScope, env, sessionID, clientTurnID string) (*Plan, error) + GetActiveWrite(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) + GetActive(ctx context.Context, scope uow.TxScope, p Principal) (*Plan, error) + CASPlan(ctx context.Context, scope uow.TxScope, planID string, expectedRev int64, from []PlanStatus, to PlanStatus, now time.Time) (newRevision int64, err error) + InsertDecision(ctx context.Context, scope uow.TxScope, d *Decision) error + GetDecisionByNonce(ctx context.Context, scope uow.TxScope, planID, nonce string) (*Decision, error) + ReplaceSteps(ctx context.Context, scope uow.TxScope, planID string, steps []Step) error + ClaimStep(ctx context.Context, scope uow.TxScope, planID, stepID, worker string, until, now time.Time, idempotencyKey string) (*Step, int64, error) + WriteStepReceipt(ctx context.Context, scope uow.TxScope, worker string, token int64, step Step, now time.Time) (bool, error) + ExpireWaiting(ctx context.Context, scope uow.TxScope, now time.Time) (int, error) + PurgeClosed(ctx context.Context, scope uow.TxScope, before time.Time) (int, error) +} + +func waitingStatuses() []PlanStatus { + return []PlanStatus{PlanProposed, PlanAwaitingClarification, PlanAwaitingConfirmation} +} + +func activeStatuses() []PlanStatus { + return []PlanStatus{ + PlanProposed, PlanAwaitingClarification, PlanAwaitingConfirmation, + PlanExecuting, PlanPartiallyCompleted, + } +} + +func isActiveStatus(s PlanStatus) bool { + for _, x := range activeStatuses() { + if x == s { + return true + } + } + return false +} + +func isWaitingStatus(s PlanStatus) bool { + for _, x := range waitingStatuses() { + if x == s { + return true + } + } + return false +} diff --git a/internal/app/intentflow/types.go b/internal/app/intentflow/types.go index d8160f9a..2373aec7 100644 --- a/internal/app/intentflow/types.go +++ b/internal/app/intentflow/types.go @@ -119,3 +119,123 @@ const ( PathPlanByID = "/api/v1/intent-plans/{plan_id}" PathPlanDecisions = "/api/v1/intent-plans/{plan_id}/decisions" ) + +const SegmenterVersionUnknown = "u3-explicit" + +// CreateCommand persists a plan without going through chat. U7 will map HandleTurn +// onto this command; U3 keeps HandleTurn a no-op so production routing stays legacy. +type CreateCommand struct { + Principal Principal + ClientTurnID string + RequestDigest string + SegmenterVersion string + PolicyVersion string + Status PlanStatus + ExecutionMode ExecutionMode + WritePlan bool + ExpiresAt time.Time + SafeSummary string + Steps []StepDraft +} + +// StepDraft is one proposed step at create time. +type StepDraft struct { + ActionKey string + DependsOn []string + Slots []byte + ExpectedResourceVersion string + ActionDigest string + SafeSummary string + ExecutionMode ExecutionMode +} + +// Plan is the owner-side aggregate. It is not returned over HTTP. +type Plan struct { + PlanID string + Principal Principal + ClientTurnID string + RequestDigest string + SegmenterVersion string + PolicyVersion string + Status PlanStatus + Revision int64 + ExpiresAt time.Time + LastFencingToken int64 + WritePlan bool + ExecutionMode ExecutionMode + SafeSummary string + DisplaySnapshot []byte + CreatedAt time.Time + UpdatedAt time.Time + Steps []Step +} + +// Step is one row of the plan ledger. +type Step struct { + PlanID string + StepID string + Ordinal int + ActionKey string + Slots []byte + DependsOn []string + Status StepStatus + ExpectedResourceVersion string + ActionDigest string + IdempotencyKey string + Attempt int + ResultRef string + ResultDigest string + SafeSummary string + LeaseOwner string + LeaseUntil *time.Time + FencingToken int64 + Version int64 + ExecutionMode ExecutionMode + CreatedAt time.Time + UpdatedAt time.Time +} + +// Decision is an append-only command receipt. +type Decision struct { + DecisionID string + PlanID string + Nonce string + PlanRevision int64 + ActorUserID int64 + Kind DecisionKind + StepIDs []string + ExcludeUnselected bool + DecisionDigest string + Outcome string + CreatedAt time.Time +} + +// ClaimedStep is a worker lease plus the fence the action owner must see. +type ClaimedStep struct { + Plan Plan + Step Step + WorkerID string + FencingToken int64 + IdempotencyKey string +} + +// Receipt is the action-owner result. Intermediate payloads stay behind ResultRef. +type Receipt struct { + IdempotencyKey string + ActionDigest string + ResultRef string + ResultDigest string + SafeSummary string +} + +// StepExec is the port payload for an action owner. Confirmation is not authorization. +type StepExec struct { + Principal Principal + PlanID string + StepID string + ActionKey string + IdempotencyKey string + ActionDigest string + ExpectedResourceVersion string + FencingToken int64 +} diff --git a/internal/db/migrate_test.go b/internal/db/migrate_test.go index e0922c81..efd02803 100644 --- a/internal/db/migrate_test.go +++ b/internal/db/migrate_test.go @@ -111,6 +111,41 @@ func TestGooseU9_00038_HasUpAndDown(t *testing.T) { } } +func TestGooseU3_00044_HasUpAndDown(t *testing.T) { + files, err := ListMigrationFiles(ResolveMigrationsDir()) + if err != nil { + t.Fatal(err) + } + var found string + for _, name := range files { + if strings.HasPrefix(name, "00044_") && strings.HasSuffix(name, ".sql") { + found = name + break + } + } + if found == "" { + t.Fatal("expected goose 00044 for intent plans") + } + raw, err := os.ReadFile(filepath.Join(ResolveMigrationsDir(), found)) + if err != nil { + t.Fatal(err) + } + text := string(raw) + if !strings.Contains(text, "-- +goose Up") || !strings.Contains(text, "-- +goose Down") { + t.Fatal("00044 must have Up and Down") + } + for _, table := range []string{"intent_plans", "intent_plan_steps", "intent_plan_decisions"} { + if !strings.Contains(text, table) { + t.Fatalf("00044 must mention %s", table) + } + } + for _, idx := range []string{"uq_intent_plans_turn", "uq_intent_plans_one_active_write", "uq_intent_plan_decisions_nonce"} { + if !strings.Contains(text, idx) { + t.Fatalf("00044 must mention %s", idx) + } + } +} + func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { if testing.Short() { t.Skip("goose down is not part of -short") @@ -132,6 +167,7 @@ func TestGooseM2_00034to00036_DownThenUp(t *testing.T) { "00036_2026-09-14-final-artifact-vault.sql", "00037_2026-09-15-task-run-manifests.sql", "00038_2026-09-15-dataset-lifecycle.sql", + "00044_2026-09-15-intent-plans.sql", } joined := strings.Join(files, "\n") for _, name := range want {