fix: atomic execution creation + RUNNING state transition

Add CreateExecutionAndSetRunning to storage.DB and Store interface,
replacing the two sequential CreateExecution/UpdateTaskState calls in
executor.go. Eliminates the crash window where a task stays PENDING
with an orphaned RUNNING execution record.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

1 files changed, 49 insertions, 0 deletions

diff --git a/internal/storage/db.go b/internal/storage/db.go
index 37d1ada..3a3e6b2 100644
--- a/internal/storage/db.go
+++ b/internal/storage/db.go
@@ -463,6 +463,55 @@ type Execution struct {
 	Summary string
 }
 
+// CreateExecutionAndSetRunning inserts an execution record and transitions the
+// task to RUNNING in a single transaction, preventing a crash-window where the
+// task stays PENDING with an orphaned RUNNING execution record.
+func (s *DB) CreateExecutionAndSetRunning(e *Execution) error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck
+
+	// Validate state transition.
+	var currentState string
+	if err := tx.QueryRow(`SELECT state FROM tasks WHERE id = ?`, e.TaskID).Scan(&currentState); err != nil {
+		if err == sql.ErrNoRows {
+			return fmt.Errorf("task %q not found", e.TaskID)
+		}
+		return err
+	}
+	if !task.ValidTransition(task.State(currentState), task.StateRunning) {
+		return fmt.Errorf("invalid state transition %s → RUNNING for task %q", currentState, e.TaskID)
+	}
+
+	// Insert execution record.
+	commitsJSON := "[]"
+	if len(e.Commits) > 0 {
+		b, err := json.Marshal(e.Commits)
+		if err != nil {
+			return fmt.Errorf("marshaling commits: %w", err)
+		}
+		commitsJSON = string(b)
+	}
+	if _, err := tx.Exec(`
+		INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)`,
+		e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status,
+		e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON,
+	); err != nil {
+		return err
+	}
+
+	// Transition task to RUNNING.
+	now := time.Now().UTC()
+	if _, err := tx.Exec(`UPDATE tasks SET state = ?, updated_at = ? WHERE id = ?`, string(task.StateRunning), now, e.TaskID); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
 // CreateExecution inserts an execution record.
 func (s *DB) CreateExecution(e *Execution) error {
 	var changestatsJSON *string