From 037377fd2760090f73718bd15ab6ecd5744d55e5 Mon Sep 17 00:00:00 2001 From: SAU-143 implementer Date: Mon, 26 Oct 2026 13:40:15 +0000 Subject: [PATCH] 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 --- PROTOCOL.md | 35 ++- agent/src/main.rs | 37 ++- agent/src/overflow.rs | 258 +++++++++++++++++++ agent/src/tools.rs | 108 ++++++-- cli/src/host.rs | 7 +- cli/src/journal.rs | 305 ++++++++++++++++++++--- cli/src/runtime.rs | 24 +- cli/tests/crates/cargo-valid/src/main.rs | 1 + cli/tests/host/agent.rs | 23 +- cli/tests/host/cli.rs | 10 +- cli/tests/host/core.rs | 86 ++++++- cli/tests/plugins/probe/src/main.rs | 1 + sdk/tests/wasi.rs | 6 +- 13 files changed, 818 insertions(+), 83 deletions(-) create mode 100644 agent/src/overflow.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index cc2f56f..3385cfc 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -34,7 +34,8 @@ Rust plugins use the SDK `sausage-plugin`. Its source is at source/ the source. `cargo.build` builds it from here. imports/.json descriptions of the imported plugins state// persistent state of a plugin, by convention - state/core/journal.sqlite the journal + logs/core/journal.sqlite the journal + state/core/ console history (state, not logs) config/ global configuration boot.yaml the services agent.yaml the API, base URL, model, and key of the agent @@ -279,9 +280,23 @@ for example the control interface of a web server. ## 7. The journal -- An SQLite file at `/state/core/journal.sqlite`. Only the core writes it. +- New entries go to `/logs/core/journal.sqlite`. Only the core writes it. + An existing `/state/core/journal.sqlite` stays in place and is opened + read-only; queries merge both databases by entry ID before applying the + limit. The host never renames a live SQLite/WAL trio. Console history stays + in `/state/core`. No journal or log files are automatically pruned. +- If a journal transaction fails, the complete batch (including original + IDs and timestamps) is saved in an exclusive, synced JSON file under + `/logs/core/failed-batches/`; the host reports its path on stderr. These + files are recovery artifacts, not query results: recovery/import is + deliberate, using IDs to avoid duplicates, and never deletes the source. + If spooling also fails, entries stay in memory and retry on the next + command (no polling); at writer shutdown their full text is reported to + host stderr. Storage failure plus host loss can still lose unsaved data. - Each entry has a UUIDv7 (it sorts by time), the process ID, the plugin, the stream (`stderr` or `core`), and the line. +- Traps retain their full diagnostic text in the journal; short exit-status + displays are UTF-8-boundary-safe. - The core writes its own events too: start (with method and params), exit status, stop, kill, upgrades, and the warning about discarded stdout. - `core.log {plugin?, pid?, since?, until?, grep?, limit?}` queries the @@ -473,3 +488,19 @@ Tests: files through `env!("CARGO_MANIFEST_DIR")`. - The test runner answers `core.lock`, `core.unlock`, `core.notify`, and `core.log`. A call to a plugin fails: a test has no other plugins. + +## 12. Embedded agent tool output + +The agent shipped by `sausage init` uses a context-size guard of 50 KiB or +2000 lines, not a data-retention limit. Oversized tool results, including +errors, are saved in full under `/logs/agent//default/` before history +and provider delivery receive an excerpt. Overflow JSON string values are +rendered as physical text lines so the reference's one-based inclusive line +ranges can be followed with `read_file {path, line_offset, line_limit}` +(zero-based offset; default 200 lines). Huge single lines give exact byte +ranges and continuation offsets. Command/log/build/test results show the +tail; other tools show the head. Reading an existing `/logs` file excerpts +that same file without making another archive. Archives are exclusive and +never automatically deleted. If archival fails, the agent reports the error +to stderr and keeps the complete result instead of discarding it. Existing +history formats and paths are unchanged. diff --git a/agent/src/main.rs b/agent/src/main.rs index 83032fc..6f0df2e 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -94,19 +94,38 @@ impl Agent { return Ok(final_text); } for (id, name, arguments) in calls { - let output = match serde_json::from_str(&arguments) { - Ok(args) => tools::run(&name, &args) + let args = serde_json::from_str::(&arguments); + let output = match &args { + Ok(args) => tools::run(&name, args) .await .unwrap_or_else(|error| json!({"error":error})), Err(error) => json!({"error":format!("arguments are not valid JSON: {error}")}), }; - append( - &mut entries, - Entry::ToolResult { - id, - output: tools::limit_output(output.to_string()), - }, - )?; + let output = if name == "read_file" + && args + .as_ref() + .ok() + .and_then(|a| a["path"].as_str()) + .is_some_and(|p| p.starts_with("/logs/")) + { + output["content"] + .as_str() + .map(str::to_owned) + .unwrap_or_else(|| output.to_string()) + } else { + output.to_string() + }; + let tail = args + .as_ref() + .ok() + .and_then(|a| a["method"].as_str()) + .is_some_and(|m| { + [".exec", ".log", ".build", ".test"] + .iter() + .any(|suffix| m.ends_with(suffix)) + }); + let output = tools::overflow::guard(output, "default", &id, tail); + append(&mut entries, Entry::ToolResult { id, output })?; } } } diff --git a/agent/src/overflow.rs b/agent/src/overflow.rs new file mode 100644 index 0000000..9e29c97 --- /dev/null +++ b/agent/src/overflow.rs @@ -0,0 +1,258 @@ +//! Context-size guard; full output is in the referenced file. Never a data limit. +use std::io::{self, Write}; +use std::sync::atomic::{AtomicU64, Ordering}; + +// One policy point: Pi's 50 KiB / 2000 lines, including room for the reference. +const CONTEXT_LIMIT: (usize, usize) = (50 * 1024, 2000); +static NEXT: AtomicU64 = AtomicU64::new(0); + +fn component(s: &str) -> String { + s.bytes() + .take(80) + .map(|b| { + if b.is_ascii_alphanumeric() || b == b'-' || b == b'_' { + b as char + } else { + '_' + } + }) + .collect() +} + +pub fn exceeds_limit(output: &str) -> bool { + output.len() > CONTEXT_LIMIT.0 || output.lines().count() > CONTEXT_LIMIT.1 +} + +/// Save before excerpting. On storage failure return all data, not a false reference. +pub fn guard(output: String, session: &str, id: &str, tail: bool) -> String { + if !exceeds_limit(&output) { + return output; + } + // Decode only results that crossed the context guard. Archive physical + // string lines, not JSON escapes, even if decoding makes the text smaller. + let output = serde_json::from_str::(&output) + .map(|value| super::render(&value)) + .unwrap_or(output); + let pid = std::env::var("SAUSAGE_PID").unwrap_or_else(|_| "test".into()); + let dir = format!("/logs/agent/{}/{}", date(), component(session)); + let saved = (|| -> io::Result { + std::fs::create_dir_all(&dir)?; + // create_new prevents overwriting even if an ID is reused or a process restarts. + loop { + let path = format!( + "{dir}/{}-{}-{}.txt", + component(id), + component(&pid), + NEXT.fetch_add(1, Ordering::Relaxed) + ); + match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + { + Ok(mut f) => { + f.write_all(output.as_bytes())?; + return Ok(path); + } + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + } + } + })(); + match saved { + Ok(path) => excerpt(&output, &path, tail, 0, output.len()), + Err(e) => { + eprintln!("overflow archive failed: {e}; retaining complete tool result"); + output + } + } +} + +fn date() -> String { + let days = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + / 86400; + let z = days as i64 + 719468; + let era = z / 146097; + let doe = z - era * 146097; + let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = mp + if mp < 10 { 3 } else { -9 }; + format!("{:04}-{m:02}-{d:02}", y + i64::from(m <= 2)) +} + +/// Excerpt an existing archive in place. Byte/line ranges refer to that file, +/// including when a caller requests an oversized line or the entire archive. +pub fn existing(bytes: &[u8], path: &str, args: &serde_json::Value) -> String { + let text = String::from_utf8_lossy(bytes); + let (mut from, mut to) = if let Some(line) = args["line_offset"].as_u64() { + let from = text + .split_inclusive('\n') + .take(line as usize) + .map(str::len) + .sum::(); + let len = text[from..] + .split_inclusive('\n') + .take(args["line_limit"].as_u64().unwrap_or(200) as usize) + .map(str::len) + .sum::(); + (from, from + len) + } else { + let from = (args["offset"].as_u64().unwrap_or(0) as usize).min(text.len()); + ( + from, + args["max_bytes"].as_u64().map_or(text.len(), |n| { + from.saturating_add(n as usize).min(text.len()) + }), + ) + }; + while !text.is_char_boundary(from) { + from += 1; + } + while to > from && !text.is_char_boundary(to) { + to -= 1; + } + to = to.max(from); + excerpt(&text, path, false, from, to) +} + +fn excerpt(output: &str, path: &str, tail: bool, from: usize, to: usize) -> String { + let selected = &output[from..to]; + let lines: Vec<&str> = selected.split_inclusive('\n').collect(); + let budget = CONTEXT_LIMIT.0.saturating_sub(1024); + let mut start = 0; + let mut end = lines.len(); + let mut bytes = 0; + if tail { + start = end; + while start > 0 + && end - start < CONTEXT_LIMIT.1.saturating_sub(2) + && bytes + lines[start - 1].len() <= budget + { + start -= 1; + bytes += lines[start].len(); + } + } else { + end = 0; + while end < lines.len() + && end < CONTEXT_LIMIT.1.saturating_sub(2) + && bytes + lines[end].len() <= budget + { + bytes += lines[end].len(); + end += 1; + } + } + let mut shown = lines[start..end].concat(); + // A single huge line: preserve valid UTF-8 and explicitly label a partial line. + let partial = start == end; + if partial && !lines.is_empty() { + if tail { + start = lines.len() - 1; + let line = lines[start]; + let mut from = line.len().saturating_sub(budget); + while !line.is_char_boundary(from) { + from += 1; + } + shown.push_str(&line[from..]); + } else { + let mut to = budget.min(lines[0].len()); + while !lines[0].is_char_boundary(to) { + to -= 1; + } + shown.push_str(&lines[0][..to]); + } + } + let mut byte_start = from + lines[..start].iter().map(|s| s.len()).sum::(); + if partial && tail { + byte_start = to - shown.len(); + } + let byte_end = byte_start + shown.len(); + let line_start = output[..byte_start].bytes().filter(|b| *b == b'\n').count(); + let line_end = line_start + shown.split_inclusive('\n').count(); + let total_lines = output.split_inclusive('\n').count(); + let partial = partial + || (from > 0 && !output[..from].ends_with('\n')) + || (to < output.len() && !output[..to].ends_with('\n')); + let read = if partial { + format!( + "Shown byte range [{byte_start},{byte_end}); continue with read_file path={path}, offset={byte_end}, max_bytes={budget} (or choose an earlier byte offset)" + ) + } else { + format!( + "Read with read_file path={path}, line_offset={line_start}, line_limit={}; next line_offset={line_end}", + line_end - line_start + ) + }; + format!( + "{shown}\n[Showing lines {}-{} of {}{}; total {} bytes. Full output: {path}. {read}]", + line_start + 1, + line_end, + total_lines, + if partial { " (partial line)" } else { "" }, + output.len() + ) +} + +/// Zero-based line offsets; byte-based reads remain unchanged when absent. +pub fn read_lines(bytes: &[u8], offset: usize, limit: usize) -> String { + String::from_utf8_lossy(bytes) + .split_inclusive('\n') + .skip(offset) + .take(limit) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn full_archive_and_context_guard() { + let original = "😀 line\n".repeat(5400); + for tail in [false, true] { + let shown = guard(original.clone(), "guard-test", "call", tail); + assert!(shown.len() <= CONTEXT_LIMIT.0); + assert!(shown.lines().count() <= CONTEXT_LIMIT.1); + assert!(shown.contains(if tail { + "Showing lines 3403-5400" + } else { + "Showing lines 1-1998" + })); + let path = shown + .split("Full output: ") + .nth(1) + .unwrap() + .split(". Read with") + .next() + .unwrap(); + assert_eq!(std::fs::read_to_string(path).unwrap(), original); + } + assert_eq!(guard("small".into(), "test", "id", false), "small"); + } + #[test] + fn archive_failure_does_not_discard_data() { + std::fs::create_dir_all(format!("/logs/agent/{}", date())).unwrap(); + std::fs::write(format!("/logs/agent/{}/not-a-directory", date()), "block").unwrap(); + let output = "no loss\n".repeat(6000); + assert_eq!( + guard(output.clone(), "not-a-directory", "test", false), + output + ); + } + #[test] + fn huge_line_and_exact_boundaries() { + for tail in [false, true] { + let result = guard("😀".repeat(40000), "test", "utf8", tail); + assert!(result.len() <= CONTEXT_LIMIT.0); + assert!(result.contains("partial line")); + } + let exact = "a\n".repeat(CONTEXT_LIMIT.1); + assert_eq!(guard(exact.clone(), "test", "exact", false), exact); + assert_eq!(read_lines(b"one\ntwo\nthree", 1, 1), "two\n"); + assert_eq!(read_lines(b"one", 3, 1), ""); + } +} diff --git a/agent/src/tools.rs b/agent/src/tools.rs index fb8b09c..3190202 100644 --- a/agent/src/tools.rs +++ b/agent/src/tools.rs @@ -4,21 +4,47 @@ use crate::llm::Tool; use serde_json::{Value, json}; use std::io; -/// Maximum UTF-8 bytes in a serialized tool result, including its truncation marker. -pub const MAX_OUTPUT_BYTES: usize = 65_536; +#[path = "overflow.rs"] +pub mod overflow; -/// Bound every tool result before it enters history or a provider request. -pub fn limit_output(mut output: String) -> String { - const MARKER: &str = "\n[tool output truncated at 65536 bytes]"; - if output.len() > MAX_OUTPUT_BYTES { - let mut end = MAX_OUTPUT_BYTES - MARKER.len(); - while !output.is_char_boundary(end) { - end -= 1; +/// Human-readable, lossless text representation of tool JSON. String values +/// retain real newlines, with JSON-quoted field/index labels for structure. +fn render(value: &Value) -> String { + fn visit(value: &Value, label: &str, out: &mut String) { + match value { + Value::Object(fields) => { + if fields.is_empty() { + out.push_str(&format!("{label}: {{}}\n")); + } + for (key, value) in fields { + visit( + value, + &format!("{label}[{}]", serde_json::to_string(key).unwrap()), + out, + ); + } + } + Value::Array(items) => { + if items.is_empty() { + out.push_str(&format!("{label}: []\n")); + } + for (i, value) in items.iter().enumerate() { + visit(value, &format!("{label}[{i}]"), out); + } + } + Value::String(text) => { + out.push_str(&format!("{label} (string, {} bytes):\n", text.len())); + out.push_str(text); + out.push('\n'); + } + _ => { + out.push_str(&format!("{label}: {value}\n")); + } } - output.truncate(end); - output.push_str(MARKER); } - output + let mut out = String::new(); + visit(value, "$", &mut out); + out } /// Describe the five tools sent to either inference API. @@ -26,7 +52,7 @@ pub fn definitions() -> Vec { vec![ Tool { name: "call", - description: "Call a plugin method through the core. Await its result. Set detach to let its process survive this caller. Output is capped at 65536 bytes with a truncation marker.", + description: "Call a plugin method through the core. Await its result. Set detach to let its process survive this caller. Large results show an excerpt with a full-output /logs reference.", parameters: schema( json!({ "method": {"type": "string", "description": "Qualified plugin method"}, @@ -38,19 +64,21 @@ pub fn definitions() -> Vec { }, Tool { name: "read_file", - description: "Read a text file in the world, including `/proc`. `offset` and `max_bytes` are optional. Output is capped at 65536 bytes with a truncation marker.", + description: "Read a text file in the world, including `/proc`. `offset`/`max_bytes` select bytes; `line_offset`/`line_limit` select zero-based lines (default 200). Large results show an excerpt with a full-output /logs reference.", parameters: schema( json!({ "path": {"type": "string"}, "offset": {"type": "integer", "description": "Byte offset (default 0)"}, - "max_bytes": {"type": "integer", "description": "Optional maximum number of bytes to read"} + "max_bytes": {"type": "integer", "description": "Optional maximum number of bytes to read"}, + "line_offset": {"type": "integer", "description": "Zero-based starting line; overrides bytes"}, + "line_limit": {"type": "integer", "description": "Number of lines; default 200"} }), &["path"], ), }, Tool { name: "write_file", - description: "Write a text file in the world. Parent folders are created. Output is capped at 65536 bytes with a truncation marker.", + description: "Write a text file in the world. Parent folders are created. Large results show an excerpt with a full-output /logs reference.", parameters: schema( json!({"path": {"type": "string"}, "content": {"type": "string"}}), &["path", "content"], @@ -58,12 +86,12 @@ pub fn definitions() -> Vec { }, Tool { name: "list_dir", - description: "List entries in a world directory. Returns each name, type, and size. Output is capped at 65536 bytes with a truncation marker.", + description: "List entries in a world directory. Returns each name, type, and size. Large results show an excerpt with a full-output /logs reference.", parameters: schema(json!({"path": {"type": "string"}}), &["path"]), }, Tool { name: "copy_dir", - description: "Copy a directory tree, for example a plugin source to a draft folder. Skips `target/` folders. Output is capped at 65536 bytes with a truncation marker.", + description: "Copy a directory tree, for example a plugin source to a draft folder. Skips `target/` folders. Large results show an excerpt with a full-output /logs reference.", parameters: schema( json!({"from": {"type": "string"}, "to": {"type": "string"}}), &["from", "to"], @@ -103,6 +131,16 @@ pub async fn run(name: &str, args: &Value) -> sausage_plugin::Result { fn read_file(args: &Value) -> io::Result { let path = args["path"].as_str().unwrap_or_default(); let bytes = std::fs::read(path)?; + if path.starts_with("/logs/") { + return Ok( + json!({"path":path,"size":bytes.len(),"content":overflow::existing(&bytes, path, args)}), + ); + } + if let Some(line) = args["line_offset"].as_u64() { + return Ok( + json!({"path":path,"size":bytes.len(),"line_offset":line,"content":overflow::read_lines(&bytes, line as usize, args["line_limit"].as_u64().unwrap_or(200) as usize)}), + ); + } let offset = (args["offset"].as_u64().unwrap_or(0) as usize).min(bytes.len()); let end = args["max_bytes"].as_u64().map_or(bytes.len(), |max| { offset.saturating_add(max as usize).min(bytes.len()) @@ -185,11 +223,35 @@ fn copy_tree(from: &str, to: &str) -> io::Result { #[cfg(test)] mod tests { #[test] - fn tool_output_limit_preserves_utf8_and_marks_truncation() { - let limited = super::limit_output("😀".repeat(40_000)); - assert!(limited.len() <= super::MAX_OUTPUT_BYTES); - assert!(limited.ends_with("[tool output truncated at 65536 bytes]")); - assert_eq!(super::limit_output("small".into()), "small"); + fn overflow_json_can_be_followed_by_line_without_rearchiving() { + let value = serde_json::json!({"stdout":"hello 😀\n".repeat(4000),"exit":1,"error":{"text":"failure\n".repeat(4000)}}); + let expected = super::render(&value); + let shown = super::overflow::guard(value.to_string(), "json-test", "call", true); + assert!(shown.len() <= 50 * 1024 && shown.lines().count() <= 2000); + let path = shown + .split("Full output: ") + .nth(1) + .unwrap() + .split(". Read with") + .next() + .unwrap(); + assert_eq!(std::fs::read_to_string(path).unwrap(), expected); + let parent = std::path::Path::new(path).parent().unwrap(); + let count = std::fs::read_dir(parent).unwrap().count(); + let args = serde_json::json!({"path":path,"line_offset":30,"line_limit":5}); + let read = super::read_file(&args).unwrap(); + let read = read["content"].as_str().unwrap(); + assert!(read.contains("Showing lines 31-35")); + assert!(read.contains(path)); + assert_eq!(std::fs::read_dir(parent).unwrap().count(), count); + assert!( + super::overflow::existing( + expected.as_bytes(), + path, + &serde_json::json!({"line_offset":u64::MAX}) + ) + .contains("Full output:") + ); } #[test] fn copy_rejects_a_destination_inside_the_source() { diff --git a/cli/src/host.rs b/cli/src/host.rs index 2cdd551..e0f4785 100644 --- a/cli/src/host.rs +++ b/cli/src/host.rs @@ -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 || { diff --git a/cli/src/journal.rs b/cli/src/journal.rs index 8d9f37a..5b26b14 100644 --- a/cli/src/journal.rs +++ b/cli/src/journal.rs @@ -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, warned: Mutex>, + spool: PathBuf, } impl Journal { pub(crate) async fn memory() -> anyhow::Result { Self::open(PathBuf::from(":memory:")).await } pub(crate) async fn open(path: PathBuf) -> anyhow::Result { + Self::open_with_legacy(path, None).await + } + pub(crate) async fn open_with_legacy( + path: PathBuf, + legacy: Option, + ) -> anyhow::Result { + 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 { CREATE INDEX IF NOT EXISTS journal_plugin ON journal(plugin);")?; Ok(db) } -fn write_batch(db: &mut Connection, batch: Vec) { - 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 { + 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, + batch: Vec, +) -> Vec { + 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 { + 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 { let mut statement = db.prepare( @@ -135,6 +244,140 @@ fn query_database(db: &Connection, filters: &Value) -> rusqlite::Result { #[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 = 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() { diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index fc6899c..a994ff1 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -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"); + } +} diff --git a/cli/tests/crates/cargo-valid/src/main.rs b/cli/tests/crates/cargo-valid/src/main.rs index 73d7a70..19f1125 100644 --- a/cli/tests/crates/cargo-valid/src/main.rs +++ b/cli/tests/crates/cargo-valid/src/main.rs @@ -92,6 +92,7 @@ mod tests { "/config", "/state", "/state/core/journal.sqlite", + "/logs/core/journal.sqlite", ] { assert!(!std::path::Path::new(path).exists(), "{path}"); } diff --git a/cli/tests/host/agent.rs b/cli/tests/host/agent.rs index a33d08a..c532869 100644 --- a/cli/tests/host/agent.rs +++ b/cli/tests/host/agent.rs @@ -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::(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"}))) diff --git a/cli/tests/host/cli.rs b/cli/tests/host/cli.rs index b38ac39..dadeeda 100644 --- a/cli/tests/host/cli.rs +++ b/cli/tests/host/cli.rs @@ -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 {}'", [], diff --git a/cli/tests/host/core.rs b/cli/tests/host/core.rs index b12a6bb..2a062d2 100644 --- a/cli/tests/host/core.rs +++ b/cli/tests/host/core.rs @@ -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 = 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; +} diff --git a/cli/tests/plugins/probe/src/main.rs b/cli/tests/plugins/probe/src/main.rs index 48f7da7..9ae4d44 100644 --- a/cli/tests/plugins/probe/src/main.rs +++ b/cli/tests/plugins/probe/src/main.rs @@ -376,6 +376,7 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option { 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" => { diff --git a/sdk/tests/wasi.rs b/sdk/tests/wasi.rs index a56e89c..b7bfcf2 100644 --- a/sdk/tests/wasi.rs +++ b/sdk/tests/wasi.rs @@ -34,10 +34,8 @@ impl Process { let mut s = String::new(); if self.output.read_line(&mut s).unwrap() == 0 { let mut stderr = String::new(); - let _ = std::io::Read::read_to_string( - self.child.stderr.as_mut().unwrap(), - &mut stderr, - ); + let _ = + std::io::Read::read_to_string(self.child.stderr.as_mut().unwrap(), &mut stderr); panic!("unexpected EOF from wasmtime. Its stderr:\n{stderr}"); } if let Ok(v) = sausage_plugin::serde_json::from_str::(&s)