fix(review round 1): daemon PAT-header auth, job pruning, session-split batches, streaming tars, seed cleanup, ping/pong, byte caps, branch switch, slug --, ctx cancel, Init:true, pagination, latest/before events; web newest-window history + load-older, offline busy gate, staleness guards, memo store, auth probe, scroll key, focus-visible, draft restore, poll dedup; 106+181 tests, coverage 96.2%/95.1%+

daemon/hub.go

@@ -44,8 +44,11 @@ const (
 	agentSendQueue    int           = 32
 	webSendQueue      int           = 128
 	webFlushInterval  time.Duration = 40 * time.Millisecond
-	webMaxPending     int           = 4096 // drop slow clients beyond this backlog
+	webMaxPending     int           = 4096     // drop slow clients beyond this backlog
+	webMaxPendingByte int           = 64 << 20 // ...or beyond this many queued bytes
 	promptSendTimeout time.Duration = 3 * time.Second
+	pongWait          time.Duration = 60 * time.Second
+	pingPeriod        time.Duration = 30 * time.Second
 	writeWait         time.Duration = 5 * time.Second
 	daemonVersion     int           = 1 // envelope "v"
 )
@@ -109,11 +112,12 @@ type webClient struct {
 	hub  *Hub
 	conn *websocket.Conn
 
-	mu      sync.Mutex
-	sub     string // subscribed sessionId, "" when none
-	control [][]byte
-	events  []pendingEvent
-	dropped bool
+	mu           sync.Mutex
+	sub          string // subscribed sessionId, "" when none
+	control      [][]byte
+	events       []pendingEvent
+	pendingBytes int
+	dropped      bool
 
 	closeOnce sync.Once
 	done      chan struct{}
@@ -126,18 +130,25 @@ func (c *webClient) drop() {
 	})
 }
 
+// overflows reports whether one more item of size n would exceed the caps.
+func (c *webClient) overflows(n int) bool {
+	return len(c.control)+len(c.events) >= webMaxPending ||
+		c.pendingBytes+n > webMaxPendingByte
+}
+
 func (c *webClient) deliverControl(b []byte) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if c.dropped {
 		return
 	}
