SSE reconnect, task switcher state/inbox/new task, stream throttle

internal/agents/manager.go

@@ -587,10 +587,17 @@ func (m *Manager) pump(l *live) {
 	ctx := context.Background()
 
 	for ev := range l.sess.Events() {
+		start := time.Now()
+		if wait := start.Sub(ev.Read); wait > 200*time.Millisecond {
+			slog.Info("agent event waited", "session", l.id, "type", ev.Type, "wait", wait)
+		}
 		err := m.handle(ctx, l, ev)
 		if err != nil {
 			slog.Error("record agent event", "session", l.id, "type", ev.Type, "err", err)
 		}
+		if took := time.Since(start); took > 200*time.Millisecond {
+			slog.Info("agent event slow", "session", l.id, "type", ev.Type, "took", took)
+		}
 	}
 
 	m.mu.Lock()

internal/claude/session.go

@@ -45,6 +45,7 @@ type Event struct {
 	Type    string
 	Subtype string
 	Raw     json.RawMessage
+	Read    time.Time
 }
 
 type Block struct {
@@ -505,6 +506,7 @@ func (s *Session) readLoop(stdout io.Reader) {
 }
 
 func (s *Session) handleLine(line []byte) {
+	read := time.Now()
 	env := envelope{}
 	err := json.Unmarshal(line, &env)
 	if err != nil {
@@ -540,7 +542,7 @@ func (s *Session) handleLine(line []byte) {
 		s.idMu.Unlock()
 	}
 
-	s.events <- Event{Type: env.Type, Subtype: env.Subtype, Raw: append(json.RawMessage{}, line...)}
+	s.events <- Event{Type: env.Type, Subtype: env.Subtype, Raw: append(json.RawMessage{}, line...), Read: read}
 }
 
 func (s *Session) learnScrub(raw json.RawMessage) {

internal/server/sse.go

@@ -48,7 +48,7 @@ func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request, params api
 		return
 	}
 
-	ping := time.NewTicker(30 * time.Second)
+	ping := time.NewTicker(15 * time.Second)
 	defer ping.Stop()
 
 	for {
@@ -68,7 +68,7 @@ func (s *Server) StreamEvents(w http.ResponseWriter, r *http.Request, params api
 			}
 
 		case <-ping.C:
-			_, err := fmt.Fprint(w, ": ping\n\n")
+			_, err := fmt.Fprint(w, "event: ping\ndata: ping\n\n")
 			if err != nil {
 				return
 			}