byre/internal/store/landing.go

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.merge_state NOT IN ('claimed', 'turn', '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
}