feat: offload large base64 media payloads to external blob files

.changeset/wire-blob-offloading.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/agent-core": patch
+"@moonshot-ai/kimi-code": patch
+---
+
+Offload large base64 media payloads from wire.jsonl into external blob files to reduce wire size and memory pressure during session replay. Includes an in-memory read-through cache on `BlobStore` so repeated rehydration avoids redundant disk reads.

apps/vis/server/src/app.ts

@@ -4,6 +4,7 @@ import { join, resolve } from 'node:path';
 
 import { Hono } from 'hono';
 
+import { blobsRoute } from './routes/blobs';
 import { contextRoute } from './routes/context';
 import { sessionDetailRoute } from './routes/session-detail';
 import { sessionsRoute } from './routes/sessions';
@@ -87,6 +88,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise<Hono> {
   api.route('/sessions', sessionDetailRoute());
   api.route('/sessions', wireRoute());
   api.route('/sessions', subagentsRoute());
+  api.route('/sessions', blobsRoute());
   // Mount contextRoute last because it currently uses a catch-all stub
   // (Phase C scope) that would otherwise shadow more specific routes
   // registered below it.

apps/vis/server/src/lib/blob-resolver.ts

@@ -0,0 +1,97 @@
+import type { ContentPart, WireEntry } from './agent-record-types';
+
+const BLOBREF_PROTOCOL = 'blobref:';
+
+function isBlobRef(url: string): boolean {
+  return url.startsWith(BLOBREF_PROTOCOL);
+}
+
+/** Convert a `blobref:<mime>;<hash>` URL into a vis-server blob route.
+ *  Non-blobref URLs are returned unchanged. */
+export function resolveBlobRefUrl(
+  url: string,
+  sessionId: string,
+  agentId: string,
+  baseUrl: string = '',
+): string {
+  if (!isBlobRef(url)) return url;
+  const rest = url.slice(BLOBREF_PROTOCOL.length);
+  const semiIdx = rest.indexOf(';');
+  if (semiIdx === -1) return url;
+  const mimeType = rest.slice(0, semiIdx);
+  const hash = rest.slice(semiIdx + 1);
+  if (hash.length === 0) return url;
+  const path = `/api/sessions/${encodeURIComponent(sessionId)}/blobs/${encodeURIComponent(hash)}?agent=${encodeURIComponent(agentId)}&mime=${encodeURIComponent(mimeType)}`;
+  return baseUrl ? `${baseUrl}${path}` : path;
+}
+
+/** Walk every record in a wire and replace blobref URLs with vis-server
+ *  blob routes so the UI can render them. Only mutates `entry.data`;
+ *  `entry.raw` is left untouched. */
+export function rehydrateWireEntries(
+  entries: readonly WireEntry[],
+  sessionId: string,
+  agentId: string,
+  baseUrl: string = '',
+): void {
+  for (const entry of entries) {
+    rehydrateRecord(entry.data as Record<string, unknown>, sessionId, agentId, baseUrl);
+  }
+}
+
+function rehydrateRecord(
+  record: Record<string, unknown>,
+  sessionId: string,
+  agentId: string,
+  baseUrl: string,
+): void {
+  const type = record['type'];
+  if (type === 'turn.prompt' || type === 'turn.steer') {
+    rehydrateParts(record['input'] as unknown as ContentPart[], sessionId, agentId, baseUrl);
+    return;
+  }
+  if (type === 'context.append_message') {
+    const message = record['message'] as { content: ContentPart[] };
+    rehydrateParts(message.content, sessionId, agentId, baseUrl);
+    return;
+  }
+  if (type === 'context.append_loop_event') {
+    const event = record['event'] as Record<string, unknown>;
+    if (event['type'] === 'tool.result') {
+      const result = event['result'] as Record<string, unknown>;
+      if (typeof result['output'] !== 'string') {
+        rehydrateParts(result['output'] as ContentPart[], sessionId, agentId, baseUrl);
+      }
+    } else if (event['type'] === 'content.part') {
+      rehydrateParts([event['part'] as ContentPart], sessionId, agentId, baseUrl);
+    }
+    return;
+  }
+}
+
+function rehydrateParts(
+  parts: ContentPart[],
+  sessionId: string,
+  agentId: string,
+  baseUrl: string,
+): void {
+  for (const part of parts) {
+    switch (part.type) {
+      case 'image_url':
+        part.imageUrl.url = resolveBlobRefUrl(part.imageUrl.url, sessionId, agentId, baseUrl);
+        break;
+      case 'audio_url':
+        part.audioUrl.url = resolveBlobRefUrl(part.audioUrl.url, sessionId, agentId, baseUrl);
+        break;
+      case 'video_url':
+        part.videoUrl.url = resolveBlobRefUrl(part.videoUrl.url, sessionId, agentId, baseUrl);
+        break;
+      default:
+        break;
+    }
+  }
+}
+
+export function isSafeBlobHash(hash: string): boolean {
+  return /^[a-f0-9]{64}$/.test(hash);
+}

apps/vis/server/src/lib/context-projector.ts

@@ -170,7 +170,7 @@ export function projectContext(entries: ReadonlyArray<WireEntry>): ContextProjec
         }];
         break;
       case 'usage.record': {
-        const scope = rec.usageScope ?? 'session';
+        const scope = (rec.usageScope ?? 'session') as 'session' | 'turn';
         addUsage(usage.byScope[scope], rec.usage);
         if (!usage.byModel[rec.model]) usage.byModel[rec.model] = { ...ZERO };
         addUsage(usage.byModel[rec.model]!, rec.usage);

apps/vis/server/src/lib/wire-reader.ts

@@ -52,28 +52,28 @@ export async function readAgentWire(path: string): Promise<WireReadResult> {
     let parsed: unknown;
     try {
       parsed = JSON.parse(line);
-    } catch (err) {
-      warnings.push(`line ${lineNo}: invalid JSON (${(err as Error).message})`);
+    } catch (error) {
+      warnings.push(`line ${lineNo}: invalid JSON (${(error as Error).message})`);
       continue;
     }
-    if (!isObject(parsed) || typeof parsed.type !== 'string') {
+    if (!isObject(parsed) || typeof parsed['type'] !== 'string') {
       warnings.push(`line ${lineNo}: missing 'type' field`);
       continue;
     }
     if (metadata === null) {
-      if (parsed.type !== 'metadata') {
+      if (parsed['type'] !== 'metadata') {
         throw new Error(`Wire file missing metadata header at line ${lineNo}`);
       }
       const pv = parsed['protocol_version'];
       const ca = parsed['created_at'];
       if (typeof pv !== 'string' || typeof ca !== 'number') {
-        throw new Error(`Wire metadata malformed at line ${lineNo}`);
+        throw new TypeError(`Wire metadata malformed at line ${lineNo}`);
       }
       try {
         migrations = resolveWireMigrations(pv);
-      } catch (err) {
+      } catch (error) {
         warnings.push(
-          `unrecognised protocol_version "${pv}" — parsing as best-effort (${(err as Error).message})`,
+          `unrecognised protocol_version "${pv}" — parsing as best-effort (${(error as Error).message})`,
         );
         migrations = bestEffortMigrations();
       }
@@ -85,18 +85,18 @@ export async function readAgentWire(path: string): Promise<WireReadResult> {
     try {
       migrated =
         migrations.length === 0
-          ? raw
+          ? (structuredClone(raw) as Record<string, unknown>)
           : (migrateWireRecord(
               raw as Record<string, unknown> & { type: string },
               migrations,
             ) as Record<string, unknown>);
-    } catch (err) {
+    } catch (error) {
       // A single record that won't migrate is not fatal — keep the raw
       // payload so the UI can still render whatever fields it understands.
       warnings.push(
-        `line ${lineNo}: migration failed (${(err as Error).message}); using raw record`,
+        `line ${lineNo}: migration failed (${(error as Error).message}); using raw record`,
       );
-      migrated = raw;
+      migrated = structuredClone(raw) as Record<string, unknown>;
     }
     records.push({ lineNo, data: migrated as AgentRecord, raw });
   }

apps/vis/server/src/routes/blobs.ts

@@ -0,0 +1,45 @@
+import { Hono } from 'hono';
+import { join } from 'node:path';
+import { readFile } from 'node:fs/promises';
+
+import { KIMI_CODE_HOME } from '../config';
+import { isSafeAgentId, readSessionDetail } from '../lib/session-store';
+import { isSafeBlobHash } from '../lib/blob-resolver';
+
+export function blobsRoute(home: string = KIMI_CODE_HOME): Hono {
+  const r = new Hono();
+  r.get('/:id/blobs/:hash', async (c) => {
+    const id = c.req.param('id');
+    const agentId = c.req.query('agent') ?? 'main';
+    const hash = c.req.param('hash');
+    if (!isSafeAgentId(agentId)) {
+      return c.json({ error: 'invalid agent id', code: 'BAD_REQUEST' }, 400);
+    }
+    if (!isSafeBlobHash(hash)) {
+      return c.json({ error: 'invalid blob hash', code: 'BAD_REQUEST' }, 400);
+    }
+    const detail = await readSessionDetail(home, id);
+    if (!detail) {
+      return c.json({ error: 'session not found', code: 'NOT_FOUND' }, 404);
+    }
+    const agent = detail.agents.find((a) => a.agentId === agentId);
+    if (!agent) {
+      return c.json(
+        { error: `agent "${agentId}" not found`, code: 'NOT_FOUND' },
+        404,
+      );
+    }
+    const blobPath = join(agent.homedir, 'blobs', hash);
+    let content: Buffer;
+    try {
+      content = await readFile(blobPath);
+    } catch {
+      return c.json({ error: 'blob not found', code: 'NOT_FOUND' }, 404);
+    }
+    const mimeType = c.req.query('mime') ?? 'application/octet-stream';
+    return new Response(content, {
+      headers: { 'content-type': mimeType },
+    });
+  });
+  return r;
+}

apps/vis/server/src/routes/context.ts

@@ -3,6 +3,7 @@ import { join } from 'node:path';
 
 import { KIMI_CODE_HOME } from '../config';
 import { isSafeAgentId, readSessionDetail } from '../lib/session-store';
+import { rehydrateWireEntries } from '../lib/blob-resolver';
 import { readAgentWire } from '../lib/wire-reader';
 import { projectContext } from '../lib/context-projector';
 
@@ -26,6 +27,8 @@ export function contextRoute(): Hono {
       const wire = await readAgentWire(
         join(detail.sessionDir, 'agents', agentId, 'wire.jsonl'),
       );
+      const baseUrl = new URL(c.req.url).origin;
+      rehydrateWireEntries(wire.records, id, agentId, baseUrl);
       const proj = projectContext(wire.records);
       return c.json({
         sessionId: id,

apps/vis/server/src/routes/wire.ts

@@ -3,6 +3,7 @@ import { join } from 'node:path';
 
 import { KIMI_CODE_HOME } from '../config';
 import { isSafeAgentId, readSessionDetail } from '../lib/session-store';
+import { rehydrateWireEntries } from '../lib/blob-resolver';
 import { readAgentWire } from '../lib/wire-reader';
 
 export function wireRoute(): Hono {
@@ -28,6 +29,8 @@ export function wireRoute(): Hono {
       const result = await readAgentWire(
         join(detail.sessionDir, 'agents', agentId, 'wire.jsonl'),
       );
+      const baseUrl = new URL(c.req.url).origin;
+      rehydrateWireEntries(result.records, id, agentId, baseUrl);
       return c.json({
         sessionId: id,
         agentId,