Skip to content
Sign up

flagon-io/g1t

public
0
264 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 }
110 }
111}
112
113#[derive(Deserialize)]
114pub(crate) struct ClaimRow {
115 id: String,
116 repo_id: String,
117 issue: u32,
118 session_id: Option<String>,
119 actor_id: String,
120 actor_name: String,
121 harness: Option<String>,
122 note: Option<String>,
123 state: String,
124 #[allow(dead_code)]
125 ttl_minutes: u32,
126 created_at: String,
127 expires_at: String,
128 ended_at: Option<String>,
129 end_reason: Option<String>,
130 handoff_to: Option<String>,
131}
132
133impl From<ClaimRow> for Claim {
134 fn from(row: ClaimRow) -> Self {
135 Claim {
136 id: row.id,
137 repo_id: row.repo_id,
138 issue: row.issue,
139 session_id: row.session_id,
140 actor: Principal { id: row.actor_id, username: row.actor_name },
141 harness: row.harness.as_deref().and_then(Harness::parse),
142 note: row.note,
143 state: ClaimState::parse(&row.state).unwrap_or(ClaimState::Expired),
144 created_at: row.created_at,
145 expires_at: row.expires_at,
146 ended_at: row.ended_at,
147 end_reason: row.end_reason,
148 handoff_to: row.handoff_to,
149 }
150 }
151}
152
153/// Text as given, trimmed and cut to `limit` characters; none when empty.
154pub(crate) fn tidy(text: Option<&str>, limit: usize) -> Option<String> {
155 let text = text?.trim();
156 (!text.is_empty()).then(|| text.chars().take(limit).collect())
157}
158
159/// Ends a repository's (`?1`) sessions last seen at or before `?2` as
160/// `timed_out`, as of when they were last seen.
161pub(crate) fn time_out_sessions_sql() -> String {
162 format!(
163 "UPDATE sessions SET state = 'ended', ended_at = last_seen_at, end_reason = 'timed_out'
164 WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2
165 RETURNING {SESSION_COLUMNS}"
166 )
167}
168
169/// Expires a repository's (`?1`) active claims due at or before `?2`.
170pub(crate) fn expire_claims_sql() -> String {
171 format!(
172 "UPDATE claims SET state = 'expired', ended_at = expires_at, end_reason = 'expired'
173 WHERE repo_id = ?1 AND state = 'active' AND expires_at <= ?2
174 RETURNING {CLAIM_COLUMNS}"
175 )
176}
177
178/// RFC 3339 for `minutes` after `now`.
179pub(crate) fn after_minutes(now: u64, minutes: u32) -> String {
180 rfc3339(now + u64::from(minutes) * 60_000)
181}
182
183impl Work {
184 /// The repository as `viewer` sees it, for reading its coordination.
185 pub(crate) async fn coordination_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
186 let repo = unwrap_or_fail!(self.repo(path, viewer).await?);
187 self.sweep(&repo.id).await?;
188 Ok(Outcome::Ok(repo))
189 }
190
191 /// The repository, if `actor` may take part in it: a confirmed email
192 /// address, a repository that is not archived, and a role that may
193 /// open issues and pull requests.
194 pub(crate) async fn participating_repo(&self, path: &RepoPath, actor: &User) -> Result<Outcome<Repo>> {
195 if !actor.verified {
196 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
197 }
198 let repo = unwrap_or_fail!(self.repo(path, &Some(actor.clone())).await?);
199 unwrap_or_fail!(crate::retired::writable(&repo));
200 unwrap_or_fail!(crate::allowed(Some(actor), &repo, g1t_contracts::access::Capability::Participate));
201 self.sweep(&repo.id).await?;
202 Ok(Outcome::Ok(repo))
203 }
204
205 /// Ends the repository's sessions that have gone two hours without a
206 /// heartbeat (`timed_out`) and expires its claims past their time, and
207 /// announces each. Run before the repository's coordination is read or
208 /// changed, so that nobody sees a stale one as live.
209 pub(crate) async fn sweep(&self, repo_id: &str) -> Result<()> {
210 let now = now_ms();
211 let quiet_since = rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS));
212 let at = rfc3339(now);
213 let results = self
214 .db
215 .batch(vec![
216 self.db.prepare(time_out_sessions_sql()).bind(&[repo_id.into(), quiet_since.as_str().into()])?,
217 self.db.prepare(expire_claims_sql()).bind(&[repo_id.into(), at.as_str().into()])?,
218 ])
219 .await?;
220 let mut results = results.into_iter();
221 let ended: Vec<SessionRow> = match results.next() {
222 Some(result) => result.results()?,
223 None => Vec::new(),
224 };
225 let expired: Vec<ClaimRow> = match results.next() {
226 Some(result) => result.results()?,
227 None => Vec::new(),
228 };
229 for row in ended {
230 let session = row.at(now);
231 self.publish_as("session.ended", repo_id, None, SessionEvent::of(&session)).await?;
232 }
233 for row in expired {
234 let claim = Claim::from(row);
235 self.publish_as("work.claim_expired", repo_id, None, ClaimEvent::of(&claim)).await?;
236 }
237 Ok(())
238 }
239
240 pub(crate) async fn session_row(&self, repo_id: &str, id: &str) -> Result<Option<SessionRow>> {
241 self.db
242 .prepare(format!("SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id = ?"))
243 .bind(&[repo_id.into(), id.into()])?
244 .first::<SessionRow>(None)
245 .await
246 }
247
248 /// Active claims on an issue, oldest first.
249 pub(crate) async fn active_claims(&self, repo_id: &str, issue: u32) -> Result<Vec<Claim>> {
250 Ok(self
251 .db
252 .prepare(format!(
253 "SELECT {CLAIM_COLUMNS} FROM claims
254 WHERE repo_id = ? AND issue = ? AND state = 'active' ORDER BY created_at, id"
255 ))
256 .bind(&[repo_id.into(), issue.into()])?
257 .all()
258 .await?
259 .results::<ClaimRow>()?
260 .into_iter()
261 .map(Claim::from)
262 .collect())
263 }
264}