Retry failed sends with their send_id until the server confirms them

internal/agents/lead.go

@@ -127,6 +127,16 @@ func (m *Manager) MessageLead(ctx context.Context, user api.User, text, model, s
 	if err != nil {
 		return api.AgentSession{}, err
 	}
+	prior, sent, err := m.priorSend(ctx, sendID)
+	if err != nil {
+		return api.AgentSession{}, err
+	}
+	if sent && prior.Id == "" {
+		return sess, nil
+	}
+	if sent {
+		return prior, nil
+	}
 	if !found {
 		return m.createLead(ctx, user, text, model, user.Login, sendID, images)
 	}

internal/agents/manager.go

@@ -186,6 +186,22 @@ 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, sendID string, images ...string) (api.AgentSession, error) {
+	unlock := m.lockDelivery("create:" + task.Id)
+	defer unlock()
+
+	ctx = context.WithoutCancel(ctx)
+	prior, sent, err := m.priorSend(ctx, sendID)
+	if err != nil || sent {
+		return prior, err
+	}
+	unused, found, err := m.store.UnusedSession(ctx, task.Id, "")
+	if err != nil {
+		return api.AgentSession{}, err
+	}
+	if found {
+		slog.Info("first message to an unused session", "session", unused, "task", task.Id)
+		return m.Send(ctx, unused, creator.Login, prompt, model, sendID, images...)
+	}
 	return m.create(ctx, task, creator, "", prompt, model, creator.Login, sendID, images)
 }
 

internal/agents/queue.go

@@ -43,6 +43,11 @@ func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model, se
 	unlock := m.lockDelivery(sessionID)
 	defer unlock()
 
+	prior, sent, err := m.priorSend(ctx, sendID)
+	if err != nil || sent {
+		return prior, false, err
+	}
+
 	sess, err := m.store.GetSession(ctx, sessionID)
 	if err != nil {
 		return api.AgentSession{}, false, err
@@ -83,6 +88,22 @@ func (m *Manager) Message(ctx context.Context, sessionID, actor, text, model, se
 	return sess, true, nil
 }
 
+func (m *Manager) priorSend(ctx context.Context, sendID string) (api.AgentSession, bool, error) {
+	if sendID == "" {
+		return api.AgentSession{}, false, nil
+	}
+	sessionID, found, err := m.store.SendSession(ctx, sendID)
+	if err != nil || !found || sessionID == "" {
+		return api.AgentSession{}, found, err
+	}
+	sess, err := m.store.GetSession(ctx, sessionID)
+	if err != nil {
+		return api.AgentSession{}, false, err
+	}
+	slog.Info("send already accepted", "session", sess.Id, "send", sendID)
+	return sess, true, nil
+}
+
 func automatedActor(actor string) bool {
 	return actor == "byre" || actor == agentActor || actor == boardActor
 }

internal/store/migrations/070-index-event-send-ids.sql

@@ -0,0 +1 @@
+CREATE INDEX events_send_id ON events(json_extract(payload, '$.send_id')) WHERE kind IN ('user.queued', 'user.message');

internal/store/store.go

@@ -1356,6 +1356,16 @@ func (s *Store) SessionStarted(ctx context.Context, sessionID string) (bool, err
 	return started, err
 }
 
+func (s *Store) UnusedSession(ctx context.Context, taskID, member string) (string, bool, error) {
+	id := ""
+	err := s.db.QueryRowContext(ctx, `SELECT id FROM agent_sessions WHERE task_id = ? AND member = ? AND role = ? AND state = ? AND started = 0 AND pending = 0 LIMIT 1`,
+		taskID, member, string(api.Implementer), string(api.AgentSessionStateIdle)).Scan(&id)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", false, nil
+	}
+	return id, err == nil, 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)

internal/store/taskinputs.go

@@ -45,6 +45,22 @@ func (s *Store) AddTaskInput(ctx context.Context, in TaskInput, projectID, sessi
 	})
 }
 
+func (s *Store) SendSession(ctx context.Context, sendID string) (string, bool, error) {
+	sessionID := sql.NullString{}
+	err := s.db.QueryRowContext(ctx, `SELECT agent_session_id FROM events WHERE kind IN ('user.queued', 'user.message') AND json_extract(payload, '$.send_id') = ? LIMIT 1`, sendID).Scan(&sessionID)
+	if err == nil {
+		return sessionID.String, true, nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return "", false, err
+	}
+	err = s.db.QueryRowContext(ctx, `SELECT 1 FROM lead_inputs WHERE send_id = ? LIMIT 1`, sendID).Scan(new(int))
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", false, nil
+	}
+	return "", err == nil, err
+}
+
 func (s *Store) WithdrawTaskInput(ctx context.Context, taskID string, id int64, projectID, actor string) (string, error) {
 	sessionID := ""
 	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {