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

Add team tasks: several agents in one worktree sharing a board

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-18 08:48:57 -07:00
parent 7f62df19d2
commit 6bcde13b2f
31 changed files with 1392 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agents/manager.go
+85 -9
View File
@@ -57,6 +57,7 @@ type Manager struct {
startRefused map[string]bool
finishing map[string]bool
handoffModel map[string]string
mergeTurns map[string]mergeTurn
deliverMu sync.Mutex
deliverLocks map[string]*sessionLock
@@ -88,6 +89,7 @@ type live struct {
taskID string
projectID string
owner string
member string
notes string
reviewID string
stage string
@@ -143,6 +145,7 @@ func New(st *store.Store, worktreeRoot, leadRoot, codexHome string, status func(
startRefused: map[string]bool{},
finishing: map[string]bool{},
handoffModel: map[string]string{},
mergeTurns: map[string]mergeTurn{},
signals: make(chan []api.Event, 1024),
snaps: map[string]*snapState{},
merges: map[string]*projectMerge{},
@@ -162,10 +165,10 @@ func (m *Manager) taskFor(ctx context.Context, sess api.AgentSession) (api.Task,
}
func (m *Manager) Create(ctx context.Context, task api.Task, creator api.User, prompt, model string, images ...string) (api.AgentSession, error) {
return m.create(ctx, task, creator, prompt, model, creator.Login, images)
return m.create(ctx, task, creator, "", prompt, model, creator.Login, images)
}
func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, prompt, model, actor string, images []string) (api.AgentSession, error) {
func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, member, prompt, model, actor string, images []string) (api.AgentSession, error) {
if creator.GitName == "" || creator.GitEmail == "" {
return api.AgentSession{}, ErrNoGitIdentity
}
@@ -174,20 +177,20 @@ func (m *Manager) create(ctx context.Context, task api.Task, creator api.User, p
return api.AgentSession{}, fmt.Errorf("%w: task %s is %s", ErrConflict, task.Id, task.Status)
}
sess, err := m.newSession(ctx, task, creator, model)
sess, err := m.newSession(ctx, task, creator, member, model)
if err != nil {
return api.AgentSession{}, err
}
return m.turn(ctx, sess, task, actor, prompt, images)
}
func (m *Manager) newSession(ctx context.Context, task api.Task, creator api.User, model string) (api.AgentSession, error) {
func (m *Manager) newSession(ctx context.Context, task api.Task, creator api.User, member, model string) (api.AgentSession, error) {
existing, err := m.store.ListSessions(ctx, task.Id)
if err != nil {
return api.AgentSession{}, err
}
for _, s := range existing {
if s.State != api.AgentSessionStateCompleted {
if s.State != api.AgentSessionStateCompleted && s.Member == member {
return api.AgentSession{}, fmt.Errorf("%w: task %s already has session %s", ErrConflict, task.Id, s.Id)
}
}
@@ -202,6 +205,7 @@ func (m *Manager) newSession(ctx context.Context, task api.Task, creator api.Use
VendorSessionId: initialVendorSessionID(vendor, id),
State: api.AgentSessionStateIdle,
Role: api.Implementer,
Member: member,
CreatedBy: creator.Login,
CreatedAt: time.Now().UTC(),
}
@@ -347,6 +351,11 @@ func (m *Manager) Recover(ctx context.Context) error {
return err
}
err = m.recoverMergeTurns(ctx, sessions)
if err != nil {
return err
}
for _, sess := range sessions {
if staged[sess.Id] {
continue
@@ -415,6 +424,7 @@ func (m *Manager) Recover(ctx context.Context) error {
func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, task api.Task, err error) error {
slog.Error("resume session", "session", sess.Id, "err", err)
m.endMergeTurn(sess.Id, true)
m.logUndelivered(ctx, sess.Id, "resume failed")
derr := m.store.DropStageOutput(ctx, sess.Id)
if derr != nil {
@@ -429,7 +439,7 @@ func (m *Manager) failResume(ctx context.Context, sess api.AgentSession, task ap
go m.finishDiscard(task.Id)
case next == store.DiscardHeld:
go m.startDiscard(task.Id)
case sess.Role == api.Implementer:
case sess.Role == api.Implementer && m.leadsTask(ctx, task.Id, sess.Id):
m.abandonMerge(task.Id, task.ProjectId)
}
return nil
@@ -540,6 +550,7 @@ func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Ta
err = l.sess.send(ctx, text, imgs)
if err != nil {
m.endMergeTurn(sess.Id, false)
return api.AgentSession{}, err
}
@@ -547,6 +558,7 @@ func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Ta
}
func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err error) {
m.endMergeTurn(sessionID, false)
cctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cancel()
@@ -589,6 +601,9 @@ func (m *Manager) message(ctx context.Context, l *live, actor, text string, imag
}
func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text string, images []string) ([]image, error) {
if l.implementer() && isMergeInPrompt(actor, text) {
m.takeMergeTurn(l)
}
imgs := []image{}
for _, id := range images {
b, data, err := m.store.ReadBlob(ctx, id)
@@ -707,6 +722,7 @@ func (m *Manager) startLive(ctx context.Context, sess api.AgentSession, task api
taskID: task.Id,
projectID: task.ProjectId,
owner: sess.Owner,
member: sess.Member,
subagents: map[string]bool{},
}
@@ -715,6 +731,7 @@ func (m *Manager) startLive(ctx context.Context, sess api.AgentSession, task api
return nil, err
}
tools := append([]mcp.Tool{m.askTool(l), m.cancelQuestionTool(l), m.setMergeTitleTool(l), m.stopMergeTool(l), m.setTaskSummaryTool(l), m.searchClosedTool(l), m.exposePortTool(l), m.takeSnapshotTool(l)}, append(m.subtaskTools(l), m.clientTools(l)...)...)
lead := true
if sess.Role == api.Implementer {
g += longTitleGuidance(task)
notes, err := m.notesText(ctx, task.CreatedBy, task.ProjectId)
@@ -724,7 +741,16 @@ func (m *Manager) startLive(ctx context.Context, sess api.AgentSession, task api
l.notes = "\n\n" + taskNotesGuidance + "\n\n" + notes
tools = append(tools, m.suggestNoteTool(l))
}
if sess.Role == api.Implementer && task.WorkflowStage != nil {
if sess.Role == api.Implementer && sess.Member != "" {
sessions, err := m.store.ListSessions(ctx, task.Id)
if err != nil {
return nil, err
}
lead = sess.Member == teamLead(sessions)
g += teamGuidance(sessions, sess.Member, task.WorkflowStage != nil)
tools = append(tools, m.boardTools(l)...)
}
if sess.Role == api.Implementer && task.WorkflowStage != nil && lead {
frozen, err := m.store.TaskWorkflow(ctx, task.Id)
if err != nil {
return nil, err
@@ -828,6 +854,7 @@ func (m *Manager) pump(l *live) {
slog.Error("drop stage output", "session", l.id, "err", err)
}
}
m.endMergeTurn(l.id, true)
next, _, err := m.store.EndTurn(ctx, l.id, agentActor, map[string]any{"error": "agent process exited mid-turn"}, nil, false, false)
if err != nil {
slog.Error("record session exit", "session", l.id, "err", err)
@@ -845,7 +872,9 @@ func (m *Manager) pump(l *live) {
return
}
if l.owner == "" {
m.abandonMerge(l.taskID, l.projectID)
if m.leadsTask(ctx, l.taskID, l.id) {
m.abandonMerge(l.taskID, l.projectID)
}
m.fileError(l, "The agent process exited mid-turn. Send a message to resume it.")
}
}
@@ -1026,11 +1055,49 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
finished := !res.isError && !interrupted && strings.TrimSpace(res.result) != ""
payload := res.payload
if l.member != "" {
payload["member"] = l.member
}
owner := m.ownsTask(l.id)
keep := false
if owner {
defer func() {
if !keep {
m.endMergeTurn(l.id, false)
}
}()
if !finished {
err := git.AbortMerge(l.dir)
if err != nil {
slog.Error("abort unfinished merge", "session", l.id, "err", err)
}
}
}
unlock := func() {}
if !owner {
locked := m.lockTask(l.taskID)
once := sync.OnceFunc(locked)
unlock = once
defer once()
task, err := m.store.GetTask(ctx, l.taskID)
if err != nil {
return err
}
if task.Status != api.TaskStatusActive {
return nil
}
}
sha, err := git.CommitAll(l.dir, commitMessage(res), l.gitName, l.gitEmail)
if err != nil {
slog.Error("commit turn", "session", l.id, "err", err)
m.fileError(l, fmt.Sprintf("byre couldn't commit the turn's work, so the turn counts as unfinished: %v", err))
finished = false
if owner {
err = git.AbortMerge(l.dir)
if err != nil {
slog.Error("abort unfinished merge", "session", l.id, "err", err)
}
}
}
handoff := compacting && finished
if sha != "" {
@@ -1084,10 +1151,14 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
if err == nil && !stageFinished {
err = m.store.DropStageOutput(ctx, l.id)
}
if err == nil && stageFinished {
err = m.recordStage(ctx, l)
}
if err != nil {
return err
}
}
unlock()
held := len(l.subagents) > 0
turn := m.event(l, agentActor, "turn.ended", payload)
@@ -1120,10 +1191,14 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
}
if finalizeBy != "" {
m.logUndelivered(ctx, l.id, "turn ended in merge")
keep = owner
go m.finishMerge(l, finalizeBy)
return nil
}
mergeBy := m.takeMerge(l.taskID)
mergeBy := ""
if m.leadsTask(ctx, l.taskID, l.id) {
mergeBy = m.takeMerge(l.taskID)
}
switch {
case mergeBy != "":
go m.releaseMerge(l.projectID, l.taskID)
@@ -1155,6 +1230,7 @@ func (m *Manager) afterTurn(ctx context.Context, l *live) {
if err != nil {
slog.Error("deliver inbox responses", "session", l.id, "err", err)
}
m.deliverTeam(ctx, l)
}
func commitMessage(res endResult) string {