agent permissions: hold hive-c0re's stored grant until the agent's own read

Until its first successful read of its key, the harness read its
permissions as the default set, `execution` included. So did the
subagent daemon. An agent the operator had narrowed ran wider than
granted after every restart while the queue was unreachable, and in the
gap before the first 10 s poll. During the meta-repo import, every
non-ruth key is still unset, and the agent read that as the default too,
for as long as the import stayed pending.

mara picked option (a) on #5293: "a, removal of the old path as a follow
up in 1w".

hive-c0re now writes each agent's entry from
swarm-agent-permissions.json into the agent's runtime dir as
swarm-permissions.json (/run/hive in the container), together with
whether the import is still pending. It writes the file at every start
(write_dropins) and after every 30 s pass of the permissions worker. The
harness and the subagent daemon hold that stored list until their own
first read succeeds. While the import is pending, an unset key keeps
the stored list rather than reverting it to the default. A queue that
is down never blocks a start: the agent runs on the stored grant.

The decision lives in swarm_queue_client::agent_permissions::HiveCopy,
shared by both readers.

Part of #5170.

swarm-queue-client/src/agent_permissions.rs

@@ -132,6 +132,78 @@ pub enum Read {
     NoBucket,
 }
 
+/// The file hive-c0re writes an agent's [`HiveCopy`] to in that agent's
+/// runtime dir.
+#[cfg(feature = "agent-permissions")]
+pub const HIVE_COPY_FILE: &str = "swarm-permissions.json";
+
+/// Where the agent's own processes read [`HIVE_COPY_FILE`]: the runtime dir
+/// is mounted at `/run/hive` in the container.
+#[cfg(feature = "agent-permissions")]
+pub const HIVE_COPY_PATH: &str = "/run/hive/swarm-permissions.json";
+
+/// What hive-c0re holds for one agent: its record as the hive last stored it
+/// (read from this bucket, or taken from the meta repo before the import),
+/// and whether the meta repo's import is still pending. The agent's harness
+/// and subagent daemon hold [`HiveCopy::before_read`] until their own first
+/// read of the key succeeds, and judge every read with
+/// [`HiveCopy::after_read`].
+#[cfg(feature = "agent-permissions")]
+#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct HiveCopy {
+    #[serde(flatten)]
+    pub stored: AgentPermissions,
+    #[serde(default)]
+    pub import_pending: bool,
+}
+
+#[cfg(feature = "agent-permissions")]
+impl HiveCopy {
+    /// The copy at `path`. A missing or unreadable file is the empty copy,
+    /// which holds the default and has no import pending.
+    #[must_use]
+    pub fn read_at(path: &std::path::Path) -> Self {
+        let bytes = match std::fs::read(path) {
+            Ok(bytes) => bytes,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Self::default(),
+            Err(e) => {
+                tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy is unreadable");
+                return Self::default();
+            }
+        };
+        serde_json::from_slice(&bytes).unwrap_or_else(|e| {
+            tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy does not decode");
+            Self::default()
+        })
+    }
+
+    /// What `agent` holds before any read of its key has succeeded: the
+    /// stored list, or the default when the hive stored none.
+    #[must_use]
+    pub fn before_read(&self, agent: &str) -> Vec<hive_sh4re::permissions::Permission> {
+        self.stored
+            .permissions
+            .clone()
+            .unwrap_or_else(|| hive_sh4re::permissions::Permission::default_for(agent))
+    }
+
+    /// What `agent` holds after reading `read` from its key. An unset key
+    /// while the import is pending may only mean the grant is not imported
+    /// yet, so the stored list stands; otherwise unset is the default.
+    #[must_use]
+    pub fn after_read(
+        &self,
+        agent: &str,
+        read: &AgentPermissions,
+    ) -> Vec<hive_sh4re::permissions::Permission> {
+        match &read.permissions {
+            Some(list) => list.clone(),
+            None if self.import_pending => self.before_read(agent),
+            None => hive_sh4re::permissions::Permission::default_for(agent),
+        }
+    }
+}
+
 /// Read `agent`'s current permissions with one direct get: the one subject
 /// an agent's own grant covers, and one of those a hive's covers.
 ///
@@ -234,6 +306,33 @@ mod tests {
         assert_eq!(ruth.permissions, Some(vec![Permission::ManageRootAgent]));
     }
 
+    /// Before a read the stored list stands; after one, a set key wins and an
+    /// unset key is the default unless the import is pending.
+    #[cfg(feature = "agent-permissions")]
+    #[test]
+    fn the_hive_copy_stands_until_read_and_while_the_import_is_pending() {
+        use hive_sh4re::permissions::Permission;
+        let record = |json: &str| -> AgentPermissions { serde_json::from_str(json).unwrap() };
+        let copy = |json: &str| -> HiveCopy { serde_json::from_str(json).unwrap() };
+        let web = vec![Permission::WebTools];
+        let default = Permission::DEFAULT.to_vec();
+
+        assert_eq!(
+            copy(r#"{"permissions":["web_tools"]}"#).before_read("atlas"),
+            web
+        );
+        assert_eq!(copy("{}").before_read("atlas"), default);
+
+        let pending = copy(r#"{"permissions":["web_tools"],"import_pending":true}"#);
+        assert_eq!(pending.after_read("atlas", &record("{}")), web);
+        assert_eq!(
+            pending.after_read("atlas", &record(r#"{"permissions":[]}"#)),
+            Vec::new()
+        );
+        let settled = copy(r#"{"permissions":["web_tools"]}"#);
+        assert_eq!(settled.after_read("atlas", &record("{}")), default);
+    }
+
     /// Fields a newer controller adds must not cost an older hive the ones it
     /// knows; a deleted key is unset.
     #[cfg(feature = "agent-permissions")]