From f9bb5ab69609e862e48b409d25d1e888ec739a70 Mon Sep 17 00:00:00 2001 From: Guillermo Casanova Date: Wed, 9 Sep 2026 11:42:34 -0300 Subject: [PATCH 1/3] fix(server): skip imports of native agent sessions --- .../src/project/AgentSessionImporter.test.ts | 82 +++++++++++++++++-- .../src/project/AgentSessionImporter.ts | 25 ++++++ 2 files changed, 101 insertions(+), 6 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 2438edca8b1b3..4e93c3ab2aa19 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -200,7 +200,7 @@ const runImport = (input: { it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { describe("importRecentAgentThreads", () => { - it.effect("uses the project root and stores provider-specific resume cursors", () => + it.effect("imports sessions owned by another provider instance", () => Effect.gen(function* () { const commands: Array = []; const bindings: Array = []; @@ -235,7 +235,16 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeedNone, listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => + Effect.succeed([ + { + threadId: ThreadId.make("native-other-instance"), + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex-other"), + resumeCursor: { threadId: "codex-session" }, + lastSeenAt: "2026-08-24T10:00:00.000Z", + }, + ]), }); const result = yield* runImport({ @@ -340,7 +349,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.die("unused"), getBinding: () => Effect.die("must not read a scanner skip binding"), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }); const result = yield* runImport({ @@ -418,7 +427,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { getBinding: () => Effect.succeed(bindings[0] === undefined ? Option.none() : Option.some(bindings[0])), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }); const snapshots = makeSnapshotsLayer({ project: makeProject(), @@ -459,7 +468,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeedSome(runningBinding), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }); const engine = OrchestrationEngine.OrchestrationEngineService.of({ dispatch: () => Effect.die("must not replay history or settle active work"), @@ -514,7 +523,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.die("unused"), getBinding: () => Effect.succeedNone, listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }); const result = yield* runImport({ @@ -582,6 +591,67 @@ const integrationLayer = Layer.mergeAll( ); it.layer(integrationLayer)("AgentSessionImporter integration", (it) => { + for (const source of ["codex", "claudeAgent"] as const) { + it.effect(`does not import a ${source} session already owned by a native thread`, () => + Effect.gen(function* () { + const engine = yield* OrchestrationEngine.OrchestrationEngineService; + const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const projectId = ProjectId.make(`native-import-${source}`); + const threadId = ThreadId.make(`native-${source}`); + const thread = { + ...makeThread(source), + providerSessionId: + source === "codex" ? "native-codex-session" : "123e4567-e89b-42d3-a456-426614174001", + }; + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make(`create-${projectId}`), + projectId, + title: "Native project", + workspaceRoot: `/tmp/${projectId}`, + defaultModelSelection: null, + createdAt: thread.createdAt, + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create-${threadId}`), + threadId, + projectId, + title: "Native conversation", + modelSelection: { instanceId: thread.providerInstanceId, model: "default" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: thread.createdAt, + }); + yield* directory.upsert({ + threadId, + provider: ProviderDriverKind.make(source), + providerInstanceId: thread.providerInstanceId, + status: "stopped", + resumeCursor: + source === "codex" + ? { threadId: thread.providerSessionId } + : { threadId, resume: thread.providerSessionId }, + }); + const before = yield* snapshots.getThreadDetailById(threadId); + const result = yield* importRecentAgentThreads({ projectId }).pipe( + Effect.provideService(AgentSessionScanner.AgentSessionScanner, { + scan: Effect.die("unused"), + recentThreads: () => Stream.succeed(makeThreadOutcome(thread)), + }), + ); + const importedId = ThreadId.make(`import:${source}:${thread.providerSessionId}`); + expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); + expect(yield* snapshots.getThreadDetailById(importedId)).toEqual(Option.none()); + expect(yield* directory.getBinding(importedId)).toEqual(Option.none()); + expect(yield* snapshots.getThreadDetailById(threadId)).toEqual(before); + }), + ); + } + it.effect("imports once after the real engine persists an old rejected receipt", () => Effect.gen(function* () { const engine = yield* OrchestrationEngine.OrchestrationEngineService; diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index bf9eb702ebe52..12bcb788439a2 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -21,6 +21,7 @@ import { normalizeProjectPathForComparison } from "@t3tools/shared/path"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -134,6 +135,26 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu completedSources.map((entry) => entry.source), ); const importedThreadIds = new Set(); + const nativeSessions = new Set(); + const bindings = yield* directory + .listBindings() + .pipe( + Effect.mapError((cause) => new AgentSessionScanError({ operation: "read-projects", cause })), + ); + for (const binding of bindings) { + if (binding.threadId.startsWith("import:") || !Predicate.isObject(binding.resumeCursor)) { + continue; + } + const sessionId = + binding.provider === "claudeAgent" + ? binding.resumeCursor.resume + : binding.provider === "codex" + ? binding.resumeCursor.threadId + : undefined; + if (typeof sessionId === "string" && binding.providerInstanceId !== undefined) { + nativeSessions.add(`${binding.providerInstanceId}\0${sessionId}`); + } + } let importedCount = 0; let skippedCount = 0; @@ -165,6 +186,10 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu return; } const thread = outcome.thread; + if (nativeSessions.has(`${thread.providerInstanceId}\0${thread.providerSessionId}`)) { + skippedCount += 1; + return; + } const threadId = ThreadId.make( `import:${thread.providerInstanceId}:${thread.providerSessionId}`, ); From 9c6ea6a466ecf6e16b3a2cd1d1144e4b02561b5d Mon Sep 17 00:00:00 2001 From: Guillermo Casanova Date: Wed, 9 Sep 2026 11:56:40 -0300 Subject: [PATCH 2/3] fix(server): check native sessions when reserving imports --- .../src/persistence/ProviderSessionRuntime.ts | 26 ++- .../src/project/AgentSessionImporter.test.ts | 150 +++++++++++------- .../src/project/AgentSessionImporter.ts | 3 +- .../Services/ProviderSessionDirectory.ts | 2 + 4 files changed, 118 insertions(+), 63 deletions(-) diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 1ec7adcd623ba..88f491a974946 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -67,6 +67,7 @@ export type RecordImportedTranscriptInput = typeof RecordImportedTranscriptInput export interface ProviderSessionRuntimeUpsertOptions { readonly onConflict?: "update" | "ignore"; + readonly unlessNativeSessionId?: string; } /** @@ -238,7 +239,9 @@ export const make = Effect.gen(function* () { }); const insertRuntimeRow = SqlSchema.void({ - Request: ProviderSessionRuntimeDbRowSchema, + Request: ProviderSessionRuntimeDbRowSchema.mapFields( + Struct.assign({ unlessNativeSessionId: Schema.NullOr(Schema.String) }), + ), execute: (runtime) => sql` INSERT INTO provider_session_runtime ( @@ -252,7 +255,7 @@ export const make = Effect.gen(function* () { resume_cursor_json, runtime_payload_json ) - VALUES ( + SELECT ${runtime.threadId}, ${runtime.providerName}, ${runtime.providerInstanceId}, @@ -266,6 +269,17 @@ export const make = Effect.gen(function* () { THEN json_remove(${runtime.runtimePayload}, '$.importedTranscripts') ELSE ${runtime.runtimePayload} END + WHERE ${runtime.unlessNativeSessionId} IS NULL OR NOT EXISTS ( + SELECT 1 FROM provider_session_runtime + WHERE thread_id NOT LIKE 'import:%' + AND provider_name = ${runtime.providerName} + AND COALESCE(provider_instance_id, provider_name) = ${runtime.providerInstanceId} + AND CASE WHEN json_valid(resume_cursor_json) THEN + json_extract(resume_cursor_json, CASE provider_name + WHEN 'claudeAgent' THEN '$.resume' + WHEN 'codex' THEN '$.threadId' + END) + END = ${runtime.unlessNativeSessionId} ) ON CONFLICT (thread_id) DO NOTHING `, @@ -369,7 +383,13 @@ export const make = Effect.gen(function* () { }); const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) => - (options?.onConflict === "ignore" ? insertRuntimeRow(runtime) : upsertRuntimeRow(runtime)).pipe( + (options?.onConflict === "ignore" + ? insertRuntimeRow({ + ...runtime, + unlessNativeSessionId: options.unlessNativeSessionId ?? null, + }) + : upsertRuntimeRow(runtime) + ).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProviderSessionRuntimeRepository.upsert:query", diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 4e93c3ab2aa19..75c9160d1d4f2 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -233,7 +233,10 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { upsert: (binding) => Effect.sync(() => void bindings.push(binding)), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, - getBinding: () => Effect.succeedNone, + getBinding: (threadId) => + Effect.succeed( + Option.fromUndefinedOr(bindings.find((binding) => binding.threadId === threadId)), + ), listThreadIds: () => Effect.die("unused"), listBindings: () => Effect.succeed([ @@ -592,64 +595,93 @@ const integrationLayer = Layer.mergeAll( it.layer(integrationLayer)("AgentSessionImporter integration", (it) => { for (const source of ["codex", "claudeAgent"] as const) { - it.effect(`does not import a ${source} session already owned by a native thread`, () => - Effect.gen(function* () { - const engine = yield* OrchestrationEngine.OrchestrationEngineService; - const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; - const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; - const projectId = ProjectId.make(`native-import-${source}`); - const threadId = ThreadId.make(`native-${source}`); - const thread = { - ...makeThread(source), - providerSessionId: - source === "codex" ? "native-codex-session" : "123e4567-e89b-42d3-a456-426614174001", - }; - yield* engine.dispatch({ - type: "project.create", - commandId: CommandId.make(`create-${projectId}`), - projectId, - title: "Native project", - workspaceRoot: `/tmp/${projectId}`, - defaultModelSelection: null, - createdAt: thread.createdAt, - }); - yield* engine.dispatch({ - type: "thread.create", - commandId: CommandId.make(`create-${threadId}`), - threadId, - projectId, - title: "Native conversation", - modelSelection: { instanceId: thread.providerInstanceId, model: "default" }, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - createdAt: thread.createdAt, - }); - yield* directory.upsert({ - threadId, - provider: ProviderDriverKind.make(source), - providerInstanceId: thread.providerInstanceId, - status: "stopped", - resumeCursor: - source === "codex" - ? { threadId: thread.providerSessionId } - : { threadId, resume: thread.providerSessionId }, - }); - const before = yield* snapshots.getThreadDetailById(threadId); - const result = yield* importRecentAgentThreads({ projectId }).pipe( - Effect.provideService(AgentSessionScanner.AgentSessionScanner, { - scan: Effect.die("unused"), - recentThreads: () => Stream.succeed(makeThreadOutcome(thread)), - }), - ); - const importedId = ThreadId.make(`import:${source}:${thread.providerSessionId}`); - expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); - expect(yield* snapshots.getThreadDetailById(importedId)).toEqual(Option.none()); - expect(yield* directory.getBinding(importedId)).toEqual(Option.none()); - expect(yield* snapshots.getThreadDetailById(threadId)).toEqual(before); - }), - ); + for (const timing of ["before scan", "before reservation"] as const) { + it.effect(`skips native ${source} sessions bound ${timing}`, () => + Effect.gen(function* () { + const engine = yield* OrchestrationEngine.OrchestrationEngineService; + const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const projectId = ProjectId.make(`native-import-${source}-${timing}`); + const threadId = ThreadId.make(`native-${source}-${timing}`); + const thread = { + ...makeThread(source), + providerSessionId: + source === "codex" + ? `native-codex-session-${timing}` + : timing === "before scan" + ? "123e4567-e89b-42d3-a456-426614174001" + : "123e4567-e89b-42d3-a456-426614174002", + }; + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make(`create-${projectId}`), + projectId, + title: "Native project", + workspaceRoot: `/tmp/${projectId}`, + defaultModelSelection: null, + createdAt: thread.createdAt, + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create-${threadId}`), + threadId, + projectId, + title: "Native conversation", + modelSelection: { instanceId: thread.providerInstanceId, model: "default" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: thread.createdAt, + }); + const bindNative = directory.upsert({ + threadId, + provider: ProviderDriverKind.make(source), + providerInstanceId: thread.providerInstanceId, + status: "stopped", + resumeCursor: + source === "codex" + ? { threadId: thread.providerSessionId } + : { threadId, resume: thread.providerSessionId }, + }); + if (timing === "before scan") yield* bindNative; + const before = yield* snapshots.getThreadDetailById(threadId); + const result = yield* importRecentAgentThreads({ projectId }).pipe( + Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, { + ...directory, + upsert: (binding, options) => + bindNative.pipe(Effect.andThen(directory.upsert(binding, options))), + }), + Effect.provideService(AgentSessionScanner.AgentSessionScanner, { + scan: Effect.die("unused"), + recentThreads: () => Stream.succeed(makeThreadOutcome(thread)), + }), + ); + const importedId = ThreadId.make(`import:${source}:${thread.providerSessionId}`); + expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); + expect(yield* snapshots.getThreadDetailById(importedId)).toEqual(Option.none()); + expect(yield* directory.getBinding(importedId)).toEqual(Option.none()); + expect(yield* snapshots.getThreadDetailById(threadId)).toEqual(before); + const otherInstance = ProviderInstanceId.make(`${source}-other`); + const otherThreadId = ThreadId.make( + `import:${otherInstance}:${thread.providerSessionId}`, + ); + yield* directory.upsert( + { + threadId: otherThreadId, + provider: ProviderDriverKind.make(source), + providerInstanceId: otherInstance, + resumeCursor: + source === "codex" + ? { threadId: thread.providerSessionId } + : { resume: thread.providerSessionId }, + }, + { onConflict: "ignore", unlessNativeSessionId: thread.providerSessionId }, + ); + expect(Option.isSome(yield* directory.getBinding(otherThreadId))).toBe(true); + }), + ); + } } it.effect("imports once after the real engine persists an old rejected receipt", () => diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 12bcb788439a2..30622828da209 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -262,8 +262,9 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu : { threadId, resume: thread.providerSessionId }, runtimePayload: { cwd: workspaceRoot }, }, - { onConflict: "ignore" }, + { onConflict: "ignore", unlessNativeSessionId: thread.providerSessionId }, ); + if (Option.isNone(yield* directory.getBinding(threadId))) return false; } if (Option.isNone(existingThread)) { diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index 1b5c47ec1d621..7799997a5f160 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -43,6 +43,8 @@ export type ProviderSessionDirectoryWriteError = export interface ProviderSessionDirectoryUpsertOptions { readonly onConflict?: "update" | "ignore"; + // For insert-ignore imports, reserve only if no native binding owns this session. + readonly unlessNativeSessionId?: string; } export interface ProviderSessionDirectoryShape { From f27bba3110dae7d0f638b11ed6c4afa13669ecb1 Mon Sep 17 00:00:00 2001 From: Guillermo Casanova Date: Fri, 11 Sep 2026 01:35:07 -0300 Subject: [PATCH 3/3] fix(server): recheck native ownership when retrying imports --- .../src/persistence/ProviderSessionRuntime.ts | 13 +- .../src/project/AgentSessionImporter.test.ts | 121 +++++++++++++++++- .../src/project/AgentSessionImporter.ts | 39 +++--- .../src/provider/Layers/CodexAdapter.test.ts | 2 +- .../provider/Layers/OpenCodeAdapter.test.ts | 2 +- .../provider/Layers/ProviderService.test.ts | 2 +- .../Layers/ProviderSessionDirectory.ts | 2 +- .../Services/ProviderSessionDirectory.ts | 5 +- apps/server/src/server.test.ts | 2 +- .../serverRuntimeStartup.reconcile.test.ts | 19 ++- 10 files changed, 164 insertions(+), 43 deletions(-) diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 88f491a974946..edf9e67960c93 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -85,7 +85,7 @@ export class ProviderSessionRuntimeRepository extends Context.Service< readonly upsert: ( runtime: ProviderSessionRuntime, options?: ProviderSessionRuntimeUpsertOptions, - ) => Effect.Effect; + ) => Effect.Effect; /** Record one source file without replacing the current session state. */ readonly recordImportedTranscript: ( @@ -238,10 +238,12 @@ export const make = Effect.gen(function* () { `, }); - const insertRuntimeRow = SqlSchema.void({ + // A no-op conflict update lets RETURNING report allowed reuse without changing the binding. + const insertRuntimeRow = SqlSchema.findOneOption({ Request: ProviderSessionRuntimeDbRowSchema.mapFields( Struct.assign({ unlessNativeSessionId: Schema.NullOr(Schema.String) }), ), + Result: Schema.Struct({ threadId: ThreadId }), execute: (runtime) => sql` INSERT INTO provider_session_runtime ( @@ -281,7 +283,8 @@ export const make = Effect.gen(function* () { END) END = ${runtime.unlessNativeSessionId} ) - ON CONFLICT (thread_id) DO NOTHING + ON CONFLICT (thread_id) DO UPDATE SET thread_id = provider_session_runtime.thread_id + RETURNING thread_id AS "threadId" `, }); @@ -387,8 +390,8 @@ export const make = Effect.gen(function* () { ? insertRuntimeRow({ ...runtime, unlessNativeSessionId: options.unlessNativeSessionId ?? null, - }) - : upsertRuntimeRow(runtime) + }).pipe(Effect.map(Option.isSome)) + : upsertRuntimeRow(runtime).pipe(Effect.as(true)) ).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 75c9160d1d4f2..254b859d64d3d 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -230,7 +230,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { latestSequence: Effect.succeed(0), }); const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ - upsert: (binding) => Effect.sync(() => void bindings.push(binding)), + upsert: (binding) => Effect.sync(() => bindings.push(binding)).pipe(Effect.as(true)), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, getBinding: (threadId) => @@ -422,8 +422,8 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { }), ); } - bindings.push(binding); - return Effect.void; + if (bindings.length === 0) bindings.push(binding); + return Effect.succeed(true); }, getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, @@ -684,6 +684,121 @@ it.layer(integrationLayer)("AgentSessionImporter integration", (it) => { } } + for (const source of ["codex", "claudeAgent"] as const) { + for (const owner of ["same instance", "other instance", "none"] as const) { + it.effect(`rechecks a reserved ${source} import with native owner in ${owner}`, () => + Effect.gen(function* () { + const engine = yield* OrchestrationEngine.OrchestrationEngineService; + const snapshots = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const projectId = ProjectId.make(`retry-native-${source}-${owner}`); + const nativeId = ThreadId.make(`native-${projectId}`); + const thread = { + ...makeThread(source), + providerSessionId: + source === "codex" + ? `retry-native-${owner}` + : owner === "same instance" + ? "123e4567-e89b-42d3-a456-426614174011" + : owner === "other instance" + ? "123e4567-e89b-42d3-a456-426614174012" + : "123e4567-e89b-42d3-a456-426614174013", + }; + const importedId = ThreadId.make(`import:${source}:${thread.providerSessionId}`); + yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make(`create-${projectId}`), + projectId, + title: "Retry native ownership", + workspaceRoot: `/tmp/${projectId}`, + defaultModelSelection: null, + createdAt: thread.createdAt, + }); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(`create-${nativeId}`), + threadId: nativeId, + projectId, + title: "Native conversation", + modelSelection: { instanceId: thread.providerInstanceId, model: "default" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: thread.createdAt, + }); + const scanner = AgentSessionScanner.AgentSessionScanner.of({ + scan: Effect.die("unused"), + recentThreads: () => Stream.succeed(makeThreadOutcome(thread)), + }); + const failed = yield* importRecentAgentThreads({ projectId }).pipe( + Effect.provideService(AgentSessionScanner.AgentSessionScanner, scanner), + Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { + ...engine, + dispatch: (command) => + command.type === "thread.create" + ? Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "Injected thread creation failure after reservation.", + }), + ) + : engine.dispatch(command), + }), + ); + expect(failed).toEqual({ importedCount: 0, skippedCount: 1 }); + expect(yield* snapshots.getThreadDetailById(importedId)).toEqual(Option.none()); + const reservation = yield* directory.getBinding(importedId); + expect(Option.getOrThrow(reservation).status).toBe("stopped"); + const nativeBefore = yield* snapshots.getThreadDetailById(nativeId); + + const result = yield* importRecentAgentThreads({ projectId }).pipe( + Effect.provideService(AgentSessionScanner.AgentSessionScanner, { + ...scanner, + recentThreads: () => + Stream.fromEffect( + Effect.gen(function* () { + if (owner !== "none") { + yield* directory.upsert({ + threadId: nativeId, + provider: ProviderDriverKind.make(source), + providerInstanceId: + owner === "same instance" + ? thread.providerInstanceId + : ProviderInstanceId.make(`${source}-other`), + status: "running", + resumeCursor: + source === "codex" + ? { threadId: thread.providerSessionId } + : { threadId: nativeId, resume: thread.providerSessionId }, + }); + } + return makeThreadOutcome(thread); + }).pipe(Effect.orDie), + ), + }), + ); + expect(yield* snapshots.getThreadDetailById(nativeId)).toEqual(nativeBefore); + if (owner === "same instance") { + expect(result).toEqual({ importedCount: 0, skippedCount: 1 }); + expect(yield* snapshots.getThreadDetailById(importedId)).toEqual(Option.none()); + expect(yield* directory.getBinding(importedId)).toEqual(reservation); + } else { + expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); + expect( + Option.getOrThrow(yield* snapshots.getThreadDetailById(importedId)).messages.map( + (message) => message.text, + ), + ).toEqual(thread.messages.map((message) => message.text)); + expect(Option.getOrThrow(yield* directory.getBinding(importedId)).resumeCursor).toEqual( + Option.getOrThrow(reservation).resumeCursor, + ); + } + }), + ); + } + } + it.effect("imports once after the real engine persists an old rejected receipt", () => Effect.gen(function* () { const engine = yield* OrchestrationEngine.OrchestrationEngineService; diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 30622828da209..28b3706c4a23b 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -245,27 +245,24 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu return yield* new AgentSessionThreadModifiedError({ threadId }); } - // Install the cursor before the thread becomes visible. A concurrent - // real session can replace it, while insert-ignore keeps this import - // from replacing that newer binding. - if (Option.isNone(existingBinding)) { - yield* directory.upsert( - { - threadId, - provider, - providerInstanceId: thread.providerInstanceId, - status: "stopped", - runtimeMode: DEFAULT_RUNTIME_MODE, - resumeCursor: - thread.source === "codex" - ? { threadId: thread.providerSessionId } - : { threadId, resume: thread.providerSessionId }, - runtimePayload: { cwd: workspaceRoot }, - }, - { onConflict: "ignore", unlessNativeSessionId: thread.providerSessionId }, - ); - if (Option.isNone(yield* directory.getBinding(threadId))) return false; - } + // Check native ownership even when retrying an old reservation, without + // replacing a binding that a real session may have updated. + const reserved = yield* directory.upsert( + { + threadId, + provider, + providerInstanceId: thread.providerInstanceId, + status: "stopped", + runtimeMode: DEFAULT_RUNTIME_MODE, + resumeCursor: + thread.source === "codex" + ? { threadId: thread.providerSessionId } + : { threadId, resume: thread.providerSessionId }, + runtimePayload: { cwd: workspaceRoot }, + }, + { onConflict: "ignore", unlessNativeSessionId: thread.providerSessionId }, + ); + if (!reserved) return false; if (Option.isNone(existingThread)) { yield* engine.dispatch({ diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 9aff859521fc1..893b79099fccd 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -221,7 +221,7 @@ function makeScopedRuntimeFactory(options?: { readonly failConstruction?: boolea } const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory, { - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")), diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 40d034089ae95..23d37d62086dc 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -575,7 +575,7 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { }; const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory, { - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die(new Error("ProviderSessionDirectory.getProvider is not used in test")), diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 7acf3fcb7b01c..e22da7ad3f3f0 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -5026,7 +5026,7 @@ const getBinding = vi.fn((threadId: ThreadId) => ); const boundedListing = makeProviderServiceLayer({ directory: { - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("ProviderService.listSessions does not use getProvider"), getBinding, diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 0e9de8fb868f0..152992de96484 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -123,7 +123,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () { issue: "providerInstanceId is required for provider session runtime bindings.", }); } - yield* repository + return yield* repository .upsert( { threadId: resolvedThreadId, diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index 7799997a5f160..002b3d8ab0242 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -43,15 +43,16 @@ export type ProviderSessionDirectoryWriteError = export interface ProviderSessionDirectoryUpsertOptions { readonly onConflict?: "update" | "ignore"; - // For insert-ignore imports, reserve only if no native binding owns this session. + // For insert-ignore imports, allow insertion or reuse only without a native owner. readonly unlessNativeSessionId?: string; } export interface ProviderSessionDirectoryShape { + // Returns false when native ownership blocks an import reservation. readonly upsert: ( binding: ProviderRuntimeBinding, options?: ProviderSessionDirectoryUpsertOptions, - ) => Effect.Effect; + ) => Effect.Effect; /** Record an imported file without changing the current provider session. */ readonly recordImportedTranscript: (input: { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 930e7e7bb369f..ee3b8ab523840 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -842,7 +842,7 @@ const buildAppUnderTest = (options?: { ...options?.layers?.antigravityInstallation, }), Layer.mock(ProviderSessionDirectory.ProviderSessionDirectory)({ - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), getBinding: () => Effect.succeedNone, listThreadIds: () => Effect.succeed([]), listBindings: () => Effect.succeed([]), diff --git a/apps/server/src/serverRuntimeStartup.reconcile.test.ts b/apps/server/src/serverRuntimeStartup.reconcile.test.ts index 050650b11515b..c9647969f6b7c 100644 --- a/apps/server/src/serverRuntimeStartup.reconcile.test.ts +++ b/apps/server/src/serverRuntimeStartup.reconcile.test.ts @@ -150,7 +150,7 @@ it.effect("marks active running sessions that have persisted resume state", () = }), ), ), - upsert: (binding) => Effect.sync(() => upserts.push(binding)), + upsert: (binding) => Effect.sync(() => upserts.push(binding)).pipe(Effect.as(true)), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), @@ -279,6 +279,7 @@ it.effect.each( Effect.flatMap((firstMarkerCleared) => firstMarkerCleared ? Deferred.succeed(continuationCleared, undefined) : Effect.void, ), + Effect.as(true), ), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), @@ -410,7 +411,7 @@ it.effect("does not continue archived or deleted marked sessions", () => { }, }); }, - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), @@ -464,7 +465,7 @@ it.effect("retries continuation preparation before settling a persistent failure continueAfterServerUpdate: thread.session.activeTurnId, }, }), - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), @@ -536,7 +537,7 @@ it.effect("reconciles multiple active and archived orphans but skips live sessio }), ), ), - upsert: (binding) => Effect.sync(() => upserts.push(binding)), + upsert: (binding) => Effect.sync(() => upserts.push(binding)).pipe(Effect.as(true)), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), @@ -652,7 +653,7 @@ it.effect("retries failed projections and continues after a persistent failure", threads: [transient, persistent, later], directory: { getBinding: () => Effect.succeedNone, - upsert: () => Effect.void, + upsert: () => Effect.succeed(true), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), listThreadIds: () => Effect.die("unused"), @@ -766,6 +767,7 @@ for (const scenario of [ upsert: (binding) => Effect.sync(() => { upserts.push(binding); + return true; }), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), @@ -838,8 +840,10 @@ for (const preparedStatus of [ upsert: (next: ProviderSessionDirectory.ProviderRuntimeBinding) => Effect.gen(function* () { binding = next; - if (binding.status !== "starting" || sends.length === 0) return; - yield* Deferred.succeed(cleared, undefined); + if (binding.status === "starting" && sends.length > 0) { + yield* Deferred.succeed(cleared, undefined); + } + return true; }), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"), @@ -944,6 +948,7 @@ it.effect("settles failed opt-in recovery without retrying the provider turn", ( upsert: (next) => Effect.sync(() => { binding = next; + return true; }), recordImportedTranscript: () => Effect.die("unused"), getProvider: () => Effect.die("unused"),