OpenAgents

Commit

coder: add exponential backoff retries for transient responses API failures (#54)

1b3fee9a955f · AtlantisPleb · · parent 858c9a68fdf2 

coder: add exponential backoff retries for transient responses API failures (#54)

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

Not deployed through the forge lane

No push, promotion, build, or deploy receipt references this commit (receipts are scanned over a bounded recent window). Changes shipped by full node replacement carry their proof in the release gate receipt instead.

Changed files

  • modified packages/openagents-cli/src/coder-responses.ts
  • modified packages/openagents-cli/test/coder-responses.test.ts

Diff

2 files changed, +169 -39

packages/openagents-cli/src/coder-responses.ts modified +83 -35

@@ -21,11 +21,16 @@ import type { ReplyChunk, ReplySource } from "./coder-session.js";

21 21 
import { tierLabel } from "./coder-tiers.js";
22 22 
import type { CoderTool } from "./coder-tools.js";
23 23
24 
/** Default backoff ladder for transient responses API failures (5xx or network drops). */
25 
const DEFAULT_RETRY_DELAYS_MS: ReadonlyArray<number> = [250, 500, 1000, 2000];
26
24 27 
export interface ResponsesOptions {
25 28 
  /** The API origin, such as `http://localhost:4000`. */
26 29 
  readonly origin: string;
27 30 
  /** The account bearer, sent when held; the surface also answers without one. */
28 31 
  readonly token?: string | undefined;
32 
  /** Retry backoff delays in milliseconds for testing or custom ladders. */
33 
  readonly retryDelaysMs?: ReadonlyArray<number> | undefined;
29 34 
}
30 35
31 36 
/** One conversation item, in the OpenResponses input shape. */

@@ -56,8 +61,11 @@ export class ResponsesReplySource implements ReplySource {

56 61 
  private readonly items: Item[] = [];
57 62 
  private tools: ReadonlyArray<CoderTool> = [];
58 63 
  private standing: string | undefined;
64 
  private readonly retryDelaysMs: ReadonlyArray<number>;
59 65
60 
  constructor(private readonly options: ResponsesOptions) {}
66 
  constructor(private readonly options: ResponsesOptions) {
67 
    this.retryDelaysMs = options.retryDelaysMs ?? DEFAULT_RETRY_DELAYS_MS;
68 
  }
61 69
62 70 
  /** The dev lane defers to the server, which is what Coder Auto names. */
63 71 
  get model(): string {

@@ -120,48 +128,88 @@ export class ResponsesReplySource implements ReplySource {

120 128 
    yield { type: "usage", calls };
121 129 
  }
122 130
131 
  /**
132 
   * Request POST /api/v1/responses with exponential backoff retries for transient failures.
133 
   */
134 
  private async request(signal: AbortSignal): Promise<Response> {
135 
    const attempts = this.retryDelaysMs.length + 1;
136
137 
    for (let attempt = 0; attempt < attempts; attempt += 1) {
138 
      let response: Response | undefined;
139 
      let error: unknown;
140
141 
      try {
142 
        response = await fetch(new URL("/api/v1/responses", this.options.origin), {
143 
          method: "POST",
144 
          headers: {
145 
            ...(this.options.token === undefined
146 
              ? {}
147 
              : { authorization: `Bearer ${this.options.token}` }),
148 
            "content-type": "application/json",
149 
            // Both named: the pipeline negotiates on json, the answer is SSE.
150 
            accept: "text/event-stream, application/json",
151 
          },
152 
          body: JSON.stringify({
153 
            input: this.items,
154 
            stream: true,
155 
            ...(this.standing === undefined ? {} : { instructions: this.standing }),
156 
            ...(this.tools.length === 0
157 
              ? {}
158 
              : {
159 
                  tools: this.tools.map((tool) => ({
160 
                    type: "function",
161 
                    name: tool.name,
162 
                    description: tool.description,
163 
                    parameters: tool.parameters,
164 
                  })),
165 
                }),
166 
          }),
167 
          signal,
168 
        });
169 
      } catch (cause) {
170 
        if (signal.aborted) throw cause;
171 
        error = cause;
172 
      }
173
174 
      if (response !== undefined && response.ok && response.body !== null) {
175 
        return response;
176 
      }
177
178 
      const status = response?.status;
179 
      const isTransient =
180 
        error !== undefined || (status !== undefined && status >= 500 && status < 600);
181
182 
      if (!isTransient || attempt === attempts - 1) {
183 
        if (response !== undefined) {
184 
          throw new Error(
185 
            `The responses API at ${this.options.origin} answered HTTP ${String(response.status)}.`,
186 
          );
187 
        }
188 
        throw new Error(
189 
          `The responses API at ${this.options.origin} could not be reached: ${String(error)}`,
190 
        );
191 
      }
192
193 
      const delayMs = this.retryDelaysMs[attempt] ?? 1000;
194 
      await new Promise((resolve) => setTimeout(resolve, delayMs));
195 
      if (signal.aborted) {
196 
        throw new Error("Aborted");
197 
      }
198 
    }
199
200 
    throw new Error(`The responses API at ${this.options.origin} request failed.`);
201 
  }
202
123 203 
  /** One request: stream the events, yield what renders, return the rest. */
124 204 
  private async *once(
125 205 
    signal: AbortSignal,
126 206 
  ): AsyncGenerator<ReplyChunk, { text: string; requested: Call[] }> {
127 
    const response = await fetch(new URL("/api/v1/responses", this.options.origin), {
128 
      method: "POST",
129 
      headers: {
130 
        ...(this.options.token === undefined
131 
          ? {}
132 
          : { authorization: `Bearer ${this.options.token}` }),
133 
        "content-type": "application/json",
134 
        // Both named: the pipeline negotiates on json, the answer is SSE.
135 
        accept: "text/event-stream, application/json",
136 
      },
137 
      body: JSON.stringify({
138 
        input: this.items,
139 
        stream: true,
140 
        ...(this.standing === undefined ? {} : { instructions: this.standing }),
141 
        ...(this.tools.length === 0
142 
          ? {}
143 
          : {
144 
              tools: this.tools.map((tool) => ({
145 
                type: "function",
146 
                name: tool.name,
147 
                description: tool.description,
148 
                parameters: tool.parameters,
149 
              })),
150 
            }),
151 
      }),
152 
      signal,
153 
    });
154
155 
    if (!response.ok || response.body === null) {
156 
      throw new Error(
157 
        `The responses API at ${this.options.origin} answered HTTP ${String(response.status)}.`,
158 
      );
159 
    }
207 
    const response = await this.request(signal);
160 208
161 209 
    let text = "";
162 210 
    const requested: Call[] = [];
163 211
164 
    for await (const data of frames(response.body, signal)) {
212 
    for await (const data of frames(response.body!, signal)) {
165 213 
      const event = parse(data);
166 214 
      if (event === undefined) continue;
167 215
packages/openagents-cli/test/coder-responses.test.ts modified +86 -4

@@ -8,7 +8,10 @@ const sse = (frames: ReadonlyArray<string>) =>

8 8 
    new ReadableStream<Uint8Array>({
9 9 
      start(controller) {
10 10 
        const encoder = new TextEncoder();
11 
        for (const frame of frames) controller.enqueue(encoder.encode(frame));
11 
        for (let i = 0; i < frames.length; i += 1) {
12 
          const frame = frames[i];
13 
          if (frame !== undefined) controller.enqueue(encoder.encode(frame));
14 
        }
12 15 
        controller.close();
13 16 
      },
14 17 
    }),

@@ -76,11 +79,90 @@ describe("ResponsesReplySource", () => {

76 79 
    expect(source.modelId).toBe("openagents-coder");
77 80 
  });
78 81
79 
  it("reports a refusal with the origin and the status", async () => {
80 
    vi.stubGlobal("fetch", vi.fn(async () => new Response("no", { status: 503 })));
81 
    const source = new ResponsesReplySource({ origin: "http://localhost:4000" });
82 
  it("reports a refusal with the origin and the status when retries exhaust", async () => {
83 
    let callCount = 0;
84 
    vi.stubGlobal(
85 
      "fetch",
86 
      vi.fn(async () => {
87 
        callCount += 1;
88 
        return new Response("no", { status: 503 });
89 
      }),
90 
    );
91 
    const source = new ResponsesReplySource({
92 
      origin: "http://localhost:4000",
93 
      retryDelaysMs: [1, 1, 1],
94 
    });
82 95
83 96 
    await expect(collect(source)).rejects.toThrow("http://localhost:4000 answered HTTP 503");
97 
    expect(callCount).toBe(4);
98 
  });
99
100 
  it("does not retry 4xx errors", async () => {
101 
    let callCount = 0;
102 
    vi.stubGlobal(
103 
      "fetch",
104 
      vi.fn(async () => {
105 
        callCount += 1;
106 
        return new Response("bad request", { status: 400 });
107 
      }),
108 
    );
109 
    const source = new ResponsesReplySource({
110 
      origin: "http://localhost:4000",
111 
      retryDelaysMs: [1, 1, 1],
112 
    });
113
114 
    await expect(collect(source)).rejects.toThrow("http://localhost:4000 answered HTTP 400");
115 
    expect(callCount).toBe(1);
116 
  });
117
118 
  it("retries transient 5xx server errors and succeeds when server recovers", async () => {
119 
    let callCount = 0;
120 
    vi.stubGlobal(
121 
      "fetch",
122 
      vi.fn(async () => {
123 
        callCount += 1;
124 
        if (callCount < 3) {
125 
          return new Response("server down briefly", { status: 500 });
126 
        }
127 
        return sse(STREAM);
128 
      }),
129 
    );
130
131 
    const source = new ResponsesReplySource({
132 
      origin: "http://localhost:4000",
133 
      retryDelaysMs: [1, 1, 1],
134 
    });
135 
    const chunks = await collect(source);
136 
    expect(callCount).toBe(3);
137 
    const text = chunks
138 
      .map((chunk) => (chunk.type === "text" ? chunk.value : ""))
139 
      .join("");
140 
    expect(text).toBe("Acknowledged.");
141 
  });
142
143 
  it("retries fetch network reachability failures and recovers", async () => {
144 
    let callCount = 0;
145 
    vi.stubGlobal(
146 
      "fetch",
147 
      vi.fn(async () => {
148 
        callCount += 1;
149 
        if (callCount === 1) {
150 
          throw new TypeError("fetch failed: ECONNREFUSED");
151 
        }
152 
        return sse(STREAM);
153 
      }),
154 
    );
155
156 
    const source = new ResponsesReplySource({
157 
      origin: "http://localhost:4000",
158 
      retryDelaysMs: [1, 1, 1],
159 
    });
160 
    const chunks = await collect(source);
161 
    expect(callCount).toBe(2);
162 
    const text = chunks
163 
      .map((chunk) => (chunk.type === "text" ? chunk.value : ""))
164 
      .join("");
165 
    expect(text).toBe("Acknowledged.");
84 166 
  });
85 167 
});
86 168

This page updates live while a promote is in flight · changelog