diff --git a/android/TerminalApp/Android.bp b/android/TerminalApp/Android.bp
index 2bac412..e1e236a 100644
--- a/android/TerminalApp/Android.bp
+++ b/android/TerminalApp/Android.bp
@@ -15,6 +15,7 @@
         "android.system.virtualizationservice_internal-java",
         "androidx-constraintlayout_constraintlayout",
         "androidx.window_window",
+        "androidx.work_work-runtime",
         "apache-commons-compress",
         "avf_aconfig_flags_java",
         "com.google.android.material_material",
diff --git a/android/TerminalApp/java/com/android/virtualization/terminal/DebianServiceImpl.kt b/android/TerminalApp/java/com/android/virtualization/terminal/DebianServiceImpl.kt
index e035ad4..e81be7f 100644
--- a/android/TerminalApp/java/com/android/virtualization/terminal/DebianServiceImpl.kt
+++ b/android/TerminalApp/java/com/android/virtualization/terminal/DebianServiceImpl.kt
@@ -18,7 +18,8 @@
 import android.content.Context
 import android.util.Log
 import androidx.annotation.Keep
-import com.android.virtualization.terminal.DebianServiceImpl.ForwarderHostCallback
+import com.android.internal.annotations.GuardedBy
+import com.android.system.virtualmachine.flags.Flags
 import com.android.virtualization.terminal.MainActivity.Companion.TAG
 import com.android.virtualization.terminal.PortsStateManager.Companion.getInstance
 import com.android.virtualization.terminal.proto.DebianServiceGrpc.DebianServiceImplBase
@@ -28,6 +29,8 @@
 import com.android.virtualization.terminal.proto.ReportVmActivePortsResponse
 import com.android.virtualization.terminal.proto.ShutdownQueueOpeningRequest
 import com.android.virtualization.terminal.proto.ShutdownRequestItem
+import com.android.virtualization.terminal.proto.StorageBalloonQueueOpeningRequest
+import com.android.virtualization.terminal.proto.StorageBalloonRequestItem
 import io.grpc.stub.ServerCallStreamObserver
 import io.grpc.stub.StreamObserver
 
@@ -35,6 +38,8 @@
     private val portsStateManager: PortsStateManager = getInstance(context)
     private var portsStateListener: PortsStateManager.Listener? = null
     private var shutdownRunnable: Runnable? = null
