agentmux_srv/backend/storage/
work_claims.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3//
4//! Work claims: an agent saying "I'm working on this" about a path, branch
5//! or topic, so others see it in `WhoIsWorkingOn` and in overlap notes
6//! (docs/specs/SPEC_AGENT_OVERLAP_AWARENESS_2026_10_10.md §3.4).
7//!
8//! A claim never blocks anyone: it is information with an expiry. It lives
9//! in the always-global identity store, like the work queue, so every
10//! channel on this computer sees every claim. It has its own table rather
11//! than being a work-queue row: a queue item whose lease lapses goes back to
12//! `open`, where `WorkClaim` would hand it to another agent as an
13//! instruction. An expired claim is simply gone: it is filtered out on every
14//! read and deleted on the next write.
15
16use rusqlite::{params, OptionalExtension, Row};
17use serde::{Deserialize, Serialize};
18
19use super::store::Store;
20use super::StoreError;
21
22/// One live claim. `repo`, `path`, `absolute_path`, `branch` and `topic` are
23/// the claimed target, resolved the way `WhoIsWorkingOn` resolves a question
24/// (`work_facts::matching::Target`); any of them may be absent.
25#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
26pub struct WorkClaim {
27    pub id: String,
28    pub agent: String,
29    /// The claimer's UID, when its request carried one; empty otherwise.
30    #[serde(default)]
31    pub agent_uid: String,
32    pub channel: String,
33    pub repo: Option<String>,
34    /// Repository-relative; `""` is the whole repository.
35    pub path: Option<String>,
36    pub absolute_path: Option<String>,
37    pub branch: Option<String>,
38    pub topic: Option<String>,
39    pub note: String,
40    pub created_at: i64,
41    pub expires_at: i64,
42}
43
44/// The table, created on every open of the identity store
45/// (`Store::open_identity_store`). A new table with no ALTER, so the store's
46/// schema version is deliberately not bumped: this runs on every open anyway,
47/// and a bump would make every older installed build refuse the shared store.
48pub(crate) const SCHEMA: &str = "
49    CREATE TABLE IF NOT EXISTS db_work_claims (
50        id            TEXT PRIMARY KEY,
51        agent         TEXT NOT NULL,
52        agent_uid     TEXT NOT NULL DEFAULT '',
53        channel       TEXT NOT NULL DEFAULT '',
54        repo          TEXT,
55        path          TEXT,
56        absolute_path TEXT,
57        branch        TEXT,
58        topic         TEXT,
59        note          TEXT NOT NULL DEFAULT '',
60        created_at    INTEGER NOT NULL,
61        expires_at    INTEGER NOT NULL
62    );
63    CREATE INDEX IF NOT EXISTS idx_ids_work_claims_expires ON db_work_claims(expires_at);";
64
65const COLS: &str =
66    "id, agent, agent_uid, channel, repo, path, absolute_path, branch, topic, note, created_at, expires_at";
67
68fn row_to_claim(row: &Row) -> rusqlite::Result<WorkClaim> {
69    Ok(WorkClaim {
70        id: row.get("id")?,
71        agent: row.get("agent")?,
72        agent_uid: row.get("agent_uid")?,
73        channel: row.get("channel")?,
74        repo: row.get("repo")?,
75        path: row.get("path")?,
76        absolute_path: row.get("absolute_path")?,
77        branch: row.get("branch")?,
78        topic: row.get("topic")?,
79        note: row.get("note")?,
80        created_at: row.get("created_at")?,
81        expires_at: row.get("expires_at")?,
82    })
83}
84
85/// The holder test, as the work queue's: by UID when the row and the caller
86/// both have one, else by name (case-insensitively). `?1` name, `?2` UID.
87const HOLDER: &str = "((agent_uid <> '' AND ?2 <> '' AND agent_uid = ?2) \
88                      OR ((agent_uid = '' OR ?2 = '') AND agent = ?1 COLLATE NOCASE))";
89
90impl Store {
91    /// Record `claim`, or renew the holder's existing claim on the same
92    /// target: its id and `created_at` stay, its note and expiry are
93    /// replaced. Expired claims are deleted first. Returns the stored claim.
94    pub fn work_claim_put(&self, claim: &WorkClaim, now_ms: i64) -> Result<WorkClaim, StoreError> {
95        let conn = self.conn().lock().unwrap();
96        conn.execute("DELETE FROM db_work_claims WHERE expires_at <= ?1", params![now_ms])?;
97        // `IS` compares NULL to NULL as equal: an absent part of the target
98        // matches an absent part.
99        let sql = format!(
100            "SELECT id, created_at FROM db_work_claims
101              WHERE {HOLDER}
102                AND repo IS ?3 AND path IS ?4 AND absolute_path IS ?5 AND branch IS ?6
103                AND topic IS ?7 COLLATE NOCASE
104              LIMIT 1"
105        );
106        let existing: Option<(String, i64)> = conn
107            .query_row(
108                &sql,
109                params![
110                    claim.agent,
111                    claim.agent_uid,
112                    claim.repo,
113                    claim.path,
114                    claim.absolute_path,
115                    claim.branch,
116                    claim.topic
117                ],
118                |r| Ok((r.get(0)?, r.get(1)?)),
119            )
120            .optional()?;
121        let mut stored = claim.clone();
122        if let Some((id, created_at)) = existing {
123            stored.id = id;
124            stored.created_at = created_at;
125        }
126        conn.execute(
127            &format!("INSERT OR REPLACE INTO db_work_claims ({COLS}) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12)"),
128            params![
129                stored.id,
130                stored.agent,
131                stored.agent_uid,
132                stored.channel,
133                stored.repo,
134                stored.path,
135                stored.absolute_path,
136                stored.branch,
137                stored.topic,
138                stored.note,
139                stored.created_at,
140                stored.expires_at
141            ],
142        )?;
143        Ok(stored)
144    }
145
146    /// Every unexpired claim, oldest first.
147    pub fn work_claims_live(&self, now_ms: i64) -> Result<Vec<WorkClaim>, StoreError> {
148        let conn = self.conn().lock().unwrap();
149        let mut stmt =
150            conn.prepare(&format!("SELECT {COLS} FROM db_work_claims WHERE expires_at > ?1 ORDER BY created_at, id"))?;
151        let rows = stmt.query_map(params![now_ms], row_to_claim)?;
152        Ok(rows.collect::<Result<Vec<_>, _>>()?)
153    }
154
155    /// Release the holder's claims: the one with `id`, or all of them when
156    /// `id` is `None`. Returns what was released; another agent's claim is
157    /// never touched.
158    pub fn work_claim_release(
159        &self,
160        agent: &str,
161        agent_uid: &str,
162        id: Option<&str>,
163        now_ms: i64,
164    ) -> Result<Vec<WorkClaim>, StoreError> {
165        let conn = self.conn().lock().unwrap();
166        let sql = format!(
167            "DELETE FROM db_work_claims
168              WHERE {HOLDER} AND (?3 IS NULL OR id = ?3) AND expires_at > ?4
169              RETURNING {COLS}"
170        );
171        let mut stmt = conn.prepare(&sql)?;
172        let rows = stmt.query_map(params![agent, agent_uid, id, now_ms], row_to_claim)?;
173        let released = rows.collect::<Result<Vec<_>, _>>()?;
174        drop(stmt);
175        conn.execute("DELETE FROM db_work_claims WHERE expires_at <= ?1", params![now_ms])?;
176        Ok(released)
177    }
178}
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183
184    const NOW: i64 = 1_000_000_000;
185    const MIN: i64 = 60_000;
186
187    fn store() -> (tempfile::TempDir, Store) {
188        let dir = tempfile::tempdir().unwrap();
189        let s = Store::open_identity_store(&dir.path().join("identity-store.db")).unwrap();
190        (dir, s)
191    }
192
193    fn claim(id: &str, agent: &str, path: &str) -> WorkClaim {
194        WorkClaim {
195            id: id.into(),
196            agent: agent.into(),
197            channel: "stable".into(),
198            repo: Some("o/r".into()),
199            path: Some(path.into()),
200            note: "presence work".into(),
201            created_at: NOW,
202            expires_at: NOW + 120 * MIN,
203            ..Default::default()
204        }
205    }
206
207    #[test]
208    fn a_claim_is_listed_until_it_expires() {
209        let (_d, s) = store();
210        s.work_claim_put(&claim("c1", "Agent5", "crates/srv/src/muxbus"), NOW).unwrap();
211        assert_eq!(s.work_claims_live(NOW).unwrap().len(), 1);
212        assert_eq!(s.work_claims_live(NOW + 120 * MIN - 1).unwrap().len(), 1);
213        assert!(s.work_claims_live(NOW + 120 * MIN).unwrap().is_empty(), "expired");
214    }
215
216    #[test]
217    fn claiming_the_same_target_again_renews_it_and_keeps_its_id() {
218        let (_d, s) = store();
219        s.work_claim_put(&claim("c1", "Agent5", "a.rs"), NOW).unwrap();
220        let again = WorkClaim {
221            note: "still on it".into(),
222            created_at: NOW + 30 * MIN,
223            expires_at: NOW + 200 * MIN,
224            ..claim("c2", "agent5", "a.rs")
225        };
226        let stored = s.work_claim_put(&again, NOW + 30 * MIN).unwrap();
227        assert_eq!((stored.id.as_str(), stored.created_at), ("c1", NOW));
228        let live = s.work_claims_live(NOW + 30 * MIN).unwrap();
229        assert_eq!(live.len(), 1);
230        assert_eq!((live[0].note.as_str(), live[0].expires_at), ("still on it", NOW + 200 * MIN));
231
232        // Another target, or another agent on the same one, is a second claim.
233        s.work_claim_put(&claim("c3", "Agent5", "b.rs"), NOW).unwrap();
234        s.work_claim_put(&claim("c4", "Agent4", "a.rs"), NOW).unwrap();
235        assert_eq!(s.work_claims_live(NOW).unwrap().len(), 3);
236    }
237
238    #[test]
239    fn only_the_holder_releases_one_claim_or_all_of_its_own() {
240        let (_d, s) = store();
241        s.work_claim_put(&claim("c1", "Agent5", "a.rs"), NOW).unwrap();
242        s.work_claim_put(&claim("c2", "Agent5", "b.rs"), NOW).unwrap();
243        s.work_claim_put(&claim("c3", "Agent4", "a.rs"), NOW).unwrap();
244
245        assert!(s.work_claim_release("Agent4", "", Some("c1"), NOW).unwrap().is_empty(), "not Agent4's");
246        let one = s.work_claim_release("agent5", "", Some("c1"), NOW).unwrap();
247        assert_eq!(one.iter().map(|c| c.id.as_str()).collect::<Vec<_>>(), vec!["c1"]);
248        let rest = s.work_claim_release("Agent5", "", None, NOW).unwrap();
249        assert_eq!(rest.iter().map(|c| c.id.as_str()).collect::<Vec<_>>(), vec!["c2"]);
250        let left = s.work_claims_live(NOW).unwrap();
251        assert_eq!(left.iter().map(|c| c.agent.as_str()).collect::<Vec<_>>(), vec!["Agent4"]);
252    }
253
254    #[test]
255    fn a_uid_decides_when_both_sides_have_one() {
256        let (_d, s) = store();
257        let mine = WorkClaim { agent_uid: "u1".into(), ..claim("c1", "Agent5", "a.rs") };
258        s.work_claim_put(&mine, NOW).unwrap();
259        assert!(s.work_claim_release("Agent5", "u2", None, NOW).unwrap().is_empty(), "same name, another UID");
260        assert_eq!(s.work_claim_release("Agent5", "u1", None, NOW).unwrap().len(), 1);
261    }
262}
agentmux_srv/backend/storage/work_claims.rs

agentmux_srv/backend/storage/
work_claims.rs
