Add task summaries, closed-task search and a summary turn before discard

2026-10-16 10:08:08 -07:00
parent 2107a43ef2
commit fd63a0dfbc
25 changed files with 1013 additions and 268 deletions
@@ -0,0 +1,160 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"g.fc.run/misc/byre/internal/api"
+)
+
+type DiscardState string
+
+const (
+	DiscardNone    DiscardState = ""
+	DiscardHeld    DiscardState = "held"
+	DiscardRunning DiscardState = "running"
+	DiscardClosing DiscardState = "closing"
+)
+
+var (
+	ErrDiscardPending = errors.New("the task is already being discarded")
+	ErrMerging        = errors.New("the task is merging")
+)
+
+type Discard struct {
+	State   DiscardState
+	By      string
+	Session string
+}
+
+func (s *Store) GetDiscard(ctx context.Context, taskID string) (Discard, error) {
+	d := Discard{}
+	err := s.db.QueryRowContext(ctx, `SELECT discard_state, discarding_by, discard_session FROM tasks WHERE id = ?`, taskID).Scan(&d.State, &d.By, &d.Session)
+	if errors.Is(err, sql.ErrNoRows) {
+		return d, ErrNotFound
+	}
+	return d, err
+}
+
+func (s *Store) RequestDiscard(ctx context.Context, taskID, actor string) (DiscardState, error) {
+	state := DiscardNone
+	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		projectID := ""
+		status := ""
+		mergingBy := ""
+		err := tx.QueryRowContext(ctx, `SELECT project_id, status, merging_by, discard_state FROM tasks WHERE id = ?`, taskID).Scan(&projectID, &status, &mergingBy, &state)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
+			return err
+		}
+		switch {
+		case status != string(api.TaskStatusActive):
+			return fmt.Errorf("%w: task %s is %s", ErrNotFound, taskID, status)
+		case state == DiscardClosing:
+			return nil
+		case state != DiscardNone:
+			return ErrDiscardPending
+		case mergingBy != "":
+			return ErrMerging
+		}
+		state = DiscardHeld
+		_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ?, discarding_by = ?, discard_session = '', merge_queued_by = '', merge_queued_at = '' WHERE id = ?`,
+			string(DiscardHeld), actor, taskID)
+		if err != nil {
+			return err
+		}
+		return emit(api.Event{Kind: "task.discard_started", ProjectId: &projectID, TaskId: &taskID, Actor: actor, Payload: map[string]any{"id": taskID}})
+	})
+	return state, err
+}
+
+func (s *Store) StartDiscardTurn(ctx context.Context, taskID, sessionID, actor string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		projectID := ""
+		state := DiscardNone
+		sessState := ""
+		err := tx.QueryRowContext(ctx, `
+			SELECT t.project_id, t.discard_state, s.state FROM tasks t JOIN agent_sessions s ON s.task_id = t.id
+			WHERE t.id = ? AND s.id = ? AND s.role = ?`, taskID, sessionID, string(api.Implementer)).Scan(&projectID, &state, &sessState)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
+			return err
+		}
+		if state != DiscardHeld || sessState != string(api.AgentSessionStateIdle) {
+			return fmt.Errorf("%w: discard is %q and session %s is %s", ErrStateChanged, state, sessionID, sessState)
+		}
+		_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ?, discard_session = ? WHERE id = ?`, string(DiscardRunning), sessionID, taskID)
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(api.AgentSessionStateRunning), sessionID)
+		if err != nil {
+			return err
+		}
+		return emit(sessionEvent("session.state", projectID, taskID, "", sessionID, actor, map[string]any{"state": string(api.AgentSessionStateRunning)}))
+	})
+}
+
+func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload map[string]any) (DiscardState, error) {
+	next := DiscardNone
+	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		taskID := ""
+		projectID := ""
+		owner := ""
+		role := ""
+		state := DiscardNone
+		discardSession := ""
+		err := tx.QueryRowContext(ctx, `
+			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.role, coalesce(t.discard_state, ''), coalesce(t.discard_session, '')
+			FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?`, sessionID).Scan(&taskID, &projectID, &owner, &role, &state, &discardSession)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
+			return err
+		}
+
+		_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(api.AgentSessionStateIdle), sessionID)
+		if err != nil {
+			return err
+		}
+		if payload == nil {
+			payload = map[string]any{}
+		}
+		payload["state"] = string(api.AgentSessionStateIdle)
+		err = emit(sessionEvent("session.state", projectID, taskID, owner, sessionID, actor, payload))
+		if err != nil {
+			return err
+		}
+
+		if role != string(api.Implementer) {
+			return nil
+		}
+		switch {
+		case state == DiscardHeld:
+			next = DiscardHeld
+		case state == DiscardRunning && discardSession == sessionID:
+			next = DiscardClosing
+			_, err = tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ? WHERE id = ?`, string(DiscardClosing), taskID)
+		}
+		return err
+	})
+	return next, err
+}
+
+func (s *Store) CloseDiscard(ctx context.Context, taskID string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		_, err := tx.ExecContext(ctx, `UPDATE tasks SET discard_state = ? WHERE id = ? AND discard_state != ''`, string(DiscardClosing), taskID)
+		return err
+	})
+}
+
+func (s *Store) PendingDiscards(ctx context.Context) ([]api.Task, error) {
+	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE status = ? AND discard_state != ''`, string(api.TaskStatusActive))
+}
@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"

