Parallelize Pi RPC title-boundary reviews

AGENTS.md

@@ -116,6 +116,40 @@ cargo run --release --manifest-path tools\schema_v2_synthetic_augment\Cargo.toml
   --manifest data\schema_v2_synthetic_aug.manifest.json
 ```
 
+When auditing Rust DMHY title-boundary recipe decisions with an LLM, prefer the
+Rust reviewer cache flow. The LM reviewer is optional evidence only; it must not
+mutate recipes or training JSONL directly. Use Pi RPC for stronger model-backed
+reviews, and keep the default deterministic Rust heuristic as the source of
+truth until reviewed decisions are explicitly consumed by a separate Rust merge
+step:
+
+```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.title_boundary_trial.seed.jsonl `
+  --title-boundary-decisions-output reports\dmhy_title_boundary_pi_rpc_decisions.title_boundary_trial.jsonl `
+  --title-boundary-min-similarity 0.25 `
+  --title-boundary-max-similarity 0.85 `
+  --threads 4
+```
+
+For reviewer calibration, run stratified samples before full review. Use
+`--title-boundary-heuristic-contains` to inspect high-similar duplicate/alias
+keeps, structural keeps, and path-boundary keeps that are normally skipped by
+the ambiguous merge review:
+
+```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.title_boundary_trial.seed.jsonl `
+  --title-boundary-decisions-output reports\dmhy_title_boundary_pi_rpc_highsim_keep_sample.jsonl `
+  --title-boundary-heuristic-contains keep_similar_duplicate_or_alias `
+  --title-boundary-min-similarity 0.85 `
+  --title-boundary-max-similarity 1.0 `
+  --threads 4 `
+  --limit 20
+```
+
 Preferred synthetic follow-up training is a second stage from the best repaired
 hard-focus checkpoint, not a replacement for hard-focus. Keep this path
 Rust-cache-first: build one combined encoded cache from hard-focus JSONL plus

tools/rust_dmhy_template_apply/README.md

@@ -77,9 +77,36 @@ cargo run --release --manifest-path tools\rust_dmhy_template_apply\Cargo.toml --
   --pi-rpc-model openai/gpt-5.5 `
   --title-boundary-min-similarity 0.25 `
   --title-boundary-max-similarity 0.85 `
+  --threads 4 `
   --limit 50
 ```
 
