diff --git a/.scratch/batch-b/issues/02-u2-transport-timeout-abort.md b/.scratch/batch-b/issues/02-u2-transport-timeout-abort.md index 5983901932..c140dc049e 100644 --- a/.scratch/batch-b/issues/02-u2-transport-timeout-abort.md +++ b/.scratch/batch-b/issues/02-u2-transport-timeout-abort.md @@ -6,10 +6,19 @@ **Evidence:** `.scratch/batch-b/evidence.md#u-2--timeout-传播到底层-http-取消` **Branch:** `test/transport-abort` **Blocked by:** 01（同一 OpenSpec 串行落地） -**Status:** blocked +**Status:** closed -- [ ] fixture 提供“请求已接收”与“response 已取消”的有界 fence，禁止用固定 sleep 猜时序 -- [ ] timeout 后断言现有 `LLMError` Transport/Timeout 形状 -- [ ] 服务端确定性观察到 response stream cancellation；request abort 只作补充信号 -- [ ] 不用内存 HttpClient 或显式 Fiber interrupt 重复现有覆盖 -- [ ] 在 `packages/llm` 连续运行目标测试至少 3 次并运行 `bun typecheck` +- [x] fixture 提供“请求已接收”与“response 已取消”的有界 fence，禁止用固定 sleep 猜时序 +- [x] timeout 后断言现有 `LLMError` Transport/Timeout 形状 +- [x] 服务端确定性观察到 response stream cancellation；request abort 只作补充信号 +- [x] 不用内存 HttpClient 或显式 Fiber interrupt 重复现有覆盖 +- [x] 在 `packages/llm` 连续运行目标测试至少 3 次并运行 `bun typecheck` + +## 验证证据 + +- 基线：`dev@55dd345491de4542dbf6fa7a4ba126a2c23104c4`；分支：`test/transport-abort`。 +- 实现提交：`8ec1ef192`；PR：[LeXwDeX/OpenCode-GraphAgent#193](https://github.com/LeXwDeX/OpenCode-GraphAgent/pull/193) → `dev`。 +- 真实 transport：公开 `LLMClient.stream(...)` 经 `FetchHttpClient.layer` 请求 loopback `Bun.serve`；2 秒有界 fence 分别证明请求已接收与服务端 response `cancel()` 已触发。 +- mutation 红灯：临时移除 response stream 的 `Stream.timeoutOrElse` 后，新增场景在 1 秒测试边界超时，0 pass / 1 fail；mutation 已恢复，生产文件无 diff。 +- `cd packages/llm && bun test test/transport-timeout.test.ts --timeout 30000`：连续 3 次均为 7 pass、0 fail、14 expect；`bun typecheck`：`tsgo --noEmit`，exit 0。 +- 现有生产实现满足 OpenSpec Requirement 2，无生产代码修改；Requirement 3 留给 03 票。 diff --git a/.scratch/batch-b/issues/03-transport-midstream-stall.md b/.scratch/batch-b/issues/03-transport-midstream-stall.md index ebc3f10eee..5303984f16 100644 --- a/.scratch/batch-b/issues/03-transport-midstream-stall.md +++ b/.scratch/batch-b/issues/03-transport-midstream-stall.md @@ -5,8 +5,8 @@ **Spec:** `.scratch/batch-b/abort-path-contracts.md` 的 Requirement 3 **Evidence:** `.scratch/batch-b/evidence.md#transport-mid-stream-stall` **Branch:** `test/midstream-timeout` -**Blocked by:** 02（共同修改 `packages/llm/test/transport-timeout.test.ts`） -**Status:** blocked +**Blocked by:** None（02 已完成） +**Status:** ready-for-agent - [ ] 用 fence 证明 timeout 前合法首帧已经交付给消费者 - [ ] 用 TestClock 越过下一帧间隔，随后得到现有 Transport/Timeout diff --git a/packages/llm/test/transport-timeout.test.ts b/packages/llm/test/transport-timeout.test.ts index bb1e19f2e1..b62d05dd99 100644 --- a/packages/llm/test/transport-timeout.test.ts +++ b/packages/llm/test/transport-timeout.test.ts @@ -1,20 +1,28 @@ import { describe, expect, test } from "bun:test" import { Cause, Duration, Effect, Exit, Fiber, Option, Stream } from "effect" +import { FetchHttpClient } from "effect/unstable/http" import * as TestClock from "effect/testing/TestClock" import { LLM, LLMError, LLMEvent } from "../src" import * as OpenAIChat from "../src/protocols/openai-chat" import { HttpOptions, Model, mergeHttpOptions } from "../src/schema" import { LLMClient } from "../src/route" import { testEffect } from "./lib/effect" -import { dynamicResponse, fixedResponse } from "./lib/http" +import { dynamicResponse, fixedResponse, runtimeLayer } from "./lib/http" import { deltaChunk } from "./lib/openai-chunks" import { sseEvents } from "./lib/sse" const model = Model.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route }) -const request = (timeout?: number) => +const request = (timeout?: number, baseURL?: string) => LLM.request({ - model, + model: + baseURL === undefined + ? model + : Model.make({ + id: "fake-model", + provider: "fake", + route: OpenAIChat.route.with({ endpoint: { baseURL } }), + }), prompt: "Say hello.", http: timeout === undefined ? undefined : { timeout: Duration.millis(timeout) }, }) @@ -37,7 +45,59 @@ const expectTimeoutExit = (exit: Exit.Exit) => { expect(error.reason).toMatchObject({ _tag: "Transport", kind: "Timeout" }) } +const waitForFence = (name: string, promise: Promise) => + Effect.promise(() => promise).pipe( + Effect.timeout(Duration.seconds(2)), + Effect.mapError(() => new Error(`${name} was not observed within 2000ms`)), + ) + +const timeoutProvider = () => { + const requestReceived = Promise.withResolvers() + const responseCanceled = Promise.withResolvers() + const encoder = new TextEncoder() + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch() { + requestReceived.resolve() + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(": connected\n\n")) + }, + cancel() { + responseCanceled.resolve() + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ) + }, + }) + return { server, requestReceived: requestReceived.promise, responseCanceled: responseCanceled.promise } +} + +const networkRuntime = runtimeLayer(FetchHttpClient.layer) + describe("http transport timeout", () => { + testEffect(networkRuntime).live( + "cancels the provider response stream when a real HTTP request times out", + () => + Effect.gen(function* () { + const provider = yield* Effect.acquireRelease( + Effect.sync(timeoutProvider), + (fixture) => Effect.promise(() => fixture.server.stop(true)), + ) + const fiber = yield* LLMClient.stream(request(200, provider.server.url.origin)).pipe( + Stream.runCollect, + Effect.forkScoped, + ) + + yield* waitForFence("provider request", provider.requestReceived) + expectTimeoutExit(yield* Fiber.join(fiber).pipe(Effect.exit)) + yield* waitForFence("provider response cancellation", provider.responseCanceled) + }), + ) + testEffect(hangingHeaders).effect( "ends the stream with a Timeout error when the provider never sends response headers", () =>