feat(daemon): storage::policy + PolicyChanged signal + inotify watcher

Lands plan tasks 2.6 (PolicyStore wraps load_from_dir / save_local), 2.7
(PolicyChanged D-Bus signal on the AuthForge interface), and 2.8 (notify-based
inotify watcher in main.rs that emits the signal on any change in the policy.d
directory). Bundled because watcher → emit signal → wraps PolicyStore is one
data flow.

- daemon/src/storage/{mod,policy}.rs — PolicyStore::{load,save,watch}; watch
  returns a (RecommendedWatcher, watch::Receiver) so the caller keeps the
  watcher alive.
- daemon/src/dbus.rs — adds #[zbus(signal)] policy_changed; integration test
  via p2p connection asserts the signal arrives within 2s.
- daemon/src/main.rs — spawns a task that ticks PolicyChanged on every
  rx.changed(), keyed off AUTHFORGE_POLICY_DIR env var (default
  /etc/authforge/policy.d). Watcher leaked via std::mem::forget; daemon
  lifetime = process lifetime.

Test count: 17/17 daemon (was 14) + 13/13 common.

Cargo.lock

@@ -219,6 +219,7 @@ version = "0.1.0"
 dependencies = [
  "serde",
  "serde_json",
+ "tempfile",
  "thiserror",
  "toml",
  "zvariant",

daemon/src/dbus.rs

@@ -115,6 +115,13 @@ impl AuthForge {
         let n: u32 = rand::rng().random_range(0..100_000_000);
         Ok(format!("{n:08}"))
     }
+
+    /// Emitted by main.rs whenever the inotify watcher on the policy.d/ dir
+    /// reports any change. Subscribers (GUI) reload via `GetPolicy`.
+    #[zbus(signal)]
+    pub async fn policy_changed(
+        signal_ctxt: &zbus::object_server::SignalContext<'_>,
+    ) -> zbus::Result<()>;
 }
 
 #[cfg(test)]
@@ -263,4 +270,27 @@ mod tests {
         assert_eq!(code.len(), 8);
         assert!(code.chars().all(|c| c.is_ascii_digit()));
     }
+
+    #[tokio::test]
+    async fn policy_changed_signal_fires_on_emit() {
+        use futures_util::stream::StreamExt;
+        let (server, client, _state) = p2p_pair().await;
+
+        let proxy = proxy(&client).await;
+        let mut stream = proxy.receive_signal("PolicyChanged").await.unwrap();
+
+        let iface_ref: zbus::object_server::InterfaceRef<AuthForge> = server
+            .object_server()
+            .interface("/io/dangerousthings/AuthForge")
+            .await
+            .unwrap();
+        AuthForge::policy_changed(iface_ref.signal_context())
+            .await
+            .unwrap();
+
+        let _msg = tokio::time::timeout(std::time::Duration::from_secs(2), stream.next())
+            .await
+            .expect("signal not received within 2s")
+            .expect("stream closed before signal");
+    }
 }

daemon/src/main.rs

@@ -1,10 +1,12 @@
 use anyhow::{Context, Result};
+use std::path::PathBuf;
 use std::sync::Arc;
 use tracing::{info, warn};
 
 mod dbus;
 mod polkit;
 mod state;
+mod storage;
 
 const BUS_NAME: &str = "io.dangerousthings.AuthForge";
 const OBJECT_PATH: &str = "/io/dangerousthings/AuthForge";
