| 1 | //! g1t taking on what it is asked in an issue or a pull request: assigned |
| 2 | //! the issue (`issue.assigned` adding `g1t`), or mentioned in a comment |
| 3 | //! (`comment.created` whose text says `@g1t`). |
| 4 | //! |
| 5 | //! g1t reacts 👀 to what asked it (the comment, or the issue for an |
| 6 | //! assignment) and starts a session on the workspace's own Claude, as the |
| 7 | //! person who asked ([`Integrations::start_agent_session`]): a pull |
| 8 | //! request's mention carries on with that pull request on its branch, and |
| 9 | //! an issue's with the issue. The comment is the session's instruction, so |
| 10 | //! "that's not what I meant" reaches it, along with where to find what the |
| 11 | //! sessions before it did. A mention on a thread where that same person's |
| 12 | //! session is still going is passed to that session instead of starting |
| 13 | //! another. The session shows in the thread's timeline as it goes. |
| 14 | //! |
| 15 | //! Only the workspace's members can ask; anyone else's mention is an |
| 16 | //! ordinary one. When the session cannot start (no Anthropic key, say), |
| 17 | //! g1t says why in the thread. Each ask is acted on once (`summons`), and |
| 18 | //! a thread is answered at most [`MAX_PER_HOUR`] times an hour, so a |
| 19 | //! session that writes `@g1t` itself cannot start sessions without end. |
| 20 | |
| 21 | use g1t_contracts::agent_sessions::{AgentMode, Asked, StartAgentSessionArgs}; |
| 22 | use g1t_contracts::events::Event; |
| 23 | use g1t_contracts::identity::UserIdArgs; |
| 24 | use g1t_contracts::repos::{PathByIdArgs, RepoPath}; |
| 25 | use g1t_contracts::time::rfc3339; |
| 26 | use g1t_contracts::updates::update_command; |
| 27 | use g1t_contracts::work::{Comment, CommentKind, IssueDetail, PullDetail, ReactArgs, ReactionContent, ViewArgs}; |
| 28 | use g1t_contracts::{FailureCode, Outcome, User, Viewer, system}; |
| 29 | use g1t_kit::now_ms; |
| 30 | use serde::Deserialize; |
| 31 | use serde_json::Value; |
| 32 | use worker::Result; |
| 33 | |
| 34 | use crate::Integrations; |
| 35 | |
| 36 | /// The most times g1t answers one thread in an hour. |
| 37 | pub const MAX_PER_HOUR: u32 = 6; |
| 38 | const HOUR_MS: u64 = 60 * 60 * 1000; |
| 39 | |
| 40 | /// What g1t was asked, read from an event before anything is looked up. |
| 41 | #[derive(Debug, PartialEq, Eq)] |
| 42 | pub enum Ask { |
| 43 | /// Assigned issue `number`. |
| 44 | Assigned { number: u32 }, |
| 45 | /// Comment `comment_id` on issue or pull request `number` may mention it. |
| 46 | Commented { number: u32, comment_id: String }, |
| 47 | } |
| 48 | |
| 49 | /// What an event asks of g1t, if anything. |
| 50 | pub fn ask_in(event: &Event) -> Option<Ask> { |
| 51 | let actor = event.actor.as_deref()?; |
| 52 | // g1t's own doings never ask it anything. |
| 53 | if system::is_system_id(actor) { |
| 54 | return None; |
| 55 | } |
| 56 | let number = event.data["number"].as_u64().and_then(|n| u32::try_from(n).ok())?; |
| 57 | match event.kind.as_str() { |
| 58 | "issue.assigned" => { |
| 59 | let added = event.data["added"].as_array()?; |
| 60 | added.iter().any(|name| name.as_str() == Some(system::USERNAME)).then_some(Ask::Assigned { number }) |
| 61 | } |
| 62 | "comment.created" => { |
| 63 | // An agent writing as itself, from before agents were removed, never asks. |
| 64 | if !event.data["agent"].is_null() { |
| 65 | return None; |
| 66 | } |
| 67 | let comment_id = event.data["commentId"].as_str()?.to_owned(); |
| 68 | Some(Ask::Commented { number, comment_id }) |
| 69 | } |
| 70 | _ => None, |
| 71 | } |
| 72 | } |
| 73 | |
| 74 | /// Whether `text` mentions `@g1t`: outside code and quotes, and not as part |
| 75 | /// of a longer name (`@g1t-agent`) or a team (`@g1t/core`). |
| 76 | pub fn mentions_g1t(text: &str) -> bool { |
| 77 | let text = text.to_ascii_lowercase(); |
| 78 | let mut fenced = false; |
| 79 | for line in text.lines() { |
| 80 | let trimmed = line.trim_start(); |
| 81 | if trimmed.starts_with("```") || trimmed.starts_with("~~~") { |
| 82 | fenced = !fenced; |
| 83 | continue; |
| 84 | } |
| 85 | if fenced || trimmed.starts_with('>') || line.starts_with(" ") || line.starts_with('\t') { |
| 86 | continue; |
| 87 | } |
| 88 | // Inline code: every other run between backticks. |
| 89 | for (at, part) in trimmed.split('`').enumerate() { |
| 90 | if at % 2 == 1 { |
| 91 | continue; |
| 92 | } |
| 93 | let bytes = part.as_bytes(); |
| 94 | let mut from = 0; |
| 95 | while let Some(found) = part[from..].find("@g1t") { |
| 96 | let start = from + found; |
| 97 | let end = start + 4; |
| 98 | let before = start.checked_sub(1).map(|i| bytes[i]); |
| 99 | let after = bytes.get(end).copied(); |
| 100 | let word = |b: u8| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'/' | b'.' | b'@'); |
| 101 | let ends = |b: u8| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'/'); |
| 102 | if !before.is_some_and(word) && !after.is_some_and(ends) { |
| 103 | return true; |
| 104 | } |
| 105 | from = end; |
| 106 | } |
| 107 | } |
| 108 | } |
| 109 | false |
| 110 | } |
| 111 | |
| 112 | #[derive(Deserialize)] |
| 113 | struct Counted { |
| 114 | n: u32, |
| 115 | } |
| 116 | |
| 117 | #[derive(Deserialize)] |
| 118 | struct Running { |
| 119 | provider_session_id: String, |
| 120 | connection_id: String, |
| 121 | } |
| 122 | |
| 123 | /// The thread asked in, as the asker sees it. |
| 124 | struct Thread { |
| 125 | /// `issues` or `pull`, for links. |
| 126 | page: &'static str, |
| 127 | comments: Vec<Comment>, |
| 128 | /// The issue's number, or the pull request's issue's, if any. |
| 129 | issue: Option<u32>, |
| 130 | /// Set for a pull request: its number. |
| 131 | pull: Option<u32>, |
| 132 | /// A pull request g1t opened (an update), whose `@g1t` commands are |
| 133 | /// the security service's. |
| 134 | g1t_pull: bool, |
| 135 | open: bool, |
| 136 | } |
| 137 | |
| 138 | impl Integrations { |
| 139 | /// Acts on what an event asks of g1t. Never fails the batch: a problem |
| 140 | /// is logged, and the ask is not tried again. |
| 141 | pub(crate) async fn summon(&self, event: &Event) { |
| 142 | let Some(ask) = ask_in(event) else { return }; |
| 143 | if let Err(error) = self.answer_ask(event, ask).await { |
| 144 | worker::console_error!("summon: {} {}: {error}", event.kind, event.id); |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | async fn answer_ask(&self, event: &Event, ask: Ask) -> Result<()> { |
| 149 | let (Some(repo_id), Some(actor_id)) = (event.repo_id.as_deref(), event.actor.as_deref()) else { return Ok(()) }; |
| 150 | let Some(repo): Option<RepoPath> = g1t_kit::call(&self.repos, "path_by_id", &PathByIdArgs { id: repo_id.to_owned() }).await? else { |
| 151 | return Ok(()); |
| 152 | }; |
| 153 | let workspace = repo.namespace.to_lowercase(); |
| 154 | // Who asked, as they are now: only the workspace's members may. |
| 155 | let actor: Viewer = g1t_kit::call(&self.identity, "user_by_id", &UserIdArgs { id: actor_id.to_owned() }).await?; |
| 156 | let Some(actor) = actor.filter(|actor| actor.verified && actor.is_member(&workspace)) else { return Ok(()) }; |
| 157 | let number = match &ask { |
| 158 | Ask::Assigned { number } | Ask::Commented { number, .. } => *number, |
| 159 | }; |
| 160 | let Some(thread) = self.thread(&actor, &repo, number).await? else { return Ok(()) }; |
| 161 | let site = self.site_url.trim_end_matches('/'); |
| 162 | let thread_url = format!("{site}/{}/{}/{}/{number}", repo.namespace, repo.name, thread.page); |
| 163 | let (comment, url) = match &ask { |
| 164 | Ask::Assigned { .. } => (None, thread_url.clone()), |
| 165 | Ask::Commented { comment_id, .. } => { |
| 166 | let Some(comment) = thread.comments.iter().find(|comment| &comment.id == comment_id && comment.kind == CommentKind::Comment) else { |
| 167 | return Ok(()); |
| 168 | }; |
| 169 | if !mentions_g1t(&comment.body) { |
| 170 | return Ok(()); |
| 171 | } |
| 172 | // `@g1t rebase` on an update g1t opened is the security service's. |
| 173 | if thread.g1t_pull && update_command(&comment.body).is_some() { |
| 174 | return Ok(()); |
| 175 | } |
| 176 | (Some(comment), format!("{thread_url}#{}", comment.id)) |
| 177 | } |
| 178 | }; |
| 179 | if !thread.open { |
| 180 | return Ok(()); |
| 181 | } |
| 182 | // Once per ask, and a few times an hour per thread. |
| 183 | if !self.first_time(&event.id, &repo, number).await? { |
| 184 | return Ok(()); |
| 185 | } |
| 186 | let react_to = comment.map(|comment| comment.id.clone()); |
| 187 | let system = User::system(&workspace); |
| 188 | let _: Outcome<Value> = g1t_kit::call( |
| 189 | &self.work, |
| 190 | "react", |
| 191 | &ReactArgs { actor: system.clone(), repo: repo.clone(), number, comment_id: react_to, content: ReactionContent::Eyes, remove: false }, |
| 192 | ) |
| 193 | .await?; |
| 194 | let text = comment.map(|comment| comment.body.clone()); |
| 195 | // The asker's session on this thread, still going: told, not doubled. |
| 196 | if let Some(text) = &text |
| 197 | && self.tell_running(&repo, &actor, &[Some(number), thread.issue].into_iter().flatten().collect::<Vec<_>>(), text, &url).await? |
| 198 | { |
| 199 | return Ok(()); |
| 200 | } |
| 201 | let started = self |
| 202 | .start_agent_session(StartAgentSessionArgs { |
| 203 | actor: actor.clone(), |
| 204 | repo: repo.clone(), |
| 205 | issue: thread.issue.filter(|_| thread.pull.is_none()), |
| 206 | pull: thread.pull, |
| 207 | asked: Some(Asked { by: actor.username.clone(), comment: text, url }), |
| 208 | title: None, |
| 209 | body: None, |
| 210 | instruction: None, |
| 211 | mode: Some(AgentMode::Implement), |
| 212 | handoff: Some(false), |
| 213 | }) |
| 214 | .await?; |
| 215 | if let Outcome::Fail(failure) = started { |
| 216 | let mut body = format!("@{}, I can't take this on: {}.", actor.username, failure.message.trim().trim_end_matches('.')); |
| 217 | // No Anthropic account to start on: where to add one. |
| 218 | if failure.code == FailureCode::Conflict { |
| 219 | body.push_str(&format!(" Add an Anthropic key in [the workspace's AI settings]({site}/{workspace}/-/ai), then ask again.")); |
| 220 | } |
| 221 | self.comment(&system, &repo, number, body).await?; |
| 222 | } |
| 223 | Ok(()) |
| 224 | } |
| 225 | |
| 226 | /// The thread `number`, issue or pull request, as `actor` may see it. |
| 227 | async fn thread(&self, actor: &User, repo: &RepoPath, number: u32) -> Result<Option<Thread>> { |
| 228 | let view = ViewArgs { repo: repo.clone(), number, viewer: Some(actor.clone()), after_seq: 0 }; |
| 229 | let issue: Outcome<IssueDetail> = g1t_kit::call(&self.work, "get_issue", &view).await?; |
| 230 | if let Outcome::Ok(detail) = issue { |
| 231 | return Ok(Some(Thread { |
| 232 | page: "issues", |
| 233 | open: detail.issue.state == g1t_contracts::work::State::Open, |
| 234 | comments: detail.comments, |
| 235 | issue: Some(number), |
| 236 | pull: None, |
| 237 | g1t_pull: false, |
| 238 | })); |
| 239 | } |
| 240 | let pull: Outcome<PullDetail> = g1t_kit::call(&self.work, "get_pull", &view).await?; |
| 241 | Ok(match pull { |
| 242 | Outcome::Ok(detail) => Some(Thread { |
| 243 | page: "pull", |
| 244 | open: detail.pull.status.is_active(), |
| 245 | g1t_pull: detail.pull.author.is_system() || system::is_system_id(&detail.pull.author.id), |
| 246 | issue: detail.pull.issue, |
| 247 | pull: Some(number), |
| 248 | comments: detail.comments, |
| 249 | }), |
| 250 | Outcome::Fail(_) => None, |
| 251 | }) |
| 252 | } |
| 253 | |
| 254 | /// Records ask `event_id` on thread `number`: whether it is new and the |
| 255 | /// thread has not been answered [`MAX_PER_HOUR`] times this hour. |
| 256 | async fn first_time(&self, event_id: &str, repo: &RepoPath, number: u32) -> Result<bool> { |
| 257 | let now = now_ms(); |
| 258 | let thread = format!("{}/{}#{number}", repo.namespace.to_lowercase(), repo.name); |
| 259 | let recent = self |
| 260 | .db |
| 261 | .prepare("SELECT COUNT(*) AS n FROM summons WHERE thread = ? AND at > ?") |
| 262 | .bind(&[thread.as_str().into(), rfc3339(now.saturating_sub(HOUR_MS)).into()])? |
| 263 | .first::<Counted>(None) |
| 264 | .await? |
| 265 | .map_or(0, |row| row.n); |
| 266 | if recent >= MAX_PER_HOUR { |
| 267 | worker::console_log!("summon: {thread} asked {recent} times this hour; {event_id} let be"); |
| 268 | return Ok(false); |
| 269 | } |
| 270 | let inserted = self |
| 271 | .db |
| 272 | .prepare("INSERT INTO summons (event_id, thread, at) VALUES (?, ?, ?) ON CONFLICT DO NOTHING RETURNING event_id") |
| 273 | .bind(&[event_id.into(), thread.as_str().into(), rfc3339(now).into()])? |
| 274 | .first::<Value>(None) |
| 275 | .await?; |
| 276 | Ok(inserted.is_some()) |
| 277 | } |
| 278 | |
| 279 | /// Passes `text` to `actor`'s own session on `numbers` in `repo` that |
| 280 | /// g1t started and is still going. Whether there was one. |
| 281 | async fn tell_running(&self, repo: &RepoPath, actor: &User, numbers: &[u32], text: &str, url: &str) -> Result<bool> { |
| 282 | let repo_name = format!("{}/{}", repo.namespace.to_lowercase(), repo.name); |
| 283 | let running = self |
| 284 | .db |
| 285 | .prepare( |
| 286 | "SELECT provider_session_id, connection_id FROM agent_runs |
| 287 | WHERE repo = ? AND status = 'running' AND json_extract(actor, '$.id') = ? |
| 288 | AND issue IN (SELECT value FROM json_each(?)) |
| 289 | ORDER BY created_at DESC LIMIT 1", |
| 290 | ) |
| 291 | .bind(&[repo_name.as_str().into(), actor.id.as_str().into(), serde_json::to_string(numbers)?.into()])? |
| 292 | .first::<Running>(None) |
| 293 | .await?; |
| 294 | let Some(running) = running else { return Ok(false) }; |
| 295 | let Some(row) = self.ai_row(&running.connection_id).await? else { return Ok(false) }; |
| 296 | let key = self.ai_secret(&row).unwrap_or_default(); |
| 297 | let message = format!( |
| 298 | "@{} said, in a comment on g1t ({url}):\n\n<comment>\n{}\n</comment>\n\nTake it into account, and answer them in that thread.", |
| 299 | actor.username, |
| 300 | text.trim() |
| 301 | ); |
| 302 | let sent = self |
| 303 | .call_anthropic(&key, worker::Method::Post, &format!("/v1/sessions/{}/events", running.provider_session_id), Some(crate::agent_sessions::message_event(&message))) |
| 304 | .await?; |
| 305 | Ok(sent.is_ok()) |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | /// The cron: asks older than a week are forgotten. |
| 310 | pub async fn forget_old(service: &Integrations) { |
| 311 | let before = rfc3339(now_ms().saturating_sub(7 * 24 * HOUR_MS)); |
| 312 | let forgot = async { service.db.prepare("DELETE FROM summons WHERE at < ?").bind(&[before.into()])?.run().await }; |
| 313 | if let Err(error) = forgot.await { |
| 314 | worker::console_error!("summon: old asks not cleared: {error}"); |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | #[cfg(test)] |
| 319 | mod tests { |
| 320 | use super::*; |
| 321 | use serde_json::json; |
| 322 | |
| 323 | fn event(kind: &str, actor: &str, data: Value) -> Event { |
| 324 | Event { |
| 325 | id: "evt_1".into(), |
| 326 | kind: kind.into(), |
| 327 | source: "work".into(), |
| 328 | time: String::new(), |
| 329 | repo_id: Some("rep_1".into()), |
| 330 | actor: Some(actor.into()), |
| 331 | data, |
| 332 | } |
| 333 | } |
| 334 | |
| 335 | #[test] |
| 336 | fn a_mention_is_an_at_g1t_outside_code_and_quotes() { |
| 337 | assert!(mentions_g1t("@g1t try again, the test is still red")); |
| 338 | assert!(mentions_g1t("Thanks. @g1t: use the other API.")); |
| 339 | assert!(mentions_g1t("(@g1t) please")); |
| 340 | assert!(mentions_g1t("Hey @G1T")); |
| 341 | assert!(!mentions_g1t("@g1t-agent was retired"), "a longer name"); |
| 342 | assert!(!mentions_g1t("ask @g1t/core"), "a team"); |
| 343 | assert!(!mentions_g1t("mail g1t@users.noreply.g1t.sh")); |
| 344 | assert!(!mentions_g1t("run `@g1t rebase` on updates"), "inline code"); |
| 345 | assert!(!mentions_g1t("```\n@g1t\n```"), "a code block"); |
| 346 | assert!(!mentions_g1t("> @g1t do it\n\nI disagree."), "a quote"); |
| 347 | assert!(mentions_g1t("```\nlog\n```\n@g1t look at the log above")); |
| 348 | } |
| 349 | |
| 350 | #[test] |
| 351 | fn assigning_g1t_and_commenting_ask_and_nothing_g1t_does_does() { |
| 352 | assert_eq!( |
| 353 | ask_in(&event("issue.assigned", "usr_ana", json!({ "number": 4, "added": ["g1t"] }))), |
| 354 | Some(Ask::Assigned { number: 4 }) |
| 355 | ); |
| 356 | assert_eq!(ask_in(&event("issue.assigned", "usr_ana", json!({ "number": 4, "added": ["bo"] }))), None); |
| 357 | assert_eq!( |
| 358 | ask_in(&event("comment.created", "usr_ana", json!({ "number": 4, "commentId": "cmt_1" }))), |
| 359 | Some(Ask::Commented { number: 4, comment_id: "cmt_1".into() }) |
| 360 | ); |
| 361 | assert_eq!(ask_in(&event("comment.created", "g1t", json!({ "number": 4, "commentId": "cmt_1" }))), None); |
| 362 | assert_eq!( |
| 363 | ask_in(&event("comment.created", "usr_ana", json!({ "number": 4, "commentId": "cmt_1", "agent": { "id": "agt_1" } }))), |
| 364 | None |
| 365 | ); |
| 366 | assert_eq!(ask_in(&event("issue.closed", "usr_ana", json!({ "number": 4 }))), None); |
| 367 | } |
| 368 | } |