refactor(storage): keep event bodies out of prefix sort records

Order canonical headers and references before resolving bodies by primary key
on the same SQLite worker and transaction. Reuse one body statement while
decoding borrowed rows and recording original canonical evidence. Reject
missing bodies without publishing partial canonical bytes.

Generated-by: Codex

Author: M4n5ter

platform/event-log/README.md

@@ -73,7 +73,7 @@ Canonical databases use UTF-8. Successful tool results store their original `mod
 
 Bodies are directly seekable raw bytes, shared by SHA-256 of their original token with exact byte comparison on a repeated identity. Reads validate logical lengths and selected ranges. Framed compression was evaluated against raw storage and rejected because it repeatedly slowed supported window and sequential reads; it is not a persisted format. Space savings come from compact directories and sharing identical projection tokens, not from compressing unique bodies.
 
-Complete-event queries join the referenced body into the selected SQL row within the original snapshot. JSON readers reconstruct the original bytes; typed prefix readers decode the borrowed header and original projection bytes directly on the original SQLite worker. They avoid copying the body into a temporary string and finish decoding before advancing its row. The existing event/byte preflight and transaction cover the complete result; no SQLite borrow or unfinished decoder leaves that worker. Typed headers reuse fact field types and defaults, ignore unknown metadata, and verify that the projection placeholder is exactly at its stored offset. Root evidence hashes the three original byte slices as one length-prefixed JSON field. Metadata predicates use the header and retain their indexes. Semantic spans retain original canonical coordinates; tool spans map those logical coordinates to header/body ranges, while inline model parts retain physical SQLite BLOB bounds. Resolution stays on the original SQLite worker. Hot range reads use its existing prepared-statement cache. Body references follow the existing history-owner collection proof; clearing or deleting the final reference removes its body in the same transaction. The event row publishes the header, logical length, offset and reference atomically; live references cannot be redirected. Neither a body digest nor a derived directory grants access or extends lifetime. Derived archive placeholders and other fragment bodies keep their chunked storage.
+Complete-event JSON queries join the referenced body into the selected SQL row within the original snapshot and reconstruct the original bytes. Typed prefix readers order headers and body references, keeping large payloads out of SQLite's sort records. One prepared primary-key lookup resolves each referenced body on the same connection and transaction; inline events require no body lookup. Readers decode the borrowed header and original projection bytes directly on the original SQLite worker. They avoid copying the body into a temporary string and finish decoding before advancing either row. The existing event/byte preflight and transaction cover the complete result; no SQLite borrow or unfinished decoder leaves that worker. Typed headers reuse fact field types and defaults, ignore unknown metadata, and verify that the projection placeholder is exactly at its stored offset. Root evidence hashes the three original byte slices as one length-prefixed JSON field. Metadata predicates use the header and retain their indexes. Semantic spans retain original canonical coordinates; tool spans map those logical coordinates to header/body ranges, while inline model parts retain physical SQLite BLOB bounds. Resolution stays on the original SQLite worker. Hot range reads use its existing prepared-statement cache. Body references follow the existing history-owner collection proof; clearing or deleting the final reference removes its body in the same transaction. The event row publishes the header, logical length, offset and reference atomically; live references cannot be redirected. Neither a body digest nor a derived directory grants access or extends lifetime. Derived archive placeholders and other fragment bodies keep their chunked storage.
 
 Semantic row, payload and child directories are finite derived descriptors encoded with Postcard. Writers encode them once in the publication transaction. Pages copy their stored bytes alongside fixed little-endian locator fields into the caller's charged allocation; consumers decode the original typed descriptors without navigating whole bodies. Original byte positions and leaf commitments remain unchanged. Directory encoding never supplies a canonical fact or source identity. The SQL migration converts former JSON descriptors atomically; its conversion functions exist only during migration.
 

platform/event-log/README.zh-CN.md

@@ -67,7 +67,7 @@ Control 中的调用意图、项目登记、workspace attachment、Session 路
 
 正文保存为可直接 seek 的原始字节，按原始 token 的 SHA-256 共享，重复身份还需逐字节核对。读取校验逻辑长度与选中区间。对照实验中，分帧压缩稳定拖慢了受支持的窗口和顺序读取，因此未作为持久化格式保留。空间收益来自紧凑目录与相同投影 token 的共享，不能把压缩实验中独有正文的压缩率套到最终实现。
 
-完整事件查询在原快照内的同一 SQL 行 JOIN 引用的正文。JSON reader 重建原始字节；类型化前缀 reader 在原 SQLite worker 上直接解码借用的 header 和原始投影字节，避免把正文复制到临时字符串中，并在推进当前行之前完成解码。原有事件／字节预检和事务覆盖完整结果；SQLite 借用和未完成的解码都不会离开该 worker。Header 复用事实字段的类型和默认值、忽略未知元数据，并验证投影占位符恰好位于已存偏移。Root 证据把原始三段字节作为一个带长度前缀的 JSON 字段计算摘要；元数据谓词继续读取 header 并使用原索引。语义区间保留原规范坐标，工具区间映射至实际 header／body，内联 model part 则保留 SQLite BLOB 的物理边界。读取仍由原 SQLite worker 执行。热区间读取复用其已有 statement cache。正文引用的存活由既有历史所有者回收证明决定，最后一个引用清除或删除时，在同一事务删除正文。事件行原子发布 header、逻辑长度、偏移和引用，存活引用不可重定向；正文摘要和派生目录都不授予访问权，也不延长存活期。归档占位内容和其他片段正文继续使用分块存储。
+完整事件的 JSON 查询在原快照内的同一 SQL 行 JOIN 引用的正文，并重建原始字节。类型化前缀 reader 只排列 header 和正文引用，避免大正文进入 SQLite 排序记录；随后在同一连接、同一事务内复用一条 prepared 主键查询读取引用的正文，内联事件无需查询正文。Reader 在原 SQLite worker 上直接解码借用的 header 和原始投影字节，避免把正文复制到临时字符串中，并在推进任一行之前完成解码。原有事件／字节预检和事务覆盖完整结果；SQLite 借用和未完成的解码都不会离开该 worker。Header 复用事实字段的类型和默认值、忽略未知元数据，并验证投影占位符恰好位于已存偏移。Root 证据把原始三段字节作为一个带长度前缀的 JSON 字段计算摘要；元数据谓词继续读取 header 并使用原索引。语义区间保留原规范坐标，工具区间映射至实际 header／body，内联 model part 则保留 SQLite BLOB 的物理边界。读取仍由原 SQLite worker 执行。热区间读取复用其已有 statement cache。正文引用的存活由既有历史所有者回收证明决定，最后一个引用清除或删除时，在同一事务删除正文。事件行原子发布 header、逻辑长度、偏移和引用，存活引用不可重定向；正文摘要和派生目录都不授予访问权，也不延长存活期。归档占位内容和其他片段正文继续使用分块存储。
 
 语义行、payload 和子项目录是有限的派生描述符，采用 Postcard 编码。writer 在发布事务内编码一次；分页把已存字节和固定小端序定位字段直接复制到调用者计费的缓冲区，消费端恢复原有类型，无需读取整个正文。原始字节位置与叶子摘要保持不变，目录编码不参与规范事实或来源身份。SQL 迁移在同一事务内转换原 JSON 描述符；转换函数仅在迁移期间存在。
 

platform/event-log/src/canonical/read.rs

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-//! Complete canonical reads join one immutable body into the same selected row.
+//! Canonical reads resolve immutable bodies within the selected event snapshot.
 use super::recipe::{Recipe, Slot};
 use crate::StoreError;
 use sqlx::{Row, sqlite::SqliteRow};
@@ -126,16 +126,30 @@ pub(crate) fn json(row: &SqliteRow) -> Result<Cow<'_, str>, StoreError> {
 pub(crate) fn json_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Cow<'a, str>, StoreError> {
     reconstruct(fields_on_handle(row)?)
 }
-pub(crate) fn decode_event_on_handle<'a>(
-    row: &'a rusqlite::Row<'_>,
-    visit: impl FnOnce([&'a [u8]; 3]),
+pub(crate) fn decode_event_on_handle(
+    row: &rusqlite::Row<'_>,
+    bodies: &mut rusqlite::Statement<'_>,
+    visit: impl FnOnce([&[u8]; 3]),
 ) -> Result<maka_agent_facts::event::RuntimeEvent, StoreError> {
-    let (recipe, body) = material(fields_on_handle(row)?)?;
+    let mut fields = header_fields_on_handle(row)?;
+    let mut body_rows = match fields.reference {
+        Some(reference) => Some(bodies.query([reference])?),
+        None => None,
+    };
+    if let Some(rows) = &mut body_rows {
+        let body = rows
+            .next()?
+            .ok_or_else(|| invalid("canonical body is missing"))?;
+        fields.body = body.get("canonical_body_id")?;
+        fields.raw_bytes = body.get("canonical_raw_bytes")?;
+        fields.payload = payload_on_handle(body)?;
+    }
+    let (recipe, body) = material(fields)?;
     let (event, parts) = recipe.decode(body)?;
     visit(parts);
     Ok(event)
 }
-fn fields_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Fields<'a>, StoreError> {
+fn header_fields_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Fields<'a>, StoreError> {
     let header = row
         .get_ref("event_header")?
         .as_str_or_null()
@@ -150,18 +164,22 @@ fn fields_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Fields<'a>, StoreE
         })?;
     let length = row.get("canonical_length")?;
     let reference: Option<i64> = row.get("canonical_body_reference")?;
-    if reference.is_none() {
-        return Ok(Fields {
-            header,
-            length,
-            reference,
-            offset: None,
-            body: None,
-            raw_bytes: None,
-            payload: None,
-        });
-    }
-    let payload = row
+    Ok(Fields {
+        header,
+        length,
+        reference,
+        offset: if reference.is_some() {
+            row.get("canonical_offset")?
+        } else {
+            None
+        },
+        body: None,
+        raw_bytes: None,
+        payload: None,
+    })
+}
+fn payload_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Option<&'a [u8]>, StoreError> {
+    Ok(row
         .get_ref("canonical_payload")?
         .as_blob_or_null()
         .map_err(|error| {
@@ -172,16 +190,16 @@ fn fields_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Fields<'a>, StoreE
                 rusqlite::types::Type::Blob,
                 Box::new(error),
             )
-        })?;
-    Ok(Fields {
-        header,
-        length,
-        reference,
-        offset: row.get("canonical_offset")?,
-        body: row.get("canonical_body_id")?,
-        raw_bytes: row.get("canonical_raw_bytes")?,
-        payload,
-    })
+        })?)
+}
+fn fields_on_handle<'a>(row: &'a rusqlite::Row<'_>) -> Result<Fields<'a>, StoreError> {
+    let mut fields = header_fields_on_handle(row)?;
+    if fields.reference.is_some() {
+        fields.body = row.get("canonical_body_id")?;
+        fields.raw_bytes = row.get("canonical_raw_bytes")?;
+        fields.payload = payload_on_handle(row)?;
+    }
+    Ok(fields)
 }
 pub(crate) async fn one(
     connection: &mut sqlx::SqliteConnection,
@@ -199,6 +217,33 @@ pub(crate) async fn one(
 #[cfg(test)]
 mod tests {
     use super::*;
+    #[test]
+    fn missing_body_lookup_fails_without_visiting_partial_canonical_bytes() {
+        let db = rusqlite::Connection::open_in_memory().unwrap();
+        let mut headers = db
+            .prepare(
+                "SELECT 'null' AS event_header,4 AS canonical_length,
+             1 AS canonical_body_reference,0 AS canonical_offset",
+            )
+            .unwrap();
+        let mut bodies = db
+            .prepare(
+                "SELECT 1 AS canonical_body_id,4 AS canonical_raw_bytes,
+             X'6E756C6C' AS canonical_payload WHERE ?1 IS NULL AND 0",
+            )
+            .unwrap();
+        let mut rows = headers.query([]).unwrap();
+        let row = rows.next().unwrap().unwrap();
+        let error = decode_event_on_handle(row, &mut bodies, |_| {
+            panic!("missing body must not publish canonical bytes");
+        })
+        .unwrap_err();
+        assert!(
+            error.to_string().contains("canonical body is missing"),
+            "{error}"
+        );
+    }
+
     #[test]
     fn corrupt_large_length_is_rejected_before_reserving_the_event() {
         let result = reconstruct(Fields {

platform/event-log/src/prefix.rs

@@ -77,8 +77,9 @@ impl EventLog {
             return Err(StoreError::PrefixTooLarge);
         }
         let query = format!(
-            "SELECT e.sequence,e.event_header,{columns} FROM {source} e {joins} WHERE {filter} ORDER BY e.sequence",
-            columns=crate::canonical::COLUMNS,joins=crate::canonical::JOINS
+            "SELECT e.sequence,e.event_header,e.canonical_bytes AS canonical_length,
+             e.body_id AS canonical_body_reference,e.header_offset AS canonical_offset
+             FROM {source} e WHERE {filter} ORDER BY e.sequence"
         );
         let session = selection.session.clone();
         let selected_lineage = selection.lineage.clone();
@@ -93,10 +94,16 @@ impl EventLog {
                 let mut digest = session.is_none().then(crate::context::evidence::RootDigest::new);
                 let mut events = Vec::new();
                 let mut statement = raw.prepare(&query)?;
+                // Keep large body payloads out of the ordered event records.
+                // Both statements borrow the original transaction's snapshot.
+                let mut bodies = raw.prepare(
+                    "SELECT id AS canonical_body_id,raw_bytes AS canonical_raw_bytes,
+                     payload AS canonical_payload FROM event_bodies WHERE id=?"
+                )?;
                 let mut rows = statement.query(rusqlite::params![session, selected_lineage])?;
                 while let Some(row) = rows.next()? {
                     let sequence: i64 = row.get("sequence")?;
-                    let event = crate::canonical::decode_event_on_handle(row, |parts| {
+                    let event = crate::canonical::decode_event_on_handle(row, &mut bodies, |parts| {
                         if let Some(digest) = &mut digest { digest.record(sequence, &parts); }
                     })?;
                     events.push(StoredEvent { sequence: sequence_number(sequence)?, event });