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