diff --git a/internal/process/manager.go b/internal/process/manager.go index 022031f..46ba7e5 100644 --- a/internal/process/manager.go +++ b/internal/process/manager.go @@ -2,19 +2,27 @@ package process import ( "fmt" + "io" "os" "os/exec" "sync" + "syscall" "codespace/internal/util" ) +const ( + outputBufferSize = 4 * 1024 + ringBufferSize = 64 * 1024 +) + // Manager manages OpenCode processes per workspace. type Manager interface { Start(workspaceID string, workspaceRoot string) error Stop(workspaceID string) error Restart(workspaceID string, workspaceRoot string) error Status(workspaceID string) Status + Subscribe(workspaceID string) (Subscription, error) } // LocalManager implements Manager using local OS processes. @@ -51,19 +59,108 @@ func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error { cmd.Dir = workspaceRoot cmd.Env = append(os.Environ(), fmt.Sprintf("HOME=%s", workspaceRoot)) + stdinR, stdinW, err := os.Pipe() + if err != nil { + return util.Wrap(util.CodeInternal, "failed to create stdin pipe", err) + } + + stdoutR, stdoutW, err := os.Pipe() + if err != nil { + stdinR.Close() + stdinW.Close() + return util.Wrap(util.CodeInternal, "failed to create stdout pipe", err) + } + + cmd.Stdin = stdinR + cmd.Stdout = stdoutW + cmd.Stderr = stdoutW + if err := cmd.Start(); err != nil { + stdinR.Close() + stdinW.Close() + stdoutR.Close() + stdoutW.Close() return util.Wrap(util.CodeInternal, "failed to start process", err) } - m.sessions[workspaceID] = &Session{ + stdinR.Close() + stdoutW.Close() + + sess := &Session{ WorkspaceID: workspaceID, Root: workspaceRoot, Cmd: cmd, + Stdin: stdinW, } + m.sessions[workspaceID] = sess + + outputDone := make(chan struct{}) + go m.captureOutput(sess, stdoutR, outputDone) + go m.waitExit(sess, outputDone) 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 +// 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 { + sub.(*subscription).send(chunk) + } + } + if err != nil { + if err != io.EOF { + // Ignore read errors; the pipe is closing. + } + break + } + } +} + +// waitExit waits for the process to exit, then records the exit status, closes +// the stdin writer, and closes every subscriber channel exactly once. +func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) { + _ = sess.Cmd.Wait() + <-outputDone + + sess.mu.Lock() + defer sess.mu.Unlock() + + if sess.Cmd.ProcessState != nil { + sess.Exit.Code = sess.Cmd.ProcessState.ExitCode() + if ws, ok := sess.Cmd.ProcessState.Sys().(syscall.WaitStatus); ok && ws.Signaled() { + sess.Exit.Signal = ws.Signal().String() + } + } + + if sess.Stdin != nil { + _ = sess.Stdin.Close() + } + + for _, sub := range sess.Subscribers { + sub.(*subscription).closeChan() + } +} + // Stop kills the process for the given workspace. // Returns CodeNotFound if no session exists. func (m *LocalManager) Stop(workspaceID string) error { @@ -117,3 +214,51 @@ func (m *LocalManager) Status(workspaceID string) Status { PID: sess.Cmd.Process.Pid, } } + +// Subscribe creates a new output subscription for the workspace. +// Returns CodeNotFound if the workspace has no session. +func (m *LocalManager) Subscribe(workspaceID string) (Subscription, error) { + m.mu.Lock() + sess, ok := m.sessions[workspaceID] + m.mu.Unlock() + + if !ok { + return nil, util.New(util.CodeNotFound, "no running process") + } + + sess.mu.Lock() + sub := newSubscription(sess) + sess.Subscribers = append(sess.Subscribers, sub) + if sess.Cmd.ProcessState != nil { + sub.closeChan() + } + sess.mu.Unlock() + + return sub, 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 1122eee..4a42152 100644 --- a/internal/process/session.go +++ b/internal/process/session.go @@ -1,6 +1,10 @@ package process -import "os/exec" +import ( + "io" + "os/exec" + "sync" +) // Status represents the status of a process for a workspace. type Status struct { @@ -9,9 +13,32 @@ type Status struct { PID int `json:"pid,omitempty"` } +// ExitInfo holds the exit reason of a process. +type ExitInfo struct { + Code int + Signal string +} + // Session holds the runtime state of a started OpenCode process. type Session struct { WorkspaceID string Root string Cmd *exec.Cmd + Stdin io.WriteCloser + Subscribers []Subscription + Exit ExitInfo + mu sync.Mutex +} + +// removeSubscription removes sub from s.Subscribers. It is safe to call when +// the subscription is no longer present. +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 + } + } } diff --git a/internal/process/subscription.go b/internal/process/subscription.go new file mode 100644 index 0000000..641dd14 --- /dev/null +++ b/internal/process/subscription.go @@ -0,0 +1,66 @@ +package process + +import "sync" + +// Subscription is a consumer handle for a process output stream. +type Subscription interface { + Output() <-chan []byte + Close() error +} + +// subscription is a buffered, closable fan-out channel for a Session. +type subscription struct { + session *Session + ch chan []byte + once sync.Once + mu sync.Mutex + closed bool +} + +func newSubscription(sess *Session) *subscription { + return &subscription{ + session: sess, + ch: make(chan []byte, 32), + } +} + +func (s *subscription) Output() <-chan []byte { + return s.ch +} + +func (s *subscription) Close() error { + s.once.Do(func() { + s.mu.Lock() + s.closed = true + close(s.ch) + s.mu.Unlock() + s.session.removeSubscription(s) + }) + return nil +} + +// closeChan closes the underlying channel exactly once without removing the +// subscription from the session's subscriber list. It is used by the manager +// when the process exits. +func (s *subscription) closeChan() { + s.once.Do(func() { + s.mu.Lock() + s.closed = true + close(s.ch) + s.mu.Unlock() + }) +} + +// send performs a non-blocking send to the subscription channel. If the +// channel is full or already closed, the chunk is dropped. +func (s *subscription) send(chunk []byte) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + select { + case s.ch <- chunk: + default: + } +}