Keep durable chat images once, out of Pi Durable's transcript rows

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`.

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 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:
 
@@ -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. 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] …".

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;

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';
@@ -16,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<ImageContent | undefined> {
+  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. */

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 } 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';

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<M extends AnyMessage>(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<M extends AnyMessage>(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;
 }