refactor(storage): decode canonical prefixes on the SQLite worker

Decode borrowed headers and original projection slices before advancing each
SQLite row. Remove the temporary body string, transfer queue and independent
decoder while preserving the original snapshot, budgets and raw evidence.

Align the cleanup scheduling test with upstream's readable Root contract,
retaining sibling drainage, confirmed outcomes and failed shutdown evidence.

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 prepare the header and original projection bytes on the SQLite worker, then decode owned values on a joined blocking task while reading continues. They avoid a full temporary JSON string and keep the original transaction until both stages finish, including error and panic paths. The transfer contains only the prefix admitted by the existing event/byte budgets; no SQLite borrow crosses to the decoder. 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 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.
 
 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 和原始投影字节，由等待完成的阻塞任务解码已拥有的数据，同时继续读取。它避免构造完整临时 JSON 字符串；包括错误和 panic 在内，都等待两段结束后才结束原事务。传递范围仅限已通过原有事件／字节预算的前缀，不把 SQLite 借用交给解码任务。Header 复用事实字段的类型和默认值、忽略未知元数据，并验证投影占位符恰好位于已存偏移。Root 证据把原始三段字节作为一个带长度前缀的 JSON 字段计算摘要；元数据谓词继续读取 header 并使用原索引。语义区间保留原规范坐标，工具区间映射至实际 header／body，内联 model part 则保留 SQLite BLOB 的物理边界。读取仍由原 SQLite worker 执行。热区间读取复用其已有 statement cache。正文引用的存活由既有历史所有者回收证明决定，最后一个引用清除或删除时，在同一事务删除正文。事件行原子发布 header、逻辑长度、偏移和引用，存活引用不可重定向；正文摘要和派生目录都不授予访问权，也不延长存活期。归档占位内容和其他片段正文继续使用分块存储。
+完整事件查询在原快照内的同一 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、逻辑长度、偏移和引用，存活引用不可重定向；正文摘要和派生目录都不授予访问权，也不延长存活期。归档占位内容和其他片段正文继续使用分块存储。
 
 语义行、payload 和子项目录是有限的派生描述符，采用 Postcard 编码。writer 在发布事务内编码一次；分页把已存字节和固定小端序定位字段直接复制到调用者计费的缓冲区，消费端恢复原有类型，无需读取整个正文。原始字节位置与叶子摘要保持不变，目录编码不参与规范事实或来源身份。SQL 迁移在同一事务内转换原 JSON 描述符；转换函数仅在迁移期间存在。
 

platform/event-log/src/canonical.rs

@@ -25,8 +25,7 @@ mod recipe;
 mod source;
 mod write;
 
-pub(crate) use event::PreparedEvent;
-pub(crate) use read::{COLUMNS, JOINS, json, json_on_handle, one, prepare_event_on_handle};
+pub(crate) use read::{COLUMNS, JOINS, decode_event_on_handle, json, json_on_handle, one};
 pub(crate) use recipe::projection;
 pub(crate) use source::{
     SPAN_COLUMNS, SpanMetadata, with_span, with_span_metadata, with_span_on_row,

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

@@ -53,64 +53,7 @@ struct HeaderOutcome<'a> {
     artifacts: Vec<ArtifactEvidence>,
 }
 
-pub(crate) struct PreparedEvent(Prepared);
-
-enum Prepared {
-    Inline(String),
-    Tool {
-        header: Box<ToolHeader>,
-        projection: String,
-    },
-}
-struct ToolHeader {
-    id: String,
-    recorded_at: SystemTime,
-    invocation: Invocation,
-    operation_id: String,
-    raw: RawToolResultRef,
-    artifacts: Vec<ArtifactEvidence>,
-}
-
-impl PreparedEvent {
-    pub(super) fn inline(json: &str) -> Self {
-        Self(Prepared::Inline(json.to_owned()))
-    }
-    pub(crate) fn decode(self) -> Result<RuntimeEvent, StoreError> {
-        match self.0 {
-            Prepared::Inline(json) => Ok(serde_json::from_str(&json)?),
-            Prepared::Tool { header, projection } => {
-                let model_projection: DurableToolProjection = serde_json::from_str(&projection)?;
-                let ToolHeader {
-                    id,
-                    recorded_at,
-                    invocation,
-                    operation_id,
-                    raw,
-                    artifacts,
-                } = *header;
-                Ok(RuntimeEvent {
-                    id,
-                    recorded_at,
-                    invocation,
-                    fact: Fact::ToolSettled {
-                        operation_id,
-                        outcome: ToolOutcome::Succeeded {
-                            raw,
-                            model_projection,
-                            artifacts,
-                        },
-                    },
-                })
-            }
-        }
-    }
-}
-
-pub(super) fn prepare(
-    header: &str,
-    offset: usize,
-    body: &[u8],
-) -> Result<PreparedEvent, StoreError> {
+pub(super) fn decode(header: &str, offset: usize, body: &[u8]) -> Result<RuntimeEvent, StoreError> {
     let Header {
         id,
         recorded_at,
@@ -142,17 +85,20 @@ pub(super) fn prepare(
     }
     let body = std::str::from_utf8(body)
         .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
-    Ok(PreparedEvent(Prepared::Tool {
-        header: Box::new(ToolHeader {
-            id,
-            recorded_at,
-            invocation,
+    let model_projection: DurableToolProjection = serde_json::from_str(body)?;
+    Ok(RuntimeEvent {
+        id,
+        recorded_at,
+        invocation,
+        fact: Fact::ToolSettled {
             operation_id,
-            raw,
-            artifacts,
-        }),
-        projection: body.to_owned(),
-    }))
+            outcome: ToolOutcome::Succeeded {
+                raw,
+                model_projection,
+                artifacts,
+            },
+        },
+    })
 }
 
 #[cfg(test)]
@@ -205,8 +151,7 @@ mod tests {
                 }),
             )
             .unwrap();
