Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
636 linesCodeBlameRaw
1//! The statements coordination runs, against the schema every migration
2//! leaves, in SQLite (which D1 is), with foreign keys on as D1 has them.
3
4use rusqlite::{Connection, OptionalExtension, params};
5
6use super::links::{BRANCH_SESSIONS, FAMILY, FIRST_PROMPTS, INSERT_LINK, INSERT_OVERLAP, INSERT_TOUCH, TOUCHED_BY_OTHERS, end_tree_sql, tree_sql};
7use super::sessions::{INSERT_ENTRY, MOVE_ENTRIES, RENEW_CLAIM, harness_session_sql, insert_session_sql, refresh_sql, renewal, resume_sql};
8use super::{after_minutes, expire_claims_sql, time_out_sessions_sql};
9use super::updates::deliver_sql;
10use g1t_contracts::time::rfc3339;
11use g1t_contracts::work::{claim_is_mine, session_overlaps, session_pair, subagent_harness_id};
12use std::collections::HashSet;
13
14const NOW: u64 = 1_791_800_000_000;
15
16/// Starts a subagent of `parent` the way `start_session` does, on issue 4.
17fn start_child(db: &Connection, id: &str, parent: &str, harness_id: &str, agent: &str) -> Option<String> {
18 let at = rfc3339(NOW);
19 db.query_row(
20 &insert_session_sql(),
21 params![
22 id, "rep_1", 4, None::<u32>, "main", "usr_1", "ana", "claude_code", harness_id, None::<String>, None::<String>, at, at,
23 parent, agent, "Find the endpoints", "Find every API endpoint and list them", None::<String>
24 ],
25 |row| row.get::<_, String>(0),
26 )
27 .optional()
28 .unwrap()
29}
30
31fn column(db: &Connection, sql: &str, params: impl rusqlite::Params) -> Vec<String> {
32 let mut statement = db.prepare(sql).unwrap();
33 statement.query_map(params, |row| row.get::<_, String>(0)).unwrap().map(Result::unwrap).collect()
34}
35
36fn touch(db: &Connection, session: &str, kind: &str, value: &str) -> bool {
37 db.query_row(INSERT_TOUCH, params![session, "rep_1", kind, value, rfc3339(NOW)], |_| Ok(())).optional().unwrap().is_some()
38}
39
40fn database() -> Connection {
41 database_at(":memory:")
42}
43
44fn database_at(path: impl AsRef<std::path::Path>) -> Connection {
45 let db = Connection::open(path).unwrap();
46 db.execute_batch("PRAGMA foreign_keys = ON;").unwrap();
47 let dir = concat!(env!("CARGO_MANIFEST_DIR"), "/migrations");
48 let mut names: Vec<String> = std::fs::read_dir(dir)
49 .unwrap()
50 .map(|entry| entry.unwrap().file_name().into_string().unwrap())
51 .filter(|name| name.ends_with(".sql"))
52 .collect();
53 names.sort();
54 for name in names {
55 let sql = std::fs::read_to_string(format!("{dir}/{name}")).unwrap();
56 db.execute_batch(&sql).unwrap_or_else(|error| panic!("{name}: {error}"));
57 }
58 db.execute(
59 "INSERT INTO pulls (id, repo_id, number, title, agent, runtime, author_id, author_name, created_at, updated_at)
60 VALUES ('pr_9', 'rep_1', 9, 'Greet', 'claude-code', 'external', 'usr_1', 'ana', '', '')",
61 [],
62 )
63 .unwrap();
64 db
65}
66
67fn claim(db: &Connection, id: &str, session: Option<&str>, actor: &str) -> String {
68 db.query_row(
69 &super::claims::upsert_claim_sql(),
70 params![id, "rep_1", 4, session, actor, "ana", "codex", "Working on it", 30, rfc3339(NOW), after_minutes(NOW, 30)],
71 |row| row.get(0),
72 ).unwrap()
73}
74
75#[test]
76fn simultaneous_claims_converge_without_locking_out_other_agents() {
77 let path = std::env::temp_dir().join(format!("g1t-claims-{}-{}.sqlite", std::process::id(),
78 std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
79 let db = database_at(&path);
80 let barrier = std::sync::Arc::new(std::sync::Barrier::new(16));
81 let threads: Vec<_> = (0..16).map(|n| {
82 let path = path.clone();
83 let barrier = barrier.clone();
84 std::thread::spawn(move || {
85 let db = Connection::open(path).unwrap();
86 db.busy_timeout(std::time::Duration::from_secs(10)).unwrap();
87 // Every caller's preliminary read precedes every write.
88 assert_eq!(db.query_row("SELECT COUNT(*) FROM claims", [], |r| r.get::<_, i64>(0)).unwrap(), 0);
89 barrier.wait();
90 claim(&db, &format!("clm_{n}"), Some("ses_a"), "usr_1")
91 })
92 }).collect();
93 let ids: HashSet<_> = threads.into_iter().map(|t| t.join().unwrap()).collect();
94 assert_eq!(ids.len(), 1);
95 let first = ids.into_iter().next().unwrap();
96 assert_ne!(claim(&db, "sibling", Some("ses_b"), "usr_1"), first);
97 assert_ne!(claim(&db, "person", Some("ses_a"), "usr_2"), first);
98 db.execute("UPDATE claims SET state = 'released' WHERE id = ?", [&first]).unwrap();
99 assert_eq!(claim(&db, "replacement", Some("ses_a"), "usr_1"), "replacement");
100 assert_eq!(db.query_row("SELECT COUNT(*) FROM claims WHERE state = 'active'", [], |r| r.get::<_, i64>(0)).unwrap(), 3);
101 drop(db);
102 std::fs::remove_file(path).unwrap();
103}
104
105#[test]
106fn a_session_adopts_a_sessionless_claim_and_handoff_allows_a_new_claim() {
107 let db = database();
108 assert_eq!(claim(&db, "first", None, "usr_1"), "first");
109 assert_eq!(claim(&db, "retry", None, "usr_1"), "first");
110 assert_eq!(claim(&db, "attached", Some("ses_a"), "usr_1"), "first");
111 assert_eq!(claim(&db, "sibling", Some("ses_b"), "usr_1"), "sibling");
112 db.execute("UPDATE claims SET state = 'handed_off' WHERE id = 'first'", []).unwrap();
113 assert_eq!(claim(&db, "successor", Some("ses_a"), "usr_1"), "successor");
114 db.execute("UPDATE claims SET state = 'expired' WHERE id = 'successor'", []).unwrap();
115 assert_eq!(claim(&db, "after_expiry", Some("ses_a"), "usr_1"), "after_expiry");
116}
117
118fn start(db: &Connection, id: &str, harness_id: Option<&str>) -> Option<String> {
119 let at = rfc3339(NOW);
120 db.query_row(
121 &insert_session_sql(),
122 params![
123 id, "rep_1", 4, None::<u32>, "main", "usr_1", "ana", "claude_code", harness_id, None::<String>, None::<String>, at, at,
124 None::<String>, None::<String>, None::<String>, None::<String>, None::<String>
125 ],
126 |row| row.get::<_, String>(0),
127 )
128 .optional()
129 .unwrap()
130}
131
132/// Records an entry the way `record_session_entries` does: the session
133/// number it took, or none when its key was recorded before.
134fn record(db: &Connection, pull: Option<&str>, session: &str, text: &str, key: Option<&str>) -> Option<u32> {
135 db.query_row(INSERT_ENTRY, params![pull, "note", text, None::<String>, None::<String>, rfc3339(NOW), session, key], |row| {
136 row.get(0)
137 })
138 .optional()
139 .unwrap()
140}
141
142fn pull_entries(db: &Connection) -> Vec<(u32, String)> {
143 let mut rows = db.prepare("SELECT seq, text FROM session_entries WHERE pull_id = 'pr_9' ORDER BY seq").unwrap();
144 rows.query_map([], |row| Ok((row.get(0)?, row.get(1)?))).unwrap().map(Result::unwrap).collect()
145}
146
147#[test]
148fn starting_a_harness_session_again_returns_the_first() {
149 let db = database();
150 assert_eq!(start(&db, "ses_1", Some("claude-abc")).as_deref(), Some("ses_1"));
151 // The hook fired twice: nothing new, and the first is found by its id.
152 assert_eq!(start(&db, "ses_2", Some("claude-abc")), None);
153 let found: String = db
154 .query_row(&harness_session_sql(), params!["rep_1", "usr_1", "claude_code", "claude-abc"], |row| row.get(0))
155 .unwrap();
156 assert_eq!(found, "ses_1");
157 // Without a harness id every start is a session of its own.
158 assert_eq!(start(&db, "ses_3", None).as_deref(), Some("ses_3"));
159 assert_eq!(start(&db, "ses_4", None).as_deref(), Some("ses_4"));
160}
161
162#[test]
163fn a_quiet_session_times_out_and_a_heartbeat_brings_it_back() {
164 let db = database();
165 start(&db, "ses_1", Some("a"));
166 // Two hours and more without a heartbeat.
167 let late = rfc3339(NOW + g1t_contracts::work::SESSION_TIMEOUT_MS);
168 let ended: Vec<String> = db
169 .prepare(&time_out_sessions_sql())
170 .unwrap()
171 .query_map(params!["rep_1", late, "ses_1"], |row| row.get(0))
172 .unwrap()
173 .map(Result::unwrap)
174 .collect();
175 assert_eq!(ended, ["ses_1"]);
176 let reason: String = db.query_row("SELECT end_reason FROM sessions WHERE id = 'ses_1'", [], |row| row.get(0)).unwrap();
177 assert_eq!(reason, "timed_out");
178 // Noticed once only.
179 assert_eq!(db.prepare(&time_out_sessions_sql()).unwrap().query_map(params!["rep_1", late, "ses_1"], |_| Ok(())).unwrap().count(), 0);
180 // The harness reconnects.
181 let state: String = db
182 .query_row(
183 &resume_sql(),
184 params![
185 None::<u32>, 9, None::<String>, "abc123", None::<String>, None::<String>, None::<String>, None::<String>,
186 None::<String>, None::<String>, late, "ses_1"
187 ],
188 |row| row.get(10),
189 )
190 .unwrap();
191 assert_eq!(state, "active");
192}
193
194#[test]
195fn a_subagent_starts_once_and_once_finished_stays_finished() {
196 let db = database();
197 start(&db, "ses_parent", Some("claude-abc"));
198 let child = subagent_harness_id("claude-abc", "agent-def456");
199 assert_eq!(start_child(&db, "ses_c1", "ses_parent", &child, "Explore").as_deref(), Some("ses_c1"));
200 // SubagentStart, then the Agent tool's answer naming it without its prefix.
201 assert_eq!(start_child(&db, "ses_c2", "ses_parent", &subagent_harness_id("claude-abc", "def456"), "Explore"), None);
202 let found: String = db.query_row(&harness_session_sql(), params!["rep_1", "usr_1", "claude_code", child], |row| row.get(0)).unwrap();
203 assert_eq!(found, "ses_c1");
204 // SubagentStop ended it with its result.
205 db.execute("UPDATE sessions SET state = 'ended', end_reason = 'done', result = 'Found 12 endpoints' WHERE id = 'ses_c1'", [])
206 .unwrap();
207 // A late hook only fills in what it says: it stays ended, its result kept.
208 let (state, task, result): (String, String, String) = db
209 .query_row(
210 &refresh_sql(),
211 params![
212 None::<u32>, None::<u32>, None::<String>, None::<String>, None::<String>, None::<String>, "List endpoints",
213 None::<String>, None::<String>, None::<String>, rfc3339(NOW), "ses_c1"
214 ],
215 |row| Ok((row.get(10)?, row.get(18)?, row.get(20)?)),
216 )
217 .unwrap();
218 assert_eq!((state.as_str(), task.as_str(), result.as_str()), ("ended", "List endpoints", "Found 12 endpoints"));
219 // Its tree, from the parent.
220 assert_eq!(column(&db, &tree_sql(), params!["ses_parent"]), ["ses_c1"]);
221}
222
223#[test]
224fn ending_a_session_ends_the_subagents_it_left_going() {
225 let db = database();
226 start(&db, "ses_parent", Some("p"));
227 start_child(&db, "ses_c1", "ses_parent", "p/agent/1", "Explore");
228 start_child(&db, "ses_c2", "ses_parent", "p/agent/2", "Plan");
229 // A subagent of a subagent.
230 start_child(&db, "ses_c3", "ses_c1", "p/agent/3", "Explore");
231 db.execute("UPDATE sessions SET state = 'ended', end_reason = 'done' WHERE id = 'ses_c2'", []).unwrap();
232 // Not before the parent itself ended, at that moment.
233 assert!(column(&db, &end_tree_sql(), params!["ses_parent", rfc3339(NOW), "parent ended"]).is_empty());
234 db.execute("UPDATE sessions SET state = 'ended', ended_at = ?1 WHERE id = 'ses_parent'", [rfc3339(NOW)]).unwrap();
235 let mut ended = column(&db, &end_tree_sql(), params!["ses_parent", rfc3339(NOW), "parent ended"]);
236 ended.sort();
237 assert_eq!(ended, ["ses_c1", "ses_c3"], "the one already done keeps its own end");
238 let reason: String = db.query_row("SELECT end_reason FROM sessions WHERE id = 'ses_c2'", [], |row| row.get(0)).unwrap();
239 assert_eq!(reason, "done");
240 let mut family = column(&db, FAMILY, params!["ses_c1"]);
241 family.sort();
242 assert_eq!(family, ["ses_c1", "ses_c3", "ses_parent"], "its parents up and its subagents down, not its sibling");
243}
244
245#[test]
246fn links_and_touches_are_kept_once_each() {
247 let db = database();
248 start(&db, "ses_1", Some("a"));
249 assert!(touch(&db, "ses_1", "file", "src/lib.rs"));
250 assert!(!touch(&db, "ses_1", "file", "src/lib.rs"), "said again: nothing new");
251 assert!(touch(&db, "ses_1", "branch", "4-greet"));
252 let link = |via: &str| db.execute(INSERT_LINK, params!["ses_1", "rep_1", "issue", 4, via, rfc3339(NOW)]).unwrap();
253 assert_eq!(link("claim"), 1);
254 assert_eq!(link("claim"), 0);
255 assert_eq!(link("ref"), 1);
256 // A pull request opened on the branch after the push finds the session.
257 db.execute("UPDATE pulls SET source_branch = '4-greet' WHERE id = 'pr_9'", []).unwrap();
258 assert_eq!(column(&db, BRANCH_SESSIONS, params!["rep_1", "[9]"]), ["ses_1"]);
259 assert!(column(&db, BRANCH_SESSIONS, params!["rep_1", "[10]"]).is_empty());
260}
261
262/// Twelve subagents of one session work on one issue, some on the same
263/// files: each is listed on the issue beside the others, each is told of
264/// every other that touched a file it touched (once per pair, for both
265/// sides), and each holds a claim of its own that the others see.
266#[test]
267fn twelve_subagents_on_one_issue_see_each_other() {
268 let db = database();
269 start(&db, "ses_parent", Some("lead"));
270 let children: Vec<String> = (1..=12).map(|n| format!("ses_c{n:02}")).collect();
271 for (n, id) in children.iter().enumerate() {
272 start_child(&db, id, "ses_parent", &subagent_harness_id("lead", &format!("agent-{n}")), "general-purpose").unwrap();
273 // Every one edits the shared router; each pair also shares a file.
274 touch(&db, id, "file", "src/router.rs");
275 touch(&db, id, "file", &format!("src/part{}.rs", n / 2));
276 touch(&db, id, "file", &format!("src/only{n}.rs"));
277 }
278 // All twelve, and their parent, are on the issue.
279 let on_issue = column(&db, "SELECT id FROM sessions WHERE repo_id = 'rep_1' AND issue = 4 ORDER BY id", []);
280 assert_eq!(on_issue.len(), 13);
281 let mut announced = 0;
282 for id in &children {
283 let n: usize = id[5..].parse().unwrap();
284 let mine = vec!["src/router.rs".to_owned(), format!("src/part{}.rs", (n - 1) / 2), format!("src/only{}.rs", n - 1)];
285 let touched: Vec<(String, String)> = {
286 let mut statement = db.prepare(TOUCHED_BY_OTHERS).unwrap();
287 statement
288 .query_map(params!["rep_1", serde_json::to_string(&mine).unwrap(), id], |row| Ok((row.get(0)?, row.get(1)?)))
289 .unwrap()
290 .map(Result::unwrap)
291 .collect()
292 };
293 let family: HashSet<String> = column(&db, FAMILY, params![id]).into_iter().collect();
294 let overlaps = session_overlaps(id, &family, &touched);
295 // Every sibling, and never its own parent.
296 assert_eq!(overlaps.len(), 11, "{id} sees its eleven siblings");
297 assert!(overlaps.iter().all(|(other, _)| other != "ses_parent" && other != id));
298 let partner = overlaps.iter().find(|(_, files)| files.len() == 2).expect("the sibling sharing a part file");
299 assert_eq!(partner.1, vec![format!("src/part{}.rs", (n - 1) / 2), "src/router.rs".to_owned()]);
300 for (other, files) in overlaps {
301 let (low, high) = session_pair(id, &other);
302 let new = db
303 .query_row(INSERT_OVERLAP, params!["rep_1", low, high, serde_json::to_string(&files).unwrap(), rfc3339(NOW)], |_| Ok(()))
304 .optional()
305 .unwrap()
306 .is_some();
307 // Each new pair is two overlap.detected events, one for each side.
308 if new {
309 announced += 2;
310 }
311 }
312 }
313 assert_eq!(announced, 12 * 11, "66 pairs, each announced to both sides once");
314 // Each claims the issue from its own session: the others see eleven.
315 let claims: Vec<(&str, Option<&str>)> = children.iter().map(|id| ("usr_1", Some(id.as_str()))).collect();
316 for id in &children {
317 let others = claims.iter().filter(|(actor, session)| !claim_is_mine(actor, *session, "usr_1", Some(id))).count();
318 assert_eq!(others, 11, "{id}");
319 }
320}
321
322#[test]
323fn entries_are_recorded_once_per_key_and_numbered_in_the_session() {
324 let db = database();
325 start(&db, "ses_1", Some("a"));
326 assert_eq!(record(&db, None, "ses_1", "planning", Some("k1")), Some(1));
327 // The hook retried: skipped.
328 assert_eq!(record(&db, None, "ses_1", "planning", Some("k1")), None);
329 assert_eq!(record(&db, None, "ses_1", "no key", None), Some(2));
330 assert_eq!(record(&db, None, "ses_1", "no key", None), Some(3));
331 // Another session numbers its own, and may reuse a key.
332 start(&db, "ses_2", Some("b"));
333 assert_eq!(record(&db, None, "ses_2", "other", Some("k1")), Some(1));
334}
335
336#[test]
337fn a_sessions_entries_join_its_pull_request_after_what_it_has() {
338 let db = database();
339 // Recorded on the pull request the old way.
340 db.execute(
341 "INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_9', 1, 'note', 'legacy', '')",
342 [],
343 )
344 .unwrap();
345 start(&db, "ses_1", Some("a"));
346 record(&db, None, "ses_1", "before the pull request", Some("k1"));
347 record(&db, None, "ses_1", "still before", Some("k2"));
348 // The session names #9: what it recorded moves onto it, in order.
349 let base: u32 = db.query_row("SELECT COALESCE(MAX(seq), 0) FROM session_entries WHERE pull_id = 'pr_9'", [], |row| row.get(0)).unwrap();
350 db.execute(MOVE_ENTRIES, params!["pr_9", base, "ses_1"]).unwrap();
351 // Then it records on the pull request as it goes.
352 assert_eq!(record(&db, Some("pr_9"), "ses_1", "after", Some("k3")), Some(3));
353 assert_eq!(record(&db, Some("pr_9"), "ses_1", "after", Some("k3")), None);
354 assert_eq!(
355 pull_entries(&db),
356 [(1, "legacy".into()), (2, "before the pull request".into()), (3, "still before".into()), (4, "after".into())]
357 );
358}
359
360#[test]
361fn claims_expire_when_due_and_heartbeats_only_extend_them() {
362 let db = database();
363 let insert = |id: &str, expires_at: &str| {
364 db.execute(
365 "INSERT INTO claims (id, repo_id, issue, actor_id, actor_name, state, ttl_minutes, created_at, expires_at)
366 VALUES (?1, 'rep_1', 4, 'usr_1', 'ana', 'active', 30, '', ?2)",
367 params![id, expires_at],
368 )
369 .unwrap();
370 };
371 insert("clm_due", &rfc3339(NOW - 1));
372 insert("clm_live", &after_minutes(NOW, 10));
373 let expire = |id: &str| -> Vec<String> {
374 db.prepare(&expire_claims_sql())
375 .unwrap()
376 .query_map(params!["rep_1", rfc3339(NOW), id], |row| row.get(0))
377 .unwrap()
378 .map(Result::unwrap)
379 .collect()
380 };
381 assert!(expire("clm_live").is_empty(), "a claim not yet due stays");
382 assert_eq!(expire("clm_due"), ["clm_due"]);
383 let (state, ended_at): (String, String) =
384 db.query_row("SELECT state, ended_at FROM claims WHERE id = 'clm_due'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap();
385 assert_eq!((state.as_str(), ended_at), ("expired", rfc3339(NOW - 1)));
386 // A heartbeat renews the live one by its own length.
387 let renewed = renewal(NOW, 30, &after_minutes(NOW, 10)).unwrap();
388 assert_eq!(renewed, after_minutes(NOW, 30));
389 assert_eq!(db.execute(RENEW_CLAIM, params![renewed, "clm_live"]).unwrap(), 1);
390 // Never an expired one, and never shorter than it is.
391 assert_eq!(db.execute(RENEW_CLAIM, params![renewed, "clm_due"]).unwrap(), 0);
392 assert_eq!(renewal(NOW, 30, &after_minutes(NOW, 240)), None);
393}
394
395#[test]
396fn an_untitled_sessions_first_prompt_is_found_by_its_own_numbering() {
397 let db = database();
398 start(&db, "ses_1", Some("claude-1")).unwrap();
399 start(&db, "ses_2", Some("claude-2")).unwrap();
400 let entry = |session: &str, kind: &str, text: &str| {
401 db.query_row(INSERT_ENTRY, params![None::<String>, kind, text, None::<String>, None::<String>, rfc3339(NOW), session, None::<String>], |_| Ok(()))
402 .unwrap();
403 };
404 entry("ses_1", "tool_call", "Bash: ls");
405 entry("ses_1", "prompt", "Fix the preview builds that time out");
406 entry("ses_1", "prompt", "and the second one");
407 // ses_2 recorded no prompt.
408 entry("ses_2", "note", "a note");
409 let mut statement = db.prepare(FIRST_PROMPTS).unwrap();
410 let found: Vec<(String, String)> = statement
411 .query_map(params![r#"["ses_1","ses_2","ses_none"]"#, 12], |row| Ok((row.get(0)?, row.get(1)?)))
412 .unwrap()
413 .map(Result::unwrap)
414 .collect();
415 assert_eq!(found, vec![("ses_1".to_owned(), "Fix the prev".to_owned())]);
416}
417
418#[test]
419fn inbox_delivery_acknowledgement_and_reconnect_are_session_scoped() {
420 use super::updates::{PENDING, ACK};
421 let db = database();
422 start(&db, "ses_a", Some("a"));
423 start(&db, "ses_b", Some("b"));
424 start(&db, "ses_ended", Some("ended"));
425 start(&db, "ses_other", Some("other"));
426 db.execute("UPDATE sessions SET state = 'ended' WHERE id = 'ses_ended'", []).unwrap();
427 db.execute("UPDATE sessions SET repo_id = 'other' WHERE id = 'ses_other'", []).unwrap();
428 let event = params!["upd_1", "overlap.detected", "{}", rfc3339(NOW), "rep_1", "[]"];
429 assert_eq!(db.execute(&deliver_sql("1"), event).unwrap(), 2);
430 assert_eq!(db.execute(&deliver_sql("1"), event).unwrap(), 0);
431 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
432 // Reading repeatedly, including after a reconnect, never clears updates.
433 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
434 assert!(column(&db, PENDING, ["ses_other"]).is_empty());
435 assert!(column(&db, PENDING, ["ses_ended"]).is_empty());
436 db.execute(ACK, params!["first", "ses_a", "upd_1"]).unwrap();
437 db.execute(ACK, params!["retry", "ses_a", "upd_1"]).unwrap();
438 assert!(column(&db, PENDING, ["ses_a"]).is_empty());
439 assert_eq!(column(&db, PENDING, ["ses_b"]), ["upd_1"]);
440 assert_eq!(column(&db, "SELECT acknowledged_at FROM session_updates WHERE session_id = 'ses_a'", []), ["first"]);
441 db.execute(&deliver_sql("1"), params!["upd_2", "dependency.changed", "{}", rfc3339(NOW+1), "rep_1", "[]"]).unwrap();
442 // An old acknowledgement cannot swallow a later update.
443 db.execute(ACK, params!["retry", "ses_a", "upd_1"]).unwrap();
444 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_2"]);
445}
446
447#[test]
448fn inbox_drains_bounded_pages_without_skipping_updates() {
449 use super::updates::{PENDING, ACK};
450 let db = database();
451 start(&db, "ses_a", Some("a"));
452 for i in 0..105 {
453 db.execute(&deliver_sql("1"), params![format!("upd_{i:03}"), "work.claimed", "{}", rfc3339(NOW), "rep_1", "[]"]).unwrap();
454 }
455 let first = column(&db, PENDING, ["ses_a"]);
456 assert_eq!(first.len(), 100);
457 for id in first { db.execute(ACK, params!["now", "ses_a", id]).unwrap(); }
458 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_100", "upd_101", "upd_102", "upd_103", "upd_104"]);
459}
460
461#[test]
462fn pending_updates_and_receipts_survive_database_reopen() {
463 use super::updates::{PENDING, ACK};
464 let path = std::env::temp_dir().join(format!("g1t-updates-{}-{}.sqlite", std::process::id(),
465 std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
466 let db = database_at(&path);
467 start(&db, "ses_a", Some("a"));
468 db.execute(&deliver_sql("1"), params!["upd_1", "overlap.detected", "{}", rfc3339(NOW), "rep_1", "[]"]).unwrap();
469 drop(db);
470 let db = Connection::open(&path).unwrap();
471 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
472 db.execute(ACK, params!["considered", "ses_a", "upd_1"]).unwrap();
473 drop(db);
474 let db = Connection::open(&path).unwrap();
475 assert!(column(&db, PENDING, ["ses_a"]).is_empty());
476 assert_eq!(column(&db, "SELECT acknowledged_at FROM session_updates", []), ["considered"]);
477 drop(db);
478 std::fs::remove_file(path).unwrap();
479}
480
481#[test]
482fn inbox_targets_session_overlaps_and_excludes_own_lifecycle_events() {
483 use super::updates::{PENDING};
484 let db = database();
485 start(&db, "ses_a", Some("a"));
486 start(&db, "ses_b", Some("b"));
487 db.execute(&deliver_sql("1"), params!["upd_overlap", "overlap.detected", r#"{"sessionId":"ses_a","otherSessionId":"ses_b"}"#, rfc3339(NOW), "rep_1", "[]"]).unwrap();
488 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_overlap"]);
489 assert!(column(&db, PENDING, ["ses_b"]).is_empty());
490 db.execute(&deliver_sql("1"), params!["upd_start", "session.started", r#"{"session":{"id":"ses_b"}}"#, rfc3339(NOW+1), "rep_1", "[]"]).unwrap();
491 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_overlap", "upd_start"]);
492 assert!(column(&db, PENDING, ["ses_b"]).is_empty());
493}
494
495#[test]
496fn a_session_is_not_told_of_its_own_claims_but_others_are() {
497 use super::updates::{PENDING};
498 let db = database();
499 start(&db, "ses_a", Some("a"));
500 start(&db, "ses_b", Some("b"));
501 for (id, kind) in [("upd_claim", "work.claimed"), ("upd_release", "work.released"), ("upd_handoff", "work.handoff_requested")] {
502 db.execute(&deliver_sql("1"), params![id, kind, r#"{"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW), "rep_1", "[]"]).unwrap();
503 }
504 // An expiry happened to the session; it hears of it.
505 db.execute(&deliver_sql("1"), params!["upd_expired", "work.claim_expired", r#"{"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW + 1), "rep_1", "[]"]).unwrap();
506 // A sessionless claim is news to every session.
507 db.execute(&deliver_sql("1"), params!["upd_bare", "work.claimed", r#"{"claim":{"sessionId":null}}"#, rfc3339(NOW + 2), "rep_1", "[]"]).unwrap();
508 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_expired", "upd_bare"]);
509 assert_eq!(column(&db, PENDING, ["ses_b"]), ["upd_claim", "upd_handoff", "upd_release", "upd_expired", "upd_bare"]);
510}
511
512#[test]
513fn acknowledging_counts_only_this_sessions_updates() {
514 use super::updates::{ACK, KNOWN};
515 let db = database();
516 start(&db, "ses_a", Some("a"));
517 start(&db, "ses_b", Some("b"));
518 db.execute(&deliver_sql("1"), params!["upd_1", "dependency.changed", "{}", rfc3339(NOW), "rep_1", "[]"]).unwrap();
519 let known = |session: &str, ids: &str| db.query_row(KNOWN, params![session, ids], |row| row.get::<_, u32>(0)).unwrap();
520 assert_eq!(known("ses_a", r#"["upd_1","upd_typo"]"#), 1);
521 db.execute(ACK, params!["now", "ses_a", "upd_1"]).unwrap();
522 // A receipt still counts on a retry, and never for another session.
523 assert_eq!(known("ses_a", r#"["upd_1"]"#), 1);
524 assert_eq!(known("ses_b", r#"["upd_x"]"#), 0);
525}
526
527#[test]
528fn updates_reach_the_sessions_on_their_subject_and_those_not_on_anything_yet() {
529 use super::updates::{PENDING, deliver_sql, subjects};
530 let db = database();
531 for id in ["ses_on4", "ses_on9", "ses_free", "ses_linked", "ses_pull"] {
532 start(&db, id, Some(id));
533 }
534 // start() puts a session on #4: only ses_on4 stays there.
535 db.execute("UPDATE sessions SET issue = NULL WHERE id <> 'ses_on4'", []).unwrap();
536 db.execute("UPDATE sessions SET issue = 9 WHERE id = 'ses_on9'", []).unwrap();
537 db.execute("UPDATE sessions SET pull = 14 WHERE id = 'ses_pull'", []).unwrap();
538 // Tied to #4 another way (it named it), though attached to nothing.
539 db.execute(INSERT_LINK, params!["ses_linked", "rep_1", "issue", 4, "ref", rfc3339(NOW)]).unwrap();
540 let data = serde_json::json!({ "number": 14, "issue": 4, "claim": { "sessionId": null } });
541 assert_eq!(subjects(&data), [4, 14]);
542 db.execute(&deliver_sql("1"), params!["upd_claim", "work.claimed", data.to_string(), rfc3339(NOW), "rep_1", "[4,14]"]).unwrap();
543 let heard: Vec<&str> = ["ses_on4", "ses_on9", "ses_free", "ses_linked", "ses_pull"]
544 .into_iter()
545 .filter(|id| !column(&db, PENDING, [*id]).is_empty())
546 .collect();
547 assert_eq!(heard, ["ses_on4", "ses_free", "ses_linked", "ses_pull"], "the session on #9 is not interrupted");
548 // Tied to another issue only, a session is on something: not told of #4.
549 start(&db, "ses_elsewhere", Some("elsewhere"));
550 db.execute("UPDATE sessions SET issue = NULL WHERE id = 'ses_elsewhere'", []).unwrap();
551 db.execute(INSERT_LINK, params!["ses_elsewhere", "rep_1", "issue", 9, "claim", rfc3339(NOW)]).unwrap();
552 db.execute(&deliver_sql("1"), params!["upd_claim_2", "work.claimed", data.to_string(), rfc3339(NOW), "rep_1", "[4,14]"]).unwrap();
553 assert!(column(&db, PENDING, ["ses_elsewhere"]).is_empty());
554 // An event about nothing in particular reaches everyone.
555 db.execute(&deliver_sql("1"), params!["upd_any", "session.started", r#"{"session":{"id":"ses_x"}}"#, rfc3339(NOW + 1), "rep_1", "[]"]).unwrap();
556 assert_eq!(column(&db, PENDING, ["ses_on9"]), ["upd_any"]);
557}
558
559#[test]
560fn an_update_is_written_only_with_its_change_and_once() {
561 use super::claims::upsert_claim_sql;
562 use super::updates::{PENDING, deliver_sql, update_id};
563 let mut db = database();
564 start(&db, "ses_a", Some("a"));
565 start(&db, "ses_b", Some("b"));
566 let guard = deliver_sql("EXISTS (SELECT 1 FROM claims WHERE id = ?7)");
567 let claim_with_update = |db: &mut Connection, proposed: &str| {
568 // As one D1 batch: the claim and its update commit together.
569 let tx = db.transaction().unwrap();
570 let id: String = tx
571 .query_row(
572 &upsert_claim_sql(),
573 params![proposed, "rep_1", 4, "ses_a", "usr_1", "ana", "codex", "Scope: api", 30, rfc3339(NOW), after_minutes(NOW, 30)],
574 |row| row.get(0),
575 )
576 .unwrap();
577 let update = update_id("work.claimed", &format!("{proposed}:claimed"));
578 tx.execute(&guard, params![update, "work.claimed", r#"{"issue":4,"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW), "rep_1", "[4]", proposed]).unwrap();
579 tx.commit().unwrap();
580 id
581 };
582 assert_eq!(claim_with_update(&mut db, "clm_first"), "clm_first");
583 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
584 assert!(column(&db, PENDING, ["ses_a"]).is_empty(), "not told of its own claim");
585 // Claiming again renews: no new claim, so no update.
586 assert_eq!(claim_with_update(&mut db, "clm_retry"), "clm_first");
587 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
588 // The same change written twice (a retried or concurrent request) is one update.
589 let update = update_id("work.claimed", "clm_first:claimed");
590 db.execute(&guard, params![update, "work.claimed", "{}", rfc3339(NOW), "rep_1", "[4]", "clm_first"]).unwrap();
591 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
592 // A transaction that fails takes its update with it.
593 {
594 let tx = db.transaction().unwrap();
595 tx.execute(&guard, params!["upd_lost", "work.claimed", "{}", rfc3339(NOW), "rep_1", "[4]", "clm_first"]).unwrap();
596 drop(tx);
597 }
598 assert_eq!(column(&db, PENDING, ["ses_b"]), [update]);
599}
600
601#[test]
602fn a_pair_is_told_again_when_it_starts_changing_another_file_in_common() {
603 let db = database();
604 let announce = |files: &[&str], at: u64| {
605 db.query_row(INSERT_OVERLAP, params!["rep_1", "ses_a", "ses_b", serde_json::to_string(files).unwrap(), rfc3339(at)], |_| Ok(()))
606 .optional()
607 .unwrap()
608 .is_some()
609 };
610 assert!(announce(&["CLAUDE.md"], NOW), "first met on a harness file");
611 assert!(!announce(&["CLAUDE.md"], NOW + 1), "the same file again is not news");
612 assert!(announce(&["src/report.ts"], NOW + 2), "the code they both change is");
613 assert!(!announce(&["src/report.ts", "CLAUDE.md"], NOW + 3));
614 let (files, noticed): (String, String) = db
615 .query_row("SELECT files, noticed_at FROM session_overlaps WHERE low = 'ses_a'", [], |row| Ok((row.get(0)?, row.get(1)?)))
616 .unwrap();
617 let mut files: Vec<String> = serde_json::from_str(&files).unwrap();
618 files.sort();
619 assert_eq!(files, ["CLAUDE.md", "src/report.ts"]);
620 assert_eq!(noticed, rfc3339(NOW + 2), "noticed when it last grew, which guards its delivery");
621}
622
623#[test]
624fn a_queue_result_told_twice_reaches_each_session_once() {
625 use super::updates::{PENDING, deliver_sql, update_id};
626 let db = database();
627 start(&db, "ses_a", Some("a"));
628 let id = update_id("integration.failed", "qe_1:att_2:abc123");
629 for _ in 0..2 {
630 db.execute(&deliver_sql("1"), params![id, "integration.failed", r#"{"number":14,"issue":4}"#, rfc3339(NOW), "rep_1", "[4,14]"]).unwrap();
631 }
632 assert_eq!(column(&db, PENDING, ["ses_a"]), [id.clone()]);
633 // A later attempt's result is news of its own.
634 let next = update_id("integration.failed", "qe_1:att_3:abc123");
635 assert_ne!(next, id);
636}