merge: fix/event-notify-cutover

.doc/ops/CURRENT.md

@@ -5,7 +5,7 @@
 
 ## 焦点
 
-1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B PASS；下一开工 E1A notify shadow→live，默认 dispatcher mode 仍 off）
+1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B/E1A PASS；下一开工 E1B workflow advance，默认 dispatcher mode 仍 off）
 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；编制任务链仍开放）
 
@@ -60,6 +60,7 @@
 | Cursor | leftover-branch-cleanup | chore/leftover-branch-cleanup | - | ops | done |
 | 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 | in_progress |
 
 > 新任务开始前登记。`branch` 必须为 `feature/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。
 
@@ -71,7 +72,7 @@
 
 | holder | phase | since | scope | queue |
 |--------|-------|-------|-------|-------|
-| Cursor | merging | 2026-09-15 | E0B dispatcher 基座；mode=off | - |
+| Cursor | merging | 2026-09-15 | E1A notify cutover；mode=off | - |
 
 **phase**: `idle` | `merging` | `smoke` | `e2e`
 
@@ -81,7 +82,7 @@
 
 | 环境 | 端口 | schema | 状态 |
 |------|------|--------|------|
-| dev | 8080 | **38** | healthy（`pi-agent`；U11–U12 rebuild digest `a8e3e940f01111c4298c67947ae9f4cd6105f13cf58134672f4a87daa19e27ab`；health+ready PASS） |
+| dev | 8080 | **41** | healthy（`pi-agent`；events worker `:8088`；dispatcher mode=off；health+ready PASS） |
 | test | 8081 | 25→27* | healthy |
 | main | 8000 | 25→27* | healthy（app + runner + DB） |
 

.doc/ops/short-term/2026-09-15-durable-event-track-e.md

@@ -3,8 +3,8 @@ title: "Track E 持久事件交付"
 status: in_progress
 created: 2026-09-15
 updated: 2026-09-15
-owner: TBD
-branch: fix/event-production-scope-contract
+owner: Cursor
+branch: fix/event-notify-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。
+3. [ ] 依次交付 E1A notify、E1B workflow advance、E1C projection；每步独立 go/no-go。E1A 报告已过，下一开工 E1B。
 4. [ ] 一个兼容发布窗口后 contract 删除闭包 API。
 5. [ ] 同步 test report、CURRENT、FEATURE-MATRIX、gap-matrix。
 

.doc/ops/short-term/2026-09-15-event-e1a-notify-cutover.md

@@ -0,0 +1,48 @@
+---
+title: "E1A notify shadow 与切换"
+status: in_progress
+created: 2026-09-15
+updated: 2026-09-15
+owner: Cursor
+branch: fix/event-notify-cutover
+worktree: "-"
+mutex: events
+---
+
+# E1A notify shadow 与切换
+
+## 目标
+
+notify 消费者具备 shadow 覆盖门、live 唯一可见结果与显式 rollback；生产集群保持 off。
+
+## 范围
+
+- Goose 00041、`EmitVisibleDocumentEvent`、`docnotify.EventConsumer`、Handler 与 Ack 同事务
+- cmd/events 注册 notify（mode=off 时不 claim）
+
+## 非目标
+
+- 不把共享环境 notify 切 live
+- 不删除 OnOutbox 闭包
+- 不激活 FAV / workflow advance / projection live
+
+## 步骤
+
+1. [x] 从 E0B 后 `dev` 开 `fix/event-notify-cutover`
+2. [x] additive `source_event_id` 与可见结果唯一索引
+3. [x] shadow dry-run + live 同事务 ack
+4. [x] fixture N=30、quarantine、correlation、rollback
+5. [x] 隔离库 PG + race + 报告
+
+## 验收标准
+
+- [x] [E1A 测试报告](../../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md)
+
+## 相关链接
+
+- Track E: [./2026-09-15-durable-event-track-e.md](./2026-09-15-durable-event-track-e.md)
+- Report: [../../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md](../../test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md)
+
+## 证据 / 备注
+
+隔离库 `agent_rel_e1a` 测完已 DROP。共享 `event_consumer_modes.notify=off`。

.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) | 下一 go/no-go 仅为 Track E E0A：生产者、event contract、历史数据与 sequence 门；dispatcher 保持 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) | E0A/E0B/E1A 已过隔离库门。下一 go/no-go 为 E1B workflow advance。dispatcher/notify 保持 off。FAV、结构整理、HTTP/PI 协议不纳入本 blocker 关闭条件。项目 143 只读 |
 
 ---
 

.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) | E0A/E0B 已过隔离库门；dispatcher 默认 off。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) | E0A/E0B/E1A 已过隔离库门；dispatcher 与 notify 默认 off。下一波 E1B。Track S 与 HTTP/PI 协议分别规划；FAV 保持独立审计 | `architecture-docs` |
 
 ---
 

.doc/system-design/module-boundaries.yaml

@@ -88,6 +88,7 @@ imports:
     - docgov
     - docrevision
     - notification
+    - events
   app/docflow:
     - app/uow
     - docgov
@@ -247,6 +248,8 @@ table_owners:
     - events
   event_reconciliations:
     - events
+  notifications:
+    - notification
   final_artifacts:
     - artifactvault
   artifact_payload_staging:

.doc/test-reports/2026-09-15-event-e1a-notify-cutover-test-report.md

@@ -0,0 +1,96 @@
+# E1A notify shadow/live/rollback 测试报告
+
+## 一、 测试背景 (Background)
+
+E0B 已交付独立 `cmd/events` 与原子 lease。本波把 `notify` 接到同一 Handler：副作用与 `AckLease` 同事务，用 `(source_event_id, user_id, type)` 保证重复投递只有一条可见通知。生产集群 mode 保持 **off**；闭包仍是当前唯一写者，直到显式 live 切换。
+
+## 二、 测试目标 (Test Objectives)
+
+- Goose 00041 additive：`notifications.source_event_id` + 部分唯一索引。
+- Shadow 对 Handler 做 dry-run（事务 Rollback），只写 `event_reconciliations.intended_digest`，不写 inbox、不 ack。
+- 规范化器比较收件人、类型、去重键 `evt:{eventID}:{type}:{user}`；live 指纹与 shadow intended 一致。
+- Fixture N=30（近 7 天文档事件流量为 0，沿用 E0A 低流量策略）。
+- 缺文档 → quarantine，无通知泄露；live 重放仍一条；mode=off 回滚不再 claim。
+- Handler+Ack 同事务：未提交的二次 Handle 回滚不增加行。
+- `off|shadow|live` 带 revision；FAV 仍 off。
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+| 项 | 值 |
+|----|-----|
+| OS | Linux 7.0.0-31-generic |
+| Go | go1.27.1 linux/amd64 |
+| 分支 | `fix/event-notify-cutover` |
+| Schema | **41** |
+| PostgreSQL | 隔离库 `agent_rel_e1a`（`shuimu-system-pg-test:5434`）；测完 DROP |
+| 项目 143 | 只读 |
+| 默认 mode | off（共享 `agent_system_dev` 未切 notify live） |
+| 证据 | `.doc/test-reports/assets/2026-09-15-event-e1a-notify-cutover/` |
+
+语义：至少一次投递 + 恰好一个可见结果。不是 exactly-once delivery。
+
+## 四、 预期结果 (Expected Results)
+
+| 用例 | 期望 |
+|------|------|
+| shadow | notifications=0；recon intended = `VisibleOutcomeDigest` |
+| live 两次 RunOnce | 同一 `(source_event_id, user_id, type)` 仅 1 行 |
+| live hash | 等于 shadow intended |
+| correlation | envelope → `event_delivery_attempts` → `notifications.source_event_id` |
+| fixture 30 | 30 条独立 approved fixture，shadow/live 各 30 |
+| 缺文档 | delivery=`quarantine`，通知=0 |
+| rollback tx | 已提交行仍为 1 |
+| mode=off | RunOnce claim=0 |
+| 共享库 mode | notify/projection/workflow_advance/fav 全 off |
+
+闭包 `OnOutbox` 不产生 `document.approved` 通知；approved 可见结果只走耐久 envelope。其它文档 outbox 类型仍由闭包写旧 `dedupe_key`，直到 E2 删除 hook。规范化比较用新 canonical key，不要求与闭包旧键逐字相等。
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+### 5.1 规范化器
+
+- **测试输入**: `unset TEST_DATABASE_URL; go test -short ./internal/app/docnotify -run TestIntendedNotifyDigestStable`
+- **实际输出**: PASS；`IntendedNotifyDigest` 与 `events.VisibleOutcomeDigest` 一致
+- **评价判定**: **PASS**
+- **性能与耗时**: <1ms
+
+### 5.2 隔离库 shadow → live → rollback
+
+- **测试输入**: `TEST_DATABASE_URL` → `127.0.0.1:5434/agent_rel_e1a`；`go test -v ./internal/app/docnotify -run 'TestE1A_|TestIntended'`
+- **实际输出**: 见 `assets/.../pg-test.txt`。goose 至 41；`TestE1A_NotifyShadowLiveRollback_PG` PASS（0.29s）；`TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG` PASS（0.06s）
+- **评价判定**: **PASS**
+- **性能与耗时**: 含 goose 0.36s
+
+### 5.3 race
+
+- **测试输入**: 同上隔离库；`go test -race ./internal/app/docnotify -run 'TestE1A_|TestIntended'`
+- **实际输出**: `assets/.../race.txt` PASS（1.14s）
+- **评价判定**: **PASS**
+- **性能与耗时**: 1.14s
+
+### 5.4 共享库默认 mode（只读）
+
+- **测试输入**: `SELECT environment, consumer, mode, revision FROM event_consumer_modes`
+- **实际输出**: 12 行均为 `off` / revision=1（`assets/.../consumer-modes.txt`）
+- **评价判定**: **PASS**（未对共享库 live claim）
+
+### 5.5 全仓 short / arch-check
+
+- **测试输入**: `unset TEST_DATABASE_URL; go test -short -count=1 ./...`；`./scripts/arch-check.sh --mode block`
+- **实际输出**: 全包 PASS；arch-check block PASS
+- **评价判定**: **PASS**
+- **性能与耗时**: 约 15s
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+| 命令 | 结果 |
+|------|------|
+| `go test -short ./...` | PASS |
+| `arch-check --mode block` | PASS |
+| 隔离库 `TestE1A_*` | PASS schema 41 |
+| `go test -race ./internal/app/docnotify` | PASS |
+| 生产 notify mode | **off**；闭包仍是 inbox 唯一写者 |
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+E1A **PASS**。兼容代码已注册 `ConsumerNotify`；集群开关仍 off。下一波 E1B：workflow advance shadow/live/rollback，并发批准只推进一次。不激活 FAV。不删除 `OnOutbox`（E2）。

.doc/test-reports/assets/2026-09-15-event-e1a-notify-cutover/consumer-modes.txt

@@ -0,0 +1,16 @@
+ environment |     consumer     | mode | revision 
+-------------+------------------+------+----------
+ dev         | fav              | off  |        1
+ dev         | notify           | off  |        1
+ dev         | projection       | off  |        1
+ dev         | workflow_advance | off  |        1
+ main        | fav              | off  |        1
+ main        | notify           | off  |        1
+ main        | projection       | off  |        1
+ main        | workflow_advance | off  |        1
+ test        | fav              | off  |        1
+ test        | notify           | off  |        1
+ test        | projection       | off  |        1
+ test        | workflow_advance | off  |        1
+(12 rows)
+

.doc/test-reports/assets/2026-09-15-event-e1a-notify-cutover/pg-test.txt

@@ -0,0 +1,50 @@
+=== RUN   TestIntendedNotifyDigestStable
+--- PASS: TestIntendedNotifyDigestStable (0.00s)
+=== RUN   TestE1A_NotifyShadowLiveRollback_PG
+2026/09/15 09:24:52 OK   00001_baseline.sql (61.04ms)
+2026/09/15 09:24:52 OK   00002_2026-09-09-doc-edit-lease.sql (1.74ms)
+2026/09/15 09:24:52 OK   00003_2026-09-09-uq-context-memory-chunk-hash.sql (1.72ms)
+2026/09/15 09:24:52 OK   00004_2026-09-10-chat-daily-digests-and-search.sql (700.35µs)
+2026/09/15 09:24:52 OK   00005_2026-09-10-chat-messages-day-window.sql (1.11ms)
+2026/09/15 09:24:52 OK   00006_2026-09-11-chat-read-positions.sql (505.41µs)
+2026/09/15 09:24:52 OK   00007_2026-09-11-chat-turns.sql (522µs)
+2026/09/15 09:24:52 OK   00008_2026-09-11-freeform-documents.sql (8.57ms)
+2026/09/15 09:24:52 OK   00009_2026-09-11-message-export-policy.sql (3.06ms)
+2026/09/15 09:24:52 OK   00010_2026-09-11-project-workflows-and-lineage.sql (12.21ms)
+2026/09/15 09:24:52 OK   00011_2026-09-11-seed-ten-projects.sql (5.15ms)
+2026/09/15 09:24:52 OK   00012_2026-09-11-seed-ten-projects-chat.sql (29ms)
+2026/09/15 09:24:52 OK   00013_2026-09-11-task-run-lineage.sql (1.28ms)
+2026/09/15 09:24:52 OK   00014_project_members.sql (3.69ms)
+2026/09/15 09:24:52 OK   00015_2026-09-12-iam-core.sql (16.86ms)
+2026/09/15 09:24:52 OK   00016_2026-09-12-workflow-definitions-and-templates.sql (10.92ms)
+2026/09/15 09:24:52 OK   00017_2026-09-12-project-workflow-instances.sql (14.32ms)
+2026/09/15 09:24:52 OK   00018_2026-09-12-outbox-notifications-stage-advance.sql (8.15ms)
+2026/09/15 09:24:52 OK   00019_2026-09-12-task-runs-events.sql (4.98ms)
+2026/09/15 09:24:52 OK   00020_2026-09-12-task-run-worker-lease.sql (1.07ms)
+2026/09/15 09:24:52 OK   00021_2026-09-12-iam-hardening.sql (2.55ms)
+2026/09/15 09:24:52 OK   00022_2026-09-12-workflow-schema-v2.sql (1.79ms)
+2026/09/15 09:24:52 OK   00023_2026-09-12-instance-integrity.sql (984.4µs)
+2026/09/15 09:24:52 OK   00024_2026-09-12-workflow-catalog-governance.sql (5.91ms)
+2026/09/15 09:24:52 OK   00025_2026-09-13-sync-seed-sequences.sql (885.15µs)
+2026/09/15 09:24:52 OK   00026_2026-09-13-chat-message-scope.sql (2.46ms)
+2026/09/15 09:24:52 OK   00027_2026-09-13-chat-message-scope-backfill.sql (2.02ms)
+2026/09/15 09:24:52 OK   00028_2026-09-13-document-revisions.sql (16.26ms)
+2026/09/15 09:24:52 OK   00029_2026-09-13-freeform-client-key.sql (1.13ms)
+2026/09/15 09:24:52 OK   00030_2026-09-13-task-run-pending-author-confirmation.sql (1.03ms)
+2026/09/15 09:24:52 OK   00031_2026-09-13-project-member-document-duty.sql (1.71ms)
+2026/09/15 09:24:52 OK   00032_2026-09-14-document-revision-migration-repair.sql (732.12µs)
+2026/09/15 09:24:52 OK   00033_2026-09-14-task-run-blocked-pending-governance.sql (567.62µs)
+2026/09/15 09:24:52 OK   00034_2026-09-14-workspace-lease-fencing.sql (621.59µs)
+2026/09/15 09:24:52 OK   00035_2026-09-14-durable-event-delivery.sql (5.09ms)
+2026/09/15 09:24:52 OK   00036_2026-09-14-final-artifact-vault.sql (10.17ms)
+2026/09/15 09:24:52 OK   00037_2026-09-15-task-run-manifests.sql (3.05ms)
+2026/09/15 09:24:52 OK   00038_2026-09-15-dataset-lifecycle.sql (7.75ms)
+2026/09/15 09:24:52 OK   00039_2026-09-15-event-inventory-and-principals.sql (6.37ms)
+2026/09/15 09:24:52 OK   00040_2026-09-15-event-dispatcher.sql (6.45ms)
+2026/09/15 09:24:52 OK   00041_2026-09-15-notification-visible-outcome.sql (1.04ms)
+2026/09/15 09:24:52 goose: successfully migrated database to version: 41
+--- PASS: TestE1A_NotifyShadowLiveRollback_PG (0.29s)
+=== RUN   TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG
+--- PASS: TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG (0.06s)
+PASS
+ok  	chat-agent-system/internal/app/docnotify	0.357s

.doc/test-reports/assets/2026-09-15-event-e1a-notify-cutover/race.txt

@@ -0,0 +1 @@
+ok  	chat-agent-system/internal/app/docnotify	1.137s

cmd/events/main.go

@@ -11,9 +11,12 @@ import (
 	"syscall"
 	"time"
 
+	"chat-agent-system/internal/app/docnotify"
 	"chat-agent-system/internal/config"
 	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docgov"
 	"chat-agent-system/internal/events"
+	"chat-agent-system/internal/notification"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -57,6 +60,10 @@ func main() {
 
 	owner := fmt.Sprintf("events-%d", os.Getpid())
 	disp := events.NewDispatcher(pool, cfg.Env, owner)
+	notify := notification.NewService(pool)
+	_ = disp.Register(events.ConsumerNotify, (&docnotify.EventConsumer{
+		Notify: notify, Docs: docgov.NewGovernanceEngine(pool), 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})

deploy/postgres/goose/00041_2026-09-15-notification-visible-outcome.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+-- E1A: correlate inbox notifications to envelope event_id for exactly-one visible outcome.
+ALTER TABLE notifications
+    ADD COLUMN IF NOT EXISTS source_event_id TEXT NOT NULL DEFAULT '';
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_notification_visible_outcome
+    ON notifications (source_event_id, user_id, type)
+    WHERE source_event_id <> '';
+
+-- +goose Down
+DROP INDEX IF EXISTS uq_notification_visible_outcome;
+ALTER TABLE notifications DROP COLUMN IF EXISTS source_event_id;

internal/app/docnotify/consumer.go

@@ -0,0 +1,82 @@
+package docnotify
+
+import (
+	"context"
+	"encoding/json"
+
+	"chat-agent-system/internal/events"
+	"chat-agent-system/internal/notification"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// EventConsumer is the durable notify worker. Closures remain the unique writer until live cutover.
+type EventConsumer struct {
+	Notify *notification.Service
+	Docs   DocumentLookup
+	Env    string
+}
+
+func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
+	if c == nil || c.Notify == nil {
+		return events.ErrUnknownConsumer
+	}
+	principal := events.ServicePrincipal{
+		ID: "notify-" + c.Env, Environment: c.Env, Consumer: events.ConsumerNotify,
+		Actions: []string{"notify.emit"},
+	}
+	if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "notify.emit"); err != nil {
+		return events.ErrQuarantine
+	}
+	kind, title, body := notifyCopy(lease.EventName)
+	if kind == "" {
+		return nil
+	}
+	docID := lease.Envelope.AggregateID
+	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
+		}
+	}
+	var recipient int64
+	projectID := lease.ProjectID
+	if c.Docs != nil && docID != "" {
+		doc, err := c.Docs.GetDocument(ctx, docID)
+		if err != nil || doc == nil {
+			return events.ErrQuarantine
+		}
+		if doc.ProjectID != lease.ProjectID {
+			return events.ErrQuarantine
+		}
+		recipient = doc.AuthorID
+		projectID = doc.ProjectID
+	}
+	if recipient <= 0 {
+		return events.ErrQuarantine
+	}
+	p := notification.DocumentEventPayload{ProjectID: projectID, DocID: docID}
+	return c.Notify.EmitVisibleDocumentEvent(ctx, tx, lease.EventID, kind, title, body, recipient, p)
+}
+
+func IntendedNotifyDigest(eventID, kind string, recipient int64) string {
+	return events.VisibleOutcomeDigest(eventID, kind, recipient)
+}
+
+func notifyCopy(eventName string) (kind, title, body string) {
+	switch eventName {
+	case events.EventCandidateReady:
+		return notification.TypeDocAgentCandidateReady, "智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。"
+	case events.EventVersionCommitted:
+		return notification.TypeDocAgentVersionCommitted, "智能体版本已生效", "新的智能体文档版本已提交。"
+	case events.EventDownstreamImpact:
+		return notification.TypeDocDownstreamImpact, "下游文档受影响", "上游文档变更，请复核关联文档。"
+	case events.EventChangeReportDegraded:
+		return notification.TypeDocChangeReportDegraded, "变更报告未完整评估", "变更报告降级，受控文档暂不可送审。"
+	case events.EventApproved:
+		return notification.TypeDocApproved, "文档已批准", "文档已完成终审批准。"
+	default:
+		return "", "", ""
+	}
+}

internal/app/docnotify/e1a_pg_test.go

@@ -0,0 +1,349 @@
+package docnotify_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/docnotify"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/events"
+	"chat-agent-system/internal/notification"
+
+	"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 TestIntendedNotifyDigestStable(t *testing.T) {
+	a := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9)
+	b := docnotify.IntendedNotifyDigest("e1", notification.TypeDocApproved, 9)
+	if a == "" || a != b {
+		t.Fatalf("%s %s", a, b)
+	}
+	want := events.VisibleOutcomeDigest("e1", notification.TypeDocApproved, 9)
+	if a != want {
+		t.Fatalf("normalizer drift: %s vs %s", a, want)
+	}
+}
+
+func TestE1A_NotifyShadowLiveRollback_PG(t *testing.T) {
+	pool, ctx := e1aPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	userID, projectID := seedUserProject(t, ctx, pool, suffix)
+	docID := "e1a_" + suffix
+	seedDoc(t, ctx, pool, docID, projectID, userID)
+
+	env := approvedEnv(suffix, docID, projectID, 1)
+	appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify, events.ConsumerWorkflowAdvance})
+
+	d := newNotifyDispatcher(t, pool, "e1a-w")
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
+		t.Fatal(err)
+	}
+	_, revShadow, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify)
+	if err != nil || revShadow < 1 {
+		t.Fatalf("shadow revision=%d err=%v", revShadow, err)
+	}
+	if n, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil || n < 1 {
+		t.Fatalf("shadow n=%d err=%v", n, err)
+	}
+	var notes, recon int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, env.EventID).Scan(&notes); err != nil {
+		t.Fatal(err)
+	}
+	if notes != 0 {
+		t.Fatalf("shadow wrote notifications=%d", notes)
+	}
+	var intended string
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*), MAX(intended_digest) FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, env.EventID).Scan(&recon, &intended); err != nil {
+		t.Fatal(err)
+	}
+	if recon < 1 {
+		t.Fatal("missing shadow recon")
+	}
+	want := docnotify.IntendedNotifyDigest(env.EventID, notification.TypeDocApproved, userID)
+	if intended != want {
+		t.Fatalf("intended digest %s want %s", intended, want)
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
+		t.Fatal(err)
+	}
+	_, revLive, err := events.LoadMode(ctx, pool, "dev", events.ConsumerNotify)
+	if err != nil || revLive <= revShadow {
+		t.Fatalf("live revision %d must exceed shadow %d", revLive, revShadow)
+	}
+	if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1 AND user_id=$2 AND type=$3`,
+		env.EventID, userID, notification.TypeDocApproved).Scan(&notes); err != nil {
+		t.Fatal(err)
+	}
+	if notes != 1 {
+		t.Fatalf("live visible notifications=%d", notes)
+	}
+	liveHash, err := hashCommitted(ctx, pool, env.EventID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if liveHash != intended {
+		t.Fatalf("live hash %s != shadow intended %s", liveHash, intended)
+	}
+
+	var attempts, receipts int
+	if err := pool.QueryRow(ctx, `
+		SELECT
+		  (SELECT COUNT(*) FROM event_delivery_attempts WHERE event_id=$1 AND consumer=$2),
+		  (SELECT COUNT(*) FROM notifications WHERE source_event_id=$1)
+	`, env.EventID, events.ConsumerNotify).Scan(&attempts, &receipts); err != nil {
+		t.Fatal(err)
+	}
+	if attempts < 1 || receipts != 1 {
+		t.Fatalf("correlation attempts=%d receipts=%d", attempts, receipts)
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
+		t.Fatal(err)
+	}
+	n, err := d.RunOnce(ctx, events.ConsumerNotify)
+	if err != nil || n != 0 {
+		t.Fatalf("rollback off claimed=%d err=%v", n, err)
+	}
+}
+
+func TestE1A_NotifyFixtureThirtyAuthzAndRollbackTx_PG(t *testing.T) {
+	pool, ctx := e1aPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	userID, projectID := seedUserProject(t, ctx, pool, suffix)
+	d := newNotifyDispatcher(t, pool, "e1a-n30")
+	d.Batch = 64
+
+	const nFix = 30
+	ids := make([]string, 0, nFix)
+	for i := 0; i < nFix; i++ {
+		docID := fmt.Sprintf("e1a30_%s_%d", suffix, i)
+		seedDoc(t, ctx, pool, docID, projectID, userID)
+		env := approvedEnv(fmt.Sprintf("%s-%d", suffix, i), docID, projectID, int64(i+1))
+		appendEnv(t, ctx, pool, env, []string{events.ConsumerNotify})
+		ids = append(ids, env.EventID)
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
+		t.Fatal(err)
+	}
+	var scanned int
+	for i := 0; i < 8; i++ {
+		n, err := d.RunOnce(ctx, events.ConsumerNotify)
+		if err != nil {
+			t.Fatalf("shadow err=%v", err)
+		}
+		scanned += n
+		var reconNow int
+		if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&reconNow); err != nil {
+			t.Fatal(err)
+		}
+		if reconNow >= nFix {
+			break
+		}
+		if n == 0 {
+			t.Fatalf("shadow stalled at recon=%d scanned=%d", reconNow, scanned)
+		}
+	}
+	var shadowNotes int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1`, "%"+suffix+"%").Scan(&shadowNotes); err != nil {
+		t.Fatal(err)
+	}
+	if shadowNotes != 0 {
+		t.Fatalf("fixture shadow wrote %d notifications", shadowNotes)
+	}
+	var recon int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE mode='shadow' AND event_id LIKE $1`, "%"+suffix+"%").Scan(&recon); err != nil {
+		t.Fatal(err)
+	}
+	if recon < nFix {
+		t.Fatalf("fixture shadow recon=%d want >=%d", recon, nFix)
+	}
+	for _, id := range ids {
+		want := docnotify.IntendedNotifyDigest(id, notification.TypeDocApproved, userID)
+		var got string
+		if err := pool.QueryRow(ctx, `SELECT intended_digest FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, id).Scan(&got); err != nil {
+			t.Fatal(err)
+		}
+		if got != want {
+			t.Fatalf("event %s intended %s want %s", id, got, want)
+		}
+	}
+
+	d.Batch = 64
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 8; i++ {
+		n, err := d.RunOnce(ctx, events.ConsumerNotify)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var liveNow int
+		if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNow); err != nil {
+			t.Fatal(err)
+		}
+		if liveNow >= nFix {
+			break
+		}
+		if n == 0 && i > 0 {
+			t.Fatalf("live stalled at notifications=%d", liveNow)
+		}
+	}
+	var liveNotes int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id LIKE $1 AND type=$2`, "%"+suffix+"%", notification.TypeDocApproved).Scan(&liveNotes); err != nil {
+		t.Fatal(err)
+	}
+	if liveNotes != nFix {
+		t.Fatalf("live unique outcomes=%d want %d", liveNotes, nFix)
+	}
+
+	// same-tx rollback: handle+ack then rollback leaves no extra row
+	missingDoc := approvedEnv("missing-"+suffix, "no-such-doc-"+suffix, projectID, 99)
+	appendEnv(t, ctx, pool, missingDoc, []string{events.ConsumerNotify})
+	if _, err := d.RunOnce(ctx, events.ConsumerNotify); err != nil {
+		t.Fatal(err)
+	}
+	var qStatus string
+	if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, missingDoc.EventID, events.ConsumerNotify).Scan(&qStatus); err != nil {
+		t.Fatal(err)
+	}
+	if qStatus != "quarantine" {
+		t.Fatalf("missing doc status=%s want quarantine", qStatus)
+	}
+	var qNotes int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, missingDoc.EventID).Scan(&qNotes); err != nil {
+		t.Fatal(err)
+	}
+	if qNotes != 0 {
+		t.Fatalf("quarantine leaked notifications=%d", qNotes)
+	}
+
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := (&docnotify.EventConsumer{
+		Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev",
+	}).HandleLease(ctx, events.Lease{
+		EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, Environment: "dev",
+		Envelope: events.Envelope{EventID: ids[0], EventName: events.EventApproved, ProjectID: projectID, AggregateID: "e1a30_" + suffix + "_0", Payload: []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d}`, "e1a30_"+suffix+"_0", projectID))},
+	}, tx); err != nil {
+		t.Fatal(err)
+	}
+	_ = tx.Rollback(ctx)
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1`, ids[0]).Scan(&liveNotes); err != nil {
+		t.Fatal(err)
+	}
+	if liveNotes != 1 {
+		t.Fatalf("rollback of extra handle changed committed count=%d", liveNotes)
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func e1aPool(t *testing.T) (*pgxpool.Pool, context.Context) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("e1a")
+	}
+	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 newNotifyDispatcher(t *testing.T, pool *pgxpool.Pool, owner string) *events.Dispatcher {
+	t.Helper()
+	d := events.NewDispatcher(pool, "dev", owner)
+	if err := d.Register(events.ConsumerNotify, (&docnotify.EventConsumer{
+		Notify: notification.NewService(pool), Docs: docgov.NewGovernanceEngine(pool), Env: "dev",
+	}).HandleLease); err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+
+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`, "e1a-"+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`, "e1a-"+suffix, "e1a").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
+		) VALUES ($1,$2,'intake','t','V0.5',1,'approved',true,false,$3,'A',now(),now())
+	`, docID, projectID, userID); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func approvedEnv(id, docID string, projectID, ver int64) events.Envelope {
+	body := []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d,"task_key":"intake"}`, docID, projectID))
+	return events.Envelope{
+		EventID: "approved-" + id, EventName: events.EventApproved, SchemaVersion: events.EnvelopeSchema,
+		Environment: "dev", ProjectID: projectID, AggregateType: "document_approval", AggregateID: docID,
+		AggregateVersion: ver, ActorRef: "user:3", IdempotencyKey: "approved-" + 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)
+	}
+}
+
+func hashCommitted(ctx context.Context, pool *pgxpool.Pool, eventID string) (string, error) {
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	return events.HashVisibleNotifications(ctx, tx, eventID)
+}

