Skip to content
Sign up

flagon-io/g1t

public
0

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
347 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 super::{CLAIM_COLUMNS, ClaimRow, after_minutes, tidy, unwrap_or_fail};
17use crate::Work;
18
19/// The most claims one list returns.
20const LIST_LIMIT: u32 = 100;
21
22fn text_or_null(value: &Option<String>) -> JsValue {
23 value.as_deref().map_or(JsValue::NULL, JsValue::from)
24}
25
26impl Work {
27 /// The caller's active claim on an issue, if they hold one.
28 async fn own_claim(&self, repo_id: &str, issue: u32, actor: &User, session: Option<&str>) -> Result<Option<Claim>> {
29 Ok(self
30 .active_claims(repo_id, issue)
31 .await?
32 .into_iter()
33 .find(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &actor.id, session)))
34 }
35
36 /// `claim_issue`: the issue named, or with a `title` and no number, a
37 /// new one opened for it, with the claim's session attached to it.
38 pub(crate) async fn claim_issue(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
39 if a.number == 0 {
40 return self.open_and_claim(a).await;
41 }
42 self.claim_number(a).await
43 }
44
45 /// Opens an issue from `title` and `body`, claims it, and attaches the
46 /// claim's session to it. The session is checked before anything is
47 /// opened, so a claim that would be refused opens no issue.
48 async fn open_and_claim(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
49 let Some(title) = a.title.as_deref().map(str::trim).filter(|title| !title.is_empty()).map(str::to_owned) else {
50 return Ok(Outcome::fail(FailureCode::Invalid, "Name the issue by its number, or give a title (and body) to open one."));
51 };
52 let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?);
53 let session_id = a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()).map(str::to_owned);
54 if let Some(id) = &session_id {
55 match self.session_row(&repo.id, id).await? {
56 Some(row) if row.actor_id == a.actor.id => {
57 if row.at(now_ms()).state == SessionState::Ended {
58 return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
59 }
60 }
61 _ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
62 }
63 }
64 let opened = unwrap_or_fail!(
65 self.open_issue(OpenIssueArgs {
66 actor: a.actor.clone(),
67 repo: a.repo.clone(),
68 title,
69 body: a.body.clone().unwrap_or_default(),
70 labels: Vec::new(),
71 checks: Vec::new(),
72 milestone: None,
73 })
74 .await?
75 );
76 let claimed = unwrap_or_fail!(
77 self.claim_number(ClaimArgs { number: opened.number, title: None, body: None, session_id: session_id.clone(), ..a.clone() })
78 .await?
79 );
80 // The session is on the issue now, not only tied to it by the claim.
81 if let Some(id) = session_id {
82 let attached = self
83 .update_session(UpdateSessionArgs {
84 actor: a.actor.clone(),
85 repo: a.repo.clone(),
86 id,
87 issue: Some(opened.number),
88 pull: None,
89 branch: None,
90 head_commit: None,
91 title: None,
92 agent_type: None,
93 task: None,
94 prompt: None,
95 model: None,
96 result: None,
97 })
98 .await?;
99 if let Outcome::Fail(failure) = attached {
100 worker::console_error!("#{}: the session was not attached: {}", opened.number, failure.message);
101 }
102 }
103 Ok(Outcome::Ok(Claimed { issue: Some(opened), ..claimed }))
104 }
105
106 async fn claim_number(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
107 let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?);
108 let Some(issue) = self.issue(&repo.id, a.number).await? else {
109 return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
110 };
111 if issue.state == State::Closed {
112 return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", issue.number)));
113 }
114 // The session it is made from, whose heartbeats renew it.
115 let session = match a.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) {
116 Some(id) => match self.session_row(&repo.id, id).await? {
117 Some(row) if row.actor_id == a.actor.id => Some(row.at(now_ms())),
118 _ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
119 },
120 None => None,
121 };
122 if session.as_ref().is_some_and(|session| session.state == SessionState::Ended) {
123 return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
124 }
125 let minutes = claim_minutes(a.ttl_minutes);
126 let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS);
127 let session_id = session.as_ref().map(|session| session.id.clone());
128 let harness = session.as_ref().map(|session| session.harness.as_str().to_owned());
129 let now = now_ms();
130 let expires_at = after_minutes(now, minutes);
131 let standing = self.active_claims(&repo.id, issue.number).await?;
132 let (mine, others): (Vec<Claim>, Vec<Claim>) = standing
133 .into_iter()
134 .partition(|claim| claim_is_mine(&claim.actor.id, claim.session_id.as_deref(), &a.actor.id, session_id.as_deref()));
135 let blocked_by = self.blockers(&repo.id, &issue.blocked_by).await?;
136 // Claiming what you hold renews it.
137 if let Some(held) = mine.into_iter().next() {
138 let renewed = self
139 .db
140 .prepare(format!(
141 "UPDATE claims
142 SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note),
143 session_id = COALESCE(?, session_id), harness = COALESCE(?, harness)
144 WHERE id = ? AND state = 'active'
145 RETURNING {CLAIM_COLUMNS}"
146 ))
147 .bind(&[
148 expires_at.as_str().into(),
149 minutes.into(),
150 text_or_null(&note),
151 text_or_null(&session_id),
152 text_or_null(&harness),
153 held.id.as_str().into(),
154 ])?
155 .first::<ClaimRow>(None)
156 .await?;
157 if let Some(renewed) = renewed {
158 if let Some(session_id) = &session_id {
159 self.link(session_id, &repo.id, LinkKind::Issue, issue.number, LinkVia::Claim).await;
160 }
161 return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by, issue: None }));
162 }
163 }
164 // Taking over what was handed off to you, or to anyone.
165 let handed: Option<ClaimRow> = self
166 .db
167 .prepare(format!(
168 "SELECT {CLAIM_COLUMNS} FROM claims
169 WHERE repo_id = ? AND issue = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1"
170 ))
171 .bind(&[repo.id.as_str().into(), issue.number.into()])?
172 .first::<ClaimRow>(None)
173 .await?;
174 let took_over = handed
175 .map(Claim::from)
176 .filter(|last| last.state == ClaimState::HandedOff)
177 .filter(|last| last.handoff_to.as_deref().is_none_or(|to| to == a.actor.username))
178 .map(|last| last.id);
179 let at = rfc3339(now);
180 let created = self
181 .db
182 .prepare(format!(
183 "INSERT INTO claims
184 (id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state, ttl_minutes,
185 created_at, expires_at)
186 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)
187 RETURNING {CLAIM_COLUMNS}"
188 ))
189 .bind(&[
190 new_id("clm", now).into(),
191 repo.id.as_str().into(),
192 issue.number.into(),
193 text_or_null(&session_id),
194 a.actor.id.as_str().into(),
195 a.actor.username.as_str().into(),
196 text_or_null(&harness),
197 text_or_null(&note),
198 minutes.into(),
199 at.as_str().into(),
200 expires_at.as_str().into(),
201 ])?
202 .first::<ClaimRow>(None)
203 .await?;
204 let Some(created) = created else {
205 return Err(worker::Error::RustError("the claim was not recorded".into()));
206 };
207 let claim = Claim::from(created);
208 if let Some(session_id) = &claim.session_id {
209 self.link(session_id, &repo.id, LinkKind::Issue, claim.issue, LinkVia::Claim).await;
210 }
211 self.publish(
212 "work.claimed",
213 &repo.id,
214 &a.actor,
215 ClaimEvent { others: Some(others.clone()), took_over, ..ClaimEvent::of(&claim) },
216 )
217 .await?;
218 Ok(Outcome::Ok(Claimed { claim, others, blocked_by, issue: None }))
219 }
220
221 /// Ends the caller's active claim on an issue as `state`, or refuses
222 /// when they hold none.
223 async fn end_own_claim(
224 &self,
225 actor: &User,
226 repo: &g1t_contracts::repos::RepoPath,
227 number: u32,
228 state: ClaimState,
229 reason: &str,
230 handoff: Option<(&Option<String>, &str)>,
231 session: Option<&str>,
232 ) -> Result<Outcome<Claim>> {
233 let repo = unwrap_or_fail!(self.participating_repo(repo, actor).await?);
234 let Some(held) = self.own_claim(&repo.id, number, actor, session.map(str::trim).filter(|id| !id.is_empty())).await? else {
235 return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
236 };
237 let at = rfc3339(now_ms());
238 let (handoff_to, note) = match handoff {
239 Some((to, note)) => (text_or_null(to), JsValue::from(note)),
240 None => (JsValue::NULL, JsValue::NULL),
241 };
242 let ended = self
243 .db
244 .prepare(format!(
245 "UPDATE claims
246 SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note)
247 WHERE id = ? AND state = 'active'
248 RETURNING {CLAIM_COLUMNS}"
249 ))
250 .bind(&[state.as_str().into(), at.into(), reason.into(), handoff_to, note, held.id.as_str().into()])?
251 .first::<ClaimRow>(None)
252 .await?;
253 Ok(match ended {
254 Some(row) => Outcome::Ok(row.into()),
255 None => Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")),
256 })
257 }
258
259 pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> {
260 let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "released".to_owned());
261 let claim = unwrap_or_fail!(
262 self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::Released, &reason, None, a.session_id.as_deref()).await?
263 );
264 self.publish(
265 "work.released",
266 &claim.repo_id.clone(),
267 &a.actor,
268 ClaimEvent { reason: Some(reason), ..ClaimEvent::of(&claim) },
269 )
270 .await?;
271 Ok(Outcome::Ok(claim))
272 }
273
274 pub(crate) async fn hand_off_issue(&self, a: HandOffArgs) -> Result<Outcome<Claim>> {
275 let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else {
276 return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up."));
277 };
278 let to = tidy(a.to.as_deref().map(|to| to.trim().trim_start_matches('@')), 100).map(|to| to.to_lowercase());
279 if let Some(name) = &to {
280 let account: Viewer =
281 g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?;
282 if account.is_none() {
283 return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}.")));
284 }
285 }
286 let claim = unwrap_or_fail!(
287 self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::HandedOff, "handed_off", Some((&to, &note)), a.session_id.as_deref())
288 .await?
289 );
290 self.publish(
291 "work.handoff_requested",
292 &claim.repo_id.clone(),
293 &a.actor,
294 ClaimEvent { from: Some(a.actor.username.clone()), to, note: Some(note), ..ClaimEvent::of(&claim) },
295 )
296 .await?;
297 Ok(Outcome::Ok(claim))
298 }
299
300 pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> {
301 let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
302 let state = a.state.unwrap_or(ClaimState::Active);
303 let rows = self
304 .db
305 .prepare(format!(
306 "SELECT {CLAIM_COLUMNS} FROM claims
307 WHERE repo_id = ?1 AND state = ?2 AND (?3 IS NULL OR issue = ?3)
308 ORDER BY created_at DESC, id DESC LIMIT ?4"
309 ))
310 .bind(&[
311 repo.id.as_str().into(),
312 state.as_str().into(),
313 a.issue.filter(|number| *number > 0).map_or(JsValue::NULL, JsValue::from),
314 LIST_LIMIT.into(),
315 ])?
316 .all()
317 .await?
318 .results::<ClaimRow>()?;
319 Ok(Outcome::Ok(rows.into_iter().map(Claim::from).collect()))
320 }
321
322 /// An issue closed: its active claims end as released (`issue_closed`),
323 /// each announced.
324 pub(crate) async fn release_claims_on_close(&self, repo_id: &str, issue: u32, actor: Option<&User>) -> Result<()> {
325 let at = rfc3339(now_ms());
326 let released = self
327 .db
328 .prepare(format!(
329 "UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'issue_closed'
330 WHERE repo_id = ? AND issue = ? AND state = 'active'
331 RETURNING {CLAIM_COLUMNS}"
332 ))
333 .bind(&[at.into(), repo_id.into(), issue.into()])?
334 .all()
335 .await?
336 .results::<ClaimRow>()?;
337 for row in released {
338 let claim = Claim::from(row);
339 let data = ClaimEvent { reason: Some("issue_closed".to_owned()), ..ClaimEvent::of(&claim) };
340 match actor {
341 Some(actor) => self.publish("work.released", repo_id, actor, data).await?,
342 None => self.publish_as("work.released", repo_id, None, data).await?,
343 }
344 }
345 Ok(())
346 }
347}