From f6c39e0260ddcd73d38bbd112f39417fa5b19173 Mon Sep 17 00:00:00 2001 From: maco Date: Fri, 25 Sep 2026 08:37:04 +0000 Subject: [PATCH] fix(server): continue threads whose background tasks outlived the turn A thread can finish its turn while subagents or background shells are still running. Restart continuation only resumed sessions that were still running with an active turn, so those threads stayed ready and the work disappeared. Startup now treats unterminated task rows like an interrupted turn. When continue-after-restart is on, the continuation prompt names the stopped tasks. Otherwise those tasks are settled as stopped and the thread gets the existing restart notice. A retry after those tasks were already settled restores the recorded names from the continuation marker. Malformed activity payloads are skipped, and plan or dream tasks stay excluded when a later status row omits taskType. Fixes #13400 Grok 4.7 via Cursor cloud agent. Co-authored-by: maco --- .../checkpointing/CheckpointDiffQuery.test.ts | 5 + .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.test.ts | 151 +++++++++++ .../Layers/ProjectionSnapshotQuery.ts | 111 ++++++++ .../Services/ProjectionSnapshotQuery.ts | 17 ++ .../src/project/AgentSessionScanner.test.ts | 1 + .../project/ProjectSetupScriptRunner.test.ts | 1 + .../provider/Layers/ProviderService.test.ts | 1 + .../Layers/ProviderSessionReaper.test.ts | 1 + .../serverRuntimeStartup.reconcile.test.ts | 246 +++++++++++++++++- apps/server/src/serverRuntimeStartup.test.ts | 4 + apps/server/src/serverRuntimeStartup.ts | 201 ++++++++++++-- 12 files changed, 710 insertions(+), 30 deletions(-) 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, }); });