1import { assert, describe, it } from "@effect/vitest";
2import {
3 ProviderDriverKind,
4 ProviderInstanceId,
5 ProviderSessionId,
6 ProviderSetupError,
7 ProviderThreadId,
8 ThreadId,
9 type OrchestrationV2ProviderSession,
10 type OrchestrationV2ThreadShell,
11 type ProviderAuthState,
12} from "@t3tools/contracts";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Deferred from "effect/Deferred";
16import * as Fiber from "effect/Fiber";
17import * as Layer from "effect/Layer";
18import * as PubSub from "effect/PubSub";
19import * as Queue from "effect/Queue";
20import * as Stream from "effect/Stream";
21import * as SubscriptionRef from "effect/SubscriptionRef";
22
23import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
24import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts";
25import { AcpProviderCapabilitiesV2 } from "@t3tools/provider-acp/server/adapter";
26import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
27import * as ProviderAuthService from "./ProviderAuthService.ts";
28import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts";
29
30const instanceId = ProviderInstanceId.make("antigravity-personal");
31const otherInstanceId = ProviderInstanceId.make("antigravity-work");
32const unsupportedInstanceId = ProviderInstanceId.make("codex");
33const driverKind = ProviderDriverKind.make("antigravity");
34const owner = "paired-client-owner";
35const otherOwner = "paired-client-other";
36const flowId = "test-sign-in-flow";
37const callbackUrl = "http://127.0.0.1:48123/?state=test-state&code=test-code";
38const now = "2026-09-02T00:00:00.000Z";
39const idleAuthState: ProviderAuthState = {
40 instanceId,
41 phase: "idle",
42 flowId: null,
43 authorizationUrl: null,
44 expiresAt: null,
45 message: null,
46};
47const waitingAuthState: ProviderAuthState = {
48 ...idleAuthState,
49 phase: "waiting",
50 flowId,
51 authorizationUrl: "https://accounts.google.com/o/oauth2/v2/auth?state=test-state",
52 expiresAt: "2026-09-02T00:05:00.000Z",
53};
54
55function makeInstance(input: {
56 instanceId: ProviderInstanceId;
57 enabled: boolean;
58 auth?: ProviderAuthService.ProviderAuthController;
59}): ProviderInstance {
60 return {
61 ...input,
62 driverKind,
63 displayName: undefined,
64 continuationIdentity: { driverKind, continuationKey: input.instanceId },
65 get snapshot(): never {
66 throw new Error("Auth routing must not refresh the provider snapshot.");
67 },
68 get orchestrationAdapter(): never {
69 throw new Error("Auth routing must not start an adapter session.");
70 },
71 get textGeneration(): never {
72 throw new Error("Auth routing must not generate text.");
73 },
74 };
75}
76
77const nowUtc = DateTime.makeUnsafe(now);
78
79/**
80 * A thread bound to a provider instance with (or without) a live native
81 * thread. The auth service reads only the routing fields, so the rest of the
82 * shell row is left out.
83 */
84function makeThread(
85 thread: string,
86 input: { readonly providerInstanceId?: ProviderInstanceId; readonly active?: boolean } = {},
87): OrchestrationV2ThreadShell {
88 const shell: Pick<
89 OrchestrationV2ThreadShell,
90 "id" | "providerInstanceId" | "activeProviderThreadId"
91 > = {
92 id: ThreadId.make(thread),
93 providerInstanceId: input.providerInstanceId ?? instanceId,
94 activeProviderThreadId:
95 input.active === false ? null : ProviderThreadId.make(`${thread}:native`),
96 };
97 return shell as unknown as OrchestrationV2ThreadShell;
98}
99
100function makeSession(
101 session: string,
102 status: OrchestrationV2ProviderSession["status"] = "ready",
103 providerInstanceId = instanceId,
104): OrchestrationV2ProviderSession {
105 return {
106 id: ProviderSessionId.make(session),
107 driver: driverKind,
108 providerInstanceId,
109 status,
110 cwd: "/workspace",
111 model: null,
112 capabilities: AcpProviderCapabilitiesV2,
113 createdAt: nowUtc,
114 updatedAt: nowUtc,
115 lastError: null,
116 };
117}
118
119function sessionThreads(list: ReadonlyArray<OrchestrationV2ProviderSession>) {
120 return {
121 threads: list.map((session) => makeThread(session.id)),
122 sessions: new Map(list.map((session) => [ThreadId.make(session.id), [session]])),
123 };
124}
125
126const makeHarness = Effect.fn("ProviderAuthService.test.makeHarness")(function* (
127 input: {
128 enabled?: boolean;
129 threads?: ReadonlyArray<OrchestrationV2ThreadShell>;
130 sessions?: ReadonlyMap<ThreadId, ReadonlyArray<OrchestrationV2ProviderSession>>;
131 shellError?: boolean;
132 stopError?: boolean;
133 logoutError?: ProviderSetupError;
134 sharedCredentials?: boolean;
135 sharedBusy?: boolean;
136 sharedBusyEffect?: Effect.Effect<boolean>;
137 responds?: boolean;
138 beforeLogout?: Effect.Effect<void>;
139 beforeStop?: Effect.Effect<void>;
140 beforeListSessions?: Effect.Effect<void>;
141 onLookup?: Effect.Effect<void>;
142 onListThreads?: (instances: ProviderInstance[]) => void;
143 } = {},
144) {
145 const actions: string[] = [];
146 const registryChanges = yield* PubSub.unbounded<void>();
147 const threads = input.threads ?? [];
148 const sessions = new Map(
149 [...(input.sessions?.entries() ?? [])].map(([threadId, list]) => [threadId, [...list]]),
150 );
151 const released: string[] = [];
152 const idle = idleAuthState;
153 let state = idle;
154 let flowOwner: string | undefined;
155 let gateClosed = false;
156
157 const checkOwner = Effect.fn("ProviderAuthService.test.checkOwner")(function* (
158 ownerSessionId: string,
159 requestedFlowId: string,
160 operation: string,
161 ) {
162 if (ownerSessionId !== flowOwner || requestedFlowId !== state.flowId) {
163 return yield* new ProviderSetupError({
164 instanceId,
165 operation,
166 detail: "This sign-in belongs to another client or has expired.",
167 });
168 }
169 });
170
171 const auth: ProviderAuthService.ProviderAuthController = {
172 ...(input.sharedCredentials
173 ? { credentialBinding: { owner: "provider" as const, key: "shared" } }
174 : {}),
175 ...(input.responds
176 ? {
177 respond: Effect.fn(function* (ownerSessionId, request) {
178 yield* checkOwner(ownerSessionId, request.flowId, "respond");
179 actions.push(`respond:${request.response.type}`);
180 return state;
181 }),
182 }
183 : {}),
184 start: Effect.fn(function* (ownerSessionId, stopSessions) {
185 gateClosed = true;
186 actions.push("close-gate");
187 yield* stopSessions ?? Effect.void;
188 flowOwner = ownerSessionId;
189 state = waitingAuthState;
190 actions.push("start-sign-in");
191 return state;
192 }),
193 complete: Effect.fn(function* (ownerSessionId, request) {
194 yield* checkOwner(ownerSessionId, request.flowId, "complete");
195 if (request.callbackUrl !== callbackUrl) {
196 return yield* new ProviderSetupError({
197 instanceId,
198 operation: "complete",
199 detail: "The redirect URL does not match this sign-in.",
200 });
201 }
202 state = { ...idle, flowId, phase: "succeeded" };
203 return state;
204 }),
205 cancel: Effect.fn(function* (ownerSessionId, requestedFlowId) {
206 yield* checkOwner(ownerSessionId, requestedFlowId, "cancel");
207 state = { ...idle, flowId, phase: "cancelled" };
208 return state;
209 }),
210 logout: Effect.fn(function* (stopSessions) {
211 gateClosed = true;
212 actions.push("close-gate");
213 yield* stopSessions;
214 if (input.logoutError) return yield* input.logoutError;
215 yield* input.beforeLogout ?? Effect.void;
216 actions.push("native-logout");
217 state = idle;
218 flowOwner = undefined;
219 return state;
220 }),
221 subscribe: (ownerSessionId) =>
222 Stream.fromEffect(Effect.sync(() => (ownerSessionId === flowOwner ? state : idle))),
223 isLogoutPrompt: (text, hasAttachments) => !hasAttachments && text.trim() === "/logout",
224 };
225 const instances = [
226 makeInstance({ instanceId, enabled: input.enabled ?? true, auth }),
227 makeInstance({ instanceId: unsupportedInstanceId, enabled: true }),
228 ...(input.sharedCredentials
229 ? [
230 makeInstance({
231 instanceId: otherInstanceId,
232 enabled: true,
233 auth: {
234 ...auth,
235 isChangingCredentials:
236 input.sharedBusyEffect ?? Effect.succeed(input.sharedBusy ?? false),
237 invalidate: Effect.sync(() => {
238 actions.push("invalidate-shared");
239 }),
240 },
241 }),
242 ]
243 : []),
244 ];
245 const service = yield* ProviderAuthService.makeProviderAuthService.pipe(
246 Effect.provide(
247 Layer.mergeAll(
248 Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({
249 getInstance: (id) =>
250 Effect.gen(function* () {
251 const found = instances.find((instance) => instance.instanceId === id);
252 yield* input.onLookup ?? Effect.void;
253 return found;
254 }),
255 listInstances: Effect.succeed(instances),
256 subscribeChanges: PubSub.subscribe(registryChanges),
257 }),
258 Layer.mock(ProjectionStore.ProjectionStoreV2)({
259 getRecoveryThreadIds: () =>
260 Effect.gen(function* () {
261 assert.isTrue(gateClosed);
262 actions.push("list-threads");
263 input.onListThreads?.(instances);
264 yield* input.beforeListSessions ?? Effect.void;
265 return yield* input.shellError
266 ? Effect.fail(
267 new ProjectionStore.ProjectionStoreReadError({
268 threadId: ThreadId.make("shell"),
269 cause: new Error("private database diagnostics"),
270 }),
271 )
272 : Effect.succeed(threads.map((thread) => thread.id));
273 }),
274 getThreadRecords: (threadId) =>
275 Effect.sync(() => {
276 actions.push(`sessions:${threadId}`);
277 return {
278 providerSessions: sessions.get(threadId) ?? [],
279 } as never;
280 }),
281 }),
282 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
283 release: ({ providerSessionId, reason }) =>
284 Effect.gen(function* () {
285 assert.isTrue(gateClosed);
286 actions.push(`stop:${providerSessionId}`);
287 yield* input.beforeStop ?? Effect.void;
288 if (input.stopError) {
289 return yield* Effect.fail(
290 new ProviderSessionManager.ProviderSessionReleaseError({
291 providerSessionId,
292 reason,
293 cause: new Error("private process diagnostics"),
294 }),
295 );
296 }
297 released.push(providerSessionId);
298 for (const [threadId, list] of sessions) {
299 const remaining = list.filter((session) => session.id !== providerSessionId);
300 if (remaining.length === 0) sessions.delete(threadId);
301 else sessions.set(threadId, remaining);
302 }
303 }),
304 }),
305 ),
306 ),
307 );
308 return {
309 service,
310 actions,
311 released,
312 sessions,
313 auth,
314 addInstance: (instance: ProviderInstance) => instances.push(instance),
315 replaceInstance: (replacement: ProviderInstance) => {
316 const index = instances.findIndex(
317 (instance) => instance.instanceId === replacement.instanceId,
318 );
319 assert.isAtLeast(index, 0);
320 instances[index] = replacement;
321 },
322 replaceAuth: (next: ProviderAuthService.ProviderAuthController) => {
323 instances[0] = makeInstance({ instanceId, enabled: input.enabled ?? true, auth: next });
324 },
325 };
326});
327
328const makeStreamingController = Effect.fn("ProviderAuthService.test.makeStreamingController")(
329 function* (flowOwner: string) {
330 const state = yield* SubscriptionRef.make(idleAuthState);
331 const close = yield* Deferred.make<void>();
332 const closedSubscriptions = yield* Queue.unbounded<string>();
333 const unused = () => Effect.die("Unexpected auth operation in a subscription test.");
334 const auth: ProviderAuthService.ProviderAuthController = {
335 start: unused,
336 complete: unused,
337 cancel: unused,
338 logout: unused,
339 subscribe: (ownerSessionId) =>
340 SubscriptionRef.changes(state).pipe(
341 Stream.map((current) => (ownerSessionId === flowOwner ? current : idleAuthState)),
342 Stream.interruptWhen(Deferred.await(close)),
343 Stream.ensuring(Queue.offer(closedSubscriptions, ownerSessionId)),
344 ),
345 };
346 return {
347 auth,
348 state,
349 closedSubscriptions,
350 close: Deferred.succeed(close, undefined),
351 };
352 },
353);
354
355const makeSubscriptionHarness = Effect.fn("ProviderAuthService.test.makeSubscriptionHarness")(
356 function* (initial: ProviderInstance, replaceDuringFirstLookup?: ProviderInstance) {
357 const changes = yield* PubSub.unbounded<void>();
358 let current: ProviderInstance | undefined = initial;
359 let pendingReplacement = replaceDuringFirstLookup;
360 let subscribed = false;
361 const service = yield* ProviderAuthService.makeProviderAuthService.pipe(
362 Effect.provide(
363 Layer.mergeAll(
364 Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({
365 subscribeChanges: Effect.gen(function* () {
366 const subscription = yield* PubSub.subscribe(changes);
367 subscribed = true;
368 return subscription;
369 }),
370 getInstance: () =>
371 Effect.gen(function* () {
372 const instance = current;
373 if (pendingReplacement) {
374 assert.isTrue(subscribed, "Registry changes must be subscribed before lookup.");
375 current = pendingReplacement;
376 pendingReplacement = undefined;
377 yield* PubSub.publish(changes, undefined);
378 }
379 return instance;
380 }),
381 }),
382 Layer.mock(ProjectionStore.ProjectionStoreV2)({}),
383 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
384 ),
385 ),
386 );
387 return {
388 service,
389 replace: Effect.fn(function* (replacement: ProviderInstance | undefined) {
390 current = replacement;
391 yield* PubSub.publish(changes, undefined);
392 }),
393 };
394 },
395);
396
397const observeAuth = Effect.fn("ProviderAuthService.test.observeAuth")(function* (
398 stream: Stream.Stream<ProviderAuthState, ProviderSetupError>,
399) {
400 const states = yield* Queue.unbounded<ProviderAuthState>();
401 const fiber = yield* stream.pipe(
402 Stream.runForEach((state) => Queue.offer(states, state)),
403 Effect.forkScoped,
404 );
405 return { states, fiber };
406});
407
408describe("ProviderAuthService", () => {
409 it.effect(
410 "stops sessions sharing credentials and invalidates their processes before logout",
411 () =>
412 Effect.gen(function* () {
413 const { service, actions, sessions } = yield* makeHarness({
414 sharedCredentials: true,
415 ...sessionThreads([
416 makeSession("target"),
417 makeSession("shared", "ready", otherInstanceId),
418 makeSession("unrelated", "ready", unsupportedInstanceId),
419 ]),
420 });
421 yield* service.logout({ instanceId });
422 assert.deepStrictEqual([...sessions.keys()], [ThreadId.make("unrelated")]);
423 assert.isBelow(actions.indexOf("stop:shared"), actions.indexOf("invalidate-shared"));
424 assert.isBelow(actions.indexOf("invalidate-shared"), actions.indexOf("native-logout"));
425 }),
426 );
427 it.effect("rejects overlapping changes to a shared sign-in", () =>
428 Effect.gen(function* () {
429 const { service, actions } = yield* makeHarness({
430 sharedCredentials: true,
431 sharedBusy: true,
432 });
433 for (const task of [service.start({ instanceId }, owner), service.logout({ instanceId })]) {
434 const error = yield* task.pipe(Effect.flip);
435 assert.include(error.detail, "shared sign-in");
436 }
437 assert.deepStrictEqual(actions, []);
438 }),
439 );
440 it.effect("routes typed interactions to the flow owner and rejects unsupported controllers", () =>
441 Effect.gen(function* () {
442 const { service, actions } = yield* makeHarness({ responds: true });
443 yield* service.start({ instanceId }, owner);
444 const request = {
445 instanceId,
446 flowId,
447 interactionId: "consent",
448 response: { type: "browser" as const, action: "accept" as const },
449 };
450 const rejected = yield* service.respond(request, otherOwner).pipe(Effect.flip);
451 assert.strictEqual(rejected.operation, "respond");
452 yield* service.respond(request, owner);
453 assert.strictEqual(actions.at(-1), "respond:browser");
454 const unsupported = yield* makeHarness();
455 const error = yield* unsupported.service.respond(request, owner).pipe(Effect.flip);
456 assert.include(error.detail, "does not accept");
457 }),
458 );
459 it.effect("stops routed sessions before sign-in, including for a disabled instance", () =>
460 Effect.gen(function* () {
461 const { service, actions, released } = yield* makeHarness({
462 enabled: false,
463 threads: [makeThread("active")],
464 sessions: new Map([[ThreadId.make("active"), [makeSession("active-session")]]]),
465 });
466 const state = yield* service.start({ instanceId }, owner);
467
468 assert.strictEqual(state.instanceId, instanceId);
469 assert.strictEqual(state.phase, "waiting");
470 assert.deepStrictEqual(released, ["active-session"]);
471 assert.deepStrictEqual(actions, [
472 "close-gate",
473 "list-threads",
474 "sessions:active",
475 "stop:active-session",
476 "start-sign-in",
477 ]);
478 }),
479 );
480
481 it.effect("keeps sign-in state private and accepts the owner's redirect URL", () =>
482 Effect.gen(function* () {
483 const { service } = yield* makeHarness();
484 const waiting = yield* service.start({ instanceId }, owner);
485 const ownerStates = yield* service
486 .subscribe({ instanceId }, owner)
487 .pipe(Stream.take(1), Stream.runCollect);
488 const otherStates = yield* service
489 .subscribe({ instanceId }, otherOwner)
490 .pipe(Stream.take(1), Stream.runCollect);
491
492 assert.deepStrictEqual(ownerStates, [waiting]);
493 assert.strictEqual(otherStates[0]?.authorizationUrl, null);
494 assert.strictEqual(otherStates[0]?.flowId, null);
495
496 const otherError = yield* Effect.flip(
497 service.complete({ instanceId, flowId, callbackUrl }, otherOwner),
498 );
499 assert.strictEqual(otherError.operation, "complete");
500 const complete = yield* service.complete({ instanceId, flowId, callbackUrl }, owner);
501 assert.strictEqual(complete.phase, "succeeded");
502 assert.strictEqual(complete.authorizationUrl, null);
503 }),
504 );
505
506 it.effect("lets only the flow owner cancel sign-in", () =>
507 Effect.gen(function* () {
508 const { service } = yield* makeHarness();
509 yield* service.start({ instanceId }, owner);
510 const error = yield* Effect.flip(service.cancel({ instanceId, flowId }, otherOwner));
511 assert.strictEqual(error.operation, "cancel");
512
513 const cancelled = yield* service.cancel({ instanceId, flowId }, owner);
514 assert.strictEqual(cancelled.phase, "cancelled");
515 assert.strictEqual(cancelled.authorizationUrl, null);
516 }),
517 );
518
519 it.effect.each([
520 { change: "enable", initialEnabled: false, closeBeforeReplacement: true },
521 { change: "config", initialEnabled: true, closeBeforeReplacement: false },
522 ])(
523 "keeps private auth subscriptions current after an instance $change change",
524 ({ initialEnabled, closeBeforeReplacement }) =>
525 Effect.gen(function* () {
526 const first = yield* makeStreamingController(owner);
527 const replacement = yield* makeStreamingController(otherOwner);
528 yield* SubscriptionRef.set(first.state, waitingAuthState);
529 const { service, replace } = yield* makeSubscriptionHarness(
530 makeInstance({ instanceId, enabled: initialEnabled, auth: first.auth }),
531 );
532 const firstClient = yield* observeAuth(service.subscribe({ instanceId }, owner));
533 const secondClient = yield* observeAuth(service.subscribe({ instanceId }, otherOwner));
534 assert.deepStrictEqual(yield* Queue.take(firstClient.states), waitingAuthState);
535 assert.deepStrictEqual(yield* Queue.take(secondClient.states), idleAuthState);
536
537 if (closeBeforeReplacement) {
538 yield* first.close;
539 const closed = yield* Queue.takeN(first.closedSubscriptions, 2);
540 assert.deepStrictEqual(new Set(closed), new Set([owner, otherOwner]));
541 }
542 yield* replace(makeInstance({ instanceId, enabled: true, auth: replacement.auth }));
543 assert.deepStrictEqual(yield* Queue.take(firstClient.states), idleAuthState);
544 assert.deepStrictEqual(yield* Queue.take(secondClient.states), idleAuthState);
545 if (!closeBeforeReplacement) {
546 const closed = yield* Queue.takeN(first.closedSubscriptions, 2);
547 assert.deepStrictEqual(new Set(closed), new Set([owner, otherOwner]));
548 }
549
550 const staleState: ProviderAuthState = { ...idleAuthState, phase: "cancelled" };
551 yield* SubscriptionRef.set(first.state, staleState);
552 const newWaiting: ProviderAuthState = {
553 ...waitingAuthState,
554 flowId: "replacement-flow",
555 authorizationUrl: "https://accounts.google.com/o/oauth2/v2/auth?state=replacement-state",
556 };
557 yield* SubscriptionRef.set(replacement.state, newWaiting);
558 assert.deepStrictEqual(yield* Queue.take(firstClient.states), idleAuthState);
559 assert.deepStrictEqual(yield* Queue.take(secondClient.states), newWaiting);
560 }),
561 );
562
563 it.effect("does not miss a replacement during the first instance lookup", () =>
564 Effect.gen(function* () {
565 const first = yield* makeStreamingController(owner);
566 const replacement = yield* makeStreamingController(owner);
567 yield* SubscriptionRef.set(replacement.state, waitingAuthState);
568 const { service } = yield* makeSubscriptionHarness(
569 makeInstance({ instanceId, enabled: false, auth: first.auth }),
570 makeInstance({ instanceId, enabled: true, auth: replacement.auth }),
571 );
572
573 const states = yield* service.subscribe({ instanceId }, owner).pipe(
574 Stream.filter((state) => state.phase === "waiting"),
575 Stream.take(1),
576 Stream.runCollect,
577 );
578
579 assert.deepStrictEqual(states, [waitingAuthState]);
580 }),
581 );
582
583 it.effect("ends the subscription with a setup error when the instance is removed", () =>
584 Effect.gen(function* () {
585 const controller = yield* makeStreamingController(owner);
586 const { service, replace } = yield* makeSubscriptionHarness(
587 makeInstance({ instanceId, enabled: true, auth: controller.auth }),
588 );
589 const { states, fiber } = yield* observeAuth(service.subscribe({ instanceId }, owner));
590 assert.deepStrictEqual(yield* Queue.take(states), idleAuthState);
591
592 yield* replace(undefined);
593 const error = yield* Effect.flip(Fiber.join(fiber));
594
595 assert.instanceOf(error, ProviderSetupError);
596 assert.strictEqual(error.instanceId, instanceId);
597 assert.strictEqual(error.operation, "subscribe");
598 assert.include(error.detail, "no longer available");
599 assert.strictEqual(yield* Queue.take(controller.closedSubscriptions), owner);
600 }),
601 );
602
603 it.effect.each([
604 { id: ProviderInstanceId.make("missing"), detail: "no longer available" },
605 { id: unsupportedInstanceId, detail: "does not support sign-in" },
606 ])("rejects setup for unavailable or unsupported instance $id", ({ id, detail }) =>
607 Effect.gen(function* () {
608 const { service, actions } = yield* makeHarness();
609 const operations: Effect.Effect<
610 ProviderAuthState | ReadonlyArray<ProviderAuthState>,
611 ProviderSetupError
612 >[] = [
613 service.start({ instanceId: id }, owner),
614 service.complete({ instanceId: id, flowId, callbackUrl }, owner),
615 service.cancel({ instanceId: id, flowId }, owner),
616 service.logout({ instanceId: id }),
617 Stream.runCollect(service.subscribe({ instanceId: id }, owner)),
618 ];
619 for (const operation of operations) {
620 const error = yield* Effect.flip(operation);
621 assert.instanceOf(error, ProviderSetupError);
622 assert.strictEqual(error.instanceId, id);
623 assert.include(error.detail, detail);
624 }
625 assert.deepStrictEqual(actions, []);
626 }),
627 );
628
629 it.effect("stops every live session on the instance before native logout", () =>
630 Effect.gen(function* () {
631 const { service, actions, released } = yield* makeHarness({
632 threads: [
633 makeThread("running"),
634 makeThread("starting"),
635 makeThread("idle", { active: false }),
636 makeThread("other-instance", { providerInstanceId: otherInstanceId }),
637 ],
638 sessions: new Map([
639 [
640 ThreadId.make("running"),
641 [
642 makeSession("running-session", "running"),
643 makeSession("stopped-session", "stopped"),
644 makeSession("errored-session", "error"),
645 makeSession("foreign-session", "ready", otherInstanceId),
646 ],
647 ],
648 [ThreadId.make("starting"), [makeSession("starting-session", "starting")]],
649 [ThreadId.make("idle"), [makeSession("idle-session")]],
650 [
651 ThreadId.make("other-instance"),
652 [makeSession("other-session", "ready", otherInstanceId)],
653 ],
654 ]),
655 });
656 const state = yield* service.logout({ instanceId });
657
658 assert.strictEqual(state.phase, "idle");
659 assert.deepStrictEqual(actions, [
660 "close-gate",
661 "list-threads",
662 "sessions:running",
663 "stop:running-session",
664 "sessions:starting",
665 "stop:starting-session",
666 "sessions:idle",
667 "stop:idle-session",
668 "sessions:other-instance",
669 "native-logout",
670 ]);
671 assert.deepStrictEqual(released, ["running-session", "starting-session", "idle-session"]);
672 }),
673 );
674
675 it.effect(
676 "releases a shared native session once after a thread's selected provider changes",
677 () =>
678 Effect.gen(function* () {
679 const shared = makeSession("shared-native-session");
680 const { service, released } = yield* makeHarness({
681 threads: [
682 makeThread("changed-selection", { providerInstanceId: otherInstanceId }),
683 makeThread("same-session"),
684 ],
685 sessions: new Map([
686 [ThreadId.make("changed-selection"), [shared]],
687 [ThreadId.make("same-session"), [shared]],
688 ]),
689 });
690
691 yield* service.logout({ instanceId });
692
693 assert.deepStrictEqual(released, [shared.id]);
694 }),
695 );
696
697 it.effect.each([
698 { text: "/logout", hasAttachments: false, handled: true },
699 { text: " \n/logout\t", hasAttachments: false, handled: true },
700 { text: "/logout", hasAttachments: true, handled: false },
701 { text: "/logout please", hasAttachments: false, handled: false },
702 { text: "Explain /logout", hasAttachments: false, handled: false },
703 { text: "/Logout", hasAttachments: false, handled: false },
704 ])("handles only a standalone logout command %#", ({ text, hasAttachments, handled }) =>
705 Effect.gen(function* () {
706 const { service, actions } = yield* makeHarness();
707 const result = yield* service.tryHandlePromptCommand({ instanceId, text, hasAttachments });
708
709 assert.strictEqual(result, handled);
710 assert.deepStrictEqual(
711 actions,
712 handled ? ["close-gate", "list-threads", "native-logout"] : [],
713 );
714 }),
715 );
716
717 it.effect("does not intercept commands for providers without an auth controller", () =>
718 Effect.gen(function* () {
719 const { service, actions } = yield* makeHarness();
720 for (const id of [unsupportedInstanceId, ProviderInstanceId.make("missing")]) {
721 assert.isFalse(
722 yield* service.tryHandlePromptCommand({
723 instanceId: id,
724 text: "/logout",
725 hasAttachments: false,
726 }),
727 );
728 }
729 assert.deepStrictEqual(actions, []);
730 }),
731 );
732
733 it.effect("does not log out when the thread shell cannot be read", () =>
734 Effect.gen(function* () {
735 const { service, actions } = yield* makeHarness({ shellError: true });
736 const error = yield* Effect.flip(service.logout({ instanceId }));
737
738 assert.instanceOf(error, ProviderSetupError);
739 assert.strictEqual(error.instanceId, instanceId);
740 assert.strictEqual(error.operation, "stopSessions");
741 assert.notInclude(error.detail, "private database diagnostics");
742 assert.deepStrictEqual(actions, ["close-gate", "list-threads"]);
743 }),
744 );
745
746 it.effect("does not log out or consume the command when stopping a session fails", () =>
747 Effect.gen(function* () {
748 const { service, actions, released } = yield* makeHarness({
749 threads: [makeThread("active")],
750 sessions: new Map([[ThreadId.make("active"), [makeSession("active-session")]]]),
751 stopError: true,
752 });
753 const error = yield* Effect.flip(
754 service.tryHandlePromptCommand({ instanceId, text: "/logout", hasAttachments: false }),
755 );
756
757 assert.instanceOf(error, ProviderSetupError);
758 assert.strictEqual(error.operation, "stopSessions");
759 assert.notInclude(error.detail, "private process diagnostics");
760 assert.deepStrictEqual(released, []);
761 assert.deepStrictEqual(actions, [
762 "close-gate",
763 "list-threads",
764 "sessions:active",
765 "stop:active-session",
766 ]);
767 }),
768 );
769
770 it.effect("returns native logout failures to the command caller", () =>
771 Effect.gen(function* () {
772 const logoutError = new ProviderSetupError({
773 instanceId,
774 operation: "logout",
775 detail: "Native sign-out failed. Try again.",
776 });
777 const { service } = yield* makeHarness({ logoutError });
778 const error = yield* Effect.flip(
779 service.tryHandlePromptCommand({ instanceId, text: "/logout", hasAttachments: false }),
780 );
781
782 assert.strictEqual(error, logoutError);
783 }),
784 );
785});
786
787it.effect("queued logout prompts resolve the current controller after provider replacement", () =>
788 Effect.gen(function* () {
789 const entered = yield* Deferred.make<void>();
790 const release = yield* Deferred.make<void>();
791 const lookup = yield* Deferred.make<void>();
792 let observeLookup = false;
793 const harness = yield* makeHarness({
794 beforeLogout: Deferred.succeed(entered, undefined).pipe(
795 Effect.andThen(Deferred.await(release)),
796 ),
797 onLookup: Effect.suspend(() =>
798 observeLookup ? Deferred.succeed(lookup, undefined).pipe(Effect.asVoid) : Effect.void,
799 ),
800 });
801 const first = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild);
802 yield* Deferred.await(entered);
803 observeLookup = true;
804 const queued = yield* harness.service
805 .tryHandlePromptCommand({ instanceId, text: "/logout", hasAttachments: false })
806 .pipe(Effect.forkChild);
807 yield* Deferred.await(lookup);
808 let replacementLoggedOut = false;
809 harness.replaceAuth({
810 ...harness.auth,
811 logout: (stopSessions) =>
812 stopSessions.pipe(
813 Effect.andThen(
814 Effect.sync(() => {
815 replacementLoggedOut = true;
816 return idleAuthState;
817 }),
818 ),
819 ),
820 });
821 yield* Deferred.succeed(release, undefined);
822 yield* Fiber.join(first);
823 assert.isTrue(yield* Fiber.join(queued));
824 assert.isTrue(replacementLoggedOut);
825 assert.strictEqual(harness.actions.filter((action) => action === "native-logout").length, 1);
826 }).pipe(Effect.scoped),
827);
828
829it.effect.each(["start", "logout", "prompt"] as const)(
830 "%s uses the same credential controller for shared exclusion and mutation during replacement",
831 (action) =>
832 Effect.gen(function* () {
833 const checked = yield* Deferred.make<void>();
834 const continueCheck = yield* Deferred.make<void>();
835 const replacementPeerId = ProviderInstanceId.make("replacement-shared-peer");
836 const harness = yield* makeHarness({
837 sharedCredentials: true,
838 sharedBusyEffect: Deferred.succeed(checked, undefined).pipe(
839 Effect.andThen(Deferred.await(continueCheck)),
840 Effect.as(false),
841 ),
842 ...sessionThreads([
843 makeSession("old-shared", "ready", otherInstanceId),
844 makeSession("replacement-shared", "ready", replacementPeerId),
845 ]),
846 });
847 harness.addInstance(
848 makeInstance({
849 instanceId: replacementPeerId,
850 enabled: true,
851 auth: {
852 ...harness.auth,
853 credentialBinding: { owner: "provider", key: "replacement-binding" },
854 isChangingCredentials: Effect.succeed(true),
855 invalidate: Effect.die(
856 "The unrelated replacement credential binding must stay intact.",
857 ),
858 },
859 }),
860 );
861 const operation =
862 action === "start"
863 ? harness.service.start({ instanceId }, owner)
864 : action === "logout"
865 ? harness.service.logout({ instanceId })
866 : harness.service.tryHandlePromptCommand({
867 instanceId,
868 text: "/logout",
869 hasAttachments: false,
870 });
871 const running = yield* operation.pipe(Effect.forkChild);
872 yield* Deferred.await(checked);
873 let replacementMutations = 0;
874 harness.replaceAuth({
875 ...harness.auth,
876 credentialBinding: { owner: "provider", key: "replacement-binding" },
877 start: () =>
878 Effect.sync(() => {
879 replacementMutations++;
880 return waitingAuthState;
881 }),
882 logout: () =>
883 Effect.sync(() => {
884 replacementMutations++;
885 return idleAuthState;
886 }),
887 });
888 yield* Deferred.succeed(continueCheck, undefined);
889 yield* Fiber.join(running);
890 assert.equal(replacementMutations, 0);
891 assert.include(harness.actions, action === "start" ? "start-sign-in" : "native-logout");
892 assert.isFalse(harness.sessions.has(ThreadId.make("old-shared")));
893 assert.isTrue(harness.sessions.has(ThreadId.make("replacement-shared")));
894 const blocked = yield* Effect.flip(harness.service.logout({ instanceId }));
895 assert.include(blocked.detail, "shared sign-in");
896 assert.equal(replacementMutations, 0);
897 }).pipe(Effect.scoped),
898);
899
900it.effect.each([
901 { owner: "provider" as const, key: "different-binding" },
902 { owner: "t3" as const, key: "shared" },
903])(
904 "does not invalidate a peer that switches credential binding during session draining %#",
905 (binding) =>
906 Effect.gen(function* () {
907 const draining = yield* Deferred.make<void>();
908 const continueDrain = yield* Deferred.make<void>();
909 const harness = yield* makeHarness({
910 sharedCredentials: true,
911 ...sessionThreads([makeSession("shared-draining", "ready", otherInstanceId)]),
912 beforeStop: Deferred.succeed(draining, undefined).pipe(
913 Effect.andThen(Deferred.await(continueDrain)),
914 ),
915 });
916 const logout = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild);
917 yield* Deferred.await(draining);
918 let replacementInvalidated = false;
919 harness.replaceInstance(
920 makeInstance({
921 instanceId: otherInstanceId,
922 enabled: true,
923 auth: {
924 ...harness.auth,
925 credentialBinding: binding,
926 invalidate: Effect.sync(() => {
927 replacementInvalidated = true;
928 }),
929 },
930 }),
931 );
932 yield* Deferred.succeed(continueDrain, undefined);
933 yield* Fiber.join(logout);
934 assert.isFalse(replacementInvalidated);
935 assert.isFalse(harness.sessions.has(ThreadId.make("shared-draining")));
936 assert.include(harness.actions, "native-logout");
937 }).pipe(Effect.scoped),
938);
939
940it.effect.each(["selection", "drain"] as const)(
941 "preserves replacement peer sessions when credentials change during session %s",
942 (phase) =>
943 Effect.gen(function* () {
944 const entered = yield* Deferred.make<void>();
945 const proceed = yield* Deferred.make<void>();
946 const block = Deferred.succeed(entered, undefined).pipe(
947 Effect.andThen(Deferred.await(proceed)),
948 );
949 const harness = yield* makeHarness({
950 sharedCredentials: true,
951 ...sessionThreads([
952 makeSession("target-draining"),
953 makeSession("peer-replacement", "ready", otherInstanceId),
954 makeSession("peer-persisted", "running", otherInstanceId),
955 ]),
956 ...(phase === "selection" ? { beforeListSessions: block } : { beforeStop: block }),
957 });
958 const logout = yield* harness.service.logout({ instanceId }).pipe(Effect.forkChild);
959 yield* Deferred.await(entered);
960 harness.replaceInstance(
961 makeInstance({
962 instanceId: otherInstanceId,
963 enabled: true,
964 auth: {
965 ...harness.auth,
966 credentialBinding: { owner: "provider", key: "replacement-credentials" },
967 invalidate: Effect.die("The replacement peer's credentials must not be invalidated."),
968 },
969 }),
970 );
971 yield* Deferred.succeed(proceed, undefined);
972 yield* Fiber.join(logout);
973 assert.isTrue(harness.sessions.has(ThreadId.make("peer-replacement")));
974 assert.notInclude(harness.actions, "stop:peer-replacement");
975 assert.notInclude(harness.actions, "stop:peer-persisted");
976 assert.isFalse(harness.sessions.has(ThreadId.make("target-draining")));
977 assert.include(harness.actions, "native-logout");
978 }).pipe(Effect.scoped),
979);