pushkin_core/
board.rs

1//! Peer board (spec §9): persisted, restart-surviving coordination in
2//! `SQLite`. Every operation is scoped by a run ID so concurrent runs never
3//! intersect. Verbs: register / status / peers / broadcast / send / read
4//! (auto-cursor per recipient), plus claim / release on paths — the
5//! pre-write gate consults claims to deny cross-agent edits. Embedded
6//! `SQLite` only (spec §10); schema is a versioned, append-friendly step.
7
8use globset::Glob;
9use rusqlite::{params, Connection};
10use std::path::Path;
11use thiserror::Error;
12
13#[derive(Debug, Error)]
14pub enum BoardError {
15    #[error("board storage error: {0}")]
16    Storage(#[from] rusqlite::Error),
17    #[error("invalid claim glob '{glob}': {message}")]
18    BadGlob { glob: String, message: String },
19}
20
21/// One agent's registration row.
22#[derive(Debug)]
23pub struct Peer {
24    pub agent: String,
25    pub status: Option<String>,
26}
27
28/// One message delivered to a reader.
29#[derive(Debug)]
30pub struct Message {
31    pub id: i64,
32    pub from_agent: String,
33    pub body: String,
34}
35
36/// An active path claim.
37#[derive(Debug)]
38pub struct Claim {
39    pub agent: String,
40    pub path_glob: String,
41}
42
43pub struct Board {
44    conn: Connection,
45    run_id: String,
46}
47
48impl Board {
49    /// Opens (creating as needed) the board for one run.
50    ///
51    /// # Errors
52    /// Returns `BoardError` when the database cannot be opened or migrated.
53    pub fn open(path: &Path, run_id: &str) -> Result<Self, BoardError> {
54        let conn = Connection::open(path)?;
55        conn.pragma_update(None, "journal_mode", "WAL")?;
56        // Versioned, append-only migration steps (AGENTS.md SQLite rule).
57        // Timestamps: epoch ms UTC, one convention for every board table.
58        conn.execute_batch(
59            "CREATE TABLE IF NOT EXISTS board_schema (version INTEGER NOT NULL);
60             CREATE TABLE IF NOT EXISTS agents (
61               run_id TEXT NOT NULL,
62               agent  TEXT NOT NULL,
63               status TEXT,
64               registered_at_ms INTEGER NOT NULL,
65               PRIMARY KEY (run_id, agent)
66             );
67             CREATE TABLE IF NOT EXISTS messages (
68               id INTEGER PRIMARY KEY AUTOINCREMENT,
69               run_id TEXT NOT NULL,
70               from_agent TEXT NOT NULL,
71               to_agent TEXT,
72               body TEXT NOT NULL,
73               sent_at_ms INTEGER NOT NULL
74             );
75             CREATE TABLE IF NOT EXISTS cursors (
76               run_id TEXT NOT NULL,
77               agent  TEXT NOT NULL,
78               last_read_id INTEGER NOT NULL,
79               PRIMARY KEY (run_id, agent)
80             );
81             CREATE TABLE IF NOT EXISTS claims (
82               run_id TEXT NOT NULL,
83               agent  TEXT NOT NULL,
84               path_glob TEXT NOT NULL,
85               claimed_at_ms INTEGER NOT NULL,
86               PRIMARY KEY (run_id, path_glob)
87             );",
88        )?;
89        let versions: i64 =
90            conn.query_row("SELECT COUNT(*) FROM board_schema", [], |row| row.get(0))?;
91        if versions == 0 {
92            conn.execute("INSERT INTO board_schema (version) VALUES (1)", [])?;
93        }
94        Ok(Self {
95            conn,
96            run_id: run_id.to_owned(),
97        })
98    }
99
100    /// Registers (or re-registers, idempotently) an agent in this run.
101    ///
102    /// # Errors
103    /// Returns `BoardError` on storage failure.
104    pub fn register(&self, agent: &str) -> Result<(), BoardError> {
105        self.conn.execute(
106            "INSERT INTO agents (run_id, agent, status, registered_at_ms)
107             VALUES (?1, ?2, NULL, ?3)
108             ON CONFLICT (run_id, agent) DO NOTHING",
109            params![self.run_id, agent, now_ms()],
110        )?;
111        Ok(())
112    }
113
114    /// Updates an agent's free-text status line.
115    ///
116    /// # Errors
117    /// Returns `BoardError` on storage failure.
118    pub fn set_status(&self, agent: &str, status: &str) -> Result<(), BoardError> {
119        self.register(agent)?;
120        self.conn.execute(
121            "UPDATE agents SET status = ?3 WHERE run_id = ?1 AND agent = ?2",
122            params![self.run_id, agent, status],
123        )?;
124        Ok(())
125    }
126
127    /// All agents in this run except `agent` itself.
128    ///
129    /// # Errors
130    /// Returns `BoardError` on storage failure.
131    pub fn peers(&self, agent: &str) -> Result<Vec<Peer>, BoardError> {
132        let mut statement = self.conn.prepare(
133            "SELECT agent, status FROM agents
134             WHERE run_id = ?1 AND agent != ?2 ORDER BY agent",
135        )?;
136        let peers = statement
137            .query_map(params![self.run_id, agent], |row| {
138                Ok(Peer {
139                    agent: row.get(0)?,
140                    status: row.get(1)?,
141                })
142            })?
143            .collect::<Result<Vec<_>, _>>()?;
144        Ok(peers)
145    }
146}
147
148impl Board {
149    /// Queues a message to every current peer (`to_agent` NULL = broadcast).
150    ///
151    /// # Errors
152    /// Returns `BoardError` on storage failure.
153    pub fn broadcast(&self, from_agent: &str, body: &str) -> Result<(), BoardError> {
154        self.conn.execute(
155            "INSERT INTO messages (run_id, from_agent, to_agent, body, sent_at_ms)
156             VALUES (?1, ?2, NULL, ?3, ?4)",
157            params![self.run_id, from_agent, body, now_ms()],
158        )?;
159        Ok(())
160    }
161
162    /// Queues a directed message.
163    ///
164    /// # Errors
165    /// Returns `BoardError` on storage failure.
166    pub fn send(&self, from_agent: &str, to_agent: &str, body: &str) -> Result<(), BoardError> {
167        self.conn.execute(
168            "INSERT INTO messages (run_id, from_agent, to_agent, body, sent_at_ms)
169             VALUES (?1, ?2, ?3, ?4, ?5)",
170            params![self.run_id, from_agent, to_agent, body, now_ms()],
171        )?;
172        Ok(())
173    }
174
175    /// Unread messages for `agent` (broadcasts + directed), advancing the
176    /// auto-cursor so a second read never redelivers.
177    ///
178    /// # Errors
179    /// Returns `BoardError` on storage failure.
180    pub fn read(&self, agent: &str) -> Result<Vec<Message>, BoardError> {
181        let cursor: i64 = self
182            .conn
183            .query_row(
184                "SELECT last_read_id FROM cursors WHERE run_id = ?1 AND agent = ?2",
185                params![self.run_id, agent],
186                |row| row.get(0),
187            )
188            .unwrap_or(0);
189        let mut statement = self.conn.prepare(
190            "SELECT id, from_agent, body FROM messages
191             WHERE run_id = ?1 AND id > ?2
192               AND from_agent != ?3
193               AND (to_agent IS NULL OR to_agent = ?3)
194             ORDER BY id",
195        )?;
196        let messages = statement
197            .query_map(params![self.run_id, cursor, agent], |row| {
198                Ok(Message {
199                    id: row.get(0)?,
200                    from_agent: row.get(1)?,
201                    body: row.get(2)?,
202                })
203            })?
204            .collect::<Result<Vec<_>, _>>()?;
205        if let Some(last) = messages.last() {
206            self.conn.execute(
207                "INSERT INTO cursors (run_id, agent, last_read_id) VALUES (?1, ?2, ?3)
208                 ON CONFLICT (run_id, agent) DO UPDATE SET last_read_id = ?3",
209                params![self.run_id, agent, last.id],
210            )?;
211        }
212        Ok(messages)
213    }
214
215    /// Claims a path glob for `agent`. Idempotent for the holder; a claim
216    /// held by another agent in this run is a loud conflict.
217    ///
218    /// # Errors
219    /// Returns `BoardError::BadGlob` for an invalid glob, `Storage` for
220    /// conflicts (constraint violation) and other database failures.
221    pub fn claim(&self, agent: &str, path_glob: &str) -> Result<(), BoardError> {
222        Glob::new(path_glob).map_err(|error| BoardError::BadGlob {
223            glob: path_glob.to_owned(),
224            message: error.to_string(),
225        })?;
226        self.register(agent)?;
227        let changed = self.conn.execute(
228            "INSERT INTO claims (run_id, agent, path_glob, claimed_at_ms)
229             VALUES (?1, ?2, ?3, ?4)
230             ON CONFLICT (run_id, path_glob) DO NOTHING",
231            params![self.run_id, agent, path_glob, now_ms()],
232        )?;
233        if changed == 0 {
234            let holder: String = self.conn.query_row(
235                "SELECT agent FROM claims WHERE run_id = ?1 AND path_glob = ?2",
236                params![self.run_id, path_glob],
237                |row| row.get(0),
238            )?;
239            if holder != agent {
240                return Err(BoardError::Storage(rusqlite::Error::SqliteFailure(
241                    rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT),
242                    Some(format!("path already claimed by {holder}")),
243                )));
244            }
245        }
246        Ok(())
247    }
248
249    /// Releases a claim; only the holder can release.
250    ///
251    /// # Errors
252    /// Returns `BoardError` on storage failure.
253    pub fn release(&self, agent: &str, path_glob: &str) -> Result<bool, BoardError> {
254        let changed = self.conn.execute(
255            "DELETE FROM claims WHERE run_id = ?1 AND agent = ?2 AND path_glob = ?3",
256            params![self.run_id, agent, path_glob],
257        )?;
258        Ok(changed > 0)
259    }
260
261    /// Active claims in this run.
262    ///
263    /// # Errors
264    /// Returns `BoardError` on storage failure.
265    pub fn claims(&self) -> Result<Vec<Claim>, BoardError> {
266        let mut statement = self
267            .conn
268            .prepare("SELECT agent, path_glob FROM claims WHERE run_id = ?1 ORDER BY path_glob")?;
269        let claims = statement
270            .query_map(params![self.run_id], |row| {
271                Ok(Claim {
272                    agent: row.get(0)?,
273                    path_glob: row.get(1)?,
274                })
275            })?
276            .collect::<Result<Vec<_>, _>>()?;
277        Ok(claims)
278    }
279
280    /// The holder of a claim covering `file`, when that holder is not
281    /// `agent` — i.e. the gate-side question "is this write blocked?".
282    ///
283    /// # Errors
284    /// Returns `BoardError` on storage failure.
285    pub fn blocking_holder(&self, agent: &str, file: &str) -> Result<Option<String>, BoardError> {
286        for claim in self.claims()? {
287            if claim.agent != agent && glob_matches(&claim.path_glob, file) {
288                return Ok(Some(claim.agent));
289            }
290        }
291        Ok(None)
292    }
293}
294
295/// Whether `glob` matches `file`; an invalid glob matches nothing.
296/// Public so CLI-side policy layers share one matching semantics with
297/// the board and waivers instead of growing a second glob dependency.
298#[must_use]
299pub fn glob_matches(glob: &str, file: &str) -> bool {
300    Glob::new(glob).is_ok_and(|g| g.compile_matcher().is_match(file))
301}
302
303fn now_ms() -> i64 {
304    i64::try_from(
305        std::time::SystemTime::now()
306            .duration_since(std::time::UNIX_EPOCH)
307            .map_or(0, |duration| duration.as_millis()),
308    )
309    .unwrap_or(i64::MAX)
310}
pushkin_core/board.rs

pushkin_core/
board.rs
