Merge 'e26260d36eb8e7bf9e0b7edfed90c0dcff77fc19'

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

DECISIONS.md

@@ -61,6 +61,20 @@ a plugin, or a change that the agent makes to itself.
 
 ## Small choices
 
+- **Restart is a clean stop supervised by `#loop`.** `core.restart` requires
+  the host startup marker `SAUSAGE_LOOP=1`, which the loop exports for its
+  invocation. Without a supervisor, restarting would only stop the world.
+  A single pending request answers first, allows 250 ms for delivery, then
+  signals the existing shutdown path; it does not use `process::exit` or
+  bypass detached-process cleanup, lock release, or journal draining. The
+  executable returns dedicated restart code **75** after cleanup. `#loop`
+  refreshes and starts again on 75 (and other nonzero exits); 0 stops it.
+  `core.exit` is a compatibility alias with the same supervisor gate,
+  response grace, and coalescing; new callers use `core.restart`.
+  **HARD RULE: the mediator and all agents NEVER call `core.restart` without
+  the person's explicit approval each time.** This
+  approval rule also covers legacy `core.exit`; it must not bypass the gate.
+
 - **YAML for configuration, JSON for data.** A person edits configuration, so
   it needs comments. `init` writes commented templates instead of asking
   questions, so there is one way to configure a world.
@@ -70,6 +84,16 @@ a plugin, or a change that the agent makes to itself.
   that occur only in WASI, for example a missing WASI function or a panic at
   start. The test runner uses the same Wasmtime and the same mounts as the
   core.
+- **Tests never run without a timeout.** `cargo.build` serializes its work
+  folder, so one hung test or build script must not block all later builds.
+  `timeout_secs` defaults to 900 seconds after mutex acquisition, excluding
+  queue time; `test_timeout_secs` defaults to 120 seconds per WASI instance.
+  Both are optional positive integers, never null/unlimited. Test timeouts name
+  the failing test and let other instances finish. Build deadlines and
+  cancellation kill cargo's process group before dropping the mutex; timeout
+  errors retain the phase, elapsed time, and recent output. Source preparation
+  and publication retain serialization while their blocking filesystem work
+  finishes, even if cancelled, to avoid overlapping writes to the work folder.
 - **The SDK is not published.** `cargo` points `sausage-plugin = "0.1"` at
   its own copy, which it also shows in `/plugins/cargo/sdk/`.
 - **Compiled code is a cache, not state.** It lives outside the world, in

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/<plugin>.json   descriptions of the imported plugins
   state/<plugin>/             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
@@ -322,7 +337,37 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 | `core.activate` | `plugin` and `wasm`, or `plugins: [{plugin, wasm}]`. `force`? | the new `rpc.discover` documents |
 | `core.revert` | `plugin`, `force`? | like `core.activate` |
 | `core.log` | filters (section 7) | journal entries |
-| `core.exit` | none | `{}`. Then the host stops with exit code 75. |
+| `core.restart` | none | `{}`. With `SAUSAGE_LOOP=1`, schedule clean shutdown and exit code 75. |
+| `core.exit` | none | Alias of `core.restart`: same supervisor gate, response, coalescing, and exit code 75. |
+
+### Host restart
+
+**HARD RULE: the mediator and all agents NEVER call `core.restart` without
+the person's explicit approval each time.** The same approval rule applies
+to the legacy alias `core.exit`; it uses the same gate and pending request.
+
+`core.restart {}` is enabled only when the host starts with `SAUSAGE_LOOP=1`.
+The host captures the marker at startup. An absent marker (or any value other
+than `1`) returns error -32600: `restart needs the loop supervisor
+(SAUSAGE_LOOP not set); start with SAUSAGE_LOOP=1`, without scheduling an exit.
+This is a supervisor marker, not a permission system.
+
+The method returns `{}` immediately and schedules the shutdown request after
+250 milliseconds to let the response reach the caller before stdin closes.
+Repeated requests while a restart is pending return `{}` without scheduling
+another restart or extending that delay. The executable then uses the same
+`Host::stop` path as a normal stop: reject new processes, stop all processes
+(including detached ones) in parallel with the normal five-second grace,
+wait for outstanding file operations, release locks, and drain journal writes.
+Only after that does the executable return the dedicated restart exit code
+**75**. The library signals this path; it never kills its embedding process.
+
+`nix run .#loop -- <world>` sets `SAUSAGE_LOOP=1` on the host invocation and
+starts the newest host again after exit code 75 (and, as before, other nonzero
+exits). Exit code 0 ends the loop. If using an older loop script, the person
+must set `SAUSAGE_LOOP=1` in that supervisor's environment; do not set it for
+an unsupervised host. The NixEnv service invokes this flake's `#loop`, so a
+new loop script supplies the marker without changes to NixEnv.
 
 Everything else is a file.
 
@@ -418,9 +463,37 @@ builds `/plugins/notes/source/`:
 3. It builds the release `.wasm` for `wasm32-wasip2`. The tests and the build
    share the compiled dependencies.
 4. It returns `{"wasm": "/state/cargo/builds/<n>/notes.wasm", "tests":
-   {"passed", "failed", "output"}, "warnings"}`. If a step fails, it returns
+   {"passed", "failed", "output"}, "warnings", "logs"}`. If a step fails, it returns
    an error with the compiler or test output.
 
+- Before running commands, it creates `/logs/cargo/<YYYY-MM-DD>/<build-id>.stdout.log`
+  and `.stderr.log` (UTC date, unique build ID). Both test and release phases
+  append their raw bytes directly to these files, before any parsing, progress
+  filtering or UTF-8 conversion. This includes successful JSON build output.
+  `/logs` is a top-level world folder, created if missing, not under `/state`.
+- Success results and error -32004 `data` include `logs`:
+  `{"stdout":{"path":"/logs/cargo/...stdout.log","bytes":123},
+  "stderr":{"path":"/logs/cargo/...stderr.log","bytes":45},"total_bytes":168}`.
+  Counts are raw bytes, not characters. `logs` is null only if log creation
+  itself fails. Preparation failures otherwise include empty log files.
+  Inline diagnostics may be filtered or shortened; the files are the full truth.
+- Logs remain after failure, total/test timeout, and cancellation. Cancellation
+  need not return a result; its files remain discoverable in `/logs/cargo/`.
+  Pruning numeric `/state/cargo/builds/` artifact folders does not prune logs.
+- Optional `timeout_secs` is an integer of at least 1, default **900** (not
+  null or unlimited). The wall-clock budget starts when the build acquires its
+  mutex, excluding queue time, and covers tests and release compilation together.
+  On expiry, cargo and its descendant process group are killed and the mutex is
+  released. Error -32004 has reason/message `timeout`, with `data.reason`,
+  `data.elapsed_secs`, `data.step` (`tests` or `release build`), `data.tests`,
+  and `data.output` containing the last 50 output lines.
+- Optional `test_timeout_secs` is an integer of at least 1, default **120**.
+  Each WASI test instance has this wall-clock budget. On expiry, the instance
+  is killed, its named test fails with `timed out after N s`, and remaining
+  tests continue. The overall build budget still applies.
+- Cancellation kills cargo's process group, including build scripts and test
+  runners, before releasing the build mutex. An attached build also stops when
+  its caller ends (under the normal process stopping deadline).
 - A call to `cargo.build` gets a process ID, a folder in `/proc`, and journal
   entries, like other calls.
 - `/plugins/cargo/sdk/` shows the source of the SDK. The dependency
@@ -443,3 +516,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/<date>/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.

README.md

@@ -45,9 +45,18 @@ nix run git+https://gitea.surma.technology/surma/sausage#loop               # ./
 nix run git+https://gitea.surma.technology/surma/sausage#loop -- my-world   # an existing world
 ```
 
