From d424e6a9135411bd0aa2a3c56e89b6564669d8f6 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Mon, 19 Oct 2026 23:23:01 -0500 Subject: [PATCH 1/9] fix(server): show the native history of imported ACP sessions An imported ACP session opened as an empty thread. The import created the thread with only the native session reference, and loading the session replayed its history only into the ACP adapter's own snapshot, which no server code read. After the import, open the session once, read that snapshot and write its messages into the thread as run-less history, the same events the Claude and Codex transcript import writes. Refs #22015 Co-Authored-By: Claude Opus 5.5 --- .../src/project/AgentSessionImporter.test.ts | 81 ++++++++++++++++++- .../src/project/AgentSessionImporter.ts | 66 ++++++++++++++- apps/server/src/ws.ts | 18 ++++- 3 files changed, 162 insertions(+), 3 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 4248e51b55977..66dbdede1a159 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -5,12 +5,14 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + MessageId, ThreadId, type ApplicationStoredEvent, type OrchestrationV2AppThread, type OrchestrationV2ProviderThread, } from "@t3tools/contracts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -240,7 +242,11 @@ const makeImporter = Effect.fn(function* ( ), ); const importer = Context.get(context, AgentSessionImporter.AgentSessionImporter); - return { run: importer.importRecentAgentThreads({ projectId }), published }; + return { + run: importer.importRecentAgentThreads({ projectId }), + importNativeHistory: importer.importNativeHistory, + published, + }; }); it.effect( @@ -400,3 +406,76 @@ it.effect( }); }).pipe(Effect.scoped, Effect.provide(persistence)), ); + +it.effect("copies the history a native session load replays into the imported thread", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const source = candidate(CODEX_MODEL_SELECTION, "native-history"); + const threadId = yield* seedOwner(source); + const importer = yield* makeImporter([]); + const opened: Array = []; + let closed = false; + const replayed = (role: "user" | "assistant", text: string) => ({ + createdBy: role === "user" ? ("user" as const) : ("agent" as const), + creationSource: "provider" as const, + id: MessageId.make(`replayed:${text}`), + threadId, + runId: null, + nodeId: null, + role, + text, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }); + const adapter = { + openSession: (input: ProviderAdapter.ProviderAdapterV2OpenSessionInput) => + Effect.acquireRelease( + Effect.sync(() => { + opened.push(input); + return { + readThreadSnapshot: ({ providerThread }) => + Effect.succeed({ + providerThread, + providerTurns: [], + runtimeRequests: [], + messages: [ + replayed("user", "Fix the build"), + replayed("assistant", "Fixed it"), + replayed("assistant", " "), + ], + }), + } as Pick; + }), + () => + Effect.sync(() => { + closed = true; + }), + ), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + + expect( + yield* importer.importNativeHistory({ + threadId, + adapter, + workspaceRoot: "/workspace/project", + }), + ).toBe(2); + expect(opened).toMatchObject([ + { + threadId, + initialNativeThreadId: source.source.providerSessionId, + runtimePolicy: { cwd: "/workspace/project" }, + }, + ]); + expect(closed).toBe(true); + const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); + expect( + page.items.map(({ item }) => [item.type, item.runId, "text" in item ? item.text : undefined]), + ).toEqual([ + ["user_message", null, "Fix the build"], + ["assistant_message", null, "Fixed it"], + ]); + }).pipe(Effect.scoped, Effect.provide(persistence)), +); diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 038b488635053..405adb746470f 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -33,6 +33,7 @@ import * as Stream from "effect/Stream"; import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts"; import * as AgentSessionScanner from "./AgentSessionScanner.ts"; @@ -404,7 +405,70 @@ const make = Effect.gen(function* () { return { importedCount, skippedCount } satisfies AgentSessionImportResult; }); - return { importRecentAgentThreads }; + /** + * A thread bound to an existing native session starts empty. Loading the + * session replays its history only into the adapter's own snapshot, so copy + * that snapshot into the thread as run-less history, as transcript imports do. + */ + const importNativeHistory = Effect.fn("importNativeHistoryV2")(function* (input: { + readonly threadId: ThreadId; + readonly adapter: ProviderAdapter.ProviderAdapterV2["Service"]; + readonly workspaceRoot: string; + }) { + const { thread, providerThreads } = yield* orchestrator.getThreadRecords(input.threadId, [ + "providerThreads", + ]); + const providerThread = providerThreads.find( + (candidate) => candidate.appThreadId === input.threadId && candidate.nativeThreadRef !== null, + ); + const nativeThreadId = providerThread?.nativeThreadRef?.nativeId; + if (providerThread === undefined || nativeThreadId == null) return 0; + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: providerThread.providerInstanceId, + threadId: input.threadId, + }); + const snapshot = yield* input.adapter + .openSession({ + threadId: input.threadId, + providerSessionId, + modelSelection: thread.modelSelection, + runtimePolicy: { + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + cwd: thread.worktreePath ?? input.workspaceRoot, + }, + initialNativeThreadId: nativeThreadId, + ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined + ? {} + : { + initialProviderItemIdentityVersion: providerThread.nativeMetadata.itemIdentityVersion, + }), + }) + .pipe( + Effect.flatMap((session) => session.readThreadSnapshot({ providerThread })), + Effect.scoped, + ); + const messages = snapshot.messages.flatMap((message) => + (message.role === "user" || message.role === "assistant") && message.text.trim() !== "" + ? [ + { + role: message.role, + text: message.text, + createdAt: DateTime.formatIso(message.createdAt), + }, + ] + : [], + ); + if (messages.length === 0) return 0; + yield* eventSink.write({ + events: messages.flatMap((message, index) => + messageEvents({ threadId: input.threadId, index, message }), + ), + }); + return messages.length; + }); + + return { importRecentAgentThreads, importNativeHistory }; }); type AgentSessionImporterShape = Effect.Success; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4781fcf78dde6..ccf5af744d6a1 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1488,7 +1488,7 @@ export const layerWsRpc = ( ) { return yield* acpRegistryRuntimeCoordinator.withSessionMutation( Effect.gen(function* () { - yield* acpRegistryProject(input.projectId); + const project = yield* acpRegistryProject(input.projectId); const { instance } = yield* acpSessionManager(input.instanceId); const providerSnapshot = yield* instance.snapshot.getSnapshot; if ( @@ -1571,6 +1571,22 @@ export const layerWsRpc = ( cause: launched.failure, }); } + // The thread is usable without its history, so a failed read only logs. + yield* agentSessionImporter + .importNativeHistory({ + threadId, + adapter: instance.orchestrationAdapter, + workspaceRoot: project.workspaceRoot, + }) + .pipe( + Effect.catch((cause) => + Effect.logWarning("Could not import ACP session history", { + instanceId: input.instanceId, + sessionId: input.sessionId, + cause, + }), + ), + ); return { threadId, imported: true } as const; }), ); From 526d7930d0d392bd27ae22493abec7f1d0aec8ae Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Tue, 20 Oct 2026 00:09:24 -0500 Subject: [PATCH 2/9] fix(server): bound the ACP session close and open history imports through the registry The ACP adapter's session finalizer waited for session/close with no deadline. Finalizers can't be interrupted, so an agent that never answered kept its process running and held the history import, and the ACP session lock, even after the import was interrupted. The close now gets 10 seconds, the bound the Stop path already uses, and the runtime closes after it. The history import now gets its adapter from the provider adapter registry, so the session open has the same sign-in admission as every other open. Refs #22015 Co-Authored-By: Claude Opus 5.5 --- apps/server/scripts/acp-mock-agent.ts | 3 + .../Adapters/AcpAdapterV2.test.ts | 61 +++++++++++++++++++ .../src/orchestration-v2/runtimeLayer.ts | 1 + .../src/project/AgentSessionImporter.test.ts | 20 ++++-- .../src/project/AgentSessionImporter.ts | 8 ++- apps/server/src/ws.ts | 6 +- packages/provider-acp/src/server/adapter.ts | 6 +- 7 files changed, 90 insertions(+), 15 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index c166340d89a8e..0318dbb9f4e28 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -696,6 +696,9 @@ const program = Effect.gen(function* () { yield* agent.handleCloseSession(() => Effect.gen(function* () { yield* requireAuthentication(); + if (process.env.T3_ACP_HANG_CLOSE_SESSION === "1") { + return yield* Effect.never; + } return {}; }), ); diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index b36c8de7fdf88..2571f67ff7b3d 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -4613,6 +4613,67 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("closes the ACP runtime when the agent never answers session/close", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const selfInvocation = yield* resolveSelfInvocation(); + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const protocolEvents = yield* Queue.bounded(256); + const makeRuntime = makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + protocolEvents, + environment: { T3_ACP_HANG_CLOSE_SESSION: "1" }, + }); + let runtimeClosed = false; + const instanceId = ProviderInstanceId.make("acp-test"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + makeRuntime: (runtimeInput) => + Effect.addFinalizer(() => + Effect.sync(() => { + runtimeClosed = true; + }), + ).pipe(Effect.andThen(makeRuntime(runtimeInput))), + }, + selfInvocation, + }); + const sessionScope = yield* Scope.make(); + yield* adapter + .openSession({ + threadId: ThreadId.make("thread-acp-unanswered-close"), + providerSessionId: ProviderSessionId.make("provider-session-acp-unanswered-close"), + modelSelection: { instanceId, model: "default" }, + runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }), + }) + .pipe(Effect.provideService(Scope.Scope, sessionScope)); + const closing = yield* Scope.close(sessionScope, Exit.void).pipe(Effect.forkDetach); + yield* Stream.fromQueue(protocolEvents).pipe( + Stream.filter( + (event) => event.direction === "outgoing" && rawProtocolMethod(event) === "session/close", + ), + Stream.runHead, + ); + yield* TestClock.adjust("10 seconds"); + const closed = yield* Fiber.await(closing).pipe( + Effect.timeoutOption("5 seconds"), + TestClock.withLive, + ); + assert.isTrue(Option.isSome(closed)); + assert.isTrue(runtimeClosed); + }).pipe(Effect.provide(layerTest)), + ); + // Oracle: Grok incident frames. Command completion must release the turn even // when the CLI keeps the prompt RPC and process alive. Command polls stay visible. it.live( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 1b54627595372..f60dac1986067 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -243,6 +243,7 @@ const layerAgentSessionImporterProvided = AgentSessionImporter.layer.pipe( layerEventSink, IdAllocator.layer, ProviderSessionRuntime.layer, + layerProviderAdapterRegistryProvided, ), ), ); diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 66dbdede1a159..ea5876bf2352a 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -27,6 +27,7 @@ import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as EventStore from "../orchestration-v2/EventStore.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; import { CLAUDE_MODEL_SELECTION, CODEX_MODEL_SELECTION, @@ -171,6 +172,7 @@ const seedOwner = Effect.fn(function* ( const makeImporter = Effect.fn(function* ( candidates: ReadonlyArray>, beforeTransaction: Effect.Effect = Effect.void, + getAdapter?: ProviderAdapterRegistry.ProviderAdapterRegistryV2Shape["get"], ) { const sql = yield* SqlClient.SqlClient; const events = yield* EventStore.EventStoreV2; @@ -237,6 +239,9 @@ const makeImporter = Effect.fn(function* ( ), ), }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)( + getAdapter === undefined ? {} : { get: getAdapter }, + ), ), ), ), @@ -412,7 +417,7 @@ it.effect("copies the history a native session load replays into the imported th const projections = yield* ProjectionStore.ProjectionStoreV2; const source = candidate(CODEX_MODEL_SELECTION, "native-history"); const threadId = yield* seedOwner(source); - const importer = yield* makeImporter([]); + const resolved: Array = []; const opened: Array = []; let closed = false; const replayed = (role: "user" | "assistant", text: string) => ({ @@ -454,14 +459,17 @@ it.effect("copies the history a native session load replays into the imported th }), ), } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter([], Effect.void, (instanceId) => + Effect.sync(() => { + resolved.push(instanceId); + return adapter; + }), + ); expect( - yield* importer.importNativeHistory({ - threadId, - adapter, - workspaceRoot: "/workspace/project", - }), + yield* importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }), ).toBe(2); + expect(resolved).toEqual([CODEX_MODEL_SELECTION.instanceId]); expect(opened).toMatchObject([ { threadId, diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 405adb746470f..85dc9842cad54 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -33,8 +33,8 @@ import * as Stream from "effect/Stream"; import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts"; import * as AgentSessionScanner from "./AgentSessionScanner.ts"; import * as ProjectService from "./ProjectService.ts"; @@ -173,6 +173,7 @@ const make = Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* ( input: AgentSessionImportInput, ) { @@ -412,7 +413,6 @@ const make = Effect.gen(function* () { */ const importNativeHistory = Effect.fn("importNativeHistoryV2")(function* (input: { readonly threadId: ThreadId; - readonly adapter: ProviderAdapter.ProviderAdapterV2["Service"]; readonly workspaceRoot: string; }) { const { thread, providerThreads } = yield* orchestrator.getThreadRecords(input.threadId, [ @@ -427,7 +427,9 @@ const make = Effect.gen(function* () { providerInstanceId: providerThread.providerInstanceId, threadId: input.threadId, }); - const snapshot = yield* input.adapter + // The registry admits this open against sign-in changes, like any other open. + const adapter = yield* registry.get(thread.modelSelection.instanceId); + const snapshot = yield* adapter .openSession({ threadId: input.threadId, providerSessionId, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ccf5af744d6a1..27ec375c903bc 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1573,11 +1573,7 @@ export const layerWsRpc = ( } // The thread is usable without its history, so a failed read only logs. yield* agentSessionImporter - .importNativeHistory({ - threadId, - adapter: instance.orchestrationAdapter, - workspaceRoot: project.workspaceRoot, - }) + .importNativeHistory({ threadId, workspaceRoot: project.workspaceRoot }) .pipe( Effect.catch((cause) => Effect.logWarning("Could not import ACP session history", { diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index bd2552ab42183..4f763328bd586 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7921,7 +7921,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( !restartRequired && !transportHadOutstandingResponses ) { - yield* runtime.closeSession().pipe(Effect.ignore); + // Finalizers can't be interrupted, so an agent that never + // answers would keep its process and this scope open. + yield* runtime + .closeSession() + .pipe(Effect.timeoutOption("10 seconds"), Effect.ignore); } }), ); From da62b612f36057c7f9a0b31490ceb40163360c8b Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Fri, 23 Oct 2026 21:49:14 -0500 Subject: [PATCH 3/9] fix(server): imported ACP history survives forks and failed reads Imported ACP history was written as run-less items without marking the thread, so forks and portable-fork context dropped it. The history write now sets historyOrigin "v1_import" in the same batch, as transcript imports do. That marker also records the import as done. A thread whose history read failed lists as importable again, and importing it again retries the read. Threads already marked, or with a run, are left alone. Co-Authored-By: Claude Opus 5.5 --- .../src/project/AgentSessionImporter.test.ts | 170 +++++++++++------- .../src/project/AgentSessionImporter.ts | 50 +++++- apps/server/src/ws.ts | 27 ++- 3 files changed, 169 insertions(+), 78 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index ea5876bf2352a..be16ec556e9ae 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -12,7 +12,7 @@ import { type OrchestrationV2ProviderThread, } from "@t3tools/contracts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -412,78 +412,110 @@ it.effect( }).pipe(Effect.scoped, Effect.provide(persistence)), ); -it.effect("copies the history a native session load replays into the imported thread", () => - Effect.gen(function* () { - const projections = yield* ProjectionStore.ProjectionStoreV2; - const source = candidate(CODEX_MODEL_SELECTION, "native-history"); - const threadId = yield* seedOwner(source); - const resolved: Array = []; - const opened: Array = []; - let closed = false; - const replayed = (role: "user" | "assistant", text: string) => ({ - createdBy: role === "user" ? ("user" as const) : ("agent" as const), - creationSource: "provider" as const, - id: MessageId.make(`replayed:${text}`), - threadId, - runId: null, - nodeId: null, - role, - text, - attachments: [], - streaming: false, - createdAt: now, - updatedAt: now, - }); - const adapter = { - openSession: (input: ProviderAdapter.ProviderAdapterV2OpenSessionInput) => - Effect.acquireRelease( - Effect.sync(() => { - opened.push(input); - return { - readThreadSnapshot: ({ providerThread }) => - Effect.succeed({ - providerThread, - providerTurns: [], - runtimeRequests: [], - messages: [ - replayed("user", "Fix the build"), - replayed("assistant", "Fixed it"), - replayed("assistant", " "), - ], - }), - } as Pick; - }), - () => +it.effect( + "copies native session history into the imported thread once, retrying a failed read", + () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const source = candidate(CODEX_MODEL_SELECTION, "native-history"); + const threadId = yield* seedOwner(source); + const resolved: Array = []; + const opened: Array = []; + let closed = false; + let failRead = true; + const replayed = (role: "user" | "assistant", text: string) => ({ + createdBy: role === "user" ? ("user" as const) : ("agent" as const), + creationSource: "provider" as const, + id: MessageId.make(`replayed:${text}`), + threadId, + runId: null, + nodeId: null, + role, + text, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }); + const adapter = { + openSession: (input: ProviderAdapter.ProviderAdapterV2OpenSessionInput) => + Effect.acquireRelease( Effect.sync(() => { - closed = true; + opened.push(input); + return { + readThreadSnapshot: ({ providerThread }) => + failRead + ? Effect.fail( + new ProviderAdapter.ProviderAdapterReadThreadSnapshotError({ + driver: providerThread.driver, + providerThreadId: providerThread.id, + }), + ) + : Effect.succeed({ + providerThread, + providerTurns: [], + runtimeRequests: [], + messages: [ + replayed("user", "Fix the build"), + replayed("assistant", "Fixed it"), + replayed("assistant", " "), + ], + }), + } as Pick; }), - ), - } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; - const importer = yield* makeImporter([], Effect.void, (instanceId) => - Effect.sync(() => { - resolved.push(instanceId); - return adapter; - }), - ); + () => + Effect.sync(() => { + closed = true; + }), + ), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter([], Effect.void, (instanceId) => + Effect.sync(() => { + resolved.push(instanceId); + return adapter; + }), + ); - expect( - yield* importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }), - ).toBe(2); - expect(resolved).toEqual([CODEX_MODEL_SELECTION.instanceId]); - expect(opened).toMatchObject([ - { + const importHistory = importer.importNativeHistory({ + threadId, + workspaceRoot: "/workspace/project", + }); + + expect((yield* Effect.flip(importHistory))._tag).toBe( + "ProviderAdapterReadThreadSnapshotError", + ); + expect(closed).toBe(true); + expect( + (yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin, + ).toBeUndefined(); + + failRead = false; + expect(yield* importHistory).toBe(2); + expect(resolved).toEqual([ + CODEX_MODEL_SELECTION.instanceId, + CODEX_MODEL_SELECTION.instanceId, + ]); + expect(opened[1]).toMatchObject({ threadId, initialNativeThreadId: source.source.providerSessionId, runtimePolicy: { cwd: "/workspace/project" }, - }, - ]); - expect(closed).toBe(true); - const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); - expect( - page.items.map(({ item }) => [item.type, item.runId, "text" in item ? item.text : undefined]), - ).toEqual([ - ["user_message", null, "Fix the build"], - ["assistant_message", null, "Fixed it"], - ]); - }).pipe(Effect.scoped, Effect.provide(persistence)), + }); + // Forks keep run-less history only for this origin. + expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe( + "v1_import", + ); + expect(yield* importHistory).toBe(0); + expect(opened).toHaveLength(2); + const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); + expect( + page.items.map(({ item }) => [ + item.type, + item.runId, + "text" in item ? item.text : undefined, + ]), + ).toEqual([ + ["user_message", null, "Fix the build"], + ["assistant_message", null, "Fixed it"], + ]); + }).pipe(Effect.scoped, Effect.provide(persistence)), ); diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 85dc9842cad54..a0d9cd279a62e 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -83,6 +83,18 @@ class AgentSessionThreadModifiedError extends Schema.TaggedError + isNativeHistoryPending({ + historyOrigin: records.thread.historyOrigin, + hasRun: records.runs.length > 0, + }) + ? records + : null, + ); + const records = yield* pendingThread; + if (records === null) return 0; + const { thread, providerThreads } = records; const providerThread = providerThreads.find( (candidate) => candidate.appThreadId === input.threadId && candidate.nativeThreadRef !== null, ); @@ -461,11 +486,22 @@ const make = Effect.gen(function* () { ] : [], ); - if (messages.length === 0) return 0; + // Loading the session can take a while, and the thread is usable meanwhile. + const current = yield* pendingThread; + if (current === null) return 0; yield* eventSink.write({ - events: messages.flatMap((message, index) => - messageEvents({ threadId: input.threadId, index, message }), - ), + events: [ + ...messages.flatMap((message, index) => + messageEvents({ threadId: input.threadId, index, message }), + ), + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${input.threadId}:native-history`), + type: "thread.metadata-updated", + threadId: input.threadId, + occurredAt: yield* DateTime.now, + payload: { ...current.thread, historyOrigin: "v1_import" }, + }, + ], }); return messages.length; }); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 27ec375c903bc..60eb064b810f9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1466,7 +1466,16 @@ export const layerWsRpc = ( return threadManagement.getThreadShell(threadId).pipe( Effect.map((thread) => ({ ...session, - importedThreadId: thread === null ? null : threadId, + // A thread still missing its history lists as importable, so + // importing again repairs it. + importedThreadId: + thread === null || + AgentSessionImporter.isNativeHistoryPending({ + historyOrigin: thread.historyOrigin, + hasRun: thread.latestRunId !== null, + }) + ? null + : threadId, })), Effect.mapError( (cause) => @@ -1515,7 +1524,21 @@ export const layerWsRpc = ( }), ), ); - if (existing !== null) return { threadId, imported: false } as const; + if (existing !== null) { + yield* agentSessionImporter + .importNativeHistory({ threadId, workspaceRoot: project.workspaceRoot }) + .pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "session_import_failed", + message: "Could not import the ACP session history.", + cause, + }), + ), + ); + return { threadId, imported: false } as const; + } const provider = (yield* providerRegistry.getProviders).find( (candidate) => candidate.instanceId === input.instanceId, From b14157fd178e5d2326c549a2a3c47fc084f77338 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Fri, 23 Oct 2026 22:21:29 -0500 Subject: [PATCH 4/9] fix(server): repair ACP history without losing thread edits or Delete guards The session list cleared importedThreadId for threads still missing their history, which also re-enabled Delete and dropped the Imported label. Keep importedThreadId meaning "a T3 thread exists" and report the repair state in a separate optional nativeHistoryPending flag. The web row offers Import for a pending thread while Delete stays disabled, and a repair now reports imported: true. Threads imported by the PR's earlier code hold the copied messages but no marker, and their deterministic event ids made every repair fail on the UNIQUE event_id. A pending thread that already has those items now gets the marker alone. The marker write replaces the whole thread row, so it now holds the per-thread command lock from its re-read through the write. A rename made while the session loads survives, and a run that starts meanwhile still stops the late import. The marker event also carries providerInstanceId. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.ts | 1 + .../src/project/AgentSessionImporter.test.ts | 234 ++++++++++++++++ .../src/project/AgentSessionImporter.ts | 57 ++-- apps/server/src/ws.test.ts | 257 +++++++++++++++++- apps/server/src/ws.ts | 24 +- ...pSessionManagementSection.browser.test.tsx | 27 ++ .../settings/AcpSessionManagementSection.tsx | 10 +- packages/contracts/src/acpRegistry.ts | 2 + 8 files changed, 577 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index f60dac1986067..9bdf634de5ce3 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -244,6 +244,7 @@ const layerAgentSessionImporterProvided = AgentSessionImporter.layer.pipe( IdAllocator.layer, ProviderSessionRuntime.layer, layerProviderAdapterRegistryProvided, + ThreadCommandExecutor.layer, ), ), ); diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index be16ec556e9ae..b8db0e09c8f32 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -6,12 +6,17 @@ import { ProviderDriverKind, ProviderInstanceId, MessageId, + RunId, ThreadId, + TurnItemId, type ApplicationStoredEvent, type OrchestrationV2AppThread, + type OrchestrationV2ConversationMessage, + type OrchestrationV2DomainEvent, type OrchestrationV2ProviderThread, } from "@t3tools/contracts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as KeyedLock from "@t3tools/shared/KeyedLock"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -28,6 +33,7 @@ import * as EventStore from "../orchestration-v2/EventStore.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts"; import { CLAUDE_MODEL_SELECTION, CODEX_MODEL_SELECTION, @@ -173,6 +179,7 @@ const makeImporter = Effect.fn(function* ( candidates: ReadonlyArray>, beforeTransaction: Effect.Effect = Effect.void, getAdapter?: ProviderAdapterRegistry.ProviderAdapterRegistryV2Shape["get"], + threadCommands?: KeyedLock.KeyedLock, ) { const sql = yield* SqlClient.SqlClient; const events = yield* EventStore.EventStoreV2; @@ -242,6 +249,10 @@ const makeImporter = Effect.fn(function* ( Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)( getAdapter === undefined ? {} : { get: getAdapter }, ), + Layer.succeed( + ThreadCommandExecutor.ThreadCommandExecutor, + threadCommands ?? (yield* KeyedLock.make()), + ), ), ), ), @@ -519,3 +530,226 @@ it.effect( ]); }).pipe(Effect.scoped, Effect.provide(persistence)), ); + +it.effect("marks history that an earlier build copied without its origin", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, "native-history-unmarked")); + // Earlier builds wrote these deterministic ids and no origin. + const suffix = "000000"; + const messageId = MessageId.make(`${threadId}:${suffix}`); + const message = { + createdBy: "user", + creationSource: "server", + id: messageId, + threadId, + runId: null, + nodeId: null, + role: "user", + text: "Fix the build", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + } satisfies OrchestrationV2ConversationMessage; + yield* sink.write({ + events: [ + { + id: EventId.make(`agent-session-import:v2:message:${threadId}:${suffix}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: message, + }, + { + id: EventId.make(`agent-session-import:v2:turn-item:${threadId}:${suffix}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`agent-session-import:v2:turn-item:${threadId}:${suffix}`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + createdBy: "user", + creationSource: "server", + type: "user_message", + messageId, + inputIntent: "turn_start", + text: "Fix the build", + attachments: [], + }, + }, + ], + }); + let opened = 0; + // Reading the session again would replay the same message. + const adapter = { + openSession: () => + Effect.sync(() => { + opened += 1; + return { + readThreadSnapshot: ({ providerThread }) => + Effect.succeed({ + providerThread, + providerTurns: [], + runtimeRequests: [], + messages: [{ ...message, creationSource: "provider" }], + }), + } as Pick; + }), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter([], Effect.void, () => Effect.succeed(adapter)); + + expect( + yield* importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }), + ).toBe(0); + expect(opened).toBe(0); + expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe( + "v1_import", + ); + const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); + expect(page.items.map(({ item }) => item.id)).toEqual([ + `agent-session-import:v2:turn-item:${threadId}:${suffix}`, + ]); + }).pipe(Effect.scoped, Effect.provide(persistence)), +); + +/** + * Runs a thread command that read the thread before the history load ended and + * writes once the importer is about to commit, as an unserialized write would. + */ +const importDuringCommand = Effect.fn(function* ( + name: string, + command: (thread: OrchestrationV2AppThread) => ReadonlyArray, +) { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, name)); + const lock = yield* KeyedLock.make(); + const loaded = yield* Deferred.make(); + const importerCommitting = yield* Deferred.make(); + const commandWritten = yield* Deferred.make(); + const committing = Deferred.succeed(importerCommitting, undefined); + const adapter = { + openSession: () => + Effect.succeed({ + readThreadSnapshot: ({ providerThread }) => + Deferred.await(loaded).pipe( + Effect.as({ + providerThread, + providerTurns: [], + runtimeRequests: [], + messages: [ + { + createdBy: "user" as const, + creationSource: "provider" as const, + id: MessageId.make(`replayed:${name}`), + threadId, + runId: null, + nodeId: null, + role: "user" as const, + text: "Fix the build", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + ], + }), + ), + } as Pick), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter( + [], + committing.pipe(Effect.andThen(Deferred.await(commandWritten))), + () => Effect.succeed(adapter), + { + ...lock, + withLock: (key, effect) => committing.pipe(Effect.andThen(lock.withLock(key, effect))), + }, + ); + const commanding = yield* Effect.forkScoped( + lock.withLock( + threadId, + Effect.gen(function* () { + const { thread } = yield* projections.getThreadRecords(threadId, []); + yield* Deferred.succeed(loaded, undefined); + yield* Deferred.await(importerCommitting); + yield* sink.write({ events: command(thread) }); + yield* Deferred.succeed(commandWritten, undefined); + }), + ), + ); + const imported = yield* importer.importNativeHistory({ + threadId, + workspaceRoot: "/workspace/project", + }); + yield* Fiber.join(commanding); + const records = yield* projections.getThreadRecords(threadId, ["runs"]); + const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); + return { imported, thread: records.thread, runs: records.runs, items: page.items }; +}); + +it.effect("keeps a thread edit made while the native history loads", () => + Effect.gen(function* () { + const result = yield* importDuringCommand("renamed-during-load", (thread) => [ + { + id: EventId.make(`${thread.id}:renamed`), + type: "thread.metadata-updated", + threadId: thread.id, + occurredAt: now, + payload: { ...thread, title: "Renamed" }, + }, + ]); + expect(result.imported).toBe(1); + expect(result.thread.title).toBe("Renamed"); + expect(result.thread.historyOrigin).toBe("v1_import"); + expect(result.items).toHaveLength(1); + }).pipe(Effect.scoped, Effect.provide(persistence)), +); + +it.effect("skips the native history when a run starts while it loads", () => + Effect.gen(function* () { + const result = yield* importDuringCommand("run-during-load", (thread) => [ + { + id: EventId.make(`${thread.id}:run`), + type: "run.updated", + threadId: thread.id, + occurredAt: now, + payload: { + id: RunId.make(`${thread.id}:run`), + threadId: thread.id, + ordinal: 1, + providerInstanceId: thread.providerInstanceId, + modelSelection: thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`${thread.id}:prompt`), + rootNodeId: null, + activeAttemptId: null, + status: "queued", + requestedAt: now, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }, + ]); + expect(result.imported).toBe(0); + expect(result.runs).toHaveLength(1); + expect(result.thread.historyOrigin).toBeUndefined(); + expect(result.items).toEqual([]); + }).pipe(Effect.scoped, Effect.provide(persistence)), +); diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index a0d9cd279a62e..e446491b8f5dd 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -35,6 +35,7 @@ import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts"; import * as ProviderSessionRuntime from "../persistence/ProviderSessionRuntime.ts"; import * as AgentSessionScanner from "./AgentSessionScanner.ts"; import * as ProjectService from "./ProjectService.ts"; @@ -186,6 +187,7 @@ const make = Effect.gen(function* () { const idAllocator = yield* IdAllocator.IdAllocatorV2; const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; + const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* ( input: AgentSessionImportInput, ) { @@ -431,7 +433,7 @@ const make = Effect.gen(function* () { readonly workspaceRoot: string; }) { const pendingThread = Effect.map( - orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs"]), + orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs", "turnItems"]), (records) => isNativeHistoryPending({ historyOrigin: records.thread.historyOrigin, @@ -440,9 +442,42 @@ const make = Effect.gen(function* () { ? records : null, ); + const importedItemPrefix = `${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:`; + // Thread commands write whole thread rows, so hold their lock from the + // re-read through the write to keep both this marker and their changes. + const commit = (messages: ReadonlyArray) => + threadCommands.withLock( + input.threadId, + Effect.gen(function* () { + const current = yield* pendingThread; + if (current === null) return 0; + // Earlier builds copied the history without the marker. Its ids are + // deterministic, so copying it again would collide; add the marker alone. + const copied = current.turnItems.some((item) => item.id.startsWith(importedItemPrefix)); + yield* eventSink.write({ + events: [ + ...(copied + ? [] + : messages.flatMap((message, index) => + messageEvents({ threadId: input.threadId, index, message }), + )), + { + id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${input.threadId}:native-history`), + type: "thread.metadata-updated", + threadId: input.threadId, + providerInstanceId: current.thread.providerInstanceId, + occurredAt: yield* DateTime.now, + payload: { ...current.thread, historyOrigin: "v1_import" }, + }, + ], + }); + return copied ? 0 : messages.length; + }), + ); const records = yield* pendingThread; if (records === null) return 0; - const { thread, providerThreads } = records; + const { thread, providerThreads, turnItems } = records; + if (turnItems.some((item) => item.id.startsWith(importedItemPrefix))) return yield* commit([]); const providerThread = providerThreads.find( (candidate) => candidate.appThreadId === input.threadId && candidate.nativeThreadRef !== null, ); @@ -487,23 +522,7 @@ const make = Effect.gen(function* () { : [], ); // Loading the session can take a while, and the thread is usable meanwhile. - const current = yield* pendingThread; - if (current === null) return 0; - yield* eventSink.write({ - events: [ - ...messages.flatMap((message, index) => - messageEvents({ threadId: input.threadId, index, message }), - ), - { - id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${input.threadId}:native-history`), - type: "thread.metadata-updated", - threadId: input.threadId, - occurredAt: yield* DateTime.now, - payload: { ...current.thread, historyOrigin: "v1_import" }, - }, - ], - }); - return messages.length; + return yield* commit(messages); }); return { importRecentAgentThreads, importNativeHistory }; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 641845f322ea6..22c05f70a3e93 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -2,15 +2,23 @@ import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretReso import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, AuthSessionId, EnvironmentId, + EventId, + MessageId, ProjectId, - ThreadId, ProviderDriverKind, ProviderInstanceId, + RunId, + ThreadId, WS_METHODS, WsServerRefreshProvidersRpc, + WsRpcGroup, WsSubscribeServerConfigRpc, + type OrchestrationV2AppThread, + type OrchestrationV2DomainEvent, type OrchestrationV2StoredEvent, type OrchestrationV2ThreadShell, type OrchestrationProjectShell, @@ -18,8 +26,10 @@ import { type ServerConfigStreamEvent, type ServerProvider, } from "@t3tools/contracts"; -import * as Deferred from "effect/Deferred"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; @@ -28,9 +38,11 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; import { RpcGroup, RpcMessage, RpcServer } from "effect/rpc"; +import * as RpcTest from "effect/rpc/RpcTest"; import * as SqlClient from "effect/sql/SqlClient"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; +import * as RpcAuthorization from "./auth/RpcAuthorization.ts"; import * as ServerBrowser from "./preview/ServerBrowser.ts"; import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; @@ -111,6 +123,7 @@ import * as ServerConfigModule from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ServerSettings from "./serverSettings.ts"; +import * as EventSink from "./orchestration-v2/EventSink.ts"; import * as EventStore from "./orchestration-v2/EventStore.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; @@ -979,3 +992,243 @@ it.effect("thread resume across compacted history keeps message order and final ), ), ); + +const acpDriver = ProviderDriverKind.make("acpRegistry"); +const acpInstanceId = ProviderInstanceId.make("acpRegistry_test"); +const acpProjectId = ProjectId.make("project:acp"); +const acpThreadId = (sessionId: string) => + IdAllocator.deriveThreadFromProviderThread({ + driver: acpDriver, + providerInstanceId: acpInstanceId, + nativeThreadId: sessionId, + }); +const acpSessionTags = [ + WS_METHODS.serverListAcpRegistrySessions, + WS_METHODS.serverImportAcpRegistrySession, +] as const; +const acpSessionRpcs = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + ( + tag, + ): tag is Exclude< + keyof typeof RpcAuthorization.RPC_REQUIRED_SCOPES, + (typeof acpSessionTags)[number] + > => !(acpSessionTags as ReadonlyArray).includes(tag), + ), +); +const acpStores = Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +const seedAcpThread = Effect.fn(function* ( + sessionId: string, + state: { readonly imported: boolean; readonly ran: boolean }, +) { + const sink = yield* EventSink.EventSinkV2; + const now = DateTime.makeUnsafe("2026-10-23T00:00:00.000Z"); + const threadId = acpThreadId(sessionId); + const thread: OrchestrationV2AppThread = { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: acpProjectId, + title: sessionId, + providerInstanceId: acpInstanceId, + modelSelection: { instanceId: acpInstanceId, model: "default" }, + runtimeMode: "approval-required", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + ...(state.imported ? { historyOrigin: "v1_import" as const } : {}), + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + deletedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + }; + const events: Array = [ + { + id: EventId.make(`${threadId}:created`), + type: "thread.created", + threadId, + occurredAt: now, + payload: thread, + }, + ]; + if (state.ran) { + events.push({ + id: EventId.make(`${threadId}:run`), + type: "run.updated", + threadId, + occurredAt: now, + payload: { + id: RunId.make(`${threadId}:run`), + threadId, + ordinal: 1, + providerInstanceId: acpInstanceId, + modelSelection: thread.modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`${threadId}:prompt`), + rootNodeId: null, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + } + yield* sink.write({ events }); + return threadId; +}); + +it.effect("ACP session RPCs list a thread missing its history and repair it on import", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const pending = yield* seedAcpThread("pending", { imported: false, ran: false }); + const complete = yield* seedAcpThread("complete", { imported: true, ran: false }); + const ran = yield* seedAcpThread("ran", { imported: false, ran: true }); + const failing = yield* seedAcpThread("failing", { imported: false, ran: false }); + const repaired: Array = []; + const session = (sessionId: string) => ({ + sessionId, + cwd: "/workspace/acp", + additionalDirectories: [], + title: null, + updatedAt: null, + importedThreadId: null, + }); + const instance = { + driverKind: acpDriver, + snapshot: { getSnapshot: Effect.succeed({ nativeSessions: { canLoad: true } }) }, + acpSessionManagement: { + listSessions: () => + Effect.succeed({ + sessions: ["new", "pending", "complete", "ran"].map(session), + nextCursor: null, + canLoad: true, + canResume: true, + canDelete: true, + }), + }, + } as unknown as ProviderInstance; + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.fresh(layerConfigSources), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({}), + Layer.mock(UsageLimitSources.UsageLimitSources)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + Layer.mock(ProjectService.ProjectService)({ + getById: () => + Effect.succeedSome({ + id: acpProjectId, + title: "ACP", + workspaceRoot: "/workspace/acp", + defaultModelSelection: null, + scripts: [], + createdAt: "2026-10-23T00:00:00.000Z", + updatedAt: "2026-10-23T00:00:00.000Z", + deletedAt: null, + }), + }), + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: () => Effect.succeed(instance), + }), + Layer.mock(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator)({ + withSessionMutation: (effect) => effect, + }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (threadId) => + projections + .getThreadShell(threadId) + .pipe( + Effect.mapError( + (cause) => new Orchestrator.OrchestratorProjectionError({ threadId, cause }), + ), + ), + }), + // Repairing an existing thread must not launch another one. + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(AgentSessionImporter.AgentSessionImporter)({ + importNativeHistory: ({ threadId }) => + threadId === failing + ? Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId, cause: "read" })) + : Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), + }), + ); + const client = yield* RpcTest.makeClient(acpSessionRpcs).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("acp-test"), + subject: "acp-test", + method: "browser-session-cookie", + scopes: [], + }, + {}, + {}, + yield* PreviewAutomationBroker.PreviewAutomationBroker, + yield* ServerBrowser.ServerBrowser, + ).pipe( + Layer.provide(services), + Layer.merge( + RpcAuthorization.layer([AuthOrchestrationReadScope, AuthOrchestrationOperateScope]), + ), + ), + ), + ); + + const listed = yield* client[WS_METHODS.serverListAcpRegistrySessions]({ + instanceId: acpInstanceId, + projectId: acpProjectId, + }); + assert.deepEqual( + listed.sessions.map(({ sessionId, importedThreadId, nativeHistoryPending }) => [ + sessionId, + importedThreadId, + nativeHistoryPending, + ]), + [ + ["new", null, undefined], + ["pending", pending, true], + ["complete", complete, undefined], + ["ran", ran, undefined], + ], + ); + + const importSession = (sessionId: string) => + client[WS_METHODS.serverImportAcpRegistrySession]({ + instanceId: acpInstanceId, + projectId: acpProjectId, + sessionId, + }); + assert.deepEqual(yield* importSession("pending"), { threadId: pending, imported: true }); + assert.deepEqual(repaired, [pending]); + const failure = yield* importSession("failing").pipe(Effect.flip); + assert.equal(failure._tag, "AcpRegistryOperationError"); + if (failure._tag === "AcpRegistryOperationError") { + assert.equal(failure.reason, "session_import_failed"); + } + }).pipe( + Effect.provide( + Layer.mergeAll( + EventSink.layer.pipe(Layer.provideMerge(acpStores)), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 60eb064b810f9..e5b6a0ad73d66 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1466,16 +1466,14 @@ export const layerWsRpc = ( return threadManagement.getThreadShell(threadId).pipe( Effect.map((thread) => ({ ...session, - // A thread still missing its history lists as importable, so - // importing again repairs it. - importedThreadId: - thread === null || - AgentSessionImporter.isNativeHistoryPending({ - historyOrigin: thread.historyOrigin, - hasRun: thread.latestRunId !== null, - }) - ? null - : threadId, + importedThreadId: thread === null ? null : threadId, + ...(thread !== null && + AgentSessionImporter.isNativeHistoryPending({ + historyOrigin: thread.historyOrigin, + hasRun: thread.latestRunId !== null, + }) + ? { nativeHistoryPending: true } + : {}), })), Effect.mapError( (cause) => @@ -1525,6 +1523,10 @@ export const layerWsRpc = ( ), ); if (existing !== null) { + const repairing = AgentSessionImporter.isNativeHistoryPending({ + historyOrigin: existing.historyOrigin, + hasRun: existing.latestRunId !== null, + }); yield* agentSessionImporter .importNativeHistory({ threadId, workspaceRoot: project.workspaceRoot }) .pipe( @@ -1537,7 +1539,7 @@ export const layerWsRpc = ( }), ), ); - return { threadId, imported: false } as const; + return { threadId, imported: repairing } as const; } const provider = (yield* providerRegistry.getProviders).find( diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx index a61814ae3efa6..998e1f6aaee34 100644 --- a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx +++ b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx @@ -224,3 +224,30 @@ it.each(["Save", "Disable"])("keeps other provider edits through a %s refetch", expect(input("other base URL").value).toBe("https://api.example.test/v1"); expect(input("other write-only headers JSON").value).toBe(""); }); + +it("offers to import a pending thread's history but not to delete its session", async () => { + setup(); + commands.list.mockResolvedValue({ + _tag: "Success", + value: { + sessions: [ + { + ...session, + importedThreadId: ThreadId.make("thread-imported"), + nativeHistoryPending: true, + }, + ], + nextCursor: null, + canLoad: true, + canResume: true, + canDelete: true, + }, + }); + await using screen = await renderSection(); + await screen.click("List sessions"); + expect(screen.button("Delete").disabled).toBe(true); + await screen.click("Import"); + expect(commands.import).toHaveBeenCalledOnce(); + expect(screen.button("Imported").disabled).toBe(true); + expect(screen.button("Delete").disabled).toBe(true); +}); diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.tsx index 675294497b118..c50a81d8ade48 100644 --- a/apps/web/src/components/settings/AcpSessionManagementSection.tsx +++ b/apps/web/src/components/settings/AcpSessionManagementSection.tsx @@ -34,6 +34,10 @@ interface AcpSessionProject { readonly workspaceRoot: string; } +function needsImport(session: AcpRegistrySession) { + return session.importedThreadId === null || session.nativeHistoryPending === true; +} + function reportFailure(title: string, result: AtomCommandResult) { if (result._tag !== "Failure" || isAtomCommandInterrupted(result)) return; const error = squashAtomCommandFailure(result); @@ -173,7 +177,7 @@ export function AcpSessionManagementSection(props: { setSessions((current) => current.map((candidate) => candidate.sessionId === session.sessionId - ? { ...candidate, importedThreadId: result.value.threadId } + ? { ...candidate, importedThreadId: result.value.threadId, nativeHistoryPending: false } : candidate, ), ); @@ -409,12 +413,12 @@ export function AcpSessionManagementSection(props: { disabled={ !canOperate || !canImport || - session.importedThreadId !== null || + !needsImport(session) || importingSessionId !== null } onClick={() => void importNativeSession(session)} > - {session.importedThreadId !== null + {!needsImport(session) ? "Imported" : importingSessionId === session.sessionId ? "Importing" diff --git a/packages/contracts/src/acpRegistry.ts b/packages/contracts/src/acpRegistry.ts index 8d4ffb4852d9e..a269664fbefad 100644 --- a/packages/contracts/src/acpRegistry.ts +++ b/packages/contracts/src/acpRegistry.ts @@ -149,6 +149,8 @@ export const AcpRegistrySession = Schema.Struct({ title: Schema.NullOr(Schema.String.check(Schema.isMaxLength(1_024))), updatedAt: Schema.NullOr(Schema.String.check(Schema.isMaxLength(128))), importedThreadId: Schema.NullOr(ThreadId), + /** The imported thread still lacks the session's history, so importing again copies it. */ + nativeHistoryPending: Schema.optionalKey(Schema.Boolean), }); export type AcpRegistrySession = typeof AcpRegistrySession.Type; From a02c74a5cd8af49bae99ecb19311ad946cf0d5eb Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 31 Oct 2026 22:32:07 +0000 Subject: [PATCH 5/9] refactor(server): drop the pre-marker ACP history branch no release wrote Only this PR's own unmerged commits wrote native history without the v1_import marker, so the collision guard, its turn-item read and its test covered state that no release can contain. Co-Authored-By: Claude Opus 5.5 --- .../src/project/AgentSessionImporter.test.ts | 97 ------------------- .../src/project/AgentSessionImporter.ts | 19 ++-- 2 files changed, 6 insertions(+), 110 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index b8db0e09c8f32..32b0e99b371e4 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -8,10 +8,8 @@ import { MessageId, RunId, ThreadId, - TurnItemId, type ApplicationStoredEvent, type OrchestrationV2AppThread, - type OrchestrationV2ConversationMessage, type OrchestrationV2DomainEvent, type OrchestrationV2ProviderThread, } from "@t3tools/contracts"; @@ -531,101 +529,6 @@ it.effect( }).pipe(Effect.scoped, Effect.provide(persistence)), ); -it.effect("marks history that an earlier build copied without its origin", () => - Effect.gen(function* () { - const projections = yield* ProjectionStore.ProjectionStoreV2; - const sink = yield* EventSink.EventSinkV2; - const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, "native-history-unmarked")); - // Earlier builds wrote these deterministic ids and no origin. - const suffix = "000000"; - const messageId = MessageId.make(`${threadId}:${suffix}`); - const message = { - createdBy: "user", - creationSource: "server", - id: messageId, - threadId, - runId: null, - nodeId: null, - role: "user", - text: "Fix the build", - attachments: [], - streaming: false, - createdAt: now, - updatedAt: now, - } satisfies OrchestrationV2ConversationMessage; - yield* sink.write({ - events: [ - { - id: EventId.make(`agent-session-import:v2:message:${threadId}:${suffix}`), - type: "message.updated", - threadId, - occurredAt: now, - payload: message, - }, - { - id: EventId.make(`agent-session-import:v2:turn-item:${threadId}:${suffix}`), - type: "turn-item.updated", - threadId, - occurredAt: now, - payload: { - id: TurnItemId.make(`agent-session-import:v2:turn-item:${threadId}:${suffix}`), - threadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - createdBy: "user", - creationSource: "server", - type: "user_message", - messageId, - inputIntent: "turn_start", - text: "Fix the build", - attachments: [], - }, - }, - ], - }); - let opened = 0; - // Reading the session again would replay the same message. - const adapter = { - openSession: () => - Effect.sync(() => { - opened += 1; - return { - readThreadSnapshot: ({ providerThread }) => - Effect.succeed({ - providerThread, - providerTurns: [], - runtimeRequests: [], - messages: [{ ...message, creationSource: "provider" }], - }), - } as Pick; - }), - } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; - const importer = yield* makeImporter([], Effect.void, () => Effect.succeed(adapter)); - - expect( - yield* importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }), - ).toBe(0); - expect(opened).toBe(0); - expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe( - "v1_import", - ); - const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); - expect(page.items.map(({ item }) => item.id)).toEqual([ - `agent-session-import:v2:turn-item:${threadId}:${suffix}`, - ]); - }).pipe(Effect.scoped, Effect.provide(persistence)), -); - /** * Runs a thread command that read the thread before the history load ended and * writes once the importer is about to commit, as an unserialized write would. diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index e446491b8f5dd..08df1a1f442e4 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -433,7 +433,7 @@ const make = Effect.gen(function* () { readonly workspaceRoot: string; }) { const pendingThread = Effect.map( - orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs", "turnItems"]), + orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs"]), (records) => isNativeHistoryPending({ historyOrigin: records.thread.historyOrigin, @@ -442,7 +442,6 @@ const make = Effect.gen(function* () { ? records : null, ); - const importedItemPrefix = `${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:`; // Thread commands write whole thread rows, so hold their lock from the // re-read through the write to keep both this marker and their changes. const commit = (messages: ReadonlyArray) => @@ -451,16 +450,11 @@ const make = Effect.gen(function* () { Effect.gen(function* () { const current = yield* pendingThread; if (current === null) return 0; - // Earlier builds copied the history without the marker. Its ids are - // deterministic, so copying it again would collide; add the marker alone. - const copied = current.turnItems.some((item) => item.id.startsWith(importedItemPrefix)); yield* eventSink.write({ events: [ - ...(copied - ? [] - : messages.flatMap((message, index) => - messageEvents({ threadId: input.threadId, index, message }), - )), + ...messages.flatMap((message, index) => + messageEvents({ threadId: input.threadId, index, message }), + ), { id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${input.threadId}:native-history`), type: "thread.metadata-updated", @@ -471,13 +465,12 @@ const make = Effect.gen(function* () { }, ], }); - return copied ? 0 : messages.length; + return messages.length; }), ); const records = yield* pendingThread; if (records === null) return 0; - const { thread, providerThreads, turnItems } = records; - if (turnItems.some((item) => item.id.startsWith(importedItemPrefix))) return yield* commit([]); + const { thread, providerThreads } = records; const providerThread = providerThreads.find( (candidate) => candidate.appThreadId === input.threadId && candidate.nativeThreadRef !== null, ); From 077759db16dd86772491f31b6d0508ec80034334 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 31 Oct 2026 23:16:22 +0000 Subject: [PATCH 6/9] fix(server): fit the ACP history import to current main Main renamed the provider adapter registry layer (#26238), changed layerWsRpc to take four arguments and need ScratchThreadPromotion (#26767), and caps comment blocks at four lines (#21514). The importer layer, the ACP session RPC test and the importNativeHistory docblock follow suit. The pending-history browser test now asserts through expect.element, as AGENTS.md asks of browser tests. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/runtimeLayer.ts | 2 +- apps/server/src/project/AgentSessionImporter.ts | 10 ++++------ apps/server/src/ws.test.ts | 4 ++-- .../AcpSessionManagementSection.browser.test.tsx | 6 +++--- 4 files changed, 10 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 9bdf634de5ce3..11c134e4d79da 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -243,7 +243,7 @@ const layerAgentSessionImporterProvided = AgentSessionImporter.layer.pipe( layerEventSink, IdAllocator.layer, ProviderSessionRuntime.layer, - layerProviderAdapterRegistryProvided, + ProviderAdapterRegistry.layerFromProviderInstanceRegistry, ThreadCommandExecutor.layer, ), ), diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 08df1a1f442e4..5ffc71f1f9361 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -421,12 +421,10 @@ const make = Effect.gen(function* () { }); /** - * A thread bound to an existing native session starts empty. Loading the - * session replays its history only into the adapter's own snapshot, so copy - * that snapshot into the thread as run-less history, as transcript imports do. - * The `v1_import` origin it sets is what forks read to keep that history, and - * marks the import done, so a call after a failed read repairs the thread and - * any later call does nothing. + * Loading a native session replays its history only into the adapter's own + * snapshot, so copy that snapshot into the thread as run-less history. The + * `v1_import` origin keeps it in forks and marks the import done, so a call + * after a failed read repairs the thread and any later call does nothing. */ const importNativeHistory = Effect.fn("importNativeHistoryV2")(function* (input: { readonly threadId: ThreadId; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 22c05f70a3e93..e997d2d97f506 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1128,6 +1128,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), Layer.mock(DirectEndpoints.DirectEndpoints)({}), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), Layer.mock(ServerEnvironment.ServerEnvironment)({}), Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), Layer.mock(ProjectService.ProjectService)({ @@ -1178,9 +1179,8 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i scopes: [], }, {}, - {}, - yield* PreviewAutomationBroker.PreviewAutomationBroker, yield* ServerBrowser.ServerBrowser, + () => Effect.void, ).pipe( Layer.provide(services), Layer.merge( diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx index 998e1f6aaee34..b2779eebe02df 100644 --- a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx +++ b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx @@ -245,9 +245,9 @@ it("offers to import a pending thread's history but not to delete its session", }); await using screen = await renderSection(); await screen.click("List sessions"); - expect(screen.button("Delete").disabled).toBe(true); + await expect.element(screen.button("Delete")).toBeDisabled(); await screen.click("Import"); expect(commands.import).toHaveBeenCalledOnce(); - expect(screen.button("Imported").disabled).toBe(true); - expect(screen.button("Delete").disabled).toBe(true); + await expect.element(screen.button("Imported")).toBeDisabled(); + await expect.element(screen.button("Delete")).toBeDisabled(); }); From 62458d66b8386e6e39edfd631cca7808720e8311 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 01:00:21 +0000 Subject: [PATCH 7/9] fix(server): an ACP agent that never answers can't hold the import lock An agent that supports only session/resume gets no deadline on that request or on initialize, unlike session/load's 90 seconds, and ACP imports read the history while holding the session-mutation lock. The history read now gives up after two minutes and closes the session. A repair then reports session_import_failed, and a new import keeps its thread and logs the failure, as for any failed read. Co-Authored-By: Claude Opus 5.5 --- .../src/project/AgentSessionImporter.test.ts | 30 +++++++++++++++++++ .../src/project/AgentSessionImporter.ts | 5 ++++ 2 files changed, 35 insertions(+) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 32b0e99b371e4..4f0bba2f979c8 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -25,6 +25,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as EventStore from "../orchestration-v2/EventStore.ts"; @@ -529,6 +530,35 @@ it.effect( }).pipe(Effect.scoped, Effect.provide(persistence)), ); +it.effect("gives up on a native session that never answers and closes it", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, "native-history-hang")); + let closed = false; + // An agent that only resumes sessions and never answers session/resume. + const adapter = { + openSession: () => + Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }), + ).pipe(Effect.andThen(Effect.never)), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter([], Effect.void, () => Effect.succeed(adapter)); + + const importing = yield* importer + .importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }) + .pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("2 minutes"); + + expect((yield* Fiber.join(importing))._tag).toBe("TimeoutError"); + expect(closed).toBe(true); + expect( + (yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin, + ).toBeUndefined(); + }).pipe(Effect.scoped, Effect.provide(persistence)), +); + /** * Runs a thread command that read the thread before the history load ended and * writes once the importer is about to commit, as an unserialized write would. diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 5ffc71f1f9361..4dc39d7ad2a32 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -84,6 +84,10 @@ class AgentSessionThreadModifiedError extends Schema.TaggedError session.readThreadSnapshot({ providerThread })), + Effect.timeout(NATIVE_HISTORY_READ_TIMEOUT), Effect.scoped, ); const messages = snapshot.messages.flatMap((message) => From 77ab499bc324531923a8e31d14d4a2a8f5b109b6 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 1 Nov 2026 01:17:40 +0000 Subject: [PATCH 8/9] test(server): a hung native session releases the ACP import lock The importer test now runs the import inside the real ACP session-mutation lock. When the session never answers, the import times out, the session closes, a waiting import or delete gets the lock, and a later import repairs the thread. The ACP session RPC test checks that the timeout reaches the client as session_import_failed. Co-Authored-By: Claude Opus 5.5 --- .../src/project/AgentSessionImporter.test.ts | 70 ++++++++++++++++--- apps/server/src/ws.test.ts | 16 +++-- 2 files changed, 70 insertions(+), 16 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 4f0bba2f979c8..263359c3e7090 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -15,6 +15,7 @@ import { } from "@t3tools/contracts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as KeyedLock from "@t3tools/shared/KeyedLock"; +import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -530,33 +531,80 @@ it.effect( }).pipe(Effect.scoped, Effect.provide(persistence)), ); -it.effect("gives up on a native session that never answers and closes it", () => +it.effect("releases the ACP import lock when a native session never answers", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; + const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator; const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, "native-history-hang")); + const opened = yield* Deferred.make(); let closed = false; - // An agent that only resumes sessions and never answers session/resume. + // An agent that only resumes sessions and never answers its first session/resume. const adapter = { openSession: () => - Effect.addFinalizer(() => - Effect.sync(() => { - closed = true; - }), - ).pipe(Effect.andThen(Effect.never)), + Effect.suspend(() => { + if (!closed) { + return Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }), + ).pipe( + Effect.andThen(Deferred.succeed(opened, undefined)), + Effect.andThen(Effect.never), + ); + } + return Effect.succeed({ + readThreadSnapshot: ({ providerThread }) => + Effect.succeed({ + providerThread, + providerTurns: [], + runtimeRequests: [], + messages: [ + { + createdBy: "user", + creationSource: "provider", + id: MessageId.make("replayed:hang"), + threadId, + runId: null, + nodeId: null, + role: "user", + text: "Fix the build", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + ], + }), + } as Pick); + }), } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; const importer = yield* makeImporter([], Effect.void, () => Effect.succeed(adapter)); + const importHistory = coordinator.withSessionMutation( + importer.importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }), + ); - const importing = yield* importer - .importNativeHistory({ threadId, workspaceRoot: "/workspace/project" }) - .pipe(Effect.flip, Effect.forkChild); + const importing = yield* importHistory.pipe(Effect.flip, Effect.forkChild); + yield* Deferred.await(opened); + // Any other ACP import or delete waits for the same lock. + const next = yield* coordinator + .withSessionMutation(Effect.succeed("next")) + .pipe(Effect.forkChild); yield* TestClock.adjust("2 minutes"); expect((yield* Fiber.join(importing))._tag).toBe("TimeoutError"); expect(closed).toBe(true); + expect(yield* Fiber.join(next)).toBe("next"); expect( (yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin, ).toBeUndefined(); - }).pipe(Effect.scoped, Effect.provide(persistence)), + expect(yield* importHistory).toBe(1); + expect((yield* projections.getThreadRecords(threadId, [])).thread.historyOrigin).toBe( + "v1_import", + ); + }).pipe( + Effect.scoped, + Effect.provide(Layer.merge(persistence, AcpRegistryRuntimeCoordinator.layer)), + ), ); /** diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index e997d2d97f506..a3de35b24f218 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -29,6 +29,7 @@ import { import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import * as DateTime from "effect/DateTime"; +import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -1096,6 +1097,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i const complete = yield* seedAcpThread("complete", { imported: true, ran: false }); const ran = yield* seedAcpThread("ran", { imported: false, ran: true }); const failing = yield* seedAcpThread("failing", { imported: false, ran: false }); + const timingOut = yield* seedAcpThread("timing-out", { imported: false, ran: false }); const repaired: Array = []; const session = (sessionId: string) => ({ sessionId, @@ -1166,7 +1168,9 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i importNativeHistory: ({ threadId }) => threadId === failing ? Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId, cause: "read" })) - : Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), + : threadId === timingOut + ? Effect.fail(new Cause.TimeoutError()) + : Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), }), ); const client = yield* RpcTest.makeClient(acpSessionRpcs).pipe( @@ -1216,10 +1220,12 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i }); assert.deepEqual(yield* importSession("pending"), { threadId: pending, imported: true }); assert.deepEqual(repaired, [pending]); - const failure = yield* importSession("failing").pipe(Effect.flip); - assert.equal(failure._tag, "AcpRegistryOperationError"); - if (failure._tag === "AcpRegistryOperationError") { - assert.equal(failure.reason, "session_import_failed"); + for (const sessionId of ["failing", "timing-out"]) { + const failure = yield* importSession(sessionId).pipe(Effect.flip); + assert.equal(failure._tag, "AcpRegistryOperationError"); + if (failure._tag === "AcpRegistryOperationError") { + assert.equal(failure.reason, "session_import_failed"); + } } }).pipe( Effect.provide( From 3735605596ce3402b6a13cb31ae4657cfe63c2b9 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 2 Nov 2026 01:39:41 +0000 Subject: [PATCH 9/9] fix(server): avoid replaying imported ACP history into its native session --- .../src/orchestration-v2/Orchestrator.migration.test.ts | 8 ++++++++ apps/server/src/orchestration-v2/Orchestrator.ts | 9 ++++++++- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.migration.test.ts b/apps/server/src/orchestration-v2/Orchestrator.migration.test.ts index 911c1b8de7df5..7f2a3cb3a753a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.migration.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.migration.test.ts @@ -10,6 +10,14 @@ it("reissues imported context until a V2 run completes", () => { legacyImportItemCount: 2, }), ); + assert.isFalse( + shouldPrepareLegacyImportHandoff({ + historyOrigin: "v1_import", + hasCompletedRun: false, + legacyImportItemCount: 2, + hasNativeHistory: true, + }), + ); assert.isFalse( shouldPrepareLegacyImportHandoff({ historyOrigin: "v1_import", diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1fed13e08ad0b..29eae7dcd4ab8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -956,12 +956,16 @@ function latestContextTransfer( } export function shouldPrepareLegacyImportHandoff(input: { + readonly hasNativeHistory?: boolean; readonly hasCompletedRun: boolean; readonly historyOrigin: OrchestrationV2AppThread["historyOrigin"]; readonly legacyImportItemCount: number; }): boolean { return ( - input.historyOrigin === "v1_import" && !input.hasCompletedRun && input.legacyImportItemCount > 0 + input.historyOrigin === "v1_import" && + !input.hasNativeHistory && + !input.hasCompletedRun && + input.legacyImportItemCount > 0 ); } @@ -6248,6 +6252,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio nativeThreadId: `pending:${runId}`, }); const legacyImportHandoff = shouldPrepareLegacyImportHandoff({ + hasNativeHistory: + activeProviderThread?.driver === "acpRegistry" && + activeProviderThread.nativeThreadRef !== null, historyOrigin: projection.thread.historyOrigin, hasCompletedRun: latestCompletedRun !== undefined, legacyImportItemCount: legacyImportItems.length,