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