merge: fix/event-workflow-advance

2026-09-15 09:43:09 +00:00
parent ebf16aa094 7d1b7e736b
commit 1b1d0a1cba
18 changed files with 693 additions and 10 deletions
@@ -5,7 +5,7 @@

 ## 焦点

-1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B/E1A PASS；下一开工 E1B workflow advance，默认 dispatcher mode 仍 off）
+1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B/E1A/E1B PASS；下一开工 E1C projection，默认 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,7 +60,8 @@
 | 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 |
+| Cursor | event-e1a-notify-cutover | fix/event-notify-cutover | - | events | done |
+| Cursor | event-e1b-workflow-advance | fix/event-workflow-advance | - | events | in_progress |

 > 新任务开始前登记。`branch` 必须为 `feature/<slug>` 或 `fix/<slug>`，**禁止** `dev` / `test` / `main`。同一 `mutex` 同时仅一个 `in_progress`。

@@ -72,7 +73,7 @@

 | holder | phase | since | scope | queue |
 |--------|-------|-------|-------|-------|
-| Cursor | merging | 2026-09-15 | E1A notify cutover；mode=off | - |
+| Cursor | merging | 2026-09-15 | E1B workflow advance；mode=off | - |

 **phase**: `idle` | `merging` | `smoke` | `e2e`

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

@@ -1,6 +1,6 @@
 ---
 title: "E1A notify shadow 与切换"
-status: in_progress
+status: done
 created: 2026-09-15
 updated: 2026-09-15
 owner: Cursor
@@ -0,0 +1,44 @@
+---
+title: "E1B workflow advance shadow 与切换"
+status: in_progress
+created: 2026-09-15
+updated: 2026-09-15
+owner: Cursor
+branch: fix/event-workflow-advance
+worktree: "-"
+mutex: events
+---
+
+# E1B workflow advance shadow 与切换
+
+## 目标
+
+workflow_advance 具备 shadow/live/rollback；并发或重放批准只产生一次阶段可推进可见结果。生产集群保持 off。
+
+## 范围
+
+- Goose 00042、`docadvance.EventConsumer`、`EmitVisibleStageReady`、MarkDocTaskApproved 走 fenced tx
+- cmd/events 注册 workflow_advance
+
+## 非目标
+
+- 不把共享环境 workflow_advance 切 live
+- 不删除 AddOnApprovedHook
+- 不激活 FAV / projection live
+
+## 步骤
+
+1. [x] 从 E1A 后 `dev` 开 `fix/event-workflow-advance`
+2. [x] additive source_event_id on stage_advance_requests
+3. [x] consumer 重读文档并同事务 ack
+4. [x] 隔离库并发/重放/quarantine/rollback
+5. [x] 报告
+
+## 验收标准
+
+- [x] [E1B 测试报告](../../test-reports/2026-09-15-event-e1b-workflow-advance-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-e1b-workflow-advance-test-report.md](../../test-reports/2026-09-15-event-e1b-workflow-advance-test-report.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) | E0A/E0B/E1A 已过隔离库门；dispatcher 与 notify 默认 off。下一波 E1B。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) | E0A–E1B 已过隔离库门；dispatcher/notify/advance 默认 off。下一波 E1C。Track S 与 HTTP/PI 协议分别规划；FAV 保持独立审计 | `architecture-docs` |

 ---

@@ -59,6 +59,7 @@ imports:
  app:
    - agent
    - app/chatflow
+    - app/docadvance
    - app/docflow
    - app/docnotify
    - app/stageflow
@@ -89,6 +90,12 @@ imports:
    - docrevision
    - notification
    - events
+  app/docadvance:
+    - docgov
+    - events
+    - notification
+    - projectinstance
+    - projectworkflow
  app/docflow:
    - app/uow
    - docgov
@@ -250,6 +257,8 @@ table_owners:
    - events
  notifications:
    - notification
+  stage_advance_requests:
+    - notification
  final_artifacts:
    - artifactvault
  artifact_payload_staging:
