Add Pi RPC title-boundary reviewer backend

tools/rust_dmhy_template_apply/README.md

@@ -53,9 +53,9 @@ cargo run --release --manifest-path tools\rust_dmhy_template_apply\Cargo.toml --
   --audit-max-count 50
 ```
 
-Review ambiguous title-boundary decisions with a local LM Studio model. This is
-an optional final-pass reviewer: it writes a decision cache JSONL and does not
-modify recipes or training data directly.
+Review ambiguous title-boundary decisions with a local LM Studio model or with
+Pi RPC mode. This is an optional final-pass reviewer: it writes a decision cache
+JSONL and does not modify recipes or training data directly.
 
 ```powershell
 cargo run --release --manifest-path tools\rust_dmhy_template_apply\Cargo.toml -- `
@@ -69,6 +69,17 @@ cargo run --release --manifest-path tools\rust_dmhy_template_apply\Cargo.toml --
   --limit 50
 ```
 
+```powershell
+cargo run --release --manifest-path tools\rust_dmhy_template_apply\Cargo.toml -- `
+  --review-title-boundaries-pi-rpc `
+  --recipes reports\dmhy_template_recipes.full_top5000.seed.jsonl `
+  --title-boundary-decisions-output reports\dmhy_title_boundary_pi_rpc_decisions.jsonl `
+  --pi-rpc-model openai/gpt-5.5 `
+  --title-boundary-min-similarity 0.25 `
+  --title-boundary-max-similarity 0.85 `
+  --limit 50
+```
+
 Optional controls:
 
 ```powershell

tools/rust_dmhy_template_apply/src/main.rs

@@ -10,6 +10,7 @@ use std::collections::{HashMap, HashSet};
 use std::fs::{self, File};
 use std::io::{BufRead, BufReader, BufWriter, Read, Write};
 use std::path::PathBuf;
