From c6394b97c6ab9d177e70e74fac57b778545af570 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 5 Sep 2026 21:05:08 +1000 Subject: [PATCH 1/2] fix(orchestration): exclude rolled-back work from bounded recovery --- .../orchestration-v2/ProjectionStore.test.ts | 288 ++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 11 +- .../ProviderRuntimeRecoveryService.test.ts | 59 ++++ 3 files changed, 355 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 7ac010f1900b..bf09eb80141a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -28,6 +28,7 @@ import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import { isTurnItemAtOrBeforeRun, + layerMemory as projectionStoreMemoryLayer, ProjectionStoreV2, ProjectionStoreThreadNotFoundError, layer as projectionStoreLayer, @@ -51,6 +52,190 @@ const driver = ProviderDriverKind.make("codex"); const providerInstanceId = modelSelection.instanceId; const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate")(function* ( + suffix: string, +) { + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:${suffix}:rolled-back`); + const runId = RunId.make(`run:${suffix}:rolled-back`); + const rootNodeId = NodeId.make(`node:${suffix}:rolled-back`); + const run = { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${suffix}:rolled-back`), + rootNodeId, + activeAttemptId: null, + status: "running" as const, + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Rolled-back recovery candidate", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:run-created`), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: run, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:item-running`), + type: "turn-item.updated", + threadId, + runId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:${suffix}:rolled-back`), + threadId, + runId, + nodeId: rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "abandoned command", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "command_execution", + input: "sleep 60", + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:run-rolled-back`), + type: "run.updated", + threadId, + runId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { ...run, status: "rolled_back", completedAt: now }, + }); + + return threadId; +}); + +const addOrphanedRecoveryCandidate = Effect.fn("addOrphanedRecoveryCandidate")(function* ( + suffix: string, +) { + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:${suffix}:orphaned`); + const runId = RunId.make(`run:${suffix}:missing`); + const rootNodeId = NodeId.make(`node:${suffix}:orphaned`); + + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:thread-created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${suffix}`), + title: "Orphaned recovery candidate", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + yield* projectionStore.apply({ + id: EventId.make(`event:${suffix}:item-running`), + type: "turn-item.updated", + threadId, + runId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: TurnItemId.make(`item:${suffix}:orphaned`), + threadId, + runId, + nodeId: rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "orphaned command", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "command_execution", + input: "sleep 60", + }, + }); + + return threadId; +}); + it("includes imported runless history when selecting fork context through a run", () => { const firstRunId = RunId.make("run:projection-imported-fork:1"); const secondRunId = RunId.make("run:projection-imported-fork:2"); @@ -93,6 +278,24 @@ it("includes imported runless history when selecting fork context through a run" ); }); +it.effect("memory recovery selection ignores unfinished items from rolled-back runs", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("memory-recovery-candidates"); + + assert.notInclude(yield* projectionStore.getRecoveryThreadIds("runtime"), threadId); + }).pipe(Effect.provide(projectionStoreMemoryLayer)), +); + +it.effect("memory recovery selection includes unfinished items from missing runs", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStoreV2; + const threadId = yield* addOrphanedRecoveryCandidate("memory-recovery-candidates"); + + assert.include(yield* projectionStore.getRecoveryThreadIds("runtime"), threadId); + }).pipe(Effect.provide(projectionStoreMemoryLayer)), +); + it.layer(TestLayer)("ProjectionStoreV2", (it) => { it.effect("preserves stored provider usage when a terminal update omits it", () => Effect.gen(function* () { @@ -1267,6 +1470,91 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => { }), ); + it.effect("selects only threads with runtime state that needs recovery", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStoreV2; + const now = yield* DateTime.now; + const settledThreadId = ThreadId.make("thread:recovery-candidates:settled"); + const runningThreadId = ThreadId.make("thread:recovery-candidates:running"); + const rolledBackThreadId = yield* addRolledBackRecoveryCandidate("recovery-candidates"); + const orphanedThreadId = yield* addOrphanedRecoveryCandidate("recovery-candidates"); + const projectId = ProjectId.make("project:recovery-candidates"); + const makeThread = (threadId: ThreadId) => ({ + createdBy: "user" as const, + creationSource: "web" as const, + id: threadId, + projectId, + title: "Recovery candidate", + providerInstanceId, + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }); + + for (const threadId of [settledThreadId, runningThreadId]) { + yield* projectionStore.apply({ + id: EventId.make(`event:recovery-candidates:${threadId}:created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: makeThread(threadId), + }); + } + + const runId = RunId.make("run:recovery-candidates:running"); + const rootNodeId = NodeId.make("node:recovery-candidates:running"); + yield* projectionStore.apply({ + id: EventId.make("event:recovery-candidates:run-created"), + type: "run.created", + threadId: runningThreadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: { + id: runId, + threadId: runningThreadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:recovery-candidates:running"), + rootNodeId, + activeAttemptId: null, + status: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }); + + const recoveryThreadIds = yield* projectionStore.getRecoveryThreadIds("runtime"); + assert.include(recoveryThreadIds, runningThreadId); + assert.include(recoveryThreadIds, orphanedThreadId); + assert.notInclude(recoveryThreadIds, settledThreadId); + assert.notInclude(recoveryThreadIds, rolledBackThreadId); + }), + ); + it.effect("projects one shared provider session into multiple thread bindings", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 960da963b11e..1008d33dd1e0 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -359,7 +359,8 @@ function needsRecovery( projection.turnItems.some( (item) => ["command_execution", "dynamic_tool", "subagent"].includes(item.type) && - ["pending", "running", "waiting"].includes(item.status), + ["pending", "running", "waiting"].includes(item.status) && + !projection.runs.some((run) => run.id === item.runId && run.status === "rolled_back"), ) ); } @@ -2950,8 +2951,12 @@ export const layer: Layer.Layer = CROSS JOIN orchestration_v2_projection_subagents AS subagents ON subagents.provider_thread_id = pending_provider_threads.provider_thread_id UNION - SELECT thread_id FROM orchestration_v2_projection_turn_items - WHERE type IN ('command_execution', 'dynamic_tool', 'subagent') + SELECT item.thread_id FROM orchestration_v2_projection_turn_items AS item + WHERE NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS run + WHERE run.run_id = item.run_id AND run.status = 'rolled_back' + ) + AND type IN ('command_execution', 'dynamic_tool', 'subagent') AND status IN ('pending', 'running', 'waiting') UNION SELECT thread_id FROM orchestration_v2_effect_outbox diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 6aab67fc603b..dbb0cd113568 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -64,6 +64,65 @@ it.effect("leaves durable effects for the worker after runtime reconciliation", }), ); +it.effect("reads full projections only for threads that need runtime recovery", () => { + const settledThreadIds = Array.from({ length: 1_000 }, (_, index) => + ThreadId.make(`thread_recovery_settled_${index}`), + ); + const recoveryThreadId = ThreadId.make("thread_recovery_candidate"); + const projectionReads = vi.fn<(threadId: ThreadId) => void>(); + const layer = ProviderRuntimeRecovery.layer.pipe( + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.mergeAll( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getShellSnapshot: () => + Effect.succeed({ + schemaVersion: 2, + snapshotSequence: 0, + threads: [...settledThreadIds, recoveryThreadId].map((id) => ({ id })), + archivedThreads: [], + } as never), + getRecoveryThreadIds: () => Effect.succeed([recoveryThreadId]), + getThreadProjection: (threadId) => { + projectionReads(threadId); + return Effect.succeed({ + thread: { id: threadId }, + runtimeRequests: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runs: [], + attempts: [], + nodes: [], + subagents: [], + messages: [], + turnItems: [], + } as unknown as OrchestrationV2ThreadProjection); + }, + }), + Layer.mock(EventSink.EventSinkV2)({}), + IdAllocator.layer, + Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({ + runRecoveryOnce: Effect.succeed(false), + }), + Layer.mock(EffectOutbox.EffectOutboxV2)({ + cancelUnsettled: () => Effect.succeed([]), + signalCancellations: () => Effect.void, + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + + return Effect.gen(function* () { + yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover; + assert.deepEqual( + projectionReads.mock.calls.map(([threadId]) => threadId), + [recoveryThreadId], + ); + }).pipe(Effect.provide(layer)); +}); + it.effect("expires orphaned runtime requests before command readiness", () => { const threadId = ThreadId.make("thread_recovery_requests"); let committedInput: Parameters[0] | null = From ac52f0db67c5de619b6fdabf5bdcd5e3a8e726f1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 10 Sep 2026 16:26:26 -0700 Subject: [PATCH 2/2] test(server): adapt recovery scan coverage to bounded projections --- .../orchestration-v2/ProviderRuntimeRecoveryService.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index dbb0cd113568..c9c4a820e611 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -64,7 +64,7 @@ it.effect("leaves durable effects for the worker after runtime reconciliation", }), ); -it.effect("reads full projections only for threads that need runtime recovery", () => { +it.effect("reads recovery projections only for threads that need runtime recovery", () => { const settledThreadIds = Array.from({ length: 1_000 }, (_, index) => ThreadId.make(`thread_recovery_settled_${index}`), ); @@ -83,7 +83,7 @@ it.effect("reads full projections only for threads that need runtime recovery", archivedThreads: [], } as never), getRecoveryThreadIds: () => Effect.succeed([recoveryThreadId]), - getThreadProjection: (threadId) => { + getRuntimeRecoveryProjection: (threadId) => { projectionReads(threadId); return Effect.succeed({ thread: { id: threadId },