feat(mailbox): add trusted-host local durable journal and receipts

crates/mvm-agentd/README.md

@@ -34,9 +34,47 @@ Specialized binaries keep privilege and dependency boundaries small:
   `mvm-setpriv` crate, applies the final identity.
 
 The guest treats host replies as protocol input, while the host treats every
-guest request as untrusted. Secrets and authorization decisions remain
-host-side; the agent receives only the capabilities and values needed for the
-admitted execution.
+guest request as untrusted. Host sealing and destination signing material
+remain host-side; an opt-in mailbox reader additionally holds its own per-boot
+ACK signing key inside the guest agent, not in the workload.
+
+## Guest mailbox reader
+
+Only a signed plan granting this workload's own mailbox can enable Receive.
+The agent binds `/run/mvm/mailbox/receive.sock` when its separate boot ACK key
+was provisioned. Dev's ephemeral host helper refuses reader frames; a production
+mailbox with signed retention uses a locally anchored journal and boot-bound
+Receive authority. The local protocol
+is versioned, length-prefixed, and bounded; use `ApplicationReaderRequest` and
+`ReaderResponse` from `mvm_core::mailbox::reader_wire`. Once backed by that
+local destination, `Receive` returns a signed lease,
+content type, and payload. Process the payload **before** sending
+`Ack { lease }` on a new connection. An empty inbox and a refusal are not
+successful delivery; reading never automatically ACKs.
+
+For a guest application example, build:
+
+```sh
+cargo build -p mvm-agentd --example mailbox_reader_client --target aarch64-unknown-linux-musl
+```
+
+Run it without arguments to read without acknowledging; pass `--ack` only when
+writing and flushing its stdout counts as completed processing. The returned
+`Acknowledged { ack }` echoes the exact guest-signed operation after the local
+journal and anchor commit. This is a trusted-local confirmation, **not** a
+destination-signed, exportable ACK receipt. A lost delivery or ACK response is
+**unknown**, not success. Retry with the same lease and signed ACK identity
+while this boot's agent remains active, or reconcile against the durable
+journal; a different ACK for the same lease conflicts.
+
+`mailbox_writer_client <message>` stages into the Dev/legacy spool.
+`mailbox_writer_client --local <message>` uses the separate immutable
+`outbox-local` request for the production local-durable tier; the guest agent
+retains it until it verifies the boot-bound signed V2 receipt. A response it
+cannot verify is not acceptance. Both commands report **local staging only**,
+not a destination receipt; verify the host-side receipt or journal before
+claiming destination acceptance. Do not use the legacy spool for a production
+local-durable claim.
 
 ## Bounded entrypoint capture
 

crates/mvm-agentd/examples/mailbox_reader_client.rs

@@ -0,0 +1,70 @@
+//! Guest application example: read one message from this workload's own inbox.
+//! Pass `--ack` only after the application has processed and flushed its output.
+
+use std::io::{self, Write as _};
+use std::os::unix::net::UnixStream;
+use std::time::Duration;
+
+use mvm_core::mailbox::reader_wire::{
+    ApplicationReaderRequest, GUEST_READER_SOCKET_PATH, ReaderResponse, read_response,
+    write_application_request,
+};
+
+fn connect() -> io::Result<UnixStream> {
+    let stream = UnixStream::connect(GUEST_READER_SOCKET_PATH)?;
+    stream.set_read_timeout(Some(Duration::from_secs(12)))?;
+    stream.set_write_timeout(Some(Duration::from_secs(12)))?;
+    Ok(stream)
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let ack = match std::env::args().skip(1).collect::<Vec<_>>().as_slice() {
+        [] => false,
+        [flag] if flag == "--ack" => true,
+        _ => return Err("usage: mailbox_reader_client [--ack]".into()),
+    };
+    let mut stream = connect()?;
+    write_application_request(&mut stream, &ApplicationReaderRequest::Receive)?;
+    let response = read_response(&mut stream)?;
+    let ReaderResponse::Delivery {
+        lease,
+        content_type,
+        payload,
+    } = response
+    else {
+        match response {
+            ReaderResponse::Empty => eprintln!("inbox empty"),
+            ReaderResponse::Refused { reason } => return Err(reason.into()),
+            _ => return Err("unexpected reader response".into()),
+        }
+        return Ok(());
+    };
+    // This example's application processing is writing and flushing stdout.
+    // A real application ACKs only after its own durable processing completes.
+    io::stdout().write_all(&payload)?;
+    io::stdout().flush()?;
+    eprintln!(
+        "received {} bytes as {} (message {:?}, delivery {:?})",
+        payload.len(),
+        content_type.name(),
+        lease.lease.message_id,
+        lease.lease.delivery_id
+    );
+    if ack {
+        let mut stream = connect()?;
+        write_application_request(
+            &mut stream,
+            &ApplicationReaderRequest::Ack {
+                lease: lease.clone(),
+            },
+        )?;
+        match read_response(&mut stream)? {
+            ReaderResponse::Acknowledged { ack: signed } if signed.ack.lease == *lease => {
+                eprintln!("locally acknowledged by MVM");
+            }
+            ReaderResponse::Refused { reason } => return Err(reason.into()),
+            _ => return Err("ACK response did not identify this lease".into()),
+        }
+    }
+    Ok(())
+}

