use crate::ollama_chat::{HttpMethod, NetworkPolicy, ReplayBoundary, ReplayDecision};
use router_protocol::{
CanonicalRequest, ContentBlock, InputItem, MessageRole, PhaseKind, RouteIdentity,
RouteIdentityError, ToolChoice, TransportKind,
};
use serde_json::{Map, Value, json};
use std::fmt;
pub const GEMINI_API_KEY_HEADER: &str = router_protocol::GEMINI_API_KEY_HEADER;
pub const DEFAULT_MAX_OUTPUT_TOKENS: u32 = 1024;
pub const fn is_gemini_phase(phase: PhaseKind) -> bool {
matches!(phase, PhaseKind::GeminiSubscription | PhaseKind::GeminiApi)
}
pub const fn capability_projection() -> router_protocol::MultiAgentCapabilityProjection {
router_protocol::MultiAgentCapabilityProjection {
namespace_tools: false,
multi_agent_version: router_protocol::MultiAgentVersion::V2,
wire_api: router_protocol::MultiAgentWireApi::Responses,
observed_namespace_calls:
router_protocol::ObservedNamespaceCallBehavior::BareNameNoNamespace,
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct GeminiSecretReferenceMarker(String);
impl GeminiSecretReferenceMarker {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for GeminiSecretReferenceMarker {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("GeminiSecretReferenceMarker([REDACTED])")
}
}
#[derive(Clone, PartialEq)]
pub struct GeminiRequestSpec {
pub method: HttpMethod,
pub origin: String,
pub path: String,
pub content_type: &'static str,
pub accept: &'static str,
pub api_key_secret: GeminiSecretReferenceMarker,
pub network: NetworkPolicy,
body: Value,
}
impl GeminiRequestSpec {
pub fn body(&self) -> &Value {
&self.body
}
pub fn body_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&self.body)
}
}
impl fmt::Debug for GeminiRequestSpec {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("GeminiRequestSpec")
.field("method", &self.method)
.field("origin", &self.origin)
.field("path", &self.path)
.field("content_type", &self.content_type)
.field("accept", &self.accept)
.field("api_key_secret", &self.api_key_secret)
.field("network", &self.network)
.field("body", &"[REDACTED]")
.finish()
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GeminiRequestErrorKind {
InvalidRouteIdentity,
InvalidRouteEndpoint,
UnsupportedRoute,
InvalidCanonicalRequest,
AmbiguousToolName,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeminiRequestError {
pub kind: GeminiRequestErrorKind,
}
impl fmt::Display for GeminiRequestError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Gemini generateContent request could not be prepared")
}
}
impl std::error::Error for GeminiRequestError {}
impl From<RouteIdentityError> for GeminiRequestError {
fn from(_: RouteIdentityError) -> Self {
Self {
kind: GeminiRequestErrorKind::InvalidRouteIdentity,
}
}
}
pub fn build_request(
route: &RouteIdentity,
request: &CanonicalRequest,
) -> Result<GeminiRequestSpec, GeminiRequestError> {
route.validate()?;
if !is_gemini_phase(route.phase) || route.transport != TransportKind::DirectHttps {
return Err(GeminiRequestError {
kind: GeminiRequestErrorKind::UnsupportedRoute,
});
}
request.validate().map_err(|_| GeminiRequestError {
kind: GeminiRequestErrorKind::InvalidCanonicalRequest,
})?;
let secret_ref = route.credential_ref.clone().ok_or(GeminiRequestError {
kind: GeminiRequestErrorKind::InvalidRouteEndpoint,
})?;
Ok(GeminiRequestSpec {
method: HttpMethod::Post,
origin: route.endpoint.origin.clone(),
path: materialize_stream_path(&route.endpoint.path, &route.model_id),
content_type: "application/json",
accept: "text/event-stream",
api_key_secret: GeminiSecretReferenceMarker(secret_ref),
network: NetworkPolicy {
allow_proxy: false,
allow_dns: true,
follow_redirects: false,
require_tls: true,
},
body: build_body(request)?,
})
}
fn materialize_stream_path(template: &str, model_id: &str) -> String {
let encoded = encode_model_id(model_id);
let with_model = template.replace("{model}", &encoded);
with_model.replace(":generateContent", ":streamGenerateContent")
}
fn encode_model_id(model_id: &str) -> String {
let mut out = String::with_capacity(model_id.len());
for byte in model_id.bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'.' | b'_' | b'-' => {
out.push(byte as char);
}
other => {
out.push('%');
out.push_str(&format!("{other:02X}"));
}
}
}
out
}
fn build_body(request: &CanonicalRequest) -> Result<Value, GeminiRequestError> {
let mut object = Map::new();
object.insert("contents".to_owned(), Value::Array(build_contents(request)));
let mut generation_config = Map::new();
generation_config.insert(
"maxOutputTokens".to_owned(),
Value::from(
request
.max_output_tokens
.unwrap_or(DEFAULT_MAX_OUTPUT_TOKENS),
),
);
object.insert(
"generationConfig".to_owned(),
Value::Object(generation_config),
);
let system = build_system_instruction(request);
if !system.is_empty() {
object.insert(
"systemInstruction".to_owned(),
json!({"parts": [{"text": system}]}),
);
}
if !request.tools.is_empty() {
let mut names = std::collections::HashSet::new();
let mut declarations = Vec::with_capacity(request.tools.len());
for tool in &request.tools {
let name = provider_tool_name(tool.namespace.as_deref(), &tool.name);
if !names.insert(name.clone()) {
return Err(GeminiRequestError {
kind: GeminiRequestErrorKind::AmbiguousToolName,
});
}
declarations.push(json!({
"name": name,
"description": tool.description,
"parameters": tool.parameters,
}));
}
object.insert(
"tools".to_owned(),
json!([{"functionDeclarations": declarations}]),
);
object.insert(
"toolConfig".to_owned(),
json!({"functionCallingConfig": {"mode": gemini_tool_mode(&request.tool_choice)}}),
);
}
Ok(Value::Object(object))
}
fn build_system_instruction(request: &CanonicalRequest) -> String {
request
.input
.iter()
.filter_map(|item| match item {
InputItem::Message {
role: MessageRole::Developer | MessageRole::System,
content,
} => Some(content),
_ => None,
})
.flat_map(|content| {
content.iter().filter_map(|block| match block {
ContentBlock::Text { text } => Some(text.as_str()),
ContentBlock::ImageReference { .. } => None,
})
})
.collect::<Vec<_>>()
.join("\n")
}
fn build_contents(request: &CanonicalRequest) -> Vec<Value> {
let mut call_name = std::collections::HashMap::new();
for item in &request.input {
if let InputItem::FunctionCall {
call_id,
namespace,
name,
..
} = item
{
call_name.insert(
call_id.as_str(),
provider_tool_name(namespace.as_deref(), name),
);
}
}
request
.input
.iter()
.map(|item| match item {
InputItem::Message {
role: MessageRole::User,
content,
} => json!({
"role": "user",
"parts": gemini_parts(content),
}),
InputItem::Message {
role: MessageRole::Assistant,
content,
} => json!({
"role": "model",
"parts": gemini_parts(content),
}),
InputItem::Message {
role: MessageRole::Developer | MessageRole::System,
..
} => json!({
"role": "user",
"parts": [],
}),
InputItem::FunctionCall {
call_id: _,
namespace,
name,
arguments,
} => json!({
"role": "model",
"parts": [{
"functionCall": {
"name": provider_tool_name(namespace.as_deref(), name),
"args": arguments,
}
}]
}),
InputItem::FunctionCallOutput { call_id, output } => {
let name = call_name
.get(call_id.as_str())
.map(String::as_str)
.unwrap_or("result");
json!({
"role": "user",
"parts": [{
"functionResponse": {
"name": name,
"response": {"result": output.to_string()},
}
}]
})
}
})
.collect()
}
fn gemini_parts(content: &[ContentBlock]) -> Value {
Value::Array(
content
.iter()
.filter_map(|block| match block {
ContentBlock::Text { text } => Some(json!({"text": text})),
ContentBlock::ImageReference { .. } => {
None
}
})
.collect(),
)
}
fn provider_tool_name(namespace: Option<&str>, name: &str) -> String {
namespace.map_or_else(
|| name.to_owned(),
|namespace| format!("{namespace}__{name}"),
)
}
fn gemini_tool_mode(tool_choice: &ToolChoice) -> &'static str {
match tool_choice {
ToolChoice::Auto => "AUTO",
ToolChoice::None => "NONE",
ToolChoice::Required | ToolChoice::Named(_) => "ANY",
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GeminiUsage {
pub prompt_tokens: u64,
pub candidates_tokens: u64,
pub total_tokens: u64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum GeminiStreamEvent {
ContentDelta { text: String },
FinishReason { reason: String },
Usage(GeminiUsage),
Done,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GeminiStreamErrorKind {
InvalidUtf8,
InvalidJson,
InvalidFrame,
MissingTerminal,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeminiStreamError {
pub kind: GeminiStreamErrorKind,
}
impl fmt::Display for GeminiStreamError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Gemini stream could not be normalized")
}
}
impl std::error::Error for GeminiStreamError {}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct GeminiReplayGuard {
boundary: Option<ReplayBoundary>,
body_written: bool,
}
impl GeminiReplayGuard {
pub fn mark_body_written(&mut self) {
self.body_written = true;
}
pub fn observe(&mut self, event: &GeminiStreamEvent) {
match event {
GeminiStreamEvent::ContentDelta { text } if !text.is_empty() => {
self.cross(ReplayBoundary::PublicContent);
}
GeminiStreamEvent::ContentDelta { .. }
| GeminiStreamEvent::FinishReason { .. }
| GeminiStreamEvent::Usage(_)
| GeminiStreamEvent::Done => {}
}
}
pub fn mark_transport_outcome_unknown(&mut self) {
if self.body_written {
self.cross(ReplayBoundary::AmbiguousDispatch);
}
}
pub fn decision(&self) -> ReplayDecision {
self.boundary
.map_or(ReplayDecision::Allowed, ReplayDecision::Denied)
}
fn cross(&mut self, boundary: ReplayBoundary) {
if self.boundary.is_none() {
self.boundary = Some(boundary);
}
}
}
pub struct GeminiSseParser {
_attempt_id: router_protocol::AttemptId,
replay: GeminiReplayGuard,
terminal: bool,
prompt_tokens: u64,
candidates_tokens: u64,
total_tokens: u64,
}
impl GeminiSseParser {
pub fn new(attempt_id: &router_protocol::AttemptId) -> Self {
Self {
_attempt_id: attempt_id.clone(),
replay: GeminiReplayGuard::default(),
terminal: false,
prompt_tokens: 0,
candidates_tokens: 0,
total_tokens: 0,
}
}
pub fn replay_decision(&self) -> ReplayDecision {
self.replay.decision()
}
pub fn mark_body_written(&mut self) {
self.replay.mark_body_written();
}
pub fn mark_transport_outcome_unknown(&mut self) {
self.replay.mark_transport_outcome_unknown();
}
pub fn feed(&mut self, chunk: &[u8]) -> Result<Vec<GeminiStreamEvent>, GeminiStreamError> {
let mut events = Vec::new();
for frame in split_sse_frames(chunk)? {
let value: Value = serde_json::from_slice(&frame).map_err(|_| GeminiStreamError {
kind: GeminiStreamErrorKind::InvalidJson,
})?;
events.extend(self.observe_json(value)?);
}
for event in &events {
self.replay.observe(event);
}
Ok(events)
}
pub fn finish(&mut self) -> Result<Vec<GeminiStreamEvent>, GeminiStreamError> {
if !self.terminal {
return Err(GeminiStreamError {
kind: GeminiStreamErrorKind::MissingTerminal,
});
}
Ok(Vec::new())
}
fn observe_json(&mut self, value: Value) -> Result<Vec<GeminiStreamEvent>, GeminiStreamError> {
let mut events = Vec::new();
if let Some(candidates) = value.get("candidates").and_then(Value::as_array) {
for candidate in candidates {
if let Some(text) = candidate
.pointer("/content/parts")
.and_then(Value::as_array)
.and_then(|parts| {
parts
.iter()
.find_map(|part| part.get("text").and_then(Value::as_str))
})
&& !text.is_empty()
{
events.push(GeminiStreamEvent::ContentDelta {
text: text.to_owned(),
});
}
if let Some(reason) = candidate.get("finishReason").and_then(Value::as_str) {
events.push(GeminiStreamEvent::FinishReason {
reason: reason.to_owned(),
});
if !self.terminal {
self.terminal = true;
if let Some(usage) = value.get("usageMetadata") {
self.capture_usage(usage);
}
events.push(GeminiStreamEvent::Usage(GeminiUsage {
prompt_tokens: self.prompt_tokens,
candidates_tokens: self.candidates_tokens,
total_tokens: self.total_tokens,
}));
events.push(GeminiStreamEvent::Done);
}
}
}
}
if let Some(usage) = value.get("usageMetadata") {
self.capture_usage(usage);
}
Ok(events)
}
fn capture_usage(&mut self, usage: &Value) {
if let Some(prompt) = usage.get("promptTokenCount").and_then(Value::as_u64) {
self.prompt_tokens = prompt;
}
if let Some(candidates) = usage.get("candidatesTokenCount").and_then(Value::as_u64) {
self.candidates_tokens = candidates;
}
if let Some(total) = usage.get("totalTokenCount").and_then(Value::as_u64) {
self.total_tokens = total;
}
}
}
fn split_sse_frames(buffer: &[u8]) -> Result<Vec<Vec<u8>>, GeminiStreamError> {
let text = std::str::from_utf8(buffer).map_err(|_| GeminiStreamError {
kind: GeminiStreamErrorKind::InvalidUtf8,
})?;
let mut frames = Vec::new();
let mut current = Vec::new();
for line in text.split('\n') {
let line = line.strip_suffix('\r').unwrap_or(line);
if line.is_empty() {
if !current.is_empty() {
frames.push(std::mem::take(&mut current));
}
continue;
}
if let Some(data) = line.strip_prefix("data:") {
let data = data.strip_prefix(' ').unwrap_or(data);
if !current.is_empty() {
current.push(b'\n');
}
current.extend_from_slice(data.as_bytes());
}
}
if !current.is_empty() {
frames.push(current);
}
if frames.is_empty() {
return Err(GeminiStreamError {
kind: GeminiStreamErrorKind::InvalidFrame,
});
}
Ok(frames)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GeminiFailureClass {
InvalidRequest,
Authentication,
PaymentRequired,
Authorization,
NotFound,
Timeout,
Unprocessable,
RateLimited,
UpstreamServer,
UnexpectedStatus,
MismatchedContentType,
RedirectRejected,
Cancelled,
TransportTimeout,
AmbiguousDispatch,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeminiFailure {
pub class: GeminiFailureClass,
pub status: Option<u16>,
pub retry_after: Option<crate::ollama_chat::BoundedRetryAfter>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GeminiBodyWriteState {
NotWritten,
Written,
Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GeminiTransportFailureSignal {
Cancelled,
Timeout,
}
pub fn normalize_http_response(
status: u16,
content_type: Option<&str>,
retry_after: Option<&str>,
redirected: bool,
) -> Result<(), GeminiFailure> {
if redirected || (300..400).contains(&status) {
return Err(gemini_failure(
GeminiFailureClass::RedirectRejected,
Some(status),
retry_after,
));
}
if status != 200 {
let class = match status {
400 => GeminiFailureClass::InvalidRequest,
401 => GeminiFailureClass::Authentication,
402 => GeminiFailureClass::PaymentRequired,
403 => GeminiFailureClass::Authorization,
404 => GeminiFailureClass::NotFound,
408 => GeminiFailureClass::Timeout,
422 => GeminiFailureClass::Unprocessable,
429 => GeminiFailureClass::RateLimited,
500..=599 => GeminiFailureClass::UpstreamServer,
_ => GeminiFailureClass::UnexpectedStatus,
};
return Err(gemini_failure(class, Some(status), retry_after));
}
let media_type = content_type
.and_then(|value| value.split(';').next())
.map(str::trim);
if media_type != Some("text/event-stream") {
return Err(gemini_failure(
GeminiFailureClass::MismatchedContentType,
Some(status),
None,
));
}
Ok(())
}
pub fn normalize_transport_failure(
signal: GeminiTransportFailureSignal,
body_write: GeminiBodyWriteState,
) -> GeminiFailure {
let class = match body_write {
GeminiBodyWriteState::Written | GeminiBodyWriteState::Unknown => {
GeminiFailureClass::AmbiguousDispatch
}
GeminiBodyWriteState::NotWritten => match signal {
GeminiTransportFailureSignal::Cancelled => GeminiFailureClass::Cancelled,
GeminiTransportFailureSignal::Timeout => GeminiFailureClass::TransportTimeout,
},
};
gemini_failure(class, None, None)
}
fn gemini_failure(
class: GeminiFailureClass,
status: Option<u16>,
retry_after: Option<&str>,
) -> GeminiFailure {
GeminiFailure {
class,
status,
retry_after: crate::ollama_chat::BoundedRetryAfter::parse(retry_after),
}
}
#[cfg(test)]
mod tests {
use super::*;
use router_protocol::{
BillingAuthorityClass, CapabilityRequirements, CatalogSourceKind, CredentialSourceKind,
EndpointIdentity, ExecutionLocation, FunctionTool, HarnessInstanceId, M1_WIRE_VERSION,
MessageRole, ModelNamespace, PhaseKind, PhasePlacement, ProviderService, ReasoningIntent,
SessionId, TurnId,
};
fn gemini_route(phase: PhaseKind, billing: BillingAuthorityClass) -> RouteIdentity {
RouteIdentity {
phase,
placement: PhasePlacement::Builtin,
provider: ProviderService::Gemini,
transport: TransportKind::DirectHttps,
endpoint: EndpointIdentity {
origin: router_protocol::GEMINI_ORIGIN.to_owned(),
path: router_protocol::GEMINI_GENERATE_CONTENT_PATH.to_owned(),
},
credential_source: CredentialSourceKind::SecretRef,
credential_ref: Some("env:STEADYMUX_GEMINI_KEY".to_owned()),
credential_refs: vec!["env:STEADYMUX_GEMINI_KEY".to_owned()],
catalog_source: CatalogSourceKind::GeminiCatalog,
catalog_ref: "catalog:gemini:verified".to_owned(),
billing_authority: billing,
model_namespace: ModelNamespace::Gemini,
model_id: "gemini-2.5-flash".to_owned(),
execution_location: ExecutionLocation::RemoteCloud,
account_ref: "account:gemini".to_owned(),
capacity_domain_ref: "capacity:gemini".to_owned(),
local_execution_proof: false,
approved_local_inventory: Vec::new(),
reasoning_modes: Vec::new(),
}
}
fn request(prompt: &str) -> CanonicalRequest {
CanonicalRequest {
wire_version: M1_WIRE_VERSION.to_owned(),
model: "router-auto".to_owned(),
stream: true,
harness_instance: HarnessInstanceId::parse("fixture-cli").expect("fixture"),
session_id: SessionId::parse("session-gemini").expect("fixture"),
turn_id: TurnId::parse("turn-gemini").expect("fixture"),
input: vec![InputItem::Message {
role: MessageRole::User,
content: vec![ContentBlock::Text {
text: prompt.to_owned(),
}],
}],
tools: vec![FunctionTool {
namespace: Some("forecast".to_owned()),
name: "current".to_owned(),
description: "current forecast".to_owned(),
parameters: json!({"type":"object"}),
strict: true,
}],
tool_choice: ToolChoice::Auto,
structured_output: None,
reasoning: ReasoningIntent::None,
max_output_tokens: Some(256),
max_spend_micro_usd: None,
capabilities: CapabilityRequirements::default(),
}
}
#[test]
fn gemini_routes_build_verified_generate_content_wire_shape() {
for (phase, billing) in [
(
PhaseKind::GeminiSubscription,
BillingAuthorityClass::GeminiSubscription,
),
(
PhaseKind::GeminiApi,
BillingAuthorityClass::GeminiMeteredApi,
),
] {
let route = gemini_route(phase, billing);
route.validate().expect("valid gemini route");
let spec = build_request(&route, &request("gemini-canary")).expect("request");
assert_eq!(spec.method, HttpMethod::Post);
assert_eq!(spec.origin, router_protocol::GEMINI_ORIGIN);
assert_eq!(
spec.path,
"/v1beta/models/gemini-2.5-flash:streamGenerateContent"
);
assert_eq!(spec.accept, "text/event-stream");
assert_eq!(spec.api_key_secret.as_str(), "env:STEADYMUX_GEMINI_KEY");
assert!(spec.network.require_tls);
assert!(!spec.network.follow_redirects);
assert!(!spec.network.allow_proxy);
assert!(spec.body().get("model").is_none());
assert_eq!(spec.body()["contents"][0]["role"], json!("user"));
assert_eq!(
spec.body()["contents"][0]["parts"][0]["text"],
json!("gemini-canary")
);
assert_eq!(
spec.body()["generationConfig"]["maxOutputTokens"],
json!(256)
);
assert_eq!(
spec.body()["tools"][0]["functionDeclarations"][0]["name"],
json!("forecast__current")
);
assert_eq!(
spec.body()["toolConfig"]["functionCallingConfig"]["mode"],
json!("AUTO")
);
}
}
#[test]
fn gemini_adapter_has_no_hardcoded_model_id_and_exposes_projection() {
let adapter_core = include_str!("gemini_chat.rs")
.split("#[cfg(test)]")
.next()
.expect("adapter body precedes the test module");
for model_literal in [
concat!("gemini", "-2"),
concat!("gemini", "-3"),
concat!("gemini", "-pro"),
concat!("gemini", "-flash"),
] {
assert!(
!adapter_core.contains(model_literal),
"adapter hardcodes the model literal {model_literal}"
);
}
assert!(is_gemini_phase(PhaseKind::GeminiSubscription));
assert!(is_gemini_phase(PhaseKind::GeminiApi));
assert!(!is_gemini_phase(PhaseKind::ClaudeApi));
assert!(!capability_projection().namespace_tools);
assert_eq!(
PhaseKind::GeminiApi.multi_agent_v2_projection(),
capability_projection()
);
}
#[test]
fn gemini_sse_parses_text_finish_and_usage_once() {
let transcript = concat!(
"data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"hello \"}]},\"finishReason\":null}],\"usageMetadata\":{\"promptTokenCount\":9,\"candidatesTokenCount\":0,\"totalTokenCount\":9}}\n\n",
"data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"world\"}]},\"finishReason\":null}]}\n\n",
"data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"\"}]},\"finishReason\":\"STOP\"}],\"usageMetadata\":{\"promptTokenCount\":9,\"candidatesTokenCount\":5,\"totalTokenCount\":14}}\n\n",
);
let attempt = router_protocol::AttemptId::parse("attempt-gemini-1").expect("attempt");
let mut parser = GeminiSseParser::new(&attempt);
let events = parser
.feed(transcript.as_bytes())
.expect("feed gemini stream");
assert_eq!(
events,
vec![
GeminiStreamEvent::ContentDelta {
text: "hello ".to_owned()
},
GeminiStreamEvent::ContentDelta {
text: "world".to_owned()
},
GeminiStreamEvent::FinishReason {
reason: "STOP".to_owned()
},
GeminiStreamEvent::Usage(GeminiUsage {
prompt_tokens: 9,
candidates_tokens: 5,
total_tokens: 14,
}),
GeminiStreamEvent::Done,
]
);
assert_eq!(parser.finish(), Ok(vec![]));
assert!(parser.replay_decision() != ReplayDecision::Allowed);
}
#[test]
fn gemini_stream_requires_terminal() {
let attempt = router_protocol::AttemptId::parse("attempt-gemini-2").expect("attempt");
let mut parser = GeminiSseParser::new(&attempt);
parser
.feed(
b"data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"partial\"}]},\"finishReason\":null}]}\n\n",
)
.expect("feed");
assert_eq!(
parser.finish(),
Err(GeminiStreamError {
kind: GeminiStreamErrorKind::MissingTerminal
})
);
}
}