From d424e6a9135411bd0aa2a3c56e89b6564669d8f6 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Mon, 19 Oct 2026 23:23:01 -0500 Subject: [PATCH 01/15] 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 02/15] 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 03/15] 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 04/15] 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 05/15] 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 06/15] 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 07/15] 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 08/15] 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 09/15] 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, From fa34498cb2919665d91a6cf1820f00f994876b82 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 10:13:37 +0000 Subject: [PATCH 10/15] refactor(server): move the ACP session import into a service The serverImportAcpRegistrySession handler ran the whole import itself: project and instance lookup, the existing-thread repair, the thread launch and the native history import. It now calls AcpRegistrySessions.importSession, which holds that sequence unchanged. The project and session-manager lookups move with it, and the other ACP session handlers reuse them. Co-Authored-By: Claude Opus 5.5 --- .../src/provider/AcpRegistrySessions.ts | 229 ++++++++++++++++++ apps/server/src/ws.test.ts | 5 +- apps/server/src/ws.ts | 217 ++--------------- 3 files changed, 254 insertions(+), 197 deletions(-) create mode 100644 apps/server/src/provider/AcpRegistrySessions.ts diff --git a/apps/server/src/provider/AcpRegistrySessions.ts b/apps/server/src/provider/AcpRegistrySessions.ts new file mode 100644 index 0000000000000..57fb4c168e34c --- /dev/null +++ b/apps/server/src/provider/AcpRegistrySessions.ts @@ -0,0 +1,229 @@ +import { + AcpRegistryOperationError, + CommandId, + type AcpRegistryImportSessionInput, + type AcpRegistryImportSessionResult, + type Project, + type ProjectId, + type ProviderDriverKind, + type ProviderInstanceId, +} from "@t3tools/contracts"; +import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Result from "effect/Result"; + +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as AgentSessionImporter from "../project/AgentSessionImporter.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ServerRuntimeStartup from "../serverRuntimeStartup.ts"; +import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts"; +import * as ProviderRegistry from "./ProviderRegistry.ts"; + +export const importedThreadId = (input: { + readonly driver: ProviderDriverKind; + readonly instanceId: ProviderInstanceId; + readonly sessionId: string; +}) => + IdAllocator.deriveThreadFromProviderThread({ + driver: input.driver, + providerInstanceId: input.instanceId, + nativeThreadId: input.sessionId, + }); + +export class AcpRegistrySessions extends Context.Service< + AcpRegistrySessions, + { + readonly project: (projectId: ProjectId) => Effect.Effect; + readonly sessionManager: (instanceId: ProviderInstanceId) => Effect.Effect< + { + readonly instance: ProviderInstance; + readonly manager: NonNullable; + }, + AcpRegistryOperationError + >; + /** + * Binds a native ACP session to a T3 thread and copies its history in. Importing a + * session whose thread still lacks its history repairs that thread instead. + */ + readonly importSession: ( + input: AcpRegistryImportSessionInput, + ) => Effect.Effect; + } +>()("t3/provider/AcpRegistrySessions") {} + +const make = Effect.gen(function* () { + const projectService = yield* ProjectService.ProjectService; + const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; + const providerRegistry = yield* ProviderRegistry.ProviderRegistry; + const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator; + const threadManagement = yield* ThreadManagementService.ThreadManagementService; + const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; + const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; + const agentSessionImporter = yield* AgentSessionImporter.AgentSessionImporter; + const crypto = yield* Crypto.Crypto; + + const project = Effect.fn("AcpRegistrySessions.project")(function* (projectId: ProjectId) { + const found = yield* projectService.getById(projectId).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "project_not_found", + message: `Project ${projectId} is unavailable.`, + cause, + }), + ), + ); + return yield* Effect.fromOption( + found, + () => + new AcpRegistryOperationError({ + reason: "project_not_found", + message: `Project ${projectId} was not found.`, + }), + ); + }); + + const sessionManager = Effect.fn("AcpRegistrySessions.sessionManager")(function* ( + instanceId: ProviderInstanceId, + ) { + const instance = yield* providerInstances.getInstance(instanceId); + if (instance === undefined) { + return yield* new AcpRegistryOperationError({ + reason: "instance_not_found", + message: `Provider instance ${instanceId} was not found.`, + }); + } + if (instance.acpSessionManagement === undefined) { + return yield* new AcpRegistryOperationError({ + reason: "session_list_unsupported", + message: `Provider instance ${instanceId} does not expose ACP session management.`, + }); + } + return { instance, manager: instance.acpSessionManagement }; + }); + + const importSession = Effect.fn("AcpRegistrySessions.importSession")(function* ( + input: AcpRegistryImportSessionInput, + ) { + const { workspaceRoot } = yield* project(input.projectId); + const { instance } = yield* sessionManager(input.instanceId); + const providerSnapshot = yield* instance.snapshot.getSnapshot; + if ( + providerSnapshot.nativeSessions?.canLoad !== true && + providerSnapshot.nativeSessions?.canResume !== true + ) { + return yield* new AcpRegistryOperationError({ + reason: "session_resume_unsupported", + message: "The ACP agent cannot load or resume native sessions.", + }); + } + const threadId = importedThreadId({ + driver: instance.driverKind, + instanceId: input.instanceId, + sessionId: input.sessionId, + }); + const existing = yield* threadManagement.getThreadShell(threadId).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "session_import_failed", + message: "Could not inspect the imported ACP session mapping.", + cause, + }), + ), + ); + if (existing !== null) { + const repairing = AgentSessionImporter.isNativeHistoryPending({ + historyOrigin: existing.historyOrigin, + hasRun: existing.latestRunId !== null, + }); + yield* agentSessionImporter.importNativeHistory({ threadId, workspaceRoot }).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "session_import_failed", + message: "Could not import the ACP session history.", + cause, + }), + ), + ); + return { threadId, imported: repairing } as const; + } + + const provider = (yield* providerRegistry.getProviders).find( + (candidate) => candidate.instanceId === input.instanceId, + ); + const model = + provider?.models.find((candidate) => candidate.isDefault)?.slug ?? + provider?.models[0]?.slug ?? + "default"; + const commandId = CommandId.make(yield* crypto.randomUUIDv4().pipe(Effect.orDie)); + const launched = yield* Effect.result( + startup.enqueueCommand( + threadLaunch.launch({ + commandId, + threadId, + projectId: input.projectId, + title: input.title ?? "Imported ACP session", + modelSelection: { instanceId: input.instanceId, model }, + runtimeMode: "approval-required", + interactionMode: "default", + workspaceStrategy: { type: "root" }, + importedNativeThread: { + ref: { + driver: instance.driverKind, + nativeId: input.sessionId, + strength: "strong", + }, + metadata: { + itemIdentityVersion: 2, + ...(input.title === undefined ? {} : { title: input.title }), + ...(input.updatedAt === undefined ? {} : { updatedAt: input.updatedAt }), + }, + }, + createdBy: "user", + creationSource: "web", + }), + ), + ); + if (Result.isFailure(launched)) { + const racedImport = yield* threadManagement.getThreadShell(threadId).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "session_import_failed", + message: "Could not inspect the imported ACP session after launch failed.", + cause, + }), + ), + ); + if (racedImport !== null) return { threadId, imported: false } as const; + return yield* new AcpRegistryOperationError({ + reason: "session_import_failed", + message: "Could not create a T3 thread for the ACP session.", + cause: launched.failure, + }); + } + // The thread is usable without its history, so a failed read only logs. + yield* agentSessionImporter.importNativeHistory({ threadId, 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; + }, coordinator.withSessionMutation); + + return AcpRegistrySessions.of({ project, sessionManager, importSession }); +}); + +export const layer = Layer.effect(AcpRegistrySessions, make); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ae36c4675638e..57015dcb428ec 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -79,6 +79,7 @@ import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts import * as ProjectCloneTracker from "./project/ProjectCloneTracker.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as ProviderAuthService from "./provider/ProviderAuthService.ts"; +import * as AcpRegistrySessions from "./provider/AcpRegistrySessions.ts"; import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; @@ -549,6 +550,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(ProjectEnrichment.ProjectEnrichmentService)({}), Layer.mock(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator)({}), Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({}), + Layer.mock(AcpRegistrySessions.AcpRegistrySessions)({}), Layer.mock(AgentSessionImporter.AgentSessionImporter)({}), Layer.mock(AgentSessionScanner.AgentSessionScanner)({}), Layer.mock(AnalyticsService.AnalyticsService)({}), @@ -1195,7 +1197,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i }), }, } as unknown as ProviderInstance; - const services = Layer.mergeAll( + const dependencies = Layer.mergeAll( layerUnusedConfigRpcServices, Layer.fresh(layerConfigSources), Layer.mock(ExternalLauncher.ExternalLauncher)({}), @@ -1247,6 +1249,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i : Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), }), ); + const services = AcpRegistrySessions.layer.pipe(Layer.provideMerge(dependencies)); const client = yield* RpcTest.makeClient(acpSessionRpcs).pipe( Effect.provide( layerWsRpc( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 942aada56e751..9a0ec0f4db0f3 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -17,7 +17,6 @@ import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import { rpcInitialItems } from "./rpcInitialItems.ts"; import { subscribeChatGptHandoff } from "./provider/CodexChatGptHandoff.ts"; @@ -43,7 +42,6 @@ import { type GitActionProgressEvent, type GitManagerServiceError, type MessageId, - type AcpRegistryImportSessionInput, type AcpRegistryDeleteSessionInput, type AcpRegistryDisableProviderInput, type AcpRegistryListProvidersInput, @@ -91,9 +89,6 @@ import { PersistChatAttachmentsError, RpcClientId, EnvironmentAuthorizationError, - type ProjectId, - type ProviderDriverKind, - type ProviderInstanceId, ThreadId, type TerminalMetadataStreamEvent, type PullRequestRef, @@ -120,7 +115,6 @@ import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManag import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts"; import * as DocumentText from "./provider/DocumentText.ts"; -import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; import * as GuestShareService from "./guest/GuestShareService.ts"; @@ -169,6 +163,7 @@ import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import { userFacingDispatchErrorMessage } from "./orchestration-v2/UserFacingErrors.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; +import * as AcpRegistrySessions from "./provider/AcpRegistrySessions.ts"; import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; @@ -1369,6 +1364,7 @@ export const layerWsRpc = ( const acpRegistryCatalog = yield* AcpRegistrySupport.AcpRegistryCatalog; const acpRegistryRuntimeCoordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator; + const acpRegistrySessions = yield* AcpRegistrySessions.AcpRegistrySessions; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const providerAuth = yield* ProviderAuthService.ProviderAuthService; const providerInstallation = yield* makeProviderInstallation(); @@ -1432,64 +1428,11 @@ export const layerWsRpc = ( ? effect : Effect.fail(rpcAuthorizationError(requiredScope)); - const acpRegistryProject = Effect.fn("ws.acpRegistry.project")(function* ( - projectId: ProjectId, - ) { - const project = yield* projectService.getById(projectId).pipe( - Effect.mapError( - (cause) => - new AcpRegistryOperationError({ - reason: "project_not_found", - message: `Project ${projectId} is unavailable.`, - cause, - }), - ), - ); - return yield* Effect.fromOption( - project, - () => - new AcpRegistryOperationError({ - reason: "project_not_found", - message: `Project ${projectId} was not found.`, - }), - ); - }); - - const acpSessionManager = Effect.fn("ws.acpRegistry.sessionManager")(function* ( - instanceId: ProviderInstanceId, - ) { - const instance = yield* providerInstances.getInstance(instanceId); - if (instance === undefined) { - return yield* new AcpRegistryOperationError({ - reason: "instance_not_found", - message: `Provider instance ${instanceId} was not found.`, - }); - } - if (instance.acpSessionManagement === undefined) { - return yield* new AcpRegistryOperationError({ - reason: "session_list_unsupported", - message: `Provider instance ${instanceId} does not expose ACP session management.`, - }); - } - return { instance, manager: instance.acpSessionManagement }; - }); - - const importedAcpThreadId = (input: { - readonly driver: ProviderDriverKind; - readonly instanceId: ProviderInstanceId; - readonly sessionId: string; - }) => - IdAllocator.deriveThreadFromProviderThread({ - driver: input.driver, - providerInstanceId: input.instanceId, - nativeThreadId: input.sessionId, - }); - const listAcpRegistrySessions = Effect.fn("ws.acpRegistry.listSessions")(function* ( input: AcpRegistryListSessionsInput, ) { - const project = yield* acpRegistryProject(input.projectId); - const { instance, manager } = yield* acpSessionManager(input.instanceId); + const project = yield* acpRegistrySessions.project(input.projectId); + const { instance, manager } = yield* acpRegistrySessions.sessionManager(input.instanceId); const listed = yield* manager.listSessions({ cwd: project.workspaceRoot, ...(input.cursor === undefined ? {} : { cursor: input.cursor }), @@ -1497,7 +1440,7 @@ export const layerWsRpc = ( const sessions = yield* Effect.forEach( listed.sessions, (session) => { - const threadId = importedAcpThreadId({ + const threadId = AcpRegistrySessions.importedThreadId({ driver: instance.driverKind, instanceId: input.instanceId, sessionId: session.sessionId, @@ -1529,136 +1472,15 @@ export const layerWsRpc = ( return { ...listed, sessions }; }); - const importAcpRegistrySession = Effect.fn("ws.acpRegistry.importSession")(function* ( - input: AcpRegistryImportSessionInput, - ) { - return yield* acpRegistryRuntimeCoordinator.withSessionMutation( - Effect.gen(function* () { - const project = yield* acpRegistryProject(input.projectId); - const { instance } = yield* acpSessionManager(input.instanceId); - const providerSnapshot = yield* instance.snapshot.getSnapshot; - if ( - providerSnapshot.nativeSessions?.canLoad !== true && - providerSnapshot.nativeSessions?.canResume !== true - ) { - return yield* new AcpRegistryOperationError({ - reason: "session_resume_unsupported", - message: "The ACP agent cannot load or resume native sessions.", - }); - } - const threadId = importedAcpThreadId({ - driver: instance.driverKind, - instanceId: input.instanceId, - sessionId: input.sessionId, - }); - const existing = yield* threadManagement.getThreadShell(threadId).pipe( - Effect.mapError( - (cause) => - new AcpRegistryOperationError({ - reason: "session_import_failed", - message: "Could not inspect the imported ACP session mapping.", - cause, - }), - ), - ); - if (existing !== null) { - const repairing = AgentSessionImporter.isNativeHistoryPending({ - historyOrigin: existing.historyOrigin, - hasRun: existing.latestRunId !== 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: repairing } as const; - } - - const provider = (yield* providerRegistry.getProviders).find( - (candidate) => candidate.instanceId === input.instanceId, - ); - const model = - provider?.models.find((candidate) => candidate.isDefault)?.slug ?? - provider?.models[0]?.slug ?? - "default"; - const commandId = CommandId.make(yield* crypto.randomUUIDv4().pipe(Effect.orDie)); - const launched = yield* Effect.result( - startup.enqueueCommand( - threadLaunch.launch({ - commandId, - threadId, - projectId: input.projectId, - title: input.title ?? "Imported ACP session", - modelSelection: { instanceId: input.instanceId, model }, - runtimeMode: "approval-required", - interactionMode: "default", - workspaceStrategy: { type: "root" }, - importedNativeThread: { - ref: { - driver: instance.driverKind, - nativeId: input.sessionId, - strength: "strong", - }, - metadata: { - itemIdentityVersion: 2, - ...(input.title === undefined ? {} : { title: input.title }), - ...(input.updatedAt === undefined ? {} : { updatedAt: input.updatedAt }), - }, - }, - createdBy: "user", - creationSource: "web", - }), - ), - ); - if (Result.isFailure(launched)) { - const racedImport = yield* threadManagement.getThreadShell(threadId).pipe( - Effect.mapError( - (cause) => - new AcpRegistryOperationError({ - reason: "session_import_failed", - message: "Could not inspect the imported ACP session after launch failed.", - cause, - }), - ), - ); - if (racedImport !== null) return { threadId, imported: false } as const; - return yield* new AcpRegistryOperationError({ - reason: "session_import_failed", - message: "Could not create a T3 thread for the ACP session.", - cause: launched.failure, - }); - } - // The thread is usable without its history, so a failed read only logs. - yield* agentSessionImporter - .importNativeHistory({ threadId, 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; - }), - ); - }); - const deleteAcpRegistrySession = Effect.fn("ws.acpRegistry.deleteSession")(function* ( input: AcpRegistryDeleteSessionInput, ) { return yield* acpRegistryRuntimeCoordinator.withSessionMutation( Effect.gen(function* () { - const project = yield* acpRegistryProject(input.projectId); - const { instance, manager } = yield* acpSessionManager(input.instanceId); + const project = yield* acpRegistrySessions.project(input.projectId); + const { instance, manager } = yield* acpRegistrySessions.sessionManager( + input.instanceId, + ); const snapshot = yield* instance.snapshot.getSnapshot; if (snapshot.nativeSessions?.canDelete !== true) { return yield* new AcpRegistryOperationError({ @@ -1666,7 +1488,7 @@ export const layerWsRpc = ( message: "The ACP agent does not advertise session deletion.", }); } - const threadId = importedAcpThreadId({ + const threadId = AcpRegistrySessions.importedThreadId({ driver: instance.driverKind, instanceId: input.instanceId, sessionId: input.sessionId, @@ -1699,8 +1521,8 @@ export const layerWsRpc = ( const listAcpRegistryProviders = Effect.fn("ws.acpRegistry.listProviders")(function* ( input: AcpRegistryListProvidersInput, ) { - const project = yield* acpRegistryProject(input.projectId); - const { instance, manager } = yield* acpSessionManager(input.instanceId); + const project = yield* acpRegistrySessions.project(input.projectId); + const { instance, manager } = yield* acpRegistrySessions.sessionManager(input.instanceId); const snapshot = yield* instance.snapshot.getSnapshot; if (snapshot.configurableProviders !== true) { return yield* new AcpRegistryOperationError({ @@ -1714,8 +1536,8 @@ export const layerWsRpc = ( const setAcpRegistryProvider = Effect.fn("ws.acpRegistry.setProvider")(function* ( input: AcpRegistrySetProviderInput, ) { - const project = yield* acpRegistryProject(input.projectId); - const { manager } = yield* acpSessionManager(input.instanceId); + const project = yield* acpRegistrySessions.project(input.projectId); + const { manager } = yield* acpRegistrySessions.sessionManager(input.instanceId); if (input.headers !== undefined && Object.keys(input.headers).length > 32) { return yield* new AcpRegistryOperationError({ reason: "provider_configuration_failed", @@ -1756,8 +1578,8 @@ export const layerWsRpc = ( const disableAcpRegistryProvider = Effect.fn("ws.acpRegistry.disableProvider")(function* ( input: AcpRegistryDisableProviderInput, ) { - const project = yield* acpRegistryProject(input.projectId); - const { manager } = yield* acpSessionManager(input.instanceId); + const project = yield* acpRegistrySessions.project(input.projectId); + const { manager } = yield* acpRegistrySessions.sessionManager(input.instanceId); const listed = yield* manager.listProviders(project.workspaceRoot); const provider = listed.providers.find( (candidate) => candidate.providerId === input.providerId, @@ -2301,7 +2123,7 @@ export const layerWsRpc = ( Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId, "project.id": input.projectId, - }).pipe(Effect.andThen(importAcpRegistrySession(input))), + }).pipe(Effect.andThen(acpRegistrySessions.importSession(input))), [WS_METHODS.serverDeleteAcpRegistrySession]: (input) => Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId, @@ -2326,7 +2148,9 @@ export const layerWsRpc = ( Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId }).pipe( Effect.andThen( Effect.gen(function* () { - const { instance, manager } = yield* acpSessionManager(input.instanceId); + const { instance, manager } = yield* acpRegistrySessions.sessionManager( + input.instanceId, + ); const snapshot = yield* instance.snapshot.getSnapshot; if (snapshot.auth.canLogout !== true) { return yield* new AcpRegistryOperationError({ @@ -3445,6 +3269,7 @@ export const layer = Layer.unwrap( Layer.provide(DefectReporter.layer), Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), Layer.provide(AgentSessionScanner.layer), + Layer.provide(AcpRegistrySessions.layer), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide(Layer.succeed(NpmRelease.NpmRelease, npmRelease)), Layer.provide(Layer.succeed(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)), From de09c91a7888874adbfaea5827347c31a03fc514 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sun, 8 Nov 2026 12:46:45 +0000 Subject: [PATCH 11/15] test(server): adapt the ACP import fixtures to main Main's layerWsRpc takes the client's settings capabilities, so the ACP session RPC test passes them. Effect 4.1.2 flags process.env reads inside Effect code, so the mock agent reads its close-session flag at module load like its other flags. Co-Authored-By: Claude Opus 5.5 --- apps/server/scripts/acp-mock-agent.ts | 3 ++- apps/server/src/ws.test.ts | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index a2bd4448d6903..b631029d7f798 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -28,6 +28,7 @@ const emitInterleavedAssistantToolCalls = process.env.T3_ACP_EMIT_INTERLEAVED_ASSISTANT_TOOL_CALLS === "1"; const emitV2Fidelity = process.env.T3_ACP_EMIT_V2_FIDELITY === "1"; const emitNotice = process.env.T3_ACP_EMIT_NOTICE === "1"; +const hangCloseSession = process.env.T3_ACP_HANG_CLOSE_SESSION === "1"; const vibeRetryOutcome = process.env.T3_ACP_VIBE_RETRY_OUTCOME; const emitGenericToolPlaceholders = process.env.T3_ACP_EMIT_GENERIC_TOOL_PLACEHOLDERS === "1"; const emitInTurnTaskOutputThenLateDuplicate = @@ -731,7 +732,7 @@ const program = Effect.gen(function* () { yield* agent.handleCloseSession(() => Effect.gen(function* () { yield* requireAuthentication(); - if (process.env.T3_ACP_HANG_CLOSE_SESSION === "1") { + if (hangCloseSession) { return yield* Effect.never; } return {}; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index aafce510a251c..7b2cec54cbea3 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1815,6 +1815,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i {}, yield* ServerBrowser.ServerBrowser, () => Effect.void, + { secretReferences: true }, ).pipe( Layer.provide(services), Layer.merge( From 061f6e9d65ac30312e692ca67599c11a22675b34 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Tue, 10 Nov 2026 17:58:05 +0000 Subject: [PATCH 12/15] fix(server): imported ACP thinking shows as reasoning, not replies Loading an ACP session replays agent_thought_chunk updates into the adapter's snapshot as assistant text, so the native history import wrote the agent's thinking as chat replies. The snapshot now lists which of its messages came from thought chunks, and the importer writes those as run-less reasoning items, the item type live ACP turns use. Co-authored-by: Alan Justino da Silva <155623+alanjds@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 10 +++- .../src/project/AgentSessionImporter.test.ts | 7 ++- .../src/project/AgentSessionImporter.ts | 59 +++++++++++-------- packages/provider-acp/src/server/adapter.ts | 56 +++++++----------- .../src/server/ProviderAdapter.ts | 2 + 5 files changed, 72 insertions(+), 62 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 89b3187ffbeca..d5332b016cbf9 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -2620,12 +2620,20 @@ describe("AcpAdapterV2", () => { content: { type: "text", text: "after plan" }, }, }); + yield* sessionUpdateHandler!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: "weighing it" }, + }, + }); const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); assert.deepEqual( snapshot.messages.map((message) => message.text), - ["before plan", "after plan"], + ["before plan", "after plan", "weighing it"], ); + assert.deepEqual(snapshot.reasoningMessageIds, [snapshot.messages[2]!.id]); }).pipe(Effect.provide(layerTest), Effect.scoped), ); diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index f10b7477bd877..c659622a2f1e8 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -489,8 +489,10 @@ it.effect( providerThread, providerTurns: [], runtimeRequests: [], + reasoningMessageIds: [MessageId.make("replayed:Check the build log")], messages: [ replayed("user", "Fix the build"), + replayed("assistant", "Check the build log"), replayed("assistant", "Fixed it"), replayed("assistant", " "), ], @@ -524,7 +526,7 @@ it.effect( ).toBeUndefined(); failRead = false; - expect(yield* importHistory).toBe(2); + expect(yield* importHistory).toBe(3); expect(resolved).toEqual([ CODEX_MODEL_SELECTION.instanceId, CODEX_MODEL_SELECTION.instanceId, @@ -540,7 +542,7 @@ it.effect( ); expect(yield* importHistory).toBe(0); expect(opened).toHaveLength(2); - const page = yield* projections.getTimelinePage(threadId, { view: "messages", limit: 10 }); + const page = yield* projections.getTimelinePage(threadId, { view: "activity", limit: 10 }); expect( page.items.map(({ item }) => [ item.type, @@ -549,6 +551,7 @@ it.effect( ]), ).toEqual([ ["user_message", null, "Fix the build"], + ["reasoning", null, "Check the build log"], ["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 f4119d87a5739..79b72e524e096 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -107,10 +107,14 @@ function dateTime(value: string): DateTime.Utc { return DateTime.makeUnsafe(value); } +type HistoryMessage = Omit & { + readonly role: AgentSessionScanner.AgentSessionThreadMessage["role"] | "reasoning"; +}; + function messageEvents(input: { readonly threadId: ThreadId; readonly index: number; - readonly message: AgentSessionScanner.AgentSessionThreadMessage; + readonly message: HistoryMessage; }): ReadonlyArray { const ordinal = input.index + 1; const suffix = String(input.index).padStart(6, "0"); @@ -119,20 +123,6 @@ function messageEvents(input: { `${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`, ); const at = dateTime(input.message.createdAt); - const message: OrchestrationV2ConversationMessage = { - createdBy: input.message.role === "user" ? "user" : "agent", - creationSource: "server", - id: messageId, - threadId: input.threadId, - runId: null, - nodeId: null, - role: input.message.role, - text: input.message.text, - attachments: [], - streaming: false, - createdAt: at, - updatedAt: at, - }; const common = { id: turnItemId, threadId: input.threadId, @@ -149,6 +139,32 @@ function messageEvents(input: { completedAt: at, updatedAt: at, }; + const turnItemEvent = (payload: OrchestrationV2TurnItem): OrchestrationV2DomainEvent => ({ + id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`), + type: "turn-item.updated", + threadId: input.threadId, + occurredAt: at, + payload, + }); + if (input.message.role === "reasoning") { + return [ + turnItemEvent({ ...common, type: "reasoning", text: input.message.text, streaming: false }), + ]; + } + const message: OrchestrationV2ConversationMessage = { + createdBy: input.message.role === "user" ? "user" : "agent", + creationSource: "server", + id: messageId, + threadId: input.threadId, + runId: null, + nodeId: null, + role: input.message.role, + text: input.message.text, + attachments: [], + streaming: false, + createdAt: at, + updatedAt: at, + }; const turnItem: OrchestrationV2TurnItem = input.message.role === "user" ? { @@ -176,13 +192,7 @@ function messageEvents(input: { occurredAt: at, payload: message, }, - { - id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`), - type: "turn-item.updated", - threadId: input.threadId, - occurredAt: at, - payload: turnItem, - }, + turnItemEvent(turnItem), ]; } @@ -494,7 +504,7 @@ const make = Effect.gen(function* () { ); // 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) => + const commit = (messages: ReadonlyArray) => threadCommands.withLock( input.threadId, Effect.gen(function* () { @@ -554,11 +564,12 @@ const make = Effect.gen(function* () { Effect.timeout(NATIVE_HISTORY_READ_TIMEOUT), Effect.scoped, ); + const reasoningIds = new Set(snapshot.reasoningMessageIds); const messages = snapshot.messages.flatMap((message) => (message.role === "user" || message.role === "assistant") && message.text.trim() !== "" ? [ { - role: message.role, + role: reasoningIds.has(message.id) ? ("reasoning" as const) : message.role, text: message.text, createdAt: DateTime.formatIso(message.createdAt), }, diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 730c5d589aff1..9a8709af1a738 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1641,11 +1641,22 @@ type PendingRuntimeRequest = { interface SnapshotMessageState { readonly order: Array; readonly messages: Map; + /** Messages loaded from thought chunks; `messages` stores them as assistant text. */ + readonly reasoningIds: ReadonlySet; loadingRole: "user" | "assistant" | "thought" | null; loadingMessageId: string | null; loadingIndex: number; } +const emptySnapshotState = (): SnapshotMessageState => ({ + order: [], + messages: new Map(), + reasoningIds: new Set(), + loadingRole: null, + loadingMessageId: null, + loadingIndex: 0, +}); + /** How long a Stop waits for the agent to confirm before the user is offered Force stop. */ const ACP_STOP_CONFIRM_TIMEOUT_MS = 10_000; const TOOL_UPDATE_PERSIST_EVERY = 10; @@ -1962,13 +1973,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( options.testHooks?.onNativeResponseLifecycle ?? (() => Effect.void); const nextElicitationOrdinal = yield* Ref.make(0); const providerTurns = yield* Ref.make(new Map()); - const snapshot = yield* Ref.make({ - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); + const snapshot = yield* Ref.make(emptySnapshotState()); const awaitRuntimeTeardown = Effect.fnUntraced(function* () { const state = yield* Ref.get(runtimeTeardownState); @@ -4548,6 +4553,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return { order: current.order.includes(key) ? current.order : [...current.order, key], messages, + reasoningIds: + role === "thought" + ? new Set(current.reasoningIds).add(messageId) + : current.reasoningIds, loadingRole: role, loadingMessageId: sourceMessageId ?? (startsNew ? null : current.loadingMessageId), loadingIndex, @@ -7105,13 +7114,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(promptInstructionStates, new Map()); yield* Ref.set(compactionMarkers, new Map()); yield* Ref.set(providerTurns, new Map()); - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); + yield* Ref.set(snapshot, emptySnapshotState()); yield* continuationPermit.withPermit( Effect.gen(function* () { yield* Ref.update(continuationGeneration, (value) => value + 1); @@ -8049,13 +8052,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* Ref.set(activeSessionSetup, null); yield* Ref.set(activeSelection, null); yield* Ref.set(activeInteractionMode, null); - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); + yield* Ref.set(snapshot, emptySnapshotState()); return true; }); @@ -8737,13 +8734,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( threadInput.providerThread.appThreadId, ).pipe(Effect.tapError(() => restorePreviousItemIdentity)); if ((yield* Ref.get(activeSessionId)) !== sessionId || restartAfterInterrupt) { - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); + yield* Ref.set(snapshot, emptySnapshotState()); const activated = yield* activateSession( sessionId, threadInput.providerThread.appThreadId, @@ -9241,13 +9232,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( detail: "ACP driver does not support session/load snapshots", }); } - yield* Ref.set(snapshot, { - order: [], - messages: new Map(), - loadingRole: null, - loadingMessageId: null, - loadingIndex: 0, - }); + yield* Ref.set(snapshot, emptySnapshotState()); const mcpContext = yield* readMcpContext( snapshotInput.providerThread.appThreadId, ); @@ -9279,6 +9264,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const message = state.messages.get(key); return message === undefined ? [] : [message]; }), + reasoningMessageIds: [...state.reasoningIds], runtimeRequests: [], providerPayload: { protocol: ACP_PROTOCOL, sessionId }, }; diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index e94740e34712f..8e65b18ed5b64 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -517,6 +517,8 @@ export interface ProviderAdapterV2ThreadSnapshot { readonly providerThread: OrchestrationV2ProviderThread; readonly providerTurns: ReadonlyArray; readonly messages: ReadonlyArray; + /** Ids in `messages` that hold the agent's reasoning, not its reply. */ + readonly reasoningMessageIds?: ReadonlyArray; readonly runtimeRequests: ReadonlyArray; readonly providerPayload?: unknown; } From 54f92235c322de03076090cc9818931a2c1bcc52 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Tue, 10 Nov 2026 18:24:12 +0000 Subject: [PATCH 13/15] test(server): cover whole replayed ACP thoughts as reasoning A replayed agent_thought (not only agent_thought_chunk) is now covered by the snapshot test. The reasoning id set is copied only when a new thought starts, and the import event id reuses the turn item id. Co-authored-by: Alan Justino da Silva <155623+alanjds@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpAdapterV2.test.ts | 15 +++++++++++++-- apps/server/src/project/AgentSessionImporter.ts | 2 +- packages/provider-acp/src/server/adapter.ts | 2 +- 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index d5332b016cbf9..cf3fcc03fda6a 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -2627,13 +2627,24 @@ describe("AcpAdapterV2", () => { content: { type: "text", text: "weighing it" }, }, }); + yield* sessionUpdateHandler!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_thought", + messageId: "whole-thought", + content: [{ type: "text", text: "settled on it" }], + }, + }); const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); assert.deepEqual( snapshot.messages.map((message) => message.text), - ["before plan", "after plan", "weighing it"], + ["before plan", "after plan", "weighing it", "settled on it"], + ); + assert.deepEqual( + snapshot.reasoningMessageIds, + snapshot.messages.slice(2).map((message) => message.id), ); - assert.deepEqual(snapshot.reasoningMessageIds, [snapshot.messages[2]!.id]); }).pipe(Effect.provide(layerTest), Effect.scoped), ); diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 79b72e524e096..ec3e9e3bbe21a 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -140,7 +140,7 @@ function messageEvents(input: { updatedAt: at, }; const turnItemEvent = (payload: OrchestrationV2TurnItem): OrchestrationV2DomainEvent => ({ - id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${input.threadId}:${suffix}`), + id: EventId.make(turnItemId), type: "turn-item.updated", threadId: input.threadId, occurredAt: at, diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 9a8709af1a738..a57899fb1472e 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -4554,7 +4554,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( order: current.order.includes(key) ? current.order : [...current.order, key], messages, reasoningIds: - role === "thought" + role === "thought" && !current.reasoningIds.has(messageId) ? new Set(current.reasoningIds).add(messageId) : current.reasoningIds, loadingRole: role, From 57c141ab983ca5e0305b032a97fc0b3d03a3a147 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 23:13:58 +0000 Subject: [PATCH 14/15] fix(server): drop imports the main merge duplicated in ws tests --- apps/server/src/ws.test.ts | 3 --- 1 file changed, 3 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index e4756375c2188..32a0055e2a0b6 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -146,11 +146,8 @@ import * as EventStore from "./orchestration-v2/EventStore.ts"; import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; -import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; -import * as EventSink from "./orchestration-v2/EventSink.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; import * as ThreadCommandExecutor from "./orchestration-v2/ThreadCommandExecutor.ts"; -import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; From 19ade0208185ae574f59a08040ac4049f88510c1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 05:55:09 +0000 Subject: [PATCH 15/15] test: trim ACP import tests to the regressions they catch Drop the settings browser test that only checked a button label derived from a flag, drop the ws test's error-mapping branches, and revert the unrelated import reordering in ws.test.ts. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/ws.test.ts | 36 ++++++------------- ...pSessionManagementSection.browser.test.tsx | 27 -------------- 2 files changed, 11 insertions(+), 52 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 32a0055e2a0b6..3776992685498 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -30,10 +30,9 @@ import { type ServerConfigStreamEvent, type ServerProvider, } from "@t3tools/contracts"; -import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as Deferred from "effect/Deferred"; 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 Exit from "effect/Exit"; import * as Cause from "effect/Cause"; @@ -141,13 +140,14 @@ import { USAGE_LIMITS_COMMAND } from "@t3tools/shared/usageLimits"; import * as ServerSettings from "./serverSettings.ts"; import * as ManagedPolicy from "./policy/ManagedPolicy.ts"; -import * as EventSink from "./orchestration-v2/EventSink.ts"; import * as EventStore from "./orchestration-v2/EventStore.ts"; -import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; +import * as EventSink from "./orchestration-v2/EventSink.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; import * as ThreadCommandExecutor from "./orchestration-v2/ThreadCommandExecutor.ts"; +import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; @@ -1921,8 +1921,6 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i 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 timingOut = yield* seedAcpThread("timing-out", { imported: false, ran: false }); const repaired: Array = []; const session = (sessionId: string) => ({ sessionId, @@ -1991,11 +1989,7 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), Layer.mock(AgentSessionImporter.AgentSessionImporter)({ importNativeHistory: ({ threadId }) => - threadId === failing - ? Effect.fail(new Orchestrator.OrchestratorProjectionError({ threadId, cause: "read" })) - : threadId === timingOut - ? Effect.fail(new Cause.TimeoutError()) - : Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), + Effect.sync(() => repaired.push(threadId)).pipe(Effect.as(1)), }), ); const services = AcpRegistrySessions.layer.pipe(Layer.provideMerge(dependencies)); @@ -2039,21 +2033,13 @@ it.effect("ACP session RPCs list a thread missing its history and repair it on i ], ); - const importSession = (sessionId: string) => - client[WS_METHODS.serverImportAcpRegistrySession]({ - instanceId: acpInstanceId, - projectId: acpProjectId, - sessionId, - }); - assert.deepEqual(yield* importSession("pending"), { threadId: pending, imported: true }); + const imported = yield* client[WS_METHODS.serverImportAcpRegistrySession]({ + instanceId: acpInstanceId, + projectId: acpProjectId, + sessionId: "pending", + }); + assert.deepEqual(imported, { threadId: pending, imported: true }); assert.deepEqual(repaired, [pending]); - 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( Layer.mergeAll( diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx index 7bd56d72f633f..824d4a28f99a1 100644 --- a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx +++ b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx @@ -239,33 +239,6 @@ it.each(["Save", "Disable"])("keeps other provider edits through a %s refetch", 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"); - await expect.element(screen.button("Delete")).toBeDisabled(); - await screen.click("Import"); - expect(commands.import).toHaveBeenCalledOnce(); - await expect.element(screen.button("Imported")).toBeDisabled(); - await expect.element(screen.button("Delete")).toBeDisabled(); -}); - it.each([true, false])( "enables project actions after projects arrive (sessions: %s)", async (canList) => {