Rename compute to pPilot

.github/actions/setup-build-env/action.yml

@@ -42,8 +42,11 @@ runs:
       shell: bash
       run: |
         sudo apt-get update -qq
-        sudo apt-get install -y -qq --no-install-recommends protobuf-compiler
+        sudo apt-get install -y -qq --no-install-recommends \
+          protobuf-compiler \
+          libprotobuf-dev
         protoc --version
+        test -f /usr/include/google/protobuf/empty.proto
 
     - name: Install protoc
       if: inputs.install-protoc == 'true' && runner.os == 'macOS'

Cargo.toml

@@ -7,7 +7,7 @@ members = [
     "crates/persisting-capture",
     "crates/persisting-cli",
     "crates/persisting-dlcapt",
-    "crates/persisting-compute",
+    "crates/persisting-ppilot",
     "crates/persisting-pvisor",
 ]
 

README.md

@@ -112,7 +112,7 @@ arr = kv["s1", 0, 2, 0:512].tensor()
 |------------|--------|-------------|
 | **Agent Trajectory Capture** | ✅ Stable | Proxy + record LLM traffic as Lance + Markdown |
 | **Streaming Queue** | ✅ Stable | Lance-backed append/consume, KV API, samplers |
-| **Compute Orchestration** | ✅ Stable | `plan()` + `execute()`, local/torchrun |
+| **pPilot Orchestration** | ✅ Stable | `plan()` + `execute()`, local/torchrun |
 | **Agent Search** | ✅ Stable | Document indexing, IVF-PQ, hybrid search |
 | **Tensor Memory (TTAS)** | 🧪 Experimental | Multi-dim tensor subscript, tiered backends |
 | **Cross-node KV Cache** | 📋 Planned | Pulsing + RDMA data plane |
@@ -159,7 +159,7 @@ pip install persisting[lance]        # Full
 pip install persisting               # Minimal
 ```
 
-For the CLI tools (`persisting traj`, `persisting compute`, `persisting search`):
+For the CLI tools (`persisting traj`, `persisting ppilot`, `persisting search`):
 
 ```bash
 git clone https://github.com/DeepLink-org/Persisting.git

crates/persisting-capture/tests/fixtures/tlv/demo-run-001.md

@@ -17,5 +17,5 @@ block: |+
 
 <!-- persisting:block:note {"type":"markdown","length":89,"kind":"note","role":"note","seq":2,"session_id":"demo-run-001","source":"markdown","timestamp":"2026-01-01T00:00:00Z","turn":2,"v":1} -->
 
-本次会话由 `persisting traj capture` 代理记录；元数据在块注释 JSON 中。
+本次会话由 `persisting capture run` 代理记录；元数据在块注释 JSON 中。
 

crates/persisting-cli/Cargo.toml

@@ -25,7 +25,7 @@ persisting-capture = { path = "../persisting-capture" }
 persisting-dlcapt = { path = "../persisting-dlcapt", optional = true }
 persisting-engine = { path = "../persisting-engine" }
 persisting-proto = { path = "../persisting-proto" }
-persisting-compute = { path = "../persisting-compute", features = ["traj-sink"] }
+persisting-ppilot = { path = "../persisting-ppilot", features = ["traj-sink"] }
 ron = "0.8"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

crates/persisting-cli/src/main.rs

@@ -299,8 +299,9 @@ enum Command {
     /// Agent trajectory: capture, proxy, inspect, repair（短名 `traj`）
     #[command(visible_alias = "traj", long_about = TRAJ_LONG_ABOUT)]
     Trajectory(TrajectoryArgs),
-    /// Run a compute plan (default). Use `--check` to validate locally first.
-    Compute(persisting_compute::ComputeArgs),
+    /// pPilot durable Run orchestration. `compute` remains a compatibility alias.
+    #[command(name = "ppilot", visible_alias = "compute")]
+    PPilot(persisting_ppilot::PPilotArgs),
 }
 
 #[derive(Debug, Args)]
