Add a pause that holds an agent after its current step
This commit is contained in:
+84
-62
@@ -62,6 +62,7 @@ type Manager struct {
|
||||
startHolds map[string]showHold
|
||||
handoffModel map[string]string
|
||||
mergeTurns map[string]mergeTurn
|
||||
paused map[string]chan struct{}
|
||||
|
||||
deliverMu sync.Mutex
|
||||
deliverLocks map[string]*sessionLock
|
||||
@@ -165,6 +166,7 @@ func New(st *store.Store, worktreeRoot, leadRoot, handsRoot, codexHome, antigrav
|
||||
startHolds: map[string]showHold{},
|
||||
handoffModel: map[string]string{},
|
||||
mergeTurns: map[string]mergeTurn{},
|
||||
paused: map[string]chan struct{}{},
|
||||
signals: make(chan []api.Event, 1024),
|
||||
snaps: map[string]*snapState{},
|
||||
pokes: make(chan struct{}, 1),
|
||||
@@ -326,7 +328,9 @@ func (m *Manager) StopTask(ctx context.Context, taskID, actor string) error {
|
||||
if sess.State == api.AgentSessionStateCompleted {
|
||||
continue
|
||||
}
|
||||
unlock := m.lockDelivery(sess.Id)
|
||||
err = m.retire(ctx, sess, actor)
|
||||
unlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -337,7 +341,9 @@ func (m *Manager) StopTask(ctx context.Context, taskID, actor string) error {
|
||||
|
||||
func (m *Manager) retire(ctx context.Context, sess api.AgentSession, actor string) error {
|
||||
m.stopLive(sess.Id)
|
||||
return m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateCompleted, actor, nil)
|
||||
err := m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateCompleted, actor, nil)
|
||||
m.dropPause(sess.Id)
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *Manager) stopLive(sessionID string) {
|
||||
@@ -357,12 +363,79 @@ func (m *Manager) stopLive(sessionID string) {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) recoverSession(ctx context.Context, sess api.AgentSession) error {
|
||||
task, err := m.taskFor(ctx, sess)
|
||||
if err != nil {
|
||||
slog.Error("resume session: load task", "session", sess.Id, "task", sess.TaskId, "err", err)
|
||||
return nil
|
||||
}
|
||||
|
||||
pending, err := m.store.SessionPending(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
started, err := m.store.SessionStarted(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
finished, err := m.store.StageFinished(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if finished || !pending && !started {
|
||||
m.endMergeTurn(sess.Id, true)
|
||||
err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
slog.Info("session left idle at startup: its stage finished or it never got its first message", "session", sess.Id, "task", task.Id)
|
||||
return nil
|
||||
}
|
||||
if pending {
|
||||
found, err := m.deliverRecorded(ctx, sess, task)
|
||||
if err != nil {
|
||||
slog.Error("resume session: deliver its recorded message", "session", sess.Id, "task", task.Id, "err", err)
|
||||
}
|
||||
if found {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
l, err := m.ensureLive(ctx, sess, task)
|
||||
if err != nil {
|
||||
return m.failResume(ctx, sess, task, err)
|
||||
}
|
||||
|
||||
notice := restartNotice
|
||||
if sess.Role == api.Implementer {
|
||||
discard, err := m.store.GetDiscard(ctx, task.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if discard.State == store.DiscardRunning && discard.Session == sess.Id {
|
||||
notice = discardRestartPrompt
|
||||
}
|
||||
}
|
||||
err = m.message(ctx, sess, task, l, "byre", notice, nil)
|
||||
if err != nil {
|
||||
return m.failResume(ctx, sess, task, err)
|
||||
}
|
||||
slog.Info("resumed session", "session", sess.Id, "task", task.Id, "owner", sess.Owner)
|
||||
m.logUndelivered(ctx, sess.Id, "resumed with restart notice")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) Recover(ctx context.Context) error {
|
||||
err := m.store.ClearMainPushes(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = m.loadPauses(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = m.recoverPreviews(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -390,72 +463,18 @@ func (m *Manager) Recover(ctx context.Context) error {
|
||||
}
|
||||
|
||||
for _, sess := range sessions {
|
||||
task, err := m.taskFor(ctx, sess)
|
||||
if err != nil {
|
||||
slog.Error("resume session: load task", "session", sess.Id, "task", sess.TaskId, "err", err)
|
||||
m.mu.Lock()
|
||||
resumed := m.paused[sess.Id]
|
||||
m.mu.Unlock()
|
||||
if resumed != nil {
|
||||
slog.Info("session resume held: paused", "session", sess.Id, "task", sess.TaskId)
|
||||
go m.recoverAfterPause(sess.Id)
|
||||
continue
|
||||
}
|
||||
|
||||
pending, err := m.store.SessionPending(ctx, sess.Id)
|
||||
err = m.recoverSession(ctx, sess)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
started, err := m.store.SessionStarted(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
finished, err := m.store.StageFinished(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if finished || !pending && !started {
|
||||
m.endMergeTurn(sess.Id, true)
|
||||
err = m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, "byre", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
slog.Info("session left idle at startup: its stage finished or it never got its first message", "session", sess.Id, "task", task.Id)
|
||||
continue
|
||||
}
|
||||
if pending {
|
||||
found, err := m.deliverRecorded(ctx, sess, task)
|
||||
if err != nil {
|
||||
slog.Error("resume session: deliver its recorded message", "session", sess.Id, "task", task.Id, "err", err)
|
||||
}
|
||||
if found {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
l, err := m.ensureLive(ctx, sess, task)
|
||||
if err != nil {
|
||||
ferr := m.failResume(ctx, sess, task, err)
|
||||
if ferr != nil {
|
||||
return ferr
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
notice := restartNotice
|
||||
if sess.Role == api.Implementer {
|
||||
discard, err := m.store.GetDiscard(ctx, task.Id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if discard.State == store.DiscardRunning && discard.Session == sess.Id {
|
||||
notice = discardRestartPrompt
|
||||
}
|
||||
}
|
||||
err = m.message(ctx, sess, task, l, "byre", notice, nil)
|
||||
if err != nil {
|
||||
ferr := m.failResume(ctx, sess, task, err)
|
||||
if ferr != nil {
|
||||
return ferr
|
||||
}
|
||||
continue
|
||||
}
|
||||
slog.Info("resumed session", "session", sess.Id, "task", task.Id, "owner", sess.Owner)
|
||||
m.logUndelivered(ctx, sess.Id, "resumed with restart notice")
|
||||
}
|
||||
|
||||
idle, err := m.store.SessionsInState(ctx, api.AgentSessionStateIdle)
|
||||
@@ -576,6 +595,9 @@ 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) {
|
||||
if m.isPaused(sess.Id) {
|
||||
return api.AgentSession{}, fmt.Errorf("%w: session %s is paused; resume it first", ErrConflict, sess.Id)
|
||||
}
|
||||
err := m.startRunning(ctx, sess, actor)
|
||||
if err != nil {
|
||||
return api.AgentSession{}, err
|
||||
|
||||
Reference in New Issue
Block a user