fix(serve): stop trigger-C stuck false-fires + boot-burst hub drop

Two v0.1 post-ship nits that the user's browser session surfaced:

attention engine: trigger C (stuck) would instantly fire on daemon
restart because the hub ring's replay delivered old tool_call events
with their original (pre-restart) timestamps. The idle check used the
raw lastCall, so a session with a tool_call 10 min before the restart
already tripped the 5 min threshold at t=0.

Fix: Engine.bootTime is stamped at NewEngine, and trigger C uses
max(st.lastCall, bootTime) as the idle reference. A session truly
idle for IdleMinutes post-boot still fires; replay-only signals no
longer pre-age it. Regression test added.

hub buffer: subChanBuffer raised from 128 to 1024. The boot-burst
WARN ("hub subscriber dropping events filter=\"\" dropped_total=1")
came from the quota ingester's catch-up scan + tailer initial replay
fanning out tens of events faster than a newly-subscribed internal
consumer could drain its 128-slot channel. 1024 absorbs the burst
with negligible memory cost (~200 KB per subscriber) and the
SlowConsumerDrop test is now anchored to subChanBuffer so future
tweaks stay covered.

Verification: go test ./... 493/24, race-clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: aksOps

internal/serve/attention/engine.go

@@ -92,6 +92,13 @@ type Engine struct {
 	thr      Thresholds
 	now      func() time.Time
 
+	// bootTime is the clock at NewEngine. Trigger C (stuck) uses
+	// max(st.lastCall, bootTime) as its reference so sessions whose
+	// only tool_call history came from a hub-ring replay don't
+	// instantly trip stuck with an ancient timestamp. A session that
+	// truly goes idle after boot will still fire after IdleMinutes.
+	bootTime time.Time
+
 	mu       sync.Mutex
 	sessions_state map[string]*sessionState
 }
@@ -116,6 +123,7 @@ func NewEngine(hub *events.Hub, quota QuotaSource, sessions SessionSource, thr T
 		sessions:       sessions,
 		thr:            thr,
 		now:            clock,
+		bootTime:       clock(),
 		sessions_state: make(map[string]*sessionState),
 	}
 }
@@ -464,9 +472,16 @@ func (e *Engine) pickState(
 		}
 	}
 
-	// C · stuck — idle > threshold while tmux alive.
+	// C · stuck — idle > threshold while tmux alive. Reference clamps
+	// to bootTime so an ancient tool_call recovered from the hub ring
+	// on daemon restart can't pre-age the signal; the session has to
+	// go idle for IdleMinutes *since we started watching*.
 	if e.thr.IdleMinutes > 0 && !st.lastCall.IsZero() {
-		if now.Sub(st.lastCall) > time.Duration(e.thr.IdleMinutes)*time.Minute {
+		ref := st.lastCall
+		if ref.Before(e.bootTime) {
+			ref = e.bootTime
+		}
+		if now.Sub(ref) > time.Duration(e.thr.IdleMinutes)*time.Minute {
 			return Snapshot{
 				State:   StateStuck,
 				Since:   e.sinceOr(st, StateStuck, now),

internal/serve/attention/engine_test.go

@@ -240,6 +240,44 @@ func TestTriggerC_Stuck(t *testing.T) {
 	}
 }
 
+// Regression: daemon restart replays old tool_calls from the hub ring
+// with timestamps > IdleMinutes ago. The previous implementation used
+// those timestamps directly and pre-tripped stuck on boot. Reference
+// now clamps to bootTime, so stuck can only fire on idle *since boot*.
+func TestTriggerC_Stuck_BootReplayDoesNotInstantlyFire(t *testing.T) {
+	now := time.Unix(1_700_000_000, 0).UTC()
+	hub := events.NewHub(50)
+	q := newFakeQuota()
+	s := newFakeSessions("alpha")
+	thr := Defaults()
+	thr.IdleMinutes = 5
+	eng := newEngineAt(&now, hub, q, s, thr)
+
+	// Simulate ring replay: a tool_call from 10 minutes before boot.
+	ancient := now.Add(-10 * time.Minute)
+	eng.handleEvent(toolCallEv(t, "alpha", false, ancient))
+
+	// Immediately at boot: must NOT fire stuck.
+	eng.evaluateAll()
+	if snap, ok := eng.Snapshot("alpha"); ok {
+		t.Fatalf("stuck fired on replay at boot; snap=%+v", snap)
+	}
+
+	// 4 min after boot: still under the 5 min threshold since boot.
+	now = now.Add(4 * time.Minute)
+	eng.evaluateAll()
+	if snap, ok := eng.Snapshot("alpha"); ok {
+		t.Fatalf("stuck fired at 4 min since boot; snap=%+v", snap)
+	}
+
+	// 6 min since boot with no fresh tool_call → stuck fires now.
+	now = now.Add(2 * time.Minute)
+	eng.evaluateAll()
+	if snap, ok := eng.Snapshot("alpha"); !ok || snap.State != StateStuck {
+		t.Fatalf("want stuck at 6 min since boot; got %+v ok=%v", snap, ok)
+	}
+}
+
 func TestTriggerD_TmuxDead(t *testing.T) {
 	now := time.Unix(1_700_000_000, 0).UTC()
 	hub := events.NewHub(50)

internal/serve/events/hub.go

@@ -11,7 +11,13 @@ import (
 
 const (
 	defaultRingSize = 500
-	subChanBuffer   = 128
+	// subChanBuffer sized to absorb boot bursts without dropping: the
+	// quota ingester's catch-up scan + the tailer's initial JSONL
+	// replay can fan out dozens of events in a few ms before an
+	// internal subscriber's consumer loop gets scheduled. 1024 covers
+	// that comfortably at ~200 KB per subscriber worst-case, and still
+	// gives ordinary browser SSE tabs plenty of headroom.
+	subChanBuffer   = 1024
 	dropLogInterval = time.Minute
 )
 

internal/serve/events/hub_test.go

@@ -134,7 +134,9 @@ func TestHub_SlowConsumerDrop(t *testing.T) {
 	slow, _ := h.Subscribe("", "")
 	defer slow.Close()
 
-	const n = 1000
+	// Anchor to the buffer size so this keeps testing "overflow"
+	// behaviour if subChanBuffer changes in the future.
+	n := subChanBuffer + 500
 	done := make(chan struct{})
 	go func() {
 		defer close(done)
@@ -149,7 +151,7 @@ func TestHub_SlowConsumerDrop(t *testing.T) {
 		t.Fatalf("publisher blocked on slow consumer")
 	}
 
-	// Drain what slow can hold (subChanBuffer=128) and verify the rest were dropped.
+	// Drain what slow can hold (subChanBuffer) and verify the rest were dropped.
 	delivered := 0
 drain:
 	for {
@@ -166,7 +168,7 @@ drain:
 	if slow.Dropped() == 0 {
 		t.Fatal("expected non-zero dropped counter")
 	}
-	if uint64(delivered)+slow.Dropped() != n {
+	if uint64(delivered)+slow.Dropped() != uint64(n) {
 		t.Fatalf("delivered(%d) + dropped(%d) != %d", delivered, slow.Dropped(), n)
 	}
 	t.Logf("slow consumer: published=%d delivered=%d dropped=%d", n, delivered, slow.Dropped())