@@ -1035,12 +1036,12 @@ fn engine_lib_names() -> [&'static str; 3] {
 fn main() -> Result<()> {
     let cli = Cli::parse_from(normalize_cli_args(std::env::args().collect()));
     match &cli.command {
-        Command::Compute(args) => {
-            persisting_compute::cli::init_tracing_with_verbose(args.verbose);
+        Command::PPilot(args) => {
+            persisting_ppilot::cli::init_tracing_with_verbose(args.verbose);
             let args = args.clone();
             let code = tokio::runtime::Runtime::new()
                 .context("tokio runtime")?
-                .block_on(persisting_compute::run_compute(args))?;
+                .block_on(persisting_ppilot::run_ppilot(args))?;
             if code != std::process::ExitCode::SUCCESS {
                 std::process::exit(1);
             }

crates/persisting-core/src/tiered_loop.rs

@@ -6,8 +6,15 @@ use pyo3::prelude::*;
 use std::sync::{mpsc, Mutex};
 use std::thread;
 
+/// 分区坐标值。保留整数与字符串的类型信息，与 Python BlockId 对应。
+#[allow(dead_code)]
+enum CoordValue {
+    Int(i64),
+    Str(String),
+}
+
 /// 单条预取任务：(partition_key, block_id)，与 Python BlockId 对应。
-pub type BlockRef = (Vec<i64>, i64);
+type BlockRef = (Vec<CoordValue>, i64);
 
 /// Rust 侧主事件循环：单线程消费预取队列，填页逻辑（fill_blocks）全在 Rust，不持 GIL。
 #[pyclass]
@@ -93,14 +100,10 @@ fn parse_single_block(item: &Bound<'_, PyAny>) -> PyResult<BlockRef> {
     let mut key = Vec::with_capacity(len);
     for i in 0..len {
         let v = pk_seq.get_item(i)?;
-        let v: i64 = if let Ok(n) = v.extract::<i64>() {
-            n
+        let v = if let Ok(n) = v.extract::<i64>() {
+            CoordValue::Int(n)
         } else if let Ok(s) = v.extract::<String>() {
-            // TODO: BlockRef 当前使用 Vec<i64>，无法忠实表示字符串 partition key。
-            // 改为支持 CoordValue（Int/Str/Bytes）后再启用字符串 key。
-            return Err(PyTypeError::new_err(format!(
-                "string partition keys not yet supported (got {s:?}); use integer keys"
-            )));
+            CoordValue::Str(s)
         } else {
             return Err(PyTypeError::new_err(
                 "partition_key elements must be int or str",

crates/persisting-core/src/uffd.rs

@@ -68,7 +68,7 @@ fn uffd_create() -> std::io::Result<i32> {
         features: 0,
         ioctl_bits: 0,
     };
-    if unsafe { ioctl(fd, UFFDIO_API as i32, &api) } < 0 {
+    if unsafe { ioctl(fd, UFFDIO_API as libc::Ioctl, &api) } < 0 {
         let e = std::io::Error::last_os_error();
         unsafe { libc::close(fd) };
         return Err(e);
@@ -83,7 +83,7 @@ fn uffd_register(fd: i32, start: u64, len: u64) -> std::io::Result<()> {
         mode: UFFDIO_REGISTER_MODE_MISSING,
         ioctls: 0,
     };
-    if unsafe { ioctl(fd, UFFDIO_REGISTER as i32, &reg) } < 0 {
+    if unsafe { ioctl(fd, UFFDIO_REGISTER as libc::Ioctl, &reg) } < 0 {
         return Err(std::io::Error::last_os_error());
     }
     Ok(())
@@ -98,7 +98,7 @@ fn uffd_copy(fd: i32, dst: u64, src: *const u8, len: usize) -> std::io::Result<(
         mode: 0,
         copy: 0,
     };
-    if unsafe { ioctl(fd, UFFDIO_COPY as i32, &mut copy) } < 0 {
+    if unsafe { ioctl(fd, UFFDIO_COPY as libc::Ioctl, &mut copy) } < 0 {
         return Err(std::io::Error::last_os_error());
     }
     Ok(())

crates/persisting-engine/src/abi.rs

@@ -106,11 +106,11 @@ mod tests {
         RequestBody, RpcRequest, RpcResponse, SearchAddRequest, PROTOCOL_VERSION,
     };
 
-    fn sample_request_ron() -> String {
+    fn sample_request_ron(dataset: &str) -> String {
         let req = RpcRequest {
             version: PROTOCOL_VERSION,
             body: RequestBody::SearchAdd(SearchAddRequest {
-                dataset: "ds".into(),
+                dataset: dataset.into(),
                 id: None,
                 text: "hi".into(),
                 metadata: None,
@@ -126,7 +126,9 @@ mod tests {
 
     #[test]
     fn submit_poll_take_smoke() {
-        let ron_in = sample_request_ron();
+        let temp = tempfile::tempdir().unwrap();
+        let dataset = temp.path().join("search.lance");
+        let ron_in = sample_request_ron(&dataset.to_string_lossy());
         let mut handle: u64 = 0;
         let st =
             unsafe { persisting_engine_submit(ron_in.as_ptr(), ron_in.len() as u64, &mut handle) };
@@ -162,7 +164,9 @@ mod tests {
 
     #[test]
     fn take_out_too_small_then_retry() {
-        let ron_in = sample_request_ron();
+        let temp = tempfile::tempdir().unwrap();
+        let dataset = temp.path().join("search.lance");
+        let ron_in = sample_request_ron(&dataset.to_string_lossy());
         let mut handle: u64 = 0;
         unsafe { persisting_engine_submit(ron_in.as_ptr(), ron_in.len() as u64, &mut handle) };
         loop {

crates/persisting-engine/src/ron_api.rs

@@ -41,10 +41,16 @@ mod tests {
 
     #[test]
     fn invoke_ron_search_add_roundtrip_envelope() {
+        let temp = tempfile::tempdir().unwrap();
+        let dataset = temp
+            .path()
+            .join("search.lance")
+            .to_string_lossy()
+            .into_owned();
         let req = RpcRequest {
             version: PROTOCOL_VERSION,
             body: RequestBody::SearchAdd(SearchAddRequest {
-                dataset: "ds".into(),
+                dataset: dataset.clone(),
                 id: Some("i1".into()),
                 text: "hello".into(),
                 metadata: None,
@@ -57,7 +63,7 @@ mod tests {
         assert_eq!(resp.version, PROTOCOL_VERSION);
         match resp.body {
             ResponseBody::SearchAdd(r) => {
-                assert_eq!(r.dataset, "ds");
+                assert_eq!(r.dataset, dataset);
                 assert_eq!(r.id, "i1");
             }
             _ => panic!("expected SearchAdd response"),

crates/persisting-compute/Cargo.toml → crates/persisting-ppilot/Cargo.toml

@@ -1,10 +1,10 @@
 [package]
-name = "persisting-compute"
+name = "persisting-ppilot"
 version.workspace = true
 edition.workspace = true
 authors.workspace = true
 license.workspace = true
-description = "Thin compute control plane: plan script → streamed tasks → Pulsing workers (via `persisting compute`)"
+description = "pPilot durable Run orchestrator: plan, schedule, resume, reconcile, and collect independent Runs"
 
 [dependencies]
 anyhow = "1"
@@ -26,7 +26,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 default = []
-# Append compute results to Lance via TrajectoryAppend (Tee with JsonlFileSink).
+# Append pPilot results to Lance via TrajectoryAppend (Tee with JsonlFileSink).
 traj-sink = ["dep:chrono", "dep:persisting-capture", "dep:persisting-engine", "dep:persisting-proto"]
 
 [dev-dependencies]

crates/persisting-compute/src/blocks.rs → crates/persisting-ppilot/src/blocks.rs

@@ -19,7 +19,7 @@
 //! | Async sink writer | [`SinkSubmitter`], [`spawn_sink_writer`] | [`crate::sink_writer`] |
 //! | Checkpoint | [`CheckpointLedger`], [`CheckpointTracker`] | [`crate::checkpoint`] |
 //! | Plan emit | [`stream_plan_tasks`] | [`crate::plan`] |
-//! | Execute host | [`Executor`], [`ExecutorRouter`] | [`crate::executor`] |
+//! | Execute host | Executor router | `executor` |
 //! | Worker seam | [`WorkerActor`], supervised spawn | [`crate::worker`] |
 //! | Job cancel | [`JobControlActor`], DeathWatch | [`crate::job_control`] |
 //! | Pulsing helpers | resolve / ask_timeout / spawn_supervised | [`crate::pulsing_ext`] |
@@ -43,8 +43,6 @@
 //! [`CheckpointLedger`]: crate::checkpoint::CheckpointLedger
 //! [`CheckpointTracker`]: crate::checkpoint::CheckpointTracker
 //! [`stream_plan_tasks`]: crate::plan::stream_plan_tasks
-//! [`Executor`]: crate::executor::Executor
-//! [`ExecutorRouter`]: crate::executor::ExecutorRouter
 //! [`WorkerActor`]: crate::worker::WorkerActor
 //! [`WorkerCommand`]: crate::worker::WorkerCommand
 //! [`JobControlActor`]: crate::job_control::JobControlActor

crates/persisting-compute/src/check.rs → crates/persisting-ppilot/src/check.rs

@@ -1,7 +1,7 @@
 //! Local validation: prove env + plan + execute before scale-out.
 //!
 //! ```text
-//! persisting compute plan.py --check --python python3
+//! persisting ppilot plan.py --check --python python3
 //! ```
 
 use crate::plan::stream_plan_tasks;
@@ -150,7 +150,7 @@ pub async fn run_check(opts: CheckOptions) -> Result<CheckReport> {
             tasks.len() - to_run,
             tasks.len()
         );
-        crate::skip::SkipSet::from_iter(tasks[to_run..].iter().map(|t| t.id.clone()))
+        tasks[to_run..].iter().map(|t| t.id.clone()).collect()
     } else {
         crate::skip::SkipSet::new()
     };
@@ -306,10 +306,8 @@ print("ok", flush=True)
 
 /// Built-in smoke: no user plan. Proves `plan()` + `execute(item)` end-to-end.
 pub async fn run_self_test(python: PathBuf, workers: usize, verbose: bool) -> Result<CheckReport> {
-    let dir = std::env::temp_dir().join(format!(
-        "persisting-compute-selftest-{}",
-        std::process::id()
-    ));
+    let dir =
+        std::env::temp_dir().join(format!("persisting-ppilot-selftest-{}", std::process::id()));
     std::fs::create_dir_all(&dir).with_context(|| format!("mkdir {}", dir.display()))?;
 
     let plan_py = dir.join("plan_smoke.py");