From d4875bc79477bd8d344698c459a9424e86e8dfa5 Mon Sep 17 00:00:00 2001 From: Jonas Silva <7734262+jonaslsaa@users.noreply.github.com> Date: Tue, 3 Nov 2026 01:42:58 +0100 Subject: [PATCH 1/3] Keep durable chat images once, out of Pi's transcript rows Tool results' images go to memory's store in afterTool and leave their name in Pi's row; offloadImages does the same for a user's input. beforeRequest puts the images back for each request. Closes #282 --- docs/durable.md | 5 ++-- src/core/index.ts | 2 +- src/durable/optchat.ts | 60 +++++++++++++++++++++++++++++++++++++----- test/durable.test.ts | 59 ++++++++++++++++++++++++++++++++++++++++- 4 files changed, 116 insertions(+), 10 deletions(-) diff --git a/docs/durable.md b/docs/durable.md index 83967296..fb30758d 100644 --- a/docs/durable.md +++ b/docs/durable.md @@ -20,7 +20,7 @@ With the Agents SDK's `PiHarness` this goes in its `harness` factory, which supp ## Child conversations -`memoryExtension.snapshot(task, ctx)` builds a child conversation's first input to submit (not `init` entries): the view with `afterView`'s text, the previous exchange, the run so far (in a tool's `execute`, up to the call that made it), then `task`. It is one text block, `[{ type: 'text', text }]`, so cache marks reach it. Tool calls and results can't go into another conversation as they are, so the exchange and run are `kind: text` lines as memory logs them, images as `[image]`. +`memoryExtension.snapshot(task, ctx)` builds a child conversation's first input to submit (not `init` entries): the view with `afterView`'s text, the previous exchange, the run so far (in a tool's `execute`, up to the call that made it), then `task`. It is one text block, `[{ type: 'text', text }]`, so cache marks reach it. Tool calls and results can't go into another conversation as they are, so the exchange and run are `kind: text` lines as memory logs them, images as `[image]` (one offloaded from Pi's transcript by its name). It isn't stable across restarts, by design: take it once. With Pi Durable's subagent recipe (find the child by `ownerTaskId`, submit with a `requestId`, `replay: 'safe'`), a rerun finds the child and its input and doesn't snapshot again: @@ -61,7 +61,8 @@ Keep `optchat` selected in the child, as its copy of the owner's agent does. It - **A failed save** stops memory's log: requests keep the view it has, and after a restart logging resumes from the receipts and Pi's transcript. Cloudflare resets the object itself only when a commit fails after the code ran (the output gate, which drops Pi's writes too); a write that throws at once, like `SQLITE_FULL` at the storage limit, doesn't, so the restart waits for idle eviction, a deploy, or the host's `ctx.abort()`, which the example calls from `warn`. `SQLITE_FULL` stops Pi's writes as well; the errors only memory's writes could hit (a row over 2 MB, a message index saved twice) are ruled out by the store's parts and by one Memory per database. - **Compaction:** turn Pi's off; the extension also declines any that starts. OptChat's summaries replace it. - **Large tool results** are sent as their head and tail (`boundedMessage`), capped as Pi does. -- **Images:** each image in a user message or tool result is named in memory's text as Pi's OptChat names it, `[image <16 hex digits>]` (SHA-256 of its base64, by Web Crypto), and kept in the store before the message is logged, so `zoom(id, 1)` returns it and the previous exchange names it instead of sending it again. OptChat doesn't resize images: Pi Durable's Photon processor (`createCloudflarePhotonImages()`) runs in Workers, but the memory it decodes in never shrinks again. Shrink photos when they arrive instead (Cloudflare Images, say), so Pi Durable's transcript and the model get the small copy too; Pi Durable keeps each transcript entry, base64 images included, in one row, so on Cloudflare a message's images must stay well under 2 MB anyway. `SqliteStore` keeps images in `optchat_images`; to keep large binary data out of the database (its row size limit and storage cost), give it `{ images }`, an `ImageStore` with `putImage` and `image`. The example keeps them in R2. +- **Images:** each image in a user message or tool result is named in memory's text as Pi's OptChat names it, `[image <16 hex digits>]` (SHA-256 of its base64, by Web Crypto), and kept in the store before the message is logged, so `zoom(id, 1)` returns it and the previous exchange names it instead of sending it again. OptChat doesn't resize images: Pi Durable's Photon processor (`createCloudflarePhotonImages()`) runs in Workers, but the memory it decodes in never shrinks again. Shrink photos when they arrive instead (Cloudflare Images, say), so the store and the model get the small copy too. `SqliteStore` keeps images in `optchat_images`; to keep large binary data out of the database (its row size limit and storage cost), give it `{ images }`, an `ImageStore` with `putImage` and `image`. The example keeps them in R2. +- **Images in Pi's transcript:** Pi Durable keeps each transcript entry, base64 images included, in one row. So each image goes to memory's store once, and Pi's row keeps only its name as a text block, `[image <16 hex digits>]`. A model-issued tool's images go there by themselves (`afterTool`); a user's go when the app passes the input through `offloadImages` before submitting it: `root.submit({ type: 'input', content: await memoryExtension.offloadImages(content) }, ctx)`. Each request gets the images back (`beforeRequest`), read from the store once per request; memory's text is the same either way. An image the store refuses stays inline, as do a nested call's images, a result whose text would pass 30,000 bytes with the names (Pi Durable bounds a result's text by keeping one of its blocks, which would drop names), and an input the app didn't pass through `offloadImages`; on Cloudflare a message's inline images must stay well under 2 MB. A name the store lacks or can't read is sent as the text it is. Hooks run in the order extensions are selected (install order by default): a host's own `beforeRequest` or `afterTool` before OptChat's sees images, one after sees names. A host that reads Pi's transcript itself, such as a chat UI, sees the names too: get an image with `memory.store.image(name)`. - **Tools:** `zoom` and `date` (plus `search` with `search: true`), replay-safe. - **Cache marks:** `withCacheMarks(models)` adds OptChat's marks to Anthropic requests that carry a view, so a view that grows at its end is read from the cache. A request with its own `onPayload` is left alone. - **Summaries of a group chat:** a chat that isn't one user talking to OptChat, such as a housing board with several people, gives `createCompressor` its own `system` prompt and names each speaker in the text, like "[Kari] …". diff --git a/src/core/index.ts b/src/core/index.ts index 5267c71b..4b2ee55f 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -1,7 +1,7 @@ /** OptChat's memory for any host: the log, the summary tree and its views, zoom/date/search, the compactor, cache marks and * memory's prompts. It imports nothing from Pi or Node at runtime: the host gives it a `Store` and the compactor's model. * Only what a host needs is exported here; Pi's own code imports the modules directly. */ -export { isEntry, Memory, openMemory, Refused, VIEW, type Compression, type Compressor, type Entry, type Kind, type MemoryOptions, type Origin, type Part, type Summary } from './memory.ts'; +export { CAP, isEntry, Memory, openMemory, Refused, VIEW, type Compression, type Compressor, type Entry, type Kind, type MemoryOptions, type Origin, type Part, type Summary } from './memory.ts'; export { isMime, type ImageStore, type Lease, type Mime, type Saved, type Store } from './store.ts'; export { ACCEPTED_BYTES, createCompressor, type CompactorModel, type CompressorOptions, type Retryable } from './compactor.ts'; export { cachePayload } from './cache.ts'; diff --git a/src/durable/optchat.ts b/src/durable/optchat.ts index 4516a2c1..5573152b 100644 --- a/src/durable/optchat.ts +++ b/src/durable/optchat.ts @@ -1,24 +1,32 @@ import { AssistantEntry, CompactionTask, defineExtension, defineTask, defineTool, GenerationTask, hook, InboxDoc, LiveDoc, ResetEntry, ROOT_CONVERSATION_ID, - section, ToolResultEntry, UserEntry, type Conversation, type ConversationId, type EntryId, type EntryRecord, type GenerationHooks, type Harness, - type HarnessOptions, type ToolExecutionResult, type UserInput, + section, ToolResultEntry, ToolTask, UserEntry, type Conversation, type ConversationId, type EntryId, type EntryRecord, type GenerationHooks, type Harness, + type HarnessOptions, type ToolExecutionResult, type ToolHooks, type UserInput, } from '@earendil-works/pi-durable'; import type { Static, TSchema } from 'typebox'; import { - allowSearch, boundedMessage, cachePayload, CHILD, childInput, completedExchange, CONTINUITY, decode, isImage, isMime, logMessage, memoryTools, refreshBudget, refreshedRun, + allowSearch, boundedMessage, cachePayload, CAP, CHILD, childInput, completedExchange, CONTINUITY, decode, isImage, isMime, loadImages, logMessage, memoryTools, refreshBudget, refreshedRun, refreshPoint, SEARCH_DOC, searchTool, textContent, TURN_REFRESHED, VIEW_DOC, withView, type AnyMessage, type ImageContent, type Memory, } from '../core/index.ts'; export type Context = Parameters[1]; type Message = Parameters[0]['messages'][number]; type Models = Exclude unknown>; +type Block = NonNullable[number]; +type TextContent = Extract; /** The entries that are the chat: other kinds (system prompt changes, resets, compactions, other extensions' own) stay out of memory. */ const CHAT = new Set([UserEntry.kind, AssistantEntry.kind, ToolResultEntry.kind]); const chat = (entry: EntryRecord) => CHAT.has(entry.kind) ? entry.model ?? [] : []; +/** A text block that is only an image's name: what an image offloaded to memory's store leaves in Pi's transcript. */ +const MARKER = /^\[image [0-9a-f]{16}\]$/; +const isMarker = (block: Block): block is TextContent => block.type === 'text' && MARKER.test(block.text); +/** A zoom's page names the images it returns already, so memory leaves out the names its offloaded images leave behind. */ +const unmarkZoom = (m: Message) => m.role === 'toolResult' && m.toolName === 'zoom' ? { ...m, content: m.content.filter(block => !isMarker(block)) } : m; +const failure = (error: unknown) => error instanceof Error ? error.message : String(error); /** What memory logs of an entry: its chat, or a reset's note (`Conversation.reset(text)`, a tool's handoff) marked as one. Pi sends * the note in the context it starts, which is logged only once it ends, so the model never sees it twice. */ -const remembered = (entry: EntryRecord) => !ResetEntry.is(entry) ? chat(entry) +const remembered = (entry: EntryRecord) => !ResetEntry.is(entry) ? chat(entry).map(unmarkZoom) : (entry.model ?? []).map(note => ({ role: 'user' as const, content: `[Handoff note] ${textContent(note.content)}`, timestamp: note.timestamp })); /** Each message OptChat logs from a Pi entry carries the entry's id, so the log itself says how far it has read Pi's transcript. */ const RECEIPT = 'pi:'; @@ -101,6 +109,40 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp } }; + /** `content` with each image kept in memory's store and replaced by its name, as memory names it; one the store refuses stays. */ + const offload = async (content: readonly Block[]) => { + const names = await imageNames(content.filter(isImage)), kept = new Set(); + await Promise.all([...names.values()].map(async ({ image, name }) => { + if (!isMime(image.mimeType)) return; + try { await memory.store.putImage(name, image.mimeType, decode(image.data)); kept.add(image.data); } + catch (error) { console.error(`OptChat could not keep an image: ${failure(error)}`); } + })); + return content.map(block => isImage(block) && kept.has(block.data) ? { type: 'text' as const, text: `[image ${names.get(block.data)?.name}]` } : block); + }; + /** A model-issued tool's images go to the store, so Pi's transcript holds their names. A result whose text would pass CAP keeps + * them: Pi Durable bounds a result's text by keeping one of its text blocks, and a text over CAP is sent as its head and tail, + * either of which can drop a name. */ + const afterTool: ToolHooks['afterTool'] = async (call, result, api) => { + if (api.conversationId !== id || call.parent || !result.output?.some(isImage)) return undefined; + const output = await offload(result.output); + return encoder.encode(textContent(output)).length <= CAP ? { ...result, output } : undefined; + }; + /** The request's messages with each image name in a user message or tool result replaced by the image, read from the store once + * per request. A name the store lacks or can't read stays as it is. */ + const restore = async (messages: readonly Message[]) => { + const loads = new Map>(); + const load = (block: TextContent) => { + let image = loads.get(block.text); + if (!image) loads.set(block.text, image = loadImages(memory.store, block.text).then(([found]) => found ?? block, error => { + console.error(`OptChat could not read an image: ${failure(error)}`); + return block; + })); + return image; + }; + return Promise.all(messages.map(async m => m.role !== 'user' && m.role !== 'toolResult' || typeof m.content === 'string' || !m.content.some(isMarker) ? m + : { ...m, content: await Promise.all(m.content.map(block => isMarker(block) ? load(block) : block)) })); + }; + const handle = async (ctx: Context): Promise => { const conversation = await opened().harness.conversation(id, ctx); if (!conversation) throw new Error(`No conversation ${id}.`); @@ -210,8 +252,9 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp if (cut && await logThrough(cut.entry.id)) keep = cut.i; return { system: messages.slice(0, at), previous: shared.previous, run: keep ? refreshedRun(own, keep, TURN_REFRESHED) : whole, view: shared.view }; }; - const beforeRequest: GenerationHooks['beforeRequest'] = async ({ messages }, api, ctx) => { + const beforeRequest: GenerationHooks['beforeRequest'] = async (request, api, ctx) => { if (api.conversationId !== id) return undefined; + const messages = await restore(request.messages); const { system, previous, run, view } = await compose(ctx, { messages, models: api.models }); return system.length < messages.length ? { messages: withView([...system, ...previous, ...run], view) } : undefined; }; @@ -268,7 +311,7 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp name: 'optchat', tools, sections: [section('optchat', ({ conversationId }) => conversationId === id ? text : childPrompt, { tag: false })], - hooks: [hook(GenerationTask, { beforeRequest }), hook(CompactionTask, { beforeCompact: (_compaction, api) => api.conversationId === id ? { decline: true } : undefined })], + hooks: [hook(GenerationTask, { beforeRequest }), hook(ToolTask, { afterTool }), hook(CompactionTask, { beforeCompact: (_compaction, api) => api.conversationId === id ? { decline: true } : undefined })], tasks: [Remember], }), /** The first input of a child conversation that sees OptChat's memory as the conversation's next request would: the view (with @@ -280,6 +323,11 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp // Text blocks, which cachePayload marks; a plain string is sent as one. return [{ type: 'text', text: childInput({ view, previous, run, task }) }]; }, + /** `input` with its images kept in memory's store, each replaced by its name, as tool results' are: pass a user input through it + * before submitting it, so Pi's transcript holds no image bytes; requests get the images back. An image the store refuses stays. */ + async offloadImages(input: UserInput): Promise { + return typeof input === 'string' ? input : offload(input); + }, /** Starts following the conversation, and resets it now if a run ended before a restart. Returns a function that stops. */ attach(harness: Harness, ctx: Context) { attached.set(harness, ctx); diff --git a/test/durable.test.ts b/test/durable.test.ts index cd9a40c3..964441b6 100644 --- a/test/durable.test.ts +++ b/test/durable.test.ts @@ -12,7 +12,7 @@ import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context'; import { NodeSqliteDatabase } from '@earendil-works/chord/sqlite/node'; import { createModels } from '@earendil-works/pi-ai/models'; import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall, type FauxResponseStep } from '@earendil-works/pi-ai/providers/faux'; -import { CHILD, cachePayload, openMemory, textContent, type Compressor, type ImageStore, type ImportedEntry, type Mime } from '../src/core/index.ts'; +import { CHILD, cachePayload, isImage, openMemory, textContent, type Compressor, type ImageStore, type ImportedEntry, type Mime } from '../src/core/index.ts'; import { optchat, optchatImport, SqliteStore, withCacheMarks, type OptChatOptions, type Read, type SqliteStoreOptions } from '../src/durable/index.ts'; import { importTables } from '../src/durable/store.ts'; import { imageRef } from '../src/images.ts'; @@ -267,6 +267,63 @@ test('with an images store, images go there, and zoom reads them from it', async } finally { await agent.close(); } }); +/** A tool that returns `images` after a line of text. */ +const camera = (...images: (typeof photo)[]): Extension => defineExtension({ name: 'camera', tools: [defineTool({ + name: 'camera', description: 'Takes a photo.', parameters: Type.Object({}), replay: 'safe', + execute: async () => ({ output: [{ type: 'text', text: 'snap' }, ...images] }), +})] }); +/** Whether any of Pi Durable's rows holds `data`. */ +const inPiRows = (agent: Awaited>, data: string) => agent.sql.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name LIKE 'pi\\_%' ESCAPE '\\'").all() + .some(({ name }) => agent.sql.prepare(`SELECT * FROM "${String(name)}"`).all().some(row => JSON.stringify(row).includes(data))); +const imagesIn = (request: Sent | undefined, role: string) => (request ?? []).flatMap(m => m.role === role && Array.isArray(m.content) ? m.content.filter(isImage) : []); +const memoryImages = () => { + const kept = new Map(); + const images: ImageStore = { putImage: async (name, mimeType, data) => void kept.set(name, { mimeType, data }), image: name => kept.get(name) }; + return { kept, images }; +}; + +test('a tool\'s images and a user\'s sent through offloadImages are kept once, in the images store: Pi\'s rows hold their names, requests the images', async () => { + const { kept, images } = memoryImages(), f = file(); + const user = { ...photo, data: Buffer.alloc(1000, 9).toString('base64') }; + const agent = await open(f, [ + fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' }), fauxAssistantMessage('nice'), + ], undefined, { images, extensions: () => [camera(photo)] }); + try { + await (await agent.root.submit({ type: 'input', content: await agent.chat.offloadImages([{ type: 'text', text: 'take one like this' }, user]) }, context)).wait(context); + await agent.idle(); + assert.ok(!inPiRows(agent, photo.data) && !inPiRows(agent, user.data)); + assert.deepEqual(imagesIn(agent.sent[0], 'user'), [user]); + assert.deepEqual(imagesIn(agent.sent[1], 'user'), [user]); + assert.deepEqual(imagesIn(agent.sent[1], 'toolResult'), [photo]); + // Memory names them as it names inline images. + assert.deepEqual(agent.memory.entries().map(e => e.text), [`take one like this\n${imageRef(user)}`, 'camera {}', `camera: snap\n${imageRef(photo)}`, 'nice']); + assert.equal(kept.size, 2); + } finally { await agent.close(); } + // After a restart, zoom returns the tool's image; its own result is offloaded too, and memory logs zoom's page as it is. + const again = await open(f, [fauxAssistantMessage(fauxToolCall('zoom', { id: 2 }), { stopReason: 'toolUse' }), fauxAssistantMessage('here')], undefined, { images }); + try { + await again.ask('show it again'); + assert.deepEqual(imagesIn(again.sent[1], 'toolResult'), [photo]); + assert.ok(!inPiRows(again, photo.data)); + assert.equal(again.memory.text(6).split(imageRef(photo)).length, 2); + } finally { await again.close(); } +}); + +test('an image the store refuses stays inline in Pi\'s row and still reaches the model', async (t) => { + t.mock.method(console, 'error', () => {}); + const { kept, images } = memoryImages(), refused = { ...photo, data: Buffer.alloc(1000, 8).toString('base64') }; + const putImage = images.putImage; + images.putImage = (name, mimeType, data) => name === imageRef(refused).slice(7, -1) ? Promise.reject(new Error('full')) : putImage(name, mimeType, data); + const agent = await open(file(), [fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' }), fauxAssistantMessage('nice')], + undefined, { images, extensions: () => [camera(photo, refused)] }); + try { + await agent.ask('take two'); + assert.ok(!inPiRows(agent, photo.data) && inPiRows(agent, refused.data)); + assert.deepEqual(imagesIn(agent.sent[1], 'toolResult'), [photo, refused]); + assert.equal(kept.size, 1); + } finally { await agent.close(); } +}); + test('cache marks go on an Anthropic request that carries the view; a request with its own onPayload is left alone', async () => { const faux = fauxProvider({ api: 'anthropic-messages' }), models = createModels(); models.setProvider(faux.provider); From be680c4c37b84151f0f954592a00a34177a3fe54 Mon Sep 17 00:00:00 2001 From: Jonas Silva <7734262+jonaslsaa@users.noreply.github.com> Date: Tue, 3 Nov 2026 01:53:45 +0100 Subject: [PATCH 2/3] Offload a tool's images only inside Pi Durable's bound; move image code to durable/images.ts Pi Durable bounds a tool result after afterTool by keeping one text block, which dropped the refs of a result over 50 KB or 2,000 lines. Decide first, then put; zoom's images skip the put. One ref format from core (ref, refName), one keep helper, tests for the bound, a failed read and the restart split out. --- docs/durable.md | 4 +- src/core/images.ts | 5 +++ src/core/index.ts | 4 +- src/durable/images.ts | 85 ++++++++++++++++++++++++++++++++++++++++++ src/durable/optchat.ts | 81 ++++++++-------------------------------- src/images.ts | 4 +- test/durable.test.ts | 65 ++++++++++++++++++++++---------- 7 files changed, 156 insertions(+), 92 deletions(-) create mode 100644 src/durable/images.ts diff --git a/docs/durable.md b/docs/durable.md index fb30758d..c1859621 100644 --- a/docs/durable.md +++ b/docs/durable.md @@ -20,7 +20,7 @@ With the Agents SDK's `PiHarness` this goes in its `harness` factory, which supp ## Child conversations -`memoryExtension.snapshot(task, ctx)` builds a child conversation's first input to submit (not `init` entries): the view with `afterView`'s text, the previous exchange, the run so far (in a tool's `execute`, up to the call that made it), then `task`. It is one text block, `[{ type: 'text', text }]`, so cache marks reach it. Tool calls and results can't go into another conversation as they are, so the exchange and run are `kind: text` lines as memory logs them, images as `[image]` (one offloaded from Pi's transcript by its name). +`memoryExtension.snapshot(task, ctx)` builds a child conversation's first input to submit (not `init` entries): the view with `afterView`'s text, the previous exchange, the run so far (in a tool's `execute`, up to the call that made it), then `task`. It is one text block, `[{ type: 'text', text }]`, so cache marks reach it. Tool calls and results can't go into another conversation as they are, so the exchange and run are `kind: text` lines as memory logs them, images as `[image]` (one kept out of Pi's transcript by its ref). It isn't stable across restarts, by design: take it once. With Pi Durable's subagent recipe (find the child by `ownerTaskId`, submit with a `requestId`, `replay: 'safe'`), a rerun finds the child and its input and doesn't snapshot again: @@ -62,7 +62,7 @@ Keep `optchat` selected in the child, as its copy of the owner's agent does. It - **Compaction:** turn Pi's off; the extension also declines any that starts. OptChat's summaries replace it. - **Large tool results** are sent as their head and tail (`boundedMessage`), capped as Pi does. - **Images:** each image in a user message or tool result is named in memory's text as Pi's OptChat names it, `[image <16 hex digits>]` (SHA-256 of its base64, by Web Crypto), and kept in the store before the message is logged, so `zoom(id, 1)` returns it and the previous exchange names it instead of sending it again. OptChat doesn't resize images: Pi Durable's Photon processor (`createCloudflarePhotonImages()`) runs in Workers, but the memory it decodes in never shrinks again. Shrink photos when they arrive instead (Cloudflare Images, say), so the store and the model get the small copy too. `SqliteStore` keeps images in `optchat_images`; to keep large binary data out of the database (its row size limit and storage cost), give it `{ images }`, an `ImageStore` with `putImage` and `image`. The example keeps them in R2. -- **Images in Pi's transcript:** Pi Durable keeps each transcript entry, base64 images included, in one row. So each image goes to memory's store once, and Pi's row keeps only its name as a text block, `[image <16 hex digits>]`. A model-issued tool's images go there by themselves (`afterTool`); a user's go when the app passes the input through `offloadImages` before submitting it: `root.submit({ type: 'input', content: await memoryExtension.offloadImages(content) }, ctx)`. Each request gets the images back (`beforeRequest`), read from the store once per request; memory's text is the same either way. An image the store refuses stays inline, as do a nested call's images, a result whose text would pass 30,000 bytes with the names (Pi Durable bounds a result's text by keeping one of its blocks, which would drop names), and an input the app didn't pass through `offloadImages`; on Cloudflare a message's inline images must stay well under 2 MB. A name the store lacks or can't read is sent as the text it is. Hooks run in the order extensions are selected (install order by default): a host's own `beforeRequest` or `afterTool` before OptChat's sees images, one after sees names. A host that reads Pi's transcript itself, such as a chat UI, sees the names too: get an image with `memory.store.image(name)`. +- **Images in Pi's transcript:** Pi Durable keeps each transcript entry, base64 images included, in one row. So each image goes to memory's store once, and Pi's row keeps only its ref as a text block, `[image <16 hex digits>]`. A model-issued tool's images go there by themselves (`afterTool`); a user's go when the app passes the input through `offloadImages` before submitting it: `root.submit({ type: 'input', content: await memoryExtension.offloadImages(content) }, ctx)`. Each request gets the images back (`beforeRequest`), each read from the store once per request; memory's text is the same either way. `putImage` must keep the image or throw: one it refuses stays inline. So do a nested call's images, an input the app didn't pass through `offloadImages`, and a tool result whose text could pass Pi Durable's default bound with the refs (50 KB or 2,000 lines): Pi bounds a result by keeping one of its text blocks, which would drop the refs. The hook can't see a tool's own `outputLimits`, so a tool with lower limits must stay inside them to keep its images. On Cloudflare, a message's inline images must stay well under 2 MB. A ref the store lacks or can't read is sent as its text. Hooks run in the order extensions are selected (install order by default): a host's own `beforeRequest` or `afterTool` before OptChat's sees images, one after sees refs. A host that reads Pi's transcript itself, such as a chat UI, sees the refs too: get an image with `memory.store.image(name)`. - **Tools:** `zoom` and `date` (plus `search` with `search: true`), replay-safe. - **Cache marks:** `withCacheMarks(models)` adds OptChat's marks to Anthropic requests that carry a view, so a view that grows at its end is read from the cache. A request with its own `onPayload` is left alone. - **Summaries of a group chat:** a chat that isn't one user talking to OptChat, such as a housing board with several people, gives `createCompressor` its own `system` prompt and names each speaker in the text, like "[Kari] …". diff --git a/src/core/images.ts b/src/core/images.ts index ac1a53b7..d179219e 100644 --- a/src/core/images.ts +++ b/src/core/images.ts @@ -3,6 +3,11 @@ import type { Store } from './store.ts'; /** How a message's image reads in memory: `[image <16 hex digits>]`, named by the host (Pi: a hash of the image as sent). */ const REF = /\[image ([0-9a-f]{16})\]/g; +const ONLY_REF = new RegExp(`^${REF.source}$`); +/** The ref that names an image in memory's text. */ +export const ref = (name: string) => `[image ${name}]`; +/** The image name a text is the ref of, when it is that ref and nothing else. */ +export const refName = (text: string) => ONLY_REF.exec(text)?.[1]; export const isImage = (part: unknown): part is ImageContent => typeof part === 'object' && part !== null && 'type' in part && part.type === 'image' && 'data' in part && typeof part.data === 'string' && 'mimeType' in part && typeof part.mimeType === 'string'; diff --git a/src/core/index.ts b/src/core/index.ts index 4b2ee55f..a47954bc 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -1,13 +1,13 @@ /** OptChat's memory for any host: the log, the summary tree and its views, zoom/date/search, the compactor, cache marks and * memory's prompts. It imports nothing from Pi or Node at runtime: the host gives it a `Store` and the compactor's model. * Only what a host needs is exported here; Pi's own code imports the modules directly. */ -export { CAP, isEntry, Memory, openMemory, Refused, VIEW, type Compression, type Compressor, type Entry, type Kind, type MemoryOptions, type Origin, type Part, type Summary } from './memory.ts'; +export { isEntry, Memory, openMemory, Refused, VIEW, type Compression, type Compressor, type Entry, type Kind, type MemoryOptions, type Origin, type Part, type Summary } from './memory.ts'; export { isMime, type ImageStore, type Lease, type Mime, type Saved, type Store } from './store.ts'; export { ACCEPTED_BYTES, createCompressor, type CompactorModel, type CompressorOptions, type Retryable } from './compactor.ts'; export { cachePayload } from './cache.ts'; export { allowSearch, CHILD, COMPACT, CONTINUITY, SEARCH_DOC, TURN_REFRESHED, VIEW_DOC } from './prompts.ts'; export { boundedMessage, childInput, completedExchange, logMessage, refreshBudget, refreshedRun, refreshPoint, textContent, withView, type AnyMessage, type ImageRef } from './messages.ts'; export { memoryTools, searchTool } from './tools.ts'; -export { decode, isImage, loadImages } from './images.ts'; +export { decode, isImage, loadImages, ref, refName } from './images.ts'; export type { ImageContent } from '@earendil-works/pi-ai'; export { chronological, deduplicate, planImport, runImport, type ImportedEntry, type ImportMode } from './import.ts'; diff --git a/src/durable/images.ts b/src/durable/images.ts new file mode 100644 index 00000000..47f4ceb0 --- /dev/null +++ b/src/durable/images.ts @@ -0,0 +1,85 @@ +import type { GenerationHooks, ToolExecutionResult } from '@earendil-works/pi-durable'; +import { decode, isImage, isMime, loadImages, ref, refName, type AnyMessage, type ImageContent, type ImageRef, type Store } from '../core/index.ts'; + +export type Message = Parameters[0]['messages'][number]; +type Block = NonNullable[number]; +type Text = Extract; +const encoder = new TextEncoder(); + +/** A text block that is only an image's ref: what an image kept in memory's store leaves in Pi's transcript. */ +const isRef = (block: Block): block is Text => block.type === 'text' && refName(block.text) !== undefined; + +/** The images memory names, as Pi keeps them: a user's and a tool's, but not zoom's, which are kept already. */ +export const keptImages = (message: AnyMessage) => message.role === 'user' || message.role === 'toolResult' && !('toolName' in message && message.toolName === 'zoom') + ? Array.isArray(message.content) ? message.content.filter(isImage) : [] : []; +/** A zoom's page names the images it returns already, so memory leaves out the refs its images leave behind. */ +export const dropZoomRefs = (m: Message) => m.role === 'toolResult' && m.toolName === 'zoom' ? { ...m, content: m.content.filter(block => !isRef(block)) } : m; + +/** Each distinct image, named as Pi's OptChat names it: the first 16 hex digits of the SHA-256 of its base64; and the ImageRef that + * names them. Web Crypto is asynchronous and core's ImageRef is not, so images are named before core reads the names. */ +export async function imageNames(images: readonly ImageContent[]) { + const distinct = [...new Map(images.map(image => [image.data, image])).values()]; + const named = await Promise.all(distinct.map(async image => { + const hash = new Uint8Array(await crypto.subtle.digest('SHA-256', encoder.encode(image.data)), 0, 8); + return { image, name: Array.from(hash, byte => byte.toString(16).padStart(2, '0')).join('') }; + })); + const names = new Map(named.map(({ image, name }) => [image.data, name])); + const imageRef: ImageRef = image => { + const name = names.get(image.data); + if (name === undefined) throw new Error('OptChat was asked to name an image it was not given.'); + return ref(name); + }; + return { named, imageRef }; +} + +/** Puts an image in the store; false, and logged, when the store refuses it or its type. */ +export async function keep(store: Store, image: ImageContent, name: string) { + if (!isMime(image.mimeType)) return false; + try { + await store.putImage(name, image.mimeType, decode(image.data)); + return true; + } catch (error) { + console.error(`OptChat could not keep an image: ${error instanceof Error ? error.message : String(error)}`); + return false; + } +} + +/** `content` with each image the store keeps replaced by its ref; one it refuses stays. `put: false` for images it has already, + * such as zoom's. */ +export async function offload(store: Store, content: readonly Block[], put = true): Promise { + const { named, imageRef } = await imageNames(content.filter(isImage)); + const kept = await Promise.all(named.map(async ({ image, name }) => !put || await keep(store, image, name) ? [image.data] : [])); + const offloaded = new Set(kept.flat()); + return content.map(block => isImage(block) && offloaded.has(block.data) ? { type: 'text', text: imageRef(block) } : block); +} + +/** Pi Durable's default bound on a tool result's text, its DEFAULT_MAX_BYTES and DEFAULT_MAX_LINES, which it doesn't export. A + * result over the tool's bound keeps one of its text blocks, so its refs would go. A hook can't see a tool's own `outputLimits`. */ +const BOUND = { bytes: 50 * 1024, lines: 2_000 }; +const REF_BYTES = ref('0'.repeat(16)).length; +/** Whether a tool result's text stays inside Pi Durable's default bound once its images are refs. Pi joins its text blocks with + * nothing between them, and a ref has no newline. */ +export function fitsBound(output: readonly Block[]) { + const text = output.map(block => block.type === 'text' ? block.text : '').join(''); + return encoder.encode(text).length + output.filter(isImage).length * REF_BYTES <= BOUND.bytes && text.split('\n').length <= BOUND.lines; +} + +/** `messages` with each ref in a user message or tool result replaced by its image from the store, each read once. A ref the + * store lacks or can't read stays as it is. */ +export async function restore(store: Store, messages: readonly Message[]) { + const reads = new Map>(); + const read = (block: Text) => { + const pending = reads.get(block.text); + if (pending) return pending; + const image = loadImages(store, block.text).then(([found]) => found ?? block, error => { + console.error(`OptChat could not read an image: ${error instanceof Error ? error.message : String(error)}`); + return block; + }); + reads.set(block.text, image); + return image; + }; + return Promise.all(messages.map(async m => { + if ((m.role !== 'user' && m.role !== 'toolResult') || typeof m.content === 'string') return m; + return { ...m, content: await Promise.all(m.content.map(block => isRef(block) ? read(block) : block)) }; + })); +} diff --git a/src/durable/optchat.ts b/src/durable/optchat.ts index 5573152b..b4576f9c 100644 --- a/src/durable/optchat.ts +++ b/src/durable/optchat.ts @@ -5,28 +5,20 @@ import { } from '@earendil-works/pi-durable'; import type { Static, TSchema } from 'typebox'; import { - allowSearch, boundedMessage, cachePayload, CAP, CHILD, childInput, completedExchange, CONTINUITY, decode, isImage, isMime, loadImages, logMessage, memoryTools, refreshBudget, refreshedRun, - refreshPoint, SEARCH_DOC, searchTool, textContent, TURN_REFRESHED, VIEW_DOC, withView, type AnyMessage, type ImageContent, type Memory, + allowSearch, boundedMessage, cachePayload, CHILD, childInput, completedExchange, CONTINUITY, isImage, logMessage, memoryTools, refreshBudget, refreshedRun, + refreshPoint, SEARCH_DOC, searchTool, textContent, TURN_REFRESHED, VIEW_DOC, withView, type Memory, } from '../core/index.ts'; +import { dropZoomRefs, fitsBound, imageNames, keep, keptImages, offload, restore, type Message } from './images.ts'; export type Context = Parameters[1]; -type Message = Parameters[0]['messages'][number]; type Models = Exclude unknown>; -type Block = NonNullable[number]; -type TextContent = Extract; /** The entries that are the chat: other kinds (system prompt changes, resets, compactions, other extensions' own) stay out of memory. */ const CHAT = new Set([UserEntry.kind, AssistantEntry.kind, ToolResultEntry.kind]); const chat = (entry: EntryRecord) => CHAT.has(entry.kind) ? entry.model ?? [] : []; -/** A text block that is only an image's name: what an image offloaded to memory's store leaves in Pi's transcript. */ -const MARKER = /^\[image [0-9a-f]{16}\]$/; -const isMarker = (block: Block): block is TextContent => block.type === 'text' && MARKER.test(block.text); -/** A zoom's page names the images it returns already, so memory leaves out the names its offloaded images leave behind. */ -const unmarkZoom = (m: Message) => m.role === 'toolResult' && m.toolName === 'zoom' ? { ...m, content: m.content.filter(block => !isMarker(block)) } : m; -const failure = (error: unknown) => error instanceof Error ? error.message : String(error); /** What memory logs of an entry: its chat, or a reset's note (`Conversation.reset(text)`, a tool's handoff) marked as one. Pi sends * the note in the context it starts, which is logged only once it ends, so the model never sees it twice. */ -const remembered = (entry: EntryRecord) => !ResetEntry.is(entry) ? chat(entry).map(unmarkZoom) +const remembered = (entry: EntryRecord) => !ResetEntry.is(entry) ? chat(entry).map(dropZoomRefs) : (entry.model ?? []).map(note => ({ role: 'user' as const, content: `[Handoff note] ${textContent(note.content)}`, timestamp: note.timestamp })); /** Each message OptChat logs from a Pi entry carries the entry's id, so the log itself says how far it has read Pi's transcript. */ const RECEIPT = 'pi:'; @@ -34,19 +26,6 @@ const RECEIPT = 'pi:'; const endsRun = (m: Message) => m.role === 'assistant' && m.stopReason !== 'toolUse'; const encoder = new TextEncoder(); -/** The images memory names, as Pi keeps them: a user's and a tool's, but not zoom's, which are kept already. */ -const keptImages = (message: AnyMessage) => message.role === 'user' || message.role === 'toolResult' && !('toolName' in message && message.toolName === 'zoom') - ? Array.isArray(message.content) ? message.content.filter(isImage) : [] : []; -/** Each distinct image by its base64, named as Pi's OptChat names it: the first 16 hex digits of the SHA-256 of its base64. Web - * Crypto is asynchronous and core's ImageRef is not, so images are named before core reads the names. */ -async function imageNames(images: readonly ImageContent[]) { - const distinct = new Map(images.map(image => [image.data, image])); - return new Map(await Promise.all([...distinct.values()].map(async image => { - const hash = new Uint8Array(await crypto.subtle.digest('SHA-256', encoder.encode(image.data)), 0, 8); - return [image.data, { image, name: Array.from(hash, byte => byte.toString(16).padStart(2, '0')).join('') }] as const; - }))); -} - /** One of core's memory tools as a Pi Durable tool, its arguments typed by its own schema. */ const durableTool =

