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
472 linesCodeBlameRaw
1//! Claims: an advisory, expiring "I'm working on #N".
2//!
3//! A claim is never a lock: anyone may claim an issue someone else has
4//! claimed, and is told who else is on it. It lapses unless renewed, by
5//! claiming again or by a heartbeat of the session it was made from, and
6//! ends when released, handed off, or when its issue closes.
7
8use g1t_contracts::identity::UsernameArgs;
9use g1t_contracts::time::rfc3339;
10use g1t_contracts::work::*;
11use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
12use g1t_kit::now_ms;
13use worker::Result;
14use worker::wasm_bindgen::JsValue;
15
16use g1t_contracts::credentials::Principal;
17
18use super::updates::push_delivery;
19use super::{CLAIM_COLUMNS, ClaimRow, after_minutes, tidy, unwrap_or_fail};
20use crate::Work;
21
22/// The most claims one list returns.
23const LIST_LIMIT: u32 = 100;
24
25/// Resolve ownership and insert/renew in one SQLite statement. In particular,
26/// two requests that both read no claim must converge on the same row.
27/// Sessionless claims retain the caller-wide semantics of claim_is_mine.
28pub(super) fn upsert_claim_sql() -> String {
29 format!(
30 "INSERT INTO claims
31 (id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state, ttl_minutes,
32 created_at, expires_at)
33 VALUES (COALESCE((SELECT id FROM claims
34 WHERE repo_id = ?2 AND issue = ?3 AND actor_id = ?5 AND state = 'active'
35 AND (session_id IS NULL OR ?4 IS NULL OR session_id = ?4)
36 ORDER BY created_at, id LIMIT 1), ?1),
37 ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'active', ?9, ?10, ?11)
38 ON CONFLICT(id) DO UPDATE SET
39 expires_at = MAX(claims.expires_at, excluded.expires_at), ttl_minutes = excluded.ttl_minutes,
40 note = COALESCE(excluded.note, claims.note),
41 session_id = COALESCE(excluded.session_id, claims.session_id),
42 harness = COALESCE(excluded.harness, claims.harness)
43 WHERE claims.state = 'active'
44 RETURNING {CLAIM_COLUMNS}"
45 )
46}
47
48fn text_or_null(value: &Option<String>) -> JsValue {
49 value.as_deref().map_or(JsValue::NULL, JsValue::from)
50}
51
52impl Work {
53 /// The caller's active claim on an issue, if they hold one.
54 async fn own_claim(&self, repo_id: &str, issue: u32, actor: &User, session: Option<&str>) -> Result<Option<Claim>> {
55 Ok(self
56 .active_claims(repo_id, issue)
57 .await?
58 .into_iter()
59 .find(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &actor.id, session)))
60 }
61
62 /// `claim_issue`: the issue named, or with a `title` and no number, a
63 /// new one opened for it, with the claim's session attached to it.
64 pub(crate) async fn claim_issue(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
65 if a.number == 0 {
66 return self.open_and_claim(a).await;
67 }
68 self.claim_number(a).await
69 }
70
71 /// Opens an issue from `title` and `body`, claims it, and attaches the
72 /// claim's session to it. The session is checked before anything is
73 /// opened, so a claim that would be refused opens no issue.
74 async fn open_and_claim(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
75 let Some(title) = a.title.as_deref().map(str::trim).filter(|title| !title.is_empty()).map(str::to_owned) else {
76 return Ok(Outcome::fail(FailureCode::Invalid, "Name the issue by its number, or give a title (and body) to open one."));
77 };
78 let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?);
79 let session_id = a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()).map(str::to_owned);
80 if let Some(id) = &session_id {
81 match self.session_row(&repo.id, id).await? {
82 Some(row) if row.actor_id == a.actor.id => {
83 if row.at(now_ms()).state == SessionState::Ended {
84 return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
85 }
86 }
87 _ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
88 }
89 }
90 let opened = unwrap_or_fail!(
91 self.open_issue(OpenIssueArgs {
92 actor: a.actor.clone(),
93 repo: a.repo.clone(),
94 title,
95 body: a.body.clone().unwrap_or_default(),
96 labels: Vec::new(),
97 checks: Vec::new(),
98 milestone: None,
99 })
100 .await?
101 );
102 let claimed = unwrap_or_fail!(
103 self.claim_number(ClaimArgs { number: opened.number, title: None, body: None, session_id: session_id.clone(), ..a.clone() })
104 .await?
105 );
106 // The session is on the issue now, not only tied to it by the claim.
107 if let Some(id) = session_id {
108 let attached = self
109 .update_session(UpdateSessionArgs {
110 actor: a.actor.clone(),
111 repo: a.repo.clone(),
112 id,
113 issue: Some(opened.number),
114 pull: None,
115 branch: None,
116 head_commit: None,
117 title: None,
118 agent_type: None,
119 task: None,
120 prompt: None,
121 model: None,
122 result: None,
123 })
124 .await?;
125 if let Outcome::Fail(failure) = attached {
126 worker::console_error!("#{}: the session was not attached: {}", opened.number, failure.message);
127 }
128 }
129 Ok(Outcome::Ok(Claimed { issue: Some(opened), ..claimed }))
130 }
131
132 async fn claim_number(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
133 let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?);
134 let Some(issue) = self.issue(&repo.id, a.number).await? else {
135 return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
136 };
137 if issue.state == State::Closed {
138 return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", issue.number)));
139 }
140 // The session it is made from, whose heartbeats renew it.
141 let session = match a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) {
142 Some(id) => match self.session_row(&repo.id, id).await? {
143 Some(row) if row.actor_id == a.actor.id => Some(row.at(now_ms())),
144 _ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
145 },
146 None => None,
147 };
148 if session.as_ref().is_some_and(|session| session.state == SessionState::Ended) {
149 return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
150 }
151 let minutes = claim_minutes(a.ttl_minutes);
152 let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS);
153 let session_id = session.as_ref().map(|session| session.id.clone());
154 let harness = session.as_ref().map(|session| session.harness.as_str().to_owned());
155 let now = now_ms();
156 let expires_at = after_minutes(now, minutes);
157 let standing = self.active_claims(&repo.id, issue.number).await?;
158 let (mine, others): (Vec<Claim>, Vec<Claim>) = standing
159 .into_iter()
160 .partition(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &a.actor.id, session_id.as_deref()));
161 let blocked_by = self.blockers(&repo.id, &issue.blocked_by).await?;
162 // Claiming what you hold renews it.
163 if let Some(held) = mine.into_iter().next() {
164 let renewed = self
165 .db
166 .prepare(format!(
167 "UPDATE claims
168 SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note),
169 session_id = COALESCE(?, session_id), harness = COALESCE(?, harness)
170 WHERE id = ? AND state = 'active'
171 RETURNING {CLAIM_COLUMNS}"
172 ))
173 .bind(&[
174 expires_at.as_str().into(),
175 minutes.into(),
176 text_or_null(&note),
177 text_or_null(&session_id),
178 text_or_null(&harness),
179 held.id.as_str().into(),
180 ])?
181 .first::<ClaimRow>(None)
182 .await?;
183 if let Some(renewed) = renewed {
184 if let Some(session_id) = &session_id {
185 self.link(session_id, &repo.id, LinkKind::Issue, issue.number, LinkVia::Claim).await;
186 }
187 return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by, issue: None }));
188 }
189 }
190 // Taking over what was handed off to you, or to anyone.
191 let handed: Option<ClaimRow> = self
192 .db
193 .prepare(format!(
194 "SELECT {CLAIM_COLUMNS} FROM claims
195 WHERE repo_id = ? AND issue = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1"
196 ))
197 .bind(&[repo.id.as_str().into(), issue.number.into()])?
198 .first::<ClaimRow>(None)
199 .await?;
200 let took_over = handed
201 .map(Claim::from)
202 .filter(|last| last.state == ClaimState::HandedOff)
203 .filter(|last| last.handoff_to.as_deref().is_none_or(|to| to == a.actor.username))
204 .map(|last| last.id);
205 let at = rfc3339(now);
206 let proposed_id = new_id("clm", now);
207 // A new claim is announced with itself; a renewal (the claim found
208 // is an older one of the caller's) is not.
209 let proposed = Claim {
210 id: proposed_id.clone(),
211 repo_id: repo.id.clone(),
212 issue: issue.number,
213 session_id: session_id.clone(),
214 actor: Principal { id: a.actor.id.clone(), username: a.actor.username.clone() },
215 harness: harness.as_deref().and_then(Harness::parse),
216 note: note.clone(),
217 state: ClaimState::Active,
218 created_at: at.clone(),
219 expires_at: expires_at.clone(),
220 ended_at: None,
221 end_reason: None,
222 handoff_to: None,
223 };
224 let event = |claim: &Claim| {
225 g1t_contracts::events::marked(
226 ClaimEvent { others: Some(others.clone()), took_over: took_over.clone(), ..ClaimEvent::of(claim) },
227 Some(&a.actor),
228 )
229 };
230 let mut statements = vec![self
231 .db
232 .prepare(upsert_claim_sql())
233 .bind(&[
234 proposed_id.as_str().into(),
235 repo.id.as_str().into(),
236 issue.number.into(),
237 text_or_null(&session_id),
238 a.actor.id.as_str().into(),
239 a.actor.username.as_str().into(),
240 text_or_null(&harness),
241 text_or_null(&note),
242 minutes.into(),
243 at.as_str().into(),
244 expires_at.as_str().into(),
245 ])?];
246 let sent = push_delivery(
247 &mut statements,
248 self.delivery(
249 "work.claimed",
250 &repo.id,
251 event(&proposed),
252 &format!("{proposed_id}:claimed"),
253 "EXISTS (SELECT 1 FROM claims WHERE id = ?7)",
254 vec![proposed_id.as_str().into()],
255 )?,
256 );
257 let created = match self.db.batch(statements).await?.into_iter().next() {
258 Some(result) => result.results::<ClaimRow>()?.into_iter().next(),
259 None => None,
260 };
261 let Some(created) = created else {
262 return Err(worker::Error::RustError("the claim was not recorded".into()));
263 };
264 let claim = Claim::from(created);
265 if let Some(session_id) = &claim.session_id {
266 self.link(session_id, &repo.id, LinkKind::Issue, claim.issue, LinkVia::Claim).await;
267 }
268 if claim.id != proposed_id {
269 return Ok(Outcome::Ok(Claimed { claim, others, blocked_by, issue: None }));
270 }
271 self.delivered(sent, "work.claimed", &repo.id, Some(a.actor.id.clone()), event(&claim)).await?;
272 Ok(Outcome::Ok(Claimed { claim, others, blocked_by, issue: None }))
273 }
274
275 /// Ends the caller's active claim on an issue as `state`, announced as
276 /// `kind` with the event `event` makes of it, or refuses when they hold
277 /// none.
278 #[allow(clippy::too_many_arguments)]
279 async fn end_own_claim(
280 &self,
281 actor: &User,
282 repo: &g1t_contracts::repos::RepoPath,
283 number: u32,
284 state: ClaimState,
285 reason: &str,
286 handoff: Option<(&Option<String>, &str)>,
287 session: Option<&str>,
288 kind: &'static str,
289 event: impl Fn(&Claim) -> ClaimEvent,
290 ) -> Result<Outcome<Claim>> {
291 let repo = unwrap_or_fail!(self.participating_repo(repo, actor).await?);
292 let Some(held) = self.own_claim(&repo.id, number, actor, session.map(str::trim).filter(|id| !id.is_empty())).await? else {
293 return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
294 };
295 let at = rfc3339(now_ms());
296 let expected = Claim {
297 state,
298 ended_at: Some(at.clone()),
299 end_reason: Some(reason.to_owned()),
300 handoff_to: handoff.and_then(|(to, _)| to.clone()),
301 note: handoff.map(|(_, note)| note.to_owned()).or(held.note.clone()),
302 ..held.clone()
303 };
304 let (handoff_to, note) = match handoff {
305 Some((to, note)) => (text_or_null(to), JsValue::from(note)),
306 None => (JsValue::NULL, JsValue::NULL),
307 };
308 let mut statements = vec![self
309 .db
310 .prepare(format!(
311 "UPDATE claims
312 SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note)
313 WHERE id = ? AND state = 'active'
314 RETURNING {CLAIM_COLUMNS}"
315 ))
316 .bind(&[state.as_str().into(), at.as_str().into(), reason.into(), handoff_to, note, held.id.as_str().into()])?];
317 // A claim ends once: its id and how it ended name the change.
318 let sent = push_delivery(
319 &mut statements,
320 self.delivery(
321 kind,
322 &repo.id,
323 g1t_contracts::events::marked(event(&expected), Some(actor)),
324 &format!("{}:{}", held.id, state.as_str()),
325 "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = ?8)",
326 vec![held.id.as_str().into(), state.as_str().into()],
327 )?,
328 );
329 let ended = match self.db.batch(statements).await?.into_iter().next() {
330 Some(result) => result.results::<ClaimRow>()?.into_iter().next(),
331 None => None,
332 };
333 let Some(row) = ended else {
334 return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
335 };
336 let claim = Claim::from(row);
337 let data = g1t_contracts::events::marked(event(&claim), Some(actor));
338 self.delivered(sent, kind, &repo.id, Some(actor.id.clone()), data).await?;
339 Ok(Outcome::Ok(claim))
340 }
341
342 pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> {
343 let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "released".to_owned());
344 let claim = unwrap_or_fail!(
345 self.end_own_claim(
346 &a.actor,
347 &a.repo,
348 a.number,
349 ClaimState::Released,
350 &reason,
351 None,
352 a.session_id.as_deref(),
353 "work.released",
354 |claim| ClaimEvent { reason: Some(reason.clone()), ..ClaimEvent::of(claim) },
355 )
356 .await?
357 );
358 Ok(Outcome::Ok(claim))
359 }
360
361 pub(crate) async fn hand_off_issue(&self, a: HandOffArgs) -> Result<Outcome<Claim>> {
362 let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else {
363 return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up."));
364 };
365 let to = tidy(a.to.as_deref().map(|to| to.trim().trim_start_matches('@')), 100).map(|to| to.to_lowercase());
366 if let Some(name) = &to {
367 let account: Viewer =
368 g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?;
369 if account.is_none() {
370 return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}.")));
371 }
372 }
373 let claim = unwrap_or_fail!(
374 self.end_own_claim(
375 &a.actor,
376 &a.repo,
377 a.number,
378 ClaimState::HandedOff,
379 "handed_off",
380 Some((&to, &note)),
381 a.session_id.as_deref(),
382 "work.handoff_requested",
383 |claim| ClaimEvent {
384 from: Some(a.actor.username.clone()),
385 to: to.clone(),
386 note: Some(note.clone()),
387 ..ClaimEvent::of(claim)
388 },
389 )
390 .await?
391 );
392 Ok(Outcome::Ok(claim))
393 }
394
395 pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> {
396 let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
397 let state = a.state.unwrap_or(ClaimState::Active);
398 let rows = self
399 .db
400 .prepare(format!(
401 "SELECT {CLAIM_COLUMNS} FROM claims
402 WHERE repo_id = ?1 AND state = ?2 AND (?3 IS NULL OR issue = ?3)
403 ORDER BY created_at DESC, id DESC LIMIT ?4"
404 ))
405 .bind(&[
406 repo.id.as_str().into(),
407 state.as_str().into(),
408 a.issue.filter(|number| *number > 0).map_or(JsValue::NULL, JsValue::from),
409 LIST_LIMIT.into(),
410 ])?
411 .all()
412 .await?
413 .results::<ClaimRow>()?;
414 Ok(Outcome::Ok(rows.into_iter().map(Claim::from).collect()))
415 }
416
417 /// An issue closed: its active claims end as released (`issue_closed`),
418 /// each announced.
419 pub(crate) async fn release_claims_on_close(&self, repo_id: &str, issue: u32, actor: Option<&User>) -> Result<()> {
420 let at = rfc3339(now_ms());
421 let active = self.active_claims(repo_id, issue).await?;
422 if active.is_empty() {
423 return Ok(());
424 }
425 let event = |claim: &Claim| {
426 g1t_contracts::events::marked(
427 ClaimEvent { reason: Some("issue_closed".to_owned()), ..ClaimEvent::of(claim) },
428 actor,
429 )
430 };
431 // Each claim's release and its updates are one change, in one batch.
432 let mut statements = Vec::new();
433 let mut sent = Vec::new();
434 for held in &active {
435 statements.push(
436 self.db
437 .prepare(format!(
438 "UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'issue_closed'
439 WHERE id = ? AND state = 'active'
440 RETURNING {CLAIM_COLUMNS}"
441 ))
442 .bind(&[at.as_str().into(), held.id.as_str().into()])?,
443 );
444 let expected = Claim {
445 state: ClaimState::Released,
446 ended_at: Some(at.clone()),
447 end_reason: Some("issue_closed".to_owned()),
448 ..held.clone()
449 };
450 let delivery = self.delivery(
451 "work.released",
452 repo_id,
453 event(&expected),
454 &format!("{}:{}", held.id, ClaimState::Released.as_str()),
455 "EXISTS (SELECT 1 FROM claims WHERE id = ?7 AND state = 'released')",
456 vec![held.id.as_str().into()],
457 )?;
458 let index = statements.len() - 1;
459 sent.push((index, push_delivery(&mut statements, delivery)));
460 }
461 let results = self.db.batch(statements).await?;
462 for (index, sent) in sent {
463 let Some(row) = results.get(index).map(|result| result.results::<ClaimRow>()).transpose()?.and_then(|rows| rows.into_iter().next())
464 else {
465 continue;
466 };
467 let claim = Claim::from(row);
468 self.delivered(sent, "work.released", repo_id, actor.map(|actor| actor.id.clone()), event(&claim)).await?;
469 }
470 Ok(())
471 }
472}