diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 44011bc7f2d4b..03f69beb57ccd 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,6 +732,9 @@ const program = Effect.gen(function* () { yield* agent.handleCloseSession(() => Effect.gen(function* () { yield* requireAuthentication(); + if (hangCloseSession) { + 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 f63a220fb5191..cf3fcc03fda6a 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -2620,11 +2620,30 @@ 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" }, + }, + }); + 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"], + ["before plan", "after plan", "weighing it", "settled on it"], + ); + assert.deepEqual( + snapshot.reasoningMessageIds, + snapshot.messages.slice(2).map((message) => message.id), ); }).pipe(Effect.provide(layerTest), Effect.scoped), ); @@ -5453,6 +5472,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 07c1a544798be..6846d24274a2e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1007,12 +1007,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 ); } @@ -6872,6 +6876,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 ba1dadad1864c..6ea169ee98bd0 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -244,6 +244,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..c659622a2f1e8 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,321 @@ 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: [], + 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", " "), + ], + }), + } 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(3); + 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: "activity", limit: 10 }); + expect( + page.items.map(({ item }) => [ + item.type, + item.runId, + "text" in item ? item.text : undefined, + ]), + ).toEqual([ + ["user_message", null, "Fix the build"], + ["reasoning", null, "Check the build log"], + ["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 +770,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..ec3e9e3bbe21a 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,14 +87,34 @@ class AgentSessionThreadModifiedError extends Schema.TaggedError & { + 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"); @@ -101,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, @@ -131,6 +139,32 @@ function messageEvents(input: { completedAt: at, updatedAt: at, }; + const turnItemEvent = (payload: OrchestrationV2TurnItem): OrchestrationV2DomainEvent => ({ + id: EventId.make(turnItemId), + 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" ? { @@ -158,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), ]; } @@ -175,6 +203,8 @@ 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 threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const settingsService = yield* ServerSettings.ServerSettingsService; const ceiling = settingsService.managedPolicy?.limits?.maxRuntimeMode; const importRecentAgentThreads = Effect.fn("importRecentAgentThreadsV2")(function* ( @@ -452,7 +482,105 @@ const make = Effect.gen(function* () { return { importedCount, skippedCount } satisfies AgentSessionImportResult; }); - return { importRecentAgentThreads }; + /** + * 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; + readonly workspaceRoot: string; + }) { + const pendingThread = Effect.map( + orchestrator.getThreadRecords(input.threadId, ["providerThreads", "runs"]), + (records) => + 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 reasoningIds = new Set(snapshot.reasoningMessageIds); + const messages = snapshot.messages.flatMap((message) => + (message.role === "user" || message.role === "assistant") && message.text.trim() !== "" + ? [ + { + role: reasoningIds.has(message.id) ? ("reasoning" as const) : 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/provider/AcpRegistrySessions.ts b/apps/server/src/provider/AcpRegistrySessions.ts new file mode 100644 index 0000000000000..9cf869cb78123 --- /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 SecureRandom from "effect/crypto/SecureRandom"; +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 random = yield* SecureRandom.SecureRandom; + + 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* random.uuidV4()); + 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 6374204471964..3776992685498 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -3,6 +3,8 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, AuthSessionId, AuthSourceControlWriteScope, EnvironmentId, @@ -13,11 +15,14 @@ import { TurnItemId, ProviderDriverKind, ProviderInstanceId, + RunId, SourceControlRepositoryError, WS_METHODS, WsRpcGroup, WsServerRefreshProvidersRpc, WsSubscribeServerConfigRpc, + type OrchestrationV2AppThread, + type OrchestrationV2DomainEvent, type OrchestrationV2StoredEvent, type OrchestrationV2ThreadShell, type OrchestrationProjectShell, @@ -26,6 +31,7 @@ import { type ServerProvider, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -40,6 +46,7 @@ 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 Statement from "effect/sql/Statement"; import { vi } from "vite-plus/test"; @@ -83,6 +90,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"; @@ -132,6 +140,7 @@ import { USAGE_LIMITS_COMMAND } from "@t3tools/shared/usageLimits"; import * as ServerSettings from "./serverSettings.ts"; import * as ManagedPolicy from "./policy/ManagedPolicy.ts"; +import * as EventStore from "./orchestration-v2/EventStore.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"; @@ -782,6 +791,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)({}), @@ -1808,3 +1818,236 @@ it.effect("thread resume at its own high-water below the pruned boundary skips t assert.isTrue(Exit.isSuccess(exit)); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); + +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 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 dependencies = 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 }) => + 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( + { + sessionId: AuthSessionId.make("acp-test"), + subject: "acp-test", + method: "browser-session-cookie", + scopes: [], + }, + {}, + yield* ServerBrowser.ServerBrowser, + () => Effect.void, + { secretReferences: true }, + ).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 imported = yield* client[WS_METHODS.serverImportAcpRegistrySession]({ + instanceId: acpInstanceId, + projectId: acpProjectId, + sessionId: "pending", + }); + assert.deepEqual(imported, { threadId: pending, imported: true }); + assert.deepEqual(repaired, [pending]); + }).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 9721864fcdfd5..5f668ab1ba0c8 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, @@ -94,9 +92,6 @@ import { PersistChatAttachmentsError, RpcClientId, EnvironmentAuthorizationError, - type ProjectId, - type ProviderDriverKind, - type ProviderInstanceId, ThreadId, type TerminalMetadataStreamEvent, type PullRequestRef, @@ -124,7 +119,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"; @@ -174,6 +168,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"; @@ -1503,6 +1498,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* ProviderInstallation.ProviderInstallation; @@ -1567,64 +1563,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 }), @@ -1632,7 +1575,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, @@ -1641,6 +1584,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) => @@ -1657,106 +1607,15 @@ export const layerWsRpc = ( return { ...listed, sessions }; }); - const importAcpRegistrySession = Effect.fn("ws.acpRegistry.importSession")(function* ( - input: AcpRegistryImportSessionInput, - ) { - return yield* acpRegistryRuntimeCoordinator.withSessionMutation( - Effect.gen(function* () { - 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) return { threadId, imported: false } 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* random.uuidV4()); - 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, - }); - } - 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({ @@ -1764,7 +1623,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, @@ -1797,8 +1656,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({ @@ -1812,8 +1671,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", @@ -1854,8 +1713,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, @@ -2429,7 +2288,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, @@ -2454,7 +2313,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({ @@ -3610,6 +3471,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)), diff --git a/apps/web/src/components/settings/AcpSessionManagementSection.tsx b/apps/web/src/components/settings/AcpSessionManagementSection.tsx index d187633aa3172..c7085f5b4b44a 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); @@ -223,7 +227,7 @@ function InstanceSessionManagement( setSessions((current) => current.map((candidate) => candidate.sessionId === session.sessionId - ? { ...candidate, importedThreadId: result.value.threadId } + ? { ...candidate, importedThreadId: result.value.threadId, nativeHistoryPending: false } : candidate, ), ); @@ -472,12 +476,12 @@ function InstanceSessionManagement( 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 a2c2c429fa88c..a57899fb1472e 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" && !current.reasoningIds.has(messageId) + ? 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; }); @@ -8575,7 +8572,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); } }), ); @@ -8733,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, @@ -9237,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, ); @@ -9275,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 e925f10c8f154..98f71a7473a84 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -540,6 +540,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; }