diff --git a/docs/api-reference/openapi.json b/docs/api-reference/openapi.json index 783508cb..7cf3777a 100644 --- a/docs/api-reference/openapi.json +++ b/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", diff --git a/platform/src/agents/providers/youtube/tools/analyze-video-frames.ts b/platform/src/agents/providers/youtube/tools/analyze-video-frames.ts index c24deb64..134d8fa5 100644 --- a/platform/src/agents/providers/youtube/tools/analyze-video-frames.ts +++ b/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)}`, diff --git a/platform/src/agents/providers/youtube/tools/analyze-video-storyboard.ts b/platform/src/agents/providers/youtube/tools/analyze-video-storyboard.ts index 5197aef0..5f5e727e 100644 --- a/platform/src/agents/providers/youtube/tools/analyze-video-storyboard.ts +++ b/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({ diff --git a/platform/src/agents/providers/youtube/tools/analyze-video-transcripts.ts b/platform/src/agents/providers/youtube/tools/analyze-video-transcripts.ts index b3764e71..7eca1589 100644 --- a/platform/src/agents/providers/youtube/tools/analyze-video-transcripts.ts +++ b/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', diff --git a/platform/src/agents/providers/youtube/tools/stored-analysis.ts b/platform/src/agents/providers/youtube/tools/stored-analysis.ts index 1b5315c8..c3429dc4 100644 --- a/platform/src/agents/providers/youtube/tools/stored-analysis.ts +++ b/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 }; } diff --git a/platform/src/agents/runtime/session-evidence.ts b/platform/src/agents/runtime/session-evidence.ts index ea33781a..b6c3fdbf 100644 --- a/platform/src/agents/runtime/session-evidence.ts +++ b/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 }>; diff --git a/platform/src/lib/extraction-diagnostics.ts b/platform/src/lib/extraction-diagnostics.ts index 8e19a0bc..1ec023fb 100644 --- a/platform/src/lib/extraction-diagnostics.ts +++ b/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(), diff --git a/platform/src/lib/video-resources.ts b/platform/src/lib/video-resources.ts index c5b49ccb..88edc45c 100644 --- a/platform/src/lib/video-resources.ts +++ b/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>(); 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,7 +381,11 @@ 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. @@ -387,7 +393,8 @@ export async function loadVideoResource( 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 }); } } } diff --git a/platform/test/fixtures/analysis-session.ts b/platform/test/fixtures/analysis-session.ts index 84d974e7..a5be6545 100644 --- a/platform/test/fixtures/analysis-session.ts +++ b/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(); 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 }; } diff --git a/platform/test/session-evidence.integration.test.ts b/platform/test/session-evidence.integration.test.ts index 65de0e1a..1f96f6f0 100644 --- a/platform/test/session-evidence.integration.test.ts +++ b/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); + }); + } + }, +); diff --git a/platform/test/video-catalog.test.ts b/platform/test/video-catalog.test.ts index 60b728b4..d5da6463 100644 --- a/platform/test/video-catalog.test.ts +++ b/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(resolve => { release = resolve; }); + let writeFinished!: () => void; + const finished = new Promise(resolve => { writeFinished = resolve; }); + let extractionFinished!: () => void; + const extracted = new Promise(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(); } +}); diff --git a/platform/test/youtube-stored-analysis.test.ts b/platform/test/youtube-stored-analysis.test.ts index c0bc6d8b..29210b87 100644 --- a/platform/test/youtube-stored-analysis.test.ts +++ b/platform/test/youtube-stored-analysis.test.ts @@ -192,6 +192,12 @@ describe('retrieval and saved-asset analysis boundary', () => { expect(analyst(context, kind)).toHaveBeenCalledTimes(2); for (const fetch of Object.values(provider)) expect(fetch).not.toHaveBeenCalled(); }); + it.each(kinds)('reads %s payloads only before inference', async (kind) => { + const { context, versions } = setup(); + const read = vi.spyOn(context.session!, 'readAsset'); + await analyze(kind, versions[kind], context); + expect(read).toHaveBeenCalledTimes(1); + }); it.each(kinds)('rejects missing, wrong-kind, and out-of-scope %s assets before inference', async (kind) => { const { context, provider, versions } = setup(); await expect(analyze(kind, 'f'.repeat(64), context)).rejects.toThrow('unavailable or deleted'); @@ -207,12 +213,78 @@ describe('retrieval and saved-asset analysis boundary', () => { const fn = vi.mocked(analyst(context, kind) as (...args: unknown[]) => Promise); const implementation = fn.getMockImplementation()!; fn.mockImplementationOnce(async (input: unknown) => { - store.saved.delete(versions[kind]); + store.remove(versions[kind]); + return implementation(input); + }); + await expect(analyze(kind, versions[kind], context)).rejects.toThrow('unavailable or deleted'); + expect(packets).toEqual([]); + }); + it.each(kinds)('fences %s analysis after deletion even if its version is restored', async (kind) => { + const { context, store, versions, packets } = setup(); + const fn = vi.mocked(analyst(context, kind) as (...args: unknown[]) => Promise); + const implementation = fn.getMockImplementation()!; + fn.mockImplementationOnce(async (input: unknown) => { + const saved = store.saved.get(versions[kind])!; + store.remove(versions[kind]); + store.saved.set(versions[kind], saved); + return implementation(input); + }); + await expect(analyze(kind, versions[kind], context)).rejects.toThrow('unavailable or deleted'); + expect(packets).toEqual([]); + }); + it.each(kinds)('fences %s analysis after an unrelated session deletion', async (kind) => { + const { context, store, versions, packets } = setup(); + const fn = vi.mocked(analyst(context, kind) as (...args: unknown[]) => Promise); + const implementation = fn.getMockImplementation()!; + fn.mockImplementationOnce(async (input: unknown) => { + store.remove(kind === 'frame' ? versions.transcript : versions.frame); return implementation(input); }); await expect(analyze(kind, versions[kind], context)).rejects.toThrow('unavailable or deleted'); expect(packets).toEqual([]); }); + it.each(kinds)('rejects %s deleted while inputs load before inference', async (kind) => { + const { context, store, versions, packets } = setup(); + const read = context.session!.readAsset!; + vi.spyOn(context.session!, 'readAsset').mockImplementationOnce(async version => { + const loaded = await read(version); + store.remove(version); + return loaded; + }); + await expect(analyze(kind, versions[kind], context)).rejects.toThrow('unavailable or deleted'); + expect(analyst(context, kind)).not.toHaveBeenCalled(); + expect(packets).toEqual([]); + }); + it.each(kinds)('rejects %s canceled during inference without another payload read', async (kind) => { + const { context, versions, packets } = setup(); + const controller = new AbortController(); + context.signal = controller.signal; + const read = vi.spyOn(context.session!, 'readAsset'); + const fn = vi.mocked(analyst(context, kind) as (...args: unknown[]) => Promise); + const implementation = fn.getMockImplementation()!; + fn.mockImplementationOnce(async (input: unknown) => { + controller.abort(new Error('Canceled analysis')); + return implementation(input); + }); + await expect(analyze(kind, versions[kind], context)).rejects.toThrow('Canceled analysis'); + expect(read).toHaveBeenCalledTimes(1); + expect(packets).toEqual([]); + }); + it.each(kinds)('records %s input read timing without source payloads', async kind => { + const { context, versions } = setup(); + const log = vi.spyOn(console, 'log').mockImplementation(() => {}); + try { + await analyze(kind, versions[kind], context); + const timings = log.mock.calls.map(([line]) => JSON.parse(line as string)) + .filter(event => event.stage === 'analysis_asset_read'); + expect(timings).toEqual([ + { event: 'agent_operation', runId: 'test-run', toolCallId: 'Describe the chart.', + stage: 'analysis_asset_read', kind, assetCount: 1, outcome: 'started' }, + { event: 'agent_operation', runId: 'test-run', toolCallId: 'Describe the chart.', + stage: 'analysis_asset_read', kind, assetCount: 1, outcome: 'succeeded', elapsedMs: expect.any(Number) }, + ]); + } finally { log.mockRestore(); } + }); it.each(kinds)('requires explicit retrieval in a refresh run before analyzing %s', async (kind) => { const { context, versions, packets } = setup(); context.refreshEvidence = true; diff --git a/reference/engineering/FRAME_EXTRACTION.md b/reference/engineering/FRAME_EXTRACTION.md index c9e43608..bffca5d9 100644 --- a/reference/engineering/FRAME_EXTRACTION.md +++ b/reference/engineering/FRAME_EXTRACTION.md @@ -26,6 +26,17 @@ are now stored separately. Old split-image records remain readable; storyboards retain separate deduplicated JPEGs. Frame storage, session attachment, and preview writes admit six images together; storyboard I/O remains bounded to four. +The catalog-phase extraction diagnostic records `catalog_write` as the elapsed window +from the first progressive write to the final drain. This window can include decoding +and gaps between writes; it is not the sum of storage service time. `catalog_write_wait` +records the publication tail after extraction settles, including remaining writes and +the final drain before returning. It is a subset of the write window, not another cost +to add. A fully overlapped write has zero remaining wait. On extraction failure, it +records the time spent draining already-started writes before propagating the error. +The existing per-write visual spans and D1/R2 counters retain the individual operations. +These measurements support a later decision about response/write overlap; persistence +still completes before retrieval returns. + After the private coordinator confirms successful publication, the agent provider creates request-local frame receipts tied to the bucket, immutable catalog reference, and exact frame payload. Session attachment can consume a matching receipt without diff --git a/reference/engineering/SESSION_EVIDENCE.md b/reference/engineering/SESSION_EVIDENCE.md index c0df34eb..3e280da0 100644 --- a/reference/engineering/SESSION_EVIDENCE.md +++ b/reference/engineering/SESSION_EVIDENCE.md @@ -4,6 +4,29 @@ The existing conversation-scoped AgentRuntimeDO owns source availability and der Descriptions are retained as separate citable excerpts of at most 2,000 characters, without the generic 800-character model projection cut. Historical metadata retains descriptions within a combined 48,000-character snapshot budget. Model-facing transcript analyses retain title, channel, caption provenance and supported entity quotes, but omit the repeated full description. The original description stays in saved evidence and its separate citable metadata packet. This keeps publisher prose from displacing completed findings. Finalization still applies its overall evidence budget and reports omitted context. Existing saved answers and analyses are not rewritten. Older metadata without a retained description can be loaded again before new analysis. +## Analysis availability and read timing + +Frame, storyboard and transcript analysis load their saved inputs once. Before loading, +they capture the session's SQLite deletion generation. After loading and again after +inference, they synchronously check that the generation is unchanged and every selected +asset row still exists. The post-inference check does not read D1 or R2. Deleting and +retrieving the same version during inference cannot restore that in-flight analysis. +Any generation change fences the analysis, including deletion of unrelated evidence or +memory. Normal retrieval and refresh do not advance the generation. + +The initial read still verifies the shared catalog payload. The final check establishes +session availability only; it does not detect a shared catalog version or R2 object +removed during inference. Shared sources currently have no deletion or retention policy. +A future shared-source deletion feature must revisit this boundary. + +`agent_operation` logs with `stage=analysis_asset_read` measure the wall-clock time to +load and validate the selected input batch, including session checks, catalog lookups, +R2 downloads and payload hydration. Each event carries `runId`, `toolCallId`, asset kind, +count and outcome, with `elapsedMs` on completion. Concurrent reads contribute their +batch duration, not the sum of individual read durations. These logs contain no source +payloads. They are separate from inference timing. This change adds no run-scoped cache; +analysis still performs its initial saved-payload read even after same-run retrieval. + ## Flow 1. Before classification, original user messages and completed answers are indexed into the SDK Session history. Existing run records remain authoritative for execution and dashboard restoration. Failed-run user messages are included; answers whose evidence was deleted are excluded. Backfill reads changed runs incrementally, and subsequent updates only rewrite changed text. The classifier receives eight completed conversation turns, a compact inventory with coverage and collection times, the indexed message count, and bounded derived memory. It never receives raw transcripts or images through the inventory.