feat(acp): serve owner-approved sandbox terminals over the relay

crates/buzz-acp/Cargo.toml

@@ -86,7 +86,7 @@ nix = { version = "0.31", default-features = false, features = ["signal", "user"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 pty-process = { version = "=0.5.3", features = ["async"] }
-rustix = { version = "1.1.4", features = ["process"] }
+rustix = { version = "1.1.4", features = ["process", "fs"] }
 
 [dev-dependencies]
 axum = { workspace = true }

crates/buzz-acp/src/lib.rs

@@ -3411,6 +3411,9 @@ async fn run_harness(
     //      `IN_FLIGHT_DEADLINE_SECS` expires.
     let (steer_ack_tx, mut steer_ack_rx) = mpsc::unbounded_channel::<SteerAckEvent>();
 
+    let mut terminal_task =
+        sandbox_terminal::spawn(config, relay.rest_client(), shutdown_rx.clone())?;
+
     // Startup is complete; the main loop now owns graceful shutdown.
     let _ = startup_ready.send(());
 
@@ -3616,6 +3619,25 @@ async fn run_harness(
                     tracing::info!("shutting down");
                     break;
                 }
+                result = async {
+                    match terminal_task.as_mut() {
+                        Some(task) => task.await,
+                        None => std::future::pending().await,
+                    }
+                }, if terminal_task.is_some() => {
+                    terminal_task = None;
+                    match result {
+                        Ok(Ok(())) => {},
+                        Ok(Err(error)) if !sandbox_terminal::containment_failure(&error) => {
+                            tracing::warn!("terminal service unavailable; restart the sandbox to retry");
+                        }
+                        Ok(Err(error)) => return Err(error),
+                        Err(_) => {
+                            anyhow::bail!("terminal task failed without confirming containment");
+                        }
+                    }
+                    None
+                }
                 wake = recovery_wake::wait(&mut respawn_rx, retry_at, maintenance_at) => {
                     Some(PoolEvent::Recovery(wake))
                 }
@@ -4595,6 +4617,9 @@ async fn run_harness(
     // just as promptly. Timeout is a backstop for a slot stuck outside the
     // select (e.g. in spawn); only then do we fall back to aborting.
     let _ = shutdown_tx.send(());
+    // PTY teardown precedes checkpoint scans. A failed containment boundary is
+    // fatal even when ordinary model work could otherwise continue.
+    sandbox_terminal::stop(terminal_task).await?;
     let wake_drain = tokio::time::timeout(Duration::from_secs(30), async {
         while wake_tasks.join_next().await.is_some() {}
     })

crates/buzz-acp/src/sandbox_terminal.rs

@@ -1,11 +1,73 @@
 //! Local terminal process boundary. Relay authorization is applied by the
 //! harness before this worker is started. The worker uses local IPC only.
 
+#[cfg(target_os = "linux")]
+mod authority;
+#[cfg(target_os = "linux")]
+mod client;
+#[cfg(target_os = "linux")]
+mod journal;
 #[cfg(target_os = "linux")]
 mod process_tree;
 #[cfg(target_os = "linux")]
+mod server;
+#[cfg(target_os = "linux")]
 mod worker;
 
+/// Start terminal service only for a Linux owner-gated, immutable channel thread.
+/// Unsupported transports and scopes never subscribe to terminal requests.
+pub(crate) fn spawn(
+    config: &crate::config::Config,
+    rest: crate::relay::RestClient,
+    shutdown: tokio::sync::watch::Receiver<()>,
+) -> anyhow::Result<Option<tokio::task::JoinHandle<anyhow::Result<()>>>> {
+    #[cfg(target_os = "linux")]
+    {
+        Ok(server::Config::from_harness(config)?
+            .map(|config| tokio::spawn(server::serve(config, rest, shutdown))))
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        let _ = (config, rest, shutdown);
+        Ok(None)
+    }
+}
+
+/// Only loss of process containment requires stopping the whole harness.
+pub(crate) fn containment_failure(error: &anyhow::Error) -> bool {
+    #[cfg(target_os = "linux")]
+    {
+        error.is::<client::ContainmentError>()
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        let _ = error;
+        false
+    }
+}
+
+/// Shutdown cannot abandon a worker while the harness checkpoints its workspace.
+pub(crate) async fn stop(
+    task: Option<tokio::task::JoinHandle<anyhow::Result<()>>>,
+) -> anyhow::Result<()> {
+    let Some(mut task) = task else { return Ok(()) };
+    // The maximum admission (15s), confirmed worker cleanup (15s), final audit
+    // attempt (5s), and socket close (1s) fit inside this absolute bound.
+    match tokio::time::timeout(std::time::Duration::from_secs(40), &mut task).await {
+        Ok(Ok(Ok(()))) => Ok(()),
+        Ok(Ok(Err(error))) if !containment_failure(&error) => {
+            tracing::warn!("terminal service stopped with its audit journal retained");
+            Ok(())
+        }
+        Ok(Ok(Err(error))) => Err(error),
+        Ok(Err(_)) => anyhow::bail!("terminal service exited without confirming containment"),
+        Err(_) => {
+            task.abort();
+            anyhow::bail!("terminal service shutdown could not confirm containment")
+        }
+    }
+}
+
 /// Dispatch the isolated worker before ordinary harness configuration or Tokio
 /// threads exist. Exit 2 means containment could not be confirmed: the caller
 /// must terminate the harness instead of leaving an uncontained sandbox alive.

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

@@ -0,0 +1,190 @@
+//! Current owner approval and independent channel/delegation authority.
+
+use anyhow::{ensure, Context, Result};
+use buzz_sdk::sandbox_attach::{Authority, Grant, Pending, Scope};
+use nostr::{Event, EventId, PublicKey, Timestamp};
+use serde_json::{json, Value};
+
+use crate::relay::RestClient;
+
+pub(super) const AUTH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
+pub(super) struct Snapshot {
+    owner: PublicKey,
+    relay: PublicKey,
+    roster: Event,
+    delegations: Vec<Event>,
+    profiles: Vec<Event>,
+}
+
+impl Snapshot {
+    fn authority(&self) -> Authority<'_> {
+        Authority {
+            owner: self.owner,
+            relay: self.relay,
+            roster: &self.roster,
+            delegations: &self.delegations,
+            profiles: &self.profiles,
+        }
+    }
+
+    pub(super) fn authors(&self, scope: Scope) -> Vec<PublicKey> {
+        let mut authors = vec![self.owner];
+        authors.extend(
+            buzz_sdk::gateway::resolve(
+                self.owner,
+                &self.delegations,
+                &self.profiles,
+                &[self.owner, scope.agent],
+                Timestamp::now().as_secs(),
+            )
+            .into_iter()
+            .filter(|live| {
+                buzz_sdk::gateway::channel_member(
+                    &self.roster,
+                    self.relay,
+                    scope.channel,
+                    live.gateway(),
+                )
+            })
+            .map(|live| live.gateway()),
+        );
+        authors
+    }
+}
+
+async fn query(rest: &RestClient, mut filter: Value, limit: usize) -> Result<Vec<Event>> {
+    filter["consistency"] = json!("strong");
+    let events: Vec<Event> = serde_json::from_value(rest.query_raw(&[filter]).await?)?;
+    ensure!(
+        events.len() <= limit,
+        "terminal authority response exceeds bound"
+    );
+    Ok(events)
+}
+
+async fn read(rest: &RestClient, owner: PublicKey, scope: Scope) -> Result<Snapshot> {
+    let relay = rest
+        .relay_self()
+        .await?
+        .and_then(|value| PublicKey::from_hex(&value).ok())
+        .context("terminal relay identity unavailable")?;
+    let delegations = query(
+        rest,
+        json!({"kinds":[30181], "authors":[owner],
+        "limit":buzz_sdk::gateway::LOOKUP_LIMIT}),
+        buzz_sdk::gateway::LOOKUP_LIMIT,
+    )
+    .await?;
+    let candidates: Vec<_> =
+        buzz_sdk::gateway::live_heads(owner, &delegations, Timestamp::now().as_secs())
+            .into_iter()
+            .map(|live| live.gateway())
+            .collect();
+    let profiles = if candidates.is_empty() {
+        Vec::new()
+    } else {
+        query(
+            rest,
+            json!({"kinds":[0], "authors":candidates, "limit":candidates.len()}),
+            candidates.len(),
+        )
+        .await?
+    };
+    let mut rosters = query(
+        rest,
+        json!({"kinds":[39002], "authors":[relay], "#d":[scope.channel], "limit":1}),
+        1,
+    )
+    .await?;
+    let roster = rosters
+        .pop()
+        .context("terminal channel membership unavailable")?;
+    ensure!(
+        [owner, scope.agent]
+            .into_iter()
+            .all(|key| buzz_sdk::gateway::channel_member(&roster, relay, scope.channel, key)),
+        "terminal owner or agent no longer has posting membership"
+    );
+    Ok(Snapshot {
+        owner,
+        relay,
+        roster,
+        delegations,
+        profiles,
+    })
+}
+
+pub(super) async fn snapshot(
+    rest: &RestClient,
+    owner: PublicKey,
+    scope: Scope,
+) -> Result<Snapshot> {
+    tokio::time::timeout(AUTH_TIMEOUT, read(rest, owner, scope)).await?
+}
+
+async fn current(rest: &RestClient, snapshot: &Snapshot, grant: &Grant) -> Result<()> {
+    let coordinate = |author, phase| {
+        json!({"kinds":[30624], "authors":[author],
+        "#d":[format!("terminal:{}:{phase}", grant.session())], "limit":1})
+    };
+    let (approvals, pending) = tokio::try_join!(
+        query(rest, coordinate(snapshot.owner, "grant"), 1),
+        query(rest, coordinate(grant.requester(), "request"), 1),
+    )?;
+    ensure!(
+        approvals
+            .as_slice()
+            .first()
+            .is_some_and(|event| event.id == grant.id() && event.verify().is_ok()),
+        "terminal approval was revoked or replaced"
+    );
+    let pending = Pending::parse(
+        pending.first().context("terminal request was revoked")?,
+        Timestamp::now().as_secs(),
+    )?;
+    ensure!(
+        grant.authorized(&pending, snapshot.authority(), Timestamp::now().as_secs()),
+        "terminal approval is not currently authorized"
+    );
+    Ok(())
+}
+
+pub(super) async fn approve(
+    rest: &RestClient,
+    owner: PublicKey,
+    scope: Scope,
+    id: EventId,
+    requester: PublicKey,
+) -> Result<Grant> {
+    tokio::time::timeout(AUTH_TIMEOUT, async {
+        let snapshot = read(rest, owner, scope).await?;
+        let events = query(rest, json!({"kinds":[30624], "ids":[id], "limit":1}), 1).await?;
+        let event = events.first().context("terminal approval unavailable")?;
+        ensure!(
+            event.id == id && event.pubkey == owner,
+            "terminal approval signer mismatch"
+        );
+        let grant = Grant::parse(event, scope, Timestamp::now().as_secs())?;
+        ensure!(
+            grant.requester() == requester,
+            "terminal requester mismatch"
+        );
+        current(rest, &snapshot, &grant).await?;
+        Ok(grant)
+    })
+    .await?
+}
+
+pub(super) async fn refresh(
+    rest: &RestClient,
+    owner: PublicKey,
+    grant: &Grant,
+) -> Result<Snapshot> {
+    tokio::time::timeout(AUTH_TIMEOUT, async {
+        let snapshot = read(rest, owner, grant.scope()).await?;
+        current(rest, &snapshot, grant).await?;
+        Ok(snapshot)
+    })
+    .await?
+}