@@ -413,6 +414,13 @@ INSERT INTO users_next (login, display_name, git_name, git_email, created_at, no
 	FROM users;
 DROP TABLE users;
 ALTER TABLE users_next RENAME TO users;
+`}, {"add-task-summary-and-discarding-by", `
+ALTER TABLE tasks ADD COLUMN summary TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN discarding_by TEXT NOT NULL DEFAULT '';
+`}, {"add-task-discard-state", `
+ALTER TABLE tasks ADD COLUMN discard_state TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN discard_session TEXT NOT NULL DEFAULT '';
+UPDATE tasks SET discard_state = 'held' WHERE discarding_by != '' AND status = 'active';
 `}}

 const (
@@ -774,7 +782,7 @@ func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) erro
 func (s *Store) SetMergingBy(ctx context.Context, id, login string) error {
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		projectID := ""
-		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ? RETURNING project_id`, login, id).Scan(&projectID)
+		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ? AND discard_state = '' RETURNING project_id`, login, id).Scan(&projectID)
 		if errors.Is(err, sql.ErrNoRows) {
 			return ErrNotFound
 		}
@@ -797,15 +805,22 @@ func (s *Store) MergingBy(ctx context.Context, id string) (string, error) {
 func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) {
 	login := ""
 	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
-		err := tx.QueryRowContext(ctx, `SELECT merging_by FROM tasks WHERE id = ?`, id).Scan(&login)
+		projectID := ""
+		err := tx.QueryRowContext(ctx, `SELECT project_id, merging_by FROM tasks WHERE id = ?`, id).Scan(&projectID, &login)
 		if errors.Is(err, sql.ErrNoRows) {
 			return ErrNotFound
 		}
 		if err != nil {
 			return err
 		}
+		if login == "" {
+			return nil
+		}
 		_, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = '' WHERE id = ?`, id)
-		return err
+		if err != nil {
+			return err
+		}
+		return emit(api.Event{Kind: "task.merge_ended", ProjectId: &projectID, TaskId: &id, Actor: "byre", Payload: map[string]any{"id": id}})
 	})
 	return login, err
 }