crates/mvm-agentd/examples/mailbox_writer_client.rs

@@ -0,0 +1,46 @@
+//! Stage one guest-to-own-mailbox message. Local staging is not a signed
+//! destination acceptance receipt; inspect the host journal for durability.
+
+use std::path::Path;
+
+use mvm_agentd::mailbox_outbox::{LOCAL_OUTBOX_PATH, OUTBOX_PATH, write_outbox};
+use mvm_contract::protocol::mailbox::PrincipalId;
+use mvm_core::mailbox::frame::{OWN_MAILBOX_RECIPIENT, OutboxMessage};
+
+fn destination(args: &[String]) -> Option<(&'static str, &str)> {
+    match args {
+        [body] if !body.is_empty() && body != "--local" => Some((OUTBOX_PATH, body)),
+        [flag, body] if flag == "--local" && !body.is_empty() => Some((LOCAL_OUTBOX_PATH, body)),
+        _ => None,
+    }
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    let (outbox, body) =
+        destination(&args).ok_or("usage: mailbox_writer_client [--local] <message>")?;
+    let message = OutboxMessage::new(
+        PrincipalId::from_bytes(OWN_MAILBOX_RECIPIENT),
+        body.as_bytes().to_vec(),
+    );
+    write_outbox(Path::new(outbox), &message)?;
+    eprintln!("staged locally: message {:?}", message.message_id);
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_local_option_chooses_the_receipt_bearing_outbox_only_when_explicit() {
+        assert_eq!(destination(&["hello".into()]), Some((OUTBOX_PATH, "hello")));
+        assert_eq!(
+            destination(&["--local".into(), "hello".into()]),
+            Some((LOCAL_OUTBOX_PATH, "hello"))
+        );
+        assert_eq!(destination(&["--local".into()]), None);
+        assert_eq!(destination(&["".into()]), None);
+        assert_eq!(destination(&["one".into(), "two".into()]), None);
+    }
+}

crates/mvm-agentd/src/bin/mvm-guest-agent.rs

@@ -34,6 +34,8 @@ mod init;
 mod interactive;
 #[path = "mvm-guest-agent/mailbox.rs"]
 mod mailbox;
+#[path = "mvm-guest-agent/mailbox_reader.rs"]
+mod mailbox_reader;
 #[path = "mvm-guest-agent/monitoring.rs"]
 mod monitoring;
 #[path = "mvm-guest-agent/port_forward.rs"]
@@ -221,7 +223,7 @@ impl ColdContinuation {
 }
 
 fn complete_cold_response(
-    file: &mut std::fs::File,
+    file: &mut impl Write,
     session: &mut AuthenticatedSession,
     next: Option<ColdContinuation>,
     response: &GuestResponse,
@@ -1656,6 +1658,17 @@ mod tests {
 
     #[test]
     fn cold_ack_write_failure_cannot_enable_continuation() {
+        struct RejectWrite;
+        impl Write for RejectWrite {
+            fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
+                Err(std::io::ErrorKind::BrokenPipe.into())
+            }
+
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+
         let (mut host, mut guest) = UnixStream::pair().unwrap();
         host.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
         guest
@@ -1672,18 +1685,19 @@ mod tests {
         let (mut session, _) =
             AuthenticatedSession::host_retaining_peer(&mut host, "ack-failure", host_key).unwrap();
         peer.join().unwrap();
-        let descriptor: std::os::fd::OwnedFd = host.into();
-        let mut file = std::fs::File::from(descriptor);
         // This tests the real protected write seam, not activation or native
         // ownership. A candidate ACK alone cannot create a continuation.
+        let Err(error) = complete_cold_response(
+            &mut RejectWrite,
+            &mut session,
+            Some(ColdContinuation::Exec),
+            &GuestResponse::ActivateEnvironmentAck,
+        ) else {
+            panic!("the injected writer must fail before continuation");
+        };
         assert!(
-            complete_cold_response(
-                &mut file,
-                &mut session,
-                Some(ColdContinuation::Exec),
-                &GuestResponse::ActivateEnvironmentAck,
-            )
-            .is_err()
+            format!("{error:#}").contains("control frame write failed"),
+            "the injected writer must fail before continuation"
         );
     }
 

crates/mvm-agentd/src/bin/mvm-guest-agent/init.rs

@@ -184,7 +184,9 @@ pub(crate) fn apply_activation(
     // The telemetry signing key is the egress client's, mode 0400; read it now
     // or the listener can never answer a session.
     crate::telemetry::capture_session_keys_before_privilege_drop();
+    crate::mailbox::capture_ack_key_before_privilege_drop();
     guest_mount::drop_guest_agent_privilege(guest_mount::WORKLOAD_UID, guest_mount::WORKLOAD_GID)?;
+    crate::mailbox::protect_ack_key_after_privilege_drop()?;
 
     boot_state.set_activation(ActivationState::Activated);
     Ok(())