+use std::process::{Child, ChildStdin, Command, Stdio};
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 #[derive(Parser, Debug)]
@@ -25,6 +26,8 @@ struct Args {
     rich_annotations: bool,
     #[arg(long)]
     review_title_boundaries_lmstudio: bool,
+    #[arg(long)]
+    review_title_boundaries_pi_rpc: bool,
     #[arg(long, default_value = "datasets/AnimeName/dmhy_list.jsonl")]
     input: PathBuf,
     #[arg(long, default_value = "reports/dmhy_template_recipes.seed.jsonl")]
@@ -76,6 +79,8 @@ struct Args {
     lmstudio_temperature: f64,
     #[arg(long, default_value_t = 120)]
     lmstudio_timeout_seconds: u64,
+    #[arg(long)]
+    pi_rpc_model: Option<String>,
     #[arg(long, default_value_t = 50)]
     audit_max_count: u64,
     #[arg(long)]
@@ -329,8 +334,8 @@ fn main() -> Result<()> {
     if args.rich_annotations {
         return run_rich_annotations(&args);
     }
-    if args.review_title_boundaries_lmstudio {
-        return run_review_title_boundaries_lmstudio(&args);
+    if args.review_title_boundaries_lmstudio || args.review_title_boundaries_pi_rpc {
+        return run_review_title_boundaries(&args);
     }
     if args.expand != "all" && args.expand != "sample" {
         bail!("--expand must be all or sample");
@@ -1171,7 +1176,7 @@ fn run_rich_annotations(args: &Args) -> Result<()> {
     Ok(())
 }
 
-fn run_review_title_boundaries_lmstudio(args: &Args) -> Result<()> {
+fn run_review_title_boundaries(args: &Args) -> Result<()> {
     if !(0.0..=1.0).contains(&args.title_boundary_min_similarity)
         || !(0.0..=1.0).contains(&args.title_boundary_max_similarity)
         || args.title_boundary_min_similarity > args.title_boundary_max_similarity
@@ -1188,10 +1193,18 @@ fn run_review_title_boundaries_lmstudio(args: &Args) -> Result<()> {
     let mut writer = BufWriter::new(File::create(&args.title_boundary_decisions_output)?);
     let mut reviewed = 0usize;
     let mut skipped = 0usize;
+    if args.review_title_boundaries_lmstudio && args.review_title_boundaries_pi_rpc {
+        bail!("choose only one title-boundary review backend: LM Studio or Pi RPC");
+    }
     let endpoint = format!(
         "{}/chat/completions",
         args.lmstudio_base_url.trim_end_matches('/')
     );
+    let mut pi_rpc = if args.review_title_boundaries_pi_rpc {
+        Some(PiRpcClient::start(args.pi_rpc_model.as_deref())?)
+    } else {
+        None
+    };
 
     for recipe in recipes {
         let template_id = recipe
@@ -1226,27 +1239,34 @@ fn run_review_title_boundaries_lmstudio(args: &Args) -> Result<()> {
                 }
             }
             let prompt = title_boundary_review_prompt(&recipe, &decision);
-            let response = call_lmstudio_title_boundary(
-                &endpoint,
-                &args.lmstudio_model,
-                args.lmstudio_temperature,
-                args.lmstudio_timeout_seconds,
-                &prompt,
-            )
-            .with_context(|| {
-                format!(
-                    "LM Studio title-boundary review failed for template {template_id} via {endpoint}"
+            let content = if let Some(client) = pi_rpc.as_mut() {
+                client.review(&prompt).with_context(|| {
+                    format!("Pi RPC title-boundary review failed for template {template_id}")
+                })?
+            } else {
+                let response = call_lmstudio_title_boundary(
+                    &endpoint,
+                    &args.lmstudio_model,
+                    args.lmstudio_temperature,
+                    args.lmstudio_timeout_seconds,
+                    &prompt,
                 )
-            })?;
-            let content = response
-                .get("choices")
-                .and_then(Value::as_array)
-                .and_then(|choices| choices.first())
-                .and_then(|choice| choice.get("message"))
-                .and_then(|message| message.get("content"))
-                .and_then(Value::as_str)
-                .unwrap_or("");
-            let parsed = parse_lmstudio_json_object(content).unwrap_or_else(|| {
+                .with_context(|| {
+                    format!(
+                        "LM Studio title-boundary review failed for template {template_id} via {endpoint}"
+                    )
+                })?;
+                response
+                    .get("choices")
+                    .and_then(Value::as_array)
+                    .and_then(|choices| choices.first())
+                    .and_then(|choice| choice.get("message"))
+                    .and_then(|message| message.get("content"))
+                    .and_then(Value::as_str)
+                    .unwrap_or("")
+                    .to_string()
+            };
+            let parsed = parse_lmstudio_json_object(&content).unwrap_or_else(|| {
                 json!({
                     "decision": "skip",
                     "kind": "parse_error",
@@ -1265,8 +1285,8 @@ fn run_review_title_boundaries_lmstudio(args: &Args) -> Result<()> {
                 "model_kind": parsed.get("kind").cloned().unwrap_or(Value::Null),
                 "confidence": parsed.get("confidence").cloned().unwrap_or(Value::Null),
                 "reason": parsed.get("reason").cloned().unwrap_or(Value::Null),
-                "source": "lmstudio",
-                "model": args.lmstudio_model,
+                "source": if args.review_title_boundaries_pi_rpc { "pi-rpc" } else { "lmstudio" },
+                "model": if args.review_title_boundaries_pi_rpc { args.pi_rpc_model.as_deref().unwrap_or("pi-default") } else { args.lmstudio_model.as_str() },
                 "reviewed_at": Utc::now().to_rfc3339(),
             });
             serde_json::to_writer(&mut writer, &output)?;
@@ -1280,8 +1300,9 @@ fn run_review_title_boundaries_lmstudio(args: &Args) -> Result<()> {
         "output": args.title_boundary_decisions_output.to_string_lossy(),
         "reviewed": reviewed,
         "skipped": skipped,
-        "model": args.lmstudio_model,
-        "base_url": args.lmstudio_base_url,
+        "model": if args.review_title_boundaries_pi_rpc { args.pi_rpc_model.as_deref().unwrap_or("pi-default") } else { args.lmstudio_model.as_str() },
+        "backend": if args.review_title_boundaries_pi_rpc { "pi-rpc" } else { "lmstudio" },
+        "base_url": if args.review_title_boundaries_pi_rpc { "pi --mode rpc" } else { args.lmstudio_base_url.as_str() },
         "similarity_range": [args.title_boundary_min_similarity, args.title_boundary_max_similarity],
         "generated_at": Utc::now().to_rfc3339(),
     });
@@ -1369,6 +1390,149 @@ fn title_boundary_review_prompt(recipe: &Value, decision: &Value) -> String {
     )
 }
 
+struct PiRpcClient {
+    child: Child,
+    stdin: Option<ChildStdin>,
+    stdout: BufReader<std::process::ChildStdout>,
+    next_id: usize,
+}
+
+impl PiRpcClient {
+    fn start(model: Option<&str>) -> Result<Self> {
+        let mut command = Command::new("pi");
+        command.args(["--mode", "rpc", "--no-session"]);
+        if let Some(model) = model {
+            command.args(["--model", model]);
+        }
+        let mut child = command
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .context("failed to start pi --mode rpc")?;
+        let stdin = child
+            .stdin
+            .take()
+            .ok_or_else(|| anyhow::anyhow!("failed to open Pi RPC stdin"))?;
+        let stdout = child
+            .stdout
+            .take()
+            .ok_or_else(|| anyhow::anyhow!("failed to open Pi RPC stdout"))?;
+        Ok(Self {
+            child,
+            stdin: Some(stdin),
+            stdout: BufReader::new(stdout),
+            next_id: 0,
+        })
+    }
+
+    fn review(&mut self, prompt: &str) -> Result<String> {
+        let prompt_id = self.next_request_id("prompt");
+        self.send(json!({"id": prompt_id, "type": "prompt", "message": prompt}))?;
+        let mut accepted = false;
+        loop {
+            let event = self.read_event()?;
+            if event.get("type").and_then(Value::as_str) == Some("response")
+                && event.get("id").and_then(Value::as_str) == Some(prompt_id.as_str())
+            {
+                if !event
+                    .get("success")
+                    .and_then(Value::as_bool)
+                    .unwrap_or(false)
+                {
+                    bail!("Pi RPC prompt rejected: {event}");
+                }
+                accepted = true;
+                continue;
+            }
+            if event.get("type").and_then(Value::as_str) == Some("agent_end") && accepted {
+                break;
+            }
+        }
+        let text = self.get_last_assistant_text()?;
+        self.new_session()?;
+        Ok(text)
+    }
+
+    fn get_last_assistant_text(&mut self) -> Result<String> {
+        let id = self.next_request_id("last");
+        self.send(json!({"id": id, "type": "get_last_assistant_text"}))?;
+        loop {
+            let event = self.read_event()?;
+            if event.get("type").and_then(Value::as_str) == Some("response")
+                && event.get("id").and_then(Value::as_str) == Some(id.as_str())
+            {
+                if !event
+                    .get("success")
+                    .and_then(Value::as_bool)
+                    .unwrap_or(false)
+                {
+                    bail!("Pi RPC get_last_assistant_text failed: {event}");
+                }
+                return Ok(event
+                    .get("data")
+                    .and_then(|data| data.get("text"))
+                    .and_then(Value::as_str)
+                    .unwrap_or("")
+                    .to_string());
+            }
+        }
+    }
+
+    fn new_session(&mut self) -> Result<()> {
+        let id = self.next_request_id("new");
+        self.send(json!({"id": id, "type": "new_session"}))?;
+        loop {
+            let event = self.read_event()?;
+            if event.get("type").and_then(Value::as_str) == Some("response")
+                && event.get("id").and_then(Value::as_str) == Some(id.as_str())
+            {
+                if !event
+                    .get("success")
+                    .and_then(Value::as_bool)
+                    .unwrap_or(false)
+                {
+                    bail!("Pi RPC new_session failed: {event}");
+                }
+                return Ok(());
+            }
+        }
+    }
+
+    fn next_request_id(&mut self, prefix: &str) -> String {
+        self.next_id += 1;
+        format!("{prefix}-{}", self.next_id)
+    }
+
+    fn send(&mut self, value: Value) -> Result<()> {
+        let stdin = self
+            .stdin
+            .as_mut()
+            .ok_or_else(|| anyhow::anyhow!("Pi RPC stdin is closed"))?;
+        serde_json::to_writer(&mut *stdin, &value)?;
+        stdin.write_all(b"\n")?;
+        stdin.flush()?;
+        Ok(())
+    }
+
+    fn read_event(&mut self) -> Result<Value> {
+        let mut line = String::new();
+        let bytes = self.stdout.read_line(&mut line)?;
+        if bytes == 0 {
+            bail!("Pi RPC process exited");
+        }
+        let line = line.trim_end_matches(['\r', '\n']);
+        serde_json::from_str(line).with_context(|| format!("invalid Pi RPC JSON line: {line}"))
+    }
+}
+
+impl Drop for PiRpcClient {
+    fn drop(&mut self) {
+        let _ = self.stdin.take();
+        let _ = self.child.wait();
+    }
+}
+
 fn call_lmstudio_title_boundary(
     endpoint: &str,
     model: &str,