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

Move lead delivery and the Merge & continue hold onto the coordinator

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-21 13:59:37 -07:00
parent 1734057293
commit 5c50fa857f
16 changed files with 270 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agents/compact.go
+9 -5
View File
@@ -33,16 +33,20 @@ func (m *Manager) Compact(ctx context.Context, sessionID, model string) (api.Age
if err != nil {
return api.AgentSession{}, err
}
err = m.startRunning(ctx, sess, "byre")
if err != nil {
return api.AgentSession{}, err
}
return m.launchCompaction(ctx, sess, task, model)
}
func (m *Manager) launchCompaction(ctx context.Context, sess api.AgentSession, task api.Task, model string) (api.AgentSession, error) {
prompt := handoffPrompt
if sess.Owner != "" {
prompt = leadHandoffPrompt
}
err = m.startRunning(ctx, sess, task.Id, "byre")
if err != nil {
return api.AgentSession{}, err
}
m.mu.Lock()
m.handoffModel[sessionID] = model
m.handoffModel[sess.Id] = model
m.mu.Unlock()
return m.launch(ctx, sess, task, "byre", prompt, "", "", nil)
}
internal/agents/coordinator.go
+6
View File
@@ -75,6 +75,12 @@ func (m *Manager) coordinatePass(ctx context.Context, started map[string]bool) {
for _, id := range work.AdvanceStages {
once("advance:"+id, func() { m.advanceStage(id) })
}
for _, d := range work.LeadDeliveries {
once(fmt.Sprintf("lead:%s:%d:%d", d.Owner, d.First, d.Last), func() { m.deliverLead(d.Owner) })
}
for _, q := range work.QueuedDelivery {
once(fmt.Sprintf("queue:%s:%d:%s", q.SessionID, q.Input, q.Response), func() { m.deliverQueuedTurn(q.SessionID) })
}
for _, p := range work.ReadyPrompts {
once("ready:"+p.TaskID, func() { m.promptReady(p) })
}
internal/agents/lead.go
+46 -41
View File
@@ -163,14 +163,30 @@ func (m *Manager) MessageLead(ctx context.Context, user api.User, text, model, s
if err != nil {
return api.AgentSession{}, err
}
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: user.Login, Kind: store.LeadMessage, Actor: user.Login, Text: text, Images: images, SendID: sendID, Client: client})
err = m.addLeadInput(ctx, store.LeadInput{Owner: user.Login, Kind: store.LeadMessage, Actor: user.Login, Text: text, Images: images, SendID: sendID, Client: client})
if err != nil {
return api.AgentSession{}, err
}
return sess, nil
}
func (m *Manager) DeliverLead(ctx context.Context, owner string) error {
func (m *Manager) addLeadInput(ctx context.Context, in store.LeadInput) error {
err := m.store.AddLeadInput(ctx, in)
if err != nil {
return err
}
m.poke()
return nil
}
func (m *Manager) deliverLead(owner string) {
err := m.tryDeliverLead(context.Background(), owner)
if err != nil {
slog.Error("deliver to lead", "owner", owner, "err", err)
}
}
func (m *Manager) tryDeliverLead(ctx context.Context, owner string) error {
m.leadMu.Lock()
defer m.leadMu.Unlock()
@@ -257,24 +273,6 @@ func (m *Manager) leadStateText(ctx context.Context, owner string) (string, erro
return fmt.Sprintf("Tasks now:\n%s\n%s", tasks, pane), nil
}
func (m *Manager) deliverLead(owner string) {
err := m.DeliverLead(context.Background(), owner)
if err != nil {
slog.Error("deliver to lead", "owner", owner, "err", err)
}
}
func (m *Manager) deliverAllLeads() {
owners, err := m.store.LeadOwners(context.Background())
if err != nil {
slog.Error("list lead owners", "err", err)
return
}
for _, owner := range owners {
m.deliverLead(owner)
}
}
func (m *Manager) parkLeads(ctx context.Context) error {
running, err := m.store.SessionsInState(ctx, api.AgentSessionStateRunning)
if err != nil {
@@ -288,7 +286,7 @@ func (m *Manager) parkLeads(ctx context.Context) error {
if err != nil {
return err
}
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: sess.Owner, Kind: store.LeadSignal, Actor: "byre", Text: leadRestartNotice})
err = m.addLeadInput(ctx, store.LeadInput{Owner: sess.Owner, Kind: store.LeadSignal, Actor: "byre", Text: leadRestartNotice})
if err != nil {
return err
}
@@ -322,6 +320,15 @@ func (m *Manager) leadTurnEnded(ctx context.Context, l *live, res endResult) err
}
}
if handoff {
go m.continueFromHandoff(l.id, res.result)
return nil
}
if wantCompact && finished && !compacting {
go m.compactLead(l.id)
return nil
}
var state map[string]any
if res.isError && !interrupted {
state = map[string]any{"error": fmt.Sprintf("the turn ended with an error (%s): %s", res.subtype, res.result)}
@@ -331,24 +338,23 @@ func (m *Manager) leadTurnEnded(ctx context.Context, l *live, res endResult) err
return err
}
switch {
case handoff:
go m.continueFromHandoff(l.id, res.result)
case wantCompact && finished && !compacting:
go func() {
_, err := m.Compact(context.Background(), l.id, "")
if err != nil {
slog.Error("compact lead", "session", l.id, "err", err)
}
}()
default:
err = m.warnLeadContext(ctx, l)
if err != nil {
return err
}
go m.deliverLead(l.owner)
return m.warnLeadContext(ctx, l)
}
func (m *Manager) compactLead(sessionID string) {
ctx := context.Background()
sess, err := m.store.GetSession(ctx, sessionID)
if err == nil {
_, err = m.launchCompaction(ctx, sess, api.Task{}, "")
}
if err == nil {
return
}
slog.Error("compact lead", "session", sessionID, "err", err)
err = m.store.TransitionSession(ctx, sessionID, api.AgentSessionStateRunning, api.AgentSessionStateIdle, "byre")
if err != nil && !errors.Is(err, store.ErrStateChanged) {
slog.Error("end failed lead compaction", "session", sessionID, "err", err)
}
return nil
}
func (m *Manager) warnLeadContext(ctx context.Context, l *live) error {
@@ -357,7 +363,7 @@ func (m *Manager) warnLeadContext(ctx context.Context, l *live) error {
}
l.contextWarned = true
text := fmt.Sprintf("Your context is at %d%% of its window (%d of %d tokens). Call compact_self at the next logical point, this turn if nothing is in progress.", l.contextUsed*100/l.contextWindow, l.contextUsed, l.contextWindow)
return m.store.AddLeadInput(ctx, store.LeadInput{Owner: l.owner, Kind: store.LeadSignal, Actor: "byre", Text: text})
return m.addLeadInput(ctx, store.LeadInput{Owner: l.owner, Kind: store.LeadSignal, Actor: "byre", Text: text})
}
func (m *Manager) Observe(events []api.Event) {
@@ -472,11 +478,10 @@ func (m *Manager) signal(ctx context.Context, batch []api.Event) error {
continue
}
text := strings.Join(append(states, lines...), "\n")
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: owner, Kind: store.LeadSignal, Actor: "byre", Text: text})
err = m.addLeadInput(ctx, store.LeadInput{Owner: owner, Kind: store.LeadSignal, Actor: "byre", Text: text})
if err != nil {
return err
}
m.deliverLead(owner)
}
return nil
}
internal/agents/leadnotes.go
+2 -8
View File
@@ -122,11 +122,10 @@ func (m *Manager) suggestNoteTool(l *live) mcp.Tool {
}
signal := fmt.Sprintf("%s suggests a note on how %s works:\n%s\nEvidence: %s\nSave it with remember if it holds and your notes don't already cover it, under project_id %s or for every project as fits; otherwise drop it.",
m.taskLabel(ctx, task.Id), task.CreatedBy, text, strings.TrimSpace(args.Evidence), task.ProjectId)
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: task.CreatedBy, Kind: store.LeadSignal, Actor: "byre", Text: signal})
err = m.addLeadInput(ctx, store.LeadInput{Owner: task.CreatedBy, Kind: store.LeadSignal, Actor: "byre", Text: signal})
if err != nil {
return "", err
}
m.deliverLead(task.CreatedBy)
return "Suggestion sent to the lead.", nil
},
}
@@ -142,12 +141,7 @@ func (m *Manager) DeleteLeadNote(ctx context.Context, login, id string) error {
return err
}
signal := fmt.Sprintf("The human deleted your note %s: %q. Stop applying it from now on, in recommendations, answers and handoffs, and don't write it again unless they say so. If this needs nothing from you to the human, make your final message NO_REPLY.", id, text)
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadMessage, Actor: "byre", Text: signal})
if err != nil {
return err
}
m.deliverLead(login)
return nil
return m.addLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadMessage, Actor: "byre", Text: signal})
}
func (m *Manager) notesText(ctx context.Context, owner, onlyProject string) (string, error) {
internal/agents/leadtools.go
+3 -18
View File
@@ -668,12 +668,7 @@ func (m *Manager) endWatchHold(ctx context.Context, login string, at time.Time)
return err
}
text := fmt.Sprintf("Your right pane is free: the human has watched %s's agent work for 10 seconds without typing.", m.taskLabel(ctx, *f.TaskId))
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: text})
if err != nil {
return err
}
m.deliverLead(login)
return nil
return m.addLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: text})
}
func sameView(a, b api.Focus) bool {
@@ -751,12 +746,7 @@ func (m *Manager) endMoveHold(ctx context.Context, login string, at time.Time) e
if !refused {
return nil
}
err := m.store.AddLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: moveHoldEnded})
if err != nil {
return err
}
m.deliverLead(login)
return nil
return m.addLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: moveHoldEnded})
}
const startHold = 10 * time.Second
@@ -802,12 +792,7 @@ func (m *Manager) endStartHold(ctx context.Context, login string, at time.Time)
if f.Tab != api.TabProject {
return nil
}
err = m.store.AddLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: startHoldEnded})
if err != nil {
return err
}
m.deliverLead(login)
return nil
return m.addLeadInput(ctx, store.LeadInput{Owner: login, Kind: store.LeadSignal, Actor: "byre", Text: startHoldEnded})
}
func (m *Manager) modelOr(model string, l *live) string {
internal/agents/manager.go
+8 -25
View File
@@ -63,7 +63,6 @@ type Manager struct {
startRefused map[string]bool
handoffModel map[string]string
mergeTurns map[string]mergeTurn
landing map[string]landing
deliverMu sync.Mutex
deliverLocks map[string]*sessionLock
@@ -168,7 +167,6 @@ func New(st *store.Store, worktreeRoot, leadRoot, handsRoot, codexHome, antigrav
startRefused: map[string]bool{},
handoffModel: map[string]string{},
mergeTurns: map[string]mergeTurn{},
landing: map[string]landing{},
signals: make(chan []api.Event, 1024),
snaps: map[string]*snapState{},
pokes: make(chan struct{}, 1),
@@ -475,8 +473,6 @@ func (m *Manager) Recover(ctx context.Context) error {
return err
}
go m.deliverAllLeads()
go m.deliverAllQueued(context.Background())
go m.coordinate()
m.poke()
m.MarkFacts(allTasks)
@@ -582,30 +578,15 @@ func (m *Manager) Usage(ctx context.Context) (Usage, error) {
}
func (m *Manager) turn(ctx context.Context, sess api.AgentSession, task api.Task, actor, text, sendID, client string, images []string) (api.AgentSession, error) {
err := m.startRunning(ctx, sess, task.Id, actor)
err := m.startRunning(ctx, sess, actor)
if err != nil {
return api.AgentSession{}, err
}
return m.launch(ctx, sess, task, actor, text, sendID, client, images)
}
func (m *Manager) startRunning(ctx context.Context, sess api.AgentSession, taskID, actor string) error {
held := fmt.Errorf("%w: held while this task merges and continues", ErrConflict)
if m.heldForLanding(taskID, sess.Id) {
return held
}
err := transition(ctx, m.store, sess.Id, api.AgentSessionStateIdle, api.AgentSessionStateRunning, actor)
if err != nil {
return err
}
if m.heldForLanding(taskID, sess.Id) {
err = transition(ctx, m.store, sess.Id, api.AgentSessionStateRunning, api.AgentSessionStateIdle, actor)
if err != nil {
return err
}
return held
}
return nil
func (m *Manager) startRunning(ctx context.Context, sess api.AgentSession, actor string) error {
return transition(ctx, m.store, sess.Id, api.AgentSessionStateIdle, api.AgentSessionStateRunning, actor)
}
func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Task, actor, text, sendID, client string, images []string) (api.AgentSession, error) {
@@ -1187,8 +1168,6 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
return m.reviewTurnEnded(ctx, l, res)
}
defer m.endLanding(l.taskID, l.id)
m.mu.Lock()
interrupted := l.interrupted
rateLimited := l.rateLimited
@@ -1206,7 +1185,11 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
if owner {
defer m.endMergeTurn(l.id, false)
}
if (owner || m.landingBy(l.taskID, l.id) != "") && !finished {
landingBy, err := m.store.LandingBy(ctx, l.taskID, l.id)
if err != nil {
return err
}
if (owner || landingBy != "") && !finished {
err := git.AbortMerge(l.dir)
if err != nil {
slog.Error("abort unfinished merge", "session", l.id, "err", err)
internal/agents/mergefacts.go
+8
View File
@@ -3,7 +3,9 @@ package agents
import (
"context"
"errors"
"io/fs"
"log/slog"
"os"
"g.fc.run/ai/byre/internal/api"
"g.fc.run/ai/byre/internal/git"
@@ -98,6 +100,12 @@ func (m *Manager) refreshFacts(ctx context.Context, taskID string) {
func (m *Manager) settleTaskFacts(ctx context.Context, t api.Task, p api.Project) error {
m.factsRun.Lock()
defer m.factsRun.Unlock()
if t.WorktreePath != "" {
_, err := os.Stat(t.WorktreePath)
if errors.Is(err, fs.ErrNotExist) {
return nil
}
}
g, err := m.gitFacts(ctx, t, p)
if err != nil {
slog.Error("compute merge facts", "task", t.Id, "err", err)
internal/agents/mergemain.go
+16 -69
View File
@@ -55,7 +55,11 @@ func (m *Manager) mergeToMain(ctx context.Context, l *live, raw json.RawMessage)
return "", err
}
login := sess.CreatedBy
if by := m.landingBy(l.taskID, l.id); by != "" {
by, err := m.store.LandingBy(ctx, l.taskID, l.id)
if err != nil {
return "", err
}
if by != "" {
login = by
}
user, err := m.store.GetUser(ctx, login)
@@ -187,29 +191,16 @@ func (m *Manager) MergeToMain(ctx context.Context, task api.Task, project api.Pr
unlock := m.lockDelivery(sess.Id)
defer unlock()
m.mu.Lock()
holder := m.landing[task.Id]
if holder.session == "" {
m.landing[task.Id] = landing{session: sess.Id, login: user.Login}
}
m.mu.Unlock()
if holder.session != "" {
return api.Task{}, fmt.Errorf("%w: task %s is already merging and continuing", ErrConflict, task.Id)
}
sessions, err := m.store.ListSessions(ctx, task.Id)
if err == nil {
for _, s := range sessions {
if s.Role == api.Implementer && s.State == api.AgentSessionStateRunning {
err = fmt.Errorf("%w: an agent on this task is mid-turn; merge when they're all idle", ErrConflict)
}
}
}
if err == nil {
m.clearMergeAfterReview(ctx, task.Id)
_, err = m.Send(ctx, sess.Id, "byre", landPrompt(project.DefaultBranch, title), "", "", "")
err = m.store.StartLanding(ctx, task.Id, sess.Id, user.Login)
if errors.Is(err, store.ErrStateChanged) {
return api.Task{}, fmt.Errorf("%w: %w", ErrConflict, err)
}
if err != nil {
m.endLanding(task.Id, sess.Id)
return api.Task{}, err
}
m.clearMergeAfterReview(ctx, task.Id)
_, err = m.launch(ctx, sess, task, "byre", landPrompt(project.DefaultBranch, title), "", "", nil)
if err != nil {
return api.Task{}, err
}
@@ -220,53 +211,9 @@ func (m *Manager) MergeToMain(ctx context.Context, task api.Task, project api.Pr
return m.store.GetTask(ctx, task.Id)
}
type landing struct {
session string
login string
}
func (m *Manager) heldForLanding(taskID, sessionID string) bool {
m.mu.Lock()
defer m.mu.Unlock()
holder := m.landing[taskID].session
return holder != "" && holder != sessionID
}
func (m *Manager) landingBy(taskID, sessionID string) string {
m.mu.Lock()
defer m.mu.Unlock()
if m.landing[taskID].session != sessionID {
return ""
}
return m.landing[taskID].login
}
func (m *Manager) endLanding(taskID, sessionID string) {
m.mu.Lock()
held := m.landing[taskID].session == sessionID
if held {
delete(m.landing, taskID)
}
m.mu.Unlock()
if held {
go m.releaseTeam(taskID, sessionID)
}
}
func (m *Manager) releaseTeam(taskID, landingID string) {
ctx := context.Background()
sessions, err := m.store.ListSessions(ctx, taskID)
func (m *Manager) deliverQueuedTurn(sessionID string) {
err := m.Deliver(context.Background(), sessionID)
if err != nil {
slog.Error("list sessions after merge and continue", "task", taskID, "err", err)
return
}
for _, s := range sessions {
if s.Id == landingID || s.Role != api.Implementer || s.State != api.AgentSessionStateIdle || m.interrupted(s.Id) {
continue
}
err = m.Deliver(ctx, s.Id)
if err != nil {
slog.Error("deliver after merge and continue", "session", s.Id, "err", err)
}
slog.Error("deliver queued message", "session", sessionID, "err", err)
}
}
internal/agents/queue.go
+5 -20
View File
@@ -58,7 +58,11 @@ func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model, se
return api.AgentSession{}, false, err
}
}
held := sess.State == api.AgentSessionStateIdle && sess.Role == api.Implementer && (automatedActor(actor) && m.interrupted(sess.Id) || m.heldForLanding(sess.TaskId, sess.Id))
landing, err := m.store.HeldForLanding(ctx, sess.TaskId, sess.Id)
if err != nil {
return api.AgentSession{}, false, err
}
held := sess.State == api.AgentSessionStateIdle && sess.Role == api.Implementer && (automatedActor(actor) && m.interrupted(sess.Id) || landing)
if !held && (sess.State != api.AgentSessionStateRunning || sess.Role != api.Implementer) {
sess, err = m.Send(ctx, sessionID, actor, text, model, sendID, client, images...)
return sess, false, err
@@ -165,25 +169,6 @@ func (m *Manager) deliverQueued(ctx context.Context, sess api.AgentSession) (boo
return true, m.store.DeleteTaskInputs(ctx, []int64{in.ID})
}
func (m *Manager) deliverAllQueued(ctx context.Context) {
idle, err := m.store.SessionsInState(ctx, api.AgentSessionStateIdle)
if err != nil {
slog.Error("list idle sessions", "err", err)
return
}
for _, sess := range idle {
if sess.Role != api.Implementer {
continue
}
unlock := m.lockDelivery(sess.Id)
_, err = m.deliverQueued(ctx, sess)
unlock()
if err != nil {
slog.Error("deliver queued message", "session", sess.Id, "err", err)
}
}
}
type queued struct {
actor string
text string
internal/agents/tasks.go
+1 -1
View File
@@ -29,7 +29,7 @@ func (m *Manager) StartTask(ctx context.Context, project api.Project, creator ap
}
sess, err := m.newSession(ctx, task, creator, "", model)
if err == nil {
err = m.startRunning(ctx, sess, task.Id, creator.Login)
err = m.startRunning(ctx, sess, creator.Login)
}
if err != nil {
return api.Task{}, fmt.Errorf("task %s created, but its agent did not start: %w", task.Id, err)
internal/store/coordinate.go
+14
View File
@@ -17,6 +17,8 @@ type Work struct {
SettleReviews []SnapshotRef
ReadyPrompts []ReadyPrompt
AdvanceStages []string
LeadDeliveries []LeadDelivery
QueuedDelivery []QueuedDelivery
}
func (s *Store) Coordinate(ctx context.Context) (Work, error) {
@@ -74,6 +76,18 @@ func (s *Store) Coordinate(ctx context.Context) (Work, error) {
return err
}
work.AdvanceStages, err = finishedStages(ctx, tx)
if err != nil {
return err
}
work.LeadDeliveries, err = leadDeliveries(ctx, tx)
if err != nil {
return err
}
err = endLandings(ctx, tx)
if err != nil {
return err
}
work.QueuedDelivery, err = queuedDeliveries(ctx, tx)
return err
})
return work, err
internal/store/landing.go
+114
View File
@@ -0,0 +1,114 @@
package store
import (
"context"
"database/sql"
"errors"
"fmt"
"g.fc.run/ai/byre/internal/api"
)
func (s *Store) StartLanding(ctx context.Context, taskID, sessionID, login string) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
landing := ""
running := false
err := tx.QueryRowContext(ctx, `SELECT landing_session, `+implementerRunning+` FROM tasks WHERE id = ?`, taskID).Scan(&landing, &running)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
if err != nil {
return err
}
switch {
case landing != "":
return fmt.Errorf("%w: task %s is already merging and continuing", ErrStateChanged, taskID)
case running:
return fmt.Errorf("%w: an agent on task %s is mid-turn; merge when they're all idle", ErrStateChanged, taskID)
}
idle := api.AgentSessionStateIdle
err = transitionSessionTx(ctx, tx, emit, sessionID, &idle, api.AgentSessionStateRunning, "byre", nil)
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `UPDATE tasks SET landing_session = ?, landing_by = ? WHERE id = ?`, sessionID, login, taskID)
return err
})
}
func (s *Store) LandingBy(ctx context.Context, taskID, sessionID string) (string, error) {
by := ""
err := s.db.QueryRowContext(ctx, `SELECT landing_by FROM tasks WHERE id = ? AND landing_session = ?`, taskID, sessionID).Scan(&by)
if errors.Is(err, sql.ErrNoRows) {
return "", nil
}
return by, err
}
func (s *Store) HeldForLanding(ctx context.Context, taskID, sessionID string) (bool, error) {
held := false
err := s.db.QueryRowContext(ctx, `SELECT landing_session != '' AND landing_session != ? FROM tasks WHERE id = ?`, sessionID, taskID).Scan(&held)
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
return held, err
}
func endLandings(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, `UPDATE tasks SET landing_session = '', landing_by = ''
WHERE landing_session != '' AND NOT EXISTS (SELECT 1 FROM agent_sessions s WHERE s.id = tasks.landing_session AND s.state = 'running')`)
return err
}
type QueuedDelivery struct {
SessionID string
Input int64
Response string
}
func queuedDeliveries(ctx context.Context, tx *sql.Tx) ([]QueuedDelivery, error) {
rows, err := tx.QueryContext(ctx, `
SELECT s.id, s.task_id,
coalesce((SELECT min(q.id) FROM task_inputs q WHERE q.task_id = s.task_id AND q.member = s.member), 0),
coalesce((SELECT r.id FROM inbox_responses r JOIN inbox_items i ON i.id = r.item_id
WHERE i.agent_session_id = s.id AND r.delivered_at IS NULL ORDER BY r.created_at LIMIT 1), '')
FROM agent_sessions s JOIN tasks t ON t.id = s.task_id
WHERE s.role = 'implementer' AND s.state = 'idle'
AND t.status = 'active' AND t.discard_state = '' AND t.merging_by = '' AND t.finalizing = '' AND t.landing_session = ''
AND (EXISTS (SELECT 1 FROM task_inputs q WHERE q.task_id = s.task_id AND q.member = s.member)
OR EXISTS (SELECT 1 FROM inbox_responses r JOIN inbox_items i ON i.id = r.item_id WHERE i.agent_session_id = s.id AND r.delivered_at IS NULL))
AND s.id = (SELECT l.id FROM agent_sessions l WHERE l.task_id = s.task_id AND l.member = s.member AND l.role = 'implementer' AND l.state != 'completed' ORDER BY l.created_at DESC LIMIT 1)`)
if err != nil {
return nil, err
}
type candidate struct {
QueuedDelivery
taskID string
}
found := []candidate{}
for rows.Next() {
c := candidate{}
err = rows.Scan(&c.SessionID, &c.taskID, &c.Input, &c.Response)
if err != nil {
rows.Close()
return nil, err
}
found = append(found, c)
}
err = errors.Join(rows.Err(), rows.Close())
if err != nil {
return nil, err
}
deliveries := []QueuedDelivery{}
for _, c := range found {
stopped, err := sessionStopped(ctx, tx, c.taskID, c.SessionID)
if err != nil {
return nil, err
}
if !stopped {
deliveries = append(deliveries, c.QueuedDelivery)
}
}
return deliveries, nil
}
internal/store/lead.go
+28
View File
@@ -89,6 +89,34 @@ func (s *Store) RestoreLeadInputs(ctx context.Context, inputs []LeadInput) error
})
}
type LeadDelivery struct {
Owner string
First int64
Last int64
}
func leadDeliveries(ctx context.Context, tx *sql.Tx) ([]LeadDelivery, error) {
rows, err := tx.QueryContext(ctx, `
SELECT i.owner, min(i.id), max(i.id) FROM lead_inputs i
WHERE (SELECT l.state FROM agent_sessions l WHERE l.owner = i.owner AND l.state != 'completed' ORDER BY l.created_at DESC LIMIT 1) = 'idle'
GROUP BY i.owner`)
if err != nil {
return nil, err
}
defer rows.Close()
found := []LeadDelivery{}
for rows.Next() {
d := LeadDelivery{}
err = rows.Scan(&d.Owner, &d.First, &d.Last)
if err != nil {
return nil, err
}
found = append(found, d)
}
return found, rows.Err()
}
func (s *Store) DeleteLeadInputs(ctx context.Context, ids []int64) error {
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
for _, id := range ids {
internal/store/migrations/095-add-task-landing.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE tasks ADD COLUMN landing_session TEXT NOT NULL DEFAULT '';
ALTER TABLE tasks ADD COLUMN landing_by TEXT NOT NULL DEFAULT '';
internal/store/store.go
+7 -2
View File
@@ -1315,9 +1315,11 @@ func transitionSessionTx(ctx context.Context, tx *sql.Tx, emit emitFunc, id stri
role := ""
discard := ""
finalizing := ""
landing := ""
err := tx.QueryRowContext(ctx, `
SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.finalizing, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard, &finalizing)
SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, ''), coalesce(t.finalizing, ''), coalesce(t.landing_session, '')
FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard, &finalizing, &landing)
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
}
@@ -1334,6 +1336,9 @@ func transitionSessionTx(ctx context.Context, tx *sql.Tx, emit emitFunc, id stri
if state == api.AgentSessionStateRunning && role == string(api.Implementer) && finalizing != "" {
return fmt.Errorf("%w: task %s is finishing its merge", ErrStateChanged, taskID)
}
if state == api.AgentSessionStateRunning && role == string(api.Implementer) && landing != "" && landing != id {
return fmt.Errorf("%w: held while task %s merges and continues", ErrStateChanged, taskID)
}
_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(state), id)
if err != nil {