fix(tasks): detect stalled sessions and heal orphaned ones in sweep (#3832)

Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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
 

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 {

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)
 
 	// ============================================

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

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"},

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"},

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.

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)

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")...)

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

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

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
 }

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{} {

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

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)

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 == "" {

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

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()

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

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)
+	}
+}

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)
+	}
+}

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{})
+}

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")
+		}
+	}
+}

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)
 	}
 }

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

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

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) {

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) {

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)

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.

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.

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: ""

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)

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.

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 |