Speed up inbound WebSocket validation (#1895)
* perf(protocol): generate inbound ws validators * perf(client): use generated ws validation * docs(protocol): document generated validation * fix(protocol): make validator generation source-only * test(protocol): cover explicit provider model normalization * fix(protocol): preserve inbound compat defaults * fix(protocol): make validator import rewrite portable * fix(protocol): harden validation safety checks * fix(protocol): guard generated validator boundaries * fix(protocol): normalize legacy inbound defaults * fix(protocol): simplify generated validation safety net * fix(protocol): keep cold typecheck source-only * fix(protocol): encapsulate validator codegen * fix(protocol): bootstrap source-alias typechecks * fix(app): keep source aliases out of bundler config * fix(protocol): harden validator codegen packaging * fix(protocol): own zod-aot patches in generator * fix(protocol): trim validator safety net * fix(client): normalize provider updates before dispatch * fix(protocol): keep validator generation out of install
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
import type { AgentModelDefinition, ProviderSnapshotEntry } from "@getpaseo/protocol/agent-types";
|
||||
import { normalizeAgentModelDefinition } from "@getpaseo/protocol/agent-types";
|
||||
import type {
|
||||
GetProvidersSnapshotResponseMessage,
|
||||
ListProviderModelsResponseMessage,
|
||||
SessionOutboundMessage,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
|
||||
type ListProviderModelsPayload = ListProviderModelsResponseMessage["payload"];
|
||||
type GetProvidersSnapshotPayload = GetProvidersSnapshotResponseMessage["payload"];
|
||||
type ProvidersSnapshotUpdatePayload = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "providers_snapshot_update" }
|
||||
>["payload"];
|
||||
|
||||
// COMPAT(model-normalize): daemon normalizes at source (provider-registry) — shim covers older daemons; drop when floor >= v0.1.104
|
||||
function normalizeAgentModels(
|
||||
models: AgentModelDefinition[] | undefined,
|
||||
): AgentModelDefinition[] | undefined {
|
||||
if (!models) {
|
||||
return models;
|
||||
}
|
||||
|
||||
let changed = false;
|
||||
const normalized = models.map((model) => {
|
||||
const next = normalizeAgentModelDefinition(model);
|
||||
changed ||= next !== model;
|
||||
return next;
|
||||
});
|
||||
|
||||
return changed ? normalized : models;
|
||||
}
|
||||
|
||||
function normalizeProviderSnapshotEntry(entry: ProviderSnapshotEntry): ProviderSnapshotEntry {
|
||||
const models = normalizeAgentModels(entry.models);
|
||||
return models === entry.models ? entry : { ...entry, models };
|
||||
}
|
||||
|
||||
function normalizeProviderSnapshotEntries(
|
||||
entries: ProviderSnapshotEntry[],
|
||||
): ProviderSnapshotEntry[] {
|
||||
let changed = false;
|
||||
const normalized = entries.map((entry) => {
|
||||
const next = normalizeProviderSnapshotEntry(entry);
|
||||
changed ||= next !== entry;
|
||||
return next;
|
||||
});
|
||||
|
||||
return changed ? normalized : entries;
|
||||
}
|
||||
|
||||
export function normalizeListProviderModelsPayload(
|
||||
payload: ListProviderModelsPayload,
|
||||
): ListProviderModelsPayload {
|
||||
const models = normalizeAgentModels(payload.models);
|
||||
return models === payload.models ? payload : { ...payload, models };
|
||||
}
|
||||
|
||||
export function normalizeProvidersSnapshotPayload<
|
||||
T extends GetProvidersSnapshotPayload | ProvidersSnapshotUpdatePayload,
|
||||
>(payload: T): T {
|
||||
const entries = normalizeProviderSnapshotEntries(payload.entries);
|
||||
return entries === payload.entries ? payload : { ...payload, entries };
|
||||
}
|
||||
|
||||
export function normalizeProviderSnapshotUpdateMessage(
|
||||
msg: SessionOutboundMessage,
|
||||
): SessionOutboundMessage {
|
||||
if (msg.type !== "providers_snapshot_update") {
|
||||
return msg;
|
||||
}
|
||||
|
||||
return { ...msg, payload: normalizeProvidersSnapshotPayload(msg.payload) };
|
||||
}
|
||||
@@ -13,8 +13,8 @@ import {
|
||||
DaemonUpdateResponseSchema,
|
||||
SessionInboundMessageSchema,
|
||||
type ServerInfoStatusPayload,
|
||||
WSOutboundMessageSchema,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
import { validateWSOutboundMessage } from "@getpaseo/protocol/validation/ws-outbound";
|
||||
import type {
|
||||
AgentStreamEventPayload,
|
||||
AgentSnapshotPayload,
|
||||
@@ -118,6 +118,11 @@ import {
|
||||
type WebSocketFactory,
|
||||
} from "./daemon-client-transport.js";
|
||||
import { DaemonClientRuntimeMetrics } from "./daemon-client-runtime-metrics.js";
|
||||
import {
|
||||
normalizeListProviderModelsPayload,
|
||||
normalizeProviderSnapshotUpdateMessage,
|
||||
normalizeProvidersSnapshotPayload,
|
||||
} from "./compat/normalize-provider-models.js";
|
||||
import { TerminalStreamRouter, type TerminalStreamEvent } from "./terminal-stream-router.js";
|
||||
import type {
|
||||
BrowserAutomationExecuteRequest,
|
||||
@@ -370,6 +375,7 @@ type WriteProjectConfigPayload = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "write_project_config_response" }
|
||||
>["payload"];
|
||||
|
||||
type ListCommandsPayload = ListCommandsResponse["payload"];
|
||||
type ListCommandsDraftConfig = Pick<
|
||||
AgentSessionConfig,
|
||||
@@ -3728,7 +3734,7 @@ export class DaemonClient {
|
||||
provider: AgentProvider,
|
||||
options?: { cwd?: string; requestId?: string },
|
||||
): Promise<ListProviderModelsPayload> {
|
||||
return this.sendCorrelatedSessionRequest({
|
||||
const payload = await this.sendCorrelatedSessionRequest({
|
||||
requestId: options?.requestId,
|
||||
message: {
|
||||
type: "list_provider_models_request",
|
||||
@@ -3739,6 +3745,7 @@ export class DaemonClient {
|
||||
// Provider SDK cold starts (especially model discovery) can exceed 60s.
|
||||
timeout: 90000,
|
||||
});
|
||||
return normalizeListProviderModelsPayload(payload);
|
||||
}
|
||||
|
||||
async listProviderModes(
|
||||
@@ -3788,7 +3795,7 @@ export class DaemonClient {
|
||||
cwd?: string;
|
||||
requestId?: string;
|
||||
}): Promise<GetProvidersSnapshotPayload> {
|
||||
return this.sendCorrelatedSessionRequest({
|
||||
const payload = await this.sendCorrelatedSessionRequest({
|
||||
requestId: options?.requestId,
|
||||
message: {
|
||||
type: "get_providers_snapshot_request",
|
||||
@@ -3796,6 +3803,7 @@ export class DaemonClient {
|
||||
},
|
||||
responseType: "get_providers_snapshot_response",
|
||||
});
|
||||
return normalizeProvidersSnapshotPayload(payload);
|
||||
}
|
||||
|
||||
async getDaemonConfig(
|
||||
@@ -4754,7 +4762,7 @@ export class DaemonClient {
|
||||
return;
|
||||
}
|
||||
|
||||
const parsed = WSOutboundMessageSchema.safeParse(parsedJson);
|
||||
const parsed = validateWSOutboundMessage(parsedJson);
|
||||
if (!parsed.success) {
|
||||
const msgType =
|
||||
parsedJson != null &&
|
||||
@@ -5017,8 +5025,10 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
private handleSessionMessage(msg: SessionOutboundMessage): void {
|
||||
if (msg.type === "status") {
|
||||
const serverInfo = parseServerInfoStatusPayload(msg.payload);
|
||||
const consumerMessage = normalizeProviderSnapshotUpdateMessage(msg);
|
||||
|
||||
if (consumerMessage.type === "status") {
|
||||
const serverInfo = parseServerInfoStatusPayload(consumerMessage.payload);
|
||||
if (serverInfo) {
|
||||
this.lastServerInfoMessage = serverInfo;
|
||||
if (this.connectionState.status === "connecting") {
|
||||
@@ -5034,39 +5044,39 @@ export class DaemonClient {
|
||||
}
|
||||
}
|
||||
|
||||
if (msg.type === "terminal_stream_exit") {
|
||||
this.terminalStreams.removeTerminal(msg.payload.terminalId);
|
||||
if (consumerMessage.type === "terminal_stream_exit") {
|
||||
this.terminalStreams.removeTerminal(consumerMessage.payload.terminalId);
|
||||
}
|
||||
|
||||
if (this.rawMessageListeners.size > 0) {
|
||||
for (const handler of this.rawMessageListeners) {
|
||||
try {
|
||||
handler(msg);
|
||||
handler(consumerMessage);
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const handlers = this.messageHandlers.get(msg.type);
|
||||
const handlers = this.messageHandlers.get(consumerMessage.type);
|
||||
if (handlers) {
|
||||
for (const handler of handlers) {
|
||||
try {
|
||||
handler(msg);
|
||||
handler(consumerMessage);
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const event = this.toEvent(msg);
|
||||
const event = this.toEvent(consumerMessage);
|
||||
if (event) {
|
||||
for (const handler of this.eventListeners) {
|
||||
handler(event);
|
||||
}
|
||||
}
|
||||
|
||||
this.resolveWaiters(msg);
|
||||
this.resolveWaiters(consumerMessage);
|
||||
}
|
||||
|
||||
private resolveWaiters(msg: SessionOutboundMessage): void {
|
||||
|
||||
@@ -629,20 +629,37 @@ test("provider actions delegate to existing provider RPCs and local snapshot upd
|
||||
});
|
||||
|
||||
const snapshotUpdates: string[] = [];
|
||||
const snapshotModelDefaults: Array<string | undefined> = [];
|
||||
const unsubscribe = client.providers.subscribe((update) => {
|
||||
snapshotUpdates.push(update.generatedAt);
|
||||
snapshotModelDefaults.push(update.entries[0]?.models?.[0]?.defaultThinkingOptionId);
|
||||
});
|
||||
ws.message(
|
||||
sessionMessage({
|
||||
type: "providers_snapshot_update",
|
||||
payload: {
|
||||
cwd: "/repo/sdk",
|
||||
entries: [{ provider: "codex", status: "ready", enabled: true }],
|
||||
entries: [
|
||||
{
|
||||
provider: "codex",
|
||||
status: "ready",
|
||||
enabled: true,
|
||||
models: [
|
||||
{
|
||||
provider: "codex",
|
||||
id: "gpt-5.5",
|
||||
label: "GPT-5.5",
|
||||
thinkingOptions: [{ id: "high", label: "High", isDefault: true }],
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
generatedAt: "2026-05-16T01:00:00.000Z",
|
||||
},
|
||||
}),
|
||||
);
|
||||
expect(snapshotUpdates).toEqual(["2026-05-16T01:00:00.000Z"]);
|
||||
expect(snapshotModelDefaults).toEqual(["high"]);
|
||||
|
||||
unsubscribe();
|
||||
await client.close();
|
||||
|
||||
Reference in New Issue
Block a user