From ebefa73d095c440cea0bef7fac5c1836f8b0eef9 Mon Sep 17 00:00:00 2001 From: maco Date: Fri, 25 Sep 2026 10:32:23 +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 and stays within the provider send limit. Otherwise those tasks are settled once 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, and an empty open-task list clears any stale marker. Malformed activity payloads are skipped, and plan or dream tasks stay excluded when a later status row omits taskType. Fixes #13400 Co-authored-by: maco --- .../checkpointing/CheckpointDiffQuery.test.ts | 5 + .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.test.ts | 151 +++++++ .../Layers/ProjectionSnapshotQuery.ts | 112 +++++ .../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 | 397 +++++++++++++++++- apps/server/src/serverRuntimeStartup.test.ts | 4 + apps/server/src/serverRuntimeStartup.ts | 259 ++++++++++-- 12 files changed, 920 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..6a5af19cbcea 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1527,6 +1527,117 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ); + // Latest decisive task row per task. Status-free progress is not decisive: + // a late heartbeat must not resurrect a task that already finished. `stopped` + // is included because task.completed uses it for an interrupted task. + const listUnterminatedTaskRows = SqlSchema.findAll({ + Request: Schema.Struct({}), + Result: Schema.Struct({ + threadId: ThreadId, + taskId: Schema.String, + label: Schema.String, + }), + execute: () => sql` + WITH decisive AS ( + SELECT + a.thread_id AS threadId, + trim(json_extract(a.payload_json, '$.taskId')) AS taskId, + a.kind AS kind, + json_extract(a.payload_json, '$.status') AS status, + json_extract(a.payload_json, '$.taskType') AS taskType, + CASE + WHEN json_type(a.payload_json, '$.title') = 'text' + AND length(trim(json_extract(a.payload_json, '$.title'))) > 0 + THEN trim(json_extract(a.payload_json, '$.title')) + WHEN a.kind = 'task.started' + AND json_type(a.payload_json, '$.detail') = 'text' + AND length(trim(json_extract(a.payload_json, '$.detail'))) > 0 + THEN trim(json_extract(a.payload_json, '$.detail')) + ELSE NULL + END AS label, + COALESCE(a.sequence, -1) AS sequence, + a.created_at AS createdAt, + a.activity_id AS activityId + FROM projection_thread_activities a + JOIN projection_threads t ON t.thread_id = a.thread_id + WHERE t.deleted_at IS NULL + AND t.archived_at IS NULL + AND json_valid(a.payload_json) + AND json_type(a.payload_json, '$.taskId') = 'text' + AND length(trim(json_extract(a.payload_json, '$.taskId'))) > 0 + AND ( + a.kind = 'task.completed' + OR a.kind = 'task.started' + OR ( + a.kind IN ('task.progress', 'task.updated') + AND json_type(a.payload_json, '$.status') = 'text' + ) + ) + ), + latest AS ( + SELECT + threadId, + taskId, + kind, + status, + taskType, + label, + ROW_NUMBER() OVER ( + PARTITION BY threadId, taskId + ORDER BY sequence DESC, createdAt DESC, activityId DESC + ) AS rn + FROM decisive + ), + open_tasks AS ( + SELECT threadId, taskId, label + FROM latest + WHERE rn = 1 + AND kind != 'task.completed' + AND COALESCE(status, '') NOT IN ( + 'completed', 'failed', 'stopped', 'cancelled', 'interrupted', 'idle' + ) + AND NOT EXISTS ( + SELECT 1 FROM decisive d2 + WHERE d2.threadId = latest.threadId + AND d2.taskId = latest.taskId + AND d2.taskType IN ('plan', 'dream') + ) + ), + labels AS ( + SELECT + d.threadId AS threadId, + d.taskId AS taskId, + d.label AS label, + ROW_NUMBER() OVER ( + PARTITION BY d.threadId, d.taskId + ORDER BY d.sequence DESC, d.createdAt DESC, d.activityId DESC + ) AS labelRn + FROM decisive d + INNER JOIN open_tasks o + ON o.threadId = d.threadId AND o.taskId = d.taskId + WHERE d.label IS NOT NULL + ) + SELECT + o.threadId AS "threadId", + o.taskId AS "taskId", + COALESCE(l.label, o.label, o.taskId) AS "label" + FROM open_tasks o + LEFT JOIN labels l + ON l.threadId = o.threadId AND l.taskId = o.taskId AND l.labelRn = 1 + `, + }); + + /** Open background tasks whose latest decisive row is still non-terminal. */ + const listUnterminatedTasks: ProjectionSnapshotQueryShape["listUnterminatedTasks"] = () => + listUnterminatedTaskRows({}).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.listUnterminatedTasks:query", + "ProjectionSnapshotQuery.listUnterminatedTasks:decodeRow", + ), + ), + ); + const listThreadActivityIdsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionThreadActivityIdRowSchema, @@ -3776,6 +3887,7 @@ pending_approval_requests AS ( getCommandReadModel, getUserInputActivity, listActivitiesByKind, + listUnterminatedTasks, getSnapshot, getShellSnapshot, getArchivedShellSnapshot, diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts 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..fa288b691ab1 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -4,6 +4,7 @@ import { type OrchestrationSessionStatus, ProviderDriverKind, ProviderInstanceId, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, type ProviderSendTurnInput, ThreadId, TurnId, @@ -18,6 +19,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 +74,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 +96,7 @@ const runReconciliation = (input: { ServerRuntimeStartup.reconcileProviderSessions.pipe( Effect.provideService( ProjectionSnapshotQuery.ProjectionSnapshotQuery, - queryWithThreads(input.threads), + queryWithThreads(input.threads, input.tasks), ), Effect.provideService( ProviderService.ProviderService, @@ -164,6 +171,7 @@ it.effect("marks active running sessions that have persisted resume state", () = activeTurnId: "turn-mark-active", continueAfterServerUpdate: active.session.activeTurnId, continueAfterServerUpdatePrepared: null, + continueAfterServerUpdateTasks: null, }); }), ), @@ -347,6 +355,7 @@ it.effect.each( continueAfterServerUpdate: continuationTurnId, continueAfterServerUpdatePrepared: true, activeTurnId: null, + continueAfterServerUpdateTasks: null, }, ); } @@ -875,6 +884,7 @@ for (const preparedStatus of [ activeTurnId: null, continueAfterServerUpdate: turnId, continueAfterServerUpdatePrepared: true, + continueAfterServerUpdateTasks: null, }); if (preparedStatus === "completed after update marking") { @@ -904,6 +914,7 @@ for (const preparedStatus of [ activeTurnId: null, continueAfterServerUpdate: null, continueAfterServerUpdatePrepared: null, + continueAfterServerUpdateTasks: null, }); }), ); @@ -966,6 +977,7 @@ it.effect("settles failed opt-in recovery without retrying the provider turn", ( activeTurnId: null, continueAfterServerUpdate: turnId, continueAfterServerUpdatePrepared: true, + continueAfterServerUpdateTasks: null, }, ]); assert.deepStrictEqual( @@ -989,3 +1001,384 @@ 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[] = []; + const dispatched: OrchestrationCommand[] = []; + 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: (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", + }, + ]); + assert.equal( + dispatched.some((command) => command.type === "thread.activity.append"), + false, + ); + }), +); + +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.filter((command) => command.type === "thread.activity.append"); + assert.equal(stopped.length, 1); + const [onlyStopped] = stopped; + assert.equal(onlyStopped?.type, "thread.activity.append"); + if (onlyStopped?.type === "thread.activity.append") { + assert.equal(onlyStopped.activity.kind, "task.completed"); + assert.equal((onlyStopped.activity.payload as { status?: string }).status, "stopped"); + } + }), +); + +it.effect("clears a stale continuation task list when no background tasks are open", () => { + const active = makeThread("thread-mark-stale-tasks", "running", TurnId.make("turn-mark-stale")); + const upserts: ProviderSessionDirectory.ProviderRuntimeBinding[] = []; + return ServerRuntimeStartup.markRunningProviderSessionsForContinuation.pipe( + Effect.provideService( + ProjectionSnapshotQuery.ProjectionSnapshotQuery, + queryWithThreads([active], []), + ), + Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, { + getBinding: (threadId) => + Effect.succeedSome({ + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId, + resumeCursor: { threadId }, + runtimePayload: { + activeTurnId: "turn-mark-stale", + continueAfterServerUpdateTasks: [{ taskId: "task-old", label: "Already finished" }], + }, + }), + upsert: (binding) => Effect.sync(() => upserts.push(binding)), + recordImportedTranscript: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), + }), + Effect.tap(() => + Effect.sync(() => { + assert.deepStrictEqual(upserts[0]?.runtimePayload, { + activeTurnId: "turn-mark-stale", + continueAfterServerUpdate: active.session.activeTurnId, + continueAfterServerUpdatePrepared: null, + continueAfterServerUpdateTasks: null, + }); + }), + ), + ); +}); + +it.effect("bounds an oversized stopped-task label inside the provider send limit", () => + Effect.gen(function* () { + const thread = makeThread("thread-background-long-label", "ready"); + const continued = yield* Deferred.make(); + const sends: ProviderSendTurnInput[] = []; + const label = "x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS + 500); + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + tasks: [{ threadId: thread.id, taskId: "task-long", label }], + 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-long-label") }; + }), + }, + 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: () => Effect.succeed({ sequence: 1 }), + }); + yield* Deferred.await(continued); + const input = sends[0]?.input ?? ""; + assert.equal(sends.length, 1); + assert.isAtMost(input.length, PROVIDER_SEND_TURN_MAX_INPUT_CHARS); + assert.equal(input.startsWith("Continue where you left off."), true); + assert.equal(input.includes("These background tasks were stopped"), true); + }), +); + +it.effect("settles an open background task once when continuation fails", () => + Effect.gen(function* () { + const thread = makeThread("thread-background-settle-once", "ready"); + const settled = yield* Deferred.make(); + const dispatched: OrchestrationCommand[] = []; + yield* runReconciliation({ + threads: [thread], + continueAfterRestart: true, + tasks: [{ threadId: thread.id, taskId: "task-once", label: "tail the logs" }], + providerService: { + ...makeProviderService(), + getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), + sendTurn: () => Effect.die("continuation failed"), + }, + 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.gen(function* () { + dispatched.push(command); + if (command.type === "thread.session.set" && command.session.status === "error") { + yield* Deferred.succeed(settled, undefined); + } + return { sequence: dispatched.length }; + }), + }); + yield* Deferred.await(settled); + const stopped = dispatched.filter( + (command) => + command.type === "thread.activity.append" && command.activity.kind === "task.completed", + ); + assert.equal(stopped.length, 1); + }), +); 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..4f77aec97e6b 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/apps/server/src/serverRuntimeStartup.ts @@ -9,6 +9,7 @@ import { type OrchestrationProjectShell, ProjectId, ProviderInstanceId, + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, ThreadId, TurnId, WORKTREE_SETUP_ACTIVITY_KIND, @@ -344,7 +345,18 @@ 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; +} + +const STOPPED_TASKS_PROMPT_HEADER = `${SERVER_UPDATE_CONTINUATION_PROMPT}\n\nThese background tasks were stopped by the server restart and did not finish:\n`; class ProviderSessionContinuationError extends Schema.TaggedError()( "ProviderSessionContinuationError", @@ -387,6 +399,117 @@ function readRuntimePayload(runtimePayload: unknown): Record { : {}; } +/** Task names recorded beside a continuation marker, ignoring malformed entries. */ +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; +} + +/** Shorten one task label so it fits the characters still left in the prompt. */ +function boundedTaskLabel(label: string, maxChars: number): string { + if (maxChars <= 0) { + return ""; + } + if (label.length <= maxChars) { + return label; + } + if (maxChars === 1) { + return "…"; + } + return `${label.slice(0, maxChars - 1)}…`; +} + +/** + * Continuation prompt naming tasks a restart stopped. + * The fixed text and every label stay within the provider send limit. + */ +function continuationPromptForTasks(tasks: ReadonlyArray): string { + const limit = PROVIDER_SEND_TURN_MAX_INPUT_CHARS; + if (tasks.length === 0) { + return SERVER_UPDATE_CONTINUATION_PROMPT.length <= limit + ? SERVER_UPDATE_CONTINUATION_PROMPT + : SERVER_UPDATE_CONTINUATION_PROMPT.slice(0, limit); + } + if (STOPPED_TASKS_PROMPT_HEADER.length >= limit) { + return STOPPED_TASKS_PROMPT_HEADER.slice(0, limit); + } + const named = tasks.slice(0, MAX_NAMED_STOPPED_TASKS); + const overflowCount = tasks.length - named.length; + const overflowLine = overflowCount > 0 ? `- and ${overflowCount} more` : ""; + const overflowCost = overflowLine.length === 0 ? 0 : overflowLine.length + 1; + let remaining = limit - STOPPED_TASKS_PROMPT_HEADER.length - overflowCost; + const lines: string[] = []; + for (const task of named) { + const prefix = lines.length === 0 ? "- " : "\n- "; + if (remaining <= prefix.length) { + break; + } + const label = boundedTaskLabel(task.label, remaining - prefix.length); + if (label.length === 0) { + break; + } + lines.push(`${prefix}${label}`); + remaining -= prefix.length + label.length; + } + const overflow = + overflowLine.length === 0 ? "" : `${lines.length > 0 ? "\n" : ""}${overflowLine}`; + const prompt = `${STOPPED_TASKS_PROMPT_HEADER}${lines.join("")}${overflow}`; + return prompt.length <= limit ? prompt : prompt.slice(0, limit); +} + +/** Drop continuation markers, including a recorded task list, after recovery finishes. */ +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 } + : {}), + }; +} + +/** Open background tasks grouped by thread. A query failure yields an empty map. */ +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 { @@ -402,23 +525,32 @@ const toServerUpdateThreadContinuationError = (cause: unknown) => ? cause : new ServerUpdateThreadContinuationError({ cause }); +/** + * Remember running turns and threads whose background tasks are still open + * so the next process can continue them after an update. + */ 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 +564,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, + [SERVER_UPDATE_CONTINUATION_TASKS_KEY]: openTasks.length > 0 ? openTasks : null, }, }); marked.push(thread.id); @@ -446,6 +579,7 @@ export const markRunningProviderSessionsForContinuation = Effect.gen(function* ( ); }).pipe(Effect.mapError(toServerUpdateThreadContinuationError)); +/** Remove continuation markers from the listed provider session bindings. */ const clearContinuationMarkers = ( directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"], threadIds: ReadonlyArray, @@ -460,11 +594,7 @@ const clearContinuationMarkers = ( onSome: (binding) => directory.upsert({ ...binding, - runtimePayload: { - ...readRuntimePayload(binding.runtimePayload), - [SERVER_UPDATE_CONTINUATION_KEY]: null, - continueAfterServerUpdatePrepared: null, - }, + runtimePayload: clearedContinuationPayload(binding.runtimePayload), }), }), ), @@ -478,6 +608,10 @@ const clearProviderSessionContinuationMarkers = (threadIds: ReadonlyArray 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 +673,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 +721,81 @@ 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 tasksToSettle = openTasks; + let tasksSettled = false; + const continueBackgroundTasks = + continueAfterRestartFor(thread.projectId) && + stoppedTasks.length > 0 && + session.activeTurnId === null; + /** Write one stopped activity per open task. Recorded names are not settled again. */ + const settleStoppedTasks = Effect.gen(function* () { + if (tasksSettled || tasksToSettle.length === 0) { + return; + } + tasksSettled = true; + const settledAt = DateTime.formatIso(yield* DateTime.now); + yield* Effect.forEach( + tasksToSettle, + (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 +840,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 +856,13 @@ 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, + [SERVER_UPDATE_CONTINUATION_TASKS_KEY]: stoppedTasks.length > 0 ? stoppedTasks : null, }, }); const resumedAt = DateTime.formatIso(yield* DateTime.now); @@ -700,11 +902,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, }); });