| ... | ... | @@ -431,6 +431,60 @@ describe("AgentAwarenessRelay", () => { |
| 431 | 431 | }), |
| 432 | 432 | ); |
| 433 | 433 | |
| 434 | it.effect("a stalled response body releases later threads and uses bounded retries", () => |
| 435 | Effect.gen(function* () { |
| 436 | const bodyStarted = Promise.withResolvers<void>(); |
| 437 | const secondRead = yield* Deferred.make<void>(); |
| 438 | const currentShell = yield* Ref.make(shell()); |
| 439 | const { relay, publications } = yield* makeTestRelay({ |
| 440 | readShell: (threadId) => |
| 441 | threadId === SECOND_THREAD_ID |
| 442 | ? Deferred.succeed(secondRead, undefined).pipe( |
| 443 | Effect.as(shell({ id: SECOND_THREAD_ID })), |
| 444 | ) |
| 445 | : Ref.get(currentShell), |
| 446 | respond: (attempt) => { |
| 447 | if (attempt !== 1) return new Response("relay unavailable", { status: 503 }); |
| 448 | const response = new Response(null, { |
| 449 | headers: { "content-type": "application/json" }, |
| 450 | }); |
| 451 | return Object.assign(response, { |
| 452 | arrayBuffer: () => { |
| 453 | bodyStarted.resolve(); |
| 454 | return new Promise<ArrayBuffer>(() => {}); |
| 455 | }, |
| 456 | }); |
| 457 | }, |
| 458 | }); |
| 459 | const first = yield* relay.publishThread(THREAD_ID).pipe(Effect.forkChild); |
| 460 | yield* Effect.promise(() => bodyStarted.promise); |
| 461 | const second = yield* relay.publishThread(SECOND_THREAD_ID).pipe(Effect.forkChild); |
| 462 | yield* TestClock.adjust("9999 millis"); |
| 463 | assert.equal(publications.length, 1); |
| 464 | yield* Ref.set(currentShell, shell({ title: "Latest title" })); |
| 465 | yield* TestClock.adjust("1 millis"); |
| 466 | // Assert progress before draining so the base fails instead of hanging. |
| 467 | assert.isTrue(yield* Deferred.isDone(secondRead)); |
| 468 | yield* Fiber.join(first); |
| 469 | yield* Fiber.join(second); |
| 470 | assert.equal(publications.length, 2); |
| 471 | assert.isTrue(publications[1]?.url.includes(SECOND_THREAD_ID) ?? false); |
| 472 | for (const delay of [1_000, 2_000, 4_000, 8_000, 16_000]) { |
| 473 | const attempts = publications.length; |
| 474 | yield* TestClock.adjust(delay - 1); |
| 475 | yield* relay.drain; |
| 476 | assert.equal(publications.length, attempts); |
| 477 | yield* TestClock.adjust(1); |
| 478 | yield* relay.drain; |
| 479 | assert.equal(publications.length, attempts + 2); |
| 480 | assert.equal(publications[attempts]?.state?.threadTitle, "Latest title"); |
| 481 | } |
| 482 | yield* TestClock.adjust("1 minute"); |
| 483 | yield* relay.drain; |
| 484 | assert.equal(publications.length, 12); |
| 485 | }), |
| 486 | ); |
| 487 | |
| 434 | 488 | it.effect("retries a failed final state without another thread event", () => |
| 435 | 489 | Effect.gen(function* () { |
| 436 | 490 | const { relay, currentShell, shellReads, publications } = yield* makeTestRelay({ |
Alex Southwell <saphid@gmail.com> 2026-10-14 19:04:46+11:00
GitHub <noreply@github.com> 2026-10-14 01:04:46-07:00