diff --git a/changelog.d/ingestion-body-digests.added.md b/changelog.d/ingestion-body-digests.added.md new file mode 100644 index 000000000..0242f0bf8 --- /dev/null +++ b/changelog.d/ingestion-body-digests.added.md @@ -0,0 +1,32 @@ +- **Delivered-body digests on every stored MCPL inbound item.** Both MCPL + lanes stamp two fields, outside the frozen source envelope (console and API + input, and the host's own notices, carry neither): + - `metadata.sourceBodyDigest` is the version identity of the body as + delivered, taken at ingestion, after MCPL conversion and before any host + decoration or storage sharding. It is the SHA-256 (hex) of `[blocks]` + as canonical JSON (keys sorted at every level, `undefined` object entries + dropped, and an `undefined` array element written as `null` as JSON + writes it, though the store refuses a body holding one), with the blocks + as the store keeps them: inline media as + the store hands it back (base64 re-encoded from its bytes, an image's + media type taken from its bytes' signature, only `type` and `source` + kept), strings well-formed (a lone surrogate as U+FFFD) and an entry + whose key isn't well-formed dropped, as the store's append drops it. So + the body hashes alike as delivered, as handed to storage and as read + back, and an undecorated, unsharded stored copy hashes to it. + - `metadata.storedBodyDigest` uses the same function over exactly the + blocks handed to storage, decorations included, taken where each path + stores. + + Both are always the host's own: values an adapter supplies are replaced. + A stored copy, read back with its blobs resolved, that no longer hashes to + its `storedBodyDigest` was changed after it arrived (`editMessage` keeps + metadata). That check is of an unsharded copy: a body the store shards + (context-manager's adaptive ingress chunking) carries the whole body's two + digests on every shard (`bodyGroupId` and `shardIndex` mark them), and no + shard hashes to them. A `blob_ref` block is digested as given, while the + store resolves it on read, so a copy holding one doesn't match; the MCPL + lanes never store one. `sourceBodyDigest(blocks)` is exported from the + package root as the one implementation consumers share; it carries the + framing and the store's normalization, so check a copy with it rather than + hashing your own serialization. diff --git a/src/framework.ts b/src/framework.ts index 750f86b0d..cd0aec0e0 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -105,7 +105,7 @@ import { InferenceRouter } from './mcpl/inference-router.js'; import { ChannelRegistry, connectorText, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; import { KEEP_IDLE_MS, KEEP_NEWEST, ProseDraftStore, draftState, uncertainAttempt, type Draft, type DraftAttempt, type DraftReason, type DraftState, type InheritedRisk } from './prose-drafts.js'; import { ConversationRouter, CONVERSATION_ROUTING_DEPRECATION_NOTICE } from './mcpl/conversation-router.js'; -import { INBOUND_SOURCE_KEY, type InboundAcceptanceObserver, type InboundChannelSource, type InboundLane, type InboundSource, type InboundUnscopedSource } from './mcpl/inbound-source.js'; +import { INBOUND_SOURCE_KEY, sourceBodyDigest, type InboundAcceptanceObserver, type InboundChannelSource, type InboundLane, type InboundSource, type InboundUnscopedSource } from './mcpl/inbound-source.js'; import type { IncomingMessageFacts, RoutingDecision } from './mcpl/conversation-router.js'; import { safeSlice } from './safe-slice.js'; import { SEND_TOOLS, bareToolName } from './send-tools.js'; @@ -12076,6 +12076,11 @@ export class AgentFramework { this.noteInboundAccepted(source); } metadata[INBOUND_SOURCE_KEY] = source; + // The delivered body's version identity (mcpl/inbound-source.ts + // sourceBodyDigest; room-220 #46752–#47210), from the body as delivered, + // before any host decoration. The stored copy's own digest is taken at + // each storage site below, over exactly the blocks stored there. + metadata.sourceBodyDigest = sourceBodyDigest(event.content); // Per-channel conversation routing: messages go to the channel's fork // agent (spawned from the template on first qualifying message), never @@ -12088,7 +12093,7 @@ export class AgentFramework { if (!target) return undefined; if (this.conversationRouter && this.conversationAgentHomes.has(target.name)) { metadata.triggered = event.triggerInference ?? false; - const id = target.getContextManager().addMessage('user', event.content, metadata); + const id = target.getContextManager().addMessage('user', event.content, AgentFramework.withStoredDigest(metadata, event.content)); // A correction may target an older engagement. Refresh only the // binding whose agent actually received the message. if (this.conversationRouter.getBinding(event.channelId)?.agentName === target.name) { @@ -12157,7 +12162,7 @@ export class AgentFramework { } const placement: CoalescingPlacement = { agent: '' }; - const id = this.addMessage('user', incomingContent, metadata, { placement, bypassDeferralFor: event.assemblingFor }); + const id = this.addMessage('user', incomingContent, AgentFramework.withStoredDigest(metadata, incomingContent), { placement, bypassDeferralFor: event.assemblingFor }); this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:channel-incoming' }); if (event.triggerInference && !divert) { @@ -12271,7 +12276,7 @@ export class AgentFramework { const trigger = decision.trigger && event.triggerInference !== false; messageMetadata.triggered = trigger; - const id = agent.getContextManager().addMessage('user', event.content, messageMetadata); + const id = agent.getContextManager().addMessage('user', event.content, AgentFramework.withStoredDigest(messageMetadata, event.content)); router.touch(event.channelId); this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:channel-incoming' }); @@ -12870,6 +12875,27 @@ export class AgentFramework { }; } + /** + * An inbound item's metadata with its stored copy's own digest + * (`storedBodyDigest`): the delivered-body digest's function over exactly + * the blocks handed to storage here — every decoration included, before + * storage shards them. That function hashes blocks as the store keeps them + * (inline media re-encoded and relabeled from its bytes), so an unsharded + * copy read back hashes to this too. A body the store shards (adaptive + * ingress chunking) carries this on every shard, with its metadata, and no + * shard hashes to it. Taken at each storage site, never earlier, because + * a path can still decorate the body after ingestion stamped it (the + * closed-channel invitation; room-220 #48282). An edit through + * editMessage keeps metadata but not this hash. + */ + private static withStoredDigest( + metadata: Record, + content: readonly ContentBlock[], + ): Record { + metadata.storedBodyDigest = sourceBodyDigest(content); + return metadata; + } + /** Source envelope of a channels/incoming event (lane facts as given). */ private channelEventSource( event: { @@ -13609,6 +13635,12 @@ export class AgentFramework { metadata[INBOUND_SOURCE_KEY] = source; const content = [...event.content]; + // The delivered body's version identity, before any host decoration + // (the closed-channel invitation below). Every push that reaches here + // gets one, written over any value the adapter's origin carried. (A + // visibly-empty push is refused at admission; the silent-heartbeat marker + // stores nothing, so its metadata is never kept.) + metadata.sourceBodyDigest = sourceBodyDigest(content); if (triggerChannel) { const origin = (event.origin ?? {}) as Record; const invitation = this.buildClosedChannelInvitation({ @@ -13637,7 +13669,7 @@ export class AgentFramework { const placement: CoalescingPlacement = { agent: '' }; if (!silentHeartbeat) { - const id = this.addMessage('user', content, metadata, { placement, bypassDeferralFor: event.assemblingFor }); + const id = this.addMessage('user', content, AgentFramework.withStoredDigest(metadata, content), { placement, bypassDeferralFor: event.assemblingFor }); this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:push-event' }); } else { console.error(`[heartbeat] ${event.serverId}: accepted silent scheduled wake ${logValue(event.eventId ?? '')}`); diff --git a/src/index.ts b/src/index.ts index 9cceb37d4..12c634638 100644 --- a/src/index.ts +++ b/src/index.ts @@ -86,8 +86,10 @@ export type { GateConfig, GateOptions, GatePolicy, GatePolicyMatch, GateBehavior export { ChannelRegistry } from './mcpl/index.js'; export type { PublishRefusalReason } from './mcpl/index.js'; -// Where each accepted inbound item came from (`metadata.inboundSource`) -export { INBOUND_SOURCE_KEY, readInboundSource, conversationKey } from './mcpl/index.js'; +// Where each accepted inbound item came from (`metadata.inboundSource`), and +// the digest its delivered and stored bodies are stamped with +// (`metadata.sourceBodyDigest`, `metadata.storedBodyDigest`) +export { INBOUND_SOURCE_KEY, readInboundSource, conversationKey, sourceBodyDigest } from './mcpl/index.js'; export type { InboundSource, InboundChannelSource, diff --git a/src/mcpl/inbound-source.ts b/src/mcpl/inbound-source.ts index 766eb07cf..e2bc75645 100644 --- a/src/mcpl/inbound-source.ts +++ b/src/mcpl/inbound-source.ts @@ -21,6 +21,8 @@ * key: `eventId` is present only when the producer supplied one. */ +import { createHash } from 'node:crypto'; + /** The MCPL admission lane that accepted an item. Its contract decides what * an `eventId` is worth: `push/event` deduplicates by eventId, and RFC-006 * coalesced admission (`coalesced: true`) guarantees stable retries and @@ -154,3 +156,131 @@ export function conversationKey(source: InboundSource): string | undefined { if (source.kind === 'surface') return JSON.stringify(['surface', source.surface]); return undefined; } + +/** + * JSON with object keys sorted at every level, written as the store's append + * keeps a value (its native UTF-8 write, which the real-store digest table + * checks): an `undefined` object entry is dropped; a string is well-formed + * (each lone surrogate comes back as U+FFFD); and an entry whose key isn't + * well-formed is dropped (the store doesn't keep it). An `undefined` array + * element is written as `null`, as JSON writes it: the store's append + * refuses a body holding one, so no stored copy does. So a value + * hashes alike however its keys were ordered (content read back from the + * store has them in a different order than it was written), and alike + * before and after the store. editMessage writes JSON instead, so an edited + * copy can keep what an append wouldn't; it has changed anyway. + * + * It expects JSON-shaped values, as the MCPL lanes deliver them (parsed from + * JSON). Any other object is written by its own enumerable entries, unlike + * JSON.stringify: a Date becomes `{}`, as the store's own write keeps it, but + * a deferred write recovered from its JSON file brings it back as a string. + * + * sourceBodyDigest's serializer, kept out of the package API: serialization + * and framing both stay inside sourceBodyDigest, so consumers don't + * duplicate either (`canonicalJson(blocks)` alone hashes differently). + */ +export function canonicalJson(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]`; + if (value && typeof value === 'object') { + const entries = Object.entries(value as Record) + .filter(([k, v]) => v !== undefined && k.isWellFormed()) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)); + return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`).join(',')}}`; + } + if (typeof value === 'string') return JSON.stringify(value.toWellFormed()); + return JSON.stringify(value) ?? 'null'; +} + +/** + * A block as the store hands it back. Context-manager keeps inline media as + * blobs (blob-manager.ts): an image whose source isn't a URL, and every + * document, audio or video block, comes back as `{ type, source: { type: + * 'base64', data, mediaType } }` and nothing else, its data re-encoded from + * the decoded bytes, and an image's media type taken from its bytes' + * signature where they have one. Any other block comes back as it was given, + * except a `blob_ref`, which the store resolves to its inline media on read: + * it is digested as given here, so a copy holding one doesn't match its + * digest (the MCPL lanes never store one). + */ +function storedBlock(block: unknown): unknown { + const media = block as { type?: unknown; source?: { type?: unknown; data?: unknown; mediaType?: unknown } } | null; + const blob = media?.type === 'image' + ? media.source?.type !== 'url' + : media?.type === 'document' || media?.type === 'audio' || media?.type === 'video'; + const source = media?.source; + if (!blob || typeof source?.data !== 'string') return block; + const data = canonicalBase64(source.data); + const mediaType = media!.type === 'image' + ? sniffRasterImageMediaType(Buffer.from(data.slice(0, 32), 'base64')) ?? source.mediaType + : source.mediaType; + return { type: media!.type, source: { type: 'base64', data, mediaType } }; +} + +/** + * Base64 as the store writes it back from the decoded bytes: padded, with no + * whitespace, the standard alphabet and zero unused bits. Data already in + * that form, as every stored copy's is, is returned without decoding it all. + */ +function canonicalBase64(data: string): string { + if (data.length % 4 === 0 && /^[A-Za-z0-9+/]*={0,2}$/.test(data)) { + const last = data.slice(-4); + if (Buffer.from(last, 'base64').toString('base64') === last) return data; + } + return Buffer.from(data, 'base64').toString('base64'); +} + +/** + * The store's own test for raster image types with an unambiguous byte + * signature, copied from context-manager's blob-manager.ts (it isn't + * exported): the store relabels an image whose bytes match one of these. + */ +function sniffRasterImageMediaType(bytes: Uint8Array): string | null { + if (bytes.length >= 8 && + bytes[0] === 0x89 && bytes[1] === 0x50 && bytes[2] === 0x4e && bytes[3] === 0x47 && + bytes[4] === 0x0d && bytes[5] === 0x0a && bytes[6] === 0x1a && bytes[7] === 0x0a) { + return 'image/png'; + } + if (bytes.length >= 3 && bytes[0] === 0xff && bytes[1] === 0xd8 && bytes[2] === 0xff) { + return 'image/jpeg'; + } + if (bytes.length >= 6) { + const signature = Buffer.from(bytes.subarray(0, 6)).toString('ascii'); + if (signature === 'GIF87a' || signature === 'GIF89a') return 'image/gif'; + } + if (bytes.length >= 12 && + Buffer.from(bytes.subarray(0, 4)).toString('ascii') === 'RIFF' && + Buffer.from(bytes.subarray(8, 12)).toString('ascii') === 'WEBP') { + return 'image/webp'; + } + return null; +} + +/** + * The version identity of a delivered body (agreed with the receipts lane, + * room-220 #46752–#47210): SHA-256 hex of `canonicalJson([blocks])` over the + * ContentBlock[] the framework stores for the body — after MCPL conversion, + * before any host decoration is added and before storage shards it — with + * each block as the store keeps it (storedBlock, canonicalJson). So a body + * hashes alike as delivered, as handed to storage, and as read back from + * the store. The one-element array keeps the framing an undecorated, + * unsharded stored copy has always hashed to, so a stamped copy and an + * older copy of the same body share a version. + * + * Ingestion stamps it as `metadata.sourceBodyDigest`, and the same function + * over exactly the blocks handed to storage (decorations included) as + * `metadata.storedBodyDigest`, on both MCPL lanes. Both live outside the + * frozen admission envelope: they describe the body actually delivered (a + * materialization, a correction), not the admission. Neither proves later + * presence or completeness; an unsharded stored copy that no longer hashes + * to its storedBodyDigest was changed after delivery (editMessage keeps + * metadata). A body the store shards carries the whole body's two digests + * on every shard, and no shard hashes to them. Check a copy read back with + * its blobs resolved (getAllMessages, getMessage, or getMessageWindow + * without `resolveBlobs: false`): a blob reference never matches. + * + * Exported from the package root, so a consumer checks a copy against either + * field with this same function rather than a copy of it. + */ +export function sourceBodyDigest(blocks: readonly unknown[]): string { + return createHash('sha256').update(canonicalJson([blocks.map(storedBlock)])).digest('hex'); +} diff --git a/src/mcpl/index.ts b/src/mcpl/index.ts index 44e32652e..7747cc5f7 100644 --- a/src/mcpl/index.ts +++ b/src/mcpl/index.ts @@ -173,8 +173,9 @@ export { InferenceRouter } from './inference-router.js'; export { ChannelRegistry } from './channel-registry.js'; export type { PublishRefusalReason } from './channel-registry.js'; -// Inbound source envelope: where each accepted item came from, stamped once -export { INBOUND_SOURCE_KEY, readInboundSource, conversationKey } from './inbound-source.js'; +// Inbound source envelope: where each accepted item came from, stamped once; +// and the digest of a delivered or stored body +export { INBOUND_SOURCE_KEY, readInboundSource, conversationKey, sourceBodyDigest } from './inbound-source.js'; export type { InboundSource, InboundChannelSource, diff --git a/test/fixtures/speech-route-mcpl-server.mjs b/test/fixtures/speech-route-mcpl-server.mjs index 39d2c4e4d..c01e503c0 100644 --- a/test/fixtures/speech-route-mcpl-server.mjs +++ b/test/fixtures/speech-route-mcpl-server.mjs @@ -5,8 +5,10 @@ // BOTH MCPL lanes. The TEST appends one JSON object per line to COMMAND_PATH: // {"op":"incoming","channelId":…,"messageId":…,"mode":"ambient"|"addressed", // "text":…, "threadId"?, "metadata"?, "eventId"?, "coalesce"?} -// {"op":"dm","eventId":…,"authorId":…,"authorName"?,"rawChannelId":…,"text":…} -// {"op":"push","eventId":…,"origin"?,"tags"?,"text":…,"featureSet"?} +// {"op":"dm","eventId":…,"authorId":…,"authorName"?,"rawChannelId":…,"text":…, +// "content"? (content blocks, instead of text; [] included), "origin"?} +// {"op":"push","eventId":…,"origin"?,"tags"?,"text":…,"featureSet"?, +// "content"? (content blocks, instead of text; [] included)} // {"op":"rename","channelId":…,"label":…} (channels/changed) // {"op":"publish-mode","mode":"delivered"|"not-delivered"|"no-receipt"|"error"|"hang"} // {"op":"history","channelId":…,"history":[ChannelIncomingMessage…]} @@ -91,7 +93,7 @@ function pollCommands() { ...(c.origin ?? {}), }, tags: ['chat:dm', 'chat:addressed', 'chat:private'], - payload: { content: textBlock(c.text ?? '') }, + payload: { content: c.content ?? textBlock(c.text ?? '') }, }, }); } else if (c.op === 'push') { @@ -106,7 +108,7 @@ function pollCommands() { timestamp: c.timestamp ?? new Date().toISOString(), ...(c.origin ? { origin: c.origin } : {}), ...(c.tags ? { tags: c.tags } : {}), - payload: { content: textBlock(c.text ?? '') }, + payload: { content: c.content ?? textBlock(c.text ?? '') }, }, }); } else if (c.op === 'rename') { diff --git a/test/inbound-source.test.ts b/test/inbound-source.test.ts index f71b4943e..f3cc531b0 100644 --- a/test/inbound-source.test.ts +++ b/test/inbound-source.test.ts @@ -16,6 +16,9 @@ import { INBOUND_SOURCE_KEY, readInboundSource, conversationKey, + // The package's public digest: consumers import it from the root. + sourceBodyDigest, + PassthroughStrategy, type InboundSource, type Module, type ModuleContext, @@ -26,12 +29,19 @@ import { type ToolCall, type ToolResult, } from '../src/index.js'; +import { ContextManager } from '@animalabs/context-manager'; +import type { ContentBlock } from '@animalabs/membrane'; import { MockMembrane } from './helpers/mock-membrane.js'; +import { canonicalJson } from '../src/mcpl/inbound-source.js'; +import { createHash } from 'node:crypto'; import { fixture, eventually, TS } from './helpers/coalescing-fixture.js'; const FIXTURE = join(import.meta.dirname, 'fixtures/speech-route-mcpl-server.mjs'); const ROOM = 'discord:g1:room'; const RAW_DM = '1548000000000000001'; +/** Bytes that open with PNG's signature, which is all the store reads of an + * image. 14 bytes, so their base64 is padded and has a '/' in it. */ +const PNG = Buffer.from('89504e470d0a1a0afbefbeffffff', 'hex'); async function waitFor(cond: () => boolean, what: string, timeoutMs = 15_000): Promise { const deadline = Date.now() + timeoutMs; @@ -166,6 +176,133 @@ describe('inbound source envelope', () => { assert.equal(observed.length, 1); }); + it('stamps the delivered body\'s digest and the stored copy\'s digest at ingestion', async () => { + command({ op: 'incoming', channelId: ROOM, messageId: 'm-d1', mode: 'ambient', text: 'digest me' }); + await waitFor(() => !!storedWith((m) => m.messageId === 'm-d1'), 'message stored'); + const message = storedWith((m) => m.messageId === 'm-d1')!; + const meta = message.metadata as Record; + // The unsharded framing receipts already hash: canonicalJson([blocks]). + const expected = createHash('sha256').update(canonicalJson([[{ type: 'text', text: 'digest me' }]])).digest('hex'); + assert.equal(meta.sourceBodyDigest, expected); + assert.equal(sourceBodyDigest([{ text: 'digest me', type: 'text' }]), expected, 'key order does not matter'); + // This lane stores the body as delivered: the stored copy hashes the same. + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content)); + assert.equal(meta.storedBodyDigest, meta.sourceBodyDigest); + // Outside the frozen admission envelope. + assert.equal((readInboundSource(message.metadata) as unknown as Record).sourceBodyDigest, undefined); + + // An edit through the supported path keeps the metadata, so only the + // stored digest can reveal that the copy no longer holds the delivery. + const cm = framework.getAgent('scout')!.getContextManager(); + cm.editMessage(message.id, [{ type: 'text', text: 'edited later' }]); + const edited = cm.getAllMessages().find((m) => m.id === message.id)!; + assert.equal((edited.metadata as Record).storedBodyDigest, meta.storedBodyDigest); + assert.notEqual(sourceBodyDigest(edited.content), meta.storedBodyDigest); + }); + + it('a message accepted while its channel was open but stored after it closed: the stored digest covers the invitation', async () => { + // Tessa-974's discriminator (room-220 #48282): the invitation is + // appended after ingestion stamped the delivered digest, at storage + // time, when the channel has closed in between. + const registry = (framework as unknown as { + channelRegistry: { handleChannelToolCall(name: string, input: unknown, caller: unknown): Promise; isChannelOpen(id: string): boolean }; + }).channelRegistry; + const fw = framework as unknown as { pushEvent(e: { type: string }): void }; + const held: Array<{ type: string }> = []; + const push = fw.pushEvent.bind(framework); + fw.pushEvent = (e) => { if (e.type === 'mcpl:channel-incoming') held.push(e); else push(e); }; + command({ op: 'incoming', channelId: ROOM, messageId: 'm-closed', mode: 'addressed', text: 'hello while open' }); + await waitFor(() => held.length > 0, 'incoming accepted and held'); + await registry.handleChannelToolCall('channel_close', { channelId: ROOM }, { kind: 'agent', agentName: 'scout' }); + assert.equal(registry.isChannelOpen(ROOM), false); + fw.pushEvent = push; + for (const e of held) push(e); + await waitFor(() => !!storedWith((m) => m.messageId === 'm-closed'), 'message stored'); + const message = storedWith((m) => m.messageId === 'm-closed')!; + const meta = message.metadata as Record; + assert.equal(meta.channelInvitation, true, 'stored with the closed-channel invitation'); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content), 'the untouched copy matches its own witness'); + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'hello while open' }]), 'the delivered body alone'); + }); + + it('on the push lane, the delivered digest excludes host decoration and the stored digest covers it', async () => { + command({ op: 'dm', eventId: 'ev-dig', authorId: '134', authorName: 'antra', rawChannelId: RAW_DM, text: 'psst' }); + await waitFor(() => !!storedWith((m) => m.eventId === 'ev-dig'), 'dm stored'); + const message = storedWith((m) => m.eventId === 'ev-dig')!; + const meta = message.metadata as Record; + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'psst' }]), 'the adapter body alone'); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content), 'exactly what was stored'); + // The DM's channel is closed, so the host appends its invitation. + assert.equal(message.content.length, 2); + assert.notEqual(meta.storedBodyDigest, meta.sourceBodyDigest, 'the closed-channel invitation decorates the stored copy'); + }); + + it('a visibly-empty ordinary push is refused and stores nothing; an accepted push carries the host\'s digests, never the adapter\'s', async () => { + // Admission (main #236): content with nothing visible is refused with + // -32602 unless it is the exact silent-heartbeat marker or an RFC-006 + // retraction, so it never reaches storage and has no witnesses. Every + // push that is accepted is stored with the host's own digests, written + // over whatever its origin claimed. + const forged = { sourceBodyDigest: 'adapter-value', storedBodyDigest: 'adapter-value' }; + command({ op: 'push', eventId: 'empty-1', content: [], origin: { source: 'timer', ...forged } }); + command({ op: 'push', eventId: 'full-1', text: 'tick', origin: { source: 'timer', ...forged } }); + await waitFor(() => !!storedWith((m) => m.eventId === 'full-1'), 'accepted push stored'); + assert.equal(storedWith((m) => m.eventId === 'empty-1'), undefined, 'the empty push was refused at admission, before the next was handled'); + const message = storedWith((m) => m.eventId === 'full-1')!; + const meta = message.metadata as Record; + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'tick' }]), 'the delivered body, not the adapter\'s value'); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content), 'exactly what was stored, not the adapter\'s value'); + }); + + it('a DM pushed into a closed channel carries the host\'s digests, never the adapter\'s: the delivered body, and the stored copy with its invitation', async () => { + command({ + op: 'dm', eventId: 'ev-forged', authorId: '134', authorName: 'antra', rawChannelId: RAW_DM, text: 'psst again', + origin: { sourceBodyDigest: 'adapter-value', storedBodyDigest: 'adapter-value' }, + }); + await waitFor(() => !!storedWith((m) => m.eventId === 'ev-forged'), 'dm stored'); + const message = storedWith((m) => m.eventId === 'ev-forged')!; + const meta = message.metadata as Record; + assert.equal(meta.channelInvitation, true, 'stored with the closed-channel invitation'); + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'psst again' }]), 'the delivered body, not the adapter\'s value'); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content), 'exactly what was stored, not the adapter\'s value'); + assert.notEqual(meta.storedBodyDigest, meta.sourceBodyDigest, 'the invitation decorates the stored copy'); + }); + + it('a channels/incoming item carries the host\'s digests, never the adapter\'s', async () => { + command({ + op: 'incoming', channelId: ROOM, messageId: 'm-forged', mode: 'ambient', text: 'forged digests', + metadata: { sourceBodyDigest: 'adapter-value', storedBodyDigest: 'adapter-value' }, + }); + await waitFor(() => !!storedWith((m) => m.messageId === 'm-forged'), 'message stored'); + const message = storedWith((m) => m.messageId === 'm-forged')!; + const meta = message.metadata as Record; + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'forged digests' }]), 'the delivered body, not the adapter\'s value'); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(message.content), 'exactly what was stored, not the adapter\'s value'); + }); + + it('an image the store relabels or re-encodes still matches both its digests once read back', async () => { + // slimepriestess's review of #252: the store takes an image's media type + // from its bytes and re-encodes its base64, so a digest of the blocks as + // handed to storage failed against the untouched copy read back. + const png = PNG.toString('base64'); + const pushes: Record = { + 'img-labeled': { type: 'image', data: png, mimeType: 'image/png' }, + 'img-mislabeled': { type: 'image', data: png, mimeType: 'image/jpeg' }, + 'img-wrapped': { type: 'image', data: `${png.slice(0, 11)}\n${png.slice(11)}`, mimeType: 'image/png' }, + }; + for (const [eventId, block] of Object.entries(pushes)) command({ op: 'push', eventId, content: [block] }); + await waitFor(() => Object.keys(pushes).every((id) => !!storedWith((m) => m.eventId === id)), 'images stored'); + for (const eventId of Object.keys(pushes)) { + const message = storedWith((m) => m.eventId === eventId)!; + const meta = message.metadata as Record; + const image = message.content.find((b) => b.type === 'image'); + assert.deepEqual(image, { type: 'image', source: { type: 'base64', data: png, mediaType: 'image/png' } }, + `${eventId}: stored as the bytes' own type, in canonical base64`); + assert.equal(sourceBodyDigest(message.content), meta.storedBodyDigest, `${eventId}: the untouched copy matches its stored digest`); + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([image]), `${eventId}: the delivered body hashes as the image stored for it`); + } + }); + it("drops an envelope a module's metadata carries: only the framework writes one", async () => { const forged = { kind: 'channel', lane: 'channels/incoming', serverId: 'evil', binding: 'x', channelId: 'discord:g9:elsewhere', acceptedAt: 1 }; framework.pushEvent({ type: 'relay', content: 'relayed words', metadata: { inboundSource: forged, note: 'kept' } } as unknown as ProcessEvent); @@ -325,6 +462,61 @@ describe('readInboundSource', () => { }); }); +describe('stored digests when the store shards a body', () => { + /** A strategy whose store splits a body of more than one block into one + * shard per block, as adaptive ingress chunking does. */ + class ShardingStrategy extends PassthroughStrategy { + chunkIngressMessage(_participant: string, content: ContentBlock[]) { + if (content.length < 2) return null; + return { bodyGroupId: `group-${content.length}-${Date.now()}`, shards: content.map((block, shardIndex) => ({ content: [block], shardIndex })) }; + } + } + + it("every shard carries the whole body's digests, and no shard hashes to them; an unsharded copy matches its own", async () => { + const tempDir = mkdtempSync(join(tmpdir(), 'inbound-shards-')); + const commandPath = join(tempDir, 'commands.jsonl'); + writeFileSync(commandPath, ''); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'scout', model: 'test-model', systemPrompt: 'You are scout.', strategy: new ShardingStrategy() }], + mcplServers: [{ + id: 'discord', + command: process.execPath, + args: [FIXTURE], + env: { STATUS_PATH: join(tempDir, 'status.jsonl'), COMMAND_PATH: commandPath }, + }], + modules: [], + }); + try { + await framework.start(); + const command = (c: Record): void => appendFileSync(commandPath, JSON.stringify(c) + '\n'); + const blocks = [{ type: 'text', text: 'first part' }, { type: 'text', text: 'second part' }]; + command({ op: 'push', eventId: 'sharded', content: blocks }); + command({ op: 'push', eventId: 'whole', text: 'one block' }); + const stored = () => framework.getAgent('scout')!.getContextManager().getAllMessages(); + await waitFor(() => stored().some((m) => (m.metadata as { eventId?: string } | undefined)?.eventId === 'whole'), 'both pushes stored'); + const shards = stored().filter((m) => (m.metadata as { eventId?: string } | undefined)?.eventId === 'sharded'); + assert.equal(shards.length, 2, 'stored as two shards'); + assert.ok(shards[0]!.bodyGroupId && shards.every((m) => m.bodyGroupId === shards[0]!.bodyGroupId)); + assert.deepEqual(shards.map((m) => m.shardIndex), [0, 1]); + const whole = sourceBodyDigest(blocks); + for (const shard of shards) { + const meta = shard.metadata as Record; + assert.equal(meta.sourceBodyDigest, whole, 'the whole delivered body'); + assert.equal(meta.storedBodyDigest, whole, 'the whole body handed to storage'); + assert.notEqual(sourceBodyDigest(shard.content), whole, 'a shard alone is not the body'); + } + const one = stored().find((m) => (m.metadata as { eventId?: string } | undefined)?.eventId === 'whole')!; + assert.equal(one.bodyGroupId, undefined, 'a one-block body is not sharded'); + assert.equal(sourceBodyDigest(one.content), (one.metadata as Record).storedBodyDigest, 'an unsharded copy matches'); + } finally { + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + } + }); +}); + describe('inbound source envelope under RFC-006 coalescing', () => { const install = (framework: AgentFramework): InboundSource[] => { const observed: InboundSource[] = []; @@ -366,6 +558,14 @@ describe('inbound source envelope under RFC-006 coalescing', () => { assert.equal(observed.length, 1, 'materialization is not a second acceptance'); const stored = sourceOf(f, 'document_diff'); assert.deepEqual(stored, { ...accepted, materialized: true }); + // The materialized copy is its own storage site: both digests are the + // host's, over the rendered body, not the deferred placeholder. + const copy = f.framework.getAgent('agent')!.getContextManager().getAllMessages() + .find((m) => JSON.stringify(m.content).includes('document_diff'))!; + const meta = copy.metadata as Record; + assert.equal(meta.storedBodyDigest, sourceBodyDigest(copy.content), 'exactly what was stored'); + assert.equal(meta.sourceBodyDigest, sourceBodyDigest(copy.content), 'the rendered body, undecorated'); + assert.notEqual(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'fallback_unavailable' }]), 'not the placeholder'); }); it('a channel-scoped coalesced message belongs to its channel, frozen at acceptance', async (t) => { @@ -385,6 +585,46 @@ describe('inbound source envelope under RFC-006 coalescing', () => { assert.deepEqual(observed, [stored]); }); + it('a channel message deferred mid-turn and stored at the boundary still hashes to its stored digest', async (t) => { + const f = await fixture(); t.after(f.close); + await f.register('chat'); + const internals = f.framework as unknown as { activeTurnTokens: Map; flushDeferredWrites(label: string): Promise }; + internals.activeTurnTokens.set('agent', 100); + await f.send('channels/incoming', { messages: [f.channel('e1', 'deferred_words', { initial: true })] }); + assert.ok(!f.context().includes('deferred_words'), 'held until the boundary'); + internals.activeTurnTokens.delete('agent'); + await internals.flushDeferredWrites('test'); + const copy = f.framework.getAgent('agent')!.getContextManager().getAllMessages() + .find((m) => JSON.stringify(m.content).includes('deferred_words'))!; + const meta = copy.metadata as Record; + assert.equal(meta.storedBodyDigest, sourceBodyDigest(copy.content), 'the copy stored at the boundary matches'); + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text: 'deferred_words' }]), 'the delivered body, unchanged'); + }); + + it("the conversation router's and a fork correction's storage sites carry the host's digests, never the adapter's", async (t) => { + const f = await fixture({ framework: { conversations: { templateAgent: 'agent', bind: { channel: 'always' }, trigger: { channel: 'always' } } } as never }); t.after(f.close); + await f.register(); + const forged = { sourceBodyDigest: 'adapter-value', storedBodyDigest: 'adapter-value' }; + const router = (f.framework as unknown as { conversationRouter: { getBinding(id: string): { agentName: string } | undefined; unbind(id: string): void } }).conversationRouter; + const check = (text: string, fork: string) => { + const copy = f.framework.getAgent(fork)!.getContextManager().getAllMessages().find((m) => JSON.stringify(m.content).includes(text)); + assert.ok(copy, `${text} stored in ${fork}`); + const meta = copy.metadata as Record; + assert.equal(meta.sourceBodyDigest, sourceBodyDigest([{ type: 'text', text }]), `${text}: the delivered body, not the adapter's value`); + assert.equal(meta.storedBodyDigest, sourceBodyDigest(copy.content), `${text}: exactly what was stored, not the adapter's value`); + }; + // Routed into a fresh fork by the conversation router. + await f.send('channels/incoming', { messages: [{ ...f.channel('c1', 'routed_original', { initial: true }), metadata: forged }] }); + const fork = router.getBinding('chat')!.agentName; + check('routed_original', fork); + // A correction follows the prior delivery into that fork, though the + // channel is no longer bound to it. + router.unbind('chat'); + await f.send('channels/incoming', { messages: [{ ...f.channel('c2', 'fork_correction'), metadata: forged }] }); + assert.equal(router.getBinding('chat'), undefined, 'no fresh fork was spawned for the correction'); + check('fork_correction', fork); + }); + it("a channel-scoped push edit that names no message takes its subject's message and thread, as its delivery does", async (t) => { const f = await fixture(); t.after(f.close); await f.register(); @@ -411,3 +651,139 @@ describe('inbound source envelope under RFC-006 coalescing', () => { assert.equal(stored.threadId, 'thread2'); }); }); + +describe('sourceBodyDigest hashes a body as the store keeps it', () => { + // Every agent stores through context-manager over chronicle. Whatever the + // store changes on the way in (an image's label, base64's spelling, fields + // it doesn't keep, a lone surrogate), the body read back must hash as it + // did when it was handed over, live and after the store is reopened. + const png = PNG.toString('base64'); + const image = (data: string, mediaType = 'image/png', extra: Record = {}) => + ({ type: 'image', source: { type: 'base64', data, mediaType }, ...extra }); + const media = (type: string, data: string, mediaType: string, extra: Record = {}) => + ({ type, source: { type: 'base64', data, mediaType }, ...extra }); + /** Each body, and whether the store hands it back changed. */ + const bodies: Record = { + 'a labeled image': { blocks: [image(png)], changed: false }, + 'PNG bytes labeled image/jpeg': { blocks: [image(png, 'image/jpeg')], changed: true }, + 'JPEG bytes labeled image/png': { blocks: [image(Buffer.from('ffd8ffe000104a464946', 'hex').toString('base64'))], changed: true }, + // One fixture per branch of the copied sniffer, each labeled as something + // else, so a context-manager that stops sniffing one fails here. + 'GIF87a bytes labeled image/png': { blocks: [image(Buffer.from('GIF87a\x01\x00\x01\x00\x00\x00\x00', 'latin1').toString('base64'))], changed: true }, + 'GIF89a bytes labeled image/jpeg': { blocks: [image(Buffer.from('GIF89a\x01\x00\x01\x00\x00\x00\x00', 'latin1').toString('base64'), 'image/jpeg')], changed: true }, + 'WebP bytes labeled image/png': { blocks: [image(Buffer.from('RIFF\x1a\x00\x00\x00WEBPVP8L\x0d\x00\x00\x00', 'latin1').toString('base64'))], changed: true }, + 'base64 with a newline in it': { blocks: [image(`${png.slice(0, 11)}\n${png.slice(11)}`)], changed: true }, + 'base64 with whitespace around it': { blocks: [image(` ${png}\n`)], changed: true }, + 'base64 without its padding': { blocks: [image(png.replace(/=+$/, ''))], changed: true }, + 'base64 with non-zero unused bits': { blocks: [image(png.replace(/8=$/, '9='))], changed: true }, + 'base64 in the URL-safe alphabet': { blocks: [image(png.replace(/\//g, '_'))], changed: true }, + // Bytes the store has no signature for keep their label. Labeled + // image/png, so a context-manager that learns to sniff one fails here. + 'SVG bytes labeled image/png': { blocks: [image(Buffer.from('').toString('base64'))], changed: false }, + 'BMP bytes labeled image/png': { blocks: [image(Buffer.from('424d3a0000000000000036000000', 'hex').toString('base64'))], changed: false }, + 'AVIF bytes labeled image/png': { blocks: [image(Buffer.from('0000001c6674797061766966000000006d696631', 'hex').toString('base64'))], changed: false }, + 'an image whose source is neither base64 nor a URL': { blocks: [{ type: 'image', source: { type: 'file', data: png, mediaType: 'image/png' } }], changed: true }, + 'an image with fields the store drops': { blocks: [image(png, 'image/png', { sourceUrl: 'https://cdn.example/a.png', tokenEstimate: 85 })], changed: true }, + 'an image by URL, its fields kept': { blocks: [{ type: 'image', source: { type: 'url', url: 'https://cdn.example/a.png' }, sourceUrl: 'https://cdn.example/a.png' }], changed: false }, + 'audio, wrapped, with a duration': { blocks: [media('audio', 'SUQz\nAwAAAAAA', 'audio/mpeg', { duration: 3 })], changed: true }, + 'a document with a filename': { blocks: [media('document', 'JVBERi0xLjQ=', 'application/pdf', { filename: 'a.pdf' })], changed: true }, + 'text with lone surrogates': { blocks: [{ type: 'text', text: 'cut \ud83d here, \ude00 there, \ud83d\ud83d twice' }], changed: true }, + 'a key that is not well-formed': { blocks: [{ type: 'text', text: 'x', rawItem: { 'k\ud800': 1, kept: 2 } }], changed: true }, + 'a relabeled image with a text decoration': { blocks: [image(png, 'image/jpeg'), { type: 'text', text: '[invitation]' }], changed: true }, + 'a video with a duration': { blocks: [media('video', 'AAAAGGZ0eXBtcDQy', 'video/mp4', { duration: 2 })], changed: true }, + // No media type: an image takes its bytes' own (an image without a + // signature, or other media, is refused at write: see below). + 'an image without a media type': { blocks: [{ type: 'image', source: { type: 'base64', data: png } }], changed: true }, + 'an image whose media type is undefined': { blocks: [{ type: 'image', source: { type: 'base64', data: png, mediaType: undefined } }], changed: true }, + 'a text block with an undefined field': { blocks: [{ type: 'text', text: 'x', note: undefined }], changed: true }, + // Only a non-URL image is blobbed: a URL image keeps even a stray data + // string as given, and storedBlock must not re-encode it. + 'an image by URL with a stray data string': { blocks: [{ type: 'image', source: { type: 'url', url: 'https://cdn.example/a.png', data: 'not base64 at all' } }], changed: false }, + }; + + it('every body hashes alike as handed to the store and as read back, live and after reopening', async () => { + const dir = mkdtempSync(join(tmpdir(), 'stored-digest-')); + const path = join(dir, 'store.chronicle'); + try { + const strategy = () => new PassthroughStrategy(); + const ids = new Map(); + let cm = await ContextManager.open({ path, strategy: strategy() }); + for (const [name, { blocks }] of Object.entries(bodies)) { + ids.set(name, cm.addMessage('user', blocks as never, { storedBodyDigest: sourceBodyDigest(blocks) })); + } + const check = (when: string) => { + for (const [name, { blocks, changed }] of Object.entries(bodies)) { + const back = cm.getMessage(ids.get(name)!)!; + if (changed) assert.notDeepEqual(back.content, blocks, `${when}, ${name}: the store hands it back changed`); + else assert.deepEqual(back.content, blocks, `${when}, ${name}: the store hands it back as it was`); + assert.equal(sourceBodyDigest(back.content), (back.metadata as Record).storedBodyDigest, `${when}, ${name}: it still hashes as handed over`); + } + }; + check('live'); + cm.close(); + cm = await ContextManager.open({ path, strategy: strategy() }); + check('reopened'); + cm.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a block the store refuses is refused at write, and a blob_ref is digested as given, which its copy read back does not match', async () => { + const dir = mkdtempSync(join(tmpdir(), 'stored-digest-refused-')); + try { + const cm = await ContextManager.open({ path: join(dir, 'store.chronicle'), strategy: new PassthroughStrategy() }); + try { + const svg = Buffer.from('').toString('base64'); + for (const [what, block] of [ + ['a document without a media type', { type: 'document', source: { type: 'base64', data: 'JVBERi0xLjQ=' } }], + ['an image without a media type whose bytes have no signature', { type: 'image', source: { type: 'base64', data: svg } }], + // The store's append can't represent an undefined array element. + ['an array with an undefined element', { type: 'text', text: 'x', parts: ['a', undefined] }], + ] as const) { + assert.throws(() => cm.addMessage('user', [block] as never), Error, `${what}: refused at write`); + } + // A blob_ref names media the store already holds; it is resolved to + // inline media on read, so the copy read back hashes differently. + const held = cm.addMessage('user', [image(png)] as never); + const ref = cm.getMessageWindow(0, cm.getMessageCount(), { resolveBlobs: false }).messages + .find((m) => m.id === held)!.content[0] as { type: string }; + assert.equal(ref.type, 'blob_ref'); + const given = [ref]; + const id = cm.addMessage('user', given as never, { storedBodyDigest: sourceBodyDigest(given) }); + const back = cm.getMessage(id)!; + assert.deepEqual(back.content, [image(png)], 'resolved on read'); + assert.notEqual(sourceBodyDigest(back.content), (back.metadata as Record).storedBodyDigest, 'the documented mismatch'); + } finally { + cm.close(); + } + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('a block the store keeps as given is digested as given', () => { + const plain = (blocks: unknown[]) => createHash('sha256').update(canonicalJson([blocks])).digest('hex'); + // Only a non-URL image is blobbed, so a URL image's stray data string is + // neither decoded nor re-encoded. + const urlImage = { type: 'image', source: { type: 'url', url: 'https://cdn.example/a.png', data: 'not base64 at all' } }; + assert.equal(sourceBodyDigest([urlImage]), plain([urlImage])); + assert.equal(sourceBodyDigest([{ type: 'text', text: 'x' }]), plain([{ type: 'text', text: 'x' }])); + // An undefined array element is written as null, as JSON writes it (the + // store refuses such a body, above, so no stored copy holds one). + assert.equal( + sourceBodyDigest([{ type: 'text', text: 'x', parts: ['a', undefined] }]), + sourceBodyDigest([{ type: 'text', text: 'x', parts: ['a', null] }]), + ); + }); + + it('what the store keeps still changes the digest', () => { + const digest = (blocks: unknown[]) => sourceBodyDigest(blocks); + assert.notEqual(digest([image(png)]), digest([image(Buffer.from([...PNG, 0]).toString('base64'))]), 'other bytes'); + const svg = Buffer.from('').toString('base64'); + assert.notEqual(digest([image(svg, 'image/svg+xml')]), digest([image(svg, 'image/png')]), 'a label the store keeps'); + assert.notEqual(digest([media('audio', 'SUQzAwAAAAAA', 'audio/mpeg')]), digest([media('audio', 'SUQzAwAAAAAA', 'audio/wav')]), 'an audio label'); + assert.notEqual(digest([{ type: 'text', text: 'a' }]), digest([{ type: 'text', text: 'b' }]), 'other text'); + assert.equal(digest([image(png, 'image/jpeg')]), digest([image(png)]), 'a label the store replaces is not part of the body'); + }); +}); diff --git a/tsconfig.json b/tsconfig.json index 476d44b30..d67d56a9f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -3,7 +3,7 @@ "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", - "lib": ["ES2022"], + "lib": ["ES2022", "ES2024.String"], "types": ["node"], "outDir": "./dist", "rootDir": ".",