Feat/multi rollout viewer

.gitignore

@@ -1,27 +1 @@
-# Rust
-/target/
-
-# Frontend
-/ui/node_modules/
-/ui/dist/
-
-# Editor / OS
-.DS_Store
-*.swp
-*.swo
-.idea/
-.vscode/
-
-# Local cluster configs (paths and SLURM accounts vary per user/site)
-/labctl.toml
-cluster.*.local.toml
-
-# Local registries / artifacts (should never be checked in)
-*.db
-*.db-journal
-*.db-wal
-*.db-shm
-
-# Env / secrets
-.env
-.env.*
+*.log

README.md

@@ -1,89 +1 @@
-# labctl
-
-Reproducible lab run envelope, artifact lineage, and async eval control
-plane for ML workflows on a SLURM cluster.
-
-`labctl` wraps recipe TOML files into versioned SLURM jobs, captures their
-inputs/outputs in a filesystem-truth registry, and provides a small
-read-only web UI for monitoring runs, comparing metrics, and tracing
-artifact lineage. It is multi-user by design: every action runs under
-the invoking user's own uid, and SLURM job ownership matches.
-
-## Architecture in one paragraph
-
-The filesystem under `runs_base/` is the source of truth. Each user's
-runs land at `runs_base/runs/<user>/<run_id>/.lab/`; artifacts at
-`<artifact_root>/<kind>/<user>/<alias>/`; aliases, eval-request
-dedup, pipelines and events live in their own subdirs. The CLI is the
-only writer: `labctl run` opens the registry directly, snapshots the
-source repo, renders the sbatch script, and shells out to `sbatch`
-under its own uid. A per-user `labctl agent` runs reconcile + evald +
-throttle as a systemd unit. `labctl serve` is a read-only HTTP server
-that anyone can run; it builds an in-memory SQLite cache from the tree
-on startup.
-
-## Workflow philosophy
-
-**Stable artifacts, composable pipelines, ephemeral runs.** Data
-preparation lives in long-running pipelines that produce named
-artifacts; experiments are small pipeline files that extend a specific
-historical run via `from = "<run_id>"`. Stage-level cache-hit
-short-circuits any stage whose key already has a succeeded run on disk;
-in-flight coalescing routes parallel submissions that share an upstream
-key onto a single SLURM job instead of duplicating it. The net effect:
-fan out experiments without duplicating work, and pin them to frozen
-upstream state without freezing the registry. See
-`examples/pipelines/from-pinned.toml`.
-
-## Install
-
-```bash
-./scripts/install.sh
-```
-
-Builds the embedded frontend, runs `cargo install --path . --features ui`
-(so `labctl` lands in `~/.cargo/bin/` and is on PATH for any normal
-Rust setup), and points git at `scripts/hooks/` so `cargo test
---all-features` runs before every push.
-
-Re-run after `git pull` to refresh the installed binary.
-
-## Quick start
-
-```bash
-./scripts/install.sh   # cargo install + git hook setup
-labctl init            # interactive bootstrap — config, dirs, agent, doctor
-labctl run path/to/recipe.toml
-```
-
-`labctl init` is a full setup wizard, not just a config writer. It
-picks one of four modes — interactively or via flag — and does
-*everything* needed to leave you with a working setup:
-
-| Situation | Command | What it does |
-| --- | --- | --- |
-| Brand-new cluster, no template | `labctl init` | Greenfield: SLURM probe + prompts → fresh cluster.toml, per-user dirs, agent unit, doctor. |
-| You already wrote a cluster.toml | `labctl init --use ~/cluster.toml` | Symlinks it into the default config location; creates dirs; installs agent; doctor. |
-| Standing labctl up at a new site (had one at site A, now at site B) | `labctl init --migrate-from /path/to/cluster.berlin.toml` | Schema carries over; site-local paths reviewed interactively. |
-| Joining a colleague's shared registry on this cluster | `labctl init --join /shared/cluster.berlin.toml` | Paths kept verbatim; per-user agent + per-user subdirs only. |
-
-The config is written to `~/.config/labctl/cluster.toml` by default,
-so all later `labctl <cmd>` invocations work without `--cluster`.
-Override with `--cluster <path>` or `$LABCTL_CLUSTER`.
-
-Once set up:
-
-```bash
-labctl run path/to/recipe.toml                   # submit
-labctl serve --bind 127.0.0.1:8765               # UI (ssh -L from your laptop)
-labctl doctor                                    # re-verify anytime
-```
-
-See `docs/ONBOARDING.md` for the full walkthrough,
-`docs/RECIPE_CONTRACT.md` for the contract between labctl and your
-recipes, and `examples/` for cluster / recipe / policy templates.
-
-## Status
-
-Pre-1.0. Multi-user from day one of the rewrite, single trust domain
-(loopback + uid).
+hello

