1import {
2 MessageId,
3 type ModelSelection,
4 NodeId,
5 ProjectId,
6 type ProviderInstanceId,
7 RunAttemptId,
8 RunId,
9 type OrchestrationV2ProviderThread,
10 type ThreadId,
11 ProviderDriverKind,
12 ProviderReplayEntry,
13 type ProviderReplayTranscript,
14} from "@t3tools/contracts";
15import type * as DateTime from "effect/DateTime";
16import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
17import * as Effect from "effect/Effect";
18import * as SecureRandom from "effect/crypto/SecureRandom";
19import * as FileSystem from "effect/FileSystem";
20import * as Layer from "effect/Layer";
21import * as Schema from "effect/Schema";
22import type * as Scope from "effect/Scope";
23import * as Stream from "effect/Stream";
24import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
25import * as EffectAcpErrors from "effect-acp/errors";
26
27import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime";
28import type { ProviderReplayGate } from "@t3tools/provider-testing/replayGate";
29import { ACP_PROTOCOL, type AcpAdapterV2RuntimeInput } from "@t3tools/provider-acp/server/adapter";
30
31export const AcpReplayTranscript = Schema.Struct({
32 provider: ProviderDriverKind,
33 protocol: Schema.Literal(ACP_PROTOCOL),
34 version: Schema.String,
35 scenario: Schema.String,
36 metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
37 entries: Schema.Array(ProviderReplayEntry),
38});
39export type AcpReplayTranscript = typeof AcpReplayTranscript.Type;
40
41const decodeAcpReplayTranscriptSchema = Schema.decodeUnknownEffect(AcpReplayTranscript);
42const encodeReplayTranscriptJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
43
44export class AcpReplayTranscriptDecodeError extends Schema.TaggedError<AcpReplayTranscriptDecodeError>()(
45 "AcpReplayTranscriptDecodeError",
46 {
47 expectedProvider: ProviderDriverKind,
48 driver: Schema.optional(Schema.String),
49 protocol: Schema.optional(Schema.String),
50 scenario: Schema.optional(Schema.String),
51 cause: Schema.Defect(),
52 },
53) {
54 override get message(): string {
55 return `Failed to decode ${this.expectedProvider} ACP replay transcript for scenario ${this.scenario ?? "<unknown>"}.`;
56 }
57}
58
59const isAcpReplayTranscriptDecodeError = Schema.is(AcpReplayTranscriptDecodeError);
60
61function metadataFromTranscript(transcript: ProviderReplayTranscript): {
62 readonly provider?: string;
63 readonly protocol?: string;
64 readonly scenario?: string;
65} {
66 return {
67 provider: transcript.provider,
68 protocol: transcript.protocol,
69 scenario: transcript.scenario,
70 };
71}
72
73export function decodeAcpReplayTranscript(
74 transcript: ProviderReplayTranscript,
75 expectedProvider: ProviderDriverKind,
76 options: { readonly retargetProvider?: boolean } = {},
77): Effect.Effect<AcpReplayTranscript, AcpReplayTranscriptDecodeError> {
78 const candidate = options.retargetProvider
79 ? { ...transcript, provider: expectedProvider }
80 : transcript;
81 return decodeAcpReplayTranscriptSchema(candidate).pipe(
82 Effect.filterOrFail(
83 (decoded) => decoded.provider === expectedProvider,
84 () =>
85 new AcpReplayTranscriptDecodeError({
86 expectedProvider,
87 ...metadataFromTranscript(transcript),
88 cause: `Expected provider ${expectedProvider}, received ${transcript.provider}`,
89 }),
90 ),
91 Effect.mapError((cause) =>
92 isAcpReplayTranscriptDecodeError(cause)
93 ? cause
94 : new AcpReplayTranscriptDecodeError({
95 expectedProvider,
96 ...metadataFromTranscript(transcript),
97 cause,
98 }),
99 ),
100 );
101}
102
103interface ReplayStatus {
104 readonly cursor: number;
105 readonly total: number;
106 readonly failure?: unknown;
107}
108
109function decodeReplayStatus(raw: string): ReplayStatus {
110 const value: unknown = JSON.parse(raw);
111 if (
112 typeof value !== "object" ||
113 value === null ||
114 typeof Reflect.get(value, "cursor") !== "number" ||
115 typeof Reflect.get(value, "total") !== "number"
116 ) {
117 throw new Error("ACP replay status is malformed.");
118 }
119 return value as ReplayStatus;
120}
121
122export function makeAcpReplayCompletenessAssertion(
123 fileSystem: FileSystem.FileSystem,
124 statusPath: string,
125 transcript: AcpReplayTranscript,
126): Effect.Effect<void, EffectAcpErrors.AcpError> {
127 return fileSystem.readFileString(statusPath).pipe(
128 Effect.mapError(
129 (cause) =>
130 new EffectAcpErrors.AcpTransportError({
131 detail: `Failed to read ACP replay status for ${transcript.scenario}`,
132 cause,
133 }),
134 ),
135 Effect.flatMap((raw) =>
136 Effect.try({
137 try: () => decodeReplayStatus(raw),
138 catch: (cause) =>
139 new EffectAcpErrors.AcpTransportError({
140 detail: `Failed to decode ACP replay status for ${transcript.scenario}`,
141 cause,
142 }),
143 }),
144 ),
145 Effect.flatMap((status) => {
146 if (
147 status.failure === undefined &&
148 status.cursor === transcript.entries.length &&
149 status.total === transcript.entries.length
150 ) {
151 return Effect.void;
152 }
153 return Effect.fail(
154 new EffectAcpErrors.AcpTransportError({
155 detail: `ACP replay did not consume all frames for ${transcript.scenario}`,
156 cause: status,
157 }),
158 );
159 }),
160 );
161}
162
163function acpReplayAgentArgs(scriptPath: string): ReadonlyArray<string> {
164 return ["--experimental-strip-types", scriptPath];
165}
166
167function holdGatedReplayLines(
168 transcript: AcpReplayTranscript,
169 replayGate: ProviderReplayGate,
170): NonNullable<AcpSessionRuntime.AcpSessionRuntimeOptions["transformStdout"]> {
171 const inboundLabels = transcript.entries.flatMap((entry) =>
172 entry.type === "emit_inbound" ? [entry.label] : [],
173 );
174 let inboundIndex = 0;
175 const encoder = new TextEncoder();
176 return (stdout) =>
177 stdout.pipe(
178 Stream.decodeText,
179 Stream.splitLines,
180 Stream.mapEffect((line) => {
181 const label = line.trim().length === 0 ? undefined : inboundLabels[inboundIndex++];
182 return Effect.promise((signal) => replayGate.beforeEmit(label, signal)).pipe(
183 Effect.as(encoder.encode(`${line}\n`)),
184 );
185 }),
186 );
187}
188
189export function makeAcpReplayRuntime(input: {
190 readonly transcript: AcpReplayTranscript;
191 readonly statusPath: string;
192 readonly scriptPath: string;
193 readonly childProcessSpawner: ChildProcessSpawner.ChildProcessSpawner["Service"];
194 readonly fileSystem: FileSystem.FileSystem;
195 readonly customModels?: AcpSessionRuntime.AcpSessionRuntimeOptions["customModels"];
196 readonly cancelMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelMeta"];
197 readonly cancelBehavior?: AcpSessionRuntime.AcpSessionRuntimeOptions["cancelBehavior"];
198 readonly initializeMeta?: AcpSessionRuntime.AcpSessionRuntimeOptions["initializeMeta"];
199 readonly replayGate?: ProviderReplayGate;
200}): (
201 runtimeInput: AcpAdapterV2RuntimeInput,
202) => Effect.Effect<
203 AcpSessionRuntime.AcpSessionRuntime["Service"],
204 EffectAcpErrors.AcpError,
205 SecureRandom.SecureRandom | Scope.Scope
206> {
207 // Transcripts exceed the kernel's 128 KiB per-environment-variable limit.
208 const transcriptPath = `${input.statusPath}.transcript.json`;
209 return (runtimeInput) =>
210 Effect.gen(function* () {
211 yield* input.fileSystem
212 .writeFileString(transcriptPath, encodeReplayTranscriptJson(input.transcript))
213 .pipe(
214 Effect.mapError(
215 (cause) =>
216 new EffectAcpErrors.AcpTransportError({
217 detail: `Failed to write ACP replay transcript for ${input.transcript.scenario}`,
218 cause,
219 }),
220 ),
221 );
222 const context = yield* Layer.build(
223 AcpSessionRuntime.layer({
224 ...runtimeInput,
225 spawn: {
226 command: process.execPath,
227 args: acpReplayAgentArgs(input.scriptPath),
228 cwd: runtimeInput.cwd,
229 env: {
230 ...process.env,
231 T3_ACP_REPLAY_TRANSCRIPT_PATH: transcriptPath,
232 T3_ACP_REPLAY_STATUS_PATH: input.statusPath,
233 T3_ACP_REPLAY_WORKSPACE: runtimeInput.cwd,
234 },
235 },
236 authMethodId: "replay",
237 ...(input.customModels === undefined ? {} : { customModels: input.customModels }),
238 ...(input.cancelMeta === undefined ? {} : { cancelMeta: input.cancelMeta }),
239 ...(input.cancelBehavior === undefined ? {} : { cancelBehavior: input.cancelBehavior }),
240 ...(input.initializeMeta === undefined ? {} : { initializeMeta: input.initializeMeta }),
241 ...(input.replayGate === undefined
242 ? {}
243 : { transformStdout: holdGatedReplayLines(input.transcript, input.replayGate) }),
244 }).pipe(
245 Layer.provide(
246 Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, input.childProcessSpawner),
247 ),
248 ),
249 );
250 return yield* Effect.service(AcpSessionRuntime.AcpSessionRuntime).pipe(
251 Effect.provide(context),
252 );
253 });
254}
255
256export function makeTurnInput(input: {
257 readonly threadId: ThreadId;
258 readonly providerThread: OrchestrationV2ProviderThread;
259 readonly instanceId: ProviderInstanceId;
260 readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
261 readonly now: DateTime.Utc;
262 readonly ordinal?: number;
263 readonly modelSelection?: ModelSelection;
264 readonly messageCreatedBy?: "user" | "agent";
265 readonly messageCreationSource?: "web" | "mobile" | "mcp" | "provider" | "server";
266 readonly messageText?: string;
267}): ProviderAdapter.ProviderAdapterV2TurnInput {
268 const ordinal = input.ordinal ?? 1;
269 const suffix = `${input.threadId}:${ordinal}`;
270 const modelSelection =
271 input.modelSelection ?? ({ instanceId: input.instanceId, model: "default" } as const);
272 return {
273 appThread: {
274 createdBy: "user",
275 creationSource: "web",
276 id: input.threadId,
277 projectId: ProjectId.make(`project:${input.threadId}`),
278 title: "ACP adapter test",
279 providerInstanceId: input.instanceId,
280 modelSelection,
281 runtimeMode: "approval-required",
282 interactionMode: "default",
283 branch: null,
284 worktreePath: null,
285 activeProviderThreadId: input.providerThread.id,
286 lineage: {
287 parentThreadId: null,
288 relationshipToParent: null,
289 rootThreadId: input.threadId,
290 },
291 forkedFrom: null,
292 createdAt: input.now,
293 updatedAt: input.now,
294 archivedAt: null,
295 settledOverride: null,
296 settledAt: null,
297 lastVisitedAt: null,
298 deletedAt: null,
299 },
300 threadId: input.threadId,
301 runId: RunId.make(`run:${suffix}`),
302 runOrdinal: ordinal,
303 providerTurnOrdinal: ordinal,
304 attemptId: RunAttemptId.make(`attempt:${suffix}`),
305 rootNodeId: NodeId.make(`node:${suffix}`),
306 providerThread: input.providerThread,
307 message: {
308 createdBy: input.messageCreatedBy ?? "user",
309 creationSource: input.messageCreationSource ?? "web",
310 messageId: MessageId.make(`message:${suffix}`),
311 text: input.messageText ?? "test prompt",
312 attachments: [],
313 },
314 modelSelection,
315 runtimePolicy: input.runtimePolicy,
316 };
317}