From be4b4dda0df26a8500a3aa7840211a1bdc36ef97 Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Fri, 16 Oct 2026 12:56:53 -0700 Subject: [PATCH] Agent startup holds the manager's global lock --- TODO.md | 2 - internal/agents/inbox.go | 8 ++-- internal/agents/manager.go | 88 +++++++++++++++++++++++++++++++++----- internal/agents/process.go | 11 +++++ internal/agents/queue.go | 36 ++++++++++++++-- 5 files changed, 125 insertions(+), 20 deletions(-) diff --git a/TODO.md b/TODO.md index ca23c66..0f5398e 100644 --- a/TODO.md +++ b/TODO.md @@ -8,8 +8,6 @@ `CheckResponse` reads open state outside the write transaction and `Respond` updates by id alone (`internal/agents/respond.go:26–34`, `:93`; `internal/store/inbox.go:140`), so two concurrent answers both land, or a withdrawal is overwritten. Condition the update on `state='open'`, check RowsAffected, roll back the batch on conflict and return 409. - A Codex RPC can deadlock the reader it waits on `drive` calls `Steer` synchronously while `readLoop` pushes notifications into a 256-entry channel (`internal/codex/session.go:186`, `:440–476`, `:598`); if it fills before the response is read, each waits on the other. Route RPC responses separately from notification consumption and give RPCs bounded deadlines, without replacing backpressure with an unbounded queue. -- Agent startup holds the manager's global lock - `ensureLive` holds `m.mu` through process start and protocol init (`internal/agents/manager.go:538–629`), and `deliverMu` serializes every session (`internal/agents/queue.go:15–16`), so one slow start stalls health, interrupt and event handling everywhere. Coordinate starts per session, release m.mu during process I/O and publish the result under the lock, and narrow delivery serialization to the session. - Git network operations have no deadline and run under a cross-project lock `exec.Command` has no context (`internal/git/git.go:348–364`) and Merge holds the shared `mergeMu` through fetch (`internal/agents/mergein.go:68`, `:101`), so one stalled remote blocks merges in every project. Add contextual deadlines and non-interactive git/SSH, and serialize the network phase per project while keeping same-project ordering. - A malformed file image URL breaks transcript rendering diff --git a/internal/agents/inbox.go b/internal/agents/inbox.go index 6fb8f8f..ae22796 100644 --- a/internal/agents/inbox.go +++ b/internal/agents/inbox.go @@ -281,8 +281,8 @@ func (m *Manager) newItem(l *live, id, thread string, kind api.InboxItemKind, pa } func (m *Manager) Deliver(ctx context.Context, sessionID string) error { - m.deliverMu.Lock() - defer m.deliverMu.Unlock() + unlock := m.lockDelivery(sessionID) + defer unlock() sess, err := m.store.GetSession(ctx, sessionID) if err != nil { @@ -356,8 +356,8 @@ func (m *Manager) deliverMidTurn(ctx context.Context, l *live) string { return m.recordQueued(ctx, l, msgs) } - m.deliverMu.Lock() - defer m.deliverMu.Unlock() + unlock := m.lockDelivery(l.id) + defer unlock() discard, err := m.store.GetDiscard(ctx, l.taskID) if err != nil { diff --git a/internal/agents/manager.go b/internal/agents/manager.go index fb87506..1bcee6b 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -40,6 +40,7 @@ type Manager struct { mu sync.Mutex live map[string]*live + starting map[string]*start closing bool rateLimits map[api.AgentSessionVendor]map[string]any leadAnswered map[string]bool @@ -49,9 +50,10 @@ type Manager struct { startedTodo map[string]time.Time finishing map[string]bool - deliverMu sync.Mutex - leadMu sync.Mutex - signals chan []api.Event + deliverMu sync.Mutex + deliverLocks map[string]*sessionLock + leadMu sync.Mutex + signals chan []api.Event snapMu sync.Mutex snaps map[string]*snapState @@ -96,6 +98,14 @@ type live struct { contextWarned bool } +type start struct { + done chan struct{} + cancel context.CancelFunc + l *live + err error + stopping bool +} + func (l *live) implementer() bool { return l.owner == "" && l.reviewID == "" } @@ -107,6 +117,8 @@ func New(st *store.Store, worktreeRoot, leadRoot, codexHome string) *Manager { leadRoot: leadRoot, codexHome: codexHome, live: map[string]*live{}, + starting: map[string]*start{}, + deliverLocks: map[string]*sessionLock{}, rateLimits: map[api.AgentSessionVendor]map[string]any{}, leadAnswered: map[string]bool{}, paneFocus: map[string]api.PaneFocus{}, @@ -270,6 +282,10 @@ func (m *Manager) stopLive(sessionID string) { if l != nil { l.stopping = true } + st := m.starting[sessionID] + if st != nil { + st.stopping = true + } m.mu.Unlock() if l != nil { @@ -406,11 +422,21 @@ func (m *Manager) Close() { for _, l := range m.live { all = append(all, l) } + pending := []*start{} + for _, st := range m.starting { + pending = append(pending, st) + } m.mu.Unlock() + for _, st := range pending { + st.cancel() + } for _, l := range all { l.sess.kill() } + for _, st := range pending { + <-st.done + } err := m.parkLeads(context.Background()) if err != nil { @@ -576,23 +602,68 @@ func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text strin func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) { m.mu.Lock() - defer m.mu.Unlock() - if m.closing { + m.mu.Unlock() return nil, errors.New("shutting down") } - l := m.live[sess.Id] if l != nil { + m.mu.Unlock() return l, nil } + st := m.starting[sess.Id] + if st != nil { + m.mu.Unlock() + select { + case <-st.done: + return st.l, st.err + case <-ctx.Done(): + return nil, ctx.Err() + } + } + sctx, cancel := context.WithCancel(ctx) + st = &start{done: make(chan struct{}), cancel: cancel} + m.starting[sess.Id] = st + m.mu.Unlock() + l, err := m.startLive(sctx, sess, task) + cancel() + + m.mu.Lock() + delete(m.starting, sess.Id) + closing := m.closing + stopping := st.stopping + if err == nil && !closing { + l.stopping = stopping + m.live[sess.Id] = l + } + m.mu.Unlock() + + if err == nil && closing { + go l.sess.discard() + l.sess.kill() + l, err = nil, errors.New("shutting down") + } + st.l, st.err = l, err + close(st.done) + if err != nil { + return nil, err + } + + go m.pump(l) + if stopping { + l.sess.kill() + } + return l, nil +} + +func (m *Manager) startLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) { creator, err := m.store.GetUser(ctx, sess.CreatedBy) if err != nil { return nil, err } - l = &live{ + l := &live{ id: sess.Id, dir: task.WorktreePath, model: sess.Model, @@ -661,9 +732,6 @@ func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task ap if err != nil { return nil, err } - - m.live[sess.Id] = l - go m.pump(l) return l, nil } diff --git a/internal/agents/process.go b/internal/agents/process.go index 10abfa3..5cdc80a 100644 --- a/internal/agents/process.go +++ b/internal/agents/process.go @@ -28,6 +28,7 @@ type agentProcess interface { lastLine() time.Time exited() bool drive(m *Manager, l *live) + discard() } func initialVendorSessionID(vendor api.AgentSessionVendor, id string) string { @@ -121,6 +122,11 @@ func (p claudeProcess) pid() int { return p.s.Pid() } func (p claudeProcess) lastLine() time.Time { return p.s.LastLine() } func (p claudeProcess) exited() bool { return p.s.Exited() } +func (p claudeProcess) discard() { + for range p.s.Events() { + } +} + func (p claudeProcess) drive(m *Manager, l *live) { ctx := context.Background() for ev := range p.s.Events() { @@ -208,6 +214,11 @@ func (p codexProcess) pid() int { return p.s.Pid() } func (p codexProcess) lastLine() time.Time { return p.s.LastLine() } func (p codexProcess) exited() bool { return p.s.Exited() } +func (p codexProcess) discard() { + for range p.s.Events() { + } +} + func (p codexProcess) drive(m *Manager, l *live) { ctx := context.Background() for ev := range p.s.Events() { diff --git a/internal/agents/queue.go b/internal/agents/queue.go index 030b438..df7996c 100644 --- a/internal/agents/queue.go +++ b/internal/agents/queue.go @@ -6,14 +6,42 @@ import ( "fmt" "log/slog" "strings" + "sync" "g.fc.run/misc/byre/internal/api" "g.fc.run/misc/byre/internal/store" ) -func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model string, images ...string) (api.AgentSession, error) { +type sessionLock struct { + mu sync.Mutex + refs int +} + +func (m *Manager) lockDelivery(sessionID string) func() { m.deliverMu.Lock() - defer m.deliverMu.Unlock() + sl := m.deliverLocks[sessionID] + if sl == nil { + sl = &sessionLock{} + m.deliverLocks[sessionID] = sl + } + sl.refs++ + m.deliverMu.Unlock() + + sl.mu.Lock() + return func() { + sl.mu.Unlock() + m.deliverMu.Lock() + sl.refs-- + if sl.refs == 0 { + delete(m.deliverLocks, sessionID) + } + m.deliverMu.Unlock() + } +} + +func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model string, images ...string) (api.AgentSession, error) { + unlock := m.lockDelivery(sessionID) + defer unlock() sess, err := m.store.GetSession(ctx, sessionID) if err != nil { @@ -77,9 +105,9 @@ func (m *Manager) deliverAllQueued(ctx context.Context) { if sess.Role != api.Implementer { continue } - m.deliverMu.Lock() + unlock := m.lockDelivery(sess.Id) _, err = m.deliverQueued(ctx, sess) - m.deliverMu.Unlock() + unlock() if err != nil { slog.Error("deliver queued message", "session", sess.Id, "err", err) }