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