+For stratified audits, select heuristic strata explicitly:
+
+```powershell
+# High-similarity duplicate/alias keep candidates.
+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_highsim_keep.jsonl `
+  --title-boundary-heuristic-contains keep_similar_duplicate_or_alias `
+  --title-boundary-min-similarity 0.85 `
+  --title-boundary-max-similarity 1.0 `
+  --threads 4 `
+  --limit 20
+
+# Structural keep candidates normally excluded from ambiguous merge review.
+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_structural_keep.jsonl `
+  --title-boundary-heuristic-contains keep_structural_boundary `
+  --title-boundary-min-similarity 0.0 `
+  --title-boundary-max-similarity 1.0 `
+  --threads 4 `
+  --limit 20
+```
+
 Optional controls:
 
 ```powershell

tools/rust_dmhy_template_apply/src/main.rs

@@ -81,6 +81,8 @@ struct Args {
     lmstudio_timeout_seconds: u64,
     #[arg(long)]
     pi_rpc_model: Option<String>,
+    #[arg(long)]
+    title_boundary_heuristic_contains: Option<String>,
     #[arg(long, default_value_t = 50)]
     audit_max_count: u64,
     #[arg(long)]
@@ -1176,6 +1178,23 @@ fn run_rich_annotations(args: &Args) -> Result<()> {
     Ok(())
 }
 
+#[derive(Clone, Copy)]
+enum TitleBoundaryReviewBackend {
+    LmStudio,
+    PiRpc,
+}
+
+#[derive(Clone)]
+struct TitleBoundaryReviewTask {
+    sequence: usize,
+    recipe: Value,
+    decision: Value,
+    similarity: f64,
+    heuristic_decision: String,
+    template_id: String,
+    prompt: String,
+}
+
 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)
@@ -1183,34 +1202,38 @@ fn run_review_title_boundaries(args: &Args) -> Result<()> {
     {
         bail!("title boundary similarity thresholds must satisfy 0 <= min <= max <= 1");
     }
-    let recipes = load_recipe_values(&args.recipes)?;
-    if recipes.is_empty() {
-        bail!("no recipe rows found in {}", args.recipes.display());
-    }
-    if let Some(parent) = args.title_boundary_decisions_output.parent() {
-        fs::create_dir_all(parent)?;
-    }
-    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 backend = if args.review_title_boundaries_pi_rpc {
+        TitleBoundaryReviewBackend::PiRpc
+    } else {
+        TitleBoundaryReviewBackend::LmStudio
+    };
+    let source = match backend {
+        TitleBoundaryReviewBackend::LmStudio => "lmstudio",
+        TitleBoundaryReviewBackend::PiRpc => "pi-rpc",
+    };
+    let model_name = match backend {
+        TitleBoundaryReviewBackend::LmStudio => args.lmstudio_model.as_str(),
+        TitleBoundaryReviewBackend::PiRpc => args.pi_rpc_model.as_deref().unwrap_or("pi-default"),
+    };
     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
-    };
+    let worker_count = args.threads.unwrap_or(1).max(1);
+    let recipes = load_recipe_values(&args.recipes)?;
+    if recipes.is_empty() {
+        bail!("no recipe rows found in {}", args.recipes.display());
+    }
+    if let Some(parent) = args.title_boundary_decisions_output.parent() {
+        fs::create_dir_all(parent)?;
+    }
 
-    for recipe in recipes {
-        let template_id = recipe
-            .get("template_id")
-            .and_then(Value::as_str)
-            .unwrap_or("unknown");
+    let mut skipped = 0usize;
+    let mut tasks = Vec::new();
+    'recipes: for recipe in recipes {
         let decisions = recipe
             .get("title_boundary_decisions")
             .and_then(Value::as_array)
@@ -1224,92 +1247,188 @@ fn run_review_title_boundaries(args: &Args) -> Result<()> {
             let heuristic_decision = decision
                 .get("decision")
                 .and_then(Value::as_str)
-                .unwrap_or("unknown");
-            if heuristic_decision.contains("path_boundary")
+                .unwrap_or("unknown")
+                .to_string();
+            if let Some(needle) = args.title_boundary_heuristic_contains.as_deref() {
+                if !heuristic_decision.contains(needle) {
+                    skipped += 1;
+                    continue;
+                }
+            } else if heuristic_decision.contains("path_boundary")
                 || heuristic_decision.contains("structural_boundary")
-                || similarity < args.title_boundary_min_similarity
+            {
+                skipped += 1;
+                continue;
+            }
+            if similarity < args.title_boundary_min_similarity
                 || similarity > args.title_boundary_max_similarity
             {
                 skipped += 1;
                 continue;
             }
             if let Some(limit) = args.limit {
-                if reviewed >= limit {
-                    break;
+                if tasks.len() >= limit {
+                    break 'recipes;
                 }
             }
-            let prompt = title_boundary_review_prompt(&recipe, &decision);
-            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,
-                )
-                .with_context(|| {
-                    format!(
-                        "LM Studio title-boundary review failed for template {template_id} via {endpoint}"
+            let template_id = recipe
+                .get("template_id")
+                .and_then(Value::as_str)
+                .unwrap_or("unknown")
+                .to_string();
+            tasks.push(TitleBoundaryReviewTask {
+                sequence: tasks.len(),
+                prompt: title_boundary_review_prompt(&recipe, &decision),
+                recipe: recipe.clone(),
+                decision,
+                similarity,
+                heuristic_decision,
+                template_id,
+            });
+        }
+    }
+
+    let mut outputs = if worker_count == 1 || tasks.len() <= 1 {
+        review_title_boundary_chunk(
+            &tasks,
+            backend,
+            &endpoint,
+            model_name,
+            args.pi_rpc_model.as_deref(),
+            args.lmstudio_temperature,
+            args.lmstudio_timeout_seconds,
+            source,
+        )?
+    } else {
+        let chunk_size = ((tasks.len() + worker_count - 1) / worker_count).max(1);
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(worker_count)
+            .build()
+            .context("failed to configure title-boundary review thread pool")?;
+        let chunk_results: Vec<Result<Vec<(usize, Value)>>> = pool.install(|| {
+            tasks
+                .par_chunks(chunk_size)
+                .map(|chunk| {
+                    review_title_boundary_chunk(
+                        chunk,
+                        backend,
+                        &endpoint,
+                        model_name,
+                        args.pi_rpc_model.as_deref(),
+                        args.lmstudio_temperature,
+                        args.lmstudio_timeout_seconds,
+                        source,
                     )
-                })?;
-                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",
-                    "confidence": 0.0,
-                    "reason": content,
                 })
-            });
-            let output = json!({
-                "template_id": recipe.get("template_id").cloned().unwrap_or(Value::Null),
-                "template": recipe.get("template").cloned().unwrap_or(Value::Null),
-                "left_role_index": decision.get("left_role_index").cloned().unwrap_or(Value::Null),
-                "right_role_index": decision.get("right_role_index").cloned().unwrap_or(Value::Null),
-                "similarity": similarity,
-                "heuristic_decision": heuristic_decision,
-                "model_decision": parsed.get("decision").cloned().unwrap_or(Value::Null),
-                "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": 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)?;
-            writer.write_all(b"\n")?;
-            reviewed += 1;
+                .collect()
+        });
+        let mut combined = Vec::new();
+        for result in chunk_results {
+            combined.extend(result?);
         }
+        combined
+    };
+    outputs.sort_by_key(|(sequence, _)| *sequence);
+
+    let mut writer = BufWriter::new(File::create(&args.title_boundary_decisions_output)?);
+    for (_, output) in &outputs {
+        serde_json::to_writer(&mut writer, output)?;
+        writer.write_all(b"\n")?;
     }
     writer.flush()?;
     let summary = json!({
         "recipes": args.recipes.to_string_lossy(),
         "output": args.title_boundary_decisions_output.to_string_lossy(),
-        "reviewed": reviewed,
+        "reviewed": outputs.len(),
         "skipped": skipped,
-        "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" },
+        "model": model_name,
+        "backend": source,
         "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],
+        "heuristic_contains": args.title_boundary_heuristic_contains.as_deref(),
+        "threads": worker_count,
         "generated_at": Utc::now().to_rfc3339(),
     });
     println!("{}", serde_json::to_string_pretty(&summary)?);
     Ok(())
 }
 
+fn review_title_boundary_chunk(
+    tasks: &[TitleBoundaryReviewTask],
+    backend: TitleBoundaryReviewBackend,
+    endpoint: &str,
+    model_name: &str,
+    pi_rpc_model: Option<&str>,
+    lmstudio_temperature: f64,
+    lmstudio_timeout_seconds: u64,
+    source: &str,
+) -> Result<Vec<(usize, Value)>> {
+    let mut pi_rpc = if matches!(backend, TitleBoundaryReviewBackend::PiRpc) {
+        Some(PiRpcClient::start(pi_rpc_model)?)
+    } else {
+        None
+    };
+    let mut outputs = Vec::with_capacity(tasks.len());
+    for task in tasks {
+        let content = if let Some(client) = pi_rpc.as_mut() {
+            client.review(&task.prompt).with_context(|| {
+                format!(
+                    "Pi RPC title-boundary review failed for template {}",
+                    task.template_id
+                )
+            })?
+        } else {
+            let response = call_lmstudio_title_boundary(
+                endpoint,
+                model_name,
+                lmstudio_temperature,
+                lmstudio_timeout_seconds,
+                &task.prompt,
+            )
+            .with_context(|| {
+                format!(
+                    "LM Studio title-boundary review failed for template {} via {endpoint}",
+                    task.template_id
+                )
+            })?;
+            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",
+                "confidence": 0.0,
+                "reason": content,
+            })
+        });
+        let output = json!({
+            "template_id": task.recipe.get("template_id").cloned().unwrap_or(Value::Null),
+            "template": task.recipe.get("template").cloned().unwrap_or(Value::Null),
+            "left_role_index": task.decision.get("left_role_index").cloned().unwrap_or(Value::Null),
+            "right_role_index": task.decision.get("right_role_index").cloned().unwrap_or(Value::Null),
+            "similarity": task.similarity,
+            "heuristic_decision": task.heuristic_decision,
+            "model_decision": parsed.get("decision").cloned().unwrap_or(Value::Null),
+            "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": source,
+            "model": model_name,
+            "reviewed_at": Utc::now().to_rfc3339(),
+        });
+        outputs.push((task.sequence, output));
+    }
+    Ok(outputs)
+}
+
 fn load_recipe_values(path: &PathBuf) -> Result<Vec<Value>> {
     let file = File::open(path).with_context(|| format!("failed to open {}", path.display()))?;
     let reader = BufReader::new(file);