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
377 linesCodeBlameRaw
1//! Coordinating work between agents and people
2//! (`g1t_contracts::work::coordination`): sessions of their own harnesses
3//! on a repository (sessions.rs), advisory claims on work items
4//! (claims.rs), what a work item is blocked by (dependencies.rs), pull
5//! requests that start changing the same files (overlaps.rs), and the one
6//! call that gathers it all for a work item (context.rs).
7//!
8//! Nothing here is a lock and nothing waits on a harness: sessions and
9//! claims that went quiet are found when the repository's coordination is
10//! next touched (`sweep`), and announced then as ended or expired.
11
12pub(crate) mod updates;
13mod claims;
14mod context;
15mod dependencies;
16mod links;
17mod overlaps;
18mod sessions;
19pub(crate) mod shown;
20#[cfg(test)]
21mod tests;
22
23use g1t_contracts::credentials::Principal;
24use g1t_contracts::repos::{Repo, RepoPath};
25use g1t_contracts::time::{parse_rfc3339, rfc3339};
26use g1t_contracts::work::*;
27use g1t_contracts::{FailureCode, Outcome, User, Viewer};
28use g1t_kit::now_ms;
29use serde::Deserialize;
30use worker::Result;
31
32use crate::Work;
33
34/// Unwraps an `Outcome`, returning its failure from the enclosing method.
35macro_rules! unwrap_or_fail {
36 ($outcome:expr) => {
37 match $outcome {
38 g1t_contracts::Outcome::Ok(value) => value,
39 g1t_contracts::Outcome::Fail(failure) => return Ok(g1t_contracts::Outcome::Fail(failure)),
40 }
41 };
42}
43pub(crate) use unwrap_or_fail;
44
45/// A session's columns, with its work item's number beside its id.
46const SESSION_COLUMNS: &str = concat!(
47 "id, repo_id, item_id,
48 ",
49 crate::items::sql::item_number!("sessions.item_id"),
50 " AS item,
51 pull, branch, actor_id, actor_name, harness, harness_session_id,
52 title, state, head_commit, started_at, last_seen_at, ended_at, end_reason,
53 parent_id, agent_type, task, prompt, result, model"
54);
55
56/// A claim's columns, with its work item's workspace and number.
57const CLAIM_COLUMNS: &str = concat!(
58 "id, item_id, repo_id,
59 ",
60 crate::items::sql::item_workspace!("claims.item_id"),
61 " AS workspace,
62 ",
63 crate::items::sql::item_number!("claims.item_id"),
64 " AS item,
65 session_id, actor_id, actor_name, harness, note, state,
66 ttl_minutes, created_at, expires_at, ended_at, end_reason, handoff_to"
67);
68
69#[derive(Clone, Deserialize)]
70pub(crate) struct SessionRow {
71 id: String,
72 repo_id: String,
73 #[serde(default)]
74 item_id: Option<String>,
75 #[serde(default)]
76 item: Option<u32>,
77 pull: Option<u32>,
78 branch: Option<String>,
79 actor_id: String,
80 actor_name: String,
81 harness: String,
82 harness_session_id: Option<String>,
83 title: Option<String>,
84 state: String,
85 head_commit: Option<String>,
86 started_at: String,
87 last_seen_at: String,
88 ended_at: Option<String>,
89 end_reason: Option<String>,
90 #[serde(default)]
91 parent_id: Option<String>,
92 #[serde(default)]
93 agent_type: Option<String>,
94 #[serde(default)]
95 task: Option<String>,
96 #[serde(default)]
97 prompt: Option<String>,
98 #[serde(default)]
99 result: Option<String>,
100 #[serde(default)]
101 model: Option<String>,
102}
103
104impl SessionRow {
105 /// The session as it reads at `now`.
106 pub(crate) fn at(self, now: u64) -> Session {
107 let last_seen = parse_rfc3339(&self.last_seen_at).unwrap_or(now);
108 Session {
109 state: SessionState::at(self.state == "ended", last_seen, now),
110 id: self.id,
111 repo_id: self.repo_id,
112 item: self.item,
113 pull: self.pull,
114 branch: self.branch,
115 actor: Principal { id: self.actor_id, username: self.actor_name },
116 harness: Harness::parse(&self.harness).unwrap_or(Harness::Other),
117 harness_session_id: self.harness_session_id,
118 title: self.title,
119 head_commit: self.head_commit,
120 started_at: self.started_at,
121 last_seen_at: self.last_seen_at,
122 ended_at: self.ended_at,
123 end_reason: self.end_reason,
124 parent_id: self.parent_id,
125 agent_type: self.agent_type,
126 task: self.task,
127 prompt: self.prompt,
128 result: self.result,
129 model: self.model,
130 subagents: 0,
131 first_prompt: None,
132 }
133 }
134}
135
136#[derive(Deserialize)]
137pub(crate) struct ClaimRow {
138 id: String,
139 pub(crate) item_id: String,
140 #[serde(default)]
141 repo_id: Option<String>,
142 #[serde(default)]
143 workspace: Option<String>,
144 #[serde(default)]
145 item: Option<u32>,
146 session_id: Option<String>,
147 actor_id: String,
148 actor_name: String,
149 harness: Option<String>,
150 note: Option<String>,
151 state: String,
152 #[allow(dead_code)]
153 ttl_minutes: u32,
154 created_at: String,
155 expires_at: String,
156 ended_at: Option<String>,
157 end_reason: Option<String>,
158 handoff_to: Option<String>,
159}
160
161impl From<ClaimRow> for Claim {
162 fn from(row: ClaimRow) -> Self {
163 Claim {
164 id: row.id,
165 repo_id: row.repo_id,
166 workspace: row.workspace.unwrap_or_default(),
167 item: row.item.unwrap_or_default(),
168 session_id: row.session_id,
169 actor: Principal { id: row.actor_id, username: row.actor_name },
170 harness: row.harness.as_deref().and_then(Harness::parse),
171 note: row.note,
172 state: ClaimState::parse(&row.state).unwrap_or(ClaimState::Expired),
173 created_at: row.created_at,
174 expires_at: row.expires_at,
175 ended_at: row.ended_at,
176 end_reason: row.end_reason,
177 handoff_to: row.handoff_to,
178 }
179 }
180}
181
182/// Text as given, trimmed and cut to `limit` characters; none when empty.
183pub(crate) fn tidy(text: Option<&str>, limit: usize) -> Option<String> {
184 let text = text?.trim();
185 (!text.is_empty()).then(|| text.chars().take(limit).collect())
186}
187
188/// The most sessions, and claims, one sweep ends; the rest wait for the next.
189const SWEEP_LIMIT: u32 = 50;
190
191/// Ends a repository's (`?1`) session `?3` if last seen at or before `?2`,
192/// as `timed_out`, as of when it was last seen.
193pub(crate) fn time_out_sessions_sql() -> String {
194 format!(
195 "UPDATE sessions SET state = 'ended', ended_at = last_seen_at, end_reason = 'timed_out'
196 WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2 AND id = ?3
197 RETURNING {SESSION_COLUMNS}"
198 )
199}
200
201/// Expires active claim `?3` if due at or before `?2` (`?1` unused, so
202/// the binds are the sessions').
203pub(crate) fn expire_claims_sql() -> String {
204 format!(
205 "UPDATE claims SET state = 'expired', ended_at = expires_at, end_reason = 'expired'
206 WHERE ?1 IS NOT NULL AND state = 'active' AND expires_at <= ?2 AND id = ?3
207 RETURNING {CLAIM_COLUMNS}"
208 )
209}
210
211/// RFC 3339 for `minutes` after `now`.
212pub(crate) fn after_minutes(now: u64, minutes: u32) -> String {
213 rfc3339(now + u64::from(minutes) * 60_000)
214}
215
216impl Work {
217 /// The repository as `viewer` sees it, for reading its coordination.
218 pub(crate) async fn coordination_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
219 let repo = unwrap_or_fail!(self.repo(path, viewer).await?);
220 self.sweep(&repo.id).await?;
221 Ok(Outcome::Ok(repo))
222 }
223
224 /// The repository, if `actor` may take part in it: a confirmed email
225 /// address, a repository that is not archived, and a role that may
226 /// open work items and pull requests.
227 pub(crate) async fn participating_repo(&self, path: &RepoPath, actor: &User) -> Result<Outcome<Repo>> {
228 if !actor.verified {
229 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
230 }
231 let repo = unwrap_or_fail!(self.repo(path, &Some(actor.clone())).await?);
232 unwrap_or_fail!(crate::retired::writable(&repo));
233 unwrap_or_fail!(crate::allowed(Some(actor), &repo, g1t_contracts::access::Capability::Participate));
234 self.sweep(&repo.id).await?;
235 Ok(Outcome::Ok(repo))
236 }
237
238 /// Ends the repository's sessions that have gone 30 minutes without a
239 /// heartbeat (`timed_out`) and expires the claims made from it or on
240 /// work items linked to it past their time, and announces each. Run
241 /// before the repository's coordination is read or changed, so that
242 /// nobody sees a stale one as live.
243 pub(crate) async fn sweep(&self, repo_id: &str) -> Result<()> {
244 let now = now_ms();
245 let quiet_since = rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS));
246 let at = rfc3339(now);
247 // Read first: most of the time nothing is due, and nothing is written.
248 let due = self
249 .db
250 .batch(vec![
251 self.db
252 .prepare(format!(
253 "SELECT {SESSION_COLUMNS} FROM sessions
254 WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2 LIMIT ?3"
255 ))
256 .bind(&[repo_id.into(), quiet_since.as_str().into(), SWEEP_LIMIT.into()])?,
257 self.db
258 .prepare(format!(
259 "SELECT {CLAIM_COLUMNS} FROM claims
260 WHERE state = 'active' AND expires_at <= ?2 AND (repo_id = ?1 OR item_id IN {linked})
261 LIMIT ?3",
262 linked = crate::items::sql::items_linked_to!("?1"),
263 ))
264 .bind(&[repo_id.into(), at.as_str().into(), SWEEP_LIMIT.into()])?,
265 ])
266 .await?;
267 let mut due = due.into_iter();
268 let quiet: Vec<SessionRow> = match due.next() {
269 Some(result) => result.results()?,
270 None => Vec::new(),
271 };
272 let lapsed: Vec<ClaimRow> = match due.next() {
273 Some(result) => result.results()?,
274 None => Vec::new(),
275 };
276 if quiet.is_empty() && lapsed.is_empty() {
277 return Ok(());
278 }
279 // Each one's end and its updates are one change: a sweep running at
280 // the same moment ends it, and delivers it, once.
281 let mut statements = Vec::new();
282 let mut sessions = Vec::new();
283 for row in quiet {
284 statements.push(
285 self.db
286 .prepare(time_out_sessions_sql())
287 .bind(&[repo_id.into(), quiet_since.as_str().into(), row.id.as_str().into()])?,
288 );
289 let index = statements.len() - 1;
290 let ended_at = row.last_seen_at.clone();
291 let expected = SessionRow {
292 state: "ended".to_owned(),
293 ended_at: Some(ended_at.clone()),
294 end_reason: Some("timed_out".to_owned()),
295 ..row
296 }
297 .at(now);
298 let delivery = self.delivery(
299 "session.ended",
300 repo_id,
301 serde_json::to_value(SessionEvent::of(&expected))?,
302 &format!("{}:ended:{ended_at}", expected.id),
303 "EXISTS (SELECT 1 FROM sessions WHERE id = ?7 AND state = 'ended' AND ended_at = ?8)",
304 vec![expected.id.as_str().into(), ended_at.as_str().into()],
305 )?;
306 sessions.push((index, updates::push_delivery(&mut statements, delivery)));
307 }
308 let mut claims = Vec::new();
309 for row in lapsed {
310 statements.push(
311 self.db.prepare(expire_claims_sql()).bind(&[repo_id.into(), at.as_str().into(), row.id.as_str().into()])?,
312 );
313 let index = statements.len() - 1;
314 let held = Claim::from(row);
315 let expected = Claim {
316 state: ClaimState::Expired,
317 ended_at: Some(held.expires_at.clone()),
318 end_reason: Some("expired".to_owned()),
319 ..held
320 };
321 let delivery = self.delivery(
322 "work.claim_expired",
323 repo_id,
324 serde_json::to_value(ClaimEvent::of(&expected, &[repo_id.to_owned()]))?,
325 &format!("{}:{}", expected.id, ClaimState::Expired.as_str()),
326 "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = 'expired')",
327 vec![expected.id.as_str().into()],
328 )?;
329 claims.push((index, updates::push_delivery(&mut statements, delivery)));
330 }
331 let results = self.db.batch(statements).await?;
332 let returned = |index: usize| results.get(index);
333 for (index, sent) in sessions {
334 let Some(row) = returned(index).map(|result| result.results::<SessionRow>()).transpose()?.and_then(|rows| rows.into_iter().next())
335 else {
336 continue;
337 };
338 let session = row.at(now);
339 self.delivered(sent, "session.ended", repo_id, None, serde_json::to_value(SessionEvent::of(&session))?).await?;
340 }
341 for (index, sent) in claims {
342 let Some(row) = returned(index).map(|result| result.results::<ClaimRow>()).transpose()?.and_then(|rows| rows.into_iter().next())
343 else {
344 continue;
345 };
346 let claim = Claim::from(row);
347 let data = serde_json::to_value(ClaimEvent::of(&claim, &[repo_id.to_owned()]))?;
348 self.delivered(sent, "work.claim_expired", repo_id, None, data).await?;
349 }
350 Ok(())
351 }
352
353 pub(crate) async fn session_row(&self, repo_id: &str, id: &str) -> Result<Option<SessionRow>> {
354 self.db
355 .prepare(format!("SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id = ?"))
356 .bind(&[repo_id.into(), id.into()])?
357 .first::<SessionRow>(None)
358 .await
359 }
360
361 /// Active claims on a work item, oldest first.
362 pub(crate) async fn active_claims(&self, item_id: &str) -> Result<Vec<Claim>> {
363 Ok(self
364 .db
365 .prepare(format!(
366 "SELECT {CLAIM_COLUMNS} FROM claims
367 WHERE item_id = ? AND state = 'active' ORDER BY created_at, id"
368 ))
369 .bind(&[item_id.into()])?
370 .all()
371 .await?
372 .results::<ClaimRow>()?
373 .into_iter()
374 .map(Claim::from)
375 .collect())
376 }
377}