new task lag

internal/agents/manager.go

@@ -380,6 +380,20 @@ func (m *Manager) Recover(ctx context.Context) error {
 			continue
 		}
 
+		pending, err := m.store.SessionPending(ctx, sess.Id)
+		if err != nil {
+			return err
+		}
+		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)
@@ -399,7 +413,7 @@ func (m *Manager) Recover(ctx context.Context) error {
 				notice = discardRestartPrompt
 			}
 		}
-		err = m.message(ctx, l, "byre", notice, nil)
+		err = m.message(ctx, sess, task, l, "byre", notice, nil)
 		if err != nil {
 			ferr := m.failResume(ctx, sess, task, err)
 			if ferr != nil {
@@ -566,28 +580,67 @@ func (m *Manager) startRunning(ctx context.Context, sess api.AgentSession, taskI
 }
 
 func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Task, actor, text string, images []string) (api.AgentSession, error) {
+	imgs, compacting, err := m.recordMessage(ctx, sess, task, actor, text, images)
+	if err != nil {
+		m.failStart(ctx, sess.Id, actor, err)
+		return api.AgentSession{}, err
+	}
+	err = m.deliver(ctx, sess, task, actor, text, imgs, compacting)
+	if err != nil {
+		return api.AgentSession{}, err
+	}
+	return m.store.GetSession(ctx, sess.Id)
+}
+
+func (m *Manager) deliver(ctx context.Context, sess api.AgentSession, task api.Task, actor, text string, imgs []image, compacting bool) error {
 	l, err := m.ensureLive(ctx, sess, task)
 	if err == nil {
 		err = m.syncModel(ctx, l, sess.Model)
 	}
 	if err != nil {
 		m.failStart(ctx, sess.Id, actor, err)
-		return api.AgentSession{}, err
+		return err
 	}
 
-	imgs, err := m.prepareMessage(ctx, l, actor, text, images)
+	m.beginTurn(l, compacting)
-	if err != nil {
+	err = m.send(ctx, l, text, imgs)
-		m.failStart(ctx, sess.Id, actor, err)
-		return api.AgentSession{}, err
-	}
-
-	err = l.sess.send(ctx, text, imgs)
 	if err != nil {
 		m.endMergeTurn(sess.Id, false)
-		return api.AgentSession{}, err
+		return err
+	}
+	return nil
 }
 
-	return m.store.GetSession(ctx, sess.Id)
+func (m *Manager) deliverRecorded(ctx context.Context, sess api.AgentSession, task api.Task) (bool, error) {
+	ev, found, err := m.store.LastMessage(ctx, sess.Id)
+	if err != nil || !found {
+		return false, err
+	}
+	text, _ := ev.Payload["text"].(string)
+	ids := []string{}
+	raw, _ := ev.Payload["images"].([]any)
+	for _, id := range raw {
+		s, _ := id.(string)
+		ids = append(ids, s)
+	}
+	imgs, err := m.readImages(ctx, ids)
+	if err != nil {
+		m.failStart(ctx, sess.Id, ev.Actor, err)
+		return true, err
+	}
+	return true, m.deliver(ctx, sess, task, ev.Actor, text, imgs, isHandoffPrompt(ev.Actor, text))
+}
+
+func (m *Manager) readImages(ctx context.Context, ids []string) ([]image, error) {
+	imgs := []image{}
+	for _, id := range ids {
+		b, data, err := m.store.ReadBlob(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+		imgs = append(imgs, image{mediaType: b.ContentType, data: data})
+	}
+	return imgs, nil
 }
 
 func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err error) {
@@ -625,32 +678,38 @@ func transition(ctx context.Context, st *store.Store, id string, from, to api.Ag
 	return err
 }
 
-func (m *Manager) message(ctx context.Context, l *live, actor, text string, images []string) error {
+func (m *Manager) message(ctx context.Context, sess api.AgentSession, task api.Task, l *live, actor, text string, images []string) error {
-	imgs, err := m.prepareMessage(ctx, l, actor, text, images)
+	imgs, compacting, err := m.recordMessage(ctx, sess, task, actor, text, images)
 	if err != nil {
 		return err
 	}
-	return l.sess.send(ctx, text, imgs)
+	m.beginTurn(l, compacting)
+	return m.send(ctx, l, text, imgs)
 }
 
-func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text string, images []string) ([]image, error) {
+func (m *Manager) send(ctx context.Context, l *live, text string, imgs []image) error {
-	if l.implementer() && isMergeInPrompt(actor, text) {
+	err := l.sess.send(ctx, text, imgs)
-		m.takeMergeTurn(l)
-	}
-	imgs := []image{}
-	for _, id := range images {
-		b, data, err := m.store.ReadBlob(ctx, id)
 	if err != nil {
-			return nil, err
+		return err
 	}
-		imgs = append(imgs, image{mediaType: b.ContentType, data: data})
+	return m.store.MarkSessionDelivered(ctx, l.id)
+}
+
+func (m *Manager) recordMessage(ctx context.Context, sess api.AgentSession, task api.Task, actor, text string, images []string) ([]image, bool, error) {
+	implementer := sess.Role == api.Implementer
+	if implementer && isMergeInPrompt(actor, text) {
+		m.ownMerge(sess.Id, task.Id, task.WorktreePath)
+	}
+	imgs, err := m.readImages(ctx, images)
+	if err != nil {
+		return nil, false, err
 	}
 
 	payload := map[string]any{"text": text}
-	if l.implementer() {
+	if implementer {
-		head, err := git.Head(l.dir)
+		head, err := git.Head(task.WorktreePath)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		payload["head"] = head
 	}
@@ -664,11 +723,14 @@ func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text strin
 	if isContinuation(actor, text) {
 		payload["continuation"] = true
 	}
-	err := m.emit(ctx, l, actor, "user.message", payload)
+	err = m.store.RecordTurnMessage(ctx, sess.Id, newEvent(sess.Id, sess.Owner, task.ProjectId, task.Id, actor, "user.message", payload))
 	if err != nil {
-		return nil, err
+		return nil, false, err
+	}
+	return imgs, compacting, nil
 }
 
+func (m *Manager) beginTurn(l *live, compacting bool) {
 	m.mu.Lock()
 	l.interrupted = false
 	l.asked = false
@@ -678,8 +740,6 @@ func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text strin
 	l.rateLimited = nil
 	l.lastText = ""
 	m.mu.Unlock()
-
-	return imgs, nil
 }
 
 func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) {
@@ -1357,18 +1417,22 @@ func (m *Manager) emit(ctx context.Context, l *live, actor, kind string, payload
 }
 
 func (m *Manager) event(l *live, actor, kind string, payload map[string]any) api.Event {
+	return newEvent(l.id, l.owner, l.projectID, l.taskID, actor, kind, payload)
+}
+
+func newEvent(sessionID, owner, projectID, taskID, actor, kind string, payload map[string]any) api.Event {
 	ev := api.Event{
 		Kind:           kind,
-		AgentSessionId: &l.id,
+		AgentSessionId: &sessionID,
 		Actor:          actor,
 		Payload:        payload,
 	}
-	if l.owner != "" {
+	if owner != "" {
-		ev.Owner = &l.owner
+		ev.Owner = &owner
 		return ev
 	}
-	ev.ProjectId = &l.projectID
+	ev.ProjectId = &projectID
-	ev.TaskId = &l.taskID
+	ev.TaskId = &taskID
 	return ev
 }
 

internal/agents/mergein.go

@@ -189,6 +189,10 @@ func (m *Manager) launchMerge(sess api.AgentSession, task api.Task, project api.
 }
 
 func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task api.Task, project api.Project, prompt string) error {
+	_, compacting, err := m.recordMessage(ctx, sess, task, "byre", prompt, nil)
+	if err != nil {
+		return err
+	}
 	l, err := m.ensureLive(ctx, sess, task)
 	if err != nil {
 		return err
@@ -197,10 +201,7 @@ func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task
 	if err != nil {
 		return err
 	}
-	_, err = m.prepareMessage(ctx, l, "byre", prompt, nil)
+	m.beginTurn(l, compacting)
-	if err != nil {
-		return err
-	}
 	m.mu.Lock()
 	l.fetching = true
 	m.mu.Unlock()
@@ -218,7 +219,7 @@ func (m *Manager) deliverMerge(ctx context.Context, sess api.AgentSession, task
 	if interrupted {
 		return errors.New("interrupted before the merge started")
 	}
-	return l.sess.send(ctx, prompt, nil)
+	return m.send(ctx, l, prompt, nil)
 }
 
 func (m *Manager) recoverMerges(ctx context.Context) error {

internal/agents/tasks.go

@@ -26,10 +26,24 @@ func (m *Manager) StartTask(ctx context.Context, project api.Project, creator ap
 	if err != nil {
 		return api.Task{}, err
 	}
-	_, err = m.create(ctx, task, creator, "", prompt, model, creator.Login, images)
+	sess, err := m.newSession(ctx, task, creator, "", model)
+	if err == nil {
+		err = m.startRunning(ctx, sess, task.Id, creator.Login)
+	}
 	if err != nil {
 		return api.Task{}, fmt.Errorf("task %s created, but its agent did not start: %w", task.Id, err)
 	}
+	imgs, compacting, err := m.recordMessage(ctx, sess, task, creator.Login, prompt, images)
+	if err != nil {
+		m.failStart(ctx, sess.Id, creator.Login, err)
+		return api.Task{}, fmt.Errorf("task %s created, but its agent did not start: %w", task.Id, err)
+	}
+	go func() {
+		err := m.deliver(context.Background(), sess, task, creator.Login, prompt, imgs, compacting)
+		if err != nil {
+			slog.Error("start task agent", "task", task.Id, "err", err)
+		}
+	}()
 	return task, nil
 }
 

internal/store/migrations/067-add-session-started.sql

@@ -0,0 +1,3 @@
+ALTER TABLE agent_sessions ADD COLUMN started INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE agent_sessions ADD COLUMN pending INTEGER NOT NULL DEFAULT 0;
+UPDATE agent_sessions SET started = 1 WHERE id IN (SELECT agent_session_id FROM events WHERE kind = 'user.message' AND agent_session_id IS NOT NULL);

internal/store/store.go

@@ -1006,6 +1006,21 @@ func (s *Store) FirstPrompt(ctx context.Context, taskID string) (string, error)
 	return text, err
 }
 
+func (s *Store) LastMessage(ctx context.Context, sessionID string) (api.Event, bool, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+eventColumns+`
+		FROM events WHERE agent_session_id = ? AND kind = 'user.message' ORDER BY seq DESC LIMIT 1
+	`, sessionID)
+	if err != nil {
+		return api.Event{}, false, err
+	}
+	evs, err := scanEvents(rows)
+	if err != nil || len(evs) == 0 {
+		return api.Event{}, false, err
+	}
+	return evs[0], true, nil
+}
+
 func (s *Store) BoardPosts(ctx context.Context, taskID string, after int64) ([]api.Event, error) {
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT `+eventColumns+`
@@ -1330,10 +1345,37 @@ func (s *Store) SetSubagents(ctx context.Context, id string, n int, actor string
 
 func (s *Store) SessionStarted(ctx context.Context, sessionID string) (bool, error) {
 	started := false
-	err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events WHERE agent_session_id = ? AND kind = 'user.message')`, sessionID).Scan(&started)
+	err := s.db.QueryRowContext(ctx, `SELECT started FROM agent_sessions WHERE id = ?`, sessionID).Scan(&started)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, ErrNotFound
+	}
 	return started, err
 }
 
+func (s *Store) SessionPending(ctx context.Context, sessionID string) (bool, error) {
+	pending := false
+	err := s.db.QueryRowContext(ctx, `SELECT pending FROM agent_sessions WHERE id = ?`, sessionID).Scan(&pending)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, ErrNotFound
+	}
+	return pending, err
+}
+
+func (s *Store) RecordTurnMessage(ctx context.Context, sessionID string, ev api.Event) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		_, err := tx.ExecContext(ctx, `UPDATE agent_sessions SET pending = 1 WHERE id = ?`, sessionID)
+		if err != nil {
+			return err
+		}
+		return emit(ev)
+	})
+}
+
+func (s *Store) MarkSessionDelivered(ctx context.Context, sessionID string) error {
+	_, err := s.db.ExecContext(ctx, `UPDATE agent_sessions SET started = 1, pending = 0 WHERE id = ?`, sessionID)
+	return err
+}
+
 type Turn struct {
 	Actor string
 	At    time.Time