Open History to see the last commit
1import {
2 AcpRegistryOperationError,
3 CommandId,
4 type AcpRegistryImportSessionInput,
5 type AcpRegistryImportSessionResult,
6 type Project,
7 type ProjectId,
8 type ProviderDriverKind,
9 type ProviderInstanceId,
10} from "@t3tools/contracts";
11import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator";
12import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
13import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
14import * as Context from "effect/Context";
15import * as SecureRandom from "effect/crypto/SecureRandom";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Result from "effect/Result";
19
20import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
21import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
22import * as AgentSessionImporter from "../project/AgentSessionImporter.ts";
23import * as ProjectService from "../project/ProjectService.ts";
24import * as ServerRuntimeStartup from "../serverRuntimeStartup.ts";
25import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts";
26import * as ProviderRegistry from "./ProviderRegistry.ts";
27
28export const importedThreadId = (input: {
29 readonly driver: ProviderDriverKind;
30 readonly instanceId: ProviderInstanceId;
31 readonly sessionId: string;
32}) =>
33 IdAllocator.deriveThreadFromProviderThread({
34 driver: input.driver,
35 providerInstanceId: input.instanceId,
36 nativeThreadId: input.sessionId,
37 });
38
39export class AcpRegistrySessions extends Context.Service<
40 AcpRegistrySessions,
41 {
42 readonly project: (projectId: ProjectId) => Effect.Effect<Project, AcpRegistryOperationError>;
43 readonly sessionManager: (instanceId: ProviderInstanceId) => Effect.Effect<
44 {
45 readonly instance: ProviderInstance;
46 readonly manager: NonNullable<ProviderInstance["acpSessionManagement"]>;
47 },
48 AcpRegistryOperationError
49 >;
50 /**
51 * Binds a native ACP session to a T3 thread and copies its history in. Importing a
52 * session whose thread still lacks its history repairs that thread instead.
53 */
54 readonly importSession: (
55 input: AcpRegistryImportSessionInput,
56 ) => Effect.Effect<AcpRegistryImportSessionResult, AcpRegistryOperationError>;
57 }
58>()("t3/provider/AcpRegistrySessions") {}
59
60const make = Effect.gen(function* () {
61 const projectService = yield* ProjectService.ProjectService;
62 const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
63 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
64 const coordinator = yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
65 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
66 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;
67 const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup;
68 const agentSessionImporter = yield* AgentSessionImporter.AgentSessionImporter;
69 const random = yield* SecureRandom.SecureRandom;
70
71 const project = Effect.fn("AcpRegistrySessions.project")(function* (projectId: ProjectId) {
72 const found = yield* projectService.getById(projectId).pipe(
73 Effect.mapError(
74 (cause) =>
75 new AcpRegistryOperationError({
76 reason: "project_not_found",
77 message: `Project ${projectId} is unavailable.`,
78 cause,
79 }),
80 ),
81 );
82 return yield* Effect.fromOption(
83 found,
84 () =>
85 new AcpRegistryOperationError({
86 reason: "project_not_found",
87 message: `Project ${projectId} was not found.`,
88 }),
89 );
90 });
91
92 const sessionManager = Effect.fn("AcpRegistrySessions.sessionManager")(function* (
93 instanceId: ProviderInstanceId,
94 ) {
95 const instance = yield* providerInstances.getInstance(instanceId);
96 if (instance === undefined) {
97 return yield* new AcpRegistryOperationError({
98 reason: "instance_not_found",
99 message: `Provider instance ${instanceId} was not found.`,
100 });
101 }
102 if (instance.acpSessionManagement === undefined) {
103 return yield* new AcpRegistryOperationError({
104 reason: "session_list_unsupported",
105 message: `Provider instance ${instanceId} does not expose ACP session management.`,
106 });
107 }
108 return { instance, manager: instance.acpSessionManagement };
109 });
110
111 const importSession = Effect.fn("AcpRegistrySessions.importSession")(function* (
112 input: AcpRegistryImportSessionInput,
113 ) {
114 const { workspaceRoot } = yield* project(input.projectId);
115 const { instance } = yield* sessionManager(input.instanceId);
116 const providerSnapshot = yield* instance.snapshot.getSnapshot;
117 if (
118 providerSnapshot.nativeSessions?.canLoad !== true &&
119 providerSnapshot.nativeSessions?.canResume !== true
120 ) {
121 return yield* new AcpRegistryOperationError({
122 reason: "session_resume_unsupported",
123 message: "The ACP agent cannot load or resume native sessions.",
124 });
125 }
126 const threadId = importedThreadId({
127 driver: instance.driverKind,
128 instanceId: input.instanceId,
129 sessionId: input.sessionId,
130 });
131 const existing = yield* threadManagement.getThreadShell(threadId).pipe(
132 Effect.mapError(
133 (cause) =>
134 new AcpRegistryOperationError({
135 reason: "session_import_failed",
136 message: "Could not inspect the imported ACP session mapping.",
137 cause,
138 }),
139 ),
140 );
141 if (existing !== null) {
142 const repairing = AgentSessionImporter.isNativeHistoryPending({
143 historyOrigin: existing.historyOrigin,
144 hasRun: existing.latestRunId !== null,
145 });
146 yield* agentSessionImporter.importNativeHistory({ threadId, workspaceRoot }).pipe(
147 Effect.mapError(
148 (cause) =>
149 new AcpRegistryOperationError({
150 reason: "session_import_failed",
151 message: "Could not import the ACP session history.",
152 cause,
153 }),
154 ),
155 );
156 return { threadId, imported: repairing } as const;
157 }
158
159 const provider = (yield* providerRegistry.getProviders).find(
160 (candidate) => candidate.instanceId === input.instanceId,
161 );
162 const model =
163 provider?.models.find((candidate) => candidate.isDefault)?.slug ??
164 provider?.models[0]?.slug ??
165 "default";
166 const commandId = CommandId.make(yield* random.uuidV4());
167 const launched = yield* Effect.result(
168 startup.enqueueCommand(
169 threadLaunch.launch({
170 commandId,
171 threadId,
172 projectId: input.projectId,
173 title: input.title ?? "Imported ACP session",
174 modelSelection: { instanceId: input.instanceId, model },
175 runtimeMode: "approval-required",
176 interactionMode: "default",
177 workspaceStrategy: { type: "root" },
178 importedNativeThread: {
179 ref: {
180 driver: instance.driverKind,
181 nativeId: input.sessionId,
182 strength: "strong",
183 },
184 metadata: {
185 itemIdentityVersion: 2,
186 ...(input.title === undefined ? {} : { title: input.title }),
187 ...(input.updatedAt === undefined ? {} : { updatedAt: input.updatedAt }),
188 },
189 },
190 createdBy: "user",
191 creationSource: "web",
192 }),
193 ),
194 );
195 if (Result.isFailure(launched)) {
196 const racedImport = yield* threadManagement.getThreadShell(threadId).pipe(
197 Effect.mapError(
198 (cause) =>
199 new AcpRegistryOperationError({
200 reason: "session_import_failed",
201 message: "Could not inspect the imported ACP session after launch failed.",
202 cause,
203 }),
204 ),
205 );
206 if (racedImport !== null) return { threadId, imported: false } as const;
207 return yield* new AcpRegistryOperationError({
208 reason: "session_import_failed",
209 message: "Could not create a T3 thread for the ACP session.",
210 cause: launched.failure,
211 });
212 }
213 // The thread is usable without its history, so a failed read only logs.
214 yield* agentSessionImporter.importNativeHistory({ threadId, workspaceRoot }).pipe(
215 Effect.catch((cause) =>
216 Effect.logWarning("Could not import ACP session history", {
217 instanceId: input.instanceId,
218 sessionId: input.sessionId,
219 cause,
220 }),
221 ),
222 );
223 return { threadId, imported: true } as const;
224 }, coordinator.withSessionMutation);
225
226 return AcpRegistrySessions.of({ project, sessionManager, importSession });
227});
228
229export const layer = Layer.effect(AcpRegistrySessions, make);