c0re: reject apply_commit on duplicate flake inputs (#317)

hive-c0re/src/flake_check.rs

@@ -0,0 +1,284 @@
+//! Pre-apply validation for agent `flake.lock` files (closes part of #317).
+//!
+//! Every `request_apply_commit` lands a `proposal/<id>` tag in the
+//! agent's applied repo before the operator sees the approval. We
+//! parse `flake.lock` from that tag's tree and reject the request if
+//! two or more nodes share an identical `original` field — that
+//! signals a missing `inputs.<X>.inputs.nixpkgs.follows = "nixpkgs"`
+//! directive in `flake.nix` and would inflate meta's lock with
+//! duplicates after deploy.
+//!
+//! Per mara's scope note on #317 (comment 4189): the check runs on
+//! the agent repo, not meta, and catches *new* violations only.
+//! Existing agents whose lock already has duplicates are out of
+//! scope here and get a coordinated config-change pass via the
+//! manager instead.
+
+use std::collections::BTreeMap;
+use std::fmt::Write as _;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use serde_json::Value;
+
+use crate::lifecycle::git_command;
+
+/// One group of `flake.lock` nodes that all share the same canonical
+/// `original` reference. Surfaced in the rejection message so the
+/// operator (and the manager that submitted the apply) can see
+/// exactly which input pair needs a `follows` directive.
+#[derive(Debug, Clone)]
+pub struct DuplicateGroup {
+    /// One of the original `Value`s from the lock — used for
+    /// pretty-printing in the error message. Canonical equivalence
+    /// is enforced by the `BTreeMap` key in `duplicate_groups`, so
+    /// we don't need to keep the canonicalised form on the struct.
+    pub original: Value,
+    /// Names of the flake.lock nodes that share this `original`,
+    /// sorted for stable error output.
+    pub keys: Vec<String>,
+}
+
+/// Read `flake.lock` from `<tag>:flake.lock` in `repo`. Returns
+/// `Ok(None)` when the file isn't tracked in that tag (no inputs ⇒
+/// nothing to dedup); `Err` only on real git plumbing failures.
+async fn read_lock_at_tag(repo: &Path, tag: &str) -> Result<Option<String>> {
+    let spec = format!("{tag}:flake.lock");
+    let out = git_command()
+        .current_dir(repo)
+        .args(["show", &spec])
+        .output()
+        .await
+        .with_context(|| format!("git show {spec} in {}", repo.display()))?;
+    if !out.status.success() {
+        let stderr = String::from_utf8_lossy(&out.stderr);
+        // git uses different messages for "path not in tree" depending
+        // on version + whether the path collides with an on-disk file.
+        // Treat any of them as "no flake.lock in this commit" rather
+        // than a hard failure — an agent with empty `inputs = { }` is
+        // a legitimate, dedup-clean state.
+        if stderr.contains("does not exist")
+            || stderr.contains("exists on disk, but not in")
+            || stderr.contains("Path '")
+        {
+            return Ok(None);
+        }
+        anyhow::bail!("git show {spec} failed: {}", stderr.trim());
+    }
+    Ok(Some(String::from_utf8_lossy(&out.stdout).into_owned()))
+}
+
+/// Recursively serialise `v` with object keys sorted, so two
+/// JSON values that differ only in key insertion order produce the
+/// same string. `serde_json::Value` preserves `IndexMap` order by
+/// default, which is fine for parsing but breaks our group-by-key
+/// idea — hence this hand-rolled canonicaliser.
+fn canonical_json(v: &Value) -> String {
+    match v {
+        Value::Object(map) => {
+            let mut keys: Vec<&String> = map.keys().collect();
+            keys.sort();
+            let mut s = String::from("{");
+            for (i, k) in keys.iter().enumerate() {
+                if i > 0 {
+                    s.push(',');
+                }
+                s.push_str(&serde_json::to_string(k).unwrap_or_default());
+                s.push(':');
+                s.push_str(&canonical_json(&map[*k]));
+            }
+            s.push('}');
+            s
+        }
+        Value::Array(arr) => {
+            let mut s = String::from("[");
+            for (i, x) in arr.iter().enumerate() {
+                if i > 0 {
+                    s.push(',');
+                }
+                s.push_str(&canonical_json(x));
+            }
+            s.push(']');
+            s
+        }
+        _ => serde_json::to_string(v).unwrap_or_default(),
+    }
+}
+
+/// Parse `raw` (a `flake.lock` JSON document) and return every group
+/// of nodes whose `original` field is identical. Nodes without an
+/// `original` (the synthetic `root`, or anomalous entries) are
+/// skipped. Groups with only one member are filtered out — only true
+/// duplicates surface.
+///
+/// Pure function, no I/O — covered by the unit tests below.
+pub fn duplicate_groups(raw: &str) -> Result<Vec<DuplicateGroup>> {
+    let json: Value = serde_json::from_str(raw).context("parse flake.lock")?;
+    let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else {
+        return Ok(Vec::new());
+    };
+    let mut groups: BTreeMap<String, DuplicateGroup> = BTreeMap::new();
+    for (name, node) in nodes {
+        let Some(original) = node.get("original") else {
+            continue;
+        };
+        let key = canonical_json(original);
+        let entry = groups.entry(key).or_insert_with(|| DuplicateGroup {
+            original: original.clone(),
+            keys: Vec::new(),
+        });
+        entry.keys.push(name.clone());
+    }
+    let mut dups: Vec<DuplicateGroup> = groups
+        .into_values()
+        .filter(|g| g.keys.len() > 1)
+        .collect();
+    for g in &mut dups {
+        g.keys.sort();
+    }
+    Ok(dups)
+}
+
+/// Run the dedup check against the agent's freshly-applied tree.
+///
+/// `Ok(())` means either the commit doesn't carry a `flake.lock` (no
+/// inputs declared) or every node has a unique `original`. `Err`
+/// carries a multi-line message listing every duplicate group with
+/// the offending node names, suitable for surfacing on the failed
+/// approval row.
+pub async fn check_no_duplicate_inputs(repo: &Path, tag: &str) -> Result<()> {
+    let Some(raw) = read_lock_at_tag(repo, tag).await? else {
+        return Ok(());
+    };
+    let dups = duplicate_groups(&raw)?;
+    if dups.is_empty() {
+        return Ok(());
+    }
+    let mut msg = String::from(
+        "flake.lock has duplicate flake inputs — add a `follows` directive in flake.nix to collapse them:\n",
+    );
+    for g in &dups {
+        let original = serde_json::to_string(&g.original).unwrap_or_else(|_| "?".into());
+        let _ = writeln!(msg, "  - {original} → nodes [{}]", g.keys.join(", "));
+    }
+    anyhow::bail!("{}", msg.trim_end());
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const CLEAN_LOCK: &str = r#"{
+        "nodes": {
+            "nixpkgs": {
+                "locked": {"rev": "aaa"},
+                "original": {"owner": "NixOS", "repo": "nixpkgs", "ref": "nixos-25.11", "type": "github"}
+            },
+            "nixpkgs-unstable": {
+                "locked": {"rev": "bbb"},
+                "original": {"owner": "NixOS", "repo": "nixpkgs", "ref": "nixpkgs-unstable", "type": "github"}
+            },
+            "root": {"inputs": {"nixpkgs": "nixpkgs"}}
+        },
+        "root": "root",
+        "version": 7
+    }"#;
+
+    const DUPLICATE_LOCK: &str = r#"{
+        "nodes": {
+            "nixpkgs": {
+                "locked": {"rev": "aaa"},
+                "original": {"owner": "NixOS", "repo": "nixpkgs", "ref": "nixos-25.11", "type": "github"}
+            },
+            "nixpkgs_2": {
+                "locked": {"rev": "ccc"},
+                "original": {"owner": "NixOS", "repo": "nixpkgs", "ref": "nixos-25.11", "type": "github"}
+            },
+            "nixpkgs_3": {
+                "locked": {"rev": "ddd"},
+                "original": {"owner": "NixOS", "repo": "nixpkgs", "ref": "nixos-25.11", "type": "github"}
+            },
+            "treefmt-nix": {
+                "locked": {"rev": "eee"},
+                "original": {"owner": "numtide", "repo": "treefmt-nix", "type": "github"}
+            },
+            "treefmt-nix_2": {
+                "locked": {"rev": "fff"},
+                "original": {"type": "github", "owner": "numtide", "repo": "treefmt-nix"}
+            },
+            "root": {"inputs": {"nixpkgs": "nixpkgs"}}
+        },
+        "root": "root",
+        "version": 7
+    }"#;
+
+    #[test]
+    fn clean_lock_has_no_duplicates() {
+        let dups = duplicate_groups(CLEAN_LOCK).expect("parse");
+        assert!(dups.is_empty(), "expected no dups, got {dups:#?}");
+    }
+
+    #[test]
+    fn duplicate_lock_reports_groups() {
+        let dups = duplicate_groups(DUPLICATE_LOCK).expect("parse");
+        assert_eq!(dups.len(), 2, "expected nixpkgs + treefmt-nix groups");
+
+        // Order is BTreeMap-stable: sorted by canonical_json key. The
+        // numtide treefmt-nix key sorts before the NixOS nixpkgs one
+        // because `numtide` < `NixOS` lexicographically (case-sensitive,
+        // capitals come first... wait — capital N is 0x4e, lowercase n
+        // is 0x6e, so capitals come first). So nixpkgs group sorts
+        // first. Verify by content instead of position to avoid coupling
+        // to that subtlety.
+        let nixpkgs_group = dups
+            .iter()
+            .find(|g| g.original.get("ref").is_some())
+            .expect("nixpkgs group present");
+        assert_eq!(
+            nixpkgs_group.keys,
+            vec!["nixpkgs", "nixpkgs_2", "nixpkgs_3"]
+        );
+
+        let treefmt_group = dups
+            .iter()
+            .find(|g| g.original.get("ref").is_none())
+            .expect("treefmt-nix group present");
+        assert_eq!(treefmt_group.keys, vec!["treefmt-nix", "treefmt-nix_2"]);
+    }
+
+    #[test]
+    fn key_order_in_original_does_not_matter() {
+        // The treefmt-nix and treefmt-nix_2 entries above use different
+        // key orderings for `original` ({owner,repo,type} vs
+        // {type,owner,repo}); duplicate_groups should still merge them.
+        let dups = duplicate_groups(DUPLICATE_LOCK).expect("parse");
+        let treefmt = dups
+            .iter()
+            .find(|g| g.keys.iter().any(|k| k == "treefmt-nix"))
+            .expect("treefmt-nix group");
+        assert!(treefmt.keys.contains(&"treefmt-nix_2".to_owned()));
+    }
+
+    #[test]
+    fn nodes_without_original_are_ignored() {
+        // The synthetic `root` node has no `original` and must not be
+        // grouped against anything.
+        let dups = duplicate_groups(CLEAN_LOCK).expect("parse");
+        assert!(dups.iter().all(|g| !g.keys.iter().any(|k| k == "root")));
+    }
+
+    #[test]
+    fn missing_nodes_object_is_not_an_error() {
+        // A lock file that's syntactically JSON but lacks `nodes` (e.g.
+        // a partial test fixture) should fail open — no dups reported.
+        let raw = r#"{"root": "root", "version": 7}"#;
+        let dups = duplicate_groups(raw).expect("parse");
+        assert!(dups.is_empty());
+    }
+
+    #[test]
+    fn invalid_json_is_a_hard_error() {
+        let err = duplicate_groups("not-json-at-all").unwrap_err();
+        assert!(format!("{err:#}").contains("parse flake.lock"));
+    }
+}