-            let (prepared, parts) = recipe.prepare(Some(projection.json.as_bytes())).unwrap();
-            let event = prepared.decode().unwrap();
+            let (event, parts) = recipe.decode(Some(projection.json.as_bytes())).unwrap();
             assert_eq!(event, expected);
             assert_eq!(parts.concat(), json.as_bytes());
         }
@@ -230,7 +175,7 @@ mod tests {
             }),
         )
         .unwrap();
-        assert!(recipe.prepare(Some(projection.json.as_bytes())).is_err());
+        assert!(recipe.decode(Some(projection.json.as_bytes())).is_err());
     }
 
     #[test]
@@ -252,11 +197,7 @@ mod tests {
         ] {
             let position = changed.find("\"model_projection\": null").unwrap()
                 + "\"model_projection\": ".len();
-            assert!(
-                prepare(&changed, position, projection.json.as_bytes())
-                    .and_then(PreparedEvent::decode)
-                    .is_err()
-            );
+            assert!(decode(&changed, position, projection.json.as_bytes()).is_err());
         }
     }
 }

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

@@ -126,12 +126,12 @@ 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 prepare_event_on_handle<'a>(
+pub(crate) fn decode_event_on_handle<'a>(
     row: &'a rusqlite::Row<'_>,
     visit: impl FnOnce([&'a [u8]; 3]),
-) -> Result<super::event::PreparedEvent, StoreError> {
+) -> Result<maka_agent_facts::event::RuntimeEvent, StoreError> {
     let (recipe, body) = material(fields_on_handle(row)?)?;
-    let (event, parts) = recipe.prepare(body)?;
+    let (event, parts) = recipe.decode(body)?;
     visit(parts);
     Ok(event)
 }

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

@@ -244,13 +244,13 @@ impl<'a> Recipe<'a> {
             Ok(())
         })
     }
-    pub fn prepare(
+    pub fn decode(
         &self,
         body: Option<&'a [u8]>,
-    ) -> Result<(super::event::PreparedEvent, [&'a [u8]; 3]), StoreError> {
+    ) -> Result<(maka_agent_facts::event::RuntimeEvent, [&'a [u8]; 3]), StoreError> {
         match (self.layout.slot, body) {
             (None, None) => Ok((
-                super::event::PreparedEvent::inline(self.header),
+                serde_json::from_str(self.header)?,
                 [self.header.as_bytes(), &[], &[]],
             )),
             (Some(slot), Some(body)) => {
@@ -259,7 +259,7 @@ impl<'a> Recipe<'a> {
                 }
                 let position = usize::try_from(slot.offset)
                     .map_err(|_| invalid("slot offset exceeds memory range"))?;
-                let event = super::event::prepare(self.header, position, body)?;
+                let event = super::event::decode(self.header, position, body)?;
                 Ok((
                     event,
                     [

platform/event-log/src/prefix.rs

@@ -19,12 +19,11 @@
 
 use maka_agent_facts::event::LogPrefix;
 use maka_agent_facts::event::LogScope;
+use maka_agent_facts::event::StoredEvent;
 use sqlx::Connection;
 
 use crate::{EventLog, StoreError, context::selection::Selection, sequence_number};
 
-mod decode;
-
 impl EventLog {
     /// Read one complete bounded execution prefix, or fail without returning a partial
     /// history that a caller could accidentally treat as complete. Session-owned
@@ -86,28 +85,31 @@ impl EventLog {
         let root_scope = session.is_none();
         let (events, root_digest) = if count == 0 {
             (Vec::new(), root_scope.then(crate::context::evidence::RootDigest::new))
-        } else { decode::run(count as usize, |sender| crate::connection::worker_query(&mut transaction, move |worker| Box::pin(async move {
+        } else { crate::connection::worker_query(&mut transaction, move |worker| Box::pin(async move {
             // SAFETY: this callback exclusively owns the original SQLite worker.
             // The wrapper and all borrowed rows/statements close before return.
             let raw = unsafe { rusqlite::Connection::from_handle(worker.as_raw_handle().as_ptr()) }?;
             let result = (|| {
                 let mut digest = session.is_none().then(crate::context::evidence::RootDigest::new);
+                let mut events = Vec::new();
                 let mut statement = raw.prepare(&query)?;
                 let mut rows = statement.query(rusqlite::params![session, selected_lineage])?;
                 while let Some(row) = rows.next()? {
                     let sequence: i64 = row.get("sequence")?;
-                    let prepared = crate::canonical::prepare_event_on_handle(row, |parts| {
+                    let event = crate::canonical::decode_event_on_handle(row, |parts| {
                         if let Some(digest) = &mut digest { digest.record(sequence, &parts); }
                     })?;
-                    if sender.send((sequence_number(sequence)?, prepared)).is_err() {
-                        return Ok(decode::Produced::DecoderStopped);
-                    }
+                    events.push(StoredEvent { sequence: sequence_number(sequence)?, event });
+                }
+                if events.len() as u64 != sequence_number(count)? {
+                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidData,
+                        "prefix decoder stopped before completing the admitted snapshot").into());
                 }
-                Ok(decode::Produced::Complete(digest))
+                Ok((events, digest))
             })();
             drop(raw);
             result
-        }))).await? };
+        })).await? };
         let digest = match root_digest {
             Some(digest) => digest.finish(high_water)?,
             None => crate::context::evidence::selected(