src/pg_store.rs

@@ -860,6 +860,14 @@ impl PgStore {
                  er.eval_run_id,
                  COALESCE(er_run.status, 'pending') AS state,
                  cp.metadata_json AS checkpoint_metadata,
+                 (
+                     SELECT a.id
+                     FROM run_outputs ro
+                     JOIN artifacts a ON a.id = ro.artifact_id
+                     WHERE ro.run_id = er.eval_run_id AND a.kind = 'eval_result'
+                     ORDER BY a.created_at, a.id
+                     LIMIT 1
+                 ) AS eval_result_artifact_id,
                  (
                      SELECT a.metadata_json
                      FROM run_outputs ro
@@ -892,6 +900,143 @@ impl PgStore {
                     eval_run_id: r.try_get("eval_run_id")?,
                     state: r.try_get("state")?,
                     checkpoint_metadata: checkpoint_metadata.map(|j| j.0),
+                    eval_result_artifact_id: r.try_get("eval_result_artifact_id")?,
+                    eval_result_metadata: eval_result_metadata.map(|j| j.0),
+                })
+            })
+            .collect()
+    }
+
+    /// Like [`Self::eval_series_rows`], but for the rollout *browser*: it
+    /// aggregates a training run's evals one export hop down, not just its
+    /// direct checkpoints. In the per-checkpoint-export workflow the
+    /// rollout evals reference HF-export checkpoints whose `producer_run_id`
+    /// is a per-checkpoint *export* run, not the training run — so the
+    /// one-hop `eval_series_rows` (cp.producer_run_id = run) misses them.
+    ///
+    /// Matches an eval when its checkpoint was produced either directly by
+    /// `run_id`, or by a run that consumed a checkpoint produced by `run_id`
+    /// (the export run). Direct-eval workflows (text SFT) still match via the
+    /// first clause. The `step` still comes from the eval's checkpoint
+    /// metadata, which carries it across the export.
+    pub async fn rollout_series_rows(&self, run_id: &str) -> Result<Vec<EvalSeriesRow>> {
+        let rows = sqlx::query(
+            "SELECT
+                 er.eval_key,
+                 er.checkpoint_artifact_id,
+                 er.eval_recipe_hash,
+                 er.policy_id,
+                 er.eval_run_id,
+                 COALESCE(er_run.status, 'pending') AS state,
+                 cp.metadata_json AS checkpoint_metadata,
+                 (
+                     SELECT a.id
+                     FROM run_outputs ro
+                     JOIN artifacts a ON a.id = ro.artifact_id
+                     WHERE ro.run_id = er.eval_run_id AND a.kind = 'eval_result'
+                     ORDER BY a.created_at, a.id
+                     LIMIT 1
+                 ) AS eval_result_artifact_id,
+                 (
+                     SELECT a.metadata_json
+                     FROM run_outputs ro
+                     JOIN artifacts a ON a.id = ro.artifact_id
+                     WHERE ro.run_id = er.eval_run_id AND a.kind = 'eval_result'
+                     ORDER BY a.created_at, a.id
+                     LIMIT 1
+                 ) AS eval_result_metadata
+             FROM eval_requests er
+             LEFT JOIN artifacts cp ON cp.id = er.checkpoint_artifact_id
+             LEFT JOIN runs er_run ON er_run.id = er.eval_run_id
+             WHERE cp.producer_run_id = $1
+                OR er.eval_run_id = $1
+                OR cp.producer_run_id IN (
+                       SELECT ri.run_id
+                       FROM run_inputs ri
+                       JOIN artifacts mc ON mc.id = ri.artifact_id
+                       WHERE mc.producer_run_id = $1
+                   )
+             ORDER BY er.created_at",
+        )
+        .bind(run_id)
+        .fetch_all(&self.pool)
+        .await
+        .with_context(|| format!("rollout_series_rows({run_id})"))?;
+        rows.into_iter()
+            .map(|r| {
+                let checkpoint_metadata: Option<sqlx::types::Json<Value>> =
+                    r.try_get("checkpoint_metadata")?;
+                let eval_result_metadata: Option<sqlx::types::Json<Value>> =
+                    r.try_get("eval_result_metadata")?;
+                Ok(EvalSeriesRow {
+                    eval_key: r.try_get("eval_key")?,
+                    checkpoint_artifact_id: r.try_get("checkpoint_artifact_id")?,
+                    eval_recipe_hash: r.try_get("eval_recipe_hash")?,
+                    policy_id: r.try_get("policy_id")?,
+                    eval_run_id: r.try_get("eval_run_id")?,
+                    state: r.try_get("state")?,
+                    checkpoint_metadata: checkpoint_metadata.map(|j| j.0),
+                    eval_result_artifact_id: r.try_get("eval_result_artifact_id")?,
+                    eval_result_metadata: eval_result_metadata.map(|j| j.0),
+                })
+            })
+            .collect()
+    }
+
+    /// Rollout evals discovered purely by *lineage*, not via `eval_requests`.
+    /// Picks up manually-submitted freeroll evals (which never create an
+    /// eval_request row) so they still appear in the rollout browser. Finds
+    /// every `eval_result` artifact whose producing run consumed a checkpoint
+    /// that is produced directly by `run_id`, or by a run that consumed a
+    /// checkpoint produced by `run_id` (the one export hop) — the same
+    /// reachability as [`Self::rollout_series_rows`].
+    ///
+    /// `policy_id` is set to the eval run's `recipe_name`; the caller
+    /// normalizes it (collapsing the `step<N>` token) so freerolls of one
+    /// family at different checkpoints group into a single browsable series.
+    /// Rows already covered by `eval_requests` are deduped by the caller.
+    pub async fn lineage_rollout_rows(&self, run_id: &str) -> Result<Vec<EvalSeriesRow>> {
+        let rows = sqlx::query(
+            "SELECT DISTINCT
+                 a.id  AS eval_result_artifact_id,
+                 a.metadata_json AS eval_result_metadata,
+                 a.producer_run_id AS eval_run_id,
+                 COALESCE(er_run.status, 'pending') AS state,
+                 COALESCE(er_run.recipe_name, '') AS recipe_name,
+                 cp.id AS checkpoint_artifact_id,
+                 cp.metadata_json AS checkpoint_metadata
+             FROM artifacts a
+             JOIN runs er_run ON er_run.id = a.producer_run_id
+             JOIN run_inputs ri ON ri.run_id = a.producer_run_id
+             JOIN artifacts cp ON cp.id = ri.artifact_id AND cp.kind = 'checkpoint'
+             WHERE a.kind = 'eval_result'
+               AND ( cp.producer_run_id = $1
+                     OR cp.producer_run_id IN (
+                            SELECT ri2.run_id
+                            FROM run_inputs ri2
+                            JOIN artifacts mc ON mc.id = ri2.artifact_id
+                            WHERE mc.producer_run_id = $1
+                        ) )",
+        )
+        .bind(run_id)
+        .fetch_all(&self.pool)
+        .await
+        .with_context(|| format!("lineage_rollout_rows({run_id})"))?;
+        rows.into_iter()
+            .map(|r| {
+                let checkpoint_metadata: Option<sqlx::types::Json<Value>> =
+                    r.try_get("checkpoint_metadata")?;
+                let eval_result_metadata: Option<sqlx::types::Json<Value>> =
+                    r.try_get("eval_result_metadata")?;
+                Ok(EvalSeriesRow {
+                    eval_key: String::new(),
+                    checkpoint_artifact_id: r.try_get("checkpoint_artifact_id")?,
+                    eval_recipe_hash: String::new(),
+                    policy_id: r.try_get("recipe_name")?,
+                    eval_run_id: r.try_get("eval_run_id")?,
+                    state: r.try_get("state")?,
+                    checkpoint_metadata: checkpoint_metadata.map(|j| j.0),
+                    eval_result_artifact_id: r.try_get("eval_result_artifact_id")?,
                     eval_result_metadata: eval_result_metadata.map(|j| j.0),
                 })
             })