refactor(app): remove legacy stream buffer adapter
Removed the deprecated and compatibility API from . Runtime app code already uses the canonical head/tail reducer, so this dead path added maintenance overhead and duplicated behavior.\n\nReplaced the mixed compatibility test suite with a focused that verifies the canonical flow (head accumulation, flush on non-streamable events, completion flushing, and no-op reference stability).\n\nVerification:\n- npm run test --workspace=@getpaseo/app -- src/types/stream.test.ts src/types/stream-event.test.ts src/types/stream.harness.test.ts\n- npm run typecheck\n\nFor next agent:\n- Accomplished: removed legacy stream buffer migration shim and aligned tests to the canonical model only.\n- Challenge/Watchout: if any external consumer still imports removed symbols, it will now fail at compile-time; current repo has no remaining references.
This commit is contained in:
Mohamed Boudra
+21
-123
@@ -1,11 +1,7 @@
|
|||||||
import { describe, expect, it } from "vitest";
|
import { describe, expect, it } from "vitest";
|
||||||
import type { AgentStreamEventPayload } from "@server/shared/messages";
|
import type { AgentStreamEventPayload } from "@server/shared/messages";
|
||||||
import type { StreamItem, ThoughtItem } from "@/types/stream";
|
import type { ThoughtItem } from "@/types/stream";
|
||||||
import {
|
import { applyStreamEvent } from "@/types/stream";
|
||||||
applyStreamEvent,
|
|
||||||
applyStreamEventWithBuffer,
|
|
||||||
type StreamingBufferEntry,
|
|
||||||
} from "@/types/stream";
|
|
||||||
|
|
||||||
const baseTimestamp = new Date(0);
|
const baseTimestamp = new Date(0);
|
||||||
|
|
||||||
@@ -23,7 +19,7 @@ const toolCallEvent = (): AgentStreamEventPayload => ({
|
|||||||
provider: "codex",
|
provider: "codex",
|
||||||
item: {
|
item: {
|
||||||
type: "tool_call",
|
type: "tool_call",
|
||||||
callId: "buffer-tool-call",
|
callId: "head-tail-tool-call",
|
||||||
name: "run",
|
name: "run",
|
||||||
status: "running",
|
status: "running",
|
||||||
detail: {
|
detail: {
|
||||||
@@ -60,122 +56,7 @@ const reasoningChunk = (text: string): AgentStreamEventPayload => ({
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
describe("applyStreamEventWithBuffer", () => {
|
describe("applyStreamEvent", () => {
|
||||||
it("buffers assistant chunks without changing stream", () => {
|
|
||||||
const stream: StreamItem[] = [];
|
|
||||||
const result = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: null,
|
|
||||||
event: assistantChunk("Hel"),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(result.stream).toBe(stream);
|
|
||||||
expect(result.changedStream).toBe(false);
|
|
||||||
expect(result.buffer?.text).toBe("Hel");
|
|
||||||
});
|
|
||||||
|
|
||||||
it("appends assistant chunks to the buffer", () => {
|
|
||||||
const stream: StreamItem[] = [];
|
|
||||||
const initial = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: null,
|
|
||||||
event: assistantChunk("Hel"),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
const next = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: initial.buffer,
|
|
||||||
event: assistantChunk("lo"),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(next.buffer?.text).toBe("Hello");
|
|
||||||
expect(next.changedStream).toBe(false);
|
|
||||||
});
|
|
||||||
|
|
||||||
it("commits buffered message on completion", () => {
|
|
||||||
const stream: StreamItem[] = [];
|
|
||||||
const buffered = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: null,
|
|
||||||
event: assistantChunk("Hello"),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
const result = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: buffered.buffer,
|
|
||||||
event: completionEvent(),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(result.buffer).toBe(null);
|
|
||||||
expect(result.stream).toHaveLength(1);
|
|
||||||
expect(result.stream[0].kind).toBe("assistant_message");
|
|
||||||
expect((result.stream[0] as { text: string }).text).toBe("Hello");
|
|
||||||
});
|
|
||||||
|
|
||||||
it("commits buffer before non-assistant timeline items", () => {
|
|
||||||
const stream: StreamItem[] = [];
|
|
||||||
const buffered = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: null,
|
|
||||||
event: assistantChunk("Hello"),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
const result = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: buffered.buffer,
|
|
||||||
event: toolCallEvent(),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(result.buffer).toBe(null);
|
|
||||||
expect(result.stream).toHaveLength(2);
|
|
||||||
expect(result.stream[0].kind).toBe("assistant_message");
|
|
||||||
expect(result.stream[1].kind).toBe("tool_call");
|
|
||||||
});
|
|
||||||
|
|
||||||
it("keeps stream reference for no-op events", () => {
|
|
||||||
const stream: StreamItem[] = [];
|
|
||||||
const result = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer: null,
|
|
||||||
event: permissionEvent(),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(result.stream).toBe(stream);
|
|
||||||
expect(result.changedStream).toBe(false);
|
|
||||||
});
|
|
||||||
|
|
||||||
it("avoids double-committing an already flushed buffer", () => {
|
|
||||||
const stream: StreamItem[] = [
|
|
||||||
{
|
|
||||||
kind: "assistant_message",
|
|
||||||
id: "dup",
|
|
||||||
text: "Hello",
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
},
|
|
||||||
];
|
|
||||||
const buffer: StreamingBufferEntry = {
|
|
||||||
id: "dup",
|
|
||||||
text: "Hello",
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
};
|
|
||||||
const result = applyStreamEventWithBuffer({
|
|
||||||
state: stream,
|
|
||||||
buffer,
|
|
||||||
event: completionEvent(),
|
|
||||||
timestamp: baseTimestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(result.stream).toBe(stream);
|
|
||||||
expect(result.buffer).toBe(null);
|
|
||||||
});
|
|
||||||
});
|
|
||||||
|
|
||||||
describe("applyStreamEvent (head/tail model)", () => {
|
|
||||||
it("buffers reasoning chunks in head", () => {
|
it("buffers reasoning chunks in head", () => {
|
||||||
const result = applyStreamEvent({
|
const result = applyStreamEvent({
|
||||||
tail: [],
|
tail: [],
|
||||||
@@ -270,4 +151,21 @@ describe("applyStreamEvent (head/tail model)", () => {
|
|||||||
expect(result.head).toHaveLength(1);
|
expect(result.head).toHaveLength(1);
|
||||||
expect(result.head[0].kind).toBe("assistant_message");
|
expect(result.head[0].kind).toBe("assistant_message");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("keeps references stable for no-op events", () => {
|
||||||
|
const tail: ReturnType<typeof applyStreamEvent>["tail"] = [];
|
||||||
|
const head: ReturnType<typeof applyStreamEvent>["head"] = [];
|
||||||
|
|
||||||
|
const result = applyStreamEvent({
|
||||||
|
tail,
|
||||||
|
head,
|
||||||
|
event: permissionEvent(),
|
||||||
|
timestamp: baseTimestamp,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(result.tail).toBe(tail);
|
||||||
|
expect(result.head).toBe(head);
|
||||||
|
expect(result.changedTail).toBe(false);
|
||||||
|
expect(result.changedHead).toBe(false);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
@@ -899,56 +899,3 @@ export function applyStreamEvent(params: {
|
|||||||
|
|
||||||
return { tail: nextTail, head: nextHead, changedTail, changedHead };
|
return { tail: nextTail, head: nextHead, changedTail, changedHead };
|
||||||
}
|
}
|
||||||
|
|
||||||
// Legacy export for backwards compatibility during migration
|
|
||||||
// TODO: Remove after all consumers are updated
|
|
||||||
export type StreamingBufferEntry = {
|
|
||||||
id: string;
|
|
||||||
text: string;
|
|
||||||
timestamp: Date;
|
|
||||||
};
|
|
||||||
|
|
||||||
export function applyStreamEventWithBuffer(params: {
|
|
||||||
state: StreamItem[];
|
|
||||||
buffer: StreamingBufferEntry | null;
|
|
||||||
event: AgentStreamEventPayload;
|
|
||||||
timestamp: Date;
|
|
||||||
}): {
|
|
||||||
stream: StreamItem[];
|
|
||||||
buffer: StreamingBufferEntry | null;
|
|
||||||
changedStream: boolean;
|
|
||||||
changedBuffer: boolean;
|
|
||||||
} {
|
|
||||||
// Convert legacy buffer to head
|
|
||||||
const head: StreamItem[] = params.buffer
|
|
||||||
? [
|
|
||||||
{
|
|
||||||
kind: "assistant_message",
|
|
||||||
id: params.buffer.id,
|
|
||||||
text: params.buffer.text,
|
|
||||||
timestamp: params.buffer.timestamp,
|
|
||||||
},
|
|
||||||
]
|
|
||||||
: [];
|
|
||||||
|
|
||||||
const result = applyStreamEvent({
|
|
||||||
tail: params.state,
|
|
||||||
head,
|
|
||||||
event: params.event,
|
|
||||||
timestamp: params.timestamp,
|
|
||||||
});
|
|
||||||
|
|
||||||
// Convert head back to legacy buffer format
|
|
||||||
const lastHead = result.head[result.head.length - 1];
|
|
||||||
const newBuffer: StreamingBufferEntry | null =
|
|
||||||
lastHead && lastHead.kind === "assistant_message"
|
|
||||||
? { id: lastHead.id, text: lastHead.text, timestamp: lastHead.timestamp }
|
|
||||||
: null;
|
|
||||||
|
|
||||||
return {
|
|
||||||
stream: result.tail,
|
|
||||||
buffer: newBuffer,
|
|
||||||
changedStream: result.changedTail,
|
|
||||||
changedBuffer: result.changedHead,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user