backend.rs · sigitcodereview/sigit

Code Commits Issues Pulls Cloud Agent CI/CD

feature/acp-cancel-request

claude/github-issue-40-0au5ua claude/linked-issue-acp-tui-u4vvpa claude/minor-release-cut-vkugow claude/sigit-acp-local-chat-cx6380 claude/sigit-cloud-agent-expansion-reox0a claude/tool-permission-system claude/zen-feynman-0u78dk development feature/acp-cancel-request feature/acp-session-list-cursor feature/acp-v1-conformance feature/agent-tools-multiedit-glob-todos-remember feature/cargo-workspace feature/hooks feature/load-local-model-explicitly feature/sigit-agent-profile feature/sigit-code-cloud feature/xcode-mcp-server fix/acp-replay-loaded-session-history fix/acp-replay-reasoning-and-attachments fix/dsml-doubled-bar-markers main release/v1.5.10 release/v1.5.6 release/v1.6.1

rs 3,775 lines 148 KB

Raw

1	//! Inference backend abstraction.
2	//!
3	//! The agent loop only needs to send a turn (optionally with tools) and return
4	//! tool results. This module defines that seam as the `InferenceBackend` trait
5	//! plus a few neutral types, with two implementations:
6	//!
7	//! - `LocalBackend` runs on-device through the `onde` crate (`ChatEngine`).
8	//! - `OpenAiBackend` talks to any OpenAI-compatible HTTP endpoint, configured by
9	//! `base_url`, `api_key`, and `model`.
10	//!
11	//! The trait exposes neither `onde` nor OpenAI types, so the loop does not depend
12	//! on a specific backend.
13	//!
14	//! The seam is consumed by both surfaces: the interactive client (`#[cfg(unix)]`,
15	//! see `run_interactive` in `main.rs` and `mod tui` in `chat.rs`) and the ACP
16	//! server's prompt loop. Some items are still reached only through the
17	//! Unix-only interactive paths, so the dead-code lint stays suppressed on
18	//! non-Unix targets only — Unix builds keep full coverage.
19	#![cfg_attr(not(unix), allow(dead_code))]
20
21	use std::sync::Arc;
22	use std::sync::atomic::{AtomicBool, Ordering};
23
24	use async_trait::async_trait;
25	use onde::inference::{ChatEngine, ChatMessage, ChatRole, EngineStatus, ToolDefinition};
26	use serde::Deserialize;
27	use tokio::sync::Mutex;
28
29	// ── Neutral types ───────────────────────────────────────────────────────────────
30
31	/// A tool the model may call, in a provider-neutral form. `parameters_schema` is
32	/// a JSON Schema encoded as a string (matching how siGit Code already declares
33	/// tools).
34	#[derive(Debug, Clone)]
35	pub struct ToolSpec {
36	pub name: String,
37	pub description: String,
38	pub parameters_schema: String,
39	}
40
41	/// A tool call requested by the model.
42	#[derive(Debug, Clone)]
43	pub struct ToolCall {
44	pub id: String,
45	pub name: String,
46	/// Arguments as a JSON-encoded string.
47	pub arguments: String,
48	}
49
50	/// The output of executing one tool call, fed back to the model.
51	#[derive(Debug, Clone)]
52	pub struct ToolResult {
53	pub tool_call_id: String,
54	pub content: String,
55	}
56
57	/// The result of one assistant turn: free text and/or tool calls.
58	#[derive(Debug, Clone, Default)]
59	pub struct TurnResult {
60	pub text: String,
61	pub tool_calls: Vec<ToolCall>,
62	/// Why the model stopped. Only the last round of a turn decides how the
63	/// turn is reported; a round that goes on to run tools is not the end.
64	pub finish: FinishReason,
65	/// The context in use once this reply is in, as the endpoint counted it:
66	/// prompt plus completion tokens. `None` when nothing was reported, which
67	/// is always the case on-device.
68	pub context_tokens: Option<u64>,
69	}
70
71	/// Why the model stopped generating, reduced to what the agent loop acts on.
72	/// Endpoints report it as `finish_reason`; without it a reply cut off at the
73	/// token limit is indistinguishable from one the model finished.
74	#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
75	pub enum FinishReason {
76	/// The model finished on its own: `stop`, `tool_calls`, or no reason given.
77	#[default]
78	Complete,
79	/// The output ran into the token limit, so the reply is cut short.
80	Length,
81	/// The endpoint withheld the reply instead of answering.
82	ContentFilter,
83	/// The harness cut the reply off because it had degenerated into
84	/// repetition (see `repetition`).
85	Repetition,
86	}
87
88	impl FinishReason {
89	/// Map a wire `finish_reason`. OpenAI's spellings are `length` and
90	/// `content_filter`; gateways in front of other providers sometimes pass
91	/// the upstream's own word through, so those are accepted too.
92	fn from_wire(reason: Option<&str>) -> Self {
93	match reason {
94	Some("length" \| "max_tokens") => Self::Length,
95	Some("content_filter" \| "refusal") => Self::ContentFilter,
96	_ => Self::Complete,
97	}
98	}
99	}
100
101	/// Backend errors are plain strings. Callers map them to ACP errors.
102	pub type BackendError = String;
103
104	/// Rough context budget for a conversation, in estimated tokens (see
105	/// [`estimate_tokens`]). When a snapshot exceeds this, the agent loops compact
106	/// history before the next tool round.
107	pub const DEFAULT_CONTEXT_TOKEN_BUDGET: usize = 24_000;
108
109	/// How many trailing messages survive a compaction verbatim (the rest are
110	/// folded into the summary).
111	pub const COMPACT_KEEP_LAST: usize = 6;
112
113	/// The summarization request sent to the model when compacting history.
114	const SUMMARIZE_PROMPT: &str = "Summarize this coding session so far: decisions made, \
115	files touched, current state, open items. Be concise and factual.";
116
117	/// Largest transcript (in estimated tokens, see [`estimate_tokens`]) sent to
118	/// the model for summarization in one go.
119	///
120	/// Compaction runs because the conversation has outgrown the context budget,
121	/// so shipping the whole transcript verbatim asks the endpoint to process
122	/// exactly the payload that is already too big — on siGit Code Cloud that
123	/// request sits in inference until the gateway gives up (504), which is the
124	/// failure behind issue #125. Capping the transcript keeps the summarization
125	/// round inside the window the model actually has; the oldest and newest
126	/// exchanges carry the intent and the current state, while the truncated
127	/// middle is the part a summary can afford to lose.
128	const SUMMARY_TRANSCRIPT_TOKEN_CAP: usize = 12_000;
129
130	/// Trim `transcript` to about `cap` estimated tokens by cutting whole lines
131	/// from the middle, keeping the opening and closing exchanges.
132	///
133	/// A middle cut preserves the two ends that matter most to a summary: how the
134	/// session started (the original request) and where it stands now. Cutting by
135	/// whole lines keeps individual messages intact. A transcript already under
136	/// the cap is returned unchanged.
137	///
138	/// Line boundaries are best-effort. A head with no newline in its half is
139	/// dropped and the tail keeps its own half. A tail with no newline in its half
140	/// (a single huge message, or a giant final tool result) can't be cut on a
141	/// line, so the newest `cap` worth of text is kept instead and the head goes:
142	/// the newest exchanges are the ones the summary can least afford to lose.
143	fn truncate_transcript_middle(transcript: &str, cap: usize) -> String {
144	let budget = cap * 4; // estimate_tokens is chars / 4; invert it.
145	if transcript.len() <= budget {
146	return transcript.to_string();
147	}
148
149	let total = transcript.len();
150	// Each side gets half the budget, so the two together stay inside the cap.
151	// Walk each half to a line boundary. A head with no boundary in reach
152	// contributes nothing; a tail with none takes the hard cut below.
153	let half = budget / 2;
154	let head_end = {
155	let mut cut = half.min(total);
156	while cut < total && !transcript.is_char_boundary(cut) {
157	cut += 1;
158	}
159	cut = cut.min(total);
160	if transcript[..cut].contains('\n') {
161	transcript[..cut].rfind('\n').unwrap() + 1
162	} else {
163	0
164	}
165	};
166	let tail_start = {
167	let mut cut = total.saturating_sub(half);
168	while cut < total && !transcript.is_char_boundary(cut) {
169	cut += 1;
170	}
171	if transcript[cut..].contains('\n') {
172	transcript[cut..]
173	.find('\n')
174	.map(\|offset\| cut + offset + 1)
175	.unwrap()
176	} else {
177	total
178	}
179	};
180
181	if tail_start == total {
182	// The closing exchange has no line boundary in reach (a transcript of
183	// one huge message, or a giant newline-less tool result at the end):
184	// keep the newest budget. A head-only cut would invert the contract
185	// while the placeholder claims older messages were dropped.
186	let mut start = total.saturating_sub(budget);
187	while start > 0 && !transcript.is_char_boundary(start) {
188	start -= 1;
189	}
190	return format!(
191	"{}\n[…older messages omitted…]",
192	transcript[start..].trim_start()
193	);
194	}
195
196	if head_end >= tail_start {
197	// No line boundary in reach anywhere: keep the newest budget.
198	let mut start = total.saturating_sub(budget);
199	while start > 0 && !transcript.is_char_boundary(start) {
200	start -= 1;
201	}
202	return format!(
203	"{}\n[…older messages omitted…]",
204	transcript[start..].trim_start()
205	);
206	}
207
208	let omitted_chars = transcript[head_end..tail_start].chars().count();
209	format!(
210	"{}\n[…{} characters of older messages omitted…]\n{}",
211	transcript[..head_end].trim_end(),
212	omitted_chars,
213	transcript[tail_start..].trim_start(),
214	)
215	}
216
217	/// Render a history snapshot as a plain-text transcript, with tool calls and
218	/// tool results spelled out as prose rather than left in their wire shapes.
219	///
220	/// Compaction summarizes a conversation that is, by definition, thick with
221	/// `tool_calls` and `role: "tool"` messages — but the summarization round asks
222	/// for a plain answer and so sends no `tools` array. Forwarding the raw shapes
223	/// in that request produces tool blocks with no schema to validate against,
224	/// which strict endpoints reject outright: Anthropic answers 400, so every
225	/// compaction of a session that had ever run a tool failed, permanently, no
226	/// matter how small the history was. Flattening to text keeps everything the
227	/// summary actually needs and drops the shapes that only make sense alongside
228	/// a tool schema. It also sidesteps orphaned `tool_call_id`s and role-
229	/// alternation rules, neither of which a transcript can violate.
230	fn transcript_for_summary(history: &[serde_json::Value]) -> String {
231	let mut lines: Vec<String> = Vec::new();
232	for message in history {
233	let role = message["role"].as_str().unwrap_or("user");
234	// The system prompt is carried over verbatim, so it needn't be summarized.
235	if role == "system" {
236	continue;
237	}
238
239	let mut parts: Vec<String> = Vec::new();
240	let text = message_text(message);
241	if !text.trim().is_empty() {
242	parts.push(text);
243	}
244	// A summary cannot carry a picture, but it should say one was there.
245	match message_image_count(message) {
246	0 => {}
247	1 => parts.push("[attached an image]".to_string()),
248	count => parts.push(format!("[attached {count} images]")),
249	}
250	for call in message["tool_calls"].as_array().into_iter().flatten() {
251	parts.push(format!(
252	"called {}({})",
253	call["function"]["name"].as_str().unwrap_or("tool"),
254	call["function"]["arguments"].as_str().unwrap_or_default(),
255	));
256	}
257	if parts.is_empty() {
258	continue;
259	}
260
261	let label = if role == "tool" { "tool result" } else { role };
262	lines.push(format!("{label}: {}", parts.join("\n")));
263	}
264	lines.join("\n\n")
265	}
266
267	/// Crude token estimate for a history snapshot: serialized characters / 4.
268	/// Deliberately model-agnostic — it only needs to be in the right ballpark to
269	/// decide when compaction is worth an extra inference round.
270	pub fn estimate_tokens(history: &[serde_json::Value]) -> usize {
271	let mut chars = 0;
272	let mut images = 0;
273	for message in history {
274	let image_count = message_image_count(message);
275	if image_count == 0 {
276	chars += message.to_string().chars().count();
277	} else {
278	// An image travels as base64, megabytes of it, but a model bills
279	// it by resolution. Counting the payload as text would put every
280	// session with one screenshot permanently over budget.
281	chars += message_text(message).chars().count();
282	images += image_count;
283	}
284	}
285	chars / 4 + images * IMAGE_TOKEN_ESTIMATE
286	}
287
288	/// What one attached image is assumed to cost. Providers land between a few
289	/// hundred and a couple of thousand tokens depending on resolution; like the
290	/// rest of [`estimate_tokens`] this only has to be in the right ballpark.
291	const IMAGE_TOKEN_ESTIMATE: usize = 1_000;
292
293	/// The text of a history message. `content` is a plain string for almost every
294	/// message; a user message that carries an image uses OpenAI's content-part
295	/// array instead, and its text parts are joined here.
296	pub fn message_text(message: &serde_json::Value) -> String {
297	match &message["content"] {
298	serde_json::Value::String(text) => text.clone(),
299	serde_json::Value::Array(parts) => parts
300	.iter()
301	.filter_map(\|part\| part["text"].as_str())
302	.collect::<Vec<_>>()
303	.join("\n"),
304	_ => String::new(),
305	}
306	}
307
308	/// How many content parts of `part_type` a history message carries.
309	fn message_part_count(message: &serde_json::Value, part_type: &str) -> usize {
310	message["content"]
311	.as_array()
312	.map(\|parts\| {
313	parts
314	.iter()
315	.filter(\|part\| part["type"] == part_type)
316	.count()
317	})
318	.unwrap_or(0)
319	}
320
321	/// How many images a history message carries as `image_url` content parts.
322	pub fn message_image_count(message: &serde_json::Value) -> usize {
323	message_part_count(message, "image_url")
324	}
325
326	/// How many audio clips a history message carries as `input_audio` content
327	/// parts.
328	pub fn message_audio_count(message: &serde_json::Value) -> usize {
329	message_part_count(message, "input_audio")
330	}
331
332	/// One streamed fragment, tagged by whether it is visible prose or the model's
333	/// reasoning.
334	///
335	/// Reasoning reaches a backend two ways: inline `<think>…</think>` tags inside
336	/// the ordinary content stream (local models, and some OpenAI-compatible
337	/// endpoints), or a separate `reasoning_content` delta field (DeepSeek-style
338	/// endpoints). Inline tags can't be classified at the source — a tag may span
339	/// chunk boundaries — so they travel as [`TokenChunk::Visible`] and are split
340	/// out downstream by `StreamedReply`. The `reasoning_content` field is
341	/// unambiguous at the source, so the backend tags it [`TokenChunk::Reasoning`]
342	/// directly.
343	#[derive(Debug, Clone, PartialEq, Eq)]
344	pub enum TokenChunk {
345	/// Content-stream text. May still contain inline `<think>` tags that a
346	/// downstream consumer splits into reasoning and prose.
347	Visible(String),
348	/// Text from a dedicated reasoning channel, already known to be reasoning.
349	Reasoning(String),
350	}
351
352	/// A sink for streaming assistant fragments to the UI as they are produced.
353	///
354	/// When a caller passes `Some(sink)`, a streaming-capable backend forwards each
355	/// fragment through it as the model emits it, tagged [`TokenChunk::Visible`] or
356	/// [`TokenChunk::Reasoning`]; the returned [`TurnResult`] still carries the
357	/// fully assembled (visible) text and any tool calls. When the sink is `None`,
358	/// the backend runs in non-streaming mode. Unbounded so the inference task
359	/// never blocks on a slow consumer.
360	pub type TokenSink = tokio::sync::mpsc::UnboundedSender<TokenChunk>;
361
362	/// An image attached to a user message: base64 data and its media type, as an
363	/// ACP client sends it.
364	#[derive(Debug, Clone, PartialEq, Eq)]
365	pub struct ImageInput {
366	pub mime_type: String,
367	pub data: String,
368	}
369
370	/// An audio clip attached to a user message: base64 data and its media type,
371	/// as an ACP client sends it.
372	#[derive(Debug, Clone, PartialEq, Eq)]
373	pub struct AudioInput {
374	pub mime_type: String,
375	pub data: String,
376	}
377
378	impl AudioInput {
379	/// The `format` OpenAI's `input_audio` part expects (`wav`, `mp3`, ...),
380	/// from the MIME type an ACP client sends (`audio/wav`, `audio/mpeg`, ...).
381	fn format(&self) -> String {
382	let subtype = self
383	.mime_type
384	.split(';')
385	.next()
386	.unwrap_or_default()
387	.trim()
388	.to_lowercase();
389	let subtype = subtype.strip_prefix("audio/").unwrap_or(&subtype);
390	match subtype.strip_prefix("x-").unwrap_or(subtype) {
391	"mpeg" \| "mp3" => "mp3".to_string(),
392	"wave" \| "wav" \| "vnd.wave" => "wav".to_string(),
393	other => other.to_string(),
394	}
395	}
396	}
397
398	/// What a model that cannot read images is shown in place of one.
399	const IMAGE_OMITTED_NOTE: &str = "[image omitted: this model cannot read images]";
400
401	/// What a model that cannot take audio is shown in place of a clip.
402	const AUDIO_OMITTED_NOTE: &str = "[audio omitted: this model cannot take audio]";
403
404	/// A user message in history form. Plain text keeps the string `content` every
405	/// other message uses; with attachments it becomes OpenAI's content-part
406	/// array, text first, each image as a base64 `data:` URL and each audio clip
407	/// as an `input_audio` part.
408	fn user_message(text: &str, images: &[ImageInput], audio: &[AudioInput]) -> serde_json::Value {
409	if images.is_empty() && audio.is_empty() {
410	return serde_json::json!({ "role": "user", "content": text });
411	}
412	let mut parts = Vec::with_capacity(images.len() + audio.len() + 1);
413	if !text.is_empty() {
414	parts.push(serde_json::json!({ "type": "text", "text": text }));
415	}
416	for image in images {
417	parts.push(serde_json::json!({
418	"type": "image_url",
419	"image_url": {
420	"url": format!("data:{};base64,{}", image.mime_type, image.data),
421	},
422	}));
423	}
424	for clip in audio {
425	parts.push(serde_json::json!({
426	"type": "input_audio",
427	"input_audio": { "data": clip.data, "format": clip.format() },
428	}));
429	}
430	serde_json::json!({ "role": "user", "content": parts })
431	}
432
433	/// `history` with every content part of `part_type` replaced by `note`, for a
434	/// model that cannot take that kind of attachment. A message left with only
435	/// text is flattened back to a plain string.
436	///
437	/// History itself keeps the attachments. A thread can move to a model that
438	/// reads them (or back to one), so what was attached is not thrown away just
439	/// because the model active right now cannot use it.
440	fn without_parts(
441	history: &[serde_json::Value],
442	part_type: &str,
443	note: &str,
444	) -> Vec<serde_json::Value> {
445	history
446	.iter()
447	.map(\|message\| {
448	let Some(parts) = message["content"].as_array() else {
449	return message.clone();
450	};
451	if !parts.iter().any(\|part\| part["type"] == part_type) {
452	return message.clone();
453	}
454	let mut kept: Vec<serde_json::Value> = Vec::with_capacity(parts.len());
455	let mut notes = Vec::new();
456	for part in parts {
457	if part["type"] == part_type {
458	notes.push(serde_json::json!({ "type": "text", "text": note }));
459	} else {
460	kept.push(part.clone());
461	}
462	}
463	kept.extend(notes);
464	let mut stripped = message.clone();
465	stripped["content"] = if kept.iter().all(\|part\| part["type"] == "text") {
466	let text = kept
467	.iter()
468	.filter_map(\|part\| part["text"].as_str())
469	.filter(\|text\| !text.is_empty())
470	.collect::<Vec<_>>()
471	.join("\n");
472	serde_json::Value::String(text)
473	} else {
474	serde_json::Value::Array(kept)
475	};
476	stripped
477	})
478	.collect()
479	}
480
481	/// `history` as a model that cannot read images has to receive it, with a
482	/// note where each image was.
483	fn without_images(history: &[serde_json::Value]) -> Vec<serde_json::Value> {
484	without_parts(history, "image_url", IMAGE_OMITTED_NOTE)
485	}
486
487	/// `history` as a model that cannot take audio has to receive it, with a note
488	/// where each clip was.
489	fn without_audio(history: &[serde_json::Value]) -> Vec<serde_json::Value> {
490	without_parts(history, "input_audio", AUDIO_OMITTED_NOTE)
491	}
492
493	// ── The trait ───────────────────────────────────────────────────────────────────
494
495	/// A swappable inference backend driving siGit Code's agent loop.
496	#[async_trait]
497	pub trait InferenceBackend: Send + Sync {
498	/// Start an assistant turn from a new user message, offering `tools`.
499	///
500	/// If `sink` is `Some`, text is streamed through it as it is generated. A
501	/// backend may decline to stream a given round (for example, on-device
502	/// inference cannot stream while it is still deciding whether to call a
503	/// tool); in that case the text is delivered only via the returned result.
504	async fn send_message_with_tools(
505	&self,
506	text: &str,
507	tools: &[ToolSpec],
508	sink: Option<&TokenSink>,
509	) -> Result<TurnResult, BackendError>;
510
511	/// Continue the turn by returning tool results. `allow_tool_calls` controls
512	/// whether `tools` are offered again; the complete catalog remains available
513	/// so a disabled round can recognize and suppress tool-shaped model output.
514	/// `sink` streams assistant text when set.
515	async fn send_tool_results(
516	&self,
517	results: Vec<ToolResult>,
518	tools: &[ToolSpec],
519	allow_tool_calls: bool,
520	sink: Option<&TokenSink>,
521	) -> Result<TurnResult, BackendError>;
522
523	/// Record tool results in the conversation history without asking the
524	/// model to continue the turn. Used when a turn is abandoned mid-round
525	/// (the user cancelled at the permission gate): by then the assistant
526	/// message carrying the tool calls is already in history, and leaving them
527	/// unanswered makes strict OpenAI-compatible endpoints reject every later
528	/// request in the session.
529	async fn record_cancelled_tool_results(&self, results: Vec<ToolResult>);
530
531	/// Start a turn from a user message that carries images or audio.
532	///
533	/// The default drops the attachments and sends the text, which is right
534	/// for a backend that cannot read them. Callers check
535	/// [`Self::accepts_images`] and [`Self::accepts_audio`] first so the user
536	/// can be told, instead of the attachment vanishing.
537	async fn send_message_with_attachments(
538	&self,
539	text: &str,
540	images: &[ImageInput],
541	audio: &[AudioInput],
542	tools: &[ToolSpec],
543	sink: Option<&TokenSink>,
544	) -> Result<TurnResult, BackendError> {
545	let _ = (images, audio);
546	self.send_message_with_tools(text, tools, sink).await
547	}
548
549	/// Whether the model behind this backend reads images. On-device models do
550	/// not; a remote one answers from its model id.
551	fn accepts_images(&self) -> bool {
552	false
553	}
554
555	/// Whether the model behind this backend takes audio. On-device models do
556	/// not; a remote one answers from its model id.
557	fn accepts_audio(&self) -> bool {
558	false
559	}
560
561	/// Whether inference runs over the network (a configured provider) rather
562	/// than on-device. Drives UI labelling so the displayed model can't claim a
563	/// local model while requests actually go to the cloud.
564	fn is_remote(&self) -> bool;
565
566	/// A backend for the same endpoint and model with an empty conversation,
567	/// or `None` when the conversation cannot be separated from the backend
568	/// (on-device, the engine holds it). An editor keeps several threads open
569	/// in one process; this is what lets each of them own its history instead
570	/// of taking turns on a shared one.
571	fn fresh(&self) -> Option<Arc<dyn InferenceBackend>> {
572	None
573	}
574
575	/// A serializable snapshot of the conversation history, one JSON object per
576	/// message (`{"role": ..., "content": ...}` at minimum). The snapshot is
577	/// what the session store persists; it includes any seeded system message
578	/// so [`InferenceBackend::restore_history`] can replace state wholesale.
579	async fn history_snapshot(&self) -> Vec<serde_json::Value>;
580
581	/// Replace the conversation history with a previously saved snapshot.
582	/// Backends that cannot represent every entry (e.g. on-device history has
583	/// no tool-call structure) flatten what they can and drop the rest.
584	async fn restore_history(&self, history: Vec<serde_json::Value>);
585
586	/// Shrink the conversation history: summarize everything so far with one
587	/// extra (non-streaming) inference round, then rebuild history as
588	/// `[system message, summary, last keep_last non-system messages]`. When
589	/// the summarization round itself fails, a backend should fall back to a
590	/// deterministic shrink (dropping the oldest messages) rather than leave
591	/// an over-budget session stuck; an `Err` here means even that fallback
592	/// could not fit the history in the budget.
593	async fn compact_history(&self, keep_last: usize) -> Result<(), BackendError>;
594	}
595
596	// ── Local backend (onde ChatEngine) ──────────────────────────────────────────────
597
598	/// On-device inference. A thin adapter over `onde::ChatEngine`.
599	pub struct LocalBackend {
600	engine: Arc<ChatEngine>,
601	/// History restored before an on-device model is loaded. Onde ignores
602	/// `push_history` while unloaded, so keep the snapshot here until the
603	/// engine can accept it instead of silently dropping a loaded session.
604	pending_history: Mutex<Option<Vec<serde_json::Value>>>,
605	}
606
607	impl LocalBackend {
608	pub fn new(engine: Arc<ChatEngine>) -> Self {
609	Self {
610	engine,
611	pending_history: Mutex::new(None),
612	}
613	}
614
615	async fn apply_pending_history(&self) -> Result<(), BackendError> {
616	if self.pending_history.lock().await.is_none() {
617	return Ok(());
618	}
619	let status = self.engine.info().await.status;
620	if status != EngineStatus::Ready {
621	return Err(format!(
622	"on-device model is not ready to restore session history (status: {status})"
623	));
624	}
625	let Some(history) = self.pending_history.lock().await.take() else {
626	return Ok(());
627	};
628
629	self.engine.clear_history().await;
630	for entry in history {
631	let role = entry["role"].as_str().unwrap_or("");
632	// The on-device engine keeps text only, so an image attached
633	// while a cloud model was active stays behind here.
634	let content = message_text(&entry);
635	// Tool-call-only assistant entries and empty tool results carry no
636	// text a plain chat history can replay; drop them.
637	if content.is_empty() && role != "user" && role != "system" {
638	continue;
639	}
640	let message = match role {
641	"system" => ChatMessage::system(content),
642	"user" => ChatMessage::user(content),
643	"assistant" => ChatMessage::assistant(content),
644	// Tool results flatten to plain text (MVP; acceptable loss).
645	"tool" => ChatMessage::user(format!("[tool result]\n{content}")),
646	_ => continue,
647	};
648	self.engine.push_history(message).await;
649	}
650	Ok(())
651	}
652	}
653
654	fn to_onde_tools(tools: &[ToolSpec]) -> Vec<ToolDefinition> {
655	tools
656	.iter()
657	.map(\|tool\| ToolDefinition {
658	name: tool.name.clone(),
659	description: tool.description.clone(),
660	parameters_schema: tool.parameters_schema.clone(),
661	})
662	.collect()
663	}
664
665	#[async_trait]
666	impl InferenceBackend for LocalBackend {
667	async fn send_message_with_tools(
668	&self,
669	text: &str,
670	tools: &[ToolSpec],
671	sink: Option<&TokenSink>,
672	) -> Result<TurnResult, BackendError> {
673	self.apply_pending_history().await?;
674	// onde's tool-aware path is non-streaming: it has to buffer the whole
675	// reply to detect tool calls. We can only stream when no tools are on
676	// offer (a plain answer), which is exactly the tools-disabled case.
677	if let Some(sink) = sink
678	&& tools.is_empty()
679	{
680	let rx = self
681	.engine
682	.stream_message(text)
683	.await
684	.map_err(\|error\| error.to_string())?;
685	return drain_onde_stream(rx, sink).await;
686	}
687
688	let onde_tools = to_onde_tools(tools);
689	let result = self
690	.engine
691	.send_message_with_tools(text, &onde_tools)
692	.await
693	.map_err(\|error\| error.to_string())?;
694	Ok(onde_result_to_turn(result))
695	}
696
697	async fn send_tool_results(
698	&self,
699	results: Vec<ToolResult>,
700	tools: &[ToolSpec],
701	allow_tool_calls: bool,
702	sink: Option<&TokenSink>,
703	) -> Result<TurnResult, BackendError> {
704	self.apply_pending_history().await?;
705	let onde_results: Vec<onde::inference::ToolResult> = results
706	.into_iter()
707	.map(\|result\| onde::inference::ToolResult {
708	tool_call_id: result.tool_call_id,
709	content: result.content,
710	})
711	.collect();
712
713	// A forced-text round is the only round onde can stream, since no
714	// further tool calls are parsed.
715	if let Some(sink) = sink
716	&& !allow_tool_calls
717	{
718	let rx = self
719	.engine
720	.stream_tool_results(onde_results, None)
721	.await
722	.map_err(\|error\| error.to_string())?;
723	return drain_onde_stream(rx, sink).await;
724	}
725
726	let onde_tools = allow_tool_calls.then(\|\| to_onde_tools(tools));
727	let result = self
728	.engine
729	.send_tool_results(onde_results, onde_tools.as_deref())
730	.await
731	.map_err(\|error\| error.to_string())?;
732	Ok(onde_result_to_turn(result))
733	}
734
735	async fn record_cancelled_tool_results(&self, _results: Vec<ToolResult>) {
736	// onde's public API cannot append tool-result history entries without
737	// running another inference round, so the dangling tool call stays in
738	// its history. The chat template replays it as-is, which local models
739	// tolerate — worst case the model re-issues the call next turn.
740	}
741
742	fn is_remote(&self) -> bool {
743	false
744	}
745
746	async fn history_snapshot(&self) -> Vec<serde_json::Value> {
747	if let Some(history) = self.pending_history.lock().await.as_ref() {
748	return history.clone();
749	}
750	// onde's `history()` already flattens tool entries: assistant tool
751	// calls become plain assistant text and tool results are omitted, so
752	// the snapshot is lossy for tool-heavy turns (acceptable in this MVP).
753	self.engine
754	.history()
755	.await
756	.iter()
757	.map(\|message\| {
758	serde_json::json!({
759	"role": message.role.to_string(),
760	"content": message.content,
761	})
762	})
763	.collect()
764	}
765
766	async fn restore_history(&self, history: Vec<serde_json::Value>) {
767	*self.pending_history.lock().await = Some(history);
768	if self.engine.info().await.status == EngineStatus::Ready {
769	// Ready was just observed, so failure here can only mean a status
770	// transition; keep the pending copy for the next inference call.
771	let _ = self.apply_pending_history().await;
772	}
773	}
774
775	async fn compact_history(&self, keep_last: usize) -> Result<(), BackendError> {
776	self.apply_pending_history().await?;
777	let snapshot = self.engine.history().await;
778
779	// Leading system messages carry session context; keep them all.
780	let system: Vec<ChatMessage> = snapshot
781	.iter()
782	.take_while(\|message\| message.role == ChatRole::System)
783	.cloned()
784	.collect();
785	let non_system: Vec<ChatMessage> = snapshot
786	.into_iter()
787	.filter(\|message\| message.role != ChatRole::System)
788	.collect();
789	let tail_start = non_system.len().saturating_sub(keep_last);
790	let tail: Vec<ChatMessage> = non_system[tail_start..].to_vec();
791
792	// One plain (tool-free) inference round produces the summary. Note the
793	// cap only bounds this request's own payload: the engine still replays
794	// its full (over-budget) history underneath it, so an already
795	// over-budget local session usually fails here and lands in the
796	// fallback below — the cap is about not making the doomed round worse.
797	let transcript = truncate_transcript_middle(
798	&chat_messages_transcript(&non_system),
799	SUMMARY_TRANSCRIPT_TOKEN_CAP,
800	);
801	let result = self
802	.engine
803	.send_message(format!("{transcript}\n\n{SUMMARIZE_PROMPT}"))
804	.await
805	.map_err(\|error\| error.to_string());
806
807	let rebuild: Vec<ChatMessage> = match result {
808	Ok(result) => {
809	// Local models may reason in <think> blocks; keep only the visible part.
810	let (_think, summary) = crate::chat::strip_think_blocks(&result.text);
811	let mut rebuilt = system;
812	rebuilt.push(ChatMessage::user(format!(
813	"[Conversation summary]\n{summary}"
814	)));
815	rebuilt.extend(tail);
816	rebuilt
817	}
818	Err(error) => {
819	log::warn!(
820	"local summarization round failed ({error}); falling back to plain truncation"
821	);
822	// Same deterministic shrink as the remote path, on the plain
823	// chat shapes the engine holds (its history has no tool-call
824	// structure, so there are no orphaned tool results to guard).
825	match truncate_chat_messages_to_budget(system, tail) {
826	Some(rebuilt) => rebuilt,
827	// The system messages alone overflow the window: unrecoverable.
828	None => return Err(error),
829	}
830	}
831	};
832
833	self.engine.clear_history().await;
834	for message in rebuild {
835	self.engine.push_history(message).await;
836	}
837	Ok(())
838	}
839	}
840
841	/// The local counterpart of `truncate_history_to_budget`: system messages, the
842	/// truncation placeholder, and the newest of `tail` that fit the budget.
843	///
844	/// The tail may drain to empty. A newest message that overflows the window on
845	/// its own (a giant tool result, the very thing that triggers compaction) is
846	/// dropped rather than failing the session, so `None` only means the system
847	/// messages alone don't fit.
848	fn truncate_chat_messages_to_budget(
849	system: Vec<ChatMessage>,
850	mut tail: Vec<ChatMessage>,
851	) -> Option<Vec<ChatMessage>> {
852	let budget = DEFAULT_CONTEXT_TOKEN_BUDGET * 4; // estimate_tokens is chars / 4.
853	let fixed_chars: usize = system
854	.iter()
855	.map(\|message\| message.content.len())
856	.sum::<usize>()
857	+ TRUNCATION_PLACEHOLDER.len();
858	if fixed_chars > budget {
859	return None;
860	}
861
862	let mut tail_chars: usize = tail.iter().map(\|message\| message.content.len()).sum();
863	let mut drop = 0;
864	while fixed_chars + tail_chars > budget {
865	tail_chars -= tail[drop].content.len();
866	drop += 1;
867	}
868	tail.drain(..drop);
869
870	let mut rebuilt = system;
871	rebuilt.push(ChatMessage::user(TRUNCATION_PLACEHOLDER.to_string()));
872	rebuilt.extend(tail);
873	Some(rebuilt)
874	}
875
876	/// Flatten `onde` chat messages into the transcript shape summarization
877	/// expects (the local engine's history has no tool-call structure, so this is
878	/// a plain role/content rendering).
879	fn chat_messages_transcript(history: &[ChatMessage]) -> String {
880	history
881	.iter()
882	.filter(\|message\| message.role != ChatRole::System && !message.content.trim().is_empty())
883	.map(\|message\| format!("{}: {}", message.role, message.content))
884	.collect::<Vec<_>>()
885	.join("\n\n")
886	}
887
888	/// Drain an onde streaming receiver, forwarding each token to `sink` and
889	/// assembling the full text. onde reports stream failures as a final chunk whose
890	/// `finish_reason` is `"error: …"`; surface those as a backend error.
891	async fn drain_onde_stream(
892	mut rx: tokio::sync::mpsc::Receiver<onde::inference::StreamChunk>,
893	sink: &TokenSink,
894	) -> Result<TurnResult, BackendError> {
895	let mut text = String::new();
896	let mut finish = FinishReason::default();
897	while let Some(chunk) = rx.recv().await {
898	if !chunk.delta.is_empty() {
899	text.push_str(&chunk.delta);
900	// The receiver is the UI; if it's gone the turn is being cancelled,
901	// so stop assembling rather than spinning the model to completion.
902	// onde reports reasoning as inline `<think>` tags in `delta`, so it
903	// is visible content that a downstream consumer splits out.
904	if sink.send(TokenChunk::Visible(chunk.delta)).is_err() {
905	break;
906	}
907	}
908	if chunk.done {
909	if let Some(reason) = chunk.finish_reason.as_deref()
910	&& let Some(message) = reason.strip_prefix("error: ")
911	{
912	return Err(message.to_string());
913	}
914	finish = FinishReason::from_wire(chunk.finish_reason.as_deref());
915	break;
916	}
917	}
918	Ok(TurnResult {
919	text,
920	tool_calls: Vec::new(),
921	finish,
922	context_tokens: None,
923	})
924	}
925
926	/// Convert an `onde` tool-aware result into the neutral [`TurnResult`].
927	fn onde_result_to_turn(result: onde::inference::ToolAwareResult) -> TurnResult {
928	TurnResult {
929	finish: FinishReason::from_wire(Some(&result.finish_reason)),
930	context_tokens: None,
931	text: result.text,
932	tool_calls: result
933	.tool_calls
934	.into_iter()
935	.map(\|call\| ToolCall {
936	id: call.id,
937	name: call.function_name,
938	arguments: call.arguments,
939	})
940	.collect(),
941	}
942	}
943
944	// ── OpenAI-compatible backend ─────────────────────────────────────────────────────
945
946	/// Inference against any OpenAI-compatible Chat Completions endpoint.
947	///
948	/// Conversation state is held client-side and replayed on every request, so the
949	/// endpoint can be stateless. Standard OpenAI function-calling is used end to
950	/// end (`tools`, `choices[].message.tool_calls`, `role: "tool"` follow-ups).
951	pub struct OpenAiBackend {
952	base_url: String,
953	api_key: String,
954	model: String,
955	http: reqwest::Client,
956	/// The full message list sent on each request (system + turns + tool results).
957	history: Mutex<Vec<serde_json::Value>>,
958	/// Whether `model` reads images (see `provider::model_accepts_images`).
959	/// When it does not, images already in `history` are left out of each
960	/// request rather than sent to an endpoint that would refuse them.
961	accepts_images: bool,
962	/// Whether `model` takes audio (see `provider::model_accepts_audio`), with
963	/// the same effect on audio already in `history`.
964	accepts_audio: bool,
965	/// Set once the endpoint has refused `stream_options`, so streamed
966	/// requests stop asking it for token usage.
967	stream_usage_refused: Arc<AtomicBool>,
968	}
969
970	impl OpenAiBackend {
971	/// Build a backend for `{base_url, api_key, model}`, seeding the optional
972	/// system prompt. `base_url` should include the API root (e.g. ending in
973	/// `/v1`); the chat path is appended.
974	pub fn new(
975	base_url: impl Into<String>,
976	api_key: impl Into<String>,
977	model: impl Into<String>,
978	system_prompt: Option<String>,
979	) -> Self {
980	let mut history = Vec::new();
981	if let Some(prompt) = system_prompt {
982	history.push(serde_json::json!({ "role": "system", "content": prompt }));
983	}
984	let model = model.into();
985	Self {
986	base_url: base_url.into(),
987	api_key: api_key.into(),
988	accepts_images: crate::provider::model_accepts_images(&model),
989	accepts_audio: crate::provider::model_accepts_audio(&model),
990	model,
991	http: reqwest::Client::new(),
992	history: Mutex::new(history),
993	stream_usage_refused: Arc::default(),
994	}
995	}
996
997	fn tools_json(tools: &[ToolSpec]) -> Vec<serde_json::Value> {
998	tools
999	.iter()
1000	.map(\|tool\| {
1001	// parameters_schema is a JSON string; parse it, defaulting to an
1002	// empty object schema if malformed.
1003	let parameters: serde_json::Value = serde_json::from_str(&tool.parameters_schema)
1004	.unwrap_or_else(\|_\| serde_json::json!({ "type": "object", "properties": {} }));
1005	serde_json::json!({
1006	"type": "function",
1007	"function": {
1008	"name": tool.name,
1009	"description": tool.description,
1010	"parameters": parameters,
1011	}
1012	})
1013	})
1014	.collect()
1015	}
1016
1017	/// POST the current history and apply the assistant reply to history.
1018	/// `tools` is always the known catalog, while `allow_tool_calls` determines
1019	/// whether it is advertised to the model and whether returned calls may run.
1020	/// Streams via SSE when `sink` is set; otherwise reads a single JSON response.
1021	///
1022	/// A reply whose only attempt at a tool call was a block that couldn't be
1023	/// parsed gets one retry: the model is told the call didn't run and asked
1024	/// to make it again through the structured interface. Without that the
1025	/// turn ends on whatever prose preceded the broken block, and the call is
1026	/// silently lost.
1027	async fn complete(
1028	&self,
1029	tools: &[ToolSpec],
1030	allow_tool_calls: bool,
1031	sink: Option<&TokenSink>,
1032	) -> Result<TurnResult, BackendError> {
1033	let (result, malformed) = self.request(tools, allow_tool_calls, sink).await?;
1034	let retry_note = if !allow_tool_calls \|\| !result.tool_calls.is_empty() {
1035	None
1036	} else if malformed > 0 {
1037	log::warn!(
1038	"dropped {malformed} unparseable inline tool call(s); asking the model to retry"
1039	);
1040	Some(MALFORMED_TOOL_CALL_RETRY)
1041	} else if result.finish == FinishReason::Repetition {
1042	log::warn!("reply was cut off as a repetition loop; asking the model to retry");
1043	Some(REPETITION_RETRY)
1044	} else {
1045	None
1046	};
1047	let Some(retry_note) = retry_note else {
1048	return Ok(result);
1049	};
1050
1051	self.history.lock().await.push(serde_json::json!({
1052	"role": "user",
1053	"content": retry_note,
1054	}));
1055	if let Some(sink) = sink
1056	&& !result.text.trim().is_empty()
1057	{
1058	// The retry's text follows what already streamed; keep it from
1059	// running on into the previous sentence.
1060	let _ = sink.send(TokenChunk::Visible("\n\n".to_string()));
1061	}
1062	let (retry, retry_malformed) = self.request(tools, allow_tool_calls, sink).await?;
1063	if retry_malformed > 0 && retry.tool_calls.is_empty() {
1064	log::warn!(
1065	"the retry also wrote {retry_malformed} unparseable inline tool call(s); \
1066	ending the turn without them"
1067	);
1068	}
1069	Ok(TurnResult {
1070	text: join_reply_text(&result.text, &retry.text),
1071	tool_calls: retry.tool_calls,
1072	finish: retry.finish,
1073	context_tokens: retry.context_tokens,
1074	})
1075	}
1076
1077	/// One chat-completion round trip. Returns the turn plus how many inline
1078	/// tool-call blocks were dropped because they couldn't be parsed.
1079	async fn request(
1080	&self,
1081	tools: &[ToolSpec],
1082	allow_tool_calls: bool,
1083	sink: Option<&TokenSink>,
1084	) -> Result<(TurnResult, usize), BackendError> {
1085	let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
1086	let streaming = sink.is_some();
1087
1088	let messages = {
1089	let history = self.history.lock().await;
1090	let messages = if self.accepts_images {
1091	history.clone()
1092	} else {
1093	without_images(&history)
1094	};
1095	if self.accepts_audio {
1096	messages
1097	} else {
1098	without_audio(&messages)
1099	}
1100	};
1101	let mut body = serde_json::json!({
1102	"model": self.model,
1103	"messages": messages,
1104	"stream": streaming,
1105	});
1106	// A stream reports no token usage unless asked to. It comes in a last
1107	// chunk with no choices, which feeds the client's context meter.
1108	let ask_for_usage = streaming && !self.stream_usage_refused.load(Ordering::Relaxed);
1109	if ask_for_usage {
1110	body["stream_options"] = serde_json::json!({ "include_usage": true });
1111	}
1112	if allow_tool_calls && !tools.is_empty() {
1113	body["tools"] = serde_json::Value::Array(Self::tools_json(tools));
1114	// OpenAI specifies `auto` as the default when tools are present,
1115	// but not every OpenAI-compatible gateway implements that default.
1116	// Sending it explicitly is important for the cloud path: otherwise
1117	// a model can describe the action it intends to take as text and
1118	// end the turn without returning a tool call at all.
1119	body["tool_choice"] = serde_json::Value::String("auto".to_string());
1120	}
1121
1122	let mut response = self
1123	.http
1124	.post(&url)
1125	.bearer_auth(&self.api_key)
1126	.json(&body)
1127	.send()
1128	.await
1129	.map_err(\|error\| format!("request to {url} failed: {error}"))?;
1130
1131	if !response.status().is_success() {
1132	let status = response.status();
1133	let error_body = response.text().await.unwrap_or_default();
1134	// Usage is a nicety; an endpoint that does not know the option is
1135	// asked again without it, and never asked for it again.
1136	if !(ask_for_usage && refuses_stream_options(status, &error_body)) {
1137	return Err(describe_api_error(status, &error_body));
1138	}
1139	log::info!("{url} refused stream_options; streaming without token usage");
1140	self.stream_usage_refused.store(true, Ordering::Relaxed);
1141	if let Some(fields) = body.as_object_mut() {
1142	fields.remove("stream_options");
1143	}
1144	response = self
1145	.http
1146	.post(&url)
1147	.bearer_auth(&self.api_key)
1148	.json(&body)
1149	.send()
1150	.await
1151	.map_err(\|error\| format!("request to {url} failed: {error}"))?;
1152	if !response.status().is_success() {
1153	let status = response.status();
1154	let error_body = response.text().await.unwrap_or_default();
1155	return Err(describe_api_error(status, &error_body));
1156	}
1157	}
1158
1159	if let Some(sink) = sink {
1160	self.consume_stream(response, sink, tools, allow_tool_calls)
1161	.await
1162	} else {
1163	self.consume_json(response, tools, allow_tool_calls).await
1164	}
1165	}
1166
1167	/// One self-contained summarization round: summarize `transcript` without
1168	/// touching the live history.
1169	///
1170	/// Unlike [`complete`], which replays and mutates `self.history`, this
1171	/// sends its own one-off message list. Compaction only rewrites history
1172	/// once it has a summary in hand, so a failed round leaves the session
1173	/// exactly as it was — and the caller can fall back to plain truncation
1174	/// instead of failing the session (issue #125).
1175	async fn summarize(
1176	&self,
1177	system: &[serde_json::Value],
1178	transcript: &str,
1179	) -> Result<String, BackendError> {
1180	let url = format!("{}/chat/completions", self.base_url.trim_end_matches('/'));
1181	let mut messages = Vec::with_capacity(system.len() + 1);
1182	messages.extend(system.iter().cloned());
1183	messages.push(serde_json::json!({
1184	"role": "user",
1185	"content": format!("{transcript}\n\n{SUMMARIZE_PROMPT}"),
1186	}));
1187	let body = serde_json::json!({
1188	"model": self.model,
1189	"messages": messages,
1190	"stream": false,
1191	});
1192
1193	let response = self
1194	.http
1195	.post(&url)
1196	.bearer_auth(&self.api_key)
1197	.json(&body)
1198	.send()
1199	.await
1200	.map_err(\|error\| format!("request to {url} failed: {error}"))?;
1201
1202	if !response.status().is_success() {
1203	let status = response.status();
1204	let body = response.text().await.unwrap_or_default();
1205	return Err(describe_api_error(status, &body));
1206	}
1207
1208	let parsed: ChatCompletion = response
1209	.json()
1210	.await
1211	.map_err(\|error\| format!("response parse error: {error}"))?;
1212	parsed
1213	.choices
1214	.into_iter()
1215	.next()
1216	.and_then(\|choice\| choice.message.content)
1217	.filter(\|text\| !text.trim().is_empty())
1218	.ok_or_else(\|\| "endpoint returned no summary".to_string())
1219	}
1220
1221	/// Parse a single non-streaming chat-completion response.
1222	async fn consume_json(
1223	&self,
1224	response: reqwest::Response,
1225	tools: &[ToolSpec],
1226	allow_tool_calls: bool,
1227	) -> Result<(TurnResult, usize), BackendError> {
1228	let parsed: ChatCompletion = response
1229	.json()
1230	.await
1231	.map_err(\|error\| format!("response parse error: {error}"))?;
1232	let context_tokens = parsed.usage.as_ref().map(Usage::context_tokens);
1233
1234	let choice = parsed
1235	.choices
1236	.into_iter()
1237	.next()
1238	.ok_or_else(\|\| "endpoint returned no choices".to_string())?;
1239	let finish = FinishReason::from_wire(choice.finish_reason.as_deref());
1240	let mut message = choice.message;
1241
1242	let text = message.content.clone().unwrap_or_default();
1243	let mut tool_calls: Vec<ToolCall> = message
1244	.tool_calls
1245	.iter()
1246	.flatten()
1247	.map(\|call\| ToolCall {
1248	id: call.id.clone(),
1249	name: call.function.name.clone(),
1250	arguments: call.function.arguments.clone(),
1251	})
1252	.collect();
1253
1254	let mut extracted = crate::inline_tool_calls::extract(&text, tools);
1255	// Markup the model made up in the harness's shape goes too, from the
1256	// reply and the history alike. See `harness_markup`.
1257	extracted.text = crate::harness_markup::strip(&extracted.text);
1258	let malformed = extracted.malformed;
1259	if malformed > 0 {
1260	log::warn!("dropped {malformed} unparseable inline tool call(s) from the reply");
1261	}
1262
1263	if !allow_tool_calls {
1264	let recovered = extracted.calls;
1265	let text = extracted.text;
1266	let suppressed = tool_calls.len() + recovered.len();
1267	if suppressed > 0 {
1268	let names = tool_calls
1269	.iter()
1270	.map(\|call\| call.name.as_str())
1271	.chain(recovered.iter().map(\|call\| call.name.as_str()))
1272	.collect::<Vec<_>>()
1273	.join(", ");
1274	log::warn!(
1275	"suppressed {suppressed} tool call(s) from a forced-text response: {names}"
1276	);
1277	}
1278	self.history
1279	.lock()
1280	.await
1281	.push(streamed_assistant_history(&text, &[]));
1282	return Ok((
1283	TurnResult {
1284	text,
1285	tool_calls: Vec::new(),
1286	finish,
1287	context_tokens,
1288	},
1289	malformed,
1290	));
1291	}
1292
1293	// Some models write a tool call out as literal `<tool_call>` text
1294	// instead of using the structured field (see `inline_tool_calls`).
1295	// Recover it, or the turn ends with the tag rendered as prose and
1296	// whatever the model meant to do is dropped.
1297	if tool_calls.is_empty() && (malformed > 0 \|\| !extracted.calls.is_empty()) {
1298	let cleaned = extracted.text;
1299	let recovered = extracted.calls;
1300	if !recovered.is_empty() {
1301	log::warn!(
1302	"recovered {} tool call(s) the model emitted as text instead of a structured call",
1303	recovered.len()
1304	);
1305	let tool_calls: Vec<ToolCall> = recovered
1306	.into_iter()
1307	.enumerate()
1308	.map(\|(index, call)\| ToolCall {
1309	id: format!("call_recovered_{index}"),
1310	name: call.name,
1311	arguments: call.arguments,
1312	})
1313	.collect();
1314	// Record the recovered shape, not the raw tag: the tool
1315	// results that follow have to answer an assistant message
1316	// that actually carries these calls, or the next request is
1317	// rejected for orphaned tool results.
1318	self.history
1319	.lock()
1320	.await
1321	.push(streamed_assistant_history(&cleaned, &tool_calls));
1322	return Ok((
1323	TurnResult {
1324	text: cleaned,
1325	tool_calls,
1326	finish,
1327	context_tokens,
1328	},
1329	malformed,
1330	));
1331	}
1332	// Nothing recovered, so what was found is a broken block. Keep it
1333	// out of history as well as the reply; see `inline_tool_calls` for
1334	// why it poisons later turns.
1335	self.push_reply_without_calls(&cleaned).await;
1336	return Ok((
1337	TurnResult {
1338	text: cleaned,
1339	tool_calls,
1340	finish,
1341	context_tokens,
1342	},
1343	malformed,
1344	));
1345	} else if malformed > 0 \|\| !extracted.calls.is_empty() {
1346	// Structured calls arrived with inline blocks beside them. As on
1347	// the streaming path, inline calls that parse run alongside the
1348	// structured ones, broken blocks are dropped, and neither kind of
1349	// block is left in the reply or the history.
1350	let offset = tool_calls.len();
1351	tool_calls.extend(
1352	extracted
1353	.calls
1354	.into_iter()
1355	.enumerate()
1356	.map(\|(index, call)\| ToolCall {
1357	id: format!("call_recovered_{}", offset + index),
1358	name: call.name,
1359	arguments: call.arguments,
1360	}),
1361	);
1362	self.history
1363	.lock()
1364	.await
1365	.push(streamed_assistant_history(&extracted.text, &tool_calls));
1366	return Ok((
1367	TurnResult {
1368	text: extracted.text,
1369	tool_calls,
1370	finish,
1371	context_tokens,
1372	},
1373	malformed,
1374	));
1375	}
1376
1377	// Record the assistant turn so later tool results have context. No
1378	// inline block was found, so the only change `extracted.text` can
1379	// carry is the stripped markup.
1380	let text = extracted.text;
1381	if message.content.is_some() {
1382	message.content = Some(text.clone());
1383	}
1384	self.history.lock().await.push(message.into_history_value());
1385
1386	Ok((
1387	TurnResult {
1388	text,
1389	tool_calls,
1390	finish,
1391	context_tokens,
1392	},
1393	0,
1394	))
1395	}
1396
1397	/// Record a text-only assistant reply. An empty one is skipped: strict
1398	/// endpoints reject an assistant message with neither content nor tool
1399	/// calls, and there is nothing in it worth replaying.
1400	async fn push_reply_without_calls(&self, text: &str) {
1401	if !text.is_empty() {
1402	self.history
1403	.lock()
1404	.await
1405	.push(streamed_assistant_history(text, &[]));
1406	}
1407	}
1408
1409	/// Consume an OpenAI Server-Sent Events stream, forwarding content deltas to
1410	/// `sink` and reassembling any tool calls (which arrive fragmented across
1411	/// chunks, keyed by `index`).
1412	async fn consume_stream(
1413	&self,
1414	response: reqwest::Response,
1415	sink: &TokenSink,
1416	tools: &[ToolSpec],
1417	allow_tool_calls: bool,
1418	) -> Result<(TurnResult, usize), BackendError> {
1419	use futures::StreamExt;
1420
1421	let mut stream = response.bytes_stream();
1422	// Newlines are ASCII, so splitting raw bytes on `\n` never bisects a
1423	// multibyte UTF-8 sequence; we only lossily decode whole lines.
1424	let mut buffer: Vec<u8> = Vec::new();
1425	let mut text = String::new();
1426	let mut tool_accum: Vec<StreamingToolCall> = Vec::new();
1427	let mut done = false;
1428	// Arrives on a late chunk, usually one with an empty delta.
1429	let mut finish = FinishReason::default();
1430	// Arrives on the last chunk, which has no choices, when asked for.
1431	let mut context_tokens = None;
1432	// Recovers a tool call the model wrote as literal `<tool_call>` text
1433	// instead of a structured delta. Scanning here (rather than after the
1434	// stream) keeps the tag off the UI: content goes straight to `sink` as
1435	// it arrives, so by the time a whole turn is assembled the tag has
1436	// already been rendered. See `inline_tool_calls`.
1437	let mut scanner = crate::inline_tool_calls::StreamScanner::new(tools);
1438	// What it lets through then loses any markup the model made up in the
1439	// harness's shape, like a `<system_warning>` block. See
1440	// `harness_markup`.
1441	let mut markup = crate::harness_markup::MarkupFilter::new();
1442	let mut recovered: Vec<ToolCall> = Vec::new();
1443	let mut malformed = 0usize;
1444	// Stops a reply that falls into a repetition loop. See `repetition`.
1445	let mut guard = crate::repetition::RepetitionGuard::new();
1446	let mut degenerate = false;
1447
1448	while let Some(item) = stream.next().await {
1449	let bytes = item.map_err(\|error\| format!("stream read error: {error}"))?;
1450	buffer.extend_from_slice(&bytes);
1451
1452	while let Some(pos) = buffer.iter().position(\|&b\| b == b'\n') {
1453	let line: Vec<u8> = buffer.drain(..=pos).collect();
1454	let line = String::from_utf8_lossy(&line);
1455	let line = line.trim();
1456
1457	let Some(data) = line.strip_prefix("data:") else {
1458	continue;
1459	};
1460	let data = data.trim();
1461	if data == "[DONE]" {
1462	done = true;
1463	break;
1464	}
1465	if data.is_empty() {
1466	continue;
1467	}
1468
1469	// An endpoint that fails after the status line is already sent
1470	// reports it in the stream instead, as an ordinary `data:`
1471	// frame holding an error envelope. siGit Code Cloud does this
1472	// when the upstream fails mid-turn. It has no `choices`, so
1473	// without this it parses as an empty chunk and is skipped, and
1474	// the turn ends looking like the model simply said nothing.
1475	if let Some(message) = api_error_message(data) {
1476	return Err(message);
1477	}
1478
1479	let chunk: StreamCompletion = match serde_json::from_str(data) {
1480	Ok(chunk) => chunk,
1481	// Skip keep-alive comments and anything we can't parse rather
1482	// than aborting a turn over one malformed frame.
1483	Err(_) => continue,
1484	};
1485
1486	if let Some(usage) = &chunk.usage {
1487	context_tokens = Some(usage.context_tokens());
1488	}
1489	let Some(choice) = chunk.choices.into_iter().next() else {
1490	continue;
1491	};
1492	if let Some(reason) = choice.finish_reason.as_deref() {
1493	finish = FinishReason::from_wire(Some(reason));
1494	}
1495	// Reasoning streamed in its own field is tagged at the source;
1496	// it bypasses the inline-tag scanner and is never kept in the
1497	// assistant message text.
1498	if let Some(reasoning) = choice.delta.reasoning_content
1499	&& !reasoning.is_empty()
1500	&& sink.send(TokenChunk::Reasoning(reasoning)).is_err()
1501	{
1502	// Consumer dropped (turn cancelled).
1503	done = true;
1504	break;
1505	}
1506	if let Some(content) = choice.delta.content
1507	&& !content.is_empty()
1508	{
1509	let mut cancelled = false;
1510	for event in scanner.push(&content) {
1511	match event {
1512	crate::inline_tool_calls::ScanEvent::Text(chunk) => {
1513	let chunk = markup.push(&chunk);
1514	if chunk.is_empty() {
1515	continue;
1516	}
1517	text.push_str(&chunk);
1518	let looping = guard.push(&chunk);
1519	if sink.send(TokenChunk::Visible(chunk)).is_err() {
1520	// Consumer dropped (turn cancelled).
1521	cancelled = true;
1522	break;
1523	}
1524	if looping {
1525	degenerate = true;
1526	break;
1527	}
1528	}
1529	crate::inline_tool_calls::ScanEvent::ToolCall(call) => {
1530	if allow_tool_calls {
1531	log::warn!(
1532	"recovered tool call '{}' the model emitted as text instead of a structured call",
1533	call.name
1534	);
1535	}
1536	recovered.push(ToolCall {
1537	id: format!("call_recovered_{}", recovered.len()),
1538	name: call.name,
1539	arguments: call.arguments,
1540	});
1541	}
1542	crate::inline_tool_calls::ScanEvent::Malformed(block) => {
1543	log_malformed_block(&block);
1544	malformed += 1;
1545	}
1546	}
1547	}
1548	if cancelled \|\| degenerate {
1549	done = true;
1550	break;
1551	}
1552	}
1553	for delta in choice.delta.tool_calls.into_iter().flatten() {
1554	let index = delta.index.unwrap_or(0) as usize;
1555	if tool_accum.len() <= index {
1556	tool_accum.resize_with(index + 1, StreamingToolCall::default);
1557	}
1558	let slot = &mut tool_accum[index];
1559	if let Some(id) = delta.id {
1560	slot.id = id;
1561	}
1562	if let Some(function) = delta.function {
1563	if let Some(name) = function.name {
1564	slot.name = name;
1565	}
1566	if let Some(arguments) = function.arguments {
1567	slot.arguments.push_str(&arguments);
1568	}
1569	}
1570	}
1571	}
1572
1573	if done {
1574	break;
1575	}
1576	}
1577
1578	// Flush what was held back. A partial marker is just text; a
1579	// tool-call block that never closed is dropped like any other
1580	// unparseable one.
1581	let mut leftover = String::new();
1582	if degenerate {
1583	// Dropping the stream below closes the connection, so the endpoint
1584	// stops generating. Nothing held back is worth flushing.
1585	let keep = guard.loop_start().unwrap_or(text.len()).min(text.len());
1586	log::warn!(
1587	"reply degenerated into repetition; cut {} of {} bytes",
1588	text.len() - keep,
1589	text.len()
1590	);
1591	text.truncate(keep);
1592	text.truncate(text.trim_end().len());
1593	finish = FinishReason::Repetition;
1594	let _ = sink.send(TokenChunk::Visible(REPETITION_NOTICE.to_string()));
1595	} else {
1596	match scanner.finish() {
1597	Some(crate::inline_tool_calls::ScanEvent::Text(rest)) => {
1598	leftover = markup.push(&rest);
1599	}
1600	Some(crate::inline_tool_calls::ScanEvent::Malformed(block)) => {
1601	log_malformed_block(&block);
1602	malformed += 1;
1603	}
1604	Some(crate::inline_tool_calls::ScanEvent::ToolCall(_)) \| None => {}
1605	}
1606	leftover.push_str(&markup.finish());
1607	if !leftover.is_empty() {
1608	text.push_str(&leftover);
1609	let _ = sink.send(TokenChunk::Visible(leftover));
1610	}
1611	}
1612
1613	let mut tool_calls: Vec<ToolCall> = tool_accum
1614	.iter()
1615	.filter(\|call\| !call.name.is_empty())
1616	.enumerate()
1617	.map(\|(index, call)\| ToolCall {
1618	id: if call.id.is_empty() {
1619	format!("call_{index}")
1620	} else {
1621	call.id.clone()
1622	},
1623	name: call.name.clone(),
1624	arguments: call.arguments.clone(),
1625	})
1626	.collect();
1627	tool_calls.extend(recovered);
1628
1629	if !allow_tool_calls && !tool_calls.is_empty() {
1630	let names = tool_calls
1631	.iter()
1632	.map(\|call\| call.name.as_str())
1633	.collect::<Vec<_>>()
1634	.join(", ");
1635	log::warn!(
1636	"suppressed {} tool call(s) from a forced-text response: {names}",
1637	tool_calls.len()
1638	);
1639	tool_calls.clear();
1640	}
1641
1642	// Record the assistant turn so later tool results have context.
1643	if (malformed > 0 \|\| degenerate) && tool_calls.is_empty() {
1644	self.push_reply_without_calls(&text).await;
1645	} else {
1646	self.history
1647	.lock()
1648	.await
1649	.push(streamed_assistant_history(&text, &tool_calls));
1650	}
1651
1652	Ok((
1653	TurnResult {
1654	text,
1655	tool_calls,
1656	finish,
1657	context_tokens,
1658	},
1659	malformed,
1660	))
1661	}
1662	}
1663
1664	/// Sent as a user turn after a reply whose tool call couldn't be parsed.
1665	const MALFORMED_TOOL_CALL_RETRY: &str = "[siGit Code] Your last reply tried to call a tool by \
1666	writing the call out as text, and it could not be parsed, so nothing ran and the user did \
1667	not see it. Do not write tool calls or tool results as text. Make the call again using the \
1668	tool-calling interface, or answer in plain prose if no tool is needed.";
1669
1670	/// Sent as a user turn after a reply was cut off as a repetition loop.
1671	const REPETITION_RETRY: &str = "[siGit Code] Your last reply started repeating the same few \
1672	words and was cut off. Do not continue it. Make the next tool call you need, or answer \
1673	briefly in plain prose.";
1674
1675	/// Shown to the user, who has already watched the loop stream. Not kept in
1676	/// history.
1677	const REPETITION_NOTICE: &str =
1678	"\n\n[siGit Code stopped this reply because the model started repeating itself.]";
1679
1680	fn log_malformed_block(block: &str) {
1681	log::warn!(
1682	"dropped an unparseable inline tool call ({} chars): {}",
1683	block.len(),
1684	block.chars().take(200).collect::<String>()
1685	);
1686	}
1687
1688	/// Join the visible text of a reply and its retry the way the stream showed
1689	/// them: as separate paragraphs.
1690	fn join_reply_text(first: &str, retry: &str) -> String {
1691	match (first.trim().is_empty(), retry.trim().is_empty()) {
1692	(true, _) => retry.to_string(),
1693	(false, true) => first.to_string(),
1694	(false, false) => format!("{first}\n\n{retry}"),
1695	}
1696	}
1697
1698	/// One tool call being reassembled from streamed deltas.
1699	#[derive(Default)]
1700	struct StreamingToolCall {
1701	id: String,
1702	name: String,
1703	arguments: String,
1704	}
1705
1706	/// Rebuild the assistant message for replay in history after a streamed turn,
1707	/// preserving any tool calls so the follow-up request is well-formed. Mirrors
1708	/// [`ResponseMessage::into_history_value`] for the non-streaming path.
1709	fn streamed_assistant_history(text: &str, tool_calls: &[ToolCall]) -> serde_json::Value {
1710	let mut message = serde_json::json!({ "role": "assistant" });
1711	message["content"] = if text.is_empty() {
1712	serde_json::Value::Null
1713	} else {
1714	serde_json::Value::String(text.to_string())
1715	};
1716	if !tool_calls.is_empty() {
1717	message["tool_calls"] = serde_json::json!(
1718	tool_calls
1719	.iter()
1720	.map(\|call\| serde_json::json!({
1721	"id": call.id,
1722	"type": "function",
1723	"function": {
1724	"name": call.name,
1725	"arguments": call.arguments,
1726	}
1727	}))
1728	.collect::<Vec<_>>()
1729	);
1730	}
1731	message
1732	}
1733
1734	#[async_trait]
1735	impl InferenceBackend for OpenAiBackend {
1736	async fn send_message_with_tools(
1737	&self,
1738	text: &str,
1739	tools: &[ToolSpec],
1740	sink: Option<&TokenSink>,
1741	) -> Result<TurnResult, BackendError> {
1742	self.history
1743	.lock()
1744	.await
1745	.push(serde_json::json!({ "role": "user", "content": text }));
1746	self.complete(tools, true, sink).await
1747	}
1748
1749	async fn send_tool_results(
1750	&self,
1751	results: Vec<ToolResult>,
1752	tools: &[ToolSpec],
1753	allow_tool_calls: bool,
1754	sink: Option<&TokenSink>,
1755	) -> Result<TurnResult, BackendError> {
1756	{
1757	let mut history = self.history.lock().await;
1758	for result in results {
1759	history.push(serde_json::json!({
1760	"role": "tool",
1761	"tool_call_id": result.tool_call_id,
1762	"content": result.content,
1763	}));
1764	}
1765	}
1766	self.complete(tools, allow_tool_calls, sink).await
1767	}
1768
1769	async fn record_cancelled_tool_results(&self, results: Vec<ToolResult>) {
1770	let mut history = self.history.lock().await;
1771	for result in results {
1772	// Cancellation races the in-flight continuation request. That
1773	// request appends its tool results before awaiting HTTP, so cleanup
1774	// must be safe whether cancellation won just before or just after
1775	// the append.
1776	let matching_call = history.iter().rposition(\|message\| {
1777	message["role"] == "assistant"
1778	&& message["tool_calls"].as_array().is_some_and(\|calls\| {
1779	calls.iter().any(\|call\| call["id"] == result.tool_call_id)
1780	})
1781	});
1782	let already_recorded = matching_call.is_some_and(\|call_index\| {
1783	history[call_index + 1..].iter().any(\|message\| {
1784	message["role"] == "tool" && message["tool_call_id"] == result.tool_call_id
1785	})
1786	});
1787	if already_recorded {
1788	continue;
1789	}
1790	history.push(serde_json::json!({
1791	"role": "tool",
1792	"tool_call_id": result.tool_call_id,
1793	"content": result.content,
1794	}));
1795	}
1796	}
1797
1798	async fn send_message_with_attachments(
1799	&self,
1800	text: &str,
1801	images: &[ImageInput],
1802	audio: &[AudioInput],
1803	tools: &[ToolSpec],
1804	sink: Option<&TokenSink>,
1805	) -> Result<TurnResult, BackendError> {
1806	self.history
1807	.lock()
1808	.await
1809	.push(user_message(text, images, audio));
1810	self.complete(tools, true, sink).await
1811	}
1812
1813	fn accepts_images(&self) -> bool {
1814	self.accepts_images
1815	}
1816
1817	fn accepts_audio(&self) -> bool {
1818	self.accepts_audio
1819	}
1820
1821	fn is_remote(&self) -> bool {
1822	true
1823	}
1824
1825	fn fresh(&self) -> Option<Arc<dyn InferenceBackend>> {
1826	Some(Arc::new(Self {
1827	base_url: self.base_url.clone(),
1828	api_key: self.api_key.clone(),
1829	model: self.model.clone(),
1830	// The client is a handle on one connection pool; sharing it keeps
1831	// connections to the endpoint warm across sessions.
1832	http: self.http.clone(),
1833	history: Mutex::new(Vec::new()),
1834	accepts_images: self.accepts_images,
1835	accepts_audio: self.accepts_audio,
1836	// It is the same endpoint, so what it refused it refuses here.
1837	stream_usage_refused: Arc::clone(&self.stream_usage_refused),
1838	}))
1839	}
1840
1841	async fn history_snapshot(&self) -> Vec<serde_json::Value> {
1842	self.history.lock().await.clone()
1843	}
1844
1845	async fn restore_history(&self, history: Vec<serde_json::Value>) {
1846	// The snapshot includes the seeded system message, so a wholesale
1847	// replacement restores exactly what was saved.
1848	*self.history.lock().await = history;
1849	}
1850
1851	async fn compact_history(&self, keep_last: usize) -> Result<(), BackendError> {
1852	let snapshot: Vec<serde_json::Value> = self.history.lock().await.clone();
1853
1854	// Leading system messages carry session context (the seeded prompt,
1855	// pushed context); keep them all, like the summarization-success path.
1856	let system: Vec<serde_json::Value> = snapshot
1857	.iter()
1858	.take_while(\|message\| message["role"] == "system")
1859	.cloned()
1860	.collect();
1861
1862	let non_system: Vec<serde_json::Value> = snapshot
1863	.iter()
1864	.filter(\|message\| message["role"] != "system")
1865	.cloned()
1866	.collect();
1867	let tail_start = non_system.len().saturating_sub(keep_last);
1868	let mut tail = non_system[tail_start..].to_vec();
1869	// Drop leading tool results whose assistant tool-call message was
1870	// summarized away — strict endpoints reject orphaned `role: "tool"`
1871	// entries on the very next request.
1872	while tail
1873	.first()
1874	.is_some_and(\|message\| message["role"] == "tool")
1875	{
1876	tail.remove(0);
1877	}
1878
1879	// Ask the endpoint for a summary of the conversation so far, through a
1880	// direct completion call. The request carries the conversation as a
1881	// flattened transcript in a single user message rather than the live
1882	// history: this round offers no tools, and a tool-shaped history sent
1883	// without a tool schema is rejected upstream (see
1884	// `transcript_for_summary`). The transcript is capped so the request
1885	// cannot be as large as the over-budget conversation that triggered
1886	// compaction (see `SUMMARY_TRANSCRIPT_TOKEN_CAP`).
1887	let transcript = truncate_transcript_middle(
1888	&transcript_for_summary(&snapshot),
1889	SUMMARY_TRANSCRIPT_TOKEN_CAP,
1890	);
1891	let summary = match self.summarize(&system, &transcript).await {
1892	Ok(summary) => summary,
1893	Err(error) => {
1894	log::warn!(
1895	"summarization round failed ({error}); falling back to plain truncation"
1896	);
1897	// Compaction must not fail the session: fold the overflow into
1898	// a placeholder and drop the oldest messages until the rebuild
1899	// fits the budget. Needs no endpoint, so it cannot fail the way
1900	// the inference round just did (issue #125).
1901	let rebuilt = truncate_history_to_budget(&system, tail.clone()).ok_or(error)?;
1902	*self.history.lock().await = rebuilt;
1903	return Ok(());
1904	}
1905	};
1906
1907	let mut rebuilt = Vec::with_capacity(system.len() + tail.len() + 1);
1908	rebuilt.extend(system);
1909	rebuilt.push(serde_json::json!({
1910	"role": "user",
1911	"content": format!("[Conversation summary]\n{summary}"),
1912	}));
1913	rebuilt.extend(tail);
1914	*self.history.lock().await = rebuilt;
1915	Ok(())
1916	}
1917	}
1918
1919	/// The placeholder message replacing dropped history when compaction falls
1920	/// back to plain truncation (the model reads it, so it explains the gap).
1921	const TRUNCATION_PLACEHOLDER: &str = "[Older conversation omitted: the session outgrew \
1922	the context window and the summarization service was unavailable. Earlier \
1923	messages were dropped, so answers may lack that context.]";
1924
1925	/// Last-resort compaction when the summarization round cannot run: rebuild the
1926	/// history as `[system, placeholder, newest messages that fit the budget]`.
1927	///
1928	/// Returns `None` when even the smallest rebuild exceeds the budget (a
1929	/// pathological case — e.g. one message larger than the whole window); the
1930	/// caller then reports the original summarization error and leaves history
1931	/// untouched.
1932	fn truncate_history_to_budget(
1933	system: &[serde_json::Value],
1934	mut tail: Vec<serde_json::Value>,
1935	) -> Option<Vec<serde_json::Value>> {
1936	/// A `role: "tool"` entry only makes sense behind the assistant message
1937	/// that requested it; dropping that message strands it.
1938	fn drop_orphaned_tool_results(tail: &mut Vec<serde_json::Value>) {
1939	while tail
1940	.first()
1941	.is_some_and(\|message\| message["role"] == "tool")
1942	{
1943	tail.remove(0);
1944	}
1945	}
1946
1947	// A caller splitting mid-round can hand in a tail that starts with tool
1948	// results whose call message already fell outside the kept window.
1949	drop_orphaned_tool_results(&mut tail);
1950
1951	loop {
1952	let mut candidate = Vec::with_capacity(tail.len() + system.len() + 1);
1953	candidate.extend(system.iter().cloned());
1954	candidate.push(serde_json::json!({
1955	"role": "user",
1956	"content": TRUNCATION_PLACEHOLDER,
1957	}));
1958	candidate.extend(tail.iter().cloned());
1959	if estimate_tokens(&candidate) <= DEFAULT_CONTEXT_TOKEN_BUDGET {
1960	return Some(candidate);
1961	}
1962
1963	// Nothing left to drop: even the minimal rebuild (system + placeholder,
1964	// no conversation) overflows — the session is genuinely unrecoverable.
1965	if tail.is_empty() {
1966	return None;
1967	}
1968	if tail[0]["role"] == "assistant"
1969	&& tail[0]["tool_calls"]
1970	.as_array()
1971	.is_some_and(\|calls\| !calls.is_empty())
1972	&& tail.get(1).is_some_and(\|message\| message["role"] == "tool")
1973	{
1974	// Dropping an assistant tool-call message would orphan its
1975	// results; drop the whole round (call + results) together.
1976	let mut end = 1;
1977	while tail
1978	.get(end)
1979	.is_some_and(\|message\| message["role"] == "tool")
1980	{
1981	end += 1;
1982	}
1983	tail.drain(..end);
1984	} else {
1985	tail.remove(0);
1986	drop_orphaned_tool_results(&mut tail);
1987	}
1988	}
1989	}
1990
1991	// ── OpenAI error shape ────────────────────────────────────────────────────────
1992
1993	/// The OpenAI error envelope: `{"error": {"message": ..., "type": ...}}`.
1994	///
1995	/// Every endpoint siGit talks to speaks it, including siGit Code Cloud, whose
1996	/// messages are written for the person reading them ("Monthly siGit Code Cloud
1997	/// allowance reached..."). Worth unwrapping rather than pasting the raw body
1998	/// into the editor's error banner.
1999	#[derive(Debug, Deserialize)]
2000	struct ApiErrorEnvelope {
2001	error: ApiErrorBody,
2002	}
2003
2004	#[derive(Debug, Deserialize)]
2005	struct ApiErrorBody {
2006	#[serde(default)]
2007	message: Option<String>,
2008	}
2009
2010	/// How much of an unparseable error body to keep. An endpoint behind a proxy
2011	/// can answer with a full HTML page, and the whole thing ends up in the
2012	/// editor's error banner.
2013	const ERROR_BODY_LIMIT: usize = 500;
2014
2015	/// Turn an error response into something worth showing a person.
2016	///
2017	/// The endpoint's own message wins when there is one: it is written for the
2018	/// user, and the status code repeats what it already says. Anything else falls
2019	/// back to the status plus whatever the body held, which is all there is to go
2020	/// on — except an HTML page, which is a proxy or Rails error page and tells
2021	/// the person nothing the status doesn't (pasting it into chat is the dump
2022	/// seen in issue #125's screenshots).
2023	fn describe_api_error(status: reqwest::StatusCode, body: &str) -> String {
2024	if let Some(message) = api_error_message(body) {
2025	return message;
2026	}
2027
2028	let body = body.trim();
2029	if body.is_empty() \|\| is_probable_html(body) {
2030	return format!("endpoint returned {status}");
2031	}
2032
2033	let mut detail = body;
2034	if detail.len() > ERROR_BODY_LIMIT {
2035	let mut cut = ERROR_BODY_LIMIT;
2036	while !detail.is_char_boundary(cut) {
2037	cut -= 1;
2038	}
2039	detail = &detail[..cut];
2040	return format!("endpoint returned {status}: {detail}…");
2041	}
2042	format!("endpoint returned {status}: {detail}")
2043	}
2044
2045	/// The human-readable message out of an OpenAI error envelope, if the body is
2046	/// one and carries a non-empty message.
2047	fn api_error_message(body: &str) -> Option<String> {
2048	let envelope: ApiErrorEnvelope = serde_json::from_str(body).ok()?;
2049	let message = envelope.error.message?;
2050	let message = message.trim();
2051	(!message.is_empty()).then(\|\| message.to_string())
2052	}
2053
2054	/// Whether `body` looks like an HTML page rather than prose: a doctype or an
2055	/// opening tag in the first chunk. Gateways and Rails apps answer failures
2056	/// with full pages; none of that markup is worth showing.
2057	fn is_probable_html(body: &str) -> bool {
2058	let head: String = body.chars().take(512).collect();
2059	let head = head.trim_start().to_ascii_lowercase();
2060	head.starts_with("<!doctype") \|\| head.starts_with("<html") \|\| head.starts_with("<head")
2061	}
2062
2063	// ── OpenAI response shapes ────────────────────────────────────────────────────────
2064
2065	#[derive(Debug, Deserialize)]
2066	struct ChatCompletion {
2067	#[serde(default)]
2068	choices: Vec<CompletionChoice>,
2069	#[serde(default)]
2070	usage: Option<Usage>,
2071	}
2072
2073	/// Token usage as an OpenAI-compatible endpoint reports it.
2074	#[derive(Debug, Deserialize)]
2075	struct Usage {
2076	#[serde(default)]
2077	prompt_tokens: u64,
2078	#[serde(default)]
2079	completion_tokens: u64,
2080	#[serde(default)]
2081	total_tokens: Option<u64>,
2082	}
2083
2084	impl Usage {
2085	/// The context in use once the reply is in.
2086	fn context_tokens(&self) -> u64 {
2087	self.total_tokens
2088	.unwrap_or(self.prompt_tokens + self.completion_tokens)
2089	}
2090	}
2091
2092	/// Whether a failed request was turned down for carrying `stream_options`,
2093	/// which some OpenAI-compatible servers predate and reject as unknown.
2094	fn refuses_stream_options(status: reqwest::StatusCode, body: &str) -> bool {
2095	matches!(status.as_u16(), 400 \| 422) && body.contains("stream_options")
2096	}
2097
2098	#[derive(Debug, Deserialize)]
2099	struct CompletionChoice {
2100	message: ResponseMessage,
2101	#[serde(default)]
2102	finish_reason: Option<String>,
2103	}
2104
2105	#[derive(Debug, Deserialize)]
2106	struct ResponseMessage {
2107	#[serde(default)]
2108	content: Option<String>,
2109	#[serde(default)]
2110	tool_calls: Option<Vec<ResponseToolCall>>,
2111	}
2112
2113	impl ResponseMessage {
2114	/// Reconstruct the assistant message for replay in history, preserving any
2115	/// tool calls so the follow-up request is well-formed.
2116	fn into_history_value(self) -> serde_json::Value {
2117	let mut message = serde_json::json!({ "role": "assistant" });
2118	message["content"] = match self.content {
2119	Some(text) => serde_json::Value::String(text),
2120	None => serde_json::Value::Null,
2121	};
2122	if let Some(tool_calls) = self.tool_calls {
2123	message["tool_calls"] = serde_json::json!(
2124	tool_calls
2125	.into_iter()
2126	.map(\|call\| serde_json::json!({
2127	"id": call.id,
2128	"type": "function",
2129	"function": {
2130	"name": call.function.name,
2131	"arguments": call.function.arguments,
2132	}
2133	}))
2134	.collect::<Vec<_>>()
2135	);
2136	}
2137	message
2138	}
2139	}
2140
2141	#[derive(Debug, Deserialize)]
2142	struct ResponseToolCall {
2143	id: String,
2144	function: ResponseFunction,
2145	}
2146
2147	#[derive(Debug, Deserialize)]
2148	struct ResponseFunction {
2149	name: String,
2150	#[serde(default)]
2151	arguments: String,
2152	}
2153
2154	// ── OpenAI streaming (SSE) chunk shapes ─────────────────────────────────────────
2155
2156	#[derive(Debug, Deserialize)]
2157	struct StreamCompletion {
2158	#[serde(default)]
2159	choices: Vec<StreamChoice>,
2160	#[serde(default)]
2161	usage: Option<Usage>,
2162	}
2163
2164	#[derive(Debug, Deserialize)]
2165	struct StreamChoice {
2166	#[serde(default)]
2167	delta: StreamDelta,
2168	#[serde(default)]
2169	finish_reason: Option<String>,
2170	}
2171
2172	#[derive(Debug, Default, Deserialize)]
2173	struct StreamDelta {
2174	#[serde(default)]
2175	content: Option<String>,
2176	// Some endpoints stream model reasoning in a dedicated field rather than
2177	// inline `<think>` tags. It is unambiguously reasoning at the source, so it
2178	// travels as `TokenChunk::Reasoning` and never enters the content scanner.
2179	#[serde(default, alias = "reasoning")]
2180	reasoning_content: Option<String>,
2181	#[serde(default)]
2182	tool_calls: Option<Vec<StreamToolCallDelta>>,
2183	}
2184
2185	#[derive(Debug, Deserialize)]
2186	struct StreamToolCallDelta {
2187	#[serde(default)]
2188	index: Option<u32>,
2189	#[serde(default)]
2190	id: Option<String>,
2191	#[serde(default)]
2192	function: Option<StreamFunctionDelta>,
2193	}
2194
2195	#[derive(Debug, Deserialize)]
2196	struct StreamFunctionDelta {
2197	#[serde(default)]
2198	name: Option<String>,
2199	#[serde(default)]
2200	arguments: Option<String>,
2201	}
2202
2203	// ── Carrying a conversation across a model switch ────────────────────────────
2204
2205	/// The part of a history snapshot that survives a model switch: every
2206	/// non-system message, with half-finished tool plumbing repaired.
2207	///
2208	/// System messages are dropped because the backend being switched to seeds
2209	/// its own — a different model's prompt, or a freshly pushed session-context
2210	/// message. A switch can land mid-turn, after the assistant asked for a tool
2211	/// but before the results came back; a tool call with no matching result (or a
2212	/// result with no matching call) makes strict OpenAI-compatible endpoints
2213	/// reject every later request in the session, so those halves are stripped
2214	/// instead of carried.
2215	pub fn carryover_history(snapshot: Vec<serde_json::Value>) -> Vec<serde_json::Value> {
2216	let answered: std::collections::HashSet<String> = snapshot
2217	.iter()
2218	.filter(\|message\| message["role"] == "tool")
2219	.filter_map(\|message\| message["tool_call_id"].as_str().map(str::to_string))
2220	.collect();
2221
2222	let mut carried: Vec<serde_json::Value> = Vec::with_capacity(snapshot.len());
2223	let mut called: std::collections::HashSet<String> = std::collections::HashSet::new();
2224
2225	for mut message in snapshot {
2226	match message["role"].as_str().unwrap_or_default() {
2227	"system" => continue,
2228	"assistant" => {
2229	if let Some(calls) = message["tool_calls"].as_array() {
2230	let kept: Vec<serde_json::Value> = calls
2231	.iter()
2232	.filter(\|call\| call["id"].as_str().is_some_and(\|id\| answered.contains(id)))
2233	.cloned()
2234	.collect();
2235	if kept.is_empty() {
2236	// Only unanswered requests: keep whatever text came
2237	// with them, drop the message if there was none.
2238	if let Some(object) = message.as_object_mut() {
2239	object.remove("tool_calls");
2240	}
2241	if message_text(&message).is_empty() {
2242	continue;
2243	}
2244	} else {
2245	for call in &kept {
2246	called.insert(call["id"].as_str().unwrap_or_default().to_string());
2247	}
2248	message["tool_calls"] = serde_json::Value::Array(kept);
2249	}
2250	}
2251	carried.push(message);
2252	}
2253	"tool" => {
2254	let answers_a_kept_call = message["tool_call_id"]
2255	.as_str()
2256	.is_some_and(\|id\| called.contains(id));
2257	if answers_a_kept_call {
2258	carried.push(message);
2259	}
2260	}
2261	_ => carried.push(message),
2262	}
2263	}
2264
2265	carried
2266	}
2267
2268	/// Take a cancelled prompt's user message back out of the live history.
2269	///
2270	/// The client drops a cancelled prompt from its thread, so the model must not
2271	/// keep answering it. Only the first inference round can cancel with the user
2272	/// message as the last entry; later rounds end on tool or assistant output.
2273	/// A prompt that fails with an error is not taken back: the client keeps it,
2274	/// so the history keeps it too and a model switch carries it over.
2275	pub async fn forget_trailing_user_message(backend: &dyn InferenceBackend) {
2276	let mut history = backend.history_snapshot().await;
2277	if history
2278	.last()
2279	.is_some_and(\|message\| message["role"] == "user")
2280	{
2281	history.pop();
2282	backend.restore_history(history).await;
2283	}
2284	}
2285
2286	/// Replay `carried` (from [`carryover_history`]) into `backend`, on top of the
2287	/// system messages `backend` seeded for itself. Used when a model switch
2288	/// installs a new backend — or reloads the on-device engine, which wipes its
2289	/// history — so the thread continues under the new model instead of restarting.
2290	pub async fn adopt_carryover(backend: &dyn InferenceBackend, carried: Vec<serde_json::Value>) {
2291	if carried.is_empty() {
2292	return;
2293	}
2294	let mut rebuilt: Vec<serde_json::Value> = backend
2295	.history_snapshot()
2296	.await
2297	.into_iter()
2298	.take_while(\|message\| message["role"] == "system")
2299	.collect();
2300	rebuilt.extend(carried);
2301	backend.restore_history(rebuilt).await;
2302	}
2303
2304	#[cfg(test)]
2305	mod tests {
2306	use super::*;
2307
2308	/// The case from the issue: the endpoint says something a person can act
2309	/// on, and the status code and JSON around it are noise.
2310	#[test]
2311	fn an_endpoints_own_message_is_what_the_user_sees() {
2312	let body = r#"{"error":{"message":"Monthly siGit Code Cloud allowance reached. It resets at the start of your next billing period.","type":"server_error"}}"#;
2313
2314	let described = describe_api_error(reqwest::StatusCode::TOO_MANY_REQUESTS, body);
2315
2316	assert_eq!(
2317	described,
2318	"Monthly siGit Code Cloud allowance reached. \
2319	It resets at the start of your next billing period."
2320	);
2321	}
2322
2323	#[test]
2324	fn a_body_that_is_not_an_error_envelope_keeps_the_status() {
2325	let described = describe_api_error(reqwest::StatusCode::BAD_GATEWAY, "upstream said no");
2326
2327	assert!(described.contains("502"), "{described}");
2328	assert!(described.contains("upstream said no"), "{described}");
2329	}
2330
2331	/// The issue-#125 screenshots: a gateway's HTML error page pasted into
2332	/// chat tells the person nothing the status doesn't already say.
2333	#[test]
2334	fn an_html_error_page_is_not_pasted_into_the_message() {
2335	let page = "<!doctype html>\n<html lang=\"en\">\n<head>\n\
2336	<title>We're sorry, but something went wrong (500 Internal Server Error)</title>\n\
2337	</head>\n<body>…</body>\n</html>";
2338
2339	let described = describe_api_error(reqwest::StatusCode::GATEWAY_TIMEOUT, page);
2340
2341	assert_eq!(described, "endpoint returned 504 Gateway Timeout");
2342	}
2343
2344	#[test]
2345	fn an_empty_body_still_says_something() {
2346	let described = describe_api_error(reqwest::StatusCode::INTERNAL_SERVER_ERROR, " ");
2347
2348	assert_eq!(described, "endpoint returned 500 Internal Server Error");
2349	}
2350
2351	#[test]
2352	fn an_oversized_body_is_cut_down() {
2353	let body = "x".repeat(ERROR_BODY_LIMIT * 3);
2354
2355	let described = describe_api_error(reqwest::StatusCode::BAD_GATEWAY, &body);
2356
2357	assert!(described.len() < ERROR_BODY_LIMIT + 100, "{described}");
2358	assert!(described.ends_with('…'), "{described}");
2359	}
2360
2361	/// An envelope with nothing useful in it must not shadow the status, which
2362	/// would leave the user with an empty error.
2363	#[test]
2364	fn an_envelope_with_a_blank_message_falls_back() {
2365	let described = describe_api_error(
2366	reqwest::StatusCode::FORBIDDEN,
2367	r#"{"error":{"message":" "}}"#,
2368	);
2369
2370	assert!(described.contains("403"), "{described}");
2371	}
2372
2373	#[test]
2374	fn tools_json_wraps_function_schema() {
2375	let tools = vec![ToolSpec {
2376	name: "read_file".to_string(),
2377	description: "Read a file".to_string(),
2378	parameters_schema: r#"{"type":"object","properties":{"path":{"type":"string"}}}"#
2379	.to_string(),
2380	}];
2381	let json = OpenAiBackend::tools_json(&tools);
2382	assert_eq!(json[0]["type"], "function");
2383	assert_eq!(json[0]["function"]["name"], "read_file");
2384	assert_eq!(
2385	json[0]["function"]["parameters"]["properties"]["path"]["type"],
2386	"string"
2387	);
2388	}
2389
2390	#[test]
2391	fn malformed_schema_falls_back_to_empty_object() {
2392	let tools = vec![ToolSpec {
2393	name: "x".to_string(),
2394	description: String::new(),
2395	parameters_schema: "not json".to_string(),
2396	}];
2397	let json = OpenAiBackend::tools_json(&tools);
2398	assert_eq!(json[0]["function"]["parameters"]["type"], "object");
2399	}
2400
2401	#[test]
2402	fn streamed_assistant_history_omits_empty_tool_calls() {
2403	let value = streamed_assistant_history("hello", &[]);
2404	assert_eq!(value["role"], "assistant");
2405	assert_eq!(value["content"], "hello");
2406	assert!(value.get("tool_calls").is_none());
2407	}
2408
2409	#[test]
2410	fn streamed_assistant_history_preserves_tool_calls() {
2411	let calls = vec![ToolCall {
2412	id: "call_0".to_string(),
2413	name: "read_file".to_string(),
2414	arguments: r#"{"path":"a.rs"}"#.to_string(),
2415	}];
2416	let value = streamed_assistant_history("", &calls);
2417	assert!(value["content"].is_null());
2418	assert_eq!(value["tool_calls"][0]["id"], "call_0");
2419	assert_eq!(value["tool_calls"][0]["type"], "function");
2420	assert_eq!(value["tool_calls"][0]["function"]["name"], "read_file");
2421	assert_eq!(
2422	value["tool_calls"][0]["function"]["arguments"],
2423	r#"{"path":"a.rs"}"#
2424	);
2425	}
2426
2427	#[tokio::test]
2428	async fn cancelled_tool_results_close_out_history() {
2429	let backend = OpenAiBackend::new("http://localhost", "", "test-model", None);
2430	backend
2431	.history
2432	.lock()
2433	.await
2434	.push(streamed_assistant_history(
2435	"",
2436	&[ToolCall {
2437	id: "call_9".to_string(),
2438	name: "run_command".to_string(),
2439	arguments: r#"{"command":"ls"}"#.to_string(),
2440	}],
2441	));
2442
2443	backend
2444	.record_cancelled_tool_results(vec![ToolResult {
2445	tool_call_id: "call_9".to_string(),
2446	content: "cancelled by the user".to_string(),
2447	}])
2448	.await;
2449
2450	let history = backend.history.lock().await;
2451	let last = history.last().unwrap();
2452	assert_eq!(last["role"], "tool");
2453	assert_eq!(last["tool_call_id"], "call_9");
2454	assert_eq!(last["content"], "cancelled by the user");
2455
2456	drop(history);
2457	backend
2458	.record_cancelled_tool_results(vec![ToolResult {
2459	tool_call_id: "call_9".to_string(),
2460	content: "duplicate cleanup".to_string(),
2461	}])
2462	.await;
2463	let history = backend.history.lock().await;
2464	assert_eq!(
2465	history
2466	.iter()
2467	.filter(\|message\| message["tool_call_id"] == "call_9")
2468	.count(),
2469	1
2470	);
2471	}
2472
2473	#[test]
2474	fn carryover_drops_system_messages_and_keeps_the_turns() {
2475	let snapshot = vec![
2476	serde_json::json!({ "role": "system", "content": "old prompt" }),
2477	serde_json::json!({ "role": "user", "content": "hello" }),
2478	serde_json::json!({ "role": "assistant", "content": "hi" }),
2479	];
2480
2481	let carried = carryover_history(snapshot);
2482
2483	assert_eq!(carried.len(), 2);
2484	assert_eq!(carried[0]["role"], "user");
2485	assert_eq!(carried[1]["content"], "hi");
2486	}
2487
2488	#[test]
2489	fn carryover_keeps_a_trailing_user_message_from_a_failed_turn() {
2490	// The client keeps a prompt whose inference errored, so a switch must
2491	// not drop it (issue #124).
2492	let carried = carryover_history(vec![
2493	serde_json::json!({ "role": "system", "content": "prompt" }),
2494	serde_json::json!({ "role": "user", "content": "completed question" }),
2495	serde_json::json!({ "role": "assistant", "content": "completed answer" }),
2496	serde_json::json!({ "role": "user", "content": "failed question" }),
2497	]);
2498
2499	assert_eq!(carried.len(), 3, "{carried:#?}");
2500	assert_eq!(carried.last().unwrap()["content"], "failed question");
2501	}
2502
2503	#[tokio::test]
2504	async fn forget_trailing_user_message_drops_only_a_cancelled_prompt() {
2505	let backend = OpenAiBackend::new("http://localhost", "", "m", None);
2506	backend
2507	.restore_history(vec![
2508	serde_json::json!({ "role": "user", "content": "completed question" }),
2509	serde_json::json!({ "role": "assistant", "content": "completed answer" }),
2510	serde_json::json!({ "role": "user", "content": "cancelled question" }),
2511	])
2512	.await;
2513
2514	forget_trailing_user_message(&backend).await;
2515
2516	let history = backend.history_snapshot().await;
2517	assert_eq!(history.len(), 2, "{history:#?}");
2518	assert_eq!(history[1]["content"], "completed answer");
2519
2520	forget_trailing_user_message(&backend).await;
2521	assert_eq!(backend.history_snapshot().await.len(), 2);
2522	}
2523
2524	#[test]
2525	fn carryover_strips_a_tool_call_that_never_got_a_result() {
2526	// Switching mid-turn: the assistant asked for a tool, nothing answered.
2527	let snapshot = vec![
2528	serde_json::json!({ "role": "user", "content": "read a.rs" }),
2529	streamed_assistant_history(
2530	"on it",
2531	&[ToolCall {
2532	id: "call_1".to_string(),
2533	name: "read_file".to_string(),
2534	arguments: "{}".to_string(),
2535	}],
2536	),
2537	];
2538
2539	let carried = carryover_history(snapshot);
2540
2541	assert_eq!(carried.len(), 2);
2542	assert_eq!(carried[1]["content"], "on it");
2543	assert!(
2544	carried[1].get("tool_calls").is_none(),
2545	"an unanswered tool call must not survive the switch"
2546	);
2547	}
2548
2549	#[test]
2550	fn carryover_drops_a_textless_unanswered_tool_call_and_its_late_result() {
2551	let snapshot = vec![
2552	streamed_assistant_history(
2553	"",
2554	&[ToolCall {
2555	id: "call_1".to_string(),
2556	name: "read_file".to_string(),
2557	arguments: "{}".to_string(),
2558	}],
2559	),
2560	// An orphan: its assistant message is gone with the line above.
2561	serde_json::json!({
2562	"role": "tool",
2563	"tool_call_id": "call_other",
2564	"content": "file contents",
2565	}),
2566	];
2567
2568	assert!(carryover_history(snapshot).is_empty());
2569	}
2570
2571	#[test]
2572	fn carryover_keeps_an_answered_tool_call_paired_with_its_result() {
2573	let snapshot = vec![
2574	streamed_assistant_history(
2575	"",
2576	&[ToolCall {
2577	id: "call_1".to_string(),
2578	name: "read_file".to_string(),
2579	arguments: "{}".to_string(),
2580	}],
2581	),
2582	serde_json::json!({
2583	"role": "tool",
2584	"tool_call_id": "call_1",
2585	"content": "file contents",
2586	}),
2587	];
2588
2589	let carried = carryover_history(snapshot);
2590
2591	assert_eq!(carried.len(), 2);
2592	assert_eq!(carried[0]["tool_calls"][0]["id"], "call_1");
2593	assert_eq!(carried[1]["tool_call_id"], "call_1");
2594	}
2595
2596	#[tokio::test]
2597	async fn adopt_carryover_replays_the_thread_under_the_new_system_prompt() {
2598	let new_backend =
2599	OpenAiBackend::new("http://localhost", "", "m", Some("new prompt".into()));
2600
2601	let carried = carryover_history(vec![
2602	serde_json::json!({ "role": "system", "content": "old prompt" }),
2603	serde_json::json!({ "role": "user", "content": "hello" }),
2604	serde_json::json!({ "role": "assistant", "content": "hi" }),
2605	]);
2606	adopt_carryover(&new_backend, carried).await;
2607
2608	let history = new_backend.history_snapshot().await;
2609	assert_eq!(history.len(), 3);
2610	assert_eq!(history[0]["role"], "system");
2611	assert_eq!(history[0]["content"], "new prompt");
2612	assert_eq!(history[1]["content"], "hello");
2613	assert_eq!(history[2]["content"], "hi");
2614	}
2615
2616	#[tokio::test]
2617	async fn adopt_carryover_leaves_a_fresh_backend_alone() {
2618	let new_backend =
2619	OpenAiBackend::new("http://localhost", "", "m", Some("new prompt".into()));
2620
2621	adopt_carryover(&new_backend, Vec::new()).await;
2622
2623	let history = new_backend.history_snapshot().await;
2624	assert_eq!(history.len(), 1);
2625	assert_eq!(history[0]["content"], "new prompt");
2626	}
2627
2628	fn user_with_image(text: &str, payload_chars: usize) -> serde_json::Value {
2629	serde_json::json!({
2630	"role": "user",
2631	"content": [
2632	{ "type": "text", "text": text },
2633	{ "type": "image_url", "image_url": {
2634	"url": format!("data:image/png;base64,{}", "A".repeat(payload_chars)),
2635	}},
2636	],
2637	})
2638	}
2639
2640	#[test]
2641	fn user_message_uses_content_parts_only_when_there_is_an_image() {
2642	assert_eq!(
2643	user_message("hello", &[], &[]),
2644	serde_json::json!({ "role": "user", "content": "hello" })
2645	);
2646
2647	let image = ImageInput {
2648	mime_type: "image/png".to_string(),
2649	data: "AAAA".to_string(),
2650	};
2651	assert_eq!(
2652	user_message("what is this?", std::slice::from_ref(&image), &[]),
2653	serde_json::json!({
2654	"role": "user",
2655	"content": [
2656	{ "type": "text", "text": "what is this?" },
2657	{ "type": "image_url", "image_url": { "url": "data:image/png;base64,AAAA" } },
2658	],
2659	})
2660	);
2661	// An image with no text sends no empty text part.
2662	let only_image = user_message("", &[image], &[]);
2663	assert_eq!(only_image["content"].as_array().unwrap().len(), 1);
2664	}
2665
2666	#[test]
2667	fn audio_is_sent_as_an_input_audio_part() {
2668	let clip = AudioInput {
2669	mime_type: "audio/mpeg".to_string(),
2670	data: "BBBB".to_string(),
2671	};
2672	assert_eq!(
2673	user_message("transcribe this", &[], &[clip]),
2674	serde_json::json!({
2675	"role": "user",
2676	"content": [
2677	{ "type": "text", "text": "transcribe this" },
2678	{ "type": "input_audio", "input_audio": { "data": "BBBB", "format": "mp3" } },
2679	],
2680	})
2681	);
2682	}
2683
2684	#[test]
2685	fn audio_format_follows_the_mime_type() {
2686	let format = \|mime: &str\| {
2687	AudioInput {
2688	mime_type: mime.to_string(),
2689	data: String::new(),
2690	}
2691	.format()
2692	};
2693	assert_eq!(format("audio/wav"), "wav");
2694	assert_eq!(format("audio/x-wav"), "wav");
2695	assert_eq!(format("audio/wave"), "wav");
2696	assert_eq!(format("audio/mpeg"), "mp3");
2697	assert_eq!(format("Audio/MP3; codecs=mp3"), "mp3");
2698	assert_eq!(format("audio/flac"), "flac");
2699	}
2700
2701	#[test]
2702	fn a_model_without_audio_gets_a_note_and_keeps_the_image() {
2703	let history = vec![serde_json::json!({
2704	"role": "user",
2705	"content": [
2706	{ "type": "text", "text": "what is this?" },
2707	{ "type": "image_url", "image_url": { "url": "data:image/png;base64,AAAA" } },
2708	{ "type": "input_audio", "input_audio": { "data": "BBBB", "format": "wav" } },
2709	],
2710	})];
2711
2712	let sent = without_audio(&history);
2713	assert_eq!(
2714	sent[0]["content"],
2715	serde_json::json!([
2716	{ "type": "text", "text": "what is this?" },
2717	{ "type": "image_url", "image_url": { "url": "data:image/png;base64,AAAA" } },
2718	{ "type": "text", "text": AUDIO_OMITTED_NOTE },
2719	])
2720	);
2721	// With neither, the message is plain text again.
2722	assert_eq!(
2723	without_images(&sent)[0]["content"],
2724	format!("what is this?\n{AUDIO_OMITTED_NOTE}\n{IMAGE_OMITTED_NOTE}")
2725	);
2726	// History keeps the clip for a model that can take it later.
2727	assert_eq!(message_audio_count(&history[0]), 1);
2728	}
2729
2730	#[test]
2731	fn a_text_only_model_is_sent_a_note_in_place_of_each_image() {
2732	let history = vec![
2733	serde_json::json!({ "role": "system", "content": "prompt" }),
2734	user_with_image("what is this?", 16),
2735	serde_json::json!({ "role": "assistant", "content": "a cat" }),
2736	];
2737	let sent = without_images(&history);
2738
2739	assert_eq!(sent[0], history[0]);
2740	assert_eq!(sent[2], history[2]);
2741	assert_eq!(
2742	sent[1]["content"],
2743	format!("what is this?\n{IMAGE_OMITTED_NOTE}")
2744	);
2745	// History keeps the image for a model that can read it later.
2746	assert_eq!(message_image_count(&history[1]), 1);
2747	}
2748
2749	#[test]
2750	fn a_remote_backend_reads_images_by_model_id() {
2751	let own_endpoint = OpenAiBackend::new("http://localhost", "", "gpt-4o-mini", None);
2752	assert!(own_endpoint.accepts_images());
2753	let text_tier = OpenAiBackend::new("http://localhost", "", "onde-nova", None);
2754	assert!(!text_tier.accepts_images());
2755	let image_tier = OpenAiBackend::new("http://localhost", "", "onde-large", None);
2756	assert!(image_tier.accepts_images());
2757	}
2758
2759	#[test]
2760	fn a_remote_backend_takes_audio_only_on_the_users_own_endpoint() {
2761	let own_endpoint = OpenAiBackend::new("http://localhost", "", "gpt-4o-audio", None);
2762	assert!(own_endpoint.accepts_audio());
2763	let cloud_tier = OpenAiBackend::new("http://localhost", "", "onde-large", None);
2764	assert!(!cloud_tier.accepts_audio());
2765	}
2766
2767	#[test]
2768	fn message_text_reads_both_content_shapes() {
2769	assert_eq!(
2770	message_text(&serde_json::json!({ "role": "user", "content": "hello" })),
2771	"hello"
2772	);
2773	let with_image = user_with_image("what is this?", 16);
2774	assert_eq!(message_text(&with_image), "what is this?");
2775	assert_eq!(message_image_count(&with_image), 1);
2776	// A tool-call-only assistant message has no content at all.
2777	let no_content = serde_json::json!({ "role": "assistant", "content": null });
2778	assert_eq!(message_text(&no_content), "");
2779	assert_eq!(message_image_count(&no_content), 0);
2780	}
2781
2782	#[test]
2783	fn estimate_tokens_does_not_count_an_image_payload_as_text() {
2784	// Two megabytes of base64 would read as half a million tokens.
2785	let history = vec![user_with_image("what is this?", 2_000_000)];
2786	let estimate = estimate_tokens(&history);
2787	assert!(
2788	(IMAGE_TOKEN_ESTIMATE..IMAGE_TOKEN_ESTIMATE + 100).contains(&estimate),
2789	"{estimate}"
2790	);
2791	}
2792
2793	#[test]
2794	fn summary_transcript_keeps_the_text_of_a_message_with_an_image() {
2795	let transcript = transcript_for_summary(&[user_with_image("what is this?", 64)]);
2796	assert!(transcript.contains("what is this?"), "{transcript}");
2797	assert!(transcript.contains("[attached an image]"), "{transcript}");
2798	assert!(!transcript.contains("base64"), "{transcript}");
2799	}
2800
2801	#[test]
2802	fn carryover_keeps_a_user_message_that_carries_an_image() {
2803	let carried = carryover_history(vec![
2804	serde_json::json!({ "role": "system", "content": "prompt" }),
2805	user_with_image("what is this?", 16),
2806	serde_json::json!({ "role": "assistant", "content": "a cat" }),
2807	]);
2808	assert_eq!(carried.len(), 2);
2809	assert_eq!(message_image_count(&carried[0]), 1);
2810	}
2811
2812	#[test]
2813	fn estimate_tokens_scales_with_serialized_size() {
2814	assert_eq!(estimate_tokens(&[]), 0);
2815
2816	let short = vec![serde_json::json!({ "role": "user", "content": "hi" })];
2817	let long = vec![serde_json::json!({ "role": "user", "content": "x".repeat(4_000) })];
2818	let short_estimate = estimate_tokens(&short);
2819	let long_estimate = estimate_tokens(&long);
2820
2821	assert!(short_estimate > 0, "non-empty history estimates > 0 tokens");
2822	assert!(long_estimate > short_estimate, "longer history costs more");
2823	// 4,000 content chars / 4 ≈ 1,000 tokens, plus a little JSON framing.
2824	assert!((1_000..1_100).contains(&long_estimate), "{long_estimate}");
2825	}
2826
2827	#[tokio::test]
2828	async fn openai_snapshot_restore_round_trips_exactly() {
2829	let backend = OpenAiBackend::new("http://localhost", "", "m", Some("be helpful".into()));
2830	{
2831	let mut history = backend.history.lock().await;
2832	history.push(serde_json::json!({ "role": "user", "content": "hello" }));
2833	history.push(streamed_assistant_history(
2834	"",
2835	&[ToolCall {
2836	id: "call_1".to_string(),
2837	name: "read_file".to_string(),
2838	arguments: r#"{"path":"a.rs"}"#.to_string(),
2839	}],
2840	));
2841	history.push(serde_json::json!({
2842	"role": "tool", "tool_call_id": "call_1", "content": "fn main() {}",
2843	}));
2844	history.push(serde_json::json!({ "role": "assistant", "content": "done" }));
2845	}
2846	let snapshot = backend.history_snapshot().await;
2847	assert_eq!(
2848	snapshot[0]["role"], "system",
2849	"snapshot keeps the system message"
2850	);
2851
2852	// Restoring into a backend seeded with a different system prompt must
2853	// replace everything, including that seed.
2854	let restored = OpenAiBackend::new("http://localhost", "", "m", Some("other seed".into()));
2855	restored.restore_history(snapshot.clone()).await;
2856	assert_eq!(restored.history_snapshot().await, snapshot);
2857	}
2858
2859	#[tokio::test]
2860	async fn local_restore_replaces_engine_history_and_system_context() {
2861	let engine = Arc::new(ChatEngine::new());
2862	let backend = LocalBackend::new(Arc::clone(&engine));
2863	backend
2864	.restore_history(vec![
2865	serde_json::json!({ "role": "system", "content": "thread A context" }),
2866	serde_json::json!({ "role": "user", "content": "thread A question" }),
2867	])
2868	.await;
2869
2870	backend
2871	.restore_history(vec![
2872	serde_json::json!({ "role": "system", "content": "thread B context" }),
2873	serde_json::json!({ "role": "user", "content": "thread B question" }),
2874	])
2875	.await;
2876
2877	let restored = backend.history_snapshot().await;
2878	assert_eq!(restored.len(), 2, "{restored:#?}");
2879	assert_eq!(restored[0]["content"], "thread B context");
2880	assert_eq!(restored[1]["content"], "thread B question");
2881	assert!(
2882	restored.iter().all(\|message\| !message["content"]
2883	.as_str()
2884	.unwrap_or_default()
2885	.contains("thread A")),
2886	"restore_history must replace, not append to, the local engine: {restored:#?}"
2887	);
2888
2889	let error = backend
2890	.send_message_with_tools("must not run", &[], None)
2891	.await
2892	.unwrap_err();
2893	assert!(error.contains("not ready"), "{error}");
2894	assert_eq!(backend.history_snapshot().await, restored);
2895	}
2896
2897	/// Minimal scripted OpenAI-compatible endpoint: accepts one HTTP request on
2898	/// a std listener and answers with a fixed non-streaming completion. The
2899	/// receiver yields the request body the backend actually put on the wire.
2900	fn spawn_completion_stub(
2901	summary: &str,
2902	) -> (std::net::SocketAddr, std::sync::mpsc::Receiver<String>) {
2903	spawn_message_stub(serde_json::json!({ "role": "assistant", "content": summary }))
2904	}
2905
2906	/// Like [`spawn_completion_stub`], answering with an arbitrary message.
2907	fn spawn_message_stub(
2908	message: serde_json::Value,
2909	) -> (std::net::SocketAddr, std::sync::mpsc::Receiver<String>) {
2910	spawn_choice_stub(serde_json::json!({ "message": message }))
2911	}
2912
2913	/// Like [`spawn_message_stub`], answering with a whole choice, so a test
2914	/// can set what sits beside the message.
2915	fn spawn_choice_stub(
2916	choice: serde_json::Value,
2917	) -> (std::net::SocketAddr, std::sync::mpsc::Receiver<String>) {
2918	use std::io::{Read, Write};
2919
2920	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
2921	let addr = listener.local_addr().unwrap();
2922	let (sender, receiver) = std::sync::mpsc::channel();
2923	let body = serde_json::json!({ "choices": [choice] }).to_string();
2924	std::thread::spawn(move \|\| {
2925	let (mut stream, _) = listener.accept().unwrap();
2926	// Read until the full request (headers + content-length body) is in.
2927	let mut request = Vec::new();
2928	let mut chunk = [0u8; 4096];
2929	loop {
2930	let n = stream.read(&mut chunk).unwrap_or(0);
2931	if n == 0 {
2932	break;
2933	}
2934	request.extend_from_slice(&chunk[..n]);
2935	if let Some(headers_end) =
2936	request.windows(4).position(\|window\| window == b"\r\n\r\n")
2937	{
2938	let headers = String::from_utf8_lossy(&request[..headers_end]);
2939	let content_length = headers
2940	.lines()
2941	.find_map(\|line\| {
2942	line.to_ascii_lowercase()
2943	.strip_prefix("content-length:")
2944	.map(\|value\| value.trim().parse::<usize>().unwrap_or(0))
2945	})
2946	.unwrap_or(0);
2947	if request.len() >= headers_end + 4 + content_length {
2948	let _ = sender.send(
2949	String::from_utf8_lossy(&request[headers_end + 4..]).into_owned(),
2950	);
2951	break;
2952	}
2953	}
2954	}
2955	let response = format!(
2956	"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
2957	content-length: {}\r\nconnection: close\r\n\r\n{}",
2958	body.len(),
2959	body
2960	);
2961	let _ = stream.write_all(response.as_bytes());
2962	});
2963	(addr, receiver)
2964	}
2965
2966	/// Scripted SSE endpoint: one response per connection, in order. The
2967	/// receiver yields each request body.
2968	fn spawn_sse_sequence_stub(
2969	replies: Vec<Vec<String>>,
2970	) -> (std::net::SocketAddr, std::sync::mpsc::Receiver<String>) {
2971	use std::io::{Read, Write};
2972
2973	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
2974	let addr = listener.local_addr().unwrap();
2975	let (sender, receiver) = std::sync::mpsc::channel();
2976	std::thread::spawn(move \|\| {
2977	for deltas in replies {
2978	let (mut stream, _) = listener.accept().unwrap();
2979	let mut request = Vec::new();
2980	let mut buf = [0u8; 4096];
2981	loop {
2982	let n = stream.read(&mut buf).unwrap_or(0);
2983	if n == 0 {
2984	break;
2985	}
2986	request.extend_from_slice(&buf[..n]);
2987	if let Some(end) = request.windows(4).position(\|w\| w == b"\r\n\r\n") {
2988	let headers = String::from_utf8_lossy(&request[..end]).to_ascii_lowercase();
2989	let length = headers
2990	.lines()
2991	.find_map(\|l\| l.strip_prefix("content-length:")?.trim().parse().ok())
2992	.unwrap_or(0usize);
2993	if request.len() >= end + 4 + length {
2994	let _ = sender
2995	.send(String::from_utf8_lossy(&request[end + 4..]).into_owned());
2996	break;
2997	}
2998	}
2999	}
3000	let mut body = String::new();
3001	for delta in deltas {
3002	let frame = serde_json::json!({
3003	"choices": [{ "delta": { "content": delta } }]
3004	});
3005	body.push_str(&format!("data: {frame}\n\n"));
3006	}
3007	body.push_str("data: [DONE]\n\n");
3008	let response = format!(
3009	"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\n\
3010	content-length: {}\r\nconnection: close\r\n\r\n{body}",
3011	body.len()
3012	);
3013	let _ = stream.write_all(response.as_bytes());
3014	}
3015	});
3016	(addr, receiver)
3017	}
3018
3019	/// Issue #123: a streamed reply that degenerates into a repetition loop is
3020	/// cut off, history keeps only what came before the loop, and the model
3021	/// gets one retry.
3022	#[tokio::test]
3023	async fn a_repetition_loop_is_cut_and_retried_once() {
3024	let intro = "Let me look at the commits. ";
3025	let looped: Vec<String> = std::iter::once(intro.to_string())
3026	.chain(
3027	crate::repetition::issue_loop_text()
3028	.as_bytes()
3029	.chunks(40)
3030	.map(\|c\| String::from_utf8_lossy(c).into_owned()),
3031	)
3032	.collect();
3033	let (addr, requests) =
3034	spawn_sse_sequence_stub(vec![looped, vec!["Here is the answer.".to_string()]]);
3035	let backend =
3036	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3037	let (sink, mut tokens) = tokio::sync::mpsc::unbounded_channel();
3038	let tools = vec![ToolSpec {
3039	name: "run_command".to_string(),
3040	description: "Run a command".to_string(),
3041	parameters_schema: r#"{"type":"object","properties":{}}"#.to_string(),
3042	}];
3043
3044	let result = backend
3045	.send_message_with_tools("what changed", &tools, Some(&sink))
3046	.await
3047	.unwrap();
3048
3049	assert_eq!(result.finish, FinishReason::Complete);
3050	assert!(result.text.starts_with(intro.trim_end()), "{}", result.text);
3051	assert!(result.text.ends_with("Here is the answer."));
3052	assert!(
3053	result.text.len() < 600,
3054	"junk kept: {} bytes",
3055	result.text.len()
3056	);
3057
3058	let first = requests.recv().unwrap();
3059	assert!(!first.contains(REPETITION_RETRY));
3060	let second = requests.recv().unwrap();
3061	assert!(second.contains(REPETITION_RETRY));
3062
3063	let history = backend.history_snapshot().await;
3064	let kept = history
3065	.iter()
3066	.find(\|m\| m["role"] == "assistant")
3067	.and_then(\|m\| m["content"].as_str())
3068	.unwrap();
3069	assert!(
3070	kept.len() < 600,
3071	"history kept the loop: {} bytes",
3072	kept.len()
3073	);
3074
3075	drop(sink);
3076	let mut shown = String::new();
3077	while let Ok(chunk) = tokens.try_recv() {
3078	if let TokenChunk::Visible(text) = chunk {
3079	shown.push_str(&text);
3080	}
3081	}
3082	assert!(shown.contains("stopped this reply"), "{shown}");
3083	}
3084
3085	/// Serves `responses` in order, one per connection, as `(status,
3086	/// content type, body)`, and hands back each request body.
3087	fn spawn_scripted_stub(
3088	responses: Vec<(u16, &'static str, String)>,
3089	) -> (std::net::SocketAddr, std::sync::mpsc::Receiver<String>) {
3090	use std::io::{Read, Write};
3091
3092	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
3093	let addr = listener.local_addr().unwrap();
3094	let (sender, receiver) = std::sync::mpsc::channel();
3095	std::thread::spawn(move \|\| {
3096	for (status, content_type, body) in responses {
3097	let Ok((mut stream, _)) = listener.accept() else {
3098	return;
3099	};
3100	let mut request = Vec::new();
3101	let mut chunk = [0u8; 4096];
3102	loop {
3103	let n = stream.read(&mut chunk).unwrap_or(0);
3104	if n == 0 {
3105	break;
3106	}
3107	request.extend_from_slice(&chunk[..n]);
3108	if let Some(headers_end) =
3109	request.windows(4).position(\|window\| window == b"\r\n\r\n")
3110	{
3111	let headers = String::from_utf8_lossy(&request[..headers_end]);
3112	let content_length = headers
3113	.lines()
3114	.find_map(\|line\| {
3115	line.to_ascii_lowercase()
3116	.strip_prefix("content-length:")
3117	.map(\|value\| value.trim().parse::<usize>().unwrap_or(0))
3118	})
3119	.unwrap_or(0);
3120	if request.len() >= headers_end + 4 + content_length {
3121	let _ = sender.send(
3122	String::from_utf8_lossy(&request[headers_end + 4..]).into_owned(),
3123	);
3124	break;
3125	}
3126	}
3127	}
3128	let response = format!(
3129	"HTTP/1.1 {status} X\r\ncontent-type: {content_type}\r\n\
3130	content-length: {}\r\nconnection: close\r\n\r\n{}",
3131	body.len(),
3132	body
3133	);
3134	let _ = stream.write_all(response.as_bytes());
3135	}
3136	});
3137	(addr, receiver)
3138	}
3139
3140	fn sse(events: &[serde_json::Value]) -> String {
3141	let mut body: String = events
3142	.iter()
3143	.map(\|event\| format!("data: {event}\n\n"))
3144	.collect();
3145	body.push_str("data: [DONE]\n\n");
3146	body
3147	}
3148
3149	/// A stream asks for token usage and reads it off the last chunk, the
3150	/// one with no choices, for the client's context meter (issue #194).
3151	#[tokio::test]
3152	async fn a_stream_reports_the_usage_it_was_asked_for() {
3153	let (addr, requests) = spawn_scripted_stub(vec![(
3154	200,
3155	"text/event-stream",
3156	sse(&[
3157	serde_json::json!({"choices": [{"delta": {"content": "Hi."}}]}),
3158	serde_json::json!({
3159	"choices": [],
3160	"usage": {"prompt_tokens": 1200, "completion_tokens": 34, "total_tokens": 1234},
3161	}),
3162	]),
3163	)]);
3164	let backend =
3165	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3166	let (sink, _chunks) = tokio::sync::mpsc::unbounded_channel();
3167
3168	let result = backend
3169	.send_message_with_tools("hello", &[], Some(&sink))
3170	.await
3171	.unwrap();
3172
3173	assert_eq!(result.text, "Hi.");
3174	assert_eq!(result.context_tokens, Some(1234));
3175	let request: serde_json::Value = serde_json::from_str(&requests.recv().unwrap()).unwrap();
3176	assert_eq!(request["stream_options"]["include_usage"], true);
3177	}
3178
3179	#[tokio::test]
3180	async fn a_json_reply_reports_its_usage() {
3181	let body = serde_json::json!({
3182	"choices": [{"message": {"role": "assistant", "content": "Hi."}}],
3183	"usage": {"prompt_tokens": 90, "completion_tokens": 10},
3184	})
3185	.to_string();
3186	let (addr, requests) = spawn_scripted_stub(vec![(200, "application/json", body)]);
3187	let backend =
3188	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3189
3190	let result = backend
3191	.send_message_with_tools("hello", &[], None)
3192	.await
3193	.unwrap();
3194
3195	// No total given: prompt plus completion.
3196	assert_eq!(result.context_tokens, Some(100));
3197	let request: serde_json::Value = serde_json::from_str(&requests.recv().unwrap()).unwrap();
3198	assert!(request.get("stream_options").is_none(), "{request}");
3199	}
3200
3201	/// An endpoint that predates `stream_options` is asked again without it,
3202	/// and not asked for usage again after that.
3203	#[tokio::test]
3204	async fn an_endpoint_that_refuses_stream_options_still_streams() {
3205	let refusal = serde_json::json!({
3206	"error": {"message": "Unrecognized request argument supplied: stream_options"},
3207	})
3208	.to_string();
3209	let reply = sse(&[serde_json::json!({"choices": [{"delta": {"content": "Hi."}}]})]);
3210	let (addr, requests) = spawn_scripted_stub(vec![
3211	(400, "application/json", refusal),
3212	(200, "text/event-stream", reply.clone()),
3213	(200, "text/event-stream", reply),
3214	]);
3215	let backend =
3216	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3217	let (sink, _chunks) = tokio::sync::mpsc::unbounded_channel();
3218
3219	let first = backend
3220	.send_message_with_tools("hello", &[], Some(&sink))
3221	.await
3222	.unwrap();
3223	assert_eq!(first.text, "Hi.");
3224	assert_eq!(first.context_tokens, None);
3225	let second = backend
3226	.send_message_with_tools("again", &[], Some(&sink))
3227	.await
3228	.unwrap();
3229	assert_eq!(second.text, "Hi.");
3230
3231	let bodies: Vec<serde_json::Value> = requests
3232	.try_iter()
3233	.map(\|body\| serde_json::from_str(&body).unwrap())
3234	.collect();
3235	assert_eq!(bodies.len(), 3);
3236	assert!(bodies[0].get("stream_options").is_some());
3237	assert!(bodies[1].get("stream_options").is_none());
3238	assert!(bodies[2].get("stream_options").is_none());
3239	}
3240
3241	#[test]
3242	fn finish_reasons_map_from_the_wire() {
3243	for reason in [None, Some("stop"), Some("tool_calls"), Some("unheard-of")] {
3244	assert_eq!(FinishReason::from_wire(reason), FinishReason::Complete);
3245	}
3246	assert_eq!(
3247	FinishReason::from_wire(Some("length")),
3248	FinishReason::Length
3249	);
3250	assert_eq!(
3251	FinishReason::from_wire(Some("content_filter")),
3252	FinishReason::ContentFilter
3253	);
3254	}
3255
3256	/// The non-streaming path (headless runs, subagents) reads the reason off
3257	/// the choice; a reply cut off at the token limit must not look finished.
3258	#[tokio::test]
3259	async fn a_truncated_json_reply_reports_length() {
3260	let (addr, _requests) = spawn_choice_stub(serde_json::json!({
3261	"message": { "role": "assistant", "content": "The answer is" },
3262	"finish_reason": "length",
3263	}));
3264	let backend =
3265	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3266
3267	let result = backend
3268	.send_message_with_tools("explain", &[], None)
3269	.await
3270	.unwrap();
3271
3272	assert_eq!(result.text, "The answer is");
3273	assert_eq!(result.finish, FinishReason::Length);
3274	}
3275
3276	/// A structured call can come back with a broken inline block in the same
3277	/// message. The call runs; the block reaches neither the reply nor the
3278	/// history, where the model would read back a call that never ran.
3279	#[tokio::test]
3280	async fn a_broken_inline_block_beside_a_structured_call_is_dropped() {
3281	let (addr, _requests) = spawn_message_stub(serde_json::json!({
3282	"role": "assistant",
3283	"content": "Checking.<tool_call>run_command CheckStatus=nope</tool_call>",
3284	"tool_calls": [{
3285	"id": "call_1",
3286	"type": "function",
3287	"function": { "name": "run_command", "arguments": "{\"command\":\"pwd\"}" },
3288	}],
3289	}));
3290	let backend =
3291	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3292	let tools = vec![ToolSpec {
3293	name: "run_command".to_string(),
3294	description: "Run a command".to_string(),
3295	parameters_schema: r#"{"type":"object","properties":{"command":{"type":"string"}}}"#
3296	.to_string(),
3297	}];
3298
3299	let result = backend
3300	.send_message_with_tools("where am I", &tools, None)
3301	.await
3302	.unwrap();
3303
3304	assert_eq!(result.text, "Checking.");
3305	assert_eq!(result.tool_calls.len(), 1);
3306	assert_eq!(result.tool_calls[0].id, "call_1");
3307	let history = backend.history_snapshot().await;
3308	let reply = history.last().unwrap();
3309	assert_eq!(reply["content"], "Checking.");
3310	assert_eq!(reply["tool_calls"][0]["id"], "call_1");
3311	}
3312
3313	/// A structured call plus a well-formed inline block: both run, and the
3314	/// inline markup leaves the reply and history, as on the streaming path.
3315	#[tokio::test]
3316	async fn a_well_formed_inline_block_beside_a_structured_call_also_runs() {
3317	let (addr, _requests) = spawn_message_stub(serde_json::json!({
3318	"role": "assistant",
3319	"content": "Checking.<tool_call>run_command<arg_key>command</arg_key><arg_value>ls</arg_value></tool_call>",
3320	"tool_calls": [{
3321	"id": "call_1",
3322	"type": "function",
3323	"function": { "name": "run_command", "arguments": "{\"command\":\"pwd\"}" },
3324	}],
3325	}));
3326	let backend =
3327	OpenAiBackend::new(format!("http://{addr}/v1"), "test-key", "test-model", None);
3328	let tools = vec![ToolSpec {
3329	name: "run_command".to_string(),
3330	description: "Run a command".to_string(),
3331	parameters_schema: r#"{"type":"object","properties":{"command":{"type":"string"}}}"#
3332	.to_string(),
3333	}];
3334
3335	let result = backend
3336	.send_message_with_tools("where am I", &tools, None)
3337	.await
3338	.unwrap();
3339
3340	assert_eq!(result.text, "Checking.");
3341	let ids: Vec<&str> = result.tool_calls.iter().map(\|c\| c.id.as_str()).collect();
3342	assert_eq!(ids, ["call_1", "call_recovered_1"]);
3343	assert_eq!(result.tool_calls[1].arguments, r#"{"command":"ls"}"#);
3344	let history = backend.history_snapshot().await;
3345	let reply = history.last().unwrap();
3346	assert_eq!(reply["content"], "Checking.");
3347	assert_eq!(reply["tool_calls"].as_array().unwrap().len(), 2);
3348	}
3349
3350	#[tokio::test]
3351	async fn compact_history_rebuilds_system_summary_and_tail() {
3352	let (addr, _requests) = spawn_completion_stub("We refactored backend.rs; tests pass.");
3353	let backend = OpenAiBackend::new(
3354	format!("http://{addr}/v1"),
3355	"test-key",
3356	"test-model",
3357	Some("be helpful".into()),
3358	);
3359	{
3360	let mut history = backend.history.lock().await;
3361	for i in 0..5 {
3362	let role = if i % 2 == 0 { "user" } else { "assistant" };
3363	history.push(serde_json::json!({
3364	"role": role, "content": format!("message {i}"),
3365	}));
3366	}
3367	}
3368
3369	backend.compact_history(2).await.unwrap();
3370
3371	let history = backend.history_snapshot().await;
3372	assert_eq!(history.len(), 4, "system + summary + last 2: {history:?}");
3373	assert_eq!(history[0]["role"], "system");
3374	assert_eq!(history[0]["content"], "be helpful");
3375	assert_eq!(history[1]["role"], "user");
3376	let summary_text = history[1]["content"].as_str().unwrap();
3377	assert!(summary_text.starts_with("[Conversation summary]\n"));
3378	assert!(summary_text.contains("We refactored backend.rs; tests pass."));
3379	assert_eq!(
3380	history[2],
3381	serde_json::json!({ "role": "assistant", "content": "message 3" })
3382	);
3383	assert_eq!(
3384	history[3],
3385	serde_json::json!({ "role": "user", "content": "message 4" })
3386	);
3387	}
3388
3389	/// Compacting a tool-heavy session must not put tool shapes on the wire.
3390	/// The summarization round offers no `tools`, and endpoints reject tool
3391	/// calls and tool results that arrive without a schema — which used to make
3392	/// compaction fail forever in any session that had run a single tool.
3393	#[tokio::test]
3394	async fn compact_history_sends_no_tool_artifacts() {
3395	let (addr, requests) = spawn_completion_stub("Ran git status on main.");
3396	let backend = OpenAiBackend::new(
3397	format!("http://{addr}/v1"),
3398	"test-key",
3399	"test-model",
3400	Some("be helpful".into()),
3401	);
3402	{
3403	let mut history = backend.history.lock().await;
3404	history.push(serde_json::json!({ "role": "user", "content": "check the repo" }));
3405	history.push(serde_json::json!({
3406	"role": "assistant",
3407	"content": null,
3408	"tool_calls": [{
3409	"id": "call_1",
3410	"type": "function",
3411	"function": { "name": "run_command", "arguments": "{\"command\":\"git status\"}" },
3412	}],
3413	}));
3414	history.push(serde_json::json!({
3415	"role": "tool",
3416	"tool_call_id": "call_1",
3417	"content": "on branch main",
3418	}));
3419	}
3420
3421	backend.compact_history(2).await.unwrap();
3422
3423	let body: serde_json::Value =
3424	serde_json::from_str(&requests.recv().unwrap()).expect("request body is JSON");
3425	assert!(
3426	body.get("tools").is_none(),
3427	"summarization offers no tools: {body}"
3428	);
3429	for message in body["messages"].as_array().unwrap() {
3430	assert!(
3431	message.get("tool_calls").is_none(),
3432	"no tool_calls may be sent without a schema: {message}"
3433	);
3434	assert_ne!(
3435	message["role"], "tool",
3436	"no tool results may be sent without a schema: {message}"
3437	);
3438	}
3439
3440	// The tool round still has to survive into the summary request as prose,
3441	// or the summary loses the work the session actually did.
3442	let transcript = body["messages"].as_array().unwrap().last().unwrap()["content"]
3443	.as_str()
3444	.unwrap();
3445	assert!(transcript.contains("called run_command({\"command\":\"git status\"})"));
3446	assert!(transcript.contains("tool result: on branch main"));
3447	}
3448
3449	/// Compaction only fails when even the truncation fallback cannot fit the
3450	/// history in the budget — here the newest message alone is bigger than
3451	/// the whole window.
3452	#[tokio::test]
3453	async fn compact_history_failure_leaves_history_intact() {
3454	// No listener at this address, so summarization fails; the system
3455	// prompt alone is bigger than the window, so even the minimal
3456	// fallback rebuild cannot fit — the genuinely unrecoverable case.
3457	let backend = OpenAiBackend::new(
3458	"http://127.0.0.1:9",
3459	"",
3460	"test-model",
3461	Some("s".repeat(DEFAULT_CONTEXT_TOKEN_BUDGET * 4 + 1000)),
3462	);
3463	backend
3464	.history
3465	.lock()
3466	.await
3467	.push(serde_json::json!({ "role": "user", "content": "hello" }));
3468	let before = backend.history_snapshot().await;
3469
3470	assert!(backend.compact_history(2).await.is_err());
3471	assert_eq!(backend.history_snapshot().await, before);
3472	}
3473
3474	/// The issue-#125 path: the summarization endpoint is down, and the
3475	/// session must still shrink rather than die with "start a new thread".
3476	#[tokio::test]
3477	async fn failed_summarization_falls_back_to_truncation() {
3478	// Nothing listens here, so the summarization round always fails.
3479	let backend =
3480	OpenAiBackend::new("http://127.0.0.1:9", "", "test-model", Some("sys".into()));
3481	{
3482	let mut history = backend.history.lock().await;
3483	for i in 0..20 {
3484	history.push(serde_json::json!({
3485	"role": "user",
3486	"content": format!("message {i}: {}", "padding ".repeat(200)),
3487	}));
3488	}
3489	}
3490
3491	backend.compact_history(4).await.unwrap();
3492
3493	let history = backend.history_snapshot().await;
3494	assert_eq!(history[0]["role"], "system", "{history:?}");
3495	assert_eq!(history[1]["content"], TRUNCATION_PLACEHOLDER, "{history:?}");
3496	// The newest messages survive; the oldest were dropped to fit.
3497	assert_eq!(
3498	history.last().unwrap()["content"]
3499	.as_str()
3500	.unwrap()
3501	.split(':')
3502	.next()
3503	.unwrap(),
3504	"message 19",
3505	"{history:?}"
3506	);
3507	assert!(
3508	estimate_tokens(&history) <= DEFAULT_CONTEXT_TOKEN_BUDGET,
3509	"fallback result must fit the budget: ~{} tokens",
3510	estimate_tokens(&history)
3511	);
3512	}
3513
3514	/// A history small enough to begin with keeps its whole tail: the
3515	/// fallback drops nothing beyond what `keep_last` already folded away.
3516	#[tokio::test]
3517	async fn truncation_fallback_keeps_the_newest_messages_that_fit() {
3518	let system = vec![serde_json::json!({ "role": "system", "content": "sys" })];
3519	let tail: Vec<serde_json::Value> = (0..4)
3520	.map(\|i\| serde_json::json!({ "role": "user", "content": format!("m{i}") }))
3521	.collect();
3522
3523	let rebuilt = truncate_history_to_budget(&system, tail).unwrap();
3524
3525	assert_eq!(rebuilt[0]["role"], "system");
3526	assert_eq!(rebuilt[1]["content"], TRUNCATION_PLACEHOLDER);
3527	assert_eq!(rebuilt.len(), 6, "{rebuilt:?}");
3528	assert_eq!(rebuilt[2]["content"], "m0");
3529	assert_eq!(rebuilt[5]["content"], "m3");
3530	}
3531
3532	/// A multi-system-message session keeps every leading system message in
3533	/// the fallback, exactly like the summarization-success path.
3534	#[test]
3535	fn truncation_fallback_keeps_all_leading_system_messages() {
3536	let system = vec![
3537	serde_json::json!({ "role": "system", "content": "seed prompt" }),
3538	serde_json::json!({ "role": "system", "content": "session context" }),
3539	];
3540	let tail = vec![serde_json::json!({ "role": "user", "content": "hi" })];
3541
3542	let rebuilt = truncate_history_to_budget(&system, tail).unwrap();
3543
3544	assert_eq!(
3545	rebuilt[0]["content"], "seed prompt",
3546	"first system message kept: {rebuilt:?}"
3547	);
3548	assert_eq!(
3549	rebuilt[1]["content"], "session context",
3550	"second system message kept: {rebuilt:?}"
3551	);
3552	}
3553
3554	#[test]
3555	fn truncation_fallback_drops_orphaned_tool_results_with_their_call() {
3556	// A caller that split mid-round hands in a tail starting with tool
3557	// results whose assistant call message already fell outside the kept
3558	// window; those orphans must go before anything else, or strict
3559	// endpoints reject the next request.
3560	let tail = vec![
3561	serde_json::json!({ "role": "tool", "tool_call_id": "call_1", "content": "ok" }),
3562	serde_json::json!({ "role": "user", "content": "thanks" }),
3563	serde_json::json!({ "role": "assistant", "content": "done" }),
3564	];
3565
3566	let rebuilt = truncate_history_to_budget(&[], tail).unwrap();
3567
3568	let roles: Vec<&str> = rebuilt.iter().filter_map(\|m\| m["role"].as_str()).collect();
3569	assert_eq!(
3570	roles,
3571	["user", "user", "assistant"],
3572	"placeholder + the surviving exchange, orphan gone: {rebuilt:?}"
3573	);
3574	}
3575
3576	/// Review follow-up: a tail that drains to empty (one oversized tool
3577	/// round) must still test the minimal rebuild — system + placeholder fits,
3578	/// so the session is recoverable, not "unfixable".
3579	#[test]
3580	fn truncation_fallback_recovers_when_the_whole_tail_was_one_oversized_round() {
3581	let system = vec![serde_json::json!({ "role": "system", "content": "sys" })];
3582	let tail = vec![
3583	serde_json::json!({
3584	"role": "assistant",
3585	"content": null,
3586	"tool_calls": [{
3587	"id": "call_1", "type": "function",
3588	"function": { "name": "read_file", "arguments": "{}" },
3589	}],
3590	}),
3591	serde_json::json!({
3592	"role": "tool",
3593	"tool_call_id": "call_1",
3594	"content": "y".repeat(DEFAULT_CONTEXT_TOKEN_BUDGET * 4 + 1000),
3595	}),
3596	];
3597
3598	let rebuilt = truncate_history_to_budget(&system, tail).unwrap();
3599
3600	assert_eq!(rebuilt.len(), 2, "{rebuilt:?}");
3601	assert_eq!(rebuilt[0]["role"], "system");
3602	assert_eq!(rebuilt[1]["content"], TRUNCATION_PLACEHOLDER);
3603	}
3604
3605	#[test]
3606	fn truncation_fallback_reports_when_nothing_fits() {
3607	// Even the minimal rebuild (system + placeholder, no conversation)
3608	// overflows: the system messages alone are bigger than the window.
3609	let system = vec![serde_json::json!({
3610	"role": "system",
3611	"content": "s".repeat(DEFAULT_CONTEXT_TOKEN_BUDGET * 4 + 1000),
3612	})];
3613
3614	assert!(truncate_history_to_budget(&system, Vec::new()).is_none());
3615	}
3616
3617	/// Review follow-up: the local fallback recovers from a newest message
3618	/// that overflows the window on its own, the same way the remote one does.
3619	#[test]
3620	fn local_truncation_fallback_drops_an_oversized_final_message() {
3621	let system = vec![ChatMessage::system("sys")];
3622	let tail = vec![
3623	ChatMessage::user("hi"),
3624	ChatMessage::user("y".repeat(DEFAULT_CONTEXT_TOKEN_BUDGET * 4 + 1000)),
3625	];
3626
3627	let rebuilt = truncate_chat_messages_to_budget(system, tail).unwrap();
3628
3629	assert_eq!(rebuilt.len(), 2, "{rebuilt:?}");
3630	assert_eq!(rebuilt[0].content, "sys");
3631	assert_eq!(rebuilt[1].content, TRUNCATION_PLACEHOLDER);
3632	}
3633
3634	#[test]
3635	fn local_truncation_fallback_keeps_the_newest_messages_that_fit() {
3636	let big = DEFAULT_CONTEXT_TOKEN_BUDGET * 4 / 2;
3637	let tail = vec![
3638	ChatMessage::user("a".repeat(big)),
3639	ChatMessage::user("b".repeat(big)),
3640	ChatMessage::user("newest"),
3641	];
3642
3643	let rebuilt = truncate_chat_messages_to_budget(Vec::new(), tail).unwrap();
3644
3645	assert_eq!(rebuilt.len(), 3, "oldest dropped, rest kept");
3646	assert_eq!(rebuilt[0].content, TRUNCATION_PLACEHOLDER);
3647	assert!(rebuilt[1].content.starts_with('b'));
3648	assert_eq!(rebuilt[2].content, "newest");
3649	}
3650
3651	#[test]
3652	fn local_truncation_fallback_reports_when_the_system_messages_overflow() {
3653	let system = vec![ChatMessage::system(
3654	"s".repeat(DEFAULT_CONTEXT_TOKEN_BUDGET * 4 + 1000),
3655	)];
3656
3657	assert!(truncate_chat_messages_to_budget(system, Vec::new()).is_none());
3658	}
3659
3660	/// Review follow-up: head and tail share the budget, so the trimmed
3661	/// transcript stays near the cap instead of reaching twice it.
3662	#[test]
3663	fn truncate_transcript_middle_stays_within_the_cap() {
3664	let lines: Vec<String> = (0..2_000)
3665	.map(\|i\| format!("user: message {i} with some padding text"))
3666	.collect();
3667	let transcript = lines.join("\n");
3668
3669	let trimmed = truncate_transcript_middle(&transcript, 200); // ~800 chars
3670
3671	assert!(trimmed.contains("message 0 "), "opening kept: {trimmed}");
3672	assert!(trimmed.contains("message 1999"), "closing kept: {trimmed}");
3673	assert!(
3674	trimmed.len() <= 800 + 80,
3675	"cap plus the omission marker, got {} chars",
3676	trimmed.len()
3677	);
3678	}
3679
3680	#[test]
3681	fn truncate_transcript_middle_keeps_short_transcripts_verbatim() {
3682	let transcript = "user: hello\n\nassistant: hi";
3683
3684	assert_eq!(truncate_transcript_middle(transcript, 1_000), transcript);
3685	}
3686
3687	/// Review follow-up: a huge final message with no newline in the tail
3688	/// half must not be silently dropped — the closing exchange is what the
3689	/// summary most needs. The head gives up its share instead.
3690	#[test]
3691	fn truncate_transcript_middle_keeps_a_newlineless_final_message() {
3692	let mut lines: Vec<String> = (0..100).map(\|i\| format!("user: message {i}")).collect();
3693	lines.push(format!(
3694	"user: {}current state: pending review",
3695	"noise ".repeat(600)
3696	));
3697	let transcript = lines.join("\n");
3698
3699	let trimmed = truncate_transcript_middle(&transcript, 400); // ~1600 chars
3700
3701	assert!(
3702	trimmed.contains("current state: pending review"),
3703	"the closing exchange survives: {trimmed}"
3704	);
3705	assert!(
3706	trimmed.contains("older messages omitted"),
3707	"the gap is marked: {trimmed}"
3708	);
3709	assert!(
3710	!trimmed.contains("message 0"),
3711	"the head gives up its share to the unbreakable tail: {trimmed}"
3712	);
3713	}
3714
3715	/// Review follow-up: no newline anywhere in reach — keep the newest
3716	/// budget, not the oldest (which would invert the contract while the
3717	/// placeholder claims older messages were dropped).
3718	#[test]
3719	fn truncate_transcript_middle_without_any_newline_keeps_the_newest() {
3720	let transcript = format!("old decisions {} current state", "x".repeat(4_000));
3721
3722	let trimmed = truncate_transcript_middle(&transcript, 200); // ~800 chars
3723
3724	assert!(
3725	trimmed.contains("current state"),
3726	"the newest content survives: {trimmed}"
3727	);
3728	assert!(
3729	!trimmed.contains("old decisions"),
3730	"the oldest goes when only one side can survive: {trimmed}"
3731	);
3732	}
3733
3734	#[test]
3735	fn truncate_transcript_middle_cuts_the_middle_not_the_ends() {
3736	let mut lines: Vec<String> = Vec::new();
3737	for i in 0..200 {
3738	lines.push(format!("user: message {i} with some padding text"));
3739	}
3740	let transcript = lines.join("\n");
3741
3742	let trimmed = truncate_transcript_middle(&transcript, 200); // ~800 chars
3743
3744	assert!(trimmed.contains("message 0"), "opening kept: {trimmed}");
3745	assert!(trimmed.contains("message 199"), "closing kept: {trimmed}");
3746	assert!(
3747	trimmed.contains("older messages omitted"),
3748	"gap is marked: {trimmed}"
3749	);
3750	assert!(
3751	trimmed.len() <= transcript.len(),
3752	"never grows the transcript"
3753	);
3754	}
3755
3756	#[test]
3757	fn assistant_message_with_tool_calls_round_trips() {
3758	let message = ResponseMessage {
3759	content: None,
3760	tool_calls: Some(vec![ResponseToolCall {
3761	id: "call_1".to_string(),
3762	function: ResponseFunction {
3763	name: "read_file".to_string(),
3764	arguments: r#"{"path":"a.rs"}"#.to_string(),
3765	},
3766	}]),
3767	};
3768	let value = message.into_history_value();
3769	assert_eq!(value["role"], "assistant");
3770	assert!(value["content"].is_null());
3771	assert_eq!(value["tool_calls"][0]["id"], "call_1");
3772	assert_eq!(value["tool_calls"][0]["type"], "function");
3773	assert_eq!(value["tool_calls"][0]["function"]["name"], "read_file");
3774	}
3775	}