ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Agent startup holds the manager's global lock

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-16 12:56:53 -07:00
parent 3da0394116
commit be4b4dda0d
5 changed files with 125 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TODO.md
-2
View File
@@ -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
internal/agents/inbox.go
+4 -4
View File
@@ -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 {
internal/agents/manager.go
+78 -10
View File
@@ -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
}
internal/agents/process.go
+11
View File
@@ -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() {
internal/agents/queue.go
+32 -4
View File
@@ -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)
}