Distillation

Date: Sep 16, 2026

• 🟡 (22:41) In packages/gateway/src/pipeline.ts:9950-9959, stream-abort cleanup resumes pending demand, clears keepaliveTimer, and either calls cancelAndReleaseReader(activeReader, signal.reason) or cancels currentPrincipalResponse.body; onStreamAbort is registered once and invoked immediately if signal.aborted.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:9960-9974, the ReadableStream<Uint8Array> implements backpressure through waitForDemand(), which waits while (controller.desiredSize ?? 1) <= 0, unless canceled or aborted, and then calls signal.throwIfAborted().
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:9975-10002, safeEnqueue(...) waits for demand, enqueues sequenceChunk(chunk), marks the stream canceled if controller.enqueue(...) throws, and invokes its callback only after successful enqueue; enqueuePrincipal(...) then sets principalEventEmitted = true, optionally sets ordinaryToolEmitted = true, and invokes its own callback.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10003-10020, safeClose() and safeError(error) both call cleanupAbort(), return when already canceled, and suppress exceptions caused by an already closed or canceled controller.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10022-10037, resetKeepalive() schedules a recurring timer at KEEPALIVE_INACTIVITY_MS; each tick emits keepaliveComment through safeEnqueue(...) only when (controller.desiredSize ?? 1) > 0, then reschedules unless aborted. clearKeepalive() clears the timer and sets it to null.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10038-10056, principal-stream state tracks principalReader, principalTransportRetries, principalRetrySucceededReported, principalReadFinished, continuationAttempted, continuationFailureCategory, continuationFailureReported, and recallDetected; PrincipalFailureCategory supports "principal_transport", "principal_resource_limit", "principal_protocol", "principal_missing_terminal", and "principal_unexpected", initially "principal_unexpected".
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10057-10067, classifyPrincipalFailure(error) maps SSEStreamTransportError to "principal_transport" and SSEStreamLimitError to "principal_resource_limit"; all other errors use the current principalFailureCategory.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10068-10080, principalTransportStage() returns "post_tool" after an ordinary tool is emitted, otherwise "post_output" after any principal event, otherwise "pre_output"; reportContinuationFailure(category) reports through reportRecallContinuationFailure(category) at most once using continuationFailureReported.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10081-10088, recall items are designated gateway-internal and hidden on every exit, including failures before marker replacement; lifecycle tracking uses recallIndices: Set<number>, unresolvedToolIndices: Set<number>, and referenceIndices: Map<number, ReferenceLifecycle>, with snapshots retainedStateBaseline = retainedStateBytes and hiddenRecallBaseline = hiddenRecallBytes.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10089-10097, runPrincipalAttempt() resets principalReadFinished, throws "Upstream response has no body" when needed, obtains a reader from currentPrincipalResponse.body, and assigns it to both principalReader and activeReader.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10098-10117, each principal attempt initializes parsedRecallInputs: Map<number, RecallArguments>, ordered pendingRecalls: PendingResponsesRecall[], otherToolSeen = false, unresolvedToolBytes: Map<number, number>, and deferredEvents entries containing chunk plus optional candidateIndex; discardDeferredCandidate(outputIndex) removes all deferred entries for that candidate by iterating backward.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10118-10126, promoteDeferredCandidate(outputIndex) transfers that candidate’s byte count from unresolvedToolBytes into hiddenRecallBytes, deletes the unresolved entry, and throws new SSEStreamLimitError("recall stream exceeded deferred event limit") if hiddenRecallBytes > maxHiddenRecallBytes.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10128-10145, principal SSE parsing calls parseSSEStream(reader, { maxFrames: maxSSEFrames, inactivityMs: sseInactivityMs, signal, frameCounter }), resets keepalive on each upstream event, ignores empty data and "[DONE]", adds encoded formatResponsesEvent(event, data) size to streamBytes, and throws new SSEStreamLimitError("Responses stream exceeded byte limit") when streamBytes > maxStreamBytes.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10147-10171, each parsed frame first sets principalFailureCategory = "principal_protocol" and attempts JSON.parse(data); malformed JSON for an event whose name starts with response. throws malformed JSON in Responses event ${event}, while malformed non-message events are forwarded directly unless recalls or unresolved tools exist, in which case they are buffered in deferredEvents.
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10156-10169, buffered malformed non-JSON events increase deferredBytes and throw new SSEStreamLimitError("recall stream exceeded deferred event limit") if deferredBytes > maxDeferredBytes; otherwise direct forwarding uses enqueuePrincipal(chunk, otherToolSeen).
• 🟡 (22:41) In packages/gateway/src/pipeline.ts:10172-10183, a mismatch between parsed.type and the SSE event name throws Responses payload type does not match ${event}; response.output_item.added or response.output_item.done carrying a function_call whose name === RECALL_TOOL_NAME sets recallDetected = true.