mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-18 21:22:28 +03:00
c545855b26
Diagnosed while chasing the reused-output-index incident (see 705ac7335 / OpenClaw issue #123342): every call-log artifact showed a wire-clean response, even for requests that actually failed, because withEarlyStreamKeepalive injects its startup/keepalive/error frames directly into the outer response stream, entirely outside the request handler's own reqLogger. reqLogger.appendConvertedChunk (which populates pipeline.streamChunks.client) never sees those bytes — only what chatCore.ts's own SSE writer produced. The persisted artifact was answering "what did the handler generate," not "what did the client actually receive," which is the wrong question when diagnosing a client-visible stream defect. withEarlyStreamKeepalive wraps the handler's Promise from OUTSIDE its call tree; the reqLogger it needs to feed is created deep inside chatCore.ts, after routing/model/provider resolution, and doesn't exist yet when the keepalive frames are written. The two sides share no reference — only an identifier, if one is deliberately threaded through both. Fix: responses/route.ts now generates a correlationId before calling handleChat, passes it as handleChat's existing (already-supported, previously-unused-here) 4th positional arg — which chatCore.ts already threads into trackPendingRequest's metadata as entry.correlationId, zero changes needed there — and also into withEarlyStreamKeepalive's options. The wrapper buffers every direct- to-client write (startup frame, periodic ticks, in-band error frames) via the new earlyKeepaliveByteBuffer module, keyed by that same id. chatCore/attemptLogging.ts, which already has correlationId in scope right where it assembles the final pipeline payload before saveCallLog, takes the buffered bytes and prepends them into streamChunks.client in send order. The verbatim-forwarded real response body is deliberately NOT re-recorded here — the handler's own reqLogger already captures that; recording it twice would duplicate it in the artifact. The buffer is consumed exactly once per correlationId and swept on a 10-minute TTL so a request that never reaches the persist call (aborted, detailed logging disabled, a route that doesn't opt in) cannot leak entries forever. Scoped to /v1/responses only, where the incident actually happened. /v1/chat/completions and /v1/messages call withEarlyStreamKeepalive the same way and would need the identical two-line route change to opt in; left as a follow-up rather than bundled in sight-unseen. Test plan: - tests/unit/early-keepalive-byte-buffer.test.ts (new): record/take ordering, single-consumption, per-id isolation, empty-input no-ops, unbounded-growth cap - tests/unit/early-stream-keepalive.test.ts: two new tests — a correlationId records the startup frame and keepalive ticks but NOT the forwarded body; omitting correlationId is a true no-op - tests/unit/attempt-logging-early-keepalive-merge.test.ts (new): real temp-DB end-to-end proof against the actual persisted call-log row — early bytes prepended in send order, consumed exactly once, no-op without a correlationId, gated by detailedLoggingEnabled matching the existing streamChunks capture gate - tests/unit/chatcore-attempt-logging.test.ts (existing): unchanged, still passing — confirms the merge addition doesn't disturb existing persistence behavior - 44 passed total across the above plus earlyStreamKeepalive.test.ts, 2 pre-existing skips unrelated to this change - tsgo --noEmit: clean on all touched files
52 lines
1.9 KiB
TypeScript
52 lines
1.9 KiB
TypeScript
import test from "node:test";
|
|
import assert from "node:assert/strict";
|
|
|
|
import {
|
|
recordEarlyKeepaliveBytes,
|
|
takeEarlyKeepaliveBytes,
|
|
} from "../../open-sse/utils/earlyKeepaliveByteBuffer.ts";
|
|
|
|
test("records are returned in the order they were written", () => {
|
|
const id = "buf-order-1";
|
|
recordEarlyKeepaliveBytes(id, "a");
|
|
recordEarlyKeepaliveBytes(id, "b");
|
|
recordEarlyKeepaliveBytes(id, "c");
|
|
assert.deepEqual(takeEarlyKeepaliveBytes(id), ["a", "b", "c"]);
|
|
});
|
|
|
|
test("take clears the entry — a second take for the same id is empty", () => {
|
|
const id = "buf-once-1";
|
|
recordEarlyKeepaliveBytes(id, "a");
|
|
takeEarlyKeepaliveBytes(id);
|
|
assert.deepEqual(takeEarlyKeepaliveBytes(id), []);
|
|
});
|
|
|
|
test("take on an id that was never recorded returns an empty array, not undefined", () => {
|
|
assert.deepEqual(takeEarlyKeepaliveBytes("buf-never-recorded"), []);
|
|
});
|
|
|
|
test("different correlation ids do not leak into each other", () => {
|
|
recordEarlyKeepaliveBytes("buf-iso-a", "only-a");
|
|
recordEarlyKeepaliveBytes("buf-iso-b", "only-b");
|
|
assert.deepEqual(takeEarlyKeepaliveBytes("buf-iso-a"), ["only-a"]);
|
|
assert.deepEqual(takeEarlyKeepaliveBytes("buf-iso-b"), ["only-b"]);
|
|
});
|
|
|
|
test("empty chunk and empty correlationId are both no-ops", () => {
|
|
const id = "buf-noop-1";
|
|
recordEarlyKeepaliveBytes(id, "");
|
|
recordEarlyKeepaliveBytes("", "should-not-record");
|
|
assert.deepEqual(takeEarlyKeepaliveBytes(id), []);
|
|
assert.deepEqual(takeEarlyKeepaliveBytes(""), []);
|
|
});
|
|
|
|
test("a pathologically long-lived slow path does not grow the buffer unbounded", () => {
|
|
const id = "buf-cap-1";
|
|
for (let i = 0; i < 1000; i++) {
|
|
recordEarlyKeepaliveBytes(id, `tick-${i}`);
|
|
}
|
|
const chunks = takeEarlyKeepaliveBytes(id);
|
|
assert.ok(chunks.length < 1000, "buffer must cap well below an unbounded 1000 items");
|
|
assert.ok(chunks.length > 0, "capping must not silently drop everything");
|
|
});
|