diff --git a/.github/actions/build_example_persistent_worker/action.yml b/.github/actions/build_example_persistent_worker/action.yml
new file mode 100644
index 0000000000000..2943b0a84ca4f
--- /dev/null
+++ b/.github/actions/build_example_persistent_worker/action.yml
@@ -0,0 +1,16 @@
+name: build_example_persistent_worker
+inputs:
+  buildbuddyApiKey:
+    description: "The API key for BuildBuddy remote cache and execution."
+    required: true
+runs:
+  using: composite
+  steps:
+  - name: Build examples/persistent_worker directory
+    env:
+      BUILDBUDDY_API_KEY: ${{ inputs.buildbuddyApiKey }}
+    run: |-
+      cd examples/persistent_worker
+      export PATH="$RUNNER_TEMP/artifacts:$PATH"
+      ./test.sh
+    shell: bash
diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml
index 089de65260bcd..885d43332813a 100644
--- a/.github/workflows/build-and-test.yml
+++ b/.github/workflows/build-and-test.yml
@@ -2,6 +2,7 @@ name: Build and test
 on:
   push:
   pull_request:
+  workflow_dispatch: # allows manual triggering
 jobs:
   linux-build-and-test:
     runs-on: 4-core-ubuntu
@@ -69,6 +70,9 @@ jobs:
         $RUNNER_TEMP/artifacts/buck2 test //... -v 2
     - uses: ./.github/actions/build_example_conan
     - uses: ./.github/actions/build_example_no_prelude
+    - uses: ./.github/actions/build_example_persistent_worker
+      with:
+        buildbuddyApiKey: ${{ secrets.BUILDBUDDY_API_KEY }}
     - uses: ./.github/actions/setup_reindeer
     - uses: ./.github/actions/build_bootstrap
   windows-build-examples:
diff --git a/app/buck2_action_impl/src/actions/impls/run.rs b/app/buck2_action_impl/src/actions/impls/run.rs
index c1307b4aa113e..7627fbb1a5069 100644
--- a/app/buck2_action_impl/src/actions/impls/run.rs
+++ b/app/buck2_action_impl/src/actions/impls/run.rs
@@ -40,6 +40,7 @@ use buck2_core::execution_types::executor_config::RemoteExecutorDependency;
 use buck2_core::fs::artifact_path_resolver::ArtifactFs;
 use buck2_core::fs::buck_out_path::BuildArtifactPath;
 use buck2_core::fs::paths::forward_rel_path::ForwardRelativePathBuf;
+use buck2_error::buck2_error;
 use buck2_error::BuckErrorContext;
 use buck2_events::dispatch::span_async_simple;
 use buck2_execute::artifact::fs::ExecutorFs;
@@ -263,6 +264,7 @@ struct UnpackedWorkerValues<'v> {
     id: WorkerId,
     concurrency: Option<usize>,
     streaming: bool,
+    supports_bazel_remote_persistent_worker_protocol: bool,
 }
 
 struct UnpackedRunActionValues<'v> {
@@ -320,6 +322,7 @@ impl RunAction {
             id: WorkerId(worker.id),
             concurrency: worker.concurrency(),
             streaming: worker.streaming(),
+            supports_bazel_remote_persistent_worker_protocol: worker.supports_bazel_remote_persistent_worker_protocol(),
         });
 
         Ok(UnpackedRunActionValues {
@@ -352,11 +355,32 @@ impl RunAction {
                 .exe
                 .add_to_command_line(&mut worker_rendered, &mut cli_ctx)?;
             worker.exe.visit_artifacts(artifact_visitor)?;
+            let worker_key = if worker.supports_bazel_remote_persistent_worker_protocol {
+                let mut worker_visitor = SimpleCommandLineArtifactVisitor::new();
+                worker.exe.visit_artifacts(&mut worker_visitor)?;
+                if !worker_visitor.outputs.is_empty() {
+                    // TODO[AH] create appropriate error enum value.
+                    return Err(buck2_error!(
+                        buck2_error::ErrorTag::Input,
+                        "remote persistent worker command should not produce an output"
+                    ));
+                }
+                let worker_inputs: Vec<&ArtifactGroupValues> = worker_visitor
+                    .inputs()
+                    .map(|group| action_execution_ctx.artifact_values(group))
+                    .collect();
+                let (_, worker_digest) =
+                    metadata_content(fs.fs(), &worker_inputs, action_execution_ctx.digest_config())?;
+                Some(worker_digest)
+            } else {
+                None
+            };
             Some(WorkerSpec {
                 exe: worker_rendered,
                 id: worker.id,
                 concurrency: worker.concurrency,
                 streaming: worker.streaming,
+                remote_key: worker_key,
             })
         } else {
             None
diff --git a/app/buck2_build_api/src/actions/execute/action_executor.rs b/app/buck2_build_api/src/actions/execute/action_executor.rs
index 8297c9ca28566..c59a496ee874b 100644
--- a/app/buck2_build_api/src/actions/execute/action_executor.rs
+++ b/app/buck2_build_api/src/actions/execute/action_executor.rs
@@ -761,6 +761,7 @@ mod tests {
                 CommandGenerationOptions {
                     path_separator: PathSeparatorKind::Unix,
                     output_paths_behavior: Default::default(),
+                    use_bazel_protocol_remote_persistent_workers: false,
                 },
                 Default::default(),
             ),
diff --git a/app/buck2_build_api/src/interpreter/rule_defs/command_executor_config.rs b/app/buck2_build_api/src/interpreter/rule_defs/command_executor_config.rs
index ba65c66f2bf4b..51a1ea8f38fdf 100644
--- a/app/buck2_build_api/src/interpreter/rule_defs/command_executor_config.rs
+++ b/app/buck2_build_api/src/interpreter/rule_defs/command_executor_config.rs
@@ -86,6 +86,7 @@ pub fn register_command_executor_config(builder: &mut GlobalsBuilder) {
     /// * `allow_hybrid_fallbacks_on_failure`: Whether to allow fallbacks when the result is failure (i.e. the command failed on the primary, but the infra worked)
     /// * `use_windows_path_separators`: Whether to use Windows path separators in command line arguments
     /// * `use_persistent workers`: Whether to use persistent workers for local execution if they are available
+    /// * `use_bazel_protocol_remote_persistent_workers`: Whether to use persistent workers for remote execution via the Bazel remote persistent worker protocol if they are available
     /// * `allow_cache_uploads`: Whether to upload local actions to the RE cache
     /// * `max_cache_upload_mebibytes`: Maximum size to upload in cache uploads
     /// * `experimental_low_pass_filter`: Whether to use the experimental low pass filter
@@ -111,6 +112,7 @@ pub fn register_command_executor_config(builder: &mut GlobalsBuilder) {
         #[starlark(default = false, require = named)] allow_hybrid_fallbacks_on_failure: bool,
         #[starlark(default = false, require = named)] use_windows_path_separators: bool,
         #[starlark(default = false, require = named)] use_persistent_workers: bool,
+        #[starlark(default = false, require = named)] use_bazel_protocol_remote_persistent_workers: bool,
         #[starlark(default = false, require = named)] allow_cache_uploads: bool,
         #[starlark(default = NoneOr::None, require = named)] max_cache_upload_mebibytes: NoneOr<
             i32,
@@ -323,6 +325,7 @@ pub fn register_command_executor_config(builder: &mut GlobalsBuilder) {
                         PathSeparatorKind::Unix
                     },
                     output_paths_behavior,
+                    use_bazel_protocol_remote_persistent_workers,
                 },
             }
         };
diff --git a/app/buck2_build_api/src/interpreter/rule_defs/provider/builtin/worker_info.rs b/app/buck2_build_api/src/interpreter/rule_defs/provider/builtin/worker_info.rs
index 51753d1be41a0..9106c0891e822 100644
--- a/app/buck2_build_api/src/interpreter/rule_defs/provider/builtin/worker_info.rs
+++ b/app/buck2_build_api/src/interpreter/rule_defs/provider/builtin/worker_info.rs
@@ -50,6 +50,8 @@ pub struct WorkerInfoGen<V: ValueLifetimeless> {
     pub concurrency: ValueOfUncheckedGeneric<V, NoneOr<usize>>,
     // Whether to always run actions using this worker via the streaming API
     pub streaming: ValueOfUncheckedGeneric<V, bool>,
+    // Bazel remote persistent worker protocol capable worker.
+    pub supports_bazel_remote_persistent_worker_protocol: ValueOfUncheckedGeneric<V, bool>,
 
     pub id: u64,
 }
@@ -68,6 +70,7 @@ fn worker_info_creator(globals: &mut GlobalsBuilder) {
             ValueOf<'v, usize>,
         >,
         #[starlark(require = named, default = NoneType)] streaming: Value<'v>,
+        #[starlark(require = named, default = false)] supports_bazel_remote_persistent_worker_protocol: bool,
         eval: &mut Evaluator<'v, '_, '_>,
     ) -> starlark::Result<WorkerInfo<'v>> {
         let heap = eval.heap();
@@ -79,6 +82,7 @@ fn worker_info_creator(globals: &mut GlobalsBuilder) {
             id,
             concurrency: heap.alloc_typed_unchecked(concurrency).cast(),
             streaming: ValueOfUnchecked::new(streaming),
+            supports_bazel_remote_persistent_worker_protocol: heap.alloc_typed_unchecked(supports_bazel_remote_persistent_worker_protocol).cast(),
         })
     }
 }
