diff --git a/apps/backend/internal/automation/models.go b/apps/backend/internal/automation/models.go index 1f9cf22f8b..01dd8b74f9 100644 --- a/apps/backend/internal/automation/models.go +++ b/apps/backend/internal/automation/models.go @@ -45,13 +45,25 @@ const ( RunStatusSucceeded RunStatus = "succeeded" RunStatusFailed RunStatus = "failed" RunStatusSkipped RunStatus = "skipped" + // RunStatusArchived is a read-time-derived terminal status: it is never + // written by a trigger-firing code path, only computed by ListRuns when + // a task_created run's generated task has been archived (archived_at + // set) — whether archived through the UI or by the agent itself (e.g. + // via an "archive this task" instruction). The task's outcome may still + // be known (succeeded/failed already recorded); this only overrides the + // still-open task_created status, never a real terminal one. + RunStatusArchived RunStatus = "archived" // RunStatusCancelled is a read-time-derived terminal status: it is never // written by a trigger-firing code path, only computed by ListRuns (and - // implied by CountActiveRuns' filter) when a task_created run's - // generated task has been archived or no longer exists. The task's - // outcome is unknown at that point — the user closed it out some other - // way — so this is deliberately distinct from succeeded/failed rather - // than guessing one. + // implied by CountActiveRuns' filter) for a task_created run whose + // generated task either no longer exists (deleted — outcome + // unrecoverable) or whose task's *current* session (task_sessions row + // with is_primary = 1) was explicitly put into the CANCELLED state — + // set only when an agent run was manually stopped (a deliberate + // cancellation, distinct from archiving; the task's own state is not + // touched by a stop and so isn't consulted here). Archived_at takes + // precedence over the cancelled-session check when both apply — see + // listRunsWithTaskState. RunStatusCancelled RunStatus = "cancelled" ) diff --git a/apps/backend/internal/automation/store.go b/apps/backend/internal/automation/store.go index c3d13af679..f4d48ee40e 100644 --- a/apps/backend/internal/automation/store.go +++ b/apps/backend/internal/automation/store.go @@ -14,6 +14,7 @@ import ( "github.com/jmoiron/sqlx" "github.com/kandev/kandev/internal/db" + taskmodels "github.com/kandev/kandev/internal/task/models" ) // Store provides SQLite persistence for automations. @@ -438,12 +439,13 @@ func (s *Store) updateRunTerminalStatus(ctx context.Context, taskID string, stat } // ListRuns returns recent runs for an automation. A task_created run whose -// generated task has been archived or no longer exists is reported as -// cancelled — the task's outcome is unknown and it's no longer outstanding -// work — without touching runs that already reached a real terminal -// status. Falls back to the raw stored status when the tasks table isn't -// present (isolated automation-only tests; production always has it, -// migrated by the task repository before automation triggers can fire). +// generated task has been archived is reported as archived; one whose task +// is gone entirely or explicitly cancelled is reported as cancelled — see +// listRunsWithTaskState for the full precedence — without touching runs +// that already reached a real terminal status. Falls back to the raw +// stored status when the tasks table isn't present (isolated +// automation-only tests; production always has it, migrated by the task +// repository before automation triggers can fire). func (s *Store) ListRuns(ctx context.Context, automationID string, limit int) ([]*AutomationRun, error) { if limit <= 0 { limit = 50 @@ -470,11 +472,36 @@ func (s *Store) listRunsWithTaskState(ctx context.Context, automationID string, // and displays as cancelled rather than its raw stored status — // reachable today only through the e2e run-seeding endpoint, never in // production. + // + // Three read-time overrides of a still-open task_created run, in + // priority order: (1) the task row is gone entirely — deleted, outcome + // unrecoverable — shown as cancelled; (2) the task was archived + // (archived_at set, via the UI or by the agent itself, e.g. an + // "archive this task" instruction) — shown as archived, which takes + // precedence over the session check below; (3) the task's current + // (is_primary) session is CANCELLED — set only when an agent run was + // manually stopped (coordinator/MCP stop_task, or the UI Stop button) + // — a deliberate cancellation, shown as cancelled. Task.state is + // deliberately NOT consulted here: stopping an agent leaves the task + // itself at whatever state the stop caller chose (e.g. REVIEW) and + // only ever marks the *session* CANCELLED — see + // orchestrator.handleAgentStopped's "we do NOT update task state + // here" note. Filtering on is_primary picks the task's current + // session, so a resumed-and-completed task isn't misclassified by a + // stale cancelled session left over from an earlier stop. EXISTS + // (rather than a LEFT JOIN) keeps the query correct even if the + // "at most one is_primary=1 row per task" invariant is ever violated + // — a join would fan out and duplicate the run. var runs []*AutomationRun err := s.ro.SelectContext(ctx, &runs, ` SELECT ar.id, ar.automation_id, ar.trigger_id, ar.trigger_type, ar.task_id, CASE - WHEN ar.status = ? AND (t.id IS NULL OR t.archived_at IS NOT NULL) THEN ? + WHEN ar.status = ? AND t.id IS NULL THEN ? + WHEN ar.status = ? AND t.archived_at IS NOT NULL THEN ? + WHEN ar.status = ? AND EXISTS ( + SELECT 1 FROM task_sessions ts + WHERE ts.task_id = ar.task_id AND ts.is_primary = 1 AND ts.state = ? + ) THEN ? ELSE ar.status END AS status, ar.dedup_key, ar.trigger_data, ar.error_message, ar.created_at @@ -482,7 +509,10 @@ func (s *Store) listRunsWithTaskState(ctx context.Context, automationID string, LEFT JOIN tasks t ON t.id = ar.task_id WHERE ar.automation_id = ? ORDER BY ar.created_at DESC LIMIT ?`, - string(RunStatusTaskCreated), string(RunStatusCancelled), automationID, limit) + string(RunStatusTaskCreated), string(RunStatusCancelled), + string(RunStatusTaskCreated), string(RunStatusArchived), + string(RunStatusTaskCreated), string(taskmodels.TaskSessionStateCancelled), string(RunStatusCancelled), + automationID, limit) return runs, err } @@ -508,11 +538,11 @@ func (s *Store) HasRunWithDedupKey(ctx context.Context, automationID, dedupKey s // CountActiveRuns returns the number of runs with task_created status for // an automation whose generated task is still open. A task_created run -// whose task was archived or deleted no longer represents outstanding -// work — the user closed it out some other way — so it must not keep -// counting against max_concurrent_runs forever. Falls back to a plain -// count when the tasks table isn't present (isolated automation-only -// tests; production always has it). +// whose task was archived, deleted, or explicitly cancelled no longer +// represents outstanding work — the user (or agent) closed it out some +// other way — so it must not keep counting against max_concurrent_runs +// forever. Falls back to a plain count when the tasks table isn't present +// (isolated automation-only tests; production always has it). func (s *Store) CountActiveRuns(ctx context.Context, automationID string) (int, error) { count, err := s.countActiveRunsWithTaskState(ctx, automationID) if db.IsMissingTableError(err) { @@ -525,13 +555,20 @@ func (s *Store) countActiveRunsWithTaskState(ctx context.Context, automationID s // Same non-empty-task_id assumption as listRunsWithTaskState above: an // empty ar.task_id never matches a real task row either, so such a run // silently falls out of the active count below instead of erroring. + // See listRunsWithTaskState for why the current (is_primary) session's + // state, not the task's own state, is the cancellation signal, and why + // NOT EXISTS rather than a LEFT JOIN. var count int err := s.ro.GetContext(ctx, &count, ` SELECT COUNT(*) FROM automation_runs ar LEFT JOIN tasks t ON t.id = ar.task_id WHERE ar.automation_id = ? AND ar.status = ? - AND t.id IS NOT NULL AND t.archived_at IS NULL`, - automationID, string(RunStatusTaskCreated)) + AND t.id IS NOT NULL AND t.archived_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM task_sessions ts + WHERE ts.task_id = ar.task_id AND ts.is_primary = 1 AND ts.state = ? + )`, + automationID, string(RunStatusTaskCreated), string(taskmodels.TaskSessionStateCancelled)) return count, err } diff --git a/apps/backend/internal/automation/store_test.go b/apps/backend/internal/automation/store_test.go index 093d923394..2f8753f11d 100644 --- a/apps/backend/internal/automation/store_test.go +++ b/apps/backend/internal/automation/store_test.go @@ -301,16 +301,21 @@ func TestDeleteAllRuns(t *testing.T) { } // createTasksTable adds a minimal shadow of the task repository's `tasks` -// table (id, archived_at) to the store's DB. The automation package never -// owns this table — apps/backend/internal/task/repository/sqlite is the -// canonical owner — so only tests that exercise the CountActiveRuns/ListRuns -// task-state join create it; production always has the real table already -// migrated by the task repository before automation triggers can fire. +// and `task_sessions` tables (id/archived_at, and task_id/is_primary/state +// respectively) to the store's DB. The automation package never owns +// these tables — apps/backend/internal/task/repository/sqlite is the +// canonical owner — so only tests that exercise the CountActiveRuns/ +// ListRuns task-state join create them; production always has the real +// tables already migrated by the task repository before automation +// triggers can fire. func createTasksTable(t *testing.T, store *Store) { t.Helper() if _, err := store.db.Exec(`CREATE TABLE tasks (id TEXT PRIMARY KEY, archived_at DATETIME)`); err != nil { t.Fatal(err) } + if _, err := store.db.Exec(`CREATE TABLE task_sessions (id TEXT PRIMARY KEY, task_id TEXT NOT NULL, is_primary INTEGER NOT NULL DEFAULT 0, state TEXT NOT NULL DEFAULT 'CREATED')`); err != nil { + t.Fatal(err) + } } func insertTask(t *testing.T, store *Store, id string, archived bool) { @@ -324,15 +329,47 @@ func insertTask(t *testing.T, store *Store, id string, archived bool) { } } -// TestCountActiveRuns_ExcludesArchivedOrMissingTask reproduces the reported -// bug: an automation-generated task that gets archived (manually, via -// auto-archive, or via cascade) leaves its automation run stuck at -// task_created forever unless CountActiveRuns checks the task's current -// state. A run whose task is archived, gone entirely, or never recorded -// (empty task_id — see countActiveRunsWithTaskState's docstring in -// store.go) no longer represents outstanding work and must not count -// against max_concurrent_runs. -func TestCountActiveRuns_ExcludesArchivedOrMissingTask(t *testing.T) { +// insertPrimarySession seeds the given task's *current* session (is_primary +// = 1) with an explicit models.TaskSessionState string (e.g. "CANCELLED"), +// so tests can exercise the genuine-cancellation branch of +// listRunsWithTaskState/countActiveRunsWithTaskState independently of +// archived_at. Mirrors production: a task has at most one is_primary = 1 +// session at a time (SetPrimarySession unsets the rest on resume). +func insertPrimarySession(t *testing.T, store *Store, taskID, state string) { + t.Helper() + if _, err := store.db.Exec( + `INSERT INTO task_sessions (id, task_id, is_primary, state) VALUES (?, ?, 1, ?)`, + taskID+"-session", taskID, state, + ); err != nil { + t.Fatal(err) + } +} + +// insertStaleSession seeds a non-primary (is_primary = 0) session for the +// given task — e.g. a CANCELLED session left over from a stop, before the +// task was resumed and its is_primary flag moved to a fresh session. Used +// to prove the current-session filter isn't fooled by cancellation +// history that no longer reflects the task's live state. +func insertStaleSession(t *testing.T, store *Store, taskID, state string) { + t.Helper() + if _, err := store.db.Exec( + `INSERT INTO task_sessions (id, task_id, is_primary, state) VALUES (?, ?, 0, ?)`, + taskID+"-stale-session", taskID, state, + ); err != nil { + t.Fatal(err) + } +} + +// TestCountActiveRuns_ExcludesArchivedCancelledOrMissingTask reproduces the +// reported bug: an automation-generated task that gets archived (manually, +// via auto-archive, via cascade, or by the agent itself) or is explicitly +// cancelled leaves its automation run stuck at task_created forever unless +// CountActiveRuns checks the task's current state. A run whose task is +// archived, cancelled, gone entirely, or never recorded (empty task_id — +// see countActiveRunsWithTaskState's docstring in store.go) no longer +// represents outstanding work and must not count against +// max_concurrent_runs. +func TestCountActiveRuns_ExcludesArchivedCancelledOrMissingTask(t *testing.T) { store := setupTestStore(t) createTasksTable(t, store) ctx := context.Background() @@ -344,11 +381,18 @@ func TestCountActiveRuns_ExcludesArchivedOrMissingTask(t *testing.T) { insertTask(t, store, "task-active", false) insertTask(t, store, "task-archived", true) + insertTask(t, store, "task-cancelled", false) + insertPrimarySession(t, store, "task-cancelled", "CANCELLED") + // Resumed after a stop: the stale CANCELLED session is no longer + // primary, so this task must still count as active. + insertTask(t, store, "task-resumed", false) + insertStaleSession(t, store, "task-resumed", "CANCELLED") + insertPrimarySession(t, store, "task-resumed", "RUNNING") // "task-missing" deliberately has no row in tasks at all; "" (the // task_id column default) exercises the same no-live-task branch // through a different route. - for _, tid := range []string{"task-active", "task-archived", "task-missing", ""} { + for _, tid := range []string{"task-active", "task-archived", "task-cancelled", "task-resumed", "task-missing", ""} { if err := store.CreateRun(ctx, &AutomationRun{ AutomationID: a.ID, TriggerType: TriggerTypeScheduled, @@ -364,8 +408,8 @@ func TestCountActiveRuns_ExcludesArchivedOrMissingTask(t *testing.T) { if err != nil { t.Fatal(err) } - if count != 1 { - t.Fatalf("expected 1 active run (only task-active is open), got %d", count) + if count != 2 { + t.Fatalf("expected 2 active runs (task-active and task-resumed are open), got %d", count) } } @@ -400,12 +444,16 @@ func TestCountActiveRuns_FallsBackWhenTasksTableAbsent(t *testing.T) { } } -// TestListRuns_ArchivedOrMissingTaskShowsCancelled ensures the "Recent Runs" -// list the settings UI reads stops labeling a run "Running" once its -// generated task is archived, gone, or never recorded (empty task_id — see -// listRunsWithTaskState's docstring in store.go), without touching the -// stored status of runs that already reached a real terminal outcome. -func TestListRuns_ArchivedOrMissingTaskShowsCancelled(t *testing.T) { +// TestListRuns_DerivesArchivedCancelledAndActiveStatus ensures the "Recent +// Runs" list the settings UI reads stops labeling a run "Running" once its +// generated task is archived, cancelled, gone, or never recorded (empty +// task_id — see listRunsWithTaskState's docstring in store.go). Archived +// (regardless of whether the UI or the agent itself triggered it) must be +// visually distinct from a genuine user cancellation (task state +// CANCELLED), and archived_at takes precedence when a task is both +// cancelled and archived. Runs that already reached a real terminal +// outcome are left untouched. +func TestListRuns_DerivesArchivedCancelledAndActiveStatus(t *testing.T) { store := setupTestStore(t) createTasksTable(t, store) ctx := context.Background() @@ -416,13 +464,28 @@ func TestListRuns_ArchivedOrMissingTaskShowsCancelled(t *testing.T) { } insertTask(t, store, "task-active", false) insertTask(t, store, "task-archived", true) + insertTask(t, store, "task-cancelled", false) + insertPrimarySession(t, store, "task-cancelled", "CANCELLED") + insertTask(t, store, "task-cancelled-and-archived", true) + insertPrimarySession(t, store, "task-cancelled-and-archived", "CANCELLED") + // Stopped once (stale, non-primary CANCELLED session), then resumed + // with a fresh primary session that completed. Must read as its raw + // stored status, not cancelled — the stale session no longer reflects + // the task's live state. + insertTask(t, store, "task-resumed", false) + insertStaleSession(t, store, "task-resumed", "CANCELLED") + insertPrimarySession(t, store, "task-resumed", "COMPLETED") active := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-active", TriggerData: json.RawMessage(`{}`)} archived := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-archived", TriggerData: json.RawMessage(`{}`)} + cancelled := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-cancelled", TriggerData: json.RawMessage(`{}`)} + cancelledAndArchived := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-cancelled-and-archived", TriggerData: json.RawMessage(`{}`)} + resumed := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-resumed", TriggerData: json.RawMessage(`{}`)} missing := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-missing", TriggerData: json.RawMessage(`{}`)} emptyTaskID := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "", TriggerData: json.RawMessage(`{}`)} succeededOnArchived := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusSucceeded, TaskID: "task-archived", TriggerData: json.RawMessage(`{}`)} - for _, r := range []*AutomationRun{active, archived, missing, emptyTaskID, succeededOnArchived} { + allRuns := []*AutomationRun{active, archived, cancelled, cancelledAndArchived, resumed, missing, emptyTaskID, succeededOnArchived} + for _, r := range allRuns { if err := store.CreateRun(ctx, r); err != nil { t.Fatal(err) } @@ -439,8 +502,17 @@ func TestListRuns_ArchivedOrMissingTaskShowsCancelled(t *testing.T) { if s := statusByID[active.ID]; s != RunStatusTaskCreated { t.Errorf("active task run: expected task_created, got %q", s) } - if s := statusByID[archived.ID]; s != RunStatusCancelled { - t.Errorf("archived task run: expected cancelled, got %q", s) + if s := statusByID[archived.ID]; s != RunStatusArchived { + t.Errorf("archived task run: expected archived, got %q", s) + } + if s := statusByID[cancelled.ID]; s != RunStatusCancelled { + t.Errorf("cancelled task run: expected cancelled, got %q", s) + } + if s := statusByID[cancelledAndArchived.ID]; s != RunStatusArchived { + t.Errorf("cancelled-and-archived task run: expected archived_at to take precedence (archived), got %q", s) + } + if s := statusByID[resumed.ID]; s != RunStatusTaskCreated { + t.Errorf("resumed-after-cancel task run: expected task_created (stale session ignored), got %q", s) } if s := statusByID[missing.ID]; s != RunStatusCancelled { t.Errorf("missing task run: expected cancelled, got %q", s) diff --git a/apps/backend/internal/backendapp/e2e_reset.go b/apps/backend/internal/backendapp/e2e_reset.go index e22a64c168..e6f4327320 100644 --- a/apps/backend/internal/backendapp/e2e_reset.go +++ b/apps/backend/internal/backendapp/e2e_reset.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/gitlab" + taskmodels "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" ) @@ -49,6 +50,10 @@ func registerE2EResetRoutes( api.POST("/automations", handleE2ECreateAutomation(automationSvc, log)) api.POST("/automation-runs", handleE2ECreateAutomationRun(automationSvc, log)) } + // Seeds a task_sessions row directly (e.g. a CANCELLED primary session) + // so tests can assert on session-state-derived behavior without a full + // agent stop/resume cycle. + api.POST("/task-sessions", handleE2ECreateTaskSession(repo, log)) log.Info("registered E2E endpoints (test-only)") } @@ -321,3 +326,59 @@ func handleE2ECreateAutomationRun(svc *automation.Service, log *logger.Logger) g }) } } + +type e2eCreateTaskSessionRequest struct { + TaskID string `json:"task_id"` + State string `json:"state"` + // IsPrimary defaults to true: tests seeding a single session almost + // always want it to be the task's *current* session (the one + // listRunsWithTaskState/countActiveRunsWithTaskState read). Set to + // false to seed a stale, superseded session for resume scenarios. + IsPrimary *bool `json:"is_primary"` +} + +// handleE2ECreateTaskSession seeds a task_sessions row directly for E2E +// tests that need to assert on session-state-derived behavior (e.g. the +// automation Recent Runs "Cancelled" status, which is keyed off the task's +// primary session state, not the task's own state — see +// internal/automation.Store.listRunsWithTaskState) without driving a real +// agent through a full stop/resume cycle. +func handleE2ECreateTaskSession(repo *sqliterepo.Repository, log *logger.Logger) gin.HandlerFunc { + return func(c *gin.Context) { + var body e2eCreateTaskSessionRequest + if err := c.ShouldBindJSON(&body); err != nil || body.TaskID == "" || body.State == "" { + c.JSON(http.StatusBadRequest, gin.H{errKey: "task_id and state are required"}) + return + } + isPrimary := true + if body.IsPrimary != nil { + isPrimary = *body.IsPrimary + } + session := &taskmodels.TaskSession{ + TaskID: body.TaskID, + State: taskmodels.TaskSessionState(body.State), + } + if err := repo.CreateTaskSession(c.Request.Context(), session); err != nil { + log.Error("e2e: failed to create task session", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()}) + return + } + // SetSessionPrimary clears is_primary on every other session for this + // task before setting it here, mirroring production's resume path + // (Repository.SetSessionPrimary) so seeding a second primary session + // for a task that already has one can't leave two is_primary = 1 rows + // behind — that would corrupt the run-status query's read of the + // task's "current" session (see listRunsWithTaskState). + if isPrimary { + if err := repo.SetSessionPrimary(c.Request.Context(), session.ID); err != nil { + log.Error("e2e: failed to set primary task session", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()}) + return + } + session.IsPrimary = true + } + c.JSON(http.StatusCreated, gin.H{ + "id": session.ID, taskIDPayloadKey: session.TaskID, statusKey: session.State, + }) + } +} diff --git a/apps/backend/internal/task/repository/session_repository_test.go b/apps/backend/internal/task/repository/session_repository_test.go index f4699ae1d6..a5cd5f774b 100644 --- a/apps/backend/internal/task/repository/session_repository_test.go +++ b/apps/backend/internal/task/repository/session_repository_test.go @@ -771,3 +771,54 @@ func TestListWorktreesBySessionIDs_ChunksOverPlaceholderLimit(t *testing.T) { t.Errorf("expected empty map for non-existent session IDs, got %d entries", len(got)) } } + +// TestSetSessionPrimary_DemotesOthersAndPromotesExactlyOne pins the core +// invariant SetSessionPrimary exists to maintain: after promoting a second +// session, the first is demoted and exactly one session is primary. See +// internal/automation.Store.listRunsWithTaskState, which reads a task's +// current session via is_primary = 1 and would misclassify a run if two +// sessions were ever left primary. +func TestSetSessionPrimary_DemotesOthersAndPromotesExactlyOne(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-primary", Name: "WF"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-primary", WorkflowID: "wf-primary", WorkflowStepID: "step-1", Title: "T", + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + s1 := &models.TaskSession{ID: "sess-primary-1", TaskID: "task-primary", State: models.TaskSessionStateCancelled} + if err := repo.CreateTaskSession(ctx, s1); err != nil { + t.Fatalf("CreateTaskSession s1: %v", err) + } + s2 := &models.TaskSession{ID: "sess-primary-2", TaskID: "task-primary", State: models.TaskSessionStateRunning} + if err := repo.CreateTaskSession(ctx, s2); err != nil { + t.Fatalf("CreateTaskSession s2: %v", err) + } + + if err := repo.SetSessionPrimary(ctx, s1.ID); err != nil { + t.Fatalf("SetSessionPrimary(s1): %v", err) + } + if err := repo.SetSessionPrimary(ctx, s2.ID); err != nil { + t.Fatalf("SetSessionPrimary(s2): %v", err) + } + + got1, err := repo.GetTaskSession(ctx, s1.ID) + if err != nil { + t.Fatalf("GetTaskSession(s1): %v", err) + } + got2, err := repo.GetTaskSession(ctx, s2.ID) + if err != nil { + t.Fatalf("GetTaskSession(s2): %v", err) + } + if got1.IsPrimary { + t.Errorf("s1 still primary after promoting s2 — exactly-one-primary invariant broken") + } + if !got2.IsPrimary { + t.Errorf("s2 not primary after being promoted") + } +} diff --git a/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go index b8bb24c519..97f012e9c1 100644 --- a/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go +++ b/apps/backend/internal/task/repository/sqlite/postgres_schema_test.go @@ -2,9 +2,16 @@ package sqlite import ( "context" + "fmt" + "net/url" + "strings" + "sync" "testing" "time" + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + "github.com/kandev/kandev/internal/agentruntime" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/testutil" @@ -314,3 +321,124 @@ func TestPostgresTaskEnvironmentReposMultiBranchMigration(t *testing.T) { t.Fatal("expected duplicate env/repo/branch insert to fail") } } + +// openIsolatedPostgresMultiConn is like testutil.OpenIsolatedPostgres but +// supports a real multi-connection pool. OpenIsolatedPostgres scopes its +// isolated schema via a session-level `SET search_path` issued on one +// connection — fine for its single-connection pool, but a second pooled +// connection never sees that SET and falls back to the default "public" +// schema. Baking search_path into the DSN's libpq `options` param instead +// makes every new connection resolve unqualified table names against the +// isolated schema, so the pool can be sized for genuine concurrency tests. +func openIsolatedPostgresMultiConn(t *testing.T, dsn string, maxConns int) *sqlx.DB { + t.Helper() + schema := "kandev_test_" + strings.ReplaceAll(uuid.NewString(), "-", "") + + setup, err := sqlx.Open("pgx", dsn) + if err != nil { + t.Fatalf("open postgres (schema setup): %v", err) + } + if _, err := setup.Exec("CREATE SCHEMA " + schema); err != nil { + _ = setup.Close() + t.Fatalf("create postgres schema %s: %v", schema, err) + } + _ = setup.Close() + // Register schema-drop cleanup immediately: any t.Fatalf between here and + // the end of this function (e.g. sqlx.Open below failing) must still not + // leak the schema on the Postgres test instance. + t.Cleanup(func() { + if cleanup, cerr := sqlx.Open("pgx", dsn); cerr == nil { + _, _ = cleanup.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE") + _ = cleanup.Close() + } + }) + + // dsn is either a URL ("postgres://user:pass@host/db?sslmode=...") or a + // libpq keyword/value string ("host=... port=... sslmode=disable" — + // what CI's KANDEV_TEST_POSTGRES_DSN uses). Appending "?options=..." to + // the latter corrupts the last keyword's value instead of adding a new + // one, so the two forms need different separators. + var scopedDSN string + if strings.Contains(dsn, "://") { + sep := "?" + if strings.Contains(dsn, "?") { + sep = "&" + } + scopedDSN = dsn + sep + "options=" + url.QueryEscape("-c search_path="+schema) + } else { + scopedDSN = dsn + " options='-c search_path=" + schema + "'" + } + db, err := sqlx.Open("pgx", scopedDSN) + if err != nil { + t.Fatalf("open postgres (scoped, %d conns): %v", maxConns, err) + } + db.SetMaxOpenConns(maxConns) + db.SetMaxIdleConns(maxConns) + t.Cleanup(func() { _ = db.Close() }) + return db +} + +// TestPostgresSetSessionPrimary_ConcurrentPromotionsLeaveExactlyOnePrimary +// exercises the race Greptile flagged on PR #1860: SetSessionPrimary demotes +// every other session for a task and promotes the target in one +// transaction, plus a `SELECT ... FOR UPDATE` row lock on the owning task +// so concurrent promotions on separate Postgres connections serialize +// instead of interleaving their demote/promote pairs. Uses +// openIsolatedPostgresMultiConn (not testutil.OpenIsolatedPostgres, which +// caps the pool at one connection) because genuine cross-connection +// concurrency is the whole point — a one-connection pool would trivially +// serialize even the pre-fix code and prove nothing. +func TestPostgresSetSessionPrimary_ConcurrentPromotionsLeaveExactlyOnePrimary(t *testing.T) { + const concurrency = 8 + db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), concurrency) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + + now := time.Now().UTC() + if _, err := db.Exec(db.Rebind(` + INSERT INTO tasks (id, workspace_id, title, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + `), "task-primary-race", "ws-primary-race", "Task primary race", now, now); err != nil { + t.Fatalf("seed task: %v", err) + } + + sessionIDs := make([]string, concurrency) + for i := range sessionIDs { + sessionIDs[i] = fmt.Sprintf("session-primary-race-%d", i) + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionIDs[i], TaskID: "task-primary-race", State: models.TaskSessionStateRunning, + }); err != nil { + t.Fatalf("CreateTaskSession(%s): %v", sessionIDs[i], err) + } + } + + var wg sync.WaitGroup + errs := make([]error, concurrency) + for i, sessionID := range sessionIDs { + wg.Add(1) + go func(i int, sessionID string) { + defer wg.Done() + errs[i] = repo.SetSessionPrimary(ctx, sessionID) + }(i, sessionID) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("SetSessionPrimary(%s) concurrent call failed: %v", sessionIDs[i], err) + } + } + + var primaryCount int + if err := db.Get(&primaryCount, db.Rebind( + `SELECT COUNT(*) FROM task_sessions WHERE task_id = ? AND is_primary = 1`, + ), "task-primary-race"); err != nil { + t.Fatalf("count primary sessions: %v", err) + } + if primaryCount != 1 { + t.Errorf("expected exactly 1 primary session after %d concurrent promotions, got %d — FOR UPDATE lock not serializing across connections", concurrency, primaryCount) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index f49338d10d..1e10e4e1de 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -1914,13 +1914,32 @@ func (r *Repository) GetPrimarySessionInfoByTaskIDs(ctx context.Context, taskIDs return result, rows.Err() } -// SetSessionPrimary marks a session as primary and clears primary flag on other sessions for the same task +// SetSessionPrimary marks a session as primary and clears the primary flag +// on every other session for the same task, atomically. Both writes go +// through a single transaction so a concurrent caller can't observe (or +// write) a half-applied state — e.g. two sessions racing on is_primary=1, +// or a reader seeing zero primary sessions mid-swap. +// +// On SQLite the writer pool is a single connection (see +// internal/db.NewSQLiteDB), so only one transaction can hold it at a time — +// the transaction alone fully serializes concurrent callers. On Postgres, +// separate connections could otherwise run two of these transactions truly +// concurrently, each seeing zero primary sessions and both promoting; to +// close that window we take an exclusive row lock on the owning task +// (`SELECT ... FOR UPDATE`) before touching its sessions, so a second +// concurrent promotion for the same task blocks until the first commits. func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) error { now := time.Now().UTC() + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + // First, get the task_id for this session var taskID string - err := r.db.QueryRowContext(ctx, r.db.Rebind(`SELECT task_id FROM task_sessions WHERE id = ?`), sessionID).Scan(&taskID) + err = tx.QueryRowContext(ctx, r.db.Rebind(`SELECT task_id FROM task_sessions WHERE id = ?`), sessionID).Scan(&taskID) if err == sql.ErrNoRows { return fmt.Errorf("session not found: %s", sessionID) } @@ -1928,8 +1947,19 @@ func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) er return err } + // Serialize concurrent promotions for the same task across Postgres + // connections. SQLite has no FOR UPDATE / row-level locking and doesn't + // need it — the single-connection writer pool already serializes here. + if dialect.IsPostgres(r.db.DriverName()) { + var lockedTaskID string + err := tx.QueryRowContext(ctx, r.db.Rebind(`SELECT id FROM tasks WHERE id = ? FOR UPDATE`), taskID).Scan(&lockedTaskID) + if err != nil && err != sql.ErrNoRows { + return err + } + } + // Clear primary flag on all sessions for this task - _, err = r.db.ExecContext(ctx, r.db.Rebind(` + _, err = tx.ExecContext(ctx, r.db.Rebind(` UPDATE task_sessions SET is_primary = 0, updated_at = ? WHERE task_id = ? `), now, taskID) if err != nil { @@ -1937,7 +1967,7 @@ func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) er } // Set primary flag on the specified session - result, err := r.db.ExecContext(ctx, r.db.Rebind(` + result, err := tx.ExecContext(ctx, r.db.Rebind(` UPDATE task_sessions SET is_primary = 1, updated_at = ? WHERE id = ? `), now, sessionID) if err != nil { @@ -1948,5 +1978,5 @@ func (r *Repository) SetSessionPrimary(ctx context.Context, sessionID string) er if rows == 0 { return fmt.Errorf("session not found: %s", sessionID) } - return nil + return tx.Commit() } diff --git a/apps/web/components/automations/runs-section.test.tsx b/apps/web/components/automations/runs-section.test.tsx new file mode 100644 index 0000000000..b711528fdd --- /dev/null +++ b/apps/web/components/automations/runs-section.test.tsx @@ -0,0 +1,76 @@ +import { cleanup, fireEvent, render, screen } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { AutomationRun } from "@/lib/types/automation"; + +const mockUseAutomationRuns = vi.fn(); + +vi.mock("@/hooks/domains/settings/use-automation-runs", () => ({ + useAutomationRuns: (...args: unknown[]) => mockUseAutomationRuns(...args), +})); + +const mockPush = vi.fn(); +vi.mock("@/lib/routing/client-router", () => ({ + useRouter: () => ({ push: mockPush }), +})); + +import { RunsSection } from "./runs-section"; + +function mkRun(overrides: Partial = {}): AutomationRun { + return { + id: "run-1", + automation_id: "auto-1", + trigger_id: "trig-1", + trigger_type: "scheduled", + task_id: "task-1", + status: "task_created", + dedup_key: "", + trigger_data: {}, + error_message: "", + created_at: new Date().toISOString(), + ...overrides, + }; +} + +function setup(runs: AutomationRun[]) { + mockUseAutomationRuns.mockReturnValue({ + runs, + loading: false, + refresh: vi.fn(), + deleteRun: vi.fn(), + deleteAllRuns: vi.fn(), + }); + render(); + // The runs table only renders once the "Recent Runs" disclosure is expanded. + fireEvent.click(screen.getByText(/Recent Runs/)); +} + +describe("RunsSection status badges", () => { + afterEach(() => { + cleanup(); + vi.clearAllMocks(); + }); + + it("labels an archived-task run as Archived, not Cancelled", () => { + setup([mkRun({ id: "run-archived", status: "archived" })]); + + expect(screen.getByText("Archived")).toBeTruthy(); + expect(screen.queryByText("Cancelled")).toBeNull(); + }); + + it("labels a genuinely cancelled-task run as Cancelled", () => { + setup([mkRun({ id: "run-cancelled", status: "cancelled" })]); + + expect(screen.getByText("Cancelled")).toBeTruthy(); + expect(screen.queryByText("Archived")).toBeNull(); + }); + + it("distinguishes archived and cancelled runs side by side", () => { + setup([ + mkRun({ id: "run-archived", status: "archived" }), + mkRun({ id: "run-cancelled", status: "cancelled" }), + ]); + + expect(screen.getByText("Archived")).toBeTruthy(); + expect(screen.getByText("Cancelled")).toBeTruthy(); + }); +}); diff --git a/apps/web/components/automations/runs-section.tsx b/apps/web/components/automations/runs-section.tsx index ff19bb6bbf..bdb14c4229 100644 --- a/apps/web/components/automations/runs-section.tsx +++ b/apps/web/components/automations/runs-section.tsx @@ -37,9 +37,14 @@ const STATUS_BADGE: Record< succeeded: { variant: "default", label: "Succeeded" }, failed: { variant: "destructive", label: "Failed" }, skipped: { variant: "outline", label: "Skipped" }, - // The generating task was archived or no longer exists — its outcome is - // unknown, so this is deliberately distinct from succeeded/failed rather - // than guessing one. See internal/automation.RunStatusCancelled. + // The generating task was archived — via the UI or by the agent itself + // (e.g. an "archive this task" instruction). Distinct from a genuine + // user cancellation: archiving just closes the task out, it doesn't + // mean the run's work was rejected. See internal/automation.RunStatusArchived. + archived: { variant: "outline", label: "Archived" }, + // The generating task no longer exists, or its current primary session + // is CANCELLED — a real cancellation, distinct from archived. + // See internal/automation.RunStatusCancelled. cancelled: { variant: "outline", label: "Cancelled" }, }; diff --git a/apps/web/e2e/helpers/api-client.ts b/apps/web/e2e/helpers/api-client.ts index b11a96b0b8..ded283e8fa 100644 --- a/apps/web/e2e/helpers/api-client.ts +++ b/apps/web/e2e/helpers/api-client.ts @@ -2446,6 +2446,26 @@ export class ApiClient { task_id: taskId ?? "", }); } + + /** + * Seed a task_sessions row directly via the E2E HTTP endpoint — + * deterministic state seeding, not a substitute for driving a real + * agent stop/resume cycle. Used to assert on session-state-derived + * behavior (e.g. the automation Recent Runs "Cancelled" status, which + * is keyed off the task's *primary* session state, not the task's own + * state). Only works when KANDEV_MOCK_AGENT is active. + */ + async seedAutomationTaskSession( + taskId: string, + state: string, + isPrimary = true, + ): Promise<{ id: string; task_id: string; status: string }> { + return this.request("POST", "/api/v1/e2e/task-sessions", { + task_id: taskId, + state, + is_primary: isPrimary, + }); + } } // --- Jira / Linear mock payload types --- diff --git a/apps/web/e2e/tests/automations-settings.spec.ts b/apps/web/e2e/tests/automations-settings.spec.ts index 077d934837..e39ecb6ce5 100644 --- a/apps/web/e2e/tests/automations-settings.spec.ts +++ b/apps/web/e2e/tests/automations-settings.spec.ts @@ -358,17 +358,21 @@ test.describe("Automations settings page", () => { await expect(testPage.getByText("No runs yet")).toBeVisible({ timeout: 5_000 }); }); - test("archived task's run shows Cancelled instead of Running", async ({ + test("archived task's run shows Archived, cancelled task's run shows Cancelled", async ({ testPage, seedData, apiClient, }) => { // Regression test: an automation-generated task that gets archived - // (manually, via auto-archive, or via cascade) before its run is - // otherwise finalized used to leave the run stuck at "task_created" - // forever, displayed as "Running" and permanently pinned against - // max_concurrent_runs. See internal/automation.Store.CountActiveRuns / - // ListRuns. + // (manually, via auto-archive, via cascade, or by the agent itself) + // before its run is otherwise finalized used to leave the run stuck at + // "task_created" forever, displayed as "Running" and permanently + // pinned against max_concurrent_runs. It also used to be + // indistinguishable from a genuine user cancellation (the task's + // primary session state going CANCELLED — the signal a real Stop + // leaves behind; the task's own state is untouched by a stop), both + // showing as "Cancelled". See internal/automation.Store.CountActiveRuns + // / ListRuns. const automation = await apiClient.seedAutomation({ workspaceId: seedData.workspaceId, name: "Archived Task Run Test", @@ -381,13 +385,20 @@ test.describe("Automations settings page", () => { "Archived Automation Task", { workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId }, ); + const cancelledTask = await apiClient.createTask( + seedData.workspaceId, + "Cancelled Automation Task", + { workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId }, + ); const openTask = await apiClient.createTask(seedData.workspaceId, "Open Automation Task", { workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId, }); await apiClient.seedAutomationRun(automation.id, "task_created", archivedTask.id); + await apiClient.seedAutomationRun(automation.id, "task_created", cancelledTask.id); await apiClient.seedAutomationRun(automation.id, "task_created", openTask.id); await apiClient.archiveTask(archivedTask.id); + await apiClient.seedAutomationTaskSession(cancelledTask.id, "CANCELLED"); await testPage.goto(`/settings/workspace/${seedData.workspaceId}/automations/${automation.id}`); await testPage.getByTestId("automation-editor").waitFor({ state: "visible", timeout: 15_000 }); @@ -401,13 +412,22 @@ test.describe("Automations settings page", () => { const tbody = testPage.locator("table tbody"); await tbody.waitFor({ state: "visible", timeout: 5_000 }); - await expect(tbody.locator("tr")).toHaveCount(2, { timeout: 10_000 }); + await expect(tbody.locator("tr")).toHaveCount(3, { timeout: 10_000 }); - // The archived task's run is no longer outstanding work. + // The archived task's run is no longer outstanding work, and reads + // "Archived" rather than being conflated with a real cancellation. const archivedRow = testPage.locator("table tbody tr", { hasText: archivedTask.id.slice(0, 8), }); - await expect(archivedRow.getByText("Cancelled", { exact: true })).toBeVisible(); + await expect(archivedRow.getByText("Archived", { exact: true })).toBeVisible(); + + // A task whose current (primary) session was genuinely cancelled — + // the signal a real Stop leaves behind — reads "Cancelled", distinct + // from "Archived". + const cancelledRow = testPage.locator("table tbody tr", { + hasText: cancelledTask.id.slice(0, 8), + }); + await expect(cancelledRow.getByText("Cancelled", { exact: true })).toBeVisible(); // The still-open task's run is unaffected. const openRow = testPage.locator("table tbody tr", { hasText: openTask.id.slice(0, 8) }); diff --git a/apps/web/lib/types/automation.ts b/apps/web/lib/types/automation.ts index e978c5514c..2bbaddf3ea 100644 --- a/apps/web/lib/types/automation.ts +++ b/apps/web/lib/types/automation.ts @@ -8,6 +8,7 @@ export type RunStatus = | "succeeded" | "failed" | "skipped" + | "archived" | "cancelled"; // ExecutionMode controls whether an automation firing creates a visible