@@ -42,6 +44,41 @@ async fn main() -> Result<()> {
 
     info!("listening on D-Bus as {BUS_NAME} at {OBJECT_PATH}");
 
+    // Inotify watcher → PolicyChanged signal.
+    let policy_dir = std::env::var_os("AUTHFORGE_POLICY_DIR")
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from("/etc/authforge/policy.d"));
+    let policy_store = storage::policy::PolicyStore::new(policy_dir.clone());
+    match policy_store.watch() {
+        Ok((watcher, mut rx)) => {
+            let conn_for_signal = conn.clone();
+            tokio::spawn(async move {
+                while rx.changed().await.is_ok() {
+                    let iface_ref = match conn_for_signal
+                        .object_server()
+                        .interface::<_, dbus::AuthForge>(OBJECT_PATH)
+                        .await
+                    {
+                        Ok(r) => r,
+                        Err(e) => {
+                            warn!(error=%e, "no AuthForge interface; skipping signal");
+                            continue;
+                        }
+                    };
+                    if let Err(e) =
+                        dbus::AuthForge::policy_changed(iface_ref.signal_context()).await
+                    {
+                        warn!(error=%e, "PolicyChanged emit failed");
+                    }
+                }
+            });
+            // Daemon lives until SIGTERM; deliberately leak to keep watcher alive.
+            std::mem::forget(watcher);
+            info!("watching {} for policy changes", policy_dir.display());
+        }
+        Err(e) => warn!(error=%e, "failed to start policy.d watcher"),
+    }
+
     // Park forever. systemd will SIGTERM the process when stopping the unit.
     std::future::pending::<()>().await;
     Ok(())

daemon/src/storage/mod.rs

@@ -0,0 +1 @@
+pub(crate) mod policy;

daemon/src/storage/policy.rs

@@ -0,0 +1,88 @@
+use authforge_common::policy::{Policy, PolicyError};
+use notify::{RecommendedWatcher, RecursiveMode, Watcher};
+use std::path::PathBuf;
+
+pub(crate) struct PolicyStore {
+    dir: PathBuf,
+}
+
+// `save` and `load` are wired through AppState in Task 2.15. Keep them now so
+// the storage module is the single owner of policy I/O — main.rs constructs a
+// store for the watcher in 2.8.
+#[allow(dead_code)]
+impl PolicyStore {
+    pub fn new(dir: PathBuf) -> Self {
+        Self { dir }
+    }
+
+    pub fn load(&self) -> Result<Policy, PolicyError> {
+        Policy::load_from_dir(&self.dir)
+    }
+
+    pub fn save(&self, p: &Policy) -> Result<(), PolicyError> {
+        p.save_local(&self.dir)
+    }
+
+    /// Watch `self.dir` for any change. The returned receiver yields `()` each
+    /// time something changes. The caller must keep the `RecommendedWatcher`
+    /// alive — drop it to stop watching.
+    pub fn watch(&self) -> notify::Result<(RecommendedWatcher, tokio::sync::watch::Receiver<()>)> {
+        let (tx, rx) = tokio::sync::watch::channel(());
+        let mut watcher =
+            notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
+                if res.is_ok() {
+                    let _ = tx.send(());
+                }
+            })?;
+        // Safe even if the dir doesn't yet exist — daemon may install before
+        // anyone drops a config file in.
+        std::fs::create_dir_all(&self.dir).ok();
+        watcher.watch(&self.dir, RecursiveMode::NonRecursive)?;
+        Ok((watcher, rx))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use authforge_common::policy::StackPolicy;
+    use authforge_common::types::{Method, Mode};
+    use tempfile::tempdir;
+
+    #[test]
+    fn save_then_load_roundtrip() {
+        let d = tempdir().unwrap();
+        let store = PolicyStore::new(d.path().to_path_buf());
+
+        let mut p = Policy::default();
+        p.stacks.insert(
+            "sudo".to_string(),
+            StackPolicy {
+                mode: Mode::Required,
+                methods: vec![Method::Fido2],
+            },
+        );
+        store.save(&p).unwrap();
+
+        let back = store.load().unwrap();
+        assert_eq!(back, p);
+    }
+
+    #[tokio::test]
+    async fn watcher_ticks_on_file_change() {
+        let d = tempdir().unwrap();
+        let store = PolicyStore::new(d.path().to_path_buf());
+        let (_watcher, mut rx) = store.watch().unwrap();
+
+        std::fs::write(
+            d.path().join("00-base.conf"),
+            "[stacks.sudo]\nmode = \"required\"\nmethods = [\"fido2\"]\n",
+        )
+        .unwrap();
+
+        tokio::time::timeout(std::time::Duration::from_secs(2), rx.changed())
+            .await
+            .expect("no change within 2s")
+            .unwrap();
+    }
+}