From 5d3ddff8b632ae2b89952f9d98a414f8b125318a Mon Sep 17 00:00:00 2001 From: Nicolas Lenepveu Date: Thu, 11 Jun 2026 15:51:34 +0200 Subject: [PATCH] fix(task): reap active sessions when archiving a task (#1275) --- .../task/handlers/process_handlers_test.go | 3 + .../internal/task/repository/interface.go | 1 + .../task/repository/sqlite/session.go | 19 +++ .../task/repository/sqlite/session_test.go | 112 ++++++++++++++++++ .../internal/task/service/service_tasks.go | 13 ++ 5 files changed, 148 insertions(+) diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 2caf91784f..d757971a3d 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -267,6 +267,9 @@ func (m *mockRepository) ListActiveTaskSessions(ctx context.Context) ([]*models. func (m *mockRepository) ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) { return nil, nil } +func (m *mockRepository) CancelActiveTaskSessionsByTaskID(ctx context.Context, taskID, reason string) (int64, error) { + return 0, nil +} func (m *mockRepository) HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) { return false, nil } diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index c22938205c..3f00a55062 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -140,6 +140,7 @@ type SessionRepository interface { ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) + CancelActiveTaskSessionsByTaskID(ctx context.Context, taskID, reason string) (int64, error) HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) GetActiveTaskInfoByAgentProfile(ctx context.Context, agentProfileID string) ([]agentdto.ActiveTaskInfo, error) HasActiveTaskSessionsByExecutor(ctx context.Context, executorID string) (bool, error) diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index f3d97c9bf0..2e4b9705ab 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -514,6 +514,25 @@ func (r *Repository) UpdateTaskSessionState(ctx context.Context, id string, stat return nil } +// CancelActiveTaskSessionsByTaskID transitions every active session of a task +// (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT) to CANCELLED, returning the +// number of rows changed. The transition is a pure DB state change and does not +// require a live agent execution, making it the authoritative way to finalize a +// task's sessions independent of agent-process teardown. +func (r *Repository) CancelActiveTaskSessionsByTaskID(ctx context.Context, taskID, reason string) (int64, error) { + now := time.Now().UTC() + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE task_sessions + SET state = ?, error_message = ?, completed_at = ?, updated_at = ? + WHERE task_id = ? + AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT') + `), string(models.TaskSessionStateCancelled), reason, now, now, taskID) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + // UpdateSessionMetadata updates only the metadata column of a session, // avoiding a full-row overwrite that could clobber concurrent field updates. func (r *Repository) UpdateSessionMetadata(ctx context.Context, sessionID string, metadata map[string]interface{}) error { diff --git a/apps/backend/internal/task/repository/sqlite/session_test.go b/apps/backend/internal/task/repository/sqlite/session_test.go index dc79c6b69f..9fe0f260ee 100644 --- a/apps/backend/internal/task/repository/sqlite/session_test.go +++ b/apps/backend/internal/task/repository/sqlite/session_test.go @@ -455,3 +455,115 @@ func TestHasActiveTaskSessionsByRepository_ExcludesArchivedTasks(t *testing.T) { t.Error("live task session must mark repository as active") } } + +// insertSession inserts a task_session row in the given state directly, for +// seeding multiple sessions on a single task (seedRepoLink can only create one +// session per task because its task_repositories PK is keyed on task+repo). +func insertSession(t *testing.T, repo *Repository, sessionID, taskID, state string) { + t.Helper() + now := time.Now().UTC() + if _, err := repo.db.Exec(repo.db.Rebind(` + INSERT INTO task_sessions (id, task_id, state, started_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), sessionID, taskID, state, now, now); err != nil { + t.Fatalf("seed session %s: %v", sessionID, err) + } +} + +func sessionState(t *testing.T, repo *Repository, sessionID string) string { + t.Helper() + var state string + if err := repo.ro.QueryRowx(repo.ro.Rebind( + `SELECT state FROM task_sessions WHERE id = ?`), sessionID).Scan(&state); err != nil { + t.Fatalf("read state for %s: %v", sessionID, err) + } + return state +} + +func sessionCancellationMetadata(t *testing.T, repo *Repository, sessionID string) (string, sql.NullTime, time.Time) { + t.Helper() + var errorMessage string + var completedAt sql.NullTime + var updatedAt time.Time + if err := repo.ro.QueryRowx(repo.ro.Rebind( + `SELECT error_message, completed_at, updated_at FROM task_sessions WHERE id = ?`), + sessionID, + ).Scan(&errorMessage, &completedAt, &updatedAt); err != nil { + t.Fatalf("read cancellation metadata for %s: %v", sessionID, err) + } + return errorMessage, completedAt, updatedAt +} + +func assertReapedSession(t *testing.T, repo *Repository, sessionID string, reapedAfter time.Time) { + t.Helper() + if got := sessionState(t, repo, sessionID); got != "CANCELLED" { + t.Errorf("%s = %q, want CANCELLED", sessionID, got) + } + errorMessage, completedAt, updatedAt := sessionCancellationMetadata(t, repo, sessionID) + if errorMessage != "task archived" { + t.Errorf("%s error_message = %q, want task archived", sessionID, errorMessage) + } + if !completedAt.Valid { + t.Errorf("%s completed_at should be set", sessionID) + } + if updatedAt.Before(reapedAfter) { + t.Errorf("%s updated_at = %s, want >= %s", sessionID, updatedAt, reapedAfter) + } +} + +// TestCancelActiveTaskSessionsByTaskID verifies the archive reaper transitions +// only the target task's still-active sessions to CANCELLED, leaves terminal +// sessions and other tasks untouched, and reports the rows changed. It also +// confirms the repo-delete guard reports the repository as free afterward — +// the end-to-end purpose of the reap. +func TestCancelActiveTaskSessionsByTaskID(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + + // Target task: one active session via the link helper, plus a second active + // session, two pre-run sessions, and an already-terminal session inserted directly. + seedRepoLink(t, repo, "ws-r", "repo-r", "task-r", "sess-r1", "WAITING_FOR_INPUT") + insertSession(t, repo, "sess-r2", "task-r", "RUNNING") + insertSession(t, repo, "sess-r3", "task-r", "CREATED") + insertSession(t, repo, "sess-r4", "task-r", "STARTING") + insertSession(t, repo, "sess-r5", "task-r", "COMPLETED") + // A different task on a different repo — must be untouched. + seedRepoLink(t, repo, "ws-r", "repo-other", "task-other", "sess-o1", "RUNNING") + + reapedAfter := time.Now().UTC() + reaped, err := repo.CancelActiveTaskSessionsByTaskID(ctx, "task-r", "task archived") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if reaped != 4 { + t.Errorf("expected 4 active sessions reaped, got %d", reaped) + } + for _, sessionID := range []string{"sess-r1", "sess-r2", "sess-r3", "sess-r4"} { + assertReapedSession(t, repo, sessionID, reapedAfter) + } + if got := sessionState(t, repo, "sess-r5"); got != "COMPLETED" { + t.Errorf("sess-r5 (terminal) = %q, want unchanged COMPLETED", got) + } + if got := sessionState(t, repo, "sess-o1"); got != "RUNNING" { + t.Errorf("sess-o1 (other task) = %q, want unchanged RUNNING", got) + } + + // End-to-end: the repository that only the reaped task referenced is now + // reported as free by the delete guard. + active, err := repo.HasActiveTaskSessionsByRepository(ctx, "repo-r") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if active { + t.Error("repo-r should have no active sessions after reaping its task") + } + + // Idempotent: a second call changes nothing. + reaped, err = repo.CancelActiveTaskSessionsByTaskID(ctx, "task-r", "task archived") + if err != nil { + t.Fatalf("unexpected error on second call: %v", err) + } + if reaped != 0 { + t.Errorf("expected 0 rows on idempotent re-run, got %d", reaped) + } +} diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 7911b66db4..f348981da9 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -788,6 +788,19 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { return err } + // 3b. Finalize active sessions in the DB. The async cleanup below tears down + // the agent processes; this records the terminal session state, which + // process teardown does not persist on its own. + if reaped, rerr := s.sessions.CancelActiveTaskSessionsByTaskID(ctx, id, "task archived"); rerr != nil { + s.logger.Warn("failed to reap active sessions on archive", + zap.String("task_id", id), + zap.Error(rerr)) + } else if reaped > 0 { + s.logger.Info("reaped active sessions on archive", + zap.String("task_id", id), + zap.Int64("count", reaped)) + } + // 4. Re-read task for updated archived_at field task, err = s.tasks.GetTask(ctx, id) if err != nil {