| 1 | //! The plan reads itself (migrations/0057_plan_reads_itself.sql), on the |
| 2 | //! service's cron, a little at a time: |
| 3 | //! |
| 4 | //! - A work item in Triage gets a suggestion of where it goes: a horizon, |
| 5 | //! an appetite, an open work item it seems to repeat, and why, asked once |
| 6 | //! of the workspace's own AI provider through integrations' `ai_brief` |
| 7 | //! (g1t runs no model of its own). A person accepts it or not. |
| 8 | //! - A work item in Now gets a sentence or two on where it stands, drafted |
| 9 | //! from its record (sessions, decisions, pull requests) whenever |
| 10 | //! something new has happened on it. |
| 11 | //! - A work item in Later with nothing happening on it for |
| 12 | //! [`SET_ASIDE_DAYS`] is set aside, saying so in its conversation. |
| 13 | //! |
| 14 | //! A workspace with no provider chosen gets no suggestions and no drafts; |
| 15 | //! the rest of the plan works the same without them. |
| 16 | |
| 17 | use std::collections::HashSet; |
| 18 | |
| 19 | use g1t_contracts::ai::{AiBriefArgs, MAX_BRIEF_INPUT_CHARS}; |
| 20 | use g1t_contracts::events::WorkItemEvent; |
| 21 | use g1t_contracts::time::rfc3339; |
| 22 | use g1t_contracts::work::*; |
| 23 | use g1t_contracts::Outcome; |
| 24 | use g1t_kit::now_ms; |
| 25 | use serde::Deserialize; |
| 26 | use worker::Result; |
| 27 | |
| 28 | use super::plan::days_before; |
| 29 | use crate::Work; |
| 30 | |
| 31 | /// The most the provider is asked in one run of the cron, across |
| 32 | /// workspaces: suggestions and drafts together. |
| 33 | const CALLS_PER_RUN: u32 = 12; |
| 34 | /// The most set aside in one run. |
| 35 | const SET_ASIDE_PER_RUN: u32 = 50; |
| 36 | /// The most of the plan a suggestion is asked with. |
| 37 | const PLAN_LINES: u32 = 80; |
| 38 | /// The longest suggestion reason and standing kept, in characters. |
| 39 | const WHY_CHARS: usize = 240; |
| 40 | const STANDING_CHARS: usize = 320; |
| 41 | |
| 42 | const SUGGEST_SYSTEM: &str = "You place a new work item in a software team's plan. \ |
| 43 | The horizons: now (what the team is on; only for something urgent or blocking), \ |
| 44 | next (ranked, taken soon), later (wanted, not soon). \ |
| 45 | The appetites: small (one session), medium (a few sessions, about a week), large (a phase to break into smaller items). \ |
| 46 | If it repeats one of the open work items listed, give that one's number. \ |
| 47 | Answer in exactly four lines and nothing else:\n\ |
| 48 | horizon: now|next|later\n\ |
| 49 | appetite: small|medium|large\n\ |
| 50 | duplicate: <number or none>\n\ |
| 51 | why: <one plain sentence>"; |
| 52 | |
| 53 | const STANDING_SYSTEM: &str = "From the record given, say where this piece of work stands now and what comes next, \ |
| 54 | in one or two plain sentences. No preamble, no lists, no headings. Name pull requests and work items as #N."; |
| 55 | |
| 56 | /// What one run may still ask the provider, and the workspaces that have |
| 57 | /// none (asked again on the next run). |
| 58 | struct Budget { |
| 59 | calls: u32, |
| 60 | without: HashSet<String>, |
| 61 | } |
| 62 | |
| 63 | /// A suggestion read from the provider's answer: none when it does not |
| 64 | /// say a horizon it may suggest. `known` are the open work items it may |
| 65 | /// say this one repeats. |
| 66 | pub(crate) fn read_suggestion(answer: &str, known: &[u32], at: &str) -> Option<PlanSuggestion> { |
| 67 | let mut horizon = None; |
| 68 | let mut appetite = None; |
| 69 | let mut duplicate_of = None; |
| 70 | let mut why = String::new(); |
| 71 | for line in answer.lines() { |
| 72 | let Some((key, value)) = line.split_once(':') else { |
| 73 | continue; |
| 74 | }; |
| 75 | let value = value.trim().trim_matches(|c: char| c == '*' || c == '`' || c == '"').trim(); |
| 76 | match key.trim().trim_matches(|c: char| c == '*' || c == '-').trim().to_ascii_lowercase().as_str() { |
| 77 | "horizon" => horizon = Horizon::parse(value).filter(|h| matches!(h, Horizon::Now | Horizon::Next | Horizon::Later)), |
| 78 | "appetite" => appetite = Appetite::parse(value), |
| 79 | "duplicate" => { |
| 80 | duplicate_of = value.trim_start_matches('#').parse::<u32>().ok().filter(|number| known.contains(number)); |
| 81 | } |
| 82 | "why" => why = clip(value, WHY_CHARS), |
| 83 | _ => {} |
| 84 | } |
| 85 | } |
| 86 | Some(PlanSuggestion { horizon: horizon?, appetite, duplicate_of, why, at: at.to_owned() }) |
| 87 | } |
| 88 | |
| 89 | /// `text` on one line, at most `limit` characters, cut at a word. |
| 90 | pub(crate) fn clip(text: &str, limit: usize) -> String { |
| 91 | let flat = text.split_whitespace().collect::<Vec<_>>().join(" "); |
| 92 | if flat.chars().count() <= limit { |
| 93 | return flat; |
| 94 | } |
| 95 | let cut: String = flat.chars().take(limit).collect(); |
| 96 | let cut = cut.rsplit_once(' ').map_or(cut.as_str(), |(head, _)| head).trim_end_matches([',', ';', ':', '.']); |
| 97 | format!("{cut}…") |
| 98 | } |
| 99 | |
| 100 | /// A drafted standing as kept: one paragraph, bounded; none when the |
| 101 | /// answer has nothing in it. |
| 102 | pub(crate) fn read_standing(answer: &str) -> Option<String> { |
| 103 | // A list marker or bold the model put in front, never a #N it begins with. |
| 104 | let trimmed = answer.trim(); |
| 105 | let trimmed = trimmed.strip_prefix("- ").or_else(|| trimmed.strip_prefix("* ")).unwrap_or(trimmed).trim_matches('*'); |
| 106 | let text = clip(trimmed, STANDING_CHARS); |
| 107 | (!text.is_empty()).then_some(text) |
| 108 | } |
| 109 | |
| 110 | #[derive(Deserialize)] |
| 111 | struct Pending { |
| 112 | id: String, |
| 113 | workspace: String, |
| 114 | number: u32, |
| 115 | title: String, |
| 116 | body: String, |
| 117 | #[serde(default)] |
| 118 | origin: Option<String>, |
| 119 | } |
| 120 | |
| 121 | #[derive(Deserialize)] |
| 122 | struct PlanLine { |
| 123 | number: u32, |
| 124 | horizon: String, |
| 125 | title: String, |
| 126 | } |
| 127 | |
| 128 | #[derive(Deserialize)] |
| 129 | struct Line { |
| 130 | #[serde(default)] |
| 131 | title: Option<String>, |
| 132 | #[serde(default)] |
| 133 | state: Option<String>, |
| 134 | #[serde(default)] |
| 135 | result: Option<String>, |
| 136 | #[serde(default)] |
| 137 | summary: Option<String>, |
| 138 | #[serde(default)] |
| 139 | number: Option<u32>, |
| 140 | #[serde(default)] |
| 141 | status: Option<String>, |
| 142 | } |
| 143 | |
| 144 | #[derive(Deserialize)] |
| 145 | struct Stale { |
| 146 | id: String, |
| 147 | } |
| 148 | |
| 149 | impl Work { |
| 150 | /// One run of the plan's upkeep: sets aside what has waited too long in |
| 151 | /// Later, then suggests and drafts within the run's budget. |
| 152 | pub(crate) async fn settle_plan(&self) -> Result<()> { |
| 153 | self.set_aside_stale().await?; |
| 154 | let mut budget = Budget { calls: CALLS_PER_RUN, without: HashSet::new() }; |
| 155 | self.suggest_triage(&mut budget).await?; |
| 156 | self.draft_standings(&mut budget).await?; |
| 157 | Ok(()) |
| 158 | } |
| 159 | |
| 160 | /// Asks the workspace's provider, within `budget`. None when it has no |
| 161 | /// provider, or could not answer (said in the log). |
| 162 | async fn brief(&self, workspace: &str, system: &str, input: String, max_tokens: u32, budget: &mut Budget) -> Option<String> { |
| 163 | if budget.calls == 0 || budget.without.contains(workspace) { |
| 164 | return None; |
| 165 | } |
| 166 | let integrations = self.integrations.as_ref()?; |
| 167 | budget.calls -= 1; |
| 168 | let input: String = input.chars().take(MAX_BRIEF_INPUT_CHARS).collect(); |
| 169 | let args = AiBriefArgs { workspace: workspace.to_owned(), system: system.to_owned(), input, max_tokens }; |
| 170 | match g1t_kit::call::<_, Outcome<Option<String>>>(integrations, "ai_brief", &args).await { |
| 171 | Ok(Outcome::Ok(Some(answer))) => Some(answer), |
| 172 | Ok(Outcome::Ok(None)) => { |
| 173 | budget.without.insert(workspace.to_owned()); |
| 174 | None |
| 175 | } |
| 176 | Ok(Outcome::Fail(failure)) => { |
| 177 | worker::console_error!("{workspace}: the plan got no answer from the AI provider: {}", failure.message); |
| 178 | budget.without.insert(workspace.to_owned()); |
| 179 | None |
| 180 | } |
| 181 | Err(error) => { |
| 182 | worker::console_error!("{workspace}: the plan got no answer from the AI provider: {error}"); |
| 183 | budget.without.insert(workspace.to_owned()); |
| 184 | None |
| 185 | } |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | /// Sets aside what has had nothing happen on it in Later for |
| 190 | /// [`SET_ASIDE_DAYS`]. |
| 191 | async fn set_aside_stale(&self) -> Result<()> { |
| 192 | let now = now_ms(); |
| 193 | let before = days_before(now, SET_ASIDE_DAYS); |
| 194 | let stale = self |
| 195 | .db |
| 196 | .prepare( |
| 197 | "SELECT id FROM work_items |
| 198 | WHERE state = 'open' AND horizon = 'later' AND number IS NOT NULL |
| 199 | AND max(updated_at, COALESCE(planned_at, '')) < ?1 |
| 200 | LIMIT ?2", |
| 201 | ) |
| 202 | .bind(&[before.as_str().into(), SET_ASIDE_PER_RUN.into()])? |
| 203 | .all() |
| 204 | .await? |
| 205 | .results::<Stale>()?; |
| 206 | let at = rfc3339(now); |
| 207 | for row in stale { |
| 208 | self.db |
| 209 | .prepare("UPDATE work_items SET horizon = 'set_aside', rank = NULL, planned_at = ?1 WHERE id = ?2 AND horizon = 'later'") |
| 210 | .bind(&[at.as_str().into(), row.id.as_str().into()])? |
| 211 | .run() |
| 212 | .await?; |
| 213 | let Some(held) = self.item_by_id(&row.id).await? else { |
| 214 | continue; |
| 215 | }; |
| 216 | self.publish_item( |
| 217 | "work_item.planned", |
| 218 | &held, |
| 219 | None, |
| 220 | WorkItemEvent { |
| 221 | horizon: Some(Horizon::SetAside), |
| 222 | from_horizon: Some(Horizon::Later), |
| 223 | appetite: held.item.appetite, |
| 224 | urgent: Some(false), |
| 225 | ..Self::item_event(&held) |
| 226 | }, |
| 227 | ) |
| 228 | .await?; |
| 229 | self.note_item( |
| 230 | &row.id, |
| 231 | (g1t_contracts::identity::AGENT_ID, "g1t"), |
| 232 | &format!("set this aside: nothing happened on it in Later for {SET_ASIDE_DAYS} days"), |
| 233 | ) |
| 234 | .await?; |
| 235 | } |
| 236 | Ok(()) |
| 237 | } |
| 238 | |
| 239 | /// Suggests where new work items in Triage go, the newest first. |
| 240 | async fn suggest_triage(&self, budget: &mut Budget) -> Result<()> { |
| 241 | let pending = self |
| 242 | .db |
| 243 | .prepare( |
| 244 | "SELECT w.id, w.workspace, w.number, w.title, w.body, |
| 245 | (SELECT COALESCE(rw.path, o.path) FROM work_item_repos o LEFT JOIN repo_workspaces rw ON rw.repo_id = o.repo_id |
| 246 | WHERE o.item_id = w.id AND o.role = 'origin') AS origin |
| 247 | FROM work_items w |
| 248 | WHERE w.state = 'open' AND COALESCE(w.horizon, 'triage') = 'triage' AND w.suggested_at IS NULL |
| 249 | AND w.number IS NOT NULL |
| 250 | ORDER BY w.created_at DESC LIMIT ?1", |
| 251 | ) |
| 252 | .bind(&[budget.calls.into()])? |
| 253 | .all() |
| 254 | .await? |
| 255 | .results::<Pending>()?; |
| 256 | for item in pending { |
| 257 | if budget.calls == 0 { |
| 258 | break; |
| 259 | } |
| 260 | let plan = self |
| 261 | .db |
| 262 | .prepare( |
| 263 | "SELECT number, COALESCE(horizon, 'triage') AS horizon, title FROM work_items |
| 264 | WHERE workspace = ?1 AND state = 'open' AND number IS NOT NULL AND id <> ?2 |
| 265 | ORDER BY CASE COALESCE(horizon, 'triage') WHEN 'now' THEN 0 WHEN 'next' THEN 1 WHEN 'later' THEN 2 |
| 266 | WHEN 'triage' THEN 3 ELSE 4 END, rank IS NULL, rank, number DESC |
| 267 | LIMIT ?3", |
| 268 | ) |
| 269 | .bind(&[item.workspace.as_str().into(), item.id.as_str().into(), PLAN_LINES.into()])? |
| 270 | .all() |
| 271 | .await? |
| 272 | .results::<PlanLine>()?; |
| 273 | let known: Vec<u32> = plan.iter().map(|line| line.number).collect(); |
| 274 | let listed: Vec<String> = plan.iter().map(|line| format!("#{} [{}] {}", line.number, line.horizon, line.title)).collect(); |
| 275 | let input = format!( |
| 276 | "New work item #{}{}: {}\n\n{}\n\nOpen work items:\n{}", |
| 277 | item.number, |
| 278 | item.origin.as_deref().map(|path| format!(" (filed in {path})")).unwrap_or_default(), |
| 279 | item.title, |
| 280 | clip(&item.body, 1_500), |
| 281 | listed.join("\n") |
| 282 | ); |
| 283 | let at = rfc3339(now_ms()); |
| 284 | let answer = self.brief(&item.workspace, SUGGEST_SYSTEM, input, 120, budget).await; |
| 285 | if answer.is_none() && !budget.without.contains(&item.workspace) { |
| 286 | // Out of calls for this run: asked again on the next. |
| 287 | break; |
| 288 | } |
| 289 | let suggestion = answer.as_deref().and_then(|answer| read_suggestion(answer, &known, &at)); |
| 290 | // Asked once: with no provider, or no usable answer, it is not |
| 291 | // asked again. |
| 292 | self.db |
| 293 | .prepare("UPDATE work_items SET suggestion = ?1, suggested_at = ?2 WHERE id = ?3") |
| 294 | .bind(&[ |
| 295 | suggestion.as_ref().map(serde_json::to_string).transpose()?.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), |
| 296 | at.as_str().into(), |
| 297 | item.id.as_str().into(), |
| 298 | ])? |
| 299 | .run() |
| 300 | .await?; |
| 301 | } |
| 302 | Ok(()) |
| 303 | } |
| 304 | |
| 305 | /// Drafts where work items in Now stand, those with something new |
| 306 | /// since their last draft first. |
| 307 | async fn draft_standings(&self, budget: &mut Budget) -> Result<()> { |
| 308 | #[derive(Deserialize)] |
| 309 | struct Due { |
| 310 | id: String, |
| 311 | workspace: String, |
| 312 | number: u32, |
| 313 | title: String, |
| 314 | } |
| 315 | let due = self |
| 316 | .db |
| 317 | .prepare( |
| 318 | "SELECT id, workspace, number, title FROM ( |
| 319 | SELECT w.id, w.workspace, w.number, w.title, w.standing_at, |
| 320 | max(COALESCE((SELECT max(s.last_seen_at) FROM sessions s WHERE s.item_id = w.id), ''), |
| 321 | COALESCE((SELECT max(d.created_at) FROM decisions d WHERE d.item_id = w.id), ''), |
| 322 | COALESCE((SELECT max(p.updated_at) FROM pulls p WHERE p.item_id = w.id), '')) AS active_at |
| 323 | FROM work_items w WHERE w.state = 'open' AND w.horizon = 'now' AND w.number IS NOT NULL) |
| 324 | WHERE active_at <> '' AND (standing_at IS NULL OR standing_at < active_at) |
| 325 | ORDER BY active_at DESC LIMIT ?1", |
| 326 | ) |
| 327 | .bind(&[budget.calls.into()])? |
| 328 | .all() |
| 329 | .await? |
| 330 | .results::<Due>()?; |
| 331 | for item in due { |
| 332 | if budget.calls == 0 { |
| 333 | break; |
| 334 | } |
| 335 | let key = [worker::wasm_bindgen::JsValue::from(item.id.as_str())]; |
| 336 | let statements = vec![ |
| 337 | self.db |
| 338 | .prepare("SELECT title, state, result FROM sessions WHERE item_id = ?1 ORDER BY last_seen_at DESC LIMIT 5") |
| 339 | .bind(&key)?, |
| 340 | self.db.prepare("SELECT summary FROM decisions WHERE item_id = ?1 ORDER BY created_at DESC LIMIT 5").bind(&key)?, |
| 341 | self.db.prepare("SELECT number, status, title FROM pulls WHERE item_id = ?1 ORDER BY created_at DESC LIMIT 5").bind(&key)?, |
| 342 | ]; |
| 343 | let results = self.db.batch(statements).await?; |
| 344 | let lines = |index: usize| -> Result<Vec<Line>> { |
| 345 | results.get(index).map_or_else(|| Ok(Vec::new()), |result| result.results::<Line>()) |
| 346 | }; |
| 347 | let sessions: Vec<String> = lines(0)? |
| 348 | .into_iter() |
| 349 | .map(|line| { |
| 350 | let result = line.result.as_deref().map(|text| format!(": {}", clip(text, 300))).unwrap_or_default(); |
| 351 | format!("- {} ({}){result}", line.title.unwrap_or_else(|| "Untitled session".into()), line.state.unwrap_or_default()) |
| 352 | }) |
| 353 | .collect(); |
| 354 | let decisions: Vec<String> = lines(1)?.into_iter().filter_map(|line| line.summary).map(|summary| format!("- {}", clip(&summary, 300))).collect(); |
| 355 | let pulls: Vec<String> = lines(2)? |
| 356 | .into_iter() |
| 357 | .map(|line| format!("- #{} {} ({})", line.number.unwrap_or_default(), line.title.unwrap_or_default(), line.status.unwrap_or_default())) |
| 358 | .collect(); |
| 359 | let none = || "- none".to_owned(); |
| 360 | let input = format!( |
| 361 | "Work item #{}: {}\n\nSessions, newest first:\n{}\n\nDecisions, newest first:\n{}\n\nPull requests:\n{}", |
| 362 | item.number, |
| 363 | item.title, |
| 364 | if sessions.is_empty() { none() } else { sessions.join("\n") }, |
| 365 | if decisions.is_empty() { none() } else { decisions.join("\n") }, |
| 366 | if pulls.is_empty() { none() } else { pulls.join("\n") }, |
| 367 | ); |
| 368 | let answer = self.brief(&item.workspace, STANDING_SYSTEM, input, 120, budget).await; |
| 369 | if answer.is_none() && !budget.without.contains(&item.workspace) { |
| 370 | break; |
| 371 | } |
| 372 | // With no provider, the time is still kept, so it is asked again |
| 373 | // only when something new happens. |
| 374 | let standing = answer.as_deref().and_then(read_standing); |
| 375 | self.db |
| 376 | .prepare("UPDATE work_items SET standing = COALESCE(?1, standing), standing_at = ?2 WHERE id = ?3") |
| 377 | .bind(&[ |
| 378 | standing.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), |
| 379 | rfc3339(now_ms()).into(), |
| 380 | item.id.as_str().into(), |
| 381 | ])? |
| 382 | .run() |
| 383 | .await?; |
| 384 | } |
| 385 | Ok(()) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | #[cfg(test)] |
| 390 | mod tests { |
| 391 | use super::*; |
| 392 | |
| 393 | #[test] |
| 394 | fn a_suggestion_is_read_from_the_four_lines() { |
| 395 | let at = "2026-10-20T00:00:00Z"; |
| 396 | let answer = "horizon: next\nappetite: small\nduplicate: #12\nwhy: It is a small fix to Home that others wait on."; |
| 397 | let read = read_suggestion(answer, &[12, 14], at).unwrap(); |
| 398 | assert_eq!(read.horizon, Horizon::Next); |
| 399 | assert_eq!(read.appetite, Some(Appetite::Small)); |
| 400 | assert_eq!(read.duplicate_of, Some(12)); |
| 401 | assert_eq!(read.why, "It is a small fix to Home that others wait on."); |
| 402 | // Markdown around it, and a number not in the plan. |
| 403 | let read = read_suggestion("**Horizon:** Later\n- Appetite: L\nDuplicate: 99\nWhy: Big.", &[12], at).unwrap(); |
| 404 | assert_eq!((read.horizon, read.appetite, read.duplicate_of), (Horizon::Later, Some(Appetite::Large), None)); |
| 405 | // A horizon it may not suggest, or none: no suggestion. |
| 406 | assert_eq!(read_suggestion("horizon: set_aside\nwhy: old", &[], at), None); |
| 407 | assert_eq!(read_suggestion("I think this should go next.", &[], at), None); |
| 408 | } |
| 409 | |
| 410 | #[test] |
| 411 | fn what_is_kept_is_one_bounded_line() { |
| 412 | assert_eq!(clip("Two words\nhere", 50), "Two words here"); |
| 413 | let long = "word ".repeat(100); |
| 414 | let clipped = clip(&long, 30); |
| 415 | assert!(clipped.chars().count() <= 31 && clipped.ends_with('…'), "{clipped}"); |
| 416 | assert_eq!(read_standing(" \n "), None); |
| 417 | assert_eq!(read_standing("- #205 merged; the plan is live. Next: place the backlog.").as_deref(), Some("#205 merged; the plan is live. Next: place the backlog.")); |
| 418 | } |
| 419 | } |