Open History to see the last commit
1import { presentThreadShell } from "./models.ts";
2import { threadKey } from "./entities.ts";
3import {
4 EnvironmentId,
5 EventId,
6 MessageId,
7 ThreadId,
8 type ScopedThreadRef,
9 NodeId,
10 ProviderSessionId,
11 RuntimeRequestId,
12 TurnItemId,
13 type OrchestrationV2ThreadProjection,
14} from "@t3tools/contracts";
15import { describe, expect, it, vi } from "vite-plus/test";
16import { Atom, AtomRegistry, AsyncResult } from "effect/reactivity";
17import * as Cause from "effect/Cause";
18import { applyOrchestrationV2ProjectionEvent } from "./orchestrationV2Projection.ts";
19import { v2Now, v2Projection, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
20import {
21 createQuestionHistoryProjector,
22 createChildThreadApprovalsAtom,
23 type PendingThreadRequests,
24 derivePendingThreadRequests,
25 declinePendingApprovals,
26 sendApprovalFeedback,
27 makeThreadRequestsResolvedAnnouncer,
28 onThreadRequestsResolved,
29 pruneResolvedUserInputDrafts,
30 resolvedQuestionDraftKeys,
31 seedUserInputDraftAnswers,
32 userInputDraftKey,
33 type ApprovalResponder,
34 type ThreadRequestsResolved,
35 type ThreadPendingApproval,
36} from "./threadRequests.ts";
37
38const requestId = RuntimeRequestId.make("async-question");
39const nodeId = NodeId.make("async-question-node");
40
41describe("declining approvals before sending", () => {
42 const approval: ThreadPendingApproval = {
43 requestId,
44 requestKind: "command",
45 createdAt: "2026-10-19T00:00:00Z",
46 responseCapability: "live",
47 };
48 const second: ThreadPendingApproval = {
49 ...approval,
50 requestId: RuntimeRequestId.make("second-approval"),
51 };
52
53 it("declines every pending approval before the send continues", async () => {
54 let resolve!: (value: Awaited<ReturnType<ApprovalResponder>>) => void;
55 const firstResponse = new Promise<Awaited<ReturnType<ApprovalResponder>>>((done) => {
56 resolve = done;
57 });
58 const respond = vi.fn<ApprovalResponder>(async (id) =>
59 id === requestId ? firstResponse : AsyncResult.success(undefined),
60 );
61 const send = vi.fn();
62 const submission = (async () => {
63 if (await declinePendingApprovals([approval, second], respond)) send();
64 })();
65 expect(respond).toHaveBeenCalledWith(requestId, "decline");
66 expect(send).not.toHaveBeenCalled();
67 resolve(AsyncResult.success(undefined));
68 await submission;
69 expect(respond).toHaveBeenCalledWith(second.requestId, "decline");
70 expect(send).toHaveBeenCalledOnce();
71 });
72
73 it.each([AsyncResult.failure(Cause.fail("decline failed")), undefined])(
74 "does not send when a decline is unsuccessful: %s",
75 async (response) => {
76 const send = vi.fn();
77 if (await declinePendingApprovals([approval], async () => response)) send();
78 expect(send).not.toHaveBeenCalled();
79 },
80 );
81
82 it("skips approvals the provider can no longer answer", async () => {
83 const respond = vi.fn(async () => AsyncResult.success(undefined));
84 expect(
85 await declinePendingApprovals(
86 [{ ...approval, responseCapability: "not_resumable" }],
87 respond,
88 ),
89 ).toBe(true);
90 expect(respond).not.toHaveBeenCalled();
91 });
92
93 it("sends feedback only to the displayed request, leaving other approvals pending", async () => {
94 const respond = vi.fn<ApprovalResponder>(async () => AsyncResult.success(undefined));
95 const pending = derivePendingThreadRequests({
96 runtimeRequests: [
97 {
98 ...projection.runtimeRequests[0]!,
99 kind: "command",
100 responseCapability: {
101 type: "live",
102 providerSessionId: ProviderSessionId.make("claude-session"),
103 supportsFeedback: true,
104 },
105 },
106 { ...projection.runtimeRequests[0]!, id: second.requestId, kind: "command" },
107 ],
108 turnItems: [],
109 });
110 expect(await sendApprovalFeedback(pending.approvals[0]!, " Safer please. ", respond)).toBe(
111 true,
112 );
113 expect(respond.mock.calls).toEqual([[approval.requestId, "decline", "Safer please."]]);
114 });
115
116 it("keeps feedback unsent when unsupported, stale, empty or rejected", async () => {
117 const respond = vi.fn<ApprovalResponder>(async () => AsyncResult.failure(Cause.fail("failed")));
118 expect(await sendApprovalFeedback(approval, "Safer please.", respond)).toBe(false);
119 expect(
120 await sendApprovalFeedback(
121 { ...approval, supportsFeedback: true, responseCapability: "not_resumable" },
122 "Safer please.",
123 respond,
124 ),
125 ).toBe(false);
126 expect(await sendApprovalFeedback({ ...approval, supportsFeedback: true }, " ", respond)).toBe(
127 false,
128 );
129 expect(respond).not.toHaveBeenCalled();
130 expect(
131 await sendApprovalFeedback({ ...approval, supportsFeedback: true }, "Safer please.", respond),
132 ).toBe(false);
133 });
134});
135
136describe("pruning resolved user input drafts", () => {
137 const answered = userInputDraftKey("env", "thread", "answered");
138 const pending = userInputDraftKey("env", "thread", "pending");
139 const otherThread = userInputDraftKey("env", "thread-2", "answered");
140 const otherEnvironment = userInputDraftKey("env-2", "thread", "answered");
141 const drafts = {
142 [answered]: 1,
143 [pending]: 2,
144 [otherThread]: 3,
145 [otherEnvironment]: 4,
146 };
147
148 it("drops only the thread's requests that are no longer pending", () => {
149 expect(pruneResolvedUserInputDrafts(drafts, "env", "thread", ["pending"])).toEqual({
150 [pending]: 2,
151 [otherThread]: 3,
152 [otherEnvironment]: 4,
153 });
154 });
155
156 it("returns the same record when nothing resolved", () => {
157 expect(pruneResolvedUserInputDrafts(drafts, "env", "thread", ["answered", "pending"])).toBe(
158 drafts,
159 );
160 });
161});
162
163describe("resolved question draft keys", () => {
164 const prefix = "env:question-thread-";
165 const key = (request: string, question: string) =>
166 `${prefix}${encodeURIComponent(JSON.stringify([request, question]))}`;
167
168 it("returns the thread's keys for requests that are no longer pending", () => {
169 const keys = [
170 key("pending", "first"),
171 key("pending", "second"),
172 key("pending-2", "first"),
173 key("answered", "first"),
174 "env:question-other-thread-x",
175 ];
176 expect(resolvedQuestionDraftKeys(keys, prefix, [RuntimeRequestId.make("pending")])).toEqual([
177 key("pending-2", "first"),
178 key("answered", "first"),
179 ]);
180 });
181});
182
183describe("announcing resolved thread requests", () => {
184 const environmentId = EnvironmentId.make("env");
185 const threadId = v2Projection.thread.id;
186 const first = RuntimeRequestId.make("first");
187 const second = RuntimeRequestId.make("second");
188
189 function record() {
190 const events: ThreadRequestsResolved[] = [];
191 const unsubscribe = onThreadRequestsResolved((event) => events.push(event));
192 return { events, [Symbol.dispose]: unsubscribe };
193 }
194
195 it("announces the first projection, then only when a request leaves pending", () => {
196 using recorder = record();
197 const announce = makeThreadRequestsResolvedAnnouncer(environmentId, threadId);
198
199 const pending = (id: RuntimeRequestId) => ({ id, status: "pending" as const });
200 announce([pending(first)]);
201 announce([pending(first), pending(second)]);
202 announce([pending(first), pending(second)]);
203 announce([{ id: first, status: "cancelled" }, pending(second)]);
204 announce([
205 { id: first, status: "cancelled" },
206 { id: second, status: "resolved" },
207 ]);
208
209 expect(recorder.events).toEqual([
210 { environmentId, threadId, pendingRequestIds: [first], unansweredRequestIds: [] },
211 { environmentId, threadId, pendingRequestIds: [second], unansweredRequestIds: [first] },
212 { environmentId, threadId, pendingRequestIds: [], unansweredRequestIds: [first] },
213 ]);
214 });
215});
216
217describe("initial user input drafts", () => {
218 const questions = [
219 { id: "editor", initialAnswer: " Proposed commit\n\n" },
220 { id: "empty", initialAnswer: "" },
221 { id: "ordinary" },
222 { id: "choices", initialAnswer: "Ignore this", allowCustomAnswer: false },
223 ];
224
225 it("seeds editable answers and preserves edits and selections on later snapshots", () => {
226 const seeded = seedUserInputDraftAnswers(questions, {});
227 expect(seeded).toEqual({
228 editor: { customAnswer: " Proposed commit\n\n" },
229 empty: { customAnswer: "" },
230 });
231 const edits = {
232 ...seeded,
233 editor: { customAnswer: "" },
234 empty: { selectedOptionValues: ["empty"], customAnswer: "" },
235 };
236 expect(seedUserInputDraftAnswers(questions, edits)).toEqual(edits);
237 expect(
238 seedUserInputDraftAnswers(questions, { editor: { customAnswer: "Edited commit" } }).editor,
239 ).toEqual({ customAnswer: "Edited commit" });
240 expect(seedUserInputDraftAnswers(questions, {}).editor).toEqual({
241 customAnswer: " Proposed commit\n\n",
242 });
243 });
244
245 it("seeds a question with the id __proto__ as its own draft", () => {
246 const seeded = seedUserInputDraftAnswers(
247 [
248 { id: "__proto__", initialAnswer: "Proposed" },
249 { id: "name", initialAnswer: "Control" },
250 ],
251 {},
252 );
253
254 expect(JSON.stringify(seeded)).toBe(
255 '{"__proto__":{"customAnswer":"Proposed"},"name":{"customAnswer":"Control"}}',
256 );
257 });
258});
259const projection: OrchestrationV2ThreadProjection = {
260 ...v2Projection,
261 runtimeRequests: [
262 {
263 id: requestId,
264 nodeId,
265 providerTurnId: null,
266 nativeRequestRef: null,
267 kind: "user_input",
268 status: "pending",
269 responseCapability: { type: "message" },
270 createdAt: v2Now,
271 resolvedAt: null,
272 },
273 ],
274 turnItems: [
275 {
276 id: TurnItemId.make("async-question-item"),
277 threadId: v2Projection.thread.id,
278 runId: null,
279 nodeId,
280 providerThreadId: null,
281 providerTurnId: null,
282 nativeItemRef: null,
283 parentItemId: null,
284 ordinal: 0,
285 status: "completed",
286 title: null,
287 startedAt: v2Now,
288 completedAt: v2Now,
289 updatedAt: v2Now,
290 type: "user_input_request",
291 requestId,
292 responseMode: "message",
293 questions: [
294 {
295 id: "next",
296 header: "Next",
297 question: "What should happen next?",
298 required: true,
299 allowCustomAnswer: false,
300 options: [{ label: "Continue", value: " continue ", description: "Resume work" }],
301 },
302 ],
303 },
304 ],
305};
306
307describe("pending v2 questions", () => {
308 it("keeps message responses available after the originating runtime exits", () => {
309 expect(projection.providerSessions).toEqual([]);
310 expect(derivePendingThreadRequests(projection).userInputs).toEqual([
311 {
312 requestId,
313 createdAt: "2026-06-20T00:00:00.000Z",
314 responseCapability: "message",
315 responseMode: "message",
316 dismissible: true,
317 questions: [
318 {
319 id: "next",
320 header: "Next",
321 question: "What should happen next?",
322 required: true,
323 allowCustomAnswer: false,
324 multiSelect: false,
325 options: [{ label: "Continue", value: " continue ", description: "Resume work" }],
326 },
327 ],
328 },
329 ]);
330 });
331
332 it("only marks asynchronous questions as dismissible", () => {
333 const live = {
334 ...projection,
335 runtimeRequests: projection.runtimeRequests.map((request) => ({
336 ...request,
337 responseCapability: {
338 type: "live" as const,
339 providerSessionId: ProviderSessionId.make("live-session"),
340 },
341 })),
342 turnItems: projection.turnItems.map((item) =>
343 item.type === "user_input_request" ? { ...item, responseMode: undefined } : item,
344 ),
345 };
346 expect(derivePendingThreadRequests(live).userInputs[0]?.dismissible).toBe(false);
347 });
348});
349
350it("restores old text answers without mutating history or replacing unchanged rows", () => {
351 const project = createQuestionHistoryProjector();
352 const item = projection.turnItems[0]!;
353 const row = {
354 item,
355 position: 0,
356 sourceItemId: item.id,
357 sourceThreadId: item.threadId,
358 visibility: "local" as const,
359 };
360 const rows = [row];
361 const pending = { visibleTurnItems: rows, runtimeRequests: projection.runtimeRequests };
362 expect(project(pending)).toBe(rows);
363 const answered = {
364 ...pending,
365 runtimeRequests: [
366 {
367 ...projection.runtimeRequests[0]!,
368 status: "resolved" as const,
369 answers: { next: "Continue" },
370 },
371 ],
372 };
373 const result = project(answered);
374 expect(result[0]?.item).toMatchObject({
375 questionAnswer: {
376 answers: { next: "Continue" },
377 attachmentsByQuestionId: {},
378 questionTextById: { next: "What should happen next?" },
379 },
380 });
381 expect(row.item).not.toHaveProperty("questionAnswer");
382 expect(project(answered)).toBe(result);
383 expect(project({ ...answered, visibleTurnItems: [...rows] })[0]).toBe(result[0]);
384});
385
386type QuestionItem = Extract<
387 OrchestrationV2ThreadProjection["turnItems"][number],
388 { readonly type: "user_input_request" }
389>;
390
391describe("question history through exact row replacements", () => {
392 const baseItem = {
393 threadId: v2Projection.thread.id,
394 runId: null,
395 nodeId,
396 providerThreadId: null,
397 providerTurnId: null,
398 nativeItemRef: null,
399 parentItemId: null,
400 status: "running" as const,
401 title: null,
402 startedAt: v2Now,
403 completedAt: null,
404 updatedAt: v2Now,
405 };
406 const assistantItem = (ordinal: number, text = "Hel") => ({
407 ...baseItem,
408 id: TurnItemId.make(`assistant-${ordinal}`),
409 ordinal,
410 type: "assistant_message" as const,
411 messageId: MessageId.make(`message-${ordinal}`),
412 text,
413 streaming: true,
414 });
415 const localRow = (
416 item: OrchestrationV2ThreadProjection["turnItems"][number],
417 position: number,
418 ) => ({
419 item,
420 position,
421 sourceItemId: item.id,
422 sourceThreadId: item.threadId,
423 visibility: "local" as const,
424 });
425 const updated = (
426 current: OrchestrationV2ThreadProjection,
427 item: OrchestrationV2ThreadProjection["turnItems"][number],
428 ) =>
429 applyOrchestrationV2ProjectionEvent(current, {
430 id: EventId.make(`event-${item.id}`),
431 type: "turn-item.updated",
432 threadId: current.thread.id,
433 occurredAt: v2Now,
434 payload: item,
435 })!;
436
437 it("reads only the replaced row and leaves the request index alone", () => {
438 const size = 1_000;
439 const counts = { items: 0, requestIds: 0 };
440 const items = Array.from({ length: size }, (_, index) => assistantItem(index));
441 const countedRows = items.map((item, index) => {
442 const row = localRow(item, index);
443 return Object.defineProperty(row, "item", {
444 enumerable: true,
445 get: () => {
446 counts.items += 1;
447 return item;
448 },
449 });
450 });
451 const countedRequests = Array.from({ length: 400 }, (_, index) => {
452 const request = {
453 ...projection.runtimeRequests[0]!,
454 id: RuntimeRequestId.make(`request-${index}`),
455 };
456 return Object.defineProperty(request, "id", {
457 enumerable: true,
458 get: () => {
459 counts.requestIds += 1;
460 return RuntimeRequestId.make(`request-${index}`);
461 },
462 });
463 });
464 const initial: OrchestrationV2ThreadProjection = {
465 ...v2Projection,
466 turnItems: items,
467 visibleTurnItems: countedRows,
468 runtimeRequests: countedRequests,
469 };
470 const project = createQuestionHistoryProjector();
471 const scan = (current: OrchestrationV2ThreadProjection) => {
472 counts.items = 0;
473 counts.requestIds = 0;
474 const result = project(current);
475 return { result, ...counts };
476 };
477 expect(scan(initial)).toMatchObject({ items: size, requestIds: 400 });
478
479 const adjacent = updated(initial, assistantItem(size - 1, "Hello"));
480 const hinted = scan(adjacent);
481 expect(hinted.result).toBe(adjacent.visibleTurnItems);
482 expect(hinted).toMatchObject({ items: 0, requestIds: 0 });
483
484 // A skipped delta has no hint for this exact pair, so it scans again.
485 const skipped = updated(
486 updated(adjacent, assistantItem(size - 1, "Hello t")),
487 assistantItem(size - 1, "Hello there"),
488 );
489 const fallback = scan(skipped);
490 expect(fallback.result).toBe(skipped.visibleTurnItems);
491 expect(fallback).toMatchObject({ items: size - 1, requestIds: 0 });
492 });
493
494 it("matches a full projection through streamed rows, request changes and answers", () => {
495 const question = projection.turnItems[0] as QuestionItem;
496 const resolved = (answers: Record<string, string>) => ({
497 ...projection.runtimeRequests[0]!,
498 status: "resolved" as const,
499 answers,
500 });
501 const explicitRequestId = RuntimeRequestId.make("explicit-question");
502 const explicitAnswer = {
503 requestId: explicitRequestId,
504 answers: { next: "Explicit" },
505 attachmentsByQuestionId: {},
506 questionTextById: { next: "What should happen next?" },
507 };
508 const unanswered: QuestionItem = {
509 ...question,
510 id: TurnItemId.make("explicit-question-item"),
511 ordinal: 1,
512 requestId: explicitRequestId,
513 };
514 const explicit: QuestionItem = { ...unanswered, questionAnswer: explicitAnswer };
515 const explicitRequest = {
516 ...resolved({ next: "From request" }),
517 id: explicitRequestId,
518 };
519 const stream = assistantItem(2);
520 const inherited = {
521 ...localRow({ ...question, id: TurnItemId.make("inherited-question-item") }, 3),
522 visibility: "inherited" as const,
523 sourceThreadId: ThreadId.make("parent-thread"),
524 };
525 const turnItems = [question, explicit, stream];
526 let current: OrchestrationV2ThreadProjection = {
527 ...v2Projection,
528 turnItems,
529 visibleTurnItems: [...turnItems.map(localRow), inherited],
530 runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest],
531 };
532 const project = createQuestionHistoryProjector();
533 const step = (next: OrchestrationV2ThreadProjection) => {
534 current = next;
535 const result = project(next);
536 expect(result).toEqual(createQuestionHistoryProjector()(next));
537 return result;
538 };
539 const answerOf = (rows: ReadonlyArray<{ readonly item: unknown }>, index: number) =>
540 (rows[index]!.item as { readonly questionAnswer?: { readonly answers: unknown } })
541 .questionAnswer?.answers;
542
543 expect(step(current)).toBe(current.visibleTurnItems);
544 expect(step(updated(current, assistantItem(2, "Hello")))).toBe(current.visibleTurnItems);
545
546 // The last duplicate request wins, and every row asking it gets the answer.
547 const answered = step({
548 ...current,
549 runtimeRequests: [
550 projection.runtimeRequests[0]!,
551 resolved({ next: "Continue" }),
552 explicitRequest,
553 ],
554 });
555 expect(answerOf(answered, 0)).toEqual({ next: "Continue" });
556 expect(answerOf(answered, 1)).toEqual({ next: "Explicit" });
557 expect(answerOf(answered, 3)).toEqual({ next: "Continue" });
558
559 const whileOverlaid = step(updated(current, assistantItem(2, "Hello there")));
560 expect(whileOverlaid[0]).toBe(answered[0]);
561 expect(whileOverlaid[3]).toBe(answered[3]);
562
563 const changedAnswer = step({
564 ...current,
565 runtimeRequests: [resolved({ next: "Stop" }), explicitRequest],
566 });
567 expect(answerOf(changedAnswer, 0)).toEqual({ next: "Stop" });
568
569 const reopened = step({
570 ...current,
571 runtimeRequests: [projection.runtimeRequests[0]!, explicitRequest],
572 });
573 expect(reopened).toBe(current.visibleTurnItems);
574
575 // Dropping an explicit answer falls back to its resolved request.
576 const restored = step(updated(current, unanswered));
577 expect(answerOf(restored, 1)).toEqual({ next: "From request" });
578
579 const retitled: QuestionItem = {
580 ...unanswered,
581 questions: [{ ...question.questions[0]!, question: "Now what?" }],
582 };
583 const retitledRows = step(updated(current, retitled));
584 expect(retitledRows[1]!.item).toMatchObject({
585 questionAnswer: { questionTextById: { next: "Now what?" } },
586 });
587 });
588});
589
590it("loads only child approvals and retains their destination across environments", () => {
591 const local = EnvironmentId.make("local");
592 const remote = EnvironmentId.make("remote");
593 const parent = { environmentId: local, threadId: ThreadId.make("parent") };
594 const childId = ThreadId.make("child");
595 const children = Atom.make(
596 [local, remote].map((environmentId) => ({
597 ...presentThreadShell(environmentId, v2ThreadShell),
598 id: childId,
599 title: environmentId,
600 hasPendingApprovals: true,
601 })),
602 );
603 const details = Atom.family((key: string) =>
604 Atom.make<PendingThreadRequests | null>({
605 approvals: [
606 {
607 requestId: RuntimeRequestId.make(key),
608 requestKind: "command",
609 createdAt: "2026-10-19T00:00:00Z",
610 responseCapability: "live",
611 },
612 ],
613 userInputs: [],
614 }),
615 );
616 const readDetails = vi.fn((ref: ScopedThreadRef) => details(threadKey(ref)));
617 const approvalsAtom = createChildThreadApprovalsAtom({
618 childThreadInputsAtom: () => children,
619 pendingRequestsAtom: readDetails,
620 });
621 const registry = AtomRegistry.make();
622 const unmount = registry.mount(approvalsAtom(parent));
623 const pending = registry.get(approvalsAtom(parent));
624 expect(pending.map((entry) => entry.thread)).toEqual(
625 [local, remote].map((environmentId) => ({ environmentId, threadId: childId })),
626 );
627 expect(pending.map((entry) => entry.title)).toEqual(["local", "remote"]);
628 registry.set(details(threadKey({ environmentId: remote, threadId: childId })), {
629 approvals: [],
630 userInputs: [],
631 });
632 expect(registry.get(approvalsAtom(parent))).toEqual([pending[0]]);
633 readDetails.mockClear();
634 registry.set(
635 children,
636 registry
637 .get(children)
638 .map((child) => ({ ...child, hasPendingApprovals: false, hasPendingUserInput: true })),
639 );
640 expect(registry.get(approvalsAtom(parent))).toEqual([]);
641 expect(readDetails).not.toHaveBeenCalled();
642 unmount();
643 registry.dispose();
644});