perf(app): reduce startup websocket work

Make provider catalogs compact and content-addressed so repeat launches can reuse the persisted snapshot instead of transferring it again. Remove eager global catalog loading and the stopped-agent defensive tail fetch, while keeping sequence-gap and reconnect recovery as the correctness paths. Add an opt-in production-ID Android profiling build for local startup traces.
This commit is contained in:
Mohamed Boudra
2026-08-04 15:45:55 +02:00
parent 8e5b6876b4
commit 7dbeff8290
36 changed files with 1209 additions and 171 deletions
+4
View File
@@ -1,9 +1,11 @@
const fs = require("node:fs");
const path = require("node:path");
const pkg = require("./package.json");
const withAndroidProfileable = require("./plugins/with-android-profileable");
const withFdroidAutolinking = require("./plugins/with-fdroid-autolinking");
const appVariant = process.env.APP_VARIANT ?? "production";
const isFdroidBuild = process.env.PASEO_FDROID_BUILD === "1";
const isProfileBuild = process.env.PASEO_PROFILE_BUILD === "1";
const buildProfile = isFdroidBuild
? {
@@ -193,6 +195,7 @@ export default {
},
],
...buildProfile.fdroidPlugins,
...(isProfileBuild ? [withAndroidProfileable] : []),
],
experiments: {
typedRoutes: true,
@@ -201,6 +204,7 @@ export default {
},
extra: {
fdroidBuild: isFdroidBuild,
profileBuild: isProfileBuild,
router: {},
eas: {
projectId: "0e7f65ce-0367-46c8-a238-2b65963d235a",
@@ -915,39 +915,6 @@ test.describe("Agent message submission", () => {
}
});
test("reconciles an interrupted submission when live acknowledgements are missed", async ({
page,
}) => {
const gate = await installDaemonWebSocketGate(page);
const title = "Missed submission acknowledgements";
const prompt = "Interrupt without delivering live acknowledgements.";
const agent = await seedMockAgentWorkspace({
repoPrefix: "submission-missed-acknowledgements-",
title,
model: "one-minute-stream",
});
try {
await openAgentRoute(page, agent);
await expectComposerVisible(page);
gate.setAgentStreamItemSuppressed("user_message", true);
gate.setAgentStreamEventSuppressed("turn_canceled", true);
await submitMessage(page, prompt);
await gate.waitForAgentStreamItem("user_message");
await expectRunningAgentChrome(page, title);
await cancelAgent(page);
await gate.waitForAgentStreamEvent("turn_canceled");
await agent.client.waitForFinish(agent.agentId, 30_000);
await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(1);
await expectAgentSurfacesIdle(page, title);
} finally {
gate.setAgentStreamItemSuppressed("user_message", false);
gate.setAgentStreamEventSuppressed("turn_canceled", false);
await agent.cleanup();
}
});
test("keeps layout stable when submitting to an agent with existing history", async ({
page,
submissionScenario,
@@ -0,0 +1,15 @@
apply plugin: 'com.android.library'
group = 'sh.paseo.trace'
version = '0.1.0'
def expoModulesCorePlugin = new File(project(":expo-modules-core").projectDir.absolutePath, "ExpoModulesCorePlugin.gradle")
apply from: expoModulesCorePlugin
applyKotlinExpoModulesCorePlugin()
useCoreDependencies()
useExpoPublishing()
useDefaultAndroidSdkVersions()
android {
namespace "sh.paseo.trace"
}
@@ -0,0 +1 @@
<manifest xmlns:android="http://schemas.android.com/apk/res/android" />
@@ -0,0 +1,19 @@
package sh.paseo.trace
import android.os.Trace
import expo.modules.kotlin.modules.Module
import expo.modules.kotlin.modules.ModuleDefinition
class PaseoNativeTraceModule : Module() {
override fun definition() = ModuleDefinition {
Name("PaseoNativeTrace")
Function("beginSection") { name: String ->
Trace.beginSection(name.take(127))
}
Function("endSection") {
Trace.endSection()
}
}
}
@@ -0,0 +1,6 @@
{
"platforms": ["android"],
"android": {
"modules": ["sh.paseo.trace.PaseoNativeTraceModule"]
}
}
@@ -0,0 +1,5 @@
{
"name": "paseo-native-trace",
"version": "0.1.0",
"private": true
}
@@ -0,0 +1,15 @@
const { withAndroidManifest } = require("expo/config-plugins");
function withAndroidProfileable(config) {
return withAndroidManifest(config, (modConfig) => {
const application = modConfig.modResults.manifest.application?.[0];
if (!application) {
throw new Error("Could not enable Android profiling without an application manifest entry");
}
application.profileable = [{ $: { "android:shell": "true" } }];
return modConfig;
});
}
module.exports = withAndroidProfileable;
@@ -2,3 +2,6 @@ import Constants from "expo-constants";
/** F-Droid build without proprietary camera or notification dependencies. */
export const isFdroidBuild = Constants.expoConfig?.extra?.fdroidBuild === true;
/** Production-like Android build with local profiling enabled. */
export const isProfileBuild = Constants.expoConfig?.extra?.profileBuild === true;
@@ -10,7 +10,6 @@ import {
createSetAgentInitializing,
refreshAgentInitializationTimeout,
} from "@/hooks/use-agent-initialization";
import { prefetchProvidersSnapshot } from "@/hooks/use-providers-snapshot";
import { generateMessageId, type StreamItem } from "@/types/stream";
import {
createSessionAgentStreamReducerQueue,
@@ -494,19 +493,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
});
}, [client, serverId, updateSessionServerInfo]);
useEffect(() => {
if (!isConnected) {
return;
}
const serverInfo = client.getLastServerInfoMessage();
if (!serverInfo?.features?.providersSnapshot) {
return;
}
prefetchProvidersSnapshot(serverId, client);
}, [client, isConnected, serverId]);
useEffect(() => {
const unregister = voiceRuntime?.registerSession({
serverId,
@@ -739,19 +725,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
viewedTimelineSyncRef.current?.setConnected(isConnected);
}, [isConnected]);
useEffect(
() =>
getHostRuntimeStore().subscribeAgentStoppedRunning(serverId, (agentId) => {
const session = useSessionStore.getState().sessions[serverId];
const timeline = selectAgentTimelineState(session, agentId);
if (timeline.status === "synced" && timeline.newer === "available") {
return;
}
viewedTimelineSyncRef.current?.reconcileAgent(agentId);
}),
[serverId],
);
// Daemon message handlers - directly update Zustand store
useEffect(() => {
const agentStreamReducerQueue = createSessionAgentStreamReducerQueue({
@@ -0,0 +1,65 @@
import { describe, expect, it } from "vitest";
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
import { compactProviderSnapshot } from "@getpaseo/protocol/provider-snapshot-codec";
import { createProviderSnapshotCache } from "./provider-snapshot-cache";
function createStorage() {
const values = new Map<string, string>();
return {
values,
async getItem(key: string) {
return values.get(key) ?? null;
},
async setItem(key: string, value: string) {
values.set(key, value);
},
async removeItem(key: string) {
values.delete(key);
},
};
}
describe("provider snapshot cache", () => {
it("round-trips compact snapshots with shared thinking option references", async () => {
const thinkingOptions = [{ id: "high", label: "High", isDefault: true }];
const entries: ProviderSnapshotEntry[] = [
{
provider: "pi",
status: "ready",
enabled: true,
models: ["one", "two"].map((id) => ({
provider: "pi",
id,
label: id,
thinkingOptions,
defaultThinkingOptionId: "high",
})),
},
];
const storage = createStorage();
const cache = createProviderSnapshotCache(storage);
await cache.write({
serverId: "server-1",
cwd: "/repo",
hash: "snapshot-hash",
generatedAt: "2026-08-04T00:00:00.000Z",
compactSnapshot: compactProviderSnapshot(entries),
});
const cached = await cache.read("server-1", "/repo");
expect(cached?.entries).toEqual(entries);
expect(cached?.entries[0]?.models?.[0]?.thinkingOptions).toBe(
cached?.entries[0]?.models?.[1]?.thinkingOptions,
);
});
it("discards an invalid cache record", async () => {
const storage = createStorage();
const cache = createProviderSnapshotCache(storage);
storage.values.set('@paseo/provider-snapshot/v1:["server-1","/repo"]', "not json");
await expect(cache.read("server-1", "/repo")).resolves.toBeNull();
expect(storage.values.size).toBe(0);
});
});
@@ -0,0 +1,129 @@
import AsyncStorage from "@react-native-async-storage/async-storage";
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
import {
expandProviderSnapshot,
type CompactProviderSnapshot,
} from "@getpaseo/protocol/provider-snapshot-codec";
import { CompactProviderSnapshotSchema } from "@getpaseo/protocol/messages";
const CACHE_VERSION = 1;
const CACHE_KEY_PREFIX = "@paseo/provider-snapshot/v1";
interface ProviderSnapshotStorage {
getItem(key: string): Promise<string | null>;
setItem(key: string, value: string): Promise<void>;
removeItem(key: string): Promise<void>;
}
interface StoredProviderSnapshot {
version: typeof CACHE_VERSION;
hash: string;
generatedAt: string;
compactSnapshot: CompactProviderSnapshot;
}
export interface CachedProviderSnapshot extends StoredProviderSnapshot {
entries: ProviderSnapshotEntry[];
}
export interface ProviderSnapshotCache {
read(serverId: string, cwd: string | null): Promise<CachedProviderSnapshot | null>;
write(input: {
serverId: string;
cwd: string | null;
hash: string;
generatedAt: string;
compactSnapshot: CompactProviderSnapshot;
}): Promise<void>;
}
export class ProviderSnapshotCacheMissError extends Error {
constructor() {
super("Daemon returned not-modified without a cached provider snapshot");
this.name = "ProviderSnapshotCacheMissError";
}
}
function cacheKey(serverId: string, cwd: string | null): string {
return `${CACHE_KEY_PREFIX}:${JSON.stringify([serverId, cwd])}`;
}
function parseStoredProviderSnapshot(value: string): StoredProviderSnapshot | null {
const parsed: unknown = JSON.parse(value);
if (typeof parsed !== "object" || parsed === null) return null;
const version = Reflect.get(parsed, "version");
const hash = Reflect.get(parsed, "hash");
const generatedAt = Reflect.get(parsed, "generatedAt");
const compactSnapshot = CompactProviderSnapshotSchema.safeParse(
Reflect.get(parsed, "compactSnapshot"),
);
if (
version !== CACHE_VERSION ||
typeof hash !== "string" ||
typeof generatedAt !== "string" ||
!compactSnapshot.success
) {
return null;
}
return {
version,
hash,
generatedAt,
compactSnapshot: compactSnapshot.data,
};
}
async function discardCachedSnapshot(storage: ProviderSnapshotStorage, key: string): Promise<void> {
try {
await storage.removeItem(key);
} catch (error) {
if (!(error instanceof Error)) throw error;
}
}
export function createProviderSnapshotCache(
storage: ProviderSnapshotStorage,
): ProviderSnapshotCache {
return {
async read(serverId, cwd) {
const key = cacheKey(serverId, cwd);
let value: string | null;
try {
value = await storage.getItem(key);
} catch (error) {
if (error instanceof Error) return null;
throw error;
}
if (value === null) return null;
try {
const stored = parseStoredProviderSnapshot(value);
if (!stored) {
await discardCachedSnapshot(storage, key);
return null;
}
return { ...stored, entries: expandProviderSnapshot(stored.compactSnapshot) };
} catch (error) {
if (!(error instanceof SyntaxError) && !(error instanceof RangeError)) {
throw error;
}
await discardCachedSnapshot(storage, key);
return null;
}
},
async write(input) {
const stored: StoredProviderSnapshot = {
version: CACHE_VERSION,
hash: input.hash,
generatedAt: input.generatedAt,
compactSnapshot: input.compactSnapshot,
};
try {
await storage.setItem(cacheKey(input.serverId, input.cwd), JSON.stringify(stored));
} catch (error) {
if (!(error instanceof Error)) throw error;
}
},
};
}
export const providerSnapshotCache = createProviderSnapshotCache(AsyncStorage);
@@ -21,11 +21,13 @@ export function providersSnapshotQueryKey(serverId: string | null, cwd?: string
export function providersSnapshotRequestOptions(input: {
cwd?: string | null;
providers?: AgentProvider[];
ifNoneMatch?: string;
}) {
const normalizedCwd = normalizeProvidersSnapshotCwd(input.cwd);
return {
...(normalizedCwd ? { cwd: normalizedCwd } : {}),
...(input.providers ? { providers: input.providers } : {}),
...(input.ifNoneMatch ? { ifNoneMatch: input.ifNoneMatch } : {}),
};
}
+17 -1
View File
@@ -8,7 +8,12 @@ import { agentCommandsQueryRoot } from "@/hooks/agent-commands-query";
import { orderCheckoutDiffFiles } from "@/git/diff-order";
import { daemonConfigQueryKey } from "@/data/daemon-config";
import { daemonPairingOfferQueryKey } from "@/data/daemon-pairing";
import { providersSnapshotQueryKey, providersSnapshotQueryRoot } from "@/data/providers-snapshot";
import { providerSnapshotCache, type ProviderSnapshotCache } from "@/data/provider-snapshot-cache";
import {
normalizeProvidersSnapshotCwd,
providersSnapshotQueryKey,
providersSnapshotQueryRoot,
} from "@/data/providers-snapshot";
type ProvidersSnapshotUpdateMessage = Extract<
SessionOutboundMessage,
@@ -184,6 +189,7 @@ export function applyProvidersSnapshotUpdate(input: {
serverId: string;
queryClient: QueryClient;
message: ProvidersSnapshotUpdate;
cache?: ProviderSnapshotCache;
}): void {
if (input.message.type !== "providers_snapshot_update") {
return;
@@ -194,6 +200,16 @@ export function applyProvidersSnapshotUpdate(input: {
generatedAt: input.message.payload.generatedAt,
requestId: "providers_snapshot_update",
});
const { compactSnapshot, snapshotHash } = input.message.payload;
if (compactSnapshot && snapshotHash) {
void (input.cache ?? providerSnapshotCache).write({
serverId: input.serverId,
cwd: normalizeProvidersSnapshotCwd(input.message.payload.cwd),
hash: snapshotHash,
generatedAt: input.message.payload.generatedAt,
compactSnapshot,
});
}
void input.queryClient.invalidateQueries({
queryKey: agentCommandsQueryRoot(input.serverId),
exact: false,
@@ -3,6 +3,8 @@ import { beforeEach, describe, expect, it } from "vitest";
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
import type { ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
import { compactProviderSnapshot } from "@getpaseo/protocol/provider-snapshot-codec";
import type { CachedProviderSnapshot, ProviderSnapshotCache } from "@/data/provider-snapshot-cache";
import { draftAgentCommandsQueryKey } from "@/hooks/agent-commands-query";
import { applyProvidersSnapshotUpdate, type ProvidersSnapshotUpdate } from "@/data/push-router";
import {
@@ -79,6 +81,21 @@ function codexEntry(
const readyCodexModel = { provider: "codex", id: "gpt-5.4", label: "GPT-5.4" } as const;
const serverId = "server-1";
function createCache(initial: CachedProviderSnapshot | null = null): ProviderSnapshotCache & {
writes: Parameters<ProviderSnapshotCache["write"]>[0][];
} {
const writes: Parameters<ProviderSnapshotCache["write"]>[0][] = [];
return {
writes,
async read() {
return initial;
},
async write(input) {
writes.push(input);
},
};
}
describe("providersSnapshotQueryKey", () => {
it("uses separate keys for home and workspace scopes", () => {
expect(providersSnapshotQueryKey(serverId)).toEqual(["providersSnapshot", serverId, "home"]);
@@ -95,7 +112,7 @@ describe("fetchProvidersSnapshot", () => {
it("sends no cwd for the home scope", async () => {
const client = createClient({ snapshots: [providersSnapshot([])] });
await fetchProvidersSnapshot({ client, cwd: null });
await fetchProvidersSnapshot({ client, serverId, cwd: null, cache: createCache() });
expect(client.getCalls).toEqual([{}]);
});
@@ -103,10 +120,78 @@ describe("fetchProvidersSnapshot", () => {
it("sends the workspace cwd for the workspace scope", async () => {
const client = createClient({ snapshots: [providersSnapshot([])] });
await fetchProvidersSnapshot({ client, cwd: "/repo-a" });
await fetchProvidersSnapshot({
client,
serverId,
cwd: "/repo-a",
cache: createCache(),
});
expect(client.getCalls).toEqual([{ cwd: "/repo-a" }]);
});
it("reuses a cached snapshot when the daemon reports its hash unchanged", async () => {
const entries = [codexEntry("ready", [readyCodexModel])];
const compactSnapshot = compactProviderSnapshot(entries);
const cache = createCache({
version: 1,
hash: "snapshot-hash",
generatedAt: "2026-01-01T00:00:00.000Z",
compactSnapshot,
entries,
});
const client = createClient({
snapshots: [
{
entries: [],
snapshotHash: "snapshot-hash",
notModified: true,
generatedAt: "2026-01-02T00:00:00.000Z",
requestId: "snapshot-2",
},
],
});
const snapshot = await fetchProvidersSnapshot({
client,
serverId,
cwd: "/repo-a",
cache,
});
expect(client.getCalls).toEqual([{ cwd: "/repo-a", ifNoneMatch: "snapshot-hash" }]);
expect(snapshot.entries).toBe(entries);
expect(cache.writes).toEqual([]);
});
it("persists a changed compact snapshot for the next launch", async () => {
const entries = [codexEntry("ready", [readyCodexModel])];
const compactSnapshot = compactProviderSnapshot(entries);
const cache = createCache();
const client = createClient({
snapshots: [
{
entries,
compactSnapshot,
snapshotHash: "next-hash",
generatedAt: "2026-01-02T00:00:00.000Z",
requestId: "snapshot-2",
},
],
});
await fetchProvidersSnapshot({ client, serverId, cwd: "/repo-a", cache });
expect(cache.writes).toEqual([
{
serverId,
cwd: "/repo-a",
hash: "next-hash",
generatedAt: "2026-01-02T00:00:00.000Z",
compactSnapshot,
},
]);
});
});
describe("refreshAndApplyProvidersSnapshot", () => {
@@ -127,6 +212,7 @@ describe("refreshAndApplyProvidersSnapshot", () => {
serverId,
cwd: null,
providers: ["codex"],
cache: createCache(),
});
expect(client.refreshCalls).toEqual([{ providers: ["codex"] }]);
@@ -147,6 +233,7 @@ describe("refreshAndApplyProvidersSnapshot", () => {
serverId,
cwd: "/repo-a",
providers: ["codex"],
cache: createCache(),
});
expect(client.refreshCalls).toEqual([{ cwd: "/repo-a", providers: ["codex"] }]);
@@ -166,6 +253,7 @@ describe("refreshAndApplyProvidersSnapshot", () => {
queryClient,
serverId,
cwd: null,
cache: createCache(),
});
expect(
@@ -186,6 +274,7 @@ describe("refreshAndApplyProvidersSnapshot", () => {
queryClient,
serverId,
cwd: "/repo-a",
cache: createCache(),
});
expect(queryClient.getQueryState(providersSnapshotQueryKey(serverId))?.isInvalidated).toBe(
@@ -209,6 +298,7 @@ describe("refreshAndApplyProvidersSnapshot", () => {
queryClient,
serverId,
cwd: "/repo-a",
cache: createCache(),
});
expect(queryClient.getQueryState(commandsKey)?.isInvalidated).toBe(true);
@@ -266,6 +356,27 @@ describe("applyProvidersSnapshotUpdate", () => {
);
});
it("persists compact push updates", () => {
const entries = [codexEntry("ready", [readyCodexModel])];
const compactSnapshot = compactProviderSnapshot(entries);
const cache = createCache();
const message = updateMessage(entries, "/repo-a");
message.payload.compactSnapshot = compactSnapshot;
message.payload.snapshotHash = "push-hash";
applyProvidersSnapshotUpdate({ serverId, queryClient, message, cache });
expect(cache.writes).toEqual([
{
serverId,
cwd: "/repo-a",
hash: "push-hash",
generatedAt: "2026-01-01T00:00:01.000Z",
compactSnapshot,
},
]);
});
it("applies Windows daemon updates to app-normalized workspace paths", () => {
const workspaceCwd = "C:/Users/Ezekiel Bulver/project";
const daemonCwd = "C:\\Users\\Ezekiel Bulver\\project";
@@ -8,6 +8,11 @@ import { useHostRuntimeClient, useHostRuntimeIsConnected } from "@/runtime/host-
import { useSessionStore } from "@/stores/session-store";
import { useReplicaQuery } from "@/data/query";
import { queryClient as singletonQueryClient } from "@/data/query-client";
import {
providerSnapshotCache,
ProviderSnapshotCacheMissError,
type ProviderSnapshotCache,
} from "@/data/provider-snapshot-cache";
import { agentCommandsQueryRoot } from "@/hooks/agent-commands-query";
import {
isProvidersSnapshotHomeScope,
@@ -29,9 +34,31 @@ export type ProvidersSnapshotClient = Pick<
export async function fetchProvidersSnapshot(input: {
client: ProvidersSnapshotClient;
serverId: string;
cwd: string | null;
cache?: ProviderSnapshotCache;
}): Promise<GetProvidersSnapshotResult> {
return input.client.getProvidersSnapshot(providersSnapshotRequestOptions({ cwd: input.cwd }));
const cache = input.cache ?? providerSnapshotCache;
const cached = await cache.read(input.serverId, input.cwd);
const snapshot = await input.client.getProvidersSnapshot(
providersSnapshotRequestOptions({ cwd: input.cwd, ifNoneMatch: cached?.hash }),
);
if (snapshot.notModified) {
if (!cached) {
throw new ProviderSnapshotCacheMissError();
}
return { ...snapshot, entries: cached.entries };
}
if (snapshot.compactSnapshot && snapshot.snapshotHash) {
await cache.write({
serverId: input.serverId,
cwd: input.cwd,
hash: snapshot.snapshotHash,
generatedAt: snapshot.generatedAt,
compactSnapshot: snapshot.compactSnapshot,
});
}
return snapshot;
}
export async function refreshAndApplyProvidersSnapshot(input: {
@@ -40,11 +67,17 @@ export async function refreshAndApplyProvidersSnapshot(input: {
serverId: string;
cwd: string | null;
providers?: AgentProvider[];
cache?: ProviderSnapshotCache;
}): Promise<RefreshProvidersSnapshotResult> {
const refreshResult = await input.client.refreshProvidersSnapshot(
providersSnapshotRequestOptions({ cwd: input.cwd, providers: input.providers }),
);
const snapshot = await fetchProvidersSnapshot({ client: input.client, cwd: input.cwd });
const snapshot = await fetchProvidersSnapshot({
client: input.client,
serverId: input.serverId,
cwd: input.cwd,
cache: input.cache,
});
input.queryClient.setQueryData(providersSnapshotQueryKey(input.serverId, input.cwd), snapshot);
void input.queryClient.invalidateQueries({
queryKey: agentCommandsQueryRoot(input.serverId),
@@ -113,10 +146,10 @@ export function useProvidersSnapshot(
enabled: Boolean(enabled && supportsSnapshot && serverId && client && isConnected),
pushEvent: "providers_snapshot_update",
queryFn: async () => {
if (!client) {
if (!client || !serverId) {
throw new Error(t("workspace.terminal.hostDisconnected"));
}
return fetchProvidersSnapshot({ client, cwd });
return fetchProvidersSnapshot({ client, serverId, cwd });
},
});
@@ -180,6 +213,6 @@ export function prefetchProvidersSnapshot(
void singletonQueryClient.prefetchQuery({
queryKey,
staleTime: Infinity,
queryFn: () => fetchProvidersSnapshot({ client, cwd }),
queryFn: () => fetchProvidersSnapshot({ client, serverId, cwd }),
});
}
@@ -0,0 +1,38 @@
import type { DaemonClientTrace } from "@getpaseo/client/internal/daemon-client";
import { requireOptionalNativeModule } from "expo-modules-core";
import { isProfileBuild } from "@/constants/build-profile";
interface PaseoNativeTraceModule {
beginSection(name: string): void;
endSection(): void;
}
const traceModule = requireOptionalNativeModule<PaseoNativeTraceModule>("PaseoNativeTrace");
export const nativePerformanceTrace: DaemonClientTrace = {
isEnabled() {
return isProfileBuild && traceModule !== null;
},
beginSection(name, args) {
traceModule?.beginSection(formatSectionName(name, args));
},
endSection() {
traceModule?.endSection();
},
};
function formatSectionName(name: string, args?: Record<string, string>): string {
if (!args) {
return name;
}
const fields = Object.entries(args).map(([key, value]) => `${key}=${value}`);
return `${name}|${fields.join(";")}`;
}
export function traceInstant(name: string, args?: Record<string, string>): void {
if (!nativePerformanceTrace.isEnabled()) {
return;
}
nativePerformanceTrace.beginSection(name, args);
nativePerformanceTrace.endSection();
}
+5 -23
View File
@@ -56,6 +56,7 @@ import { resolveComposerAttachmentSubmitFormat } from "@/composer/attachments/su
import { encodeImages } from "@/utils/encode-images";
import { DirectorySync, type RefreshAgentDirectoryResult } from "@/runtime/directory-sync";
import { ReplicaCache } from "@/runtime/replica-cache";
import { nativePerformanceTrace } from "@/performance/native-trace";
export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error";
export type HostRegistryStatus = "loading" | "ready";
@@ -483,6 +484,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
appVersion: resolveAppVersion() ?? undefined,
runtimeGeneration,
capabilities: appCapabilities,
trace: nativePerformanceTrace,
};
if (connection.type === "directSocket" || connection.type === "directPipe") {
return new DaemonClient({
@@ -521,6 +523,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps {
...(host.serverId ? { serverId: host.serverId } : {}),
...(timeoutMs !== undefined ? { timeoutMs } : {}),
capabilities: appCapabilities,
trace: nativePerformanceTrace,
}),
getClientId: () => getOrCreateClientId(),
mountClientHandlers: ({ client, host }) => {
@@ -1341,7 +1344,6 @@ interface AgentDirectoryRefreshInput {
export class HostRuntimeStore {
private controllers = new Map<string, HostRuntimeController>();
private serverListeners = new Map<string, Set<() => void>>();
private agentStoppedRunningListeners = new Map<string, Set<(agentId: string) => void>>();
private globalListeners = new Set<() => void>();
private hostListListeners = new Set<() => void>();
private version = 0;
@@ -1589,12 +1591,11 @@ export class HostRuntimeStore {
rekeyMap(this.controllers, oldServerId, newServerId);
rekeyMap(this.lastConnectionStatusByServer, oldServerId, newServerId);
rekeyMap(this.directoryBootstrapInFlight, oldServerId, newServerId);
rekeyMap(this.agentStoppedRunningListeners, oldServerId, newServerId);
this.replicaCache.reconcileServerId(oldServerId, newServerId);
this.directorySyncByServer.get(oldServerId)?.dispose();
this.directorySyncByServer.delete(oldServerId);
const directory = new DirectorySync(newServerId, {
onAgentStoppedRunning: (agentId) => this.onAgentStoppedRunning(newServerId, agentId),
onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId),
markAgentLoading: () => controller.markAgentDirectorySyncLoading(),
markAgentReady: () => controller.markAgentDirectorySyncReady(),
markAgentError: (error) => controller.markAgentDirectorySyncError(error),
@@ -1977,7 +1978,7 @@ export class HostRuntimeStore {
this.directorySyncByServer.set(
host.serverId,
new DirectorySync(host.serverId, {
onAgentStoppedRunning: (agentId) => this.onAgentStoppedRunning(host.serverId, agentId),
onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(host.serverId, agentId),
markAgentLoading: () => controller.markAgentDirectorySyncLoading(),
markAgentReady: () => controller.markAgentDirectorySyncReady(),
markAgentError: (error) => controller.markAgentDirectorySyncError(error),
@@ -2176,18 +2177,6 @@ export class HostRuntimeStore {
};
}
subscribeAgentStoppedRunning(serverId: string, listener: (agentId: string) => void): () => void {
const listeners = this.agentStoppedRunningListeners.get(serverId) ?? new Set();
listeners.add(listener);
this.agentStoppedRunningListeners.set(serverId, listeners);
return () => {
const current = this.agentStoppedRunningListeners.get(serverId);
if (!current) return;
current.delete(listener);
if (current.size === 0) this.agentStoppedRunningListeners.delete(serverId);
};
}
subscribeAll(listener: () => void): () => void {
this.globalListeners.add(listener);
return () => {
@@ -2296,13 +2285,6 @@ export class HostRuntimeStore {
listener();
}
}
private onAgentStoppedRunning(serverId: string, agentId: string): void {
this.drainQueuedAgentMessage(serverId, agentId);
for (const listener of this.agentStoppedRunningListeners.get(serverId) ?? []) {
listener(agentId);
}
}
}
let singletonHostRuntimeStore: HostRuntimeStore | null = null;
@@ -74,6 +74,7 @@ import { ImportSessionSheet } from "@/components/import-session-sheet";
import { useToast } from "@/contexts/toast-context";
import { selectIsFileExplorerOpen, usePanelStore } from "@/stores/panel-store";
import { type ExplorerCheckoutContext } from "@/stores/explorer-checkout-context";
import { traceInstant } from "@/performance/native-trace";
import { useSessionStore, type WorkspaceDescriptor } from "@/stores/session-store";
import {
collectAllTabs,
@@ -938,6 +939,12 @@ export const WorkspaceScreen = memo(function WorkspaceScreen({
recoveryAgentId,
}: WorkspaceScreenProps) {
const navigationFocused = useIsFocused();
useEffect(() => {
traceInstant("paseo.workspace.mount", { serverId, workspaceId });
return () => {
traceInstant("paseo.workspace.unmount", { serverId, workspaceId });
};
}, [serverId, workspaceId]);
return (
<WorkspaceScreenContent
serverId={serverId}
@@ -190,39 +190,6 @@ test("a gap absorbed by a running tail is recovered after the tail completes", a
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("an idle reconciliation waits for an in-flight tail and fetches a fresh snapshot", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const initialTail = await world.nextFetch("agent-a");
world.sync.reconcileAgent("agent-a");
world.expectNoPendingFetch();
initialTail.respond({ hasNewer: false });
const reconciliation = await world.nextFetch("agent-a");
expect(reconciliation.request).toEqual({
direction: "tail",
limit: 40,
projection: "projected",
});
reconciliation.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("does not reconcile an agent outside the viewed membership", () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.reconcileAgent("agent-a");
world.expectNoPendingFetch();
world.expectNoPendingMembership();
});
test("unchanged visible-set publication does not cancel paged catch-up", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -33,7 +33,6 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge {
setActive(active: boolean): void;
setConnected(connected: boolean): void;
setDeliveryMode(mode: TimelineDeliveryMode): void;
reconcileAgent(agentId: string): void;
recoverGap(agentId: string, cursor: { epoch: string; endSeq: number }): void;
dispose(): void;
}
@@ -451,15 +450,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
if (deliveryMode === "selective" && connected) void reconcileMembership();
else if (connected) startAcknowledgedCatchUps();
},
reconcileAgent(agentId) {
if (!isDesired(agentId)) return;
const request = planTimelineTailFetch();
if (catchUps.get(agentId)?.status === "running") {
pendingCatchUps.set(agentId, request);
return;
}
startCatchUp(agentId, { request, supersede: true });
},
recoverGap(agentId, cursor) {
if (!isDesired(agentId)) return;
startCatchUp(agentId, {
@@ -134,6 +134,27 @@ describe("test-daemon-connection connectToDaemon", () => {
expect(probe.createdConfigs()[0]?.password).toBe("shared-secret");
});
it("passes performance tracing into the connected client", async () => {
const { connectToDaemon } = await import("./test-daemon-connection");
const trace = {
isEnabled: () => true,
beginSection: vi.fn(),
endSection: vi.fn(),
};
const result = await connectToDaemon(
{
id: "direct:lan:6767",
type: "directTcp",
endpoint: "lan:6767",
},
{ trace },
probe.deps,
);
await result.client.close();
expect(probe.createdConfigs()[0]?.trace).toBe(trace);
});
it("uses relay TLS from the stored connection", async () => {
const { connectToDaemon } = await import("./test-daemon-connection");
const tlsResult = await connectToDaemon(
@@ -97,7 +97,10 @@ export class DaemonConnectionTestError extends Error {
export async function buildClientConfig(
connection: HostConnection,
serverId?: string,
options?: { capabilities?: DaemonClientConfig["capabilities"] },
options?: {
capabilities?: DaemonClientConfig["capabilities"];
trace?: DaemonClientConfig["trace"];
},
deps: Pick<
DaemonConnectionDependencies<DaemonProbeClient>,
"getClientId" | "resolveAppVersion" | "createLocalTransportFactory" | "buildLocalTransportUrl"
@@ -112,6 +115,7 @@ export async function buildClientConfig(
suppressSendErrors: true,
reconnect: { enabled: false },
...(options?.capabilities ? { capabilities: options.capabilities } : {}),
...(options?.trace ? { trace: options.trace } : {}),
...((connection.type === "directSocket" || connection.type === "directPipe") &&
localTransportFactory
? { transportFactory: localTransportFactory }
@@ -224,6 +228,7 @@ interface ProbeOptions {
serverId?: string;
timeoutMs?: number;
capabilities?: DaemonClientConfig["capabilities"];
trace?: DaemonClientConfig["trace"];
}
function resolveTimeout(connection: HostConnection, options?: ProbeOptions): number {
@@ -1,5 +1,6 @@
import type { AgentModelDefinition, ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
import { normalizeAgentModelDefinition } from "@getpaseo/protocol/agent-types";
import { expandProviderSnapshot } from "@getpaseo/protocol/provider-snapshot-codec";
import type {
GetProvidersSnapshotResponseMessage,
ListProviderModelsResponseMessage,
@@ -59,7 +60,9 @@ export function normalizeListProviderModelsPayload(
export function normalizeProvidersSnapshotPayload<
T extends GetProvidersSnapshotPayload | ProvidersSnapshotUpdatePayload,
>(payload: T): T {
const entries = normalizeProviderSnapshotEntries(payload.entries);
const entries = payload.compactSnapshot
? expandProviderSnapshot(payload.compactSnapshot)
: normalizeProviderSnapshotEntries(payload.entries);
return entries === payload.entries ? payload : { ...payload, entries };
}
+150 -1
View File
@@ -1,6 +1,11 @@
import { afterEach, expect, expectTypeOf, test, vi } from "vitest";
import { z } from "zod";
import { DaemonClient, type DaemonTransport, type Logger } from "./daemon-client";
import {
DaemonClient,
type DaemonClientTrace,
type DaemonTransport,
type Logger,
} from "./daemon-client";
import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities";
import { BROWSER_AUTOMATION_COMMAND_NAMES } from "@getpaseo/protocol/browser-automation/rpc-schemas";
import {
@@ -34,6 +39,24 @@ function createMockLogger() {
};
}
interface TraceRecord {
phase: "begin" | "end";
name?: string;
args?: Record<string, string>;
}
function createTraceRecorder(): { trace: DaemonClientTrace; records: TraceRecord[] } {
const records: TraceRecord[] = [];
return {
trace: {
isEnabled: () => true,
beginSection: (name, args) => records.push({ phase: "begin", name, args }),
endSection: () => records.push({ phase: "end" }),
},
records,
};
}
function createMockTransport() {
const sent: Array<string | Uint8Array | ArrayBuffer> = [];
@@ -161,6 +184,56 @@ afterEach(async () => {
vi.unstubAllGlobals();
});
test("traces WebSocket frames, message types, and JSON parse duration", async () => {
const mock = createMockTransport();
const recorder = createTraceRecorder();
const client = new DaemonClient({
url: "ws://test",
clientId: "trace_unit_test",
transportFactory: () => mock.transport,
reconnect: { enabled: false },
trace: recorder.trace,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen({ preserveSent: true });
await connectPromise;
expect(recorder.records).toEqual([
{
phase: "begin",
name: "paseo.ws.message.outbound",
args: { envelopeType: "hello", messageType: "hello" },
},
{ phase: "end" },
{
phase: "begin",
name: "paseo.ws.frame.outbound",
args: { kind: "text", size: expect.any(String) },
},
{ phase: "end" },
{
phase: "begin",
name: "paseo.ws.frame.inbound",
args: { kind: "text", size: expect.any(String) },
},
{
phase: "begin",
name: "paseo.ws.json.parse",
args: { size: expect.any(String) },
},
{ phase: "end" },
{
phase: "begin",
name: "paseo.ws.message.inbound",
args: { envelopeType: "session", messageType: "status" },
},
{ phase: "end" },
{ phase: "end" },
]);
});
test("does not infer browser automation capabilities from Electron runtime", async () => {
vi.stubGlobal("navigator", {
userAgent:
@@ -664,6 +737,7 @@ test("advertises client capabilities in hello", async () => {
clientType: "cli",
protocolVersion: 1,
capabilities: {
compact_provider_snapshots: true,
custom_mode_icons: true,
project_updates: true,
provider_subagents: true,
@@ -4384,6 +4458,81 @@ test("lists available providers via RPC", async () => {
});
});
test("requests provider snapshots conditionally and expands the compact response", async () => {
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "clsk_provider_snapshot_test",
reconnect: { enabled: false },
transportFactory: () => mock.transport,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
const promise = client.getProvidersSnapshot({ cwd: "/repo", ifNoneMatch: "previous-hash" });
const request = parseSentFrame(mock.sent[0]);
expect(request).toMatchObject({
type: "get_providers_snapshot_request",
cwd: "/repo",
ifNoneMatch: "previous-hash",
});
mock.triggerMessage(
wrapSessionMessage({
type: "get_providers_snapshot_response",
payload: {
entries: [],
compactSnapshot: {
entries: [
{
provider: "pi",
status: "ready",
enabled: true,
models: [
{ id: "model-a", label: "Model A", thinkingSet: 0 },
{ id: "model-b", label: "Model B", thinkingSet: 0 },
],
},
],
thinkingSets: [
{
options: [{ id: "high", label: "High", isDefault: true }],
defaultOptionId: "high",
},
],
},
snapshotHash: "next-hash",
generatedAt: "2026-08-04T00:00:00.000Z",
requestId: request.requestId,
},
}),
);
const response = await promise;
expect(response.entries[0]?.models).toEqual([
{
provider: "pi",
id: "model-a",
label: "Model A",
thinkingOptions: [{ id: "high", label: "High", isDefault: true }],
defaultThinkingOptionId: "high",
},
{
provider: "pi",
id: "model-b",
label: "Model B",
thinkingOptions: [{ id: "high", label: "High", isDefault: true }],
defaultThinkingOptionId: "high",
},
]);
expect(response.entries[0]?.models?.[0]?.thinkingOptions).toBe(
response.entries[0]?.models?.[1]?.thinkingOptions,
);
});
test("lists commands with draft config via RPC", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
+136 -30
View File
@@ -317,9 +317,16 @@ export interface DaemonClientConfig {
};
runtimeMetricsIntervalMs?: number;
runtimeMetricsWindowMs?: number;
trace?: DaemonClientTrace;
capabilities?: Partial<Record<ClientCapability, unknown>>;
}
export interface DaemonClientTrace {
isEnabled(): boolean;
beginSection(name: string, args?: Record<string, string>): void;
endSection(): void;
}
export interface SendMessageOptions {
messageId?: string;
images?: Array<{ data: string; mimeType: string }>;
@@ -1035,6 +1042,26 @@ function concatByteChunks(chunks: Uint8Array[], size: number): Uint8Array {
return bytes;
}
function getTransportFrameSize(frame: string | Uint8Array | ArrayBuffer): number {
if (typeof frame === "string") {
return frame.length;
}
return frame.byteLength;
}
function describeInboundTransportFrame(
frame: unknown,
rawBytes: Uint8Array | null,
): Record<string, string> {
if (typeof frame === "string") {
return { kind: "text", size: String(frame.length) };
}
if (rawBytes) {
return { kind: "binary", size: String(rawBytes.byteLength) };
}
return { kind: "unknown", size: "0" };
}
function hashForLog(value: string): string {
let hash = 0;
for (let index = 0; index < value.length; index += 1) {
@@ -1535,6 +1562,49 @@ export class DaemonClient {
// Core Send Helpers
// ============================================================================
private beginTraceSection(name: string, args?: Record<string, string>): boolean {
const trace = this.config.trace;
if (!trace?.isEnabled()) {
return false;
}
trace.beginSection(name, args);
return true;
}
private endTraceSection(isOpen: boolean): void {
if (isOpen) {
this.config.trace?.endSection();
}
}
private traceInstant(name: string, args?: Record<string, string>): void {
const isOpen = this.beginTraceSection(name, args);
this.endTraceSection(isOpen);
}
private sendJsonMessage(envelopeType: string, messageType: string, message: unknown): void {
this.traceInstant("paseo.ws.message.outbound", {
envelopeType,
messageType,
});
this.sendTransportFrame(JSON.stringify(message));
}
private sendTransportFrame(frame: string | Uint8Array | ArrayBuffer): void {
if (!this.transport) {
throw new Error("Transport not connected");
}
const isOpen = this.beginTraceSection("paseo.ws.frame.outbound", {
kind: typeof frame === "string" ? "text" : "binary",
size: String(getTransportFrameSize(frame)),
});
try {
this.transport.send(frame);
} finally {
this.endTraceSection(isOpen);
}
}
/**
* Send a session message. For fire-and-forget messages (heartbeats, etc.),
* failures are suppressed if `suppressSendErrors` is configured.
@@ -1549,7 +1619,7 @@ export class DaemonClient {
}
const payload = SessionInboundMessageSchema.parse(message);
try {
this.transport.send(JSON.stringify({ type: "session", message: payload }));
this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
} catch (error) {
if (this.config.suppressSendErrors) {
return;
@@ -1566,7 +1636,11 @@ export class DaemonClient {
throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
}
try {
this.transport.send(frame);
this.traceInstant("paseo.ws.message.outbound", {
envelopeType: "binary",
messageType: "binary",
});
this.sendTransportFrame(frame);
} catch (error) {
if (this.config.suppressSendErrors) {
return;
@@ -1587,7 +1661,7 @@ export class DaemonClient {
// If connected, send immediately
if (this.transport && status === "connected") {
const payload = SessionInboundMessageSchema.parse(message);
this.transport.send(JSON.stringify({ type: "session", message: payload }));
this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
return Promise.resolve();
}
@@ -1623,7 +1697,7 @@ export class DaemonClient {
try {
if (this.transport && this.connectionState.status === "connected") {
const payload = SessionInboundMessageSchema.parse(pending.message);
this.transport.send(JSON.stringify({ type: "session", message: payload }));
this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
pending.resolve();
} else {
pending.reject(new Error("Connection lost before message could be sent"));
@@ -1777,7 +1851,7 @@ export class DaemonClient {
}
const payload = SessionInboundMessageSchema.parse(message);
try {
this.transport.send(JSON.stringify({ type: "session", message: payload }));
this.sendJsonMessage("session", payload.type, { type: "session", message: payload });
} catch (error) {
throw error instanceof Error ? error : new Error(String(error));
}
@@ -1952,7 +2026,7 @@ export class DaemonClient {
this.pingProbe = probe;
try {
this.transport.send(JSON.stringify({ type: "ping" }));
this.sendJsonMessage("ping", "ping", { type: "ping" });
} catch (error) {
this.clearPingProbe();
const sendError = error instanceof Error ? error : new Error(String(error));
@@ -4421,6 +4495,7 @@ export class DaemonClient {
async getProvidersSnapshot(options?: {
cwd?: string;
ifNoneMatch?: string;
requestId?: string;
}): Promise<GetProvidersSnapshotPayload> {
const payload = await this.sendCorrelatedSessionRequest({
@@ -4428,6 +4503,7 @@ export class DaemonClient {
message: {
type: "get_providers_snapshot_request",
cwd: options?.cwd,
ifNoneMatch: options?.ifNoneMatch,
},
responseType: "get_providers_snapshot_response",
});
@@ -5326,23 +5402,22 @@ export class DaemonClient {
}
try {
this.transport.send(
JSON.stringify({
type: "hello",
clientId: this.config.clientId,
clientType: this.config.clientType ?? "cli",
protocolVersion: 1,
capabilities: {
[CLIENT_CAPS.customModeIcons]: true,
[CLIENT_CAPS.reasoningMergeEnum]: true,
[CLIENT_CAPS.terminalReflowableSnapshot]: true,
[CLIENT_CAPS.providerSubagents]: true,
[CLIENT_CAPS.projectUpdates]: true,
...this.config.capabilities,
},
...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}),
}),
);
this.sendJsonMessage("hello", "hello", {
type: "hello",
clientId: this.config.clientId,
clientType: this.config.clientType ?? "cli",
protocolVersion: 1,
capabilities: {
[CLIENT_CAPS.customModeIcons]: true,
[CLIENT_CAPS.reasoningMergeEnum]: true,
[CLIENT_CAPS.terminalReflowableSnapshot]: true,
[CLIENT_CAPS.providerSubagents]: true,
[CLIENT_CAPS.projectUpdates]: true,
[CLIENT_CAPS.compactProviderSnapshots]: true,
...this.config.capabilities,
},
...(this.config.appVersion ? { appVersion: this.config.appVersion } : {}),
});
} catch (error) {
const message = error instanceof Error ? error.message : "Failed to send hello message";
this.lastErrorValue = message;
@@ -5413,24 +5488,37 @@ export class DaemonClient {
}
const rawBytes = asUint8Array(rawData);
if (rawBytes && this.tryHandleBinaryFrame(rawBytes)) {
return;
const isOpen = this.beginTraceSection(
"paseo.ws.frame.inbound",
describeInboundTransportFrame(rawData, rawBytes),
);
try {
if (rawBytes && this.tryHandleBinaryFrame(rawBytes)) {
return;
}
const payload = decodeMessageData(rawData);
if (!payload) {
return;
}
this.handleJsonPayload(payload, rawBytes?.byteLength);
} finally {
this.endTraceSection(isOpen);
}
const payload = decodeMessageData(rawData);
if (!payload) {
return;
}
this.handleJsonPayload(payload, rawBytes?.byteLength);
}
private handleJsonPayload(payload: string, rawBytesLength: number | undefined): void {
const bytes = rawBytesLength ?? payload.length;
const startMs = perfNow();
let parsedJson: unknown;
const parseTraceOpen = this.beginTraceSection("paseo.ws.json.parse", {
size: String(bytes),
});
try {
parsedJson = JSON.parse(payload);
} catch {
return;
} finally {
this.endTraceSection(parseTraceOpen);
}
const parsed = validateWSOutboundMessage(parsedJson);
@@ -5457,11 +5545,19 @@ export class DaemonClient {
this.consecutiveLivenessFailures = 0;
if (parsed.data.type === "pong") {
this.traceInstant("paseo.ws.message.inbound", {
envelopeType: "pong",
messageType: "pong",
});
this.resolvePingProbe();
this.runtimeMetrics?.recordMessage("pong", bytes, perfNow() - startMs);
return;
}
this.traceInstant("paseo.ws.message.inbound", {
envelopeType: "session",
messageType: parsed.data.message.type,
});
this.handleSessionMessage(parsed.data.message);
const msgType = parsed.data.message.type;
this.runtimeMetrics?.recordMessage(msgType, bytes, perfNow() - startMs);
@@ -5473,6 +5569,11 @@ export class DaemonClient {
private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean {
const fileFrame = decodeFileTransferFrame(rawBytes);
if (fileFrame) {
this.traceInstant("paseo.ws.message.inbound", {
envelopeType: "binary",
messageType: "file",
opcode: String(fileFrame.opcode),
});
this.consecutiveLivenessFailures = 0;
this.handleFileTransferFrame(fileFrame);
this.runtimeMetrics?.recordBinaryFrame("other", rawBytes.byteLength, 0);
@@ -5483,6 +5584,11 @@ export class DaemonClient {
if (!frame) {
return false;
}
this.traceInstant("paseo.ws.message.inbound", {
envelopeType: "binary",
messageType: "terminal",
opcode: String(frame.opcode),
});
this.consecutiveLivenessFailures = 0;
const binaryStartMs = perfNow();
this.terminalStreams.handleFrame(frame);
@@ -20,6 +20,10 @@ export const CLIENT_CAPS = {
providerSubagents: "provider_subagents",
// COMPAT(projectUpdates): added in v0.1.109, remove gate after 2027-01-15.
projectUpdates: "project_updates",
// COMPAT(compactProviderSnapshots): added in v0.2.X. Capable clients receive
// provider catalogs with shared thinking sets and may revalidate by content hash.
// Remove the legacy snapshot encoding after 2027-02-04.
compactProviderSnapshots: "compact_provider_snapshots",
browserHost: "browser_host",
} as const;
+33 -2
View File
@@ -256,7 +256,7 @@ export const AgentFeatureSchema = z.discriminatedUnion("type", [
AgentFeatureSelectSchema,
]);
const AgentModelDefinitionSchema: z.ZodType<AgentModelDefinition> = z.object({
const AgentModelDefinitionSchema = z.object({
provider: AgentProviderSchema,
id: z.string(),
label: z.string(),
@@ -266,7 +266,7 @@ const AgentModelDefinitionSchema: z.ZodType<AgentModelDefinition> = z.object({
contextWindowMaxTokens: z.number().optional(),
thinkingOptions: z.array(AgentSelectOptionSchema).optional(),
defaultThinkingOptionId: z.string().optional(),
});
}) satisfies z.ZodType<AgentModelDefinition>;
export const ProviderSnapshotEntrySchema = z.object({
provider: AgentProviderSchema,
@@ -282,6 +282,29 @@ export const ProviderSnapshotEntrySchema = z.object({
defaultModeId: z.string().nullable().optional(),
});
export const CompactProviderSnapshotModelSchema = AgentModelDefinitionSchema.omit({
provider: true,
thinkingOptions: true,
}).extend({
thinkingSet: z.number().int().nonnegative().optional(),
});
export const ProviderSnapshotThinkingSetSchema = z.object({
options: z.array(AgentSelectOptionSchema),
defaultOptionId: z.string().optional(),
});
export const CompactProviderSnapshotEntrySchema = ProviderSnapshotEntrySchema.omit({
models: true,
}).extend({
models: z.array(CompactProviderSnapshotModelSchema).optional(),
});
export const CompactProviderSnapshotSchema = z.object({
entries: z.array(CompactProviderSnapshotEntrySchema),
thinkingSets: z.array(ProviderSnapshotThinkingSetSchema),
});
const AgentCapabilityFlagsSchema: z.ZodType<AgentCapabilityFlags> = z
.object({
supportsStreaming: z.boolean(),
@@ -1326,6 +1349,8 @@ export const ListAvailableProvidersRequestMessageSchema = z.object({
export const GetProvidersSnapshotRequestMessageSchema = z.object({
type: z.literal("get_providers_snapshot_request"),
cwd: z.string().optional(),
// COMPAT(compactProviderSnapshots): old daemons ignore this field and return a full snapshot.
ifNoneMatch: z.string().optional(),
requestId: z.string(),
});
@@ -4956,6 +4981,9 @@ export const GetProvidersSnapshotResponseMessageSchema = z.object({
type: z.literal("get_providers_snapshot_response"),
payload: z.object({
entries: z.array(ProviderSnapshotEntrySchema),
compactSnapshot: CompactProviderSnapshotSchema.optional(),
snapshotHash: z.string().optional(),
notModified: z.boolean().optional(),
generatedAt: z.string(),
requestId: z.string(),
}),
@@ -4967,6 +4995,8 @@ export const ProvidersSnapshotUpdateMessageSchema = z.object({
payload: z.object({
cwd: z.string().optional(),
entries: z.array(ProviderSnapshotEntrySchema),
compactSnapshot: CompactProviderSnapshotSchema.optional(),
snapshotHash: z.string().optional(),
generatedAt: z.string(),
}),
});
@@ -5938,6 +5968,7 @@ export const WSHelloMessageSchema = z.object({
[CLIENT_CAPS.terminalReflowableSnapshot]: z.boolean().optional(),
[CLIENT_CAPS.providerSubagents]: z.boolean().optional(),
[CLIENT_CAPS.projectUpdates]: z.boolean().optional(),
[CLIENT_CAPS.compactProviderSnapshots]: z.boolean().optional(),
[CLIENT_CAPS.browserHost]: BrowserAutomationHostCapabilitySchema.optional(),
})
.passthrough()
@@ -0,0 +1,79 @@
import { describe, expect, it } from "vitest";
import type { ProviderSnapshotEntry } from "./agent-types.js";
import { compactProviderSnapshot, expandProviderSnapshot } from "./provider-snapshot-codec.js";
const sharedThinking = [
{ id: "off", label: "Off", description: "No extra reasoning" },
{ id: "minimal", label: "Minimal", description: "Light reasoning" },
{ id: "low", label: "Low" },
{ id: "medium", label: "Medium", description: "Balanced reasoning" },
{ id: "high", label: "High", isDefault: true },
{ id: "xhigh", label: "XHigh", description: "Very deep reasoning" },
{ id: "max", label: "Max", description: "Extreme reasoning" },
];
function providerEntry(): ProviderSnapshotEntry {
return {
provider: "pi",
status: "ready",
enabled: true,
models: [
{
provider: "pi",
id: "openai/gpt-a",
label: "GPT A",
description: "openai/gpt-a",
metadata: { provider: "openai", modelId: "gpt-a" },
thinkingOptions: sharedThinking,
defaultThinkingOptionId: "high",
},
{
provider: "pi",
id: "openai/gpt-b",
label: "GPT B",
metadata: { provider: "openai", modelId: "gpt-b" },
thinkingOptions: structuredClone(sharedThinking),
defaultThinkingOptionId: "high",
},
],
modes: [{ id: "build", label: "Build" }],
};
}
describe("provider snapshot codec", () => {
it("shares repeated thinking options while preserving the public snapshot shape", () => {
const original = [providerEntry()];
const compact = compactProviderSnapshot(original);
const expanded = expandProviderSnapshot(compact);
expect(compact.thinkingSets).toHaveLength(1);
expect(compact.entries[0]?.models?.map((model) => model.thinkingSet)).toEqual([0, 0]);
expect(expanded).toEqual(original);
expect(expanded[0]?.models?.[0]?.thinkingOptions).toBe(
expanded[0]?.models?.[1]?.thinkingOptions,
);
});
it("shrinks catalogs dominated by one repeated thinking set", () => {
const model = providerEntry().models?.[0];
if (!model) throw new Error("fixture model missing");
const entries: ProviderSnapshotEntry[] = [
{
provider: "pi",
status: "ready",
enabled: true,
models: Array.from({ length: 200 }, (_, index) => ({
...model,
id: `openai/model-${index}`,
label: `Model ${index}`,
metadata: { provider: "openai", modelId: `model-${index}` },
})),
},
];
const legacyBytes = JSON.stringify(entries).length;
const compactBytes = JSON.stringify(compactProviderSnapshot(entries)).length;
expect(compactBytes).toBeLessThan(legacyBytes * 0.5);
});
});
@@ -0,0 +1,119 @@
import type {
AgentModelDefinition,
AgentSelectOption,
ProviderSnapshotEntry,
} from "./agent-types.js";
export interface CompactProviderSnapshotModel extends Omit<
AgentModelDefinition,
"provider" | "thinkingOptions"
> {
thinkingSet?: number;
}
export interface CompactProviderSnapshotEntry extends Omit<ProviderSnapshotEntry, "models"> {
models?: CompactProviderSnapshotModel[];
}
export interface ProviderSnapshotThinkingSet {
options: AgentSelectOption[];
defaultOptionId?: string;
}
export interface CompactProviderSnapshot {
entries: CompactProviderSnapshotEntry[];
thinkingSets: ProviderSnapshotThinkingSet[];
}
function thinkingSetKey(set: ProviderSnapshotThinkingSet): string {
return JSON.stringify(set);
}
function compactModel(
{
provider: _provider,
thinkingOptions,
defaultThinkingOptionId,
...modelFields
}: AgentModelDefinition,
thinkingSets: ProviderSnapshotThinkingSet[],
thinkingSetIndexes: Map<string, number>,
): CompactProviderSnapshotModel {
const compact: CompactProviderSnapshotModel = modelFields;
if (thinkingOptions === undefined) {
if (defaultThinkingOptionId !== undefined) {
compact.defaultThinkingOptionId = defaultThinkingOptionId;
}
return compact;
}
const thinkingSet: ProviderSnapshotThinkingSet = {
options: thinkingOptions,
...(defaultThinkingOptionId !== undefined ? { defaultOptionId: defaultThinkingOptionId } : {}),
};
const key = thinkingSetKey(thinkingSet);
const existingIndex = thinkingSetIndexes.get(key);
if (existingIndex !== undefined) {
compact.thinkingSet = existingIndex;
return compact;
}
const index = thinkingSets.length;
thinkingSets.push(thinkingSet);
thinkingSetIndexes.set(key, index);
compact.thinkingSet = index;
return compact;
}
export function compactProviderSnapshot(entries: ProviderSnapshotEntry[]): CompactProviderSnapshot {
const thinkingSets: ProviderSnapshotThinkingSet[] = [];
const thinkingSetIndexes = new Map<string, number>();
const compactEntries = entries.map((entry): CompactProviderSnapshotEntry => {
if (entry.models === undefined) {
return entry;
}
return {
...entry,
models: entry.models.map((model) => compactModel(model, thinkingSets, thinkingSetIndexes)),
};
});
return { entries: compactEntries, thinkingSets };
}
function expandModel(
provider: string,
model: CompactProviderSnapshotModel,
thinkingSets: ProviderSnapshotThinkingSet[],
): AgentModelDefinition {
const { thinkingSet: thinkingSetIndex, ...modelFields } = model;
if (thinkingSetIndex === undefined) {
return { provider, ...modelFields };
}
const thinkingSet = thinkingSets[thinkingSetIndex];
if (!thinkingSet) {
throw new RangeError(`Provider snapshot references missing thinking set ${thinkingSetIndex}`);
}
return {
provider,
...modelFields,
thinkingOptions: thinkingSet.options,
...(thinkingSet.defaultOptionId !== undefined
? { defaultThinkingOptionId: thinkingSet.defaultOptionId }
: {}),
};
}
export function expandProviderSnapshot(snapshot: CompactProviderSnapshot): ProviderSnapshotEntry[] {
return snapshot.entries.map((entry): ProviderSnapshotEntry => {
const { models, ...entryFields } = entry;
if (models === undefined) return entryFields;
return {
...entryFields,
models: models.map((model) =>
expandModel(entryFields.provider, model, snapshot.thinkingSets),
),
};
});
}
@@ -138,6 +138,42 @@ const SourceSchema = z.object({
);
});
it("accepts a compact provider snapshot envelope", () => {
const envelope = {
type: "session",
message: {
type: "get_providers_snapshot_response",
payload: {
entries: [],
compactSnapshot: {
entries: [
{
provider: "pi",
status: "ready",
enabled: true,
models: [{ id: "model-a", label: "Model A", thinkingSet: 0 }],
},
],
thinkingSets: [
{
options: [{ id: "high", label: "High", isDefault: true }],
defaultOptionId: "high",
},
],
},
snapshotHash: "snapshot-hash",
generatedAt: "2026-08-04T00:00:00.000Z",
requestId: "provider-snapshot",
},
},
};
expect(GeneratedWSOutboundMessageSchema.safeParse(envelope)).toEqual({
success: true,
data: envelope,
});
});
it.each([
{
name: "dedicated attention message",
+1
View File
@@ -841,6 +841,7 @@ export class Session {
emit: (msg) => this.emit(msg),
isProviderVisibleToClient: (provider) => this.isProviderVisibleToClient(provider),
supportsCustomModeIcons: () => this.supports(CLIENT_CAPS.customModeIcons),
supportsCompactProviderSnapshots: () => this.supports(CLIENT_CAPS.compactProviderSnapshots),
listProviderAvailability: () => this.agentManager.listProviderAvailability(),
listDraftFeatures: (config) => this.agentManager.listDraftFeatures(config),
},
@@ -13,12 +13,14 @@ import {
} from "../../agent/provider-snapshot-manager.js";
import type { ProviderSnapshotEntry } from "../../agent/agent-sdk-types.js";
import type { ProviderUsageService } from "../../../services/quota-fetcher/service.js";
import { expandProviderSnapshot } from "@getpaseo/protocol/provider-snapshot-codec";
type SnapshotChangeHandler = (entries: ProviderSnapshotEntry[], cwd: string) => void;
interface MakeOptions {
visibleProviders?: Set<string>;
supportsCustomModeIcons?: boolean;
supportsCompactProviderSnapshots?: boolean;
snapshot?: { [K in keyof ProviderSnapshotManager]?: unknown };
usage?: { [K in keyof ProviderUsageService]?: unknown };
host?: Partial<ProviderCatalogSessionHost>;
@@ -49,6 +51,7 @@ function makeSubsystem(options: MakeOptions = {}) {
emit: (msg) => emitted.push(msg),
isProviderVisibleToClient: (provider) => visible.has(provider),
supportsCustomModeIcons: () => options.supportsCustomModeIcons ?? false,
supportsCompactProviderSnapshots: () => options.supportsCompactProviderSnapshots ?? false,
listProviderAvailability: async () => [],
listDraftFeatures: async () => [],
...options.host,
@@ -113,6 +116,21 @@ describe("ProviderCatalogSession", () => {
expect(pull?.payload.entries).toEqual(push?.payload.entries);
});
it("pushes the compact encoding to capable clients", () => {
const { subsystem, emitted, pushSnapshotChange } = makeSubsystem({
supportsCustomModeIcons: true,
supportsCompactProviderSnapshots: true,
});
subsystem.start();
pushSnapshotChange(makeEntries());
const push = findByType(emitted, "providers_snapshot_update");
expect(push?.payload.entries).toEqual([]);
expect(push?.payload.snapshotHash).toEqual(expect.any(String));
expect(expandProviderSnapshot(push!.payload.compactSnapshot!)).toEqual([makeEntries()[0]]);
});
it("preserves custom mode icons when the client supports them", async () => {
const { subsystem, emitted } = makeSubsystem({
supportsCustomModeIcons: true,
@@ -128,6 +146,60 @@ describe("ProviderCatalogSession", () => {
expect(pull?.payload.entries[0]?.modes?.[0]?.icon).toBe("Sparkles");
});
it("sends capable clients a compact snapshot and returns not-modified for its hash", async () => {
const thinkingOptions = [
{ id: "low", label: "Low" },
{ id: "high", label: "High", isDefault: true },
];
const entries: ProviderSnapshotEntry[] = [
{
provider: "codex",
status: "ready",
enabled: true,
models: ["one", "two"].map((id) => ({
provider: "codex",
id,
label: id,
thinkingOptions,
defaultThinkingOptionId: "high",
})),
},
];
const { subsystem, emitted } = makeSubsystem({
supportsCompactProviderSnapshots: true,
snapshot: { getSnapshot: () => entries },
});
await subsystem.handleGetProvidersSnapshotRequest({
type: "get_providers_snapshot_request",
requestId: "compact-1",
});
const first = findByType(emitted, "get_providers_snapshot_response");
expect(first?.payload.entries).toEqual([]);
expect(first?.payload.snapshotHash).toEqual(expect.any(String));
expect(first?.payload.compactSnapshot?.thinkingSets).toHaveLength(1);
expect(expandProviderSnapshot(first!.payload.compactSnapshot!)).toEqual(entries);
await subsystem.handleGetProvidersSnapshotRequest({
type: "get_providers_snapshot_request",
requestId: "compact-2",
ifNoneMatch: first?.payload.snapshotHash,
});
const responses = emitted.filter(
(message) => message.type === "get_providers_snapshot_response",
);
const second = responses[1];
expect(second?.payload).toMatchObject({
entries: [],
snapshotHash: first?.payload.snapshotHash,
notModified: true,
requestId: "compact-2",
});
expect(second?.payload.compactSnapshot).toBeUndefined();
});
it("reports a disabled provider on list_provider_models without warming the snapshot", async () => {
// warmUpSnapshotForCwd is intentionally unstubbed: createStub throws if it is called,
// so the disabled short-circuit is proven by the absence of a throw.
@@ -1,5 +1,10 @@
import type pino from "pino";
import { createHash } from "node:crypto";
import { getErrorMessage } from "@getpaseo/protocol/error-utils";
import {
compactProviderSnapshot,
type CompactProviderSnapshot,
} from "@getpaseo/protocol/provider-snapshot-codec";
import type { SessionInboundMessage, SessionOutboundMessage } from "../../messages.js";
import {
isGlobalProviderSnapshotKey,
@@ -39,6 +44,7 @@ export interface ProviderCatalogSessionHost {
isProviderVisibleToClient(provider: string): boolean;
// COMPAT(customModeIcons): reads clientCapabilities live.
supportsCustomModeIcons(): boolean;
supportsCompactProviderSnapshots(): boolean;
listProviderAvailability(): Promise<ProviderAvailability[]>;
listDraftFeatures(config: AgentSessionConfig): Promise<AgentFeature[]>;
}
@@ -50,6 +56,19 @@ export interface ProviderCatalogSessionOptions {
logger: pino.Logger;
}
interface EncodedProviderSnapshot {
compactSnapshot: CompactProviderSnapshot;
snapshotHash: string;
}
function encodeProviderSnapshot(entries: ProviderSnapshotEntry[]): EncodedProviderSnapshot {
const compactSnapshot = compactProviderSnapshot(entries);
const snapshotHash = createHash("sha256")
.update(JSON.stringify(compactSnapshot))
.digest("base64url");
return { compactSnapshot, snapshotHash };
}
/**
* A client's provider catalog surface: model / mode / feature listing, the providers
* snapshot push + pull, provider diagnostics, and usage. The snapshot PUSH (start) and
@@ -78,11 +97,25 @@ export class ProviderCatalogSession {
this.host.isProviderVisibleToClient(entry.provider),
);
const snapshotCwd = isGlobalProviderSnapshotKey(cwd) ? undefined : cwd;
const clientEntries = this.downgradeEntryModesForClient(visibleEntries);
if (this.host.supportsCompactProviderSnapshots()) {
const encoded = encodeProviderSnapshot(clientEntries);
this.host.emit({
type: "providers_snapshot_update",
payload: {
...(snapshotCwd ? { cwd: snapshotCwd } : {}),
entries: [],
...encoded,
generatedAt: new Date().toISOString(),
},
});
return;
}
this.host.emit({
type: "providers_snapshot_update",
payload: {
...(snapshotCwd ? { cwd: snapshotCwd } : {}),
entries: this.downgradeEntryModesForClient(visibleEntries),
entries: clientEntries,
generatedAt: new Date().toISOString(),
},
});
@@ -360,11 +393,29 @@ export class ProviderCatalogSession {
const entries = this.providerSnapshotManager
.getSnapshot(msg.cwd ? expandTilde(msg.cwd) : undefined)
.filter((entry) => this.host.isProviderVisibleToClient(entry.provider));
const clientEntries = this.downgradeEntryModesForClient(entries);
if (this.host.supportsCompactProviderSnapshots()) {
const encoded = encodeProviderSnapshot(clientEntries);
const notModified = msg.ifNoneMatch === encoded.snapshotHash;
this.host.emit({
type: "get_providers_snapshot_response",
payload: {
entries: [],
...(!notModified ? { compactSnapshot: encoded.compactSnapshot } : {}),
snapshotHash: encoded.snapshotHash,
...(notModified ? { notModified: true } : {}),
generatedAt: new Date().toISOString(),
requestId: msg.requestId,
},
});
return;
}
this.host.emit({
type: "get_providers_snapshot_response",
payload: {
entries: this.downgradeEntryModesForClient(entries),
entries: clientEntries,
generatedAt: new Date().toISOString(),
requestId: msg.requestId,
},