mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 21:23:04 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
41d9bb4294 | ||
|
|
b044201f62 | ||
|
|
cacf186a94 | ||
|
|
47b18ffb50 | ||
|
|
ab55377458 | ||
|
|
f3d8718c6a | ||
|
|
26db1a4672 | ||
|
|
f28f0fee3d | ||
|
|
98f667f93e | ||
|
|
8c2a2a9a88 | ||
|
|
a4f696620b | ||
|
|
184ab66cde |
@@ -52,6 +52,33 @@ On capable daemons, connecting creates no timeline or event demand. Client subsc
|
||||
release it on unsubscribe, and restore it after reconnect. Raw message observers inspect traffic
|
||||
without requesting streams. Application caches and which agents are visible remain caller-owned.
|
||||
|
||||
## Binary chunking
|
||||
|
||||
`binary_chunking_v1` in hello and optional `server_info.features.binaryChunkingV1`
|
||||
negotiate connection-scoped framing. Application serializers and routed envelopes use
|
||||
the same boundary. Small messages and handshakes keep their existing representation.
|
||||
Older peers retain existing frame limits.
|
||||
|
||||
The reserved binary opcode `0x31` is followed by version `1`, original kind
|
||||
(`0` text, `1` binary), and START/END flag bits `0`/`1`. Remaining bytes are an
|
||||
unchanged slice. There is no total length or chunk count; a future receiver can
|
||||
consume slices incrementally. Today the receiver buffers up to the existing
|
||||
100 MiB logical limit and decodes UTF-8 only after END. Transfers cannot interleave.
|
||||
Disconnect clears partial state and negotiation.
|
||||
|
||||
The lower transport supplies its plaintext frame budget, including encryption
|
||||
overhead. The sender waits for drain between chunks while ordinary messages keep
|
||||
the existing delivery path. Slow-client protections count messages queued behind
|
||||
an active transfer as well as bytes already in the socket.
|
||||
The framing module never selects an encryption construction.
|
||||
|
||||
Native uploads retain the platform's existing queue limits. React Native's
|
||||
WebSocket does not report its queued bytes, so the client cannot wait for native
|
||||
drain. Android's installed OkHttp transport closes when a send would put more
|
||||
than 16 MiB in its queue. The native transport disables outbound chunking, including
|
||||
on routed channels, and retains its existing single-frame admission errors.
|
||||
Inbound chunking remains enabled; daemon-to-native responses use the daemon's paced writes.
|
||||
|
||||
## Owned observations
|
||||
|
||||
`owned_subscriptions` and `server_info.features.ownedSubscriptions` negotiate the source-owned
|
||||
|
||||
+10
-4
@@ -21,10 +21,16 @@ The invariants are:
|
||||
> agent establishes the daemon's current tail in one complete-turn request, with older history reachable
|
||||
> through backward pagination.
|
||||
|
||||
Tool output is bounded before it enters either delivery path. Canonical shell tool output is sliced
|
||||
to 64 KiB, and the same bounded item is used for runtime timeline rows and live stream events.
|
||||
Provider history hydration applies the same rule so reopening an agent cannot restore an oversized
|
||||
tool payload.
|
||||
Every item is bounded before it enters either delivery path, because a timeline page is one
|
||||
WebSocket message and the daemon closes a socket whose outbound buffer passes 64 MiB. Tool calls get
|
||||
64 KiB and other items 256 KiB. Over budget, the largest strings are cut to a shared length with a
|
||||
`…[truncated 12.3 MB]` marker, and an oversized `data:` URL is replaced by the marker. Strings up
|
||||
to 1 KiB stay whole, so ids and paths survive. Arrays drop their trailing elements without a marker
|
||||
element, so typed arrays still match the wire schema. Limiting an item twice gives the same item. The
|
||||
budget does not bound wide objects or multi-byte text. The same
|
||||
bounded item is used for runtime timeline rows and live stream events. Provider history hydration
|
||||
applies the same rule so reopening an agent cannot restore an oversized payload. Fetches apply it
|
||||
again to assistant and reasoning rows the projection merges from many chunks.
|
||||
|
||||
## Presence is not delivery
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ function visibleMainPane(page: Page) {
|
||||
}
|
||||
|
||||
function composerChangesPill(page: Page) {
|
||||
return page.getByTestId("composer-diff-stat-pill");
|
||||
return page.getByRole("button", { name: "Open Changes tab", exact: true });
|
||||
}
|
||||
|
||||
async function revealComposerChangesInExplorer(page: Page) {
|
||||
@@ -498,6 +498,7 @@ test.describe("send review feedback", () => {
|
||||
test("only open tabs in this workspace appear in compact and wide recipient menus", async ({
|
||||
page,
|
||||
}) => {
|
||||
test.setTimeout(120000);
|
||||
const workspace = await seedChangedAgent("review-feedback-multiple-");
|
||||
const other = await seedChangedAgent("review-feedback-other-");
|
||||
try {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { chooseModel, startWithoutRememberedModel } from "../support/helpers/model-memory";
|
||||
import {
|
||||
composerLocator,
|
||||
enterComposerFullscreen,
|
||||
@@ -97,6 +98,7 @@ test.describe("New workspace composer draft", () => {
|
||||
const project = await seedWorkspace({ repoPrefix: "new-workspace-submitted-draft-" });
|
||||
|
||||
try {
|
||||
await startWithoutRememberedModel(page);
|
||||
await gotoAppShell(page);
|
||||
await waitForSidebarHydration(page);
|
||||
await openNewWorkspaceComposer(page, {
|
||||
@@ -104,6 +106,7 @@ test.describe("New workspace composer draft", () => {
|
||||
projectDisplayName: project.projectDisplayName,
|
||||
});
|
||||
|
||||
await chooseModel(page, "mock", "Ten second stream");
|
||||
const composer = page.getByRole("textbox", { name: "Message agent..." });
|
||||
const createButton = page.getByTestId("workspace-create-submit");
|
||||
await expect(composer).toBeEditable({ timeout: 30_000 });
|
||||
|
||||
@@ -64,7 +64,7 @@ function resolveShouldFill(
|
||||
): boolean {
|
||||
if (!fillAvailableHeight) return false;
|
||||
const t = detail?.type;
|
||||
return t === "shell" || t === "edit" || t === "write" || t === "read" || t === "sub_agent";
|
||||
return t === "shell" || t === "edit" || t === "write" || t === "read";
|
||||
}
|
||||
|
||||
function useDetailStyles(
|
||||
@@ -186,184 +186,6 @@ function ShellDetailSection({ command, output, ds }: ShellDetailProps) {
|
||||
);
|
||||
}
|
||||
|
||||
function resolveSubAgentFallbackHeader(
|
||||
subAgentType: string | null | undefined,
|
||||
description: string | null | undefined,
|
||||
fallbackText: string,
|
||||
): string {
|
||||
if (subAgentType && description) {
|
||||
return `${subAgentType}: ${description}`;
|
||||
}
|
||||
return subAgentType ?? description ?? fallbackText;
|
||||
}
|
||||
|
||||
interface SubAgentDetailProps {
|
||||
log: string;
|
||||
childSessionId: string | null | undefined;
|
||||
subAgentType: string | null | undefined;
|
||||
description: string | null | undefined;
|
||||
ds: DetailStyles;
|
||||
}
|
||||
|
||||
interface SubAgentActivityRow {
|
||||
index: number;
|
||||
toolName: string;
|
||||
summary?: string;
|
||||
}
|
||||
|
||||
interface ParsedSubAgentLog {
|
||||
actions: SubAgentActivityRow[];
|
||||
remainingLog: string;
|
||||
}
|
||||
|
||||
function parseBracketedSubAgentLine(line: string, index: number): SubAgentActivityRow | null {
|
||||
const match = line.match(/^\[([^\]]+)\](?:\s+(.*))?$/);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const toolName = match[1]?.trim();
|
||||
if (!toolName) {
|
||||
return null;
|
||||
}
|
||||
const summary = match[2]?.trim();
|
||||
return {
|
||||
index,
|
||||
toolName,
|
||||
...(summary ? { summary } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function parseSubAgentLog(log: string): ParsedSubAgentLog {
|
||||
const actions: SubAgentActivityRow[] = [];
|
||||
const remainingLines: string[] = [];
|
||||
for (const line of log.replace(/^\n+/, "").split("\n")) {
|
||||
const normalizedLine = line.trim();
|
||||
if (!normalizedLine) {
|
||||
continue;
|
||||
}
|
||||
const parsedAction = parseBracketedSubAgentLine(normalizedLine, actions.length + 1);
|
||||
if (parsedAction) {
|
||||
actions.push(parsedAction);
|
||||
} else {
|
||||
remainingLines.push(line);
|
||||
}
|
||||
}
|
||||
return {
|
||||
actions,
|
||||
remainingLog: remainingLines.join("\n").replace(/^\n+/, ""),
|
||||
};
|
||||
}
|
||||
|
||||
function SubAgentActionRow({ action }: { action: SubAgentActivityRow }) {
|
||||
return (
|
||||
<View style={styles.subAgentActionRow}>
|
||||
<Text selectable style={styles.subAgentActionTool}>
|
||||
{formatSubAgentToolName(action.toolName)}
|
||||
</Text>
|
||||
{action.summary ? (
|
||||
<Text selectable style={styles.subAgentActionSummary}>
|
||||
{action.summary}
|
||||
</Text>
|
||||
) : null}
|
||||
</View>
|
||||
);
|
||||
}
|
||||
|
||||
function formatSubAgentToolName(toolName: string): string {
|
||||
const trimmed = toolName.trim();
|
||||
if (!trimmed) {
|
||||
return toolName;
|
||||
}
|
||||
return trimmed
|
||||
.replace(/[._-]+/g, " ")
|
||||
.split(" ")
|
||||
.filter((segment) => segment.length > 0)
|
||||
.map((segment) => `${segment[0]?.toUpperCase() ?? ""}${segment.slice(1)}`)
|
||||
.join(" ");
|
||||
}
|
||||
|
||||
function SubAgentLogText({
|
||||
activityLog,
|
||||
fallbackHeader,
|
||||
hasActions,
|
||||
}: {
|
||||
activityLog: string;
|
||||
fallbackHeader: string;
|
||||
hasActions: boolean;
|
||||
}) {
|
||||
if (activityLog.length > 0) {
|
||||
return (
|
||||
<Text selectable style={styles.scrollText}>
|
||||
{activityLog}
|
||||
</Text>
|
||||
);
|
||||
}
|
||||
if (!hasActions) {
|
||||
return (
|
||||
<Text selectable style={styles.scrollText}>
|
||||
{fallbackHeader}
|
||||
</Text>
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function SubAgentDetailSection({
|
||||
log,
|
||||
childSessionId,
|
||||
subAgentType,
|
||||
description,
|
||||
ds,
|
||||
}: SubAgentDetailProps) {
|
||||
const { t } = useTranslation();
|
||||
const { actions, remainingLog } = useMemo(() => parseSubAgentLog(log), [log]);
|
||||
const fallbackHeader = resolveSubAgentFallbackHeader(
|
||||
subAgentType,
|
||||
description,
|
||||
t("toolCallDetails.subAgentActivity"),
|
||||
);
|
||||
const hasActions = actions.length > 0;
|
||||
return (
|
||||
<View style={ds.sectionFillStyle}>
|
||||
<View style={ds.codeBlockFillStyle}>
|
||||
<ScrollView
|
||||
style={ds.codeVerticalScrollStyle}
|
||||
contentContainerStyle={styles.codeVerticalContent}
|
||||
nestedScrollEnabled
|
||||
showsVerticalScrollIndicator
|
||||
>
|
||||
<ScrollView
|
||||
horizontal
|
||||
nestedScrollEnabled
|
||||
showsHorizontalScrollIndicator
|
||||
contentContainerStyle={styles.codeHorizontalContent}
|
||||
>
|
||||
<View style={styles.codeLine} dataSet={CODE_SURFACE_DATASET}>
|
||||
{childSessionId ? (
|
||||
<Text selectable style={styles.subAgentSessionText}>
|
||||
session {childSessionId}
|
||||
</Text>
|
||||
) : null}
|
||||
{hasActions ? (
|
||||
<View style={styles.subAgentActions}>
|
||||
{actions.map((action) => (
|
||||
<SubAgentActionRow key={action.index} action={action} />
|
||||
))}
|
||||
</View>
|
||||
) : null}
|
||||
<SubAgentLogText
|
||||
activityLog={remainingLog}
|
||||
fallbackHeader={fallbackHeader}
|
||||
hasActions={hasActions}
|
||||
/>
|
||||
</View>
|
||||
</ScrollView>
|
||||
</ScrollView>
|
||||
</View>
|
||||
</View>
|
||||
);
|
||||
}
|
||||
|
||||
interface EditDetailProps {
|
||||
diffLines: DiffLine[] | undefined;
|
||||
ds: DetailStyles;
|
||||
@@ -636,19 +458,7 @@ function buildDetailSections(
|
||||
<ShellDetailSection key="shell" command={detail.command} output={detail.output} ds={ds} />,
|
||||
];
|
||||
}
|
||||
if (detail.type === "worktree_setup") return [];
|
||||
if (detail.type === "sub_agent") {
|
||||
return [
|
||||
<SubAgentDetailSection
|
||||
key="sub-agent"
|
||||
log={detail.log}
|
||||
childSessionId={detail.childSessionId}
|
||||
subAgentType={detail.subAgentType}
|
||||
description={detail.description}
|
||||
ds={ds}
|
||||
/>,
|
||||
];
|
||||
}
|
||||
if (detail.type === "worktree_setup" || detail.type === "sub_agent") return [];
|
||||
if (detail.type === "edit") {
|
||||
return [<EditDetailSection key="edit" diffLines={diffLines} ds={ds} />];
|
||||
}
|
||||
@@ -900,34 +710,6 @@ const styles = StyleSheet.create((theme) => {
|
||||
shellPrompt: {
|
||||
color: theme.colors.foregroundMuted,
|
||||
},
|
||||
subAgentSessionText: {
|
||||
fontFamily: theme.fontFamily.mono,
|
||||
fontSize: theme.fontSize.code,
|
||||
color: theme.colors.foregroundMuted,
|
||||
lineHeight: 18,
|
||||
marginBottom: theme.spacing[2],
|
||||
},
|
||||
subAgentActions: {
|
||||
gap: theme.spacing[1],
|
||||
marginBottom: theme.spacing[2],
|
||||
},
|
||||
subAgentActionRow: {
|
||||
flexDirection: "row",
|
||||
alignItems: "center",
|
||||
gap: theme.spacing[2],
|
||||
},
|
||||
subAgentActionTool: {
|
||||
fontFamily: theme.fontFamily.mono,
|
||||
fontSize: theme.fontSize.code,
|
||||
color: theme.colors.foregroundMuted,
|
||||
lineHeight: 18,
|
||||
},
|
||||
subAgentActionSummary: {
|
||||
fontFamily: theme.fontFamily.mono,
|
||||
fontSize: theme.fontSize.code,
|
||||
color: theme.colors.foreground,
|
||||
lineHeight: 18,
|
||||
},
|
||||
jsonScroll: {
|
||||
borderWidth: theme.borderWidth[1],
|
||||
borderColor: theme.colors.border,
|
||||
|
||||
@@ -832,7 +832,6 @@ describe("translation resources", () => {
|
||||
expect(en.common.placeholders.search).toBe("Search...");
|
||||
expect(en.common.empty.noResults).toBe("No results found");
|
||||
expect(en.common.empty.noOptionsMatchSearch).toBe("No options match your search.");
|
||||
expect(en.toolCallDetails.subAgentActivity).toBe("Sub-agent activity");
|
||||
expect(en.panels.file.failedToLoadPreview).toBe("Failed to load file preview");
|
||||
});
|
||||
|
||||
|
||||
@@ -2013,7 +2013,6 @@ export const ar: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "خطأ",
|
||||
empty: "لا توجد تفاصيل إضافية متاحة",
|
||||
subAgentActivity: "نشاط الوكيل الفرعي",
|
||||
input: "مدخل",
|
||||
output: "الإخراج",
|
||||
},
|
||||
|
||||
@@ -2038,7 +2038,6 @@ export const en = {
|
||||
toolCallDetails: {
|
||||
error: "Error",
|
||||
empty: "No additional details available",
|
||||
subAgentActivity: "Sub-agent activity",
|
||||
input: "Input",
|
||||
output: "Output",
|
||||
},
|
||||
|
||||
@@ -2061,7 +2061,6 @@ export const es: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "Error",
|
||||
empty: "No hay detalles adicionales disponibles",
|
||||
subAgentActivity: "Actividad de subagente",
|
||||
input: "Aporte",
|
||||
output: "Producción",
|
||||
},
|
||||
|
||||
@@ -2060,7 +2060,6 @@ export const fr: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "Erreur",
|
||||
empty: "Aucun détail supplémentaire disponible",
|
||||
subAgentActivity: "Activité du sous-agent",
|
||||
input: "Entrée",
|
||||
output: "Sortie",
|
||||
},
|
||||
|
||||
@@ -2033,7 +2033,6 @@ export const ja: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "エラー",
|
||||
empty: "追加の詳細はありません",
|
||||
subAgentActivity: "サブエージェントアクティビティ",
|
||||
input: "入力",
|
||||
output: "出力",
|
||||
},
|
||||
|
||||
@@ -2023,7 +2023,6 @@ export const ko: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "오류",
|
||||
empty: "추가 세부 정보가 없습니다",
|
||||
subAgentActivity: "서브에이전트 활동",
|
||||
input: "입력",
|
||||
output: "출력",
|
||||
},
|
||||
|
||||
@@ -2046,7 +2046,6 @@ export const ptBR: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "Erro",
|
||||
empty: "Nenhum detalhe adicional disponível",
|
||||
subAgentActivity: "Atividade do subagente",
|
||||
input: "Entrada",
|
||||
output: "Saída",
|
||||
},
|
||||
|
||||
@@ -2046,7 +2046,6 @@ export const ru: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "Ошибка",
|
||||
empty: "Дополнительные сведения отсутствуют",
|
||||
subAgentActivity: "Активность субагента",
|
||||
input: "Входные данные",
|
||||
output: "Выходные данные",
|
||||
},
|
||||
|
||||
@@ -1991,7 +1991,6 @@ export const zhCN: TranslationResources = {
|
||||
toolCallDetails: {
|
||||
error: "错误",
|
||||
empty: "没有可用的更多详情",
|
||||
subAgentActivity: "Sub-agent 活动",
|
||||
input: "输入",
|
||||
output: "输出",
|
||||
},
|
||||
|
||||
@@ -76,6 +76,25 @@ describe("test-daemon-connection connectToDaemon", () => {
|
||||
probe = new FakeDaemonProbe();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ id: "direct:lan:6767", type: "directTcp" as const, endpoint: "lan:6767" },
|
||||
{
|
||||
id: "relay:test",
|
||||
type: "relay" as const,
|
||||
relayEndpoint: "relay.paseo.sh:443",
|
||||
daemonPublicKeyB64: "pubkey",
|
||||
},
|
||||
])("adoptable $type probe clients use the app WebSocket factory", async (connection) => {
|
||||
const { connectToDaemon } = await import("./test-daemon-connection");
|
||||
const { createAppWebSocketFactory } = await import("@/runtime/websocket-factory");
|
||||
const result = await connectToDaemon(connection, { serverId: "srv_probe_test" }, probe.deps);
|
||||
try {
|
||||
expect(result.client.config.webSocketFactory).toBe(createAppWebSocketFactory());
|
||||
} finally {
|
||||
await result.client.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("reuses the app clientId for direct connections", async () => {
|
||||
const { connectToDaemon } = await import("./test-daemon-connection");
|
||||
const first = await connectToDaemon(
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
} from "@getpaseo/client/internal/daemon-client";
|
||||
import type { DaemonClientConfig } from "@getpaseo/client/internal/daemon-client";
|
||||
import type { HostConnection } from "@/types/host-connection";
|
||||
import { createAppWebSocketFactory } from "@/runtime/websocket-factory";
|
||||
import { getOrCreateClientId } from "./client-id";
|
||||
import { resolveAppVersion } from "./app-version";
|
||||
import {
|
||||
@@ -156,6 +157,7 @@ export async function buildClientConfig(
|
||||
clientType: "mobile" as const,
|
||||
appVersion: deps.resolveAppVersion() ?? undefined,
|
||||
suppressSendErrors: true,
|
||||
webSocketFactory: createAppWebSocketFactory(),
|
||||
reconnect: { enabled: false },
|
||||
...resolveConnectionCredentials(connection, options),
|
||||
...(options.capabilities ? { capabilities: options.capabilities } : {}),
|
||||
|
||||
@@ -40,12 +40,6 @@ function hasMeaningfulFetchDetail(detail: Extract<ToolCallDetail, { type: "fetch
|
||||
return Boolean(detail.url || detail.result || detail.codeText);
|
||||
}
|
||||
|
||||
function hasMeaningfulSubAgentDetail(
|
||||
detail: Extract<ToolCallDetail, { type: "sub_agent" }>,
|
||||
): boolean {
|
||||
return Boolean(detail.subAgentType || detail.description || detail.log);
|
||||
}
|
||||
|
||||
function hasMeaningfulPlainTextDetail(
|
||||
detail: Extract<ToolCallDetail, { type: "plain_text" }>,
|
||||
): boolean {
|
||||
@@ -75,9 +69,9 @@ export function hasMeaningfulToolCallDetail(detail: ToolCallDetail | undefined):
|
||||
case "fetch":
|
||||
return hasMeaningfulFetchDetail(detail);
|
||||
case "worktree_setup":
|
||||
return false;
|
||||
// A sub-agent card is a launch marker; the child's activity lives in the sub-agent panel.
|
||||
case "sub_agent":
|
||||
return hasMeaningfulSubAgentDetail(detail);
|
||||
return false;
|
||||
case "plain_text":
|
||||
return hasMeaningfulPlainTextDetail(detail);
|
||||
case "plan":
|
||||
@@ -92,7 +86,7 @@ export function isPendingToolCallDetail(params: {
|
||||
status: "executing" | "running" | "completed" | "failed" | "canceled";
|
||||
error: unknown;
|
||||
}): boolean {
|
||||
if (params.detail?.type === "worktree_setup") return false;
|
||||
if (params.detail?.type === "worktree_setup" || params.detail?.type === "sub_agent") return false;
|
||||
const isRunning = params.status === "running" || params.status === "executing";
|
||||
return isRunning && params.error == null && !hasMeaningfulToolCallDetail(params.detail);
|
||||
}
|
||||
|
||||
@@ -125,6 +125,7 @@ export class ConnectionSubscriptions extends OwnedSubscriptions {
|
||||
// Protocol support belongs to the installed client. Only browser hosting needs
|
||||
// a resource supplied by the caller. Keep this exhaustive as the protocol evolves.
|
||||
export const DEFAULT_CLIENT_CAPABILITIES = {
|
||||
[CLIENT_CAPS.binaryChunkingV1]: true,
|
||||
[CLIENT_CAPS.helloRejection]: true,
|
||||
[CLIENT_CAPS.ownedSubscriptions]: true,
|
||||
[CLIENT_CAPS.allProviders]: true,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { maxSecuredPlaintextByteLength } from "@getpaseo/relay";
|
||||
import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit";
|
||||
import {
|
||||
openClientSecureChannel,
|
||||
@@ -155,18 +156,15 @@ export function createEncryptedTransport(
|
||||
const relayTransport: RelayTransport = {
|
||||
send: (data) => {
|
||||
if (typeof data === "string") {
|
||||
base.send(data);
|
||||
return;
|
||||
return base.send(data);
|
||||
}
|
||||
if (ArrayBuffer.isView(data)) {
|
||||
base.send(normalizeTransportPayload(data));
|
||||
return;
|
||||
return base.send(normalizeTransportPayload(data));
|
||||
}
|
||||
if (data instanceof ArrayBuffer) {
|
||||
base.send(data);
|
||||
return;
|
||||
return base.send(data);
|
||||
}
|
||||
base.send(String(data));
|
||||
return base.send(String(data));
|
||||
},
|
||||
// The channel decides the close code. Report it now: a socket close event
|
||||
// arrives asynchronously, and the consumer must not act on an earlier
|
||||
@@ -226,6 +224,8 @@ export function createEncryptedTransport(
|
||||
});
|
||||
|
||||
return {
|
||||
supportsDrain: base.supportsDrain,
|
||||
maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES),
|
||||
send: (data) => {
|
||||
if (!channel) {
|
||||
throw new Error("Encrypted channel not ready");
|
||||
@@ -239,8 +239,9 @@ export function createEncryptedTransport(
|
||||
// Encryption and its transport handoff are synchronous up to send's first await.
|
||||
// Reject capacity failures here so a routed sender loses only its own channel.
|
||||
assertOutboundCapacity(wireBytes);
|
||||
void channel.send(payload).catch((error) => {
|
||||
return channel.send(payload).catch((error) => {
|
||||
emitError(error);
|
||||
throw error;
|
||||
});
|
||||
},
|
||||
close: (code?: number, reason?: string) => {
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
export interface DaemonTransport {
|
||||
send: (data: string | Uint8Array | ArrayBuffer) => void;
|
||||
/** False when the underlying socket cannot report write drain. */
|
||||
supportsDrain?: boolean;
|
||||
maxFrameBytes?: number;
|
||||
send: (data: string | Uint8Array | ArrayBuffer) => void | Promise<void>;
|
||||
close: (code?: number, reason?: string) => void;
|
||||
onMessage: (handler: (data: unknown, isBinary: boolean) => void) => () => void;
|
||||
onOpen: (handler: () => void) => () => void;
|
||||
@@ -13,12 +16,16 @@ export type DaemonTransportFactory = (options: {
|
||||
protocols?: string[];
|
||||
}) => DaemonTransport;
|
||||
|
||||
export type WebSocketFactory = (
|
||||
url: string,
|
||||
options?: { headers?: Record<string, string>; protocols?: string[] },
|
||||
) => WebSocketLike;
|
||||
export interface WebSocketFactory {
|
||||
(
|
||||
url: string,
|
||||
options?: { headers?: Record<string, string>; protocols?: string[] },
|
||||
): WebSocketLike;
|
||||
supportsDrain?: boolean;
|
||||
}
|
||||
|
||||
export interface WebSocketLike {
|
||||
bufferedAmount?: number;
|
||||
readyState: number;
|
||||
send: (data: string | Uint8Array | ArrayBuffer) => void;
|
||||
close: (code?: number, reason?: string) => void;
|
||||
|
||||
@@ -175,3 +175,18 @@ describe("daemon-client transport helpers", () => {
|
||||
expect(decodeMessageData(bytes)).toBe("hello");
|
||||
});
|
||||
});
|
||||
|
||||
test("native drain capability survives WebSocket and encryption wrappers", async () => {
|
||||
const { nativeWebSocketFactory } = await import("./daemon-client-websocket-transport.js");
|
||||
expect(nativeWebSocketFactory.supportsDrain).toBe(false);
|
||||
const factory = Object.assign(() => ({ readyState: 1, send() {}, close() {} }), {
|
||||
supportsDrain: false,
|
||||
});
|
||||
const base = createWebSocketTransportFactory(factory)({ url: "ws://test" });
|
||||
expect(base.supportsDrain).toBe(false);
|
||||
const encrypted = createEncryptedTransport(base, {
|
||||
daemonPublicKeyB64: "unused-key",
|
||||
logger: { warn() {} },
|
||||
});
|
||||
expect(encrypted.supportsDrain).toBe(false);
|
||||
});
|
||||
|
||||
@@ -39,6 +39,9 @@ export function nativeWebSocketFactory(
|
||||
return new (getGlobalWebSocket())(url, options?.protocols, { headers: options?.headers });
|
||||
}
|
||||
|
||||
// React Native leaves bufferedAmount at zero, including while native output is queued.
|
||||
nativeWebSocketFactory.supportsDrain = false;
|
||||
|
||||
export function createWebSocketTransportFactory(factory: WebSocketFactory): DaemonTransportFactory {
|
||||
return ({ url, headers, protocols }) => {
|
||||
const ws = factory(url, { headers, protocols });
|
||||
@@ -50,11 +53,28 @@ export function createWebSocketTransportFactory(factory: WebSocketFactory): Daem
|
||||
}
|
||||
}
|
||||
return {
|
||||
supportsDrain: factory.supportsDrain !== false,
|
||||
send: (data) => {
|
||||
if (typeof ws.readyState === "number" && ws.readyState !== 1) {
|
||||
throw new Error(`WebSocket not open (readyState=${ws.readyState})`);
|
||||
}
|
||||
ws.send(data);
|
||||
if ((ws.bufferedAmount ?? 0) > 1024 * 1024) {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const drain = () => {
|
||||
if (ws.readyState !== 1) {
|
||||
reject(new Error("WebSocket closed while draining"));
|
||||
return;
|
||||
}
|
||||
if ((ws.bufferedAmount ?? 0) <= 1024 * 1024) {
|
||||
resolve();
|
||||
return;
|
||||
}
|
||||
setTimeout(drain, 10);
|
||||
};
|
||||
drain();
|
||||
});
|
||||
}
|
||||
},
|
||||
close: (code?: number, reason?: string) => {
|
||||
// Node's `ws` may emit an `error` when a connecting socket is closed before the
|
||||
|
||||
@@ -1094,6 +1094,7 @@ test("advertises client capabilities in hello", async () => {
|
||||
provider_snapshot_references: true,
|
||||
explicit_event_subscriptions: true,
|
||||
owned_subscriptions: true,
|
||||
binary_chunking_v1: true,
|
||||
compact_provider_snapshots: true,
|
||||
custom_mode_icons: true,
|
||||
project_updates: true,
|
||||
@@ -6817,91 +6818,102 @@ test("drops legacy agent_stream tool_call payloads and logs validation warning",
|
||||
expect(logger.warn).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("parses canonical fetch_agent_timeline_response payloads without crashing", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
test.each([false, true])(
|
||||
"parses canonical timeline responses with chunking=%s",
|
||||
async (chunked) => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen({ features: { ownedSubscriptions: true, binaryChunkingV1: chunked } });
|
||||
await connectPromise;
|
||||
|
||||
const received: unknown[] = [];
|
||||
const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => {
|
||||
received.push(msg);
|
||||
});
|
||||
const received: unknown[] = [];
|
||||
const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => {
|
||||
received.push(msg);
|
||||
});
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "fetch_agent_timeline_response",
|
||||
const command = chunked ? "x".repeat(50 * 1024 * 1024) : "pwd";
|
||||
const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport");
|
||||
const sender = new ChunkedTransport({
|
||||
frameBytes: 24 * 1024 * 1024,
|
||||
send: (frame) => mock.triggerMessage(frame),
|
||||
deliver: () => {},
|
||||
});
|
||||
sender.enabled = chunked;
|
||||
await sender.send(
|
||||
wrapSessionMessage({
|
||||
type: "fetch_agent_timeline_response",
|
||||
payload: {
|
||||
requestId: "req-1",
|
||||
agentId: "agent_cli",
|
||||
agent: null,
|
||||
direction: "tail",
|
||||
projection: "projected",
|
||||
epoch: "epoch-1",
|
||||
reset: false,
|
||||
staleCursor: false,
|
||||
gap: false,
|
||||
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
|
||||
startCursor: { epoch: "epoch-1", seq: 1 },
|
||||
endCursor: { epoch: "epoch-1", seq: 1 },
|
||||
hasOlder: false,
|
||||
hasNewer: false,
|
||||
entries: [
|
||||
{
|
||||
timestamp: "2026-02-08T20:20:00.000Z",
|
||||
provider: "codex",
|
||||
seqStart: 1,
|
||||
seqEnd: 1,
|
||||
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
|
||||
collapsed: [],
|
||||
item: {
|
||||
type: "tool_call",
|
||||
callId: "call_cli_snapshot",
|
||||
name: "shell",
|
||||
status: "running",
|
||||
detail: {
|
||||
type: "shell",
|
||||
command,
|
||||
},
|
||||
error: null,
|
||||
},
|
||||
},
|
||||
],
|
||||
error: null,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
unsubscribe();
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0]).toMatchObject({
|
||||
payload: {
|
||||
requestId: "req-1",
|
||||
agentId: "agent_cli",
|
||||
agent: null,
|
||||
direction: "tail",
|
||||
projection: "projected",
|
||||
epoch: "epoch-1",
|
||||
reset: false,
|
||||
staleCursor: false,
|
||||
gap: false,
|
||||
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
|
||||
startCursor: { epoch: "epoch-1", seq: 1 },
|
||||
endCursor: { epoch: "epoch-1", seq: 1 },
|
||||
hasOlder: false,
|
||||
hasNewer: false,
|
||||
entries: [
|
||||
{
|
||||
timestamp: "2026-02-08T20:20:00.000Z",
|
||||
provider: "codex",
|
||||
seqStart: 1,
|
||||
seqEnd: 1,
|
||||
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
|
||||
collapsed: [],
|
||||
item: {
|
||||
type: "tool_call",
|
||||
callId: "call_cli_snapshot",
|
||||
name: "shell",
|
||||
status: "running",
|
||||
detail: {
|
||||
type: "shell",
|
||||
command: "pwd",
|
||||
},
|
||||
error: null,
|
||||
detail: { type: "shell", command },
|
||||
},
|
||||
},
|
||||
],
|
||||
error: null,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
unsubscribe();
|
||||
|
||||
expect(received).toHaveLength(1);
|
||||
expect(received[0]).toMatchObject({
|
||||
payload: {
|
||||
entries: [
|
||||
{
|
||||
item: {
|
||||
type: "tool_call",
|
||||
status: "running",
|
||||
error: null,
|
||||
detail: { type: "shell" },
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
expect(logger.warn).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
expect(logger.warn).not.toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
|
||||
test("rejects and logs a correlated response that violates the protocol schema", async () => {
|
||||
const mock = createMockTransport();
|
||||
@@ -8471,3 +8483,302 @@ test("closes before admission when a protected-capable daemon is reached in lega
|
||||
expect(daemonReceived).toHaveLength(1);
|
||||
expect(JSON.parse(daemonReceived[0] as string)).toMatchObject({ type: "hello" });
|
||||
});
|
||||
|
||||
test("uses the frame budget selected by an asynchronously opened transport", async () => {
|
||||
const mock = createMockTransport();
|
||||
mock.transport.maxFrameBytes = 1024;
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "dynamic-budget",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.transport.maxFrameBytes = 16;
|
||||
mock.triggerOpen({ features: { binaryChunkingV1: true } });
|
||||
await connecting;
|
||||
client.sendHostFrame("route", new Uint8Array(100));
|
||||
await vi.waitFor(() => expect(mock.sent.length).toBeGreaterThan(1));
|
||||
expect(mock.sent.every((frame) => typeof frame !== "string" && frame.byteLength <= 16)).toBe(
|
||||
true,
|
||||
);
|
||||
});
|
||||
|
||||
test.each([false, true])("routed sends wait for origin drain (binary=%s)", async (binary) => {
|
||||
const mock = createMockTransport();
|
||||
let drain!: () => void;
|
||||
let blocked = false;
|
||||
const send = mock.transport.send;
|
||||
mock.transport.send = (frame) => {
|
||||
send(frame);
|
||||
if (blocked)
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "routed-drain",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen({ features: { binaryChunkingV1: true } });
|
||||
await connecting;
|
||||
blocked = true;
|
||||
let finished = false;
|
||||
const write = Promise.resolve(
|
||||
client.sendHostFrame("route", binary ? new Uint8Array([1]) : "hello"),
|
||||
).then(() => {
|
||||
finished = true;
|
||||
return undefined;
|
||||
});
|
||||
await Promise.resolve();
|
||||
expect(finished).toBe(false);
|
||||
drain();
|
||||
await write;
|
||||
expect(finished).toBe(true);
|
||||
});
|
||||
|
||||
test("a queued admission failure rejects its RPC without closing the client", async () => {
|
||||
const mock = createMockTransport();
|
||||
const send = mock.transport.send;
|
||||
let block = false;
|
||||
let drain!: () => void;
|
||||
mock.transport.send = (frame) => {
|
||||
if (!block) return send(frame);
|
||||
if (typeof frame !== "string")
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
throw new Error("capacity");
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "queued-admission",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connecting;
|
||||
block = true;
|
||||
const first = client.sendHostFrame("route", new Uint8Array([1]));
|
||||
const ping = client.ping({ timeoutMs: 1000 });
|
||||
void ping.catch(() => {});
|
||||
drain();
|
||||
await first;
|
||||
await expect(ping).rejects.toThrow("capacity");
|
||||
expect(client.isConnected).toBe(true);
|
||||
});
|
||||
|
||||
test("RPC deadlines remain active while a preceding frame waits for drain", async () => {
|
||||
const mock = createMockTransport();
|
||||
const send = mock.transport.send;
|
||||
let blocked = false;
|
||||
let drain!: () => void;
|
||||
mock.transport.send = (frame) => {
|
||||
if (!blocked) return send(frame);
|
||||
if (typeof frame !== "string")
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "drain-deadline",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connecting;
|
||||
blocked = true;
|
||||
const first = client.sendHostFrame("route", new Uint8Array([1]));
|
||||
try {
|
||||
await expect(client.ping({ timeoutMs: 20 })).rejects.toThrow("Timeout waiting for message");
|
||||
expect(client.isConnected).toBe(true);
|
||||
} finally {
|
||||
drain();
|
||||
await first;
|
||||
}
|
||||
});
|
||||
|
||||
test.each([false, true])(
|
||||
"event listener errors do not close the transport (chunked=%s)",
|
||||
async (chunked) => {
|
||||
const mock = createMockTransport();
|
||||
const closed = vi.fn();
|
||||
mock.transport.close = closed;
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "listener-error",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen({ features: { binaryChunkingV1: chunked } });
|
||||
await connecting;
|
||||
const unsubscribe = client.subscribe(() => {
|
||||
throw new Error("listener failed");
|
||||
});
|
||||
const text = wrapSessionMessage({
|
||||
type: "agent_deleted",
|
||||
payload: { agentId: "agent-test", requestId: "listener-test" },
|
||||
});
|
||||
const payload = new TextEncoder().encode(text);
|
||||
const frame = chunked ? new Uint8Array([0x31, 1, 0, 3, ...payload]) : text;
|
||||
expect(() => mock.triggerMessage(frame)).toThrow("listener failed");
|
||||
expect(closed).not.toHaveBeenCalled();
|
||||
unsubscribe();
|
||||
const events: string[] = [];
|
||||
client.subscribe((event) => {
|
||||
events.push(event.type);
|
||||
});
|
||||
mock.triggerMessage(frame);
|
||||
expect(events).toEqual(["agent_deleted"]);
|
||||
await client.sendHostFrame("route", "still connected");
|
||||
expect(client.isConnected).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
test("malformed chunk metadata still closes the client transport", async () => {
|
||||
const mock = createMockTransport();
|
||||
const closed = vi.fn();
|
||||
mock.transport.close = closed;
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "malformed-chunk",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen({ features: { binaryChunkingV1: true } });
|
||||
await connecting;
|
||||
mock.triggerMessage(new Uint8Array([0x31, 2, 0, 3, 65]));
|
||||
expect(closed).toHaveBeenCalledWith(4003, "Invalid chunked transport");
|
||||
});
|
||||
|
||||
test.each([FileTransferOpcode.FileBegin, FileTransferOpcode.FileChunk, FileTransferOpcode.FileEnd])(
|
||||
"upload stops at a queued pre-write rejection (opcode=%s)",
|
||||
async (opcode) => {
|
||||
const mock = createMockTransport();
|
||||
const send = mock.transport.send;
|
||||
let drain!: () => void;
|
||||
let blocked = false;
|
||||
const attempts: number[] = [];
|
||||
mock.transport.send = (frame) => {
|
||||
if (typeof frame !== "string") {
|
||||
const decoded = decodeFileTransferFrame(assertUint8Array(frame));
|
||||
if (decoded) {
|
||||
attempts.push(decoded.opcode);
|
||||
if (decoded.opcode === opcode) throw new Error("upload capacity");
|
||||
} else if (blocked)
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
}
|
||||
return send(frame);
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "upload-admission",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connecting;
|
||||
blocked = true;
|
||||
const first = client.sendHostFrame("route", new Uint8Array([1]));
|
||||
const upload = client.uploadFile({
|
||||
fileName: "file.bin",
|
||||
mimeType: "application/octet-stream",
|
||||
bytes: new Uint8Array([1, 2]),
|
||||
chunkSize: 1,
|
||||
});
|
||||
let rejected: unknown;
|
||||
void upload.catch((error) => {
|
||||
rejected = error;
|
||||
});
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
||||
drain();
|
||||
await first;
|
||||
await vi.waitFor(() => expect(rejected).toMatchObject({ message: "upload capacity" }), {
|
||||
timeout: 500,
|
||||
});
|
||||
expect(attempts.at(-1)).toBe(opcode);
|
||||
expect(client.isConnected).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
test.each(["latency", "dictation start", "dictation finish"])(
|
||||
"queued control send rejection reaches its caller (%s)",
|
||||
async (operation) => {
|
||||
const mock = createMockTransport();
|
||||
const send = mock.transport.send;
|
||||
let blocked = false;
|
||||
let drain!: () => void;
|
||||
mock.transport.send = (frame) => {
|
||||
if (!blocked) return send(frame);
|
||||
if (typeof frame !== "string")
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
throw new Error("control capacity");
|
||||
};
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "control-admission",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connecting;
|
||||
blocked = true;
|
||||
const first = client.sendHostFrame("route", new Uint8Array([1]));
|
||||
let result: Promise<unknown>;
|
||||
if (operation === "latency") result = client.measureLatency({ timeoutMs: 1000 });
|
||||
else if (operation === "dictation start")
|
||||
result = client.startDictationStream("dictation", "pcm");
|
||||
else result = client.finishDictationStream("dictation", 0);
|
||||
const rejected = expect(result).rejects.toThrow("control capacity");
|
||||
drain();
|
||||
await first;
|
||||
await rejected;
|
||||
expect(client.isConnected).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
test("routed channels inherit an origin without drain support", async () => {
|
||||
const { createHostChannelTransport } = await import("./hosts/index.js");
|
||||
const mock = createMockTransport();
|
||||
mock.transport.supportsDrain = false;
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "unpaced-route",
|
||||
transportFactory: () => mock.transport,
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
clients.push(client);
|
||||
const connecting = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connecting;
|
||||
const route = createHostChannelTransport(
|
||||
client,
|
||||
"srv_destination",
|
||||
"scope",
|
||||
"route",
|
||||
)({ url: "ws://route" });
|
||||
expect(route.supportsDrain).toBe(false);
|
||||
route.close();
|
||||
});
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import {
|
||||
ChunkedTransport,
|
||||
ChunkedTransportProtocolError,
|
||||
} from "@getpaseo/protocol/chunked-transport";
|
||||
import {
|
||||
HOST_ROUTE_OPCODE,
|
||||
encodeHostRouteFrame,
|
||||
@@ -1266,7 +1270,7 @@ function toReasonCode(reason: string | null | undefined): string | null {
|
||||
}
|
||||
|
||||
interface PendingSend {
|
||||
send: () => void;
|
||||
send: () => void | Promise<void>;
|
||||
resolve: () => void;
|
||||
reject: (error: Error) => void;
|
||||
timeoutHandle: ReturnType<typeof setTimeout>;
|
||||
@@ -1366,6 +1370,7 @@ export class DaemonClient {
|
||||
failed: (error) => this.logger.error({ err: error }, "Subscription failed"),
|
||||
});
|
||||
private transport: DaemonTransport | null = null;
|
||||
private chunkedTransport: ChunkedTransport | null = null;
|
||||
private helloAuth: HelloAuth;
|
||||
private transportCleanup: Array<() => void> = [];
|
||||
private rawMessageListeners: Set<(message: SessionOutboundMessage) => void> = new Set();
|
||||
@@ -1560,6 +1565,17 @@ export class DaemonClient {
|
||||
});
|
||||
this.traceConnectionAttempt += 1;
|
||||
this.transport = transport;
|
||||
this.chunkedTransport = new ChunkedTransport({
|
||||
get supportsDrain() {
|
||||
return transport.supportsDrain;
|
||||
},
|
||||
get frameBytes() {
|
||||
return transport.maxFrameBytes ?? 32 * 1024 * 1024;
|
||||
},
|
||||
send: (frame) => transport.send(frame),
|
||||
onFailure: () => transport.close(4003, "Transport send failed"),
|
||||
deliver: (frame) => this.handleTransportMessage(frame),
|
||||
});
|
||||
this.lastServerInfoMessage = null;
|
||||
this.authFailureReasonValue = null;
|
||||
|
||||
@@ -1666,7 +1682,16 @@ export class DaemonClient {
|
||||
});
|
||||
}),
|
||||
transport.onMessage((data) => {
|
||||
if (this.transport === transport) this.handleTransportMessage(data);
|
||||
if (this.transport !== transport) return;
|
||||
const bytes = asUint8Array(data);
|
||||
try {
|
||||
if (typeof data === "string" || bytes)
|
||||
this.chunkedTransport!.receive(typeof data === "string" ? data : bytes!);
|
||||
else this.handleTransportMessage(data);
|
||||
} catch (error) {
|
||||
if (!(error instanceof ChunkedTransportProtocolError)) throw error;
|
||||
transport.close(4003, "Invalid chunked transport");
|
||||
}
|
||||
}),
|
||||
];
|
||||
} catch (error) {
|
||||
@@ -1867,9 +1892,9 @@ export class DaemonClient {
|
||||
})
|
||||
).routeId;
|
||||
}
|
||||
sendHostFrame(routeId: string, data: string | Uint8Array): void {
|
||||
sendHostFrame(routeId: string, data: string | Uint8Array): Promise<void> {
|
||||
if (typeof data === "string") {
|
||||
this.sendSessionMessageStrict({
|
||||
return this.sendSessionMessageStrict({
|
||||
type: "hosts.route.frame.request",
|
||||
routeId,
|
||||
data,
|
||||
@@ -1877,7 +1902,7 @@ export class DaemonClient {
|
||||
});
|
||||
} else {
|
||||
if (!this.isConnected || !this.transport) throw this.unavailableError();
|
||||
this.sendTransportFrame(encodeHostRouteFrame(routeId, data));
|
||||
return this.sendTransportFrame(encodeHostRouteFrame(routeId, data));
|
||||
}
|
||||
}
|
||||
private readonly hostFrameListeners = new Set<(frame: HostRouteFrame) => void>();
|
||||
@@ -1935,6 +1960,11 @@ export class DaemonClient {
|
||||
// Message Subscription
|
||||
// ============================================================================
|
||||
|
||||
/** Drain capability inherited by transports routed through this connection. */
|
||||
get supportsDrain(): boolean {
|
||||
return this.transport?.supportsDrain !== false;
|
||||
}
|
||||
|
||||
subscribe(handler: DaemonEventHandler): () => void {
|
||||
this.eventListeners.add(handler);
|
||||
return () => {
|
||||
@@ -2043,7 +2073,7 @@ export class DaemonClient {
|
||||
messageType: string,
|
||||
message: unknown,
|
||||
submitted?: () => void,
|
||||
): void {
|
||||
): Promise<void> {
|
||||
const isOpen =
|
||||
this.config.trace?.isEnabled() === true &&
|
||||
this.beginTraceSection("paseo.ws.message.outbound", {
|
||||
@@ -2052,25 +2082,30 @@ export class DaemonClient {
|
||||
...describeRequestIdentity(message),
|
||||
});
|
||||
try {
|
||||
this.sendTransportFrame(JSON.stringify(message), submitted);
|
||||
return this.sendTransportFrame(JSON.stringify(message), submitted);
|
||||
} finally {
|
||||
this.endTraceSection(isOpen);
|
||||
}
|
||||
}
|
||||
|
||||
private sendSessionPayload(payload: SessionInboundMessage): void {
|
||||
this.sendJsonMessage("session", payload.type, { type: "session", message: payload }, () => {
|
||||
if (!("requestId" in payload) || typeof payload.requestId !== "string") return;
|
||||
for (const waiter of this.waiters) {
|
||||
if (waiter.requestId === payload.requestId) waiter.submitted = true;
|
||||
}
|
||||
});
|
||||
private sendSessionPayload(payload: SessionInboundMessage): Promise<void> {
|
||||
return this.sendJsonMessage(
|
||||
"session",
|
||||
payload.type,
|
||||
{ type: "session", message: payload },
|
||||
() => {
|
||||
if (!("requestId" in payload) || typeof payload.requestId !== "string") return;
|
||||
for (const waiter of this.waiters) {
|
||||
if (waiter.requestId === payload.requestId) waiter.submitted = true;
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
private sendTransportFrame(
|
||||
frame: string | Uint8Array | ArrayBuffer,
|
||||
submitted?: () => void,
|
||||
): void {
|
||||
): Promise<void> {
|
||||
if (!this.transport) {
|
||||
throw this.unavailableError();
|
||||
}
|
||||
@@ -2083,7 +2118,8 @@ export class DaemonClient {
|
||||
});
|
||||
try {
|
||||
submitted?.();
|
||||
this.transport.send(frame);
|
||||
const payload = typeof frame === "string" ? frame : asUint8Array(frame)!;
|
||||
return this.chunkedTransport!.send(payload);
|
||||
} finally {
|
||||
this.endTraceSection(isOpen);
|
||||
}
|
||||
@@ -2112,10 +2148,10 @@ export class DaemonClient {
|
||||
}
|
||||
}
|
||||
|
||||
private sendBinaryFrame(frame: Uint8Array): void {
|
||||
private sendBinaryFrame(frame: Uint8Array): Promise<void> {
|
||||
if (!this.transport || this.connectionState.status !== "connected") {
|
||||
if (this.config.suppressSendErrors) {
|
||||
return;
|
||||
return Promise.resolve();
|
||||
}
|
||||
throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
|
||||
}
|
||||
@@ -2124,10 +2160,10 @@ export class DaemonClient {
|
||||
envelopeType: "binary",
|
||||
messageType: "binary",
|
||||
});
|
||||
this.sendTransportFrame(frame);
|
||||
return this.sendTransportFrame(frame);
|
||||
} catch (error) {
|
||||
if (this.config.suppressSendErrors) {
|
||||
return;
|
||||
return Promise.resolve();
|
||||
}
|
||||
throw error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
@@ -2142,7 +2178,7 @@ export class DaemonClient {
|
||||
private sendSessionMessageOrThrow(message: SessionInboundMessage): Promise<void> {
|
||||
return this.sendWhenConnected(() => {
|
||||
const payload = SessionInboundMessageSchema.parse(message);
|
||||
this.sendSessionPayload(payload);
|
||||
return this.sendSessionPayload(payload);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -2151,13 +2187,12 @@ export class DaemonClient {
|
||||
return this.sendWhenConnected(() => undefined);
|
||||
}
|
||||
|
||||
private sendWhenConnected(send: () => void): Promise<void> {
|
||||
private sendWhenConnected(send: () => void | Promise<void>): Promise<void> {
|
||||
const status = this.connectionState.status;
|
||||
|
||||
// If connected, send immediately
|
||||
if (this.transport && status === "connected") {
|
||||
send();
|
||||
return Promise.resolve();
|
||||
return Promise.resolve(send());
|
||||
}
|
||||
|
||||
// If connecting, queue the message to be sent once connected
|
||||
@@ -2196,8 +2231,7 @@ export class DaemonClient {
|
||||
clearTimeout(pending.timeoutHandle);
|
||||
try {
|
||||
if (this.transport && this.connectionState.status === "connected") {
|
||||
pending.send();
|
||||
pending.resolve();
|
||||
void Promise.resolve(pending.send()).then(pending.resolve, pending.reject);
|
||||
} else {
|
||||
pending.reject(new DaemonConnectionError("Connection lost before message could be sent"));
|
||||
}
|
||||
@@ -2253,19 +2287,17 @@ export class DaemonClient {
|
||||
{ ...params.options, requestId: params.requestId },
|
||||
);
|
||||
|
||||
// The response deadline still applies while the transport waits for drain.
|
||||
void promise.catch(() => undefined);
|
||||
try {
|
||||
await this.sendSessionMessageOrThrow(wire.message);
|
||||
const sent = this.sendSessionMessageOrThrow(wire.message);
|
||||
const result = await Promise.race([promise, sent.then(() => promise)]);
|
||||
if (result.kind === "error") throw result.error;
|
||||
return result.value;
|
||||
} catch (error) {
|
||||
const err = error instanceof Error ? error : new Error(String(error));
|
||||
cancel(err);
|
||||
void promise.catch(() => undefined);
|
||||
throw err;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await promise;
|
||||
if (result.kind === "error") throw result.error;
|
||||
return result.value;
|
||||
} finally {
|
||||
await wire.finish();
|
||||
}
|
||||
@@ -2355,13 +2387,13 @@ export class DaemonClient {
|
||||
});
|
||||
}
|
||||
|
||||
private sendSessionMessageStrict(message: SessionInboundMessage): void {
|
||||
private sendSessionMessageStrict(message: SessionInboundMessage): Promise<void> {
|
||||
if (!this.transport || this.connectionState.status !== "connected") {
|
||||
throw this.unavailableError();
|
||||
}
|
||||
const payload = SessionInboundMessageSchema.parse(message);
|
||||
try {
|
||||
this.sendSessionPayload(payload);
|
||||
return this.sendSessionPayload(payload);
|
||||
} catch (error) {
|
||||
throw error instanceof Error ? error : new Error(String(error));
|
||||
}
|
||||
@@ -2559,15 +2591,16 @@ export class DaemonClient {
|
||||
};
|
||||
this.pingProbe = probe;
|
||||
|
||||
try {
|
||||
this.sendJsonMessage("ping", "ping", { type: "ping" });
|
||||
} catch (error) {
|
||||
this.clearPingProbe();
|
||||
const rejectSend = (error: unknown) => {
|
||||
if (this.pingProbe !== probe) return;
|
||||
const sendError = error instanceof Error ? error : new Error(String(error));
|
||||
if (probe.drivesLivenessFailure) {
|
||||
this.recordLivenessFailure(sendError);
|
||||
}
|
||||
return Promise.reject(sendError);
|
||||
this.rejectPingProbe(sendError);
|
||||
if (probe.drivesLivenessFailure) this.recordLivenessFailure(sendError);
|
||||
};
|
||||
try {
|
||||
void this.sendJsonMessage("ping", "ping", { type: "ping" }).catch(rejectSend);
|
||||
} catch (error) {
|
||||
rejectSend(error);
|
||||
}
|
||||
|
||||
return promise;
|
||||
@@ -4222,7 +4255,7 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
async sendVoiceAudioChunk(audio: string, format: string, isLast = false): Promise<void> {
|
||||
this.sendSessionMessage({ type: "voice_audio_chunk", audio, format, isLast });
|
||||
return this.sendSessionMessageStrict({ type: "voice_audio_chunk", audio, format, isLast });
|
||||
}
|
||||
|
||||
async startDictationStream(dictationId: string, format: string): Promise<void> {
|
||||
@@ -4263,8 +4296,13 @@ export class DaemonClient {
|
||||
|
||||
const cleanupError = new Error("Cancelled dictation start waiter");
|
||||
try {
|
||||
this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format });
|
||||
await Promise.race([ackPromise, errorPromise]);
|
||||
const sent = this.sendSessionMessageStrict({
|
||||
type: "dictation_stream_start",
|
||||
dictationId,
|
||||
format,
|
||||
});
|
||||
const response = Promise.race([ackPromise, errorPromise]);
|
||||
await Promise.race([response, sent.then(() => response)]);
|
||||
} finally {
|
||||
ack.cancel(cleanupError);
|
||||
streamError.cancel(cleanupError);
|
||||
@@ -4399,12 +4437,17 @@ export class DaemonClient {
|
||||
|
||||
const cleanupError = new Error("Cancelled dictation finish waiter");
|
||||
try {
|
||||
this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq });
|
||||
const firstOutcome = await Promise.race([
|
||||
const sent = this.sendSessionMessageStrict({
|
||||
type: "dictation_stream_finish",
|
||||
dictationId,
|
||||
finalSeq,
|
||||
});
|
||||
const response = Promise.race([
|
||||
finalOutcomePromise,
|
||||
errorOutcomePromise,
|
||||
finishAcceptedOutcomePromise,
|
||||
]);
|
||||
const firstOutcome = await Promise.race([response, sent.then(() => response)]);
|
||||
|
||||
if (firstOutcome.kind === "final") {
|
||||
return firstOutcome.payload;
|
||||
@@ -5320,8 +5363,10 @@ export class DaemonClient {
|
||||
return undefined;
|
||||
},
|
||||
);
|
||||
const sendFrame = (frame: Uint8Array) =>
|
||||
Promise.race([this.sendBinaryFrame(frame), responsePromise]);
|
||||
try {
|
||||
this.sendBinaryFrame(
|
||||
await sendFrame(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileBegin,
|
||||
requestId: resolvedRequestId,
|
||||
@@ -5335,6 +5380,7 @@ export class DaemonClient {
|
||||
}),
|
||||
);
|
||||
|
||||
if (settled) return await responsePromise;
|
||||
const chunkSize = input.chunkSize ?? 128 * 1024;
|
||||
for (let offset = 0; offset < bytes.byteLength; offset += chunkSize) {
|
||||
// Native WebSocket.send encodes binary synchronously. Let rendering and
|
||||
@@ -5344,7 +5390,7 @@ export class DaemonClient {
|
||||
if (this.transport !== uploadTransport || this.connectionState.status !== "connected") {
|
||||
throw new DaemonConnectionError("Connection changed during file upload");
|
||||
}
|
||||
this.sendBinaryFrame(
|
||||
await sendFrame(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileChunk,
|
||||
requestId: resolvedRequestId,
|
||||
@@ -5353,7 +5399,7 @@ export class DaemonClient {
|
||||
);
|
||||
}
|
||||
|
||||
this.sendBinaryFrame(
|
||||
await sendFrame(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileEnd,
|
||||
requestId: resolvedRequestId,
|
||||
@@ -6763,6 +6809,8 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
private disposeTransport(code = 1001, reason = "Reconnecting"): void {
|
||||
this.chunkedTransport?.close();
|
||||
this.chunkedTransport = null;
|
||||
this.owned.disconnected();
|
||||
this.providerSnapshotUpdates.clear();
|
||||
this.stopLivenessHeartbeat();
|
||||
@@ -7232,15 +7280,6 @@ export class DaemonClient {
|
||||
probe.resolve(perfNow() - probe.startedAt);
|
||||
}
|
||||
|
||||
private clearPingProbe(): void {
|
||||
const probe = this.pingProbe;
|
||||
if (!probe) {
|
||||
return;
|
||||
}
|
||||
this.pingProbe = null;
|
||||
clearTimeout(probe.timeoutHandle);
|
||||
}
|
||||
|
||||
private rejectPingProbe(error: Error): void {
|
||||
const probe = this.pingProbe;
|
||||
if (!probe) {
|
||||
@@ -7289,6 +7328,10 @@ export class DaemonClient {
|
||||
if (consumerMessage.type === "status") {
|
||||
const serverInfo = parseServerInfoStatusPayload(consumerMessage.payload);
|
||||
if (serverInfo) {
|
||||
if (this.chunkedTransport)
|
||||
this.chunkedTransport.enabled =
|
||||
serverInfo.features?.binaryChunkingV1 === true &&
|
||||
this.config.capabilities?.[CLIENT_CAPS.binaryChunkingV1] !== false;
|
||||
this.lastServerInfoMessage = serverInfo;
|
||||
this.destinationServerId = serverInfo.serverId;
|
||||
this.connectionFailure = undefined;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { MAX_HOST_CHANNEL_FRAME_BYTES } from "@getpaseo/protocol/binary-frames/index";
|
||||
import { safeRandomId } from "../daemon-client-transport-utils.js";
|
||||
import { DaemonClient } from "../daemon-client.js";
|
||||
import type { DaemonTransportFactory } from "../daemon-client-transport-types.js";
|
||||
@@ -84,13 +85,17 @@ export function createHostChannelTransport(
|
||||
})
|
||||
.catch(close);
|
||||
return {
|
||||
get supportsDrain() {
|
||||
return origin.supportsDrain;
|
||||
},
|
||||
maxFrameBytes: MAX_HOST_CHANNEL_FRAME_BYTES,
|
||||
send(data) {
|
||||
if (closed || !routeId)
|
||||
throw Object.assign(new Error(`Host ${serverId} is offline`), {
|
||||
code: "host_offline",
|
||||
serverId,
|
||||
});
|
||||
origin.sendHostFrame(
|
||||
return origin.sendHostFrame(
|
||||
routeId,
|
||||
typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data),
|
||||
);
|
||||
|
||||
@@ -44,12 +44,14 @@ test("a slow consumer cannot exceed the shared byte budget and close releases re
|
||||
socket.pause();
|
||||
await ready;
|
||||
const frame = new Uint8Array(1024 * 1024);
|
||||
for (let index = 0; index < 4; index++) transport.send(frame);
|
||||
const writes = Array.from({ length: 4 }, () => transport.send(frame));
|
||||
const drained = Promise.allSettled(writes);
|
||||
expect(budget.bufferedBytes).toBe(4 * 1024 * 1024);
|
||||
expect(() => transport.send(frame)).toThrow("Connected host buffer limit reached");
|
||||
expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit);
|
||||
transport.close();
|
||||
await expect.poll(() => budget.bufferedBytes).toBe(0);
|
||||
await drained;
|
||||
});
|
||||
test("direct connector preserves 25 MiB binary frames in both directions", async () => {
|
||||
const server = await endpoint();
|
||||
@@ -65,12 +67,13 @@ test("direct connector preserves 25 MiB binary frames in both directions", async
|
||||
await ready;
|
||||
const bytes = new Uint8Array(25 * 1024 * 1024).fill(173);
|
||||
const echoed = once(socket, "message");
|
||||
transport.send(bytes);
|
||||
const sending = transport.send(bytes);
|
||||
const [sent, binary] = await echoed;
|
||||
expect(binary).toBe(true);
|
||||
expect(Buffer.from(sent as Uint8Array).equals(bytes)).toBe(true);
|
||||
socket.send(bytes);
|
||||
expect(Buffer.from(await received).equals(bytes)).toBe(true);
|
||||
await sending;
|
||||
});
|
||||
|
||||
import {
|
||||
@@ -124,24 +127,29 @@ test("encrypted connector rejects overload synchronously and preserves the physi
|
||||
});
|
||||
await new Promise<void>((resolve) => transport.onOpen(resolve));
|
||||
await ready;
|
||||
expect(transport.maxFrameBytes).toBeLessThan(32 * 1024 * 1024);
|
||||
peer.pause();
|
||||
const frame = new Uint8Array(1024 * 1024).fill(173);
|
||||
let rejected: unknown;
|
||||
const writes: (void | Promise<void>)[] = [];
|
||||
for (let index = 0; index < 8; index++) {
|
||||
try {
|
||||
transport.send(frame);
|
||||
writes.push(transport.send(frame));
|
||||
} catch (error) {
|
||||
rejected = error;
|
||||
break;
|
||||
}
|
||||
}
|
||||
const drained = Promise.allSettled(writes);
|
||||
expect(rejected).toMatchObject({ code: "resource_limit" });
|
||||
expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit);
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(errors).toEqual([]);
|
||||
expect(closed).toBe(false);
|
||||
transport.send("sibling control");
|
||||
const sibling = transport.send("sibling control");
|
||||
peer.resume();
|
||||
await drained;
|
||||
await sibling;
|
||||
await expect.poll(() => received.includes("sibling control")).toBe(true);
|
||||
expect(received.filter((data) => data instanceof ArrayBuffer).length).toBeGreaterThan(0);
|
||||
expect(
|
||||
|
||||
@@ -172,9 +172,15 @@ export function createHostTransportFactory(
|
||||
close();
|
||||
});
|
||||
return {
|
||||
get supportsDrain() {
|
||||
return transport?.supportsDrain;
|
||||
},
|
||||
get maxFrameBytes() {
|
||||
return transport?.maxFrameBytes;
|
||||
},
|
||||
send(data) {
|
||||
if (closed || !transport) throw new Error("Host transport is unavailable");
|
||||
transport.send(data);
|
||||
return transport.send(data);
|
||||
},
|
||||
close,
|
||||
onOpen(handler) {
|
||||
|
||||
@@ -288,7 +288,7 @@ export type ToolCallDetail =
|
||||
subAgentType?: string;
|
||||
description?: string;
|
||||
childSessionId?: string;
|
||||
log: string;
|
||||
log?: string;
|
||||
actions?: Array<{
|
||||
index: number;
|
||||
toolName: string;
|
||||
|
||||
@@ -3,6 +3,11 @@ import { MAX_WEBSOCKET_MESSAGE_BYTES } from "../websocket-message-limit.js";
|
||||
export const HOST_ROUTE_OPCODE = 0x30;
|
||||
export const MAX_HOST_ROUTE_ID_BYTES = 256;
|
||||
export const MAX_HOST_ROUTE_HEADER_BYTES = 3 + MAX_HOST_ROUTE_ID_BYTES;
|
||||
// A logical host channel crosses the 32 MiB route admission boundary. Text can
|
||||
// expand sixfold when enclosed in JSON; binary adds the bounded route header.
|
||||
export const MAX_HOST_CHANNEL_FRAME_BYTES = Math.floor(
|
||||
(32 * 1024 * 1024 - MAX_HOST_ROUTE_HEADER_BYTES) / 6,
|
||||
);
|
||||
export const MAX_ROUTED_PAYLOAD_BYTES = MAX_WEBSOCKET_MESSAGE_BYTES - MAX_HOST_ROUTE_HEADER_BYTES;
|
||||
export interface HostRouteFrame {
|
||||
routeId: string;
|
||||
|
||||
@@ -0,0 +1,392 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { ChunkedTransport } from "./chunked-transport.js";
|
||||
|
||||
describe("chunked transport", () => {
|
||||
it("delivers a large UTF-8 logical message once, preserving order", async () => {
|
||||
const delivered: (string | Uint8Array)[] = [];
|
||||
const receiver = new ChunkedTransport({
|
||||
frameBytes: 17,
|
||||
send: () => {},
|
||||
deliver: (v) => delivered.push(v),
|
||||
});
|
||||
receiver.enabled = true;
|
||||
const frames: (string | Uint8Array)[] = [];
|
||||
const sender = new ChunkedTransport({
|
||||
frameBytes: 17,
|
||||
send: (v) => {
|
||||
frames.push(v);
|
||||
receiver.receive(v);
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
sender.enabled = true;
|
||||
const text = "🌍".repeat(100);
|
||||
await sender.send(text);
|
||||
expect(delivered).toEqual([text]);
|
||||
expect(frames.length).toBeGreaterThan(1);
|
||||
expect(frames.every((v) => v instanceof Uint8Array && v.byteLength <= 17)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves binary kind and leaves legacy frames unchanged", async () => {
|
||||
const delivered: (string | Uint8Array)[] = [];
|
||||
const receiver = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: () => {},
|
||||
deliver: (v) => delivered.push(v),
|
||||
});
|
||||
const frames: (string | Uint8Array)[] = [];
|
||||
const sender = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: (v) => {
|
||||
frames.push(v);
|
||||
receiver.receive(v);
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
const data = new Uint8Array(40).fill(5);
|
||||
await sender.send(data);
|
||||
expect(frames).toEqual([data]);
|
||||
frames.length = 0;
|
||||
sender.enabled = receiver.enabled = true;
|
||||
await sender.send(data);
|
||||
expect(frames).toHaveLength(10);
|
||||
expect(delivered).toEqual([data, data]);
|
||||
});
|
||||
|
||||
it("rejects malformed sequences and drops incomplete transfers on close", () => {
|
||||
const delivered: unknown[] = [];
|
||||
const receiver = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: () => {},
|
||||
deliver: (v) => delivered.push(v),
|
||||
});
|
||||
receiver.enabled = true;
|
||||
receiver.receive(new Uint8Array([0x31, 1, 0, 1, 65]));
|
||||
expect(() => receiver.receive("interrupted")).toThrow("Interrupted");
|
||||
receiver.receive(new Uint8Array([0x31, 1, 0, 2, 66]));
|
||||
expect(delivered).toEqual([]);
|
||||
});
|
||||
|
||||
it("waits for drain before the next chunk and preserves logical ordering", async () => {
|
||||
const frames: (string | Uint8Array)[] = [];
|
||||
let drain!: () => void;
|
||||
const sender = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: (v) => {
|
||||
frames.push(v);
|
||||
if (frames.length === 1)
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
sender.enabled = true;
|
||||
const first = sender.send("abcdefghij");
|
||||
const next = new Uint8Array([1, 2, 3, 4]);
|
||||
const second = sender.send(next);
|
||||
await Promise.resolve();
|
||||
expect(frames).toHaveLength(1);
|
||||
expect(sender.queuedByteLength).toBe(4);
|
||||
drain();
|
||||
await Promise.all([first, second]);
|
||||
expect(frames).toHaveLength(4);
|
||||
expect(frames[3]).toEqual(next);
|
||||
expect(sender.queuedByteLength).toBe(0);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[0x31, 2, 0, 3, 65],
|
||||
[0x31, 1, 2, 3, 65],
|
||||
[0x31, 1, 0, 4, 65],
|
||||
[0x31, 1, 0, 2, 65],
|
||||
[0x31, 1, 0, 1],
|
||||
])("rejects malformed chunk metadata %j", (...bytes) => {
|
||||
const receiver = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: () => {},
|
||||
deliver: () => {
|
||||
throw new Error("Unexpected delivery");
|
||||
},
|
||||
});
|
||||
receiver.enabled = true;
|
||||
expect(() => receiver.receive(new Uint8Array(bytes))).toThrow();
|
||||
});
|
||||
|
||||
it("reconnect discards partial state and requires negotiation again", () => {
|
||||
const delivered: unknown[] = [];
|
||||
const create = () =>
|
||||
new ChunkedTransport({ frameBytes: 8, send: () => {}, deliver: (v) => delivered.push(v) });
|
||||
const old = create();
|
||||
old.enabled = true;
|
||||
old.receive(new Uint8Array([0x31, 1, 0, 1, 65]));
|
||||
old.close();
|
||||
old.receive(new Uint8Array([0x31, 1, 0, 2, 66]));
|
||||
const fresh = create();
|
||||
expect(fresh.enabled).toBe(false);
|
||||
fresh.enabled = true;
|
||||
fresh.receive(new Uint8Array([0x31, 1, 0, 3, 67]));
|
||||
expect(delivered).toEqual(["C"]);
|
||||
});
|
||||
|
||||
it("bounds logical messages waiting behind a blocked writer", async () => {
|
||||
let drain!: () => void;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 32 * 1024 * 1024,
|
||||
send: () =>
|
||||
new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
}),
|
||||
deliver: () => {},
|
||||
});
|
||||
const first = transport.send(new Uint8Array(100 * 1024 * 1024));
|
||||
const overflow = transport.send("one more byte");
|
||||
try {
|
||||
await expect(overflow).rejects.toThrow("queue");
|
||||
} finally {
|
||||
drain();
|
||||
await first;
|
||||
}
|
||||
});
|
||||
|
||||
it("requests drain for chunks while ordinary frames keep the immediate write path", async () => {
|
||||
const waits: boolean[] = [];
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
send: (_frame, wait) => {
|
||||
waits.push(wait);
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
transport.enabled = true;
|
||||
await transport.send("small");
|
||||
await transport.send("abcdefghij");
|
||||
expect(waits).toEqual([false, true, true, true]);
|
||||
});
|
||||
|
||||
it("charges queued ASCII messages by their UTF-8 bytes", async () => {
|
||||
let drain!: () => void;
|
||||
let blocked = true;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 32 * 1024 * 1024,
|
||||
send: () => {
|
||||
if (blocked) {
|
||||
blocked = false;
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
}
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
transport.enabled = true;
|
||||
const first = transport.send(new Uint8Array([1]));
|
||||
const text = "x".repeat(10 * 1024 * 1024);
|
||||
const queued = Array.from({ length: 4 }, () => transport.send(text));
|
||||
const settled = Promise.allSettled(queued);
|
||||
expect(transport.queuedByteLength).toBe(40 * 1024 * 1024);
|
||||
drain();
|
||||
await first;
|
||||
expect((await settled).map((result) => result.status)).toEqual([
|
||||
"fulfilled",
|
||||
"fulfilled",
|
||||
"fulfilled",
|
||||
"fulfilled",
|
||||
]);
|
||||
});
|
||||
|
||||
it("notifies the socket owner when an awaited send overflows the queue", async () => {
|
||||
let drain!: () => void;
|
||||
const failures: unknown[] = [];
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 32 * 1024 * 1024,
|
||||
send: () =>
|
||||
new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
}),
|
||||
deliver: () => {},
|
||||
onFailure: (error) => {
|
||||
failures.push(error);
|
||||
},
|
||||
});
|
||||
const first = transport.send(new Uint8Array(100 * 1024 * 1024));
|
||||
const overflow = transport.send("extra");
|
||||
await expect(overflow).rejects.toThrow("queue");
|
||||
drain();
|
||||
await first;
|
||||
expect(failures).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("bounds retained reassembly memory for many one-byte chunks", async () => {
|
||||
const { spawnSync } = await import("node:child_process");
|
||||
// Exercise the built package as a Node consumer; CI builds it before protocol tests.
|
||||
const entry = new URL("../dist/chunked-transport.js", import.meta.url).href;
|
||||
const result = spawnSync(
|
||||
process.execPath,
|
||||
[
|
||||
"--expose-gc",
|
||||
"--input-type=module",
|
||||
"-e",
|
||||
`
|
||||
import { ChunkedTransport } from ${JSON.stringify(entry)};
|
||||
let delivered = 0;
|
||||
const receiver = new ChunkedTransport({ frameBytes: 8, send() {}, deliver(frame) { delivered = frame.byteLength; } });
|
||||
receiver.enabled = true;
|
||||
global.gc();
|
||||
const before = process.memoryUsage().heapUsed;
|
||||
receiver.receive(new Uint8Array([0x31, 1, 1, 1, 65]));
|
||||
for (let index = 0; index < 300000; index++) receiver.receive(new Uint8Array([0x31, 1, 1, 0, 65]));
|
||||
global.gc();
|
||||
const growth = process.memoryUsage().heapUsed - before;
|
||||
receiver.receive(new Uint8Array([0x31, 1, 1, 2, 66]));
|
||||
console.log(JSON.stringify({ growth, delivered }));
|
||||
`,
|
||||
],
|
||||
{ encoding: "utf8", timeout: 15000 },
|
||||
);
|
||||
expect(result.status, result.stderr).toBe(0);
|
||||
const observed = JSON.parse(result.stdout);
|
||||
expect(observed.delivered).toBe(300002);
|
||||
expect(observed.growth).toBeLessThan(8 * 1024 * 1024);
|
||||
}, 20000);
|
||||
|
||||
it.each(["ASCII", "中", "🌍", "\ud800", "\udc00", "\ud800x"])(
|
||||
"accounts for queued UTF-8 text %j exactly",
|
||||
async (text) => {
|
||||
let drain!: () => void;
|
||||
let blocked = true;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 1024,
|
||||
send: () => {
|
||||
if (blocked) {
|
||||
blocked = false;
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
}
|
||||
},
|
||||
deliver: () => {},
|
||||
});
|
||||
const first = transport.send(new Uint8Array([1]));
|
||||
const second = transport.send(text);
|
||||
expect(transport.queuedByteLength).toBe(new TextEncoder().encode(text).byteLength);
|
||||
drain();
|
||||
await Promise.all([first, second]);
|
||||
},
|
||||
);
|
||||
|
||||
it("notifies the owner once when an accepted writer fails", async () => {
|
||||
const failures: unknown[] = [];
|
||||
const error = new Error("write failed");
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 1024,
|
||||
deliver: () => {},
|
||||
onFailure: (failure) => {
|
||||
failures.push(failure);
|
||||
},
|
||||
send: () => Promise.reject(error),
|
||||
});
|
||||
await expect(transport.send("hello", { waitForDrain: true })).rejects.toThrow("write failed");
|
||||
expect(failures).toEqual([error]);
|
||||
expect(() => transport.send("again")).toThrow("Transport closed");
|
||||
expect(failures).toEqual([error]);
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
"a pre-write rejection preserves the sender (chunked=%s)",
|
||||
async (chunked) => {
|
||||
const failures: unknown[] = [];
|
||||
let reject = true;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
deliver: () => {},
|
||||
onFailure: (error) => {
|
||||
failures.push(error);
|
||||
},
|
||||
send: () => {
|
||||
if (reject) throw new Error("capacity");
|
||||
},
|
||||
});
|
||||
transport.enabled = chunked;
|
||||
expect(() => transport.send("abcdefghij")).toThrow("capacity");
|
||||
expect(failures).toEqual([]);
|
||||
reject = false;
|
||||
await transport.send("sibling");
|
||||
},
|
||||
);
|
||||
|
||||
it("a queued pre-write rejection does not poison the next logical send", async () => {
|
||||
let drain!: () => void;
|
||||
const failures: unknown[] = [];
|
||||
const delivered: string[] = [];
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 1024,
|
||||
deliver: () => {},
|
||||
onFailure: (error) => {
|
||||
failures.push(error);
|
||||
},
|
||||
send: (frame) => {
|
||||
if (frame === "blocked")
|
||||
return new Promise<void>((resolve) => {
|
||||
drain = resolve;
|
||||
});
|
||||
if (frame === "reject") throw new Error("capacity");
|
||||
delivered.push(frame as string);
|
||||
},
|
||||
});
|
||||
const first = transport.send("blocked");
|
||||
const rejected = transport.send("reject");
|
||||
const sibling = transport.send("sibling");
|
||||
const results = Promise.allSettled([first, rejected, sibling]);
|
||||
drain();
|
||||
expect((await results).map((result) => result.status)).toEqual([
|
||||
"fulfilled",
|
||||
"rejected",
|
||||
"fulfilled",
|
||||
]);
|
||||
expect(delivered).toEqual(["sibling"]);
|
||||
expect(failures).toEqual([]);
|
||||
});
|
||||
|
||||
it.each([false, true])("delivery errors leave framing usable (chunked=%s)", async (chunked) => {
|
||||
let fail = true;
|
||||
const delivered: (string | Uint8Array)[] = [];
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 64,
|
||||
send: () => {},
|
||||
deliver: (frame) => {
|
||||
if (fail) throw new Error("listener failed");
|
||||
delivered.push(frame);
|
||||
},
|
||||
});
|
||||
transport.enabled = chunked;
|
||||
const frame = chunked ? new Uint8Array([0x31, 1, 0, 3, 65]) : "A";
|
||||
expect(() => transport.receive(frame)).toThrow("listener failed");
|
||||
fail = false;
|
||||
transport.receive(frame);
|
||||
expect(delivered).toEqual(["A"]);
|
||||
await transport.send("still connected");
|
||||
});
|
||||
|
||||
it("unpaced transports retain ordinary outbound admission and accept inbound chunks", async () => {
|
||||
const frames: unknown[] = [];
|
||||
const delivered: unknown[] = [];
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 8,
|
||||
supportsDrain: false,
|
||||
send: (frame) => {
|
||||
frames.push(frame);
|
||||
if (typeof frame === "string" && frame.length > 8) throw new Error("single frame limit");
|
||||
},
|
||||
deliver: (frame) => {
|
||||
delivered.push(frame);
|
||||
},
|
||||
});
|
||||
transport.enabled = true;
|
||||
expect(() => transport.send("too large for one frame")).toThrow("single frame limit");
|
||||
expect(frames).toEqual(["too large for one frame"]);
|
||||
transport.receive(new Uint8Array([0x31, 1, 0, 3, 65]));
|
||||
expect(delivered).toEqual(["A"]);
|
||||
await transport.send("small");
|
||||
});
|
||||
@@ -0,0 +1,214 @@
|
||||
import { MAX_WEBSOCKET_MESSAGE_BYTES } from "./websocket-message-limit.js";
|
||||
|
||||
type Frame = string | Uint8Array;
|
||||
|
||||
/** A malformed wire frame, distinct from a consumer delivery failure. */
|
||||
export class ChunkedTransportProtocolError extends Error {}
|
||||
// Reserved outside the application binary opcode space. Version, kind, flags follow.
|
||||
const OPCODE = 0x31;
|
||||
const HEADER = 4;
|
||||
|
||||
/** Connection-scoped ordered framing, below serialization and routing, above encryption. */
|
||||
export class ChunkedTransport {
|
||||
enabled = false;
|
||||
private assembly: Uint8Array | undefined;
|
||||
private size = 0;
|
||||
private kind = 0;
|
||||
private closed = false;
|
||||
private pending: Promise<void> | undefined;
|
||||
private queuedBytes = 0;
|
||||
private activeBytes = 0;
|
||||
|
||||
constructor(
|
||||
private readonly io: {
|
||||
frameBytes: number;
|
||||
supportsDrain?: boolean;
|
||||
send: (frame: Frame, waitForDrain: boolean) => void | Promise<void>;
|
||||
deliver: (frame: Frame) => void;
|
||||
onFailure?: (error: unknown) => void;
|
||||
},
|
||||
) {
|
||||
if (io.frameBytes <= HEADER) throw new Error("Invalid transport frame budget");
|
||||
}
|
||||
|
||||
/** Bytes waiting behind the active write; physical buffering owns the active frame. */
|
||||
get queuedByteLength(): number {
|
||||
return this.queuedBytes - this.activeBytes;
|
||||
}
|
||||
get isWriting(): boolean {
|
||||
return this.pending !== undefined;
|
||||
}
|
||||
|
||||
send(frame: Frame, options: { waitForDrain?: boolean; byteLength?: number } = {}): Promise<void> {
|
||||
const byteLength = options.byteLength ?? frameByteLength(frame);
|
||||
if (byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) throw new Error("Logical message too large");
|
||||
const chunk =
|
||||
this.enabled && this.io.supportsDrain !== false && byteLength > this.io.frameBytes;
|
||||
if (this.closed) throw new Error("Transport closed");
|
||||
const write = () => {
|
||||
if (this.closed) throw new Error("Transport closed");
|
||||
this.activeBytes = byteLength;
|
||||
// A synchronous rejection did not admit a frame. Only accepted writes can
|
||||
// fail the connection; queued admission failures still belong to their sender.
|
||||
const result = chunk
|
||||
? this.sendChunks(
|
||||
typeof frame === "string" ? new TextEncoder().encode(frame) : frame,
|
||||
typeof frame === "string",
|
||||
)
|
||||
: this.io.send(frame, options.waitForDrain === true);
|
||||
if (!result) return;
|
||||
return result.catch((error) => {
|
||||
this.fail(error);
|
||||
throw error;
|
||||
});
|
||||
};
|
||||
if (this.queuedBytes + byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) {
|
||||
const error = new Error("Logical output queue exceeded its byte limit");
|
||||
this.fail(error);
|
||||
const rejected = Promise.reject<void>(error);
|
||||
void rejected.catch(() => {});
|
||||
return rejected;
|
||||
}
|
||||
this.queuedBytes += byteLength;
|
||||
let result: void | Promise<void>;
|
||||
try {
|
||||
result = this.pending ? this.pending.then(write) : write();
|
||||
} catch (error) {
|
||||
this.queuedBytes -= byteLength;
|
||||
this.activeBytes = 0;
|
||||
throw error;
|
||||
}
|
||||
if (!result) {
|
||||
this.queuedBytes -= byteLength;
|
||||
this.activeBytes = 0;
|
||||
return Promise.resolve();
|
||||
}
|
||||
const completion = Promise.resolve(result).finally(() => {
|
||||
this.queuedBytes -= byteLength;
|
||||
this.activeBytes = 0;
|
||||
});
|
||||
const tail = completion.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
this.pending = tail;
|
||||
void tail.then(() => {
|
||||
if (this.pending === tail) this.pending = undefined;
|
||||
return undefined;
|
||||
});
|
||||
return completion;
|
||||
}
|
||||
|
||||
private sendChunks(bytes: Uint8Array, text: boolean): Promise<void> {
|
||||
const capacity = this.io.frameBytes - HEADER;
|
||||
const writeChunk = (offset: number) => {
|
||||
if (this.closed) throw new Error("Transport closed");
|
||||
const end = Math.min(bytes.byteLength, offset + capacity);
|
||||
const wire = new Uint8Array(HEADER + end - offset);
|
||||
wire.set([
|
||||
OPCODE,
|
||||
1,
|
||||
text ? 0 : 1,
|
||||
(offset === 0 ? 1 : 0) | (end === bytes.byteLength ? 2 : 0),
|
||||
]);
|
||||
wire.set(bytes.subarray(offset, end), HEADER);
|
||||
return this.io.send(wire, true);
|
||||
};
|
||||
const first = writeChunk(0);
|
||||
return (async () => {
|
||||
await first;
|
||||
for (let offset = capacity; offset < bytes.byteLength; offset += capacity)
|
||||
await writeChunk(offset);
|
||||
})();
|
||||
}
|
||||
|
||||
receive(frame: Frame): void {
|
||||
if (this.closed) return;
|
||||
let message: Frame | undefined;
|
||||
try {
|
||||
message = this.assemble(frame);
|
||||
} catch (error) {
|
||||
this.close();
|
||||
throw new ChunkedTransportProtocolError(
|
||||
error instanceof Error ? error.message : String(error),
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
if (message !== undefined) this.io.deliver(message);
|
||||
}
|
||||
|
||||
private assemble(frame: Frame): Frame | undefined {
|
||||
if (typeof frame === "string" || frame[0] !== OPCODE) {
|
||||
if (this.assembly) throw new Error("Interrupted chunked message");
|
||||
return frame;
|
||||
}
|
||||
if (
|
||||
!this.enabled ||
|
||||
frame.length <= HEADER ||
|
||||
frame[1] !== 1 ||
|
||||
frame[2]! > 1 ||
|
||||
(frame[3]! & ~3) !== 0
|
||||
)
|
||||
throw new Error("Invalid chunk header");
|
||||
const start = (frame[3]! & 1) !== 0;
|
||||
const end = (frame[3]! & 2) !== 0;
|
||||
if (start === !!this.assembly) throw new Error("Invalid chunk sequence");
|
||||
if (start) {
|
||||
this.assembly = new Uint8Array(0);
|
||||
this.kind = frame[2]!;
|
||||
this.size = 0;
|
||||
}
|
||||
if (frame[2] !== this.kind) throw new Error("Chunk kind changed");
|
||||
this.size += frame.length - HEADER;
|
||||
if (this.size > MAX_WEBSOCKET_MESSAGE_BYTES) throw new Error("Logical message too large");
|
||||
const payload = frame.subarray(HEADER);
|
||||
if (this.assembly!.byteLength < this.size) {
|
||||
const capacity = Math.min(
|
||||
MAX_WEBSOCKET_MESSAGE_BYTES,
|
||||
Math.max(this.size, this.assembly!.byteLength * 2, 16384),
|
||||
);
|
||||
const grown = new Uint8Array(capacity);
|
||||
grown.set(this.assembly!.subarray(0, this.size - payload.byteLength));
|
||||
this.assembly = grown;
|
||||
}
|
||||
this.assembly!.set(payload, this.size - payload.byteLength);
|
||||
if (!end) return;
|
||||
const data = this.assembly!.subarray(0, this.size);
|
||||
this.assembly = undefined;
|
||||
this.size = 0;
|
||||
return this.kind === 0 ? new TextDecoder("utf-8", { fatal: true }).decode(data) : data.slice();
|
||||
}
|
||||
|
||||
private fail(error: unknown): void {
|
||||
if (this.closed) return;
|
||||
this.close();
|
||||
this.io.onFailure?.(error);
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.closed = true;
|
||||
this.enabled = false;
|
||||
this.assembly = undefined;
|
||||
this.size = 0;
|
||||
}
|
||||
}
|
||||
|
||||
function frameByteLength(frame: Frame): number {
|
||||
if (typeof frame !== "string") return frame.byteLength;
|
||||
let bytes = 0;
|
||||
for (let index = 0; index < frame.length; index++) {
|
||||
const code = frame.charCodeAt(index);
|
||||
if (code <= 0x7f) bytes++;
|
||||
else if (code <= 0x7ff) bytes += 2;
|
||||
else if (
|
||||
code >= 0xd800 &&
|
||||
code <= 0xdbff &&
|
||||
frame.charCodeAt(index + 1) >= 0xdc00 &&
|
||||
frame.charCodeAt(index + 1) <= 0xdfff
|
||||
) {
|
||||
bytes += 4;
|
||||
index++;
|
||||
} else bytes += 3;
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
export const CLIENT_CAPS = {
|
||||
binaryChunkingV1: "binary_chunking_v1",
|
||||
helloRejection: "hello_rejection",
|
||||
// COMPAT(ownedSubscriptions): added in v0.8.0, remove legacy ownership after 2027-03-09.
|
||||
ownedSubscriptions: "owned_subscriptions",
|
||||
|
||||
@@ -641,7 +641,9 @@ const ToolCallDetailPayloadSchema: z.ZodType<ToolCallDetail, unknown> = z.discri
|
||||
subAgentType: z.string().optional(),
|
||||
description: z.string().optional(),
|
||||
childSessionId: z.string().optional(),
|
||||
log: z.string(),
|
||||
// COMPAT(subagent-log): added in v0.14.1, remove after 2027-04-30. A sub-agent card is a
|
||||
// launch marker; apps before v0.14.1 required `log`, so daemons still send it empty.
|
||||
log: z.string().optional(),
|
||||
// Compat cruft for clients <= 0.1.65-beta.3 that required this field. Producers still
|
||||
// emit `[]`; nothing reads it. Drop the field (and the `[]` emissions) once those
|
||||
// clients are no longer in the field.
|
||||
@@ -3832,6 +3834,7 @@ export const ServerInfoStatusPayloadSchema = z
|
||||
selectiveAgentTimeline: z.boolean().optional(),
|
||||
explicitEventSubscriptions: z.boolean().optional(),
|
||||
ownedSubscriptions: z.boolean().optional(),
|
||||
binaryChunkingV1: z.boolean().optional(),
|
||||
workspaceSetupObservation: z.boolean().optional(),
|
||||
workspaceSortTimes: z.boolean().optional(),
|
||||
// COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30.
|
||||
|
||||
@@ -200,8 +200,8 @@ describe("wire schema compatibility", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("sub_agent tool-call payload still parses against the v0.1.65-beta.3 schema", () => {
|
||||
const parsed = LegacySubAgentToolCallSchema.parse({
|
||||
test("a sub-agent launch marker still parses against the v0.1.65-beta.3 schema", () => {
|
||||
const card = {
|
||||
type: "tool_call",
|
||||
callId: "call-sub-agent-1",
|
||||
name: "Task",
|
||||
@@ -212,12 +212,26 @@ describe("wire schema compatibility", () => {
|
||||
subAgentType: "Explore",
|
||||
description: "Inspect repository structure",
|
||||
childSessionId: "child-session-1",
|
||||
log: "[Read] README.md",
|
||||
log: "",
|
||||
actions: [],
|
||||
},
|
||||
});
|
||||
} as const;
|
||||
|
||||
expect(parsed.detail.actions).toEqual([]);
|
||||
const parsed = LegacySubAgentToolCallSchema.parse(JSON.parse(JSON.stringify(card)));
|
||||
expect(parsed.detail).toMatchObject({ log: "", actions: [] });
|
||||
const { log: _log, ...withoutLog } = card.detail;
|
||||
expect(LegacySubAgentToolCallSchema.safeParse({ ...card, detail: withoutLog }).success).toBe(
|
||||
false,
|
||||
);
|
||||
expect(AgentTimelineItemPayloadSchema.parse({ ...card, detail: withoutLog })).toMatchObject({
|
||||
detail: { type: "sub_agent", description: "Inspect repository structure" },
|
||||
});
|
||||
expect(
|
||||
AgentTimelineItemPayloadSchema.parse({
|
||||
...card,
|
||||
detail: { ...card.detail, log: "[Read] README.md" },
|
||||
}),
|
||||
).toMatchObject({ detail: { log: "[Read] README.md" } });
|
||||
});
|
||||
|
||||
test("old clients parse agent snapshots with rewind capabilities", () => {
|
||||
|
||||
@@ -247,7 +247,7 @@ second line'`,
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps nested sub-agent logs beside the sub-agent that produced them", () => {
|
||||
it("lists each sub-agent as its launch marker", () => {
|
||||
const timeline: AgentTimelineItem[] = [
|
||||
{ type: "assistant_message", text: "Before first child." },
|
||||
toolCallItem({
|
||||
@@ -257,7 +257,7 @@ second line'`,
|
||||
type: "sub_agent",
|
||||
subAgentType: "Child one",
|
||||
description: "First investigation",
|
||||
log: "[Assistant] First child result.",
|
||||
log: "",
|
||||
},
|
||||
}),
|
||||
{ type: "assistant_message", text: "Between children." },
|
||||
@@ -268,7 +268,7 @@ second line'`,
|
||||
type: "sub_agent",
|
||||
subAgentType: "Child two",
|
||||
description: "Second investigation",
|
||||
log: "[Assistant] Second child result.",
|
||||
log: "",
|
||||
},
|
||||
}),
|
||||
{ type: "assistant_message", text: "After second child." },
|
||||
@@ -279,10 +279,8 @@ second line'`,
|
||||
expect(result.split("\n")).toEqual([
|
||||
"[Assistant] Before first child.",
|
||||
"[Child one] First investigation",
|
||||
"[Assistant] First child result.",
|
||||
"[Assistant] Between children.",
|
||||
"[Child two] Second investigation",
|
||||
"[Assistant] Second child result.",
|
||||
"[Assistant] After second child.",
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -164,9 +164,6 @@ function curateProjectedActivityEntries(
|
||||
break;
|
||||
}
|
||||
entries.push(formatToolCallEntry(item, options));
|
||||
if (item.detail.type === "sub_agent" && item.detail.log.trim()) {
|
||||
entries.push(activityEntry(item.detail.log.trim()));
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "todo":
|
||||
|
||||
@@ -6,6 +6,7 @@ import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
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 { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
|
||||
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import type {
|
||||
@@ -31,7 +32,6 @@ import type {
|
||||
|
||||
const COALESCE_WINDOW_MS = AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS;
|
||||
const BEFORE_COALESCE_WINDOW_MS = Math.max(COALESCE_WINDOW_MS - 1, 0);
|
||||
const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024;
|
||||
|
||||
const TEST_CAPABILITIES: AgentCapabilityFlags = {
|
||||
supportsStreaming: false,
|
||||
@@ -400,10 +400,7 @@ describe("target coalesced behavior", () => {
|
||||
try {
|
||||
const { agentId, session } = await createManagedSession(harness);
|
||||
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
|
||||
const expectedItem = toolCall({
|
||||
status: "completed",
|
||||
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
});
|
||||
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
|
||||
|
||||
session.pushEvent(timelineEvent(toolCall({ status: "completed", output })));
|
||||
await waitForSessionEventQueue();
|
||||
@@ -425,10 +422,7 @@ describe("target coalesced behavior", () => {
|
||||
try {
|
||||
const { agentId } = await createManagedSession(harness);
|
||||
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
|
||||
const expectedItem = toolCall({
|
||||
status: "completed",
|
||||
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
});
|
||||
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
|
||||
|
||||
await harness.manager.appendTimelineItem(agentId, toolCall({ status: "completed", output }));
|
||||
|
||||
@@ -449,10 +443,7 @@ describe("target coalesced behavior", () => {
|
||||
try {
|
||||
const { agentId, session } = await createManagedSession(harness);
|
||||
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
|
||||
const expectedItem = toolCall({
|
||||
status: "completed",
|
||||
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
});
|
||||
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
|
||||
session.setHistory([timelineEvent(toolCall({ status: "completed", output }))]);
|
||||
|
||||
await harness.manager.hydrateTimelineFromProvider(agentId, { force: true });
|
||||
@@ -470,10 +461,7 @@ describe("target coalesced behavior", () => {
|
||||
try {
|
||||
const { agentId } = await createManagedSession(harness);
|
||||
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
|
||||
const expectedItem = toolCall({
|
||||
status: "completed",
|
||||
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
});
|
||||
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
|
||||
|
||||
await harness.manager.emitLiveTimelineItem(
|
||||
agentId,
|
||||
@@ -495,10 +483,9 @@ describe("target coalesced behavior", () => {
|
||||
try {
|
||||
const { agentId, session } = await createManagedSession(harness);
|
||||
const content = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
|
||||
const expectedItem = toolCall({
|
||||
status: "failed",
|
||||
error: { content: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH) },
|
||||
});
|
||||
const expectedItem = limitAgentTimelineItemContent(
|
||||
toolCall({ status: "failed", error: { content } }),
|
||||
);
|
||||
|
||||
session.pushEvent(timelineEvent(toolCall({ status: "failed", error: { content } })));
|
||||
await waitForSessionEventQueue();
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
} from "./agent-manager.js";
|
||||
import { AgentStorage } from "./agent-storage.js";
|
||||
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
|
||||
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
|
||||
import { toAgentPayload } from "./agent-projections.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
|
||||
@@ -4592,7 +4593,7 @@ test("importProviderSession imports the selected session without listing and pub
|
||||
expect(manager.getTimeline(imported.id)).toEqual([
|
||||
{ type: "user_message", text: "Trace provider imports" },
|
||||
{ type: "assistant_message", text: "Done" },
|
||||
{
|
||||
limitAgentTimelineItemContent({
|
||||
type: "tool_call",
|
||||
callId: "large-shell-result",
|
||||
name: "shell",
|
||||
@@ -4601,10 +4602,10 @@ test("importProviderSession imports the selected session without listing and pub
|
||||
detail: {
|
||||
type: "shell",
|
||||
command: "print output",
|
||||
output: "x".repeat(64 * 1024),
|
||||
output: "x".repeat(1024 * 1024),
|
||||
exitCode: 0,
|
||||
},
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(manager.listProviderSubagents(imported.id)).toEqual([
|
||||
expect.objectContaining({ id: "thread-child", title: "Imported child", status: "completed" }),
|
||||
@@ -11957,10 +11958,7 @@ test("commits startup notices once on create and after restored history", async
|
||||
exitCode: 0,
|
||||
},
|
||||
};
|
||||
const limitedToolOutput: AgentTimelineItem = {
|
||||
...toolOutput,
|
||||
detail: { type: "shell", command: "print output", output: "x".repeat(64 * 1024), exitCode: 0 },
|
||||
};
|
||||
const limitedToolOutput = limitAgentTimelineItemContent(toolOutput);
|
||||
class NotifyingSession extends TestAgentSession {
|
||||
readonly initialTimeline = [{ item: notice, timestamp: "2026-09-22T00:00:00.000Z" }];
|
||||
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
|
||||
@@ -313,7 +313,10 @@ export type ToolCallDetail =
|
||||
subAgentType?: string;
|
||||
description?: string;
|
||||
childSessionId?: string;
|
||||
log: string;
|
||||
// COMPAT(subagent-log): added in v0.14.1, remove after 2027-04-30. Apps before v0.14.1
|
||||
// require `log`; every producer sends it empty and carries the same tag. Optional here because
|
||||
// protocol timeline items, where `log` is optional, flow into this type.
|
||||
log?: string;
|
||||
actions?: Array<{
|
||||
index: number;
|
||||
toolName: string;
|
||||
|
||||
@@ -1,12 +1,30 @@
|
||||
import { AgentTimelineItemPayloadSchema } from "@getpaseo/protocol/messages";
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
|
||||
|
||||
function serializedBytes(item: AgentTimelineItem): number {
|
||||
return Buffer.byteLength(JSON.stringify(item), "utf8");
|
||||
}
|
||||
|
||||
describe("agent timeline content", () => {
|
||||
test("returns an item within budget unchanged", () => {
|
||||
const item: AgentTimelineItem = {
|
||||
type: "tool_call",
|
||||
callId: "small",
|
||||
name: "Bash",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: { type: "shell", command: "ls", output: "x".repeat(60 * 1024) },
|
||||
};
|
||||
expect(limitAgentTimelineItemContent(item)).toBe(item);
|
||||
});
|
||||
|
||||
test.each([false, true])(
|
||||
"preserves the exact UTF-16 prefix for oversized shell output (failed=%s)",
|
||||
"cuts oversized shell output and keeps short fields whole (failed=%s)",
|
||||
(failed) => {
|
||||
const text = "x".repeat(64 * 1024 - 1) + "🙂";
|
||||
const text = "x".repeat(1024 * 1024) + "🙂";
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "tool_call",
|
||||
callId: "large-shell",
|
||||
@@ -15,45 +33,116 @@ describe("agent timeline content", () => {
|
||||
detail: { type: "shell", command: "echo large", output: text },
|
||||
error: failed ? { content: text } : null,
|
||||
});
|
||||
expect(item).toEqual({
|
||||
type: "tool_call",
|
||||
expect(item).toMatchObject({
|
||||
callId: "large-shell",
|
||||
name: "Bash",
|
||||
status: failed ? "failed" : "completed",
|
||||
detail: { type: "shell", command: "echo large", output: text.slice(0, 64 * 1024) },
|
||||
error: failed ? { content: text.slice(0, 64 * 1024) } : null,
|
||||
detail: { type: "shell", command: "echo large" },
|
||||
});
|
||||
if (item.type !== "tool_call" || item.detail.type !== "shell") throw new Error("shape");
|
||||
expect(item.detail.output).toMatch(/^x+…\[truncated \d{3}\.\d KB\]$/);
|
||||
expect(serializedBytes(item)).toBeLessThanOrEqual(64 * 1024);
|
||||
},
|
||||
);
|
||||
|
||||
test("marks truncated plain-text tool content within the content budget", () => {
|
||||
const oversizedInput = "x".repeat(64 * 1024 + 1);
|
||||
|
||||
test("replaces an oversized data URL in an unknown tool result with a marker", () => {
|
||||
const url = `data:image/jpeg;base64,${"A".repeat(14 * 1024 * 1024)}`;
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "tool_call",
|
||||
callId: "terminal-session-4242",
|
||||
name: "terminal",
|
||||
callId: "screenshot",
|
||||
name: "node_repl.js",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: {
|
||||
type: "plain_text",
|
||||
text: oversizedInput,
|
||||
icon: "square_terminal",
|
||||
type: "unknown",
|
||||
input: { index: 1 },
|
||||
output: { content: [{ type: "image", url }], note: "captured" },
|
||||
},
|
||||
});
|
||||
|
||||
expect(limitAgentTimelineItemContent(item)).toEqual(item);
|
||||
expect(item).toEqual({
|
||||
type: "tool_call",
|
||||
callId: "terminal-session-4242",
|
||||
name: "terminal",
|
||||
callId: "screenshot",
|
||||
name: "node_repl.js",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: {
|
||||
type: "plain_text",
|
||||
text: "x".repeat(64 * 1024 - "\n\n[Content truncated]".length) + "\n\n[Content truncated]",
|
||||
icon: "square_terminal",
|
||||
type: "unknown",
|
||||
input: { index: 1 },
|
||||
output: { content: [{ type: "image", url: "…[truncated 14.0 MB]" }], note: "captured" },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
test("gives non-tool items a 256 KiB budget", () => {
|
||||
const text = "a".repeat(10 * 1024 * 1024);
|
||||
const item = limitAgentTimelineItemContent({ type: "assistant_message", text });
|
||||
if (item.type !== "assistant_message") throw new Error("shape");
|
||||
expect(item.text).toMatch(/^a+…\[truncated 9\.\d MB\]$/);
|
||||
expect(item.text.length).toBeGreaterThan(200 * 1024);
|
||||
expect(serializedBytes(item)).toBeLessThanOrEqual(256 * 1024);
|
||||
});
|
||||
|
||||
test("bounds an item whose structure alone exceeds the budget", () => {
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "tool_call",
|
||||
callId: "many",
|
||||
name: "query",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: { type: "unknown", input: null, output: Array.from({ length: 1_000_000 }, () => 1) },
|
||||
});
|
||||
if (item.type !== "tool_call" || item.detail.type !== "unknown") throw new Error("shape");
|
||||
const output = item.detail.output as unknown[];
|
||||
expect(output.length).toBeLessThan(1_000_000);
|
||||
expect(output.every((value) => value === 1)).toBe(true);
|
||||
expect(serializedBytes(item)).toBeLessThanOrEqual(64 * 1024);
|
||||
});
|
||||
|
||||
test("keeps short strings whole and limits idempotently when an item has many strings", () => {
|
||||
const callId = "3f2b8c1e-9a4d-4e6f-8b2a-1c3d5e7f9a0b";
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "tool_call",
|
||||
callId,
|
||||
name: "mcp__search__files",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: {
|
||||
type: "unknown",
|
||||
input: { query: "useWorkspaceSomethingVeryDescriptive" },
|
||||
output: Array.from(
|
||||
{ length: 10_000 },
|
||||
(_, index) => `${String(index).padStart(5, "0")}${"p".repeat(28)}`,
|
||||
),
|
||||
},
|
||||
});
|
||||
if (item.type !== "tool_call" || item.detail.type !== "unknown") throw new Error("shape");
|
||||
expect(item.callId).toBe(callId);
|
||||
expect(item.detail.input).toEqual({ query: "useWorkspaceSomethingVeryDescriptive" });
|
||||
const output = item.detail.output as string[];
|
||||
expect(output.length).toBeLessThan(10_000);
|
||||
expect(output.every((path) => path.length === 33)).toBe(true);
|
||||
expect(serializedBytes(item)).toBeLessThanOrEqual(64 * 1024);
|
||||
expect(limitAgentTimelineItemContent(item)).toBe(item);
|
||||
});
|
||||
|
||||
test("keeps a long typed array valid against the wire schema", () => {
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "todo",
|
||||
items: Array.from({ length: 20_000 }, (_, index) => ({
|
||||
text: `Task ${index}`,
|
||||
completed: false,
|
||||
})),
|
||||
});
|
||||
if (item.type !== "todo") throw new Error("shape");
|
||||
expect(item.items.length).toBeLessThan(20_000);
|
||||
expect(AgentTimelineItemPayloadSchema.safeParse(item).success).toBe(true);
|
||||
expect(serializedBytes(item)).toBeLessThanOrEqual(256 * 1024);
|
||||
});
|
||||
|
||||
test("is stable on an already-limited item", () => {
|
||||
const item = limitAgentTimelineItemContent({
|
||||
type: "reasoning",
|
||||
text: "r".repeat(1024 * 1024),
|
||||
});
|
||||
expect(limitAgentTimelineItemContent(item)).toBe(item);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,9 +1,19 @@
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
import type { JsonValue } from "@getpaseo/protocol/agent-types";
|
||||
|
||||
const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024;
|
||||
export const PLUGIN_TIMELINE_DATA_MAX_BYTES = 64 * 1024;
|
||||
|
||||
// One timeline fetch sends its entries in one WebSocket message, so every item must stay small.
|
||||
// Budgets count UTF-16 units plus a fixed cost per node; neither is exact JSON bytes. They bound
|
||||
// long strings and long arrays, not wide objects (every entry is kept) or multi-byte text.
|
||||
const TOOL_CALL_BUDGET = 64 * 1024;
|
||||
const ITEM_BUDGET = 256 * 1024;
|
||||
const NODE_COST = 8;
|
||||
const MARKER_COST = 32;
|
||||
// Ids, names, paths and queries stay whole, so an item keeps its identity across passes.
|
||||
const SHORT_STRING = 1024;
|
||||
const MAX_DEPTH = 64;
|
||||
|
||||
export function assertPluginTimelineDataSize(data: JsonValue): void {
|
||||
const serializedBytes = Buffer.byteLength(JSON.stringify(data), "utf8");
|
||||
if (serializedBytes > PLUGIN_TIMELINE_DATA_MAX_BYTES) {
|
||||
@@ -17,66 +27,119 @@ function retainTextPrefix(text: string, length: number): string {
|
||||
return Buffer.from(text.slice(0, length), "utf16le").toString("utf16le");
|
||||
}
|
||||
|
||||
function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem {
|
||||
if (
|
||||
item.type !== "tool_call" ||
|
||||
item.detail.type !== "shell" ||
|
||||
item.status !== "failed" ||
|
||||
typeof item.error !== "object" ||
|
||||
item.error === null ||
|
||||
!("content" in item.error) ||
|
||||
typeof item.error.content !== "string" ||
|
||||
item.error.content.length <= TOOL_CALL_CONTENT_MAX_LENGTH
|
||||
) {
|
||||
return item;
|
||||
}
|
||||
return {
|
||||
...item,
|
||||
error: {
|
||||
...item.error,
|
||||
content: retainTextPrefix(item.error.content, TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
},
|
||||
};
|
||||
function formatSize(units: number): string {
|
||||
if (units < 1024) return `${units} B`;
|
||||
if (units < 1024 * 1024) return `${(units / 1024).toFixed(1)} KB`;
|
||||
return `${(units / (1024 * 1024)).toFixed(1)} MB`;
|
||||
}
|
||||
|
||||
function limitPlainText(item: AgentTimelineItem): AgentTimelineItem {
|
||||
if (
|
||||
item.type !== "tool_call" ||
|
||||
item.detail.type !== "plain_text" ||
|
||||
typeof item.detail.text !== "string" ||
|
||||
item.detail.text.length <= TOOL_CALL_CONTENT_MAX_LENGTH
|
||||
) {
|
||||
return item;
|
||||
}
|
||||
const notice = "\n\n[Content truncated]";
|
||||
return {
|
||||
...item,
|
||||
detail: {
|
||||
...item.detail,
|
||||
text:
|
||||
retainTextPrefix(item.detail.text, TOOL_CALL_CONTENT_MAX_LENGTH - notice.length) + notice,
|
||||
},
|
||||
};
|
||||
function truncatedMarker(units: number): string {
|
||||
return `…[truncated ${formatSize(units)}]`;
|
||||
}
|
||||
|
||||
interface ItemSize {
|
||||
nodeCost: number;
|
||||
stringLengths: number[];
|
||||
tooDeep: boolean;
|
||||
}
|
||||
|
||||
function measure(value: unknown, depth: number, size: ItemSize): void {
|
||||
size.nodeCost += NODE_COST;
|
||||
if (depth > MAX_DEPTH) {
|
||||
size.tooDeep = true;
|
||||
} else if (typeof value === "string") {
|
||||
size.stringLengths.push(value.length);
|
||||
} else if (Array.isArray(value)) {
|
||||
for (const element of value) measure(element, depth + 1, size);
|
||||
} else if (typeof value === "object" && value !== null) {
|
||||
for (const [key, child] of Object.entries(value)) {
|
||||
size.nodeCost += key.length;
|
||||
measure(child, depth + 1, size);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The largest per-string length that keeps all strings, once cut, within `available`. */
|
||||
function findStringCap(lengths: readonly number[], available: number): number {
|
||||
let low = 0;
|
||||
let high = 0;
|
||||
for (const length of lengths) high = Math.max(high, length);
|
||||
while (low < high) {
|
||||
const cap = Math.ceil((low + high) / 2);
|
||||
let total = 0;
|
||||
for (const length of lengths) total += Math.min(length, cap);
|
||||
if (total <= available) low = cap;
|
||||
else high = cap - 1;
|
||||
}
|
||||
return low;
|
||||
}
|
||||
|
||||
interface CutState {
|
||||
stringCap: number;
|
||||
remaining: number;
|
||||
}
|
||||
|
||||
function cutString(text: string, cap: number): string {
|
||||
if (text.length <= cap) return text;
|
||||
// A partial data URL is useless to every reader, so drop it whole.
|
||||
if (text.startsWith("data:")) return truncatedMarker(text.length);
|
||||
return retainTextPrefix(text, cap) + truncatedMarker(text.length - cap);
|
||||
}
|
||||
|
||||
function cut(value: unknown, depth: number, state: CutState): unknown {
|
||||
state.remaining -= NODE_COST;
|
||||
if (depth > MAX_DEPTH) return "…[truncated]";
|
||||
if (typeof value === "string") {
|
||||
const text = cutString(value, state.stringCap);
|
||||
state.remaining -= text.length;
|
||||
return text;
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
let changed = false;
|
||||
const output: unknown[] = [];
|
||||
for (const element of value) {
|
||||
const remainingBefore = state.remaining;
|
||||
const next = cut(element, depth + 1, state);
|
||||
if (state.remaining < 0) {
|
||||
// Drop this and later elements. A marker element would break typed arrays.
|
||||
state.remaining = remainingBefore;
|
||||
changed = true;
|
||||
break;
|
||||
}
|
||||
changed ||= next !== element;
|
||||
output.push(next);
|
||||
}
|
||||
return changed ? output : value;
|
||||
}
|
||||
if (typeof value === "object" && value !== null) {
|
||||
let changed = false;
|
||||
const output: Record<string, unknown> = {};
|
||||
for (const [key, child] of Object.entries(value)) {
|
||||
state.remaining -= key.length;
|
||||
const next = cut(child, depth + 1, state);
|
||||
changed ||= next !== child;
|
||||
output[key] = next;
|
||||
}
|
||||
return changed ? output : value;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Returns the item itself when it fits its budget; otherwise a copy with its largest strings cut. */
|
||||
export function limitAgentTimelineItemContent(item: AgentTimelineItem): AgentTimelineItem {
|
||||
item = limitFailedShellError(item);
|
||||
item = limitPlainText(item);
|
||||
if (
|
||||
item.type !== "tool_call" ||
|
||||
item.detail.type !== "shell" ||
|
||||
typeof item.detail.output !== "string"
|
||||
) {
|
||||
return item;
|
||||
}
|
||||
if (item.detail.output.length <= TOOL_CALL_CONTENT_MAX_LENGTH) {
|
||||
return item;
|
||||
}
|
||||
return {
|
||||
...item,
|
||||
detail: {
|
||||
...item.detail,
|
||||
output: retainTextPrefix(item.detail.output, TOOL_CALL_CONTENT_MAX_LENGTH),
|
||||
},
|
||||
};
|
||||
const budget = item.type === "tool_call" ? TOOL_CALL_BUDGET : ITEM_BUDGET;
|
||||
const size: ItemSize = { nodeCost: 0, stringLengths: [], tooDeep: false };
|
||||
measure(item, 0, size);
|
||||
let stringTotal = 0;
|
||||
for (const length of size.stringLengths) stringTotal += length;
|
||||
if (!size.tooDeep && size.nodeCost + stringTotal <= budget) return item;
|
||||
|
||||
// Strings get what structure leaves, but at least half the budget, less a margin for JSON
|
||||
// escaping. Arrays drop the elements that do not fit, which bounds items whose short strings
|
||||
// or structure alone are too big.
|
||||
const markerCost =
|
||||
MARKER_COST * size.stringLengths.filter((length) => length > SHORT_STRING).length;
|
||||
const available = (budget * 15) / 16 - Math.min(size.nodeCost, budget / 2) - markerCost;
|
||||
const stringCap = Math.max(SHORT_STRING, findStringCap(size.stringLengths, available));
|
||||
return cut(item, 0, { stringCap, remaining: budget }) as AgentTimelineItem;
|
||||
}
|
||||
|
||||
@@ -181,6 +181,20 @@ describe("projected timeline retention", () => {
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("serves a merged assistant message within the item budget", () => {
|
||||
const store = new InMemoryAgentTimelineStore();
|
||||
store.initialize("agent", { epoch: "e" });
|
||||
const chunk = "a".repeat(200 * 1024);
|
||||
for (let index = 0; index < 10; index++) {
|
||||
store.append("agent", { type: "assistant_message", messageId: "m", text: chunk });
|
||||
}
|
||||
const [row] = store.fetch("agent", { limit: 1 }).rows;
|
||||
expect(row).toMatchObject({ seqStart: 1, seqEnd: 10 });
|
||||
expect(row.item).toMatchObject({ type: "assistant_message", messageId: "m" });
|
||||
expect(Buffer.byteLength(JSON.stringify(row.item))).toBeLessThanOrEqual(256 * 1024);
|
||||
expect(store.getLastAssistantMessage("agent")).toHaveLength(10 * chunk.length);
|
||||
});
|
||||
});
|
||||
|
||||
describe("projected sequence ownership", () => {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { TimelineProjection, type ProjectedTimelineRow } from "./timeline-projection.js";
|
||||
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
import type {
|
||||
AgentTimelineFetchOptions,
|
||||
@@ -120,7 +121,12 @@ export class InMemoryAgentTimelineStore {
|
||||
hasNewer: page.hasNewer,
|
||||
startSeq: page.startSeq,
|
||||
endSeq: page.endSeq,
|
||||
rows: page.entries.map((entry) => Object.assign({ seq: entry.seqEnd }, entry)),
|
||||
// Stored rows are already within budget; this bounds assistant and reasoning merges.
|
||||
rows: page.entries.map((entry) =>
|
||||
Object.assign({ seq: entry.seqEnd }, entry, {
|
||||
item: limitAgentTimelineItemContent(entry.item),
|
||||
}),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -80,7 +80,7 @@ describe("ProviderSubagentStore", () => {
|
||||
const [row] = subagents.fetchTimeline("parent-a", "child-1").rows;
|
||||
expect(row?.item).toMatchObject({
|
||||
type: "tool_call",
|
||||
detail: { type: "shell", output: "x".repeat(64 * 1024) },
|
||||
detail: { type: "shell", output: expect.stringMatching(/^x+…\[truncated \d+\.\d KB\]$/) },
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -466,13 +466,13 @@ describe("background Claude subagents", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "general-purpose",
|
||||
description: "Count files here",
|
||||
log: "[Bash] ls -1",
|
||||
log: "",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
test.each(["before", "after"])(
|
||||
"keeps one logged Task card when child tools start %s the parent finishes",
|
||||
"keeps one Task card when child tools start %s the parent finishes",
|
||||
async (timing) => {
|
||||
// A child can stream across the main session's turn boundary. Claude opens the next
|
||||
// main-session turn with system/init; bare child frames do not open a turn.
|
||||
@@ -605,7 +605,7 @@ describe("background Claude subagents", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "general-purpose",
|
||||
description: "Count files here",
|
||||
log: "[Bash] ls -1",
|
||||
log: "",
|
||||
},
|
||||
});
|
||||
},
|
||||
|
||||
@@ -737,7 +737,7 @@ describe("ClaudeAgentSession persisted subagent replay", () => {
|
||||
);
|
||||
}, 120_000);
|
||||
|
||||
test("limits retained child shell output to the timeline's existing content limit", async () => {
|
||||
test("limits retained child shell output to the timeline item budget", async () => {
|
||||
const output = "x".repeat(128 * 1024);
|
||||
writeSession({
|
||||
parentLines: [taskToolUse(), taskToolResult()],
|
||||
@@ -782,7 +782,10 @@ describe("ClaudeAgentSession persisted subagent replay", () => {
|
||||
type: "timeline",
|
||||
id: TOOL_USE_ID,
|
||||
item: expect.objectContaining({
|
||||
detail: expect.objectContaining({ type: "shell", output: output.slice(0, 64 * 1024) }),
|
||||
detail: expect.objectContaining({
|
||||
type: "shell",
|
||||
output: expect.stringMatching(/^x+…\[truncated \d+\.\d KB\]$/),
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
|
||||
+23
-25
@@ -273,7 +273,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
queryFactory.mockReset();
|
||||
});
|
||||
|
||||
test("accumulates lightweight sub_agent detail and preserves callId lifecycle collapse", async () => {
|
||||
test("labels the Task card from its input and preserves callId lifecycle collapse", async () => {
|
||||
const session = await new ClaudeAgentClient({
|
||||
logger,
|
||||
queryFactory,
|
||||
@@ -307,9 +307,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
|
||||
expect(latest.detail.subAgentType).toBe("Explore");
|
||||
expect(latest.detail.description).toBe("Inspect repository structure");
|
||||
expect(latest.detail.log).toContain("[Read] README.md");
|
||||
expect(latest.detail.log).toContain("[Edit] src/index.ts");
|
||||
expect(latest.detail.log).not.toContain("VERY_LARGE_OLD_STRING");
|
||||
expect(latest.detail.log).toBe("");
|
||||
|
||||
const rows: AgentTimelineRow[] = timelineToolCalls.map((item, index) => ({
|
||||
seq: index + 1,
|
||||
@@ -396,10 +394,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
.map((event) => event.item)
|
||||
.at(-1);
|
||||
|
||||
expect(latestSubAgentUpdate?.detail).toMatchObject({
|
||||
type: "sub_agent",
|
||||
log: expect.stringContaining("[Read] README.md"),
|
||||
});
|
||||
expect(latestSubAgentUpdate?.detail).toMatchObject({ type: "sub_agent", log: "" });
|
||||
});
|
||||
|
||||
test("routes a resumed SendMessage sidechain to the original child without a second Task card", async () => {
|
||||
@@ -553,7 +548,7 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
).toMatchObject({ type: "upsert", id: "task-tail-1", status: "failed" });
|
||||
});
|
||||
|
||||
test("tails sub-agent actions instead of dropping latest entries at cap", async () => {
|
||||
test("keeps the Task card the same size while the child runs", async () => {
|
||||
queryFactory.mockImplementation(() => buildQueryMock(buildTailScenarioEvents(205)));
|
||||
|
||||
const session = await new ClaudeAgentClient({
|
||||
@@ -568,24 +563,27 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
const events = await collectUntilTerminal(streamSession(session, "delegate work"));
|
||||
await session.close();
|
||||
|
||||
const timelineToolCalls = events
|
||||
.filter(
|
||||
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
|
||||
event.type === "timeline" && event.item.type === "tool_call",
|
||||
const runningCards = events
|
||||
.flatMap((event) =>
|
||||
event.type === "timeline" &&
|
||||
event.item.type === "tool_call" &&
|
||||
event.item.callId === "task-tail-1" &&
|
||||
event.item.status === "running" &&
|
||||
event.item.detail.type === "sub_agent"
|
||||
? [event.item]
|
||||
: [],
|
||||
)
|
||||
.map((event) => event.item)
|
||||
.filter((item) => item.callId === "task-tail-1");
|
||||
const subAgentUpdates = timelineToolCalls.filter((item) => item.detail.type === "sub_agent");
|
||||
const latest = subAgentUpdates[subAgentUpdates.length - 1];
|
||||
expect(latest).toBeDefined();
|
||||
if (!latest || latest.detail.type !== "sub_agent") {
|
||||
throw new Error("expected sub_agent detail");
|
||||
}
|
||||
.map((item) => JSON.stringify(item));
|
||||
expect(runningCards).toHaveLength(1);
|
||||
|
||||
expect(latest.detail.log).not.toContain("[Read] file-1.md");
|
||||
expect(latest.detail.log).not.toContain("[Read] file-5.md");
|
||||
expect(latest.detail.log).toContain("[Read] file-6.md");
|
||||
expect(latest.detail.log).toContain("[Read] file-205.md");
|
||||
const childToolCalls = events.filter(
|
||||
(event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "task-tail-1" &&
|
||||
event.event.item.type === "tool_call",
|
||||
);
|
||||
expect(childToolCalls).toHaveLength(205);
|
||||
});
|
||||
|
||||
test("keeps a long main-thread tool call out of the subagents track", async () => {
|
||||
|
||||
@@ -4667,7 +4667,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
/**
|
||||
* The Task card in the parent transcript, built from the declaration.
|
||||
*
|
||||
* The sidechain tracker also emits this card, enriched with the child's action log, and the two
|
||||
* The sidechain tracker also emits this card when it first sees the child, and the two
|
||||
* collapse on the shared tool-call id. Emitting it at declaration matters because a
|
||||
* backgrounded subagent produces no sidechain frames at all — verified on the wire — so the
|
||||
* tracker never runs for one and its card would otherwise stay an unlabeled "Task".
|
||||
@@ -6276,6 +6276,7 @@ function buildClaudeSubagentCardDetail(
|
||||
type: "sub_agent",
|
||||
...(facts.title ? { subAgentType: facts.title } : {}),
|
||||
...(facts.description ? { description: facts.description } : {}),
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
actions: [],
|
||||
};
|
||||
|
||||
@@ -4,7 +4,7 @@ import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
|
||||
import { ClaudeSidechainTracker } from "./sidechain-tracker.js";
|
||||
|
||||
describe("ClaudeSidechainTracker", () => {
|
||||
it("retains a background child's actions across the parent finishing", () => {
|
||||
it("keeps a background child's card across the parent finishing", () => {
|
||||
const tracker = new ClaudeSidechainTracker({
|
||||
getToolInput: () => ({ subagent_type: "general-purpose", description: "List files" }),
|
||||
isDescriptorOwnedElsewhere: () => true,
|
||||
@@ -15,27 +15,67 @@ describe("ClaudeSidechainTracker", () => {
|
||||
parent_tool_use_id: "task-1",
|
||||
message: { content: [{ type: "tool_use", id, name: "Bash", input: { command } }] },
|
||||
}) as unknown as SDKMessage;
|
||||
tracker.handleMessage(action("ls", "ls -1"), "task-1");
|
||||
const parentCards = (events: ReturnType<typeof tracker.handleMessage>) =>
|
||||
events.filter((event) => event.type === "timeline");
|
||||
const childCallIds = (events: ReturnType<typeof tracker.handleMessage>) =>
|
||||
events.flatMap((event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.item.type === "tool_call"
|
||||
? [event.event.item.callId]
|
||||
: [],
|
||||
);
|
||||
|
||||
expect(parentCards(tracker.handleMessage(action("ls", "ls -1"), "task-1"))).toHaveLength(1);
|
||||
expect(tracker.finishAll("completed")).toEqual([]);
|
||||
const events = tracker.handleMessage(action("pwd", "pwd"), "task-1");
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "timeline",
|
||||
item: expect.objectContaining({
|
||||
detail: expect.objectContaining({ log: "[Bash] ls -1\n[Bash] pwd" }),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
const afterFinish = tracker.handleMessage(action("pwd", "pwd"), "task-1");
|
||||
expect(parentCards(afterFinish)).toEqual([]);
|
||||
expect(childCallIds(afterFinish)).toEqual(["pwd"]);
|
||||
|
||||
tracker.delete("task-1");
|
||||
const next = tracker.handleMessage(action("next", "ls"), "task-1");
|
||||
expect(next).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "timeline",
|
||||
item: expect.objectContaining({
|
||||
detail: expect.objectContaining({ log: "[Bash] ls" }),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
expect(parentCards(next)).toHaveLength(1);
|
||||
expect(childCallIds(next)).toEqual(["next"]);
|
||||
});
|
||||
|
||||
it("shows a streamed child tool call's input once the assistant message brings it", () => {
|
||||
const tracker = new ClaudeSidechainTracker({ getToolInput: () => null });
|
||||
const frames = [
|
||||
{
|
||||
type: "stream_event",
|
||||
parent_tool_use_id: "task-1",
|
||||
event: {
|
||||
type: "content_block_start",
|
||||
index: 0,
|
||||
content_block: { type: "tool_use", id: "bash-1", name: "Bash", input: {} },
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "assistant",
|
||||
parent_tool_use_id: "task-1",
|
||||
message: {
|
||||
content: [
|
||||
{ type: "tool_use", id: "bash-1", name: "Bash", input: { command: "echo child" } },
|
||||
],
|
||||
},
|
||||
},
|
||||
] as unknown as SDKMessage[];
|
||||
|
||||
const childCards = frames
|
||||
.flatMap((frame) => tracker.handleMessage(frame, "task-1"))
|
||||
.flatMap((event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.item.type === "tool_call"
|
||||
? [event.event.item]
|
||||
: [],
|
||||
);
|
||||
|
||||
expect(childCards.at(-1)).toMatchObject({
|
||||
callId: "bash-1",
|
||||
status: "running",
|
||||
detail: { type: "shell", command: "echo child" },
|
||||
});
|
||||
});
|
||||
|
||||
it("uses Claude's native agent name for the provider subagent title", () => {
|
||||
|
||||
@@ -5,7 +5,6 @@ import {
|
||||
mapClaudeFailedToolCall,
|
||||
mapClaudeRunningToolCall,
|
||||
} from "./tool-call-mapper.js";
|
||||
import { buildToolCallDisplayModel } from "@getpaseo/protocol/tool-call-display";
|
||||
import {
|
||||
isClaudeSubagentHandbackToolName,
|
||||
readClaudeSubagentHandback,
|
||||
@@ -20,21 +19,17 @@ interface ClaudeContentChunk {
|
||||
|
||||
type ClaudeToolCallItem = Extract<AgentTimelineItem, { type: "tool_call" }>;
|
||||
|
||||
interface SubAgentActionEntry {
|
||||
index: number;
|
||||
interface SubAgentChildCall {
|
||||
toolName: string;
|
||||
input: unknown;
|
||||
summary?: string;
|
||||
}
|
||||
|
||||
interface SubAgentActivityState {
|
||||
name?: string;
|
||||
subAgentType?: string;
|
||||
description?: string;
|
||||
actions: SubAgentActionEntry[];
|
||||
actionKeys: string[];
|
||||
nextActionIndex: number;
|
||||
actionIndexByKey: Map<string, number>;
|
||||
/** The child's tool calls awaiting results, which carry no input of their own. */
|
||||
runningCalls: Map<string, SubAgentChildCall>;
|
||||
completedActionKeys: Set<string>;
|
||||
}
|
||||
|
||||
@@ -44,7 +39,6 @@ interface SubAgentActionCandidate {
|
||||
input: unknown;
|
||||
}
|
||||
|
||||
const MAX_SUB_AGENT_LOG_ENTRIES = 200;
|
||||
const MAX_SUB_AGENT_SUMMARY_CHARS = 160;
|
||||
|
||||
function readTrimmedString(value: unknown): string | undefined {
|
||||
@@ -79,8 +73,7 @@ export class ClaudeSidechainTracker {
|
||||
needsSyntheticParentToolCard?: (toolUseId: string) => boolean;
|
||||
/**
|
||||
* The parent's tool call once Claude has answered it. A background child streams its frames
|
||||
* after that answer, so its card keeps the settled status and output and gains the child's
|
||||
* action log.
|
||||
* after that answer, so its card keeps the settled status and output.
|
||||
*/
|
||||
getSettledParentToolCall?: (toolUseId: string) => ClaudeToolCallItem | undefined;
|
||||
}) {
|
||||
@@ -91,28 +84,27 @@ export class ClaudeSidechainTracker {
|
||||
}
|
||||
|
||||
handleMessage(message: SDKMessage, parentToolUseId: string): AgentStreamEvent[] {
|
||||
const existingState = this.activeSidechains.get(parentToolUseId);
|
||||
const state =
|
||||
this.activeSidechains.get(parentToolUseId) ??
|
||||
existingState ??
|
||||
({
|
||||
actions: [],
|
||||
actionKeys: [],
|
||||
nextActionIndex: 1,
|
||||
actionIndexByKey: new Map<string, number>(),
|
||||
runningCalls: new Map<string, SubAgentChildCall>(),
|
||||
completedActionKeys: new Set<string>(),
|
||||
} satisfies SubAgentActivityState);
|
||||
this.activeSidechains.set(parentToolUseId, state);
|
||||
|
||||
const contextUpdated = this.updateSubAgentContextFromTaskInput(state, parentToolUseId);
|
||||
// The parent card is a launch marker: it changes only when the child is first seen or the
|
||||
// task input names it differently, never with the child's activity.
|
||||
const cardUpdated =
|
||||
this.updateSubAgentContextFromTaskInput(state, parentToolUseId) || !existingState;
|
||||
const actionCandidates = this.extractSubAgentActionCandidates(message);
|
||||
const childTimelineItems = [
|
||||
...this.extractSubAgentTimelineItems(message, state),
|
||||
...this.extractSubAgentToolResults(message, state),
|
||||
];
|
||||
let actionUpdated = false;
|
||||
for (const action of actionCandidates) {
|
||||
if (state.completedActionKeys.has(action.key)) continue;
|
||||
if (this.appendSubAgentAction(state, action)) {
|
||||
actionUpdated = true;
|
||||
if (this.trackChildCall(state, action)) {
|
||||
const toolCall = mapClaudeRunningToolCall({
|
||||
name: action.toolName,
|
||||
callId: action.key,
|
||||
@@ -125,7 +117,7 @@ export class ClaudeSidechainTracker {
|
||||
}
|
||||
}
|
||||
|
||||
if (!contextUpdated && !actionUpdated && childTimelineItems.length === 0) {
|
||||
if (!cardUpdated && childTimelineItems.length === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
@@ -145,11 +137,8 @@ export class ClaudeSidechainTracker {
|
||||
type: "sub_agent",
|
||||
...(state.subAgentType ? { subAgentType: state.subAgentType } : {}),
|
||||
...(state.description ? { description: state.description } : {}),
|
||||
log: state.actions
|
||||
.map((action) =>
|
||||
action.summary ? `[${action.toolName}] ${action.summary}` : `[${action.toolName}]`,
|
||||
)
|
||||
.join("\n"),
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
actions: [],
|
||||
};
|
||||
|
||||
@@ -170,18 +159,19 @@ export class ClaudeSidechainTracker {
|
||||
},
|
||||
];
|
||||
|
||||
const parentCard: AgentStreamEvent[] = this.needsSyntheticParentToolCard(parentToolUseId)
|
||||
? [
|
||||
{
|
||||
type: "timeline",
|
||||
item: {
|
||||
...toolCall,
|
||||
detail,
|
||||
const parentCard: AgentStreamEvent[] =
|
||||
cardUpdated && this.needsSyntheticParentToolCard(parentToolUseId)
|
||||
? [
|
||||
{
|
||||
type: "timeline",
|
||||
item: {
|
||||
...toolCall,
|
||||
detail,
|
||||
},
|
||||
provider: "claude",
|
||||
},
|
||||
provider: "claude",
|
||||
},
|
||||
]
|
||||
: [];
|
||||
]
|
||||
: [];
|
||||
|
||||
return [
|
||||
...descriptorEvents,
|
||||
@@ -303,8 +293,7 @@ export class ClaudeSidechainTracker {
|
||||
if (!isClaudeContentChunk(block) || !block.type.endsWith("tool_result")) continue;
|
||||
const callId = readTrimmedString(block.tool_use_id);
|
||||
if (!callId || state.completedActionKeys.has(callId)) continue;
|
||||
const actionIndex = state.actionIndexByKey.get(callId);
|
||||
const action = actionIndex === undefined ? undefined : state.actions[actionIndex];
|
||||
const action = state.runningCalls.get(callId);
|
||||
const toolName = action?.toolName ?? readTrimmedString(block.tool_name);
|
||||
if (!toolName) continue;
|
||||
const params = {
|
||||
@@ -317,6 +306,7 @@ export class ClaudeSidechainTracker {
|
||||
? mapClaudeFailedToolCall({ ...params, error: block })
|
||||
: mapClaudeCompletedToolCall(params);
|
||||
if (toolCall) {
|
||||
state.runningCalls.delete(callId);
|
||||
state.completedActionKeys.add(callId);
|
||||
items.push(toolCall);
|
||||
}
|
||||
@@ -412,13 +402,12 @@ export class ClaudeSidechainTracker {
|
||||
const key =
|
||||
readTrimmedString(block.id) ??
|
||||
`stream:${block.name}:${typeof event.index === "number" ? event.index : 0}`;
|
||||
return [
|
||||
{
|
||||
key,
|
||||
toolName: block.name,
|
||||
input: block.input ?? null,
|
||||
},
|
||||
];
|
||||
// A streamed call starts with `{}` until its input arrives, so an empty input is not final.
|
||||
const input =
|
||||
block.input && typeof block.input === "object" && Object.keys(block.input).length > 0
|
||||
? block.input
|
||||
: null;
|
||||
return [{ key, toolName: block.name, input }];
|
||||
}
|
||||
|
||||
private extractSubAgentActionCandidates(message: SDKMessage): SubAgentActionCandidate[] {
|
||||
@@ -448,84 +437,20 @@ export class ClaudeSidechainTracker {
|
||||
return [];
|
||||
}
|
||||
|
||||
private appendSubAgentAction(
|
||||
/** Returns whether the call is new or has gained its input, so its running card changes. */
|
||||
private trackChildCall(
|
||||
state: SubAgentActivityState,
|
||||
candidate: SubAgentActionCandidate,
|
||||
): boolean {
|
||||
const normalizedToolName = readTrimmedString(candidate.toolName);
|
||||
if (!normalizedToolName) {
|
||||
const toolName = readTrimmedString(candidate.toolName);
|
||||
if (!toolName) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const summary = this.deriveSubAgentActionSummary(normalizedToolName, candidate.input);
|
||||
const existingIndex = state.actionIndexByKey.get(candidate.key);
|
||||
|
||||
if (existingIndex !== undefined) {
|
||||
const existing = state.actions[existingIndex];
|
||||
if (!existing) {
|
||||
return false;
|
||||
}
|
||||
const nextSummary = existing.summary ?? summary;
|
||||
if (existing.toolName === normalizedToolName && existing.summary === nextSummary) {
|
||||
return false;
|
||||
}
|
||||
state.actions[existingIndex] = {
|
||||
...existing,
|
||||
toolName: normalizedToolName,
|
||||
input: existing.input ?? candidate.input,
|
||||
...(nextSummary ? { summary: nextSummary } : {}),
|
||||
};
|
||||
return true;
|
||||
const existing = state.runningCalls.get(candidate.key);
|
||||
if (existing && (existing.input !== null || candidate.input === null)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
state.actions.push({
|
||||
index: state.nextActionIndex,
|
||||
toolName: normalizedToolName,
|
||||
input: candidate.input,
|
||||
...(summary ? { summary } : {}),
|
||||
});
|
||||
state.nextActionIndex += 1;
|
||||
state.actionKeys.push(candidate.key);
|
||||
this.trimSubAgentTail(state);
|
||||
this.rebuildSubAgentActionIndex(state);
|
||||
state.runningCalls.set(candidate.key, { toolName, input: candidate.input });
|
||||
return true;
|
||||
}
|
||||
|
||||
private trimSubAgentTail(state: SubAgentActivityState): void {
|
||||
while (state.actions.length > MAX_SUB_AGENT_LOG_ENTRIES) {
|
||||
state.actions.shift();
|
||||
const removedKey = state.actionKeys.shift();
|
||||
if (removedKey) state.completedActionKeys.delete(removedKey);
|
||||
}
|
||||
}
|
||||
|
||||
private rebuildSubAgentActionIndex(state: SubAgentActivityState): void {
|
||||
state.actionIndexByKey.clear();
|
||||
for (let index = 0; index < state.actionKeys.length; index += 1) {
|
||||
const key = state.actionKeys[index];
|
||||
if (key) {
|
||||
state.actionIndexByKey.set(key, index);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private deriveSubAgentActionSummary(toolName: string, input: unknown): string | undefined {
|
||||
const runningToolCall = mapClaudeRunningToolCall({
|
||||
name: toolName,
|
||||
callId: `sub-agent-summary-${toolName}`,
|
||||
input,
|
||||
output: null,
|
||||
});
|
||||
if (!runningToolCall) {
|
||||
return undefined;
|
||||
}
|
||||
const display = buildToolCallDisplayModel({
|
||||
name: runningToolCall.name,
|
||||
status: runningToolCall.status,
|
||||
error: runningToolCall.error,
|
||||
detail: runningToolCall.detail,
|
||||
metadata: runningToolCall.metadata,
|
||||
});
|
||||
return this.normalizeSubAgentText(display.summary);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -77,12 +77,19 @@ CHILD_SENTINEL in your own response.
|
||||
|
||||
const completedChildIndex = events.findIndex(
|
||||
(event) =>
|
||||
event.type === "timeline" &&
|
||||
event.item.type === "tool_call" &&
|
||||
event.item.status === "completed" &&
|
||||
event.item.detail.type === "sub_agent" &&
|
||||
event.item.detail.log.includes("CHILD_SENTINEL"),
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "upsert" &&
|
||||
event.event.status === "completed",
|
||||
);
|
||||
const childAssistantText = events
|
||||
.flatMap((event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.item.type === "assistant_message"
|
||||
? [event.event.item.text]
|
||||
: [],
|
||||
)
|
||||
.join("");
|
||||
const rootMessageIndexes = events.flatMap((event, index) =>
|
||||
event.type === "timeline" && event.item.type === "assistant_message" ? [index] : [],
|
||||
);
|
||||
@@ -105,6 +112,7 @@ CHILD_SENTINEL in your own response.
|
||||
.map((item) => item.text)
|
||||
.join("");
|
||||
expect(topLevelAssistantText).not.toContain("CHILD_SENTINEL");
|
||||
expect(childAssistantText).toContain("CHILD_SENTINEL");
|
||||
expect(completedChildIndex).toBeGreaterThanOrEqual(0);
|
||||
expect(rootMessageIndexes[0]).toBeGreaterThan(completedChildIndex);
|
||||
expect(terminalIndexes).toHaveLength(1);
|
||||
|
||||
@@ -3524,7 +3524,7 @@ describe("Codex app-server provider", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
test("folds child-thread Codex activity into the parent sub-agent tool call", async () => {
|
||||
test("keeps child-thread Codex activity out of the parent sub-agent card", async () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
@@ -3562,7 +3562,6 @@ describe("Codex app-server provider", () => {
|
||||
});
|
||||
|
||||
const timelineEvents = events.filter((event) => event.type === "timeline");
|
||||
expect(timelineEvents).toHaveLength(4);
|
||||
expect(timelineEvents.every((event) => event.item.type === "tool_call")).toBe(true);
|
||||
const finalItem = timelineEvents.at(-1)?.item;
|
||||
expect(finalItem).toMatchObject({
|
||||
@@ -3574,7 +3573,7 @@ describe("Codex app-server provider", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "Sub-agent",
|
||||
description: "Report findings.",
|
||||
log: "[Assistant] Found the path.",
|
||||
log: "",
|
||||
actions: [],
|
||||
},
|
||||
});
|
||||
@@ -3651,13 +3650,9 @@ describe("Codex app-server provider", () => {
|
||||
id: "settled-child-thread",
|
||||
status: "completed",
|
||||
});
|
||||
expect(events.at(-1)).toMatchObject({
|
||||
expect(events.findLast((event) => event.type === "timeline")).toMatchObject({
|
||||
type: "timeline",
|
||||
item: {
|
||||
callId: "call-settled-child",
|
||||
status: "completed",
|
||||
detail: { type: "sub_agent", log: "[Assistant] Late trailing output." },
|
||||
},
|
||||
item: { callId: "call-settled-child", status: "completed" },
|
||||
});
|
||||
|
||||
appServer.completesSubAgentActivity({
|
||||
@@ -3962,6 +3957,8 @@ describe("Codex app-server provider", () => {
|
||||
createTestLogger(),
|
||||
async () => appServer.child,
|
||||
);
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
|
||||
try {
|
||||
const resultPromise = session.run("Delegate the investigation, then report the result.");
|
||||
@@ -4005,11 +4002,18 @@ describe("Codex app-server provider", () => {
|
||||
type: "tool_call",
|
||||
callId: "spawn-child-early",
|
||||
status: "completed",
|
||||
detail: {
|
||||
type: "sub_agent",
|
||||
log: "[Assistant] Early child findings.",
|
||||
},
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
});
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "provider_subagent",
|
||||
event: expect.objectContaining({
|
||||
type: "timeline",
|
||||
id: "child-thread-early",
|
||||
item: expect.objectContaining({ type: "assistant_message" }),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
appServer.assertNoErrors();
|
||||
} finally {
|
||||
await session.close();
|
||||
@@ -4080,10 +4084,7 @@ describe("Codex app-server provider", () => {
|
||||
expect(toolCalls.at(-1)).toMatchObject({
|
||||
callId: "spawn-child-stable",
|
||||
status: "canceled",
|
||||
detail: {
|
||||
type: "sub_agent",
|
||||
log: "[Assistant] Still on the same card.",
|
||||
},
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
});
|
||||
appServer.assertNoErrors();
|
||||
} finally {
|
||||
@@ -4163,11 +4164,10 @@ describe("Codex app-server provider", () => {
|
||||
|
||||
const result = await resultPromise;
|
||||
const toolCalls = result.timeline.filter((item) => item.type === "tool_call");
|
||||
expect(toolCalls.map((item) => item.status)).toEqual(["running", "completed", "completed"]);
|
||||
expect(toolCalls.map((item) => item.status)).toEqual(["running", "completed"]);
|
||||
expect(toolCalls.at(-1)).toMatchObject({
|
||||
callId: "spawn-child-late-compaction",
|
||||
status: "completed",
|
||||
detail: { type: "sub_agent", log: "[Compacted]" },
|
||||
});
|
||||
appServer.assertNoErrors();
|
||||
} finally {
|
||||
@@ -4175,7 +4175,7 @@ describe("Codex app-server provider", () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("projects legacy child tools into one stable sub-agent log", async () => {
|
||||
test("routes legacy child tools to the child timeline", async () => {
|
||||
const appServer = createFakeCodexAppServer();
|
||||
const session = new CodexAppServerAgentSession(
|
||||
createConfig({ cwd: "/workspace/project" }),
|
||||
@@ -4183,6 +4183,8 @@ describe("Codex app-server provider", () => {
|
||||
createTestLogger(),
|
||||
async () => appServer.child,
|
||||
);
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
|
||||
try {
|
||||
const resultPromise = session.run("Delegate the implementation.");
|
||||
@@ -4225,19 +4227,26 @@ describe("Codex app-server provider", () => {
|
||||
expect(finalToolCall).toMatchObject({
|
||||
callId: "spawn-legacy-tool-child",
|
||||
status: "completed",
|
||||
detail: { type: "sub_agent" },
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
});
|
||||
if (finalToolCall?.detail.type === "sub_agent") {
|
||||
expect(finalToolCall.detail.log.match(/\[Shell\]/g)).toHaveLength(1);
|
||||
expect(finalToolCall.detail.log).toContain("[Edit]");
|
||||
}
|
||||
const childToolCalls = events.flatMap((event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "legacy-tool-child" &&
|
||||
event.event.item.type === "tool_call"
|
||||
? [event.event.item]
|
||||
: [],
|
||||
);
|
||||
expect(new Set(childToolCalls.map((item) => item.detail.type))).toEqual(
|
||||
new Set(["shell", "edit"]),
|
||||
);
|
||||
appServer.assertNoErrors();
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps nested MultiAgentV2 output inside the root sub-agent card", async () => {
|
||||
test("keeps nested MultiAgentV2 output out of the root sub-agent card", async () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
@@ -4297,8 +4306,22 @@ describe("Codex app-server provider", () => {
|
||||
expect(beforeParentCompletes.at(-1)).toMatchObject({
|
||||
callId: "spawn-child-root",
|
||||
status: "running",
|
||||
detail: { type: "sub_agent", log: expect.stringContaining("Grandchild findings.") },
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
});
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "provider_subagent",
|
||||
event: {
|
||||
type: "timeline",
|
||||
id: "grandchild-thread",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "grandchild-message",
|
||||
text: "Grandchild findings.",
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
await asInternals(session).handleNotification("turn/completed", {
|
||||
threadId: "child-thread-root",
|
||||
@@ -4365,13 +4388,20 @@ describe("Codex app-server provider", () => {
|
||||
"child-thread-root",
|
||||
"child-thread-root",
|
||||
]);
|
||||
expect(events.at(-1)).toMatchObject({
|
||||
type: "timeline",
|
||||
item: {
|
||||
callId: "spawn-child-root",
|
||||
detail: { type: "sub_agent", log: expect.stringContaining("Still nested.") },
|
||||
},
|
||||
});
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "provider_subagent",
|
||||
event: {
|
||||
type: "timeline",
|
||||
id: "grandchild-thread",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "grandchild-after-wait",
|
||||
text: "Still nested.",
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
test("never treats an unmapped foreign terminal as the root terminal", async () => {
|
||||
@@ -4791,7 +4821,7 @@ describe("Codex app-server provider", () => {
|
||||
expect(events).toHaveLength(eventCountAfterSpawn);
|
||||
});
|
||||
|
||||
test("keeps child context compaction inside the child card", async () => {
|
||||
test("keeps child context compaction on the child timeline", async () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
@@ -4824,10 +4854,70 @@ describe("Codex app-server provider", () => {
|
||||
(item) => item.type === "tool_call" && item.callId === "spawn-child-compaction",
|
||||
),
|
||||
).toBe(true);
|
||||
expect(timelineItems.at(-1)).toMatchObject({
|
||||
type: "tool_call",
|
||||
detail: { type: "sub_agent", log: "[Compacted]" },
|
||||
expect(
|
||||
events.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "timeline" ? [event.event] : [],
|
||||
),
|
||||
).toEqual([
|
||||
{
|
||||
type: "timeline",
|
||||
id: "child-thread-compaction",
|
||||
item: { type: "compaction", status: "loading" },
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
id: "child-thread-compaction",
|
||||
item: { type: "compaction", status: "completed" },
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("settles a child compaction once when the item completion is missing or duplicated", async () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
|
||||
for (const child of ["compacted-child", "turn-ended-child"]) {
|
||||
await asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "test-thread",
|
||||
item: {
|
||||
type: "subAgentActivity",
|
||||
id: `spawn-${child}`,
|
||||
kind: "started",
|
||||
agentThreadId: child,
|
||||
agentPath: `/root/${child}`,
|
||||
},
|
||||
});
|
||||
await asInternals(session).handleNotification("item/started", {
|
||||
threadId: child,
|
||||
item: { type: "contextCompaction", id: `${child}-compaction` },
|
||||
});
|
||||
}
|
||||
await asInternals(session).handleNotification("thread/compacted", {
|
||||
threadId: "compacted-child",
|
||||
});
|
||||
await asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "compacted-child",
|
||||
item: { type: "contextCompaction", id: "compacted-child-compaction" },
|
||||
});
|
||||
await asInternals(session).handleNotification("turn/completed", {
|
||||
threadId: "turn-ended-child",
|
||||
turn: { status: "completed" },
|
||||
});
|
||||
|
||||
const compactions = events.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "timeline"
|
||||
? [
|
||||
`${event.event.id} ${event.event.item.type === "compaction" ? event.event.item.status : ""}`,
|
||||
]
|
||||
: [],
|
||||
);
|
||||
expect(compactions).toEqual([
|
||||
"compacted-child loading",
|
||||
"turn-ended-child loading",
|
||||
"compacted-child completed",
|
||||
"turn-ended-child completed",
|
||||
]);
|
||||
});
|
||||
|
||||
test("keeps the parent sub-agent running when a child command fails during the child turn", async () => {
|
||||
@@ -4862,7 +4952,7 @@ describe("Codex app-server provider", () => {
|
||||
},
|
||||
});
|
||||
|
||||
expect(events.at(-1)?.item).toMatchObject({
|
||||
expect(events.findLast((event) => event.type === "timeline")?.item).toMatchObject({
|
||||
type: "tool_call",
|
||||
callId: "call-sub-agent-child-command-failure",
|
||||
name: "Sub-agent",
|
||||
@@ -5157,7 +5247,7 @@ describe("Codex app-server provider", () => {
|
||||
);
|
||||
expect(liveToolCalls.at(-1)).toMatchObject({
|
||||
status: "running",
|
||||
detail: { type: "sub_agent", log: "[Assistant] More findings after resume." },
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
});
|
||||
|
||||
liveEvents.length = 0;
|
||||
@@ -5170,12 +5260,12 @@ describe("Codex app-server provider", () => {
|
||||
itemId: "legacy-child-message-after-resume",
|
||||
delta: "Legacy findings after resume.",
|
||||
});
|
||||
expect(liveEvents.at(-1)).toMatchObject({
|
||||
expect(liveEvents.findLast((event) => event.type === "timeline")).toMatchObject({
|
||||
type: "timeline",
|
||||
item: {
|
||||
callId: "legacy-spawn-history",
|
||||
status: "running",
|
||||
detail: { type: "sub_agent", log: "[Assistant] Legacy findings after resume." },
|
||||
detail: { type: "sub_agent", log: "" },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
@@ -49,10 +49,10 @@ import { Dirent } from "node:fs";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { z } from "zod";
|
||||
import { renderPromptAttachmentAsText } from "../prompt-attachments.js";
|
||||
import { composeSystemPromptParts } from "../system-prompt.js";
|
||||
import { curateAgentActivity } from "../activity-curator.js";
|
||||
import { CodexAsyncQuestions, codexAsyncQuestionToTimeline } from "./codex/async-questions.js";
|
||||
import {
|
||||
mapCodexToolCallEnvelope,
|
||||
@@ -3384,8 +3384,6 @@ interface CodexSubAgentCallState {
|
||||
activityItemIds: Set<string>;
|
||||
pendingCommandOutputDeltas: Map<string, string[]>;
|
||||
pendingFileChangeOutputDeltas: Map<string, string[]>;
|
||||
childItemOrder: string[];
|
||||
childItems: Map<string, AgentTimelineItem>;
|
||||
childThreadIds: Set<string>;
|
||||
}
|
||||
|
||||
@@ -3487,6 +3485,9 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
private compactionTriggerByItemId = new Map<string, "auto" | "manual">();
|
||||
private pendingRootCompactionItemIds = new Set<string>();
|
||||
private pendingAnonymousRootCompactions = 0;
|
||||
// Child threads with a compaction shown as loading. Some Codex builds settle it only through
|
||||
// `thread/compacted` or the end of the child's turn.
|
||||
private pendingChildCompactionThreadIds = new Set<string>();
|
||||
// Codex can report one completed compaction through both channels:
|
||||
// `thread/compacted` and a completed `contextCompaction` item.
|
||||
private unpairedCompactionNotificationCompletions = 0;
|
||||
@@ -5516,6 +5517,9 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
case "patch_apply_completed":
|
||||
this.handlePatchApplyCompletedNotification(parsed, callId);
|
||||
return;
|
||||
case "context_compacted":
|
||||
this.settleChildCompaction(parsed.threadId);
|
||||
return;
|
||||
default:
|
||||
// Aggregate child telemetry is redundant and must not leak into the
|
||||
// root timeline. Concrete legacy tools are projected above for Codex
|
||||
@@ -5692,20 +5696,10 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
activityItemIds: new Set<string>(),
|
||||
pendingCommandOutputDeltas: new Map<string, string[]>(),
|
||||
pendingFileChangeOutputDeltas: new Map<string, string[]>(),
|
||||
childItemOrder: [],
|
||||
childItems: new Map<string, AgentTimelineItem>(),
|
||||
childThreadIds: new Set<string>(),
|
||||
} satisfies CodexSubAgentCallState);
|
||||
|
||||
state.toolCall = {
|
||||
...timelineItem,
|
||||
detail: {
|
||||
...timelineItem.detail,
|
||||
log:
|
||||
timelineItem.detail.log ||
|
||||
(state.toolCall.detail.type === "sub_agent" ? state.toolCall.detail.log : ""),
|
||||
},
|
||||
};
|
||||
state.toolCall = timelineItem;
|
||||
state.parentCallId ??= parentCallId;
|
||||
state.parentSubagentId ??= parentSubagentId;
|
||||
const activity = readCodexSubAgentActivity(rawItem);
|
||||
@@ -5758,25 +5752,17 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
const activityToolCall = mapCodexToolCallFromThreadItem(rawItem, {
|
||||
cwd: this.config.cwd ?? null,
|
||||
});
|
||||
if (
|
||||
activityToolCall?.detail.type === "sub_agent" &&
|
||||
state.toolCall.detail.type === "sub_agent"
|
||||
) {
|
||||
state.toolCall = {
|
||||
...state.toolCall,
|
||||
detail: {
|
||||
...state.toolCall.detail,
|
||||
subAgentType: activityToolCall.detail.subAgentType,
|
||||
},
|
||||
};
|
||||
}
|
||||
const subAgentType =
|
||||
activityToolCall?.detail.type === "sub_agent"
|
||||
? activityToolCall.detail.subAgentType
|
||||
: undefined;
|
||||
let nextStatus: ToolCallTimelineItem["status"] | undefined = "running";
|
||||
if (activity.kind === "interrupted") {
|
||||
nextStatus = "canceled";
|
||||
} else if (isTerminalSubAgentStatus(state.toolCall.status)) {
|
||||
nextStatus = undefined;
|
||||
}
|
||||
this.emitSubAgentActivityUpdate(callId, nextStatus);
|
||||
this.emitSubAgentActivityUpdate(callId, nextStatus, { subAgentType });
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -5817,7 +5803,13 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
): boolean {
|
||||
if (
|
||||
childSubAgentCallId &&
|
||||
this.handleSubAgentContextCompactionItem(childSubAgentCallId, parsed.item, "completed")
|
||||
parsed.threadId &&
|
||||
this.handleSubAgentContextCompactionItem(
|
||||
childSubAgentCallId,
|
||||
parsed.threadId,
|
||||
parsed.item,
|
||||
"completed",
|
||||
)
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
@@ -5827,17 +5819,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
);
|
||||
}
|
||||
|
||||
private upsertSubAgentChildItem(callId: string, itemId: string, item: AgentTimelineItem): void {
|
||||
const state = this.subAgentCallsByCallId.get(callId);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
if (!state.childItems.has(itemId)) {
|
||||
state.childItemOrder.push(itemId);
|
||||
}
|
||||
state.childItems.set(itemId, item);
|
||||
}
|
||||
|
||||
private emitCodexToolTimelineItem(
|
||||
timelineItem: ToolCallTimelineItem,
|
||||
subAgentCallId: string | null,
|
||||
@@ -5847,7 +5828,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
|
||||
return;
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, timelineItem.callId, timelineItem);
|
||||
const state = this.subAgentCallsByCallId.get(subAgentCallId);
|
||||
if (state) {
|
||||
const targetThreadIds = childThreadId ? [childThreadId] : state.childThreadIds;
|
||||
@@ -5861,26 +5841,15 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
);
|
||||
}
|
||||
|
||||
private getSubAgentChildTimeline(state: CodexSubAgentCallState): AgentTimelineItem[] {
|
||||
return state.childItemOrder
|
||||
.map((itemId) => state.childItems.get(itemId))
|
||||
.filter((item): item is AgentTimelineItem => Boolean(item));
|
||||
}
|
||||
|
||||
private emitSubAgentActivityUpdate(
|
||||
callId: string,
|
||||
status?: ToolCallTimelineItem["status"],
|
||||
options?: { reopen?: boolean },
|
||||
options?: { reopen?: boolean; subAgentType?: string },
|
||||
): void {
|
||||
const state = this.subAgentCallsByCallId.get(callId);
|
||||
if (!state || state.toolCall.detail.type !== "sub_agent") {
|
||||
return;
|
||||
}
|
||||
const childTimeline = this.getSubAgentChildTimeline(state);
|
||||
const log =
|
||||
childTimeline.length > 0
|
||||
? curateAgentActivity(childTimeline, { labelAssistantMessages: true })
|
||||
: "";
|
||||
let resolvedStatus = status ?? state.toolCall.status;
|
||||
if (
|
||||
status === "running" &&
|
||||
@@ -5889,16 +5858,12 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
) {
|
||||
resolvedStatus = state.toolCall.status;
|
||||
}
|
||||
for (const childThreadId of state.childThreadIds) {
|
||||
this.emitProviderSubagentUpsert(childThreadId, state, resolvedStatus);
|
||||
}
|
||||
const baseToolCall = {
|
||||
...state.toolCall,
|
||||
detail: {
|
||||
...state.toolCall.detail,
|
||||
log,
|
||||
},
|
||||
};
|
||||
const baseToolCall = options?.subAgentType
|
||||
? {
|
||||
...state.toolCall,
|
||||
detail: { ...state.toolCall.detail, subAgentType: options.subAgentType },
|
||||
}
|
||||
: state.toolCall;
|
||||
const nextToolCall: ToolCallTimelineItem =
|
||||
resolvedStatus === "failed"
|
||||
? {
|
||||
@@ -5911,10 +5876,14 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
status: resolvedStatus,
|
||||
error: null,
|
||||
};
|
||||
// The card is a launch marker: the child's activity streams to its own timeline, so the card
|
||||
// is re-sent only when its status or label changes.
|
||||
const cardChanged = !isDeepStrictEqual(nextToolCall, state.toolCall);
|
||||
state.toolCall = nextToolCall;
|
||||
if (state.parentCallId && state.parentCallId !== callId) {
|
||||
this.upsertSubAgentChildItem(state.parentCallId, state.callId, nextToolCall);
|
||||
this.emitSubAgentActivityUpdate(state.parentCallId);
|
||||
for (const childThreadId of state.childThreadIds) {
|
||||
this.emitProviderSubagentUpsert(childThreadId, state, resolvedStatus);
|
||||
}
|
||||
if (!cardChanged || (state.parentCallId && state.parentCallId !== callId)) {
|
||||
return;
|
||||
}
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: nextToolCall });
|
||||
@@ -6022,7 +5991,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
): void {
|
||||
this.applyBufferedDeltaTextToTimelineItem(timelineItem, itemId);
|
||||
if (itemId) {
|
||||
this.upsertSubAgentChildItem(callId, itemId, timelineItem);
|
||||
this.pendingAgentMessages.delete(itemId);
|
||||
this.pendingReasoning.delete(itemId);
|
||||
this.pendingCommandOutputDeltas.delete(itemId);
|
||||
@@ -6033,19 +6001,29 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
|
||||
private handleSubAgentContextCompactionItem(
|
||||
callId: string,
|
||||
childThreadId: string,
|
||||
item: { id?: string; type?: string; [key: string]: unknown },
|
||||
status: "loading" | "completed",
|
||||
): boolean {
|
||||
if (!this.isContextCompactionItem(item)) {
|
||||
return false;
|
||||
}
|
||||
if (item.id) {
|
||||
this.upsertSubAgentChildItem(callId, item.id, { type: "compaction", status });
|
||||
if (status === "loading") {
|
||||
this.pendingChildCompactionThreadIds.add(childThreadId);
|
||||
this.emitProviderSubagentTimeline(childThreadId, { type: "compaction", status });
|
||||
} else {
|
||||
this.settleChildCompaction(childThreadId);
|
||||
}
|
||||
this.emitSubAgentActivityUpdate(callId);
|
||||
return true;
|
||||
}
|
||||
|
||||
private settleChildCompaction(childThreadId: string): void {
|
||||
if (this.pendingChildCompactionThreadIds.delete(childThreadId)) {
|
||||
this.emitProviderSubagentTimeline(childThreadId, { type: "compaction", status: "completed" });
|
||||
}
|
||||
}
|
||||
|
||||
private shouldSkipCompletedThreadItem(
|
||||
timelineItem: AgentTimelineItem,
|
||||
normalizedItemType: string | undefined,
|
||||
@@ -6076,11 +6054,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
text: parsed.delta,
|
||||
});
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, {
|
||||
type: "assistant_message",
|
||||
messageId: parsed.itemId,
|
||||
text,
|
||||
});
|
||||
this.emitSubAgentActivityUpdate(subAgentCallId, "running");
|
||||
return;
|
||||
}
|
||||
@@ -6114,10 +6087,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
text: parsed.delta,
|
||||
});
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, {
|
||||
type: "reasoning",
|
||||
text: prev.join(""),
|
||||
});
|
||||
this.emitSubAgentActivityUpdate(subAgentCallId, "running");
|
||||
return;
|
||||
}
|
||||
@@ -6185,6 +6154,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
): void {
|
||||
const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
|
||||
if (subAgentCallId) {
|
||||
if (parsed.threadId) this.settleChildCompaction(parsed.threadId);
|
||||
let status: ToolCallTimelineItem["status"] = "completed";
|
||||
if (parsed.status === "failed") {
|
||||
status = "failed";
|
||||
@@ -6771,7 +6741,13 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
|
||||
if (
|
||||
childSubAgentCallId &&
|
||||
this.handleSubAgentContextCompactionItem(childSubAgentCallId, parsed.item, "loading")
|
||||
parsed.threadId &&
|
||||
this.handleSubAgentContextCompactionItem(
|
||||
childSubAgentCallId,
|
||||
parsed.threadId,
|
||||
parsed.item,
|
||||
"loading",
|
||||
)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
@@ -6802,9 +6778,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
});
|
||||
if (childSubAgentCallId) {
|
||||
this.emitStartedProviderSubagentItem(parsed.threadId, timelineItem);
|
||||
if (parsed.item.id) {
|
||||
this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem);
|
||||
}
|
||||
this.emitSubAgentActivityUpdate(childSubAgentCallId, "running");
|
||||
await this.replayPendingSubAgentNotifications(registeredChildThreadIds);
|
||||
return;
|
||||
@@ -6858,9 +6831,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
if (parsed.threadId) {
|
||||
this.emitProviderSubagentTimeline(parsed.threadId, timelineItem);
|
||||
}
|
||||
if (itemId) {
|
||||
this.upsertSubAgentChildItem(childSubAgentCallId, itemId, timelineItem);
|
||||
}
|
||||
this.emitSubAgentActivityUpdate(childSubAgentCallId, "running");
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -950,6 +950,7 @@ function mapCollabAgentToolCallItem(
|
||||
type: "sub_agent",
|
||||
subAgentType: "Sub-agent",
|
||||
...(item.prompt ? { description: item.prompt } : {}),
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
actions: [],
|
||||
};
|
||||
@@ -1006,6 +1007,7 @@ function mapSubAgentActivityItem(
|
||||
type: "sub_agent",
|
||||
subAgentType: nativeName,
|
||||
description,
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
actions: [],
|
||||
},
|
||||
|
||||
@@ -69,7 +69,7 @@ import {
|
||||
type OmpRuntimeOptions,
|
||||
} from "./provider-config.js";
|
||||
export { formatOmpVersionSupport, resolveOmpDiagnosticPaths } from "./provider-config.js";
|
||||
import { OmpSubagentCardTracker, type OmpSubagentCardScheduler } from "./subagent-card-tracker.js";
|
||||
import { OmpSubagentCardTracker } from "./subagent-card-tracker.js";
|
||||
import { ompSkillPromptUserText, shouldDisplayOmpCustomMessage } from "./custom-message.js";
|
||||
import { getUserMessageText } from "./message-history.js";
|
||||
import { mapOmpSystemNoticeToNotification } from "./system-notice.js";
|
||||
@@ -130,7 +130,6 @@ export interface OmpAgentClientOptions {
|
||||
logger: Logger;
|
||||
runtimeSettings?: ProviderRuntimeSettings;
|
||||
runtime?: OmpRuntime;
|
||||
subagentCardScheduler?: OmpSubagentCardScheduler;
|
||||
providerIdleScheduler?: OmpProviderIdleScheduler;
|
||||
noTurnScheduler?: OmpNoTurnScheduler;
|
||||
usagePollScheduler?: OmpUsagePollScheduler;
|
||||
@@ -205,7 +204,6 @@ interface OmpAgentSessionOptions {
|
||||
initialState: OmpSessionState;
|
||||
currentModeId?: string | null;
|
||||
logger: Logger;
|
||||
subagentCardScheduler?: OmpSubagentCardScheduler;
|
||||
providerIdleScheduler?: OmpProviderIdleScheduler;
|
||||
noTurnScheduler?: OmpNoTurnScheduler;
|
||||
usagePollScheduler?: OmpUsagePollScheduler;
|
||||
@@ -773,9 +771,7 @@ export class OmpAgentSession implements AgentSession {
|
||||
this.logger.debug({ err: error }, "OMP context usage poll failed");
|
||||
},
|
||||
});
|
||||
this.subagentCardTracker = new OmpSubagentCardTracker({
|
||||
scheduler: options.subagentCardScheduler,
|
||||
});
|
||||
this.subagentCardTracker = new OmpSubagentCardTracker();
|
||||
this.lastKnownThinkingOptionId =
|
||||
normalizeOmpThinkingOption(options.config.thinkingOptionId) ??
|
||||
this.state.thinkingLevel ??
|
||||
@@ -2155,8 +2151,7 @@ export class OmpAgentSession implements AgentSession {
|
||||
return mapOmpToolDetail(toolCall, result, {
|
||||
toolCallId,
|
||||
bridgedTool: this.hostTools?.bridge?.tool(toolCall.toolName),
|
||||
mapSubagentDetail: (detail) =>
|
||||
this.subagentCardTracker.detailFor(toolCallId, detail) ?? detail,
|
||||
mapSubagentDetail: (detail) => this.subagentCardTracker.detailFor(toolCallId, detail),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -2246,7 +2241,6 @@ export class OmpAgentClient implements AgentClient {
|
||||
|
||||
private readonly logger: Logger;
|
||||
private readonly runtimeSettings?: ProviderRuntimeSettings;
|
||||
private readonly subagentCardScheduler?: OmpSubagentCardScheduler;
|
||||
private readonly providerIdleScheduler?: OmpProviderIdleScheduler;
|
||||
private readonly noTurnScheduler?: OmpNoTurnScheduler;
|
||||
private readonly usagePollScheduler?: OmpUsagePollScheduler;
|
||||
@@ -2265,7 +2259,6 @@ export class OmpAgentClient implements AgentClient {
|
||||
);
|
||||
this.logger = options.logger;
|
||||
this.runtimeSettings = runtimeSettings;
|
||||
this.subagentCardScheduler = options.subagentCardScheduler;
|
||||
this.providerIdleScheduler = options.providerIdleScheduler;
|
||||
this.noTurnScheduler = options.noTurnScheduler;
|
||||
this.usagePollScheduler = options.usagePollScheduler;
|
||||
@@ -2349,7 +2342,6 @@ export class OmpAgentClient implements AgentClient {
|
||||
initialState,
|
||||
currentModeId: launchMode.modeId,
|
||||
logger: this.logger,
|
||||
subagentCardScheduler: this.subagentCardScheduler,
|
||||
providerIdleScheduler: this.providerIdleScheduler,
|
||||
noTurnScheduler: this.noTurnScheduler,
|
||||
usagePollScheduler: this.usagePollScheduler,
|
||||
@@ -2421,7 +2413,6 @@ export class OmpAgentClient implements AgentClient {
|
||||
initialState,
|
||||
currentModeId: launchMode.modeId,
|
||||
logger: this.logger,
|
||||
subagentCardScheduler: this.subagentCardScheduler,
|
||||
providerIdleScheduler: this.providerIdleScheduler,
|
||||
noTurnScheduler: this.noTurnScheduler,
|
||||
usagePollScheduler: this.usagePollScheduler,
|
||||
|
||||
@@ -637,7 +637,7 @@ describe("OMP history mapper", () => {
|
||||
subAgentType: "explore",
|
||||
description: "Inspect files",
|
||||
childSessionId: "/tmp/omp-task/Explore.jsonl",
|
||||
log: "done\ntranscript: /tmp/omp-task/Explore.jsonl",
|
||||
log: "",
|
||||
},
|
||||
error: null,
|
||||
},
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import { OmpSubagentCardTracker, type OmpSubagentCardScheduler } from "./subagent-card-tracker.js";
|
||||
import { OmpSubagentCardTracker } from "./subagent-card-tracker.js";
|
||||
import type { OmpSubagentLifecyclePayload, OmpSubagentProgressPayload } from "./rpc-types.js";
|
||||
|
||||
const PARENT_TOOL_CALL_ID = "task-1";
|
||||
@@ -59,184 +59,84 @@ const PROGRESS: OmpSubagentProgressPayload[] = [
|
||||
},
|
||||
];
|
||||
|
||||
class ManualScheduler implements OmpSubagentCardScheduler {
|
||||
private currentMs = 0;
|
||||
private nextId = 1;
|
||||
private readonly timers = new Map<number, { dueMs: number; callback: () => void }>();
|
||||
|
||||
now(): number {
|
||||
return this.currentMs;
|
||||
}
|
||||
|
||||
setTimeout(callback: () => void, delayMs: number) {
|
||||
const id = this.nextId;
|
||||
this.nextId += 1;
|
||||
this.timers.set(id, { dueMs: this.currentMs + delayMs, callback });
|
||||
return { token: id };
|
||||
}
|
||||
|
||||
clearTimeout(timer: { token: unknown }): void {
|
||||
if (typeof timer.token === "number") {
|
||||
this.timers.delete(timer.token);
|
||||
}
|
||||
}
|
||||
|
||||
advance(ms: number): void {
|
||||
this.currentMs += ms;
|
||||
const dueTimers = [...this.timers.entries()]
|
||||
.filter(([, timer]) => timer.dueMs <= this.currentMs)
|
||||
.sort((left, right) => left[1].dueMs - right[1].dueMs);
|
||||
for (const [id, timer] of dueTimers) {
|
||||
if (this.timers.delete(id)) {
|
||||
timer.callback();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe("OmpSubagentCardTracker", () => {
|
||||
test("folds lifecycle and progress into one throttled sub-agent detail", () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
test("keeps the card a launch marker while the child runs", () => {
|
||||
const emitted: string[] = [];
|
||||
const tracker = new OmpSubagentCardTracker({
|
||||
scheduler,
|
||||
emitToolCall: (toolCallId) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
},
|
||||
});
|
||||
tracker.handleLifecycle(LIFECYCLE);
|
||||
const emit = (toolCallId: string) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
};
|
||||
const tracker = new OmpSubagentCardTracker();
|
||||
tracker.handleLifecycle(LIFECYCLE, emit);
|
||||
for (const payload of PROGRESS) {
|
||||
tracker.handleProgress(payload);
|
||||
tracker.handleProgress(payload, emit);
|
||||
}
|
||||
|
||||
expect(emitted).toEqual([PARENT_TOOL_CALL_ID]);
|
||||
const detailBeforeTrailing = tracker.detailFor(PARENT_TOOL_CALL_ID, {
|
||||
type: "sub_agent",
|
||||
subAgentType: "task",
|
||||
description: "Task arg description wins",
|
||||
log: "",
|
||||
});
|
||||
expect(detailBeforeTrailing).toEqual({
|
||||
type: "sub_agent",
|
||||
subAgentType: "task",
|
||||
description: "Task arg description wins",
|
||||
childSessionId: SESSION_FILE,
|
||||
log: [
|
||||
"EchoSubagent started",
|
||||
"[bash] echo subagent-hi",
|
||||
"[yield]",
|
||||
"EchoSubagent completed",
|
||||
].join("\n"),
|
||||
});
|
||||
|
||||
scheduler.advance(500);
|
||||
|
||||
expect(emitted).toEqual([PARENT_TOOL_CALL_ID, PARENT_TOOL_CALL_ID]);
|
||||
});
|
||||
|
||||
test("aggregates batch progress streams into one index-prefixed log", () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
const emitted: string[] = [];
|
||||
const tracker = new OmpSubagentCardTracker({
|
||||
scheduler,
|
||||
emitToolCall: (toolCallId) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
tracker.handleLifecycle({
|
||||
id: "Explore",
|
||||
agent: "task",
|
||||
description: "Inspect files",
|
||||
status: "started",
|
||||
sessionFile: "/tmp/one.jsonl",
|
||||
parentToolCallId: "task.batch-1",
|
||||
index: 0,
|
||||
});
|
||||
scheduler.advance(600);
|
||||
tracker.handleProgress({
|
||||
index: 5,
|
||||
agent: "task",
|
||||
task: "Run tests",
|
||||
parentToolCallId: "task.batch-1",
|
||||
progress: {
|
||||
id: "Test",
|
||||
status: "running",
|
||||
description: "Run tests",
|
||||
recentTools: [{ tool: "bash", args: "npm test", endMs: 10 }],
|
||||
},
|
||||
sessionFile: "/tmp/two.jsonl",
|
||||
});
|
||||
|
||||
expect(emitted).toEqual(["task.batch-1", "task.batch-1"]);
|
||||
expect(
|
||||
tracker.detailFor("task.batch-1", {
|
||||
tracker.detailFor(PARENT_TOOL_CALL_ID, {
|
||||
type: "sub_agent",
|
||||
subAgentType: "batch",
|
||||
actions: [],
|
||||
subAgentType: "task",
|
||||
description: "Task arg description wins",
|
||||
log: "",
|
||||
}),
|
||||
).toEqual({
|
||||
type: "sub_agent",
|
||||
subAgentType: "batch",
|
||||
description: "Inspect files",
|
||||
childSessionId: "/tmp/one.jsonl",
|
||||
log: "[1/6] Explore started\n[6/6] [bash] npm test",
|
||||
actions: [],
|
||||
subAgentType: "task",
|
||||
description: "Task arg description wins",
|
||||
childSessionId: SESSION_FILE,
|
||||
log: "",
|
||||
});
|
||||
});
|
||||
|
||||
test("keeps dirty trailing state available for final detail and cancels it on cleanup", () => {
|
||||
const scheduler = new ManualScheduler();
|
||||
test("re-emits the card when a later event names the child", () => {
|
||||
const emitted: string[] = [];
|
||||
const tracker = new OmpSubagentCardTracker({
|
||||
scheduler,
|
||||
emitToolCall: (toolCallId) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
const emit = (toolCallId: string) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
};
|
||||
const tracker = new OmpSubagentCardTracker();
|
||||
tracker.handleLifecycle({ ...LIFECYCLE, description: undefined, sessionFile: undefined }, emit);
|
||||
tracker.handleProgress(PROGRESS[0], emit);
|
||||
|
||||
expect(emitted).toEqual([PARENT_TOOL_CALL_ID, PARENT_TOOL_CALL_ID]);
|
||||
expect(tracker.detailFor(PARENT_TOOL_CALL_ID, { type: "sub_agent", log: "" })).toEqual({
|
||||
type: "sub_agent",
|
||||
childSessionId: SESSION_FILE,
|
||||
log: "",
|
||||
});
|
||||
|
||||
tracker.delete(PARENT_TOOL_CALL_ID);
|
||||
expect(tracker.detailFor(PARENT_TOOL_CALL_ID, { type: "sub_agent", log: "" })).toEqual({
|
||||
type: "sub_agent",
|
||||
log: "",
|
||||
});
|
||||
});
|
||||
test("leaves the card alone when a later child of the same task starts", () => {
|
||||
const emitted: string[] = [];
|
||||
const emit = (toolCallId: string) => {
|
||||
emitted.push(toolCallId);
|
||||
return true;
|
||||
};
|
||||
const tracker = new OmpSubagentCardTracker();
|
||||
tracker.handleLifecycle(LIFECYCLE, emit);
|
||||
tracker.handleLifecycle(
|
||||
{
|
||||
...LIFECYCLE,
|
||||
id: "Second",
|
||||
index: 1,
|
||||
description: "Second child",
|
||||
sessionFile: "/s.jsonl",
|
||||
},
|
||||
emit,
|
||||
);
|
||||
|
||||
expect(emitted).toEqual([PARENT_TOOL_CALL_ID]);
|
||||
expect(tracker.detailFor(PARENT_TOOL_CALL_ID, { type: "sub_agent", log: "" })).toEqual({
|
||||
type: "sub_agent",
|
||||
description: "Run echo in subagent",
|
||||
childSessionId: SESSION_FILE,
|
||||
log: "",
|
||||
});
|
||||
|
||||
tracker.handleLifecycle({
|
||||
id: "Explore",
|
||||
agent: "task",
|
||||
description: "Inspect files",
|
||||
status: "started",
|
||||
parentToolCallId: "task-1",
|
||||
index: 0,
|
||||
});
|
||||
scheduler.advance(100);
|
||||
tracker.handleProgress({
|
||||
index: 0,
|
||||
agent: "task",
|
||||
task: "Inspect files",
|
||||
parentToolCallId: "task-1",
|
||||
progress: {
|
||||
id: "Explore",
|
||||
status: "running",
|
||||
recentOutput: ["found target file"],
|
||||
},
|
||||
});
|
||||
|
||||
expect(emitted).toEqual(["task-1"]);
|
||||
expect(
|
||||
tracker.detailFor("task-1", {
|
||||
type: "sub_agent",
|
||||
log: "",
|
||||
}).log,
|
||||
).toBe("Explore started\nfound target file");
|
||||
|
||||
tracker.delete("task-1");
|
||||
scheduler.advance(500);
|
||||
|
||||
expect(emitted).toEqual(["task-1"]);
|
||||
expect(
|
||||
tracker.detailFor("task-1", {
|
||||
type: "sub_agent",
|
||||
log: "static",
|
||||
}).log,
|
||||
).toBe("static");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,361 +1,106 @@
|
||||
import type { ToolCallDetail } from "../../agent-sdk-types.js";
|
||||
import type { OmpSubagentLifecyclePayload, OmpSubagentProgressPayload } from "./rpc-types.js";
|
||||
|
||||
export interface OmpSubagentCardTimer {
|
||||
readonly token: unknown;
|
||||
}
|
||||
|
||||
export interface OmpSubagentCardScheduler {
|
||||
now(): number;
|
||||
setTimeout(callback: () => void, delayMs: number): OmpSubagentCardTimer;
|
||||
clearTimeout(timer: OmpSubagentCardTimer): void;
|
||||
}
|
||||
|
||||
export interface OmpSubagentCardTrackerOptions {
|
||||
scheduler?: OmpSubagentCardScheduler;
|
||||
emitToolCall?: (toolCallId: string) => boolean;
|
||||
}
|
||||
|
||||
interface OmpSubagentLogLine {
|
||||
key: string;
|
||||
text: string;
|
||||
}
|
||||
|
||||
interface OmpSubagentCardItem {
|
||||
interface OmpSubagentCardChild {
|
||||
index: number;
|
||||
description?: string;
|
||||
agent?: string;
|
||||
status?: "pending" | "running" | "completed" | "failed" | "aborted";
|
||||
childSessionId?: string;
|
||||
lines: OmpSubagentLogLine[];
|
||||
lineKeys: Set<string>;
|
||||
}
|
||||
|
||||
interface OmpSubagentCardState {
|
||||
items: Map<number, OmpSubagentCardItem>;
|
||||
emitToolCall: (toolCallId: string) => boolean;
|
||||
lastEmitMs: number | null;
|
||||
trailingTimer: OmpSubagentCardTimer | null;
|
||||
dirty: boolean;
|
||||
}
|
||||
|
||||
const MAX_LOG_LINES_PER_ITEM = 200;
|
||||
const MAX_LOG_LINE_CHARS = 240;
|
||||
const THROTTLE_INTERVAL_MS = 500;
|
||||
|
||||
export function createOmpSubagentCardScheduler(): OmpSubagentCardScheduler {
|
||||
return {
|
||||
now: () => Date.now(),
|
||||
setTimeout: (callback, delayMs) => ({ token: setTimeout(callback, delayMs) }),
|
||||
clearTimeout: (timer) => {
|
||||
clearTimeout(timer.token as ReturnType<typeof setTimeout>);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Fills a task card's description and child session from OMP's sub-agent events. The card is a
|
||||
* launch marker that names the task's first child; the children's activity belongs to their own
|
||||
* timelines.
|
||||
*/
|
||||
export class OmpSubagentCardTracker {
|
||||
private readonly scheduler: OmpSubagentCardScheduler;
|
||||
private readonly emitToolCall: (toolCallId: string) => boolean;
|
||||
private readonly states = new Map<string, OmpSubagentCardState>();
|
||||
|
||||
constructor(options: OmpSubagentCardTrackerOptions) {
|
||||
this.scheduler = options.scheduler ?? createOmpSubagentCardScheduler();
|
||||
this.emitToolCall = options.emitToolCall ?? (() => false);
|
||||
}
|
||||
private readonly firstChildren = new Map<string, OmpSubagentCardChild>();
|
||||
|
||||
handleLifecycle(
|
||||
payload: OmpSubagentLifecyclePayload,
|
||||
emitToolCall: (toolCallId: string) => boolean = this.emitToolCall,
|
||||
emitToolCall: (toolCallId: string) => boolean,
|
||||
): void {
|
||||
const parentToolCallId = readTrimmedString(payload.parentToolCallId);
|
||||
if (!parentToolCallId) {
|
||||
return;
|
||||
}
|
||||
this.stateFor(parentToolCallId).emitToolCall = emitToolCall;
|
||||
const item = this.upsertItem(parentToolCallId, {
|
||||
index: payload.index,
|
||||
agent: payload.agent,
|
||||
description: payload.description,
|
||||
status: payload.status === "started" ? "running" : payload.status,
|
||||
childSessionId: payload.sessionFile,
|
||||
});
|
||||
this.appendLine(
|
||||
item,
|
||||
`lifecycle:${payload.status}:${payload.id}`,
|
||||
`${payload.id} ${payload.status}`,
|
||||
this.update(
|
||||
payload.parentToolCallId,
|
||||
{
|
||||
index: payload.index,
|
||||
description: payload.description,
|
||||
childSessionId: payload.sessionFile,
|
||||
},
|
||||
emitToolCall,
|
||||
);
|
||||
this.requestEmit(parentToolCallId);
|
||||
}
|
||||
|
||||
handleProgress(
|
||||
payload: OmpSubagentProgressPayload,
|
||||
emitToolCall: (toolCallId: string) => boolean = this.emitToolCall,
|
||||
emitToolCall: (toolCallId: string) => boolean,
|
||||
): void {
|
||||
const parentToolCallId = readTrimmedString(payload.parentToolCallId);
|
||||
if (!parentToolCallId) {
|
||||
return;
|
||||
}
|
||||
this.stateFor(parentToolCallId).emitToolCall = emitToolCall;
|
||||
const item = this.upsertItem(parentToolCallId, {
|
||||
index: payload.index,
|
||||
agent: payload.agent,
|
||||
description: payload.progress.description,
|
||||
status: payload.progress.status,
|
||||
childSessionId: payload.sessionFile,
|
||||
});
|
||||
|
||||
const currentToolLine = summarizeTool(readRecord(payload.progress.currentTool));
|
||||
if (currentToolLine) {
|
||||
this.appendLine(item, `current:${currentToolLine.key}`, currentToolLine.text);
|
||||
}
|
||||
|
||||
for (const tool of readRecordArray(payload.progress.recentTools)) {
|
||||
const line = summarizeTool(tool);
|
||||
if (line) {
|
||||
this.appendLine(item, `tool:${line.key}`, line.text);
|
||||
}
|
||||
}
|
||||
|
||||
for (const output of readOutputLines(payload.progress.recentOutput)) {
|
||||
this.appendLine(item, `output:${output}`, output);
|
||||
}
|
||||
|
||||
if (payload.progress.status !== "running") {
|
||||
this.appendLine(
|
||||
item,
|
||||
`status:${payload.progress.status}:${payload.progress.id}`,
|
||||
`${payload.progress.id} ${payload.progress.status}`,
|
||||
);
|
||||
}
|
||||
|
||||
this.requestEmit(parentToolCallId);
|
||||
this.update(
|
||||
payload.parentToolCallId,
|
||||
{
|
||||
index: payload.index,
|
||||
description: payload.progress.description,
|
||||
childSessionId: payload.sessionFile,
|
||||
},
|
||||
emitToolCall,
|
||||
);
|
||||
}
|
||||
|
||||
detailFor(toolCallId: string, baseDetail: ToolCallDetail): ToolCallDetail {
|
||||
if (baseDetail.type !== "sub_agent") {
|
||||
return baseDetail;
|
||||
}
|
||||
const state = this.states.get(toolCallId);
|
||||
if (!state) {
|
||||
const first = this.firstChildren.get(toolCallId);
|
||||
if (!first) {
|
||||
return baseDetail;
|
||||
}
|
||||
const items = [...state.items.values()].sort((left, right) => left.index - right.index);
|
||||
const firstItem = items[0];
|
||||
if (!firstItem) {
|
||||
return baseDetail;
|
||||
}
|
||||
|
||||
const detail: ToolCallDetail = {
|
||||
type: "sub_agent",
|
||||
...(baseDetail.subAgentType ? { subAgentType: baseDetail.subAgentType } : {}),
|
||||
...((baseDetail.description ?? firstItem.description)
|
||||
? { description: baseDetail.description ?? firstItem.description }
|
||||
: {}),
|
||||
...((firstItem.childSessionId ?? baseDetail.childSessionId)
|
||||
? { childSessionId: firstItem.childSessionId ?? baseDetail.childSessionId }
|
||||
: {}),
|
||||
log: buildLog(items, baseDetail.log),
|
||||
const description = baseDetail.description ?? first.description;
|
||||
const childSessionId = first.childSessionId ?? baseDetail.childSessionId;
|
||||
return {
|
||||
...baseDetail,
|
||||
...(description ? { description } : {}),
|
||||
...(childSessionId ? { childSessionId } : {}),
|
||||
};
|
||||
return baseDetail.actions ? { ...detail, actions: baseDetail.actions } : detail;
|
||||
}
|
||||
|
||||
flush(toolCallId: string): void {
|
||||
const state = this.states.get(toolCallId);
|
||||
if (!state || !state.dirty) {
|
||||
return;
|
||||
}
|
||||
this.emitNow(toolCallId, state);
|
||||
}
|
||||
|
||||
delete(toolCallId: string): void {
|
||||
const state = this.states.get(toolCallId);
|
||||
if (state?.trailingTimer) {
|
||||
this.scheduler.clearTimeout(state.trailingTimer);
|
||||
}
|
||||
this.states.delete(toolCallId);
|
||||
this.firstChildren.delete(toolCallId);
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
for (const toolCallId of this.states.keys()) {
|
||||
this.delete(toolCallId);
|
||||
}
|
||||
this.firstChildren.clear();
|
||||
}
|
||||
|
||||
private upsertItem(
|
||||
parentToolCallId: string,
|
||||
input: {
|
||||
index: number;
|
||||
agent?: string;
|
||||
status?: "pending" | "running" | "completed" | "failed" | "aborted";
|
||||
description?: string;
|
||||
childSessionId?: string;
|
||||
},
|
||||
): OmpSubagentCardItem {
|
||||
const state = this.stateFor(parentToolCallId);
|
||||
const existing = state.items.get(input.index);
|
||||
if (existing) {
|
||||
existing.agent = readTrimmedString(input.agent) ?? existing.agent;
|
||||
existing.status = input.status ?? existing.status;
|
||||
existing.description = readTrimmedString(input.description) ?? existing.description;
|
||||
existing.childSessionId = readTrimmedString(input.childSessionId) ?? existing.childSessionId;
|
||||
return existing;
|
||||
private update(
|
||||
parentToolCallId: string | undefined,
|
||||
input: { index: number; description?: string; childSessionId?: string },
|
||||
emitToolCall: (toolCallId: string) => boolean,
|
||||
): void {
|
||||
const toolCallId = readTrimmedString(parentToolCallId);
|
||||
if (!toolCallId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const item: OmpSubagentCardItem = {
|
||||
const first = this.firstChildren.get(toolCallId);
|
||||
if (first && input.index > first.index) {
|
||||
return;
|
||||
}
|
||||
const existing = first?.index === input.index ? first : undefined;
|
||||
const next: OmpSubagentCardChild = {
|
||||
index: input.index,
|
||||
lines: [],
|
||||
lineKeys: new Set<string>(),
|
||||
...(readTrimmedString(input.agent) ? { agent: readTrimmedString(input.agent) } : {}),
|
||||
...(input.status ? { status: input.status } : {}),
|
||||
...(readTrimmedString(input.description)
|
||||
? { description: readTrimmedString(input.description) }
|
||||
: {}),
|
||||
...(readTrimmedString(input.childSessionId)
|
||||
? { childSessionId: readTrimmedString(input.childSessionId) }
|
||||
: {}),
|
||||
description: readTrimmedString(input.description) ?? existing?.description,
|
||||
childSessionId: readTrimmedString(input.childSessionId) ?? existing?.childSessionId,
|
||||
};
|
||||
state.items.set(input.index, item);
|
||||
return item;
|
||||
}
|
||||
|
||||
private stateFor(parentToolCallId: string): OmpSubagentCardState {
|
||||
const existing = this.states.get(parentToolCallId);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
const state: OmpSubagentCardState = {
|
||||
items: new Map(),
|
||||
emitToolCall: this.emitToolCall,
|
||||
lastEmitMs: null,
|
||||
trailingTimer: null,
|
||||
dirty: false,
|
||||
};
|
||||
this.states.set(parentToolCallId, state);
|
||||
return state;
|
||||
}
|
||||
|
||||
private appendLine(item: OmpSubagentCardItem, key: string, text: string): void {
|
||||
const normalizedText = normalizeLine(text);
|
||||
if (!normalizedText || item.lineKeys.has(key)) {
|
||||
if (
|
||||
existing &&
|
||||
existing.description === next.description &&
|
||||
existing.childSessionId === next.childSessionId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
item.lines.push({ key, text: normalizedText });
|
||||
item.lineKeys.add(key);
|
||||
while (item.lines.length > MAX_LOG_LINES_PER_ITEM) {
|
||||
const removed = item.lines.shift();
|
||||
if (removed) {
|
||||
item.lineKeys.delete(removed.key);
|
||||
}
|
||||
}
|
||||
this.firstChildren.set(toolCallId, next);
|
||||
emitToolCall(toolCallId);
|
||||
}
|
||||
|
||||
private requestEmit(toolCallId: string): void {
|
||||
const state = this.stateFor(toolCallId);
|
||||
const now = this.scheduler.now();
|
||||
const elapsedMs = state.lastEmitMs === null ? THROTTLE_INTERVAL_MS : now - state.lastEmitMs;
|
||||
if (elapsedMs >= THROTTLE_INTERVAL_MS) {
|
||||
this.emitNow(toolCallId, state);
|
||||
return;
|
||||
}
|
||||
|
||||
state.dirty = true;
|
||||
if (state.trailingTimer) {
|
||||
return;
|
||||
}
|
||||
state.trailingTimer = this.scheduler.setTimeout(() => {
|
||||
state.trailingTimer = null;
|
||||
if (state.dirty) {
|
||||
this.emitNow(toolCallId, state);
|
||||
}
|
||||
}, THROTTLE_INTERVAL_MS - elapsedMs);
|
||||
}
|
||||
|
||||
private emitNow(toolCallId: string, state: OmpSubagentCardState): void {
|
||||
if (state.trailingTimer) {
|
||||
this.scheduler.clearTimeout(state.trailingTimer);
|
||||
state.trailingTimer = null;
|
||||
}
|
||||
state.dirty = false;
|
||||
state.lastEmitMs = this.scheduler.now();
|
||||
state.emitToolCall(toolCallId);
|
||||
}
|
||||
}
|
||||
|
||||
function buildLog(items: OmpSubagentCardItem[], fallback: string): string {
|
||||
const itemCount = Math.max(...items.map((item) => item.index)) + 1;
|
||||
const lines = items.flatMap((item) => {
|
||||
const prefix = itemCount > 1 ? `[${item.index + 1}/${itemCount}] ` : "";
|
||||
return item.lines.map((line) => `${prefix}${line.text}`);
|
||||
});
|
||||
return lines.length > 0 ? lines.join("\n") : fallback;
|
||||
}
|
||||
|
||||
function summarizeTool(tool: Record<string, unknown> | null): OmpSubagentLogLine | null {
|
||||
if (!tool) {
|
||||
return null;
|
||||
}
|
||||
const toolName = readTrimmedString(tool.tool) ?? readTrimmedString(tool.name);
|
||||
if (!toolName) {
|
||||
return null;
|
||||
}
|
||||
const args = readTrimmedString(tool.args) ?? summarizeUnknown(tool.input);
|
||||
const endMs = typeof tool.endMs === "number" ? `:${tool.endMs}` : "";
|
||||
return {
|
||||
key: `${toolName}:${args ?? ""}${endMs}`,
|
||||
text: args ? `[${toolName}] ${args}` : `[${toolName}]`,
|
||||
};
|
||||
}
|
||||
|
||||
function readOutputLines(value: unknown): string[] {
|
||||
if (!Array.isArray(value)) {
|
||||
return [];
|
||||
}
|
||||
return value.flatMap((item) => {
|
||||
const text =
|
||||
readTrimmedString(item) ??
|
||||
(isRecord(item)
|
||||
? (readTrimmedString(item.text) ??
|
||||
readTrimmedString(item.output) ??
|
||||
readTrimmedString(item.content))
|
||||
: undefined);
|
||||
return text ? [text] : [];
|
||||
});
|
||||
}
|
||||
|
||||
function readRecord(value: unknown): Record<string, unknown> | null {
|
||||
return isRecord(value) ? value : null;
|
||||
}
|
||||
|
||||
function readRecordArray(value: unknown): Record<string, unknown>[] {
|
||||
return Array.isArray(value) ? value.filter(isRecord) : [];
|
||||
}
|
||||
|
||||
function summarizeUnknown(value: unknown): string | undefined {
|
||||
if (value === undefined || value === null) {
|
||||
return undefined;
|
||||
}
|
||||
if (typeof value === "string") {
|
||||
return readTrimmedString(value);
|
||||
}
|
||||
const json = JSON.stringify(value);
|
||||
return typeof json === "string" ? normalizeLine(json) : undefined;
|
||||
}
|
||||
|
||||
function normalizeLine(value: string): string | undefined {
|
||||
const normalized = value.trim().replace(/\s+/g, " ");
|
||||
if (!normalized) {
|
||||
return undefined;
|
||||
}
|
||||
if (normalized.length <= MAX_LOG_LINE_CHARS) {
|
||||
return normalized;
|
||||
}
|
||||
return `${normalized.slice(0, MAX_LOG_LINE_CHARS)}...`;
|
||||
}
|
||||
|
||||
function readTrimmedString(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
@@ -85,7 +85,7 @@ describe("OMP tool call mapper", () => {
|
||||
expect(mapOmpToolDetail(parseToolArgs("todo", { op: "view" }), null)).toBeNull();
|
||||
});
|
||||
|
||||
test("uses task result text and transcript path as the best static replay detail", () => {
|
||||
test("takes the child session from the task transcript path", () => {
|
||||
expect(
|
||||
mapOmpToolDetail(
|
||||
parseToolArgs("task", {
|
||||
@@ -106,7 +106,7 @@ describe("OMP tool call mapper", () => {
|
||||
subAgentType: "explore",
|
||||
description: "Inspect the target files",
|
||||
childSessionId: "/tmp/omp-task-static/Explore.jsonl",
|
||||
log: "done\ntranscript: /tmp/omp-task-static/Explore.jsonl",
|
||||
log: "",
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -179,7 +179,6 @@ const OPAQUE_TEXT_TOOLS = new Set([
|
||||
function mapOmpTaskDetail(args: unknown, result: OmpToolResult): ToolCallDetail {
|
||||
const argRecord = isRecord(args) ? args : {};
|
||||
const firstTask = Array.isArray(argRecord.tasks) ? argRecord.tasks.find(isRecord) : undefined;
|
||||
const resultText = extractTextFromToolResult(result);
|
||||
const childSessionId = readChildSessionId(result);
|
||||
return {
|
||||
type: "sub_agent",
|
||||
@@ -200,7 +199,8 @@ function mapOmpTaskDetail(args: unknown, result: OmpToolResult): ToolCallDetail
|
||||
argRecord.assignment,
|
||||
),
|
||||
...(childSessionId ? { childSessionId } : {}),
|
||||
log: resultText?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -4846,15 +4846,6 @@ class OpenCodeAgentSession implements AgentSession {
|
||||
...item,
|
||||
status: "failed",
|
||||
error,
|
||||
detail:
|
||||
item.detail.type === "sub_agent"
|
||||
? {
|
||||
...item.detail,
|
||||
log: [item.detail.log, error.message]
|
||||
.filter((entry) => entry.trim().length > 0)
|
||||
.join("\n"),
|
||||
}
|
||||
: item.detail,
|
||||
},
|
||||
},
|
||||
turnId,
|
||||
|
||||
@@ -153,21 +153,6 @@ const OpencodeGrepOutputSchema = z
|
||||
])
|
||||
.nullable();
|
||||
|
||||
function formatLogEntry(value: unknown): string | undefined {
|
||||
const outputText = readOutputText(value);
|
||||
if (outputText) {
|
||||
return outputText;
|
||||
}
|
||||
if (value === null || value === undefined) {
|
||||
return undefined;
|
||||
}
|
||||
try {
|
||||
return JSON.stringify(value, null, 2);
|
||||
} catch {
|
||||
return String(value);
|
||||
}
|
||||
}
|
||||
|
||||
function extractOpenCodeTaskSessionId(value: unknown): string | undefined {
|
||||
const text = readOutputText(value);
|
||||
if (text) {
|
||||
@@ -195,7 +180,6 @@ function extractOpenCodeTaskSessionId(value: unknown): string | undefined {
|
||||
function deriveOpencodeTaskDetail(
|
||||
input: unknown,
|
||||
output: unknown,
|
||||
error: unknown,
|
||||
metadata: Record<string, unknown> | undefined,
|
||||
): ToolCallDetail | null {
|
||||
if (!isRecord(input)) {
|
||||
@@ -208,7 +192,6 @@ function deriveOpencodeTaskDetail(
|
||||
return null;
|
||||
}
|
||||
|
||||
const log = [formatLogEntry(output), formatLogEntry(error)].filter((entry) => entry).join("\n");
|
||||
const childSessionId =
|
||||
normalizeSubAgentText(metadata?.sessionId) ?? extractOpenCodeTaskSessionId(output);
|
||||
return {
|
||||
@@ -216,7 +199,8 @@ function deriveOpencodeTaskDetail(
|
||||
...(subAgentType ? { subAgentType } : {}),
|
||||
...(description ? { description } : {}),
|
||||
...(childSessionId ? { childSessionId } : {}),
|
||||
log,
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
actions: [],
|
||||
};
|
||||
}
|
||||
@@ -373,11 +357,10 @@ export function deriveOpencodeToolDetail(
|
||||
toolName: string,
|
||||
input: unknown,
|
||||
output: unknown,
|
||||
error: unknown = null,
|
||||
metadata?: Record<string, unknown>,
|
||||
): ToolCallDetail {
|
||||
if (toolName.trim().toLowerCase() === "task") {
|
||||
const taskDetail = deriveOpencodeTaskDetail(input, output, error, metadata);
|
||||
const taskDetail = deriveOpencodeTaskDetail(input, output, metadata);
|
||||
if (taskDetail) {
|
||||
return taskDetail;
|
||||
}
|
||||
|
||||
@@ -460,7 +460,7 @@ describe("opencode tool-call mapper", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("maps completed task calls with final output log to sub_agent detail", () => {
|
||||
it("maps completed task calls to sub_agent detail", () => {
|
||||
const item = expectMapped(
|
||||
mapOpencodeToolCall({
|
||||
toolName: "task",
|
||||
@@ -480,7 +480,7 @@ describe("opencode tool-call mapper", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "explore",
|
||||
description: "Explore agent-tools codebase",
|
||||
log: "Found the CLI entrypoint and provider registry.",
|
||||
log: "",
|
||||
actions: [],
|
||||
});
|
||||
});
|
||||
@@ -505,12 +505,12 @@ describe("opencode tool-call mapper", () => {
|
||||
subAgentType: "explore",
|
||||
description: "Explore current directory",
|
||||
childSessionId: "ses_2268db431ffe299vL1bbot8R7Z",
|
||||
log: '<task id="ses_2268db431ffe299vL1bbot8R7Z">done</task>',
|
||||
log: "",
|
||||
actions: [],
|
||||
});
|
||||
});
|
||||
|
||||
it("maps aborted task calls with preserved error log to sub_agent detail", () => {
|
||||
it("maps aborted task calls to sub_agent detail", () => {
|
||||
const item = expectMapped(
|
||||
mapOpencodeToolCall({
|
||||
toolName: "task",
|
||||
@@ -531,7 +531,7 @@ describe("opencode tool-call mapper", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "explore",
|
||||
description: "Explore agent-tools codebase",
|
||||
log: "Tool execution aborted",
|
||||
log: "",
|
||||
actions: [],
|
||||
});
|
||||
});
|
||||
|
||||
@@ -43,7 +43,7 @@ export function mapOpencodeToolCall(params: OpencodeToolCallParams): ToolCallTim
|
||||
const error = raw.error ?? null;
|
||||
const rawStatus = typeof raw.status === "string" ? raw.status : undefined;
|
||||
const status = normalizeToolCallStatus(rawStatus, error, output);
|
||||
const detail = deriveOpencodeToolDetail(name, input, output, error, raw.metadata);
|
||||
const detail = deriveOpencodeToolDetail(name, input, output, raw.metadata);
|
||||
|
||||
if (status === "failed") {
|
||||
return {
|
||||
|
||||
+2
-2
@@ -1,5 +1,4 @@
|
||||
import { z } from "zod";
|
||||
import { extractTextFromToolResult } from "../../tool-call-mapper.js";
|
||||
import type { PiChildSessionFile, PiExtension } from "../contract.js";
|
||||
import type { ProviderSubagentInputEvent } from "../../../../provider-subagents/store.js";
|
||||
|
||||
@@ -99,7 +98,8 @@ export const edxethPiSubagents: PiExtension = {
|
||||
detail: {
|
||||
type: "sub_agent",
|
||||
description,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
},
|
||||
subagents,
|
||||
childSessions,
|
||||
|
||||
+12
@@ -130,6 +130,18 @@ describe("@gotgenes/pi-subagents adapter", () => {
|
||||
.filter((event) => event.event.type === "upsert")
|
||||
.map((event) => (event.event.type === "upsert" ? event.event.status : null)),
|
||||
).toEqual(["running", "completed"]);
|
||||
expect(
|
||||
events.flatMap((event) => (event.event.type === "timeline" ? [event.event] : [])),
|
||||
).toEqual([
|
||||
{
|
||||
type: "timeline",
|
||||
id: "call_502323",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: "Agent completed in 4.7s (1 tool uses, 9.0k token).\nAgent ID: 848583f5-6749-4a1\n\nThe workspace is empty.",
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
test("correlates a background notification and result collection", async () => {
|
||||
const source = readSubagentFixture(new URL("./fixtures/background.json", import.meta.url));
|
||||
|
||||
+18
-3
@@ -64,7 +64,8 @@ export const gotgenesPiSubagents: PiExtension = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: title,
|
||||
description,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
if (call.status === "running")
|
||||
return {
|
||||
@@ -92,6 +93,10 @@ export const gotgenesPiSubagents: PiExtension = {
|
||||
outputFileFromToolResult(call.result) ?? pendingFilesByAgent.get(details.data.agentId);
|
||||
pendingFilesByAgent.delete(details.data.agentId);
|
||||
const childSessions = childSession(call.callId, file);
|
||||
const childStatus = status(details.data.status);
|
||||
// Without a session file, as on replay, the child's answer is the tool result.
|
||||
const answer =
|
||||
!file && childStatus !== "running" ? extractTextFromToolResult(call.result)?.trim() : "";
|
||||
return {
|
||||
detail,
|
||||
subagents: [
|
||||
@@ -101,8 +106,17 @@ export const gotgenesPiSubagents: PiExtension = {
|
||||
title,
|
||||
description,
|
||||
toolCallId: call.callId,
|
||||
status: status(details.data.status),
|
||||
status: childStatus,
|
||||
},
|
||||
...(answer
|
||||
? [
|
||||
{
|
||||
type: "timeline" as const,
|
||||
id: call.callId,
|
||||
item: { type: "assistant_message" as const, text: answer },
|
||||
},
|
||||
]
|
||||
: []),
|
||||
],
|
||||
childSessions,
|
||||
};
|
||||
@@ -118,7 +132,8 @@ export const gotgenesPiSubagents: PiExtension = {
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
description: details.success ? details.data.description : undefined,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
if (call.status === "running" || !details.success) return { detail };
|
||||
const file = details.data.transcriptPath;
|
||||
|
||||
@@ -210,7 +210,7 @@ describe("Pi extension host", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "scout",
|
||||
description: "Inspect files",
|
||||
log: "Found two files",
|
||||
log: "",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -137,7 +137,7 @@ describe("pi-subagents adapter", () => {
|
||||
type: "sub_agent",
|
||||
subAgentType: "reviewer",
|
||||
description: "Review the Pi mapper change",
|
||||
log: "The mapper change preserves provider status.",
|
||||
log: "",
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -131,7 +131,8 @@ export const piSubagents: PiExtension = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: args.data.agent,
|
||||
description: args.data.task,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
const base = {
|
||||
type: "upsert" as const,
|
||||
|
||||
@@ -54,8 +54,6 @@ export async function verifySubagentFixture(
|
||||
try {
|
||||
await session.startTurn("Delegate work");
|
||||
for (const event of fixture.events) pi.latestSession().emit(event);
|
||||
const immediate = live.filter((event) => event.type === "provider_subagent");
|
||||
expect(immediate.some((event) => event.event.type === "timeline")).toBe(false);
|
||||
pi.latestSession().finishTurn();
|
||||
const replay: AgentStreamEvent[] = [];
|
||||
for await (const event of streamPiHistory("pi", fixture.messages)) replay.push(event);
|
||||
|
||||
+19
-2
@@ -36,12 +36,17 @@ describe("@tintinweb/pi-subagents adapter", () => {
|
||||
});
|
||||
expect(mapping?.subagents).toEqual([
|
||||
expect.objectContaining({ id: "call-1", status: "completed" }),
|
||||
{
|
||||
type: "timeline",
|
||||
id: "call-1",
|
||||
item: { type: "assistant_message", text: "Wrapped up (turn limit)" },
|
||||
},
|
||||
]);
|
||||
expect(mapping?.detail).toEqual({
|
||||
type: "sub_agent",
|
||||
subAgentType: "Explore",
|
||||
description: "Inspect",
|
||||
log: "Wrapped up (turn limit)",
|
||||
log: "",
|
||||
});
|
||||
});
|
||||
test("maps captured foreground lifecycle live and on replay", async () => {
|
||||
@@ -53,6 +58,18 @@ describe("@tintinweb/pi-subagents adapter", () => {
|
||||
.filter((event) => event.event.type === "upsert")
|
||||
.map((event) => (event.event.type === "upsert" ? event.event.status : null)),
|
||||
).toEqual(["running", "completed"]);
|
||||
expect(
|
||||
events.flatMap((event) => (event.event.type === "timeline" ? [event.event] : [])),
|
||||
).toEqual([
|
||||
{
|
||||
type: "timeline",
|
||||
id: "call_458926",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: "Agent completed in 6.6s (1 tool uses, 18.8k token).\n\nThe workspace directory `/tmp/paseo-pi-extension-phase3/workspace` is empty (no files or hidden dotfiles).",
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
test("uses a structured notification to complete a background child", async () => {
|
||||
const source = readSubagentFixture(new URL("./fixtures/background.json", import.meta.url));
|
||||
@@ -147,7 +164,7 @@ describe("@tintinweb/pi-subagents adapter", () => {
|
||||
expect(mapping?.detail).toEqual({
|
||||
type: "sub_agent",
|
||||
description: undefined,
|
||||
log: "Wrapped up (turn limit)",
|
||||
log: "",
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
+18
-3
@@ -57,7 +57,8 @@ export const tintinwebPiSubagents: PiExtension = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: title,
|
||||
description,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
if (call.status === "running")
|
||||
return {
|
||||
@@ -84,6 +85,10 @@ export const tintinwebPiSubagents: PiExtension = {
|
||||
const file = outputFileFromToolResult(call.result);
|
||||
const childSessions = file && !readSessions.has(file) ? [{ id: call.callId, file }] : [];
|
||||
if (file) readSessions.add(file);
|
||||
const childStatus = status(details.data.status);
|
||||
// A foreground child has no session file; its answer is the tool result.
|
||||
const answer =
|
||||
!file && childStatus !== "running" ? extractTextFromToolResult(call.result)?.trim() : "";
|
||||
return {
|
||||
detail,
|
||||
subagents: [
|
||||
@@ -93,8 +98,17 @@ export const tintinwebPiSubagents: PiExtension = {
|
||||
title,
|
||||
description,
|
||||
toolCallId: call.callId,
|
||||
status: status(details.data.status),
|
||||
status: childStatus,
|
||||
},
|
||||
...(answer
|
||||
? [
|
||||
{
|
||||
type: "timeline" as const,
|
||||
id: call.callId,
|
||||
item: { type: "assistant_message" as const, text: answer },
|
||||
},
|
||||
]
|
||||
: []),
|
||||
],
|
||||
childSessions,
|
||||
};
|
||||
@@ -110,7 +124,8 @@ export const tintinwebPiSubagents: PiExtension = {
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
description: details.success ? details.data.description : undefined,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
// COMPAT(subagent-log): remove after 2027-04-30.
|
||||
log: "",
|
||||
};
|
||||
if (call.status === "running" || !details.success) return { detail };
|
||||
return {
|
||||
|
||||
@@ -2091,3 +2091,80 @@ test.each(["removed", "disabled"] as const)(
|
||||
).toEqual([]);
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"a 50 MiB timeline crosses a routed host without dropping content or the route",
|
||||
{ timeout: 30000 },
|
||||
async () => {
|
||||
const b = await daemon();
|
||||
const a = await daemon();
|
||||
const destination = await api(b);
|
||||
const origin = await api(a);
|
||||
const workspace = await destination.paseo.workspaces.create({
|
||||
source: { kind: "directory", path: b.staticDir },
|
||||
});
|
||||
const agent = await workspace.agents.create({
|
||||
config: { provider: "claude/claude-test", modeId: "full-access" },
|
||||
title: "Large history",
|
||||
});
|
||||
// Keep each row below the established item budget; exercise transport size across rows.
|
||||
const text = "x".repeat(240 * 1024);
|
||||
const rowCount = 220;
|
||||
for (let index = 0; index < rowCount; index++) {
|
||||
await b.daemon.agentManager.appendTimelineItem(agent.id, { type: "user_message", text });
|
||||
}
|
||||
await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` });
|
||||
const serverId = (await online(origin.paseo)).hosts[0]!.serverId!;
|
||||
const remote = await origin.paseo.hosts.client(serverId);
|
||||
const response = await remote.agents.ref(agent.id).timeline.refetch({ limit: 0 });
|
||||
const messages = response.entries.flatMap((row) =>
|
||||
row.item.type === "user_message" ? [row.item.text] : [],
|
||||
);
|
||||
expect(messages).toHaveLength(rowCount);
|
||||
expect(messages.every((message) => message === text)).toBe(true);
|
||||
expect(
|
||||
messages.reduce((bytes, message) => bytes + Buffer.byteLength(message), 0),
|
||||
).toBeGreaterThan(50 * 1024 * 1024);
|
||||
expect((await remote.workspaces.list()).entries).toHaveLength(1);
|
||||
await remote.dispose();
|
||||
},
|
||||
);
|
||||
|
||||
test("a pre-write host budget rejection leaves sibling routes connected", async () => {
|
||||
const { HostTransportBudget } = await import("@getpaseo/client/node");
|
||||
const b = await daemon();
|
||||
const origin = createHostClient(
|
||||
`127.0.0.1:${b.port}`,
|
||||
{ clientId: "budget-origin", reconnect: { enabled: false } },
|
||||
new HostTransportBudget(1024 * 1024),
|
||||
);
|
||||
cleanup.push(() => origin.close());
|
||||
await origin.connect();
|
||||
const serverId = origin.getLastServerInfoMessage()!.serverId;
|
||||
const routeIds = [
|
||||
await origin.openHostLink("budget-first"),
|
||||
await origin.openHostLink("budget-sibling"),
|
||||
];
|
||||
const peers = routeIds.map(
|
||||
(id, index) =>
|
||||
new DaemonClient({
|
||||
url: "ws://host-link/ws",
|
||||
clientId: `budget-child-${index}`,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: createHostChannelTransport(origin, serverId, `budget-${index}`, id),
|
||||
}),
|
||||
);
|
||||
for (const peer of peers) {
|
||||
cleanup.push(() => peer.close());
|
||||
await peer.connect();
|
||||
}
|
||||
await expect(async () =>
|
||||
origin.sendHostFrame(routeIds[0]!, new Uint8Array(2 * 1024 * 1024)),
|
||||
).rejects.toMatchObject({ code: "resource_limit" });
|
||||
await peers[1]!.ping({ timeoutMs: 2000 });
|
||||
expect([origin.isConnected, ...peers.map((peer) => peer.isConnected)]).toEqual([
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
]);
|
||||
}, 30000);
|
||||
|
||||
@@ -510,10 +510,13 @@ export class ConnectedHosts {
|
||||
return;
|
||||
}
|
||||
if (!verified) throw new HostError("host_offline", "Host identity has not been verified");
|
||||
try {
|
||||
transport.send(data);
|
||||
} catch {
|
||||
const rejectWrite = () =>
|
||||
route.close("resource_limit", "Host buffer limit reached or transport unavailable");
|
||||
try {
|
||||
const write = transport.send(data);
|
||||
if (write) void write.catch(rejectWrite);
|
||||
} catch {
|
||||
rejectWrite();
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
import type { SessionDelivery } from "../session/owned-subscriptions/index.js";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { createHash } from "node:crypto";
|
||||
import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index";
|
||||
import {
|
||||
MAX_HOST_CHANNEL_FRAME_BYTES,
|
||||
encodeHostRouteFrame,
|
||||
type HostRouteFrame,
|
||||
} from "@getpaseo/protocol/binary-frames/index";
|
||||
import type {
|
||||
DaemonPermission,
|
||||
SessionInboundMessage,
|
||||
@@ -10,6 +14,7 @@ import type {
|
||||
|
||||
/** A logical socket has no network resources or authority of its own. */
|
||||
export class VirtualHostSocket extends EventEmitter {
|
||||
readonly maxFrameBytes = MAX_HOST_CHANNEL_FRAME_BYTES;
|
||||
readyState = 1;
|
||||
constructor(
|
||||
private readonly output: (data: string | Uint8Array) => Promise<void>,
|
||||
|
||||
+1
-23
@@ -1,6 +1,3 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
@@ -10,7 +7,7 @@ import { ClaudeAgentClient } from "../agent/providers/claude/agent.js";
|
||||
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
|
||||
import { DaemonClient } from "../test-utils/daemon-client.js";
|
||||
|
||||
test("background cards retain child actions after the parent finishes (#6253)", async () => {
|
||||
test("background children keep their actions after the parent finishes (#6253)", async () => {
|
||||
const cwd = mkdtempSync(path.join(tmpdir(), "paseo-background-card-"));
|
||||
const logger = pino({ level: "silent" });
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
@@ -62,22 +59,6 @@ test("background cards retain child actions after the parent finishes (#6253)",
|
||||
limit: 0,
|
||||
projection: "projected",
|
||||
});
|
||||
const { stdout } = await promisify(execFile)(
|
||||
process.execPath,
|
||||
[
|
||||
"--conditions=source",
|
||||
"--import",
|
||||
"tsx",
|
||||
fileURLToPath(new URL("../../../../cli/src/index.ts", import.meta.url)),
|
||||
"--host",
|
||||
`127.0.0.1:${daemon.port}`,
|
||||
"logs",
|
||||
agent.id,
|
||||
],
|
||||
{ env: { ...process.env, PASEO_HOME: daemon.paseoHome }, timeout: 30_000 },
|
||||
);
|
||||
expect(stdout).toContain("[Bash] ls -1");
|
||||
expect(stdout).toContain("[Bash] pwd");
|
||||
expect(
|
||||
atFinish.entries.filter(
|
||||
(row) => row.item.type === "tool_call" && row.item.detail?.type === "sub_agent",
|
||||
@@ -94,9 +75,6 @@ test("background cards retain child actions after the parent finishes (#6253)",
|
||||
),
|
||||
).toBe(true);
|
||||
}
|
||||
for (const card of cards) {
|
||||
expect(card.detail?.type === "sub_agent" ? card.detail.log : "").toContain("[Bash]");
|
||||
}
|
||||
} finally {
|
||||
await client.close();
|
||||
await daemon.close();
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { expect, test } from "vitest";
|
||||
import { createDaemonTestContext } from "../test-utils/index.js";
|
||||
|
||||
// #2220: a timeline page is one WebSocket message, so a few huge entries used to close the socket.
|
||||
test.each([40, 0])(
|
||||
"fetching image-heavy history with limit %i keeps the direct connection usable",
|
||||
async (limit) => {
|
||||
const ctx = await createDaemonTestContext();
|
||||
const cwd = mkdtempSync(path.join(tmpdir(), "oversized-timeline-"));
|
||||
try {
|
||||
const agent = await ctx.client.createAgent({ provider: "codex", cwd });
|
||||
const imageUrl = `data:image/jpeg;base64,${"A".repeat(14 * 1024 * 1024)}`;
|
||||
for (let index = 0; index < 5; index += 1) {
|
||||
await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, {
|
||||
type: "tool_call",
|
||||
callId: `screenshot-${index}`,
|
||||
name: "node_repl.js",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: {
|
||||
type: "unknown",
|
||||
input: { index },
|
||||
output: { _meta: { "codex/toolSurface": { screenshot: { url: imageUrl } } } },
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const page = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit });
|
||||
const screenshots = page.entries.filter((entry) => entry.item.type === "tool_call");
|
||||
expect(screenshots).toHaveLength(5);
|
||||
for (const entry of screenshots) {
|
||||
expect(Buffer.byteLength(JSON.stringify(entry.item))).toBeLessThanOrEqual(64 * 1024);
|
||||
}
|
||||
|
||||
const next = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1 });
|
||||
expect(next.entries).toHaveLength(1);
|
||||
} finally {
|
||||
await ctx.cleanup();
|
||||
rmSync(cwd, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
@@ -9,7 +9,7 @@ import {
|
||||
} from "@getpaseo/protocol/binary-frames/index";
|
||||
import {
|
||||
getDaemonAuthFailureReason,
|
||||
type DaemonClient as HostClient,
|
||||
DaemonClient as HostClient,
|
||||
} from "@getpaseo/client/internal/daemon-client";
|
||||
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
|
||||
import { describe, expect, test } from "vitest";
|
||||
@@ -593,72 +593,112 @@ async function waitForCapturedLog(
|
||||
}
|
||||
}, 90000);
|
||||
|
||||
test("oversized attachments fail without disconnecting or resubmitting over the relay", async () => {
|
||||
await startRelay({ useLocalRelay: true });
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
relayEnabled: true,
|
||||
relayEndpoint: `127.0.0.1:${relayPort}`,
|
||||
relayUseTls: false,
|
||||
relayPublicUseTls: false,
|
||||
});
|
||||
let client: DaemonClient | null = null;
|
||||
try {
|
||||
const offerUrl = await getPairingOfferUrl({
|
||||
paseoHome: daemon.paseoHome,
|
||||
test.each([true, false])(
|
||||
"large messages cross the relay with unpaced upload admission preserved (supportsDrain=%s)",
|
||||
async (supportsDrain) => {
|
||||
await startRelay({ useLocalRelay: true });
|
||||
const daemon = await createTestPaseoDaemon({
|
||||
relayEnabled: true,
|
||||
relayEndpoint: daemon.config.relayEndpoint,
|
||||
relayEndpoint: `127.0.0.1:${relayPort}`,
|
||||
relayUseTls: false,
|
||||
relayPublicUseTls: false,
|
||||
});
|
||||
const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
|
||||
client = new DaemonClient({
|
||||
url: buildRelayWebSocketUrl({
|
||||
endpoint: `127.0.0.1:${relayPort}`,
|
||||
useTls: false,
|
||||
serverId,
|
||||
role: "client",
|
||||
}),
|
||||
e2ee: { enabled: true, daemonPublicKeyB64 },
|
||||
reconnect: { baseDelayMs: 100, maxDelayMs: 100 },
|
||||
});
|
||||
await client.connect();
|
||||
const statuses: string[] = [];
|
||||
client.subscribeConnectionStatus((state) => statuses.push(state.status));
|
||||
for (const imageMiB of [30, 61]) {
|
||||
// Attachments are base64 in the JSON, then E2EE base64 expands the frame again.
|
||||
const images = [
|
||||
{ data: "A".repeat(4 * Math.ceil((imageMiB * 1024 * 1024) / 3)), mimeType: "image/png" },
|
||||
];
|
||||
let client: HostClient | null = null;
|
||||
try {
|
||||
const offerUrl = await getPairingOfferUrl({
|
||||
paseoHome: daemon.paseoHome,
|
||||
relayEnabled: true,
|
||||
relayEndpoint: daemon.config.relayEndpoint,
|
||||
});
|
||||
const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
|
||||
client = new HostClient({
|
||||
clientId: "paced-relay-regression",
|
||||
webSocketFactory: Object.assign((url: string) => new WebSocket(url), { supportsDrain }),
|
||||
url: buildRelayWebSocketUrl({
|
||||
endpoint: `127.0.0.1:${relayPort}`,
|
||||
useTls: false,
|
||||
serverId,
|
||||
role: "client",
|
||||
}),
|
||||
e2ee: { enabled: true, daemonPublicKeyB64 },
|
||||
reconnect: { baseDelayMs: 100, maxDelayMs: 100 },
|
||||
});
|
||||
await client.connect();
|
||||
const statuses: string[] = [];
|
||||
client.subscribeConnectionStatus((state) => statuses.push(state.status));
|
||||
for (const imageMiB of [30, 61]) {
|
||||
const images = [
|
||||
{
|
||||
data: "A".repeat(4 * Math.ceil((imageMiB * 1024 * 1024) / 3)),
|
||||
mimeType: "image/png",
|
||||
},
|
||||
];
|
||||
const creation = client.createWorkspace({
|
||||
source: { kind: "directory", path: daemon.staticDir },
|
||||
agent: {
|
||||
provider: "claude",
|
||||
cwd: daemon.staticDir,
|
||||
initialPrompt: "Describe these screens.",
|
||||
images,
|
||||
},
|
||||
});
|
||||
if (supportsDrain)
|
||||
await withTimeout(creation, 30000, "Chunked workspace creation did not settle");
|
||||
else await expect(creation).rejects.toThrow(/too large.*32\.0 MiB/);
|
||||
}
|
||||
expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 2 : 0);
|
||||
await expect(
|
||||
withTimeout(
|
||||
client.createWorkspace({
|
||||
source: { kind: "directory", path: daemon.staticDir },
|
||||
agent: {
|
||||
provider: "claude",
|
||||
cwd: daemon.staticDir,
|
||||
initialPrompt: "Describe these screens.",
|
||||
images,
|
||||
},
|
||||
}),
|
||||
10000,
|
||||
"Workspace creation did not settle",
|
||||
),
|
||||
).rejects.toThrow(/too large.*32\.0 MiB/);
|
||||
expect((await client.fetchAgents()).entries).toEqual([]);
|
||||
}
|
||||
const agent = await client.createAgent({ provider: "claude", cwd: daemon.staticDir });
|
||||
await expect(
|
||||
client.sendMessage(agent.id, "Describe these screens.", {
|
||||
client.createWorkspace({
|
||||
source: { kind: "directory", path: daemon.staticDir },
|
||||
agent: {
|
||||
provider: "claude",
|
||||
cwd: daemon.staticDir,
|
||||
initialPrompt: "Describe these screens.",
|
||||
images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow(/too large.*100\.0 MiB/);
|
||||
expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 2 : 0);
|
||||
const agent = await client.createAgent({ provider: "claude", cwd: daemon.staticDir });
|
||||
const attachment = client.sendMessage(agent.id, "Describe these screens.", {
|
||||
images: [{ data: "A".repeat(40 * 1024 * 1024), mimeType: "image/png" }],
|
||||
}),
|
||||
).rejects.toThrow(/too large.*32\.0 MiB/);
|
||||
await client.sendMessage(agent.id, "A small follow-up still works.");
|
||||
expect((await client.fetchAgents()).entries).toHaveLength(1);
|
||||
expect(statuses).toEqual(["connected"]);
|
||||
} finally {
|
||||
await client?.close();
|
||||
await daemon.close();
|
||||
await stopRelay();
|
||||
}
|
||||
}, 60000);
|
||||
});
|
||||
if (supportsDrain) await attachment;
|
||||
else await expect(attachment).rejects.toThrow(/too large.*32\.0 MiB/);
|
||||
await expect(
|
||||
client.sendMessage(agent.id, "Describe these screens.", {
|
||||
images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
|
||||
}),
|
||||
).rejects.toThrow(/too large.*100\.0 MiB/);
|
||||
await client.sendMessage(agent.id, "A small follow-up still works.");
|
||||
expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 3 : 1);
|
||||
// Respect the per-item content budget while exceeding the relay frame limit.
|
||||
const text = "x".repeat(240 * 1024);
|
||||
for (let row = 0; row < 220; row++)
|
||||
await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
|
||||
type: "user_message",
|
||||
text,
|
||||
});
|
||||
const timeline = await client.fetchAgentTimeline(agent.id, {
|
||||
projection: "canonical",
|
||||
limit: 0,
|
||||
timeout: 30000,
|
||||
});
|
||||
const delivered = timeline.entries.filter(
|
||||
(row) => row.item.type === "user_message" && row.item.text === text,
|
||||
);
|
||||
expect(delivered).toHaveLength(220);
|
||||
expect(delivered.length * Buffer.byteLength(text)).toBeGreaterThan(50 * 1024 * 1024);
|
||||
await client.ping();
|
||||
expect(statuses).toEqual(["connected"]);
|
||||
} finally {
|
||||
await client?.close();
|
||||
await daemon.close();
|
||||
await stopRelay();
|
||||
}
|
||||
},
|
||||
120000,
|
||||
);
|
||||
|
||||
test("daemon connects to relay and client ping/pong works through relay", async () => {
|
||||
process.env.PASEO_PRIMARY_LAN_IP = "192.168.1.12";
|
||||
|
||||
@@ -1836,3 +1836,58 @@ describe("list terminals across directories", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
test("a paused terminal consumer catches up through a snapshot instead of retaining every output frame", async () => {
|
||||
const { existsSync } = await import("node:fs");
|
||||
const cwd = tmpCwd();
|
||||
tempDirs.push(cwd);
|
||||
const marker = path.join(cwd, "output-complete");
|
||||
const created = await createTerminalInWorkspace(ctx.client, {
|
||||
cwd,
|
||||
options: {
|
||||
command: process.execPath,
|
||||
args: [
|
||||
"-e",
|
||||
`
|
||||
const fs = require('node:fs');
|
||||
process.stdin.setRawMode?.(true);
|
||||
process.stdin.resume();
|
||||
process.stdin.once('data', () => {
|
||||
const chunk = Buffer.alloc(256 * 1024, 120);
|
||||
let remaining = 128;
|
||||
const write = () => {
|
||||
if (remaining-- === 0) { fs.writeFileSync(${JSON.stringify(marker)}, 'done'); return; }
|
||||
process.stdout.write(chunk, write);
|
||||
};
|
||||
write();
|
||||
});
|
||||
`,
|
||||
],
|
||||
},
|
||||
});
|
||||
const ws = await connectRawWebSocket(ctx.daemon.port);
|
||||
try {
|
||||
const slot = await subscribeRawTerminal(ws, created.terminal!.id, "slow-terminal");
|
||||
await waitForRawBinaryFrame(
|
||||
ws,
|
||||
(frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Snapshot,
|
||||
);
|
||||
const capture = startTerminalOutputCapture(ws, slot);
|
||||
try {
|
||||
ws.pause();
|
||||
sendRawTerminalInput(ws, slot, "go\n");
|
||||
await waitForCondition(() => existsSync(marker), 20000);
|
||||
ws.resume();
|
||||
await waitForCondition(() => capture.snapshotCount > 0, 10000);
|
||||
expect(ws.readyState).toBe(WebSocket.OPEN);
|
||||
expect(capture.frames.reduce((total, frame) => total + frame.payloadBytes, 0)).toBeLessThan(
|
||||
32 * 1024 * 1024,
|
||||
);
|
||||
} finally {
|
||||
ws.resume();
|
||||
capture.stop();
|
||||
}
|
||||
} finally {
|
||||
await closeWebSocket(ws);
|
||||
}
|
||||
}, 40000);
|
||||
|
||||
@@ -90,7 +90,11 @@ test(
|
||||
TEST_TIMEOUT_MS,
|
||||
);
|
||||
|
||||
async function connectSocket(port: number, clientId: string): Promise<WebSocket> {
|
||||
async function connectSocket(
|
||||
port: number,
|
||||
clientId: string,
|
||||
capabilities?: Record<string, unknown>,
|
||||
): Promise<WebSocket> {
|
||||
const socket = new WebSocket(`ws://127.0.0.1:${port}/ws`);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
socket.once("open", resolve);
|
||||
@@ -98,7 +102,7 @@ async function connectSocket(port: number, clientId: string): Promise<WebSocket>
|
||||
});
|
||||
await sendAndWait(
|
||||
socket,
|
||||
{ type: "hello", clientId, clientType: "browser", protocolVersion: 1 },
|
||||
{ type: "hello", clientId, clientType: "browser", protocolVersion: 1, capabilities },
|
||||
(message) =>
|
||||
message.type === "session" &&
|
||||
message.message.type === "status" &&
|
||||
@@ -166,3 +170,97 @@ function sendAndWait(
|
||||
socket.send(JSON.stringify(message));
|
||||
return response;
|
||||
}
|
||||
|
||||
test("an awaited timeline framing overflow closes the actual daemon socket", async () => {
|
||||
const { createHostClient } = await import("@getpaseo/client/node");
|
||||
const { createPaseoApi } = await import("@getpaseo/client");
|
||||
const { default: pino } = await import("pino");
|
||||
const { Writable } = await import("node:stream");
|
||||
const messages: string[] = [];
|
||||
const logger = pino(
|
||||
{ level: "warn" },
|
||||
new Writable({
|
||||
write(chunk, _encoding, done) {
|
||||
messages.push(JSON.parse(chunk.toString()).msg);
|
||||
done();
|
||||
},
|
||||
}),
|
||||
);
|
||||
daemon = await createTestPaseoDaemon({ logger });
|
||||
const client = createHostClient(`127.0.0.1:${daemon.port}`, {
|
||||
clientId: "framing-seed",
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
await client.connect();
|
||||
const api = createPaseoApi(client);
|
||||
try {
|
||||
const workspace = await api.workspaces.create({
|
||||
source: { kind: "directory", path: daemon.staticDir },
|
||||
});
|
||||
const agents = [];
|
||||
const text = "x".repeat(240 * 1024);
|
||||
for (const [index, rows] of [350, 90].entries()) {
|
||||
const agent = await workspace.agents.create({
|
||||
config: { provider: "claude/claude-test", modeId: "full-access" },
|
||||
title: `Framing overflow ${index}`,
|
||||
});
|
||||
agents.push(agent.id);
|
||||
for (let row = 0; row < rows; row++)
|
||||
await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
|
||||
type: "user_message",
|
||||
text,
|
||||
});
|
||||
}
|
||||
const source = await connectSocket(daemon.port, "framing-overflow", {
|
||||
binary_chunking_v1: true,
|
||||
});
|
||||
sockets.push(source);
|
||||
const closed = new Promise<void>((resolve) => source.once("close", () => resolve()));
|
||||
const firstChunk = new Promise<void>((resolve) => {
|
||||
const listener = (data: RawData, binary: boolean) => {
|
||||
if (!binary || (data as Buffer)[0] !== 0x31) return;
|
||||
source.off("message", listener);
|
||||
source.pause();
|
||||
resolve();
|
||||
};
|
||||
source.on("message", listener);
|
||||
});
|
||||
const fetch = (agentId: string, requestId: string) =>
|
||||
source.send(
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "fetch_agent_timeline_request",
|
||||
requestId,
|
||||
agentId,
|
||||
direction: "tail",
|
||||
projection: "canonical",
|
||||
limit: 0,
|
||||
},
|
||||
}),
|
||||
);
|
||||
fetch(agents[0]!, "first");
|
||||
await firstChunk;
|
||||
fetch(agents[1]!, "overflow");
|
||||
await expect
|
||||
.poll(() => messages.includes("Closing physical WebSocket after outbound framing failure"), {
|
||||
timeout: 15000,
|
||||
})
|
||||
.toBe(true);
|
||||
expect(messages).not.toContain("Closing physical WebSocket at outbound high-water mark");
|
||||
source.resume();
|
||||
await closed;
|
||||
} finally {
|
||||
await api.dispose();
|
||||
await client.close();
|
||||
}
|
||||
}, 60000);
|
||||
|
||||
test("malformed chunk metadata closes the daemon socket with a protocol error", async () => {
|
||||
daemon = await createTestPaseoDaemon();
|
||||
const source = await connectSocket(daemon.port, "invalid-framing", { binary_chunking_v1: true });
|
||||
sockets.push(source);
|
||||
const closed = new Promise<number>((resolve) => source.once("close", (code) => resolve(code)));
|
||||
source.send(new Uint8Array([0x31, 2, 0, 3, 65]));
|
||||
expect(await closed).toBe(1002);
|
||||
});
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
import {
|
||||
ChunkedTransport,
|
||||
ChunkedTransportProtocolError,
|
||||
} from "@getpaseo/protocol/chunked-transport";
|
||||
import { type WorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
|
||||
import {
|
||||
createAgentMessageDelivery,
|
||||
@@ -124,8 +128,8 @@ import {
|
||||
MAX_PHYSICAL_SOCKET_BUFFERED_BYTES,
|
||||
outboundFrameByteLength,
|
||||
physicalSocketHasCapacity,
|
||||
sendBoundedPhysicalFrame,
|
||||
sendBoundedPhysicalFrameAndWait,
|
||||
sendBoundedPhysicalFrame,
|
||||
} from "./websocket/physical-socket.js";
|
||||
|
||||
const WS_CLOSE_DAEMON_AUTH_FAILED = 4401;
|
||||
@@ -440,6 +444,7 @@ function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffe
|
||||
}
|
||||
|
||||
export interface WebSocketLike {
|
||||
maxFrameBytes?: number;
|
||||
readyState: number;
|
||||
bufferedAmount?: number;
|
||||
send: (
|
||||
@@ -1273,6 +1278,67 @@ export class VoiceAssistantWebSocketServer {
|
||||
return { sockets: writableSockets, payload, bytes: outboundFrameByteLength(payload) };
|
||||
}
|
||||
|
||||
private sendHelloResponse(ws: WebSocketLike, session: Session, hello: WSHelloMessage): void {
|
||||
// The peer learns our support from this ordinary response before receiving chunks.
|
||||
this.sendToClient(ws, this.createServerInfoMessage(session));
|
||||
this.chunkedSocket(ws).enabled = hello.capabilities?.[CLIENT_CAPS.binaryChunkingV1] === true;
|
||||
}
|
||||
|
||||
private readonly chunkedSockets = new WeakMap<WebSocketLike, ChunkedTransport>();
|
||||
|
||||
private chunkedSocket(ws: WebSocketLike): ChunkedTransport {
|
||||
let transport = this.chunkedSockets.get(ws);
|
||||
if (!transport) {
|
||||
transport = new ChunkedTransport({
|
||||
frameBytes: ws.maxFrameBytes ?? 32 * 1024 * 1024,
|
||||
onFailure: (err) =>
|
||||
this.closePhysicalSocket({
|
||||
ws,
|
||||
logMessage: "Closing physical WebSocket after outbound framing failure",
|
||||
logFields: { err },
|
||||
}),
|
||||
send: (frame, waitForDrain) => {
|
||||
const params = {
|
||||
socket: ws,
|
||||
frame,
|
||||
onHighWater: () => this.closeAtOutboundHighWater(ws),
|
||||
};
|
||||
if (waitForDrain)
|
||||
return sendBoundedPhysicalFrameAndWait(params).then((sent) => {
|
||||
if (!sent) throw new Error("Source is closed");
|
||||
return undefined;
|
||||
});
|
||||
if (!sendBoundedPhysicalFrame(params)) throw new Error("Source is closed");
|
||||
},
|
||||
deliver: (frame) =>
|
||||
this.handleRawMessage(ws, typeof frame === "string" ? frame : Buffer.from(frame)),
|
||||
});
|
||||
this.chunkedSockets.set(ws, transport);
|
||||
}
|
||||
return transport;
|
||||
}
|
||||
|
||||
private socketBufferedAmount(ws: WebSocketLike): number | null {
|
||||
const queued = this.chunkedSockets.get(ws)?.queuedByteLength ?? 0;
|
||||
if (typeof ws.bufferedAmount === "number") return ws.bufferedAmount + queued;
|
||||
return queued > 0 ? queued : null;
|
||||
}
|
||||
|
||||
private sendSocketFrame(
|
||||
ws: WebSocketLike,
|
||||
frame: string | Uint8Array,
|
||||
bytes: number,
|
||||
waitForDrain = false,
|
||||
): Promise<void> {
|
||||
const transport = this.chunkedSocket(ws);
|
||||
const admittedBytes = transport.isWriting
|
||||
? bytes
|
||||
: Math.min(bytes, ws.maxFrameBytes ?? 32 * 1024 * 1024);
|
||||
if (!this.ensureOutboundCapacity(ws, admittedBytes))
|
||||
return Promise.reject(new Error("Source is closed"));
|
||||
return transport.send(frame, { waitForDrain, byteLength: bytes });
|
||||
}
|
||||
|
||||
private async sendMessageToClientAndWait(
|
||||
ws: WebSocketLike,
|
||||
message: WSOutboundMessage,
|
||||
@@ -1280,13 +1346,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
const prepared = this.prepareJsonDelivery([ws], message, true);
|
||||
if (!prepared) throw new Error("Source message delivery unavailable");
|
||||
try {
|
||||
const sent = await sendBoundedPhysicalFrameAndWait({
|
||||
socket: ws,
|
||||
frame: prepared.payload,
|
||||
frameBytes: prepared.bytes,
|
||||
onHighWater: () => this.closeAtOutboundHighWater(ws),
|
||||
});
|
||||
if (!sent) throw new Error("Source is closed");
|
||||
await this.sendSocketFrame(ws, prepared.payload, prepared.bytes, true);
|
||||
this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount);
|
||||
} catch (err) {
|
||||
this.logger.warn({ err }, "ws_send_failed");
|
||||
@@ -1313,14 +1373,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const sent = await sendBoundedPhysicalFrameAndWait({
|
||||
socket: ws,
|
||||
frame,
|
||||
onHighWater: () => this.closeAtOutboundHighWater(ws),
|
||||
});
|
||||
if (!sent) {
|
||||
throw new Error("Physical WebSocket is not open");
|
||||
}
|
||||
await this.sendSocketFrame(ws, frame, frame.byteLength, true);
|
||||
this.runtimeMetrics.recordOutboundBinaryFrame(ws.bufferedAmount);
|
||||
} catch (err) {
|
||||
this.logger.warn({ err }, "ws_send_failed");
|
||||
@@ -1335,13 +1388,9 @@ export class VoiceAssistantWebSocketServer {
|
||||
recordSent: () => void,
|
||||
): void {
|
||||
try {
|
||||
const sent = sendBoundedPhysicalFrame({
|
||||
socket: ws,
|
||||
frame,
|
||||
frameBytes,
|
||||
onHighWater: () => this.closeAtOutboundHighWater(ws),
|
||||
void this.sendSocketFrame(ws, frame, frameBytes).then(recordSent, (err) => {
|
||||
this.logger.warn({ err }, "ws_send_failed");
|
||||
});
|
||||
if (sent) recordSent();
|
||||
} catch (err) {
|
||||
this.logger.warn({ err }, "ws_send_failed");
|
||||
}
|
||||
@@ -1349,7 +1398,13 @@ export class VoiceAssistantWebSocketServer {
|
||||
|
||||
private ensureOutboundCapacity(ws: WebSocketLike, frameBytes: number): boolean {
|
||||
if (ws.readyState !== 1) return false;
|
||||
if (physicalSocketHasCapacity(ws, frameBytes)) return true;
|
||||
if (
|
||||
physicalSocketHasCapacity(
|
||||
{ bufferedAmount: this.socketBufferedAmount(ws) ?? undefined },
|
||||
frameBytes,
|
||||
)
|
||||
)
|
||||
return true;
|
||||
|
||||
this.closeAtOutboundHighWater(ws);
|
||||
return false;
|
||||
@@ -1522,16 +1577,15 @@ export class VoiceAssistantWebSocketServer {
|
||||
if (!connection) {
|
||||
return null;
|
||||
}
|
||||
if (source) return (source as WebSocketLike).bufferedAmount ?? null;
|
||||
if (source) return this.socketBufferedAmount(source as WebSocketLike);
|
||||
// Relay-attached sockets are a WebSocketLike that doesn't expose
|
||||
// bufferedAmount. Return null when no socket gives a signal so the
|
||||
// terminal fallback can't mistake "no signal" for "client keeping up";
|
||||
// a direct ws reports its real buffered bytes (0 when drained).
|
||||
let maxBuffered: number | null = null;
|
||||
for (const socket of connection.sockets) {
|
||||
if (typeof socket.bufferedAmount === "number") {
|
||||
maxBuffered = Math.max(maxBuffered ?? 0, socket.bufferedAmount);
|
||||
}
|
||||
const buffered = this.socketBufferedAmount(socket);
|
||||
if (buffered !== null) maxBuffered = Math.max(maxBuffered ?? 0, buffered);
|
||||
}
|
||||
return maxBuffered;
|
||||
},
|
||||
@@ -1568,8 +1622,9 @@ export class VoiceAssistantWebSocketServer {
|
||||
let total = 0;
|
||||
for (const socket of this.sessions.keys()) {
|
||||
if (socket instanceof VirtualHostSocket) continue;
|
||||
if (typeof socket.bufferedAmount !== "number") return null;
|
||||
total += socket.bufferedAmount;
|
||||
const buffered = this.socketBufferedAmount(socket);
|
||||
if (buffered === null) return null;
|
||||
total += buffered;
|
||||
}
|
||||
return total;
|
||||
},
|
||||
@@ -1770,7 +1825,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.externalSessionsByKey.set(sessionKey, connection);
|
||||
}
|
||||
pending.identity.sessionId = connection.session.getSessionId();
|
||||
this.sendToClient(ws, this.createServerInfoMessage(connection.session));
|
||||
this.sendHelloResponse(ws, connection.session, message);
|
||||
connection.connectionLogger.info(
|
||||
{
|
||||
...toConnectionLogFields(pending.identity),
|
||||
@@ -1906,7 +1961,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
existing.sockets.add(ws);
|
||||
this.sessions.set(ws, existing);
|
||||
pending.identity.sessionId = existing.session.getSessionId();
|
||||
this.sendToClient(ws, this.createServerInfoMessage(existing.session));
|
||||
this.sendHelloResponse(ws, existing.session, message);
|
||||
pending.connectionLogger.info(
|
||||
{
|
||||
...toConnectionLogFields(pending.identity),
|
||||
@@ -1952,6 +2007,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
usageSetting: true,
|
||||
usageSources: true,
|
||||
ownedSubscriptions: true,
|
||||
binaryChunkingV1: true,
|
||||
agentRequestReceipts: true,
|
||||
workspaceRequestReceipts: true,
|
||||
creationLifecycle: true,
|
||||
@@ -2165,10 +2221,16 @@ export class VoiceAssistantWebSocketServer {
|
||||
private bindSocketHandlers(ws: WebSocketLike): void {
|
||||
ws.on("message", (...args: unknown[]) => {
|
||||
const data = args[0] as Buffer | ArrayBuffer | Buffer[] | string;
|
||||
this.handleRawMessage(ws, data);
|
||||
try {
|
||||
this.chunkedSocket(ws).receive(typeof data === "string" ? data : bufferFromWsData(data));
|
||||
} catch (error) {
|
||||
if (!(error instanceof ChunkedTransportProtocolError)) throw error;
|
||||
ws.close(1002, "Invalid chunked transport");
|
||||
}
|
||||
});
|
||||
|
||||
ws.on("close", async (...args: unknown[]) => {
|
||||
this.chunkedSockets.get(ws)?.close();
|
||||
const code = args[0];
|
||||
const reason = args[1];
|
||||
await this.detachSocket(ws, {
|
||||
@@ -2568,7 +2630,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
identity: JSON.stringify([connection.principalId, connection.clientId]),
|
||||
permissions: () => connection.session.getPermissions(),
|
||||
routed: () => connection.session.isRouted(),
|
||||
bufferedAmount: () => ws.bufferedAmount ?? null,
|
||||
bufferedAmount: () => this.socketBufferedAmount(ws),
|
||||
delivery: connection.session.delivery,
|
||||
admit: (socket, permissions) =>
|
||||
this.attachSocket(socket, undefined, undefined, false, {
|
||||
|
||||
@@ -140,3 +140,74 @@ test("encrypted sockets do not double-count bytes already buffered by the transp
|
||||
|
||||
expect(socket.bufferedAmount).toBe(payload.byteLength + 40);
|
||||
});
|
||||
|
||||
test("50 MiB logical text and routed binary cross encrypted relay frames in both directions", async () => {
|
||||
const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport");
|
||||
const { openClientSecureChannel, openDaemonSecureChannel } = await import("@getpaseo/relay/e2ee");
|
||||
const { generateKeyPair, exportPublicKey, maxSecuredPlaintextByteLength } =
|
||||
await import("@getpaseo/relay");
|
||||
const { encodeHostRouteFrame, decodeHostRouteFrame } =
|
||||
await import("@getpaseo/protocol/binary-frames/index");
|
||||
type Transport = import("@getpaseo/relay/e2ee").Transport;
|
||||
const limit = 32 * 1024 * 1024;
|
||||
const wireSizes: number[] = [];
|
||||
const endpoint = (): Transport => ({
|
||||
send: () => {},
|
||||
close: () => {},
|
||||
onmessage: null,
|
||||
onclose: null,
|
||||
onerror: null,
|
||||
});
|
||||
const left = endpoint();
|
||||
const right = endpoint();
|
||||
const wire = (destination: Transport) => (data: string | ArrayBuffer) => {
|
||||
const size = typeof data === "string" ? Buffer.byteLength(data) : data.byteLength;
|
||||
expect(size).toBeLessThanOrEqual(limit);
|
||||
wireSizes.push(size);
|
||||
queueMicrotask(() => destination.onmessage?.({ data, isBinary: typeof data !== "string" }));
|
||||
};
|
||||
left.send = wire(right);
|
||||
right.send = wire(left);
|
||||
const key = generateKeyPair();
|
||||
const delivered: (string | Uint8Array)[][] = [[], []];
|
||||
let receivers: InstanceType<typeof ChunkedTransport>[] = [];
|
||||
const incoming = (index: number) => (data: string | ArrayBuffer) =>
|
||||
receivers[index]!.receive(typeof data === "string" ? data : new Uint8Array(data));
|
||||
const daemonPromise = openDaemonSecureChannel(right, {
|
||||
daemonKeyPair: key,
|
||||
events: { onmessage: incoming(1) },
|
||||
});
|
||||
const client = await openClientSecureChannel(left, {
|
||||
daemonPublicKeyB64: exportPublicKey(key.publicKey),
|
||||
events: { onmessage: incoming(0) },
|
||||
});
|
||||
const daemon = await daemonPromise;
|
||||
receivers = [client, daemon].map(
|
||||
(channel, index) =>
|
||||
new ChunkedTransport({
|
||||
frameBytes: maxSecuredPlaintextByteLength(limit),
|
||||
send: (frame) => channel.send(typeof frame === "string" ? frame : frame.slice().buffer),
|
||||
deliver: (frame) => {
|
||||
delivered[index]!.push(frame);
|
||||
},
|
||||
}),
|
||||
);
|
||||
for (const receiver of receivers) receiver.enabled = true;
|
||||
const text = "x".repeat(50 * 1024 * 1024);
|
||||
wireSizes.length = 0;
|
||||
await receivers[1]!.send(text);
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(delivered[0]).toEqual([text]);
|
||||
expect(wireSizes.length).toBeGreaterThan(1);
|
||||
const routed = encodeHostRouteFrame("route", new Uint8Array(50 * 1024 * 1024).fill(7));
|
||||
await receivers[0]!.send(routed);
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(delivered[1]).toHaveLength(1);
|
||||
const received = decodeHostRouteFrame(delivered[1]![0] as Uint8Array)!;
|
||||
expect(received.routeId).toBe("route");
|
||||
expect(
|
||||
Buffer.from(received.payload).equals(Buffer.from(new Uint8Array(50 * 1024 * 1024).fill(7))),
|
||||
).toBe(true);
|
||||
client.close();
|
||||
daemon.close();
|
||||
}, 30_000);
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { maxSecuredPlaintextByteLength } from "@getpaseo/relay";
|
||||
import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { MAX_PHYSICAL_SOCKET_BUFFERED_BYTES } from "./physical-socket.js";
|
||||
|
||||
@@ -8,6 +10,7 @@ export interface EncryptedRelayChannel {
|
||||
}
|
||||
|
||||
export interface EncryptedRelaySocket {
|
||||
readonly maxFrameBytes: number;
|
||||
readonly readyState: number;
|
||||
readonly bufferedAmount: number;
|
||||
send: (data: string | Uint8Array | ArrayBuffer) => void | Promise<void>;
|
||||
@@ -43,6 +46,7 @@ export function createEncryptedRelaySocket(params: {
|
||||
});
|
||||
|
||||
return {
|
||||
maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES),
|
||||
get readyState() {
|
||||
return readyState;
|
||||
},
|
||||
|
||||
@@ -120,3 +120,96 @@ test("the awaitable physical send rejects callback errors", async () => {
|
||||
}),
|
||||
).rejects.toThrow("send failed");
|
||||
});
|
||||
|
||||
test("ordinary frames expose a paused real socket's backlog and terminate at the physical backstop", async () => {
|
||||
const { once } = await import("node:events");
|
||||
const { WebSocket, WebSocketServer } = await import("ws");
|
||||
const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport");
|
||||
const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
|
||||
await once(server, "listening");
|
||||
const connected = once(server, "connection");
|
||||
const address = server.address();
|
||||
if (typeof address === "string" || !address) throw new Error("Missing server address");
|
||||
const client = new WebSocket(`ws://127.0.0.1:${address.port}`);
|
||||
await once(client, "open");
|
||||
const [socket] = (await connected) as [InstanceType<typeof WebSocket>];
|
||||
const tcp = (socket as unknown as { _socket: import("node:net").Socket })._socket;
|
||||
let terminated = false;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 32 * 1024 * 1024,
|
||||
deliver: () => {},
|
||||
send: (frame, wait) => {
|
||||
const options = {
|
||||
socket,
|
||||
frame,
|
||||
onHighWater: () => {
|
||||
terminated = true;
|
||||
socket.terminate();
|
||||
},
|
||||
};
|
||||
if (wait) return sendBoundedPhysicalFrameAndWait(options).then(() => undefined);
|
||||
sendBoundedPhysicalFrame(options);
|
||||
},
|
||||
});
|
||||
try {
|
||||
client.pause();
|
||||
tcp.cork();
|
||||
const frame = new Uint8Array(256 * 1024);
|
||||
for (let count = 0; count < 200; count++) await transport.send(frame);
|
||||
expect(socket.bufferedAmount).toBeGreaterThan(4 * 1024 * 1024);
|
||||
expect(transport.queuedByteLength).toBe(0);
|
||||
for (let count = 0; count < 100; count++) {
|
||||
if (terminated) break;
|
||||
await transport.send(frame);
|
||||
}
|
||||
expect(terminated).toBe(true);
|
||||
} finally {
|
||||
transport.close();
|
||||
client.terminate();
|
||||
socket.terminate();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
test("an awaited framing failure terminates its real socket immediately", async () => {
|
||||
const { once } = await import("node:events");
|
||||
const { WebSocket, WebSocketServer } = await import("ws");
|
||||
const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport");
|
||||
const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
|
||||
await once(server, "listening");
|
||||
const connected = once(server, "connection");
|
||||
const address = server.address();
|
||||
if (typeof address === "string" || !address) throw new Error("Missing server address");
|
||||
const client = new WebSocket(`ws://127.0.0.1:${address.port}`);
|
||||
await once(client, "open");
|
||||
const [socket] = (await connected) as [InstanceType<typeof WebSocket>];
|
||||
const tcp = (socket as unknown as { _socket: import("node:net").Socket })._socket;
|
||||
const transport = new ChunkedTransport({
|
||||
frameBytes: 32 * 1024 * 1024,
|
||||
deliver: () => {},
|
||||
onFailure: () => socket.terminate(),
|
||||
send: (frame) =>
|
||||
sendBoundedPhysicalFrameAndWait({
|
||||
socket,
|
||||
frame,
|
||||
onHighWater: () => socket.terminate(),
|
||||
}).then(() => undefined),
|
||||
});
|
||||
transport.enabled = true;
|
||||
try {
|
||||
client.pause();
|
||||
tcp.cork();
|
||||
const closed = once(socket, "close");
|
||||
const first = transport.send(new Uint8Array(100 * 1024 * 1024), { waitForDrain: true });
|
||||
const overflow = transport.send(new Uint8Array([1]), { waitForDrain: true });
|
||||
const settled = Promise.allSettled([first, overflow]);
|
||||
await closed;
|
||||
expect(socket.readyState).toBe(WebSocket.CLOSED);
|
||||
expect((await settled).map((result) => result.status)).toEqual(["rejected", "rejected"]);
|
||||
} finally {
|
||||
transport.close();
|
||||
client.terminate();
|
||||
socket.terminate();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user