diff --git a/AGENTS.md b/AGENTS.md index f14087d87d..c39304b3ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -174,6 +174,7 @@ history and remains immutable. - ADR 0015's step-completion-signal telemetry lives under `workflow_*`: `workflow_step_completion_signal_received_total` (`internal/workflow/signalmetrics/`), labelled by `source` and `agent_type`, counts accepted `step_complete_kandev` signals. Its separate `workflow_step_completion_signal_fallback_used_total` counter is labelled by `agent_type` and counts manual fallback uses. The fallback button does not have a production increment site yet, so the counter remains zero until that UI ships. - Office stall detection (REQ-OFFICE-STALL-VISIBILITY) lives under `office_stall_*`: `office_stall_stranded_signal_total` (labelled by `gate`, which names which of the two watchdog gate sites saw it), `office_stall_decision_waiting_total`, and `office_stall_detector_skipped_total` (labelled by `reason`, so a detector that fails closed is visible rather than silent). Detection only: these counters never accompany a transition, a synthesized decision, or a queued run, because Office tasks are surfaced and never reclaimed. Also emitted as structured zap logs. - Office session-termination suppression (REQ-OFFICE-SESSION-TERM) lives under `office_session_term_suppressed_total`, labelled by `reason` (one of `task_reassigned`, `participant_removed`, `participant_seat_claimed`) and `outcome` (`runner` or `seat` when the agent still holds that capacity, `read_failed` when the capacity read itself failed). It counts guarded terminations that left a session live, so a suppression is distinguishable from a path that never ran. Both dimensions are closed sets; no task, agent, step or session identifier is ever a label. Also emitted as structured zap logs. +- Task-level stall detection (issue #3712) runs in the session reconciliation sweep (`internal/task/service/`, one-minute tick shared with the archived-session pass): an unarchived task holding an active session with no live execution behind it and no events/messages past `tasks.stallDetectionThreshold` (default 2h) emits a `task.stalled` event plus a `stalled_task` WARN, once per stall episode. At or beyond twice the threshold the sweep cancels the orphaned sessions (`error_message` = "orphaned session") through the archived pass's `finalizeCancelledSessions` transition scoped to the classified session IDs; the heal runs only when every active session of the task is orphaned and past the grace window (a live or merely-stalled sibling cancels nothing), and liveness is re-checked at the cancellation boundary. A stall past the grace window can therefore emit `task.stalled` and a `session.state_changed` in the same sweep; the cancel never depends on the detection event. ### GitHub Operations diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index 6759d7e1e3..b79b8bfc11 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/apps/backend/internal/backendapp/adapters.go @@ -960,6 +960,24 @@ func (a *lifecycleAdapter) ListExecutionsForTask(taskID string) []lifecycle.Exec return a.mgr.ListExecutionsForTask(taskID) } +// LiveSessionIDsForTask satisfies taskservice.SessionExecutionRegistry: the +// session IDs under taskID that currently have a live in-memory execution +// registered by the agent runtime. The session reconciliation sweep uses the +// snapshot to tell an active session whose backing actor is alive from one +// whose actor is gone. Pinned here so a signature drift is a build error. +var _ taskservice.SessionExecutionRegistry = (*lifecycleAdapter)(nil) + +func (a *lifecycleAdapter) LiveSessionIDsForTask(taskID string) []string { + references := a.mgr.ListExecutionsForTask(taskID) + sessionIDs := make([]string, 0, len(references)) + for _, reference := range references { + if reference.SessionID != "" { + sessionIDs = append(sessionIDs, reference.SessionID) + } + } + return sessionIDs +} + func (a *lifecycleAdapter) GetRemoteRuntimeStatusBySession(ctx context.Context, sessionID string) (*executor.RemoteRuntimeStatus, error) { status, ok := a.mgr.GetRemoteStatusBySessionID(ctx, sessionID) if !ok || status == nil { diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 600fbbf08a..daa9a1250d 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -1199,7 +1199,8 @@ func startGatewayAndServe( } services.Task.StartAutoArchiveLoop(ctx) - services.Task.StartArchivedSessionReconciliationLoop(ctx) + services.Task.SetStallDetectionThreshold(cfg.Tasks.StallDetectionThreshold) + services.Task.StartSessionReconciliationLoop(ctx) services.Task.StartQuickChatExpirationLoop(ctx) // ============================================ diff --git a/apps/backend/internal/backendapp/orchestrator.go b/apps/backend/internal/backendapp/orchestrator.go index be42eb4373..e1e17effba 100644 --- a/apps/backend/internal/backendapp/orchestrator.go +++ b/apps/backend/internal/backendapp/orchestrator.go @@ -170,6 +170,10 @@ func provideOrchestrator( // Runtime-aware liveness lets durable cleanup treat a not-found stop for a // confirmed-dead local runtime as already stopped instead of retrying forever. taskSvc.SetRowLivenessProber(agentManagerClient) + // The session reconciliation sweep's active-task pass (stall detection and + // orphaned-session healing) verifies "no live execution" against the agent + // runtime's in-memory execution store through this registry. + taskSvc.SetSessionExecutionRegistry(agentManagerClient) taskSvc.SetContextWindowResetter(orchestratorSvc.ResetContextWindow) taskSvc.SetGitArchiveCapture(orchestratorSvc) // Automation runs keep their worktrees so they stay repliable, which makes diff --git a/apps/backend/internal/common/config/catalog.go b/apps/backend/internal/common/config/catalog.go index a6af5add7a..6e006e312a 100644 --- a/apps/backend/internal/common/config/catalog.go +++ b/apps/backend/internal/common/config/catalog.go @@ -92,6 +92,7 @@ var startupCatalog = []CatalogEntry{ {Key: "office.schedulerTickMs", EnvVars: []string{"KANDEV_OFFICE_SCHEDULER_TICK_MS"}, Owner: "office", Default: "5000"}, {Key: "githubCredentialBroker.publicBaseUrl", EnvVars: []string{"KANDEV_GITHUB_CREDENTIAL_BROKER_PUBLIC_BASE_URL"}, Owner: "github credential broker", Default: ""}, {Key: "tasks.preparationTimeout", EnvVars: []string{"KANDEV_TASK_PREPARATION_TIMEOUT"}, Owner: "task lifecycle", Default: "10m"}, + {Key: "tasks.stallDetectionThreshold", EnvVars: []string{"KANDEV_TASK_STALL_DETECTION_THRESHOLD"}, Owner: "task lifecycle", Default: "2h"}, {Key: "credentials.file", EnvVars: []string{"KANDEV_CREDENTIALS_FILE"}, Owner: "credentials", Default: ""}, {Key: "limits.ghMaxConcurrent", EnvVars: []string{"KANDEV_GH_MAX_CONCURRENT"}, Owner: "subprocess admission", Default: "8"}, {Key: "limits.gitMaxConcurrent", EnvVars: []string{"KANDEV_GIT_MAX_CONCURRENT"}, Owner: "subprocess admission", Default: "12"}, diff --git a/apps/backend/internal/common/config/catalog_test.go b/apps/backend/internal/common/config/catalog_test.go index f9087f859c..6ce878b407 100644 --- a/apps/backend/internal/common/config/catalog_test.go +++ b/apps/backend/internal/common/config/catalog_test.go @@ -115,6 +115,7 @@ func auditedStartupEnvironmentInventory() []auditedStartupEnvironment { {envVar: "KANDEV_OFFICE_SCHEDULER_TICK_MS", class: "catalog"}, {envVar: "KANDEV_GITHUB_CREDENTIAL_BROKER_PUBLIC_BASE_URL", class: "catalog"}, {envVar: "KANDEV_TASK_PREPARATION_TIMEOUT", class: "catalog"}, + {envVar: "KANDEV_TASK_STALL_DETECTION_THRESHOLD", class: "catalog"}, {envVar: "KANDEV_CREDENTIALS_FILE", class: "catalog"}, {envVar: "KANDEV_GH_MAX_CONCURRENT", class: "catalog"}, {envVar: "KANDEV_GIT_MAX_CONCURRENT", class: "catalog"}, diff --git a/apps/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go index f307932515..0775967ee4 100644 --- a/apps/backend/internal/common/config/config.go +++ b/apps/backend/internal/common/config/config.go @@ -59,6 +59,11 @@ type Config struct { // TasksConfig contains task lifecycle startup limits. type TasksConfig struct { PreparationTimeout time.Duration `mapstructure:"preparationTimeout"` + // StallDetectionThreshold is the event-silence window after which the + // session reconciliation sweep classifies an active session with no live + // execution as stalled (issue #3712). The orphaned-session healing grace + // window is twice this value. + StallDetectionThreshold time.Duration `mapstructure:"stallDetectionThreshold"` } // CredentialsConfig contains operator-managed credential file settings. diff --git a/apps/backend/internal/common/config/source.go b/apps/backend/internal/common/config/source.go index caa33af107..aea8ea96f1 100644 --- a/apps/backend/internal/common/config/source.go +++ b/apps/backend/internal/common/config/source.go @@ -67,6 +67,7 @@ type configSelection struct { var yamlOnlyStartupKeys = map[string]struct{}{ "server.trustedProxies": {}, "tasks.preparationTimeout": {}, + "tasks.stallDetectionThreshold": {}, "credentials.file": {}, "limits.ghMaxConcurrent": {}, "limits.gitMaxConcurrent": {}, @@ -286,6 +287,7 @@ func applyStartupDefaults(cfg *Config, yamlKeys map[string]bool, profileDefaults cfg.Server.TrustedProxies = nil } setDefaultDuration("tasks.preparationTimeout", &cfg.Tasks.PreparationTimeout, 10*time.Minute) + setDefaultDuration("tasks.stallDetectionThreshold", &cfg.Tasks.StallDetectionThreshold, 2*time.Hour) if !yamlKeys["credentials.file"] { cfg.Credentials.File = "" } @@ -316,6 +318,7 @@ func applyStartupDefaults(cfg *Config, yamlKeys map[string]bool, profileDefaults func applyStartupEnvironment(cfg *Config, envSnapshot map[string]string, sources map[string]SettingSource) { applyTrustedProxiesEnv(cfg, envSnapshot, sources) applyDurationEnv("tasks.preparationTimeout", &cfg.Tasks.PreparationTimeout, 10*time.Minute, envSnapshot, sources) + applyDurationEnv("tasks.stallDetectionThreshold", &cfg.Tasks.StallDetectionThreshold, 2*time.Hour, envSnapshot, sources) applyStringEnvAllowEmpty("credentials.file", &cfg.Credentials.File, envSnapshot, sources) applyPositiveIntEnv("limits.ghMaxConcurrent", &cfg.Limits.GHMaxConcurrent, 8, envSnapshot, sources) applyPositiveIntEnv("limits.gitMaxConcurrent", &cfg.Limits.GitMaxConcurrent, 12, envSnapshot, sources) diff --git a/apps/backend/internal/common/config/validation.go b/apps/backend/internal/common/config/validation.go index e4fdc8b4db..1f08f4305f 100644 --- a/apps/backend/internal/common/config/validation.go +++ b/apps/backend/internal/common/config/validation.go @@ -10,6 +10,7 @@ import ( func validateStartupSettings(cfg *Config) error { errs := validateTrustedProxies(cfg) errs = append(errs, configuredValidation(cfg, "tasks.preparationTimeout", cfg.Tasks.PreparationTimeout <= 0, "tasks.preparationTimeout must be positive")...) + errs = append(errs, configuredValidation(cfg, "tasks.stallDetectionThreshold", cfg.Tasks.StallDetectionThreshold <= 0, "tasks.stallDetectionThreshold must be positive")...) errs = append(errs, configuredValidation(cfg, "limits.ghMaxConcurrent", cfg.Limits.GHMaxConcurrent <= 0, "limits.ghMaxConcurrent must be positive")...) errs = append(errs, configuredValidation(cfg, "limits.gitMaxConcurrent", cfg.Limits.GitMaxConcurrent <= 0, "limits.gitMaxConcurrent must be positive")...) errs = append(errs, configuredValidation(cfg, "limits.lspMaxConnections", cfg.Limits.LSPMaxConnections <= 0, "limits.lspMaxConnections must be positive")...) diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 5c612c26b7..d43d3c9a05 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -23,6 +23,13 @@ const ( // predecessor failed or was cancelled. Payload: // {task_id, failed_task_id, failed_state}. TaskDependencyFailed = "task.dependency_failed" + // TaskStalled fires when the session reconciliation sweep observes a task + // holding an active session with no live execution behind it and no + // session events or messages for longer than the stall threshold. + // Detection only: the event never accompanies a state transition, a + // synthesized decision, or a queued run. Payload: + // {task_id, workspace_id, session_ids, stalled_for, last_event_at}. + TaskStalled = "task.stalled" ) // Event types for plugin-backed canvas lifecycle changes. Payloads contain diff --git a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go index 55bdfbc7eb..da53feb6c6 100644 --- a/apps/backend/internal/orchestrator/executor/executor_mocks_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_mocks_test.go @@ -805,6 +805,9 @@ func (m *mockRepository) ListTasksForAutoArchive(ctx context.Context) ([]*models func (m *mockRepository) ListArchivedTasksWithActiveSessions(ctx context.Context) ([]string, error) { return nil, nil } +func (m *mockRepository) ListUnarchivedTasksWithActiveSessions(ctx context.Context) ([]*models.Task, error) { + return nil, nil +} func (m *mockRepository) GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) { return nil, nil diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 380315abfc..1461399a80 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -124,6 +124,12 @@ func (m *mockRepository) ListTasksForAutoArchive(ctx context.Context) ([]*models func (m *mockRepository) ListArchivedTasksWithActiveSessions(ctx context.Context) ([]string, error) { return nil, nil } +func (m *mockRepository) ListUnarchivedTasksWithActiveSessions(ctx context.Context) ([]*models.Task, error) { + return nil, nil +} +func (m *mockRepository) GetLastMessageTimeBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]time.Time, error) { + return nil, nil +} func (m *mockRepository) ListExpiredQuickChatTasks(ctx context.Context, cutoff time.Time) ([]*models.Task, error) { return nil, nil } @@ -467,6 +473,12 @@ func (m *mockRepository) ListLiveWorkspaceSessions(ctx context.Context) ([]*mode func (m *mockRepository) CancelActiveTaskSessionsByTaskID(ctx context.Context, taskID, reason string) ([]*models.TaskSession, error) { return nil, nil } +func (m *mockRepository) CancelActiveTaskSessionsByIDs(ctx context.Context, taskID string, sessionIDs []string, reason string) ([]*models.TaskSession, error) { + return nil, nil +} +func (m *mockRepository) CancelActiveTaskSessionsByCandidates(ctx context.Context, taskID string, candidates []models.ActiveSessionCancellationCandidate, reason string) ([]*models.TaskSession, error) { + return nil, nil +} func (m *mockRepository) HasActiveTaskSessionsByAgentProfile(ctx context.Context, agentProfileID string) (bool, error) { return false, nil } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index a95f6a22cc..404a58bd77 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -1893,6 +1893,16 @@ type TaskSession struct { TokensOut int64 `json:"tokens_out"` } +// ActiveSessionCancellationCandidate is the compare-and-set snapshot used by +// the active-session stall healer. An empty ExpectedTurnID means that no turn +// may be active when the cancellation is written. +type ActiveSessionCancellationCandidate struct { + SessionID string + ExpectedUpdatedAt time.Time + ExpectedLastEventAt time.Time + ExpectedTurnID string +} + // ToAPI converts internal TaskSession to API type // TODO: Add v1.TaskSession type to pkg/api/v1/ func (s *TaskSession) ToAPI() map[string]interface{} { diff --git a/apps/backend/internal/task/models/resume_safety.go b/apps/backend/internal/task/models/resume_safety.go index ab36dee976..3e06290755 100644 --- a/apps/backend/internal/task/models/resume_safety.go +++ b/apps/backend/internal/task/models/resume_safety.go @@ -81,6 +81,14 @@ const ( SessionArchiveTreeCancelReason = "task tree archived" ) +// SessionOrphanedCancelReason is the TaskSession.ErrorMessage value written +// when the session reconciliation sweep cancels an active session that no +// longer has a live execution behind it (e.g. after a backend restart) and +// has been event-silent beyond the sweep's grace window. It is deliberately +// not an archive reason: unarchived tasks must treat the cancelled session +// like an explicit stop, not like archive cancellation awaiting unarchive. +const SessionOrphanedCancelReason = "orphaned session" + // IsArchiveCancelReason reports whether reason came from an archive path. func IsArchiveCancelReason(reason string) bool { return reason == SessionArchiveCancelReason || reason == SessionArchiveTreeCancelReason diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index b5345ff87d..da109949aa 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -95,6 +95,14 @@ type TaskRepository interface { // Returns whether the row was updated. UnarchiveTask(ctx context.Context, id string) (bool, error) ListTasksForAutoArchive(ctx context.Context) ([]*models.Task, error) + // ListUnarchivedTasksWithActiveSessions returns the unarchived tasks + // (archived_at IS NULL) that still have at least one task_sessions row + // in an active DB state (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT). + // This is the candidate list for the session reconciliation sweep's + // active-task pass: unarchived tasks holding active sessions whose + // backing execution may be gone (e.g. after a backend restart). The + // archived counterpart is ListArchivedTasksWithActiveSessions. + ListUnarchivedTasksWithActiveSessions(ctx context.Context) ([]*models.Task, error) // ListArchivedTasksWithActiveSessions returns the IDs of archived tasks // (archived_at IS NOT NULL) that still have at least one task_sessions // row in an active DB state (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT). @@ -284,6 +292,12 @@ type WorkflowRepository interface { type MessageRepository interface { CreateMessage(ctx context.Context, message *models.Message) error GetMessage(ctx context.Context, id string) (*models.Message, error) + // GetLastMessageTimeBySessionIDs returns the newest task_session_messages + // updated_at for each requested session, in one chunked query. Sessions + // with no messages are absent from the result; callers fall back to the + // session row's own timestamps. Used by the session reconciliation sweep + // to measure per-session event silence. + GetLastMessageTimeBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]time.Time, error) // HasUserPromptHistory reports whether the session has ever accepted a user // prompt. The durable prompt sequence remains after message deletion. HasUserPromptHistory(ctx context.Context, sessionID string) (bool, error) @@ -454,6 +468,21 @@ type SessionRepository interface { // behavior by picking up IDLE sessions. ListLiveWorkspaceSessions(ctx context.Context) ([]*models.TaskSession, error) CancelActiveTaskSessionsByTaskID(ctx context.Context, taskID, reason string) ([]*models.TaskSession, error) + // CancelActiveTaskSessionsByIDs transitions exactly the listed active + // sessions (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT) to CANCELLED, + // returning the full row of each session actually transitioned. It is + // the session-scoped counterpart of CancelActiveTaskSessionsByTaskID: + // sessions outside the ID list — including ones that became active + // after the caller classified its set — are never touched. Callers that + // classified a stale or partial snapshot use it so a mid-sweep + // registration of new live work cannot be cancelled by a bulk + // task-scoped write. Same RETURNING contract as the task-scoped method. + CancelActiveTaskSessionsByIDs(ctx context.Context, taskID string, sessionIDs []string, reason string) ([]*models.TaskSession, error) + // ActiveSessionCancellationCandidate captures the activity and current-turn + // identity observed by a reconciliation pass. Implementations must cancel a + // candidate only when the session row, message activity clock, and active + // turn still match this snapshot at the write boundary. + CancelActiveTaskSessionsByCandidates(ctx context.Context, taskID string, candidates []models.ActiveSessionCancellationCandidate, reason string) ([]*models.TaskSession, 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/message.go b/apps/backend/internal/task/repository/sqlite/message.go index 6a8f9a13e7..9313b6864b 100644 --- a/apps/backend/internal/task/repository/sqlite/message.go +++ b/apps/backend/internal/task/repository/sqlite/message.go @@ -19,6 +19,54 @@ import ( // Message operations +// GetLastMessageTimeBySessionIDs returns the newest task_session_messages +// updated_at for each requested session, one chunked query over +// idx_messages_session_updated. Sessions with no messages are absent from +// the result; callers fall back to the session row's own timestamps. The +// session reconciliation sweep uses this to measure per-session event +// silence without loading any transcript content. +func (r *Repository) GetLastMessageTimeBySessionIDs(ctx context.Context, sessionIDs []string) (map[string]time.Time, error) { + result := make(map[string]time.Time, len(sessionIDs)) + if len(sessionIDs) == 0 { + return result, nil + } + + for _, chunk := range chunkIDs(sessionIDs, sqliteMaxHostParams) { + placeholders, ids := buildInPlaceholders(chunk) + query := ` + SELECT task_session_id, MAX(updated_at) + FROM task_session_messages + WHERE task_session_id IN (` + placeholders + `) + GROUP BY task_session_id` + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), ids...) + if err != nil { + return nil, fmt.Errorf("load last message time by session: %w", err) + } + for rows.Next() { + var ( + sessionID string + rawLastMessage interface{} + ) + if err := rows.Scan(&sessionID, &rawLastMessage); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("scan last message time by session: %w", err) + } + lastMessage, err := parseTaskActivityTime(rawLastMessage) + if err != nil { + _ = rows.Close() + return nil, fmt.Errorf("parse last message time for session %q: %w", sessionID, err) + } + result[sessionID] = lastMessage + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, fmt.Errorf("read last message time by session: %w", err) + } + _ = rows.Close() + } + return result, nil +} + // CreateMessage creates a new message func (r *Repository) CreateMessage(ctx context.Context, message *models.Message) error { if message.ID == "" { diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 850ec659f3..15726f9f5e 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -103,7 +103,11 @@ func stampTurnDefaults(turn *models.Turn) { // insertTurnRow inserts turn's row via execer, which is either r.db (a plain, // non-transactional insert) or a *sql.Tx (participating in the caller's -// transaction). +// transaction). A new turn is session activity, so the same statement batch +// also refreshes task_sessions.updated_at: the session reconciliation sweep +// measures event silence from that row (and the newest message), and a turn +// dispatched without a user message (auto-start, workflow on_enter) must +// reset the clock even though no message row exists yet. func (r *Repository) insertTurnRow(ctx context.Context, execer taskSessionExecutor, turn *models.Turn) error { metadataJSON := "{}" if turn.Metadata != nil { @@ -114,20 +118,23 @@ func (r *Repository) insertTurnRow(ctx context.Context, execer taskSessionExecut metadataJSON = string(metadataBytes) } - _, err := execer.ExecContext(ctx, r.db.Rebind(` + if _, err := execer.ExecContext(ctx, r.db.Rebind(` INSERT INTO task_session_turns (id, task_session_id, task_id, execution_profile_id, route_generation, started_at, completed_at, metadata, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `), turn.ID, turn.TaskSessionID, turn.TaskID, turn.ExecutionProfileID, turn.RouteGeneration, turn.StartedAt, turn.CompletedAt, metadataJSON, turn.CreatedAt, turn.UpdatedAt) + `), turn.ID, turn.TaskSessionID, turn.TaskID, turn.ExecutionProfileID, turn.RouteGeneration, turn.StartedAt, turn.CompletedAt, metadataJSON, turn.CreatedAt, turn.UpdatedAt); err != nil { + return err + } + _, err := execer.ExecContext(ctx, r.db.Rebind(` + UPDATE task_sessions SET updated_at = ? WHERE id = ? + `), turn.UpdatedAt, turn.TaskSessionID) return err } // insertTurnWithSessionLock serializes successor-turn creation with every -// current-turn clarification decision on PostgreSQL. SQLite's writer pool -// already provides the equivalent serialization. +// current-turn clarification decision on every database. The transaction is +// also required for SQLite: inserting a turn and refreshing its parent +// session's activity clock must commit or roll back as one unit. func (r *Repository) insertTurnWithSessionLock(ctx context.Context, turn *models.Turn) error { - if !dialect.IsPostgres(r.db.DriverName()) { - return r.insertTurnRow(ctx, r.db, turn) - } tx, err := r.db.BeginTxx(ctx, nil) if err != nil { return fmt.Errorf("begin turn creation: %w", err) @@ -2142,6 +2149,157 @@ func (r *Repository) CancelActiveTaskSessionsByTaskID(ctx context.Context, taskI return sessions, rows.Err() } +// CancelActiveTaskSessionsByIDs is documented on the SessionRepository +// interface. It shares CancelActiveTaskSessionsByTaskID's atomic +// UPDATE ... RETURNING shape and detached-but-bounded write context, so the +// returned rows are exactly the sessions this call transitioned, and a +// client disconnect mid-write cannot lose the committed cancellation. +// Chunking keeps the placeholder count below the host-parameter limit; the +// active set of one task is far below it, so multi-chunk calls are a +// correctness backstop, not an expected path. +func (r *Repository) CancelActiveTaskSessionsByIDs(ctx context.Context, taskID string, sessionIDs []string, reason string) ([]*models.TaskSession, error) { + if len(sessionIDs) == 0 { + return nil, nil + } + now := time.Now().UTC() + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + var sessions []*models.TaskSession + for _, chunk := range chunkIDs(sessionIDs, sqliteMaxHostParams) { + placeholders, args := buildInPlaceholders(chunk) + args = append([]interface{}{string(models.TaskSessionStateCancelled), reason, now, now, taskID}, args...) + rows, err := r.db.QueryContext(writeCtx, r.db.Rebind(` + UPDATE task_sessions + SET state = ?, error_message = ?, completed_at = ?, updated_at = ? + WHERE task_id = ? + AND id IN (`+placeholders+`) + AND state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT') + RETURNING id, agent_profile_id, agent_profile_snapshot, is_passthrough, name, + review_status, metadata, task_environment_id, state, updated_at + `), args...) + if err != nil { + return nil, err + } + for rows.Next() { + session, err := scanCancelledTaskSessionRow(rows, taskID) + if err != nil { + _ = rows.Close() + return nil, err + } + sessions = append(sessions, session) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + } + return sessions, nil +} + +// CancelActiveTaskSessionsByCandidates cancels only candidates whose session +// activity and current-turn identity are unchanged since classification. The +// compare-and-set predicates run in the same UPDATE that transitions the row, +// so a resumed session, a newer message, or a successor turn wins the race and +// remains active for the next reconciliation pass. +func (r *Repository) CancelActiveTaskSessionsByCandidates( + ctx context.Context, + taskID string, + candidates []models.ActiveSessionCancellationCandidate, + reason string, +) ([]*models.TaskSession, error) { + if len(candidates) == 0 { + return nil, nil + } + now := time.Now().UTC() + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + // Each candidate contributes at most five bind parameters. Keep a margin + // below SQLite's host-parameter limit and use the same bounded-chunk shape + // as the ID-scoped cancellation path. + chunkSize := sqliteMaxHostParams / 5 + if chunkSize < 1 { + chunkSize = 1 + } + var sessions []*models.TaskSession + for start := 0; start < len(candidates); start += chunkSize { + end := start + chunkSize + if end > len(candidates) { + end = len(candidates) + } + chunk := candidates[start:end] + predicates := make([]string, 0, len(chunk)) + predicateArgs := make([]interface{}, 0, len(chunk)*5) + for _, candidate := range chunk { + predicate, args := activeSessionCancellationCandidatePredicate(candidate) + predicates = append(predicates, predicate) + predicateArgs = append(predicateArgs, args...) + } + args := []interface{}{string(models.TaskSessionStateCancelled), reason, now, now, taskID} + args = append(args, predicateArgs...) + rows, err := r.db.QueryContext(writeCtx, 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') + AND (`+strings.Join(predicates, " OR ")+`) + RETURNING id, agent_profile_id, agent_profile_snapshot, is_passthrough, name, + review_status, metadata, task_environment_id, state, updated_at + `), args...) + if err != nil { + return nil, err + } + for rows.Next() { + session, err := scanCancelledTaskSessionRow(rows, taskID) + if err != nil { + _ = rows.Close() + return nil, err + } + sessions = append(sessions, session) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + } + return sessions, nil +} + +func activeSessionCancellationCandidatePredicate( + candidate models.ActiveSessionCancellationCandidate, +) (string, []interface{}) { + predicate := "(id = ? AND updated_at = ?" + args := []interface{}{candidate.SessionID, candidate.ExpectedUpdatedAt} + if !candidate.ExpectedLastEventAt.IsZero() { + predicate += ` + AND NOT EXISTS ( + SELECT 1 FROM task_session_messages newer_message + WHERE newer_message.task_session_id = task_sessions.id + AND newer_message.updated_at > ? + )` + args = append(args, candidate.ExpectedLastEventAt) + } + if candidate.ExpectedTurnID == "" { + predicate += ` + AND NOT EXISTS ( + SELECT 1 FROM task_session_turns active_turn + WHERE active_turn.task_session_id = task_sessions.id + AND active_turn.completed_at IS NULL + )` + } else { + predicate += ` + AND EXISTS ( + SELECT 1 FROM task_session_turns active_turn + WHERE active_turn.id = ? + AND active_turn.task_session_id = task_sessions.id + AND active_turn.completed_at IS NULL + )` + args = append(args, candidate.ExpectedTurnID) + } + return predicate + ")", args +} + // scanCancelledTaskSessionRow scans one row produced by // CancelActiveTaskSessionsByTaskID's UPDATE ... RETURNING into a // *models.TaskSession, mirroring scanTaskSessionRow's JSON-unmarshal and diff --git a/apps/backend/internal/task/repository/sqlite/session_test.go b/apps/backend/internal/task/repository/sqlite/session_test.go index 9e46345fc0..1c60c61d60 100644 --- a/apps/backend/internal/task/repository/sqlite/session_test.go +++ b/apps/backend/internal/task/repository/sqlite/session_test.go @@ -2187,6 +2187,105 @@ func TestCancelActiveTaskSessionsByTaskID(t *testing.T) { } } +// TestCancelActiveTaskSessionsByIDs verifies the session-scoped cancel: only +// the listed still-active sessions of the target task transition to +// CANCELLED. Active siblings outside the ID list — including ones that +// became active after the caller classified its set — and other tasks' +// sessions survive, which is the structural guard the reconciliation +// sweep's heal path relies on. +func TestCancelActiveTaskSessionsByIDs(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + + 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", "COMPLETED") + seedRepoLink(t, repo, "ws-r", "repo-other", "task-other", "sess-o1", "RUNNING") + + sessions, err := repo.CancelActiveTaskSessionsByIDs(ctx, "task-r", []string{"sess-r1", "sess-r3", "sess-r4", "sess-o1"}, "orphaned session") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + var gotIDs []string + for _, sess := range sessions { + gotIDs = append(gotIDs, sess.ID) + if sess.TaskID != "task-r" { + t.Errorf("session %s TaskID = %q, want task-r", sess.ID, sess.TaskID) + } + if sess.State != models.TaskSessionStateCancelled { + t.Errorf("session %s State = %q, want CANCELLED", sess.ID, sess.State) + } + } + if !sameStringSet(gotIDs, []string{"sess-r1", "sess-r3"}) { + t.Errorf("cancelled ids = %v, want [sess-r1 sess-r3]", gotIDs) + } + // The unlisted active sibling keeps running: the write's predicate is + // the ID list, not the task. + if got := sessionState(t, repo, "sess-r2"); got != "RUNNING" { + t.Errorf("sess-r2 (unlisted) = %q, want unchanged RUNNING", got) + } + // Other tasks' sessions listed by mistake are never touched. + if got := sessionState(t, repo, "sess-o1"); got != "RUNNING" { + t.Errorf("sess-o1 (other task) = %q, want unchanged RUNNING", got) + } + + // Idempotent: repeating the same scope changes nothing. + sessions, err = repo.CancelActiveTaskSessionsByIDs(ctx, "task-r", []string{"sess-r1", "sess-r3"}, "orphaned session") + if err != nil { + t.Fatalf("unexpected error on second call: %v", err) + } + if len(sessions) != 0 { + t.Errorf("expected 0 sessions on idempotent re-run, got %v", sessions) + } +} + +func TestCancelActiveTaskSessionsByCandidatesRejectsNewActivityAndTurn(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedRepoLink(t, repo, "ws-candidate", "repo-candidate", "task-candidate", "session-candidate", "RUNNING") + + turn := &models.Turn{ + ID: "turn-candidate", + TaskSessionID: "session-candidate", + TaskID: "task-candidate", + } + if err := repo.CreateTurn(ctx, turn); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + session, err := repo.GetTaskSession(ctx, "session-candidate") + if err != nil { + t.Fatalf("GetTaskSession after turn: %v", err) + } + candidate := models.ActiveSessionCancellationCandidate{ + SessionID: session.ID, + ExpectedUpdatedAt: session.UpdatedAt, + ExpectedLastEventAt: session.UpdatedAt, + ExpectedTurnID: turn.ID, + } + + // A successor turn refreshes the session activity clock and changes the + // current-turn identity. The old candidate must lose the write race. + newTurn := &models.Turn{ + ID: "turn-candidate-successor", + TaskSessionID: "session-candidate", + TaskID: "task-candidate", + } + if err := repo.CreateTurn(ctx, newTurn); err != nil { + t.Fatalf("Create successor turn: %v", err) + } + cancelled, err := repo.CancelActiveTaskSessionsByCandidates(ctx, "task-candidate", []models.ActiveSessionCancellationCandidate{candidate}, "orphaned session") + if err != nil { + t.Fatalf("guarded cancellation: %v", err) + } + if len(cancelled) != 0 { + t.Fatalf("cancelled sessions = %v, want none after successor turn", cancelled) + } + if got := sessionState(t, repo, "session-candidate"); got != "RUNNING" { + t.Fatalf("session state = %q, want RUNNING", got) + } +} + // sameStringSet reports whether got and want contain the same strings, // ignoring order and duplicates count-for-count — used to compare the set of // cancelled session IDs against an expected set regardless of return order. @@ -2288,6 +2387,103 @@ func TestCreateTurnRoundTripsEveryFieldAndDefaultsTimestamps(t *testing.T) { } } +// TestCreateTurnRefreshesSessionUpdatedAt verifies that turn persistence +// refreshes the parent session row's updated_at: the session reconciliation +// sweep measures event silence from task_sessions.updated_at (and the newest +// message), so a turn dispatched without a user message — auto-start, +// workflow on_enter — must reset the clock even when no message row exists +// yet. Both entry points (CreateTurn and the step-stamped transaction) are +// covered. +func TestCreateTurnRefreshesSessionUpdatedAt(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedSessionForTurns(t, repo, "task-turn-clock", "session-turn-clock") + + oldRowTime := time.Now().UTC().Add(-3 * time.Hour) + if _, err := repo.DB().ExecContext(ctx, `UPDATE task_sessions SET updated_at = ? WHERE id = ?`, oldRowTime, "session-turn-clock"); err != nil { + t.Fatalf("age session row: %v", err) + } + + turn := &models.Turn{ + ID: "turn-clock", TaskSessionID: "session-turn-clock", TaskID: "task-turn-clock", + } + if err := repo.CreateTurn(ctx, turn); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + session, err := repo.GetTaskSession(ctx, "session-turn-clock") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if !session.UpdatedAt.After(oldRowTime.Add(time.Hour)) { + t.Errorf("updated_at = %v after CreateTurn, want refreshed well past the aged %v", session.UpdatedAt, oldRowTime) + } + + // The step-stamped transactional path refreshes it too. + if _, err := repo.DB().ExecContext(ctx, `UPDATE task_sessions SET updated_at = ? WHERE id = ?`, oldRowTime, "session-turn-clock"); err != nil { + t.Fatalf("age session row again: %v", err) + } + stamped := &models.Turn{ + ID: "turn-clock-2", TaskSessionID: "session-turn-clock", TaskID: "task-turn-clock", + } + if _, err := repo.CreateTurnWithStepStamp(ctx, stamped); err != nil { + t.Fatalf("CreateTurnWithStepStamp: %v", err) + } + session, err = repo.GetTaskSession(ctx, "session-turn-clock") + if err != nil { + t.Fatalf("GetTaskSession (second): %v", err) + } + if !session.UpdatedAt.After(oldRowTime.Add(time.Hour)) { + t.Errorf("updated_at = %v after CreateTurnWithStepStamp, want refreshed well past the aged %v", session.UpdatedAt, oldRowTime) + } +} + +func TestCreateTurnRollsBackWhenSessionActivityRefreshFails(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedSessionForTurns(t, repo, "task-turn-atomic", "session-turn-atomic") + oldRowTime := time.Now().UTC().Add(-3 * time.Hour) + if _, err := repo.DB().ExecContext(ctx, + `UPDATE task_sessions SET updated_at = ? WHERE id = ?`, oldRowTime, "session-turn-atomic"); err != nil { + t.Fatalf("age session row: %v", err) + } + if _, err := repo.DB().ExecContext(ctx, ` + CREATE TRIGGER fail_turn_session_clock + BEFORE UPDATE OF updated_at ON task_sessions + WHEN OLD.id = 'session-turn-atomic' + BEGIN + SELECT RAISE(ABORT, 'session activity refresh rejected'); + END; + `); err != nil { + t.Fatalf("create failure trigger: %v", err) + } + + err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-atomic", + TaskSessionID: "session-turn-atomic", + TaskID: "task-turn-atomic", + }) + if err == nil { + t.Fatal("CreateTurn succeeded despite a failed session activity refresh") + } + + var turnCount int + if err := repo.DB().QueryRowContext(ctx, + `SELECT COUNT(*) FROM task_session_turns WHERE id = ?`, "turn-atomic").Scan(&turnCount); err != nil { + t.Fatalf("count inserted turn: %v", err) + } + if turnCount != 0 { + t.Fatalf("turn row count = %d after failed session refresh, want 0", turnCount) + } + var gotSessionTime time.Time + if err := repo.DB().QueryRowContext(ctx, + `SELECT updated_at FROM task_sessions WHERE id = ?`, "session-turn-atomic").Scan(&gotSessionTime); err != nil { + t.Fatalf("read session clock: %v", err) + } + if !gotSessionTime.Equal(oldRowTime) { + t.Fatalf("session updated_at = %v after rollback, want %v", gotSessionTime, oldRowTime) + } +} + func TestGetTurnAndActiveTurnReturnErrNoRowsWhenAbsent(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 21721abe7a..6d621d4d64 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -4240,11 +4240,36 @@ func (r *Repository) ListTasksForAutoArchive(ctx context.Context) ([]*models.Tas return r.scanTasks(rows) } +// ListUnarchivedTasksWithActiveSessions returns the unarchived tasks that +// still have at least one task_sessions row in an active DB state +// (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT). This is the candidate list +// for the session reconciliation sweep's active-task pass (see +// service.StartSessionReconciliationLoop): an active session on an +// unarchived task whose execution is absent from the in-memory execution +// store is left over from a lost actor (e.g. a backend restart) and must be +// detected or healed by the sweep, not by any request path. The sweep +// re-derives this list every pass, so tasks that regain a live execution +// between passes are simply no longer candidates. +func (r *Repository) ListUnarchivedTasksWithActiveSessions(ctx context.Context) ([]*models.Task, error) { + rows, err := r.ro.QueryContext(ctx, ` + SELECT DISTINCT `+taskSelectColumns("t")+` + FROM tasks t + JOIN task_sessions ts ON ts.task_id = t.id + WHERE t.archived_at IS NULL + AND ts.state IN ('CREATED', 'STARTING', 'RUNNING', 'WAITING_FOR_INPUT') + `) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + return r.scanTasks(rows) +} + // ListArchivedTasksWithActiveSessions returns the IDs of archived tasks that // still have at least one task_sessions row in an active DB state // (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT). This is the candidate list // for the periodic reconciliation sweep (see -// service.StartArchivedSessionReconciliationLoop): finalizeCancelledSessions +// service.StartSessionReconciliationLoop): finalizeCancelledSessions // bounds its session-cancellation retry to a handful of attempts, so // sustained SQLite writer contention can exhaust it and leave an archived // task's sessions stuck active with no session.state_changed event ever diff --git a/apps/backend/internal/task/repository/sqlite/task_active_sessions_postgres_test.go b/apps/backend/internal/task/repository/sqlite/task_active_sessions_postgres_test.go new file mode 100644 index 0000000000..fc1b155179 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/task_active_sessions_postgres_test.go @@ -0,0 +1,71 @@ +package sqlite + +// Postgres parity coverage for the session reconciliation sweep's reads. +// GetLastMessageTimeBySessionIDs aggregates MAX(updated_at) over +// task_session_messages and scans the result per dialect, and +// ListUnarchivedTasksWithActiveSessions projects full task rows through a +// DISTINCT join. The SQLite-only run never exercises PostgreSQL's own +// timestamp and DISTINCT-interaction behavior on these paths. +// Skips unless KANDEV_TEST_POSTGRES_DSN is set; CI runs these in postgres-boot. + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/testutil" +) + +func TestPostgresSessionSweepReads(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + seedPostgresTask(t, repo, "task-sweep-pg") + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-sweep-pg", TaskID: "task-sweep-pg", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + oldest := time.Now().Add(-3 * time.Hour).UTC().Truncate(time.Microsecond) + newest := time.Now().Add(-1 * time.Hour).UTC().Truncate(time.Microsecond) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-sweep-pg", TaskSessionID: "session-sweep-pg", TaskID: "task-sweep-pg", StartedAt: oldest, + }); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + for _, msg := range []*models.Message{ + {ID: "msg-sweep-pg-old", TaskSessionID: "session-sweep-pg", TaskID: "task-sweep-pg", TurnID: "turn-sweep-pg", Content: "old", UpdatedAt: oldest}, + {ID: "msg-sweep-pg-new", TaskSessionID: "session-sweep-pg", TaskID: "task-sweep-pg", TurnID: "turn-sweep-pg", Content: "new", UpdatedAt: newest}, + } { + if err := repo.CreateMessage(ctx, msg); err != nil { + t.Fatalf("CreateMessage(%s): %v", msg.ID, err) + } + } + + tasks, err := repo.ListUnarchivedTasksWithActiveSessions(ctx) + if err != nil { + t.Fatalf("ListUnarchivedTasksWithActiveSessions: %v", err) + } + found := false + for _, task := range tasks { + if task.ID == "task-sweep-pg" { + found = true + } + } + if !found { + t.Fatalf("task-sweep-pg not returned among %d candidates", len(tasks)) + } + + lastMessage, err := repo.GetLastMessageTimeBySessionIDs(ctx, []string{"session-sweep-pg"}) + if err != nil { + t.Fatalf("GetLastMessageTimeBySessionIDs: %v", err) + } + if got, ok := lastMessage["session-sweep-pg"]; !ok || !got.Equal(newest) { + t.Fatalf("last message time = %v (present=%v), want the newest message time %v", got, ok, newest) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/task_active_sessions_test.go b/apps/backend/internal/task/repository/sqlite/task_active_sessions_test.go new file mode 100644 index 0000000000..b37dd1c1e5 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/task_active_sessions_test.go @@ -0,0 +1,106 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +func TestListUnarchivedTasksWithActiveSessionsReturnsOnlyHolders(t *testing.T) { + repo := newRepoForArchiveTests(t, "task-active-waiting", "task-active-done", "task-arch-running") + ctx := context.Background() + + sessions := []struct { + id string + taskID string + state models.TaskSessionState + }{ + {"session-active-waiting", "task-active-waiting", models.TaskSessionStateWaitingForInput}, + {"session-active-done", "task-active-done", models.TaskSessionStateCompleted}, + {"session-arch-running", "task-arch-running", models.TaskSessionStateRunning}, + } + for _, session := range sessions { + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: session.id, TaskID: session.taskID, State: session.state, + }); err != nil { + t.Fatalf("CreateTaskSession(%s): %v", session.id, err) + } + } + if err := repo.ArchiveTask(ctx, "task-arch-running"); err != nil { + t.Fatalf("ArchiveTask(task-arch-running): %v", err) + } + + tasks, err := repo.ListUnarchivedTasksWithActiveSessions(ctx) + if err != nil { + t.Fatalf("ListUnarchivedTasksWithActiveSessions: %v", err) + } + if len(tasks) != 1 { + t.Fatalf("candidate tasks = %d (%v), want exactly task-active-waiting", len(tasks), taskIDsOf(tasks)) + } + if tasks[0].ID != "task-active-waiting" { + t.Fatalf("candidate task ID = %q, want task-active-waiting", tasks[0].ID) + } + if tasks[0].WorkspaceID != archiveWorkspaceID { + t.Fatalf("candidate workspace ID = %q, want %q", tasks[0].WorkspaceID, archiveWorkspaceID) + } +} + +func taskIDsOf(tasks []*models.Task) []string { + ids := make([]string, 0, len(tasks)) + for _, task := range tasks { + ids = append(ids, task.ID) + } + return ids +} + +func TestGetLastMessageTimeBySessionIDsReturnsNewestPerSession(t *testing.T) { + repo := newRepoForArchiveTests(t, "task-msg") + ctx := context.Background() + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-msg", TaskID: "task-msg", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-silent", TaskID: "task-msg", State: models.TaskSessionStateRunning, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + oldest := time.Now().Add(-3 * time.Hour).UTC().Truncate(time.Second) + newest := time.Now().Add(-1 * time.Hour).UTC().Truncate(time.Second) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-msg", TaskSessionID: "session-msg", TaskID: "task-msg", StartedAt: oldest, + }); err != nil { + t.Fatalf("CreateTurn: %v", err) + } + for _, msg := range []*models.Message{ + {ID: "msg-old", TaskSessionID: "session-msg", TaskID: "task-msg", TurnID: "turn-msg", Content: "old", UpdatedAt: oldest}, + {ID: "msg-new", TaskSessionID: "session-msg", TaskID: "task-msg", TurnID: "turn-msg", Content: "new", UpdatedAt: newest}, + } { + if err := repo.CreateMessage(ctx, msg); err != nil { + t.Fatalf("CreateMessage(%s): %v", msg.ID, err) + } + } + + got, err := repo.GetLastMessageTimeBySessionIDs(ctx, []string{"session-msg", "session-silent"}) + if err != nil { + t.Fatalf("GetLastMessageTimeBySessionIDs: %v", err) + } + if len(got) != 1 { + t.Fatalf("result sessions = %v, want only session-msg", got) + } + if !got["session-msg"].Equal(newest) { + t.Fatalf("last message time = %v, want the newest message time %v", got["session-msg"], newest) + } + + empty, err := repo.GetLastMessageTimeBySessionIDs(ctx, nil) + if err != nil { + t.Fatalf("GetLastMessageTimeBySessionIDs(nil): %v", err) + } + if len(empty) != 0 { + t.Fatalf("result for empty input = %v, want empty", empty) + } +} diff --git a/apps/backend/internal/task/service/active_session_stall.go b/apps/backend/internal/task/service/active_session_stall.go new file mode 100644 index 0000000000..425cb049c7 --- /dev/null +++ b/apps/backend/internal/task/service/active_session_stall.go @@ -0,0 +1,432 @@ +package service + +import ( + "context" + "sort" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/archivecascade" + "github.com/kandev/kandev/internal/task/models" + "go.uber.org/zap" +) + +const ( + // defaultStallDetectionThreshold is the event-silence window after which + // an execution-less active session is classified as stalled + // (tasks.stallDetectionThreshold defaults to 2h). + defaultStallDetectionThreshold = 2 * time.Hour + // stallHealGraceMultiplier derives the orphaned-session healing grace + // window from the stall threshold. Healing waits twice as long as + // detection so an operator has a full threshold window to act on the + // task.stalled event before the sweep cancels the orphaned session, and + // so a live in-flight launch that has not yet registered its execution + // never races the grace window. + stallHealGraceMultiplier = 2 +) + +// stallThreshold returns the configured stall-detection threshold, falling +// back to the default when the setter was never called (zero value). +func (s *Service) stallThreshold() time.Duration { + if s.stallDetectionThreshold > 0 { + return s.stallDetectionThreshold + } + return defaultStallDetectionThreshold +} + +// runActiveSessionSweep is the session reconciliation sweep's active-task +// pass: for every unarchived task that still holds an active session +// (CREATED/STARTING/RUNNING/WAITING_FOR_INPUT), it classifies each session +// against the in-memory execution store. A session with a registered live +// execution is healthy regardless of silence. A session with no live +// execution has lost its actor (backend restart, lost executor, or a launch +// that never registered) and nothing in any request path will ever advance +// it again, so the pass: +// +// - emits one task.stalled event per stall episode once the session has +// been event-silent beyond the stall threshold (detection only), and +// - cancels the session via the same finalizeCancelledSessions transition +// the archived pass uses once the silence exceeds the grace window +// (twice the threshold), making the DB state truthful, delivering the +// session.state_changed event clients key off, and unblocking the task's +// step lifecycle. +// +// The execution check is fail-closed: without the registry the pass cannot +// prove "no live execution", so it skips rather than healing or alerting on +// a guess. The event-silence clock is the newest of the session row's +// updated_at and its newest task_session_messages row, so any state +// transition or message resets it. +func (s *Service) runActiveSessionSweep(ctx context.Context, now time.Time) { + if s.sessionExecutionRegistry == nil { + return + } + tasks, err := s.tasks.ListUnarchivedTasksWithActiveSessions(ctx) + if err != nil { + s.logger.Error("active-session sweep: failed to list candidates", zap.Error(err)) + return + } + if len(tasks) == 0 { + s.clearAllStallNotifications() + return + } + s.pruneStallNotificationsToCandidates(tasks) + + threshold := s.stallThreshold() + grace := threshold * stallHealGraceMultiplier + for _, task := range tasks { + if task == nil || task.ID == "" { + continue + } + s.sweepTaskSessions(ctx, task, now, threshold, grace) + } +} + +// sweepTaskSessions runs the sweep classification for one candidate task. +func (s *Service) sweepTaskSessions( + ctx context.Context, + task *models.Task, + now time.Time, + threshold, grace time.Duration, +) { + activeSessions, err := s.sessions.ListActiveTaskSessionsByTaskID(ctx, task.ID) + if err != nil { + s.logger.Warn("active-session sweep: failed to list active sessions", + zap.String("task_id", task.ID), + zap.Error(err)) + return + } + if len(activeSessions) == 0 { + // Reconciled by a concurrent pass between the candidate list query + // and this read. + return + } + + liveSessions := s.liveSessionSet(task.ID) + orphaned := make([]*models.TaskSession, 0, len(activeSessions)) + for _, session := range activeSessions { + if session == nil { + continue + } + if _, live := liveSessions[session.ID]; !live { + orphaned = append(orphaned, session) + } + } + if len(orphaned) == 0 { + s.clearStallNotifications(task.ID) + return + } + + stalled, healable, lastEventBySession, classErr := s.classifyOrphanedSessions( + ctx, orphaned, now, threshold, grace, + ) + if classErr != nil { + // The per-session activity clock could not be read. updated_at alone + // understates activity (message writes never touch it), so classifying + // from the incomplete clock could stall-report or heal recently + // active work. Fail closed: skip this task this tick and retry next + // sweep. + return + } + s.notifyStalledSessions(ctx, task, stalled, lastEventBySession, now, threshold) + s.healOrphanedSessions(ctx, task, activeSessions, orphaned, healable, lastEventBySession) +} + +// classifyOrphanedSessions splits execution-less sessions into stalled +// (event-silent beyond threshold) and healable (silent beyond the grace +// window), and reports each stalled session's resolved event time for the +// task.stalled payload. It returns an error when a session's activity clock +// cannot be read: silence classification is only sound on a complete clock, +// so callers must skip the task rather than fall back to a stale timestamp. +func (s *Service) classifyOrphanedSessions( + ctx context.Context, + orphaned []*models.TaskSession, + now time.Time, + threshold, grace time.Duration, +) (stalled, healable []*models.TaskSession, lastEventBySession map[string]time.Time, err error) { + times, readErr := s.messages.GetLastMessageTimeBySessionIDs(ctx, sessionIDs(orphaned)) + if readErr != nil { + s.logger.Warn("active-session sweep: failed to load last message times; skipping task this tick", + zap.Error(readErr)) + return nil, nil, nil, readErr + } + lastEventBySession = make(map[string]time.Time, len(orphaned)) + for _, session := range orphaned { + lastEvent := lastSessionEventAt(session, times) + lastEventBySession[session.ID] = lastEvent + silence := now.Sub(lastEvent) + if silence < threshold { + continue + } + stalled = append(stalled, session) + if silence >= grace { + healable = append(healable, session) + } + } + return stalled, healable, lastEventBySession, nil +} + +// sessionIDs extracts the non-empty session IDs from sessions. +func sessionIDs(sessions []*models.TaskSession) []string { + ids := make([]string, 0, len(sessions)) + for _, session := range sessions { + ids = append(ids, session.ID) + } + return ids +} + +// lastSessionEventAt resolves one session's event clock: the newer of its +// newest task_session_messages row (when one exists) and the session row's +// own updated_at, so any persisted state transition resets the silence. +func lastSessionEventAt(session *models.TaskSession, lastEventBySession map[string]time.Time) time.Time { + lastEvent := session.UpdatedAt + if lastMessage, ok := lastEventBySession[session.ID]; ok && lastMessage.After(lastEvent) { + lastEvent = lastMessage + } + return lastEvent +} + +// liveSessionSet reads the registry's live-execution snapshot for one task. +func (s *Service) liveSessionSet(taskID string) map[string]struct{} { + ids := s.sessionExecutionRegistry.LiveSessionIDsForTask(taskID) + live := make(map[string]struct{}, len(ids)) + for _, id := range ids { + if id != "" { + live[id] = struct{}{} + } + } + return live +} + +// stallPayload field keys (task_id / workspace_id exist as constants for +// other features; these name this payload's own keys). +const ( + stallPayloadTaskIDKey = "task_id" + stallPayloadWorkspaceIDKey = "workspace_id" +) + +// notifyStalledSessions emits the task.stalled warning and event for +// sessions not yet reported in this stall episode. See stallNotifiedSessions +// for the episode semantics. The payload's timing fields describe exactly +// the sessions it names: a session already reported earlier in the episode +// must not skew stalled_for/last_event_at for a newly reported sibling. +// +// Delivery is recorded only after the event publish succeeds, so a transient +// publish failure retries on the next sweep tick. +func (s *Service) notifyStalledSessions( + ctx context.Context, + task *models.Task, + stalled []*models.TaskSession, + lastEventBySession map[string]time.Time, + now time.Time, + threshold time.Duration, +) { + if s.stallNotifiedSessions == nil { + s.stallNotifiedSessions = make(map[string]map[string]struct{}) + } + notified := s.stallNotifiedSessions[task.ID] + if len(stalled) == 0 { + if notified != nil { + delete(s.stallNotifiedSessions, task.ID) + } + return + } + notified = s.pruneStallEpisode(task.ID, stalled) + newlyStalled, oldestNewEvent := selectNewlyStalled(stalled, notified, lastEventBySession) + if len(newlyStalled) == 0 { + return + } + + sessionIDs := make([]string, 0, len(newlyStalled)) + for _, session := range newlyStalled { + sessionIDs = append(sessionIDs, session.ID) + } + sort.Strings(sessionIDs) + silence := now.Sub(oldestNewEvent) + s.logger.Warn("stalled_task detected: active session with no live execution and no recent events", + zap.String("task_id", task.ID), + zap.Strings("session_ids", sessionIDs), + zap.Duration("silent_for", silence), + zap.Duration("threshold", threshold)) + if s.eventBus == nil { + return + } + payload := map[string]interface{}{ + stallPayloadTaskIDKey: task.ID, + stallPayloadWorkspaceIDKey: task.WorkspaceID, + "session_ids": sessionIDs, + "stalled_for": silence.String(), + "last_event_at": oldestNewEvent.UTC().Format(time.RFC3339Nano), + "detection_only": true, + } + event := bus.NewEvent(events.TaskStalled, "task-reconciliation", payload) + if err := s.eventBus.Publish(ctx, events.TaskStalled, event); err != nil { + s.logger.Error("failed to publish stalled task event", + zap.String("task_id", task.ID), + zap.Error(err)) + return + } + for _, sessionID := range sessionIDs { + notified[sessionID] = struct{}{} + } +} + +// healOrphanedSessions cancels execution-less sessions that have been silent +// beyond the grace window, reusing the archived pass's +// finalizeCancelledSessions transition (cancellation with the orphaned +// reason, clarification expiry, parked-projection cleanup, ceiling release, +// and the session.state_changed event) so clients keying off that event see +// the session stop. Repeat passes are no-ops: the underlying UPDATE only +// matches rows still in an active state. +// +// Healing is task-scoped at the eligibility boundary, so it only runs when +// every active session of the task is execution-less and past the grace +// window. The write itself is candidate-scoped, so a same-session resume or a +// newer turn cannot be cancelled by a stale sweep snapshot. +// +// The guard is re-evaluated at the cancellation boundary itself: the +// liveness snapshot was taken before the silence reads, and an execution +// that registered (or a session that started) in between must abort the +// heal. Re-checking LiveSessionIDsForTask immediately before the write is +// not a lock — a registration can still land after it — but it closes the +// window from "seconds of reads ago" to "instantaneous", and any residual +// race must also co-occur with every session of the task having been +// silent past twice the stall threshold. The next sweep tick heals once +// the condition genuinely holds. +func (s *Service) healOrphanedSessions( + ctx context.Context, + task *models.Task, + activeSessions, orphaned, healable []*models.TaskSession, + lastEventBySession map[string]time.Time, +) { + if len(healable) == 0 || len(orphaned) != len(activeSessions) || len(healable) != len(orphaned) { + return + } + if live := s.liveSessionSet(task.ID); len(live) > 0 { + s.logger.Info("active-session sweep: heal aborted, a live execution appeared for the task", + zap.String("task_id", task.ID), + zap.Int("live_sessions", len(live))) + return + } + candidates := make([]models.ActiveSessionCancellationCandidate, 0, len(healable)) + for _, session := range healable { + turn, err := s.GetActiveTurn(ctx, session.ID) + if err != nil { + s.logger.Warn("active-session sweep: failed to inspect current turn; skipping heal", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.Error(err)) + return + } + lastEvent := lastEventBySession[session.ID] + if lastEvent.IsZero() { + lastEvent = session.UpdatedAt + } + candidate := models.ActiveSessionCancellationCandidate{ + SessionID: session.ID, + ExpectedUpdatedAt: session.UpdatedAt, + ExpectedLastEventAt: lastEvent, + } + if turn != nil { + candidate.ExpectedTurnID = turn.ID + } + candidates = append(candidates, candidate) + } + s.logger.Info("active-session sweep: healing orphaned sessions", + zap.String("task_id", task.ID), + zap.Strings("session_ids", cancellationCandidateIDs(candidates))) + deadline := archivecascade.ArchiveDeadline(ctx) + healCtx, cancel := archivecascade.ContinuationContextUntil(ctx, deadline) + defer cancel() + // Candidates scope the cancellation to exactly the sessions this pass + // classified as orphaned and compare their activity/turn identity in the + // UPDATE. A same-session resume or successor turn therefore survives the + // write and is retried by the next sweep. + s.finalizeCancelledSessionCandidates( + healCtx, task.ID, activeSessions, candidates, deadline, models.SessionOrphanedCancelReason, + ) +} + +func cancellationCandidateIDs(candidates []models.ActiveSessionCancellationCandidate) []string { + ids := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + ids = append(ids, candidate.SessionID) + } + return ids +} + +// pruneStallNotificationsToCandidates removes stale task entries while the +// candidate query is healthy. A task that disappears from the query can no +// longer have an active stall episode, so retaining its entry leaks memory and +// suppresses a future episode if the task later becomes active again. Query +// failures return before this helper and therefore preserve the prior map. +func (s *Service) pruneStallNotificationsToCandidates(tasks []*models.Task) { + candidateIDs := make(map[string]struct{}, len(tasks)) + for _, task := range tasks { + if task != nil && task.ID != "" { + candidateIDs[task.ID] = struct{}{} + } + } + for taskID := range s.stallNotifiedSessions { + if _, present := candidateIDs[taskID]; !present { + delete(s.stallNotifiedSessions, taskID) + } + } +} + +// pruneStallEpisode maintains the per-session episode set for one task: +// entries for sessions no longer in the current stalled set are removed (so +// a later stall on that session reports a fresh episode even while a +// sibling of the same task remains stalled), and the task's entry is +// created when it does not exist yet. It returns the live set to consult. +func (s *Service) pruneStallEpisode(taskID string, stalled []*models.TaskSession) map[string]struct{} { + notified := s.stallNotifiedSessions[taskID] + if notified == nil { + notified = make(map[string]struct{}, len(stalled)) + s.stallNotifiedSessions[taskID] = notified + } + stalledIDs := make(map[string]struct{}, len(stalled)) + for _, session := range stalled { + stalledIDs[session.ID] = struct{}{} + } + for sessionID := range notified { + if _, still := stalledIDs[sessionID]; !still { + delete(notified, sessionID) + } + } + return notified +} + +// selectNewlyStalled picks the stalled sessions not yet reported in the +// open episode and reports the oldest event time across that selection, so +// the payload's timing fields describe exactly the sessions it names. +func selectNewlyStalled( + stalled []*models.TaskSession, + notified map[string]struct{}, + lastEventBySession map[string]time.Time, +) (newlyStalled []*models.TaskSession, oldestNewEvent time.Time) { + newlyStalled = make([]*models.TaskSession, 0, len(stalled)) + for _, session := range stalled { + if _, seen := notified[session.ID]; seen { + continue + } + newlyStalled = append(newlyStalled, session) + lastEvent := lastEventBySession[session.ID] + if oldestNewEvent.IsZero() || lastEvent.Before(oldestNewEvent) { + oldestNewEvent = lastEvent + } + } + return newlyStalled, oldestNewEvent +} + +// clearStallNotifications ends a task's stall episode once it no longer has +// any stalled session, so a later stall on the same task reports again. +func (s *Service) clearStallNotifications(taskID string) { + delete(s.stallNotifiedSessions, taskID) +} + +// clearAllStallNotifications resets episode tracking when no candidates +// remain at all. +func (s *Service) clearAllStallNotifications() { + s.stallNotifiedSessions = make(map[string]map[string]struct{}) +} diff --git a/apps/backend/internal/task/service/active_session_stall_test.go b/apps/backend/internal/task/service/active_session_stall_test.go new file mode 100644 index 0000000000..b0b71c8e18 --- /dev/null +++ b/apps/backend/internal/task/service/active_session_stall_test.go @@ -0,0 +1,534 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// stubExecutionRegistry is the test double for SessionExecutionRegistry: it +// reports a fixed set of session IDs as having a live in-memory execution. +// With switchToLiveAfterFirstCall set, the first LiveSessionIDsForTask call +// reports nothing (classification sees no live execution) and every later +// call reports liveSessions — reproducing a mid-sweep registration race. +type stubExecutionRegistry struct { + liveSessions []string + switchToLiveAfterFirstCall bool + calls int +} + +func (r *stubExecutionRegistry) LiveSessionIDsForTask(taskID string) []string { + r.calls++ + if r.switchToLiveAfterFirstCall && r.calls > 1 { + return r.liveSessions + } + if r.switchToLiveAfterFirstCall { + return nil + } + return r.liveSessions +} + +// sweepFixture seeds one workspace, workflow, task, and session, wires the +// stub registry, and returns the fixed sweep clock so tests can place the +// session's UpdatedAt relative to it deterministically. +type sweepFixture struct { + svc *Service + eventBus *MockEventBus + repo interface { + GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) + CreateTaskSession(ctx context.Context, session *models.TaskSession) error + } + registry *stubExecutionRegistry + now time.Time + rawRepo *sqliterepo.Repository +} + +// newSweepFixture creates a task holding one RUNNING session whose row went +// quiet silence before the sweep clock. +func newSweepFixture(t *testing.T, silence time.Duration) *sweepFixture { + t.Helper() + registry := &stubExecutionRegistry{} + svc, eventBus, repo := createTestService(t) + svc.SetSessionExecutionRegistry(registry) + svc.SetStallDetectionThreshold(2 * time.Hour) + + now := time.Now() + ctx := context.Background() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Workflow"}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: "task-1", WorkspaceID: "ws-1", WorkflowID: "wf-1", WorkflowStepID: "step-1", + Title: "Sweep test", Priority: "medium", + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateRunning, + AgentProfileID: "agent-1", IsPrimary: true, UpdatedAt: now.Add(-silence), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + return &sweepFixture{ + svc: svc, + eventBus: eventBus, + repo: repo, + registry: registry, + now: now, + rawRepo: repo, + } +} + +func findTaskStalledEvents(eventBus *MockEventBus) []*bus.Event { + var found []*bus.Event + for _, evt := range eventBus.GetPublishedEvents() { + if evt.Type == events.TaskStalled { + found = append(found, evt) + } + } + return found +} + +// TestService_ActiveSessionSweepEmitsStalledTaskEvent is the #3712 +// regression: an unarchived task holding an active session with no live +// execution and no events for longer than the stall threshold must surface a +// task.stalled event instead of stalling silently. +func TestService_ActiveSessionSweepEmitsStalledTaskEvent(t *testing.T) { + // 3h of silence: past the 2h threshold, still inside the 4h healing + // grace window. + fixture := newSweepFixture(t, 3*time.Hour) + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + stalled := findTaskStalledEvents(fixture.eventBus) + if len(stalled) != 1 { + t.Fatalf("task.stalled events = %d, want 1", len(stalled)) + } + data, ok := stalled[0].Data.(map[string]interface{}) + if !ok { + t.Fatalf("task.stalled payload type = %T, want map", stalled[0].Data) + } + if got := data["task_id"]; got != "task-1" { + t.Errorf("task_id = %v, want task-1", got) + } + if got := data["workspace_id"]; got != "ws-1" { + t.Errorf("workspace_id = %v, want ws-1", got) + } + ids, _ := data["session_ids"].([]string) + if len(ids) != 1 || ids[0] != "session-1" { + t.Errorf("session_ids = %v, want [session-1]", data["session_ids"]) + } + + // Detection only: the session must still be active in the DB. + session, err := fixture.repo.GetTaskSession(context.Background(), "session-1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session state after detection = %q, want RUNNING (detection must not heal)", session.State) + } +} + +// TestService_ActiveSessionSweepSkipsLiveExecution proves the ExecutionStore +// guard: an event-silent session that still has a registered live execution +// is healthy and must not fire a stall event. +func TestService_ActiveSessionSweepSkipsLiveExecution(t *testing.T) { + fixture := newSweepFixture(t, 3*time.Hour) + fixture.registry.liveSessions = []string{"session-1"} + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 0 { + t.Fatalf("task.stalled events = %d, want 0 for a session with a live execution", len(stalled)) + } +} + +// TestService_ActiveSessionSweepSkipsRecentSessions proves the silence +// window: an execution-less session that went quiet inside the threshold is +// not yet stalled (e.g. a launch that has not registered its execution). +func TestService_ActiveSessionSweepSkipsRecentSessions(t *testing.T) { + // 30 minutes of silence: below the 2h threshold. + fixture := newSweepFixture(t, 30*time.Minute) + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 0 { + t.Fatalf("task.stalled events = %d, want 0 for a session inside the threshold", len(stalled)) + } +} + +// TestService_ActiveSessionSweepStallEventEmittedOncePerEpisode proves the +// episode dedupe: a sweep ticking every minute must not re-emit task.stalled +// for the same stalled session, but must report it again after the episode +// ends (a live execution reappears) and a new stall begins. +func TestService_ActiveSessionSweepStallEventEmittedOncePerEpisode(t *testing.T) { + fixture := newSweepFixture(t, 3*time.Hour) + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now.Add(time.Minute)) + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 1 { + t.Fatalf("task.stalled events after two passes = %d, want 1 (once per episode)", len(stalled)) + } + + // Episode ends: the session gains a live execution. + fixture.registry.liveSessions = []string{"session-1"} + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now.Add(2*time.Minute)) + fixture.registry.liveSessions = nil + + // A later pass in a new episode reports the stall again. + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now.Add(3*time.Minute)) + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 2 { + t.Fatalf("task.stalled events across two episodes = %d, want 2", len(stalled)) + } +} + +// TestService_ActiveSessionSweepHealsOrphanedSessions is the #3711 half of +// the sweep: once an execution-less session has been silent beyond the grace +// window (twice the threshold), the sweep cancels it through the same +// finalizeCancelledSessions transition the archived pass uses, so the DB +// state becomes truthful and the session.state_changed event fires. +func TestService_ActiveSessionSweepHealsOrphanedSessions(t *testing.T) { + // 5h of silence: past the 4h grace window (2 x 2h threshold). + fixture := newSweepFixture(t, 5*time.Hour) + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + session, err := fixture.repo.GetTaskSession(context.Background(), "session-1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.State != models.TaskSessionStateCancelled { + t.Fatalf("session state after healing = %q, want CANCELLED", session.State) + } + if session.ErrorMessage != models.SessionOrphanedCancelReason { + t.Fatalf("session error_message = %q, want %q", session.ErrorMessage, models.SessionOrphanedCancelReason) + } + + var stateChanged *bus.Event + for _, evt := range fixture.eventBus.GetPublishedEvents() { + if evt.Type != events.TaskSessionStateChanged { + continue + } + if data, ok := evt.Data.(map[string]interface{}); ok && data["session_id"] == "session-1" { + stateChanged = evt + } + } + if stateChanged == nil { + t.Fatal("expected a session.state_changed event for session-1 from the healing pass, got none") + } + data := stateChanged.Data.(map[string]interface{}) + if got := data["new_state"]; got != string(models.TaskSessionStateCancelled) { + t.Errorf("new_state = %v, want CANCELLED", got) + } +} + +// TestService_ActiveSessionSweepDoesNotHealAroundLiveSibling proves the +// task-scoped cancel guard: while any active session of the task still has a +// live execution, the sweep must not cancel the task's other (orphaned) +// sessions, because CancelActiveTaskSessionsByTaskID is task-scoped and would +// kill the live work too. +func TestService_ActiveSessionSweepDoesNotHealAroundLiveSibling(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + ctx := context.Background() + if err := fixture.repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-live", TaskID: "task-1", State: models.TaskSessionStateRunning, + AgentProfileID: "agent-1", UpdatedAt: fixture.now.Add(-5 * time.Hour), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + fixture.registry.liveSessions = []string{"session-live"} + + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + + for _, sessionID := range []string{"session-1", "session-live"} { + session, err := fixture.repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession(%s): %v", sessionID, err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session %s state = %q, want RUNNING (healing must not run while a sibling is live)", sessionID, session.State) + } + } + // The orphaned sibling is still detected. + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 1 { + t.Fatalf("task.stalled events = %d, want 1 for the orphaned sibling", len(stalled)) + } +} + +// TestService_ActiveSessionSweepSkipsArchivedTasks proves the pass boundary: +// archived tasks with active sessions belong to the archived pass, not the +// active-task pass, and must not produce task.stalled events. +func TestService_ActiveSessionSweepSkipsArchivedTasks(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + + if err := fixture.svc.ArchiveTask(context.Background(), "task-1"); err != nil { + t.Fatalf("ArchiveTask: %v", err) + } + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 0 { + t.Fatalf("task.stalled events = %d, want 0 for an archived task (owned by the archived pass)", len(stalled)) + } +} + +// TestService_ActiveSessionSweepWithoutRegistryIsSkipped proves the +// fail-closed guard: without the live-execution registry the pass cannot +// prove "no live execution" and must do nothing rather than guess. +func TestService_ActiveSessionSweepWithoutRegistryIsSkipped(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + fixture.svc.sessionExecutionRegistry = nil + + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 0 { + t.Fatalf("task.stalled events = %d, want 0 without the registry", len(stalled)) + } +} + +// TestService_ActiveSessionSweepHealAbortedWhenExecutionRegistersMidSweep is +// the stale-snapshot regression: an execution that registers between the +// sweep's classification reads and the cancellation boundary must abort the +// heal. The cancellation is scoped to the classified session IDs, so the +// late-registered session survives even when the task qualifies for healing. +func TestService_ActiveSessionSweepHealAbortedWhenExecutionRegistersMidSweep(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + ctx := context.Background() + // A second active session on the same task, equally silent. + if err := fixture.repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-late", TaskID: "task-1", State: models.TaskSessionStateRunning, + AgentProfileID: "agent-1", UpdatedAt: fixture.now.Add(-5 * time.Hour), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + // The registry reports no live executions during classification (the + // fixture's default) but a live execution for session-late by the time + // the heal boundary re-checks — the exact mid-sweep registration race. + fixture.registry.liveSessions = []string{"session-late"} + fixture.registry.switchToLiveAfterFirstCall = true + + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + + for _, sessionID := range []string{"session-1", "session-late"} { + session, err := fixture.repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession(%s): %v", sessionID, err) + } + if session.State != models.TaskSessionStateRunning { + t.Fatalf("session %s state = %q, want RUNNING (heal must abort when a live execution appears)", sessionID, session.State) + } + } +} + +// TestService_ActiveSessionSweepHealCancelsOnlyClassifiedSessions proves the +// ID-scoped cancellation: when the task genuinely qualifies for healing, the +// cancellation only touches the sessions the sweep classified. A session +// that became active after classification — outside the classified set — +// keeps running. +func TestService_ActiveSessionSweepHealCancelsOnlyClassifiedSessions(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + ctx := context.Background() + + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + + // Both active sessions of the task were classified as healable orphans, + // so the ID scope covers everything and the heal still completes. + session, err := fixture.repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.State != models.TaskSessionStateCancelled { + t.Fatalf("session-1 state = %q, want CANCELLED", session.State) + } +} + +// TestService_ActiveSessionSweepPrunesRecoveredSessionsFromDedupe proves the +// per-session episode semantics: when one of two stalled sessions recovers +// (gains a live execution) while its sibling stays stalled, the recovered +// session's dedupe entry is dropped. If it stalls again, its new episode is +// reported instead of being silently suppressed. +func TestService_ActiveSessionSweepPrunesRecoveredSessionsFromDedupe(t *testing.T) { + fixture := newSweepFixture(t, 3*time.Hour) + ctx := context.Background() + if err := fixture.repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-2", TaskID: "task-1", State: models.TaskSessionStateRunning, + AgentProfileID: "agent-1", UpdatedAt: fixture.now.Add(-3 * time.Hour), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + // First sweep: both sessions stall; both are reported. + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + first := findTaskStalledEvents(fixture.eventBus) + if len(first) != 1 { + t.Fatalf("task.stalled events after first sweep = %d, want 1", len(first)) + } + ids, _ := first[0].Data.(map[string]interface{})["session_ids"].([]string) + if len(ids) != 2 { + t.Fatalf("first report session_ids = %v, want both sessions", ids) + } + + // session-2 recovers; session-1 stays stalled. The next sweep reports + // nothing new (session-1 was already reported in this episode). + fixture.registry.liveSessions = []string{"session-2"} + fixture.svc.runActiveSessionSweep(ctx, fixture.now.Add(time.Minute)) + if stalled := findTaskStalledEvents(fixture.eventBus); len(stalled) != 1 { + t.Fatalf("task.stalled events after recovery = %d, want 1 (no new report)", len(stalled)) + } + + // session-2 stalls again: its dedupe entry was pruned when it left the + // stalled set, so the new episode reports it again — alongside nothing + // else, because session-1's episode is still open. + fixture.registry.liveSessions = nil + fixture.svc.runActiveSessionSweep(ctx, fixture.now.Add(2*time.Minute)) + all := findTaskStalledEvents(fixture.eventBus) + if len(all) != 2 { + t.Fatalf("task.stalled events across both sweeps = %d, want 2 (recovered session reports again)", len(all)) + } + ids, _ = all[1].Data.(map[string]interface{})["session_ids"].([]string) + if len(ids) != 1 || ids[0] != "session-2" { + t.Fatalf("second report session_ids = %v, want [session-2] only", ids) + } +} + +// TestService_ActiveSessionSweepRetriesFailedStallPublish proves the +// delivery semantics: a session is recorded as notified only after the +// task.stalled publish succeeds, so a transient publish failure is retried +// on the next sweep tick instead of being deduplicated away. +func TestService_ActiveSessionSweepRetriesFailedStallPublish(t *testing.T) { + fixture := newSweepFixture(t, 3*time.Hour) + fixture.eventBus.publishErrors = map[string]error{ + events.TaskStalled: errors.New("transient publish failure"), + } + + // The publish fails; the session must not be recorded as notified. + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now) + + // The publish succeeds on the retry; the same episode must now be + // reported exactly once. + fixture.eventBus.publishErrors = nil + fixture.svc.runActiveSessionSweep(context.Background(), fixture.now.Add(time.Minute)) + + stalled := findTaskStalledEvents(fixture.eventBus) + if len(stalled) != 1 { + t.Fatalf("task.stalled events after publish recovery = %d, want 1 (failed publish retried)", len(stalled)) + } +} + +// TestService_ActiveSessionSweepStallPayloadTimesNewlyReportedOnly proves +// the payload alignment: when a previously reported session stays stalled +// and a newer, quieter silence qualifies a different session mid-episode, +// the report's timing fields describe the newly reported session, not the +// whole stalled set. +func TestService_ActiveSessionSweepStallPayloadTimesNewlyReportedOnly(t *testing.T) { + fixture := newSweepFixture(t, 3*time.Hour) + ctx := context.Background() + + // First sweep reports session-1 (3h of silence). + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + + // session-2 has been silent much longer (6h) but only exists now. + if err := fixture.repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-old", TaskID: "task-1", State: models.TaskSessionStateRunning, + AgentProfileID: "agent-1", UpdatedAt: fixture.now.Add(-6 * time.Hour), + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + fixture.svc.runActiveSessionSweep(ctx, fixture.now.Add(time.Minute)) + + stalled := findTaskStalledEvents(fixture.eventBus) + if len(stalled) != 2 { + t.Fatalf("task.stalled events = %d, want 2", len(stalled)) + } + data, _ := stalled[1].Data.(map[string]interface{}) + ids, _ := data["session_ids"].([]string) + if len(ids) != 1 || ids[0] != "session-old" { + t.Fatalf("second report session_ids = %v, want [session-old]", ids) + } + // stalled_for must describe session-old's 6h1m silence (its row went + // quiet 6h before the original clock; the report tick is 1m later), not + // session-1's 3h1m. + if got := data["stalled_for"]; got != (6*time.Hour + time.Minute).String() { + t.Errorf("stalled_for = %v, want %s (the newly reported session's silence)", got, (6*time.Hour + time.Minute).String()) + } +} + +func TestService_ActiveSessionSweepPrunesRemovedTaskNotifications(t *testing.T) { + fixture := newSweepFixture(t, 30*time.Minute) + fixture.svc.stallNotifiedSessions = map[string]map[string]struct{}{ + "task-1": {"session-1": {}}, + "task-gone": {"session-gone": {}}, + } + + fixture.svc.pruneStallNotificationsToCandidates([]*models.Task{{ID: "task-1"}}) + + if _, ok := fixture.svc.stallNotifiedSessions["task-gone"]; ok { + t.Fatal("stale notification entry for a removed task was retained") + } + if _, ok := fixture.svc.stallNotifiedSessions["task-1"]; !ok { + t.Fatal("notification entry for a current task was pruned") + } +} + +func TestService_ActiveSessionSweepClosesOrphanTurnWithoutCompletionEvent(t *testing.T) { + fixture := newSweepFixture(t, 5*time.Hour) + ctx := context.Background() + turn, err := fixture.svc.StartTurn(ctx, "session-1") + if err != nil { + t.Fatalf("StartTurn: %v", err) + } + old := fixture.now.Add(-5 * time.Hour) + if err := fixture.svc.messages.CreateMessage(ctx, &models.Message{ + ID: "pending-tool-message", + TaskSessionID: "session-1", + TaskID: "task-1", + TurnID: turn.ID, + AuthorType: models.MessageAuthorAgent, + Type: models.MessageTypeToolExecute, + Content: "tool output", + Metadata: map[string]interface{}{"tool_call_id": "call-1", "status": "pending"}, + CreatedAt: old, + UpdatedAt: old, + }); err != nil { + t.Fatalf("CreateMessage: %v", err) + } + if _, err := fixture.rawRepo.DB().ExecContext(ctx, + `UPDATE task_sessions SET updated_at = ? WHERE id = ?`, old, "session-1"); err != nil { + t.Fatalf("age session row: %v", err) + } + + fixture.svc.runActiveSessionSweep(ctx, fixture.now) + + storedTurn, err := fixture.svc.turns.GetTurn(ctx, turn.ID) + if err != nil { + t.Fatalf("GetTurn: %v", err) + } + if storedTurn.CompletedAt == nil || !storedTurn.CompletedAt.Equal(storedTurn.StartedAt) { + t.Fatalf("orphan turn completion = %v, want zero-duration completion at %v", storedTurn.CompletedAt, storedTurn.StartedAt) + } + message, err := fixture.svc.messages.GetMessage(ctx, "pending-tool-message") + if err != nil { + t.Fatalf("GetMessage: %v", err) + } + if got := message.Metadata["status"]; got != "complete" { + t.Fatalf("pending tool status = %v, want complete", got) + } + for _, event := range fixture.eventBus.GetPublishedEvents() { + if event.Type != events.TurnCompleted { + continue + } + if data, ok := event.Data.(map[string]interface{}); ok && data["id"] == turn.ID { + t.Fatal("system orphan cleanup published turn.completed and may advance workflow") + } + } +} diff --git a/apps/backend/internal/task/service/archived_session_reconciliation.go b/apps/backend/internal/task/service/archived_session_reconciliation.go index 3825242676..5c6784112a 100644 --- a/apps/backend/internal/task/service/archived_session_reconciliation.go +++ b/apps/backend/internal/task/service/archived_session_reconciliation.go @@ -5,12 +5,19 @@ import ( "time" "github.com/kandev/kandev/internal/task/archivecascade" + "github.com/kandev/kandev/internal/task/models" "go.uber.org/zap" ) -// StartArchivedSessionReconciliationLoop starts a background goroutine that -// periodically re-finalizes archived tasks whose sessions never made it to a -// terminal DB state. +// StartSessionReconciliationLoop starts the background goroutine that +// periodically reconciles task sessions that no request path owns anymore. +// Each tick runs two passes: +// +// 1. The archived-task pass: re-finalize archived tasks whose sessions +// never made it to a terminal DB state (see runArchivedSessionReconciliation). +// 2. The active-task pass: detect and heal unarchived tasks holding active +// sessions whose backing execution is gone (see runActiveSessionSweep in +// active_session_stall.go). // // finalizeCancelledSessions (see service_tasks.go) already bounds its // session-cancellation retry to a handful of fixed attempts inside @@ -24,8 +31,8 @@ import ( // "is running" indicators off session.state_changed never learn the // sessions actually stopped. // -// This sweep is the eventual-consistency backstop for that residual gap, -// following the exact periodic-sweep shape StartAutoArchiveLoop already +// The archived pass is the eventual-consistency backstop for that residual +// gap, following the exact periodic-sweep shape StartAutoArchiveLoop already // uses: on a fixed interval, list every archived task that still has an // active session, and re-invoke finalizeCancelledSessions for each. Calling // finalizeCancelledSessions again is always safe — its underlying @@ -35,7 +42,7 @@ import ( // slice and is a no-op. As long as the process keeps running, a later pass // retries once the SQLite contention that exhausted the in-line retry has // cleared — unlike the fixed 3-attempt budget, this sweep never gives up. -func (s *Service) StartArchivedSessionReconciliationLoop(ctx context.Context) { +func (s *Service) StartSessionReconciliationLoop(ctx context.Context) { ticker := time.NewTicker(1 * time.Minute) go func() { defer ticker.Stop() @@ -43,12 +50,13 @@ func (s *Service) StartArchivedSessionReconciliationLoop(ctx context.Context) { select { case <-ctx.Done(): return - case <-ticker.C: + case now := <-ticker.C: s.runArchivedSessionReconciliation(ctx) + s.runActiveSessionSweep(ctx, now) } } }() - s.logger.Info("archived-session reconciliation loop started (every 1 minute)") + s.logger.Info("session reconciliation loop started (every 1 minute)") } func (s *Service) runArchivedSessionReconciliation(ctx context.Context) { @@ -78,6 +86,6 @@ func (s *Service) runArchivedSessionReconciliation(ctx context.Context) { // between the candidate list query and this read. continue } - s.finalizeCancelledSessions(reconcileCtx, taskID, activeSessions, deadline) + s.finalizeCancelledSessions(reconcileCtx, taskID, activeSessions, deadline, models.SessionArchiveCancelReason) } } diff --git a/apps/backend/internal/task/service/handoff_cascade_test.go b/apps/backend/internal/task/service/handoff_cascade_test.go index 1a0a3079ab..c799e0e37f 100644 --- a/apps/backend/internal/task/service/handoff_cascade_test.go +++ b/apps/backend/internal/task/service/handoff_cascade_test.go @@ -1037,6 +1037,10 @@ type fakeDeleteRepo struct { *fakeCascadeRepo } +func (r *fakeDeleteRepo) ListUnarchivedTasksWithActiveSessions(ctx context.Context) ([]*models.Task, error) { + return nil, nil +} + func (r *fakeDeleteRepo) DeleteTask(ctx context.Context, id string) error { _, err := r.DeleteTaskWithVacatedStep(ctx, id) return err diff --git a/apps/backend/internal/task/service/handoff_workspace_test.go b/apps/backend/internal/task/service/handoff_workspace_test.go index aa1cc10b8f..0b6ccd97fd 100644 --- a/apps/backend/internal/task/service/handoff_workspace_test.go +++ b/apps/backend/internal/task/service/handoff_workspace_test.go @@ -746,6 +746,10 @@ func (r *phase4TaskRepo) ListArchivedTasksWithActiveSessions(context.Context) ([ r.panicNotUsed("ListArchivedTasksWithActiveSessions") return nil, nil } +func (r *phase4TaskRepo) ListUnarchivedTasksWithActiveSessions(context.Context) ([]*models.Task, error) { + r.panicNotUsed("ListUnarchivedTasksWithActiveSessions") + return nil, nil +} func (r *phase4TaskRepo) ListExpiredQuickChatTasks(context.Context, time.Time) ([]*models.Task, error) { r.panicNotUsed("ListExpiredQuickChatTasks") return nil, nil diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index d1c4db6338..6a2f7e351d 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -125,6 +125,17 @@ type TaskExecutionStopper interface { RegisterExecutionStopOwner(sessionID, executionID string, force bool) } +// SessionExecutionRegistry reports which sessions of a task currently have a +// live in-memory execution registered by the agent runtime's execution store. +// The session reconciliation sweep uses it to tell an active session whose +// backing actor is alive from one whose actor is gone (e.g. after a backend +// restart), independent of the persisted session state. +type SessionExecutionRegistry interface { + // LiveSessionIDsForTask returns the session IDs under taskID that have a + // registered in-memory execution. The snapshot is read-only. + LiveSessionIDsForTask(taskID string) []string +} + // synchronousTaskExecutionStopper is an optional cleanup-only extension. The // normal StopSession contract schedules process teardown asynchronously, but // destructive resource cleanup must wait until the process exits. @@ -497,10 +508,19 @@ type Service struct { // host process detection. Nil selects the real platform implementation // (resource_cleanup_orphan_reap_host_*.go); tests override them // directly since they are unexported and this is a whitebox package. - orphanReapHostSnapshotter orphanReapHostSnapshotter - orphanReapVerifier orphanReapVerifier - orphanReapSignaler orphanReapSignaler - sessionRunningChecker SessionRunningChecker + orphanReapHostSnapshotter orphanReapHostSnapshotter + orphanReapVerifier orphanReapVerifier + orphanReapSignaler orphanReapSignaler + sessionRunningChecker SessionRunningChecker + sessionExecutionRegistry SessionExecutionRegistry + stallDetectionThreshold time.Duration + // stallNotifiedSessions dedupes task.stalled events per stall episode: + // task ID -> session IDs already reported. A session is reported at most + // once per episode; an episode ends when the session leaves the stalled + // set (terminal, healed, or a live execution reappeared), which clears + // its entry so a later stall on the same session reports again. Accessed + // only from the reconciliation sweep's single goroutine. + stallNotifiedSessions map[string]map[string]struct{} remoteBranchLister RemoteBranchLister repositorySelectionResolver RepositorySelectionResolver repoCloneLocation RepoCloneLocation @@ -746,6 +766,7 @@ func NewService(repos Repos, eventBus bus.EventBus, log *logger.Logger, discover branchFetcher: newBranchFetcher(log.Zap()), lastTaskActivity: make(map[string]v1.ForegroundActivity), lastTaskSubagentCount: make(map[string]int), + stallNotifiedSessions: make(map[string]map[string]struct{}), // Focused service tests do not run backend composition. Production // replaces this fallback with a database-allocated generation. pendingActionProjectionEpoch: "1", @@ -812,6 +833,28 @@ func (s *Service) SetExecutionStopper(stopper TaskExecutionStopper) { s.executionStopper = stopper } +// SetSessionExecutionRegistry wires the live-execution reader (agent runtime +// lifecycle manager) used by the session reconciliation sweep to distinguish +// an active session whose backing actor is still registered in this process +// from one whose actor is gone. Optional: without it the sweep's active-task +// pass (stall detection and orphaned-session healing) is skipped rather than +// guessing, because "no live execution" cannot be verified. +func (s *Service) SetSessionExecutionRegistry(registry SessionExecutionRegistry) { + s.sessionExecutionRegistry = registry +} + +// SetStallDetectionThreshold configures the event-silence window after which +// the session reconciliation sweep classifies an execution-less active +// session as stalled (tasks.stallDetectionThreshold, default 2h). Non-positive +// values keep the default. The orphaned-session healing grace window is +// derived from this threshold (twice its value). +func (s *Service) SetStallDetectionThreshold(threshold time.Duration) { + if threshold <= 0 { + return + } + s.stallDetectionThreshold = threshold +} + // SetClarificationCanceller wires terminal clarification cleanup for session // transitions owned by the task service. func (s *Service) SetClarificationCanceller(canceller TerminalClarificationCanceller) { diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index b9274ece75..3a00c2c572 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -2461,7 +2461,7 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { // 3b. Finalize active sessions in the DB and publish their cancellation // events. See finalizeCancelledSessions for the detailed rationale. - s.finalizeCancelledSessions(finalizeCtx, id, activeSessions, archiveDeadline) + s.finalizeCancelledSessions(finalizeCtx, id, activeSessions, archiveDeadline, models.SessionArchiveCancelReason) var postCommitErr error // 4. Re-read task for updated archived_at field. The archive row is @@ -2623,15 +2623,44 @@ func waitForCancellationRetry(ctx context.Context, delay time.Duration) bool { const maxCancelAttempts = 3 func (s *Service) cancelActiveTaskSessionsWithRetry( - ctx context.Context, taskID string, + ctx context.Context, taskID, reason string, sessionIDs []string, ) ([]*models.TaskSession, error) { const cancelRetryBackoff = 250 * time.Millisecond var cancelledSessions []*models.TaskSession var cancelErr error for attempt := 1; attempt <= maxCancelAttempts; attempt++ { - cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByTaskID( - ctx, taskID, models.SessionArchiveCancelReason, + if len(sessionIDs) == 0 { + cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByTaskID( + ctx, taskID, reason, + ) + } else { + cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByIDs( + ctx, taskID, sessionIDs, reason, + ) + } + if cancelErr == nil { + return cancelledSessions, nil + } + if attempt < maxCancelAttempts && !waitForCancellationRetry(ctx, cancelRetryBackoff) { + return nil, ctx.Err() + } + } + return nil, cancelErr +} + +func (s *Service) cancelActiveTaskSessionsByCandidatesWithRetry( + ctx context.Context, + taskID, reason string, + candidates []models.ActiveSessionCancellationCandidate, +) ([]*models.TaskSession, error) { + const cancelRetryBackoff = 250 * time.Millisecond + + var cancelledSessions []*models.TaskSession + var cancelErr error + for attempt := 1; attempt <= maxCancelAttempts; attempt++ { + cancelledSessions, cancelErr = s.sessions.CancelActiveTaskSessionsByCandidates( + ctx, taskID, candidates, reason, ) if cancelErr == nil { return cancelledSessions, nil @@ -2643,6 +2672,93 @@ func (s *Service) cancelActiveTaskSessionsWithRetry( return nil, cancelErr } +// finalizeCancelledSessionCandidates is the guarded cancellation path used +// by active-session healing. It closes the exact orphan turn that was part of +// the compare-and-set snapshot, and completes its pending tool calls without +// publishing turn.completed. Automatic system cancellation must not enter the +// normal turn/workflow completion path. +func (s *Service) finalizeCancelledSessionCandidates( + ctx context.Context, + taskID string, + activeSessions []*models.TaskSession, + candidates []models.ActiveSessionCancellationCandidate, + deadline time.Time, + reason string, +) { + cancelledSessions, cancelErr := s.cancelActiveTaskSessionsByCandidatesWithRetry( + ctx, taskID, reason, candidates, + ) + if cancelErr != nil { + s.logger.Error("failed to reap guarded active sessions after retries", + zap.String("task_id", taskID), + zap.Int("attempts", maxCancelAttempts), + zap.Error(cancelErr)) + return + } + if len(cancelledSessions) == 0 { + return + } + s.abandonCancelledSessionTurns(ctx, candidates, cancelledSessions) + s.logger.Info("reaped guarded active sessions", + zap.String("task_id", taskID), + zap.Int("count", len(cancelledSessions))) + s.runCancelledSessionEffects(ctx, taskID, activeSessions, cancelledSessions, deadline, reason) +} + +func (s *Service) abandonCancelledSessionTurns( + ctx context.Context, + candidates []models.ActiveSessionCancellationCandidate, + cancelledSessions []*models.TaskSession, +) { + turnIDs := make(map[string]string, len(candidates)) + for _, candidate := range candidates { + if candidate.SessionID != "" && candidate.ExpectedTurnID != "" { + turnIDs[candidate.SessionID] = candidate.ExpectedTurnID + } + } + for _, session := range cancelledSessions { + if session == nil { + continue + } + turnID := turnIDs[session.ID] + if turnID == "" { + continue + } + activeTurn, err := s.GetActiveTurn(ctx, session.ID) + if err != nil { + s.logger.Warn("failed to inspect orphan turn after session cancellation", + zap.String("session_id", session.ID), + zap.String("turn_id", turnID), + zap.Error(err)) + continue + } + if activeTurn == nil || activeTurn.ID != turnID { + // A different turn owns the session now. Never abandon it using a + // stale session-only lookup; the next reconciliation pass can + // inspect the new identity. + continue + } + if err := s.turns.AbandonTurn(ctx, turnID); err != nil { + s.logger.Warn("failed to abandon orphan turn after session cancellation", + zap.String("session_id", session.ID), + zap.String("turn_id", turnID), + zap.Error(err)) + continue + } + if affected, err := s.turns.CompletePendingToolCallsForTurn(ctx, turnID); err != nil { + s.logger.Warn("failed to complete pending tool calls for cancelled session turn", + zap.String("session_id", session.ID), + zap.String("turn_id", turnID), + zap.Error(err)) + } else if affected > 0 { + s.logger.Info("completed pending tool calls for cancelled session turn", + zap.String("session_id", session.ID), + zap.String("turn_id", turnID), + zap.Int64("affected", affected)) + } + } +} + // finalizeCancelledSessions finalizes an archived task's active sessions in // the DB and publishes a session.state_changed event for each one actually // cancelled. The async cleanup that follows tears down the agent processes; @@ -2652,6 +2768,12 @@ func (s *Service) cancelActiveTaskSessionsWithRetry( // exclusively by that event, and would otherwise show a live spinner // forever after archive. // +// reason is the TaskSession.ErrorMessage the cancellation carries ("task +// archived", "orphaned session", ...). Callers choose the value that names +// their path; models.IsArchiveCancelReason consumers key off the archive +// values for unarchive resume semantics, so a non-archive reason keeps the +// cancellation treated like an explicit stop. +// // CancelActiveTaskSessionsByTaskID is bounded by its own internal 10s // timeout, so a single attempt can time out under SQLite writer contention // alone — with archived_at already committed by the caller, that would @@ -2668,8 +2790,25 @@ func (s *Service) finalizeCancelledSessions( taskID string, activeSessions []*models.TaskSession, deadline time.Time, + reason string, ) { - cancelledSessions, cancelErr := s.cancelActiveTaskSessionsWithRetry(ctx, taskID) + s.finalizeCancelledSessionIDs(ctx, taskID, activeSessions, nil, deadline, reason) +} + +// finalizeCancelledSessionIDs is finalizeCancelledSessions with an optional +// session-ID scope: when sessionIDs is non-empty only those sessions are +// cancelled (a nil/empty scope means every active session of the task, the +// archive semantics). Callers that need activity and turn compare-and-set +// protection use finalizeCancelledSessionCandidates instead. +func (s *Service) finalizeCancelledSessionIDs( + ctx context.Context, + taskID string, + activeSessions []*models.TaskSession, + sessionIDs []string, + deadline time.Time, + reason string, +) { + cancelledSessions, cancelErr := s.cancelActiveTaskSessionsWithRetry(ctx, taskID, reason, sessionIDs) if cancelErr != nil { s.logger.Error("failed to reap active sessions on archive after retries", zap.String("task_id", taskID), @@ -2683,6 +2822,21 @@ func (s *Service) finalizeCancelledSessions( s.logger.Info("reaped active sessions on archive", zap.String("task_id", taskID), zap.Int("count", len(cancelledSessions))) + s.runCancelledSessionEffects(ctx, taskID, activeSessions, cancelledSessions, deadline, reason) +} + +// runCancelledSessionEffects owns every post-cancellation effect shared by +// the archive and heal paths: clarification expiry, parked-projection +// cleanup, session-ceiling release, and one session.state_changed per +// cancelled session. CancelledSessionEffects run on a detached-but-bounded +// context because the DB write already committed. +func (s *Service) runCancelledSessionEffects( + ctx context.Context, + taskID string, + activeSessions, cancelledSessions []*models.TaskSession, + deadline time.Time, + reason string, +) { // Detach from ctx via WithoutCancel: the DB write above already // committed on a detached context, so a client disconnect here must // not also suppress the event publish below — event-driven clients @@ -2731,7 +2885,7 @@ func (s *Service) finalizeCancelledSessions( s.sessionCeilingReleaser.ReleaseCeilingReservation(session.ID) } } - s.publishSessionsCancelled(detachedCtx, taskID, activeSessions, cancelledSessions, models.SessionArchiveCancelReason) + s.publishSessionsCancelled(detachedCtx, taskID, activeSessions, cancelledSessions, reason) } func (s *Service) registerTaskRuntimeStopOwners(stopTargets []taskStopTarget, force bool) { diff --git a/apps/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go index a693b64509..44f0e330f9 100644 --- a/apps/backend/internal/task/service/service_test.go +++ b/apps/backend/internal/task/service/service_test.go @@ -33,6 +33,9 @@ type MockEventBus struct { mu sync.Mutex publishedEvents []*bus.Event closed bool + // publishErrors, when non-nil, makes Publish fail for the listed + // subjects, for tests of publish-failure retry semantics. + publishErrors map[string]error } type recordingTaskClarificationCanceller struct { @@ -84,6 +87,9 @@ func NewMockEventBus() *MockEventBus { func (m *MockEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error { m.mu.Lock() defer m.mu.Unlock() + if err, inject := m.publishErrors[subject]; inject { + return err + } m.publishedEvents = append(m.publishedEvents, event) return nil } @@ -2428,7 +2434,7 @@ func TestFinalizeCancelledSessionsStopsRetryAtDeadline(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() - svc.finalizeCancelledSessions(ctx, "task-deadline", nil, time.Now().Add(20*time.Millisecond)) + svc.finalizeCancelledSessions(ctx, "task-deadline", nil, time.Now().Add(20*time.Millisecond), models.SessionArchiveCancelReason) if got := flaky.callCount(); got != 1 { t.Fatalf("CancelActiveTaskSessionsByTaskID call count = %d, want 1 after deadline", got) diff --git a/docs/plans/session-stall-visibility/plan.md b/docs/plans/session-stall-visibility/plan.md new file mode 100644 index 0000000000..9b340ca962 --- /dev/null +++ b/docs/plans/session-stall-visibility/plan.md @@ -0,0 +1,54 @@ +--- +created: 2026-09-20 +status: implemented +requirements: + - REQ-TASKS-SESSION-STALL-VISIBILITY-001 +system_design: + - ../../specs/tasks/system-design/session-stall-visibility.md +legacy_specs: [] +--- + +# Implementation Plan: Session Stall Visibility and Orphan Healing + +## Overview + +Close the silent-stall incident behind issue #3712 (and the sweep extension +of #3711) with one vertical correction in the session reconciliation sweep: +detect unarchived tasks holding active sessions whose backing execution is +gone, report each stall once per episode, and heal the orphans after a grace +window. Detection and healing share one pass; the fix plan approved in the +issue rejected review-specific watchdogs, verdict-sniffing, and +dashboard-only signaling. + +## Evidence and root cause + +Four review tasks (issue #3712, 2026-09-16) sat in `WAITING_FOR_INPUT` for +hours with finished agent reports in `task_session_messages` but no events +after creation. The same orphaned-turn failure as #3711: the turn-complete +actor died with a backend restart, the session row stayed in an active +state, and the event-driven stall warning ("no events received") never fired +because no events flowed at all. Reviews specifically never advance while +their session looks active, so the workflow also froze. + +## Work orders + +Sequential; one work order owns the sweep. + +- [x] [Task 01: Detect and heal orphaned active sessions in the sweep](task-01-stall-detection-and-healing.md) + +## Verification strategy + +Backend unit regressions colocated with the sweep +(`internal/task/service/active_session_stall_test.go`) and the repository +(`internal/task/repository/sqlite/`); no browser surface changes. See the +work order for exact commands and results. + +## Risks and non-goals + +- Healing is deliberately conservative: all-or-nothing per task, liveness + re-checked at the cancel boundary, ID-scoped writes. A false heal needs + every session of a task execution-less and silent past twice the + threshold while a live execution registers mid-sweep. +- No front-end consumption of `task.stalled` is included; any UI is a + separate initiative. +- Agent-runtime stall detection for live executions is unchanged. diff --git a/docs/plans/session-stall-visibility/task-01-stall-detection-and-healing.md b/docs/plans/session-stall-visibility/task-01-stall-detection-and-healing.md new file mode 100644 index 0000000000..d1fc64d44d --- /dev/null +++ b/docs/plans/session-stall-visibility/task-01-stall-detection-and-healing.md @@ -0,0 +1,88 @@ +--- +id: "01-stall-detection-and-healing" +title: "Detect and heal orphaned active sessions in the sweep" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +system_design: + - ../../specs/tasks/system-design/session-stall-visibility.md +requirements: + - REQ-TASKS-SESSION-STALL-VISIBILITY-001 +acceptance_criteria: + - AC-TASKS-SESSION-STALL-VISIBILITY-001.1 + - AC-TASKS-SESSION-STALL-VISIBILITY-001.2 + - AC-TASKS-SESSION-STALL-VISIBILITY-001.3 + - AC-TASKS-SESSION-STALL-VISIBILITY-001.4 + - AC-TASKS-SESSION-STALL-VISIBILITY-001.5 + - AC-TASKS-SESSION-STALL-VISIBILITY-001.6 +--- + +# Task 01: Detect and Heal Orphaned Active Sessions in the Sweep + +## Outcome + +The session reconciliation sweep's active-task pass detects and reports +execution-less, event-silent active sessions on unarchived tasks, and heals +them after a grace window, per +[REQ-TASKS-SESSION-STALL-VISIBILITY-001](../../specs/tasks/requirements/session-stall-visibility.md). + +## In scope + +- Active-task pass on the existing one-minute reconciliation tick. +- `task.stalled` event with one report per session per stall episode; + delivery recorded only after publish succeeds. +- Grace-window healing through the archived pass's cancelled-session + transition, scoped to the classified session IDs, with liveness re-checked + at the cancellation boundary. +- Turn persistence refreshes the session row's `updated_at` so + auto-dispatched turns reset the silence clock. +- `tasks.stallDetectionThreshold` startup setting with catalog, validation, + environment alias, and public documentation. + +## Exclusions + +- Front-end consumption of `task.stalled`; archived-task reconciliation; + live-execution stall detection; review-workflow advancement logic. + +## Acceptance conditions + +1. An orphaned, silent active session on an unarchived task emits exactly + one `task.stalled` per episode and is not healed inside the grace + window. +2. Healing never touches sessions with live executions or sessions outside + the classified set, and never runs while a sibling session is live or + merely stalled. +3. Read failures on the activity clock skip the task rather than + classifying from an incomplete clock; a missing registry skips the pass. + +## Verification + +```sh +cd apps/backend && go test ./internal/task/service/ -run 'TestService_ActiveSessionSweep' -count=1 +cd apps/backend && go test ./internal/task/repository/sqlite/ -run 'TestCancelActiveTaskSessionsByIDs|TestCreateTurnRefreshesSessionUpdatedAt' -count=1 +cd apps/backend && go test ./internal/common/config/ ./internal/events/... -count=1 +cd apps/backend && go vet ./internal/... +``` + +## Files touched + +- `internal/task/service/active_session_stall.go`, `active_session_stall_test.go` +- `internal/task/service/archived_session_reconciliation.go`, `service_tasks.go`, `service.go` +- `internal/task/repository/sqlite/{session,task,message}.go`, `session_test.go`, `task_active_sessions_test.go` +- `internal/task/repository/interface.go` +- `internal/events/types.go` +- `internal/common/config/{catalog,config,source,validation}.go` +- `internal/backendapp/{main,orchestrator,adapters}.go` +- `docs/public/configuration.md`, root `AGENTS.md` + +## Results + +Implemented; 13 sweep regressions (detection, live-execution skip, +threshold skip, episode dedupe, healing, sibling guard, archived skip, +registry fail-closed, mid-sweep registration abort, ID-scoped cancel, +per-session episode pruning, publish-retry, payload timing alignment) plus +repository regressions for the ID-scoped cancel and the turn-clock refresh +all pass. CI and review-driven remediations (session-scoped cancellation, +fail-closed activity reads, episode pruning, publish retry, turn clock) are +included; see PR #3832. diff --git a/docs/public/configuration.md b/docs/public/configuration.md index 0a79117d6f..e30d963287 100644 --- a/docs/public/configuration.md +++ b/docs/public/configuration.md @@ -159,6 +159,7 @@ The launcher starts `agentctl`, performs a one-time nonce handshake, and supplie | YAML key | Environment variable | Default | Current behavior | | -------------------------- | --------------------------------- | ------- | ------------------------------------------------------------------------------- | | `tasks.preparationTimeout` | `KANDEV_TASK_PREPARATION_TIMEOUT` | `10m` | Positive Go duration for repository setup and executor-profile prepare scripts. | +| `tasks.stallDetectionThreshold` | `KANDEV_TASK_STALL_DETECTION_THRESHOLD` | `2h` | Positive Go duration for the session reconciliation sweep's stall threshold. | `tasks.preparationTimeout` controls how long Kandev allows repository setup and executor-profile prepare scripts to run. The value uses Go duration syntax, @@ -176,6 +177,22 @@ context, so earlier work such as Sprite uploads does not reduce their full `10m` preparation budget. The environment variable overrides YAML. This is a startup setting, not a database or Settings value. +`tasks.stallDetectionThreshold` controls the periodic session reconciliation +sweep. When an unarchived task holds an active session with no live execution +behind it and no session events or messages for longer than this threshold, +Kandev emits a `task.stalled` event and logs a warning. After twice the +threshold of silence, the sweep cancels the orphaned sessions so the task +stops showing a phantom active session. The cancel is all-or-nothing per +task: it runs only when every active session of the task is orphaned and +past the grace window, so a session that still has a live execution blocks +the cancellation of its siblings rather than being swept along with them. +The value is source-specific: an invalid YAML duration fails configuration +parsing, while a zero or negative YAML duration is rejected at startup with +`tasks.stallDetectionThreshold must be positive`. An invalid or non-positive +environment value falls back to `2h`, and non-positive values from a profile +default are ignored, leaving the `2h` default in effect. The environment +variable overrides YAML, and changes require a backend restart. + ### Capacity and managed-process startup settings | YAML key | Environment variable | Type and default | Current behavior | @@ -439,6 +456,7 @@ agent: tasks: preparationTimeout: "10m" + stallDetectionThreshold: "2h" credentials: file: "" diff --git a/docs/specs/tasks/README.md b/docs/specs/tasks/README.md index 339c80a0e2..82bc6a2170 100644 --- a/docs/specs/tasks/README.md +++ b/docs/specs/tasks/README.md @@ -114,6 +114,7 @@ signals, and task-scoped scheduling contracts. - [Task Dependencies and Auto-Start Chains](requirements/task-dependencies.md) - [Dependency Gate Skip Visibility](requirements/dependency-gate-skip-visibility.md) - [Session Launch Repository Resolution](requirements/launch-repository-resolution.md) +- [Session stall visibility and orphan healing](requirements/session-stall-visibility.md) - [Task Launch Failure Recovery](requirements/task-launch-failure-recovery.md) - [Worktree metadata recovery](requirements/worktree-metadata-recovery.md) - [Task priority visibility](requirements/task-priority-visibility.md) @@ -194,6 +195,7 @@ signals, and task-scoped scheduling contracts. - [Dependency Gate Skip Visibility](system-design/dependency-gate-skip-visibility.md) - [Edit task dependencies](system-design/task-dependency-detail-editing.md) - [Session Launch Repository Resolution](system-design/launch-repository-resolution.md) +- [Session stall visibility and orphan healing](system-design/session-stall-visibility.md) - [Task Launch Failure Recovery](system-design/task-launch-failure-recovery.md) - [Worktree metadata recovery](system-design/worktree-metadata-recovery.md) - [Task Create Launch Preview](system-design/task-create-launch-preview.md) diff --git a/docs/specs/tasks/requirements/session-stall-visibility.md b/docs/specs/tasks/requirements/session-stall-visibility.md new file mode 100644 index 0000000000..a4c9a61d56 --- /dev/null +++ b/docs/specs/tasks/requirements/session-stall-visibility.md @@ -0,0 +1,78 @@ +--- +status: active +system: tasks +created: 2026-09-20 +owners: + - kandev +--- + +# Session Stall Visibility and Orphan Healing + +## Overview + +An unarchived task whose session is in an active DB state (CREATED, STARTING, +RUNNING, WAITING_FOR_INPUT) but whose backing execution is gone — after a +backend restart, a lost executor, or a launch that never registered — is +invisible to every event-driven surface: no request path owns the session, +no actor will ever advance it, and silence produces no signal. This document +defines the observable contract for detecting that state and healing it, +closing the incident where four finished review tasks sat silently in +`WAITING_FOR_INPUT` for hours with no operator-visible signal. + +## Terms + +- **Active session**: a `task_sessions` row in CREATED, STARTING, RUNNING, + or WAITING_FOR_INPUT. +- **Live execution**: an execution registered in the backend's in-memory + execution store for the session. +- **Event silence**: no persisted activity for the session — the newer of the + session row's last update and its newest persisted message. +- **Stall episode**: the continuous period during which one session is + active, execution-less, and event-silent past the stall threshold. An + episode ends when the session leaves that state (terminal transition, + healing, or a live execution reappearing). + +## Requirements + +### REQ-TASKS-SESSION-STALL-VISIBILITY-001: Stall detection and orphan healing + +**Intent:** A task whose session looks active but has no live backing +execution is surfaced promptly and eventually healed, instead of stalling +silently forever. + +#### Acceptance criteria + +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.1:** When an unarchived task holds + an active session with no live execution and no persisted session activity + for longer than the configurable stall threshold (default 2h), the system + shall emit a `task.stalled` event and a warning log naming the task and + its stalled sessions, at most once per session per stall episode. +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.2:** When a session gains a live + execution, transitions to a terminal state, or otherwise shows persisted + activity again, its stall episode shall end, and a later stall of the same + session shall be reported again as a new episode. +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.3:** When an active session with + no live execution has been silent for at least twice the stall threshold, + and every active session of the same task is equally orphaned and silent, + the system shall cancel those sessions with the reason `orphaned session` + and publish the session-state change event clients observe. +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.4:** The healing cancellation + shall never cancel a session with a live execution, including one that + registered while the sweep was evaluating the task, and shall never cancel + a session outside the set the sweep classified as orphaned. +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.5:** When a session's activity + clock cannot be read (for example a transient database read failure), the + system shall neither report a stall nor heal for that evaluation, rather + than classifying silence from an incomplete clock. +- **AC-TASKS-SESSION-STALL-VISIBILITY-001.6:** When the stall detection + threshold cannot be verified against the live-execution registry, the + sweep shall do nothing rather than guess that executions are gone. + +## Exclusions + +- Detecting stalls of sessions that still have a live execution: the + agent-runtime stall watchdog owns that case. +- Healing archived tasks' sessions: the archived-session reconciliation pass + owns that case. +- Front-end presentation of `task.stalled`: any dashboard or notification UI + built on the event is a separate initiative. diff --git a/docs/specs/tasks/system-design/session-stall-visibility.md b/docs/specs/tasks/system-design/session-stall-visibility.md new file mode 100644 index 0000000000..b1bb8e7de8 --- /dev/null +++ b/docs/specs/tasks/system-design/session-stall-visibility.md @@ -0,0 +1,95 @@ +--- +status: current +system: tasks +created: 2026-09-20 +owners: + - kandev +requirements: + - REQ-TASKS-SESSION-STALL-VISIBILITY-001 +--- + +# Session Stall Visibility and Orphan Healing Design + +## Context and boundaries + +The [session reconciliation sweep](#control-flow) in +`internal/task/service/` is the single owner of this behavior. It already +reconciles archived tasks' sessions +(`archived_session_reconciliation.go`); the active-task pass +(`active_session_stall.go`) extends the same one-minute tick to unarchived +tasks holding active sessions. The agent runtime's in-memory execution store +is the liveness oracle, injected as `SessionExecutionRegistry` +(`service.go`); the task system never guesses liveness from DB state alone. + +Archived tasks remain owned by the archived pass. Sessions with a live +execution remain owned by the agent runtime's own stall watchdog; the sweep +never reports them. + +## Control flow + +Each tick: + +1. `ListUnarchivedTasksWithActiveSessions` returns candidate tasks. +2. Per task, `ListActiveTaskSessionsByTaskID` loads the active sessions and + the registry reports live ones (`LiveSessionIDsForTask`); sessions + outside that snapshot are orphaned. +3. Each orphaned session's activity clock is the newer of its row's + `updated_at` and its newest `task_session_messages` row + (`GetLastMessageTimeBySessionIDs`). A read failure skips the task this + tick entirely (AC .5). +4. Silence past `tasks.stallDetectionThreshold` (default 2h; healed grace is + twice that) classifies the session as stalled; silence past the grace + window also classifies it as healable. +5. Stalled sessions not yet reported in this episode publish one + `task.stalled` event; delivery is recorded only after the publish + succeeds, so a failed publish retries next tick (AC .1). +6. A healable set covering every active session of the task re-checks + liveness at the cancellation boundary and then cancels exactly the + classified session IDs through `finalizeCancelledSessionIDs`, reusing the + archived pass's transition: `CANCELLED` with `SessionOrphanedCancelReason` + (`"orphaned session"`), clarification expiry, parked-projection cleanup, + session-ceiling release, and one `session.state_changed` per session + (AC .3, .4). + +## Session activity clock + +Turn persistence (`CreateTurn`, `CreateTurnWithStepStamp`) refreshes the +parent `task_sessions.updated_at` in the same statement batch, so an +auto-dispatched turn with no user message still resets the silence clock. +Message writes are already covered because the clock reads the newest +message directly. + +## Episode dedupe + +`stallNotifiedSessions` maps task ID to the sessions reported in the open +episode. Entries are pruned per session when it leaves the stalled set, so +a recovered session that stalls again reports again (AC .2). The map lives +on the sweep's single goroutine and resets when no candidates remain. + +## Failure and recovery behavior + +- No registry wired: the pass skips entirely (AC .6). +- Activity read failure: the task is skipped this tick (AC .5). +- Cancellation write failure: retried with the archived pass's bounded + in-line retry, and the next tick retries again. +- A live execution appearing between classification and the cancellation + boundary aborts the heal; the ID-scoped predicate structurally excludes + sessions that became active after classification (AC .4). + +## Configuration + +`tasks.stallDetectionThreshold` / `KANDEV_TASK_STALL_DETECTION_THRESHOLD` +(default `2h`), a startup setting in the typed config catalog +(`internal/common/config/`). Invalid YAML durations fail parsing; zero or +negative YAML durations are rejected at startup; invalid or non-positive +environment values fall back to `2h`. + +## Requirement mapping + +| Requirement | Satisfied by | +| --- | --- | +| REQ-TASKS-SESSION-STALL-VISIBILITY-001 (AC .1, .2) | Control flow steps 4-5; Episode dedupe | +| REQ-TASKS-SESSION-STALL-VISIBILITY-001 (AC .3) | Control flow step 6 | +| REQ-TASKS-SESSION-STALL-VISIBILITY-001 (AC .4) | Cancellation boundary re-check; ID-scoped cancel | +| REQ-TASKS-SESSION-STALL-VISIBILITY-001 (AC .5) | Control flow step 3 | +| REQ-TASKS-SESSION-STALL-VISIBILITY-001 (AC .6) | Failure and recovery: no registry |