1/**
2 * SecretRequests - secrets an agent asks the user for.
3 *
4 * The user's answer goes straight to the server's secret store under a
5 * one-use SecretRef; orchestration only records the request and its status.
6 * A tool that needs the value takes the ref and consumes it, so the value
7 * never reaches the transcript, projections, clients, or model context.
8 *
9 * @module SecretRequests
10 */
11import {
12 CommandId,
13 SecretRef,
14 SecretRequestError,
15 type ProjectId,
16 type SecretRequestAnswerInput,
17 type ThreadId,
18} from "@t3tools/contracts";
19// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no createHmac.
20import * as NodeCrypto from "node:crypto";
21
22import * as Clock from "effect/Clock";
23import * as Context from "effect/Context";
24import * as Effect from "effect/Effect";
25import * as FileSystem from "effect/FileSystem";
26import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";
28import * as Schedule from "effect/Schedule";
29import * as Schema from "effect/Schema";
30import * as Semaphore from "effect/Semaphore";
31
32import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
33import * as Metrics from "../observability/Metrics.ts";
34import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
35
36const SECRET_REF_PREFIX = "secret-ref:";
37/** Store name for a ref's value; refs are fixed-length hex, so names stay short and safe. */
38const storeName = (ref: SecretRef) => `secret-request-${ref.slice(SECRET_REF_PREFIX.length)}`;
39const REF_PATTERN = /^secret-ref:[0-9a-f]{32}$/;
40/** A value nobody used within this long is dropped; the agent can ask again. */
41const SECRET_REF_TTL_MS = 24 * 60 * 60 * 1000;
42const STORE_NAME_PATTERN = /^(secret-request-[0-9a-f]{32})\.bin$/;
43
44/**
45 * Each request has exactly one ref, derived from where it was asked. Its
46 * length never depends on the thread id, so the store's file names stay
47 * within filesystem limits, and the requesting tool finds it without a
48 * second record. Unguessable without the server's own salt.
49 */
50const refFor = (salt: string, threadId: ThreadId, turnItemId: string) =>
51 SecretRef.make(
52 `${SECRET_REF_PREFIX}${NodeCrypto.createHmac("sha256", salt)
53 .update(`${threadId}\u0000${turnItemId}`)
54 .digest("hex")
55 .slice(0, 32)}`,
56 );
57
58/** A ref's value, the project it was entered for, and when it was saved. */
59const StoredSecret = Schema.fromJsonString(
60 Schema.Struct({ projectId: Schema.String, value: Schema.String, savedAt: Schema.Number }),
61);
62const encodeStored = Schema.encodeEffect(StoredSecret);
63const decodeStored = Schema.decodeUnknownOption(StoredSecret);
64
65export class SecretRequests extends Context.Service<
66 SecretRequests,
67 {
68 /**
69 * Answers a pending request in a thread. Saving stores the value under a
70 * new ref that the requesting tool reads from the request's status.
71 */
72 readonly answer: (input: SecretRequestAnswerInput) => Effect.Effect<void, SecretRequestError>;
73 /** The ref minted when this request was saved, for the tool that asked. */
74 readonly savedRef: (input: {
75 readonly threadId: ThreadId;
76 readonly turnItemId: string;
77 }) => Effect.Effect<Option.Option<SecretRef>>;
78 /**
79 * Reads and deletes a ref's value. Fails for unknown or used refs, and
80 * for refs entered in another project.
81 */
82 readonly consume: (input: {
83 readonly ref: SecretRef;
84 readonly projectId: ProjectId;
85 }) => Effect.Effect<string, SecretRequestError>;
86 }
87>()("t3/secrets/SecretRequests") {}
88
89const make = Effect.gen(function* () {
90 const store = yield* ServerSecretStore.ServerSecretStore;
91 const fileSystem = yield* FileSystem.FileSystem;
92 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
93
94 const salt = Buffer.from(
95 yield* store.getOrCreateRandom("secret-request-salt", 32).pipe(Effect.orDie),
96 ).toString("hex");
97
98 const answer: SecretRequests["Service"]["answer"] = (input) =>
99 Effect.gen(function* () {
100 yield* Effect.annotateCurrentSpan({
101 "orchestration_v2.thread_id": input.threadId,
102 "secret_request.answer": input.answer.type,
103 });
104 const records = yield* threadManagement
105 .getThreadRecords(input.threadId, ["runs", "turnItems"], {
106 turnItemTypes: ["secret_request"],
107 messageRoles: [],
108 })
109 .pipe(Effect.mapError((cause) => new SecretRequestError({ reason: "load_failed", cause })));
110 const item = records.turnItems.find((candidate) => candidate.id === input.turnItemId);
111 if (item?.type !== "secret_request" || item.runId === null || item.nodeId === null) {
112 return yield* new SecretRequestError({ reason: "not_found" });
113 }
114 if (item.secretStatus !== "pending") {
115 return yield* new SecretRequestError({ reason: "already_answered" });
116 }
117 // The agent is waiting inside the run that asked; once it has ended,
118 // nobody will ever receive the ref, so a value saved now would be lost.
119 const run = records.runs.find((candidate) => candidate.id === item.runId);
120 if (run === undefined || ThreadManagementService.isTerminalRunStatus(run.status)) {
121 return yield* new SecretRequestError({ reason: "agent_stopped" });
122 }
123 // Store first: the card only says saved once the value is kept. Create,
124 // not set: a second answer racing this one must not replace the value.
125 // A value already there on a card still pending is an earlier save whose
126 // record failed and could not be cleaned up; recording it finishes that save.
127 if (input.answer.type === "save") {
128 const encoded = yield* encodeStored({
129 projectId: records.thread.projectId,
130 value: input.answer.secret,
131 savedAt: yield* Clock.currentTimeMillis,
132 }).pipe(Effect.orDie);
133 yield* store
134 .create(
135 storeName(refFor(salt, input.threadId, item.id)),
136 new TextEncoder().encode(encoded),
137 )
138 .pipe(
139 Effect.catchIf(ServerSecretStore.isSecretAlreadyExistsError, () => Effect.void),
140 Effect.mapError(
141 (error) => new SecretRequestError({ reason: "store_failed", cause: error }),
142 ),
143 );
144 }
145 const secretStatus = input.answer.type === "save" ? "saved" : "declined";
146 yield* threadManagement
147 .dispatch({
148 type: "secret_request.record",
149 commandId: CommandId.make(`secret-request:${item.id}:${secretStatus}`),
150 threadId: input.threadId,
151 runId: item.runId,
152 nodeId: item.nodeId,
153 turnItemId: item.id,
154 label: item.label,
155 reason: item.reason,
156 ...(item.placeholder === undefined ? {} : { placeholder: item.placeholder }),
157 secretStatus,
158 })
159 .pipe(
160 Effect.mapError((cause) => new SecretRequestError({ reason: "record_failed", cause })),
161 // The card still says pending, so the user can save again; the value
162 // stored above would make that retry look already answered.
163 Effect.tapError(() =>
164 input.answer.type === "save"
165 ? removeLogged(storeName(refFor(salt, input.threadId, item.id)))
166 : Effect.void,
167 ),
168 );
169 if (input.answer.type !== "save") return;
170 // A request is answered once: if the agent's wait closed the card between
171 // the checks above and this record, the record changed nothing. Nobody
172 // will receive the ref, so the value is deleted rather than left to expire.
173 const recorded = yield* threadManagement
174 .getThreadRecords(input.threadId, ["turnItems"], {
175 turnItemTypes: ["secret_request"],
176 messageRoles: [],
177 })
178 .pipe(
179 Effect.mapError((cause) => new SecretRequestError({ reason: "record_failed", cause })),
180 );
181 const card = recorded.turnItems.find((candidate) => candidate.id === item.id);
182 if (card?.type === "secret_request" && card.secretStatus !== "saved") {
183 yield* removeLogged(storeName(refFor(salt, input.threadId, item.id)));
184 return yield* new SecretRequestError({ reason: "agent_stopped" });
185 }
186 }).pipe(Effect.withSpan("SecretRequests.answer"));
187
188 const savedRef: SecretRequests["Service"]["savedRef"] = (input) =>
189 Effect.gen(function* () {
190 const ref = refFor(salt, input.threadId, input.turnItemId);
191 const stored = yield* store.get(storeName(ref)).pipe(Effect.orElseSucceed(Option.none));
192 return Option.map(stored, () => ref);
193 });
194
195 /** Removes a stored value; a failure is logged, since the value is still on disk. */
196 const removeLogged = (name: string) =>
197 store.remove(name).pipe(
198 Effect.as(true),
199 Effect.catch((error) =>
200 Effect.logWarning("Could not delete a secret request value", {
201 errorTag: error._tag,
202 }).pipe(Effect.as(false)),
203 ),
204 );
205
206 // get and remove are separate store calls; one consumer at a time keeps two
207 // concurrent calls from both reading a ref before either deletes it.
208 const consumeLock = yield* Semaphore.make(1);
209 const consume: SecretRequests["Service"]["consume"] = (input) =>
210 consumeRef(input).pipe(
211 consumeLock.withPermits(1),
212 Effect.tap(() => Metrics.increment(Metrics.secretRefsConsumedTotal, { result: "used" })),
213 Effect.tapError(() =>
214 Metrics.increment(Metrics.secretRefsConsumedTotal, { result: "rejected" }),
215 ),
216 Effect.withSpan("SecretRequests.consume"),
217 );
218
219 const consumeRef = (input: { readonly ref: SecretRef; readonly projectId: ProjectId }) =>
220 Effect.gen(function* () {
221 if (!REF_PATTERN.test(input.ref))
222 return yield* new SecretRequestError({ reason: "invalid_ref" });
223 const stored = yield* store
224 .get(storeName(input.ref))
225 .pipe(Effect.mapError((cause) => new SecretRequestError({ reason: "read_failed", cause })));
226 const decoded = Option.flatMap(stored, (bytes) =>
227 decodeStored(new TextDecoder().decode(bytes)),
228 );
229 if (Option.isNone(decoded) || decoded.value.projectId !== input.projectId) {
230 return yield* new SecretRequestError({ reason: "ref_unavailable" });
231 }
232 if ((yield* Clock.currentTimeMillis) - decoded.value.savedAt > SECRET_REF_TTL_MS) {
233 // The hourly sweep retries a removal that fails here.
234 yield* removeLogged(storeName(input.ref));
235 return yield* new SecretRequestError({ reason: "ref_expired" });
236 }
237 // One use: the value moves into whatever consumed it. If it cannot be
238 // deleted, it is not handed out, so a ref is never used twice.
239 if (!(yield* removeLogged(storeName(input.ref)))) {
240 return yield* new SecretRequestError({ reason: "consume_failed" });
241 }
242 return decoded.value.value;
243 });
244
245 /**
246 * Drops values nobody used before they expired, so an agent that never
247 * consumed its ref does not leave the user's secret on disk.
248 */
249 const sweepExpired = Effect.gen(function* () {
250 if (store.directory === undefined) return;
251 const now = yield* Clock.currentTimeMillis;
252 const names = (yield* fileSystem.readDirectory(store.directory)).flatMap((file) => {
253 const match = STORE_NAME_PATTERN.exec(file);
254 return match?.[1] === undefined ? [] : [match[1]];
255 });
256 let removed = 0;
257 for (const name of names) {
258 const stored = yield* store.get(name).pipe(Effect.orElseSucceed(Option.none));
259 const decoded = Option.flatMap(stored, (bytes) =>
260 decodeStored(new TextDecoder().decode(bytes)),
261 );
262 if (Option.isSome(decoded) && now - decoded.value.savedAt <= SECRET_REF_TTL_MS) continue;
263 if (yield* removeLogged(name)) removed += 1;
264 }
265 yield* Effect.annotateCurrentSpan({ "secret_request.expired_removed": removed });
266 }).pipe(
267 Effect.catchCause((cause) => Effect.logWarning("Could not sweep expired secret refs", cause)),
268 Effect.withSpan("SecretRequests.sweepExpired"),
269 );
270 // Once at startup, then hourly; a value lingers at most an hour past expiry.
271 yield* sweepExpired;
272 yield* sweepExpired.pipe(
273 Effect.delay("1 hour"),
274 Effect.repeat(Schedule.spaced("1 hour")),
275 Effect.forkScoped,
276 );
277
278 return SecretRequests.of({ answer, savedRef, consume });
279});
280
281export const layer = Layer.effect(SecretRequests, make);