host: preserve journals, full traps and embedded agent output

Write new journal entries under logs/core and merge read-only legacy
history without moving SQLite/WAL files. Atomically spool failed batches,
retaining unspoolable entries for event-driven retries. Journal full trap
diagnostics before UTF-8-safe status display. Port session overflow archives
and followable line/byte excerpts to the still-shipped embedded agent.
Document retention/recovery and cover the contracts with regressions.
Include one rustfmt-only normalization in the existing SDK WASI test.

Card: SAU-143
This commit is contained in:
SAU-143 implementer
2026-10-26 13:40:15 +00:00
parent 168795b408
commit 037377fd27
13 changed files with 818 additions and 83 deletions
+5 -2
View File
@@ -66,8 +66,11 @@ impl Host {
Ok((World::open(&path)?, Runtime::new(&cache)?))
})
.await??;
let journal =
journal::Journal::open(world.state_dir("core").join("journal.sqlite")).await?;
let journal = journal::Journal::open_with_legacy(
world.root.join("logs/core/journal.sqlite"),
Some(world.state_dir("core").join("journal.sqlite")),
)
.await?;
let host = Self::with_parts(world, runtime, journal);
let cargo_world = host.0.world.clone();
let cargo = tokio::task::spawn_blocking(move || {
+274 -31
View File
@@ -1,13 +1,17 @@
//! A dedicated SQLite writer batches entries and serves ordered queries without propagating disk failures into the host.
use rusqlite::{Connection, params};
use rusqlite::{Connection, OpenFlags, params};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::io::Write;
use std::{
collections::HashSet,
path::PathBuf,
sync::{Mutex, mpsc},
};
use tokio::sync::oneshot;
#[derive(Debug, Serialize, Deserialize)]
struct Entry {
time: String,
id: String,
pid: String,
plugin: String,
@@ -21,44 +25,83 @@ enum Command {
pub(crate) struct Journal {
sender: mpsc::Sender<Command>,
warned: Mutex<HashSet<String>>,
spool: PathBuf,
}
impl Journal {
pub(crate) async fn memory() -> anyhow::Result<Self> {
Self::open(PathBuf::from(":memory:")).await
}
pub(crate) async fn open(path: PathBuf) -> anyhow::Result<Self> {
Self::open_with_legacy(path, None).await
}
pub(crate) async fn open_with_legacy(
path: PathBuf,
legacy: Option<PathBuf>,
) -> anyhow::Result<Self> {
let spool = path
.parent()
.unwrap_or(std::path::Path::new("."))
.join("failed-batches");
let worker_spool = spool.clone();
let (sender, receiver) = mpsc::channel();
let (ready, initialized) = oneshot::channel();
std::thread::spawn(move || match open_database(path) {
Ok(mut db) => {
let _ = ready.send(Ok(()));
while let Ok(first) = receiver.recv() {
let mut batch = vec![first];
batch.extend(receiver.try_iter().take(255));
write_batch(&mut db, batch);
std::thread::spawn(move || {
match (|| -> anyhow::Result<_> {
let db = open_database(path)?;
// Never rename, checkpoint, or write the old SQLite/WAL files.
let old = legacy
.filter(|p| p.exists())
.map(|p| Connection::open_with_flags(p, OpenFlags::SQLITE_OPEN_READ_ONLY))
.transpose()?;
Ok((db, old))
})() {
Ok((mut db, old)) => {
let _ = ready.send(Ok(()));
let mut pending = Vec::new();
while let Ok(first) = receiver.recv() {
let mut batch = vec![first];
batch.extend(receiver.try_iter().take(255));
pending = write_batch(&mut db, old.as_ref(), &worker_spool, pending, batch);
}
if !pending.is_empty() {
// Both storage locations failed. Last resort: retain full text in host stderr.
eprintln!(
"journal shutdown with unpersisted entries: {}",
serde_json::to_string(&pending).unwrap()
);
}
}
Err(error) => {
eprintln!("journal open failed: {error}");
let _ = ready.send(Err(error));
}
}
Err(error) => {
eprintln!("journal open failed: {error}");
let _ = ready.send(Err(error));
}
});
initialized.await??;
Ok(Self {
sender,
warned: Mutex::new(HashSet::new()),
spool,
})
}
pub(crate) fn write(&self, pid: &str, plugin: &str, stream: &str, line: &str) {
let entry = Entry {
time: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
id: uuid::Uuid::now_v7().to_string(),
pid: pid.into(),
plugin: plugin.into(),
stream: stream.into(),
line: line.into(),
};
if self.sender.send(Command::Write(entry)).is_err() {
eprintln!("journal writer stopped");
if let Err(mpsc::SendError(Command::Write(entry))) = self.sender.send(Command::Write(entry))
{
eprintln!("journal writer stopped; saving entry outside SQLite");
if let Err(error) = spool_batch(&self.spool, std::slice::from_ref(&entry)) {
eprintln!(
"journal spool failed: {error}; unpersisted entry: {}",
serde_json::to_string(&entry).unwrap()
);
}
}
}
pub(crate) fn warning(&self, pid: &str, plugin: &str) {
@@ -94,28 +137,94 @@ fn open_database(path: PathBuf) -> anyhow::Result<Connection> {
CREATE INDEX IF NOT EXISTS journal_plugin ON journal(plugin);")?;
Ok(db)
}
fn write_batch(db: &mut Connection, batch: Vec<Command>) {
let write = (|| -> rusqlite::Result<()> {
let transaction = db.transaction()?;
for command in &batch {
if let Command::Write(e) = command {
transaction.execute("INSERT INTO journal (id,time,pid,plugin,stream,line) VALUES (?1,strftime('%Y-%m-%dT%H:%M:%fZ','now'),?2,?3,?4,?5)",params![e.id,e.pid,e.plugin,e.stream,e.line])?;
/// Exclusive, synced JSON files preserve failed transactions without a busy retry loop.
/// Recovery is deliberate: files are never automatically removed or replayed.
fn spool_batch(dir: &std::path::Path, entries: &[Entry]) -> anyhow::Result<PathBuf> {
std::fs::create_dir_all(dir)?;
let path = dir.join(format!("{}.json", uuid::Uuid::now_v7()));
// A failed/partial write must not look like a complete recovery batch.
let mut file = tempfile::NamedTempFile::new_in(dir)?;
serde_json::to_writer(&mut file, entries)?;
file.write_all(b"\n")?;
file.as_file().sync_all()?;
file.persist_noclobber(&path)?;
std::fs::File::open(dir)?.sync_all()?;
eprintln!(
"journal failed batch preserved: {} ({} entries)",
path.display(),
entries.len()
);
Ok(path)
}
fn write_batch(
db: &mut Connection,
old: Option<&Connection>,
spool: &std::path::Path,
mut pending: Vec<Entry>,
batch: Vec<Command>,
) -> Vec<Entry> {
let mut queries = Vec::new();
for command in batch {
match command {
Command::Write(entry) => pending.push(entry),
Command::Query(filters, response) => queries.push((filters, response)),
}
}
if !pending.is_empty() {
let write = (|| -> rusqlite::Result<()> {
let transaction = db.transaction()?;
for e in &pending {
transaction.execute("INSERT OR IGNORE INTO journal (id,time,pid,plugin,stream,line) VALUES (?1,?2,?3,?4,?5,?6)",params![e.id,e.time,e.pid,e.plugin,e.stream,e.line])?;
}
transaction.commit()
})();
match write {
Ok(()) => pending.clear(),
Err(error) => {
eprintln!("journal write failed: {error}");
match spool_batch(spool, &pending) {
Ok(_) => pending.clear(),
Err(error) => eprintln!(
"journal spool failed: {error}; retaining {} entries in memory until next command",
pending.len()
),
}
}
}
transaction.commit()
})();
if let Err(error) = write {
eprintln!("journal write failed: {error}");
}
for command in batch {
if let Command::Query(filters, response) = command {
let value = query_database(db, &filters).unwrap_or_else(|error| {
eprintln!("journal query failed: {error}");
json!([])
});
let _ = response.send(value);
for (filters, response) in queries {
let value = query_all(db, old, &filters).unwrap_or_else(|error| {
eprintln!("journal query failed: {error}");
json!([])
});
let _ = response.send(value);
}
pending
}
fn query_all(
db: &Connection,
old: Option<&Connection>,
filters: &Value,
) -> rusqlite::Result<Value> {
let mut entries = query_database(db, filters)?.as_array().unwrap().clone();
if let Some(old) = old {
entries.extend(
query_database(old, filters)?
.as_array()
.unwrap()
.iter()
.cloned(),
);
entries.sort_by(|a, b| a["id"].as_str().cmp(&b["id"].as_str()));
entries.dedup_by(|a, b| a["id"] == b["id"]);
let limit = filters["limit"].as_i64().unwrap_or(100);
if limit >= 0 {
entries.drain(..entries.len().saturating_sub(limit as usize));
}
}
Ok(json!(entries))
}
fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result<Value> {
let mut statement = db.prepare(
@@ -135,6 +244,140 @@ fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result<Value> {
#[cfg(test)]
mod tests {
use super::*;
fn entry(line: &str) -> Entry {
Entry {
id: uuid::Uuid::now_v7().to_string(),
time: "2026-10-26T12:00:00.000Z".into(),
pid: "pid".into(),
plugin: "test".into(),
stream: "core".into(),
line: line.into(),
}
}
#[test]
fn failed_transaction_is_spooled_in_full_and_queries_finish() {
let dir = tempfile::tempdir().unwrap();
let mut db = open_database(PathBuf::from(":memory:")).unwrap();
db.execute_batch("PRAGMA query_only=ON").unwrap();
let original = "😀 full failure\n".repeat(6000);
let (tx, mut rx) = oneshot::channel();
let pending = write_batch(
&mut db,
None,
dir.path(),
vec![],
vec![
Command::Write(entry(&original)),
Command::Query(json!({}), tx),
],
);
assert!(pending.is_empty());
assert_eq!(rx.try_recv().unwrap(), json!([]));
let path = std::fs::read_dir(dir.path())
.unwrap()
.next()
.unwrap()
.unwrap()
.path();
let saved: Vec<Entry> = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
assert_eq!(saved.len(), 1);
assert_eq!(saved[0].line, original);
assert_eq!(saved[0].time, "2026-10-26T12:00:00.000Z");
}
#[test]
fn failed_spool_retries_on_next_command_without_duplicates() {
let dir = tempfile::tempdir().unwrap();
let blocked = dir.path().join("blocked");
std::fs::write(&blocked, "not a directory").unwrap();
let mut db = open_database(PathBuf::from(":memory:")).unwrap();
db.execute_batch("PRAGMA query_only=ON").unwrap();
let pending = write_batch(
&mut db,
None,
&blocked,
vec![],
vec![Command::Write(entry("retained"))],
);
assert_eq!(pending.len(), 1);
db.execute_batch("PRAGMA query_only=OFF").unwrap();
let (tx, mut rx) = oneshot::channel();
let pending = write_batch(
&mut db,
None,
&blocked,
pending,
vec![Command::Query(json!({}), tx)],
);
assert!(pending.is_empty());
assert_eq!(rx.try_recv().unwrap()[0]["line"], "retained");
assert_eq!(
query_all(&db, None, &json!({}))
.unwrap()
.as_array()
.unwrap()
.len(),
1
);
}
#[tokio::test]
async fn legacy_wal_and_new_journal_share_filters_order_and_limit() {
let dir = tempfile::tempdir().unwrap();
let legacy = dir.path().join("state/core/journal.sqlite");
let mut old = open_database(legacy.clone()).unwrap();
write_batch(
&mut old,
None,
dir.path(),
vec![],
vec![Command::Write(entry("old match"))],
);
// Leave the WAL connection open: opening the new journal must not move its files.
let wal = legacy.with_extension("sqlite-wal");
let before = std::fs::read(&wal).unwrap();
let new = dir.path().join("logs/core/journal.sqlite");
let journal = Journal::open_with_legacy(new.clone(), Some(legacy.clone()))
.await
.unwrap();
journal.write("pid", "test", "core", "new match");
let entries = tokio::time::timeout(
std::time::Duration::from_secs(5),
journal.query(json!({"pid":"pid","grep":"match","limit":-1})),
)
.await
.unwrap();
assert_eq!(entries.as_array().unwrap().len(), 2);
assert_eq!(entries[0]["line"], "old match");
assert_eq!(entries[1]["line"], "new match");
assert_eq!(
journal.query(json!({"limit":1})).await[0]["line"],
"new match"
);
assert_eq!(
journal.query(json!({"until":entries[0]["id"]})).await[0]["line"],
"old match"
);
assert_eq!(
journal
.query(json!({"since":entries[1]["id"],"plugin":"test"}))
.await[0]["line"],
"new match"
);
assert_eq!(journal.query(json!({"grep":"absent"})).await, json!([]));
assert_eq!(std::fs::read(wal).unwrap(), before);
assert!(legacy.exists() && new.exists());
assert_eq!(
query_database(&old, &json!({}))
.unwrap()
.as_array()
.unwrap()
.len(),
1
);
}
/// PROTOCOL 7: completed processes do not leave warning identities in memory.
#[tokio::test]
async fn completion_forgets_warning_identity() {
+23 -1
View File
@@ -212,7 +212,11 @@ fn short_trap(full: &str) -> String {
.find_map(|l| l.find("wasm trap:").map(|i| l[i..].to_string()))
.unwrap_or_else(|| full.lines().next().unwrap_or("trap").to_string());
let mut line = line.trim().to_string();
line.truncate(300);
let mut end = line.len().min(300);
while !line.is_char_boundary(end) {
end -= 1;
}
line.truncate(end);
line
}
@@ -361,6 +365,7 @@ impl Runtime {
});
let (err_tx, err_rx) = mpsc::channel(1024);
let trap_log = log.clone();
let stderr_task = tokio::spawn(async move {
let mut reader = BufReader::new(host_err);
let mut bytes = Vec::new();
@@ -400,6 +405,9 @@ impl Runtime {
Some(I32Exit(c)) => Exit::Code(*c),
None => {
let full = format!("{e:?}");
if let Some((journal, pid, plugin)) = &trap_log {
journal.write(pid, plugin, "core", &full);
}
Exit::Trap(short_trap(&full))
}
},
@@ -416,3 +424,17 @@ impl Runtime {
})
}
}
#[cfg(test)]
mod trap_text_tests {
use super::*;
#[test]
fn short_trap_display_is_utf8_safe() {
let full = format!("backtrace\nwasm trap: {}\nmore detail", "😀".repeat(1000));
let status = short_trap(&full);
assert!(status.len() <= 300);
assert!(status.starts_with("wasm trap:"));
assert_eq!(short_trap(""), "trap");
assert_eq!(short_trap("first line\nsecond"), "first line");
}
}
+1
View File
@@ -92,6 +92,7 @@ mod tests {
"/config",
"/state",
"/state/core/journal.sqlite",
"/logs/core/journal.sqlite",
] {
assert!(!std::path::Path::new(path).exists(), "{path}");
}
+22 -1
View File
@@ -489,7 +489,28 @@ async fn oversized_current_turn_preserves_the_previous_cursor() {
.unwrap();
let (body, reply) = server.next().await;
let serialized = input(&body, api).to_string();
assert!(serialized.contains("[tool output truncated"));
assert!(serialized.contains("Full output: /logs/agent/"));
let history = world.history();
let result: Value = history
.lines()
.map(|line| serde_json::from_str::<Value>(line).unwrap())
.find(|entry| {
entry["output"]
.as_str()
.is_some_and(|out| out.contains("Full output: "))
})
.unwrap();
let output = result["output"].as_str().unwrap();
let path = output
.split("Full output: ")
.nth(1)
.unwrap()
.split(". ")
.next()
.unwrap();
let archive = std::fs::read_to_string(world.0.join(path.trim_start_matches('/'))).unwrap();
assert!(archive.contains(&"x".repeat(200_000)));
assert!(output.len() <= 50 * 1024);
assert!(serialized.len() < 70_000);
reply
.send((413, json!({"type":"request_too_large"})))
+5 -5
View File
@@ -36,7 +36,7 @@ async fn status(child: &mut Child) -> std::process::ExitStatus {
}
async fn journal_exists(world: &Path) {
tokio::time::timeout(Duration::from_secs(15), async {
while !world.join("state/core/journal.sqlite").exists() {
while !world.join("logs/core/journal.sqlite").exists() {
tokio::task::yield_now().await;
}
})
@@ -251,7 +251,7 @@ async fn signals_work_during_boot() {
tokio::time::timeout(Duration::from_secs(15), async {
loop {
let db =
rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap();
rusqlite::Connection::open(world.join("logs/core/journal.sqlite")).unwrap();
if let Ok(2) = db.query_row(
"SELECT count(*) FROM journal WHERE line='service start core.lock'",
[],
@@ -336,7 +336,7 @@ async fn console_error_stops_host() {
.await
.unwrap();
assert_eq!(status(&mut child).await.code(), Some(1));
let db = rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap();
let db = rusqlite::Connection::open(world.join("logs/core/journal.sqlite")).unwrap();
let stopped: i64 = db
.query_row(
"SELECT count(*) FROM journal WHERE plugin='probe' AND line='stop'",
@@ -363,9 +363,9 @@ async fn interrupt_leaves_call_running() {
input.write_all(b"probe.busy\n").await.unwrap();
let pid = tokio::time::timeout(Duration::from_secs(20), async {
loop {
if world.join("state/core/journal.sqlite").exists() {
if world.join("logs/core/journal.sqlite").exists() {
let db =
rusqlite::Connection::open(world.join("state/core/journal.sqlite")).unwrap();
rusqlite::Connection::open(world.join("logs/core/journal.sqlite")).unwrap();
if let Ok(pid) = db.query_row(
"SELECT pid FROM journal WHERE plugin='probe' AND line='start busy {}'",
[],
+81 -5
View File
@@ -129,7 +129,7 @@ async fn stderr_flood_does_not_block_stdin() {
let payload = json!({"large":"x".repeat(1024*1024)});
assert_eq!(call(&h, "probe.echo", payload.clone()).await, payload);
h.stop().await;
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
let db = rusqlite::Connection::open(w.0.join("logs/core/journal.sqlite")).unwrap();
let count: i64 = db
.query_row(
"SELECT count(*) FROM journal WHERE line LIKE 'flood %'",
@@ -198,7 +198,7 @@ async fn null_error_is_success() {
async fn journal_recovers_after_sqlite_write_failure() {
let w = TestWorld::new();
let h = w.start().await;
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
let db = rusqlite::Connection::open(w.0.join("logs/core/journal.sqlite")).unwrap();
db.busy_timeout(std::time::Duration::from_secs(5)).unwrap();
db.execute_batch("CREATE TRIGGER reject_writes BEFORE INSERT ON journal BEGIN SELECT RAISE(FAIL,'test failure'); END;").unwrap();
assert_eq!(
@@ -206,6 +206,16 @@ async fn journal_recovers_after_sqlite_write_failure() {
json!({"failed_log":true})
);
call(&h, "core.log", json!({})).await;
let batches: Vec<Value> = std::fs::read_dir(w.0.join("logs/core/failed-batches"))
.unwrap()
.map(|file| serde_json::from_slice(&std::fs::read(file.unwrap().path()).unwrap()).unwrap())
.collect();
assert!(
batches
.iter()
.flat_map(|batch| batch.as_array().unwrap())
.any(|entry| entry["line"].as_str().unwrap().contains("failed_log"))
);
db.execute_batch("DROP TRIGGER reject_writes;").unwrap();
assert_eq!(
call(&h, "probe.echo", json!({"recovered":true})).await,
@@ -1068,7 +1078,7 @@ async fn journal_filters_and_entries_persist_in_sqlite() {
);
h.stop().await;
assert!(
std::fs::read(w.0.join("state/core/journal.sqlite"))
std::fs::read(w.0.join("logs/core/journal.sqlite"))
.unwrap()
.starts_with(b"SQLite format 3")
);
@@ -1254,8 +1264,9 @@ async fn journal_entries_have_uuid_time_identity_stream_and_line() {
async fn journal_has_wal_and_identity_indexes() {
let w = TestWorld::new();
let h = w.start().await;
assert!(!w.0.join("state/core/journal.sqlite").exists());
h.stop().await;
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
let db = rusqlite::Connection::open(w.0.join("logs/core/journal.sqlite")).unwrap();
assert_eq!(
db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
.unwrap(),
@@ -1941,7 +1952,7 @@ async fn core_restart_cleanly_stops_detached_caller() {
std::fs::read_to_string(w.0.join("state/restart-stop")).unwrap(),
"persisted after EOF"
);
let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
let db = rusqlite::Connection::open(w.0.join("logs/core/journal.sqlite")).unwrap();
let count: i64 = db
.query_row(
"SELECT count(*) FROM journal WHERE line = 'restart stop hook finished'",
@@ -1995,3 +2006,68 @@ async fn core_exit_alias_sets_executable_status_75() {
String::from_utf8_lossy(&output.stderr)
);
}
/// PROTOCOL 7: logs move, state and the old journal do not; core.log merges both.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn journal_location_preserves_legacy_history() {
let w = TestWorld::new();
let old = w.0.join("state/core/journal.sqlite");
let db = rusqlite::Connection::open(&old).unwrap();
db.execute_batch("CREATE TABLE journal (id TEXT PRIMARY KEY, time TEXT, pid TEXT, plugin TEXT, stream TEXT, line TEXT);
INSERT INTO journal VALUES ('0000','old time','old-pid','old-plugin','core','legacy history');").unwrap();
drop(db);
let before = std::fs::read(&old).unwrap();
let history = w.0.join("state/core/console-history");
std::fs::write(&history, "keep console history").unwrap();
let h = w.start().await;
call(&h, "probe.echo", json!({})).await;
assert!(w.0.join("logs/core/journal.sqlite").exists());
let entries = call(&h, "core.log", json!({"limit":-1})).await;
assert_eq!(entries[0]["line"], "legacy history");
assert!(entries.as_array().unwrap().len() > 1);
assert_eq!(
call(&h, "core.log", json!({"pid":"old-pid"})).await[0]["line"],
"legacy history"
);
h.stop().await;
assert_eq!(std::fs::read(&old).unwrap(), before);
assert_eq!(
std::fs::read_to_string(history).unwrap(),
"keep console history"
);
}
/// PROTOCOL 7: the journal retains the full trap, not just its short status reason.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn full_trap_diagnostic_is_journaled() {
let w = TestWorld::new();
let h = w.start().await;
let error = tokio::time::timeout(
std::time::Duration::from_secs(20),
h.call("probe.trap", json!({})),
)
.await
.unwrap()
.unwrap_err();
assert_eq!(error["code"], -32002);
assert!(
error["data"]["status"]
.as_str()
.unwrap()
.starts_with("trap: wasm trap:")
);
let entries = call(
&h,
"core.log",
json!({"plugin":"probe","grep":"wasm backtrace"}),
)
.await;
assert!(
entries.as_array().unwrap().iter().any(|e| {
let text = e["line"].as_str().unwrap();
text.contains("Caused by:") && text.contains("wasm trap:")
}),
"{entries}"
);
h.stop().await;
}
+1
View File
@@ -376,6 +376,7 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
let params = &request["params"];
let result = match request["method"].as_str().unwrap() {
"batch" => batch(lines),
"trap" => std::arch::wasm32::unreachable(),
"nested_environment" => nested_environment(lines),
"forget" => forget(lines),
"forgotten" => {