@@ -310,6 +319,7 @@ event_producers:
  document.approved:
    - app/docflow
    - events
+    - app/docadvance
  taskrun.completed:
    - taskrun
  taskrun.cancelled:
@@ -0,0 +1,79 @@
+# E1B workflow advance shadow/live/rollback 测试报告
+
+## 一、 测试背景 (Background)
+
+E1A 已把 notify 接到同事务 Handler。本波把 `document.approved` 的 workflow_advance 接到同一 dispatcher：重读文档/项目/阶段、service principal 鉴权，用已有 `stage_advance_requests (stage_instance_id, readiness_revision)` 唯一约束保证并发/重放只产生一次阶段可推进可见结果。生产集群 mode 保持 **off**。
+
+## 二、 测试目标 (Test Objectives)
+
+- Goose 00042 additive：`stage_advance_requests.source_event_id` + 部分唯一索引。
+- Shadow dry-run 不写 `stage_advance_requests`；MarkDocTaskApproved 走 fenced tx，Rollback 可撤销。
+- Live 两次 RunOnce + 双 worker 并行：同一 workflow instance 仅 1 条 stage-ready 请求与 1 条通知。
+- 缺文档 / 跨项目 → quarantine。
+- 第二份 approved envelope 不新增 stage_advance_requests。
+- mode=off 回滚不再 claim。
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+| 项 | 值 |
+|----|-----|
+| OS | Linux 7.0.0-31-generic |
+| Go | go1.27.1 linux/amd64 |
+| 分支 | `fix/event-workflow-advance` |
+| Schema | **42** |
+| PostgreSQL | 隔离库 `agent_rel_e1b`（`:5434`）；测完 DROP |
+| 项目 143 | 只读 |
+| 默认 mode | off |
+| 证据 | `.doc/test-reports/assets/2026-09-15-event-e1b-workflow-advance/` |
+
+语义：至少一次投递 + 恰好一个可见结果。不是 exactly-once delivery。
+
+## 四、 预期结果 (Expected Results)
+
+| 用例 | 期望 |
+|------|------|
+| shadow | stage_advance_requests=0；有 recon |
+| live 并行 + 再 RunOnce | requests=1；notifications=1 |
+| 重复 approved 事件 | requests 仍为 1 |
+| missing doc | delivery=quarantine |
+| mode=off | claim=0 |
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+### 5.1 隔离库 PG
+
+- **测试输入**: `TEST_DATABASE_URL` → `127.0.0.1:5434/agent_rel_e1b`；`go test -v ./internal/app/docadvance -run TestE1B_`
+- **实际输出**: `assets/.../pg-test.txt` goose 至 42；`TestE1B_AdvanceShadowLiveRollback_PG` PASS（0.33s）
+- **评价判定**: **PASS**
+- **性能与耗时**: 含 goose 0.33s
+
+### 5.2 race
+
+- **测试输入**: 同上；`go test -race ./internal/app/docadvance -run TestE1B_`
+- **实际输出**: `assets/.../race.txt` PASS（1.12s）
+- **评价判定**: **PASS**
+
+### 5.3 共享库 mode（只读）
+
+- **测试输入**: `SELECT environment, consumer, mode FROM event_consumer_modes`
+- **实际输出**: 全部 off（`assets/.../consumer-modes.txt`）
+- **评价判定**: **PASS**
+
+### 5.4 arch-check / short
+
+- **测试输入**: `./scripts/arch-check.sh --mode block`；`go test -short ./internal/events ./internal/archtest ./internal/app/docadvance`
+- **实际输出**: PASS
+- **评价判定**: **PASS**
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+| 命令 | 结果 |
+|------|------|
+| 隔离库 `TestE1B_*` | PASS schema 42 |
+| `go test -race ./internal/app/docadvance` | PASS |
+| `arch-check --mode block` | PASS |
+| 生产 workflow_advance mode | **off**；闭包 `AddOnApprovedHook` 仍是 HTTP 进程写者 |
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+E1B **PASS**。cmd/events 已注册 workflow_advance，集群开关仍 off。下一波 E1C：projection no-replace + 三类 worker 重启 + 授权页面刷新。不激活 FAV。不删除批准 hook（E2）。
@@ -0,0 +1,16 @@
+ environment |     consumer     | mode 
+-------------+------------------+------
+ dev         | fav              | off
+ dev         | notify           | off
+ dev         | projection       | off
+ dev         | workflow_advance | off
+ main        | fav              | off
+ main        | notify           | off
+ main        | projection       | off
+ main        | workflow_advance | off
+ test        | fav              | off
+ test        | notify           | off
+ test        | projection       | off
+ test        | workflow_advance | off
+(12 rows)
+
@@ -0,0 +1,47 @@
+=== RUN   TestE1B_AdvanceShadowLiveRollback_PG
+2026/09/15 09:40:43 OK   00001_baseline.sql (53.34ms)
+2026/09/15 09:40:43 OK   00002_2026-09-09-doc-edit-lease.sql (2.64ms)
+2026/09/15 09:40:43 OK   00003_2026-09-09-uq-context-memory-chunk-hash.sql (1.37ms)
+2026/09/15 09:40:43 OK   00004_2026-09-10-chat-daily-digests-and-search.sql (851.49µs)
+2026/09/15 09:40:43 OK   00005_2026-09-10-chat-messages-day-window.sql (937.18µs)
+2026/09/15 09:40:43 OK   00006_2026-09-11-chat-read-positions.sql (432.67µs)
+2026/09/15 09:40:43 OK   00007_2026-09-11-chat-turns.sql (459.93µs)
+2026/09/15 09:40:43 OK   00008_2026-09-11-freeform-documents.sql (8.96ms)
+2026/09/15 09:40:43 OK   00009_2026-09-11-message-export-policy.sql (4.22ms)
+2026/09/15 09:40:43 OK   00010_2026-09-11-project-workflows-and-lineage.sql (12.49ms)
+2026/09/15 09:40:44 OK   00011_2026-09-11-seed-ten-projects.sql (5.39ms)
+2026/09/15 09:40:44 OK   00012_2026-09-11-seed-ten-projects-chat.sql (30.46ms)
+2026/09/15 09:40:44 OK   00013_2026-09-11-task-run-lineage.sql (1.47ms)
+2026/09/15 09:40:44 OK   00014_project_members.sql (3.58ms)
+2026/09/15 09:40:44 OK   00015_2026-09-12-iam-core.sql (16.97ms)
+2026/09/15 09:40:44 OK   00016_2026-09-12-workflow-definitions-and-templates.sql (11.85ms)
+2026/09/15 09:40:44 OK   00017_2026-09-12-project-workflow-instances.sql (14.83ms)
+2026/09/15 09:40:44 OK   00018_2026-09-12-outbox-notifications-stage-advance.sql (9.49ms)
+2026/09/15 09:40:44 OK   00019_2026-09-12-task-runs-events.sql (5.74ms)
+2026/09/15 09:40:44 OK   00020_2026-09-12-task-run-worker-lease.sql (1.05ms)
+2026/09/15 09:40:44 OK   00021_2026-09-12-iam-hardening.sql (2.48ms)
+2026/09/15 09:40:44 OK   00022_2026-09-12-workflow-schema-v2.sql (1.76ms)
+2026/09/15 09:40:44 OK   00023_2026-09-12-instance-integrity.sql (999.81µs)
+2026/09/15 09:40:44 OK   00024_2026-09-12-workflow-catalog-governance.sql (6ms)
+2026/09/15 09:40:44 OK   00025_2026-09-13-sync-seed-sequences.sql (882.03µs)
+2026/09/15 09:40:44 OK   00026_2026-09-13-chat-message-scope.sql (2.52ms)
+2026/09/15 09:40:44 OK   00027_2026-09-13-chat-message-scope-backfill.sql (2.02ms)
+2026/09/15 09:40:44 OK   00028_2026-09-13-document-revisions.sql (14.72ms)
+2026/09/15 09:40:44 OK   00029_2026-09-13-freeform-client-key.sql (1.13ms)
+2026/09/15 09:40:44 OK   00030_2026-09-13-task-run-pending-author-confirmation.sql (987.41µs)
+2026/09/15 09:40:44 OK   00031_2026-09-13-project-member-document-duty.sql (1.65ms)
+2026/09/15 09:40:44 OK   00032_2026-09-14-document-revision-migration-repair.sql (615.34µs)
+2026/09/15 09:40:44 OK   00033_2026-09-14-task-run-blocked-pending-governance.sql (467.41µs)
+2026/09/15 09:40:44 OK   00034_2026-09-14-workspace-lease-fencing.sql (457.44µs)
+2026/09/15 09:40:44 OK   00035_2026-09-14-durable-event-delivery.sql (4.93ms)
+2026/09/15 09:40:44 OK   00036_2026-09-14-final-artifact-vault.sql (8.41ms)
+2026/09/15 09:40:44 OK   00037_2026-09-15-task-run-manifests.sql (2.33ms)
+2026/09/15 09:40:44 OK   00038_2026-09-15-dataset-lifecycle.sql (7.16ms)
+2026/09/15 09:40:44 OK   00039_2026-09-15-event-inventory-and-principals.sql (5.73ms)
+2026/09/15 09:40:44 OK   00040_2026-09-15-event-dispatcher.sql (6.9ms)
+2026/09/15 09:40:44 OK   00041_2026-09-15-notification-visible-outcome.sql (1.13ms)
+2026/09/15 09:40:44 OK   00042_2026-09-15-workflow-advance-outcome.sql (1.03ms)
+2026/09/15 09:40:44 goose: successfully migrated database to version: 42
+--- PASS: TestE1B_AdvanceShadowLiveRollback_PG (0.33s)
+PASS
+ok  	chat-agent-system/internal/app/docadvance	0.334s
@@ -0,0 +1 @@
+ok  	chat-agent-system/internal/app/docadvance	1.115s
@@ -11,12 +11,16 @@ import (
 	"syscall"
 	"time"

+	"chat-agent-system/internal/app/docadvance"
 	"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"
+	"chat-agent-system/internal/projectinstance"
+	"chat-agent-system/internal/projectworkflow"
+	"chat-agent-system/internal/workflowdef"

 	"github.com/jackc/pgx/v5/pgxpool"
 )
