Update .gitignore, remove uuid dependency, and refactor judge-related…

.gitignore

@@ -1,6 +1,6 @@
 docs/site/
 target/
-store/
+/store/
 *.pyc
 __pycache__/
 .pytest_cache/

crates/persisting-cli/src/judge_manual.rs

@@ -8,7 +8,7 @@ use crate::terminal_markdown::{format_turn_markdown, print_section};
 use persisting_capture::engine::TurnKind;
 use persisting_capture::engine::{rebuild_session_story, Story};
 use persisting_capture::record::CaptureRecord;
-use persisting_engine::trajectory::layers::MANUAL_RATIONALE_PREFIX;
+use persisting_engine::trajectory::MANUAL_RATIONALE_PREFIX;
 use persisting_proto::{JudgeSampleMode, JudgeScope, JudgeScoreInput};
 
 /// Pick up to `limit` sessions from a scan list (`limit == 0` → keep all).

crates/persisting-cli/src/main.rs

@@ -627,9 +627,9 @@ enum TrajectoryCommand {
     Extract(TrajectoryExtractArgs),
     /// Lance → TLV Markdown（有损物化，维护用）。
     Materialize(TrajectoryMaterializeArgs),
-    /// LLM-as-judge：读 canonical Lance，写 `{run}/layers/judge_*.lance` sidecar。
+    /// LLM-as-judge：读 canonical Lance，把分数写为 `events.lance` 上的原生列。
     Judge(TrajectoryJudgeArgs),
-    /// 汇总 judge sidecar 分数（按 session + rubric）。
+    /// 汇总 `events.lance` 上的 judge 列分数（按 session + rubric）。
     #[command(name = "judge-stats")]
     JudgeStats(TrajectoryJudgeStatsArgs),
 }

crates/persisting-compute/Cargo.toml

