merge: feature/events-dispatcher-pg

2026-09-15 09:08:25 +00:00
parent 5f3b117c00 d926f64c2e
commit 06b824ad7c
20 changed files with 1374 additions and 42 deletions
@@ -5,7 +5,7 @@

 ## 焦点

-1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A PASS，见 [E0A 报告](../test-reports/2026-09-15-event-e0a-production-scope-contract-test-report.md)；下一开工为 E0B PG dispatcher，默认 mode=off）
+1. **TRACK-E**: [持久事件交付](./short-term/2026-09-15-durable-event-track-e.md)（E0A/E0B PASS；下一开工 E1A notify shadow→live，默认 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；编制任务链仍开放）

@@ -59,6 +59,7 @@
 | implementer_cursor | governance-action-level-authz | fix/governance-action-level-authz | - | workflow | done |
 | 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 |

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

@@ -70,7 +71,7 @@

 | holder | phase | since | scope | queue |
 |--------|-------|-------|-------|-------|
-| Cursor | merging | 2026-09-15 | E0A 生产者/契约/历史数据门；dispatcher 保持 off | - |
+| Cursor | merging | 2026-09-15 | E0B dispatcher 基座；mode=off | - |

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

@@ -29,7 +29,7 @@ mutex: events
 ## 步骤

 1. [x] E0A 单独建 short-term/分支，dispatcher 保持 off。
-2. [ ] E0A 报告通过后再认领 E0B。
+2. [x] E0A 报告通过后再认领 E0B。
 3. [ ] 依次交付 E1A notify、E1B workflow advance、E1C projection；每步独立 go/no-go。
 4. [ ] 一个兼容发布窗口后 contract 删除闭包 API。
 5. [ ] 同步 test report、CURRENT、FEATURE-MATRIX、gap-matrix。
@@ -0,0 +1,43 @@
+---
+title: "E0B PG dispatcher 基座"
+status: done
+created: 2026-09-15
+updated: 2026-09-15
+owner: Cursor
+branch: feature/events-dispatcher-pg
+worktree: "-"
+mutex: events
+---
+
+# E0B PG dispatcher 基座
+
+## 目标
+
+原子 lease/attempt/heartbeat/fencing、独立 `cmd/events`、默认 mode=off、1/2/4 worker 与崩溃注入。
+
+## 范围
+
+- Goose 00040、ClaimLeases、cmd/events、compose events 服务
+
+## 非目标
+
+- 不把任何 consumer 切 live
+- 不激活 FAV
+- 不删除历史
+
+## 步骤
+
+1. [x] 从 E0A 后 `dev` 开 `feature/events-dispatcher-pg`
+2. [x] additive schema + atomic claim
+3. [x] worker/heartbeat/fencing/shadow scan
+4. [x] cmd/events + compose
+5. [x] 隔离库 PG + race + 报告
+
+## 验收标准
+
+- [x] [E0B 测试报告](../../test-reports/2026-09-15-event-e0b-dispatcher-pg-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-e0b-dispatcher-pg-test-report.md](../../test-reports/2026-09-15-event-e0b-dispatcher-pg-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)；[U11–U12 报告](../test-reports/2026-09-15-app-composition-u11-u12-test-report.md)；[E0A 报告](../test-reports/2026-09-15-event-e0a-production-scope-contract-test-report.md) | U5–U12 已合入 `dev@bf59cce`。E0A 生产者/契约/inventory 已过隔离库门；PG dispatcher 仍 off。Track S 与 HTTP/PI 协议分别规划，不阻塞 Track E；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) | E0A/E0B 已过隔离库门；dispatcher 默认 off。Track S 与 HTTP/PI 协议分别规划；FAV 保持独立审计 | `architecture-docs` |

 ---

@@ -239,6 +239,14 @@ table_owners:
    - events
  event_service_principals:
    - events
+  event_delivery_attempts:
+    - events
+  event_consumer_modes:
+    - events
+  event_consumer_checkpoints:
+    - events
+  event_reconciliations:
+    - events
  final_artifacts:
    - artifactvault
  artifact_payload_staging:
