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
616 linesCodeBlameRaw

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.

README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise1//! The events service: the bus every state change in g1t is published on,
2//! and its durable log.
3//!
4//! Publishing puts events on a queue and returns. The queue consumer writes
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails5//! them to the log and passes each batch on to each subscriber's own
6//! queue, so a slow or failing subscriber holds up nobody else. Each
7//! subscriber is sent only the types it acts on
8//! (`g1t_contracts::subscribers`), in batches the queues take (fanout.rs).
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise9//!
Inbox: the events service tells people what needs them as events arrive10//! It keeps two more things beside the log: the audit log (audit.rs) and
Inbox: threads, reasons, subscriptions and watching11//! each person's inbox (inbox.rs), written as events arrive, with who
12//! follows what (subscriptions.rs).
Inbox: the events service tells people what needs them as events arrive13//!
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise14//! Other services reach it over `POST /rpc/<method>`; see
Inbox: the events service tells people what needs them as events arrive15//! `g1t_contracts::events`, `audit` and `inbox` for the methods and their
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.16//! arguments. `activity_digest` (digest.rs) counts a span of the log for
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r917//! Home. Security alerts that are not urgent wait for each person's
18//! security digest (security_digest.rs), sent by a cron of its own.
Merge branch 'worktree-agent-a5c94df323642630e'19//!
20//! Repository pages open in someone's browser refresh themselves: each
21//! batch's events that such a page shows are passed to the notify service
22//! (live.rs), which tells those watching the repository.
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise23
24mod audit;
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.25mod digest;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails26mod fanout;
Inbox: the events service tells people what needs them as events arrive27mod inbox;
Merge branch 'worktree-agent-a5c94df323642630e'28mod live;
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r929mod security_digest;
Inbox: threads, reasons, subscriptions and watching30mod subscriptions;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise31
g1t coordinates the work and the person's own sessions do it: Ask g1t reads g1t with the same tools any agent gets, answers in one conversation shared with the @g1t direct message in Chat, and hands work to sessions in the person's own harness, whose subagents g1t records, ties to issues and pull requests, and lets see each other.32use g1t_contracts::events::{Event, ListArgs, MAX_LIST_REPOS, PublishArgs};
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise33use g1t_contracts::new_id;
34use g1t_contracts::time::rfc3339;
35use g1t_kit::{args, js, now_ms, reply, rpc_method};
36use serde::Deserialize;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails37use std::collections::{HashMap, HashSet};
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise38use worker::js_sys::{Array, Object};
39use worker::wasm_bindgen::{JsCast, JsValue};
40use worker::{Context, D1Database, Env, MessageBatch, Request, Response, Result, event};
41
42const DEFAULT_PAGE: u32 = 50;
43const MAX_PAGE: u32 = 200;
Merge checks: statuses and check runs on every commit44/// Statuses and check runs reported on commits: delivered to webhooks, and
45/// left out of a timeline unless asked for by type.
46const REPORTING: [&str; 7] = [
47 "status.created",
48 "check_run.created",
49 "check_run.completed",
50 "check_run.rerequested",
51 "check_run.requested_action",
52 "check_suite.completed",
53 "check_suite.rerequested",
54];
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails55/// Every binding whose name starts with this is a queue that receives
56/// events: one per subscribing service, sent the types it routes.
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise57const SUBSCRIBER_PREFIX: &str = "SUBSCRIBER_";
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails58/// How long a record of which queues a batch reached is kept for its retries.
59const FANOUT_KEEP_MS: u64 = 24 * 60 * 60 * 1000;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise60
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.61/// An event as the log's table holds it.
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise62#[derive(Deserialize)]
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.63pub(crate) struct EventRow {
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise64 id: String,
65 #[serde(rename = "type")]
66 kind: String,
67 source: String,
68 time: String,
69 repo_id: Option<String>,
70 actor: Option<String>,
71 /// JSON.
72 data: String,
73}
74
75impl From<EventRow> for Event {
76 fn from(row: EventRow) -> Self {
77 Event {
78 id: row.id,
79 kind: row.kind,
80 source: row.source,
81 time: row.time,
82 repo_id: row.repo_id,
83 actor: row.actor,
84 data: serde_json::from_str(&row.data).unwrap_or_default(),
85 }
86 }
87}
88
89fn optional(value: &Option<String>) -> JsValue {
90 value.as_deref().map_or(JsValue::NULL, JsValue::from)
91}
92
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails93#[derive(Deserialize)]
94struct FanoutRow {
95 event_id: String,
96 /// JSON array of binding names.
97 bindings: String,
98}
99
100/// Sends `events` to a queue binding, one message each, in as many
101/// `sendBatch` calls as the queue's limits need (fanout.rs). Events are
102/// expected to have been fit to a message.
103async fn send(queue: &JsValue, events: &[&Event]) -> Result<()> {
104 for chunk in fanout::chunks(events) {
105 let messages = Array::new();
106 for event in chunk {
107 let message = Object::new();
108 js::set(&message, "body", &js::to_js(event)?);
109 messages.push(&message);
110 }
111 js::call(queue, "sendBatch", &[messages.into()]).await?;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise112 }
113 Ok(())
114}
115
116struct Events {
117 db: D1Database,
118 env: Env,
119}
120
121impl Events {
122 /// Assigns each event its id and time and puts it on the bus.
123 async fn publish(&self, a: PublishArgs) -> Result<()> {
124 if a.events.is_empty() {
125 return Ok(());
126 }
127 let now = now_ms();
128 let events: Vec<Event> = a
129 .events
130 .into_iter()
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails131 .map(|event| {
132 let mut event = Event {
133 id: new_id("evt", now),
134 kind: event.kind,
135 source: event.source,
136 time: rfc3339(now),
137 repo_id: event.repo_id,
138 actor: event.actor,
139 data: event.data,
140 };
141 // Too large for one message: its long text is shortened.
142 fanout::fit(&mut event);
143 event
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise144 })
145 .collect();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails146 let events: Vec<&Event> = events.iter().collect();
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise147 send(&js::binding(&self.env, "BUS")?, &events).await
148 }
149
Merge branch 'worktree-agent-ad531323a21f4c975'150 /// Newest first, or with `after` oldest first in the order the log
151 /// wrote them. Callers must have checked that the viewer may see the
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise152 /// repository asked about.
153 async fn list(&self, a: ListArgs) -> Result<Vec<Event>> {
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)154 let (filter, binds) = list_filter(&a);
155 let mut values: Vec<JsValue> = binds
156 .into_iter()
157 .map(|bind| match bind {
158 Bind::Text(text) => JsValue::from(text),
159 Bind::Number(number) => JsValue::from(number),
160 })
161 .collect();
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise162 values.push(a.limit.unwrap_or(DEFAULT_PAGE).min(MAX_PAGE).into());
Merge branch 'worktree-agent-ad531323a21f4c975'163 let order = if a.after.is_some() { "seq ASC" } else { "id DESC" };
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise164 let rows = self
165 .db
166 .prepare(format!(
Merge branch 'worktree-agent-ad531323a21f4c975'167 "SELECT * FROM events {filter} ORDER BY {order} LIMIT ?"
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise168 ))
169 .bind(&values)?
170 .all()
171 .await?
172 .results::<EventRow>()?;
173 Ok(rows.into_iter().map(Event::from).collect())
174 }
175
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails176 /// Writes a batch from the bus to the log, notes who now follows what
177 /// (inbox.rs), hands each subscriber the events it routes, then tells
178 /// the people it concerns.
179 ///
180 /// Everything before the hand-off is safe to repeat. A retried batch
181 /// skips the queues that already have it: when a hand-off fails part
182 /// way, which queues each event reached is written down first.
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise183 async fn deliver(&self, events: &[Event]) -> Result<()> {
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails184 let mut statements = Vec::with_capacity(events.len() + 1);
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise185 for event in events {
186 statements.push(
187 self.db
188 .prepare(
189 // Redelivered batches must not duplicate log rows.
Merge branch 'worktree-agent-ad531323a21f4c975'190 // `seq` is the order of writing, which `after` follows
191 // (0010_events_seq.sql): the log has one writer.
192 "INSERT OR IGNORE INTO events (id, type, source, time, repo_id, actor, data, seq)
193 VALUES (?, ?, ?, ?, ?, ?, ?, (SELECT COALESCE(MAX(seq), 0) + 1 FROM events))",
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise194 )
195 .bind(&[
196 event.id.as_str().into(),
197 event.kind.as_str().into(),
198 event.source.as_str().into(),
199 event.time.as_str().into(),
200 optional(&event.repo_id),
201 optional(&event.actor),
202 serde_json::to_string(&event.data)?.into(),
203 ])?,
204 );
205 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails206 let ids: Vec<&str> = events.iter().map(|event| event.id.as_str()).collect();
207 // Read in the same round trip: nearly always nothing.
208 statements.push(
209 self.db
210 .prepare("SELECT event_id, bindings FROM fanout_sent WHERE event_id IN (SELECT value FROM json_each(?))")
211 .bind(&[serde_json::to_string(&ids)?.into()])?,
212 );
213 let results = self.db.batch(statements).await?;
214 let mut sent: HashMap<String, HashSet<String>> = HashMap::new();
215 if let Some(rows) = results.last() {
216 for row in rows.results::<FanoutRow>()? {
217 let bindings: Vec<String> = serde_json::from_str(&row.bindings).unwrap_or_default();
218 sent.insert(row.event_id, bindings.into_iter().collect());
219 }
220 }
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise221 audit::follow_renames(&self.db, events).await?;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails222 // Before the hand-off, so its failing does not send the batch to
223 // every queue again.
224 inbox::follow(&self.db, events).await?;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise225
226 let bindings: &JsValue = self.env.as_ref();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails227 let mut reached: Vec<String> = Vec::new();
228 let mut failed = None;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise229 for name in Object::keys(bindings.unchecked_ref::<Object>()).iter() {
230 let Some(name) = name
231 .as_string()
232 .filter(|name| name.starts_with(SUBSCRIBER_PREFIX))
233 else {
234 continue;
235 };
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails236 let routed: Vec<&Event> = events
237 .iter()
238 .filter(|event| g1t_contracts::subscribers::routed(&name, &event.kind))
239 .filter(|event| !sent.get(&event.id).is_some_and(|reached| reached.contains(&name)))
240 .collect();
241 if routed.is_empty() {
242 continue;
243 }
244 match send(&js::binding(&self.env, &name)?, &routed).await {
245 Ok(()) => reached.push(name),
246 Err(error) => {
247 worker::console_error!("events: passing {} events to {name} failed: {error}", routed.len());
248 failed = Some(error);
249 }
250 }
251 }
252 if let Some(error) = failed {
253 self.note_reached(events, &sent, &reached).await?;
254 return Err(error);
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise255 }
Inbox: the events service tells people what needs them as events arrive256 let (work, repos, identity) = (
257 self.env.service("WORK")?,
258 self.env.service("REPOS")?,
259 self.env.service("IDENTITY")?,
260 );
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)261 // Optional: without it the inbox still fills, nobody is told live.
262 let notify = self.env.service("NOTIFY").ok();
Merge branch 'worktree-agent-a5c94df323642630e'263 // Pages open on the repositories: told first, as nobody waits longer.
264 if let Some(notify) = notify.as_ref() {
265 live::tell(notify, events).await;
266 }
Inbox: the events service tells people what needs them as events arrive267 let sources = inbox::Sources {
268 work: &work,
269 repos: &repos,
270 identity: &identity,
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)271 notify: notify.as_ref(),
Inbox: the events service tells people what needs them as events arrive272 };
273 inbox::deliver(&self.db, &sources, events).await;
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise274 Ok(())
275 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails276
277 /// Writes down which queues each event of a batch has reached, with
278 /// those it had reached before, for the batch's retry.
279 async fn note_reached(
280 &self,
281 events: &[Event],
282 sent: &HashMap<String, HashSet<String>>,
283 reached: &[String],
284 ) -> Result<()> {
285 let now = rfc3339(now_ms());
286 let mut statements = Vec::with_capacity(events.len());
287 for event in events {
288 let mut bindings: Vec<&str> = reached.iter().map(String::as_str).collect();
289 if let Some(before) = sent.get(&event.id) {
290 bindings.extend(before.iter().map(String::as_str));
291 }
292 bindings.sort_unstable();
293 bindings.dedup();
294 statements.push(
295 self.db
296 .prepare(
297 "INSERT INTO fanout_sent (event_id, bindings, created_at) VALUES (?, ?, ?)
298 ON CONFLICT (event_id) DO UPDATE SET bindings = excluded.bindings",
299 )
300 .bind(&[event.id.as_str().into(), serde_json::to_string(&bindings)?.into(), now.as_str().into()])?,
301 );
302 }
303 self.db.batch(statements).await?;
304 Ok(())
305 }
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise306}
307
308#[event(fetch)]
309async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
310 let Some(method) = rpc_method(&request) else {
311 return Response::error("Not found", 404);
312 };
313 let body: serde_json::Value = request.json().await?;
314 let events = Events {
315 db: env.d1("DB")?,
316 env,
317 };
318 match method.as_str() {
319 "publish" => reply(&events.publish(args(body)?).await?),
320 "list" => reply(&events.list(args(body)?).await?),
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.321 // Home's digest of a span: counted, by who did it (digest.rs).
322 "activity_digest" => reply(&digest::activity_digest(&events.db, args(body)?).await?),
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise323 "audit_record" => reply(&audit::record(&events.db, args(body)?).await?),
324 "audit_list" => reply(&audit::list(&events.db, args(body)?).await?),
Inbox: the events service tells people what needs them as events arrive325 "inbox_list" => {
326 let repos = events.env.service("REPOS")?;
327 reply(&inbox::list(&events.db, &repos, args(body)?).await?)
328 }
329 "inbox_counts" => reply(&inbox::counts(&events.db, args(body)?).await?),
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)330 "inbox_mark" => {
331 let marked: g1t_contracts::inbox::MarkInboxArgs = args(body)?;
332 let username = marked.username.clone();
333 let changed = inbox::mark(&events.db, marked).await?;
334 // Notify: the new count in every open tab, whoever marked (the site, the API, MCP).
335 if changed > 0
336 && let Ok(notify) = events.env.service("NOTIFY")
337 {
338 inbox::tell_inbox_count(&events.db, &notify, &username).await;
339 }
340 reply(&changed)
341 }
Inbox: threads, reasons, subscriptions and watching342 "inbox_thread" => {
343 let (repos, work) = (events.env.service("REPOS")?, events.env.service("WORK")?);
344 reply(&inbox::thread(&events.db, &repos, &work, args(body)?).await?)
345 }
346 "inbox_subscription" => {
347 let work = events.env.service("WORK")?;
348 reply(&subscriptions::subscription(&events.db, &work, args(body)?).await?)
349 }
350 "inbox_subscribe" => {
351 let work = events.env.service("WORK")?;
352 reply(&subscriptions::subscribe(&events.db, &work, args(body)?).await?)
353 }
354 "inbox_watching" => reply(&subscriptions::watching(&events.db, args(body)?).await?),
355 "inbox_watch" => reply(&subscriptions::watch(&events.db, args(body)?).await?),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97356 "inbox_watchers" => reply(&subscriptions::watchers_count(&events.db, args(body)?).await?),
Inbox: threads, reasons, subscriptions and watching357 "inbox_watched" => reply(&subscriptions::watched(&events.db, args(body)?).await?),
358 "inbox_settings" => reply(&subscriptions::settings(&events.db, args(body)?).await?),
359 "inbox_update_settings" => reply(&subscriptions::update_settings(&events.db, args(body)?).await?),
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise360 _ => Response::error("Unknown method", 404),
361 }
362}
363
364/// Events from the bus. A batch that fails is retried whole, which is safe
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails365/// because the log ignores an event it has seen, and queues that already
366/// have it are skipped (`deliver`). After its retries it goes to the
367/// dead-letter queue (wrangler.jsonc).
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise368#[event(queue)]
369async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
370 let events = Events {
371 db: env.d1("DB")?,
372 env,
373 };
374 let delivered: Vec<Event> = batch
375 .messages()?
376 .into_iter()
377 .map(|message| message.into_body())
378 .collect();
379 events.deliver(&delivered).await?;
380 batch.ack_all();
381 Ok(())
382}
383
384/// Once a day, old audit entries are removed: first everything older than
385/// any workspace keeps (`AUDIT_MAX_DAYS`, 400 days), then each workspace's
386/// entries older than its own plan keeps, as billing says (7 days free, 90
387/// on the plan, or what staff set). Workspaces with nothing older than the
388/// shortest (`AUDIT_MIN_DAYS`, 7) are left alone.
389#[event(scheduled)]
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r9390async fn scheduled(event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
391 // The morning cron sends the security digests that are due, and nothing else.
392 if event.cron() == security_digest::DIGEST_CRON {
393 let (Ok(db), Ok(identity)) = (env.d1("DB"), env.service("IDENTITY")) else { return };
394 let notify = env.service("NOTIFY").ok();
395 match security_digest::send_due(&db, &identity, notify.as_ref(), now_ms()).await {
396 Ok(sent) if sent > 0 => worker::console_log!("sent {sent} security digests"),
397 Ok(_) => {}
398 Err(error) => worker::console_error!("security digests not sent: {error}"),
399 }
400 return;
401 }
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise402 let days = |name: &str, default: u32| {
403 env.var(name)
404 .ok()
405 .and_then(|v| v.to_string().trim().parse::<u32>().ok())
406 .filter(|days| *days > 0)
407 .unwrap_or(default)
408 };
409 let max_days = days("AUDIT_MAX_DAYS", audit::DEFAULT_MAX_DAYS);
410 let min_days = days("AUDIT_MIN_DAYS", audit::DEFAULT_MIN_DAYS).min(max_days);
411 let Ok(db) = env.d1("DB") else { return };
412 let now = now_ms();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails413 let cutoff = rfc3339(now.saturating_sub(FANOUT_KEEP_MS));
414 let purged = match db.prepare("DELETE FROM fanout_sent WHERE created_at < ?").bind(&[cutoff.into()]) {
415 Ok(statement) => statement.run().await.map(|_| ()),
416 Err(error) => Err(error),
417 };
418 if let Err(error) = purged {
419 worker::console_error!("could not remove old fan-out records: {error}");
420 }
Inbox: the events service tells people what needs them as events arrive421 match inbox::purge(&db, now).await {
422 Ok(removed) if removed > 0 => worker::console_log!("removed {removed} old inbox items"),
423 Ok(_) => {}
424 Err(error) => worker::console_error!("could not remove old inbox items: {error}"),
425 }
README: run the core forge locally with Docker Compose; forks described as copies, not copy-on-write, until Cloudflare says otherwise426 match audit::purge(&db, &audit::keep_from(now, max_days), 20).await {
427 Ok(removed) if removed > 0 => {
428 worker::console_log!("removed {removed} audit entries older than {max_days} days")
429 }
430 Ok(_) => {}
431 Err(error) => worker::console_error!("could not remove old audit entries: {error}"),
432 }
433 // Without billing nobody's plan is known, so nothing younger than the
434 // ceiling is removed.
435 let billing = match env.service("BILLING") {
436 Ok(billing) => billing,
437 Err(error) => {
438 worker::console_error!(
439 "audit entries kept past their plan's days: no billing: {error}"
440 );
441 return;
442 }
443 };
444 match audit::purge_by_plan(&db, &billing, now, min_days, max_days).await {
445 Ok(removed) if removed > 0 => {
446 worker::console_log!("removed {removed} audit entries older than their plan keeps")
447 }
448 Ok(_) => {}
449 Err(error) => worker::console_error!("audit entries kept past their plan's days: {error}"),
450 }
451}
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)452
Merge branch 'worktree-agent-ad531323a21f4c975'453/// The condition of an `after` cursor, binding its id twice.
454const AFTER: &str = concat!(
455 "seq > COALESCE(",
456 "(SELECT seq FROM events WHERE id = ?), ",
457 "(SELECT seq FROM events WHERE id > ? AND seq IS NOT NULL ORDER BY id LIMIT 1) - 1, ",
458 "(SELECT MAX(seq) FROM events), 0)"
459);
460
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)461/// A value bound to a `?` in [`list_filter`]'s clause.
462#[derive(Debug, PartialEq)]
463enum Bind {
464 Text(String),
465 Number(f64),
466}
467
468/// The `WHERE` clause `list` reads with, and what it binds, in order.
469fn list_filter(a: &ListArgs) -> (String, Vec<Bind>) {
470 let mut conditions = Vec::new();
471 let mut values = Vec::new();
472 if let Some(repo_id) = &a.repo_id {
473 conditions.push("repo_id = ?".to_owned());
474 values.push(Bind::Text(repo_id.clone()));
475 }
g1t coordinates the work and the person's own sessions do it: Ask g1t reads g1t with the same tools any agent gets, answers in one conversation shared with the @g1t direct message in Chat, and hands work to sessions in the person's own harness, whose subagents g1t records, ties to issues and pull requests, and lets see each other.476 if !a.repo_ids.is_empty() {
477 // A workspace's activity across its repositories: one list, bound
478 // as JSON so the statement stays small however many there are.
479 let ids: Vec<&String> = a.repo_ids.iter().take(MAX_LIST_REPOS).collect();
480 conditions.push("repo_id IN (SELECT value FROM json_each(?))".to_owned());
481 values.push(Bind::Text(serde_json::to_string(&ids).unwrap_or_else(|_| "[]".to_owned())));
482 }
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)483 if !a.types.is_empty() {
484 let marks = vec!["?"; a.types.len()].join(", ");
485 conditions.push(format!("type IN ({marks})"));
486 values.extend(a.types.iter().map(|kind| Bind::Text(kind.clone())));
487 } else {
488 // What CI and integrations report on commits goes to webhooks,
489 // and is read from each commit's checks; a timeline asked for
490 // everything would be little else on a busy repository.
491 let marks = vec!["?"; REPORTING.len()].join(", ");
492 conditions.push(format!("type NOT IN ({marks})"));
493 values.extend(REPORTING.iter().map(|kind| Bind::Text((*kind).to_owned())));
494 }
495 if let Some(actor) = &a.actor {
496 conditions.push("actor = ?".to_owned());
497 values.push(Bind::Text(actor.clone()));
498 }
499 if !a.numbers.is_empty() {
500 // An issue or pull request's own events name it as `number`; a
501 // comment, review or link on it names it as `issue`.
502 let marks = vec!["?"; a.numbers.len()].join(", ");
503 conditions.push(format!(
504 "(json_extract(data, '$.number') IN ({marks}) OR json_extract(data, '$.issue') IN ({marks}))"
505 ));
506 for _ in 0..2 {
507 values.extend(a.numbers.iter().map(|number| Bind::Number(f64::from(*number))));
508 }
509 }
510 if let Some(since) = &a.since {
511 conditions.push("time >= ?".to_owned());
512 values.push(Bind::Text(since.clone()));
513 }
514 if let Some(before) = &a.before {
515 conditions.push("id < ?".to_owned());
516 values.push(Bind::Text(before.clone()));
517 }
Merge branch 'worktree-agent-ad531323a21f4c975'518 if let Some(after) = &a.after {
519 // What the log wrote after the cursor's event. A cursor the log
520 // does not have (yet: an id from a webhook can arrive before its
521 // row is written) starts at the first event with a later id, and
522 // one past them all at the end of the log.
523 conditions.push(AFTER.to_owned());
524 values.push(Bind::Text(after.clone()));
525 values.push(Bind::Text(after.clone()));
526 }
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)527 let filter = if conditions.is_empty() {
528 String::new()
529 } else {
530 format!("WHERE {}", conditions.join(" AND "))
531 };
532 (filter, values)
533}
534
535#[cfg(test)]
536mod tests {
537 use super::*;
538
539 #[test]
540 fn a_plain_list_leaves_out_what_is_reported_on_commits() {
541 let (filter, binds) = list_filter(&ListArgs { repo_id: Some("rep_1".into()), ..ListArgs::default() });
542 assert!(filter.starts_with("WHERE repo_id = ? AND type NOT IN ("));
543 assert_eq!(binds[0], Bind::Text("rep_1".into()));
544 assert_eq!(binds.len(), 1 + REPORTING.len());
545 }
546
547 #[test]
g1t coordinates the work and the person's own sessions do it: Ask g1t reads g1t with the same tools any agent gets, answers in one conversation shared with the @g1t direct message in Chat, and hands work to sessions in the person's own harness, whose subagents g1t records, ties to issues and pull requests, and lets see each other.548 fn a_workspaces_repositories_are_read_as_one_list() {
549 let many: Vec<String> = (0..150).map(|n| format!("rep_{n}")).collect();
550 let (filter, binds) = list_filter(&ListArgs {
551 repo_ids: many,
552 types: vec!["pull.merged".into(), "release.published".into()],
553 since: Some("2026-10-12T00:00:00Z".into()),
554 ..ListArgs::default()
555 });
556 assert_eq!(filter, "WHERE repo_id IN (SELECT value FROM json_each(?)) AND type IN (?, ?) AND time >= ?");
557 let Bind::Text(ids) = &binds[0] else { panic!("the ids are one JSON text") };
558 let ids: Vec<String> = serde_json::from_str(ids).unwrap();
559 assert_eq!(ids.len(), MAX_LIST_REPOS);
560 assert_eq!(ids[0], "rep_0");
561 }
562
563 #[test]
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)564 fn numbers_match_an_item_or_what_is_said_on_it_since_a_time() {
565 let (filter, binds) = list_filter(&ListArgs {
566 repo_id: Some("rep_1".into()),
567 types: vec!["issue.opened".into()],
568 numbers: vec![4, 9],
569 since: Some("2026-10-01T00:00:00Z".into()),
570 actor: Some("usr_g1t_agent".into()),
571 ..ListArgs::default()
572 });
573 assert_eq!(
574 filter,
575 "WHERE repo_id = ? AND type IN (?) AND actor = ? AND \
576 (json_extract(data, '$.number') IN (?, ?) OR json_extract(data, '$.issue') IN (?, ?)) AND time >= ?"
577 );
578 assert_eq!(
579 binds,
580 vec![
581 Bind::Text("rep_1".into()),
582 Bind::Text("issue.opened".into()),
583 Bind::Text("usr_g1t_agent".into()),
584 Bind::Number(4.0),
585 Bind::Number(9.0),
586 Bind::Number(4.0),
587 Bind::Number(9.0),
588 Bind::Text("2026-10-01T00:00:00Z".into()),
589 ]
590 );
591 }
Merge branch 'worktree-agent-ad531323a21f4c975'592
593 #[test]
594 fn an_after_cursor_reads_what_the_log_wrote_later_with_every_other_filter() {
595 let (filter, binds) = list_filter(&ListArgs {
596 repo_id: Some("rep_1".into()),
597 types: vec!["work.claimed".into(), "integration.failed".into()],
598 numbers: vec![7],
599 since: Some("2026-10-12T00:00:00Z".into()),
600 after: Some("evt_9".into()),
601 ..ListArgs::default()
602 });
603 assert_eq!(
604 filter,
605 format!(
606 "WHERE repo_id = ? AND type IN (?, ?) AND \
607 (json_extract(data, '$.number') IN (?) OR json_extract(data, '$.issue') IN (?)) AND time >= ? AND {AFTER}"
608 )
609 );
610 // The cursor's id is bound once to find its place, and once for a
611 // cursor the log does not have.
612 assert_eq!(&binds[binds.len() - 2..], &[Bind::Text("evt_9".into()), Bind::Text("evt_9".into())]);
613 assert_eq!(binds.len(), 8);
614 assert_eq!(AFTER.matches('?').count(), 2);
615 }
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)616}

This file's history is long; its oldest lines are credited to the oldest commit read.