+    private val mLock = Object()
+    @GuardedBy("mLock") private var storageBalloonCallback: StorageBalloonCallback? = null
 
     override fun reportVmActivePorts(
         request: ReportVmActivePortsRequest,
@@ -80,10 +85,9 @@
         request: ShutdownQueueOpeningRequest?,
         responseObserver: StreamObserver<ShutdownRequestItem?>,
     ) {
-        val serverCallStreamObserver = responseObserver as ServerCallStreamObserver<ShutdownRequestItem?>
-        serverCallStreamObserver.setOnCancelHandler {
-            shutdownRunnable = null
-        }
+        val serverCallStreamObserver =
+            responseObserver as ServerCallStreamObserver<ShutdownRequestItem?>
+        serverCallStreamObserver.setOnCancelHandler { shutdownRunnable = null }
         Log.d(TAG, "openShutdownRequestQueue")
         shutdownRunnable = Runnable {
             if (serverCallStreamObserver.isCancelled()) {
@@ -95,6 +99,60 @@
         }
     }
 
+    private class StorageBalloonCallback(
+        private val responseObserver: StreamObserver<StorageBalloonRequestItem?>
+    ) {
+        fun setAvailableStorageBytes(availableBytes: Long) {
+            Log.d(TAG, "send setStorageBalloon: $availableBytes")
+            val item =
+                StorageBalloonRequestItem.newBuilder().setAvailableBytes(availableBytes).build()
+            responseObserver.onNext(item)
+        }
+
+        fun closeConnection() {
+            Log.d(TAG, "close StorageBalloonQueue")
+            responseObserver.onCompleted()
+        }
+    }
+
+    fun setAvailableStorageBytes(availableBytes: Long): Boolean {
+        synchronized(mLock) {
+            if (storageBalloonCallback == null) {
+                Log.d(TAG, "storageBalloonCallback is not ready.")
+                return false
+            }
+            storageBalloonCallback!!.setAvailableStorageBytes(availableBytes)
+        }
+        return true
+    }
+
+    override fun openStorageBalloonRequestQueue(
+        request: StorageBalloonQueueOpeningRequest?,
+        responseObserver: StreamObserver<StorageBalloonRequestItem?>,
+    ) {
+        if (!Flags.terminalStorageBalloon()) {
+            return
+        }
+        Log.d(TAG, "openStorageRequestQueue")
+        synchronized(mLock) {
+            if (storageBalloonCallback != null) {
+                Log.d(TAG, "RequestQueue already exists. Closing connection.")
+                storageBalloonCallback!!.closeConnection()
+            }
+            storageBalloonCallback = StorageBalloonCallback(responseObserver)
+        }
+    }
+
+    fun closeStorageBalloonRequestQueue() {
+        Log.d(TAG, "Stopping storage balloon queue")
+        synchronized(mLock) {
+            if (storageBalloonCallback != null) {
+                storageBalloonCallback!!.closeConnection()
+                storageBalloonCallback = null
+            }
+        }
+    }
+
     @Keep
     private class ForwarderHostCallback(
         private val responseObserver: StreamObserver<ForwardingRequestItem?>
diff --git a/android/TerminalApp/java/com/android/virtualization/terminal/StorageBalloonWorker.kt b/android/TerminalApp/java/com/android/virtualization/terminal/StorageBalloonWorker.kt
new file mode 100644
index 0000000..345bf75
--- /dev/null
+++ b/android/TerminalApp/java/com/android/virtualization/terminal/StorageBalloonWorker.kt
@@ -0,0 +1,111 @@
+/*
+ * Copyright (C) 2025 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.android.virtualization.terminal
+
+import android.content.Context
+import android.os.storage.StorageManager
+import android.os.storage.StorageManager.UUID_DEFAULT
+import android.util.Log
+import androidx.work.WorkManager
+import androidx.work.Worker
+import androidx.work.WorkerParameters
+import com.android.virtualization.terminal.MainActivity.Companion.TAG
+import java.util.concurrent.TimeUnit
+
+class StorageBalloonWorker(appContext: Context, workerParams: WorkerParameters) :
+    Worker(appContext, workerParams) {
+
+    override fun doWork(): Result {
+        Log.d(TAG, "StorageBalloonWorker.doWork() called")
+
+        var storageManager =
+            applicationContext.getSystemService(Context.STORAGE_SERVICE) as StorageManager
+        val hostAllocatableBytes = storageManager.getAllocatableBytes(UUID_DEFAULT)
+
+        val guestAvailableBytes = calculateGuestAvailableStorageSize(hostAllocatableBytes)
+        // debianService must be set when this function is called.
+        debianService!!.setAvailableStorageBytes(guestAvailableBytes)
+
+        val delaySeconds = calculateDelaySeconds(hostAllocatableBytes)
+        scheduleNextTask(delaySeconds)
+
+        return Result.success()
+    }
+
+    private fun calculateGuestAvailableStorageSize(hostAllocatableBytes: Long): Long {
+        return hostAllocatableBytes - HOST_RESERVED_BYTES
+    }
+
+    private fun calculateDelaySeconds(hostAvailableBytes: Long): Long {
+        return when {
+            hostAvailableBytes < CRITICAL_STORAGE_THRESHOLD_BYTES -> CRITICAL_DELAY_SECONDS
+            hostAvailableBytes < LOW_STORAGE_THRESHOLD_BYTES -> LOW_STORAGE_DELAY_SECONDS
+            hostAvailableBytes < MODERATE_STORAGE_THRESHOLD_BYTES -> MODERATE_STORAGE_DELAY_SECONDS
+            else -> NORMAL_DELAY_SECONDS
+        }
+    }
+
+    private fun scheduleNextTask(delaySeconds: Long) {
+        val storageBalloonTaskRequest =
+            androidx.work.OneTimeWorkRequest.Builder(StorageBalloonWorker::class.java)
+                .setInitialDelay(delaySeconds, TimeUnit.SECONDS)
+                .build()
+        androidx.work.WorkManager.getInstance(applicationContext)
+            .enqueueUniqueWork(
+                "storageBalloonTask",
+                androidx.work.ExistingWorkPolicy.REPLACE,
+                storageBalloonTaskRequest,
+            )
+        Log.d(TAG, "next storage balloon task is scheduled in $delaySeconds seconds")
+    }
+
+    companion object {
+        private var debianService: DebianServiceImpl? = null
+
+        // Reserve 1GB as host-only region.
+        private const val HOST_RESERVED_BYTES = 1024L * 1024 * 1024
+
+        // Thresholds for deciding time period to report storage information to the guest.
+        // Less storage is available on the host, more frequently the host will report storage
+        // information to the guest.
+        //
+        // Critical: (host storage < 1GB) => report every 5 seconds
+        private const val CRITICAL_STORAGE_THRESHOLD_BYTES = 1L * 1024 * 1024 * 1024
+        private const val CRITICAL_DELAY_SECONDS = 5L
+        // Low: (1GB <= storage < 5GB) => report every 60 seconds
+        private const val LOW_STORAGE_THRESHOLD_BYTES = 5L * 1024 * 1024 * 1024
+        private const val LOW_STORAGE_DELAY_SECONDS = 60L
+        // Moderate: (5GB <= storage < 10GB) => report every 15 minutes
+        private const val MODERATE_STORAGE_THRESHOLD_BYTES = 10L * 1024 * 1024 * 1024
+        private const val MODERATE_STORAGE_DELAY_SECONDS = 15L * 60
+        // Normal: report every 60 minutes
+        private const val NORMAL_DELAY_SECONDS = 60L * 60
+
+        internal fun start(ctx: Context, ds: DebianServiceImpl) {
+            debianService = ds
+            val storageBalloonTaskRequest =
+                androidx.work.OneTimeWorkRequest.Builder(StorageBalloonWorker::class.java)
+                    .setInitialDelay(1, TimeUnit.SECONDS)
+                    .build()
+            androidx.work.WorkManager.getInstance(ctx)
+                .enqueueUniqueWork(
+                    "storageBalloonTask",
+                    androidx.work.ExistingWorkPolicy.REPLACE,
+                    storageBalloonTaskRequest,
+                )
+        }
+    }
+}
diff --git a/android/TerminalApp/java/com/android/virtualization/terminal/VmLauncherService.kt b/android/TerminalApp/java/com/android/virtualization/terminal/VmLauncherService.kt
index 1857175..0a1f0ee 100644
--- a/android/TerminalApp/java/com/android/virtualization/terminal/VmLauncherService.kt
+++ b/android/TerminalApp/java/com/android/virtualization/terminal/VmLauncherService.kt
@@ -38,10 +38,9 @@
 import android.system.virtualmachine.VirtualMachineException
 import android.util.Log
 import android.widget.Toast
-import com.android.system.virtualmachine.flags.Flags.terminalGuiSupport
+import com.android.system.virtualmachine.flags.Flags
 import com.android.virtualization.terminal.MainActivity.Companion.TAG
 import com.android.virtualization.terminal.Runner.Companion.create
-import com.android.virtualization.terminal.VmLauncherService.VmLauncherServiceCallback
 import io.grpc.Grpc
 import io.grpc.InsecureServerCredentials
 import io.grpc.Metadata
@@ -54,7 +53,6 @@
 import java.io.File
 import java.io.FileOutputStream
 import java.io.IOException
-import java.lang.RuntimeException
 import java.net.InetSocketAddress
 import java.net.SocketAddress
 import java.nio.file.Files
@@ -285,7 +283,7 @@
 
         // Set the initial display size
         // TODO(jeongik): set up the display size on demand
-        if (terminalGuiSupport() && displayInfo != null) {
+        if (Flags.terminalGuiSupport() && displayInfo != null) {
             builder
                 .setDisplayConfig(
                     VirtualMachineCustomImageConfig.DisplayConfig.Builder()
@@ -360,6 +358,10 @@
                 }
             }
         )
+
+        if (Flags.terminalStorageBalloon()) {
+            StorageBalloonWorker.start(this, debianService!!)
+        }
     }
 
     override fun onDestroy() {
@@ -383,6 +385,7 @@
 
     private fun stopDebianServer() {
         debianService?.killForwarderHost()
+        debianService?.closeStorageBalloonRequestQueue()
         server?.shutdown()
     }
 
diff --git a/build/avf_flags.aconfig b/build/avf_flags.aconfig
index 921c374..571c359 100644
--- a/build/avf_flags.aconfig
+++ b/build/avf_flags.aconfig
@@ -16,4 +16,12 @@
   namespace: "virtualization"
   description: "Flag for GUI support in terminal"
   bug: "386296118"
+}
+
+flag {
+  name: "terminal_storage_balloon"
+  is_exported: true
+  namespace: "virtualization"
+  description: "Flag for storage ballooning support in terminal"
+  bug: "382174138"
 }
\ No newline at end of file
diff --git a/build/debian/build.sh b/build/debian/build.sh
index 9c4d4b1..8c1345c 100755
--- a/build/debian/build.sh
+++ b/build/debian/build.sh
@@ -204,6 +204,7 @@
 	build_rust_as_deb forwarder_guest
 	build_rust_as_deb forwarder_guest_launcher
 	build_rust_as_deb shutdown_runner
+	build_rust_as_deb storage_balloon_agent
 }
 
 package_custom_kernel() {
diff --git a/build/debian/fai_config/package_config/AVF b/build/debian/fai_config/package_config/AVF
index 3aa8ab0..f1ee065 100644
--- a/build/debian/fai_config/package_config/AVF
+++ b/build/debian/fai_config/package_config/AVF
@@ -8,6 +8,7 @@
 forwarder-guest
 forwarder-guest-launcher
 shutdown-runner
+storage-balloon-agent
 weston
 xwayland
 mesa-vulkan-drivers
diff --git a/guest/storage_balloon_agent/.cargo/config.toml b/guest/storage_balloon_agent/.cargo/config.toml
new file mode 100644
index 0000000..a451cda
--- /dev/null
+++ b/guest/storage_balloon_agent/.cargo/config.toml
@@ -0,0 +1,6 @@
+[target.aarch64-unknown-linux-gnu]
+linker = "aarch64-linux-gnu-gcc"
+rustflags = ["-C", "target-feature=+crt-static"]
+
+[target.x86_64-unknown-linux-gnu]
+rustflags = ["-C", "target-feature=+crt-static"]
diff --git a/guest/storage_balloon_agent/.gitignore b/guest/storage_balloon_agent/.gitignore
new file mode 100644
index 0000000..ea8c4bf
--- /dev/null
+++ b/guest/storage_balloon_agent/.gitignore
@@ -0,0 +1 @@
+/target
diff --git a/guest/storage_balloon_agent/Cargo.toml b/guest/storage_balloon_agent/Cargo.toml
new file mode 100644
index 0000000..ce0e5d7
--- /dev/null
+++ b/guest/storage_balloon_agent/Cargo.toml
@@ -0,0 +1,26 @@
+[package]
+name = "storage_balloon_agent"
+version = "0.1.0"
+edition = "2021"
+license = "Apache-2.0"
+
+[dependencies]
+anyhow = "1.0.94"
+clap = { version = "4.5.20", features = ["derive"] }
+env_logger = "0.10.2"
+log = "0.4.22"
+netdev = "0.31.0"
+nix = { version = "0.28.0", features = ["fs"] }
+prost = "0.13.3"
+tokio = { version = "1.40.0", features = ["rt-multi-thread"] }
+tonic = "0.12.3"
+
+[build-dependencies]
+tonic-build = "0.12.3"
+
+[package.metadata.deb]
+maintainer = "ferrochrome-dev@google.com"
+copyright = "2025, The Android Open Source Project"
+depends = "$auto"
+maintainer-scripts = "debian/"
+systemd-units = { }
diff --git a/guest/storage_balloon_agent/build.rs b/guest/storage_balloon_agent/build.rs
new file mode 100644
index 0000000..e3939d4
--- /dev/null
+++ b/guest/storage_balloon_agent/build.rs
@@ -0,0 +1,7 @@
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let proto_file = "../../libs/debian_service/proto/DebianService.proto";
+
+    tonic_build::compile_protos(proto_file).unwrap();
+
+    Ok(())
+}
diff --git a/guest/storage_balloon_agent/debian/service b/guest/storage_balloon_agent/debian/service
new file mode 100644
index 0000000..0e9b03a
--- /dev/null
+++ b/guest/storage_balloon_agent/debian/service
@@ -0,0 +1,17 @@
+[Unit]
+After=syslog.target
+After=network.target
+After=virtiofs_internal.service
+
+[Service]
+ExecStart=/usr/bin/bash -c '/usr/bin/storage_balloon_agent --grpc_port_file /mnt/internal/debian_service_port'
+Type=simple
+Restart=on-failure
+RestartSec=1
+User=root
+Group=root
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=multi-user.target
diff --git a/guest/storage_balloon_agent/rustfmt.toml b/guest/storage_balloon_agent/rustfmt.toml
new file mode 120000
index 0000000..be3dbe2
--- /dev/null
+++ b/guest/storage_balloon_agent/rustfmt.toml
@@ -0,0 +1 @@
+../../../../../build/soong/scripts/rustfmt.toml
\ No newline at end of file
diff --git a/guest/storage_balloon_agent/src/main.rs b/guest/storage_balloon_agent/src/main.rs
new file mode 100644
index 0000000..817b337
--- /dev/null
+++ b/guest/storage_balloon_agent/src/main.rs
@@ -0,0 +1,141 @@
+// Copyright 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! gRPC daemon for the storage ballooning feature.
+
+use anyhow::anyhow;
+use anyhow::Context;
+use anyhow::Result;
+use api::debian_service_client::DebianServiceClient;
+use api::StorageBalloonQueueOpeningRequest;
+use api::StorageBalloonRequestItem;
+use clap::Parser;
+use log::debug;
+use log::error;
+use log::info;
+use nix::sys::statvfs::statvfs;
+pub mod api {
+    tonic::include_proto!("com.android.virtualization.terminal.proto");
+}
+
+#[derive(Parser)]
+/// Flags for running command
+pub struct Args {
+    /// IP address
+    #[arg(long)]
+    addr: Option<String>,
+
+    /// path to a file where grpc port number is written
+    #[arg(long)]
+    #[arg(alias = "grpc_port_file")]
+    grpc_port_file: String,
+}
+
+// Calculates how many blocks to be reserved.
+fn calculate_clusters_count(guest_available_bytes: u64) -> Result<u64> {
+    let stat = statvfs("/").context("failed to get statvfs")?;
+    let fr_size = stat.fragment_size() as u64;
+
+    if fr_size == 0 {
+        return Err(anyhow::anyhow!("fragment size is zero, fr_size: {}", fr_size));
+    }
+
+    let total = fr_size.checked_mul(stat.blocks() as u64).context(format!(
+        "overflow in total size calculation, fr_size: {}, blocks: {}",
+        fr_size,
+        stat.blocks()
+    ))?;
+
+    let free = fr_size.checked_mul(stat.blocks_available() as u64).context(format!(
+        "overflow in free size calculation, fr_size: {}, blocks_available: {}",
+        fr_size,
+        stat.blocks_available()
+    ))?;
+
+    let used = total
+        .checked_sub(free)
+        .context(format!("underflow in used size calculation (free > total), which should not happen, total: {}, free: {}", total, free))?;
+
+    let avail = std::cmp::min(free, guest_available_bytes);
+    let balloon_size_bytes = free - avail;
+
+    let reserved_clusters_count = balloon_size_bytes.div_ceil(fr_size);
+
+    debug!("total: {total}, free: {free}, used: {used}, avail: {avail}, balloon: {balloon_size_bytes}, clusters_count: {reserved_clusters_count}");
+
+    Ok(reserved_clusters_count)
+}
+
+fn set_reserved_clusters(clusters_count: u64) -> anyhow::Result<()> {
+    const ROOTFS_DEVICE_NAME: &str = "vda1";
+    std::fs::write(
+        format!("/sys/fs/ext4/{ROOTFS_DEVICE_NAME}/reserved_clusters"),
+        clusters_count.to_string(),
+    )
+    .context("failed to write reserved_clusters")?;
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::builder().filter_level(log::LevelFilter::Debug).init();
+
+    let args = Args::parse();
+    let gateway_ip_addr = netdev::get_default_gateway()?.ipv4[0];
+    let addr = args.addr.unwrap_or_else(|| gateway_ip_addr.to_string());
+
+    // Wait for `grpc_port_file` becomes available.
+    const GRPC_PORT_MAX_RETRY_COUNT: u32 = 10;
+    for _ in 0..GRPC_PORT_MAX_RETRY_COUNT {
+        if std::path::Path::new(&args.grpc_port_file).exists() {
+            break;
+        }
+        debug!("{} does not exist. Wait 1 second", args.grpc_port_file);
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    }
+    let grpc_port = std::fs::read_to_string(&args.grpc_port_file)?.trim().to_string();
+    let server_addr = format!("http://{}:{}", addr, grpc_port);
+
+    info!("connect to grpc server {}", server_addr);
+    let mut client = DebianServiceClient::connect(server_addr)
+        .await
+        .map_err(|e| anyhow!("failed to connect to grpc server: {:#}", e))?;
+    info!("connection established");
+
+    let mut res_stream = client
+        .open_storage_balloon_request_queue(tonic::Request::new(
+            StorageBalloonQueueOpeningRequest {},
+        ))
+        .await
+        .map_err(|e| anyhow!("failed to open storage balloon queue: {:#}", e))?
+        .into_inner();
+
+    while let Some(StorageBalloonRequestItem { available_bytes }) =
+        res_stream.message().await.map_err(|e| anyhow!("failed to receive message: {:#}", e))?
+    {
+        let clusters_count = match calculate_clusters_count(available_bytes) {
+            Ok(c) => c,
+            Err(e) => {
+                error!("failed to calculate cluster size to be reserved: {:#}", e);
+                continue;
+            }
+        };
+
+        if let Err(e) = set_reserved_clusters(clusters_count) {
+            error!("failed to set storage balloon size: {}", e);
+        }
+    }
+
+    Ok(())
+}
diff --git a/libs/debian_service/proto/DebianService.proto b/libs/debian_service/proto/DebianService.proto
index 43955fa..e52b28a 100644
--- a/libs/debian_service/proto/DebianService.proto
+++ b/libs/debian_service/proto/DebianService.proto
@@ -25,6 +25,7 @@
   rpc ReportVmActivePorts (ReportVmActivePortsRequest) returns (ReportVmActivePortsResponse) {}
   rpc OpenForwardingRequestQueue (QueueOpeningRequest) returns (stream ForwardingRequestItem) {}
   rpc OpenShutdownRequestQueue (ShutdownQueueOpeningRequest) returns (stream ShutdownRequestItem) {}
+  rpc OpenStorageBalloonRequestQueue (StorageBalloonQueueOpeningRequest) returns (stream StorageBalloonRequestItem) {}
 }
 
 message QueueOpeningRequest {
@@ -52,3 +53,9 @@
 message ShutdownQueueOpeningRequest {}
 
 message ShutdownRequestItem {}
+
+message StorageBalloonQueueOpeningRequest {}
+
+message StorageBalloonRequestItem {
+  uint64 available_bytes = 1;
+}