@@ -105,6 +109,13 @@ impl<'v, V: ValueLike<'v>> WorkerInfoGen<V> {
             .into_option()
             .unwrap_or(false)
     }
+
+    pub fn supports_bazel_remote_persistent_worker_protocol(&self) -> bool {
+        self.supports_bazel_remote_persistent_worker_protocol
+            .to_value()
+            .unpack()
+            .expect("validated at construction")
+    }
 }
 
 fn validate_worker_info<'v, V>(info: &WorkerInfoGen<V>) -> buck2_error::Result<()>
diff --git a/app/buck2_build_api_tests/src/interpreter/rule_defs/provider/builtin/worker_info.rs b/app/buck2_build_api_tests/src/interpreter/rule_defs/provider/builtin/worker_info.rs
index 1124c9e1629bc..97b5ff92aaf40 100644
--- a/app/buck2_build_api_tests/src/interpreter/rule_defs/provider/builtin/worker_info.rs
+++ b/app/buck2_build_api_tests/src/interpreter/rule_defs/provider/builtin/worker_info.rs
@@ -24,7 +24,7 @@ fn run_display() {
         .run_starlark_bzl_test(
             r#"
 def test():
-    assert_eq('WorkerInfo(exe=cmd_args("x"), concurrency=None, streaming=None)', str(WorkerInfo(exe="x")))
+    assert_eq('WorkerInfo(exe=cmd_args("x"), concurrency=None, streaming=None, supports_bazel_remote_persistent_worker_protocol=False)', str(WorkerInfo(exe="x")))
 "#,
         )
         .unwrap();
diff --git a/app/buck2_core/src/execution_types/executor_config.rs b/app/buck2_core/src/execution_types/executor_config.rs
index 113bc92acece0..48c9406ea1954 100644
--- a/app/buck2_core/src/execution_types/executor_config.rs
+++ b/app/buck2_core/src/execution_types/executor_config.rs
@@ -283,6 +283,7 @@ impl Default for CacheUploadBehavior {
 pub struct CommandGenerationOptions {
     pub path_separator: PathSeparatorKind,
     pub output_paths_behavior: OutputPathsBehavior,
+    pub use_bazel_protocol_remote_persistent_workers: bool,
 }
 
 #[derive(Debug, Eq, PartialEq, Hash, Allocative, Clone)]
@@ -314,6 +315,7 @@ impl CommandExecutorConfig {
             options: CommandGenerationOptions {
                 path_separator: PathSeparatorKind::system_default(),
                 output_paths_behavior: Default::default(),
+                use_bazel_protocol_remote_persistent_workers: false,
             },
         })
     }
diff --git a/app/buck2_execute/src/execute/command_executor.rs b/app/buck2_execute/src/execute/command_executor.rs
index 1859e6fb00211..d8bed0d28dbc3 100644
--- a/app/buck2_execute/src/execute/command_executor.rs
+++ b/app/buck2_execute/src/execute/command_executor.rs
@@ -20,7 +20,6 @@ use buck2_core::fs::artifact_path_resolver::ArtifactFs;
 use buck2_core::fs::project_rel_path::ProjectRelativePath;
 use buck2_core::fs::project_rel_path::ProjectRelativePathBuf;
 use buck2_directory::directory::fingerprinted_directory::FingerprintedDirectory;
-#[cfg(fbcode_build)]
 use buck2_error::buck2_error;
 use buck2_error::BuckErrorContext;
 use buck2_futures::cancellation::CancellationContext;
@@ -187,15 +186,45 @@ impl CommandExecutor {
                 }
                 CommandExecutionInput::ScratchPath(_) => None,
             });
+            let mut platform = self.0.re_platform.clone();
+            let args = if self.0.options.use_bazel_protocol_remote_persistent_workers
+                && let Some(worker) = request.worker()
+                && let Some(key) = worker.remote_key.as_ref()
+            {
+                platform.properties.push(RE::Property {
+                    name: "persistentWorkerKey".to_owned(),
+                    value: key.to_string(),
+                });
+                // TODO[AH] Ideally, Buck2 could generate an argfile on the fly.
+                for arg in request.args() {
+                    if !(arg.starts_with("@")
+                        || arg.starts_with("-flagfile")
+                        || arg.starts_with("--flagfile"))
+                    {
+                        return Err(buck2_error!(
+                            buck2_error::ErrorTag::Input,
+                            "Remote persistent worker arguments must be passed as `@argfile`, `-flagfile=argfile`, or `--flagfile=argfile`."
+                        ));
+                    }
+                }
+                worker
+                    .exe
+                    .iter()
+                    .chain(request.args().iter())
+                    .cloned()
+                    .collect()
+            } else {
+                request.all_args_vec()
+            };
             let action = re_create_action(
-                request.all_args_vec(),
+                args,
                 request.paths().output_paths(),
                 request.working_directory(),
                 request.env(),
                 input_digest,
                 action_metadata_blobs,
                 request.timeout(),
-                self.0.re_platform.clone(),
+                platform,
                 false,
                 digest_config,
                 self.0.options.output_paths_behavior,
diff --git a/app/buck2_execute/src/execute/request.rs b/app/buck2_execute/src/execute/request.rs
index 7c17d4c946af0..a205892c4fac3 100644
--- a/app/buck2_execute/src/execute/request.rs
+++ b/app/buck2_execute/src/execute/request.rs
@@ -278,12 +278,13 @@ impl CommandExecutionPaths {
 #[derive(Copy, Clone, Dupe, Debug, Display, Allocative, Hash, PartialEq, Eq)]
 pub struct WorkerId(pub u64);
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct WorkerSpec {
     pub id: WorkerId,
     pub exe: Vec<String>,
     pub concurrency: Option<usize>,
     pub streaming: bool,
+    pub remote_key: Option<TrackedFileDigest>,
 }
 
 /// The data contains the information about the command to be executed.
diff --git a/app/buck2_execute/src/lib.rs b/app/buck2_execute/src/lib.rs
index aebb7770d0381..59bc110e4ae05 100644
--- a/app/buck2_execute/src/lib.rs
+++ b/app/buck2_execute/src/lib.rs
@@ -14,6 +14,7 @@
 #![feature(try_trait_v2)]
 #![feature(used_with_arg)]
 #![feature(trait_upcasting)]
+#![feature(let_chains)]
 
 pub mod artifact;
 pub mod artifact_utils;
diff --git a/app/buck2_server/src/daemon/common.rs b/app/buck2_server/src/daemon/common.rs
index 2bfa6204f9933..b39bfe0394a71 100644
--- a/app/buck2_server/src/daemon/common.rs
+++ b/app/buck2_server/src/daemon/common.rs
@@ -500,6 +500,7 @@ pub fn get_default_executor_config(host_platform: HostPlatformOverride) -> Comma
         options: CommandGenerationOptions {
             path_separator: get_default_path_separator(host_platform),
             output_paths_behavior: Default::default(),
+            use_bazel_protocol_remote_persistent_workers: false,
         },
     }
 }
