diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index 7fa6065109c3..cd8e53d90e8a 100644 --- 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 index f855771b8f01..8dac13ea9c2f 100644 --- 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 index 890c8ae55c53..48702066955c 100644 --- 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 index 1b44054c7a32..5535e34968da 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1527,6 +1527,116 @@ 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 + `, + }); + + 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 +3886,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 index eac3ede9c1ee..d284a913c44a 100644 --- 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, 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, + 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 index 792ef92310ff..79ff24a598bf 100644 --- 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 index dd341a7f7859..eef3d863b4ee 100644 --- 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 index 36d884d8ac98..29f7a92c16e6 100644 --- 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 index d8226648e9f3..a788cfc7eacd 100644 --- 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"), diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 050650b11515..ba7c2627d350 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -18,6 +18,7 @@ import * as Stream from "effect/Stream"; import { OrchestrationCommandInvariantError } from "./orchestration/Errors.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { UnterminatedProviderTask } from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { ProviderSessionDirectoryPersistenceError, ProviderSessionNotFoundError, @@ -72,14 +73,19 @@ const makeProviderService = (liveThreadIds: ReadonlyArray = []) => streamEvents: Stream.empty, }) satisfies ProviderService.ProviderService["Service"]; -const queryWithThreads = (threads: ReadonlyArray>) => +const queryWithThreads = ( + threads: ReadonlyArray>, + tasks: ReadonlyArray = [], +) => ({ getUserInputActivity: () => Effect.die("unused"), getCommandReadModel: () => Effect.succeed({ threads } as never), + listUnterminatedTasks: () => Effect.succeed(tasks), }) as unknown as ProjectionSnapshotQuery.ProjectionSnapshotQuery["Service"]; const runReconciliation = (input: { readonly threads: ReadonlyArray>; + readonly tasks?: ReadonlyArray; readonly continueAfterRestart?: boolean; readonly liveThreadIds?: ReadonlyArray; readonly providerService?: ProviderService.ProviderService["Service"]; @@ -89,7 +95,7 @@ const runReconciliation = (input: { ServerRuntimeStartup.reconcileProviderSessions.pipe( Effect.provideService( ProjectionSnapshotQuery.ProjectionSnapshotQuery, - queryWithThreads(input.threads), + queryWithThreads(input.threads, input.tasks), ), Effect.provideService( ProviderService.ProviderService, @@ -989,3 +995,239 @@ it.effect("settles failed opt-in recovery without retrying the provider turn", ( }); }), ); + +it.effect("marks a ready thread whose background tasks were still running", () => { + const ready = makeThread("thread-mark-background", "ready"); + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + return ServerRuntimeStartup.markRunningProviderSessionsForContinuation.pipe( + Effect.provideService( + ProjectionSnapshotQuery.ProjectionSnapshotQuery, + queryWithThreads( + [ready], + [ + { + threadId: ready.id, + taskId: "task-explore", + label: "Explore the repo", + }, + ], + ), + ), + Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, { + getBinding: (threadId) => + Effect.succeedSome({ + threadId, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId, + resumeCursor: { threadId }, + runtimePayload: { activeTurnId: null }, + }), + upsert: (binding) => Effect.sync(() => upserts.push(binding)), + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }), + Effect.tap((marked) => + Effect.sync(() => { + assert.deepStrictEqual(marked, [ready.id]); + assert.deepStrictEqual(upserts[0]?.runtimePayload, { + activeTurnId: null, + continueAfterServerUpdate: "restart-background-tasks", + continueAfterServerUpdatePrepared: null, + continueAfterServerUpdateTasks: [{ taskId: "task-explore", label: "Explore the repo" }], + }); + }), + ), + ); +}); + +it.effect("continues a ready thread by naming the background tasks a restart stopped", () => + Effect.gen(function* () { + const thread = makeThread("thread-background-continue", "ready"); + const continued = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const dispatched: OrchestrationCommand[] = []; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + tasks: [{ threadId: thread.id, taskId: "task-shell", label: "tail the logs" }], + providerService: { + ...makeProviderService(), + getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), + sendTurn: (input) => + Effect.gen(function* () { + sends.push(input); + yield* Deferred.succeed(continued, undefined); + return { threadId: input.threadId, turnId: TurnId.make("turn-continued-background") }; + }), + }, + directory: { + getBinding: () => + Effect.succeedSome({ + threadId: thread.id, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId, + status: "running" as const, + resumeCursor: { threadId: thread.id }, + runtimePayload: { activeTurnId: null }, + }), + upsert: () => Effect.void, + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.sync(() => { + dispatched.push(command); + return { sequence: dispatched.length }; + }), + }); + yield* Deferred.await(continued); + assert.deepStrictEqual(sends, [ + { + threadId: thread.id, + input: + "Continue where you left off.\n\nThese background tasks were stopped by the server restart and did not finish:\n- tail the logs", + interactionMode: "default", + }, + ]); + const stopped = dispatched.find((command) => command.type === "thread.activity.append"); + assert.equal(stopped?.type, "thread.activity.append"); + if (stopped?.type === "thread.activity.append") { + assert.equal(stopped.activity.kind, "task.completed"); + assert.deepStrictEqual(stopped.activity.payload, { + taskId: "task-shell", + status: "stopped", + title: "tail the logs", + summary: "Stopped because the server restarted.", + detail: "Stopped because the server restarted.", + }); + } + assert.equal( + dispatched.some( + (command) => command.type === "thread.session.set" && command.session.status === "error", + ), + false, + ); + }), +); + +it.effect( + "names recorded background tasks when retrying a starting session after they were settled", + () => + Effect.gen(function* () { + const thread = makeThread("thread-background-retry", "starting"); + const continued = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + tasks: [], + providerService: { + ...makeProviderService(), + getCapabilities: () => + Effect.succeed({ + sessionModelSwitch: "in-session" as const, + promptlessTurnContinuation: true, + }), + sendTurn: (input) => + Effect.gen(function* () { + sends.push(input); + yield* Deferred.succeed(continued, undefined); + return { threadId: input.threadId, turnId: TurnId.make("turn-retried-background") }; + }), + }, + directory: { + getBinding: () => + Effect.succeedSome({ + threadId: thread.id, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId, + status: "starting" as const, + resumeCursor: { threadId: thread.id }, + runtimePayload: { + activeTurnId: null, + continueAfterServerUpdate: "restart-background-tasks", + continueAfterServerUpdatePrepared: true, + continueAfterServerUpdateTasks: [{ taskId: "task-shell", label: "tail the logs" }], + }, + }), + upsert: () => Effect.void, + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: () => Effect.succeed({ sequence: 1 }), + }); + yield* Deferred.await(continued); + assert.deepStrictEqual(sends, [ + { + threadId: thread.id, + input: + "Continue where you left off.\n\nThese background tasks were stopped by the server restart and did not finish:\n- tail the logs", + interactionMode: "default", + }, + ]); + }), +); + +it.effect("settles unterminated background tasks when restart continuation is off", () => + Effect.gen(function* () { + const thread = makeThread("thread-background-settle", "ready"); + const sends: ProviderSendTurnInput[] = []; + const dispatched: OrchestrationCommand[] = []; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: false, + tasks: [{ threadId: thread.id, taskId: "task-agent", label: "Review the diff" }], + providerService: { + ...makeProviderService(), + sendTurn: (input) => + Effect.sync(() => { + sends.push(input); + return { threadId: input.threadId, turnId: TurnId.make("turn-should-not-send") }; + }), + }, + directory: { + getBinding: () => + Effect.succeedSome({ + threadId: thread.id, + provider: ProviderDriverKind.make("claudeAgent"), + providerInstanceId, + status: "running" as const, + resumeCursor: { threadId: thread.id }, + runtimePayload: { activeTurnId: null }, + }), + upsert: () => Effect.void, + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }, + dispatch: (command) => + Effect.sync(() => { + dispatched.push(command); + return { sequence: dispatched.length }; + }), + }); + assert.deepStrictEqual(sends, []); + const session = dispatched.find((command) => command.type === "thread.session.set"); + assert.equal(session?.type, "thread.session.set"); + if (session?.type === "thread.session.set") { + assert.equal(session.session.status, "error"); + assert.equal( + session.session.lastError, + "Provider session did not survive a server restart. Send a new message to continue.", + ); + } + const stopped = dispatched.find((command) => command.type === "thread.activity.append"); + assert.equal(stopped?.type, "thread.activity.append"); + if (stopped?.type === "thread.activity.append") { + assert.equal(stopped.activity.kind, "task.completed"); + assert.equal((stopped.activity.payload as { status?: string }).status, "stopped"); + } + }), +); diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index df879a2cf307..2d90b9b1a526 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -165,6 +165,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getUserInputActivity: () => Effect.die("unused"), listActivitiesByKind: () => Effect.succeed([]), + listUnterminatedTasks: () => Effect.succeed([]), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -294,6 +295,7 @@ it.effect.each([ Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getUserInputActivity: () => Effect.die("unused"), listActivitiesByKind: () => Effect.succeed([]), + listUnterminatedTasks: () => Effect.succeed([]), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -381,6 +383,7 @@ it.effect( Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getUserInputActivity: () => Effect.die("unused"), listActivitiesByKind: () => Effect.succeed([]), + listUnterminatedTasks: () => Effect.succeed([]), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), @@ -446,6 +449,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa Effect.provideService(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getUserInputActivity: () => Effect.die("unused"), listActivitiesByKind: () => Effect.succeed([]), + listUnterminatedTasks: () => Effect.succeed([]), getCommandReadModel: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 138dbcedfa85..02d44b050bfe 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -344,7 +344,16 @@ const runStartupPhase = (phase: string, effect: Effect.Effect) const ORPHANED_PROVIDER_SESSION_ERROR = "Provider session did not survive a server restart. Send a new message to continue."; const SERVER_UPDATE_CONTINUATION_KEY = "continueAfterServerUpdate"; +const SERVER_UPDATE_CONTINUATION_TASKS_KEY = "continueAfterServerUpdateTasks"; const SERVER_UPDATE_CONTINUATION_PROMPT = "Continue where you left off."; +const MAX_NAMED_STOPPED_TASKS = 12; +/** Marker turn used when background tasks outlived the turn that started them. */ +const BACKGROUND_RESTART_TURN_ID = TurnId.make("restart-background-tasks"); + +interface StoppedBackgroundTask { + readonly taskId: string; + readonly label: string; +} class ProviderSessionContinuationError extends Schema.TaggedError()( "ProviderSessionContinuationError", @@ -387,6 +396,73 @@ function readRuntimePayload(runtimePayload: unknown): Record { : {}; } +function readStoppedBackgroundTasks(runtimePayload: unknown): ReadonlyArray { + const value = readRuntimePayload(runtimePayload)[SERVER_UPDATE_CONTINUATION_TASKS_KEY]; + if (!Array.isArray(value)) { + return []; + } + const tasks: StoppedBackgroundTask[] = []; + for (const entry of value) { + if (entry === null || typeof entry !== "object" || Array.isArray(entry)) { + continue; + } + const taskId = "taskId" in entry && typeof entry.taskId === "string" ? entry.taskId.trim() : ""; + const label = "label" in entry && typeof entry.label === "string" ? entry.label.trim() : ""; + if (taskId.length === 0) { + continue; + } + tasks.push({ taskId, label: label.length > 0 ? label : taskId }); + } + return tasks; +} + +function continuationPromptForTasks(tasks: ReadonlyArray): string { + if (tasks.length === 0) { + return SERVER_UPDATE_CONTINUATION_PROMPT; + } + const lines = tasks.slice(0, MAX_NAMED_STOPPED_TASKS).map((task) => `- ${task.label}`); + if (tasks.length > MAX_NAMED_STOPPED_TASKS) { + lines.push(`- and ${tasks.length - MAX_NAMED_STOPPED_TASKS} more`); + } + return `${SERVER_UPDATE_CONTINUATION_PROMPT}\n\nThese background tasks were stopped by the server restart and did not finish:\n${lines.join("\n")}`; +} + +function clearedContinuationPayload(runtimePayload: unknown): Record { + const payload = readRuntimePayload(runtimePayload); + return { + ...payload, + [SERVER_UPDATE_CONTINUATION_KEY]: null, + continueAfterServerUpdatePrepared: null, + ...(SERVER_UPDATE_CONTINUATION_TASKS_KEY in payload + ? { [SERVER_UPDATE_CONTINUATION_TASKS_KEY]: null } + : {}), + }; +} + +const loadUnterminatedTasksByThread = Effect.gen(function* () { + const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const tasks = yield* query + .listUnterminatedTasks() + .pipe( + Effect.catch((cause) => + Effect.logWarning("failed to read unterminated provider tasks", { cause }).pipe( + Effect.as([]), + ), + ), + ); + const tasksByThread = new Map(); + for (const task of tasks) { + const label = task.label.trim(); + const existing = tasksByThread.get(task.threadId) ?? []; + existing.push({ + taskId: task.taskId, + label: label.length > 0 ? label : task.taskId, + }); + tasksByThread.set(task.threadId, existing); + } + return tasksByThread; +}); + const isServerUpdateThreadContinuationError = Schema.is(ServerUpdateThreadContinuationError); function readServerUpdateContinuationTurnId(runtimePayload: unknown): TurnId | null { @@ -406,19 +482,24 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const { threads } = yield* query.getCommandReadModel(); - const running = threads.filter( - (thread) => - thread.archivedAt === null && - thread.deletedAt === null && - thread.session?.status === "running" && - thread.session.activeTurnId !== null, - ); + const tasksByThread = yield* loadUnterminatedTasksByThread; + const running = threads.filter((thread) => { + if (thread.archivedAt !== null || thread.deletedAt !== null || thread.session === null) { + return false; + } + const openTasks = tasksByThread.get(thread.id) ?? []; + return ( + (thread.session.status === "running" && thread.session.activeTurnId !== null) || + openTasks.length > 0 + ); + }); const marked: ThreadId[] = []; return yield* Effect.gen(function* () { for (const thread of running) { - const activeTurnId = thread.session?.activeTurnId; - if (activeTurnId === null || activeTurnId === undefined) { + const activeTurnId = thread.session?.activeTurnId ?? null; + const openTasks = tasksByThread.get(thread.id) ?? []; + if (activeTurnId === null && openTasks.length === 0) { continue; } const binding = yield* directory.getBinding(thread.id); @@ -432,8 +513,9 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( ...binding.value, runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), - [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId, + [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId ?? BACKGROUND_RESTART_TURN_ID, continueAfterServerUpdatePrepared: null, + ...(openTasks.length > 0 ? { [SERVER_UPDATE_CONTINUATION_TASKS_KEY]: openTasks } : {}), }, }); marked.push(thread.id); @@ -460,11 +542,7 @@ const clearContinuationMarkers = ( onSome: (binding) => directory.upsert({ ...binding, - runtimePayload: { - ...readRuntimePayload(binding.runtimePayload), - [SERVER_UPDATE_CONTINUATION_KEY]: null, - continueAfterServerUpdatePrepared: null, - }, + runtimePayload: clearedContinuationPayload(binding.runtimePayload), }), }), ), @@ -503,6 +581,7 @@ export const reconcileProviderSessions = Effect.gen(function* () { (yield* providerService.listSessions()).map((session) => session.threadId), ); const { threads } = yield* query.getCommandReadModel(); + const tasksByThread = yield* loadUnterminatedTasksByThread; // Provider startup can report ready before the continuation is submitted. // Find those markers in one read rather than querying every idle thread. const preparedThreadIds = new Set( @@ -538,7 +617,8 @@ export const reconcileProviderSessions = Effect.gen(function* () { (thread.session.status === "starting" || thread.session.status === "running" || thread.session.activeTurnId !== null || - (thread.session.status === "ready" && preparedThreadIds.has(thread.id))) && + (thread.session.status === "ready" && + (preparedThreadIds.has(thread.id) || (tasksByThread.get(thread.id)?.length ?? 0) > 0))) && !liveThreadIds.has(thread.id), ); @@ -585,22 +665,77 @@ export const reconcileProviderSessions = Effect.gen(function* () { Option.isSome(binding) && binding.value.status === "running" && binding.value.resumeCursor != null; + const openTasks = tasksByThread.get(thread.id) ?? []; + const recordedTasks = Option.isSome(binding) + ? readStoppedBackgroundTasks(binding.value.runtimePayload) + : []; + // Projection rows are the crash signal. A continuation marker keeps the + // names when this process already settled those rows and then exited again. + const stoppedTasks = + openTasks.length > 0 ? openTasks : continuationMarkerPresent ? recordedTasks : []; + const continueBackgroundTasks = + continueAfterRestartFor(thread.projectId) && + stoppedTasks.length > 0 && + session.activeTurnId === null; + const settleStoppedTasks = Effect.gen(function* () { + if (stoppedTasks.length === 0) { + return; + } + const settledAt = DateTime.formatIso(yield* DateTime.now); + yield* Effect.forEach( + stoppedTasks, + (task) => + Effect.gen(function* () { + yield* orchestrationEngine + .dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId: thread.id, + activity: { + id: EventId.make(`task-restart:${thread.id}:${task.taskId}`), + tone: "info", + kind: "task.completed", + summary: "Task stopped", + payload: { + taskId: task.taskId, + status: "stopped", + title: task.label, + summary: "Stopped because the server restarted.", + detail: "Stopped because the server restarted.", + }, + turnId: null, + createdAt: settledAt, + }, + createdAt: settledAt, + }) + .pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterrupts(cause), + (cause) => + Effect.logWarning("failed to settle interrupted background task", { + threadId: thread.id, + taskId: task.taskId, + cause, + }), + ), + ); + }), + { discard: true }, + ); + }); const settleAsError = (lastError: string) => Effect.gen(function* () { + yield* settleStoppedTasks; yield* Effect.gen(function* () { if (Option.isSome(binding)) { yield* directory.upsert({ ...binding.value, status: "stopped", runtimePayload: { - ...readRuntimePayload(binding.value.runtimePayload), + ...(continuationMarkerPresent || interruptedByRestart || continueBackgroundTasks + ? clearedContinuationPayload(binding.value.runtimePayload) + : readRuntimePayload(binding.value.runtimePayload)), activeTurnId: null, - ...(continuationMarkerPresent || interruptedByRestart - ? { - [SERVER_UPDATE_CONTINUATION_KEY]: null, - continueAfterServerUpdatePrepared: null, - } - : {}), }, }); } @@ -645,8 +780,11 @@ export const reconcileProviderSessions = Effect.gen(function* () { if ( Option.isSome(binding) && - (continuationMarked || interruptedByRestart) && - (session.status === "running" || session.status === "starting" || preparedWhileReady) && + (continuationMarked || interruptedByRestart || continueBackgroundTasks) && + (session.status === "running" || + session.status === "starting" || + preparedWhileReady || + (session.status === "ready" && continueBackgroundTasks)) && binding.value.resumeCursor != null && thread.archivedAt === null && thread.deletedAt === null @@ -658,9 +796,15 @@ export const reconcileProviderSessions = Effect.gen(function* () { runtimePayload: { ...readRuntimePayload(binding.value.runtimePayload), // Keep recovery durable if this process also exits before sending. - [SERVER_UPDATE_CONTINUATION_KEY]: session.activeTurnId ?? continuationTurnId, + [SERVER_UPDATE_CONTINUATION_KEY]: + session.activeTurnId ?? + continuationTurnId ?? + (stoppedTasks.length > 0 ? BACKGROUND_RESTART_TURN_ID : null), continueAfterServerUpdatePrepared: true, activeTurnId: null, + ...(stoppedTasks.length > 0 + ? { [SERVER_UPDATE_CONTINUATION_TASKS_KEY]: stoppedTasks } + : {}), }, }); const resumedAt = DateTime.formatIso(yield* DateTime.now); @@ -700,11 +844,12 @@ export const reconcileProviderSessions = Effect.gen(function* () { }); } const capabilities = yield* providerService.getCapabilities(providerInstanceId); + yield* settleStoppedTasks; yield* providerService.sendTurn({ threadId: thread.id, - ...(capabilities.promptlessTurnContinuation === true + ...(capabilities.promptlessTurnContinuation === true && stoppedTasks.length === 0 ? { continuation: true } - : { input: SERVER_UPDATE_CONTINUATION_PROMPT }), + : { input: continuationPromptForTasks(stoppedTasks) }), interactionMode: thread.interactionMode, }); });