1import type {
2 EnvironmentId,
3 ScopedThreadRef,
4 ThreadId,
5 OrchestrationV2ThreadProjection,
6 OrchestrationV2RuntimeRequest,
7 OrchestrationV2UserInputQuestion,
8 ProviderApprovalFullAccessOverride,
9 ProviderApprovalOption,
10 ProviderApprovalDecision,
11 ProviderRequestKind,
12 RuntimeRequestId,
13} from "@t3tools/contracts";
14import { Atom } from "effect/reactivity";
15import { parseThreadKey, threadKey } from "./entities.ts";
16import { getProjectionArrayReplacementIndex } from "./orchestrationV2Projection.ts";
17import type { EnvironmentThreadShell } from "./models.ts";
18import * as DateTime from "effect/DateTime";
19import type { AtomCommandResult } from "./runtime.ts";
20import type { ThreadStructure, ThreadStructureVisibleTurnItem } from "./models.ts";
21
22export type ApprovalResponder = (
23 requestId: RuntimeRequestId,
24 decision: ProviderApprovalDecision,
25 feedback?: string,
26) => Promise<AtomCommandResult<unknown, unknown> | undefined>;
27
28/** Feedback belongs to the displayed request, which may be a native child's tool call. */
29export async function sendApprovalFeedback(
30 approval: ThreadPendingApproval,
31 feedback: string,
32 respond: ApprovalResponder,
33): Promise<boolean> {
34 if (!approval.supportsFeedback || approval.responseCapability !== "live" || !feedback.trim())
35 return false;
36 return (await respond(approval.requestId, "decline", feedback.trim()))?._tag === "Success";
37}
38
39/** Declines every approval a redirect replaces; any one still pending would block the agent. */
40export async function declinePendingApprovals(
41 approvals: ReadonlyArray<ThreadPendingApproval>,
42 respond: ApprovalResponder,
43): Promise<boolean> {
44 for (const approval of approvals) {
45 // A request the provider can no longer answer has nothing to decline.
46 if (approval.responseCapability !== "live") continue;
47 if ((await respond(approval.requestId, "decline"))?._tag !== "Success") return false;
48 }
49 return true;
50}
51
52export interface ThreadPendingApproval {
53 readonly requestId: RuntimeRequestId;
54 readonly requestKind: ProviderRequestKind;
55 readonly createdAt: string;
56 readonly detail?: string;
57 readonly reason?: string;
58 readonly fullAccessOverride?: ProviderApprovalFullAccessOverride;
59 readonly appName?: string;
60 readonly options?: ReadonlyArray<ProviderApprovalOption>;
61 readonly responseCapability: "live" | "not_resumable";
62 readonly supportsFeedback?: boolean;
63}
64
65export const FULL_ACCESS_OVERRIDE_QUESTION = "Why am I being asked this?";
66
67export function fullAccessOverrideExplanation(
68 override: ProviderApprovalFullAccessOverride,
69): string {
70 return override === "ask_rule"
71 ? "Full access skips approvals, but an ask rule in the agent's own settings matches this action, so it still asks. Remove or narrow that rule to stop these prompts."
72 : "Full access skips approvals, but the agent still stops for a few safety checks of its own that no setting can skip, such as a delete whose path it can't verify. T3 Code can't turn these off, so check the request and approve it if it's what you expect.";
73}
74
75export interface ThreadUserInputQuestion extends Omit<
76 OrchestrationV2UserInputQuestion,
77 "multiSelect"
78> {
79 readonly multiSelect: boolean;
80}
81
82export interface ThreadPendingUserInput {
83 readonly requestId: RuntimeRequestId;
84 readonly createdAt: string;
85 readonly questions: ReadonlyArray<ThreadUserInputQuestion>;
86 readonly responseCapability: OrchestrationV2RuntimeRequest["responseCapability"]["type"];
87 readonly responseMode?: "message";
88 readonly dismissible: boolean;
89}
90
91export interface PendingThreadRequests {
92 readonly approvals: ReadonlyArray<ThreadPendingApproval>;
93 readonly userInputs: ReadonlyArray<ThreadPendingUserInput>;
94}
95
96export function seedUserInputDraftAnswers<Draft extends { readonly customAnswer?: string }>(
97 questions: ReadonlyArray<
98 Pick<OrchestrationV2UserInputQuestion, "id" | "allowCustomAnswer" | "initialAnswer">
99 >,
100 drafts: Record<string, Draft>,
101): Record<string, Draft | { customAnswer: string }> {
102 // Entries, not assignment, so an id such as `__proto__` becomes an own draft.
103 const seeds = questions.flatMap((question) =>
104 question.initialAnswer === undefined ||
105 question.allowCustomAnswer === false ||
106 Object.hasOwn(drafts, question.id)
107 ? []
108 : [[question.id, { customAnswer: question.initialAnswer }] as const],
109 );
110 return seeds.length === 0 ? drafts : { ...drafts, ...Object.fromEntries(seeds) };
111}
112
113/** Keys a user-input draft by thread so a thread's resolved requests can be pruned together. */
114export function userInputDraftKey(
115 environmentId: string,
116 threadId: string | null,
117 requestId: string | undefined,
118): string {
119 return JSON.stringify([environmentId, threadId, requestId]);
120}
121
122/**
123 * Drops the thread's drafts for requests outside `pendingRequestIds`, leaving
124 * other threads alone. Returns the input itself when nothing is dropped.
125 */
126export function pruneResolvedUserInputDrafts<T>(
127 draftsByRequestKey: Record<string, T>,
128 environmentId: string,
129 threadId: string,
130 pendingRequestIds: ReadonlyArray<string>,
131): Record<string, T> {
132 const threadPrefix = JSON.stringify([environmentId, threadId]).slice(0, -1);
133 const pendingKeys = new Set(
134 pendingRequestIds.map((requestId) => userInputDraftKey(environmentId, threadId, requestId)),
135 );
136 const retained = Object.entries(draftsByRequestKey).filter(
137 ([key]) => !key.startsWith(threadPrefix) || pendingKeys.has(key),
138 );
139 return retained.length === Object.keys(draftsByRequestKey).length
140 ? draftsByRequestKey
141 : Object.fromEntries(retained);
142}
143
144/**
145 * Question attachment drafts are keyed `<thread prefix><encoded [requestId, questionId]>`.
146 * Returns the keys under `threadPrefix` whose request is outside `pendingRequestIds`.
147 */
148export function resolvedQuestionDraftKeys(
149 keys: Iterable<string>,
150 threadPrefix: string,
151 pendingRequestIds: ReadonlyArray<string>,
152): string[] {
153 // encodeURIComponent is per character, so an encoded JSON prefix prefixes the encoded key.
154 const pendingPrefixes = pendingRequestIds.map(
155 (requestId) =>
156 `${threadPrefix}${encodeURIComponent(`${JSON.stringify([requestId]).slice(0, -1)},`)}`,
157 );
158 const resolved: string[] = [];
159 for (const key of keys) {
160 if (key.startsWith(threadPrefix) && !pendingPrefixes.some((prefix) => key.startsWith(prefix)))
161 resolved.push(key);
162 }
163 return resolved;
164}
165
166export interface ThreadRequestsResolved {
167 readonly environmentId: EnvironmentId;
168 readonly threadId: ThreadId;
169 /** Every request the thread still waits on; drafts for any other request are stale. */
170 readonly pendingRequestIds: ReadonlyArray<RuntimeRequestId>;
171 /** Requests that closed without an answer (cancelled or expired), so typed drafts were never sent. */
172 readonly unansweredRequestIds: ReadonlyArray<RuntimeRequestId>;
173}
174
175const threadRequestsResolvedListeners = new Set<(event: ThreadRequestsResolved) => void>();
176
177/**
178 * Clients register draft pruners once at module level. Thread state announces
179 * when its synchronized projection first lands and whenever a request leaves
180 * pending, wherever the resolution came from (another device, interrupt, turn end).
181 */
182export function onThreadRequestsResolved(
183 listener: (event: ThreadRequestsResolved) => void,
184): () => void {
185 threadRequestsResolvedListeners.add(listener);
186 return () => threadRequestsResolvedListeners.delete(listener);
187}
188
189function notifyThreadRequestsResolved(event: ThreadRequestsResolved): void {
190 for (const listener of threadRequestsResolvedListeners) {
191 try {
192 listener(event);
193 } catch (error) {
194 // A failing pruner must neither stop the others nor break thread sync.
195 queueMicrotask(() => {
196 throw error;
197 });
198 }
199 }
200}
201
202/**
203 * Tracks one thread's announced pending requests. Feed it the runtime requests
204 * of each authoritative projection; it notifies on the first one and whenever an
205 * announced request is no longer pending. New requests alone stay quiet.
206 */
207export function makeThreadRequestsResolvedAnnouncer(
208 environmentId: EnvironmentId,
209 threadId: ThreadId,
210) {
211 let announced: ReadonlySet<RuntimeRequestId> | null = null;
212 return (
213 runtimeRequests: ReadonlyArray<
214 Pick<OrchestrationV2ThreadProjection["runtimeRequests"][number], "id" | "status">
215 >,
216 ) => {
217 const pendingRequestIds = runtimeRequests.flatMap((request) =>
218 request.status === "pending" ? [request.id] : [],
219 );
220 const pending = new Set(pendingRequestIds);
221 const resolvedSome = announced === null || [...announced].some((id) => !pending.has(id));
222 announced = pending;
223 if (!resolvedSome) return;
224 const unansweredRequestIds = runtimeRequests.flatMap((request) =>
225 request.status === "cancelled" || request.status === "expired" ? [request.id] : [],
226 );
227 notifyThreadRequestsResolved({
228 environmentId,
229 threadId,
230 pendingRequestIds,
231 unansweredRequestIds,
232 });
233 };
234}
235
236export function derivePendingThreadRequests(
237 projection: Pick<ThreadStructure, "runtimeRequests" | "turnItems">,
238): PendingThreadRequests {
239 const approvals: ThreadPendingApproval[] = [];
240 const userInputs: ThreadPendingUserInput[] = [];
241
242 for (const request of projection.runtimeRequests) {
243 if (request.status !== "pending") continue;
244 const responseCapability = request.responseCapability.type;
245 if (request.kind === "user_input") {
246 const item = projection.turnItems.findLast(
247 (candidate) =>
248 candidate.type === "user_input_request" && candidate.requestId === request.id,
249 );
250 if (item === undefined || item.type !== "user_input_request") continue;
251 userInputs.push({
252 requestId: request.id,
253 createdAt: DateTime.formatIso(request.createdAt),
254 questions: item.questions.map((question) => ({
255 ...question,
256 multiSelect: question.multiSelect ?? false,
257 })),
258 responseCapability,
259 dismissible: item.responseMode === "message" || responseCapability === "message",
260 ...(item.responseMode === "message" || responseCapability === "message"
261 ? { responseMode: "message" as const }
262 : {}),
263 });
264 continue;
265 }
266
267 if (request.kind === "auth_refresh" || request.kind === "dynamic_tool_call") continue;
268 const item = projection.turnItems.findLast(
269 (candidate) => candidate.type === "approval_request" && candidate.requestId === request.id,
270 );
271 approvals.push({
272 requestId: request.id,
273 requestKind: request.kind,
274 createdAt: DateTime.formatIso(request.createdAt),
275 ...(item?.type === "approval_request" && item.prompt ? { detail: item.prompt } : {}),
276 ...(item?.type === "approval_request" && item.reason ? { reason: item.reason } : {}),
277 ...(item?.type === "approval_request" && item.fullAccessOverride
278 ? { fullAccessOverride: item.fullAccessOverride }
279 : {}),
280 ...(item?.type === "approval_request" && item.appName ? { appName: item.appName } : {}),
281 ...(item?.type === "approval_request" && item.options !== undefined
282 ? { options: item.options }
283 : {}),
284 responseCapability: responseCapability === "live" ? "live" : "not_resumable",
285 ...(request.responseCapability.type === "live" && request.responseCapability.supportsFeedback
286 ? { supportsFeedback: true }
287 : {}),
288 });
289 }
290
291 return { approvals, userInputs };
292}
293
294export function createQuestionHistoryProjector<
295 Row extends ThreadStructureVisibleTurnItem =
296 OrchestrationV2ThreadProjection["visibleTurnItems"][number],
297>() {
298 type Rows = ReadonlyArray<Row>;
299 type Requests = OrchestrationV2ThreadProjection["runtimeRequests"];
300 let previousRows: Rows | undefined;
301 let previousRequests: Requests | undefined;
302 let previousResult: Rows | undefined;
303 let byId = new Map<RuntimeRequestId, Requests[number]>();
304 const cache = new WeakMap<Row, { request: Requests[number]; row: Row }>();
305 const projectRow = (row: Row): Row => {
306 const item = row.item;
307 if (item.type !== "user_input_request" || item.questionAnswer) return row;
308 const request = byId.get(item.requestId);
309 if (request?.status !== "resolved" || request.answers === undefined) return row;
310 const cached = cache.get(row);
311 const next: Row =
312 cached?.request === request
313 ? cached.row
314 : {
315 ...row,
316 item: {
317 ...item,
318 questionAnswer: {
319 requestId: request.id,
320 answers: request.answers,
321 attachmentsByQuestionId: {},
322 questionTextById: Object.fromEntries(
323 item.questions.map((question) => [question.id, question.question]),
324 ),
325 },
326 },
327 };
328 cache.set(row, { request, row: next });
329 return next;
330 };
331 const projectRows = (rows: Rows): Rows => {
332 let result: Row[] | undefined;
333 for (const [index, row] of rows.entries()) {
334 const next = projectRow(row);
335 if (next !== row) result ??= rows.slice(0, index);
336 result?.push(next);
337 }
338 return result ?? rows;
339 };
340 return (projection: {
341 readonly visibleTurnItems: Rows;
342 readonly runtimeRequests: Requests;
343 }): Rows => {
344 const { visibleTurnItems: rows, runtimeRequests: requests } = projection;
345 const sameRequests = requests === previousRequests;
346 if (rows === previousRows && sameRequests && previousResult) return previousResult;
347 if (!sameRequests) byId = new Map(requests.map((request) => [request.id, request]));
348 // Nothing was overlaid and requests are unchanged, so only an exactly replaced row can differ.
349 const replacedIndex =
350 sameRequests && previousRows !== undefined && previousResult === previousRows
351 ? getProjectionArrayReplacementIndex(previousRows, rows)
352 : undefined;
353 const unchanged =
354 replacedIndex !== undefined && projectRow(rows[replacedIndex]!) === rows[replacedIndex];
355 previousRows = rows;
356 previousRequests = requests;
357 previousResult = unchanged ? rows : projectRows(rows);
358 return previousResult;
359 };
360}
361
362export interface ChildThreadApproval {
363 readonly thread: ScopedThreadRef;
364 readonly title: string;
365 readonly approval: ThreadPendingApproval;
366}
367
368export function createChildThreadApprovalsAtom(input: {
369 readonly childThreadInputsAtom: (
370 ref: ScopedThreadRef,
371 ) => Atom.Atom<ReadonlyArray<EnvironmentThreadShell>>;
372 readonly pendingRequestsAtom: (ref: ScopedThreadRef) => Atom.Atom<PendingThreadRequests | null>;
373}) {
374 const family = Atom.family((key: string) => {
375 let previous: ReadonlyArray<ChildThreadApproval> = [];
376 return Atom.make((get) => {
377 const next = get(input.childThreadInputsAtom(parseThreadKey(key))).flatMap((child) => {
378 if (!child.hasPendingApprovals) return [];
379 const thread = { environmentId: child.environmentId, threadId: child.id };
380 // Only children asking for approval need a detail subscription.
381 return (get(input.pendingRequestsAtom(thread))?.approvals ?? []).map((approval) => ({
382 thread,
383 title: child.title,
384 approval,
385 }));
386 });
387 if (
388 next.length === previous.length &&
389 next.every((entry, index) => {
390 const old = previous[index]!;
391 return (
392 old.approval === entry.approval &&
393 old.title === entry.title &&
394 threadKey(old.thread) === threadKey(entry.thread)
395 );
396 })
397 )
398 return previous;
399 previous = next;
400 return next;
401 });
402 });
403 const empty = Atom.make<ReadonlyArray<ChildThreadApproval>>([]);
404 return (ref: ScopedThreadRef | null) => (ref === null ? empty : family(threadKey(ref)));
405}