mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-07 00:44:49 +00:00
Stream assistant text at a steady rate instead of in arrival lumps (#3612)
* Stream assistant text at a steady rate instead of in arrival lumps Assistant text arrived in lumps and was painted that way. The daemon coalesces one message per 60ms window, and that window carries however many characters the model produced in it, which swings by an order of magnitude within a turn. Painting each delta as it landed made the size of those lumps visible. Three changes: - The agent stream coalescer is now leading + trailing. The first delta after an idle window flushes synchronously instead of waiting a full window, matching the invariant TerminalOutputCoalescer already documents. The window stays at 60ms; shrinking it would raise message rate on a loop that already contends with terminal frames and relay encryption, without making the stream smoother. - The client reducer queue commits on a frame instead of an arbitrary 48ms timer, with the timer kept as a ceiling so a hidden tab still advances. - Assistant and reasoning text are revealed at a rate derived from the backlog rather than from arrival timing. Arrival changes the target, not the rate. The store keeps the full text and only the rendered slice is paced, so copy, selection, the chat outline, and scroll geometry all stay consistent with what is on screen. The leading flush means a burst's first chunk lands as its own timeline row. The projection already merges contiguous same-turn rows and clients read the projected timeline, so history is unchanged; the server tests that assert on raw rows were updated for the extra row. Measured on Expo web against a local dev daemon. Mock bursty-stream provider, 6s samples, paint-on-arrival vs paced: frames that advanced the text 55 -> 244, chars-per-frame CV 2.66 -> 1.44, gap p50 100ms -> 17ms. Real Claude Haiku agent over 52s: 84% of frames advance the text, CV 1.29, gap p50 and p95 both 17ms (one frame at 60Hz). Adds a bursty-stream mock model with seeded burst sizes, a gated browser perf spec, and docs/agent-stream-performance.md. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * Address review: safe reveal boundaries and a testable seam Two findings from the PR review. Unicode boundaries. Slicing the revealed text at an arbitrary index could strand a lone surrogate, which paints as a replacement glyph, or split an emoji ZWJ sequence so a family briefly rendered as its members. Both were visible for a frame or two mid-stream — the exact flicker the paced reveal exists to remove. The rendered cut is now pulled back off surrogate pairs, combining marks, variation selectors, skin tone modifiers, and ZWJ joins. The reveal counter stays monotonic and only the painted boundary moves, so a long cluster cannot stall the reveal. Test placement. The hook test used JSDOM, @testing-library renderHook, and monkey-patched requestAnimationFrame, all of which docs/testing.md rules out; it prescribes fixing the missing seam instead. The reveal policy is now a pure state machine in text-reveal.ts (begin/retarget/advance/ complete/visible) covered by plain unit tests with no environment, and the hook is reduced to the frame-clock binding. Rendered behavior is left to the browser spec, which no longer carries frame sampling and statistics in its body — those moved to e2e/support/helpers/stream-smoothness.ts. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * Measure total painted text, not the tail message The smoothness sampler read the length of the last assistant-message element. A real turn emits many assistant messages, so that element keeps changing identity and its length is not monotonic — handovers read as resets, and a stream that was plainly moving on screen reported almost no growth at all. Sampling the total across every assistant message fixes it. This invalidated the numbers previously recorded for the real-agent run, so the doc table is replaced with a fresh A/B taken with the corrected sampler against a Claude Haiku agent: frames that advanced the text 6% -> 87%, chars-per-frame CV 4.11 -> 1.86, gap p50 317ms -> 17ms, p95 383ms -> 17ms. The paced column also reported zero Unicode defects across a stream containing emoji, which exercises the reveal boundary clamp end to end. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * perf(agent-stream): harden paced text reveal Use standards-compliant grapheme segmentation when available and preserve full arrivals on older runtimes. Cap reveal commits at 60Hz and measure the real browser rendering path so high-refresh displays stay smooth without rendering at hardware frame rate. * Hold back trailing fragments split by arrival boundaries The reveal boundary is not the only place a grapheme gets cut. The daemon's coalescing window ends a delta wherever the window closes, so a caught-up reveal can be handed half a cluster and paint it. Measured on a mixed-script Haiku stream: five frames in ~350 ending in a single regional indicator, i.e. half a flag on screen for a frame. While the turn is streaming, a trailing fragment that cannot stand alone — a dangling high surrogate, a trailing joiner, an odd regional indicator — now waits for the rest of its cluster. Once the turn ends everything received is painted, because nothing more is coming. Complements the grapheme-boundary clamp, which handles the reveal cut; this handles the arrival cut. The same stream now measures one such frame in ~400 rather than five. The remainder is a flush boundary that coincides with the end of an assistant message, putting the two halves in different messages, which the client cannot rejoin — that one belongs in the coalescer. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
This commit is contained in:
Thomas Beer
co-authored by
Claude Opus 5
Mohamed Boudra
Claude Opus 5
Mohamed Boudra
parent
e19d3914ae
commit
e22ad117fd
@@ -7,6 +7,7 @@ import { createTestLogger } from "../../test-utils/test-logger.js";
|
||||
import { AgentManager, type AgentManagerEvent } from "./agent-manager.js";
|
||||
import { AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS } from "./agent-stream-coalescer.js";
|
||||
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import type {
|
||||
AgentCapabilityFlags,
|
||||
AgentClient,
|
||||
@@ -357,6 +358,13 @@ function getTimelineItems(rows: AgentTimelineRow[]): AgentTimelineItem[] {
|
||||
return rows.map((row) => row.item);
|
||||
}
|
||||
|
||||
// The leading-edge flush emits the first chunk of a burst as its own row. Clients
|
||||
// read the projected timeline, where mergeAssistantChunks/mergeReasoningChunks
|
||||
// join those contiguous rows back together, so history reads the same either way.
|
||||
function getProjectedTimelineItems(rows: AgentTimelineRow[]): AgentTimelineItem[] {
|
||||
return projectTimelineRows({ rows, mode: "projected" }).map((entry) => entry.item);
|
||||
}
|
||||
|
||||
function expectContiguousRowSeqs(rows: AgentTimelineRow[], expected: number[]): void {
|
||||
expect(rows.map((row) => row.seq)).toEqual(expected);
|
||||
}
|
||||
@@ -505,29 +513,40 @@ describe("target coalesced behavior", () => {
|
||||
}
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
// Leading edge: the first chunk is already out, the other 999 are buffered.
|
||||
await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0);
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "x" },
|
||||
]);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(events).toHaveLength(1);
|
||||
expect(rows[0]?.item).toEqual({
|
||||
type: "assistant_message",
|
||||
text: "x".repeat(1000),
|
||||
});
|
||||
expect(rows).toHaveLength(2);
|
||||
expect(events).toHaveLength(2);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "x" },
|
||||
{ type: "assistant_message", text: "x".repeat(999) },
|
||||
]);
|
||||
expect(events.map((event) => (event.type === "agent_stream" ? event.event : null))).toEqual([
|
||||
{
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: { type: "assistant_message", text: "x".repeat(1000) },
|
||||
item: { type: "assistant_message", text: "x" },
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
item: { type: "assistant_message", text: "x".repeat(999) },
|
||||
},
|
||||
]);
|
||||
expectContiguousRowSeqs(rows, [1]);
|
||||
expectContiguousLiveSeqs(events, [1]);
|
||||
expectContiguousRowSeqs(rows, [1, 2]);
|
||||
expectContiguousLiveSeqs(events, [1, 2]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "x".repeat(1000) },
|
||||
]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -548,10 +567,16 @@ describe("target coalesced behavior", () => {
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([{ type: "reasoning", text: "r".repeat(100) }]);
|
||||
expect(events).toHaveLength(1);
|
||||
expectContiguousRowSeqs(rows, [1]);
|
||||
expectContiguousLiveSeqs(events, [1]);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "reasoning", text: "r" },
|
||||
{ type: "reasoning", text: "r".repeat(99) },
|
||||
]);
|
||||
expect(events).toHaveLength(2);
|
||||
expectContiguousRowSeqs(rows, [1, 2]);
|
||||
expectContiguousLiveSeqs(events, [1, 2]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "reasoning", text: "r".repeat(100) },
|
||||
]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -570,34 +595,47 @@ describe("target coalesced behavior", () => {
|
||||
session.pushEvent(timelineEvent(runningToolCall));
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0);
|
||||
// Only the leading chunk is out; the running tool call stays buffered with
|
||||
// the rest of the text rather than forcing an early flush.
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "a" },
|
||||
]);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(COALESCE_WINDOW_MS);
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "a".repeat(500) },
|
||||
{ type: "assistant_message", text: "a" },
|
||||
{ type: "assistant_message", text: "a".repeat(499) },
|
||||
runningToolCall,
|
||||
]);
|
||||
expectContiguousLiveSeqs(getTimelineStreamEvents(harness.events, agentId), [1, 2]);
|
||||
expectContiguousLiveSeqs(getTimelineStreamEvents(harness.events, agentId), [1, 2, 3]);
|
||||
|
||||
for (let i = 0; i < 500; i++) {
|
||||
session.pushEvent(assistant("b"));
|
||||
}
|
||||
await waitForSessionEventQueue();
|
||||
// The "b" burst arrives inside the window that just flushed, so it gets no
|
||||
// leading flush of its own and coalesces whole.
|
||||
await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(2);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(3);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "a" },
|
||||
{ type: "assistant_message", text: "a".repeat(499) },
|
||||
runningToolCall,
|
||||
{ type: "assistant_message", text: "b".repeat(500) },
|
||||
]);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3, 4]);
|
||||
expectContiguousLiveSeqs(events, [1, 2, 3, 4]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "a".repeat(500) },
|
||||
runningToolCall,
|
||||
{ type: "assistant_message", text: "b".repeat(500) },
|
||||
]);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3]);
|
||||
expectContiguousLiveSeqs(events, [1, 2, 3]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -614,18 +652,27 @@ describe("target coalesced behavior", () => {
|
||||
}
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
// The leading edge flushes the first snapshot; the remaining 199 collapse to
|
||||
// the latest one. Both carry the same callId, so the projection folds them
|
||||
// into a single tool call row.
|
||||
await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0);
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
toolCall({ output: "chunk-0" }),
|
||||
]);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([toolCall({ output: "chunk-199" })]);
|
||||
expect(events).toHaveLength(1);
|
||||
expectContiguousRowSeqs(rows, [1]);
|
||||
expectContiguousLiveSeqs(events, [1]);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
toolCall({ output: "chunk-0" }),
|
||||
toolCall({ output: "chunk-199" }),
|
||||
]);
|
||||
expect(events).toHaveLength(2);
|
||||
expectContiguousRowSeqs(rows, [1, 2]);
|
||||
expectContiguousLiveSeqs(events, [1, 2]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([toolCall({ output: "chunk-199" })]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -648,12 +695,17 @@ describe("target coalesced behavior", () => {
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
toolCall({ callId: "tool-1", output: "one-a" }),
|
||||
toolCall({ callId: "tool-2", output: "two-b" }),
|
||||
toolCall({ callId: "tool-1", output: "one-b" }),
|
||||
]);
|
||||
expect(events).toHaveLength(3);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3]);
|
||||
expectContiguousLiveSeqs(events, [1, 2, 3]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
toolCall({ callId: "tool-1", output: "one-b" }),
|
||||
toolCall({ callId: "tool-2", output: "two-b" }),
|
||||
]);
|
||||
expect(events).toHaveLength(2);
|
||||
expectContiguousRowSeqs(rows, [1, 2]);
|
||||
expectContiguousLiveSeqs(events, [1, 2]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -668,7 +720,10 @@ describe("target coalesced behavior", () => {
|
||||
session.pushEvent(assistant("before"));
|
||||
session.pushEvent(timelineEvent(toolCall({ output: "running" })));
|
||||
await waitForSessionEventQueue();
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0);
|
||||
// "before" led the burst; the running tool call is still buffered.
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "before" },
|
||||
]);
|
||||
|
||||
session.pushEvent(timelineEvent(toolCall({ status: "completed", output: "done" })));
|
||||
await waitForSessionEventQueue();
|
||||
@@ -821,14 +876,26 @@ describe("target coalesced behavior", () => {
|
||||
const firstEvents = getTimelineStreamEvents(harness.events, first.agentId);
|
||||
const secondEvents = getTimelineStreamEvents(harness.events, second.agentId);
|
||||
|
||||
expect(getTimelineItems(firstRows)).toEqual([{ type: "assistant_message", text: "aa" }]);
|
||||
expect(getTimelineItems(secondRows)).toEqual([{ type: "assistant_message", text: "bb" }]);
|
||||
expect(firstEvents).toHaveLength(1);
|
||||
expect(secondEvents).toHaveLength(1);
|
||||
expectContiguousRowSeqs(firstRows, [1]);
|
||||
expectContiguousRowSeqs(secondRows, [1]);
|
||||
expectContiguousLiveSeqs(firstEvents, [1]);
|
||||
expectContiguousLiveSeqs(secondEvents, [1]);
|
||||
expect(getTimelineItems(firstRows)).toEqual([
|
||||
{ type: "assistant_message", text: "a" },
|
||||
{ type: "assistant_message", text: "a" },
|
||||
]);
|
||||
expect(getTimelineItems(secondRows)).toEqual([
|
||||
{ type: "assistant_message", text: "b" },
|
||||
{ type: "assistant_message", text: "b" },
|
||||
]);
|
||||
expect(firstEvents).toHaveLength(2);
|
||||
expect(secondEvents).toHaveLength(2);
|
||||
expectContiguousRowSeqs(firstRows, [1, 2]);
|
||||
expectContiguousRowSeqs(secondRows, [1, 2]);
|
||||
expectContiguousLiveSeqs(firstEvents, [1, 2]);
|
||||
expectContiguousLiveSeqs(secondEvents, [1, 2]);
|
||||
expect(getProjectedTimelineItems(firstRows)).toEqual([
|
||||
{ type: "assistant_message", text: "aa" },
|
||||
]);
|
||||
expect(getProjectedTimelineItems(secondRows)).toEqual([
|
||||
{ type: "assistant_message", text: "bb" },
|
||||
]);
|
||||
rmSync(secondWorkdir, { recursive: true, force: true });
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
@@ -923,10 +990,17 @@ describe("target coalesced behavior", () => {
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: " \n\tdone" }]);
|
||||
expect(events).toHaveLength(1);
|
||||
expectContiguousRowSeqs(rows, [1]);
|
||||
expectContiguousLiveSeqs(events, [1]);
|
||||
// The empty chunk is dropped before the coalescer, so " " leads the burst.
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: " " },
|
||||
{ type: "assistant_message", text: "\n\tdone" },
|
||||
]);
|
||||
expect(events).toHaveLength(2);
|
||||
expectContiguousRowSeqs(rows, [1, 2]);
|
||||
expectContiguousLiveSeqs(events, [1, 2]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: " \n\tdone" },
|
||||
]);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -1115,13 +1189,18 @@ describe("target coalesced behavior", () => {
|
||||
const timelineEvents = getTimelineStreamEvents(idempotencyHarness.events, agentId);
|
||||
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "abc" },
|
||||
{ type: "assistant_message", text: "a" },
|
||||
{ type: "assistant_message", text: "bc" },
|
||||
TOOL_CALL,
|
||||
TOOL_CALL,
|
||||
]);
|
||||
expect(timelineEvents).toHaveLength(3);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3]);
|
||||
expectContiguousLiveSeqs(timelineEvents, [1, 2, 3]);
|
||||
expect(timelineEvents).toHaveLength(4);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3, 4]);
|
||||
expectContiguousLiveSeqs(timelineEvents, [1, 2, 3, 4]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "abc" },
|
||||
TOOL_CALL,
|
||||
]);
|
||||
} finally {
|
||||
idempotencyHarness.cleanup();
|
||||
}
|
||||
@@ -1150,9 +1229,8 @@ describe("target coalesced behavior", () => {
|
||||
session.pushEvent(assistant("A"));
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(COALESCE_WINDOW_MS);
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
// "A" flushes on the leading edge; the chunk the subscriber pushes during
|
||||
// that flush must land in the next window, not corrupt this one.
|
||||
const rowsAfterFirstFlush = await reentryHarness.manager.getTimelineRows(agentId);
|
||||
expect(getTimelineItems(rowsAfterFirstFlush)).toEqual([
|
||||
{ type: "assistant_message", text: "A" },
|
||||
@@ -1184,19 +1262,26 @@ describe("target coalesced behavior", () => {
|
||||
const { agentId, session } = await createManagedSession(harness);
|
||||
|
||||
session.pushEvent(assistant("flush"));
|
||||
session.pushEvent(assistant("-pending"));
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0);
|
||||
// "flush" left on the leading edge; "-pending" is what manager.flush() drains.
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "flush" },
|
||||
]);
|
||||
await harness.manager.flush();
|
||||
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const events = getTimelineStreamEvents(harness.events, agentId);
|
||||
expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: "flush" }]);
|
||||
expect(events).toHaveLength(1);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "flush" },
|
||||
{ type: "assistant_message", text: "-pending" },
|
||||
]);
|
||||
expect(events).toHaveLength(2);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(1);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1);
|
||||
expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(2);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(2);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
@@ -1257,7 +1342,16 @@ describe("target coalesced behavior", () => {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
turnId: "turn-1",
|
||||
item: { type: "assistant_message", text: "hello world" },
|
||||
item: { type: "assistant_message", text: "hello " },
|
||||
},
|
||||
});
|
||||
await expect(stream.next()).resolves.toEqual({
|
||||
done: false,
|
||||
value: {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
turnId: "turn-1",
|
||||
item: { type: "assistant_message", text: "world" },
|
||||
},
|
||||
});
|
||||
await expect(stream.next()).resolves.toEqual({
|
||||
@@ -1268,7 +1362,13 @@ describe("target coalesced behavior", () => {
|
||||
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
const streamEvents = getStreamEvents(harness.events, agentId);
|
||||
expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: "hello world" }]);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "hello " },
|
||||
{ type: "assistant_message", text: "world" },
|
||||
]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "hello world" },
|
||||
]);
|
||||
expect(streamEvents[0]).toMatchObject({
|
||||
type: "agent_stream",
|
||||
event: { type: "turn_started", provider: "codex", turnId: "turn-1" },
|
||||
@@ -1279,10 +1379,19 @@ describe("target coalesced behavior", () => {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
turnId: "turn-1",
|
||||
item: { type: "assistant_message", text: "hello world" },
|
||||
item: { type: "assistant_message", text: "hello " },
|
||||
},
|
||||
});
|
||||
expect(streamEvents[2]).toMatchObject({
|
||||
type: "agent_stream",
|
||||
event: {
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
turnId: "turn-1",
|
||||
item: { type: "assistant_message", text: "world" },
|
||||
},
|
||||
});
|
||||
expect(streamEvents[3]).toMatchObject({
|
||||
type: "agent_stream",
|
||||
event: { type: "turn_completed", provider: "codex", turnId: "turn-1" },
|
||||
});
|
||||
@@ -1308,22 +1417,29 @@ describe("target coalesced behavior", () => {
|
||||
|
||||
const rows = await harness.manager.getTimelineRows(agentId);
|
||||
expect(getTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "a1" },
|
||||
{ type: "assistant_message", text: "a2" },
|
||||
{ type: "reasoning", text: "r1r2" },
|
||||
{ type: "assistant_message", text: "b1b2" },
|
||||
]);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3, 4]);
|
||||
const timelineEvents = getTimelineStreamEvents(harness.events, agentId);
|
||||
expect(timelineEvents).toHaveLength(4);
|
||||
expectContiguousLiveSeqs(timelineEvents, [1, 2, 3, 4]);
|
||||
expect(getProjectedTimelineItems(rows)).toEqual([
|
||||
{ type: "assistant_message", text: "a1a2" },
|
||||
{ type: "reasoning", text: "r1r2" },
|
||||
{ type: "assistant_message", text: "b1b2" },
|
||||
]);
|
||||
expectContiguousRowSeqs(rows, [1, 2, 3]);
|
||||
const timelineEvents = getTimelineStreamEvents(harness.events, agentId);
|
||||
expect(timelineEvents).toHaveLength(3);
|
||||
expectContiguousLiveSeqs(timelineEvents, [1, 2, 3]);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
|
||||
{ type: "assistant_message", text: "a1a2" },
|
||||
{ type: "assistant_message", text: "a1" },
|
||||
{ type: "assistant_message", text: "a2" },
|
||||
{ type: "reasoning", text: "r1r2" },
|
||||
{ type: "assistant_message", text: "b1b2" },
|
||||
]);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(3);
|
||||
expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(4);
|
||||
} finally {
|
||||
harness.cleanup();
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import { AgentStorage } from "./agent-storage.js";
|
||||
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
|
||||
import { toAgentPayload } from "./agent-projections.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
|
||||
import { formatSystemNotificationPrompt, startAgentRun } from "./agent-prompt.js";
|
||||
import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js";
|
||||
@@ -4969,34 +4970,52 @@ test("coalesces assistant chunks and persists the canonical row", async () => {
|
||||
}
|
||||
}
|
||||
|
||||
// The coalescer flushes the first chunk on the leading edge, so "final " ships
|
||||
// as its own row and "reply" follows on the trailing window. Clients read the
|
||||
// projected timeline, which merges the two back into one assistant message.
|
||||
const assistantTimelineEvents = streamEvents.filter(
|
||||
(event) => event.itemType === "assistant_message",
|
||||
);
|
||||
expect(assistantTimelineEvents).toHaveLength(1);
|
||||
expect(assistantTimelineEvents).toHaveLength(2);
|
||||
expect(assistantTimelineEvents[0]).toMatchObject({
|
||||
eventType: "timeline",
|
||||
itemType: "assistant_message",
|
||||
text: "final reply",
|
||||
text: "final ",
|
||||
seq: 1,
|
||||
epoch: expect.any(String),
|
||||
});
|
||||
expect(assistantTimelineEvents[1]).toMatchObject({
|
||||
eventType: "timeline",
|
||||
itemType: "assistant_message",
|
||||
text: "reply",
|
||||
seq: 2,
|
||||
epoch: expect.any(String),
|
||||
});
|
||||
|
||||
expect(manager.getTimeline(snapshot.id)).toEqual([
|
||||
{
|
||||
type: "assistant_message",
|
||||
text: "final reply",
|
||||
text: "final ",
|
||||
},
|
||||
{
|
||||
type: "assistant_message",
|
||||
text: "reply",
|
||||
},
|
||||
]);
|
||||
const fetched = await manager.fetchTimeline(snapshot.id, {
|
||||
direction: "tail",
|
||||
limit: 0,
|
||||
});
|
||||
expect(fetched.rows).toHaveLength(1);
|
||||
expect(fetched.rows).toHaveLength(2);
|
||||
expect(assistantTimelineEvents[0]?.epoch).toBe(fetched.epoch);
|
||||
expect(fetched.rows[0]?.item).toEqual({
|
||||
type: "assistant_message",
|
||||
text: "final reply",
|
||||
});
|
||||
expect(projectTimelineRows({ rows: fetched.rows, mode: "projected" }).map((e) => e.item)).toEqual(
|
||||
[
|
||||
{
|
||||
type: "assistant_message",
|
||||
text: "final reply",
|
||||
},
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
test("fetchTimeline supports older-history pagination with before seq", async () => {
|
||||
|
||||
@@ -25,6 +25,24 @@ function createHarness(windowMs?: number) {
|
||||
return { coalescer, flushes };
|
||||
}
|
||||
|
||||
// Consume the leading-edge flush for an agent and clear it from the record, so a
|
||||
// test can assert trailing-window batching on its own. Leaves the coalescer
|
||||
// inside the window, which is where the trailing timer governs.
|
||||
function primeLeadingEdge(
|
||||
coalescer: AgentStreamCoalescer,
|
||||
flushes: AgentStreamCoalescerFlush[],
|
||||
agentId = "agent-1",
|
||||
): void {
|
||||
coalescer.handle(agentId, assistant("prime"));
|
||||
const index = flushes.findIndex(
|
||||
(flush) => flush.agentId === agentId && flush.item.type === "assistant_message",
|
||||
);
|
||||
if (index === -1) {
|
||||
throw new Error(`expected a leading-edge flush for ${agentId}`);
|
||||
}
|
||||
flushes.splice(index, 1);
|
||||
}
|
||||
|
||||
function timeline(
|
||||
item: Extract<AgentStreamEvent, { type: "timeline" }>["item"],
|
||||
options?: {
|
||||
@@ -104,9 +122,23 @@ describe("AgentStreamCoalescer", () => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
test("coalesces same-tick assistant chunks after the configured window", async () => {
|
||||
test("flushes the first chunk of a burst on the leading edge", () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
|
||||
expect(coalescer.handle("agent-1", assistant("hel"))).toBe(true);
|
||||
expect(flushes).toEqual([
|
||||
{
|
||||
agentId: "agent-1",
|
||||
item: { type: "assistant_message", text: "hel" },
|
||||
provider: "codex",
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("coalesces the rest of a burst until the configured window elapses", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
expect(coalescer.handle("agent-1", assistant("hel"))).toBe(true);
|
||||
expect(coalescer.handle("agent-1", assistant("lo"))).toBe(true);
|
||||
|
||||
@@ -124,10 +156,26 @@ describe("AgentStreamCoalescer", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
test("leads again after an idle window", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
|
||||
coalescer.handle("agent-1", assistant("first"));
|
||||
expect(flushes).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60);
|
||||
coalescer.handle("agent-1", assistant("second"));
|
||||
|
||||
expect(flushes.map((flush) => flush.item)).toEqual([
|
||||
{ type: "assistant_message", text: "first" },
|
||||
{ type: "assistant_message", text: "second" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("uses constructor windowMs instead of a hard-coded value", async () => {
|
||||
const { coalescer, flushes } = createHarness(10);
|
||||
|
||||
expect(AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS).toBe(60);
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
expect(coalescer.handle("agent-1", assistant("fast"))).toBe(true);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(9);
|
||||
@@ -321,6 +369,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("preserves whitespace byte-exactly", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
coalescer.handle("agent-1", assistant(" "));
|
||||
coalescer.handle("agent-1", assistant("\n"));
|
||||
@@ -342,6 +391,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("reconstructs bytes exactly for fragmented text", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
const text = "Hello, 世界.\nPunctuation: ,.!?;: — but JS strings stay intact.\n";
|
||||
const chunks = [
|
||||
"Hello",
|
||||
@@ -369,6 +419,8 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("isolates buffers per agent", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes, "agent-1");
|
||||
primeLeadingEdge(coalescer, flushes, "agent-2");
|
||||
|
||||
coalescer.handle("agent-1", assistant("a"));
|
||||
coalescer.handle("agent-2", assistant("x"));
|
||||
@@ -445,6 +497,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
},
|
||||
});
|
||||
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
coalescer.handle("agent-1", assistant("first"));
|
||||
coalescer.flushFor("agent-1");
|
||||
|
||||
@@ -522,6 +575,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
},
|
||||
});
|
||||
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
coalescer.handle("agent-1", assistant("old"));
|
||||
const oldTimerCallback = scheduled[0];
|
||||
coalescer.flushAndDiscard("agent-1");
|
||||
@@ -546,6 +600,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("preserves first chunk item shape and replaces only text on collapse", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
const firstItem = {
|
||||
type: "assistant_message" as const,
|
||||
text: "he",
|
||||
@@ -571,6 +626,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("coalesces tool call updates by callId with latest snapshot winning", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
expect(coalescer.handle("agent-1", toolCall({ output: "first" }))).toBe(true);
|
||||
expect(coalescer.handle("agent-1", toolCall({ output: "second" }))).toBe(true);
|
||||
@@ -600,6 +656,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("coalesces interleaved tool call updates independently by callId", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
coalescer.handle("agent-1", toolCall({ callId: "tool-1", output: "one-a" }));
|
||||
coalescer.handle("agent-1", toolCall({ callId: "tool-2", output: "two-a" }));
|
||||
@@ -640,6 +697,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("terminal tool call statuses flush immediately", () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
coalescer.handle("agent-1", toolCall({ output: "running" }));
|
||||
expect(flushes).toEqual([]);
|
||||
@@ -667,6 +725,7 @@ describe("AgentStreamCoalescer", () => {
|
||||
|
||||
test("preserves mixed text and tool call arrival order within a flush", async () => {
|
||||
const { coalescer, flushes } = createHarness();
|
||||
primeLeadingEdge(coalescer, flushes);
|
||||
|
||||
coalescer.handle("agent-1", assistant("a"));
|
||||
coalescer.handle("agent-1", toolCall({ output: "running" }));
|
||||
|
||||
@@ -25,6 +25,7 @@ export interface AgentStreamCoalescerFlush {
|
||||
export interface AgentStreamCoalescerOptions {
|
||||
windowMs?: number;
|
||||
timers: AgentStreamCoalescerTimers;
|
||||
now?: () => number;
|
||||
onFlush: (payload: AgentStreamCoalescerFlush) => void;
|
||||
}
|
||||
|
||||
@@ -51,6 +52,7 @@ interface PendingAgentStreamBuffer {
|
||||
toolCallEntryIndexes: Map<string, number>;
|
||||
timer: ReturnType<typeof setTimeout> | null;
|
||||
flushing: boolean;
|
||||
lastFlushAt: number | null;
|
||||
}
|
||||
|
||||
function isCoalescableTimelineEvent(event: AgentStreamEvent): event is CoalescableTimelineEvent {
|
||||
@@ -88,10 +90,12 @@ export class AgentStreamCoalescer {
|
||||
private readonly onFlush: (payload: AgentStreamCoalescerFlush) => void;
|
||||
private readonly timers: AgentStreamCoalescerTimers;
|
||||
private readonly windowMs: number;
|
||||
private readonly now: () => number;
|
||||
|
||||
constructor(options: AgentStreamCoalescerOptions) {
|
||||
this.windowMs = options.windowMs ?? AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS;
|
||||
this.timers = options.timers;
|
||||
this.now = options.now ?? Date.now;
|
||||
this.onFlush = options.onFlush;
|
||||
}
|
||||
|
||||
@@ -112,7 +116,17 @@ export class AgentStreamCoalescer {
|
||||
return true;
|
||||
}
|
||||
|
||||
// Leading edge: the first event after an idle window flushes synchronously so
|
||||
// the first token of a turn isn't delayed a full window. Sustained bursts fall
|
||||
// through to the trailing timer, which keeps the message rate at one per
|
||||
// window. Same shape as TerminalOutputCoalescer.
|
||||
if (!buffer.timer) {
|
||||
const elapsed =
|
||||
buffer.lastFlushAt === null ? Number.POSITIVE_INFINITY : this.now() - buffer.lastFlushAt;
|
||||
if (elapsed >= this.windowMs) {
|
||||
this.flushBuffer(agentId);
|
||||
return true;
|
||||
}
|
||||
this.scheduleFlush(buffer);
|
||||
}
|
||||
|
||||
@@ -150,6 +164,7 @@ export class AgentStreamCoalescer {
|
||||
toolCallEntryIndexes: new Map(),
|
||||
timer: null,
|
||||
flushing: false,
|
||||
lastFlushAt: null,
|
||||
};
|
||||
this.buffers.set(agentId, buffer);
|
||||
return buffer;
|
||||
@@ -221,6 +236,7 @@ export class AgentStreamCoalescer {
|
||||
buffer.entries = [];
|
||||
buffer.toolCallEntryIndexes.clear();
|
||||
buffer.flushing = true;
|
||||
buffer.lastFlushAt = this.now();
|
||||
|
||||
try {
|
||||
for (const entry of this.collapseEntries(entries)) {
|
||||
|
||||
@@ -424,6 +424,31 @@ describe("MockLoadTestAgentClient", () => {
|
||||
unsubscribe();
|
||||
});
|
||||
|
||||
test("uses one continuous assistant stream for bursty rendering measurements", async () => {
|
||||
vi.useFakeTimers();
|
||||
const client = new MockLoadTestAgentClient();
|
||||
const session = await client.createSession({
|
||||
provider: "mock",
|
||||
cwd: process.cwd(),
|
||||
model: "bursty-stream",
|
||||
});
|
||||
const events: AgentStreamEvent[] = [];
|
||||
const unsubscribe = session.subscribe((event) => events.push(event));
|
||||
|
||||
await session.startTurn("Measure paced rendering.");
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
const timelineItems = events.flatMap((event) =>
|
||||
event.type === "timeline" ? [event.item] : [],
|
||||
);
|
||||
expect(timelineItems.filter((item) => item.type === "assistant_message")).not.toHaveLength(0);
|
||||
expect(
|
||||
timelineItems.filter((item) => item.type === "reasoning" || item.type === "tool_call"),
|
||||
).toEqual([]);
|
||||
await session.interrupt();
|
||||
unsubscribe();
|
||||
});
|
||||
|
||||
test("emits a settled assistant Markdown image path selected by prompt", async () => {
|
||||
vi.useFakeTimers();
|
||||
const client = new MockLoadTestAgentClient();
|
||||
@@ -505,11 +530,13 @@ describe("MockLoadTestAgentClient", () => {
|
||||
.reduce((max, length) => Math.max(max, length), 0);
|
||||
expect(longestMessage).toBeGreaterThan(20);
|
||||
|
||||
// First message includes the cycle header.
|
||||
expect(assistantMessages[0]).toMatchObject({
|
||||
type: "assistant_message",
|
||||
text: expect.stringContaining("## Cycle 1"),
|
||||
});
|
||||
// The cycle header is at the start of the stream. It can straddle the first
|
||||
// two messages, because the coalescer flushes the leading token of a burst
|
||||
// on its own before batching the rest.
|
||||
const assistantText = assistantMessages
|
||||
.map((item) => (item.type === "assistant_message" ? item.text : ""))
|
||||
.join("");
|
||||
expect(assistantText).toContain("## Cycle 1");
|
||||
|
||||
// Tool calls land in expected order at least once.
|
||||
const runningTools = toolCalls
|
||||
|
||||
@@ -115,6 +115,20 @@ const MODELS: AgentModelDefinition[] = [
|
||||
intervalMs: 40,
|
||||
},
|
||||
},
|
||||
{
|
||||
provider: MOCK_LOAD_TEST_PROVIDER_ID,
|
||||
id: "bursty-stream",
|
||||
label: "Bursty stream",
|
||||
description:
|
||||
"Emits tokens in uneven bursts separated by idle gaps, reproducing the lumpy arrival pattern real models produce. Use this to measure streaming smoothness.",
|
||||
metadata: {
|
||||
durationMs: 60_000,
|
||||
intervalMs: 0,
|
||||
burstMinTokens: 1,
|
||||
burstMaxTokens: 40,
|
||||
burstGapMs: 90,
|
||||
},
|
||||
},
|
||||
{
|
||||
provider: MOCK_LOAD_TEST_PROVIDER_ID,
|
||||
id: "ten-second-stream",
|
||||
@@ -144,6 +158,28 @@ const MODELS: AgentModelDefinition[] = [
|
||||
},
|
||||
];
|
||||
|
||||
/**
|
||||
* Bursty emission: instead of one token per interval, emit a run of tokens
|
||||
* back-to-back and then idle. Real models arrive this way, and it is the arrival
|
||||
* pattern the client's paced reveal exists to smooth out. Burst sizes come from a
|
||||
* seeded generator so a run is reproducible.
|
||||
*/
|
||||
interface BurstProfile {
|
||||
minTokens: number;
|
||||
maxTokens: number;
|
||||
gapMs: number;
|
||||
}
|
||||
|
||||
function nextBurstSize(profile: BurstProfile, sequence: number): number {
|
||||
// Deterministic hash of the burst index; no Math.random so runs are repeatable.
|
||||
let hash = (sequence + 1) * 2654435761;
|
||||
hash ^= hash >>> 15;
|
||||
hash = Math.imul(hash, 2246822519);
|
||||
hash ^= hash >>> 13;
|
||||
const span = profile.maxTokens - profile.minTokens + 1;
|
||||
return profile.minTokens + ((hash >>> 0) % span);
|
||||
}
|
||||
|
||||
interface ActiveTurn {
|
||||
turnId: string;
|
||||
assistantMessageId: string;
|
||||
@@ -158,6 +194,8 @@ interface ActiveTurn {
|
||||
queue: CycleEvent[];
|
||||
emittedTokens: number;
|
||||
turnStarted: boolean;
|
||||
burst: BurstProfile | null;
|
||||
burstIndex: number;
|
||||
}
|
||||
|
||||
type CycleEvent =
|
||||
@@ -277,15 +315,28 @@ function resolveModelProfile(modelId: string | null | undefined): {
|
||||
modelId: string;
|
||||
durationMs: number;
|
||||
intervalMs: number;
|
||||
burst: BurstProfile | null;
|
||||
} {
|
||||
const model = MODELS.find((entry) => entry.id === modelId) ?? MODELS[0];
|
||||
const metadata = model.metadata ?? {};
|
||||
const burstMinTokens =
|
||||
typeof metadata.burstMinTokens === "number" ? metadata.burstMinTokens : null;
|
||||
const burstMaxTokens =
|
||||
typeof metadata.burstMaxTokens === "number" ? metadata.burstMaxTokens : null;
|
||||
return {
|
||||
modelId: model.id,
|
||||
durationMs:
|
||||
typeof metadata.durationMs === "number" ? metadata.durationMs : MOCK_LOAD_TEST_DURATION_MS,
|
||||
intervalMs:
|
||||
typeof metadata.intervalMs === "number" ? metadata.intervalMs : MOCK_LOAD_TEST_INTERVAL_MS,
|
||||
burst:
|
||||
burstMinTokens !== null && burstMaxTokens !== null
|
||||
? {
|
||||
minTokens: Math.max(1, burstMinTokens),
|
||||
maxTokens: Math.max(Math.max(1, burstMinTokens), burstMaxTokens),
|
||||
gapMs: typeof metadata.burstGapMs === "number" ? metadata.burstGapMs : 90,
|
||||
}
|
||||
: null,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -575,6 +626,12 @@ function buildCycleQueue(turnId: string, cycle: number): CycleEvent[] {
|
||||
return queue;
|
||||
}
|
||||
|
||||
function buildBurstyStreamQueue(cycle: number): CycleEvent[] {
|
||||
return tokenize(
|
||||
[buildIntroParagraph(cycle), buildMidParagraph(), buildClosingParagraph()].join("\n\n"),
|
||||
).map((text) => ({ kind: "assistant_token", text }));
|
||||
}
|
||||
|
||||
function createToolCall(input: {
|
||||
callId: string;
|
||||
name: string;
|
||||
@@ -755,6 +812,8 @@ export class MockLoadTestAgentSession implements AgentSession {
|
||||
queue: [],
|
||||
emittedTokens: 0,
|
||||
turnStarted: false,
|
||||
burst: profile.burst,
|
||||
burstIndex: 0,
|
||||
};
|
||||
this.activeTurn = turn;
|
||||
const largePayload = parseLargeAgentStreamPayloadPrompt(prompt);
|
||||
@@ -1425,17 +1484,24 @@ export class MockLoadTestAgentSession implements AgentSession {
|
||||
return;
|
||||
}
|
||||
|
||||
if (turn.queue.length === 0) {
|
||||
turn.cycle += 1;
|
||||
turn.queue = buildCycleQueue(turn.turnId, turn.cycle);
|
||||
}
|
||||
const eventsThisTick = turn.burst ? nextBurstSize(turn.burst, turn.burstIndex) : 1;
|
||||
turn.burstIndex += 1;
|
||||
|
||||
const event = turn.queue.shift();
|
||||
if (event) {
|
||||
for (let emitted = 0; emitted < eventsThisTick; emitted += 1) {
|
||||
if (turn.queue.length === 0) {
|
||||
turn.cycle += 1;
|
||||
turn.queue = turn.burst
|
||||
? buildBurstyStreamQueue(turn.cycle)
|
||||
: buildCycleQueue(turn.turnId, turn.cycle);
|
||||
}
|
||||
const event = turn.queue.shift();
|
||||
if (!event) {
|
||||
break;
|
||||
}
|
||||
this.dispatchCycleEvent(turn, event);
|
||||
}
|
||||
|
||||
this.schedule(turn, turn.intervalMs);
|
||||
this.schedule(turn, turn.burst ? turn.burst.gapMs : turn.intervalMs);
|
||||
}
|
||||
|
||||
private dispatchCycleEvent(turn: ActiveTurn, event: CycleEvent): void {
|
||||
|
||||
Reference in New Issue
Block a user