diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 642a642a80df1..e628fc5bcc4bd 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -793,6 +793,7 @@ it.effect( ["turn-item.updated", "cancelled"], ["turn-item.updated", "cancelled"], ["turn-item.updated", "cancelled"], + ["turn-item.updated", "completed"], ["provider-thread.updated", "idle"], ["provider-session.updated", "stopped"], ], diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 89e43e5b1ede4..46b739bec5732 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -634,6 +634,36 @@ export const make = Effect.gen(function* () { payload: cancelledItem(item, now), }); } + // Queued runs and unsent resumes leave no visible turn behind, so there is nothing to explain. + if (run.status === "queued" || unsentResumeRunIds.includes(run.id)) continue; + events.push({ + id: yield* allocateEventId(), + type: "turn-item.updated", + threadId: projection.thread.id, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { + id: ids.derive.runSignalTurnItem({ runId: run.id, signal: "server-recovery" }), + threadId: projection.thread.id, + runId: run.id, + nodeId: run.rootNodeId, + providerThreadId: run.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + // EventSink allocates the real position, after the run's existing items. + ordinal: 0, + status: "completed", + title: trigger === "startup" ? "Server restarted" : "Server shut down", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "system_notice", + message: detail, + }, + }); } // Blocking waits die with their turn; let child results wake the parent continuation as async tasks. const cutRunIds = new Set(runs.map((run) => run.id)); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a49e323770cc1..be6a414a1017d 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1591,10 +1591,16 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { assert.equal(native.runtimeRequests[0]?.responseCapability.type, "not_resumable"); yield* recovery.reconcile("shutdown"); yield* recovery.reconcile("startup"); - assert.deepEqual( - (yield* orchestrator.getThreadProjection(threadId)).runtimeRequests, - recovered.runtimeRequests, - ); + const rerecovered = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(rerecovered.runtimeRequests, recovered.runtimeRequests); + const runItems = rerecovered.turnItems.filter((item) => item.runId === runId); + const notices = runItems.filter((item) => item.type === "system_notice"); + assert.lengthOf(notices, scope === "run" ? 1 : 0); + if (scope === "run") { + assert.equal(notices[0]?.title, "Server restarted"); + assert.equal(notices[0]?.nodeId, recovered.runs[0]?.rootNodeId); + assert.equal(notices[0]?.ordinal, Math.max(...runItems.map((item) => item.ordinal))); + } yield* orchestrator.dispatch({ type: "runtime-request.respond", commandId: CommandId.make(`recovery-answer-${trigger}-${scope}`), @@ -8951,6 +8957,12 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ), [run.id, resumed.runs.at(-1)!.id], ); + assert.deepEqual( + resumed.turnItems.flatMap((item) => + item.type === "system_notice" && item.title === "Server restarted" ? [item.runId] : [], + ), + scenario === "resume" ? [] : [continuation.id], + ); assert.deepEqual( resumed.runs.map((row) => row.status), scenario === "resume" ? ["failed", "cancelled", "starting"] : ["failed", "cancelled"],