diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2673512edf10..f1ce61c73809 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; } /** @@ -235,7 +236,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 ( @@ -249,7 +252,7 @@ export const make = Effect.gen(function* () { resume_cursor_json, runtime_payload_json ) - VALUES ( + SELECT ${runtime.threadId}, ${runtime.providerName}, ${runtime.providerInstanceId}, @@ -263,6 +266,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 `, @@ -365,7 +379,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 38d6ad1d4331..f28b02b9d476 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -198,7 +198,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 = []; @@ -231,9 +231,21 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { upsert: (binding) => Effect.sync(() => void bindings.push(binding)), getProvider: () => Effect.die("unused"), recordImportedTranscript: () => Effect.void, - getBinding: () => Effect.succeed(Option.none()), + getBinding: (threadId) => + Effect.succeed( + Option.fromUndefinedOr(bindings.find((binding) => binding.threadId === threadId)), + ), 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({ @@ -338,7 +350,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({ @@ -416,7 +428,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(), @@ -457,7 +469,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.void, getBinding: () => Effect.succeed(Option.some(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"), @@ -512,7 +524,7 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { recordImportedTranscript: () => Effect.die("unused"), getBinding: () => Effect.succeed(Option.none()), listThreadIds: () => Effect.die("unused"), - listBindings: () => Effect.die("unused"), + listBindings: () => Effect.succeed([]), }); const result = yield* runImport({ @@ -580,6 +592,96 @@ const integrationLayer = Layer.mergeAll( ); it.layer(integrationLayer)("AgentSessionImporter integration", (it) => { + for (const source of ["codex", "claudeAgent"] as const) { + 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", () => 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 5ebb41a1bb54..e431097a2d34 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"; @@ -133,6 +134,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; @@ -164,6 +185,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}`, ); @@ -236,8 +261,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 9dbafd3e804e..cd4304713294 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 {