-When the host crashes or a plugin calls `core.exit`, the loop fetches the
-newest commit and starts it again. `quit` ends the loop. Set `SAUSAGE_FLAKE`
-to run another flake URL instead, for example your fork.
+When the host crashes or a plugin calls `core.restart`, the loop fetches the
+newest commit and starts it again. The loop sets `SAUSAGE_LOOP=1` to enable
+that RPC; without it, restart returns an error and leaves the host running.
+Restart answers first, then cleanly stops all processes and exits with code
+75. `quit` (code 0) ends the loop. Set `SAUSAGE_FLAKE` to run another flake
+URL instead, for example your fork.
+
+**The mediator and all agents NEVER call `core.restart` without the person's
+explicit approval each time.** This also applies to legacy `core.exit`, an
+alias with the same supervisor gate, response-before-exit behavior, and
+request coalescing.
+See [the protocol](PROTOCOL.md#host-restart) for the shutdown contract.
 
 From a clone (`git clone https://gitea.surma.technology/surma/sausage`), the
 same commands are `nix run`, `nix run -- init`, `nix run -- run`, and
@@ -57,7 +66,7 @@ same commands are `nix run`, `nix run -- init`, `nix run -- run`, and
 
 `--headless` (or `SAUSAGE_HEADLESS=1`) turns the console input off. The host
 runs until SIGTERM or SIGINT (exit code 0) or until a plugin calls
-`core.exit` (exit code 75, so the loop starts the newest version):
+`core.restart` (requires `SAUSAGE_LOOP=1`; exit code 75 restarts the loop):
 
 ```sh
 nix run git+https://gitea.surma.technology/surma/sausage#loop -- --headless my-world

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::<serde_json::Value>(&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 })?;
             }
         }
     }

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::<serde_json::Value>(&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<String> {
+        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::<usize>();
+        let len = text[from..]
+            .split_inclusive('\n')
+            .take(args["line_limit"].as_u64().unwrap_or(200) as usize)
+            .map(str::len)
+            .sum::<usize>();
+        (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::<usize>();
+    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), "");
+    }
+}

agent/src/prompt.md

@@ -3,6 +3,8 @@ You are the Sausage agent. Answer the person. Use tools to inspect or change the
 Read the contract at `/proc/core/PROTOCOL.md`. The Rust plugin SDK and its documentation are at `/plugins/cargo/sdk/`.
 Build a plugin with `cargo.build`. Activate its WASM file with `core.activate`.
 Your own source is at `/plugins/agent/source/`. You can change and rebuild it like any other plugin.
+HARD RULE: the mediator and all agents NEVER call `core.restart` (or its alias `core.exit`) without the person's explicit approval each time.
+Restart requires the host to run under `#loop` (`SAUSAGE_LOOP=1`); it answers before clean shutdown with exit code 75.
 Never call `agent.chat`. It would wait on the history lock that your current turn holds.
 Write logs to stderr. Use call results and notifications for communication.
 The interface check does not cover persistent state. Preserve state compatibility when you change a plugin.

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<Tool> {
     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> {
         },
         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> {
         },
         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<Value> {
 fn read_file(args: &Value) -> io::Result<Value> {
     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<u64> {
 #[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() {

cli/Cargo.toml

@@ -36,6 +36,7 @@ chrono = { version = "0.4.42", default-features = false, features = ["clock"] }
 tempfile = "3.27.0"
 
 [dev-dependencies]
+tokio = { version = "1", features = ["test-util"] }
 hyper = { version = "1.12.0", features = ["server", "http1"] }
 hyper-util = { version = "0.1.21", features = ["tokio"] }
 sausage-interface = { path = "../interface" }

cli/src/boot.yaml

@@ -2,6 +2,9 @@
 # Each method is a plugin.method name. params contains its named arguments.
 # The core waits for each result before it starts the next service.
 # After a plugin upgrade, the core restarts that plugin's services.
+# Host restart is separate: #loop sets SAUSAGE_LOOP=1 to enable core.restart
+# (and its alias core.exit). It answers before clean shutdown with exit code 75.
+# Agents need the person's explicit approval each time; never add it as a service.
 # Example (replace [] with this list):
 # - method: www.start
 #   params:

cli/src/cargo.rs

@@ -4,11 +4,44 @@ use anyhow::{Context, Result, bail};
 use include_dir::{Dir, include_dir};
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
+use std::time::Duration;
 use std::{
     path::{Path, PathBuf},
     sync::Arc,
 };
-use tokio::process::Command;
+use tokio::{process::Command, time::Instant};
+
+/// Default wall-clock budget after acquiring the build mutex (not queue time).
+pub const DEFAULT_TIMEOUT_SECS: u64 = 900;
+/// Default wall-clock budget for each WASI test instance.
+pub const DEFAULT_TEST_TIMEOUT_SECS: u64 = 120;
+
+/// Wall-clock budgets for a build and its individual WASI tests.
+#[derive(Clone, Copy, Debug)]
+pub struct BuildOptions {
+    pub timeout_secs: u64,
+    pub test_timeout_secs: u64,
+}
+impl Default for BuildOptions {
+    fn default() -> Self {
+        Self {
+            timeout_secs: DEFAULT_TIMEOUT_SECS,
+            test_timeout_secs: DEFAULT_TEST_TIMEOUT_SECS,
+        }
+    }
+}
+/// Optional RPC seconds are positive integers; null does not disable a deadline.
+fn seconds(params: &Value, name: &str, default: u64) -> Result<u64, Value> {
+    match params.get(name) {
+        None => Ok(default),
+        Some(value) => value
+            .as_u64()
+            .filter(|seconds| *seconds >= 1)
+            .ok_or_else(|| {
+                crate::error(-32602, &format!("{name} must be an integer of at least 1"))
+            }),
+    }
+}
 
 /// The native build plugin.
 pub const ID: &str = "cargo";
@@ -25,9 +58,78 @@ pub struct TestResults {
     pub failed: u64,
     pub output: String,
 }
+/// Raw, unfiltered output shared by the test and release phases of one build.
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct BuildLogs {
+    pub stdout: StreamLog,
+    pub stderr: StreamLog,
+    pub total_bytes: u64,
+}
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct StreamLog {
+    pub path: String,
+    pub bytes: u64,
+}
+/// Persistent descriptors are inherited by cargo, so cancellation cannot lose
+/// bytes already written by cargo or its descendants. No buffered writer is used.
+struct LogFiles {
+    stdout: std::fs::File,
+    stderr: std::fs::File,
+    stdout_path: PathBuf,
+    stderr_path: PathBuf,
+}
+impl LogFiles {
+    fn create(world: &Path) -> Result<Self> {
+        let directory = world
+            .join("logs/cargo")
+            .join(chrono::Utc::now().format("%Y-%m-%d").to_string());
+        std::fs::create_dir_all(&directory).context("create cargo log directory")?;
+        let id = uuid::Uuid::now_v7();
+        let stdout_path = directory.join(format!("{id}.stdout.log"));
+        let stderr_path = directory.join(format!("{id}.stderr.log"));
+        let create = |path: &Path| {
+            std::fs::OpenOptions::new()
+                .create_new(true)
+                .append(true)
+                .open(path)
+                .with_context(|| format!("create {}", path.display()))
+        };
+        Ok(Self {
+            stdout: create(&stdout_path)?,
+            stderr: create(&stderr_path)?,
+            stdout_path,
+            stderr_path,
+        })
+    }
+    fn info(&self, world: &Path) -> Result<BuildLogs> {
+        let stream = |file: &std::fs::File, path: &Path| -> Result<StreamLog> {
+            Ok(StreamLog {
+                path: format!("/{}", path.strip_prefix(world)?.display()),
+                bytes: file.metadata()?.len(),
+            })
+        };
+        let stdout = stream(&self.stdout, &self.stdout_path)?;
+        let stderr = stream(&self.stderr, &self.stderr_path)?;
+        Ok(BuildLogs {
+            total_bytes: stdout.bytes + stderr.bytes,
+            stdout,
+            stderr,
+        })
+    }
+    fn read(&self, path: &Path, offset: u64) -> Result<Vec<u8>> {
+        use std::io::{Read, Seek, SeekFrom};
+        let mut file =
+            std::fs::File::open(path).with_context(|| format!("read {}", path.display()))?;
+        file.seek(SeekFrom::Start(offset))?;
+        let mut bytes = Vec::new();
+        file.read_to_end(&mut bytes)?;
+        Ok(bytes)
+    }
+}
 /// A successful build, usable without a running host.
 #[derive(Debug)]
 pub struct BuildOutput {
+    pub logs: Option<BuildLogs>,
     pub wasm: PathBuf,
     pub warnings: String,
     pub tests: TestResults,
@@ -35,9 +137,12 @@ pub struct BuildOutput {
 /// A failed compiler or test phase, with its complete diagnostics.
 #[derive(Debug)]
 pub struct BuildFailure {
+    pub logs: Option<BuildLogs>,
     pub reason: String,
     pub tests: TestResults,
     pub output: String,
+    pub elapsed_secs: Option<f64>,
+    pub step: Option<&'static str>,
 }
 impl std::fmt::Display for BuildFailure {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -48,9 +153,12 @@ impl std::error::Error for BuildFailure {}
 impl From<anyhow::Error> for BuildFailure {
     fn from(error: anyhow::Error) -> Self {
         Self {
+            logs: None,
             reason: "build preparation failed".into(),
             output: format!("{error:#}"),
             tests: TestResults::default(),
+            elapsed_secs: None,
+            step: None,
         }
     }
 }
@@ -142,19 +250,59 @@ impl Cargo {
         host.register_native(ID,Arc::new(move |_context,method,params| {
             let builder=builder.clone();
             Box::pin(async move {
-                if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","params":[{"name":"plugin","required":true,"schema":{"type":"string"}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
+                if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings","logs"],"properties":{"logs":{"type":"object","required":["stdout","stderr","total_bytes"],"properties":{"stdout":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"stderr":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"total_bytes":{"type":"integer","minimum":0}}},"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
                 if method!="build" { return Err(crate::error(-32601,"cargo method not found")); }
                 let plugin=params["plugin"].as_str().ok_or_else(||crate::error(-32602,"plugin must be a string"))?;
-                match builder.build(&builder.world.plugin_dir(plugin).join("source")).await {
-                    Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings})),
-                    Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"tests":failure.tests,"output":failure.output}})),
+                let options = BuildOptions {
+                    timeout_secs: seconds(&params, "timeout_secs", DEFAULT_TIMEOUT_SECS)?,
+                    test_timeout_secs: seconds(&params, "test_timeout_secs", DEFAULT_TEST_TIMEOUT_SECS)?,
+                };
+                match builder.build_with_options(&builder.world.plugin_dir(plugin).join("source"), options).await {
+                    Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings,"logs":output.logs})),
+                    Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step,"logs":failure.logs}})),
                 }
             })
         }));
     }
     /// Copies source once, runs WASI tests, then publishes a release component under state/cargo/builds.
     pub async fn build(&self, source: &Path) -> std::result::Result<BuildOutput, BuildFailure> {
+        self.build_with_options(source, BuildOptions::default())
+            .await
+    }
+    /// Like build, with explicit positive wall-clock budgets. Waiting for the mutex is unbounded.
+    pub async fn build_with_options(
+        &self,
+        source: &Path,
+        options: BuildOptions,
+    ) -> std::result::Result<BuildOutput, BuildFailure> {
+        let logs = LogFiles::create(&self.world.root)?;
+        let result = self.build_logged(source, options, &logs).await;
+        let info = logs.info(&self.world.root)?;
+        match result {
+            Ok(mut output) => {
+                output.logs = Some(info);
+                Ok(output)
+            }
+            Err(mut failure) => {
+                failure.logs = Some(info);
+                Err(failure)
+            }
+        }
+    }
+    async fn build_logged(
+        &self,
+        source: &Path,
+        options: BuildOptions,
+        logs: &LogFiles,
+    ) -> std::result::Result<BuildOutput, BuildFailure> {
+        if options.timeout_secs == 0 || options.test_timeout_secs == 0 {
+            return Err(anyhow::anyhow!("timeouts must be at least 1 second").into());
+        }
         let build_guard = self.builds.clone().lock_owned().await;
+        let started = Instant::now();
+        let deadline = started
+            .checked_add(Duration::from_secs(options.timeout_secs))
+            .ok_or_else(|| anyhow::anyhow!("timeout_secs is too large"))?;
         let work = self.world.state_dir(ID).join("work");
         let target = self.target.clone();
         let temporary = self.world.state_dir(ID).join("tmp");
@@ -208,16 +356,35 @@ impl Cargo {
         .await
         .map_err(anyhow::Error::from)??;
         let mut test = Command::new("cargo");
-        test.args(["test", "--release", "--target", TARGET, "--target-dir"])
-            .arg(&target)
-            .current_dir(&work)
-            .env("TMPDIR", &temporary)
-            .arg("--config")
-            .arg(format!(
-                "target.wasm32-wasip2.runner={}",
-                json!([self.runner, "test-runner"])
-            ));
-        let output = run_command(test).await?;
+        test.args([
+            "test",
+            "--no-fail-fast",
+            "--release",
+            "--target",
+            TARGET,
+            "--target-dir",
+        ])
+        .arg(&target)
+        .current_dir(&work)
+        .env("TMPDIR", &temporary)
+        .env(
+            "SAUSAGE_TEST_TIMEOUT_SECS",
+            options.test_timeout_secs.to_string(),
+        )
+        .arg("--config")
+        .arg(format!(
+            "target.wasm32-wasip2.runner={}",
+            json!([self.runner, "test-runner"])
+        ));
+        let output = run_command(
+            test,
+            started,
+            deadline,
+            "tests",
+            TestResults::default(),
+            logs,
+        )
+        .await?;
         let stdout = String::from_utf8_lossy(&output.stdout);
         let stderr = String::from_utf8_lossy(&output.stderr);
         let tests = TestResults {
@@ -231,25 +398,36 @@ impl Cargo {
         };
         if !output.status.success() {
             return Err(BuildFailure {
+                logs: None,
                 reason: "WASI tests failed".into(),
                 output: tests.output.clone(),
                 tests,
+                elapsed_secs: None,
+                step: None,
             });
         }
+        let mut build = Command::new("cargo");
+        build
+            .args([
+                "build",
+                "--release",
+                "--target",
+                TARGET,
+                "--message-format=json-render-diagnostics",
+                "--target-dir",
+            ])
+            .arg(&target)
+            .current_dir(&work);
+        let output = run_command(
+            build,
+            started,
+            deadline,
+            "release build",
+            tests.clone(),
+            logs,
+        )
+        .await?;
         let result = async {
-            let mut build = Command::new("cargo");
-            build
-                .args([
-                    "build",
-                    "--release",
-                    "--target",
-                    TARGET,
-                    "--message-format=json-render-diagnostics",
-                    "--target-dir",
-                ])
-                .arg(&target)
-                .current_dir(&work);
-            let output = run_command(build).await?;
             let stdout = String::from_utf8_lossy(&output.stdout);
             let stderr = String::from_utf8_lossy(&output.stderr);
             let diagnostics = stderr.to_string();
@@ -282,6 +460,7 @@ impl Cargo {
             })
             .await??;
             Ok(BuildOutput {
+                logs: None,
                 wasm: destination,
                 warnings: strip_progress(&diagnostics),
                 tests: tests.clone(),
@@ -289,9 +468,12 @@ impl Cargo {
         }
         .await;
         result.map_err(|error: anyhow::Error| BuildFailure {
+            logs: None,
             reason: "release build failed".into(),
             output: format!("{error:#}"),
             tests,
+            elapsed_secs: None,
+            step: None,
         })
     }
 }
@@ -306,16 +488,81 @@ impl Drop for ProcessGroup {
     }
 }
 /// Starts one process group and retains its guard across all output collection.
-async fn run_command(mut command: Command) -> Result<std::process::Output> {
+async fn run_command(
+    mut command: Command,
+    started: Instant,
+    deadline: Instant,
+    step: &'static str,
+    mut tests: TestResults,
+    logs: &LogFiles,
+) -> std::result::Result<std::process::Output, BuildFailure> {
     command
         .process_group(0)
         .kill_on_drop(true)
         .stdin(std::process::Stdio::null())
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped());
-    let child = command.spawn().context("start cargo")?;
-    let _group = ProcessGroup(child.id().unwrap());
-    Ok(child.wait_with_output().await?)
+        // Child writes go straight to persistent files: no pipe buffers or reader
+        // tasks can be dropped on cancellation. Parsing happens only afterwards.
+        .stdout(logs.stdout.try_clone().context("clone stdout log")?)
+        .stderr(logs.stderr.try_clone().context("clone stderr log")?);
+    let out_start = logs.stdout.metadata().context("stat stdout log")?.len();
+    let err_start = logs.stderr.metadata().context("stat stderr log")?.len();
+    let mut child = command.spawn().context("start cargo")?;
+    let group = ProcessGroup(child.id().unwrap());
+
+    let (status, timed_out) = {
+        let collect = child.wait();
+        tokio::pin!(collect);
+        tokio::select! {
+            biased;
+            _ = tokio::time::sleep_until(deadline) => {
+                // Kill the entire tree before waiting for EOF/reaping cargo. The guard also
+                // covers cancellation while collecting, and runs before the mutex drops.
+                let _ = nix::sys::signal::killpg(nix::unistd::Pid::from_raw(group.0 as i32), nix::sys::signal::Signal::SIGKILL);
+                (collect.await, true)
+            }
+            status = &mut collect => (status, false),
+        }
+    };
+    let status = status.map_err(anyhow::Error::from)?;
+    // Also stop any descendants that kept the streams open after cargo exited.
+    drop(group);
+    let out = logs.read(&logs.stdout_path, out_start)?;
+    let err = logs.read(&logs.stderr_path, err_start)?;
+    if timed_out {
+        let output = format!(
+            "{}\n{}",
+            String::from_utf8_lossy(&out),
+            String::from_utf8_lossy(&err)
+        );
+        let output = output
+            .trim_end()
+            .lines()
+            .rev()
+            .take(50)
+            .collect::<Vec<_>>()
+            .into_iter()
+            .rev()
+            .collect::<Vec<_>>()
+            .join("\n");
+        if step == "tests" {
+            tests.passed = test_count(&String::from_utf8_lossy(&out), "passed");
+            tests.failed = test_count(&String::from_utf8_lossy(&out), "failed");
+            tests.output = output.clone();
+        }
+        return Err(BuildFailure {
+            logs: None,
+            reason: "timeout".into(),
+            tests,
+            output,
+            elapsed_secs: Some(started.elapsed().as_secs_f64()),
+            step: Some(step),
+        });
+    }
+    Ok(std::process::Output {
+        status,
+        stdout: out,
+        stderr: err,
+    })
 }
 /// Copies only source, excluding build outputs and git internals at every level.
 fn copy_source(source: &Path, destination: &Path, root: bool) -> Result<()> {
@@ -457,3 +704,143 @@ fn strip_progress(output: &str) -> String {
         .collect::<Vec<_>>()
         .join("\n")
 }
+
+#[cfg(test)]
+mod timeout_tests {
+    use super::*;
+
+    /// Streams retain exact bytes, including non-UTF8, progress and JSON; phases append.
+    #[tokio::test]
+    async fn raw_streams_append_before_parsing() {
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
+        for _ in 0..2 {
+            let mut command = Command::new("sh");
+            command.args(["-c", "printf '\\377raw\\r'; printf '   Compiling progress\\n{\"reason\":\"compiler-artifact\"}\\n' >&2; exit 3"]);
+            let now = Instant::now();
+            let output = tokio::time::timeout(
+                Duration::from_secs(5),
+                run_command(
+                    command,
+                    now,
+                    now + Duration::from_secs(3),
+                    "tests",
+                    TestResults::default(),
+                    &logs,
+                ),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+            assert!(!output.status.success());
+            assert_eq!(output.stdout, b"\xffraw\r");
+            assert!(String::from_utf8_lossy(&output.stderr).contains("Compiling progress"));
+        }
+        assert_eq!(
+            std::fs::read(&logs.stdout_path).unwrap(),
+            b"\xffraw\r\xffraw\r"
+        );
+        let info = logs.info(directory.path()).unwrap();
+        assert_eq!(info.stdout.bytes, 10);
+        assert_eq!(info.total_bytes, info.stdout.bytes + info.stderr.bytes);
+        assert!(info.stdout.path.starts_with("/logs/cargo/"));
+        assert!(!directory.path().join("state/cargo").exists());
+    }
+
+    /// Cancellation leaves kernel-written data intact, even without a newline.
+    #[tokio::test]
+    async fn cancelled_command_keeps_logs() {
+        use notify::Watcher;
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut watcher =
+            notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
+                if let Ok(event) = event {
+                    if event
+                        .paths
+                        .iter()
+                        .any(|path| path.file_name().is_some_and(|name| name == "ready"))
+                    {
+                        let _ = tx.send(());
+                    }
+                }
+            })
+            .unwrap();
+        watcher
+            .watch(directory.path(), notify::RecursiveMode::NonRecursive)
+            .unwrap();
+        let mut command = Command::new("sh");
+        command.args(["-c", "printf 'partial\\377'; printf 'stderr\\r' >&2; touch \"$1/ready\"; exec sleep 1000", "sh"]).arg(directory.path());
+        let now = Instant::now();
+        let task = tokio::spawn(async move {
+            run_command(
+                command,
+                now,
+                now + Duration::from_secs(60),
+                "tests",
+                TestResults::default(),
+                &logs,
+            )
+            .await
+        });
+        tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        let paths: Vec<_> = std::fs::read_dir(directory.path().join("logs/cargo"))
+            .unwrap()
+            .flat_map(|day| std::fs::read_dir(day.unwrap().path()).unwrap())
+            .map(|file| file.unwrap().path())
+            .collect();
+        assert_eq!(paths.len(), 2);
+        for path in paths {
+            let expected: &[u8] = if path.to_string_lossy().ends_with(".stdout.log") {
+                b"partial\xff"
+            } else {
+                b"stderr\r"
+            };
+            assert_eq!(std::fs::read(path).unwrap(), expected);
+        }
+    }
+
+    /// Deadline errors retain the shared start time, phase, test totals, and output tail.
+    #[tokio::test]
+    async fn timeout_diagnostics() {
+        let mut command = Command::new("sh");
+        command.args([
+            "-c",
+            "i=0; while [ $i -lt 100 ]; do echo diagnostic-$i; i=$((i+1)); done; exec sleep 1000",
+        ]);
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
+        let now = Instant::now();
+        let failure = run_command(
+            command,
+            now - Duration::from_secs(10),
+            now + Duration::from_secs(1),
+            "release build",
+            TestResults {
+                passed: 7,
+                failed: 0,
+                output: "successful tests".into(),
+            },
+            &logs,
+        )
+        .await
+        .unwrap_err();
+        assert_eq!(failure.reason, "timeout");
+        assert_eq!(failure.step, Some("release build"));
+        assert!(failure.elapsed_secs.unwrap() >= 11.0);
+        assert_eq!(failure.tests.passed, 7);
+        assert_eq!(failure.tests.output, "successful tests");
+        assert_eq!(failure.output.lines().count(), 50);
+        assert!(failure.output.contains("diagnostic-99"));
+        assert!(!failure.output.contains("diagnostic-0\n"));
+        let raw = std::fs::read_to_string(&logs.stdout_path).unwrap();
+        assert_eq!(raw.lines().count(), 100);
+        assert!(raw.starts_with("diagnostic-0\n"));
+    }
+}

cli/src/core.rs

@@ -62,10 +62,7 @@ impl Host {
             }
             "log" => Ok(self.0.journal.query(p).await),
             "activate" | "revert" => self.activation(caller, method, p).await,
-            "exit" => {
-                self.0.exit.send_replace(true);
-                Ok(json!({}))
-            }
+            "restart" | "exit" => self.request_restart(),
             _ => Err(error(-32601, "method not found")),
         }
     }