@@ -61,8 +65,17 @@ func main() {
 	owner := fmt.Sprintf("events-%d", os.Getpid())
 	disp := events.NewDispatcher(pool, cfg.Env, owner)
 	notify := notification.NewService(pool)
+	gov := docgov.NewGovernanceEngine(pool)
+	wfDef := workflowdef.NewService(pool, nil)
+	pwSvc := projectworkflow.NewService(projectworkflow.NewStore(pool), nil)
+	instSvc := projectinstance.NewService(
+		projectinstance.NewStore(pool), wfDef.Store(), pwSvc.Store(), projectinstance.NewDocGovChecker(gov),
+	)
 	_ = disp.Register(events.ConsumerNotify, (&docnotify.EventConsumer{
-		Notify: notify, Docs: docgov.NewGovernanceEngine(pool), Env: cfg.Env,
+		Notify: notify, Docs: gov, Env: cfg.Env,
+	}).HandleLease)
+	_ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{
+		Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env,
 	}).HandleLease)
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
@@ -0,0 +1,12 @@
+-- +goose Up
+-- E1B: correlate stage-ready unique outcome to envelope event_id.
+ALTER TABLE stage_advance_requests
+    ADD COLUMN IF NOT EXISTS source_event_id TEXT NOT NULL DEFAULT '';
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_sar_source_event
+    ON stage_advance_requests (source_event_id)
+    WHERE source_event_id <> '';
+
+-- +goose Down
+DROP INDEX IF EXISTS uq_sar_source_event;
+ALTER TABLE stage_advance_requests DROP COLUMN IF EXISTS source_event_id;
@@ -0,0 +1,105 @@
+package docadvance
+
+import (
+	"context"
+	"encoding/json"
+
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/events"
+	"chat-agent-system/internal/notification"
+	"chat-agent-system/internal/projectinstance"
+	"chat-agent-system/internal/projectworkflow"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// EventConsumer applies document.approved onto workflow task bookkeeping.
+// Closures remain the unique writer until live cutover.
+type EventConsumer struct {
+	Inst   *projectinstance.Service
+	Docs   DocumentLookup
+	PW     ProjectLookup
+	Notify *notification.Service
+	Env    string
+}
+
+type DocumentLookup interface {
+	GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
+}
+
+type ProjectLookup interface {
+	GetProject(ctx context.Context, projectID, userID int64) (*projectworkflow.Project, error)
+}
+
+func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
+	if c == nil || c.Inst == nil {
+		return events.ErrUnknownConsumer
+	}
+	principal := events.ServicePrincipal{
+		ID: "advance-" + c.Env, Environment: c.Env, Consumer: events.ConsumerWorkflowAdvance,
+		Actions: []string{"workflow.advance"},
+	}
+	if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "workflow.advance"); err != nil {
+		return events.ErrQuarantine
+	}
+	if lease.EventName != events.EventApproved {
+		return nil
+	}
+	docID := lease.Envelope.AggregateID
+	taskKey := ""
+	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 k, ok := payload["task_key"].(string); ok {
+			taskKey = k
+		}
+	}
+	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
+	}
+	if doc.CurrentState != docgov.StateApproved {
+		return events.ErrQuarantine
+	}
+	if taskKey == "" {
+		taskKey = doc.TaskKey
+	}
+	if taskKey == "" {
+		return events.ErrQuarantine
+	}
+
+	ctx = projectinstance.WithTx(ctx, tx)
+	result, err := c.Inst.HandleDocumentApproved(ctx, doc.ProjectID, taskKey)
+	if err != nil {
+		return err
+	}
+	if result == nil || !result.StageReady {
+		return nil
+	}
+	if c.Notify == nil || c.PW == nil {
+		return nil
+	}
+	project, err := c.PW.GetProject(ctx, doc.ProjectID, 0)
+	if err != nil || project == nil || project.CreatorUserID <= 0 {
+		return events.ErrQuarantine
+	}
+	return c.Notify.EmitVisibleStageReady(ctx, tx, lease.EventID, project.CreatorUserID, notification.StageReadyPayload{
+		ProjectID:          doc.ProjectID,
+		WorkflowInstanceID: result.InstanceID,
+		StageInstanceID:    result.StageInstanceID,
+		StageKey:           result.StageKey,
+		StageName:          result.StageName,
+		ConfigRevision:     result.ConfigRevision,
+		RequiredDocCount:   result.RequiredDocCount,
+		ApprovedDocCount:   result.ApprovedDocCount,
+	})
+}
@@ -0,0 +1,260 @@
+package docadvance_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/docadvance"
+	"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"
+	"chat-agent-system/internal/projectinstance"
+	"chat-agent-system/internal/projectworkflow"
+	"chat-agent-system/internal/workflowdef"
+
+	"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 TestE1B_AdvanceShadowLiveRollback_PG(t *testing.T) {
+	pool, ctx := e1bPool(t)
+	fix := seedAdvanceFixture(t, ctx, pool, "one")
+
+	env := approvedEnvelope(fix.suffix+"-a", fix.docID, fix.projectID, 1)
+	appendEnv(t, ctx, pool, env, []string{events.ConsumerWorkflowAdvance})
+
+	d := newAdvanceDispatcher(t, pool, "e1b-w")
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeShadow); err != nil {
+		t.Fatal(err)
+	}
+	if n, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil || n < 1 {
+		t.Fatalf("shadow n=%d err=%v", n, err)
+	}
+	var reqs int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE source_event_id=$1`, env.EventID).Scan(&reqs); err != nil {
+		t.Fatal(err)
+	}
+	if reqs != 0 {
+		t.Fatalf("shadow wrote stage_advance_requests=%d", reqs)
+	}
+	var recon int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_reconciliations WHERE event_id=$1 AND mode='shadow'`, env.EventID).Scan(&recon); err != nil {
+		t.Fatal(err)
+	}
+	if recon < 1 {
+		t.Fatal("missing shadow recon")
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeLive); err != nil {
+		t.Fatal(err)
+	}
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, _ = d.RunOnce(ctx, events.ConsumerWorkflowAdvance)
+		}()
+	}
+	wg.Wait()
+	if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE workflow_instance_id=$1`, fix.instanceID).Scan(&reqs); err != nil {
+		t.Fatal(err)
+	}
+	if reqs != 1 {
+		t.Fatalf("concurrent/live unique stage_advance_requests=%d", reqs)
+	}
+	var notes int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM notifications WHERE source_event_id=$1 AND type=$2`,
+		env.EventID, notification.TypeStageReadyToAdvance).Scan(&notes); err != nil {
+		t.Fatal(err)
+	}
+	if notes != 1 {
+		t.Fatalf("visible stage-ready notifications=%d", notes)
+	}
+
+	dup := approvedEnvelope(fix.suffix+"-dup", fix.docID, fix.projectID, 2)
+	appendEnv(t, ctx, pool, dup, []string{events.ConsumerWorkflowAdvance})
+	if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM stage_advance_requests WHERE workflow_instance_id=$1`, fix.instanceID).Scan(&reqs); err != nil {
+		t.Fatal(err)
+	}
+	if reqs != 1 {
+		t.Fatalf("second approved event created extra request=%d", reqs)
+	}
+
+	stale := approvedEnvelope(fix.suffix+"-stale", "missing-doc-"+fix.suffix, fix.projectID, 3)
+	appendEnv(t, ctx, pool, stale, []string{events.ConsumerWorkflowAdvance})
+	if _, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance); 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.ConsumerWorkflowAdvance).Scan(&status); err != nil {
+		t.Fatal(err)
+	}
+	if status != "quarantine" {
+		t.Fatalf("stale/missing doc status=%s", status)
+	}
+
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerWorkflowAdvance, events.ModeOff); err != nil {
+		t.Fatal(err)
+	}
+	n, err := d.RunOnce(ctx, events.ConsumerWorkflowAdvance)
+	if err != nil || n != 0 {
+		t.Fatalf("rollback claimed=%d err=%v", n, err)
+	}
+}
+
+func e1bPool(t *testing.T) (*pgxpool.Pool, context.Context) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("e1b")
+	}
+	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 newAdvanceDispatcher(t *testing.T, pool *pgxpool.Pool, owner string) *events.Dispatcher {
+	t.Helper()
+	gov := docgov.NewGovernanceEngine(pool)
+	wfDef := workflowdef.NewService(pool, nil)
+	pwSvc := projectworkflow.NewService(projectworkflow.NewStore(pool), nil)
+	inst := projectinstance.NewService(projectinstance.NewStore(pool), wfDef.Store(), pwSvc.Store(), projectinstance.NewDocGovChecker(gov))
+	d := events.NewDispatcher(pool, "dev", owner)
+	if err := d.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{
+		Inst: inst, Docs: gov, PW: pwSvc, Notify: notification.NewService(pool), Env: "dev",
+	}).HandleLease); err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+
+type advanceFix struct {
+	suffix, docID string
+	projectID     int64
+	instanceID    int64
+}
+
+func seedAdvanceFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool, tag string) advanceFix {
+	t.Helper()
+	suffix := fmt.Sprintf("%s-%d", tag, time.Now().UnixNano())
+	var author, reviewer, approver int64
+	if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'A','engineer') RETURNING id`, "e1b-a-"+suffix).Scan(&author); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'R','engineer') RETURNING id`, "e1b-r-"+suffix).Scan(&reviewer); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `INSERT INTO users (account_id, name, role) VALUES ($1,'P','engineer') RETURNING id`, "e1b-p-"+suffix).Scan(&approver); err != nil {
+		t.Fatal(err)
+	}
+	if err := workflowdef.BootstrapSeeds(ctx, pool, nil, "templates/docs", author); err != nil {
+		t.Fatal(err)
+	}
+	wfStore := workflowdef.NewStore(pool)
+	pwStore := projectworkflow.NewStore(pool)
+	instStore := projectinstance.NewStore(pool)
+	gov := docgov.NewGovernanceEngine(pool)
+	svc := projectinstance.NewService(instStore, wfStore, pwStore, projectinstance.NewDocGovChecker(gov))
+
+	var projectID int64
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision)
+		VALUES ($1, 'E1B', 'active', $2, 'planning', '', 1, 0) RETURNING id
+	`, "e1b-"+suffix, author).Scan(&projectID); err != nil {
+		t.Fatal(err)
+	}
+	plan, err := svc.BuildDefaultPlan(ctx, workflowdef.OfficialKeyMain, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := range plan.Stages {
+		for j := range plan.Stages[i].DocTasks {
+			plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
+				AuthorUserID: author, ReviewerUserID: reviewer, ApproverUserID: approver,
+			}
+		}
+	}
+	if _, err := svc.SaveWorkflowPlan(ctx, projectID, author, true, projectinstance.SavePlanInput{
+		Plan: *plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	preview, err := svc.ActivationPreview(ctx, projectID, author, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	activated, err := svc.Activate(ctx, projectID, author, true, projectinstance.ActivateInput{
+		ConfirmationToken:      preview.ConfirmationToken,
+		ExpectedProjectVersion: preview.ExpectedProjectVersion,
+		ExpectedPlanRevision:   preview.ExpectedPlanRevision,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	docID := "doc_e1b_intake_" + suffix
+	if _, err := gov.CreateOrGetDocument(ctx, docID, projectID, "intake", "客户需求采集", "V0.1", "", author, "A", reviewer, "R", approver, "P"); err != nil {
+		t.Fatal(err)
+	}
+	_ = gov.MarkDocumentContentUpdated(ctx, docID, "hash_e1b", false)
+	if _, err := gov.SubmitForReview(ctx, docID, author, "author", "A"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.ReviewApprove(ctx, docID, reviewer, "reviewer", "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.ApproverApprove(ctx, docID, approver, "approver", "P", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	return advanceFix{suffix: suffix, docID: docID, projectID: projectID, instanceID: activated.Instance.ID}
+}
+
+func approvedEnvelope(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:4", 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)
+	}
+}
@@ -40,7 +40,7 @@ func StaticContracts() []Contract {
 		{EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, 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"}, Consumers: []string{ConsumerWorkflowAdvance, ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id", "task_key"}},
+		{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"}},
 		{EventName: EventTaskRunCompleted, SchemaVersion: 1, Enabled: true, Producers: []string{"taskrun"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"run_id"}},
 		{EventName: EventTaskRunCancelled, SchemaVersion: 1, Enabled: true, Producers: []string{"taskrun"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"run_id"}},
 		{EventName: EventFAVArtifactStaged, SchemaVersion: 1, Enabled: false, Producers: []string{"app/docflow"}, Consumers: []string{ConsumerFAV}, PayloadFields: []string{"doc_id", "snapshot_id"}},
@@ -7,6 +7,7 @@ import (
 	"fmt"

 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )

@@ -107,6 +108,69 @@ func (s *Service) EmitStageReadyToAdvance(ctx context.Context, creatorUserID int
 	return true, nil
 }

+// EmitVisibleStageReady writes the unique stage-ready request on the caller's fenced tx.
+// Duplicate (stage, revision) or replayed source_event_id is success.
+func (s *Service) EmitVisibleStageReady(ctx context.Context, tx pgx.Tx, sourceEventID string, creatorUserID int64, payload StageReadyPayload) error {
+	if tx == nil || sourceEventID == "" || creatorUserID <= 0 || payload.StageInstanceID <= 0 {
+		return fmt.Errorf("notification: visible stage ready requires source event, creator, and stage")
+	}
+	var advanceID int64
+	err := tx.QueryRow(ctx, `
+		INSERT INTO stage_advance_requests (
+			stage_instance_id, workflow_instance_id, stage_key, readiness_revision, status, source_event_id
+		) VALUES ($1, $2, $3, $4, 'pending', $5)
+		ON CONFLICT (stage_instance_id, readiness_revision) DO NOTHING
+		RETURNING id
+	`, payload.StageInstanceID, payload.WorkflowInstanceID, payload.StageKey, payload.ConfigRevision, sourceEventID).Scan(&advanceID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil
+	}
+	if err != nil {
+		return ignoreUnique(err)
+	}
+
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	var eventID int64
+	aggregateID := fmt.Sprintf("stage_instance:%d", payload.StageInstanceID)
+	if err := tx.QueryRow(ctx, `
+		INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at)
+		VALUES ('stage_instance', $1, $2, $3::jsonb, $4, now())
+		RETURNING id
+	`, aggregateID, EventStageReadyToAdvance, string(raw), payload.ProjectID).Scan(&eventID); err != nil {
+		return err
+	}
+	stageLabel := payload.StageName
+	if stageLabel == "" {
+		stageLabel = payload.StageKey
+	}
+	title := fmt.Sprintf("阶段「%s」可推进", stageLabel)
+	body := fmt.Sprintf(
+		"%d/%d 个必需文档已批准，没有阻断项。请确认是否推进到下一阶段。",
+		payload.ApprovedDocCount, payload.RequiredDocCount,
+	)
+	dedupeKey := fmt.Sprintf("stage_ready:%d:%d", payload.StageInstanceID, payload.ConfigRevision)
+	_, err = tx.Exec(ctx, `
+		INSERT INTO notifications (user_id, project_id, event_id, type, title, body, dedupe_key, status, source_event_id)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,'unread',$8)
+		ON CONFLICT (user_id, dedupe_key) DO NOTHING
+	`, creatorUserID, payload.ProjectID, eventID, TypeStageReadyToAdvance, title, body, dedupeKey, sourceEventID)
+	return ignoreUnique(err)
+}
+
+func ignoreUnique(err error) error {
+	if err == nil {
+		return nil
+	}
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+		return nil
+	}
+	return err
+}
+
 // ListProjectNotifications 列出项目下当前用户的通知（可选 since_id 增量）。
 func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectID int64, sinceID int64, limit int) ([]Notification, error) {
 	if !s.Available() {
@@ -7,6 +7,7 @@ import (
 	"time"

 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 )

 func (s *pgStore) GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error) {
@@ -184,7 +185,11 @@ func (s *pgStore) MarkDocTaskApproved(ctx context.Context, projectID int64, task
 	if projectID <= 0 || taskKey == "" {
 		return nil
 	}
-	_, err := s.pool.Exec(ctx, `
+	q := dbExec(s.pool)
+	if tx := txFrom(ctx); tx != nil {
+		q = tx
+	}
+	_, err := q.Exec(ctx, `
 		UPDATE project_doc_task_instances dti
 		SET status = 'approved', updated_at = now()
 		FROM project_stage_instances psi
@@ -197,3 +202,7 @@ func (s *pgStore) MarkDocTaskApproved(ctx context.Context, projectID int64, task
 	`, projectID, taskKey)
 	return err
 }
+
+type dbExec interface {
+	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
+}
@@ -0,0 +1,22 @@
+package projectinstance
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+)
+
+type txKey struct{}
+
+// WithTx binds a fenced worker transaction so bookkeeping writes join the same UoW.
+func WithTx(ctx context.Context, tx pgx.Tx) context.Context {
+	if tx == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, txKey{}, tx)
+}
+
+func txFrom(ctx context.Context) pgx.Tx {
+	tx, _ := ctx.Value(txKey{}).(pgx.Tx)
+	return tx
+}
				`@@ -0,0 +1 @@`
				`ok chat-agent-system/internal/app/docadvance 1.115s`