7018e79 fix(agent): bound startup alert buffering · zut

M packages/agent/cli.go +6 -1


@@ -32,6 +32,9 @@ import (
32	32		// manager call back into the Interactive instance built later in
33	33		// runInteractive. Alerts are buffered until the instance is attached so
34	34		// an extension that signals attention during startup is not silently lost.
	35	+	// Keep the startup buffer bounded because extensions can flood their host.
	36	+	const maxBufferedInteractiveAlerts = 64
	37	+
35	38		type interactiveExtHooks struct {
36	39		mu sync.Mutex
37	40		interactive *modes.Interactive

@@ -77,7 +80,9 @@ func (h *interactiveExtHooks) Alert(extName string, alert extproto.AlertRequest)
77	80		}
78	81		h.mu.Lock()
79	82		if h.interactive == nil {
80		-	h.pendingAlerts = append(h.pendingAlerts, interactiveAlert{extName: extName, alert: alert})
	83	+	if len(h.pendingAlerts) < maxBufferedInteractiveAlerts {
	84	+	h.pendingAlerts = append(h.pendingAlerts, interactiveAlert{extName: extName, alert: alert})
	85	+	}
81	86		h.mu.Unlock()
82	87		return
83	88		}

 // manager call back into the Interactive instance built later in
 // runInteractive. Alerts are buffered until the instance is attached so
 // an extension that signals attention during startup is not silently lost.
+// Keep the startup buffer bounded because extensions can flood their host.
+const maxBufferedInteractiveAlerts = 64
 type interactiveExtHooks struct {
 	mu            sync.Mutex
 	interactive   *modes.Interactive
 	}
 	h.mu.Lock()
 	if h.interactive == nil {
-		h.pendingAlerts = append(h.pendingAlerts, interactiveAlert{extName: extName, alert: alert})
+		if len(h.pendingAlerts) < maxBufferedInteractiveAlerts {
+			h.pendingAlerts = append(h.pendingAlerts, interactiveAlert{extName: extName, alert: alert})
+		}
 		h.mu.Unlock()
 		return
 	}

M packages/agent/cli_alert_test.go +5 -2


@@ -10,12 +10,15 @@ import (
10	10		func TestInteractiveExtHooksBufferStartupAlerts(t *testing.T) {
11	11		hooks := &interactiveExtHooks{}
12	12		hooks.Alert("question-ext", extproto.AlertRequest{Kind: extproto.AlertKindBell, Reason: "question_ready"})
	13	+	for n := 0; n < maxBufferedInteractiveAlerts+8; n++ {
	14	+	hooks.Alert("question-ext", extproto.AlertRequest{Kind: extproto.AlertKindBell, Reason: "flood"})
	15	+	}
13	16
14	17		hooks.mu.Lock()
15	18		pending := len(hooks.pendingAlerts)
16	19		hooks.mu.Unlock()
17		-	if pending != 1 {
18		-	t.Fatalf("pending startup alerts = %d, want 1", pending)
	20	+	if pending != maxBufferedInteractiveAlerts {
	21	+	t.Fatalf("pending startup alerts = %d, want bounded capacity %d", pending, maxBufferedInteractiveAlerts)
19	22		}
20	23
21	24		iv := &modes.Interactive{}

 func TestInteractiveExtHooksBufferStartupAlerts(t *testing.T) {
 	hooks := &interactiveExtHooks{}
 	hooks.Alert("question-ext", extproto.AlertRequest{Kind: extproto.AlertKindBell, Reason: "question_ready"})
+	for n := 0; n < maxBufferedInteractiveAlerts+8; n++ {
+		hooks.Alert("question-ext", extproto.AlertRequest{Kind: extproto.AlertKindBell, Reason: "flood"})
+	}
 	hooks.mu.Lock()
 	pending := len(hooks.pendingAlerts)
 	hooks.mu.Unlock()
-	if pending != 1 {
+	if pending != maxBufferedInteractiveAlerts {
-		t.Fatalf("pending startup alerts = %d, want 1", pending)
+		t.Fatalf("pending startup alerts = %d, want bounded capacity %d", pending, maxBufferedInteractiveAlerts)
 	}
 	iv := &modes.Interactive{}

M packages/agent/ext/ext_test.go +13 -2


@@ -222,7 +222,19 @@ func TestOpenPanelEmitsCorrectFrame(t *testing.T) {
222	222
223	223		func TestAlertEmitsStructuredFrame(t *testing.T) {
224	224		h := newHarness("alert-ext")
225		-	go h.ext.Run()
	225	+	runDone := make(chan error, 1)
	226	+	go func() { runDone <- h.ext.Run() }()
	227	+	t.Cleanup(func() {
	228	+	_ = h.hostW.Close()
	229	+	select {
	230	+	case err := <-runDone:
	231	+	if err != nil {
	232	+	t.Errorf("extension Run: %v", err)
	233	+	}
	234	+	case <-time.After(2 * time.Second):
	235	+	t.Error("extension Run did not exit after closing host input")
	236	+	}
	237	+	})
226	238		h.handshake(t)
227	239
228	240		go h.ext.Alert(AlertRequest{Kind: AlertKindBell, Reason: "question_ready"})

@@ -235,7 +247,6 @@ func TestAlertEmitsStructuredFrame(t *testing.T) {
235	247		if alert.Kind != AlertKindBell \|\| alert.Reason != "question_ready" {
236	248		t.Fatalf("alert = %+v, want bell/question_ready", alert)
237	249		}
238		-	h.hostW.Close()
239	250		}
240	251
241	252		func TestToolConfirmationRequestedEvent(t *testing.T) {

 func TestAlertEmitsStructuredFrame(t *testing.T) {
 	h := newHarness("alert-ext")
-	go h.ext.Run()
+	runDone := make(chan error, 1)
+	go func() { runDone <- h.ext.Run() }()
+	t.Cleanup(func() {
+		_ = h.hostW.Close()
+		select {
+		case err := <-runDone:
+			if err != nil {
+				t.Errorf("extension Run: %v", err)
+			}
+		case <-time.After(2 * time.Second):
+			t.Error("extension Run did not exit after closing host input")
+		}
+	})
 	h.handshake(t)
 	go h.ext.Alert(AlertRequest{Kind: AlertKindBell, Reason: "question_ready"})
 	if alert.Kind != AlertKindBell || alert.Reason != "question_ready" {
 		t.Fatalf("alert = %+v, want bell/question_ready", alert)
 	}
-	h.hostW.Close()
 }
 func TestToolConfirmationRequestedEvent(t *testing.T) {

M packages/agent/modes/alert_test.go +10 -0


@@ -89,27 +89,37 @@ func TestCancelledMainTurnDoesNotBell(t *testing.T) {
89	89		i.startTurn(context.Background(), "question")
90	90
91	91		deadline := time.Now().Add(time.Second)
	92	+	started := false
92	93		for time.Now().Before(deadline) {
93	94		i.mu.Lock()
94	95		busy := i.busy
95	96		i.mu.Unlock()
96	97		if busy {
	98	+	started = true
97	99		break
98	100		}
99	101		time.Sleep(time.Millisecond)
100	102		}
	103	+	if !started {
	104	+	t.Fatal("turn did not enter busy state")
	105	+	}
101	106		i.CancelTurn()
102	107
103	108		deadline = time.Now().Add(time.Second)
	109	+	stopped := false
104	110		for time.Now().Before(deadline) {
105	111		i.mu.Lock()
106	112		busy := i.busy
107	113		i.mu.Unlock()
108	114		if !busy {
	115	+	stopped = true
109	116		break
110	117		}
111	118		time.Sleep(time.Millisecond)
112	119		}
	120	+	if !stopped {
	121	+	t.Fatal("turn did not finish after cancellation")
	122	+	}
113	123		if got := term.String(); got != "" {
114	124		t.Fatalf("cancelled main turn alert output = %q, want empty", got)
115	125		}


@@ -89,27 +89,37 @@ func TestCancelledMainTurnDoesNotBell(t *testing.T) {
89	i.startTurn(context.Background(), "question")	89		i.startTurn(context.Background(), "question")
90		90
91	deadline := time.Now().Add(time.Second)	91		deadline := time.Now().Add(time.Second)
		92	+	started := false
92	for time.Now().Before(deadline) {	93		for time.Now().Before(deadline) {
93	i.mu.Lock()	94		i.mu.Lock()
94	busy := i.busy	95		busy := i.busy
95	i.mu.Unlock()	96		i.mu.Unlock()
96	if busy {	97		if busy {
		98	+	started = true
97	break	99		break
98	}	100		}
99	time.Sleep(time.Millisecond)	101		time.Sleep(time.Millisecond)
100	}	102		}
		103	+	if !started {
		104	+	t.Fatal("turn did not enter busy state")
		105	+	}
101	i.CancelTurn()	106		i.CancelTurn()
102		107
103	deadline = time.Now().Add(time.Second)	108		deadline = time.Now().Add(time.Second)
		109	+	stopped := false
104	for time.Now().Before(deadline) {	110		for time.Now().Before(deadline) {
105	i.mu.Lock()	111		i.mu.Lock()
106	busy := i.busy	112		busy := i.busy
107	i.mu.Unlock()	113		i.mu.Unlock()
108	if !busy {	114		if !busy {
		115	+	stopped = true
109	break	116		break
110	}	117		}
111	time.Sleep(time.Millisecond)	118		time.Sleep(time.Millisecond)
112	}	119		}
		120	+	if !stopped {
		121	+	t.Fatal("turn did not finish after cancellation")
		122	+	}
113	if got := term.String(); got != "" {	123		if got := term.String(); got != "" {
114	t.Fatalf("cancelled main turn alert output = %q, want empty", got)	124		t.Fatalf("cancelled main turn alert output = %q, want empty", got)
115	}	125		}

M packages/agent/rpc.go +5 -1


@@ -171,6 +171,8 @@ func (h *rpcExtHooks) OpenPanel(string, extproto.PanelSpec) {}
171	171		func (h *rpcExtHooks) UpdatePanel(string, string, string, []string, string) {}
172	172		func (h *rpcExtHooks) ClosePanel(string, string) {}
173	173
	174	+	const maxPendingRPCExtEvents = 64
	175	+
174	176		type rpcServer struct {
175	177		ctx context.Context
176	178		args Args

@@ -486,7 +488,9 @@ func (s *rpcServer) writeExtensionEvent(payload map[string]any) {
486	488		s.writeMu.Lock()
487	489		defer s.writeMu.Unlock()
488	490		if !s.started \|\| !s.authed {
489		-	s.pendingExtEvents = append(s.pendingExtEvents, b)
	491	+	if len(s.pendingExtEvents) < maxPendingRPCExtEvents {
	492	+	s.pendingExtEvents = append(s.pendingExtEvents, b)
	493	+	}
490	494		return
491	495		}
492	496		s.writeLocked(b)

 func (h *rpcExtHooks) UpdatePanel(string, string, string, []string, string) {}
 func (h *rpcExtHooks) ClosePanel(string, string)                            {}
+const maxPendingRPCExtEvents = 64
 type rpcServer struct {
 	ctx      context.Context
 	args     Args
 	s.writeMu.Lock()
 	defer s.writeMu.Unlock()
 	if !s.started || !s.authed {
-		s.pendingExtEvents = append(s.pendingExtEvents, b)
+		if len(s.pendingExtEvents) < maxPendingRPCExtEvents {
+			s.pendingExtEvents = append(s.pendingExtEvents, b)
+		}
 		return
 	}
 	s.writeLocked(b)