Cargo.lock

@@ -399,6 +399,7 @@ dependencies = [
  "tower-http 0.5.2",
  "tracing",
  "tracing-subscriber",
+ "uuid",
  "zstd",
 ]
 
@@ -2184,6 +2185,17 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
+[[package]]
+name = "uuid"
+version = "1.23.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ea5fab0d6c3c01ae70085a09cb03d4c7a1d6314e2b3e075392783396d724ca0a"
+dependencies = [
+ "getrandom 0.4.2",
+ "js-sys",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "valuable"
 version = "0.1.1"

claw-store/Cargo.toml

@@ -55,6 +55,9 @@ axum       = { version = "0.7", features = ["macros"] }
 tower-http = { version = "0.5", features = ["cors", "fs"] }
 tokio-stream = "0.1"
 serde_json = "1"
+# v1 — session IDs (Phase 9 S1). v4 random hex; no persistence
+# concerns beyond "opaque URL-safe id".
+uuid = { version = "1", features = ["v4"] }
 # v0.2.0 — flock(2) wrapper for atomic+locked manifest writes
 # (manifest.rs). Already a transitive dep; declaring it directly
 # makes the call site obvious.

claw-store/src/lib.rs

@@ -28,6 +28,7 @@ pub mod manifest;
 pub mod restore;
 pub mod serve;
 pub mod serve_v2;
+pub mod sessions;
 pub mod snapshot;
 pub mod sync;
 pub mod zfs;

claw-store/src/main.rs

@@ -9,6 +9,7 @@ mod manifest;
 mod restore;
 mod serve;
 mod serve_v2;
+mod sessions;
 mod snapshot;
 mod sync;
 mod zfs;

claw-store/src/serve_v2.rs

@@ -31,6 +31,7 @@ use crate::cluster::rpc::{
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
 use crate::config::{Config, PeerEntry, TokenEntry};
+use crate::sessions::{LeasedTag, Session, SessionStore};
 
 /// Aggregator runtime: one QuicClient, one peer list, one identity.
 ///
@@ -63,6 +64,10 @@ pub struct V2State {
     /// entries here can be admin (no `namespace`) or scoped (must
     /// match `<namespace>:*` on tag names). First match wins.
     pub token_entries: Vec<TokenEntry>,
+    /// TTL-scoped session store (Phase 9 S1-S3). Sessions own tag
+    /// leases and are reaped by a background sweeper when their
+    /// `expires_at_unix` passes without a `renew` or `commit`.
+    pub sessions: SessionStore,
 }
 
 impl V2State {
@@ -79,6 +84,13 @@ impl V2State {
             .map_err(|e| anyhow::anyhow!("loading fleet-CA identity: {e}"))?;
         let client = QuicClient::new("0.0.0.0:0".parse()?, identity)
             .map_err(|e| anyhow::anyhow!("building QUIC client: {e}"))?;
+        // Session store persists next to warm.projects_path — the
+        // aggregator already has a writable directory there. Fall
+        // back to /tmp so tests don't need config.warm set.
+        let sessions_path = std::path::PathBuf::from(&cfg.warm.projects_path)
+            .join("aggregator-sessions.json");
+        let sessions = SessionStore::load(sessions_path)
+            .map_err(|e| anyhow::anyhow!("loading session store: {e}"))?;
         Ok(Self {
             aggregator_name: cfg.node.name.clone(),
             peers: cluster.peers.clone(),
@@ -90,6 +102,7 @@ impl V2State {
                 .as_ref()
                 .map(|a| a.tokens.clone())
                 .unwrap_or_default(),
+            sessions,
         })
     }
 
@@ -495,6 +508,7 @@ impl V2State {
             default_rpc_port_offset: self.default_rpc_port_offset,
             api_token: self.api_token.clone(),
             token_entries: self.token_entries.clone(),
+            sessions: self.sessions.clone(),
         }
     }
 }
@@ -778,13 +792,275 @@ async fn v2_auth(
     next.run(request).await
 }
 
+// ── session endpoints (Phase 9 S1-S3) ───────────────────────────
+
+#[derive(Deserialize)]
+pub struct CreateSessionBody {
+    /// Absolute-from-now TTL in seconds. Callers should send the
+    /// same value on `renew` to reset the clock; there's no additive
+    /// mode. Cap enforced at 24h to protect the sweeper from a
+    /// caller pinning tags forever.
+    pub ttl_secs: u64,
+    /// Optional human note for operator/debugging visibility.
+    #[serde(default)]
+    pub note: Option<String>,
+}
+
+#[derive(Deserialize)]
+pub struct PinInSessionBody {
+    pub tag: String,
+    pub blob_id: String,
+}
+
+#[derive(Deserialize)]
+pub struct RenewBody {
+    pub ttl_secs: u64,
+}
+
+/// Max TTL a session may live before requiring renewal or commit.
+/// Prevents a caller from creating an effectively-immortal session
+/// with, say, ttl_secs = u64::MAX.
+const MAX_SESSION_TTL_SECS: u64 = 24 * 60 * 60;
+
+/// Namespace filter that pairs a caller with the sessions they may
+/// see or modify. Admin/Open see everything; namespaced callers only
+/// see their own namespace's sessions.
+fn caller_ns(caller: &AuthedCaller) -> Option<String> {
+    match caller {
+        AuthedCaller::Admin | AuthedCaller::Open => None,
+        AuthedCaller::Namespaced { namespace } => Some(namespace.clone()),
+    }
+}
+
+/// True iff `caller` may inspect / mutate `session`.
+fn caller_may_access(caller: &AuthedCaller, session: &Session) -> bool {
+    match caller {
+        AuthedCaller::Admin | AuthedCaller::Open => true,
+        AuthedCaller::Namespaced { namespace } => {
+            session.namespace.as_deref() == Some(namespace.as_str())
+        }
+    }
+}
+
+async fn handle_create_session(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<CreateSessionBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    if body.ttl_secs == 0 || body.ttl_secs > MAX_SESSION_TTL_SECS {
+        return Err((
+            StatusCode::BAD_REQUEST,
+            format!("ttl_secs must be 1..={MAX_SESSION_TTL_SECS}"),
+        ));
+    }
+    let ns = caller_ns(&caller);
+    let sess = s
+        .sessions
+        .create(ns, body.ttl_secs, body.note)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(sess))
+}
+
+async fn handle_list_sessions(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Json<Vec<Session>> {
+    // caller_ns == None means admin/open → no filter → see all.
+    let filter = caller_ns(&caller);
+    let list = s.sessions.list(filter.as_deref()).await;
+    Json(list)
+}
+
+async fn handle_get_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Session>, StatusCode> {
+    let sess = s.sessions.get(&id).await.ok_or(StatusCode::NOT_FOUND)?;
+    if !caller_may_access(&caller, &sess) {
+        // Preserve the "does it exist" secret from other tenants —
+        // NOT_FOUND, not FORBIDDEN. Same trick the wizard-side
+        // publish-approval endpoint uses.
+        return Err(StatusCode::NOT_FOUND);
+    }
+    Ok(Json(sess))
+}
+
+/// Pin a tag *within* a session: fan the pin out to peers, then
+/// record it in the session so a later commit/cancel/expiry can
+/// reverse it. Same tag scope rule as bare `POST /tags/:name` —
+/// namespaced callers must stay in their namespace.
+async fn handle_session_pin(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<PinInSessionBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    check_tag_scope(&caller, &body.tag)?;
+    let value = decode_blob_id(&body.blob_id)
+        .ok_or_else(|| (StatusCode::BAD_REQUEST, "blob_id must be 64-char hex".into()))?;
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    if sess.committed {
+        return Err((
+            StatusCode::CONFLICT,
+            "cannot attach leases to a committed session".into(),
+        ));
+    }
+    // Fan out the pin BEFORE attaching to the session — if the fleet
+    // rejects it, we don't record a phantom lease that can never be
+    // unpinned. Same shape as the standalone put-tag path.
+    let peer_results = fanout_put_tag(&s, &body.tag, &value).await;
+    if !peer_results.iter().all(|r| r.ok) {
+        // Return the per-peer error detail so the caller can act.
+        let err = serde_json::to_string(&peer_results).unwrap_or_default();
+        return Err((StatusCode::BAD_GATEWAY, err));
+    }
+    let lease = LeasedTag {
+        tag: body.tag,
+        blob_id_hex: body.blob_id,
+        pinned_at_unix: now_unix_u64(),
+    };
+    let updated = s
+        .sessions
+        .attach_lease(&id, lease)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_renew_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<RenewBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    if body.ttl_secs == 0 || body.ttl_secs > MAX_SESSION_TTL_SECS {
+        return Err((
+            StatusCode::BAD_REQUEST,
+            format!("ttl_secs must be 1..={MAX_SESSION_TTL_SECS}"),
+        ));
+    }
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    let updated = s
+        .sessions
+        .renew(&id, body.ttl_secs)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_commit_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    let updated = s
+        .sessions
+        .commit(&id)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_delete_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Vec<PeerResult>>, StatusCode> {
+    let sess = s.sessions.get(&id).await.ok_or(StatusCode::NOT_FOUND)?;
+    if !caller_may_access(&caller, &sess) {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    // Unpin every tag on every peer FIRST — if we removed the session
+    // first, a mid-deletion crash would strand tags on the fleet.
+    let mut all_results = Vec::new();
+    for lease in &sess.leases {
+        let r = fanout_delete_tag(&s, &lease.tag).await;
+        all_results.extend(r);
+    }
+    let _ = s.sessions.remove(&id).await;
+    Ok(Json(all_results))
+}
+
+fn now_unix_u64() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+/// Reap callback for the sessions sweeper. Owns the actual fan-out.
+/// Kept a free function so `sessions::spawn_sweeper` doesn't have
+/// to know about `V2State` or its RPC client.
+async fn reap_expired(state: Arc<V2State>, sess: Session) {
+    for lease in &sess.leases {
+        let results = fanout_delete_tag(&state, &lease.tag).await;
+        // Log per-peer failures; the sweeper is best-effort. A
+        // failing peer will be retried on the next lease that
+        // touches it because the tag stays in the store until
+        // successfully removed everywhere (well, the sweeper drops
+        // the session either way — this is a known tradeoff:
+        // durability of unpin vs. never-ending sweeper retries).
+        for r in results {
+            if !r.ok {
+                eprintln!(
+                    "sweeper: session {} tag {} peer {} unpin failed: {}",
+                    sess.id, lease.tag, r.peer,
+                    r.error.as_deref().unwrap_or("<no error message>")
+                );
+            }
+        }
+    }
+}
+
 // ── route registration ──────────────────────────────────────────
 
 /// Assemble the aggregator router with state and middleware baked in.
 /// The auth middleware needs the concrete `Arc<V2State>` at layer time
 /// (so it can read `api_token`), which is why this returns a fully-
 /// stated `Router<()>` instead of a state-generic router.
+///
+/// Also spawns the background TTL sweeper (Phase 9 S1). The task is
+/// detached — its lifetime is the process lifetime.
 pub fn build(state: Arc<V2State>) -> Router {
+    // Spawn the sweeper. 15s tick is a reasonable balance: quick
+    // enough that a mid-wizard-close cleanup feels prompt, slow
+    // enough that idle aggregators aren't burning cycles.
+    {
+        let state_for_sweeper = state.clone();
+        let store = state.sessions.clone();
+        crate::sessions::spawn_sweeper(
+            store,
+            Duration::from_secs(15),
+            move |sess| {
+                let state = state_for_sweeper.clone();
+                async move { reap_expired(state, sess).await }
+            },
+        );
+    }
     Router::new()
         .route("/api/v2/fleet", get(handle_fleet))
         .route("/api/v2/node/:name/status", get(handle_node_status))
@@ -798,6 +1074,17 @@ pub fn build(state: Arc<V2State>) -> Router {
             "/api/v2/tags/:name",
             post(handle_put_tag).delete(handle_delete_tag),
         )
+        .route(
+            "/api/v2/sessions",
+            post(handle_create_session).get(handle_list_sessions),
+        )
+        .route(
+            "/api/v2/sessions/:id",
+            get(handle_get_session).delete(handle_delete_session),
+        )
+        .route("/api/v2/sessions/:id/pin", post(handle_session_pin))
+        .route("/api/v2/sessions/:id/renew", post(handle_renew_session))
+        .route("/api/v2/sessions/:id/commit", post(handle_commit_session))
         .route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
         .with_state(state)
 }

claw-store/src/sessions.rs

@@ -0,0 +1,266 @@
+//! Aggregator-side session + lease store (Phase 9 S1-S3).
+//!
+//! A **session** is a TTL-bounded container that owns one or more
+//! **tag leases**. Callers create a session, attach pins to it, and
+//! either `commit` (make the tags permanent, drop tracking) or let
+//! the session `expire` / `DELETE` it (unpin every tag on every peer).
+//!
+//! This solves the wizard-cancel problem: the client picks a repo,
+//! we mint a session + pin under it, and if the browser closes we
+//! reap automatically. No orphaned fleet-wide state.
+//!
+//! ## Persistence
+//!
+//! One JSON file at `path`. Rewritten on every mutation (small store,
+//! aggregator only, correctness > throughput). Survives aggregator
+//! restart so mid-flight leases don't leak.
+//!
+//! ## Concurrency
+//!
+//! One `RwLock<HashMap<SessionId, Session>>`. Handlers take the write
+//! lock briefly (mutation + `save()`), the sweeper takes it for the
+//! per-tick scan + reap. Peer fan-out RPCs happen **outside** the
+//! lock so a slow peer never blocks other API calls.
+//!
+//! ## Auth
+//!
+//! Every session carries the caller's `namespace` at creation. The
+//! aggregator enforces "same-namespace only" access on every session
+//! endpoint. Admin callers see all sessions.
+
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+/// Opaque session id — UUID v4 hex, stable in the URL. Public so
+/// HTTP handlers can parse from path params.
+pub type SessionId = String;
+
+/// A single tag that a session owns. On reap, the aggregator issues
+/// a fleet-wide `DELETE /api/v2/tags/:name` for each of these.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LeasedTag {
+    pub tag: String,
+    /// 64-char lowercase hex — the blob the tag was pinned to. Kept
+    /// so listing endpoints can show what the session is holding.
+    pub blob_id_hex: String,
+    /// When the pin actually landed on the fleet, unix seconds. For
+    /// operator inspection; not consulted for reap decisions.
+    pub pinned_at_unix: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Session {
+    pub id: SessionId,
+    /// Namespace the session belongs to. `None` means an admin-owned
+    /// session (created via the root `api_token`). Enforced by the
+    /// HTTP layer, not this module.
+    pub namespace: Option<String>,
+    pub created_at_unix: u64,
+    /// When the sweeper will reap this session (unix seconds).
+    /// Extended by `renew`; frozen by `commit`.
+    pub expires_at_unix: u64,
+    pub leases: Vec<LeasedTag>,
+    /// Optional operator note, e.g. `"research-wizard draft"`.
+    #[serde(default)]
+    pub note: Option<String>,
+    /// After `commit`, the session sticks around for a short grace
+    /// window so the UI can still show "committed" state, but the
+    /// sweeper stops treating expiry as a reap trigger. Reapable
+    /// only via explicit `DELETE`.
+    #[serde(default)]
+    pub committed: bool,
+}
+
+/// Handle to the on-disk + in-memory session store.
+#[derive(Clone)]
+pub struct SessionStore {
+    inner: Arc<RwLock<HashMap<SessionId, Session>>>,
+    path: PathBuf,
+}
+
+impl SessionStore {
+    /// Load from disk (empty map if the file doesn't exist). Called
+    /// once at aggregator boot; concurrent handlers share the returned
+    /// clone via `Arc`.
+    pub fn load(path: PathBuf) -> Result<Self> {
+        let map: HashMap<SessionId, Session> = if path.exists() {
+            let bytes = std::fs::read(&path)
+                .with_context(|| format!("read sessions store at {}", path.display()))?;
+            serde_json::from_slice(&bytes).context("parse sessions store")?
+        } else {
+            HashMap::new()
+        };
+        Ok(Self {
+            inner: Arc::new(RwLock::new(map)),
+            path,
+        })
+    }
+
+    /// Rewrite the whole store to disk. Called under the write lock.
+    /// Small store, correctness > throughput.
+    fn save_locked(&self, map: &HashMap<SessionId, Session>) -> Result<()> {
+        if let Some(parent) = self.path.parent() {
+            std::fs::create_dir_all(parent).ok();
+        }
+        let tmp = self.path.with_extension("json.tmp");
+        std::fs::write(&tmp, serde_json::to_vec_pretty(map)?)?;
+        std::fs::rename(&tmp, &self.path)?;
+        Ok(())
+    }
+
+    pub async fn create(
+        &self,
+        namespace: Option<String>,
+        ttl_secs: u64,
+        note: Option<String>,
+    ) -> Result<Session> {
+        let now = now_unix();
+        let sess = Session {
+            id: Uuid::new_v4().simple().to_string(),
+            namespace,
+            created_at_unix: now,
+            expires_at_unix: now.saturating_add(ttl_secs),
+            leases: Vec::new(),
+            note,
+            committed: false,
+        };
+        let mut w = self.inner.write().await;
+        w.insert(sess.id.clone(), sess.clone());
+        self.save_locked(&w)?;
+        Ok(sess)
+    }
+
+    pub async fn get(&self, id: &str) -> Option<Session> {
+        self.inner.read().await.get(id).cloned()
+    }
+
+    /// List all sessions visible to a caller. `None` namespace ==
+    /// admin (sees everything); `Some(ns)` filters to sessions owned
+    /// by that namespace.
+    pub async fn list(&self, namespace_filter: Option<&str>) -> Vec<Session> {
+        let r = self.inner.read().await;
+        let mut out: Vec<Session> = r
+            .values()
+            .filter(|s| match namespace_filter {
+                None => true,
+                Some(ns) => s.namespace.as_deref() == Some(ns),
+            })
+            .cloned()
+            .collect();
+        // Newest first — mirrors the pattern in dashboards.
+        out.sort_by(|a, b| b.created_at_unix.cmp(&a.created_at_unix));
+        out
+    }
+
+    /// Attach a tag lease to an existing session. Caller must have
+    /// already fanned the actual pin out to peers. This just records
+    /// the tag so the sweeper can reap it on expiry.
+    pub async fn attach_lease(&self, id: &str, lease: LeasedTag) -> Result<Session> {
+        let mut w = self.inner.write().await;
+        let s = w
+            .get_mut(id)
+            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
+        // Idempotent: dedupe by tag name so retries don't double-record.
+        if !s.leases.iter().any(|l| l.tag == lease.tag) {
+            s.leases.push(lease);
+        }
+        let out = s.clone();
+        self.save_locked(&w)?;
+        Ok(out)
+    }
+
+    /// Extend expiry. `ttl_secs` is absolute-from-now, not additive,
+    /// so heartbeats stay idempotent — sending "1 hour" repeatedly
+    /// pins expiry to `now + 1h` no matter how many arrive.
+    pub async fn renew(&self, id: &str, ttl_secs: u64) -> Result<Session> {
+        let mut w = self.inner.write().await;
+        let s = w
+            .get_mut(id)
+            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
+        s.expires_at_unix = now_unix().saturating_add(ttl_secs);
+        let out = s.clone();
+        self.save_locked(&w)?;
+        Ok(out)
+    }
+
+    /// Freeze the session's tags: mark committed, stop reaping on
+    /// expiry, but keep the record for a while so the UI can show
+    /// history. Explicit `DELETE` still works.
+    pub async fn commit(&self, id: &str) -> Result<Session> {
+        let mut w = self.inner.write().await;
+        let s = w
+            .get_mut(id)
+            .ok_or_else(|| anyhow::anyhow!("session {id} not found"))?;
+        s.committed = true;
+        let out = s.clone();
+        self.save_locked(&w)?;
+        Ok(out)
+    }
+
+    /// Remove a session from the store and return the tags that
+    /// need to be unpinned across the fleet. Callers must actually
+    /// issue the unpin fan-out — this module doesn't dial peers.
+    pub async fn remove(&self, id: &str) -> Result<Option<Session>> {
+        let mut w = self.inner.write().await;
+        let removed = w.remove(id);
+        if removed.is_some() {
+            self.save_locked(&w)?;
+        }
+        Ok(removed)
+    }
+
+    /// Snapshot the currently-expired sessions (used by the sweeper).
+    /// Split into snapshot-then-reap so the peer RPCs happen outside
+    /// the store's write lock.
+    pub async fn snapshot_expired(&self) -> Vec<Session> {
+        let now = now_unix();
+        let r = self.inner.read().await;
+        r.values()
+            .filter(|s| !s.committed && s.expires_at_unix <= now)
+            .cloned()
+            .collect()
+    }
+}
+
+fn now_unix() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+/// Spawn a background task that scans for expired sessions every
+/// `tick`, calls `reap` for each expired one (which is responsible
+/// for the actual per-peer unpin), then removes them from the store.
+/// The reap callback owns fan-out so this module stays free of
+/// clawstor RPC + config types.
+pub fn spawn_sweeper<F, Fut>(
+    store: SessionStore,
+    tick: Duration,
+    reap: F,
+) -> tokio::task::JoinHandle<()>
+where
+    F: Fn(Session) -> Fut + Send + Sync + 'static,
+    Fut: std::future::Future<Output = ()> + Send + 'static,
+{
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(tick);
+        // Skip the immediate first tick — nothing to reap at boot.
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            let expired = store.snapshot_expired().await;
+            for sess in expired {
+                let id = sess.id.clone();
+                reap(sess).await;
+                let _ = store.remove(&id).await;
+            }
+        }
+    })
+}