Sign in

flagon-io/g1t

public

Git for AI scale: a forge for thousands of agents working on the same code at once.

gitagentscloudflarerustforgemcp

g1t/services/work/src/retired.rs

311 lines12,898 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1//! Repositories that are archived, deleted or purged, and branches
2//! renamed: what issues, pull requests and agents do about each.
3//!
4//! An archived repository is read-only: its issues and pull requests are
5//! locked and nothing new starts on it. A deleted one looks missing (repos
6//! hides it) and keeps its rows for a restore. A purged one is gone, and
7//! every row kept for it goes with it.
8
9use g1t_contracts::events::{BranchRenamed, Event, RepoArchived, RepoDeleted};
10use g1t_contracts::repos::{Repo, RepoStatus, StatusByIdArgs, archived_message};
11use g1t_contracts::{FailureCode, Outcome};
12use g1t_contracts::time::rfc3339;
13use g1t_kit::now_ms;
14use serde::{Deserialize, Serialize};
15use worker::Result;
16
17use crate::Work;
18
19/// `Ok` when `repo` may be changed; refused, saying why, when it is
20/// archived.
21pub(crate) fn writable(repo: &Repo) -> Outcome<()> {
22 if repo.archived() {
23 Outcome::fail(FailureCode::Forbidden, archived_message(&repo.namespace, &repo.name))
24 } else {
25 Outcome::Ok(())
26 }
27}
28
29/// `repo` as found, unless it is archived: then refused, as [`writable`]
30/// says. For the steps g1t starts by itself, which stop on a refusal.
31pub(crate) fn unless_archived(repo: Outcome<Repo>) -> Outcome<Repo> {
32 match repo {
33 Outcome::Ok(repo) => match writable(&repo) {
34 Outcome::Ok(()) => Outcome::Ok(repo),
35 Outcome::Fail(failure) => Outcome::Fail(failure),
36 },
37 failed => failed,
38 }
39}
40
41/// `repo.purged`: every row kept for the repository, `?1` its id. Rows
42/// kept by pull request go first, while the pull requests still name them.
43pub(crate) const PURGED: &[&str] = &[
44 "DELETE FROM session_entries WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
45 "DELETE FROM check_runs WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
46 "DELETE FROM review_runs WHERE pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
47 "DELETE FROM agent_messages WHERE repo_id = ?1 OR pull_id IN (SELECT id FROM pulls WHERE repo_id = ?1)",
48 "DELETE FROM queue_entries WHERE repo_id = ?1",
49 "DELETE FROM agent_mentions WHERE repo_id = ?1",
50 "DELETE FROM agent_rules WHERE repo_id = ?1",
51 "DELETE FROM agent_runs WHERE repo_id = ?1",
52 "DELETE FROM commit_statuses WHERE repo_id = ?1",
53 "DELETE FROM plans WHERE repo_id = ?1",
54 "DELETE FROM comments WHERE repo_id = ?1",
55 "DELETE FROM pulls WHERE repo_id = ?1",
56 "DELETE FROM issues WHERE repo_id = ?1",
57 "DELETE FROM counters WHERE repo_id = ?1",
58 "DELETE FROM repo_settings WHERE repo_id = ?1",
59 "DELETE FROM memories WHERE scope = 'project' AND scope_key = ?1",
60 "DELETE FROM guardrails WHERE scope <> 'workspace' AND scope_key = ?1",
61];
62
63/// A repository renamed or transferred (see `g1t_kit::transfer`): runs
64/// waiting for an agent slot name it by path in their payload, at `$.repo`
65/// or `$.job.repo`; they follow it, into its workspace's queue now.
66pub(crate) const WAITS_MOVED: &[&str] = &[
67 "UPDATE agent_waits SET workspace = lower(?3),
68 payload = json_set(payload, '$.repo.namespace', ?3, '$.repo.name', ?6)
69 WHERE json_valid(payload)
70 AND lower(json_extract(payload, '$.repo.namespace')) = lower(?4)
71 AND lower(json_extract(payload, '$.repo.name')) = lower(?7)",
72 "UPDATE agent_waits SET workspace = lower(?3),
73 payload = json_set(payload, '$.job.repo.namespace', ?3, '$.job.repo.name', ?6)
74 WHERE json_valid(payload)
75 AND lower(json_extract(payload, '$.job.repo.namespace')) = lower(?4)
76 AND lower(json_extract(payload, '$.job.repo.name')) = lower(?7)",
77];
78
79/// What a run stopped because its repository was archived or deleted says
80/// as its last step; [`Work::runs_in_repo`] finds such runs by it.
81const STOPPED_STEP: &str = "Stopped: the repository was archived or deleted.";
82
83/// `repo.deleted` and `repo.archived`: the repository's agent runs that
84/// are queued or running stop, `?1` the step they end on, `?2` the time,
85/// `?3` its id.
86const STOP_RUNS: &str = "UPDATE agent_runs SET status = 'stopped', step = ?1, finished_at = ?2, updated_at = ?2
87 WHERE repo_id = ?3 AND status IN ('queued', 'running')";
88
89/// And runs waiting for an agent slot in it are dropped: `?1` its
90/// workspace, `?2` its name, as the payload names it (see [`WAITS_MOVED`]).
91const DROP_WAITS: &[&str] = &[
92 "DELETE FROM agent_waits WHERE json_valid(payload)
93 AND lower(json_extract(payload, '$.repo.namespace')) = lower(?1)
94 AND lower(json_extract(payload, '$.repo.name')) = lower(?2)",
95 "DELETE FROM agent_waits WHERE json_valid(payload)
96 AND lower(json_extract(payload, '$.job.repo.namespace')) = lower(?1)
97 AND lower(json_extract(payload, '$.job.repo.name')) = lower(?2)",
98];
99
100/// `runs_in_repo` (internal, for the runner): the agent runs of a
101/// repository whose sandboxes should be stopped: those queued or running,
102/// and those work stopped in the last hour because the repository was
103/// archived or deleted. Takes `{ "repoId" }`; returns `[RepoRun]`.
104#[derive(Deserialize)]
105#[serde(rename_all = "camelCase")]
106pub(crate) struct RunsInRepoArgs {
107 pub repo_id: String,
108}
109
110#[derive(Debug, Deserialize, Serialize)]
111#[serde(rename_all = "camelCase")]
112pub(crate) struct RepoRun {
113 pub run_id: String,
114 pub sandbox: String,
115 pub kind: String,
116 pub status: String,
117 pub pull_id: Option<String>,
118}
119
120/// The repository, by id and path, that `event` says was deleted or
121/// archived (not unarchived): its runs stop.
122fn stopping(event: &Event) -> Option<(String, String, String)> {
123 let data = event.data.clone();
124 match event.kind.as_str() {
125 "repo.deleted" => serde_json::from_value::<RepoDeleted>(data).ok().map(|e| (e.repo_id, e.namespace, e.name)),
126 "repo.archived" => serde_json::from_value::<RepoArchived>(data)
127 .ok()
128 .filter(|e| e.archived)
129 .map(|e| (e.repo_id, e.namespace, e.name)),
130 _ => None,
131 }
132}
133
134/// `branch.renamed`: open pull requests from `?2` in repository `?1`
135/// come from `?3` now. A fork carries its change on its own default
136/// branch, so only branches of the repository itself are named.
137const BRANCH_RENAMED: &str = "UPDATE pulls SET source_branch = ?3
138 WHERE repo_id = ?1 AND source_branch = ?2 AND fork_repo_id IS NULL
139 AND status IN ('draft', 'open')";
140
141impl Work {
142 /// Whether work may start on the repository: neither archived nor
143 /// deleted. When repos cannot say, it is taken as active, and the
144 /// repos calls that follow (which hide deleted repositories) decide.
145 pub(crate) async fn repo_active(&self, repo_id: &str) -> Result<bool> {
146 let status: Result<RepoStatus> = g1t_kit::call(
147 &self.repos,
148 "status_by_id",
149 &StatusByIdArgs {
150 id: repo_id.to_owned(),
151 },
152 )
153 .await;
154 Ok(match status {
155 Ok(status) => status.active(),
156 Err(error) => {
157 worker::console_error!("status_by_id {repo_id}: {error}");
158 true
159 }
160 })
161 }
162
163 /// See [`RunsInRepoArgs`].
164 pub(crate) async fn runs_in_repo(&self, a: RunsInRepoArgs) -> Result<Vec<RepoRun>> {
165 let hour_ago = rfc3339(now_ms().saturating_sub(60 * 60 * 1000));
166 self.db
167 .prepare(
168 "SELECT id AS run_id, sandbox, kind, status, pull_id FROM agent_runs
169 WHERE repo_id = ?1 AND (status IN ('queued', 'running')
170 OR (status = 'stopped' AND step = ?2 AND finished_at >= ?3))
171 ORDER BY created_at LIMIT 200",
172 )
173 .bind(&[a.repo_id.as_str().into(), STOPPED_STEP.into(), hour_ago.as_str().into()])?
174 .all()
175 .await?
176 .results::<RepoRun>()
177 }
178
179 /// `repo.deleted` and `repo.archived`: its agent runs stop (the runner
180 /// stops their sandboxes, see [`RunsInRepoArgs`]) and runs waiting for
181 /// a slot in it are dropped. Rows are kept, for a restore.
182 async fn stop_runs_in(&self, repo_id: &str, namespace: &str, name: &str) -> Result<()> {
183 let now = rfc3339(now_ms());
184 let mut batch = vec![self
185 .db
186 .prepare(STOP_RUNS)
187 .bind(&[STOPPED_STEP.into(), now.as_str().into(), repo_id.into()])?];
188 if !namespace.is_empty() && !name.is_empty() {
189 for sql in DROP_WAITS {
190 batch.push(self.db.prepare(*sql).bind(&[namespace.into(), name.into()])?);
191 }
192 }
193 self.db.batch(batch).await?;
194 Ok(())
195 }
196
197 /// `repo.purged`, `repo.deleted`, `repo.archived` and `branch.renamed`.
198 /// Says whether `event` was one this handles completely.
199 pub(crate) async fn on_retired(&self, event: &Event) -> Result<bool> {
200 if g1t_kit::lifecycle::on_purged(&self.db, event, PURGED).await? {
201 return Ok(true);
202 }
203 if let Some((repo_id, namespace, name)) = stopping(event) {
204 self.stop_runs_in(&repo_id, &namespace, &name).await?;
205 return Ok(true);
206 }
207 if event.kind != "branch.renamed" {
208 return Ok(false);
209 }
210 let Ok(renamed) = serde_json::from_value::<BranchRenamed>(event.data.clone()) else {
211 worker::console_error!("branch.renamed {} could not be read", event.id);
212 return Ok(true);
213 };
214 if renamed.from == renamed.to {
215 return Ok(true);
216 }
217 self.db
218 .prepare(BRANCH_RENAMED)
219 .bind(&[
220 renamed.repo_id.as_str().into(),
221 renamed.from.as_str().into(),
222 renamed.to.as_str().into(),
223 ])?
224 .run()
225 .await?;
226 Ok(true)
227 }
228}
229
230#[cfg(test)]
231mod tests {
232 use super::*;
233
234 fn repo(archived_at: Option<&str>) -> Repo {
235 Repo {
236 id: "rep_1".into(),
237 namespace: "acme".into(),
238 name: "web".into(),
239 description: None,
240 is_private: false,
241 owner_id: "usr_1".into(),
242 default_branch: "main".into(),
243 fork_of: None,
244 protected: false,
245 created_at: "2026-10-05T00:00:00Z".into(),
246 topics: Vec::new(),
247 website: None,
248 archived_at: archived_at.map(str::to_owned),
249 }
250 }
251
252 #[test]
253 fn an_archived_repository_refuses_changes() {
254 assert!(matches!(writable(&repo(None)), Outcome::Ok(())));
255 match writable(&repo(Some("2026-10-05T00:00:00Z"))) {
256 Outcome::Fail(failure) => {
257 assert_eq!(failure.code, FailureCode::Forbidden);
258 assert!(failure.message.contains("acme/web is archived"));
259 }
260 Outcome::Ok(()) => panic!("an archived repository was writable"),
261 }
262 }
263
264 #[test]
265 fn purging_takes_only_the_repository_id() {
266 for sql in PURGED {
267 assert_eq!(g1t_kit::transfer::parameters(sql), 1, "{sql}");
268 }
269 // Rows kept by pull request go before the pull requests do.
270 let pulls = PURGED.iter().position(|sql| sql.starts_with("DELETE FROM pulls")).unwrap();
271 for table in ["session_entries", "check_runs", "review_runs", "agent_messages"] {
272 let at = PURGED.iter().position(|sql| sql.contains(&format!("FROM {table} "))).unwrap();
273 assert!(at < pulls, "{table}");
274 }
275 }
276
277 fn event(kind: &str, data: serde_json::Value) -> Event {
278 Event {
279 id: "evt_1".into(),
280 kind: kind.into(),
281 source: "repos".into(),
282 time: "2026-10-05T00:00:00Z".into(),
283 repo_id: Some("rep_1".into()),
284 actor: None,
285 data,
286 }
287 }
288
289 #[test]
290 fn deleting_or_archiving_stops_runs_and_unarchiving_does_not() {
291 let named = Some(("rep_1".to_owned(), "acme".to_owned(), "web".to_owned()));
292 let data = serde_json::json!({ "repoId": "rep_1", "namespace": "acme", "name": "web", "archived": true });
293 assert_eq!(stopping(&event("repo.deleted", data.clone())), named);
294 assert_eq!(stopping(&event("repo.archived", data)), named);
295 let unarchived = serde_json::json!({ "repoId": "rep_1", "namespace": "acme", "name": "web", "archived": false });
296 assert_eq!(stopping(&event("repo.unarchived", unarchived.clone())), None);
297 assert_eq!(stopping(&event("repo.archived", unarchived)), None);
298 assert_eq!(STOP_RUNS.matches('?').count(), 4);
299 for sql in DROP_WAITS {
300 assert_eq!(g1t_kit::transfer::parameters(sql), 2, "{sql}");
301 }
302 }
303
304 #[test]
305 fn waiting_runs_follow_with_the_names_they_bind() {
306 for sql in WAITS_MOVED {
307 assert_eq!(g1t_kit::transfer::parameters(sql), 7, "{sql}");
308 }
309 assert_eq!(g1t_kit::transfer::parameters(BRANCH_RENAMED), 3);
310 }
311}