fix: reuse prepared session on explicit start

apps/backend/internal/orchestrator/session_ensure.go

@@ -45,6 +45,23 @@ func acquireEnsureLock(taskID string) func() {
 	return mu.Unlock
 }
 
+// ensureLockHeldKey marks a context whose caller already owns the per-task
+// ensureLocks mutex. launchStart consults it to skip a recursive acquire when
+// EnsureSession re-enters via LaunchSession — sync.Mutex isn't reentrant, so
+// without this the second Lock would deadlock the request.
+type ensureLockHeldCtxKey struct{}
+
+var ensureLockHeldKey = ensureLockHeldCtxKey{}
+
+func withEnsureLockHeld(ctx context.Context) context.Context {
+	return context.WithValue(ctx, ensureLockHeldKey, true)
+}
+
+func ensureLockHeld(ctx context.Context) bool {
+	v, _ := ctx.Value(ensureLockHeldKey).(bool)
+	return v
+}
+
 // EnsureSession is the server-authoritative idempotent entry point for opening
 // a task: it returns the existing primary (or newest) session if any, otherwise
 // resolves the agent profile from the task's full context and creates a session
@@ -61,6 +78,7 @@ func (s *Service) EnsureSession(ctx context.Context, taskID string, opts ...Ensu
 	}
 	release := acquireEnsureLock(taskID)
 	defer release()