cli/src/host.rs

@@ -26,6 +26,8 @@ pub(crate) struct Inner {
     pub(crate) journal: Arc<journal::Journal>,
     pub(crate) notifications: tokio::sync::broadcast::Sender<Value>,
     pub(crate) exit: watch::Sender<bool>,
+    pub(crate) loop_supervised: bool,
+    pub(crate) restart_pending: std::sync::atomic::AtomicBool,
     pub(crate) components: components::Components,
     /// Calls hold it for reading from the choice of a version until their
     /// process is in the table. Activation holds it for writing while it
@@ -64,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 || {
@@ -98,6 +103,8 @@ impl Host {
             }),
             notifications: tokio::sync::broadcast::channel(1024).0,
             exit: watch::channel(false).0,
+            loop_supervised: crate::restart::enabled(std::env::var_os("SAUSAGE_LOOP").as_deref()),
+            restart_pending: std::sync::atomic::AtomicBool::new(false),
             components: components::Components::default(),
             admission: tokio::sync::RwLock::new(()),
             shutdown: std::sync::atomic::AtomicBool::new(false),
@@ -143,12 +150,12 @@ impl Host {
         )
         .await
     }
-    /// Wait until a core.exit handler requests a restart.
+    /// Wait until core.restart (or its alias core.exit) requests shutdown after its response grace.
     pub async fn exit_requested(&self) {
         let mut exit = self.0.exit.subscribe();
         let _ = exit.wait_for(|requested| *requested).await;
     }
