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
889 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, SET_USAGE, 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", None::<String>
184 ],
185 |row| row.get(11),
186 )
187 .unwrap();
188 assert_eq!(state, "active");
189}
190
191/// Reports usage the way `update_session` does, then heartbeats: the cost and usage the session reads with.
192fn report_usage(db: &Connection, id: &str, usage: serde_json::Value) -> (Option<f64>, Option<String>) {
193 let usage = g1t_contracts::model_prices::parse_usage(&usage).unwrap();
194 let cost = g1t_contracts::model_prices::estimate(&usage).cost_usd;
195 db.execute(SET_USAGE, params![serde_json::to_string(&usage).unwrap(), cost, id]).unwrap();
196 db.query_row(
197 &resume_sql(),
198 params![
199 None::<String>, None::<u32>, None::<String>, None::<String>, None::<String>, None::<String>, None::<String>, None::<String>,
200 None::<String>, None::<String>, rfc3339(NOW), id, None::<String>
201 ],
202 |row| Ok((row.get("cost_usd")?, row.get("usage")?)),
203 )
204 .unwrap()
205}
206
207#[test]
208fn a_sessions_usage_replaces_the_last_and_is_priced_at_list_price() {
209 let db = database();
210 start(&db, "ses_1", Some("a"));
211 let (cost, usage) = report_usage(&db, "ses_1", serde_json::json!([{ "model": "claude-opus-5-5", "input_tokens": 1_000_000 }]));
212 assert_eq!(cost, Some(4.0));
213 assert!(!super::cost_partial(usage.as_deref()));
214 // Cumulative: the next report replaces it, it is not added.
215 let (cost, usage) = report_usage(
216 &db,
217 "ses_1",
218 serde_json::json!([{ "model": "claude-opus-5-5", "input_tokens": 2_000_000 }, { "model": "gpt-5", "output_tokens": 10 }]),
219 );
220 assert_eq!(cost, Some(8.0));
221 assert!(super::cost_partial(usage.as_deref()));
222 // Only models priced as unknown: no cost, and partial.
223 let (cost, usage) = report_usage(&db, "ses_1", serde_json::json!([{ "model": "gpt-5", "output_tokens": 10 }]));
224 assert_eq!(cost, None);
225 assert!(super::cost_partial(usage.as_deref()));
226 let read: Option<f64> = db.query_row("SELECT sessions.cost_usd FROM sessions WHERE id = 'ses_1'", [], |row| row.get(0)).unwrap();
227 assert_eq!(read, None);
228 assert!(!super::cost_partial(None));
229}
230
231#[test]
232fn a_subagent_starts_once_and_once_finished_stays_finished() {
233 let db = database();
234 start(&db, "ses_parent", Some("claude-abc"));
235 let child = subagent_harness_id("claude-abc", "agent-def456");
236 assert_eq!(start_child(&db, "ses_c1", "ses_parent", &child, "Explore").as_deref(), Some("ses_c1"));
237 // SubagentStart, then the Agent tool's answer naming it without its prefix.
238 assert_eq!(start_child(&db, "ses_c2", "ses_parent", &subagent_harness_id("claude-abc", "def456"), "Explore"), None);
239 let found: String = db.query_row(&harness_session_sql(), params!["rep_1", "usr_1", "claude_code", child], |row| row.get(0)).unwrap();
240 assert_eq!(found, "ses_c1");
241 // SubagentStop ended it with its result.
242 db.execute("UPDATE sessions SET state = 'ended', end_reason = 'done', result = 'Found 12 endpoints' WHERE id = 'ses_c1'", [])
243 .unwrap();
244 // A late hook only fills in what it says: it stays ended, its result kept.
245 let (state, task, result): (String, String, String) = db
246 .query_row(
247 &refresh_sql(),
248 params![
249 None::<String>, None::<u32>, None::<String>, None::<String>, None::<String>, None::<String>, "List endpoints",
250 None::<String>, None::<String>, None::<String>, rfc3339(NOW), "ses_c1", None::<String>
251 ],
252 |row| Ok((row.get(11)?, row.get(19)?, row.get(21)?)),
253 )
254 .unwrap();
255 assert_eq!((state.as_str(), task.as_str(), result.as_str()), ("ended", "List endpoints", "Found 12 endpoints"));
256 // Its tree, from the parent.
257 assert_eq!(column(&db, &tree_sql(), params!["ses_parent"]), ["ses_c1"]);
258}
259
260#[test]
261fn ending_a_session_ends_the_subagents_it_left_going() {
262 let db = database();
263 start(&db, "ses_parent", Some("p"));
264 start_child(&db, "ses_c1", "ses_parent", "p/agent/1", "Explore");
265 start_child(&db, "ses_c2", "ses_parent", "p/agent/2", "Plan");
266 // A subagent of a subagent.
267 start_child(&db, "ses_c3", "ses_c1", "p/agent/3", "Explore");
268 db.execute("UPDATE sessions SET state = 'ended', end_reason = 'done' WHERE id = 'ses_c2'", []).unwrap();
269 // Not before the parent itself ended, at that moment.
270 assert!(column(&db, &end_tree_sql(), params!["ses_parent", rfc3339(NOW), "parent ended"]).is_empty());
271 db.execute("UPDATE sessions SET state = 'ended', ended_at = ?1 WHERE id = 'ses_parent'", [rfc3339(NOW)]).unwrap();
272 let mut ended = column(&db, &end_tree_sql(), params!["ses_parent", rfc3339(NOW), "parent ended"]);
273 ended.sort();
274 assert_eq!(ended, ["ses_c1", "ses_c3"], "the one already done keeps its own end");
275 let reason: String = db.query_row("SELECT end_reason FROM sessions WHERE id = 'ses_c2'", [], |row| row.get(0)).unwrap();
276 assert_eq!(reason, "done");
277 let mut family = column(&db, FAMILY, params!["ses_c1"]);
278 family.sort();
279 assert_eq!(family, ["ses_c1", "ses_c3", "ses_parent"], "its parents up and its subagents down, not its sibling");
280}
281
282#[test]
283fn links_and_touches_are_kept_once_each() {
284 let db = database();
285 start(&db, "ses_1", Some("a"));
286 assert!(touch(&db, "ses_1", "file", "src/lib.rs"));
287 assert!(!touch(&db, "ses_1", "file", "src/lib.rs"), "said again: nothing new");
288 assert!(touch(&db, "ses_1", "branch", "4-greet"));
289 let link = |via: &str| db.execute(INSERT_LINK, params!["ses_1", "rep_1", "item", 0, via, rfc3339(NOW), "wki_4"]).unwrap();
290 assert_eq!(link("claim"), 1);
291 assert_eq!(link("claim"), 0);
292 assert_eq!(link("ref"), 1);
293 // A pull request opened on the branch after the push finds the session.
294 db.execute("UPDATE pulls SET source_branch = '4-greet' WHERE id = 'pr_9'", []).unwrap();
295 assert_eq!(column(&db, BRANCH_SESSIONS, params!["rep_1", "[9]"]), ["ses_1"]);
296 assert!(column(&db, BRANCH_SESSIONS, params!["rep_1", "[10]"]).is_empty());
297}
298
299/// Home marks what landed with the sessions that made it: tied by a link,
300/// by the pull request a session named, or by pushing its branch, and a
301/// subagent's as the session that started it.
302#[test]
303fn a_pull_requests_sessions_are_found_every_way_they_are_tied() {
304 let db = database();
305 start(&db, "ses_1", Some("a"));
306 start(&db, "ses_2", Some("b"));
307 start_child(&db, "ses_2c", "ses_2", "b/agent/1", "general-purpose");
308 db.execute("UPDATE pulls SET source_branch = '9-fix' WHERE id = 'pr_9'", []).unwrap();
309 // ses_1 pushed the branch; ses_2's subagent named the pull request in a trailer.
310 touch(&db, "ses_1", "branch", "9-fix");
311 db.execute(INSERT_LINK, params!["ses_2c", "rep_1", "pull", 9, "trailer", rfc3339(NOW), ""]).unwrap();
312 let rows = |numbers: &str| -> Vec<PullSessionRow> {
313 let mut statement = db.prepare(PULL_SESSIONS).unwrap();
314 statement
315 .query_map(params!["rep_1", numbers], |row| {
316 Ok(PullSessionRow {
317 number: row.get(0)?,
318 id: row.get(1)?,
319 harness: row.get(2)?,
320 parent_id: row.get(3)?,
321 parent_harness: row.get(5)?,
322 })
323 })
324 .unwrap()
325 .map(Result::unwrap)
326 .collect()
327 };
328 let found = pull_sessions_of(rows("[9, 10]"));
329 assert_eq!(found.len(), 1, "only pull requests with sessions");
330 assert_eq!(found[0].number, 9);
331 let mut ids: Vec<&str> = found[0].sessions.iter().map(|session| session.id.as_str()).collect();
332 ids.sort_unstable();
333 assert_eq!(ids, ["ses_1", "ses_2"], "the subagent counts as its parent, once");
334 assert!(pull_sessions_of(rows("[10]")).is_empty());
335}
336
337/// Twelve subagents of one session work on one work item, some on the
338/// same files: each is listed on it beside the others, each is told of
339/// every other that touched a file it touched (once per pair, for both
340/// sides), and each holds a claim of its own that the others see.
341#[test]
342fn twelve_subagents_on_one_work_item_see_each_other() {
343 let db = database();
344 start(&db, "ses_parent", Some("lead"));
345 let children: Vec<String> = (1..=12).map(|n| format!("ses_c{n:02}")).collect();
346 for (n, id) in children.iter().enumerate() {
347 start_child(&db, id, "ses_parent", &subagent_harness_id("lead", &format!("agent-{n}")), "general-purpose").unwrap();
348 // Every one edits the shared router; each pair also shares a file.
349 touch(&db, id, "file", "src/router.rs");
350 touch(&db, id, "file", &format!("src/part{}.rs", n / 2));
351 touch(&db, id, "file", &format!("src/only{n}.rs"));
352 }
353 // All twelve, and their parent, are on the work item.
354 let on_item = column(&db, "SELECT id FROM sessions WHERE repo_id = 'rep_1' AND item_id = 'wki_4' ORDER BY id", []);
355 assert_eq!(on_item.len(), 13);
356 let mut announced = 0;
357 for id in &children {
358 let n: usize = id[5..].parse().unwrap();
359 let mine = vec!["src/router.rs".to_owned(), format!("src/part{}.rs", (n - 1) / 2), format!("src/only{}.rs", n - 1)];
360 let touched: Vec<(String, String)> = {
361 let mut statement = db.prepare(TOUCHED_BY_OTHERS).unwrap();
362 statement
363 .query_map(params!["rep_1", serde_json::to_string(&mine).unwrap(), id], |row| Ok((row.get(0)?, row.get(1)?)))
364 .unwrap()
365 .map(Result::unwrap)
366 .collect()
367 };
368 let family: HashSet<String> = column(&db, FAMILY, params![id]).into_iter().collect();
369 let overlaps = session_overlaps(id, &family, &touched);
370 // Every sibling, and never its own parent.
371 assert_eq!(overlaps.len(), 11, "{id} sees its eleven siblings");
372 assert!(overlaps.iter().all(|(other, _)| other != "ses_parent" && other != id));
373 let partner = overlaps.iter().find(|(_, files)| files.len() == 2).expect("the sibling sharing a part file");
374 assert_eq!(partner.1, vec![format!("src/part{}.rs", (n - 1) / 2), "src/router.rs".to_owned()]);
375 for (other, files) in overlaps {
376 let (low, high) = session_pair(id, &other);
377 let new = db
378 .query_row(INSERT_OVERLAP, params!["rep_1", low, high, serde_json::to_string(&files).unwrap(), rfc3339(NOW)], |_| Ok(()))
379 .optional()
380 .unwrap()
381 .is_some();
382 // Each new pair is two overlap.detected events, one for each side.
383 if new {
384 announced += 2;
385 }
386 }
387 }
388 assert_eq!(announced, 12 * 11, "66 pairs, each announced to both sides once");
389 // Each claims the work item from its own session: the others see eleven.
390 let claims: Vec<(&str, Option<&str>)> = children.iter().map(|id| ("usr_1", Some(id.as_str()))).collect();
391 for id in &children {
392 let others = claims.iter().filter(|(actor, session)| !claim_is_mine(actor, *session, "usr_1", Some(id))).count();
393 assert_eq!(others, 11, "{id}");
394 }
395}
396
397#[test]
398fn entries_are_recorded_once_per_key_and_numbered_in_the_session() {
399 let db = database();
400 start(&db, "ses_1", Some("a"));
401 assert_eq!(record(&db, None, "ses_1", "planning", Some("k1")), Some(1));
402 // The hook retried: skipped.
403 assert_eq!(record(&db, None, "ses_1", "planning", Some("k1")), None);
404 assert_eq!(record(&db, None, "ses_1", "no key", None), Some(2));
405 assert_eq!(record(&db, None, "ses_1", "no key", None), Some(3));
406 // Another session numbers its own, and may reuse a key.
407 start(&db, "ses_2", Some("b"));
408 assert_eq!(record(&db, None, "ses_2", "other", Some("k1")), Some(1));
409}
410
411#[test]
412fn a_sessions_entries_join_its_pull_request_after_what_it_has() {
413 let db = database();
414 // Recorded on the pull request the old way.
415 db.execute(
416 "INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_9', 1, 'note', 'legacy', '')",
417 [],
418 )
419 .unwrap();
420 start(&db, "ses_1", Some("a"));
421 record(&db, None, "ses_1", "before the pull request", Some("k1"));
422 record(&db, None, "ses_1", "still before", Some("k2"));
423 // The session names #9: what it recorded moves onto it, in order.
424 let base: u32 = db.query_row("SELECT COALESCE(MAX(seq), 0) FROM session_entries WHERE pull_id = 'pr_9'", [], |row| row.get(0)).unwrap();
425 db.execute(MOVE_ENTRIES, params!["pr_9", base, "ses_1"]).unwrap();
426 // Then it records on the pull request as it goes.
427 assert_eq!(record(&db, Some("pr_9"), "ses_1", "after", Some("k3")), Some(3));
428 assert_eq!(record(&db, Some("pr_9"), "ses_1", "after", Some("k3")), None);
429 assert_eq!(
430 pull_entries(&db),
431 [(1, "legacy".into()), (2, "before the pull request".into()), (3, "still before".into()), (4, "after".into())]
432 );
433}
434
435#[test]
436fn claims_expire_when_due_and_heartbeats_only_extend_them() {
437 let db = database();
438 let insert = |id: &str, expires_at: &str| {
439 db.execute(
440 "INSERT INTO claims (id, item_id, repo_id, actor_id, actor_name, state, ttl_minutes, created_at, expires_at)
441 VALUES (?1, 'wki_4', 'rep_1', 'usr_1', 'ana', 'active', 30, '', ?2)",
442 params![id, expires_at],
443 )
444 .unwrap();
445 };
446 insert("clm_due", &rfc3339(NOW - 1));
447 insert("clm_live", &after_minutes(NOW, 10));
448 let expire = |id: &str| -> Vec<String> {
449 db.prepare(&expire_claims_sql())
450 .unwrap()
451 .query_map(params!["rep_1", rfc3339(NOW), id], |row| row.get(0))
452 .unwrap()
453 .map(Result::unwrap)
454 .collect()
455 };
456 assert!(expire("clm_live").is_empty(), "a claim not yet due stays");
457 assert_eq!(expire("clm_due"), ["clm_due"]);
458 let (state, ended_at): (String, String) =
459 db.query_row("SELECT state, ended_at FROM claims WHERE id = 'clm_due'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap();
460 assert_eq!((state.as_str(), ended_at), ("expired", rfc3339(NOW - 1)));
461 // A heartbeat renews the live one by its own length.
462 let renewed = renewal(NOW, 30, &after_minutes(NOW, 10)).unwrap();
463 assert_eq!(renewed, after_minutes(NOW, 30));
464 assert_eq!(db.execute(RENEW_CLAIM, params![renewed, "clm_live"]).unwrap(), 1);
465 // Never an expired one, and never shorter than it is.
466 assert_eq!(db.execute(RENEW_CLAIM, params![renewed, "clm_due"]).unwrap(), 0);
467 assert_eq!(renewal(NOW, 30, &after_minutes(NOW, 240)), None);
468}
469
470#[test]
471fn an_untitled_sessions_first_prompt_is_found_by_its_own_numbering() {
472 let db = database();
473 start(&db, "ses_1", Some("claude-1")).unwrap();
474 start(&db, "ses_2", Some("claude-2")).unwrap();
475 let entry = |session: &str, kind: &str, text: &str| {
476 db.query_row(INSERT_ENTRY, params![None::<String>, kind, text, None::<String>, None::<String>, rfc3339(NOW), session, None::<String>], |_| Ok(()))
477 .unwrap();
478 };
479 entry("ses_1", "tool_call", "Bash: ls");
480 entry("ses_1", "prompt", "Fix the preview builds that time out");
481 entry("ses_1", "prompt", "and the second one");
482 // ses_2 recorded no prompt.
483 entry("ses_2", "note", "a note");
484 let mut statement = db.prepare(FIRST_PROMPTS).unwrap();
485 let found: Vec<(String, String)> = statement
486 .query_map(params![r#"["ses_1","ses_2","ses_none"]"#, 12], |row| Ok((row.get(0)?, row.get(1)?)))
487 .unwrap()
488 .map(Result::unwrap)
489 .collect();
490 assert_eq!(found, vec![("ses_1".to_owned(), "Fix the prev".to_owned())]);
491}
492
493#[test]
494fn inbox_delivery_acknowledgement_and_reconnect_are_session_scoped() {
495 use super::updates::{PENDING, ACK};
496 let db = database();
497 start(&db, "ses_a", Some("a"));
498 start(&db, "ses_b", Some("b"));
499 start(&db, "ses_ended", Some("ended"));
500 start(&db, "ses_other", Some("other"));
501 db.execute("UPDATE sessions SET state = 'ended' WHERE id = 'ses_ended'", []).unwrap();
502 db.execute("UPDATE sessions SET repo_id = 'other' WHERE id = 'ses_other'", []).unwrap();
503 let event = params!["upd_1", "overlap.detected", "{}", rfc3339(NOW), REPO, "[]"];
504 assert_eq!(db.execute(&deliver_sql("1"), event).unwrap(), 2);
505 assert_eq!(db.execute(&deliver_sql("1"), event).unwrap(), 0);
506 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
507 // Reading repeatedly, including after a reconnect, never clears updates.
508 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
509 assert!(column(&db, PENDING, ["ses_other"]).is_empty());
510 assert!(column(&db, PENDING, ["ses_ended"]).is_empty());
511 db.execute(ACK, params!["first", "ses_a", "upd_1"]).unwrap();
512 db.execute(ACK, params!["retry", "ses_a", "upd_1"]).unwrap();
513 assert!(column(&db, PENDING, ["ses_a"]).is_empty());
514 assert_eq!(column(&db, PENDING, ["ses_b"]), ["upd_1"]);
515 assert_eq!(column(&db, "SELECT acknowledged_at FROM session_updates WHERE session_id = 'ses_a'", []), ["first"]);
516 db.execute(&deliver_sql("1"), params!["upd_2", "dependency.changed", "{}", rfc3339(NOW+1), REPO, "[]"]).unwrap();
517 // An old acknowledgement cannot swallow a later update.
518 db.execute(ACK, params!["retry", "ses_a", "upd_1"]).unwrap();
519 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_2"]);
520}
521
522#[test]
523fn inbox_drains_bounded_pages_without_skipping_updates() {
524 use super::updates::{PENDING, ACK};
525 let db = database();
526 start(&db, "ses_a", Some("a"));
527 for i in 0..105 {
528 db.execute(&deliver_sql("1"), params![format!("upd_{i:03}"), "work.claimed", "{}", rfc3339(NOW), REPO, "[]"]).unwrap();
529 }
530 let first = column(&db, PENDING, ["ses_a"]);
531 assert_eq!(first.len(), 100);
532 for id in first { db.execute(ACK, params!["now", "ses_a", id]).unwrap(); }
533 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_100", "upd_101", "upd_102", "upd_103", "upd_104"]);
534}
535
536#[test]
537fn pending_updates_and_receipts_survive_database_reopen() {
538 use super::updates::{PENDING, ACK};
539 let path = std::env::temp_dir().join(format!("g1t-updates-{}-{}.sqlite", std::process::id(),
540 std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
541 let db = database_at(&path);
542 start(&db, "ses_a", Some("a"));
543 db.execute(&deliver_sql("1"), params!["upd_1", "overlap.detected", "{}", rfc3339(NOW), REPO, "[]"]).unwrap();
544 drop(db);
545 let db = Connection::open(&path).unwrap();
546 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_1"]);
547 db.execute(ACK, params!["considered", "ses_a", "upd_1"]).unwrap();
548 drop(db);
549 let db = Connection::open(&path).unwrap();
550 assert!(column(&db, PENDING, ["ses_a"]).is_empty());
551 assert_eq!(column(&db, "SELECT acknowledged_at FROM session_updates", []), ["considered"]);
552 drop(db);
553 std::fs::remove_file(path).unwrap();
554}
555
556#[test]
557fn inbox_targets_session_overlaps_and_excludes_own_lifecycle_events() {
558 use super::updates::{PENDING};
559 let db = database();
560 start(&db, "ses_a", Some("a"));
561 start(&db, "ses_b", Some("b"));
562 db.execute(&deliver_sql("1"), params!["upd_overlap", "overlap.detected", r#"{"sessionId":"ses_a","otherSessionId":"ses_b"}"#, rfc3339(NOW), REPO, "[]"]).unwrap();
563 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_overlap"]);
564 assert!(column(&db, PENDING, ["ses_b"]).is_empty());
565 db.execute(&deliver_sql("1"), params!["upd_start", "session.started", r#"{"session":{"id":"ses_b"}}"#, rfc3339(NOW+1), REPO, "[]"]).unwrap();
566 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_overlap", "upd_start"]);
567 assert!(column(&db, PENDING, ["ses_b"]).is_empty());
568}
569
570#[test]
571fn a_session_is_not_told_of_its_own_claims_but_others_are() {
572 use super::updates::{PENDING};
573 let db = database();
574 start(&db, "ses_a", Some("a"));
575 start(&db, "ses_b", Some("b"));
576 for (id, kind) in [("upd_claim", "work.claimed"), ("upd_release", "work.released"), ("upd_handoff", "work.handoff_requested")] {
577 db.execute(&deliver_sql("1"), params![id, kind, r#"{"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW), REPO, "[]"]).unwrap();
578 }
579 // An expiry happened to the session; it hears of it.
580 db.execute(&deliver_sql("1"), params!["upd_expired", "work.claim_expired", r#"{"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW + 1), REPO, "[]"]).unwrap();
581 // A sessionless claim is news to every session.
582 db.execute(&deliver_sql("1"), params!["upd_bare", "work.claimed", r#"{"claim":{"sessionId":null}}"#, rfc3339(NOW + 2), REPO, "[]"]).unwrap();
583 assert_eq!(column(&db, PENDING, ["ses_a"]), ["upd_expired", "upd_bare"]);
584 assert_eq!(column(&db, PENDING, ["ses_b"]), ["upd_claim", "upd_handoff", "upd_release", "upd_expired", "upd_bare"]);
585}
586
587#[test]
588fn acknowledging_counts_only_this_sessions_updates() {
589 use super::updates::{ACK, KNOWN};
590 let db = database();
591 start(&db, "ses_a", Some("a"));
592 start(&db, "ses_b", Some("b"));
593 db.execute(&deliver_sql("1"), params!["upd_1", "dependency.changed", "{}", rfc3339(NOW), REPO, "[]"]).unwrap();
594 let known = |session: &str, ids: &str| db.query_row(KNOWN, params![session, ids], |row| row.get::<_, u32>(0)).unwrap();
595 assert_eq!(known("ses_a", r#"["upd_1","upd_typo"]"#), 1);
596 db.execute(ACK, params!["now", "ses_a", "upd_1"]).unwrap();
597 // A receipt still counts on a retry, and never for another session.
598 assert_eq!(known("ses_a", r#"["upd_1"]"#), 1);
599 assert_eq!(known("ses_b", r#"["upd_x"]"#), 0);
600}
601
602#[test]
603fn updates_reach_the_sessions_on_their_subject_and_those_not_on_anything_yet() {
604 use super::updates::{PENDING, deliver_sql, subjects};
605 let db = database();
606 for id in ["ses_on4", "ses_on9", "ses_free", "ses_linked", "ses_pull"] {
607 start(&db, id, Some(id));
608 }
609 // start() puts a session on #4: only ses_on4 stays there.
610 db.execute("UPDATE sessions SET item_id = NULL WHERE id <> 'ses_on4'", []).unwrap();
611 db.execute("UPDATE sessions SET item_id = 'wki_9' WHERE id = 'ses_on9'", []).unwrap();
612 db.execute("UPDATE sessions SET pull = 14 WHERE id = 'ses_pull'", []).unwrap();
613 // Tied to #4 another way (it named it), though attached to nothing.
614 db.execute(INSERT_LINK, params!["ses_linked", "rep_1", "item", 0, "ref", rfc3339(NOW), "wki_4"]).unwrap();
615 let data = serde_json::json!({ "number": 14, "item": 4, "claim": { "sessionId": null } });
616 assert_eq!(subjects(&data), [4, 14]);
617 db.execute(&deliver_sql("1"), params!["upd_claim", "work.claimed", data.to_string(), rfc3339(NOW), REPO, "[4,14]"]).unwrap();
618 let heard: Vec<&str> = ["ses_on4", "ses_on9", "ses_free", "ses_linked", "ses_pull"]
619 .into_iter()
620 .filter(|id| !column(&db, PENDING, [*id]).is_empty())
621 .collect();
622 assert_eq!(heard, ["ses_on4", "ses_free", "ses_linked", "ses_pull"], "the session on #9 is not interrupted");
623 // Tied to another work item only, a session is on something: not told of #4.
624 start(&db, "ses_elsewhere", Some("elsewhere"));
625 db.execute("UPDATE sessions SET item_id = NULL WHERE id = 'ses_elsewhere'", []).unwrap();
626 db.execute(INSERT_LINK, params!["ses_elsewhere", "rep_1", "item", 0, "claim", rfc3339(NOW), "wki_9"]).unwrap();
627 db.execute(&deliver_sql("1"), params!["upd_claim_2", "work.claimed", data.to_string(), rfc3339(NOW), REPO, "[4,14]"]).unwrap();
628 assert!(column(&db, PENDING, ["ses_elsewhere"]).is_empty());
629 // An event about nothing in particular reaches everyone.
630 db.execute(&deliver_sql("1"), params!["upd_any", "session.started", r#"{"session":{"id":"ses_x"}}"#, rfc3339(NOW + 1), REPO, "[]"]).unwrap();
631 assert_eq!(column(&db, PENDING, ["ses_on9"]), ["upd_any"]);
632}
633
634#[test]
635fn an_update_is_written_only_with_its_change_and_once() {
636 use super::claims::upsert_claim_sql;
637 use super::updates::{PENDING, deliver_sql, update_id};
638 let mut db = database();
639 start(&db, "ses_a", Some("a"));
640 start(&db, "ses_b", Some("b"));
641 let guard = deliver_sql("EXISTS (SELECT 1 FROM claims WHERE id = ?7)");
642 let claim_with_update = |db: &mut Connection, proposed: &str| {
643 // As one D1 batch: the claim and its update commit together.
644 let tx = db.transaction().unwrap();
645 let id: String = tx
646 .query_row(
647 &upsert_claim_sql(),
648 params![proposed, "wki_4", "rep_1", "ses_a", "usr_1", "ana", "codex", "Scope: api", 30, rfc3339(NOW), after_minutes(NOW, 30)],
649 |row| row.get(0),
650 )
651 .unwrap();
652 let update = update_id("work.claimed", &format!("{proposed}:claimed"));
653 tx.execute(&guard, params![update, "work.claimed", r#"{"item":4,"claim":{"sessionId":"ses_a"}}"#, rfc3339(NOW), REPO, "[4]", proposed]).unwrap();
654 tx.commit().unwrap();
655 id
656 };
657 assert_eq!(claim_with_update(&mut db, "clm_first"), "clm_first");
658 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
659 assert!(column(&db, PENDING, ["ses_a"]).is_empty(), "not told of its own claim");
660 // Claiming again renews: no new claim, so no update.
661 assert_eq!(claim_with_update(&mut db, "clm_retry"), "clm_first");
662 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
663 // The same change written twice (a retried or concurrent request) is one update.
664 let update = update_id("work.claimed", "clm_first:claimed");
665 db.execute(&guard, params![update, "work.claimed", "{}", rfc3339(NOW), REPO, "[4]", "clm_first"]).unwrap();
666 assert_eq!(column(&db, PENDING, ["ses_b"]).len(), 1);
667 // A transaction that fails takes its update with it.
668 {
669 let tx = db.transaction().unwrap();
670 tx.execute(&guard, params!["upd_lost", "work.claimed", "{}", rfc3339(NOW), REPO, "[4]", "clm_first"]).unwrap();
671 drop(tx);
672 }
673 assert_eq!(column(&db, PENDING, ["ses_b"]), [update]);
674}
675
676#[test]
677fn a_pair_is_told_again_when_it_starts_changing_another_file_in_common() {
678 let db = database();
679 let announce = |files: &[&str], at: u64| {
680 db.query_row(INSERT_OVERLAP, params!["rep_1", "ses_a", "ses_b", serde_json::to_string(files).unwrap(), rfc3339(at)], |_| Ok(()))
681 .optional()
682 .unwrap()
683 .is_some()
684 };
685 assert!(announce(&["CLAUDE.md"], NOW), "first met on a harness file");
686 assert!(!announce(&["CLAUDE.md"], NOW + 1), "the same file again is not news");
687 assert!(announce(&["src/report.ts"], NOW + 2), "the code they both change is");
688 assert!(!announce(&["src/report.ts", "CLAUDE.md"], NOW + 3));
689 let (files, noticed): (String, String) = db
690 .query_row("SELECT files, noticed_at FROM session_overlaps WHERE low = 'ses_a'", [], |row| Ok((row.get(0)?, row.get(1)?)))
691 .unwrap();
692 let mut files: Vec<String> = serde_json::from_str(&files).unwrap();
693 files.sort();
694 assert_eq!(files, ["CLAUDE.md", "src/report.ts"]);
695 assert_eq!(noticed, rfc3339(NOW + 2), "noticed when it last grew, which guards its delivery");
696}
697
698#[test]
699fn a_queue_result_told_twice_reaches_each_session_once() {
700 use super::updates::{PENDING, deliver_sql, update_id};
701 let db = database();
702 start(&db, "ses_a", Some("a"));
703 let id = update_id("integration.failed", "qe_1:att_2:abc123");
704 for _ in 0..2 {
705 db.execute(&deliver_sql("1"), params![id, "integration.failed", r#"{"number":14,"item":4}"#, rfc3339(NOW), REPO, "[4,14]"]).unwrap();
706 }
707 assert_eq!(column(&db, PENDING, ["ses_a"]), [id.clone()]);
708 // A later attempt's result is news of its own.
709 let next = update_id("integration.failed", "qe_1:att_3:abc123");
710 assert_ne!(next, id);
711}
712
713// --- Titles that settle (titles.rs) ----------------------------------------
714
715/// A session's title as stored: its text, source and when it is due.
716fn titled(db: &Connection, id: &str) -> (Option<String>, Option<String>, Option<String>) {
717 db.query_row("SELECT title, title_source, title_due_at FROM sessions WHERE id = ?", [id], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
718 .unwrap()
719}
720
721/// Records `n` prompts the first of which is for `intent`, at `at`.
722fn prompted(db: &Connection, id: &str, n: u32, intent: Option<&str>, at: u64) {
723 db.execute(&super::titles::prompted_sql(), params![id, n, intent, rfc3339(at), after_minutes(at, super::titles::SETTLE_AFTER_MINUTES)]).unwrap();
724}
725
726/// A heartbeat giving the title `title` from `source`, at `at`.
727fn heartbeat_titled(db: &Connection, id: &str, title: &str, source: &str, at: u64) {
728 db.query_row(
729 &resume_sql(),
730 params![
731 None::<String>, None::<u32>, None::<String>, None::<String>, title, None::<String>, None::<String>, None::<String>,
732 None::<String>, None::<String>, rfc3339(at), id, source
733 ],
734 |_| Ok(()),
735 )
736 .unwrap();
737}
738
739const MINUTE: u64 = 60_000;
740
741#[test]
742fn the_first_prompt_names_a_session_for_now_and_enough_of_them_settle_it() {
743 let db = database();
744 start(&db, "ses_1", Some("a"));
745 prompted(&db, "ses_1", 1, Some("This project used to be in syntaqx/g1t"), NOW);
746 let due_later = after_minutes(NOW, super::titles::SETTLE_AFTER_MINUTES);
747 assert_eq!(titled(&db, "ses_1"), (Some("This project used to be in syntaqx/g1t".into()), Some("prompt".into()), Some(due_later.clone())));
748 // A later prompt does not rename it, nor bring it forward yet.
749 prompted(&db, "ses_1", 1, Some("Fix the remotes"), NOW + MINUTE);
750 assert_eq!(titled(&db, "ses_1").0.as_deref(), Some("This project used to be in syntaqx/g1t"));
751 assert_eq!(titled(&db, "ses_1").2, Some(due_later));
752 // The third: due now.
753 prompted(&db, "ses_1", 1, None, NOW + 2 * MINUTE);
754 assert_eq!(titled(&db, "ses_1").2, Some(rfc3339(NOW + 2 * MINUTE)));
755 // A subagent is never titled from prompts.
756 start_child(&db, "ses_c", "ses_1", "a/agent/1", "Explore");
757 prompted(&db, "ses_c", 3, Some("Look around"), NOW);
758 assert_eq!(titled(&db, "ses_c"), (None, None, None));
759}
760
761#[test]
762fn an_agents_title_is_kept_and_a_pinned_one_never_replaced() {
763 let db = database();
764 start(&db, "ses_1", Some("a"));
765 prompted(&db, "ses_1", 1, Some("Fix login"), NOW);
766 heartbeat_titled(&db, "ses_1", "Rewrite the auth flow", "agent", NOW + MINUTE);
767 assert_eq!(titled(&db, "ses_1"), (Some("Rewrite the auth flow".into()), Some("agent".into()), None));
768 // Its prompts and links leave it be until it ends.
769 prompted(&db, "ses_1", 5, None, NOW + 2 * MINUTE);
770 db.execute(&super::titles::linked_sql(), params!["ses_1", rfc3339(NOW + 2 * MINUTE)]).unwrap();
771 assert_eq!(titled(&db, "ses_1").2, None);
772 // A person pins it: neither the agent nor g1t replaces it.
773 db.execute(super::titles::PIN_TITLE, params!["Auth, rewritten", rfc3339(NOW + 3 * MINUTE), "ses_1"]).unwrap();
774 heartbeat_titled(&db, "ses_1", "Something else", "agent", NOW + 4 * MINUTE);
775 prompted(&db, "ses_1", 5, None, NOW + 4 * MINUTE);
776 db.execute(&super::titles::linked_sql(), params!["ses_1", rfc3339(NOW + 4 * MINUTE)]).unwrap();
777 assert_eq!(titled(&db, "ses_1"), (Some("Auth, rewritten".into()), Some("person".into()), None));
778 // Not even as it ends.
779 let late = rfc3339(NOW + g1t_contracts::work::SESSION_TIMEOUT_MS * 2);
780 db.prepare(&time_out_sessions_sql()).unwrap().query_map(params!["rep_1", late, "ses_1"], |_| Ok(())).unwrap().count();
781 assert_eq!(titled(&db, "ses_1").2, None);
782 // Unpinned: g1t names it again.
783 db.execute(super::titles::UNPIN_TITLE, params![rfc3339(NOW + 5 * MINUTE), "ses_1"]).unwrap();
784 assert_eq!(titled(&db, "ses_1"), (Some("Auth, rewritten".into()), Some("prompt".into()), Some(rfc3339(NOW + 5 * MINUTE))));
785}
786
787#[test]
788fn an_agent_saying_its_title_again_leaves_when_it_settled() {
789 let db = database();
790 start(&db, "ses_1", Some("a"));
791 heartbeat_titled(&db, "ses_1", "Rewrite the auth flow", "agent", NOW + MINUTE);
792 let settled = |db: &Connection| -> Option<String> {
793 db.query_row("SELECT title_settled_at FROM sessions WHERE id = 'ses_1'", [], |row| row.get(0)).unwrap()
794 };
795 assert_eq!(settled(&db), Some(rfc3339(NOW + MINUTE)));
796 // Every heartbeat says it again: it settled when first given.
797 heartbeat_titled(&db, "ses_1", "Rewrite the auth flow", "agent", NOW + 30 * MINUTE);
798 assert_eq!(settled(&db), Some(rfc3339(NOW + MINUTE)));
799 // A new title settles anew.
800 heartbeat_titled(&db, "ses_1", "Rewrite auth and sessions", "agent", NOW + 40 * MINUTE);
801 assert_eq!(settled(&db), Some(rfc3339(NOW + 40 * MINUTE)));
802}
803
804#[test]
805fn an_agent_echoing_the_provisional_title_keeps_it_provisional() {
806 let db = database();
807 start(&db, "ses_1", Some("a"));
808 prompted(&db, "ses_1", 1, Some("Fix login"), NOW);
809 heartbeat_titled(&db, "ses_1", "Fix login", "prompt", NOW + MINUTE);
810 let (_, source, due) = titled(&db, "ses_1");
811 assert_eq!((source.as_deref(), due), (Some("prompt"), Some(after_minutes(NOW, super::titles::SETTLE_AFTER_MINUTES))));
812}
813
814#[test]
815fn a_settled_title_is_due_again_only_after_a_while_and_a_change() {
816 let db = database();
817 start(&db, "ses_1", Some("a"));
818 let settled = rfc3339(NOW);
819 db.execute(
820 "UPDATE sessions SET title = 'Move issues to Work items', title_source = 'item', title_settled_at = ?1, title_due_at = NULL,
821 title_prompts = 3, prompts = 3 WHERE id = 'ses_1'",
822 [settled.as_str()],
823 )
824 .unwrap();
825 // Tied to more work five minutes on: due, but not before twenty minutes after it settled.
826 db.execute(&super::titles::linked_sql(), params!["ses_1", rfc3339(NOW + 5 * MINUTE)]).unwrap();
827 assert_eq!(titled(&db, "ses_1").2, Some(rfc3339(NOW + 20 * MINUTE)));
828 // Later still: as soon as it is tied.
829 db.execute("UPDATE sessions SET title_due_at = NULL WHERE id = 'ses_1'", []).unwrap();
830 db.execute(&super::titles::linked_sql(), params!["ses_1", rfc3339(NOW + 45 * MINUTE)]).unwrap();
831 assert_eq!(titled(&db, "ses_1").2, Some(rfc3339(NOW + 45 * MINUTE)));
832 // Prompts: due again only once they grew past more_prompts(3) = 8.
833 db.execute("UPDATE sessions SET title_due_at = NULL WHERE id = 'ses_1'", []).unwrap();
834 prompted(&db, "ses_1", 4, None, NOW + 50 * MINUTE);
835 assert_eq!(titled(&db, "ses_1").2, None);
836 prompted(&db, "ses_1", 1, None, NOW + 51 * MINUTE);
837 assert_eq!(titled(&db, "ses_1").2, Some(rfc3339(NOW + 51 * MINUTE)));
838 // Settled for good as it ended: nothing makes it due but ending again.
839 db.execute("UPDATE sessions SET title_due_at = NULL, title_final = 1, state = 'ended' WHERE id = 'ses_1'", []).unwrap();
840 db.execute(&super::titles::linked_sql(), params!["ses_1", rfc3339(NOW + 90 * MINUTE)]).unwrap();
841 assert_eq!(titled(&db, "ses_1").2, None);
842 // Brought back, it is named again as it ends again.
843 heartbeat_titled(&db, "ses_1", "x", "agent", NOW + 91 * MINUTE);
844 let kept: u32 = db.query_row("SELECT title_final FROM sessions WHERE id = 'ses_1'", [], |row| row.get(0)).unwrap();
845 assert_eq!(kept, 0);
846}
847
848#[test]
849fn a_session_timing_out_is_named_a_last_time_its_subagents_not() {
850 let db = database();
851 start(&db, "ses_1", Some("a"));
852 start_child(&db, "ses_c", "ses_1", "a/agent/1", "Explore");
853 let late = rfc3339(NOW + g1t_contracts::work::SESSION_TIMEOUT_MS);
854 for id in ["ses_1", "ses_c"] {
855 db.prepare(&time_out_sessions_sql()).unwrap().query_map(params!["rep_1", late, id], |_| Ok(())).unwrap().count();
856 }
857 assert_eq!(titled(&db, "ses_1").2, Some(rfc3339(NOW)));
858 assert_eq!(titled(&db, "ses_c").2, None);
859}
860
861#[test]
862fn sessions_from_before_titles_settled_are_settled_once() {
863 let db = crate::migrated::database_before("0051_session_titles.sql");
864 let at = rfc3339(NOW);
865 db.execute(
866 "INSERT INTO sessions (id, repo_id, actor_id, actor_name, harness, title, state, started_at, last_seen_at, ended_at)
867 VALUES ('ses_old', 'rep_1', 'usr_1', 'ana', 'claude_code', 'This project used to be in syntaqx/g1t', 'ended', ?1, ?1, ?1)",
868 [at.as_str()],
869 )
870 .unwrap();
871 db.execute(
872 "INSERT INTO sessions (id, repo_id, actor_id, actor_name, harness, state, started_at, last_seen_at, parent_id)
873 VALUES ('ses_sub', 'rep_1', 'usr_1', 'ana', 'claude_code', 'ended', ?1, ?1, 'ses_old')",
874 [at.as_str()],
875 )
876 .unwrap();
877 for n in 1..=2u32 {
878 db.execute(
879 "INSERT INTO session_entries (seq, kind, text, at, session_id, session_seq) VALUES (?1, 'prompt', 'hi', ?2, 'ses_old', ?1)",
880 params![n, at],
881 )
882 .unwrap();
883 }
884 crate::migrated::migrate_from(&db, "0051_session_titles.sql");
885 let (prompts, due): (u32, Option<String>) =
886 db.query_row("SELECT prompts, title_due_at FROM sessions WHERE id = 'ses_old'", [], |row| Ok((row.get(0)?, row.get(1)?))).unwrap();
887 assert_eq!((prompts, due), (2, Some(at)));
888 assert_eq!(titled(&db, "ses_sub").2, None);
889}