Skip to content
Sign up

flagon-io/g1t

public
0
265 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
12mod claims;
13mod context;
14mod dependencies;
15mod links;
16mod overlaps;
17mod sessions;
18#[cfg(test)]
19mod tests;
20
21use g1t_contracts::credentials::Principal;
22use g1t_contracts::repos::{Repo, RepoPath};
23use g1t_contracts::time::{parse_rfc3339, rfc3339};
24use g1t_contracts::work::*;
25use g1t_contracts::{FailureCode, Outcome, User, Viewer};
26use g1t_kit::now_ms;
27use serde::Deserialize;
28use worker::Result;
29
30use crate::Work;
31
32/// Unwraps an `Outcome`, returning its failure from the enclosing method.
33macro_rules! unwrap_or_fail {
34 ($outcome:expr) => {
35 match $outcome {
36 g1t_contracts::Outcome::Ok(value) => value,
37 g1t_contracts::Outcome::Fail(failure) => return Ok(g1t_contracts::Outcome::Fail(failure)),
38 }
39 };
40}
41pub(crate) use unwrap_or_fail;
42
43const SESSION_COLUMNS: &str = "id, repo_id, issue, pull, branch, actor_id, actor_name, harness, harness_session_id,
44 title, state, head_commit, started_at, last_seen_at, ended_at, end_reason,
45 parent_id, agent_type, task, prompt, result, model";
46
47const CLAIM_COLUMNS: &str = "id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state,
48 ttl_minutes, created_at, expires_at, ended_at, end_reason, handoff_to";
49
50#[derive(Deserialize)]
51pub(crate) struct SessionRow {
52 id: String,
53 repo_id: String,
54 issue: Option<u32>,
55 pull: Option<u32>,
56 branch: Option<String>,
57 actor_id: String,
58 actor_name: String,
59 harness: String,
60 harness_session_id: Option<String>,
61 title: Option<String>,
62 state: String,
63 head_commit: Option<String>,
64 started_at: String,
65 last_seen_at: String,
66 ended_at: Option<String>,
67 end_reason: Option<String>,
68 #[serde(default)]
69 parent_id: Option<String>,
70 #[serde(default)]
71 agent_type: Option<String>,
72 #[serde(default)]
73 task: Option<String>,
74 #[serde(default)]
75 prompt: Option<String>,
76 #[serde(default)]
77 result: Option<String>,
78 #[serde(default)]
79 model: Option<String>,
80}
81
82impl SessionRow {
83 /// The session as it reads at `now`.
84 pub(crate) fn at(self, now: u64) -> Session {
85 let last_seen = parse_rfc3339(&self.last_seen_at).unwrap_or(now);
86 Session {
87 state: SessionState::at(self.state == "ended", last_seen, now),
88 id: self.id,
89 repo_id: self.repo_id,
90 issue: self.issue,
91 pull: self.pull,
92 branch: self.branch,
93 actor: Principal { id: self.actor_id, username: self.actor_name },
94 harness: Harness::parse(&self.harness).unwrap_or(Harness::Other),
95 harness_session_id: self.harness_session_id,
96 title: self.title,
97 head_commit: self.head_commit,
98 started_at: self.started_at,
99 last_seen_at: self.last_seen_at,
100 ended_at: self.ended_at,
101 end_reason: self.end_reason,
102 parent_id: self.parent_id,
103 agent_type: self.agent_type,
104 task: self.task,
105 prompt: self.prompt,
106 result: self.result,
107 model: self.model,
108 subagents: 0,
109 first_prompt: None,
110 }
111 }
112}
113
114#[derive(Deserialize)]
115pub(crate) struct ClaimRow {
116 id: String,
117 repo_id: String,
118 issue: u32,
119 session_id: Option<String>,
120 actor_id: String,
121 actor_name: String,
122 harness: Option<String>,
123 note: Option<String>,
124 state: String,
125 #[allow(dead_code)]
126 ttl_minutes: u32,
127 created_at: String,
128 expires_at: String,
129 ended_at: Option<String>,
130 end_reason: Option<String>,
131 handoff_to: Option<String>,
132}
133
134impl From<ClaimRow> for Claim {
135 fn from(row: ClaimRow) -> Self {
136 Claim {
137 id: row.id,
138 repo_id: row.repo_id,
139 issue: row.issue,
140 session_id: row.session_id,
141 actor: Principal { id: row.actor_id, username: row.actor_name },
142 harness: row.harness.as_deref().and_then(Harness::parse),
143 note: row.note,
144 state: ClaimState::parse(&row.state).unwrap_or(ClaimState::Expired),
145 created_at: row.created_at,
146 expires_at: row.expires_at,
147 ended_at: row.ended_at,
148 end_reason: row.end_reason,
149 handoff_to: row.handoff_to,
150 }
151 }
152}
153
154/// Text as given, trimmed and cut to `limit` characters; none when empty.
155pub(crate) fn tidy(text: Option<&str>, limit: usize) -> Option<String> {
156 let text = text?.trim();
157 (!text.is_empty()).then(|| text.chars().take(limit).collect())
158}
159
160/// Ends a repository's (`?1`) sessions last seen at or before `?2` as
161/// `timed_out`, as of when they were last seen.
162pub(crate) fn time_out_sessions_sql() -> String {
163 format!(
164 "UPDATE sessions SET state = 'ended', ended_at = last_seen_at, end_reason = 'timed_out'
165 WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2
166 RETURNING {SESSION_COLUMNS}"
167 )
168}
169
170/// Expires a repository's (`?1`) active claims due at or before `?2`.
171pub(crate) fn expire_claims_sql() -> String {
172 format!(
173 "UPDATE claims SET state = 'expired', ended_at = expires_at, end_reason = 'expired'
174 WHERE repo_id = ?1 AND state = 'active' AND expires_at <= ?2
175 RETURNING {CLAIM_COLUMNS}"
176 )
177}
178
179/// RFC 3339 for `minutes` after `now`.
180pub(crate) fn after_minutes(now: u64, minutes: u32) -> String {
181 rfc3339(now + u64::from(minutes) * 60_000)
182}
183
184impl Work {
185 /// The repository as `viewer` sees it, for reading its coordination.
186 pub(crate) async fn coordination_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
187 let repo = unwrap_or_fail!(self.repo(path, viewer).await?);
188 self.sweep(&repo.id).await?;
189 Ok(Outcome::Ok(repo))
190 }
191
192 /// The repository, if `actor` may take part in it: a confirmed email
193 /// address, a repository that is not archived, and a role that may
194 /// open issues and pull requests.
195 pub(crate) async fn participating_repo(&self, path: &RepoPath, actor: &User) -> Result<Outcome<Repo>> {
196 if !actor.verified {
197 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
198 }
199 let repo = unwrap_or_fail!(self.repo(path, &Some(actor.clone())).await?);
200 unwrap_or_fail!(crate::retired::writable(&repo));
201 unwrap_or_fail!(crate::allowed(Some(actor), &repo, g1t_contracts::access::Capability::Participate));
202 self.sweep(&repo.id).await?;
203 Ok(Outcome::Ok(repo))
204 }
205
206 /// Ends the repository's sessions that have gone two hours without a
207 /// heartbeat (`timed_out`) and expires its claims past their time, and
208 /// announces each. Run before the repository's coordination is read or
209 /// changed, so that nobody sees a stale one as live.
210 pub(crate) async fn sweep(&self, repo_id: &str) -> Result<()> {
211 let now = now_ms();
212 let quiet_since = rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS));
213 let at = rfc3339(now);
214 let results = self
215 .db
216 .batch(vec![
217 self.db.prepare(time_out_sessions_sql()).bind(&[repo_id.into(), quiet_since.as_str().into()])?,
218 self.db.prepare(expire_claims_sql()).bind(&[repo_id.into(), at.as_str().into()])?,
219 ])
220 .await?;
221 let mut results = results.into_iter();
222 let ended: Vec<SessionRow> = match results.next() {
223 Some(result) => result.results()?,
224 None => Vec::new(),
225 };
226 let expired: Vec<ClaimRow> = match results.next() {
227 Some(result) => result.results()?,
228 None => Vec::new(),
229 };
230 for row in ended {
231 let session = row.at(now);
232 self.publish_as("session.ended", repo_id, None, SessionEvent::of(&session)).await?;
233 }
234 for row in expired {
235 let claim = Claim::from(row);
236 self.publish_as("work.claim_expired", repo_id, None, ClaimEvent::of(&claim)).await?;
237 }
238 Ok(())
239 }
240
241 pub(crate) async fn session_row(&self, repo_id: &str, id: &str) -> Result<Option<SessionRow>> {
242 self.db
243 .prepare(format!("SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id = ?"))
244 .bind(&[repo_id.into(), id.into()])?
245 .first::<SessionRow>(None)
246 .await
247 }
248
249 /// Active claims on an issue, oldest first.
250 pub(crate) async fn active_claims(&self, repo_id: &str, issue: u32) -> Result<Vec<Claim>> {
251 Ok(self
252 .db
253 .prepare(format!(
254 "SELECT {CLAIM_COLUMNS} FROM claims
255 WHERE repo_id = ? AND issue = ? AND state = 'active' ORDER BY created_at, id"
256 ))
257 .bind(&[repo_id.into(), issue.into()])?
258 .all()
259 .await?
260 .results::<ClaimRow>()?
261 .into_iter()
262 .map(Claim::from)
263 .collect())
264 }
265}