diff --git a/.doc/ops/CURRENT.md b/.doc/ops/CURRENT.md index f46d49fe..23da993c 100644 --- a/.doc/ops/CURRENT.md +++ b/.doc/ops/CURRENT.md @@ -5,7 +5,7 @@ ## 焦点 -1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B/E1A/E1B PASS；下一开工 E1C projection，默认 dispatcher mode 仍 off） +1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A–E1C 隔离库 PASS；dispatcher 默认 off。E2 删除 hook 须等 live 窗口） 2. **CHAT-INTENT**: [对话意图运行时状态](../test-reports/2026-09-13-conversation-intent-runtime-state-test-report.md)（双标签 busy 已签：项目 **6275** / `DOC_BUSY` 单 run；不删 `sess-p143-u9` 探针） 3. **CHAT**: [对话历史刷新活验已补](../test-reports/2026-09-13-chat-history-refresh-test-report.md)（项目 6252 闲聊刷新同集 + guest 403；编制任务链仍开放） @@ -61,7 +61,8 @@ | Cursor | event-e0a-production-scope-contract | fix/event-production-scope-contract | - | events | done | | Cursor | event-e0b-dispatcher-pg | feature/events-dispatcher-pg | - | events | done | | Cursor | event-e1a-notify-cutover | fix/event-notify-cutover | - | events | done | -| Cursor | event-e1b-workflow-advance | fix/event-workflow-advance | - | events | in_progress | +| Cursor | event-e1b-workflow-advance | fix/event-workflow-advance | - | events | done | +| Cursor | event-e1c-projection-cutover | fix/event-projection-cutover | - | events | in_progress | > 新任务开始前登记。`branch` 必须为 `feature/` 或 `fix/`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。 @@ -73,7 +74,7 @@ | holder | phase | since | scope | queue | |--------|-------|-------|-------|-------| -| Cursor | merging | 2026-09-15 | E1B workflow advance；mode=off | - | +| Cursor | merging | 2026-09-15 | E1C projection no-replace；mode=off | - | **phase**: `idle` | `merging` | `smoke` | `e2e` diff --git a/.doc/ops/short-term/2026-09-15-durable-event-track-e.md b/.doc/ops/short-term/2026-09-15-durable-event-track-e.md index 7d28cbe6..a675f9b7 100644 --- a/.doc/ops/short-term/2026-09-15-durable-event-track-e.md +++ b/.doc/ops/short-term/2026-09-15-durable-event-track-e.md @@ -4,7 +4,7 @@ status: in_progress created: 2026-09-15 updated: 2026-09-15 owner: Cursor -branch: fix/event-workflow-advance +branch: fix/event-projection-cutover worktree: "-" mutex: events --- @@ -30,7 +30,7 @@ mutex: events 1. [x] E0A 单独建 short-term/分支，dispatcher 保持 off。 2. [x] E0A 报告通过后再认领 E0B。 -3. [ ] 依次交付 E1A notify、E1B workflow advance、E1C projection；每步独立 go/no-go。E1A/E1B 报告已过，下一开工 E1C。 +3. [x] 依次交付 E1A notify、E1B workflow advance、E1C projection；每步独立 go/no-go。E2 删除 hook 待 live 窗口。 4. [ ] 一个兼容发布窗口后 contract 删除闭包 API。 5. [ ] 同步 test report、CURRENT、FEATURE-MATRIX、gap-matrix。 diff --git a/.doc/ops/short-term/2026-09-15-event-e1b-workflow-advance.md b/.doc/ops/short-term/2026-09-15-event-e1b-workflow-advance.md index b7d42548..90f92a20 100644 --- a/.doc/ops/short-term/2026-09-15-event-e1b-workflow-advance.md +++ b/.doc/ops/short-term/2026-09-15-event-e1b-workflow-advance.md @@ -1,6 +1,6 @@ --- title: "E1B workflow advance shadow 与切换" -status: in_progress +status: done created: 2026-09-15 updated: 2026-09-15 owner: Cursor diff --git a/.doc/ops/short-term/2026-09-15-event-e1c-projection.md b/.doc/ops/short-term/2026-09-15-event-e1c-projection.md new file mode 100644 index 00000000..67d8983c --- /dev/null +++ b/.doc/ops/short-term/2026-09-15-event-e1c-projection.md @@ -0,0 +1,41 @@ +--- +title: "E1C projection no-replace" +status: in_progress +created: 2026-09-15 +updated: 2026-09-15 +owner: Cursor +branch: fix/event-projection-cutover +worktree: "-" +mutex: events +--- + +# E1C projection no-replace + +## 目标 + +projection 消费者幂等只前进；共享集群保持 off。 + +## 范围 + +- Goose 00043、ProjectHeadNoReplace、docproject.EventConsumer + +## 非目标 + +- 不把共享 projection 切 live +- 不激活 FAV +- 不删除闭包 + +## 步骤 + +1. [x] 分支与 additive schema +2. [x] no-replace SQL + consumer +3. [x] 隔离库 + race + worker 重启 +4. [x] 报告 + +## 验收标准 + +- [x] [E1C 测试报告](../../test-reports/2026-09-15-event-e1c-projection-test-report.md) + +## 相关链接 + +- Track E: [./2026-09-15-durable-event-track-e.md](./2026-09-15-durable-event-track-e.md) diff --git a/.doc/remediation/gap-matrix.md b/.doc/remediation/gap-matrix.md index 529b219c..44d7dc0b 100644 --- a/.doc/remediation/gap-matrix.md +++ b/.doc/remediation/gap-matrix.md @@ -70,7 +70,7 @@ | DOCREV-ACTIVE-DOC-ROUTING | UI-05「当前受控文档」被法规关键词改路由到 `feasibility_reg`，intake 无 pending | **PASS** | [audit](../audit/2026-09-13-document-revision-active-doc-routing.md)；Round20 UI-05 pending 停在 intake V2 | **不改弱 E2E**；不删证据 | | DOCREV-QUALITY-STATUS-FP | 行动计划表状态列「待续」被门禁当敷衍，两轮 repair 后 quality_failed 80 | **PASS** | [audit](../audit/2026-09-13-document-revision-quality-status-false-positive.md)；Round20 UI-03 | **不删禁词、不放宽正文**；不删证据 | | GOVERNANCE-ACTION-AUTHZ | 文档动作按职责鉴权；未激活 202-blocked；关闭创建直通 | **PASS** | [test-report](../test-reports/2026-09-14-governance-action-level-authz-test-report.md)；[audit 异态](../audit/2026-09-14-planning-with-workflow-key-anomaly.md) | 不改 83 条 planning+key 夹具；none 长 PI 审批链另开 | -| DOC-RELIABILITY-PILOT | 受控文档写路径与 PG 事件交付尚未形成生产闭环；闭包副作用在崩溃/重启后不可重试审计，错误 scope 或 projection 缺失仍可能导致刷新失败 | OPEN | [初审](../audit/2026-09-14-module-decoupling-plan-review.md)；[Phase 1](./2026-09-14-module-decoupling-refactor-plan.md)；[Phase 2 复核](../audit/2026-09-15-module-decoupling-phase2-plan-review.md)；[Phase 2 修订](./2026-09-15-module-decoupling-ipc-phase2-remediation-plan.md)；[U11–U12 报告](../test-reports/2026-09-15-app-composition-u11-u12-test-report.md)；[E1A](../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md) | E0A/E0B/E1A 已过隔离库门。下一 go/no-go 为 E1B workflow advance。dispatcher/notify 保持 off。FAV、结构整理、HTTP/PI 协议不纳入本 blocker 关闭条件。项目 143 只读 | +| DOC-RELIABILITY-PILOT | 受控文档写路径与 PG 事件交付尚未形成生产闭环；闭包副作用在崩溃/重启后不可重试审计，错误 scope 或 projection 缺失仍可能导致刷新失败 | OPEN | [初审](../audit/2026-09-14-module-decoupling-plan-review.md)；[Phase 1](./2026-09-14-module-decoupling-refactor-plan.md)；[Phase 2 复核](../audit/2026-09-15-module-decoupling-phase2-plan-review.md)；[Phase 2 修订](./2026-09-15-module-decoupling-ipc-phase2-remediation-plan.md)；[U11–U12 报告](../test-reports/2026-09-15-app-composition-u11-u12-test-report.md)；[E1A](../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md)；[E1B](../test-reports/2026-09-15-event-e1b-workflow-advance-test-report.md)；[E1C](../test-reports/2026-09-15-event-e1c-projection-test-report.md) | E0A–E1C 隔离库已过。E2 删除闭包须等 live 窗口。dispatcher 保持 off。FAV、结构整理、HTTP/PI 协议不纳入本 blocker 关闭条件。项目 143 只读 | --- diff --git a/.doc/status/FEATURE-MATRIX.md b/.doc/status/FEATURE-MATRIX.md index 1845220b..60c795a9 100644 --- a/.doc/status/FEATURE-MATRIX.md +++ b/.doc/status/FEATURE-MATRIX.md @@ -27,7 +27,7 @@ | PI Runner / 配置链 | pi-agent + MiniMax 真实链路 | done | [test piagent](../test-reports/2026-09-13-codex-audit-p0-remediation-test-report.md)；[container-env](../test-reports/2026-09-13-container-env-configuration-test-report.md) | — | `pi-agent` | | 三环境运维 | dev/test/main + 共享 KB | done | [PROJECT.md](../../PROJECT.md) §4；四环境 healthy | — | `infra` | | CI / E2E | PG 集成 + 真实 PI E2E | partial | [gap-matrix BUG-08](../remediation/gap-matrix.md)；[test pg-ci](../test-reports/2026-09-12-pg-ci-soak-e2e-test-report.md)；[运行时 UI Round20](../test-reports/2026-09-14-document-revision-runtime-ui-test-report.md) | CI `TEST_DATABASE_URL` 隔离库仍缺；文档运行时真实 UI 已按 UI_PHASE 分文件且 post-restart 断言最终 A1 正文。`-race` 全包并行 TempDir flake 为基建残留 | `ci` | -| 模块解耦与可靠事件 | 文档可靠性纵向试点、PG 持久事件交付与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[Phase 1](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[Phase 2 审阅](../audit/2026-09-15-module-decoupling-phase2-plan-review.md)；[Phase 2 修订计划](../remediation/2026-09-15-module-decoupling-ipc-phase2-remediation-plan.md)；[E0A 报告](../test-reports/2026-09-15-event-e0a-production-scope-contract-test-report.md)；[E0B 报告](../test-reports/2026-09-15-event-e0b-dispatcher-pg-test-report.md)；[E1A 报告](../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md)；[E1B 报告](../test-reports/2026-09-15-event-e1b-workflow-advance-test-report.md) | E0A–E1B 已过隔离库门；dispatcher/notify/advance 默认 off。下一波 E1C。Track S 与 HTTP/PI 协议分别规划；FAV 保持独立审计 | `architecture-docs` | +| 模块解耦与可靠事件 | 文档可靠性纵向试点、PG 持久事件交付与选择性模块化 | partial | [架构](../system-design/2026-09-14-module-decoupling-architecture.md)；[ADR](../adr/2026-09-14-modular-monolith-decoupling.md)；[Phase 1](../remediation/2026-09-14-module-decoupling-refactor-plan.md)；[Phase 2 审阅](../audit/2026-09-15-module-decoupling-phase2-plan-review.md)；[Phase 2 修订计划](../remediation/2026-09-15-module-decoupling-ipc-phase2-remediation-plan.md)；[E0A](../test-reports/2026-09-15-event-e0a-production-scope-contract-test-report.md)；[E0B](../test-reports/2026-09-15-event-e0b-dispatcher-pg-test-report.md)；[E1A](../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md)；[E1B](../test-reports/2026-09-15-event-e1b-workflow-advance-test-report.md)；[E1C](../test-reports/2026-09-15-event-e1c-projection-test-report.md) | E0A–E1C 隔离库已过；dispatcher 默认 off。E2 删除闭包须等 live。Track S/P 与 FAV 另列 | `architecture-docs` | --- diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 06031c3c..d4fb184c 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -62,6 +62,7 @@ imports: - app/docadvance - app/docflow - app/docnotify + - app/docproject - app/stageflow - app/uow - artifactvault @@ -96,6 +97,9 @@ imports: - notification - projectinstance - projectworkflow + app/docproject: + - docgov + - events app/docflow: - app/uow - docgov @@ -255,6 +259,8 @@ table_owners: - events event_reconciliations: - events + event_projection_outcomes: + - app/docproject notifications: - notification stage_advance_requests: @@ -308,6 +314,7 @@ event_producers: - docrevision - app/docflow - events + - app/docproject document.downstream_impact_detected: - docrevision - app/docflow diff --git a/.doc/test-reports/2026-09-15-event-e1c-projection-test-report.md b/.doc/test-reports/2026-09-15-event-e1c-projection-test-report.md new file mode 100644 index 00000000..b888dd0e --- /dev/null +++ b/.doc/test-reports/2026-09-15-event-e1c-projection-test-report.md @@ -0,0 +1,77 @@ +# E1C projection no-replace 测试报告 + +## 一、 测试背景 (Background) + +notify 与 workflow_advance 已接到独立 worker。本波把 `document.agent_version_committed` 的 projection 做成幂等 **只前进不回退** 的 head 指针应用，并用 `event_projection_outcomes` 保证每个 envelope / 每个 (doc, version) 只有一个可见结果。生产集群 projection 保持 **off**；HTTP 进程仍由提交 UoW 写 head。 + +## 二、 测试目标 (Test Objectives) + +- Goose 00043 additive outcomes 表。 +- Shadow 不写 outcomes / 不改 head。 +- Live 重放仍一个 outcome；旧 version 事件不得覆盖更新的 head。 +- 缺文档 quarantine。 +- mode=off 回滚。 +- events worker 重启后 HTTP 与 worker ready 仍可用。 + +## 三、 测试方案与环境 (Test Plan & Environment) + +| 项 | 值 | +|----|-----| +| OS | Linux 7.0.0-31-generic | +| Go | go1.27.1 linux/amd64 | +| 分支 | `fix/event-projection-cutover` | +| Schema | **43**（隔离库） | +| PostgreSQL | `agent_rel_e1c` `:5434`；测完 DROP | +| 项目 143 | 只读 | +| 默认 mode | off | +| 证据 | `.doc/test-reports/assets/2026-09-15-event-e1c-projection/` | + +三类崩溃：Handler 与 Ack 同事务，副作用后/ack 前崩溃等于 Rollback，与 E0B fencing 叠加。本波未在共享库把 projection 切 live，因此 **未** 用真实授权页做 live 消费者纵切；授权页仍走提交 UoW 投影。 + +## 四、 预期结果 (Expected Results) + +| 用例 | 期望 | +|------|------| +| shadow | outcomes=0 | +| live 两次 | outcomes=1；head_revision_no=2 | +| 更旧 version 事件 | head 仍为 2 | +| missing doc | quarantine | +| off | claim=0 | +| worker restart | :8088 health/ready ok；dev modes off | + +## 五、 实际测试结果与判定 (Actual Results & Evaluation) + +### 5.1 隔离库 + +- **测试输入**: `TEST_DATABASE_URL=.../agent_rel_e1c`；`go test -v ./internal/app/docproject -run TestE1C_` +- **实际输出**: goose 至 43；PASS 0.30s（`pg-test.txt` 为同库二次运行 0.04s PASS） +- **评价判定**: **PASS** +- **性能与耗时**: 0.30s 含迁移 + +### 5.2 race + +- **实际输出**: `race.txt` PASS 1.07s +- **评价判定**: **PASS** + +### 5.3 worker 重启 + +- **测试输入**: `docker restart shuimu-agent-events-dev`；curl health/ready +- **实际输出**: `restart.txt`；8088/8080 ok；modes 全 off +- **评价判定**: **PASS**（未切 live，故无 projection-missing 由本消费者引起） + +### 5.4 arch-check + +- **评价判定**: **PASS** + +## 六、 自动化测试套件汇总 (Test Suite Execution) + +| 命令 | 结果 | +|------|------| +| 隔离库 TestE1C | PASS schema 43 | +| race | PASS | +| worker restart | PASS | +| 共享 projection mode | **off** | + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +E1C **隔离库 + no-replace + worker 重启 PASS**。共享环境未 live。E2 删除 `OnOutbox` / `AddOnApprovedHook` 必须等 live 窗口；当前闭包仍是 HTTP 唯一写者。FAV 仍 off。 diff --git a/.doc/test-reports/assets/2026-09-15-event-e1c-projection/consumer-modes.txt b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/consumer-modes.txt new file mode 100644 index 00000000..5b6b0dc9 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/consumer-modes.txt @@ -0,0 +1,8 @@ + consumer | mode +------------------+------ + fav | off + notify | off + projection | off + workflow_advance | off +(4 rows) + diff --git a/.doc/test-reports/assets/2026-09-15-event-e1c-projection/pg-test.txt b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/pg-test.txt new file mode 100644 index 00000000..18da3bac --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/pg-test.txt @@ -0,0 +1,5 @@ +=== RUN TestE1C_ProjectionShadowLiveNoReplace_PG +2026/09/15 09:57:31 goose: no migrations to run. current version: 43 +--- PASS: TestE1C_ProjectionShadowLiveNoReplace_PG (0.04s) +PASS +ok chat-agent-system/internal/app/docproject 0.040s diff --git a/.doc/test-reports/assets/2026-09-15-event-e1c-projection/race.txt b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/race.txt new file mode 100644 index 00000000..90fff1f7 --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/race.txt @@ -0,0 +1 @@ +ok chat-agent-system/internal/app/docproject 1.069s diff --git a/.doc/test-reports/assets/2026-09-15-event-e1c-projection/restart.txt b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/restart.txt new file mode 100644 index 00000000..44b86a3e --- /dev/null +++ b/.doc/test-reports/assets/2026-09-15-event-e1c-projection/restart.txt @@ -0,0 +1,7 @@ +events worker restart 2026-09-15T09:57:31Z +GET :8088/health -> {"env":"dev","owner":"events-1","status":"ok"} +GET :8088/ready -> {"status":"ok"} +GET :8080/health -> ok schema_version 42 (00043 not yet on shared app image) +dev consumer modes: all off +FAV off +projection-missing not observed on HTTP health/ready after restart diff --git a/cmd/events/main.go b/cmd/events/main.go index da67da6c..3102625c 100644 --- a/cmd/events/main.go +++ b/cmd/events/main.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/app/docadvance" "chat-agent-system/internal/app/docnotify" + "chat-agent-system/internal/app/docproject" "chat-agent-system/internal/config" "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" @@ -77,6 +78,9 @@ func main() { _ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{ Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env, }).HandleLease) + _ = disp.Register(events.ConsumerProjection, (&docproject.EventConsumer{ + Docs: gov, Env: cfg.Env, + }).HandleLease) mux := http.NewServeMux() mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner}) diff --git a/deploy/postgres/goose/00043_2026-09-15-projection-outcome.sql b/deploy/postgres/goose/00043_2026-09-15-projection-outcome.sql new file mode 100644 index 00000000..fc6bdb42 --- /dev/null +++ b/deploy/postgres/goose/00043_2026-09-15-projection-outcome.sql @@ -0,0 +1,17 @@ +-- +goose Up +-- E1C: unique projection apply per envelope; no-replace is enforced in SQL + consumer. +CREATE TABLE IF NOT EXISTS event_projection_outcomes ( + source_event_id TEXT PRIMARY KEY, + doc_id TEXT NOT NULL, + version_no INT NOT NULL, + snapshot_id BIGINT NOT NULL DEFAULT 0, + applied BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_projection_doc_version + ON event_projection_outcomes (doc_id, version_no); + +-- +goose Down +DROP INDEX IF EXISTS uq_projection_doc_version; +DROP TABLE IF EXISTS event_projection_outcomes; diff --git a/internal/app/docproject/consumer.go b/internal/app/docproject/consumer.go new file mode 100644 index 00000000..a9766814 --- /dev/null +++ b/internal/app/docproject/consumer.go @@ -0,0 +1,96 @@ +package docproject + +import ( + "context" + "encoding/json" + "errors" + + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/events" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +// EventConsumer applies document.agent_version_committed onto ledger head pointers. +// Closures/producer already project in the approve/commit UoW; this path is idempotent no-replace. +type EventConsumer struct { + Docs DocumentLookup + Env string +} + +type DocumentLookup interface { + GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error) +} + +func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error { + if c == nil { + return events.ErrUnknownConsumer + } + principal := events.ServicePrincipal{ + ID: "projection-" + c.Env, Environment: c.Env, Consumer: events.ConsumerProjection, + Actions: []string{"projection.apply"}, + } + if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "projection.apply"); err != nil { + return events.ErrQuarantine + } + if lease.EventName != events.EventVersionCommitted { + return nil + } + docID := lease.Envelope.AggregateID + versionNo := int(lease.Envelope.AggregateVersion) + var payload map[string]any + _ = json.Unmarshal(lease.Envelope.Payload, &payload) + if payload != nil { + if id, ok := payload["doc_id"].(string); ok && id != "" { + docID = id + } + if v, ok := payload["version_no"].(float64); ok { + versionNo = int(v) + } + } + if c.Docs == nil || docID == "" { + return events.ErrQuarantine + } + doc, err := c.Docs.GetDocument(ctx, docID) + if err != nil || doc == nil { + return events.ErrQuarantine + } + if doc.ProjectID != lease.ProjectID { + return events.ErrQuarantine + } + var versionID, snapshotID int64 + var sha string + err = tx.QueryRow(ctx, ` + SELECT v.id, COALESCE(v.head_snapshot_id,0), COALESCE(s.content_sha256,'') + FROM document_versions v + LEFT JOIN document_snapshots s ON s.id = v.head_snapshot_id + WHERE v.doc_id=$1 AND v.version_no=$2 + `, docID, versionNo).Scan(&versionID, &snapshotID, &sha) + if err != nil { + return events.ErrQuarantine + } + tag, err := tx.Exec(ctx, ` + INSERT INTO event_projection_outcomes (source_event_id, doc_id, version_no, snapshot_id, applied) + VALUES ($1,$2,$3,$4,false) + ON CONFLICT (source_event_id) DO NOTHING + `, lease.EventID, docID, versionNo, snapshotID) + if err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" { + return nil + } + return err + } + if tag.RowsAffected() == 0 { + return nil + } + applied, err := (docgov.Repository{}).ProjectHeadNoReplace(ctx, tx, docID, versionID, snapshotID, versionNo, sha) + if err != nil { + return err + } + _, err = tx.Exec(ctx, ` + UPDATE event_projection_outcomes SET applied=$2, snapshot_id=$3 WHERE source_event_id=$1 + `, lease.EventID, applied, snapshotID) + return err +} diff --git a/internal/app/docproject/e1c_pg_test.go b/internal/app/docproject/e1c_pg_test.go new file mode 100644 index 00000000..3fe5f87c --- /dev/null +++ b/internal/app/docproject/e1c_pg_test.go @@ -0,0 +1,208 @@ +package docproject_test + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "chat-agent-system/internal/app/docproject" + "chat-agent-system/internal/app/uow" + "chat-agent-system/internal/db" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/events" + + "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 TestE1C_ProjectionShadowLiveNoReplace_PG(t *testing.T) { + pool, ctx := e1cPool(t) + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + userID, projectID := seedUserProject(t, ctx, pool, suffix) + docID := "e1c_" + suffix + seedDoc(t, ctx, pool, docID, projectID, userID) + snapID, verID := seedVersion(t, ctx, pool, docID, 2) + + env := committedEnv("c1-"+suffix, docID, projectID, 2) + appendEnv(t, ctx, pool, env, []string{events.ConsumerProjection}) + + d := events.NewDispatcher(pool, "dev", "e1c-w") + if err := d.Register(events.ConsumerProjection, (&docproject.EventConsumer{ + Docs: docgov.NewGovernanceEngine(pool), Env: "dev", + }).HandleLease); err != nil { + t.Fatal(err) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeShadow); err != nil { + t.Fatal(err) + } + if n, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil || n < 1 { + t.Fatalf("shadow n=%d err=%v", n, err) + } + var outcomes, head int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_projection_outcomes WHERE source_event_id=$1`, env.EventID).Scan(&outcomes); err != nil { + t.Fatal(err) + } + if outcomes != 0 { + t.Fatalf("shadow wrote outcomes=%d", outcomes) + } + + if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeLive); err != nil { + t.Fatal(err) + } + if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil { + t.Fatal(err) + } + if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_projection_outcomes WHERE doc_id=$1 AND version_no=2`, docID).Scan(&outcomes); err != nil { + t.Fatal(err) + } + if outcomes != 1 { + t.Fatalf("unique projection outcomes=%d", outcomes) + } + if err := pool.QueryRow(ctx, `SELECT head_revision_no FROM document_records WHERE doc_id=$1`, docID).Scan(&head); err != nil { + t.Fatal(err) + } + if head != 2 { + t.Fatalf("head_revision_no=%d want 2 snap=%d ver=%d", head, snapID, verID) + } + + older := committedEnv("old-"+suffix, docID, projectID, 1) + appendEnv(t, ctx, pool, older, []string{events.ConsumerProjection}) + seedVersion(t, ctx, pool, docID, 1) + if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `SELECT head_revision_no FROM document_records WHERE doc_id=$1`, docID).Scan(&head); err != nil { + t.Fatal(err) + } + if head != 2 { + t.Fatalf("no-replace failed, head=%d", head) + } + + stale := committedEnv("stale-"+suffix, "missing-"+suffix, projectID, 1) + appendEnv(t, ctx, pool, stale, []string{events.ConsumerProjection}) + if _, err := d.RunOnce(ctx, events.ConsumerProjection); err != nil { + t.Fatal(err) + } + var status string + if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, stale.EventID, events.ConsumerProjection).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "quarantine" { + t.Fatalf("missing doc status=%s", status) + } + + tx, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + replay := committedEnv("replay-"+suffix, docID, projectID, 3) + appendEnv(t, ctx, pool, replay, []string{events.ConsumerProjection}) + seedVersion(t, ctx, pool, docID, 3) + if err := events.SetMode(ctx, pool, "dev", events.ConsumerProjection, events.ModeOff); err != nil { + t.Fatal(err) + } + n, err := d.RunOnce(ctx, events.ConsumerProjection) + _ = tx.Rollback(ctx) + if err != nil || n != 0 { + t.Fatalf("rollback off claimed=%d err=%v", n, err) + } +} + +func e1cPool(t *testing.T) (*pgxpool.Pool, context.Context) { + t.Helper() + if testing.Short() { + t.Skip("e1c") + } + 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) + } + return pool, ctx +} + +func seedUserProject(t *testing.T, ctx context.Context, pool *pgxpool.Pool, suffix string) (userID, projectID int64) { + t.Helper() + if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'A','engineer') RETURNING id`, "e1c-"+suffix).Scan(&userID); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`, "e1c-"+suffix, "e1c").Scan(&projectID); err != nil { + t.Fatal(err) + } + return userID, projectID +} + +func seedDoc(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, projectID, userID int64) { + t.Helper() + if _, err := pool.Exec(ctx, ` + INSERT INTO document_records ( + doc_id, project_id, task_key, title, stage, version, current_state, is_locked, is_template_only, + author_id, author_name, created_at, updated_at, head_revision_no + ) VALUES ($1,$2,'intake','t','V0.5',1,'drafting',false,false,$3,'A',now(),now(),0) + `, docID, projectID, userID); err != nil { + t.Fatal(err) + } +} + +func seedVersion(t *testing.T, ctx context.Context, pool *pgxpool.Pool, docID string, versionNo int) (snapID, verID int64) { + t.Helper() + sha := fmt.Sprintf("%064d", versionNo) + if _, err := pool.Exec(ctx, `INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size) VALUES ($1,$2,1) ON CONFLICT (content_sha256) DO NOTHING`, sha, "body"); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO document_snapshots (doc_id, content_sha256, block_map, source_kind) + VALUES ($1,$2,'[]','agent') RETURNING id + `, docID, sha).Scan(&snapID); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO document_versions (doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id, sealed_at) + VALUES ($1,$2,'sealed','agent',$3,$3,now()) RETURNING id + `, docID, versionNo, snapID).Scan(&verID); err != nil { + t.Fatal(err) + } + return snapID, verID +} + +func committedEnv(id, docID string, projectID, ver int64) events.Envelope { + body := []byte(fmt.Sprintf(`{"doc_id":%q,"version_no":%d}`, docID, ver)) + return events.Envelope{ + EventID: "commit-" + id, EventName: events.EventVersionCommitted, SchemaVersion: events.EnvelopeSchema, + Environment: "dev", ProjectID: projectID, AggregateType: "document", AggregateID: docID, + AggregateVersion: ver, ActorRef: "user:1", IdempotencyKey: "commit-" + id, + PayloadDigest: events.Digest(body), Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body, + } +} + +func appendEnv(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) { + t.Helper() + if err := uow.NewPG(pool).Within(ctx, func(scope uow.TxScope) error { + return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers) + }); err != nil { + t.Fatal(err) + } +} diff --git a/internal/docgov/repository.go b/internal/docgov/repository.go index e14ae50a..a6f8c9f0 100644 --- a/internal/docgov/repository.go +++ b/internal/docgov/repository.go @@ -6,6 +6,8 @@ import ( "time" "chat-agent-system/internal/app/uow" + + "github.com/jackc/pgx/v5" ) // Repository is the document_records writer bound to a UnitOfWork scope. @@ -106,3 +108,23 @@ func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID stri } return nil } + +// ProjectHeadNoReplace moves ledger pointers only forward. Equal or newer heads stay. +func (Repository) ProjectHeadNoReplace(ctx context.Context, tx pgx.Tx, docID string, versionID, snapshotID int64, revisionNo int, sha string) (bool, error) { + if tx == nil || docID == "" { + return false, fmt.Errorf("docgov: project head requires tx and doc_id") + } + tag, err := tx.Exec(ctx, ` + UPDATE document_records + SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + row_revision=COALESCE(row_revision,0)+1, + content_sha256=COALESCE(NULLIF($5,''), content_sha256), + updated_at=now() + WHERE doc_id=$1 + AND COALESCE(head_revision_no, 0) < $4 + `, docID, versionID, snapshotID, revisionNo, sha) + if err != nil { + return false, fmt.Errorf("docgov: project head no-replace: %w", err) + } + return tag.RowsAffected() == 1, nil +} diff --git a/internal/events/registry.go b/internal/events/registry.go index ba25ea57..9028f1d5 100644 --- a/internal/events/registry.go +++ b/internal/events/registry.go @@ -12,24 +12,24 @@ const ( ConsumerWorkflowAdvance = "workflow_advance" ConsumerFAV = "fav" - EventCandidateReady = "document.agent_candidate_ready" - EventVersionCommitted = "document.agent_version_committed" - EventDownstreamImpact = "document.downstream_impact_detected" + EventCandidateReady = "document.agent_candidate_ready" + EventVersionCommitted = "document.agent_version_committed" + EventDownstreamImpact = "document.downstream_impact_detected" EventChangeReportDegraded = "document.change_report_degraded" - EventApproved = "document.approved" - EventTaskRunCompleted = "taskrun.completed" - EventTaskRunCancelled = "taskrun.cancelled" - EventFAVArtifactStaged = "fav.artifact_staged" + EventApproved = "document.approved" + EventTaskRunCompleted = "taskrun.completed" + EventTaskRunCancelled = "taskrun.cancelled" + EventFAVArtifactStaged = "fav.artifact_staged" ) // Contract is the static allowlist row for one event name. type Contract struct { - EventName string - SchemaVersion int - Enabled bool - Producers []string - Consumers []string - PayloadFields []string + EventName string + SchemaVersion int + Enabled bool + Producers []string + Consumers []string + PayloadFields []string } // StaticContracts is the runtime copy of module-boundaries event sections. @@ -37,7 +37,7 @@ type Contract struct { func StaticContracts() []Contract { return []Contract{ {EventName: EventCandidateReady, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "candidate_id", "risk"}}, - {EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}}, + {EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events", "app/docproject"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}}, {EventName: EventDownstreamImpact, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "impacted_doc_id", "version_no"}}, {EventName: EventChangeReportDegraded, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "report_id"}}, {EventName: EventApproved, SchemaVersion: 1, Enabled: true, Producers: []string{"app/docflow", "events", "app/docadvance"}, Consumers: []string{ConsumerWorkflowAdvance, ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id", "task_key"}},