diff --git a/app/buck2_test/src/orchestrator.rs b/app/buck2_test/src/orchestrator.rs
index b66144f307951..5c1184901b531 100644
--- a/app/buck2_test/src/orchestrator.rs
+++ b/app/buck2_test/src/orchestrator.rs
@@ -1191,6 +1191,7 @@ impl<'b> BuckTestOrchestrator<'b> {
             options: CommandGenerationOptions {
                 path_separator: PathSeparatorKind::system_default(),
                 output_paths_behavior: Default::default(),
+                use_bazel_protocol_remote_persistent_workers: false,
             },
         };
         let CommandExecutorResponse {
@@ -1733,6 +1734,7 @@ impl<'a> Execute2RequestExpander<'a> {
                     id: WorkerId(worker.id),
                     concurrency: worker.concurrency(),
                     streaming: worker.streaming(),
+                    remote_key: None,
                 })
             }
             _ => None,
diff --git a/examples/persistent_worker/.buckconfig b/examples/persistent_worker/.buckconfig
new file mode 100644
index 0000000000000..34a9d2e17746c
--- /dev/null
+++ b/examples/persistent_worker/.buckconfig
@@ -0,0 +1,17 @@
+[cells]
+root = .
+prelude = prelude
+toolchains = toolchains
+none = none
+
+[cell_aliases]
+config = prelude
+fbcode = none
+fbsource = none
+buck = none
+
+[external_cells]
+prelude = bundled
+
+[parser]
+target_platform_detector_spec = target:root//...->prelude//platforms:default
diff --git a/examples/persistent_worker/.buckconfig.buildbuddy b/examples/persistent_worker/.buckconfig.buildbuddy
new file mode 100644
index 0000000000000..66227e97615e4
--- /dev/null
+++ b/examples/persistent_worker/.buckconfig.buildbuddy
@@ -0,0 +1,13 @@
+[buck2]
+digest_algorithms = SHA256
+
+[buck2_re_client]
+engine_address       = grpc://remote.buildbuddy.io
+action_cache_address = grpc://remote.buildbuddy.io
+cas_address          = grpc://remote.buildbuddy.io
+tls                  = true
+http_headers         = \
+  x-buildbuddy-api-key:$BUILDBUDDY_API_KEY
+
+[build]
+execution_platforms = root//platforms:buildbuddy
diff --git a/examples/persistent_worker/.buckconfig.buildbuddy-persistent-workers b/examples/persistent_worker/.buckconfig.buildbuddy-persistent-workers
new file mode 100644
index 0000000000000..436af6c9a504e
--- /dev/null
+++ b/examples/persistent_worker/.buckconfig.buildbuddy-persistent-workers
@@ -0,0 +1,13 @@
+[buck2]
+digest_algorithms = SHA256
+
+[buck2_re_client]
+engine_address       = grpc://remote.buildbuddy.io
+action_cache_address = grpc://remote.buildbuddy.io
+cas_address          = grpc://remote.buildbuddy.io
+tls                  = true
+http_headers         = \
+  x-buildbuddy-api-key:$BUILDBUDDY_API_KEY
+
+[build]
+execution_platforms = root//platforms:buildbuddy-persistent-workers
diff --git a/examples/persistent_worker/.buckconfig.local-persistent-workers b/examples/persistent_worker/.buckconfig.local-persistent-workers
new file mode 100644
index 0000000000000..fef3def8a1f38
--- /dev/null
+++ b/examples/persistent_worker/.buckconfig.local-persistent-workers
@@ -0,0 +1,2 @@
+[build]
+execution_platforms = root//platforms:local-persistent-workers
diff --git a/examples/persistent_worker/.buckconfig.no-workers b/examples/persistent_worker/.buckconfig.no-workers
new file mode 100644
index 0000000000000..c731b17c77567
--- /dev/null
+++ b/examples/persistent_worker/.buckconfig.no-workers
@@ -0,0 +1,2 @@
+[build]
+execution_platforms = root//platforms:local
diff --git a/examples/persistent_worker/.buckroot b/examples/persistent_worker/.buckroot
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/examples/persistent_worker/.envrc b/examples/persistent_worker/.envrc
new file mode 100644
index 0000000000000..1ab1354014c7f
--- /dev/null
+++ b/examples/persistent_worker/.envrc
@@ -0,0 +1,3 @@
+# specify the following:
+# - BUILDBUDDY_API_KEY
+source_env_if_exists .envrc.private
diff --git a/examples/persistent_worker/.gitignore b/examples/persistent_worker/.gitignore
new file mode 100644
index 0000000000000..a82178e703cc5
--- /dev/null
+++ b/examples/persistent_worker/.gitignore
@@ -0,0 +1,4 @@
+.buckconfig.local
+.direnv
+.envrc.private
+prelude
diff --git a/examples/persistent_worker/BUCK b/examples/persistent_worker/BUCK
new file mode 100644
index 0000000000000..df7c7bba32410
--- /dev/null
+++ b/examples/persistent_worker/BUCK
@@ -0,0 +1,26 @@
+load("defs.bzl", "demo", "worker")
+
+python_binary(
+    name = "one_shot",
+    main = "one_shot.py",
+)
+
+python_binary(
+    name = "worker_py",
+    main = "persistent_worker.py",
+    deps = [
+        "//proto/bazel:worker_protocol_pb2",
+        "//proto/buck2:worker_pb2",
+    ],
+)
+
+worker(
+    name = "worker",
+    visibility = ["PUBLIC"],
+    worker = ":worker_py",
+)
+
+[
+    demo(name = "demo-" + str(i))
+    for i in range(4)
+]
diff --git a/examples/persistent_worker/README.md b/examples/persistent_worker/README.md
new file mode 100644
index 0000000000000..f381050cbcc3e
--- /dev/null
+++ b/examples/persistent_worker/README.md
@@ -0,0 +1,145 @@
+# Persistent Worker Demo
+
+At the time of writing (2024-09-25) Buck2 supports persistent workers for local
+builds through a dedicated Buck2 persistent worker gRPC protocol. However, Buck2
+does not support persistent workers for builds that use remote execution. This
+demo is part of a patch-set that adds support for remote persistent workers to
+Buck2, see [#776].
+
+[#776]: https://github.com/facebook/buck2/issues/776
+
+## Requirements
+
+This demo uses BuildBuddy remote execution to demonstrate remote persistent
+workers. You will need an API token for at least a free open source account. You
+can use [direnv] to set up the environment:
+
+Credentials for [BuildBuddy] stored in `.envrc.private`:
+
+```
+export BUILDBUDDY_API_KEY=...
+```
+
+On CI the API key is not available for pipelines initiated from forks of the
+main Buck2 repository. The corresponding tests will be skipped in that case. A
+Meta engineer can manually initiate a pipeline run with the token set.
+
+[direnv]: https://direnv.net/
+[BuildBuddy]: https://www.buildbuddy.io/
+
+## Local Build
+
+Configure a local build without persistent workers:
+
+```
+$ cd examples/persistent_worker
+$ echo '<file:.buckconfig.no-workers>' > .buckconfig.local
+```
+
+Run a clean build:
+
+```
+$ buck2 clean; buck2 build : -vstderr
+...
+stderr for root//:demo-7 (demo):
+...
+ONE-SHOT START
+...
+```
+
+## Local Persistent Worker
+
+Configure a local build with persistent workers:
+
+```
+$ cd examples/persistent_worker
+$ echo '<file:.buckconfig.local-persistent-workers>' > .buckconfig.local
+```
+
+Run a clean build:
+
+```
+$ buck2 clean; buck2 build : -vstderr
+...
+stderr for root//:demo-7 (demo):
+...
+Buck2 persistent worker ...
+...
+```
+
+## Remote Execution
+
+Configure a remote build without persistent workers:
+
+```
+$ cd examples/persistent_worker
+$ echo '<file:.buckconfig.buildbuddy>' > .buckconfig.local
+```
+
+Run a clean build:
+
+```
+$ buck2 clean; buck2 build : -vstderr
+...
+stderr for root//:demo-7 (demo):
+...
+ONE-SHOT START
+...
+```
+
+## Remote Persistent Worker
+
+Configure a remote build with persistent workers:
+
+```
+$ cd examples/persistent_worker
+$ echo '<file:.buckconfig.buildbuddy-persistent-workers>' > .buckconfig.local
+```
+
+Run a clean build:
+
+```
+$ buck2 clean; buck2 build : -vstderr
+...
+stderr for root//:demo-7 (demo):
+...
+Bazel persistent worker ...
+...
+```
+
+## Protocol
+
+### Starlark
+
+A Buck2 persistent worker is created by a rule that emits the `WorkerInfo`
+provider. Setting `remote = True` on this provider indicates that this worker is
+remote execution capable.
+
+Buck2 actions indicate that they can utilize a persistent worker by setting the
+`exe` parameter to `ctx.actions.run` to `WorkerRunInfo(worker, exe)`, where
+`worker` is a `WorkerInfo` provider, and `exe` defines the fallback executable
+for non persistent-worker execution.
+
+Buck2 actions that want to utilize a remote persistent worker must pass
+command-line arguments in an argument file specified as `@argfile`,
+`-flagfile=argfile`, or `--flagfile=argfile` on the command-line.
+
+### Local Persistent Worker
+
+A locally executed Buck2 persistent worker falls under the
+[Buck2 persistent worker protocol](./proto/buck2/worker.proto): It is started
+and managed by Buck2 and passed a file path in the `WORKER_SOCKET` environment
+variable where it should create a gRPC Unix domain socket to serve worker
+requests over. Multiple requests may be sent in parallel and expected to be
+served at the same time depending on the `concurrency` attribute of the
+`WorkerInfo` provider.
+
+### Remote Persistent Worker
+
+A remotely executed Buck2 persistent worker falls under the
+[Bazel persistent worker protocol](./proto/bazel/worker_protocol.proto): It is
+started and managed by the remote execution system. Work requests are sent as
+length prefixed protobuf objects to the standard input of the worker process.
+Work responses are expected as length prefixed protobuf objects on the standard
+output of the worker process. The worker process may not use standard output for
+anything else.
diff --git a/examples/persistent_worker/defs.bzl b/examples/persistent_worker/defs.bzl
new file mode 100644
index 0000000000000..55e978a2f3d18
--- /dev/null
+++ b/examples/persistent_worker/defs.bzl
@@ -0,0 +1,63 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+load("@prelude//utils:argfile.bzl", "at_argfile")
+
+def _worker_impl(ctx: AnalysisContext) -> list[Provider]:
+    return [
+        DefaultInfo(),
+        WorkerInfo(
+            exe = ctx.attrs.worker[RunInfo].args,
+            concurrency = None,
+            supports_bazel_remote_persistent_worker_protocol = True,
+        ),
+    ]
+
+worker = rule(
+    impl = _worker_impl,
+    attrs = {
+        "worker": attrs.dep(providers = [RunInfo]),
+    },
+)
+
+def _demo_impl(ctx: AnalysisContext) -> list[Provider]:
+    output = ctx.actions.declare_output(ctx.label.name)
+    argfile = at_argfile(
+        actions = ctx.actions,
+        name = "demo." + ctx.label.name + ".args",
+        args = cmd_args(output.as_output()),
+    )
+    ctx.actions.run(
+        cmd_args(argfile),
+        category = "demo",
+        env = {
+            # modify this value to force an action rerun even if caching is enabled.
+            # `--no-remote-cache` does not have the desired effect, because it also causes
+            # the action to be omitted from `buck2 log what-ran`, which interferes with the
+            # test setup.
+            "CACHE_SILO_KEY": read_root_config("build", "cache_silo_key", "0"),
+        },
+        exe = WorkerRunInfo(
+            worker = ctx.attrs._worker[WorkerInfo],
+            exe = ctx.attrs._one_shot[RunInfo].args,
+        ),
+    )
+    return [DefaultInfo(default_output = output)]
+
+demo = rule(
+    impl = _demo_impl,
+    attrs = {
+        "_one_shot": attrs.exec_dep(
+            default = "//:one_shot",
+            providers = [RunInfo],
+        ),
+        "_worker": attrs.exec_dep(
+            default = "//:worker",
+            providers = [WorkerInfo],
+        ),
+    },
+)
diff --git a/examples/persistent_worker/one_shot.py b/examples/persistent_worker/one_shot.py
new file mode 100644
index 0000000000000..6efa97162908a
--- /dev/null
+++ b/examples/persistent_worker/one_shot.py
@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+import argparse
+import os
+import sys
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        fromfile_prefix_chars="@", prog="one_shot", description="One-shot command"
+    )
+    parser.add_argument("outfile", type=argparse.FileType("w"), help="Output file.")
+
+    args = parser.parse_args()
+
+    print("one-shot.py", file=sys.stderr)
+    print("ONE-SHOT START", file=sys.stderr)
+    name = os.path.basename(args.outfile.name)
+    args.outfile.write(name + "\n")
+    args.outfile.close()
+    print("ONE-SHOT END", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/persistent_worker/persistent_worker.py b/examples/persistent_worker/persistent_worker.py
new file mode 100644
index 0000000000000..931971280ba2c
--- /dev/null
+++ b/examples/persistent_worker/persistent_worker.py
@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+"""Buck2 local and remote persistent worker and action runner.
+
+This script can:
+- Execute build requests as a Buck2 local persistent worker.
+- Execute build requests as a remote persistent worker through Bazel protocol.
+"""
+
+import argparse
+import os
+import shlex
+import socket
+import sys
+import time
+from concurrent import futures
+from dataclasses import dataclass
+
+import google.protobuf.proto as proto
+import grpc
+import proto.bazel.worker_protocol_pb2 as bazel_pb2
+import proto.buck2.worker_pb2 as buck2_pb2
+import proto.buck2.worker_pb2_grpc as buck2_pb2_grpc
+
+
+@dataclass
+class Request:
+    """Universal worker request, independent of Buck2 or Bazel protocol."""
+
+    argv: list[str]
+
+
+@dataclass
+class Response:
+    """Universal worker response, independent of Buck2 or Bazel protocol."""
+
+    exit_code: int
+    stderr: str
+
+
+class ArgumentParserError(Exception):
+    pass
+
+
+class RecoverableArgumentParser(argparse.ArgumentParser):
+    def error(self, message):
+        raise ArgumentParserError(f"{self.prog}: error: {message}\n")
+
+
+class Implementation:
+    """Universal worker implementation, independent of Buck2 or Bazel protocol."""
+
+    def __init__(self):
+        self.parser = RecoverableArgumentParser(
+            fromfile_prefix_chars="@",
+            prog="worker_py_handler",
+            description="Persistent Worker Request Handler",
+        )
+        self.parser.add_argument(
+            "outfile", type=argparse.FileType("w"), help="Output file."
+        )
+
+    def execute(self, request: Request) -> Response:
+        try:
+            print(
+                "WORKER",
+                socket.gethostname(),
+                os.getpid(),
+                os.getcwd(),
+                file=sys.stderr,
+            )
+            print("REQUEST", request.argv, file=sys.stderr)
+            args = self.parser.parse_args(request.argv)
+            print("ARGS", args, file=sys.stderr)
+            output = args.outfile
+            name = os.path.basename(output.name)
+            print("WRITE", name, file=sys.stderr)
+            output.write(name + "\n")
+            print("SLEEP", name, file=sys.stderr)
+            time.sleep(1)
+            print("COMPLETED", name, file=sys.stderr)
+            output.close()
+            return Response(exit_code=0, stderr=f"wrote to {output.name}")
+        except ArgumentParserError as e:
+            return Response(exit_code=2, stderr=str(e))
+
+
+class Buck2Servicer(buck2_pb2_grpc.WorkerServicer):
+    """Buck2 remote persistent worker implementation."""
+
+    def __init__(self):
+        self.impl = Implementation()
+
+    def Execute(self, request, context):
+        _ = context
+        print("BUCK2", request, file=sys.stderr)
+        # Decode arguments as UTF-8 strings.
+        argv = [arg.decode("utf-8") for arg in request.argv]
+        response = self.impl.execute(Request(argv=argv))
+        host = socket.gethostname()
+        pid = os.getpid()
+        cwd = os.getcwd()
+        return buck2_pb2.ExecuteResponse(
+            exit_code=response.exit_code,
+            stderr=f"Buck2 persistent worker {host} {pid} {cwd}\n" + response.stderr,
+        )
+
+
+class BazelServicer:
+    def __init__(self):
+        self.impl = Implementation()
+
+    def Execute(self, request: bazel_pb2.WorkRequest) -> bazel_pb2.WorkResponse:
+        print("BAZEL", request, file=sys.stderr)
+        response = self.impl.execute(Request(argv=request.arguments))
+        host = socket.gethostname()
+        pid = os.getpid()
+        cwd = os.getcwd()
+        return bazel_pb2.WorkResponse(
+            exit_code=response.exit_code,
+            output=f"Bazel persistent worker {host} {pid} {cwd} {request.request_id}\n"
+            + response.stderr,
+            request_id=request.request_id,
+        )
+
+
+def main():
+    print("MAIN", socket.gethostname(), os.getpid(), os.getcwd(), file=sys.stderr)
+    parser = argparse.ArgumentParser(
+        fromfile_prefix_chars="@",
+        prog="worker",
+        description="Buck2/Bazel Local/Remote Persistent Worker",
+    )
+    parser.add_argument(
+        "--persistent_worker",
+        action="store_true",
+        help="Enable persistent worker (Bazel protocol).",
+    )
+
+    (args, rest) = parser.parse_known_args()
+
+    if socket_path := os.getenv("WORKER_SOCKET"):
+        # Buck2 persistent worker mode
+        print("BUCK2 WORKER START", file=sys.stderr)
+        if rest:
+            rest_joined = " ".join(map(shlex.quote, rest))
+            print(f"Unexpected arguments: {rest_joined}\n", file=sys.stderr)
+            parser.print_usage()
+            sys.exit(2)
+
+        server = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=os.cpu_count() or 1)
+        )
+        buck2_pb2_grpc.add_WorkerServicer_to_server(Buck2Servicer(), server)
+        server.add_insecure_port(f"unix://{socket_path}")
+        server.start()
+        server.wait_for_termination()
+    elif args.persistent_worker:
+        # Bazel persistent worker mode
+        print("BAZEL WORKER START", file=sys.stderr)
+        if rest:
+            rest_joined = " ".join(map(shlex.quote, rest))
+            print(f"Unexpected arguments: {rest_joined}\n", file=sys.stderr)
+            parser.print_usage()
+            sys.exit(2)
+
+        servicer = BazelServicer()
+        # uses length prefixed serialization features added in proto version 5.28.0.
+        # https://github.com/protocolbuffers/protobuf/pull/16965
+        while request := proto.parse_length_prefixed(
+            bazel_pb2.WorkRequest, sys.stdin.buffer
+        ):
+            response = servicer.Execute(request)
+            proto.serialize_length_prefixed(response, sys.stdout.buffer)
+            sys.stdout.flush()
+    else:
+        print(
+            "Expected either 'WORKER_SOCKET' environment variable or '--persistent_worker' argument.",
+            file=sys.stderr,
+        )
+        sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/persistent_worker/platforms/BUCK b/examples/persistent_worker/platforms/BUCK
new file mode 100644
index 0000000000000..f1d81b78c4b28
--- /dev/null
+++ b/examples/persistent_worker/platforms/BUCK
@@ -0,0 +1,34 @@
+load(":buildbuddy.bzl", "buildbuddy")
+load(":local.bzl", "local")
+
+host_cpu = "prelude//cpu:" + ("arm64" if host_info().arch.is_aarch64 else "x86_64")
+
+host_os = "prelude//os:" + ("macos" if host_info().os.is_macos else "linux")
+
+buildbuddy(
+    name = "buildbuddy",
+    cpu_configuration = host_cpu,
+    os_configuration = host_os,
+    use_persistent_workers = False,
+)
+
+buildbuddy(
+    name = "buildbuddy-persistent-workers",
+    cpu_configuration = host_cpu,
+    os_configuration = host_os,
+    use_persistent_workers = True,
+)
+
+local(
+    name = "local",
+    cpu_configuration = host_cpu,
+    os_configuration = host_os,
+    use_persistent_workers = False,
+)
+
+local(
+    name = "local-persistent-workers",
+    cpu_configuration = host_cpu,
+    os_configuration = host_os,
+    use_persistent_workers = True,
+)
diff --git a/examples/persistent_worker/platforms/buildbuddy.bzl b/examples/persistent_worker/platforms/buildbuddy.bzl
new file mode 100644
index 0000000000000..a913ea1aae165
--- /dev/null
+++ b/examples/persistent_worker/platforms/buildbuddy.bzl
@@ -0,0 +1,44 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+def _platforms(ctx):
+    constraints = dict()
+    constraints.update(ctx.attrs.cpu_configuration[ConfigurationInfo].constraints)
+    constraints.update(ctx.attrs.os_configuration[ConfigurationInfo].constraints)
+    configuration = ConfigurationInfo(constraints = constraints, values = {})
+
+    platform = ExecutionPlatformInfo(
+        label = ctx.label.raw_target(),
+        configuration = configuration,
+        executor_config = CommandExecutorConfig(
+            local_enabled = True,
+            remote_enabled = True,
+            remote_cache_enabled = True,
+            allow_cache_uploads = True,
+            use_limited_hybrid = True,
+            use_persistent_workers = ctx.attrs.use_persistent_workers,
+            use_bazel_protocol_remote_persistent_workers = ctx.attrs.use_persistent_workers,
+            remote_execution_properties = {
+                "OSFamily": "Linux",
+                "nonroot-workspace": True,
+                "recycle-runner": True,  # required for remote persistent workers
+            },
+            remote_execution_use_case = "buck2-default",
+            remote_output_paths = "output_paths",
+        ),
+    )
+
+    return [DefaultInfo(), ExecutionPlatformRegistrationInfo(platforms = [platform])]
+
+buildbuddy = rule(
+    attrs = {
+        "cpu_configuration": attrs.dep(providers = [ConfigurationInfo]),
+        "os_configuration": attrs.dep(providers = [ConfigurationInfo]),
+        "use_persistent_workers": attrs.bool(default = False),
+    },
+    impl = _platforms,
+)
diff --git a/examples/persistent_worker/platforms/local.bzl b/examples/persistent_worker/platforms/local.bzl
new file mode 100644
index 0000000000000..15e43a3d57785
--- /dev/null
+++ b/examples/persistent_worker/platforms/local.bzl
@@ -0,0 +1,36 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+def _platforms(ctx):
+    constraints = dict()
+    constraints.update(ctx.attrs.cpu_configuration[ConfigurationInfo].constraints)
+    constraints.update(ctx.attrs.os_configuration[ConfigurationInfo].constraints)
+    configuration = ConfigurationInfo(constraints = constraints, values = {})
+
+    platform = ExecutionPlatformInfo(
+        label = ctx.label.raw_target(),
+        configuration = configuration,
+        executor_config = CommandExecutorConfig(
+            local_enabled = True,
+            remote_enabled = False,
+            remote_cache_enabled = False,
+            allow_cache_uploads = False,
+            use_persistent_workers = ctx.attrs.use_persistent_workers,
+            use_bazel_protocol_remote_persistent_workers = False,
+        ),
+    )
+
+    return [DefaultInfo(), ExecutionPlatformRegistrationInfo(platforms = [platform])]
+
+local = rule(
+    attrs = {
+        "cpu_configuration": attrs.dep(providers = [ConfigurationInfo]),
+        "os_configuration": attrs.dep(providers = [ConfigurationInfo]),
+        "use_persistent_workers": attrs.bool(default = False),
+    },
+    impl = _platforms,
+)
diff --git a/examples/persistent_worker/proto/BUCK b/examples/persistent_worker/proto/BUCK
new file mode 100644
index 0000000000000..fef4f5e675616
--- /dev/null
+++ b/examples/persistent_worker/proto/BUCK
@@ -0,0 +1,6 @@
+python_binary(
+    name = "protoc",
+    main_module = "grpc_tools.protoc",
+    visibility = ["PUBLIC"],
+    deps = ["//python:grpcio-tools"],
+)
diff --git a/examples/persistent_worker/proto/bazel/BUCK b/examples/persistent_worker/proto/bazel/BUCK
new file mode 100644
index 0000000000000..5728ce76db4af
--- /dev/null
+++ b/examples/persistent_worker/proto/bazel/BUCK
@@ -0,0 +1,7 @@
+load("//proto:defs.bzl", "proto_python_library")
+
+proto_python_library(
+    name = "worker_protocol_pb2",
+    src = "worker_protocol.proto",
+    visibility = ["PUBLIC"],
+)
diff --git a/examples/persistent_worker/proto/bazel/worker_protocol.proto b/examples/persistent_worker/proto/bazel/worker_protocol.proto
new file mode 100644
index 0000000000000..4bca44874b5ed
--- /dev/null
+++ b/examples/persistent_worker/proto/bazel/worker_protocol.proto
@@ -0,0 +1,98 @@
+// Copyright 2015 The Bazel Authors. All rights reserved.
+//
+// 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.
+
+syntax = "proto3";
+
+package blaze.worker;
+
+option java_package = "com.google.devtools.build.lib.worker";
+
+// An input file.
+message Input {
+  // The path in the file system where to read this input artifact from. This is
+  // either a path relative to the execution root (the worker process is
+  // launched with the working directory set to the execution root), or an
+  // absolute path.
+  string path = 1;
+
+  // A hash-value of the contents. The format of the contents is unspecified and
+  // the digest should be treated as an opaque token. This can be empty in some
+  // cases.
+  bytes digest = 2;
+}
+
+// This represents a single work unit that Blaze sends to the worker.
+message WorkRequest {
+  repeated string arguments = 1;
+
+  // The inputs that the worker is allowed to read during execution of this
+  // request.
+  repeated Input inputs = 2;
+
+  // Each WorkRequest must have either a unique
+  // request_id or request_id = 0. If request_id is 0, this WorkRequest must be
+  // processed alone (singleplex), otherwise the worker may process multiple
+  // WorkRequests in parallel (multiplexing). As an exception to the above, if
+  // the cancel field is true, the request_id must be the same as a previously
+  // sent WorkRequest. The request_id must be attached unchanged to the
+  // corresponding WorkResponse. Only one singleplex request may be sent to a
+  // worker at a time.
+  int32 request_id = 3;
+
+  // EXPERIMENTAL: When true, this is a cancel request, indicating that a
+  // previously sent WorkRequest with the same request_id should be cancelled.
+  // The arguments and inputs fields must be empty and should be ignored.
+  bool cancel = 4;
+
+  // Values greater than 0 indicate that the worker may output extra debug
+  // information to stderr (which will go into the worker log). Setting the
+  // --worker_verbose flag for Bazel makes this flag default to 10.
+  int32 verbosity = 5;
+
+  // The relative directory inside the workers working directory where the
+  // inputs and outputs are placed, for sandboxing purposes. For singleplex
+  // workers, this is unset, as they can use their working directory as sandbox.
+  // For multiplex workers, this will be set when the
+  // --experimental_worker_multiplex_sandbox flag is set _and_ the execution
+  // requirements for the worker includes 'supports-multiplex-sandbox'.
+  // The paths in `inputs` will not contain this prefix, but the actual files
+  // will be placed/must be written relative to this directory. The worker
+  // implementation is responsible for resolving the file paths.
+  string sandbox_dir = 6;
+}
+
+// The worker sends this message to Blaze when it finished its work on the
+// WorkRequest message.
+message WorkResponse {
+  int32 exit_code = 1;
+
+  // This is printed to the user after the WorkResponse has been received and is
+  // supposed to contain compiler warnings / errors etc. - thus we'll use a
+  // string type here, which gives us UTF-8 encoding.
+  string output = 2;
+
+  // This field must be set to the same request_id as the WorkRequest it is a
+  // response to. Since worker processes which support multiplex worker will
+  // handle multiple WorkRequests in parallel, this ID will be used to
+  // determined which WorkerProxy does this WorkResponse belong to.
+  int32 request_id = 3;
+
+  // EXPERIMENTAL When true, indicates that this response was sent due to
+  // receiving a cancel request. The exit_code and output fields should be empty
+  // and will be ignored. Exactly one WorkResponse must be sent for each
+  // non-cancelling WorkRequest received by the worker, but if the worker
+  // received a cancel request, it doesn't matter if it replies with a regular
+  // WorkResponse or with one where was_cancelled = true.
+  bool was_cancelled = 4;
+}
diff --git a/examples/persistent_worker/proto/buck2/BUCK b/examples/persistent_worker/proto/buck2/BUCK
new file mode 100644
index 0000000000000..87b120176bbd1
--- /dev/null
+++ b/examples/persistent_worker/proto/buck2/BUCK
@@ -0,0 +1,7 @@
+load("//proto:defs.bzl", "proto_python_library")
+
+proto_python_library(
+    name = "worker_pb2",
+    src = "worker.proto",
+    visibility = ["PUBLIC"],
+)
diff --git a/examples/persistent_worker/proto/buck2/worker.proto b/examples/persistent_worker/proto/buck2/worker.proto
new file mode 100644
index 0000000000000..0fb4ced5062b4
--- /dev/null
+++ b/examples/persistent_worker/proto/buck2/worker.proto
@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under both the MIT license found in the
+ * LICENSE-MIT file in the root directory of this source tree and the Apache
+ * License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+ * of this source tree.
+ */
+
+syntax = "proto3";
+
+option java_multiple_files = true;
+option java_package = "com.facebook.buck.worker.model";
+option java_outer_classname = "WorkerProto";
+
+package worker;
+
+message ExecuteCommand {
+  message EnvironmentEntry {
+    bytes key = 1;
+    bytes value = 2;
+  }
+
+  repeated bytes argv = 1;
+  repeated EnvironmentEntry env = 2;
+}
+
+message ExecuteResponse {
+  int32 exit_code = 1;
+  string stderr = 2;
+}
+
+message ExecuteCancel {}
+
+message ExecuteEvent {
+  oneof data {
+    ExecuteCommand command = 1;
+    ExecuteCancel cancel = 2;
+  }
+}
+
+service Worker {
+  // TODO(ctolliday) delete once workers switch to Exec
+  rpc Execute(ExecuteCommand) returns (ExecuteResponse) {};
+
+  rpc Exec(stream ExecuteEvent) returns (ExecuteResponse) {};
+}
diff --git a/examples/persistent_worker/proto/defs.bzl b/examples/persistent_worker/proto/defs.bzl
new file mode 100644
index 0000000000000..cf7e3a4f39f24
--- /dev/null
+++ b/examples/persistent_worker/proto/defs.bzl
@@ -0,0 +1,55 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+def _proto_python_library_impl(ctx: AnalysisContext) -> list[Provider]:
+    prefix = ctx.label.package
+    depth = len(prefix.split("/"))
+    libname = ctx.attrs.src.basename.removesuffix(".proto") + "_pb2"
+    python_out = ctx.actions.declare_output(prefix, "{}.py".format(libname))
+    pyi_out = ctx.actions.declare_output(prefix, "{}.pyi".format(libname))
+    grpc_python_out = ctx.actions.declare_output(prefix, "{}_grpc.py".format(libname))
+    ctx.actions.run(
+        cmd_args(
+            ctx.attrs._protoc[RunInfo],
+            cmd_args(ctx.attrs.src, format = "-I{}={}", parent = 1),
+            cmd_args(python_out.as_output(), format = "--python_out={}", parent = depth + 1),
+            cmd_args(pyi_out.as_output(), format = "--pyi_out={}", parent = depth + 1),
+            cmd_args(grpc_python_out.as_output(), format = "--grpc_python_out={}", parent = depth + 1),
+            ctx.attrs.src,
+        ),
+        category = "protoc",
+    )
+
+    # protoc does not let us control the import prefix and path prefix separately.
+    # So, we need to copy the generated files into the correct location after the fact.
+    python_out_copied = ctx.actions.declare_output("{}.py".format(libname))
+    pyi_out_copied = ctx.actions.declare_output("{}.pyi".format(libname))
+    grpc_python_out_copied = ctx.actions.declare_output("{}_grpc.py".format(libname))
+    ctx.actions.copy_file(python_out_copied, python_out)
+    ctx.actions.copy_file(pyi_out_copied, pyi_out)
+    ctx.actions.copy_file(grpc_python_out_copied, grpc_python_out)
+    return [DefaultInfo(default_outputs = [python_out_copied, pyi_out_copied, grpc_python_out_copied])]
+
+_proto_python_library = rule(
+    impl = _proto_python_library_impl,
+    attrs = {
+        "src": attrs.source(),
+        "_protoc": attrs.exec_dep(default = "//proto:protoc", providers = [RunInfo]),
+    },
+)
+
+def proto_python_library(*, name, src, **kwargs):
+    _proto_python_library(
+        name = "{}-gen".format(name),
+        src = src,
+    )
+    native.python_library(
+        name = name,
+        srcs = [":{}-gen".format(name)],
+        deps = ["//python:grpcio"],
+        **kwargs
+    )
diff --git a/examples/persistent_worker/python/BUCK b/examples/persistent_worker/python/BUCK
new file mode 100644
index 0000000000000..a0b16bbc95e09
--- /dev/null
+++ b/examples/persistent_worker/python/BUCK
@@ -0,0 +1,26 @@
+load(":defs.bzl", "fetch_python_libraries")
+
+fetch_python_libraries(pkgs = {
+    "grpcio": {
+        "deps": ["protobuf"],
+        "sha256": "f145cc21836c332c67baa6fc81099d1d27e266401565bf481948010d6ea32d46",
+        "url": "https://files.pythonhosted.org/packages/2f/86/a86742f3deaa22385c3bff984c5947fc62d47d3fab26c508730037d027e5/grpcio-1.66.2-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl",
+    },
+    "grpcio-tools": {
+        "deps": [
+            "grpcio",
+            "protobuf",
+            "setuptools",
+        ],
+        "sha256": "c68642829368f4f83929e0df571dbbc99f1f1553555d8f98d0582da9f6743d9e",
+        "url": "https://files.pythonhosted.org/packages/1d/0f/273d7ac9c7d99b56abb5841d8aff7ffd148fe01b48c2913c8da3de9438e7/grpcio_tools-1.66.2-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl",
+    },
+    "protobuf": {
+        "sha256": "52235802093bd8a2811abbe8bf0ab9c5f54cca0a751fdd3f6ac2a21438bffece",
+        "url": "https://files.pythonhosted.org/packages/9b/55/f24e3b801d2e108c48aa2b1b59bb791b5cffba89465cbbf66fc98de89270/protobuf-5.28.2-py3-none-any.whl",
+    },
+    "setuptools": {
+        "sha256": "35ab7fd3bcd95e6b7fd704e4a1539513edad446c097797f2985e0e4b960772f2",
+        "url": "https://files.pythonhosted.org/packages/ff/ae/f19306b5a221f6a436d8f2238d5b80925004093fa3edea59835b514d9057/setuptools-75.1.0-py3-none-any.whl",
+    },
+})
diff --git a/examples/persistent_worker/python/defs.bzl b/examples/persistent_worker/python/defs.bzl
new file mode 100644
index 0000000000000..66f1a719f520c
--- /dev/null
+++ b/examples/persistent_worker/python/defs.bzl
@@ -0,0 +1,21 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+def fetch_python_libraries(pkgs):
+    for name, pkg in pkgs.items():
+        native.remote_file(
+            name = "{}-download".format(name),
+            url = pkg["url"],
+            sha256 = pkg["sha256"],
+            out = "{}.whl".format(name),
+        )
+        native.prebuilt_python_library(
+            name = name,
+            binary_src = ":{}-download".format(name),
+            deps = [":{}".format(dep) for dep in pkg.get("deps", [])],
+            visibility = ["PUBLIC"],
+        )
diff --git a/examples/persistent_worker/test.sh b/examples/persistent_worker/test.sh
new file mode 100755
index 0000000000000..06d77fbd814e4
--- /dev/null
+++ b/examples/persistent_worker/test.sh
@@ -0,0 +1,127 @@
+#!/usr/bin/env bash
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+set -euo pipefail
+
+echo "::group::Local build without persistent worker" >&2
+cat >.buckconfig.local <<EOF
+<file:.buckconfig.no-workers>
+EOF
+buck2 clean; buck2 build : -vstderr
+echo "# Verifying Buck2 log" >&2
+buck2 log what-ran --show-std-err --format json \
+  | jq -s '
+      [
+        .[]
+        | select(.identity | startswith("root//:demo-"))
+      ]
+      | if length == 4 then
+          .
+        else
+          error("expected 4 demo targets, got " + (length | tostring))
+        end
+      | .[]
+      | if .reproducer.executor == "Local" and (.std_err | startswith("one-shot.py")) then
+          true
+        else
+          error("expected local without persistent worker, got " + ([.reproducer.executor, .std_err] | tostring))
+        end
+    '
+echo "::endgroup::" >&2
+
+echo "::group::Local build with persistent worker" >&2
+cat >.buckconfig.local <<EOF
+<file:.buckconfig.local-persistent-workers>
+EOF
+buck2 clean; buck2 build : -vstderr
+echo "# Verifying Buck2 log" >&2
+buck2 log what-ran --show-std-err --format json \
+  | jq -s '
+      [
+        .[]
+        | select(.identity | startswith("root//:demo-"))
+      ]
+      | if length == 5 then
+          .
+        else
+          error("expected 5 demo targets, got " + (length | tostring))
+        end
+      | .[]
+      | if (.reproducer.executor == "Worker" or .reproducer.executor == "WorkerInit") and (.std_err | startswith("Buck2 persistent worker")) then
+          true
+        else
+          error("expected local without persistent worker, got " + ([.reproducer.executor, .std_err] | tostring))
+        end
+    '
+echo "::endgroup::" >&2
+
+echo "::group::Remote build without persistent worker" >&2
+if [[ -z ${BUILDBUDDY_API_KEY:+x} ]]; then
+  echo "::notice file=$(realpath --relative-to=../.. ${BASH_SOURCE[0]}),line=${LINENO}::SKIPPED Missing BuildBuddy token. See examples/persistent_worker/README.md" >&2
+else
+  cat >.buckconfig.local <<EOF
+<file:.buckconfig.buildbuddy>
+
+[build]
+cache_silo_key=$(date +%s.%N).${GITHUB_RUN_ID-0}
+EOF
+  buck2 clean; buck2 build : -vstderr
+  echo "# Verifying Buck2 log" >&2
+  buck2 log what-ran --show-std-err --format json \
+    | jq -s '
+        [
+          .[]
+          | select(.identity | startswith("root//:demo-"))
+        ]
+        | if length == 4 then
+            .
+          else
+            error("expected 4 demo targets, got " + (length | tostring))
+          end
+        | .[]
+        | if .reproducer.executor == "Re" and (.std_err | startswith("one-shot.py")) then
+            true
+          else
+            error("expected local without persistent worker, got " + ([.reproducer.executor, .std_err] | tostring))
+          end
+      '
+fi
+echo "::endgroup::" >&2
+
+echo "::group::Remote build with persistent worker" >&2
+if [[ -z ${BUILDBUDDY_API_KEY:+x} ]]; then
+  echo "::notice file=$(realpath --relative-to=../.. ${BASH_SOURCE[0]}),line=${LINENO}::SKIPPED Missing BuildBuddy token. See examples/persistent_worker/README.md" >&2
+else
+  cat >.buckconfig.local <<EOF
+<file:.buckconfig.buildbuddy-persistent-workers>
+
+[build]
+cache_silo_key=$(date +%s.%N).${GITHUB_RUN_ID-0}
+EOF
+  buck2 clean; buck2 build : -vstderr
+  echo "# Verifying Buck2 log" >&2
+  buck2 log what-ran --show-std-err --format json \
+    | jq -s '
+        [
+          .[]
+          | select(.identity | startswith("root//:demo-"))
+        ]
+        | if length == 4 then
+            .
+          else
+            error("expected 4 demo targets, got " + (length | tostring))
+          end
+        | .[]
+        | if .reproducer.executor == "Re" and (.std_err | startswith("Bazel persistent worker")) then
+            true
+          else
+            error("expected remote persistent worker, got " + ([.reproducer.executor, .std_err] | tostring))
+          end
+      '
+fi
+echo "::endgroup::" >&2
diff --git a/examples/persistent_worker/toolchains/BUCK b/examples/persistent_worker/toolchains/BUCK
new file mode 100644
index 0000000000000..1b7ca8f52ebe0
--- /dev/null
+++ b/examples/persistent_worker/toolchains/BUCK
@@ -0,0 +1,38 @@
+load("@prelude//toolchains:cxx.bzl", "system_cxx_toolchain")
+load("@prelude//toolchains:genrule.bzl", "system_genrule_toolchain")
+load("@prelude//toolchains:python.bzl", "system_python_bootstrap_toolchain")
+load(":defs.bzl", "python_toolchain")
+
+system_cxx_toolchain(
+    name = "cxx",
+    visibility = ["PUBLIC"],
+)
+
+system_genrule_toolchain(
+    name = "genrule",
+    visibility = ["PUBLIC"],
+)
+
+system_python_bootstrap_toolchain(
+    name = "python_bootstrap",
+    interpreter = "python3",
+    visibility = ["PUBLIC"],
+)
+
+remote_file(
+    name = "python-download",
+    out = "cpython.tar.gz",
+    sha256 = "445156c61e1cc167f7b8777ad08cc36e5598e12cd27e07453f6e6dc0f62e421e",
+    url = "https://github.com/indygreg/python-build-standalone/releases/download/20241002/cpython-3.13.0rc3+20241002-x86_64-unknown-linux-gnu-install_only_stripped.tar.gz",
+)
+
+extract_archive(
+    name = "python-extract",
+    src = ":python-download",
+)
+
+python_toolchain(
+    name = "python",
+    distribution = ":python-extract",
+    visibility = ["PUBLIC"],
+)
diff --git a/examples/persistent_worker/toolchains/defs.bzl b/examples/persistent_worker/toolchains/defs.bzl
new file mode 100644
index 0000000000000..c6594923a09e4
--- /dev/null
+++ b/examples/persistent_worker/toolchains/defs.bzl
@@ -0,0 +1,59 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under both the MIT license found in the
+# LICENSE-MIT file in the root directory of this source tree and the Apache
+# License, Version 2.0 found in the LICENSE-APACHE file in the root directory
+# of this source tree.
+
+load(
+    "@prelude//:artifacts.bzl",
+    "ArtifactGroupInfo",
+)
+load(
+    "@prelude//python:toolchain.bzl",
+    "PythonPlatformInfo",
+    "PythonToolchainInfo",
+)
+
+def _python_toolchain_impl(ctx):
+    distribution = ctx.attrs.distribution[DefaultInfo].default_outputs[0]
+    interpreter = cmd_args(distribution, absolute_suffix = "/python/bin/python")
+    return [
+        DefaultInfo(),
+        PythonToolchainInfo(
+            binary_linker_flags = ctx.attrs.binary_linker_flags,
+            linker_flags = ctx.attrs.linker_flags,
+            fail_with_message = ctx.attrs.fail_with_message[RunInfo],
+            generate_static_extension_info = ctx.attrs.generate_static_extension_info,
+            make_source_db = ctx.attrs.make_source_db[RunInfo],
+            make_source_db_no_deps = ctx.attrs.make_source_db_no_deps[RunInfo],
+            host_interpreter = RunInfo(args = ["python3"]),
+            interpreter = RunInfo(args = [interpreter]),
+            make_py_package_modules = ctx.attrs.make_py_package_modules[RunInfo],
+            make_py_package_inplace = ctx.attrs.make_py_package_inplace[RunInfo],
+            compile = RunInfo(args = ["echo", "COMPILEINFO"]),
+            package_style = "inplace",
+            pex_extension = ctx.attrs.pex_extension,
+            native_link_strategy = "separate",
+            runtime_library = ctx.attrs.runtime_library,
+        ),
+        PythonPlatformInfo(name = "x86_64"),
+    ]
+
+python_toolchain = rule(
+    impl = _python_toolchain_impl,
+    attrs = {
+        "binary_linker_flags": attrs.default_only(attrs.list(attrs.arg(), default = [])),
+        "distribution": attrs.exec_dep(),
+        "fail_with_message": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:fail_with_message")),
+        "generate_static_extension_info": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:generate_static_extension_info")),
+        "linker_flags": attrs.default_only(attrs.list(attrs.arg(), default = [])),
+        "make_py_package_inplace": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:make_py_package_inplace")),
+        "make_py_package_modules": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:make_py_package_modules")),
+        "make_source_db": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:make_source_db")),
+        "make_source_db_no_deps": attrs.default_only(attrs.dep(providers = [RunInfo], default = "prelude//python/tools:make_source_db_no_deps")),
+        "pex_extension": attrs.string(default = ".pex"),
+        "runtime_library": attrs.default_only(attrs.dep(providers = [ArtifactGroupInfo], default = "prelude//python/runtime:bootstrap_files")),
+    },
+    is_toolchain_rule = True,
+)