Avoid reloading analysis inputs after inference and measure storage waits

docs/api-reference/openapi.json

@@ -6742,6 +6742,7 @@
                             "enum": [
                               "catalog_lookup",
                               "catalog_write",
+                              "catalog_write_wait",
                               "player",
                               "player_response",
                               "download",
@@ -8002,6 +8003,7 @@
                                 "enum": [
                                   "catalog_lookup",
                                   "catalog_write",
+                                  "catalog_write_wait",
                                   "player",
                                   "player_response",
                                   "download",
@@ -9582,6 +9584,7 @@
                             "enum": [
                               "catalog_lookup",
                               "catalog_write",
+                              "catalog_write_wait",
                               "player",
                               "player_response",
                               "download",
@@ -10549,6 +10552,7 @@
                             "enum": [
                               "catalog_lookup",
                               "catalog_write",
+                              "catalog_write_wait",
                               "player",
                               "player_response",
                               "download",

platform/src/agents/providers/youtube/tools/analyze-video-frames.ts

@@ -33,7 +33,7 @@ export function executeAnalyzeVideoFrames(
     semanticKey: `frame-analysis:${JSON.stringify({ assetVersions: versions, focus: parsed.focus })}`,
     execute: async () => {
       if (!context.analyzeFrames) throw new Error('Frame analysis is unavailable.');
-      const assets = await readAnalysisAssets(context, versions, 'frame');
+      const { assets, assertAvailable } = await readAnalysisAssets(context, versions, 'frame', toolCallId);
       const values = assets.map(({ asset, value }) => {
         const frames = framesSchema.parse(value);
         if (frames.videoId !== asset.videoId || frames.frames.length !== 1)
@@ -62,7 +62,7 @@ export function executeAnalyzeVideoFrames(
       });
       context.signal.throwIfAborted();
       // Recheck availability after inference so deletion cannot recreate derived evidence.
-      await readAnalysisAssets(context, versions, 'frame');
+      assertAvailable();
       const sourceId = `youtube:${videoId}:frames`;
       return evidencePacketSchema.parse({
         packetId: `packet:${context.runId}:${safeIdPart(toolCallId)}`,

platform/src/agents/providers/youtube/tools/analyze-video-storyboard.ts

@@ -35,7 +35,7 @@ export function executeAnalyzeVideoStoryboard(
       if (!canAnalyzeStoryboard(context.researchDeadlineAt))
         throw new Error('Not enough research time to analyze storyboards. Finish with the evidence available and report missing visual evidence.');
       if (!context.analyzeStoryboard) throw new Error('Storyboard analysis is unavailable.');
-      const assets = await readAnalysisAssets(context, versions, 'storyboard_sheet');
+      const { assets, assertAvailable } = await readAnalysisAssets(context, versions, 'storyboard_sheet', toolCallId);
       const values = assets.map(({ asset, value }) => {
         const parsed = storyboardSchema.parse(value);
         if (parsed.videoId !== asset.videoId || parsed.sheets.length !== 1)
@@ -77,7 +77,7 @@ export function executeAnalyzeVideoStoryboard(
         modelCallId: `visual-analyst:${context.runId}:${toolCallId}`,
       }, context.researchDeadlineAt);
       context.signal.throwIfAborted();
-      await readAnalysisAssets(context, versions, 'storyboard_sheet');
+      assertAvailable();
       const videoId = storyboard.videoId;
       const sourceId = `youtube:${videoId}:storyboard`;
       return evidencePacketSchema.parse({

platform/src/agents/providers/youtube/tools/analyze-video-transcripts.ts

@@ -76,7 +76,8 @@ export async function executeAnalyzeVideoTranscript(
       operation: 'transcript',
       semanticKey,
       execute: async () => {
-        const { asset, value } = (await readAnalysisAssets(context, [parsed.assetVersion], 'transcript'))[0]!;
+        const { assets, assertAvailable } = await readAnalysisAssets(context, [parsed.assetVersion], 'transcript', toolCallId);
+        const { asset, value } = assets[0]!;
         const transcript = value as Transcript;
         if (
           transcript.videoId !== asset.videoId ||
@@ -107,7 +108,7 @@ export async function executeAnalyzeVideoTranscript(
               },
             ),
         );
-        await readAnalysisAssets(context, [parsed.assetVersion], 'transcript');
+        assertAvailable();
         return evidencePacketSchema.parse({
           packetId: `packet:${context.runId}:${safeIdPart(toolCallId)}`,
           kind: 'youtube_transcript',

platform/src/agents/providers/youtube/tools/stored-analysis.ts

@@ -1,3 +1,4 @@
+import { observeAgentOperation } from '../../../runtime/diagnostics';
 import { mapInBatches } from '../../../../lib/map-in-batches';
 import { z } from 'zod';
 import type { AgentToolContext } from '../tool-context';
@@ -18,37 +19,56 @@ export async function readAnalysisAssets(
   context: AgentToolContext,
   versions: string[],
   kind: SessionAssetKind,
+  toolCallId: string,
 ) {
   context.signal.throwIfAborted();
-  if (!context.session?.readAsset) throw new Error('Saved session asset access is unavailable.');
-  const readAsset = context.session.readAsset.bind(context.session);
-  return mapInBatches(versions, async (version) => {
+  const session = context.session;
+  if (!session?.readAsset || !session.generation || !session.has)
+    throw new Error('Saved session asset access is unavailable.');
+  const readAsset = session.readAsset.bind(session);
+  const generation = session.generation.bind(session);
+  const has = session.has.bind(session);
+  const initialGeneration = generation();
+  // Capture before loading so deletion during either storage I/O or inference
+  // fences the result, even if the same immutable version is retrieved again.
+  const assertAvailable = () => {
     context.signal.throwIfAborted();
-    if (
-      context.refreshEvidence &&
-      !(context.getEvidence?.() ?? []).some(
-        (packet) =>
-          packet.packetId.startsWith(`packet:${context.runId}:`) &&
-          packet.assetVersions?.includes(version) &&
-          packet.artifacts.some((artifact) =>
-            [
-              'youtube_complete_transcript',
-              'youtube_frame_retrieval',
-              'youtube_storyboard_retrieval',
-            ].includes(artifact.type),
-          ),
-      )
-    )
-      throw new Error(
-        'Fresh evidence was requested. Retrieve this asset in the current run before analysis.',
-      );
-    const stored = await readAsset(version);
-    context.signal.throwIfAborted();
-    if (!stored)
+    if (generation() !== initialGeneration || versions.some(version => !has(version)))
       throw new Error('Saved asset is unavailable or deleted. Retrieve it explicitly before analysis.');
-    if (stored.asset.kind !== kind) throw new Error(`Analysis requires saved ${kind} assets.`);
-    if (context.pinnedVideoId && stored.asset.videoId !== context.pinnedVideoId)
-      throw new Error(`inspect_video is pinned to video ${context.pinnedVideoId}.`);
-    return stored;
-  });
+  };
+  const assets = await observeAgentOperation(
+    { runId: context.runId, toolCallId, stage: 'analysis_asset_read', kind, assetCount: versions.length },
+    context.signal,
+    () => mapInBatches(versions, async (version) => {
+      context.signal.throwIfAborted();
+      if (
+        context.refreshEvidence &&
+        !(context.getEvidence?.() ?? []).some(
+          (packet) =>
+            packet.packetId.startsWith(`packet:${context.runId}:`) &&
+            packet.assetVersions?.includes(version) &&
+            packet.artifacts.some((artifact) =>
+              [
+                'youtube_complete_transcript',
+                'youtube_frame_retrieval',
+                'youtube_storyboard_retrieval',
+              ].includes(artifact.type),
+            ),
+        )
+      )
+        throw new Error(
+          'Fresh evidence was requested. Retrieve this asset in the current run before analysis.',
+        );
+      const stored = await readAsset(version);
+      context.signal.throwIfAborted();
+      if (!stored)
+        throw new Error('Saved asset is unavailable or deleted. Retrieve it explicitly before analysis.');
+      if (stored.asset.kind !== kind) throw new Error(`Analysis requires saved ${kind} assets.`);
+      if (context.pinnedVideoId && stored.asset.videoId !== context.pinnedVideoId)
+        throw new Error(`inspect_video is pinned to video ${context.pinnedVideoId}.`);
+      return stored;
+    }),
+  );
+  assertAvailable();
+  return { assets, assertAvailable };
 }

platform/src/agents/runtime/session-evidence.ts

@@ -59,6 +59,9 @@ export interface SessionBrief {
 }
 export interface SessionAccess {
   brief(): SessionBrief;
+  /** Synchronous SQLite checks for deletion while analysis is in flight. */
+  generation?(): number;
+  has?(version: string): boolean;
   readAsset?(version: string): Promise<{ asset: SessionAsset; value: unknown } | null>;
   evidence(version?: string): EvidencePacket[];
   readTranscriptEvidence?(version: string): Promise<{ packets: EvidencePacket[]; nextOffset?: number }>;

platform/src/lib/extraction-diagnostics.ts

@@ -4,7 +4,7 @@ const count = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER);
 const metric = z.number().nonnegative().max(Number.MAX_SAFE_INTEGER);
 // Deliberately no free-form strings, error messages, URLs, headers, or stderr.
 export const extractionEventSchema = z.object({
-  stage: z.enum(['catalog_lookup', 'catalog_write', 'player', 'player_response', 'download', 'complete', 'request', 'image_normalized',
+  stage: z.enum(['catalog_lookup', 'catalog_write', 'catalog_write_wait', 'player', 'player_response', 'download', 'complete', 'request', 'image_normalized',
     'caption_metadata', 'caption_retry', 'media_candidates', 'media_http', 'media_transfer', 'media_retry', 'media_retry_skipped', 'ffmpeg', 'ffmpeg_success', 'job', 'proxy', 'media_decode', 'media_admission', 'media_source']),
   failureReason: z.enum(['live_or_unconfirmed_broadcast', 'invalid_proxy_configuration', 'proxy_tunnel_failed', 'bot_challenge']).optional(),
   egress: z.enum(['direct', 'proxy']).optional(), proxySlot: z.number().int().min(0).max(3).optional(),

platform/src/lib/video-resources.ts

@@ -323,6 +323,7 @@ export async function loadVideoResource(
     let catalogWriteMs: number | undefined;
     let catalogSucceeded = false;
     let firstWrite: number | undefined;
+    let writeWaitStarted: number | undefined;
     let publicationClosed = false;
     const publishing = new Map<number, Promise<{ versions: VideoAssetReference[]; error?: never } | { error: unknown; versions?: never }>>();
     const ready = (frame: VideoFrames['frames'][number]) => {
@@ -353,6 +354,7 @@ export async function loadVideoResource(
           )
         : undefined;
       const writeStarted = Date.now();
+      writeWaitStarted = writeStarted;
       let versions: VideoAssetReference[] = [];
       if (fetched) {
         try {
@@ -379,15 +381,20 @@ export async function loadVideoResource(
       });
     } finally {
       publicationClosed = true;
+      // Extraction has settled. Only this remaining publication tail delays return;
+      // catalogWriteMs includes time that may overlap decoding. Also drain on failure.
+      writeWaitStarted ??= Date.now();
       await Promise.all(publishing.values());
+      const catalogWriteWaitMs = Math.max(0, Date.now() - writeWaitStarted);
       if (firstWrite !== undefined) catalogWriteMs = Date.now() - firstWrite;
       if (lastAttempt && catalogWriteMs !== undefined) {
         // Publication overlaps extraction for Media. These durations must not be added.
         emitExtractionDiagnostic(diagnostic, { ...lastAttempt, phase: 'catalog',
           recordedAt: Date.now(), elapsedMs: lookupMs + catalogWriteMs,
           outcome: catalogSucceeded ? 'success' : 'failed', status: undefined, failureKind: undefined,
           events: [{ stage: 'catalog_lookup', elapsedMs: lookupMs },
-            { stage: 'catalog_write', elapsedMs: catalogWriteMs }], droppedEvents: 0 });
+            { stage: 'catalog_write', elapsedMs: catalogWriteMs },
+            { stage: 'catalog_write_wait', elapsedMs: catalogWriteWaitMs }], droppedEvents: 0 });
       }
     }
   }

platform/test/fixtures/analysis-session.ts

@@ -6,6 +6,7 @@ import type { SessionAsset } from '../../src/agents/runtime/session-evidence';
 export function attachTestAssetStore(context: AgentToolContext) {
   const saved = new Map<string, { asset: SessionAsset; value: unknown }>();
   let ordinal = 0;
+  let generation = 0;
   const put = (
     kind: SessionAsset['kind'],
     videoId: string,
@@ -17,6 +18,8 @@ export function attachTestAssetStore(context: AgentToolContext) {
     return version;
   };
   context.session = {
+    generation: () => generation,
+    has: version => saved.has(version),
     brief: () => ({ assets: [...saved.values()].map(({ asset }) => asset), memories: [] }),
     readAsset: async (version) => saved.get(version) ?? null,
     evidence: () => [],
@@ -80,5 +83,9 @@ export function attachTestAssetStore(context: AgentToolContext) {
         }
       : {}),
   };
-  return { saved, put };
+  const remove = (version?: string) => {
+    generation++;
+    if (version) saved.delete(version); else saved.clear();
+  };
+  return { saved, put, remove };
 }

platform/test/session-evidence.integration.test.ts

@@ -1065,3 +1065,54 @@ test('single-sheet storyboard reuses the processor middle sheet across restored
       }
     }
   }));
+
+// Real SQLite availability/generation checks, with R2 used only to load analyst inputs.
+test.each(['transcript', 'frame', 'storyboard_sheet'] as const)(
+  'saved %s analysis reads once and fences deletion, clearing, and same-version restoration', async kind => {
+    for (const mutation of ['none', 'delete', 'clear', 'restore'] as const) {
+      await within(`analysis-recheck-${kind}-${mutation}`, async store => {
+        const value = kind === 'transcript' ? transcript() : kind === 'frame' ? oneFrame(id) : oneSheet(id);
+        const retrieve = () => store.retrieve(`recheck:${kind}`, kind, id, false,
+          async () => ({ value, cacheStatus: 'miss' as const }),
+          () => kind === 'storyboard_sheet' ? { sheetIndex: 0, manifestVersion: 'manifest' } : {});
+        const version = (await retrieve()).assetVersions![0]!;
+        const ctx = context(store, provider());
+        const duringInference = async () => {
+          if (mutation === 'none') return;
+          await store.delete(mutation === 'clear' ? undefined : version);
+          if (mutation === 'restore') expect((await retrieve()).assetVersions).toEqual([version]);
+        };
+        ctx.analyzeFrames = async () => {
+          await duringInference();
+          return { findings: [{ observation: 'A chart.', timestampsMs: [1000] }], warnings: [] };
+        };
+        ctx.analyzeStoryboard = async () => {
+          await duringInference();
+          return { findings: [{ observation: 'A chart.', frameIndexes: [0] }], warnings: [] };
+        };
+        ctx.transcriptPolicy = { mode: 'contextual_analysis', researchQuestion: 'Describe the opening.',
+          analyze: async () => {
+            await duringInference();
+            return { summary: 'A clear opening sentence.',
+              findings: [{ claim: 'A clear opening sentence.', excerptIds: ['opening'] }],
+              excerpts: [{ id: 'opening', text: 'A clear opening sentence.', startMs: 0, endMs: 5600 }],
+              coverage: { completeTranscriptRead: true, segmentCount: 1, startMs: 0, endMs: 5600 }, warnings: [] };
+          } };
+        const read = vi.spyOn(store, 'read');
+        const pending = kind === 'transcript'
+          ? executeAnalyzeVideoTranscript({ assetVersion: version, focus: 'Describe the opening.' }, ctx, 'analysis')
+          : kind === 'frame'
+            ? executeAnalyzeVideoFrames({ assetVersions: [version], focus: 'Describe the chart.' }, ctx, 'analysis')
+            : executeAnalyzeVideoStoryboard({ assetVersions: [version], focus: 'Describe the chart.' }, ctx, 'analysis');
+        if (mutation === 'none') {
+          await expect(pending).resolves.toMatchObject({ assetVersions: [version] });
+          expect(store.evidence()).toHaveLength(1);
+        } else {
+          await expect(pending).rejects.toThrow('unavailable or deleted');
+          expect(store.evidence()).toEqual([]);
+        }
+        expect(read).toHaveBeenCalledExactlyOnceWith(version);
+      });
+    }
+  },
+);

platform/test/video-catalog.test.ts

@@ -1,3 +1,4 @@
+import { extractionFixture } from './fixtures/extraction-diagnostic';
 import { captureVisualWork } from '../src/lib/visual-diagnostics';
 import { DatabaseSync } from 'node:sqlite';
 import { readFileSync } from 'node:fs';
@@ -1154,3 +1155,46 @@ test('stores a Media frame atomically with one R2 write and no separate image ob
   expect(restored?.value).toEqual(value);
   expect(restored?.verifiedImages?.[0]?.previewReference()).toEqual({sharedFrameKey:vi.mocked(f.bucket.put).mock.calls[0]![0]});
 });
+
+test.each([false, true])('reports only the remaining frame publication wait, overlap=%s', async overlap => {
+  const f = fixture();
+  f.env.YOUTUBE_FRAMES_BACKEND = 'media';
+  let now = 1000;
+  const clock = vi.spyOn(Date, 'now').mockImplementation(() => now);
+  let release!: () => void;
+  const gate = new Promise<void>(resolve => { release = resolve; });
+  let writeFinished!: () => void;
+  const finished = new Promise<void>(resolve => { writeFinished = resolve; });
+  let extractionFinished!: () => void;
+  const extracted = new Promise<void>(resolve => { extractionFinished = resolve; });
+  const save = vi.spyOn(VideoCatalog.prototype, 'saveMany').mockImplementation(async () => {
+    await gate;
+    writeFinished();
+    return [];
+  });
+  vi.mocked(getVideoFrames).mockImplementation(async (_env, _request, _signal, _limits, diagnostic, ready) => {
+    ready!(frame(1000));
+    if (overlap) {
+      now = 1050;
+      release();
+      await finished;
+    }
+    now = 1100;
+    diagnostic!({ ...extractionFixture, kind: 'frames', backend: 'media', elapsedMs: 100 });
+    extractionFinished();
+    return { videoId: id, frames: [frame(1000)], failures: [], meta: { partial: false, warnings: [] } };
+  });
+  const diagnostic = vi.fn();
+  try {
+    const pending = loadVideoResource(f.env,
+      { kind: 'frames', id, timestampsMs: [1000], maxWidth: 640, extractionTimeoutMs: 5000 }, diagnostic);
+    await extracted;
+    // Let the caller enter its post-extraction publication wait.
+    await Promise.resolve();
+    if (!overlap) { now = 1400; release(); }
+    await pending;
+    const catalog = diagnostic.mock.calls.map(([event]) => event).find(event => event.phase === 'catalog');
+    expect(catalog.events).toContainEqual({ stage: 'catalog_write_wait', elapsedMs: overlap ? 0 : 300 });
+    expect(catalog.events).toContainEqual({ stage: 'catalog_write', elapsedMs: overlap ? 100 : 400 });
+  } finally { release(); save.mockRestore(); clock.mockRestore(); }
+});