@@ -0,0 +1,85 @@
+# E0B PG Dispatcher 基座测试报告
+
+## 一、 测试背景 (Background)
+
+E0A 已证明三条文档链在同一 UoW 写 envelope/delivery，dispatcher 保持 off。本波交付独立 `cmd/events` 进程、原子 lease claim、不可变 attempt、fencing 与心跳，并禁止把 MemoryBus gap wait 搬到 PostgreSQL。
+
+## 二、 测试目标 (Test Objectives)
+
+- Goose 00040 只 additive：attempts、consumer modes、checkpoints、reconciliations。
+- Claim 为 `WITH candidates … FOR UPDATE SKIP LOCKED` + 同事务 attempt 行；SELECT-only `ClaimIDs` 禁用。
+- 状态机 `pending → leased → acked`，可回 pending / dead / quarantine。
+- 1/2/4 worker 无双 lease；同 aggregate 串行；FAV 不可 claim；quarantine 不可 claim。
+- 副作用后/ack 前崩溃：旧 token ack 失败，重放后恰好一个可见结果。
+- Shadow 只写 reconciliation，不 ack。
+- 默认集群 mode=off；HTTP Runtime 不启动 dispatcher。
+- 连接池 MaxConns=8。
+
+## 三、 测试方案与环境 (Test Plan & Environment)
+
+| 项 | 值 |
+|----|-----|
+| OS | Linux 7.0.0-31-generic |
+| Go | go1.27.1 linux/amd64 |
+| 分支 | `feature/events-dispatcher-pg` |
+| Schema | **40** |
+| PostgreSQL | 隔离库 `agent_rel_e0b`（`shuimu-system-pg-test:5434`）；测完 DROP |
+| 项目 143 | 只读 |
+| 默认 mode | off（FAV 永久 off） |
+| 证据 | `.doc/test-reports/assets/2026-09-15-event-e0b-dispatcher-pg/` |
+
+默认旋钮：batch 50、lease 30s、heartbeat 10s、max attempts 12、backoff ceiling 15m。
+
+## 四、 预期结果 (Expected Results)
+
+| 用例 | 期望 |
+|------|------|
+| mode=off | RunOnce claim=0 |
+| FAV | ClaimLeases 失败；SetMode live 失败 |
+| 2/4 worker | 无双 lease；可见结果不加倍 |
+| crash before ack | 旧 fencing ack 失败；最终 n=1 |
+| quarantine | 不可 claim |
+| shadow | delivery 仍 pending/leased，不 ack |
+
+## 五、 实际测试结果与判定 (Actual Results & Evaluation)
+
+### 5.1 单元 / MemoryBus characterization
+
+- **测试输入**: `unset TEST_DATABASE_URL; go test -short ./internal/events`
+- **实际输出**: PASS（含禁用 `ClaimIDs`、claim SQL 仅一处 SKIP LOCKED）
+- **评价判定**: **PASS**
+- **性能与耗时**: 5ms
+
+### 5.2 隔离库 PG + race
+
+- **测试输入**: `TEST_DATABASE_URL` → `127.0.0.1:5434/agent_rel_e0b`；`go test -v ./internal/events -run TestE0B_`；`go test -race ./internal/events -run 'TestE0B_|TestTwoAndFour|TestCrash|TestClaim'`
+- **实际输出**（`assets/.../pg-test.txt`）: goose 至 40；`TestE0B_ModeOffDoesNotClaim_PG`、`FAVNeverClaimed`、`WorkersNoDoubleLeaseAndSerialAggregate`、`CrashAfterSideEffectBeforeAck`、`QuarantineNotClaimed`、`ShadowDoesNotAck` 全部 PASS。
+- **评价判定**: **PASS**
+- **性能与耗时**: 含 goose 0.35s；worker 用例 0.04s（8+2 事件，作为 attempt latency 基线 p50 < 50ms）。`-race` 1.16s PASS。
+
+### 5.3 默认 mode
+
+- **测试输入**: 查询 `event_consumer_modes`
+- **实际输出**: 各环境 notify/projection/workflow_advance/fav 均为 `off`（见 `consumer-modes.txt`）
+- **评价判定**: **PASS**
+
+### 5.4 全仓 short / arch-check
+
+- **测试输入**: `unset TEST_DATABASE_URL; go test -short -count=1 ./...`；`./scripts/arch-check.sh --mode block`
+- **实际输出**: 全包 PASS；arch-check block PASS
+- **评价判定**: **PASS**
+- **性能与耗时**: 约 12s
+
+## 六、 自动化测试套件汇总 (Test Suite Execution)
+
+| 命令 | 结果 |
+|------|------|
+| `go test -short ./...` | PASS |
+| `arch-check --mode block` | PASS |
+| 隔离库 `TestE0B_*` | PASS schema 40 |
+| `go test -race ./internal/events` | PASS |
+| 生产 mode | **off**；未对共享库 live claim |
+
+## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
+
+E0B **PASS**。独立 `cmd/events` + compose `events` 服务已接线；HTTP app 不启动 dispatcher。下一波 E1A 在集群 `notify=shadow` 后才允许覆盖门，再 `live`。不激活 FAV。
@@ -0,0 +1,16 @@
+ environment |     consumer     | mode | revision 
+-------------+------------------+------+----------
+ dev         | fav              | off  |        1
+ dev         | notify           | off  |       13
+ 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)
+
@@ -0,0 +1 @@
+go test -short ./... PASS; arch-check block PASS; go test -race ./internal/events TestE0B_ PASS
@@ -0,0 +1,15 @@
+=== RUN   TestE0B_ModeOffDoesNotClaim_PG
+2026/09/15 09:07:21 goose: no migrations to run. current version: 40
+--- PASS: TestE0B_ModeOffDoesNotClaim_PG (0.01s)
+=== RUN   TestE0B_FAVNeverClaimed_PG
+--- PASS: TestE0B_FAVNeverClaimed_PG (0.00s)
+=== RUN   TestE0B_WorkersNoDoubleLeaseAndSerialAggregate_PG
+--- PASS: TestE0B_WorkersNoDoubleLeaseAndSerialAggregate_PG (0.03s)
+=== RUN   TestE0B_CrashAfterSideEffectBeforeAck_PG
+--- PASS: TestE0B_CrashAfterSideEffectBeforeAck_PG (0.01s)
+=== RUN   TestE0B_QuarantineNotClaimed_PG
+--- PASS: TestE0B_QuarantineNotClaimed_PG (0.01s)
+=== RUN   TestE0B_ShadowDoesNotAck_PG
+--- PASS: TestE0B_ShadowDoesNotAck_PG (0.01s)
+PASS
+ok  	chat-agent-system/internal/events	0.075s
@@ -0,0 +1,127 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"chat-agent-system/internal/config"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/events"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func main() {
+	envPath := flag.String("env", ".env", "环境配置文件路径")
+	flag.Parse()
+
+	cfg, err := config.Load(*envPath)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "events: load config: %v\n", os.ErrInvalid)
+		os.Exit(1)
+	}
+	port := cfg.HTTPPort
+	if v := os.Getenv("EVENTS_HTTP_PORT"); v != "" {
+		if p, perr := parsePort(v); perr == nil {
+			port = p
+		}
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	pc, err := pgxpool.ParseConfig(cfg.ContextDSN())
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "events: dsn\n")
+		os.Exit(1)
+	}
+	pc.MaxConns = 8
+	pool, err := pgxpool.NewWithConfig(ctx, pc)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "events: pool\n")
+		os.Exit(1)
+	}
+	defer pool.Close()
+
+	if err := db.RunMigrations(ctx, pool, db.ResolveMigrationsDir()); err != nil {
+		fmt.Fprintf(os.Stderr, "events: migrate\n")
+		os.Exit(1)
+	}
+
+	owner := fmt.Sprintf("events-%d", os.Getpid())
+	disp := events.NewDispatcher(pool, cfg.Env, owner)
+	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})
+	})
+	mux.HandleFunc("GET /ready", func(w http.ResponseWriter, r *http.Request) {
+		if err := events.PingPool(r.Context(), pool); err != nil {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "DB"})
+			return
+		}
+		if err := events.SchemaReady(r.Context(), pool); err != nil {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "SCHEMA"})
+			return
+		}
+		reg := events.NewRegistry()
+		if err := reg.Register(events.ConsumerNotify); err != nil {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "code": "REGISTRY"})
+			return
+		}
+		writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
+	})
+	mux.HandleFunc("GET /live", func(w http.ResponseWriter, _ *http.Request) {
+		if !disp.Alive(30 * time.Second) {
+			writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "stale"})
+			return
+		}
+		writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
+	})
+	mux.HandleFunc("GET /metrics", func(w http.ResponseWriter, _ *http.Request) {
+		m := disp.Metrics
+		writeJSON(w, http.StatusOK, map[string]any{
+			"claimed": m.Claimed.Load(), "acked": m.Acked.Load(), "failed": m.Failed.Load(),
+			"fenced": m.Fenced.Load(), "dead": m.Dead.Load(), "quarantine": m.Quarantine.Load(),
+			"heartbeat": m.Heartbeat.Load(),
+		})
+	})
+
+	srv := &http.Server{Addr: fmt.Sprintf("%s:%d", cfg.HTTPHost, port), Handler: mux}
+	go func() {
+		consumers := []string{events.ConsumerNotify, events.ConsumerProjection, events.ConsumerWorkflowAdvance}
+		_ = disp.Serve(ctx, consumers)
+	}()
+	go func() {
+		<-ctx.Done()
+		shut, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		disp.StopClaim()
+		_ = srv.Shutdown(shut)
+	}()
+	fmt.Printf("events worker env=%s listen=%s:%d owner=%s mode=cluster-table\n", cfg.Env, cfg.HTTPHost, port, owner)
+	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+		os.Exit(1)
+	}
+}
+
+func parsePort(v string) (int, error) {
+	var p int
+	_, err := fmt.Sscanf(v, "%d", &p)
+	if err != nil || p <= 0 {
+		return 0, os.ErrInvalid
+	}
+	return p, nil
+}
+
+func writeJSON(w http.ResponseWriter, status int, body map[string]any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(body)
+}
@@ -18,6 +18,7 @@ COPY go.mod go.sum ./
 RUN go mod download
 COPY . .
 RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /server ./cmd/server
+RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /events ./cmd/events

 FROM debian:bookworm-slim

@@ -28,6 +29,7 @@ RUN set -eux; \

 WORKDIR /app
 COPY --from=go-builder /server /app/server
+COPY --from=go-builder /events /app/events
 COPY --from=web-builder /web/dist /app/web/dist
 COPY deploy/postgres/goose /app/deploy/postgres/goose
 # Official markdown must be on disk at /app/templates/docs and /app/configs.
@@ -0,0 +1,65 @@
+-- +goose Up
+-- E0B: atomic lease claim, immutable attempts, cluster consumer mode, checkpoints.
+-- Dispatcher defaults to off. Do not drop historical envelopes or deliveries.
+
+CREATE TABLE IF NOT EXISTS event_delivery_attempts (
+    id BIGSERIAL PRIMARY KEY,
+    event_id TEXT NOT NULL,
+    consumer TEXT NOT NULL,
+    attempt INT NOT NULL,
+    lease_owner TEXT NOT NULL,
+    fencing_token TEXT NOT NULL,
+    started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    finished_at TIMESTAMPTZ,
+    outcome TEXT NOT NULL DEFAULT 'leased',
+    error_code TEXT NOT NULL DEFAULT '',
+    correlation_id TEXT NOT NULL DEFAULT '',
+    CONSTRAINT uq_event_delivery_attempt UNIQUE (event_id, consumer, attempt, fencing_token),
+    CONSTRAINT chk_event_attempt_outcome CHECK (outcome IN ('leased','acked','pending','dead','quarantine','fenced'))
+);
+
+CREATE INDEX IF NOT EXISTS idx_event_delivery_attempts_consumer
+    ON event_delivery_attempts (consumer, started_at DESC);
+
+CREATE TABLE IF NOT EXISTS event_consumer_modes (
+    environment TEXT NOT NULL,
+    consumer TEXT NOT NULL,
+    mode TEXT NOT NULL,
+    revision BIGINT NOT NULL DEFAULT 1,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (environment, consumer),
+    CONSTRAINT chk_event_consumer_mode CHECK (mode IN ('off','shadow','live')),
+    CONSTRAINT chk_event_fav_mode_off CHECK (consumer <> 'fav' OR mode = 'off')
+);
+
+CREATE TABLE IF NOT EXISTS event_consumer_checkpoints (
+    consumer TEXT NOT NULL,
+    aggregate_type TEXT NOT NULL,
+    aggregate_id TEXT NOT NULL,
+    last_event_id TEXT NOT NULL,
+    last_aggregate_version BIGINT,
+    fencing_token TEXT NOT NULL DEFAULT '',
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (consumer, aggregate_type, aggregate_id)
+);
+
+CREATE TABLE IF NOT EXISTS event_reconciliations (
+    event_id TEXT NOT NULL,
+    consumer TEXT NOT NULL,
+    mode TEXT NOT NULL,
+    intended_digest CHAR(64) NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (event_id, consumer, mode)
+);
+
+INSERT INTO event_consumer_modes (environment, consumer, mode, revision)
+SELECT e.env, c.consumer, 'off', 1
+FROM (VALUES ('dev'), ('test'), ('main')) AS e(env)
+CROSS JOIN (VALUES ('notify'), ('projection'), ('workflow_advance'), ('fav')) AS c(consumer)
+ON CONFLICT (environment, consumer) DO NOTHING;
+
+-- +goose Down
+DROP TABLE IF EXISTS event_reconciliations;
+DROP TABLE IF EXISTS event_consumer_checkpoints;
+DROP TABLE IF EXISTS event_consumer_modes;
+DROP TABLE IF EXISTS event_delivery_attempts;
@@ -143,6 +143,35 @@ services:
      - dev-net
      - shared-net

+  events:
+    build:
+      context: .
+      dockerfile: deploy/app/Dockerfile
+    container_name: shuimu-agent-events-dev
+    entrypoint: ["/app/events"]
+    command: ["-env", "/dev/null"]
+    depends_on:
+      system-db:
+        condition: service_healthy
+    environment:
+      ENV: dev
+      HTTP_HOST: "0.0.0.0"
+      HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}"
+      EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8088}"
+      DATA_DIR: /app/data
+      APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-dev_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_dev}?sslmode=disable
+    ports:
+      - "127.0.0.1:${EVENTS_HTTP_PORT:-8088}:${EVENTS_HTTP_PORT:-8088}"
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8088}/health"]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 15s
+    restart: unless-stopped
+    networks:
+      - dev-net
+
 networks:
  dev-net:
    name: chat-agent-dev-net
@@ -142,6 +142,35 @@ services:
      - main-net
      - shared-net

