You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/10/22 07:40:22 UTC
(maka) branch rust-preview updated: perf(event-log): keep transcript bodies in a private reclaimable cache
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new c3167c09e3 perf(event-log): keep transcript bodies in a private reclaimable cache
c3167c09e3 is described below
commit c3167c09e3fc024b3e5275ce86a2ac95305b99ae
Author: Wang <m4...@gmail.com>
AuthorDate: Thu Oct 22 15:38:27 2026 +0800
perf(event-log): keep transcript bodies in a private reclaimable cache
---
.../0038_private_transcript_cache.sql} | 24 +-
platform/event-log/src/lib.rs | 28 ++
platform/event-log/src/schema.rs | 1 +
platform/event-log/src/sessions/copy/lifecycle.rs | 4 +-
platform/event-log/src/sessions/material.rs | 2 +-
platform/event-log/src/sessions/removal.rs | 4 +-
platform/event-log/src/transcript.rs | 1 +
platform/event-log/src/transcript/bodycache.rs | 118 ++++++++-
.../event-log/src/transcript/bodycache_tests.rs | 294 +++++++++++++++++++++
platform/event-log/src/transcript/directory.rs | 12 +-
platform/event-log/src/transcript/history.rs | 66 ++---
platform/event-log/src/transcript/image.rs | 30 +--
platform/event-log/src/transcript/navigation.rs | 32 ++-
platform/event-log/src/transcript/prompt.rs | 14 +-
platform/event-log/src/transcript/read.rs | 45 ++--
platform/event-log/src/transcript/search.rs | 19 +-
platform/event-log/src/transcript/source.rs | 17 +-
.../tests/compact_presentation/automatic.rs | 21 +-
.../tests/compact_presentation/fixtures.rs | 2 +-
.../mod.rs} | 27 +-
.../mod.rs} | 42 +--
platform/event-log/tests/migrations.rs | 39 +--
.../event-log/tests/migrations/prior_artifacts.rs | 4 +-
platform/event-log/tests/navigation.rs | 23 +-
platform/event-log/tests/root_messages.rs | 3 +-
.../{row_directory.rs => row_directory/mod.rs} | 15 +-
.../tests/{session_copy.rs => session_copy/mod.rs} | 25 +-
platform/event-log/tests/steering.rs | 2 +-
.../{src/transcript.rs => tests/support/owner.rs} | 43 +--
.../transcript_index/accepted_observations.rs | 46 +++-
.../event-log/tests/transcript_index/fixtures.rs | 2 +-
.../event-log/tests/transcript_index/history.rs | 157 +++++++----
platform/event-log/tests/transcript_index/image.rs | 3 +-
.../tests/transcript_index/interrupted_ancestry.rs | 12 +-
.../tests/transcript_index/large_siblings.rs | 19 +-
.../mod.rs} | 190 +++++++------
.../tests/transcript_index/observation_count.rs | 8 +-
.../tests/transcript_index/performance.rs | 10 +-
.../tests/transcript_index/performance_executor.rs | 4 +-
.../tests/transcript_index/prefix_replay.rs | 65 +++--
.../tests/transcript_index/sparse_directory.rs | 27 +-
.../mod.rs} | 39 +--
.../event-log/tests/transcript_tools/payloads.rs | 106 +++++---
43 files changed, 1159 insertions(+), 486 deletions(-)
diff --git a/platform/event-log/src/transcript.rs b/platform/event-log/migrations/0038_private_transcript_cache.sql
similarity index 66%
copy from platform/event-log/src/transcript.rs
copy to platform/event-log/migrations/0038_private_transcript_cache.sql
index 5447ba2870..b2cf339229 100644
--- a/platform/event-log/src/transcript.rs
+++ b/platform/event-log/migrations/0038_private_transcript_cache.sql
@@ -17,24 +17,6 @@
* under the License.
*/
-//! Durable row metadata and optional body caches; authority stays in the canonical ledger.
-mod active;
-mod compaction;
-pub(crate) use active::interrupted_messages;
-mod bodycache;
-pub(crate) mod directory;
-mod emit;
-mod evidence;
-mod final_output;
-mod history;
-mod image;
-mod model;
-pub mod navigation;
-mod prompt;
-mod read;
-pub mod search;
-mod source;
-mod timing;
-mod tool_metadata;
-mod tools;
-pub use read::*;
+-- Bodies are disposable connection-private projections; canonical metadata remains.
+DROP TABLE transcript_text;
+DROP TABLE transcript_rows;
diff --git a/platform/event-log/src/lib.rs b/platform/event-log/src/lib.rs
index 01ff6a4ba3..fa189afeca 100644
--- a/platform/event-log/src/lib.rs
+++ b/platform/event-log/src/lib.rs
@@ -380,3 +380,31 @@ fn sequence_number(value: i64) -> Result<u64, StoreError> {
u64::try_from(value)
.map_err(|_| StoreError::InvalidTransition("negative ledger sequence".into()))
}
+
+// Cache fault injection must run on the original private-cache owner.
+#[cfg(test)]
+extern crate self as maka_event_log;
+#[cfg(test)]
+#[path = "../tests/compact_presentation/mod.rs"]
+mod compact_presentation_tests;
+#[cfg(test)]
+#[path = "../tests/executor_identity/mod.rs"]
+mod executor_identity_tests;
+#[cfg(test)]
+#[path = "../tests/row_directory/mod.rs"]
+mod row_directory_tests;
+#[cfg(test)]
+#[path = "../tests/session_copy/mod.rs"]
+mod session_copy_tests;
+#[cfg(test)]
+#[path = "../tests/support/owner.rs"]
+mod test_store;
+#[cfg(test)]
+#[path = "../tests/support/transcript.rs"]
+mod transcript_cache;
+#[cfg(test)]
+#[path = "../tests/transcript_index/mod.rs"]
+mod transcript_index_tests;
+#[cfg(test)]
+#[path = "../tests/transcript_tools/mod.rs"]
+mod transcript_tools_tests;
diff --git a/platform/event-log/src/schema.rs b/platform/event-log/src/schema.rs
index a475c94435..a637b91b17 100644
--- a/platform/event-log/src/schema.rs
+++ b/platform/event-log/src/schema.rs
@@ -87,6 +87,7 @@ pub(crate) async fn initialize_connection(
.await?;
crate::message_sources::initialize(connection).await?;
crate::sessions::initialize_execution(connection).await?;
+ crate::transcript::initialize_cache(connection).await?;
Ok(())
}
diff --git a/platform/event-log/src/sessions/copy/lifecycle.rs b/platform/event-log/src/sessions/copy/lifecycle.rs
index ed4262f959..a7e2ea91d3 100644
--- a/platform/event-log/src/sessions/copy/lifecycle.rs
+++ b/platform/event-log/src/sessions/copy/lifecycle.rs
@@ -83,8 +83,8 @@ impl EventLog {
"DELETE FROM session_history_artifacts WHERE session_id=?",
"DELETE FROM session_history_members WHERE session_id=?",
"DELETE FROM session_revision_sources WHERE session_id=?",
- "DELETE FROM transcript_rows WHERE session_id=?",
- "DELETE FROM transcript_text WHERE session_id=?",
+ "DELETE FROM transcript_cache.rows WHERE session_id=?",
+ "DELETE FROM transcript_cache.text WHERE session_id=?",
"DELETE FROM transcript_directory WHERE session_id=?",
"DELETE FROM transcript_failures WHERE session_id=?",
"DELETE FROM catalog_messages WHERE session_id=?",
diff --git a/platform/event-log/src/sessions/material.rs b/platform/event-log/src/sessions/material.rs
index 8aaa64e1c7..8f2331f60f 100644
--- a/platform/event-log/src/sessions/material.rs
+++ b/platform/event-log/src/sessions/material.rs
@@ -99,7 +99,7 @@ impl EventLog {
}
// No live owner remains. Retained facts/membership, not UI metadata,
// remain the material-collection authority; descendants were checked above.
- for table in ["transcript_directory", "transcript_failures", "transcript_rows", "transcript_text"] {
+ for table in ["transcript_directory", "transcript_failures", "transcript_cache.rows", "transcript_cache.text"] {
let sql = format!("DELETE FROM {table} WHERE session_id=?");
sqlx::query(sqlx::AssertSqlSafe(sql.as_str())).bind(&session).execute(&mut *tx).await?;
}
diff --git a/platform/event-log/src/sessions/removal.rs b/platform/event-log/src/sessions/removal.rs
index 7011ebf052..abb62c371f 100644
--- a/platform/event-log/src/sessions/removal.rs
+++ b/platform/event-log/src/sessions/removal.rs
@@ -258,8 +258,8 @@ impl EventLog {
for statement in [
"DELETE FROM session_history_artifacts WHERE session_id=?",
"DELETE FROM session_revision_sources WHERE session_id=?",
- "DELETE FROM transcript_rows WHERE session_id=?",
- "DELETE FROM transcript_text WHERE session_id=?",
+ "DELETE FROM transcript_cache.rows WHERE session_id=?",
+ "DELETE FROM transcript_cache.text WHERE session_id=?",
"DELETE FROM transcript_directory WHERE session_id=?",
"DELETE FROM transcript_failures WHERE session_id=?",
"DELETE FROM catalog_messages WHERE session_id=?",
diff --git a/platform/event-log/src/transcript.rs b/platform/event-log/src/transcript.rs
index 5447ba2870..d90e139dc0 100644
--- a/platform/event-log/src/transcript.rs
+++ b/platform/event-log/src/transcript.rs
@@ -22,6 +22,7 @@ mod active;
mod compaction;
pub(crate) use active::interrupted_messages;
mod bodycache;
+pub(crate) use bodycache::initialize as initialize_cache;
pub(crate) mod directory;
mod emit;
mod evidence;
diff --git a/platform/event-log/src/transcript/bodycache.rs b/platform/event-log/src/transcript/bodycache.rs
index 08295f59d3..aaacb82d5e 100644
--- a/platform/event-log/src/transcript/bodycache.rs
+++ b/platform/event-log/src/transcript/bodycache.rs
@@ -22,7 +22,111 @@ use super::read::sql_position;
use crate::StoreError;
use maka_agent_facts::history::RowPosition;
use sha2::{Digest, Sha256};
-use sqlx::SqliteConnection;
+use sqlx::{Connection, SqliteConnection};
+
+// The ordinary retained database is at most 32 MiB. One active producer may
+// exceed that target so tiny fragment/sibling reads do not repeatedly emit it.
+// This does not bound a complete operation's projection or live journal peak.
+const RETAINED_PAGES: i64 = 8192;
+
+pub(crate) async fn initialize(db: &mut SqliteConnection) -> Result<(), StoreError> {
+ sqlx::raw_sql(
+ "PRAGMA temp_store=FILE;
+ ATTACH DATABASE '' AS transcript_cache;
+ PRAGMA transcript_cache.page_size=4096;
+ PRAGMA transcript_cache.auto_vacuum=FULL;
+ PRAGMA transcript_cache.cache_size=-4096;
+ PRAGMA transcript_cache.synchronous=OFF;
+ PRAGMA transcript_cache.journal_mode=DELETE;
+ CREATE TABLE transcript_cache.rows (
+ session_id TEXT NOT NULL, source_sequence INTEGER NOT NULL CHECK(source_sequence>=1),
+ item_index INTEGER NOT NULL CHECK(item_index>=0), turn_id TEXT NOT NULL,
+ message_id TEXT NOT NULL, payload BLOB NOT NULL, digest TEXT NOT NULL,
+ total_bytes INTEGER NOT NULL CHECK(total_bytes>0),
+ PRIMARY KEY(session_id,source_sequence,item_index), UNIQUE(session_id,message_id)
+ );
+ CREATE TABLE transcript_cache.text (
+ session_id TEXT NOT NULL, source_sequence INTEGER NOT NULL CHECK(source_sequence>=1),
+ item_index INTEGER NOT NULL CHECK(item_index>=0),
+ field TEXT NOT NULL CHECK(field IN ('body','thinking')),
+ timestamp INTEGER NOT NULL, role TEXT NOT NULL, body BLOB NOT NULL,
+ PRIMARY KEY(session_id,source_sequence,item_index,field)
+ );
+ CREATE TABLE transcript_cache.active_source (
+ singleton INTEGER PRIMARY KEY CHECK(singleton=1),
+ session_id TEXT NOT NULL, source_sequence INTEGER NOT NULL
+ );",
+ )
+ .execute(&mut *db)
+ .await?;
+ Ok(())
+}
+
+/// Keep every ensured row pinned until the original operation has consumed it,
+/// including final history boundaries. Accepted work stays on the writer owner.
+pub(super) fn run<'c, T: Send + 'c>(
+ db: &'c mut SqliteConnection,
+ read: impl for<'a> FnOnce(
+ &'a mut SqliteConnection,
+ ) -> crate::connection::BoxFuture<'a, Result<T, StoreError>>
+ + Send
+ + 'c,
+) -> crate::connection::BoxFuture<'c, Result<T, StoreError>> {
+ Box::pin(async move {
+ let mut tx = db.begin().await?;
+ match read(&mut tx).await {
+ Ok(value) => {
+ tx.commit().await?;
+ trim(db).await?;
+ Ok(value)
+ }
+ Err(error) => {
+ // Preserve the canonical read failure. No maintenance follows an
+ // unsuccessful read or a rollback with an uncertain result.
+ let _ = tx.rollback().await;
+ Err(error)
+ }
+ }
+ })
+}
+
+async fn trim(db: &mut SqliteConnection) -> Result<(), StoreError> {
+ let pages: i64 = sqlx::query_scalar("PRAGMA transcript_cache.page_count")
+ .fetch_one(&mut *db)
+ .await?;
+ if pages <= RETAINED_PAGES {
+ return Ok(());
+ }
+ let mut tx = db.begin().await?;
+ sqlx::raw_sql(
+ "DELETE FROM transcript_cache.text WHERE NOT EXISTS (
+ SELECT 1 FROM transcript_cache.active_source a
+ WHERE a.session_id=text.session_id AND a.source_sequence=text.source_sequence
+ );
+ DELETE FROM transcript_cache.rows WHERE NOT EXISTS (
+ SELECT 1 FROM transcript_cache.active_source a
+ WHERE a.session_id=rows.session_id AND a.source_sequence=rows.source_sequence
+ );",
+ )
+ .execute(&mut *tx)
+ .await?;
+ // FULL auto-vacuum reclaims free database pages at this commit, including
+ // an opened temporary backing file. DELETE alone would retain high-water.
+ tx.commit().await?;
+ Ok(())
+}
+
+async fn retain_source(
+ db: &mut SqliteConnection,
+ session: &str,
+ source: i64,
+) -> Result<(), StoreError> {
+ sqlx::query("INSERT INTO transcript_cache.active_source VALUES(1,?1,?2)
+ ON CONFLICT(singleton) DO UPDATE SET session_id=excluded.session_id,source_sequence=excluded.source_sequence
+ WHERE active_source.session_id!=excluded.session_id OR active_source.source_sequence!=excluded.source_sequence")
+ .bind(session).bind(source).execute(db).await?;
+ Ok(())
+}
pub(super) async fn ensure(
db: &mut SqliteConnection,
@@ -35,12 +139,13 @@ pub(super) async fn ensure(
).bind(session).bind(source).bind(item).fetch_optional(&mut *db).await?
.ok_or_else(|| StoreError::InvalidTransition("history row was not found".into()))?;
let cached: Option<(String, String)> = sqlx::query_as(
- "SELECT turn_id,message_id FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3",
+ "SELECT turn_id,message_id FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3",
).bind(session).bind(source).bind(item).fetch_optional(&mut *db).await?;
if let Some(identity) = cached {
if identity != (turn, message) {
return Err(StoreError::TranscriptConflict);
}
+ retain_source(db, session, source).await?;
return Ok(());
}
// S is the cache unit: emission already owns every original row from this
@@ -62,13 +167,18 @@ pub(super) async fn ensure(
let payload = serde_json::to_vec(&row.message)?;
let bytes = i64::try_from(payload.len()).map_err(|_| StoreError::PrefixTooLarge)?;
let digest = format!("sha256:{}", hex::encode(Sha256::digest(&payload)));
- sqlx::query("INSERT OR IGNORE INTO transcript_rows (session_id,source_sequence,item_index,turn_id,message_id,payload,digest,total_bytes) VALUES (?1,?2,?3,?4,?5,?6,?7,?8)")
+ sqlx::query("INSERT OR IGNORE INTO transcript_cache.rows (session_id,source_sequence,item_index,turn_id,message_id,payload,digest,total_bytes) VALUES (?1,?2,?3,?4,?5,?6,?7,?8)")
.bind(session).bind(source).bind(item).bind(&row.message.turn_id).bind(&row.message.id).bind(&payload).bind(&digest).bind(bytes).execute(&mut *db).await?;
- let matches: bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3 AND turn_id=?4 AND message_id=?5 AND payload=?6 AND digest=?7 AND total_bytes=?8)")
+ let matches: bool=sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3 AND turn_id=?4 AND message_id=?5 AND payload=?6 AND digest=?7 AND total_bytes=?8)")
.bind(session).bind(source).bind(item).bind(&row.message.turn_id).bind(&row.message.id).bind(payload).bind(digest).bind(bytes).fetch_one(&mut *db).await?;
if !matches {
return Err(StoreError::TranscriptConflict);
}
}
+ retain_source(db, session, source).await?;
Ok(())
}
+
+#[cfg(test)]
+#[path = "bodycache_tests.rs"]
+mod tests;
diff --git a/platform/event-log/src/transcript/bodycache_tests.rs b/platform/event-log/src/transcript/bodycache_tests.rs
new file mode 100644
index 0000000000..93ace0ae31
--- /dev/null
+++ b/platform/event-log/src/transcript/bodycache_tests.rs
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{EventLog, test_store::inspect};
+use maka_agent_facts::{
+ event::{EventWrite, Fact, Invocation, InvocationInput, InvocationOutcome, RuntimeEvent},
+ history::RowPosition,
+};
+use maka_agent_ports::session::history::{HistoryField, Page, Position};
+
+async fn user(log: &EventLog, turn: &str, text: String) -> u64 {
+ let write = EventWrite::plain(RuntimeEvent::new(
+ Invocation {
+ session_id: "cache".into(),
+ turn_id: turn.into(),
+ run_id: format!("run-{turn}"),
+ invocation_id: format!("inv-{turn}"),
+ },
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ source_messages: vec![],
+ content: text.into(),
+ request_fingerprint: None,
+ },
+ },
+ ))
+ .unwrap();
+ let source = log.append(&write).await.unwrap();
+ log.append(
+ &EventWrite::plain(RuntimeEvent::new(
+ write.event().invocation.clone(),
+ Fact::InvocationEnded {
+ outcome: InvocationOutcome::Completed,
+ },
+ ))
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ source
+}
+
+async fn read(log: &EventLog, through: u64, position: Position) -> Page {
+ log.history_text("cache", through, position, false, 8192, 1, 32768)
+ .await
+ .unwrap()
+}
+
+#[tokio::test]
+async fn private_body_reads_do_not_write_main_and_reopen_reconstructs_exact_bytes() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("history.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("cache", "create", &serde_json::json!({}), 1)
+ .await
+ .unwrap();
+ let expected = "hello 中🙂\n".repeat(200);
+ let through = user(&log, "first", expected.clone()).await;
+ let main = rusqlite::Connection::open(&path).unwrap();
+ let version = || {
+ main.query_row("PRAGMA data_version", [], |row| row.get::<_, i64>(0))
+ .unwrap()
+ };
+ let before = version();
+ let page = read(&log, through, Position::Start).await;
+ let Page::Ready { chunks, .. } = &page;
+ assert_eq!(
+ chunks
+ .iter()
+ .map(|chunk| chunk.text.as_str())
+ .collect::<String>(),
+ expected
+ );
+ let position = RowPosition {
+ source_sequence: through,
+ item_index: 0,
+ };
+ let header = log
+ .transcript_headers(
+ "cache",
+ &crate::transcript::TranscriptRead {
+ through,
+ position: Some(position),
+ direction: crate::transcript::TranscriptDirection::Newer,
+ limit: 1,
+ },
+ )
+ .await
+ .unwrap()
+ .remove(0);
+ let payload = log
+ .transcript_fragment("cache", position, 0, header.total_bytes)
+ .await
+ .unwrap();
+ assert_eq!(
+ serde_json::from_slice::<serde_json::Value>(&payload).unwrap()["text"],
+ expected
+ );
+ assert_eq!(
+ version(),
+ before,
+ "cold body/text filling must not commit to main"
+ );
+ inspect(&log, |db| {
+ assert!(db.query_row("SELECT count(*) FROM transcript_cache.rows", [], |row| row.get::<_, i64>(0)).unwrap()>0);
+ assert_eq!(db.query_row("SELECT count(*) FROM main.sqlite_schema WHERE name IN ('transcript_rows','transcript_text')", [], |row| row.get::<_, i64>(0)).unwrap(),0);
+ assert_eq!(db.query_row("PRAGMA main.journal_mode", [], |row| row.get::<_, String>(0)).unwrap(), "wal");
+ for (query, expected) in [("PRAGMA main.synchronous",2), ("PRAGMA transcript_cache.page_size",4096), ("PRAGMA transcript_cache.auto_vacuum",1), ("PRAGMA transcript_cache.cache_size",-4096), ("PRAGMA transcript_cache.synchronous",0), ("PRAGMA temp_store",1)] {
+ assert_eq!(db.query_row(query, [], |row| row.get::<_, i64>(0)).unwrap(), expected, "{query}");
+ }
+ assert_eq!(db.query_row("SELECT count(*) FROM pragma_compile_options WHERE compile_options='TEMP_STORE=3'", [], |row| row.get::<_, i64>(0)).unwrap(),0,"temporary storage must be spillable");
+ }).await;
+ log.close().await.unwrap();
+ let reopened = EventLog::open(&path).await.unwrap();
+ inspect(&reopened, |db| {
+ assert_eq!(
+ db.query_row("SELECT count(*) FROM transcript_cache.rows", [], |row| row
+ .get::<_, i64>(
+ 0
+ ))
+ .unwrap(),
+ 0
+ )
+ })
+ .await;
+ let before = version();
+ assert_eq!(
+ serde_json::to_value(read(&reopened, through, Position::Start).await).unwrap(),
+ serde_json::to_value(page).unwrap()
+ );
+ assert_eq!(version(), before);
+ reopened.close().await.unwrap();
+}
+
+/// Observe actual VFS backing size while holding the original SQLite worker.
+/// An unspilled temporary database has no open file methods yet.
+fn backing_bytes(db: &rusqlite::Connection) -> Option<i64> {
+ use rusqlite::ffi;
+ let mut file: *mut ffi::sqlite3_file = std::ptr::null_mut();
+ // SAFETY: db and its VFS file remain owned and exclusively accessed by this
+ // callback; neither the handle nor method pointers escape it.
+ unsafe {
+ assert_eq!(
+ ffi::sqlite3_file_control(
+ db.handle(),
+ c"transcript_cache".as_ptr(),
+ ffi::SQLITE_FCNTL_FILE_POINTER,
+ (&mut file as *mut *mut ffi::sqlite3_file).cast()
+ ),
+ ffi::SQLITE_OK
+ );
+ if file.is_null() || (*file).pMethods.is_null() {
+ return None;
+ }
+ let mut size = 0;
+ assert_eq!(
+ ((*(*file).pMethods).xFileSize.unwrap())(file, &mut size),
+ ffi::SQLITE_OK
+ );
+ Some(size)
+ }
+}
+
+#[tokio::test]
+async fn oversized_source_reuses_fragments_then_reclaims_backing_file_on_replacement() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("oversized.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("cache", "create", &serde_json::json!({}), 1)
+ .await
+ .unwrap();
+ let through = user(&log, "large", "α".repeat(17 * 1024 * 1024)).await;
+ let position = RowPosition {
+ source_sequence: through,
+ item_index: 0,
+ };
+ let page = read(
+ &log,
+ through,
+ Position::Row {
+ position,
+ field: HistoryField::Body,
+ },
+ )
+ .await;
+ let Page::Ready { chunks, after, .. } = page;
+ assert_eq!(chunks[0].text, "α".repeat(4096));
+ let before = inspect(&log, |db| {
+ assert!(
+ db.query_row("PRAGMA transcript_cache.page_count", [], |row| row
+ .get::<_, i64>(0))
+ .unwrap()
+ > super::RETAINED_PAGES
+ );
+ (
+ db.total_changes(),
+ backing_bytes(db).expect("large source must spill with a 4 MiB pager target"),
+ )
+ })
+ .await;
+ let Page::Ready { chunks, .. } = read(
+ &log,
+ through,
+ Position::After {
+ cursor: after.unwrap(),
+ },
+ )
+ .await;
+ assert_eq!(chunks[0].text, "α".repeat(4096));
+ inspect(&log, move |db| {
+ assert_eq!(
+ db.total_changes(),
+ before.0,
+ "another fragment must reuse the same source and text"
+ )
+ })
+ .await;
+ let small = user(&log, "small", "replacement 中🙂".into()).await;
+ let Page::Ready { chunks, .. } = read(
+ &log,
+ small,
+ Position::Row {
+ position: RowPosition {
+ source_sequence: small,
+ item_index: 0,
+ },
+ field: HistoryField::Body,
+ },
+ )
+ .await;
+ assert_eq!(chunks[0].text, "replacement 中🙂");
+ inspect(&log, move |db| {
+ let pages = db
+ .query_row("PRAGMA transcript_cache.page_count", [], |row| {
+ row.get::<_, i64>(0)
+ })
+ .unwrap();
+ assert!(pages <= super::RETAINED_PAGES);
+ assert_eq!(
+ db.query_row("PRAGMA transcript_cache.freelist_count", [], |row| row
+ .get::<_, i64>(0))
+ .unwrap(),
+ 0
+ );
+ assert_eq!(
+ db.query_row(
+ "SELECT count(*) FROM transcript_cache.rows WHERE source_sequence=?",
+ [through as i64],
+ |row| row.get::<_, i64>(0)
+ )
+ .unwrap(),
+ 0
+ );
+ let size = backing_bytes(db).unwrap();
+ assert!(
+ size <= 32 * 1024 * 1024 && size < before.1,
+ "actual temporary file must shrink: {} -> {size}",
+ before.1
+ );
+ })
+ .await;
+ let Page::Ready { chunks, .. } = read(
+ &log,
+ through,
+ Position::Row {
+ position,
+ field: HistoryField::Body,
+ },
+ )
+ .await;
+ assert_eq!(
+ chunks[0].text,
+ "α".repeat(4096),
+ "evicted source reconstructs without changing its identity or bytes"
+ );
+ log.close().await.unwrap();
+}
diff --git a/platform/event-log/src/transcript/directory.rs b/platform/event-log/src/transcript/directory.rs
index c3c6ff16fc..054aa9778f 100644
--- a/platform/event-log/src/transcript/directory.rs
+++ b/platform/event-log/src/transcript/directory.rs
@@ -330,8 +330,14 @@ fn copy_on_handle(
[target],
)?;
// An earlier inherited ID may replace a later cached bare claimant.
- db.execute("DELETE FROM transcript_rows WHERE session_id=?1", [target])?;
- db.execute("DELETE FROM transcript_text WHERE session_id=?1", [target])?;
+ db.execute(
+ "DELETE FROM transcript_cache.rows WHERE session_id=?1",
+ [target],
+ )?;
+ db.execute(
+ "DELETE FROM transcript_cache.text WHERE session_id=?1",
+ [target],
+ )?;
let mut after = 0i64;
loop {
let next: Option<i64> = db.prepare_cached("SELECT source_sequence FROM (SELECT source_sequence FROM temp.transcript_copy_rows WHERE source_sequence>?1 UNION SELECT source_sequence FROM temp.transcript_copy_failures WHERE source_sequence>?1) ORDER BY source_sequence LIMIT 1")?.query_row([after],|r|r.get(0)).optional()?;
@@ -449,7 +455,7 @@ mod tests {
check_read(db,"bare",end).await?;
let rows: Vec<(i64,String)> = sqlx::query_as("SELECT item_index,role FROM transcript_directory WHERE session_id='bare' AND source_sequence=? ORDER BY item_index").bind(first as i64).fetch_all(&mut *db).await?;
assert_eq!(rows,vec![(0,"user".into()),(1,"turn_state".into())]);
- assert_eq!(sqlx::query_scalar::<_,i64>("SELECT count(*) FROM transcript_rows").fetch_one(&mut *db).await?,0);
+ assert_eq!(sqlx::query_scalar::<_,i64>("SELECT count(*) FROM transcript_cache.rows").fetch_one(&mut *db).await?,0);
Ok(())
})).await.unwrap();
log.close().await.unwrap();
diff --git a/platform/event-log/src/transcript/history.rs b/platform/event-log/src/transcript/history.rs
index 73a518d6a1..d679b6405e 100644
--- a/platform/event-log/src/transcript/history.rs
+++ b/platform/event-log/src/transcript/history.rs
@@ -29,7 +29,7 @@ use maka_agent_ports::session::history::Page;
use maka_agent_ports::session::history::Position;
use maka_agent_ports::session::history::Role;
use maka_agent_ports::session::history::{HistoryField, HistoryView};
-use sqlx::{Connection, Row, SqliteConnection};
+use sqlx::{Row, SqliteConnection};
// Text windows and exact source reads retain material labels in their body.
pub(super) const BODY: &str = "COALESCE(CASE json_extract(payload, '$.type')
@@ -125,10 +125,9 @@ impl EventLog {
// therefore bounds rows even when a caller needs no item-count limit.
let max_chunks = max_chunks.unwrap_or(max_json_bytes).min(max_json_bytes);
let session = session.to_owned();
- self.connection.run(move |connection| Box::pin(async move {
- let mut tx = connection.begin().await?;
+ self.connection.run(move |connection| super::bodycache::run(connection, move |tx| Box::pin(async move {
let mut final_outputs = std::collections::HashMap::new();
- super::directory::check_read(&mut tx, &session, fence).await?;
+ super::directory::check_read(tx, &session, fence).await?;
let exists: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM session_control WHERE id = ?)"
).bind(&session).fetch_one(&mut *tx).await?;
@@ -138,10 +137,10 @@ impl EventLog {
|| (!include_thinking && field == HistoryField::Thinking) {
return Err(invalid("history row exceeds its fence"));
}
- materialize(&mut tx, &session, position, view, through_sql).await?;
+ materialize(tx, &session, position, view, through_sql).await?;
// Header lookup requires the actual field; a tool Body cannot
// impersonate a Thinking field that was never recorded.
- header(&mut tx, &session, position, field, view, through_sql).await?;
+ header(tx, &session, position, field, view, through_sql).await?;
Some((position, field))
} else { None };
let anchor = match position {
@@ -150,13 +149,13 @@ impl EventLog {
return Err(invalid("history cursor exceeds its fence"));
}
sql_number(cursor.offset)?;
- materialize(&mut tx, &session, cursor.position, view, through_sql).await?;
- let row = header(&mut tx, &session, cursor.position, cursor.field, view, through_sql).await?;
+ materialize(tx, &session, cursor.position, view, through_sql).await?;
+ let row = header(tx, &session, cursor.position, cursor.field, view, through_sql).await?;
if cursor.offset > row.total {
return Err(invalid("history offset exceeds message"));
}
let byte = if cursor.offset == row.total { Vec::new() } else {
- fragment(&mut tx, &session, cursor, 1,
+ fragment(tx, &session, cursor, 1,
view == HistoryView::Conversation && row.role == "user").await?
};
if byte.first().is_some_and(|byte| byte & 0xc0 == 0x80) {
@@ -205,7 +204,7 @@ impl EventLog {
let locator = row_position(locator.0, locator.1)?;
// Materialize only actually visited rows, never a long message's
// following rows. One oversized first row still progresses.
- projected += materialize(&mut tx, &session, locator, view, through_sql).await?;
+ projected += materialize(tx, &session, locator, view, through_sql).await?;
let has_thinking: bool = include_thinking && sqlx::query_scalar(
"SELECT has_thinking FROM transcript_directory WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
).bind(&session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_one(&mut *tx).await?;
@@ -217,7 +216,7 @@ impl EventLog {
if target == Some((locator, HistoryField::Body)) && field == HistoryField::Thinking {
continue;
}
- let row = header(&mut tx, &session, locator, field, view, through_sql).await?;
+ let row = header(tx, &session, locator, field, view, through_sql).await?;
let exact_target = target == Some((locator, field));
if let Some((cursor, _)) = &anchor && cursor.position == locator {
let before = field == HistoryField::Thinking && cursor.field == HistoryField::Body;
@@ -235,7 +234,7 @@ impl EventLog {
(start, row.total.min(start + (max_bytes - bytes) as u64))
};
let fragment = if row.total == 0 { Vec::new() } else {
- fragment(&mut tx, &session, Cursor { field, position: locator, offset: start }, end - start,
+ fragment(tx, &session, Cursor { field, position: locator, offset: start }, end - start,
view == HistoryView::Conversation && row.role == "user").await?
};
let text = if reverse {
@@ -258,7 +257,7 @@ impl EventLog {
let candidate_right = if reverse { right.or(Some(ending)) } else { Some(ending) };
let attachments: Vec<maka_agent_facts::attachment::AttachmentRef> = if start == 0 && row.role == "user" {
let encoded: Option<String> = sqlx::query_scalar(
- "SELECT json_extract(payload, '$.attachments') FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
+ "SELECT json_extract(payload, '$.attachments') FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
).bind(&session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_one(&mut *tx).await?;
encoded.map(|value| serde_json::from_str(&value)).transpose()?.unwrap_or_default()
} else { Vec::new() };
@@ -275,12 +274,12 @@ impl EventLog {
_ => return Err(invalid("invalid history text role")),
};
let tool = if matches!(role, Role::ToolCall | Role::ToolResult) {
- Some(tool_metadata(&mut tx, &session, locator, through_sql).await?)
+ Some(tool_metadata(tx, &session, locator, through_sql).await?)
} else { None };
- let timing = super::timing::read(&mut tx, &session, &row.turn, through_sql).await?;
+ let timing = super::timing::read(tx, &session, &row.turn, through_sql).await?;
let compaction = if role == Role::Compaction {
let duration: i64 = sqlx::query_scalar(
- "SELECT json_extract(payload, '$.durationMs') FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
+ "SELECT json_extract(payload, '$.durationMs') FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
).bind(&session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_one(&mut *tx).await?;
Some(maka_agent_ports::session::history::Compaction { duration_ms: sequence_number(duration)? })
} else { None };
@@ -288,9 +287,9 @@ impl EventLog {
let output = match final_outputs.entry(row.turn.clone()) {
std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
std::collections::hash_map::Entry::Vacant(entry) => entry.insert(
- super::final_output::resolve(&mut tx, &session, &row.turn, through_sql).await?),
+ super::final_output::resolve(tx, &session, &row.turn, through_sql).await?),
};
- let metadata = super::model::resolve(&mut tx, &session, locator, through_sql).await?;
+ let metadata = super::model::resolve(tx, &session, locator, through_sql).await?;
let final_output = field == HistoryField::Body && row.total > 0 && metadata.producer.as_ref().is_some_and(|producer|
output.as_ref() == Some(producer));
(metadata.model, final_output)
@@ -305,7 +304,7 @@ impl EventLog {
// Bound the complete canonical delivery, including attachment
// descriptors and JSON escaping. A rejected candidate does not
// advance the source boundary or silently discard any metadata.
- let (before, after) = boundaries(&mut tx, &session, through_sql, include_thinking, candidate_left, candidate_right).await?;
+ let (before, after) = boundaries(tx, &session, through_sql, include_thinking, candidate_left, candidate_right).await?;
// The empty envelope includes []; each chunk adds its exact
// encoded bytes, plus commas between elements. Cursor changes
// still count even when this boundary emits no chunk.
@@ -332,17 +331,17 @@ impl EventLog {
).bind(&session).bind(through_sql).fetch_optional(&mut *tx).await?;
if let Some(locator) = last {
let locator = row_position(locator.0, locator.1)?;
- materialize(&mut tx, &session, locator, view, through_sql).await?;
- let row = header(&mut tx, &session, locator, HistoryField::Body, view, through_sql).await?;
+ materialize(tx, &session, locator, view, through_sql).await?;
+ let row = header(tx, &session, locator, HistoryField::Body, view, through_sql).await?;
let cursor = Cursor { field: HistoryField::Body, position: locator, offset: row.total };
left = Some(cursor); right = Some((cursor, row.total));
}
}
}
- let (before, after) = boundaries(&mut tx, &session, through_sql, include_thinking, left, right).await?;
- tx.commit().await?;
+ let (before, after) = boundaries(tx, &session, through_sql, include_thinking, left, right).await?;
+
Ok(Page::Ready { through, chunks, before, after })
- })).await
+ }))).await
}
}
@@ -401,7 +400,7 @@ async fn header(
let row = sqlx::query(
"SELECT source.message_id, source.turn_id, source.timestamp, source.role, length(cached.body),
CASE WHEN source.role = 'user' THEN source.user_origin END
- FROM transcript_directory source JOIN transcript_text cached USING(session_id, source_sequence, item_index)
+ FROM transcript_directory source JOIN transcript_cache.text cached USING(session_id, source_sequence, item_index)
WHERE source.session_id = ?1 AND source.source_sequence = ?2 AND item_index = ?3 AND cached.field = ?4"
).bind(session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).bind(field_name(field)).fetch_one(&mut *tx).await?;
let mut header = Header {
@@ -418,7 +417,7 @@ async fn header(
};
if view == HistoryView::Conversation && field == HistoryField::Body && header.role == "user" {
let query = format!(
- "SELECT length(CAST({CONVERSATION_USER} AS BLOB)) FROM transcript_rows
+ "SELECT length(CAST({CONVERSATION_USER} AS BLOB)) FROM transcript_cache.rows
WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
);
header.total = sequence_number(
@@ -442,11 +441,11 @@ async fn fragment(
) -> Result<Vec<u8>, StoreError> {
let query = if conversation_user {
format!(
- "SELECT substr(CAST({CONVERSATION_USER} AS BLOB), ?4, ?5) FROM transcript_rows
+ "SELECT substr(CAST({CONVERSATION_USER} AS BLOB), ?4, ?5) FROM transcript_cache.rows
WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
)
} else {
- "SELECT substr(body, ?4, ?5) FROM transcript_text
+ "SELECT substr(body, ?4, ?5) FROM transcript_cache.text
WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3 AND field = ?6"
.into()
};
@@ -557,8 +556,8 @@ async fn materialize(
}
super::bodycache::ensure(tx, session, locator).await?;
let source: Option<(i64, bool)> = sqlx::query_as(
- "SELECT total_bytes, EXISTS(SELECT 1 FROM transcript_text cached WHERE cached.session_id = source.session_id AND cached.source_sequence = source.source_sequence AND cached.item_index = source.item_index AND cached.field = 'body')
- FROM transcript_rows source WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
+ "SELECT total_bytes, EXISTS(SELECT 1 FROM transcript_cache.text cached WHERE cached.session_id = source.session_id AND cached.source_sequence = source.source_sequence AND cached.item_index = source.item_index AND cached.field = 'body')
+ FROM transcript_cache.rows source WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3"
).bind(session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_optional(&mut *tx).await?;
let Some((bytes, cached)) = source else {
return Err(invalid("history row was not found"));
@@ -567,14 +566,14 @@ async fn materialize(
return Ok(0);
}
let query = format!(
- "INSERT INTO transcript_text
+ "INSERT INTO transcript_cache.text
SELECT session_id, source_sequence, item_index, 'body', json_extract(payload, '$.ts'), json_extract(payload, '$.type'),
CAST({BODY} AS BLOB)
- FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
+ FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
UNION ALL
SELECT session_id, source_sequence, item_index, 'thinking', json_extract(payload, '$.ts'), 'assistant',
CAST(json_extract(payload, '$.thinking.text') AS BLOB)
- FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
+ FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
AND json_extract(payload, '$.type') = 'assistant' AND json_type(payload, '$.thinking.text') = 'text'"
);
sqlx::query(sqlx::AssertSqlSafe(query.as_str()))
@@ -601,6 +600,7 @@ fn field_name(field: HistoryField) -> &'static str {
mod tests {
use super::*;
use serde_json::json;
+ use sqlx::Connection;
#[tokio::test]
async fn conversation_retains_trace_parents_and_hides_transport_helpers() {
diff --git a/platform/event-log/src/transcript/image.rs b/platform/event-log/src/transcript/image.rs
index c7de12da3c..4ae365f425 100644
--- a/platform/event-log/src/transcript/image.rs
+++ b/platform/event-log/src/transcript/image.rs
@@ -76,12 +76,11 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, through).await?;
- super::bodycache::ensure(&mut tx, &session, position).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, through).await?;
+ super::bodycache::ensure(tx, &session, position).await?;
let raw: String = sqlx::query_scalar(
- "SELECT a.value FROM transcript_rows r, json_each(r.payload, '$.attachments') a
+ "SELECT a.value FROM transcript_cache.rows r, json_each(r.payload, '$.attachments') a
WHERE r.session_id = ?1 AND r.source_sequence = ?2 AND r.item_index = ?3
AND json_extract(r.payload, '$.type') = 'user' AND a.key = ?4",
)
@@ -104,7 +103,7 @@ impl EventLog {
return Err(invalid("image attachment is outside this Session"));
}
let chunk = crate::artifacts::read_attachment_image(
- &mut tx,
+ tx,
&session,
relative_path,
Some(&attachment),
@@ -112,9 +111,9 @@ impl EventLog {
limit,
)
.await?;
- tx.commit().await?;
+
Ok(chunk)
- })
+ }))
})
.await
}
@@ -139,13 +138,12 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, through).await?;
- super::bodycache::ensure(&mut tx, &session, position).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, through).await?;
+ super::bodycache::ensure(tx, &session, position).await?;
let raw: String = sqlx::query_scalar(
"SELECT json_remove(json_extract(payload, '$.content'), '$.kind')
- FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
+ FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3
AND json_extract(payload, '$.type') = 'tool_result'
AND json_extract(payload, '$.content.kind') = 'image'
AND length(CAST(json_extract(payload, '$.content') AS BLOB)) <= 16384",
@@ -168,7 +166,7 @@ impl EventLog {
return Err(invalid("tool image belongs to another Session"));
}
let chunk = crate::artifacts::read_image(
- &mut tx,
+ tx,
&session,
&relative_path,
&image.mime_type,
@@ -177,9 +175,9 @@ impl EventLog {
position.source_sequence,
)
.await?;
- tx.commit().await?;
+
Ok(chunk)
- })
+ }))
})
.await
}
diff --git a/platform/event-log/src/transcript/navigation.rs b/platform/event-log/src/transcript/navigation.rs
index 00dff3c3f6..3969a6967b 100644
--- a/platform/event-log/src/transcript/navigation.rs
+++ b/platform/event-log/src/transcript/navigation.rs
@@ -81,9 +81,8 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, through).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, through).await?;
let rows = sqlx::query(queries::ROWS)
.bind(&session)
.bind(through_sql)
@@ -130,12 +129,12 @@ impl EventLog {
let role: String = row.try_get(3)?;
let kind: String = row.try_get(4)?;
if role == "user" && matches!(kind.as_str(), "invocation_opened" | "message_steered") {
- contribution.user_prompt_preview = preview(&mut tx,&session,position,256).await?;
+ contribution.user_prompt_preview = preview(tx,&session,position,256).await?;
}
}
if row.try_get::<String,_>(3)? == "turn_state" {
- super::bodycache::ensure(&mut tx,&session,position).await?;
- let bytes: Vec<u8> = sqlx::query_scalar("SELECT payload FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ super::bodycache::ensure(tx,&session,position).await?;
+ let bytes: Vec<u8> = sqlx::query_scalar("SELECT payload FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(sql_position(position)?.0).bind(sql_position(position)?.1).fetch_one(&mut *tx).await?;
let mut message: TurnStateMessage = serde_json::from_slice(&bytes)?;
if message.turn_id != contribution.turn_id {
@@ -162,12 +161,12 @@ impl EventLog {
contributions.push(contribution);
}
}
- tx.commit().await?;
+
Ok(TurnPage {
contributions,
next_position,
})
- })
+ }))
})
.await
}
@@ -193,9 +192,8 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, through).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, through).await?;
if let Some(turn) = turn {
let bounds: Option<(i64, i64, i64, i64)> = sqlx::query_as(queries::BOUNDS)
.bind(&session).bind(&turn).bind(through_sql).fetch_optional(&mut *tx).await?;
@@ -205,8 +203,8 @@ impl EventLog {
WHERE row.session_id=?1 AND row.turn_id=?2 AND row.source_sequence<=?3 AND row.role='user'
AND source.kind IN ('invocation_opened','message_steered') ORDER BY row.source_sequence,row.item_index LIMIT 1")
.bind(&session).bind(&turn).bind(through_sql).fetch_optional(&mut *tx).await?;
- let label = if let Some((source,item)) = prompt { preview(&mut tx,&session,row_position(source,item)?,96).await? } else { None };
- tx.commit().await?;
+ let label = if let Some((source,item)) = prompt { preview(tx,&session,row_position(source,item)?,96).await? } else { None };
+
return Ok(vec![TurnLandmark {
turn_id: turn, position: row_position(first_source, first_index)?, last_position: row_position(last_source, last_index)?,
label: label.unwrap_or_default(),
@@ -227,7 +225,7 @@ impl EventLog {
.fetch_optional(&mut *tx)
.await?
{
- let label = preview(&mut tx,&session,row_position(row.try_get(1)?,row.try_get(2)?)?,96).await?;
+ let label = preview(tx,&session,row_position(row.try_get(1)?,row.try_get(2)?)?,96).await?;
if let Some(label) = label {
let turn_id: String = row.try_get(0)?;
let (first_source, first_index, last_source, last_index): (i64, i64, i64, i64) = sqlx::query_as(queries::BOUNDS)
@@ -241,9 +239,9 @@ impl EventLog {
}
}
}
- tx.commit().await?;
+
Ok(result)
- })
+ }))
})
.await
}
@@ -256,7 +254,7 @@ async fn preview(
budget: i64,
) -> Result<Option<String>, StoreError> {
super::bodycache::ensure(db, session, position).await?;
- sqlx::query_scalar("SELECT navigation_preview(COALESCE(json_extract(payload,'$.displayText'),json_extract(payload,'$.text')),?4) FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ sqlx::query_scalar("SELECT navigation_preview(COALESCE(json_extract(payload,'$.displayText'),json_extract(payload,'$.text')),?4) FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(session).bind(sql_position(position)?.0).bind(sql_position(position)?.1).bind(budget).fetch_one(db).await.map_err(Into::into)
}
diff --git a/platform/event-log/src/transcript/prompt.rs b/platform/event-log/src/transcript/prompt.rs
index f0c256b6d8..ae0e0969e9 100644
--- a/platform/event-log/src/transcript/prompt.rs
+++ b/platform/event-log/src/transcript/prompt.rs
@@ -20,7 +20,6 @@
use crate::{EventLog, StoreError};
use maka_agent_facts::history::RowPosition;
use maka_agent_ports::session::history::Prompt;
-use sqlx::Connection;
const QUERY: &str = "SELECT source_sequence,item_index FROM transcript_directory
WHERE session_id=?1 AND (source_sequence,item_index)<=(?2,?3) AND role='user'
@@ -47,20 +46,19 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, through).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, through).await?;
let row: Option<(i64,i64)> = sqlx::query_as(QUERY)
.bind(&session).bind(source_sequence).bind(item_index)
.fetch_optional(&mut *tx).await?;
let Some((source,item)) = row else { return Ok(None) };
let position = super::read::row_position(source,item)?;
- super::bodycache::ensure(&mut tx,&session,position).await?;
- let text: String = sqlx::query_scalar("SELECT COALESCE(navigation_preview(COALESCE(json_extract(payload,'$.displayText'),json_extract(payload,'$.text'),''),4096),'[attachments]') FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ super::bodycache::ensure(tx,&session,position).await?;
+ let text: String = sqlx::query_scalar("SELECT COALESCE(navigation_preview(COALESCE(json_extract(payload,'$.displayText'),json_extract(payload,'$.text'),''),4096),'[attachments]') FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(source).bind(item).fetch_one(&mut *tx).await?;
- tx.commit().await?;
+
Ok(Some(Prompt { position,text }))
- })
+ }))
})
.await
}
diff --git a/platform/event-log/src/transcript/read.rs b/platform/event-log/src/transcript/read.rs
index 79cdad87d0..af5aef6adc 100644
--- a/platform/event-log/src/transcript/read.rs
+++ b/platform/event-log/src/transcript/read.rs
@@ -19,7 +19,7 @@
use crate::{EventLog, StoreError, sequence_number, sessions};
use maka_agent_facts::history::RowPosition;
-use sqlx::{Connection, Row};
+use sqlx::Row;
#[derive(Clone, Copy, Debug)]
pub enum TranscriptDirection {
@@ -107,9 +107,8 @@ impl EventLog {
let fence = read.through;
let direction = read.direction;
let limit = read.limit as i64;
- self.connection.run(move |connection| Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, fence).await?;
+ self.connection.run(move |connection| super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, fence).await?;
let query = match (direction, position.is_some()) {
(TranscriptDirection::Older, true) => "SELECT source_sequence,item_index,turn_id FROM transcript_directory WHERE session_id=?1 AND (source_sequence,item_index)<=(?3,?4) ORDER BY source_sequence DESC,item_index DESC LIMIT ?2",
(TranscriptDirection::Newer, true) => "SELECT source_sequence,item_index,turn_id FROM transcript_directory WHERE session_id=?1 AND source_sequence<=?2 AND (source_sequence,item_index)>=(?4,?5) ORDER BY source_sequence,item_index LIMIT ?3",
@@ -126,14 +125,14 @@ impl EventLog {
let mut result = Vec::with_capacity(selected.len());
for row in selected {
let position = row_position(row.try_get(0)?, row.try_get(1)?)?;
- super::bodycache::ensure(&mut tx, &session, position).await?;
- let (bytes, digest): (i64, String) = sqlx::query_as("SELECT total_bytes,digest FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ super::bodycache::ensure(tx, &session, position).await?;
+ let (bytes, digest): (i64, String) = sqlx::query_as("SELECT total_bytes,digest FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(sql_position(position)?.0).bind(sql_position(position)?.1).fetch_one(&mut *tx).await?;
result.push(TranscriptRowHeader { position, turn_id: row.try_get(2)?, total_bytes: sequence_number(bytes)?, digest });
}
- tx.commit().await?;
+
Ok(result)
- })).await
+ }))).await
}
/// Slice immutable UTF-8 JSON bytes; assembly may cross code points.
pub async fn transcript_fragment(
@@ -155,20 +154,19 @@ impl EventLog {
.ok_or_else(|| invalid("invalid transcript byte offset"))?,
)?;
let session = session.to_owned();
- self.connection.run(move |connection| Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, position.source_sequence).await?;
- super::bodycache::ensure(&mut tx, &session, position).await?;
- let total:i64=sqlx::query_scalar("SELECT total_bytes FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ self.connection.run(move |connection| super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, position.source_sequence).await?;
+ super::bodycache::ensure(tx, &session, position).await?;
+ let total:i64=sqlx::query_scalar("SELECT total_bytes FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(source).bind(item).fetch_optional(&mut *tx).await?.ok_or_else(||invalid("transcript message was not found"))?;
let total = sequence_number(total)?;
if offset.checked_add(length).is_none_or(|end|end>total) { return Err(invalid("transcript fragment exceeds message")); }
- let bytes:Vec<u8>=sqlx::query_scalar("SELECT substr(payload,?4,?5) FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ let bytes:Vec<u8>=sqlx::query_scalar("SELECT substr(payload,?4,?5) FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(source).bind(item).bind(start).bind(length as i64).fetch_one(&mut *tx).await?;
if bytes.len() as u64 != length { return Err(invalid("transcript payload length changed")); }
- tx.commit().await?;
+
Ok(bytes)
- })).await
+ }))).await
}
pub async fn transcript_turn_bounds(
&self,
@@ -182,12 +180,11 @@ impl EventLog {
let fence = sql_number(through)?;
let session = session.to_owned();
let turn = turn.to_owned();
- self.connection.run(move |connection| Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx,&session,through).await?;
+ self.connection.run(move |connection| super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx,&session,through).await?;
let first:Option<(i64,i64)>=sqlx::query_as("SELECT source_sequence,item_index FROM transcript_directory WHERE session_id=?1 AND turn_id=?2 AND source_sequence<=?3 ORDER BY source_sequence,item_index LIMIT 1")
.bind(&session).bind(&turn).bind(fence).fetch_optional(&mut *tx).await?;
- let Some((source,item))=first else { tx.commit().await?; return Ok(None) };
+ let Some((source,item))=first else { return Ok(None) };
let (last_source,last_item):(i64,i64)=sqlx::query_as("SELECT source_sequence,item_index FROM transcript_directory WHERE session_id=?1 AND turn_id=?2 AND source_sequence<=?3 ORDER BY source_sequence DESC,item_index DESC LIMIT 1")
.bind(&session).bind(&turn).bind(fence).fetch_one(&mut *tx).await?;
// Exact serialized bytes intentionally visit the whole requested Turn.
@@ -203,17 +200,17 @@ impl EventLog {
.bind(&session).bind(&turn).bind(fence).bind(cursor.0).bind(cursor.1).fetch_all(&mut *tx).await?;
if selected.is_empty() { break; }
for (source,item) in selected {
- super::bodycache::ensure(&mut tx,&session,row_position(source,item)?).await?;
- let size: i64=sqlx::query_scalar("SELECT total_bytes FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ super::bodycache::ensure(tx,&session,row_position(source,item)?).await?;
+ let size: i64=sqlx::query_scalar("SELECT total_bytes FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(source).bind(item).fetch_one(&mut *tx).await?;
bytes=bytes.checked_add(sequence_number(size)?).ok_or(StoreError::PrefixTooLarge)?;
cursor=(source,item);
}
}
let bounds = TranscriptTurnBounds { first:row_position(source,item)?,last:row_position(last_source,last_item)?,rows,bytes };
- tx.commit().await?;
+
Ok(Some(bounds))
- })).await
+ }))).await
}
}
pub(super) fn sql_number(value: u64) -> Result<i64, StoreError> {
diff --git a/platform/event-log/src/transcript/search.rs b/platform/event-log/src/transcript/search.rs
index ba8a902caa..ca0eb8cbe9 100644
--- a/platform/event-log/src/transcript/search.rs
+++ b/platform/event-log/src/transcript/search.rs
@@ -23,7 +23,7 @@ use maka_agent_facts::history::RowPosition;
use maka_agent_ports::session::history::SourceField;
use regex::{Regex, RegexBuilder};
use serde_json::Value;
-use sqlx::{Connection, Row, SqliteConnection};
+use sqlx::{Row, SqliteConnection};
pub struct TranscriptSearch {
pub through: u64,
@@ -81,9 +81,8 @@ impl EventLog {
let session = session.to_owned();
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, input.through).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, input.through).await?;
let query = if after.is_some() {
"SELECT source_sequence,item_index FROM transcript_directory WHERE session_id=?1 AND source_sequence<=?2 AND (source_sequence,item_index)>(?3,?4) ORDER BY source_sequence,item_index LIMIT 65"
} else {
@@ -104,8 +103,8 @@ impl EventLog {
break;
}
let locator = row_position(header.try_get(0)?,header.try_get(1)?)?;
- super::bodycache::ensure(&mut tx,&session,locator).await?;
- let size: i64 = sqlx::query_scalar("SELECT total_bytes FROM transcript_rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
+ super::bodycache::ensure(tx,&session,locator).await?;
+ let size: i64 = sqlx::query_scalar("SELECT total_bytes FROM transcript_cache.rows WHERE session_id=?1 AND source_sequence=?2 AND item_index=?3")
.bind(&session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_one(&mut *tx).await?;
let size = sequence_number(size)?;
if index > 0 && bytes + size > 16 * 1024 * 1024 {
@@ -115,11 +114,11 @@ impl EventLog {
bytes += size;
position = Some(locator);
let payload: Vec<u8> = sqlx::query_scalar(
- "SELECT payload FROM transcript_rows WHERE session_id = ? AND source_sequence = ? AND item_index = ?"
+ "SELECT payload FROM transcript_cache.rows WHERE session_id = ? AND source_sequence = ? AND item_index = ?"
).bind(&session).bind(sql_position(locator)?.0).bind(sql_position(locator)?.1).fetch_one(&mut *tx).await?;
let row: Value = serde_json::from_slice(&payload)?;
if !input.include_internal
- && internal(&mut tx, &session, through, locator, &row).await?
+ && internal(tx, &session, through, locator, &row).await?
{
continue;
}
@@ -131,9 +130,9 @@ impl EventLog {
});
}
}
- tx.commit().await?;
+
Ok(result)
- })
+ }))
})
.await
}
diff --git a/platform/event-log/src/transcript/source.rs b/platform/event-log/src/transcript/source.rs
index 03127bdc51..1fa3bae033 100644
--- a/platform/event-log/src/transcript/source.rs
+++ b/platform/event-log/src/transcript/source.rs
@@ -55,7 +55,7 @@ fn source_sql() -> String {
json_type(payload, '$.failureMessage') IS 'text' AS failure_message,
json_type(payload, '$.errorClass') IS 'text' AS error_class,
json_extract(payload, '$.content.kind') IS 'operation' AS operation
- FROM transcript_rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3),
+ FROM transcript_cache.rows WHERE session_id = ?1 AND source_sequence = ?2 AND item_index = ?3),
source AS (SELECT
CASE WHEN ?5 IS NULL THEN body WHEN format = 'json' THEN
CASE json_type(body, ?5) WHEN 'text' THEN json_extract(body, ?5) ELSE body -> ?5 END END AS body,
@@ -137,9 +137,8 @@ impl EventLog {
};
self.connection
.run(move |connection| {
- Box::pin(async move {
- let mut tx = connection.begin().await?;
- super::directory::check_read(&mut tx, &session, fence).await?;
+ super::bodycache::run(connection, move |tx| Box::pin(async move {
+ super::directory::check_read(tx, &session, fence).await?;
let exists: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM session_control WHERE id = ?)",
)
@@ -149,7 +148,7 @@ impl EventLog {
if !exists {
return Err(invalid("Session no longer exists"));
}
- super::bodycache::ensure(&mut tx, &session, locator).await?;
+ super::bodycache::ensure(tx, &session, locator).await?;
let sql = source_sql();
let header = format!(
"{sql} SELECT format, length(CAST(body AS BLOB)), role, thinking, failure_message, error_class, timestamp, operation, message_origin FROM source"
@@ -261,12 +260,12 @@ impl EventLog {
return Err(invalid("history source made no progress"));
}
let tool = if matches!(role, SourceRole::ToolCall | SourceRole::ToolResult) {
- Some(tool_metadata(&mut tx, &session, locator, fence_sql).await?)
+ Some(tool_metadata(tx, &session, locator, fence_sql).await?)
} else {
None
};
let model = if role == SourceRole::Assistant {
- super::model::resolve(&mut tx, &session, locator, sql_number(through)?).await?.model
+ super::model::resolve(tx, &session, locator, sql_number(through)?).await?.model
} else { None };
let source = HistorySource {
model,
@@ -283,9 +282,9 @@ impl EventLog {
text: text.to_owned(),
tool,
};
- tx.commit().await?;
+
Ok(source)
- })
+ }))
})
.await
}
diff --git a/platform/event-log/tests/compact_presentation/automatic.rs b/platform/event-log/tests/compact_presentation/automatic.rs
index 73c53fc13e..77c3b19ad8 100644
--- a/platform/event-log/tests/compact_presentation/automatic.rs
+++ b/platform/event-log/tests/compact_presentation/automatic.rs
@@ -126,7 +126,7 @@ async fn automatic_summary_is_hidden_without_interrupting_main_delivery_or_read_
.await
.unwrap();
let db = Connection::open(&path).unwrap();
- let before_rows = transcript(&db);
+ let before_rows = crate::test_store::inspect(&log, transcript).await;
if ending == "success" {
assert!(
finish(&log, &mut view, &inv, "summary", &"secret ".repeat(1024))
@@ -223,7 +223,7 @@ async fn automatic_summary_is_hidden_without_interrupting_main_delivery_or_read_
crate::transcript_cache::materialize_all_rows(&log, "session", end.sequence)
.await
.unwrap();
- let projected = transcript(&db);
+ let projected = crate::test_store::inspect(&log, transcript).await;
assert_eq!(&projected[..before_rows.len()], before_rows.as_slice());
assert_eq!(projected.len(), expected.len());
for ((sequence, bytes, _), row) in projected.iter().zip(expected) {
@@ -233,9 +233,15 @@ async fn automatic_summary_is_hidden_without_interrupting_main_delivery_or_read_
serde_json::to_value(row.message).unwrap()
);
}
- db.execute("DELETE FROM transcript_rows", []).unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.rows", []).unwrap()
+ })
+ .await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
db.execute("DELETE FROM catalog_message_watermark", [])
.unwrap();
log.close().await.unwrap();
@@ -243,7 +249,10 @@ async fn automatic_summary_is_hidden_without_interrupting_main_delivery_or_read_
crate::transcript_cache::materialize_all_rows(&reopened, "session", end.sequence)
.await
.unwrap();
- assert_eq!(transcript(&db), projected);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, transcript).await,
+ projected
+ );
assert!(
reopened
.observe_session::<Value>("session")
diff --git a/platform/event-log/tests/compact_presentation/fixtures.rs b/platform/event-log/tests/compact_presentation/fixtures.rs
index b24aa10f0c..88adc426b9 100644
--- a/platform/event-log/tests/compact_presentation/fixtures.rs
+++ b/platform/event-log/tests/compact_presentation/fixtures.rs
@@ -78,7 +78,7 @@ pub(super) fn accepted(text: &str) -> Fact {
pub(super) fn transcript(
db: &Connection,
) -> Vec<(maka_agent_facts::history::RowPosition, Vec<u8>, String)> {
- db.prepare("SELECT source_sequence, item_index, payload, digest FROM transcript_rows ORDER BY source_sequence, item_index")
+ db.prepare("SELECT source_sequence, item_index, payload, digest FROM transcript_cache.rows ORDER BY source_sequence, item_index")
.unwrap()
.query_map([], |row| Ok((maka_agent_facts::history::RowPosition { source_sequence: row.get::<_, i64>(0)? as u64, item_index: row.get::<_, i64>(1)? as u64 }, row.get(2)?, row.get(3)?)))
.unwrap()
diff --git a/platform/event-log/tests/compact_presentation.rs b/platform/event-log/tests/compact_presentation/mod.rs
similarity index 94%
rename from platform/event-log/tests/compact_presentation.rs
rename to platform/event-log/tests/compact_presentation/mod.rs
index eb89c5f816..337d3cc89a 100644
--- a/platform/event-log/tests/compact_presentation.rs
+++ b/platform/event-log/tests/compact_presentation/mod.rs
@@ -17,9 +17,6 @@
* under the License.
*/
-#[path = "support/transcript.rs"]
-mod transcript_cache;
-
use maka_agent_facts::context::CheckpointMode;
use maka_agent_facts::context::CompactOutcome;
use maka_agent_facts::context::ModelPurpose;
@@ -44,10 +41,10 @@ use maka_presentation::InvocationView;
use rusqlite::Connection;
use serde_json::{Value, json};
-#[path = "compact_presentation/fixtures.rs"]
+#[path = "fixtures.rs"]
mod fixtures;
use fixtures::*;
-#[path = "compact_presentation/automatic.rs"]
+#[path = "automatic.rs"]
mod automatic;
#[tokio::test]
@@ -90,7 +87,7 @@ async fn compact_attempts_preserve_visible_history_read_markers_and_rebuild() {
.await
.unwrap();
let db = Connection::open(&path).unwrap();
- let old_rows = transcript(&db);
+ let old_rows = crate::test_store::inspect(&log, transcript).await;
let compact = identity(mode);
let mut view = InvocationView::new();
hidden(
@@ -190,7 +187,7 @@ async fn compact_attempts_preserve_visible_history_read_markers_and_rebuild() {
crate::transcript_cache::materialize_all_rows(&log, "session", delta.sequence)
.await
.unwrap();
- assert_eq!(transcript(&db), old_rows);
+ assert_eq!(crate::test_store::inspect(&log, transcript).await, old_rows);
let outcome = match mode {
"accepted" | "rejected" => {
hidden(
@@ -293,7 +290,7 @@ async fn compact_attempts_preserve_visible_history_read_markers_and_rebuild() {
crate::transcript_cache::materialize_all_rows(&log, "session", terminal.sequence)
.await
.unwrap();
- assert_eq!(transcript(&db), old_rows);
+ assert_eq!(crate::test_store::inspect(&log, transcript).await, old_rows);
let acknowledged = log
.set_session_read_marker::<Value>("session", &opened.event.id)
.await
@@ -302,9 +299,15 @@ async fn compact_attempts_preserve_visible_history_read_markers_and_rebuild() {
let canonical = log.prefix(100, 1024 * 1024).await.unwrap().digest;
// Reopening rebuilds the discarded projection, preserving canonical bytes.
db.execute("DELETE FROM catalog_messages", []).unwrap();
- db.execute("DELETE FROM transcript_rows", []).unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.rows", []).unwrap()
+ })
+ .await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
let rebuilt = log.get_session::<Value>("session").await.unwrap().unwrap();
@@ -313,7 +316,7 @@ async fn compact_attempts_preserve_visible_history_read_markers_and_rebuild() {
crate::transcript_cache::materialize_all_rows(&log, "session", terminal.sequence)
.await
.unwrap();
- assert_eq!(transcript(&db), old_rows);
+ assert_eq!(crate::test_store::inspect(&log, transcript).await, old_rows);
assert_eq!(
log.prefix(100, 1024 * 1024).await.unwrap().digest,
canonical
diff --git a/platform/event-log/tests/executor_identity.rs b/platform/event-log/tests/executor_identity/mod.rs
similarity index 94%
rename from platform/event-log/tests/executor_identity.rs
rename to platform/event-log/tests/executor_identity/mod.rs
index b721e1518a..fb873a33ad 100644
--- a/platform/event-log/tests/executor_identity.rs
+++ b/platform/event-log/tests/executor_identity/mod.rs
@@ -18,8 +18,6 @@
*/
//! Canonical executor result identities cannot alias accepted client messages.
-#[path = "support/transcript.rs"]
-mod transcript_cache;
use maka_agent_facts::{
event::{
@@ -117,7 +115,7 @@ fn raw(db: &Connection) -> Vec<(String, Vec<u8>)> {
.unwrap()
}
fn cached_rows(db: &Connection) -> Vec<(i64, i64, String, Vec<u8>)> {
- db.prepare("SELECT source_sequence,item_index,message_id,payload FROM transcript_rows WHERE session_id='session' ORDER BY source_sequence,item_index")
+ db.prepare("SELECT source_sequence,item_index,message_id,payload FROM transcript_cache.rows WHERE session_id='session' ORDER BY source_sequence,item_index")
.unwrap().query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))
.unwrap().collect::<Result<_, _>>().unwrap()
}
@@ -327,10 +325,12 @@ async fn executor_closing_rejects_prior_client_result_identity_and_rolls_back()
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert!(
- cached_rows(&db).is_empty(),
+ crate::test_store::inspect(&log, cached_rows)
+ .await
+ .is_empty(),
"canonical guard does not depend on transcript cache"
);
- let before = snapshot(&db);
+ let before = crate::test_store::inspect(&log, snapshot).await;
assert_eq!(before.receipts.len(), 1);
assert!(
before.raw.is_empty(),
@@ -350,12 +350,12 @@ async fn executor_closing_rejects_prior_client_result_identity_and_rolls_back()
.unwrap_err(),
);
assert_eq!(
- snapshot(&db),
+ crate::test_store::inspect(&log, snapshot).await,
before,
"canonical bytes, receipt and transaction prefix must roll back"
);
rejected(log.append(&target).await.unwrap_err());
- assert_eq!(snapshot(&db), before);
+ assert_eq!(crate::test_store::inspect(&log, snapshot).await, before);
assert_eq!(
log.message_submit_receipt("epoch", "session", &id)
.await
@@ -410,8 +410,12 @@ async fn client_source_rejects_executor_result_identity_after_reopen() {
let (legitimate, receipt) = queued(&log, &client, "receipt-control").await;
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
- assert!(cached_rows(&db).is_empty());
- let before = snapshot(&db);
+ assert!(
+ crate::test_store::inspect(&log, cached_rows)
+ .await
+ .is_empty()
+ );
+ let before = crate::test_store::inspect(&log, snapshot).await;
assert_eq!(before.receipts.len(), 1);
assert_eq!(before.pending.len(), 1);
preflight_rejected(
@@ -427,9 +431,9 @@ async fn client_source_rejects_executor_result_identity_after_reopen() {
.await
.unwrap_err(),
);
- assert_eq!(snapshot(&db), before);
+ assert_eq!(crate::test_store::inspect(&log, snapshot).await, before);
rejected(log.append(&target).await.unwrap_err());
- assert_eq!(snapshot(&db), before);
+ assert_eq!(crate::test_store::inspect(&log, snapshot).await, before);
assert_eq!(
log.message_submit_receipt("epoch", "session", "receipt-control")
.await
@@ -520,8 +524,8 @@ async fn executor_result_identity_reserves_only_actual_pending_slots() {
log.append(&claim(&other, &occupied)).await.unwrap();
}
materialize_all(&log, through).await;
- let synthetic: Vec<String> = db.prepare("SELECT message_id FROM transcript_rows WHERE session_id='session' AND source_sequence=? AND json_extract(payload,'$.type')='tool_result' ORDER BY item_index").unwrap()
- .query_map([i64::try_from(complete_sequence).unwrap()], |row| row.get(0)).unwrap().collect::<Result<_,_>>().unwrap();
+ let synthetic: Vec<String> = crate::test_store::inspect(&log, move |db| { db.prepare("SELECT message_id FROM transcript_cache.rows WHERE session_id='session' AND source_sequence=? AND json_extract(payload,'$.type')='tool_result' ORDER BY item_index").unwrap()
+ .query_map([i64::try_from(complete_sequence).unwrap()], |row| row.get(0)).unwrap().collect::<Result<_,_>>().unwrap() }).await;
assert_eq!(
synthetic,
(0..n)
@@ -608,8 +612,12 @@ async fn executor_tool_call_identity_is_atomic_session_scoped_and_reopens() {
};
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
- assert!(cached_rows(&db).is_empty());
- let before = snapshot(&db);
+ assert!(
+ crate::test_store::inspect(&log, cached_rows)
+ .await
+ .is_empty()
+ );
+ let before = crate::test_store::inspect(&log, snapshot).await;
assert_eq!(before.receipts.len(), 1);
if !source_first {
preflight_rejected(
@@ -624,9 +632,9 @@ async fn executor_tool_call_identity_is_atomic_session_scoped_and_reopens() {
.await
.unwrap_err(),
);
- assert_eq!(snapshot(&db), before);
+ assert_eq!(crate::test_store::inspect(&log, snapshot).await, before);
rejected(log.append(&target).await.unwrap_err());
- assert_eq!(snapshot(&db), before);
+ assert_eq!(crate::test_store::inspect(&log, snapshot).await, before);
assert_eq!(
log.message_submit_receipt("epoch", "session", &receipt_id)
.await
diff --git a/platform/event-log/tests/migrations.rs b/platform/event-log/tests/migrations.rs
index 9575fa33cf..15bf9ebeb2 100644
--- a/platform/event-log/tests/migrations.rs
+++ b/platform/event-log/tests/migrations.rs
@@ -109,37 +109,40 @@ async fn initialization_and_reopen_preserve_facts_payloads_and_schema_identity()
|row| row.get(0),
)
.unwrap();
- // Dropping optional bodies preserves committed metadata and accepted work.
- connection
- .execute("DELETE FROM transcript_rows", [])
- .unwrap();
+ // Closing the owner drops optional bodies while preserving accepted work.
+ assert_eq!(connection.query_row("SELECT count(*) FROM sqlite_schema WHERE name IN ('transcript_rows','transcript_text')", [], |row| row.get::<_,i64>(0)).unwrap(), 0);
drop(connection);
let log = EventLog::open(&path).await.unwrap();
let cache = Connection::open(&path).unwrap();
assert_eq!(
cache
- .query_row("SELECT count(*) FROM transcript_rows", [], |row| row
+ .query_row("SELECT count(*) FROM sqlite_schema WHERE name IN ('transcript_rows','transcript_text')", [], |row| row
.get::<_, i64>(0))
.unwrap(),
0
);
- log.transcript_headers(
- "session",
- &maka_event_log::transcript::TranscriptRead {
- through: 1,
- position: None,
- direction: maka_event_log::transcript::TranscriptDirection::Newer,
- limit: 256,
- },
- )
- .await
- .unwrap();
+ let headers = log
+ .transcript_headers(
+ "session",
+ &maka_event_log::transcript::TranscriptRead {
+ through: 1,
+ position: None,
+ direction: maka_event_log::transcript::TranscriptDirection::Newer,
+ limit: 256,
+ },
+ )
+ .await
+ .unwrap();
assert!(
cache
- .query_row("SELECT count(*) FROM transcript_rows", [], |row| row
+ .query_row("SELECT count(*) FROM sqlite_schema WHERE name IN ('transcript_rows','transcript_text')", [], |row| row
.get::<_, i64>(0))
.unwrap()
- > 0
+ == 0
+ );
+ assert!(
+ !headers.is_empty(),
+ "public reads reconstruct rows from canonical facts"
);
drop(cache);
assert_eq!(
diff --git a/platform/event-log/tests/migrations/prior_artifacts.rs b/platform/event-log/tests/migrations/prior_artifacts.rs
index 59504a915c..13f653b4b2 100644
--- a/platform/event-log/tests/migrations/prior_artifacts.rs
+++ b/platform/event-log/tests/migrations/prior_artifacts.rs
@@ -479,7 +479,7 @@ async fn migration_37_preserves_interaction_receipts_and_relations_without_json_
0
))
.unwrap(),
- 37
+ 38
);
for ((table, columns), original) in tables.iter().zip(original) {
assert_eq!(rows(&db, table, columns), original);
@@ -718,7 +718,7 @@ async fn migration_37_preserves_artifact_metadata_and_copied_image_provenance()
0
))
.unwrap(),
- 37
+ 38
);
assert_eq!(artifacts(&db), prior);
assert_eq!(mappings(&db), provenance);
diff --git a/platform/event-log/tests/navigation.rs b/platform/event-log/tests/navigation.rs
index 34b4835095..a32c5dfacc 100644
--- a/platform/event-log/tests/navigation.rs
+++ b/platform/event-log/tests/navigation.rs
@@ -338,24 +338,33 @@ async fn navigation_anchors_fixed_fence_byte_pages_and_reopen() {
.unwrap();
assert!(first_page.next_position.is_some());
// Every landmark is an actual persisted user row, not an opening/header guess.
- let db = rusqlite::Connection::open(&path).unwrap();
for landmark in log
.navigation_landmarks("session", current, 64, None)
.await
.unwrap()
{
- let payload: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE source_sequence=? AND item_index=? AND session_id='session'",
- [landmark.position.source_sequence as i64, landmark.position.item_index as i64],
- |row| row.get(0),
+ let header = log
+ .transcript_headers(
+ "session",
+ &maka_event_log::transcript::TranscriptRead {
+ through: current,
+ position: Some(landmark.position),
+ direction: maka_event_log::transcript::TranscriptDirection::Newer,
+ limit: 1,
+ },
)
+ .await
+ .unwrap()
+ .remove(0);
+ assert_eq!(header.position, landmark.position);
+ let payload = log
+ .transcript_fragment("session", header.position, 0, header.total_bytes)
+ .await
.unwrap();
let row: serde_json::Value = serde_json::from_slice(&payload).unwrap();
assert_eq!(row["type"], "user");
assert_eq!(row["turnId"], landmark.turn_id);
}
- drop(db);
log.close().await.unwrap();
let reopened = EventLog::open(&path).await.unwrap();
assert_eq!(pages(&reopened, fence, 1).await, initial);
diff --git a/platform/event-log/tests/root_messages.rs b/platform/event-log/tests/root_messages.rs
index 98ee602f8b..27dd4d4e58 100644
--- a/platform/event-log/tests/root_messages.rs
+++ b/platform/event-log/tests/root_messages.rs
@@ -353,7 +353,8 @@ async fn root_sources_are_atomic_exclusive_delivery_proofs_and_rebuild_exact_vis
let rows = support::transcript(&log, "a").await;
let canonical = serde_json::to_vec(&log.prefix(100, 1024 * 1024).await.unwrap()).unwrap();
log.close().await.unwrap();
- db.execute_batch("DELETE FROM message_sources; DELETE FROM catalog_messages; DELETE FROM transcript_rows; DELETE FROM transcript_text; DELETE FROM transcript_rows;").unwrap();
+ db.execute_batch("DELETE FROM message_sources; DELETE FROM catalog_messages;")
+ .unwrap();
drop(db);
let log = EventLog::open(&path).await.unwrap();
assert_eq!(
diff --git a/platform/event-log/tests/row_directory.rs b/platform/event-log/tests/row_directory/mod.rs
similarity index 97%
rename from platform/event-log/tests/row_directory.rs
rename to platform/event-log/tests/row_directory/mod.rs
index b23b566443..1425d2353b 100644
--- a/platform/event-log/tests/row_directory.rs
+++ b/platform/event-log/tests/row_directory/mod.rs
@@ -238,12 +238,15 @@ mod bare_copy {
.unwrap();
assert_eq!(members, 2);
assert_eq!(
- db.query_row(
- "SELECT count(*) FROM transcript_rows WHERE session_id='target'",
- [],
- |r| r.get::<_, i64>(0)
- )
- .unwrap(),
+ crate::test_store::inspect(&log, |db| {
+ db.query_row(
+ "SELECT count(*) FROM transcript_cache.rows WHERE session_id='target'",
+ [],
+ |r| r.get::<_, i64>(0),
+ )
+ .unwrap()
+ })
+ .await,
0
);
if collision {
diff --git a/platform/event-log/tests/session_copy.rs b/platform/event-log/tests/session_copy/mod.rs
similarity index 98%
rename from platform/event-log/tests/session_copy.rs
rename to platform/event-log/tests/session_copy/mod.rs
index bc2a09450d..185b2d47fb 100644
--- a/platform/event-log/tests/session_copy.rs
+++ b/platform/event-log/tests/session_copy/mod.rs
@@ -17,9 +17,6 @@
* under the License.
*/
-#[path = "support/transcript.rs"]
-mod transcript_cache;
-
use maka_agent_facts::artifact::Artifact;
use maka_agent_facts::artifact::ArtifactKind;
use maka_agent_facts::artifact::ArtifactSource;
@@ -698,16 +695,18 @@ async fn copies_own_history_and_files_without_replaying_execution_across_retries
relative_path: files[0].id.clone(),
}
);
- let inspect = rusqlite::Connection::open(&path).unwrap();
- for table in ["transcript_text", "transcript_rows"] {
- inspect
- .execute(
- &format!("DELETE FROM {table} WHERE session_id = ?"),
- [session],
- )
- .unwrap();
- }
- drop(inspect);
+ let cache_session = session.to_owned();
+ crate::test_store::inspect(&log, move |inspect| {
+ for table in ["transcript_cache.text", "transcript_cache.rows"] {
+ inspect
+ .execute(
+ &format!("DELETE FROM {table} WHERE session_id = ?"),
+ [&cache_session],
+ )
+ .unwrap();
+ }
+ })
+ .await;
crate::transcript_cache::materialize_all_rows(&log, session, root.high_water)
.await
.unwrap();
diff --git a/platform/event-log/tests/steering.rs b/platform/event-log/tests/steering.rs
index 3d4be113a6..bd7ba71907 100644
--- a/platform/event-log/tests/steering.rs
+++ b/platform/event-log/tests/steering.rs
@@ -261,7 +261,7 @@ async fn steering_is_exact_once_session_scoped_and_atomic_with_rebuildable_user_
log.close().await.unwrap();
// Both read models are disposable; the log is the only message-delivery proof.
db.execute_batch(
- "DELETE FROM transcript_rows; DELETE FROM transcript_text; DELETE FROM transcript_rows;
+ "
DELETE FROM catalog_messages; DELETE FROM catalog_message_watermark;",
)
.unwrap();
diff --git a/platform/event-log/src/transcript.rs b/platform/event-log/tests/support/owner.rs
similarity index 50%
copy from platform/event-log/src/transcript.rs
copy to platform/event-log/tests/support/owner.rs
index 5447ba2870..26edaea476 100644
--- a/platform/event-log/src/transcript.rs
+++ b/platform/event-log/tests/support/owner.rs
@@ -17,24 +17,25 @@
* under the License.
*/
-//! Durable row metadata and optional body caches; authority stays in the canonical ledger.
-mod active;
-mod compaction;
-pub(crate) use active::interrupted_messages;
-mod bodycache;
-pub(crate) mod directory;
-mod emit;
-mod evidence;
-mod final_output;
-mod history;
-mod image;
-mod model;
-pub mod navigation;
-mod prompt;
-mod read;
-pub mod search;
-mod source;
-mod timing;
-mod tool_metadata;
-mod tools;
-pub use read::*;
+use crate::EventLog;
+
+/// Run test inspection/fault injection on the original serialized writer. No
+/// borrowed SQLite state or connection escapes the native worker callback.
+pub(crate) async fn inspect<T: Send + 'static>(
+ log: &EventLog,
+ operation: impl FnOnce(&rusqlite::Connection) -> T + Send + 'static,
+) -> T {
+ log.connection
+ .run(move |db| {
+ Box::pin(async move {
+ db.run_on_worker(move |handle| {
+ // SAFETY: SQLx owns this callback's handle and fences its statements.
+ let borrowed = unsafe { rusqlite::Connection::from_handle(handle.as_ptr()) }?;
+ Ok(operation(&borrowed))
+ })
+ .await?
+ })
+ })
+ .await
+ .unwrap()
+}
diff --git a/platform/event-log/tests/transcript_index/accepted_observations.rs b/platform/event-log/tests/transcript_index/accepted_observations.rs
index 7b9dcccd42..99ff89e63f 100644
--- a/platform/event-log/tests/transcript_index/accepted_observations.rs
+++ b/platform/event-log/tests/transcript_index/accepted_observations.rs
@@ -183,8 +183,16 @@ async fn accepted_kinds_publish_at_observation_and_unaccepted_kinds_wait_for_sea
crate::transcript_cache::materialize_all_rows(&log, "a", unaccepted)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected(&events, unaccepted));
- assert_eq!(rows(&db, "a").len(), 2);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected(&events, unaccepted)
+ );
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .len(),
+ 2
+ );
let unaccepted_page = page(&log, unaccepted).await;
let mut cold = Vec::new();
for (index, &cut) in cuts.iter().enumerate() {
@@ -192,7 +200,7 @@ async fn accepted_kinds_publish_at_observation_and_unaccepted_kinds_wait_for_sea
.await
.unwrap();
- let actual = rows(&db, "a");
+ let actual = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(actual, expected(&events, cut));
assert_eq!(actual.len(), if accepted { 3 + index } else { 2 });
if accepted {
@@ -220,7 +228,10 @@ async fn accepted_kinds_publish_at_observation_and_unaccepted_kinds_wait_for_sea
crate::transcript_cache::materialize_all_rows(&log, "a", later)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected(&events, later));
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected(&events, later)
+ );
for (&cut, original) in cuts.iter().zip(&cold) {
assert_eq!(page(&log, cut).await, *original);
}
@@ -233,7 +244,13 @@ async fn accepted_kinds_publish_at_observation_and_unaccepted_kinds_wait_for_sea
assert_eq!(page(&reopened, cut).await, *original);
}
let db = Connection::open(&path).unwrap();
- assert_eq!(rows(&db, "a"), expected(&events, later));
+ crate::transcript_cache::materialize_all_rows(&reopened, "a", later)
+ .await
+ .unwrap();
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ expected(&events, later)
+ );
assert_eq!(page(&reopened, unaccepted).await, unaccepted_page);
assert_eq!(canonical_bytes(&db), canonical);
@@ -283,6 +300,11 @@ async fn accepted_observation_directory_survives_missing_optional_bodies_and_reo
let (cuts, _, events) = seed(&log, true, false).await;
let old = cuts[0];
let original = page(&log, old).await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let db = Connection::open(&path).unwrap();
let canonical = canonical_bytes(&db);
@@ -293,15 +315,21 @@ async fn accepted_observation_directory_survives_missing_optional_bodies_and_reo
|r| r.get(0),
)
.unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+
let reopened = EventLog::open(&path).await.unwrap();
- assert!(rows(&db, "a").is_empty());
+ assert!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a"))
+ .await
+ .is_empty()
+ );
assert_eq!(page(&reopened, old).await, original);
crate::transcript_cache::materialize_all_rows(&reopened, "a", old)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected(&events, old));
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ expected(&events, old)
+ );
assert_eq!(
db.query_row(
"SELECT count(*) FROM transcript_directory WHERE session_id='a'",
diff --git a/platform/event-log/tests/transcript_index/fixtures.rs b/platform/event-log/tests/transcript_index/fixtures.rs
index ac2aaa49f1..aa33bc1c96 100644
--- a/platform/event-log/tests/transcript_index/fixtures.rs
+++ b/platform/event-log/tests/transcript_index/fixtures.rs
@@ -116,7 +116,7 @@ pub(super) async fn end(log: &EventLog, session: &str, outcome: InvocationOutcom
pub(super) type Row = (RowPosition, Vec<u8>, String, i64);
pub(super) fn rows(db: &Connection, session: &str) -> Vec<Row> {
db.prepare(
- "SELECT source_sequence, item_index, payload, digest, total_bytes FROM transcript_rows
+ "SELECT source_sequence, item_index, payload, digest, total_bytes FROM transcript_cache.rows
WHERE session_id = ? ORDER BY source_sequence, item_index",
)
.unwrap()
diff --git a/platform/event-log/tests/transcript_index/history.rs b/platform/event-log/tests/transcript_index/history.rs
index f7869ca8b6..699dcca11f 100644
--- a/platform/event-log/tests/transcript_index/history.rs
+++ b/platform/event-log/tests/transcript_index/history.rs
@@ -269,22 +269,22 @@ async fn history_pages_preserve_utf8_nul_suffixes_archive_and_frozen_fences() {
.await
.unwrap();
let cache = rusqlite::Connection::open(&path).unwrap();
- cache
+ crate::test_store::inspect(&log, move |cache| { cache
.execute(
- "INSERT INTO transcript_rows (source_sequence, item_index, session_id, turn_id, message_id, payload, digest, total_bytes)
+ "INSERT INTO transcript_cache.rows (source_sequence, item_index, session_id, turn_id, message_id, payload, digest, total_bytes)
SELECT source_sequence, item_index, 'b', turn_id, message_id, payload, digest, total_bytes
- FROM transcript_rows WHERE session_id = 'a'",
+ FROM transcript_cache.rows WHERE session_id = 'a'",
[],
)
- .unwrap();
- cache
+ .unwrap() }).await;
+ crate::test_store::inspect(&log, move |cache| { cache
.execute(
- "INSERT INTO transcript_text
+ "INSERT INTO transcript_cache.text
SELECT 'b', source_sequence, item_index, field, timestamp, role, CASE WHEN role IN ('user', 'assistant', 'tool_call', 'tool_result') THEN CAST('destination cache' AS BLOB) ELSE body END
- FROM transcript_text WHERE session_id = 'a'",
+ FROM transcript_cache.text WHERE session_id = 'a'",
[],
)
- .unwrap();
+ .unwrap() }).await;
let Page::Ready {
chunks: destination,
..
@@ -304,9 +304,15 @@ async fn history_pages_preserve_utf8_nul_suffixes_archive_and_frozen_fences() {
serde_json::to_value(source).unwrap(),
serde_json::to_value(chunks).unwrap()
);
- cache
- .execute("DELETE FROM transcript_text WHERE session_id = 'b'", [])
- .unwrap();
+ crate::test_store::inspect(&log, move |cache| {
+ cache
+ .execute(
+ "DELETE FROM transcript_cache.text WHERE session_id = 'b'",
+ [],
+ )
+ .unwrap()
+ })
+ .await;
let rebuilt = log
.history_text("b", old, Position::Start, false, 16 * 1024, 8, 1024 * 1024)
.await
@@ -617,12 +623,15 @@ async fn tool_metadata_preserves_empty_calls_and_non_text_results_through_frozen
}
let original_events =
serde_json::to_vec(&log.prefix(1000, 1024 * 1024).await.unwrap()).unwrap();
- log.close().await.unwrap();
let db = Connection::open(&path).unwrap();
- let canonical = rows(&db, "numeric");
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ let canonical = crate::test_store::inspect(&log, |db| rows(db, "numeric")).await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
drop(db);
+ log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert_eq!(
serde_json::to_vec(&log.prefix(1000, 1024 * 1024).await.unwrap()).unwrap(),
@@ -630,9 +639,13 @@ async fn tool_metadata_preserves_empty_calls_and_non_text_results_through_frozen
);
let db = Connection::open(&path).unwrap();
assert_eq!(
- db.query_row("SELECT count(*) FROM transcript_rows", [], |row| row
- .get::<_, i64>(0))
- .unwrap(),
+ crate::test_store::inspect(&log, move |db| {
+ db.query_row("SELECT count(*) FROM transcript_cache.rows", [], |row| {
+ row.get::<_, i64>(0)
+ })
+ .unwrap()
+ })
+ .await,
0
);
drop(db);
@@ -642,7 +655,10 @@ async fn tool_metadata_preserves_empty_calls_and_non_text_results_through_frozen
crate::transcript_cache::materialize_all_rows(&log, "numeric", through)
.await
.unwrap();
- assert_eq!(rows(&db, "numeric"), canonical);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "numeric")).await,
+ canonical
+ );
drop(db);
let Page::Ready { chunks, .. } = log
.history_text(
@@ -943,7 +959,10 @@ async fn bounded_history_windows_cover_sparse_rows_both_directions_and_validate_
for budget in [4, 5, 7, 13, 128 * 1024] {
for reverse in [false, true] {
let db = Connection::open(&path).unwrap();
- db.execute("DELETE FROM transcript_text", []).unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.text", []).unwrap()
+ })
+ .await;
drop(db);
let mut position = if reverse {
Position::Tail
@@ -1581,34 +1600,65 @@ async fn message_material_labels_and_time_survive_exact_paging_and_text_cache_lo
"display paths perform no I/O"
);
let before = serde_json::to_vec(&log.prefix(10, 65536).await.unwrap()).unwrap();
- log.close().await.unwrap();
let db = Connection::open(&path).unwrap();
- let canonical = sources
+ let sessions: Vec<String> = sources
.iter()
- .map(|(session, ..)| rows(&db, session))
- .collect::<Vec<_>>();
- db.execute(
- "UPDATE transcript_text SET body = CAST('old derived text' AS BLOB)",
- [],
- )
- .unwrap();
- db.execute("DELETE FROM transcript_text", []).unwrap();
+ .map(|(session, ..)| session.to_string())
+ .collect();
+ let canonical = crate::test_store::inspect(&log, move |db| {
+ sessions
+ .iter()
+ .map(|session| rows(db, session))
+ .collect::<Vec<_>>()
+ })
+ .await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute(
+ "UPDATE transcript_cache.text SET body = CAST('old derived text' AS BLOB)",
+ [],
+ )
+ .unwrap()
+ })
+ .await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.text", []).unwrap()
+ })
+ .await;
drop(db);
+ log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
let db = Connection::open(&path).unwrap();
assert_eq!(
- db.query_row("SELECT count(*) FROM transcript_text", [], |row| row
- .get::<_, i64>(0))
- .unwrap(),
+ crate::test_store::inspect(&log, move |db| {
+ db.query_row("SELECT count(*) FROM transcript_cache.text", [], |row| {
+ row.get::<_, i64>(0)
+ })
+ .unwrap()
+ })
+ .await,
0
);
- assert_eq!(
- sources
- .iter()
- .map(|(session, ..)| rows(&db, session))
- .collect::<Vec<_>>(),
- canonical
- );
+ for ((session, through, ..), expected_rows) in sources.iter().zip(&canonical) {
+ assert!(
+ crate::test_store::inspect(&log, {
+ let session = session.to_string();
+ move |db| rows(db, &session)
+ })
+ .await
+ .is_empty()
+ );
+ crate::transcript_cache::materialize_all_rows(&log, session, *through)
+ .await
+ .unwrap();
+ assert_eq!(
+ crate::test_store::inspect(&log, {
+ let session = session.to_string();
+ move |db| rows(db, &session)
+ })
+ .await,
+ *expected_rows
+ );
+ }
assert_eq!(
serde_json::to_vec(&log.prefix(10, 65536).await.unwrap()).unwrap(),
before
@@ -1746,7 +1796,10 @@ async fn thinking_windows_preserve_field_order_bytes_and_explicit_body_only_read
for reverse in [false, true] {
for budget in [4, 13, 8192] {
let db = Connection::open(&path).unwrap();
- db.execute("DELETE FROM transcript_text", []).unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.text", []).unwrap()
+ })
+ .await;
drop(db);
let mut position = if reverse {
Position::Tail
@@ -1929,13 +1982,16 @@ async fn thinking_windows_preserve_field_order_bytes_and_explicit_body_only_read
.is_err(),
"Thinking offsets must not split UTF-8"
);
- log.close().await.unwrap();
// Reopen with optional bodies/text missing; preserve canonical identities,
// fences and both fields without rewriting directory metadata.
let db = Connection::open(&path).unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
drop(db);
+ log.close().await.unwrap();
let reopened = EventLog::open(&path).await.unwrap();
let Page::Ready {
@@ -2081,16 +2137,19 @@ async fn failed_turn_and_retry_diagnostics_are_visible_with_exact_sources_and_fe
["Connection lost 中😀; retry permitted"]
);
}
- log.close().await.unwrap();
// Discard diagnostic text only; reconstruct the requested failure body.
let db = Connection::open(&path).unwrap();
- db.execute(
- "UPDATE transcript_text SET body = x'' WHERE role IN ('turn_state', 'model_interruption')",
+ crate::test_store::inspect(&log, move |db| { db.execute(
+ "UPDATE transcript_cache.text SET body = x'' WHERE role IN ('turn_state', 'model_interruption')",
[],
)
- .unwrap();
- db.execute("DELETE FROM transcript_text", []).unwrap();
+ .unwrap() }).await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.text", []).unwrap()
+ })
+ .await;
drop(db);
+ log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
let Page::Ready { chunks, .. } = log
.history_text("failed", through, Position::Tail, false, 8192, 16, 8192)
diff --git a/platform/event-log/tests/transcript_index/image.rs b/platform/event-log/tests/transcript_index/image.rs
index 41bd4c48fe..a35e148554 100644
--- a/platform/event-log/tests/transcript_index/image.rs
+++ b/platform/event-log/tests/transcript_index/image.rs
@@ -248,6 +248,7 @@ async fn actual_tool_images_are_immutable_scoped_and_reconstructed_without_paths
.bytes,
bytes[..8192]
);
+ let copied_id: String = crate::test_store::inspect(&log, move |db| { db.query_row("SELECT json_extract(payload,'$.content.ref.relativePath') FROM transcript_cache.rows WHERE session_id='copy' AND source_sequence=? AND item_index=?",[copied_sequence.source_sequence as i64, copied_sequence.item_index as i64],|row|row.get(0)).unwrap() }).await;
assert!(!dir.path().join(&artifact.id).exists());
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
@@ -294,7 +295,7 @@ async fn actual_tool_images_are_immutable_scoped_and_reconstructed_without_paths
.is_err()
);
db.execute("UPDATE artifacts SET record_json=json_set(record_json,'$.mimeType','image/gif','$.source','user_upload') WHERE session_id='images' AND id=?",[&artifact.id]).unwrap();
- let copied_id: String = db.query_row("SELECT json_extract(payload,'$.content.ref.relativePath') FROM transcript_rows WHERE session_id='copy' AND source_sequence=? AND item_index=?",[copied_sequence.source_sequence as i64, copied_sequence.item_index as i64],|row|row.get(0)).unwrap();
+
db.execute("UPDATE artifacts SET record_json=json_set(record_json,'$.source','user_upload') WHERE session_id='copy' AND id=?",[&copied_id]).unwrap();
assert!(
log.tool_image("copy", copy_fence, copied_sequence, 0, 8192)
diff --git a/platform/event-log/tests/transcript_index/interrupted_ancestry.rs b/platform/event-log/tests/transcript_index/interrupted_ancestry.rs
index 60f51bf6d9..1eee7057d1 100644
--- a/platform/event-log/tests/transcript_index/interrupted_ancestry.rs
+++ b/platform/event-log/tests/transcript_index/interrupted_ancestry.rs
@@ -93,7 +93,12 @@ async fn interrupted_streaming_tool_ancestry_and_later_model_step_replay_exactly
.unwrap();
let db = Connection::open(&path).unwrap();
- assert_eq!(rows(&db, "a").len(), 4);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .len(),
+ 4
+ );
// The ordinary Message main-request append contract permits this order.
// Current engine direct-tool cleanup is stricter; this test covers the log.
@@ -175,7 +180,10 @@ async fn interrupted_streaming_tool_ancestry_and_later_model_step_replay_exactly
let prepared = crate::transcript_cache::materialize_all_rows(&log, "a", through).await;
prepared.expect("append-accepted evidence must preserve the selected-boundary contract");
- assert_eq!(rows(&db, "a"), expected);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected
+ );
drop(db);
log.close().await.unwrap();
}
diff --git a/platform/event-log/tests/transcript_index/large_siblings.rs b/platform/event-log/tests/transcript_index/large_siblings.rs
index a13917283f..da61283929 100644
--- a/platform/event-log/tests/transcript_index/large_siblings.rs
+++ b/platform/event-log/tests/transcript_index/large_siblings.rs
@@ -77,7 +77,7 @@ async fn source(
// Retain only identities/digests, never a second copy of all large row payloads.
fn fingerprints(db: &Connection) -> Vec<(i64, i64, String, i64)> {
db.prepare(
- "SELECT source_sequence, item_index, digest, total_bytes FROM transcript_rows
+ "SELECT source_sequence, item_index, digest, total_bytes FROM transcript_cache.rows
WHERE session_id = ? ORDER BY source_sequence, item_index",
)
.unwrap()
@@ -197,8 +197,7 @@ async fn large_sibling_arguments_preserve_exact_history_sources_across_fences_an
crate::transcript_cache::materialize_all_rows(&log, SESSION, through)
.await
.unwrap();
- let db = Connection::open(&path).unwrap();
- let original = fingerprints(&db);
+ let original = crate::test_store::inspect(&log, fingerprints).await;
assert_eq!(
original
.iter()
@@ -409,15 +408,21 @@ async fn large_sibling_arguments_preserve_exact_history_sources_across_fences_an
);
};
check(&log).await;
+ crate::test_store::inspect(&log, |db| {
+ for table in ["transcript_cache.text", "transcript_cache.rows"] {
+ db.execute(&format!("DELETE FROM {table}"), []).unwrap();
+ }
+ })
+ .await;
log.close().await.unwrap();
- for table in ["transcript_text", "transcript_rows"] {
- db.execute(&format!("DELETE FROM {table}"), []).unwrap();
- }
let reopened = EventLog::open(&path).await.unwrap();
crate::transcript_cache::materialize_all_rows(&reopened, SESSION, through)
.await
.unwrap();
- assert_eq!(fingerprints(&db), original);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, fingerprints).await,
+ original
+ );
check(&reopened).await;
reopened.close().await.unwrap();
}
diff --git a/platform/event-log/tests/transcript_index.rs b/platform/event-log/tests/transcript_index/mod.rs
similarity index 85%
rename from platform/event-log/tests/transcript_index.rs
rename to platform/event-log/tests/transcript_index/mod.rs
index 7f0ad05236..879d090365 100644
--- a/platform/event-log/tests/transcript_index.rs
+++ b/platform/event-log/tests/transcript_index/mod.rs
@@ -17,9 +17,6 @@
* under the License.
*/
-#[path = "support/transcript.rs"]
-mod transcript_cache;
-
use maka_agent_facts::event::Fact;
use maka_agent_facts::event::Invocation;
use maka_agent_facts::event::InvocationInput;
@@ -38,29 +35,29 @@ use rusqlite::Connection;
use serde_json::Value;
use sha2::{Digest, Sha256};
-#[path = "transcript_index/accepted_observations.rs"]
+#[path = "accepted_observations.rs"]
mod accepted_observations;
-#[path = "transcript_index/fixtures.rs"]
+#[path = "fixtures.rs"]
mod fixtures;
-#[path = "transcript_index/history.rs"]
+#[path = "history.rs"]
mod history;
-#[path = "transcript_index/image.rs"]
+#[path = "image.rs"]
mod image;
-#[path = "transcript_index/interrupted_ancestry.rs"]
+#[path = "interrupted_ancestry.rs"]
mod interrupted_ancestry;
-#[path = "transcript_index/large_siblings.rs"]
+#[path = "large_siblings.rs"]
mod large_siblings;
-#[path = "transcript_index/observation_count.rs"]
+#[path = "observation_count.rs"]
mod observation_count;
-#[path = "transcript_index/performance.rs"]
+#[path = "performance.rs"]
mod performance;
-#[path = "transcript_index/performance_executor.rs"]
+#[path = "performance_executor.rs"]
mod performance_executor;
-#[path = "transcript_index/prefix_replay.rs"]
+#[path = "prefix_replay.rs"]
mod prefix_replay;
-#[path = "transcript_index/source.rs"]
+#[path = "source.rs"]
mod source;
-#[path = "transcript_index/sparse_directory.rs"]
+#[path = "sparse_directory.rs"]
mod sparse_directory;
use fixtures::*;
@@ -131,7 +128,7 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
.unwrap();
assert!(!wake.has_changed().unwrap());
let db = Connection::open(&path).unwrap();
- let original = rows(&db, "a");
+ let original = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(original.len(), 5);
let assistant: Value = serde_json::from_slice(&original[2].1).unwrap();
assert_eq!(assistant["id"], id);
@@ -144,18 +141,27 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
);
}
// Optional bodies may be discarded; selected reconstruction preserves the same rows.
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), original);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ original
+ );
log.close().await.unwrap();
let reopened = EventLog::open(&path).await.unwrap();
crate::transcript_cache::materialize_all_rows(&reopened, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), original);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ original
+ );
let canonical = serde_json::to_vec(&reopened.prefix(32, 65536).await.unwrap()).unwrap();
// This real completed producer emits assistant then usage. Derive positions
// from the emitted fixture, rather than assuming fixed event/ordinal numbers.
@@ -182,13 +188,13 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
let usage_item = i64::try_from(usage_row.0.item_index).unwrap();
// Preserve the original byte-identity refusal: a missing sibling triggers
// whole-S reconstruction, which must not overwrite an existing bad payload.
- db.execute("UPDATE transcript_rows SET payload=X'00' WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[assistant_source,assistant_item]).unwrap();
- db.execute(
- "DELETE FROM transcript_rows WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",
+ crate::test_store::inspect(&reopened, move |db| { db.execute("UPDATE transcript_cache.rows SET payload=X'00' WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[assistant_source,assistant_item]).unwrap() }).await;
+ crate::test_store::inspect(&reopened, move |db| { db.execute(
+ "DELETE FROM transcript_cache.rows WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",
[usage_source, usage_item],
)
- .unwrap();
- let damaged = rows(&db, "a");
+ .unwrap() }).await;
+ let damaged = crate::test_store::inspect(&reopened, |db| rows(db, "a")).await;
assert!(matches!(
reopened
.transcript_headers(
@@ -203,22 +209,30 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
.await,
Err(StoreError::TranscriptConflict)
));
- assert_eq!(rows(&db, "a"), damaged);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ damaged
+ );
assert_eq!(
serde_json::to_vec(&reopened.prefix(32, 65536).await.unwrap()).unwrap(),
canonical
);
- db.execute("UPDATE transcript_rows SET payload=?1 WHERE session_id='a' AND source_sequence=?2 AND item_index=?3",rusqlite::params![&assistant_row.1,assistant_source,assistant_item]).unwrap();
+ let restore_payload = assistant_row.1.clone();
+ crate::test_store::inspect(&reopened, move |db| { db.execute("UPDATE transcript_cache.rows SET payload=?1 WHERE session_id='a' AND source_sequence=?2 AND item_index=?3",rusqlite::params![restore_payload,assistant_source,assistant_item]).unwrap() }).await;
crate::transcript_cache::materialize_all_rows(&reopened, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), original);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ original
+ );
// Cache hits retain their separate directory-identity check without hashing
// every cached payload on each fragment.
- db.execute(
- "UPDATE transcript_rows SET turn_id = 'wrong-turn' WHERE source_sequence = ? AND item_index = ?",
- [i64::try_from(original[1].0.source_sequence).unwrap(),i64::try_from(original[1].0.item_index).unwrap()],
- ).unwrap();
+ let identity_position = original[1].0;
+ crate::test_store::inspect(&reopened, move |db| { db.execute(
+ "UPDATE transcript_cache.rows SET turn_id = 'wrong-turn' WHERE source_sequence = ? AND item_index = ?",
+ [i64::try_from(identity_position.source_sequence).unwrap(),i64::try_from(identity_position.item_index).unwrap()],
+ ).unwrap() }).await;
assert!(matches!(
crate::transcript_cache::materialize_all_rows(&reopened, "a", through).await,
Err(StoreError::TranscriptConflict)
@@ -228,25 +242,21 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
canonical
);
reopened.close().await.unwrap();
- db.execute(
- "UPDATE transcript_rows SET turn_id='turn-a' WHERE session_id='a'",
- [],
- )
- .unwrap();
db.close().map_err(|(_, error)| error).unwrap();
// All connections are now closed (including the last SQLite connection's
// WAL close/checkpoint). Copy this task fixture, never a live-WAL DB or hardlink.
for operation in ["headers", "fragment", "turn_bounds"] {
let copy = temp.path().join(format!("{operation}.sqlite"));
std::fs::copy(&path, ©).unwrap();
- let fault = Connection::open(©).unwrap();
+ let copied = EventLog::open(©).await.unwrap();
+ crate::transcript_cache::materialize_all_rows(&copied, "a", through)
+ .await
+ .unwrap();
// A miss inserts the first emitted row, then the existing second row
// fails the exact byte comparison. Only optional cache bytes are damaged.
- fault.execute("DELETE FROM transcript_rows WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[assistant_source,assistant_item]).unwrap();
- fault.execute("UPDATE transcript_rows SET payload=X'00' WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[usage_source,usage_item]).unwrap();
- let before = rows(&fault, "a");
- fault.close().map_err(|(_, error)| error).unwrap();
- let copied = EventLog::open(©).await.unwrap();
+ crate::test_store::inspect(&copied, move |fault| { fault.execute("DELETE FROM transcript_cache.rows WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[assistant_source,assistant_item]).unwrap() }).await;
+ crate::test_store::inspect(&copied, move |fault| { fault.execute("UPDATE transcript_cache.rows SET payload=X'00' WHERE session_id='a' AND source_sequence=?1 AND item_index=?2",[usage_source,usage_item]).unwrap() }).await;
+ let before = crate::test_store::inspect(&copied, |db| rows(db, "a")).await;
let error = match operation {
"headers" => copied
.transcript_headers(
@@ -278,13 +288,12 @@ async fn completed_bytes_digest_identity_reopen_and_selected_rebuild() {
serde_json::to_vec(&copied.prefix(32, 65536).await.unwrap()).unwrap(),
canonical
);
- copied.close().await.unwrap();
- let after = Connection::open(©).unwrap();
assert_eq!(
- rows(&after, "a"),
+ crate::test_store::inspect(&copied, |db| rows(db, "a")).await,
before,
"{operation}: failed source fill must roll back its first row"
);
+ copied.close().await.unwrap();
}
}
@@ -322,8 +331,7 @@ async fn interleaved_stream_preserves_order_and_excludes_other_session_evidence(
.await
.unwrap();
- let db = Connection::open(&path).unwrap();
- let projected = rows(&db, "a");
+ let projected = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(projected.len(), 3);
let assistant: Value = serde_json::from_slice(&projected[2].1).unwrap();
assert_eq!(assistant["id"], id);
@@ -333,13 +341,16 @@ async fn interleaved_stream_preserves_order_and_excludes_other_session_evidence(
crate::transcript_cache::materialize_all_rows(&log, "b", other)
.await
.unwrap();
- let other_rows = rows(&db, "b");
+ let other_rows = crate::test_store::inspect(&log, |db| rows(db, "b")).await;
let assistant: Value = serde_json::from_slice(&other_rows[2].1).unwrap();
assert_eq!(
assistant["text"],
format!("unrelated{}", "not part of a".repeat(1024))
);
- assert_eq!(rows(&db, "a"), projected);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ projected
+ );
}
#[tokio::test]
@@ -357,7 +368,6 @@ async fn partial_freezes_at_interruption_and_old_fence_excludes_later_boundaries
},
)
.await;
- let db = Connection::open(&path).unwrap();
let invocation = event("a", opening()).event().invocation.clone();
let overlay = log.active_transcript(&invocation, live).await.unwrap();
assert_eq!(overlay.len(), 1);
@@ -377,7 +387,7 @@ async fn partial_freezes_at_interruption_and_old_fence_excludes_later_boundaries
crate::transcript_cache::materialize_all_rows(&log, "a", live)
.await
.unwrap();
- let running = rows(&db, "a");
+ let running = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(running.len(), 2);
let state: Value = serde_json::from_slice(&running[1].1).unwrap();
assert_eq!(state["type"], "turn_state");
@@ -389,7 +399,7 @@ async fn partial_freezes_at_interruption_and_old_fence_excludes_later_boundaries
crate::transcript_cache::materialize_all_rows(&log, "a", interrupted)
.await
.unwrap();
- let frozen = rows(&db, "a");
+ let frozen = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(frozen.len(), 3);
let assistant: Value = serde_json::from_slice(&frozen[2].1).unwrap();
assert_eq!(assistant["id"], id);
@@ -397,7 +407,7 @@ async fn partial_freezes_at_interruption_and_old_fence_excludes_later_boundaries
crate::transcript_cache::materialize_all_rows(&log, "a", ended)
.await
.unwrap();
- let final_rows = rows(&db, "a");
+ let final_rows = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(&final_rows[..3], frozen.as_slice());
assert_eq!(final_rows.len(), 4); // no invented unknown usage
let terminal: Value = serde_json::from_slice(&final_rows[3].1).unwrap();
@@ -413,19 +423,33 @@ async fn directory_fences_and_sparse_body_session_isolation() {
log.append(&event("b", opening())).await.unwrap();
let a_end = end(&log, "a", InvocationOutcome::Completed).await;
let through = end(&log, "b", InvocationOutcome::Completed).await;
- let db = Connection::open(&path).unwrap();
- assert!(rows(&db, "a").is_empty() && rows(&db, "b").is_empty());
+ assert!(
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .is_empty()
+ && crate::test_store::inspect(&log, |db| rows(db, "b"))
+ .await
+ .is_empty()
+ );
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
assert_eq!(
- rows(&db, "a").last().unwrap().0,
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .last()
+ .unwrap()
+ .0,
RowPosition {
source_sequence: a_end,
item_index: 0
}
);
- assert!(rows(&db, "b").is_empty());
+ assert!(
+ crate::test_store::inspect(&log, |db| rows(db, "b"))
+ .await
+ .is_empty()
+ );
assert!(
crate::transcript_cache::materialize_all_rows(&log, "a", through + 1)
.await
@@ -467,8 +491,7 @@ async fn terminal_without_model_interruption_seals_only_committed_partial_eviden
.unwrap()
.is_empty()
);
- let db = Connection::open(&path).unwrap();
- let original = rows(&db, "a");
+ let original = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(original.len(), 4); // user, running, factual partial, terminal; no usage
let assistant: Value = serde_json::from_slice(&original[2].1).unwrap();
assert_eq!(assistant["id"], id);
@@ -502,14 +525,20 @@ async fn terminal_without_model_interruption_seals_only_committed_partial_eviden
}
);
// Index reset and process reopen must not drop or relabel the prefix.
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let reopened = EventLog::open(&path).await.unwrap();
crate::transcript_cache::materialize_all_rows(&reopened, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), original);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ original
+ );
assert_eq!(
reopened.active_transcript(&invocation, live).await.unwrap(),
overlay
@@ -525,11 +554,10 @@ async fn complete_projection_and_malformed_source_preserve_canonical_admission()
let expected = "x".repeat(16 * 1024 * 1024);
let (id, _) = partial(&log, "a", &expected).await;
let through = interrupt(&log, "a").await;
- let db = Connection::open(&path).unwrap();
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
- let original = rows(&db, "a");
+ let original = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert_eq!(original.len(), 3);
let assistant: Value = serde_json::from_slice(&original[2].1).unwrap();
assert_eq!(assistant["id"], id);
@@ -537,13 +565,16 @@ async fn complete_projection_and_malformed_source_preserve_canonical_admission()
assistant["text"].as_str() == Some(expected.as_str()),
"complete assistant text must survive projection"
);
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
assert!(
- rows(&db, "a") == original,
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await == original,
"cold rebuild must preserve full rows, digests and positions"
);
log.append(&event("b", opening())).await.unwrap();
@@ -566,7 +597,11 @@ async fn complete_projection_and_malformed_source_preserve_canonical_admission()
Err(StoreError::InvalidTransition(message)) if message.contains(&format!("history source {through}:"))
));
- assert!(rows(&db, "b").is_empty());
+ assert!(
+ crate::test_store::inspect(&log, |db| rows(db, "b"))
+ .await
+ .is_empty()
+ );
log.close().await.unwrap();
}
@@ -669,8 +704,7 @@ async fn completion_fanout_positions_preserve_usage_fences_sources_and_cold_rebu
crate::transcript_cache::materialize_all_rows(&log, "fanout", through)
.await
.unwrap();
- let db = Connection::open(&path).unwrap();
- let original = rows(&db, "fanout");
+ let original = crate::test_store::inspect(&log, |db| rows(db, "fanout")).await;
let completed: Vec<_> = original
.iter()
.filter(|row| row.0.source_sequence == completion)
@@ -958,15 +992,21 @@ async fn completion_fanout_positions_preserve_usage_fences_sources_and_cold_rebu
assert_eq!(search.matches.len(), 1);
assert_eq!(search.matches[0].position, result_position);
let canonical = log.prefix(1024, 8 * 1024 * 1024).await.unwrap().digest;
+ crate::test_store::inspect(&log, |db| {
+ for table in ["transcript_cache.text", "transcript_cache.rows"] {
+ db.execute(&format!("DELETE FROM {table}"), []).unwrap();
+ }
+ })
+ .await;
log.close().await.unwrap();
- for table in ["transcript_text", "transcript_rows"] {
- db.execute(&format!("DELETE FROM {table}"), []).unwrap();
- }
let reopened = EventLog::open(&path).await.unwrap();
crate::transcript_cache::materialize_all_rows(&reopened, "fanout", through)
.await
.unwrap();
- assert_eq!(rows(&db, "fanout"), original);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "fanout")).await,
+ original
+ );
assert_eq!(
reopened.prefix(1024, 8 * 1024 * 1024).await.unwrap().digest,
canonical
diff --git a/platform/event-log/tests/transcript_index/observation_count.rs b/platform/event-log/tests/transcript_index/observation_count.rs
index f93b23abbb..bd0d7ead99 100644
--- a/platform/event-log/tests/transcript_index/observation_count.rs
+++ b/platform/event-log/tests/transcript_index/observation_count.rs
@@ -132,8 +132,12 @@ async fn more_than_ten_thousand_actual_observations_remain_readable() {
crate::transcript_cache::materialize_all_rows(&log, session, through)
.await
.unwrap();
- let db = Connection::open(&path).unwrap();
- assert_eq!(rows(&db, session).len(), 4);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, session))
+ .await
+ .len(),
+ 4
+ );
log.close().await.unwrap();
}
diff --git a/platform/event-log/tests/transcript_index/performance.rs b/platform/event-log/tests/transcript_index/performance.rs
index 2532f0b0be..e0581536d0 100644
--- a/platform/event-log/tests/transcript_index/performance.rs
+++ b/platform/event-log/tests/transcript_index/performance.rs
@@ -97,9 +97,9 @@ async fn one_invocation_transcript() {
.unwrap();
let body_materialize_ms = started.elapsed().as_secs_f64() * 1000.0;
let db = Connection::open(&path).unwrap();
- let texts: Vec<String> = db
- .prepare(
- "SELECT json_extract(payload, '$.text') FROM transcript_rows
+ let texts: Vec<String> = crate::test_store::inspect(&log, move |db| {
+ db.prepare(
+ "SELECT json_extract(payload, '$.text') FROM transcript_cache.rows
WHERE session_id='a' AND json_extract(payload, '$.type')='assistant'
ORDER BY source_sequence, item_index",
)
@@ -107,7 +107,9 @@ async fn one_invocation_transcript() {
.query_map([], |row| row.get(0))
.unwrap()
.collect::<Result<_, _>>()
- .unwrap();
+ .unwrap()
+ })
+ .await;
assert_eq!(texts.len(), steps);
for (step, text) in texts.iter().enumerate() {
assert_eq!(text, &format!("one-invocation-step-{step}. ").repeat(13));
diff --git a/platform/event-log/tests/transcript_index/performance_executor.rs b/platform/event-log/tests/transcript_index/performance_executor.rs
index 745cfe6696..69df7352a1 100644
--- a/platform/event-log/tests/transcript_index/performance_executor.rs
+++ b/platform/event-log/tests/transcript_index/performance_executor.rs
@@ -136,9 +136,9 @@ async fn executor_invocation_transcript() {
.unwrap();
let body_materialize_ms = started.elapsed().as_secs_f64() * 1000.0;
let db = Connection::open(&path).unwrap();
- let rows: Vec<Value> = db.prepare("SELECT CAST(payload AS TEXT) FROM transcript_rows WHERE session_id='a' ORDER BY source_sequence, item_index").unwrap()
+ let rows: Vec<Value> = crate::test_store::inspect(&log, move |db| { db.prepare("SELECT CAST(payload AS TEXT) FROM transcript_cache.rows WHERE session_id='a' ORDER BY source_sequence, item_index").unwrap()
.query_map([], |row| row.get::<_, String>(0)).unwrap()
- .map(|row| serde_json::from_str(&row.unwrap()).unwrap()).collect();
+ .map(|row| serde_json::from_str(&row.unwrap()).unwrap()).collect() }).await;
assert_eq!(rows, expected);
println!(
diff --git a/platform/event-log/tests/transcript_index/prefix_replay.rs b/platform/event-log/tests/transcript_index/prefix_replay.rs
index 6c34249243..6e390d0695 100644
--- a/platform/event-log/tests/transcript_index/prefix_replay.rs
+++ b/platform/event-log/tests/transcript_index/prefix_replay.rs
@@ -154,22 +154,39 @@ async fn sibling_sources_match_full_canonical_replay_with_empty_cache_after_reop
let path = temp.path().join("log.sqlite");
let log = EventLog::open(&path).await.unwrap();
let (through, expected) = seed(&log).await;
- let db = Connection::open(&path).unwrap();
- assert!(rows(&db, "a").is_empty());
+ assert!(
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .is_empty()
+ );
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected
+ );
let canonical = log.prefix(32, 65536).await.unwrap().digest;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+
let reopened = EventLog::open(&path).await.unwrap();
- assert!(rows(&db, "a").is_empty());
+ assert!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a"))
+ .await
+ .is_empty()
+ );
crate::transcript_cache::materialize_all_rows(&reopened, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected);
+ assert_eq!(
+ crate::test_store::inspect(&reopened, |db| rows(db, "a")).await,
+ expected
+ );
assert_eq!(reopened.prefix(32, 65536).await.unwrap().digest, canonical);
reopened.close().await.unwrap();
}
@@ -193,14 +210,17 @@ async fn selected_raw_damage_and_cached_identity_conflict_are_not_repaired() {
crate::transcript_cache::materialize_all_rows(&log, "a", 6)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected[..5]);
- let before = rows(&db, "a");
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected[..5]
+ );
+ let before = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert!(
matches!(crate::transcript_cache::materialize_all_rows(&log,"a",through).await,
Err(StoreError::InvalidTransition(message)) if message=="missing or inconsistent durable tool payload")
);
assert_eq!(
- rows(&db, "a"),
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
before,
"failed full header batch must roll back its new rows"
);
@@ -212,17 +232,26 @@ async fn selected_raw_damage_and_cached_identity_conflict_are_not_repaired() {
crate::transcript_cache::materialize_all_rows(&log, "a", through)
.await
.unwrap();
- assert_eq!(rows(&db, "a"), expected);
- db.execute(
- "UPDATE transcript_rows SET turn_id='wrong-turn' WHERE source_sequence=9",
- [],
- )
- .unwrap();
- let damaged = rows(&db, "a");
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ expected
+ );
+ crate::test_store::inspect(&log, move |db| {
+ db.execute(
+ "UPDATE transcript_cache.rows SET turn_id='wrong-turn' WHERE source_sequence=9",
+ [],
+ )
+ .unwrap()
+ })
+ .await;
+ let damaged = crate::test_store::inspect(&log, |db| rows(db, "a")).await;
assert!(matches!(
crate::transcript_cache::materialize_all_rows(&log, "a", through).await,
Err(StoreError::TranscriptConflict)
));
- assert_eq!(rows(&db, "a"), damaged);
+ assert_eq!(
+ crate::test_store::inspect(&log, |db| rows(db, "a")).await,
+ damaged
+ );
log.close().await.unwrap();
}
diff --git a/platform/event-log/tests/transcript_index/sparse_directory.rs b/platform/event-log/tests/transcript_index/sparse_directory.rs
index b4a8a96334..bdbeb568d9 100644
--- a/platform/event-log/tests/transcript_index/sparse_directory.rs
+++ b/platform/event-log/tests/transcript_index/sparse_directory.rs
@@ -60,7 +60,11 @@ async fn cold_windows_use_complete_directory_and_preserve_all_anchor_kinds_after
}
);
let db = Connection::open(&path).unwrap();
- assert!(rows(&db, "a").is_empty());
+ assert!(
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .is_empty()
+ );
let directory = || {
db.prepare("SELECT source_sequence,item_index,message_id,role FROM transcript_directory WHERE session_id='a' ORDER BY source_sequence,item_index")
.unwrap().query_map([],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?))).unwrap().collect::<Result<Vec<_>,_>>().unwrap()
@@ -80,7 +84,9 @@ async fn cold_windows_use_complete_directory_and_preserve_all_anchor_kinds_after
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert!(
- rows(&db, "a").is_empty(),
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
+ .is_empty(),
"ordinary reopen does not rebuild optional bodies"
);
// Each anchor starts independently with complete metadata and no optional bytes.
@@ -103,8 +109,11 @@ async fn cold_windows_use_complete_directory_and_preserve_all_anchor_kinds_after
(Position::Before { cursor }, vec!["你好", "answ"]),
(Position::After { cursor }, vec!["er😀"]),
] {
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
let Page::Ready { chunks, .. } = log
.history_text("a", through, position, false, 8192, 32, 65536)
.await
@@ -134,8 +143,11 @@ async fn cold_windows_use_complete_directory_and_preserve_all_anchor_kinds_after
"reading cannot rebuild or rewrite metadata"
);
}
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
let Page::Ready {
chunks,
before,
@@ -149,7 +161,8 @@ async fn cold_windows_use_complete_directory_and_preserve_all_anchor_kinds_after
assert_eq!(chunks[0].message_id, assistant_id);
assert!(before.is_some() && after.is_none());
assert!(
- rows(&db, "a")
+ crate::test_store::inspect(&log, |db| rows(db, "a"))
+ .await
.iter()
.all(|row| row.0.source_sequence >= sealed),
"tail must not hydrate unrelated opening rows"
diff --git a/platform/event-log/tests/transcript_tools.rs b/platform/event-log/tests/transcript_tools/mod.rs
similarity index 92%
rename from platform/event-log/tests/transcript_tools.rs
rename to platform/event-log/tests/transcript_tools/mod.rs
index 8c620f5b3a..d872366ce8 100644
--- a/platform/event-log/tests/transcript_tools.rs
+++ b/platform/event-log/tests/transcript_tools/mod.rs
@@ -17,9 +17,6 @@
* under the License.
*/
-#[path = "support/transcript.rs"]
-mod transcript_cache;
-
use maka_agent_facts::event::EventWrite;
use maka_agent_facts::event::Fact;
use maka_agent_facts::event::Invocation;
@@ -40,7 +37,7 @@ use maka_presentation::tool_message_id;
use rusqlite::Connection;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
-#[path = "transcript_tools/payloads.rs"]
+#[path = "payloads.rs"]
mod payloads;
fn event(fact: Fact) -> EventWrite {
@@ -118,7 +115,7 @@ async fn accepted(log: &EventLog, step: &str, name: &str) -> u64 {
.unwrap()
}
fn rows(db: &Connection) -> Vec<(maka_agent_facts::history::RowPosition, Vec<u8>, String)> {
- db.prepare("SELECT source_sequence, item_index, payload, digest FROM transcript_rows ORDER BY source_sequence, item_index")
+ db.prepare("SELECT source_sequence, item_index, payload, digest FROM transcript_cache.rows ORDER BY source_sequence, item_index")
.unwrap()
.query_map([], |row| Ok((maka_agent_facts::history::RowPosition { source_sequence: row.get::<_, i64>(0)? as u64, item_index: row.get::<_, i64>(1)? as u64 }, row.get(2)?, row.get(3)?)))
.unwrap()
@@ -210,11 +207,10 @@ async fn tool_boundaries_page_independently_and_rebuild_exact_ids_without_readin
.await
.unwrap();
let notices = log.subscribe_commits();
- let db = Connection::open(&path).unwrap();
// A historical fence sees the planned provider call, but no later dispatch,
// nested call, result or rejection from the already-existing future suffix.
materialize_all(&log, accepted_fence).await;
- let first = rows(&db);
+ let first = crate::test_store::inspect(&log, rows).await;
assert_eq!(first.len(), 3);
let parent_id = tool_message_id("invocation", parent);
let planned: Value = serde_json::from_slice(&first[2].1).unwrap();
@@ -229,7 +225,7 @@ async fn tool_boundaries_page_independently_and_rebuild_exact_ids_without_readin
!notices.has_changed().unwrap(),
"derived cache must not publish execution commits"
);
- let original = rows(&db);
+ let original = crate::test_store::inspect(&log, rows).await;
assert_eq!(&original[..3], first.as_slice());
assert_eq!(original.len(), 17);
let values: Vec<Value> = original
@@ -331,13 +327,16 @@ async fn tool_boundaries_page_independently_and_rebuild_exact_ids_without_readin
"search emits no execution commits"
);
let canonical = log.prefix(128, 32 * 1024 * 1024).await.unwrap().digest;
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let reopened = EventLog::open(&path).await.unwrap();
materialize_all(&reopened, through).await;
assert_eq!(
- rows(&db),
+ crate::test_store::inspect(&reopened, rows).await,
original,
"replaying earlier call evidence must be idempotent"
);
@@ -345,9 +344,19 @@ async fn tool_boundaries_page_independently_and_rebuild_exact_ids_without_readin
reopened.prefix(128, 32 * 1024 * 1024).await.unwrap().digest,
canonical
);
- db.execute("DELETE FROM transcript_rows", []).unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&reopened, move |db| {
+ db.execute("DELETE FROM transcript_cache.rows", []).unwrap()
+ })
+ .await;
+ crate::test_store::inspect(&reopened, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
materialize_all(&reopened, through).await;
- assert_eq!(rows(&db), original, "the complete index is disposable");
+ assert_eq!(
+ crate::test_store::inspect(&reopened, rows).await,
+ original,
+ "the complete index is disposable"
+ );
}
diff --git a/platform/event-log/tests/transcript_tools/payloads.rs b/platform/event-log/tests/transcript_tools/payloads.rs
index 5855f0f3c6..4ef7e96103 100644
--- a/platform/event-log/tests/transcript_tools/payloads.rs
+++ b/platform/event-log/tests/transcript_tools/payloads.rs
@@ -62,14 +62,16 @@ async fn ten_mib_raw_fragments_rebuild_after_restart_from_durable_facts() {
drop(write);
materialize_all(&log, through).await;
assert!(log.prefix(100, 16 * 1024 * 1024).await.is_ok());
- let db = Connection::open(&path).unwrap();
- let original: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE message_id = ?",
- [&id],
+ let cache_id = id.clone();
+ let original: Vec<u8> = crate::test_store::inspect(&log, move |db| {
+ db.query_row(
+ "SELECT payload FROM transcript_cache.rows WHERE message_id = ?",
+ [&cache_id],
|row| row.get(0),
)
- .unwrap();
+ .unwrap()
+ })
+ .await;
let mut assembled = Vec::new();
while assembled.len() < original.len() {
let length = (original.len() - assembled.len()).min(512 * 1024);
@@ -92,9 +94,15 @@ async fn ten_mib_raw_fragments_rebuild_after_restart_from_durable_facts() {
serde_json::from_slice::<Value>(&assembled).unwrap()["content"]["value"],
raw
);
- db.execute("DELETE FROM transcript_rows", []).unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute("DELETE FROM transcript_cache.rows", []).unwrap()
+ })
+ .await;
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert_eq!(
@@ -105,13 +113,15 @@ async fn ten_mib_raw_fragments_rebuild_after_restart_from_durable_facts() {
raw
);
materialize_all(&log, through).await;
- let rebuilt: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE message_id = ?",
+ let rebuilt: Vec<u8> = crate::test_store::inspect(&log, move |db| {
+ db.query_row(
+ "SELECT payload FROM transcript_cache.rows WHERE message_id = ?",
[&id],
|row| row.get(0),
)
- .unwrap();
+ .unwrap()
+ })
+ .await;
assert_eq!(rebuilt, original);
}
@@ -140,13 +150,15 @@ async fn selected_boundary_does_not_hydrate_an_earlier_sibling_payload() {
.await
.is_err()
);
- let value: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE message_id = ?",
+ let value: Vec<u8> = crate::test_store::inspect(&log, move |db| {
+ db.query_row(
+ "SELECT payload FROM transcript_cache.rows WHERE message_id = ?",
[&target.event().id],
|row| row.get(0),
)
- .unwrap();
+ .unwrap()
+ })
+ .await;
assert_eq!(
serde_json::from_slice::<Value>(&value).unwrap()["content"]["value"]["text"],
"target"
@@ -164,21 +176,26 @@ async fn escaped_raw_and_complete_ui_row_survive_reopen_and_require_original_bod
let write = success(&operation, raw.clone());
let through = log.append(&write).await.unwrap();
materialize_all(&log, through).await;
- let db = Connection::open(&path).unwrap();
- let projected: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE message_id = ?",
- [&write.event().id],
+ let cache_id = write.event().id.clone();
+ let projected: Vec<u8> = crate::test_store::inspect(&log, move |db| {
+ db.query_row(
+ "SELECT payload FROM transcript_cache.rows WHERE message_id = ?",
+ [&cache_id],
|row| row.get(0),
)
- .unwrap();
+ .unwrap()
+ })
+ .await;
let row: Value = serde_json::from_slice(&projected).unwrap();
assert!(
row["content"]["value"] == raw,
"escaped raw must be projected completely"
);
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert!(
@@ -191,13 +208,18 @@ async fn escaped_raw_and_complete_ui_row_survive_reopen_and_require_original_bod
);
assert!(log.prefix(100, 32 * 1024 * 1024).await.is_ok());
materialize_all(&log, through).await;
- let rebuilt: Vec<u8> = db
- .query_row(
- "SELECT payload FROM transcript_rows WHERE message_id = ?",
- [&write.event().id],
- |row| row.get(0),
- )
- .unwrap();
+ let rebuilt: Vec<u8> = crate::test_store::inspect(&log, {
+ let cache_id = write.event().id.clone();
+ move |db| {
+ db.query_row(
+ "SELECT payload FROM transcript_cache.rows WHERE message_id = ?",
+ [&cache_id],
+ |row| row.get(0),
+ )
+ .unwrap()
+ }
+ })
+ .await;
assert!(
rebuilt == projected,
"reopened projection must retain exact escaped bytes"
@@ -209,8 +231,11 @@ async fn escaped_raw_and_complete_ui_row_survive_reopen_and_require_original_bod
[&write.event().id],
)
.unwrap();
- db.execute_batch("DELETE FROM transcript_text; DELETE FROM transcript_rows")
- .unwrap();
+ crate::test_store::inspect(&log, move |db| {
+ db.execute_batch("DELETE FROM transcript_cache.text; DELETE FROM transcript_cache.rows")
+ .unwrap()
+ })
+ .await;
let error = crate::transcript_cache::materialize_all_rows(&log, "session", through)
.await
.expect_err(
@@ -250,11 +275,14 @@ async fn multiple_large_results_keep_conversation_metadata_readable() {
through = log.append(&result).await.unwrap();
ids.push(result.event().id.clone());
}
- let db = Connection::open(&path).unwrap();
assert_eq!(
- db.query_row("SELECT count(*) FROM transcript_rows", [], |row| row
- .get::<_, i64>(0))
- .unwrap(),
+ crate::test_store::inspect(&log, move |db| {
+ db.query_row("SELECT count(*) FROM transcript_cache.rows", [], |row| {
+ row.get::<_, i64>(0)
+ })
+ .unwrap()
+ })
+ .await,
0
);
@@ -288,6 +316,6 @@ async fn multiple_large_results_keep_conversation_metadata_readable() {
.iter()
.all(|chunk| chunk.tool.as_ref().unwrap().result.is_some() && chunk.text.is_empty())
);
- assert_eq!(db.query_row("SELECT count(*) FROM transcript_rows body JOIN transcript_directory row USING(session_id,source_sequence,item_index) WHERE row.role IN ('tool_call','tool_result')",[],|row|row.get::<_,i64>(0)).unwrap(),0,"conversation metadata must not hydrate large tool bodies");
+ assert_eq!(crate::test_store::inspect(&log, move |db| { db.query_row("SELECT count(*) FROM transcript_cache.rows body JOIN transcript_directory row USING(session_id,source_sequence,item_index) WHERE row.role IN ('tool_call','tool_result')",[],|row|row.get::<_,i64>(0)).unwrap() }).await,0,"conversation metadata must not hydrate large tool bodies");
log.close().await.unwrap();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org