macodev00 · macodev00 · Sep 25, 2026
diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
@@ -77,6 +77,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getUserInputActivity: () => Effect.die("unused"),
             listActivitiesByKind: () => Effect.die("unused"),
+            listUnterminatedTasks: () => Effect.die("unused"),
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
             getSnapshot: () =>
@@ -194,6 +195,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getUserInputActivity: () => Effect.die("unused"),
             listActivitiesByKind: () => Effect.die("unused"),
+            listUnterminatedTasks: () => Effect.die("unused"),
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
             getSnapshot: () =>
@@ -286,6 +288,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getUserInputActivity: () => Effect.die("unused"),
             listActivitiesByKind: () => Effect.die("unused"),
+            listUnterminatedTasks: () => Effect.die("unused"),
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
             getSnapshot: () =>
@@ -363,6 +366,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getUserInputActivity: () => Effect.die("unused"),
             listActivitiesByKind: () => Effect.die("unused"),
+            listUnterminatedTasks: () => Effect.die("unused"),
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
             getSnapshot: () =>
@@ -425,6 +429,7 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getUserInputActivity: () => Effect.die("unused"),
             listActivitiesByKind: () => Effect.die("unused"),
+            listUnterminatedTasks: () => Effect.die("unused"),
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
             getSnapshot: () =>

diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
@@ -421,6 +421,7 @@ describe("OrchestrationEngine", () => {
         Layer.succeed(ProjectionSnapshotQuery, {
           getUserInputActivity: () => Effect.die("unused"),
           listActivitiesByKind: () => Effect.die("unused"),
+          listUnterminatedTasks: () => Effect.die("unused"),
           getCommandReadModel: () => Effect.succeed(commandReadModel),
           getSnapshot: () =>
             Effect.sync(() => {

diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts
@@ -3549,4 +3549,155 @@ projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {
       assert.deepEqual(yield* query.listActivitiesByKind("nope"), []);
     }),
   );
+
+  it.effect("lists background tasks that never reached a terminal status", () =>
+    Effect.gen(function* () {
+      const query = yield* ProjectionSnapshotQuery;
+      const sql = yield* SqlClient.SqlClient;
+      const timestamp = "2026-03-02T00:00:00.000Z";
+      yield* sql`
+        INSERT INTO projection_projects (
+          project_id, title, workspace_root, scripts_json, created_at, updated_at
+        ) VALUES ('project-tasks', 'Project', '/tmp/project-tasks', '[]', ${timestamp}, ${timestamp})
+      `;
+      yield* sql`
+        INSERT INTO projection_threads (
+          thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
+          created_at, updated_at
+        ) VALUES
+          ('thread-open', 'project-tasks', 'Open', '{"instanceId":"claudeAgent","model":"claude"}',
+            'full-access', 'default', ${timestamp}, ${timestamp}),
+          ('thread-done', 'project-tasks', 'Done', '{"instanceId":"claudeAgent","model":"claude"}',
+            'full-access', 'default', ${timestamp}, ${timestamp})
+      `;
+      yield* sql`
+        INSERT INTO projection_thread_activities (
+          activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at, sequence
+        ) VALUES
+          ('open-started', 'thread-open', NULL, 'info', 'task.started', 'Task started',
+            '{"taskId":"task-explore","title":"Explore the repo","taskType":"local_agent"}',
+            ${timestamp}, 1),
+          ('open-progress', 'thread-open', NULL, 'info', 'task.progress', 'Still looking',
+            '{"taskId":"task-explore","summary":"reading"}', ${timestamp}, 2),
+          ('done-started', 'thread-done', NULL, 'info', 'task.started', 'Task started',
+            '{"taskId":"task-done","title":"Finished"}', ${timestamp}, 1),
+          ('done-completed', 'thread-done', NULL, 'info', 'task.completed', 'Task completed',
+            '{"taskId":"task-done","status":"completed"}', ${timestamp}, 2),
+          ('idle-updated', 'thread-open', NULL, 'info', 'task.updated', 'Task idle',
+            '{"taskId":"task-idle","status":"idle","title":"Resting"}', ${timestamp}, 1),
+          ('plan-started', 'thread-open', NULL, 'info', 'task.started', 'Plan',
+            '{"taskId":"task-plan","taskType":"plan","title":"Plan"}', ${timestamp}, 1)
+      `;
+
+      assert.deepEqual(
+        (yield* query.listUnterminatedTasks()).filter(
+          (task) => task.threadId === ThreadId.make("thread-open"),
+        ),
+        [
+          {
+            threadId: ThreadId.make("thread-open"),
+            taskId: "task-explore",
+            label: "Explore the repo",
+          },
+        ],
+      );
+    }),
+  );
+
+  it.effect("skips a malformed activity payload without dropping other open tasks", () =>
+    Effect.gen(function* () {
+      const query = yield* ProjectionSnapshotQuery;
+      const sql = yield* SqlClient.SqlClient;
+      const timestamp = "2026-03-02T00:00:00.000Z";
+      yield* sql`
+        INSERT INTO projection_projects (
+          project_id, title, workspace_root, scripts_json, created_at, updated_at
+        ) VALUES ('project-malformed', 'Project', '/tmp/project-malformed', '[]', ${timestamp}, ${timestamp})
+      `;
+      yield* sql`
+        INSERT INTO projection_threads (
+          thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
+          created_at, updated_at
+        ) VALUES (
+          'thread-malformed', 'project-malformed', 'Open', '{"instanceId":"claudeAgent","model":"claude"}',
+          'full-access', 'default', ${timestamp}, ${timestamp}
+        )
+      `;
+      yield* sql`
+        INSERT INTO projection_thread_activities (
+          activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at, sequence
+        ) VALUES
+          ('bad-payload', 'thread-malformed', NULL, 'info', 'task.started', 'Broken',
+            '{not-json', ${timestamp}, 1),
+          ('good-started', 'thread-malformed', NULL, 'info', 'task.started', 'Task started',
+            '{"taskId":"task-good","title":"Keep going","taskType":"local_bash"}',
+            ${timestamp}, 2)
+      `;
+
+      assert.deepEqual(
+        (yield* query.listUnterminatedTasks()).filter(
+          (task) => task.threadId === ThreadId.make("thread-malformed"),
+        ),
+        [
+          {
+            threadId: ThreadId.make("thread-malformed"),
+            taskId: "task-good",
+            label: "Keep going",
+          },
+        ],
+      );
+    }),
+  );
+
+  it.effect("keeps plan and dream tasks excluded when a later status row omits taskType", () =>
+    Effect.gen(function* () {
+      const query = yield* ProjectionSnapshotQuery;
+      const sql = yield* SqlClient.SqlClient;
+      const timestamp = "2026-03-02T00:00:00.000Z";
+      yield* sql`
+        INSERT INTO projection_projects (
+          project_id, title, workspace_root, scripts_json, created_at, updated_at
+        ) VALUES ('project-plan', 'Project', '/tmp/project-plan', '[]', ${timestamp}, ${timestamp})
+      `;
+      yield* sql`
+        INSERT INTO projection_threads (
+          thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
+          created_at, updated_at
+        ) VALUES (
+          'thread-plan', 'project-plan', 'Open', '{"instanceId":"claudeAgent","model":"claude"}',
+          'full-access', 'default', ${timestamp}, ${timestamp}
+        )
+      `;
+      yield* sql`
+        INSERT INTO projection_thread_activities (
+          activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at, sequence
+        ) VALUES
+          ('plan-later-started', 'thread-plan', NULL, 'info', 'task.started', 'Plan',
+            '{"taskId":"task-plan-later","taskType":"plan","title":"Draft the plan"}',
+            ${timestamp}, 1),
+          ('plan-later-updated', 'thread-plan', NULL, 'info', 'task.updated', 'Still planning',
+            '{"taskId":"task-plan-later","status":"running"}', ${timestamp}, 2),
+          ('dream-later-started', 'thread-plan', NULL, 'info', 'task.started', 'Dream',
+            '{"taskId":"task-dream-later","taskType":"dream","title":"Dream"}', ${timestamp}, 1),
+          ('dream-later-updated', 'thread-plan', NULL, 'info', 'task.updated', 'Still dreaming',
+            '{"taskId":"task-dream-later","status":"running"}', ${timestamp}, 2),
+          ('agent-later-started', 'thread-plan', NULL, 'info', 'task.started', 'Agent',
+            '{"taskId":"task-agent-later","taskType":"local_agent","title":"Review the diff"}',
+            ${timestamp}, 1)
+      `;
+
+      assert.deepEqual(
+        (yield* query.listUnterminatedTasks()).filter(
+          (task) => task.threadId === ThreadId.make("thread-plan"),
+        ),
+        [
+          {
+            threadId: ThreadId.make("thread-plan"),
+            taskId: "task-agent-later",
+            label: "Review the diff",
+          },
+        ],
+      );
+    }),
+  );
 });
diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
@@ -1527,6 +1527,117 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
       ),
     );
 
