diff --git a/apps/backend/internal/orchestrator/session_ensure.go b/apps/backend/internal/orchestrator/session_ensure.go index de4f47fa8e..3d3dfa77cc 100644 --- a/apps/backend/internal/orchestrator/session_ensure.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/session_launch.go b/apps/backend/internal/orchestrator/session_launch.go index aea429fe75..e67db021de 100644 --- a/apps/backend/internal/orchestrator/session_launch.go +++ b/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 && + profilesCompatible(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,60 @@ 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 == "" +} + +// profilesCompatible reports whether two agent profile IDs are compatible for +// reuse: identical, or at least one is empty (start_created can resolve the +// missing side from the other). Different non-empty profiles are intentionally +// distinct launches and must NOT share a session row. +func profilesCompatible(caller, session string) bool { + if caller == "" || session == "" { + return true + } + return caller == session +} + +// 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 { + // Returning nil here means launchStart still falls through to + // StartTask — same behavior as before this PR landed when the lookup + // failed. The warn just keeps a transient DB failure visible in the + // logs rather than letting it race silently into a fresh session. + 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. diff --git a/apps/backend/internal/orchestrator/session_launch_test.go b/apps/backend/internal/orchestrator/session_launch_test.go index dfa7b2a6da..2bab6a8de5 100644 --- a/apps/backend/internal/orchestrator/session_launch_test.go +++ b/apps/backend/internal/orchestrator/session_launch_test.go @@ -400,7 +400,7 @@ func TestLaunchPrepare_PassthroughDoesNotRecurse(t *testing.T) { select { case <-done: // returned without recursing - case <-time.After(2 * time.Second): + case <-time.After(150 * time.Millisecond): t.Fatal("LaunchSession recursed indefinitely (timed out)") } } @@ -478,3 +478,301 @@ 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) + } + }) +} + +// TestProfilesCompatible covers the agent-profile matching rule used by +// launchStart's reuse path: identical or either-side-empty is fine; differing +// non-empty profiles must NOT share a session row. +func TestProfilesCompatible(t *testing.T) { + cases := []struct { + name string + caller, sess string + want bool + }{ + {"both empty", "", "", true}, + {"caller empty", "", "p", true}, + {"session empty", "p", "", true}, + {"same profile", "p", "p", true}, + {"different profiles", "p1", "p2", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := profilesCompatible(tc.caller, tc.sess); got != tc.want { + t.Errorf("profilesCompatible(%q,%q)=%v, want %v", tc.caller, tc.sess, got, tc.want) + } + }) + } +} + +// 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(150 * time.Millisecond): + 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(150 * time.Millisecond): + 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(150 * time.Millisecond): + t.Fatal("EnsureSession deadlocked: launchStart re-acquired ensureLock that EnsureSession already holds") + } +}