(tool: { name: string; description: string; parameters: P; @@ -109,38 +88,11 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp } }; - /** `content` with each image kept in memory's store and replaced by its name, as memory names it; one the store refuses stays. */ - const offload = async (content: readonly Block[]) => { - const names = await imageNames(content.filter(isImage)), kept = new Set(); - await Promise.all([...names.values()].map(async ({ image, name }) => { - if (!isMime(image.mimeType)) return; - try { await memory.store.putImage(name, image.mimeType, decode(image.data)); kept.add(image.data); } - catch (error) { console.error(`OptChat could not keep an image: ${failure(error)}`); } - })); - return content.map(block => isImage(block) && kept.has(block.data) ? { type: 'text' as const, text: `[image ${names.get(block.data)?.name}]` } : block); - }; - /** A model-issued tool's images go to the store, so Pi's transcript holds their names. A result whose text would pass CAP keeps - * them: Pi Durable bounds a result's text by keeping one of its text blocks, and a text over CAP is sent as its head and tail, - * either of which can drop a name. */ + /** A model-issued tool's images go to the store, so Pi's transcript keeps their refs, unless its text could pass Pi Durable's + * bound (images.ts). Zoom's images are in the store already. */ const afterTool: ToolHooks['afterTool'] = async (call, result, api) => { - if (api.conversationId !== id || call.parent || !result.output?.some(isImage)) return undefined; - const output = await offload(result.output); - return encoder.encode(textContent(output)).length <= CAP ? { ...result, output } : undefined; - }; - /** The request's messages with each image name in a user message or tool result replaced by the image, read from the store once - * per request. A name the store lacks or can't read stays as it is. */ - const restore = async (messages: readonly Message[]) => { - const loads = new Map>(); - const load = (block: TextContent) => { - let image = loads.get(block.text); - if (!image) loads.set(block.text, image = loadImages(memory.store, block.text).then(([found]) => found ?? block, error => { - console.error(`OptChat could not read an image: ${failure(error)}`); - return block; - })); - return image; - }; - return Promise.all(messages.map(async m => m.role !== 'user' && m.role !== 'toolResult' || typeof m.content === 'string' || !m.content.some(isMarker) ? m - : { ...m, content: await Promise.all(m.content.map(block => isMarker(block) ? load(block) : block)) })); + if (api.conversationId !== id || call.parent || !result.output?.some(isImage) || !fitsBound(result.output)) return undefined; + return { ...result, output: await offload(memory.store, result.output, call.name !== 'zoom') }; }; const handle = async (ctx: Context): Promise => { @@ -157,17 +109,14 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp if (logged + 1 > head - 1) return; const page = await conversation.entries({ order: 'ascending', minEntryId: logged + 1 as EntryId, maxEntryId: head - 1 as EntryId }, 256, cursor, ctx); for (const entry of page.items) { - const messages = remembered(entry), names = await imageNames(messages.flatMap(keptImages)); + const messages = remembered(entry), { named, imageRef } = await imageNames(messages.flatMap(keptImages)); // Kept before the message is logged, so the image its text names is there for zoom; one that can't be kept is only named. // An image sent again, or read again after a restart, is written again under its name. - for (const { image, name } of names.values()) if (isMime(image.mimeType)) { - try { await memory.store.putImage(name, image.mimeType, decode(image.data)); } - catch (error) { console.error(`OptChat could not keep an image: ${error instanceof Error ? error.message : String(error)}`); } - } + for (const { image, name } of named) await keep(memory.store, image, name); // Once a save failed (also one queued before, failing during the awaits above), memory refuses appends: requests keep the // view it has, and a restart logs the rest from Pi's transcript. if (memory.saveError) return; - for (const message of messages) logMessage(resumed, message, { receipt: `${RECEIPT}${entry.id}`, imageRef: image => `[image ${names.get(image.data)?.name}]` }); + for (const message of messages) logMessage(resumed, message, { receipt: `${RECEIPT}${entry.id}`, imageRef }); logged = entry.id; } cursor = page.next; @@ -185,8 +134,8 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp if (!last) return []; const messages = (await conversation.context(ctx, { at: last.id })).entries.flatMap(chat); const start = messages.slice(0, messages.findLastIndex(m => m.role === 'user')).findLastIndex(endsRun) + 1; - const run = messages.slice(start), names = await imageNames(run.filter(m => m.role === 'user').flatMap(keptImages)); - const exchange = completedExchange(run, image => `[image ${names.get(image.data)?.name}]`); + const run = messages.slice(start), { imageRef } = await imageNames(run.filter(m => m.role === 'user').flatMap(keptImages)); + const exchange = completedExchange(run, imageRef); const bytes = exchange.reduce((sum, m) => sum + encoder.encode(textContent(m.content)).length, 0); return bytes <= previousExchangeBytes ? exchange : []; }; @@ -254,7 +203,7 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp }; const beforeRequest: GenerationHooks['beforeRequest'] = async (request, api, ctx) => { if (api.conversationId !== id) return undefined; - const messages = await restore(request.messages); + const messages = await restore(memory.store, request.messages); const { system, previous, run, view } = await compose(ctx, { messages, models: api.models }); return system.length < messages.length ? { messages: withView([...system, ...previous, ...run], view) } : undefined; }; @@ -326,7 +275,7 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp /** `input` with its images kept in memory's store, each replaced by its name, as tool results' are: pass a user input through it * before submitting it, so Pi's transcript holds no image bytes; requests get the images back. An image the store refuses stays. */ async offloadImages(input: UserInput): Promise { - return typeof input === 'string' ? input : offload(input); + return typeof input === 'string' ? input : offload(memory.store, input); }, /** Starts following the conversation, and resets it now if a run ended before a restart. Returns a function that stops. */ attach(harness: Harness, ctx: Context) { diff --git a/src/images.ts b/src/images.ts index f00a980c..ab467b34 100644 --- a/src/images.ts +++ b/src/images.ts @@ -1,7 +1,7 @@ import { createHash } from 'node:crypto'; import type { ImageContent } from '@earendil-works/pi-ai'; import { resizeImage } from '@earendil-works/pi-coding-agent'; -import { isImage } from './core/images.ts'; +import { isImage, ref } from './core/images.ts'; import { isMime, type Store } from './core/store.ts'; /** Kept images fit 2048 px on the long side and ~1.5 MB (2 MB as base64), re-encoded as JPEG 85 only when they don't already. */ @@ -9,7 +9,7 @@ export const IMAGE_LIMITS = { maxWidth: 2048, maxHeight: 2048, maxBytes: 2 * 102 const hash = (image: ImageContent) => createHash('sha256').update(image.data).digest('hex').slice(0, 16); /** How a message's image reads in memory: named by the image as the message holds it, before any shrinking, so it always gets the same name. */ -export const imageRef = (image: ImageContent) => `[image ${hash(image)}]`; +export const imageRef = (image: ImageContent) => ref(hash(image)); /** Keeps each image of a message's content in the store, once per distinct image, shrunk to IMAGE_LIMITS. */ export async function saveImages(store: Store, content: unknown) { diff --git a/test/durable.test.ts b/test/durable.test.ts index 964441b6..bf031835 100644 --- a/test/durable.test.ts +++ b/test/durable.test.ts @@ -12,7 +12,7 @@ import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context'; import { NodeSqliteDatabase } from '@earendil-works/chord/sqlite/node'; import { createModels } from '@earendil-works/pi-ai/models'; import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall, type FauxResponseStep } from '@earendil-works/pi-ai/providers/faux'; -import { CHILD, cachePayload, isImage, openMemory, textContent, type Compressor, type ImageStore, type ImportedEntry, type Mime } from '../src/core/index.ts'; +import { CHILD, cachePayload, isImage, openMemory, refName, textContent, type Compressor, type ImageStore, type ImportedEntry, type Mime } from '../src/core/index.ts'; import { optchat, optchatImport, SqliteStore, withCacheMarks, type OptChatOptions, type Read, type SqliteStoreOptions } from '../src/durable/index.ts'; import { importTables } from '../src/durable/store.ts'; import { imageRef } from '../src/images.ts'; @@ -267,39 +267,46 @@ test('with an images store, images go there, and zoom reads them from it', async } finally { await agent.close(); } }); -/** A tool that returns `images` after a line of text. */ -const camera = (...images: (typeof photo)[]): Extension => defineExtension({ name: 'camera', tools: [defineTool({ +/** A tool that returns `images` after `text`. */ +const camera = (images: (typeof photo)[], text = 'snap'): Extension => defineExtension({ name: 'camera', tools: [defineTool({ name: 'camera', description: 'Takes a photo.', parameters: Type.Object({}), replay: 'safe', - execute: async () => ({ output: [{ type: 'text', text: 'snap' }, ...images] }), + execute: async () => ({ output: [{ type: 'text', text }, ...images] }), })] }); +const snap = () => [fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' }), fauxAssistantMessage('nice')]; /** Whether any of Pi Durable's rows holds `data`. */ const inPiRows = (agent: Awaited>, data: string) => agent.sql.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name LIKE 'pi\\_%' ESCAPE '\\'").all() .some(({ name }) => agent.sql.prepare(`SELECT * FROM "${String(name)}"`).all().some(row => JSON.stringify(row).includes(data))); const imagesIn = (request: Sent | undefined, role: string) => (request ?? []).flatMap(m => m.role === role && Array.isArray(m.content) ? m.content.filter(isImage) : []); +const contentOf = (request: Sent | undefined, role: string) => request?.find(m => m.role === role)?.content; +const nameOf = (image: typeof photo) => refName(imageRef(image)) ?? ''; const memoryImages = () => { const kept = new Map(); const images: ImageStore = { putImage: async (name, mimeType, data) => void kept.set(name, { mimeType, data }), image: name => kept.get(name) }; return { kept, images }; }; -test('a tool\'s images and a user\'s sent through offloadImages are kept once, in the images store: Pi\'s rows hold their names, requests the images', async () => { - const { kept, images } = memoryImages(), f = file(); - const user = { ...photo, data: Buffer.alloc(1000, 9).toString('base64') }; - const agent = await open(f, [ - fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' }), fauxAssistantMessage('nice'), - ], undefined, { images, extensions: () => [camera(photo)] }); +test('a tool\'s images and a user\'s sent through offloadImages are kept once, in the images store: Pi\'s rows hold their refs, requests the images', async () => { + const { kept, images } = memoryImages(), user = { ...photo, data: Buffer.alloc(1000, 9).toString('base64') }; + const agent = await open(file(), snap(), undefined, { images, extensions: () => [camera([photo])] }); try { - await (await agent.root.submit({ type: 'input', content: await agent.chat.offloadImages([{ type: 'text', text: 'take one like this' }, user]) }, context)).wait(context); + const content = await agent.chat.offloadImages([{ type: 'text', text: 'take one like this' }, user]); + assert.deepEqual(content, [{ type: 'text', text: 'take one like this' }, { type: 'text', text: imageRef(user) }]); + await (await agent.root.submit({ type: 'input', content }, context)).wait(context); await agent.idle(); assert.ok(!inPiRows(agent, photo.data) && !inPiRows(agent, user.data)); assert.deepEqual(imagesIn(agent.sent[0], 'user'), [user]); assert.deepEqual(imagesIn(agent.sent[1], 'user'), [user]); - assert.deepEqual(imagesIn(agent.sent[1], 'toolResult'), [photo]); + assert.deepEqual(contentOf(agent.sent[1], 'toolResult'), [{ type: 'text', text: 'snap' }, photo]); // Memory names them as it names inline images. assert.deepEqual(agent.memory.entries().map(e => e.text), [`take one like this\n${imageRef(user)}`, 'camera {}', `camera: snap\n${imageRef(photo)}`, 'nice']); - assert.equal(kept.size, 2); + assert.deepEqual([...kept.keys()].sort(), [nameOf(user), nameOf(photo)].sort()); } finally { await agent.close(); } - // After a restart, zoom returns the tool's image; its own result is offloaded too, and memory logs zoom's page as it is. +}); + +test('after a restart, zoom returns a tool\'s kept image; its own result keeps only refs, and memory logs zoom\'s page as it is', async () => { + const { images } = memoryImages(), f = file(); + const agent = await open(f, snap(), undefined, { images, extensions: () => [camera([photo])] }); + try { await agent.ask('take one'); } finally { await agent.close(); } const again = await open(f, [fauxAssistantMessage(fauxToolCall('zoom', { id: 2 }), { stopReason: 'toolUse' }), fauxAssistantMessage('here')], undefined, { images }); try { await again.ask('show it again'); @@ -311,16 +318,34 @@ test('a tool\'s images and a user\'s sent through offloadImages are kept once, i test('an image the store refuses stays inline in Pi\'s row and still reaches the model', async (t) => { t.mock.method(console, 'error', () => {}); - const { kept, images } = memoryImages(), refused = { ...photo, data: Buffer.alloc(1000, 8).toString('base64') }; - const putImage = images.putImage; - images.putImage = (name, mimeType, data) => name === imageRef(refused).slice(7, -1) ? Promise.reject(new Error('full')) : putImage(name, mimeType, data); - const agent = await open(file(), [fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' }), fauxAssistantMessage('nice')], - undefined, { images, extensions: () => [camera(photo, refused)] }); + const { kept, images } = memoryImages(), refused = { ...photo, data: Buffer.alloc(1000, 8).toString('base64') }, putImage = images.putImage; + images.putImage = (name, mimeType, data) => name === nameOf(refused) ? Promise.reject(new Error('full')) : putImage(name, mimeType, data); + const agent = await open(file(), snap(), undefined, { images, extensions: () => [camera([photo, refused])] }); try { await agent.ask('take two'); assert.ok(!inPiRows(agent, photo.data) && inPiRows(agent, refused.data)); assert.deepEqual(imagesIn(agent.sent[1], 'toolResult'), [photo, refused]); - assert.equal(kept.size, 1); + assert.deepEqual([...kept.keys()], [nameOf(photo)]); + } finally { await agent.close(); } +}); + +test('a tool result that could pass Pi Durable\'s bound keeps its images inline: bounding it would drop their refs', async () => { + const { images } = memoryImages(), lines = Array.from({ length: 2_500 }, (_, i) => `line ${i}`).join('\n'); + const agent = await open(file(), snap(), undefined, { images, extensions: () => [camera([photo], lines)] }); + try { + await agent.ask('take one'); + assert.deepEqual(imagesIn(agent.sent[1], 'toolResult'), [photo]); + } finally { await agent.close(); } +}); + +test('a ref the store lacks or can\'t read is sent as its text, and the run goes on', async (t) => { + t.mock.method(console, 'error', () => {}); + const missing = { ...photo, data: Buffer.alloc(1000, 5).toString('base64') }; + const images: ImageStore = { putImage: async () => {}, image: name => { if (name === nameOf(photo)) throw new Error('down'); return undefined; } }; + const agent = await open(file(), snap(), undefined, { images, extensions: () => [camera([photo, missing])] }); + try { + assert.equal((await agent.ask('take two')).status, 'done'); + assert.deepEqual(contentOf(agent.sent[1], 'toolResult'), [{ type: 'text', text: 'snap' }, { type: 'text', text: imageRef(photo) }, { type: 'text', text: imageRef(missing) }]); } finally { await agent.close(); } }); From d1e6bb52fb326474c2995dbee3266af33d946e5e Mon Sep 17 00:00:00 2001 From: Jonas Silva <7734262+jonaslsaa@users.noreply.github.com> Date: Tue, 3 Nov 2026 02:21:06 +0100 Subject: [PATCH 3/3] Restore images only into what a request sends; put zoom's images again; one ref format with the compactor Images go back after the request is built, so steps a refresh dropped are never read; a ref counts as an image in the refresh estimate and survives boundedMessage's cap. Zoom's images are put again, since the store gives them back as canonical base64 under another name. The compactor labels images with core's ref(). Docs: core's ref helpers, the compactor's image cost, beforeRequest hook order. --- docs/core.md | 4 ++-- docs/durable.md | 4 ++-- src/core/compactor.ts | 3 ++- src/core/images.ts | 11 ++++++++--- src/core/index.ts | 2 +- src/core/messages.ts | 15 +++++++++------ src/durable/images.ts | 24 ++++++++++++++---------- src/durable/optchat.ts | 10 +++++----- test/durable.test.ts | 32 ++++++++++++++++++++++++++++++++ 9 files changed, 75 insertions(+), 30 deletions(-) diff --git a/docs/core.md b/docs/core.md index e08225db..0f8aac5f 100644 --- a/docs/core.md +++ b/docs/core.md @@ -10,7 +10,7 @@ Core imports nothing from Pi or Node at runtime: only `typebox` (the tools' sche - **A `Store`:** `load`, `append`, `appendNode`, `saveView`, `putImage` and `image`, sync or async, plus `text` and `summary`, which must be synchronous. Memory loads only metadata when it opens, reads texts through `text`/`summary` as it needs them, and issues its writes in order. Pi's is `FileStore` (`src/store.ts`); Pi Durable's is `SqliteStore` (`pi-optchat/durable`, see [durable.md](durable.md)). - **The compactor's model:** `createCompressor({ model, retryable })`, where `model()` returns `{ model, reasoning, stream }` for each call (pi-ai's `Models.streamSimple` or one with its shape) and `retryable` is pi-ai's `isRetryableAssistantError`. -- **Image names**, if it keeps images: `logMessage` and `textContent` take an `imageRef` that names an image `[image <16 hex digits>]`; `loadImages` reads those back from the store. +- **Image names**, if it keeps images: `logMessage` and `textContent` take an `imageRef` that names an image by its ref, `[image <16 hex digits>]`: `ref(name)` builds one and `refName(text)` reads the name back from a text that is only a ref. `loadImage(store, name)` and `loadImages(store, text)` read the images back from the store. ## What it exposes @@ -30,7 +30,7 @@ const [zoom, date] = memoryTools(() => memory); // VIEW_DOC tells the model h - `Store`, `Saved`, `Lease`, `ImageStore` (a store's `putImage` and `image`, for keeping images elsewhere), `Mime`, `isMime`, and the record types `Entry`, `Summary`, `Origin`, `Kind`. An entry with an `origin` is imported history, which the summarizer treats as historical; `origin.source` is any label the host picks ("email"). - `createCompressor({ model, retryable, onUsage?, accepted?, retryDelays?, system? })` (`system`: the compactor's prompt, `COMPACT` by default), `ACCEPTED_BYTES`. - `logMessage(memory, message, { kind?, receipt?, imageRef? })` (every entry it appends gets `receipt`, so a host can use it as a cursor), `textContent`, `withView`, `boundedMessage`, `completedExchange`, `childInput({ view, previous?, run?, task })` (a child conversation's first input, as Pi's subagents and durable `snapshot` build it), and the mid-turn refresh helpers (`refreshBudget`, `refreshPoint`, `refreshedRun`). -- `memoryTools(memory, runs?)` (zoom and date; `runs` adds zooming into a subagent's chat), `searchTool`, `loadImages`, `isImage`, `decode`. +- `memoryTools(memory, runs?)` (zoom and date; `runs` adds zooming into a subagent's chat), `searchTool`, `ref`, `refName`, `loadImage`, `loadImages`, `isImage`, `decode`. - `cachePayload` (Anthropic: the view's cache marks, as a request's `onPayload`). - Imports: `planImport(existing, incoming, 'append' | 'rebuild', copyOf?)` dedupes by `receipt` (each starts with `import:`) and orders the new history by date: append keeps the existing log as the plan's first `kept` entries, rebuild orders everything and summarizes it again. `runImport({ entry, total, open, activate }, signal)` builds the memory `open` gives from the plan's `total` entries, read one at a time through `entry(i)` so the plan need not stay in RAM, resuming where it stopped, and calls `activate` once the whole tree is built and the memory is closed. The host keeps the plan, seeds the staging store (append: the kept entries, the summaries and the saved view), and switches to it in `activate`. Until then the live memory must take no messages, or the switch drops them: queue them and log them after. `deduplicate` alone tells what would be new, and `chronological` is the order rebuild uses. `pi-optchat/durable`'s `optchatImport` does all this on SQLite. - Prompts: `COMPACT`, `VIEW_DOC`, `CHILD`, `CONTINUITY`, `SEARCH_DOC`, `allowSearch`, `TURN_REFRESHED`. diff --git a/docs/durable.md b/docs/durable.md index c1859621..cd89885a 100644 --- a/docs/durable.md +++ b/docs/durable.md @@ -61,8 +61,8 @@ Keep `optchat` selected in the child, as its copy of the owner's agent does. It - **A failed save** stops memory's log: requests keep the view it has, and after a restart logging resumes from the receipts and Pi's transcript. Cloudflare resets the object itself only when a commit fails after the code ran (the output gate, which drops Pi's writes too); a write that throws at once, like `SQLITE_FULL` at the storage limit, doesn't, so the restart waits for idle eviction, a deploy, or the host's `ctx.abort()`, which the example calls from `warn`. `SQLITE_FULL` stops Pi's writes as well; the errors only memory's writes could hit (a row over 2 MB, a message index saved twice) are ruled out by the store's parts and by one Memory per database. - **Compaction:** turn Pi's off; the extension also declines any that starts. OptChat's summaries replace it. - **Large tool results** are sent as their head and tail (`boundedMessage`), capped as Pi does. -- **Images:** each image in a user message or tool result is named in memory's text as Pi's OptChat names it, `[image <16 hex digits>]` (SHA-256 of its base64, by Web Crypto), and kept in the store before the message is logged, so `zoom(id, 1)` returns it and the previous exchange names it instead of sending it again. OptChat doesn't resize images: Pi Durable's Photon processor (`createCloudflarePhotonImages()`) runs in Workers, but the memory it decodes in never shrinks again. Shrink photos when they arrive instead (Cloudflare Images, say), so the store and the model get the small copy too. `SqliteStore` keeps images in `optchat_images`; to keep large binary data out of the database (its row size limit and storage cost), give it `{ images }`, an `ImageStore` with `putImage` and `image`. The example keeps them in R2. -- **Images in Pi's transcript:** Pi Durable keeps each transcript entry, base64 images included, in one row. So each image goes to memory's store once, and Pi's row keeps only its ref as a text block, `[image <16 hex digits>]`. A model-issued tool's images go there by themselves (`afterTool`); a user's go when the app passes the input through `offloadImages` before submitting it: `root.submit({ type: 'input', content: await memoryExtension.offloadImages(content) }, ctx)`. Each request gets the images back (`beforeRequest`), each read from the store once per request; memory's text is the same either way. `putImage` must keep the image or throw: one it refuses stays inline. So do a nested call's images, an input the app didn't pass through `offloadImages`, and a tool result whose text could pass Pi Durable's default bound with the refs (50 KB or 2,000 lines): Pi bounds a result by keeping one of its text blocks, which would drop the refs. The hook can't see a tool's own `outputLimits`, so a tool with lower limits must stay inside them to keep its images. On Cloudflare, a message's inline images must stay well under 2 MB. A ref the store lacks or can't read is sent as its text. Hooks run in the order extensions are selected (install order by default): a host's own `beforeRequest` or `afterTool` before OptChat's sees images, one after sees refs. A host that reads Pi's transcript itself, such as a chat UI, sees the refs too: get an image with `memory.store.image(name)`. +- **Images:** each image in a user message or tool result is named in memory's text as Pi's OptChat names it, `[image <16 hex digits>]` (SHA-256 of its base64, by Web Crypto), and kept in the store before the message is logged, so `zoom(id, 1)` returns it and the previous exchange names it instead of sending it again. OptChat doesn't resize images: Pi Durable's Photon processor (`createCloudflarePhotonImages()`) runs in Workers, but the memory it decodes in never shrinks again. Shrink photos when they arrive instead (Cloudflare Images, say), so the store and the model get the small copy too. Each image is also sent once to the compactor for its message's own summary (about 1¢ each), so shrinking on arrival helps there too; one the provider rejects is summarized by name. `SqliteStore` keeps images in `optchat_images`; to keep large binary data out of the database (its row size limit and storage cost), give it `{ images }`, an `ImageStore` with `putImage` and `image`. The example keeps them in R2. +- **Images in Pi's transcript:** Pi Durable keeps each transcript entry, base64 images included, in one row. So each image goes to memory's store once, and Pi's row keeps only its ref as a text block, `[image <16 hex digits>]`. A model-issued tool's images go there by themselves (`afterTool`); a user's go when the app passes the input through `offloadImages` before submitting it: `root.submit({ type: 'input', content: await memoryExtension.offloadImages(content) }, ctx)`. Each request gets the images back (`beforeRequest`), each read from the store once per request; memory's text is the same either way. `putImage` must keep the image or throw: one it refuses stays inline. So do a nested call's images, an input the app didn't pass through `offloadImages`, and a tool result whose text could pass Pi Durable's default bound with the refs (50 KB or 2,000 lines): Pi bounds a result by keeping one of its text blocks, which would drop the refs. The hook can't see a tool's own `outputLimits`, so a tool with lower limits must stay inside them to keep its images. On Cloudflare, a message's inline images must stay well under 2 MB. A ref the store lacks or can't read is sent as its text. Hooks run in the order extensions are selected (install order by default): a host's own `afterTool` before OptChat's sees images and one after sees refs; a host's own `beforeRequest` before OptChat's sees refs and one after sees images. A host that reads Pi's transcript itself, such as a chat UI, sees the refs too: get an image with `memory.store.image(name)`. - **Tools:** `zoom` and `date` (plus `search` with `search: true`), replay-safe. - **Cache marks:** `withCacheMarks(models)` adds OptChat's marks to Anthropic requests that carry a view, so a view that grows at its end is read from the cache. A request with its own `onPayload` is left alone. - **Summaries of a group chat:** a chat that isn't one user talking to OptChat, such as a housing board with several people, gives `createCompressor` its own `system` prompt and names each speaker in the text, like "[Kari] …". diff --git a/src/core/compactor.ts b/src/core/compactor.ts index 56d479ed..80b2cd86 100644 --- a/src/core/compactor.ts +++ b/src/core/compactor.ts @@ -2,6 +2,7 @@ import type { Api, AssistantMessage, Message, Model, Models, SimpleStreamOptions import { COMPACT, compaction, IMPORT_GUIDANCE, tooLong } from './prompts.ts'; import { bytes, cut, NODE, Refused, type Compressor } from './memory.ts'; import { cachePayload, splitView } from './cache.ts'; +import { ref } from './images.ts'; /** The longest line kept without asking again, unless the host sets its own (Pi's "Summary size tolerance"). */ export const ACCEPTED_BYTES = NODE * 1.25; @@ -83,7 +84,7 @@ export function createCompressor({ model: resolve, retryable, onUsage = () => {} // OpenAI joins text blocks with nothing between them, so the task starts its own paragraph after the view. { type: 'text', text: input.context }, { type: 'text', text: `\n\n${input.historical ? IMPORT_GUIDANCE + '\n\n' : ''}${compaction({ ...input, images })}` }, - ...images.flatMap(({ name, image }) => [{ type: 'text' as const, text: `\n[image ${name}]` }, image])] }); + ...images.flatMap(({ name, image }) => [{ type: 'text' as const, text: `\n${ref(name)}` }, image])] }); let messages: Message[] = [task()]; const view = splitView(input.context); const prefix = model.api === 'anthropic-messages' && view.length > 1 ? `${model.provider}/${model.id}/${thinking ?? 'off'}\n${view.slice(0, -1).join('')}` : undefined; diff --git a/src/core/images.ts b/src/core/images.ts index d179219e..2b8a1762 100644 --- a/src/core/images.ts +++ b/src/core/images.ts @@ -21,12 +21,17 @@ const base64 = (data: Uint8Array) => { /** An image's base64 as bytes. */ export const decode = (data: string) => Uint8Array.from(atob(data), char => char.charCodeAt(0)); +/** The image kept under `name`; none if it never was. */ +export async function loadImage(store: Store, name: string): Promise { + const kept = await store.image(name); + return kept && { type: 'image', data: base64(kept.data), mimeType: kept.mimeType }; +} /** The kept images a text refers to, with their names, in order, each once; ones never kept are skipped. */ export async function namedImages(store: Store, text: string): Promise<{ name: string; image: ImageContent }[]> { - const names = [...new Set(Array.from(text.matchAll(REF), match => match[1]))], kept = await Promise.all(names.map(name => store.image(name))); + const names = [...new Set(Array.from(text.matchAll(REF), match => match[1]))], images = await Promise.all(names.map(name => loadImage(store, name))); return names.flatMap((name, k) => { - const image = kept[k]; - return image ? [{ name, image: { type: 'image' as const, data: base64(image.data), mimeType: image.mimeType } }] : []; + const image = images[k]; + return image ? [{ name, image }] : []; }); } /** The kept images a text refers to, in order, each once; ones never kept are skipped. */ diff --git a/src/core/index.ts b/src/core/index.ts index a47954bc..79eb2a6a 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -8,6 +8,6 @@ export { cachePayload } from './cache.ts'; export { allowSearch, CHILD, COMPACT, CONTINUITY, SEARCH_DOC, TURN_REFRESHED, VIEW_DOC } from './prompts.ts'; export { boundedMessage, childInput, completedExchange, logMessage, refreshBudget, refreshedRun, refreshPoint, textContent, withView, type AnyMessage, type ImageRef } from './messages.ts'; export { memoryTools, searchTool } from './tools.ts'; -export { decode, isImage, loadImages, ref, refName } from './images.ts'; +export { decode, isImage, loadImage, loadImages, ref, refName } from './images.ts'; export type { ImageContent } from '@earendil-works/pi-ai'; export { chronological, deduplicate, planImport, runImport, type ImportedEntry, type ImportMode } from './import.ts'; diff --git a/src/core/messages.ts b/src/core/messages.ts index c3824eb8..b90bff8b 100644 --- a/src/core/messages.ts +++ b/src/core/messages.ts @@ -1,6 +1,6 @@ import type { ImageContent, Message, TextContent, UserMessage } from '@earendil-works/pi-ai'; import { CAP, cap, type Kind, type Memory } from './memory.ts'; -import { isImage } from './images.ts'; +import { isImage, refName } from './images.ts'; /** A turn's messages as core reads them: pi-ai's, plus a host's own (Pi's custom messages), which pass through untouched. */ export type AnyMessage = Message | { role: string; timestamp: number; content?: string | (TextContent | ImageContent)[] }; @@ -52,12 +52,14 @@ export function childInput({ view, previous = [], run = [], task }: { view: stri }; return `${[view, ...part('The exchange before this run', previous), ...part('This run so far', run)].join('\n\n')}${TASK_HEADER}${task}`; } -/** A tool result over CAP keeps its head and tail, and its images. */ +/** A text block that is only an image's ref, which a host may send in the image's place until the request is built (durable). */ +const isRef = (block: TextContent | ImageContent) => block.type === 'text' && refName(block.text) !== undefined; +/** A tool result over CAP keeps its head and tail, and its images and their refs. */ export function boundedMessage(message: M): M { if (!is(message, 'toolResult')) return message; - const text = message.content.filter(c => c.type === 'text').map(c => c.text).join('\n'); + const text = message.content.flatMap(c => c.type === 'text' && !isRef(c) ? [c.text] : []).join('\n'); if (text.length <= CAP) return message; - return { ...message, content: [{ type: 'text', text: cap(text) }, ...message.content.filter(c => c.type === 'image')] }; + return { ...message, content: [{ type: 'text', text: cap(text) }, ...message.content.filter(c => c.type === 'image' || isRef(c))] }; } /** The view goes before the first user message's words, in that message, so everything up to them is one cached prefix. */ export function withView(messages: readonly M[], view: string): (M | UserMessage)[] { @@ -87,11 +89,12 @@ export function completedExchange(history: readonly AnyMessage[], imageRef?: Ima /** Pi's own estimate for an image when it decides to compact: 4,800 characters, about 1,200 tokens. */ const IMAGE_CHARS = 4_800; -/** Characters a message adds to a request: its text, thinking, tool-call arguments and images, as Pi estimates them. Framing is left out. */ +/** Characters a message adds to a request: its text, thinking, tool-call arguments and images (an image's ref too), as Pi estimates + * them. Framing is left out. */ export function messageChars({ content = [] }: AnyMessage) { if (typeof content === 'string') return content.length; let sum = 0; - for (const block of content) sum += block.type === 'text' ? block.text.length : block.type === 'thinking' ? block.thinking.length + for (const block of content) sum += block.type === 'text' ? refName(block.text) === undefined ? block.text.length : IMAGE_CHARS : block.type === 'thinking' ? block.thinking.length : block.type === 'toolCall' ? JSON.stringify(block.arguments).length : block.type === 'image' ? IMAGE_CHARS : 0; return sum; } diff --git a/src/durable/images.ts b/src/durable/images.ts index 47f4ceb0..cff56dfb 100644 --- a/src/durable/images.ts +++ b/src/durable/images.ts @@ -1,5 +1,5 @@ import type { GenerationHooks, ToolExecutionResult } from '@earendil-works/pi-durable'; -import { decode, isImage, isMime, loadImages, ref, refName, type AnyMessage, type ImageContent, type ImageRef, type Store } from '../core/index.ts'; +import { decode, isImage, isMime, loadImage, ref, refName, type AnyMessage, type ImageContent, type ImageRef, type Store } from '../core/index.ts'; export type Message = Parameters[0]['messages'][number]; type Block = NonNullable[number]; @@ -44,11 +44,11 @@ export async function keep(store: Store, image: ImageContent, name: string) { } } -/** `content` with each image the store keeps replaced by its ref; one it refuses stays. `put: false` for images it has already, - * such as zoom's. */ -export async function offload(store: Store, content: readonly Block[], put = true): Promise { +/** `content` with each image the store keeps replaced by its ref; one it refuses stays. Zoom's images are put again too: the store + * gives them back as canonical base64, whose name differs from the stored one when the image came in otherwise. */ +export async function offload(store: Store, content: readonly Block[]): Promise { const { named, imageRef } = await imageNames(content.filter(isImage)); - const kept = await Promise.all(named.map(async ({ image, name }) => !put || await keep(store, image, name) ? [image.data] : [])); + const kept = await Promise.all(named.map(async ({ image, name }) => await keep(store, image, name) ? [image.data] : [])); const offloaded = new Set(kept.flat()); return content.map(block => isImage(block) && offloaded.has(block.data) ? { type: 'text', text: imageRef(block) } : block); } @@ -68,18 +68,22 @@ export function fitsBound(output: readonly Block[]) { * store lacks or can't read stays as it is. */ export async function restore(store: Store, messages: readonly Message[]) { const reads = new Map>(); - const read = (block: Text) => { - const pending = reads.get(block.text); + const read = (name: string, block: Text) => { + const pending = reads.get(name); if (pending) return pending; - const image = loadImages(store, block.text).then(([found]) => found ?? block, error => { + const image = loadImage(store, name).then(found => found ?? block, error => { console.error(`OptChat could not read an image: ${error instanceof Error ? error.message : String(error)}`); return block; }); - reads.set(block.text, image); + reads.set(name, image); return image; }; return Promise.all(messages.map(async m => { if ((m.role !== 'user' && m.role !== 'toolResult') || typeof m.content === 'string') return m; - return { ...m, content: await Promise.all(m.content.map(block => isRef(block) ? read(block) : block)) }; + return { ...m, content: await Promise.all(m.content.map(block => { + if (block.type !== 'text') return block; + const name = refName(block.text); + return name === undefined ? block : read(name, block); + })) }; })); } diff --git a/src/durable/optchat.ts b/src/durable/optchat.ts index b4576f9c..bf920e54 100644 --- a/src/durable/optchat.ts +++ b/src/durable/optchat.ts @@ -89,10 +89,10 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp }; /** A model-issued tool's images go to the store, so Pi's transcript keeps their refs, unless its text could pass Pi Durable's - * bound (images.ts). Zoom's images are in the store already. */ + * bound (images.ts). */ const afterTool: ToolHooks['afterTool'] = async (call, result, api) => { if (api.conversationId !== id || call.parent || !result.output?.some(isImage) || !fitsBound(result.output)) return undefined; - return { ...result, output: await offload(memory.store, result.output, call.name !== 'zoom') }; + return { ...result, output: await offload(memory.store, result.output) }; }; const handle = async (ctx: Context): Promise => { @@ -203,9 +203,9 @@ export function optchat({ memory, conversation: id = ROOT_CONVERSATION_ID, promp }; const beforeRequest: GenerationHooks['beforeRequest'] = async (request, api, ctx) => { if (api.conversationId !== id) return undefined; - const messages = await restore(memory.store, request.messages); - const { system, previous, run, view } = await compose(ctx, { messages, models: api.models }); - return system.length < messages.length ? { messages: withView([...system, ...previous, ...run], view) } : undefined; + const { system, previous, run, view } = await compose(ctx, { messages: request.messages, models: api.models }); + // Images go back in last, into only what is sent: a refresh drops older steps' refs unread. + return system.length < request.messages.length ? { messages: withView(await restore(memory.store, [...system, ...previous, ...run]), view) } : undefined; }; const Remember = defineTask, { phase: 'remember' }, null>({ diff --git a/test/durable.test.ts b/test/durable.test.ts index bf031835..9a796190 100644 --- a/test/durable.test.ts +++ b/test/durable.test.ts @@ -316,6 +316,17 @@ test('after a restart, zoom returns a tool\'s kept image; its own result keeps o } finally { await again.close(); } }); +test('zoom returns a user\'s image that came in as unpadded base64: the store gives it back padded, under another name', async () => { + const { images } = memoryImages(), unpadded = { ...photo, data: photo.data.replace(/=+$/, '') }; + const agent = await open(file(), [fauxAssistantMessage('nice'), fauxAssistantMessage(fauxToolCall('zoom', { id: 0 }), { stopReason: 'toolUse' }), fauxAssistantMessage('here')], undefined, { images }); + try { + await (await agent.root.submit({ type: 'input', content: await agent.chat.offloadImages([{ type: 'text', text: 'look' }, unpadded]) }, context)).wait(context); + await agent.idle(); + await agent.ask('show it again'); + assert.deepEqual(imagesIn(agent.sent.at(-1), 'toolResult'), [photo]); + } finally { await agent.close(); } +}); + test('an image the store refuses stays inline in Pi\'s row and still reaches the model', async (t) => { t.mock.method(console, 'error', () => {}); const { kept, images } = memoryImages(), refused = { ...photo, data: Buffer.alloc(1000, 8).toString('base64') }, putImage = images.putImage; @@ -349,6 +360,27 @@ test('a ref the store lacks or can\'t read is sent as its text, and the run goes } finally { await agent.close(); } }); +test('a step a refresh dropped has its image read once more, for its summary, not for each request after', async () => { + // A window of 10,000 tokens: a refresh once the run passes 12,000 characters (30%), each step here an image of about 4,800. + let shot = 0; + const camera: Extension = defineExtension({ name: 'camera', tools: [defineTool({ + name: 'camera', description: 'Takes a photo.', parameters: Type.Object({}), replay: 'safe', + execute: async () => ({ output: [{ type: 'text', text: 'snap' }, { ...photo, data: Buffer.alloc(1000, ++shot).toString('base64') }] }), + })] }); + const { images } = memoryImages(), store = images.image, reads: { request: number; name: string }[] = []; + const agent = await open(file(), [...Array.from({ length: 5 }, () => fauxAssistantMessage(fauxToolCall('camera', {}), { stopReason: 'toolUse' })), fauxAssistantMessage('done')], + undefined, { images: { ...images, image: name => { reads.push({ request: agent.sent.length, name }); return store(name); } }, extensions: () => [camera], contextWindow: 10_000 }); + try { + await agent.ask('take five'); + assert.ok(agent.sent.some(request => request.some(m => text(m.content).includes('this turn grew long'))), 'a request was refreshed'); + // The first photo is read for the two requests that send it and once for its summary, not again for the requests after the + // refresh that dropped it. + const first = nameOf({ ...photo, data: Buffer.alloc(1000, 1).toString('base64') }); + assert.deepEqual(reads.filter(read => read.name === first).map(read => read.request), [1, 2, 3]); + assert.deepEqual(agent.sent.map(request => imagesIn(request, 'toolResult').some(image => nameOf(image) === first)), [false, true, true, false, false, false]); + } finally { await agent.close(); } +}); + test('cache marks go on an Anthropic request that carries the view; a request with its own onPayload is left alone', async () => { const faux = fauxProvider({ api: 'anthropic-messages' }), models = createModels(); models.setProvider(faux.provider);