+  // Latest decisive task row per task. Status-free progress is not decisive:
+  // a late heartbeat must not resurrect a task that already finished. `stopped`
+  // is included because task.completed uses it for an interrupted task.
+  const listUnterminatedTaskRows = SqlSchema.findAll({
+    Request: Schema.Struct({}),
+    Result: Schema.Struct({
+      threadId: ThreadId,
+      taskId: Schema.String,
+      label: Schema.String,
+    }),
+    execute: () => sql`
+      WITH decisive AS (
+        SELECT
+          a.thread_id AS threadId,
+          trim(json_extract(a.payload_json, '$.taskId')) AS taskId,
+          a.kind AS kind,
+          json_extract(a.payload_json, '$.status') AS status,
+          json_extract(a.payload_json, '$.taskType') AS taskType,
+          CASE
+            WHEN json_type(a.payload_json, '$.title') = 'text'
+              AND length(trim(json_extract(a.payload_json, '$.title'))) > 0
+              THEN trim(json_extract(a.payload_json, '$.title'))
+            WHEN a.kind = 'task.started'
+              AND json_type(a.payload_json, '$.detail') = 'text'
+              AND length(trim(json_extract(a.payload_json, '$.detail'))) > 0
+              THEN trim(json_extract(a.payload_json, '$.detail'))
+            ELSE NULL
+          END AS label,
+          COALESCE(a.sequence, -1) AS sequence,
+          a.created_at AS createdAt,
+          a.activity_id AS activityId
+        FROM projection_thread_activities a
+        JOIN projection_threads t ON t.thread_id = a.thread_id
+        WHERE t.deleted_at IS NULL
+          AND t.archived_at IS NULL
+          AND json_valid(a.payload_json)
+          AND json_type(a.payload_json, '$.taskId') = 'text'
+          AND length(trim(json_extract(a.payload_json, '$.taskId'))) > 0
+          AND (
+            a.kind = 'task.completed'
+            OR a.kind = 'task.started'
+            OR (
+              a.kind IN ('task.progress', 'task.updated')
+              AND json_type(a.payload_json, '$.status') = 'text'
+            )
+          )
+      ),
+      latest AS (
+        SELECT
+          threadId,
+          taskId,
+          kind,
+          status,
+          taskType,
+          label,
+          ROW_NUMBER() OVER (
+            PARTITION BY threadId, taskId
+            ORDER BY sequence DESC, createdAt DESC, activityId DESC
+          ) AS rn
+        FROM decisive
+      ),
+      open_tasks AS (
+        SELECT threadId, taskId, label
+        FROM latest
+        WHERE rn = 1
+          AND kind != 'task.completed'
+          AND COALESCE(status, '') NOT IN (
+            'completed', 'failed', 'stopped', 'cancelled', 'interrupted', 'idle'
+          )
+          AND NOT EXISTS (
+            SELECT 1 FROM decisive d2
+            WHERE d2.threadId = latest.threadId
+              AND d2.taskId = latest.taskId
+              AND d2.taskType IN ('plan', 'dream')
+          )
+      ),
+      labels AS (
+        SELECT
+          d.threadId AS threadId,
+          d.taskId AS taskId,
+          d.label AS label,
+          ROW_NUMBER() OVER (
+            PARTITION BY d.threadId, d.taskId
+            ORDER BY d.sequence DESC, d.createdAt DESC, d.activityId DESC
+          ) AS labelRn
+        FROM decisive d
+        INNER JOIN open_tasks o
+          ON o.threadId = d.threadId AND o.taskId = d.taskId
+        WHERE d.label IS NOT NULL
+      )
+      SELECT
+        o.threadId AS "threadId",
+        o.taskId AS "taskId",
+        COALESCE(l.label, o.label, o.taskId) AS "label"
+      FROM open_tasks o
+      LEFT JOIN labels l
+        ON l.threadId = o.threadId AND l.taskId = o.taskId AND l.labelRn = 1
+    `,
+  });
+
+  /** Open background tasks whose latest decisive row is still non-terminal. */
+  const listUnterminatedTasks: ProjectionSnapshotQueryShape["listUnterminatedTasks"] = () =>
+    listUnterminatedTaskRows({}).pipe(
+      Effect.mapError(
+        toPersistenceSqlOrDecodeError(
+          "ProjectionSnapshotQuery.listUnterminatedTasks:query",
+          "ProjectionSnapshotQuery.listUnterminatedTasks:decodeRow",
+        ),
+      ),
+    );
+
   const listThreadActivityIdsByThread = SqlSchema.findAll({
     Request: ThreadIdLookupInput,
     Result: ProjectionThreadActivityIdRowSchema,
@@ -3776,6 +3887,7 @@ pending_approval_requests AS (
     getCommandReadModel,
     getUserInputActivity,
     listActivitiesByKind,
+    listUnterminatedTasks,
     getSnapshot,
     getShellSnapshot,
     getArchivedShellSnapshot,

diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
@@ -38,6 +38,13 @@ export interface ProjectionSnapshotCounts {
   readonly threadCount: number;
 }
 
+/** One background task still running when its provider process died. */
+export interface UnterminatedProviderTask {
+  readonly threadId: ThreadId;
+  readonly taskId: string;
+  readonly label: string;
+}
+
 export interface ProjectionSnapshotSequence {
   readonly snapshotSequence: number;
 }
@@ -92,6 +99,16 @@ export interface ProjectionSnapshotQueryShape {
     kind: string,
   ) => Effect.Effect<ReadonlyArray<OrchestrationThreadActivity>, ProjectionRepositoryError>;
 
+  /**
+   * Background work that outlived its turn: the latest decisive task row is
+   * still non-terminal. Idle tasks and plan/dream bookkeeping are omitted.
+   * Used at startup, where in-memory `liveTaskIds` are already gone.
+   */
+  readonly listUnterminatedTasks: () => Effect.Effect<
+    ReadonlyArray<UnterminatedProviderTask>,
+    ProjectionRepositoryError
+  >;
+
   /**
    * Read the lightweight command snapshot used to bootstrap the in-memory
    * orchestration engine without hydrating message/activity/checkpoint bodies.

diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts
@@ -39,6 +39,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray<
     getCommandReadModel: () => Effect.die("unused"),
     getUserInputActivity: () => Effect.die("unused"),
     listActivitiesByKind: () => Effect.die("unused"),
+    listUnterminatedTasks: () => Effect.die("unused"),
     getSnapshot: () => Effect.die("unused"),
     getShellSnapshot: () =>
       Effect.succeed({

diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts
@@ -30,6 +30,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
   Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
     getUserInputActivity: () => Effect.die("unused"),
     listActivitiesByKind: () => Effect.die("unused"),
+    listUnterminatedTasks: () => Effect.die("unused"),
     getCommandReadModel: () => Effect.die("unused"),
     getSnapshot: () => Effect.die("unused"),
     getShellSnapshot: () => Effect.die("unused"),

diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts
@@ -4965,6 +4965,7 @@ describe("agent browser access", () => {
         getImportedAgentSessionSources: () => Effect.die("unused"),
         getUserInputActivity: () => Effect.die("unused"),
         listActivitiesByKind: () => Effect.die("unused"),
+        listUnterminatedTasks: () => Effect.die("unused"),
         getCommandReadModel: () => Effect.die("unused"),
         getSnapshot: () => Effect.die("unused"),
         getShellSnapshot: () => Effect.die("unused"),

diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
@@ -235,6 +235,7 @@ describe("ProviderSessionReaper", () => {
         Layer.succeed(ProjectionSnapshotQuery, {
           getUserInputActivity: () => Effect.die("unused"),
           listActivitiesByKind: () => Effect.die("unused"),
+          listUnterminatedTasks: () => Effect.die("unused"),
           getCommandReadModel: () => Effect.die("unused"),
           getSnapshot: () => Effect.die("unused"),
           getShellSnapshot: () => Effect.die("unused"),