diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index cfb88a06cd2..98209f6fe30 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/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, }); }); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 155e9ab0013..f87e4e5ae52 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/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( diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd31..ff900cac5b3 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/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"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index a8a51b30260..03bee919967 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/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, diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index cba967afc7c..b81344d778d 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/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. diff --git a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts b/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts index dcb750983a0..630c97b037a 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadSessions.ts +++ b/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} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index cacb2c85b83..98f009545ad 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -46,6 +46,7 @@ import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes. import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts"; +import Migration0034 from "./Migrations/034_ProjectionThreadSessionPendingBackgroundTasks.ts"; /** * Migration loader with all migrations defined inline. @@ -91,6 +92,7 @@ export const migrationEntries = [ [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], [33, "ProjectionThreadsSettled", Migration0033], + [34, "ProjectionThreadSessionPendingBackgroundTasks", Migration0034], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/034_ProjectionThreadSessionPendingBackgroundTasks.ts b/apps/server/src/persistence/Migrations/034_ProjectionThreadSessionPendingBackgroundTasks.ts new file mode 100644 index 00000000000..96f2ede191b --- /dev/null +++ b/apps/server/src/persistence/Migrations/034_ProjectionThreadSessionPendingBackgroundTasks.ts @@ -0,0 +1,11 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + ALTER TABLE projection_thread_sessions + ADD COLUMN pending_background_tasks_json TEXT + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts index 7cecac33eb6..0ef2b5d122d 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadSessions.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadSessions.ts @@ -9,6 +9,7 @@ import { RuntimeMode, IsoDateTime, + OrchestrationPendingBackgroundTask, OrchestrationSessionStatus, ProviderInstanceId, ThreadId, @@ -29,6 +30,7 @@ export const ProjectionThreadSession = Schema.Struct({ runtimeMode: RuntimeMode, activeTurnId: Schema.NullOr(TurnId), lastError: Schema.NullOr(Schema.String), + pendingBackgroundTasks: Schema.NullOr(Schema.Array(OrchestrationPendingBackgroundTask)), updatedAt: IsoDateTime, }); export type ProjectionThreadSession = typeof ProjectionThreadSession.Type; diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 2735b69a187..2a41256c1f0 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1838,6 +1838,201 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("surfaces pending background tasks as a waiting state after turn completion", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => runtimeEvents.push(event)), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "review the plan", + attachments: [], + }); + + // Background shell launched mid-turn: roster snapshot + task_started. + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "bg-codex", + tool_use_id: "tool-bg", + description: "Run Codex review of the plan", + task_type: "local_bash", + session_id: "sdk-session-bg", + uuid: "bg-task-started", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "background_tasks_changed", + tasks: [ + { + task_id: "bg-codex", + task_type: "local_bash", + description: "Run Codex review of the plan", + }, + ], + session_id: "sdk-session-bg", + uuid: "bg-roster-1", + } as unknown as SDKMessage); + + // Turn ends while the background task is still running. + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-bg", + uuid: "bg-result-1", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + const turnCompletedIndex = runtimeEvents.findIndex( + (event) => event.type === "turn.completed", + ); + assert.notEqual(turnCompletedIndex, -1); + const waitingEvent = runtimeEvents + .slice(turnCompletedIndex) + .find( + (event) => + event.type === "session.state.changed" && event.payload.reason === "background-tasks", + ); + assert.equal(waitingEvent?.type, "session.state.changed"); + if (waitingEvent?.type === "session.state.changed") { + assert.equal(waitingEvent.payload.state, "waiting"); + assert.deepEqual(waitingEvent.payload.backgroundTasks, [ + { + taskId: "bg-codex", + description: "Run Codex review of the plan", + taskType: "local_bash", + }, + ]); + } + + // The task finishing between turns drains the roster back to ready. + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "bg-codex", + tool_use_id: "tool-bg", + status: "completed", + output_file: "/tmp/tasks/bg-codex.output", + session_id: "sdk-session-bg", + uuid: "bg-task-done", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "background_tasks_changed", + tasks: [], + session_id: "sdk-session-bg", + uuid: "bg-roster-2", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + const drainedEvent = runtimeEvents.find( + (event) => + event.type === "session.state.changed" && + event.payload.reason === "background-tasks-drained", + ); + assert.equal(drainedEvent?.type, "session.state.changed"); + if (drainedEvent?.type === "session.state.changed") { + assert.equal(drainedEvent.payload.state, "ready"); + assert.deepEqual(drainedEvent.payload.backgroundTasks, []); + } + + runtimeEventsFiber.interruptUnsafe(); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("drops the background-task roster when a turn fails", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => runtimeEvents.push(event)), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "kick off a background task", + attachments: [], + }); + + harness.query.emit({ + type: "system", + subtype: "background_tasks_changed", + tasks: [{ task_id: "bg-doomed", task_type: "local_bash", description: "Long job" }], + session_id: "sdk-session-fail", + uuid: "fail-roster-1", + } as unknown as SDKMessage); + + // The turn fails; the roster is no longer trustworthy and a stale + // entry must not resurface as a waiting state after later activity. + harness.query.emit({ + type: "result", + subtype: "error_during_execution", + is_error: true, + errors: ["boom"], + session_id: "sdk-session-fail", + uuid: "fail-result-1", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + // A terminal notification for the doomed task between turns must + // report ready (empty roster), not a waiting state with stale tasks. + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "bg-doomed", + tool_use_id: "tool-doomed", + status: "stopped", + session_id: "sdk-session-fail", + uuid: "fail-task-done", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + const waitingEvents = runtimeEvents.filter( + (event) => + event.type === "session.state.changed" && event.payload.reason === "background-tasks", + ); + assert.deepEqual(waitingEvents, []); + const drained = runtimeEvents.find( + (event) => + event.type === "session.state.changed" && + event.payload.reason === "background-tasks-drained", + ); + assert.equal(drained?.type, "session.state.changed"); + if (drained?.type === "session.state.changed") { + assert.deepEqual(drained.payload.backgroundTasks, []); + } + + runtimeEventsFiber.interruptUnsafe(); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("emits thread token usage updates from Claude task progress", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 757d7a00eb2..c4691f01db7 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -178,6 +178,12 @@ interface ClaudeTaskState { readonly blockedBy: Set; } +interface ClaudeBackgroundTask { + readonly taskId: string; + readonly description?: string; + readonly taskType?: string; +} + interface ClaudeSessionContext { session: ProviderSession; readonly promptQueue: Queue.Queue; @@ -195,6 +201,14 @@ interface ClaudeSessionContext { }>; readonly inFlightTools: Map; readonly claudeTasks: Map; + /** + * Live background tasks (background shells, subagents) keyed by task_id. + * Fed by `background_tasks_changed` roster snapshots with `task_started` / + * `task_notification` as incremental fallbacks. A non-empty roster at turn + * completion means the SDK will wake the session when a task finishes, so + * the session is waiting rather than done. + */ + readonly backgroundTasks: Map; turnState: ClaudeTurnState | undefined; lastKnownContextWindow: number | undefined; lastKnownTokenUsage: ThreadTokenUsageSnapshot | undefined; @@ -1729,6 +1743,57 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); }); + /** + * Surface the background-task roster as a session waiting/ready signal. + * Only meaningful between turns: while a turn is running the session is + * already "running", and emitting would fight the turn lifecycle. Called + * at turn completion (roster non-empty → waiting) and on roster changes + * that arrive between turns (roster emptied → ready; the follow-up + * synthetic turn usually flips it to running immediately after). + */ + const emitBackgroundWaitState = Effect.fn("emitBackgroundWaitState")(function* ( + context: ClaudeSessionContext, + options: { + readonly rawMethod: string; + readonly rawPayload: unknown; + }, + ) { + if (context.turnState) { + return; + } + const tasks = Array.from(context.backgroundTasks.values()); + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "session.state.changed", + eventId: stamp.eventId, + provider: PROVIDER, + createdAt: stamp.createdAt, + threadId: context.session.threadId, + payload: + tasks.length > 0 + ? { + state: "waiting", + reason: "background-tasks", + backgroundTasks: tasks.map((task) => ({ + taskId: task.taskId, + ...(task.description ? { description: task.description } : {}), + ...(task.taskType ? { taskType: task.taskType } : {}), + })), + } + : { + state: "ready", + reason: "background-tasks-drained", + backgroundTasks: [], + }, + providerRefs: nativeProviderRefs(context), + raw: { + source: "claude.sdk.message", + method: options.rawMethod, + payload: options.rawPayload, + }, + }); + }); + const emitThreadTokenUsage = Effect.fn("emitThreadTokenUsage")(function* ( context: ClaudeSessionContext, usage: ThreadTokenUsageSnapshot | undefined, @@ -2064,6 +2129,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(status === "failed" && errorMessage ? { lastError: errorMessage } : {}), }; yield* updateResumeCursor(context); + + // The turn ended but background tasks (background shells, subagents) may + // still be running — the SDK will wake the session with a follow-up turn + // when they finish. Surface that gap as a waiting state so the thread + // doesn't look done while e.g. a codex review is still in flight. + if (status === "completed" && context.backgroundTasks.size > 0) { + yield* emitBackgroundWaitState(context, { + rawMethod: "claude/result", + rawPayload: result ?? { status }, + }); + } else if (status !== "completed") { + // A failed or interrupted turn leaves the roster untrustworthy (terminal + // task notifications or the authoritative empty snapshot may never + // arrive), and stale entries would resurface as a bogus waiting state + // after a later turn. Drop it; the next background_tasks_changed + // snapshot rebuilds it from scratch. + context.backgroundTasks.clear(); + } }); const handleStreamEvent = Effect.fn("handleStreamEvent")(function* ( @@ -2585,14 +2668,40 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }; - // Undeclared-but-real subtypes (absent from the SDK's union, so they can't - // be switch cases): consumed intentionally without emitting, otherwise - // they fall through to the unknown-subtype warning and surface as spurious - // error rows in client work logs. `background_tasks_changed` is a roster - // snapshot ({tasks: [...]}) — the task_* lifecycle events carry the - // authoritative per-agent data and the typed background_tasks control - // request is the reconciliation source. + // Undeclared-but-real subtype (absent from the SDK's union, so it can't + // be a switch case): `background_tasks_changed` is the authoritative + // roster snapshot ({tasks: [...]}) of live background tasks. It replaces + // the roster wholesale; between turns the roster drives the session + // waiting/ready signal so "turn ended but codex/subagent still running" + // is visible instead of looking idle. The task_* lifecycle events remain + // the per-agent data source for work-log rows. if ((message.subtype as string) === "background_tasks_changed") { + const roster = (message as unknown as { tasks?: unknown }).tasks; + if (Array.isArray(roster)) { + context.backgroundTasks.clear(); + for (const entry of roster) { + if (!entry || typeof entry !== "object") { + continue; + } + const task = entry as { task_id?: unknown; description?: unknown; task_type?: unknown }; + if (typeof task.task_id !== "string" || task.task_id.length === 0) { + continue; + } + context.backgroundTasks.set(task.task_id, { + taskId: task.task_id, + ...(typeof task.description === "string" && task.description.trim().length > 0 + ? { description: task.description } + : {}), + ...(typeof task.task_type === "string" && task.task_type.trim().length > 0 + ? { taskType: task.task_type } + : {}), + }); + } + yield* emitBackgroundWaitState(context, { + rawMethod: "claude/system/background_tasks_changed", + rawPayload: message, + }); + } return; } @@ -2677,6 +2786,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); return; case "task_started": + // Incremental fallback for the roster (authoritative snapshots come + // from background_tasks_changed, which is not in the SDK's typed + // union and could disappear in an SDK update). + context.backgroundTasks.set(message.task_id, { + taskId: message.task_id, + ...(message.description ? { description: message.description } : {}), + ...(message.task_type ? { taskType: message.task_type } : {}), + }); yield* offerRuntimeEvent({ ...base, type: "task.started", @@ -2714,6 +2831,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( case "task_updated": return; case "task_notification": + context.backgroundTasks.delete(message.task_id); yield* emitThreadTokenUsage( context, normalizeClaudeTaskProgressTokenUsage(message.usage, context), @@ -2732,6 +2850,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...(message.usage ? { usage: message.usage } : {}), }, }); + yield* emitBackgroundWaitState(context, { + rawMethod: "claude/system/task_notification", + rawPayload: message, + }); return; case "files_persisted": yield* offerRuntimeEvent({ @@ -3190,6 +3312,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const pendingUserInputs = new Map(); const inFlightTools = new Map(); const claudeTasks = new Map(); + const backgroundTasks = new Map(); const contextRef = yield* Ref.make(undefined); @@ -3631,6 +3754,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( turns: [], inFlightTools, claudeTasks, + backgroundTasks, turnState: undefined, lastKnownContextWindow: initialContextWindow, lastKnownTokenUsage: undefined, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ad303f607e7..be1f1d96e32 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -5253,6 +5253,7 @@ function ChatViewContent(props: ChatViewProps) { isWorking={isWorking} activeTurnInProgress={isWorking || !latestTurnSettled} activeTurnStartedAt={activeWorkStartedAt} + pendingBackgroundTasks={activeThread.session?.pendingBackgroundTasks ?? null} listRef={legendListRef} timelineEntries={timelineEntries} latestTurn={activeLatestTurn} diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index a9b32a887c0..eba244f0c56 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -678,6 +678,31 @@ describe("resolveSidebarV2Status", () => { ).toBe("working"); }); + it("reports working for a ready session waiting on background tasks", () => { + expect( + resolveSidebarV2Status({ + ...idle, + session: { + ...session, + status: "ready" as const, + activeTurnId: null, + pendingBackgroundTasks: [{ taskId: "bg-1", description: "Run Codex review" }], + }, + }), + ).toBe("working"); + expect( + resolveSidebarV2Status({ + ...idle, + session: { + ...session, + status: "ready" as const, + activeTurnId: null, + pendingBackgroundTasks: [], + }, + }), + ).toBe("ready"); + }); + it("reports failed only while the session status is error", () => { expect( resolveSidebarV2Status({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index b8ae19c3e5f..4cd8e41de94 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -441,6 +441,12 @@ export function resolveSidebarV2Status(thread: SidebarV2StatusInput): SidebarV2S if (thread.session?.status === "running" || thread.session?.status === "starting") { return "working"; } + // Ready on the wire but waiting on provider background tasks (e.g. a + // backgrounded codex review that outlived its turn): the provider wakes + // the session when the task finishes, so the thread is still in motion. + if ((thread.session?.pendingBackgroundTasks?.length ?? 0) > 0) { + return "working"; + } if (thread.session?.status === "error") { return "failed"; } diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 6d74204bc1c..fb3bec49152 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -744,6 +744,60 @@ describe("deriveMessagesTimelineRows", () => { expect(finalRow?.kind === "message" && finalRow.showAssistantMeta).toBe(true); }); + it("shows a waiting row when the turn settled with pending background tasks", () => { + const rows = deriveMessagesTimelineRows({ + timelineEntries: [ + { + id: "assistant-final-entry", + kind: "message", + createdAt: "2026-01-01T00:00:20Z", + message: { + id: "assistant-final" as never, + role: "assistant" as const, + text: "Codex is reviewing the plan; I'll report back when it completes.", + turnId: "turn-1" as never, + createdAt: "2026-01-01T00:00:20Z", + updatedAt: "2026-01-01T00:00:20Z", + streaming: false, + }, + }, + ], + latestTurn: { + turnId: "turn-1" as never, + state: "completed", + startedAt: "2026-01-01T00:00:00Z", + completedAt: "2026-01-01T00:00:20Z", + }, + isWorking: false, + activeTurnStartedAt: null, + pendingBackgroundTasks: [{ taskId: "bg-1", description: "Run Codex review of the plan" }], + turnDiffSummaryByAssistantMessageId: new Map(), + revertTurnCountByUserMessageId: new Map(), + }); + + expect(rows.at(-1)).toEqual({ + kind: "waiting-background", + id: "waiting-background-row", + createdAt: null, + description: "Run Codex review of the plan", + taskCount: 1, + }); + }); + + it("suppresses the waiting row while a turn is actively working", () => { + const rows = deriveMessagesTimelineRows({ + timelineEntries: [], + latestTurn: null, + isWorking: true, + activeTurnStartedAt: "2026-01-01T00:00:00Z", + pendingBackgroundTasks: [{ taskId: "bg-1" }], + turnDiffSummaryByAssistantMessageId: new Map(), + revertTurnCountByUserMessageId: new Map(), + }); + + expect(rows.map((row) => row.kind)).toEqual(["working"]); + }); + it("does not fold the active in-progress turn", () => { const rows = deriveMessagesTimelineRows({ timelineEntries: [ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 3227bac2413..3a6c447a359 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -175,7 +175,14 @@ export type MessagesTimelineRow = createdAt: string; proposedPlan: ProposedPlan; } - | { kind: "working"; id: string; createdAt: string | null }; + | { kind: "working"; id: string; createdAt: string | null } + | { + kind: "waiting-background"; + id: string; + createdAt: string | null; + description: string | null; + taskCount: number; + }; export interface StableMessagesTimelineRowsState { byId: Map; @@ -410,6 +417,10 @@ export function deriveMessagesTimelineRows(input: { expandedWorkGroupIds?: ReadonlySet; isWorking: boolean; activeTurnStartedAt: string | null; + pendingBackgroundTasks?: ReadonlyArray<{ + readonly taskId: string; + readonly description?: string | undefined; + }> | null; turnDiffSummaryByAssistantMessageId: ReadonlyMap; revertTurnCountByUserMessageId: ReadonlyMap; }): MessagesTimelineRow[] { @@ -569,6 +580,18 @@ export function deriveMessagesTimelineRows(input: { id: "working-indicator-row", createdAt: input.activeTurnStartedAt, }); + } else if (input.pendingBackgroundTasks && input.pendingBackgroundTasks.length > 0) { + // The turn ended but provider background tasks (background shells, + // subagents) are still running — the provider wakes the session when + // they finish, so show a waiting row instead of looking done. + const firstTask = input.pendingBackgroundTasks[0]; + nextRows.push({ + kind: "waiting-background", + id: "waiting-background-row", + createdAt: input.activeTurnStartedAt, + description: firstTask?.description ?? null, + taskCount: input.pendingBackgroundTasks.length, + }); } return nextRows; @@ -602,6 +625,15 @@ function isRowUnchanged(a: MessagesTimelineRow, b: MessagesTimelineRow): boolean case "working": return a.createdAt === (b as typeof a).createdAt; + case "waiting-background": { + const bw = b as typeof a; + return ( + a.createdAt === bw.createdAt && + a.description === bw.description && + a.taskCount === bw.taskCount + ); + } + case "turn-fold": { const bf = b as typeof a; return a.createdAt === bf.createdAt && a.label === bf.label && a.expanded === bf.expanded; diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index f759aa150be..f2035fe3c7e 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -161,6 +161,10 @@ interface MessagesTimelineProps { isWorking: boolean; activeTurnInProgress: boolean; activeTurnStartedAt: string | null; + pendingBackgroundTasks?: ReadonlyArray<{ + readonly taskId: string; + readonly description?: string | undefined; + }> | null; listRef: React.RefObject; timelineEntries: ReturnType; latestTurn: TimelineLatestTurn | null; @@ -195,6 +199,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ isWorking, activeTurnInProgress, activeTurnStartedAt, + pendingBackgroundTasks, listRef, timelineEntries, latestTurn, @@ -308,6 +313,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ expandedWorkGroupIds, isWorking, activeTurnStartedAt, + pendingBackgroundTasks: pendingBackgroundTasks ?? null, turnDiffSummaryByAssistantMessageId, revertTurnCountByUserMessageId, }), @@ -319,6 +325,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ expandedWorkGroupIds, isWorking, activeTurnStartedAt, + pendingBackgroundTasks, turnDiffSummaryByAssistantMessageId, revertTurnCountByUserMessageId, ], @@ -858,6 +865,7 @@ const TimelineRowContent = memo(function TimelineRowContent({ row }: { row: Time ) : null} {row.kind === "proposed-plan" ? : null} {row.kind === "working" ? : null} + {row.kind === "waiting-background" ? : null} ); }); @@ -1087,6 +1095,37 @@ function ProposedPlanTimelineRow({ ); } +/** + * Shown when the turn is settled but provider background tasks (background + * shells, subagents) are still running — the provider wakes the session when + * they finish, so the thread is waiting, not done. + */ +function WaitingBackgroundTimelineRow({ + row, +}: { + row: Extract; +}) { + const label = row.description + ? row.taskCount > 1 + ? `Waiting on ${row.taskCount} background tasks: ${row.description}, …` + : `Waiting on background task: ${row.description}` + : row.taskCount > 1 + ? `Waiting on ${row.taskCount} background tasks` + : "Waiting on a background task"; + return ( +
+
+ + + + + + {label} +
+
+ ); +} + function WorkingTimelineRow({ row }: { row: Extract }) { return (
diff --git a/packages/client-runtime/src/state/threadSettled.test.ts b/packages/client-runtime/src/state/threadSettled.test.ts index 59f9f1d6f55..60c5acd0bbf 100644 --- a/packages/client-runtime/src/state/threadSettled.test.ts +++ b/packages/client-runtime/src/state/threadSettled.test.ts @@ -22,7 +22,8 @@ const STALE = "2026-04-06T23:59:59.999Z"; function makeShell(input: { readonly settledOverride?: "settled" | "active" | null; readonly activityAt: string | null; - readonly sessionStatus?: "starting" | "running"; + readonly sessionStatus?: "starting" | "running" | "ready"; + readonly pendingBackgroundTasks?: ReadonlyArray<{ readonly taskId: string }>; readonly pending?: "approval" | "user-input"; }): OrchestrationThreadShell { const threadId = ThreadId.make("thread-1"); @@ -52,15 +53,18 @@ function makeShell(input: { settledOverride: input.settledOverride ?? null, settledAt: input.settledOverride === "settled" ? NOW : null, session: - input.sessionStatus === undefined + input.sessionStatus === undefined && input.pendingBackgroundTasks === undefined ? null : { threadId, - status: input.sessionStatus, + status: input.sessionStatus ?? "ready", providerName: "Codex", runtimeMode: "full-access", activeTurnId: null, lastError: null, + ...(input.pendingBackgroundTasks !== undefined + ? { pendingBackgroundTasks: input.pendingBackgroundTasks } + : {}), updatedAt: NOW, }, latestUserMessageAt: null, @@ -322,6 +326,29 @@ describe("canSettle", () => { ); }); + it("blocks settling while the session waits on provider background tasks", () => { + const waiting = makeShell({ + activityAt: FRESH, + sessionStatus: "ready", + pendingBackgroundTasks: [{ taskId: "bg-1" }], + }); + expect(canSettle(waiting, { now: NOW })).toBe(false); + // The background wait also overrides an explicit settled pin, exactly + // like a running session: blocked-in-motion work must remain visible. + expect( + effectiveSettled( + { ...waiting, settledOverride: "settled", settledAt: NOW }, + { now: NOW, autoSettleAfterDays: 3 }, + ), + ).toBe(false); + const drained = makeShell({ + activityAt: FRESH, + sessionStatus: "ready", + pendingBackgroundTasks: [], + }); + expect(canSettle(drained, { now: NOW })).toBe(true); + }); + it("blocks settling a queued turn start, only within the grace window", () => { const queued = { ...makeShell({ activityAt: FRESH }), diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index 3cb4f7e65f7..4740679db53 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -75,6 +75,18 @@ export function hasQueuedTurnStart( * The server enforces its own invariants; this client-side twin exists so * the UI can disable/reject before a round trip. */ +/** + * A session waiting on provider background tasks (e.g. a Claude background + * shell that outlived its turn) is "ready" on the wire but not done: the + * provider wakes it with a follow-up turn when the task finishes. Treated + * like a running session for settling purposes. + */ +export function hasPendingBackgroundTasks( + shell: Pick, +): boolean { + return (shell.session?.pendingBackgroundTasks?.length ?? 0) > 0; +} + export function canSettle( shell: Pick< OrchestrationThreadShell, @@ -84,6 +96,7 @@ export function canSettle( ): boolean { if (shell.hasPendingApprovals || shell.hasPendingUserInput) return false; if (shell.session?.status === "starting" || shell.session?.status === "running") return false; + if (hasPendingBackgroundTasks(shell)) return false; // Queued work is as blocked-on-progress as a live session: settling it // (or auto-settling it on a closed PR) would hide a just-requested turn. if (hasQueuedTurnStart(shell, options)) return false; @@ -109,6 +122,7 @@ export function effectiveSettled( // Blocked work must remain visible even when a user explicitly settled it. if (shell.hasPendingApprovals || shell.hasPendingUserInput) return false; if (shell.session?.status === "starting" || shell.session?.status === "running") return false; + if (hasPendingBackgroundTasks(shell)) return false; if (hasQueuedTurnStart(shell, { now: options.now })) { // The queued-turn blocker alone is forgivable: it is clock-derived, and // list callers pass a coarser `now` than the settle action used. When diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index b01df310062..647365e7a96 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -268,6 +268,19 @@ export const OrchestrationSessionStatus = Schema.Literals([ ]); export type OrchestrationSessionStatus = typeof OrchestrationSessionStatus.Type; +/** + * A provider background task (e.g. a Claude background shell or subagent) + * still running after the turn that spawned it completed. The provider will + * start a follow-up turn when the task finishes, so a session with pending + * background tasks is "ready" on the wire but visibly waiting in the UI. + */ +export const OrchestrationPendingBackgroundTask = Schema.Struct({ + taskId: TrimmedNonEmptyString, + description: Schema.optional(TrimmedNonEmptyString), + taskType: Schema.optional(TrimmedNonEmptyString), +}); +export type OrchestrationPendingBackgroundTask = typeof OrchestrationPendingBackgroundTask.Type; + export const OrchestrationSession = Schema.Struct({ threadId: ThreadId, status: OrchestrationSessionStatus, @@ -276,6 +289,9 @@ export const OrchestrationSession = Schema.Struct({ runtimeMode: RuntimeMode.pipe(Schema.withDecodingDefault(Effect.succeed(DEFAULT_RUNTIME_MODE))), activeTurnId: Schema.NullOr(TurnId), lastError: Schema.NullOr(TrimmedNonEmptyString), + // Optional and additive: old clients ignore it and degrade to today's + // idle-looking behavior; old servers never send it. + pendingBackgroundTasks: Schema.optional(Schema.Array(OrchestrationPendingBackgroundTask)), updatedAt: IsoDateTime, }); export type OrchestrationSession = typeof OrchestrationSession.Type; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index eb2563eff00..25ebd6b43f4 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -273,10 +273,23 @@ const SessionConfiguredPayload = Schema.Struct({ }); export type SessionConfiguredPayload = typeof SessionConfiguredPayload.Type; +/** + * Provider background task still running while no turn is active (e.g. a + * Claude background shell or subagent that outlives its turn). Carried on + * `session.state.changed` so ingestion can surface a waiting state. + */ +const RuntimeBackgroundTask = Schema.Struct({ + taskId: TrimmedNonEmptyStringSchema, + description: Schema.optional(TrimmedNonEmptyStringSchema), + taskType: Schema.optional(TrimmedNonEmptyStringSchema), +}); +export type RuntimeBackgroundTask = typeof RuntimeBackgroundTask.Type; + const SessionStateChangedPayload = Schema.Struct({ state: RuntimeSessionState, reason: Schema.optional(TrimmedNonEmptyStringSchema), detail: Schema.optional(Schema.Unknown), + backgroundTasks: Schema.optional(Schema.Array(RuntimeBackgroundTask)), }); export type SessionStateChangedPayload = typeof SessionStateChangedPayload.Type;