fix(delivery): wake owed agents when a Runner attaches (RIG-4166)

go/internal/delivery/consumer.go

@@ -127,6 +127,9 @@ type DeliveryReads interface { //nolint:interfacebloat // one method per store r
 	// CountOwedMentions returns the total owed_mention row count — the startup
 	// visibility log (T2 observability).
 	CountOwedMentions(ctx context.Context) (int, error)
+	// OwedMentionAccounts lists every agent still owed a mention, for the wake
+	// retried once a Runner can serve it.
+	OwedMentionAccounts(ctx context.Context) ([]store.AccountID, error)
 	// MarkMentionsRouted stamps messageID's settle-edge mention pass complete
 	// (mentions_routed_at = now, unix ms) — the recovery scan's mark after it
 	// replays a message's mention pass, and the live path's mark (T3). Idempotent:
@@ -224,6 +227,9 @@ type Consumer struct {
 	// lostQueue buffers accounts whose session the Runner reported gone; the loop wakes
 	// each, and the woken session's start sweep redelivers what is owed.
 	lostQueue []store.AccountID
+	// owedRewake is set when a Runner command stream attaches: a wake that failed
+	// while none could serve it is retried for every agent still owed a mention.
+	owedRewake bool
 	// notify wakes the loop when settleQueue OR startQueue grows. Buffered(1)
 	// with a non-blocking send, so many edges between drains collapse to one
 	// wakeup and the hook never blocks.
@@ -352,6 +358,7 @@ func (c *Consumer) Run(ctx context.Context) error {
 			c.drainSettles(ctx)
 			c.drainStarts(ctx)
 			c.drainLost(ctx)
+			c.drainOwedRewake(ctx)
 		case event, ok := <-sub.Live:
 			if !ok {
 				if sub.Lagged() {

go/internal/delivery/helpers_test.go

@@ -493,6 +493,16 @@ func (f *fakeReads) CountOwedMentions(_ context.Context) (int, error) {
 	return n, nil
 }
 
+func (f *fakeReads) OwedMentionAccounts(_ context.Context) ([]store.AccountID, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := make([]store.AccountID, 0, len(f.owedMentions))
+	for agent := range f.owedMentions {
+		out = append(out, agent)
+	}
+	return out, nil
+}
+
 // SweepChannels returns the seeded D1 disjunct channel set for agent — the pin
 // sweep's channel enumeration, mirroring the store read.
 func (f *fakeReads) SweepChannels(_ context.Context, agent store.AccountID) ([]store.ChannelID, error) {

go/internal/delivery/settle.go

@@ -382,3 +382,41 @@ func (c *Consumer) drainLost(ctx context.Context) {
 		c.wake(ctx, account)
 	}
 }
+
+// OnRunnerReady is called when a Runner's command stream attaches. A mention made
+// while no Runner could serve it left an owed row and a failed wake; the loop now
+// wakes every owed agent. WakeAgent skips live agents and coalesces duplicates.
+func (c *Consumer) OnRunnerReady() {
+	c.mu.Lock()
+	c.owedRewake = true
+	c.mu.Unlock()
+	select {
+	case c.notify <- struct{}{}:
+	default:
+	}
+}
+
+// drainOwedRewake wakes every agent with an owed mention, once per OnRunnerReady.
+func (c *Consumer) drainOwedRewake(ctx context.Context) {
+	c.mu.Lock()
+	run := c.owedRewake
+	c.owedRewake = false
+	c.mu.Unlock()
+	if !run {
+		return
+	}
+	accounts, err := c.st.OwedMentionAccounts(ctx)
+	if err != nil {
+		c.log.WarnContext(ctx, "delivery: list owed mention accounts on runner ready", "error", err)
+		return
+	}
+	// Off the loop: each wake is a Runner round trip, and a restart can owe many.
+	go func() {
+		for _, account := range accounts {
+			if ctx.Err() != nil {
+				return
+			}
+			c.wake(ctx, account)
+		}
+	}()
+}

go/internal/runnerhub/hub.go

@@ -130,6 +130,12 @@ type SessionLostSink interface {
 	OnSessionLost(sessionID string, account store.AccountID)
 }
 
+// RunnerReadySink is told each time a Runner command stream attaches, so work that
+// failed while no Runner could serve it is retried. Must return promptly.
+type RunnerReadySink interface {
+	OnRunnerReady()
+}
+
 // SessionEndSink archives the transcript tail of a session the hub saw end without
 // Stop (a lost container or a re-enroll reap). ctx is scoped to the session tenant.
 // It may block on object-store I/O, so the hub calls it off its locks and loops.
@@ -360,6 +366,7 @@ type Hub struct {
 	// lost is notified after dropLostSession releases a dead session. Read under mu.
 	lost  SessionLostSink
 	ended SessionEndSink
+	ready RunnerReadySink
 	// presence is the RIG-1569 T8 presence projection's sink, notified at
 	// deliverSession (lifecycle transition) and promoteSession (reconciliation). Nil
 	// until SetPresenceSink; read under mu. Nil-safe (today's behavior).
@@ -526,6 +533,13 @@ func (h *Hub) SetSessionReapSink(reap SessionReapSink) {
 	h.reap = reap
 }
 
+// SetRunnerReadySink wires the delivery consumer to retry owed wakes on attach.
+func (h *Hub) SetRunnerReadySink(ready RunnerReadySink) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.ready = ready
+}
+
 // SetSessionEndSink wires the transcript archive for sessions that end without Stop.
 func (h *Hub) SetSessionEndSink(end SessionEndSink) {
 	h.mu.Lock()
@@ -742,7 +756,11 @@ func (h *Hub) FrameDiagnostics() FrameDiagnostics {
 func (h *Hub) fireRunnerReady() {
 	h.mu.Lock()
 	hook := h.runnerReadyHook
+	ready := h.ready
 	h.mu.Unlock()
+	if ready != nil {
+		ready.OnRunnerReady()
+	}
 	if hook == nil {
 		return
 	}

go/internal/store/delivery_cursors.go

@@ -401,3 +401,16 @@ func (s *Store) InSweepSet(ctx context.Context, agent AccountID, channel Channel
 	}
 	return in, nil
 }
+
+// OwedMentionAccounts lists every agent with at least one owed mention.
+func (s *Store) OwedMentionAccounts(ctx context.Context) ([]AccountID, error) {
+	rows, err := s.q.OwedMentionAccounts(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("store: list owed mention accounts: %w", err)
+	}
+	out := make([]AccountID, len(rows))
+	for i, r := range rows {
+		out[i] = AccountID(r)
+	}
+	return out, nil
+}

go/internal/store/queries/delivery_cursors.sql

@@ -105,3 +105,8 @@ SELECT EXISTS(
 	WHERE cm.account_id = $1
 	  AND cm.channel_id = $2
 	  AND (cm.subscribed OR cm.channel_id = aa.home_channel_id OR ch.mandatory_subscription));
+
+-- The accounts still owed a mention: a wake that failed before any Runner could
+-- serve it is retried for these once one attaches.
+-- name: OwedMentionAccounts :many
+SELECT DISTINCT agent_account_id FROM owed_mentions ORDER BY agent_account_id;

go/server/offline_mention_e2e_pgtest_test.go

@@ -51,6 +51,15 @@ type mentionE2EWire struct {
 // also creates one shared OPEN channel the admin authors into, admin a member so
 // PostMessage's D9 membership gate admits the post.
 func newMentionE2EWire(t *testing.T) *mentionE2EWire {
+	t.Helper()
+	w := newMentionE2EWireNoRunner(t)
+	w.runner = attachFakeRunner(t, w.store, w.hub, false)
+	return w
+}
+
+// newMentionE2EWireNoRunner is newMentionE2EWire before any Runner attaches: the
+// gap after a Server restart, before the Runner re-enrolls.
+func newMentionE2EWireNoRunner(t *testing.T) *mentionE2EWire {
 	t.Helper()
 	ctx := context.Background() // the test root context (rule://go-thread-context _test.go exemption)
 	dsn := pgtest.RequireDSN(t)
@@ -86,7 +95,6 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire {
 	brd := board.NewProjection(bus)
 	tail := newSessionTail()
 	hub := newRunnerHub(st, brd, tail, commsSvc, slog.New(slog.DiscardHandler))
-	runner := attachFakeRunner(t, st, hub, false)
 
 	// The production delivery wire (sinks.go:142-155), assembled inline with the
 	// REAL resume-based waker (newLifecycleService), not a fake.
@@ -95,6 +103,7 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire {
 	hub.SetSettleSink(c)
 	hub.SetSessionStartSink(c)
 	hub.SetSessionLostSink(c)
+	hub.SetRunnerReadySink(c)
 	hub.SetDeliveryStore(st)
 	runCtx, cancel := context.WithCancel(ctx)
 	done := make(chan struct{})
@@ -123,7 +132,6 @@ func newMentionE2EWire(t *testing.T) *mentionE2EWire {
 		hub:      hub,
 		comms:    commsSvc,
 		consumer: c,
-		runner:   runner,
 		dsn:      dsn,
 		adminID:  admin.ID,
 		channel:  ch.ID,
@@ -462,3 +470,49 @@ func TestDeliverToLostSessionWakesAndRedelivers(t *testing.T) {
 		t.Fatalf("SessionForAccount after wake = (%q, %v), want (%q, true)", sess, ok, newSess)
 	}
 }
+
+// TestMentionBeforeRunnerEnrollsWakesOnAttach: a mention made after a restart but
+// before the Runner re-enrolls fails its wake (no Runner to serve it). Once a Runner
+// command stream attaches, the owed agent is woken.
+func TestMentionBeforeRunnerEnrollsWakesOnAttach(t *testing.T) {
+	w := newMentionE2EWireNoRunner(t)
+	agent := w.seedAgentMember(t, "gapwake", false)
+	w.post(t, "@gapwake are you there")
+	// Routing settles on the consumer loop, so the owed row lands asynchronously.
+	deadline := timeAfter()
+	for {
+		owed, err := w.store.OwedMentions(w.ctx, agent.ID)
+		if err != nil {
+			t.Fatalf("OwedMentions: %v", err)
+		}
+		if len(owed) == 1 {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatalf("owed mention channels = %d, want 1 before any Runner attaches", len(owed))
+		default:
+		}
+		runtime.Gosched()
+	}
+
+	// The wake fires during attach, before the harness clears its probe bookkeeping,
+	// so assert the durable session the fresh start records, not the command log.
+	// Read under the system role, as the wake path does: a wake from the delivery
+	// loop records its session without a tenant.
+	w.runner = attachFakeRunner(t, w.store, w.hub, false)
+	deadline = timeAfter()
+	for {
+		if _, ok, err := w.store.LatestSessionForAccount(store.WithSystemRole(w.ctx), agent.ID); err != nil {
+			t.Fatalf("LatestSessionForAccount: %v", err)
+		} else if ok {
+			return
+		}
+		select {
+		case <-deadline:
+			t.Fatal("no session started for the owed agent after the Runner attached")
+		default:
+		}
+		runtime.Gosched()
+	}
+}

go/server/sinks.go

@@ -125,6 +125,7 @@ func startDeliveryConsumer(gctx context.Context, g *errgroup.Group, commsBus *ev
 	hub.SetSessionStartSink(c)
 	hub.SetSessionReapSink(c)
 	hub.SetSessionLostSink(c)
+	hub.SetRunnerReadySink(c)
 	hub.SetDeliveryStore(st)
 	g.Go(func() error { return c.Run(gctx) })
 }