diff --git a/src/server/responses/responses-field-backfill.ts b/src/server/responses/responses-field-backfill.ts index 48019670f2..ccd79cf471 100644 --- a/src/server/responses/responses-field-backfill.ts +++ b/src/server/responses/responses-field-backfill.ts @@ -91,6 +91,29 @@ function nextSyntheticItemSlot(): ItemIdSlot { return { kind: "fallback", ordinal: syntheticItemOrdinal }; } +/** + * Backfill `status` on a message output item if missing. + * + * The Responses API spec defines `status` as a required field on + * `OutputMessage`. Some upstream relays omit it, which causes strict + * deserializers (e.g. grok-build's serde types) to fail with + * `missing field 'status'`. Only message items carry this field in the + * Responses schema; reasoning, function_call, and other item types do not. + * + * The value is inferred from the event context: `output_item.added` and + * `response.created` / `response.in_progress` mean the message is still + * being generated (`in_progress`); `output_item.done` and + * `response.completed` / `response.incomplete` mean the message is + * finalized (`completed` / `incomplete` respectively). + * + * Returns the same object reference if no change is needed. + */ +function backfillItemStatus(item: Record, inferredStatus: string): Record { + if (item.type !== "message") return item; + if ("status" in item) return item; + return { ...item, status: inferredStatus }; +} + /** * Backfill annotations: [] on an output_text content part if missing. * Returns the same object reference if no change is needed. @@ -122,10 +145,10 @@ function backfillContentArray(content: unknown): unknown { /** * Walk an output item and backfill output_text parts in its content. - * Also backfills a missing required id on the item itself. + * Also backfills a missing required id and status on the item itself. * Returns the same object reference if nothing changed. */ -function backfillOutputItem(item: unknown, slot: ItemIdSlot): unknown { +function backfillOutputItem(item: unknown, slot: ItemIdSlot, inferredStatus: string): unknown { if (!isPlainObject(item)) return item; // The compact wire family is the `/v1/responses/compact` format, not a Responses output item. // Those items have no `id` in that contract, so synthesizing one changes a response body the @@ -136,34 +159,93 @@ function backfillOutputItem(item: unknown, slot: ItemIdSlot): unknown { const content = item.content; const repaired = backfillContentArray(content); const withId = backfillItemId(item, slot); - if (repaired === content && withId === item) return item; - return { ...withId, ...(repaired === content ? {} : { content: repaired }) }; + const withStatus = backfillItemStatus(withId, inferredStatus); + if (repaired === content && withStatus === item) return item; + return { ...withStatus, ...(repaired === content ? {} : { content: repaired }) }; } /** * Walk a response object's output[] and backfill output_text parts. + * Also backfills `created_at` on the response itself when absent. + * + * `createdAt` is captured once per rewrite factory (SSE) or per call (JSON) + * so every event in the same stream carries the same timestamp, even if the + * stream spans a second boundary. + * + * `inferredItemStatus` is the status to backfill on message items that lack + * one — derived from the event type so `output_item.added` / `response.created` + * gets `in_progress` while `output_item.done` / `response.completed` gets + * `completed`. + * * Returns the same object reference if nothing changed. */ -function backfillResponseOutput(response: unknown): unknown { +function backfillResponseOutput(response: unknown, createdAt: number, inferredItemStatus: string): unknown { if (!isPlainObject(response)) return response; - const output = response.output; - if (!Array.isArray(output)) return response; + let current = response; + + // Backfill created_at on the Response object. Strict Responses decoders (e.g. grok-build's + // serde types) require `created_at: u64` — no `#[serde(default)]` — so an upstream relay that + // omits it causes `missing field 'created_at'`. Use a timestamp captured once per rewrite + // factory so every event in the same stream agrees, even if the stream spans a second + // boundary. + if (!("created_at" in current)) { + current = { ...current, created_at: createdAt }; + } + + const output = current.output; + if (!Array.isArray(output)) return current === response ? response : current; let changed = false; const repaired = output.map((item, idx) => { if (!isPlainObject(item)) return item; - const next = backfillOutputItem(item, { kind: "index", index: idx }); + const next = backfillOutputItem(item, { kind: "index", index: idx }, inferredItemStatus); if (next !== item) changed = true; return next; }); - return changed ? { ...response, output: repaired } : response; + if (!changed && current === response) return response; + return { ...current, output: repaired }; +} + +/** + * Infer the status to backfill on a message item from the event type. + * + * `output_item.added` means the item is still being generated (`in_progress`); + * `output_item.done` means it is finalized (`completed`). Response-level events + * infer from the response's own status field — which is authoritative when present. + * If the response status is also absent, the event type itself determines the phase: + * `response.created` / `response.in_progress` → `in_progress`, + * `response.completed` → `completed`, `response.incomplete` → `incomplete`. + */ +function inferredStatusForEventType(eventType: string): string { + if (eventType === "response.output_item.added") return "in_progress"; + if (eventType === "response.output_item.done") return "completed"; + if (eventType === "response.created" || eventType === "response.in_progress") return "in_progress"; + if (eventType === "response.incomplete" || eventType === "response.failed") return "incomplete"; + return "completed"; +} + +/** + * Map a response-level lifecycle status to a valid OutputMessage status. + * + * `OutputMessage.status` accepts only `in_progress`, `completed`, or + * `incomplete`. Response-level statuses like `failed` or `cancelled` have no + * direct message-level equivalent, but `incomplete` is the correct semantic + * mapping: the message did not finish generating. Writing `completed` would + * assert something the upstream never claimed — a client branching on + * `status === "completed"` would treat a truncated message as whole. + */ +function messageStatusFromResponseStatus(status: string): string | null { + if (status === "in_progress" || status === "completed" || status === "incomplete") return status; + if (status === "failed" || status === "cancelled") return "incomplete"; + return null; } /** * Statelessly rewrite one SSE event: backfill annotations * on any output_text content part found in the event payload. */ -function rewriteEvent(event: Record): Record { +function rewriteEvent(event: Record, createdAt: number): Record { const type = typeof event.type === "string" ? event.type : ""; + const inferredItemStatus = inferredStatusForEventType(type); let next = event; let changed = false; @@ -177,8 +259,8 @@ function rewriteEvent(event: Record): Record { // is not recoverable in that case, but a unique id is what strict decoders require, and a // well-formed stream still gets the stable index-derived id. const item = typeof rawIndex === "number" && Number.isInteger(rawIndex) && rawIndex >= 0 - ? backfillOutputItem(event.item, { kind: "index", index: rawIndex }) - : backfillOutputItem(event.item, nextSyntheticItemSlot()); + ? backfillOutputItem(event.item, { kind: "index", index: rawIndex }, inferredItemStatus) + : backfillOutputItem(event.item, nextSyntheticItemSlot(), inferredItemStatus); if (item !== event.item) { next = { ...next, item }; changed = true; @@ -198,7 +280,13 @@ function rewriteEvent(event: Record): Record { // response.created / in_progress / completed / incomplete / failed: // response.output[].content[] -> output_text parts if (isPlainObject(event.response)) { - const response = backfillResponseOutput(event.response); + // For response-level events, prefer the response's own status when it is a valid + // OutputMessage status. Response lifecycle statuses like "failed" or "cancelled" + // have no message-level equivalent — fall back to the event-type inference instead. + const responseStatus = typeof event.response.status === "string" + ? messageStatusFromResponseStatus(event.response.status) ?? inferredItemStatus + : inferredItemStatus; + const response = backfillResponseOutput(event.response, createdAt, responseStatus); if (response !== event.response) { next = { ...next, response }; changed = true; @@ -213,8 +301,13 @@ function rewriteEvent(event: Record): Record { * on output_text content parts. Unconditional: the field is a required * canonical Responses field, so adding it when absent is safe for all * clients. + * + * `created_at` is captured once at factory creation time so every event in + * the same stream carries the same timestamp, even if the stream spans a + * second boundary. */ export function createResponsesFieldBackfillBlockRewrite(): SseBlockRewrite { + const createdAt = Math.floor(Date.now() / 1000); const rewrite: SseBlockRewrite = (block: string): readonly string[] => { const payload = sseDataPayload(block); if (payload === null) return [block]; @@ -225,7 +318,7 @@ export function createResponsesFieldBackfillBlockRewrite(): SseBlockRewrite { return [block]; } if (!isPlainObject(event)) return [block]; - const rewritten = rewriteEvent(event); + const rewritten = rewriteEvent(event, createdAt); if (rewritten === event) return [block]; return [replaceSseDataPayload(block, JSON.stringify(rewritten))]; }; @@ -245,7 +338,13 @@ export function backfillResponsesFieldsJson(payload: string): string { return payload; } if (!isPlainObject(response)) return payload; - const repaired = backfillResponseOutput(response); + const createdAt = Math.floor(Date.now() / 1000); + // For a non-streaming response, derive the item status from the response's own + // status field when it is a valid OutputMessage status; fall back to "completed". + const inferredItemStatus = typeof response.status === "string" + ? messageStatusFromResponseStatus(response.status) ?? "completed" + : "completed"; + const repaired = backfillResponseOutput(response, createdAt, inferredItemStatus); if (repaired === response) return payload; return JSON.stringify(repaired); } diff --git a/tests/responses-field-backfill.test.ts b/tests/responses-field-backfill.test.ts index c5ec73e690..a53d6670f2 100644 --- a/tests/responses-field-backfill.test.ts +++ b/tests/responses-field-backfill.test.ts @@ -350,6 +350,349 @@ describe("responses-field-backfill", () => { expect(result.output[0].id).toBe("msg_real"); }); + test("backfills missing status on output_item.done message", () => { + const event = { + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "hi" }], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.item.status).toBe("completed"); + }); + + test("preserves existing status on message items", () => { + const event = { + type: "response.output_item.done", + output_index: 0, + item: { + type: "message", + id: "msg_1", + role: "assistant", + status: "in_progress", + content: [{ type: "output_text", text: "hi" }], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.item.status).toBe("in_progress"); + }); + + test("backfills missing status on response.completed output items", () => { + const event = { + type: "response.completed", + sequence_number: 42, + response: { + id: "resp_1", + object: "response", + status: "completed", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.output[0].status).toBe("completed"); + }); + + test("does not add status to non-message items", () => { + const event = { + type: "response.output_item.done", + output_index: 0, + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "do_thing", + arguments: "{}", + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.item).not.toHaveProperty("status"); + }); + + test("backfillResponsesFieldsJson backfills missing status on message items", () => { + const response = { + id: "resp_1", + object: "response", + status: "completed", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "hello" }], + }, + ], + }; + const result = JSON.parse(backfillResponsesFieldsJson(JSON.stringify(response))) as typeof response; + expect(result.output[0].status).toBe("completed"); + }); + + test("backfills missing created_at on response.completed", () => { + const event = { + type: "response.completed", + sequence_number: 42, + response: { + id: "resp_1", + object: "response", + status: "completed", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "answer" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.created_at).toEqual(expect.any(Number)); + expect(parsed.response.created_at).toBeGreaterThan(0); + }); + + test("preserves existing created_at on response objects", () => { + const event = { + type: "response.created", + sequence_number: 1, + response: { + id: "resp_1", + object: "response", + created_at: 1700000000, + status: "in_progress", + model: "grok-4.5", + output: [], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.created_at).toBe(1700000000); + }); + + test("backfillResponsesFieldsJson backfills missing created_at", () => { + const response = { + id: "resp_1", + object: "response", + status: "completed", + output: [], + }; + const result = JSON.parse(backfillResponsesFieldsJson(JSON.stringify(response))) as typeof response; + expect(result.created_at).toEqual(expect.any(Number)); + expect(result.created_at).toBeGreaterThan(0); + }); + + test("backfillResponsesFieldsJson derives incomplete status on message items", () => { + const response = { + id: "resp_1", + object: "response", + status: "incomplete", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "partial" }], + }, + ], + }; + const result = JSON.parse(backfillResponsesFieldsJson(JSON.stringify(response))) as typeof response; + expect(result.output[0].status).toBe("incomplete"); + }); + + test("backfills in_progress status on output_item.added message", () => { + const event = { + type: "response.output_item.added", + output_index: 0, + item: { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "" }], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + // output_item.added means the item is still being generated; marking it "completed" + // would misrepresent the stream state to strict clients. + expect(parsed.item.status).toBe("in_progress"); + }); + + test("backfills in_progress status on response.created output items", () => { + const event = { + type: "response.created", + sequence_number: 1, + response: { + id: "resp_1", + object: "response", + status: "in_progress", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.output[0].status).toBe("in_progress"); + }); + + test("backfills in_progress when both response status and item status are absent", () => { + // response.created with no status on either the response or the message item: + // the event type alone must drive the inference, not a "completed" default. + const event = { + type: "response.created", + sequence_number: 1, + response: { + id: "resp_1", + object: "response", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.output[0].status).toBe("in_progress"); + }); + + test("backfills incomplete status on response.incomplete output items", () => { + const event = { + type: "response.incomplete", + sequence_number: 10, + response: { + id: "resp_1", + object: "response", + status: "incomplete", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "partial" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.output[0].status).toBe("incomplete"); + }); + + test("maps response.failed to incomplete message status", () => { + // response.failed carries status: "failed" on the response object, but + // OutputMessage.status only accepts in_progress/completed/incomplete. + // "failed" means the message did not finish generating, so "incomplete" + // is the correct semantic mapping — not "completed", which would falsely + // claim the message is whole. + const event = { + type: "response.failed", + sequence_number: 5, + response: { + id: "resp_1", + object: "response", + status: "failed", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "partial" }], + }, + ], + }, + }; + const [out] = apply(sseBlock(event)); + const parsed = parseData([out])[0]; + expect(parsed.response.output[0].status).toBe("incomplete"); + }); + + test("created_at stays consistent across events in the same stream", () => { + // The rewrite factory captures created_at once at creation time; every event in + // the same stream must carry the same timestamp, even if the stream spans a + // second boundary. This test mocks Date.now to make that guarantee deterministic: + // the factory is created at t=1000, and the second event is applied at t=3000. + // A per-event implementation would write 3000 for the second event and fail. + const originalNow = Date.now; + const fakeNow = (ms: number) => () => ms; + try { + Date.now = fakeNow(1_000); + const localRewrite = createResponsesFieldBackfillBlockRewrite(); + const localApply = (block: string): string[] => [...localRewrite(block)]; + + const createdEvent = { + type: "response.created", + sequence_number: 1, + response: { + id: "resp_1", + object: "response", + status: "in_progress", + model: "grok-4.5", + output: [], + }, + }; + + // Advance the clock past a second boundary before applying the second event. + Date.now = fakeNow(3_000); + const completedEvent = { + type: "response.completed", + sequence_number: 42, + response: { + id: "resp_1", + object: "response", + status: "completed", + model: "grok-4.5", + output: [ + { + type: "message", + id: "msg_1", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "answer" }], + }, + ], + }, + }; + + const createdOut = parseData(localApply(sseBlock(createdEvent)))[0]; + const completedOut = parseData(localApply(sseBlock(completedEvent)))[0]; + // Both events went through the same rewrite factory, so their created_at must + // be the factory's capture time (1), not the per-event time (3). + expect(createdOut.response.created_at).toBe(1); + expect(completedOut.response.created_at).toBe(1); + } finally { + Date.now = originalNow; + } + }); + test("the canonical image_generation_call type gets its own prefix", () => { const response = { id: "resp_1", diff --git a/tests/server-combo-failover-e2e.test.ts b/tests/server-combo-failover-e2e.test.ts index ea34844044..ba83bf0ec0 100644 --- a/tests/server-combo-failover-e2e.test.ts +++ b/tests/server-combo-failover-e2e.test.ts @@ -1306,7 +1306,7 @@ describe("server combo failover 030 activation matrix", () => { expect(hits).toEqual(["azure", "chat"]); }); - test("cross-adapter chat 503 to Responses 200 returns the exact backup response", async () => { + test("cross-adapter chat 503 to Responses 200 returns the backup response exactly apart from the backfill", async () => { const a = serve(() => Response.json({ error: { message: "down" } }, { status: 503 })); const exact = responsesSuccess("raw backup", "m2"); let bBody: Record | undefined; @@ -1320,7 +1320,14 @@ describe("server combo failover 030 activation matrix", () => { }); const response = await post(config); expect(response.status).toBe(200); - expect(await response.json()).toEqual(exact); + const received = await response.json() as Record; + // The passthrough backfill adds response-level created_at (a fresh timestamp, + // not the fixture's value) when the upstream omits it. Assert its valid + // integer shape, then drop it so the rest must match the fixture exactly. + expect(received.created_at).toEqual(expect.any(Number)); + expect(received.created_at).toBeGreaterThan(0); + delete received.created_at; + expect(received).toEqual(exact); expect(bBody?.model).toBe("m2"); });