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