-	if len(c.control)+len(c.events) >= webMaxPending {
+	if c.overflows(len(b)) {
 		c.dropped = true
 		go c.drop()
 		return
 	}
 	c.control = append(c.control, b)
+	c.pendingBytes += len(b)
 }
 
 func (c *webClient) deliverEvent(e pendingEvent) {
@@ -146,30 +157,68 @@ func (c *webClient) deliverEvent(e pendingEvent) {
 	if c.dropped {
 		return
 	}
-	if len(c.control)+len(c.events) >= webMaxPending {
+	if c.overflows(len(e.raw)) {
 		c.dropped = true
 		go c.drop()
 		return
 	}
 	c.events = append(c.events, e)
+	c.pendingBytes += len(e.raw)
+}
+
+// writeEventsFrame flushes one batched events frame for a single session.
+func (c *webClient) writeEventsFrame(batch []pendingEvent) error {
+	items := make([]json.RawMessage, 0, len(batch))
+	for _, e := range batch {
+		items = append(items, e.raw)
+	}
+	payload, err := json.Marshal(map[string]any{
+		"type":      frameEvents,
+		"sessionId": batch[0].sessionID,
+		"after":     batch[0].seq - 1,
+		"events":    items,
+	})
+	if err != nil {
+		return nil // unmarshalable raw JSON cannot happen; drop the batch
+	}
+	_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+	return c.conn.WriteMessage(websocket.TextMessage, payload)
 }
 
 // writePump flushes queued frames every webFlushInterval, batching events of
-// the subscribed session into single frames. Never blocks the hub.
+// the subscribed session into single frames (one frame per contiguous
+// sessionID run — a mid-queue resubscribe must never mix sessions).
+// Never blocks the hub.
 func (c *webClient) writePump() {
 	ticker := time.NewTicker(webFlushInterval)
+	pings := time.NewTicker(pingPeriod)
 	defer ticker.Stop()
+	defer pings.Stop()
 	for {
 		select {
 		case <-c.done:
 			return
 		case <-ticker.C:
+		case <-pings.C:
+			if !c.sendPing() {
+				c.hub.dropWeb(c)
+				return
+			}
+			continue
 		}
 		c.mu.Lock()
 		control := c.control
 		c.control = nil
 		events := c.events
 		c.events = nil
+		drained := 0
+		for _, b := range control {
+			drained += len(b)
+		}
+		for _, e := range events {
+			drained += len(e.raw)
+		}
+		c.pendingBytes -= drained
 		c.mu.Unlock()
 		for _, b := range control {
 			_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
@@ -178,35 +227,34 @@ func (c *webClient) writePump() {
 				return
 			}
 		}
-		if len(events) == 0 {
-			continue
-		}
-		items := make([]json.RawMessage, 0, len(events))
-		after := events[0].seq - 1
-		for _, e := range events {
-			items = append(items, e.raw)
-		}
-		payload, err := json.Marshal(map[string]any{
-			"type":      frameEvents,
-			"sessionId": events[0].sessionID,
-			"after":     after,
-			"events":    items,
-		})
-		if err != nil {
-			continue
-		}
-		_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
-		if err := c.conn.WriteMessage(websocket.TextMessage, payload); err != nil {
-			c.hub.dropWeb(c)
-			return
+		for i := 0; i < len(events); {
+			j := i
+			for j < len(events) && events[j].sessionID == events[i].sessionID {
+				j++
+			}
+			if err := c.writeEventsFrame(events[i:j]); err != nil {
+				c.hub.dropWeb(c)
+				return
+			}
+			i = j
 		}
 	}
 }
 
+// sendPing writes one ping frame honoring the write deadline.
+func (c *webClient) sendPing() bool {
+	_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+	return c.conn.WriteMessage(websocket.PingMessage, nil) == nil
+}
+
 // readPump consumes subscribe/unsubscribe frames; malformed input never kills the server.
 func (c *webClient) readPump() {
 	defer c.hub.dropWeb(c)
 	c.conn.SetReadLimit(maxFrameSize)
+	_ = c.conn.SetReadDeadline(time.Now().Add(pongWait))
+	c.conn.SetPongHandler(func(string) error {
+		return c.conn.SetReadDeadline(time.Now().Add(pongWait))
+	})
 	for {
 		_, data, err := c.conn.ReadMessage()
 		if err != nil {
@@ -252,8 +300,11 @@ func (a *agentConn) drop() {
 	})
 }
 
-// writePump serializes daemon→plugin frames (prompt/abort/welcome).
+// writePump serializes daemon→plugin frames (prompt/abort/welcome) and
+// keeps the conn alive with periodic pings.
 func (a *agentConn) writePump() {
+	pings := time.NewTicker(pingPeriod)
+	defer pings.Stop()
 	for {
 		select {
 		case <-a.done:
@@ -264,10 +315,21 @@ func (a *agentConn) writePump() {
 				a.drop()
 				return
 			}
+		case <-pings.C:
+			if !a.sendPing() {
+				a.drop()
+				return
+			}
 		}
 	}
 }
 
+// sendPing writes one ping frame honoring the write deadline.
+func (a *agentConn) sendPing() bool {
+	_ = a.conn.SetWriteDeadline(time.Now().Add(writeWait))
+	return a.conn.WriteMessage(websocket.PingMessage, nil) == nil
+}
+
 // Hub tracks live agent conns and web subscribers; it is the only component
 // that mutates online state.
 type Hub struct {
@@ -418,6 +480,10 @@ func (h *Hub) ServeAgentWS(w http.ResponseWriter, r *http.Request) {
 func (h *Hub) readAgentLoop(ac *agentConn) {
 	defer ac.drop()
 	ac.conn.SetReadLimit(maxFrameSize)
+	_ = ac.conn.SetReadDeadline(time.Now().Add(pongWait))
+	ac.conn.SetPongHandler(func(string) error {
+		return ac.conn.SetReadDeadline(time.Now().Add(pongWait))
+	})
 	registered := false
 	defer func() {
 		if registered {
@@ -543,11 +609,15 @@ func (h *Hub) handleEvent(f frame) {
 
 func (h *Hub) unregister(ac *agentConn) {
 	h.mu.Lock()
+	removed := false
 	if cur, ok := h.agents[ac.sessionID]; ok && cur == ac {
 		delete(h.agents, ac.sessionID)
+		removed = true
 	}
 	h.mu.Unlock()
-	if ac.sessionID != "" {
+	// Only flip the persisted flag when no newer conn replaced this one;
+	// a reconnect races this unregister path.
+	if removed && ac.sessionID != "" {
 		if err := h.store.SetOnline(ac.sessionID, false); err != nil {
 			log.Printf("hub: mark offline %s: %v", ac.sessionID, err)
 		}