Make task state store-owned and signal its changes to the lead

2026-10-16 11:15:32 -07:00
parent 34db128afe
commit d26c81ceb1
15 changed files with 451 additions and 199 deletions
@@ -114,7 +114,7 @@ func (s *Store) StartDiscardTurn(ctx context.Context, taskID, sessionID, actor s
 	})
 }

-func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload map[string]any) (DiscardState, error) {
+func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload map[string]any, turn *api.Event, reviewRequest bool) (DiscardState, error) {
 	next := DiscardNone
 	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		taskID := ""
@@ -133,6 +133,18 @@ func (s *Store) EndTurn(ctx context.Context, sessionID, actor string, payload ma
 			return err
 		}

+		if turn != nil {
+			err = emit(*turn)
+			if err != nil {
+				return err
+			}
+		}
+		if reviewRequest {
+			_, err = tx.ExecContext(ctx, `UPDATE tasks SET review_requests = review_requests + 1 WHERE id = ?`, taskID)
+			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
@@ -446,6 +446,10 @@ CREATE TABLE summary_reads (
 	seq INTEGER NOT NULL,
 	PRIMARY KEY (login, task_id)
 );
+`}, {"add-task-state", `
+ALTER TABLE tasks ADD COLUMN state TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN review_requests INTEGER NOT NULL DEFAULT 0;
+UPDATE tasks SET state = ` + taskStateExpr + `;
 `}}

 const (
@@ -459,10 +463,10 @@ type Store struct {
 	path      string
 	portFirst int
 	writeMu   sync.Mutex
-	onEvent   func(api.Event)
+	onEvent   func([]api.Event)
 }

-func Open(path string, portFirst int, onEvent func(api.Event)) (*Store, error) {
+func Open(path string, portFirst int, onEvent func([]api.Event)) (*Store, error) {
 	dsn := "file:" + path + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_txlock=immediate"
 	db, err := sql.Open("sqlite", dsn)
 	if err != nil {
@@ -1253,7 +1257,7 @@ func (s *Store) Broadcast(ev api.Event) {
 	s.writeMu.Lock()
 	defer s.writeMu.Unlock()
 	ev.CreatedAt = time.Now().UTC()
-	s.onEvent(ev)
+	s.onEvent([]api.Event{ev})
 }

 func (s *Store) CreateSession(ctx context.Context, sess api.AgentSession, projectID string) error {
@@ -1521,20 +1525,125 @@ func (s *Store) write(ctx context.Context, fn func(tx *sql.Tx, emit emitFunc) er
 		return err
 	}

+	touched := []string{}
+	seen := map[string]bool{}
+	for _, ev := range events {
+		if ev.TaskId == nil || seen[*ev.TaskId] {
+			continue
+		}
+		seen[*ev.TaskId] = true
+		touched = append(touched, *ev.TaskId)
+	}
+	for _, taskID := range touched {
+		actor := "byre"
+		for _, ev := range events {
+			if ev.TaskId != nil && *ev.TaskId == taskID {
+				actor = ev.Actor
+			}
+		}
+		err = settleTaskState(ctx, tx, emit, taskID, actor)
+		if err != nil {
+			return err
+		}
+	}
+
 	err = tx.Commit()
 	if err != nil {
 		return err
 	}

-	for _, ev := range events {
-		s.onEvent(ev)
-	}
+	s.onEvent(events)
 	return nil
 }

+func settleTaskState(ctx context.Context, tx *sql.Tx, emit emitFunc, taskID, actor string) error {
+	projectID := ""
+	before := ""
+	now := ""
+	err := tx.QueryRowContext(ctx, `SELECT project_id, state, `+taskStateExpr+` FROM tasks WHERE id = ?`, taskID).Scan(&projectID, &before, &now)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	if before == now {
+		return nil
+	}
+	_, err = tx.ExecContext(ctx, `UPDATE tasks SET state = ? WHERE id = ?`, now, taskID)
+	if err != nil {
+		return err
+	}
+	return emit(api.Event{
+		Kind:      "task.state",
+		ProjectId: &projectID,
+		TaskId:    &taskID,
+		Actor:     actor,
+		Payload:   map[string]any{"from": before, "to": now},
+	})
+}
+
+const taskStateExpr = `CASE
+	WHEN tasks.status != 'active' THEN tasks.status
+	WHEN tasks.discard_state != '' THEN 'discarding'
+	WHEN tasks.merging_by != '' THEN 'merging'
+	ELSE coalesce((
+		SELECT CASE
+			WHEN s.state != 'idle' THEN s.state
+			WHEN EXISTS (SELECT 1 FROM reviews r WHERE r.task_id = tasks.id AND r.state = 'running') THEN 'reviewing'
+			WHEN tasks.review_requests > 0 THEN 'review_pending'
+			ELSE 'idle'
+		END
+		FROM agent_sessions s
+		WHERE s.task_id = tasks.id AND s.role != 'reviewer' AND s.state != 'completed'
+		ORDER BY s.created_at DESC LIMIT 1
+	), 'no_agent')
+END`
+
+func (s *Store) EndReviewRequest(ctx context.Context, taskID string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		_, err := tx.ExecContext(ctx, `UPDATE tasks SET review_requests = max(review_requests - 1, 0) WHERE id = ?`, taskID)
+		if err != nil {
+			return err
+		}
+		return settleTaskState(ctx, tx, emit, taskID, "byre")
+	})
+}
+
+func (s *Store) ClearReviewRequests(ctx context.Context) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		rows, err := tx.QueryContext(ctx, `UPDATE tasks SET review_requests = 0 WHERE review_requests > 0 RETURNING id`)
+		if err != nil {
+			return err
+		}
+		ids := []string{}
+		for rows.Next() {
+			id := ""
+			err = rows.Scan(&id)
+			if err != nil {
+				rows.Close()
+				return err
+			}
+			ids = append(ids, id)
+		}
+		rows.Close()
+		err = rows.Err()
+		if err != nil {
+			return err
+		}
+		for _, id := range ids {
+			err = settleTaskState(ctx, tx, emit, id, "byre")
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 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, parent_task_id, parent_merge_by, EXISTS (SELECT 1 FROM reviews WHERE reviews.task_id = tasks.id AND reviews.state = 'running'), merging_by != '', discard_state != '' AND status = 'active', (SELECT coalesce(max(seq), 0) FROM events WHERE events.task_id = tasks.id AND kind = 'turn.ended' AND json_extract(payload, '$.summary') = 1)`
+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, parent_task_id, parent_merge_by, state, merging_by != '', discard_state != '' AND status = 'active', (SELECT coalesce(max(seq), 0) FROM events WHERE events.task_id = tasks.id AND kind = 'turn.ended' AND json_extract(payload, '$.summary') = 1)`

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

@@ -1568,10 +1677,12 @@ func scanTask(row scanner) (api.Task, error) {
 	workflowStage := ""
 	parentTaskID := ""
 	parentMergeBy := ""
-	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, &parentTaskID, &parentMergeBy, &t.Reviewing, &t.Merging, &t.Discarding, &t.SummarySeq)
+	taskState := ""
+	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, &parentTaskID, &parentMergeBy, &taskState, &t.Merging, &t.Discarding, &t.SummarySeq)
 	if err != nil {
 		return t, err
 	}
+	t.State = api.TaskState(taskState)
 	if parentTaskID != "" {
 		t.ParentTaskId = &parentTaskID
 	}