QVAC-13715 feat: Profiler Operation Transport + Load/Download Metrics + Stream Profiling

packages/sdk/client/api/download-asset.ts

@@ -1,6 +1,7 @@
 import { send, stream } from "@/client/rpc/rpc-client";
 import {
   type DownloadAssetOptions as BaseDownloadAssetOptions,
+  type RPCOptions,
   downloadAssetOptionsToRequestSchema,
 } from "@/schemas";
 import {
@@ -22,6 +23,7 @@ export type DownloadAssetOptions = BaseDownloadAssetOptions;
  *   - assetSrc: The location from which the asset is downloaded (local path, remote URL, or Hyperdrive URL)
  *   - seed: Optional boolean for hyperdrive seeding
  *   - onProgress: Optional callback for download progress
+ * @param rpcOptions - Optional RPC options including per-call profiling configuration
  *
  * @returns Promise that resolves to the asset ID (either the provided assetSrc or a generated ID)
  *
@@ -48,12 +50,13 @@ export type DownloadAssetOptions = BaseDownloadAssetOptions;
  */
 export async function downloadAsset(
   options: DownloadAssetOptions,
+  rpcOptions?: RPCOptions,
 ): Promise<string> {
   const request = downloadAssetOptionsToRequestSchema.parse(options);
 
   if (options.onProgress) {
     // Use streaming for progress updates
-    for await (const response of stream(request)) {
+    for await (const response of stream(request, undefined, rpcOptions)) {
       if (response.type === "modelProgress") {
         options.onProgress(response);
       } else if (response.type === "downloadAsset") {
@@ -67,7 +70,7 @@ export async function downloadAsset(
     throw new StreamEndedError();
   } else {
     // Use regular send for simple downloading
-    const response = await send(request);
+    const response = await send(request, undefined, rpcOptions);
     if (response.type !== "downloadAsset") {
       throw new InvalidResponseError("downloadAsset");
     }

packages/sdk/client/api/load-model.ts

@@ -3,6 +3,7 @@ import { startLoggingStreamForModel } from "@/client/logging-stream-registry";
 import {
   type LoadModelOptions,
   type ReloadConfigOptions,
+  type RPCOptions,
   loadModelOptionsToRequestSchema,
   reloadConfigOptionsToRequestSchema,
   isModelTypeAlias,
@@ -32,6 +33,7 @@ const logger = getClientLogger();
  *   - modelConfig: Model-specific configuration options (companion sources, model parameters, etc.)
  *   - onProgress: Callback for download progress updates
  *   - logger: Logger instance for model operation logs
+ * @param rpcOptions - Optional RPC options including per-call profiling configuration
  *
  * @returns Promise that resolves to the model ID (either the provided modelSrc or a generated ID)
  *
@@ -107,7 +109,10 @@ const logger = getClientLogger();
  * });
  * ```
  */
-export function loadModel(options: LoadModelOptions): Promise<string>;
+export function loadModel(
+  options: LoadModelOptions,
+  rpcOptions?: RPCOptions,
+): Promise<string>;
 
 /**
  * Hot-reloads configuration on an already loaded model.
@@ -116,6 +121,7 @@ export function loadModel(options: LoadModelOptions): Promise<string>;
  *   - modelId: The ID of an existing loaded model
  *   - modelType: The type of model (must match the loaded model)
  *   - modelConfig: New configuration to apply
+ * @param rpcOptions - Optional RPC options including per-call profiling configuration
  *
  * @returns Promise that resolves to the model ID
  *
@@ -139,10 +145,14 @@ export function loadModel(options: LoadModelOptions): Promise<string>;
  * });
  * ```
  */
-export function loadModel(options: ReloadConfigOptions): Promise<string>;
+export function loadModel(
+  options: ReloadConfigOptions,
+  rpcOptions?: RPCOptions,
+): Promise<string>;
 
 export async function loadModel(
   options: LoadModelOptions | ReloadConfigOptions,
+  rpcOptions?: RPCOptions,
 ): Promise<string> {
   const isReloadConfig = "modelId" in options && !("modelSrc" in options);
 
@@ -162,7 +172,7 @@ export async function loadModel(
 
   if (onProgress) {
     // Use streaming for progress updates
-    for await (const response of stream(request)) {
+    for await (const response of stream(request, undefined, rpcOptions)) {
       if (response.type === "modelProgress") {
         onProgress(response);
       } else if (response.type === "loadModel") {
@@ -191,7 +201,7 @@ export async function loadModel(
   }
 
   // Use regular send for simple loading
-  const response = await send(request);
+  const response = await send(request, undefined, rpcOptions);
   if (response.type !== "loadModel") {
     throw new InvalidResponseError("loadModel");
   }

packages/sdk/client/rpc/profiling/profiler.ts

@@ -166,6 +166,24 @@ export function recordClientEvents(
   if (serverMeta?.delegation) {
     recordDelegationBreakdownPhases(base, serverMeta.delegation);
   }
+
+  if (serverMeta?.operation) {
+    recordOperationEvent(serverMeta.operation);
+  }
+}
+
+function recordOperationEvent(op: NonNullable<ProfilingResponseMeta["operation"]>): void {
+  const event: Parameters<typeof record>[0] = {
+    ts: nowMs(),
+    op: op.op,
+    kind: op.kind,
+    ms: op.ms,
+  };
+  if (op.profileId !== undefined) event.profileId = op.profileId;
+  if (op.gauges !== undefined) event.gauges = op.gauges;
+  if (op.tags !== undefined) event.tags = op.tags;
+  if (op.count !== undefined) event.count = op.count;
+  record(event);
 }
 
 export function recordClientStreamEvents(
@@ -205,6 +223,10 @@ export function recordClientStreamEvents(
   if (serverMeta?.delegation) {
     recordDelegationBreakdownPhases(base, serverMeta.delegation);
   }
+
+  if (serverMeta?.operation) {
+    recordOperationEvent(serverMeta.operation);
+  }
 }
 
 export function resetConnectionTracking(): void {

packages/sdk/examples/profiling/basic.ts

@@ -43,6 +43,45 @@ try {
   console.log(profiler.exportTable());
 
   const json = profiler.exportJSON();
+  console.log("\n=== Load Model Metrics ===");
+  // Filter for operation-level event (kind: "handler"), not RPC phase events
+  const loadModelEvent = json.recentEvents?.find(
+    (e) => e.op === "loadModel" && e.kind === "handler",
+  );
+  if (loadModelEvent) {
+    const tags = loadModelEvent.tags ?? {};
+    const gauges = loadModelEvent.gauges ?? {};
+    console.log("  sourceType:", tags["sourceType"] ?? "(not set)");
+    console.log("  cacheHit:", tags["cacheHit"] ?? "(not set)");
+    console.log("  totalLoadTime:", gauges["totalLoadTime"], "ms");
+    console.log(
+      "  modelInitializationTime:",
+      gauges["modelInitializationTime"],
+      "ms",
+    );
+    if (tags["cacheHit"] !== "true") {
+      console.log("  downloadTime:", gauges["downloadTime"] ?? "(cached)", "ms");
+      console.log(
+        "  totalBytesDownloaded:",
+        gauges["totalBytesDownloaded"] ?? "(cached)",
+      );
+      console.log(
+        "  downloadSpeedBps:",
+        gauges["downloadSpeedBps"] ?? "(cached)",
+      );
+    } else {
+      console.log("  (download metrics omitted - cache hit)");
+    }
+    if (gauges["checksumValidationTime"] !== undefined) {
+      console.log("  checksumValidationTime:", gauges["checksumValidationTime"], "ms");
+    }
+  } else {
+    console.log("  (no loadModel handler event captured)");
+    // Debug: show what ops are available
+    const ops = [...new Set(json.recentEvents?.map((e) => `${e.op}:${e.kind}`) ?? [])];
+    console.log("  Available ops:", ops.join(", "));
+  }
+
   console.log("\n=== Profiler JSON (structure) ===");
   console.log("  aggregates:", Object.keys(json.aggregates).length, "metrics");
   console.log("  recentEvents:", json.recentEvents?.length ?? 0, "events");

packages/sdk/profiling/aggregator.ts

@@ -95,6 +95,13 @@ export function recordEvent(
     const key = event.phase ? `${event.op}.${event.phase}` : event.op;
     recordStats(key, event.ms);
   }
+
+  if (event.gauges) {
+    const baseKey = event.phase ? `${event.op}.${event.phase}` : event.op;
+    for (const [gaugeName, value] of Object.entries(event.gauges)) {
+      recordStats(`${baseKey}.${gaugeName}`, value);
+    }
+  }
 }
 
 export function getAggregates(): Record<string, AggregatedStats> {

packages/sdk/profiling/exporters.ts

@@ -78,6 +78,20 @@ function formatDuration(ms: number): string {
   return `${(ms / 60000).toFixed(2)}m`;
 }
 
+function formatBytes(bytes: number): string {
+  if (bytes < 1024) return `${bytes.toFixed(0)} B`;
+  if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`;
+  if (bytes < 1024 * 1024 * 1024) return `${(bytes / (1024 * 1024)).toFixed(2)} MB`;
+  return `${(bytes / (1024 * 1024 * 1024)).toFixed(2)} GB`;
+}
+
+function formatThroughput(bps: number): string {
+  if (bps < 1024) return `${bps.toFixed(0)} B/s`;
+  if (bps < 1024 * 1024) return `${(bps / 1024).toFixed(1)} KB/s`;
+  if (bps < 1024 * 1024 * 1024) return `${(bps / (1024 * 1024)).toFixed(2)} MB/s`;
+  return `${(bps / (1024 * 1024 * 1024)).toFixed(2)} GB/s`;
+}
+
 function formatNumber(n: number): string {
   if (Number.isInteger(n)) {
     return n.toLocaleString();
@@ -89,6 +103,60 @@ function formatNumber(n: number): string {
   return n.toFixed(1);
 }
 
+type MetricType = "duration" | "bytes" | "throughput" | "number";
+
+function detectMetricType(metricName: string): MetricType {
+  const lowerName = metricName.toLowerCase();
+
+  if (lowerName.includes("bps") || lowerName.includes("speed")) {
+    return "throughput";
+  }
+
+  if (lowerName.includes("bytes") || lowerName.includes("downloaded") || lowerName.includes("size")) {
+    return "bytes";
+  }
+
+  if (
+    lowerName.endsWith("time") ||
+    lowerName.endsWith("ms") ||
+    lowerName.endsWith("duration") ||
+    lowerName.includes("ttfb") ||
+    lowerName.includes("latency") ||
+    lowerName.includes("wait") ||
+    lowerName.includes("overhead") ||
+    lowerName.includes("parse") ||
+    lowerName.includes("stringify") ||
+    lowerName.includes("validation") ||
+    lowerName.includes("execution") ||
+    lowerName.includes("serialization") ||
+    lowerName.includes("connection")
+  ) {
+    return "duration";
+  }
+
+  if (lowerName.includes("count") || lowerName.includes("token")) {
+    return "number";
+  }
+
+  // Default to duration for unknown metrics (most profiling is timing)
+  return "duration";
+}
+
+function formatMetricValue(value: number, metricName: string): string {
+  const type = detectMetricType(metricName);
+  switch (type) {
+    case "bytes":
+      return formatBytes(value);
+    case "throughput":
+      return formatThroughput(value);
+    case "number":
+      return formatNumber(value);
+    case "duration":
+    default:
+      return formatDuration(value);
+  }
+}
+
 function pad(s: string, len: number, align: "left" | "right" = "left"): string {
   if (s.length >= len) return s.substring(0, len);
   const spaces = " ".repeat(len - s.length);
@@ -118,15 +186,15 @@ export function exportTable(): string {
   const separator = "-".repeat(header.length);
 
   const rows = entries
-    .sort((a, b) => b[1].sum - a[1].sum)
+    .sort((a, b) => a[0].localeCompare(b[0]))
     .map(([metric, stats]) => {
       return [
         pad(metric, metricWidth),
         pad(formatNumber(stats.count), numWidth, "right"),
-        pad(formatDuration(stats.min), numWidth, "right"),
-        pad(formatDuration(stats.max), numWidth, "right"),
-        pad(formatDuration(stats.avg), numWidth, "right"),
-        pad(formatDuration(stats.sum), numWidth, "right"),
+        pad(formatMetricValue(stats.min, metric), numWidth, "right"),
+        pad(formatMetricValue(stats.max, metric), numWidth, "right"),
+        pad(formatMetricValue(stats.avg, metric), numWidth, "right"),
+        pad(formatMetricValue(stats.sum, metric), numWidth, "right"),
       ].join(" | ");
     });
 

packages/sdk/profiling/ring-buffer.ts

@@ -3,6 +3,8 @@
  * Oldest events are overwritten when full.
  */
 
+import { ProfilerInvalidCapacityError } from "@/utils/errors-client";
+
 export interface RingBufferState<T> {
   buffer: (T | undefined)[];
   capacity: number;
@@ -13,7 +15,7 @@ export interface RingBufferState<T> {
 
 export function createRingBuffer<T>(capacity: number): RingBufferState<T> {
   if (capacity < 1) {
-    throw new Error("Ring buffer capacity must be at least 1");
+    throw new ProfilerInvalidCapacityError(1);
   }
 
   return {
@@ -104,7 +106,7 @@ export function ringBufferResize<T>(
   newCapacity: number,
 ): RingBufferState<T> {
   if (newCapacity < 1) {
-    throw new Error("Ring buffer capacity must be at least 1");
+    throw new ProfilerInvalidCapacityError(1);
   }
 
   const items = ringBufferToArray(state);

packages/sdk/profiling/types.ts

@@ -49,3 +49,8 @@ export interface ProfilerExport {
   recentEvents?: ProfilingEvent[];
   exportedAt: number;
 }
+
+export interface LoadTimingStats {
+  modelInitializationTimeMs?: number;
+  totalLoadTimeMs?: number;
+}

packages/sdk/schemas/download-asset.ts

@@ -45,7 +45,9 @@ export const downloadAssetResponseSchema = z.object({
   error: z.string().optional(),
 });
 
-export const downloadTypeSchema = z.enum(["hyperdrive", "http", "registry"]);
+const downloadTypes = ["hyperdrive", "http", "registry"] as const;
+export const downloadTypeSchema = z.enum(downloadTypes);
+export const sourceTypeSchema = z.enum([...downloadTypes, "filesystem"]);
 
 export const downloadMetadataSchema = z.object({
   key: z.string(),
@@ -78,6 +80,7 @@ export type DownloadAssetOptions = z.input<
 export type DownloadAssetRequest = z.infer<typeof downloadAssetRequestSchema>;
 export type DownloadAssetResponse = z.infer<typeof downloadAssetResponseSchema>;
 export type DownloadType = z.infer<typeof downloadTypeSchema>;
+export type SourceType = z.infer<typeof sourceTypeSchema>;
 export type BaseDownloadEntry = z.infer<typeof downloadMetadataSchema> & {
   promise: Promise<string>;
   abortController: AbortController;