Merge branch 'event-emitter-7z1e' into cj/refactor/event-emitter

Author: johnstcn

lib/httpapi/events.go

@@ -64,7 +64,7 @@ type EventEmitter struct {
 	agentType           mf.AgentType
 	chans               map[int]chan Event
 	chanIdx             int
-	subscriptionBufSize int
+	subscriptionBufSize uint
 	screen              string
 }
 
@@ -81,13 +81,17 @@ func convertStatus(status st.ConversationStatus) AgentStatus {
 	}
 }
 
-const defaultSubscriptionBufSize = 1024
+const defaultSubscriptionBufSize uint = 1024
 
 type EventEmitterOption func(*EventEmitter)
 
-func WithSubscriptionBufSize(size int) EventEmitterOption {
+func WithSubscriptionBufSize(size uint) EventEmitterOption {
 	return func(e *EventEmitter) {
-		e.subscriptionBufSize = size
+		if size == 0 {
+			e.subscriptionBufSize = defaultSubscriptionBufSize
+		} else {
+			e.subscriptionBufSize = size
+		}
 	}
 }
 
@@ -150,6 +154,9 @@ func (e *EventEmitter) EmitMessages(newMessages []st.ConversationMessage) {
 			newMsg = newMessages[i]
 		}
 		if oldMsg != newMsg {
+			if i >= len(newMessages) {
+				continue
+			}
 			e.notifyChannels(EventTypeMessageUpdate, MessageUpdateBody{
 				Id:      newMessages[i].Id,
 				Role:    newMessages[i].Role,

lib/screentracker/pty_conversation.go

@@ -68,7 +68,7 @@ type PTYConversationConfig struct {
 	FormatToolCall func(message string) (string, []string)
 	// InitialPrompt is the initial prompt to send to the agent once ready
 	InitialPrompt []MessagePart
-	Logger     *slog.Logger
+	Logger        *slog.Logger
 }
 
 func (cfg PTYConversationConfig) getStableSnapshotsThreshold() int {
@@ -114,10 +114,19 @@ type PTYConversation struct {
 
 var _ Conversation = &PTYConversation{}
 
+type noopEmitter struct{}
+
+func (noopEmitter) EmitMessages([]ConversationMessage) {}
+func (noopEmitter) EmitStatus(ConversationStatus)      {}
+func (noopEmitter) EmitScreen(string)                  {}
+
 func NewPTY(ctx context.Context, cfg PTYConversationConfig, emitter Emitter) *PTYConversation {
 	if cfg.Clock == nil {
 		cfg.Clock = quartz.NewReal()
 	}
+	if emitter == nil {
+		emitter = noopEmitter{}
+	}
 	threshold := cfg.getStableSnapshotsThreshold()
 	c := &PTYConversation{
 		cfg:                      cfg,

lib/screentracker/pty_conversation_test.go

@@ -20,8 +20,8 @@ const testTimeout = 10 * time.Second
 
 // testAgent is a goroutine-safe mock implementation of AgentIO.
 type testAgent struct {
-	mu      sync.Mutex
-	screen  string
+	mu     sync.Mutex
+	screen string
 	// onWrite is called during Write to simulate the agent reacting to
 	// terminal input (e.g., changing the screen), which unblocks
 	// writeStabilize's polling loops.
@@ -53,7 +53,7 @@ type testEmitter struct{}
 
 func (testEmitter) EmitMessages([]st.ConversationMessage) {}
 func (testEmitter) EmitStatus(st.ConversationStatus)      {}
-func (testEmitter) EmitScreen(string)                      {}
+func (testEmitter) EmitScreen(string)                     {}
 
 // advanceFor is a shorthand for advanceUntil with a time-based condition.
 func advanceFor(ctx context.Context, t *testing.T, mClock *quartz.Mock, total time.Duration) {
@@ -226,11 +226,11 @@ func TestMessages(t *testing.T) {
 		mClock := quartz.NewMock(t)
 		mClock.Set(now)
 		cfg := st.PTYConversationConfig{
-			Clock:                      mClock,
-			AgentIO:                    agent,
-			SnapshotInterval:           100 * time.Millisecond,
-			ScreenStabilityLength:      200 * time.Millisecond,
-			Logger:                     slog.New(slog.NewTextHandler(io.Discard, nil)),
+			Clock:                 mClock,
+			AgentIO:               agent,
+			SnapshotInterval:      100 * time.Millisecond,
+			ScreenStabilityLength: 200 * time.Millisecond,
+			Logger:                slog.New(slog.NewTextHandler(io.Discard, nil)),
 		}
 		for _, opt := range opts {
 			opt(&cfg)
@@ -519,15 +519,15 @@ func TestInitialPromptReadiness(t *testing.T) {
 			agent.screen = fmt.Sprintf("__write_%d", writeCounter)
 		}
 		cfg := st.PTYConversationConfig{
-			Clock:                      mClock,
-			SnapshotInterval:           1 * time.Second,
-			ScreenStabilityLength:      0,
-			AgentIO:                    agent,
+			Clock:                 mClock,
+			SnapshotInterval:      1 * time.Second,
+			ScreenStabilityLength: 0,
+			AgentIO:               agent,
 			ReadyForInitialPrompt: func(message string) bool {
 				return message == "ready"
 			},
-			InitialPrompt:             []st.MessagePart{st.MessagePartText{Content: "initial prompt here"}},
-			Logger:                     discardLogger,
+			InitialPrompt: []st.MessagePart{st.MessagePartText{Content: "initial prompt here"}},
+			Logger:        discardLogger,
 		}
 
 		c := st.NewPTY(ctx, cfg, &testEmitter{})
@@ -585,11 +585,11 @@ func TestInitialPromptReadiness(t *testing.T) {
 		mClock := quartz.NewMock(t)
 		agent := &testAgent{screen: "ready"}
 		cfg := st.PTYConversationConfig{
-			Clock:                      mClock,
-			SnapshotInterval:           1 * time.Second,
-			ScreenStabilityLength:      2 * time.Second, // threshold = 3
-			AgentIO:                    agent,
-			Logger:                     discardLogger,
+			Clock:                 mClock,
+			SnapshotInterval:      1 * time.Second,
+			ScreenStabilityLength: 2 * time.Second, // threshold = 3
+			AgentIO:               agent,
+			Logger:                discardLogger,
 		}
 
 		c := st.NewPTY(ctx, cfg, &testEmitter{})