diff --git a/src/reapi/action_cache.rs b/src/reapi/action_cache.rs
new file mode 100644
index 0000000000..7f850ff598
--- /dev/null
+++ b/src/reapi/action_cache.rs
@@ -0,0 +1,131 @@
+// Copyright 2026 Mozilla Foundation
+//
+// 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.
+
+//! Looks actions up in a REAPI server's action cache.
+
+use super::client::{AuthChannel, Client, RPC_TIMEOUT, with_timeout};
+use super::proto::build::bazel::remote::execution::v2 as remexec;
+use crate::errors::*;
+use remexec::action_cache_client::ActionCacheClient;
+
+/// The action cache of the REAPI server behind a [`Client`].
+pub(crate) struct ActionCache {
+    instance_name: String,
+    action_cache: ActionCacheClient<AuthChannel>,
+}
+
+impl ActionCache {
+    pub(crate) fn new(client: &Client) -> Self {
+        ActionCache {
+            instance_name: client.instance_name().to_owned(),
+            action_cache: ActionCacheClient::new(client.channel()),
+        }
+    }
+
+    /// The action cache's result for `action_digest`, or `None` on a miss.
+    pub(crate) async fn get_action_result(
+        &self,
+        action_digest: &remexec::Digest,
+    ) -> Result<Option<remexec::ActionResult>> {
+        let request = remexec::GetActionResultRequest {
+            instance_name: self.instance_name.clone(),
+            action_digest: Some(action_digest.clone()),
+            inline_stdout: true,
+            inline_stderr: true,
+            inline_output_files: vec![],
+            digest_function: remexec::digest_function::Value::Sha256 as i32,
+        };
+        with_timeout(
+            &format!(
+                "looking up action {}/{} in the REAPI action cache",
+                action_digest.hash, action_digest.size_bytes
+            ),
+            RPC_TIMEOUT,
+            async {
+                match self.action_cache.clone().get_action_result(request).await {
+                    Ok(response) => Ok(Some(response.into_inner())),
+                    Err(status) if status.code() == tonic::Code::NotFound => Ok(None),
+                    Err(status) => Err(status.into()),
+                }
+            },
+        )
+        .await
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::reapi::digest;
+    use crate::reapi::fake::FakeServer;
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
+
+    #[tokio::test]
+    async fn test_a_cached_result_is_found() {
+        let server = Arc::new(FakeServer::default());
+        let action_cache = ActionCache::new(&server.connect().await);
+        let action_digest = digest::of_bytes(b"action");
+        assert!(
+            action_cache
+                .get_action_result(&action_digest)
+                .await
+                .unwrap()
+                .is_none()
+        );
+        let result = remexec::ActionResult {
+            exit_code: 1,
+            ..Default::default()
+        };
+        server.insert_action_result(&action_digest, result.clone());
+        assert_eq!(
+            Some(result),
+            action_cache
+                .get_action_result(&action_digest)
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_lookups_use_the_instance_name() {
+        let server = Arc::new(FakeServer {
+            instance_name: "main".to_owned(),
+            ..Default::default()
+        });
+        let action_cache = ActionCache::new(&server.connect().await);
+        let action_digest = digest::of_bytes(b"action");
+        let result = remexec::ActionResult::default();
+        server.insert_action_result(&action_digest, result.clone());
+        assert_eq!(
+            Some(result),
+            action_cache
+                .get_action_result(&action_digest)
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_a_failed_lookup_is_an_error() {
+        let server = Arc::new(FakeServer::default());
+        let action_cache = ActionCache::new(&server.connect().await);
+        server.fail_lookups.store(true, Ordering::SeqCst);
+        let e = action_cache
+            .get_action_result(&digest::of_bytes(b"action"))
+            .await
+            .unwrap_err();
+        assert!(format!("{e:#}").contains("on fire"), "{e:#}");
+    }
+}
diff --git a/src/reapi/cas.rs b/src/reapi/cas.rs
new file mode 100644
index 0000000000..4e4c610ed7
--- /dev/null
+++ b/src/reapi/cas.rs
@@ -0,0 +1,382 @@
+// Copyright 2026 Mozilla Foundation
+//
+// 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.
+
+//! Downloads blobs from a REAPI server's content-addressable storage (CAS) and checks
+//! them against their digests.
+
+use super::client::{AuthChannel, Client, RPC_TIMEOUT, with_timeout};
+use super::digest;
+use super::proto::build::bazel::remote::execution::v2 as remexec;
+use super::proto::google::bytestream::ReadRequest;
+use super::proto::google::bytestream::byte_stream_client::ByteStreamClient;
+use crate::errors::*;
+use sha2::{Digest as _, Sha256};
+use std::io::{self, Write};
+use std::time::Duration;
+use tempfile::{NamedTempFile, TempPath};
+use tokio::sync::mpsc;
+
+/// The time to download `bytes` at 1 MiB/s, plus [`RPC_TIMEOUT`].
+fn download_timeout(bytes: u64) -> Duration {
+    RPC_TIMEOUT + Duration::from_secs(bytes / (1024 * 1024))
+}
+
+/// Error for a blob the REAPI CAS doesn't have, such as an evicted one.
+#[derive(Debug)]
+pub(crate) struct MissingBlob(remexec::Digest);
+
+impl std::fmt::Display for MissingBlob {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "the REAPI CAS doesn't have blob {}/{}",
+            self.0.hash, self.0.size_bytes
+        )
+    }
+}
+
+impl std::error::Error for MissingBlob {}
+
+/// Fails if `digest`'s hash isn't 64 lowercase hex digits, its size is negative, or it's
+/// empty with another hash.
+fn check_digest(digest: &remexec::Digest) -> Result<()> {
+    let is_lower_hex = |c: char| c.is_ascii_digit() || ('a'..='f').contains(&c);
+    if digest.size_bytes < 0
+        || digest.hash.len() != 64
+        || !digest.hash.chars().all(is_lower_hex)
+        || (digest.size_bytes == 0 && *digest != digest::of_bytes(b""))
+    {
+        bail!("invalid REAPI digest {}/{}", digest.hash, digest.size_bytes);
+    }
+    Ok(())
+}
+
+/// The CAS of the REAPI server behind a [`Client`]. Downloads fail with [`MissingBlob`]
+/// when the CAS doesn't have the blob.
+pub(crate) struct Cas {
+    instance_name: String,
+    bytestream: ByteStreamClient<AuthChannel>,
+    supports_zstd_bytestream: bool,
+}
+
+impl Cas {
+    pub(crate) fn new(client: &Client) -> Self {
+        Cas {
+            instance_name: client.instance_name().to_owned(),
+            bytestream: ByteStreamClient::new(client.channel()),
+            supports_zstd_bytestream: client.supports_zstd_bytestream(),
+        }
+    }
+
+    fn download_resource_name(&self, digest: &remexec::Digest) -> String {
+        format!(
+            "{}{}/{}/{}",
+            instance_prefix(&self.instance_name),
+            if self.supports_zstd_bytestream {
+                "compressed-blobs/zstd"
+            } else {
+                "blobs"
+            },
+            digest.hash,
+            digest.size_bytes
+        )
+    }
+
+    /// The digest's size, or zstd's bound on its compressed size.
+    fn max_received_bytes(&self, digest: &remexec::Digest) -> u64 {
+        let size = digest.size_bytes as u64;
+        if self.supports_zstd_bytestream {
+            zstd::zstd_safe::compress_bound(size as usize) as u64
+        } else {
+            size
+        }
+    }
+
+    pub(crate) async fn read_blob_to_memory(&self, digest: &remexec::Digest) -> Result<Vec<u8>> {
+        self.read_blob_into(digest, vec![]).await
+    }
+
+    pub(crate) async fn read_blob_to_file(
+        &self,
+        digest: &remexec::Digest,
+        file: NamedTempFile,
+    ) -> Result<TempPath> {
+        let path = file.path().to_owned();
+        let file = self
+            .read_blob_into(digest, io::BufWriter::new(file))
+            .await?
+            .into_inner()
+            .map_err(|e| e.into_error())
+            .with_context(|| format!("writing {}", path.display()))?;
+        Ok(file.into_temp_path())
+    }
+
+    /// Downloads `digest`'s blob into `out`, failing if the bytes don't match `digest`. A
+    /// blocking thread decompresses, hashes and writes each chunk.
+    async fn read_blob_into<W: Write + Send + 'static>(
+        &self,
+        digest: &remexec::Digest,
+        out: W,
+    ) -> Result<W> {
+        check_digest(digest)?;
+        if digest.size_bytes == 0 {
+            return Ok(out);
+        }
+        let resource_name = self.download_resource_name(digest);
+        let max_received_bytes = self.max_received_bytes(digest);
+        let status_to_error = |status: tonic::Status| match status.code() {
+            tonic::Code::NotFound => Error::new(MissingBlob(digest.clone())),
+            _ => status.into(),
+        };
+        let (tx, mut rx) = mpsc::channel::<Vec<u8>>(2);
+        let (supports_zstd_bytestream, size_bytes) =
+            (self.supports_zstd_bytestream, digest.size_bytes);
+        let write_task = tokio::task::spawn_blocking(move || -> Result<(W, remexec::Digest)> {
+            let mut hashing_writer = HashingWriter::new(out, size_bytes);
+            {
+                let mut sink: Box<dyn Write + '_> = if supports_zstd_bytestream {
+                    Box::new(zstd::stream::write::Decoder::new(&mut hashing_writer)?)
+                } else {
+                    Box::new(&mut hashing_writer)
+                };
+                while let Some(chunk) = rx.blocking_recv() {
+                    sink.write_all(&chunk)?;
+                }
+                sink.flush()?;
+            }
+            hashing_writer.finish()
+        });
+        let description = format!(
+            "downloading blob {}/{} from the REAPI CAS",
+            digest.hash, digest.size_bytes
+        );
+        let read_result = with_timeout(
+            &description,
+            download_timeout(digest.size_bytes as u64),
+            async {
+                let mut stream = self
+                    .bytestream
+                    .clone()
+                    .read(ReadRequest {
+                        resource_name,
+                        read_offset: 0,
+                        read_limit: 0,
+                    })
+                    .await
+                    .map_err(status_to_error)?
+                    .into_inner();
+                let mut received_bytes = 0;
+                loop {
+                    let chunk = tokio::select! {
+                        message = stream.message() => message.map_err(status_to_error)?,
+                        // The write task stopped on an error; `write_task.await` returns it.
+                        _ = tx.closed() => break,
+                    };
+                    let Some(chunk) = chunk else { break };
+                    received_bytes += chunk.data.len() as u64;
+                    if received_bytes > max_received_bytes {
+                        bail!("the REAPI server sent more than {max_received_bytes} bytes");
+                    }
+                    // The write task stopped on an error; `write_task.await` returns it.
+                    if tx.send(chunk.data).await.is_err() {
+                        break;
+                    }
+                }
+                Ok(())
+            },
+        )
+        .await;
+        drop(tx);
+        let write_result = write_task.await?;
+        read_result?;
+        let (out, received_digest) = write_result.with_context(|| description.clone())?;
+        if received_digest != *digest {
+            bail!(
+                "{description}: the REAPI server sent blob {}/{} instead",
+                received_digest.hash,
+                received_digest.size_bytes
+            );
+        }
+        Ok(out)
+    }
+}
+
+/// Hashes what passes through to `inner`, failing writes past `max_len` bytes.
+struct HashingWriter<W> {
+    inner: W,
+    hasher: Sha256,
+    len: i64,
+    max_len: i64,
+}
+
+impl<W: Write> HashingWriter<W> {
+    fn new(inner: W, max_len: i64) -> Self {
+        HashingWriter {
+            inner,
+            hasher: Sha256::new(),
+            len: 0,
+            max_len,
+        }
+    }
+
+    /// Flushes `inner` and returns it with the digest of what passed through.
+    fn finish(mut self) -> Result<(W, remexec::Digest)> {
+        self.inner.flush()?;
+        let digest = remexec::Digest {
+            hash: format!("{:x}", self.hasher.finalize()),
+            size_bytes: self.len,
+        };
+        Ok((self.inner, digest))
+    }
+}
+
+impl<W: Write> Write for HashingWriter<W> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        if self.len + buf.len() as i64 > self.max_len {
+            return Err(io::Error::other(format!(
+                "the blob is more than {} bytes",
+                self.max_len
+            )));
+        }
+        let n = self.inner.write(buf)?;
+        self.hasher.update(&buf[..n]);
+        self.len += n as i64;
+        Ok(n)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+fn instance_prefix(instance_name: &str) -> String {
+    if instance_name.is_empty() {
+        String::new()
+    } else {
+        format!("{instance_name}/")
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::reapi::fake::FakeServer;
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
+    use test_case::test_case;
+
+    fn incompressible_bytes(len: usize) -> Vec<u8> {
+        use rand::RngCore;
+        let mut data = vec![0; len];
+        rand::thread_rng().fill_bytes(&mut data);
+        data
+    }
+
+    #[tokio::test]
+    async fn test_empty_blob_needs_no_download() {
+        let server = Arc::new(FakeServer::default());
+        let cas = Cas::new(&server.connect().await);
+        assert!(
+            cas.read_blob_to_memory(&digest::of_bytes(b""))
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    #[test_case(false, "the REAPI server sent more than" ; "uncompressed")]
+    #[test_case(true, "the blob is more than" ; "zstd")]
+    #[tokio::test]
+    async fn test_downloads_are_checked(supports_zstd_bytestream: bool, oversize_error: &str) {
+        let server = Arc::new(FakeServer {
+            supports_zstd_bytestream,
+            ..Default::default()
+        });
+        let cas = Cas::new(&server.connect().await);
+        let data = incompressible_bytes(2 * 1024 * 1024);
+        let digest = server.insert_blob(&data);
+        assert_eq!(data, cas.read_blob_to_memory(&digest).await.unwrap());
+        let file = cas
+            .read_blob_to_file(&digest, NamedTempFile::new().unwrap())
+            .await
+            .unwrap();
+        assert_eq!(data, std::fs::read(&file).unwrap());
+
+        server.corrupt_reads.store(true, Ordering::SeqCst);
+        let e = cas
+            .read_blob_to_file(&digest, NamedTempFile::new().unwrap())
+            .await
+            .unwrap_err();
+        assert!(format!("{e:#}").contains("instead"), "{e:#}");
+        server.corrupt_reads.store(false, Ordering::SeqCst);
+
+        server.oversize_reads.store(true, Ordering::SeqCst);
+        let e = cas.read_blob_to_memory(&digest).await.unwrap_err();
+        assert!(format!("{e:#}").contains(oversize_error), "{e:#}");
+    }
+
+    #[tokio::test]
+    async fn test_downloads_use_the_instance_name() {
+        let server = Arc::new(FakeServer {
+            instance_name: "main".to_owned(),
+            ..Default::default()
+        });
+        let cas = Cas::new(&server.connect().await);
+        let digest = server.insert_blob(b"blob");
+        assert_eq!(
+            b"blob".to_vec(),
+            cas.read_blob_to_memory(&digest).await.unwrap()
+        );
+    }
+
+    #[test_case(false ; "when reading starts")]
+    #[test_case(true ; "in the stream")]
+    #[tokio::test]
+    async fn test_a_blob_the_cas_doesnt_have_is_missing_blob(not_found_in_stream: bool) {
+        let server = Arc::new(FakeServer::default());
+        server
+            .not_found_in_stream
+            .store(not_found_in_stream, Ordering::SeqCst);
+        let cas = Cas::new(&server.connect().await);
+        let e = cas
+            .read_blob_to_memory(&digest::of_bytes(b"absent"))
+            .await
+            .unwrap_err();
+        assert!(e.is::<MissingBlob>(), "{e:#}");
+    }
+
+    #[test]
+    fn test_download_timeout_grows_with_the_blob() {
+        assert_eq!(RPC_TIMEOUT, download_timeout(0));
+        assert_eq!(
+            RPC_TIMEOUT + Duration::from_secs(10),
+            download_timeout(10 * 1024 * 1024)
+        );
+    }
+
+    #[test_case("../../etc/passwd", 5 ; "not a sha256 hash")]
+    #[test_case(&digest::of_bytes(b"x").hash, -1 ; "negative size")]
+    #[test_case(&digest::of_bytes(b"x").hash, 0 ; "empty size with a non-empty hash")]
+    #[tokio::test]
+    async fn test_an_invalid_digest_is_refused(hash: &str, size_bytes: i64) {
+        let server = Arc::new(FakeServer::default());
+        let cas = Cas::new(&server.connect().await);
+        let invalid = remexec::Digest {
+            hash: hash.to_owned(),
+            size_bytes,
+        };
+        let e = cas.read_blob_to_memory(&invalid).await.unwrap_err();
+        assert!(e.to_string().contains("invalid REAPI digest"), "{e}");
+    }
+}
diff --git a/src/reapi/client.rs b/src/reapi/client.rs
index 425f48493a..b69aa280d8 100644
--- a/src/reapi/client.rs
+++ b/src/reapi/client.rs
@@ -23,7 +23,7 @@ use tonic::service::interceptor::InterceptedService;
 use tonic::transport::{Channel, ClientTlsConfig, Endpoint};
 
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
-const RPC_TIMEOUT: Duration = Duration::from_secs(30);
+pub(super) const RPC_TIMEOUT: Duration = Duration::from_secs(30);
 // TCP keepalive, since grpc-go and grpc-java servers close connections that send
 // HTTP/2 pings more often than every five minutes.
 const TCP_KEEPALIVE_IDLE: Duration = Duration::from_secs(30);
diff --git a/src/reapi/fake.rs b/src/reapi/fake.rs
new file mode 100644
index 0000000000..77f06b9c65
--- /dev/null
+++ b/src/reapi/fake.rs
@@ -0,0 +1,275 @@
+// Copyright 2026 Mozilla Foundation
+//
+// 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.
+
+//! An in-memory REAPI server for tests.
+
+use super::client::{Client, ConnectOptions};
+use super::digest;
+use super::proto::build::bazel::remote::execution::v2 as remexec;
+use super::proto::google::bytestream::byte_stream_server::{ByteStream, ByteStreamServer};
+use super::proto::google::bytestream::{
+    QueryWriteStatusRequest, QueryWriteStatusResponse, ReadRequest, ReadResponse, WriteRequest,
+    WriteResponse,
+};
+use futures::Stream;
+use remexec::action_cache_server::{ActionCache, ActionCacheServer};
+use remexec::capabilities_server::{Capabilities, CapabilitiesServer};
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+use tokio::net::TcpListener;
+use tonic::codegen::BoxFuture;
+use tonic::transport::server::TcpIncoming;
+use tonic::{Request, Response, Status, Streaming};
+
+type RpcResult<T> = std::result::Result<Response<T>, Status>;
+
+/// A REAPI server that keeps blobs and action results in memory.
+#[derive(Default)]
+pub(crate) struct FakeServer {
+    pub(crate) instance_name: String,
+    pub(crate) supports_zstd_bytestream: bool,
+    /// Serves blobs with their first byte flipped.
+    pub(crate) corrupt_reads: AtomicBool,
+    /// Serves blobs with a byte too many.
+    pub(crate) oversize_reads: AtomicBool,
+    /// Answers reads with a stream whose first message is NOT_FOUND.
+    pub(crate) not_found_in_stream: AtomicBool,
+    pub(crate) fail_lookups: AtomicBool,
+    pub(crate) blobs: Mutex<HashMap<(String, i64), Vec<u8>>>,
+    pub(crate) action_results: Mutex<HashMap<(String, i64), remexec::ActionResult>>,
+}
+
+impl FakeServer {
+    pub(crate) fn insert_blob(&self, data: &[u8]) -> remexec::Digest {
+        let digest = digest::of_bytes(data);
+        self.blobs
+            .lock()
+            .unwrap()
+            .insert((digest.hash.clone(), digest.size_bytes), data.to_vec());
+        digest
+    }
+
+    pub(crate) fn insert_action_result(
+        &self,
+        action_digest: &remexec::Digest,
+        result: remexec::ActionResult,
+    ) {
+        self.action_results.lock().unwrap().insert(
+            (action_digest.hash.clone(), action_digest.size_bytes),
+            result,
+        );
+    }
+
+    /// Serves on a new local port and returns a [`Client`] connected with `instance_name`.
+    pub(crate) async fn connect(self: &Arc<Self>) -> Client {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let routes = Routes {
+            capabilities: CapabilitiesServer::from_arc(self.clone()),
+            bytestream: ByteStreamServer::from_arc(self.clone()),
+            action_cache: ActionCacheServer::from_arc(self.clone()),
+        };
+        tokio::spawn(
+            tonic::transport::Server::builder()
+                .serve_with_incoming(routes, TcpIncoming::from(listener)),
+        );
+        Client::connect(&ConnectOptions {
+            endpoint: format!("grpc://{addr}"),
+            instance_name: self.instance_name.clone(),
+            ..Default::default()
+        })
+        .await
+        .unwrap()
+    }
+
+    /// The digest in a ByteStream resource name, and whether the name asks for
+    /// zstd-compressed bytes.
+    fn parse_resource_name(
+        &self,
+        resource_name: &str,
+    ) -> std::result::Result<(remexec::Digest, bool), Status> {
+        let invalid = || Status::invalid_argument(resource_name.to_owned());
+        let blob_name = if self.instance_name.is_empty() {
+            resource_name
+        } else {
+            resource_name
+                .strip_prefix(&format!("{}/", self.instance_name))
+                .ok_or_else(invalid)?
+        };
+        let (is_zstd, hash_and_size) =
+            if let Some(hash_and_size) = blob_name.strip_prefix("compressed-blobs/zstd/") {
+                (true, hash_and_size)
+            } else if let Some(hash_and_size) = blob_name.strip_prefix("blobs/") {
+                (false, hash_and_size)
+            } else {
+                return Err(invalid());
+            };
+        let (hash, size) = hash_and_size.split_once('/').ok_or_else(invalid)?;
+        let size_bytes = size.parse().map_err(|_| invalid())?;
+        Ok((
+            remexec::Digest {
+                hash: hash.to_owned(),
+                size_bytes,
+            },
+            is_zstd,
+        ))
+    }
+}
+
+/// Routes each request by its gRPC path, since sccache builds tonic without its `router`
+/// feature.
+#[derive(Clone)]
+struct Routes {
+    capabilities: CapabilitiesServer<FakeServer>,
+    bytestream: ByteStreamServer<FakeServer>,
+    action_cache: ActionCacheServer<FakeServer>,
+}
+
+impl tower_service::Service<http::Request<tonic::body::Body>> for Routes {
+    type Response = http::Response<tonic::body::Body>;
+    type Error = std::convert::Infallible;
+    type Future = BoxFuture<Self::Response, Self::Error>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, request: http::Request<tonic::body::Body>) -> Self::Future {
+        let path = request.uri().path();
+        if path.starts_with("/build.bazel.remote.execution.v2.Capabilities/") {
+            self.capabilities.call(request)
+        } else if path.starts_with("/google.bytestream.ByteStream/") {
+            self.bytestream.call(request)
+        } else if path.starts_with("/build.bazel.remote.execution.v2.ActionCache/") {
+            self.action_cache.call(request)
+        } else {
+            let response = Status::unimplemented(path.to_owned()).into_http();
+            Box::pin(async move { Ok(response) })
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl Capabilities for FakeServer {
+    async fn get_capabilities(
+        &self,
+        _: Request<remexec::GetCapabilitiesRequest>,
+    ) -> RpcResult<remexec::ServerCapabilities> {
+        let supported_compressors = if self.supports_zstd_bytestream {
+            vec![remexec::compressor::Value::Zstd as i32]
+        } else {
+            vec![]
+        };
+        Ok(Response::new(remexec::ServerCapabilities {
+            cache_capabilities: Some(remexec::CacheCapabilities {
+                supported_compressors,
+                ..Default::default()
+            }),
+            ..Default::default()
+        }))
+    }
+}
+
+#[tonic::async_trait]
+impl ActionCache for FakeServer {
+    async fn get_action_result(
+        &self,
+        request: Request<remexec::GetActionResultRequest>,
+    ) -> RpcResult<remexec::ActionResult> {
+        if self.fail_lookups.load(Ordering::SeqCst) {
+            return Err(Status::internal("the action cache is on fire"));
+        }
+        let request = request.into_inner();
+        if request.instance_name != self.instance_name {
+            return Err(Status::invalid_argument(request.instance_name));
+        }
+        let Some(action_digest) = request.action_digest else {
+            return Err(Status::invalid_argument("no action_digest"));
+        };
+        let action_results = self.action_results.lock().unwrap();
+        match action_results.get(&(action_digest.hash, action_digest.size_bytes)) {
+            Some(result) => Ok(Response::new(result.clone())),
+            None => Err(Status::not_found("no result")),
+        }
+    }
+
+    async fn update_action_result(
+        &self,
+        _: Request<remexec::UpdateActionResultRequest>,
+    ) -> RpcResult<remexec::ActionResult> {
+        Err(Status::unimplemented("UpdateActionResult"))
+    }
+}
+
+#[tonic::async_trait]
+impl ByteStream for FakeServer {
+    type ReadStream = Pin<Box<dyn Stream<Item = std::result::Result<ReadResponse, Status>> + Send>>;
+
+    async fn read(&self, request: Request<ReadRequest>) -> RpcResult<Self::ReadStream> {
+        let request = request.into_inner();
+        if request.read_offset != 0 || request.read_limit != 0 {
+            return Err(Status::invalid_argument("partial reads not supported"));
+        }
+        if self.not_found_in_stream.load(Ordering::SeqCst) {
+            let not_found = Err(Status::not_found("blob"));
+            return Ok(Response::new(Box::pin(futures::stream::iter([not_found]))));
+        }
+        let (digest, is_zstd) = self.parse_resource_name(&request.resource_name)?;
+        if is_zstd && !self.supports_zstd_bytestream {
+            return Err(Status::invalid_argument("compressed-blobs not offered"));
+        }
+        let Some(mut data) = self
+            .blobs
+            .lock()
+            .unwrap()
+            .get(&(digest.hash.clone(), digest.size_bytes))
+            .cloned()
+        else {
+            return Err(Status::not_found(digest.hash));
+        };
+        if self.corrupt_reads.load(Ordering::SeqCst) {
+            data[0] ^= 1;
+        }
+        if self.oversize_reads.load(Ordering::SeqCst) {
+            data.push(0);
+        }
+        if is_zstd {
+            data = zstd::bulk::compress(&data, 1).map_err(|e| Status::internal(e.to_string()))?;
+        }
+        // Several chunks, as servers send large blobs.
+        let chunks: Vec<_> = data
+            .chunks(1000)
+            .map(|chunk| {
+                Ok(ReadResponse {
+                    data: chunk.to_vec(),
+                })
+            })
+            .collect();
+        Ok(Response::new(Box::pin(futures::stream::iter(chunks))))
+    }
+
+    async fn write(&self, _: Request<Streaming<WriteRequest>>) -> RpcResult<WriteResponse> {
+        Err(Status::unimplemented("Write"))
+    }
+
+    async fn query_write_status(
+        &self,
+        _: Request<QueryWriteStatusRequest>,
+    ) -> RpcResult<QueryWriteStatusResponse> {
+        Err(Status::unimplemented("QueryWriteStatus"))
+    }
+}
diff --git a/src/reapi/mod.rs b/src/reapi/mod.rs
index ffb92651a5..4a8792e3e2 100644
--- a/src/reapi/mod.rs
+++ b/src/reapi/mod.rs
@@ -15,7 +15,11 @@
 //! A client for the Remote Execution API (REAPI), Bazel's protocol for remote caching
 //! and execution.
 
+mod action_cache;
+mod cas;
 mod client;
 mod digest;
+#[cfg(test)]
+mod fake;
 mod input_root;
 pub(crate) mod proto;