internal/events/dispatcher.go

@@ -82,6 +82,64 @@ func ClaimIDs(ctx context.Context, tx pgx.Tx, consumer string, limit int) ([]str
 	return nil, fmt.Errorf("events: SELECT-only claim is disabled; use ClaimLeases")
 }
 
+const peekSQL = `
+SELECT d.event_id, d.consumer, d.attempt
+FROM event_deliveries d
+INNER JOIN event_envelopes e ON e.event_id = d.event_id
+WHERE d.consumer = $1
+  AND d.consumer <> 'fav'
+  AND d.status IN ('pending','leased')
+  AND NOT EXISTS (
+    SELECT 1 FROM event_inventory_decisions inv
+    WHERE inv.event_id = d.event_id
+      AND (inv.consumer = d.consumer OR inv.consumer = '')
+      AND inv.classification IN ('quarantine','expire')
+  )
+ORDER BY d.next_attempt_at, d.event_id
+LIMIT $2
+`
+
+// PeekPendingLeases hydrates deliveries without claiming. Shadow uses this so
+// live fencing tokens and attempt rows stay untouched.
+func PeekPendingLeases(ctx context.Context, pool *pgxpool.Pool, consumer string, limit int) ([]Lease, error) {
+	if consumer == ConsumerFAV || strings.EqualFold(consumer, "fav") {
+		return nil, fmt.Errorf("%w: fav", ErrDisabledConsumer)
+	}
+	if pool == nil {
+		return nil, fmt.Errorf("events: peek requires pool")
+	}
+	if limit <= 0 {
+		limit = DefaultBatch
+	}
+	rows, err := pool.Query(ctx, peekSQL, consumer, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Lease
+	for rows.Next() {
+		var l Lease
+		if err := rows.Scan(&l.EventID, &l.Consumer, &l.Attempt); err != nil {
+			return nil, err
+		}
+		out = append(out, l)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	for i := range out {
+		if err := hydrateLease(ctx, tx, &out[i]); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
 func ClaimLeases(ctx context.Context, tx pgx.Tx, consumer, owner string, limit int, lease time.Duration) ([]Lease, error) {
 	if consumer == ConsumerFAV || strings.EqualFold(consumer, "fav") {
 		return nil, fmt.Errorf("%w: fav", ErrDisabledConsumer)
@@ -296,3 +354,47 @@ func WriteReconciliation(ctx context.Context, tx pgx.Tx, eventID, consumer, mode
 	`, eventID, consumer, mode, digest)
 	return err
 }
+
+// VisibleOutcomeDigest is the canonical notify comparison key: recipient, type, dedupe.
+func VisibleOutcomeDigest(eventID, notifType string, userID int64) string {
+	return Digest([]byte(visibleOutcomeLine(eventID, notifType, userID)))
+}
+
+func visibleOutcomeLine(eventID, notifType string, userID int64) string {
+	key := fmt.Sprintf("evt:%s:%s:%d", eventID, notifType, userID)
+	return fmt.Sprintf("%d|%s|%s", userID, notifType, key)
+}
+
+func EmptyVisibleDigest() string {
+	return Digest([]byte("empty"))
+}
+
+// HashVisibleNotifications fingerprints inbox rows on the current (possibly uncommitted) tx.
+func HashVisibleNotifications(ctx context.Context, tx pgx.Tx, eventID string) (string, error) {
+	rows, err := tx.Query(ctx, `
+		SELECT user_id, type, COALESCE(dedupe_key, '')
+		FROM notifications
+		WHERE source_event_id = $1
+		ORDER BY user_id, type, dedupe_key
+	`, eventID)
+	if err != nil {
+		return "", err
+	}
+	defer rows.Close()
+	var parts []string
+	for rows.Next() {
+		var userID int64
+		var typ, key string
+		if err := rows.Scan(&userID, &typ, &key); err != nil {
+			return "", err
+		}
+		parts = append(parts, fmt.Sprintf("%d|%s|%s", userID, typ, key))
+	}
+	if err := rows.Err(); err != nil {
+		return "", err
+	}
+	if len(parts) == 0 {
+		return EmptyVisibleDigest(), nil
+	}
+	return Digest([]byte(strings.Join(parts, "\n"))), nil
+}

internal/events/dispatcher_pg_test.go

@@ -12,6 +12,7 @@ import (
 	"chat-agent-system/internal/db"
 	"chat-agent-system/internal/events"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
@@ -113,8 +114,8 @@ func TestE0B_WorkersNoDoubleLeaseAndSerialAggregate_PG(t *testing.T) {
 	}
 	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DROP TABLE IF EXISTS `+visible) })
 
-	handler := func(ctx context.Context, lease events.Lease) error {
-		_, err := pool.Exec(ctx, `INSERT INTO `+visible+` (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, lease.EventID)
+	handler := func(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
+		_, err := tx.Exec(ctx, `INSERT INTO `+visible+` (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, lease.EventID)
 		return err
 	}
 	runWorkers := func(n int) {

internal/events/worker.go

@@ -7,11 +7,13 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
-// Handler runs outside the claim transaction. Side effects must check the fencing token.
-type Handler func(ctx context.Context, lease Lease) error
+// Handler runs outside the claim transaction but must apply side effects on the
+// provided tx so fencing/ack/inbox share one unit of work.
+type Handler func(ctx context.Context, lease Lease, tx pgx.Tx) error
 
 type Dispatcher struct {
 	Pool        *pgxpool.Pool
@@ -145,39 +147,42 @@ func (d *Dispatcher) RunOnce(ctx context.Context, consumer string) (int, error)
 }
 
 func (d *Dispatcher) shadowScan(ctx context.Context, consumer string) (int, error) {
-	rows, err := d.Pool.Query(ctx, `
-		SELECT d.event_id
-		FROM event_deliveries d
-		WHERE d.consumer=$1 AND d.status IN ('pending','leased')
-		ORDER BY d.next_attempt_at, d.event_id
-		LIMIT $2
-	`, consumer, d.Batch)
+	leases, err := PeekPendingLeases(ctx, d.Pool, consumer, d.Batch)
 	if err != nil {
 		return 0, err
 	}
-	defer rows.Close()
+	h := d.Handlers[consumer]
 	var n int
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			return n, err
+	for _, lease := range leases {
+		digest := Digest([]byte(lease.EventID + ":" + consumer + ":" + ModeShadow))
+		if h != nil {
+			tx, err := d.Pool.Begin(ctx)
+			if err != nil {
+				return n, err
+			}
+			herr := h(ctx, lease, tx)
+			if herr != nil {
+				digest = Digest([]byte("fail:" + herr.Error()))
+			} else if dgst, herr := HashVisibleNotifications(ctx, tx, lease.EventID); herr == nil {
+				digest = dgst
+			}
+			_ = tx.Rollback(ctx)
 		}
-		tx, err := d.Pool.Begin(ctx)
+		wtx, err := d.Pool.Begin(ctx)
 		if err != nil {
 			return n, err
 		}
-		digest := Digest([]byte(id + ":" + consumer + ":" + ModeShadow))
-		if err := WriteReconciliation(ctx, tx, id, consumer, ModeShadow, digest); err != nil {
-			_ = tx.Rollback(ctx)
+		if err := WriteReconciliation(ctx, wtx, lease.EventID, consumer, ModeShadow, digest); err != nil {
+			_ = wtx.Rollback(ctx)
 			return n, err
 		}
-		if err := tx.Commit(ctx); err != nil {
+		if err := wtx.Commit(ctx); err != nil {
 			return n, err
 		}
 		n++
 	}
 	d.Touch()
-	return n, rows.Err()
+	return n, nil
 }
 
 func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
@@ -231,7 +236,13 @@ func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
 		_ = d.fail(ctx, lease, "NO_HANDLER", true)
 		return
 	}
-	if err := h(hctx, lease); err != nil {
+	tx, err := d.Pool.Begin(hctx)
+	if err != nil {
+		_ = d.fail(ctx, lease, "TRANSIENT", true)
+		return
+	}
+	if err := h(hctx, lease, tx); err != nil {
+		_ = tx.Rollback(hctx)
 		if hctx.Err() != nil {
 			return
 		}
@@ -239,30 +250,21 @@ func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
 		_ = d.fail(ctx, lease, code, retry)
 		return
 	}
-	if err := d.ack(ctx, lease); err != nil {
+	if err := AckLease(hctx, tx, lease); err != nil {
+		_ = tx.Rollback(hctx)
 		if errors.Is(err, ErrStaleFencing) {
 			d.Metrics.Fenced.Add(1)
 			return
 		}
 		_ = d.fail(ctx, lease, "TRANSIENT", true)
+		return
 	}
-}
-
-func (d *Dispatcher) ack(ctx context.Context, lease Lease) error {
-	tx, err := d.Pool.Begin(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() { _ = tx.Rollback(ctx) }()
-	if err := AckLease(ctx, tx, lease); err != nil {
-		return err
-	}
-	if err := tx.Commit(ctx); err != nil {
-		return err
+	if err := tx.Commit(hctx); err != nil {
+		_ = d.fail(ctx, lease, "TRANSIENT", true)
+		return
 	}
 	d.Metrics.Acked.Add(1)
 	d.Touch()
-	return nil
 }
 
 func (d *Dispatcher) fail(ctx context.Context, lease Lease, code string, retryable bool) error {

internal/notification/document_events.go

@@ -4,19 +4,22 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
+	"github.com/jackc/pgx/v5"
 )
 
 const (
-	EventDocAgentCandidateReady     = "document.agent_candidate_ready"
-	EventDocAgentVersionCommitted   = "document.agent_version_committed"
-	EventDocHumanRevisionConflict   = "document.human_revision_conflict"
-	EventDocDownstreamImpact        = "document.downstream_impact_detected"
-	EventDocChangeReportDegraded    = "document.change_report_degraded"
+	EventDocAgentCandidateReady   = "document.agent_candidate_ready"
+	EventDocAgentVersionCommitted = "document.agent_version_committed"
+	EventDocHumanRevisionConflict = "document.human_revision_conflict"
+	EventDocDownstreamImpact      = "document.downstream_impact_detected"
+	EventDocChangeReportDegraded  = "document.change_report_degraded"
 
 	TypeDocAgentCandidateReady   = "document_agent_candidate_ready"
 	TypeDocAgentVersionCommitted = "document_agent_version_committed"
 	TypeDocDownstreamImpact      = "document_downstream_impact"
 	TypeDocChangeReportDegraded  = "document_change_report_degraded"
+	TypeDocApproved              = "document_approved"
 )
 
 // DocumentEventPayload is a minimal, permission-safe notification body (no markdown excerpts).
@@ -76,6 +79,26 @@ func (s *Service) EmitDocumentEvent(ctx context.Context, eventType, notifType, t
 	return tag.RowsAffected() > 0, nil
 }
 
+// VisibleDedupeKey is the canonical inbox identity for a durable envelope.
+func VisibleDedupeKey(sourceEventID, notifType string, recipientUserID int64) string {
+	return fmt.Sprintf("evt:%s:%s:%d", sourceEventID, notifType, recipientUserID)
+}
+
+// EmitVisibleDocumentEvent writes the unique visible notification on the caller's
+// fenced transaction. A replay that hits (user_id, dedupe_key) is success.
+func (s *Service) EmitVisibleDocumentEvent(ctx context.Context, tx pgx.Tx, sourceEventID, notifType, title, body string, recipientUserID int64, payload DocumentEventPayload) error {
+	if tx == nil || recipientUserID <= 0 || sourceEventID == "" || notifType == "" {
+		return fmt.Errorf("notification: visible outcome requires source event, type, and recipient")
+	}
+	dedupeKey := VisibleDedupeKey(sourceEventID, notifType, recipientUserID)
+	_, err := tx.Exec(ctx, `
+		INSERT INTO notifications (user_id, project_id, type, title, body, dedupe_key, status, source_event_id)
+		VALUES ($1,$2,$3,$4,$5,$6,'unread',$7)
+		ON CONFLICT (user_id, dedupe_key) DO NOTHING
+	`, recipientUserID, payload.ProjectID, notifType, title, body, dedupeKey, sourceEventID)
+	return err
+}
+
 // UpdateNotificationStatus marks a notification read/acted/dismissed for the owner.
 func (s *Service) UpdateNotificationStatus(ctx context.Context, userID, notifID int64, status string) error {
 	if !s.Available() {