feat(mcpl): stamp each inbound item's delivered-body and stored-copy digests

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.

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<string, unknown>,
+    content: readonly ContentBlock[],
+  ): Record<string, unknown> {
+    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<string, unknown>;
       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 ?? '')}`);

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,

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<string, unknown>)
+      .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');
+}

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,

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') {