mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-04 07:33:18 +00:00
* feat(plugins): stream events and bytes between plugins and the app
Plugins can declare a subscription with defineSubscription, handle it with
server.handle(contract, (input, { push, signal })), and listen from a surface
with useSubscription or client.subscribe. A top-level Uint8Array in an event,
an RPC output, or an RPC input crosses the wire as a binary frame (opcodes
0x20-0x2f) without JSON or base64; a nested one fails validation on the sender.
Each subscriber runs its own handler, owned by the session that subscribed,
so release, disconnect, plugin stop and reload need no plugin code. The
runtime buffers pushes until the subscribe response has gone out, so a client
never sees an event for a subscription it does not know yet.
Gated on server_info.features.pluginSubscriptions; binary outputs reach only
clients that advertise plugin_binary_payloads.
* fix(plugins): close subscription gaps found in review
- Losing daemon.manage now releases the session's plugin subscriptions;
binary events bypass the outbound permission check, so streams would
otherwise outlive the revocation.
- A subscribe that times out sends unsubscribe, since the plugin may
still start the handler after the daemon gave up.
- The byte limit leaves room for the largest frame header and routing,
and lives in assertPluginWireValue so events, outputs and inputs share it.
- A release right after the host accepted keeps `ready` resolved, so a
handler that returns at once reports closed rather than an error.
- Session trace logging tolerates loggers without isLevelEnabled.
* fix(plugins): withhold binary plugin results after plugin access is revoked
A binary RPC output or event skipped the outbound permission check its
JSON form passes, so an RPC in flight when daemon.manage was revoked still
delivered its bytes. Each frame now checks the permission of the JSON
message it stands in for.
* test(plugins): cover subscriptions in a real browser instead of JSDOM
The app subscription tests mounted hooks in JSDOM, which docs/testing.md
rules out. The adapter and contract behavior is now a plain unit test,
and a Playwright spec drives the diff-watch and media-stream examples:
JSON events following file changes, streamed bytes, and an upload.
* test(plugins): add the plain subscription unit test and the browser spec
* fix(plugins): apply subscription schema transforms once
Both ends parsed each subscription value, and the sender sent its parse
output, so the receiver parsed a schema's output as fresh input. An event
schema like z.string().transform((s) => s.length) closed the stream as
invalid. The sender now validates and sends the value as given; the
receiver's parse produces what the handler or callback sees.
* fix(plugins): parse a subscription's input once in the plugin process
server.handle wrapped subscription handlers in a second input parse, so a
transforming input schema rejected its own parsed output before the
handler ran.
789 lines
29 KiB
TypeScript
789 lines
29 KiB
TypeScript
import { PluginTranslationRuntime } from "@getpaseo/plugin/client/host";
|
|
import { runPluginClientBundle, type PluginClientRuntime } from "@/plugins/evaluate";
|
|
import { transformTimelineItem, type TimelineItemTransform } from "@/plugins/timeline/model";
|
|
import type { InstalledPlugin } from "@/plugins/types";
|
|
import {
|
|
applyStreamEvent,
|
|
getStreamItemMessageId,
|
|
hydrateStreamState,
|
|
type StreamItem,
|
|
type ToolCallItem,
|
|
type UserMessageItem,
|
|
} from "@/types/stream";
|
|
import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages";
|
|
import { QueryClient } from "@tanstack/react-query";
|
|
import { describe, expect, it } from "vitest";
|
|
import { buildAgentStreamRenderModel } from "./model";
|
|
import { createStreamPresentation } from "./presentation";
|
|
import { createReadingAnchor } from "./reading-anchor";
|
|
|
|
const runtime = {
|
|
paseo: {},
|
|
async rpc() {},
|
|
openSettings() {},
|
|
openSurface() {},
|
|
openPanel() {},
|
|
addHeaderButton() {
|
|
return { update() {}, remove() {} };
|
|
},
|
|
addComposerPill() {
|
|
return { update() {}, remove() {} };
|
|
},
|
|
} as unknown as PluginClientRuntime;
|
|
|
|
function installProbe(
|
|
itemType: "user_message" | "assistant_message" | "tool_call" | "reasoning",
|
|
condition = "true",
|
|
): InstalledPlugin {
|
|
const clientBundle = `(function() {
|
|
return { default: function(client) {
|
|
client.addTimelineTransformer({
|
|
id: "probe",
|
|
query: { itemType: ${JSON.stringify(itemType)} },
|
|
transform: function({ item, phase }) {
|
|
if (!(${condition})) return undefined;
|
|
return {
|
|
items: [{
|
|
type: "plugin",
|
|
kind: "probe",
|
|
version: 1,
|
|
data: item.type === "tool_call"
|
|
? { callId: item.callId, name: item.name, status: item.status,
|
|
detail: item.detail, ...(item.metadata ? { metadata: item.metadata } : {}), error: item.error, phase: phase }
|
|
: { text: item.text, phase: phase }
|
|
}]
|
|
};
|
|
}
|
|
});
|
|
return function() {};
|
|
} };
|
|
})`;
|
|
const evaluated = runPluginClientBundle("probe-plugin", clientBundle, runtime);
|
|
return {
|
|
...evaluated,
|
|
serverId: "host-1",
|
|
clientBundle,
|
|
lifetime: new AbortController(),
|
|
locale: "en",
|
|
i18n: new PluginTranslationRuntime({ getLocale: () => "en", subscribe: () => () => {} }),
|
|
queryClient: new QueryClient(),
|
|
paseo: {} as InstalledPlugin["paseo"],
|
|
invoke: async () => undefined,
|
|
subscribe: () => {
|
|
throw new Error("Subscriptions are not used here");
|
|
},
|
|
};
|
|
}
|
|
|
|
function installedTransform(plugin: InstalledPlugin): TimelineItemTransform {
|
|
return (input) => transformTimelineItem({ ...input, plugins: [plugin] });
|
|
}
|
|
|
|
function toolCall(callId: string, name: string): ToolCallItem {
|
|
return {
|
|
kind: "tool_call",
|
|
id: `tool-${callId}`,
|
|
timestamp: new Date("2026-01-01T00:00:00.000Z"),
|
|
payload: {
|
|
source: "agent",
|
|
data: {
|
|
provider: "codex",
|
|
callId,
|
|
name,
|
|
status: "completed",
|
|
error: null,
|
|
detail: { type: "unknown", input: { callId }, output: { callId } },
|
|
},
|
|
},
|
|
};
|
|
}
|
|
|
|
function pluginData(items: StreamItem[]): unknown[] {
|
|
return items.flatMap((item) => (item.kind === "plugin" ? [item.data] : []));
|
|
}
|
|
|
|
const presentationOptions = {
|
|
display: "collapseToolCalls" as const,
|
|
expandedTurnHostIds: new Set<string>(),
|
|
isTurnActive: true,
|
|
};
|
|
|
|
function streamHarness(transform?: TimelineItemTransform) {
|
|
let state: { tail: StreamItem[]; head: StreamItem[] } = { tail: [], head: [] };
|
|
const present = createStreamPresentation();
|
|
return {
|
|
send(event: AgentStreamEventPayload) {
|
|
state = applyStreamEvent({ ...state, event, timestamp: new Date(1000) });
|
|
return this.render();
|
|
},
|
|
render(nextTransform = transform) {
|
|
return present({ ...presentationOptions, ...state, transform: nextTransform });
|
|
},
|
|
source: () => state,
|
|
};
|
|
}
|
|
|
|
function assistant(text: string, messageId = "message-1"): AgentStreamEventPayload {
|
|
return {
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: { type: "assistant_message", messageId, text },
|
|
};
|
|
}
|
|
|
|
function rows(result: { tail: StreamItem[]; head: StreamItem[] }): StreamItem[] {
|
|
return [...result.tail, ...result.head];
|
|
}
|
|
|
|
describe("whole-message Markdown presentation", () => {
|
|
it("keeps paragraphs, fences and quotes in one progressively updated row", () => {
|
|
const present = createStreamPresentation();
|
|
const text = "First paragraph.\n\nSecond paragraph.\n\n```ts\nconst n = 1;\n```\n\n> Quote";
|
|
for (const length of [10, 30, text.length]) {
|
|
const source = hydrateStreamState([
|
|
{ event: assistant(text.slice(0, length)), timestamp: new Date(1000) },
|
|
]);
|
|
const result = present({
|
|
...presentationOptions,
|
|
tail: [],
|
|
head: source,
|
|
transform: undefined,
|
|
});
|
|
expect(rows(result)).toEqual(source);
|
|
}
|
|
});
|
|
});
|
|
|
|
describe("stream presentation through installed plugins", () => {
|
|
it("offers every source tool call to an installed transformer in Overview mode", () => {
|
|
const calls = [toolCall("call-1", "bash"), toolCall("call-2", "read")];
|
|
const rendered = createStreamPresentation()({
|
|
...presentationOptions,
|
|
tail: calls,
|
|
head: [],
|
|
transform: installedTransform(installProbe("tool_call")),
|
|
});
|
|
expect(pluginData([...rendered.tail, ...rendered.head])).toMatchObject([
|
|
{ callId: "call-1", name: "bash", phase: "complete" },
|
|
{ callId: "call-2", name: "read", phase: "complete" },
|
|
]);
|
|
});
|
|
|
|
it.each([
|
|
{ type: "turn_completed", provider: "claude" },
|
|
{ type: "turn_canceled", provider: "claude", reason: "test cancellation" },
|
|
{ type: "turn_failed", provider: "claude", error: "test failure" },
|
|
] satisfies AgentStreamEventPayload[])(
|
|
"streams accumulated assistant text with one identity through $type",
|
|
(completion) => {
|
|
const harness = streamHarness(installedTransform(installProbe("assistant_message")));
|
|
const first = rows(harness.send(assistant("Header")));
|
|
const fullText = "Header\n\n\n# Body\n\n```ts\nconst x = 1;\n\nconst y = 2;\n```\n";
|
|
const live = rows(harness.send(assistant(fullText.slice("Header".length))));
|
|
const completed = rows(harness.send(completion));
|
|
|
|
expect(pluginData(first)).toEqual([{ text: "Header", phase: "streaming" }]);
|
|
expect(pluginData(live)).toEqual([{ text: fullText, phase: "streaming" }]);
|
|
expect(pluginData(completed)).toEqual([{ text: fullText, phase: "complete" }]);
|
|
expect(live.map((item) => item.id)).toEqual(first.map((item) => item.id));
|
|
expect(completed.map((item) => item.id)).toEqual(first.map((item) => item.id));
|
|
expect(rows(harness.source())).toMatchObject([{ kind: "assistant_message", text: fullText }]);
|
|
},
|
|
);
|
|
|
|
it("addresses a prompt a plugin replaced by the prompt's own id", () => {
|
|
const harness = streamHarness(installedTransform(installProbe("user_message")));
|
|
const rendered = rows(
|
|
harness.send({
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: { type: "user_message", messageId: "prompt-1", text: "Earlier prompt" },
|
|
}),
|
|
);
|
|
const source = rows(harness.source());
|
|
|
|
expect(source).toMatchObject([{ kind: "user_message", id: "prompt-1" }]);
|
|
expect(rendered).toMatchObject([{ kind: "plugin", id: "probe-plugin/prompt-1/0" }]);
|
|
// The chat outline jumps to the source id; the row it lands on is the plugin's.
|
|
expect(rendered.map(getStreamItemMessageId)).toEqual(["prompt-1"]);
|
|
});
|
|
|
|
it("finishes one message before streaming a different message ID", () => {
|
|
const harness = streamHarness(installedTransform(installProbe("assistant_message")));
|
|
harness.send(assistant("First\n\nBody", "first"));
|
|
const result = harness.send(assistant("Second", "second"));
|
|
expect(pluginData(rows(result))).toEqual([
|
|
{ text: "First\n\nBody", phase: "complete" },
|
|
{ text: "Second", phase: "streaming" },
|
|
]);
|
|
expect(new Set(rows(result).map((item) => item.id)).size).toBe(2);
|
|
});
|
|
|
|
it("lets a plugin claim the accumulated message after earlier text used native Markdown", () => {
|
|
const harness = streamHarness();
|
|
harness.send(assistant("Header\n\nBody"));
|
|
const transform = installedTransform(
|
|
installProbe("assistant_message", 'item.text.includes("Body")'),
|
|
);
|
|
expect(pluginData(rows(harness.render(transform)))).toEqual([
|
|
{ text: "Header\n\nBody", phase: "streaming" },
|
|
]);
|
|
expect(rows(harness.render(transform))).toHaveLength(1);
|
|
expect(rows(harness.render()).map((item) => item.kind)).toEqual(["assistant_message"]);
|
|
});
|
|
|
|
it("retains the complete message when streaming finishes", () => {
|
|
const harness = streamHarness();
|
|
harness.send(assistant("Intro\n\n"));
|
|
const last = harness.send(assistant("\n\nAfter\n- item"));
|
|
const completed = harness.send({ type: "turn_completed", provider: "claude" });
|
|
expect(last.tail).toEqual([]);
|
|
expect(last.head).toMatchObject([{ text: "Intro\n\n\n\nAfter\n- item" }]);
|
|
expect(completed.tail).toEqual(last.head);
|
|
expect(completed.head).toEqual([]);
|
|
});
|
|
|
|
it("keeps blank lines inside an open native code fence", () => {
|
|
const harness = streamHarness();
|
|
harness.send(assistant("Intro\n\n```ts\nconst a = 1;"));
|
|
const result = harness.send(assistant("\n\nconst b = 2;"));
|
|
expect(result.tail).toEqual([]);
|
|
expect(result.head).toMatchObject([{ text: "Intro\n\n```ts\nconst a = 1;\n\nconst b = 2;" }]);
|
|
});
|
|
|
|
// Codex streams a line break and the next line's indent as their own chunks.
|
|
it("keeps a line break when a chunk ends in the next line's indent", () => {
|
|
const harness = streamHarness();
|
|
harness.send(assistant("```mermaid\nflowchart TD\n A[Start] --> B[Review]\n"));
|
|
harness.send(assistant(" "));
|
|
const result = harness.send(assistant(" B --> C[Finish]\n```"));
|
|
expect(result.head).toMatchObject([
|
|
{ text: "```mermaid\nflowchart TD\n A[Start] --> B[Review]\n B --> C[Finish]\n```" },
|
|
]);
|
|
});
|
|
|
|
// One rendering path: a fetched message is the same block group as a streamed one,
|
|
// which is what lets find, scroll-to-message and history reveal address it by id.
|
|
// A link reference definition stays with the paragraph that uses it, so the split
|
|
// never leaves an empty row behind and the reference still resolves.
|
|
it("preserves source identity through history and the live head", () => {
|
|
const source = hydrateStreamState([
|
|
{
|
|
event: assistant("[Link][docs]\n\n[docs]: https://example.com\n\nClosing paragraph."),
|
|
timestamp: new Date(1000),
|
|
},
|
|
]);
|
|
const present = (placement: "tail" | "head") =>
|
|
rows(
|
|
createStreamPresentation()({
|
|
...presentationOptions,
|
|
tail: placement === "tail" ? source : [],
|
|
head: placement === "head" ? source : [],
|
|
transform: undefined,
|
|
}),
|
|
);
|
|
expect(present("tail")).toEqual(source);
|
|
const blocks = (items: StreamItem[]) =>
|
|
items.map((item) => [item.id, item.kind === "assistant_message" ? item.text : null]);
|
|
expect(blocks(present("head"))).toEqual(blocks(present("tail")));
|
|
});
|
|
|
|
// Rows are addressed by id from outside presentation, so the same text has to land on
|
|
// the same ids whether the reader watched it arrive or opened the chat afterwards.
|
|
it("gives a streamed message the same id as its hydrated text", () => {
|
|
const text = "Intro paragraph.\n\n# Heading\n\n- first\n- second\n\nClosing paragraph.";
|
|
const harness = streamHarness();
|
|
for (let end = 8; end < text.length; end += 8) {
|
|
harness.send(assistant(text.slice(end - 8, end)));
|
|
}
|
|
harness.send(assistant(text.slice(text.length - (text.length % 8 || 8))));
|
|
const streamed = rows(harness.send({ type: "turn_completed", provider: "claude" }));
|
|
const hydrated = rows(
|
|
createStreamPresentation()({
|
|
...presentationOptions,
|
|
tail: hydrateStreamState([{ event: assistant(text), timestamp: new Date(1000) }]),
|
|
head: [],
|
|
transform: undefined,
|
|
}),
|
|
);
|
|
expect(streamed.map((item) => item.id)).toEqual(hydrated.map((item) => item.id));
|
|
expect(streamed.map((item) => item.kind === "assistant_message" && item.text)).toEqual(
|
|
hydrated.map((item) => item.kind === "assistant_message" && item.text),
|
|
);
|
|
});
|
|
|
|
it("uses the canonical source that replaced the streamed message", () => {
|
|
const harness = streamHarness();
|
|
harness.send(assistant("Streamed answer"));
|
|
const streamed = rows(harness.render());
|
|
expect(streamed[0]).not.toHaveProperty("timelineCursor");
|
|
const source = harness.source();
|
|
const live = source.head[0];
|
|
if (live?.kind !== "assistant_message") throw new Error("Expected a live assistant message");
|
|
const canonical = { ...live, timelineCursor: { epoch: "epoch-1", seq: 42 } };
|
|
const present = createStreamPresentation();
|
|
present({ ...presentationOptions, tail: [], head: [live], transform: undefined });
|
|
const repointed = rows(
|
|
present({ ...presentationOptions, tail: [], head: [canonical], transform: undefined }),
|
|
);
|
|
expect(repointed).toMatchObject([
|
|
{
|
|
id: streamed[0]!.id,
|
|
text: "Streamed answer",
|
|
timelineCursor: { epoch: "epoch-1", seq: 42 },
|
|
},
|
|
]);
|
|
});
|
|
|
|
it("keeps a single-paragraph message a one-block group", () => {
|
|
const source = hydrateStreamState([
|
|
{ event: assistant("Only one paragraph."), timestamp: new Date(1000) },
|
|
]);
|
|
const messageId = source[0]!.id;
|
|
const result = createStreamPresentation()({
|
|
...presentationOptions,
|
|
tail: source,
|
|
head: [],
|
|
transform: undefined,
|
|
});
|
|
expect(result.tail).toEqual(source);
|
|
expect(result.tail[0]!.id).toBe(messageId);
|
|
});
|
|
|
|
it("continues to stream inline reasoning with a stable plugin row", () => {
|
|
const harness = streamHarness(installedTransform(installProbe("reasoning")));
|
|
const first = harness.send({
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: { type: "reasoning", text: "First" },
|
|
});
|
|
const second = harness.send({
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: { type: "reasoning", text: "\n\nSecond" },
|
|
});
|
|
const done = harness.send({ type: "turn_completed", provider: "claude" });
|
|
expect(pluginData(rows(second))).toEqual([{ text: "First\n\nSecond", phase: "streaming" }]);
|
|
expect(pluginData(rows(done))).toEqual([{ text: "First\n\nSecond", phase: "complete" }]);
|
|
expect(rows(first)[0]?.id).toBe(rows(done)[0]?.id);
|
|
});
|
|
|
|
it("groups only unclaimed tools and passes unchanged tool details to the callback", () => {
|
|
const calls = [
|
|
toolCall("call-1", "read"),
|
|
toolCall("call-2", "read"),
|
|
toolCall("call-3", "bash"),
|
|
];
|
|
const result = createStreamPresentation()({
|
|
...presentationOptions,
|
|
isTurnActive: false,
|
|
tail: calls,
|
|
head: [],
|
|
transform: installedTransform(installProbe("tool_call", 'item.name === "bash"')),
|
|
});
|
|
expect(result.groupsByHostId.get(result.tail[0]!.id)?.run.calls).toEqual(calls.slice(0, 2));
|
|
expect(pluginData(rows(result))).toEqual([
|
|
{
|
|
callId: "call-3",
|
|
name: "bash",
|
|
status: "completed",
|
|
error: null,
|
|
detail: { type: "unknown", input: { callId: "call-3" }, output: { callId: "call-3" } },
|
|
phase: "complete",
|
|
},
|
|
]);
|
|
});
|
|
|
|
it.each([
|
|
{ status: "completed", error: null },
|
|
{ status: "failed", error: { message: "Command failed" } },
|
|
{ status: "canceled", error: null },
|
|
] as const)("keeps a tool card's identity and full payload through $status", (completion) => {
|
|
const harness = streamHarness(installedTransform(installProbe("tool_call")));
|
|
const source = {
|
|
type: "tool_call" as const,
|
|
callId: "call-1",
|
|
name: "bash",
|
|
detail: { type: "shell" as const, command: "pwd", cwd: "/repo" },
|
|
metadata: { sequence: 1 },
|
|
};
|
|
const running = rows(
|
|
harness.send({
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: {
|
|
...source,
|
|
status: "running",
|
|
error: null,
|
|
},
|
|
}),
|
|
);
|
|
const terminalItem = { ...source, ...completion };
|
|
const terminal = rows(
|
|
harness.send({
|
|
type: "timeline",
|
|
provider: "claude",
|
|
item: {
|
|
...terminalItem,
|
|
},
|
|
}),
|
|
);
|
|
expect(pluginData(running)).toMatchObject([
|
|
{ callId: "call-1", status: "running", phase: "streaming" },
|
|
]);
|
|
expect(pluginData(terminal)).toEqual([
|
|
{
|
|
callId: source.callId,
|
|
name: source.name,
|
|
detail: source.detail,
|
|
metadata: source.metadata,
|
|
phase: "complete",
|
|
status: completion.status,
|
|
error: terminalItem.error,
|
|
},
|
|
]);
|
|
expect(terminal.map((item) => item.id)).toEqual(running.map((item) => item.id));
|
|
});
|
|
|
|
it("offers one complete live source assistant message to an installed transformer", () => {
|
|
const sourceText =
|
|
'[IMPORTANT: User invoked the "diagnose" skill; follow its instructions. Full skill below.]\n\n# Diagnose\n\nFind the root cause.';
|
|
const event = {
|
|
type: "timeline" as const,
|
|
provider: "opencode" as const,
|
|
item: { type: "assistant_message" as const, messageId: "message-1", text: sourceText },
|
|
};
|
|
const timestamp = new Date("2026-01-01T00:00:00.000Z");
|
|
const transform = installedTransform(installProbe("assistant_message"));
|
|
const present = createStreamPresentation();
|
|
|
|
const fetched = hydrateStreamState([{ event, timestamp }]);
|
|
const history = present({ ...presentationOptions, tail: fetched, head: [], transform });
|
|
expect(pluginData(history.tail)).toEqual([{ text: sourceText, phase: "complete" }]);
|
|
|
|
const stream = applyStreamEvent({
|
|
tail: [],
|
|
head: [],
|
|
event,
|
|
timestamp,
|
|
});
|
|
|
|
const rendered = present({ ...presentationOptions, ...stream, transform });
|
|
expect(pluginData([...rendered.tail, ...rendered.head])).toEqual([
|
|
{ text: sourceText, phase: "streaming" },
|
|
]);
|
|
});
|
|
});
|
|
|
|
function createTimestamp(seed: number): Date {
|
|
return new Date(`2026-01-01T00:00:${seed.toString().padStart(2, "0")}.000Z`);
|
|
}
|
|
|
|
function userMessage(id: string, seed: number): UserMessageItem {
|
|
return {
|
|
kind: "user_message",
|
|
id,
|
|
text: id,
|
|
timestamp: createTimestamp(seed),
|
|
};
|
|
}
|
|
|
|
function assistantMessage(
|
|
id: string,
|
|
seed: number,
|
|
): Extract<StreamItem, { kind: "assistant_message" }> {
|
|
return {
|
|
kind: "assistant_message",
|
|
id,
|
|
text: id,
|
|
timestamp: createTimestamp(seed),
|
|
};
|
|
}
|
|
|
|
describe("timeline presentation", () => {
|
|
const present = createStreamPresentation();
|
|
function projectTimelineItems(items: StreamItem[], transform?: TimelineItemTransform) {
|
|
return present({ ...presentationOptions, tail: items, head: [], transform }).tail;
|
|
}
|
|
const envelope =
|
|
"<spoken-input>\nPlease fix the voice chat.\n</spoken-input>\n<instruction>This message was spoken by the user. Respond using the speak tool only, not normal messages, because the user may not be looking at the chat.</instruction>";
|
|
|
|
it.each(["live", "history"])(
|
|
"shows only spoken words from a %s user message without mutating its source",
|
|
(source) => {
|
|
const item: UserMessageItem = Object.freeze({
|
|
...userMessage(source, 1),
|
|
text: envelope,
|
|
messageId: "provider-message",
|
|
clientMessageId: "client-message",
|
|
turnId: "turn-1",
|
|
timelineCursor: { epoch: "epoch", seq: 12 },
|
|
});
|
|
const presentMessage = () =>
|
|
present({
|
|
...presentationOptions,
|
|
tail: source === "history" ? [item] : [],
|
|
head: source === "live" ? [item] : [],
|
|
transform: undefined,
|
|
});
|
|
const projected = rows(presentMessage());
|
|
expect(projected).toEqual([{ ...item, text: "Please fix the voice chat." }]);
|
|
const model = buildAgentStreamRenderModel({
|
|
tail: source === "history" ? projected : [],
|
|
head: source === "live" ? projected : [],
|
|
isTurnActive: source === "live",
|
|
activeTurnStartedAt: item.timestamp,
|
|
platform: "native",
|
|
isMobileBreakpoint: true,
|
|
});
|
|
const rendered = [...model.history, ...model.segments.liveHead];
|
|
expect(rendered).toContainEqual({ ...item, text: "Please fix the voice chat." });
|
|
expect(item.text).toBe(envelope);
|
|
expect(rows(presentMessage())[0]).toBe(projected[0]);
|
|
},
|
|
);
|
|
|
|
it("supports older envelopes and preserves multiline spoken content", () => {
|
|
const item = {
|
|
...userMessage("legacy", 1),
|
|
text: "<spoken-input>\nFirst line.\nSecond line with <example>XML</example>.\n</spoken-input>",
|
|
};
|
|
expect(projectTimelineItems([item])).toEqual([
|
|
{ ...item, text: "First line.\nSecond line with <example>XML</example>." },
|
|
]);
|
|
});
|
|
|
|
it("leaves ordinary messages, assistant examples and incomplete wrappers unchanged", () => {
|
|
const items: StreamItem[] = [
|
|
userMessage("ordinary", 1),
|
|
{ ...assistantMessage("example", 2), text: envelope },
|
|
{ ...userMessage("incomplete", 3), text: "<spoken-input>unfinished" },
|
|
{ ...userMessage("quoted", 4), text: "Explain this example: " + envelope },
|
|
{
|
|
...userMessage("xml", 5),
|
|
text: "<spoken-input>Example</spoken-input><instruction>Explain this XML.</instruction>",
|
|
},
|
|
];
|
|
expect(projectTimelineItems(items)).toEqual(items);
|
|
expect(projectTimelineItems(items)[0]).toBe(items[0]);
|
|
});
|
|
|
|
it("keeps plugin transforms on the original source and respects replacements", () => {
|
|
const source: StreamItem = { ...userMessage("spoken", 1), text: envelope };
|
|
const inputs: string[] = [];
|
|
const transformed = projectTimelineItems([source], ({ item, sourceId }) => {
|
|
if (item.type === "user_message") inputs.push(item.text);
|
|
return [
|
|
{
|
|
type: "plugin",
|
|
pluginId: "test",
|
|
id: sourceId,
|
|
kind: "voice",
|
|
version: 1,
|
|
data: { text: "Custom voice row" },
|
|
},
|
|
];
|
|
});
|
|
expect(inputs).toEqual([envelope]);
|
|
expect(transformed).toMatchObject([{ kind: "plugin", data: { text: "Custom voice row" } }]);
|
|
expect(projectTimelineItems([source], () => undefined)).toEqual([
|
|
{ ...source, text: "Please fix the voice chat." },
|
|
]);
|
|
expect(projectTimelineItems([source], () => [])).toEqual([]);
|
|
});
|
|
});
|
|
|
|
it("opening a turn reveals individual calls rather than another tool group", () => {
|
|
const present = createStreamPresentation();
|
|
const timestamp = new Date();
|
|
const tail: StreamItem[] = [
|
|
{ kind: "user_message", id: "prompt", text: "Work", timestamp },
|
|
toolCall("one", "Bash"),
|
|
toolCall("two", "Bash"),
|
|
{ kind: "assistant_message", id: "reply", text: "Done", timestamp },
|
|
];
|
|
const result = present({
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail,
|
|
head: [],
|
|
display: "collapseTurns",
|
|
isTurnActive: false,
|
|
expandedTurnHostIds: new Set(["prompt"]),
|
|
});
|
|
expect(result.tail.map((item) => item.id)).toEqual(tail.map((item) => item.id));
|
|
expect(result.groupsByHostId.size).toBe(0);
|
|
expect(result.readingRowReplacements.get("tool-two")).toBeUndefined();
|
|
});
|
|
|
|
it.each(["collapseToolCalls", "collapseTurns"] as const)(
|
|
"preserves the reader when a prepend extends a group in %s",
|
|
(display) => {
|
|
const present = createStreamPresentation();
|
|
const anchor = createReadingAnchor();
|
|
const calls = [toolCall("one", "Bash"), toolCall("two", "Bash")];
|
|
const input = {
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
display,
|
|
head: [],
|
|
isTurnActive: false,
|
|
};
|
|
const partial = present({ ...input, tail: calls.slice(1) });
|
|
anchor.reconcile(0, [{ id: partial.tail[0]!.id, top: 48, height: 200 }]);
|
|
const completed = present({ ...input, tail: calls });
|
|
const groupId = completed.tail[0]!.id;
|
|
expect(anchor.getRowId(completed.readingRowReplacements)).toBe(groupId);
|
|
expect(
|
|
anchor.reconcile(
|
|
0,
|
|
[{ id: groupId, top: 448, height: 200 }],
|
|
false,
|
|
completed.readingRowReplacements,
|
|
),
|
|
).toBe(400);
|
|
expect(anchor.getRowId()).toBe(groupId);
|
|
const timestamp = new Date();
|
|
const folded = present({
|
|
...input,
|
|
display: "collapseTurns",
|
|
tail: [
|
|
{ kind: "user_message", id: "prompt", text: "Work", timestamp },
|
|
...calls,
|
|
{ kind: "assistant_message", id: "reply", text: "Done", timestamp },
|
|
],
|
|
});
|
|
// Both the original partial group and its replacement resolve directly to the turn.
|
|
expect(folded.readingRowReplacements.get(calls[1]!.id)).toBe("prompt");
|
|
expect(anchor.getRowId(folded.readingRowReplacements)).toBe("prompt");
|
|
const opened = present({ ...input, tail: calls, display: "full" });
|
|
expect(opened.readingRowReplacements.get(calls[1]!.id)).toBeUndefined();
|
|
},
|
|
);
|
|
|
|
it("presents received agent messages as attributed bubbles with the original turn identity", () => {
|
|
const call = toolCall("received", "agent_message");
|
|
if (call.payload.source !== "agent") throw new Error("Expected agent tool");
|
|
call.turnId = "turn";
|
|
call.payload.data.agentMessage = {
|
|
event: "message",
|
|
sender: { id: "sender", title: "Reviewer" },
|
|
text: "Ready",
|
|
};
|
|
const result = createStreamPresentation()({
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail: [call],
|
|
head: [],
|
|
});
|
|
expect(result.tail[0]).toMatchObject({
|
|
kind: "user_message",
|
|
id: call.id,
|
|
turnId: "turn",
|
|
text: "Ready",
|
|
sender: { id: "sender", title: "Reviewer" },
|
|
});
|
|
});
|
|
|
|
it("does not reopen the previous turn while a new run waits for its first row", () => {
|
|
const timestamp = new Date();
|
|
const tail: StreamItem[] = [
|
|
{ kind: "user_message", id: "prompt", text: "Work", timestamp },
|
|
toolCall("one", "Bash"),
|
|
{ kind: "assistant_message", id: "reply", text: "Done", timestamp },
|
|
];
|
|
for (const item of tail) item.turnId = "completed";
|
|
const result = createStreamPresentation()({
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail,
|
|
head: [],
|
|
display: "collapseTurns",
|
|
isTurnActive: true,
|
|
activeTurnId: "new",
|
|
});
|
|
expect(result.tail.map((item) => item.id)).toEqual(["prompt", "reply"]);
|
|
});
|
|
|
|
// A sent prompt counts as activity before the daemon opens its turn, so the only known turn is
|
|
// the completed one; it must stay folded until the new turn's rows arrive.
|
|
it("keeps the previous turn folded while a sent prompt waits for its turn to open", () => {
|
|
const timestamp = new Date();
|
|
const tail: StreamItem[] = [
|
|
{ kind: "user_message", id: "prompt", text: "Work", timestamp },
|
|
toolCall("one", "Bash"),
|
|
{ kind: "assistant_message", id: "reply", text: "Done", timestamp },
|
|
];
|
|
for (const item of tail) item.turnId = "completed";
|
|
const result = createStreamPresentation()({
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail,
|
|
head: [],
|
|
display: "collapseTurns",
|
|
isTurnActive: true,
|
|
activeTurnId: null,
|
|
});
|
|
expect(result.tail.map((item) => item.id)).toEqual(["prompt", "reply"]);
|
|
});
|
|
|
|
it("groups reasoning with tools unless Expand thinking is enabled, including setting changes", () => {
|
|
const present = createStreamPresentation();
|
|
const timestamp = new Date();
|
|
const thought: StreamItem = {
|
|
kind: "thought",
|
|
id: "thought",
|
|
text: "Consider this",
|
|
status: "ready",
|
|
timestamp,
|
|
};
|
|
const tail: StreamItem[] = [toolCall("one", "Bash"), thought, toolCall("two", "Bash")];
|
|
for (const item of tail) item.turnId = "run";
|
|
const input = {
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail,
|
|
head: [],
|
|
display: "collapseToolCalls" as const,
|
|
isTurnActive: false,
|
|
};
|
|
const grouped = present({ ...input, autoExpandReasoning: false });
|
|
expect(grouped.tail).toHaveLength(1);
|
|
expect(grouped.groupsByHostId.get("tool-one")?.run.calls).toEqual(tail);
|
|
const separate = present({ ...input, autoExpandReasoning: true });
|
|
expect(separate.tail.map((item) => item.id)).toEqual(["tool-one", "thought", "tool-two"]);
|
|
expect(separate.groupsByHostId.has("thought")).toBe(false);
|
|
const next = { ...thought, id: "next", turnId: "next-run" };
|
|
const bounded = present({ ...input, tail: [...tail, next], autoExpandReasoning: false });
|
|
expect(bounded.tail.map((item) => item.id)).toEqual(["tool-one", "next"]);
|
|
});
|
|
|
|
it.each([false, true])(
|
|
"reveals raw reasoning when opening a turn (Expand thinking %s)",
|
|
(autoExpandReasoning) => {
|
|
const timestamp = new Date();
|
|
const tail: StreamItem[] = [
|
|
{ kind: "user_message", id: "prompt", text: "Work", timestamp },
|
|
{ kind: "thought", id: "thought", text: "Consider this", status: "ready", timestamp },
|
|
toolCall("one", "Bash"),
|
|
{ kind: "assistant_message", id: "reply", text: "Done", timestamp },
|
|
];
|
|
const input = {
|
|
...presentationOptions,
|
|
transform: undefined,
|
|
tail,
|
|
head: [],
|
|
display: "collapseTurns" as const,
|
|
isTurnActive: false,
|
|
autoExpandReasoning,
|
|
};
|
|
const present = createStreamPresentation();
|
|
const closed = present(input);
|
|
expect(closed.tail.some((item) => item.id === "thought")).toBe(false);
|
|
const opened = present({ ...input, expandedTurnHostIds: new Set(["prompt"]) });
|
|
expect(opened.tail).toEqual(tail);
|
|
expect(opened.turns.nestedItemIds.has("thought")).toBe(true);
|
|
expect(opened.groupsByHostId.size).toBe(0);
|
|
},
|
|
);
|