diff --git a/centipede/BUILD b/centipede/BUILD
index 607811f8d..6508bc1bb 100644
--- a/centipede/BUILD
+++ b/centipede/BUILD
@@ -991,8 +991,6 @@ cc_library(
         ":runner_result",
         ":shared_memory_blob_sequence",
         "@abseil-cpp//absl/base:nullability",
-        "@abseil-cpp//absl/random",
-        "@abseil-cpp//absl/random:bit_gen_ref",
         "@com_google_fuzztest//common:defs",
     ],
 )
@@ -1140,6 +1138,7 @@ cc_library(
     linkstatic = True,  # Must be linked statically even when dynamic_mode=on.
     deps = [
         ":centipede_runner_no_main",
+        ":execution_metadata",
         ":mutation_data",
         "@abseil-cpp//absl/base:nullability",
         "@abseil-cpp//absl/types:span",
@@ -1957,7 +1956,7 @@ cc_test(
     timeout = "long",
     srcs = ["centipede_test.cc"],
     data = [
-        "@com_google_fuzztest//centipede",
+        ":centipede",
         "@com_google_fuzztest//centipede/testing:abort_fuzz_target",
         "@com_google_fuzztest//centipede/testing:async_failing_target",
         "@com_google_fuzztest//centipede/testing:expensive_startup_fuzz_target",
diff --git a/centipede/centipede_callbacks.cc b/centipede/centipede_callbacks.cc
index b03b48aad..aae910449 100644
--- a/centipede/centipede_callbacks.cc
+++ b/centipede/centipede_callbacks.cc
@@ -23,6 +23,7 @@
 
 #include <algorithm>
 #include <cerrno>
+#include <cmath>
 #include <cstddef>
 #include <cstdlib>
 #include <cstring>
@@ -194,8 +195,11 @@ class CentipedeCallbacks::PersistentModeServer {
     int poll_ret = -1;
     do {
       poll_fd = {fd, static_cast<short>(event)};
-      const int poll_timeout_ms = static_cast<int>(absl::ToInt64Milliseconds(
-          std::max(deadline - absl::Now(), kPollMinimalTimeout)));
+      const int poll_timeout_ms =
+          deadline == absl::InfiniteFuture()
+              ? -1
+              : static_cast<int>(std::ceil(absl::ToDoubleMilliseconds(
+                    std::max(deadline - absl::Now(), kPollMinimalTimeout))));
       poll_ret = poll(&poll_fd, 1, poll_timeout_ms);
     } while (poll_ret < 0 && errno == EINTR);
     if (poll_ret == 1 && (poll_fd.revents & (event | POLLHUP)) == event) {
@@ -343,7 +347,6 @@ std::string CentipedeCallbacks::ConstructRunnerFlags(
   std::vector<std::string> flags = {
       "CENTIPEDE_RUNNER_FLAGS=",
       absl::StrCat("timeout_per_input=", env_.timeout_per_input),
-      absl::StrCat("timeout_per_batch=", env_.timeout_per_batch),
       absl::StrCat("address_space_limit_mb=", env_.address_space_limit_mb),
       absl::StrCat("rss_limit_mb=", env_.rss_limit_mb),
       absl::StrCat("stack_limit_kb=", env_.stack_limit_kb),
@@ -565,8 +568,12 @@ int CentipedeCallbacks::ExecuteCentipedeSancovBinaryWithShmem(
   }
 
   // Run.
+  const auto batch_start_time = absl::Now();
   const int exit_code = RunBatchForBinary(binary);
   inputs_blobseq_.ReleaseSharedMemory();  // Inputs are already consumed.
+  const bool batch_timed_out =
+      env_.timeout_per_batch > 0 &&
+      absl::Now() - batch_start_time > absl::Seconds(env_.timeout_per_batch);
 
   // Get results.
   batch_result.exit_code() = exit_code;
@@ -591,6 +598,16 @@ int CentipedeCallbacks::ExecuteCentipedeSancovBinaryWithShmem(
 
   if (env_.print_runner_log) PrintExecutionLog();
 
+  if (batch_timed_out) {
+    FUZZTEST_LOG(INFO)
+        << "Batch timeout detected. Replacing the original exit code: "
+        << exit_code
+        << ", failure description: " << batch_result.failure_description();
+    batch_result.failure_description() = kExecutionFailurePerBatchTimeout;
+    batch_result.exit_code() = EXIT_FAILURE;
+    return EXIT_FAILURE;
+  }
+
   // TODO: b/467103298 - Handle failures when the exit code is zero, e.g., when
   // the target exits via `std::_Exit(0)`.
   if (exit_code != EXIT_SUCCESS) {
diff --git a/centipede/centipede_test.cc b/centipede/centipede_test.cc
index 17383e8f3..9391e271f 100644
--- a/centipede/centipede_test.cc
+++ b/centipede/centipede_test.cc
@@ -86,7 +86,7 @@ class CentipedeMock : public CentipedeCallbacks {
     // i-th element is the number of bytes with the value 'i' in the input.
     // `counters` is converted to FeatureVec and added to
     // `batch_result.results()`.
-    for (auto &input : inputs) {
+    for (auto& input : inputs) {
       ByteArray counters(256);
       for (uint8_t byte : input) {
         counters[byte]++;
@@ -496,7 +496,7 @@ TEST_F(CentipedeWithTemporaryLocalDir, MutateViaExternalBinary) {
     // Must contain normal mutants, but not the ones from crossover.
     const auto mutant_data = GetDataFromMutants(result.mutants());
     EXPECT_THAT(mutant_data, IsSupersetOf(some_of_expected_mutants));
-    for (const auto &crossover_mutant : expected_crossover_mutants) {
+    for (const auto& crossover_mutant : expected_crossover_mutants) {
       EXPECT_THAT(mutant_data, Not(Contains(crossover_mutant)));
     }
   }
@@ -525,7 +525,7 @@ class MergeMock : public CentipedeCallbacks {
   std::vector<Mutant> Mutate(absl::Span<const MutationInputRef> inputs,
                              size_t num_mutants) override {
     std::vector<Mutant> mutants(num_mutants);
-    for (auto &mutant : mutants) {
+    for (auto& mutant : mutants) {
       mutant.data.resize(1);
       mutant.data[0] = ++number_of_mutations_;
       mutant.origin = Mutant::kOriginNone;
@@ -611,7 +611,7 @@ class FunctionFilterMock : public CentipedeCallbacks {
   // Sets the inputs to one of 3 pre-defined values.
   std::vector<Mutant> Mutate(absl::Span<const MutationInputRef> inputs,
                              size_t num_mutants) override {
-    for (auto &input : inputs) {
+    for (auto& input : inputs) {
       if (!seed_inputs_.contains(input.data)) {
         observed_inputs_.insert(input.data);
       }
@@ -628,8 +628,8 @@ class FunctionFilterMock : public CentipedeCallbacks {
   // Returns one of 3 pre-defined values, that trigger different code paths in
   // the test target.
   static ByteArray GetMutant(size_t idx) {
-    const char *mutants[3] = {"func1", "func2-A", "foo"};
-    const char *mutant = mutants[idx % 3];
+    const char* mutants[3] = {"func1", "func2-A", "foo"};
+    const char* mutant = mutants[idx % 3];
     return {mutant, mutant + strlen(mutant)};
   }
 
@@ -646,7 +646,7 @@ class FunctionFilterMock : public CentipedeCallbacks {
 // Runs a short fuzzing session with the provided `function_filter`.
 // Returns a sorted array of observed inputs.
 static std::vector<ByteArray> RunWithFunctionFilter(
-    std::string_view function_filter, const TempDir &tmp_dir) {
+    std::string_view function_filter, const TempDir& tmp_dir) {
   Environment env;
   env.workdir = tmp_dir.path();
   env.seed = 1;  // make the runs predictable.
@@ -717,9 +717,9 @@ class ExtraBinariesMock : public CentipedeCallbacks {
   bool Execute(std::string_view binary, absl::Span<const ByteSpan> inputs,
                BatchResult& batch_result) override {
     bool res = true;
-    for (const auto &input : inputs) {
+    for (const auto& input : inputs) {
       if (input.size() != 1) continue;
-      for (const Crash &crash : crashes_) {
+      for (const Crash& crash : crashes_) {
         if (binary == crash.binary && input[0] == crash.input) {
           batch_result.exit_code() = EXIT_FAILURE;
           batch_result.failure_description() = crash.description;
@@ -736,7 +736,7 @@ class ExtraBinariesMock : public CentipedeCallbacks {
   std::vector<Mutant> Mutate(absl::Span<const MutationInputRef> inputs,
                              size_t num_mutants) override {
     std::vector<Mutant> mutants(num_mutants);
-    for (auto &mutant : mutants) {
+    for (auto& mutant : mutants) {
       mutant.data.resize(1);
       mutant.data[0] = ++number_of_mutations_;
       mutant.origin = Mutant::kOriginNone;
@@ -754,20 +754,20 @@ struct FileAndContents {
   std::string file;
   std::string contents;
 
-  bool operator==(const FileAndContents &other) const {
+  bool operator==(const FileAndContents& other) const {
     return file == other.file && contents == other.contents;
   }
 
   template <typename Sink>
-  friend void AbslStringify(Sink &sink, const FileAndContents &f) {
+  friend void AbslStringify(Sink& sink, const FileAndContents& f) {
     absl::Format(&sink, "FileAndContents{%s, \"%s\"}", f.file, f.contents);
   }
 };
 
 MATCHER_P(HasFilesWithContents, expected_files_and_contents, "") {
-  const std::string &dir_path = arg;
+  const std::string& dir_path = arg;
   std::vector<FileAndContents> files_and_contents;
-  for (const auto &dir_ent : std::filesystem::directory_iterator(dir_path)) {
+  for (const auto& dir_ent : std::filesystem::directory_iterator(dir_path)) {
     auto file_and_contents = FileAndContents{dir_ent.path().filename()};
     ReadFromLocalFile(dir_ent.path().c_str(), file_and_contents.contents);
     files_and_contents.push_back(std::move(file_and_contents));
@@ -843,7 +843,7 @@ class UndetectedCrashingInputMock : public CentipedeCallbacks {
     if (!first_pass_) {
       num_inputs_triaged_ += inputs.size();
     }
-    for (const auto &input : inputs) {
+    for (const auto& input : inputs) {
       FUZZTEST_CHECK_EQ(input.size(), 1);  // By construction in `Mutate()`.
       // The contents of each mutant is its sequential number.
       if (input[0] == crashing_input_idx_) {
@@ -933,7 +933,7 @@ TEST(Centipede, UndetectedCrashingInput) {
         absl::StrCat("crashing_batch-", crashing_input_hash);
     EXPECT_TRUE(std::filesystem::exists(crashes_dir_path)) << crashes_dir_path;
     std::vector<std::string> found_crash_file_names;
-    for (auto const &dir_ent :
+    for (auto const& dir_ent :
          std::filesystem::directory_iterator(crashes_dir_path)) {
       found_crash_file_names.push_back(dir_ent.path().filename());
     }
@@ -963,14 +963,8 @@ TEST_F(CentipedeWithTemporaryLocalDir, GetsSeedInputs) {
   CentipedeDefaultCallbacks callbacks(env, stop_condition);
 
   std::vector<ByteArray> seeds;
-  EXPECT_EQ(callbacks.GetSeeds(10, seeds), 10);
-  EXPECT_THAT(seeds, testing::ContainerEq(std::vector<ByteArray>{
-                         {0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}}));
-  EXPECT_EQ(callbacks.GetSeeds(5, seeds), 10);
-  EXPECT_THAT(seeds, testing::ContainerEq(
-                         std::vector<ByteArray>{{0}, {1}, {2}, {3}, {4}}));
-  EXPECT_EQ(callbacks.GetSeeds(100, seeds), 10);
-  EXPECT_THAT(seeds, testing::ContainerEq(std::vector<ByteArray>{
+  callbacks.GetSeeds(10, seeds);
+  EXPECT_THAT(seeds, testing::IsSupersetOf(std::vector<ByteArray>{
                          {0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}}));
 }
 
@@ -1277,12 +1271,13 @@ TEST_F(CentipedeWithTemporaryLocalDir, UsesProvidedCustomMutator) {
       "centipede/testing/fuzz_target_with_custom_mutator");
   CentipedeDefaultCallbacks callbacks(env, stop_condition);
 
-  const std::vector<ByteArray> inputs = {{1}, {2}, {3}, {4}, {5}, {6}};
-  const std::vector<Mutant> mutants = callbacks.Mutate(
-      GetMutationInputRefsFromDataInputs(inputs), inputs.size());
+  const std::vector<ByteArray> inputs = {{99}};
+  const std::vector<Mutant> mutants =
+      callbacks.Mutate(GetMutationInputRefsFromDataInputs(inputs), 5);
 
-  // The custom mutator just returns the original inputs as mutants.
-  EXPECT_EQ(inputs, GetDataFromMutants(mutants));
+  // The custom mutator just duplicates the original inputs as mutants.
+  EXPECT_EQ(GetDataFromMutants(mutants),
+            (std::vector<ByteArray>{{99}, {99}, {99}, {99}, {99}}));
 }
 
 TEST_F(CentipedeWithTemporaryLocalDir, FailsOnMisbehavingCustomMutator) {
@@ -1400,14 +1395,14 @@ TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInWorkerMode) {
   env.test_name = "some_test";
   env.populate_binary_info = false;
   env.fork_server = false;
-  env.persistent_mode = false;
+  env.persistent_mode = true;
   env.exit_on_crash = true;
+  env.stop_at = absl::Now() + absl::Seconds(10);
   fuzztest::internal::DefaultCallbacksFactory<
       fuzztest::internal::CentipedeDefaultCallbacks>
       callbacks;
-  EXPECT_DEATH(
-      [&] { std::exit(CentipedeMain(env, callbacks)); }(),
-      ContainsRegex("Failure *: INPUT FAILURE: some_failure_description"));
+  EXPECT_DEATH([&] { std::exit(CentipedeMain(env, callbacks)); }(),
+               ContainsRegex("Failure *: some_failure_description"));
 }
 
 TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInStandaloneMode) {
@@ -1427,7 +1422,7 @@ TEST_F(CentipedeWithTemporaryLocalDir, EngineWorksInStandaloneMode) {
         const int status = std::system(test_command.c_str());
         std::exit(WEXITSTATUS(status));
       }(),
-      ContainsRegex("Failure *: INPUT FAILURE: some_failure_description"));
+      ContainsRegex("Failure *: some_failure_description"));
 }
 
 }  // namespace
diff --git a/centipede/command.cc b/centipede/command.cc
index 4afab0abb..308839468 100644
--- a/centipede/command.cc
+++ b/centipede/command.cc
@@ -30,6 +30,7 @@
 
 #include <algorithm>
 #include <atomic>
+#include <cmath>
 #include <csignal>
 #include <cstdint>
 #include <cstdlib>
@@ -146,8 +147,11 @@ struct Command::ForkServerProps {
           /*fd=*/pipe_[1],
           /*events=*/POLLIN,
       };
-      const int poll_timeout_ms = static_cast<int>(absl::ToInt64Milliseconds(
-          std::max(deadline - absl::Now(), absl::Milliseconds(1))));
+      const int poll_timeout_ms =
+          deadline == absl::InfiniteFuture()
+              ? -1
+              : static_cast<int>(std::ceil(absl::ToDoubleMilliseconds(
+                    std::max(deadline - absl::Now(), absl::Milliseconds(1)))));
       poll_ret = poll(&poll_fd, 1, poll_timeout_ms);
       // The `poll()` syscall can get interrupted: it sets errno==EINTR in that
       // case. We should tolerate that.
@@ -207,7 +211,7 @@ std::string Command::ToString() const {
   ss.reserve(/*env*/ 1 + options_.env_diff.size() + /*path*/ 1 +
              /*args*/ options_.args.size() + /*out/err*/ 2);
   // env.
-  ss.push_back("env");
+  ss.push_back("exec env");
   std::vector<std::string> env_to_set;
   env_to_set.reserve(options_.env_diff.size());
   // Arguments that unset environment variables must appear first.
@@ -287,7 +291,7 @@ bool Command::StartForkServer(std::string_view temp_dir_path,
   {
     CENTIPEDE_FORK_SERVER_FIFO0="%s" \
     CENTIPEDE_FORK_SERVER_FIFO1="%s" \
-    exec %s
+    %s
   } &
   printf "%%s" $! > "%s"
 )sh";
diff --git a/centipede/command_test.cc b/centipede/command_test.cc
index 4e638b4b2..48f4d0730 100644
--- a/centipede/command_test.cc
+++ b/centipede/command_test.cc
@@ -39,18 +39,18 @@ namespace {
 using ::testing::Optional;
 
 TEST(CommandTest, ToString) {
-  EXPECT_EQ(Command{"x"}.ToString(), "env \\\nx");
+  EXPECT_EQ(Command{"x"}.ToString(), "exec env \\\nx");
   {
     Command::Options cmd_options;
     cmd_options.args = {"arg1", "arg2"};
     EXPECT_EQ((Command{"path", std::move(cmd_options)}.ToString()),
-              "env \\\npath \\\narg1 \\\narg2");
+              "exec env \\\npath \\\narg1 \\\narg2");
   }
   {
     Command::Options cmd_options;
     cmd_options.env_diff = {"K1=V1", "K2=V2", "-K3"};
     EXPECT_EQ((Command{"x", std::move(cmd_options)}.ToString()),
-              "env \\\n-u K3 \\\nK1=V1 \\\nK2=V2 \\\nx");
+              "exec env \\\n-u K3 \\\nK1=V1 \\\nK2=V2 \\\nx");
   }
 }
 
@@ -80,7 +80,7 @@ TEST(CommandTest, InputFileWildCard) {
   Command::Options cmd_options;
   cmd_options.temp_file_path = "TEMP_FILE";
   Command cmd{"foo bar @@ baz", std::move(cmd_options)};
-  EXPECT_EQ(cmd.ToString(), "env \\\nfoo bar TEMP_FILE baz");
+  EXPECT_EQ(cmd.ToString(), "exec env \\\nfoo bar TEMP_FILE baz");
 }
 
 TEST(CommandTest, ForkServer) {
diff --git a/centipede/engine_worker.cc b/centipede/engine_worker.cc
index 01030d1eb..d58a24a47 100644
--- a/centipede/engine_worker.cc
+++ b/centipede/engine_worker.cc
@@ -12,6 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 #include <fcntl.h>
+#include <sys/resource.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <sys/un.h>
 #include <unistd.h>
 
 #include <algorithm>
@@ -23,14 +27,13 @@
 #include <cstdlib>
 #include <cstring>
 #include <memory>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <type_traits>
 #include <vector>
 
 #include "absl/base/nullability.h"
-#include "absl/random/bit_gen_ref.h"
-#include "absl/random/random.h"
 #include "./centipede/engine_abi.h"
 #include "./centipede/engine_worker_abi.h"
 #include "./centipede/execution_metadata.h"
@@ -105,7 +108,7 @@ struct WorkerFlags {
 };
 
 // The first call of this function must be outside of signal handlers since it
-// allocates memory (enforced by `GetWorkerFlagsEarly`). After that it would be
+// allocates memory (enforced by `WorkerInitEarly`). After that it would be
 // signal-safe.
 //
 // The worker flags format is `:(NAME=VALUE|SWITCH:)+`. `GetWorkerFlags`
@@ -137,10 +140,6 @@ const WorkerFlags& GetWorkerFlags() {
   return worker_flags;
 }
 
-__attribute__((constructor(200))) void GetWorkerFlagsEarly() {
-  (void)GetWorkerFlags();
-}
-
 // `header` should be in the form of `FLAG_NAME=`.
 //
 // Extracts "value" as a null-terminated string from "\0FLAG_NAME=value\0" in
@@ -220,6 +219,7 @@ enum class WorkerAction {
   kTestGetSeeds,
   kTestMutate,
   kTestExecute,
+  kNoOp,
 };
 
 constexpr std::string_view kWorkerBinaryIdOutputFlagHeader =
@@ -237,6 +237,10 @@ constexpr std::string_view kWorkerInputsBlobSequencePathFlagHeader =
     "arg1=";  // TODO: Use better flag names when standardizing the protocol.
 constexpr std::string_view kWorkerOutputsBlobSequencePathFlagHeader =
     "arg2=";  // TODO: Use better flag names when standardizing the protocol.
+constexpr std::string_view kWorkerPersistentModeSocketPathFlagHeader =
+    "persistent_mode_socket=";  // TODO: Use better flag names when
+                                // standardizing the protocol.
+constexpr std::string_view kWorkerCrossOverLevel = "crossover_level=";
 
 struct WorkerState {
   std::atomic<bool> has_failure_output = false;
@@ -244,6 +248,11 @@ struct WorkerState {
   std::atomic<bool> in_adapter_execute = false;
   std::atomic<bool> has_finding = false;
   std::atomic<bool> saved_binary_id = false;
+
+  void ResetForPersistentMode() {
+    has_failure_output.store(false, std::memory_order_relaxed);
+    has_finding.store(false, std::memory_order_relaxed);
+  }
 };
 
 WorkerState& GetWorkerState() {
@@ -290,14 +299,26 @@ void WorkerEmitError(std::string_view message) {
   WorkerEmitFailureOutput("SETUP FAILURE: ", message);
 }
 
+static constexpr std::array<std::string_view, 3> kNonFindingPrefixes = {
+    "SETUP FAILURE:",
+    "IGNORED FAILURE:",
+    "SKIPPED TEST:",
+};
+
 void WorkerEmitFinding(std::string_view description,
                        std::string_view signature) {
   WorkerCheck(
       GetWorkerState().in_adapter_execute.load(std::memory_order_relaxed),
       "Must emit finding in adapter execute");
+  for (auto bad_prefix : kNonFindingPrefixes) {
+    if (description.substr(0, bad_prefix.size()) == bad_prefix) {
+      WorkerEmitError("Bad prefix in the error description");
+      return;
+    }
+  }
   const bool ignored = GetWorkerState().has_finding.exchange(true);
   if (!ignored) {
-    WorkerCheck(WorkerEmitFailureOutput("INPUT FAILURE: ", description),
+    WorkerCheck(WorkerEmitFailureOutput("", description),
                 "Failed to emit failure output for the finding");
     if (const char* finding_signature_path =
             GetWorkerFlag(kWorkerFailureSignaturePathFlagHeader);
@@ -323,6 +344,64 @@ inline std::string_view ToStringView(const std::vector<uint8_t>& bytes) {
   return {reinterpret_cast<const char*>(bytes.data()), bytes.size()};
 }
 
+// Zero initialized.
+static int persistent_mode_socket;
+
+__attribute__((constructor(200))) void WorkerInitEarly() {
+  const char* persistent_mode_socket_path =
+      GetWorkerFlag(kWorkerPersistentModeSocketPathFlagHeader);
+  if (persistent_mode_socket_path == nullptr) return;
+  persistent_mode_socket = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (persistent_mode_socket < 0) {
+    WorkerLog(
+        "Failed to create persistent mode socket - not running persistent "
+        "mode.",
+        LogLnSync{});
+    return;
+  }
+
+  struct sockaddr_un addr{};
+  addr.sun_family = AF_UNIX;
+  const size_t socket_path_len = strlen(persistent_mode_socket_path);
+  WorkerCheck(
+      socket_path_len < sizeof(addr.sun_path),
+      "persistent mode socket path string must be fit in sockaddr_un.sun_path");
+  std::memcpy(addr.sun_path, persistent_mode_socket_path, socket_path_len);
+
+  int connect_ret = 0;
+  do {
+    connect_ret =
+        connect(persistent_mode_socket, (struct sockaddr*)&addr, sizeof(addr));
+  } while (connect_ret == -1 && errno == EINTR);
+  if (connect_ret == -1) {
+    WorkerLog("Failed to connect the persistent mode socket to ",
+              persistent_mode_socket_path, LogLnSync{});
+    (void)close(persistent_mode_socket);
+    persistent_mode_socket = -1;
+  }
+
+  int flags = fcntl(persistent_mode_socket, F_GETFD);
+  if (flags == -1) {
+    WorkerLog(
+        "fcntl(F_GETFD) failed on the persistent mode socket - exiting "
+        "persistent mode",
+        LogLnSync{});
+    (void)close(persistent_mode_socket);
+    persistent_mode_socket = -1;
+  }
+  flags |= FD_CLOEXEC;
+  if (fcntl(persistent_mode_socket, F_SETFD, flags) == -1) {
+    WorkerLog(
+        "fcntl(F_SETFD) failed on the persistent mode socket - exiting "
+        "persistent mode",
+        LogLnSync{});
+    (void)close(persistent_mode_socket);
+    persistent_mode_socket = -1;
+  }
+  WorkerLog("Persistent mode: connected to ", persistent_mode_socket_path,
+            LogLnSync{});
+}
+
 BlobSequence* GetInputsBlobSequence() {
   static auto result = []() -> BlobSequence* {
     if (!HasWorkerSwitchFlag("shmem")) {
@@ -349,8 +428,25 @@ BlobSequence* GetOutputsBlobSequence() {
   return result;
 }
 
-WorkerAction GetWorkerAction() {
-  static WorkerAction worker_action = [] {
+int GetCrossOverLevel() {
+  static int result = []() {
+    const char* cross_over_level_str = GetWorkerFlag(kWorkerCrossOverLevel);
+    if (cross_over_level_str != nullptr) {
+      const int parsed =
+          atoi(cross_over_level_str);  // NOLINT: can't use strto64, etc.
+      if (0 <= parsed && parsed <= 100) return parsed;
+    }
+    // Default
+    return 50;
+  }();
+  return result;
+}
+
+std::optional<WorkerAction> GetWorkerAction() {
+  static auto worker_action = []() -> std::optional<WorkerAction> {
+    if (HasWorkerSwitchFlag("dump_configuration")) {
+      return WorkerAction::kNoOp;
+    }
     if (HasWorkerSwitchFlag("dump_binary_id")) {
       return WorkerAction::kGetBinaryId;
     }
@@ -361,7 +457,9 @@ WorkerAction GetWorkerAction() {
       return WorkerAction::kTestGetSeeds;
     }
     auto* inputs_blobseq = GetInputsBlobSequence();
-    WorkerCheck(inputs_blobseq != nullptr, "input blob sequence is not found");
+    if (inputs_blobseq == nullptr) {
+      return std::nullopt;
+    }
     auto request_type_blob = inputs_blobseq->Read();
     if (IsMutationRequest(request_type_blob)) {
       inputs_blobseq->Reset();
@@ -371,9 +469,7 @@ WorkerAction GetWorkerAction() {
       inputs_blobseq->Reset();
       return WorkerAction::kTestExecute;
     }
-    WorkerCheck(false, "unknown worker action from the flags");
-    // should not reach here.
-    std::abort();
+    return std::nullopt;
   }();
   return worker_action;
 }
@@ -457,14 +553,6 @@ void WorkerDoGetSeeds(const FuzzTestAdapter& adapter) {
   }
 }
 
-absl::BitGenRef GetBitGen() {
-  static thread_local std::unique_ptr<absl::BitGen> bitgen;
-  if (bitgen == nullptr) {
-    bitgen = std::make_unique<absl::BitGen>();
-  }
-  return *bitgen;
-}
-
 void WorkerDoMutate(const FuzzTestAdapter& adapter) {
   auto* inputs_blobseq = GetInputsBlobSequence();
   auto* outputs_blobseq = GetOutputsBlobSequence();
@@ -516,15 +604,23 @@ void WorkerDoMutate(const FuzzTestAdapter& adapter) {
 
   std::vector<uint8_t> mutant_bytes;
   const auto mutant_bytes_sink = GetBytesSinkTo(mutant_bytes);
+  static unsigned int seed = 0;
   for (size_t i = 0; i < num_mutants; ++i) {
-    const auto origin =
-        absl::Uniform<size_t>(GetBitGen(), 0, origin_inputs.size());
+    // Assume RAND_MAX is large enough and not worry about fairness for now.
+    const auto origin = rand_r(&seed) % origin_inputs.size();
     emitted_inputs.clear();
-    adapter.Mutate(adapter.ctx, origin_inputs[origin], /*shrink=*/0,
-                   &input_sink);
+    if (adapter.CrossOver != nullptr &&
+        rand_r(&seed) % 100 < GetCrossOverLevel()) {
+      const auto other = rand_r(&seed) % origin_inputs.size();
+      adapter.CrossOver(adapter.ctx, origin_inputs[origin],
+                        origin_inputs[other], &input_sink);
+    } else {
+      adapter.Mutate(adapter.ctx, origin_inputs[origin], /*shrink=*/0,
+                     &input_sink);
+    }
     WORKER_CHECK_FOR_ERROR();
     WorkerCheck(emitted_inputs.size() == 1,
-                "Mutate must emit exactly one input");
+                "Mutate/CrossOver must emit exactly one input");
     mutant_bytes.clear();
     adapter.SerializeInputContent(adapter.ctx, emitted_inputs[0],
                                   &mutant_bytes_sink);
@@ -596,6 +692,32 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) {
     return true;
   }();
 
+  // Utils to get execution stats.
+
+  size_t last_time_usec = 0;
+  auto UsecSinceLast = [&last_time_usec]() {
+    struct timeval tv = {};
+    constexpr size_t kUsecInSec = 1000000;
+    gettimeofday(&tv, nullptr);
+    uint64_t t = tv.tv_sec * kUsecInSec + tv.tv_usec;
+    uint64_t ret_val = t - last_time_usec;
+    last_time_usec = t;
+    return ret_val;
+  };
+
+  auto GetPeakRSSMb = []() -> size_t {
+    struct rusage usage = {};
+    if (getrusage(RUSAGE_SELF, &usage) != 0) return 0;
+#ifdef __APPLE__
+    // On MacOS, the unit seems to be byte according to experiment, while some
+    // documents mentioned KiB. This could depend on OS variants.
+    return usage.ru_maxrss >> 20;
+#else   // __APPLE__
+    // On Linux, ru_maxrss is in KiB
+    return usage.ru_maxrss >> 10;
+#endif  // __APPLE__
+  };
+
   // In-loop variables declared outside to save allocations.
   std::vector<uint64_t> features;
   std::vector<uint8_t> serialized_metadata;
@@ -603,6 +725,11 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) {
   const auto input_sink = GetInputSinkTo(emitted_inputs);
 
   for (size_t i = 0; i < num_inputs; i++) {
+    // NOTE: the values stored in stats here are slightly different than the
+    // definition of the original stats. Here the exec_time_usec will include
+    // the coverage processing.
+    ExecutionResult::Stats stats = {};
+    UsecSinceLast();
     auto blob = inputs_blobseq->Read();
     if (!blob.IsValid()) return;  // no more blobs to read.
     WorkerCheck(IsDataInput(blob), "Must read data input");
@@ -631,8 +758,10 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) {
         }};
 
     GetWorkerState().in_adapter_execute = true;
+    stats.prep_time_usec = UsecSinceLast();
     adapter.Execute(adapter.ctx, input, &feedback_sink);
     GetWorkerState().in_adapter_execute = false;
+    stats.exec_time_usec = UsecSinceLast();
     WORKER_CHECK_FOR_ERROR();
 
     serialized_metadata.clear();
@@ -681,6 +810,12 @@ void WorkerDoExecute(const FuzzTestAdapter& adapter) {
       WorkerLog("failed to write input metadata");
       break;
     }
+    stats.post_time_usec = UsecSinceLast();
+    stats.peak_rss_mb = GetPeakRSSMb();
+    if (!BatchResult::WriteStats(stats, *outputs_blobseq)) {
+      WorkerLog("failed to write stats");
+      break;
+    }
     if (!BatchResult::WriteInputEnd(*outputs_blobseq)) {
       WorkerLog("failed to write input end");
       break;
@@ -695,6 +830,72 @@ const char* FuzzTestWorkerGetTestName() {
   return test_name;
 }
 
+void HandlePersistentMode(const FuzzTestAdapter& adapter) {
+  auto* inputs_blobseq = GetInputsBlobSequence();
+  auto* outputs_blobseq = GetOutputsBlobSequence();
+  bool first = true;
+  while (true) {
+    PersistentModeRequest req;
+    if (!ReadAll(persistent_mode_socket, reinterpret_cast<char*>(&req), 1)) {
+      WorkerLog("Failed to read request from persistent mode socket: ",
+                LogErrNo{}, LogLnSync{});
+      return;
+    }
+    if (first) {
+      first = false;
+      WorkerLog("FuzzTest engine worker enter persistent mode", LogLnSync{});
+    } else {
+      // Reset stdout/stderr.
+      for (int fd = 1; fd <= 2; fd++) {
+        lseek(fd, 0, SEEK_SET);
+        // NOTE: Allow ftruncate() to fail by ignoring its return; that's okay
+        // to happen when the stdout/stderr are not redirected to a file.
+        (void)ftruncate(fd, 0);
+      }
+      WorkerLog(
+          "FuzzTest engine worker (",
+          req == PersistentModeRequest::kExit ? "exiting persistent mode"
+                                              : "persistent mode batch",
+          "); flags: ",
+          GetWorkerFlags().present
+              ? std::string_view{GetWorkerFlags().str, GetWorkerFlags().len}
+              : "",
+          LogLnSync{});
+    }
+    if (req == PersistentModeRequest::kExit) break;
+    WorkerCheck(req == PersistentModeRequest::kRunBatch,
+                "Unknown persistent mode request");
+
+    inputs_blobseq->Reset();
+    outputs_blobseq->Reset();
+
+    GetWorkerState().ResetForPersistentMode();
+
+    // Read the first blob. It indicates what further actions to take.
+    auto request_type_blob = inputs_blobseq->Read();
+    if (IsMutationRequest(request_type_blob)) {
+      inputs_blobseq->Reset();
+      WorkerDoMutate(adapter);
+    } else if (IsExecutionRequest(request_type_blob)) {
+      inputs_blobseq->Reset();
+      WorkerDoExecute(adapter);
+    } else {
+      WorkerCheck(false, "Unknown shmem request");
+    }
+
+    const int result =
+        GetWorkerState().has_finding.load(std::memory_order_relaxed)
+            ? EXIT_FAILURE
+            : EXIT_SUCCESS;
+    if (!WriteAll(persistent_mode_socket,
+                  reinterpret_cast<const char*>(&result), sizeof(result))) {
+      WorkerLog("Failed to write response to the persistent mode socket: ",
+                LogErrNo{}, LogLnSync{});
+      return;
+    }
+  }
+}
+
 FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) {
   const auto& flags = GetWorkerFlags();
   WorkerCheck(flags.present, "worker flags must present");
@@ -704,6 +905,12 @@ FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) {
   }
 
   const auto action = GetWorkerAction();
+  WorkerCheck(action.has_value(), "No worker action to run");
+
+  if (action == WorkerAction::kNoOp) {
+    return kFuzzTestWorkerSuccess;
+  }
+
   if (action == WorkerAction::kGetBinaryId) {
     WorkerDoGetBinaryId(manager);
     return kFuzzTestWorkerSuccess;
@@ -748,8 +955,8 @@ FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) {
   WorkerCheck(manager.ConstructAdapter != nullptr,
               "ConstructAdapter is not defined");
   FuzzTestAdapter adapter = {};
-  manager.ConstructAdapter(manager.ctx, /*diagnostic_sink=*/&diagnostic_sink,
-                           &adapter);
+  manager.ConstructAdapter(manager.ctx,
+                           /*diagnostic_sink=*/&diagnostic_sink, &adapter);
   WORKER_CHECK_FOR_ERROR();
   WorkerCheck(adapter.SetUpCoverageDomains != nullptr,
               "SetUpCoverageDomains must be defined");
@@ -764,7 +971,9 @@ FuzzTestWorkerStatus WorkerRun(const FuzzTestAdapterManager& manager) {
   WorkerCheck(adapter.FreeInput != nullptr, "FreeInput must be defined");
   WorkerCheck(adapter.FreeCtx != nullptr, "FreeCtx must be defined");
 
-  if (action == WorkerAction::kTestGetSeeds) {
+  if (persistent_mode_socket > 0) {
+    HandlePersistentMode(adapter);
+  } else if (action == WorkerAction::kTestGetSeeds) {
     WorkerDoGetSeeds(adapter);
   } else if (action == WorkerAction::kTestMutate) {
     WorkerDoMutate(adapter);
@@ -794,7 +1003,11 @@ using ::fuzztest::internal::WorkerRun;
 
 }  // namespace
 
-int FuzzTestWorkerIsRequired() { return GetWorkerFlags().present; }
+int FuzzTestWorkerIsRequired() {
+  static int result = GetWorkerFlags().present &&
+                      fuzztest::internal::GetWorkerAction().has_value();
+  return result;
+}
 
 FuzzTestWorkerStatus FuzzTestWorkerMaybeRun(
     const FuzzTestAdapterManager* manager) {
diff --git a/centipede/puzzles/run_puzzle.sh b/centipede/puzzles/run_puzzle.sh
index 0aee6243c..d6d12c853 100755
--- a/centipede/puzzles/run_puzzle.sh
+++ b/centipede/puzzles/run_puzzle.sh
@@ -94,7 +94,6 @@ function ExpectPerInputTimeout() {
 # Expects that Centipede found a per-batch timeout.
 function ExpectPerBatchTimeout() {
   echo "======= ${FUNCNAME[0]}"
-  fuzztest::internal::assert_regex_in_file "Per-batch timeout exceeded" "${log}"
   fuzztest::internal::assert_regex_in_file "Failure.*: per-batch-timeout-exceeded" "${log}"
   fuzztest::internal::assert_regex_in_file \
     "Failure applies to entire batch: not executing inputs one-by-one" "${log}"
diff --git a/centipede/runner.cc b/centipede/runner.cc
index e085b54e9..f3abed9eb 100644
--- a/centipede/runner.cc
+++ b/centipede/runner.cc
@@ -124,16 +124,15 @@ static void WaitWatchdogThreadIdle() {
 static void CheckWatchdogLimits() {
   const uint64_t curr_time = time(nullptr);
   struct Resource {
-    const char *what;
-    const char *units;
+    const char* what;
+    const char* units;
     uint64_t value;
     uint64_t limit;
     bool ignore_report;
-    const char *failure;
+    const char* failure;
   };
   const uint64_t input_start_time = state->input_start_time;
-  const uint64_t batch_start_time = state->batch_start_time;
-  if (input_start_time == 0 || batch_start_time == 0) return;
+  if (input_start_time == 0) return;
   const Resource resources[] = {
       {Resource{
           /*what=*/"Per-input timeout",
@@ -144,15 +143,6 @@ static void CheckWatchdogLimits() {
           state->run_time_flags.ignore_timeout_reports != 0,
           /*failure=*/kExecutionFailurePerInputTimeout.data(),
       }},
-      {Resource{
-          /*what=*/"Per-batch timeout",
-          /*units=*/"sec",
-          /*value=*/curr_time - batch_start_time,
-          /*limit=*/state->run_time_flags.timeout_per_batch,
-          /*ignore_report=*/
-          state->run_time_flags.ignore_timeout_reports != 0,
-          /*failure=*/kExecutionFailurePerBatchTimeout.data(),
-      }},
       {Resource{
           /*what=*/"RSS limit",
           /*units=*/"MB",
@@ -162,7 +152,7 @@ static void CheckWatchdogLimits() {
           /*failure=*/kExecutionFailureRssLimitExceeded.data(),
       }},
   };
-  for (const auto &resource : resources) {
+  for (const auto& resource : resources) {
     if (resource.limit != 0 && resource.value > resource.limit) {
       if (!watchdog_failure_found.exchange(true)) {
         if (resource.ignore_report) {
@@ -198,7 +188,7 @@ static void CheckWatchdogLimits() {
 
 // Watchdog thread. Periodically checks if it's time to abort due to a
 // timeout/OOM.
-[[noreturn]] static void *WatchdogThread(void *unused) {
+[[noreturn]] static void* WatchdogThread(void* unused) {
   // Since the watchdog is internal and does not execute user code, disable
   // SanCov tracing and TLS traversal.
   tls.traced = false;
@@ -240,10 +230,10 @@ __attribute__((noinline)) void CheckStackLimit(size_t stack_usage,
 void GlobalRunnerState::StartWatchdogThread() {
   fprintf(stderr,
           "Starting watchdog thread: timeout_per_input: %" PRIu64
-          " sec; timeout_per_batch: %" PRIu64 " sec; rss_limit_mb: %" PRIu64
-          " MB; stack_limit_kb: %" PRIu64 " KB\n",
+          " sec; rss_limit_mb: %" PRIu64 " MB; stack_limit_kb: %" PRIu64
+          " KB\n",
           run_time_flags.timeout_per_input.load(),
-          run_time_flags.timeout_per_batch, run_time_flags.rss_limit_mb.load(),
+          run_time_flags.rss_limit_mb.load(),
           state->run_time_flags.stack_limit_kb.load());
   pthread_t watchdog_thread;
   pthread_create(&watchdog_thread, nullptr, WatchdogThread, nullptr);
@@ -257,17 +247,12 @@ void GlobalRunnerState::StartWatchdogThread() {
 void GlobalRunnerState::ResetTimers() {
   const auto curr_time = time(nullptr);
   state->input_start_time = curr_time;
-  // batch_start_time is set only once -- just before the first input of the
-  // batch is about to start running.
-  if (batch_start_time == 0) {
-    batch_start_time = curr_time;
-  }
 }
 
 // Byte array mutation fallback for a custom mutator, as defined here:
 // https://github.com/google/fuzzing/blob/master/docs/structure-aware-fuzzing.md
 extern "C" __attribute__((weak)) size_t
-CentipedeLLVMFuzzerMutateCallback(uint8_t *data, size_t size, size_t max_size) {
+CentipedeLLVMFuzzerMutateCallback(uint8_t* data, size_t size, size_t max_size) {
   // TODO(kcc): [as-needed] fix the interface mismatch.
   // LLVMFuzzerMutate is an array-based interface (for compatibility reasons)
   // while ByteArray has a vector-based interface.
@@ -290,7 +275,7 @@ CentipedeLLVMFuzzerMutateCallback(uint8_t *data, size_t size, size_t max_size) {
   return array.size();
 }
 
-extern "C" size_t LLVMFuzzerMutate(uint8_t *data, size_t size,
+extern "C" size_t LLVMFuzzerMutate(uint8_t* data, size_t size,
                                    size_t max_size) {
   return CentipedeLLVMFuzzerMutateCallback(data, size, max_size);
 }
@@ -298,7 +283,7 @@ extern "C" size_t LLVMFuzzerMutate(uint8_t *data, size_t size,
 // An arbitrary large size for input data.
 static const size_t kMaxDataSize = 1 << 20;
 
-static void WriteFeaturesToFile(FILE *file, const feature_t *features,
+static void WriteFeaturesToFile(FILE* file, const feature_t* features,
                                 size_t size) {
   if (!size) return;
   auto bytes_written = fwrite(features, 1, sizeof(features[0]) * size, file);
@@ -322,19 +307,33 @@ static void PrepareCoverage(bool full_clear) {
   PrepareSancov(full_clear);
 }
 
-void RunnerCallbacks::GetSeeds(std::function<void(ByteSpan)> seed_callback) {
-  seed_callback({0});
+void RunnerCallbacks::GetPresetSeedInputs(
+    std::function<void(void*)> seed_callback) {}
+
+void RunnerCallbacks::GetRandomSeedInput(
+    std::function<void(void*)> seed_callback) {
+  seed_callback(DeserializeInput({0}));
 }
 
 std::string RunnerCallbacks::GetSerializedTargetConfig() { return ""; }
 
-bool RunnerCallbacks::Mutate(
-    absl::Span<const MutationInputRef> /*inputs*/, size_t /*num_mutants*/,
-    std::function<void(MutantRef)> /*new_mutant_callback*/) {
+void* RunnerCallbacks::Mutate(void* origin,
+                              const ExecutionMetadata& origin_metadata) {
+  RunnerCheck(HasCustomMutator(),
+              "RunnerCallbacks::Mutate called despite HasCustomMutator() "
+              "returning false. This is a runner bug!");
   RunnerCheck(!HasCustomMutator(),
               "Class deriving from RunnerCallbacks must implement Mutate() if "
               "HasCustomMutator() returns true.");
-  return true;
+  // Should be unreachable here.
+  return nullptr;
+}
+
+void* RunnerCallbacks::CrossOver(void* origin,
+                                 const ExecutionMetadata& origin_metadata,
+                                 void* other,
+                                 const ExecutionMetadata& other_metadata) {
+  return Mutate(origin, origin_metadata);
 }
 
 class LegacyRunnerCallbacks : public RunnerCallbacks {
@@ -346,10 +345,11 @@ class LegacyRunnerCallbacks : public RunnerCallbacks {
         custom_mutator_cb_(custom_mutator_cb),
         custom_crossover_cb_(custom_crossover_cb) {}
 
-  bool Execute(ByteSpan input) override {
+  bool Execute(void* input) override {
     PrintErrorAndExitIf(test_one_input_cb_ == nullptr,
                         "missing test_on_input_cb");
-    const int retval = test_one_input_cb_(input.data(), input.size());
+    const auto* input_ba = reinterpret_cast<const ByteArray*>(input);
+    const int retval = test_one_input_cb_(input_ba->data(), input_ba->size());
     PrintErrorAndExitIf(
         retval != -1 && retval != 0,
         "test_on_input_cb returns invalid value other than -1 and 0");
@@ -360,8 +360,15 @@ class LegacyRunnerCallbacks : public RunnerCallbacks {
     return custom_mutator_cb_ != nullptr;
   }
 
-  bool Mutate(absl::Span<const MutationInputRef> inputs, size_t num_mutants,
-              std::function<void(MutantRef)> new_mutant_callback) override;
+  void SerializeInput(void* input,
+                      std::function<void(ByteSpan)> bytes_sink) override;
+  void* DeserializeInput(ByteSpan input_bytes) override;
+  void FreeInput(void* input) override;
+
+  void* Mutate(void* origin, const ExecutionMetadata& origin_metadata) override;
+  void* CrossOver(void* origin, const ExecutionMetadata& origin_metadata,
+                  void* other,
+                  const ExecutionMetadata& other_metadata) override;
 
  private:
   FuzzerTestOneInputCallback test_one_input_cb_;
@@ -377,8 +384,7 @@ std::unique_ptr<RunnerCallbacks> CreateLegacyRunnerCallbacks(
       test_one_input_cb, custom_mutator_cb, custom_crossover_cb);
 }
 
-static void RunOneInput(const uint8_t *data, size_t size,
-                        RunnerCallbacks &callbacks) {
+static void RunOneInput(void* input, RunnerCallbacks& callbacks) {
   state->stats = {};
   size_t last_time_usec = 0;
   auto UsecSinceLast = [&last_time_usec]() {
@@ -391,7 +397,7 @@ static void RunOneInput(const uint8_t *data, size_t size,
   PrepareCoverage(/*full_clear=*/false);
   state->stats.prep_time_usec = UsecSinceLast();
   state->ResetTimers();
-  int target_return_value = callbacks.Execute({data, size}) ? 0 : -1;
+  int target_return_value = callbacks.Execute(input) ? 0 : -1;
   state->stats.exec_time_usec = UsecSinceLast();
   CheckWatchdogLimits();
   if (fuzztest::internal::state->input_start_time.exchange(0) != 0) {
@@ -405,18 +411,20 @@ static void RunOneInput(const uint8_t *data, size_t size,
 // Runs one input provided in file `input_path`.
 // Produces coverage data in file `input_path`-features.
 __attribute__((noinline))  // so that we see it in profile.
-static void ReadOneInputExecuteItAndDumpCoverage(const char *input_path,
-                                                 RunnerCallbacks &callbacks) {
+static void ReadOneInputExecuteItAndDumpCoverage(const char* input_path,
+                                                 RunnerCallbacks& callbacks) {
   // Read the input.
   auto data = ReadBytesFromFilePath<uint8_t>(input_path);
 
-  RunOneInput(data.data(), data.size(), callbacks);
+  void* input = callbacks.DeserializeInput(data);
+  RunOneInput(input, callbacks);
+  callbacks.FreeInput(input);
 
   // Dump features to a file.
   char features_file_path[PATH_MAX];
   snprintf(features_file_path, sizeof(features_file_path), "%s-features",
            input_path);
-  FILE *features_file = fopen(features_file_path, "w");
+  FILE* features_file = fopen(features_file_path, "w");
   PrintErrorAndExitIf(features_file == nullptr, "can't open coverage file");
 
   const SanCovRuntimeRawFeatureParts sancov_features =
@@ -428,13 +436,13 @@ static void ReadOneInputExecuteItAndDumpCoverage(const char *input_path,
 
 // Starts sending the outputs (coverage, etc.) to `outputs_blobseq`.
 // Returns true on success.
-static bool StartSendingOutputsToEngine(BlobSequence &outputs_blobseq) {
+static bool StartSendingOutputsToEngine(BlobSequence& outputs_blobseq) {
   return BatchResult::WriteInputBegin(outputs_blobseq);
 }
 
 // Copy all the sancov features to `data` with given `capacity` in bytes.
 // Returns the byte size of sancov features.
-static size_t CopyFeatures(uint8_t *data, size_t capacity) {
+static size_t CopyFeatures(uint8_t* data, size_t capacity) {
   const SanCovRuntimeRawFeatureParts sancov_features =
       SanCovRuntimeGetFeatures();
   const size_t features_len_in_bytes =
@@ -446,7 +454,7 @@ static size_t CopyFeatures(uint8_t *data, size_t capacity) {
 
 // Finishes sending the outputs (coverage, etc.) to `outputs_blobseq`.
 // Returns true on success.
-static bool FinishSendingOutputsToEngine(BlobSequence &outputs_blobseq) {
+static bool FinishSendingOutputsToEngine(BlobSequence& outputs_blobseq) {
   {
     LockGuard lock(state->execution_result_override_mu);
     bool has_overridden_execution_result = false;
@@ -491,30 +499,31 @@ static bool FinishSendingOutputsToEngine(BlobSequence &outputs_blobseq) {
 // Handles an ExecutionRequest, see RequestExecution(). Reads inputs from
 // `inputs_blobseq`, runs them, saves coverage features to `outputs_blobseq`.
 // Returns EXIT_SUCCESS on success and EXIT_FAILURE otherwise.
-static int ExecuteInputsFromShmem(BlobSequence &inputs_blobseq,
-                                  BlobSequence &outputs_blobseq,
-                                  RunnerCallbacks &callbacks) {
+static int ExecuteInputsFromShmem(BlobSequence& inputs_blobseq,
+                                  BlobSequence& outputs_blobseq,
+                                  RunnerCallbacks& callbacks) {
   size_t num_inputs = 0;
   if (!IsExecutionRequest(inputs_blobseq.Read())) return EXIT_FAILURE;
   if (!IsNumInputs(inputs_blobseq.Read(), num_inputs)) return EXIT_FAILURE;
 
-  CentipedeBeginExecutionBatch();
-
+  std::vector<void*> inputs;
   for (size_t i = 0; i < num_inputs; i++) {
     auto blob = inputs_blobseq.Read();
     // TODO(kcc): distinguish bad input from end of stream.
-    if (!blob.IsValid()) return EXIT_SUCCESS;  // no more blobs to read.
+    if (!blob.IsValid()) break;  // no more blobs to read.
     if (!IsDataInput(blob)) return EXIT_FAILURE;
 
     // TODO(kcc): [impl] handle sizes larger than kMaxDataSize.
     size_t size = std::min(kMaxDataSize, blob.size);
-    // Copy from blob to data so that to not pass the shared memory further.
-    std::vector<uint8_t> data(blob.data, blob.data + size);
+    inputs.push_back(callbacks.DeserializeInput({blob.data, size}));
+  }
+
+  CentipedeBeginExecutionBatch();
 
-    // Starting execution of one more input.
+  for (void* input : inputs) {
     if (!StartSendingOutputsToEngine(outputs_blobseq)) break;
 
-    RunOneInput(data.data(), data.size(), callbacks);
+    RunOneInput(input, callbacks);
 
     if (state->has_failure_description.load()) break;
 
@@ -523,13 +532,24 @@ static int ExecuteInputsFromShmem(BlobSequence &inputs_blobseq,
 
   CentipedeEndExecutionBatch();
 
+  for (void* input : inputs) {
+    callbacks.FreeInput(input);
+  }
+
   return state->has_failure_description.load() ? EXIT_FAILURE : EXIT_SUCCESS;
 }
 
 // Dumps seed inputs to `output_dir`. Also see `GetSeedsViaExternalBinary()`.
-static void DumpSeedsToDir(RunnerCallbacks &callbacks, const char *output_dir) {
+static void DumpSeedsToDir(RunnerCallbacks& callbacks, const char* output_dir) {
   size_t seed_index = 0;
-  callbacks.GetSeeds([&](ByteSpan seed) {
+  // Declare it on the outer scope to save allocations.
+  ByteArray serialized;
+  auto dump_seed_callback = [&](void* seed) {
+    serialized.clear();
+    callbacks.SerializeInput(seed, [&](ByteSpan bytes) {
+      serialized.insert(serialized.end(), bytes.begin(), bytes.end());
+    });
+    callbacks.FreeInput(seed);
     // Cap seed index within 9 digits. If this was triggered, the dumping would
     // take forever..
     if (seed_index >= 1000000000) return;
@@ -538,22 +558,29 @@ static void DumpSeedsToDir(RunnerCallbacks &callbacks, const char *output_dir) {
         snprintf(seed_path_buf, PATH_MAX, "%s/%09lu", output_dir, seed_index);
     PrintErrorAndExitIf(num_path_chars >= PATH_MAX,
                         "seed path reaches PATH_MAX");
-    FILE *output_file = fopen(seed_path_buf, "w");
+    FILE* output_file = fopen(seed_path_buf, "w");
     const size_t num_bytes_written =
-        fwrite(seed.data(), 1, seed.size(), output_file);
-    PrintErrorAndExitIf(num_bytes_written != seed.size(),
+        fwrite(serialized.data(), 1, serialized.size(), output_file);
+    PrintErrorAndExitIf(num_bytes_written != serialized.size(),
                         "wrong number of bytes written for cf table");
     fclose(output_file);
     ++seed_index;
-  });
+  };
+  callbacks.GetPresetSeedInputs(dump_seed_callback);
+  while (seed_index < 32) {
+    const auto prev_index = seed_index;
+    callbacks.GetRandomSeedInput(dump_seed_callback);
+    RunnerCheck(seed_index == prev_index + 1,
+                "GetRandomSeedInput must provide at least one input");
+  }
 }
 
 // Dumps serialized target config to `output_file_path`. Also see
 // `GetSerializedTargetConfigViaExternalBinary()`.
-static void DumpSerializedTargetConfigToFile(RunnerCallbacks &callbacks,
-                                             const char *output_file_path) {
+static void DumpSerializedTargetConfigToFile(RunnerCallbacks& callbacks,
+                                             const char* output_file_path) {
   const std::string config = callbacks.GetSerializedTargetConfig();
-  FILE *output_file = fopen(output_file_path, "w");
+  FILE* output_file = fopen(output_file_path, "w");
   const size_t num_bytes_written =
       fwrite(config.data(), 1, config.size(), output_file);
   PrintErrorAndExitIf(
@@ -566,6 +593,51 @@ static void DumpSerializedTargetConfigToFile(RunnerCallbacks &callbacks,
 // TODO(kcc): [as-needed] optionally pass an external seed.
 static unsigned GetRandomSeed() { return time(nullptr); }
 
+void MutateInputs(RunnerCallbacks& callbacks,
+                  absl::Span<const MutationInputRef> inputs, size_t num_mutants,
+                  std::function<void(MutantRef)> new_mutant_callback) {
+  unsigned int seed = GetRandomSeed();
+  std::vector<void*> input_objects;
+  input_objects.resize(inputs.size(), nullptr);
+  for (size_t i = 0; i < inputs.size(); ++i) {
+    input_objects[i] = callbacks.DeserializeInput(inputs[i].data);
+  }
+  ByteArray mutant_data;
+  ExecutionMetadata empty_metadata;
+  for (size_t i = 0; i < num_mutants; ++i) {
+    const size_t origin_index = rand_r(&seed) % inputs.size();
+    void* mutant = nullptr;
+    if (rand_r(&seed) % 100 < state->run_time_flags.crossover_level) {
+      // Perform crossover `crossover_level`% of the time.
+      const size_t other_index = rand_r(&seed) % inputs.size();
+      mutant = callbacks.CrossOver(input_objects[origin_index],
+                                   inputs[origin_index].metadata != nullptr
+                                       ? *inputs[origin_index].metadata
+                                       : empty_metadata,
+                                   input_objects[other_index],
+                                   inputs[other_index].metadata != nullptr
+                                       ? *inputs[other_index].metadata
+                                       : empty_metadata);
+    } else {
+      mutant = callbacks.Mutate(input_objects[origin_index],
+                                inputs[origin_index].metadata != nullptr
+                                    ? *inputs[origin_index].metadata
+                                    : empty_metadata);
+    }
+    mutant_data.clear();
+    callbacks.SerializeInput(mutant, [&mutant_data](ByteSpan bytes) {
+      mutant_data.insert(mutant_data.end(), bytes.begin(), bytes.end());
+    });
+    new_mutant_callback(
+        MutantRef{{(unsigned char*)mutant_data.data(), mutant_data.size()},
+                  origin_index});
+    callbacks.FreeInput(mutant);
+  }
+  for (void* input_object : input_objects) {
+    callbacks.FreeInput(input_object);
+  }
+}
+
 // Handles a Mutation Request, see RequestMutation().
 // Mutates inputs read from `inputs_blobseq`,
 // writes the mutants to `outputs_blobseq`
@@ -575,9 +647,9 @@ static unsigned GetRandomSeed() { return time(nullptr); }
 // returns EXIT_FAILURE.
 //
 // TODO(kcc): [impl] make use of custom_crossover_cb, if available.
-static int MutateInputsFromShmem(BlobSequence &inputs_blobseq,
-                                 BlobSequence &outputs_blobseq,
-                                 RunnerCallbacks &callbacks) {
+static int MutateInputsFromShmem(BlobSequence& inputs_blobseq,
+                                 BlobSequence& outputs_blobseq,
+                                 RunnerCallbacks& callbacks) {
   // Read max_num_mutants.
   size_t num_mutants = 0;
   size_t num_inputs = 0;
@@ -625,60 +697,96 @@ static int MutateInputsFromShmem(BlobSequence &inputs_blobseq,
   }
   if (!callbacks.HasCustomMutator()) return EXIT_SUCCESS;
 
-  if (!callbacks.Mutate(input_refs, num_mutants, [&](MutantRef mutant) {
-        MutationResult::WriteMutant(mutant, outputs_blobseq);
-      })) {
-    return EXIT_FAILURE;
+  {
+    bool succ = true;
+    MutateInputs(callbacks, input_refs, num_mutants, [&](MutantRef mutant) {
+      succ = succ && MutationResult::WriteMutant(mutant, outputs_blobseq);
+    });
+    if (!succ) return EXIT_FAILURE;
   }
+
   return EXIT_SUCCESS;
 }
 
-bool LegacyRunnerCallbacks::Mutate(
-    absl::Span<const MutationInputRef> inputs, size_t num_mutants,
-    std::function<void(MutantRef)> new_mutant_callback) {
-  if (custom_mutator_cb_ == nullptr) return false;
-  unsigned int seed = GetRandomSeed();
-  const size_t num_inputs = inputs.size();
+void LegacyRunnerCallbacks::SerializeInput(
+    void* input, std::function<void(ByteSpan)> bytes_sink) {
+  const auto* input_object = reinterpret_cast<const ByteArray*>(input);
+  bytes_sink(*input_object);
+}
+
+void* LegacyRunnerCallbacks::DeserializeInput(ByteSpan input_bytes) {
+  return reinterpret_cast<void*>(
+      new ByteArray{input_bytes.begin(), input_bytes.end()});
+}
+
+void LegacyRunnerCallbacks::FreeInput(void* input) {
+  delete reinterpret_cast<const ByteArray*>(input);
+}
+
+void* LegacyRunnerCallbacks::Mutate(void* origin,
+                                    const ExecutionMetadata& origin_metadata) {
+  const auto* origin_ba = reinterpret_cast<const ByteArray*>(origin);
+  state->byte_array_mutator->SetMetadata(origin_metadata);
   const size_t max_mutant_size = state->run_time_flags.max_len;
-  constexpr size_t kAverageMutationAttempts = 2;
-  // Reused across iterations to save memory allocations.
-  Mutant mutant;
-  for (size_t attempt = 0, num_outputs = 0;
-       attempt < num_mutants * kAverageMutationAttempts &&
-       num_outputs < num_mutants;
-       ++attempt) {
-    mutant.origin = rand_r(&seed) % num_inputs;
-    const auto& input_data = inputs[mutant.origin].data;
-
-    size_t size = std::min(input_data.size(), max_mutant_size);
-    mutant.data.resize(max_mutant_size);
-    std::copy(input_data.cbegin(), input_data.cbegin() + size,
-              mutant.data.begin());
-    size_t new_size = 0;
-    if ((custom_crossover_cb_ != nullptr) &&
-        rand_r(&seed) % 100 < state->run_time_flags.crossover_level) {
-      // Perform crossover `crossover_level`% of the time.
-      const auto &other_data = inputs[rand_r(&seed) % num_inputs].data;
-      new_size = custom_crossover_cb_(input_data.data(), input_data.size(),
-                                      other_data.data(), other_data.size(),
-                                      mutant.data.data(), max_mutant_size,
-                                      rand_r(&seed));
-    } else {
-      new_size = custom_mutator_cb_(mutant.data.data(), size, max_mutant_size,
-                                    rand_r(&seed));
-    }
-    if (new_size == 0) continue;
-    if (new_size > max_mutant_size) new_size = max_mutant_size;
-    mutant.data.resize(new_size);
-    new_mutant_callback(MutantRef{mutant});
-    ++num_outputs;
+  const size_t size = std::min(max_mutant_size, origin_ba->size());
+  auto* mutant = new ByteArray();
+  mutant->resize(size);
+  std::copy(origin_ba->begin(), origin_ba->begin() + size, mutant->begin());
+  static unsigned int seed = GetRandomSeed();
+  size_t new_size = 0;
+  mutant->resize(max_mutant_size);
+  new_size =
+      custom_mutator_cb_(mutant->data(), size, max_mutant_size, rand_r(&seed));
+  if (new_size == 0) {
+    new_size = 1;
+    mutant->assign({0});
+  } else if (new_size == max_mutant_size) {
+    new_size = max_mutant_size;
+    mutant->resize(new_size);
+  } else {
+    mutant->resize(new_size);
   }
-  return true;
+  return reinterpret_cast<void*>(mutant);
+}
+
+void* LegacyRunnerCallbacks::CrossOver(
+    void* origin, const ExecutionMetadata& origin_metadata, void* other,
+    const ExecutionMetadata& other_metadata) {
+  const auto* origin_ba = reinterpret_cast<const ByteArray*>(origin);
+  const auto* other_ba = reinterpret_cast<const ByteArray*>(other);
+  // Do not use other_metadata.
+  state->byte_array_mutator->SetMetadata(origin_metadata);
+  const size_t max_mutant_size = state->run_time_flags.max_len;
+  const size_t size = std::min(max_mutant_size, origin_ba->size());
+  auto* mutant = new ByteArray();
+  mutant->resize(size);
+  std::copy(origin_ba->begin(), origin_ba->begin() + size, mutant->begin());
+  static unsigned int seed = GetRandomSeed();
+  size_t new_size = 0;
+  if (custom_crossover_cb_ != nullptr) {
+    mutant->resize(max_mutant_size);
+    new_size = custom_crossover_cb_(
+        origin_ba->data(), origin_ba->size(), other_ba->data(),
+        other_ba->size(), mutant->data(), max_mutant_size, rand_r(&seed));
+  } else {
+    state->byte_array_mutator->CrossOver(*mutant, *other_ba);
+    new_size = mutant->size();
+  }
+  if (new_size == 0) {
+    new_size = 1;
+    mutant->assign({0});
+  } else if (new_size == max_mutant_size) {
+    new_size = max_mutant_size;
+    mutant->resize(new_size);
+  } else {
+    mutant->resize(new_size);
+  }
+  return reinterpret_cast<void*>(mutant);
 }
 
 // Returns the current process VmSize, in bytes.
 static size_t GetVmSizeInBytes() {
-  FILE *f = fopen("/proc/self/statm", "r");  // man proc
+  FILE* f = fopen("/proc/self/statm", "r");  // man proc
   if (!f) return 0;
   size_t vm_size = 0;
   // NOTE: Ignore any (unlikely) failures to suppress a compiler warning.
@@ -889,7 +997,7 @@ static int HandlePersistentMode(RunnerCallbacks& callbacks,
 //  Default: Execute ReadOneInputExecuteItAndDumpCoverage() for all inputs.//
 //
 //  Note: argc/argv are used for only ReadOneInputExecuteItAndDumpCoverage().
-int RunnerMain(int argc, char **argv, RunnerCallbacks &callbacks) {
+int RunnerMain(int argc, char** argv, RunnerCallbacks& callbacks) {
   state->centipede_runner_main_executed = true;
 
   fprintf(stderr, "Centipede fuzz target runner; argv[0]: %s flags: %s\n",
@@ -930,7 +1038,7 @@ int RunnerMain(int argc, char **argv, RunnerCallbacks &callbacks) {
 }  // namespace fuzztest::internal
 
 extern "C" int LLVMFuzzerRunDriver(
-    int *absl_nonnull argc, char ***absl_nonnull argv,
+    int* absl_nonnull argc, char*** absl_nonnull argv,
     FuzzerTestOneInputCallback test_one_input_cb) {
   if (LLVMFuzzerInitialize) LLVMFuzzerInitialize(argc, argv);
   return RunnerMain(*argc, *argv,
@@ -960,9 +1068,9 @@ extern "C" void CentipedeSetTimeoutPerInput(uint64_t timeout_per_input) {
       timeout_per_input;
 }
 
-extern "C" __attribute__((weak)) const char *absl_nullable
+extern "C" __attribute__((weak)) const char* absl_nullable
 CentipedeGetRunnerFlags() {
-  if (const char *runner_flags_env = getenv("CENTIPEDE_RUNNER_FLAGS"))
+  if (const char* runner_flags_env = getenv("CENTIPEDE_RUNNER_FLAGS"))
     return strdup(runner_flags_env);
   return nullptr;
 }
@@ -994,7 +1102,6 @@ extern "C" void CentipedeEndExecutionBatch() {
   }
   in_execution_batch = false;
   fuzztest::internal::state->input_start_time = 0;
-  fuzztest::internal::state->batch_start_time = 0;
 }
 
 extern "C" void CentipedePrepareProcessing() {
@@ -1015,7 +1122,7 @@ extern "C" int CentipedeSetCurrentThreadTraced(int traced) {
   return old_traced;
 }
 
-extern "C" size_t CentipedeGetExecutionResult(uint8_t *data, size_t capacity) {
+extern "C" size_t CentipedeGetExecutionResult(uint8_t* data, size_t capacity) {
   fuzztest::internal::BlobSequence outputs_blobseq(data, capacity);
   if (!fuzztest::internal::StartSendingOutputsToEngine(outputs_blobseq))
     return 0;
@@ -1024,11 +1131,11 @@ extern "C" size_t CentipedeGetExecutionResult(uint8_t *data, size_t capacity) {
   return outputs_blobseq.offset();
 }
 
-extern "C" size_t CentipedeGetCoverageData(uint8_t *data, size_t capacity) {
+extern "C" size_t CentipedeGetCoverageData(uint8_t* data, size_t capacity) {
   return fuzztest::internal::CopyFeatures(data, capacity);
 }
 
-extern "C" void CentipedeSetExecutionResult(const uint8_t *data, size_t size) {
+extern "C" void CentipedeSetExecutionResult(const uint8_t* data, size_t size) {
   using fuzztest::internal::state;
   fuzztest::internal::LockGuard lock(state->execution_result_override_mu);
   if (!state->execution_result_override)
@@ -1036,14 +1143,14 @@ extern "C" void CentipedeSetExecutionResult(const uint8_t *data, size_t size) {
   state->execution_result_override->ClearAndResize(1);
   if (data == nullptr) return;
   // Removing const here should be fine as we don't write to `blobseq`.
-  fuzztest::internal::BlobSequence blobseq(const_cast<uint8_t *>(data), size);
+  fuzztest::internal::BlobSequence blobseq(const_cast<uint8_t*>(data), size);
   state->execution_result_override->Read(blobseq);
   fuzztest::internal::RunnerCheck(
       state->execution_result_override->num_outputs_read() == 1,
       "Failed to set execution result from CentipedeSetExecutionResult");
 }
 
-extern "C" void CentipedeSetFailureDescription(const char *description) {
+extern "C" void CentipedeSetFailureDescription(const char* description) {
   using fuzztest::internal::state;
   if (state->failure_description_path == nullptr) return;
   if (state->has_failure_description.exchange(true)) return;
diff --git a/centipede/runner.h b/centipede/runner.h
index fbe094166..5ddc3b3e5 100644
--- a/centipede/runner.h
+++ b/centipede/runner.h
@@ -33,7 +33,6 @@ namespace fuzztest::internal {
 // Flags derived from CENTIPEDE_RUNNER_FLAGS.
 struct RunTimeFlags {
   std::atomic<uint64_t> timeout_per_input;
-  uint64_t timeout_per_batch;
   std::atomic<uint64_t> rss_limit_mb;
   uint64_t crossover_level;
   uint64_t ignore_timeout_reports : 1;
@@ -67,7 +66,6 @@ struct GlobalRunnerState {
   // flags can change later (if wrapped with std::atomic).
   RunTimeFlags run_time_flags = {
       /*timeout_per_input=*/flag_helper.HasIntFlag(":timeout_per_input=", 0),
-      /*timeout_per_batch=*/flag_helper.HasIntFlag(":timeout_per_batch=", 0),
       /*rss_limit_mb=*/flag_helper.HasIntFlag(":rss_limit_mb=", 0),
       /*crossover_level=*/flag_helper.HasIntFlag(":crossover_level=", 50),
       /*ignore_timeout_reports=*/
@@ -112,10 +110,6 @@ struct GlobalRunnerState {
   // time.
   std::atomic<time_t> input_start_time;
 
-  // Per-batch timer. Initially, zero. ResetInputTimer() sets it to the current
-  // time before the first input and never resets it.
-  std::atomic<time_t> batch_start_time;
-
   // The Watchdog thread sets this to true.
   std::atomic<bool> watchdog_thread_started;
 };
diff --git a/centipede/runner_interface.h b/centipede/runner_interface.h
index c622daf28..7d78551c8 100644
--- a/centipede/runner_interface.h
+++ b/centipede/runner_interface.h
@@ -26,36 +26,37 @@
 
 #include "absl/base/nullability.h"
 #include "absl/types/span.h"
+#include "./centipede/execution_metadata.h"
 #include "./centipede/mutation_data.h"
 #include "./common/defs.h"
 
 // Typedefs for the libFuzzer API, https://llvm.org/docs/LibFuzzer.html
-using FuzzerTestOneInputCallback = int (*)(const uint8_t *data, size_t size);
-using FuzzerInitializeCallback = int (*)(int *argc, char ***argv);
-using FuzzerCustomMutatorCallback = size_t (*)(uint8_t *data, size_t size,
+using FuzzerTestOneInputCallback = int (*)(const uint8_t* data, size_t size);
+using FuzzerInitializeCallback = int (*)(int* argc, char*** argv);
+using FuzzerCustomMutatorCallback = size_t (*)(uint8_t* data, size_t size,
                                                size_t max_size,
                                                unsigned int seed);
 using FuzzerCustomCrossOverCallback = size_t (*)(
-    const uint8_t *data1, size_t size1, const uint8_t *data2, size_t size2,
-    uint8_t *out, size_t max_out_size, unsigned int seed);
+    const uint8_t* data1, size_t size1, const uint8_t* data2, size_t size2,
+    uint8_t* out, size_t max_out_size, unsigned int seed);
 
 // This is the header-less interface of libFuzzer, see
 // https://llvm.org/docs/LibFuzzer.html.
 extern "C" {
-int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size);
-__attribute__((weak)) int LLVMFuzzerInitialize(int *absl_nonnull argc,
-                                               char ***absl_nonnull argv);
-__attribute__((weak)) size_t LLVMFuzzerCustomMutator(uint8_t *data, size_t size,
+int LLVMFuzzerTestOneInput(const uint8_t* data, size_t size);
+__attribute__((weak)) int LLVMFuzzerInitialize(int* absl_nonnull argc,
+                                               char*** absl_nonnull argv);
+__attribute__((weak)) size_t LLVMFuzzerCustomMutator(uint8_t* data, size_t size,
                                                      size_t max_size,
                                                      unsigned int seed);
 __attribute__((weak)) size_t LLVMFuzzerCustomCrossOver(
-    const uint8_t *data1, size_t size1, const uint8_t *data2, size_t size2,
-    uint8_t *out, size_t max_out_size, unsigned int seed);
+    const uint8_t* data1, size_t size1, const uint8_t* data2, size_t size2,
+    uint8_t* out, size_t max_out_size, unsigned int seed);
 }  // extern "C"
 
 // https://llvm.org/docs/LibFuzzer.html#using-libfuzzer-as-a-library
 extern "C" int LLVMFuzzerRunDriver(
-    int *absl_nonnull argc, char ***absl_nonnull argv,
+    int* absl_nonnull argc, char*** absl_nonnull argv,
     FuzzerTestOneInputCallback test_one_input_cb);
 
 // Reconfigures the RSS limit to `rss_limit_mb` - 0 indicates no limit.
@@ -73,10 +74,10 @@ extern "C" void CentipedeSetTimeoutPerInput(uint64_t timeout_per_input);
 //
 // It should return either a nullptr or a constant string that is valid
 // throughout the entire process life-time.
-extern "C" const char *absl_nullable CentipedeGetRunnerFlags();
+extern "C" const char* absl_nullable CentipedeGetRunnerFlags();
 
 // An overridable function to override `LLVMFuzzerMutate` behavior.
-extern "C" size_t CentipedeLLVMFuzzerMutateCallback(uint8_t *data, size_t size,
+extern "C" size_t CentipedeLLVMFuzzerMutateCallback(uint8_t* data, size_t size,
                                                     size_t max_size);
 
 // Prepares to run a batch of test executions that ends with calling
@@ -109,26 +110,26 @@ extern "C" int CentipedeSetCurrentThreadTraced(int traced);
 // processing an input. This function saves the data to the provided buffer and
 // returns the size of the saved data. It may be called after
 // CentipedeFinalizeProcessing().
-extern "C" size_t CentipedeGetExecutionResult(uint8_t *data, size_t capacity);
+extern "C" size_t CentipedeGetExecutionResult(uint8_t* data, size_t capacity);
 
 // Retrieves the coverage data collected during the processing of an input.
 // This function saves the raw coverage data to the provided buffer and returns
 // the size of the saved data. It may be called after
 // CentipedeFinalizeProcessing().
-extern "C" size_t CentipedeGetCoverageData(uint8_t *data, size_t capacity);
+extern "C" size_t CentipedeGetCoverageData(uint8_t* data, size_t capacity);
 
 // Set the current execution result to the opaque memory `data` with `size`.
 // Such data is retrieved using `CentipedeGetExecutionResult`, possibly from
 // another process. When `data` is `nullptr`, will set the execution result to
 // "empty" with no features or metadata.
-extern "C" void CentipedeSetExecutionResult(const uint8_t *data, size_t size);
+extern "C" void CentipedeSetExecutionResult(const uint8_t* data, size_t size);
 
 // Set the failure description for the runner to propagate further. Only the
 // description from the first call will be used.
 //
 // If used during executing batch inputs, the rest of the inputs would be
 // skipped and the batch would be considered as failed.
-extern "C" void CentipedeSetFailureDescription(const char *description);
+extern "C" void CentipedeSetFailureDescription(const char* description);
 
 namespace fuzztest::internal {
 
@@ -140,23 +141,29 @@ class RunnerCallbacks {
  public:
   // Attempts to execute the test logic using `input`, and returns false if the
   // input should be ignored from the corpus, true otherwise.
-  virtual bool Execute(ByteSpan input) = 0;
-  // Generates seed inputs by calling `seed_callback` for each input.
+  virtual bool Execute(void* input) = 0;
+  // Provide the preset seed inputs by calling `seed_callback` on each of them.
+  // The default implementation provides no inputs.
+  virtual void GetPresetSeedInputs(std::function<void(void*)> seed_callback);
+  // Generates a random seed input by calling `seed_callback` on it.
   // The default implementation generates a single-byte input {0}.
-  virtual void GetSeeds(std::function<void(ByteSpan)> seed_callback);
+  virtual void GetRandomSeedInput(std::function<void(void*)> seed_callback);
   // Returns the serialized configuration from the test target. The default
   // implementation returns the empty string.
   virtual std::string GetSerializedTargetConfig();
   // Returns true if and only if the test target has a custom mutator.
   virtual bool HasCustomMutator() const = 0;
-  // Generates at most `num_mutants` mutants by calling `new_mutant_callback`
-  // for each mutant. Returns true on success, false otherwise.
-  //
-  // TODO(xinhaoyuan): Consider supporting only_shrink to speed up
-  // input shrinking.
-  virtual bool Mutate(absl::Span<const MutationInputRef> inputs,
-                      size_t num_mutants,
-                      std::function<void(MutantRef)> new_mutant_callback);
+  // Mutates `origin` with `origin_metadata`. Must be overridden if
+  // `HasCustomMutator()` returns true.
+  virtual void* Mutate(void* origin, const ExecutionMetadata& origin_metadata);
+  // Default is return Mutate(origin, origin_metadata)
+  virtual void* CrossOver(void* origin,
+                          const ExecutionMetadata& origin_metadata, void* other,
+                          const ExecutionMetadata& other_metadata);
+  virtual void SerializeInput(void* input,
+                              std::function<void(ByteSpan)> bytes_sink) = 0;
+  virtual void* DeserializeInput(ByteSpan input_bytes) = 0;
+  virtual void FreeInput(void* input) = 0;
   virtual ~RunnerCallbacks() = default;
 };
 
@@ -173,7 +180,7 @@ std::unique_ptr<RunnerCallbacks> CreateLegacyRunnerCallbacks(
 //
 // As an *experiment* we want to allow user code to call RunnerMain().
 // This is not a guaranteed public interface (yet) and may disappear w/o notice.
-int RunnerMain(int argc, char **argv, RunnerCallbacks &callbacks);
+int RunnerMain(int argc, char** argv, RunnerCallbacks& callbacks);
 
 }  // namespace fuzztest::internal
 
diff --git a/centipede/testing/async_failing_target.cc b/centipede/testing/async_failing_target.cc
index a3fc7dd69..6f418d77f 100644
--- a/centipede/testing/async_failing_target.cc
+++ b/centipede/testing/async_failing_target.cc
@@ -26,24 +26,34 @@ namespace {
 class AsyncFailingTargetRunnerCallbacks
     : public fuzztest::internal::RunnerCallbacks {
  public:
-  bool Execute(fuzztest::internal::ByteSpan input) override {
+  bool Execute(void* input) override {
     to_fail_in_mutation = true;
     return true;
   }
 
-  bool Mutate(absl::Span<const fuzztest::internal::MutationInputRef> inputs,
-              size_t num_mutants,
-              std::function<void(fuzztest::internal::MutantRef)>
-                  new_mutant_callback) override {
+  void* Mutate(void* inputs,
+               const fuzztest::internal::ExecutionMetadata& metadata) override {
     if (to_fail_in_mutation) {
       fprintf(stderr, "Fail in mutation\n");
       std::abort();
     }
-    return true;
+    return nullptr;
   }
 
   bool HasCustomMutator() const override { return true; }
 
+  void* DeserializeInput(fuzztest::internal::ByteSpan input) override {
+    return nullptr;
+  }
+
+  void SerializeInput(
+      void* input,
+      std::function<void(fuzztest::internal::ByteSpan)> bytes_sink) override {
+    bytes_sink({0});
+  }
+
+  void FreeInput(void* input) override {}
+
  private:
   bool to_fail_in_mutation = false;
 };
diff --git a/centipede/testing/external_target.cc b/centipede/testing/external_target.cc
index b77ea6483..c2162ad45 100644
--- a/centipede/testing/external_target.cc
+++ b/centipede/testing/external_target.cc
@@ -52,7 +52,9 @@ void sendall(int sock, const uint8_t* data, size_t size) {
 class ExternalTargetRunnerCallbacks
     : public fuzztest::internal::RunnerCallbacks {
  public:
-  bool Execute(fuzztest::internal::ByteSpan input) override {
+  bool Execute(void* input_raw) override {
+    const auto* input =
+        reinterpret_cast<const fuzztest::internal::ByteArray*>(input_raw);
     const char* port_env = getenv("TARGET_PORT");
     int port = 0;
     CHECK(port_env && absl::SimpleAtoi(port_env, &port))
@@ -72,10 +74,10 @@ class ExternalTargetRunnerCallbacks
     const int enable_nodelay = 1;
     setsockopt(conn_sock, SOL_TCP, TCP_NODELAY, &enable_nodelay,
                sizeof(enable_nodelay));
-    const uint64_t input_size = input.size();
+    const uint64_t input_size = input->size();
     sendall(conn_sock, reinterpret_cast<const uint8_t*>(&input_size),
             sizeof(input_size));
-    sendall(conn_sock, input.data(), input_size);
+    sendall(conn_sock, input->data(), input_size);
     int match_result = 0;
     recvall(conn_sock, reinterpret_cast<uint8_t*>(&match_result),
             sizeof(match_result));
@@ -96,6 +98,23 @@ class ExternalTargetRunnerCallbacks
   }
 
   bool HasCustomMutator() const override { return false; }
+
+  void* DeserializeInput(fuzztest::internal::ByteSpan input) override {
+    return reinterpret_cast<void*>(
+        new fuzztest::internal::ByteArray{input.begin(), input.end()});
+  }
+
+  void SerializeInput(
+      void* input,
+      std::function<void(fuzztest::internal::ByteSpan)> bytes_sink) override {
+    const auto* ba =
+        reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+    bytes_sink(*ba);
+  }
+
+  void FreeInput(void* input) override {
+    delete reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+  }
 };
 
 }  // namespace
diff --git a/centipede/testing/fuzz_target_with_config.cc b/centipede/testing/fuzz_target_with_config.cc
index 77788c065..524f002f2 100644
--- a/centipede/testing/fuzz_target_with_config.cc
+++ b/centipede/testing/fuzz_target_with_config.cc
@@ -32,12 +32,24 @@ class FakeSerializedConfigRunnerCallbacks
  public:
   // Trivial implementations for the execution and mutation logic, even though
   // they should not be used in the tests that use this test binary.
-  bool Execute(ByteSpan input) override { return true; }
+  bool Execute(void* input) override { return true; }
   bool HasCustomMutator() const override { return false; }
 
   std::string GetSerializedTargetConfig() override {
     return "fake serialized config";
   }
+
+  void* DeserializeInput(fuzztest::internal::ByteSpan input) override {
+    return nullptr;
+  }
+
+  void SerializeInput(
+      void* input,
+      std::function<void(fuzztest::internal::ByteSpan)> bytes_sink) override {
+    bytes_sink({0});
+  }
+
+  void FreeInput(void* input) override {}
 };
 
 int main(int argc, char** absl_nonnull argv) {
diff --git a/centipede/testing/fuzz_target_with_custom_mutator.cc b/centipede/testing/fuzz_target_with_custom_mutator.cc
index e6c28af8e..c6142f117 100644
--- a/centipede/testing/fuzz_target_with_custom_mutator.cc
+++ b/centipede/testing/fuzz_target_with_custom_mutator.cc
@@ -33,18 +33,32 @@ using fuzztest::internal::MutantRef;
 class CustomMutatorRunnerCallbacks
     : public fuzztest::internal::RunnerCallbacks {
  public:
-  bool Execute(ByteSpan input) override { return true; }
+  bool Execute(void* input) override { return true; }
 
   bool HasCustomMutator() const override { return true; }
 
-  bool Mutate(absl::Span<const fuzztest::internal::MutationInputRef> inputs,
-              size_t num_mutants,
-              std::function<void(MutantRef)> new_mutant_callback) override {
-    for (size_t i = 0; i < inputs.size() && i < num_mutants; ++i) {
-      // Just return the original input as a mutant.
-      new_mutant_callback(MutantRef{inputs[i].data, i});
-    }
-    return true;
+  void* Mutate(void* input,
+               const fuzztest::internal::ExecutionMetadata& metadata) override {
+    const auto* ba =
+        reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+    return reinterpret_cast<void*>(new fuzztest::internal::ByteArray{*ba});
+  }
+
+  void* DeserializeInput(fuzztest::internal::ByteSpan input) override {
+    return reinterpret_cast<void*>(
+        new fuzztest::internal::ByteArray{input.begin(), input.end()});
+  }
+
+  void SerializeInput(
+      void* input,
+      std::function<void(fuzztest::internal::ByteSpan)> bytes_sink) override {
+    const auto* ba =
+        reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+    bytes_sink(*ba);
+  }
+
+  void FreeInput(void* input) override {
+    delete reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
   }
 };
 
diff --git a/centipede/testing/seeded_fuzz_target.cc b/centipede/testing/seeded_fuzz_target.cc
index 0530109dd..41f69a1d2 100644
--- a/centipede/testing/seeded_fuzz_target.cc
+++ b/centipede/testing/seeded_fuzz_target.cc
@@ -24,18 +24,36 @@ using fuzztest::internal::ByteSpan;
 
 class SeededRunnerCallbacks : public fuzztest::internal::RunnerCallbacks {
  public:
-  bool Execute(ByteSpan input) override {
+  bool Execute(void* input) override {
     // Should not be called in the test, but return true anyway.
     return true;
   }
 
-  void GetSeeds(std::function<void(ByteSpan)> seed_callback) override {
+  void GetPresetSeedInputs(std::function<void(void*)> seed_callback) override {
     constexpr size_t kNumAvailSeeds = 10;
     for (size_t i = 0; i < kNumAvailSeeds; ++i)
-      seed_callback({static_cast<uint8_t>(i)});
+      seed_callback(reinterpret_cast<void*>(
+          new fuzztest::internal::ByteArray{static_cast<uint8_t>(i)}));
   }
 
   bool HasCustomMutator() const override { return false; }
+
+  void* DeserializeInput(fuzztest::internal::ByteSpan input) override {
+    return reinterpret_cast<void*>(
+        new fuzztest::internal::ByteArray{input.begin(), input.end()});
+  }
+
+  void SerializeInput(
+      void* input,
+      std::function<void(fuzztest::internal::ByteSpan)> bytes_sink) override {
+    const auto* ba =
+        reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+    bytes_sink(*ba);
+  }
+
+  void FreeInput(void* input) override {
+    delete reinterpret_cast<const fuzztest::internal::ByteArray*>(input);
+  }
 };
 
 int main(int argc, char** absl_nonnull argv) {
diff --git a/centipede/testing/test_binary_for_engine_testing.cc b/centipede/testing/test_binary_for_engine_testing.cc
index 75c1a42a5..071f2232c 100644
--- a/centipede/testing/test_binary_for_engine_testing.cc
+++ b/centipede/testing/test_binary_for_engine_testing.cc
@@ -202,10 +202,10 @@ int main(int argc, char** argv) {
     return worker_status == kFuzzTestWorkerSuccess ? EXIT_SUCCESS
                                                    : EXIT_FAILURE;
   }
-  return ControllerRun(&manager, {absl::StrCat("--binary=", argv[0]),
-                                  "--test_name=some_test",
-                                  "--populate_binary_info=0", "--fork_server=0",
-                                  "--persistent_mode=0", "--exit_on_crash"}) ==
+  return ControllerRun(&manager,
+                       {absl::StrCat("--binary=", argv[0]),
+                        "--test_name=some_test", "--populate_binary_info=0",
+                        "--fork_server=0", "--exit_on_crash"}) ==
                  kFuzzTestControllerSuccess
              ? EXIT_SUCCESS
              : EXIT_FAILURE;
diff --git a/fuzztest/internal/BUILD b/fuzztest/internal/BUILD
index 2bf470ff8..3bea5e9e6 100644
--- a/fuzztest/internal/BUILD
+++ b/fuzztest/internal/BUILD
@@ -81,6 +81,7 @@ cc_library(
         "@abseil-cpp//absl/algorithm:container",
         "@abseil-cpp//absl/base:no_destructor",
         "@abseil-cpp//absl/cleanup",
+        "@abseil-cpp//absl/container:node_hash_map",
         "@abseil-cpp//absl/functional:any_invocable",
         "@abseil-cpp//absl/memory",
         "@abseil-cpp//absl/random",
@@ -97,6 +98,7 @@ cc_library(
         "@com_google_fuzztest//centipede:centipede_interface",
         "@com_google_fuzztest//centipede:centipede_runner_no_main",
         "@com_google_fuzztest//centipede:environment",
+        "@com_google_fuzztest//centipede:execution_metadata",
         "@com_google_fuzztest//centipede:fuzztest_mutator",
         "@com_google_fuzztest//centipede:mutation_data",
         "@com_google_fuzztest//centipede:runner_result",
diff --git a/fuzztest/internal/centipede_adaptor.cc b/fuzztest/internal/centipede_adaptor.cc
index 7c23407e6..8d3f9ffb1 100644
--- a/fuzztest/internal/centipede_adaptor.cc
+++ b/fuzztest/internal/centipede_adaptor.cc
@@ -48,6 +48,7 @@
 #include "absl/algorithm/container.h"
 #include "absl/base/no_destructor.h"
 #include "absl/cleanup/cleanup.h"
+#include "absl/container/node_hash_map.h"
 #include "absl/functional/any_invocable.h"
 #include "absl/memory/memory.h"
 #include "absl/random/distributions.h"
@@ -68,6 +69,7 @@
 #include "./centipede/centipede_default_callbacks.h"
 #include "./centipede/centipede_interface.h"
 #include "./centipede/environment.h"
+#include "./centipede/execution_metadata.h"
 #include "./centipede/fuzztest_mutator.h"
 #include "./centipede/mutation_data.h"
 #include "./centipede/runner_interface.h"
@@ -493,7 +495,10 @@ class CentipedeAdaptorRunnerCallbacks
         configuration_(*configuration),
         prng_(GetRandomSeed()) {}
 
-  bool Execute(fuzztest::internal::ByteSpan input) override {
+  bool Execute(void* input) override {
+    const auto* input_object =
+        reinterpret_cast<const FuzzTestFuzzerImpl::Input*>(input);
+
     // Disable tracing until running the property function in
     // `CentipedeFxitureDriver::RunFuzzTestIteration()`
     const int old_traced = CentipedeSetCurrentThreadTraced(/*traced=*/0);
@@ -517,84 +522,78 @@ class CentipedeAdaptorRunnerCallbacks
     }
     // We should avoid doing anything other than executing the input here so
     // that we don't affect the execution time.
-    auto parsed_input =
-        fuzzer_impl_.TryParse({(char*)input.data(), input.size()});
-    if (parsed_input.ok()) {
-      fuzzer_impl_.RunOneInput({*std::move(parsed_input)});
-      if (runtime_.external_failure_detected()) {
-        // This would take effect only if no previous description is set.
-        CentipedeSetFailureDescription(
-            "INPUT FAILURE: external failure detected.");
-      }
-      return true;
+    fuzzer_impl_.RunOneInput(*input_object);
+    if (runtime_.external_failure_detected()) {
+      // This would take effect only if no previous description is set.
+      CentipedeSetFailureDescription(
+          "INPUT FAILURE: external failure detected.");
     }
-    return false;
+    return true;
   }
 
-  void GetSeeds(std::function<void(fuzztest::internal::ByteSpan)> seed_callback)
-      override {
+  void GetPresetSeedInputs(std::function<void(void*)> seed_callback) override {
     std::vector<GenericDomainCorpusType> seeds =
         fuzzer_impl_.fixture_driver_->GetSeeds();
-    constexpr int kInitialValuesInSeeds = 32;
-    for (int i = 0; i < kInitialValuesInSeeds; ++i) {
-      seeds.push_back(fuzzer_impl_.params_domain_.Init(prng_));
-    }
-    absl::c_shuffle(seeds, prng_);
     for (const auto& seed : seeds) {
-      const auto seed_serialized =
-          SerializeIRObject(fuzzer_impl_.params_domain_.SerializeCorpus(seed));
-      seed_callback(fuzztest::internal::AsByteSpan(seed_serialized));
+      auto* input = new FuzzTestFuzzerImpl::Input{std::move(seed)};
+      seed_callback(reinterpret_cast<void*>(input));
     }
   }
 
+  void GetRandomSeedInput(std::function<void(void*)> seed_callback) override {
+    auto seed = fuzzer_impl_.params_domain_.Init(prng_);
+    auto* input = new FuzzTestFuzzerImpl::Input{std::move(seed)};
+    seed_callback(reinterpret_cast<void*>(input));
+  }
+
   std::string GetSerializedTargetConfig() override {
     return configuration_.Serialize();
   }
 
   bool HasCustomMutator() const override { return true; }
 
-  bool Mutate(absl::Span<const MutationInputRef> inputs, size_t num_mutants,
-              std::function<void(MutantRef)> new_mutant_callback) override {
-    if (inputs.empty()) return false;
-    cmp_tables.resize(inputs.size());
-    absl::Cleanup cmp_tables_cleaner = [this]() { cmp_tables.clear(); };
-    for (size_t i = 0; i < num_mutants; ++i) {
-      const auto choice = absl::Uniform<double>(prng_, 0, 1);
-      size_t origin_index = Mutant::kOriginNone;
-      std::string mutant_data;
-      constexpr double kDomainInitRatio = 0.0001;
-      if (choice < kDomainInitRatio) {
-        mutant_data =
-            SerializeIRObject(fuzzer_impl_.params_domain_.SerializeCorpus(
-                fuzzer_impl_.params_domain_.Init(prng_)));
-      } else {
-        origin_index = absl::Uniform<size_t>(prng_, 0, inputs.size());
-        const auto& origin = inputs[origin_index].data;
-        auto parsed_origin =
-            fuzzer_impl_.TryParse({(const char*)origin.data(), origin.size()});
-        if (!parsed_origin.ok()) {
-          parsed_origin = fuzzer_impl_.params_domain_.Init(prng_);
-        }
-        auto mutant = FuzzTestFuzzerImpl::Input{*std::move(parsed_origin)};
-        if (runtime_.run_mode() == RunMode::kFuzz &&
-            !cmp_tables[origin_index].has_value() &&
-            inputs[origin_index].metadata != nullptr) {
-          cmp_tables[origin_index].emplace(/*compact=*/true);
-          PopulateCmpEntries(*inputs[origin_index].metadata,
-                             *cmp_tables[origin_index]);
-        }
-        fuzzer_impl_.MutateValue(
-            mutant, prng_,
-            {cmp_tables[origin_index].has_value() ? &*cmp_tables[origin_index]
-                                                  : nullptr});
-        mutant_data = SerializeIRObject(
-            fuzzer_impl_.params_domain_.SerializeCorpus(mutant.args));
-      }
-      new_mutant_callback(
-          MutantRef{{(unsigned char*)mutant_data.data(), mutant_data.size()},
-                    origin_index});
+  void SerializeInput(void* input,
+                      std::function<void(ByteSpan)> bytes_sink) override {
+    auto* input_object = reinterpret_cast<FuzzTestFuzzerImpl::Input*>(input);
+    std::string serialized_input = SerializeIRObject(
+        fuzzer_impl_.params_domain_.SerializeCorpus(input_object->args));
+    bytes_sink({reinterpret_cast<const uint8_t*>(serialized_input.data()),
+                serialized_input.size()});
+  }
+
+  void* DeserializeInput(ByteSpan input_bytes) override {
+    auto parse_result = fuzzer_impl_.TryParse(
+        {(const char*)input_bytes.data(), input_bytes.size()});
+    if (parse_result.ok()) {
+      return reinterpret_cast<void*>(
+          new FuzzTestFuzzerImpl::Input{*std::move(parse_result)});
     }
-    return true;
+    return reinterpret_cast<void*>(
+        new FuzzTestFuzzerImpl::Input{fuzzer_impl_.params_domain_.Init(prng_)});
+  }
+
+  void FreeInput(void* input) override {
+    auto it = cmp_tables_.find(input);
+    if (it != cmp_tables_.end()) {
+      delete it->second;
+      cmp_tables_.erase(it);
+    }
+    delete reinterpret_cast<FuzzTestFuzzerImpl::Input*>(input);
+  }
+
+  void* Mutate(void* origin,
+               const ExecutionMetadata& origin_metadata) override {
+    const auto* origin_object =
+        reinterpret_cast<const FuzzTestFuzzerImpl::Input*>(origin);
+    auto* mutant = new FuzzTestFuzzerImpl::Input{*origin_object};
+    if (cmp_tables_.find(origin) == cmp_tables_.end()) {
+      auto* cmp_table =
+          new fuzztest::internal::TablesOfRecentCompares{/*compact=*/true};
+      PopulateCmpEntries(origin_metadata, *cmp_table);
+      cmp_tables_[origin] = cmp_table;
+    }
+    fuzzer_impl_.MutateValue(*mutant, prng_, {cmp_tables_[origin]});
+    return mutant;
   }
 
   ~CentipedeAdaptorRunnerCallbacks() override { runtime_.UnsetCurrentArgs(); }
@@ -604,8 +603,8 @@ class CentipedeAdaptorRunnerCallbacks
   FuzzTestFuzzerImpl& fuzzer_impl_;
   const Configuration& configuration_;
   absl::BitGen prng_;
-  std::vector<std::optional<fuzztest::internal::TablesOfRecentCompares>>
-      cmp_tables;
+  absl::node_hash_map<void*, fuzztest::internal::TablesOfRecentCompares*>
+      cmp_tables_;
 };
 
 namespace {