From cd5199cd14679fbc9423ca7d1df49bacd9c85a43 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Fiere?= Date: Fri, 7 Aug 2026 18:17:57 +0200 Subject: [PATCH] feat: show red interrupted icon for tasks cut short by a restart (#2358) --- .../internal/backendapp/boot_state_routes.go | 1 + .../orchestrator/event_handlers_streaming.go | 40 ++++ .../event_handlers_streaming_test.go | 91 +++++++++ apps/backend/internal/orchestrator/service.go | 25 +++ .../orchestrator/task_operations_test.go | 78 ++++++++ .../internal/task/dto/converters_test.go | 46 +++++ apps/backend/internal/task/dto/dto.go | 5 + apps/backend/internal/task/models/models.go | 7 + .../internal/task/repository/sqlite/task.go | 29 +++ .../internal/task/repository/task_wip_test.go | 47 +++++ apps/backend/pkg/api/v1/task.go | 11 +- apps/web/app/tasks/[id]/kanban-task-shell.tsx | 2 + apps/web/app/tasks/rich-task-list-row.tsx | 13 +- apps/web/components/kanban-card-content.tsx | 12 +- apps/web/components/kanban-card.tsx | 2 + .../components/kanban/graph2-step-node.tsx | 13 +- .../kanban/swimlane-graph-content.tsx | 26 ++- .../session-task-switcher-sheet-hooks.ts | 7 +- apps/web/components/task/task-item.test.tsx | 49 +++++ apps/web/components/task/task-item.tsx | 24 ++- .../task/task-session-sidebar-item.ts | 5 + .../components/task/task-state-actions.tsx | 10 +- .../task/task-switcher-click.test.ts | 16 ++ .../web/components/task/task-switcher-row.tsx | 1 + .../components/task/task-switcher-types.ts | 2 + .../tests/task/task-interrupted-icon.spec.ts | 97 ++++++++++ apps/web/lib/kanban/map-task.test.ts | 12 ++ apps/web/lib/kanban/map-task.ts | 3 + apps/web/lib/state/slices/kanban/types.ts | 2 + apps/web/lib/types/http.ts | 3 + apps/web/lib/ui/state-icons.test.tsx | 170 ++++++++++++---- apps/web/lib/ui/state-icons.tsx | 109 +++++++++-- apps/web/lib/ws/handlers/kanban.ts | 2 + .../lib/ws/handlers/tasks-activity.test.ts | 85 +++++--- apps/web/lib/ws/handlers/tasks.ts | 45 +++-- apps/web/src/locales/en/common.json | 1 + apps/web/src/locales/pseudo/common.json | 1 + docs/plans/interrupted-task-indicator/plan.md | 182 ++++++++++++++++++ .../task-01-backend-startup-marker.md | 68 +++++++ .../task-02-backend-dto-and-clearing.md | 72 +++++++ .../task-03-frontend-data-plumbing.md | 71 +++++++ .../task-04-frontend-icon-rendering.md | 75 ++++++++ .../task-05-e2e-verification.md | 68 +++++++ docs/specs/INDEX.md | 1 + .../specs/tasks/interrupted-task-indicator.md | 169 ++++++++++++++++ 45 files changed, 1657 insertions(+), 141 deletions(-) create mode 100644 apps/web/e2e/tests/task/task-interrupted-icon.spec.ts create mode 100644 docs/plans/interrupted-task-indicator/plan.md create mode 100644 docs/plans/interrupted-task-indicator/task-01-backend-startup-marker.md create mode 100644 docs/plans/interrupted-task-indicator/task-02-backend-dto-and-clearing.md create mode 100644 docs/plans/interrupted-task-indicator/task-03-frontend-data-plumbing.md create mode 100644 docs/plans/interrupted-task-indicator/task-04-frontend-icon-rendering.md create mode 100644 docs/plans/interrupted-task-indicator/task-05-e2e-verification.md create mode 100644 docs/specs/tasks/interrupted-task-indicator.md diff --git a/apps/backend/internal/backendapp/boot_state_routes.go b/apps/backend/internal/backendapp/boot_state_routes.go index b40fc94e0a..c6c71f468c 100644 --- a/apps/backend/internal/backendapp/boot_state_routes.go +++ b/apps/backend/internal/backendapp/boot_state_routes.go @@ -572,6 +572,7 @@ func mapKanbanTaskState(task taskdto.TaskDTO) map[string]any { "primarySessionState": task.PrimarySessionState, "primarySessionPendingAction": task.PrimarySessionPendingAction, "taskPendingAction": task.TaskPendingAction, + "interrupted": task.Interrupted, "statusSummary": task.StatusSummary, "sessionCount": task.SessionCount, "reviewStatus": nullString(string(task.ReviewStatus)), diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index f8b129e9e4..3ffc509c44 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -836,6 +836,12 @@ func (s *Service) updateTaskSessionStateWithHook( if onChanged != nil { onChanged() } + // Work has resumed: a session entering STARTING/RUNNING clears the + // startup interruption marker and republishes the task so open clients + // drop the red interruption icon. No-op when the marker is absent. + if nextState == models.TaskSessionStateStarting || nextState == models.TaskSessionStateRunning { + s.clearTaskInterruptedMarker(ctx, taskID) + } if authoritativeUpdatedAt == nil { s.logger.Warn("skipping session state_changed publish; could not read authoritative updated_at", zap.String("task_id", taskID), @@ -1412,12 +1418,46 @@ func (s *Service) setSessionStarting( s.writeTaskInProgressForRuntime(ctx, taskID, session.ID) } + // The launch path moves a session to STARTING without going through + // updateTaskSessionStateWithHook, so clear the interruption marker here + // too (no-op when absent). + s.clearTaskInterruptedMarker(ctx, taskID) + if publishSession != nil { s.publishTaskSessionStateChanged(ctx, taskID, session.ID, oldState, session.State, session.ErrorMessage, stateUpdatedAt, publishSession) } return nil } +// clearTaskInterruptedMarker removes the startup interruption marker from a +// task and republishes task.updated when it was actually present, so open +// clients drop the red interruption icon. Called from the session-start +// funnel when a session enters STARTING/RUNNING — work has resumed, so the +// task is no longer interrupted. No-op when the marker is absent. +func (s *Service) clearTaskInterruptedMarker(ctx context.Context, taskID string) { + if taskID == "" { + return + } + removed, err := s.repo.RemoveTaskMetadataKey(ctx, taskID, models.MetaKeyInterruptedAt) + if err != nil { + s.logger.Warn("failed to clear interrupted marker", + zap.String("task_id", taskID), + zap.Error(err)) + return + } + if !removed { + return + } + task, err := s.repo.GetTask(ctx, taskID) + if err != nil || task == nil { + s.logger.Warn("failed to load task for interrupted-clear publish", + zap.String("task_id", taskID), + zap.Error(err)) + return + } + s.publishTaskUpdated(ctx, task) +} + func (s *Service) persistFullTaskSessionIfCurrent( ctx context.Context, session *models.TaskSession, diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index 845547805c..3600b24b18 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -2100,6 +2100,97 @@ func TestSetSessionStartingCanDeferTaskInProgress(t *testing.T) { require.Empty(t, taskRepo.stateWrites) } +// recordingTaskUpdatedPublisher captures task.updated publishes so tests can +// assert the interrupted-marker clear republishes the task. +type recordingTaskUpdatedPublisher struct { + updatedTaskIDs []string +} + +func (r *recordingTaskUpdatedPublisher) PublishTaskUpdated(_ context.Context, task *models.Task, _ ...string) { + if task != nil { + r.updatedTaskIDs = append(r.updatedTaskIDs, task.ID) + } +} +func (r *recordingTaskUpdatedPublisher) PublishTaskStateChanged(context.Context, *models.Task, v1.TaskState) { +} +func (r *recordingTaskUpdatedPublisher) PublishTaskActivityIfChanged(context.Context, string) {} + +func seedInterruptedMarker(t *testing.T, repo *sqliterepo.Repository, taskID string) { + t.Helper() + if err := repo.SetTaskMetadataKey(context.Background(), taskID, models.MetaKeyInterruptedAt, time.Now().UTC().Format(time.RFC3339)); err != nil { + t.Fatalf("seed interrupted marker: %v", err) + } +} + +func TestSessionStartClearsInterruptedMarker(t *testing.T) { + ctx := context.Background() + + t.Run("state hook transition to STARTING clears and republishes", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedInterruptedMarker(t, repo, "t1") + + publisher := &recordingTaskUpdatedPublisher{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetTaskEventPublisher(publisher) + + updated, changed := svc.updateTaskSessionStateWithHook( + ctx, "t1", "s1", models.TaskSessionStateStarting, "", false, nil, + ) + require.NotNil(t, updated) + require.True(t, changed) + + task, err := repo.GetTask(ctx, "t1") + require.NoError(t, err) + _, marked := task.Metadata[models.MetaKeyInterruptedAt] + require.False(t, marked, "marker must be cleared when the session enters STARTING") + require.Equal(t, []string{"t1"}, publisher.updatedTaskIDs, + "task.updated must be republished after clearing the marker") + }) + + t.Run("launch path via setSessionStarting clears and republishes", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + seedInterruptedMarker(t, repo, "t1") + + session, err := repo.GetTaskSession(ctx, "s1") + require.NoError(t, err) + session.State = models.TaskSessionStateStarting + session.ErrorMessage = "" + session.UpdatedAt = time.Now().UTC() + + publisher := &recordingTaskUpdatedPublisher{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetTaskEventPublisher(publisher) + + require.NoError(t, svc.setSessionStarting(ctx, "t1", session, models.TaskSessionStateRunning, true)) + + task, err := repo.GetTask(ctx, "t1") + require.NoError(t, err) + _, marked := task.Metadata[models.MetaKeyInterruptedAt] + require.False(t, marked, "marker must be cleared on the launch path") + require.Equal(t, []string{"t1"}, publisher.updatedTaskIDs, + "task.updated must be republished after clearing the marker") + }) + + t.Run("no marker means no republish", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + publisher := &recordingTaskUpdatedPublisher{} + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.SetTaskEventPublisher(publisher) + + updated, changed := svc.updateTaskSessionStateWithHook( + ctx, "t1", "s1", models.TaskSessionStateStarting, "", false, nil, + ) + require.NotNil(t, updated) + require.True(t, changed) + require.Empty(t, publisher.updatedTaskIDs, + "no task.updated may be published when no marker was removed") + }) +} + func TestSetSessionStartingRejectsTerminalSession(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index bef3b29eb7..4a21866996 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -246,6 +246,16 @@ type sessionExecutorStore interface { // than an idempotent same-owner observation. ClaimTaskTitleSession(ctx context.Context, taskID, sessionID string) (owned bool, newlyClaimed bool, err error) UpdateTask(ctx context.Context, task *models.Task) error + // SetTaskMetadataKey / RemoveTaskMetadataKey are concurrent-key-safe JSON + // patch helpers on tasks.metadata (implemented by the sqlite/Postgres + // repository). Used by startup reconciliation to mark interrupted tasks + // and by the session-start funnel to clear the marker. + SetTaskMetadataKey(ctx context.Context, taskID, key string, value interface{}) error + // SetTaskMetadataKeyIfNotArchived writes one metadata key only when the + // task row still has archived_at IS NULL (single statement — the archive + // check and the write cannot be separated by a concurrent archive). + SetTaskMetadataKeyIfNotArchived(ctx context.Context, taskID, key string, value interface{}) (bool, error) + RemoveTaskMetadataKey(ctx context.Context, taskID, key string) (bool, error) ListChildCompletionRows(ctx context.Context, parentID string) ([]models.ChildCompletionRow, error) // Git snapshots and commits GetLatestGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) @@ -1811,6 +1821,21 @@ func (s *Service) reconcileOneSessionOnStartup(ctx context.Context, running *mod } } + // Mark the task interrupted when its session was mid-turn when the backend + // died (STARTING/RUNNING) so task-list surfaces can show the red + // interruption icon. WAITING_FOR_INPUT sessions were idle, not interrupted. + // The write is archive-atomic (SetTaskMetadataKeyIfNotArchived), so an + // archive that commits after this check cannot leave a stale marker on an + // archived task. The marker is cleared when a session of the task next + // enters STARTING/RUNNING (see updateTaskSessionStateWithHook). + if running.TaskID != "" && (previousState == models.TaskSessionStateStarting || previousState == models.TaskSessionStateRunning) { + if _, setErr := s.repo.SetTaskMetadataKeyIfNotArchived(ctx, running.TaskID, models.MetaKeyInterruptedAt, time.Now().UTC().Format(time.RFC3339)); setErr != nil { + s.logger.Warn("failed to mark task interrupted on startup", + zap.String("task_id", running.TaskID), + zap.Error(setErr)) + } + } + // PRESERVE the ExecutorRunning record — it holds the resume token and worktree info // needed for lazy recovery when the user opens the session. But make the row // TRUE: after a restart the spawned process is normally gone, so if the local diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 2dd4ee9778..a8564b6c17 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -6382,6 +6382,84 @@ func TestReconcileSessionsOnStartup(t *testing.T) { t.Fatalf("expected REVIEW write to use UpdateTaskStateIfCurrentIn, got %d unconditional UpdateTaskState call(s)", n) } }) + + // Interrupted-task marker: sessions that were mid-turn (STARTING/RUNNING) + // when the backend died must mark their task with the interrupted_at + // metadata key so the task DTO reports interrupted and the task list can + // show the red icon. Idle and archived tasks must never be marked. + for _, tc := range []struct { + name string + sessionState models.TaskSessionState + wantInterrupt bool + }{ + {"running_session_marks_task_interrupted", models.TaskSessionStateRunning, true}, + {"starting_session_marks_task_interrupted", models.TaskSessionStateStarting, true}, + {"waiting_for_input_session_not_marked", models.TaskSessionStateWaitingForInput, false}, + } { + t.Run(tc.name, func(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + seedTaskAndSession(t, repo, "task1", "session1", tc.sessionState) + + err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "er1", + SessionID: "session1", + TaskID: "task1", + CreatedAt: now, + UpdatedAt: now, + }) + if err != nil { + t.Fatalf("failed to upsert executor running: %v", err) + } + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + svc.reconcileSessionsOnStartup(ctx) + + task, err := repo.GetTask(ctx, "task1") + if err != nil { + t.Fatalf("failed to load task: %v", err) + } + _, marked := task.Metadata[models.MetaKeyInterruptedAt] + if marked != tc.wantInterrupt { + t.Fatalf("interrupted_at marker present=%v, want %v", marked, tc.wantInterrupt) + } + }) + } + + t.Run("archived_running_session_not_marked", func(t *testing.T) { + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning) + if err := repo.ArchiveTask(ctx, "task1"); err != nil { + t.Fatalf("failed to archive task: %v", err) + } + + err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "er1", + SessionID: "session1", + TaskID: "task1", + CreatedAt: now, + UpdatedAt: now, + }) + if err != nil { + t.Fatalf("failed to upsert executor running: %v", err) + } + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + svc.reconcileSessionsOnStartup(ctx) + + task, err := repo.GetTask(ctx, "task1") + if err != nil { + t.Fatalf("failed to load task: %v", err) + } + if _, marked := task.Metadata[models.MetaKeyInterruptedAt]; marked { + t.Fatal("archived task must not be marked interrupted") + } + }) } // --- ensureSessionRunning: prepared workspace --- diff --git a/apps/backend/internal/task/dto/converters_test.go b/apps/backend/internal/task/dto/converters_test.go index c72ae8b5c6..250d7c0080 100644 --- a/apps/backend/internal/task/dto/converters_test.go +++ b/apps/backend/internal/task/dto/converters_test.go @@ -133,3 +133,49 @@ func TestFromTaskSession_IncludesAllWorktrees(t *testing.T) { t.Fatalf("summary WorkspacePath = %q, want /task-root", summary.WorkspacePath) } } + +func TestFromTask_DerivesInterruptedFromMetadata(t *testing.T) { + now := time.Now().UTC() + + t.Run("marked", func(t *testing.T) { + task := &models.Task{ + ID: "t1", + CreatedAt: now, + UpdatedAt: now, + Metadata: map[string]interface{}{ + models.MetaKeyInterruptedAt: "2026-08-02T10:00:00Z", + }, + } + got := FromTask(task) + if !got.Interrupted { + t.Fatal("Interrupted = false, want true when interrupted_at metadata is present") + } + }) + + t.Run("unmarked", func(t *testing.T) { + task := &models.Task{ID: "t1", CreatedAt: now, UpdatedAt: now} + got := FromTask(task) + if got.Interrupted { + t.Fatal("Interrupted = true, want false when interrupted_at metadata is absent") + } + }) +} + +func TestTaskToAPI_DerivesInterruptedFromMetadata(t *testing.T) { + now := time.Now().UTC() + + marked := (&models.Task{ + ID: "t1", + CreatedAt: now, + UpdatedAt: now, + Metadata: map[string]interface{}{models.MetaKeyInterruptedAt: "2026-08-02T10:00:00Z"}, + }).ToAPI() + if !marked.Interrupted { + t.Fatal("ToAPI Interrupted = false, want true when interrupted_at metadata is present") + } + + unmarked := (&models.Task{ID: "t1", CreatedAt: now, UpdatedAt: now}).ToAPI() + if unmarked.Interrupted { + t.Fatal("ToAPI Interrupted = true, want false when interrupted_at metadata is absent") + } +} diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index 99d7c1e893..d34316f4d4 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -182,6 +182,10 @@ type TaskDTO struct { CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` Metadata map[string]interface{} `json:"metadata,omitempty"` + // Interrupted reports that the task's session was mid-turn when the backend + // died and has not been resumed since. Derived from the interrupted_at + // metadata key at DTO conversion time (see FromTaskWithSessionInfo). + Interrupted bool `json:"interrupted,omitempty"` // Office extensions AssigneeAgentProfileID string `json:"assignee_agent_profile_id,omitempty"` @@ -757,6 +761,7 @@ func FromTaskWithSessionInfo( CreatedAt: task.CreatedAt, UpdatedAt: task.UpdatedAt, Metadata: task.Metadata, + Interrupted: task.Metadata[models.MetaKeyInterruptedAt] != nil, // Office extensions. AssigneeAgentProfileID is a read-time // projection from workflow_step_participants (ADR 0005 Wave F); // the repo's task SELECTs hydrate it via a correlated subquery. diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 7abda86948..783bc46d48 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -110,6 +110,12 @@ const ( // because the winner will (or already did) handle it. // Absent on ordinary (non-watcher) auto-start tasks, which launch normally. MetaKeyAutoStartClaimed = "auto_start_claimed" + // MetaKeyInterruptedAt is set by startup reconciliation when a task's + // session was mid-turn (STARTING/RUNNING) when the backend died. Its + // presence makes the task DTO report `interrupted: true` so task-list + // surfaces show the red interruption icon; the orchestrator removes the + // key when a session of the task next enters STARTING/RUNNING. + MetaKeyInterruptedAt = "interrupted_at" // MetaKeyAgentTitlePending marks tasks created in prompt-first mode whose // provisional title still needs the first eligible agent session to replace it. MetaKeyAgentTitlePending = "agent_title_pending" @@ -1798,6 +1804,7 @@ func (t *Task) ToAPI() *v1.Task { CreatedAt: t.CreatedAt, UpdatedAt: t.UpdatedAt, Metadata: t.Metadata, + Interrupted: t.Metadata[MetaKeyInterruptedAt] != nil, IsEphemeral: t.IsEphemeral, ParentID: t.ParentID, } diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index e54c93fab1..a6c5a514ca 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -653,6 +653,35 @@ func (r *Repository) SetTaskMetadataKey(ctx context.Context, taskID, key string, return err } +// SetTaskMetadataKeyIfNotArchived writes one metadata key atomically with the +// archived_at guard: the write only lands when the task row still has +// archived_at IS NULL, and reports whether it landed. Startup reconciliation +// uses it for the interrupted_at marker so an archive that commits between the +// guard read and the metadata write can never leave a marker on an archived +// task (the check and the write are one statement). +func (r *Repository) SetTaskMetadataKeyIfNotArchived(ctx context.Context, taskID, key string, value interface{}) (bool, error) { + payload, err := json.Marshal(value) + if err != nil { + return false, err + } + var query string + if dialect.IsPostgres(r.db.DriverName()) { + query = `UPDATE tasks SET metadata = jsonb_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END, ARRAY[?]::text[], ?::jsonb, true)::text, updated_at = ? WHERE id = ? AND archived_at IS NULL` + } else { + query = `UPDATE tasks SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ? WHERE id = ? AND archived_at IS NULL` + } + path := key + if !dialect.IsPostgres(r.db.DriverName()) { + path = jsonPath(key) + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(query), path, string(payload), time.Now().UTC(), taskID) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + return rows > 0, err +} + func jsonPath(key string) string { return "$." + key } func agentTitlePendingPredicate(driver string) string { diff --git a/apps/backend/internal/task/repository/task_wip_test.go b/apps/backend/internal/task/repository/task_wip_test.go index 0ca397e755..f89967c773 100644 --- a/apps/backend/internal/task/repository/task_wip_test.go +++ b/apps/backend/internal/task/repository/task_wip_test.go @@ -287,3 +287,50 @@ func TestTaskMetadataKeyHelpersRoundTripNestedValue(t *testing.T) { t.Fatalf("deferred launch metadata still present: %#v", got.Metadata) } } + +// TestSetTaskMetadataKeyIfNotArchivedSkipsArchivedTasks pins the archive-atomic +// contract of the interrupted-marker write: an archive that commits between a +// guard read and the metadata write must not leave a marker on an archived +// task, so the conditional write itself must be the guard (the check and the +// write are one statement). +func TestSetTaskMetadataKeyIfNotArchivedSkipsArchivedTasks(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + live := &models.Task{ID: "live-task", WorkspaceID: "metadata-workspace", Title: "Live"} + if err := repo.CreateTask(ctx, live); err != nil { + t.Fatalf("create live task: %v", err) + } + archived := &models.Task{ID: "archived-task", WorkspaceID: "metadata-workspace", Title: "Archived"} + if err := repo.CreateTask(ctx, archived); err != nil { + t.Fatalf("create archived task: %v", err) + } + if err := repo.ArchiveTask(ctx, archived.ID); err != nil { + t.Fatalf("archive task: %v", err) + } + + changed, err := repo.SetTaskMetadataKeyIfNotArchived(ctx, live.ID, models.MetaKeyInterruptedAt, "2026-08-07T00:00:00Z") + if err != nil || !changed { + t.Fatalf("live task write changed=%t err=%v, want true/nil", changed, err) + } + changed, err = repo.SetTaskMetadataKeyIfNotArchived(ctx, archived.ID, models.MetaKeyInterruptedAt, "2026-08-07T00:00:00Z") + if err != nil || changed { + t.Fatalf("archived task write changed=%t err=%v, want false/nil", changed, err) + } + + got, err := repo.GetTask(ctx, archived.ID) + if err != nil { + t.Fatalf("reload archived task: %v", err) + } + if _, marked := got.Metadata[models.MetaKeyInterruptedAt]; marked { + t.Fatalf("archived task must not carry the interrupted marker: %#v", got.Metadata) + } + got, err = repo.GetTask(ctx, live.ID) + if err != nil { + t.Fatalf("reload live task: %v", err) + } + if _, marked := got.Metadata[models.MetaKeyInterruptedAt]; !marked { + t.Fatalf("live task must carry the interrupted marker: %#v", got.Metadata) + } +} diff --git a/apps/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go index bf9bf72f31..0b9bfd21e6 100644 --- a/apps/backend/pkg/api/v1/task.go +++ b/apps/backend/pkg/api/v1/task.go @@ -148,9 +148,14 @@ type Task struct { StartedAt *time.Time `json:"started_at,omitempty"` CompletedAt *time.Time `json:"completed_at,omitempty"` Metadata map[string]interface{} `json:"metadata,omitempty"` - IsEphemeral bool `json:"is_ephemeral"` // Ephemeral tasks are not shown in kanban, used for quick chat - ParentID string `json:"parent_id,omitempty"` // FK to parent task for subtasks - Identifier string `json:"identifier,omitempty"` + // Interrupted reports that the task's session was mid-turn (STARTING/RUNNING) + // when the backend died and has not been resumed since. Derived from the + // interrupted_at metadata key at DTO conversion time; the orchestrator + // clears it when a session of the task next enters STARTING/RUNNING. + Interrupted bool `json:"interrupted,omitempty"` + IsEphemeral bool `json:"is_ephemeral"` // Ephemeral tasks are not shown in kanban, used for quick chat + ParentID string `json:"parent_id,omitempty"` // FK to parent task for subtasks + Identifier string `json:"identifier,omitempty"` } // TaskRepositoryInput for creating/updating task repositories diff --git a/apps/web/app/tasks/[id]/kanban-task-shell.tsx b/apps/web/app/tasks/[id]/kanban-task-shell.tsx index 371734e1c9..fa29029f0a 100644 --- a/apps/web/app/tasks/[id]/kanban-task-shell.tsx +++ b/apps/web/app/tasks/[id]/kanban-task-shell.tsx @@ -116,6 +116,7 @@ function simpleTaskHeaderData(task: Task | null) { title: task?.title ?? t("tasks:loading"), state: task?.state ?? null, foregroundActivity: task?.foreground_activity, + interrupted: task?.interrupted ?? false, }; } @@ -130,6 +131,7 @@ function SimpleTaskHeaderRow({ task }: { task: Task | null }) { foregroundActivity={data.foregroundActivity} hasPendingClarification={pendingInput.clarification} hasPendingPermission={pendingInput.permission} + interrupted={data.interrupted} /> - {getTaskStateIcon( - task.state, - "h-4 w-4 shrink-0", - pendingInput.clarification, - task.foreground_activity, - pendingInput.permission, - )} + {getTaskStateIcon(task.state, "h-4 w-4 shrink-0", { + hasPendingClarification: pendingInput.clarification, + foregroundActivity: task.foreground_activity, + hasPendingPermission: pendingInput.permission, + interrupted: task.interrupted, + })} {task.title} diff --git a/apps/web/components/kanban-card-content.tsx b/apps/web/components/kanban-card-content.tsx index e10b6ba243..f21ccffe35 100644 --- a/apps/web/components/kanban-card-content.tsx +++ b/apps/web/components/kanban-card-content.tsx @@ -284,9 +284,10 @@ export function renderTaskStatusIcon( const showQuestionIcon = shouldUseQuestionTaskIcon(task.state, hasPendingClarification); const showPermissionIcon = shouldUsePermissionTaskIcon(hasPendingPermission); const needsMe = showQuestionIcon || showPermissionIcon; + const showInterrupted = !!task.interrupted; const hasActivity = task.foregroundActivity === "generating" || task.foregroundActivity === "background"; - if (!showRunningSpinner && !needsMe && !hasActivity) { + if (!showRunningSpinner && !needsMe && !hasActivity && !showInterrupted) { return null; } // A "needs me" prompt (pending clarification / permission) must not be masked @@ -296,13 +297,12 @@ export function renderTaskStatusIcon( if (showRunningSpinner && !needsMe && task.foregroundActivity !== "background") { return ; } - return getTaskStateIcon( - task.state, - "h-4 w-4", + return getTaskStateIcon(task.state, "h-4 w-4", { hasPendingClarification, - task.foregroundActivity, + foregroundActivity: task.foregroundActivity, hasPendingPermission, - ); + interrupted: showInterrupted, + }); } // The board's only window into a fan-out. `activeSubagentCount` is derived from diff --git a/apps/web/components/kanban-card.tsx b/apps/web/components/kanban-card.tsx index 8208bef610..7ddb8135b0 100644 --- a/apps/web/components/kanban-card.tsx +++ b/apps/web/components/kanban-card.tsx @@ -63,6 +63,8 @@ export interface Task { * affordance on the card status icon. */ foregroundActivity?: ForegroundActivity | null; + /** True when the task's session was mid-turn when the backend died. */ + interrupted?: boolean; /** Live subagents summed across this task's sessions; drives the count chip. */ activeSubagentCount?: number; reviewStatus?: "pending" | "approved" | "changes_requested" | "rejected" | null; diff --git a/apps/web/components/kanban/graph2-step-node.tsx b/apps/web/components/kanban/graph2-step-node.tsx index 16a0bd068f..cbb64499c7 100644 --- a/apps/web/components/kanban/graph2-step-node.tsx +++ b/apps/web/components/kanban/graph2-step-node.tsx @@ -146,13 +146,12 @@ export function Graph2StepNode({ >
- {getTaskStateIcon( - task.state, - "h-3 w-3", - pendingInput.clarification, - task.foregroundActivity, - pendingInput.permission, - )} + {getTaskStateIcon(task.state, "h-3 w-3", { + hasPendingClarification: pendingInput.clarification, + foregroundActivity: task.foregroundActivity, + hasPendingPermission: pendingInput.permission, + interrupted: task.interrupted, + })}
{step.title}
diff --git a/apps/web/components/kanban/swimlane-graph-content.tsx b/apps/web/components/kanban/swimlane-graph-content.tsx index e8fae3dff2..407f122645 100644 --- a/apps/web/components/kanban/swimlane-graph-content.tsx +++ b/apps/web/components/kanban/swimlane-graph-content.tsx @@ -81,13 +81,12 @@ function DraggableTaskChip({ primarySessionState: task.primarySessionState, primarySessionPendingAction: task.primarySessionPendingAction, }); - const statusIcon = getTaskStateIcon( - task.state, - "h-3 w-3", - pendingInput.clarification, - task.foregroundActivity, - pendingInput.permission, - ); + const statusIcon = getTaskStateIcon(task.state, "h-3 w-3", { + hasPendingClarification: pendingInput.clarification, + foregroundActivity: task.foregroundActivity, + hasPendingPermission: pendingInput.permission, + interrupted: task.interrupted, + }); return (