| 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 | |
| 14 | use std::collections::VecDeque; |
| 15 | |
| 16 | use futures_util::StreamExt; |
| 17 | use g1t_contracts::identity::RevokeAskTokenArgs; |
| 18 | use 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 | }; |
| 22 | use serde_json::Value; |
| 23 | use worker::{ByteStream, Fetcher, Method}; |
| 24 | |
| 25 | use crate::ask::{self, Answered, Call, Request, Round}; |
| 26 | use crate::ask_tools::{self, Decision, Mcp, Sources, Toolbox}; |
| 27 | use crate::http; |
| 28 | |
| 29 | /// What the model is told when a call is not run because the answer used |
| 30 | /// what it may. |
| 31 | const 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. |
| 34 | const 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. |
| 37 | const 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. |
| 40 | const MAX_PROPOSALS: usize = 5; |
| 41 | |
| 42 | /// What an answer has spent of its bounds. |
| 43 | #[derive(Debug, Default)] |
| 44 | pub 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 | |
| 54 | impl 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. |
| 91 | pub 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 | |
| 119 | enum 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. |
| 129 | pub fn refused_by(provider: AiProvider, status: u16, body: String) -> String { |
| 130 | http::Answer { status, body }.problem(provider.label()) |
| 131 | } |
| 132 | |
| 133 | impl 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 | // Nothing there is an answer (a name checked free), not a read that failed. |
| 323 | Ok((text, true)) if ask_tools::is_not_found(&text) => (text, true, AskToolStatus::Done), |
| 324 | Ok((text, true)) => (text, true, AskToolStatus::Failed), |
| 325 | Err(problem) => (problem, true, AskToolStatus::Failed), |
| 326 | } |
| 327 | } |
| 328 | } |
| 329 | } |
| 330 | }); |
| 331 | let results = futures_util::future::join_all(runs).await; |
| 332 | let mut answered = Vec::with_capacity(calls.len()); |
| 333 | for ((call, id), (text, is_error, status)) in calls.iter().zip(&ids).zip(results) { |
| 334 | let summary = ask_tools::summary(&text, is_error); |
| 335 | let event = self.step(id, call, status, Some(summary)); |
| 336 | if status == AskToolStatus::Done { |
| 337 | let direct = match (&event, call.input["number"].is_u64()) { |
| 338 | (AskEvent::Tool { link: Some(link), label, .. }, true) => Some((link.clone(), label.clone())), |
| 339 | _ => None, |
| 340 | }; |
| 341 | self.sources.read(&self.site, &call.name, &call.input, &text, direct.as_ref().map(|(url, title)| (url.as_str(), title.as_str()))); |
| 342 | } |
| 343 | self.out.push_back(ask::line(&event)); |
| 344 | // What the model is given: trimmed to fit, within what the |
| 345 | // answer may read in all. |
| 346 | let room = self.budget.room(); |
| 347 | let content = if room < 400 && !is_error { |
| 348 | SPENT.to_owned() |
| 349 | } else { |
| 350 | ask_tools::trim_result(&text, room.max(400), ask_tools::Keep::of(&call.name, &call.input)) |
| 351 | }; |
| 352 | self.budget.chars += content.chars().count(); |
| 353 | answered.push(Answered { id: call.id.clone(), content, is_error }); |
| 354 | } |
| 355 | self.rounds.extend(ask::tool_turns(self.provider.api(), &text, &calls, &answered)); |
| 356 | self.budget.rounds += 1; |
| 357 | self.phase = Phase::Asking; |
| 358 | } |
| 359 | |
| 360 | /// Asks the provider for the next round. |
| 361 | async fn ask(&mut self) { |
| 362 | let built = match self.request() { |
| 363 | Ok(built) => built, |
| 364 | Err(problem) => { |
| 365 | self.out.push_back(ask::line(&AskEvent::Error { message: problem })); |
| 366 | return self.finish().await; |
| 367 | } |
| 368 | }; |
| 369 | let headers: Vec<(&str, &str)> = built.headers.iter().map(|(name, value)| (name.as_str(), value.as_str())).collect(); |
| 370 | let label = self.provider.label(); |
| 371 | let opened = http::open(Method::Post, &built.url, &headers, Some(built.body)).await; |
| 372 | let mut response = match opened { |
| 373 | Ok(response) => response, |
| 374 | Err(error) => { |
| 375 | self.out.push_back(ask::line(&AskEvent::Error { message: format!("{label} could not be reached: {error}") })); |
| 376 | return self.finish().await; |
| 377 | } |
| 378 | }; |
| 379 | let status = response.status_code(); |
| 380 | if !(200..300).contains(&status) { |
| 381 | let body = response.text().await.unwrap_or_default(); |
| 382 | self.out.push_back(ask::line(&AskEvent::Error { message: refused_by(self.provider, status, body) })); |
| 383 | return self.finish().await; |
| 384 | } |
| 385 | match response.stream() { |
| 386 | Ok(upstream) => { |
| 387 | self.phase = Phase::Streaming { upstream, round: Round::new(self.provider, !self.written.trim().is_empty()) }; |
| 388 | } |
| 389 | Err(error) => { |
| 390 | self.out.push_back(ask::line(&AskEvent::Error { message: format!("The answer from {label} broke off: {error}") })); |
| 391 | self.finish().await; |
| 392 | } |
| 393 | } |
| 394 | } |
| 395 | |
| 396 | /// The answer is over: its credential ends with it. |
| 397 | async fn finish(&mut self) { |
| 398 | self.phase = Phase::Finished; |
| 399 | if let Some(id) = self.credential.take() { |
| 400 | let _: worker::Result<bool> = g1t_kit::call(&self.identity, "revoke_ask_token", &RevokeAskTokenArgs { id }).await; |
| 401 | } |
| 402 | } |
| 403 | } |
| 404 | |
| 405 | /// The answer's text in some lines, for finding what it cites. |
| 406 | fn texts(lines: &str) -> String { |
| 407 | lines |
| 408 | .lines() |
| 409 | .filter_map(|line| match serde_json::from_str::<AskEvent>(line) { |
| 410 | Ok(AskEvent::Text { text }) => Some(text), |
| 411 | _ => None, |
| 412 | }) |
| 413 | .collect() |
| 414 | } |
| 415 | |
| 416 | #[cfg(test)] |
| 417 | mod tests { |
| 418 | use super::*; |
| 419 | |
| 420 | #[test] |
| 421 | fn the_answers_text_is_read_back_from_its_lines() { |
| 422 | let lines = format!( |
| 423 | "{}{}", |
| 424 | ask::line(&AskEvent::Text { text: "See ".into() }), |
| 425 | ask::line(&AskEvent::Text { text: "https://g1t.sh/a/b/pull/1".into() }) |
| 426 | ); |
| 427 | assert_eq!(texts(&lines), "See https://g1t.sh/a/b/pull/1"); |
| 428 | assert_eq!(texts(""), ""); |
| 429 | } |
| 430 | |
| 431 | #[test] |
| 432 | fn an_answer_stops_calling_tools_at_its_bounds() { |
| 433 | let start = 1_000_000; |
| 434 | let mut budget = Budget::new(start); |
| 435 | assert!(!budget.spent(start)); |
| 436 | // Calls run out. |
| 437 | for _ in 0..ASK_MAX_TOOL_CALLS { |
| 438 | assert!(budget.take_call(start)); |
| 439 | } |
| 440 | assert!(!budget.take_call(start)); |
| 441 | assert!(budget.spent(start)); |
| 442 | // So do rounds. |
| 443 | let mut budget = Budget::new(start); |
| 444 | budget.rounds = ASK_MAX_TOOL_ROUNDS - 1; |
| 445 | assert!(!budget.spent(start)); |
| 446 | budget.rounds += 1; |
| 447 | assert!(budget.spent(start)); |
| 448 | assert!(!budget.overrun(), "one more round, to answer"); |
| 449 | budget.rounds += 1; |
| 450 | assert!(budget.overrun()); |
| 451 | // And time. |
| 452 | let budget = Budget::new(start); |
| 453 | assert!(!budget.spent(start + ASK_MAX_TOOL_SECONDS * 1000)); |
| 454 | assert!(budget.spent(start + ASK_MAX_TOOL_SECONDS * 1000 + 1)); |
| 455 | // And what the model may be given: one result at most, then what is left. |
| 456 | let mut budget = Budget::new(start); |
| 457 | assert_eq!(budget.room(), ASK_MAX_TOOL_RESULT_CHARS); |
| 458 | budget.chars = ASK_MAX_TOOL_TOTAL_CHARS - 1_000; |
| 459 | assert_eq!(budget.room(), 1_000); |
| 460 | budget.chars = ASK_MAX_TOOL_TOTAL_CHARS; |
| 461 | assert_eq!(budget.room(), 0); |
| 462 | assert!(budget.spent(start)); |
| 463 | } |
| 464 | } |