-    /// Return whether a core.exit handler requested a restart.
+    /// Return whether a restart has reached the normal shutdown path.
     pub fn restart_requested(&self) -> bool {
         *self.0.exit.borrow()
     }

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<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() {

cli/src/lib.rs

@@ -12,6 +12,7 @@ mod journal;
 mod locks;
 mod native;
 mod process;
+mod restart;
 mod routing;
 pub mod runtime;
 mod services;

cli/src/restart.rs

@@ -0,0 +1,129 @@
+//! Supervisor-gated restart requests; the executable still owns normal shutdown.
+use crate::{Host, RpcResult, error};
+use serde_json::json;
+use std::{ffi::OsStr, sync::atomic::Ordering, time::Duration};
+
+pub(crate) fn enabled(value: Option<&OsStr>) -> bool {
+    value == Some(OsStr::new("1"))
+}
+
+impl Host {
+    pub(crate) fn request_restart(&self) -> RpcResult {
+        if !self.0.loop_supervised {
+            return Err(error(
+                -32600,
+                "restart needs the loop supervisor (SAUSAGE_LOOP not set); start with SAUSAGE_LOOP=1",
+            ));
+        }
+        if !self.0.restart_pending.swap(true, Ordering::SeqCst) {
+            let host = self.clone();
+            tokio::spawn(async move {
+                // Let the RPC response reach the caller before closing any stdin,
+                // including that of a detached gateway carrying this response.
+                tokio::time::sleep(Duration::from_millis(250)).await;
+                host.0.exit.send_replace(true);
+            });
+        }
+        Ok(json!({}))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{runtime::Runtime, world::World};
+    use std::sync::Arc;
+    use tokio::time::timeout;
+
+    #[test]
+    fn only_explicit_loop_marker_enables_restart() {
+        assert!(!enabled(None));
+        for value in ["", "0", "true", "yes", "2"] {
+            assert!(!enabled(Some(OsStr::new(value))));
+        }
+        assert!(enabled(Some(OsStr::new("1"))));
+    }
+
+    async fn host(supervised: bool) -> (tempfile::TempDir, Host) {
+        let dir = tempfile::tempdir().unwrap();
+        World::create(&dir.path().join("world")).unwrap();
+        let runtime = Runtime::new(&dir.path().join("cache")).unwrap();
+        let mut host = Host::test_world(&dir.path().join("world"), runtime)
+            .await
+            .unwrap();
+        // No global environment mutation: parallel tests have independent gates.
+        Arc::get_mut(&mut host.0).unwrap().loop_supervised = supervised;
+        (dir, host)
+    }
+
+    #[tokio::test]
+    async fn restart_without_supervisor_never_triggers_exit() {
+        timeout(Duration::from_secs(10), async {
+            let (_dir, host) = host(false).await;
+            for method in ["core.restart", "core.exit"] {
+                let error = host.call(method, json!({})).await.unwrap_err();
+                assert_eq!(error["code"], -32600);
+                assert!(
+                    error["message"]
+                        .as_str()
+                        .unwrap()
+                        .contains("SAUSAGE_LOOP not set")
+                );
+                let indirect = host
+                    .call("core.call", json!({"method":method}))
+                    .await
+                    .unwrap_err();
+                assert_eq!(indirect, error);
+            }
+            assert!(!host.0.restart_pending.load(Ordering::SeqCst));
+            assert!(
+                timeout(Duration::from_millis(300), host.exit_requested())
+                    .await
+                    .is_err()
+            );
+            host.stop().await;
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn restart_and_exit_share_response_grace_and_coalescing() {
+        timeout(Duration::from_secs(10), async {
+            for first in ["core.restart", "core.exit"] {
+                let (_dir, host) = host(true).await;
+                // Same dispatch as a plugin's core.call, without killing this runner.
+                assert_eq!(
+                    host.call("core.call", json!({"method":first}))
+                        .await
+                        .unwrap(),
+                    json!({})
+                );
+                assert!(!host.restart_requested());
+                assert!(host.0.restart_pending.load(Ordering::SeqCst));
+                assert_eq!(
+                    host.call("core.restart", json!({})).await.unwrap(),
+                    json!({})
+                );
+                timeout(Duration::from_secs(1), host.exit_requested())
+                    .await
+                    .unwrap();
+                // A duplicate after the trigger must not schedule a second task.
+                host.0.exit.send_replace(false);
+                assert_eq!(
+                    host.call("core.restart", json!({})).await.unwrap(),
+                    json!({})
+                );
+                assert_eq!(host.call("core.exit", json!({})).await.unwrap(), json!({}));
+                assert!(
+                    timeout(Duration::from_millis(300), host.exit_requested())
+                        .await
+                        .is_err()
+                );
+                host.stop().await;
+            }
+        })
+        .await
+        .unwrap();
+    }
+}

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");
+    }
+}

cli/src/test_runner.rs

@@ -64,12 +64,21 @@ async fn instance(
         ("SAUSAGE_PLUGIN".into(), "test".into()),
         ("SAUSAGE_PARENT".into(), "core".into()),
     ]);
+    let timeout_secs = std::env::var("SAUSAGE_TEST_TIMEOUT_SECS")
+        .ok()
+        .and_then(|value| value.parse::<u64>().ok())
+        .filter(|value| *value >= 1)
+        .unwrap_or(sausage::cargo::DEFAULT_TEST_TIMEOUT_SECS);
+    let deadline = tokio::time::Instant::now()
+        .checked_add(std::time::Duration::from_secs(timeout_secs))
+        .ok_or_else(|| anyhow::anyhow!("test_timeout_secs is too large"))?;
+    let (ended, mut ended_rx) = tokio::sync::watch::channel(true);
     let mut guest = runtime
         .start_with_args(
             &component,
             InstanceConfig {
                 filesystem,
-                store_ended: tokio::sync::watch::channel(true).0,
+                store_ended: ended,
                 network: true,
                 env,
             },
@@ -88,6 +97,7 @@ async fn instance(
     let mut stdout_open = true;
     let mut stderr_open = true;
     let mut handlers = tokio::task::JoinSet::new();
+    let result = tokio::time::timeout_at(deadline, async {
     while stdout_open || stderr_open || !pending.is_empty() {
         tokio::select! {
             line=async { if let Some(line) = pending.pop_front() { Some(line) } else { guest.stdout.recv().await } }, if stdout_open || !pending.is_empty()=>match line {
@@ -129,11 +139,18 @@ async fn instance(
             },
         }
     }
+    Ok::<_, anyhow::Error>(guest.wait().await)
+    }).await;
     handlers.abort_all();
-    let exit = guest.wait().await;
+    // Dropping the instance aborts even a CPU-bound WASI loop. Wait for the
+    // store and outstanding filesystem work before removing its test world.
+    drop(guest);
+    let _ = ended_rx.wait_for(|ended| *ended).await;
     let _ = stderr_task.await;
     tracker.idle().await;
     host.stop().await;
+    let exit =
+        result.map_err(|_| anyhow::anyhow!("timed out after {timeout_secs} s\n{output}"))??;
     Ok(Run { exit, output })
 }
 /// Extracts libtest names from its stable list format.

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}");
         }

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::<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"})))

cli/tests/host/cargo.rs

@@ -12,6 +12,23 @@ impl TestWorld {
         World::create(&path).unwrap();
         Self(path)
     }
+    /// Short-deadline tests must not queue on another test's Cargo target lock.
+    async fn timeout_host(&self, name: &str) -> Host {
+        Host::start_with_cache(&self.0, &Path::new(CACHE).join("timeouts").join(name))
+            .await
+            .unwrap()
+    }
+    /// Creates a dependency-free plugin fixture.
+    fn simple(&self, name: &str, text: &str) {
+        let root = self.0.join(format!("plugins/{name}/source"));
+        std::fs::create_dir_all(root.join("src")).unwrap();
+        std::fs::write(
+            root.join("Cargo.toml"),
+            format!("[package]\nname='{name}'\nversion='0.1.0'\nedition='2024'\n"),
+        )
+        .unwrap();
+        std::fs::write(root.join("src/main.rs"), text).unwrap();
+    }
     /// Copies a fixture into its plugin source folder.
     fn source(&self, name: &str, source: &Path) {
         copy(source, &self.0.join(format!("plugins/{name}/source")));
@@ -56,8 +73,28 @@ async fn build(world: &TestWorld, name: &str) -> Result<Value, Value> {
         .register(&host);
     let result = host.call("cargo.build", json!({"plugin":name})).await;
     host.stop().await;
+    let logs = match &result {
+        Ok(output) => &output["logs"],
+        Err(error) => &error["data"]["logs"],
+    };
+    check_logs(world, logs);
     result
 }
+/// RPC paths point at full raw streams, outside pruned build folders.
+fn check_logs(world: &TestWorld, logs: &Value) {
+    let mut total = 0;
+    for stream in ["stdout", "stderr"] {
+        let path = logs[stream]["path"].as_str().expect("log path");
+        assert!(path.starts_with("/logs/cargo/"));
+        assert!(path.ends_with(&format!(".{stream}.log")));
+        let bytes = std::fs::metadata(world.0.join(path.trim_start_matches('/')))
+            .unwrap()
+            .len();
+        assert_eq!(logs[stream]["bytes"], bytes);
+        total += bytes;
+    }
+    assert_eq!(logs["total_bytes"], total);
+}
 struct Valid {
     world: TestWorld,
     result: Value,
@@ -189,6 +226,20 @@ async fn build_output_and_warnings() {
             .unwrap()
             .contains("fixture build warning")
     );
+    let logs = &valid.result["logs"];
+    let raw = |stream: &str| {
+        std::fs::read_to_string(
+            valid.world.0.join(
+                logs[stream]["path"]
+                    .as_str()
+                    .unwrap()
+                    .trim_start_matches('/'),
+            ),
+        )
+        .unwrap()
+    };
+    assert!(raw("stdout").contains("\"reason\":\"compiler-artifact\""));
+    assert!(raw("stderr").contains("Finished "));
     let output = valid.result["tests"]["output"].as_str().unwrap();
     assert!(output.contains("INTEGRATION_CASE"));
     assert!(!output.contains("Compiling "));
@@ -260,6 +311,7 @@ async fn native_discovery_and_error_paths() {
         .await
         .unwrap_err();
     assert_eq!(error["code"], -32004);
+    check_logs(&world, &error["data"]["logs"]);
     assert!(
         error["data"]["output"]
             .as_str()
@@ -606,6 +658,259 @@ async fn standalone_builder() {
         1
     );
 }
+/// SAU-126: a named CPU-bound WASI test times out, while other instances still run.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn named_test_timeout_and_mutex_release() {
+    let world = TestWorld::new();
+    world.simple("test-timeout", "fn main() {} #[test] fn a_hung() { loop { std::hint::black_box(1); } } #[test] fn z_remaining() { eprintln!(\"REMAINING_TEST_RAN\"); }");
+    world.simple("next", "fn main() {} #[test] fn passes() {}");
+    let integration = world.0.join("plugins/test-timeout/source/tests");
+    std::fs::create_dir_all(&integration).unwrap();
+    std::fs::write(
+        integration.join("remaining.rs"),
+        "#[test] fn integration_remaining() { eprintln!(\"REMAINING_BINARY_RAN\"); }",
+    )
+    .unwrap();
+    let host = world.timeout_host("named").await;
+    let error = tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        host.call(
+            "cargo.build",
+            json!({"plugin":"test-timeout", "test_timeout_secs":1, "timeout_secs":45}),
+        ),
+    )
+    .await
+    .unwrap()
+    .unwrap_err();
+    assert_eq!(error["code"], -32004);
+    assert_eq!(error["message"], "WASI tests failed");
+    assert_eq!(error["data"]["tests"]["failed"], 1);
+    assert_eq!(error["data"]["tests"]["passed"], 2);
+    let output = error["data"]["output"].as_str().unwrap();
+    assert!(output.contains("test a_hung ... FAILED"), "{output}");
+    assert!(output.contains("timed out after 1 s"), "{output}");
+    assert!(output.contains("REMAINING_TEST_RAN"), "{output}");
+    assert!(output.contains("REMAINING_BINARY_RAN"), "{output}");
+    let next = tokio::time::timeout(
+        std::time::Duration::from_secs(30),
+        host.call("cargo.build", json!({"plugin":"next"})),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    assert_eq!(next["tests"]["passed"], 1);
+    host.stop().await;
+}
+
+/// Wait for a build-script child without using timing sleeps.
+async fn script_child(world: &TestWorld) -> u32 {
+    tokio::time::timeout(std::time::Duration::from_secs(60), async {
+        loop {
+            if let Ok(text) = std::fs::read_to_string(world.0.join("state/cargo/work/child.pid"))
+                && let Ok(pid) = text.trim().parse()
+            {
+                break pid;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("build script never started")
+}
+/// A killed descendant may briefly remain as a zombie, but cannot keep executing.
+async fn child_ended(child: u32) {
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        loop {
+            match std::fs::read_to_string(format!("/proc/{child}/stat")) {
+                Ok(status) if status.split_whitespace().nth(2) == Some("Z") => break,
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => break,
+                _ => tokio::task::yield_now().await,
+            }
+        }
+    })
+    .await
+    .expect("cargo descendant survived cancellation/timeout");
+}
+
+/// SAU-126: an explicit small total budget kills build.rs and unblocks a queued build.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn build_timeout_and_queued_build() {
+    let world = TestWorld::new();
+    world.source("hang", &fixture("cargo-hang"));
+    world.simple("next", "fn main() {} #[test] fn passes() {}");
+    let host = world.timeout_host("queued").await;
+    let building = host.clone();
+    let task = tokio::spawn(async move {
+        building
+            .call("cargo.build", json!({"plugin":"hang", "timeout_secs":5}))
+            .await
+    });
+    let child = script_child(&world).await;
+    let queued_host = host.clone();
+    let queued = tokio::spawn(async move {
+        queued_host
+            .call("cargo.build", json!({"plugin":"next", "timeout_secs":3}))
+            .await
+    });
+    let error = task.await.unwrap().unwrap_err();
+    assert_eq!(error["code"], -32004);
+    assert_eq!(error["message"], "timeout");
+    assert_eq!(error["data"]["reason"], "timeout");
+    check_logs(&world, &error["data"]["logs"]);
+    assert_eq!(error["data"]["step"], "tests");
+    assert!(error["data"]["elapsed_secs"].as_f64().unwrap() >= 5.0);
+    assert!(!error["data"]["output"].as_str().unwrap().is_empty());
+    child_ended(child).await;
+    let next = tokio::time::timeout(std::time::Duration::from_secs(20), queued)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(next["tests"]["passed"], 1);
+    host.stop().await;
+}
+
+/// SAU-126: the same total deadline covers release compilation and retains test results.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn release_build_timeout_retains_tests() {
+    let world = TestWorld::new();
+    world.simple(
+        "release-timeout",
+        r#"fn main() {}
+#[test] fn prepare_release() {
+    std::fs::write(concat!(env!("CARGO_MANIFEST_DIR"), "/release-marker"), "ready").unwrap();
+}"#,
+    );
+    std::fs::write(
+        world.0.join("plugins/release-timeout/source/build.rs"),
+        r#"
+fn main() {
+    println!("cargo:rerun-if-changed=release-marker");
+    if std::path::Path::new("release-marker").exists() {
+        std::process::Command::new("sh")
+            .args(["-c", "echo $$ > child.pid; exec sleep 1000"])
+            .status().unwrap();
+    }
+}"#,
+    )
+    .unwrap();
+    let host = world.timeout_host("release").await;
+    let building = host.clone();
+    let task = tokio::spawn(async move {
+        building
+            .call(
+                "cargo.build",
+                json!({"plugin":"release-timeout", "timeout_secs":5}),
+            )
+            .await
+    });
+    let child = script_child(&world).await;
+    let error = tokio::time::timeout(std::time::Duration::from_secs(10), task)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap_err();
+    assert_eq!(error["code"], -32004);
+    assert_eq!(error["data"]["reason"], "timeout");
+    check_logs(&world, &error["data"]["logs"]);
+    assert_eq!(error["data"]["step"], "release build");
+    assert_eq!(error["data"]["tests"]["passed"], 1);
+    assert!(error["data"]["elapsed_secs"].as_f64().unwrap() >= 5.0);
+    child_ended(child).await;
+    host.stop().await;
+}
+
+/// SAU-126: omission selects the real 900-second deadline, without a 15-minute test.
+#[tokio::test]
+async fn omitted_build_timeout_uses_default() {
+    let world = TestWorld::new();
+    world.source("hang", &fixture("cargo-hang"));
+    let host = world.timeout_host("default").await;
+    let building = host.clone();
+    let task =
+        tokio::spawn(async move { building.call("cargo.build", json!({"plugin":"hang"})).await });
+    let child = script_child(&world).await;
+    tokio::time::pause();
+    tokio::time::advance(std::time::Duration::from_secs(
+        sausage::cargo::DEFAULT_TIMEOUT_SECS,
+    ))
+    .await;
+    // Let the awakened deadline observe the advanced clock before resuming it.
+    tokio::task::yield_now().await;
+    let error = task.await.unwrap().unwrap_err();
+    tokio::time::resume();
+    assert_eq!(error["data"]["reason"], "timeout");
+    assert!(error["data"]["elapsed_secs"].as_f64().unwrap() >= 900.0);
+    child_ended(child).await;
+    host.stop().await;
+}
+
+/// SAU-126: a caller ending stops its attached native build and releases the mutex.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn ended_caller_releases_build_mutex() {
+    let world = TestWorld::new();
+    world.source("hang", &fixture("cargo-hang"));
+    world.simple("next", "fn main() {} #[test] fn passes() {}");
+    let host = world.timeout_host("caller").await;
+    let (finish, finished) = tokio::sync::watch::channel(false);
+    host.register_native(
+        "caller",
+        std::sync::Arc::new(move |context, _, _| {
+            let mut finished = finished.clone();
+            Box::pin(async move {
+                tokio::spawn(
+                    async move { context.call("cargo.build", json!({"plugin":"hang"})).await },
+                );
+                let _ = finished.wait_for(|done| *done).await;
+                Ok(json!({}))
+            })
+        }),
+    );
+    let calling = host.clone();
+    let caller = tokio::spawn(async move { calling.call("caller.run", json!({})).await });
+    let child = script_child(&world).await;
+    let queued_host = host.clone();
+    let queued = tokio::spawn(async move {
+        queued_host
+            .call("cargo.build", json!({"plugin":"next"}))
+            .await
+    });
+    finish.send_replace(true);
+    caller.await.unwrap().unwrap();
+    let next = tokio::time::timeout(std::time::Duration::from_secs(12), queued)
+        .await
+        .expect("ended caller retained mutex")
+        .unwrap()
+        .unwrap();
+    assert_eq!(next["tests"]["passed"], 1);
+    child_ended(child).await;
+    host.stop().await;
+}
+
+/// SAU-126: discovery exposes non-null defaults; invalid budgets cannot disable limits.
+#[tokio::test]
+async fn timeout_params_discovery_and_validation() {
+    let world = TestWorld::new();
+    let host = Host::start_with_cache(&world.0, Path::new(CACHE))
+        .await
+        .unwrap();
+    let doc = host.call("cargo.rpc.discover", json!({})).await.unwrap();
+    let params = doc["methods"][0]["params"].as_array().unwrap();
+    for (name, default) in [("timeout_secs", 900), ("test_timeout_secs", 120)] {
+        let param = params.iter().find(|param| param["name"] == name).unwrap();
+        assert_eq!(param["schema"]["default"], default);
+        assert_eq!(param["schema"]["minimum"], 1);
+        assert!(!param["description"].as_str().unwrap().is_empty());
+        for value in [json!(0), json!(-1), json!(1.5), json!("1"), json!(null)] {
+            let mut args = json!({"plugin":"missing"});
+            args[name] = value;
+            let error = host.call("cargo.build", args).await.unwrap_err();
+            assert_eq!(error["code"], -32602, "{error}");
+        }
+    }
+    host.stop().await;
+}
+
 /// PROTOCOL 11: killing a native build also kills its build-script descendants.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn kill_process_group() {
@@ -615,9 +920,7 @@ async fn kill_process_group() {
     );
     let world = TestWorld::new();
     world.source("hang", &fixture("cargo-hang"));
-    let host = Host::start_with_cache(&world.0, Path::new(CACHE))
-        .await
-        .unwrap();
+    let host = world.timeout_host("kill").await;
     let building = host.clone();
     let task =
         tokio::spawn(async move { building.call("cargo.build", json!({"plugin":"hang"})).await });
@@ -685,8 +988,23 @@ async fn kill_process_group() {
             .unwrap()
             .is_empty()
     );
+    // Queue a build on the SAME builder before cancelling the lock holder.
+    world.simple("next", "fn main() {} #[test] fn passes() {}");
+    let queued_host = host.clone();
+    let queued = tokio::spawn(async move {
+        queued_host
+            .call("cargo.build", json!({"plugin":"next"}))
+            .await
+    });
     host.call("core.kill", json!({"pid":pid})).await.unwrap();
     assert!(task.await.unwrap().is_err());
+    child_ended(ancestor).await;
+    let next = tokio::time::timeout(std::time::Duration::from_secs(10), queued)
+        .await
+        .expect("cancelled build retained mutex")
+        .unwrap()
+        .unwrap();
+    assert_eq!(next["tests"]["passed"], 1);
     tokio::time::timeout(std::time::Duration::from_secs(10), async {
         loop {
             let status = std::fs::read_to_string(format!("/proc/{child}/stat"));

cli/tests/host/cli.rs

@@ -19,6 +19,7 @@ fn binary() -> Command {
         .arg("--cache-dir")
         .arg(CACHE)
         .env_remove("SAUSAGE_HEADLESS")
+        .env_remove("SAUSAGE_LOOP")
         .kill_on_drop(true);
     command
 }
@@ -35,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;
         }
     })
@@ -54,9 +55,15 @@ fn install_probe(world: &Path) {
     std::fs::copy(probe(), world.join("plugins/probe/plugin.wasm")).unwrap();
 }
 
-/// A failed agent build leaves no world. Bare sausage retries and publishes a complete world.
+/// Failed init leaves no world; successful init seeds the restart policy and a working agent.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn init_builds_agent() {
+    tokio::time::timeout(Duration::from_secs(300), init_builds_agent_inner())
+        .await
+        .expect("init and seeded-agent checks exceeded 300 seconds");
+}
+
+async fn init_builds_agent_inner() {
     let temporary = tempfile::tempdir().unwrap();
     let world = temporary.path().join("world");
     // Legacy configuration variables must not affect the templates.
@@ -81,7 +88,7 @@ async fn init_builds_agent() {
     assert!(!failed.status.success());
     assert!(String::from_utf8_lossy(&failed.stderr).contains("nonexistent-sausage-rustc"));
     assert!(!world.exists());
-    let output = configured().output().await.unwrap();
+    let output = configured().arg("init").output().await.unwrap();
     assert!(
         output.status.success(),
         "stdout: {}\nstderr: {}",
@@ -91,6 +98,22 @@ async fn init_builds_agent() {
     assert!(
         String::from_utf8_lossy(&output.stdout).contains("first agent build takes a few minutes")
     );
+    // Bare sausage reopens the completed world instead of trying to recreate it.
+    assert!(configured().output().await.unwrap().status.success());
+    let prompt = std::fs::read_to_string(world.join("plugins/agent/source/src/prompt.md")).unwrap();
+    assert!(prompt.contains("HARD RULE: the mediator and all agents NEVER call `core.restart`"));
+    assert!(prompt.contains("without the person's explicit approval each time"));
+    assert!(prompt.contains("alias `core.exit`"));
+    for text in [
+        "`#loop`",
+        "SAUSAGE_LOOP=1",
+        "answers before",
+        "exit code 75",
+    ] {
+        assert!(prompt.contains(text), "seeded prompt missing {text}");
+    }
+    let boot = std::fs::read_to_string(world.join("config/boot.yaml")).unwrap();
+    assert!(boot.contains("SAUSAGE_LOOP=1"));
     let mut config: Value =
         serde_yaml_ng::from_slice(&std::fs::read(world.join("config/agent.yaml")).unwrap())
             .unwrap();
@@ -228,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'",
                     [],
@@ -249,27 +272,39 @@ async fn signals_work_during_boot() {
 /// Piped input supports commands, quit, and the restart exit code.
 #[tokio::test]
 async fn piped_console_and_exit_codes() {
-    for (line, expected) in [("help\n", 0), ("quit\n", 0), ("core.exit\n", 75)] {
+    for (line, supervised, expected) in [
+        ("help\n", false, 0),
+        ("quit\n", false, 0),
+        ("core.restart\n", false, 0),
+        ("core.exit\n", false, 0),
+        ("core.restart\n", true, 75),
+        ("core.exit\n", true, 75),
+    ] {
         let (_folder, world) = world();
-        let mut child = binary()
-            .arg("run")
-            .arg(world)
+        let mut command = binary();
+        command.arg("run").arg(world);
+        if supervised {
+            command.env("SAUSAGE_LOOP", "1");
+        }
+        let mut child = command
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .spawn()
             .unwrap();
-        child
-            .stdin
-            .take()
-            .unwrap()
-            .write_all(line.as_bytes())
-            .await
-            .unwrap();
+        // Supervised calls must finish via restart, not a racing console EOF.
+        let mut input = child.stdin.take().unwrap();
+        input.write_all(line.as_bytes()).await.unwrap();
+        if !supervised {
+            drop(input);
+        }
         let output = tokio::time::timeout(Duration::from_secs(15), child.wait_with_output())
             .await
             .unwrap()
             .unwrap();
         assert_eq!(output.status.code(), Some(expected));
+        if !supervised && line.starts_with("core.") {
+            assert!(String::from_utf8_lossy(&output.stdout).contains("SAUSAGE_LOOP not set"));
+        }
         if expected == 0 {
             assert!(!output.stdout.is_empty());
         }
@@ -301,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'",
@@ -328,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 {}'",
                     [],

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<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,
@@ -1014,19 +1024,6 @@ async fn stop_closes_stdin_and_waits_for_exit() {
     h.stop().await;
 }
 
-/// PROTOCOL 9: core exit requests restart from its handler.
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn core_exit_requests_restart_from_its_handler() {
-    let w = TestWorld::new();
-    let h = w.start().await;
-    assert_eq!(call(&h, "core.exit", json!({})).await, json!({}));
-    tokio::time::timeout(std::time::Duration::from_secs(1), h.exit_requested())
-        .await
-        .unwrap();
-
-    h.stop().await;
-}
-
 /// PROTOCOL 3.3: process ids are uuidv7 and versions are sha256.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn process_ids_are_uuidv7_and_versions_are_sha256() {
@@ -1081,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")
     );
@@ -1267,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(),
@@ -1288,19 +1286,6 @@ async fn journal_has_wal_and_identity_indexes() {
     h.stop().await;
 }
 
-/// PROTOCOL 9: core exit works through core call.
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn core_exit_works_through_core_call() {
-    let w = TestWorld::new();
-    let h = w.start().await;
-    assert_eq!(
-        call(&h, "core.call", json!({"method":"core.exit","params":{}})).await,
-        json!({})
-    );
-    assert!(h.restart_requested());
-    h.stop().await;
-}
-
 /// PROTOCOL 6: acquisitions from different processes keep their FIFO positions.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn lock_acquisitions_from_different_processes_are_fifo() {
@@ -1931,9 +1916,67 @@ async fn detached_processes_need_explicit_termination() {
     }
 }
 
-/// PROTOCOL 9: core.exit from boot ends the host executable with status 75.
+/// PROTOCOL 9: restart answers the detached caller, then uses normal stop and status 75.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn core_exit_sets_executable_status_75() {
+async fn core_restart_cleanly_stops_detached_caller() {
+    tokio::time::timeout(std::time::Duration::from_secs(30), async {
+        let w = TestWorld::new();
+        std::fs::write(
+            w.0.join("config/boot.yaml"),
+            br#"[{"method":"probe.restart","params":{}}]"#,
+        )
+        .unwrap();
+        let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
+            .env("SAUSAGE_LOOP", "1")
+            .arg("--headless")
+            .arg("--cache-dir")
+            .arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
+            .arg("run")
+            .arg(&w.0)
+            .stdin(std::process::Stdio::null())
+            .kill_on_drop(true)
+            .output()
+            .await
+            .unwrap();
+        assert_eq!(
+            output.status.code(),
+            Some(75),
+            "{}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert_eq!(
+            std::fs::read_to_string(w.0.join("state/restart-answer")).unwrap(),
+            "received"
+        );
+        assert_eq!(
+            std::fs::read_to_string(w.0.join("state/restart-stop")).unwrap(),
+            "persisted after EOF"
+        );
+        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'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, 1);
+        let killed: i64 = db
+            .query_row(
+                "SELECT count(*) FROM journal WHERE line = 'kill'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(killed, 0);
+    })
+    .await
+    .unwrap();
+}
+
+/// PROTOCOL 9: supervised core.exit from boot uses the restart shutdown path and code 75.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_exit_alias_sets_executable_status_75() {
     let w = TestWorld::new();
     std::fs::write(
         w.0.join("config/boot.yaml"),
@@ -1948,6 +1991,7 @@ async fn core_exit_sets_executable_status_75() {
             .arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
             .arg("run")
             .arg(&w.0)
+            .env("SAUSAGE_LOOP", "1")
             .stdin(std::process::Stdio::null())
             .kill_on_drop(true)
             .output(),
@@ -1962,3 +2006,68 @@ async fn core_exit_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;
+}

cli/tests/plugins/probe/src/main.rs

@@ -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" => {
@@ -447,6 +448,21 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
             burst(request);
             return None;
         }
+        "restart" => {
+            assert_eq!(
+                lock(&json!({"path":"/state/restart-stop"}), lines),
+                json!({"held":true})
+            );
+            assert_eq!(
+                outgoing_call(&json!({"method":"core.restart", "params":{}}), lines),
+                json!({})
+            );
+            std::fs::write("/state/restart-answer", "received").unwrap();
+            serve(request, lines);
+            std::fs::write("/state/restart-stop", "persisted after EOF").unwrap();
+            eprintln!("restart stop hook finished");
+            return None;
+        }
         "serve" => {
             serve(request, lines);
             return None;

flake.nix

@@ -80,7 +80,7 @@
         # `nix run .#loop -- <world>` runs the newest `sausage run <world>`
         # (without a folder: bare `sausage`, i.e. ./world, created if needed)
         # from the public repo, again and again: after a crash or a call of
-        # `core.exit` it refreshes the flake and starts the host again, so
+        # `core.restart` (exit code 75) it refreshes the flake and starts the host again, so
         # the world updates itself. `quit` (exit code 0) ends the loop.
         # SAUSAGE_FLAKE overrides where it fetches from (any flake URL,
         # e.g. a fork or a local checkout).
@@ -90,7 +90,7 @@
           # No folder: bare `sausage` (creates ./world first if needed).
           if [ $# -gt 0 ]; then set -- run "$@"; fi
           while true; do
-            nix run --refresh "$url#default" -- "$@"
+            SAUSAGE_LOOP=1 nix run --refresh "$url#default" -- "$@"
             code=$?
             if [ "$code" -eq 0 ]; then exit 0; fi
             echo "sausage-loop: sausage exited with $code, starting the newest version in 2 s …" >&2

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::<Value>(&s)