fix(server): skip imports of native agent sessions

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",

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<OrchestrationCommand> = [];
         const bindings: Array<ProviderSessionDirectory.ProviderRuntimeBinding> = [];
@@ -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;

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<ThreadId>();
+  const nativeSessions = new Set<string>();
+  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)) {

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 {