diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index ae07db783c921..1387bd38ca0f3 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -711,6 +711,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 bb84c1a2c7492..44a62856e2729 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -4652,6 +4652,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/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, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 1b54627595372..11c134e4d79da 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -243,6 +243,8 @@ const layerAgentSessionImporterProvided = AgentSessionImporter.layer.pipe( layerEventSink, IdAllocator.layer, ProviderSessionRuntime.layer, + ProviderAdapterRegistry.layerFromProviderInstanceRegistry, + ThreadCommandExecutor.layer, ), ), ); diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 140dff31486b1..f10b7477bd877 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -6,12 +6,18 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + MessageId, + RunId, ThreadId, type ApplicationStoredEvent, type OrchestrationV2AppThread, + 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 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"; import * as Deferred from "effect/Deferred"; @@ -21,11 +27,14 @@ 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"; 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, @@ -172,6 +181,8 @@ const seedOwner = Effect.fn(function* ( const makeImporter = Effect.fn(function* ( candidates: ReadonlyArray>, beforeTransaction: Effect.Effect = Effect.void, + getAdapter?: ProviderAdapterRegistry.ProviderAdapterRegistryV2Shape["get"], + threadCommands?: KeyedLock.KeyedLock, settings: Layer.Layer< ServerSettings.ServerSettingsService, ServerSettingsError @@ -243,12 +254,23 @@ const makeImporter = Effect.fn(function* ( ), ), }), + Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)( + getAdapter === undefined ? {} : { get: getAdapter }, + ), + Layer.succeed( + ThreadCommandExecutor.ThreadCommandExecutor, + threadCommands ?? (yield* KeyedLock.make()), + ), ), ), ), ); const importer = Context.get(context, AgentSessionImporter.AgentSessionImporter); - return { run: importer.importRecentAgentThreads({ projectId }), published }; + return { + run: importer.importRecentAgentThreads({ projectId }), + importNativeHistory: importer.importNativeHistory, + published, + }; }); it.effect( @@ -424,6 +446,318 @@ it.effect( }).pipe(Effect.scoped, Effect.provide(persistence)), ); +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(() => { + 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; + }), + () => + Effect.sync(() => { + closed = true; + }), + ), + } as unknown as ProviderAdapter.ProviderAdapterV2["Service"]; + const importer = yield* makeImporter([], Effect.void, (instanceId) => + Effect.sync(() => { + resolved.push(instanceId); + return adapter; + }), + ); + + 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" }, + }); + // 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)), +); + +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 its first session/resume. + const adapter = { + openSession: () => + 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* 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(); + 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)), + ), +); + +/** + * 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)), +); + it.effect("imports under a policy ceiling in the project's default, kept under the ceiling", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -433,6 +767,8 @@ it.effect("imports under a policy ceiling in the project's default, kept under t const importer = yield* makeImporter( [source], Effect.void, + undefined, + undefined, ServerSettings.layerTest( { defaultRuntimeMode: "full-access" }, { diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 211ad06fc267a..f4119d87a5739 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -36,6 +36,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 * 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 ServerSettings from "../serverSettings.ts"; import * as AgentSessionScanner from "./AgentSessionScanner.ts"; @@ -85,6 +87,22 @@ class AgentSessionThreadModifiedError extends Schema.TaggedError + isNativeHistoryPending({ + historyOrigin: records.thread.historyOrigin, + hasRun: records.runs.length > 0, + }) + ? records + : null, + ); + // 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; + 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, + providerInstanceId: current.thread.providerInstanceId, + occurredAt: yield* DateTime.now, + payload: { ...current.thread, historyOrigin: "v1_import" }, + }, + ], + }); + return messages.length; + }), + ); + 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, + ); + const nativeThreadId = providerThread?.nativeThreadRef?.nativeId; + if (providerThread === undefined || nativeThreadId == null) return 0; + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: providerThread.providerInstanceId, + threadId: input.threadId, + }); + // 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, + 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.timeout(NATIVE_HISTORY_READ_TIMEOUT), + 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), + }, + ] + : [], + ); + // Loading the session can take a while, and the thread is usable meanwhile. + return yield* commit(messages); + }); + + return { importRecentAgentThreads, importNativeHistory }; }); type AgentSessionImporterShape = Effect.Success; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 97ed92fe21bbc..ae36c4675638e 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,11 @@ 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 Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -29,9 +40,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"; @@ -113,6 +126,7 @@ import * as Keybindings from "./keybindings.ts"; 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 ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; @@ -1053,3 +1067,248 @@ 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 timingOut = yield* seedAcpThread("timing-out", { 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(ScratchThreadPromotion.ScratchThreadPromotion)({}), + 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" })) + : threadId === timingOut + ? Effect.fail(new Cause.TimeoutError()) + : 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* ServerBrowser.ServerBrowser, + () => Effect.void, + ).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]); + 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( + 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 cc06273187a9a..942aada56e751 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1506,6 +1506,13 @@ export const layerWsRpc = ( Effect.map((thread) => ({ ...session, importedThreadId: thread === null ? null : threadId, + ...(thread !== null && + AgentSessionImporter.isNativeHistoryPending({ + historyOrigin: thread.historyOrigin, + hasRun: thread.latestRunId !== null, + }) + ? { nativeHistoryPending: true } + : {}), })), Effect.mapError( (cause) => @@ -1527,7 +1534,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 ( @@ -1554,7 +1561,25 @@ export const layerWsRpc = ( }), ), ); - if (existing !== null) return { threadId, imported: false } as const; + 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, @@ -1610,6 +1635,18 @@ export const layerWsRpc = ( 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; }), ); diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.browser.test.tsx index a61814ae3efa6..b2779eebe02df 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"); + 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(); +}); 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; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 7b24d9ef09adf..2a82901f41014 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -7965,7 +7965,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); } }), );