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.

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