hive-c0re/src/main.rs

@@ -18,6 +18,7 @@ mod dashboard;
 mod dashboard_events;
 mod events_vacuum;
 mod stats_vacuum;
+mod flake_check;
 mod forge;
 mod lifecycle;
 mod limits;

hive-c0re/src/manager_server.rs

@@ -633,14 +633,34 @@ async fn submit_apply_commit(
         .approvals
         .set_fetched_sha(id, &sha)
         .map_err(|e| anyhow::anyhow!("persist fetched_sha: {e:#}"))?;
+    // #317 dedup gate: parse the just-fetched flake.lock and reject if
+    // the agent declared two inputs that resolve to the same upstream
+    // (a missing `follows` directive). Runs after `set_fetched_sha` so
+    // the failed approval row carries the sha that broke — handy when
+    // the manager needs to inspect the bad commit.
+    let sha_short = sha[..sha.len().min(12)].to_owned();
+    if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await {
+        let note = format!("{e:#}");
+        let _ = coord.approvals.mark_failed(id, &note);
+        coord.emit_approval_resolved(
+            id,
+            agent,
+            "apply_commit",
+            Some(sha_short.clone()),
+            "failed",
+            Some(note),
+            description.map(str::to_owned),
+        );
+        return Err(anyhow::anyhow!("flake dedup check: {e:#}"));
+    }
     // Mirror the freshly-planted proposal/<id> tag to the forge.
     if let Err(e) = crate::forge::push_config(agent).await {
         tracing::warn!(%agent, %id, error = ?e, "forge: push_config after submit failed");
     }
     // Phase 5b: surface the new pending approval on the dashboard
     // event channel. Compute the diff once here so live subscribers
-    // get a fully-formed row without a snapshot refetch.
+    // get a fully-formed row without a snapshot refetch. `sha_short`
-    let sha_short = sha[..sha.len().min(12)].to_owned();
+    // is reused from the dedup gate above.
     let diff = crate::dashboard::approval_diff(agent, id).await;
     coord.emit_approval_added(
         id,