feat(cli): restore snapshots into bounded read-only mirrors

crates/buzz-acp/src/sandbox_terminal/server.rs

@@ -24,6 +24,7 @@ const IDLE: Duration = Duration::from_secs(300);
 const DRAIN: Duration = Duration::from_secs(6);
 const SUBSCRIPTION: &str = "sandbox-terminal";
 
+#[derive(Clone)]
 pub(super) struct Config {
     pub(super) relay: String,
     pub(super) keys: Keys,
@@ -110,6 +111,13 @@ fn elapsed(start: Instant) -> u64 {
     u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX)
 }
 
+fn frame_admitted(event: &Event, budget: &mut Rate, ms: u64) -> bool {
+    // NIP44 padding/base64 can exceed twice the decoded packet budget. Use
+    // the shared codec's bounded envelope before any signature/decryption;
+    // Packet::bytes separately enforces the 16KiB decoded data limit.
+    buzz_core::observer::content_looks_like_nip44(&event.content) && budget.allow(ms)
+}
+
 fn authority_deadline(grant: &Grant, checked_at: Instant) -> Instant {
     let expiry = std::time::UNIX_EPOCH + Duration::from_secs(grant.expires_at());
     let remaining = expiry
@@ -223,7 +231,7 @@ async fn open(
     // Bound ciphertext before decryption. The second typed decryption binds
     // the discovered approval ID to the complete signed envelope.
     ensure!(
-        event.content.len() <= 32 * 1024 && event.verify().is_ok(),
+        buzz_core::observer::content_looks_like_nip44(&event.content) && event.verify().is_ok(),
         "invalid terminal open"
     );
     let discovered: Frame = buzz_core::observer::decrypt_observer_payload(&config.keys, event)?;
@@ -421,7 +429,7 @@ async fn run(config: &Config, rest: &RestClient, mut shutdown: watch::Receiver<(
                         RelayMessage::Closed { subscription_id, .. } if subscription_id == SUBSCRIPTION => bail!("terminal subscription closed"),
                         RelayMessage::Event { subscription_id, event } if subscription_id == SUBSCRIPTION => {
                             let ms = elapsed(start);
-                            if event.content.len() > 32 * 1024 || !incoming.allow(ms) { continue; }
+                            if !frame_admitted(&event, &mut incoming, ms) { continue; }
                             if let Some(session) = &mut active {
                                 if event.pubkey != session.grant.requester() { continue; }
                                 let Ok(frame) = attach::decrypt(&config.keys, event.pubkey, config.scope, session.grant.id(), &event, Timestamp::now().as_secs()) else { continue; };

crates/buzz-acp/src/sandbox_terminal/server/tests.rs

@@ -134,7 +134,7 @@ impl Fixture {
         std::fs::write(
             &executable,
             r#"#!/usr/bin/env python3
-import os, sys, signal
+import os, sys, signal, json
 os.chdir(sys.argv[2])
 with open('marker', 'w') as marker: marker.write('spawned')
 def end(*_):
@@ -143,7 +143,10 @@ def end(*_):
 signal.signal(signal.SIGTERM, end)
 sys.stdin.readline()
 print('{"type":"ready"}', flush=True)
-for _ in sys.stdin: pass
+for line in sys.stdin:
+    packet = json.loads(line)
+    if packet.get('type') == 'data':
+        print(json.dumps({'type':'packet', 'packet':packet}), flush=True)
 end()
 "#,
         )
@@ -541,3 +544,180 @@ async fn harness_shutdown_propagates_unconfirmed_containment_and_task_panics() {
         tokio::spawn(async { panic!("deliberate terminal task panic") });
     assert!(crate::sandbox_terminal::stop(Some(panic)).await.is_err());
 }
+
+struct EngineRelay {
+    url: String,
+    inject: tokio::sync::mpsc::Sender<Event>,
+    events: tokio::sync::mpsc::Receiver<Event>,
+    ready: tokio::sync::oneshot::Receiver<()>,
+    task: tokio::task::JoinHandle<()>,
+}
+
+async fn engine_relay(agent: PublicKey) -> EngineRelay {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let url = format!("ws://{}", listener.local_addr().unwrap());
+    let (inject, mut injected) = tokio::sync::mpsc::channel::<Event>(16);
+    let (events_tx, events) = tokio::sync::mpsc::channel(16);
+    let (ready_tx, ready) = tokio::sync::oneshot::channel();
+    let task = tokio::spawn(async move {
+        let (socket, _) = listener.accept().await.unwrap();
+        let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap();
+        ws.send(Message::Text(
+            json!(["AUTH", "terminal-engine-fixture"])
+                .to_string()
+                .into(),
+        ))
+        .await
+        .unwrap();
+        let mut ready = Some(ready_tx);
+        loop {
+            tokio::select! {
+                message=ws.next()=>{
+                    let Some(Ok(Message::Text(text)))=message else {break};
+                    let value:Value=serde_json::from_str(&text).unwrap();
+                    match value[0].as_str() {
+                        Some("AUTH")=>{
+                            let event:Event=serde_json::from_value(value[1].clone()).unwrap();
+                            assert_eq!(event.pubkey,agent);
+                            event.verify().unwrap();
+                            ws.send(Message::Text(json!(["OK",event.id,true,""]).to_string().into())).await.unwrap();
+                        }
+                        Some("REQ")=>{if let Some(ready)=ready.take(){let _=ready.send(());}},
+                        Some("EVENT")=>{
+                            let event:Event=serde_json::from_value(value[1].clone()).unwrap();
+                            event.verify().unwrap();
+                            if event.kind.as_u16()==30624 {
+                                let body:Value=serde_json::from_str(&event.content).unwrap();
+                                let receipt=if body["status"]=="accepted" {attach::CLAIM_OK_MESSAGE}else{"saved"};
+                                ws.send(Message::Text(json!(["OK",event.id,true,receipt]).to_string().into())).await.unwrap();
+                            } else {events_tx.send(event).await.unwrap();}
+                        }
+                        _=>{},
+                    }
+                }
+                Some(event)=injected.recv()=>{
+                    ws.send(Message::Text(json!(["EVENT",SUBSCRIPTION,event]).to_string().into())).await.unwrap();
+                }
+            }
+        }
+    });
+    EngineRelay {
+        url,
+        inject,
+        events,
+        ready,
+        task,
+    }
+}
+
+#[tokio::test]
+async fn production_engine_delivers_a_full_sdk_data_packet_through_worker_ipc() {
+    let f = Fixture::new(false).await;
+    let mut relay = engine_relay(f.config.scope.agent).await;
+    let mut config = f.config.clone();
+    config.relay = relay.url.clone();
+    let rest = f.rest.clone();
+    let (shutdown_tx, shutdown) = watch::channel(());
+    let service = tokio::spawn(serve(config, rest, shutdown));
+    let bytes = vec![0xff; attach::MAX_DATA_BYTES];
+    let journey = timeout(Duration::from_secs(5), async {
+        (&mut relay.ready).await.unwrap();
+        relay.inject.send(f.open_event()).await.unwrap();
+        loop {
+            let event = relay.events.recv().await.unwrap();
+            let frame = attach::decrypt(
+                &f.requester,
+                f.config.scope.agent,
+                f.config.scope,
+                f.approval.id,
+                &event,
+                Timestamp::now().as_secs(),
+            )
+            .unwrap();
+            if frame.ack >= 1 {
+                break;
+            }
+        }
+        let data = attach::encrypt(
+            &f.requester,
+            f.config.scope.agent,
+            f.config.scope,
+            &Frame {
+                grant: f.approval.id,
+                seq: 2,
+                ack: 0,
+                packet: Packet::data(&bytes).unwrap(),
+            },
+        )
+        .unwrap();
+        assert!(
+            data.content.len() > 32 * 1024,
+            "fixture did not cross the old encrypted-content guard"
+        );
+        relay.inject.send(data).await.unwrap();
+        loop {
+            let event = relay.events.recv().await.unwrap();
+            let frame = attach::decrypt(
+                &f.requester,
+                f.config.scope.agent,
+                f.config.scope,
+                f.approval.id,
+                &event,
+                Timestamp::now().as_secs(),
+            )
+            .unwrap();
+            if matches!(frame.packet, Packet::Data { .. }) {
+                assert_eq!(frame.packet.bytes().unwrap(), bytes);
+                break;
+            }
+        }
+    })
+    .await;
+    // Always confirm cleanup before asserting, including with the old guard
+    // restored by a mutation that deliberately prevents the packet's delivery.
+    shutdown_tx.send(()).unwrap();
+    let stopped = timeout(Duration::from_secs(5), service).await;
+    relay.task.abort();
+    assert!(stopped.unwrap().unwrap().is_ok());
+    assert!(
+        journey.is_ok(),
+        "maximum SDK packet never reached/returned from the worker"
+    );
+    let record = f.journal().records().unwrap().remove(0);
+    assert_eq!(record.bytes_in, attach::MAX_DATA_BYTES as u64);
+    assert_eq!(record.bytes_out, attach::MAX_DATA_BYTES as u64);
+}
+
+#[test]
+fn production_admission_rejects_oversize_before_crypto_without_spending_rate_budget() {
+    let (owner, agent) = (Keys::generate(), Keys::generate());
+    let scope = Scope {
+        channel: Uuid::new_v4(),
+        thread: EventId::all_zeros(),
+        agent: agent.public_key(),
+    };
+    let event = attach::encrypt(
+        &owner,
+        scope.agent,
+        scope,
+        &Frame {
+            grant: EventId::all_zeros(),
+            seq: u64::MAX,
+            ack: u64::MAX,
+            packet: Packet::data(&vec![0xff; attach::MAX_DATA_BYTES]).unwrap(),
+        },
+    )
+    .unwrap();
+    assert!(event.content.len() > 32 * 1024);
+    assert!(event.content.len() <= buzz_core::observer::NIP44_MAX_CONTENT_LEN);
+    let mut oversized = event.clone();
+    oversized.content = "z".repeat(buzz_core::observer::NIP44_MAX_CONTENT_LEN + 1);
+    let mut budget = Rate::default();
+    // This deliberately has an invalid signature: the production gate only
+    // inspects size/admission and cannot invoke signature/decryption work.
+    assert!(!frame_admitted(&oversized, &mut budget, 0));
+    for _ in 0..attach::MAX_FRAMES_PER_SECOND {
+        assert!(frame_admitted(&event, &mut budget, 0));
+    }
+    assert!(!frame_admitted(&event, &mut budget, 0));
+}

crates/buzz-cli/Cargo.toml

@@ -86,14 +86,18 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std"
 tar = "0.4.46"
 flate2 = "1"
 tokio-util = { workspace = true }
+tempfile = "3"
 
 # Random number generation — full jitter for exponential backoff in with_retry
 rand = { workspace = true }
 
 [target.'cfg(unix)'.dependencies]
 # Safe local TTY handling; remote PTYs remain subordinate to buzz-acp.
 crossterm = { version = "0.29", default-features = false }
-nix = { version = "0.31", default-features = false, features = ["fs"] }
+nix = { version = "0.31", default-features = false, features = ["fs", "process", "signal"] }
+
+[target.'cfg(any(target_os = "linux", target_os = "macos"))'.dependencies]
+rustix = { version = "1", features = ["fs", "process"] }
 
 [dev-dependencies]
 # Scratch files for channel-templates.json fixtures in tests

crates/buzz-cli/src/sandbox_archive.rs

@@ -11,14 +11,25 @@ use buzz_sdk::sandbox_attach::snapshot::{
 use serde::{Deserialize, Serialize};
 
 mod format;
+mod restore;
 pub use format::{inspect_archive, ArchivePlan};
+pub use restore::{check_git, extract, RestoreBudget, Restored};
 
 /// Local archive refusal; file contents and scanner diagnostics are never included.
 #[derive(Debug, thiserror::Error)]
 pub enum ArchiveError {
     /// Cancellation completed after staging and child cleanup.
     #[error("snapshot extraction cancelled")]
     Cancelled,
+    /// The platform cannot guarantee atomic publication.
+    #[error("snapshot extraction requires Linux or macOS")]
+    Unsupported,
+    /// A bounded Git operation failed or Git is unavailable/denied.
+    #[error("snapshot Git operation failed: {0}")]
+    Git(&'static str),
+    /// Cleanup failed and its caller must retain this failure.
+    #[error("snapshot staging cleanup failed: {0}")]
+    Cleanup(std::io::Error),
     /// Filesystem access failed.
     #[error("archive file access failed: {0}")]
     Io(#[from] std::io::Error),

crates/buzz-cli/src/sandbox_archive/format.rs

@@ -125,6 +125,8 @@ pub(super) struct Item {
     pub path: String,
     pub size: u64,
     pub kind: Kind,
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    pub mode: u32,
 }
 /// Validated snapshot structure; parsing performs no filesystem writes or Git execution.
 /// Consumers must still validate imported Git objects and enforce local capacity limits.
@@ -277,7 +279,13 @@ where
         if !fields.is_empty() {
             return Err(invalid("PAX link metadata on non-link"));
         }
-        let item = Item { path, size, kind };
+        let item = Item {
+            path,
+            size,
+            kind,
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
+            mode: entry.header().mode()?,
+        };
         visit(item, &mut entry)?;
         std::io::copy(&mut entry, &mut std::io::sink())?;
     }
@@ -616,4 +624,4 @@ impl ArchivePlan {
     }
 }
 #[cfg(test)]
-mod tests;
+pub(in crate::sandbox_archive) mod tests;

crates/buzz-cli/src/sandbox_archive/format/tests.rs

@@ -3,7 +3,7 @@ use flate2::{write::GzEncoder, Compression};
 use serde_json::{json, Value};
 use std::{fs, io::Write, process::Command};
 
-fn git(repo: &Path, args: &[&str]) -> Vec<u8> {
+pub(crate) fn git(repo: &Path, args: &[&str]) -> Vec<u8> {
     let output = Command::new("git")
         .arg("-C")
         .arg(repo)
@@ -23,7 +23,7 @@ fn git(repo: &Path, args: &[&str]) -> Vec<u8> {
     );
     output.stdout
 }
-fn fixture(sha256: bool) -> (tempfile::TempDir, Value, Vec<u8>) {
+pub(crate) fn fixture(sha256: bool) -> (tempfile::TempDir, Value, Vec<u8>) {
     let temp = tempfile::tempdir().unwrap();
     let repo = temp.path();
     git(
@@ -65,7 +65,7 @@ fn fixture(sha256: bool) -> (tempfile::TempDir, Value, Vec<u8>) {
     let bytes = fs::read(bundle).unwrap();
     (temp, manifest, bytes)
 }
-fn header(path: &str, kind: u8, size: u64) -> tar::Header {
+pub(crate) fn header(path: &str, kind: u8, size: u64) -> tar::Header {
     let mut h = tar::Header::new_ustar();
     h.set_size(size);
     h.set_mode(0o644);
@@ -75,15 +75,15 @@ fn header(path: &str, kind: u8, size: u64) -> tar::Header {
     h.set_cksum();
     h
 }
-fn packed(entries: Vec<(tar::Header, Vec<u8>)>) -> Vec<u8> {
+pub(crate) fn packed(entries: Vec<(tar::Header, Vec<u8>)>) -> Vec<u8> {
     let gzip = GzEncoder::new(Vec::new(), Compression::fast());
     let mut tar = tar::Builder::new(gzip);
     for (h, data) in entries {
         tar.append(&h, data.as_slice()).unwrap();
     }
     tar.into_inner().unwrap().finish().unwrap()
 }
-fn entries(manifest: &Value, bundle: &[u8]) -> Vec<(tar::Header, Vec<u8>)> {
+pub(crate) fn entries(manifest: &Value, bundle: &[u8]) -> Vec<(tar::Header, Vec<u8>)> {
     let data = serde_json::to_vec(manifest).unwrap();
     vec![
         (