@@ -23,7 +23,6 @@ tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["rt"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-uuid = { version = "1", features = ["v4"] }
 
 [features]
 default = []

crates/persisting-compute/src/checkpoint.rs

@@ -33,6 +33,43 @@ impl CheckpointLedger {
         let failed = load_task_ids(&root.join("failures.ndjson")).await?;
         Ok(Self { ready, failed })
     }
+
+    /// Task ids in `failures.ndjson` whose stable `error_kind` matches one of
+    /// `kinds`. Legacy failure rows without the field are treated as `execute`.
+    pub async fn failed_ids_matching(
+        &self,
+        root: &Path,
+        kinds: &[String],
+    ) -> Result<HashSet<String>> {
+        let wanted: HashSet<&str> = kinds.iter().map(String::as_str).collect();
+        if wanted.is_empty() {
+            return Ok(HashSet::new());
+        }
+        let path = root.join("failures.ndjson");
+        let mut matched = HashSet::new();
+        if !path.exists() {
+            return Ok(matched);
+        }
+        let f = fs::File::open(&path)
+            .await
+            .with_context(|| format!("open {}", path.display()))?;
+        let mut lines = BufReader::new(f).lines();
+        while let Some(line) = lines.next_line().await? {
+            let Ok(value) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
+                continue;
+            };
+            let kind = value
+                .get("error_kind")
+                .and_then(|kind| kind.as_str())
+                .unwrap_or("execute");
+            if wanted.contains(kind) {
+                if let Some(id) = value.get("task_id").and_then(|id| id.as_str()) {
+                    matched.insert(id.to_string());
+                }
+            }
+        }
+        Ok(matched)
+    }
 }
 
 async fn load_task_ids(path: &Path) -> Result<HashSet<String>> {
@@ -264,6 +301,31 @@ mod tests {
         assert_eq!(tracker.snapshot().fail, 1);
     }
 
+    #[tokio::test]
+    async fn failed_ids_can_be_filtered_by_error_kind() {
+        let dir = tempfile::tempdir().unwrap();
+        fs::write(
+            dir.path().join("failures.ndjson"),
+            concat!(
+                r#"{"task_id":"execute","error_kind":"execute"}"#,
+                "\n",
+                r#"{"task_id":"infra","error_kind":"infra"}"#,
+                "\n",
+                r#"{"task_id":"legacy"}"#,
+                "\n"
+            ),
+        )
+        .await
+        .unwrap();
+        let ledger = CheckpointLedger::load(dir.path()).await.unwrap();
+        let ids = ledger
+            .failed_ids_matching(dir.path(), &["execute".into()])
+            .await
+            .unwrap();
+        assert!(ids.contains("execute") && ids.contains("legacy"));
+        assert!(!ids.contains("infra"));
+    }
+
     #[tokio::test]
     async fn load_skips_corrupt_lines_and_keeps_valid() {
         let dir = tempfile::tempdir().unwrap();

crates/persisting-compute/src/cli.rs

@@ -13,6 +13,7 @@ use crate::skip::SkipSet;
 use crate::task::TaskResult;
 use anyhow::{bail, Context, Result};
 use clap::{Args, ValueEnum};
+use std::collections::BTreeMap;
 use std::path::PathBuf;
 use std::process::ExitCode;
 use std::sync::Arc;
@@ -55,10 +56,25 @@ pub struct ComputeArgs {
     #[arg(long, value_name = "DIR")]
     pub sink: Option<PathBuf>,
 
+    /// Logical run identifier exposed to workers through `persisting_compute.context()`.
+    /// Defaults to the sink directory name, or the plan filename for ephemeral runs.
+    #[arg(long)]
+    pub job_id: Option<String>,
+
+    /// Capability labels exposed as `context()["labels"]` (comma-separated).
+    /// They are informational in this release; scheduling remains least-loaded.
+    #[arg(long, value_delimiter = ',')]
+    pub worker_label: Vec<String>,
+
     /// Resume from `--sink`: skip task ids already in ready/failures.
     #[arg(long)]
     pub resume: bool,
 
+    /// With `--resume`, run failures of these kinds again (`execute`, `infra`,
+    /// or `cancelled`). May be repeated or comma-separated.
+    #[arg(long, value_delimiter = ',')]
+    pub rerun_failed: Vec<String>,
+
     /// Also append terminal results to a Lance trajectory (requires `--sink`).
     /// Writes `compute.result` / `compute.failure` events under traj storage.
     #[cfg(feature = "traj-sink")]
@@ -160,6 +176,31 @@ pub async fn run_compute(args: ComputeArgs) -> Result<ExitCode> {
     }
 
     let under_torch = std::env::var_os("RANK").is_some();
+    let job_id = args.job_id.clone().unwrap_or_else(|| {
+        args.sink
+            .as_ref()
+            .and_then(|path| path.file_name())
+            .and_then(|name| name.to_str())
+            .filter(|name| !name.is_empty())
+            .map(str::to_string)
+            .or_else(|| {
+                script
+                    .file_stem()
+                    .and_then(|name| name.to_str())
+                    .map(str::to_string)
+            })
+            .unwrap_or_else(|| "compute".into())
+    });
+    std::env::set_var("PERSISTING_COMPUTE_JOB_ID", &job_id);
+    if let Some(dir) = &args.sink {
+        std::env::set_var("PERSISTING_COMPUTE_OUTPUT_DIR", dir);
+    }
+    if !args.worker_label.is_empty() {
+        std::env::set_var(
+            "PERSISTING_COMPUTE_WORKER_LABELS",
+            args.worker_label.join(","),
+        );
+    }
     let per_worker = args.per_worker.max(1);
     let max_inflight = args.max_inflight.unwrap_or_else(|| {
         if under_torch {
@@ -225,6 +266,14 @@ pub async fn run_compute(args: ComputeArgs) -> Result<ExitCode> {
     if args.resume && file_sink.is_none() {
         bail!("--resume requires --sink DIR");
     }
+    if !args.rerun_failed.is_empty() && !args.resume {
+        bail!("--rerun-failed requires --resume --sink DIR");
+    }
+    for kind in &args.rerun_failed {
+        if !matches!(kind.as_str(), "execute" | "infra" | "cancelled") {
+            bail!("--rerun-failed expects execute, infra, or cancelled (got {kind:?})");
+        }
+    }
 
     #[cfg(feature = "traj-sink")]
     if args.traj && args.sink.is_none() {
@@ -237,11 +286,19 @@ pub async fn run_compute(args: ComputeArgs) -> Result<ExitCode> {
             let ledger = CheckpointLedger::load(dir)
                 .await
                 .context("load checkpoint ledger")?;
-            let skip = ledger.skip_ids();
+            let mut skip = ledger.skip_ids();
+            let rerun = ledger
+                .failed_ids_matching(dir, &args.rerun_failed)
+                .await
+                .context("filter failed task ids")?;
+            for id in &rerun {
+                skip.remove(id);
+            }
             eprintln!(
-                "[ckpt] resume: ready={} fail={} skip_total={}",
+                "[ckpt] resume: ready={} fail={} rerun={} skip_total={}",
                 ledger.ready.len(),
                 ledger.failed.len(),
+                rerun.len(),
                 skip.len()
             );
             tracker.seed_from_ledger(&ledger);
@@ -329,19 +386,18 @@ pub async fn run_compute(args: ComputeArgs) -> Result<ExitCode> {
     }
 
     let failed = collected.iter().filter(|r| !r.ok || r.cancelled).count();
+    let summary = build_run_summary(&collected, args.sink.as_ref());
+    if let Some(dir) = &args.sink {
+        tokio::fs::write(
+            dir.join("summary.json"),
+            serde_json::to_vec_pretty(&summary).context("encode run summary")?,
+        )
+        .await
+        .context("write summary.json")?;
+    }
 
     if matches!(results_fmt, ResultsFormat::Summary) {
-        let ok = collected.iter().filter(|r| r.ok && !r.cancelled).count();
-        println!(
-            "{}",
-            serde_json::json!({
-                "total": collected.len(),
-                "ok": ok,
-                "failed": failed,
-                "cancelled": collected.iter().filter(|r| r.cancelled).count(),
-                "sink": args.sink.as_ref().map(|p| p.display().to_string()),
-            })
-        );
+        println!("{summary}");
         for r in &collected {
             if !r.ok {
                 if let Ok(line) = r.to_ndjson() {
@@ -358,6 +414,44 @@ pub async fn run_compute(args: ComputeArgs) -> Result<ExitCode> {
     })
 }
 
+fn build_run_summary(results: &[TaskResult], sink: Option<&PathBuf>) -> serde_json::Value {
+    let mut aggregates: BTreeMap<String, (f64, u64)> = BTreeMap::new();
+    let mut error_kinds: BTreeMap<String, u64> = BTreeMap::new();
+    let mut artifacts = 0u64;
+    for result in results {
+        for (name, value) in &result.metrics {
+            let entry = aggregates.entry(name.clone()).or_insert((0.0, 0));
+            entry.0 += value;
+            entry.1 += 1;
+        }
+        artifacts += result.artifacts.len() as u64;
+        if let Some(kind) = &result.error_kind {
+            *error_kinds
+                .entry(format!("{kind:?}").to_lowercase())
+                .or_default() += 1;
+        }
+    }
+    let metrics: BTreeMap<_, _> = aggregates
+        .into_iter()
+        .map(|(name, (sum, count))| {
+            (
+                name,
+                serde_json::json!({"count": count, "sum": sum, "mean": sum / count as f64}),
+            )
+        })
+        .collect();
+    serde_json::json!({
+        "total": results.len(),
+        "ok": results.iter().filter(|r| r.ok && !r.cancelled).count(),
+        "failed": results.iter().filter(|r| !r.ok || r.cancelled).count(),
+        "cancelled": results.iter().filter(|r| r.cancelled).count(),
+        "error_kinds": error_kinds,
+        "metrics": metrics,
+        "artifact_count": artifacts,
+        "sink": sink.map(|p| p.display().to_string()),
+    })
+}
+
 /// Ensure tracing is initialized once (safe to call from nested CLI).
 ///
 /// Default is quiet: hush Pulsing actor lifecycle noise. Override with `RUST_LOG`,

crates/persisting-compute/src/driver.rs

@@ -16,7 +16,7 @@ use crate::pulsing_ext::{ask_timeout, ASK_TIMEOUT};
 use crate::scheduler::{AcquireError, Scheduler, StickyLost, WorkerPool};
 use crate::sink_writer::SinkSubmitter;
 use crate::skip::SkipSet;
-use crate::task::{unix_now, TaskExpr, TaskResult};
+use crate::task::{unix_now, ErrorKind, TaskExpr, TaskResult};
 use crate::worker::{WorkerCommand, WorkerReply};
 use anyhow::Result;
 use futures::stream::FuturesUnordered;
@@ -241,8 +241,15 @@ async fn execute_with_placement(
                 observer
                     .task_finished(&task_id, false, false, Some(err.clone()), &sched)
                     .await;
-                let mut r =
-                    TaskResult::failure(task_id, format!("infra: {err}"), None, "infra", started);
+                let mut r = TaskResult::failure_with_kind(
+                    task_id,
+                    format!("infra: {err}"),
+                    None,
+                    "infra",
+                    started,
+                    ErrorKind::Infra,
+                    true,
+                );
                 r.infra_retries = attempt;
                 return Ok(r);
             }
@@ -252,8 +259,15 @@ async fn execute_with_placement(
                 observer
                     .task_finished(&task_id, false, false, Some(err.clone()), &sched)
                     .await;
-                let mut r =
-                    TaskResult::failure(task_id, format!("infra: {err}"), None, "infra", started);
+                let mut r = TaskResult::failure_with_kind(
+                    task_id,
+                    format!("infra: {err}"),
+                    None,
+                    "infra",
+                    started,
+                    ErrorKind::Infra,
+                    true,
+                );
                 r.infra_retries = attempt;
                 return Ok(r);
             }
@@ -337,12 +351,14 @@ async fn execute_with_placement(
     observer
         .task_finished(&task_id, false, false, Some(err.clone()), &sched)
         .await;
-    let mut r = TaskResult::failure(
+    let mut r = TaskResult::failure_with_kind(
         task_id,
         format!("infra retries exhausted: {err}"),
         None,
         "infra",
         started,
+        ErrorKind::Infra,
+        true,
     );
     r.infra_retries = infra_retries;
     Ok(r)