Dashboard › opencode-lore › Session 0fRIUUrKLk7L
0fRIUUrKLk7LS6HEKREAD-ONLY compare two sibling accumulation corrections in /home/byk/Code/opencode-lore-responses-accumulation. Common parent tklkqxyt/ce88c614. Candidate A lrqlyykx/ca83ea5361d7 fix(responses): close lifecycle gaps, validated diff hash child 7c56d9f251cc66f2e3d46944fb9c82f95f4949d9ef5a19a255c174e0608d7b92. Candidate B rwunmtrq/3cbc99c3cf38 fix(responses): complete accumulation boundaries, currently bookmark fix/responses-accumulation-order. Do not edit/format/install/VCS mutation/server/network/DB/test. Read exact immutable files/diffs using jj file show/diff, verify identities/conflicts. Determine precise overlap and behavioral differences: omitted normalized continuation coordinate bounds, forged lifecycle event allowlist, sparse missing-done message/reasoning materialization, mixed incomplete parts, item_reference accumulation/limits, any additional review findings. Map tests and source guards in each. Classify B as superseding A, A superseding B, complementary, or conflicting. Identify whether a clean merge/rebase is sound, exact conflicts/duplicated tests, and recommended canonical lineage/bookmark with evidence. Report concise file:line/symbol evidence from each revision and never mutate. If lore warning use lore recall/local UI and continue; never warning-only/empty.
<skill_content name="jj-guide"> # Skill: jj-guide # jj Guide for AI Agents Jujutsu (jj) is a Git-compatible VCS with mutable commits, automatic snapshotting, no staging area, and first-class conflicts. This skill teaches you how to use it safely from a non-interactive agent environment. If `.jj/` exists in the repo root, this is a jj repo. **Use `jj` commands, not `git`.** In a colocated repo (`.jj/` *and* `.git/`), git tools can read the state, but mutations should go through `jj` so the operation log stays consistent. ## Critical Rules - **NEVER** use interactive flags (`-i`, `--interactive`). TUI prompts hang in agent environments. This applies to `jj split -i`, `jj squash -i`, `jj commit -i`, `jj resolve`, `jj diffedit`, etc. - **ALWAYS** pass `-m "msg"` when describing/committing. Without `-m`, an editor opens and hangs. - **VERIFY** mutations with `jj st` and `jj log` after `squash`, `abandon`, `rebase`, `restore`, `commit`. ...
lrqlyykxvwoq ca83ea5361d7dc0b059fb9a8e855e892af1bd343 parents: tklkqxytmpkz/ce88c614c0ae description: fix(responses): close lifecycle gaps conflict: no bookmarks: rwunmtrqvptr 3cbc99c3cf3844c67f5132411a5dcee3eda1723c parents: tklkqxytmpkz/ce88c614c0ae description: fix(responses): complete accumulation boundaries conflict: no bookmarks: fix/responses-accumulation-order tklkqxytmpkz ce88c614c0ae55d229cb605aa7c7369d918d3650 parents: kvpmnlwnznqx/683dff02101f description: fix(responses): complete lifecycle reconciliation conflict: no bookmarks:
M packages/gateway/src/pipeline.ts M packages/gateway/test/openai-responses-recall-aware-stream.test.ts packages/gateway/src/pipeline.ts | 101 ++++-- ...est/openai-responses-recall-aware-stream.test.ts | 364 ++++++++++++++++++++++ 2 files changed, 447 insertions(+), 18 deletions(-)
M packages/gateway/src/pipeline.ts M packages/gateway/test/openai-responses-recall-aware-stream.test.ts packages/gateway/src/pipeline.ts | 155 +++--- ...est/openai-responses-recall-aware-stream.test.ts | 513 ++++++++++++++++++---- 2 files changed, 486 insertions(+), 182 deletions(-)
A from-to: 8ffab7d6744e61a8970a5dc0865a27ad3302c905571205b6e0a24ee86cdc209b - A revision: 8ffab7d6744e61a8970a5dc0865a27ad3302c905571205b6e0a24ee86cdc209b - B from-to: ed233280273da65a84bc3138a30d6ff8b3a7daf2e8763db2517dc2f15fa104ee - B revision: ed233280273da65a84bc3138a30d6ff8b3a7daf2e8763db2517dc2f15fa104ee -
A plain from-to: 7c56d9f251cc66f2e3d46944fb9c82f95f4949d9ef5a19a255c174e0608d7b92 - A show patch: 7c56d9f251cc66f2e3d46944fb9c82f95f4949d9ef5a19a255c174e0608d7b92 - B plain from-to: e62ce85478a3c6bb026b669679c86db87df2f32e7680e1959e9980b1dfec56a6 -
<!-- This section is maintained by the coding agent via lore (https://github.com/BYK/loreai) --> ## Long-term Knowledge For long-term knowledge entries managed by [lore](https://github.com/BYK/loreai) (gotchas, patterns, decisions, architecture), see [`.lore.md`](.lore.md) in the project root. <!-- End lore-managed section --> ## Review See [`quality/REVIEW.md`](quality/REVIEW.md) for the review heuristics and workflow — regression-test discipline, adversarial-order state setup, fan-out registry coverage, recurring bug-class batteries, and the two-reviewer rule. Every non-trivial PR requires an adversarial correctness review before merge. # Project Overview Lore is a **three-tier memory architecture** for AI coding agents. It intercepts LLM API calls (as a transparent proxy or native plugin), distills conversation history into compressed summaries, and extracts long-term knowledge entries that persist across sessions. **Runtime:** Node.js >= 22.5 (development/tests/production). ...
Modified regular file packages/gateway/src/pipeline.ts:
...
7968 7968: }
7969 7969: return value;
7970 7970: };
7971 : const finalizedMessageContent = (
7972 : lifecycle: OutputLifecycle,
7971: const materializeCompletedTextParts = (
7972: parts: ReadonlyMap<number, TextPartLifecycle>,
7973: description: string,
7974: outputIndex: number,
7973 7975: ): Array<Record<string, unknown>> =>
7974 : Array.from(lifecycle.content).reduce<Array<Record<string, unknown>>>(
7975 : (content, [contentIndex, part]) => {
7976 : if (!part.valueDone && !part.partDone) return content;
7977 : content[contentIndex] =
7976: Array.from(parts).reduce<Array<Record<string, unknown>>>(
7977: (materialized, [partIndex, part]) => {
7978: if (
7979: (!part.valueDone && !part.partDone) ||
7980: (part.deltaSeen && !part.valueDone) ||
7981: ...
Modified regular file packages/gateway/src/pipeline.ts:
...
7933 7933: acc: ResponsesAccState,
7934 7934: event: string,
7935 7935: parsed: Record<string, unknown>,
7936: outputOffset = 0,
7936 7937: ): ResponsesAccState | undefined => {
7937 7938: if (opts.validation !== "codex") return undefined;
7938 7939: let normalizationState = codexNormalizationStates.get(acc);
7939 7940: if (!normalizationState) {
7940 7941: normalizationState = makeResponsesAccState();
7941 7942: codexNormalizationStates.set(acc, normalizationState);
7942 7943: }
7944: // Tighten Codex's local exclusive ceiling by the continuation shift. This
7945: // rejects an omitted index while the normalizer derives it, before either
7946: // normalization or accumulation can retain an out-of-range item.
7947: const localSparseCeiling = Math.max(
7948: 0,
7949: Math.min(
7950: opts.maxSSEFrames ?? ...
Modified regular file packages/gateway/test/openai-responses-recall-aware-stream.test.ts:
...
3361 3361: },
3362 3362: );
3363 3363:
3364: test.each(
3365: (["principal", "continuation"] as const).flatMap((stream) =>
3366: (["absent", "empty"] as const).map((terminalOutput) => ({
3367: stream,
3368: terminalOutput,
3369: })),
3370: ),
3371: )(
3372: "materializes completed $stream Codex reasoning without item done when terminal output is $terminalOutput",
3373: async ({ stream, terminalOutput }) => {
3374: const responseId = `resp_reasoning_without_done_${stream}_${terminalOutput}`;
3375: const itemId = `rs_reasoning_without_done_${stream}_${terminalOutput}`;
3376: const privateSummary = `private ${stream} completed summary`;
3377: const privateContent = `private ${stream} completed content`;
3378: const response = {
...
Modified regular file packages/gateway/test/openai-responses-recall-aware-stream.test.ts:
...
5092 5092: },
5093 5093: );
5094 5094:
5095: test.each([
5096: {
5097: name: "terminal-only output",
5098: streamed: "",
5099: terminalItem: {
5100: type: "message",
5101: id: "msg_usage_terminal_only",
5102: role: "assistant",
5103: status: "completed",
5104: content: [
5105: { type: "output_text", text: "private terminal-only output" },
5106: ],
5107: },
5108: },
5109: {
5110: name: "reasoning gap",
5111: streamed:
5112: sseEvent("response.output_item.added", {
5113: output_index: 0,
5114: item: { type: "reasoning", id: "rs_usage_gap" },
5115: }) +
5116: sseEvent("response.reasoning_summary_text.done", {
5117: ...
Modified regular file packages/gateway/src/pipeline.ts:
...
7933 7933: acc: ResponsesAccState,
7934 7934: event: string,
7935 7935: parsed: Record<string, unknown>,
7936: outputOffset = 0,
7936 7937: ): ResponsesAccState | undefined => {
7937 7938: if (opts.validation !== "codex") return undefined;
7938 7939: let normalizationState = codexNormalizationStates.get(acc);
7939 7940: if (!normalizationState) {
7940 7941: normalizationState = makeResponsesAccState();
7941 7942: codexNormalizationStates.set(acc, normalizationState);
7942 7943: }
7944: // Tighten Codex's local exclusive ceiling by the continuation shift. This
7945: // rejects an omitted index while the normalizer derives it, before either
7946: // normalization or accumulation can retain an out-of-range item.
7947: const localSparseCeiling = Math.max(
7948: 0,
7949: Math.min(
7950: opts.maxSSEFrames ?? ...
7879: const referenceIdentities = new Set<string>();
7932: const normalizeCodexEvent = (
7971: const materializeCompletedTextParts = (
7995: const materializeCompletedCodexItems = (acc: ResponsesAccState): void => {
8001: const content = materializeCompletedTextParts(
8012: ? materializeCompletedTextParts(
8020: ? materializeCompletedTextParts(
8424: const reasoningSummaryLifecycleEvents = new Set([
8435: const assertSparseCoordinates = (
8442: !reasoningSummaryLifecycleEvents.has(event)) ||
8610: referenceIdentities.has(identity) ||
8686: const content = materializeCompletedTextParts(
8737: referenceIdentities.has(finalCallId) ||
9522: materializeCompletedCodexItems(acc);
9550: const isReference = actual.type === "item_reference";
9643: item?.type === "item_reference"
9646: throw new Error("invalid Responses output_index for item_reference");
9664: referenceIdentities.has(item.id)
9669: ...
7879: const referenceIdentities = new Set<string>();
7932: const normalizeCodexEvent = (
7982: const finalizedMessageContent = (
7996: const materializeCompletedCodexMessages = (acc: ResponsesAccState): void => {
8002: const content = finalizedMessageContent(lifecycle);
8397: const assertSparseCoordinates = (
8564: referenceIdentities.has(identity) ||
8622: const content = finalizedMessageContent(lifecycle);
8669: referenceIdentities.has(finalCallId) ||
9446: materializeCompletedCodexMessages(acc);
9474: const isReference = actual.type === "item_reference";
9567: item?.type === "item_reference"
9570: throw new Error("invalid Responses output_index for item_reference");
9585: referenceIdentities.has(item.id)
9590: referenceIdentities.add(item.id);
9601: item.type !== "item_reference" ||
9606: `invalid Responses item_reference lifecycle for index ${outputIndex}`,
9618: `Responses stream ended ...
907: test("bounds retained output before recall detection", async () => {
918: created("resp_retained", "gpt-5.6-terra"),
919: textItem(0, "large retained output"),
920: completed("resp_retained"),
941: "openai-responses recall-aware stream failed category=principal_resource_limit",
945: test("does not retry a pre-output principal resource limit", async () => {
948: streamFrom([created("resp_pre_output_limit", "gpt-5.6-terra")]),
969: test("accumulates the principal event that crosses the retained-state limit", async () => {
979: const triggerCiphertext = "private principal retained-state trigger";
984: id: "rs_retained_trigger",
991: item_id: "rs_retained_trigger",
997: item_id: "rs_retained_trigger",
1005: id: "rs_retained_trigger",
1016: created("resp_retained_trigger", "gpt-5.6-terra"),
1021: completed("resp_retained_trigger"),
1042: id: "rs_retained_trigger",
1066: "openai-responses ...
907: test("bounds retained output before recall detection", async () => {
918: created("resp_retained", "gpt-5.6-terra"),
919: textItem(0, "large retained output"),
920: completed("resp_retained"),
941: "openai-responses recall-aware stream failed category=principal_resource_limit",
945: test("does not retry a pre-output principal resource limit", async () => {
948: streamFrom([created("resp_pre_output_limit", "gpt-5.6-terra")]),
969: test("accumulates the principal event that crosses the retained-state limit", async () => {
979: const triggerCiphertext = "private principal retained-state trigger";
984: id: "rs_retained_trigger",
991: item_id: "rs_retained_trigger",
997: item_id: "rs_retained_trigger",
1005: id: "rs_retained_trigger",
1016: created("resp_retained_trigger", "gpt-5.6-terra"),
1021: completed("resp_retained_trigger"),
1042: id: "rs_retained_trigger",
1066: "openai-responses ...
10: * response.output_item.done → response.function_call_arguments.delta →
11: * response.function_call_arguments.done →
75: /** Accumulating output items indexed by output_index. */
616: const outputIndex = parsed.output_index as number;
645: const outputIndex = parsed.output_index as number;
660: const outputIndex = parsed.output_index as number;
672: const outputIndex = parsed.output_index as number;
685: const outputIndex = parsed.output_index as number;
695: const outputIndex = parsed.output_index as number;
704: case "response.function_call_arguments.delta": {
705: const outputIndex = parsed.output_index as number;
716: case "response.function_call_arguments.done": {
717: const outputIndex = parsed.output_index as number;
912: maxSparseIndex: number,
914: const outputIndex = parsed.output_index;
918: (outputIndex as number) < maxSparseIndex;
978: event === "response.function_call_arguments.delta" ||
979: event === ...
/usr/bin/bash: line 1: python: command not found
1200: (typeof part.refusal === "string" && part.refusal.length > 0)),
1201: );
1202: }
1203: }
1204: if (
1205: event === "response.output_text.delta" ||
1206: event === "response.output_text.done" ||
1207: event === "response.refusal.delta" ||
1208: event === "response.refusal.done" ||
1209: event === "response.function_call_arguments.delta" ||
1210: event === "response.function_call_arguments.done"
1211: ) {
1212: return [parsed.delta, parsed.text, parsed.refusal, parsed.arguments].some(
1213: (value) => typeof value === "string" && value.length > 0,
1214: );
1215: }
1216: if (event.startsWith("response.reasoning")) {
1217: const part = isRecord(parsed.part) ? parsed.part : undefined;
1218: return [parsed.delta, parsed.text, parsed.summary_text, part?.text].some(
1219: (value) => typeof value === "string" && value.length > 0,
1220: );
1221: }
1222: return false;
1223: }
1224:
1225: function ...
9528: }
9529: if (!Array.isArray(response.output)) {
9530: throw new Error("Responses terminal output must be an array");
9531: }
9532: // ChatGPT/Codex can omit some or all streamed items from the terminal
9533: // snapshot. Treat the output_item lifecycle as authoritative while still
9534: // requiring every repeated terminal item to match in stream order.
9535: const actualOutput = response.output.map((item) => {
9536: if (!item || typeof item !== "object" || Array.isArray(item)) {
9537: throw new Error("Responses terminal output contains malformed item");
9538: }
9539: return item as Record<string, unknown>;
9540: });
9541: const expected = [...acc.rawItems.entries()].sort(([a], [b]) => a - b);
9542: if (
9543: opts.validation === "public" &&
9544: actualOutput.length !== expected.length
9545: ) {
9546: throw new Error("Responses terminal output changed streamed item");
9547: }
9548: ...
2475: model: "gpt-5.6-terra",
2476: status: "completed",
2477: output: [terminalItem],
2478: },
2479: }),
2480: ]),
2481: {
2482: onComplete: (response) => {
2483: completedResponse = response;
2484: },
2485: onRecall: async () => ({ anchorText: "", resultText: "" }),
2486: runFollowUp: async () => {
2487: throw new Error("should not run");
2488: },
2489: },
2490: );
2491:
2492: expect(await drain(client)).toContain("response.failed");
2493: expect(completedResponse?.content).toEqual([]);
2494: expect(completedResponse?.rawOutputItems).toEqual([]);
2495: });
2496:
2497: test("never forwards response-side item_reference lifecycle events", async () => {
2498: let completedResponse: GatewayResponse | undefined;
2499: const client = streamResponsesRecallAware(
2500: streamFrom([
2501: created("resp_reference", ...
8124: let retainedStateBytes = 0;
9612: const assertReferenceLifecyclesComplete = (
10258: const referenceIndices = new Map<number, ReferenceLifecycle>();
10260: const retainedStateBaseline = retainedStateBytes;
10366: if (consumeReferenceEvent(state, referenceIndices, event, parsed)) {
10443: retainedStateBytes += encoder.encode(data).byteLength;
10444: if (retainedStateBytes > maxRetainedStateBytes) {
10600: assertReferenceLifecyclesComplete(referenceIndices);
10796: retainedStateBytes,
10807: const contReferenceIndices = new Map<
10996: contReferenceIndices,
11073: retainedStateBytes += encoder.encode(cd).byteLength;
11074: if (retainedStateBytes > maxRetainedStateBytes) {
11226: assertReferenceLifecyclesComplete(
11227: ...
10325: if (event.startsWith("response.")) {
10326: throw new Error(`malformed JSON in Responses event ${event}`);
10327: }
10328: // Non-JSON keepalive/comment event — forward as-is.
10329: if (event !== "message") {
10330: const chunk = encoder.encode(formatResponsesEvent(event, data));
10331: if (recallIndices.size > 0 || unresolvedToolIndices.size > 0) {
10332: deferredBytes += chunk.byteLength;
10333: if (deferredBytes > maxDeferredBytes) {
10334: throw new SSEStreamLimitError(
10335: "recall stream exceeded deferred event limit",
10336: );
10337: }
10338: deferredEvents.push({ chunk });
10339: } else {
10340: await enqueuePrincipal(chunk, otherToolSeen);
10341: }
10342: }
10343: ...
560: stopReason: "end_turn",
561: usage: { inputTokens: 0, outputTokens: 0 },
562: items: new Map(),
563: rawItems: new Map(),
564: itemIndexById: new Map(),
565: callIndexById: new Map(),
566: effectiveToolIndexById: new Map(),
567: nextOutputIndex: 0,
568: activeTextItems: new Set(),
569: activeToolItems: new Set(),
570: unboundTextItems: new Set(),
571: unboundToolItems: new Set(),
572: textDoneItems: new Set(),
573: refusalDoneItems: new Set(),
574: argumentDoneItems: new Set(),
575: };
576: }
577:
578: /**
579: * Apply one parsed Responses SSE event to the accumulation state. Never touches
580: * I/O — safe to call while forwarding the same event verbatim to the client.
581: */
582: export function applyResponsesEvent(
583: state: ResponsesAccState,
584: event: string,
585: parsed: Record<string, unknown>,
586: ): void {
587: switch (event) {
588: case "codex.rate_limits": {
589: // Only the ...
815:export function finalizeResponsesAcc( 892: rawOutputItems: Array.from(state.rawItems.entries())
805: }
806: break;
807: }
808:
809: // Other events (response.content_part.*,
810: // response.reasoning_summary_*, etc.) — ignored for accumulation
811: }
812: }
813:
814: /** Build the final GatewayResponse from accumulated state. */
815: export function finalizeResponsesAcc(
816: state: ResponsesAccState,
817: ): GatewayResponse {
818: const content: GatewayContentBlock[] = [];
819: const sortedIndices = Array.from(state.items.keys()).sort((a, b) => a - b);
820:
821: for (const index of sortedIndices) {
822: const item = state.items.get(index);
823: if (!item) continue;
824: if (item.type === "text") {
825: if (item.content) {
826: for (const part of item.content) {
827: if (!part) continue;
828: if (part.type === "output_text" && typeof part.text === "string") {
829: content.push({ type: "text", text: part.text });
830: } else {
831: content.push({
832: ...
8480: });
8481: if (seedItem.type === "function_call") {
8482: const seededLifecycle = lifecyclesFor(state).get(outputIndex);
8483: if (!seededLifecycle) {
8484: throw new Error(
8485: `missing Responses lifecycle for index ${outputIndex}`,
8486: );
8487: }
8488: seededLifecycle.argumentsDone = true;
8489: }
8490: }
8491: const lifecycles = lifecyclesFor(state);
8492: const lifecycle = lifecycles.get(outputIndex);
8493: if (lifecycle?.outputDone && event !== "response.output_item.added") {
8494: throw new Error(
8495: `Responses event after output_item.done for index ${outputIndex}`,
8496: );
8497: }
8498: if (event === "response.output_item.added") {
8499: if (state.rawItems.has(outputIndex)) {
8500: throw new Error(`duplicate Responses output_index ${outputIndex}`);
8501: }
8502: const item = parsed.item as Record<string, unknown> | ...
8700: event.startsWith("response.refusal")
8701: ) {
8702: const contentIndex = parsed.content_index as number;
8703: const expectedKind = event.startsWith("response.output_text")
8704: ? "output_text"
8705: : event.startsWith("response.refusal")
8706: ? "refusal"
8707: : event.startsWith("response.reasoning_text")
8708: ? "reasoning_text"
8709: : undefined;
8710: const part = parsed.part as Record<string, unknown> | undefined;
8711: const partKind =
8712: event.startsWith("response.content_part") &&
8713: typeof part?.type === "string"
8714: ? part.type
8715: : undefined;
8716: const kind = expectedKind ?? partKind;
8717: if (
8718: !kind ||
8719: !["output_text", "refusal", "reasoning_text"].includes(kind)
8720: ) {
8721: throw new Error(`invalid Responses content type ...
9020: item.status !== "completed"
9021: ) {
9022: throw new Error("recall function call did not complete");
9023: }
9024: outputIdentities.add(finalFunctionIdentity.callId);
9025: normalized.callId = finalFunctionIdentity.callId;
9026: normalized.name = finalFunctionIdentity.name;
9027: normalized.args = item.arguments;
9028: }
9029: if (declaredType === "message") {
9030: const finalContent = item?.content;
9031: if (!Array.isArray(finalContent)) {
9032: throw new Error(
9033: `Responses message completed without content for index ${outputIndex}`,
9034: );
9035: }
9036: for (const [contentIndex, contentState] of lifecycle.content) {
9037: const finalPart = finalContent[contentIndex] as
9038: | Record<string, unknown>
9039: | undefined;
9040: if (!finalPart ...
10035: object: "response",
10036: created_at:
10037: terminalResponse?.created_at ?? Math.floor(Date.now() / 1000),
10038: model: res.model || state.model,
10039: status: finalStatus,
10040: output: finalOutput,
10041: usage: usageData,
10042: },
10043: }),
10044: );
10045: }
10046:
10047: function buildOutputItems(
10048: hiddenIndices: ReadonlySet<number> = new Set(),
10049: ): Array<Record<string, unknown>> {
10050: const finalOutput: Array<Record<string, unknown>> = [];
10051: const sortedIndices = [
10052: ...new Set([...state.rawItems.keys(), ...state.items.keys()]),
10053: ].sort((a, b) => a - b);
10054: for (const index of sortedIndices) {
10055: if (hiddenIndices.has(index)) continue;
10056: const item = state.items.get(index);
10057: if (!item) {
10058: const rawItem = state.rawItems.get(index);
10059: if (rawItem && ...
pipeline.ts merge-file status: 0 test merge-file status: 0
7925: }
7926: return lifecycles;
7927: };
7928: const codexNormalizationStates = new WeakMap<
7929: ResponsesAccState,
7930: ResponsesAccState
7931: >();
7932: const normalizeCodexEvent = (
7933: acc: ResponsesAccState,
7934: event: string,
7935: parsed: Record<string, unknown>,
7936: outputOffset = 0,
7937: ): ResponsesAccState | undefined => {
7938: if (opts.validation !== "codex") return undefined;
7939: let normalizationState = codexNormalizationStates.get(acc);
7940: if (!normalizationState) {
7941: normalizationState = makeResponsesAccState();
7942: codexNormalizationStates.set(acc, normalizationState);
7943: }
7944: // Tighten Codex's local exclusive ceiling by the continuation shift. This
7945: // rejects an omitted index while the normalizer derives it, before either
7946: // normalization or accumulation can retain an out-of-range item.
7947: const localSparseCeiling = Math.max(
7948: ...
7965: const value = kind === "refusal" ? part.refusal : part.text;
7966: if (typeof value !== "string") {
7967: throw new Error(`invalid Responses ${description} value`);
7968: }
7969: return value;
7970: };
7971: const materializeCompletedTextParts = (
7972: parts: ReadonlyMap<number, TextPartLifecycle>,
7973: description: string,
7974: outputIndex: number,
7975: ): Array<Record<string, unknown>> =>
7976: Array.from(parts).reduce<Array<Record<string, unknown>>>(
7977: (materialized, [partIndex, part]) => {
7978: if (
7979: (!part.valueDone && !part.partDone) ||
7980: (part.deltaSeen && !part.valueDone) ||
7981: (part.partAdded && !part.partDone)
7982: ) {
7983: throw new Error(
7984: `Responses ${description} ended before completion for index ${outputIndex}:${partIndex}`,
7985: );
7986: }
7987: materialized[partIndex] =
7988: ...
7978: throw new Error(`invalid Responses ${description} value`);
7979: }
7980: return value;
7981: };
7982: const finalizedMessageContent = (
7983: lifecycle: OutputLifecycle,
7984: ): Array<Record<string, unknown>> =>
7985: Array.from(lifecycle.content).reduce<Array<Record<string, unknown>>>(
7986: (content, [contentIndex, part]) => {
7987: if (!part.valueDone && !part.partDone) return content;
7988: content[contentIndex] =
7989: part.kind === "refusal"
7990: ? { type: "refusal", refusal: part.authoritativeValue }
7991: : { type: "output_text", text: part.authoritativeValue };
7992: return content;
7993: },
7994: [],
7995: );
7996: const materializeCompletedCodexMessages = (acc: ResponsesAccState): void => {
7997: if (opts.validation !== "codex") return;
7998: for (const [outputIndex, raw] of acc.rawItems) {
7999: if (raw.type !== "message") continue;
8000: ...
8390: const abortController = new AbortController();
8391: const signal = opts.signal
8392: ? AbortSignal.any([opts.signal, abortController.signal])
8393: : abortController.signal;
8394: let activeReader: ReadableStreamDefaultReader<Uint8Array> | null = null;
8395: let currentPrincipalResponse = upstreamResponse;
8396:
8397: const assertSparseCoordinates = (
8398: event: string,
8399: parsed: Record<string, unknown>,
8400: outputOffset = 0,
8401: ): void => {
8402: if (Object.hasOwn(parsed, "output_index")) {
8403: const outputIndex = parsed.output_index;
8404: if (
8405: !Number.isSafeInteger(outputIndex) ||
8406: (outputIndex as number) < 0 ||
8407: (outputIndex as number) + outputOffset >= maxSparseIndex
8408: ) {
8409: throw new Error(`invalid Responses output_index for ${event}`);
8410: }
8411: }
8412: if (Object.hasOwn(parsed, "content_index")) {
8413: const contentIndex = ...
9228: const validateResponseLifecycle = (
9180: (contentState.finalValue !== undefined &&
9181: contentState.finalValue !== finalValue) ||
9182: (contentState.partFinalValue !== undefined &&
9183: contentState.partFinalValue !== finalValue)
9184: ) {
9185: throw new Error(
9186: `Responses output_item.done changed reasoning content for index ${outputIndex}:${contentIndex}`,
9187: );
9188: }
9189: }
9190: }
9191: }
9192: lifecycle.outputDone = true;
9193: }
9194: }
9195: return outputIndex;
9196: };
9197: const seedImplicitCodexItem = (
9198: acc: ResponsesAccState,
9199: normalizationState: ResponsesAccState | undefined,
9200: event: string,
9201: parsed: Record<string, unknown>,
9202: ): void => {
9203: if (
9204: !normalizationState ||
9205: event === "response.output_item.added" ||
9206: ...
8127: const maxSSEFrames = opts.maxSSEFrames ?? DEFAULT_MAX_SSE_FRAMES;
8105: } catch (err) {
8106: log.error("recall transaction rollback failed:", err);
8107: }
8108: }
8109: };
8110: const encoder = new TextEncoder();
8111: const sessionID = opts.sessionID;
8112: const recallBudget = new RecallChainBudget({
8113: maxExecutions:
8114: opts.maxRecallExecutions ?? opts.maxRecallDepth ?? MAX_RECALL_EXECUTIONS,
8115: deadlineAt: opts.recallDeadlineAt,
8116: });
8117: const maxDeferredBytes = opts.maxDeferredBytes ?? 1024 * 1024;
8118: const maxHiddenRecallBytes = opts.maxHiddenRecallBytes ?? maxDeferredBytes;
8119: const maxRetainedStateBytes = opts.maxRetainedStateBytes ?? 16 * 1024 * 1024;
8120: // Validated continuation output is retained transactionally until its chain
8121: // completes, so bound its shared spool with the retained-state budget.
8122: const maxTransactionalBytes = maxRetainedStateBytes;
8123: const maxStreamBytes = opts.maxStreamBytes ?? ...
8081: const seedTextParts = (
8075: throw new Error(
8076: `Responses ${description} changed content for index ${outputIndex}:${partIndex}`,
8077: );
8078: }
8079: }
8080: };
8081: const seedTextParts = (
8082: parts: unknown,
8083: target: Map<number, TextPartLifecycle>,
8084: allowedKinds: ReadonlySet<string>,
8085: description: string,
8086: ): void => {
8087: if (parts === undefined) return;
8088: if (!Array.isArray(parts)) {
8089: throw new Error(`Responses ${description} must be an array`);
8090: }
8091: for (const [index, rawPart] of parts.entries()) {
8092: if (!rawPart || typeof rawPart !== "object" || Array.isArray(rawPart)) {
8093: throw new Error(`invalid Responses ${description} item`);
8094: }
8095: const part = rawPart as Record<string, unknown>;
8096: if (typeof part.type !== "string" || !allowedKinds.has(part.type)) {
8097: throw new Error(`invalid Responses ${description} ...
634: test("suppresses a recall function_call and emits a marker (mixed tools)", async () => {
638: created("resp_mixed", "gpt-5.6-terra"),
667: completed("resp_mixed"),
678: throw new Error("runFollowUp should not run for mixed tools");
699: "msg_resp_mixed_0",
790: throw new Error("should not run for mixed tools");
805: created("resp_parallel_mixed", "gpt-5.6-terra"),
832: completed("resp_parallel_mixed"),
2541: test("rejects an item_reference missing output_item.done", async () => {
3372: "materializes completed $stream Codex reasoning without item done when terminal output is $terminalOutput",
3465: "rejects mixed complete and incomplete %s Codex message parts without item done",
3467: const responseId = `resp_mixed_message_without_done_${stream}`;
3468: const itemId = `msg_mixed_message_without_done_${stream}`;
3475: const mixedEvents = [
3495: const followUp = streamFrom(mixedEvents);
3501: ? ...
7880: const outputIdentities = new Set<string>(); 10798: outputIdentities: new Set(outputIdentities), 11340: outputIdentities.clear(); 11537: outputIdentities: new Set(outputIdentities), 11751: outputIdentities.clear();
11730: throw new Error("principal retry did not return a stream");
11731: }
11732: } catch {
11733: if (signal.aborted) {
11734: void retryResponse?.body
11735: ?.cancel(signal.reason)
11736: .catch(() => {});
11737: throw signal.reason;
11738: }
11739: reportPrincipalTransportFailure({
11740: kind: error.kind,
11741: stage: "pre_output",
11742: outcome: "retry_exhausted",
11743: });
11744: throw error;
11745: }
11746:
11747: currentPrincipalResponse = retryResponse;
11748: state = makeResponsesAccState();
11749: syntheticIdentities.clear();
11750: referenceIdentities.clear();
11751: outputIdentities.clear();
11752: ...
lrqlyykxvwoq/ca83ea5361d7dc0b059fb9a8e855e892af1bd343 parent=tklkqxytmpkz/ce88c614c0ae conflict=no bookmarks= rwunmtrqvptr/3cbc99c3cf3844c67f5132411a5dcee3eda1723c parent=tklkqxytmpkz/ce88c614c0ae conflict=no bookmarks=fix/responses-accumulation-order tklkqxytmpkz/ce88c614c0ae55d229cb605aa7c7369d918d3650 parent=kvpmnlwnznqx/683dff02101f conflict=no bookmarks= A plain diff: 7c56d9f251cc66f2e3d46944fb9c82f95f4949d9ef5a19a255c174e0608d7b92 - B plain diff: e62ce85478a3c6bb026b669679c86db87df2f32e7680e1959e9980b1dfec56a6 -
10450: if (consumeReferenceEvent(state, referenceIndices, event, parsed)) {
10471: applyResponsesEvent(state, event, parsed);
10527: retainedStateBytes += encoder.encode(data).byteLength;
11080: consumeReferenceEvent(
11096: applyResponsesEvent(contState, ce, cparsed);
11151: retainedStateBytes += encoder.encode(cd).byteLength;
2541: test("rejects an item_reference missing output_item.done", async () => {
6169: test("rejects a recall item missing output_item.done", async () => {
418: sseEvent("response.output_text.done", {
1037: const materializedItem = finalizeSpy.mock.calls
1040: expect(materializedItem).toEqual(
1101: const materializedItem = finalizeSpy.mock.calls
1104: const materializedParts = materializedItem?.[
1107: expect(materializedParts).toHaveLength(scenario.partIndex + 1);
1108: expect(Object.keys(materializedParts)).toEqual([
1111: expect(materializedParts[scenario.partIndex]).toEqual(
1115: Object.keys(materializedParts[scenario.partIndex] as object),
1242: const materializedItem = finalizeSpy.mock.calls
1245: expect(materializedItem).toEqual(
1956: sseEvent("response.output_text.done", {
1988: sseEvent("response.output_text.done", {
2610: test("preserves an accepted sparse message content coordinate", async () => {
2631: sseEvent("response.output_text.done", {
2974: sseEvent("response.output_text.done", {
3017: sseEvent("response.output_text.done", {
3110: ...
3170: part: { type: "output_text", text: "sparse finalized text" },
3171: }),
3172: sseEvent("response.output_text.done", {
3173: output_index: 0,
3174: item_id: "msg_sparse_omitted_content",
3175: content_index: 2,
3176: text: "sparse finalized text",
3177: }),
3178: sseEvent("response.content_part.done", {
3179: output_index: 0,
3180: item_id: "msg_sparse_omitted_content",
3181: content_index: 2,
3182: part: { type: "output_text", text: "sparse finalized text" },
3183: }),
3184: sseEvent("response.output_item.done", {
3185: output_index: 0,
3186: item: {
3187: type: "message",
3188: id: "msg_sparse_omitted_content",
3189: role: "assistant",
3190: status: "completed",
3191: },
3192: }),
3193: sseEvent("response.completed", {
3194: response: ...
Classification: complementary. Neither candidate supersedes the other.
item_reference events bypass canonical accumulation and retained-state accounting.| Revision | Commit | Parent | Conflict |
|---|---|---|---|
| Base tklkqxyt | ce88c614c0ae55d229cb605aa7c7369d918d3650 | 683dff02101f | No |
| A lrqlyykx | ca83ea5361d7dc0b059fb9a8e855e892af1bd343 | Base | No |
| B rwunmtrq | 3cbc99c3cf3844c67f5132411a5dcee3eda1723c | Base | No |
7c56d9f251cc66f2e3d46944fb9c82f95f4949d9ef5a19a255c174e0608d7b92 — matches supplied hash.e62ce85478a3c6bb026b669679c86db87df2f32e7680e1959e9980b1dfec56a6.fix/responses-accumulation-order points to B.Paths below use:
P: packages/gateway/src/pipeline.tsT: packages/gateway/test/openai-responses-recall-aware-stream.test.tsA:P:11062-11071. Tests done-only and data-only omitted indices at the boundary: A:T:7683-7768.B:P:7932-7960, called with contIndex at B:P:10979-10985. Its test covers the last valid normalized index and the first invalid nested item, and asserts the invalid item was never accumulated: B:T:7921-8005.B prevents even temporary retention in normalization state. A’s post-normalization check remains valid defense in depth. ...