+  events:
+    build:
+      context: .
+      dockerfile: deploy/app/Dockerfile
+    container_name: shuimu-agent-events-main
+    entrypoint: ["/app/events"]
+    command: ["-env", "/dev/null"]
+    depends_on:
+      system-db:
+        condition: service_healthy
+    environment:
+      ENV: main
+      HTTP_HOST: "0.0.0.0"
+      HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}"
+      EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8090}"
+      DATA_DIR: /app/data
+      APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:?SYSTEM_DB_PASSWORD is required for main/production}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_main}?sslmode=disable
+    ports:
+      - "127.0.0.1:${EVENTS_HTTP_PORT:-8090}:${EVENTS_HTTP_PORT:-8090}"
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8090}/health"]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 15s
+    restart: always
+    networks:
+      - main-net
+
 networks:
  main-net:
    name: chat-agent-main-net
@@ -141,6 +141,35 @@ services:
      - test-net
      - shared-net

+  events:
+    build:
+      context: .
+      dockerfile: deploy/app/Dockerfile
+    container_name: shuimu-agent-events-test
+    entrypoint: ["/app/events"]
+    command: ["-env", "/dev/null"]
+    depends_on:
+      system-db:
+        condition: service_healthy
+    environment:
+      ENV: test
+      HTTP_HOST: "0.0.0.0"
+      HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}"
+      EVENTS_HTTP_PORT: "${EVENTS_HTTP_PORT:-8089}"
+      DATA_DIR: /app/data
+      APP_DB_DSN: postgres://${SYSTEM_DB_USER:-app}:${SYSTEM_DB_PASSWORD:-test_secret_db_pass}@system-db:5432/${SYSTEM_DB_NAME:-agent_system_test}?sslmode=disable
+    ports:
+      - "127.0.0.1:${EVENTS_HTTP_PORT:-8089}:${EVENTS_HTTP_PORT:-8089}"
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:${EVENTS_HTTP_PORT:-8089}/health"]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 15s
+    restart: unless-stopped
+    networks:
+      - test-net
+
 networks:
  test-net:
    name: chat-agent-test-net
@@ -0,0 +1,298 @@
+package events
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+const (
+	DefaultBatch          = 50
+	DefaultLease          = 30 * time.Second
+	DefaultHeartbeat      = 10 * time.Second
+	DefaultBackoffCeiling = 15 * time.Minute
+	ModeOff               = "off"
+	ModeShadow            = "shadow"
+	ModeLive              = "live"
+)
+
+// Lease is the fenced claim a worker holds while the handler runs outside the claim transaction.
+type Lease struct {
+	EventID       string
+	Consumer      string
+	Attempt       int
+	Owner         string
+	Token         string
+	LeaseUntil    time.Time
+	AggregateType string
+	AggregateID   string
+	ProjectID     int64
+	Environment   string
+	EventName     string
+	Envelope      Envelope
+}
+
+const claimSQL = `
+WITH candidates AS (
+  SELECT d.event_id, d.consumer
+  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 (d.lease_until IS NULL OR d.lease_until < now())
+    AND d.next_attempt_at <= now()
+    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')
+    )
+    AND NOT EXISTS (
+      SELECT 1
+      FROM event_deliveries x
+      JOIN event_envelopes ex ON ex.event_id = x.event_id
+      WHERE x.consumer = d.consumer
+        AND x.status = 'leased'
+        AND x.lease_until > now()
+        AND ex.aggregate_type = e.aggregate_type
+        AND ex.aggregate_id = e.aggregate_id
+    )
+  ORDER BY d.next_attempt_at, d.event_id
+  FOR UPDATE OF d SKIP LOCKED
+  LIMIT $2
+)
+UPDATE event_deliveries d
+SET status = 'leased',
+    lease_owner = $3,
+    fencing_token = md5(random()::text || clock_timestamp()::text || d.event_id),
+    lease_until = now() + $4::interval,
+    attempt = d.attempt + 1
+FROM candidates c
+WHERE d.event_id = c.event_id AND d.consumer = c.consumer
+RETURNING d.event_id, d.consumer, d.attempt, d.lease_owner, d.fencing_token, d.lease_until
+`
+
+// ClaimIDs is the retired SELECT-only path. Production workers must use ClaimLeases.
+func ClaimIDs(ctx context.Context, tx pgx.Tx, consumer string, limit int) ([]string, error) {
+	return nil, fmt.Errorf("events: SELECT-only claim is disabled; use ClaimLeases")
+}
+
+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)
+	}
+	if limit <= 0 {
+		limit = DefaultBatch
+	}
+	if lease <= 0 {
+		lease = DefaultLease
+	}
+	rows, err := tx.Query(ctx, claimSQL, consumer, limit, owner, fmt.Sprintf("%d seconds", int(lease.Seconds())))
+	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, &l.Owner, &l.Token, &l.LeaseUntil); err != nil {
+			return nil, err
+		}
+		out = append(out, l)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	for i := range out {
+		if _, err := tx.Exec(ctx, `
+			INSERT INTO event_delivery_attempts (
+				event_id, consumer, attempt, lease_owner, fencing_token, outcome, correlation_id
+			) VALUES ($1,$2,$3,$4,$5,'leased',$1)
+		`, out[i].EventID, out[i].Consumer, out[i].Attempt, out[i].Owner, out[i].Token); err != nil {
+			return nil, err
+		}
+		if err := hydrateLease(ctx, tx, &out[i]); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+func hydrateLease(ctx context.Context, tx pgx.Tx, l *Lease) error {
+	var env Envelope
+	if err := tx.QueryRow(ctx, `
+		SELECT event_id, event_name, schema_version, environment, COALESCE(project_id,0),
+		       aggregate_type, aggregate_id, aggregate_version, actor_ref,
+		       correlation_id, causation_id, idempotency_key, payload_digest,
+		       producer, occurred_at, payload
+		FROM event_envelopes WHERE event_id=$1
+	`, l.EventID).Scan(&env.EventID, &env.EventName, &env.SchemaVersion, &env.Environment, &env.ProjectID,
+		&env.AggregateType, &env.AggregateID, &env.AggregateVersion, &env.ActorRef,
+		&env.CorrelationID, &env.CausationID, &env.IdempotencyKey, &env.PayloadDigest,
+		&env.Producer, &env.OccurredAt, &env.Payload); err != nil {
+		return err
+	}
+	l.Envelope = env
+	l.AggregateType = env.AggregateType
+	l.AggregateID = env.AggregateID
+	l.ProjectID = env.ProjectID
+	l.Environment = env.Environment
+	l.EventName = env.EventName
+	return nil
+}
+
+func AckLease(ctx context.Context, tx pgx.Tx, l Lease) error {
+	tag, err := tx.Exec(ctx, `
+		UPDATE event_deliveries
+		SET status='acked', lease_owner='', fencing_token='', inbox=TRUE
+		WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
+	`, l.EventID, l.Consumer, l.Owner, l.Token)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() != 1 {
+		return ErrStaleFencing
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE event_delivery_attempts
+		SET finished_at=now(), outcome='acked'
+		WHERE event_id=$1 AND consumer=$2 AND fencing_token=$3
+	`, l.EventID, l.Consumer, l.Token); err != nil {
+		return err
+	}
+	if l.AggregateType != "" && l.AggregateID != "" {
+		if _, err := tx.Exec(ctx, `
+			INSERT INTO event_consumer_checkpoints (
+				consumer, aggregate_type, aggregate_id, last_event_id, last_aggregate_version, fencing_token
+			) VALUES ($1,$2,$3,$4,$5,$6)
+			ON CONFLICT (consumer, aggregate_type, aggregate_id) DO UPDATE SET
+				last_event_id = EXCLUDED.last_event_id,
+				last_aggregate_version = EXCLUDED.last_aggregate_version,
+				fencing_token = EXCLUDED.fencing_token,
+				updated_at = now()
+		`, l.Consumer, l.AggregateType, l.AggregateID, l.EventID, l.Envelope.AggregateVersion, l.Token); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func FailLease(ctx context.Context, tx pgx.Tx, l Lease, code string, retryable bool) error {
+	nextStatus := "pending"
+	outcome := "pending"
+	if !retryable || l.Attempt >= MaxAttempts {
+		nextStatus = "dead"
+		outcome = "dead"
+		if code == "QUARANTINE" || code == "AUTHZ" || code == "SCOPE" {
+			nextStatus = "quarantine"
+			outcome = "quarantine"
+		}
+	}
+	backoff := backoffFor(l.Attempt)
+	tag, err := tx.Exec(ctx, `
+		UPDATE event_deliveries
+		SET status=$5, lease_owner='', fencing_token='',
+		    next_attempt_at = now() + $6::interval
+		WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
+	`, l.EventID, l.Consumer, l.Owner, l.Token, nextStatus, fmt.Sprintf("%d seconds", int(backoff.Seconds())))
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() != 1 {
+		return ErrStaleFencing
+	}
+	_, err = tx.Exec(ctx, `
+		UPDATE event_delivery_attempts
+		SET finished_at=now(), outcome=$4, error_code=$5
+		WHERE event_id=$1 AND consumer=$2 AND fencing_token=$3
+	`, l.EventID, l.Consumer, l.Token, outcome, code)
+	return err
+}
+
+func HeartbeatLease(ctx context.Context, tx pgx.Tx, l Lease, lease time.Duration) error {
+	if lease <= 0 {
+		lease = DefaultLease
+	}
+	tag, err := tx.Exec(ctx, `
+		UPDATE event_deliveries
+		SET lease_until = now() + $5::interval
+		WHERE event_id=$1 AND consumer=$2 AND lease_owner=$3 AND fencing_token=$4 AND status='leased'
+	`, l.EventID, l.Consumer, l.Owner, l.Token, fmt.Sprintf("%d seconds", int(lease.Seconds())))
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() != 1 {
+		return ErrStaleFencing
+	}
+	return nil
+}
+
+func LoadMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer string) (string, int64, error) {
+	var mode string
+	var rev int64
+	err := pool.QueryRow(ctx, `
+		SELECT mode, revision FROM event_consumer_modes WHERE environment=$1 AND consumer=$2
+	`, environment, consumer).Scan(&mode, &rev)
+	if err == pgx.ErrNoRows {
+		return ModeOff, 0, nil
+	}
+	return mode, rev, err
+}
+
+func SetMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error {
+	if consumer == ConsumerFAV && mode != ModeOff {
+		return fmt.Errorf("%w: fav", ErrDisabledConsumer)
+	}
+	_, err := pool.Exec(ctx, `
+		INSERT INTO event_consumer_modes (environment, consumer, mode, revision, updated_at)
+		VALUES ($1,$2,$3,1,now())
+		ON CONFLICT (environment, consumer) DO UPDATE SET
+			mode = EXCLUDED.mode,
+			revision = event_consumer_modes.revision + 1,
+			updated_at = now()
+	`, environment, consumer, mode)
+	return err
+}
+
+func backoffFor(attempt int) time.Duration {
+	shift := attempt
+	if shift < 0 {
+		shift = 0
+	}
+	if shift > 10 {
+		shift = 10
+	}
+	d := time.Second * time.Duration(1<<shift)
+	if d > DefaultBackoffCeiling {
+		return DefaultBackoffCeiling
+	}
+	return d
+}
+
+func SchemaReady(ctx context.Context, pool *pgxpool.Pool) error {
+	if pool == nil {
+		return ErrUnknownConsumer
+	}
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_consumer_modes`).Scan(&n); err != nil {
+		return err
+	}
+	if n == 0 {
+		return fmt.Errorf("events: consumer mode table empty")
+	}
+	return nil
+}
+
+func WriteReconciliation(ctx context.Context, tx pgx.Tx, eventID, consumer, mode, digest string) error {
+	_, err := tx.Exec(ctx, `
+		INSERT INTO event_reconciliations (event_id, consumer, mode, intended_digest)
+		VALUES ($1,$2,$3,$4)
+		ON CONFLICT (event_id, consumer, mode) DO UPDATE SET intended_digest = EXCLUDED.intended_digest
+	`, eventID, consumer, mode, digest)
+	return err
+}
@@ -0,0 +1,296 @@
+package events_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+	"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 e0bPool(t *testing.T) (*pgxpool.Pool, context.Context) {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("E0B dispatcher needs isolated postgres")
+	}
+	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 seedEnvelope(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) {
+	t.Helper()
+	unit := uow.NewPG(pool)
+	if err := unit.Within(ctx, func(scope uow.TxScope) error {
+		return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers)
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func sampleEnv(id, agg string, ver int64) events.Envelope {
+	body := []byte(`{"doc_id":"` + agg + `"}`)
+	return events.Envelope{
+		EventID: id, EventName: events.EventVersionCommitted, SchemaVersion: events.EnvelopeSchema,
+		Environment: "dev", ProjectID: 9, AggregateType: "document", AggregateID: agg,
+		AggregateVersion: ver, ActorRef: "user:1", IdempotencyKey: id, PayloadDigest: events.Digest(body),
+		Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body,
+	}
+}
+
+func TestE0B_ModeOffDoesNotClaim_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	seedEnvelope(t, ctx, pool, sampleEnv("off-"+suffix, "doc-off-"+suffix, 1), []string{events.ConsumerNotify})
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
+		t.Fatal(err)
+	}
+	d := events.NewDispatcher(pool, "dev", "w-off")
+	n, err := d.RunOnce(ctx, events.ConsumerNotify)
+	if err != nil || n != 0 {
+		t.Fatalf("claimed=%d err=%v", n, err)
+	}
+}
+
+func TestE0B_FAVNeverClaimed_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	if _, err := events.ClaimLeases(ctx, tx, events.ConsumerFAV, "w", 8, events.DefaultLease); err == nil {
+		t.Fatal("fav claim must fail")
+	}
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerFAV, events.ModeLive); err == nil {
+		t.Fatal("fav live must fail")
+	}
+}
+
+func TestE0B_WorkersNoDoubleLeaseAndSerialAggregate_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 8; i++ {
+		seedEnvelope(t, ctx, pool, sampleEnv(fmt.Sprintf("par-%s-%d", suffix, i), fmt.Sprintf("doc-par-%s-%d", suffix, i), 1), []string{events.ConsumerNotify})
+	}
+	seedEnvelope(t, ctx, pool, sampleEnv("ser-a-"+suffix, "doc-ser-"+suffix, 1), []string{events.ConsumerNotify})
+	seedEnvelope(t, ctx, pool, sampleEnv("ser-b-"+suffix, "doc-ser-"+suffix, 2), []string{events.ConsumerNotify})
+
+	visible := "e0b_visible_" + suffix
+	if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS `+visible+` (event_id TEXT PRIMARY KEY, n INT NOT NULL)`); err != nil {
+		t.Fatal(err)
+	}
+	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)
+		return err
+	}
+	runWorkers := func(n int) {
+		var wg sync.WaitGroup
+		for i := 0; i < n; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				d := events.NewDispatcher(pool, "dev", fmt.Sprintf("w%d", i))
+				d.Heartbeat = 50 * time.Millisecond
+				if err := d.Register(events.ConsumerNotify, handler); err != nil {
+					return
+				}
+				for k := 0; k < 12; k++ {
+					_, _ = d.RunOnce(ctx, events.ConsumerNotify)
+				}
+			}(i)
+		}
+		wg.Wait()
+	}
+	runWorkers(2)
+	runWorkers(4)
+
+	var leased int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_deliveries WHERE consumer=$1 AND status='leased' AND event_id LIKE $2`, events.ConsumerNotify, "%"+suffix+"%").Scan(&leased); err != nil {
+		t.Fatal(err)
+	}
+	if leased != 0 {
+		t.Fatalf("leftover leases=%d", leased)
+	}
+	var vis int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM `+visible).Scan(&vis); err != nil {
+		t.Fatal(err)
+	}
+	if vis < 8 {
+		t.Fatalf("visible=%d", vis)
+	}
+	var owners int
+	if err := pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM (
+			SELECT event_id, consumer, attempt, COUNT(DISTINCT lease_owner) FROM event_delivery_attempts
+			WHERE event_id LIKE $1 GROUP BY 1,2,3 HAVING COUNT(DISTINCT lease_owner) > 1
+		) x
+	`, "%"+suffix+"%").Scan(&owners); err != nil {
+		t.Fatal(err)
+	}
+	if owners != 0 {
+		t.Fatalf("double lease attempts=%d", owners)
+	}
+}
+
+func TestE0B_CrashAfterSideEffectBeforeAck_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	eventID := "crash-" + suffix
+	seedEnvelope(t, ctx, pool, sampleEnv(eventID, "doc-crash-"+suffix, 1), []string{events.ConsumerNotify})
+	if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS e0b_crash_visible (event_id TEXT PRIMARY KEY, n INT NOT NULL)`); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM e0b_crash_visible WHERE event_id=$1`, eventID)
+	})
+
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	leases, err := events.ClaimLeases(ctx, tx, events.ConsumerNotify, "w1", 1, time.Second)
+	if err != nil || len(leases) != 1 {
+		_ = tx.Rollback(ctx)
+		t.Fatalf("claim1: %v %#v", err, leases)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `INSERT INTO e0b_crash_visible (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, eventID); err != nil {
+		t.Fatal(err)
+	}
+	_, _ = pool.Exec(ctx, `UPDATE event_deliveries SET lease_until=now() - interval '2 seconds' WHERE event_id=$1 AND consumer=$2`, eventID, events.ConsumerNotify)
+
+	tx2, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := events.ClaimLeases(ctx, tx2, events.ConsumerNotify, "w2", 1, events.DefaultLease)
+	if err != nil || len(second) != 1 {
+		_ = tx2.Rollback(ctx)
+		t.Fatalf("claim2: %v %#v", err, second)
+	}
+	if err := tx2.Commit(ctx); err != nil {
+		t.Fatal(err)
+	}
+	ackOld, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := events.AckLease(ctx, ackOld, leases[0]); err != events.ErrStaleFencing {
+		_ = ackOld.Rollback(ctx)
+		t.Fatalf("old ack: %v", err)
+	}
+	_ = ackOld.Rollback(ctx)
+
+	if _, err := pool.Exec(ctx, `INSERT INTO e0b_crash_visible (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, eventID); err != nil {
+		t.Fatal(err)
+	}
+	ackNew, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := events.AckLease(ctx, ackNew, second[0]); err != nil {
+		_ = ackNew.Rollback(ctx)
+		t.Fatal(err)
+	}
+	if err := ackNew.Commit(ctx); err != nil {
+		t.Fatal(err)
+	}
+	var n int
+	if err := pool.QueryRow(ctx, `SELECT n FROM e0b_crash_visible WHERE event_id=$1`, eventID).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("visible n=%d", n)
+	}
+}
+
+func TestE0B_QuarantineNotClaimed_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	env := sampleEnv("q-"+suffix, "doc-q-"+suffix, 1)
+	seedEnvelope(t, ctx, pool, env, []string{events.ConsumerNotify})
+	unit := uow.NewPG(pool)
+	row := events.ClassifyEnvelope(events.Envelope{EventID: env.EventID, EventName: env.EventName}, events.ConsumerNotify)
+	row.Classification = events.ClassificationQuarantine
+	row.Reason = "fixture"
+	if err := unit.Within(ctx, func(scope uow.TxScope) error {
+		return events.PersistInventory(uow.WithScope(ctx, scope), row, "ops", "e0b")
+	}); err != nil {
+		t.Fatal(err)
+	}
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	got, err := events.ClaimLeases(ctx, tx, events.ConsumerNotify, "w", 8, events.DefaultLease)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, l := range got {
+		if l.EventID == env.EventID {
+			t.Fatalf("quarantined event claimed: %+v", l)
+		}
+	}
+}
+
+func TestE0B_ShadowDoesNotAck_PG(t *testing.T) {
+	pool, ctx := e0bPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	id := "sh-" + suffix
+	seedEnvelope(t, ctx, pool, sampleEnv(id, "doc-sh-"+suffix, 1), []string{events.ConsumerNotify})
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
+		t.Fatal(err)
+	}
+	d := events.NewDispatcher(pool, "dev", "w-shadow")
+	n, err := d.RunOnce(ctx, events.ConsumerNotify)
+	if err != nil || n < 1 {
+		t.Fatalf("shadow n=%d err=%v", n, err)
+	}
+	var status string
+	if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, id, events.ConsumerNotify).Scan(&status); err != nil {
+		t.Fatal(err)
+	}
+	if status != "pending" && status != "leased" {
+		t.Fatalf("shadow must not ack, status=%s", status)
+	}
+	if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
+		t.Fatal(err)
+	}
+}
@@ -11,7 +11,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 )

