twitchyliquid64 · Jul 21, 2026 · coderabbitai · Jul 21, 2026
@@ -234,6 +234,53 @@ impl Client {
         }
         Ok(())
     }
+
+    /// Stream a zstd-compressed tarball of `dir` to the daemon's
+    /// `HomedirFilesTarZst` subsystem, which unpacks it into the
+    /// session's home directory.
+    ///
+    /// `session_id` is set as the `MINIMAL_SESSION_ID` env var on the
+    /// channel so the daemon can scope the upload to the correct session.
+    pub async fn upload_home_files(
+        &mut self,
+        session_id: sessions::SessionId,
+        dir: &Path,
+    ) -> Result<(), anyhow::Error> {
+        let subsystem = constcat::concat!(minimald_rpc::RPC_SUBSYSTEM_PREFIX, "HomedirFilesTarZst");
+
+        let mut channel = self
+            .handle
+            .channel_open_session()
+            .await
+            .context("open channel for workspace file upload")?;
-        let mut channel = self
-            .handle
-            .channel_open_session()
-            .await
-            .context("open channel for workspace file upload")?;
+        let mut channel = self
+            .handle
+            .channel_open_session()
+            .await
+            .context("open channel for homedir file upload")?;
-        let mut channel = self
-            .handle
-            .channel_open_session()
-            .await
-            .context("open channel for workspace file upload")?;
+        let mut channel = self
+            .handle
+            .channel_open_session()
+            .await
+            .context("open channel for homedir file upload")?;
+
+        channel
+            .set_env(true, "MINIMAL_SESSION_ID", session_id.to_string())
+            .await
+            .context("set MINIMAL_SESSION_ID env")?;
+
+        channel
+            .request_subsystem(true, subsystem)
+            .await
+            .context("request HomedirFilesTarZst subsystem")?;
+
+        crate::file_upload::stream_tar_zstd(dir, channel.make_writer()).await?;
+
+        // Wait for the channel to close to signal that unpacking is done.
+        let mut err = Vec::new();
+        while let Some(msg) = channel.wait().await {
+            if let russh::ChannelMsg::ExtendedData { data, ext: 1 } = msg {
+                err.extend_from_slice(&data);
+            }
+        }
+        if !err.is_empty() {
+            anyhow::bail!(
+                "daemon failed to unpack homedir files: {}",
+                String::from_utf8_lossy(&err)
+            );
+        }
+        Ok(())
+    }
 }
 
 /// Resolve the provider-instance dir (`<state dir>/providers/local-0`) the

@@ -524,21 +524,41 @@ async fn serve_stream_workspace_files(
     config: ChannelConfig,
     mut c: RuChannel<Msg>,
 ) {
-    if let Err(msg) = unpack_workspace_files(&s, &config, &mut c).await {
+    if let Err(msg) = unpack_files(&s, &config, &mut c, UnpackDir::Workspace).await {
         let _ = c.extended_data_bytes(1, msg).await;
     }
     let _ = c.close().await;
 }
 
+pub(crate) const STREAM_HOMEDIR_FILES: &str =
+    constcat::concat!(RPC_SUBSYSTEM_PREFIX, "HomedirFilesTarZst");
+
+async fn serve_stream_homedir_files(
+    s: ServerStateHandle,
+    config: ChannelConfig,
+    mut c: RuChannel<Msg>,
+) {
+    if let Err(msg) = unpack_files(&s, &config, &mut c, UnpackDir::Homedir).await {
+        let _ = c.extended_data_bytes(1, msg).await;
+    }
+    let _ = c.close().await;
+}
+
+enum UnpackDir {
+    Workspace,
+    Homedir,
+}
+
 /// Unpacks the zstd-compressed tarball streamed over `c` into the
-/// workspace directory of the session named by the channel environment.
+/// given directory of the session named by the channel environment.
 ///
 /// On failure, returns the human-readable message to relay back to the
 /// client over the channel's extended-data stream.
-async fn unpack_workspace_files(
+async fn unpack_files(
     s: &ServerStateHandle,
     config: &ChannelConfig,
     c: &mut RuChannel<Msg>,
+    into_dir: UnpackDir,
 ) -> Result<(), String> {
     let session_id_str = config
         .env_vars
@@ -562,7 +582,13 @@ async fn unpack_workspace_files(
         c.make_reader(),
     ));
     async_tar::Archive::new(reader)
-        .unpack(paths.working.as_utf8_path())
+        .unpack(
+            match into_dir {
+                UnpackDir::Workspace => paths.working,
+                UnpackDir::Homedir => paths.home,
+            }
+            .as_utf8_path(),
+        )
         .await
         .map_err(|e| format!("unpack failed: {e}"))?;
 
@@ -603,6 +629,7 @@ pub async fn handle_ssh_rpc(
         | GetSessionPolicy::NAME
         | GetMeshStatus::NAME
         | STREAM_WORKSPACE_FILES
+        | STREAM_HOMEDIR_FILES
         | IssueClientCert::NAME => {
             let mut conn_lock = c.lock().await;
             let c_hnd = match conn_lock.take(id) {
@@ -642,6 +669,7 @@ pub async fn handle_ssh_rpc(
         GetSessionPolicy::NAME => drop(spawn(serve_get_session_policy(s, channel))),
         GetMeshStatus::NAME => drop(spawn(serve_get_mesh_status(s, channel))),
         STREAM_WORKSPACE_FILES => drop(spawn(serve_stream_workspace_files(s, config, channel))),
+        STREAM_HOMEDIR_FILES => drop(spawn(serve_stream_homedir_files(s, config, channel))),
         IssueClientCert::NAME => {
             #[cfg(feature = "networking-proxy")]
             drop(spawn(serve_issue_client_cert(s, channel)));
@@ -784,6 +812,55 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn stream_homedir_files_unpacks_tarball_into_home() {
+        let server = TestServer::new().await;
+        let mut client = server.connect().await;
+        let session_id = fresh_session(&mut client).await;
+
+        let payload = tar_zst(&[
+            ("hello.txt", b"hello world\n"),
+            ("dir/nested.txt", b"nested contents"),
+        ])
+        .await;
+
+        let channel = client
+            .open_subsystem(
+                STREAM_HOMEDIR_FILES,
+                &[(MINIMAL_SESSION_ID_ENV, &session_id.to_string())],
+            )
+            .await;
+        let mut stream = channel.into_stream();
+        stream.write_all(&payload).await.unwrap();
+        // Half-close so the server's decoder sees EOF; then read to the
+        // server's channel close so the unpack has completed on-disk
+        // before we assert.
+        stream.shutdown().await.unwrap();
+        let mut trailing = Vec::new();
+        stream.read_to_end(&mut trailing).await.unwrap();
+
+        let mngr = server.state.sessions_manager().await;
+        let handle = mngr
+            .get_session(SessionKeyPredicate::Id(session_id))
+            .await
+            .unwrap()
+            .expect("freshly-created session should be retrievable");
+        let paths = handle.paths().await.unwrap();
+
+        assert_eq!(
+            tokio::fs::read(paths.home.as_utf8_path().join("hello.txt"))
+                .await
+                .unwrap(),
+            b"hello world\n",
+        );
+        assert_eq!(
+            tokio::fs::read(paths.home.as_utf8_path().join("dir/nested.txt"))
+                .await
+                .unwrap(),
+            b"nested contents",
+        );
+    }
+
     #[tokio::test]
     async fn get_version_returns_compiled_in_versions() {
         let server = TestServer::new().await;