feat: show a waiting state when background tasks outlive the turn

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -1037,6 +1037,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         runtimeMode: event.payload.session.runtimeMode,
         activeTurnId: event.payload.session.activeTurnId,
         lastError: event.payload.session.lastError,
+        pendingBackgroundTasks: event.payload.session.pendingBackgroundTasks ?? null,
         updatedAt: event.payload.session.updatedAt,
       });
     });

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -5,6 +5,7 @@ import {
   MessageId,
   NonNegativeInt,
   OrchestrationCheckpointFile,
+  OrchestrationPendingBackgroundTask,
   OrchestrationProposedPlanId,
   OrchestrationReadModel,
   OrchestrationShellSnapshot,
@@ -86,7 +87,13 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
     sequence: Schema.NullOr(NonNegativeInt),
   }),
 );
-const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
+const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession.mapFields(
+  Struct.assign({
+    pendingBackgroundTasks: Schema.NullOr(
+      Schema.fromJsonString(Schema.Array(OrchestrationPendingBackgroundTask)),
+    ),
+  }),
+);
 const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(
   Struct.assign({
     files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
@@ -220,6 +227,9 @@ function mapSessionRow(
     runtimeMode: row.runtimeMode,
     activeTurnId: row.activeTurnId,
     lastError: row.lastError,
+    ...(row.pendingBackgroundTasks !== null && row.pendingBackgroundTasks.length > 0
+      ? { pendingBackgroundTasks: row.pendingBackgroundTasks }
+      : {}),
     updatedAt: row.updatedAt,
   };
 }
@@ -488,6 +498,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           runtime_mode AS "runtimeMode",
           active_turn_id AS "activeTurnId",
           last_error AS "lastError",
+          pending_background_tasks_json AS "pendingBackgroundTasks",
           updated_at AS "updatedAt"
         FROM projection_thread_sessions
         ORDER BY thread_id ASC
@@ -509,6 +520,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           sessions.runtime_mode AS "runtimeMode",
           sessions.active_turn_id AS "activeTurnId",
           sessions.last_error AS "lastError",
+          sessions.pending_background_tasks_json AS "pendingBackgroundTasks",
           sessions.updated_at AS "updatedAt"
         FROM projection_thread_sessions sessions
         INNER JOIN projection_threads threads
@@ -534,6 +546,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           sessions.runtime_mode AS "runtimeMode",
           sessions.active_turn_id AS "activeTurnId",
           sessions.last_error AS "lastError",
+          sessions.pending_background_tasks_json AS "pendingBackgroundTasks",
           sessions.updated_at AS "updatedAt"
         FROM projection_thread_sessions sessions
         INNER JOIN projection_threads threads
@@ -853,6 +866,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
           runtime_mode AS "runtimeMode",
           active_turn_id AS "activeTurnId",
           last_error AS "lastError",
+          pending_background_tasks_json AS "pendingBackgroundTasks",
           updated_at AS "updatedAt"
         FROM projection_thread_sessions
         WHERE thread_id = ${threadId}
@@ -1150,18 +1164,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
 
               for (const row of sessionRows) {
                 updatedAt = maxIso(updatedAt, row.updatedAt);
-                sessionsByThread.set(row.threadId, {
-                  threadId: row.threadId,
-                  status: row.status,
-                  providerName: row.providerName,
-                  ...(row.providerInstanceId !== null
-                    ? { providerInstanceId: row.providerInstanceId }
-                    : {}),
-                  runtimeMode: row.runtimeMode,
-                  activeTurnId: row.activeTurnId,
-                  lastError: row.lastError,
-                  updatedAt: row.updatedAt,
-                });
+                sessionsByThread.set(row.threadId, mapSessionRow(row));
               }
 
               const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects(

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts

@@ -449,6 +449,146 @@ describe("ProviderRuntimeIngestion", () => {
     expect(thread.session?.lastError).toBeNull();
   });
 
+  it("surfaces background-task rosters as pendingBackgroundTasks on a ready session", async () => {
+    const harness = await createHarness();
+    const now = "2026-01-01T00:00:00.000Z";
+
+    // Waiting-on-background-tasks stays "ready" on the wire (additive field,
+    // no new status literal) with the roster attached.
+    harness.emit({
+      type: "session.state.changed",
+      eventId: asEventId("evt-bg-waiting"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      payload: {
+        state: "waiting",
+        reason: "background-tasks",
+        backgroundTasks: [
+          { taskId: "bg-1", description: "Run Codex review", taskType: "local_bash" },
+        ],
+      },
+    });
+
+    let thread = await waitForThread(
+      harness.readModel,
+      (entry) =>
+        entry.session?.status === "ready" &&
+        (entry.session?.pendingBackgroundTasks?.length ?? 0) === 1,
+    );
+    expect(thread.session?.pendingBackgroundTasks).toEqual([
+      { taskId: "bg-1", description: "Run Codex review", taskType: "local_bash" },
+    ]);
+    expect(thread.session?.activeTurnId).toBeNull();
+
+    // An empty roster (tasks drained) clears the pending list.
+    harness.emit({
+      type: "session.state.changed",
+      eventId: asEventId("evt-bg-drained"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      payload: {
+        state: "ready",
+        reason: "background-tasks-drained",
+        backgroundTasks: [],
+      },
+    });
+
+    thread = await waitForThread(
+      harness.readModel,
+      (entry) =>
+        entry.session?.status === "ready" &&
+        (entry.session?.pendingBackgroundTasks?.length ?? 0) === 0,
+    );
+    expect(thread.session?.pendingBackgroundTasks).toBeUndefined();
+
+    // A new turn starting also clears any stale pending roster.
+    harness.emit({
+      type: "session.state.changed",
+      eventId: asEventId("evt-bg-waiting-2"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      payload: {
+        state: "waiting",
+        reason: "background-tasks",
+        backgroundTasks: [{ taskId: "bg-2" }],
+      },
+    });
+    await waitForThread(
+      harness.readModel,
+      (entry) => (entry.session?.pendingBackgroundTasks?.length ?? 0) === 1,
+    );
+    harness.emit({
+      type: "turn.started",
+      eventId: asEventId("evt-bg-turn-started"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      turnId: asTurnId("turn-bg"),
+    });
+
+    thread = await waitForThread(
+      harness.readModel,
+      (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === "turn-bg",
+    );
+    expect(thread.session?.pendingBackgroundTasks).toBeUndefined();
+  });
+
+  it("ignores stale background-task roster events while a turn is active", async () => {
+    const harness = await createHarness();
+    const now = "2026-01-01T00:00:00.000Z";
+
+    harness.emit({
+      type: "turn.started",
+      eventId: asEventId("evt-bg-race-turn-started"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      turnId: asTurnId("turn-race"),
+    });
+    await waitForThread(
+      harness.readModel,
+      (entry) => entry.session?.status === "running" && entry.session?.activeTurnId === "turn-race",
+    );
+
+    // A delayed/replayed roster event must not settle the running turn.
+    harness.emit({
+      type: "session.state.changed",
+      eventId: asEventId("evt-bg-race-roster"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      payload: {
+        state: "waiting",
+        reason: "background-tasks",
+        backgroundTasks: [{ taskId: "bg-stale" }],
+      },
+    });
+    // A non-roster state change is still applied; use it as the sync point to
+    // know the roster event has been processed (events are handled in order).
+    harness.emit({
+      type: "session.state.changed",
+      eventId: asEventId("evt-bg-race-running"),
+      provider: ProviderDriverKind.make("codex"),
+      threadId: asThreadId("thread-1"),
+      createdAt: now,
+      payload: {
+        state: "running",
+        reason: "still-working",
+      },
+    });
+
+    const thread = await waitForThread(
+      harness.readModel,
+      (entry) => entry.session?.lastError === null && entry.session?.status === "running",
+    );
+    expect(thread.session?.status).toBe("running");
+    expect(thread.session?.activeTurnId).toBe("turn-race");
+    expect(thread.session?.pendingBackgroundTasks).toBeUndefined();
+  });
+
   it("clears active turn when provider session becomes ready", async () => {
     const harness = await createHarness();
     const now = "2026-01-01T00:00:00.000Z";

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts

@@ -1340,6 +1340,13 @@ const make = Effect.gen(function* () {
           case "session.started":
           case "thread.started":
             return true;
+          case "session.state.changed":
+            // Background-task roster events are only meaningful between
+            // turns: they map to a "ready" status, so applying a stale or
+            // replayed one mid-turn would null the active turn and settle a
+            // genuinely running turn. The turn lifecycle already clears
+            // pendingBackgroundTasks when a turn starts.
+            return event.payload.backgroundTasks === undefined || activeTurnId === null;
           case "turn.started":
             return !conflictsWithActiveTurn || conflictingTurnStartIsPendingTurnStart;
           case "turn.completed":
@@ -1369,10 +1376,20 @@ const make = Effect.gen(function* () {
         event.type === "turn.started" ||
         event.type === "turn.completed"
       ) {
+        // A session.state.changed carrying a backgroundTasks roster is the
+        // between-turns "waiting on background work" signal (e.g. a Claude
+        // background shell that outlived its turn). It keeps the wire status
+        // "ready" — old clients degrade to today's idle look — and surfaces
+        // the roster via session.pendingBackgroundTasks for new clients.
+        const backgroundTaskRoster =
+          event.type === "session.state.changed" ? event.payload.backgroundTasks : undefined;
         const status = (() => {
           switch (event.type) {
             case "session.state.changed": {
-              const runtimeStatus = orchestrationSessionStatusFromRuntimeState(event.payload.state);
+              const runtimeStatus =
+                backgroundTaskRoster !== undefined
+                  ? "ready"
+                  : orchestrationSessionStatusFromRuntimeState(event.payload.state);
               return hasPendingTurnStart && runtimeStatus === "ready" ? "starting" : runtimeStatus;
             }
             case "turn.started":
@@ -1432,6 +1449,13 @@ const make = Effect.gen(function* () {
             );
           }
 
+          // Roster events replace pendingBackgroundTasks wholesale (empty
+          // roster clears it); every other lifecycle event clears it — a
+          // turn starting or the session exiting supersedes the wait.
+          const pendingBackgroundTasks =
+            backgroundTaskRoster !== undefined && backgroundTaskRoster.length > 0
+              ? backgroundTaskRoster
+              : undefined;
           yield* orchestrationEngine.dispatch({
             type: "thread.session.set",
             commandId: yield* providerCommandId(event, "thread-session-set"),
@@ -1446,6 +1470,7 @@ const make = Effect.gen(function* () {
               runtimeMode: thread.session?.runtimeMode ?? "full-access",
               activeTurnId: nextActiveTurnId,
               lastError,
+              ...(pendingBackgroundTasks !== undefined ? { pendingBackgroundTasks } : {}),
               updatedAt: now,
             },
             createdAt: now,

apps/server/src/orchestration/decider.ts

@@ -399,6 +399,17 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
           }),
         );
       }
+      // Waiting on provider background tasks (a background shell/subagent
+      // that outlived its turn) is live work: the provider wakes the session
+      // when the task finishes, so settling now would hide it.
+      if ((thread.session?.pendingBackgroundTasks?.length ?? 0) > 0) {
+        return yield* Effect.fail(
+          new OrchestrationCommandInvariantError({
+            commandType: command.type,
+            detail: `thread ${command.threadId} is waiting on background tasks and cannot be settled`,
+          }),
+        );
+      }
       // Pending approval / user-input requests are blocked-on-you work: a
       // raced or stale client must not park them behind a settled override
       // that would surface only after the request resolves.

apps/server/src/persistence/Layers/ProjectionThreadSessions.ts

@@ -2,6 +2,9 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
 import * as SqlSchema from "effect/unstable/sql/SqlSchema";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import * as Struct from "effect/Struct";
+import { OrchestrationPendingBackgroundTask } from "@t3tools/contracts";
 
 import { toPersistenceSqlError } from "../Errors.ts";
 
@@ -13,11 +16,19 @@ import {
   GetProjectionThreadSessionInput,
 } from "../Services/ProjectionThreadSessions.ts";
 
+const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession.mapFields(
+  Struct.assign({
+    pendingBackgroundTasks: Schema.NullOr(
+      Schema.fromJsonString(Schema.Array(OrchestrationPendingBackgroundTask)),
+    ),
+  }),
+);
+
 const makeProjectionThreadSessionRepository = Effect.gen(function* () {
   const sql = yield* SqlClient.SqlClient;
 
   const upsertProjectionThreadSessionRow = SqlSchema.void({
-    Request: ProjectionThreadSession,
+    Request: ProjectionThreadSessionDbRowSchema,
     execute: (row) =>
       sql`
         INSERT INTO projection_thread_sessions (
@@ -28,6 +39,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () {
           runtime_mode,
           active_turn_id,
           last_error,
+          pending_background_tasks_json,
           updated_at
         )
         VALUES (
@@ -38,6 +50,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () {
           ${row.runtimeMode},
           ${row.activeTurnId},
           ${row.lastError},
+          ${row.pendingBackgroundTasks},
           ${row.updatedAt}
         )
         ON CONFLICT (thread_id)
@@ -48,13 +61,14 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () {
           runtime_mode = excluded.runtime_mode,
           active_turn_id = excluded.active_turn_id,
           last_error = excluded.last_error,
+          pending_background_tasks_json = excluded.pending_background_tasks_json,
           updated_at = excluded.updated_at
       `,
   });
 
   const getProjectionThreadSessionRow = SqlSchema.findOneOption({
     Request: GetProjectionThreadSessionInput,
-    Result: ProjectionThreadSession,
+    Result: ProjectionThreadSessionDbRowSchema,
     execute: ({ threadId }) =>
       sql`
         SELECT
@@ -65,6 +79,7 @@ const makeProjectionThreadSessionRepository = Effect.gen(function* () {
           runtime_mode AS "runtimeMode",
           active_turn_id AS "activeTurnId",
           last_error AS "lastError",
+          pending_background_tasks_json AS "pendingBackgroundTasks",
           updated_at AS "updatedAt"
         FROM projection_thread_sessions
         WHERE thread_id = ${threadId}