read the cache facts on the hive and let hives read the closure bucket

hive-c0re reads its cache facts from the store every five minutes and writes /run/hyperhive/attic/{netrc,substituter,public-key}, each 0600. it accepts only an https url, so the pull token never travels in clear. hives get a read grant on the agent-closures bucket, and swarm-queue-client a reader for it. part 2 of 3: nothing uses the files or the reader until the probe lands.

swarm-queue-client/Cargo.toml

@@ -59,6 +59,9 @@ agent-permissions = ["dep:hive-types", "async-nats/jetstream"]
 # `agent_archetype::read`, for hive-c0re, which reads the archetype of each
 # agent it provisions or rebuilds. Explicit for the same reason as `notices`.
 agent-archetype = ["async-nats/jetstream"]
+# `agent_closures::read`, for hive-c0re, which checks the latest closure of each
+# agent it runs can be pulled. Explicit for the same reason as `notices`.
+agent-closures = ["async-nats/jetstream"]
 # `agents_md_initial::read_own`, for the agent harness, which reads its own
 # initial `AGENTS.md` text with one direct get. Explicit for the same reason as
 # `notices`.

swarm-queue-client/src/agent_closures.rs

@@ -17,3 +17,84 @@ pub async fn open_or_create(
     )
     .await
 }
+
+/// What reading one agent's key found.
+#[cfg(feature = "agent-closures")]
+#[derive(Debug)]
+pub enum Read {
+    /// The bucket answered: the agent's latest build, or `None` for an absent
+    /// or deleted key.
+    Current(Option<swarm_wire::agent_closures::AgentClosure>),
+    /// Nothing answers for the bucket's stream: no build has been recorded
+    /// yet, or `JetStream` is not serving it right now.
+    NoBucket,
+}
+
+/// Read `agent`'s latest closure record with one direct get.
+///
+/// # Errors
+///
+/// A failed request or an undecodable value. A request the queue refused for
+/// permissions reaches the client as a timeout, not as a refusal.
+#[cfg(feature = "agent-closures")]
+pub async fn read(
+    client: &async_nats::Client,
+    agent: &str,
+) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
+    use async_nats::jetstream::stream::DirectGetErrorKind;
+    use swarm_wire::agent_closures::{STREAM, key_subject};
+
+    let js = async_nats::jetstream::new(client.clone());
+    let stream = js.get_stream_no_info(STREAM).await?;
+    let message = match stream.direct_get_last_for_subject(key_subject(agent)).await {
+        Ok(message) => message,
+        Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
+            return Ok(Read::Current(None));
+        }
+        Err(e)
+            if matches!(
+                e.kind(),
+                DirectGetErrorKind::ErrorResponse(status, _)
+                    if status == async_nats::StatusCode::NO_RESPONDERS
+            ) =>
+        {
+            return Ok(Read::NoBucket);
+        }
+        Err(e) => return Err(e.into()),
+    };
+    Ok(Read::Current(decode(
+        message
+            .headers
+            .get("KV-Operation")
+            .map(async_nats::HeaderValue::as_str),
+        &message.payload,
+    )?))
+}
+
+/// One stored message as a record: a KV delete or purge is stored as an empty
+/// marker carrying the `KV-Operation` header, and reads as no record.
+#[cfg(feature = "agent-closures")]
+fn decode(
+    operation: Option<&str>,
+    payload: &[u8],
+) -> serde_json::Result<Option<swarm_wire::agent_closures::AgentClosure>> {
+    if matches!(operation, Some("DEL" | "PURGE")) {
+        return Ok(None);
+    }
+    serde_json::from_slice(payload).map(Some)
+}
+
+#[cfg(all(test, feature = "agent-closures"))]
+mod tests {
+    use super::decode;
+
+    #[test]
+    fn a_stored_record_reads_and_a_delete_is_no_record() {
+        let payload = br#"{"meta_rev":"abc","store_path":"/nix/store/x","system":"x86_64-linux","built_unix":7,"cached":true}"#;
+        let record = decode(None, payload).expect("decodes").expect("a record");
+        assert_eq!(record.meta_rev, "abc");
+        assert!(record.cached);
+        assert!(decode(Some("DEL"), b"").expect("decodes").is_none());
+        assert!(decode(Some("PURGE"), b"").expect("decodes").is_none());
+    }
+}

swarm-queue-client/src/lib.rs

@@ -251,8 +251,9 @@ pub mod operator_profiles;
 pub mod agent_archetype;
 
 /// Opening the bucket swarm-controller records each agent's latest built
-/// system closure in. The names and the stored type are
-/// `swarm_wire::agent_closures`; writing needs `kv`.
+/// system closure in, read by the hive that runs the agent. The names and the
+/// stored type are `swarm_wire::agent_closures`; writing needs `kv` and reading
+/// needs `agent-closures`.
 pub mod agent_closures;
 
 /// The bucket swarm-controller keeps the evaluated agent archetypes in — one