Skip to content

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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