From 4a4028486cde1b55a5db0235a812a2d8ab4e61cc Mon Sep 17 00:00:00 2001
From: "tao.chen" <93983997+taochen-ct@users.noreply.github.com>
Date: Fri, 3 Jul 2026 14:08:20 +0800
Subject: [PATCH] refactor(process): use map for subscribers and drop dead ring
 buffer

Co-Authored-By: Claude <noreply@anthropic.com>
---
 internal/process/manager.go | 43 ++++++-------------------------------
 internal/process/session.go |  9 ++------
 2 files changed, 8 insertions(+), 44 deletions(-)

diff --git a/internal/process/manager.go b/internal/process/manager.go
index bac6d37..b9f17f5 100644
--- a/internal/process/manager.go
+++ b/internal/process/manager.go
@@ -13,7 +13,6 @@ import (
 
 const (
 	outputBufferSize = 4 * 1024
-	ringBufferSize   = 64 * 1024
 )
 
 // Manager manages OpenCode processes per workspace.
@@ -93,6 +92,7 @@ func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error {
 		Root:        workspaceRoot,
 		Cmd:         cmd,
 		Stdin:       stdinW,
+		Subscribers: make(map[Subscription]struct{}),
 	}
 	m.sessions[workspaceID] = sess
 
@@ -103,31 +103,25 @@ func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error {
 	return nil
 }
 
-// captureOutput reads from the merged stdout/stderr pipe, keeps the most
-// recent ringBufferSize bytes in an in-memory ring buffer, and fans out each
+// captureOutput reads from the merged stdout/stderr pipe and fans out each
 // chunk to all subscribers using non-blocking sends.
 func (m *LocalManager) captureOutput(sess *Session, stdoutR *os.File, done chan<- struct{}) {
 	defer close(done)
 	defer stdoutR.Close()
 
 	buf := make([]byte, outputBufferSize)
-	ring := newRingBuffer(ringBufferSize)
 
 	for {
 		n, err := stdoutR.Read(buf)
 		if n > 0 {
 			chunk := make([]byte, n)
 			copy(chunk, buf[:n])
-			ring.Write(chunk)
 
 			sess.mu.Lock()
-			subs := make([]Subscription, len(sess.Subscribers))
-			copy(subs, sess.Subscribers)
-			sess.mu.Unlock()
-
-			for _, sub := range subs {
+			for sub := range sess.Subscribers {
 				sub.(*subscription).send(chunk)
 			}
+			sess.mu.Unlock()
 		}
 		if err != nil {
 			if err != io.EOF {
@@ -158,7 +152,7 @@ func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) {
 		_ = sess.Stdin.Close()
 	}
 
-	for _, sub := range sess.Subscribers {
+	for sub := range sess.Subscribers {
 		sub.(*subscription).closeChan()
 	}
 }
@@ -230,7 +224,7 @@ func (m *LocalManager) Subscribe(workspaceID string) (Subscription, error) {
 
 	sess.mu.Lock()
 	sub := newSubscription(sess)
-	sess.Subscribers = append(sess.Subscribers, sub)
+	sess.Subscribers[sub] = struct{}{}
 	if sess.Cmd.ProcessState != nil {
 		sub.closeChan()
 	}
@@ -274,28 +268,3 @@ func (m *LocalManager) exitOf(workspaceID string) (ExitInfo, error) {
 	return sess.Exit, nil
 }
 
-// ringBuffer is a fixed-size byte buffer that drops the oldest bytes when full.
-type ringBuffer struct {
-	mu   sync.Mutex
-	buf  []byte
-	maxN int
-}
-
-func newRingBuffer(maxN int) *ringBuffer {
-	return &ringBuffer{maxN: maxN}
-}
-
-func (r *ringBuffer) Write(p []byte) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	if len(p) >= r.maxN {
-		r.buf = append(r.buf[:0], p[len(p)-r.maxN:]...)
-		return
-	}
-
-	r.buf = append(r.buf, p...)
-	if len(r.buf) > r.maxN {
-		r.buf = r.buf[len(r.buf)-r.maxN:]
-	}
-}
diff --git a/internal/process/session.go b/internal/process/session.go
index 4a42152..26fb63a 100644
--- a/internal/process/session.go
+++ b/internal/process/session.go
@@ -25,7 +25,7 @@ type Session struct {
 	Root        string
 	Cmd         *exec.Cmd
 	Stdin       io.WriteCloser
-	Subscribers []Subscription
+	Subscribers map[Subscription]struct{}
 	Exit        ExitInfo
 	mu          sync.Mutex
 }
@@ -35,10 +35,5 @@ type Session struct {
 func (s *Session) removeSubscription(sub *subscription) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	for i, cur := range s.Subscribers {
-		if cur == sub {
-			s.Subscribers = append(s.Subscribers[:i], s.Subscribers[i+1:]...)
-			return
-		}
-	}
+	delete(s.Subscribers, sub)
 }