@@ -859,7 +874,7 @@ func (s *Store) queryTasks(ctx context.Context, query string, args ...any) ([]ap
 func (s *Store) QueueMerge(ctx context.Context, id, login string) error {
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		projectID := ""
-		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_queued_by = ?, merge_queued_at = ? WHERE id = ? AND status = ? AND merge_queued_by = '' RETURNING project_id`,
+		err := tx.QueryRowContext(ctx, `UPDATE tasks SET merge_queued_by = ?, merge_queued_at = ? WHERE id = ? AND status = ? AND merge_queued_by = '' AND discard_state = '' RETURNING project_id`,
 			login, now(), id, string(api.TaskStatusActive)).Scan(&projectID)
 		if errors.Is(err, sql.ErrNoRows) {
 			return ErrNotFound
@@ -894,6 +909,70 @@ func (s *Store) TakeQueuedMerge(ctx context.Context, id string) (string, error)
 	return login, err
 }

+func (s *Store) SetTaskSummary(ctx context.Context, id, summary, actor string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		projectID := ""
+		err := tx.QueryRowContext(ctx, `UPDATE tasks SET summary = ? WHERE id = ? AND status = ? RETURNING project_id`,
+			summary, id, string(api.TaskStatusActive)).Scan(&projectID)
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrNotFound
+		}
+		if err != nil {
+			return err
+		}
+		return emit(api.Event{Kind: "task.summary", ProjectId: &projectID, TaskId: &id, Actor: actor, Payload: map[string]any{"id": id, "summary": summary}})
+	})
+}
+
+func (s *Store) TaskSummary(ctx context.Context, id string) (string, error) {
+	summary := ""
+	err := s.db.QueryRowContext(ctx, `SELECT summary FROM tasks WHERE id = ?`, id).Scan(&summary)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", ErrNotFound
+	}
+	return summary, err
+}
+
+type ClosedTask struct {
+	Task    api.Task
+	Project string
+	Summary string
+}
+
+func (s *Store) SearchClosedTasks(ctx context.Context, projectID, query string, limit int) ([]ClosedTask, error) {
+	pattern := "%" + strings.NewReplacer(`\`, `\\`, "%", `\%`, "_", `\_`).Replace(query) + "%"
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+taskColumns+`, (SELECT name FROM projects WHERE projects.id = tasks.project_id), summary FROM tasks
+		WHERE status != ? AND (? = '' OR project_id = ?)
+			AND (title LIKE ? ESCAPE '\' OR merge_title LIKE ? ESCAPE '\' OR todo_item LIKE ? ESCAPE '\' OR summary LIKE ? ESCAPE '\' OR merge_commit LIKE ? ESCAPE '\')
+		ORDER BY closed_at DESC LIMIT ?`,
+		string(api.TaskStatusActive), projectID, projectID, pattern, pattern, pattern, pattern, pattern, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	found := []ClosedTask{}
+	for rows.Next() {
+		c := ClosedTask{}
+		c.Task, err = scanTask(trailingScanner{rows, []any{&c.Project, &c.Summary}})
+		if err != nil {
+			return nil, err
+		}
+		found = append(found, c)
+	}
+	return found, rows.Err()
+}
+
+type trailingScanner struct {
+	row   scanner
+	extra []any
+}
+
+func (t trailingScanner) Scan(dest ...any) error {
+	return t.row.Scan(append(dest, t.extra...)...)
+}
+
 func (s *Store) MergeTask(ctx context.Context, id, commit, pushed, actor string) (api.Task, error) {
 	return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "pushed": pushed})
 }
@@ -1190,9 +1269,11 @@ func (s *Store) transitionSession(ctx context.Context, id string, from *api.Agen
 		projectID := ""
 		owner := ""
 		current := ""
+		role := ""
+		discard := ""
 		err := tx.QueryRowContext(ctx, `
-			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
-		`, id).Scan(&taskID, &projectID, &owner, &current)
+			SELECT coalesce(t.id, ''), coalesce(t.project_id, ''), s.owner, s.state, s.role, coalesce(t.discard_state, '') FROM agent_sessions s LEFT JOIN tasks t ON t.id = s.task_id WHERE s.id = ?
+		`, id).Scan(&taskID, &projectID, &owner, &current, &role, &discard)
 		if errors.Is(err, sql.ErrNoRows) {
 			return ErrNotFound
 		}
@@ -1203,6 +1284,9 @@ func (s *Store) transitionSession(ctx context.Context, id string, from *api.Agen
 		if from != nil && current != string(*from) {
 			return fmt.Errorf("%w: session %s is %s, not %s", ErrStateChanged, id, current, *from)
 		}
+		if state == api.AgentSessionStateRunning && role == string(api.Implementer) && discard != "" {
+			return fmt.Errorf("%w: task %s is being discarded", ErrStateChanged, taskID)
+		}

 		_, err = tx.ExecContext(ctx, `UPDATE agent_sessions SET state = ? WHERE id = ?`, string(state), id)
 		if err != nil {
@@ -1352,7 +1436,7 @@ func (s *Store) write(ctx context.Context, fn func(tx *sql.Tx, emit emitFunc) er

 const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at`

-const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_run, workflow_stage, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running')`
+const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_run, workflow_stage, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running'), merging_by != '', discard_state != '' AND status = 'active'`

 const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at`

@@ -1384,7 +1468,7 @@ func scanTask(row scanner) (api.Task, error) {
 	workflowName := ""
 	workflowRun := ""
 	workflowStage := ""
-	err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowRun, &workflowStage, &t.Reviewing)
+	err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowRun, &workflowStage, &t.Reviewing, &t.Merging, &t.Discarding)
 	if err != nil {
 		return t, err
 	}
@@ -135,7 +135,7 @@ func (s *Store) SetStageOutput(ctx context.Context, sessionID, output string) er
 func (s *Store) FinishStage(ctx context.Context, sessionID string) (bool, error) {
 	finished := false
 	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
-		res, err := tx.ExecContext(ctx, `UPDATE agent_sessions SET stage_finished = 1 WHERE id = ? AND stage_output != ''`, sessionID)
+		res, err := tx.ExecContext(ctx, `UPDATE agent_sessions SET stage_finished = 1 WHERE id = ? AND stage_output != '' AND NOT EXISTS (SELECT 1 FROM tasks WHERE tasks.id = agent_sessions.task_id AND tasks.discard_state != '')`, sessionID)
 		if err != nil {
 			return err
 		}