+	ctx = withEnsureLockHeld(ctx)
 
 	var o EnsureSessionOptions
 	if len(opts) > 0 {

apps/backend/internal/orchestrator/session_launch.go

@@ -170,6 +170,36 @@ func (s *Service) launchStart(ctx context.Context, req *LaunchSessionRequest) (*
 		return s.launchPrepare(ctx, req)
 	}
 
+	// Opening a task whose step lacks auto_start_agent calls session.ensure
+	// which creates a CREATED session via prepare. A subsequent explicit start
+	// (e.g. the edit-dialog "Start Agent" button) would otherwise create a
+	// duplicate session row and surface two tabs on a fresh task. Promote the
+	// prepared session to a start_created launch instead. ensureLocks is shared
+	// with EnsureSession so the two paths cannot race and both decide to create.
+	// EnsureSession holds the same mutex while re-entering through LaunchSession,
+	// so skip the acquire when its ctx marker is set — sync.Mutex isn't
+	// reentrant, and a recursive Lock would deadlock the request.
+	//
+	// Reuse is also skipped when the caller has explicitly chosen an executor
+	// or priority: start_created does not honor those, so adopting a prepared
+	// session would silently drop the caller's selection. Same logic applies
+	// when the caller's agent profile is set and disagrees with the prepared
+	// session's profile — that signals an intentionally different launch.
+	if s.canReusePreparedSession(req) {
+		if !ensureLockHeld(ctx) {
+			release := acquireEnsureLock(req.TaskID)
+			defer release()
+		}
+		if reused := s.findReusableCreatedSession(ctx, req.TaskID); reused != nil &&
+			(req.AgentProfileID == "" || req.AgentProfileID == reused.AgentProfileID) {
+			req.SessionID = reused.ID
+			if req.AgentProfileID == "" {
+				req.AgentProfileID = reused.AgentProfileID
+			}
+			return s.launchStartCreated(ctx, req)
+		}
+	}
+
 	execution, err := s.StartTask(
 		ctx, req.TaskID, req.AgentProfileID, req.ExecutorID,
 		req.ExecutorProfileID, req.Priority, req.Prompt,
@@ -181,6 +211,48 @@ func (s *Service) launchStart(ctx context.Context, req *LaunchSessionRequest) (*
 	return executionToLaunchResponse(req.TaskID, execution), nil
 }
 
+// canReusePreparedSession reports whether launchStart should consider adopting
+// an existing CREATED session. Reuse is unsafe when the caller has explicitly
+// chosen an executor or priority — start_created carries neither field, so
+// reusing would silently drop the caller's selection.
+func (s *Service) canReusePreparedSession(req *LaunchSessionRequest) bool {
+	return req.SessionID == "" &&
+		req.ExecutorID == "" &&
+		req.ExecutorProfileID == "" &&
+		req.Priority == ""
+}
+
+// findReusableCreatedSession returns a CREATED session on the task that an
+// explicit start should adopt instead of creating a new row. Primary wins; the
+// most-recently-updated CREATED session is the fallback. Non-CREATED states
+// (RUNNING, WAITING_FOR_INPUT, COMPLETED, FAILED, CANCELLED) are skipped — they
+// reflect either an active agent or an explicit terminal state and must not be
+// implicitly revived from a fresh start request.
+func (s *Service) findReusableCreatedSession(ctx context.Context, taskID string) *models.TaskSession {
+	sessions, err := s.repo.ListTaskSessions(ctx, taskID)
+	if err != nil {
+		// Swallowing the error would route the caller into StartTask and write
+		// the duplicate session row the reuse path is meant to prevent. Surface
+		// the failure so a transient DB error is visible in the logs instead.
+		s.logger.Warn("findReusableCreatedSession: list sessions failed; skipping reuse",
+			zap.String("task_id", taskID), zap.Error(err))
+		return nil
+	}
+	var best *models.TaskSession
+	for _, sess := range sessions {
+		if sess.State != models.TaskSessionStateCreated {
+			continue
+		}
+		if sess.IsPrimary {
+			return sess
+		}
+		if best == nil || sess.UpdatedAt.After(best.UpdatedAt) {
+			best = sess
+		}
+	}
+	return best
+}
+
 // shouldBlockAutoStart checks whether the task's workflow step allows auto-starting
 // the agent. Returns true when the step exists but does not have auto_start_agent
 // in its on_enter events. Tasks without a workflow step are never blocked.

apps/backend/internal/orchestrator/session_launch_test.go

@@ -478,3 +478,277 @@ func TestLaunchRestoreWorkspace_IncludesWorktreeInfo(t *testing.T) {
 		t.Errorf("expected worktree_branch 'feature/test', got %v", resp.WorktreeBranch)
 	}
 }
+
+// --- launchStart reuse of prepared sessions ---
+
+// TestFindReusableCreatedSession pins the lookup that lets an explicit Start
+// adopt a session.ensure-prepared row instead of creating a duplicate.
+func TestFindReusableCreatedSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+
+	t.Run("returns nil when task has no sessions", func(t *testing.T) {
+		if got := svc.findReusableCreatedSession(ctx, "task-missing"); got != nil {
+			t.Fatalf("expected nil, got %+v", got)
+		}
+	})
+
+	t.Run("skips non-CREATED sessions", func(t *testing.T) {
+		seedTaskAndSession(t, repo, "task-running", "s-running", models.TaskSessionStateRunning)
+		if got := svc.findReusableCreatedSession(ctx, "task-running"); got != nil {
+			t.Fatalf("expected nil for RUNNING-only task, got %+v", got)
+		}
+	})
+
+	t.Run("prefers primary CREATED over newer non-primary", func(t *testing.T) {
+		seedTaskAndSession(t, repo, "task-primary", "s-primary", models.TaskSessionStateCreated)
+		if err := repo.SetSessionPrimary(ctx, "s-primary"); err != nil {
+			t.Fatalf("set primary: %v", err)
+		}
+		newer := &models.TaskSession{
+			ID:        "s-newer",
+			TaskID:    "task-primary",
+			State:     models.TaskSessionStateCreated,
+			StartedAt: time.Now().UTC().Add(time.Minute),
+			UpdatedAt: time.Now().UTC().Add(time.Minute),
+		}
+		if err := repo.CreateTaskSession(ctx, newer); err != nil {
+			t.Fatalf("create newer: %v", err)
+		}
+		got := svc.findReusableCreatedSession(ctx, "task-primary")
+		if got == nil || got.ID != "s-primary" {
+			t.Fatalf("expected primary session, got %+v", got)
+		}
+	})
+
+	t.Run("falls back to newest CREATED when no primary", func(t *testing.T) {
+		// Stress the UpdatedAt comparison by inverting it against the SQL
+		// started_at DESC order: s-sql-newer is inserted with a LATER
+		// StartedAt but an OLDER UpdatedAt than s-updated-newer. SQL puts
+		// s-sql-newer first; only the UpdatedAt comparison in
+		// findReusableCreatedSession can pick s-updated-newer. Without that
+		// comparison, the test would erroneously accept whatever SQL returns
+		// first. CreateTaskSession honors caller-supplied StartedAt/UpdatedAt
+		// when non-zero, so this writes the desired skew directly.
+		// seedTaskAndSession creates the parent task; the seeded session is
+		// dropped immediately so only the two crafted rows remain.
+		seedTaskAndSession(t, repo, "task-no-primary", "s-throwaway", models.TaskSessionStateCreated)
+		if err := repo.DeleteTaskSession(ctx, "s-throwaway"); err != nil {
+			t.Fatalf("delete throwaway: %v", err)
+		}
+		now := time.Now().UTC()
+		sqlNewer := &models.TaskSession{
+			ID:        "s-sql-newer",
+			TaskID:    "task-no-primary",
+			State:     models.TaskSessionStateCreated,
+			StartedAt: now.Add(time.Minute),
+			UpdatedAt: now.Add(-time.Hour),
+		}
+		if err := repo.CreateTaskSession(ctx, sqlNewer); err != nil {
+			t.Fatalf("create sql-newer: %v", err)
+		}
+		updatedNewer := &models.TaskSession{
+			ID:        "s-updated-newer",
+			TaskID:    "task-no-primary",
+			State:     models.TaskSessionStateCreated,
+			StartedAt: now,
+			UpdatedAt: now,
+		}
+		if err := repo.CreateTaskSession(ctx, updatedNewer); err != nil {
+			t.Fatalf("create updated-newer: %v", err)
+		}
+		got := svc.findReusableCreatedSession(ctx, "task-no-primary")
+		if got == nil || got.ID != "s-updated-newer" {
+			t.Fatalf("expected newest-UpdatedAt session, got %+v", got)
+		}
+	})
+}
+
+// TestCanReusePreparedSession pins the gate that protects launchStart from
+// adopting a prepared session when the caller has chosen its own executor,
+// executor profile, or priority — start_created carries none of those, so
+// reusing would silently drop them.
+func TestCanReusePreparedSession(t *testing.T) {
+	repo := setupTestRepo(t)
+	svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+
+	cases := []struct {
+		name string
+		req  LaunchSessionRequest
+		want bool
+	}{
+		{"empty fields → reuse", LaunchSessionRequest{TaskID: "t"}, true},
+		{"explicit session_id → no reuse", LaunchSessionRequest{TaskID: "t", SessionID: "s"}, false},
+		{"executor_id set → no reuse", LaunchSessionRequest{TaskID: "t", ExecutorID: "e"}, false},
+		{"executor_profile_id set → no reuse", LaunchSessionRequest{TaskID: "t", ExecutorProfileID: "ep"}, false},
+		{"priority set → no reuse", LaunchSessionRequest{TaskID: "t", Priority: "high"}, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := svc.canReusePreparedSession(&tc.req); got != tc.want {
+				t.Errorf("canReusePreparedSession()=%v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestLaunchStart_ReusesPreparedSession regression-tests the duplicate-session
+// bug: opening a task whose step lacks auto_start_agent calls session.ensure
+// (creates a CREATED session via prepare); the subsequent edit-dialog "Start
+// Agent" click must adopt that session instead of letting StartTask create a
+// second row.
+func TestLaunchStart_ReusesPreparedSession(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	taskRepo := newMockTaskRepo()
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+
+	seedTaskAndSession(t, repo, "task1", "session-prepared", models.TaskSessionStateCreated)
+	taskRepo.tasks["task1"] = &v1.Task{ID: "task1", WorkspaceID: "ws1", Title: "Test", Description: "desc"}
+	if err := repo.SetSessionPrimary(ctx, "session-prepared"); err != nil {
+		t.Fatalf("set primary: %v", err)
+	}
+	prepared, err := repo.GetTaskSession(ctx, "session-prepared")
+	if err != nil {
+		t.Fatalf("load seeded session: %v", err)
+	}
+	prepared.AgentProfileID = "profile1"
+	if err := repo.UpdateTaskSession(ctx, prepared); err != nil {
+		t.Fatalf("update seeded profile: %v", err)
+	}
+
+	// Mirrors buildStartRequest from the edit dialog: explicit start, no
+	// session_id, agent profile + prompt supplied. Downstream agent launch
+	// panics on the stub scheduler — that's expected and not what this test
+	// pins. Run the call in a recovered goroutine so we can still inspect
+	// the post-state session count from the main goroutine.
+	done := make(chan struct{})
+	go func() {
+		defer func() {
+			_ = recover()
+			close(done)
+		}()
+		_, _ = svc.LaunchSession(ctx, &LaunchSessionRequest{
+			TaskID:         "task1",
+			Intent:         IntentStart,
+			AgentProfileID: "profile1",
+			Prompt:         "hello",
+		})
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("LaunchSession hung")
+	}
+
+	sessions, err := repo.ListTaskSessions(ctx, "task1")
+	if err != nil {
+		t.Fatalf("list sessions: %v", err)
+	}
+	if len(sessions) != 1 {
+		t.Fatalf("expected single session after reuse, got %d", len(sessions))
+	}
+	if sessions[0].ID != "session-prepared" {
+		t.Fatalf("expected reused session id, got %q", sessions[0].ID)
+	}
+}
+
+// TestLaunchStart_NoReuseOnProfileMismatch confirms that when the caller picks
+// an agent profile that differs from the prepared session's, launchStart falls
+// through to StartTask instead of silently overriding the prepared session's
+// profile.
+func TestLaunchStart_NoReuseOnProfileMismatch(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	taskRepo := newMockTaskRepo()
+	svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{})
+
+	seedTaskAndSession(t, repo, "task1", "session-prepared", models.TaskSessionStateCreated)
+	taskRepo.tasks["task1"] = &v1.Task{ID: "task1", WorkspaceID: "ws1", Title: "Test", Description: "desc"}
+	if err := repo.SetSessionPrimary(ctx, "session-prepared"); err != nil {
+		t.Fatalf("set primary: %v", err)
+	}
+	prepared, err := repo.GetTaskSession(ctx, "session-prepared")
+	if err != nil {
+		t.Fatalf("load seeded session: %v", err)
+	}
+	prepared.AgentProfileID = "profile-prepared"
+	if err := repo.UpdateTaskSession(ctx, prepared); err != nil {
+		t.Fatalf("update seeded profile: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer func() {
+			_ = recover()
+			close(done)
+		}()
+		_, _ = svc.LaunchSession(ctx, &LaunchSessionRequest{
+			TaskID:         "task1",
+			Intent:         IntentStart,
+			AgentProfileID: "profile-different",
+			Prompt:         "hello",
+		})
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("LaunchSession hung")
+	}
+
+	sessions, err := repo.ListTaskSessions(ctx, "task1")
+	if err != nil {
+		t.Fatalf("list sessions: %v", err)
+	}
+	if len(sessions) != 2 {
+		t.Fatalf("expected fresh session for different profile, got %d sessions", len(sessions))
+	}
+}
+
+// TestEnsureSession_AutoStartDoesNotDeadlock guards the reentrant-mutex bug:
+// EnsureSession acquires ensureLocks[taskID] then calls LaunchSession; when
+// the step has auto_start_agent, the dispatch lands in launchStart which used
+// to re-acquire the same non-reentrant mutex and freeze the request. The fix
+// threads an ensureLockHeld context marker so launchStart skips the inner
+// acquire — this test fails (hangs) without that marker.
+func TestEnsureSession_AutoStartDoesNotDeadlock(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	taskRepo := newMockTaskRepo()
+	stepGetter := newMockStepGetter()
+	stepGetter.steps["step-auto"] = &wfmodels.WorkflowStep{
+		ID:             "step-auto",
+		Name:           "auto-start",
+		AgentProfileID: "profile1",
+		Events: wfmodels.StepEvents{
+			OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}},
+		},
+	}
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, &mockAgentManager{})
+
+	seedTaskAndSessionWithStep(t, repo, "task1", "session-throwaway", "step-auto")
+	if err := repo.DeleteTaskSession(ctx, "session-throwaway"); err != nil {
+		t.Fatalf("delete seed session: %v", err)
+	}
+	taskRepo.tasks["task1"] = &v1.Task{
+		ID:          "task1",
+		WorkspaceID: "ws1",
+		Title:       "Test",
+		Description: "desc",
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer func() {
+			_ = recover()
+			close(done)
+		}()
+		_, _ = svc.EnsureSession(ctx, "task1")
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("EnsureSession deadlocked: launchStart re-acquired ensureLock that EnsureSession already holds")
+	}
+}