diff --git a/packages/spill/spill-policy/tests/spill-policy.spec.ts b/packages/spill/spill-policy/tests/spill-policy.spec.ts index add76c1882..db559bc338 100644 --- a/packages/spill/spill-policy/tests/spill-policy.spec.ts +++ b/packages/spill/spill-policy/tests/spill-policy.spec.ts @@ -329,70 +329,82 @@ describe('the durable dispatch-log arm', () => { await ctx.plugin(StubStore) await ctx.plugin(SpillPolicy, { maxInlineBytes: 100 }) await mountRuntime(ctx, {}) - // A spill backend that hangs until released. - let releaseSave!: () => void - const gate = new Promise((resolve) => { releaseSave = resolve }) + const saveStarted = Promise.withResolvers() + const saveGate = Promise.withResolvers() + const smallStarted = Promise.withResolvers() const store = ctx.spillStore as StubStore const realSave = store.saveText.bind(store) store.saveText = async (input) => { - await gate + saveStarted.resolve(undefined) + await saveGate.promise return realSave(input) } const events: { type: string; data: unknown }[] = [] - const agent = observedAgent(ctx, 'dispatch-slow-spill', (type: string, data: unknown) => { events.push({ type, data }) }) + const agent = observedAgent(ctx, 'dispatch-slow-spill', (type: string, data: unknown) => { + events.push({ type, data }) + if (type === 'tool/ptc-dispatch-start' && (data as { name: string }).name === 'small_read') smallStarted.resolve(undefined) + }) ctx.tools.register(textTool('huge_read', 'H'.repeat(2_000))) ctx.tools.register(textTool('small_read', 'tiny')) - let smallAfterHuge = false const runPromise = ctx.tools.execute({ signal: testToolSignal, callId: ToolCallId('parent-3'), name: 'run_code', arguments: { - // The program takes BOTH values while the spill backend hangs: the - // huge read's binding resolves immediately (its logged copy is side - // work), so the small read proceeds without waiting. + // Both program values remain available while the durable spill is blocked. code: 'const big = await tools.huge_read({});\nconst small = await tools.small_read({});\nreturn big[0].text.length + small[0].text.length', description: 'Prove log shaping is off the program path', }, agent: agent as never, - }).then((result) => { - return result }) - // The run cannot COMPLETE while the settle append is gated (drain waits - // for logWork), but the program itself already ran both calls; release - // the backend and observe the settle events land inside the turn. - await vi.waitFor(() => { - // The second dispatch STARTED while the first one's spill hung. - smallAfterHuge = events.some(event => event.type === 'tool/ptc-dispatch-start' - && (event.data as { name: string }).name === 'small_read') - if (!smallAfterHuge) throw new Error('small_read not started yet') - }) - releaseSave() - const result = await runPromise - expect(result.isError).toBe(false) - if (result.isError) throw new Error('expected success') - expect(result.value).toMatchObject({ result: 2_004 }) - const settles = events.filter(event => event.type === 'tool/ptc-dispatch') - expect(settles).toHaveLength(2) - expect(smallAfterHuge).toBe(true) + onTestFinished(() => { saveGate.resolve(undefined) }) + try { + await Promise.race([ + Promise.all([saveStarted.promise, smallStarted.promise]), + runPromise.then(() => { throw new Error('run finished before the blocked save and later dispatch overlapped') }), + ]) + saveGate.resolve(undefined) + const result = await runPromise + expect(result.isError).toBe(false) + if (result.isError) throw new Error('expected success') + expect(result.value).toMatchObject({ result: 2_004 }) + expect(events.filter(event => event.type === 'tool/ptc-dispatch')).toHaveLength(2) + } finally { + saveGate.resolve(undefined) + await runPromise + } }) it('a sustained slow backend backpressures the run instead of accumulating unbounded log tasks', async () => { const ctx = new Context() await ctx.plugin(SystemPrompt) - // Cap 1: once the hung shaped-append backlog exceeds the cap, the ordered - // lane holds inside the second commit, so the THIRD dispatch cannot start - // until a pending save drains — the bound is observable as its missing - // start event. await ctx.plugin(ToolRuntime, { mode: 'ptc', maxParallelSubCalls: 1 }) await ctx.plugin(StubStore) await ctx.plugin(SpillPolicy, { maxInlineBytes: 100 }) await mountRuntime(ctx, {}) + const secondSaveStarted = Promise.withResolvers() + const thirdStarted = Promise.withResolvers() const store = ctx.spillStore as StubStore const releases: (() => void)[] = [] - store.gate = () => new Promise((resolve) => { releases.push(resolve) }) + let gateNewSaves = true + let saves = 0 + store.gate = () => { + if (!gateNewSaves) return Promise.resolve() + return new Promise((resolve) => { + releases.push(resolve) + if (++saves === 2) secondSaveStarted.resolve(undefined) + }) + } + // Disabling future gates also releases a third save that has not reached the backend yet. + const releaseSaves = (): void => { + gateNewSaves = false + for (const release of releases.splice(0)) release() + } const events: { type: string; data: unknown }[] = [] - const agent = observedAgent(ctx, 'dispatch-spill-bound', (type: string, data: unknown) => { events.push({ type, data }) }) + const agent = observedAgent(ctx, 'dispatch-spill-bound', (type: string, data: unknown) => { + events.push({ type, data }) + if (type === 'tool/ptc-dispatch-start' && (data as { subCallId: string }).subCallId.endsWith(':ptc:3')) thirdStarted.resolve(undefined) + }) ctx.tools.register(textTool('huge_read', 'H'.repeat(2_000))) const started = (n: number): boolean => events.some(event => event.type === 'tool/ptc-dispatch-start' && (event.data as { subCallId: string }).subCallId.endsWith(`:ptc:${n}`)) @@ -401,32 +413,33 @@ describe('the durable dispatch-log arm', () => { callId: ToolCallId('parent-bound'), name: 'run_code', arguments: { - code: 'await tools.huge_read({}); await tools.huge_read({}); await tools.huge_read({}); return "done"', + // Queue all requests before waiting so the third is available to the bounded lane. + code: 'await Promise.all([tools.huge_read({}), tools.huge_read({}), tools.huge_read({})]); return "done"', description: 'Three oversized reads against a hung backend', }, agent: agent as never, }) - // Two hung saves = backlog above the cap: the lane must hold before - // starting dispatch 3. - await vi.waitFor(() => { - if (releases.length < 2) throw new Error('second hung save not reached yet') - }) - expect(started(2)).toBe(true) - expect(started(3)).toBe(false) - releases.shift()!() - // Draining one pending save releases the lane; dispatch 3 starts. - await vi.waitFor(() => { - if (!started(3)) throw new Error('third dispatch not started yet') - }) - while (releases.length > 0) releases.shift()!() - const result = await runPromise - expect(result.isError).toBe(false) - await vi.waitFor(() => { - if (releases.length > 0) { while (releases.length > 0) releases.shift()!() } - if (events.filter(event => event.type === 'tool/ptc-dispatch').length !== 3) { - throw new Error('settle events still pending') - } - }) + onTestFinished(releaseSaves) + try { + await Promise.race([ + secondSaveStarted.promise, + runPromise.then(() => { throw new Error('run finished before two spill saves were blocked') }), + ]) + expect(started(2)).toBe(true) + expect(started(3)).toBe(false) + releases.shift()!() + await Promise.race([ + thirdStarted.promise, + runPromise.then(() => { throw new Error('run finished before the third dispatch started') }), + ]) + releaseSaves() + const result = await runPromise + expect(result.isError).toBe(false) + expect(events.filter(event => event.type === 'tool/ptc-dispatch')).toHaveLength(3) + } finally { + releaseSaves() + await runPromise + } }) it('a saveText failure keeps the complete content in the durable log (best-effort)', async () => { diff --git a/scripts/prepare-ci-bubblewrap.sh b/scripts/prepare-ci-bubblewrap.sh index 1efaee200a..5b19d8ad6f 100755 --- a/scripts/prepare-ci-bubblewrap.sh +++ b/scripts/prepare-ci-bubblewrap.sh @@ -4,9 +4,11 @@ set -euo pipefail # Ubuntu's package transaction scans the hosted image's full dpkg database and # runs post-install hooks. CI needs only the signed-archive payload, so pin and # verify that payload before extracting it into the ephemeral runner directory. -readonly BUBBLEWRAP_VERSION='0.9.0-1ubuntu0.1' -readonly BUBBLEWRAP_SHA256='1b506492bd9c7fd0cdb4f02ac822f1d3e336b0aead5113c1239baf8db5db562a' -readonly BUBBLEWRAP_URL="https://archive.ubuntu.com/ubuntu/pool/main/b/bubblewrap/bubblewrap_${BUBBLEWRAP_VERSION}_amd64.deb" +# This amd64 build needs glibc 2.38+ and Linux 5.10+; Ubuntu 24.04 satisfies both. +readonly BUBBLEWRAP_VERSION='0.12.0-1' +readonly BUBBLEWRAP_SHA256='db4d572a7927bfd34cc9b8e24b56983efa8e0483d03da6c810f6caae081dcc36' +# The recorded build remains addressable after Ubuntu prunes its live package pool. +readonly BUBBLEWRAP_URL="https://launchpad.net/ubuntu/+source/bubblewrap/${BUBBLEWRAP_VERSION}/+build/33546835/+files/bubblewrap_${BUBBLEWRAP_VERSION}_amd64.deb" : "${RUNNER_TEMP:?prepare-ci-bubblewrap requires RUNNER_TEMP}" : "${GITHUB_PATH:?prepare-ci-bubblewrap requires GITHUB_PATH}"