-// PGSink appends envelopes inside the current UnitOfWork. Claim uses ClaimIDs.
+// PGSink appends envelopes inside the current UnitOfWork.
 type PGSink struct{}

 func (PGSink) Append(ctx context.Context, env Envelope, consumers []string) error {
@@ -29,35 +29,6 @@ func (PGSink) Append(ctx context.Context, env Envelope, consumers []string) erro
 	return InsertEnvelope(ctx, tx, env, consumers)
 }

-const claimSQL = `
-SELECT event_id
-FROM event_deliveries
-WHERE consumer = $1
-  AND status IN ('pending', 'leased')
-  AND (lease_until IS NULL OR lease_until < now())
-ORDER BY next_attempt_at, event_id
-FOR UPDATE SKIP LOCKED
-LIMIT $2
-`
-
-// ClaimIDs runs the production claim statement. SKIP LOCKED is used only here.
-func ClaimIDs(ctx context.Context, tx pgx.Tx, consumer string, limit int) ([]string, error) {
-	rows, err := tx.Query(ctx, claimSQL, consumer, limit)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var ids []string
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			return nil, err
-		}
-		ids = append(ids, id)
-	}
-	return ids, rows.Err()
-}
-
 func InsertEnvelope(ctx context.Context, tx pgx.Tx, env Envelope, consumers []string) error {
 	if err := ValidateEnvelope(env); err != nil {
 		return err
@@ -2,12 +2,14 @@ package events

 import (
 	"os"
+	"path/filepath"
+	"runtime"
 	"strings"
 	"testing"
 )

 func TestClaimSQLUsesSkipLockedOnlyForClaim(t *testing.T) {
-	if !strings.Contains(claimSQL, "FOR UPDATE SKIP LOCKED") {
+	if !strings.Contains(claimSQL, "FOR UPDATE OF d SKIP LOCKED") {
 		t.Fatal("claim must use SKIP LOCKED")
 	}
 	if strings.Count(claimSQL, "SKIP LOCKED") != 1 {
@@ -16,15 +18,22 @@ func TestClaimSQLUsesSkipLockedOnlyForClaim(t *testing.T) {
 }

 func TestSkipLockedAbsentFromInsertSQL(t *testing.T) {
-	src, err := os.ReadFile("pg.go")
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		t.Fatal("caller")
+	}
+	src, err := os.ReadFile(filepath.Join(filepath.Dir(file), "pg.go"))
 	if err != nil {
 		t.Fatal(err)
 	}
-	withoutClaim := strings.ReplaceAll(string(src), claimSQL, "")
-	for _, line := range strings.Split(withoutClaim, "\n") {
-		trim := strings.TrimSpace(line)
-		if strings.Contains(trim, "SKIP LOCKED") && !strings.HasPrefix(trim, "//") {
-			t.Fatalf("SKIP LOCKED outside claim: %s", line)
+	if strings.Contains(string(src), "SKIP LOCKED") {
+		t.Fatal("SKIP LOCKED must not appear in pg.go")
 	}
 }
+
+func TestClaimIDsDisabled(t *testing.T) {
+	_, err := ClaimIDs(nil, nil, ConsumerNotify, 1)
+	if err == nil || !strings.Contains(err.Error(), "SELECT-only claim is disabled") {
+		t.Fatalf("%v", err)
+	}
 }
@@ -0,0 +1,308 @@
+package events
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"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
+
+type Dispatcher struct {
+	Pool        *pgxpool.Pool
+	Environment string
+	Owner       string
+	Batch       int
+	Lease       time.Duration
+	Heartbeat   time.Duration
+	Handlers    map[string]Handler
+	Metrics     *Metrics
+
+	mu        sync.Mutex
+	claiming  bool
+	inflight  map[string]struct{}
+	cancelFns []context.CancelFunc
+	lastBeat  atomic.Int64
+}
+
+type Metrics struct {
+	Claimed    atomic.Int64
+	Acked      atomic.Int64
+	Failed     atomic.Int64
+	Fenced     atomic.Int64
+	Dead       atomic.Int64
+	Quarantine atomic.Int64
+	Heartbeat  atomic.Int64
+}
+
+func NewDispatcher(pool *pgxpool.Pool, environment, owner string) *Dispatcher {
+	d := &Dispatcher{
+		Pool:        pool,
+		Environment: environment,
+		Owner:       owner,
+		Batch:       DefaultBatch,
+		Lease:       DefaultLease,
+		Heartbeat:   DefaultHeartbeat,
+		Handlers:    map[string]Handler{},
+		Metrics:     &Metrics{},
+		claiming:    true,
+		inflight:    map[string]struct{}{},
+	}
+	d.Touch()
+	return d
+}
+
+func (d *Dispatcher) Touch() { d.lastBeat.Store(time.Now().UnixNano()) }
+
+func (d *Dispatcher) Alive(maxAge time.Duration) bool {
+	if maxAge <= 0 {
+		maxAge = 30 * time.Second
+	}
+	return time.Since(time.Unix(0, d.lastBeat.Load())) < maxAge
+}
+
+func (d *Dispatcher) StopClaim() {
+	d.mu.Lock()
+	d.claiming = false
+	cancels := append([]context.CancelFunc(nil), d.cancelFns...)
+	d.mu.Unlock()
+	for _, c := range cancels {
+		c()
+	}
+}
+
+func (d *Dispatcher) Register(consumer string, h Handler) error {
+	if err := NewRegistry().Register(consumer); err != nil {
+		return err
+	}
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.Handlers[consumer] = h
+	return nil
+}
+
+func (d *Dispatcher) claimingEnabled() bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return d.claiming
+}
+
+func (d *Dispatcher) Serve(ctx context.Context, consumers []string) error {
+	tick := time.NewTicker(time.Second)
+	defer tick.Stop()
+	d.Touch()
+	for {
+		select {
+		case <-ctx.Done():
+			d.StopClaim()
+			return ctx.Err()
+		case <-tick.C:
+			d.Touch()
+			for _, c := range consumers {
+				_, _ = d.RunOnce(ctx, c)
+			}
+		}
+	}
+}
+
+func (d *Dispatcher) RunOnce(ctx context.Context, consumer string) (int, error) {
+	d.Touch()
+	if !d.claimingEnabled() {
+		return 0, nil
+	}
+	mode, _, err := LoadMode(ctx, d.Pool, d.Environment, consumer)
+	if err != nil {
+		return 0, err
+	}
+	if mode == ModeOff || consumer == ConsumerFAV {
+		return 0, nil
+	}
+	if mode == ModeShadow {
+		return d.shadowScan(ctx, consumer)
+	}
+	tx, err := d.Pool.Begin(ctx)
+	if err != nil {
+		return 0, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	leases, err := ClaimLeases(ctx, tx, consumer, d.Owner, d.Batch, d.Lease)
+	if err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return 0, err
+	}
+	d.Metrics.Claimed.Add(int64(len(leases)))
+	for _, lease := range leases {
+		d.handle(ctx, lease)
+	}
+	return len(leases), nil
+}
+
+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)
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var n int
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return n, err
+		}
+		tx, 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)
+			return n, err
+		}
+		if err := tx.Commit(ctx); err != nil {
+			return n, err
+		}
+		n++
+	}
+	d.Touch()
+	return n, rows.Err()
+}
+
+func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
+	key := lease.Consumer + ":" + lease.AggregateType + ":" + lease.AggregateID
+	d.mu.Lock()
+	d.inflight[key] = struct{}{}
+	hctx, cancel := context.WithCancel(ctx)
+	d.cancelFns = append(d.cancelFns, cancel)
+	d.mu.Unlock()
+	defer func() {
+		cancel()
+		d.mu.Lock()
+		delete(d.inflight, key)
+		d.mu.Unlock()
+	}()
+
+	done := make(chan struct{})
+	go func() {
+		ticker := time.NewTicker(d.Heartbeat)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				tx, err := d.Pool.Begin(hctx)
+				if err != nil {
+					cancel()
+					return
+				}
+				err = HeartbeatLease(hctx, tx, lease, d.Lease)
+				if err != nil {
+					_ = tx.Rollback(hctx)
+					d.Metrics.Fenced.Add(1)
+					cancel()
+					return
+				}
+				if err := tx.Commit(hctx); err != nil {
+					cancel()
+					return
+				}
+				d.Metrics.Heartbeat.Add(1)
+				d.Touch()
+			}
+		}
+	}()
+	defer close(done)
+
+	h := d.Handlers[lease.Consumer]
+	if h == nil {
+		_ = d.fail(ctx, lease, "NO_HANDLER", true)
+		return
+	}
+	if err := h(hctx, lease); err != nil {
+		if hctx.Err() != nil {
+			return
+		}
+		code, retry := classifyHandlerErr(err)
+		_ = d.fail(ctx, lease, code, retry)
+		return
+	}
+	if err := d.ack(ctx, lease); err != nil {
+		if errors.Is(err, ErrStaleFencing) {
+			d.Metrics.Fenced.Add(1)
+			return
+		}
+		_ = d.fail(ctx, lease, "TRANSIENT", true)
+	}
+}
+
+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
+	}
+	d.Metrics.Acked.Add(1)
+	d.Touch()
+	return nil
+}
+
+func (d *Dispatcher) fail(ctx context.Context, lease Lease, code string, retryable bool) error {
+	tx, err := d.Pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	if err := FailLease(ctx, tx, lease, code, retryable); err != nil {
+		if errors.Is(err, ErrStaleFencing) {
+			d.Metrics.Fenced.Add(1)
+			return err
+		}
+		return err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return err
+	}
+	d.Metrics.Failed.Add(1)
+	if code == "QUARANTINE" || code == "AUTHZ" || code == "SCOPE" {
+		d.Metrics.Quarantine.Add(1)
+	}
+	if !retryable || lease.Attempt >= MaxAttempts {
+		d.Metrics.Dead.Add(1)
+	}
+	d.Touch()
+	return nil
+}
+
+func classifyHandlerErr(err error) (string, bool) {
+	switch {
+	case errors.Is(err, ErrQuarantine), errors.Is(err, ErrAuthFailure):
+		return "QUARANTINE", false
+	case errors.Is(err, ErrIntegrityFailure):
+		return "INTEGRITY", false
+	case errors.Is(err, ErrInvalidScope):
+		return "SCOPE", false
+	case errors.Is(err, ErrStaleFencing):
+		return "FENCED", false
+	default:
+		return "TRANSIENT", true
+	}
+}
				`@@ -0,0 +1 @@`
				`go test -short ./... PASS; arch-check block PASS; go test -race ./internal/events TestE0B_ PASS`