diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml
index c8f70777..5efedf29 100644
--- a/.doc/system-design/module-boundaries.yaml
+++ b/.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:
diff --git a/cmd/events/main.go b/cmd/events/main.go
index 9dba1508..21e9b9e3 100644
--- a/cmd/events/main.go
+++ b/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})
diff --git a/internal/app/docnotify/consumer.go b/internal/app/docnotify/consumer.go
new file mode 100644
index 00000000..ebaebb39
--- /dev/null
+++ b/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 "", "", ""
+	}
+}
diff --git a/internal/events/dispatcher.go b/internal/events/dispatcher.go
index 00072055..c4a71dbf 100644
--- a/internal/events/dispatcher.go
+++ b/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
+}
diff --git a/internal/events/dispatcher_pg_test.go b/internal/events/dispatcher_pg_test.go
index 492b47e1..75c923e3 100644
--- a/internal/events/dispatcher_pg_test.go
+++ b/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) {
diff --git a/internal/events/worker.go b/internal/events/worker.go
index cd54237e..5b5be912 100644
--- a/internal/events/worker.go
+++ b/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 {
diff --git a/internal/notification/document_events.go b/internal/notification/document_events.go
index fb994606..d22cffe7 100644
--- a/internal/notification/document_events.go
+++ b/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() {