Skip to content
Sign up

flagon-io/g1t

public
00

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
462 linesCodeBlameRaw
1//! One answer as it runs: the provider streams, asks for tools, Ask runs
2//! them as the person asking (ask_tools.rs) and sends the provider the
3//! results, round after round, until the model answers. Each line goes to
4//! the person as soon as it is known: the answer's text as it is written,
5//! each step as it starts and ends, then the sources and `done`.
6//!
7//! The answer is bounded: at most [`ASK_MAX_TOOL_ROUNDS`] rounds and
8//! [`ASK_MAX_TOOL_CALLS`] calls, [`ASK_MAX_TOOL_SECONDS`] of calling tools,
9//! [`ASK_MAX_TOOL_RESULT_CHARS`] of any one result and
10//! [`ASK_MAX_TOOL_TOTAL_CHARS`] of results in all. Past a bound the model is
11//! told so and answers with what it has. The person pressing Stop drops the
12//! stream, and with it whatever is in flight.
13
14use std::collections::VecDeque;
15
16use futures_util::StreamExt;
17use g1t_contracts::identity::RevokeAskTokenArgs;
18use g1t_contracts::ai::{
19 ASK_MAX_TOOL_CALLS, ASK_MAX_TOOL_RESULT_CHARS, ASK_MAX_TOOL_ROUNDS, ASK_MAX_TOOL_SECONDS, ASK_MAX_TOOL_TOTAL_CHARS,
20 AskEvent, AskMessage, AskToolStatus, AiConfig, AiProvider,
21};
22use serde_json::Value;
23use worker::{ByteStream, Fetcher, Method};
24
25use crate::ask::{self, Answered, Call, Request, Round};
26use crate::ask_tools::{self, Decision, Mcp, Sources, Toolbox};
27use crate::http;
28
29/// What the model is told when a call is not run because the answer used
30/// what it may.
31const SPENT: &str = "Not run: this answer has used the tool calls it may make. Answer now with what you have read, and say what you could not check.";
32
33/// What the model is told once a session is proposed.
34const PROPOSED: &str = "Proposed. The person now sees a card under your answer to start this session on their workspace's Anthropic account or copy a Claude Code command for it; nothing has started. Don't ask them to confirm in words: say in a sentence what the session will do.";
35
36/// What the model is told once it proposes building a new repository.
37const PROPOSED_NEW: &str = "Proposed. The person now sees a card under your answer; nothing has happened yet. Starting it creates the repository, turns on its deployments if you asked, opens the issue and starts the session, which pushes the first version to the default branch; that push deploys it, and the card then shows the address where it will be live on g1t.page. Don't ask them to confirm in words: say in a sentence or two what will be built, that it goes live on its own once the session pushes it, and that they can ask you how it is going.";
38
39/// The most sessions one answer proposes.
40const MAX_PROPOSALS: usize = 5;
41
42/// What an answer has spent of its bounds.
43#[derive(Debug, Default)]
44pub struct Budget {
45 /// When the answer started, in milliseconds.
46 pub started: u64,
47 /// Rounds of tool calls run.
48 pub rounds: usize,
49 pub calls: usize,
50 /// Characters of results given to the model.
51 pub chars: usize,
52}
53
54impl Budget {
55 pub fn new(started: u64) -> Self {
56 Budget { started, ..Budget::default() }
57 }
58
59 /// Whether the answer has used what it may of the tools, at `now`: the
60 /// model must answer with what it has.
61 pub fn spent(&self, now: u64) -> bool {
62 self.rounds >= ASK_MAX_TOOL_ROUNDS
63 || self.calls >= ASK_MAX_TOOL_CALLS
64 || self.chars >= ASK_MAX_TOOL_TOTAL_CHARS
65 || now.saturating_sub(self.started) > ASK_MAX_TOOL_SECONDS * 1000
66 }
67
68 /// Takes one call, if one is left.
69 pub fn take_call(&mut self, now: u64) -> bool {
70 if self.spent(now) {
71 return false;
72 }
73 self.calls += 1;
74 true
75 }
76
77 /// How much of a result the model may be given now.
78 pub fn room(&self) -> usize {
79 ASK_MAX_TOOL_TOTAL_CHARS.saturating_sub(self.chars).min(ASK_MAX_TOOL_RESULT_CHARS)
80 }
81
82 /// A model that keeps asking for tools after it was told to answer:
83 /// the answer ends with what it has written.
84 pub fn overrun(&self) -> bool {
85 self.rounds > ASK_MAX_TOOL_ROUNDS
86 }
87}
88
89/// Everything one answer needs, owned, so its stream outlives the request
90/// that started it.
91pub struct Answering {
92 pub provider: AiProvider,
93 pub config: AiConfig,
94 pub key: Option<String>,
95 pub model: String,
96 pub system: String,
97 pub messages: Vec<AskMessage>,
98 pub max_tokens: u32,
99 pub toolbox: Toolbox,
100 /// g1t's MCP server as the person asking; none when it could not be
101 /// reached, and the answer goes without tools.
102 pub mcp: Option<Mcp>,
103 pub site: String,
104 /// Identity, and the credential to revoke when the answer ends.
105 pub identity: Fetcher,
106 pub credential: Option<String>,
107 budget: Budget,
108 phase: Phase,
109 out: VecDeque<String>,
110 /// The rounds of tool calls so far, in the provider's shape.
111 rounds: Vec<Value>,
112 /// All the answer's text, for its sources.
113 written: String,
114 sources: Sources,
115 /// Sessions proposed so far, at most [`MAX_PROPOSALS`].
116 proposals: usize,
117}
118
119enum Phase {
120 Streaming { upstream: ByteStream, round: Round },
121 /// The model asked for tools: their `running` lines have gone; run them.
122 Calling { text: String, calls: Vec<Call>, ids: Vec<String> },
123 /// Ask the provider again, with the round's results.
124 Asking,
125 Finished,
126}
127
128/// Why a provider could not be asked, for the person.
129pub fn refused_by(provider: AiProvider, status: u16, body: String) -> String {
130 http::Answer { status, body }.problem(provider.label())
131}
132
133impl Answering {
134 #[allow(clippy::too_many_arguments)]
135 pub fn new(
136 provider: AiProvider,
137 config: AiConfig,
138 key: Option<String>,
139 model: String,
140 system: String,
141 messages: Vec<AskMessage>,
142 max_tokens: u32,
143 toolbox: Toolbox,
144 mcp: Option<Mcp>,
145 site: String,
146 identity: Fetcher,
147 credential: Option<String>,
148 started: u64,
149 ) -> Self {
150 Answering {
151 provider,
152 config,
153 key,
154 model,
155 system,
156 messages,
157 max_tokens,
158 toolbox,
159 mcp,
160 site,
161 identity,
162 credential,
163 budget: Budget::new(started),
164 phase: Phase::Asking,
165 out: VecDeque::new(),
166 rounds: Vec::new(),
167 written: String::new(),
168 sources: Sources::default(),
169 proposals: 0,
170 }
171 }
172
173 /// The provider's request for the next round.
174 pub fn request(&self) -> Result<ask::Built, String> {
175 ask::build(&Request {
176 provider: self.provider,
177 config: &self.config,
178 key: self.key.as_deref(),
179 model: &self.model,
180 system: &self.system,
181 messages: &self.messages,
182 rounds: &self.rounds,
183 tools: self.toolbox.specs(),
184 answer_now: self.budget.spent(g1t_kit::now_ms()),
185 max_tokens: self.max_tokens,
186 })
187 }
188
189 /// Starts streaming the first round's answer, already opened.
190 pub fn streaming(&mut self, upstream: ByteStream, first: String) {
191 self.out.push_back(first);
192 self.phase = Phase::Streaming { upstream, round: Round::new(self.provider, false) };
193 }
194
195 /// The next line to send, or `None` when the answer is over.
196 pub async fn next(&mut self) -> Option<String> {
197 loop {
198 if let Some(line) = self.out.pop_front() {
199 return Some(line);
200 }
201 match std::mem::replace(&mut self.phase, Phase::Finished) {
202 Phase::Finished => return None,
203 Phase::Asking => self.ask().await,
204 Phase::Streaming { mut upstream, mut round } => {
205 let label = self.provider.label();
206 let lines = match upstream.next().await {
207 Some(Ok(bytes)) => round.feed(&bytes),
208 Some(Err(error)) => {
209 round.broke(&format!("The answer from {label} broke off: {error}"));
210 String::new()
211 }
212 None => round.end(),
213 };
214 self.written.push_str(&texts(&lines));
215 if !lines.is_empty() {
216 self.out.push_back(lines);
217 }
218 if round.over() {
219 self.after(round).await;
220 } else {
221 self.phase = Phase::Streaming { upstream, round };
222 }
223 }
224 Phase::Calling { text, calls, ids } => self.call(text, calls, ids).await,
225 }
226 }
227 }
228
229 /// A round ended: run the tools it asked for, or finish.
230 async fn after(&mut self, round: Round) {
231 if let Some(problem) = round.error() {
232 self.out.push_back(ask::line(&AskEvent::Error { message: problem.to_owned() }));
233 return self.finish().await;
234 }
235 let calls = round.calls();
236 // No tools asked for, none to give, or a model that keeps asking
237 // after it was told to answer: what it wrote is the answer.
238 if calls.is_empty() || self.toolbox.is_empty() || self.budget.overrun() {
239 let sources = self.sources.finish(&self.site, &self.written);
240 if !sources.is_empty() {
241 self.out.push_back(ask::line(&AskEvent::Sources { sources }));
242 }
243 self.out.push_back(ask::line(&AskEvent::Done { stop_reason: round.stop_reason() }));
244 return self.finish().await;
245 }
246 // Each step shows as it starts.
247 let mut ids = Vec::with_capacity(calls.len());
248 for call in &calls {
249 let id = format!("step_{}_{}", self.budget.rounds, ids.len());
250 self.out.push_back(ask::line(&self.step(&id, call, AskToolStatus::Running, None)));
251 ids.push(id);
252 }
253 self.phase = Phase::Calling { text: round.text().to_owned(), calls, ids };
254 }
255
256 fn step(&self, id: &str, call: &Call, status: AskToolStatus, summary: Option<String>) -> AskEvent {
257 AskEvent::Tool {
258 id: id.to_owned(),
259 name: call.name.clone(),
260 action: call.input["action"].as_str().map(str::to_owned),
261 label: ask_tools::label(&call.name, &call.input),
262 status,
263 summary,
264 link: ask_tools::link(&self.site, &call.name, &call.input),
265 }
266 }
267
268 /// Runs a round's calls, all at once, and asks the provider again.
269 async fn call(&mut self, text: String, calls: Vec<Call>, ids: Vec<String>) {
270 enum Plan {
271 Run,
272 Skip(String, AskToolStatus),
273 }
274 let mut plans = Vec::with_capacity(calls.len());
275 for call in &calls {
276 let now = g1t_kit::now_ms();
277 let plan = if self.budget.spent(now) {
278 Plan::Skip(SPENT.to_owned(), AskToolStatus::Refused)
279 } else {
280 match self.toolbox.decide(&call.name, &call.input) {
281 Decision::Run if self.budget.take_call(now) => Plan::Run,
282 Decision::Run => Plan::Skip(SPENT.to_owned(), AskToolStatus::Refused),
283 // Dispatch: the answer shows a proposal the person
284 // starts themselves; the click is the confirmation.
285 Decision::Confirm if self.proposals >= MAX_PROPOSALS => {
286 Plan::Skip("Not run: this answer has proposed as many sessions as it may.".to_owned(), AskToolStatus::Refused)
287 }
288 Decision::Confirm => {
289 let id = format!("proposal_{}_{}", self.budget.rounds, self.proposals);
290 match ask_tools::proposal(&id, &call.input) {
291 Ok(proposal) => {
292 self.proposals += 1;
293 let told = match &proposal {
294 AskEvent::Proposal { new_repository: true, .. } => PROPOSED_NEW,
295 _ => PROPOSED,
296 };
297 self.out.push_back(ask::line(&proposal));
298 Plan::Skip(told.to_owned(), AskToolStatus::Done)
299 }
300 Err(why) => Plan::Skip(why, AskToolStatus::Failed),
301 }
302 }
303 Decision::Refuse(why) => Plan::Skip(why, AskToolStatus::Refused),
304 }
305 };
306 plans.push(plan);
307 }
308 let runs = calls.iter().zip(&plans).map(|(call, plan)| {
309 let mcp = self.mcp.clone();
310 let native = self.toolbox.is_native(&call.name);
311 async move {
312 match plan {
313 Plan::Skip(why, status) => (why.clone(), true, *status),
314 Plan::Run => {
315 let ran = match (native, mcp) {
316 (true, _) => ask_tools::run_native(&call.name, &call.input).await,
317 (false, Some(mcp)) => mcp.call(&call.name, &call.input).await,
318 (false, None) => Err("g1t's tools could not be reached.".to_owned()),
319 };
320 match ran {
321 Ok((text, false)) => (text, false, AskToolStatus::Done),
322 Ok((text, true)) => (text, true, AskToolStatus::Failed),
323 Err(problem) => (problem, true, AskToolStatus::Failed),
324 }
325 }
326 }
327 }
328 });
329 let results = futures_util::future::join_all(runs).await;
330 let mut answered = Vec::with_capacity(calls.len());
331 for ((call, id), (text, is_error, status)) in calls.iter().zip(&ids).zip(results) {
332 let summary = ask_tools::summary(&text, is_error);
333 let event = self.step(id, call, status, Some(summary));
334 if status == AskToolStatus::Done {
335 let direct = match (&event, call.input["number"].is_u64()) {
336 (AskEvent::Tool { link: Some(link), label, .. }, true) => Some((link.clone(), label.clone())),
337 _ => None,
338 };
339 self.sources.read(&self.site, &call.name, &call.input, &text, direct.as_ref().map(|(url, title)| (url.as_str(), title.as_str())));
340 }
341 self.out.push_back(ask::line(&event));
342 // What the model is given: trimmed to fit, within what the
343 // answer may read in all.
344 let room = self.budget.room();
345 let content = if room < 400 && !is_error {
346 SPENT.to_owned()
347 } else {
348 ask_tools::trim_result(&text, room.max(400), ask_tools::Keep::of(&call.name, &call.input))
349 };
350 self.budget.chars += content.chars().count();
351 answered.push(Answered { id: call.id.clone(), content, is_error });
352 }
353 self.rounds.extend(ask::tool_turns(self.provider.api(), &text, &calls, &answered));
354 self.budget.rounds += 1;
355 self.phase = Phase::Asking;
356 }
357
358 /// Asks the provider for the next round.
359 async fn ask(&mut self) {
360 let built = match self.request() {
361 Ok(built) => built,
362 Err(problem) => {
363 self.out.push_back(ask::line(&AskEvent::Error { message: problem }));
364 return self.finish().await;
365 }
366 };
367 let headers: Vec<(&str, &str)> = built.headers.iter().map(|(name, value)| (name.as_str(), value.as_str())).collect();
368 let label = self.provider.label();
369 let opened = http::open(Method::Post, &built.url, &headers, Some(built.body)).await;
370 let mut response = match opened {
371 Ok(response) => response,
372 Err(error) => {
373 self.out.push_back(ask::line(&AskEvent::Error { message: format!("{label} could not be reached: {error}") }));
374 return self.finish().await;
375 }
376 };
377 let status = response.status_code();
378 if !(200..300).contains(&status) {
379 let body = response.text().await.unwrap_or_default();
380 self.out.push_back(ask::line(&AskEvent::Error { message: refused_by(self.provider, status, body) }));
381 return self.finish().await;
382 }
383 match response.stream() {
384 Ok(upstream) => {
385 self.phase = Phase::Streaming { upstream, round: Round::new(self.provider, !self.written.trim().is_empty()) };
386 }
387 Err(error) => {
388 self.out.push_back(ask::line(&AskEvent::Error { message: format!("The answer from {label} broke off: {error}") }));
389 self.finish().await;
390 }
391 }
392 }
393
394 /// The answer is over: its credential ends with it.
395 async fn finish(&mut self) {
396 self.phase = Phase::Finished;
397 if let Some(id) = self.credential.take() {
398 let _: worker::Result<bool> = g1t_kit::call(&self.identity, "revoke_ask_token", &RevokeAskTokenArgs { id }).await;
399 }
400 }
401}
402
403/// The answer's text in some lines, for finding what it cites.
404fn texts(lines: &str) -> String {
405 lines
406 .lines()
407 .filter_map(|line| match serde_json::from_str::<AskEvent>(line) {
408 Ok(AskEvent::Text { text }) => Some(text),
409 _ => None,
410 })
411 .collect()
412}
413
414#[cfg(test)]
415mod tests {
416 use super::*;
417
418 #[test]
419 fn the_answers_text_is_read_back_from_its_lines() {
420 let lines = format!(
421 "{}{}",
422 ask::line(&AskEvent::Text { text: "See ".into() }),
423 ask::line(&AskEvent::Text { text: "https://g1t.sh/a/b/pull/1".into() })
424 );
425 assert_eq!(texts(&lines), "See https://g1t.sh/a/b/pull/1");
426 assert_eq!(texts(""), "");
427 }
428
429 #[test]
430 fn an_answer_stops_calling_tools_at_its_bounds() {
431 let start = 1_000_000;
432 let mut budget = Budget::new(start);
433 assert!(!budget.spent(start));
434 // Calls run out.
435 for _ in 0..ASK_MAX_TOOL_CALLS {
436 assert!(budget.take_call(start));
437 }
438 assert!(!budget.take_call(start));
439 assert!(budget.spent(start));
440 // So do rounds.
441 let mut budget = Budget::new(start);
442 budget.rounds = ASK_MAX_TOOL_ROUNDS - 1;
443 assert!(!budget.spent(start));
444 budget.rounds += 1;
445 assert!(budget.spent(start));
446 assert!(!budget.overrun(), "one more round, to answer");
447 budget.rounds += 1;
448 assert!(budget.overrun());
449 // And time.
450 let budget = Budget::new(start);
451 assert!(!budget.spent(start + ASK_MAX_TOOL_SECONDS * 1000));
452 assert!(budget.spent(start + ASK_MAX_TOOL_SECONDS * 1000 + 1));
453 // And what the model may be given: one result at most, then what is left.
454 let mut budget = Budget::new(start);
455 assert_eq!(budget.room(), ASK_MAX_TOOL_RESULT_CHARS);
456 budget.chars = ASK_MAX_TOOL_TOTAL_CHARS - 1_000;
457 assert_eq!(budget.room(), 1_000);
458 budget.chars = ASK_MAX_TOOL_TOTAL_CHARS;
459 assert_eq!(budget.room(), 0);
460 assert!(budget.spent(start));
461 }
462}