feat(events): skip in-process writers when a consumer is live

Keep OnOutbox and approval hooks registered while dispatcher mode is
off, but no-op them as soon as the matching durable consumer owns the
visible write so a later switch cannot dual-write.

Co-authored-by: Cursor <cursoragent@cursor.com>

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

@@ -161,6 +161,7 @@ imports:
     - docgov
     - docquery
     - docrevision
+    - events
     - freedoc
     - intent
     - notification

internal/app/app.go

@@ -153,6 +153,9 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
 		Stageflow:  stages,
 	}
 	rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()}
+	rt.DocNotify.SkipWrite = func(ctx context.Context) bool {
+		return rt.WriterLive(ctx, events.ConsumerNotify)
+	}
 	envName := ""
 	if cfg != nil {
 		envName = cfg.Env
@@ -193,7 +196,22 @@ func (r *Runtime) bindWorkflowApprovalHook() {
 	r.subscribers++
 }
 
+// WriterLive reports whether the durable consumer already owns the visible write.
+func (r *Runtime) WriterLive(ctx context.Context, consumer string) bool {
+	if r == nil || r.Biz == nil {
+		return false
+	}
+	env := ""
+	if r.Cfg != nil {
+		env = r.Cfg.Env
+	}
+	return events.WriterLive(ctx, r.Biz.Pool(), env, consumer)
+}
+
 func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) {
+	if r.WriterLive(ctx, events.ConsumerWorkflowAdvance) {
+		return
+	}
 	if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil {
 		return
 	}

internal/app/docnotify/subscriber.go

@@ -19,6 +19,9 @@ type DocumentLookup interface {
 type Subscriber struct {
 	Notify notificationEmitter
 	Docs   DocumentLookup
+	// SkipWrite is true when the durable notify consumer is live.
+	// Bind stays registered while mode=off; the closure must not dual-write after switch.
+	SkipWrite func(ctx context.Context) bool
 }
 
 type notificationEmitter interface {
@@ -37,7 +40,13 @@ func Bind(rev *docrevision.Service, sub *Subscriber) bool {
 
 // Handle maps one document outbox event onto a permission-safe inbox notification.
 func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error {
-	if s == nil || s.Notify == nil || !s.Notify.Available() {
+	if s == nil {
+		return nil
+	}
+	if s.SkipWrite != nil && s.SkipWrite(ctx) {
+		return nil
+	}
+	if s.Notify == nil || !s.Notify.Available() {
 		return nil
 	}
 	docID, _ := payload["doc_id"].(string)

internal/events/dispatcher.go

@@ -302,6 +302,17 @@ func LoadMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer str
 	return mode, rev, err
 }
 
+// WriterLive is true only when the durable consumer already owns the visible write.
+// Closures must no-op in that window; off/shadow keep the in-process hook as the unique writer.
+// A missing pool or mode row is treated as not-live so HTTP notify/advance keep working.
+func WriterLive(ctx context.Context, pool *pgxpool.Pool, environment, consumer string) bool {
+	if pool == nil {
+		return false
+	}
+	mode, _, err := LoadMode(ctx, pool, environment, consumer)
+	return err == nil && mode == ModeLive
+}
+
 func SetMode(ctx context.Context, pool *pgxpool.Pool, environment, consumer, mode string) error {
 	if consumer == ConsumerFAV && mode != ModeOff {
 		return fmt.Errorf("%w: fav", ErrDisabledConsumer)

internal/server/workflow_approval_hook.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/events"
 	"chat-agent-system/internal/notification"
 )
 
@@ -15,6 +16,9 @@ func (s *Server) registerWorkflowApprovalHook() {
 }
 
 func (s *Server) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) {
+	if s.rt != nil && s.rt.WriterLive(ctx, events.ConsumerWorkflowAdvance) {
+		return
+	}
 	if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || s.instSvc == nil {
 		return
 	}