authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-05 17:18:50-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-05 17:18:50-07:00
logbd16b86d50c1df49afeb7c0a7568a4908ade4048
treede1c5b1a3aade45383a92ab10a53f9f123198c6d
parentf12d39359f0f76a64ff2d77959c5baf821df15be
signature Signed by PGP key B5690EEEBB952194

fix(client-runtime): report terminated thread loads (#10216)

4 files changed, 498 insertions(+), 67 deletions(-)

packages/client-runtime/src/rpc/client.test.ts+87-11
......@@ -465,36 +465,112 @@ describe("environment RPC", () => {
465465 }),
466466 );
467467
468 it.effect("does not classify subscription defects as expected failures", () =>
468 it.effect.each(["input", "stream"] as const)(
469 "does not classify %s subscription defects as expected failures",
470 (where) =>
471 Effect.gen(function* () {
472 const defect = new Error("subscription invariant failed");
473 let expectedFailureCount = 0;
474 let inputs = 0;
475 let streams = 0;
476 const observedDefects: unknown[] = [];
477 const client = {
478 [WS_METHODS.subscribeTerminalEvents]: () => {
479 streams += 1;
480 return where === "stream" ? Stream.die(defect) : Stream.never;
481 },
482 } as unknown as WsRpcProtocolClient;
483 const { activeSession, supervisor } = yield* makeHarness();
484
485 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
486 const exit = yield* subscribeDynamicWithSession(
487 WS_METHODS.subscribeTerminalEvents,
488 () =>
489 Effect.sync(() => {
490 inputs += 1;
491 }).pipe(Effect.andThen(where === "input" ? Effect.die(defect) : Effect.succeed({}))),
492 {
493 onDefect: (cause) =>
494 Effect.sync(() => {
495 observedDefects.push(Cause.squash(cause));
496 }),
497 onExpectedFailure: () =>
498 Effect.sync(() => {
499 expectedFailureCount += 1;
500 }),
501 retryExpectedFailureAfter: "250 millis",
502 },
503 ).pipe(
504 Stream.runDrain,
505 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
506 Effect.exit,
507 );
508
509 expect(Exit.isFailure(exit)).toBe(true);
510 if (Exit.isFailure(exit)) {
511 expect(Cause.hasDies(exit.cause)).toBe(true);
512 expect(Cause.squash(exit.cause)).toBe(defect);
513 }
514 expect(inputs).toBe(1);
515 expect(streams).toBe(where === "input" ? 0 : 1);
516 expect(expectedFailureCount).toBe(0);
517 expect(observedDefects).toEqual([defect]);
518 }),
519 );
520
521 it.effect("reports an initializer defect once after an expected failure retries", () =>
469522 Effect.gen(function* () {
470 const defect = new Error("subscription invariant failed");
471 let expectedFailureCount = 0;
523 const defect = new Error("Synthetic retry initializer defect");
524 const expectedFailure = yield* Deferred.make<void>();
525 const observations: string[] = [];
526 const observedDefects: unknown[] = [];
527 let inputs = 0;
472528 const client = {
473 [WS_METHODS.subscribeTerminalEvents]: () => Stream.die(defect),
529 [WS_METHODS.subscribeTerminalEvents]: () => {
530 observations.push("stream");
531 return Stream.fail(new Error("subscription not ready"));
532 },
474533 } as unknown as WsRpcProtocolClient;
475534 const { activeSession, supervisor } = yield* makeHarness();
476
477535 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
478 const exit = yield* subscribe(
536 const fiber = yield* subscribeDynamicWithSession(
479537 WS_METHODS.subscribeTerminalEvents,
480 {},
538 () =>
539 Effect.sync(() => {
540 inputs += 1;
541 observations.push(`input ${inputs}`);
542 return inputs;
543 }).pipe(
544 Effect.flatMap((attempt) => (attempt === 1 ? Effect.succeed({}) : Effect.die(defect))),
545 ),
481546 {
482 onExpectedFailure: () =>
547 onDefect: (cause) =>
483548 Effect.sync(() => {
484 expectedFailureCount += 1;
549 observations.push("defect");
550 observedDefects.push(Cause.squash(cause));
485551 }),
552 onExpectedFailure: () =>
553 Effect.sync(() => {
554 observations.push("expected failure");
555 }).pipe(Effect.andThen(Deferred.succeed(expectedFailure, undefined)), Effect.asVoid),
556 retryExpectedFailureAfter: "250 millis",
486557 },
487558 ).pipe(
488559 Stream.runDrain,
489560 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
490561 Effect.exit,
562 Effect.forkChild,
491563 );
492
564 yield* Deferred.await(expectedFailure);
565 yield* TestClock.adjust("250 millis");
566 const exit = yield* Fiber.join(fiber);
493567 expect(Exit.isFailure(exit)).toBe(true);
494568 if (Exit.isFailure(exit)) {
495569 expect(Cause.hasDies(exit.cause)).toBe(true);
570 expect(Cause.squash(exit.cause)).toBe(defect);
496571 }
497 expect(expectedFailureCount).toBe(0);
572 expect(observations).toEqual(["input 1", "stream", "expected failure", "input 2", "defect"]);
573 expect(observedDefects).toEqual([defect]);
498574 }),
499575 );
500576});
packages/client-runtime/src/rpc/client.ts+56-39
......@@ -171,6 +171,10 @@ export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
171171}
172172
173173interface SubscriptionOptions<TTag extends EnvironmentSubscriptionRpcTag> {
174 /** Reports protocol or programming defects without changing their recovery policy. */
175 readonly onDefect?: (
176 cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
177 ) => Effect.Effect<void, never, never>;
174178 readonly onExpectedFailure?: (
175179 cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
176180 ) => Effect.Effect<void, never, never>;
......@@ -228,47 +232,60 @@ function subscribeDynamicMapped<TTag extends EnvironmentSubscriptionRpcTag, A>(
228232 });
229233 return mapStream(session, method(input)).pipe(
230234 Stream.ensuring(completeObservation),
231 Stream.catchCause((cause) => {
232 const hasOnlyExpectedFailures =
233 cause.reasons.length > 0 &&
234 cause.reasons.every((reason) => reason._tag === "Fail");
235 const isTransportFailure =
236 hasOnlyExpectedFailures &&
237 cause.reasons.every(
238 (reason) => reason._tag === "Fail" && isRpcClientError(reason.error),
239 );
240 if (isTransportFailure) {
241 return Stream.fromEffect(
242 Effect.logWarning(
243 "Durable RPC subscription lost its transport; waiting for the next session.",
244 {
245 cause: Cause.pretty(cause),
246 method: tag,
247 environmentId: supervisor.target.environmentId,
248 },
249 ),
250 ).pipe(Stream.drain);
251 }
252 if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) {
253 const handled = Stream.fromEffect(
254 options.onExpectedFailure(cause),
255 ).pipe(Stream.drain);
256 if (options.retryExpectedFailureAfter === undefined) {
257 return handled;
258 }
259 return handled.pipe(
260 Stream.concat(
261 Stream.fromEffect(
262 Effect.sleep(options.retryExpectedFailureAfter),
263 ).pipe(Stream.drain),
264 ),
265 Stream.concat(subscribeToSession()),
266 );
267 }
268 return Stream.failCause(cause);
269 }),
270235 );
271236 }),
237 ).pipe(
238 Stream.tapCause((cause) =>
239 options?.onDefect !== undefined &&
240 cause.reasons.some(
241 (reason) =>
242 reason._tag === "Die" ||
243 (reason._tag === "Fail" &&
244 isRpcClientError(reason.error) &&
245 reason.error.reason._tag === "RpcClientDefect"),
246 )
247 ? options.onDefect(cause)
248 : Effect.void,
249 ),
250 Stream.catchCause((cause) => {
251 const hasOnlyExpectedFailures =
252 cause.reasons.length > 0 &&
253 cause.reasons.every((reason) => reason._tag === "Fail");
254 const isTransportFailure =
255 hasOnlyExpectedFailures &&
256 cause.reasons.every(
257 (reason) => reason._tag === "Fail" && isRpcClientError(reason.error),
258 );
259 if (isTransportFailure) {
260 return Stream.fromEffect(
261 Effect.logWarning(
262 "Durable RPC subscription lost its transport; waiting for the next session.",
263 {
264 cause: Cause.pretty(cause),
265 method: tag,
266 environmentId: supervisor.target.environmentId,
267 },
268 ),
269 ).pipe(Stream.drain);
270 }
271 if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) {
272 const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe(
273 Stream.drain,
274 );
275 if (options.retryExpectedFailureAfter === undefined) {
276 return handled;
277 }
278 return handled.pipe(
279 Stream.concat(
280 Stream.fromEffect(Effect.sleep(options.retryExpectedFailureAfter)).pipe(
281 Stream.drain,
282 ),
283 ),
284 Stream.concat(subscribeToSession()),
285 );
286 }
287 return Stream.failCause(cause);
288 }),
272289 ),
273290 );
274291 return subscribeToSession();
packages/client-runtime/src/state/threads-atoms.test.ts+332-6
......@@ -11,6 +11,8 @@ import {
1111 type OrchestrationThreadStreamItem,
1212} from "@t3tools/contracts";
1313import { afterEach, describe, expect, it, vi } from "@effect/vitest";
14import * as Cause from "effect/Cause";
15import * as Clock from "effect/Clock";
1416import * as Deferred from "effect/Deferred";
1517import * as Effect from "effect/Effect";
1618import * as Layer from "effect/Layer";
......@@ -18,7 +20,10 @@ import * as Option from "effect/Option";
1820import * as Queue from "effect/Queue";
1921import * as Stream from "effect/Stream";
2022import * as SubscriptionRef from "effect/SubscriptionRef";
23import * as TestClock from "effect/testing/TestClock";
2124import { Atom, AtomRegistry } from "effect/unstable/reactivity";
25import { RpcClientError } from "effect/unstable/rpc";
26import { Socket } from "effect/unstable/socket";
2227
2328import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
2429import { EnvironmentRegistry } from "../connection/registry.ts";
......@@ -30,6 +35,7 @@ import {
3035 type SupervisorConnectionState,
3136} from "../connection/model.ts";
3237import { EnvironmentSupervisor } from "../connection/supervisor.ts";
38import { ConnectionWakeups, type ConnectionWakeup } from "../connection/wakeups.ts";
3339import { EnvironmentCacheStore } from "../platform/persistence.ts";
3440import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
3541import type { RpcSession } from "../rpc/session.ts";
......@@ -38,6 +44,7 @@ import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
3844import type { ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";
3945import {
4046 createEnvironmentThreadStateAtoms,
47 makeEnvironmentThreadState,
4148 requestOlderThreadTurns,
4249 ThreadSnapshotLoader,
4350 type EnvironmentThreadState,
......@@ -54,7 +61,7 @@ const THREAD: OrchestrationThread = {
5461 id: THREAD_ID,
5562 projectId: ProjectId.make("project-1"),
5663 title: "Cached thread",
57 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
64 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "ModelA" },
5865 runtimeMode: "full-access",
5966 interactionMode: "default",
6067 branch: "main",
......@@ -86,10 +93,15 @@ const CONNECTED_STATE: SupervisorConnectionState = {
8693const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
8794 readonly snapshot?: OrchestrationThreadDetailSnapshot;
8895 readonly connected?: boolean;
96 readonly httpNone?: boolean;
97 readonly initialLoad?: Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;
98 readonly stream?: Stream.Stream<OrchestrationThreadStreamItem, Error>;
8999}) {
100 const clock = yield* Clock.Clock;
101 const wakeups = yield* Queue.unbounded<ConnectionWakeup>();
90102 const subscriptions = yield* Queue.unbounded<{
91103 readonly afterSequence: number | undefined;
92 readonly events: Queue.Queue<OrchestrationThreadStreamItem>;
104 readonly events: Queue.Queue<OrchestrationThreadStreamItem, Error>;
93105 readonly closed: Deferred.Deferred<void>;
94106 }>();
95107 const olderLoads = yield* Queue.unbounded<{
......@@ -106,7 +118,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
106118 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>
107119 Stream.unwrap(
108120 Effect.gen(function* () {
109 const events = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
121 const events = yield* Queue.unbounded<OrchestrationThreadStreamItem, Error>();
110122 const closed = yield* Deferred.make<void>();
111123 yield* Effect.acquireRelease(
112124 Effect.sync(() => {
......@@ -119,7 +131,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
119131 }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))),
120132 );
121133 yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed });
122 return Stream.fromQueue(events);
134 return options?.stream ?? Stream.fromQueue(events);
123135 }),
124136 ),
125137 } as unknown as WsRpcProtocolClient;
......@@ -179,6 +191,8 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
179191 });
180192 const runtime = Atom.runtime(
181193 Layer.mergeAll(
194 Layer.succeed(Clock.Clock, clock),
195 Layer.succeed(ConnectionWakeups, { changes: Stream.fromQueue(wakeups) }),
182196 Layer.succeed(EnvironmentRegistry, environmentRegistry),
183197 Layer.succeed(
184198 EnvironmentCacheStore,
......@@ -208,8 +222,12 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
208222 if (window?.beforeCursor === undefined) {
209223 return Effect.sync(() => {
210224 httpLoads += 1;
211 return Option.some(snapshot);
212 });
225 }).pipe(
226 Effect.andThen(
227 options?.initialLoad ??
228 Effect.succeed(options?.httpNone ? Option.none() : Option.some(snapshot)),
229 ),
230 );
213231 }
214232 return Effect.gen(function* () {
215233 const response =
......@@ -235,6 +253,8 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
235253 const registry = yield* makeRegistry;
236254
237255 return {
256 runtime,
257 supervisor,
238258 registry,
239259 makeRegistry,
240260 rawAtoms: raw,
......@@ -246,6 +266,7 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
246266 connectionState,
247267 session,
248268 sessionRef,
269 wakeups,
249270 counts: () => ({ httpLoads, diskLoads, opened, active }),
250271 };
251272});
......@@ -275,6 +296,311 @@ describe("createEnvironmentThreadStateAtoms", () => {
275296 vi.restoreAllMocks();
276297 });
277298
299 it.effect("exposes snapshot loader defects before the RPC subscription starts", () =>
300 Effect.gen(function* () {
301 const completed = yield* Deferred.make<void>();
302 const h = yield* makeHarness({
303 connected: true,
304 initialLoad: Effect.die(
305 new Error("SYNTHETIC_RAW_SNAPSHOT_DEFECT_SHOULD_NOT_REACH_THREAD_UI"),
306 ).pipe(Effect.ensuring(Deferred.succeed(completed, undefined))),
307 });
308 const unmount = h.registry.mount(h.stateAtom);
309 yield* Deferred.await(completed);
310 const failed = yield* observeState(h.registry, h.stateAtom, (state) =>
311 Option.isSome(state.error),
312 );
313 expect(failed.status).toBe("empty");
314 expect(failed.error).toEqual(Option.some("Could not synchronize the thread."));
315 expect(failed.data).toEqual(Option.none());
316 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 0, active: 0 });
317 yield* TestClock.adjust("1 second");
318 expect(h.registry.get(h.stateAtom)).toEqual(failed);
319 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 0, active: 0 });
320 unmount();
321 }),
322 );
323
324 it.effect.each([
325 { kind: "protocol", httpNone: true },
326 { kind: "protocol", httpNone: false },
327 { kind: "fatal", httpNone: true },
328 { kind: "fatal", httpNone: false },
329 ] as const)(
330 "retains a terminated $kind load diagnostic across connection updates (empty: $httpNone)",
331 ({ kind, httpNone }) =>
332 Effect.gen(function* () {
333 const h = yield* makeHarness({ connected: true, httpNone });
334 const unmount = h.registry.mount(h.stateAtom);
335 const first = yield* Queue.take(h.subscriptions);
336 const error = new Error("SYNTHETIC_RAW_DEFECT_SHOULD_NOT_REACH_THREAD_UI");
337 yield* Queue.failCause(
338 first.events,
339 kind === "fatal"
340 ? Cause.die(error)
341 : Cause.fail(
342 new RpcClientError.RpcClientError({
343 reason: new RpcClientError.RpcClientDefect({
344 message: error.message,
345 cause: error,
346 }),
347 }),
348 ),
349 );
350 yield* Deferred.await(first.closed);
351 // The real finalizer has run; advancing the atom runtime's test clock
352 // also verifies that a defect does not enter the domain retry loop.
353 yield* TestClock.adjust("1 second");
354 const failed = h.registry.get(h.stateAtom);
355 expect(failed.status).toBe(httpNone ? "empty" : "cached");
356 expect(failed.error).toEqual(Option.some("Could not synchronize the thread."));
357 expect(failed.data).toEqual(httpNone ? Option.none() : Option.some(THREAD));
358 expect(h.counts().opened).toBe(1);
359 expect(h.counts().active).toBe(0);
360
361 // Session publication can precede connected, and a fatal child cannot
362 // restart just because its supervisor reconnects.
363 for (const connection of [
364 AVAILABLE_CONNECTION_STATE,
365 { ...CONNECTED_STATE, phase: "connecting" as const },
366 CONNECTED_STATE,
367 ]) {
368 yield* SubscriptionRef.set(h.connectionState, connection);
369 yield* TestClock.adjust("0 millis");
370 expect(h.registry.get(h.stateAtom)).toEqual(failed);
371 }
372 yield* SubscriptionRef.set(h.sessionRef, Option.some({ ...h.session }));
373 if (kind === "fatal") {
374 yield* TestClock.adjust("1 second");
375 expect(h.counts().opened).toBe(1);
376 expect(h.registry.get(h.stateAtom)).toEqual(failed);
377 unmount();
378 return;
379 }
380 const next = yield* Queue.take(h.subscriptions);
381 expect(h.registry.get(h.stateAtom).error).toEqual(Option.none());
382 expect(h.registry.get(h.stateAtom).status).toBe("synchronizing");
383 yield* Queue.offer(next.events, { kind: "snapshot", snapshot: SNAPSHOT });
384 yield* Queue.offer(next.events, { kind: "synchronized" });
385 const recovered = yield* observeState(
386 h.registry,
387 h.stateAtom,
388 (state) => state.status === "live",
389 );
390 expect(recovered.error).toEqual(Option.none());
391 expect(recovered.data).toEqual(Option.some(THREAD));
392 unmount();
393 yield* Deferred.await(next.closed);
394 }),
395 );
396
397 it.effect("retries a protocol failure on foreground without replacing the session", () =>
398 Effect.gen(function* () {
399 const h = yield* makeHarness({ connected: true, httpNone: true });
400 const unmount = h.registry.mount(h.stateAtom);
401 const first = yield* Queue.take(h.subscriptions);
402 yield* Queue.fail(
403 first.events,
404 new RpcClientError.RpcClientError({
405 reason: new RpcClientError.RpcClientDefect({
406 message: "incompatible snapshot",
407 cause: new Error("incompatible snapshot"),
408 }),
409 }),
410 );
411 yield* Deferred.await(first.closed);
412 yield* TestClock.adjust("0 millis");
413 expect(Option.isSome(h.registry.get(h.stateAtom).error)).toBe(true);
414 yield* Queue.offer(h.wakeups, "application-active");
415 const next = yield* Queue.take(h.subscriptions);
416 expect(h.registry.get(h.stateAtom).error).toEqual(Option.none());
417 yield* Queue.offer(next.events, { kind: "snapshot", snapshot: SNAPSHOT });
418 yield* Queue.offer(next.events, { kind: "synchronized" });
419 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
420 unmount();
421 yield* Deferred.await(next.closed);
422 }),
423 );
424
425 it.effect("keeps transport loss nonterminal and recovers with a replacement session", () =>
426 Effect.gen(function* () {
427 const h = yield* makeHarness({ connected: true, httpNone: true });
428 const unmount = h.registry.mount(h.stateAtom);
429 const first = yield* Queue.take(h.subscriptions);
430 yield* Queue.fail(
431 first.events,
432 new RpcClientError.RpcClientError({
433 reason: new Socket.SocketCloseError({ code: 1006, closeReason: "connection lost" }),
434 }),
435 );
436 yield* Deferred.await(first.closed);
437 yield* TestClock.adjust("1 second");
438 expect(h.registry.get(h.stateAtom)).toMatchObject({
439 status: "synchronizing",
440 error: Option.none(),
441 data: Option.none(),
442 });
443 expect(h.counts().opened).toBe(1);
444 yield* SubscriptionRef.set(h.sessionRef, Option.some({ ...h.session }));
445 const next = yield* Queue.take(h.subscriptions);
446 yield* Queue.offer(next.events, { kind: "snapshot", snapshot: SNAPSHOT });
447 yield* Queue.offer(next.events, { kind: "synchronized" });
448 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
449 unmount();
450 yield* Deferred.await(next.closed);
451 }),
452 );
453
454 it.effect("retains ordinary domain error reporting and same-session retries", () =>
455 Effect.gen(function* () {
456 const h = yield* makeHarness({ connected: true, httpNone: true });
457 const unmount = h.registry.mount(h.stateAtom);
458 const first = yield* Queue.take(h.subscriptions);
459 yield* Queue.fail(first.events, new Error("thread not found yet"));
460 yield* Deferred.await(first.closed);
461 const failed = yield* observeState(h.registry, h.stateAtom, (state) =>
462 Option.isSome(state.error),
463 );
464 expect(failed.error).toEqual(Option.some("thread not found yet"));
465 yield* TestClock.adjust("250 millis");
466 const next = yield* Queue.take(h.subscriptions);
467 expect(h.counts().opened).toBe(2);
468 yield* Queue.offer(next.events, { kind: "snapshot", snapshot: SNAPSHOT });
469 yield* Queue.offer(next.events, { kind: "synchronized" });
470 const recovered = yield* observeState(
471 h.registry,
472 h.stateAtom,
473 (state) => state.status === "live",
474 );
475 expect(recovered.error).toEqual(Option.none());
476 unmount();
477 yield* Deferred.await(next.closed);
478 }),
479 );
480
481 it.effect.each([
482 { kind: "protocol", deleted: false },
483 { kind: "fatal", deleted: false },
484 { kind: "domain", deleted: false },
485 { kind: "protocol", deleted: true },
486 ] as const)(
487 "keeps buffered outcomes after a $kind failure (deleted: $deleted)",
488 ({ kind, deleted }) =>
489 Effect.gen(function* () {
490 const burst = yield* Deferred.make<void>();
491 const error = new Error(
492 kind === "domain"
493 ? "buffered thread failure"
494 : "SYNTHETIC_BUFFERED_DEFECT_SHOULD_NOT_REACH_THREAD_UI",
495 );
496 const items: OrchestrationThreadStreamItem[] = [
497 { kind: "snapshot", snapshot: SNAPSHOT },
498 { kind: "synchronized" },
499 {
500 kind: "event",
501 event: {
502 eventId: EventId.make("buffered-event"),
503 commandId: null,
504 causationEventId: null,
505 correlationId: null,
506 metadata: {},
507 sequence: 8,
508 occurredAt: THREAD.createdAt,
509 aggregateKind: "thread",
510 aggregateId: THREAD_ID,
511 type: "thread.meta-updated",
512 payload: {
513 threadId: THREAD_ID,
514 title: "Buffer drained",
515 updatedAt: THREAD.createdAt,
516 },
517 },
518 },
519 ];
520 if (deleted) {
521 items.push({
522 kind: "event",
523 event: {
524 eventId: EventId.make("buffered-deletion"),
525 commandId: null,
526 causationEventId: null,
527 correlationId: null,
528 metadata: {},
529 sequence: 9,
530 occurredAt: THREAD.createdAt,
531 aggregateKind: "thread",
532 aggregateId: THREAD_ID,
533 type: "thread.deleted",
534 payload: { threadId: THREAD_ID, deletedAt: THREAD.createdAt },
535 },
536 });
537 }
538 const failure =
539 kind === "fatal"
540 ? Cause.die(error)
541 : Cause.fail(
542 kind === "domain"
543 ? error
544 : new RpcClientError.RpcClientError({
545 reason: new RpcClientError.RpcClientDefect({
546 message: error.message,
547 cause: error,
548 }),
549 }),
550 );
551 const h = yield* makeHarness({
552 connected: true,
553 httpNone: true,
554 stream: Stream.fromEffect(Deferred.await(burst)).pipe(
555 Stream.flatMap(() => Stream.fromIterable(items)),
556 Stream.concat(Stream.failCause(failure)),
557 ),
558 });
559 yield* Effect.gen(function* () {
560 const state = yield* makeEnvironmentThreadState(THREAD_ID);
561 const initial = yield* Deferred.make<void>();
562 const drained = yield* Deferred.make<void>();
563 yield* SubscriptionRef.changes(state).pipe(
564 Stream.runForEach((value) =>
565 Deferred.succeed(initial, undefined).pipe(
566 Effect.andThen(
567 (
568 deleted
569 ? value.status === "deleted"
570 : Option.getOrNull(value.data)?.title === "Buffer drained"
571 )
572 ? Deferred.succeed(drained, undefined)
573 : Effect.void,
574 ),
575 ),
576 ),
577 Effect.forkScoped,
578 );
579 yield* Deferred.await(initial);
580 const subscription = yield* Queue.take(h.subscriptions);
581 yield* Deferred.succeed(burst, undefined);
582 yield* Deferred.await(subscription.closed);
583 yield* Deferred.await(drained);
584 const final = yield* SubscriptionRef.get(state);
585 if (deleted) {
586 expect(final.status).toBe("deleted");
587 expect(final.data).toEqual(Option.none());
588 expect(final.error).toEqual(Option.none());
589 return;
590 }
591 expect(Option.getOrThrow(final.data).title).toBe("Buffer drained");
592 expect(final.error).toEqual(
593 Option.some(kind === "domain" ? error.message : "Could not synchronize the thread."),
594 );
595 expect(final.status).toBe("cached");
596 }).pipe(
597 Effect.provideService(EnvironmentSupervisor, h.supervisor),
598 Effect.provide(h.registry.get(h.runtime.layer)),
599 Effect.scoped,
600 );
601 }),
602 );
603
278604 it.effect("shares one live stream and closes it after the last detail consumer leaves", () =>
279605 Effect.gen(function* () {
280606 const h = yield* makeHarness();
packages/client-runtime/src/state/threads.ts+23-11
......@@ -310,8 +310,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
310310 Effect.forkScoped,
311311 );
312312
313 const setSynchronizing = SubscriptionRef.update(state, (current) =>
314 current.status === "deleted"
313 const setConnecting = SubscriptionRef.update(state, (current) =>
314 current.status === "deleted" || Option.isSome(current.error)
315315 ? current
316316 : {
317317 ...current,
......@@ -320,7 +320,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
320320 },
321321 );
322322 const setReady = SubscriptionRef.update(state, (current) =>
323 current.status === "live" || current.status === "deleted"
323 current.status === "live" || current.status === "deleted" || Option.isSome(current.error)
324324 ? current
325325 : {
326326 ...current,
......@@ -341,14 +341,14 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
341341 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
342342 }));
343343 });
344 const setStreamError = (cause: Cause.Cause<unknown>) =>
344 const setStreamError = (message: string) =>
345345 Ref.set(awaitingCompletion, false).pipe(
346346 Effect.andThen(
347347 SubscriptionRef.update(state, (current) => ({
348348 ...current,
349349 status:
350350 current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
351 error: Option.some(formatThreadError(cause)),
351 error: Option.some(message),
352352 })),
353353 ),
354354 );
......@@ -362,8 +362,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
362362 const waiting = yield* Ref.get(awaitingCompletion);
363363 yield* SubscriptionRef.update(state, (current) => ({
364364 data: Option.some(thread),
365 status: waiting ? ("synchronizing" as const) : ("live" as const),
366 error: Option.none(),
365 // Buffered values from the failed attempt can still arrive after its error.
366 status: Option.isSome(current.error)
367 ? ("cached" as const)
368 : waiting
369 ? ("synchronizing" as const)
370 : ("live" as const),
371 error: current.error,
367372 page: page === "keep" ? current.page : page,
368373 }));
369374 // Active threads can update many times per second and retain large tool
......@@ -423,7 +428,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
423428 if (item.kind === "synchronized") {
424429 yield* Ref.set(awaitingCompletion, false);
425430 yield* SubscriptionRef.update(state, (current) =>
426 Option.isSome(current.data) && current.status !== "deleted"
431 Option.isSome(current.data) && current.status !== "deleted" && Option.isNone(current.error)
427432 ? { ...current, status: "live" as const, error: Option.none() }
428433 : current,
429434 );
......@@ -639,7 +644,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
639644 Stream.runForEach((connectionState) => {
640645 switch (connectionProjectionPhase(connectionState)) {
641646 case "synchronizing":
642 return setSynchronizing;
647 return setConnecting;
643648 case "disconnected":
644649 return setDisconnected;
645650 case "ready":
......@@ -661,7 +666,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
661666 const resumingLive = yield* Ref.make(initialState.status === "live");
662667 const markSynchronizing = Effect.gen(function* () {
663668 if (yield* Ref.get(resumingLive)) return;
664 yield* setSynchronizing;
669 // Connection notifications do not establish that a terminated load restarted.
670 // Clear its diagnostic only when this subscription actually tries again.
671 yield* SubscriptionRef.update(state, (current) =>
672 current.status === "deleted"
673 ? current
674 : { ...current, status: "synchronizing" as const, error: Option.none() },
675 );
665676 });
666677
667678 yield* markSynchronizing;
......@@ -757,7 +768,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
757768 };
758769 }),
759770 {
760 onExpectedFailure: setStreamError,
771 onDefect: () => setStreamError("Could not synchronize the thread."),
772 onExpectedFailure: (cause) => setStreamError(formatThreadError(cause)),
761773 retryExpectedFailureAfter: "250 millis",
762774 resubscribe: foregroundResubscriptions,
763775 },