gproxy_protocol/adapt/
images.rs

1//! Bounded image calls with explicit progress. There is no implicit retry.
2//!
3//! The host must own a unique operation identity and scope for each generation.
4//! Resource publication is idempotent; upstream image generation is not. A new
5//! ImageProgress is not permission to repeat an uncertain billed generation.
6//! Dropping the future stops later calls. Completed bytes/usage and recorded
7//! publication identities survive in caller-owned progress; the host can query
8//! publication_status and release its owned handles explicitly. This adapter
9//! never deletes resources or silently compensates a partial publication.
10//!
11//! All requests, source options, byte caps and required delivery facts are
12//! checked before the first upstream send. All output images are validated
13//! before the first publish. Non-2xx responses retain their original HTTP body;
14//! invalid decoded 2xx responses retain the typed native response, including
15//! usage, so failed output conversion does not erase billing evidence.
16
17use crate::{
18    HttpBody, WireRequest, WireResponse,
19    adapt::{JsonInvocation, invoke_json},
20    capability::{
21        PublicationKind, PublishedResource, ResourceAccess, ResourceMetadata, ResourceReference,
22        Upstream,
23    },
24    codec::{self, CodecLimits},
25    openai::images as o,
26    transform::{
27        Report, TransformError, TransformErrorKind,
28        images::{
29            self, ImageCallResult, ImageDialect, ImageDialectBody, ImageInput, ImageResponseFacts,
30            ImageTargetModels, ResolvedImageInput,
31        },
32    },
33};
34use base64::Engine;
35
36#[derive(Debug, Clone, Copy)]
37pub struct ImageLimits {
38    pub codec: CodecLimits,
39
40    pub max_input_bytes: u64,
41    pub max_total_input_bytes: u64,
42    pub max_output_bytes: u64,
43    pub max_total_output_bytes: u64,
44}
45
46/// Caller-owned progress survives future cancellation. Once generation starts,
47/// this object cannot be reused for generation: a dropped send may have billed
48/// upstream. All completed image bytes and native usage remain available.
49/// Publication operation IDs are recorded BEFORE awaiting publish; recover an
50/// uncertain result with ResourceAccess::publication_status(scope, id).
51#[derive(Debug)]
52pub struct ImageProgress<H> {
53    started: bool,
54    attempted_calls: usize,
55    calls: Vec<ImageCallResult>,
56    publication_ids: Vec<String>,
57    published: Vec<PublishedResource<H>>,
58}
59
60impl<H> Default for ImageProgress<H> {
61    fn default() -> Self {
62        Self {
63            started: false,
64            attempted_calls: 0,
65            calls: Vec::new(),
66            publication_ids: Vec::new(),
67            published: Vec::new(),
68        }
69    }
70}
71
72impl<H> ImageProgress<H> {
73    pub fn started(&self) -> bool {
74        self.started
75    }
76    pub fn attempted_calls(&self) -> usize {
77        self.attempted_calls
78    }
79    pub fn calls(&self) -> &[ImageCallResult] {
80        &self.calls
81    }
82    pub fn publication_ids(&self) -> &[String] {
83        &self.publication_ids
84    }
85    pub fn published(&self) -> &[PublishedResource<H>] {
86        &self.published
87    }
88}
89
90#[derive(Debug)]
91pub enum ImageError {
92    Transform(TransformError),
93    Rejected(Box<WireResponse<HttpBody>>),
94    InvalidResponses {
95        error: TransformError,
96        response:
97            Box<WireResponse<crate::openai::responses::response::GenerateContentResponseBody>>,
98    },
99    InvalidGemini {
100        error: TransformError,
101        response: Box<WireResponse<crate::gemini::GenerateContentResponseBody>>,
102    },
103}
104
105impl From<TransformError> for ImageError {
106    fn from(v: TransformError) -> Self {
107        Self::Transform(v)
108    }
109}
110
111impl std::fmt::Display for ImageError {
112    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
113        match self {
114            Self::Transform(e)
115            | Self::InvalidResponses { error: e, .. }
116            | Self::InvalidGemini { error: e, .. } => e.fmt(f),
117            Self::Rejected(r) => write!(f, "image upstream returned HTTP {}", r.status),
118        }
119    }
120}
121
122impl std::error::Error for ImageError {
123    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
124        match self {
125            Self::Transform(e)
126            | Self::InvalidResponses { error: e, .. }
127            | Self::InvalidGemini { error: e, .. } => Some(e),
128            Self::Rejected(_) => None,
129        }
130    }
131}
132
133#[derive(Debug)]
134pub struct ImageOutcome {
135    pub response: o::ImagesResponse,
136    pub report: Report,
137}
138
139fn limit(context: &str) -> TransformError {
140    TransformError::new(
141        TransformErrorKind::Limit,
142        context,
143        "image operation exceeds configured cap",
144    )
145}
146
147fn clone_request<T: Clone>(r: &WireRequest<T>) -> WireRequest<T> {
148    WireRequest {
149        method: r.method.clone(),
150        path: r.path.clone(),
151        query: r.query.clone(),
152        headers: r.headers.clone(),
153        body: r.body.clone(),
154    }
155}
156
157fn request_fits<T: serde::Serialize>(
158    r: &WireRequest<T>,
159    limits: CodecLimits,
160    max_write: u64,
161) -> Result<(), TransformError> {
162    codec::encode_json(
163        &r.body,
164        CodecLimits {
165            max_body_bytes: limits.max_body_bytes.min(max_write),
166            ..limits
167        },
168    )
169    .map(|_| ())
170    .map_err(|e| {
171        TransformError::new(
172            if e.kind() == codec::CodecErrorKind::Limit {
173                TransformErrorKind::Limit
174            } else {
175                TransformErrorKind::InvalidInput
176            },
177            "image.request",
178            e.to_string(),
179        )
180    })
181}
182
183#[allow(clippy::too_many_arguments)]
184pub async fn generate<A: ResourceAccess, U: Upstream>(
185    access: &A,
186    scope: &A::Scope,
187    upstream: &U,
188    target: &U::Target,
189    input: ImageInput,
190    dialect: ImageDialect,
191    models: &ImageTargetModels,
192    facts: &ImageResponseFacts,
193    limits: ImageLimits,
194    progress: &mut ImageProgress<A::PublishedHandle>,
195) -> Result<ImageOutcome, ImageError> {
196    if progress.started {
197        return Err(TransformError::new(TransformErrorKind::MissingState,"image.progress","generation already started; recover recorded publications, never replay uncertain upstream calls").into());
198    }
199    let (context, refs, mut report) = images::preflight(&input, dialect, models)?;
200    let delivery = context
201        .response_format
202        .unwrap_or(facts.default_response_format);
203
204    if facts.created < 0 {
205        return Err(TransformError::shape(
206            "image.created",
207            "nonnegative factual Unix timestamp required",
208        )
209        .into());
210    }
211    if delivery == o::ImageResponseFormat::Url {
212        if facts.operation_id.len() as u64 > limits.codec.max_value_bytes {
213            return Err(limit("image.operation_id").into());
214        }
215        if facts.operation_id.trim().is_empty() || facts.operation_id.chars().any(char::is_control)
216        {
217            return Err(TransformError::shape(
218                "image.operation_id",
219                "nonempty control-free operation identity required",
220            )
221            .into());
222        }
223        if facts
224            .publish_expires_at
225            .is_none_or(|t| t <= facts.observed_at)
226        {
227            return Err(TransformError::shape(
228                "image.publish.expires_at",
229                "future expiry required before generation",
230            )
231            .into());
232        }
233    }
234    let mut resolved = Vec::new();
235    let mut total = 0u64;
236    for reference in refs {
237        if resolved
238            .iter()
239            .any(|r: &ResolvedImageInput| r.reference == reference)
240        {
241            continue;
242        }
243        let remaining = limits
244            .max_total_input_bytes
245            .checked_sub(total)
246            .ok_or_else(|| limit("image.input.total"))?;
247        let (image, len) = read_image(
248            access,
249            scope,
250            &reference,
251            limits.codec,
252            limits.max_input_bytes.min(remaining),
253            facts.observed_at,
254        )
255        .await?;
256        total = total
257            .checked_add(len)
258            .ok_or_else(|| limit("image.input.total"))?;
259        resolved.push(image);
260    }
261    let prepared = images::build_request(input, dialect, models, &resolved)?.value;
262    match &prepared.body {
263        ImageDialectBody::Responses(r) => {
264            request_fits(r, limits.codec, upstream.limits().write_bytes)?
265        }
266        ImageDialectBody::Gemini(r) => {
267            request_fits(r, limits.codec, upstream.limits().write_bytes)?
268        }
269    };
270    // No upstream or publication side effect occurred during preflight.
271    if limits.max_output_bytes == 0
272        || limits.max_total_output_bytes == 0
273        || (delivery == o::ImageResponseFormat::Url && access.limits().write_bytes == 0)
274    {
275        return Err(limit("image.output.bytes").into());
276    }
277    progress.started = true;
278    let mut total_output = 0u64;
279    for _ in 0..context.n {
280        let cap = limits
281            .max_output_bytes
282            .min(if delivery == o::ImageResponseFormat::Url {
283                access.limits().write_bytes
284            } else {
285                u64::MAX
286            })
287            .min(
288                limits
289                    .max_total_output_bytes
290                    .checked_sub(total_output)
291                    .ok_or_else(|| limit("image.output.total"))?,
292            );
293        if cap == 0 {
294            return Err(limit("image.output.total").into());
295        }
296        progress.attempted_calls += 1;
297        let result = match &prepared.body {
298            ImageDialectBody::Responses(r) => {
299                match invoke_json::<
300                    _,
301                    _,
302                    crate::openai::responses::response::GenerateContentResponseBody,
303                >(upstream, target, clone_request(r), limits.codec)
304                .await?
305                {
306                    JsonInvocation::Success(r) => {
307                        match images::image_response_from_responses(&r.body, &context, cap) {
308                            Ok(v) => v,
309                            Err(error) => {
310                                return Err(ImageError::InvalidResponses {
311                                    error,
312                                    response: Box::new(r),
313                                });
314                            }
315                        }
316                    }
317                    JsonInvocation::Rejected(r) => return Err(ImageError::Rejected(Box::new(r))),
318                }
319            }
320            ImageDialectBody::Gemini(r) => match invoke_json::<
321                _,
322                _,
323                crate::gemini::GenerateContentResponseBody,
324            >(
325                upstream, target, clone_request(r), limits.codec
326            )
327            .await?
328            {
329                JsonInvocation::Success(r) => {
330                    match images::image_response_from_gemini(&r.body, &context, cap) {
331                        Ok(v) => v,
332                        Err(error) => {
333                            return Err(ImageError::InvalidGemini {
334                                error,
335                                response: Box::new(r),
336                            });
337                        }
338                    }
339                }
340                JsonInvocation::Rejected(r) => return Err(ImageError::Rejected(Box::new(r))),
341            },
342        };
343        total_output = total_output
344            .checked_add(result.image.bytes.len() as u64)
345            .ok_or_else(|| limit("image.output.total"))?;
346        report
347            .diagnostics
348            .extend(result.report.diagnostics.iter().cloned());
349        progress.calls.push(result);
350    }
351    // Every output has been validated before the first publication.
352    if delivery == o::ImageResponseFormat::Url
353        && progress
354            .calls
355            .iter()
356            .any(|c| c.image.bytes.len() as u64 > access.limits().write_bytes)
357    {
358        return Err(limit("image.publish.bytes").into());
359    }
360    let mut data = Vec::new();
361    for (index, call) in progress.calls.iter().enumerate() {
362        let mut image = o::GeneratedImage::builder().build();
363        image.revised_prompt = call.image.revised_prompt.clone();
364        if delivery == o::ImageResponseFormat::Url {
365            // Length-prefixed identity makes the index suffix unambiguous.
366            let operation = format!(
367                "image:{}:{}:{}",
368                facts.operation_id.len(),
369                facts.operation_id,
370                index
371            );
372            progress.publication_ids.push(operation.clone());
373            let publication = access
374                .publish(
375                    scope,
376                    &operation,
377                    PublicationKind::Url,
378                    ResourceMetadata {
379                        mime: Some(call.image.metadata.mime().into()),
380                        length: Some(call.image.bytes.len() as u64),
381                        filename: None,
382                        expires_at: facts.publish_expires_at,
383                    },
384                    HttpBody::Bytes(call.image.bytes.clone()),
385                )
386                .await
387                .map_err(TransformError::from)?;
388            // Retain even a malformed host result's handle for compensation.
389            progress.published.push(publication);
390            let publication = progress.published.last().unwrap();
391            let ResourceReference::Url(url) = &publication.reference else {
392                return Err(TransformError::invalid_result(
393                    "image.publish",
394                    "host returned an ID instead of URL",
395                )
396                .into());
397            };
398            if url.trim().is_empty()
399                || publication.metadata.mime.as_deref() != Some(call.image.metadata.mime())
400                || publication.metadata.length != Some(call.image.bytes.len() as u64)
401                || publication
402                    .metadata
403                    .expires_at
404                    .is_none_or(|t| t <= facts.observed_at)
405            {
406                return Err(TransformError::invalid_result(
407                    "image.publish",
408                    "publication metadata contradicts image or is expired",
409                )
410                .into());
411            }
412            image.url = Some(url.clone());
413        } else {
414            image.b64_json =
415                Some(base64::engine::general_purpose::STANDARD.encode(&call.image.bytes));
416        }
417        data.push(image);
418    }
419    let first = &progress.calls[0].image.metadata;
420    let format = progress
421        .calls
422        .iter()
423        .all(|c| c.image.metadata.format == first.format)
424        .then_some(first.format);
425    let size =
426        if progress.calls.iter().all(|c| {
427            (c.image.metadata.width, c.image.metadata.height) == (first.width, first.height)
428        }) {
429            match (first.width, first.height) {
430                (1024, 1024) => Some(o::OutputImageSize::Square),
431                (1536, 1024) => Some(o::OutputImageSize::Landscape),
432                (1024, 1536) => Some(o::OutputImageSize::Portrait),
433                (width, height) => Some(o::OutputImageSize::Custom(format!("{width}x{height}"))),
434            }
435        } else {
436            None
437        };
438    if progress.calls.iter().any(|c| c.usage.is_some()) {
439        report.changed("usage","native usage is retained per call in ImageProgress; native generic token counters cannot supply Images image/text token breakdown");
440    }
441    let mut response = o::ImagesResponse::builder(facts.created).data(data).build();
442    response.output_format = format;
443    response.size = size;
444    // Quality/background are requested controls, not factual upstream results.
445    Ok(ImageOutcome { response, report })
446}
447
448async fn read_image<A: ResourceAccess>(
449    access: &A,
450    scope: &A::Scope,
451    reference: &ResourceReference,
452    codec_limits: CodecLimits,
453    max_bytes: u64,
454    observed_at: std::time::SystemTime,
455) -> Result<(ResolvedImageInput, u64), TransformError> {
456    let (bytes, mime) = if let ResourceReference::Url(url) = reference {
457        if let Some(data) = url.strip_prefix("data:") {
458            let (mime, encoded) = data.split_once(";base64,").ok_or_else(|| {
459                TransformError::shape("image.data_url", "base64 data URL required")
460            })?;
461            let image =
462                images::decode_image(encoded, Some(mime), max_bytes).map_err(input_error)?;
463            (image.bytes, image.metadata.mime().to_owned())
464        } else {
465            read_resource(
466                access,
467                scope,
468                reference,
469                codec_limits,
470                max_bytes,
471                observed_at,
472            )
473            .await?
474        }
475    } else {
476        read_resource(
477            access,
478            scope,
479            reference,
480            codec_limits,
481            max_bytes,
482            observed_at,
483        )
484        .await?
485    };
486    let len = bytes.len() as u64;
487    Ok((
488        ResolvedImageInput {
489            reference: reference.clone(),
490            bytes_base64: base64::engine::general_purpose::STANDARD.encode(bytes),
491            mime_type: mime,
492        },
493        len,
494    ))
495}
496
497async fn read_resource<A: ResourceAccess>(
498    access: &A,
499    scope: &A::Scope,
500    reference: &ResourceReference,
501    codec_limits: CodecLimits,
502    max_bytes: u64,
503    observed_at: std::time::SystemTime,
504) -> Result<(bytes::Bytes, String), TransformError> {
505    let read = access
506        .read(scope, reference)
507        .await
508        .map_err(TransformError::from)?;
509    let cap = max_bytes.min(access.limits().read_bytes);
510    if read.metadata.length.is_some_and(|n| n > cap) {
511        return Err(limit("image.read.length"));
512    }
513    if read.metadata.expires_at.is_some_and(|t| t <= observed_at) {
514        return Err(TransformError::invalid_result(
515            "image.read",
516            "host returned expired resource",
517        ));
518    }
519    let mime = read
520        .metadata
521        .mime
522        .ok_or_else(|| TransformError::missing_metadata("image.mime"))?;
523    let bytes = codec::read_http_body(
524        read.body,
525        CodecLimits {
526            max_body_bytes: cap,
527            ..codec_limits
528        },
529    )
530    .await
531    .map_err(|e| {
532        TransformError::new(
533            if e.kind() == codec::CodecErrorKind::Limit {
534                TransformErrorKind::Limit
535            } else {
536                TransformErrorKind::Host
537            },
538            "image.read",
539            e.to_string(),
540        )
541    })?;
542    if read
543        .metadata
544        .length
545        .is_some_and(|n| n != bytes.len() as u64)
546    {
547        return Err(TransformError::invalid_result(
548            "image.read.length",
549            "metadata length differs from actual bytes",
550        ));
551    }
552    let metadata = images::inspect_image(&bytes)?;
553    if metadata.mime() != mime {
554        return Err(TransformError::invalid_result(
555            "image.read.mime",
556            "MIME differs from actual image bytes",
557        ));
558    }
559    Ok((bytes, mime))
560}
561
562fn input_error(error: TransformError) -> TransformError {
563    TransformError::with_source(
564        if error.kind() == TransformErrorKind::InvalidResult {
565            TransformErrorKind::InvalidInput
566        } else {
567            error.kind()
568        },
569        "image.input",
570        error.to_string(),
571        error,
572    )
573}
gproxy_protocol/adapt/images.rs

gproxy_protocol/adapt/
images.rs
