diff --git a/docs/operations.md b/docs/operations.md index c055978..53b7d63 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -86,7 +86,7 @@ Vendor setup and protocol quirks are in [Claude Code](claude-protocol.md), [Code ## Coordination -Coordination of merges, discards, reviews, ready-for-merge prompts and workflow stages must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice. +Coordination of merges, discards, reviews, ready-for-merge prompts, workflow stages, deliveries to lead agents and queued messages, and Merge & continue's hold on teammates must decide from the database's current state, never from which event woke it. This lets startup use the same decisions as ordinary operation rather than maintaining separate recovery logic. Slow work must run outside the decision transaction, with its progress recorded so another pass cannot start it twice. ## git diff --git a/internal/agents/compact.go b/internal/agents/compact.go index 562c49e..c55a538 100644 --- a/internal/agents/compact.go +++ b/internal/agents/compact.go @@ -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) } diff --git a/internal/agents/coordinator.go b/internal/agents/coordinator.go index 7eeefb3..de5fc8c 100644 --- a/internal/agents/coordinator.go +++ b/internal/agents/coordinator.go @@ -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) }) } diff --git a/internal/agents/lead.go b/internal/agents/lead.go index 74e93c5..3841044 100644 --- a/internal/agents/lead.go +++ b/internal/agents/lead.go @@ -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 } diff --git a/internal/agents/leadnotes.go b/internal/agents/leadnotes.go index dadc2ea..4e4d160 100644 --- a/internal/agents/leadnotes.go +++ b/internal/agents/leadnotes.go @@ -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) { diff --git a/internal/agents/leadtools.go b/internal/agents/leadtools.go index 6ff7c17..ecac97b 100644 --- a/internal/agents/leadtools.go +++ b/internal/agents/leadtools.go @@ -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 { diff --git a/internal/agents/manager.go b/internal/agents/manager.go index 5a63a88..bf67e65 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -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) diff --git a/internal/agents/mergefacts.go b/internal/agents/mergefacts.go index c82ef2f..4f34f02 100644 --- a/internal/agents/mergefacts.go +++ b/internal/agents/mergefacts.go @@ -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) diff --git a/internal/agents/mergemain.go b/internal/agents/mergemain.go index 4cd3826..8f95001 100644 --- a/internal/agents/mergemain.go +++ b/internal/agents/mergemain.go @@ -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) } } diff --git a/internal/agents/queue.go b/internal/agents/queue.go index aeb16d5..9dbeeed 100644 --- a/internal/agents/queue.go +++ b/internal/agents/queue.go @@ -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 diff --git a/internal/agents/tasks.go b/internal/agents/tasks.go index 4035525..2d0c449 100644 --- a/internal/agents/tasks.go +++ b/internal/agents/tasks.go @@ -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) diff --git a/internal/store/coordinate.go b/internal/store/coordinate.go index 31fcaef..3a83718 100644 --- a/internal/store/coordinate.go +++ b/internal/store/coordinate.go @@ -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 diff --git a/internal/store/landing.go b/internal/store/landing.go new file mode 100644 index 0000000..4aed096 --- /dev/null +++ b/internal/store/landing.go @@ -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 +} diff --git a/internal/store/lead.go b/internal/store/lead.go index ef64122..44e93e5 100644 --- a/internal/store/lead.go +++ b/internal/store/lead.go @@ -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 { diff --git a/internal/store/migrations/095-add-task-landing.sql b/internal/store/migrations/095-add-task-landing.sql new file mode 100644 index 0000000..5341056 --- /dev/null +++ b/internal/store/migrations/095-add-task-landing.sql @@ -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 ''; diff --git a/internal/store/store.go b/internal/store/store.go index 6d229ad..e9a6dcc 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -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, ¤t, &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, ¤t, &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 {