1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import {
5 ProjectId,
6 type OrchestrationV2ServerCommand,
7 ThreadId,
8 TurnItemId,
9} from "@t3tools/contracts";
10import * as Clock from "effect/Clock";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Metric from "effect/Metric";
14import * as TestClock from "effect/testing/TestClock";
15import * as Tracer from "effect/Tracer";
16import * as Option from "effect/Option";
17import * as PlatformError from "effect/PlatformError";
18
19import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
20import * as ServerConfig from "../config.ts";
21import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
22import * as SecretRequests from "./SecretRequests.ts";
23
24const threadId = ThreadId.make("thread-orchestrator");
25const turnItemId = TurnItemId.make("turn-item:secret-request:1");
26const projectId = ProjectId.make("project-1");
27
28/** Runs `body` against the service with an in-memory store and a thread holding one request. */
29const withService = <A, E>(
30 body: (input: {
31 readonly service: SecretRequests.SecretRequests["Service"];
32 readonly stored: Map<string, Uint8Array>;
33 readonly dispatched: Array<OrchestrationV2ServerCommand>;
34 }) => Effect.Effect<A, E>,
35 options: {
36 readonly threadId?: ThreadId;
37 readonly runStatus?: string;
38 readonly removeFails?: boolean;
39 /** The agent's wait closes the card just before the answer's record lands. */
40 readonly closedFirst?: boolean;
41 /** How many record dispatches fail before one succeeds. */
42 readonly failedRecords?: number;
43 } = {},
44) =>
45 Effect.gen(function* () {
46 const stored = new Map<string, Uint8Array>();
47 const dispatched: Array<OrchestrationV2ServerCommand> = [];
48 let secretStatus = "pending";
49 let failedRecords = options.failedRecords ?? 0;
50 const requestThreadId = options.threadId ?? threadId;
51 const layerDependencies = Layer.mergeAll(
52 NodeCrypto.layer,
53 NodeServices.layer,
54 Layer.succeed(
55 ServerSecretStore.ServerSecretStore,
56 ServerSecretStore.ServerSecretStore.of({
57 // Yields like a real file read, so concurrent callers can interleave.
58 get: (name) => Effect.yieldNow.pipe(Effect.as(Option.fromNullishOr(stored.get(name)))),
59 set: (name, value) => Effect.sync(() => void stored.set(name, value)),
60 create: (name, value) =>
61 stored.has(name)
62 ? Effect.fail(
63 new ServerSecretStore.SecretStorePersistError({
64 resource: name,
65 cause: new PlatformError.PlatformError(
66 new PlatformError.SystemError({
67 _tag: "AlreadyExists",
68 module: "FileSystem",
69 method: "open",
70 }),
71 ),
72 } as never),
73 )
74 : Effect.sync(() => void stored.set(name, value)),
75 getOrCreateRandom: (name, bytes) =>
76 Effect.sync(() => {
77 const existing = stored.get(name);
78 if (existing) return existing;
79 const value = new Uint8Array(bytes).fill(7);
80 stored.set(name, value);
81 return value;
82 }),
83 remove: (name) =>
84 options.removeFails
85 ? Effect.fail(
86 new ServerSecretStore.SecretStorePersistError({
87 resource: name,
88 cause: new Error("read-only"),
89 }),
90 )
91 : Effect.sync(() => void stored.delete(name)),
92 }),
93 ),
94 Layer.mock(ThreadManagementService.ThreadManagementService)({
95 getThreadRecords: () =>
96 Effect.succeed({
97 thread: { projectId },
98 runs: [{ id: "run-1", status: options.runStatus ?? "running" }],
99 turnItems: [
100 {
101 id: turnItemId,
102 threadId: requestThreadId,
103 runId: "run-1",
104 nodeId: "node-root",
105 type: "secret_request",
106 label: "GitHub token",
107 reason: "Used as GH_TOKEN.",
108 secretStatus,
109 },
110 ],
111 } as never),
112 dispatch: (command) => {
113 if (command.type === "secret_request.record" && failedRecords > 0) {
114 failedRecords -= 1;
115 return Effect.fail(new Error("orchestrator unavailable") as never);
116 }
117 return Effect.sync(() => {
118 dispatched.push(command);
119 // Like the orchestrator, a card that is no longer pending keeps its answer.
120 if (options.closedFirst && command.type === "secret_request.record") {
121 secretStatus = "cancelled";
122 } else if (command.type === "secret_request.record" && secretStatus === "pending") {
123 secretStatus = command.secretStatus;
124 }
125 return {} as never;
126 });
127 },
128 }),
129 );
130 return yield* Effect.gen(function* () {
131 const service = yield* SecretRequests.SecretRequests;
132 return yield* body({ service, stored, dispatched });
133 }).pipe(Effect.provide(SecretRequests.layer.pipe(Layer.provide(layerDependencies))));
134 });
135
136/** The secret values in the store, leaving out the server's own salt. */
137const valuesOf = (stored: Map<string, Uint8Array>) =>
138 Array.from(stored.entries())
139 .filter(([name]) => name !== "secret-request-salt")
140 .map(([, bytes]) => new TextDecoder().decode(bytes));
141
142it.effect("a saved answer becomes a one-use ref, and the thread only learns it was saved", () =>
143 withService(({ service, stored, dispatched }) =>
144 Effect.gen(function* () {
145 yield* service.answer({
146 threadId,
147 turnItemId,
148 answer: { type: "save", secret: "ghp_secret" },
149 });
150 assert.equal(dispatched.length, 1);
151 assert.include(dispatched[0] as object, {
152 type: "secret_request.record",
153 secretStatus: "saved",
154 });
155 assert.notInclude(Object.values(dispatched[0] as object).map(String), "ghp_secret");
156
157 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
158 assert.equal(yield* service.consume({ ref, projectId }), "ghp_secret");
159 // Used once: the value is gone from the store and the ref fails.
160 assert.isFalse(valuesOf(stored).some((value) => value.includes("ghp_secret")));
161 const again = yield* service.consume({ ref, projectId }).pipe(Effect.flip);
162 assert.include(again.message, "already used");
163 }),
164 ),
165);
166
167it.effect("two concurrent uses of one ref hand the value out once", () =>
168 withService(({ service }) =>
169 Effect.gen(function* () {
170 yield* service.answer({
171 threadId,
172 turnItemId,
173 answer: { type: "save", secret: "ghp_secret" },
174 });
175 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
176 const results = yield* Effect.all(
177 [service.consume({ ref, projectId }), service.consume({ ref, projectId })].map(
178 Effect.result,
179 ),
180 { concurrency: "unbounded" },
181 );
182 assert.deepEqual(results.map((result) => result._tag).toSorted(), ["Failure", "Success"]);
183 }),
184 ),
185);
186
187it.effect("a save that loses to the card closing deletes the value and says so", () =>
188 withService(
189 ({ service, stored }) =>
190 Effect.gen(function* () {
191 const error = yield* service
192 .answer({ threadId, turnItemId, answer: { type: "save", secret: "ghp_secret" } })
193 .pipe(Effect.flip);
194 assert.equal(error.reason, "agent_stopped");
195 assert.isFalse(valuesOf(stored).some((value) => value.includes("ghp_secret")));
196 }),
197 { closedFirst: true },
198 ),
199);
200
201it.effect("a save whose record failed can be saved again", () =>
202 withService(
203 ({ service }) =>
204 Effect.gen(function* () {
205 const failed = yield* service
206 .answer({ threadId, turnItemId, answer: { type: "save", secret: "ghp_secret" } })
207 .pipe(Effect.flip);
208 assert.equal(failed.reason, "record_failed");
209 yield* service.answer({
210 threadId,
211 turnItemId,
212 answer: { type: "save", secret: "ghp_secret" },
213 });
214 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
215 assert.equal(yield* service.consume({ ref, projectId }), "ghp_secret");
216 }),
217 { failedRecords: 1 },
218 ),
219);
220
221it.effect("a save whose record and cleanup both failed is finished by saving again", () =>
222 withService(
223 ({ service }) =>
224 Effect.gen(function* () {
225 yield* service
226 .answer({ threadId, turnItemId, answer: { type: "save", secret: "ghp_secret" } })
227 .pipe(Effect.flip);
228 yield* service.answer({
229 threadId,
230 turnItemId,
231 answer: { type: "save", secret: "ghp_secret" },
232 });
233 assert.isTrue(Option.isSome(yield* service.savedRef({ threadId, turnItemId })));
234 }),
235 { failedRecords: 1, removeFails: true },
236 ),
237);
238
239it.effect("a ref only works in the project it was entered for", () =>
240 withService(({ service }) =>
241 Effect.gen(function* () {
242 yield* service.answer({
243 threadId,
244 turnItemId,
245 answer: { type: "save", secret: "ghp_secret" },
246 });
247 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
248 const elsewhere = yield* service
249 .consume({ ref, projectId: ProjectId.make("project-other") })
250 .pipe(Effect.flip);
251 assert.include(elsewhere.message, "does not exist");
252 // A failed attempt from another project does not burn the ref.
253 assert.equal(yield* service.consume({ ref, projectId }), "ghp_secret");
254 }),
255 ),
256);
257
258it.effect("declining stores nothing, and a request is answered once", () =>
259 withService(({ service, stored, dispatched }) =>
260 Effect.gen(function* () {
261 yield* service.answer({ threadId, turnItemId, answer: { type: "decline" } });
262 assert.deepEqual(valuesOf(stored), []);
263 assert.isTrue(Option.isNone(yield* service.savedRef({ threadId, turnItemId })));
264 const late = yield* service
265 .answer({ threadId, turnItemId, answer: { type: "save", secret: "ghp_secret" } })
266 .pipe(Effect.flip);
267 assert.include(late.message, "already answered");
268 assert.equal(dispatched.length, 1);
269 assert.deepEqual(valuesOf(stored), []);
270 }),
271 ),
272);
273
274it.effect("traces and counts a saved answer without ever recording the value", () =>
275 Effect.gen(function* () {
276 const spans: Array<Tracer.NativeSpan> = [];
277 const tracer = Tracer.make({
278 span: (options) => {
279 const span = new Tracer.NativeSpan(options);
280 spans.push(span);
281 return span;
282 },
283 });
284 const counted = Metric.snapshot.pipe(
285 Effect.map((snapshots) => {
286 const found = snapshots.find(
287 (snapshot) =>
288 snapshot.id === "t3_secret_refs_consumed_total" &&
289 snapshot.attributes?.result === "used",
290 );
291 return found?.type === "Counter" ? Number(found.state.count) : 0;
292 }),
293 );
294 const before = yield* counted;
295 yield* withService(({ service }) =>
296 Effect.gen(function* () {
297 yield* service.answer({
298 threadId,
299 turnItemId,
300 answer: { type: "save", secret: "ghp_secret" },
301 });
302 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
303 yield* service.consume({ ref, projectId });
304 }),
305 ).pipe(Effect.withTracer(tracer));
306 assert.equal((yield* counted) - before, 1);
307 const recorded = spans.flatMap((span) => [
308 span.name,
309 ...Array.from(span.attributes.values(), String),
310 ]);
311 assert.include(recorded, "SecretRequests.answer");
312 assert.include(recorded, "SecretRequests.consume");
313 assert.isFalse(recorded.some((value) => value.includes("ghp_secret")));
314 }),
315);
316
317it.effect("works in threads with long ids, such as a delegated subagent's", () => {
318 const delegatedThreadId = ThreadId.make(
319 `thread:delegated-task:command%3Amcp%3A${"a".repeat(36)}%3Adelegate-task%3Arelease-notes-v0.2.0-${"b".repeat(40)}`,
320 );
321 return withService(
322 ({ service, stored }) =>
323 Effect.gen(function* () {
324 yield* service.answer({
325 threadId: delegatedThreadId,
326 turnItemId,
327 answer: { type: "save", secret: "ghp_secret" },
328 });
329 // Store names never grow with the thread id, so they fit any filesystem.
330 assert.isTrue(Array.from(stored.keys()).every((name) => name.length < 100));
331 const ref = Option.getOrThrow(
332 yield* service.savedRef({ threadId: delegatedThreadId, turnItemId }),
333 );
334 assert.equal(yield* service.consume({ ref, projectId }), "ghp_secret");
335 }),
336 { threadId: delegatedThreadId },
337 );
338});
339
340it.effect("refuses an answer once the agent that asked has stopped", () =>
341 withService(
342 ({ service, stored, dispatched }) =>
343 Effect.gen(function* () {
344 const late = yield* service
345 .answer({ threadId, turnItemId, answer: { type: "save", secret: "ghp_secret" } })
346 .pipe(Effect.flip);
347 assert.include(late.message, "has stopped");
348 assert.deepEqual(valuesOf(stored), []);
349 assert.equal(dispatched.length, 0);
350 }),
351 { runStatus: "completed" },
352 ),
353);
354
355it.effect("drops values nobody used once they expire, and keeps the rest", () =>
356 Effect.gen(function* () {
357 const store = yield* ServerSecretStore.ServerSecretStore;
358 const encode = (savedAt: number) =>
359 new TextEncoder().encode(
360 `{"projectId":"project-1","value":"ghp_secret","savedAt":${savedAt}}`,
361 );
362 yield* TestClock.adjust("2 days");
363 const now = yield* Clock.currentTimeMillis;
364 const stale = `secret-request-${"a".repeat(32)}`;
365 const fresh = `secret-request-${"b".repeat(32)}`;
366 yield* store.set(stale, encode(now - 25 * 60 * 60 * 1000));
367 yield* store.set(fresh, encode(now));
368 yield* store.set("unrelated", new Uint8Array([1]));
369
370 // Building the service runs the first sweep.
371 yield* Effect.gen(function* () {
372 yield* SecretRequests.SecretRequests;
373 }).pipe(
374 Effect.provide(
375 SecretRequests.layer.pipe(
376 Layer.provide(Layer.mock(ThreadManagementService.ThreadManagementService)({})),
377 ),
378 ),
379 Effect.scoped,
380 );
381
382 assert.isTrue(Option.isNone(yield* store.get(stale)));
383 assert.isTrue(Option.isSome(yield* store.get(fresh)));
384 assert.isTrue(Option.isSome(yield* store.get("unrelated")));
385 assert.isTrue(Option.isSome(yield* store.get("secret-request-salt")));
386 }).pipe(
387 Effect.provide(
388 ServerSecretStore.layer.pipe(
389 Layer.provideMerge(
390 ServerConfig.layerTest(process.cwd(), { prefix: "t3-secret-requests-" }),
391 ),
392 Layer.provideMerge(NodeServices.layer),
393 ),
394 ),
395 ),
396);
397
398it.effect("a used value that cannot be deleted is not handed out", () =>
399 withService(
400 ({ service }) =>
401 Effect.gen(function* () {
402 yield* service.answer({
403 threadId,
404 turnItemId,
405 answer: { type: "save", secret: "ghp_secret" },
406 });
407 const ref = Option.getOrThrow(yield* service.savedRef({ threadId, turnItemId }));
408 const failed = yield* service.consume({ ref, projectId }).pipe(Effect.flip);
409 assert.include(failed.message, "Could not use");
410 }),
411 { removeFails: true },
412 ),
413);