mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-30 06:02:11 +00:00
test(core): synchronize retry tests on scheduled events (#46027)
This commit is contained in:
Kit Langton
GitHub
parent
803b7718b8
commit
87525e00b9
1 changed files with 51 additions and 45 deletions
|
|
@ -584,6 +584,18 @@ const scenario = (
|
|||
}),
|
||||
)
|
||||
|
||||
// Subscribe before resuming; model requests can arrive before retry backoff is scheduled.
|
||||
const subscribeRetries = (s: Scenario) =>
|
||||
Effect.gen(function* () {
|
||||
const scheduled = yield* Queue.unbounded<SessionMessage.ID>()
|
||||
yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
|
||||
Stream.filter((event) => event.data.sessionID === sessionID),
|
||||
Stream.runForEach((event) => Queue.offer(scheduled, event.data.assistantMessageID)),
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
return scheduled
|
||||
})
|
||||
|
||||
const providerUnavailable = () =>
|
||||
new AIError({
|
||||
reason: new TransportError({
|
||||
|
|
@ -4343,14 +4355,10 @@ describe("SessionRunnerLLM", () => {
|
|||
yield* s.admit("Retry transport")
|
||||
yield* s.llm.push(Stream.fail(providerUnavailable()))
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "retry-success"))
|
||||
const scheduled = yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
|
||||
Stream.filter((event) => event.data.sessionID === sessionID),
|
||||
Stream.runHead,
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* Fiber.join(scheduled)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("1599 millis")
|
||||
expect(s.requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("801 millis")
|
||||
|
|
@ -4425,14 +4433,10 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.admit("Use custom retry delay")
|
||||
yield* s.llm.push(Stream.fail(providerUnavailable()), TestLLM.text("Recovered", "hook-delay-success"))
|
||||
const scheduled = yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
|
||||
Stream.filter((event) => event.data.sessionID === sessionID),
|
||||
Stream.runHead,
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* Fiber.join(scheduled)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("4999 millis")
|
||||
expect(s.requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("1 millis")
|
||||
|
|
@ -4446,13 +4450,9 @@ describe("SessionRunnerLLM", () => {
|
|||
scenario("does not start another physical attempt after interruption during retry backoff", function* (s) {
|
||||
yield* s.admit("Interrupt retry backoff")
|
||||
yield* s.llm.push(Stream.fail(providerUnavailable()), TestLLM.text("Must not run", "unused-retry"))
|
||||
const scheduled = yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
|
||||
Stream.filter((event) => event.data.sessionID === sessionID),
|
||||
Stream.runHead,
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* Fiber.join(scheduled)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* s.session.interrupt(sessionID)
|
||||
const exit = yield* Fiber.await(run)
|
||||
expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
|
||||
|
|
@ -4492,8 +4492,9 @@ describe("SessionRunnerLLM", () => {
|
|||
yield* s.llm.push(Stream.fail(incompleteStream()))
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "incomplete-stream-success"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4513,8 +4514,9 @@ describe("SessionRunnerLLM", () => {
|
|||
])
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "unknown-finish-success"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4531,8 +4533,9 @@ describe("SessionRunnerLLM", () => {
|
|||
yield* s.llm.push(Stream.fail(rateLimited(5_000)))
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "retry-after-success"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("4999 millis")
|
||||
expect(s.requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("1 millis")
|
||||
|
|
@ -4545,8 +4548,9 @@ describe("SessionRunnerLLM", () => {
|
|||
yield* s.llm.push(Stream.fail(rateLimited(3_600_000)))
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "retry-cap-success"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("899999 millis")
|
||||
expect(s.requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("1 millis")
|
||||
|
|
@ -4567,8 +4571,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text(" continuation", "continued-text"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4616,8 +4621,9 @@ describe("SessionRunnerLLM", () => {
|
|||
])
|
||||
yield* s.llm.push(TestLLM.text(" continuation", "unknown-continuation"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4646,8 +4652,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text(" continuation", "rate-limit-continuation"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("4999 millis")
|
||||
expect(s.requests).toHaveLength(1)
|
||||
yield* TestClock.adjust("1 millis")
|
||||
|
|
@ -4684,8 +4691,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text(" continuation", "unknown-failure-continuation"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4714,8 +4722,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "reasoning-recovery"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4756,8 +4765,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "reasoning-transport-recovery"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4792,9 +4802,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "tool-recovery"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4828,9 +4838,9 @@ describe("SessionRunnerLLM", () => {
|
|||
)
|
||||
yield* s.llm.push(TestLLM.text("Recovered", "tool-defect-recovery"))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
while (!(yield* recordedEventTypes(sessionID)).includes("session.retry.scheduled.1")) yield* Effect.yieldNow
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
@ -4851,12 +4861,7 @@ describe("SessionRunnerLLM", () => {
|
|||
scenario(
|
||||
"shares retry accounting and assistant identity across transparent retries and partial continuations",
|
||||
function* (s) {
|
||||
const scheduled = yield* Queue.unbounded<SessionMessage.ID>()
|
||||
yield* s.bus.subscribe(SessionEvent.RetryScheduled).pipe(
|
||||
Stream.filter((event) => event.data.sessionID === sessionID),
|
||||
Stream.runForEach((event) => Queue.offer(scheduled, event.data.assistantMessageID)),
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
yield* s.admit("Mix retry paths")
|
||||
const failure = incompleteStream()
|
||||
const partial = TestLLM.failAfter(
|
||||
|
|
@ -4898,11 +4903,11 @@ describe("SessionRunnerLLM", () => {
|
|||
),
|
||||
)
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) {
|
||||
for (const delay of [2_400, 4_800, 9_600, 19_200]) {
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust(delay)
|
||||
yield* s.llm.wait(index + 2)
|
||||
}
|
||||
expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure)
|
||||
expect(s.requests).toHaveLength(5)
|
||||
|
|
@ -4916,11 +4921,11 @@ describe("SessionRunnerLLM", () => {
|
|||
const failure = providerUnavailable()
|
||||
yield* s.llm.always(Stream.fail(failure))
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
for (const [index, delay] of [2_400, 4_800, 9_600, 19_200].entries()) {
|
||||
for (const delay of [2_400, 4_800, 9_600, 19_200]) {
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust(delay)
|
||||
yield* s.llm.wait(index + 2)
|
||||
}
|
||||
expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure)
|
||||
expect(s.requests).toHaveLength(5)
|
||||
|
|
@ -4965,8 +4970,9 @@ describe("SessionRunnerLLM", () => {
|
|||
yield* s.llm.push(Stream.fail(failure))
|
||||
yield* s.llm.push(TestLLM.tool("call-after-retry", "echo", { text: "recovered" }), TestLLM.stop())
|
||||
|
||||
const scheduled = yield* subscribeRetries(s)
|
||||
const run = yield* s.resume.pipe(Effect.forkChild)
|
||||
yield* s.llm.wait(1)
|
||||
yield* Queue.take(scheduled)
|
||||
yield* TestClock.adjust("2400 millis")
|
||||
yield* Fiber.join(run)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue